From 342555393a4bcd663bc4e9fa3164763cbd050a5d Mon Sep 17 00:00:00 2001 From: Lex Date: Tue, 18 Aug 2026 06:57:23 +0800 Subject: [PATCH] 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 c8c6cb60b..26d1a7ae8 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 @@ -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 43d1e35cf..fce8c31d3 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") + }), + ), + ) + }) })