feat(dag): node timeout escalation — signal, adjudication, cap backstop

package.json

@@ -12,7 +12,7 @@
     "dev:console": "ulimit -n 10240 2>/dev/null; bun run --cwd packages/console/app dev",
     "dev:stats": "bun sst shell --stage=production -- bun run --cwd packages/stats/app dev",
     "dev:storybook": "bun --cwd packages/storybook storybook",
-    "lint": "oxlint --max-warnings=4734",
+    "lint": "oxlint --max-warnings=4831",
     "typecheck": "bun turbo typecheck",
     "upgrade-opentui": "bun run script/upgrade-opentui.ts",
     "postinstall": "bun run --cwd packages/core fix-node-pty",

packages/core/schema.json

@@ -1,9 +1,9 @@
 {
   "version": "7",
   "dialect": "sqlite",
-  "id": "442cdbd5-86a8-41a9-86d6-5361dbac90e0",
+  "id": "abdf5c23-7f2e-4ca3-b08b-012db47b5aa5",
   "prevIds": [
-    "a953899b-bb63-497e-8e2e-86eb5e0fdeed"
+    "442cdbd5-86a8-41a9-86d6-5361dbac90e0"
   ],
   "ddl": [
     {
@@ -658,6 +658,26 @@
       "entityType": "columns",
       "table": "workflow_node"
     },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "timeout_extensions",
+      "entityType": "columns",
+      "table": "workflow_node"
+    },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "false",
+      "generated": null,
+      "name": "escalation_pending",
+      "entityType": "columns",
+      "table": "workflow_node"
+    },
     {
       "type": "integer",
       "notNull": true,

packages/core/src/dag/core/replan.ts

@@ -186,15 +186,15 @@ export function planReplan(
   for (const n of current.nodes) {
     if (!survivingIds.has(n.id)) continue
     const frag = fragmentNodeById.get(n.id)
-    // A node takes the fragment's deps only when it is actually being replaced
-    // (pending/queued/paused) or restarted (running with restart marker). A
-    // running node present without a marker is "kept unchanged" and keeps its
-    // current deps; terminal nodes are immutable and keep their current deps.
-    if (frag && (frag.restart || (n.status !== NodeStatus.RUNNING && !isNodeTerminalStatus(n.status)))) {
-      for (const depId of frag.depends_on) tryAddEdge(n.id, depId)
-    } else {
-      for (const depId of n.depends_on) tryAddEdge(n.id, depId)
-    }
+    // P1a: the CHECK graph must equal the EXECUTION graph. A running node
+    // present without a restart marker is replaced (its definition is
+    // re-published via NodeRegistered and the projector upserts the fragment's
+    // depends_on into the durable row the runtime rebuilds from), so it takes
+    // the fragment's deps here too — otherwise a cycle only reachable through
+    // the replaced deps passes the check and crashes the runtime's
+    // rebuildGraph. Terminal nodes are immutable and keep their current deps.
+    const deps = frag && !isNodeTerminalStatus(n.status) ? frag.depends_on : n.depends_on
+    for (const depId of deps) tryAddEdge(n.id, depId)
   }
   for (const fragNode of fragment.nodes) {
     if (currentStateById.has(fragNode.id)) continue // handled above
@@ -226,7 +226,15 @@ export function planReplan(
       continue
     }
     if (n.status === NodeStatus.RUNNING) {
-      if (frag?.restart) restart.push(n.id)
+      if (frag?.restart) {
+        restart.push(n.id)
+        continue
+      }
+      // A running node present in the fragment (no restart marker) gets its
+      // definition replaced without re-executing — this is the timeout
+      // extension path: the merged config carries the new worker_config.timeout_ms,
+      // and the runtime recomputes the absolute deadline from it.
+      if (frag) replace.push(n.id)
       continue
     }
     if (n.status === NodeStatus.PENDING) {

packages/core/src/dag/projector.ts

@@ -235,6 +235,12 @@ export const layer = Layer.effectDiscard(
           deadline_ms: event.data.deadlineMs ?? null,
           wake_eligible: event.data.wakeEligible ?? false,
           wake_reported: false,
+          // S3: a fresh execution attempt starts with a fresh extension budget —
+          // restart clears timeout_extensions so the cap is per-attempt, not
+          // lifetime (a restart must not inherit a nearly-exhausted cap). The
+          // new attempt is also not awaiting adjudication.
+          timeout_extensions: 0,
+          escalation_pending: false,
           seq: event.durable!.seq,
           time_updated: toMillis(event.data.timestamp),
         })
@@ -260,6 +266,12 @@ export const layer = Layer.effectDiscard(
           status: "completed",
           output: event.data.output,
           completed_at: toMillis(event.data.timestamp),
+          // F2b: re-arm wake delivery on every status migration. A node whose
+          // escalated wake was already reported (wake_reported=true) must
+          // re-enter the snapshot on completion/failure — otherwise its result
+          // notification (and the crash-recovery failure at recovery.ts) is
+          // lost behind the earlier escalation notification.
+          wake_reported: false,
           seq: event.durable!.seq,
           time_updated: toMillis(event.data.timestamp),
         })
@@ -284,6 +296,10 @@ export const layer = Layer.effectDiscard(
           error_reason: event.data.reason,
           error_class: event.data.trigger,
           completed_at: toMillis(event.data.timestamp),
+          // F2b: same re-arm as NodeCompleted — a failure after a reported
+          // escalation (or a crash-recovery failure of an escalated node)
+          // must still reach the main agent.
+          wake_reported: false,
           seq: event.durable!.seq,
           time_updated: toMillis(event.data.timestamp),
         })
@@ -339,6 +355,11 @@ export const layer = Layer.effectDiscard(
           // abort the old session before spawning the replacement. NodeStarted
           // will overwrite it with the new child session.
           replan_attempts: sql`${WorkflowNodeTable.replan_attempts} + 1`,
+          // S3: restart opens a new attempt — reset the extension budget and
+          // clear the pending-adjudication flag so the cap and the summary are
+          // per-attempt, not lifetime.
+          timeout_extensions: 0,
+          escalation_pending: false,
           seq: event.durable!.seq,
           time_updated: toMillis(event.data.timestamp),
         })
