diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index a725cdd56..e0986c925 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -3,7 +3,7 @@ export * as DagLoop from "./loop" -import { Cause, Effect, Layer, Context, Stream, Semaphore, Fiber, Option, DateTime, Clock } from "effect" +import { Cause, Effect, Layer, Context, Stream, Semaphore, Fiber, Option, DateTime, Clock, Schema } from "effect" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { SessionV1 } from "@opencode-ai/core/v1/session" import { InstanceState } from "@/effect/instance-state" @@ -36,6 +36,12 @@ import { spawnNode, makeDeadlineWatcher } from "./spawn" import { evaluateCondition, resolveInputMapping } from "./eval" import { reconcileWorkflow, makeSessionStatusChecker } from "./recovery" +// A reporting checkpoint's replan verdict vetoes the current direction: the +// workflow pauses durably before any downstream spawn (see NodeCompleted +// handler). Only the verdict shape matters — any node whose submitted output +// matches triggers the gate, so non-reporting nodes can never trip it. +const GateReplanVerdict = Schema.Struct({ verdict: Schema.Literal("replan") }) + export interface Interface { readonly init: () => Effect.Effect } @@ -652,10 +658,36 @@ const serviceLayer = Layer.effect( // back. Mirrors the NodeFailed handler's isActive guard. if (confirmed && entry.runtime.isActive(nodeID)) { settle(entry, nodeID) + const nodeConfig = entry.config?.nodes.find((n) => n.id === nodeID) + const gateReplan = def === DagEvent.NodeCompleted + && nodeConfig?.report_to_parent === true + && Option.isSome(Schema.decodeUnknownOption(GateReplanVerdict)(node?.output)) + if (gateReplan) { + // Verdict gate (issue #322): a reporting checkpoint that + // submits verdict "replan" vetoes the direction. Pause + // durably BEFORE any spawn round so dependents can never + // run on the rejected direction; the parent is woken by + // the report_to_parent wake and control(replan) applies + // corrective nodes — a paused workflow resumes as part + // of replan (workflow tool) so corrections can run. + const paused = yield* dag.pause(dagID).pipe( + Effect.map(() => true), + Effect.catch(() => + Effect.gen(function* () { + const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) + if (wf?.status !== "paused") + yield* Effect.logWarning("DagLoop pause on replan verdict failed", { dagID, nodeID }) + return wf?.status === "paused" + }), + ), + ) + entry.runtime.setPaused(paused) + yield* Effect.logWarning("DagLoop paused workflow after gate verdict: replan", { dagID, nodeID }) + } // In stepMode, do NOT auto-advance — wait for the next // explicit step command. checkCompletion still runs so // required-node failure / early completion is detected. - if (!entry.runtime.isStepMode()) yield* spawnReady(dagID) + if (!gateReplan && !entry.runtime.isStepMode()) yield* spawnReady(dagID) } yield* checkCompletion(dagID) }), diff --git a/packages/opencode/src/tool/workflow.ts b/packages/opencode/src/tool/workflow.ts index a59259388..afa66390a 100644 --- a/packages/opencode/src/tool/workflow.ts +++ b/packages/opencode/src/tool/workflow.ts @@ -726,13 +726,37 @@ export const WorkflowTool = Tool.define< dag.replan(wfId, { nodes: result.prepared.nodes }), "The workflow reached a terminal status before the replan arrived — terminal workflows are immutable. Recover by starting a new workflow with the updated node definitions, or extend if a reporting leaf checkpoint naturally completed the graph. Next time issue control(pause) BEFORE composing the spec.", ).pipe(Effect.orDie) + // A paused workflow (explicit pause-first protocol, or the + // runtime's gate pause after a checkpoint replan verdict) must + // resume for the corrective nodes to run — the replan intent + // is "the graph changed, proceed", so resume closes the loop. + // Resume races with concurrent control ops are tolerated: the + // replan already landed, so never die on them. + const wfAfterReplan = yield* dag.store.getWorkflow(wfId).pipe(Effect.orDie) + const resumedFromPause = wfAfterReplan?.status === "paused" + const resumedOk = resumedFromPause + ? yield* dag.resume(wfId).pipe( + Effect.map(() => true), + Effect.catch((error) => + Effect.gen(function* () { + yield* Effect.logWarning("Workflow resume after replan failed", { wfId, error }) + return false + }), + ), + ) + : false const ignored = r.ignore.length > 0 ? `\nIgnored (terminal, immutable — add replacements under new ids to retry): ${r.ignore.join(", ")}` : "" + const pauseNote = !resumedFromPause + ? "" + : resumedOk + ? "\nWorkflow was paused and has been resumed — corrective nodes are now schedulable." + : "\nWorkflow was paused; automatic resume raced with another control op — check status and issue control(resume) if still paused." return { title: `Workflow replanned: +${r.add.length} -${r.cancel.length} ↻${r.restart.length}`, - output: `\nAdded: ${r.add.join(", ")}\nCancelled: ${r.cancel.join(", ")}\nRestarted: ${r.restart.join(", ")}\nReplaced: ${r.replace.join(", ")}${ignored}\n`, + output: `\nAdded: ${r.add.join(", ")}\nCancelled: ${r.cancel.join(", ")}\nRestarted: ${r.restart.join(", ")}\nReplaced: ${r.replace.join(", ")}${ignored}${pauseNote}\n`, metadata: { workflowId: wfId, ...r } as Metadata, } } @@ -741,7 +765,7 @@ export const WorkflowTool = Tool.define< yield* dag.pause(wfId).pipe(Effect.orDie) return { title: "Workflow paused", - output: `\nNote: pause stops new node spawns only — nodes already running continue to completion. To stop a running node, submit a replan spec marking it restart: true or cancel: true (replan is valid while paused).`, + output: `\nNote: pause stops new node spawns only — nodes already running continue to completion. To stop a running node, submit a replan spec marking it restart: true or cancel: true (replan is valid while paused, and a successful replan resumes the paused workflow so corrective nodes can run).`, metadata: { workflowId: wfId } as Metadata, } case "resume": diff --git a/packages/opencode/test/dag/dag-loop-guards.test.ts b/packages/opencode/test/dag/dag-loop-guards.test.ts index b2b4ab1d8..43d1e35cf 100644 --- a/packages/opencode/test/dag/dag-loop-guards.test.ts +++ b/packages/opencode/test/dag/dag-loop-guards.test.ts @@ -370,3 +370,83 @@ describe("DagLoop cancel-skip race", () => { ) }) }) + +describe("DagLoop replan verdict gate (issue #322)", () => { + it("pauses the workflow on a reporting checkpoint's replan verdict and blocks dependents until resume", async () => { + await Effect.runPromise( + runGuardTest({ instanceProject: "project-1" }, ({ dag, store, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_project-1", + title: "Gate replan verdict", + config: { + name: "gate-replan", + nodes: [ + node({ id: "gate", name: "gate", required: true, report_to_parent: true, output_schema: { type: "object" } }), + node({ id: "downstream", name: "downstream", required: false, depends_on: ["gate"] }), + ], + }, + }) + const gateChild = yield* takeWithin(childPrompts, "gate node did not start") + expect(gateChild.title).toBe("gate") + // The checkpoint submits a replan verdict (issue #322: the graph used + // to spawn the dependent anyway and spin to terminal). + yield* dag.nodeCompleted(dagID, "gate", { verdict: "replan", findings: "direction vetoed" }) + yield* pollWithTimeout( + Effect.gen(function* () { + const wf = yield* store.getWorkflow(dagID) + return wf?.status === "paused" ? (true as const) : undefined + }), + "workflow did not pause after the replan verdict", + ) + // The only prompt that may land while paused is the report_to_parent + // wake for the parent session — a dependent spawn would flip the + // durable row to queued/running first. + const woken = yield* takeWithin(childPrompts, "report_to_parent wake never delivered") + expect(woken.title).toBe("ses_project-1") + expect(Option.isNone(yield* Queue.poll(childPrompts))).toBe(true) + expect((yield* store.getNode(dagID, "downstream"))?.status).toBe("pending") + // Parent disposition: replan fragment + resume continues the graph. + yield* dag.resume(dagID) + const downstreamChild = yield* takeWithin(childPrompts, "downstream did not start after resume") + expect(downstreamChild.title).toBe("downstream") + yield* Deferred.succeed(downstreamChild.release, "done") + }), + ), + ) + }) + + it("advances normally when a reporting checkpoint submits verdict continue", async () => { + await Effect.runPromise( + runGuardTest({ instanceProject: "project-1" }, ({ dag, store, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_project-1", + title: "Gate continue verdict", + config: { + name: "gate-continue", + nodes: [ + node({ id: "gate", name: "gate", required: true, report_to_parent: true, output_schema: { type: "object" } }), + node({ id: "downstream", name: "downstream", required: false, depends_on: ["gate"] }), + ], + }, + }) + const gateChild = yield* takeWithin(childPrompts, "gate node did not start") + expect(gateChild.title).toBe("gate") + yield* dag.nodeCompleted(dagID, "gate", { verdict: "continue", findings: "direction confirmed" }) + // The report_to_parent wake and the downstream spawn can land in + // either order; accept the downstream prompt whichever comes second. + const first = yield* takeWithin(childPrompts, "no prompt after continue verdict") + const downstreamChild = first.title === "downstream" + ? first + : yield* takeWithin(childPrompts, "downstream did not spawn after continue verdict") + expect(downstreamChild.title).toBe("downstream") + expect((yield* store.getWorkflow(dagID))?.status).toBe("running") + yield* Deferred.succeed(downstreamChild.release, "done") + }), + ), + ) + }) +}) diff --git a/packages/opencode/test/dag/workflow-tool.test.ts b/packages/opencode/test/dag/workflow-tool.test.ts index c2fef15d6..97fbca720 100644 --- a/packages/opencode/test/dag/workflow-tool.test.ts +++ b/packages/opencode/test/dag/workflow-tool.test.ts @@ -1223,6 +1223,41 @@ describe("workflow tool execution", () => { }), ) + runtime.effect("replanning a gate-paused workflow resumes it so corrective nodes can run", () => + Effect.gen(function* () { + published.length = 0 + const info = yield* WorkflowTool + const workflow = yield* info.init() + const spec_path = yield* writeWorkflowSpec("paused-replan", { + fragment: { + name: "paused-replan", + nodes: [ + { + id: "corrective", + name: "Corrective", + worker_type: "general", + depends_on: [], + prompt_template: { inline: "work" }, + }, + ], + }, + }) + const result = yield* workflow.execute( + Schema.decodeUnknownSync(Parameters)({ params: { + action: "control", + workflow_id: "dag_paused", + operation: "replan", + spec_path, + }}), + toolContext(), + ) + + expect(result.title).toContain("Workflow replanned: +1") + expect(result.output).toContain("has been resumed") + expect(published.some((event) => event.type === DagEvent.WorkflowResumed.type)).toBe(true) + }), + ) + runtime.effect("rejects inline or missing spec sources before side effects", () => Effect.gen(function* () { const info = yield* WorkflowTool