Surface repair-needed age on operator metrics

durable-workflow-ops · durable-workflow-ops · commit 402470f2d64c · 2026-04-25T07:20:06.000Z
Adds `runs.oldest_repair_needed_at` and `runs.max_repair_needed_age_ms`
to `OperatorMetrics::snapshot()` so operators can read "how long has the
worst-case run been stuck without progress?" — the canonical
stuck-workflow duplicate-risk age indicator paired with the
`durable_resume_paths` health check — from the metric alone without
walking `workflow_run_summaries`.

The summary's `updated_at` is sourced by `RunSummaryProjector` from
`WorkflowRun::last_progress_at`, so it advances when the run made forward
progress and stalls when the run stopped progressing. For runs already
pinned at `liveness_state = repair_needed` it is therefore the closest
available proxy for "when this run last made progress before being
marked broken." Falls back to the run's `started_at` when the projection
has not recorded a progress boundary (a fresh run that was projected
straight into `repair_needed` without a prior progress write).

The `repair_needed` predicate matches `liveness_state = 'repair_needed'`
exactly, mirroring the existing count under `runs.repair_needed`. It
deliberately excludes the routing-blocked variant
`workflow_task_waiting_for_compatible_worker` (a wait state, not a
broken state); compatibility-blocked age is already surfaced under
`backlog.oldest_compatibility_blocked_started_at` /
`max_compatibility_blocked_age_ms`. The new signal pairs with that
routing-block age the same way `tasks.oldest_unhealthy_at` /
`max_unhealthy_age_ms` rolls up the four contributing per-path ages.

Pinned in `docs/architecture/rollout-safety.md` Frozen metric keys
table and asserted by `RolloutSafetyDocumentationTest` (frozen-keys
list plus a dedicated row regex). Covered end-to-end by two new
`V2OperatorMetricsTest` cases that verify the predicate matches
liveness_state exactly (compatibility-blocked rows do not contribute
their older `updated_at`), that the oldest stuck-since timestamp wins
across multiple repair_needed runs, and that the keys read as
`null` / 0 when no run is in repair_needed.
diff --git a/docs/architecture/rollout-safety.md b/docs/architecture/rollout-safety.md
@@ -403,6 +403,7 @@ change.
 | Section | Key | Meaning |
 | ------- | --- | ------- |
 | `runs` | `repair_needed` | open runs with `liveness_state = repair_needed` |
+| `runs` | `oldest_repair_needed_at`, `max_repair_needed_age_ms` | earliest "stuck since" timestamp across runs whose `liveness_state` is exactly `repair_needed` and the largest stuck age in milliseconds, mirroring the `backlog.oldest_compatibility_blocked_started_at` / `max_compatibility_blocked_age_ms` shape on the routing path so operators can read "how long has the worst-case run been stuck without progress?" — the canonical stuck-workflow duplicate-risk age indicator paired with the `durable_resume_paths` health check — from the metric alone without walking `workflow_run_summaries`. The summary's `updated_at` is sourced by `RunSummaryProjector` from `WorkflowRun::last_progress_at`, so it advances with forward progress and stalls when the run stops progressing; for runs already at `repair_needed` it is the closest available proxy for "when this run last made progress before being marked broken." Falls back to the run's `started_at` when the projection has not recorded a progress boundary |
 | `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`. |
