john
311 lines
10 KiB
JavaScript
311 lines
10 KiB
JavaScript
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',
|
|
],
|
|
);
|
|
});
|