diff --git a/.specgit.yaml b/.specgit.yaml index a9719a5a2..39ba8f3ba 100644 --- a/.specgit.yaml +++ b/.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 diff --git a/packages/opencode/src/dag/blocks.ts b/packages/opencode/src/dag/blocks.ts index fc010a014..38bb0a272 100644 --- a/packages/opencode/src/dag/blocks.ts +++ b/packages/opencode/src/dag/blocks.ts @@ -332,7 +332,15 @@ function node(input: { review?: NodeConfig["review"] outputSchema?: Record }): 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), ) diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 27e1fb289..0f37a9c25 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/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), ) diff --git a/packages/opencode/src/dag/runtime/recovery.ts b/packages/opencode/src/dag/runtime/recovery.ts index cd7647324..5013b39a5 100644 --- a/packages/opencode/src/dag/runtime/recovery.ts +++ b/packages/opencode/src/dag/runtime/recovery.ts @@ -32,6 +32,7 @@ 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, @@ -39,6 +40,7 @@ export function reconcileWorkflow( cancelSession?: (sessionID: string) => Effect.Effect, workflowConfig?: { nodes: Pick[] } | null, lastAssistantText?: (childSessionID: string) => Effect.Effect, + 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++ diff --git a/packages/opencode/src/tool/submit_result.txt b/packages/opencode/src/tool/submit_result.txt index 14b792908..c523c9fd2 100644 --- a/packages/opencode/src/tool/submit_result.txt +++ b/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. diff --git a/packages/opencode/test/dag/blocks.test.ts b/packages/opencode/test/dag/blocks.test.ts index a36184d27..cf6e22011 100644 --- a/packages/opencode/test/dag/blocks.test.ts +++ b/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")) + }) }) diff --git a/packages/opencode/test/dag/dag-recovery.test.ts b/packages/opencode/test/dag/dag-recovery.test.ts index 322e78b77..a24275bf0 100644 --- a/packages/opencode/test/dag/dag-recovery.test.ts +++ b/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() + }) +}) diff --git a/packages/opencode/test/dag/dag-schema-prompt-contract.test.ts b/packages/opencode/test/dag/dag-schema-prompt-contract.test.ts new file mode 100644 index 000000000..24eb8b72b --- /dev/null +++ b/packages/opencode/test/dag/dag-schema-prompt-contract.test.ts @@ -0,0 +1,241 @@ +// oxlint-disable typescript-eslint/no-unsafe-type-assertion -- harness +// deliberately mirrors dag-wake-integration.test.ts: mocked service layers and +// row fixtures use `as never` type shims (mock objects implement only the +// interface slice the scenario exercises). The shims are type-only. +// SPDX-FileCopyrightText: 2026 LeXwDeX +// SPDX-License-Identifier: AGPL-3.0-or-later + +/** + * Issue #386 — structured-output single-authority contract. + * + * A DAG node with output_schema must instruct the child, up front, that the + * submit_result payload is the single authoritative report: the summary lives + * inside the payload, prose must not duplicate it, and a successful submission + * ends the turn. Both delivery channels carry the contract: the DAG-generated + * schema instruction part (loop.ts) and the submit_result tool description. + */ +import { describe, expect, it } from "bun:test" +import path from "node:path" +import { Deferred, Effect, Layer, Option, Queue } from "effect" +import type { SessionV1 } from "@opencode-ai/core/v1/session" +import { Database } from "@opencode-ai/core/database/database" +import { DagProjector } from "@opencode-ai/core/dag/projector" +import { DagStore } from "@opencode-ai/core/dag/store" +import { EventV2 } from "@opencode-ai/core/event" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { Agent } from "@/agent/agent" +import { Dag, type NodeConfig } from "@/dag/dag" +import { DagLoop } from "@/dag/runtime/loop" +import { InstanceRef } from "@/effect/instance-ref" +import { EventV2Bridge } from "@/event-v2-bridge" +import { SessionPrompt } from "@/session/prompt" +import { MessageID } from "@/session/schema" +import { Session } from "@/session/session" +import { SessionStatus } from "@/session/status" +import { pollWithTimeout } from "../lib/effect" +import { withIdleAdmission } from "../lib/session-prompt" + +interface PromptGate { + readonly input: SessionPrompt.PromptInput + readonly release: Deferred.Deferred +} + +function reply(sessionID: string, text: string): SessionV1.WithParts { + return { + info: { + id: MessageID.ascending(), + role: "assistant", + parentID: MessageID.ascending(), + sessionID: sessionID as never, + mode: "build", + agent: "build", + cost: 0, + path: { cwd: process.cwd(), root: process.cwd() }, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: "test-model" as never, + providerID: "test" as never, + time: { created: Date.now() }, + finish: "stop", + }, + parts: text ? [{ type: "text", text }] as never : [], + } +} + +function schemaNode(): NodeConfig { + return { + id: "report", + name: "report", + worker_type: "build", + depends_on: [], + required: true, + prompt_template: { inline: "Summarize the delivery" }, + output_schema: { + type: "object", + required: ["summary"], + properties: { summary: { type: "string" } }, + }, + } +} + +function contractLayer(childPrompts: Queue.Queue) { + 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) + const session = Layer.mock(Session.Service, { + get: () => Effect.succeed({ id: "ses_parent", permission: [], agent: "build" } as never), + create: () => Effect.succeed({ id: "ses_child_1" } as never), + messages: () => Effect.succeed([]), + }) + const deliver = Effect.fn("test.SessionPrompt.deliver")(function* (value: SessionPrompt.PromptInput) { + const sessionID = value.sessionID as string + const release = yield* Deferred.make() + yield* Queue.offer(childPrompts, { input: value, release }) + return reply(sessionID, yield* Deferred.await(release)) + }) + const prompt = Layer.mock(SessionPrompt.Service, withIdleAdmission({ + cancel: () => Effect.void, + prompt: deliver, + promptIfIdle: (value) => deliver(value).pipe(Effect.map(Option.some)), + })) + const agent = Layer.mock(Agent.Service, { + get: () => Effect.succeed({ + name: "build", + mode: "all", + permission: [], + options: {}, + description: "", + prompt: "", + model: { providerID: "test" as never, modelID: "test-model" as never }, + tools: {}, + hooks: {}, + }), + }) + const loop = DagLoop.layer.pipe( + Layer.provide(base), + Layer.provide(session), + Layer.provide(prompt), + Layer.provide(agent), + ) + return Layer.merge(base, loop) +} + +function runContractTest(test: (services: { + readonly dag: Dag.Interface + readonly loop: DagLoop.Interface + readonly store: DagStore.Interface + readonly childPrompts: Queue.Queue +}) => Effect.Effect) { + return Effect.gen(function* () { + const childPrompts = yield* Queue.unbounded() + return yield* Effect.gen(function* () { + const dag = yield* Dag.Service + const loop = yield* DagLoop.Service + const store = yield* DagStore.Service + const database = yield* Database.Service + yield* database.db.insert(ProjectTable).values({ + id: "project-1" as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.db.insert(SessionTable).values({ + id: "ses_parent" as never, + project_id: "project-1" as never, + slug: "parent", + directory: process.cwd(), + title: "Parent", + version: "test", + }).run().pipe(Effect.orDie) + yield* loop.init() + return yield* test({ dag, loop, store, childPrompts }) + }).pipe( + Effect.provide(contractLayer(childPrompts)), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: "project-1" }, + } as never), + Effect.scoped, + ) + }) +} + +function promptText(input: SessionPrompt.PromptInput) { + return input.parts + .filter((part) => part.type === "text") + .map((part) => part.text) + .join("\n") +} + +describe("DAG schema prompt contract (issue #386)", () => { + it("schema instruction tells the child the payload is the single authoritative report", async () => { + await Effect.runPromise( + runContractTest(({ dag, childPrompts }) => + Effect.gen(function* () { + yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Schema prompt contract", + config: { name: "schema-prompt-contract", nodes: [schemaNode()] }, + }) + const gate = yield* Queue.take(childPrompts) + const prompt = promptText(gate.input) + expect(prompt).toContain("submit_result") + // Summary belongs inside the payload, not in prose. + expect(prompt).toContain("Put your full summary inside the payload") + // Prose must not duplicate the payload. + expect(prompt).toContain("Do not repeat the payload in your message text") + // A successful submission ends the turn. + expect(prompt).toContain("end your turn without restating the result") + yield* Deferred.succeed(gate.release, "done") + }), + ), + ) + }) + + // The issue-#386 acceptance chain does not stop at prompt construction: + // the child submits through submit_result, the capture lands durably, and + // the node settles with the payload as its durable output — prose plays no + // part in settlement. The gate replays exactly the durable write the real + // tool performs (store.setCapturedOutput); spawn's completion gate + // (settleCapturedOutput) runs unmocked below it. + it("settles the node from the submit_result payload as the durable output", async () => { + await Effect.runPromise( + runContractTest(({ dag, store, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Schema prompt contract", + config: { name: "schema-prompt-contract", nodes: [schemaNode()] }, + }) + const gate = yield* Queue.take(childPrompts) + const payload = { summary: "Delivered through submit_result only." } + yield* store.setCapturedOutput(gate.input.sessionID as string, payload) + // The contract-compliant reply: no payload duplication in prose. + yield* Deferred.succeed(gate.release, "Submitted.") + const node = yield* pollWithTimeout( + store.getNode(dagID, "report").pipe( + Effect.map((row) => row?.status === "completed" ? row : undefined), + ), + "schema node did not complete from the submitted payload", + ) + expect(node.output).toEqual(payload) + }), + ), + ) + }) + + it("submit_result tool description carries the same single-authority contract", async () => { + const description = await Bun.file( + path.join(import.meta.dir, "../../src/tool/submit_result.txt"), + ).text() + expect(description).toContain("Do not duplicate the payload") + expect(description).toContain("end your turn without restating the result") + }) +}) diff --git a/packages/opencode/test/dag/dag-wake-integration.test.ts b/packages/opencode/test/dag/dag-wake-integration.test.ts index 560e48e45..c41dc0091 100644 --- a/packages/opencode/test/dag/dag-wake-integration.test.ts +++ b/packages/opencode/test/dag/dag-wake-integration.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it } from "bun:test" +import { createHash } from "node:crypto" import * as fs from "node:fs/promises" +import * as os from "node:os" import * as path from "node:path" import { Deferred, Effect, Fiber, Layer, Option, Queue } from "effect" import type { SessionV1 } from "@opencode-ai/core/v1/session" @@ -559,6 +561,53 @@ describe("DagLoop atomic wake integration", () => { ) }) + // issue #388 live path: when a schemaless node's final reply IS one + // existing absolute file path, submit-time detection records the durable + // {content_ref, size, sha256, summary} receipt while the settlement stays + // the raw path. Keep in lockstep with dag-recovery.test.ts + // "reconcileWorkflow output file refs (issue #388)" — live and recovery + // must produce identical durable effects for the same reply. + it("captures a file_ref receipt when a schemaless reply is one absolute path (issue #388)", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "dag-live-ref-")) + const reportPath = path.join(dir, "report.md") + const content = "live report body" + await fs.writeFile(reportPath, content) + try { + await Effect.runPromise( + runWakeTest(({ dag, store, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "File-ref live capture", + config: { name: "file-ref-live-capture", nodes: [node("file-report")] }, + }) + + const report = yield* takeWithin(childPrompts, "file-report did not start") + yield* Deferred.succeed(report.release, reportPath) + const row = yield* pollWithTimeout( + store.getNode(dagID, "file-report").pipe( + Effect.map((item) => item?.status === "completed" ? item : undefined), + ), + "file-ref node did not complete", + ) + expect(row.output).toBe(reportPath) + expect(row.capturedOutput).toEqual({ + kind: "file_ref", + content_ref: reportPath, + path: reportPath, + size: Buffer.byteLength(content), + sha256: createHash("sha256").update(content).digest("hex"), + summary: content, + }) + }), + ), + ) + } finally { + await fs.rm(dir, { recursive: true, force: true }) + } + }) + integration.live("runs an additive wave after a terminal checkpoint wake", () => runWakeTest(({ dag, store, childPrompts, parentPrompts }) => Effect.gen(function* () {