From 37ee09c774c83c353888d7bd879f0e3b3a8c28c9 Mon Sep 17 00:00:00 2001 From: Lex Date: Tue, 18 Aug 2026 23:58:34 +0800 Subject: [PATCH 1/7] fix(dag): host-level supervision sweep settles nodes orphaned by instance teardown (production incident 2026-08-18) --- .../src/dag/runtime/supervision-sweep.ts | 164 +++++++++ packages/opencode/src/effect/app-runtime.ts | 6 + .../test/dag/dag-node-supervision.test.ts | 329 ++++++++++++++++++ 3 files changed, 499 insertions(+) create mode 100644 packages/opencode/src/dag/runtime/supervision-sweep.ts create mode 100644 packages/opencode/test/dag/dag-node-supervision.test.ts diff --git a/packages/opencode/src/dag/runtime/supervision-sweep.ts b/packages/opencode/src/dag/runtime/supervision-sweep.ts new file mode 100644 index 0000000000..77442be9d0 --- /dev/null +++ b/packages/opencode/src/dag/runtime/supervision-sweep.ts @@ -0,0 +1,164 @@ +// SPDX-FileCopyrightText: 2026 LeXwDeX +// SPDX-License-Identifier: AGPL-3.0-or-later + +export * as DagSupervisionSweep from "./supervision-sweep" + +import { Context, Effect, Fiber, Layer, Scope } from "effect" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Database } from "@opencode-ai/core/database/database" +import { WorkflowNodeTable } from "@opencode-ai/core/dag/sql" +import { and, eq, sql } from "drizzle-orm" +import { Dag } from "@/dag/dag" +import { DagLocation } from "@/dag/location" +import { InstanceState } from "@/effect/instance-state" +import { SessionPrompt } from "@/session/prompt" + +/** + * Host-level deadline-supervision sweep — the fallback retry for the + * production incident (2026-08-18, dag_fe5feabfcae607fqVdRh47lN1B): + * + * A per-directory instance teardown (lifecycle cleanup, directory switch, + * config change) silently reaps every fiber forked into its scope — the + * DagLoop subscriptions, the spawn execution fiber, AND the deadline + * watcher — while the durable node row stays `running`. Because the host + * process keeps running, nothing ever re-arms supervision: the node rots in + * `running` past its deadline with `timeout_extensions` frozen for hours + * (7.5h observed). Re-init/crash recovery CAN settle such rows, but only + * when something re-triggers the instance — and in a live host nothing + * does. + * + * This sweep is deliberately NOT forked into any per-directory + * InstanceState scope: its repeating fiber is forked into the LAYER scope at + * construction (per AGENTS.md's background-loop convention) and lives for + * the process lifetime. Each tick looks for the frozen signature — a + * `running` node whose deadline has passed and whose `timeout_extensions` + * did not move between two consecutive ticks (a live watcher escalates on + * its interval, so a frozen counter across a full tick window means + * supervision is gone). On detection it cancels the child session and fails + * the node durably ("timeout") — the same terminal semantics the watcher's + * cap enforcement would have applied. + * + * False-positive safety: a node whose watcher is alive always shows counter + * movement across two ticks (escalateIntervalMs == max(1s, timeoutMs), far + * below the sweep interval); a node that terminalized races safely — the + * nodeFailed guard rejects the stale write. + */ + +export interface Interface { + /** Re-arm the periodic sweep fiber (idempotent). Production layers fork it at construction; init exists for entry points that prefer explicit control. */ + readonly init: () => Effect.Effect + /** One scan pass. Exported for deterministic tests: call twice with the freeze window in between to simulate a dead watcher. */ + readonly sweepOnce: () => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/DagSupervisionSweep") {} + +export const SWEEP_INTERVAL = "60 seconds" + +const serviceLayer = Layer.effect( + Service, + Effect.gen(function* () { + const { db } = yield* Database.Service + const dag = yield* Dag.Service + const promptSvc = yield* SessionPrompt.Service + const scope = yield* Scope.Scope + + // nodeKey -> timeout_extensions observed at the previous tick. A running, + // deadline-overdue node whose counter did not advance across a tick is + // frozen (supervision dead). + const lastSeen = new Map() + let sweepFiber: Fiber.Fiber | undefined + + const sweepOnce = Effect.fn("DagSupervisionSweep.sweepOnce")(function* () { + const rows = yield* db + .select({ + workflowId: WorkflowNodeTable.workflow_id, + nodeId: WorkflowNodeTable.id, + childSessionId: WorkflowNodeTable.child_session_id, + deadlineMs: WorkflowNodeTable.deadline_ms, + extensions: WorkflowNodeTable.timeout_extensions, + }) + .from(WorkflowNodeTable) + .where( + and( + eq(WorkflowNodeTable.status, "running"), + sql`${WorkflowNodeTable.deadline_ms} IS NOT NULL AND ${WorkflowNodeTable.deadline_ms} <= ${Date.now()}`, + ), + ) + .all() + .pipe(Effect.orDie) + + const observed = new Map() + for (const row of rows) { + const key = `${row.workflowId}\0${row.nodeId}` + observed.set(key, row.extensions) + const previous = lastSeen.get(key) + if (previous === undefined) continue + if (previous !== row.extensions) continue // counter moved — watcher alive + // Frozen across a full tick: confirm this instance still owns the + // workflow before writing (a repainted or migrated identity belongs + // to whichever instance now owns it). + if (!(yield* DagLocation.ownsWorkflow(row.workflowId, yield* InstanceState.directory))) continue + if (row.childSessionId) { + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- the durable column is typed string|null; the cancel seam brands SessionID. + yield* promptSvc.cancel(row.childSessionId as never).pipe(Effect.ignore) + } + yield* dag + .nodeFailed( + row.workflowId, + row.nodeId, + `deadline supervision lost (no escalation progress across sweep window) — swept, extensions ${row.extensions}`, + "timeout", + ) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("DagSupervisionSweep nodeFailed failed", { + dagID: row.workflowId, + nodeID: row.nodeId, + cause, + }), + ), + ) + yield* Effect.logWarning("DagSupervisionSweep settled a node with dead deadline supervision", { + dagID: row.workflowId, + nodeID: row.nodeId, + extensions: row.extensions, + }) + observed.delete(key) + } + // Retain only what is still overdue-running so settled/restarted nodes + // do not accumulate. + lastSeen.clear() + for (const [key, extensions] of observed) lastSeen.set(key, extensions) + }) + + const init = Effect.fn("DagSupervisionSweep.init")(function* () { + if (sweepFiber) return + sweepFiber = yield* Effect.gen(function* () { + for (;;) { + yield* Effect.sleep(SWEEP_INTERVAL) + yield* sweepOnce() + } + }).pipe(Effect.forkIn(scope)) + }) + + // AGENTS.md background-loop convention: fork at construction so the sweep + // survives without any caller remembering to init it. + yield* init() + + return Service.of({ init, sweepOnce }) + }), +) + +/** The bare effect layer — bring your own Database/Dag/SessionPrompt. Tests compose this against their mocks; production uses `defaultLayer`. */ +export const layerWithoutDeps = serviceLayer + +export const layer = serviceLayer.pipe( + Layer.provide(Database.defaultLayer), + Layer.provide(Dag.defaultLayer), + Layer.provide(SessionPrompt.defaultLayer), +) + +export const defaultLayer = layer + +export const node = LayerNode.make(serviceLayer, [Database.node, Dag.node, SessionPrompt.node]) diff --git a/packages/opencode/src/effect/app-runtime.ts b/packages/opencode/src/effect/app-runtime.ts index d9edca954c..435b173794 100644 --- a/packages/opencode/src/effect/app-runtime.ts +++ b/packages/opencode/src/effect/app-runtime.ts @@ -60,6 +60,7 @@ import { Dag } from "@/dag/dag" import { DagStore } from "@opencode-ai/core/dag/store" import { DagLoop } from "@/dag/runtime/loop" import { DagSummaryPublisher } from "@/dag/runtime/summary-publisher" +import { DagSupervisionSweep } from "@/dag/runtime/supervision-sweep" import { Memory } from "@/memory/memory" export const AppLayer = Layer.mergeAll( @@ -132,6 +133,11 @@ export const AppLayer = Layer.mergeAll( Layer.provideMerge(GoalLoop.defaultLayer), Layer.provideMerge(DagLoop.defaultLayer), Layer.provideMerge(DagSummaryPublisher.defaultLayer), + // Host-level deadline-supervision sweep (production incident 2026-08-18): + // forks its repeating fiber at construction into the LAYER scope — unlike + // DagLoop it must NOT die with a per-directory instance teardown, or a + // `running` node with dead supervision would rot forever. + Layer.provideMerge(DagSupervisionSweep.defaultLayer), Layer.provideMerge(SettingsHook.defaultLayer), ) diff --git a/packages/opencode/test/dag/dag-node-supervision.test.ts b/packages/opencode/test/dag/dag-node-supervision.test.ts new file mode 100644 index 0000000000..d899320f6c --- /dev/null +++ b/packages/opencode/test/dag/dag-node-supervision.test.ts @@ -0,0 +1,329 @@ +// oxlint-disable typescript-eslint/no-unsafe-type-assertion -- The incident +// harness deliberately mirrors dag-loop-guards.test.ts: mocked service layers +// and seeded row fixtures use `as never` type shims (mock objects implement +// only the interface slice the scenario exercises). The shims are type-only; +// converting them would fork the template's shape without changing behavior. +// oxlint-disable eslint/no-unused-vars -- gate objects are taken for their +// readiness side effect (takeWithin), not their value. +import { describe, expect, it } from "bun:test" +import { Deferred, Effect, Layer, Option, Queue } from "effect" +import type { SessionV1 } from "@opencode-ai/core/v1/session" +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 { ProjectTable } from "@opencode-ai/core/project/sql" +import { SessionTable } from "@opencode-ai/core/session/sql" +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 { disposeInstance } from "@/effect/instance-registry" +import { DagSupervisionSweep } from "@/dag/runtime/supervision-sweep" +import { EventV2Bridge } from "@/event-v2-bridge" +import { SessionPrompt } from "@/session/prompt" +import { MessageID } from "@/session/schema" +import { Session } from "@/session/session" +import { SessionStatus } from "@/session/status" +import { pollWithTimeout } from "../lib/effect" +import { withIdleAdmission } from "../lib/session-prompt" + +// Production-incident harness (2026-08-18, dag_fe5feabfcae607fqVdRh47lN1B): +// a coding node's child session LLM stream died silently mid-turn; the node +// stayed `running` past its deadline for 7.5+ hours with escalation_pending=0 +// and timeout_extensions=0 — the deadline watcher never fired while the host +// process stayed alive. This harness reproduces the supervision shape at +// 2-second deadlines and asserts the invariant the incident violated: +// +// A running node past its deadline must leave `running` (escalate or fail) +// within a bounded window — no matter HOW the surrounding fibers die. +// +// Modes cover the candidate death paths: +// stream-hang — the child prompt never resolves (incident shape) +// dispose-instance — the per-directory instance scope closes mid-run +// healthy — control: the watcher fires normally + +interface PromptGate { + readonly title: string + readonly release: Deferred.Deferred +} + +function node(overrides: Partial = {}): NodeConfig { + return { + id: "n1", + name: "Node 1", + worker_type: "build", + depends_on: [], + required: true, + prompt_template: { inline: "work" }, + ...overrides, + } +} + +function takeWithin(queue: Queue.Queue, message: string) { + return Queue.take(queue).pipe( + Effect.timeoutOption("2 seconds"), + Effect.flatMap(Option.match({ onNone: () => Effect.fail(new Error(message)), onSome: Effect.succeed })), + ) +} + +function reply(sessionID: string, text: string): SessionV1.WithParts { + return { + info: { + id: MessageID.ascending(), + sessionID, + role: "assistant", + time: { created: Date.now() }, + }, + parts: [{ type: "text", text }], + } as never +} + +function supervisionLayer(input: { + readonly childPrompts: Queue.Queue + readonly cancels: string[] +}) { + 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 status = SessionStatus.layer.pipe(Layer.provide(bridge)) + const projector = DagProjector.layer.pipe(Layer.provide(events), Layer.provide(database)) + const dag = Dag.layer.pipe(Layer.provide(bridge), Layer.provide(store)) + const base = Layer.mergeAll(database, events, bridge, store, projector, dag, status) + const childTitles = new Map() + const created: string[] = [] + const session = Layer.mock(Session.Service, { + get: () => Effect.succeed({ id: "ses_parent", permission: [], agent: "build" } as never), + 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 + }), + messages: () => Effect.succeed([]), + }) + const deliver = Effect.fn("test.SessionPrompt.deliver")(function* (value: SessionPrompt.PromptInput) { + const sessionID = value.sessionID as string + const release = yield* Deferred.make() + yield* Queue.offer(input.childPrompts, { + title: childTitles.get(sessionID) ?? sessionID, + release, + }) + return reply(sessionID, yield* Deferred.await(release)) + }) + const prompt = Layer.mock( + SessionPrompt.Service, + withIdleAdmission({ + cancel: (sessionID: string) => + Effect.sync(() => { + input.cancels.push(sessionID) + }), + prompt: (value: SessionPrompt.PromptInput) => deliver(value), + promptIfIdle: (value: SessionPrompt.PromptInput) => deliver(value).pipe(Effect.map(Option.some)), + }), + ) + const agent = Layer.mock(Agent.Service, { + get: () => + Effect.succeed({ + name: "build", + mode: "all", + permission: [], + options: {}, + description: "", + prompt: "", + model: { providerID: "test" as never, modelID: "test-model" as never }, + tools: {}, + hooks: {}, + }), + }) + const loop = DagLoop.layer.pipe(Layer.provide(base), Layer.provide(session), Layer.provide(prompt), Layer.provide(agent)) + const sweep = DagSupervisionSweep.layerWithoutDeps.pipe(Layer.provide(base), Layer.provide(prompt)) + return Layer.merge(Layer.merge(base, loop), sweep) +} + +interface SupervisionServices { + readonly dag: Dag.Interface + readonly loop: DagLoop.Interface + readonly store: DagStore.Interface + readonly sweep: import("@/dag/runtime/supervision-sweep").Interface + readonly childPrompts: Queue.Queue + readonly cancels: string[] + readonly database: Database.Interface +} +function runSupervisionTest(options: { readonly instanceProject: string }, test: (services: SupervisionServices) => Effect.Effect) { + return Effect.gen(function* () { + const childPrompts = yield* Queue.unbounded() + const cancels: string[] = [] + return yield* Effect.gen(function* () { + const dag = yield* Dag.Service + const loop = yield* DagLoop.Service + const store = yield* DagStore.Service + const sweep = yield* DagSupervisionSweep.Service + const database = yield* Database.Service + for (const project of ["project-1", "project-2"]) { + yield* database.db + .insert(ProjectTable) + .values({ id: project as never, worktree: process.cwd() as never, sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* database.db + .insert(SessionTable) + .values({ + id: `ses_${project}` as never, + project_id: project as never, + slug: project, + directory: process.cwd() as never, + title: `Parent of ${project}`, + version: "test", + }) + .run() + .pipe(Effect.orDie) + } + yield* loop.init() + return yield* test({ dag, loop, store, sweep, childPrompts, cancels, database }) + }).pipe( + Effect.provide(supervisionLayer({ childPrompts, cancels })), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: options.instanceProject }, + } as never), + Effect.scoped, + ) + }) +} + +// Shared graph: one coding node with a 2-second worker timeout. nodeTimeoutMs +// default would make the deadline 10 minutes — unusable for a test. The cap +// is pinned to 1 escalation so cap enforcement lands inside the test window. +const incidentGraph = { + projectID: "project-1", + sessionID: "ses_project-1", + title: "incident", + config: { + name: "incident", + max_timeout_extensions: 1, + nodes: [node({ id: "worker", name: "worker", worker_config: { timeout_ms: 2_000 } })], + }, +} + +// The incident invariant, as a poll predicate: the node must leave `running` +// (any terminal status, or escalated-but-running counts as progress only if +// extensions climb — the incident had BOTH frozen at zero, so we assert on +// status change OR timeout_extensions > 0). +const supervisionProgress = (store: DagStore.Interface, dagID: string, nodeID: string) => + Effect.gen(function* () { + const row = yield* store.getNode(dagID, nodeID) + if (!row) return undefined + if (row.status !== "running") return row + if (row.timeoutExtensions > 0) return row + return undefined + }) + +// bun's default per-test timeout is 5s; the healthy cap-enforcement path +// needs ~8s at a 2s deadline — run this file with --timeout 30000 (the CI +// suite default) or keep each body under the limit. +describe("DAG node supervision — deadline enforcement (production incident)", () => { + it("healthy: a node past its deadline gets escalated by the watcher", async () => { + await Effect.runPromise( + runSupervisionTest({ instanceProject: "project-1" }, ({ dag, store, childPrompts, cancels }) => + Effect.gen(function* () { + const dagID = yield* dag.create(incidentGraph) + const child = yield* takeWithin(childPrompts, "worker did not start") + // Leave the prompt unresolved past the 2s deadline: the watcher + // must escalate (timeout_extensions climbs), then exhaust the cap + // and force-cancel the child. + yield* pollWithTimeout( + supervisionProgress(store, dagID, "worker"), + "watcher never escalated a node past its deadline (healthy control)", + "8 seconds", + ) + // Cap enforcement: max extensions default is 3 — after enough + // escalations the watcher cancels the child and fails the node. + yield* pollWithTimeout( + Effect.gen(function* () { + const row = yield* store.getNode(dagID, "worker") + return row?.status === "failed" ? row : undefined + }), + "watcher never cap-enforced (cancel + nodeFailed(timeout))", + "30 seconds", + ) + expect(cancels.length).toBeGreaterThan(0) + }), + ), + ) + }) + + it("stream-hang: the incident shape — prompt never resolves, node still must not rot in running", async () => { + await Effect.runPromise( + runSupervisionTest({ instanceProject: "project-1" }, ({ dag, store, childPrompts }: SupervisionServices ) => + Effect.gen(function* () { + const dagID = yield* dag.create(incidentGraph) + const child = yield* takeWithin(childPrompts, "worker did not start") + // Incident shape: the LLM stream died — the prompt gate is never + // released and never errors. Supervision must still progress. + void child + yield* pollWithTimeout( + supervisionProgress(store, dagID, "worker"), + "node rotted in running past its deadline with zero supervision progress (incident)", + "8 seconds", + ) + }), + ), + ) + }) + + // H5 (instance-scope harvest): closing the per-directory instance state + // mid-run interrupts every fiber forked into its scope — the DagLoop + // subscriptions, the spawn execution fiber, AND the deadline watcher — + // without touching the durable row. The production signature (a node stuck + // in running with escalation frozen at zero for hours while the host kept + // logging) is only reachable if supervision dies silently this way. + it("dispose-instance: instance teardown mid-run freezes durable supervision (incident mechanism)", async () => { + await Effect.runPromise( + runSupervisionTest({ instanceProject: "project-1" }, ({ dag, store, sweep, childPrompts }: SupervisionServices) => + Effect.gen(function* () { + const dagID = yield* dag.create(incidentGraph) + const child = yield* takeWithin(childPrompts, "worker did not start") + void child + // Wait for the first escalation so we know supervision was live. + yield* pollWithTimeout( + supervisionProgress(store, dagID, "worker"), + "watcher never escalated before dispose", + "8 seconds", + ) + const extensionsAtDispose = (yield* store.getNode(dagID, "worker"))?.timeoutExtensions ?? 0 + // Dispose the instance (the production candidate: lifecycle/ + // directory cleanup) — silently reaps every in-scope fiber. + yield* Effect.promise(() => disposeInstance(process.cwd())) + // Give any surviving supervision ample time to escalate again. + yield* Effect.sleep("4 seconds") + const row = yield* store.getNode(dagID, "worker") + // The frozen-supervision signature: still running, extensions + // frozen at the dispose-time value, no cap enforcement. + expect(row?.status).toBe("running") + expect(row?.timeoutExtensions).toBe(extensionsAtDispose) + + // The fallback-retry contract (production fix): the HOST-LEVEL + // supervision sweep — whose fiber lives in the layer scope and + // survives the instance teardown — settles the frozen node on its + // second tick (frozen counter across ticks = dead supervision). + yield* sweep.sweepOnce() + yield* Effect.sleep("200 millis") + yield* sweep.sweepOnce() + const swept = yield* pollWithTimeout( + Effect.gen(function* () { + const settled = yield* store.getNode(dagID, "worker") + return settled && settled.status !== "running" ? settled : undefined + }), + "host-level sweep never settled the node with dead supervision (fallback retry)", + "5 seconds", + ) + expect(swept?.errorClass).toBe("timeout") + }), + ), + ) + }) +}) From 9b76a87f65d529526a735619eb80a3913808db3e Mon Sep 17 00:00:00 2001 From: Lex Date: Wed, 19 Aug 2026 00:13:27 +0800 Subject: [PATCH 2/7] fix(dag): sweep survives layer context (no ambient InstanceRef), cadence-aware freeze window (review R1) --- .../src/dag/runtime/supervision-sweep.ts | 103 +++++++++++------- .../test/dag/dag-node-supervision.test.ts | 63 +++++++++-- 2 files changed, 121 insertions(+), 45 deletions(-) diff --git a/packages/opencode/src/dag/runtime/supervision-sweep.ts b/packages/opencode/src/dag/runtime/supervision-sweep.ts index 77442be9d0..56be2f05c4 100644 --- a/packages/opencode/src/dag/runtime/supervision-sweep.ts +++ b/packages/opencode/src/dag/runtime/supervision-sweep.ts @@ -9,8 +9,6 @@ import { Database } from "@opencode-ai/core/database/database" import { WorkflowNodeTable } from "@opencode-ai/core/dag/sql" import { and, eq, sql } from "drizzle-orm" import { Dag } from "@/dag/dag" -import { DagLocation } from "@/dag/location" -import { InstanceState } from "@/effect/instance-state" import { SessionPrompt } from "@/session/prompt" /** @@ -28,26 +26,31 @@ import { SessionPrompt } from "@/session/prompt" * does. * * This sweep is deliberately NOT forked into any per-directory - * InstanceState scope: its repeating fiber is forked into the LAYER scope at - * construction (per AGENTS.md's background-loop convention) and lives for - * the process lifetime. Each tick looks for the frozen signature — a - * `running` node whose deadline has passed and whose `timeout_extensions` - * did not move between two consecutive ticks (a live watcher escalates on - * its interval, so a frozen counter across a full tick window means - * supervision is gone). On detection it cancels the child session and fails - * the node durably ("timeout") — the same terminal semantics the watcher's - * cap enforcement would have applied. + * InstanceState scope: its repeating fiber is forked into the LAYER scope + * at construction and lives for the process lifetime. It must therefore + * never depend on ambient per-instance context (InstanceRef) — its fiber's + * context is the layer-build context, which has none. Ownership is decided + * from the durable rows alone: this process's Database owns every workflow + * row it can read, and the nodeFailed guard under the workflow lock + * serializes any race with another writer (including a second host sharing + * the DB — double settles collapse to one). * - * False-positive safety: a node whose watcher is alive always shows counter - * movement across two ticks (escalateIntervalMs == max(1s, timeoutMs), far - * below the sweep interval); a node that terminalized races safely — the - * nodeFailed guard rejects the stale write. + * False-positive safety: a LIVE watcher escalates on + * escalateIntervalMs == max(1s, timeout_ms ?? 10min) and nodeTimeoutEscalated + * does NOT move deadline_ms — so a live overdue node legitimately shows a + * flat timeout_extensions for up to one full escalate interval (10 minutes + * on the default config). The freeze window is therefore expressed in + * ticks: a node is only declared dead once its counter has stayed flat for + * frozenTicksNeeded(escalateIntervalMs) consecutive sweep ticks — the + * default 10-minute cadence needs 11 ticks (≈11 minutes), so a live watcher + * always moves the counter well inside the window, while a dead one (the + * incident shape: 7.5h frozen) is settled in bounded time. */ export interface Interface { - /** Re-arm the periodic sweep fiber (idempotent). Production layers fork it at construction; init exists for entry points that prefer explicit control. */ + /** Re-arm the periodic sweep fiber (idempotent; production layers fork it at construction). */ readonly init: () => Effect.Effect - /** One scan pass. Exported for deterministic tests: call twice with the freeze window in between to simulate a dead watcher. */ + /** One scan pass. Exported for deterministic tests: loop it frozenTicksNeeded times to simulate a dead watcher. */ readonly sweepOnce: () => Effect.Effect } @@ -55,6 +58,16 @@ export class Service extends Context.Service()("@opencode/Da export const SWEEP_INTERVAL = "60 seconds" +/** + * Ticks a flat timeout_extensions counter must persist across before the + * sweep declares supervision dead: ceil(escalateInterval / sweepInterval) + 1, + * evaluated against the DEFAULT node timeout (10 min) — the widest cadence a + * live watcher can legitimately sleep. Nodes configured with shorter + * timeouts escalate faster, so they are only ever settled later than + * strictly necessary, never sooner. + */ +export const FROZEN_TICKS_NEEDED = 11 + const serviceLayer = Layer.effect( Service, Effect.gen(function* () { @@ -63,11 +76,11 @@ const serviceLayer = Layer.effect( const promptSvc = yield* SessionPrompt.Service const scope = yield* Scope.Scope - // nodeKey -> timeout_extensions observed at the previous tick. A running, - // deadline-overdue node whose counter did not advance across a tick is - // frozen (supervision dead). - const lastSeen = new Map() - let sweepFiber: Fiber.Fiber | undefined + // nodeKey -> {extensions, flatTicks}: the counter value last observed and + // how many consecutive sweep ticks it has stayed flat while the node was + // running and overdue. Reset on any counter movement, terminal status, or + // disappearance from the query. + const flatStreak = new Map() const sweepOnce = Effect.fn("DagSupervisionSweep.sweepOnce")(function* () { const rows = yield* db @@ -75,7 +88,6 @@ const serviceLayer = Layer.effect( workflowId: WorkflowNodeTable.workflow_id, nodeId: WorkflowNodeTable.id, childSessionId: WorkflowNodeTable.child_session_id, - deadlineMs: WorkflowNodeTable.deadline_ms, extensions: WorkflowNodeTable.timeout_extensions, }) .from(WorkflowNodeTable) @@ -86,19 +98,28 @@ const serviceLayer = Layer.effect( ), ) .all() - .pipe(Effect.orDie) + .pipe( + // A store defect must not kill the sweep — the same silent-death + // class this service exists to eliminate. Degrade to an empty + // pass and retry next tick (mirror of spawn.ts's R13 hardening). + Effect.catchCause((cause) => + Effect.gen(function* () { + yield* Effect.logWarning("DagSupervisionSweep store query failed — skipping tick", { cause }) + return [] + }), + ), + ) - const observed = new Map() + const observed = new Map() for (const row of rows) { const key = `${row.workflowId}\0${row.nodeId}` - observed.set(key, row.extensions) - const previous = lastSeen.get(key) - if (previous === undefined) continue - if (previous !== row.extensions) continue // counter moved — watcher alive - // Frozen across a full tick: confirm this instance still owns the - // workflow before writing (a repainted or migrated identity belongs - // to whichever instance now owns it). - if (!(yield* DagLocation.ownsWorkflow(row.workflowId, yield* InstanceState.directory))) continue + const prior = flatStreak.get(key) + const flatTicks = prior && prior.extensions === row.extensions ? prior.flatTicks + 1 : 0 + observed.set(key, { extensions: row.extensions, flatTicks }) + if (flatTicks < FROZEN_TICKS_NEEDED) continue + // Frozen across the full window: cancel the (possibly dead) child and + // settle the node. The nodeFailed guard under the workflow lock + // serializes any race with a live watcher or another host's sweep. if (row.childSessionId) { // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- the durable column is typed string|null; the cancel seam brands SessionID. yield* promptSvc.cancel(row.childSessionId as never).pipe(Effect.ignore) @@ -107,7 +128,7 @@ const serviceLayer = Layer.effect( .nodeFailed( row.workflowId, row.nodeId, - `deadline supervision lost (no escalation progress across sweep window) — swept, extensions ${row.extensions}`, + `deadline supervision lost (no escalation progress across ${FROZEN_TICKS_NEEDED} sweep ticks) — swept, extensions ${row.extensions}`, "timeout", ) .pipe( @@ -128,16 +149,24 @@ const serviceLayer = Layer.effect( } // Retain only what is still overdue-running so settled/restarted nodes // do not accumulate. - lastSeen.clear() - for (const [key, extensions] of observed) lastSeen.set(key, extensions) + flatStreak.clear() + for (const [key, streak] of observed) flatStreak.set(key, streak) }) + let sweepFiber: Fiber.Fiber | undefined + const init = Effect.fn("DagSupervisionSweep.init")(function* () { if (sweepFiber) return sweepFiber = yield* Effect.gen(function* () { for (;;) { yield* Effect.sleep(SWEEP_INTERVAL) - yield* sweepOnce() + yield* sweepOnce().pipe( + // Per-tick guard: any residual defect inside a tick degrades to + // a logged skip — the loop itself must outlive every failure. + Effect.catchCause((cause) => + Effect.logWarning("DagSupervisionSweep tick failed — retrying next interval", { cause }), + ), + ) } }).pipe(Effect.forkIn(scope)) }) diff --git a/packages/opencode/test/dag/dag-node-supervision.test.ts b/packages/opencode/test/dag/dag-node-supervision.test.ts index d899320f6c..a01d61616f 100644 --- a/packages/opencode/test/dag/dag-node-supervision.test.ts +++ b/packages/opencode/test/dag/dag-node-supervision.test.ts @@ -222,9 +222,8 @@ const supervisionProgress = (store: DagStore.Interface, dagID: string, nodeID: s return undefined }) -// bun's default per-test timeout is 5s; the healthy cap-enforcement path -// needs ~8s at a 2s deadline — run this file with --timeout 30000 (the CI -// suite default) or keep each body under the limit. +// bun's default per-test timeout is 5s; several bodies here need 8-40s at a +// 2s deadline — run this file with --timeout 30000 (the CI suite default). describe("DAG node supervision — deadline enforcement (production incident)", () => { it("healthy: a node past its deadline gets escalated by the watcher", async () => { await Effect.runPromise( @@ -308,11 +307,15 @@ describe("DAG node supervision — deadline enforcement (production incident)", // The fallback-retry contract (production fix): the HOST-LEVEL // supervision sweep — whose fiber lives in the layer scope and - // survives the instance teardown — settles the frozen node on its - // second tick (frozen counter across ticks = dead supervision). - yield* sweep.sweepOnce() - yield* Effect.sleep("200 millis") - yield* sweep.sweepOnce() + // survives the instance teardown — settles the frozen node once + // its counter has stayed flat for FROZEN_TICKS_NEEDED ticks + // (dead supervision; a live watcher always moves the counter + // inside the window). + const { FROZEN_TICKS_NEEDED } = DagSupervisionSweep + for (let tick = 0; tick <= FROZEN_TICKS_NEEDED; tick++) { + yield* sweep.sweepOnce() + yield* Effect.sleep("50 millis") + } const swept = yield* pollWithTimeout( Effect.gen(function* () { const settled = yield* store.getNode(dagID, "worker") @@ -326,4 +329,48 @@ describe("DAG node supervision — deadline enforcement (production incident)", ), ) }) + + // Review R1 issue 2 (false-positive kill): a LIVE watcher on a node whose + // escalation cadence spans multiple sweep intervals must never be swept — + // the counter legitimately stays flat between escalations. The graph keeps + // the default cap (20) so the watcher's ladder is the intended path; the + // sweep passes run alongside a live watcher for well over the freeze + // window, and the settle that eventually lands must carry the WATCHER's + // own cap reason, never the sweep's. + it("freeze window: a live watcher is never swept — only its own cap enforcement ends the node", async () => { + await Effect.runPromise( + runSupervisionTest({ instanceProject: "project-1" }, ({ dag, store, sweep, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + ...incidentGraph, + config: { ...incidentGraph.config, max_timeout_extensions: 3 }, + }) + const child = yield* takeWithin(childPrompts, "worker did not start") + void child + // Supervision alive: the watcher escalates on its 2s cadence. + yield* pollWithTimeout( + supervisionProgress(store, dagID, "worker"), + "watcher never escalated before the streak test", + "8 seconds", + ) + const { FROZEN_TICKS_NEEDED } = DagSupervisionSweep + // Run more sweep passes than the freeze window alongside the live + // watcher: its 2s escalation cadence resets the streak every time, + // so the sweep must never fire. + for (let tick = 0; tick < FROZEN_TICKS_NEEDED + 2; tick++) { + yield* Effect.sleep("300 millis") + yield* sweep.sweepOnce() + } + const row = yield* store.getNode(dagID, "worker") + // Either still running (ladder ongoing) or terminalized by the + // watcher's OWN cap — never by the sweep. + if (row?.status === "failed") { + expect(row?.errorReason).toContain("timeout extensions exhausted") + } else { + expect(row?.status).toBe("running") + } + }), + ), + ) + }) }) From 0d11e05263376f5c6ea4b6b37081272056a332d8 Mon Sep 17 00:00:00 2001 From: Lex Date: Wed, 19 Aug 2026 00:25:46 +0800 Subject: [PATCH 3/7] fix(dag): per-node cadence-aware freeze window from workflow config (review R2) --- .../src/dag/runtime/supervision-sweep.ts | 46 +++++++++++++++---- .../test/dag/dag-node-supervision.test.ts | 26 ++++++----- 2 files changed, 51 insertions(+), 21 deletions(-) diff --git a/packages/opencode/src/dag/runtime/supervision-sweep.ts b/packages/opencode/src/dag/runtime/supervision-sweep.ts index 56be2f05c4..bd49d85649 100644 --- a/packages/opencode/src/dag/runtime/supervision-sweep.ts +++ b/packages/opencode/src/dag/runtime/supervision-sweep.ts @@ -6,6 +6,7 @@ export * as DagSupervisionSweep from "./supervision-sweep" import { Context, Effect, Fiber, Layer, Scope } from "effect" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Database } from "@opencode-ai/core/database/database" +import { DagStore } from "@opencode-ai/core/dag/store" import { WorkflowNodeTable } from "@opencode-ai/core/dag/sql" import { and, eq, sql } from "drizzle-orm" import { Dag } from "@/dag/dag" @@ -57,21 +58,28 @@ export interface Interface { export class Service extends Context.Service()("@opencode/DagSupervisionSweep") {} export const SWEEP_INTERVAL = "60 seconds" +const SWEEP_INTERVAL_MS = 60_000 /** - * Ticks a flat timeout_extensions counter must persist across before the - * sweep declares supervision dead: ceil(escalateInterval / sweepInterval) + 1, - * evaluated against the DEFAULT node timeout (10 min) — the widest cadence a - * live watcher can legitimately sleep. Nodes configured with shorter - * timeouts escalate faster, so they are only ever settled later than - * strictly necessary, never sooner. + * Flat ticks before declaring supervision dead, derived from the node's own + * escalation cadence: a LIVE watcher escalates every + * escalateIntervalMs == max(1s, timeout_ms ?? 10min) and never moves + * deadline_ms, so its counter can legitimately stay flat for up to one full + * interval. Requiring ceil(interval / sweep interval) + 1 consecutive flat + * ticks means a live watcher — at ANY configured timeout, including the + * doc-recommended 30-minute verifier timeouts — always moves the counter + * inside the window, while a dead one (the incident shape: hours frozen) is + * settled in bounded time. The config lookup happens once a node has been + * overdue-flat for at least one tick, so healthy graphs pay nothing. */ -export const FROZEN_TICKS_NEEDED = 11 +export const frozenTicksNeeded = (escalateIntervalMs: number) => + Math.ceil(Math.max(escalateIntervalMs, 1_000) / SWEEP_INTERVAL_MS) + 1 const serviceLayer = Layer.effect( Service, Effect.gen(function* () { const { db } = yield* Database.Service + const store = yield* DagStore.Service const dag = yield* Dag.Service const promptSvc = yield* SessionPrompt.Service const scope = yield* Scope.Scope @@ -82,6 +90,20 @@ const serviceLayer = Layer.effect( // disappearance from the query. const flatStreak = new Map() + // The node's escalation cadence, from the workflow's persisted config — + // the same source spawn.ts derived the watcher's escalateIntervalMs from. + // Cached per workflow id for the tick; a config read failure degrades to + // the DEFAULT cadence (the widest guaranteed-safe window). + const escalateIntervalFor = Effect.fnUntraced(function* (workflowId: string, nodeId: string) { + const wf = yield* store.getWorkflow(workflowId).pipe( + Effect.catchCause(() => Effect.succeed(undefined)), + ) + if (!wf) return Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs + const node = JSON.parse(wf.config).nodes?.find?.((n: { id: string }) => n.id === nodeId) + const timeoutMs = node?.worker_config?.timeout_ms + return Math.max(1_000, typeof timeoutMs === "number" ? timeoutMs : Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs) + }) + const sweepOnce = Effect.fn("DagSupervisionSweep.sweepOnce")(function* () { const rows = yield* db .select({ @@ -116,7 +138,10 @@ const serviceLayer = Layer.effect( const prior = flatStreak.get(key) const flatTicks = prior && prior.extensions === row.extensions ? prior.flatTicks + 1 : 0 observed.set(key, { extensions: row.extensions, flatTicks }) - if (flatTicks < FROZEN_TICKS_NEEDED) continue + // Only nodes already flat for a tick pay the config lookup. + if (flatTicks < 1) continue + const escalateIntervalMs = yield* escalateIntervalFor(row.workflowId, row.nodeId) + if (flatTicks < frozenTicksNeeded(escalateIntervalMs)) continue // Frozen across the full window: cancel the (possibly dead) child and // settle the node. The nodeFailed guard under the workflow lock // serializes any race with a live watcher or another host's sweep. @@ -128,7 +153,7 @@ const serviceLayer = Layer.effect( .nodeFailed( row.workflowId, row.nodeId, - `deadline supervision lost (no escalation progress across ${FROZEN_TICKS_NEEDED} sweep ticks) — swept, extensions ${row.extensions}`, + `deadline supervision lost (no escalation progress across ${flatTicks} sweep ticks, escalate cadence ${escalateIntervalMs}ms) — swept, extensions ${row.extensions}`, "timeout", ) .pipe( @@ -184,10 +209,11 @@ export const layerWithoutDeps = serviceLayer export const layer = serviceLayer.pipe( Layer.provide(Database.defaultLayer), + Layer.provide(DagStore.defaultLayer), Layer.provide(Dag.defaultLayer), Layer.provide(SessionPrompt.defaultLayer), ) export const defaultLayer = layer -export const node = LayerNode.make(serviceLayer, [Database.node, Dag.node, SessionPrompt.node]) +export const node = LayerNode.make(serviceLayer, [Database.node, DagStore.node, Dag.node, SessionPrompt.node]) diff --git a/packages/opencode/test/dag/dag-node-supervision.test.ts b/packages/opencode/test/dag/dag-node-supervision.test.ts index a01d61616f..94c791f5c7 100644 --- a/packages/opencode/test/dag/dag-node-supervision.test.ts +++ b/packages/opencode/test/dag/dag-node-supervision.test.ts @@ -308,11 +308,11 @@ describe("DAG node supervision — deadline enforcement (production incident)", // The fallback-retry contract (production fix): the HOST-LEVEL // supervision sweep — whose fiber lives in the layer scope and // survives the instance teardown — settles the frozen node once - // its counter has stayed flat for FROZEN_TICKS_NEEDED ticks - // (dead supervision; a live watcher always moves the counter - // inside the window). - const { FROZEN_TICKS_NEEDED } = DagSupervisionSweep - for (let tick = 0; tick <= FROZEN_TICKS_NEEDED; tick++) { + // its counter has stayed flat for frozenTicksNeeded(2s) = 2 ticks + // (dead supervision; a live 2s-cadence watcher always moves the + // counter inside the window). + const needed = DagSupervisionSweep.frozenTicksNeeded(2_000) + for (let tick = 0; tick <= needed; tick++) { yield* sweep.sweepOnce() yield* Effect.sleep("50 millis") } @@ -353,12 +353,16 @@ describe("DAG node supervision — deadline enforcement (production incident)", "watcher never escalated before the streak test", "8 seconds", ) - const { FROZEN_TICKS_NEEDED } = DagSupervisionSweep - // Run more sweep passes than the freeze window alongside the live - // watcher: its 2s escalation cadence resets the streak every time, - // so the sweep must never fire. - for (let tick = 0; tick < FROZEN_TICKS_NEEDED + 2; tick++) { - yield* Effect.sleep("300 millis") + const needed = DagSupervisionSweep.frozenTicksNeeded(2_000) + // Sweep passes at the PRODUCTION cadence relationship: each pass is + // spaced just past the node's 2s escalate interval, so the live + // watcher moves the counter between every pass and the streak can + // never reach `needed`. (Spacing the passes closer than the + // escalate interval would defeat the window's math — a live + // watcher's counter is legitimately flat for up to one full + // interval.) + for (let tick = 0; tick < needed + 2; tick++) { + yield* Effect.sleep("2.3 seconds") yield* sweep.sweepOnce() } const row = yield* store.getNode(dagID, "worker") From 5033463d2eb33f700052b67f3de19c4a4fc40233 Mon Sep 17 00:00:00 2001 From: Lex Date: Wed, 19 Aug 2026 00:37:07 +0800 Subject: [PATCH 4/7] fix(dag): defensive config parsing and pure cadence lookup in sweep (review R3) --- .../src/dag/runtime/supervision-sweep.ts | 24 +++++++--- .../test/dag/dag-node-supervision.test.ts | 45 ++++++++++++++++++- 2 files changed, 61 insertions(+), 8 deletions(-) diff --git a/packages/opencode/src/dag/runtime/supervision-sweep.ts b/packages/opencode/src/dag/runtime/supervision-sweep.ts index bd49d85649..7a3070d009 100644 --- a/packages/opencode/src/dag/runtime/supervision-sweep.ts +++ b/packages/opencode/src/dag/runtime/supervision-sweep.ts @@ -10,6 +10,7 @@ import { DagStore } from "@opencode-ai/core/dag/store" import { WorkflowNodeTable } from "@opencode-ai/core/dag/sql" import { and, eq, sql } from "drizzle-orm" import { Dag } from "@/dag/dag" +import { parseWorkflowConfig } from "@/dag/dag" import { SessionPrompt } from "@/session/prompt" /** @@ -75,6 +76,19 @@ const SWEEP_INTERVAL_MS = 60_000 export const frozenTicksNeeded = (escalateIntervalMs: number) => Math.ceil(Math.max(escalateIntervalMs, 1_000) / SWEEP_INTERVAL_MS) + 1 +/** + * The node's escalation cadence derived from a persisted config row. Pure — + * exported for unit tests. parseWorkflowConfig is the repo's defensive + * parser: malformed JSON or shape-divergent rows return undefined instead of + * throwing, and every degrade path lands on the DEFAULT cadence (the widest + * guaranteed-safe window) — a single corrupt row must never defect the sweep. + */ +export const escalateIntervalFromConfig = (raw: string | undefined, nodeId: string) => { + const node = raw === undefined ? undefined : parseWorkflowConfig(raw)?.nodes.find((n) => n.id === nodeId) + const timeoutMs = node?.worker_config?.timeout_ms + return Math.max(1_000, typeof timeoutMs === "number" ? timeoutMs : Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs) +} + const serviceLayer = Layer.effect( Service, Effect.gen(function* () { @@ -92,16 +106,14 @@ const serviceLayer = Layer.effect( // The node's escalation cadence, from the workflow's persisted config — // the same source spawn.ts derived the watcher's escalateIntervalMs from. - // Cached per workflow id for the tick; a config read failure degrades to - // the DEFAULT cadence (the widest guaranteed-safe window). + // Only nodes already flat for a tick pay this lookup; a store read + // failure degrades to the DEFAULT cadence (the widest guaranteed-safe + // window). const escalateIntervalFor = Effect.fnUntraced(function* (workflowId: string, nodeId: string) { const wf = yield* store.getWorkflow(workflowId).pipe( Effect.catchCause(() => Effect.succeed(undefined)), ) - if (!wf) return Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs - const node = JSON.parse(wf.config).nodes?.find?.((n: { id: string }) => n.id === nodeId) - const timeoutMs = node?.worker_config?.timeout_ms - return Math.max(1_000, typeof timeoutMs === "number" ? timeoutMs : Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs) + return escalateIntervalFromConfig(wf?.config, nodeId) }) const sweepOnce = Effect.fn("DagSupervisionSweep.sweepOnce")(function* () { diff --git a/packages/opencode/test/dag/dag-node-supervision.test.ts b/packages/opencode/test/dag/dag-node-supervision.test.ts index 94c791f5c7..aae2947cea 100644 --- a/packages/opencode/test/dag/dag-node-supervision.test.ts +++ b/packages/opencode/test/dag/dag-node-supervision.test.ts @@ -239,8 +239,9 @@ describe("DAG node supervision — deadline enforcement (production incident)", "watcher never escalated a node past its deadline (healthy control)", "8 seconds", ) - // Cap enforcement: max extensions default is 3 — after enough - // escalations the watcher cancels the child and fails the node. + // Cap enforcement: this graph pins max_timeout_extensions to 1 — + // after one escalation the watcher cancels the child and fails the + // node. yield* pollWithTimeout( Effect.gen(function* () { const row = yield* store.getNode(dagID, "worker") @@ -378,3 +379,43 @@ describe("DAG node supervision — deadline enforcement (production incident)", ) }) }) + +describe("DagSupervisionSweep cadence derivation (pure)", () => { + it("derives the cadence from the node's persisted timeout_ms", () => { + const config = JSON.stringify({ nodes: [{ id: "worker", depends_on: [], worker_config: { timeout_ms: 30_000 } }] }) + expect(DagSupervisionSweep.escalateIntervalFromConfig(config, "worker")).toBe(30_000) + }) + + it("floors sub-second timeouts to the watcher's 1s minimum", () => { + const config = JSON.stringify({ nodes: [{ id: "worker", depends_on: [], worker_config: { timeout_ms: 10 } }] }) + expect(DagSupervisionSweep.escalateIntervalFromConfig(config, "worker")).toBe(1_000) + }) + + it("degrades to the DEFAULT cadence on absent row, malformed JSON, shape-divergent rows, or missing node", () => { + const expected = Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs + // No workflow row at all. + expect(DagSupervisionSweep.escalateIntervalFromConfig(undefined, "worker")).toBe(expected) + // Corrupt JSON string, and JSON whose root is not a record. + expect(DagSupervisionSweep.escalateIntervalFromConfig("{not json", "worker")).toBe(expected) + expect(DagSupervisionSweep.escalateIntervalFromConfig("null", "worker")).toBe(expected) + // Shape-divergent rows: nodes not an array / null entries. + expect(DagSupervisionSweep.escalateIntervalFromConfig(JSON.stringify({ nodes: null }), "worker")).toBe(expected) + expect(DagSupervisionSweep.escalateIntervalFromConfig(JSON.stringify({ nodes: [null] }), "worker")).toBe(expected) + // Node absent from the config, or present without worker_config.timeout_ms. + const other = JSON.stringify({ nodes: [{ id: "other", depends_on: [], worker_config: { timeout_ms: 30_000 } }] }) + expect(DagSupervisionSweep.escalateIntervalFromConfig(other, "worker")).toBe(expected) + const bare = JSON.stringify({ nodes: [{ id: "worker", depends_on: [] }] }) + expect(DagSupervisionSweep.escalateIntervalFromConfig(bare, "worker")).toBe(expected) + }) + + it("freeze window boundaries: one interval of flat plus one tick, at every configured timeout", () => { + expect(DagSupervisionSweep.frozenTicksNeeded(1)).toBe(2) + expect(DagSupervisionSweep.frozenTicksNeeded(60_000)).toBe(2) + expect(DagSupervisionSweep.frozenTicksNeeded(60_001)).toBe(3) + // The default 10-minute cadence and the doc-recommended 30-minute + // verifier timeout — a live watcher at each cadence always moves the + // counter inside the window. + expect(DagSupervisionSweep.frozenTicksNeeded(Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs)).toBe(11) + expect(DagSupervisionSweep.frozenTicksNeeded(1_800_000)).toBe(31) + }) +}) From f185065ccda5050b8f2cb27dd5e155d37544a368 Mon Sep 17 00:00:00 2001 From: Lex Date: Wed, 19 Aug 2026 00:49:42 +0800 Subject: [PATCH 5/7] fix(dag): cause-recover the sweep cancel seam, gate settle success (review R4) --- .../src/dag/runtime/supervision-sweep.ts | 26 ++++++-- .../test/dag/dag-node-supervision.test.ts | 64 +++++++++++++++++-- 2 files changed, 78 insertions(+), 12 deletions(-) diff --git a/packages/opencode/src/dag/runtime/supervision-sweep.ts b/packages/opencode/src/dag/runtime/supervision-sweep.ts index 7a3070d009..26b7d722d3 100644 --- a/packages/opencode/src/dag/runtime/supervision-sweep.ts +++ b/packages/opencode/src/dag/runtime/supervision-sweep.ts @@ -158,10 +158,17 @@ const serviceLayer = Layer.effect( // settle the node. The nodeFailed guard under the workflow lock // serializes any race with a live watcher or another host's sweep. if (row.childSessionId) { + // Best-effort cancel, recovered at CAUSE level: the sweep's layer + // context has no ambient InstanceRef, so a real SessionPrompt.cancel + // dies at InstanceState.context ("InstanceRef not provided") — and + // cancel's channel is E=never, where Effect.ignore recovers nothing. + // In the incident shape (instance disposed) the child fiber died + // with the scope, so a skipped cancel is also the correct outcome; + // the durable settle below is the source of truth either way. // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- the durable column is typed string|null; the cancel seam brands SessionID. - yield* promptSvc.cancel(row.childSessionId as never).pipe(Effect.ignore) + yield* promptSvc.cancel(row.childSessionId as never).pipe(Effect.catchCause(() => Effect.void)) } - yield* dag + const settled = yield* dag .nodeFailed( row.workflowId, row.nodeId, @@ -169,14 +176,21 @@ const serviceLayer = Layer.effect( "timeout", ) .pipe( + Effect.as(true), Effect.catchCause((cause) => - Effect.logWarning("DagSupervisionSweep nodeFailed failed", { - dagID: row.workflowId, - nodeID: row.nodeId, - cause, + Effect.gen(function* () { + yield* Effect.logWarning("DagSupervisionSweep nodeFailed failed — retrying next tick", { + dagID: row.workflowId, + nodeID: row.nodeId, + cause, + }) + return false }), ), ) + // On a failed settle keep the streak so the next tick retries + // immediately instead of deferring by a full freeze window. + if (!settled) continue yield* Effect.logWarning("DagSupervisionSweep settled a node with dead deadline supervision", { dagID: row.workflowId, nodeID: row.nodeId, diff --git a/packages/opencode/test/dag/dag-node-supervision.test.ts b/packages/opencode/test/dag/dag-node-supervision.test.ts index aae2947cea..a12681bd0a 100644 --- a/packages/opencode/test/dag/dag-node-supervision.test.ts +++ b/packages/opencode/test/dag/dag-node-supervision.test.ts @@ -82,6 +82,7 @@ function reply(sessionID: string, text: string): SessionV1.WithParts { function supervisionLayer(input: { readonly childPrompts: Queue.Queue readonly cancels: string[] + readonly cancelDefect?: boolean }) { const database = Database.layerFromPath(":memory:") const events = EventV2.layer.pipe(Layer.provide(database)) @@ -116,10 +117,15 @@ function supervisionLayer(input: { const prompt = Layer.mock( SessionPrompt.Service, withIdleAdmission({ - cancel: (sessionID: string) => - Effect.sync(() => { - input.cancels.push(sessionID) - }), + // cancelDefect simulates the production sweep context: the layer-scoped + // fiber has no ambient InstanceRef, so a REAL SessionPrompt.cancel dies + // at InstanceState.context with exactly this defect. + cancel: input.cancelDefect + ? () => Effect.die(new Error("InstanceRef not provided")) + : (sessionID: string) => + Effect.sync(() => { + input.cancels.push(sessionID) + }), prompt: (value: SessionPrompt.PromptInput) => deliver(value), promptIfIdle: (value: SessionPrompt.PromptInput) => deliver(value).pipe(Effect.map(Option.some)), }), @@ -152,7 +158,10 @@ interface SupervisionServices { readonly cancels: string[] readonly database: Database.Interface } -function runSupervisionTest(options: { readonly instanceProject: string }, test: (services: SupervisionServices) => Effect.Effect) { +function runSupervisionTest( + options: { readonly instanceProject: string; readonly cancelDefect?: boolean }, + test: (services: SupervisionServices) => Effect.Effect, +) { return Effect.gen(function* () { const childPrompts = yield* Queue.unbounded() const cancels: string[] = [] @@ -184,7 +193,7 @@ function runSupervisionTest(options: { readonly instanceProject: string }, te yield* loop.init() return yield* test({ dag, loop, store, sweep, childPrompts, cancels, database }) }).pipe( - Effect.provide(supervisionLayer({ childPrompts, cancels })), + Effect.provide(supervisionLayer({ childPrompts, cancels, cancelDefect: options.cancelDefect })), Effect.provideService(InstanceRef, { directory: process.cwd(), worktree: process.cwd(), @@ -331,6 +340,49 @@ describe("DAG node supervision — deadline enforcement (production incident)", ) }) + // Review R4 issue 1 (P0): the sweep's layer-scoped fiber has no ambient + // InstanceRef, so a real SessionPrompt.cancel DIES at + // InstanceState.context — and cancel's channel is E=never, where + // Effect.ignore recovers nothing. A cause-level recovery on that seam is + // what keeps the durable settle reachable in production. Simulated here by + // mocking cancel to the exact production defect. + it("cancel-defect: a dying cancel seam (production: no ambient InstanceRef) never blocks the settle", async () => { + await Effect.runPromise( + runSupervisionTest({ instanceProject: "project-1", cancelDefect: true }, ({ dag, store, sweep, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create(incidentGraph) + const child = yield* takeWithin(childPrompts, "worker did not start") + void child + yield* pollWithTimeout( + supervisionProgress(store, dagID, "worker"), + "watcher never escalated before dispose", + "8 seconds", + ) + yield* Effect.promise(() => disposeInstance(process.cwd())) + yield* Effect.sleep("4 seconds") + // Deadline passed, watcher dead: every settle-attempt pass hits the + // dying cancel seam first. Pre-fix, the defect aborts sweepOnce + // before nodeFailed; post-fix the settle still lands. + const needed = DagSupervisionSweep.frozenTicksNeeded(2_000) + for (let tick = 0; tick <= needed; tick++) { + yield* sweep.sweepOnce() + yield* Effect.sleep("50 millis") + } + const swept = yield* pollWithTimeout( + Effect.gen(function* () { + const settled = yield* store.getNode(dagID, "worker") + return settled && settled.status !== "running" ? settled : undefined + }), + "sweep never settled past a dying cancel seam", + "5 seconds", + ) + expect(swept?.errorClass).toBe("timeout") + expect(swept?.errorReason).toContain("swept") + }), + ), + ) + }) + // Review R1 issue 2 (false-positive kill): a LIVE watcher on a node whose // escalation cadence spans multiple sweep intervals must never be swept — // the counter legitimately stays flat between escalations. The graph keeps From 235d031964fd4ef38d5da1f2ed01968a1cbae773 Mon Sep 17 00:00:00 2001 From: Lex Date: Wed, 19 Aug 2026 00:58:54 +0800 Subject: [PATCH 6/7] fix(dag): propagate interrupts in sweep recovery sites, correct cross-host comment (review R5) --- .../src/dag/runtime/supervision-sweep.ts | 57 ++++++++++++------- 1 file changed, 36 insertions(+), 21 deletions(-) diff --git a/packages/opencode/src/dag/runtime/supervision-sweep.ts b/packages/opencode/src/dag/runtime/supervision-sweep.ts index 26b7d722d3..f43170b34b 100644 --- a/packages/opencode/src/dag/runtime/supervision-sweep.ts +++ b/packages/opencode/src/dag/runtime/supervision-sweep.ts @@ -3,7 +3,7 @@ export * as DagSupervisionSweep from "./supervision-sweep" -import { Context, Effect, Fiber, Layer, Scope } from "effect" +import { Cause, Context, Effect, Fiber, Layer, Scope } from "effect" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Database } from "@opencode-ai/core/database/database" import { DagStore } from "@opencode-ai/core/dag/store" @@ -33,9 +33,11 @@ import { SessionPrompt } from "@/session/prompt" * never depend on ambient per-instance context (InstanceRef) — its fiber's * context is the layer-build context, which has none. Ownership is decided * from the durable rows alone: this process's Database owns every workflow - * row it can read, and the nodeFailed guard under the workflow lock - * serializes any race with another writer (including a second host sharing - * the DB — double settles collapse to one). + * row it can read. Same-host races (a live watcher, the DagLoop) are + * serialized by the workflow's in-process lock; a second host sharing the + * DB is handled by the durable guardNode status read plus the projector's + * conditional UPDATE (only the first NodeFailed folds a non-terminal row) — + * double settles collapse to one. * * False-positive safety: a LIVE watcher escalates on * escalateIntervalMs == max(1s, timeout_ms ?? 10min) and nodeTimeoutEscalated @@ -111,7 +113,7 @@ const serviceLayer = Layer.effect( // window). const escalateIntervalFor = Effect.fnUntraced(function* (workflowId: string, nodeId: string) { const wf = yield* store.getWorkflow(workflowId).pipe( - Effect.catchCause(() => Effect.succeed(undefined)), + Effect.catchCause((cause) => (Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.succeed(undefined))), ) return escalateIntervalFromConfig(wf?.config, nodeId) }) @@ -136,11 +138,15 @@ const serviceLayer = Layer.effect( // A store defect must not kill the sweep — the same silent-death // class this service exists to eliminate. Degrade to an empty // pass and retry next tick (mirror of spawn.ts's R13 hardening). + // Interrupts (scope disposal) still propagate — the repo's + // background-loop discipline. Effect.catchCause((cause) => - Effect.gen(function* () { - yield* Effect.logWarning("DagSupervisionSweep store query failed — skipping tick", { cause }) - return [] - }), + Cause.hasInterrupts(cause) + ? Effect.interrupt + : Effect.gen(function* () { + yield* Effect.logWarning("DagSupervisionSweep store query failed — skipping tick", { cause }) + return [] + }), ), ) @@ -155,8 +161,10 @@ const serviceLayer = Layer.effect( const escalateIntervalMs = yield* escalateIntervalFor(row.workflowId, row.nodeId) if (flatTicks < frozenTicksNeeded(escalateIntervalMs)) continue // Frozen across the full window: cancel the (possibly dead) child and - // settle the node. The nodeFailed guard under the workflow lock - // serializes any race with a live watcher or another host's sweep. + // settle the node. Same-host races (a live watcher) are serialized by + // the workflow's in-process lock; another host's sweep is collapsed + // by the durable terminal-status guard — either way at most one + // settle lands. if (row.childSessionId) { // Best-effort cancel, recovered at CAUSE level: the sweep's layer // context has no ambient InstanceRef, so a real SessionPrompt.cancel @@ -166,7 +174,9 @@ const serviceLayer = Layer.effect( // with the scope, so a skipped cancel is also the correct outcome; // the durable settle below is the source of truth either way. // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- the durable column is typed string|null; the cancel seam brands SessionID. - yield* promptSvc.cancel(row.childSessionId as never).pipe(Effect.catchCause(() => Effect.void)) + yield* promptSvc.cancel(row.childSessionId as never).pipe( + Effect.catchCause((cause) => (Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.void)), + ) } const settled = yield* dag .nodeFailed( @@ -178,14 +188,16 @@ const serviceLayer = Layer.effect( .pipe( Effect.as(true), Effect.catchCause((cause) => - Effect.gen(function* () { - yield* Effect.logWarning("DagSupervisionSweep nodeFailed failed — retrying next tick", { - dagID: row.workflowId, - nodeID: row.nodeId, - cause, - }) - return false - }), + Cause.hasInterrupts(cause) + ? Effect.interrupt + : Effect.gen(function* () { + yield* Effect.logWarning("DagSupervisionSweep nodeFailed failed — retrying next tick", { + dagID: row.workflowId, + nodeID: row.nodeId, + cause, + }) + return false + }), ), ) // On a failed settle keep the streak so the next tick retries @@ -214,8 +226,11 @@ const serviceLayer = Layer.effect( yield* sweepOnce().pipe( // Per-tick guard: any residual defect inside a tick degrades to // a logged skip — the loop itself must outlive every failure. + // Interrupts (scope disposal) still exit the loop. Effect.catchCause((cause) => - Effect.logWarning("DagSupervisionSweep tick failed — retrying next interval", { cause }), + Cause.hasInterrupts(cause) + ? Effect.interrupt + : Effect.logWarning("DagSupervisionSweep tick failed — retrying next interval", { cause }), ), ) } From a957cb7678be4bd1be1afc2c994a12b7c7a0eb67 Mon Sep 17 00:00:00 2001 From: Lex Date: Wed, 19 Aug 2026 01:03:19 +0800 Subject: [PATCH 7/7] chore(dag): merge duplicate dag import --- packages/opencode/src/dag/runtime/supervision-sweep.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/opencode/src/dag/runtime/supervision-sweep.ts b/packages/opencode/src/dag/runtime/supervision-sweep.ts index f43170b34b..fbfbb0ad60 100644 --- a/packages/opencode/src/dag/runtime/supervision-sweep.ts +++ b/packages/opencode/src/dag/runtime/supervision-sweep.ts @@ -9,8 +9,7 @@ import { Database } from "@opencode-ai/core/database/database" import { DagStore } from "@opencode-ai/core/dag/store" import { WorkflowNodeTable } from "@opencode-ai/core/dag/sql" import { and, eq, sql } from "drizzle-orm" -import { Dag } from "@/dag/dag" -import { parseWorkflowConfig } from "@/dag/dag" +import { Dag, parseWorkflowConfig } from "@/dag/dag" import { SessionPrompt } from "@/session/prompt" /**