fix(dag): batch A residuals — zero tech debt (phantom cancelled state, stale-read budget, lint ratchet restored)

.opencode/grill-batch-a/node-lifecycle-transitions.md

@@ -8,7 +8,9 @@
 
 ## 状态空间
 
-**主状态**（workflow_node.status）：`pending` / `queued` / `running` / 终态 `completed` / `failed` / `cancelled` / `skipped`
+**主状态**（workflow_node.status）：`pending` / `queued` / `running` / 终态 `completed` / `failed` / `skipped`
+
+> **节点级无独立 `cancelled` 终态**（method-A 对齐实现）：`NodeCancelled` 事件投影为 `status=failed` + `error_reason='cancelled via replan'`，取消语义经 error_reason 承载，行永不持有 `status='cancelled'`（`NodeStatus` 枚举无 CANCELLED，`getValidNextNodeStatuses` 对任何 from 均不返回 cancelled）。工作流级 `cancelled`（`WorkflowStatusProjection.cancelled`）是合法独立终态，与节点级无关。见 T5。
 
 **running 扩展维度**（子状态）：
 | 维度 | 语义 | 契约来源 |
@@ -26,7 +28,7 @@
 | T2 | queued | nodeStarted | runtime spawn | running | **清 escalation_pending + 重置 timeout_extensions=0**（新 attempt） | 子会话启动 | [现状] |
 | T3 | running | nodeCompleted | 子会话结果 | completed | **清 escalation_pending**（终态无裁决对象） | 结果交付（终态交付臂） | [目标] ADR-0001 |
 | T4 | running/queued | nodeFailed（reason + trigger） | 子会话失败 / watchdog cap / recovery | failed | **清 escalation_pending**；trigger 入 error 语义 | `[DAG Node Result]`/wake 承载 reason+trigger（错误即状态→处置依据） | [目标] ADR-0001 |
-| T5 | pending/queued/running | nodeCancelled | replan cancel / workflow cancel | cancelled | **清 escalation_pending**（cancel 即裁决） | 取消交付 | [目标] ADR-0001 |
+| T5 | pending/queued/running | nodeCancelled | replan cancel / workflow cancel | failed(cancelled) | **status=failed + error_reason='cancelled via replan' + 清 escalation_pending**（cancel 即裁决；节点级无独立 cancelled 终态，取消语义经 error_reason 承载） | 取消交付 | [目标] ADR-0001 |
 | T6 | pending/queued | nodeSkipped | 依赖失败级联 | skipped | — | 跳过级联 | [现状] |
 | T7 | failed | nodeRestarted | replan restart | running | 清旗 + 重置计数（新 attempt） | 重试 | [现状] |
 | T8 | running | nodeTimeoutEscalated | **watchdog（提议者）** | running | timeout_extensions+1、escalation_pending=true、wake re-arm（wake_reported=false） | `[DAG Node Timeout]` wake（extend 或 cancel 的裁决请求） | [现状] |

package.json

@@ -5,15 +5,15 @@
   "private": true,
   "type": "module",
   "packageManager": "bun@1.3.14",
