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'; import { createWorkflowShadowObserver } from './shadow-observer.mjs'; function jsonResponse(body, status = 200) { return new Response(JSON.stringify(body), { status, headers: { 'content-type': 'application/json' }, }); } 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())); }), }; } test('shadow observer skips without creating a remote client when mode does not select shadow', async () => { let fetchCalls = 0; const observer = createWorkflowShadowObserver({ configService: { async selectEngine() { return { shadowEngine: null, reason: 'mode_off', mode: 'off' }; }, async getRuntimeState() { throw new Error('must not load runtime config'); }, }, fetchImpl: async () => { fetchCalls += 1; return jsonResponse({}); }, }); const result = await observer({ runId: 'run-1', requestId: 'request-1', userId: 'user-1', }); assert.deepEqual(result, { observed: false, reason: 'mode_off', mode: 'off', executionPlan: null, }); assert.equal(fetchCalls, 0); }); test('shadow boundary returns an auditable dry-run plan without calling LangGraph', async () => { let fetchCalls = 0; const observer = createWorkflowShadowObserver({ configService: { async selectEngine() { return { engine: 'native', candidateEngine: 'langgraph', shadowEngine: null, fallbackEngine: 'native', reason: 'execution_gate_disabled', candidateReason: 'user_allowlist', mode: 'canary', configVersion: 8, bucket: 42, dryRun: true, }; }, async getRuntimeState() { throw new Error('must not load runtime config'); }, }, fetchImpl: async () => { fetchCalls += 1; return jsonResponse({}); }, }); const result = await observer({ runId: 'run-dry-1', requestId: 'request-dry-1', userId: 'user-1', taskType: 'repo_refactor', }); assert.equal(result.observed, false); assert.deepEqual(result.executionPlan, { version: 'workflow-execution-decision-v1', candidateEngine: 'langgraph', effectiveEngine: 'native', fallbackEngine: 'native', reason: 'execution_gate_disabled', candidateReason: 'user_allowlist', configVersion: 8, bucket: 42, taskType: 'repo_refactor', dryRun: true, handoffAllowed: false, }); assert.equal(fetchCalls, 0); }); test('shadow boundary records non-selected canary decisions for a complete rate denominator', async () => { const observer = createWorkflowShadowObserver({ configService: { async selectEngine() { return { engine: 'native', candidateEngine: 'native', shadowEngine: null, fallbackEngine: 'native', reason: 'canary_not_selected', candidateReason: null, mode: 'canary', configVersion: 9, bucket: 88, dryRun: false, }; }, async getRuntimeState() { throw new Error('must not load runtime config'); }, }, }); const result = await observer({ runId: 'run-native-1', requestId: 'request-native-1', userId: 'user-1', taskType: 'code_analysis', }); assert.equal(result.observed, false); assert.equal(result.executionPlan.dryRun, true); assert.equal(result.executionPlan.candidateEngine, 'native'); assert.equal(result.executionPlan.reason, 'canary_not_selected'); assert.equal(result.executionPlan.taskType, 'code_analysis'); }); test('shadow observer sends a control-plane-only RunSpec without user content or identity', async () => { let capturedUrl = null; let capturedInit = null; const observer = createWorkflowShadowObserver({ configService: { async selectEngine() { return { shadowEngine: 'langgraph', reason: 'shadow', mode: 'shadow', configVersion: 7, }; }, async getRuntimeState() { return { config: { serviceUrl: 'http://orchestrator.internal:8093', requestTimeoutMs: 1200, }, }; }, }, serviceToken: 'internal-token', fetchImpl: async (url, init) => { capturedUrl = url; capturedInit = init; return jsonResponse({ runId: 'run-2', status: 'succeeded' }, 202); }, }); const result = await observer({ runId: 'run-2', requestId: 'request-2', userId: 'user-2', sessionId: 'session-2', taskType: 'code-change', pageDataRequired: true, userMessage: { content: [{ type: 'text', text: 'Implement the service boundary' }], }, }); assert.equal(result.observed, true); assert.equal(capturedUrl, 'http://orchestrator.internal:8093/v1/runs'); assert.equal(capturedInit.headers.authorization, 'Bearer internal-token'); const body = JSON.parse(capturedInit.body); assert.equal(body.version, 'orchestrator-run-v1'); assert.equal(body.policy.executionMode, 'observe-only'); assert.equal(body.policy.sideEffectsAllowed, false); assert.equal(body.input.instruction, '[shadow-control-plane-only]'); assert.equal(body.input.pageDataRequired, true); assert.equal('sessionRef' in body.input, false); assert.deepEqual(body.subject, { tenantId: null, userId: null }); assert.equal(body.metadata.configVersion, 7); assert.equal(body.metadata.dataPolicy, 'control-plane-only-v1'); assert.equal( capturedInit.body.includes('Implement the service boundary'), false, ); assert.equal(capturedInit.body.includes('user-2'), false); assert.equal(capturedInit.body.includes('session-2'), false); }); test('isolated Portal shadow flow reaches LangGraph over HTTP and removes all terminal state', async (t) => { const checkpointer = new MemorySaver(); const runtime = createLangGraphOrchestratorRuntime({ checkpointer, checkpointKind: 'memory', durable: false, }); const server = await listen(createOrchestratorApp({ runtime, serviceToken: 'isolated-shadow-token', })); t.after(server.close); const observer = createWorkflowShadowObserver({ configService: { async selectEngine() { return { shadowEngine: 'langgraph', reason: 'shadow', mode: 'shadow', configVersion: 11, }; }, async getRuntimeState() { return { config: { serviceUrl: server.baseUrl, requestTimeoutMs: 2000, }, }; }, }, serviceToken: 'isolated-shadow-token', }); const secretUserContent = 'private user content must never cross the shadow boundary'; const result = await observer({ runId: 'isolated-shadow-run-1', requestId: 'isolated-shadow-request-1', userId: 'private-user-id', sessionId: 'private-session-id', taskType: 'repo_refactor', userMessage: { role: 'user', content: [{ type: 'text', text: secretUserContent }], }, }); assert.equal(result.observed, true); assert.equal(result.shadowRun.status, 'succeeded'); assert.equal(result.shadowRun.result.executed, false); const tuple = await checkpointer.getTuple({ configurable: { thread_id: 'isolated-shadow-run-1', checkpoint_ns: '', }, }); assert.equal( tuple.checkpoint.channel_values.spec.input.instruction, '[shadow-control-plane-only]', ); const checkpointJson = JSON.stringify(tuple.checkpoint.channel_values.spec); assert.equal(checkpointJson.includes(secretUserContent), false); assert.equal(checkpointJson.includes('private-user-id'), false); assert.equal(checkpointJson.includes('private-session-id'), false); const validation = await observer.observeValidation({ runId: 'isolated-shadow-run-1', requestId: 'isolated-shadow-request-1', userId: 'private-user-id', workflowName: 'code-run-v1', observation: { idempotencyKey: 'isolated-shadow-run-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 }, source: 'portal-agent-run', observedAt: 123, sensitive: { workspacePath: '/private/workspace', instruction: secretUserContent, }, }, }); assert.equal(validation.observed, true); assert.equal(validation.validation.verdict, 'passed'); const validatedState = await runtime.getState('isolated-shadow-run-1'); assert.equal(validatedState.validation.verdict, 'passed'); const validatedJson = JSON.stringify(validatedState.validation); assert.equal(validatedJson.includes('/private/workspace'), false); assert.equal(validatedJson.includes(secretUserContent), false); const deleted = await fetch(`${server.baseUrl}/v1/runs/isolated-shadow-run-1`, { method: 'DELETE', headers: { authorization: 'Bearer isolated-shadow-token' }, }); assert.equal(deleted.status, 200); assert.deepEqual(await deleted.json(), { runId: 'isolated-shadow-run-1', deleted: true, executorJobDeleted: true, }); assert.equal(await runtime.getState('isolated-shadow-run-1'), null); assert.equal( await runtime.getExecutorJob('isolated-shadow-run-1:executor-preview'), null, ); });