import assert from 'node:assert/strict'; import test from 'node:test'; import { MemorySaver } from '@langchain/langgraph'; import { createInMemoryExecutorJobStore } from './executor-gateway.mjs'; import { createLangGraphOrchestratorRuntime } from './runtime.mjs'; function runSpec(overrides = {}) { return { runId: 'run-shadow-1', requestId: 'request-shadow-1', workflow: { name: 'code-run-v1', version: 1 }, subject: { userId: 'user-1' }, input: { instruction: 'Add an idempotency test', taskType: 'code-change', }, policy: { executionMode: 'observe-only', sideEffectsAllowed: false, }, ...overrides, }; } test('LangGraph runtime checkpoints a deterministic observe-only code workflow', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), checkpointKind: 'memory', durable: false, }); const state = await runtime.start(runSpec()); assert.equal(state.runId, 'run-shadow-1'); assert.equal(state.status, 'succeeded'); assert.equal(state.phase, 'completed'); assert.equal(state.result.observed, true); assert.equal(state.result.executed, false); assert.equal(state.plan.executorAdapter, 'native-agent-run'); assert.deepEqual(state.plan.executorJob, { kind: 'executor-job', id: 'run-shadow-1:executor-preview', executor: 'goosed', status: 'blocked', reason: 'executor_execution_disabled', durable: false, }); assert.equal( (await runtime.getExecutorJob('run-shadow-1:executor-preview')).status, 'blocked', ); const executorEvents = await runtime.listExecutorJobEvents( 'run-shadow-1:executor-preview', ); assert.deepEqual( executorEvents.events.map((event) => event.type), ['executor_job_blocked'], ); assert.equal(executorEvents.nextCursor, 1); const restored = await runtime.getState('run-shadow-1'); assert.deepEqual(restored, state); const eventPage = await runtime.listEvents('run-shadow-1'); assert.deepEqual( eventPage.events.map((event) => event.type), ['workflow_validated', 'workflow_planned', 'workflow_completed'], ); assert.deepEqual( eventPage.events.map((event) => event.sequence), [1, 2, 3], ); assert.equal(eventPage.nextCursor, 3); assert.equal((await runtime.listEvents('run-shadow-1', { after: 2 })).events.length, 1); }); test('LangGraph runtime is idempotent by run id', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), }); const first = await runtime.start(runSpec()); const second = await runtime.start(runSpec({ input: { instruction: 'This must not replace the first checkpoint' }, })); assert.deepEqual(second, first); assert.equal((await runtime.listEvents('run-shadow-1')).events.length, 3); }); test('LangGraph runtime records idempotent Page Data validation evidence after Shadow completion', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), }); await runtime.start(runSpec({ input: { instruction: '[shadow-control-plane-only]', taskType: 'page_data_dev_complex', pageDataRequired: true, }, })); const input = { idempotencyKey: 'run-shadow-1:page-data-delivery:v1', taskType: 'page_data_dev_complex', required: true, checks: [ { id: 'agent_run_completion', status: 'passed' }, { id: 'page_data_binding', status: 'passed' }, { id: 'page_data_storage_policy', status: 'passed' }, ], metrics: { pageCount: 2, publicationCount: 1 }, observedAt: 123, }; const recorded = await runtime.recordValidationObservation('run-shadow-1', input); assert.equal(recorded.status, 'succeeded'); assert.equal(recorded.validation.verdict, 'passed'); assert.equal(recorded.validation.dataPolicy, 'control-plane-only-v1'); assert.deepEqual( (await runtime.listEvents('run-shadow-1')).events.map((event) => event.type), [ 'workflow_validated', 'workflow_planned', 'workflow_completed', 'workflow_validation_observed', ], ); const duplicate = await runtime.recordValidationObservation('run-shadow-1', { ...input, observedAt: 456, }); assert.deepEqual(duplicate, recorded); assert.equal((await runtime.listEvents('run-shadow-1')).events.length, 4); await assert.rejects( () => runtime.recordValidationObservation('run-shadow-1', { ...input, checks: [{ id: 'page_data_binding', status: 'failed' }], }), (error) => error.code === 'WORKFLOW_VALIDATION_IDEMPOTENCY_CONFLICT' && error.status === 409, ); assert.equal(await runtime.recordValidationObservation('missing-run', input), null); }); test('LangGraph runtime deletes terminal checkpoints and linked Executor Jobs', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), }); await runtime.start(runSpec()); assert.deepEqual(await runtime.deleteRun('run-shadow-1'), { runId: 'run-shadow-1', deleted: true, executorJobDeleted: true, }); assert.equal(await runtime.getState('run-shadow-1'), null); assert.equal(await runtime.getExecutorJob('run-shadow-1:executor-preview'), null); assert.equal(await runtime.deleteRun('run-shadow-1'), null); }); test('LangGraph runtime retention purge is dry-run by default and deletes only on apply', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), }); await runtime.start(runSpec({ runId: 'run-retention-1' })); const before = Date.now() + 1000; const preview = await runtime.purgeTerminalRuns({ before }); assert.equal(preview.dryRun, true); assert.equal(preview.candidates.length, 1); assert.equal(await runtime.getState('run-retention-1') != null, true); const applied = await runtime.purgeTerminalRuns({ before, dryRun: false, }); assert.equal(applied.deleted.length, 1); assert.equal(applied.failures.length, 0); assert.equal(await runtime.getState('run-retention-1'), null); await assert.rejects( () => runtime.purgeTerminalRuns({ before: 0 }), (error) => error.code === 'WORKFLOW_PURGE_CUTOFF_REQUIRED', ); }); test('LangGraph runtime persists a durable blocked Executor Job and probes both stores', async () => { const checkpointProbeCalls = []; const executorProbeCalls = []; const executorJobStore = { ...createInMemoryExecutorJobStore(), kind: 'postgres', durable: true, }; const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), checkpointKind: 'postgres', durable: true, checkpointProbe: async () => checkpointProbeCalls.push('checkpoint'), executorJobStore, executorJobProbe: async () => executorProbeCalls.push('executor-jobs'), }); const state = await runtime.start(runSpec()); assert.equal(state.plan.executorJob.durable, true); assert.equal( (await executorJobStore.getById('run-shadow-1:executor-preview')).status, 'blocked', ); const ready = await runtime.ready(); assert.equal(ready.ready, true); assert.equal(ready.executorGateway.store.kind, 'postgres'); assert.equal(ready.executorGateway.store.durable, true); assert.equal(ready.executorGateway.executionEnabled, false); assert.deepEqual(checkpointProbeCalls, ['checkpoint']); assert.deepEqual(executorProbeCalls, ['executor-jobs']); }); test('LangGraph runtime rejects execution-enabled and unsupported workflows', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), }); await assert.rejects( runtime.start(runSpec({ runId: 'run-active', policy: { executionMode: 'active', sideEffectsAllowed: true }, })), (error) => error.code === 'WORKFLOW_OBSERVE_ONLY_REQUIRED', ); await assert.rejects( runtime.start(runSpec({ runId: 'run-unsupported', workflow: { name: 'research-v1', version: 1 }, })), (error) => error.code === 'WORKFLOW_NOT_SUPPORTED' && error.status === 422, ); }); test('LangGraph runtime derives a bounded opaque Executor Job id for long run ids', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), }); const state = await runtime.start(runSpec({ runId: 'r'.repeat(128), requestId: 'request-long-run-id', })); assert.match(state.plan.executorJob.id, /^run_[a-f0-9]{64}:executor-preview$/); assert.ok(state.plan.executorJob.id.length <= 128); assert.equal((await runtime.getExecutorJob(state.plan.executorJob.id)).status, 'blocked'); }); test('LangGraph active workflow waits on a leased Executor Job and converges terminal state', async () => { const runtime = createLangGraphOrchestratorRuntime({ checkpointer: new MemorySaver(), executionEnabled: true, enabledExecutors: ['goosed'], }); const state = await runtime.start({ runId: 'run-active-1', requestId: 'request-active-1', workflow: { name: 'code-run-v1', version: 1 }, subject: { tenantId: 'tenant-1', userId: 'user-1' }, input: { instruction: 'Run a controlled canary', executor: 'goosed', workspaceRef: { kind: 'workspace-alias', id: 'canary' }, }, policy: { executionMode: 'active', sideEffectsAllowed: true, networkAllowed: false, }, }); assert.equal(state.status, 'waiting'); assert.equal(state.phase, 'executor_queued'); assert.equal(state.plan.executorJob.status, 'queued'); await assert.rejects( () => runtime.deleteRun('run-active-1'), (error) => error.code === 'WORKFLOW_RUN_NOT_TERMINAL' && error.status === 409, ); const claim = await runtime.claimExecutorJob({ workerId: 'worker-1', executors: ['goosed'], }); await runtime.startExecutorJob(claim.job.jobId, { leaseToken: claim.leaseToken }); await runtime.completeExecutorJob(claim.job.jobId, { leaseToken: claim.leaseToken, result: { summary: 'canary complete', artifactRefs: [{ kind: 'canary-result', id: 'result-1' }], }, }); const completed = await runtime.getState('run-active-1'); assert.equal(completed.status, 'succeeded'); assert.equal(completed.result.executed, true); assert.deepEqual( completed.result.executorJob.result.artifactRefs, [{ kind: 'canary-result', id: 'result-1' }], ); assert.deepEqual( (await runtime.listEvents('run-active-1')).events.map((event) => event.type), [ 'workflow_validated', 'workflow_planned', 'workflow_executor_queued', 'workflow_completed', ], ); });