fix(dag): low-severity batch from the 2026-08-19 deep-dive audit

.specgit.yaml

@@ -1,8 +1,8 @@
 version: 1
-delivery: issue370
+delivery: issue349
 context:
   kind: branch
-  branch: feat/370-issue370
+  branch: feat/349-issue349
 issues:
-  - 370
-pr: 371
+  - 349
+pr: 372

AGENTS.md

@@ -278,3 +278,13 @@ verified on its own evidence, split it before binding.
 - `--json` is the only parse surface: stdout is exactly one JSON
   document; never scrape human-readable output.
 <!-- specgit:block:end -->
+
+## Tool-call discipline (hard rules)
+
+- Never fan out duplicate or near-duplicate queries. One question, one
+  tool call; if the answer is already in context, make zero calls.
+- Parallel tool batches must contain distinct, independently justified
+  calls. Before sending a batch, verify no two calls answer the same
+  question. A repeated identical call is a bug regardless of intent.
+- Long CI waits use `sleep N && <single check>`, never repeated watches
+  of the same resource. One watch command, one result.

packages/opencode/src/dag/blocks.ts

@@ -263,10 +263,28 @@ function compileBlock(
         required: true,
         reportToParent: false,
         inputMapping: Object.fromEntries(
-          aggregation.writerIDs.flatMap((writerID: string) => [
-            [`${writerID.replace(/-/g, "_")}_changed_files`, `${writerID}.output.changed_files`],
-            [`${writerID.replace(/-/g, "_")}_summary`, `${writerID}.output.summary`],
-          ]),
+          (() => {
+            // #349/BLK-02: writer ids may mix hyphens and underscores
+            // ("foo-bar" vs "foo_bar") whose -→_ normalization collides on
+            // the same mapping key — Object.fromEntries would silently drop
+            // one writer's evidence (and its files escape the aggregator's
+            // overlap detection). Reject the shape at compile time.
+            const seen = new Map<string, string>()
+            for (const writerID of aggregation.writerIDs) {
+              const key = writerID.replace(/-/g, "_")
+              const prior = seen.get(key)
+              if (prior !== undefined) {
+                throw new Error(
+                  `Parallel implementation writers "${prior}" and "${writerID}" normalize to the same input-mapping key "${key}" — their aggregator evidence keys would collide. Rename one of the writers so the ids differ beyond hyphens vs underscores`,
+                )
+              }
+              seen.set(key, writerID)
+            }
+            return aggregation.writerIDs.flatMap((writerID: string) => [
+              [`${writerID.replace(/-/g, "_")}_changed_files`, `${writerID}.output.changed_files`],
+              [`${writerID.replace(/-/g, "_")}_summary`, `${writerID}.output.summary`],
+            ])
+          })(),
         ),
         outputSchema: IMPLEMENTATION_SCHEMA,
       }),
