From 85e888b340f08d24744bcd6383d575d5aa4e5913 Mon Sep 17 00:00:00 2001 From: john Date: Fri, 24 Jul 2026 22:09:19 +0800 Subject: [PATCH] feat: add orchestrator canary readiness gate --- admin-routes.mjs | 7 + admin-routes.test.mjs | 17 ++ .../memind-orchestrator-boundary.md | 7 + ops/src/api/admin.ts | 56 +++++ .../pages/admin/OrchestratorShadowPanel.tsx | 95 ++++++++- services/orchestrator/README.md | 13 ++ services/orchestrator/observability.mjs | 198 +++++++++++++++++- services/orchestrator/observability.test.mjs | 147 ++++++++++++- 8 files changed, 534 insertions(+), 6 deletions(-) diff --git a/admin-routes.mjs b/admin-routes.mjs index f62e9a4..66c7427 100644 --- a/admin-routes.mjs +++ b/admin-routes.mjs @@ -268,6 +268,13 @@ export function createAdminApi({ })); }); + adminApi.get('/orchestrator/canary-readiness', requireAdmin, async (_req, res) => { + if (!orchestratorObservabilityService?.getCanaryReadiness) { + return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); + } + return res.json(await orchestratorObservabilityService.getCanaryReadiness()); + }); + adminApi.get('/orchestrator/shadow-runs/:runId', requireAdmin, async (req, res) => { if (!orchestratorObservabilityService?.getShadowRun) { return res.status(503).json({ message: 'Orchestrator 观测服务未启用' }); diff --git a/admin-routes.test.mjs b/admin-routes.test.mjs index d601980..a467fd9 100644 --- a/admin-routes.test.mjs +++ b/admin-routes.test.mjs @@ -139,6 +139,14 @@ test('admin orchestrator routes expose a versioned plug-in control plane', async }, }, orchestratorObservabilityService: { + async getCanaryReadiness() { + return { + ready: false, + recommendation: 'keep_shadow', + blockers: ['sample_volume'], + samples: { eligibleObservations: 1 }, + }; + }, async listShadowRuns(options) { assert.deepEqual(options, { hours: '12', limit: '25', status: 'failed' }); return { @@ -198,6 +206,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 readinessResponse = await fetch( + `${server.baseUrl}/admin-api/orchestrator/canary-readiness`, + { headers: { cookie: 'h5_user_session=token-admin' } }, + ); + assert.equal(readinessResponse.status, 200); + const readinessBody = await readinessResponse.json(); + assert.equal(readinessBody.ready, false); + assert.deepEqual(readinessBody.blockers, ['sample_volume']); + const shadowRunResponse = await fetch( `${server.baseUrl}/admin-api/orchestrator/shadow-runs/run-shadow-1`, { headers: { cookie: 'h5_user_session=token-admin' } }, diff --git a/docs/architecture/memind-orchestrator-boundary.md b/docs/architecture/memind-orchestrator-boundary.md index e9e36c8..78ed220 100644 --- a/docs/architecture/memind-orchestrator-boundary.md +++ b/docs/architecture/memind-orchestrator-boundary.md @@ -110,6 +110,13 @@ Memind-owned observability service aggregates those events and may join them to the product run status. Only per-run detail calls cross the service boundary to read LangGraph checkpoint state and node events. +Canary readiness is also a Memind control-plane projection. It excludes +synthetic smoke runs and evaluates operational health, durable checkpoint state, +sample volume, session coverage, success rate, latency coverage, P95 latency, +Native terminal coverage, and sample freshness. The result is advisory: +`manual_canary_review` never mutates routing configuration or transfers +execution ownership. + ## Deployment evolution 1. Local native Orchestrator process with an explicit MemorySaver for debugging. diff --git a/ops/src/api/admin.ts b/ops/src/api/admin.ts index 4c8a983..74c90f8 100644 --- a/ops/src/api/admin.ts +++ b/ops/src/api/admin.ts @@ -284,6 +284,7 @@ export type OrchestratorShadowRun = { configVersion: number | null; phase: string | null; taskType: string | null; + synthetic: boolean; executorAdapter: string | null; latencyMs: number | null; error: { code: string; message: string } | null; @@ -298,6 +299,55 @@ export type OrchestratorShadowRunList = { runs: OrchestratorShadowRun[]; }; +export type OrchestratorCanaryReadinessCheck = { + id: string; + passed: boolean; + actual: string | number | boolean | null; + target: string | number | boolean | null; +}; + +export type OrchestratorCanaryReadiness = { + generatedAt: number; + ready: boolean; + recommendation: 'keep_shadow' | 'manual_canary_review'; + window: { hours: number; from: number }; + thresholds: { + hours: number; + minObservations: number; + minSuccessRate: number; + maxP95LatencyMs: number; + minLatencyCoverageRate: number; + minNativeSettledRate: number; + minDistinctSessions: number; + maxHoursSinceLastObservation: number; + }; + samples: { + totalObservations: number; + eligibleObservations: number; + excludedSynthetic: number; + successes: number; + failures: number; + successRate: number | null; + latencyCoverageRate: number | null; + latencyP95Ms: number | null; + nativeSettledRate: number | null; + distinctSessions: number; + lastObservedAt: number | null; + hoursSinceLastObservation: number | null; + sampled: boolean; + }; + service: { + status: string | null; + latencyMs: number | null; + checkpointKind: string | null; + checkpointDurable: boolean | null; + execution: string | null; + }; + checks: OrchestratorCanaryReadinessCheck[]; + blockers: string[]; + failureCodes: Array<{ code: string; count: number }>; +}; + export type OrchestratorShadowRunDetail = { native: { runId: string; @@ -386,6 +436,12 @@ export async function fetchOrchestratorShadowRuns(params: { ); } +export async function fetchOrchestratorCanaryReadiness() { + return adminFetch( + '/admin-api/orchestrator/canary-readiness', + ); +} + export async function fetchOrchestratorShadowRun(runId: string) { return adminFetch( `/admin-api/orchestrator/shadow-runs/${encodeURIComponent(runId)}`, diff --git a/ops/src/pages/admin/OrchestratorShadowPanel.tsx b/ops/src/pages/admin/OrchestratorShadowPanel.tsx index a8fa2a8..1ba4b30 100644 --- a/ops/src/pages/admin/OrchestratorShadowPanel.tsx +++ b/ops/src/pages/admin/OrchestratorShadowPanel.tsx @@ -1,7 +1,10 @@ import { useEffect, useState } from 'react'; import { + fetchOrchestratorCanaryReadiness, fetchOrchestratorShadowRun, fetchOrchestratorShadowRuns, + type OrchestratorCanaryReadiness, + type OrchestratorCanaryReadinessCheck, type OrchestratorShadowRunDetail, type OrchestratorShadowRunList, } from '../../api/admin'; @@ -22,6 +25,46 @@ function shortId(value: string) { return value.length > 16 ? `${value.slice(0, 8)}…${value.slice(-6)}` : value; } +const readinessCheckLabels: Record = { + shadow_mode: '当前保持 Shadow', + service_healthy: '服务健康', + durable_checkpoint: '持久化 checkpoint', + observe_only: '仅观察不执行', + sample_volume: '有效样本量', + shadow_success_rate: 'Shadow 成功率', + latency_coverage: '延迟数据覆盖率', + latency_p95: '延迟 P95', + native_settled_rate: 'Native 终态覆盖率', + session_coverage: '真实会话覆盖', + sample_freshness: '最近样本新鲜度', + complete_window: '统计窗口完整', +}; + +function formatReadinessValue(check: OrchestratorCanaryReadinessCheck) { + const value = check.actual; + if (value == null) return '无数据'; + if (['shadow_success_rate', 'latency_coverage', 'native_settled_rate'].includes(check.id)) { + return formatPercent(Number(value)); + } + if (check.id === 'latency_p95') return formatLatency(Number(value)); + if (check.id === 'sample_freshness') return `${Number(value).toFixed(1)} 小时`; + if (typeof value === 'boolean') return value ? '是' : '否'; + return String(value); +} + +function formatReadinessTarget(check: OrchestratorCanaryReadinessCheck) { + const target = check.target; + if (target == null) return ''; + if (['shadow_success_rate', 'latency_coverage', 'native_settled_rate'].includes(check.id)) { + return `≥ ${formatPercent(Number(target))}`; + } + if (check.id === 'latency_p95') return `≤ ${formatLatency(Number(target))}`; + if (check.id === 'sample_freshness') return `≤ ${target} 小时`; + if (['sample_volume', 'session_coverage'].includes(check.id)) return `≥ ${target}`; + if (typeof target === 'boolean') return target ? '= 是' : '= 否'; + return `= ${target}`; +} + const metricCardStyle = { border: '1px solid #e4e9e6', borderRadius: 10, @@ -33,6 +76,7 @@ export function OrchestratorShadowPanel() { const [hours, setHours] = useState(24); const [status, setStatus] = useState<'all' | 'succeeded' | 'failed'>('all'); const [data, setData] = useState(null); + const [readiness, setReadiness] = useState(null); const [selected, setSelected] = useState(null); const [loading, setLoading] = useState(true); const [detailLoading, setDetailLoading] = useState(false); @@ -42,7 +86,12 @@ export function OrchestratorShadowPanel() { setLoading(true); setError(null); try { - setData(await fetchOrchestratorShadowRuns({ hours, status, limit: 100 })); + const [runs, canaryReadiness] = await Promise.all([ + fetchOrchestratorShadowRuns({ hours, status, limit: 100 }), + fetchOrchestratorCanaryReadiness(), + ]); + setData(runs); + setReadiness(canaryReadiness); } catch (err) { setError(err instanceof Error ? err.message : 'Shadow 指标加载失败'); } finally { @@ -100,6 +149,49 @@ export function OrchestratorShadowPanel() { {error ?

{error}

: null} + {readiness ? ( +
+
+
+

Canary 准入评估

+

+ 固定 7 天窗口；smoke 样本不计入准入，不会自动切换运行模式。 +

+
+ + {readiness.ready ? '已达到门槛，等待人工审批' : '继续保持 Shadow'} + +
+

+ 有效样本 {readiness.samples.eligibleObservations} / {readiness.thresholds.minObservations} + {' · '} + 已排除 smoke {readiness.samples.excludedSynthetic} + {' · '} + 真实会话 {readiness.samples.distinctSessions} / {readiness.thresholds.minDistinctSessions} +

+
+ {readiness.checks.map((check) => ( +
+ {check.passed ? '通过' : '阻塞'} · {readinessCheckLabels[check.id] ?? check.id} + ：{formatReadinessValue(check)}（目标 {formatReadinessTarget(check)}） +
+ ))} +
+ {readiness.failureCodes.length ? ( +

+ 主要失败：{readiness.failureCodes.map((item) => `${item.code} × ${item.count}`).join('，')} +

+ ) : null} +
+ ) : null} +
观察请求 @@ -168,6 +260,7 @@ export function OrchestratorShadowPanel() { {formatLatency(run.latencyMs)} {run.taskType ?? '—'} + {run.synthetic ? '（smoke）' : ''} {run.executorAdapter ?? '—'} diff --git a/services/orchestrator/README.md b/services/orchestrator/README.md index c652f17..d2c3e60 100644 --- a/services/orchestrator/README.md +++ b/services/orchestrator/README.md @@ -60,6 +60,18 @@ Phase 2.5 adds a Memind-owned Shadow observability projection: The projection reads `h5_agent_run_events` from the Memind control plane. LangGraph still has no access to Memind business tables. +Phase 2.6 adds a read-only Canary readiness gate. It evaluates a fixed seven-day +window and never changes the runtime mode. The first gate requires: + +- at least 20 non-smoke observations across at least five real sessions; +- at least 95% Shadow success, latency coverage, and Native terminal coverage; +- Shadow P95 latency no greater than two seconds; +- a healthy observe-only service backed by a durable checkpoint; +- a fresh, uncapped metric window while the control plane remains in Shadow. + +Passing the gate means only `manual_canary_review`; it does not authorize or +activate Canary routing. + The default memindadm mode remains `off`. Canary and Active routing are not wired to the LangGraph executor in Phase 2. Native Agent Run remains the sole executor, including in Shadow mode. @@ -101,6 +113,7 @@ memindadm exposes the projection through: ```text GET /admin-api/orchestrator/shadow-runs GET /admin-api/orchestrator/shadow-runs/:runId +GET /admin-api/orchestrator/canary-readiness ``` ## Colima deployment diff --git a/services/orchestrator/observability.mjs b/services/orchestrator/observability.mjs index 0ac3377..44a332c 100644 --- a/services/orchestrator/observability.mjs +++ b/services/orchestrator/observability.mjs @@ -5,6 +5,18 @@ const SHADOW_EVENT_TYPES = Object.freeze([ 'workflow_shadow_failed', ]); const MAX_METRIC_EVENTS = 5000; +const CANARY_READINESS_THRESHOLDS = Object.freeze({ + hours: 24 * 7, + minObservations: 20, + minSuccessRate: 0.95, + 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); @@ -38,6 +50,7 @@ function projectShadowEventRow(row) { const data = parseJsonColumn(row.data_json, {}) ?? {}; const succeeded = row.event_type === 'workflow_shadow_completed'; const latency = Number(data.latencyMs); + const taskType = data.taskType ?? null; return { eventId: row.event_id, runId: row.run_id, @@ -50,7 +63,8 @@ function projectShadowEventRow(row) { engine: data.engine ?? 'langgraph', configVersion: data.configVersion == null ? null : Number(data.configVersion), phase: data.phase ?? null, - taskType: data.taskType ?? 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 ? null : { @@ -64,6 +78,170 @@ function projectShadowEventRow(row) { }; } +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 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 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 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 checks = [ + { + id: 'shadow_mode', + passed: runtimeState?.config?.mode === 'shadow', + actual: runtimeState?.config?.mode ?? null, + target: 'shadow', + }, + { + 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: '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: '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: eligibleRuns.length - successes, + successRate, + 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, + 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.length - successes; @@ -97,6 +275,7 @@ export function createOrchestratorObservabilityService({ 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) { @@ -105,7 +284,7 @@ export function createOrchestratorObservabilityService({ async function loadShadowEvents({ hours = 24 } = {}) { const normalizedHours = clampInteger(hours, 24, 1, 24 * 30); - const from = Date.now() - normalizedHours * 60 * 60 * 1000; + const from = nowMs() - normalizedHours * 60 * 60 * 1000; const [rows] = await pool.query( `SELECT e.id AS event_id, @@ -131,11 +310,19 @@ export function createOrchestratorObservabilityService({ from, hours: normalizedHours, capped: rows.length >= MAX_METRIC_EVENTS, - runs: rows.map(projectShadowEventRow), + runs: projectUniqueShadowRuns(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, @@ -148,7 +335,7 @@ export function createOrchestratorObservabilityService({ ? loaded.runs : loaded.runs.filter((run) => run.shadowStatus === normalizedStatus); return { - generatedAt: Date.now(), + generatedAt: nowMs(), window: { hours: loaded.hours, from: loaded.from, @@ -244,11 +431,14 @@ export function createOrchestratorObservabilityService({ } export const orchestratorObservabilityInternals = { + CANARY_READINESS_THRESHOLDS, MAX_METRIC_EVENTS, SHADOW_EVENT_TYPES, + buildCanaryReadiness, normalizeRunId, parseJsonColumn, percentile, projectShadowEventRow, + projectUniqueShadowRuns, summarizeRuns, }; diff --git a/services/orchestrator/observability.test.mjs b/services/orchestrator/observability.test.mjs index 5f4c99f..04b5b53 100644 --- a/services/orchestrator/observability.test.mjs +++ b/services/orchestrator/observability.test.mjs @@ -1,6 +1,9 @@ import assert from 'node:assert/strict'; import test from 'node:test'; -import { createOrchestratorObservabilityService } from './observability.mjs'; +import { + createOrchestratorObservabilityService, + orchestratorObservabilityInternals, +} from './observability.mjs'; function response(body, status = 200) { return new Response(JSON.stringify(body), { @@ -148,3 +151,145 @@ test('observability rejects unsafe run ids before querying', async () => { assert.equal(await service.getShadowRun('../unsafe'), null); assert.equal(queries, 0); }); + +test('observability counts only the newest terminal event for each run', () => { + const base = { + run_id: 'same-run', + request_id: 'same-request', + user_id: 'same-user', + native_status: 'succeeded', + native_attempts: 1, + }; + const runs = orchestratorObservabilityInternals.projectUniqueShadowRuns([ + { + ...base, + event_id: 'newer-success', + event_type: 'workflow_shadow_completed', + data_json: { latencyMs: 80 }, + event_created_at: 2000, + }, + { + ...base, + event_id: 'older-failure', + event_type: 'workflow_shadow_failed', + data_json: { latencyMs: 100, code: 'TIMEOUT' }, + event_created_at: 1000, + }, + ]); + assert.equal(runs.length, 1); + assert.equal(runs[0].eventId, 'newer-success'); + assert.equal(runs[0].shadowStatus, 'succeeded'); +}); + +test('canary readiness excludes smoke runs and reports explicit blockers', async () => { + const now = 10_000_000; + const service = createOrchestratorObservabilityService({ + pool: { + async query() { + return [[{ + event_id: 'smoke-event', + run_id: 'smoke-run', + event_type: 'workflow_shadow_completed', + data_json: { + latencyMs: 50, + taskType: 'orchestrator_shadow_smoke', + }, + event_created_at: now - 1000, + request_id: 'smoke-request', + user_id: 'smoke-user', + agent_session_id: null, + native_status: 'succeeded', + native_attempts: 0, + native_completed_at: now, + }]]; + }, + }, + configService: { + async getRuntimeState(options) { + assert.deepEqual(options, { probe: true }); + return { + config: { mode: 'shadow' }, + serviceHealth: { + ok: true, + status: 'healthy', + latencyMs: 5, + details: { + checkpoint: { kind: 'postgres', durable: true }, + execution: 'observe-only', + }, + }, + }; + }, + }, + nowMs: () => now, + }); + + const readiness = await service.getCanaryReadiness(); + assert.equal(readiness.ready, false); + assert.equal(readiness.recommendation, 'keep_shadow'); + assert.equal(readiness.samples.totalObservations, 1); + assert.equal(readiness.samples.eligibleObservations, 0); + assert.equal(readiness.samples.excludedSynthetic, 1); + assert.ok(readiness.blockers.includes('sample_volume')); + assert.ok(readiness.blockers.includes('session_coverage')); + assert.ok(!readiness.blockers.includes('service_healthy')); +}); + +test('canary readiness passes only when operational and sample gates all pass', async () => { + const now = 20_000_000; + const rows = Array.from({ length: 20 }, (_, index) => ({ + event_id: `event-${index}`, + run_id: `run-${index}`, + event_type: index === 19 + ? 'workflow_shadow_failed' + : 'workflow_shadow_completed', + data_json: { + latencyMs: 100 + index, + taskType: 'code_task', + ...(index === 19 ? { code: 'TRANSIENT', message: 'retry later' } : {}), + }, + event_created_at: now - index * 1000, + request_id: `request-${index}`, + user_id: `user-${index % 3}`, + agent_session_id: `session-${index % 5}`, + native_status: index % 2 ? 'succeeded' : 'failed', + native_attempts: 1, + native_completed_at: now, + })); + const service = createOrchestratorObservabilityService({ + pool: { + async query() { + return [rows]; + }, + }, + configService: { + async getRuntimeState() { + return { + config: { mode: 'shadow' }, + serviceHealth: { + ok: true, + status: 'healthy', + latencyMs: 4, + details: { + checkpoint: { kind: 'postgres', durable: true }, + execution: 'observe-only', + }, + }, + }; + }, + }, + nowMs: () => now, + }); + + const readiness = await service.getCanaryReadiness(); + assert.equal(readiness.ready, true); + assert.equal(readiness.recommendation, 'manual_canary_review'); + assert.equal(readiness.window.hours, 168); + assert.equal(readiness.samples.eligibleObservations, 20); + assert.equal(readiness.samples.successRate, 0.95); + assert.equal(readiness.samples.latencyCoverageRate, 1); + assert.equal(readiness.samples.nativeSettledRate, 1); + assert.equal(readiness.samples.distinctSessions, 5); + assert.equal(readiness.blockers.length, 0); + assert.deepEqual(readiness.failureCodes, [{ code: 'TRANSIENT', count: 1 }]); +});