memind/services/orchestrator/app.test.mjs

import assert from 'node:assert/strict';
import test from 'node:test';
import { MemorySaver } from '@langchain/langgraph';
import { createOrchestratorApp } from './app.mjs';
import { createLangGraphOrchestratorRuntime } from './runtime.mjs';

async function listen(app) {
  const server = await new Promise((resolve) => {
    const candidate = app.listen(0, '127.0.0.1', () => resolve(candidate));
  });
  const address = server.address();
  return {
    baseUrl: `http://127.0.0.1:${address.port}`,
    close: () => new Promise((resolve, reject) => {
      server.close((error) => (error ? reject(error) : resolve()));
    }),
  };
}

function spec() {
  return {
    runId: 'http-shadow-1',
    requestId: 'http-request-1',
    workflow: { name: 'code-run-v1', version: 1 },
    input: { instruction: 'Inspect without executing' },
    policy: { executionMode: 'observe-only', sideEffectsAllowed: false },
  };
}

test('orchestrator HTTP API exposes health and authenticated run endpoints', async (t) => {
  const runtime = createLangGraphOrchestratorRuntime({
    checkpointer: new MemorySaver(),
    checkpointKind: 'memory',
  });
  const server = await listen(createOrchestratorApp({
    runtime,
    serviceToken: 'test-service-token',
  }));
  t.after(server.close);

  const health = await fetch(`${server.baseUrl}/health`);
  assert.equal(health.status, 200);
  const healthBody = await health.json();
  assert.equal(healthBody.execution, 'observe-only');
  assert.equal(healthBody.executorGateway.executionEnabled, false);
  assert.deepEqual(healthBody.executorGateway.store, {
    kind: 'memory',
    durable: false,
  });
  const ready = await fetch(`${server.baseUrl}/ready`);
  assert.equal(ready.status, 200);
  assert.equal((await ready.json()).ready, true);
  const metrics = await fetch(`${server.baseUrl}/metrics`);
  assert.equal(metrics.status, 200);
  assert.match(await metrics.text(), /memind_orchestrator_execution_enabled 0/);

  const unauthorized = await fetch(`${server.baseUrl}/v1/runs`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify(spec()),
  });
  assert.equal(unauthorized.status, 401);
  const unauthorizedExecutorJob = await fetch(
    `${server.baseUrl}/v1/executor-jobs/http-shadow-1%3Aexecutor-preview`,
  );
  assert.equal(unauthorizedExecutorJob.status, 401);

  const created = await fetch(`${server.baseUrl}/v1/runs`, {
    method: 'POST',
    headers: {
      authorization: 'Bearer test-service-token',
      'content-type': 'application/json',
    },
    body: JSON.stringify(spec()),
  });
  assert.equal(created.status, 202);
  assert.equal((await created.json()).status, 'succeeded');

  const events = await fetch(`${server.baseUrl}/v1/runs/http-shadow-1/events?after=1`, {
    headers: { authorization: 'Bearer test-service-token' },
  });
  const eventBody = await events.json();
  assert.equal(events.status, 200);
  assert.deepEqual(eventBody.events.map((event) => event.sequence), [2, 3]);

  const jobId = 'http-shadow-1:executor-preview';
  const executorJob = await fetch(
    `${server.baseUrl}/v1/executor-jobs/${encodeURIComponent(jobId)}`,
    { headers: { authorization: 'Bearer test-service-token' } },
  );
  assert.equal(executorJob.status, 200);
  assert.equal((await executorJob.json()).status, 'blocked');

  const executorEvents = await fetch(
    `${server.baseUrl}/v1/executor-jobs/${encodeURIComponent(jobId)}/events?after=0`,
    { headers: { authorization: 'Bearer test-service-token' } },
  );
  const executorEventBody = await executorEvents.json();
  assert.equal(executorEvents.status, 200);
  assert.deepEqual(
    executorEventBody.events.map((event) => event.type),
    ['executor_job_blocked'],
  );
  assert.equal(executorEventBody.nextCursor, 1);

  const purgePreview = await fetch(
    `${server.baseUrl}/v1/maintenance/purge-terminal-runs`,
    {
      method: 'POST',
      headers: {
        authorization: 'Bearer test-service-token',
        'content-type': 'application/json',
      },
      body: JSON.stringify({ before: Date.now() + 1000 }),
    },
  );
  assert.equal(purgePreview.status, 200);
  const purgePreviewBody = await purgePreview.json();
  assert.equal(purgePreviewBody.dryRun, true);
  assert.equal(purgePreviewBody.candidates.length, 1);

  const deleted = await fetch(`${server.baseUrl}/v1/runs/http-shadow-1`, {
    method: 'DELETE',
    headers: { authorization: 'Bearer test-service-token' },
  });
  assert.equal(deleted.status, 200);
  assert.deepEqual(await deleted.json(), {
    runId: 'http-shadow-1',
    deleted: true,
    executorJobDeleted: true,
  });
  assert.equal(
    (await fetch(`${server.baseUrl}/v1/runs/http-shadow-1`, {
      headers: { authorization: 'Bearer test-service-token' },
    })).status,
    404,
  );
});

