Surface run wait age on operator metrics and durable_resume_paths health

Adds runs.waiting, runs.oldest_wait_started_at, and runs.max_wait_age_ms
to OperatorMetrics::snapshot() so operators can answer "how long has the
worst-case run been waiting at a durable resume point?" from the metric
alone, mirroring the existing tasks.oldest_lease_expired_at,
tasks.oldest_ready_due_at, and backlog.oldest_compatibility_blocked_started_at
shapes. The signal counts every kind of wait — signal, update, timer, and
compatibility-blocked — because each is a durable resume point the system
is parked on; consumers that want to isolate the non-compatibility share
can subtract runs.compatibility_blocked.

Forwards the same trio on HealthCheck::durableResumePathCheck() data as
waiting_runs / oldest_wait_started_at / max_wait_age_ms so the wait age
is legible from /healthz without re-reading the metrics snapshot. The
check's escalation predicate stays unchanged (it still escalates only on
repair_needed_runs); the wait-age data is observability, since a
long-parked wait is not by itself a stuck condition the system can repair
without application-level action.

Pins the new keys in docs/architecture/rollout-safety.md frozen metric
table and adds a wait-row regression assertion in
RolloutSafetyDocumentationTest. Extends V2OperatorMetricsTest with two
focused tests proving the metric counts non-compatibility waits and
returns null/0 when no runs are waiting, plus an existing-fixture
assertion in the broad metrics test. Extends HealthCheckTest's
testSnapshotWarnsWhenRunSummaryProjectionSchemaIsOutdated assertion to
pin the new health-check fields.

Author: durable-workflow-ops

docs/architecture/rollout-safety.md

@@ -405,6 +405,7 @@ change.
 | `runs` | `repair_needed` | open runs with `liveness_state = repair_needed` |
 | `runs` | `claim_failed` | runs whose most recent task claim failed |
 | `runs` | `compatibility_blocked` | runs blocked by compatibility mismatch |
+| `runs` | `waiting`, `oldest_wait_started_at`, `max_wait_age_ms` | running runs currently parked at a durable resume point (`status_bucket = 'running'` and `wait_started_at IS NOT NULL`), the earliest `wait_started_at` among them, and the largest wait age in milliseconds. Mirrors the `backlog.oldest_compatibility_blocked_started_at` / `max_compatibility_blocked_age_ms` and `tasks.oldest_lease_expired_at` / `max_lease_expired_age_ms` shapes so operators can answer "how long has the worst-case run been waiting at a signal, update, timer, or compatible-worker arrival?" from the metric alone. The signal is unconditional and includes compatibility-blocked waits; consumers that want the non-compatibility share can subtract `runs.compatibility_blocked` and `backlog.oldest_compatibility_blocked_started_at`. |
 | `tasks` | `ready`, `ready_due`, `delayed`, `leased` | queue depth by phase |
 | `tasks` | `dispatch_failed`, `claim_failed` | transport failure counts |
 | `tasks` | `dispatch_overdue`, `lease_expired` | lease and dispatch timing |
@@ -494,6 +495,18 @@ are authoritative and how they surface.
 - **Stale projection.** A projection behind the authoritative
   history surfaces through the `run_summary_projection` and
   `selected_run_projections` checks on `HealthCheck::snapshot()`.
+- **Long-parked wait.** A running run whose projector has recorded
+  a `wait_started_at` is counted under `runs.waiting`, and its
+  worst-case wait age is surfaced through `runs.oldest_wait_started_at`
+  and `runs.max_wait_age_ms`, both forwarded on the
+  `durable_resume_paths` health check (`waiting_runs`,
+  `oldest_wait_started_at`, `max_wait_age_ms`). The signal includes
+  every kind of wait — signal, update, timer, and compatibility-blocked
+  wait — because each is a durable resume point the system is
+  parked on. The check itself escalates only on `repair_needed_runs`;
+  the wait-age data is observability so operators can decide whether
+  the worst-case wait reflects healthy long-running work or a lost
+  signal/update that the application must resend.
 
 Guarantees:
 

src/V2/Support/HealthCheck.php

@@ -28,7 +28,11 @@ public static function snapshot(?CarbonInterface $now = null): array
             self::historyRetentionInvariantCheck($metrics['history'] ?? []),
             self::commandContractCheck($metrics['command_contracts'] ?? []),
             self::taskTransportCheck($metrics['tasks'] ?? [], $metrics['backlog'] ?? []),
