fix: reconcile dead TaskRunner DO/D1 task lifecycle promptly (Priority 5)

.claude/skills/api-reference/SKILL.md

@@ -60,6 +60,12 @@ user-invocable: false
 - `POST /api/projects/:projectId/tasks/:taskId/delegate` — Delegate ready+unblocked task to owned running workspace
 - `GET /api/projects/:projectId/tasks/:taskId/events` — List append-only task status events
 
+## Administration (Superadmin Only)
+
+- `GET /api/admin/tasks/stuck` — List tasks currently in transient states
+- `GET /api/admin/tasks/:taskId/reconciliation-diagnostics` — Read the TaskRunner probe, task-scoped runtime liveness, eligibility threshold, reconciliation decision, and whether/where the bounded cursor page selects the task, without mutating task state
+- `GET /api/admin/tasks/recent-failures` — List recent failed tasks with error details
+
 ## Agent Sessions
 
 - `GET /api/workspaces/:id/agent-sessions` — List workspace agent sessions

.claude/skills/env-reference/SKILL.md

@@ -88,6 +88,8 @@ See `apps/api/.env.example` for the full list. Key variables:
 - `MAX_PROJECTS_PER_USER` — Runtime project cap
 - `MAX_TASKS_PER_PROJECT` — Runtime task cap per project
 - `MAX_TASK_DEPENDENCIES_PER_TASK` — Runtime dependency-edge cap per task
+- `STUCK_TASK_MAX_CANDIDATES_PER_SWEEP` — Maximum active task rows inspected by one five-minute recovery sweep (default: 100)
+- `STUCK_TASK_SCAN_CURSOR_KV_KEY` — KV key used to resume the bounded recovery scan fairly across active rows (default: `scheduled:stuck-tasks:scan-cursor:v1`)
 - `PROJECT_INVITE_TOKEN_BYTES` — Random bytes used for generated project invite link tokens (default: 32)
 - `PROJECT_INVITE_DEFAULT_EXPIRY_DAYS` — Default lifetime for project invite links created without an explicit expiry (default: 7)
 - `PROJECT_INVITE_MAX_EXPIRY_DAYS` — Maximum allowed project invite link lifetime (default: 30)
@@ -113,6 +115,7 @@ See `apps/api/.env.example` for the full list. Key variables:
 - `TASK_RECONCILIATION_PROMPT_SOFT_STALL_MS` — In-flight prompt observation threshold before SAM records a non-interrupting reconciliation event (default: 1800000)
 - `TASK_RECONCILIATION_PROMPT_HARD_STALL_MS` — In-flight prompt hard-stall threshold before SAM requests prompt cancellation and retries check-in later (default: 7200000)
 - `TASK_RECONCILIATION_MIN_ALARM_DELAY_MS` — Minimum delay before the next reconciliation alarm can fire (default: 10000)
+- `TASK_RUN_ABSOLUTE_CEILING_MS` — Absolute runaway-cost ceiling that fails even a demonstrably live task (default: 86400000 / 24h)
 - `SESSION_ACTIVITY_STALE_THRESHOLD_MS` — Evidence-based fallback threshold before stale working activity can be healed to idle (default: 300000)
 - `NODE_HEARTBEAT_STALE_SECONDS` — Staleness threshold for node health
 - `NODE_AGENT_READY_TIMEOUT_MS` — Max wait for freshly provisioned node-agent health

apps/api/.env.example

@@ -344,8 +344,14 @@ BASE_DOMAIN=workspaces.example.com
 
 # Task execution timeout (stuck task recovery)
 # TASK_RUN_MAX_EXECUTION_MS=14400000     # 4 hours — max time before task is failed
-# TASK_STUCK_QUEUED_TIMEOUT_MS=600000      # 10 minutes — max time in 'queued' state
+# TASK_STUCK_QUEUED_TIMEOUT_MS=1200000     # 20 minutes — max time in 'queued' state
 # TASK_STUCK_DELEGATED_TIMEOUT_MS=1860000  # 31 minutes — max time in 'delegated' state (> workspace ready timeout)
