fix(dag): halt downstream scheduling when a reporting checkpoint returns verdict replan

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<void>
 }
@@ -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)
                   }),

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: `<workflow id="${wfId}" action="replan">\nAdded: ${r.add.join(", ")}\nCancelled: ${r.cancel.join(", ")}\nRestarted: ${r.restart.join(", ")}\nReplaced: ${r.replace.join(", ")}${ignored}\n</workflow>`,
+                  output: `<workflow id="${wfId}" action="replan">\nAdded: ${r.add.join(", ")}\nCancelled: ${r.cancel.join(", ")}\nRestarted: ${r.restart.join(", ")}\nReplaced: ${r.replace.join(", ")}${ignored}${pauseNote}\n</workflow>`,
                   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: `<workflow id="${wfId}" state="paused"/>\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: `<workflow id="${wfId}" state="paused"/>\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":

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")
+        }),
+      ),
+    )
+  })
+})

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