fix(dag): resolve DAG prompt and output duplication

.specgit.yaml

@@ -1,8 +1,10 @@
 version: 1
-delivery: issue384
+delivery: end-structured-output
 context:
   kind: branch
-  branch: feat/384-issue384
+  branch: fix/386-end-structured-output
 issues:
-  - 384
-pr: 385
+  - 386
+  - 387
+  - 388
+pr: 390

packages/opencode/src/dag/blocks.ts

@@ -332,7 +332,15 @@ function node(input: {
   review?: NodeConfig["review"]
   outputSchema?: Record<string, unknown>
 }): NodeConfig {
-  const instruction = input.instruction?.trim() ? "Block-specific instruction:\n{{instruction}}" : ""
+  // issue #387: an instruction equal to the objective (after trim and
+  // line-ending normalization) would render the same content twice in the
+  // single child prompt — the objective section already carries it, so the
+  // instruction is dropped instead of duplicated. Equivalence is exact, not
+  // fuzzy: an instruction carrying the objective plus additional detail stays.
+  const equivalent = (a: string, b: string) => a.trim().replace(/\r\n/g, "\n") === b.trim().replace(/\r\n/g, "\n")
+  const instructionText = input.instruction?.trim() ?? ""
+  const hasInstruction = instructionText !== "" && !equivalent(instructionText, input.objective)
+  const instruction = hasInstruction ? "Block-specific instruction:\n{{instruction}}" : ""
   // issue #323: a reporting checkpoint adjudicates a direction, so its
   // prompt must demand adversarial independent verification. The production
   // incident: a gate confirmed parent-supplied "defect evidence" that was a
@@ -360,7 +368,7 @@ function node(input: {
         .join("\n\n"),
       input: {
         objective: input.objective,
-        ...(input.instruction?.trim() ? { instruction: input.instruction.trim() } : {}),
+        ...(hasInstruction ? { instruction: instructionText } : {}),
       },
     },
     ...(input.condition ? { condition: input.condition } : {}),
@@ -521,7 +529,7 @@ function reviewWriterTopology(block: WorkflowBlock, blocks: WorkflowBlock[]): Re
       `Implementation review "${block.id}" requires exactly one verification ancestor; found ${verifications.length}`,
     )
   }
-  const verification = verifications[0]!
+  const verification = verifications[0]
   const verifiedImplementations = implementations.filter((candidate) =>
     dependsTransitively(blocks, verification.id, candidate.id),
   )

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

