import { createRemoteWorkflowEngine } from './engine-registry.mjs'; import { WORKFLOW_ENGINE } from './contracts.mjs'; const SHADOW_EVENT_TYPES = Object.freeze([ 'workflow_shadow_completed', 'workflow_shadow_failed', 'workflow_shadow_skipped', ]); const EXECUTION_PLAN_EVENT_TYPE = 'workflow_execution_planned'; const MAX_METRIC_EVENTS = 5000; const CANARY_READINESS_THRESHOLDS = Object.freeze({ hours: 24 * 7, minObservations: 20, minSuccessRate: 0.95, maxSkipRate: 0, maxP95LatencyMs: 2000, minLatencyCoverageRate: 0.95, minNativeSettledRate: 0.95, minDistinctSessions: 5, maxHoursSinceLastObservation: 24, }); const NATIVE_TERMINAL_STATUSES = new Set(['succeeded', 'failed']); const SYNTHETIC_TASK_TYPES = new Set(['orchestrator_shadow_smoke']); function clampInteger(value, fallback, min, max) { const parsed = Number(value); if (!Number.isFinite(parsed)) return fallback; return Math.min(max, Math.max(min, Math.floor(parsed))); } function parseJsonColumn(value, fallback = null) { if (value == null || value === '') return fallback; if (typeof value === 'object') return value; try { return JSON.parse(String(value)); } catch { return fallback; } } function percentile(values, ratio) { if (!values.length) return null; const sorted = [...values].sort((left, right) => left - right); const index = Math.max(0, Math.ceil(sorted.length * ratio) - 1); return sorted[index]; } function normalizeRunId(value) { const runId = String(value ?? '').trim(); return /^[a-zA-Z0-9_-]{1,128}$/.test(runId) ? runId : ''; } function normalizeExecutorJobId(value) { const jobId = String(value ?? '').trim(); return /^[a-zA-Z0-9:_-]{1,128}$/.test(jobId) ? jobId : ''; } function projectShadowEventRow(row) { const data = parseJsonColumn(row.data_json, {}) ?? {}; const succeeded = row.event_type === 'workflow_shadow_completed'; const skipped = row.event_type === 'workflow_shadow_skipped'; const latency = Number(data.latencyMs); const taskType = data.taskType ?? null; return { eventId: row.event_id, runId: row.run_id, requestId: row.request_id, userId: row.user_id, sessionId: row.agent_session_id ?? null, nativeStatus: row.native_status, nativeAttempts: Number(row.native_attempts ?? 0), shadowStatus: succeeded ? 'succeeded' : skipped ? 'skipped' : 'failed', engine: data.engine ?? 'langgraph', configVersion: data.configVersion == null ? null : Number(data.configVersion), phase: data.phase ?? null, taskType, synthetic: data.synthetic === true || SYNTHETIC_TASK_TYPES.has(String(taskType ?? '')), executorAdapter: data.executorAdapter ?? null, latencyMs: Number.isFinite(latency) && latency >= 0 ? latency : null, error: succeeded || skipped ? null : { code: data.code ?? 'WORKFLOW_SHADOW_FAILED', message: data.message ?? 'Shadow observation failed', }, skipReason: skipped ? data.reason ?? 'shadow_queue_full' : null, observedAt: Number(row.event_created_at ?? 0), nativeCompletedAt: row.native_completed_at == null ? null : Number(row.native_completed_at), }; } function projectUniqueShadowRuns(rows) { const seen = new Set(); const runs = []; for (const row of rows) { const run = projectShadowEventRow(row); if (seen.has(run.runId)) continue; seen.add(run.runId); runs.push(run); } return runs; } function projectExecutionPlanEventRow(row) { const data = parseJsonColumn(row.data_json, {}) ?? {}; const bucket = Number(data.bucket); return { eventId: row.event_id, runId: row.run_id, requestId: row.request_id, userId: row.user_id, sessionId: row.agent_session_id ?? null, nativeStatus: row.native_status, nativeAttempts: Number(row.native_attempts ?? 0), mode: data.mode ?? null, candidateEngine: data.candidateEngine ?? 'native', effectiveEngine: data.effectiveEngine ?? 'native', fallbackEngine: data.fallbackEngine ?? 'native', reason: data.reason ?? null, candidateReason: data.candidateReason ?? null, configVersion: data.configVersion == null ? null : Number(data.configVersion), bucket: Number.isFinite(bucket) ? bucket : null, taskType: data.taskType ?? null, dryRun: data.dryRun === true, handoffAllowed: data.handoffAllowed === true, plannedAt: Number(row.event_created_at ?? 0), nativeCompletedAt: row.native_completed_at == null ? null : Number(row.native_completed_at), }; } function projectUniqueExecutionPlans(rows) { const seen = new Set(); const plans = []; for (const row of rows) { const plan = projectExecutionPlanEventRow(row); if (seen.has(plan.runId)) continue; seen.add(plan.runId); plans.push(plan); } return plans; } function ratio(numerator, denominator) { return denominator > 0 ? numerator / denominator : null; } function buildCanaryReadiness(loaded, runtimeState, now) { const thresholds = CANARY_READINESS_THRESHOLDS; const eligibleRuns = loaded.runs.filter((run) => !run.synthetic); const successes = eligibleRuns.filter((run) => run.shadowStatus === 'succeeded').length; const failures = eligibleRuns.filter((run) => run.shadowStatus === 'failed').length; const skipped = eligibleRuns.filter((run) => run.shadowStatus === 'skipped').length; const latencies = eligibleRuns .map((run) => run.latencyMs) .filter((value) => Number.isFinite(value)); const nativeSettled = eligibleRuns.filter((run) => NATIVE_TERMINAL_STATUSES.has(run.nativeStatus)).length; const distinctSessions = new Set( eligibleRuns.map((run) => run.sessionId).filter(Boolean), ).size; const lastObservedAt = eligibleRuns[0]?.observedAt ?? null; const hoursSinceLastObservation = lastObservedAt == null ? null : Math.max(0, (now - lastObservedAt) / (60 * 60 * 1000)); const successRate = ratio(successes, eligibleRuns.length); const skipRate = ratio(skipped, eligibleRuns.length); const latencyCoverageRate = ratio(latencies.length, eligibleRuns.length); const nativeSettledRate = ratio(nativeSettled, eligibleRuns.length); const latencyP95Ms = percentile(latencies, 0.95); const serviceHealth = runtimeState?.serviceHealth ?? null; const checkpoint = serviceHealth?.details?.checkpoint ?? null; const executorStore = serviceHealth?.details?.executorGateway?.store ?? null; const checks = [ { id: 'shadow_mode', passed: runtimeState?.config?.mode === 'shadow', actual: runtimeState?.config?.mode ?? null, target: 'shadow', }, { id: 'shadow_wiring_enabled', passed: runtimeState?.runtime?.shadowObservation?.enabled === true, actual: runtimeState?.runtime?.shadowObservation?.enabled ?? null, target: true, }, { id: 'service_healthy', passed: serviceHealth?.ok === true, actual: serviceHealth?.status ?? null, target: 'healthy', }, { id: 'durable_checkpoint', passed: checkpoint?.durable === true, actual: checkpoint?.durable ?? null, target: true, }, { id: 'durable_executor_job_store', passed: executorStore?.durable === true, actual: executorStore?.durable ?? null, target: true, }, { id: 'observe_only', passed: serviceHealth?.details?.execution === 'observe-only', actual: serviceHealth?.details?.execution ?? null, target: 'observe-only', }, { id: 'sample_volume', passed: eligibleRuns.length >= thresholds.minObservations, actual: eligibleRuns.length, target: thresholds.minObservations, }, { id: 'shadow_success_rate', passed: successRate != null && successRate >= thresholds.minSuccessRate, actual: successRate, target: thresholds.minSuccessRate, }, { id: 'shadow_skip_rate', passed: skipRate != null && skipRate <= thresholds.maxSkipRate, actual: skipRate, target: thresholds.maxSkipRate, }, { id: 'latency_coverage', passed: latencyCoverageRate != null && latencyCoverageRate >= thresholds.minLatencyCoverageRate, actual: latencyCoverageRate, target: thresholds.minLatencyCoverageRate, }, { id: 'latency_p95', passed: latencyP95Ms != null && latencyP95Ms <= thresholds.maxP95LatencyMs, actual: latencyP95Ms, target: thresholds.maxP95LatencyMs, }, { id: 'native_settled_rate', passed: nativeSettledRate != null && nativeSettledRate >= thresholds.minNativeSettledRate, actual: nativeSettledRate, target: thresholds.minNativeSettledRate, }, { id: 'session_coverage', passed: distinctSessions >= thresholds.minDistinctSessions, actual: distinctSessions, target: thresholds.minDistinctSessions, }, { id: 'sample_freshness', passed: hoursSinceLastObservation != null && hoursSinceLastObservation <= thresholds.maxHoursSinceLastObservation, actual: hoursSinceLastObservation, target: thresholds.maxHoursSinceLastObservation, }, { id: 'complete_window', passed: !loaded.capped, actual: loaded.capped ? 'sampled' : 'complete', target: 'complete', }, ]; const failureCounts = new Map(); for (const run of eligibleRuns) { if (!run.error?.code) continue; failureCounts.set(run.error.code, (failureCounts.get(run.error.code) ?? 0) + 1); } return { generatedAt: now, ready: checks.every((check) => check.passed), recommendation: checks.every((check) => check.passed) ? 'manual_canary_review' : 'keep_shadow', window: { hours: loaded.hours, from: loaded.from, }, thresholds, samples: { totalObservations: loaded.runs.length, eligibleObservations: eligibleRuns.length, excludedSynthetic: loaded.runs.length - eligibleRuns.length, successes, failures, skipped, successRate, skipRate, latencyCoverageRate, latencyP95Ms, nativeSettledRate, distinctSessions, lastObservedAt, hoursSinceLastObservation, sampled: loaded.capped, }, service: { status: serviceHealth?.status ?? null, latencyMs: serviceHealth?.latencyMs ?? null, checkpointKind: checkpoint?.kind ?? null, checkpointDurable: checkpoint?.durable ?? null, executorJobStoreKind: executorStore?.kind ?? null, executorJobStoreDurable: executorStore?.durable ?? null, execution: serviceHealth?.details?.execution ?? null, }, checks, blockers: checks.filter((check) => !check.passed).map((check) => check.id), failureCodes: [...failureCounts.entries()] .map(([code, count]) => ({ code, count })) .sort((left, right) => right.count - left.count || left.code.localeCompare(right.code)) .slice(0, 10), }; } function summarizeRuns(runs, { capped = false } = {}) { const successes = runs.filter((run) => run.shadowStatus === 'succeeded').length; const failures = runs.filter((run) => run.shadowStatus === 'failed').length; const skipped = runs.filter((run) => run.shadowStatus === 'skipped').length; const latencies = runs .map((run) => run.latencyMs) .filter((value) => Number.isFinite(value)); return { observations: runs.length, successes, failures, skipped, successRate: runs.length ? successes / runs.length : null, failureRate: runs.length ? failures / runs.length : null, skipRate: runs.length ? skipped / runs.length : null, latencyP50Ms: percentile(latencies, 0.5), latencyP95Ms: percentile(latencies, 0.95), nativeSucceeded: runs.filter((run) => run.nativeStatus === 'succeeded').length, nativeFailed: runs.filter((run) => run.nativeStatus === 'failed').length, lastObservedAt: runs[0]?.observedAt ?? null, sampled: capped, }; } function countValues(values) { const counts = new Map(); for (const value of values) { const normalized = String(value ?? '').trim() || 'unknown'; counts.set(normalized, (counts.get(normalized) ?? 0) + 1); } return [...counts.entries()] .map(([value, count]) => ({ value, count })) .sort((left, right) => right.count - left.count || left.value.localeCompare(right.value)); } function summarizeExecutionPlans(plans, { capped = false } = {}) { const candidateSelections = plans.filter( (plan) => plan.candidateEngine !== WORKFLOW_ENGINE.NATIVE, ).length; const nativeSettled = plans.filter((plan) => NATIVE_TERMINAL_STATUSES.has(plan.nativeStatus)).length; return { decisions: plans.length, candidateSelections, candidateSelectionRate: ratio(candidateSelections, plans.length), nativeSelections: plans.length - candidateSelections, nativeSucceeded: plans.filter((plan) => plan.nativeStatus === 'succeeded').length, nativeFailed: plans.filter((plan) => plan.nativeStatus === 'failed').length, nativeSettledRate: ratio(nativeSettled, plans.length), handoffAllowed: plans.filter((plan) => plan.handoffAllowed).length, distinctSessions: new Set(plans.map((plan) => plan.sessionId).filter(Boolean)).size, lastPlannedAt: plans[0]?.plannedAt ?? null, candidateReasons: countValues(plans.map((plan) => plan.candidateReason ?? plan.reason)), taskTypes: countValues(plans.map((plan) => plan.taskType)), sampled: capped, }; } function safeRemoteError(error) { return { code: String(error?.code ?? 'ORCHESTRATOR_UNAVAILABLE').slice(0, 128), message: String(error instanceof Error ? error.message : error).slice(0, 1000), }; } export function createOrchestratorObservabilityService({ pool, configService, serviceToken = process.env.MEMIND_ORCHESTRATOR_SERVICE_TOKEN, fetchImpl = globalThis.fetch, nowMs = Date.now, } = {}) { if (!pool?.query) throw new Error('Orchestrator observability requires a database pool'); if (!configService?.getRuntimeState) { throw new Error('Orchestrator observability requires config service'); } async function loadShadowEvents({ hours = 24 } = {}) { const normalizedHours = clampInteger(hours, 24, 1, 24 * 30); const from = nowMs() - normalizedHours * 60 * 60 * 1000; const [rows] = await pool.query( `SELECT e.id AS event_id, e.run_id, e.event_type, e.data_json, e.created_at AS event_created_at, r.request_id, r.user_id, r.agent_session_id, r.status AS native_status, r.attempts AS native_attempts, r.completed_at AS native_completed_at FROM h5_agent_run_events e INNER JOIN h5_agent_runs r ON r.id = e.run_id WHERE e.event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')}) AND e.created_at >= ? ORDER BY e.created_at DESC, e.id DESC LIMIT ?`, [...SHADOW_EVENT_TYPES, from, MAX_METRIC_EVENTS], ); return { from, hours: normalizedHours, capped: rows.length >= MAX_METRIC_EVENTS, runs: projectUniqueShadowRuns(rows), }; } async function loadExecutionPlanEvents({ hours = 24 } = {}) { const normalizedHours = clampInteger(hours, 24, 1, 24 * 30); const from = nowMs() - normalizedHours * 60 * 60 * 1000; const [rows] = await pool.query( `SELECT e.id AS event_id, e.run_id, e.data_json, e.created_at AS event_created_at, r.request_id, r.user_id, r.agent_session_id, r.status AS native_status, r.attempts AS native_attempts, r.completed_at AS native_completed_at FROM h5_agent_run_events e INNER JOIN h5_agent_runs r ON r.id = e.run_id WHERE e.event_type = ? AND e.created_at >= ? ORDER BY e.created_at DESC, e.id DESC LIMIT ?`, [EXECUTION_PLAN_EVENT_TYPE, from, MAX_METRIC_EVENTS], ); return { from, hours: normalizedHours, capped: rows.length >= MAX_METRIC_EVENTS, plans: projectUniqueExecutionPlans(rows), }; } return { async getCanaryReadiness() { const [loaded, runtimeState] = await Promise.all([ loadShadowEvents({ hours: CANARY_READINESS_THRESHOLDS.hours }), configService.getRuntimeState({ probe: true }), ]); return buildCanaryReadiness(loaded, runtimeState, nowMs()); }, async listShadowRuns({ hours = 24, limit = 50, status = 'all', } = {}) { const loaded = await loadShadowEvents({ hours }); const normalizedLimit = clampInteger(limit, 50, 1, 200); const normalizedStatus = ['succeeded', 'failed', 'skipped'].includes(status) ? status : 'all'; const filtered = normalizedStatus === 'all' ? loaded.runs : loaded.runs.filter((run) => run.shadowStatus === normalizedStatus); return { generatedAt: nowMs(), window: { hours: loaded.hours, from: loaded.from, }, metrics: summarizeRuns(loaded.runs, { capped: loaded.capped }), runs: filtered.slice(0, normalizedLimit), }; }, async listExecutionPlans({ hours = 24, limit = 50, selection = 'all', } = {}) { const loaded = await loadExecutionPlanEvents({ hours }); const normalizedLimit = clampInteger(limit, 50, 1, 200); const normalizedSelection = ['candidate', 'native'].includes(selection) ? selection : 'all'; const filtered = normalizedSelection === 'all' ? loaded.plans : loaded.plans.filter((plan) => ( normalizedSelection === 'candidate' ? plan.candidateEngine !== WORKFLOW_ENGINE.NATIVE : plan.candidateEngine === WORKFLOW_ENGINE.NATIVE )); return { generatedAt: nowMs(), window: { hours: loaded.hours, from: loaded.from, }, metrics: summarizeExecutionPlans(loaded.plans, { capped: loaded.capped }), plans: filtered.slice(0, normalizedLimit), }; }, async getShadowRun(runId) { const normalizedRunId = normalizeRunId(runId); if (!normalizedRunId) return null; const [runRows] = await pool.query( `SELECT id, request_id, user_id, agent_session_id, status, attempts, error_message, created_at, updated_at, started_at, completed_at FROM h5_agent_runs WHERE id = ? LIMIT 1`, [normalizedRunId], ); if (!runRows[0]) return null; const [eventRows] = await pool.query( `SELECT id AS event_id, event_type, data_json, created_at FROM h5_agent_run_events WHERE run_id = ? AND event_type IN (${SHADOW_EVENT_TYPES.map(() => '?').join(', ')}) ORDER BY created_at ASC, id ASC`, [normalizedRunId, ...SHADOW_EVENT_TYPES], ); const native = runRows[0]; const localEvents = eventRows.map((row) => ({ eventId: row.event_id, type: row.event_type, data: parseJsonColumn(row.data_json, null), createdAt: Number(row.created_at), })); let remote = { available: false, state: null, events: [], error: null, executorJob: { available: false, state: null, events: [], error: null, }, }; try { const runtimeState = await configService.getRuntimeState(); if (runtimeState.config.serviceUrl) { const engine = createRemoteWorkflowEngine({ baseUrl: runtimeState.config.serviceUrl, serviceToken, timeoutMs: runtimeState.config.requestTimeoutMs, fetchImpl, }); const [state, events] = await Promise.all([ engine.getState(normalizedRunId), (async () => { const collected = []; for await (const event of engine.streamEvents(normalizedRunId)) { collected.push(event); if (collected.length >= 500) break; } return collected; })(), ]); const executorJobId = normalizeExecutorJobId(state?.plan?.executorJob?.id); let executorJob = { available: false, state: null, events: [], error: null, }; if (executorJobId) { try { const [jobState, jobEvents] = await Promise.all([ engine.getExecutorJob(executorJobId), (async () => { const collected = []; for await (const event of engine.streamExecutorJobEvents(executorJobId)) { collected.push(event); if (collected.length >= 500) break; } return collected; })(), ]); executorJob = { available: true, state: jobState, events: jobEvents, error: null, }; } catch (error) { executorJob.error = safeRemoteError(error); } } remote = { available: true, state, events, error: null, executorJob, }; } } catch (error) { remote.error = safeRemoteError(error); } return { native: { runId: native.id, requestId: native.request_id, userId: native.user_id, sessionId: native.agent_session_id ?? null, status: native.status, attempts: Number(native.attempts ?? 0), error: native.error_message ?? null, createdAt: Number(native.created_at ?? 0), updatedAt: Number(native.updated_at ?? 0), startedAt: native.started_at == null ? null : Number(native.started_at), completedAt: native.completed_at == null ? null : Number(native.completed_at), }, localEvents, remote, }; }, }; } export const orchestratorObservabilityInternals = { CANARY_READINESS_THRESHOLDS, EXECUTION_PLAN_EVENT_TYPE, MAX_METRIC_EVENTS, SHADOW_EVENT_TYPES, buildCanaryReadiness, normalizeExecutorJobId, normalizeRunId, parseJsonColumn, percentile, projectShadowEventRow, projectUniqueShadowRuns, projectExecutionPlanEventRow, projectUniqueExecutionPlans, summarizeExecutionPlans, summarizeRuns, };