feat: add workflow dry-run observability

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