@@ -276,6 +294,17 @@ function compileBlock(
 
   const verifyAggregatorIDs = verifyAggregators.get(block.id)
   const verifyAggregator = verifyAggregatorIDs && verifyAggregatorIDs.length > 0 ? verifyAggregatorIDs[0] : undefined
+  // #349/BLK-3: one verify node serving two parallel-writer review routes
+  // would be rewired onto two aggregators, but the verify contract binds ONE
+  // implementation reference and ONE fingerprint — mapping only the first
+  // (the old silent behavior) lets the second route's write-set escape the
+  // review binding. Reject the shape instead: fan the routes together
+  // first, exactly like multi-review-gate dependencies.
+  if (verifyAggregatorIDs && verifyAggregatorIDs.length > 1) {
+    throw new Error(
+      `Verify block "${block.id}" serves multiple parallel-writer review routes (${verifyAggregatorIDs.join(", ")}) — the verification contract binds a single implementation fingerprint. Fan the routes into one review block first, or give each route its own verify block`,
+    )
+  }
   // A synthesize that follows a review is the route's final gate: it must map
   // the review output so unresolvedReviewOutcomes/finalReviewGates recognize
   // an ACCEPTed review as resolved (issue #304) — the same binding contract

packages/opencode/src/dag/runtime/capture.ts

@@ -87,6 +87,15 @@ export function validateAgainstSchema(value: unknown, schema: Record<string, unk
     if (typeof maxItems === "number" && value.length > maxItems)
       return { ok: false, error: `expected maxItems ${maxItems}, got ${value.length}` }
     if (schema["uniqueItems"] === true) {
+      // #349/CAP-02: the pairwise deepEqual scan is O(n²); model outputs
+      // with more items than this are pathological — fail loudly instead of
+      // burning the validation path.
+      if (value.length > UNIQUE_ITEMS_MAX) {
+        return {
+          ok: false,
+          error: `uniqueItems validation is capped at ${UNIQUE_ITEMS_MAX} items, got ${value.length}`,
+        }
+      }
       const duplicate = value.findIndex((item, index) => value.slice(0, index).some((prev) => deepEqual(prev, item)))
       if (duplicate !== -1)
         return { ok: false, error: `expected uniqueItems, found duplicate at index ${duplicate}` }
@@ -245,9 +254,15 @@ function describeType(value: unknown): string {
 
 // Schema patterns come from workflow config; a malformed regex must not crash
 // validation, it just fails the constraint.
+// #349/CAP-02: patterns may also be PATHOLOGICAL (the draft action lets a
+// model author them) — cap the tested span so catastrophic backtracking
+// against an unbounded model output cannot hang submit_result validation.
+const REGEX_TEST_MAX_CHARS = 100_000
+// #349/CAP-02: bound for the O(n²) uniqueItems pairwise scan.
+const UNIQUE_ITEMS_MAX = 1_000
 function safeRegexTest(pattern: string, value: string): boolean {
   try {
-    return new RegExp(pattern).test(value)
+    return new RegExp(pattern).test(value.slice(0, REGEX_TEST_MAX_CHARS))
   } catch {
     return false
   }

packages/opencode/src/dag/runtime/loop.ts

@@ -321,7 +321,9 @@ const serviceLayer = Layer.effect(
               Effect.provideService(Session.Service, sessionSvc),
               Effect.provideService(SessionPrompt.Service, promptSvc),
               Effect.catchCause((cause) =>
-                dag.nodeFailed(dagID, nodeID, Cause.pretty(cause), "exec_failed"),
+                Cause.hasInterrupts(cause)
+                  ? Effect.failCause(cause)
+                  : dag.nodeFailed(dagID, nodeID, Cause.pretty(cause), "exec_failed"),
               ),
               Effect.ignore,
             )
@@ -436,18 +438,33 @@ const serviceLayer = Layer.effect(
             // explicit workflow control.
             const pausedForRecovery = recovery.ownershipLost > 0 && wf.status === "running"
             if (pausedForRecovery) {
-              // A concurrent control op (cancel/fail) can terminalize the
-              // workflow while reconciliation runs — the pause guard then
-              // rejects. Abandon adoption instead of tracking a workflow this
-              // instance no longer controls.
-              const pauseAccepted = yield* dag.pause(dagID).pipe(
-                Effect.as(true),
-                Effect.catchCause((cause) =>
-                  Effect.logWarning("DagLoop recovery pause rejected — abandoning adoption", { dagID, cause }).pipe(
-                    Effect.as(false),
+              // #349/NEW-2: a pause rejected by a CONCURRENT TERMINAL control
+              // op means this instance no longer controls the workflow —
+              // abandoning adoption is correct. But a lock-timeout or store
+              // defect used to take the same silent path: the invented
+              // NodeFailed rows were persisted with no runtime entry, events
+              // filtered by runtimes.has, wake boundaries requiring an entry —
+              // the workflow stalled until a process restart. Mirror the
+              // replan-verdict gate: retry twice, fold defects in, and only
+              // abandon when the durable row is genuinely terminal.
+              const pauseAccepted = yield* Effect.gen(function* () {
+                const attemptPause = dag.pause(dagID).pipe(
+                  Effect.map(() => true),
+                  Effect.catchCause((cause) =>
+                    Cause.hasInterrupts(cause) ? Effect.failCause(cause) : Effect.succeed(false),
                   ),
-                ),
-              )
+                )
+                if (yield* attemptPause) return true
+                if (yield* attemptPause) return true
+                const row = yield* store.getWorkflow(dagID).pipe(Effect.orDie)
+                if (row && row.status !== "paused") {
+                  yield* Effect.logError(
+                    "DagLoop recovery pause failed after retries — workflow stays unadopted; it will be re-adopted on the next instance load",
+                    { dagID, status: row.status },
+                  )
+                }
+                return row?.status === "paused"
+              })
               if (!pauseAccepted) return
               yield* Effect.logWarning("DagLoop paused workflow after recovery invented node failures", {
                 dagID,

packages/opencode/src/dag/runtime/output-ref.ts

@@ -45,6 +45,10 @@ const SUMMARY_CHARS = 200
 // The summary only needs the leading chars; decoding a bounded prefix keeps a
 // giant report from being copied twice (once for the digest, once for text).
 const SUMMARY_DECODE_BYTES = 4096
+// #349/CAP-02: whole-file capture bound — a giant or sparse referenced file
+// must not spike memory; larger files fall back to the inline path
+// (returning undefined here is the designed degradation).
+const FILE_REF_MAX_BYTES = 64 * 1024 * 1024
 const MAX_PATH_CHARS = 4096
 
 export const REPORT_AREA = path.join(".opencode", "workflow-reports")
@@ -88,6 +92,9 @@ export function captureOutputFileRef(rawText: string): Effect.Effect<OutputFileR
   return Effect.gen(function* () {
     const info = yield* Effect.promise(() => stat(candidate).catch(() => undefined))
     if (!info || !info.isFile() || info.size === 0) return undefined
+    // #349/CAP-02: refuse oversized refs — stat already told us the size, so
+    // the read never happens for a pathological file.
+    if (info.size > FILE_REF_MAX_BYTES) return undefined
     const bytes = yield* Effect.promise(() =>
       Bun.file(candidate)
         .arrayBuffer()

packages/opencode/src/dag/runtime/recovery.ts

@@ -69,7 +69,19 @@ export function reconcileWorkflow(
       // never revisit it if the workflow is about to become terminal.
       if (node.status === "pending" || node.status === "queued") {
         if (node.childSessionId && cancelSession) {
-          yield* cancelSession(node.childSessionId)
+          // #349/REC-1: same hardening as the running-node branch below — a
+          // persistent cancel failure must not abort the whole reconcile
+          // (this workflow would then never be adopted by this process).
+          yield* cancelSession(node.childSessionId).pipe(
+            Effect.catchCause((cause) =>
+              Effect.logWarning("DAG recovery failed to cancel stale child session", {
+                dagID,
+                nodeID: node.id,
+                childSessionID: node.childSessionId,
+                cause,
+              }),
+            ),
+          )
         }
         continue
       }

packages/opencode/src/dag/runtime/supervision-sweep.ts

@@ -8,11 +8,13 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { Database } from "@opencode-ai/core/database/database"
 import { DagStore } from "@opencode-ai/core/dag/store"
 import { WorkflowNodeTable } from "@opencode-ai/core/dag/sql"
+import { ProjectTable } from "@opencode-ai/core/project/sql"
 import { isNodeTerminalStatus, isTransitionRejection, isWorkflowTerminalStatus } from "@opencode-ai/core/dag/core/types"
 import { and, eq, sql } from "drizzle-orm"
 import { Dag, parseWorkflowConfig } from "@/dag/dag"
 import { SessionPrompt } from "@/session/prompt"
 import { SessionAutomationLease } from "@/session/automation-lease"
+import { InstanceRef } from "@/effect/instance-ref"
 
 /**
  * Host-level deadline-supervision sweep — the fallback retry for the
@@ -150,6 +152,37 @@ const serviceLayer = Layer.effect(
       return escalateIntervalFromConfig(wf?.config, nodeId)
     })
 
+    // #349/SW-L1: publish-side location stamping. The sweep's layer context
+    // has no ambient InstanceRef, so its durable events used to carry an
+    // empty location — live instances' summary publishers filter by
+    // directory, so the TUI never got a summary push for a swept settle
+    // (bootstrap refetch only). Providing a reference derived from the
+    // workflow's own durable row (directory + project) stamps the events so
+    // the owning directory's consumers see them. Falls back to unstamped
+    // when the row cannot be resolved — same visibility as before, never
+    // worse.
+    const withWorkflowLocation = Effect.fnUntraced(function* (workflowId: string, body: Effect.Effect<void, Error>) {
+      const wf = yield* store.getWorkflow(workflowId).pipe(
+        Effect.catchCause((cause) => (Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.succeed(undefined))),
+      )
+      if (!wf?.directory) return yield* body
+      const project = yield* db
+        .select()
+        .from(ProjectTable)
+        // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- the durable column is typed string; ProjectTable.id is branded.
+        .where(eq(ProjectTable.id, wf.projectId as never))
+        .get()
+        .pipe(Effect.catchCause((cause) => (Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.succeed(undefined))))
+      yield* body.pipe(
+        // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion -- partial InstanceContext: only directory/worktree/project.id are read by the publish-side location stamp.
+        Effect.provideService(InstanceRef, {
+          directory: wf.directory,
+          worktree: project?.worktree ?? wf.directory,
+          project: { id: wf.projectId },
+        } as never),
+      )
+    })
+
     const sweepOnce = Effect.fn("DagSupervisionSweep.sweepOnce")(function* () {
       const rows = yield* db
         .select({
@@ -219,28 +252,31 @@ const serviceLayer = Layer.effect(
             Effect.catchCause((cause) => (Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.void)),
           )
         }
-        const settled = yield* dag
-          .nodeFailed(
-            row.workflowId,
-            row.nodeId,
-            `deadline supervision lost (no escalation progress across ${flatTicks} sweep ticks, escalate cadence ${escalateIntervalMs}ms) — swept, extensions ${row.extensions}`,
-            "timeout",
-          )
-          .pipe(
-            Effect.as(true),
-            Effect.catchCause((cause) =>
-              Cause.hasInterrupts(cause)
-                ? Effect.interrupt
-                : Effect.gen(function* () {
-                    yield* Effect.logWarning("DagSupervisionSweep nodeFailed failed — retrying next tick", {
-                      dagID: row.workflowId,
-                      nodeID: row.nodeId,
-                      cause,
-                    })
-                    return false
-                  }),
-            ),
-          )
+        const settled = yield* withWorkflowLocation(
+          row.workflowId,
+          dag
+            .nodeFailed(
+              row.workflowId,
+              row.nodeId,
+              `deadline supervision lost (no escalation progress across ${flatTicks} sweep ticks, escalate cadence ${escalateIntervalMs}ms) — swept, extensions ${row.extensions}`,
+              "timeout",
+            )
+            .pipe(Effect.asVoid),
+        ).pipe(
+          Effect.as(true),
+          Effect.catchCause((cause) =>
+            Cause.hasInterrupts(cause)
+              ? Effect.interrupt
+              : Effect.gen(function* () {
+                  yield* Effect.logWarning("DagSupervisionSweep nodeFailed failed — retrying next tick", {
+                    dagID: row.workflowId,
+                    nodeID: row.nodeId,
+                    cause,
+                  })
+                  return false
+                }),
+          ),
+        )
         // On a failed settle keep the streak so the next tick retries
         // immediately instead of deferring by a full freeze window.
         if (!settled) continue
@@ -260,7 +296,7 @@ const serviceLayer = Layer.effect(
         // load. A live DagLoop racing this is serialized by the same
         // workflow lock and its terminal-status guards — double settles
         // collapse to one.
-        yield* settleWorkflowIfComplete(row.workflowId).pipe(
+        yield* withWorkflowLocation(row.workflowId, settleWorkflowIfComplete(row.workflowId)).pipe(
           Effect.catchCause((cause) => (Cause.hasInterrupts(cause) ? Effect.interrupt : Effect.void)),
         )
       }

packages/opencode/test/dag/dag-recovery.test.ts

@@ -171,22 +171,24 @@ describe("reconcileWorkflow", () => {
     expect(result).toEqual({ reconciled: 0, ownershipLost: 0 })
   })
 
-  it("aborts recovery when a stale restart-orphan session cannot be cancelled", async () => {
+  // #349/REC-1: a persistent stale-child cancel failure no longer aborts
+  // the whole reconcile — that made the workflow unadoptable in this process
+  // (its running nodes would never be scheduled until a restart). The
+  // failure is logged and the reconcile continues; this test pinned the old
+  // abort behavior.
+  it("survives a stale restart-orphan cancel failure and continues the reconcile", async () => {
     const events: TrackedEvent[] = []
     const nodes = [makeNodeRow({ id: "n1", status: "queued", childSessionId: "ses_stale" })]
     const dagLayer = makeDagLayer(nodes, events)
     const checkStatus = () => Effect.succeed("active" as const)
     const cancelSession = () => Effect.fail(new Error("cancel unavailable"))
 
-    const exit = await Effect.runPromise(
-      reconcileWorkflow("wf-1", checkStatus, cancelSession).pipe(
-        Effect.provide(dagLayer),
-        Effect.exit,
-      ),
+    const result = await Effect.runPromise(
+      reconcileWorkflow("wf-1", checkStatus, cancelSession).pipe(Effect.provide(dagLayer)),
     )
 
-    expect(Exit.isFailure(exit)).toBe(true)
     expect(events).toEqual([])
+    expect(result).toEqual({ reconciled: 0, ownershipLost: 0 })
   })
 
   it("cancels and fails a zero-message child classified as unknown exactly once", async () => {

packages/tui/src/config/keybind.ts

@@ -87,6 +87,10 @@ export const Definitions = {
   dag_resume: keybind("none", "Resume selected DAG workflow"),
   dag_step: keybind("none", "Step selected DAG workflow (run one node)"),
   dag_cancel: keybind("none", "Cancel selected DAG workflow"),
+  // #349/F6: plugin-level palette command — without a Definitions/CommandMap
+  // entry it is not rebindable and never appears in the keybind config
+  // schema.
+  dag_cancel_active: keybind("none", "Cancel the session's active DAG workflow"),
 
   editor_open: keybind("<leader>e", "Open external editor"),
   theme_list: keybind("<leader>t", "List available themes"),
@@ -306,6 +310,7 @@ export const CommandMap = {
   dag_resume: "dag.resume",
   dag_step: "dag.step",
   dag_cancel: "dag.cancel",
+  dag_cancel_active: "dag.cancel.active",
   editor_open: "prompt.editor",
   theme_list: "theme.switch",
   theme_switch_mode: "theme.switch_mode",