memind/services/orchestrator/runtime.mjs

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