memind/services/orchestrator/shadow-observer.test.mjs

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,
  );
});