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