feat: add shadow LangGraph orchestrator runtime

2026-07-24 21:25:40 +08:00
parent 46ea22b342
commit 24336d0178
29 changed files with 3247 additions and 25 deletions
@@ -0,0 +1,137 @@
+import { normalizeRunSpec } from './contracts.mjs';
+import {
+  CODE_RUN_SHADOW_WORKFLOW,
+  createCodeRunShadowGraph,
+} from './code-run-shadow-graph.mjs';
+
+function graphConfig(runId) {
+  return {
+    configurable: {
+      thread_id: runId,
+      checkpoint_ns: '',
+    },
+  };
+}
+
+function projectSnapshot(runId, snapshot) {
+  const values = snapshot?.values ?? {};
+  if (!values.spec) return null;
+  return {
+    version: 'orchestrator-state-v1',
+    runId,
+    requestId: values.spec.requestId,
+    workflow: values.spec.workflow,
+    status: values.status ?? 'unknown',
+    phase: values.phase ?? null,
+    plan: values.plan ?? null,
+    result: values.result ?? null,
+    createdAt: values.events?.[0]?.timestamp ?? null,
+    updatedAt: values.events?.at?.(-1)?.timestamp ?? null,
+  };
+}
+
+export function createLangGraphOrchestratorRuntime({
+  checkpointer,
+  checkpointKind = 'unknown',
+  durable = false,
+  checkpointProbe = null,
+} = {}) {
+  const graph = createCodeRunShadowGraph({ checkpointer });
+  const probe = typeof checkpointProbe === 'function'
+    ? checkpointProbe
+    : async () => {
+        await checkpointer.getTuple(graphConfig('__orchestrator_readiness__'));
+        return true;
+      };
+
+  async function getSnapshot(runId) {
+    return graph.getState(graphConfig(runId));
+  }
+
+  async function getState(runId) {
+    const normalizedRunId = String(runId ?? '').trim();
+    if (!normalizedRunId) return null;
+    return projectSnapshot(normalizedRunId, await getSnapshot(normalizedRunId));
+  }
+
+  function health() {
+    return {
+      status: 'ok',
+      service: 'memind-langgraph-orchestrator',
+      checkpoint: {
+        kind: checkpointKind,
+        durable: Boolean(durable),
+      },
+      execution: 'observe-only',
+    };
+  }
+
+  return {
+    async start(input) {
+      const spec = normalizeRunSpec(input);
+      if (spec.workflow.name !== CODE_RUN_SHADOW_WORKFLOW) {
+        const error = new Error(`Unsupported workflow: ${spec.workflow.name}`);
+        error.code = 'WORKFLOW_NOT_SUPPORTED';
+        error.status = 422;
+        throw error;
+      }
+      const existing = await getState(spec.runId);
+      if (existing) return existing;
+      await graph.invoke({
+        spec,
+        status: 'queued',
+        phase: 'accepted',
+        events: [],
+      }, graphConfig(spec.runId));
+      return getState(spec.runId);
+    },
+
+    getState,
+
+    async listEvents(runId, { after = 0 } = {}) {
+      const normalizedRunId = String(runId ?? '').trim();
+      const snapshot = normalizedRunId ? await getSnapshot(normalizedRunId) : null;
+      if (!snapshot?.values?.spec) return null;
+      const cursor = Math.max(0, Number(after) || 0);
+      const events = (snapshot.values.events ?? []).filter((event) => event.sequence > cursor);
+      return {
+        runId: normalizedRunId,
+        events,
+        nextCursor: events.at(-1)?.sequence ?? cursor,
+      };
+    },
+
+    async resume(runId) {
+      const state = await getState(runId);
+      if (!state) return null;
+      const error = new Error('This workflow has no interrupt point to resume');
+      error.code = 'WORKFLOW_NOT_INTERRUPTED';
+      error.status = 409;
+      throw error;
+    },
+
+    async cancel(runId) {
+      const state = await getState(runId);
+      if (!state) return null;
+      const error = new Error('Completed shadow observations cannot be cancelled');
+      error.code = 'WORKFLOW_ALREADY_TERMINAL';
+      error.status = 409;
+      throw error;
+    },
+
+    health,
+
+    async ready() {
+      await probe();
+      return {
+        ...health(),
+        ready: true,
+      };
+    },
+  };
+}
+
+export const orchestratorRuntimeInternals = {
+  graphConfig,
+  projectSnapshot,
+};