memind/services/orchestrator/runtime.test.mjs

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