From 726d4b9a2caf86fbf89b30385b15778b286310a7 Mon Sep 17 00:00:00 2001 From: Lex Date: Sun, 16 Aug 2026 19:31:15 +0800 Subject: [PATCH] fix(dag): complete workflows at rejected review checkpoints instead of failing them (issue #294) --- packages/opencode/src/dag/dag.ts | 16 ++++++++++++---- packages/opencode/src/dag/runtime/loop.ts | 15 ++++++--------- packages/opencode/src/tool/workflow.ts | 8 ++++++++ .../test/dag/dag-wake-integration.test.ts | 11 +++++++---- 4 files changed, 33 insertions(+), 17 deletions(-) diff --git a/packages/opencode/src/dag/dag.ts b/packages/opencode/src/dag/dag.ts index bb0b980538..1c83a45092 100644 --- a/packages/opencode/src/dag/dag.ts +++ b/packages/opencode/src/dag/dag.ts @@ -262,7 +262,7 @@ export interface Interface { readonly resume: (dagID: string) => Effect.Effect readonly step: (dagID: string) => Effect.Effect<{ status: "stepping"; nodeID?: string } | { status: "no_ready_nodes" }, Error> readonly cancel: (dagID: string) => Effect.Effect - readonly complete: (dagID: string) => Effect.Effect + readonly complete: (dagID: string, options?: { readonly skipReviewGate?: boolean }) => Effect.Effect readonly fail: (dagID: string, reason: string) => Effect.Effect readonly replan: (dagID: string, fragment: { nodes: NodeConfig[] }) => Effect.Effect< { cancel: string[]; restart: string[]; replace: string[]; add: string[]; ignore: string[] }, @@ -502,11 +502,19 @@ export const layer = Layer.effect( yield* events.publish(DagEvent.WorkflowCancelled, { dagID: dagID as ID, timestamp: yield* DateTime.now }) yield* terminateNonTerminalNodes(lock, dagID, "workflow_cancelled", "workflow_cancelled", false) }) - const complete = Effect.fn("Dag.complete")(function* (lock: WorkflowLock, dagID: string) { + const complete = Effect.fn("Dag.complete")(function* ( + lock: WorkflowLock, + dagID: string, + options?: { readonly skipReviewGate?: boolean }, + ) { yield* guardWorkflow(dagID, WorkflowStatus.COMPLETED) const workflow = yield* store.getWorkflow(dagID).pipe(Effect.orDie) const config = workflow ? parseWorkflowConfig(workflow.config) : undefined - const unresolvedReviews = config + // The review gate guards EXPLICIT completion (tool/HTTP shortcuts). A + // natural completion at a REJECT checkpoint must stay reachable so the + // parent can dispose of the verdict via reopen-extend (issue #294); the + // scheduling loop passes skipReviewGate for that path. + const unresolvedReviews = !options?.skipReviewGate && config ? unresolvedReviewOutcomes(config, yield* store.getNodes(dagID)) : [] if (unresolvedReviews.length > 0) yield* Effect.fail(new ReviewGateError(dagID, unresolvedReviews)) @@ -898,7 +906,7 @@ export const layer = Layer.effect( resume: (dagID) => withWorkflowLock(dagID)((lock) => resume(lock, dagID)), step: (dagID) => withWorkflowLock(dagID)((lock) => step(lock, dagID)), cancel: (dagID) => withWorkflowLock(dagID)((lock) => cancel(lock, dagID)), - complete: (dagID) => withWorkflowLock(dagID)((lock) => complete(lock, dagID)), + complete: (dagID, options) => withWorkflowLock(dagID)((lock) => complete(lock, dagID, options)), fail: (dagID, reason) => withWorkflowLock(dagID)((lock) => fail(lock, dagID, reason)), replan: (dagID, fragment) => withWorkflowLock(dagID)((lock) => _replan(lock, dagID, fragment)), extend: (dagID, nodes) => withWorkflowLock(dagID)((lock) => _extend(lock, dagID, nodes)), diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 9076807b7c..a725cdd563 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -22,7 +22,6 @@ import { reviewContractForNode, validateReviewExecutionInput, reviewEvidenceKeys, - unresolvedReviewOutcomes, } from "../review-lifecycle" import { Agent } from "@/agent/agent" import { Session } from "@/session/session" @@ -325,14 +324,12 @@ const serviceLayer = Layer.effect( yield* dag.fail(dagID, `required node(s) failed: ${entry.runtime.getRequiredFailures().join(", ")}`) return } - const unresolvedReviews = entry.config - ? unresolvedReviewOutcomes(entry.config, nodes) - : [] - if (unresolvedReviews.length > 0) { - yield* dag.fail(dagID, `unresolved review outcome(s): ${unresolvedReviews.join(", ")}`) - return - } - yield* dag.complete(dagID) + // Unresolved review outcomes terminalize the graph as COMPLETED at + // the checkpoint, not as a failure: the REJECT shape (skipped + // dependents, reporting leaf) is exactly what reopen-extend is + // designed to pick up, and a failed workflow is immutable + // (issue #294). Explicit completion shortcuts keep the review gate. + yield* dag.complete(dagID, { skipReviewGate: true }) }) const checkSessionStatus = makeSessionStatusChecker(sessionSvc) diff --git a/packages/opencode/src/tool/workflow.ts b/packages/opencode/src/tool/workflow.ts index 58c42b86de..a592593889 100644 --- a/packages/opencode/src/tool/workflow.ts +++ b/packages/opencode/src/tool/workflow.ts @@ -5,6 +5,7 @@ import { Tool } from "./tool" import { CommandPlugin } from "@opencode-ai/core/plugin/command" import { Effect, Option, Schema } from "effect" import { Dag } from "@/dag/dag" +import { DagReviewLifecycle } from "@/dag/review-lifecycle" import { DagConfig } from "@/dag/config" import { DagWorkflows } from "@/dag/workflows" import { DagModel } from "@/dag/model" @@ -451,6 +452,12 @@ export const WorkflowTool = Tool.define< // stay reachable via the result seam (getNode is unfiltered). const nodes = yield* dag.store.getCurrentNodes(params.workflow_id).pipe(Effect.orDie) const config = Dag.parseWorkflowConfig(workflow.config) + // A completed graph can still carry an unresolved review verdict + // (REJECT checkpoint, issue #294); surface it explicitly instead + // of burying it in a terminal reason string. + const unresolvedReviews = config + ? DagReviewLifecycle.unresolvedReviewOutcomes(config, nodes) + : [] return { title: `Workflow status: ${workflow.title}`, output: JSON.stringify( @@ -460,6 +467,7 @@ export const WorkflowTool = Tool.define< status: workflow.status, session_id: workflow.sessionId, mode: config?.mode ?? "standard", + ...(unresolvedReviews.length > 0 ? { unresolved_reviews: unresolvedReviews } : {}), ...(config?.admission ? { admission: { diff --git a/packages/opencode/test/dag/dag-wake-integration.test.ts b/packages/opencode/test/dag/dag-wake-integration.test.ts index 451d9574e9..fe72bf4a76 100644 --- a/packages/opencode/test/dag/dag-wake-integration.test.ts +++ b/packages/opencode/test/dag/dag-wake-integration.test.ts @@ -1260,7 +1260,10 @@ describe("DagLoop atomic wake integration", () => { ) }) - it.each(["deep", "standard"] as const)("fails a recovered %s workflow when verification skips every diff review", async (mode) => { + // Issue #294: a workflow whose verification skips every diff review settles + // as COMPLETED at the checkpoint (not failed) so the parent can dispose of + // the unresolved review verdict via reopen-extend; failed stays immutable. + it.each(["deep", "standard"] as const)("completes a recovered %s workflow when verification skips every diff review", async (mode) => { await Effect.runPromise( runWakeTest( ({ store, parentPrompts }) => @@ -1271,13 +1274,13 @@ describe("DagLoop atomic wake integration", () => { ), "recovered workflow without an accepted review did not settle", ) - expect(workflow.status).toBe("failed") + expect(workflow.status).toBe("completed") expect((yield* store.getNode(workflow.id, "review-diff"))?.status).toBe("skipped") expect((yield* store.getNode(workflow.id, "final-audit"))?.status).toBe("skipped") - const parent = yield* takeWithin(parentPrompts, "review rejection failure did not wake the parent") + const parent = yield* takeWithin(parentPrompts, "review rejection completion did not wake the parent") expect(promptText(parent.input)).toContain( - '[DAG Workflow failed] Workflow "Recovered review rejection" has reached terminal status.', + '[DAG Workflow completed] Workflow "Recovered review rejection" has reached terminal status.', ) yield* Deferred.succeed(parent.release, "success") }),