@@ -278,7 +278,7 @@ const serviceLayer = Layer.effect(
             if (nodeConfig?.output_schema) {
               promptParts.push({
                 type: "text",
-                text: `\n\nYou MUST call the submit_result tool with a JSON payload matching this schema before ending your turn:\n${JSON.stringify(nodeConfig.output_schema, null, 2)}`,
+                text: `\n\nYou MUST call the submit_result tool with a JSON payload matching this schema before ending your turn:\n${JSON.stringify(nodeConfig.output_schema, null, 2)}\nPut your full summary inside the payload. Do not repeat the payload in your message text. After submit_result succeeds, end your turn without restating the result.`,
               })
             }
 
@@ -423,6 +423,7 @@ const serviceLayer = Layer.effect(
               // fails such nodes loudly instead of undefined-completing them.
               config ?? null,
               lastAssistantText,
+              ctx.directory,
             ).pipe(
               Effect.provideService(Dag.Service, dag),
             )

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

@@ -32,13 +32,15 @@ import { reviewImplementationFingerprint } from "../review-lifecycle"
 import { resolveInputMapping } from "./eval"
 import { settleCapturedOutput } from "./capture"
 import type { CapturedSettlement } from "./capture"
+import { captureOutputFileRef, ensureReportAreaGitignore } from "./output-ref"
 
 export function reconcileWorkflow(
   dagID: string,
   checkSessionStatus: (childSessionID: string) => Effect.Effect<"active" | "completed" | "failed" | "unknown", Error>,
   cancelSession?: (sessionID: string) => Effect.Effect<void, Error>,
   workflowConfig?: { nodes: Pick<NodeConfig, "id" | "output_schema" | "review" | "input_mapping">[] } | null,
   lastAssistantText?: (childSessionID: string) => Effect.Effect<string | undefined, Error>,
+  directory?: string,
 ): Effect.Effect<{ reconciled: number; ownershipLost: number }, Error, Dag.Service> {
   return Effect.gen(function* () {
     const dag = yield* Dag.Service
@@ -143,6 +145,27 @@ export function reconcileWorkflow(
           const rawText = lastAssistantText
             ? (yield* lastAssistantText(node.childSessionId)) ?? ""
             : undefined
+          // #388 parity with the live path: when the recovered reply IS one
+          // existing absolute file path, capture the same {content_ref, size,
+          // sha256, summary} receipt submit-time detection records, so live
+          // and recovered settlement produce identical durable output
+          // metadata. Best-effort like the live path — any anomaly keeps the
+          // plain inline completion and never fails the node.
+          if (rawText) {
+            const fileRef = yield* captureOutputFileRef(rawText)
+            if (fileRef) {
+              yield* dag.store.setCapturedOutput(node.childSessionId, fileRef).pipe(
+                Effect.catchCause((cause) =>
+                  Effect.logWarning("DAG recovery output-ref capture persistence failed — inline output preserved", {
+                    dagID,
+                    nodeID: node.id,
+                    cause,
+                  }),
+                ),
+              )
+              if (directory) yield* ensureReportAreaGitignore(directory, fileRef.path)
+            }
+          }
           yield* settle(node.id, dag.nodeCompleted(dagID, node.id, rawText))
         }
         reconciled++

packages/opencode/src/tool/submit_result.txt

@@ -4,4 +4,6 @@ This tool is only relevant when you are running as a child session of a DAG work
 
 If the payload does not match the schema, the tool returns a validation error — correct the payload and call again within the same session. The result is not final until this tool succeeds.
 
+The payload is the single authoritative report: put the full result, including any summary, inside the payload itself. Do not duplicate the payload in your message text. Once submit_result succeeds, end your turn without restating the result.
+
 If you are not in a DAG workflow child session, this tool has no effect.

packages/opencode/test/dag/blocks.test.ts

@@ -456,4 +456,47 @@ describe("workflow blocks", () => {
       }),
     ).toThrow("depends on multiple review gates")
   })
+
+  // issue #387: an instruction that duplicates the objective (after
+  // trim/line-ending normalization) must not be emitted twice in the single
+  // child prompt — the objective section already carries the content.
+  it("drops an instruction that duplicates the objective (issue #387)", () => {
+    const duplicated = DagBlocks.compileWorkflowBlocks({
+      objective: "Ship the memory feature",
+      blocks: [{ id: "map", kind: "explore", instruction: "Ship the memory feature" }],
+    })
+    expect(duplicated[0]?.prompt_template.inline).not.toContain("Block-specific instruction")
+    expect(duplicated[0]?.prompt_template.input).not.toHaveProperty("instruction")
+
+    const whitespaceEquivalent = DagBlocks.compileWorkflowBlocks({
+      objective: "Ship the memory feature",
+      blocks: [{ id: "map", kind: "explore", instruction: "  Ship the memory feature\r\n" }],
+    })
+    expect(whitespaceEquivalent[0]?.prompt_template.inline).not.toContain("Block-specific instruction")
+    expect(whitespaceEquivalent[0]?.prompt_template.input).not.toHaveProperty("instruction")
+  })
+
+  // Equivalence is exact, not fuzzy: block-specific instructions survive —
+  // both a fully distinct one and one that carries the objective plus
+  // additional detail — ordered after the objective.
+  it("keeps block-specific instructions that extend the objective (issue #387)", () => {
+    const distinct = DagBlocks.compileWorkflowBlocks({
+      objective: "Ship the memory feature",
+      blocks: [{ id: "map", kind: "explore", instruction: "Focus on the persistence seam" }],
+    })
+    const distinctInline = distinct[0]?.prompt_template.inline ?? ""
+    expect(distinctInline).toContain("Block-specific instruction:\n{{instruction}}")
+    expect(distinct[0]?.prompt_template.input).toMatchObject({ instruction: "Focus on the persistence seam" })
+
+    const extended = DagBlocks.compileWorkflowBlocks({
+      objective: "Ship the memory feature",
+      blocks: [{ id: "map", kind: "explore", instruction: "Ship the memory feature, then profile the persistence seam" }],
+    })
+    const extendedInline = extended[0]?.prompt_template.inline ?? ""
+    expect(extendedInline).toContain("Block-specific instruction:\n{{instruction}}")
+    expect(extended[0]?.prompt_template.input).toMatchObject({
+      instruction: "Ship the memory feature, then profile the persistence seam",
+    })
+    expect(extendedInline.indexOf("Workflow objective")).toBeLessThan(extendedInline.indexOf("Block-specific instruction"))
+  })
 })

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

