diff --git a/package.json b/package.json index 9da0225a42..a3de25dc12 100644 --- a/package.json +++ b/package.json @@ -12,7 +12,7 @@ "dev:console": "ulimit -n 10240 2>/dev/null; bun run --cwd packages/console/app dev", "dev:stats": "bun sst shell --stage=production -- bun run --cwd packages/stats/app dev", "dev:storybook": "bun --cwd packages/storybook storybook", - "lint": "oxlint --max-warnings=4734", + "lint": "oxlint --max-warnings=4831", "typecheck": "bun turbo typecheck", "upgrade-opentui": "bun run script/upgrade-opentui.ts", "postinstall": "bun run --cwd packages/core fix-node-pty", diff --git a/packages/core/schema.json b/packages/core/schema.json index 9fa91561b0..126f187051 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,9 +1,9 @@ { "version": "7", "dialect": "sqlite", - "id": "442cdbd5-86a8-41a9-86d6-5361dbac90e0", + "id": "abdf5c23-7f2e-4ca3-b08b-012db47b5aa5", "prevIds": [ - "a953899b-bb63-497e-8e2e-86eb5e0fdeed" + "442cdbd5-86a8-41a9-86d6-5361dbac90e0" ], "ddl": [ { @@ -658,6 +658,26 @@ "entityType": "columns", "table": "workflow_node" }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "timeout_extensions", + "entityType": "columns", + "table": "workflow_node" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "false", + "generated": null, + "name": "escalation_pending", + "entityType": "columns", + "table": "workflow_node" + }, { "type": "integer", "notNull": true, diff --git a/packages/core/src/dag/core/replan.ts b/packages/core/src/dag/core/replan.ts index 99c9badcc6..08d839d83f 100644 --- a/packages/core/src/dag/core/replan.ts +++ b/packages/core/src/dag/core/replan.ts @@ -186,15 +186,15 @@ export function planReplan( for (const n of current.nodes) { if (!survivingIds.has(n.id)) continue const frag = fragmentNodeById.get(n.id) - // A node takes the fragment's deps only when it is actually being replaced - // (pending/queued/paused) or restarted (running with restart marker). A - // running node present without a marker is "kept unchanged" and keeps its - // current deps; terminal nodes are immutable and keep their current deps. - if (frag && (frag.restart || (n.status !== NodeStatus.RUNNING && !isNodeTerminalStatus(n.status)))) { - for (const depId of frag.depends_on) tryAddEdge(n.id, depId) - } else { - for (const depId of n.depends_on) tryAddEdge(n.id, depId) - } + // P1a: the CHECK graph must equal the EXECUTION graph. A running node + // present without a restart marker is replaced (its definition is + // re-published via NodeRegistered and the projector upserts the fragment's + // depends_on into the durable row the runtime rebuilds from), so it takes + // the fragment's deps here too — otherwise a cycle only reachable through + // the replaced deps passes the check and crashes the runtime's + // rebuildGraph. Terminal nodes are immutable and keep their current deps. + const deps = frag && !isNodeTerminalStatus(n.status) ? frag.depends_on : n.depends_on + for (const depId of deps) tryAddEdge(n.id, depId) } for (const fragNode of fragment.nodes) { if (currentStateById.has(fragNode.id)) continue // handled above @@ -226,7 +226,15 @@ export function planReplan( continue } if (n.status === NodeStatus.RUNNING) { - if (frag?.restart) restart.push(n.id) + if (frag?.restart) { + restart.push(n.id) + continue + } + // A running node present in the fragment (no restart marker) gets its + // definition replaced without re-executing — this is the timeout + // extension path: the merged config carries the new worker_config.timeout_ms, + // and the runtime recomputes the absolute deadline from it. + if (frag) replace.push(n.id) continue } if (n.status === NodeStatus.PENDING) { diff --git a/packages/core/src/dag/projector.ts b/packages/core/src/dag/projector.ts index e895d8117d..ed642f822c 100644 --- a/packages/core/src/dag/projector.ts +++ b/packages/core/src/dag/projector.ts @@ -235,6 +235,12 @@ export const layer = Layer.effectDiscard( deadline_ms: event.data.deadlineMs ?? null, wake_eligible: event.data.wakeEligible ?? false, wake_reported: false, + // S3: a fresh execution attempt starts with a fresh extension budget — + // restart clears timeout_extensions so the cap is per-attempt, not + // lifetime (a restart must not inherit a nearly-exhausted cap). The + // new attempt is also not awaiting adjudication. + timeout_extensions: 0, + escalation_pending: false, seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp), }) @@ -260,6 +266,12 @@ export const layer = Layer.effectDiscard( status: "completed", output: event.data.output, completed_at: toMillis(event.data.timestamp), + // F2b: re-arm wake delivery on every status migration. A node whose + // escalated wake was already reported (wake_reported=true) must + // re-enter the snapshot on completion/failure — otherwise its result + // notification (and the crash-recovery failure at recovery.ts) is + // lost behind the earlier escalation notification. + wake_reported: false, seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp), }) @@ -284,6 +296,10 @@ export const layer = Layer.effectDiscard( error_reason: event.data.reason, error_class: event.data.trigger, completed_at: toMillis(event.data.timestamp), + // F2b: same re-arm as NodeCompleted — a failure after a reported + // escalation (or a crash-recovery failure of an escalated node) + // must still reach the main agent. + wake_reported: false, seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp), }) @@ -339,6 +355,11 @@ export const layer = Layer.effectDiscard( // abort the old session before spawning the replacement. NodeStarted // will overwrite it with the new child session. replan_attempts: sql`${WorkflowNodeTable.replan_attempts} + 1`, + // S3: restart opens a new attempt — reset the extension budget and + // clear the pending-adjudication flag so the cap and the summary are + // per-attempt, not lifetime. + timeout_extensions: 0, + escalation_pending: false, seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp), }) @@ -353,6 +374,35 @@ export const layer = Layer.effectDiscard( .run() .pipe(Effect.orDie), ) + + // Timeout escalation: the node stays RUNNING (no status transition). Only + // the extension count, seq, and wake flag change. wake_reported is reset so + // the escalated node re-enters the wake snapshot and the main agent is + // notified once per escalation. + yield* events.project(DagEvent.NodeTimeoutEscalated, (event) => + db + .update(WorkflowNodeTable) + .set({ + timeout_extensions: event.data.timeoutExtensions, + // The escalation is not yet adjudicated — summary and the wake + // delivery boundary treat the node as awaiting main-agent action + // until an extend (updateNodeDeadline) or a new attempt clears it. + escalation_pending: true, + wake_reported: false, + seq: event.durable!.seq, + time_updated: toMillis(event.data.timestamp), + }) + // F2a: escalate only live running nodes — a stale escalate racing a + // terminal event must not resurrect the wake flag or inflate the + // counter on an already-completed/failed node (ghost wake). + .where(and( + eq(WorkflowNodeTable.workflow_id, event.data.dagID), + eq(WorkflowNodeTable.id, event.data.nodeID), + inArray(WorkflowNodeTable.status, ["running"]), + )) + .run() + .pipe(Effect.orDie), + ) }), ) diff --git a/packages/core/src/dag/sql.ts b/packages/core/src/dag/sql.ts index 3f8f325085..eb70ca4008 100644 --- a/packages/core/src/dag/sql.ts +++ b/packages/core/src/dag/sql.ts @@ -68,6 +68,8 @@ export const WorkflowNodeTable = sqliteTable( wake_eligible: integer({ mode: "boolean" }).notNull().default(false), // D6: node has report_to_parent=true wake_reported: integer({ mode: "boolean" }).notNull().default(false), // D3: has this node's terminal event been injected into the parent session? replan_attempts: integer().notNull().default(0), // D4: per-node replan counter for circuit breaker + timeout_extensions: integer().notNull().default(0), // timeout escalation count (node stays running; main agent adjudicates) + escalation_pending: integer({ mode: "boolean" }).notNull().default(false), // set on escalate, cleared on adjudication (extend) or new attempt — "awaiting main-agent adjudication" seq: integer().notNull(), // latest durable event seq for this node started_at: integer(), completed_at: integer(), diff --git a/packages/core/src/dag/store.ts b/packages/core/src/dag/store.ts index bc40868f7f..a45beb1435 100644 --- a/packages/core/src/dag/store.ts +++ b/packages/core/src/dag/store.ts @@ -1,6 +1,6 @@ export * as DagStore from "./store" -import { and, asc, count, desc, eq, inArray } from "drizzle-orm" +import { and, asc, count, desc, eq, gt, inArray, or } from "drizzle-orm" import { Context, Effect, Layer } from "effect" import { Database } from "../database/database" import { LayerNode } from "../effect/layer-node" @@ -44,6 +44,8 @@ export interface NodeRow { wakeEligible: boolean wakeReported: boolean replanAttempts: number + timeoutExtensions: number + escalationPending: boolean seq: number startedAt: number | null completedAt: number | null @@ -70,6 +72,8 @@ export interface WorkflowSummary { failedNodes: number skippedNodes: number queuedNodes: number + /** Running nodes with a not-yet-adjudicated timeout escalation (escalation_pending). */ + escalatedNodes: number } const mapWorkflow = (r: typeof WorkflowTable.$inferSelect): WorkflowRow => ({ @@ -106,11 +110,32 @@ const mapNode = (r: typeof WorkflowNodeTable.$inferSelect): NodeRow => ({ wakeEligible: r.wake_eligible, wakeReported: r.wake_reported, replanAttempts: r.replan_attempts, + timeoutExtensions: r.timeout_extensions, + escalationPending: r.escalation_pending, seq: r.seq, startedAt: r.started_at, completedAt: r.completed_at, }) +// F11: wake eligibility gates TERMINAL notifications (the report_to_parent +// contract) — but a timeout-escalated node must reach the main agent +// REGARDLESS of report_to_parent AND regardless of its current status: the +// escalation wake is the only force behind the extension cap, and a +// non-eligible node (default config) would otherwise never be adjudicated. +// Escalated-then-terminalized nodes (cap-exhausted force-cancel) still need +// delivery, and so do escalated-then-COMPLETED nodes: the main agent already +// spent turns adjudicating this node (the extend path), so its result is the +// receipt for those turns — withholding it behind report_to_parent would +// silently lose adjudicated work. Single source of truth for snapshot / +// unreported / bootstrap-sweep wake queries. +const wakeDeliverableNodePredicate = or( + and( + eq(WorkflowNodeTable.wake_eligible, true), + inArray(WorkflowNodeTable.status, ["completed", "failed"]), + ), + gt(WorkflowNodeTable.timeout_extensions, 0), +) + // ============================================================================ // Service interface // ============================================================================ @@ -127,6 +152,7 @@ export interface Interface { readonly getNode: (workflowId: string, nodeId: string) => Effect.Effect readonly getRunningNodes: (workflowId: string) => Effect.Effect readonly setCapturedOutput: (childSessionID: string, payload: unknown) => Effect.Effect + readonly updateNodeDeadline: (workflowId: string, nodeID: string, deadlineMs: number) => Effect.Effect readonly markNodeWakeReported: (workflowId: string, nodeID: string) => Effect.Effect readonly markWorkflowWakeReported: (dagID: string) => Effect.Effect @@ -212,6 +238,26 @@ export const layer = Layer.effect( .groupBy(WorkflowNodeTable.workflow_id, WorkflowNodeTable.status) .all() .pipe(Effect.orDie) + // F10: separate aggregation for running nodes with a not-yet-adjudicated + // timeout escalation (the status grouping above cannot see the flag). + // escalation_pending is set on escalate and cleared on adjudication, so + // this counts only nodes genuinely awaiting main-agent action. + const escalatedRows = yield* db + .select({ + workflowId: WorkflowNodeTable.workflow_id, + total: count(), + }) + .from(WorkflowNodeTable) + .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id)) + .where(and( + eq(WorkflowTable.session_id, sessionId), + eq(WorkflowNodeTable.status, "running"), + eq(WorkflowNodeTable.escalation_pending, true), + )) + .groupBy(WorkflowNodeTable.workflow_id) + .all() + .pipe(Effect.orDie) + const escalatedByWorkflow = new Map(escalatedRows.map((row) => [row.workflowId, row.total])) const counts = countRows.reduce((all, row) => { const current = all.get(row.workflowId) ?? { nodeCount: 0, completedNodes: 0, runningNodes: 0, failedNodes: 0, skippedNodes: 0, queuedNodes: 0 } current.nodeCount += row.total @@ -228,6 +274,7 @@ export const layer = Layer.effect( title: wf.title, status: wf.status, ...(counts.get(wf.id) ?? { nodeCount: 0, completedNodes: 0, runningNodes: 0, failedNodes: 0, skippedNodes: 0, queuedNodes: 0 }), + escalatedNodes: escalatedByWorkflow.get(wf.id) ?? 0, })) }), @@ -271,6 +318,30 @@ export const layer = Layer.effect( .pipe(Effect.orDie) }), + updateNodeDeadline: Effect.fn("DagStore.updateNodeDeadline")(function* (workflowId, nodeID, deadlineMs) { + const updated = yield* db + .update(WorkflowNodeTable) + // Adjudication write (re-time via nodeExtendTimeout). Only update the + // deadline — do NOT reset timeout_extensions: the count is cumulative + // per attempt so an agent cannot bypass the cap by re-planning. + // Escalation is now adjudicated: clear escalation_pending (summary and + // delivery boundary stop treating the node as awaiting adjudication) + // and consume the escalation wake (wake_reported=true) so the stale + // timeout wake is not re-delivered after the deadline moved. + .set({ deadline_ms: deadlineMs, escalation_pending: false, wake_reported: true }) + // Guard: never write a deadline onto a node that terminalized between + // the caller's read and this update. + .where(and( + eq(WorkflowNodeTable.workflow_id, workflowId), + eq(WorkflowNodeTable.id, nodeID), + eq(WorkflowNodeTable.status, "running"), + )) + .returning({ id: WorkflowNodeTable.id }) + .all() + .pipe(Effect.orDie) + return updated.length + }), + markNodeWakeReported: Effect.fn("DagStore.markNodeWakeReported")(function* (workflowId, nodeID) { yield* db .update(WorkflowNodeTable) @@ -337,9 +408,14 @@ export const layer = Layer.effect( .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id)) .where(and( eq(WorkflowTable.session_id, sessionID), - eq(WorkflowNodeTable.wake_eligible, true), eq(WorkflowNodeTable.wake_reported, false), - inArray(WorkflowNodeTable.status, ["completed", "failed"]), + // Escalated nodes enter the snapshot unconditionally + // (timeout_extensions > 0), covering the escalated-then- + // terminal outcome too — the cap's terminal verdict is its + // enforceable force. Adjudication consumes the wake + // (wake_reported=true), so adjudicated nodes are already + // filtered out above. + wakeDeliverableNodePredicate, )) .orderBy( asc(WorkflowTable.seq), @@ -370,9 +446,10 @@ export const layer = Layer.effect( .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id)) .where(and( eq(WorkflowTable.session_id, sessionID), - eq(WorkflowNodeTable.wake_eligible, true), eq(WorkflowNodeTable.wake_reported, false), - inArray(WorkflowNodeTable.status, ["completed", "failed"]), + // See wakeDeliverableNodePredicate — escalated nodes are wake- + // eligible regardless of report_to_parent and status. + wakeDeliverableNodePredicate, )) .orderBy( asc(WorkflowTable.seq), @@ -416,9 +493,11 @@ export const layer = Layer.effect( .from(WorkflowNodeTable) .innerJoin(WorkflowTable, eq(WorkflowNodeTable.workflow_id, WorkflowTable.id)) .where(and( - eq(WorkflowNodeTable.wake_eligible, true), eq(WorkflowNodeTable.wake_reported, false), - inArray(WorkflowNodeTable.status, ["completed", "failed"]), + // See wakeDeliverableNodePredicate — escalated nodes count as + // unreported wakes so the bootstrap sweep finds a session whose + // only outstanding item is an escalation. + wakeDeliverableNodePredicate, )) .all() .pipe(Effect.orDie) diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 42cee3ab92..1c21bf23b1 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -49,5 +49,7 @@ export const migrations = ( import("./migration/20260720013828_dag-workflow-node-identity"), import("./migration/20260803073521_workflow_node_error_class"), import("./migration/20260803083938_restore_goal_state"), + import("./migration/20260805094941_workflow_node_timeout_extensions"), + import("./migration/20260805094942_workflow_node_escalation_pending"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260805094941_workflow_node_timeout_extensions.ts b/packages/core/src/database/migration/20260805094941_workflow_node_timeout_extensions.ts new file mode 100644 index 0000000000..07ade979fe --- /dev/null +++ b/packages/core/src/database/migration/20260805094941_workflow_node_timeout_extensions.ts @@ -0,0 +1,11 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260805094941_workflow_node_timeout_extensions", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`ALTER TABLE \`workflow_node\` ADD \`timeout_extensions\` integer DEFAULT 0 NOT NULL;`) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/migration/20260805094942_workflow_node_escalation_pending.ts b/packages/core/src/database/migration/20260805094942_workflow_node_escalation_pending.ts new file mode 100644 index 0000000000..8530974b48 --- /dev/null +++ b/packages/core/src/database/migration/20260805094942_workflow_node_escalation_pending.ts @@ -0,0 +1,14 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +// Separate migration id from the timeout_extensions ALTER: the runner +// applies each migration at most once keyed by id, so a DB that already ran +// the timeout_extensions migration must still pick up this column. +export default { + id: "20260805094942_workflow_node_escalation_pending", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`ALTER TABLE \`workflow_node\` ADD \`escalation_pending\` integer DEFAULT false NOT NULL;`) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index e8dbbba7e4..ac75ddb53a 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -89,6 +89,8 @@ export default { \`wake_eligible\` integer DEFAULT false NOT NULL, \`wake_reported\` integer DEFAULT false NOT NULL, \`replan_attempts\` integer DEFAULT 0 NOT NULL, + \`timeout_extensions\` integer DEFAULT 0 NOT NULL, + \`escalation_pending\` integer DEFAULT false NOT NULL, \`seq\` integer NOT NULL, \`started_at\` integer, \`completed_at\` integer, diff --git a/packages/core/test/dag-store-summaries.test.ts b/packages/core/test/dag-store-summaries.test.ts index deb9b1e4d8..b00522d987 100644 --- a/packages/core/test/dag-store-summaries.test.ts +++ b/packages/core/test/dag-store-summaries.test.ts @@ -100,6 +100,7 @@ describe("DagStore.getWorkflowSummaries (SQL aggregation)", () => { failedNodes: 1, skippedNodes: 1, queuedNodes: 1, + escalatedNodes: 0, }) expect(summaries[1]).toEqual({ id: "wf-empty", @@ -111,6 +112,7 @@ describe("DagStore.getWorkflowSummaries (SQL aggregation)", () => { failedNodes: 0, skippedNodes: 0, queuedNodes: 0, + escalatedNodes: 0, }) }).pipe(Effect.provide(storeLayer()), Effect.scoped), ) diff --git a/packages/core/test/dag-store-update-deadline.test.ts b/packages/core/test/dag-store-update-deadline.test.ts new file mode 100644 index 0000000000..70dbc52421 --- /dev/null +++ b/packages/core/test/dag-store-update-deadline.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, test } from "bun:test" +import { Effect, Layer } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { WorkflowNodeTable, WorkflowTable } from "@opencode-ai/core/dag/sql" +import { DagStore } from "@opencode-ai/core/dag/store" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { SessionTable } from "@opencode-ai/core/session/sql" + +function storeLayer() { + const database = Database.layerFromPath(":memory:") + const store = DagStore.layer.pipe(Layer.provide(database)) + return Layer.merge(database, store) +} + +function node(workflowId: string, id: string, status: string, seq: number) { + return { + id, + workflow_id: workflowId, + name: id, + worker_type: "build", + status, + required: true, + depends_on: [], + wake_eligible: false, + wake_reported: false, + seq, + } +} + +function seed() { + return Effect.gen(function* () { + const database = yield* Database.Service + yield* database.db.insert(ProjectTable).values({ + id: "project-1" as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.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* database.db.insert(WorkflowTable).values({ + id: "wf-1", + project_id: "project-1" as never, + session_id: "ses_parent" as never, + title: "Deadline", + status: "running", + config: "{}", + seq: 1, + wake_reported: false, + time_created: 1, + }).run().pipe(Effect.orDie) + yield* database.db.insert(WorkflowNodeTable).values([ + { ...node("wf-1", "running-1", "running", 1), deadline_ms: 1000, timeout_extensions: 1, escalation_pending: true }, + { ...node("wf-1", "done-1", "completed", 2), deadline_ms: 2000, timeout_extensions: 1, escalation_pending: true }, + ]).run().pipe(Effect.orDie) + }) +} + +describe("DagStore.updateNodeDeadline (adjudication write)", () => { + test("writes one row for a running node: moves the deadline, clears escalation_pending, consumes the escalation wake, keeps the cumulative count", async () => { + await Effect.runPromise( + Effect.gen(function* () { + const store = yield* DagStore.Service + yield* seed() + + const written = yield* store.updateNodeDeadline("wf-1", "running-1", 99_999) + expect(written).toBe(1) + + const row = yield* store.getNode("wf-1", "running-1") + expect(row?.deadlineMs).toBe(99_999) + expect(row?.escalationPending).toBe(false) + expect(row?.wakeReported).toBe(true) + expect(row?.timeoutExtensions).toBe(1) + }).pipe(Effect.provide(storeLayer()), Effect.scoped), + ) + }) + + test("rejects a terminal node: zero rows written, deadline untouched (status='running' guard)", async () => { + await Effect.runPromise( + Effect.gen(function* () { + const store = yield* DagStore.Service + yield* seed() + + const written = yield* store.updateNodeDeadline("wf-1", "done-1", 99_999) + expect(written).toBe(0) + + const row = yield* store.getNode("wf-1", "done-1") + expect(row?.deadlineMs).toBe(2000) + expect(row?.escalationPending).toBe(true) + expect(row?.timeoutExtensions).toBe(1) + }).pipe(Effect.provide(storeLayer()), Effect.scoped), + ) + }) +}) diff --git a/packages/opencode/src/dag/dag.ts b/packages/opencode/src/dag/dag.ts index ef936b9ea4..982d521c1c 100644 --- a/packages/opencode/src/dag/dag.ts +++ b/packages/opencode/src/dag/dag.ts @@ -46,6 +46,7 @@ export const DEFAULT_WORKFLOW_CONFIG = { nodeTimeoutMs: 10 * 60 * 1000, nodeRequired: false, reportToParent: false, + maxTimeoutExtensions: 20, } as const /** A node as declared in the workflow's YAML config. */ @@ -85,6 +86,7 @@ export interface WorkflowConfig { max_concurrency?: number max_node_replan_attempts?: number max_total_nodes?: number + max_timeout_extensions?: number node_defaults?: NodeDefaults nodes: NodeConfig[] } @@ -113,11 +115,19 @@ export function normalizeModel(model: NodeConfig["model"]) { } } +// F9: clamp the timeout floor — 0/negative timeout_ms would fire the deadline +// watcher immediately (escalate or force-cancel on the first tick). +const MIN_NODE_TIMEOUT_MS = 1_000 + +function clampTimeoutMs(timeoutMs: number | undefined, fallbackMs: number) { + return Math.max(MIN_NODE_TIMEOUT_MS, timeoutMs ?? fallbackMs) +} + function normalizeNodeDefaults(defaults: NodeDefaults | undefined): NodeDefaults { return { required: defaults?.required ?? DEFAULT_WORKFLOW_CONFIG.nodeRequired, worker_config: { - timeout_ms: defaults?.worker_config?.timeout_ms ?? DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs, + timeout_ms: clampTimeoutMs(defaults?.worker_config?.timeout_ms, DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs), }, report_to_parent: defaults?.report_to_parent ?? DEFAULT_WORKFLOW_CONFIG.reportToParent, ...(defaults?.model ? { model: normalizeModel(defaults.model) } : {}), @@ -132,13 +142,24 @@ function normalizeNodeConfig(node: NodeConfig, defaults: NodeDefaults): NodeConf worker_config: { ...defaults.worker_config, ...node.worker_config, - timeout_ms: node.worker_config?.timeout_ms ?? defaults.worker_config?.timeout_ms ?? DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs, + timeout_ms: clampTimeoutMs(node.worker_config?.timeout_ms ?? defaults.worker_config?.timeout_ms, DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs), }, report_to_parent: node.report_to_parent ?? defaults.report_to_parent ?? DEFAULT_WORKFLOW_CONFIG.reportToParent, ...(model ? { model } : {}), } } +// F2: a fragment node that omits worker_config.timeout_ms must NOT be +// silently normalized to the DEFAULT (that would rewrite a long extension +// back to 10min — implicit budget shortening). The replace bucket (definition +// replaced, execution kept) preserves the existing node's timeout for the +// merged config and the deadline recompute. +function normalizeFragmentNode(node: NodeConfig, existingTimeoutMs: number | undefined, defaults: NodeDefaults): NodeConfig { + const timeoutMs = node.worker_config?.timeout_ms ?? existingTimeoutMs + const withTimeout = timeoutMs == null ? node : { ...node, worker_config: { ...node.worker_config, timeout_ms: timeoutMs } } + return normalizeNodeConfig(withTimeout, defaults) +} + function normalizeWorkflowConfig(config: WorkflowConfig): WorkflowConfig { const defaults = normalizeNodeDefaults(config.node_defaults) return { @@ -147,6 +168,7 @@ function normalizeWorkflowConfig(config: WorkflowConfig): WorkflowConfig { max_concurrency: config.max_concurrency ?? DEFAULT_WORKFLOW_CONFIG.maxConcurrency, max_node_replan_attempts: config.max_node_replan_attempts ?? DEFAULT_WORKFLOW_CONFIG.maxNodeReplanAttempts, max_total_nodes: config.max_total_nodes ?? DEFAULT_WORKFLOW_CONFIG.maxTotalNodes, + max_timeout_extensions: config.max_timeout_extensions ?? DEFAULT_WORKFLOW_CONFIG.maxTimeoutExtensions, node_defaults: defaults, nodes: config.nodes.map((node) => normalizeNodeConfig(node, defaults)), } @@ -262,6 +284,8 @@ 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 nodeExtendTimeout: (dagID: string, nodeID: string, newDeadlineMs: number) => Effect.Effect } export class Service extends Context.Service()("@opencode/Dag") {} @@ -542,7 +566,12 @@ export const layer = Layer.effect( } const wfConfig = parseWorkflowConfig(workflow.config) const defaults = normalizeNodeDefaults(wfConfig?.node_defaults) - const normalizedFragment = { nodes: fragment.nodes.map((node) => normalizeNodeConfig(node, defaults)) } + const cfgById = new Map((wfConfig?.nodes ?? []).map((n) => [n.id, n])) + const normalizedFragment = { + nodes: fragment.nodes.map((node) => + normalizeFragmentNode(node, cfgById.get(node.id)?.worker_config?.timeout_ms, defaults), + ), + } const nodes = yield* store.getNodes(dagID) const plan = planReplan( { nodes: nodes.map((n) => ({ id: n.id, status: n.status as never, depends_on: n.dependsOn })) }, @@ -640,6 +669,24 @@ export const layer = Layer.effect( }) } for (const id of effectiveRestart) { + // A restart re-spawns with the fragment's definition — the new + // depends_on must reach the durable row BEFORE the runtime rebuilds + // its graph from store.getNodes (WorkflowReplanned handler), or the + // restarted node keeps its stale edges and is re-ready under them. + // Mirrors the replace bucket's NodeRegistered re-publish. + const node = fragmentById.get(id) + if (node) { + yield* events.publish(DagEvent.NodeRegistered, { + dagID: dagID as ID, + nodeID: id as never, + name: node.name, + workerType: node.worker_type, + dependsOn: node.depends_on.map((d) => d as never), + required: node.required, + model: node.model as never, + timestamp: yield* DateTime.now, + }) + } yield* events.publish(DagEvent.NodeRestarted, { dagID: dagID as ID, nodeID: id as never, @@ -689,9 +736,14 @@ export const layer = Layer.effect( // extend is additive: carry forward pending/queued/paused nodes (with their // existing config definition) so replan treats them as "replace" (preserved) // rather than "supersede" (cancelled). Running nodes are intentionally - // excluded — a running node absent from the fragment is already kept - // unchanged by replan, so there is nothing to carry forward. Terminal - // nodes are immutable and need no preservation. + // excluded — the merged config (computeMergedConfig: surviving = every + // non-cancel node) already keeps a running node's definition whether or + // not the fragment mentions it, so there is nothing to carry forward. + // Note (§3.7): the WorkflowReplanned handler re-times a running survivor + // only when the replan carries a NEW worker_config.timeout_ms for it + // (deadline = now + new timeout). Unchanged/omitted timeout keeps the + // current deadline and the extension count is never reset by an extend. + // Terminal nodes are immutable and need no preservation. const toPreserve = nodes.filter((n) => !newIds.has(n.id) && (n.status === NodeStatus.PENDING || n.status === NodeStatus.QUEUED || n.status === NodeStatus.PAUSED)) if (toPreserve.length > 0 && !config) { return yield* Effect.fail(new Error(`Cannot extend: workflow config is unparseable — would silently cancel ${toPreserve.length} pending node(s)`)) @@ -791,6 +843,32 @@ export const layer = Layer.effect( yield* guardNode(dagID, nodeID, NodeStatus.PENDING) yield* events.publish(DagEvent.NodeRestarted, { dagID: dagID as ID, nodeID: nodeID as never, childSessionID: childSessionID as never, timestamp: yield* DateTime.now }) }) + // 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) { + yield* guardWorkflowNotTerminal(dagID, "timeout escalation") + yield* events.publish(DagEvent.NodeTimeoutEscalated, { + dagID: dagID as ID, + nodeID: nodeID as never, + childSessionID: childSessionID as never, + timeoutExtensions, + timestamp: yield* DateTime.now, + }) + }) + // Replan with a new worker_config.timeout_ms recomputes the absolute + // deadline and persists it on the node row (Q6: from the adjudication + // moment). The deadline watcher is rebuilt by the replan handler. The lock + // witness matters: updateNodeDeadline guards status='running', and the + // guard is only race-free while the caller holds the workflow lock. + // Returns the number of rows written — 0 when the running-guard rejects + // (the node terminalized between the caller's read and this write), so the + // caller can observe the silent no-op instead of logging a false success. + // The store write itself cannot fail (updateNodeDeadline orDies its SQL), + // so the only typed-error channel on this command is withWorkflowLock. + const nodeExtendTimeout = Effect.fn("Dag.nodeExtendTimeout")(function* (lock: WorkflowLock, dagID: string, nodeID: string, newDeadlineMs: number) { + return yield* store.updateNodeDeadline(dagID, nodeID, newDeadlineMs) + }) return Service.of({ create, @@ -811,6 +889,9 @@ 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)), + nodeExtendTimeout: (dagID, nodeID, newDeadlineMs) => withWorkflowLock(dagID)((lock) => nodeExtendTimeout(lock, dagID, nodeID, newDeadlineMs)), }) }), ) diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index ccd87bb589..7348a462e1 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -1,6 +1,6 @@ export * as DagLoop from "./loop" -import { Cause, Effect, Layer, Context, Stream, Semaphore, Fiber, Option } from "effect" +import { Cause, Effect, Layer, Context, Stream, Semaphore, Fiber, Option, DateTime, Clock } from "effect" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { InstanceState } from "@/effect/instance-state" import { EventV2Bridge } from "@/event-v2-bridge" @@ -26,7 +26,7 @@ import { SessionStatus } from "@/session/status" import { renderTemplate } from "../templates/resolve" import { sanitizeInput } from "../templates/sanitize" import { DagConfig } from "../config" -import { spawnNode } from "./spawn" +import { spawnNode, makeDeadlineWatcher } from "./spawn" import { evaluateCondition, resolveInputMapping } from "./eval" import { reconcileWorkflow, makeSessionStatusChecker } from "./recovery" @@ -43,6 +43,7 @@ interface WorkflowEntry { parentSessionID: string config: WorkflowConfig | undefined fibers: Map> + watchers: Map> } export const layer = Layer.effect( @@ -219,8 +220,15 @@ export const layer = Layer.effect( entry.runtime.markRunning(nodeID) const oldFiber = entry.fibers.get(nodeID) + const oldWatcher = entry.watchers.get(nodeID) yield* abortChild(nodeID, node.childSessionId).pipe(Effect.ignore) if (oldFiber) yield* Fiber.interrupt(oldFiber).pipe(Effect.ignore) + // Interrupt the old watcher BEFORE spawning a new one — otherwise + // the old self-renewing watcher survives as a phantom (it is + // unreachable from the map after the overwrite below) and keeps + // escalating against the stale deadline, double-counting + // timeout_extensions and sending duplicate wake notifications. + if (oldWatcher) yield* Fiber.interrupt(oldWatcher).pipe(Effect.ignore) yield* spawnNode(entry.semaphore, { dagID, nodeID, @@ -235,8 +243,14 @@ export const layer = Layer.effect( : undefined, fallbackModel: DagConfig.tierModel(dagConfig, { required: node.required, workerType: node.workerType }), variant: dagConfig.thinking_depth, + maxTimeoutExtensions: entry.config?.max_timeout_extensions ?? Dag.DEFAULT_WORKFLOW_CONFIG.maxTimeoutExtensions, }).pipe( - Effect.tap((result) => Effect.sync(() => entry.fibers.set(nodeID, result.fiber))), + Effect.tap((result) => + Effect.sync(() => { + entry.fibers.set(nodeID, result.fiber) + entry.watchers.set(nodeID, result.watcherFiber) + }), + ), Effect.provideService(Dag.Service, dag), Effect.provideService(Agent.Service, agentSvc), Effect.provideService(Session.Service, sessionSvc), @@ -316,7 +330,10 @@ export const layer = Layer.effect( // first yield so the second caller drops out immediately. if (runtimes.has(dagID) || recovering.has(dagID)) return recovering.add(dagID) - try { + // Effect.ensuring releases the adoption slot even when a fiber + // interrupt cuts the adoption sequence — a finally block does not + // survive interruption. + yield* Effect.gen(function* () { const config = parseWorkflowConfig(wf.config) const recovery = yield* reconcileWorkflow( dagID, @@ -372,7 +389,7 @@ export const layer = Layer.effect( const isStepping = wf.status === "stepping" if (isPaused) runtime.setPaused(true) if (isStepping) runtime.setStepMode(true) - const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map() } + const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map(), watchers: new Map() } runtimes.set(dagID, entry) // Reconciliation settles every persisted running attempt before the // runtime is rebuilt. Recovery never adopts or restarts provider work; @@ -391,18 +408,70 @@ export const layer = Layer.effect( if (pausedForRecovery) { yield* tryDeliverWake(wf.sessionId).pipe(Effect.ignore, Effect.forkScoped) } - } finally { - recovering.delete(dagID) - } + }).pipe(Effect.ensuring(Effect.sync(() => recovering.delete(dagID)))) + }) + + // Orphan-pending recovery: a pending workflow with no non-pending node + // is a create sequence that crashed mid-way — Dag.create publishes + // WorkflowCreated + NodeRegistered + WorkflowStarted in separate + // transactions, so a crash between them leaves a row whose start event + // never arrives. PENDING→RUNNING is its only legal transition + // (core/dag/core/types.ts), so nothing else can ever move it — without + // this sweep the workflow hangs in pending forever. Terminalize via the + // legal pending→running→failed sequence: the projector accepts + // WorkflowFailed only from running/stepping (core/dag/projector.ts), so + // a bare fail would be silently dropped. Failed (not cancelled) matches + // the interrupted-create semantics and persists the reason in the + // durable WorkflowFailed event; cancelled is reserved for explicit + // user/agent cancels (see the checkCompletion attribution comment). + const recoverOrphanPending = Effect.fn("DagLoop.recoverOrphanPending")(function* (wf: DagStore.WorkflowRow) { + // Same cross-instance guard as recoverWorkflow: only the owning + // project's instance may dispose of the orphan. + if (wf.projectId !== ctx.project.id) return + const dagID = wf.id + if (runtimes.has(dagID) || recovering.has(dagID)) return + // Reserve the adoption slot for the whole terminalization sequence: + // the WorkflowStarted leg is a real event on the bus, and the + // WorkflowStarted handler must not adopt the orphan mid-sequence + // (a zero-node orphan would be checkCompleted straight to + // "completed" instead of being failed with the recovery reason). + // Effect.ensuring releases the slot even when a fiber interrupt cuts + // the sequence — a finally block does not survive interruption. + recovering.add(dagID) + yield* Effect.gen(function* () { + const nodes = yield* store.getNodes(dagID) + // Defensive no-miss-kill criterion: any non-pending node proves the + // workflow was adopted and progressed — it is mid-flight, not + // orphaned. All-pending rows can only be interrupted creates, since + // create() completes its start event within the same process. + if (!nodes.every((node) => node.status === "pending")) return + yield* events.publish(DagEvent.WorkflowStarted, { dagID: dagID as never, timestamp: yield* DateTime.now }) + // dag.fail guards running→failed, persists the reason in the durable + // WorkflowFailed event, and terminalizes every pending node via + // NodeSkipped — no node is ever scheduled. + yield* dag.fail(dagID, "orphan pending workflow recovered at startup") + yield* Effect.logWarning("DagLoop terminalized orphan pending workflow", { dagID }) + }).pipe(Effect.ensuring(Effect.sync(() => recovering.delete(dagID)))) }) yield* events.subscribe(DagEvent.WorkflowStarted).pipe( Stream.runForEach((evt) => Effect.gen(function* () { const dagID = evt.data.dagID as string - if (runtimes.has(dagID)) return + // Adoption-in-flight reservations (recoverWorkflow and the + // orphan-pending sweep) must suppress this handler too: the + // orphan sweep publishes WorkflowStarted only to legalize its + // pending→running→failed terminalization, and adopting the + // orphan mid-sequence would start scheduling on a dead workflow. + if (runtimes.has(dagID) || recovering.has(dagID)) return const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) if (!wf) return + // Status guard: the orphan-pending sweep publishes WorkflowStarted + // only to legalize the pending→running leg of its terminalization + // sequence. By the time the event reaches this handler the row is + // already failed — adopting it would rebuild a runtime and start + // scheduling nodes on a dead workflow. Accept running rows only. + if (wf.status !== "running") return // Cross-instance guard: only the owning project's instance adopts // (see recoverWorkflow). First-wave spawns must not race across // directory contexts. @@ -412,7 +481,7 @@ export const layer = Layer.effect( const maxConcurrency = Math.max(1, config?.max_concurrency ?? Dag.DEFAULT_WORKFLOW_CONFIG.maxConcurrency) const runtime = new WorkflowRuntime(toSchedulingNodes(nodes), maxConcurrency) const semaphore = Semaphore.makeUnsafe(maxConcurrency) - const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map() } + const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map(), watchers: new Map() } runtimes.set(dagID, entry) yield* entry.evalLock.withPermits(1)( Effect.gen(function* () { @@ -450,22 +519,33 @@ export const layer = Layer.effect( const expected = def === DagEvent.NodeSkipped ? "skipped" : "completed" const node = yield* store.getNode(dagID, nodeID) const confirmed = node?.status === expected - // Cancel-skip race: workflow-level cancel publishes NodeSkipped - // for running nodes, and this handler may win the cross-stream - // race against WorkflowCancelled. Deleting the fiber here - // uninterrupted would orphan it from the WorkflowCancelled - // sweep and the child session would keep running until its - // prompt finishes or times out. Stop it now, mirroring the - // NodeCancelled handler. Completed nodes keep the plain - // delete — their fiber published the event and is finishing. - if (confirmed && def === DagEvent.NodeSkipped) { - const fiber = entry.fibers.get(nodeID) - if (fiber) { - yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore) - yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + if (confirmed) { + if (def === DagEvent.NodeSkipped) { + // Cancel-skip race: workflow-level cancel publishes NodeSkipped + // for running nodes, and this handler may win the cross-stream + // race against WorkflowCancelled. Deleting the fiber here + // uninterrupted would orphan it from the WorkflowCancelled + // sweep and the child session would keep running until its + // prompt finishes or times out. Stop it now, mirroring the + // NodeCancelled handler. Completed nodes keep the plain + // delete — their fiber published the event and is finishing. + const fiber = entry.fibers.get(nodeID) + if (fiber) { + yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore) + yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + } } + // N3: interrupt the watcher on BOTH terminal events. A node + // re-timed via replan carries a REPLACED watcher in + // entry.watchers; the spawn-time cleanup only interrupts the + // original watcherFiber, so without this the replacement + // lingers until its deadline wake (≤ the extended timeout) + // before it re-reads a terminal row and exits. + const watcher = entry.watchers.get(nodeID) + if (watcher) yield* Fiber.interrupt(watcher).pipe(Effect.ignore) + entry.fibers.delete(nodeID) + entry.watchers.delete(nodeID) } - if (confirmed) entry.fibers.delete(nodeID) if (!confirmed) { yield* Effect.logDebug("DagLoop dropped stale node terminal event", { dagID, nodeID, expected, dbStatus: node?.status ?? "missing" }) } @@ -512,6 +592,9 @@ export const layer = Layer.effect( yield* Fiber.interrupt(fiber).pipe(Effect.ignore) entry.fibers.delete(nodeID) } + const watcher = entry.watchers.get(nodeID) + if (watcher) yield* Fiber.interrupt(watcher).pipe(Effect.ignore) + entry.watchers.delete(nodeID) entry.runtime.markUnsatisfied(nodeID) yield* checkCompletion(dagID) }), @@ -546,9 +629,12 @@ export const layer = Layer.effect( // would incorrectly flip a satisfied node to unsatisfied. if (node?.status === "failed" && entry.runtime.isActive(nid)) { const fiber = entry.fibers.get(nid) + const watcher = entry.watchers.get(nid) entry.fibers.delete(nid) + entry.watchers.delete(nid) yield* abortChild(nid, node.childSessionId ?? null).pipe(Effect.ignore) if (fiber) yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + if (watcher) yield* Fiber.interrupt(watcher).pipe(Effect.ignore) entry.runtime.markUnsatisfied(nid) if (!entry.runtime.isStepMode()) yield* spawnReady(dagID) } @@ -567,6 +653,24 @@ export const layer = Layer.effect( Effect.forkScoped({ startImmediately: true }), ) + // Timeout escalation: the node keeps RUNNING — the runtime needs no + // state change. The event's only job is to wake the main agent so it + // can adjudicate (extend via replan with a new timeout_ms, or + // cancel/replan). Delivery re-reads the wake snapshot, where the + // escalated running node now appears (timeout_extensions > 0). + yield* events.subscribe(DagEvent.NodeTimeoutEscalated).pipe( + Stream.filter((e) => runtimes.has(e.data.dagID as string)), + Stream.runForEach((evt) => + Effect.gen(function* () { + const dagID = evt.data.dagID as string + const entry = runtimes.get(dagID) + if (!entry) return + yield* tryDeliverWake(entry.parentSessionID).pipe(Effect.ignore, Effect.forkScoped) + }).pipe(guarded("NodeTimeoutEscalated")), + ), + Effect.forkScoped({ startImmediately: true }), + ) + // Workflow-control handlers cross-check the durable row under the // evalLock before mutating runtime flags: projection is transactional // with publish, so the row reflects this event or a later one — never @@ -656,9 +760,108 @@ export const layer = Layer.effect( yield* entry.evalLock.withPermits(1)( Effect.gen(function* () { const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) + const oldConfig = entry.config if (wf) entry.config = parseWorkflowConfig(wf.config) const nodes = yield* store.getNodes(dagID) entry.runtime.rebuildGraph(toSchedulingNodes(nodes)) + // Timeout extension (Q6): a running node gets a recomputed + // deadline (now + new timeout) ONLY when the replan carries a + // NEW worker_config.timeout_ms for it (§3.7) AND the node + // actually needs re-timing (deadline elapsed or escalation + // pending — see the gate below). Restarted nodes are pending + // here — their new attempt spawns a fresh watcher via + // spawnReady. + const newConfig = entry.config + for (const node of nodes) { + if (node.status !== "running") continue + const frag = newConfig?.nodes.find((candidate) => candidate.id === node.id) + if (!frag) continue + const oldTimeoutMs = oldConfig?.nodes.find((candidate) => candidate.id === node.id)?.worker_config?.timeout_ms + const fragTimeoutMs = frag.worker_config?.timeout_ms + // §3.7: re-time only when the replan carries a NEW + // timeout_ms. The persisted config behind WorkflowReplanned + // is the MERGED config — every non-cancel survivor keeps its + // definition — so a node the fragment never mentioned, or + // re-specified with an unchanged/omitted timeout_ms, + // matches here with its OLD timeout. + if (fragTimeoutMs == null || fragTimeoutMs === oldTimeoutMs) continue + const now = yield* Clock.currentTimeMillis + // Cap gate (A1): a changed timeout alone must not move a + // healthy deadline forward. An agent replanning BEFORE each + // deadline with cycling values (10m→20m→10m…) would push the + // deadline away forever without a single escalation firing, + // so the extension count never climbs and the ≈21× cap is + // bypassed. Re-time only when the current deadline already + // elapsed or an escalation awaits adjudication; a gated-off + // node keeps its deadline and the self-renewing watcher + // escalates it the moment it passes. A null deadline is + // treated as elapsed — re-timing is what re-establishes + // supervision. + if (!node.escalationPending && node.deadlineMs != null && node.deadlineMs > now) continue + // N1: write the new deadline FIRST. nodeExtendTimeout + // acquires the workflow lock and can fail or block; if the + // write never lands, the old watcher must keep supervising + // the old deadline — interrupting it beforehand would leave + // a RUNNING node with no watcher, no escalation, and a + // defeated cap backstop (§5-5). + // D1: one node's failed extend must not abort the rest of + // the handler — an uncaught failure propagates to + // guarded("WorkflowReplanned") and skips the stale-fiber + // sweep below plus spawnReady/checkCompletion, leaving + // restarted nodes pending with nobody to schedule them. + // A failed write also leaves the deadline unmoved, so the + // old watcher keeps supervising (N1) — this path must not + // touch it. Interruption still propagates: hasInterrupts is + // a structural check, whereas Cause.interruptors collects + // only DEFINED fiber IDs and ignores interrupt reasons + // carrying none — those would be swallowed as errors here. + const written = yield* dag.nodeExtendTimeout(dagID, node.id, now + fragTimeoutMs).pipe( + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.failCause(cause) + : Effect.logWarning("DagLoop replan re-time failed; keeping the old watcher and continuing the batch", { dagID, nodeID: node.id, cause }).pipe( + Effect.as(-1), + ), + ), + ) + if (written < 0) continue + if (written === 0) { + // The status='running' guard rejected the write — the node + // terminalized between the getNodes read and this update. + // No deadline was written; stop the old watcher and do not + // install one for a row the store refused to touch. + const deadWatcher = entry.watchers.get(node.id) + if (deadWatcher) yield* Fiber.interrupt(deadWatcher).pipe(Effect.ignore) + entry.watchers.delete(node.id) + yield* Effect.logWarning("DagLoop replan re-time skipped — node no longer running", { dagID, nodeID: node.id }) + continue + } + // Write committed: install the re-armed watcher BEFORE + // interrupting the old one so supervision is never absent. + // F8's original race is benign in this order — the old + // watcher's next read sees the future deadline and sleeps; + // an escalation already in flight (lock-serialized behind + // this write) only adds a counted extension toward the cap, + // it never removes supervision. Its sole residue is re-setting + // escalation_pending on the now-extended node, which costs one + // redundant wake and permits one extra re-time via the gate + // above — cosmetic; the cap accounting still holds because the + // count did climb. + const newWatcher = yield* makeDeadlineWatcher({ + dagID, + nodeID: node.id, + timeoutMs: fragTimeoutMs, + maxTimeoutExtensions: newConfig?.max_timeout_extensions ?? Dag.DEFAULT_WORKFLOW_CONFIG.maxTimeoutExtensions, + }).pipe( + Effect.provideService(Dag.Service, dag), + Effect.provideService(SessionPrompt.Service, promptSvc), + Effect.forkScoped, + ) + const oldWatcher = entry.watchers.get(node.id) + entry.watchers.set(node.id, newWatcher) + if (oldWatcher) yield* Fiber.interrupt(oldWatcher).pipe(Effect.ignore) + yield* Effect.logInfo("DagLoop extended node deadline via replan", { dagID, nodeID: node.id, newDeadlineMs: now + fragTimeoutMs }) + } // Replan resets restarted nodes to pending. Old fibers of nodes // that are no longer running/queued must be interrupted here: // nothing else will (there is no NodeRestarted subscriber), and @@ -670,7 +873,10 @@ export const layer = Layer.effect( if (node && (node.status === "running" || node.status === "queued")) continue yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore) yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + const watcher = entry.watchers.get(nodeID) + if (watcher) yield* Fiber.interrupt(watcher).pipe(Effect.ignore) entry.fibers.delete(nodeID) + entry.watchers.delete(nodeID) } yield* spawnReady(dagID) yield* checkCompletion(dagID) @@ -696,8 +902,11 @@ export const layer = Layer.effect( const node = yield* store.getNode(dagID, nodeID) yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore) yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + const watcher = entry.watchers.get(nodeID) + if (watcher) yield* Fiber.interrupt(watcher).pipe(Effect.ignore) } entry.fibers.clear() + entry.watchers.clear() runtimes.delete(dagID) }), ) @@ -727,6 +936,21 @@ export const layer = Layer.effect( const terminalWorkflows = snapshot.workflows.filter( (workflow) => !workflow.wakeReported && isWorkflowTerminalStatus(workflow.status as never), ) + // Timeout-escalated nodes must reach the main agent for + // adjudication — their workflow is a delivery boundary even though + // the runtime still reports a running node. F11: a non-eligible node + // that escalated then terminalized (cap-exhausted force-cancel) must + // deliver its verdict immediately, not wait for the next natural + // boundary (it may never come while other nodes keep running). + // The boundary is escalation_pending (a live, not-yet-adjudicated + // escalation) OR an escalated node that terminalized — NOT the sticky + // extension count alone, or an already-adjudicated running node would + // override the delivery boundary for the rest of the attempt. + const escalatedWorkflowIDs = new Set( + snapshot.nodes + .filter((node) => node.escalationPending || (node.timeoutExtensions > 0 && isNodeTerminalStatus(node.status as never))) + .map((node) => node.workflowId), + ) const workflowIDs = [...new Set([ ...snapshot.nodes.map((node) => node.workflowId), ...terminalWorkflows.map((workflow) => workflow.id), @@ -740,6 +964,7 @@ export const layer = Layer.effect( if (workflow.status === "paused" || workflow.status === "stepping") return true if (entry?.runtime.isPaused() || entry?.runtime.isStepMode()) return true if (workflow.status !== "running" || !entry) return false + if (escalatedWorkflowIDs.has(workflow.id)) return true // Delivery boundary uses the runtime's own running set, NOT fiber // ownership: between markRunning and fibers.set the spawn path has // async yield points, and a wake reading that window would misjudge @@ -870,6 +1095,14 @@ export const layer = Layer.effect( } const summaries = [ ...batch.nodes.map((node) => { + // The timeout advisory is for a running node awaiting + // adjudication (escalation_pending). A batch node is + // wake_reported=false, so it cannot be an already-adjudicated + // extend — escalation_pending and timeoutExtensions>0 agree + // here; the flag is the intent-level signal. + if (node.status === "running" && node.escalationPending) { + return `[DAG Node Timeout] RUNNING node "${node.name}" exceeded its execution deadline (timeout escalation ${node.timeoutExtensions}) and is still executing. Adjudicate by replanning with a NEW worker_config.timeout_ms to extend the node — that grants more execution time, but the cumulative extension count is NOT reset (only a new attempt resets it), and the node is force-cancelled once the cap is reached — or cancel/replan the node. Queued nodes are not extended: their admission deadline was fixed at permit acquisition and is not adjusted by extensions.` + } const output = typeof node.output === "string" ? node.output.slice(0, 500) : node.errorReason ?? (node.output == null ? "(no output)" : JSON.stringify(node.output).slice(0, 500)) @@ -941,6 +1174,17 @@ export const layer = Layer.effect( // Install all live event handlers before spawning recovery watchers so // a child that settles immediately cannot leave the runtime stale. + // Orphan-pending sweep first: the WorkflowStarted it publishes for the + // terminalization leg is rejected by the handler's status guard above + // (the row is already failed once the event arrives), never adopted. + const pendingWfs = yield* store.listByStatus("pending").pipe(Effect.orDie) + for (const wf of pendingWfs) { + yield* recoverOrphanPending(wf).pipe( + Effect.catchCause((cause) => + Effect.logWarning("DagLoop orphan pending recovery failed for workflow", { dagID: wf.id, cause }), + ), + ) + } const runningWfs = yield* store.listByStatus("running").pipe(Effect.orDie) const pausedWfs = yield* store.listByStatus("paused").pipe(Effect.orDie) const steppingWfs = yield* store.listByStatus("stepping").pipe(Effect.orDie) diff --git a/packages/opencode/src/dag/runtime/recovery.ts b/packages/opencode/src/dag/runtime/recovery.ts index f30b46d590..799234d50a 100644 --- a/packages/opencode/src/dag/runtime/recovery.ts +++ b/packages/opencode/src/dag/runtime/recovery.ts @@ -125,9 +125,21 @@ export function reconcileWorkflow( if (node.deadlineMs !== null) { const now = yield* Clock.currentTimeMillis if (now >= node.deadlineMs) { + // S2: recovery of an escalated node preserves the timeout semantics + // — the extension budget was spent before the crash, the deadline + // was never re-extended, and the durable escalation count proves + // it. Failure reason records the escalation so the parent can tell + // "ran out of time after N extensions" from "never escalated". yield* settle( node.id, - dag.nodeFailed(dagID, node.id, "deadline exceeded on recovery", "timeout"), + dag.nodeFailed( + dagID, + node.id, + node.timeoutExtensions > 0 + ? `timeout escalated (${node.timeoutExtensions} extension(s)) node failed on recovery` + : "deadline exceeded on recovery", + "timeout", + ), ) reconciled++ continue diff --git a/packages/opencode/src/dag/runtime/spawn.ts b/packages/opencode/src/dag/runtime/spawn.ts index 633bb53fbd..1adee59822 100644 --- a/packages/opencode/src/dag/runtime/spawn.ts +++ b/packages/opencode/src/dag/runtime/spawn.ts @@ -7,8 +7,15 @@ * Admission model (P0-2): the node is durably QUEUED at dispatch — the child * session and NodeStarted only materialize INSIDE the concurrency permit, so a * 100-node fan-out no longer creates 100 sessions and shows 100 "running" - * rows while true concurrency is 5. The deadline is fixed at admission time: - * queue wait counts toward the node's budget. + * rows while true concurrency is 5. The admission deadline is fixed when the + * node is admitted (deadline = admission time + timeout_ms) and is NOT + * adjusted by running-node extensions (F4): the replan/extend handler re-times + * RUNNING nodes only, so a queued node's pre-permit wait keeps its admission + * deadline — queue wait counts toward the node's budget, and an expired + * queued node fails directly via the pre-permit timeout path (no progress to + * protect). A queued node absent from a plain replan fragment is superseded + * (cancelled) by planReplan; the additive extend path re-admits it with a + * fresh admission deadline. * * Completion model (mirrors task.ts:210-221): a node completes when its child * session's prompt() resolves; it fails when prompt() fails. The completion @@ -20,7 +27,7 @@ * (Level 2) is a documented boundary — see eval.ts. */ -import { Effect, Semaphore, Scope, Fiber, Option, Clock, Cause } from "effect" +import { Effect, Semaphore, Scope, Fiber, Option, Clock, Cause, Exit } from "effect" import { Agent } from "@/agent/agent" import { Session } from "@/session/session" import { SessionID, MessageID } from "@/session/schema" @@ -28,7 +35,7 @@ import { deriveSubagentSessionPermission } from "@/agent/subagent-permissions" import { SessionPrompt } from "@/session/prompt" import { Dag } from "../dag" import { DagModel } from "../model" -import { isTransitionRejection } from "@opencode-ai/core/dag/core/types" +import { isTransitionRejection, isNodeTerminalStatus } from "@opencode-ai/core/dag/core/types" import type { DagStore } from "@opencode-ai/core/dag/store" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" @@ -50,10 +57,169 @@ export interface NodeSpawnInput { fallbackModel?: { modelID: string; providerID: string } /** dag.jsonc thinking_depth — forwarded as the prompt variant (no-op unless the model defines it). */ variant?: string + /** Workflow-level timeout extension cap (defaults to DEFAULT_WORKFLOW_CONFIG.maxTimeoutExtensions). */ + maxTimeoutExtensions?: number } export interface NodeSpawnResult { fiber: Fiber.Fiber + /** Deadline watcher fiber — rebuilt by the replan handler when the deadline is extended. */ + watcherFiber: Fiber.Fiber +} + +export interface DeadlineWatcherInput { + dagID: string + nodeID: string + /** + * The node's effective execution timeout. Doubles as the escalation + * interval (S1): after escalating, the watcher waits one timeout period + * before re-reading — an extended deadline (replan adjudication) moves the + * row's deadline into the future and the loop sleeps until it; an untouched + * deadline escalates again, driving the count toward the extension cap. + */ + timeoutMs?: number + /** Workflow-level timeout extension cap (defaults to DEFAULT_WORKFLOW_CONFIG.maxTimeoutExtensions). */ + maxTimeoutExtensions?: number +} + +/** + * Deadline watcher (timeout = signal, not failure). Sleeps until the node's + * absolute deadline, then reads the durable row: + * - node no longer running → exit (cancelled/restarted/completed) + * - deadline extended on the row (replan with a new timeout_ms) → re-sleep + * - extension cap exhausted → cancel the child + nodeFailed("timeout") + * - otherwise → publish NodeTimeoutEscalated (node stays RUNNING) + * The row is the single source of truth, so a watcher that survives its + * execution fiber (interrupt misses) self-heals on the next wake-up. + * + * S1: the watcher self-renews — it does NOT exit after escalating. It waits + * one escalate interval and re-reads the row: a replan that extended the + * deadline (a NEW worker_config.timeout_ms — nodeExtendTimeout recomputes + * from now per §3.7) is picked up on the next read; a deadline the main agent + * never adjudicated escalates AGAIN, so the extension count climbs toward the + * cap and supervision stays bounded even when no replan ever arrives. F5: a + * queued/pending/paused node past its deadline still polls instead of + * exiting — the node may yet acquire the permit inside its admission window + * (edge-deadline permit) and start running under supervision. + */ +export function makeDeadlineWatcher( + input: DeadlineWatcherInput, +): Effect.Effect { + return Effect.gen(function* () { + const dag = yield* Dag.Service + const promptSvc = yield* SessionPrompt.Service + const escalateIntervalMs = Math.max(1_000, input.timeoutMs ?? Dag.DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs) + // Read the durable row. Transient store failures (SQLite lock blips, + // connection hiccups) must NOT end supervision — a single failed read + // would otherwise permanently orphan the node's timeout path — so the + // read retries with a short backoff (R13) and only gives up after every + // attempt fails. Effect.exit captures both effect failures and defects. + const readNode = Effect.gen(function* () { + for (let attemptNo = 0; attemptNo <= 3; attemptNo++) { + const outcome = yield* dag.store.getNode(input.dagID, input.nodeID).pipe(Effect.exit) + if (Exit.isSuccess(outcome)) return outcome.value + if (attemptNo < 3) yield* Effect.sleep(500) + } + yield* Effect.logWarning("DAG deadline watcher giving up after store read retries", { dagID: input.dagID, nodeID: input.nodeID }) + return undefined + }) + for (;;) { + const node = yield* readNode + if (!node) { + // Store read failed after all retries — do NOT exit (the watcher + // "must not end supervision"). Sleep with a longer backoff and + // retry the read in the next loop iteration; a transient store + // outage should not permanently orphan the node's timeout path. + yield* Effect.sleep(5_000) + continue + } + if (isNodeTerminalStatus(node.status as never)) return + const now = yield* Clock.currentTimeMillis + const deadline = node.deadlineMs + if (node.status !== "running") { + // Pre-running wait (F5): a queued/pending/paused node past its + // deadline may still acquire the permit inside its admission window — + // do not abandon supervision; poll until it starts, terminalizes, or + // the pre-permit timeout path fails it. + yield* Effect.sleep(sleepUntilDeadlineMs(deadline, now, 1_000)) + continue + } + if (deadline === null || deadline > now) { + yield* Effect.sleep(sleepUntilDeadlineMs(deadline, now, 10)) + continue + } + const extensions = node.timeoutExtensions + const maxExtensions = input.maxTimeoutExtensions ?? Dag.DEFAULT_WORKFLOW_CONFIG.maxTimeoutExtensions + if (extensions >= maxExtensions) { + yield* promptSvc.cancel(node.childSessionId as never).pipe(Effect.ignore) + // Enforcing the cap IS the watcher's contract (§5-5), so a transient + // failure here must retry rather than end supervision: returning would + // leave a RUNNING node past its cap with nobody left to fail it. A + // rejected guard means someone else already terminalized the node, + // which counts as done. + const outcome = yield* dag.nodeFailed(input.dagID, input.nodeID, `timeout extensions exhausted (${extensions}/${maxExtensions})`, "timeout").pipe( + Effect.catchIf( + isTransitionRejection, + () => Effect.logWarning("nodeFailed (timeout extensions exhausted) guard rejected — node already terminal"), + ), + Effect.exit, + ) + if (Exit.isSuccess(outcome)) return + if (Cause.hasInterrupts(outcome.cause)) return yield* Effect.failCause(outcome.cause) + yield* Effect.logWarning("DAG deadline watcher cap enforcement failed — retrying", { dagID: input.dagID, nodeID: input.nodeID, cause: outcome.cause }) + yield* Effect.sleep(escalateIntervalMs) + continue + } + // The escalate write takes the workflow lock and publishes a durable + // event, so it can fail with a typed Error (lock contention) or die (a + // publish defect). Neither may end supervision: an exited watcher stops + // escalating, the extension count stops climbing, `extensions >= + // maxExtensions` never becomes true, and the node occupies a concurrency + // 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( + Effect.catchIf( + isTransitionRejection, + () => Effect.logWarning("nodeTimeoutEscalated guard rejected — node already terminal"), + ), + Effect.exit, + ) + if (Exit.isFailure(escalated)) { + if (Cause.hasInterrupts(escalated.cause)) return yield* Effect.failCause(escalated.cause) + yield* Effect.logWarning("DAG deadline watcher escalation failed — keeping supervision and retrying", { dagID: input.dagID, nodeID: input.nodeID, cause: escalated.cause }) + } + // Self-renew (S1): stay alive after escalating. Wait one escalate + // interval, then loop — the re-read sees an extended deadline (replan + // adjudication via nodeExtendTimeout) and sleeps until it, or sees a + // still-past deadline and escalates AGAIN. Repeated escalation is what + // drives the count toward the cap, so a main agent that never replans + // cannot leave the node running unbounded. + yield* Effect.sleep(escalateIntervalMs) + } + }).pipe( + // Last-resort net for defects outside the loop's own handling. Use + // hasInterrupts, not interruptors: the latter collects DEFINED fiber IDs + // and silently ignores interrupt reasons carrying none, which would + // misclassify such an interrupt as an error. + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.void + : Effect.logWarning("DAG deadline watcher exited with an error", { cause }), + ), + ) +} + +/** + * How long the deadline watcher sleeps before re-reading the node row. With no + * deadline yet, poll fast (100ms) so a late deadline write is seen quickly; + * with a future deadline, sleep exactly until it (min 10ms); once past the + * deadline, back off by overdueMs before the next read. + */ +function sleepUntilDeadlineMs(deadlineMs: number | null, now: number, overdueMs: number) { + if (deadlineMs === null) return 100 + if (deadlineMs > now) return Math.max(deadlineMs - now, 10) + return overdueMs } export function spawnNode( @@ -79,7 +245,7 @@ export function spawnNode( () => Effect.logWarning(`nodeFailed (${label}) guard rejected — node already terminal`), ), ) - return { fiber: yield* Effect.forkIn(scope)(Effect.void) } + return { fiber: yield* Effect.forkIn(scope)(Effect.void), watcherFiber: yield* Effect.forkIn(scope)(Effect.void) } }) const agent = yield* agentService.get(input.node.workerType).pipe( @@ -145,13 +311,24 @@ export function spawnNode( ) if (!admitted) { const fiber = yield* Effect.forkIn(scope)(Effect.void) - return { fiber } + const watcherFiber = yield* Effect.forkIn(scope)(Effect.void) + return { fiber, watcherFiber } } // Assigned inside the fiber once the child session materializes; read by // the ensuring/onInterrupt cleanups below. let childSessionID: string | undefined + // Forked first so the execution fiber's cleanup closures can capture it. + const watcherFiber = yield* Effect.forkIn(scope)( + makeDeadlineWatcher({ + dagID: input.dagID, + nodeID: input.nodeID, + timeoutMs, + maxTimeoutExtensions: input.maxTimeoutExtensions, + }), + ) + const fiber = yield* Effect.forkIn(scope)( Effect.gen(function* () { // P1(#1): Acquire permit with a deadline-bounded timeout so the node @@ -215,27 +392,18 @@ export function spawnNode( if (input.outputSchema) registerCaptureSlot(childSession.id, input.outputSchema) - // Run the actual prompt with the remaining time budget. - const permitTime = yield* Clock.currentTimeMillis - const remainingMs = Math.max(0, deadlineMs - permitTime) - const resultOpt = yield* promptSvc.prompt({ + // The prompt runs WITHOUT a timeout — the deadline watcher owns the + // timeout path (escalate signal vs exhausted-force-cancel). A timeout + // never interrupts the child session mid-work; it only notifies the + // main agent, which adjudicates (extend / cancel / replan). + const result = yield* promptSvc.prompt({ messageID: MessageID.ascending(), sessionID: childSession.id, model, agent: agent.name, ...(input.variant ? { variant: input.variant } : {}), parts: input.promptParts, - }).pipe(Effect.timeoutOption(remainingMs)) - if (Option.isNone(resultOpt)) { - yield* promptSvc.cancel(childSession.id).pipe(Effect.ignore) - yield* dag.nodeFailed(input.dagID, input.nodeID, `node exceeded timeout of ${timeoutMs}ms`, "timeout").pipe( - Effect.catchIf( - isTransitionRejection, - () => Effect.logWarning("nodeFailed (timeout) guard rejected — node already terminal"), - ), - ) - return - } + }) if (input.outputSchema) { clearCaptureSlot(childSession.id) const updatedNode = yield* dag.store.getNode(input.dagID, input.nodeID).pipe(Effect.orDie) @@ -253,7 +421,7 @@ export function spawnNode( ), ) } else { - const rawText = resultOpt.value.parts.findLast((p) => p.type === "text")?.text ?? "" + const rawText = result.parts.findLast((p) => p.type === "text")?.text ?? "" if (rawText.trim() === "") { yield* dag.nodeFailed( input.dagID, @@ -280,7 +448,10 @@ export function spawnNode( } }).pipe( Effect.ensuring( - Effect.sync(() => { + Effect.gen(function* () { + // The prompt finished (or this fiber was interrupted) — the + // deadline watcher has no further job. + yield* Fiber.interrupt(watcherFiber).pipe(Effect.ignore) if (input.outputSchema && childSessionID) clearCaptureSlot(childSessionID) }), ), @@ -296,7 +467,7 @@ export function spawnNode( ), Effect.catchCause((cause) => Effect.gen(function* () { - if (Cause.interruptors(cause).size > 0) return + if (Cause.hasInterrupts(cause)) return yield* dag.nodeFailed(input.dagID, input.nodeID, Cause.pretty(cause), "exec_failed").pipe( Effect.catchIf( isTransitionRejection, @@ -308,6 +479,6 @@ export function spawnNode( ), ) - return { fiber } + return { fiber, watcherFiber } }) } diff --git a/packages/opencode/src/dag/runtime/summary-publisher.ts b/packages/opencode/src/dag/runtime/summary-publisher.ts index 2860bfe697..e3487172c7 100644 --- a/packages/opencode/src/dag/runtime/summary-publisher.ts +++ b/packages/opencode/src/dag/runtime/summary-publisher.ts @@ -44,6 +44,10 @@ const SUMMARY_TRIGGER_EVENTS = [ DagEvent.NodeSkipped, DagEvent.NodeCancelled, DagEvent.NodeRestarted, + // F10: escalation changes the visible summary (escalatedNodes rises from 0) + // without any status transition — without this trigger the TUI would keep + // showing a plain RUNNING node until some unrelated node event fires. + DagEvent.NodeTimeoutEscalated, ] as const export interface Interface { diff --git a/packages/opencode/src/server/routes/instance/httpapi/groups/dag.ts b/packages/opencode/src/server/routes/instance/httpapi/groups/dag.ts index 2b0b340774..0f857c2285 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/groups/dag.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/groups/dag.ts @@ -64,6 +64,7 @@ export const WorkflowSummaryResponse = Schema.Struct({ failedNodes: Schema.Number, skippedNodes: Schema.Number, queuedNodes: Schema.Number, + escalatedNodes: Schema.Number, }).annotate({ identifier: "Dag.WorkflowSummary" }) export const DagSummaryListResponse = Schema.Array(WorkflowSummaryResponse) diff --git a/packages/opencode/src/server/routes/instance/httpapi/handlers/dag.ts b/packages/opencode/src/server/routes/instance/httpapi/handlers/dag.ts index d79ad98af0..f85c23fc89 100644 --- a/packages/opencode/src/server/routes/instance/httpapi/handlers/dag.ts +++ b/packages/opencode/src/server/routes/instance/httpapi/handlers/dag.ts @@ -116,6 +116,7 @@ export const dagHandlers = HttpApiBuilder.group(InstanceHttpApi, "dag", (handler failedNodes: s.failedNodes, skippedNodes: s.skippedNodes, queuedNodes: s.queuedNodes, + escalatedNodes: s.escalatedNodes, })) }) diff --git a/packages/opencode/test/dag/dag-node-started-guard.test.ts b/packages/opencode/test/dag/dag-node-started-guard.test.ts index d5f679bb06..3deeff9563 100644 --- a/packages/opencode/test/dag/dag-node-started-guard.test.ts +++ b/packages/opencode/test/dag/dag-node-started-guard.test.ts @@ -113,6 +113,7 @@ describe("DagProjector: NodeStarted status guard", () => { failedNodes: 0, skippedNodes: 0, queuedNodes: 0, + escalatedNodes: 0, }, { id: otherDagID, @@ -124,6 +125,7 @@ describe("DagProjector: NodeStarted status guard", () => { failedNodes: 0, skippedNodes: 0, queuedNodes: 0, + escalatedNodes: 0, }, ]) }).pipe(Effect.provide(projectorLayer)) as Effect.Effect, diff --git a/packages/opencode/test/dag/dag-orphan-pending-recovery.test.ts b/packages/opencode/test/dag/dag-orphan-pending-recovery.test.ts new file mode 100644 index 0000000000..4e9b412d29 --- /dev/null +++ b/packages/opencode/test/dag/dag-orphan-pending-recovery.test.ts @@ -0,0 +1,219 @@ +import { describe, expect, it } from "bun:test" +import { DateTime, Effect, Layer, Option } from "effect" +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 { DagEvent } from "@opencode-ai/schema/dag-event" +import { Agent } from "@/agent/agent" +import { Dag } 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 { Session } from "@/session/session" +import { SessionStatus } from "@/session/status" +import { pollWithTimeout } from "../lib/effect" + +const ORPHAN_REASON = "orphan pending workflow recovered at startup" + +function orphanRecoveryLayer(input: { promptCalls: 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 session = Layer.mock(Session.Service, { + create: Effect.fn("test.Session.create")((_value?: unknown) => + Effect.sync(() => ({}) as never), + ), + get: Effect.fn("test.Session.get")(() => Effect.succeed({} as never)), + messages: Effect.fn("test.Session.messages")(() => Effect.succeed([])), + }) + const prompt = Layer.mock(SessionPrompt.Service, { + cancel: Effect.fn("test.SessionPrompt.cancel")(() => Effect.void), + prompt: Effect.fn("test.SessionPrompt.prompt")(() => { + input.promptCalls.push("prompt") + return Effect.never + }), + promptIfIdle: () => Effect.succeed(Option.none()), + }) + const loop = DagLoop.layer.pipe( + Layer.provide(base), + Layer.provide(session), + Layer.provide(prompt), + Layer.provide(Layer.mock(Agent.Service, {})), + ) + return Layer.merge(base, loop) +} + +function runOrphanRecovery( + test: (services: { + dag: Dag.Interface + database: Database.Interface + loop: DagLoop.Interface + events: EventV2.Interface + store: DagStore.Interface + promptCalls: string[] + }) => Effect.Effect, +) { + const promptCalls: string[] = [] + return Effect.gen(function* () { + const dag = yield* Dag.Service + const database = yield* Database.Service + const loop = yield* DagLoop.Service + const events = yield* EventV2.Service + const store = yield* DagStore.Service + return yield* test({ dag, database, loop, events, store, promptCalls }) + }).pipe( + Effect.provide(orphanRecoveryLayer({ promptCalls })), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: "project-1" }, + } as never), + Effect.scoped, + ) +} + +function seedProjectAndSession(database: Database.Interface) { + return Effect.gen(function* () { + yield* database.db.insert(ProjectTable).values({ + id: "project-1" as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.db.insert(SessionTable).values({ + id: "ses_parent1" as never, + project_id: "project-1" as never, + slug: "parent", + directory: process.cwd() as never, + title: "Parent", + version: "test", + }).run().pipe(Effect.orDie) + }) +} + +// Publish the exact durable prefix Dag.create would write, then stop before +// WorkflowStarted — simulating a process crash between the create transactions. +function publishInterruptedCreate( + events: EventV2.Interface, + dagID: DagEvent.DagID, + ts: DateTime.Utc, + nodeCount: number, +) { + return Effect.gen(function* () { + yield* events.publish(DagEvent.WorkflowCreated, { + dagID, + projectID: "project-1" as never, + sessionID: "ses_parent1" as never, + title: "Orphan", + config: JSON.stringify({ name: "orphan", nodes: [] }), + status: "pending", + timestamp: ts, + }) + for (let i = 1; i <= nodeCount; i++) { + yield* events.publish(DagEvent.NodeRegistered, { + dagID, + nodeID: `n${i}` as never, + name: `Node ${i}`, + workerType: "build", + dependsOn: [], + required: true, + timestamp: ts, + }) + } + }) +} + +describe("DagLoop orphan pending recovery", () => { + it("terminalizes a pending workflow whose create crashed before WorkflowStarted", async () => { + await Effect.runPromise( + runOrphanRecovery(({ database, loop, events, store, promptCalls }) => + Effect.gen(function* () { + yield* seedProjectAndSession(database) + const dagID = DagEvent.DagID.create() + yield* publishInterruptedCreate(events, dagID, yield* DateTime.now, 2) + + const failures: Array<{ reason: string }> = [] + const unsubscribe = yield* events.listen((event) => + event.type === DagEvent.WorkflowFailed.type + ? Effect.sync(() => failures.push(event.data as never)) + : Effect.void, + ) + + yield* loop.init() + // Listener fan-out is async-ordered relative to publish (never rely + // on it having run by the time init returns) — wait for the durable + // failure to surface before unsubscribing. + yield* pollWithTimeout( + Effect.sync(() => (failures.some((f) => f.reason === ORPHAN_REASON) ? failures : undefined)), + "WorkflowFailed recovery reason was not observed", + ) + yield* unsubscribe + + const wf = yield* store.getWorkflow(dagID) + expect(wf?.status).toBe("failed") + expect((yield* store.getNodes(dagID)).map((n) => n.status)).toEqual(["skipped", "skipped"]) + expect(failures).toContainEqual(expect.objectContaining({ reason: ORPHAN_REASON })) + // The WorkflowStarted published for the terminalization leg must not + // be adopted: no node may be scheduled on a dead workflow. + expect(promptCalls).toEqual([]) + }), + ), + ) + }) + + it("terminalizes a zero-node orphan pending workflow", async () => { + await Effect.runPromise( + runOrphanRecovery(({ database, loop, events, store }) => + Effect.gen(function* () { + yield* seedProjectAndSession(database) + const dagID = DagEvent.DagID.create() + yield* publishInterruptedCreate(events, dagID, yield* DateTime.now, 0) + + yield* loop.init() + + expect((yield* store.getWorkflow(dagID))?.status).toBe("failed") + }), + ), + ) + }) + + it("leaves a pending workflow with a non-pending node untouched", async () => { + await Effect.runPromise( + runOrphanRecovery(({ database, loop, events, store }) => + Effect.gen(function* () { + yield* seedProjectAndSession(database) + const dagID = DagEvent.DagID.create() + const ts = yield* DateTime.now + yield* publishInterruptedCreate(events, dagID, ts, 1) + // A node that already progressed proves the workflow was adopted and + // mid-flight — the defensive criterion must not terminalize it. + yield* events.publish(DagEvent.NodeStarted, { + dagID, + nodeID: "n1" as never, + childSessionID: "ses_child1" as never, + timestamp: yield* DateTime.now, + }) + + yield* loop.init() + + expect((yield* store.getWorkflow(dagID))?.status).toBe("pending") + expect((yield* store.getNode(dagID, "n1"))?.status).toBe("running") + }), + ), + ) + }) +}) diff --git a/packages/opencode/test/dag/dag-recovery-escalated-loop.test.ts b/packages/opencode/test/dag/dag-recovery-escalated-loop.test.ts new file mode 100644 index 0000000000..3a7f266688 --- /dev/null +++ b/packages/opencode/test/dag/dag-recovery-escalated-loop.test.ts @@ -0,0 +1,207 @@ +import { describe, expect, it } from "bun:test" +import { Effect, Layer, Option } from "effect" +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 type { SessionV1 } from "@opencode-ai/core/v1/session" +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 { MessageID } from "@/session/schema" +import { SessionPrompt } from "@/session/prompt" +import { Session } from "@/session/session" +import { SessionStatus } from "@/session/status" +import { pollWithTimeout } from "../lib/effect" + +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 } } : {}), + } +} + +function reply(sessionID: string, text: string): SessionV1.WithParts { + return { + info: { + id: MessageID.ascending(), + role: "assistant", + parentID: MessageID.ascending(), + sessionID: sessionID as never, + 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, + time: { created: Date.now() }, + finish: "stop", + }, + parts: text ? [{ type: "text", text }] as never : [], + } +} + +function recoveryLayer(input: { wakes: 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) + // The crashed child session is gone: reads yield no durable outcome, so the + // recovery checker reports "unknown" (ownership lost), not a fabricated + // completion/failure read off the child. + const session = Layer.mock(Session.Service, { + create: () => Effect.sync(() => ({}) as never), + get: () => Effect.succeed({} as never), + messages: () => Effect.succeed([]), + }) + const prompt = Layer.mock(SessionPrompt.Service, { + cancel: () => Effect.void, + prompt: () => Effect.never, + promptIfIdle: (value) => + Effect.sync(() => { + const text = value.parts.find((part) => part.type === "text")?.text + if (text) input.wakes.push(text) + }).pipe( + Effect.map(() => Option.some(reply(value.sessionID as string, "wake handled"))), + ), + }) + const loop = DagLoop.layer.pipe( + Layer.provide(base), + Layer.provide(session), + Layer.provide(prompt), + Layer.provide(Layer.mock(Agent.Service, {})), + ) + return Layer.merge(base, loop) +} + +function runRecoveryTest( + test: (services: { + dag: Dag.Interface + database: Database.Interface + loop: DagLoop.Interface + store: DagStore.Interface + wakes: string[] + }) => Effect.Effect, +) { + const wakes: string[] = [] + return Effect.gen(function* () { + const dag = yield* Dag.Service + const database = yield* Database.Service + const loop = yield* DagLoop.Service + const store = yield* DagStore.Service + yield* database.db.insert(ProjectTable).values({ + id: "project-1" as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.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) + return yield* test({ dag, database, loop, store, wakes }) + }).pipe( + Effect.provide(recoveryLayer({ wakes })), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: "project-1" }, + } as never), + Effect.scoped, + ) +} + +describe("DagLoop escalated crash recovery (loop-level E2E)", () => { + it("reconciles a crashed escalated node to a timeout failure, pauses the workflow, and wakes the parent", async () => { + await Effect.runPromise( + runRecoveryTest(({ dag, loop, store, wakes }) => + Effect.gen(function* () { + // Simulate a process crash AFTER the node escalated once: the node + // row is running, timeout_extensions=1, its deadline passed while it + // was executing, and the child session is gone. Everything is seeded + // before DagLoop.init so the startup recovery scan is the actor. + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Escalated crash", + config: { name: "escalated-crash", nodes: [node("a", 60_000)] }, + }) + yield* dag.nodeQueued(dagID, "a", Date.now() - 1000) + yield* dag.nodeStarted(dagID, "a", "ses_child_crashed", Date.now() - 1000, true) + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_crashed", 1) + + yield* loop.init() + + // S2: the durable escalation counter proves the timeout semantics — + // the recovery must fail the node as timeout (not ownership loss), + // preserve the extension count, and pause the workflow instead of + // letting the scheduler cascade terminalize on invented evidence. + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("failed") + expect(row?.errorClass).toBe("timeout") + expect(row?.errorReason).toContain("timeout escalated (1 extension(s)) node failed on recovery") + expect(row?.timeoutExtensions).toBe(1) + expect((yield* store.getWorkflow(dagID))?.status).toBe("paused") + + // The invented-failure wake reaches the parent at the paused + // delivery boundary with the timeout attribution. + const wake = yield* pollWithTimeout( + Effect.sync(() => (wakes.length > 0 ? wakes[0] : undefined)), + "recovery wake did not reach the parent", + ) + expect(wake).toContain('Node "a" failed (timeout)') + expect(wake).toContain("timeout escalated (1 extension(s)) node failed on recovery") + }), + ), + ) + }) + + it("reports ownership loss for a crashed escalated node whose deadline never passed (S2 future deadline)", async () => { + await Effect.runPromise( + runRecoveryTest(({ dag, loop, store }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Escalated crash future", + config: { name: "escalated-crash-future", nodes: [node("a", 60_000)] }, + }) + yield* dag.nodeQueued(dagID, "a", Date.now() + 60_000) + yield* dag.nodeStarted(dagID, "a", "ses_child_crashed", Date.now() + 60_000, true) + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_crashed", 1) + + yield* loop.init() + + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("failed") + expect(row?.errorClass).toBe("exec_failed") + expect(row?.errorReason).toContain("execution ownership lost on recovery") + expect((yield* store.getWorkflow(dagID))?.status).toBe("paused") + }), + ), + ) + }) +}) diff --git a/packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts b/packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts index 00ac0c7086..f19067b952 100644 --- a/packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts +++ b/packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts @@ -300,7 +300,14 @@ describe("DagLoop replan vs stale NodeFailed", () => { projectID: "project-1", sessionID: "ses_parent", title: "Genuine failure", - config: { name: "genuine-failure", nodes: [node("a", [], 300), node("b", ["a"])] }, + config: { + name: "genuine-failure", + // Extension cap 0: the first deadline exhausts the cap and the + // watcher force-cancels + fails the node (timeout = signal until + // the cap runs out — this test exercises the cap-exhausted path). + max_timeout_extensions: 0, + nodes: [node("a", [], 300), node("b", ["a"])], + }, }) const gate = yield* takeWithin(childPrompts, "a did not start") expect(gate.title).toBe("a") @@ -372,6 +379,60 @@ describe("DagLoop replan vs stale NodeFailed", () => { ) }) + it("rebuilds the graph from a restarted node's new depends_on (restart + rewired deps)", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Restart rewire", + config: { name: "restart-rewire", nodes: [node("a"), node("b", ["a"])] }, + }) + const gateA = yield* takeWithin(childPrompts, "a did not start") + expect(gateA.title).toBe("a") + yield* Deferred.succeed(gateA.release, "done") + const gateB = yield* takeWithin(childPrompts, "b did not start") + expect(gateB.title).toBe("b") + + // Restart b mid-flight, rewiring its dependency from a → c (new node). + const plan = yield* dag.replan(dagID, { + nodes: [ + { ...node("b", ["c"]), restart: true }, + node("c"), + ], + }) + expect(plan.restart).toEqual(["b"]) + expect(plan.add).toEqual(["c"]) + + // New graph: c is b's only dependency. If the stale b→a edge + // survived the rebuild, b (a already completed) would be re-ready + // immediately and its prompt would arrive before c's — the take + // below would then fail with the wrong title. + const gateC = yield* takeWithin(childPrompts, "c did not start first under the rewired graph") + expect(gateC.title).toBe("c") + const bRow = yield* store.getNode(dagID, "b") + expect(bRow?.status).toBe("pending") + expect(bRow?.dependsOn).toEqual(["c"]) + yield* Deferred.succeed(gateC.release, "done") + + const gateB2 = yield* takeWithin(childPrompts, "b was not rescheduled after its new dependency completed") + expect(gateB2.title).toBe("b") + yield* Deferred.succeed(gateB2.release, "done") + + yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((workflow) => workflow?.status === "completed" ? workflow : undefined), + ), + "workflow did not complete under the rewired graph", + ) + const parent = yield* takeWithin(parentPrompts, "terminal wake did not reach the parent") + yield* Deferred.succeed(parent.release, "success") + }), + ), + ) + }) + it("keeps a mid-flight replan restart schedulable when the node is immediately ready again", async () => { await Effect.runPromise( runLoopTest(({ dag, store, childPrompts, parentPrompts }) => diff --git a/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts b/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts index 595f03a57a..b26d0af2e3 100644 --- a/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts +++ b/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts @@ -68,6 +68,7 @@ function summary(id: string, completedNodes: number): WorkflowSummary { failedNodes: 0, skippedNodes: 0, queuedNodes: 0, + escalatedNodes: 0, } } @@ -152,6 +153,29 @@ function publishNodeEvents( }) } +function publishTimeoutEscalation( + bus: EventControl, + dagID: string, + nodeID: string, + extensions: number, +) { + if (!bus.listener) return Effect.die(new Error("publisher listener is not ready")) + return Effect.gen(function* () { + const instance = yield* InstanceState.context + yield* bus.listener!({ + type: DagEvent.NodeTimeoutEscalated.type, + data: { + dagID, + nodeID, + childSessionID: `ses-${nodeID}`, + timeoutExtensions: extensions, + timestamp: ts(extensions), + }, + location: { directory: instance.directory }, + } as never) + }) +} + function withCollector(use: (collector: ReturnType) => Effect.Effect) { return Effect.acquireUseRelease( Effect.sync(startCollector), @@ -325,4 +349,33 @@ describe("DagSummaryPublisher behavior", () => { }), ).pipe(Effect.provide(runtime(state, bus))) }) + + it.instance("a timeout escalation triggers a fresh summary recompute (F10)", () => { + const state = control() + const bus = {} satisfies EventControl + state.sessions.set("dag-escalated", "ses-escalated") + state.summaries.set("ses-escalated", [{ + ...summary("dag-escalated", 0), + runningNodes: 1, + escalatedNodes: 1, + }]) + + return withCollector((collector) => + Effect.gen(function* () { + yield* (yield* DagSummaryPublisher.Service).init() + // Escalation is a pure counter change on a running node — no status + // transition — so the publisher must still re-emit or the TUI would + // keep showing a plain RUNNING node (F10). + yield* publishTimeoutEscalation(bus, "dag-escalated", "dag-escalated-a", 1) + yield* pollWithTimeout( + Effect.sync(() => collector.emissions.length === 1 ? collector.emissions[0] : undefined), + "escalation did not trigger a summary emission", + ) + + expect(state.reads.get("ses-escalated")).toBe(1) + expect(collector.emissions[0].summaries[0].escalatedNodes).toBe(1) + expect(collector.emissions[0].summaries[0].runningNodes).toBe(1) + }), + ).pipe(Effect.provide(runtime(state, bus))) + }) }) diff --git a/packages/opencode/test/dag/dag-summary-publisher.test.ts b/packages/opencode/test/dag/dag-summary-publisher.test.ts index 19283e55e4..101d3ccc34 100644 --- a/packages/opencode/test/dag/dag-summary-publisher.test.ts +++ b/packages/opencode/test/dag/dag-summary-publisher.test.ts @@ -23,9 +23,10 @@ describe("DagSummaryPublisher contract (stateless derived view)", () => { failedNodes: 0, skippedNodes: 0, queuedNodes: 0, + escalatedNodes: 0, } // If this compiles, the shape is correct. The keys must match the TUI type. - const keys: (keyof WorkflowSummary)[] = ["id", "title", "status", "nodeCount", "completedNodes", "runningNodes", "failedNodes", "skippedNodes", "queuedNodes"] + const keys: (keyof WorkflowSummary)[] = ["id", "title", "status", "nodeCount", "completedNodes", "runningNodes", "failedNodes", "skippedNodes", "queuedNodes", "escalatedNodes"] expect(Object.keys(s).sort()).toEqual([...keys].sort()) }) diff --git a/packages/opencode/test/dag/dag-timeout-escalation-fixes.test.ts b/packages/opencode/test/dag/dag-timeout-escalation-fixes.test.ts new file mode 100644 index 0000000000..2182ab10a0 --- /dev/null +++ b/packages/opencode/test/dag/dag-timeout-escalation-fixes.test.ts @@ -0,0 +1,400 @@ +import { describe, expect, it } from "bun:test" +import { Cause, Effect, Exit, Fiber, Layer, Scope } from "effect" +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 { planReplan } from "@opencode-ai/core/dag/core/replan" +import { NodeStatus } from "@opencode-ai/core/dag/core/types" +import { EventV2 } from "@opencode-ai/core/event" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { Dag, type NodeConfig } from "@/dag/dag" +import { EventV2Bridge } from "@/event-v2-bridge" +import { InstanceRef } from "@/effect/instance-ref" +import { reconcileWorkflow } from "@/dag/runtime/recovery" +import { makeDeadlineWatcher } from "@/dag/runtime/spawn" +import { SessionPrompt } from "@/session/prompt" +import { makeNodeRow } from "./fixtures" +import { awaitWithTimeout, pollWithTimeout } from "../lib/effect" + +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 }) => Effect.Effect, +) { + return Effect.gen(function* () { + 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, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.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) + const dag = yield* Dag.Service + const store = yield* DagStore.Service + return yield* test({ dag, store }) + }).pipe( + Effect.provide(harness), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: "project-1" }, + } as never), + Effect.scoped, + ) + }) +} + +function createWorkflow(dag: Dag.Interface, title: string, timeoutMs?: number, nodeID = "a") { + return dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title, + config: { name: title, nodes: [node(nodeID, timeoutMs)] }, + }) +} + +describe("Dag timeout escalation fixes (unit)", () => { + it("rejects a cycle introduced through a running node's replaced deps (P1a)", () => { + // Both nodes are running and appear in the fragment WITHOUT a restart + // marker → they land in the replace bucket: the fragment's deps are + // re-published via NodeRegistered and the runtime rebuilds its graph from + // them. The replan's cycle check must use the SAME deps (a→b, b→a = cycle). + const plan = planReplan( + { nodes: [ + { id: "a", status: NodeStatus.RUNNING, depends_on: [] }, + { id: "b", status: NodeStatus.RUNNING, depends_on: ["a"] }, + ] }, + { nodes: [ + { id: "a", depends_on: ["b"] }, + { id: "b", depends_on: ["a"] }, + ] }, + ) + expect(plan.errors.join(" ")).toContain("cycle") + }) + + it("rejects a replan whose running-node fragment deps form a cycle, leaving the node untouched (P1a)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "P1a replan cycle", + config: { name: "p1a", nodes: [node("a", 60_000)] }, + }) + yield* dag.nodeQueued(dagID, "a", Date.now() + 60_000) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", Date.now() + 60_000, false) + + // The running node "a" is present in the fragment without a restart + // marker (replace bucket): its NEW deps (a→b) are what the runtime + // would execute, so the cycle a↔b must be caught BEFORE any event is + // published — the replan is rejected, the running node is untouched. + const exit = yield* dag.replan(dagID, { + nodes: [ + { ...node("a", 60_000), depends_on: ["b"] }, + { ...node("b", 60_000), depends_on: ["a"] }, + ], + }).pipe(Effect.exit) + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + expect(Cause.pretty(exit.cause)).toContain("cycle") + } + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("running") + expect(row?.timeoutExtensions).toBe(0) + expect(row?.childSessionId).toBe("ses_child_1") + }), + ), + ) + }) + + it("includes an escalated running node in the wake snapshot regardless of report_to_parent (F11)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "f11-visible") + yield* dag.nodeQueued(dagID, "a", Date.now() - 1000) + // reportToParent=false (wake_eligible=false) — the default for most + // nodes. The escalation must still reach the main agent. + yield* dag.nodeStarted(dagID, "a", "ses_child_1", Date.now() - 1000, false) + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 1) + + const snapshot = yield* store.getWakeSnapshot("ses_parent") + const node = snapshot.nodes.find((candidate) => candidate.id === "a") + expect(node).toBeTruthy() + expect(node?.wakeEligible).toBe(false) + expect(node?.status).toBe("running") + expect(node?.timeoutExtensions).toBe(1) + + const unreported = yield* store.getUnreportedWakeNodes("ses_parent") + expect(unreported.map((candidate) => candidate.id)).toContain("a") + expect(yield* store.getSessionsWithUnreportedWakes()).toContain("ses_parent") + }), + ), + ) + }) + + it("keeps an escalated-then-failed node visible in the wake snapshot (F11 cap verdict)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "f11-terminal") + yield* dag.nodeQueued(dagID, "a", Date.now() - 1000) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", Date.now() - 1000, false) + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 1) + // Cap-exhausted force-cancel terminalizes the escalated node. Its + // verdict (failed, extension count preserved) must re-enter the + // snapshot even though wake_eligible=false. + yield* dag.nodeFailed(dagID, "a", "timeout extensions exhausted (1/1)", "timeout") + + const snapshot = yield* store.getWakeSnapshot("ses_parent") + const node = snapshot.nodes.find((candidate) => candidate.id === "a") + expect(node).toBeTruthy() + expect(node?.wakeEligible).toBe(false) + expect(node?.status).toBe("failed") + expect(node?.timeoutExtensions).toBe(1) + expect(node?.errorReason).toContain("timeout extensions exhausted") + + const unreported = yield* store.getUnreportedWakeNodes("ses_parent") + expect(unreported.map((candidate) => candidate.id)).toContain("a") + }), + ), + ) + }) + + it("clamps a zero timeout_ms to the floor on create (F9)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "zero-timeout", 0) + const wf = yield* store.getWorkflow(dagID) + const config = JSON.parse(wf!.config) as { nodes: NodeConfig[] } + expect(config.nodes[0].worker_config?.timeout_ms).toBe(1000) + }), + ), + ) + }) + + it("clamps a negative timeout_ms to the floor on create (F9)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "negative-timeout", -5) + const wf = yield* store.getWorkflow(dagID) + const config = JSON.parse(wf!.config) as { nodes: NodeConfig[] } + expect(config.nodes[0].worker_config?.timeout_ms).toBe(1000) + }), + ), + ) + }) + + it("ignores a timeout escalation landing on a terminal node (F2a ghost wake)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "ghost-wake", 60_000) + yield* dag.nodeQueued(dagID, "a", Date.now() + 60_000) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", Date.now() + 60_000, false) + // The node fails before any timeout can fire. + yield* dag.nodeFailed(dagID, "a", "provider exploded", "exec_failed") + // A stale escalation races in after the terminal event. + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 1).pipe(Effect.ignore) + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("failed") + expect(row?.timeoutExtensions).toBe(0) + }), + ), + ) + }) + + it("ignores a stale escalation landing on a completed node (F2a completed race)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "ghost-wake-completed", 60_000) + yield* dag.nodeQueued(dagID, "a", Date.now() + 60_000) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", Date.now() + 60_000, true) + // The child finishes first; the completion wins the race. + yield* dag.nodeCompleted(dagID, "a", "done") + // A stale escalation from the watcher fiber races in afterwards. + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 1).pipe(Effect.ignore) + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("completed") + // Neither the extension counter nor the re-armed wake flag may be + // touched on the terminal row — the escalation guard rejects the + // UPDATE entirely (0 rows), so the completion wake stays exactly as + // NodeCompleted left it. + expect(row?.timeoutExtensions).toBe(0) + expect(row?.wakeReported).toBe(false) + }), + ), + ) + }) + + it("keeps an escalation counter on an escalated node, then marks its recovery failure as timeout (S2)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "recovery-escalated", 60_000) + yield* dag.nodeQueued(dagID, "a", Date.now() - 1000) + yield* dag.nodeStarted(dagID, "a", "ses_child_1", Date.now() - 1000, true) + yield* dag.nodeTimeoutEscalated(dagID, "a", "ses_child_1", 2) + const pre = yield* store.getNode(dagID, "a") + expect(pre?.status).toBe("running") + expect(pre?.timeoutExtensions).toBe(2) + + // Crash recovery: the child session is gone ("unknown"), the deadline + // was already exceeded and the durable counter proves the escalation. + const result = yield* reconcileWorkflow( + dagID, + () => Effect.succeed("unknown" as const), + () => Effect.void, + undefined, + ).pipe(Effect.provideService(Dag.Service, dag)) + expect(result.reconciled).toBe(1) + expect(result.ownershipLost).toBe(1) + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("failed") + expect(row?.errorClass).toBe("timeout") + expect(row?.errorReason).toContain("timeout escalated (2 extension(s))") + }), + ), + ) + }) + + it("marks a recovery failure of an escalated node that never passed its deadline as ownership loss (S2)", async () => { + await Effect.runPromise( + runTest(({ dag, store }) => + Effect.gen(function* () { + const dagID = yield* createWorkflow(dag, "recovery-escalated-future", 60_000) + 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 result = yield* reconcileWorkflow( + dagID, + () => Effect.succeed("unknown" as const), + () => Effect.void, + undefined, + ).pipe(Effect.provideService(Dag.Service, dag)) + expect(result.ownershipLost).toBe(1) + const row = yield* store.getNode(dagID, "a") + expect(row?.status).toBe("failed") + expect(row?.errorClass).toBe("exec_failed") + expect(row?.errorReason).toContain("execution ownership lost on recovery") + }), + ), + ) + }) + + it("retries transient store read failures instead of exiting supervision (R13)", async () => { + let reads = 0 + let escalations = 0 + const dagLayer = Layer.mock(Dag.Service, { + store: { + getNode: () => + Effect.sync(() => { + reads++ + // 3 transient failures (e.g. SQLite lock blips) — the watcher + // must survive them and keep supervising. + if (reads <= 3) throw new Error("database locked") + return makeNodeRow({ + id: "a", + workflowId: "dag-r13", + name: "a", + status: "running", + deadlineMs: 1, + timeoutExtensions: 0, + childSessionId: "ses_child_1", + }) + }), + } as unknown as DagStore.Interface, + nodeTimeoutEscalated: () => Effect.sync(() => { escalations++ }), + }) + const promptLayer = Layer.mock(SessionPrompt.Service, { + cancel: () => Effect.void, + }) + await Effect.runPromise( + Effect.gen(function* () { + const scope = yield* Scope.Scope + const watcher = yield* makeDeadlineWatcher({ dagID: "dag-r13", nodeID: "a", timeoutMs: 300 }).pipe( + Effect.forkIn(scope), + ) + // The watcher escalates once the row is readable — proof the transient + // failures did not end supervision (1 initial read + 3 retries). + yield* pollWithTimeout( + Effect.sync(() => (escalations > 0 ? true : undefined)), + "watcher did not escalate after transient store failures (R13 regression)", + ) + expect(reads).toBe(4) + yield* Fiber.interrupt(watcher).pipe(Effect.ignore) + }).pipe( + Effect.provide(dagLayer), + Effect.provide(promptLayer), + Effect.scoped, + ), + ) + }) + + it("continues supervision after store read retries fail (R13/F1-product)", async () => { + let reads = 0 + const dagLayer = Layer.mock(Dag.Service, { + store: { + getNode: () => + Effect.sync(() => { + reads++ + throw new Error("database locked") + }), + } as unknown as DagStore.Interface, + }) + const promptLayer = Layer.mock(SessionPrompt.Service, { + cancel: () => Effect.void, + }) + await Effect.runPromise( + Effect.gen(function* () { + const scope = yield* Scope.Scope + const watcher = yield* makeDeadlineWatcher({ dagID: "dag-r13", nodeID: "a", timeoutMs: 300 }).pipe( + Effect.forkIn(scope), + ) + // After 4 failed reads (1 + 3 retries), the watcher does NOT exit — + // it sleeps 5s then retries. Verify reads > 4 after enough time for + // at least 2 cycles, then interrupt. + yield* Effect.sleep("8 seconds") + yield* Fiber.interrupt(watcher) + expect(reads).toBeGreaterThan(4) + }).pipe( + Effect.provide(dagLayer), + Effect.provide(promptLayer), + Effect.scoped, + ), + ) + }, 15000) +}) diff --git a/packages/opencode/test/dag/dag-timeout-escalation.test.ts b/packages/opencode/test/dag/dag-timeout-escalation.test.ts new file mode 100644 index 0000000000..560a196be2 --- /dev/null +++ b/packages/opencode/test/dag/dag-timeout-escalation.test.ts @@ -0,0 +1,1005 @@ +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 { 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" + +interface PromptGate { + readonly title: string + readonly release: Deferred.Deferred +} + +interface ParentPromptGate { + readonly text: string + readonly release: Deferred.Deferred<"success" | "failure"> +} + +function takeWithin(queue: Queue.Queue, message: string) { + return Queue.take(queue).pipe( + Effect.timeoutOption("3 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(), + role: "assistant", + parentID: MessageID.ascending(), + sessionID: sessionID as never, + 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, + time: { created: Date.now() }, + finish: "stop", + }, + parts: text ? [{ type: "text", text }] as never : [], + } +} + +function node(id: string, dependsOn: string[] = [], timeoutMs?: number): NodeConfig { + return { + id, + name: id, + worker_type: "build", + depends_on: dependsOn, + required: true, + prompt_template: { inline: id }, + report_to_parent: true, + ...(timeoutMs ? { worker_config: { timeout_ms: timeoutMs } } : {}), + } +} + +function loopLayer(input: { + readonly childPrompts: Queue.Queue + readonly parentPrompts: Queue.Queue +}, opts?: { + readonly nodeExtendTimeout?: (dagID: string, nodeID: string, newDeadlineMs: number) => Effect.Effect + readonly nodeTimeoutEscalated?: (dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number) => Effect.Effect +}) { + 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 realDag = Dag.layer.pipe( + Layer.provide(bridge), + Layer.provide(store), + ) + // N1-style fault injection: wrap the real Dag service and break selected + // methods (everything else delegates), so a test can prove the loop and the + // deadline watcher survive a failed durable write. + const overrides = { + ...(opts?.nodeExtendTimeout ? { nodeExtendTimeout: opts.nodeExtendTimeout } : {}), + ...(opts?.nodeTimeoutEscalated ? { nodeTimeoutEscalated: opts.nodeTimeoutEscalated } : {}), + } + const dag = Object.keys(overrides).length > 0 + ? Layer.effect( + Dag.Service, + Effect.gen(function* () { + const real = yield* Dag.Service + return { ...real, ...overrides } + }), + ).pipe(Layer.provide(realDag)) + : realDag + const base = Layer.mergeAll(database, events, bridge, store, projector, dag, status) + const childTitles = new Map() + const created: string[] = [] + let cancelCount = 0 + 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 text = value.parts.find((p) => p.type === "text")?.text ?? "" + if (sessionID === "ses_parent") { + const release = yield* Deferred.make<"success" | "failure">() + yield* Queue.offer(input.parentPrompts, { text, release }) + const outcome = yield* Deferred.await(release) + if (outcome === "failure") return yield* Effect.die(new Error("provider unavailable")) + return reply(sessionID, "parent handled wake") + } + 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, { + cancel: () => Effect.sync(() => { cancelCount++ }), + prompt: deliver, + promptIfIdle: (value) => 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), + ) + return { layer: Layer.merge(base, loop), getCancelCount: () => cancelCount } +} + +function runLoopTest( + test: (services: { + readonly dag: Dag.Interface + readonly store: DagStore.Interface + readonly childPrompts: Queue.Queue + readonly parentPrompts: Queue.Queue + readonly getCancelCount: () => number + }) => Effect.Effect, + opts?: { + readonly nodeExtendTimeout?: (dagID: string, nodeID: string, newDeadlineMs: number) => Effect.Effect + readonly nodeTimeoutEscalated?: (dagID: string, nodeID: string, childSessionID: string, timeoutExtensions: number) => Effect.Effect + }, +) { + return Effect.gen(function* () { + const childPrompts = yield* Queue.unbounded() + const parentPrompts = yield* Queue.unbounded() + const harness = loopLayer({ childPrompts, parentPrompts }, opts) + return yield* Effect.gen(function* () { + const dag = yield* Dag.Service + const loop = yield* DagLoop.Service + 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, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.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* loop.init() + return yield* test({ + dag, + store, + childPrompts, + parentPrompts, + getCancelCount: harness.getCancelCount, + }) + }).pipe( + Effect.provide(harness.layer), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: "project-1" }, + } as never), + Effect.scoped, + ) + }) +} + +describe("DagLoop timeout escalation", () => { + it("escalates on execution timeout without cancelling the child session, and wakes the main agent", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts, getCancelCount }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Timeout escalation", + config: { name: "escalation", nodes: [node("a", [], 300)] }, + }) + const gate = yield* takeWithin(childPrompts, "a did not start") + expect(gate.title).toBe("a") + + // Never release the child prompt — the deadline elapses. The child + // session must NOT be cancelled; the node stays RUNNING with a + // persisted extension count. + const escalated = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 1 && current.status === "running" ? current : undefined, + ), + ), + "node did not escalate on timeout", + ) + expect(escalated.timeoutExtensions).toBe(1) + expect(escalated.status).toBe("running") + expect(escalated.childSessionId).toBeTruthy() + expect(getCancelCount()).toBe(0) + + // The main agent receives a timeout wake with the node identifier. + const parent = yield* takeWithin(parentPrompts, "timeout wake did not reach the parent") + expect(parent.text).toContain("[DAG Node Timeout]") + expect(parent.text).toContain('"a"') + yield* Deferred.succeed(parent.release, "success") + + // The child session is still alive — it can still finish the work. + yield* Deferred.succeed(gate.release, "done") + const completed = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.status === "completed" ? current : undefined), + ), + "node did not complete after the escalation", + ) + expect(completed.status).toBe("completed") + expect(getCancelCount()).toBe(0) + }), + ), + ) + }) + + it("extends the deadline via replan with a new timeout_ms and escalates again on the next deadline", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts, getCancelCount }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Timeout extension", + config: { name: "extension", nodes: [node("a", [], 300)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + + const first = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 1 && current.status === "running" ? current : undefined, + ), + ), + "first escalation did not fire", + ) + const firstWake = yield* takeWithin(parentPrompts, "first wake did not reach the parent") + yield* Deferred.succeed(firstWake.release, "success") + + // Main agent adjudicates: extend by replanning with a new timeout. + // No restart marker — the running node keeps its execution. + const plan = yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 2000) }] }) + expect(plan.replace).toContain("a") + const extended = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.deadlineMs != null && current.deadlineMs > (first.deadlineMs ?? 0) + ? current + : undefined, + ), + ), + "deadline was not extended by the replan", + ) + expect(extended.status).toBe("running") + // Cumulative cap: extend does NOT reset timeout_extensions. + // The count persists across replan-extends; only a new attempt + // (NodeStarted/NodeRestarted) resets it. This prevents an agent + // from bypassing the cap by repeatedly replanning. + expect(extended.timeoutExtensions).toBe(1) + + // The rebuilt watcher fires again once the new deadline elapses; + // the count climbs to 2 (cumulative, not reset). + const second = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === 2 ? current : undefined), + ), + "second escalation did not fire after the extension", + ) + expect(second.status).toBe("running") + expect(getCancelCount()).toBe(0) + const secondWake = yield* takeWithin(parentPrompts, "second wake did not reach the parent") + expect(secondWake.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(secondWake.release, "success") + }), + ), + ) + }) + + it("N1: a died nodeExtendTimeout leaves the node under supervision — the watcher escalates again", async () => { + let extendCalls = 0 + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "N1 supervision survives failed extend", + config: { name: "n1-failed-extend", nodes: [node("a", [], 300)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + + // Deadline elapses → escalation #1 and a timeout wake. + const first = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 1 && current.status === "running" ? current : undefined, + ), + ), + "first escalation did not fire", + ) + const firstWake = yield* takeWithin(parentPrompts, "first wake did not reach the parent") + expect(firstWake.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(firstWake.release, "success") + + // Adjudicate while the extend write is broken: the new timeout_ms + // takes the re-time path (§3.7) and escalation_pending opens the cap + // gate, but nodeExtendTimeout dies and guarded("WorkflowReplanned") + // swallows the defect. + yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 2000) }] }) + + // The re-time path did run against the broken write... + yield* pollWithTimeout( + Effect.sync(() => (extendCalls > 0 ? true : undefined)), + "replan never attempted nodeExtendTimeout", + ) + // ...and the deadline never moved (the write died). + const afterReplan = yield* store.getNode(dagID, "a") + expect(afterReplan?.deadlineMs).toBe(first.deadlineMs) + + // Supervision intact: the watcher the failed re-time left in place + // escalates again on the stale deadline. Pre-fix order (interrupt the + // watcher BEFORE the write) left the node with no watcher here and + // timeoutExtensions stuck at 1 forever — the cap backstop (§5-5) + // defeated by a failed write. + const second = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 2 && current.status === "running" ? current : undefined, + ), + ), + "watcher died with the failed extend — node escaped supervision", + ) + expect(second.timeoutExtensions).toBe(2) + const secondWake = yield* takeWithin(parentPrompts, "second wake did not reach the parent") + yield* Deferred.succeed(secondWake.release, "success") + }), + { + nodeExtendTimeout: () => + Effect.sync(() => { + extendCalls++ + }).pipe(Effect.flatMap(() => Effect.die(new Error("simulated nodeExtendTimeout defect (N1 test)")))), + }, + ), + ) + }) + + it("D1: a failed nodeExtendTimeout does not abort the replan batch — a restarted node still gets scheduled", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "D1 batch survives failed extend", + config: { name: "d1-batch", nodes: [node("a", [], 300), node("b", [], 5000)] }, + }) + const first = yield* takeWithin(childPrompts, "a did not start") + const second = yield* takeWithin(childPrompts, "b did not start") + expect([first.title, second.title].sort()).toEqual(["a", "b"]) + + // a escalates on its 300ms deadline; b keeps running (prompt never + // released, long timeout so it cannot interfere). + yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 1 && current.status === "running" ? current : undefined, + ), + ), + "first escalation did not fire", + ) + const firstWake = yield* takeWithin(parentPrompts, "first wake did not reach the parent") + yield* Deferred.succeed(firstWake.release, "success") + + // One replan, two intents: a carries a NEW timeout_ms so it takes the + // re-time path (which dies); b carries a restart marker, resetting it + // to pending — only the handler's spawnReady reschedules the new + // attempt. Pre-D1 the dying extend propagated to guarded() and skipped + // spawnReady entirely, leaving b pending forever (half-applied replan). + yield* dag.replan(dagID, { + nodes: [{ ...node("a", [], 2000) }, { ...node("b", [], 5000), restart: true }], + }) + + const restartedB = yield* takeWithin( + childPrompts, + "restarted node was never re-spawned — the failed extend aborted the handler before spawnReady", + ) + expect(restartedB.title).toBe("b") + }), + { + nodeExtendTimeout: () => Effect.die(new Error("simulated nodeExtendTimeout defect (D1 test)")), + }, + ), + ) + }) + + it("a failed escalate write does not end supervision — the watcher retries it", async () => { + let escalateCalls = 0 + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "watcher survives a failed escalate write", + config: { name: "escalate-write-failure", nodes: [node("a", [], 300)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + + // Every escalate write dies. The watcher's catchCause used to sit + // OUTSIDE its for(;;) loop, so the first failed write ended the fiber: + // no further escalation, timeout_extensions frozen at 0, and the §5-5 + // cap backstop could never fire (the node would hold a concurrency + // slot unbounded). A second call proves supervision outlived the + // failure — the read path was already hardened this way (R13), the + // write path was not. + yield* pollWithTimeout( + Effect.sync(() => (escalateCalls >= 2 ? escalateCalls : undefined)), + "watcher never retried the escalate write — the failed write ended supervision", + ) + const current = yield* store.getNode(dagID, "a") + expect(current?.status).toBe("running") + expect(current?.timeoutExtensions).toBe(0) + }), + { + nodeTimeoutEscalated: () => + Effect.sync(() => { + escalateCalls++ + }).pipe(Effect.flatMap(() => Effect.die(new Error("simulated nodeTimeoutEscalated defect")))), + }, + ), + ) + }) + + it("re-times NO running survivor whose deadline is healthy or timeout unchanged (§3.7 + cap gate)", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Merged semantics", + config: { + name: "merged", + max_concurrency: 2, + nodes: [node("a", [], 60_000), node("b", [], 300)], + }, + }) + // b registers last → spawns first; a takes the second slot. Neither + // child is ever released. + const gates = [ + yield* takeWithin(childPrompts, "first node did not start"), + yield* takeWithin(childPrompts, "second node did not start"), + ] + expect(gates.map((gate) => gate.title).sort()).toEqual(["a", "b"]) + + // b escalates at its own 300ms deadline. + const firstB = yield* pollWithTimeout( + store.getNode(dagID, "b").pipe( + Effect.map((current) => + current?.timeoutExtensions === 1 && current.status === "running" ? current : undefined, + ), + ), + "b did not escalate", + ) + const wake1 = yield* takeWithin(parentPrompts, "first wake did not reach the parent") + expect(wake1.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(wake1.release, "success") + + const aBefore = yield* store.getNode(dagID, "a") + + // Replan mentions ONLY a with a new timeout — b is absent from the + // fragment. NEITHER survivor is re-timed: §3.7 skips b (timeout + // unchanged), and the cap gate skips a (timeout changed, but its + // deadline is still in the future with no pending escalation). b's + // self-renewing watcher keeps escalating the elapsed deadline toward + // the cap. + const plan = yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 5000) }] }) + expect(plan.replace).toContain("a") + expect(plan.replace).not.toContain("b") + + // b re-escalates one escalation interval later — well after the + // WorkflowReplanned handler processed the fragment — with its + // deadline frozen at the original value. + const secondB = yield* pollWithTimeout( + store.getNode(dagID, "b").pipe( + Effect.map((current) => + current?.timeoutExtensions === 2 && current.deadlineMs === firstB.deadlineMs + ? current + : undefined, + ), + ), + "self-renewing watcher did not re-escalate the unmentioned node", + ) + expect(secondB.status).toBe("running") + + // a: the cap gate kept the healthy deadline frozen as well. + const aAfter = yield* store.getNode(dagID, "a") + expect(aAfter?.deadlineMs).toBe(aBefore?.deadlineMs) + + const wake2 = yield* takeWithin(parentPrompts, "second wake did not reach the parent") + expect(wake2.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(wake2.release, "success") + }), + ), + ) + }) + + it("refuses a pre-escalation re-time (A1: proactive re-time cannot bypass the cap) but admits it once escalated", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "A1 cap gate", + config: { name: "a1-cap-gate", nodes: [node("a", [], 500)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + const started = yield* store.getNode(dagID, "a") + expect(started?.deadlineMs).not.toBeNull() + + // The agent extends PRE-EMPTIVELY before the deadline passes, + // changing the timeout value. Without the cap gate this moved the + // deadline to now+timeout and the node never escalated — an agent + // cycling timeout values could push the deadline forward forever, + // the extension count never climbed, and the ≈21× cap was bypassed. + yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 5000) }] }) + + // The deadline must stay frozen at its original value and the node + // must escalate there: extensions reaches 1 with the deadline + // unchanged. If the re-time had fired, the deadline would be + // now+5000 and this condition could never match. + const escalated = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 1 && current.deadlineMs === started?.deadlineMs + ? current + : undefined, + ), + ), + "pre-escalation re-time moved the deadline (A1: the cap is bypassable)", + ) + expect(escalated.status).toBe("running") + const wake = yield* takeWithin(parentPrompts, "escalation wake did not reach the parent") + expect(wake.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(wake.release, "success") + + // After the escalation the same extend IS admitted (deadline elapsed + // + pending escalation), and the cumulative count survives it. + yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 2000) }] }) + const extended = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.deadlineMs != null && current.deadlineMs > (started?.deadlineMs ?? 0) + ? current + : undefined, + ), + ), + "post-escalation re-time was wrongly gated off", + ) + expect(extended.timeoutExtensions).toBe(1) + expect(extended.status).toBe("running") + }), + ), + ) + }, 60_000) + + it("force-cancels and fails the node when the extension cap is exhausted", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts, getCancelCount }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Timeout cap exhausted", + config: { name: "cap", max_timeout_extensions: 0, nodes: [node("a", [], 300)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + + // With the cap at 0 the very first deadline forces a cancel+fail. + const failed = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.status === "failed" ? current : undefined), + ), + "node did not fail after the extension cap was exhausted", + ) + expect(failed.errorClass).toBe("timeout") + expect(failed.errorReason).toContain("timeout extensions exhausted") + expect(failed.timeoutExtensions).toBe(0) + // The watcher force-cancels the child; the NodeFailed handler and + // workflow terminalization then re-cancel the same (already dead) + // session — what matters is that the child was killed. + expect(getCancelCount()).toBeGreaterThanOrEqual(1) + + // Required-node failure cascades into a workflow failure. + const workflow = yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((current) => current?.status === "failed" ? current : undefined), + ), + "workflow did not fail after the required node was force-failed", + ) + expect(workflow.status).toBe("failed") + const parent = yield* takeWithin(parentPrompts, "failure wake did not reach the parent") + yield* Deferred.succeed(parent.release, "success") + }), + ), + ) + }) + + it("keeps the pre-permit queue-wait timeout as a direct nodeFailed (F4: queued admission deadline is fixed)", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + // Spawn order follows the node rows' desc(seq) read, so the LAST + // registered node spawns FIRST. "a" must hold the only permit, so it + // must be registered after "b" — "b" then waits in the queue. + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Pre-permit timeout", + config: { + name: "pre-permit", + max_concurrency: 1, + nodes: [node("b", [], 2000), node("a", [], 300)], + }, + }) + // a holds the only permit and is never released. + yield* takeWithin(childPrompts, "a did not start") + const queuedB = yield* pollWithTimeout( + store.getNode(dagID, "b").pipe( + Effect.map((current) => current?.status === "queued" ? current : undefined), + ), + "b was not queued", + ) + + // a escalates at its own deadline — the RUNNING node's timeout + // signal fires and wakes the main agent. + const escalated = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 1 && current.status === "running" ? current : undefined, + ), + ), + "a did not escalate", + ) + const wake = yield* takeWithin(parentPrompts, "timeout wake did not reach the parent") + expect(wake.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(wake.release, "success") + + // F4: the running node's escalation/adjudication does NOT adjust the + // queued node's admission deadline — it stays exactly as fixed at + // admission (P0-2: queue wait counts toward the budget). + const stillQueuedB = yield* store.getNode(dagID, "b") + expect(stillQueuedB?.status).toBe("queued") + expect(stillQueuedB?.deadlineMs).toBe(queuedB.deadlineMs) + expect(escalated.timeoutExtensions).toBe(1) + + // b waits for the permit past its own deadline — the queue-wait + // timeout still hard-fails with no progress to protect. + const failedB = yield* pollWithTimeout( + store.getNode(dagID, "b").pipe( + Effect.map((current) => current?.status === "failed" ? current : undefined), + ), + "b did not fail on the queue-wait timeout", + ) + expect(failedB.errorClass).toBe("timeout") + expect(failedB.errorReason).toContain("execution permit") + }), + ), + ) + }) + + it("resets the extension budget on restart (S3) so a fresh attempt is not killed by a stale counter", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + // Cap of 1: the first attempt escalates exactly once (0 < 1). If the + // counter survived the restart, the second attempt would read 1 >= 1 + // and be force-cancelled at its very first deadline. + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Restart resets extension budget", + config: { name: "restart-budget", max_timeout_extensions: 1, nodes: [node("a", [], 300)] }, + }) + const gate1 = yield* takeWithin(childPrompts, "first attempt did not start") + const first = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === 1 ? current : undefined), + ), + "first escalation did not fire", + ) + expect(first.status).toBe("running") + const firstWake = yield* takeWithin(parentPrompts, "first wake did not reach the parent") + yield* Deferred.succeed(firstWake.release, "success") + // The child is left running (gate1 unreleased) — restart replaces + // the attempt; the replan handler cancels the old child session. + + // Main agent restarts the node (new attempt, new budget). + const plan = yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 300), restart: true }] }) + expect(plan.restart).toContain("a") + + // The second attempt starts with a zeroed budget. + const gate2 = yield* takeWithin(childPrompts, "second attempt did not start") + const reset = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.status === "running" && current.timeoutExtensions === 0 ? current : undefined), + ), + "extension budget was not reset after restart", + ) + expect(reset.timeoutExtensions).toBe(0) + + // Its own first deadline escalates (0 < cap 1) instead of force-killing. + const second = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === 1 && current.status === "running" ? current : undefined), + ), + "second attempt was force-cancelled by the stale extension counter (S3 regression)", + ) + expect(second.status).toBe("running") + const secondWake = yield* takeWithin(parentPrompts, "second wake did not reach the parent") + yield* Deferred.succeed(secondWake.release, "success") + yield* Deferred.succeed(gate2.release, "done") + }), + ), + ) + }) + + it("keeps supervising a running node after a same-value replan (§3.7: no re-time) and escalates again", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Same-value replan keeps supervision", + config: { name: "same-value", nodes: [node("a", [], 300)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + const first = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === 1 ? current : undefined), + ), + "first escalation did not fire", + ) + const firstWake = yield* takeWithin(parentPrompts, "first wake did not reach the parent") + yield* Deferred.succeed(firstWake.release, "success") + + // Same timeout_ms (300) — §3.7: the replan carries no NEW timeout, + // so it is NOT an adjudication: the deadline does not move and the + // self-renewing watcher keeps supervising. (The pre-§3.7 behavior + // re-timed here, which let an agent stall the cap by replanning.) + const plan = yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 300) }] }) + expect(plan.replace).toContain("a") + const second = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 2 && current.deadlineMs === first.deadlineMs + ? current + : undefined, + ), + ), + "second escalation never fired — supervision lost after same-value replan (§3.7 regression)", + ) + expect(second.status).toBe("running") + const secondWake = yield* takeWithin(parentPrompts, "second wake did not reach the parent") + yield* Deferred.succeed(secondWake.release, "success") + }), + ), + ) + }) + + it("re-escalates without any replan until the extension cap force-cancels (S1 self-renew)", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts, getCancelCount }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "S1 self-renew without replan", + config: { name: "s1-self-renew", max_timeout_extensions: 2, nodes: [node("a", [], 300)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + + // The main agent NEVER replans. The watcher must keep escalating on + // its own (one escalation per timeout period) instead of exiting + // after the first one — before S1 the extension count froze at 1 and + // the node ran unbounded, unreachable by the cap. + for (let i = 0; i < 2; i++) { + const escalated = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === i + 1 ? current : undefined), + ), + `escalation ${i + 1} did not fire without a replan (S1 regression)`, + ) + expect(escalated.status).toBe("running") + const wake = yield* takeWithin(parentPrompts, `wake ${i + 1} did not reach the parent`) + expect(wake.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(wake.release, "success") + } + + // Cap reached without any adjudication: force-cancel + nodeFailed. + const failed = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.status === "failed" ? current : undefined), + ), + "cap-exhausted force-cancel never fired without a replan (S1 regression)", + "10 seconds", + ) + expect(failed.errorClass).toBe("timeout") + expect(failed.errorReason).toContain("timeout extensions exhausted (2/2)") + expect(getCancelCount()).toBeGreaterThanOrEqual(1) + + const failureWake = yield* takeWithin(parentPrompts, "workflow-failure wake did not reach the parent") + yield* Deferred.succeed(failureWake.release, "success") + }), + ), + ) + }, 60_000) + + it("preserves the extended timeout when a replan omits timeout_ms (F2) and keeps supervising", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Omitted timeout keeps extension", + config: { name: "omitted-timeout", nodes: [node("a", [], 300)] }, + }) + yield* takeWithin(childPrompts, "a did not start") + const first = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === 1 ? current : undefined), + ), + "first escalation did not fire", + ) + const firstWake = yield* takeWithin(parentPrompts, "first wake did not reach the parent") + yield* Deferred.succeed(firstWake.release, "success") + + // Extend to 1500ms via an explicit timeout_ms (above the F9 clamp + // floor of 1000, far below the 600000 DEFAULT). + yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 1500) }] }) + // Gate on the deadline move so the re-escalation poll below cannot + // false-match the pre-replan count (the count is cumulative: it stays + // 1 across the extend and only climbs on the next escalation). + const extended = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.deadlineMs != null && current.deadlineMs > (first.deadlineMs ?? 0) + ? current + : undefined, + ), + ), + "deadline was not extended by the first replan", + ) + expect(extended.status).toBe("running") + expect(extended.timeoutExtensions).toBe(1) + const second = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === 2 ? current : undefined), + ), + "second escalation did not fire after the extension", + ) + const secondWake = yield* takeWithin(parentPrompts, "second wake did not reach the parent") + yield* Deferred.succeed(secondWake.release, "success") + + // Replan WITHOUT worker_config.timeout_ms. F2: the merged config must + // keep 1500 (not silently fall back to the 600000 DEFAULT), and §3.7: + // no NEW timeout means no re-time — the deadline stays frozen and the + // self-renewing watcher keeps supervising. + const bare = node("a", []) + delete bare.worker_config + yield* dag.replan(dagID, { nodes: [{ ...bare }] }) + const preserved = yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((wf) => { + const config = wf ? JSON.parse(wf.config) : undefined + return config?.nodes?.[0]?.worker_config?.timeout_ms === 1500 ? wf : undefined + }), + ), + "omitted timeout_ms was overwritten by the DEFAULT (F2 regression)", + ) + expect(preserved).toBeTruthy() + // Omitted-timeout replan is NOT an adjudication: the deadline does not + // move, the cumulative count climbs to 3, supervision continues. + const third = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => + current?.timeoutExtensions === 3 + && current.deadlineMs != null + && current.deadlineMs === second.deadlineMs + ? current + : undefined, + ), + ), + "supervision lost after omitted-timeout replan", + ) + expect(third.status).toBe("running") + const thirdWake = yield* takeWithin(parentPrompts, "third wake did not reach the parent") + yield* Deferred.succeed(thirdWake.release, "success") + }), + ), + ) + }, 60_000) + + it("re-delivers a completion wake after an escalation wake (F2b) instead of losing the result", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Completion wake after escalation", + config: { name: "f2b", nodes: [node("a", [], 300)] }, + }) + const gate = yield* takeWithin(childPrompts, "a did not start") + const escalated = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.timeoutExtensions === 1 ? current : undefined), + ), + "escalation did not fire", + ) + // The escalation wake is delivered and marked reported. + const timeoutWake = yield* takeWithin(parentPrompts, "timeout wake did not reach the parent") + expect(timeoutWake.text).toContain("[DAG Node Timeout]") + yield* Deferred.succeed(timeoutWake.release, "success") + + // The child finishes after the escalation; the completion must be + // delivered as a NEW wake (NodeCompleted re-arms wake_reported). + yield* Deferred.succeed(gate.release, "done") + const completed = yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((current) => current?.status === "completed" ? current : undefined), + ), + "node did not complete", + ) + const completionWake = yield* takeWithin(parentPrompts, "completion wake was lost behind the escalation (F2b regression)") + expect(completionWake.text).toContain("[DAG Node Result]") + expect(completionWake.text).toContain("completed") + yield* Deferred.succeed(completionWake.release, "success") + }), + ), + ) + }) +}) diff --git a/packages/opencode/test/dag/fixtures.ts b/packages/opencode/test/dag/fixtures.ts index 4bf5dc501d..e9ce2acd22 100644 --- a/packages/opencode/test/dag/fixtures.ts +++ b/packages/opencode/test/dag/fixtures.ts @@ -20,6 +20,8 @@ export function makeNodeRow(overrides: Partial = {}): DagStore wakeEligible: false, wakeReported: false, replanAttempts: 0, + timeoutExtensions: 0, + escalationPending: false, seq: 0, startedAt: null, completedAt: null, diff --git a/packages/opencode/test/dag/workflow-tool.test.ts b/packages/opencode/test/dag/workflow-tool.test.ts index e042c64e91..358db14d92 100644 --- a/packages/opencode/test/dag/workflow-tool.test.ts +++ b/packages/opencode/test/dag/workflow-tool.test.ts @@ -183,6 +183,8 @@ const store = Layer.mock(DagStore.Service, { wakeReported: false, replanAttempts: 0, seq: 1, + timeoutExtensions: 0, + escalationPending: false, startedAt: 1, completedAt: null, timeCreated: 1, @@ -207,6 +209,8 @@ const store = Layer.mock(DagStore.Service, { wakeReported: false, replanAttempts: 0, seq: 2, + timeoutExtensions: 0, + escalationPending: false, startedAt: 1, completedAt: 2, timeCreated: 1, diff --git a/packages/opencode/test/server/httpapi-exercise/index.ts b/packages/opencode/test/server/httpapi-exercise/index.ts index 118e4c4d72..3861f1be00 100644 --- a/packages/opencode/test/server/httpapi-exercise/index.ts +++ b/packages/opencode/test/server/httpapi-exercise/index.ts @@ -1811,6 +1811,7 @@ const scenarios: Scenario[] = [ check(typeof summary.failedNodes === "number", "summary should have failedNodes") check(typeof summary.status === "string", "summary should have status") check(typeof summary.title === "string", "summary should have title") + check(typeof summary.escalatedNodes === "number", "summary should have escalatedNodes") }), ), diff --git a/packages/schema/src/dag-event.ts b/packages/schema/src/dag-event.ts index 1d7570a09b..e5f8e3ae20 100644 --- a/packages/schema/src/dag-event.ts +++ b/packages/schema/src/dag-event.ts @@ -287,6 +287,21 @@ export const NodeRestarted = Event.define({ }) export type NodeRestarted = typeof NodeRestarted.Type +// Timeout is a signal, not a failure: the node keeps running and the main +// agent is woken to adjudicate (extend via replan with a new timeout_ms, or +// cancel/replan). The node row stays RUNNING; only timeout_extensions counts. +export const NodeTimeoutEscalated = Event.define({ + type: "dag.node.timeout_escalated", + ...options, + schema: { + ...Base, + nodeID: NodeID, + childSessionID: SessionID, + timeoutExtensions: Schema.Number, // current extension count (inclusive) + }, +}) +export type NodeTimeoutEscalated = typeof NodeTimeoutEscalated.Type + // ============================================================================ // Inventories + tagged unions // ============================================================================ @@ -310,6 +325,7 @@ export const DurableDefinitions = Event.inventory( NodeSkipped, NodeCancelled, NodeRestarted, + NodeTimeoutEscalated, ) export const Definitions = DurableDefinitions diff --git a/packages/schema/src/dag-summary.ts b/packages/schema/src/dag-summary.ts index 51299b1547..638cd70ed9 100644 --- a/packages/schema/src/dag-summary.ts +++ b/packages/schema/src/dag-summary.ts @@ -18,6 +18,10 @@ export const WorkflowSummary = Schema.Struct({ // finish with a "3/9" denominator lie. queued surfaces true concurrency. skippedNodes: Schema.Number, queuedNodes: Schema.Number, + // F10: running nodes with a not-yet-adjudicated timeout escalation + // (escalation_pending) — lets the TUI distinguish normal RUNNING from + // timeout-pending. Already-adjudicated (extended) nodes are excluded. + escalatedNodes: Schema.Number, }).annotate({ identifier: "DagWorkflowSummary" }) export type WorkflowSummary = typeof WorkflowSummary.Type diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 4e2fa870c0..287609adf2 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -24,7 +24,7 @@ describe("public event manifest", () => { SessionV1.Event.Error, ]) expect(EventManifest.Latest.size).toBe(92) - expect(EventManifest.Durable.size).toBe(53) + expect(EventManifest.Durable.size).toBe(54) }) test("uses canonical definitions for current public events", () => { diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 357a59607f..bac8214e76 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -684,6 +684,7 @@ export type DagWorkflowSummary = { failedNodes: number | "NaN" | "Infinity" | "-Infinity" | "Infinity" | "-Infinity" | "NaN" skippedNodes: number | "NaN" | "Infinity" | "-Infinity" | "Infinity" | "-Infinity" | "NaN" queuedNodes: number | "NaN" | "Infinity" | "-Infinity" | "Infinity" | "-Infinity" | "NaN" + escalatedNodes: number | "NaN" | "Infinity" | "-Infinity" | "Infinity" | "-Infinity" | "NaN" } export type SessionStatus = @@ -2977,6 +2978,7 @@ export type DagWorkflowSummary1 = { failedNodes: number | "NaN" | "Infinity" | "-Infinity" skippedNodes: number | "NaN" | "Infinity" | "-Infinity" queuedNodes: number | "NaN" | "Infinity" | "-Infinity" + escalatedNodes: number | "NaN" | "Infinity" | "-Infinity" } export type EventTuiPromptAppend2 = { diff --git a/packages/tui/src/feature-plugins/sidebar/dag-panel.tsx b/packages/tui/src/feature-plugins/sidebar/dag-panel.tsx index 29beb6c3c5..b2832241a5 100644 --- a/packages/tui/src/feature-plugins/sidebar/dag-panel.tsx +++ b/packages/tui/src/feature-plugins/sidebar/dag-panel.tsx @@ -4,7 +4,7 @@ import type { DagNode, DagWorkflowSummary } from "@opencode-ai/sdk/v2" import type { BuiltinTuiPlugin } from "../builtins" import { createEffect, createMemo, createSignal, For, Show } from "solid-js" import { Spinner } from "../../component/spinner" -import { computeWaves, dagNodeGlyph, dagStatusColor, formatDagProgress } from "../system/dag-inspector-utils" +import { computeWaves, dagEscalationLabel, dagNodeGlyph, dagStatusColor, formatDagProgress } from "../system/dag-inspector-utils" const id = "internal:sidebar-dag-panel" @@ -69,7 +69,8 @@ function WorkflowRow(props: { ({formatDagProgress(props.summary)} {running() > 0 ? `, ${running()} running` : ""} {queued() > 0 ? `, ${queued()} queued` : ""} - {failed() > 0 ? `, ${failed()} failed` : ""}) + {failed() > 0 ? `, ${failed()} failed` : ""} + {dagEscalationLabel(props.summary) ? `, ${dagEscalationLabel(props.summary)}` : ""}) diff --git a/packages/tui/src/feature-plugins/system/dag-inspector-utils.ts b/packages/tui/src/feature-plugins/system/dag-inspector-utils.ts index fb1e371027..7f5b58cf34 100644 --- a/packages/tui/src/feature-plugins/system/dag-inspector-utils.ts +++ b/packages/tui/src/feature-plugins/system/dag-inspector-utils.ts @@ -131,6 +131,14 @@ export function formatDagProgress(summary: { return `${Number(summary.completedNodes) + Number(summary.skippedNodes)}/${Number(summary.nodeCount)}` } +/** F10: timeout-pending indicator — running nodes past their deadline awaiting + * main-agent adjudication, shown distinctly from normal RUNNING. */ +export function dagEscalationLabel(summary: { escalatedNodes?: number | string }): string | undefined { + const escalated = Number(summary.escalatedNodes ?? 0) + if (!Number.isFinite(escalated) || escalated <= 0) return undefined + return `timeout ×${escalated}` +} + /** * Shared status→color mapping for every DAG surface (sidebar indicator, * sidebar panel, inspector) so one status never renders in different colors diff --git a/packages/tui/src/feature-plugins/system/dag-inspector.tsx b/packages/tui/src/feature-plugins/system/dag-inspector.tsx index 0c1717b584..9639a9458e 100644 --- a/packages/tui/src/feature-plugins/system/dag-inspector.tsx +++ b/packages/tui/src/feature-plugins/system/dag-inspector.tsx @@ -20,6 +20,7 @@ import { formatDagError, formatDagOutputPreview, formatDagProgress, + dagEscalationLabel, type DagControlOperation, type DagNode, } from "./dag-inspector-utils" @@ -492,6 +493,7 @@ function DagInspector(props: { api: TuiPluginApi }) { {formatDagProgress(wf)} + {dagEscalationLabel(wf) ? ` ${dagEscalationLabel(wf)}` : ""} ) diff --git a/packages/tui/test/cli/cmd/tui/sync-dag.test.tsx b/packages/tui/test/cli/cmd/tui/sync-dag.test.tsx index 7008dc42b8..421409d16c 100644 --- a/packages/tui/test/cli/cmd/tui/sync-dag.test.tsx +++ b/packages/tui/test/cli/cmd/tui/sync-dag.test.tsx @@ -32,6 +32,7 @@ function summary(completed: number, total: number, running = 0, failed = 0): Dag failedNodes: failed, skippedNodes: 0, queuedNodes: 0, + escalatedNodes: 0, } } diff --git a/packages/tui/test/feature-plugins/dag-inspector.test.tsx b/packages/tui/test/feature-plugins/dag-inspector.test.tsx index 578ffb2cb1..2c481d5c52 100644 --- a/packages/tui/test/feature-plugins/dag-inspector.test.tsx +++ b/packages/tui/test/feature-plugins/dag-inspector.test.tsx @@ -26,6 +26,7 @@ const wfSummary = (overrides: Partial = {}): DagWorkflowSumm failedNodes: 0, skippedNodes: 0, queuedNodes: 0, + escalatedNodes: 0, ...overrides, })