memind/services/orchestrator/shadow-observer.mjs

import {
  WORKFLOW_ENGINE,
  normalizePageDataValidationObservation,
  normalizeRunSpec,
} from './contracts.mjs';
import { createRemoteWorkflowEngine } from './engine-registry.mjs';

const VALIDATION_RETRY_DELAYS_MS = Object.freeze([25, 75, 225]);

function safeError(error) {
  return {
    code: String(error?.code ?? 'WORKFLOW_SHADOW_FAILED').slice(0, 128),
    message: String(error instanceof Error ? error.message : error).slice(0, 1000),
  };
}

async function recordValidationWithRetry(engine, runId, observation) {
  let attempt = 0;
  while (true) {
    try {
      return await engine.recordValidationObservation(runId, observation);
    } catch (error) {
      const delayMs = VALIDATION_RETRY_DELAYS_MS[attempt];
      if (error?.status !== 404 || delayMs == null) throw error;
      attempt += 1;
      await new Promise((resolve) => setTimeout(resolve, delayMs));
    }
  }
}

export function createWorkflowShadowObserver({
  configService,
  serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN,
  fetchImpl = globalThis.fetch,
  logger = console,
} = {}) {
  if (!configService?.selectEngine || !configService?.getRuntimeState) {
    throw new Error('Workflow shadow observer requires orchestrator config service');
  }

  async function selectShadowEngine({
    runId,
    requestId,
    userId,
    workflowName,
  }) {
    const selection = await configService.selectEngine({
      runId,
      requestId,
      userId,
      workflowName,
    });
    if (selection.shadowEngine !== WORKFLOW_ENGINE.LANGGRAPH) {
      return { selection, engine: null };
    }
    const state = await configService.getRuntimeState();
    return {
      selection,
      engine: createRemoteWorkflowEngine({
        id: WORKFLOW_ENGINE.LANGGRAPH,
        baseUrl: state.config.serviceUrl,
        serviceToken,
        timeoutMs: state.config.requestTimeoutMs,
        fetchImpl,
      }),
    };
  }

  async function observeWorkflowRun({
    runId,
    requestId,
    userId,
    workflowName = 'code-run-v1',
    taskType = null,
    pageDataRequired = false,
  } = {}) {
    const { selection, engine } = await selectShadowEngine({
      runId,
      requestId,
      userId,
      workflowName,
    });
    const planningMode = ['shadow', 'canary', 'active'].includes(selection.mode);
    const executionPlan = planningMode
      && (selection.dryRun || ['canary', 'active'].includes(selection.mode)) ? {
      version: 'workflow-execution-decision-v1',
      candidateEngine: selection.candidateEngine ?? WORKFLOW_ENGINE.NATIVE,
      effectiveEngine: selection.engine ?? WORKFLOW_ENGINE.NATIVE,
      fallbackEngine: selection.fallbackEngine ?? WORKFLOW_ENGINE.NATIVE,
      reason: selection.reason,
      candidateReason: selection.candidateReason ?? null,
      configVersion: selection.configVersion ?? null,
      bucket: selection.bucket ?? null,
      taskType,
      dryRun: true,
      handoffAllowed: false,
    } : null;
    if (!engine) {
      return {
        observed: false,
        reason: selection.reason,
        mode: selection.mode,
        executionPlan,
      };
    }

    const spec = normalizeRunSpec({
      runId,
      requestId,
      workflow: { name: workflowName, version: 1 },
      subject: {},
      input: {
        instruction: '[shadow-control-plane-only]',
        taskType,
        toolMode: 'code',
        pageDataRequired: pageDataRequired === true,
      },
      policy: {
        executionMode: 'observe-only',
        sideEffectsAllowed: false,
      },
      metadata: {
        source: 'memind-agent-run',
        configVersion: selection.configVersion,
        dataPolicy: 'control-plane-only-v1',
      },
    });
    try {
      const result = await engine.start(spec);
      return {
        observed: true,
        engine: WORKFLOW_ENGINE.LANGGRAPH,
        mode: selection.mode,
        configVersion: selection.configVersion,
        executionPlan,
        shadowRun: result,
      };
    } catch (error) {
      logger.warn('[orchestrator-shadow] observation failed:', safeError(error));
      if (error && typeof error === 'object') {
        error.executionPlan = executionPlan;
        error.mode = selection.mode;
      }
      throw error;
    }
  }

  observeWorkflowRun.observeValidation = async function observeWorkflowValidation({
    runId,
    requestId,
    userId,
    workflowName = 'code-run-v1',
    observation,
  } = {}) {
    const { selection, engine } = await selectShadowEngine({
      runId,
      requestId,
      userId,
      workflowName,
    });
    if (!engine) {
      return {
        observed: false,
        reason: selection.reason,
        mode: selection.mode,
      };
    }
    const normalized = normalizePageDataValidationObservation(observation);
    try {
      const result = await recordValidationWithRetry(engine, runId, normalized);
      return {
        observed: true,
        engine: WORKFLOW_ENGINE.LANGGRAPH,
        mode: selection.mode,
        configVersion: selection.configVersion,
        validation: result.validation ?? normalized,
        shadowRun: result,
      };
    } catch (error) {
      logger.warn('[orchestrator-shadow] validation observation failed:', safeError(error));
      throw error;
    }
  };

  return observeWorkflowRun;
}

export const workflowShadowObserverInternals = {
  recordValidationWithRetry,
  safeError,
};