+# TASK_DO_MISMATCH_GRACE_MS=300000          # 5 minutes — minimum age before DO/D1 liveness reconciliation
+# STUCK_TASK_MAX_CANDIDATES_PER_SWEEP=100   # Bound task rows inspected per cron invocation
+# STUCK_TASK_SCAN_CURSOR_KV_KEY=scheduled:stuck-tasks:scan-cursor:v1 # KV cursor for fair bounded scans
+# TASK_LIVENESS_MAX_ACP_SESSIONS=5          # Bound task-scoped ACP sessions inspected per candidate
+# TASK_LIVENESS_PROBE_TIMEOUT_MS=5000       # Per-candidate timeout for the ACP liveness DO probe (inconclusive on timeout, never fatal)
+# TASK_RUN_ABSOLUTE_CEILING_MS=86400000 # 24h — absolute runaway-cost ceiling; fails even a live task
 # CLAUDE_CODE_COMPACTION_LOOP_DETECTOR_ENABLED=true
 # CLAUDE_CODE_COMPACTION_LOOP_RECENT_MESSAGE_LIMIT=40
 # CLAUDE_CODE_COMPACTION_LOOP_WINDOW_MESSAGES=20

apps/api/src/durable-objects/task-runner/state-machine.ts

@@ -121,12 +121,26 @@ export async function transitionToInProgress(
   ).bind(now, now, state.taskId).run();
 
   if (!result.meta.changes || result.meta.changes === 0) {
+    const authoritative = await rc.env.DATABASE.prepare(
+      `SELECT status FROM tasks WHERE id = ?`
+    ).bind(state.taskId).first<{ status: string }>();
     log.warn('task_runner_do.aborted_by_recovery', {
       taskId: state.taskId,
       step: 'in_progress_transition',
+      authoritativeStatus: authoritative?.status ?? null,
     });
-    state.completed = true;
-    await rc.ctx.storage.put('state', state);
+    if (authoritative?.status === 'in_progress') {
+      state.currentStep = 'running';
+      state.completed = true;
+      await rc.ctx.storage.put('state', state);
+      return;
+    }
+    if (!authoritative || ['completed', 'failed', 'cancelled'].includes(authoritative.status)) {
+      state.completed = true;
+      await rc.ctx.storage.put('state', state);
+      return;
+    }
+    await failTask(state, 'Task orchestration was superseded before agent handoff completed.', rc);
     return;
   }
 
@@ -207,9 +221,12 @@ export async function failTask(
     return;
   }
 
-  // Fail the task
+  // Fail the task. The status predicate makes this idempotent against a
+  // concurrent terminal transition that lands between the check above and this
+  // write — never clobber an already-terminal row (completed/failed/cancelled).
   await rc.env.DATABASE.prepare(
-    `UPDATE tasks SET status = 'failed', execution_step = NULL, error_message = ?, completed_at = ?, updated_at = ? WHERE id = ?`
+    `UPDATE tasks SET status = 'failed', execution_step = NULL, error_message = ?, completed_at = ?, updated_at = ?
+     WHERE id = ? AND status NOT IN ('completed', 'failed', 'cancelled')`
   ).bind(errorMessage, now, now, state.taskId).run();
 
   // Sync trigger execution status (best-effort) — without this, cron triggers

apps/api/src/env.ts

