From 99dbef667edd3676ee27fa44ced27cf88a4efacd Mon Sep 17 00:00:00 2001
From: Richard R <me@richardr.dev>
Date: Wed, 3 Jun 2026 01:56:54 -0600
Subject: [PATCH] feat(worker): extend orphan recovery to handle whisper_align
 ops and improve logging

Update orphan recovery to detect and fail stale 'running' whisper_align operations
in addition to pdf_layout. Refactor recovery logic to generalize staleness checks
across operation kinds and enhance log output with detailed operation info.
Expand tests to verify correct handling of both whisper_align and pdf_layout
operations in running and queued states.
---
 compute/worker/src/runtime.ts           | 26 ++++++++-----
 compute/worker/tests/api/routes.test.ts | 49 +++++++++++++++++++------
 2 files changed, 55 insertions(+), 20 deletions(-)

diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts
index d535543..fb89764 100644
--- a/compute/worker/src/runtime.ts
+++ b/compute/worker/src/runtime.ts
@@ -753,17 +753,21 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti
     orphanRecoveryPromise = (async () => {
       const now = Date.now();
       const states = await operationStateStore.listOpStates!();
-      const stalePdfStates = states.filter((state) => {
-        if (state.kind !== 'pdf_layout' || !isInflightStatus(state.status)) return false;
+      const staleStates = states.filter((state) => {
+        if (!isInflightStatus(state.status)) return false;
         const ageMs = now - state.updatedAt;
         if (state.status === 'running') {
-          return ageMs > pdfTimeoutMs;
+          const runningTimeoutMs = state.kind === 'whisper_align' ? whisperTimeoutMs : pdfTimeoutMs;
+          return ageMs > runningTimeoutMs;
         }
+        if (state.kind !== 'pdf_layout') return false;
         return ageMs > opStaleMs;
       });
 
-      for (const state of stalePdfStates) {
-        const staleAfterMs = state.status === 'running' ? pdfTimeoutMs : opStaleMs;
+      for (const state of staleStates) {
+        const staleAfterMs = state.status === 'running'
+          ? (state.kind === 'whisper_align' ? whisperTimeoutMs : pdfTimeoutMs)
+          : opStaleMs;
         await orchestrator.markFailed({
           opId: state.opId,
           error: {
@@ -774,11 +778,15 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti
         });
       }
 
-      if (stalePdfStates.length > 0) {
+      if (staleStates.length > 0) {
         app.log.warn({
-          recoveredCount: stalePdfStates.length,
-          opIds: stalePdfStates.map((state) => state.opId),
-        }, 'recovered stale in-flight pdf operations on startup');
+          recoveredCount: staleStates.length,
+          ops: staleStates.map((state) => ({
+            opId: state.opId,
+            kind: state.kind,
+            status: state.status,
+          })),
+        }, 'recovered stale in-flight operations on startup');
       }
 
       orphanRecoveryDoneForGeneration = sessionGeneration;
diff --git a/compute/worker/tests/api/routes.test.ts b/compute/worker/tests/api/routes.test.ts
index d5d2745..6ba641d 100644
--- a/compute/worker/tests/api/routes.test.ts
+++ b/compute/worker/tests/api/routes.test.ts
@@ -120,22 +120,40 @@ describe('compute worker API routes', () => {
     expect(stream.body).toContain('"status":"succeeded"');
   });
 
-  test('marks stale running pdf ops failed during startup reconciliation but leaves queued ops on the conservative path', async () => {
+  test('marks stale running whisper and pdf ops failed during startup reconciliation but leaves queued ops on the conservative path', async () => {
     const now = Date.now();
     fake.seedState({
-      opId: 'op-stale-running',
-      opKey: 'k-stale-running',
+      opId: 'op-stale-whisper-running',
+      opKey: 'k-stale-whisper-running',
+      kind: 'whisper_align',
+      jobId: 'job-op-stale-whisper-running',
+      status: 'running',
+      queuedAt: 1,
+      updatedAt: now - 40_000,
+    });
+    fake.seedState({
+      opId: 'op-stale-whisper-queued',
+      opKey: 'k-stale-whisper-queued',
+      kind: 'whisper_align',
+      jobId: 'job-op-stale-whisper-queued',
+      status: 'queued',
+      queuedAt: 1,
+      updatedAt: now - 40_000,
+    });
+    fake.seedState({
+      opId: 'op-stale-pdf-running',
+      opKey: 'k-stale-pdf-running',
       kind: 'pdf_layout',
-      jobId: 'job-op-stale-running',
+      jobId: 'job-op-stale-pdf-running',
       status: 'running',
       queuedAt: 1,
       updatedAt: now - 310_000,
     });
     fake.seedState({
-      opId: 'op-stale-queued',
-      opKey: 'k-stale-queued',
+      opId: 'op-stale-pdf-queued',
+      opKey: 'k-stale-pdf-queued',
       kind: 'pdf_layout',
-      jobId: 'job-op-stale-queued',
+      jobId: 'job-op-stale-pdf-queued',
       status: 'queued',
       queuedAt: 1,
       updatedAt: now - 310_000,
@@ -143,25 +161,34 @@ describe('compute worker API routes', () => {
 
     const fetch = await runtime.app.inject({
       method: 'GET',
-      url: '/ops/op-stale-running',
+      url: '/ops/op-stale-whisper-running',
       headers: AUTH,
     });
 
     expect(fetch.statusCode).toBe(200);
     expect(fetch.json()).toMatchObject({
-      opId: 'op-stale-running',
+      opId: 'op-stale-whisper-running',
       status: 'failed',
       error: {
         code: 'WORKER_ORPHANED_OP',
       },
     });
-    expect(fake.getState('op-stale-running')).toMatchObject({
+    expect(fake.getState('op-stale-whisper-running')).toMatchObject({
       status: 'failed',
       error: {
         code: 'WORKER_ORPHANED_OP',
       },
     });
-    expect(fake.getState('op-stale-queued')).toMatchObject({
+    expect(fake.getState('op-stale-whisper-queued')).toMatchObject({
+      status: 'queued',
+    });
+    expect(fake.getState('op-stale-pdf-running')).toMatchObject({
+      status: 'failed',
+      error: {
+        code: 'WORKER_ORPHANED_OP',
+      },
+    });
+    expect(fake.getState('op-stale-pdf-queued')).toMatchObject({
       status: 'queued',
     });
   });