john
322 lines
9.8 KiB
JavaScript
322 lines
9.8 KiB
JavaScript
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,
|
|
);
|
|
});
|