From 67d1ca2b164c25ce736dfcb4b330bd177ada3faa Mon Sep 17 00:00:00 2001 From: lex Date: Sat, 8 Aug 2026 14:45:02 +0800 Subject: [PATCH 1/3] =?UTF-8?q?fix(dag):=20eliminate=20phantom=20cancelled?= =?UTF-8?q?=20node=20state=20=E2=80=94=20align=20transition=20table=20T5?= =?UTF-8?q?=20with=20projection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../node-lifecycle-transitions.md | 6 +- packages/core/src/dag/projector.ts | 12 ++ .../dag-node-cancelled-projection.test.ts | 106 ++++++++++++++++++ .../core/test/dag-projector-drift.test.ts | 11 ++ 4 files changed, 133 insertions(+), 2 deletions(-) create mode 100644 packages/core/test/dag-node-cancelled-projection.test.ts diff --git a/.opencode/grill-batch-a/node-lifecycle-transitions.md b/.opencode/grill-batch-a/node-lifecycle-transitions.md index c37a5c02d8..9e142df277 100644 --- a/.opencode/grill-batch-a/node-lifecycle-transitions.md +++ b/.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 的裁决请求） | [现状] | diff --git a/packages/core/src/dag/projector.ts b/packages/core/src/dag/projector.ts index e85d767787..49b901b58d 100644 --- a/packages/core/src/dag/projector.ts +++ b/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) diff --git a/packages/core/test/dag-node-cancelled-projection.test.ts b/packages/core/test/dag-node-cancelled-projection.test.ts new file mode 100644 index 0000000000..e23ebbe55b --- /dev/null +++ b/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), + ) + }) +}) diff --git a/packages/core/test/dag-projector-drift.test.ts b/packages/core/test/dag-projector-drift.test.ts index 5accf890c6..b721b97a08 100644 --- a/packages/core/test/dag-projector-drift.test.ts +++ b/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. From b356304861030f8c91b2a4961112a8b9fdccbe54 Mon Sep 17 00:00:00 2001 From: lex Date: Sat, 8 Aug 2026 14:45:23 +0800 Subject: [PATCH 2/3] fix(dag): stale watchdog read no longer consumes timeout extension budget --- packages/opencode/src/dag/dag.ts | 28 ++- packages/opencode/src/dag/runtime/spawn.ts | 8 +- .../test/dag/dag-retime-stale-read.test.ts | 222 ++++++++++++++++++ 3 files changed, 253 insertions(+), 5 deletions(-) create mode 100644 packages/opencode/test/dag/dag-retime-stale-read.test.ts diff --git a/packages/opencode/src/dag/dag.ts b/packages/opencode/src/dag/dag.ts index d203ceaa91..1545a2d5bc 100644 --- a/packages/opencode/src/dag/dag.ts +++ b/packages/opencode/src/dag/dag.ts @@ -318,7 +318,7 @@ export interface Interface { readonly nodeSkipped: (dagID: string, nodeID: string, reason: string) => Effect.Effect readonly nodeCancelled: (dagID: string, nodeID: string) => Effect.Effect readonly nodeRestarted: (dagID: string, nodeID: string, childSessionID: string) => Effect.Effect - readonly nodeTimeoutEscalated: (dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number) => Effect.Effect + readonly nodeTimeoutEscalated: (dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number, staleDeadlineMs?: number | null) => Effect.Effect readonly nodeExtendTimeout: (dagID: string, nodeID: string, newDeadlineMs: number) => Effect.Effect } @@ -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)), }) }), diff --git a/packages/opencode/src/dag/runtime/spawn.ts b/packages/opencode/src/dag/runtime/spawn.ts index 1adee59822..008ef15ad2 100644 --- a/packages/opencode/src/dag/runtime/spawn.ts +++ b/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"), diff --git a/packages/opencode/test/dag/dag-retime-stale-read.test.ts b/packages/opencode/test/dag/dag-retime-stale-read.test.ts new file mode 100644 index 0000000000..f8e8b94b19 --- /dev/null +++ b/packages/opencode/test/dag/dag-retime-stale-read.test.ts @@ -0,0 +1,222 @@ +import { describe, expect, it } from "bun:test" +import { Effect, Layer } from "effect" +import { sql } from "drizzle-orm" +import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { EventTable } from "@opencode-ai/core/event/sql" +import { DagProjector } from "@opencode-ai/core/dag/projector" +import { DagStore } from "@opencode-ai/core/dag/store" +import { Session } from "@opencode-ai/schema/session" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { Dag, type NodeConfig } from "@/dag/dag" +import { EventV2Bridge } from "@/event-v2-bridge" +import { InstanceRef } from "@/effect/instance-ref" + +// ============================================================================ +// Ticket B — spurious T8 (stale-read budget consumption), method-A. +// +// The deadline watcher reads the durable row WITHOUT the workflow lock +// (spawn.ts readNode). When that stale snapshot shows an expired deadline it +// calls dag.nodeTimeoutEscalated, which acquires the workflow lock and would +// unconditionally publish NodeTimeoutEscalated (incrementing timeout_extensions). +// If a replan's nodeExtendTimeout moved the deadline into the future BETWEEN the +// stale read and the lock acquisition, the escalation charges a budget unit for +// a node that is no longer overdue — a spurious T8 (domain 3; the cosmetic +// residue self-documented at loop.ts:870-880). +// +// Method-A fix: the watchdog passes the deadline it observed (node.deadlineMs). +// nodeTimeoutEscalated 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. +// +// N1 (running node never loses its watcher): nodeTimeoutEscalated returns void +// whether it publishes or suppresses, and the watcher's self-renewal loop +// (spawn.ts:126-199) only exits on a terminal status or fiber interrupt — a +// suppressed escalation flows into the same post-escalation sleep+re-read as a +// published one, so supervision cannot end on the suppression path. +// ============================================================================ + +function node(id: string, timeoutMs?: number): NodeConfig { + return { + id, + name: id, + worker_type: "build", + depends_on: [], + required: true, + prompt_template: { inline: id }, + ...(timeoutMs !== undefined ? { worker_config: { timeout_ms: timeoutMs } } : {}), + } +} + +const harness = (() => { + const database = Database.layerFromPath(":memory:") + const events = EventV2.layer.pipe(Layer.provide(database)) + const bridge = EventV2Bridge.layer.pipe(Layer.provide(events)) + const store = DagStore.layer.pipe(Layer.provide(database)) + const projector = DagProjector.layer.pipe(Layer.provide(events), Layer.provide(database)) + const dag = Dag.layer.pipe(Layer.provide(bridge), Layer.provide(store)) + return Layer.mergeAll(database, events, bridge, store, projector, dag) +})() + +function runTest( + test: (services: { readonly dag: Dag.Interface; readonly store: DagStore.Interface; readonly db: Database.Interface["db"] }) => Effect.Effect, +) { + return Effect.gen(function* () { + return yield* Effect.gen(function* () { + const database = yield* Database.Service + yield* database.db.insert(ProjectTable).values({ + id: Project.ID.make("project-1"), + worktree: AbsolutePath.make(process.cwd()), + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.db.insert(SessionTable).values({ + id: Session.ID.make("ses_parent"), + project_id: Project.ID.make("project-1"), + slug: "parent", + directory: AbsolutePath.make(process.cwd()), + title: "Parent", + version: "test", + }).run().pipe(Effect.orDie) + const dag = yield* Dag.Service + const store = yield* DagStore.Service + return yield* test({ dag, store, db: database.db }) + }).pipe( + Effect.provide(harness), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { + id: Project.ID.make("project-1"), + worktree: process.cwd(), + time: { created: 0, updated: 0 }, + sandboxes: [], + }, + }), + Effect.scoped, + ) + }) +} + +function createWorkflow(dag: Dag.Interface, title: string, nodeID = "a") { + return dag.create({ + projectID: Project.ID.make("project-1"), + sessionID: Session.ID.make("ses_parent"), + title, + config: { name: title, nodes: [node(nodeID)] }, + }) +} + +// Count durable NodeTimeoutEscalated rows for a node. The stored type is +// versioned (`dag.node.timeout_escalated.1`), so match the prefix. Narrowing +// the JSON `data` column to a struct is a safe downcast (not an unsafe +// assertion) — it never inflates the no-unsafe-type-assertion ratchet. +function timeoutEscalatedCount(db: Database.Interface["db"], dagID: string, nodeID: string) { + return Effect.gen(function* () { + const rows = yield* db + .select({ type: EventTable.type, data: EventTable.data }) + .from(EventTable) + .where(sql`${EventTable.aggregate_id} = ${dagID} AND ${EventTable.type} LIKE 'dag.node.timeout_escalated.%'`) + .all() + .pipe(Effect.orDie) + return rows.filter((row) => (row.data as { nodeID?: string }).nodeID === nodeID).length + }) +} + +describe("nodeTimeoutEscalated stale-read suppression (ticket B, method-A)", () => { + it("suppresses the escalation when the deadline was extended after the watcher's stale read (spurious T8)", async () => { + await Effect.runPromise( + runTest(({ dag, store, db }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "stale-suppressed") + // The node started with an EXPIRED deadline — this is the value the + // watcher's stale snapshot would have read. + const expiredDeadline = Date.now() - 5_000 + yield* dag.nodeQueued(dagID, "a", expiredDeadline) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", expiredDeadline, true) + + // A replan adjudicates the timeout by extending the deadline into the + // future AFTER the watcher's snapshot read but BEFORE its escalation + // acquires the workflow lock. nodeExtendTimeout publishes + // NodeDeadlineExtended (no budget change — extensions never increment + // timeout_extensions). + const extendedDeadline = Date.now() + 60_000 + const written = yield* dag.nodeExtendTimeout(dagID, "a", extendedDeadline) + expect(written).toBe(1) + const extended = yield* store.getNode(dagID, "a") + expect(extended?.deadlineMs).toBe(extendedDeadline) + expect(extended?.timeoutExtensions).toBe(0) + + // The watcher fires nodeTimeoutEscalated carrying the deadline it + // OBSERVED (the stale expired value). Under the workflow lock the + // command re-reads the node: its deadline is now strictly past the + // observed value → the stale read is invalidated → the escalation is + // suppressed. No NodeTimeoutEscalated event, no budget increment. + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 1, expiredDeadline) + + const eventCount = yield* timeoutEscalatedCount(db, dagID, "a") + expect(eventCount).toBe(0) + + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("running") + expect(row?.timeoutExtensions).toBe(0) + expect(row?.escalationPending).toBe(false) + expect(row?.deadlineMs).toBe(extendedDeadline) + }), + ), + ) + }) + + it("still escalates when the deadline was NOT extended (legitimate escalation — regression arm)", async () => { + await Effect.runPromise( + runTest(({ dag, store, db }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "stale-legitimate") + const expiredDeadline = Date.now() - 5_000 + yield* dag.nodeQueued(dagID, "a", expiredDeadline) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", expiredDeadline, true) + // No replan extend: the fresh in-lock deadline equals the observed + // value, so the escalation is NOT a stale read and must publish. + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 1, expiredDeadline) + + const eventCount = yield* timeoutEscalatedCount(db, dagID, "a") + expect(eventCount).toBe(1) + + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("running") + expect(row?.timeoutExtensions).toBe(1) + expect(row?.escalationPending).toBe(true) + expect(row?.wakeReported).toBe(false) + }), + ), + ) + }) + + it("preserves back-compat: a 4-arg call (no observed deadline) never suppresses", async () => { + await Effect.runPromise( + runTest(({ dag, store, db }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "stale-backcompat") + // A future-deadline node is escalated directly (the idiom existing + // tests use to set up escalation_pending). With no observed-deadline + // argument the suppression guard is inert — callers that do not opt + // into the stale-read protocol keep the unconditional-publish + // behavior, so existing 4-arg call sites are unaffected. + yield* dag.nodeQueued(dagID, "a", Date.now() + 60_000) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", Date.now() + 60_000, true) + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 1) + + const eventCount = yield* timeoutEscalatedCount(db, dagID, "a") + expect(eventCount).toBe(1) + + const row = yield* store.getNode(dagID, "a") + expect(row?.timeoutExtensions).toBe(1) + expect(row?.escalationPending).toBe(true) + }), + ), + ) + }) +}) From 222c17277d9e44e3c99dc5ddc042146720d60d26 Mon Sep 17 00:00:00 2001 From: lex Date: Sat, 8 Aug 2026 14:45:43 +0800 Subject: [PATCH 3/3] =?UTF-8?q?test(dag):=20type-safe=20fixtures=20remove?= =?UTF-8?q?=2036=20lint=20warnings,=20ratchet=20restored=204888=E2=86=9248?= =?UTF-8?q?52?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- package.json | 4 +- .../test/dag/dag-deadline-extended.test.ts | 52 +++++++++------- .../dag/dag-escalation-clear-flag.test.ts | 22 ++++--- .../test/dag/dag-timeout-escalation.test.ts | 60 ++++++++++++++----- 4 files changed, 90 insertions(+), 48 deletions(-) diff --git a/package.json b/package.json index d39dd725d4..9a8ee27513 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "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", @@ -13,7 +13,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=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", diff --git a/packages/opencode/test/dag/dag-deadline-extended.test.ts b/packages/opencode/test/dag/dag-deadline-extended.test.ts index 9ce570d42a..ce67d2a8e8 100644 --- a/packages/opencode/test/dag/dag-deadline-extended.test.ts +++ b/packages/opencode/test/dag/dag-deadline-extended.test.ts @@ -7,6 +7,7 @@ import { EventTable, EventSequenceTable } from "@opencode-ai/core/event/sql" import { DagProjector } from "@opencode-ai/core/dag/projector" import { DagStore } from "@opencode-ai/core/dag/store" import { DagEvent } from "@opencode-ai/schema/dag-event" +import { Session } from "@opencode-ai/schema/session" import { Project } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" import { SessionTable } from "@opencode-ai/core/session/sql" @@ -50,15 +51,15 @@ function runTest( return yield* Effect.gen(function* () { const database = yield* Database.Service yield* database.db.insert(ProjectTable).values({ - id: "project-1" as never, - worktree: process.cwd() as never, + id: Project.ID.make("project-1"), + worktree: AbsolutePath.make(process.cwd()), sandboxes: [], }).run().pipe(Effect.orDie) yield* database.db.insert(SessionTable).values({ - id: "ses_parent" as never, - project_id: "project-1" as never, + id: Session.ID.make("ses_parent"), + project_id: Project.ID.make("project-1"), slug: "parent", - directory: process.cwd() as never, + directory: AbsolutePath.make(process.cwd()), title: "Parent", version: "test", }).run().pipe(Effect.orDie) @@ -70,8 +71,13 @@ function runTest( Effect.provideService(InstanceRef, { directory: process.cwd(), worktree: process.cwd(), - project: { id: "project-1" }, - } as never), + project: { + id: Project.ID.make("project-1"), + worktree: process.cwd(), + time: { created: 0, updated: 0 }, + sandboxes: [], + }, + }), Effect.scoped, ) }) @@ -118,7 +124,7 @@ function setupFKs() { return Effect.gen(function* () { const { db } = yield* Database.Service yield* db.insert(ProjectTable).values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }).run().pipe(Effect.orDie) - yield* db.insert(SessionTable).values({ id: "ses_replay" as never, project_id: Project.ID.global, slug: "replay", directory: "/project", title: "replay", version: "test" }).run().pipe(Effect.orDie) + yield* db.insert(SessionTable).values({ id: Session.ID.make("ses_replay"), project_id: Project.ID.global, slug: "replay", directory: "/project", title: "replay", version: "test" }).run().pipe(Effect.orDie) }) } @@ -261,15 +267,15 @@ describe("NodeDeadlineExtended projector fold + replay (Q3)", () => { yield* setupFKs() const events = yield* EventV2.Service const store = yield* DagStore.Service - const dagID = "dag_replay_extend" as never + const dagID = DagEvent.DagID.descending("dag_replay_extend") - yield* events.publish(DagEvent.WorkflowCreated, { dagID, projectID: Project.ID.global as never, sessionID: "ses_replay" as never, title: "extend-replay", config: "{}", status: "pending", timestamp: ts(0) }) - yield* events.publish(DagEvent.NodeRegistered, { dagID, nodeID: "a" as never, name: "A", workerType: "build", dependsOn: [], required: true, timestamp: ts(1) }) + yield* events.publish(DagEvent.WorkflowCreated, { dagID, projectID: Project.ID.global, sessionID: Session.ID.make("ses_replay"), title: "extend-replay", config: "{}", status: "pending", timestamp: ts(0) }) + yield* events.publish(DagEvent.NodeRegistered, { dagID, nodeID: DagEvent.NodeID.make("a"), name: "A", workerType: "build", dependsOn: [], required: true, timestamp: ts(1) }) yield* events.publish(DagEvent.WorkflowStarted, { dagID, timestamp: ts(2) }) - yield* events.publish(DagEvent.NodeStarted, { dagID, nodeID: "a" as never, childSessionID: "ses_child" as never, deadlineMs: 5_000, wakeEligible: true, timestamp: ts(3) }) + yield* events.publish(DagEvent.NodeStarted, { dagID, nodeID: DagEvent.NodeID.make("a"), childSessionID: Session.ID.make("ses_child"), deadlineMs: 5_000, wakeEligible: true, timestamp: ts(3) }) // Escalate, then adjudicate by extending the deadline. - yield* events.publish(DagEvent.NodeTimeoutEscalated, { dagID, nodeID: "a" as never, childSessionID: "ses_child" as never, timeoutExtensions: 1, timestamp: ts(4) }) - yield* events.publish(DagEvent.NodeDeadlineExtended, { dagID, nodeID: "a" as never, deadlineMs: 99_999, timeoutExtensions: 1, timestamp: ts(5) }) + yield* events.publish(DagEvent.NodeTimeoutEscalated, { dagID, nodeID: DagEvent.NodeID.make("a"), childSessionID: Session.ID.make("ses_child"), timeoutExtensions: 1, timestamp: ts(4) }) + yield* events.publish(DagEvent.NodeDeadlineExtended, { dagID, nodeID: DagEvent.NodeID.make("a"), deadlineMs: 99_999, timeoutExtensions: 1, timestamp: ts(5) }) const before = yield* store.getNode(dagID, "a") expect(before?.deadlineMs).toBe(99_999) @@ -288,7 +294,7 @@ describe("NodeDeadlineExtended projector fold + replay (Q3)", () => { expect(replayed?.escalationPending).toBe(false) expect(replayed?.wakeReported).toBe(true) expect(replayed?.timeoutExtensions).toBe(1) - }).pipe(Effect.provide(projectorLayer)) as Effect.Effect, + }).pipe(Effect.provide(projectorLayer)), ) }) @@ -298,17 +304,17 @@ describe("NodeDeadlineExtended projector fold + replay (Q3)", () => { yield* setupFKs() const events = yield* EventV2.Service const store = yield* DagStore.Service - const dagID = "dag_replay_stale_extend" as never + const dagID = DagEvent.DagID.descending("dag_replay_stale_extend") - yield* events.publish(DagEvent.WorkflowCreated, { dagID, projectID: Project.ID.global as never, sessionID: "ses_replay" as never, title: "stale-extend", config: "{}", status: "pending", timestamp: ts(0) }) - yield* events.publish(DagEvent.NodeRegistered, { dagID, nodeID: "a" as never, name: "A", workerType: "build", dependsOn: [], required: true, timestamp: ts(1) }) + yield* events.publish(DagEvent.WorkflowCreated, { dagID, projectID: Project.ID.global, sessionID: Session.ID.make("ses_replay"), title: "stale-extend", config: "{}", status: "pending", timestamp: ts(0) }) + yield* events.publish(DagEvent.NodeRegistered, { dagID, nodeID: DagEvent.NodeID.make("a"), name: "A", workerType: "build", dependsOn: [], required: true, timestamp: ts(1) }) yield* events.publish(DagEvent.WorkflowStarted, { dagID, timestamp: ts(2) }) - yield* events.publish(DagEvent.NodeStarted, { dagID, nodeID: "a" as never, childSessionID: "ses_child" as never, deadlineMs: 5_000, wakeEligible: true, timestamp: ts(3) }) - yield* events.publish(DagEvent.NodeDeadlineExtended, { dagID, nodeID: "a" as never, deadlineMs: 50_000, timeoutExtensions: 1, timestamp: ts(4) }) + yield* events.publish(DagEvent.NodeStarted, { dagID, nodeID: DagEvent.NodeID.make("a"), childSessionID: Session.ID.make("ses_child"), deadlineMs: 5_000, wakeEligible: true, timestamp: ts(3) }) + yield* events.publish(DagEvent.NodeDeadlineExtended, { dagID, nodeID: DagEvent.NodeID.make("a"), deadlineMs: 50_000, timeoutExtensions: 1, timestamp: ts(4) }) // Node completes AFTER the extension was logged... - yield* events.publish(DagEvent.NodeCompleted, { dagID, nodeID: "a" as never, output: "done", durationMs: 0, timestamp: ts(5) }) + yield* events.publish(DagEvent.NodeCompleted, { dagID, nodeID: DagEvent.NodeID.make("a"), output: "done", durationMs: 0, timestamp: ts(5) }) // ...then a stale/late extension races in (crash-recovery replay order). - yield* events.publish(DagEvent.NodeDeadlineExtended, { dagID, nodeID: "a" as never, deadlineMs: 99_999, timeoutExtensions: 1, timestamp: ts(6) }) + yield* events.publish(DagEvent.NodeDeadlineExtended, { dagID, nodeID: DagEvent.NodeID.make("a"), deadlineMs: 99_999, timeoutExtensions: 1, timestamp: ts(6) }) const row = yield* store.getNode(dagID, "a") // The projector's status='running' WHERE guard means the stale fold is a @@ -317,7 +323,7 @@ describe("NodeDeadlineExtended projector fold + replay (Q3)", () => { expect(row?.status).toBe("completed") expect(row?.deadlineMs).toBe(50_000) expect(row?.output).toBe("done") - }).pipe(Effect.provide(projectorLayer)) as Effect.Effect, + }).pipe(Effect.provide(projectorLayer)), ) }) }) diff --git a/packages/opencode/test/dag/dag-escalation-clear-flag.test.ts b/packages/opencode/test/dag/dag-escalation-clear-flag.test.ts index e322c5ecab..235dfac9e1 100644 --- a/packages/opencode/test/dag/dag-escalation-clear-flag.test.ts +++ b/packages/opencode/test/dag/dag-escalation-clear-flag.test.ts @@ -4,8 +4,11 @@ import { Database } from "@opencode-ai/core/database/database" import { DagProjector } from "@opencode-ai/core/dag/projector" import { DagStore } from "@opencode-ai/core/dag/store" import { EventV2 } from "@opencode-ai/core/event" +import { Project } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionTable } from "@opencode-ai/core/session/sql" +import { Session } from "@opencode-ai/schema/session" import { Dag, type NodeConfig } from "@/dag/dag" import { EventV2Bridge } from "@/event-v2-bridge" import { InstanceRef } from "@/effect/instance-ref" @@ -39,15 +42,15 @@ function runTest( return yield* Effect.gen(function* () { const database = yield* Database.Service yield* database.db.insert(ProjectTable).values({ - id: "project-1" as never, - worktree: process.cwd() as never, + id: Project.ID.make("project-1"), + worktree: AbsolutePath.make(process.cwd()), sandboxes: [], }).run().pipe(Effect.orDie) yield* database.db.insert(SessionTable).values({ - id: "ses_parent" as never, - project_id: "project-1" as never, + id: Session.ID.make("ses_parent"), + project_id: Project.ID.make("project-1"), slug: "parent", - directory: process.cwd() as never, + directory: AbsolutePath.make(process.cwd()), title: "Parent", version: "test", }).run().pipe(Effect.orDie) @@ -59,8 +62,13 @@ function runTest( Effect.provideService(InstanceRef, { directory: process.cwd(), worktree: process.cwd(), - project: { id: "project-1" }, - } as never), + project: { + id: Project.ID.make("project-1"), + worktree: process.cwd(), + time: { created: 0, updated: 0 }, + sandboxes: [], + }, + }), Effect.scoped, ) }) diff --git a/packages/opencode/test/dag/dag-timeout-escalation.test.ts b/packages/opencode/test/dag/dag-timeout-escalation.test.ts index ae299a3a25..1ba5489df6 100644 --- a/packages/opencode/test/dag/dag-timeout-escalation.test.ts +++ b/packages/opencode/test/dag/dag-timeout-escalation.test.ts @@ -5,15 +5,19 @@ import { Database } from "@opencode-ai/core/database/database" import { DagProjector } from "@opencode-ai/core/dag/projector" import { DagStore } from "@opencode-ai/core/dag/store" import { EventV2 } from "@opencode-ai/core/event" +import { Project } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionTable } from "@opencode-ai/core/session/sql" +import { Model } from "@opencode-ai/schema/model" +import { Provider } from "@opencode-ai/schema/provider" import { Agent } from "@/agent/agent" import { Dag, type NodeConfig } from "@/dag/dag" import { DagLoop } from "@/dag/runtime/loop" import { InstanceRef } from "@/effect/instance-ref" import { EventV2Bridge } from "@/event-v2-bridge" import { SessionPrompt } from "@/session/prompt" -import { MessageID } from "@/session/schema" +import { MessageID, PartID, SessionID } from "@/session/schema" import { Session } from "@/session/session" import { SessionStatus } from "@/session/status" import { pollWithTimeout } from "../lib/effect" @@ -39,23 +43,24 @@ function takeWithin(queue: Queue.Queue, message: string) { } function reply(sessionID: string, text: string): SessionV1.WithParts { + const id = MessageID.ascending() return { info: { - id: MessageID.ascending(), + id, role: "assistant", parentID: MessageID.ascending(), - sessionID: sessionID as never, + sessionID: SessionID.make(sessionID), mode: "build", agent: "build", cost: 0, path: { cwd: process.cwd(), root: process.cwd() }, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, - modelID: "test-model" as never, - providerID: "test" as never, + modelID: Model.ID.make("test-model"), + providerID: Provider.ID.make("test"), time: { created: Date.now() }, finish: "stop", }, - parts: text ? [{ type: "text", text }] as never : [], + parts: text ? [{ id: PartID.ascending(), sessionID: SessionID.make(sessionID), messageID: id, type: "text", text }] : [], } } @@ -113,13 +118,31 @@ function loopLayer(input: { const created: string[] = [] let cancelCount = 0 const session = Layer.mock(Session.Service, { - get: () => Effect.succeed({ id: "ses_parent", permission: [], agent: "build" } as never), + get: () => Effect.succeed({ + id: SessionID.make("ses_parent"), + slug: "parent", + projectID: Project.ID.make("project-1"), + directory: process.cwd(), + title: "Parent", + version: "test", + time: { created: 0, updated: 0 }, + permission: [], + agent: "build", + }), create: (value) => Effect.sync(() => { const id = `ses_child_${created.length + 1}` created.push(id) childTitles.set(id, (value?.title ?? id).replace(" (DAG node)", "")) - return { id } as never + return { + id: SessionID.make(id), + slug: "child", + projectID: Project.ID.make("project-1"), + directory: process.cwd(), + title: value?.title ?? id, + version: "test", + time: { created: 0, updated: 0 }, + } }), messages: () => Effect.succeed([]), }) @@ -153,7 +176,7 @@ function loopLayer(input: { options: {}, description: "", prompt: "", - model: { providerID: "test" as never, modelID: "test-model" as never }, + model: { providerID: Provider.ID.make("test"), modelID: Model.ID.make("test-model") }, tools: {}, hooks: {}, }), @@ -190,15 +213,15 @@ function runLoopTest( const store = yield* DagStore.Service const database = yield* Database.Service yield* database.db.insert(ProjectTable).values({ - id: "project-1" as never, - worktree: process.cwd() as never, + id: Project.ID.make("project-1"), + worktree: AbsolutePath.make(process.cwd()), sandboxes: [], }).run().pipe(Effect.orDie) yield* database.db.insert(SessionTable).values({ - id: "ses_parent" as never, - project_id: "project-1" as never, + id: SessionID.make("ses_parent"), + project_id: Project.ID.make("project-1"), slug: "parent", - directory: process.cwd() as never, + directory: AbsolutePath.make(process.cwd()), title: "Parent", version: "test", }).run().pipe(Effect.orDie) @@ -215,8 +238,13 @@ function runLoopTest( Effect.provideService(InstanceRef, { directory: process.cwd(), worktree: process.cwd(), - project: { id: "project-1" }, - } as never), + project: { + id: Project.ID.make("project-1"), + worktree: process.cwd(), + time: { created: 0, updated: 0 }, + sandboxes: [], + }, + }), Effect.scoped, ) })