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 validation = await fetch( `${server.baseUrl}/v1/runs/http-shadow-1/validation-observations`, { method: 'POST', headers: { authorization: 'Bearer test-service-token', 'content-type': 'application/json', }, body: JSON.stringify({ idempotencyKey: 'http-shadow-1:page-data-delivery:v1', taskType: 'page_data_dev', required: true, checks: [ { id: 'agent_run_completion', status: 'passed' }, { id: 'page_data_binding', status: 'passed' }, { id: 'page_data_storage_policy', status: 'passed' }, ], metrics: { pageCount: 1, publicationCount: 1 }, observedAt: 123, }), }, ); assert.equal(validation.status, 200); assert.equal((await validation.json()).validation.verdict, 'passed'); 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, 4]); assert.equal(eventBody.events.at(-1).type, 'workflow_validation_observed'); 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); });