diff --git a/src/V2/Support/OperatorMetrics.php b/src/V2/Support/OperatorMetrics.php
@@ -93,6 +93,7 @@ private static function matchingRoleSnapshot(): array
     private static function runMetrics(CarbonInterface $now, ?string $namespace): array
     {
         $oldestWaitStartedAt = self::oldestRunWaitStartedAt($namespace);
+        $oldestRepairNeededAt = self::oldestRepairNeededRunAt($namespace);
 
         return [
             'total' => self::summaryQuery($namespace)->count(),
@@ -104,6 +105,10 @@ private static function runMetrics(CarbonInterface $now, ?string $namespace): ar
             'terminated' => self::summaryQuery($namespace)->where('status', RunStatus::Terminated->value)->count(),
             'archived' => self::summaryQuery($namespace)->whereNotNull('archived_at')->count(),
             'repair_needed' => self::summaryQuery($namespace)->where('liveness_state', 'repair_needed')->count(),
+            'oldest_repair_needed_at' => $oldestRepairNeededAt?->toJSON(),
+            'max_repair_needed_age_ms' => $oldestRepairNeededAt === null
+                ? 0
+                : (int) $oldestRepairNeededAt->diffInMilliseconds($now),
             'claim_failed' => self::claimFailedRuns($namespace),
             'compatibility_blocked' => self::compatibilityBlockedRuns($namespace),
             'waiting' => self::waitingRuns($namespace),
@@ -810,6 +815,39 @@ private static function claimFailedRuns(?string $namespace): int
             ->count();
     }
 
+    /**
+     * Earliest "stuck since" timestamp across runs whose liveness_state is
+     * exactly `repair_needed`. Rollout-safety surfaces this alongside
+     * `runs.repair_needed` so operators can answer "how long has the
+     * worst-case run been stuck without progress?" from the metric alone,
+     * the canonical stuck-workflow age indicator paired with the
+     * `durable_resume_paths` health check.
+     *
+     * The summary's `updated_at` is sourced by `RunSummaryProjector` from
+     * `WorkflowRun::last_progress_at`, so it advances when the run made
+     * forward progress and stalls when the run stopped progressing. For
+     * runs already pinned at `repair_needed` it is therefore the closest
+     * available proxy for "when this run last made progress before being
+     * marked broken." The summary `updated_at` is preferred; the run's
+     * `started_at` is the fallback when the projection did not record a
+     * progress boundary (a fresh run that was projected straight into
+     * `repair_needed` without a prior progress write).
+     */
+    private static function oldestRepairNeededRunAt(?string $namespace): ?CarbonInterface
+    {
+        /** @var WorkflowRunSummary|null $summary */
+        $summary = self::summaryQuery($namespace)
+            ->where('liveness_state', 'repair_needed')
+            ->orderByRaw('COALESCE(updated_at, started_at) asc')
+            ->first();
+
+        if (! $summary instanceof WorkflowRunSummary) {
+            return null;
+        }
+
+        return $summary->updated_at ?? $summary->started_at;
+    }
+
     /**
      * Open runs that are currently parked at a wait point — running runs whose
      * `RunSummaryProjector` has recorded a `wait_started_at` because they are
diff --git a/tests/Feature/V2/V2OperatorMetricsTest.php b/tests/Feature/V2/V2OperatorMetricsTest.php
@@ -1590,6 +1590,109 @@ public function testSnapshotReportsRunWaitAgeAsZeroWhenNoRunsAreWaiting(): void
         $this->assertSame(0, $snapshot['runs']['max_wait_age_ms']);
     }
 
+    public function testSnapshotSurfacesRepairNeededAgeFromOldestRepairNeededRun(): void
+    {
+        Carbon::setTestNow('2026-04-09 12:00:00');
+        $this->beforeApplicationDestroyed(static function (): void {
+            Carbon::setTestNow();
+        });
+
+        $now = Carbon::now();
+
+        // Worst-case: repair_needed run whose summary updated_at (sourced from
+        // last_progress_at) was 90s ago — the oldest stuck-since timestamp wins.
+        $stuckLongest = $this->createRunWithSummary(
+            instanceId: 'repair-needed-age-instance-stuck',
+            runId: '01JREPAIRSTKRUN0000000001',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'repair_needed',
+        );
+        WorkflowRunSummary::query()
+            ->whereKey($stuckLongest->id)
+            ->update([
+                'updated_at' => $now->copy()
+                    ->subSeconds(90),
+            ]);
+
+        // Newer repair_needed run — must be counted, but its 30s-ago updated_at
+        // must not win "oldest repair-needed since".
+        $stuckRecent = $this->createRunWithSummary(
+            instanceId: 'repair-needed-age-instance-recent',
+            runId: '01JREPAIRRECRUN0000000002',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'repair_needed',
+        );
+        WorkflowRunSummary::query()
+            ->whereKey($stuckRecent->id)
+            ->update([
+                'updated_at' => $now->copy()
+                    ->subSeconds(30),
+            ]);
+
+        // Compatibility-blocked run — has a repair_needed-shaped liveness state
+        // (`workflow_task_waiting_for_compatible_worker`) but liveness_state is
+        // NOT exactly `repair_needed`, so it must not be counted under
+        // runs.repair_needed even though its updated_at would win the oldest age.
+        $compatibilityBlocked = $this->createRunWithSummary(
+            instanceId: 'repair-needed-age-instance-compat',
+            runId: '01JREPAIRCMPRUN0000000003',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'workflow_task_waiting_for_compatible_worker',
+        );
+        WorkflowRunSummary::query()
+            ->whereKey($compatibilityBlocked->id)
+            ->update([
+                'updated_at' => $now->copy()
+                    ->subHours(2),
+            ]);
+
+        // Healthy running run — must not be counted regardless of updated_at.
+        $this->createRunWithSummary(
+            instanceId: 'repair-needed-age-instance-running',
+            runId: '01JREPAIRRUNRUN0000000004',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'running',
+        );
+
+        $snapshot = OperatorMetrics::snapshot($now);
+
+        $expectedOldestRepairNeededAt = $now->copy()
+            ->subSeconds(90)
+            ->toJSON();
+
+        $this->assertSame(2, $snapshot['runs']['repair_needed']);
+        $this->assertSame($expectedOldestRepairNeededAt, $snapshot['runs']['oldest_repair_needed_at']);
+        $this->assertSame(90 * 1000, $snapshot['runs']['max_repair_needed_age_ms']);
+    }
+
+    public function testSnapshotReportsRepairNeededAgeAsZeroWhenNoRunsAreRepairNeeded(): void
+    {
+        Carbon::setTestNow('2026-04-09 12:00:00');
+        $this->beforeApplicationDestroyed(static function (): void {
+            Carbon::setTestNow();
+        });
+
+        $now = Carbon::now();
+
+        $this->createRunWithSummary(
+            instanceId: 'repair-needed-none-instance',
+            runId: '01JREPAIRNONRUN0000000001',
+            status: 'running',
+            statusBucket: 'running',
+            livenessState: 'running',
+        );
+
+        $snapshot = OperatorMetrics::snapshot($now);
+
+        $this->assertSame(0, $snapshot['runs']['repair_needed']);
+        $this->assertNull($snapshot['runs']['oldest_repair_needed_at']);
+        $this->assertSame(0, $snapshot['runs']['max_repair_needed_age_ms']);
+    }
+
     public function testSnapshotSurfacesRetryingActivityAgeFromOldestRetryingActivity(): void
     {
         Carbon::setTestNow('2026-04-09 12:00:00');
diff --git a/tests/Unit/V2/RolloutSafetyDocumentationTest.php b/tests/Unit/V2/RolloutSafetyDocumentationTest.php
@@ -140,6 +140,8 @@ final class RolloutSafetyDocumentationTest extends TestCase
         'unhealthy',
         'oldest_unhealthy_at',
         'max_unhealthy_age_ms',
+        'oldest_repair_needed_at',
+        'max_repair_needed_age_ms',
         'runnable_tasks',
         'delayed_tasks',
         'leased_tasks',
@@ -460,6 +462,17 @@ public function testContractDocumentFreezesUnhealthyAgeRollupRow(): void
         );
     }
 
+    public function testContractDocumentFreezesRepairNeededRunAgeRow(): void
+    {
+        $contents = $this->documentContents();
+
+        $this->assertMatchesRegularExpression(
+            '/\|\s*`runs`\s*\|[^|]*`oldest_repair_needed_at`[^|]*`max_repair_needed_age_ms`/',
+            $contents,
+            'Rollout safety contract must pin the runs repair-needed age row so operators can read "how long has the worst-case run been stuck without progress?" — the canonical stuck-workflow duplicate-risk age indicator paired with the durable_resume_paths health check — from OperatorMetrics::snapshot() without walking workflow_run_summaries.',
+        );
+    }
+
     public function testContractDocumentFreezesHealthCheckNames(): void
     {
         $contents = $this->documentContents();