-            self::durableResumePathCheck($metrics['backlog'] ?? [], $metrics['repair'] ?? []),
+            self::durableResumePathCheck(
+                $metrics['backlog'] ?? [],
+                $metrics['repair'] ?? [],
+                $metrics['runs'] ?? [],
+            ),
             self::workerCompatibilityCheck($metrics['workers'] ?? []),
             self::schedulerRoleCheck($metrics['schedules'] ?? []),
             self::longPollWakeAccelerationCheck(),
@@ -247,9 +251,10 @@ private static function taskTransportCheck(array $tasks, array $backlog): array
     /**
      * @param array<string, mixed> $backlog
      * @param array<string, mixed> $repair
+     * @param array<string, mixed> $runs
      * @return array<string, mixed>
      */
-    private static function durableResumePathCheck(array $backlog, array $repair): array
+    private static function durableResumePathCheck(array $backlog, array $repair, array $runs): array
     {
         $repairNeededRuns = self::integer($backlog['repair_needed_runs'] ?? 0);
 
@@ -268,6 +273,11 @@ private static function durableResumePathCheck(array $backlog, array $repair): a
                     ? $repair['oldest_missing_run_started_at']
                     : null,
                 'max_missing_run_age_ms' => self::integer($repair['max_missing_run_age_ms'] ?? 0),
+                'waiting_runs' => self::integer($runs['waiting'] ?? 0),
+                'oldest_wait_started_at' => is_string($runs['oldest_wait_started_at'] ?? null)
+                    ? $runs['oldest_wait_started_at']
+                    : null,
+                'max_wait_age_ms' => self::integer($runs['max_wait_age_ms'] ?? 0),
             ],
         );
     }

src/V2/Support/OperatorMetrics.php

@@ -40,7 +40,7 @@ public static function snapshot(?CarbonInterface $now = null, ?string $namespace
 
         return [
             'generated_at' => $now->toJSON(),
-            'runs' => self::runMetrics($namespace),
+            'runs' => self::runMetrics($now, $namespace),
             'tasks' => self::taskMetrics($now, $namespace),
             'activities' => self::activityMetrics($namespace),
             'backlog' => self::backlogMetrics($now, $namespace),
@@ -88,10 +88,12 @@ private static function matchingRoleSnapshot(): array
     }
 
     /**
-     * @return array<string, int>
+     * @return array<string, int|string|null>
      */
-    private static function runMetrics(?string $namespace): array
+    private static function runMetrics(CarbonInterface $now, ?string $namespace): array
     {
+        $oldestWaitStartedAt = self::oldestRunWaitStartedAt($namespace);
+
         return [
             'total' => self::summaryQuery($namespace)->count(),
             'current' => self::summaryQuery($namespace)->where('is_current_run', true)->count(),
@@ -104,6 +106,11 @@ private static function runMetrics(?string $namespace): array
             'repair_needed' => self::summaryQuery($namespace)->where('liveness_state', 'repair_needed')->count(),
             'claim_failed' => self::claimFailedRuns($namespace),
             'compatibility_blocked' => self::compatibilityBlockedRuns($namespace),
+            'waiting' => self::waitingRuns($namespace),
+            'oldest_wait_started_at' => $oldestWaitStartedAt?->toJSON(),
+            'max_wait_age_ms' => $oldestWaitStartedAt === null
+                ? 0
+                : (int) $oldestWaitStartedAt->diffInMilliseconds($now),
         ];
     }
 
@@ -711,6 +718,50 @@ private static function claimFailedRuns(?string $namespace): int
             ->count();
     }
 