-  "_lint_ratchet_note": "Ratchet set to the CI type-aware baseline (4888). CI lints ~3 more files than a local run (install/platform-generated artifacts on an identical git tree), producing ~10 extra same-category type-aware warnings (4888 CI vs 4878 local, 0 errors) — NOT new code warnings. This batch raised the baseline by ~36 type-aware no-unsafe-type-assertion warnings from two new dag test files (dag-deadline-extended.test.ts, dag-escalation-clear-flag.test.ts) using the established `as never` test-data idiom — same category as the prior 4842. When you fix existing warnings locally, lower --max-warnings in the lint script to match so the gate keeps tightening. See .oxlintrc.json header for the ratchet contract.",
+  "_lint_ratchet_note": "Ratchet lowered to 4852 (the pre-batch-A CI baseline) after replacing the `as never` test-data idiom in the three dag timeout/escalation test files (dag-deadline-extended, dag-escalation-clear-flag, dag-timeout-escalation) with schema brand makers (Project.ID.make, Session.ID.make, DagEvent.NodeID.make, AbsolutePath.make) and fully-typed InstanceRef/Session mocks — their no-unsafe-type-assertion warnings are gone. CI lints ~3 extra install/platform-generated artifacts on an identical tree, adding ~10 same-category type-aware warnings (~4841 CI vs ~4831 local, 0 errors) — NOT new code warnings; 4852 keeps a small margin over the projected CI count. When you fix existing warnings locally, lower --max-warnings to match so the gate keeps tightening. See .oxlintrc.json header for the ratchet contract.",
   "scripts": {
     "dev": "bun run --cwd packages/opencode --conditions=browser src/index.ts",
     "dev:desktop": "bun --cwd packages/desktop dev",
     "dev:web": "bun --cwd packages/app dev",
     "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=4888",
+    "lint": "oxlint --max-warnings=4852",
     "typecheck": "bun turbo typecheck",
     "upgrade-opentui": "bun run script/upgrade-opentui.ts",
     "postinstall": "bun run --cwd packages/core fix-node-pty",

packages/core/src/dag/projector.ts

@@ -32,6 +32,12 @@ export const NodeStatusProjection = {
   completed: { to: "completed", from: ["running"] },
   failed: { to: "failed", from: ["running", "pending", "queued"] },
   skipped: { to: "skipped", from: ["pending", "queued", "running", "paused"] },
+  // NodeCancelled has NO independent terminal status — the NodeStatus enum has
+  // no CANCELLED and getValidNextNodeStatuses never returns it. A cancelled
+  // node lands on `failed` with the cancellation carried by `error_reason`
+  // ("cancelled via replan"), never on a phantom node-level "cancelled" status.
+  // Workflow-level cancelled (WorkflowStatusProjection.cancelled below) is a
+  // legitimate, separate terminal — this entry is node-scoped only.
   cancelled: { to: "failed", from: ["pending", "queued", "running", "paused"] },
   restarted: { to: "pending", from: ["running"] },
 } as const
@@ -342,6 +348,12 @@ export const layer = Layer.effectDiscard(
         .pipe(Effect.orDie),
     )
 
+    // NodeCancelled carries no independent terminal status: it projects to
+    // status="failed" with the cancellation marker in error_reason and clears
+    // the adjudication flag (cancel is itself an adjudication). A node row can
+    // therefore never hold status="cancelled"; see NodeStatusProjection.cancelled
+    // above and the canonical proof in
+    // packages/opencode/test/dag/dag-escalation-clear-flag.test.ts:130-148.
     yield* events.project(DagEvent.NodeCancelled, (event) =>
       db
         .update(WorkflowNodeTable)

packages/core/test/dag-node-cancelled-projection.test.ts

@@ -0,0 +1,106 @@
+/**
+ * Regression guard for the NodeCancelled projection contract (ticket A,
+ * method-A: align to implementation).
+ *
+ * NodeCancelled has NO independent terminal status. It projects to
+ * `status="failed"` carrying the cancellation marker in `error_reason`
+ * ("cancelled via replan") and clears `escalation_pending` (cancel is an
+ * adjudication). The NodeStatus enum has no CANCELLED value and
+ * getValidNextNodeStatuses never returns cancelled, so a node row can never
+ * hold status="cancelled". This test exercises the real projector SQL
+ * (projector.ts NodeCancelled handler) end-to-end at the core layer so the
+ * semantic cannot silently drift back to a phantom node-level "cancelled"
+ * status.
+ *
+ * The end-to-end canonical proof lives in
+ * packages/opencode/test/dag/dag-escalation-clear-flag.test.ts:130-148; this
+ * core-level test mirrors it without depending on the opencode Dag command
+ * layer.
+ */
+import { describe, expect, test } from "bun:test"
+import { DateTime, Effect, Layer } from "effect"
+import { Database } from "@opencode-ai/core/database/database"
+import { EventV2 } from "@opencode-ai/core/event"
+import { DagProjector } from "@opencode-ai/core/dag/projector"
+import { DagStore } from "@opencode-ai/core/dag/store"
+import { WorkflowNodeTable, WorkflowTable } from "@opencode-ai/core/dag/sql"
+import { ProjectTable } from "@opencode-ai/core/project/sql"
+import { SessionTable } from "@opencode-ai/core/session/sql"
+import { DagEvent } from "@opencode-ai/schema/dag-event"
+
+function projectorLayer() {
+  const database = Database.layerFromPath(":memory:")
+  const eventLayer = EventV2.layer.pipe(Layer.provide(database))
+  const projector = DagProjector.layer.pipe(Layer.provide(Layer.merge(database, eventLayer)))
+  const store = DagStore.layer.pipe(Layer.provide(database))
+  return Layer.mergeAll(database, eventLayer, projector, store)
+}
+
+function seed() {
+  return Effect.gen(function* () {
+    const { db } = yield* Database.Service
+    yield* db.insert(ProjectTable).values({
+      id: "project-1" as never,
+      worktree: process.cwd() as never,
+      sandboxes: [],
+    }).run().pipe(Effect.orDie)
+    yield* db.insert(SessionTable).values({
+      id: "ses_parent" as never,
+      project_id: "project-1" as never,
+      slug: "parent",
+      directory: process.cwd() as never,
+      title: "Parent",
+      version: "test",
+    }).run().pipe(Effect.orDie)
+    yield* db.insert(WorkflowTable).values({
+      id: "dag_cancel",
+      project_id: "project-1" as never,
+      session_id: "ses_parent" as never,
+      title: "Cancel projection",
+      status: "running",
+      config: "{}",
+      seq: 1,
+      wake_reported: false,
+    }).run().pipe(Effect.orDie)
+    yield* db.insert(WorkflowNodeTable).values({
+      id: "n1",
+      workflow_id: "dag_cancel",
+      name: "N1",
+      worker_type: "build",
+      status: "running",
+      required: true,
+      depends_on: [],
+      wake_eligible: false,
+      wake_reported: false,
+      // Pre-set an adjudication flag so the projection's clear is observable.
+      escalation_pending: true,
+      seq: 1,
+    }).run().pipe(Effect.orDie)
+  })
+}
+
+describe("NodeCancelled projection (no phantom node-level cancelled status)", () => {
+  test("projects NodeCancelled to status=failed + error_reason='cancelled via replan' and clears escalation_pending", async () => {
+    await Effect.runPromise(
+      Effect.gen(function* () {
+        yield* seed()
+        const events = yield* EventV2.Service
+        const store = yield* DagStore.Service
+
+        yield* events.publish(DagEvent.NodeCancelled, {
+          dagID: DagEvent.DagID.make("dag_cancel"),
+          nodeID: DagEvent.NodeID.make("n1"),
+          timestamp: yield* DateTime.now,
+        })
+
+        const row = yield* store.getNode("dag_cancel", "n1")
+        // NodeCancelled has no independent terminal status: it lands on failed
+        // with the cancellation carried by error_reason, never status="cancelled".
+        expect(row?.status).toBe("failed")
+        expect(row?.errorReason).toBe("cancelled via replan")
+        // Cancel is an adjudication — the pending-escalation flag must clear.
+        expect(row?.escalationPending).toBe(false)
+      }).pipe(Effect.provide(projectorLayer()), Effect.scoped),
+    )
+  })
+})

packages/core/test/dag-projector-drift.test.ts

@@ -67,3 +67,14 @@ describe("projector from-guards vs declared transition tables", () => {
 // third encoding of the same machine with zero production callers — a
 // capability reservoir kept for the event-semantics mapping. It is exercised
 // by dag-core.test.ts only and intentionally not welded here.
+//
+// Note (ticket A, method-A): NodeStatusProjection.cancelled.to === "failed"
+// is intentional, not a missing target. NodeCancelled has no independent
+// terminal status — the NodeStatus enum has no CANCELLED and
+// getValidNextNodeStatuses never returns cancelled, so a node row can never
+// hold status="cancelled". The drift test passes for cancelled because
+// "failed" is a legal target from every cancelled.from state; a phantom
+// node-level "cancelled" target is what this alignment rules out. The
+// cancellation marker rides on error_reason ("cancelled via replan"), locked
+// by dag-node-cancelled-projection.test.ts and the opencode canonical proof
+// at packages/opencode/test/dag/dag-escalation-clear-flag.test.ts:130-148.

packages/opencode/src/dag/dag.ts

@@ -318,7 +318,7 @@ export interface Interface {
   readonly nodeSkipped: (dagID: string, nodeID: string, reason: string) => Effect.Effect<void, Error>
   readonly nodeCancelled: (dagID: string, nodeID: string) => Effect.Effect<void, Error>
   readonly nodeRestarted: (dagID: string, nodeID: string, childSessionID: string) => Effect.Effect<void, Error>
-  readonly nodeTimeoutEscalated: (dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number) => Effect.Effect<void, Error>
+  readonly nodeTimeoutEscalated: (dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number, staleDeadlineMs?: number | null) => Effect.Effect<void, Error>
   readonly nodeExtendTimeout: (dagID: string, nodeID: string, newDeadlineMs: number) => Effect.Effect<number, Error>
 }
 
@@ -887,8 +887,28 @@ export const layer = Layer.effect(
     // Timeout escalation publishes no status transition — the node stays
     // RUNNING (see the NodeTimeoutEscalated projector). Only the extension
     // count, seq, and wake flag change.
-    const nodeTimeoutEscalated = Effect.fn("Dag.nodeTimeoutEscalated")(function* (lock: WorkflowLock, dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number) {
+    //
+    // Ticket B (method-A — stale-read suppression): the deadline watcher reads
+    // the durable row WITHOUT the workflow lock (spawn.ts readNode). Between
+    // that stale snapshot and this command acquiring the lock, a replan's
+    // nodeExtendTimeout may have moved the deadline into the future. Escalating
+    // then would charge a max_timeout_extensions budget unit for a node that is
+    // no longer overdue — a spurious T8 (the cosmetic residue self-documented
+    // at loop.ts:870-880). The caller passes the deadline it OBSERVED
+    // (node.deadlineMs); this command re-reads the node FRESH under the workflow
+    // lock and, when the deadline has moved strictly past the observed value,
+    // suppresses the escalation (no publish, no budget increment). Budget only
+    // counts a real extension (a deadline that actually moved), not a stale-read
+    // cosmetic recount. Suppression returns void, exactly like a publish, so the
+    // watcher's self-renewal loop (S1) keeps supervising — a running node is
+    // never orphaned (N1). When staleDeadlineMs is omitted (existing callers,
+    // test setups) the guard is inert: back-compat is unconditional publish.
+    const nodeTimeoutEscalated = Effect.fn("Dag.nodeTimeoutEscalated")(function* (lock: WorkflowLock, dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number, staleDeadlineMs?: number | null) {
       yield* guardWorkflowNotTerminal(dagID, "timeout escalation")
+      if (staleDeadlineMs != null) {
+        const node = yield* store.getNode(dagID, nodeID).pipe(Effect.orDie)
+        if (node && node.status === "running" && node.deadlineMs != null && node.deadlineMs > staleDeadlineMs) return
+      }
       yield* events.publish(DagEvent.NodeTimeoutEscalated, {
         dagID: dagID as ID,
         nodeID: nodeID as never,
@@ -959,8 +979,8 @@ export const layer = Layer.effect(
       nodeSkipped: (dagID, nodeID, reason) => withWorkflowLock(dagID)((lock) => nodeSkipped(lock, dagID, nodeID, reason)),
       nodeCancelled: (dagID, nodeID) => withWorkflowLock(dagID)((lock) => nodeCancelled(lock, dagID, nodeID)),
       nodeRestarted: (dagID, nodeID, childSessionID) => withWorkflowLock(dagID)((lock) => nodeRestarted(lock, dagID, nodeID, childSessionID)),
-      nodeTimeoutEscalated: (dagID, nodeID, childSessionID, timeoutExtensions) =>
-        withWorkflowLock(dagID)((lock) => nodeTimeoutEscalated(lock, dagID, nodeID, childSessionID, timeoutExtensions)),
+      nodeTimeoutEscalated: (dagID, nodeID, childSessionID, timeoutExtensions, staleDeadlineMs) =>
+        withWorkflowLock(dagID)((lock) => nodeTimeoutEscalated(lock, dagID, nodeID, childSessionID, timeoutExtensions, staleDeadlineMs)),
       nodeExtendTimeout: (dagID, nodeID, newDeadlineMs) => withWorkflowLock(dagID)((lock) => nodeExtendTimeout(lock, dagID, nodeID, newDeadlineMs)),
     })
   }),

packages/opencode/src/dag/runtime/spawn.ts

@@ -178,7 +178,13 @@ export function makeDeadlineWatcher(
       // slot unbounded. Log and fall through to the sleep — the next iteration
       // re-reads the row and escalates again. Mirrors the read path's R13
       // hardening above, which the write path previously lacked.
-      const escalated = yield* dag.nodeTimeoutEscalated(input.dagID, input.nodeID, node.childSessionId as never, extensions + 1).pipe(
+      // The deadline this watcher OBSERVED may be stale — it was read WITHOUT
+      // the workflow lock (readNode above). nodeTimeoutEscalated re-reads the
+      // node under the lock and suppresses the escalation when the deadline has
+      // moved past this observed value (ticket B — spurious T8 suppression),
+      // so a budget unit is only charged when the node is genuinely still
+      // overdue. Pass node.deadlineMs, the value this snapshot read.
+      const escalated = yield* dag.nodeTimeoutEscalated(input.dagID, input.nodeID, node.childSessionId as never, extensions + 1, node.deadlineMs).pipe(
         Effect.catchIf(
           isTransitionRejection,
           () => Effect.logWarning("nodeTimeoutEscalated guard rejected — node already terminal"),