@@ -353,6 +374,35 @@ export const layer = Layer.effectDiscard(
         .run()
         .pipe(Effect.orDie),
     )
+
+    // Timeout escalation: the node stays RUNNING (no status transition). Only
+    // the extension count, seq, and wake flag change. wake_reported is reset so
+    // the escalated node re-enters the wake snapshot and the main agent is
+    // notified once per escalation.
+    yield* events.project(DagEvent.NodeTimeoutEscalated, (event) =>
+      db
+        .update(WorkflowNodeTable)
+        .set({
+          timeout_extensions: event.data.timeoutExtensions,
+          // The escalation is not yet adjudicated — summary and the wake
+          // delivery boundary treat the node as awaiting main-agent action
+          // until an extend (updateNodeDeadline) or a new attempt clears it.
+          escalation_pending: true,
+          wake_reported: false,
+          seq: event.durable!.seq,
+          time_updated: toMillis(event.data.timestamp),
+        })
+        // F2a: escalate only live running nodes — a stale escalate racing a
+        // terminal event must not resurrect the wake flag or inflate the
+        // counter on an already-completed/failed node (ghost wake).
+        .where(and(
+          eq(WorkflowNodeTable.workflow_id, event.data.dagID),
+          eq(WorkflowNodeTable.id, event.data.nodeID),
+          inArray(WorkflowNodeTable.status, ["running"]),
+        ))
+        .run()
+        .pipe(Effect.orDie),
+    )
   }),
 )
 

packages/core/src/dag/sql.ts