@@ -217,6 +217,12 @@ export interface Env {
   TASK_RUN_HARD_TIMEOUT_MS?: string;
   TASK_STUCK_QUEUED_TIMEOUT_MS?: string;
   TASK_STUCK_DELEGATED_TIMEOUT_MS?: string;
+  TASK_DO_MISMATCH_GRACE_MS?: string;
+  STUCK_TASK_MAX_CANDIDATES_PER_SWEEP?: string;
+  STUCK_TASK_SCAN_CURSOR_KV_KEY?: string;
+  TASK_LIVENESS_MAX_ACP_SESSIONS?: string;
+  TASK_LIVENESS_PROBE_TIMEOUT_MS?: string;
+  TASK_RUN_ABSOLUTE_CEILING_MS?: string;
   CLAUDE_CODE_COMPACTION_LOOP_DETECTOR_ENABLED?: string;
   CLAUDE_CODE_COMPACTION_LOOP_RECENT_MESSAGE_LIMIT?: string;
   CLAUDE_CODE_COMPACTION_LOOP_WINDOW_MESSAGES?: string;

apps/api/src/index.ts

@@ -375,11 +375,13 @@ h1{font-size:1.4rem}code{background:#f0f0f0;padding:2px 6px;border-radius:3px;fo
   }
 
   const nodeRuntime = workspace.nodeId
-    ? (await db
-        .select({ runtime: schema.nodes.runtime })
-        .from(schema.nodes)
-        .where(eq(schema.nodes.id, workspace.nodeId))
-        .get())?.runtime ?? 'vm'
+    ? ((
+        await db
+          .select({ runtime: schema.nodes.runtime })
+          .from(schema.nodes)
+          .where(eq(schema.nodes.id, workspace.nodeId))
+          .get()
+      )?.runtime ?? 'vm')
     : 'vm';
 
   if (workspace.status !== 'running' && workspace.status !== 'recovery') {
@@ -400,10 +402,16 @@ h1{font-size:1.4rem}code{background:#f0f0f0;padding:2px 6px;border-radius:3px;fo
   if (nodeRuntime === 'cf-container') {
     const containerConfig = getVmAgentContainerConfig(c.env);
     if (!containerConfig.enabled) {
-      return c.json({ error: 'CF_CONTAINER_DISABLED', message: 'Container workspace runtime is disabled' }, 503);
+      return c.json(
+        { error: 'CF_CONTAINER_DISABLED', message: 'Container workspace runtime is disabled' },
+        503
+      );
     }
     if (!c.env.VM_AGENT_CONTAINER) {
-      return c.json({ error: 'CF_CONTAINER_UNAVAILABLE', message: 'VM agent container binding is unavailable' }, 503);
+      return c.json(
+        { error: 'CF_CONTAINER_UNAVAILABLE', message: 'VM agent container binding is unavailable' },
+        503
+      );
     }
 
     const containerId = workspace.nodeId || workspaceId;
@@ -425,7 +433,10 @@ h1{font-size:1.4rem}code{background:#f0f0f0;padding:2px 6px;border-radius:3px;fo
           workspaceId,
           ...serializeError(err),
         });
-        return c.json({ error: 'TOKEN_ERROR', message: 'Failed to generate port proxy token' }, 500);
+        return c.json(
+          { error: 'TOKEN_ERROR', message: 'Failed to generate port proxy token' },
+          500
+        );
       }
     }
 
@@ -661,7 +672,8 @@ app.get('/api/config/artifacts-enabled', (c) => {
 // only render a provider button when that provider is actually usable. Google
 // here means the LOGIN client (getGoogleLoginOAuthConfig), never the infra/GCP one.
 app.get('/api/config/login-providers', async (c) => {
-  const { getGitHubOAuthConfig, getGoogleLoginOAuthConfig } = await import('./services/platform-config');
+  const { getGitHubOAuthConfig, getGoogleLoginOAuthConfig } =
+    await import('./services/platform-config');
   const [github, google] = await Promise.all([
     getGitHubOAuthConfig(c.env),
     getGoogleLoginOAuthConfig(c.env),
@@ -929,6 +941,10 @@ export default {
     }
 
     // 5-minute operational sweep
+    // Recover stuck tasks first so an unrelated cleanup failure cannot suppress
+    // the task lifecycle safety net for another five-minute interval.
+    const stuckTasks = await recoverStuckTasks(env);
+
     // Check for stuck provisioning workspaces
     const timedOut = await checkProvisioningTimeouts(env.DATABASE, env, env.OBSERVABILITY_DATABASE);
 
@@ -939,9 +955,6 @@ export default {
     // Clean up stale warm nodes and expired auto-provisioned nodes
     const nodeCleanup = await runNodeCleanupSweep(env);
 
-    // Recover stuck tasks (queued/delegated/in_progress past timeout)
-    const stuckTasks = await recoverStuckTasks(env);
-
     // Purge expired observability errors (retention + row count limits)
     const observabilityPurge = await runObservabilityPurge(env);
 
@@ -979,6 +992,13 @@ export default {
       stuckTasksHeartbeatSkipped: stuckTasks.heartbeatSkipped,
       stuckTaskErrors: stuckTasks.errors,
       stuckTaskDoHealthChecked: stuckTasks.doHealthChecked,
+      stuckTaskDoHealthMissing: stuckTasks.doHealthMissing,
+      stuckTaskDoHealthErrors: stuckTasks.doHealthErrors,
+      stuckTaskDeadRuntimeReconciled: stuckTasks.deadRuntimeReconciled,
+      stuckTaskCandidatesScanned: stuckTasks.candidatesScanned,
+      stuckTaskCandidateCursorLoaded: stuckTasks.candidateCursorLoaded,
+      stuckTaskCandidateCursorWrapped: stuckTasks.candidateCursorWrapped,
+      stuckTaskCandidateCursorErrors: stuckTasks.candidateCursorErrors,
       observabilityPurgedByAge: observabilityPurge.deletedByAge,
       observabilityPurgedByCount: observabilityPurge.deletedByCount,
       cronTriggersChecked: cronTriggers.checked,

apps/api/src/routes/admin.ts

@@ -1,17 +1,36 @@
-import type { PlatformErrorLevel,PlatformErrorSource, UserRole, UserStatus } from '@simple-agent-manager/shared';
+import type {
+  PlatformErrorLevel,
+  PlatformErrorSource,
+  UserRole,
+  UserStatus,
+} from '@simple-agent-manager/shared';
 import { desc, eq, inArray } from 'drizzle-orm';
 import { drizzle } from 'drizzle-orm/d1';
 import { Hono } from 'hono';
 
 import * as schema from '../db/schema';
 import type { ProjectData as ProjectDataDO } from '../durable-objects/project-data';
 import type { Env } from '../env';
-import { getUserId,requireApproved, requireAuth, requireSuperadmin } from '../middleware/auth';
+import { getUserId, requireApproved, requireAuth, requireSuperadmin } from '../middleware/auth';
 import { errors } from '../middleware/error';
 import { rateLimit } from '../middleware/rate-limit';
-import { AdminLogQuerySchema,AdminUserActionSchema, AdminUserRoleSchema, jsonValidator, UpdateSignupApprovalConfigSchema } from '../schemas';
+import { getTaskReconciliationDiagnostics } from '../scheduled/stuck-tasks';
+import {
+  AdminLogQuerySchema,
+  AdminUserActionSchema,
+  AdminUserRoleSchema,
+  jsonValidator,
+  UpdateSignupApprovalConfigSchema,
+} from '../schemas';
 import { getRuntimeLimits } from '../services/limits';
-import { CfApiError,getErrorTrends, getHealthSummary, getLogQueryRateLimit, queryCloudflareLogs, queryErrors } from '../services/observability';
+import {
+  CfApiError,
+  getErrorTrends,
+  getHealthSummary,
+  getLogQueryRateLimit,
+  queryCloudflareLogs,
+  queryErrors,
+} from '../services/observability';
 import { getSignupApprovalConfig, setSignupApprovalConfig } from '../services/signup-approval';
 
 const adminRoutes = new Hono<{ Bindings: Env }>();
@@ -196,6 +215,18 @@ adminRoutes.get('/tasks/stuck', async (c) => {
   return c.json({ tasks: tasksWithAge });
 });
 
+/**
+ * GET /api/admin/tasks/:taskId/reconciliation-diagnostics - Explain the
+ * read-only evidence and decision used by scheduled task reconciliation.
+ */
+adminRoutes.get('/tasks/:taskId/reconciliation-diagnostics', async (c) => {
+  const { taskId } = c.req.param();
+  const diagnostics = await getTaskReconciliationDiagnostics(c.env, taskId);
+
+  if (!diagnostics) throw errors.notFound('Task');
+  return c.json({ diagnostics });
+});
+
 /**
  * GET /api/admin/tasks/recent-failures - List recently failed tasks with error details
  *
@@ -272,8 +303,8 @@ adminRoutes.get('/observability/errors', async (c) => {
   }
 
   const result = await queryErrors(c.env.OBSERVABILITY_DATABASE, {
-    source: source && source !== 'all' ? source as PlatformErrorSource : undefined,
-    level: level && level !== 'all' ? level as PlatformErrorLevel : undefined,
+    source: source && source !== 'all' ? (source as PlatformErrorSource) : undefined,
+    level: level && level !== 'all' ? (level as PlatformErrorLevel) : undefined,
     search: search || undefined,
     startTime: startTime ? new Date(startTime).getTime() : undefined,
     endTime: endTime ? new Date(endTime).getTime() : undefined,
@@ -327,7 +358,8 @@ adminRoutes.get('/observability/trends', async (c) => {
  *
  * Body: { timeRange: { start, end }, levels?, search?, limit?, cursor? }
  */
-adminRoutes.post('/observability/logs/query',
+adminRoutes.post(
+  '/observability/logs/query',
   // Per-admin KV-based rate limiting (1-minute window)
   async (c, next) => {
     const limiter = rateLimit({
@@ -339,54 +371,59 @@ adminRoutes.post('/observability/logs/query',
   },
   jsonValidator(AdminLogQuerySchema),
   async (c) => {
-  if (!c.env.CF_API_TOKEN || !c.env.CF_ACCOUNT_ID) {
-    throw errors.badRequest('Cloudflare API credentials not configured. Set CF_API_TOKEN and CF_ACCOUNT_ID.');
-  }
+    if (!c.env.CF_API_TOKEN || !c.env.CF_ACCOUNT_ID) {
+      throw errors.badRequest(
+        'Cloudflare API credentials not configured. Set CF_API_TOKEN and CF_ACCOUNT_ID.'
+      );
+    }
 
-  const body = c.req.valid('json');
+    const body = c.req.valid('json');
 
-  // Validate dates
-  const startDate = new Date(body.timeRange.start);
-  const endDate = new Date(body.timeRange.end);
-  if (isNaN(startDate.getTime()) || isNaN(endDate.getTime())) {
-    throw errors.badRequest('timeRange start and end must be valid ISO 8601 dates');
-  }
+    // Validate dates
+    const startDate = new Date(body.timeRange.start);
+    const endDate = new Date(body.timeRange.end);
+    if (isNaN(startDate.getTime()) || isNaN(endDate.getTime())) {
+      throw errors.badRequest('timeRange start and end must be valid ISO 8601 dates');
+    }
 
-  // Validate levels
-  if (body.levels) {
-    const validLogLevels = new Set(['error', 'warn', 'info', 'debug', 'log']);
-    for (const level of body.levels) {
-      if (!validLogLevels.has(level)) {
-        throw errors.badRequest(`Invalid level: ${level}. Must be one of: error, warn, info, debug, log`);
+    // Validate levels
+    if (body.levels) {
+      const validLogLevels = new Set(['error', 'warn', 'info', 'debug', 'log']);
+      for (const level of body.levels) {
+        if (!validLogLevels.has(level)) {
+          throw errors.badRequest(
+            `Invalid level: ${level}. Must be one of: error, warn, info, debug, log`
+          );
+        }
       }
     }
-  }
 
-  // Validate limit
-  if (body.limit !== undefined && (body.limit < 1 || body.limit > 500)) {
-    throw errors.badRequest('limit must be between 1 and 500');
-  }
+    // Validate limit
+    if (body.limit !== undefined && (body.limit < 1 || body.limit > 500)) {
+      throw errors.badRequest('limit must be between 1 and 500');
+    }
 
-  try {
-    const result = await queryCloudflareLogs({
-      cfApiToken: c.env.CF_API_TOKEN,
-      cfAccountId: c.env.CF_ACCOUNT_ID,
-      timeRange: { start: body.timeRange.start, end: body.timeRange.end },
-      levels: body.levels ?? undefined,
-      search: body.search || undefined,
-      limit: body.limit,
-      cursor: body.cursor || undefined,
-      queryId: body.queryId || undefined,
-    });
+    try {
+      const result = await queryCloudflareLogs({
+        cfApiToken: c.env.CF_API_TOKEN,
+        cfAccountId: c.env.CF_ACCOUNT_ID,
+        timeRange: { start: body.timeRange.start, end: body.timeRange.end },
+        levels: body.levels ?? undefined,
+        search: body.search || undefined,
+        limit: body.limit,
+        cursor: body.cursor || undefined,
+        queryId: body.queryId || undefined,
+      });
 
-    return c.json(result);
-  } catch (err) {
-    if (err instanceof CfApiError) {
-      return c.json({ error: 'CF_API_ERROR', message: err.message }, 502);
+      return c.json(result);
+    } catch (err) {
+      if (err instanceof CfApiError) {
+        return c.json({ error: 'CF_API_ERROR', message: err.message }, 502);
+      }
+      throw err;
     }
-    throw err;
   }
-});
+);
 
 /**
  * GET /api/admin/observability/logs/stream - WebSocket upgrade for real-time log stream
@@ -465,7 +502,10 @@ adminRoutes.post('/backfill-session-summaries', async (c) => {
       const stub = c.env.PROJECT_DATA.get(doId) as DurableObjectStub<ProjectDataDO>;
 
       // List all sessions from the DO (up to 1000)
-      const result = await stub.listSessions(null, 1000, 0) as { sessions: Record<string, unknown>[]; total: number };
+      const result = (await stub.listSessions(null, 1000, 0)) as {
+        sessions: Record<string, unknown>[];
+        total: number;
+      };
 
       if (result.sessions.length === 0) continue;