+    /**
+     * Open runs that are currently parked at a wait point — running runs whose
+     * `RunSummaryProjector` has recorded a `wait_started_at` because they are
+     * blocked on a signal, update, timer, or compatible-worker arrival.
+     *
+     * Counted unconditionally so the worst-case wait age is legible regardless
+     * of what kind of wait it is; consumers that want to exclude
+     * compatibility-blocked waits can subtract `runs.compatibility_blocked`.
+     */
+    private static function waitingRuns(?string $namespace): int
+    {
+        return self::summaryQuery($namespace)
+            ->where('status_bucket', 'running')
+            ->whereNotNull('wait_started_at')
+            ->count();
+    }
+
+    /**
+     * Earliest `wait_started_at` timestamp across runs currently parked at a
+     * wait point. Rollout-safety surfaces this alongside `runs.waiting` so
+     * operators can answer "how long has the worst-case run been waiting at
+     * a durable resume point?" from the metric alone, mirroring the existing
+     * `backlog.oldest_compatibility_blocked_started_at` and
+     * `tasks.oldest_lease_expired_at` shapes. The signal includes
+     * compatibility-blocked waits because they too are durable-resume points
+     * the system is waiting on; consumers that need to isolate the
+     * non-compatibility share can use `backlog.oldest_compatibility_blocked_started_at`.
+     */
+    private static function oldestRunWaitStartedAt(?string $namespace): ?CarbonInterface
+    {
+        /** @var WorkflowRunSummary|null $summary */
+        $summary = self::summaryQuery($namespace)
+            ->where('status_bucket', 'running')
+            ->whereNotNull('wait_started_at')
+            ->orderBy('wait_started_at')
+            ->first();
+
+        if (! $summary instanceof WorkflowRunSummary) {
+            return null;
+        }
+
+        return $summary->wait_started_at;
+    }
+
     private static function retryingActivities(?string $namespace): int
     {
         return self::scopedRunModelQuery(self::activityExecutionModel(), $namespace)

tests/Feature/V2/V2OperatorMetricsTest.php

@@ -202,6 +202,14 @@ public function testSnapshotSummarizesDurableBacklogRepairCompatibilityAndWorker
         $this->assertSame(1, $snapshot['runs']['repair_needed']);
         $this->assertSame(1, $snapshot['runs']['claim_failed']);
         $this->assertSame(1, $snapshot['runs']['compatibility_blocked']);
+        $this->assertSame(1, $snapshot['runs']['waiting']);
+        $this->assertSame(
+            Carbon::parse('2026-04-09 12:00:00')
+                ->subMinutes(4)
+                ->toJSON(),
+            $snapshot['runs']['oldest_wait_started_at'],
+        );
+        $this->assertSame(4 * 60 * 1000, $snapshot['runs']['max_wait_age_ms']);
         $this->assertSame(7, $snapshot['tasks']['open']);
         $this->assertSame(5, $snapshot['tasks']['ready']);
         $this->assertSame(4, $snapshot['tasks']['ready_due']);
@@ -931,6 +939,111 @@ public function testSnapshotSurfacesSchedulerRoleHealth(): void
         $this->assertSame(2, $snapshot['schedules']['failures_total']);
     }
 
