diff --git a/admin-routes.mjs b/admin-routes.mjs index 66c7427..3811c45 100644 --- a/admin-routes.mjs +++ b/admin-routes.mjs @@ -268,6 +268,17 @@ export function createAdminApi({ })); }); + adminApi.get('/orchestrator/execution-plans', requireAdmin, async (req, res) => { + if (!orchestratorObservabilityService?.listExecutionPlans) { + return res.status(503).json({ message: 'Orchestrator Dry-run 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.listExecutionPlans({ + hours: req.query.hours, + limit: req.query.limit, + selection: req.query.selection, + })); + }); + adminApi.get('/orchestrator/canary-readiness', requireAdmin, async (_req, res) => { if (!orchestratorObservabilityService?.getCanaryReadiness) { return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); diff --git a/admin-routes.test.mjs b/admin-routes.test.mjs index a467fd9..bd84f0f 100644 --- a/admin-routes.test.mjs +++ b/admin-routes.test.mjs @@ -139,6 +139,17 @@ test('admin orchestrator routes expose a versioned plug-in control plane', async }, }, orchestratorObservabilityService: { + async listExecutionPlans(options) { + assert.deepEqual(options, { + hours: '24', + limit: '50', + selection: 'candidate', + }); + return { + metrics: { decisions: 3, candidateSelections: 2 }, + plans: [{ runId: 'run-plan-1', candidateEngine: 'langgraph' }], + }; + }, async getCanaryReadiness() { return { ready: false, @@ -206,6 +217,15 @@ test('admin orchestrator routes expose a versioned plug-in control plane', async assert.equal(shadowRunsResponse.status, 200); assert.equal((await shadowRunsResponse.json()).metrics.failures, 1); + const executionPlansResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/execution-plans?hours=24&limit=50&selection=candidate`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(executionPlansResponse.status, 200); + const executionPlansBody = await executionPlansResponse.json(); + assert.equal(executionPlansBody.metrics.decisions, 3); + assert.equal(executionPlansBody.plans[0].candidateEngine, 'langgraph'); + const readinessResponse = await fetch( `${server.baseUrl}/admin-api/orchestrator/canary-readiness`, { headers: { cookie: 'h5_user_session=token-admin' } }, diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index c6937a1..103f91e 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -339,6 +339,30 @@ export function createAgentRunGateway({ ); } + async function appendExecutionPlanEvent(input, source) { + const executionPlan = source?.executionPlan; + if (!executionPlan?.dryRun) return; + await appendEvent(input.runId, 'workflow_execution_planned', { + version: executionPlan.version ?? 'workflow-execution-decision-v1', + mode: source.mode ?? null, + candidateEngine: executionPlan.candidateEngine ?? null, + effectiveEngine: executionPlan.effectiveEngine ?? 'native', + fallbackEngine: executionPlan.fallbackEngine ?? 'native', + reason: executionPlan.reason ?? 'execution_gate_disabled', + candidateReason: executionPlan.candidateReason ?? null, + configVersion: executionPlan.configVersion ?? null, + bucket: executionPlan.bucket ?? null, + taskType: executionPlan.taskType ?? input.taskType ?? null, + dryRun: true, + handoffAllowed: false, + }).catch((error) => { + console.warn( + '[AgentRun] workflow execution plan event skipped:', + error instanceof Error ? error.message : error, + ); + }); + } + function dispatchShadowObservation(input) { if (typeof observeWorkflowRun !== 'function' || input.toolMode !== 'code') return; setImmediate(() => { @@ -346,26 +370,7 @@ export function createAgentRunGateway({ void Promise.resolve() .then(() => observeWorkflowRun(input)) .then(async (result) => { - if (result?.executionPlan?.dryRun) { - await appendEvent(input.runId, 'workflow_execution_planned', { - version: result.executionPlan.version ?? 'workflow-execution-decision-v1', - mode: result.mode ?? null, - candidateEngine: result.executionPlan.candidateEngine ?? null, - effectiveEngine: result.executionPlan.effectiveEngine ?? 'native', - fallbackEngine: result.executionPlan.fallbackEngine ?? 'native', - reason: result.executionPlan.reason ?? 'execution_gate_disabled', - candidateReason: result.executionPlan.candidateReason ?? null, - configVersion: result.executionPlan.configVersion ?? null, - bucket: result.executionPlan.bucket ?? null, - dryRun: true, - handoffAllowed: false, - }).catch((error) => { - console.warn( - '[AgentRun] workflow execution plan event skipped:', - error instanceof Error ? error.message : error, - ); - }); - } + await appendExecutionPlanEvent(input, result); if (!result?.observed) return; await appendEvent(input.runId, 'workflow_shadow_completed', { engine: result.engine ?? 'langgraph', @@ -379,6 +384,7 @@ export function createAgentRunGateway({ }); }) .catch(async (error) => { + await appendExecutionPlanEvent(input, error); console.warn( '[AgentRun] workflow shadow observation failed:', error instanceof Error ? error.message : error, diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 024b130..094ea4b 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -443,6 +443,20 @@ test('shadow observer failure cannot fail a Native run and chat runs are not obs observations += 1; const error = new Error('shadow unavailable'); error.code = 'SHADOW_UNAVAILABLE'; + error.mode = 'shadow'; + error.executionPlan = { + version: 'workflow-execution-decision-v1', + candidateEngine: 'native', + effectiveEngine: 'native', + fallbackEngine: 'native', + reason: 'shadow', + candidateReason: 'canary_not_selected', + configVersion: 10, + bucket: 91, + taskType: 'code_task', + dryRun: true, + handoffAllowed: false, + }; throw error; }, }); @@ -466,6 +480,8 @@ test('shadow observer failure cannot fail a Native run and chat runs are not obs assert.equal(observations, 1); const failure = pool.events.find((event) => event.eventType === 'workflow_shadow_failed'); assert.equal(JSON.parse(failure.dataJson).code, 'SHADOW_UNAVAILABLE'); + const plan = pool.events.find((event) => event.eventType === 'workflow_execution_planned'); + assert.equal(JSON.parse(plan.dataJson).candidateReason, 'canary_not_selected'); }); test('canary dry-run records the candidate decision without changing Native run state', async () => { @@ -487,6 +503,7 @@ test('canary dry-run records the candidate decision without changing Native run candidateReason: 'percentage', configVersion: 9, bucket: 3, + taskType: 'repo_refactor', dryRun: true, handoffAllowed: false, }, @@ -520,6 +537,7 @@ test('canary dry-run records the candidate decision without changing Native run candidateReason: 'percentage', configVersion: 9, bucket: 3, + taskType: 'repo_refactor', dryRun: true, handoffAllowed: false, }); diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index a8bda9a..29f36b9 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -68,6 +68,16 @@ cancellation, and fallback controls, but its non-Native dispatch path is hard-disabled in code. Changing memindadm configuration or setting an environment override cannot transfer execution ownership. +The Dry-run projection records every code-run decision while the configured +mode is Shadow, Canary, or Active, including `canary_not_selected`. In Shadow, +the same allowlist and deterministic percentage are evaluated in parallel with +the remote observation, so collecting routing evidence does not interrupt +Shadow readiness samples. This prevents a selected-only dataset from reporting +a meaningless 100% candidate rate. The Memind control plane joins each decision +to the Native run terminal state and aggregates candidate rate, selection +reasons, task types, session coverage, and Native terminal coverage without +querying LangGraph. + ## Protocol The framework-neutral contracts are: diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index 878d42e..ade9106 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -306,6 +306,52 @@ export type OrchestratorShadowRunList = { runs: OrchestratorShadowRun[]; }; +export type OrchestratorExecutionPlan = { + eventId: string; + runId: string; + requestId: string; + userId: string; + sessionId: string | null; + nativeStatus: string; + nativeAttempts: number; + mode: string | null; + candidateEngine: string; + effectiveEngine: string; + fallbackEngine: string; + reason: string | null; + candidateReason: string | null; + configVersion: number | null; + bucket: number | null; + taskType: string | null; + dryRun: boolean; + handoffAllowed: boolean; + plannedAt: number; + nativeCompletedAt: number | null; +}; + +export type OrchestratorExecutionPlanMetrics = { + decisions: number; + candidateSelections: number; + candidateSelectionRate: number | null; + nativeSelections: number; + nativeSucceeded: number; + nativeFailed: number; + nativeSettledRate: number | null; + handoffAllowed: number; + distinctSessions: number; + lastPlannedAt: number | null; + candidateReasons: Array<{ value: string; count: number }>; + taskTypes: Array<{ value: string; count: number }>; + sampled: boolean; +}; + +export type OrchestratorExecutionPlanList = { + generatedAt: number; + window: { hours: number; from: number }; + metrics: OrchestratorExecutionPlanMetrics; + plans: OrchestratorExecutionPlan[]; +}; + export type OrchestratorCanaryReadinessCheck = { id: string; passed: boolean; @@ -443,6 +489,20 @@ export async function fetchOrchestratorShadowRuns(params: { ); } +export async function fetchOrchestratorExecutionPlans(params: { + hours?: number; + limit?: number; + selection?: 'all' | 'candidate' | 'native'; +} = {}) { + const query = new URLSearchParams(); + if (params.hours) query.set('hours', String(params.hours)); + if (params.limit) query.set('limit', String(params.limit)); + if (params.selection) query.set('selection', params.selection); + return adminFetch( + `/admin-api/orchestrator/execution-plans?${query}`, + ); +} + export async function fetchOrchestratorCanaryReadiness() { return adminFetch( '/admin-api/orchestrator/canary-readiness', diff --git a/ops/src/pages/admin/OrchestratorExecutionPlanPanel.tsx b/ops/src/pages/admin/OrchestratorExecutionPlanPanel.tsx new file mode 100644 index 0000000..8117870 --- /dev/null +++ b/ops/src/pages/admin/OrchestratorExecutionPlanPanel.tsx @@ -0,0 +1,205 @@ +import { useEffect, useState } from 'react'; +import { + fetchOrchestratorExecutionPlans, + type OrchestratorExecutionPlanList, +} from '../../api/admin'; + +function formatTime(value: number | null | undefined) { + return value ? new Date(value).toLocaleString('zh-CN', { hour12: false }) : '—'; +} + +function formatPercent(value: number | null) { + return value == null ? '—' : `${(value * 100).toFixed(1)}%`; +} + +function shortId(value: string) { + return value.length > 16 ? `${value.slice(0, 8)}…${value.slice(-6)}` : value; +} + +const metricCardStyle = { + border: '1px solid #e4e9e6', + borderRadius: 10, + padding: 12, + minWidth: 140, +} as const; + +export function OrchestratorExecutionPlanPanel() { + const [hours, setHours] = useState(24); + const [selection, setSelection] = useState<'all' | 'candidate' | 'native'>('all'); + const [data, setData] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + + const load = async () => { + setLoading(true); + setError(null); + try { + setData(await fetchOrchestratorExecutionPlans({ + hours, + selection, + limit: 100, + })); + } catch (err) { + setError(err instanceof Error ? err.message : 'Dry-run 决策加载失败'); + } finally { + setLoading(false); + } + }; + + useEffect(() => { + void load(); + }, [hours, selection]); + + const metrics = data?.metrics; + + return ( +
+
+
+

执行路由 Dry-run

+

+ Shadow 同时模拟 Canary 候选路由；所有记录的实际执行引擎仍必须是 Native。 +

+
+
+ + + +
+
+ + {error ?

{error}

: null} + {(metrics?.handoffAllowed ?? 0) > 0 ? ( +

+ 检测到 {metrics?.handoffAllowed} 条 handoffAllowed=true 记录，请立即检查执行闸门。 +

+ ) : null} + +
+
+ 路由决策 +
{metrics?.decisions ?? '—'}
+
+
+ 候选命中率 +
+ {formatPercent(metrics?.candidateSelectionRate ?? null)} +
+
+
+ LangGraph / Native 候选 +
+ {metrics ? `${metrics.candidateSelections} / ${metrics.nativeSelections}` : '—'} +
+
+
+ Native 终态覆盖率 +
+ {formatPercent(metrics?.nativeSettledRate ?? null)} +
+
+
+ Native 成功 / 失败 +
+ {metrics ? `${metrics.nativeSucceeded} / ${metrics.nativeFailed}` : '—'} +
+
+
+ 真实会话 +
{metrics?.distinctSessions ?? '—'}
+
+
+ + {metrics ? ( +
+
+ 选择原因 +

+ {metrics.candidateReasons.length + ? metrics.candidateReasons.map((item) => `${item.value} × ${item.count}`).join('，') + : '暂无数据'} +

+
+
+ 任务类型 +

+ {metrics.taskTypes.length + ? metrics.taskTypes.map((item) => `${item.value} × ${item.count}`).join('，') + : '暂无数据'} +

+
+
+ ) : null} + + {metrics?.sampled ? ( +

当前指标基于最近 5000 条 Dry-run 决策采样。

+ ) : null} + +
+ + + + + + + + + + + + + + {(data?.plans ?? []).map((plan) => ( + + + + + + + + + + ))} + {!loading && !data?.plans.length ? ( + + + + ) : null} + +
时间Run模式候选 → 实际原因 / Bucket任务类型Native
{formatTime(plan.plannedAt)}{shortId(plan.runId)}{plan.mode ?? '—'} + + {plan.candidateEngine} + + {' → '} + {plan.effectiveEngine} + + {plan.candidateReason ?? plan.reason ?? '—'} + + bucket {plan.bucket ?? '—'} · config v{plan.configVersion ?? '—'} + + {plan.taskType ?? '—'} + {plan.nativeStatus} + + attempts {plan.nativeAttempts} + +
+ 当前时间范围内没有 Dry-run 路由决策。 +
+
+
+ ); +} diff --git a/ops/src/pages/admin/OrchestratorPage.tsx b/ops/src/pages/admin/OrchestratorPage.tsx index e039749..352546a 100644 --- a/ops/src/pages/admin/OrchestratorPage.tsx +++ b/ops/src/pages/admin/OrchestratorPage.tsx @@ -7,6 +7,7 @@ import { type OrchestratorConfigState, type OrchestratorServiceHealth, } from '../../api/admin'; +import { OrchestratorExecutionPlanPanel } from './OrchestratorExecutionPlanPanel'; import { OrchestratorShadowPanel } from './OrchestratorShadowPanel'; function listToText(values: string[]) { @@ -170,6 +171,8 @@ export function OrchestratorPage() { + +

路由模式

diff --git a/services/orchestrator/README.md b/services/orchestrator/README.md index 47353ad..3b5447a 100644 --- a/services/orchestrator/README.md +++ b/services/orchestrator/README.md @@ -85,6 +85,14 @@ Phase 3 adds the execution handoff boundary without enabling it: - the implementation capability is hard-disabled in code, so memindadm configuration cannot accidentally unlock execution. +Phase 3.1 adds a Memind-owned Dry-run projection. Shadow mode continues its +LangGraph observation while also evaluating the configured Canary allowlist and +percentage. It records both selected and non-selected decisions so the +candidate selection rate has a complete denominator. memindadm shows candidate +reasons, task-type distribution, Native terminal coverage, recent decisions, +and an explicit alert if any stored decision ever reports +`handoffAllowed=true`. + The default memindadm mode remains `off`. Canary and Active routing are not wired to the LangGraph executor in Phase 3. Native Agent Run remains the sole executor in every mode. @@ -129,6 +137,7 @@ memindadm exposes the projection through: GET /admin-api/orchestrator/shadow-runs GET /admin-api/orchestrator/shadow-runs/:runId GET /admin-api/orchestrator/canary-readiness +GET /admin-api/orchestrator/execution-plans ``` ## Colima deployment diff --git a/services/orchestrator/admin-config.mjs b/services/orchestrator/admin-config.mjs index ca560b4..3e13fda 100644 --- a/services/orchestrator/admin-config.mjs +++ b/services/orchestrator/admin-config.mjs @@ -363,11 +363,27 @@ export function createOrchestratorAdminConfigService(pool, { if (!runtime.effective || !workflowMatched) { return { ...base, reason: runtime.reason ?? 'workflow_not_enabled' }; } + const rolloutKey = runId || requestId || `${userId ?? ''}:${normalizedWorkflow}`; + const bucket = stableRolloutBucket(rolloutKey); + const explicitlyAllowed = Boolean(userId && config.userAllowlist.includes(String(userId))); + const percentAllowed = bucket < config.rolloutPercent; + const rolloutSelected = explicitlyAllowed || percentAllowed; + const rolloutReason = explicitlyAllowed + ? 'user_allowlist' + : percentAllowed + ? 'percentage' + : 'canary_not_selected'; if (config.mode === ORCHESTRATOR_MODE.SHADOW) { return { ...base, shadowEngine: config.primaryEngine, reason: 'shadow', + candidateEngine: rolloutSelected + ? config.primaryEngine + : WORKFLOW_ENGINE.NATIVE, + candidateReason: rolloutReason, + bucket, + dryRun: true, }; } if (config.mode === ORCHESTRATOR_MODE.ACTIVE) { @@ -384,25 +400,20 @@ export function createOrchestratorAdminConfigService(pool, { : { ...selected, reason: 'execution_gate_disabled', dryRun: true }; } if (config.mode === ORCHESTRATOR_MODE.CANARY) { - const explicitlyAllowed = Boolean(userId && config.userAllowlist.includes(String(userId))); - const rolloutKey = runId || requestId || `${userId ?? ''}:${normalizedWorkflow}`; - const bucket = stableRolloutBucket(rolloutKey); - const percentAllowed = bucket < config.rolloutPercent; - if (!explicitlyAllowed && !percentAllowed) { + if (!rolloutSelected) { return { ...base, reason: 'canary_not_selected', bucket }; } - const candidateReason = explicitlyAllowed ? 'user_allowlist' : 'percentage'; const selected = { ...base, candidateEngine: config.primaryEngine, - candidateReason, + candidateReason: rolloutReason, bucket, }; if (config.primaryEngine === WORKFLOW_ENGINE.NATIVE) { - return { ...selected, reason: candidateReason }; + return { ...selected, reason: rolloutReason }; } return runtime.executionHandoff.enabled - ? { ...selected, engine: config.primaryEngine, reason: candidateReason } + ? { ...selected, engine: config.primaryEngine, reason: rolloutReason } : { ...selected, reason: 'execution_gate_disabled', dryRun: true }; } return base; diff --git a/services/orchestrator/admin-config.test.mjs b/services/orchestrator/admin-config.test.mjs index f956bc1..610ee2c 100644 --- a/services/orchestrator/admin-config.test.mjs +++ b/services/orchestrator/admin-config.test.mjs @@ -129,6 +129,38 @@ test('orchestrator keeps a Native primary as a non-dry-run Native selection', as assert.equal(selected.dryRun, false); }); +test('Shadow mode evaluates the Canary rollout while Native remains effective', async () => { + const service = createOrchestratorAdminConfigService(createPool(), { env: {} }); + await service.updateAdminConfig({ + mode: 'shadow', + serviceUrl: 'http://127.0.0.1:8093', + rolloutPercent: 0, + userAllowlist: ['user-preview'], + }); + + const selected = await service.selectEngine({ + runId: 'run-preview-selected', + userId: 'user-preview', + workflowName: 'code-run-v1', + }); + assert.equal(selected.engine, 'native'); + assert.equal(selected.shadowEngine, 'langgraph'); + assert.equal(selected.candidateEngine, 'langgraph'); + assert.equal(selected.candidateReason, 'user_allowlist'); + assert.equal(selected.dryRun, true); + + const native = await service.selectEngine({ + runId: 'run-preview-native', + userId: 'user-native', + workflowName: 'code-run-v1', + }); + assert.equal(native.engine, 'native'); + assert.equal(native.shadowEngine, 'langgraph'); + assert.equal(native.candidateEngine, 'native'); + assert.equal(native.candidateReason, 'canary_not_selected'); + assert.equal(native.dryRun, true); +}); + test('orchestrator emergency kill switch always forces native selection', async () => { const service = createOrchestratorAdminConfigService(createPool(), { env: { MEMIND_ORCHESTRATOR_KILL_SWITCH: '1' }, diff --git a/services/orchestrator/observability.mjs b/services/orchestrator/observability.mjs index 44a332c..3eb126e 100644 --- a/services/orchestrator/observability.mjs +++ b/services/orchestrator/observability.mjs @@ -1,9 +1,11 @@ import { createRemoteWorkflowEngine } from './engine-registry.mjs'; +import { WORKFLOW_ENGINE } from './contracts.mjs'; const SHADOW_EVENT_TYPES = Object.freeze([ 'workflow_shadow_completed', 'workflow_shadow_failed', ]); +const EXECUTION_PLAN_EVENT_TYPE = 'workflow_execution_planned'; const MAX_METRIC_EVENTS = 5000; const CANARY_READINESS_THRESHOLDS = Object.freeze({ hours: 24 * 7, @@ -90,6 +92,47 @@ function projectUniqueShadowRuns(rows) { 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; } @@ -263,6 +306,40 @@ function summarizeRuns(runs, { capped = false } = {}) { }; } +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), @@ -314,6 +391,37 @@ export function createOrchestratorObservabilityService({ }; } + 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([ @@ -345,6 +453,34 @@ export function createOrchestratorObservabilityService({ }; }, + 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; @@ -432,6 +568,7 @@ export function createOrchestratorObservabilityService({ export const orchestratorObservabilityInternals = { CANARY_READINESS_THRESHOLDS, + EXECUTION_PLAN_EVENT_TYPE, MAX_METRIC_EVENTS, SHADOW_EVENT_TYPES, buildCanaryReadiness, @@ -440,5 +577,8 @@ export const orchestratorObservabilityInternals = { percentile, projectShadowEventRow, projectUniqueShadowRuns, + projectExecutionPlanEventRow, + projectUniqueExecutionPlans, + summarizeExecutionPlans, summarizeRuns, }; diff --git a/services/orchestrator/observability.test.mjs b/services/orchestrator/observability.test.mjs index 04b5b53..1499275 100644 --- a/services/orchestrator/observability.test.mjs +++ b/services/orchestrator/observability.test.mjs @@ -73,6 +73,118 @@ test('observability aggregates shadow outcomes and latency percentiles', async ( assert.equal(result.runs[1].error.code, 'TIMEOUT'); }); +test('observability aggregates complete dry-run routing decisions and Native outcomes', async () => { + const rows = [ + { + event_id: 'plan-event-3', + run_id: 'plan-run-3', + data_json: { + mode: 'canary', + candidateEngine: 'langgraph', + effectiveEngine: 'native', + fallbackEngine: 'native', + reason: 'execution_gate_disabled', + candidateReason: 'percentage', + configVersion: 4, + bucket: 12, + taskType: 'repo_refactor', + dryRun: true, + handoffAllowed: false, + }, + event_created_at: 3000, + request_id: 'plan-request-3', + user_id: 'plan-user-3', + agent_session_id: 'plan-session-2', + native_status: 'failed', + native_attempts: 2, + native_completed_at: 3100, + }, + { + event_id: 'plan-event-2', + run_id: 'plan-run-2', + data_json: JSON.stringify({ + mode: 'canary', + candidateEngine: 'native', + effectiveEngine: 'native', + reason: 'canary_not_selected', + configVersion: 4, + bucket: 78, + taskType: 'code_analysis', + dryRun: true, + handoffAllowed: false, + }), + event_created_at: 2000, + request_id: 'plan-request-2', + user_id: 'plan-user-2', + agent_session_id: 'plan-session-1', + native_status: 'succeeded', + native_attempts: 1, + native_completed_at: 2100, + }, + { + event_id: 'plan-event-1', + run_id: 'plan-run-1', + data_json: { + mode: 'canary', + candidateEngine: 'langgraph', + effectiveEngine: 'native', + reason: 'execution_gate_disabled', + candidateReason: 'user_allowlist', + taskType: 'repo_refactor', + dryRun: true, + handoffAllowed: false, + }, + event_created_at: 1000, + request_id: 'plan-request-1', + user_id: 'plan-user-1', + agent_session_id: 'plan-session-1', + native_status: 'running', + native_attempts: 1, + native_completed_at: null, + }, + ]; + const service = createOrchestratorObservabilityService({ + pool: { + async query(sql, params) { + assert.match(sql, /e\.event_type = \?/); + assert.equal(params[0], 'workflow_execution_planned'); + return [rows]; + }, + }, + configService: { getRuntimeState() {} }, + nowMs: () => 5000, + }); + + const result = await service.listExecutionPlans({ + hours: 12, + limit: 10, + selection: 'candidate', + }); + assert.equal(result.window.hours, 12); + assert.equal(result.metrics.decisions, 3); + assert.equal(result.metrics.candidateSelections, 2); + assert.equal(result.metrics.candidateSelectionRate, 2 / 3); + assert.equal(result.metrics.nativeSelections, 1); + assert.equal(result.metrics.nativeSucceeded, 1); + assert.equal(result.metrics.nativeFailed, 1); + assert.equal(result.metrics.nativeSettledRate, 2 / 3); + assert.equal(result.metrics.handoffAllowed, 0); + assert.equal(result.metrics.distinctSessions, 2); + assert.deepEqual(result.metrics.candidateReasons, [ + { value: 'canary_not_selected', count: 1 }, + { value: 'percentage', count: 1 }, + { value: 'user_allowlist', count: 1 }, + ]); + assert.deepEqual(result.metrics.taskTypes, [ + { value: 'repo_refactor', count: 2 }, + { value: 'code_analysis', count: 1 }, + ]); + assert.equal(result.plans.length, 2); + assert.equal(result.plans[0].candidateEngine, 'langgraph'); + assert.equal(result.plans[0].effectiveEngine, 'native'); + assert.equal(result.plans[0].taskType, 'repo_refactor'); +}); + test('observability detail joins Native state with remote LangGraph checkpoint events', async () => { const queries = []; const service = createOrchestratorObservabilityService({ diff --git a/services/orchestrator/shadow-observer.mjs b/services/orchestrator/shadow-observer.mjs index f611cb5..5eeb376 100644 --- a/services/orchestrator/shadow-observer.mjs +++ b/services/orchestrator/shadow-observer.mjs @@ -49,23 +49,27 @@ export function createWorkflowShadowObserver({ 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 (selection.shadowEngine !== WORKFLOW_ENGINE.LANGGRAPH) { return { observed: false, reason: selection.reason, mode: selection.mode, - executionPlan: selection.dryRun ? { - 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, - dryRun: true, - handoffAllowed: false, - } : null, + executionPlan, }; } @@ -107,10 +111,15 @@ export function createWorkflowShadowObserver({ 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; } }; diff --git a/services/orchestrator/shadow-observer.test.mjs b/services/orchestrator/shadow-observer.test.mjs index 8e8d126..548a641 100644 --- a/services/orchestrator/shadow-observer.test.mjs +++ b/services/orchestrator/shadow-observer.test.mjs @@ -72,6 +72,7 @@ test('shadow boundary returns an auditable dry-run plan without calling LangGrap runId: 'run-dry-1', requestId: 'request-dry-1', userId: 'user-1', + taskType: 'repo_refactor', }); assert.equal(result.observed, false); assert.deepEqual(result.executionPlan, { @@ -83,12 +84,49 @@ test('shadow boundary returns an auditable dry-run plan without calling LangGrap candidateReason: 'user_allowlist', configVersion: 8, bucket: 42, + taskType: 'repo_refactor', dryRun: true, handoffAllowed: false, }); assert.equal(fetchCalls, 0); }); +test('shadow boundary records non-selected canary decisions for a complete rate denominator', async () => { + const observer = createWorkflowShadowObserver({ + configService: { + async selectEngine() { + return { + engine: 'native', + candidateEngine: 'native', + shadowEngine: null, + fallbackEngine: 'native', + reason: 'canary_not_selected', + candidateReason: null, + mode: 'canary', + configVersion: 9, + bucket: 88, + dryRun: false, + }; + }, + async getRuntimeState() { + throw new Error('must not load runtime config'); + }, + }, + }); + + const result = await observer({ + runId: 'run-native-1', + requestId: 'request-native-1', + userId: 'user-1', + taskType: 'code_analysis', + }); + assert.equal(result.observed, false); + assert.equal(result.executionPlan.dryRun, true); + assert.equal(result.executionPlan.candidateEngine, 'native'); + assert.equal(result.executionPlan.reason, 'canary_not_selected'); + assert.equal(result.executionPlan.taskType, 'code_analysis'); +}); + test('shadow observer sends a bounded observe-only RunSpec to LangGraph service', async () => { let capturedUrl = null; let capturedInit = null;