@@ -68,6 +68,8 @@ export const WorkflowNodeTable = sqliteTable(
     wake_eligible: integer({ mode: "boolean" }).notNull().default(false), // D6: node has report_to_parent=true
     wake_reported: integer({ mode: "boolean" }).notNull().default(false), // D3: has this node's terminal event been injected into the parent session?
     replan_attempts: integer().notNull().default(0), // D4: per-node replan counter for circuit breaker
+    timeout_extensions: integer().notNull().default(0), // timeout escalation count (node stays running; main agent adjudicates)
+    escalation_pending: integer({ mode: "boolean" }).notNull().default(false), // set on escalate, cleared on adjudication (extend) or new attempt — "awaiting main-agent adjudication"
     seq: integer().notNull(), // latest durable event seq for this node
     started_at: integer(),
     completed_at: integer(),

packages/core/src/dag/store.ts

@@ -1,6 +1,6 @@
 export * as DagStore from "./store"
 
-import { and, asc, count, desc, eq, inArray } from "drizzle-orm"
+import { and, asc, count, desc, eq, gt, inArray, or } from "drizzle-orm"
 import { Context, Effect, Layer } from "effect"
 import { Database } from "../database/database"
 import { LayerNode } from "../effect/layer-node"
@@ -44,6 +44,8 @@ export interface NodeRow {
   wakeEligible: boolean
   wakeReported: boolean
   replanAttempts: number
+  timeoutExtensions: number
+  escalationPending: boolean
   seq: number
   startedAt: number | null
   completedAt: number | null
@@ -70,6 +72,8 @@ export interface WorkflowSummary {
   failedNodes: number
   skippedNodes: number
   queuedNodes: number
+  /** Running nodes with a not-yet-adjudicated timeout escalation (escalation_pending). */
+  escalatedNodes: number
 }
 
 const mapWorkflow = (r: typeof WorkflowTable.$inferSelect): WorkflowRow => ({
@@ -106,11 +110,32 @@ const mapNode = (r: typeof WorkflowNodeTable.$inferSelect): NodeRow => ({
   wakeEligible: r.wake_eligible,
   wakeReported: r.wake_reported,
   replanAttempts: r.replan_attempts,
+  timeoutExtensions: r.timeout_extensions,
+  escalationPending: r.escalation_pending,
   seq: r.seq,
   startedAt: r.started_at,
   completedAt: r.completed_at,
 })
 
+// F11: wake eligibility gates TERMINAL notifications (the report_to_parent
+// contract) — but a timeout-escalated node must reach the main agent
+// REGARDLESS of report_to_parent AND regardless of its current status: the
+// escalation wake is the only force behind the extension cap, and a
+// non-eligible node (default config) would otherwise never be adjudicated.
+// Escalated-then-terminalized nodes (cap-exhausted force-cancel) still need
+// delivery, and so do escalated-then-COMPLETED nodes: the main agent already
+// spent turns adjudicating this node (the extend path), so its result is the
+// receipt for those turns — withholding it behind report_to_parent would
+// silently lose adjudicated work. Single source of truth for snapshot /
+// unreported / bootstrap-sweep wake queries.
+const wakeDeliverableNodePredicate = or(
+  and(
+    eq(WorkflowNodeTable.wake_eligible, true),
+    inArray(WorkflowNodeTable.status, ["completed", "failed"]),
+  ),
+  gt(WorkflowNodeTable.timeout_extensions, 0),
+)
+
 // ============================================================================
 // Service interface
 // ============================================================================
@@ -127,6 +152,7 @@ export interface Interface {
   readonly getNode: (workflowId: string, nodeId: string) => Effect.Effect<NodeRow | undefined>
   readonly getRunningNodes: (workflowId: string) => Effect.Effect<NodeRow[]>
   readonly setCapturedOutput: (childSessionID: string, payload: unknown) => Effect.Effect<void>
+  readonly updateNodeDeadline: (workflowId: string, nodeID: string, deadlineMs: number) => Effect.Effect<number>
 
   readonly markNodeWakeReported: (workflowId: string, nodeID: string) => Effect.Effect<void>
   readonly markWorkflowWakeReported: (dagID: string) => Effect.Effect<void>
@@ -212,6 +238,26 @@ export const layer = Layer.effect(
           .groupBy(WorkflowNodeTable.workflow_id, WorkflowNodeTable.status)
           .all()
           .pipe(Effect.orDie)
+        // F10: separate aggregation for running nodes with a not-yet-adjudicated
+        // timeout escalation (the status grouping above cannot see the flag).
+        // escalation_pending is set on escalate and cleared on adjudication, so
+        // this counts only nodes genuinely awaiting main-agent action.
+        const escalatedRows = yield* db
+          .select({
+            workflowId: WorkflowNodeTable.workflow_id,
+            total: count(),
+          })
+          .from(WorkflowNodeTable)
+          .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id))
+          .where(and(
+            eq(WorkflowTable.session_id, sessionId),
+            eq(WorkflowNodeTable.status, "running"),
+            eq(WorkflowNodeTable.escalation_pending, true),
+          ))
+          .groupBy(WorkflowNodeTable.workflow_id)
+          .all()
+          .pipe(Effect.orDie)
+        const escalatedByWorkflow = new Map(escalatedRows.map((row) => [row.workflowId, row.total]))
         const counts = countRows.reduce((all, row) => {
           const current = all.get(row.workflowId) ?? { nodeCount: 0, completedNodes: 0, runningNodes: 0, failedNodes: 0, skippedNodes: 0, queuedNodes: 0 }
           current.nodeCount += row.total
@@ -228,6 +274,7 @@ export const layer = Layer.effect(
           title: wf.title,
           status: wf.status,
           ...(counts.get(wf.id) ?? { nodeCount: 0, completedNodes: 0, runningNodes: 0, failedNodes: 0, skippedNodes: 0, queuedNodes: 0 }),
+          escalatedNodes: escalatedByWorkflow.get(wf.id) ?? 0,
         }))
       }),
 
@@ -271,6 +318,30 @@ export const layer = Layer.effect(
           .pipe(Effect.orDie)
       }),
 
+      updateNodeDeadline: Effect.fn("DagStore.updateNodeDeadline")(function* (workflowId, nodeID, deadlineMs) {
+        const updated = yield* db
+          .update(WorkflowNodeTable)
+          // Adjudication write (re-time via nodeExtendTimeout). Only update the
+          // deadline — do NOT reset timeout_extensions: the count is cumulative
+          // per attempt so an agent cannot bypass the cap by re-planning.
+          // Escalation is now adjudicated: clear escalation_pending (summary and
+          // delivery boundary stop treating the node as awaiting adjudication)
+          // and consume the escalation wake (wake_reported=true) so the stale
+          // timeout wake is not re-delivered after the deadline moved.
+          .set({ deadline_ms: deadlineMs, escalation_pending: false, wake_reported: true })
+          // Guard: never write a deadline onto a node that terminalized between
+          // the caller's read and this update.
+          .where(and(
+            eq(WorkflowNodeTable.workflow_id, workflowId),
+            eq(WorkflowNodeTable.id, nodeID),
+            eq(WorkflowNodeTable.status, "running"),
+          ))
+          .returning({ id: WorkflowNodeTable.id })
+          .all()
+          .pipe(Effect.orDie)
+        return updated.length
+      }),
+
       markNodeWakeReported: Effect.fn("DagStore.markNodeWakeReported")(function* (workflowId, nodeID) {
         yield* db
           .update(WorkflowNodeTable)
@@ -337,9 +408,14 @@ export const layer = Layer.effect(
                 .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id))
                 .where(and(
                   eq(WorkflowTable.session_id, sessionID),
-                  eq(WorkflowNodeTable.wake_eligible, true),
                   eq(WorkflowNodeTable.wake_reported, false),
-                  inArray(WorkflowNodeTable.status, ["completed", "failed"]),
+                  // Escalated nodes enter the snapshot unconditionally
+                  // (timeout_extensions > 0), covering the escalated-then-
+                  // terminal outcome too — the cap's terminal verdict is its
+                  // enforceable force. Adjudication consumes the wake
+                  // (wake_reported=true), so adjudicated nodes are already
+                  // filtered out above.
+                  wakeDeliverableNodePredicate,
                 ))
                 .orderBy(
                   asc(WorkflowTable.seq),
@@ -370,9 +446,10 @@ export const layer = Layer.effect(
           .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id))
           .where(and(
             eq(WorkflowTable.session_id, sessionID),
-            eq(WorkflowNodeTable.wake_eligible, true),
             eq(WorkflowNodeTable.wake_reported, false),
-            inArray(WorkflowNodeTable.status, ["completed", "failed"]),
+            // See wakeDeliverableNodePredicate — escalated nodes are wake-
+            // eligible regardless of report_to_parent and status.
+            wakeDeliverableNodePredicate,
           ))
           .orderBy(
             asc(WorkflowTable.seq),
@@ -416,9 +493,11 @@ export const layer = Layer.effect(
           .from(WorkflowNodeTable)
           .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id))
           .where(and(
-            eq(WorkflowNodeTable.wake_eligible, true),
             eq(WorkflowNodeTable.wake_reported, false),
-            inArray(WorkflowNodeTable.status, ["completed", "failed"]),
+            // See wakeDeliverableNodePredicate — escalated nodes count as
+            // unreported wakes so the bootstrap sweep finds a session whose
+            // only outstanding item is an escalation.
+            wakeDeliverableNodePredicate,
           ))
           .all()
           .pipe(Effect.orDie)

packages/core/src/database/migration/20260805094941_workflow_node_timeout_extensions.ts

@@ -0,0 +1,11 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260805094941_workflow_node_timeout_extensions",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE \`workflow_node\` ADD \`timeout_extensions\` integer DEFAULT 0 NOT NULL;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260805094942_workflow_node_escalation_pending.ts

@@ -0,0 +1,14 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+// Separate migration id from the timeout_extensions ALTER: the runner
+// applies each migration at most once keyed by id, so a DB that already ran
+// the timeout_extensions migration must still pick up this column.
+export default {
+  id: "20260805094942_workflow_node_escalation_pending",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE \`workflow_node\` ADD \`escalation_pending\` integer DEFAULT false NOT NULL;`)
+    })
+  },
+} satisfies DatabaseMigration.Migration