From e129daa891d6b8fd9a0caf0829be447013834885 Mon Sep 17 00:00:00 2001
From: Lex <sunsan05@Gmail.com>
Date: Mon, 17 Aug 2026 21:15:23 +0800
Subject: [PATCH 1/7] fix(dag): reject ungated reporting checkpoints at
 authoring (issue #320)

---
 .../src/plugin/command/workflow-routing.md    |   4 +
 packages/opencode/src/dag/CONTEXT.md          |   4 +
 .../adr/0003-reporting-checkpoint-gating.md   |  70 +++++++++
 packages/opencode/src/dag/validation.ts       |  47 +++++-
 .../test/dag/dag-checkpoint-gate.test.ts      | 139 ++++++++++++++++++
 5 files changed, 258 insertions(+), 6 deletions(-)
 create mode 100644 packages/opencode/src/dag/docs/adr/0003-reporting-checkpoint-gating.md
 create mode 100644 packages/opencode/test/dag/dag-checkpoint-gate.test.ts

diff --git a/packages/core/src/plugin/command/workflow-routing.md b/packages/core/src/plugin/command/workflow-routing.md
index 878aea801b..85b9db97b4 100644
--- a/packages/core/src/plugin/command/workflow-routing.md
+++ b/packages/core/src/plugin/command/workflow-routing.md
@@ -108,6 +108,10 @@ Top level is `title`/`mode`/`admission` (optional) and `config` (required);
 `kind` (required), `depends_on`, `instruction`, `worker_type`, `required`,
 `report_to_parent` — never `worker`, `prompt`, or `agent`.
 
+A `report_to_parent` node with dependents is a reporting checkpoint: gate each
+dependent on its output via `condition`, keep it a reporting leaf, or drop
+`report_to_parent`.
+
 Validate that `spec_path` before start. Fix every diagnostic in the same file
 and revalidate; validation creates no workflow. A successful start returns the
 exact workflow ID. The parent owns the graph, controls, and final report;
diff --git a/packages/opencode/src/dag/CONTEXT.md b/packages/opencode/src/dag/CONTEXT.md
index 3cd98f308e..38ac3de76a 100644
--- a/packages/opencode/src/dag/CONTEXT.md
+++ b/packages/opencode/src/dag/CONTEXT.md
@@ -13,6 +13,7 @@ Workflow Orchestration turns one user objective into one durable DAG. Its model-
 | Orchestration Router | The product-owned parent guidance that qualifies an objective and selects one Workflow Route without external Skill discovery. |
 | Block Composer | The Orchestration Router decision that selects the smallest Block graph justified by current evidence. |
 | Decision Checkpoint | One parent-owned confirmation for unresolved user choices that materially change behavior, scope, acceptance, or an irreversible boundary. |
+| Reporting Checkpoint | A `report_to_parent: true` node with dependents; its dependents must gate on its output via `condition`, or it must be a reporting leaf. |
 | Workflow Brief | The recommended route, scope, acceptance evidence, assumptions, risks, and material alternatives presented at a Decision Checkpoint. |
 | Block | A reusable high-level orchestration capability such as explore, plan, debug, coding, verify, or review. Blocks compile into Nodes. |
 | Node | A low-level durable unit of child-agent work with dependencies, prompt input, policy, and output contract. |
@@ -31,6 +32,7 @@ Workflow Orchestration turns one user objective into one durable DAG. Its model-
 - Model-facing graph actions expose only `spec_path`; graph fields live in YAML so provider tool-call serialization cannot turn a nested graph into a string.
 - Legacy YAML may be adapted at the file boundary without making legacy fields valid inline input.
 - Runtime Admission and Workflow Authoring Check have separate names, state, and responsibilities.
+- Dependents of a reporting checkpoint must be gated on its output; authoring rejects ungated shapes at start/validate (enforcement point: authoring boundary only, runtime create deliberately unchanged).
 
 ## Boundaries
 
@@ -43,3 +45,5 @@ Workflow Orchestration turns one user objective into one durable DAG. Its model-
 ## Decisions
 
 - [ADR-0001: One Workflow Authoring Check authority](docs/adr/0001-workflow-authoring-check.md)
+- [ADR-0002: Parallel workspace writers with an implementation aggregator](docs/adr/0002-parallel-writers-aggregator.md)
+- [ADR-0003: Reporting checkpoint gating at the authoring boundary](docs/adr/0003-reporting-checkpoint-gating.md)
diff --git a/packages/opencode/src/dag/docs/adr/0003-reporting-checkpoint-gating.md b/packages/opencode/src/dag/docs/adr/0003-reporting-checkpoint-gating.md
new file mode 100644
index 0000000000..41b1e50b6d
--- /dev/null
+++ b/packages/opencode/src/dag/docs/adr/0003-reporting-checkpoint-gating.md
@@ -0,0 +1,70 @@
+# ADR-0003: Reporting checkpoint gating at the authoring boundary
+
+- Status: Accepted
+- Date: 2026-08-17
+
+## Context
+
+On 2026-08-17 a hand-authored 15-node workflow (issue #320) ran 75 minutes to
+`completed` although every one of its five decision checkpoints returned
+`verdict: replan`. The checkpoint nodes carried `report_to_parent: true` but
+their stage dependents declared only `depends_on`, no `condition`. The engine
+spawns a dependent the moment all of its dependencies complete, so each next
+stage started milliseconds (≈12ms) after its checkpoint settled; the wake to
+the parent was terminal-only advisory signal, never a gate. The authoring
+model ignored warning-level feedback, which is how the ungated shape shipped.
+
+Block-compiled graphs already gate dependents on checkpoint verdicts (the
+issue #294 REJECT-checkpoint shape); hand-built node graphs had no equivalent
+check.
+
+## Decision
+
+A `report_to_parent: true` node with dependents is a **reporting checkpoint**.
+Each dependent must gate on the checkpoint's output via `condition`
+(`input_mapping` does not count — it feeds data, it does not gate), or the
+checkpoint must be a reporting leaf, or the node must drop `report_to_parent`.
+`node_defaults.report_to_parent` is honored: a node inheriting the default
+reports the same way.
+
+Enforcement lives in `checkpointGateDiagnostics`, wired only into
+`validatePostCompile`'s structural branch — the authoring start/validate path.
+Every ungated dependent emits one error-severity `dag.invalid` diagnostic in
+both `portable` and `environment` profiles, so `start` and `validate` reject
+the shape before any durable graph exists.
+
+Enforcement is authoring-only by design. `Dag.create` and the replan/extend
+fragment paths stay untouched: the verdict vocabulary is open, the ACCEPT path
+must not wait for the parent, and runtime enforcement would change the
+semantics of every existing graph, including issue #294's wake-chain and
+reopen-extend behavior.
+
+## Consequences
+
+- Ungated reporting checkpoints fail fast at start/validate with a diagnostic
+  naming the checkpoint, the dependent, and the three legal fixes.
+- Runtime create, wake chains, and reopen-extend semantics are unchanged;
+  trusted internal callers retain full runtime flexibility.
+- Saved and curated workflows were audited: 14 curated block workflows are
+  unaffected; only `ultra-flow-route.yaml` and `release-route.yaml` trip the
+  new check and are tracked in opencode-dag-config#14.
+
+## Alternatives Considered
+
+- Runtime enforcement at `Dag.create`: rejected — the verdict vocabulary is
+  open-ended, the ACCEPT path must not block waiting for the parent, and it
+  would change the behavior of every existing graph.
+- Warning-severity diagnostic: rejected — the authoring model ignores
+  warnings; that is precisely how the incident happened.
+- A new explicit `gate` field on dependents: rejected — `condition` already
+  expresses output gating and the block compiler already emits it; a second
+  mechanism would split the gating vocabulary.
+
+## Deferred
+
+- Replan/extend fragments are not checkpoint-gate-checked (coverage gap; no
+  date). Runtime flexibility was prioritized; fragment authoring remains
+  advisory.
+- Deprecation of advisory wake chains (no date): `report_to_parent` without
+  gated dependents stays legal but is a smell worth revisiting once fragment
+  coverage exists.
diff --git a/packages/opencode/src/dag/validation.ts b/packages/opencode/src/dag/validation.ts
index d1126e41bf..c92fae8847 100644
--- a/packages/opencode/src/dag/validation.ts
+++ b/packages/opencode/src/dag/validation.ts
@@ -575,6 +575,38 @@ function conditionDiagnostics(nodes: readonly NodeConfig[]): Diagnostic[] {
   ]
 }
 
+/** A report_to_parent node wakes the parent for adjudication; a dependent
+ * without a condition on that node's output is spawned the moment the
+ * checkpoint completes, so the checkpoint verdict can never act first.
+ * Block-compiled graphs gate dependents on the verdict (issue #294
+ * REJECT-checkpoint shape); hand-built node graphs must do the same or keep
+ * the checkpoint as a reporting leaf. */
+export function checkpointGateDiagnostics(
+  nodes: readonly NodeConfig[],
+  defaults?: { readonly report_to_parent?: boolean },
+): Diagnostic[] {
+  const reportsToParent = (node: NodeConfig) =>
+    node.report_to_parent ?? defaults?.report_to_parent ?? DEFAULT_WORKFLOW_CONFIG.reportToParent
+  return nodes.flatMap((checkpoint) => {
+    if (!reportsToParent(checkpoint)) return []
+    return nodes
+      .filter((dependent) => dependent.depends_on.includes(checkpoint.id))
+      .filter((dependent) => conditionReference(dependent.condition) !== checkpoint.id)
+      .map((dependent) =>
+        diagnostic({
+          code: DIAGNOSTIC_CODES.dagInvalid,
+          path: `nodes[${dependent.id}].condition`,
+          message:
+            `reporting checkpoint "${checkpoint.id}" has dependent "${dependent.id}" that is not gated on its output`
+            + ` — the engine spawns "${dependent.id}" as soon as "${checkpoint.id}" completes, so the checkpoint verdict cannot be acted on first`,
+          hint:
+            `Gate "${dependent.id}" with condition: "${checkpoint.id}.output.<field> == ..." (e.g. on its verdict),`
+            + ` keep "${checkpoint.id}" a reporting leaf, or set report_to_parent: false on "${checkpoint.id}" if downstream must run unconditionally`,
+        }),
+      )
+  })
+}
+
 function bindingDiagnostics(nodes: readonly NodeConfig[]): Diagnostic[] {
   return templateBindingErrors(nodes).map((error) =>
     diagnostic({
@@ -926,7 +958,7 @@ export function validatePostCompile(input: {
   config: {
     mode?: ExecutionMode
     max_total_nodes?: number
-    node_defaults?: { required?: boolean; model?: { modelID: string; providerID: string } }
+    node_defaults?: { required?: boolean; report_to_parent?: boolean; model?: { modelID: string; providerID: string } }
   }
   nodes: readonly NodeConfig[]
   /** The original blocks when the graph used the high-level interface. */
@@ -942,11 +974,14 @@ export function validatePostCompile(input: {
     const diagnostics =
       input.structural === false
         ? []
-        : structuralDiagnostics({
-            nodes: input.nodes,
-            mode: input.config.mode,
-            max_total_nodes: input.config.max_total_nodes,
-          })
+        : [
+            ...structuralDiagnostics({
+              nodes: input.nodes,
+              mode: input.config.mode,
+              max_total_nodes: input.config.max_total_nodes,
+            }),
+            ...checkpointGateDiagnostics(input.nodes, input.config.node_defaults),
+          ]
     if (input.profile === "portable") diagnostics.push(...nonportablePromptDiagnostics(input.nodes))
     if (input.profile === "environment") {
       diagnostics.push(
diff --git a/packages/opencode/test/dag/dag-checkpoint-gate.test.ts b/packages/opencode/test/dag/dag-checkpoint-gate.test.ts
new file mode 100644
index 0000000000..18fcebecf8
--- /dev/null
+++ b/packages/opencode/test/dag/dag-checkpoint-gate.test.ts
@@ -0,0 +1,139 @@
+import { expect } from "bun:test"
+import { Effect } from "effect"
+import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
+import { WorkflowAuthoring } from "../../src/dag/authoring"
+import { testEffect } from "../lib/effect"
+
+const it = testEffect(CrossSpawnSpawner.defaultLayer)
+
+// A report_to_parent (wake-eligible) checkpoint hands its verdict to the
+// parent. When a dependent lacks a condition on that checkpoint's output, the
+// engine spawns it the moment the checkpoint completes and the verdict can
+// never act first — exactly the shape that ran a 6-stage "ultra flow" to
+// completion after every decision checkpoint returned replan. Block-compiled
+// graphs gate dependents on the verdict (issue #294 REJECT-checkpoint shape);
+// hand-built node graphs must do the same.
+
+function spec(config: Record<string, unknown>) {
+  return { title: "checkpoint gate", config }
+}
+
+function checkpoint(id: string) {
+  return {
+    id,
+    name: id,
+    worker_type: "general",
+    depends_on: [],
+    report_to_parent: true,
+    prompt_template: { inline: id },
+    output_schema: {
+      type: "object",
+      properties: { verdict: { type: "string" } },
+      required: ["verdict"],
+    },
+  }
+}
+
+function stage(id: string, dependsOn: string[], condition?: string) {
+  return {
+    id,
+    name: id,
+    worker_type: "build",
+    depends_on: dependsOn,
+    prompt_template: { inline: id },
+    ...(condition ? { condition } : {}),
+  }
+}
+
+function validate(value: unknown) {
+  return WorkflowAuthoring.make().prepare({
+    action: "start",
+    source: { kind: "inline", value, source: "<test>" },
+  })
+}
+
+it.effect("rejects a reporting checkpoint whose dependent is not gated on its output", () =>
+  Effect.gen(function* () {
+    const result = yield* validate(
+      spec({
+        name: "ungated-checkpoint",
+        nodes: [checkpoint("cp-design-decision"), stage("stage-development", ["cp-design-decision"])],
+      }),
+    )
+    expect(result.valid).toBe(false)
+    expect(result.errors.some((d) => d.message.includes('"cp-design-decision"') && d.message.includes('"stage-development"'))).toBe(true)
+  }),
+)
+
+it.effect("accepts a dependent gated by a condition on the checkpoint output", () =>
+  Effect.gen(function* () {
+    const result = yield* validate(
+      spec({
+        name: "gated-checkpoint",
+        nodes: [
+          checkpoint("cp-design-decision"),
+          stage("stage-development", ["cp-design-decision"], 'cp-design-decision.output.verdict == "continue"'),
+        ],
+      }),
+    )
+    expect(result.errors.filter((d) => d.message.includes("not gated"))).toEqual([])
+  }),
+)
+
+it.effect("accepts a reporting checkpoint as a leaf", () =>
+  Effect.gen(function* () {
+    const result = yield* validate(
+      spec({
+        name: "leaf-checkpoint",
+        nodes: [stage("stage-design", []), { ...checkpoint("cp-after-design"), depends_on: ["stage-design"] }],
+      }),
+    )
+    expect(result.errors.filter((d) => d.message.includes("not gated"))).toEqual([])
+  }),
+)
+
+it.effect("accepts an ungated dependent when the checkpoint does not report to parent", () =>
+  Effect.gen(function* () {
+    const result = yield* validate(
+      spec({
+        name: "quiet-node",
+        nodes: [{ ...checkpoint("analysis"), report_to_parent: false }, stage("summary", ["analysis"])],
+      }),
+    )
+    expect(result.errors.filter((d) => d.message.includes("not gated"))).toEqual([])
+  }),
+)
+
+it.effect("flags ungated dependents inherited from node_defaults.report_to_parent", () =>
+  Effect.gen(function* () {
+    const result = yield* validate(
+      spec({
+        name: "defaults-inherited",
+        node_defaults: { report_to_parent: true },
+        nodes: [
+          { ...checkpoint("cp"), report_to_parent: undefined },
+          stage("after", ["cp"]),
+        ],
+      }),
+    )
+    expect(result.valid).toBe(false)
+    expect(result.errors.some((d) => d.message.includes('"cp"') && d.message.includes('"after"'))).toBe(true)
+  }),
+)
+
+it.effect("flags a condition that gates a different dependency than the checkpoint", () =>
+  Effect.gen(function* () {
+    const result = yield* validate(
+      spec({
+        name: "wrong-gate",
+        nodes: [
+          checkpoint("cp-design-decision"),
+          stage("stage-design", []),
+          stage("stage-development", ["cp-design-decision", "stage-design"], 'stage-design.output.ready == "yes"'),
+        ],
+      }),
+    )
+    expect(result.valid).toBe(false)
+    expect(result.errors.some((d) => d.message.includes('"cp-design-decision"') && d.message.includes('"stage-development"'))).toBe(true)
+  }),
+)

From 938a197737115aec3182ede4feecc837cdeefb7d Mon Sep 17 00:00:00 2001
From: Lex <sunsan05@Gmail.com>
Date: Mon, 17 Aug 2026 22:21:01 +0800
Subject: [PATCH 2/7] fix(dag): mark wake reported at admit time (issue #321)

wake_reported landed only after the whole wake-driven turn completed, so a
restart or mid-turn interruption left the synthetic part durable but the mark
absent, and the startup sweep re-injected a byte-identical wake. Mark the batch
(and unregister terminal workflows) immediately after admitIfIdle, before
awaiting the turn; drop the turn-failure-redelivery semantics. Mark failures
degrade to retry-later while the admitted turn still runs.

Flips the two dag-wake-integration failure-redelivery assertions and reworks
the two timeout-escalation Q2-gate tests (their "held turn => wakeReported=false"
window no longer exists; they now hold the wake via a busy parent session).
Adds regression tests for failed-turn no-redelivery and a simulated restart.
---
 packages/opencode/src/dag/runtime/loop.ts     | 120 ++++++----
 .../test/dag/dag-timeout-escalation.test.ts   |  53 +++--
 .../test/dag/dag-wake-integration.test.ts     | 218 +++++++++++++++++-
 3 files changed, 321 insertions(+), 70 deletions(-)

diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts
index a725cdd563..938d27604a 100644
--- a/packages/opencode/src/dag/runtime/loop.ts
+++ b/packages/opencode/src/dag/runtime/loop.ts
@@ -74,16 +74,15 @@ const serviceLayer = Layer.effect(
         const recovering = new Set<string>()
         const wakeInFlight = new Set<string>()
         const wakePending = new Set<string>()
-        // GOAL-FP-01-14: per-session record of the last wake summary whose
-        // transcript part was written. The durable mark runs AFTER the write
-        // (at-least-once delivery: a mark failure keeps the batch unreported
-        // for a retry), so a retry of an already-written summary would
-        // re-inject the same digest into the transcript. The retry dedupes on
-        // this map and only re-marks. In-process only — a crash between write
-        // and mark still duplicates on the restart sweep (a durable
-        // delivering-marker would need a schema change; registered, see
-        // GOAL-FP-01-14). Capped: evicting entries degrades to the pre-fix
-        // duplicate visibility, never to a lost wake.
+        // Per-session record of the last wake summary whose transcript part was
+        // written. Admit success IS the delivery (issue #321): the durable
+        // wake_reported mark lands immediately after the write, not after the
+        // wake-driven turn completes, so the in-memory map only needs to dedupe
+        // the narrow mark-retry path — if the leased mark returns None the rows
+        // stay unreported and a later trigger re-marks WITHOUT re-prompting.
+        // In-process only; restart durability now comes from wake_reported being
+        // persisted at admit time, not from this map. Capped: evicting entries
+        // degrades to a redundant re-prompt, never to a lost wake.
         const deliveredWakeSummaries = new Map<string, string>()
 
         // Seed the commented global dag.jsonc once per instance init — the
@@ -1315,24 +1314,30 @@ const serviceLayer = Layer.effect(
               )
               if (!wakeLease) return
 
-              // Persist wake_reported AFTER successful delivery only.
-              // A failure stays durable for a later idle event or restart scan;
-              // it must not spin synchronously on the same row.
               // The part is marked synthetic: model-visible (the orchestrator
               // receives the node result and can act) but NOT rendered as a user
               // message in the TUI chat — DAG data surfaces via the sidebar panel
               // and Inspector, keeping the chat conversation clean.
               //
-              // GOAL-FP-01-14: the transcript part is written BEFORE the
-              // durable mark. A mark failure (or a crash between the two)
-              // leaves the batch unreported and the retry would re-inject the
-              // SAME summary. When this session already had this exact summary
-              // written, skip the prompt and only re-mark — the write is
-              // idempotent in effect because an identical digest adds no
-              // information. A differing summary (new results committed
-              // between attempts) always prompts.
+              // ADMIT SUCCESS == DELIVERED (issue #321). The previous contract
+              // (GOAL-FP-01-14) persisted wake_reported only AFTER the whole
+              // wake-driven parent turn completed. A restart or mid-turn
+              // interruption therefore left wake_reported=false while the
+              // synthetic part was already durable in transcript, so the startup
+              // sweep re-injected a byte-identical wake (real incident: the same
+              // 4391-char wake injected twice, ~10 min apart, after a TUI
+              // restart). Redelivery adds duplicates, never information. The
+              // mark (and the terminal-workflow unregisters) now land right
+              // after admitIfIdle admits the part, BEFORE awaiting the turn.
+              //
+              // The in-memory dedup map still guards the retry path: if the
+              // leased mark below returns None (generation/owner changed), the
+              // rows stay unreported and a later trigger re-marks without
+              // re-prompting. A differing summary (new results committed between
+              // attempts) always prompts.
               if (deliveredWakeSummaries.size > 1024) deliveredWakeSummaries.clear()
               const didDeliver = yield* Effect.gen(function* () {
+                let wakeTurn: Effect.Effect<SessionV1.WithParts> | undefined
                 if (deliveredWakeSummaries.get(sessionID) !== summary) {
                   const delivered = yield* SessionPrompt.admitIfIdle(promptSvc, automation, wakeLease, {
                     sessionID: SessionID.make(sessionID),
@@ -1340,28 +1345,63 @@ const serviceLayer = Layer.effect(
                   })
                   if (Option.isNone(delivered)) return false
                   deliveredWakeSummaries.set(sessionID, summary)
-                  yield* delivered.value.pipe(
-                    Effect.onError(() =>
-                      Effect.sync(() => {
-                        if (deliveredWakeSummaries.get(sessionID) === summary) {
-                          deliveredWakeSummaries.delete(sessionID)
-                        }
-                      }),
-                    ),
-                  )
+                  wakeTurn = delivered.value
                 }
 
-                const markLease = yield* automation.claim(SessionID.make(sessionID), { kind: "dag" })
-                if (Option.isNone(markLease)) return false
-                const marked = yield* automation.use(markLease.value, store.markWakeBatchReported(batch))
-                if (Option.isNone(marked)) return false
-                plan.unresponsiveDagIDs.forEach((workflowID) => deliveredUnresponsiveDagIDs.add(workflowID))
-                yield* Effect.forEach(
-                  batch.workflows.filter((workflow) => isWorkflowTerminalStatus(workflow.status as never)),
-                  (workflow) => automation.unregister(SessionID.make(sessionID), { kind: "dag", id: workflow.id }),
-                  { discard: true },
+                // Admit success == delivered (issue #321): persist wake_reported
+                // at admit time. A leased mark returning None (generation/owner
+                // raced) is treated as retry-later — the rows stay unreported and
+                // a later trigger re-marks — but the admitted turn still runs
+                // below (its end-of-turn idle is what re-arms that retry).
+                const markLease = Option.getOrUndefined(
+                  yield* automation.claim(SessionID.make(sessionID), { kind: "dag" }),
                 )
-                return true
+                // Any mark failure (lease lost, generation raced, or the store
+                // write dying) degrades to retry-later instead of propagating:
+                // the rows stay unreported and a later trigger re-marks, while
+                // the admitted turn below still runs (its end-of-turn idle is
+                // what re-arms that retry). Only interruption propagates.
+                const markSucceeded = markLease
+                  ? Option.isSome(
+                      yield* automation.use(markLease, store.markWakeBatchReported(batch)).pipe(
+                        Effect.catchCause((cause) =>
+                          Cause.hasInterrupts(cause)
+                            ? Effect.failCause(cause)
+                            : Effect.logWarning("DAG wake batch mark failed; rows stay unreported for retry", {
+                                sessionID,
+                                cause: Cause.pretty(cause),
+                              }).pipe(Effect.as(Option.none())),
+                        ),
+                      ),
+                    )
+                  : false
+                if (markSucceeded) {
+                  plan.unresponsiveDagIDs.forEach((workflowID) => deliveredUnresponsiveDagIDs.add(workflowID))
+                  yield* Effect.forEach(
+                    batch.workflows.filter((workflow) => isWorkflowTerminalStatus(workflow.status as never)),
+                    (workflow) => automation.unregister(SessionID.make(sessionID), { kind: "dag", id: workflow.id }),
+                    { discard: true },
+                  )
+                }
+
+                // Pacing — keep one wake turn at a time. The turn runs AFTER the
+                // durable mark, so its failure can no longer lose the report;
+                // swallow non-interrupt failures instead of surfacing them as a
+                // delivery failure. It also runs when the mark raced, so its
+                // end-of-turn idle re-arms the mark retry.
+                if (wakeTurn) {
+                  yield* wakeTurn.pipe(
+                    Effect.catchCause((cause) =>
+                      Cause.hasInterrupts(cause)
+                        ? Effect.failCause(cause)
+                        : Effect.logInfo("DAG wake turn did not complete; wake already reported", {
+                            sessionID,
+                            cause: Cause.pretty(cause),
+                          }),
+                    ),
+                  )
+                }
+                return markSucceeded
               }).pipe(
                 Effect.catchCause((cause) =>
                   Effect.logWarning("DAG wake delivery failed", { sessionID, cause: Cause.pretty(cause) }).pipe(
diff --git a/packages/opencode/test/dag/dag-timeout-escalation.test.ts b/packages/opencode/test/dag/dag-timeout-escalation.test.ts
index f8c88f46a8..0167ab318d 100644
--- a/packages/opencode/test/dag/dag-timeout-escalation.test.ts
+++ b/packages/opencode/test/dag/dag-timeout-escalation.test.ts
@@ -195,6 +195,7 @@ function runLoopTest<A>(
   test: (services: {
     readonly dag: Dag.Interface
     readonly store: DagStore.Interface
+    readonly status: SessionStatus.Interface
     readonly childPrompts: Queue.Queue<PromptGate>
     readonly parentPrompts: Queue.Queue<ParentPromptGate>
     readonly getCancelCount: () => number
@@ -212,6 +213,7 @@ function runLoopTest<A>(
       const dag = yield* Dag.Service
       const loop = yield* DagLoop.Service
       const store = yield* DagStore.Service
+      const status = yield* SessionStatus.Service
       const database = yield* Database.Service
       yield* database.db.insert(ProjectTable).values({
         id: Project.ID.make("project-1"),
@@ -230,6 +232,7 @@ function runLoopTest<A>(
       return yield* test({
         dag,
         store,
+        status,
         childPrompts,
         parentPrompts,
         getCancelCount: harness.getCancelCount,
@@ -1044,7 +1047,7 @@ describe("DagLoop timeout escalation", () => {
   //   P5  escalationPending ∧ wakeReported          → proceed                [recovery — test below + L880/L567]
   it("blocks re-time while the escalation wake is undelivered (Q2: escalationPending ∧ ¬wakeReported ⇒ skip)", async () => {
     await Effect.runPromise(
-      runLoopTest(({ dag, store, childPrompts, parentPrompts }) =>
+      runLoopTest(({ dag, store, status, childPrompts, parentPrompts }) =>
         Effect.gen(function* () {
           const dagID = yield* dag.create({
             projectID: "project-1",
@@ -1054,10 +1057,16 @@ describe("DagLoop timeout escalation", () => {
           })
           yield* takeWithin(childPrompts, "a did not start")
 
-          // Acceptance #2: the deadline-driven INITIAL escalation (watchdog →
-          // nodeTimeoutEscalated → first wake) is NOT touched by the gate —
-          // the gate only governs the replan re-time path. It fires and its
-          // wake reaches the parent.
+          // issue #321: wake_reported now lands at ADMIT time, so an in-flight
+          // turn no longer holds wake_reported=false (the wake is reported as
+          // soon as it is admitted). The only window that keeps the escalation
+          // wake UNDELIVERED now is a BUSY parent session — the idle-gate never
+          // admits it. Hold the wake there.
+          yield* status.set(SessionID.make("ses_parent"), { type: "busy" })
+
+          // Initial escalation fires; while the parent is busy its wake is held
+          // UNDELIVERED (never admitted). The node sits at the public-path state
+          // [escalationPending ∧ ¬wakeReported ∧ deadline≤now].
           const escalated = yield* pollWithTimeout(
             store.getNode(dagID, "a").pipe(
               Effect.map((current) => current?.timeoutExtensions === 1 ? current : undefined),
@@ -1066,17 +1075,10 @@ describe("DagLoop timeout escalation", () => {
           )
           expect(escalated.status).toBe("running")
           expect(escalated.escalationPending).toBe(true)
+          expect(escalated.wakeReported).toBe(false)
           const baselineDeadline = escalated.deadlineMs
-          const timeoutWake = yield* takeWithin(parentPrompts, "initial escalation wake did not reach the parent")
-          expect(timeoutWake.text).toContain("[DAG Node Timeout]")
-
-          // Hold the wake UNDELIVERED: the harness blocks delivery on the
-          // release Deferred, and the loop persists wake_reported=true only
-          // AFTER successful delivery (loop.ts:1125). The node sits at the
-          // public-path state [escalationPending ∧ ¬wakeReported ∧ deadline≤now].
-          const undelivered = yield* store.getNode(dagID, "a")
-          expect(undelivered?.escalationPending).toBe(true)
-          expect(undelivered?.wakeReported).toBe(false)
+          // The wake is held by the busy gate — nothing reached the parent yet.
+          expect(Option.isNone(yield* Queue.poll(parentPrompts))).toBe(true)
 
           // Main agent replans with a NEW timeout. Q2 must SKIP the re-time:
           // adjudication cannot land before the escalation wake was delivered.
@@ -1106,7 +1108,11 @@ describe("DagLoop timeout escalation", () => {
           expect(reEscalated.wakeReported).toBe(false)
           expect(reEscalated.deadlineMs).toBe(baselineDeadline)
 
-          // Release the held wake so the loop marks delivery before teardown.
+          // Release the held wake: the parent goes idle, the wake is admitted
+          // (wake_reported lands at admit, issue #321), and the turn settles.
+          yield* status.set(SessionID.make("ses_parent"), { type: "idle" })
+          const timeoutWake = yield* takeWithin(parentPrompts, "escalation wake did not reach the parent once idle")
+          expect(timeoutWake.text).toContain("[DAG Node Timeout]")
           yield* Deferred.succeed(timeoutWake.release, "success")
         }),
       ),
@@ -1179,7 +1185,7 @@ describe("DagLoop timeout escalation", () => {
   // terminal-cleanup path.
   it("C1: nodeExtendTimeout distinguishes Q2 rejection (-2) from terminal rejection (0) — three-valued contract", async () => {
     await Effect.runPromise(
-      runLoopTest(({ dag, store, childPrompts, parentPrompts }) =>
+      runLoopTest(({ dag, store, status, childPrompts, parentPrompts }) =>
         Effect.gen(function* () {
           const dagID = yield* dag.create({
             projectID: "project-1",
@@ -1189,12 +1195,15 @@ describe("DagLoop timeout escalation", () => {
           })
           const gate = yield* takeWithin(childPrompts, "a did not start")
 
-          // Escalation fires; the wake is held UNDELIVERED so the node sits at
-          // the Q2 state [escalationPending ∧ ¬wakeReported ∧ running].
+          // issue #321: wake_reported lands at ADMIT time, so the Q2 state
+          // [escalationPending ∧ ¬wakeReported] can no longer be held by an
+          // in-flight turn — hold the escalation wake UNDELIVERED via the busy
+          // idle-gate instead, which keeps it never admitted.
+          yield* status.set(SessionID.make("ses_parent"), { type: "busy" })
           const escalated = yield* pollWithTimeout(
             store.getNode(dagID, "a").pipe(
               Effect.map((current) =>
-                current?.timeoutExtensions === 1 && current.escalationPending && !current.wakeReported
+                current && current.status === "running" && current.escalationPending && !current.wakeReported
                   ? current
                   : undefined,
               ),
@@ -1213,7 +1222,9 @@ describe("DagLoop timeout escalation", () => {
           expect(afterQ2?.status).toBe("running")
           expect(afterQ2?.deadlineMs).toBe(frozenDeadline)
 
-          // Deliver the wake and let the child finish — the node terminalizes.
+          // Release the held wake (idle admits it) and let the child finish —
+          // the node terminalizes.
+          yield* status.set(SessionID.make("ses_parent"), { type: "idle" })
           const wake = yield* takeWithin(parentPrompts, "escalation wake did not reach the parent")
           yield* Deferred.succeed(wake.release, "success")
           yield* Deferred.succeed(gate.release, "done")
diff --git a/packages/opencode/test/dag/dag-wake-integration.test.ts b/packages/opencode/test/dag/dag-wake-integration.test.ts
index 76039be7da..560e48e457 100644
--- a/packages/opencode/test/dag/dag-wake-integration.test.ts
+++ b/packages/opencode/test/dag/dag-wake-integration.test.ts
@@ -9,8 +9,12 @@ 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 { Project } from "@opencode-ai/core/project"
 import { ProjectTable } from "@opencode-ai/core/project/sql"
+import { AbsolutePath } from "@opencode-ai/core/schema"
 import { SessionTable } from "@opencode-ai/core/session/sql"
+import { Model } from "@opencode-ai/schema/model"
+import { Provider } from "@opencode-ai/schema/provider"
 import { Agent } from "@/agent/agent"
 import { fingerprintBrief } from "@/dag/admission"
 import { Dag, type NodeConfig } from "@/dag/dag"
@@ -1028,7 +1032,11 @@ describe("DagLoop atomic wake integration", () => {
     )
   })
 
-  it("leaves the whole batch unreported when parent delivery fails", async () => {
+  // FLIPPED for issue #321: previously a failed parent turn left the whole
+  // batch unreported (for later redelivery). Admit success now IS the
+  // delivery — the mark lands at admit time, so a failed/interrupted turn
+  // still leaves the batch reported.
+  it("reports the wake batch at admit time even when the parent turn then fails (issue #321)", async () => {
     await Effect.runPromise(
       runWakeTest(({ dag, store, childPrompts, parentPrompts, parentSettled }) =>
         Effect.gen(function* () {
@@ -1044,14 +1052,20 @@ describe("DagLoop atomic wake integration", () => {
           yield* Deferred.succeed(parent.release, "failure")
           yield* takeWithin(parentSettled, "failed parent prompt did not settle")
 
-          expect(yield* store.getUnreportedWakeNodes("ses_parent")).toHaveLength(1)
-          expect(yield* store.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(1)
+          // The synthetic part is durable in transcript either way; marking at
+          // admit time means a restart or mid-turn interruption has nothing to
+          // re-inject.
+          expect(yield* store.getUnreportedWakeNodes("ses_parent")).toHaveLength(0)
+          expect(yield* store.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(0)
         }),
       ),
     )
   })
 
-  it("retries the parent prompt after a provider failure instead of silently marking the wake", async () => {
+  // FLIPPED for issue #321: previously a failed parent turn was retried — a
+  // fresh idle event re-injected the SAME wake. Admit success is now the
+  // delivery, so a later trigger must inject NO duplicate prompt.
+  it("does not redeliver a wake whose parent turn failed (issue #321)", async () => {
     await Effect.runPromise(
       runWakeTest(({ dag, store, status, childPrompts, parentPrompts, parentSettled }) =>
         Effect.gen(function* () {
@@ -1066,14 +1080,16 @@ describe("DagLoop atomic wake integration", () => {
           const first = yield* takeWithin(parentPrompts, "retryable batch did not wake the parent")
           yield* Deferred.succeed(first.release, "failure")
           yield* takeWithin(parentSettled, "failed parent prompt did not settle")
-          yield* status.set(SessionID.make("ses_parent"), { type: "idle" })
 
-          const second = yield* takeWithin(parentPrompts, "failed provider wake was not prompted again")
-          expect(promptText(second.input)).toContain('Node "retryable-node" completed: retry me')
-          yield* Deferred.succeed(second.release, "success")
-          yield* takeWithin(parentSettled, "successful retry did not settle")
+          // The batch is reported at admit time even though the turn failed.
           expect(yield* store.getUnreportedWakeNodes("ses_parent")).toHaveLength(0)
           expect(yield* store.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(0)
+
+          // Re-trigger the delivery path: the idle gate must NOT inject a
+          // duplicate prompt (pre-fix this re-delivered the identical wake).
+          yield* status.set(SessionID.make("ses_parent"), { type: "idle" })
+          yield* Effect.sleep("500 millis")
+          expect(Option.isNone(yield* Queue.poll(parentPrompts))).toBe(true)
         }),
       ),
     )
@@ -1123,6 +1139,190 @@ describe("DagLoop atomic wake integration", () => {
     )
   })
 
+  // NEW for issue #321: simulates the production restart. A wake is admitted
+  // but its parent turn NEVER finishes (the process dies mid-turn). Pre-fix the
+  // mark only landed after the turn completed, so the restart sweep saw
+  // wake_reported=false and re-injected the byte-identical wake. Post-fix the
+  // mark lands at admit time, so a fresh loop's startup sweep delivers nothing.
+  it("does not redeliver an already-admitted wake across a restart (issue #321)", async () => {
+    await Effect.runPromise(
+      Effect.gen(function* () {
+        const childTitles = new Map<string, string>()
+        const created: string[] = []
+        const session = Layer.mock(Session.Service, {
+          get: () =>
+            Effect.succeed({
+              id: SessionID.make("ses_parent"),
+              slug: "parent",
+              projectID: Project.ID.make("project-1"),
+              directory: process.cwd(),
+              title: "Parent",
+              version: "test",
+              time: { created: 0, updated: 0 },
+              permission: [],
+              agent: "build",
+            }),
+          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: SessionID.make(id),
+                slug: "child",
+                projectID: Project.ID.make("project-1"),
+                directory: process.cwd(),
+                title: value?.title ?? id,
+                version: "test",
+                time: { created: 0, updated: 0 },
+              }
+            }),
+          messages: () => Effect.succeed([]),
+        })
+        const agent = Layer.mock(Agent.Service, {
+          get: () =>
+            Effect.succeed({
+              name: "build",
+              mode: "all",
+              permission: [],
+              options: {},
+              description: "",
+              prompt: "",
+              model: { providerID: Provider.ID.make("test"), modelID: Model.ID.make("test-model") },
+              tools: {},
+              hooks: {},
+            }),
+        })
+        const deliver = (queues: {
+          readonly childPrompts: Queue.Queue<PromptGate>
+          readonly parentPrompts: Queue.Queue<ParentPromptGate>
+          readonly parentSettled: Queue.Queue<void>
+        }) =>
+          Effect.fn("test.SessionPrompt.deliver")(function* (value: SessionPrompt.PromptInput) {
+            const sessionID = value.sessionID as string
+            if (sessionID === "ses_parent") {
+              const release = yield* Deferred.make<"success" | "failure">()
+              yield* Queue.offer(queues.parentPrompts, { input: value, release })
+              const outcome = yield* Deferred.await(release).pipe(
+                Effect.ensuring(Queue.offer(queues.parentSettled, undefined)),
+              )
+              if (outcome === "failure") return yield* Effect.die(new Error("provider unavailable"))
+              return reply(sessionID, "parent handled wake")
+            }
+            const release = yield* Deferred.make<string>()
+            yield* Queue.offer(queues.childPrompts, {
+              title: childTitles.get(sessionID) ?? sessionID,
+              input: value,
+              release,
+            })
+            return reply(sessionID, yield* Deferred.await(release))
+          })
+        const promptLayer = (queues: {
+          readonly childPrompts: Queue.Queue<PromptGate>
+          readonly parentPrompts: Queue.Queue<ParentPromptGate>
+          readonly parentSettled: Queue.Queue<void>
+        }) => Layer.mock(SessionPrompt.Service, withIdleAdmission({
+          cancel: () => Effect.void,
+          prompt: deliver(queues),
+          promptIfIdle: (value) => deliver(queues)(value).pipe(Effect.map(Option.some)),
+        }))
+
+        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)
+
+        yield* Effect.gen(function* () {
+          const storeSvc = yield* DagStore.Service
+          const databaseSvc = yield* Database.Service
+          yield* databaseSvc.db.insert(ProjectTable).values({
+            id: Project.ID.make("project-1"),
+            worktree: AbsolutePath.make(process.cwd()),
+            sandboxes: [],
+          }).run().pipe(Effect.orDie)
+          yield* databaseSvc.db.insert(SessionTable).values({
+            id: SessionID.make("ses_parent"),
+            project_id: Project.ID.make("project-1"),
+            slug: "parent",
+            directory: AbsolutePath.make(process.cwd()),
+            title: "Parent",
+            version: "test",
+          }).run().pipe(Effect.orDie)
+
+          // Phase 1: admit the wake, then "die" before the parent turn finishes.
+          const q1 = {
+            childPrompts: yield* Queue.unbounded<PromptGate>(),
+            parentPrompts: yield* Queue.unbounded<ParentPromptGate>(),
+            parentSettled: yield* Queue.unbounded<void>(),
+          }
+          yield* Effect.scoped(
+            Effect.gen(function* () {
+              const dagSvc = yield* Dag.Service
+              const loopSvc = yield* DagLoop.Service
+              yield* loopSvc.init()
+              yield* dagSvc.create({
+                projectID: "project-1",
+                sessionID: "ses_parent",
+                title: "Restart wake",
+                config: { name: "restart-wake", nodes: [node("restart-node")] },
+              })
+              const child = yield* takeWithin(q1.childPrompts, "restart node did not start")
+              yield* Deferred.succeed(child.release, "done")
+              // The wake was admitted (mark landed at admit). Do NOT release the
+              // parent turn — disposing the scope simulates a restart mid-turn.
+              yield* takeWithin(q1.parentPrompts, "terminal workflow did not wake the parent")
+            }).pipe(Effect.provide(DagLoop.layer.pipe(
+              Layer.provide(session),
+              Layer.provide(promptLayer(q1)),
+              Layer.provide(agent),
+            ))),
+          )
+
+          // The durable rows are already reported even though the turn never ran.
+          expect(yield* storeSvc.getUnreportedWakeNodes("ses_parent")).toHaveLength(0)
+          expect(yield* storeSvc.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(0)
+          expect(yield* storeSvc.getSessionsWithUnreportedWakes()).toHaveLength(0)
+
+          // Phase 2: a fresh loop over the SAME store runs the startup sweep.
+          const q2 = {
+            childPrompts: yield* Queue.unbounded<PromptGate>(),
+            parentPrompts: yield* Queue.unbounded<ParentPromptGate>(),
+            parentSettled: yield* Queue.unbounded<void>(),
+          }
+          yield* Effect.scoped(
+            Effect.gen(function* () {
+              const loopSvc = yield* DagLoop.Service
+              yield* loopSvc.init()
+              // Bound the window in which a (now-forbidden) redelivery could appear.
+              yield* Effect.sleep("500 millis")
+              expect(Option.isNone(yield* Queue.poll(q2.parentPrompts))).toBe(true)
+            }).pipe(Effect.provide(DagLoop.layer.pipe(
+              Layer.provide(session),
+              Layer.provide(promptLayer(q2)),
+              Layer.provide(agent),
+            ))),
+          )
+        }).pipe(
+          Effect.provide(base),
+          Effect.provideService(InstanceRef, {
+            directory: process.cwd(),
+            worktree: process.cwd(),
+            project: {
+              id: Project.ID.make("project-1"),
+              worktree: process.cwd(),
+              time: { created: 0, updated: 0 },
+              sandboxes: [],
+            },
+          }),
+        )
+      }).pipe(Effect.scoped),
+    )
+  })
+
   it("keeps a wake unreported while the parent is busy and delivers it on idle", async () => {
     await Effect.runPromise(
       runWakeTest(({ dag, store, status, childPrompts, parentPrompts }) =>

From 032a68340d4866d9df3b137068b40ba581829828 Mon Sep 17 00:00:00 2001
From: Lex <sunsan05@Gmail.com>
Date: Mon, 17 Aug 2026 22:35:56 +0800
Subject: [PATCH 3/7] fix(dag): halt downstream scheduling when a reporting
 checkpoint returns verdict replan

A reporting checkpoint submitting {verdict: "replan"} now pauses the workflow durably before any spawn round, so dependents can never run on the rejected direction; the parent is woken by the existing report_to_parent wake. control(replan) resumes a paused workflow so corrective nodes run, with TOCTOU-safe resume handling and updated pause guidance. continue verdicts and non-verdict outputs advance unchanged. Closes #322
---
 packages/opencode/src/dag/runtime/loop.ts     | 36 ++++++++-
 packages/opencode/src/tool/workflow.ts        | 28 ++++++-
 .../opencode/test/dag/dag-loop-guards.test.ts | 80 +++++++++++++++++++
 .../opencode/test/dag/workflow-tool.test.ts   | 35 ++++++++
 4 files changed, 175 insertions(+), 4 deletions(-)

diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts
index a725cdd563..e0986c9252 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<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)
                   }),
diff --git a/packages/opencode/src/tool/workflow.ts b/packages/opencode/src/tool/workflow.ts
index a592593889..afa66390a4 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: `<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":
diff --git a/packages/opencode/test/dag/dag-loop-guards.test.ts b/packages/opencode/test/dag/dag-loop-guards.test.ts
index b2b4ab1d8f..43d1e35cf4 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 c2fef15d61..97fbca7200 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

From 311091b79f182b961db7ac4d5360605ecb01ad5d Mon Sep 17 00:00:00 2001
From: Lex <sunsan05@Gmail.com>
Date: Mon, 17 Aug 2026 22:35:59 +0800
Subject: [PATCH 4/7] docs(dag): correct stale unregister-ordering comment
 after admit-time wake mark (issue #321)

---
 packages/opencode/src/session/automation-lease.ts | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/packages/opencode/src/session/automation-lease.ts b/packages/opencode/src/session/automation-lease.ts
index 1613608c39..ac848c2cb6 100644
--- a/packages/opencode/src/session/automation-lease.ts
+++ b/packages/opencode/src/session/automation-lease.ts
@@ -101,10 +101,13 @@ export const layer = Layer.effect(
     // GOAL-FP-01-02: when the dag ownership actually disappears (owner
     // transitions dag → goal/none), re-trigger the goal evaluation through
     // the EXISTING idle status event mechanism so a goal that yielded to the
-    // dag on the last idle event gets a fresh evaluation. The final dag
-    // unregister of a wake delivery (U2 in dag/runtime/loop.ts) lands AFTER
-    // the wake turn's idle event — without this re-trigger the goal silently
-    // stalls until the next external idle. This is also the GOAL-FP-01-11
+    // dag on the last idle event gets a fresh evaluation. Since issue #321
+    // the final dag unregister of a wake delivery lands MID-TURN at admit
+    // time (the mark moves with it), so the session is busy when the owner
+    // flips: the idle gate below skips the re-emit here, and the in-flight
+    // wake turn re-emits idle on completion, which re-drives the goal.
+    // Without this re-trigger path the goal silently stalls until the next
+    // external idle. This is also the GOAL-FP-01-11
     // mitigation surface: a claim that lost the ownership race gets another
     // chance once the owner actually transfers.
     //

From 073b71c7bb4964071a7c513325ebbe4a8f53e419 Mon Sep 17 00:00:00 2001
From: Lex <sunsan05@Gmail.com>
Date: Mon, 17 Aug 2026 22:36:26 +0800
Subject: [PATCH 5/7] fix(memory): repair the MEMORY write path for
 thinking-mode providers

Guarantee the literal json token at the MemoryModel seam (providers reject json_object calls without it), size match/maintain output budgets so structured replies survive reasoning, and retire the wall-clock prepare/checkpoint/in-model timeouts that killed active reasoning calls (issue #324 tracks the remaining streaming-liveness debt). Checkpoint maintenance runs in the background with an atomic per-project in-flight guard and commits under fence+lock only; prepare/search semantics unchanged.
---
 packages/opencode/src/memory/memory.ts       | 174 ++++++++++++++-----
 packages/opencode/src/memory/model.ts        |  19 +-
 packages/opencode/test/memory/memory.test.ts |  77 ++++++++
 3 files changed, 220 insertions(+), 50 deletions(-)

diff --git a/packages/opencode/src/memory/memory.ts b/packages/opencode/src/memory/memory.ts
index 6e081178ee..4a33e01c77 100644
--- a/packages/opencode/src/memory/memory.ts
+++ b/packages/opencode/src/memory/memory.ts
@@ -3,7 +3,7 @@ export * as Memory from "./memory"
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { ProjectV2 } from "@opencode-ai/core/project"
 import { SessionV1 } from "@opencode-ai/core/v1/session"
-import { Context, Duration, Effect, Layer, Option, Ref, Schema, Semaphore } from "effect"
+import { Context, Effect, Layer, Option, Ref, Schema, Scope, Semaphore } from "effect"
 import { stringify } from "yaml"
 import { Config } from "@/config/config"
 import { Provider } from "@/provider/provider"
@@ -22,8 +22,10 @@ import { MemoryStore } from "./store"
 
 const EVIDENCE_MESSAGES = 16
 const EVIDENCE_CHARS = 8_000
-const PREPARE_TIMEOUT = Duration.seconds(5)
-const CHECKPOINT_TIMEOUT = Duration.seconds(8)
+// Reasoning-heavy models spend thinking tokens against max_output_tokens; size
+// the budgets so the structured reply survives the thinking phase.
+const MATCH_OUTPUT_TOKENS = 2_048
+const MAINTAIN_OUTPUT_TOKENS = 16_384
 
 type TurnCache = {
   readonly completedTurns: number
@@ -89,6 +91,8 @@ export const layer: Layer.Layer<
     const globalStarted = yield* Ref.make(false)
     const initializationLock = Semaphore.makeUnsafe(1)
     const state = yield* InstanceState.make(() => Effect.succeed({ sessions: new Map<SessionID, SessionCache>() }))
+    const scope = yield* Scope.Scope
+    const maintenanceInFlight = yield* Ref.make(new Set<ProjectV2.ID>())
 
     const availableModels = Effect.fn("Memory.availableModels")(function* () {
       const providers = yield* provider.list()
@@ -265,7 +269,7 @@ export const layer: Layer.Layer<
           topics: MemoryStore.indexes(input.topics),
         }),
         schema: MemorySchema.MatchResponse,
-        maxOutputTokens: 256,
+        maxOutputTokens: MATCH_OUTPUT_TOKENS,
       })
       const decoded = Schema.decodeUnknownOption(MemorySchema.MatchResponse)(output)
       if (Option.isNone(decoded))
@@ -276,15 +280,25 @@ export const layer: Layer.Layer<
         .slice(0, input.config.injection.max_topics)
     })
 
-    const maintain = Effect.fn("Memory.maintain")(function* (input: {
+    // Serialize the identity-liveness recheck and the per-project lock around
+    // the store write only; the model calls that produce the update run
+    // outside the fence/lock so a long reasoning call cannot wedge or leak it.
+    const applyUpdate = (projectID: ProjectV2.ID, update: (topics: MemorySchema.Topic[]) => MemoryStore.Update<undefined>) =>
+      fence.withLiveIdentity(
+        projectID,
+        lock.withProject(projectID)(store.updateTopics(projectID, update)),
+      )
+
+    // Model-only half of maintenance: evidence → inspect match → maintenance
+    // proposal. Performs no persistence; callers own admission and the commit.
+    const proposeMaintenance = Effect.fn("Memory.proposeMaintenance")(function* (input: {
       model: Provider.Model
       config: MemorySchema.Config
       topics: MemorySchema.Topic[]
       messages: SessionV1.WithParts[]
-      projectID: Project.Info["id"]
     }) {
       const evidence = maintenanceEvidence(input.messages)
-      if (!evidence) return input.topics
+      if (!evidence) return
       const inspect = yield* match({
         model: input.model,
         config: input.config,
@@ -306,16 +320,28 @@ export const layer: Layer.Layer<
           }),
         }),
         schema: MemorySchema.MaintenanceResponse,
-        maxOutputTokens: 2_048,
+        maxOutputTokens: MAINTAIN_OUTPUT_TOKENS,
       })
       const decoded = Schema.decodeUnknownOption(MemorySchema.MaintenanceResponse)(output)
       if (Option.isNone(decoded))
         return yield* new ControllerError({ message: "MEMORY maintenance returned invalid output" })
+      return decoded.value.actions
+    })
+
+    const maintain = Effect.fn("Memory.maintain")(function* (input: {
+      model: Provider.Model
+      config: MemorySchema.Config
+      topics: MemorySchema.Topic[]
+      messages: SessionV1.WithParts[]
+      projectID: Project.Info["id"]
+    }) {
+      const actions = yield* proposeMaintenance(input)
+      if (!actions) return input.topics
       return yield* store
         .updateTopics(input.projectID, (topics) => ({
           applied: MemoryStore.applyActions({
             topics,
-            actions: decoded.value.actions,
+            actions,
             topicLimit: input.config.topic_limit,
           }),
           result: undefined,
@@ -343,6 +369,76 @@ export const layer: Layer.Layer<
       return renderSelection(selected, input.config)
     })
 
+    // Self-contained background maintenance for the checkpoint path: the
+    // matcher and maintenance model run OUTSIDE the fence/lock, and only the
+    // topic commit acquires them (applyUpdate), so a long reasoning call
+    // cannot wedge the lock, leak it on interruption, or block the caller.
+    const backgroundMaintain = Effect.fn("Memory.backgroundMaintain")(function* (input: {
+      model: Provider.Model
+      config: MemorySchema.Config
+      messages: SessionV1.WithParts[]
+      projectID: ProjectV2.ID
+    }) {
+      const topics = yield* store.readTopics(input.projectID)
+      const actions = yield* proposeMaintenance({
+        model: input.model,
+        config: input.config,
+        topics,
+        messages: input.messages,
+      })
+      if (!actions) return
+      yield* applyUpdate(input.projectID, (current) => ({
+        applied: MemoryStore.applyActions({
+          topics: current,
+          actions,
+          topicLimit: input.config.topic_limit,
+        }),
+        result: undefined,
+      }))
+    })
+
+    const releaseMaintenanceSlot = (projectID: ProjectV2.ID) =>
+      Ref.update(maintenanceInFlight, (set) => {
+        if (!set.has(projectID)) return set
+        const next = new Set(set)
+        next.delete(projectID)
+        return next
+      })
+
+    const kickMaintenance = Effect.fn("Memory.kickMaintenance")(function* (input: {
+      model: Provider.Model
+      config: MemorySchema.Config
+      messages: SessionV1.WithParts[]
+      projectID: ProjectV2.ID
+    }) {
+      const job = backgroundMaintain(input).pipe(
+        Effect.catchCause((cause) => Effect.logWarning("background MEMORY maintenance failed", { cause })),
+        Effect.ensuring(releaseMaintenanceSlot(input.projectID)),
+      )
+      // Reserve and fork atomically: an interruption between the two would
+      // leak the in-flight slot and silently skip every later maintenance for
+      // this process; a fork into a closing scope must hand the slot back.
+      yield* Effect.uninterruptible(
+        Effect.gen(function* () {
+          const reserved = yield* Ref.modify(maintenanceInFlight, (set) =>
+            set.has(input.projectID)
+              ? ([false, set] as const)
+              : ([true, new Set(set).add(input.projectID)] as const),
+          )
+          if (!reserved) return
+          yield* job.pipe(
+            Effect.forkIn(scope),
+            Effect.catchCause((cause) =>
+              Effect.gen(function* () {
+                yield* releaseMaintenanceSlot(input.projectID)
+                yield* Effect.logWarning("background MEMORY maintenance fork failed", { cause })
+              }),
+            ),
+          )
+        }),
+      )
+    })
+
     const prepareUnsafe = Effect.fn("Memory.prepareUnsafe")(function* (input: {
       sessionID: SessionID
       messages: SessionV1.WithParts[]
@@ -422,7 +518,6 @@ export const layer: Layer.Layer<
 
     const prepare: Interface["prepare"] = Effect.fn("Memory.prepare")((input) =>
       prepareUnsafe(input).pipe(
-        Effect.timeout(PREPARE_TIMEOUT),
         Effect.catchCause((cause) => Effect.logWarning("MEMORY prepare failed", { cause })),
       ),
     )
@@ -525,7 +620,6 @@ export const layer: Layer.Layer<
 
     const search: Interface["search"] = Effect.fn("Memory.search")((input) =>
       searchUnsafe(input).pipe(
-        Effect.timeout(PREPARE_TIMEOUT),
         Effect.catchCause((cause) =>
           Effect.gen(function* () {
             yield* Effect.logWarning("MEMORY search failed", { cause })
@@ -545,54 +639,40 @@ export const layer: Layer.Layer<
         return []
       }
       const user = latestRealUser(input.messages)
-      // Cross-process identity guard: a concurrent upgrade may retire this
-      // identity (row deleted, Home renamed away) while this write is in
-      // flight. Serialize on the identity lock and re-check liveness inside it;
-      // writing after retirement would re-create the retired Home and orphan
-      // the new content permanently (the identity cache already points at the
-      // successor, so no migration would ever run for this pair again).
       const live = yield* fence.withLiveIdentity(
         current.project.id,
-        Effect.gen(function* () {
-          return yield* lock.withProject(current.project.id)(
-            Effect.gen(function* () {
-              const topics = yield* store.readTopics(current.project.id)
-              const maintained = yield* maintain({
-                model: current.model,
-                config: current.loaded.config,
-                topics,
-                messages: input.messages,
-                projectID: current.project.id,
-              }).pipe(
-                Effect.catchCause((cause) =>
-                  Effect.gen(function* () {
-                    yield* Effect.logWarning("pre-compaction MEMORY maintenance failed", { cause })
-                    return topics
-                  }),
-                ),
-              )
-              const rendered = (yield* select({
-                model: current.model,
-                config: current.loaded.config,
-                topics: maintained,
-                text: user?.text ?? "",
-                projectID: current.project.id,
-              })).rendered
-              return rendered
-            }),
-          )
-        }),
+        lock.withProject(current.project.id)(
+          Effect.gen(function* () {
+            const topics = yield* store.readTopics(current.project.id)
+            return (yield* select({
+              model: current.model,
+              config: current.loaded.config,
+              topics,
+              text: user?.text ?? "",
+              projectID: current.project.id,
+            })).rendered
+          }),
+        ),
       )
       if (Option.isNone(live)) {
         yield* clearSession(input.sessionID)
         return []
       }
+      // Maintenance runs in the background AFTER the identity fence: compaction
+      // must not wait on a long reasoning call, a retired identity never burns
+      // model calls, and the injection above rendered the pre-maintenance
+      // topics. At most one job per project is in flight.
+      yield* kickMaintenance({
+        model: current.model,
+        config: current.loaded.config,
+        messages: input.messages,
+        projectID: current.project.id,
+      })
       return live.value
     })
 
     const checkpoint: Interface["checkpoint"] = Effect.fn("Memory.checkpoint")((input) =>
       checkpointUnsafe(input).pipe(
-        Effect.timeout(CHECKPOINT_TIMEOUT),
         Effect.catchCause((cause) =>
           Effect.gen(function* () {
             yield* Effect.logWarning("MEMORY checkpoint failed", { cause })
diff --git a/packages/opencode/src/memory/model.ts b/packages/opencode/src/memory/model.ts
index c328460387..88c7bcee5e 100644
--- a/packages/opencode/src/memory/model.ts
+++ b/packages/opencode/src/memory/model.ts
@@ -5,7 +5,13 @@ import { Context, Duration, Effect, Layer, Schema } from "effect"
 import { generateObject } from "ai"
 import { Provider } from "@/provider/provider"
 
-const DEFAULT_TIMEOUT = Duration.seconds(8)
+// Last-resort guard for a provider that never responds at all. This is not an
+// activity budget: maintenance reasoning runs are long, generation terminates
+// on its own via max_output_tokens, and transport timers own dead-connection
+// detection, so a call that keeps working finishes before this ever fires.
+const RESPONSE_TIMEOUT = Duration.minutes(5)
+
+const JSON_HINT = "Respond with a JSON object matching the provided schema."
 
 export interface Request {
   readonly model: Provider.Model
@@ -43,9 +49,9 @@ export function make(input: {
 }) {
   return Service.of({
     generate: Effect.fn("MemoryModel.generate")((request) =>
-      input.execute(request).pipe(
+      input.execute(requireJsonToken(request)).pipe(
         Effect.timeoutOrElse({
-          duration: input.timeout ?? DEFAULT_TIMEOUT,
+          duration: input.timeout ?? RESPONSE_TIMEOUT,
           orElse: () => Effect.fail(new TimeoutError()),
         }),
       ),
@@ -53,6 +59,13 @@ export function make(input: {
   })
 }
 
+// Providers serving response_format json_object reject prompts that do not
+// contain the literal word "json"; the maintenance prompts never mention it.
+function requireJsonToken(request: Request): Request {
+  if (/json/i.test(request.system) || /json/i.test(request.prompt)) return request
+  return { ...request, system: `${request.system}\n${JSON_HINT}` }
+}
+
 export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
diff --git a/packages/opencode/test/memory/memory.test.ts b/packages/opencode/test/memory/memory.test.ts
index 40c0f0105b..49d455f2c0 100644
--- a/packages/opencode/test/memory/memory.test.ts
+++ b/packages/opencode/test/memory/memory.test.ts
@@ -313,6 +313,7 @@ function recallFixture() {
     topics: MemorySchema.Topic[]
     failQueries: Set<string>
     maintenance: number
+    budgets: number[]
     config: MemorySchema.Config
     projectInitialized: number
     matcher?: (query: string) => Effect.Effect<unknown>
@@ -322,6 +323,7 @@ function recallFixture() {
     topics: [topic()],
     failQueries: new Set<string>(),
     maintenance: 0,
+    budgets: [],
     config,
     projectInitialized: 1,
   }
@@ -349,6 +351,7 @@ function recallFixture() {
         Layer.mock(MemoryModel.Service, {
           generate: (input) =>
             Effect.gen(function* () {
+              state.budgets.push(input.maxOutputTokens)
               if (input.system === MemoryPrompts.MATCH_SYSTEM) {
                 const request: unknown = JSON.parse(input.prompt)
                 const query =
@@ -403,6 +406,7 @@ function recallFixture() {
       state.topics = [topic()]
       state.failQueries.clear()
       state.maintenance = 0
+      state.budgets.length = 0
       state.config = config
       state.projectInitialized = 1
       state.matcher = undefined
@@ -1375,6 +1379,79 @@ describe("memory hidden model", () => {
       expect(interrupted).toBe(true)
     }),
   )
+
+  it.live("guarantees the json token reaches the model and leaves json-aware prompts untouched", () =>
+    Effect.gen(function* () {
+      const seen: MemoryModel.Request[] = []
+      const service = MemoryModel.make({
+        execute: (request) =>
+          Effect.sync(() => {
+            seen.push(request)
+            return {}
+          }),
+      })
+
+      yield* service.generate({
+        model: ProviderTest.model(),
+        system: "Select relevant topics.",
+        prompt: "plain evidence without the token",
+        schema: MemorySchema.MatchResponse,
+        maxOutputTokens: 32,
+      })
+      yield* service.generate({
+        model: ProviderTest.model(),
+        system: "Propose updates as a JSON object.",
+        prompt: "evidence",
+        schema: MemorySchema.MaintenanceResponse,
+        maxOutputTokens: 32,
+      })
+
+      expect(seen[0].system).toContain("JSON")
+      expect(seen[1].system).toBe("Propose updates as a JSON object.")
+    }),
+  )
+})
+
+describe("memory maintenance budgets", () => {
+  const recall = recallFixture()
+
+  recall.it.instance(
+    "sizes match and maintenance output for reasoning-heavy models during checkpoint",
+    () =>
+      Effect.gen(function* () {
+        recall.reset()
+        const memory = yield* Memory.Service
+        const sessionID = SessionID.make("ses_memory_budget")
+        const userID = MessageID.ascending()
+        const messages: SessionV1.WithParts[] = [
+          user(userID, sessionID, "确认一条长期偏好：回复保持简洁"),
+          {
+            info: assistant(userID, sessionID, ProviderV2.ID.make("test"), ModelV2.ID.make("test-model"), "end_turn"),
+            parts: [],
+          },
+        ]
+
+        yield* memory.checkpoint({ sessionID, messages })
+        // Checkpoint select runs synchronously; the kicked maintenance job
+        // completes in the background (its inspect-match + maintain calls).
+        yield* pollWithTimeout(
+          Effect.sync(() => (recall.state.budgets.length === 3 ? (true as const) : undefined)),
+          "background maintenance never completed",
+        )
+        expect([...recall.state.budgets].sort((a, b) => a - b)).toEqual([2_048, 2_048, 16_384])
+        // A second checkpoint must run a second maintenance: the in-flight
+        // slot is released when the first job finishes, never wedged.
+        yield* pollWithTimeout(
+          Effect.gen(function* () {
+            if (recall.state.budgets.filter((budget) => budget === 16_384).length >= 2) return true as const
+            yield* memory.checkpoint({ sessionID, messages })
+            return undefined
+          }),
+          "second background maintenance never ran — in-flight slot wedged",
+        )
+      }),
+    { git: true },
+  )
 })
 
 describe("memory bootstrap", () => {

From 8e2329d4f0cc20db97e95d871cf40c26dd3fd8c5 Mon Sep 17 00:00:00 2001
From: Lex <sunsan05@Gmail.com>
Date: Tue, 18 Aug 2026 06:36:05 +0800
Subject: [PATCH 6/7] fix(memory): replace wall-clock timeout with per-chunk
 SSE liveness

Retire the whole-call wall clock in favor of activity-based liveness on the memory model seam: switch generateObject -> streamObject and re-arm an idle watchdog on every arriving part (reasoning deltas included), so an actively streaming call is never killed however long the reasoning runs. Add a connect timeout (no first part) and an idle timeout (gap between parts), both pure non-response detectors; generation still terminates via max_output_tokens / a natural stop. Addresses the streaming-liveness half of #324; the periodic prepare lock hardening remains tracked there.
---
 packages/opencode/src/memory/model.ts        | 149 +++++++++++++++----
 packages/opencode/test/memory/memory.test.ts |  91 +++++++++++
 2 files changed, 213 insertions(+), 27 deletions(-)

diff --git a/packages/opencode/src/memory/model.ts b/packages/opencode/src/memory/model.ts
index 88c7bcee5e..f8d7c22855 100644
--- a/packages/opencode/src/memory/model.ts
+++ b/packages/opencode/src/memory/model.ts
@@ -2,14 +2,16 @@ export * as MemoryModel from "./model"
 
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { Context, Duration, Effect, Layer, Schema } from "effect"
-import { generateObject } from "ai"
+import { streamObject } from "ai"
 import { Provider } from "@/provider/provider"
 
-// Last-resort guard for a provider that never responds at all. This is not an
-// activity budget: maintenance reasoning runs are long, generation terminates
-// on its own via max_output_tokens, and transport timers own dead-connection
-// detection, so a call that keeps working finishes before this ever fires.
-const RESPONSE_TIMEOUT = Duration.minutes(5)
+// Liveness is judged per-chunk, never by a whole-call wall clock: a stream
+// that keeps delivering parts is alive, however long the reasoning runs.
+// CONNECT_TIMEOUT bounds the wait for the FIRST part; IDLE_TIMEOUT bounds the
+// silence BETWEEN parts and is re-armed by every arriving part. Generation
+// still terminates on its own via max_output_tokens / a natural stop.
+const CONNECT_TIMEOUT = Duration.seconds(60)
+const IDLE_TIMEOUT = Duration.seconds(60)
 
 const JSON_HINT = "Respond with a JSON object matching the provided schema."
 
@@ -48,14 +50,19 @@ export function make(input: {
   readonly timeout?: Duration.Input
 }) {
   return Service.of({
-    generate: Effect.fn("MemoryModel.generate")((request) =>
-      input.execute(requireJsonToken(request)).pipe(
+    generate: Effect.fn("MemoryModel.generate")((request) => {
+      const effect = input.execute(requireJsonToken(request))
+      // An injected timeout (tests) stays a hard deadline; production relies on
+      // the per-chunk liveness below, so an actively streaming call is never
+      // killed by a wall clock.
+      if (input.timeout === undefined) return effect
+      return effect.pipe(
         Effect.timeoutOrElse({
-          duration: input.timeout ?? RESPONSE_TIMEOUT,
+          duration: input.timeout,
           orElse: () => Effect.fail(new TimeoutError()),
         }),
-      ),
-    ),
+      )
+    }),
   })
 }
 
@@ -66,6 +73,101 @@ function requireJsonToken(request: Request): Request {
   return { ...request, system: `${request.system}\n${JSON_HINT}` }
 }
 
+// Signals that the stream went silent past the liveness window.
+export class Stalled extends Error {}
+
+// Drains `parts`, re-arming the idle watchdog on EVERY part (so a live stream
+// that keeps delivering — reasoning deltas included — never trips the timer).
+// Arms `connectTimeout` until the first part and `idleTimeout` between parts;
+// a silent window invokes `onStall` (abort the request) and fails with
+// `Stalled`, while an `errorOf` hit fails with that part's error.
+export const drainWithLiveness = <T>(input: {
+  parts: AsyncIterable<T>
+  connectTimeout: Duration.Duration
+  idleTimeout: Duration.Duration
+  onStall: () => void
+  errorOf: (part: T) => unknown | undefined
+}) =>
+  new Promise<void>((resolve, reject) => {
+    let timer: ReturnType<typeof setTimeout> | undefined
+    let settled = false
+    const finish = (action: () => void) => {
+      if (settled) return
+      settled = true
+      if (timer) clearTimeout(timer)
+      action()
+    }
+    const arm = (duration: Duration.Duration) => {
+      if (timer) clearTimeout(timer)
+      timer = setTimeout(
+        () =>
+          finish(() => {
+            input.onStall()
+            reject(new Stalled())
+          }),
+        Duration.toMillis(duration),
+      )
+    }
+    arm(input.connectTimeout)
+    void (async () => {
+      try {
+        for await (const part of input.parts) {
+          arm(input.idleTimeout)
+          const error = input.errorOf(part)
+          if (error !== undefined) throw error
+        }
+        finish(resolve)
+      } catch (cause) {
+        finish(() => reject(cause))
+      }
+    })()
+  })
+
+const streamGenerate = (input: {
+  language: Parameters<typeof streamObject>[0]["model"]
+  system: string
+  prompt: string
+  schema: Schema.Decoder<unknown>
+  temperature?: number
+  maxOutputTokens: number
+  connectTimeout: Duration.Duration
+  idleTimeout: Duration.Duration
+}) =>
+  Effect.tryPromise({
+    try: (signal) =>
+      (async () => {
+        const controller = new AbortController()
+        const forwardAbort = () => controller.abort()
+        signal.addEventListener("abort", forwardAbort)
+        try {
+          const result = streamObject({
+            model: input.language,
+            system: input.system,
+            prompt: input.prompt,
+            schema: Object.assign(
+              Schema.toStandardSchemaV1(input.schema),
+              Schema.toStandardJSONSchemaV1(input.schema),
+            ),
+            temperature: input.temperature,
+            maxOutputTokens: input.maxOutputTokens,
+            abortSignal: controller.signal,
+            onError: () => {},
+          })
+          await drainWithLiveness({
+            parts: result.fullStream,
+            connectTimeout: input.connectTimeout,
+            idleTimeout: input.idleTimeout,
+            onStall: () => controller.abort(),
+            errorOf: (part) => (part.type === "error" ? part.error : undefined),
+          })
+          return await result.object
+        } finally {
+          signal.removeEventListener("abort", forwardAbort)
+        }
+      })(),
+    catch: (cause) => (cause instanceof Stalled ? new TimeoutError() : new GenerateError({ cause })),
+  })
+
 export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
@@ -73,22 +175,15 @@ export const layer = Layer.effect(
     return make({
       execute: Effect.fnUntraced(function* (input) {
         const language = yield* provider.getLanguage(input.model)
-        const schema = Object.assign(
-          Schema.toStandardSchemaV1(input.schema),
-          Schema.toStandardJSONSchemaV1(input.schema),
-        )
-        return yield* Effect.tryPromise({
-          try: (signal) =>
-            generateObject({
-              model: language,
-              system: input.system,
-              prompt: input.prompt,
-              schema,
-              temperature: input.model.capabilities.temperature ? 0 : undefined,
-              maxOutputTokens: input.maxOutputTokens,
-              abortSignal: signal,
-            }).then((result) => result.object),
-          catch: (cause) => new GenerateError({ cause }),
+        return yield* streamGenerate({
+          language,
+          system: input.system,
+          prompt: input.prompt,
+          schema: input.schema,
+          temperature: input.model.capabilities.temperature ? 0 : undefined,
+          maxOutputTokens: input.maxOutputTokens,
+          connectTimeout: CONNECT_TIMEOUT,
+          idleTimeout: IDLE_TIMEOUT,
         })
       }),
     })
diff --git a/packages/opencode/test/memory/memory.test.ts b/packages/opencode/test/memory/memory.test.ts
index 49d455f2c0..2f0a9e4b3c 100644
--- a/packages/opencode/test/memory/memory.test.ts
+++ b/packages/opencode/test/memory/memory.test.ts
@@ -1410,6 +1410,97 @@ describe("memory hidden model", () => {
       expect(seen[1].system).toBe("Propose updates as a JSON object.")
     }),
   )
+
+  it.live("keeps an actively streaming call alive regardless of total duration", () =>
+    Effect.gen(function* () {
+      // Thirty parts arriving every 8ms: ~240ms total, far past the 40ms
+      // idle window — every arrival re-arms the watchdog, so the call lives.
+      async function* streaming() {
+        for (let index = 0; index < 30; index++) {
+          await new Promise((resolve) => setTimeout(resolve, 8))
+          yield { type: index % 2 === 0 ? "reasoning" : "text-delta" }
+        }
+      }
+      yield* Effect.promise(() =>
+        MemoryModel.drainWithLiveness({
+          parts: streaming(),
+          connectTimeout: Duration.millis(40),
+          idleTimeout: Duration.millis(40),
+          onStall: () => {},
+          errorOf: () => undefined,
+        }),
+      )
+    }),
+  )
+
+  it.live("stalls a silent stream after the idle window and invokes the abort hook", () =>
+    Effect.gen(function* () {
+      let aborts = 0
+      async function* onePartThenSilence() {
+        yield { type: "reasoning" }
+        await new Promise(() => {})
+      }
+      const error = yield* Effect.tryPromise({
+        try: () =>
+          MemoryModel.drainWithLiveness({
+            parts: onePartThenSilence(),
+            connectTimeout: Duration.millis(250),
+            idleTimeout: Duration.millis(40),
+            onStall: () => {
+              aborts++
+            },
+            errorOf: () => undefined,
+          }),
+        catch: (cause) => cause,
+      }).pipe(Effect.flip)
+      expect(error instanceof MemoryModel.Stalled).toBe(true)
+      expect(aborts).toBe(1)
+    }),
+  )
+
+  it.live("fails on a dead connection that never delivers a first part", () =>
+    Effect.gen(function* () {
+      async function* nothing() {
+        await new Promise(() => {})
+      }
+      const started = Date.now()
+      const error = yield* Effect.tryPromise({
+        try: () =>
+          MemoryModel.drainWithLiveness({
+            parts: nothing(),
+            connectTimeout: Duration.millis(40),
+            idleTimeout: Duration.millis(250),
+            onStall: () => {},
+            errorOf: () => undefined,
+          }),
+        catch: (cause) => cause,
+      }).pipe(Effect.flip)
+      expect(error instanceof MemoryModel.Stalled).toBe(true)
+      expect(Date.now() - started).toBeLessThan(200)
+    }),
+  )
+
+  it.live("propagates a stream error part without treating it as a stall", () =>
+    Effect.gen(function* () {
+      const boom = new Error("provider stream error")
+      async function* erroring() {
+        yield { type: "text-delta" }
+        yield { type: "error", error: boom }
+      }
+      const error = yield* Effect.tryPromise({
+        try: () =>
+          MemoryModel.drainWithLiveness({
+            parts: erroring(),
+            connectTimeout: Duration.millis(250),
+            idleTimeout: Duration.millis(250),
+            onStall: () => {},
+            errorOf: (part: { type: string; error?: unknown }) => (part.type === "error" ? part.error : undefined),
+          }),
+        catch: (cause) => cause,
+      }).pipe(Effect.flip)
+      expect(error).toBe(boom)
+    }),
+  )
 })
 
 describe("memory maintenance budgets", () => {

From 342555393a4bcd663bc4e9fa3164763cbd050a5d Mon Sep 17 00:00:00 2001
From: Lex <sunsan05@Gmail.com>
Date: Tue, 18 Aug 2026 06:57:23 +0800
Subject: [PATCH 7/7] fix(dag): harden the replan verdict gate against string
 outputs and transient pause failure

Audit follow-up to #327/#322. (1) Parse a string-typed checkpoint output as JSON before matching the replan verdict, so a report_to_parent gate without output_schema (or a string-typed child reply, reachable via replan fragments which skip the authoring check) cannot bypass the pause gate and reproduce the #322 spin. (2) Retry the gate pause once before falling back to the durable status, so a transient pause failure (e.g. the workflow lock held by a concurrent long replan) never silently strands the workflow with no spawn round and no parent wake.
---
 packages/opencode/src/dag/runtime/loop.ts     | 38 +++++++++++++------
 .../opencode/test/dag/dag-loop-guards.test.ts | 36 ++++++++++++++++++
 2 files changed, 62 insertions(+), 12 deletions(-)

diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts
index c8c6cb60bd..26d1a7ae8b 100644
--- a/packages/opencode/src/dag/runtime/loop.ts
+++ b/packages/opencode/src/dag/runtime/loop.ts
@@ -41,6 +41,7 @@ import { reconcileWorkflow, makeSessionStatusChecker } from "./recovery"
 // 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") })
+const parseJsonOption = Schema.decodeUnknownOption(Schema.UnknownFromJsonString)
 
 export interface Interface {
   readonly init: () => Effect.Effect<void>
@@ -658,9 +659,17 @@ const serviceLayer = Layer.effect(
                     if (confirmed && entry.runtime.isActive(nodeID)) {
                       settle(entry, nodeID)
                       const nodeConfig = entry.config?.nodes.find((n) => n.id === nodeID)
+                      // A checkpoint output can arrive as a raw string (no
+                      // output_schema, or a string-typed child reply); parse it
+                      // before matching the verdict so a string-typed
+                      // {"verdict":"replan"} cannot bypass the gate (the spin
+                      // behind issue #322).
+                      const gateOutput = typeof node?.output === "string"
+                        ? Option.getOrUndefined(parseJsonOption(node.output))
+                        : node?.output
                       const gateReplan = def === DagEvent.NodeCompleted
                         && nodeConfig?.report_to_parent === true
-                        && Option.isSome(Schema.decodeUnknownOption(GateReplanVerdict)(node?.output))
+                        && Option.isSome(Schema.decodeUnknownOption(GateReplanVerdict)(gateOutput))
                       if (gateReplan) {
                         // Verdict gate (issue #322): a reporting checkpoint that
                         // submits verdict "replan" vetoes the direction. Pause
@@ -669,17 +678,22 @@ const serviceLayer = Layer.effect(
                         // 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"
-                            }),
-                          ),
-                        )
+                        const paused = yield* Effect.gen(function* () {
+                          // Pause can fail transiently (e.g. the workflow lock is
+                          // held by a concurrent long replan); retry once before
+                          // falling back to the durable status, so the workflow
+                          // is never silently stranded.
+                          const attemptPause = dag.pause(dagID).pipe(
+                            Effect.map(() => true),
+                            Effect.catch(() => Effect.succeed(false)),
+                          )
+                          if (yield* attemptPause) return true
+                          if (yield* attemptPause) return true
+                          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 })
                       }
diff --git a/packages/opencode/test/dag/dag-loop-guards.test.ts b/packages/opencode/test/dag/dag-loop-guards.test.ts
index 43d1e35cf4..fce8c31d39 100644
--- a/packages/opencode/test/dag/dag-loop-guards.test.ts
+++ b/packages/opencode/test/dag/dag-loop-guards.test.ts
@@ -449,4 +449,40 @@ describe("DagLoop replan verdict gate (issue #322)", () => {
       ),
     )
   })
+
+  it("pauses on a string-typed replan verdict (no output_schema bypass)", 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 string verdict",
+            config: {
+              name: "gate-string-verdict",
+              nodes: [
+                // report_to_parent without output_schema: the child's final
+                // text lands as a raw string output.
+                node({ id: "gate", name: "gate", required: true, report_to_parent: true }),
+                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")
+          // String-typed verdict (audit SOFT-2): must still trip the gate,
+          // not slip past the Object-only decode.
+          yield* dag.nodeCompleted(dagID, "gate", JSON.stringify({ verdict: "replan", findings: "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 string-typed replan verdict",
+          )
+          expect((yield* store.getNode(dagID, "downstream"))?.status).toBe("pending")
+        }),
+      ),
+    )
+  })
 })