@@ -1,12 +1,26 @@
-import { describe, expect, it } from "bun:test"
+// oxlint-disable typescript-eslint/no-unsafe-type-assertion -- mock dag
+// layers and row fixtures use `as unknown as DagStore.Interface` shims that
+// implement only the interface slice each scenario exercises.
+import { describe, expect, it, afterAll } from "bun:test"
+import { createHash } from "node:crypto"
+import fs from "node:fs/promises"
+import os from "node:os"
+import path from "node:path"
 import { Effect, Exit, Layer } from "effect"
-import { reconcileWorkflow } from "@/dag/runtime/recovery"
+import type { SessionV1 } from "@opencode-ai/core/v1/session"
+import { reconcileWorkflow, makeLastAssistantTextReader } from "@/dag/runtime/recovery"
 import { Dag } from "@/dag/dag"
 import type { DagStore } from "@opencode-ai/core/dag/store"
 import { WorkflowRuntime, toSchedulingNodes } from "@opencode-ai/core/dag/core/scheduling"
 import { TerminalViolationError } from "@opencode-ai/core/dag/core/types"
 import { makeNodeRow } from "./fixtures"
 
+const tmpRoots: string[] = []
+
+afterAll(async () => {
+  for (const dir of tmpRoots) await fs.rm(dir, { recursive: true, force: true })
+})
+
 type TrackedEvent = {
   type: string
   nodeID: string
@@ -15,11 +29,21 @@ type TrackedEvent = {
   trigger?: string
 }
 
-function makeDagLayer(nodes: DagStore.NodeRow[], trackedEvents: TrackedEvent[], actions?: string[]) {
+function makeDagLayer(
+  nodes: DagStore.NodeRow[],
+  trackedEvents: TrackedEvent[],
+  actions?: string[],
+  capturedWrites?: { sid: string; payload: unknown }[],
+  opts?: { capturedFail?: boolean },
+) {
   return Layer.mock(Dag.Service, {
     store: {
       getNodes: () => Effect.succeed(nodes),
       getNode: (id: string) => Effect.succeed(nodes.find((n) => n.id === id)),
+      setCapturedOutput: (sid: string, payload: unknown) =>
+        opts?.capturedFail
+          ? Effect.fail(new Error("captured output persistence boom"))
+          : Effect.sync(() => capturedWrites?.push({ sid, payload })),
     } as unknown as DagStore.Interface,
     nodeCompleted: Effect.fn("stub.nodeCompleted")((dagID: string, nodeID: string, output: unknown) =>
       Effect.sync(() => trackedEvents.push({
@@ -511,3 +535,133 @@ describe("rehydration via toSchedulingNodes", () => {
     expect(events).toContainEqual({ type: "nodeCompleted", nodeID: "n1" })
   })
 })
+
+// issue #388: the live path captures {content_ref, size, sha256, summary}
+// when a schemaless node's final reply IS one existing absolute file path
+// (spawn.ts → output-ref.ts). Recovery must produce the same durable receipt
+// for the same reply instead of diverging by crash timing. Behavioral
+// lockstep with the live path is asserted side-by-side in
+// dag-wake-integration.test.ts "captures a file_ref receipt when a
+// schemaless reply is one absolute path (issue #388)" — keep both green or
+// neither ships.
+describe("reconcileWorkflow output file refs (issue #388)", () => {
+  it("captures the same file_ref receipt as the live path for an absolute-path reply", async () => {
+    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "dag-recovery-ref-"))
+    tmpRoots.push(dir)
+    const reportPath = path.join(dir, "report.md")
+    const content = "recovered report body"
+    await Bun.write(reportPath, content)
+
+    const events: TrackedEvent[] = []
+    const captured: { sid: string; payload: unknown }[] = []
+    const nodes = [makeNodeRow({ id: "n1", status: "running", childSessionId: "ses_1" })]
+    const dagLayer = makeDagLayer(nodes, events, undefined, captured)
+    const checkStatus = () => Effect.succeed<"active" | "completed" | "failed" | "unknown">("completed")
+
+    await Effect.runPromise(
+      reconcileWorkflow(
+        "wf-1",
+        checkStatus,
+        undefined,
+        { nodes: [{ id: "n1" }] },
+        () => Effect.succeed(reportPath),
+        dir,
+      ).pipe(Effect.provide(dagLayer)),
+    )
+
+    expect(events).toContainEqual({ type: "nodeCompleted", nodeID: "n1", output: reportPath })
+    expect(captured).toEqual([{
+      sid: "ses_1",
+      payload: {
+        kind: "file_ref",
+        content_ref: reportPath,
+        path: reportPath,
+        size: Buffer.byteLength(content),
+        sha256: createHash("sha256").update(content).digest("hex"),
+        summary: content,
+      },
+    }])
+  })
+
+  it("keeps the inline settlement and captures nothing when the reply is not an existing path", async () => {
+    const events: TrackedEvent[] = []
+    const captured: { sid: string; payload: unknown }[] = []
+    const nodes = [makeNodeRow({ id: "n1", status: "running", childSessionId: "ses_1" })]
+    const dagLayer = makeDagLayer(nodes, events, undefined, captured)
+    const checkStatus = () => Effect.succeed<"active" | "completed" | "failed" | "unknown">("completed")
+
+    await Effect.runPromise(
+      reconcileWorkflow(
+        "wf-1",
+        checkStatus,
+        undefined,
+        { nodes: [{ id: "n1" }] },
+        () => Effect.succeed(`Report written to ${path.join(os.tmpdir(), "dag-recovery-ghost.md")}`),
+        process.cwd(),
+      ).pipe(Effect.provide(dagLayer)),
+    )
+
+    expect(captured).toEqual([])
+    const completed = events.find((event) => event.type === "nodeCompleted")
+    expect(completed?.output).toMatch(/^Report written to /)
+  })
+
+  // #388 best-effort contract: a captured-output persistence failure logs a
+  // warning and NEVER fails the node — the inline completion survives.
+  it("completes inline even when output-ref persistence fails (issue #388)", async () => {
+    const dir = await fs.mkdtemp(path.join(os.tmpdir(), "dag-recovery-refboom-"))
+    tmpRoots.push(dir)
+    const reportPath = path.join(dir, "report.md")
+    await Bun.write(reportPath, "doomed receipt")
+    const events: TrackedEvent[] = []
+    const nodes = [makeNodeRow({ id: "n1", status: "running", childSessionId: "ses_1" })]
+    const dagLayer = makeDagLayer(nodes, events, undefined, undefined, { capturedFail: true })
+    const checkStatus = () => Effect.succeed<"active" | "completed" | "failed" | "unknown">("completed")
+
+    await Effect.runPromise(
+      reconcileWorkflow(
+        "wf-1",
+        checkStatus,
+        undefined,
+        { nodes: [{ id: "n1" }] },
+        () => Effect.succeed(reportPath),
+        dir,
+      ).pipe(Effect.provide(dagLayer)),
+    )
+
+    expect(events).toContainEqual({ type: "nodeCompleted", nodeID: "n1", output: reportPath })
+    expect(events).not.toContainEqual({ type: "nodeFailed", nodeID: "n1" })
+  })
+})
+
+// #345: the schemaless completion mirror — recovery reads the child's last
+// assistant text exactly as spawn settles it. Direct reader contract.
+describe("makeLastAssistantTextReader (#345)", () => {
+  function assistantText(text: string): SessionV1.WithParts {
+    return {
+      info: {
+        id: "m1",
+        role: "assistant",
+        sessionID: "ses_1",
+        time: { created: 0 },
+        agent: "build",
+        model: { providerID: "p", modelID: "m" },
+      },
+      parts: [{ type: "text", text }],
+    } as never
+  }
+
+  it("returns the last assistant text part from the child transcript", async () => {
+    const reader = makeLastAssistantTextReader({
+      messages: () => Effect.succeed([assistantText("attempt one"), assistantText("final verdict: GO")]),
+    } as never)
+    expect(await Effect.runPromise(reader("ses_1"))).toBe("final verdict: GO")
+  })
+
+  it("treats a missing child session as no text instead of failing recovery", async () => {
+    const reader = makeLastAssistantTextReader({
+      messages: () => Effect.fail({ _tag: "NotFoundError", message: "session gone" } as never),
+    } as never)
+    expect(await Effect.runPromise(reader("ses_ghost"))).toBeUndefined()
+  })
+})