test('orchestrator HTTP API reports missing runs and observe-only violations', async (t) => {
  const runtime = createLangGraphOrchestratorRuntime({
    checkpointer: new MemorySaver(),
  });
  const server = await listen(createOrchestratorApp({ runtime }));
  t.after(server.close);

  const missing = await fetch(`${server.baseUrl}/v1/runs/missing`);
  assert.equal(missing.status, 404);
  const missingExecutorJob = await fetch(`${server.baseUrl}/v1/executor-jobs/missing`);
  assert.equal(missingExecutorJob.status, 404);
  assert.equal((await missingExecutorJob.json()).error.code, 'EXECUTOR_JOB_NOT_FOUND');

  const activeSpec = spec();
  activeSpec.runId = 'http-active-1';
  activeSpec.policy = { executionMode: 'active', sideEffectsAllowed: true };
  const rejected = await fetch(`${server.baseUrl}/v1/runs`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify(activeSpec),
  });
  assert.equal(rejected.status, 422);
  assert.equal((await rejected.json()).error.code, 'WORKFLOW_OBSERVE_ONLY_REQUIRED');
});

test('executor worker HTTP protocol requires both service and worker credentials', async (t) => {
  const runtime = createLangGraphOrchestratorRuntime({
    checkpointer: new MemorySaver(),
    executionEnabled: true,
    enabledExecutors: ['goosed'],
  });
  const server = await listen(createOrchestratorApp({
    runtime,
    serviceToken: 'service-token',
    workerToken: 'worker-token',
  }));
  t.after(server.close);
  const serviceHeaders = {
    authorization: 'Bearer service-token',
    'content-type': 'application/json',
  };
  const submitted = await fetch(`${server.baseUrl}/v1/executor-jobs`, {
    method: 'POST',
    headers: serviceHeaders,
    body: JSON.stringify({
      jobId: 'http-worker-job-1',
      idempotencyKey: 'http-worker-idem-1',
      executor: 'goosed',
      task: {
        instruction: 'Run the canary',
        workspaceRef: { kind: 'workspace-alias', id: 'canary' },
      },
      subject: { tenantId: 'tenant-1', userId: 'user-1' },
      authorization: { executionAllowed: true },
      policy: { sideEffectsAllowed: true },
    }),
  });
  assert.equal(submitted.status, 202);
  assert.equal((await submitted.json()).job.status, 'queued');

  const unauthorizedWorker = await fetch(`${server.baseUrl}/v1/workers/claim`, {
    method: 'POST',
    headers: serviceHeaders,
    body: JSON.stringify({ workerId: 'worker-1', executors: ['goosed'] }),
  });
  assert.equal(unauthorizedWorker.status, 401);
  assert.equal((await unauthorizedWorker.json()).error.code, 'WORKER_UNAUTHORIZED');

  const workerHeaders = {
    ...serviceHeaders,
    'x-orchestrator-worker-token': 'worker-token',
  };
  const claimResponse = await fetch(`${server.baseUrl}/v1/workers/claim`, {
    method: 'POST',
    headers: workerHeaders,
    body: JSON.stringify({ workerId: 'worker-1', executors: ['goosed'] }),
  });
  assert.equal(claimResponse.status, 200);
  const claim = await claimResponse.json();
  assert.equal(claim.job.request.task.instruction, 'Run the canary');

  const started = await fetch(
    `${server.baseUrl}/v1/workers/jobs/http-worker-job-1/start`,
    {
      method: 'POST',
      headers: workerHeaders,
      body: JSON.stringify({ leaseToken: claim.leaseToken }),
    },
  );
  assert.equal(started.status, 200);

  const completed = await fetch(
    `${server.baseUrl}/v1/workers/jobs/http-worker-job-1/complete`,
    {
      method: 'POST',
      headers: workerHeaders,
      body: JSON.stringify({
        leaseToken: claim.leaseToken,
        result: {
          summary: 'completed',
          artifactRefs: [{ kind: 'canary-result', id: 'result-1' }],
        },
      }),
    },
  );
  assert.equal(completed.status, 200);
  assert.equal((await completed.json()).status, 'succeeded');

  const projected = await fetch(
    `${server.baseUrl}/v1/executor-jobs/http-worker-job-1`,
    { headers: { authorization: 'Bearer service-token' } },
  );
  const projectedBody = await projected.json();
  assert.equal(projectedBody.status, 'succeeded');
  assert.equal('request' in projectedBody, false);
  assert.equal('token' in (projectedBody.lease ?? {}), false);
});