diff --git a/package.json b/package.json index f95bf3b23c..2b0555f523 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=4690", + "lint": "oxlint --max-warnings=4711", "typecheck": "bun turbo typecheck", "upgrade-opentui": "bun run script/upgrade-opentui.ts", "postinstall": "bun run --cwd packages/core fix-node-pty", diff --git a/packages/opencode/src/dag/dag.ts b/packages/opencode/src/dag/dag.ts index 531e947a80..5d9c7fd2eb 100644 --- a/packages/opencode/src/dag/dag.ts +++ b/packages/opencode/src/dag/dag.ts @@ -254,13 +254,26 @@ export interface Interface { export class Service extends Context.Service()("@opencode/Dag") {} +// The per-workflow KeyedMutex is single-permit and NOT reentrant: a guarded +// command invoked without the lock races its status guards, and one invoked +// while the lock is already held deadlocks silently (the permit never frees). +// The witness turns that convention into a type — only withWorkflowLock mints +// a WorkflowLock, so every guarded command below carries compile-time proof +// that it runs inside the lock's critical section. Internal composition +// (extend → replan → nodeFailed) forwards the caller's witness instead of +// re-acquiring the lock. +declare const WorkflowLockHeld: unique symbol +type WorkflowLock = { readonly [WorkflowLockHeld]: true } + export const layer = Layer.effect( Service, Effect.gen(function* () { const events = yield* EventV2Bridge.Service const store = yield* DagStore.Service const workflowLocks = KeyedMutex.makeUnsafe() - const withWorkflowLock = (dagID: string) => workflowLocks.withLock(dagID) + const lockWitness = {} as WorkflowLock + const withWorkflowLock = (dagID: string) => (body: (lock: WorkflowLock) => Effect.Effect) => + workflowLocks.withLock(dagID)(Effect.suspend(() => body(lockWitness))) const guardWorkflow = Effect.fn("Dag.guardWorkflow")(function* (dagID: string, target: WorkflowStatus) { const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) @@ -418,21 +431,24 @@ export const layer = Layer.effect( return dagID }) - const pause = Effect.fn("Dag.pause")(function* (dagID: string) { + const pause = Effect.fn("Dag.pause")(function* (lock: WorkflowLock, dagID: string) { yield* guardWorkflow(dagID, WorkflowStatus.PAUSED) yield* events.publish(DagEvent.WorkflowPaused, { dagID: dagID as ID, timestamp: yield* DateTime.now }) }) - const resume = Effect.fn("Dag.resume")(function* (dagID: string) { + const resume = Effect.fn("Dag.resume")(function* (lock: WorkflowLock, dagID: string) { yield* guardWorkflow(dagID, WorkflowStatus.RUNNING) yield* events.publish(DagEvent.WorkflowResumed, { dagID: dagID as ID, timestamp: yield* DateTime.now }) }) - const step = Effect.fn("Dag.step")(function* (dagID: string) { + const step = Effect.fn("Dag.step")(function* (lock: WorkflowLock, dagID: string) { // Guard: only `running` → `stepping` is valid. yield* guardWorkflow(dagID, WorkflowStatus.STEPPING) - // Reject if a node is still in-flight (one-at-a-time stepping). + // Reject if a node is still in-flight (one-at-a-time stepping). Queued + // counts as in-flight: the node is durably admitted and will start once + // a permit frees (P0-2) — stepping alongside it would put two nodes in + // flight. const nodes = yield* store.getNodes(dagID) - const hasInFlight = nodes.some((n) => n.status === "running") + const hasInFlight = nodes.some((n) => n.status === "running" || n.status === "queued") if (hasInFlight) return yield* Effect.fail(new Error(`Node still in-flight: cannot step ${dagID}`)) // Compute ready nodes using a transient WorkflowRuntime. const schedulingNodes = toSchedulingNodes(nodes) @@ -451,7 +467,7 @@ export const layer = Layer.effect( // nodes always get NodeSkipped. The projector's status guards make this // safe against races — a node that transitioned between the read and the // publish is silently left at its current status. - const terminateNonTerminalNodes = Effect.fnUntraced(function* (dagID: string, skipReason: "agent_complete" | "workflow_cancelled" | "workflow_failed", failReason: string, failRunning: boolean) { + const terminateNonTerminalNodes = Effect.fnUntraced(function* (lock: WorkflowLock, dagID: string, skipReason: "agent_complete" | "workflow_cancelled" | "workflow_failed", failReason: string, failRunning: boolean) { const nodes = yield* store.getNodes(dagID) for (const node of nodes) { if (isNodeTerminalStatus(node.status as NodeStatus)) continue @@ -475,12 +491,12 @@ export const layer = Layer.effect( } }) - const cancel = Effect.fn("Dag.cancel")(function* (dagID: string) { + const cancel = Effect.fn("Dag.cancel")(function* (lock: WorkflowLock, dagID: string) { yield* guardWorkflow(dagID, WorkflowStatus.CANCELLED) yield* events.publish(DagEvent.WorkflowCancelled, { dagID: dagID as ID, timestamp: yield* DateTime.now }) - yield* terminateNonTerminalNodes(dagID, "workflow_cancelled", "workflow_cancelled", false) + yield* terminateNonTerminalNodes(lock, dagID, "workflow_cancelled", "workflow_cancelled", false) }) - const complete = Effect.fn("Dag.complete")(function* (dagID: string) { + const complete = Effect.fn("Dag.complete")(function* (lock: WorkflowLock, dagID: string) { yield* guardWorkflow(dagID, WorkflowStatus.COMPLETED) const workflow = yield* store.getWorkflow(dagID).pipe(Effect.orDie) const config = workflow ? parseWorkflowConfig(workflow.config) : undefined @@ -488,17 +504,18 @@ export const layer = Layer.effect( ? unresolvedReviewOutcomes(config, yield* store.getNodes(dagID)) : [] if (unresolvedReviews.length > 0) yield* Effect.fail(new ReviewGateError(dagID, unresolvedReviews)) - yield* terminateNonTerminalNodes(dagID, "agent_complete", "", false) + yield* terminateNonTerminalNodes(lock, dagID, "agent_complete", "", false) yield* events.publish(DagEvent.WorkflowCompleted, { dagID: dagID as ID, durationMs: 0 as never, timestamp: yield* DateTime.now }) }) - const fail = Effect.fn("Dag.fail")(function* (dagID: string, reason: string) { + const fail = Effect.fn("Dag.fail")(function* (lock: WorkflowLock, dagID: string, reason: string) { yield* guardWorkflow(dagID, WorkflowStatus.FAILED) yield* events.publish(DagEvent.WorkflowFailed, { dagID: dagID as ID, reason, failedNodes: [] as never, timestamp: yield* DateTime.now }) - yield* terminateNonTerminalNodes(dagID, "workflow_failed", reason, true) + yield* terminateNonTerminalNodes(lock, dagID, "workflow_failed", reason, true) }) const _replan = Effect.fn("Dag._replan")(function* ( + lock: WorkflowLock, dagID: string, fragment: { nodes: NodeConfig[] }, reopenCompleted = false, @@ -567,7 +584,7 @@ export const layer = Layer.effect( for (const id of plan.restart) { const existing = nodeById.get(id) if (existing && existing.replanAttempts >= maxReplanAttempts) { - yield* nodeFailed(dagID, id, "replan attempt ceiling exceeded", "exec_failed").pipe(Effect.ignore) + yield* nodeFailed(lock, dagID, id, "replan attempt ceiling exceeded", "exec_failed").pipe(Effect.ignore) ceilingBreached.push(id) } } @@ -650,7 +667,7 @@ export const layer = Layer.effect( return { cancel: effectivePlan.cancel, restart: effectivePlan.restart, replace: effectivePlan.replace, add: effectivePlan.add, ignore: effectivePlan.ignore } }) - const _extend = Effect.fn("Dag._extend")(function* (dagID: string, newNodes: NodeConfig[]) { + const _extend = Effect.fn("Dag._extend")(function* (lock: WorkflowLock, dagID: string, newNodes: NodeConfig[]) { const wf = yield* store.getWorkflow(dagID) if (!wf) return yield* Effect.fail(new Error(`Workflow not found: ${dagID}`)) const nodes = yield* store.getNodes(dagID) @@ -689,30 +706,30 @@ export const layer = Layer.effect( // terminal, as do public replan and non-additive terminal mutations. // Internal call to _replan — shares the caller's lock holding period, // does NOT re-acquire the per-workflow lock or go through Service.of. - return yield* _replan(dagID, { nodes: [...preserved, ...newNodes] }, reopenCompleted) + return yield* _replan(lock, dagID, { nodes: [...preserved, ...newNodes] }, reopenCompleted) }) - const nodeQueued = Effect.fn("Dag.nodeQueued")(function* (dagID: string, nodeID: string, deadlineMs?: number) { + const nodeQueued = Effect.fn("Dag.nodeQueued")(function* (lock: WorkflowLock, dagID: string, nodeID: string, deadlineMs?: number) { yield* guardNode(dagID, nodeID, NodeStatus.QUEUED) yield* events.publish(DagEvent.NodeQueued, { dagID: dagID as ID, nodeID: nodeID as never, deadlineMs, timestamp: yield* DateTime.now }) }) - const nodeStarted = Effect.fn("Dag.nodeStarted")(function* (dagID: string, nodeID: string, childSessionID: string, deadlineMs?: number, wakeEligible?: boolean) { + const nodeStarted = Effect.fn("Dag.nodeStarted")(function* (lock: WorkflowLock, dagID: string, nodeID: string, childSessionID: string, deadlineMs?: number, wakeEligible?: boolean) { yield* guardNode(dagID, nodeID, NodeStatus.RUNNING) yield* events.publish(DagEvent.NodeStarted, { dagID: dagID as ID, nodeID: nodeID as never, childSessionID: childSessionID as never, deadlineMs, wakeEligible, timestamp: yield* DateTime.now }) }) - const nodeCompleted = Effect.fn("Dag.nodeCompleted")(function* (dagID: string, nodeID: string, output: unknown) { + const nodeCompleted = Effect.fn("Dag.nodeCompleted")(function* (lock: WorkflowLock, dagID: string, nodeID: string, output: unknown) { yield* guardNode(dagID, nodeID, NodeStatus.COMPLETED) yield* events.publish(DagEvent.NodeCompleted, { dagID: dagID as ID, nodeID: nodeID as never, output, durationMs: 0 as never, timestamp: yield* DateTime.now }) }) - const nodeFailed = Effect.fn("Dag.nodeFailed")(function* (dagID: string, nodeID: string, reason: string, trigger: string) { + const nodeFailed = Effect.fn("Dag.nodeFailed")(function* (lock: WorkflowLock, dagID: string, nodeID: string, reason: string, trigger: string) { yield* guardNode(dagID, nodeID, NodeStatus.FAILED) yield* events.publish(DagEvent.NodeFailed, { dagID: dagID as ID, nodeID: nodeID as never, reason, trigger: trigger as never, timestamp: yield* DateTime.now }) }) - const nodeSkipped = Effect.fn("Dag.nodeSkipped")(function* (dagID: string, nodeID: string, reason: string) { + const nodeSkipped = Effect.fn("Dag.nodeSkipped")(function* (lock: WorkflowLock, dagID: string, nodeID: string, reason: string) { yield* guardNode(dagID, nodeID, NodeStatus.SKIPPED) yield* events.publish(DagEvent.NodeSkipped, { dagID: dagID as ID, nodeID: nodeID as never, reason: reason as never, timestamp: yield* DateTime.now }) }) - const nodeCancelled = Effect.fn("Dag.nodeCancelled")(function* (dagID: string, nodeID: string) { + const nodeCancelled = Effect.fn("Dag.nodeCancelled")(function* (lock: WorkflowLock, dagID: string, nodeID: string) { // Cancellation is valid from any non-terminal status; no single target // NodeStatus is a legal transition from all of pending/queued/running/paused // (e.g. PAUSED -> SKIPPED is not in the table), so guard on terminality. @@ -728,7 +745,7 @@ export const layer = Layer.effect( timestamp: yield* DateTime.now, }) }) - const nodeRestarted = Effect.fn("Dag.nodeRestarted")(function* (dagID: string, nodeID: string, childSessionID: string) { + const nodeRestarted = Effect.fn("Dag.nodeRestarted")(function* (lock: WorkflowLock, dagID: string, nodeID: string, childSessionID: string) { 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 }) }) @@ -736,22 +753,22 @@ export const layer = Layer.effect( return Service.of({ create, store, - pause: (dagID) => withWorkflowLock(dagID)(pause(dagID)), - resume: (dagID) => withWorkflowLock(dagID)(resume(dagID)), - step: (dagID) => withWorkflowLock(dagID)(step(dagID)), - cancel: (dagID) => withWorkflowLock(dagID)(cancel(dagID)), - complete: (dagID) => withWorkflowLock(dagID)(complete(dagID)), - fail: (dagID, reason) => withWorkflowLock(dagID)(fail(dagID, reason)), - replan: (dagID, fragment) => withWorkflowLock(dagID)(_replan(dagID, fragment)), - extend: (dagID, nodes) => withWorkflowLock(dagID)(_extend(dagID, nodes)), - nodeQueued: (dagID, nodeID, deadlineMs) => withWorkflowLock(dagID)(nodeQueued(dagID, nodeID, deadlineMs)), + pause: (dagID) => withWorkflowLock(dagID)((lock) => pause(lock, dagID)), + resume: (dagID) => withWorkflowLock(dagID)((lock) => resume(lock, dagID)), + step: (dagID) => withWorkflowLock(dagID)((lock) => step(lock, dagID)), + cancel: (dagID) => withWorkflowLock(dagID)((lock) => cancel(lock, dagID)), + complete: (dagID) => withWorkflowLock(dagID)((lock) => complete(lock, dagID)), + fail: (dagID, reason) => withWorkflowLock(dagID)((lock) => fail(lock, dagID, reason)), + replan: (dagID, fragment) => withWorkflowLock(dagID)((lock) => _replan(lock, dagID, fragment)), + extend: (dagID, nodes) => withWorkflowLock(dagID)((lock) => _extend(lock, dagID, nodes)), + nodeQueued: (dagID, nodeID, deadlineMs) => withWorkflowLock(dagID)((lock) => nodeQueued(lock, dagID, nodeID, deadlineMs)), nodeStarted: (dagID, nodeID, childSessionID, deadlineMs, wakeEligible) => - withWorkflowLock(dagID)(nodeStarted(dagID, nodeID, childSessionID, deadlineMs, wakeEligible)), - nodeCompleted: (dagID, nodeID, output) => withWorkflowLock(dagID)(nodeCompleted(dagID, nodeID, output)), - nodeFailed: (dagID, nodeID, reason, trigger) => withWorkflowLock(dagID)(nodeFailed(dagID, nodeID, reason, trigger)), - nodeSkipped: (dagID, nodeID, reason) => withWorkflowLock(dagID)(nodeSkipped(dagID, nodeID, reason)), - nodeCancelled: (dagID, nodeID) => withWorkflowLock(dagID)(nodeCancelled(dagID, nodeID)), - nodeRestarted: (dagID, nodeID, childSessionID) => withWorkflowLock(dagID)(nodeRestarted(dagID, nodeID, childSessionID)), + withWorkflowLock(dagID)((lock) => nodeStarted(lock, dagID, nodeID, childSessionID, deadlineMs, wakeEligible)), + nodeCompleted: (dagID, nodeID, output) => withWorkflowLock(dagID)((lock) => nodeCompleted(lock, dagID, nodeID, output)), + nodeFailed: (dagID, nodeID, reason, trigger) => withWorkflowLock(dagID)((lock) => nodeFailed(lock, dagID, nodeID, reason, trigger)), + 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)), }) }), ) diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index def308e103..11cbdd5138 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -59,6 +59,11 @@ export const layer = Layer.effect( const state = yield* InstanceState.make( Effect.fn("DagLoop.state")(function* (ctx) { const runtimes = new Map() + // Adoption in-flight reservations: recoverWorkflow spans async yield + // points before it publishes its entry into `runtimes`, so a plain + // runtimes.has check is check-then-act. Reserved synchronously at + // adoption entry, released when the adoption settles. + const recovering = new Set() const wakeInFlight = new Set() const wakePending = new Set() @@ -302,67 +307,92 @@ export const layer = Layer.effect( // server spawns children under a foreign directory context. if (wf.projectId !== ctx.project.id) return const dagID = wf.id - const config = parseWorkflowConfig(wf.config) - const recovery = yield* reconcileWorkflow( - dagID, - checkSessionStatus, - (sid) => promptSvc.cancel(sid as never), - config, - ).pipe( - Effect.provideService(Dag.Service, dag), - ) - // P2-2 recovery-pause: reconciliation invented failures (ownership - // lost / no child session / deadline enforced offline) without any - // durable proof of the child's outcome. Letting spawnReady cascade - // skips and checkCompletion terminalize now would weld the workflow - // into a terminal status the parent never sanctioned — and terminal - // nodes are immutable, so replan could no longer rewire downstream. - // Pause instead: pending nodes stay replannable, the durable - // NodeFailed wake rows reach the parent at the paused delivery - // boundary, and disposition (replan / resume / cancel) stays under - // explicit workflow control. - const pausedForRecovery = recovery.ownershipLost > 0 && wf.status === "running" - if (pausedForRecovery) { - yield* dag.pause(dagID) - yield* Effect.logWarning("DagLoop paused workflow after recovery invented node failures", { - dagID, - reconciled: recovery.reconciled, - ownershipLost: recovery.ownershipLost, - }) - } - if (recovery.ownershipLost > 0 && !pausedForRecovery) { - yield* Effect.logWarning("DagLoop terminalized recovered nodes after execution ownership loss", { + // Idempotency guard: the startup scan and the WorkflowReplanned + // handler's re-adoption path can both reach here for the same + // workflow (e.g. a replan call arriving while the scan is still + // reconciling it). A duplicate adoption would reconcile twice and + // overwrite the runtimes entry — orphaning the first entry's fibers + // from every interrupt sweep. Reserve synchronously before the + // first yield so the second caller drops out immediately. + if (runtimes.has(dagID) || recovering.has(dagID)) return + recovering.add(dagID) + try { + const config = parseWorkflowConfig(wf.config) + const recovery = yield* reconcileWorkflow( dagID, - reconciled: recovery.reconciled, - ownershipLost: recovery.ownershipLost, - }) - } - const nodes = yield* store.getNodes(dagID) - 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 isPaused = wf.status === "paused" || pausedForRecovery - 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() } - runtimes.set(dagID, entry) - // Reconciliation settles every persisted running attempt before the - // runtime is rebuilt. Recovery never adopts or restarts provider work; - // a new execution attempt must come from explicit workflow control. - if (!isPaused && !isStepping) { - yield* entry.evalLock.withPermits(1)( - Effect.gen(function* () { - yield* spawnReady(dagID) - yield* checkCompletion(dagID) - }), + checkSessionStatus, + (sid) => promptSvc.cancel(sid as never), + config, + ).pipe( + Effect.provideService(Dag.Service, dag), ) - } - // Deliver the invented-failure wake rows now instead of waiting for - // the next idle event — the workflow just paused itself and the - // parent is the only actor that can dispose of it. - if (pausedForRecovery) { - yield* tryDeliverWake(wf.sessionId).pipe(Effect.ignore, Effect.forkScoped) + // P2-2 recovery-pause: reconciliation invented failures (ownership + // lost / no child session / deadline enforced offline) without any + // durable proof of the child's outcome. Letting spawnReady cascade + // skips and checkCompletion terminalize now would weld the workflow + // into a terminal status the parent never sanctioned — and terminal + // nodes are immutable, so replan could no longer rewire downstream. + // Pause instead: pending nodes stay replannable, the durable + // NodeFailed wake rows reach the parent at the paused delivery + // boundary, and disposition (replan / resume / cancel) stays under + // explicit workflow control. + const pausedForRecovery = recovery.ownershipLost > 0 && wf.status === "running" + if (pausedForRecovery) { + // A concurrent control op (cancel/fail) can terminalize the + // workflow while reconciliation runs — the pause guard then + // rejects. Abandon adoption instead of tracking a workflow this + // instance no longer controls. + const pauseAccepted = yield* dag.pause(dagID).pipe( + Effect.as(true), + Effect.catchCause((cause) => + Effect.logWarning("DagLoop recovery pause rejected — abandoning adoption", { dagID, cause }).pipe( + Effect.as(false), + ), + ), + ) + if (!pauseAccepted) return + yield* Effect.logWarning("DagLoop paused workflow after recovery invented node failures", { + dagID, + reconciled: recovery.reconciled, + ownershipLost: recovery.ownershipLost, + }) + } + if (recovery.ownershipLost > 0 && !pausedForRecovery) { + yield* Effect.logWarning("DagLoop terminalized recovered nodes after execution ownership loss", { + dagID, + reconciled: recovery.reconciled, + ownershipLost: recovery.ownershipLost, + }) + } + const nodes = yield* store.getNodes(dagID) + 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 isPaused = wf.status === "paused" || pausedForRecovery + 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() } + runtimes.set(dagID, entry) + // Reconciliation settles every persisted running attempt before the + // runtime is rebuilt. Recovery never adopts or restarts provider work; + // a new execution attempt must come from explicit workflow control. + if (!isPaused && !isStepping) { + yield* entry.evalLock.withPermits(1)( + Effect.gen(function* () { + yield* spawnReady(dagID) + yield* checkCompletion(dagID) + }), + ) + } + // Deliver the invented-failure wake rows now instead of waiting for + // the next idle event — the workflow just paused itself and the + // parent is the only actor that can dispose of it. + if (pausedForRecovery) { + yield* tryDeliverWake(wf.sessionId).pipe(Effect.ignore, Effect.forkScoped) + } + } finally { + recovering.delete(dagID) } }) @@ -537,13 +567,28 @@ export const layer = Layer.effect( 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 + // an earlier one. Applying the event's implied flags blindly lets a + // cross-stream ordering (pause/resume/step race) clobber a newer + // control decision. Mirrors the node handlers' DB arbitration. + const refreshControlFlags = Effect.fnUntraced(function* (dagID: string, entry: WorkflowEntry) { + const workflow = yield* store.getWorkflow(dagID) + if (!workflow || isWorkflowTerminalStatus(workflow.status as never)) return undefined + entry.runtime.setPaused(workflow.status === "paused") + entry.runtime.setStepMode(workflow.status === "stepping") + return workflow + }) + yield* events.subscribe(DagEvent.WorkflowPaused).pipe( Stream.filter((e) => runtimes.has(e.data.dagID as string)), Stream.runForEach((evt) => Effect.gen(function* () { - const entry = runtimes.get(evt.data.dagID as string) + const dagID = evt.data.dagID as string + const entry = runtimes.get(dagID) if (!entry) return - yield* entry.evalLock.withPermits(1)(Effect.sync(() => entry.runtime.setPaused(true))) + yield* entry.evalLock.withPermits(1)(refreshControlFlags(dagID, entry)) }).pipe(guarded("WorkflowPaused")), ), Effect.forkScoped({ startImmediately: true }), @@ -558,7 +603,15 @@ export const layer = Layer.effect( if (!entry) return yield* entry.evalLock.withPermits(1)( Effect.gen(function* () { - entry.runtime.setStepMode(true) + const workflow = yield* refreshControlFlags(dagID, entry) + if (workflow?.status !== "stepping") return + // Dag.step validated "no in-flight node" on a DB snapshot + // taken outside this evalLock; a terminal-event handler can + // spawn in between (its DB status read predated the stepped + // projection). Spawning now would put a second node in + // flight under stepping — leave advancement to the next + // explicit step command instead. + if (entry.runtime.hasRunning()) return yield* spawnReady(dagID) }), ) @@ -576,9 +629,8 @@ export const layer = Layer.effect( if (!entry) return yield* entry.evalLock.withPermits(1)( Effect.gen(function* () { - entry.runtime.setPaused(false) - entry.runtime.setStepMode(false) - yield* spawnReady(dagID) + const workflow = yield* refreshControlFlags(dagID, entry) + if (workflow?.status === "running") yield* spawnReady(dagID) // A workflow can be resumed with every node already settled // (e.g. recovery-pause on a single lost node). Without this, // nothing else re-runs completion and the workflow hangs in diff --git a/packages/opencode/test/dag/dag-adoption-step-races.test.ts b/packages/opencode/test/dag/dag-adoption-step-races.test.ts new file mode 100644 index 0000000000..074f2add11 --- /dev/null +++ b/packages/opencode/test/dag/dag-adoption-step-races.test.ts @@ -0,0 +1,341 @@ +import { describe, expect, it } from "bun:test" +import { DateTime, Deferred, Effect, Fiber, 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 { WorkflowNodeTable, WorkflowTable } from "@opencode-ai/core/dag/sql" +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, 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 +} + +function takeWithin(queue: Queue.Queue, message: string) { + return Queue.take(queue).pipe( + Effect.timeoutOption("2 seconds"), + Effect.flatMap(Option.match({ + onNone: () => Effect.fail(new Error(message)), + onSome: Effect.succeed, + })), + ) +} + +function reply(sessionID: string, text: string): SessionV1.WithParts { + return { + info: { + id: MessageID.ascending(), + 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 nodeConfig(id: string, dependsOn: string[] = []): NodeConfig { + return { + id, + name: id, + worker_type: "build", + depends_on: dependsOn, + required: true, + prompt_template: { inline: id }, + report_to_parent: false, + } +} + +function raceLayer(input: { + readonly childPrompts: Queue.Queue + readonly cancelled: string[] + readonly messages: (value: { sessionID: string; limit?: 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 dag = Dag.layer.pipe( + Layer.provide(bridge), + Layer.provide(store), + ) + const base = Layer.mergeAll(database, events, bridge, store, projector, dag, status) + const childTitles = new Map() + const created: string[] = [] + const session = Layer.mock(Session.Service, { + get: () => Effect.succeed({ id: "ses_parent", permission: [], agent: "build" } as never), + create: (value) => + Effect.sync(() => { + const id = `ses_child_${created.length + 1}` + created.push(id) + childTitles.set(id, (value?.title ?? id).replace(" (DAG node)", "")) + return { id } as never + }), + messages: (value) => input.messages(value as never) as never, + }) + const prompt = Layer.mock(SessionPrompt.Service, { + cancel: (sessionID) => Effect.sync(() => void input.cancelled.push(sessionID as string)), + prompt: Effect.fn("test.SessionPrompt.prompt")(function* (value: SessionPrompt.PromptInput) { + const sessionID = value.sessionID as string + const release = yield* Deferred.make() + yield* Queue.offer(input.childPrompts, { + title: childTitles.get(sessionID) ?? sessionID, + release, + }) + return reply(sessionID, yield* Deferred.await(release)) + }), + // Keep wake delivery pending so the tests observe scheduling only. + promptIfIdle: () => Effect.succeed(Option.none()), + }) + 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.merge(base, loop) +} + +function runRaceTest( + input: { + readonly messages: (value: { sessionID: string; limit?: number }) => Effect.Effect + }, + test: (services: { + readonly dag: Dag.Interface + readonly loop: DagLoop.Interface + readonly store: DagStore.Interface + readonly events: EventV2.Interface + readonly database: Database.Interface + readonly childPrompts: Queue.Queue + readonly cancelled: string[] + }) => Effect.Effect, +) { + return Effect.gen(function* () { + const childPrompts = yield* Queue.unbounded() + const cancelled: string[] = [] + return yield* Effect.gen(function* () { + const dag = yield* Dag.Service + const loop = yield* DagLoop.Service + const store = yield* DagStore.Service + const events = yield* EventV2.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) + return yield* test({ dag, loop, store, events, database, childPrompts, cancelled }) + }).pipe( + Effect.provide(raceLayer({ childPrompts, cancelled, messages: input.messages })), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: "project-1" }, + } as never), + Effect.scoped, + ) + }) +} + +const activeReply = [{ info: { role: "assistant", finish: undefined } }] as never as SessionV1.WithParts[] + +describe("DagLoop adoption idempotency", () => { + it("adopts a workflow exactly once when a replan event races the startup scan", async () => { + // The startup scan blocks inside reconciliation (first child status + // probe); a WorkflowReplanned event arriving in that window finds no + // runtimes entry and takes the re-adoption path. Without the synchronous + // recovering reservation both adoptions run: reconciliation executes + // twice (double child-session cancel) and the second runtimes.set + // orphans the first entry's fibers. + const gate = await Effect.runPromise(Deferred.make()) + const childStatusCalls = { count: 0 } + await Effect.runPromise( + runRaceTest( + { + messages: (value) => { + if (value.sessionID !== "ses_child1") return Effect.succeed([]) + childStatusCalls.count += 1 + if (childStatusCalls.count === 1) { + return Deferred.await(gate).pipe(Effect.as(activeReply)) + } + return Effect.succeed(activeReply) + }, + }, + ({ dag, loop, store, events, cancelled }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Adoption race", + config: { name: "adoption-race", nodes: [nodeConfig("n1"), nodeConfig("n2", ["n1"])] }, + }) + yield* dag.nodeStarted(dagID, "n1", "ses_child1") + + const initFiber = yield* loop.init().pipe(Effect.forkChild) + yield* pollWithTimeout( + Effect.sync(() => childStatusCalls.count === 1 ? true as const : undefined), + "startup scan did not reach the child status probe", + ) + yield* events.publish(DagEvent.WorkflowReplanned, { + dagID: dagID as never, + added: 0 as never, + removed: 0 as never, + replaced: 0 as never, + restarted: 0 as never, + timestamp: yield* DateTime.now, + }) + // Give the WorkflowReplanned handler time to attempt re-adoption + // while the scan is still blocked, then let the scan finish. + yield* Effect.sleep("150 millis") + yield* Deferred.succeed(gate, undefined) + yield* Fiber.join(initFiber) + + expect(childStatusCalls.count).toBe(1) + expect(cancelled).toEqual(["ses_child1"]) + expect((yield* store.getWorkflow(dagID))?.status).toBe("paused") + expect((yield* store.getNode(dagID, "n1"))?.status).toBe("failed") + expect((yield* store.getNode(dagID, "n2"))?.status).toBe("pending") + }), + ), + ) + }) +}) + +describe("DagLoop stepping race window", () => { + it("does not spawn a second node when a stale stepped event lands while one is in flight", async () => { + await Effect.runPromise( + runRaceTest( + { messages: () => Effect.succeed([]) }, + ({ dag, loop, store, events, database, childPrompts }) => + Effect.gen(function* () { + const dagID = "dag_step_race" + yield* database.db.transaction((tx) => + Effect.gen(function* () { + yield* tx.insert(WorkflowTable).values({ + id: dagID, + project_id: "project-1" as never, + session_id: "ses_parent" as never, + title: "Step race", + status: "stepping", + config: JSON.stringify({ name: "step-race", nodes: [nodeConfig("a"), nodeConfig("b")] }), + seq: 2, + wake_reported: false, + }).run() + yield* tx.insert(WorkflowNodeTable).values([ + { + id: "a", + workflow_id: dagID, + name: "a", + worker_type: "build", + status: "pending", + required: true, + depends_on: [], + wake_eligible: false, + wake_reported: false, + seq: 1, + }, + { + id: "b", + workflow_id: dagID, + name: "b", + worker_type: "build", + status: "pending", + required: true, + depends_on: [], + wake_eligible: false, + wake_reported: false, + seq: 0, + }, + ]).run() + }), + ).pipe(Effect.orDie) + + yield* loop.init() + expect(yield* dag.step(dagID)).toEqual({ status: "stepping", nodeID: "a" }) + const first = yield* takeWithin(childPrompts, "stepped node a did not start") + expect(first.title).toBe("a") + + // A second stepped event admitted on a snapshot that predated a's + // spawn (the step race): the handler must re-check in-flight work + // under the evalLock and refuse to put a second node in flight. + yield* events.publish(DagEvent.WorkflowStepped, { + dagID: dagID as never, + nodeID: "b" as never, + timestamp: yield* DateTime.now, + }) + yield* Effect.sleep("150 millis") + expect(Option.isNone(yield* Queue.poll(childPrompts))).toBe(true) + expect((yield* store.getNode(dagID, "b"))?.status).toBe("pending") + expect((yield* store.getNode(dagID, "a"))?.status).toBe("running") + + // The next explicit step after a settles advances exactly one node. + yield* Deferred.succeed(first.release, "done") + yield* pollWithTimeout( + store.getNode(dagID, "a").pipe( + Effect.map((item) => item?.status === "completed" ? item : undefined), + ), + "stepped node a did not complete", + ) + expect(yield* dag.step(dagID)).toEqual({ status: "stepping", nodeID: "b" }) + const second = yield* takeWithin(childPrompts, "next step did not start node b") + expect(second.title).toBe("b") + yield* Deferred.succeed(second.release, "done") + yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((workflow) => workflow?.status === "completed" ? workflow : undefined), + ), + "step-race workflow did not complete", + ) + }), + ), + ) + }) +})