feat: recover stale running agent runs

agent-run-gateway.mjs

@@ -493,6 +493,17 @@ export function createAgentRunGateway({
     for (const row of rows) {
       statusCounts[row.status] = Number(row.count ?? 0);
     }
+    const [oldestRunningRows] = await pool.query(
+      `SELECT MIN(started_at) AS oldest_started_at
+       FROM h5_agent_runs
+       WHERE status = 'running'`,
+    );
+    const oldestRunningStartedAt = oldestRunningRows[0]?.oldest_started_at == null
+      ? null
+      : Number(oldestRunningRows[0].oldest_started_at);
+    const oldestRunningAgeMs = oldestRunningStartedAt == null
+      ? 0
+      : Math.max(0, nowMs() - oldestRunningStartedAt);
     return {
       autoDispatch,
       maxConcurrentRuns,
@@ -500,18 +511,86 @@ export function createAgentRunGateway({
       inFlight: inFlight.size,
       pendingDispatches: queuedDispatches.length,
       statusCounts,
+      oldestRunningStartedAt,
+      oldestRunningAgeMs,
       terminalStatuses: [...TERMINAL_STATUSES],
       toolGateway: toolGateway?.getStatus ? toolGateway.getStatus() : null,
     };
   }
 
+  async function recoverStaleRunningRuns({
+    staleMs = runTimeoutMs,
+    limit = maxConcurrentRuns,
+    dryRun = true,
+    reason = 'stale_running_timeout',
+  } = {}) {
+    const normalizedStaleMs = positiveInteger(staleMs, runTimeoutMs);
+    const normalizedLimit = positiveInteger(limit, maxConcurrentRuns);
+    const cutoff = nowMs() - normalizedStaleMs;
+    const [rows] = await pool.query(
+      `SELECT id, request_id, started_at, updated_at, attempts
+       FROM h5_agent_runs
+       WHERE status = 'running' AND started_at IS NOT NULL AND started_at <= ?
+       ORDER BY started_at ASC
+       LIMIT ?`,
+      [cutoff, normalizedLimit],
+    );
+    const recovered = [];
+    for (const row of rows) {
+      const ageMs = Math.max(0, nowMs() - Number(row.started_at ?? 0));
+      const item = {
+        id: row.id,
+        requestId: row.request_id,
+        startedAt: Number(row.started_at ?? 0),
+        updatedAt: Number(row.updated_at ?? 0),
+        attempts: Number(row.attempts ?? 0),
+        ageMs,
+        reason,
+      };
+      if (!dryRun) {
+        const message = `agent run recovered from stale running state after ${ageMs}ms`;
+        const completedAt = nowMs();
+        const [update] = await pool.query(
+          `UPDATE h5_agent_runs
+           SET status = 'failed', error_message = ?, updated_at = ?, completed_at = ?
+           WHERE id = ? AND status = 'running' AND started_at <= ?`,
+          [message, completedAt, completedAt, row.id, cutoff],
+        );
+        if (Number(update?.affectedRows ?? 0) === 0) continue;
+        await appendEvent(row.id, 'stale_recovered', {
+          reason,
+          staleMs: normalizedStaleMs,
+          ageMs,
+          status: 'failed',
+          error: message,
+        });
+      }
+      recovered.push(item);
+    }
+    return {
+      dryRun,
+      staleMs: normalizedStaleMs,
+      cutoff,
+      considered: rows.length,
+      recovered: dryRun ? 0 : recovered.length,
+      runs: recovered,
+    };
+  }
+
   async function dispatchQueuedRuns({ limit = maxConcurrentRuns } = {}) {
+    const staleRecovery = await recoverStaleRunningRuns({
+      staleMs: runTimeoutMs,
+      limit: Math.max(maxConcurrentRuns, Number(limit) || maxConcurrentRuns),
+      dryRun: false,
+      reason: 'worker_dispatch_stale_recovery',
+    });
     const dispatchLimit = Math.max(1, Number(limit) || maxConcurrentRuns);
     const available = Math.max(0, maxConcurrentRuns - inFlight.size - queuedDispatches.length);
     if (available <= 0) {
       return {
         considered: 0,
         dispatched: 0,
+        staleRecovery,
         queue: await getQueueStatus(),
       };
     }
@@ -530,6 +609,7 @@ export function createAgentRunGateway({
     return {
       considered: rows.length,
       dispatched: rows.length,
+      staleRecovery,
       queue: await getQueueStatus(),
     };
   }
@@ -540,5 +620,6 @@ export function createAgentRunGateway({
     dispatchRun,
     getQueueStatus,
     dispatchQueuedRuns,
+    recoverStaleRunningRuns,
   };
 }