+    public function testSnapshotSurfacesRunWaitAgeForEveryRunningWaitNotJustCompatibilityBlocked(): void
+    {
+        Carbon::setTestNow('2026-04-09 12:00:00');
+        $this->beforeApplicationDestroyed(static function (): void {
+            Carbon::setTestNow();
+        });
+
+        $now = Carbon::now();
+
+        // Signal-wait run parked the longest — earliest wait_started_at wins.
+        $signalWait = $this->createRunWithSummary(
+            instanceId: 'wait-age-instance-signal',
+            runId: '01JWAITSIGNALRUN0000000001',
+            status: 'waiting',
+            statusBucket: 'running',
+            livenessState: 'workflow_task_waiting_for_signal',
+        );
+        WorkflowRunSummary::query()
+            ->whereKey($signalWait->id)
+            ->update([
+                'wait_started_at' => $now->copy()
+                    ->subMinutes(7),
+            ]);
+
+        // Compatibility-blocked wait — counted under runs.waiting too.
+        $compatibilityWait = $this->createRunWithSummary(
+            instanceId: 'wait-age-instance-compat',
+            runId: '01JWAITCOMPATRUN0000000002',
+            status: 'waiting',
+            statusBucket: 'running',
+            livenessState: 'workflow_task_waiting_for_compatible_worker',
+        );
+        WorkflowRunSummary::query()
+            ->whereKey($compatibilityWait->id)
+            ->update([
+                'wait_started_at' => $now->copy()
+                    ->subMinutes(2),
+            ]);
+
+        // Running run that is actively executing — must not count.
+        $this->createRunWithSummary(
+            instanceId: 'wait-age-instance-active',
+            runId: '01JWAITACTIVERUN0000000003',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'running',
+        );
+
+        // Closed run that previously had a wait — must not count.
+        $closedWait = $this->createRunWithSummary(
+            instanceId: 'wait-age-instance-closed',
+            runId: '01JWAITCLOSEDRUN0000000004',
+            status: 'completed',
+            statusBucket: 'completed',
+            livenessState: 'closed',
+        );
+        WorkflowRunSummary::query()
+            ->whereKey($closedWait->id)
+            ->update([
+                'wait_started_at' => $now->copy()
+                    ->subHours(2),
+            ]);
+
+        $snapshot = OperatorMetrics::snapshot($now);
+
+        $expectedOldestWaitAt = $now->copy()
+            ->subMinutes(7)
+            ->toJSON();
+
+        $this->assertSame(2, $snapshot['runs']['waiting']);
+        $this->assertSame($expectedOldestWaitAt, $snapshot['runs']['oldest_wait_started_at']);
+        $this->assertSame(7 * 60 * 1000, $snapshot['runs']['max_wait_age_ms']);
+
+        $healthSnapshot = HealthCheck::snapshot($now);
+        $resumePaths = collect($healthSnapshot['checks'])->firstWhere('name', 'durable_resume_paths');
+        $this->assertNotNull($resumePaths);
+        $this->assertSame(2, $resumePaths['data']['waiting_runs']);
+        $this->assertSame($expectedOldestWaitAt, $resumePaths['data']['oldest_wait_started_at']);
+        $this->assertSame(7 * 60 * 1000, $resumePaths['data']['max_wait_age_ms']);
+    }
+
+    public function testSnapshotReportsRunWaitAgeAsZeroWhenNoRunsAreWaiting(): void
+    {
+        Carbon::setTestNow('2026-04-09 12:00:00');
+        $this->beforeApplicationDestroyed(static function (): void {
+            Carbon::setTestNow();
+        });
+
+        $now = Carbon::now();
+
+        $this->createRunWithSummary(
+            instanceId: 'no-wait-instance-active',
+            runId: '01JWAITNONERUN00000000001',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'running',
+        );
+
+        $snapshot = OperatorMetrics::snapshot($now);
+
+        $this->assertSame(0, $snapshot['runs']['waiting']);
+        $this->assertNull($snapshot['runs']['oldest_wait_started_at']);
+        $this->assertSame(0, $snapshot['runs']['max_wait_age_ms']);
+    }
+
     public function testSnapshotReportsInWorkerMatchingRoleShapeByDefault(): void
     {
         config()->set('workflows.v2.matching_role.queue_wake_enabled', true);

tests/Unit/V2/HealthCheckTest.php

@@ -559,6 +559,9 @@ public function testSnapshotWarnsWhenOpenRunHasNoDurableResumePath(): void
         $this->assertSame(1, $resumePaths['data']['selected_missing_task_candidates']);
         $this->assertSame('2026-04-09T11:55:00.000000Z', $resumePaths['data']['oldest_missing_run_started_at']);
         $this->assertSame(300000, $resumePaths['data']['max_missing_run_age_ms']);
+        $this->assertSame(1, $resumePaths['data']['waiting_runs']);
+        $this->assertSame('2026-04-09T11:55:00.000000Z', $resumePaths['data']['oldest_wait_started_at']);
+        $this->assertSame(300000, $resumePaths['data']['max_wait_age_ms']);
     }
 
     public function testSnapshotWarnsWhenRunSummaryProjectionSchemaIsOutdated(): void

tests/Unit/V2/RolloutSafetyDocumentationTest.php

@@ -138,6 +138,9 @@ final class RolloutSafetyDocumentationTest extends TestCase
         'compatibility_blocked_runs',
         'oldest_compatibility_blocked_started_at',
         'max_compatibility_blocked_age_ms',
+        'waiting',
+        'oldest_wait_started_at',
+        'max_wait_age_ms',
         'missing_task_candidates',
         'selected_missing_task_candidates',
         'oldest_missing_run_started_at',
@@ -347,6 +350,17 @@ public function testContractDocumentFreezesReadyDueAgeRow(): void
         );
     }
 
+    public function testContractDocumentFreezesRunWaitAgeRow(): void
+    {
+        $contents = $this->documentContents();
+
+        $this->assertMatchesRegularExpression(
+            '/\|\s*`runs`\s*\|[^|]*`waiting`[^|]*`oldest_wait_started_at`[^|]*`max_wait_age_ms`/',
+            $contents,
+            'Rollout safety contract must pin the runs wait-age row so operators can read "how long has the worst-case run been waiting at a durable resume point?" from OperatorMetrics::snapshot() without scanning workflow_run_summaries.',
+        );
+    }
+
     public function testContractDocumentFreezesHealthCheckNames(): void
     {
         $contents = $this->documentContents();