From a231f18ac451c0a2589c863f972d83ecc95a13a6 Mon Sep 17 00:00:00 2001 From: lex Date: Wed, 29 Jul 2026 15:56:37 +0800 Subject: [PATCH] fix(core): drop stale dag node terminal events via db status cross-check Loop handlers for NodeCompleted/NodeSkipped/NodeFailed now confirm the node row still reflects the event's terminal status before touching fibers or runtime state, so events from a previous generation (after NodeRestarted/replan) are dropped. Schema capture validation additionally enforces const/enum via structural equality. --- packages/opencode/src/dag/runtime/capture.ts | 30 ++ packages/opencode/src/dag/runtime/loop.ts | 80 ++++- .../dag/dag-replan-stale-nodefailed.test.ts | 317 ++++++++++++++++++ .../test/dag/dag-structured-output.test.ts | 80 +++++ .../test/dag/dag-wake-integration.test.ts | 35 ++ 5 files changed, 525 insertions(+), 17 deletions(-) create mode 100644 packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts diff --git a/packages/opencode/src/dag/runtime/capture.ts b/packages/opencode/src/dag/runtime/capture.ts index 45ae0ec08..119210169 100644 --- a/packages/opencode/src/dag/runtime/capture.ts +++ b/packages/opencode/src/dag/runtime/capture.ts @@ -49,6 +49,13 @@ export function validateAgainstSchema(value: unknown, schema: Record deepEqual(value, v))) + return { ok: false, error: `expected one of ${truncate(JSON.stringify(enumVals))}, got ${truncate(JSON.stringify(value))}` } + const required = schema["required"] if (Array.isArray(required) && typeof value === "object" && value !== null && !Array.isArray(value)) { const obj = value as Record @@ -80,3 +87,26 @@ export function validateAgainstSchema(value: unknown, schema: Record deepEqual(item, b[i])) + } + const aObj = a as Record + const bObj = b as Record + const aKeys = Object.keys(aObj) + if (aKeys.length !== Object.keys(bObj).length) return false + return aKeys.every((key) => key in bObj && deepEqual(aObj[key], bObj[key])) +} diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 6852073ed..b78cbf9fe 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -394,6 +394,14 @@ export const layer = Layer.effect( yield* entry.evalLock.withPermits(1)( Effect.gen(function* () { const nodeID = evt.data.nodeID as string + // Same DB cross-check as the NodeFailed handler: projection + // is transactional with publish, so a row that no longer + // matches the event's terminal status means a later + // restart/replan reset it — drop the stale event without + // touching the new generation's fiber. + const expected = def === DagEvent.NodeSkipped ? "skipped" : "completed" + const node = yield* store.getNode(dagID, nodeID) + const confirmed = node?.status === expected // Cancel-skip race: workflow-level cancel publishes NodeSkipped // for running nodes, and this handler may win the cross-stream // race against WorkflowCancelled. Deleting the fiber here @@ -402,23 +410,25 @@ export const layer = Layer.effect( // prompt finishes or times out. Stop it now, mirroring the // NodeCancelled handler. Completed nodes keep the plain // delete — their fiber published the event and is finishing. - if (def === DagEvent.NodeSkipped) { + if (confirmed && def === DagEvent.NodeSkipped) { const fiber = entry.fibers.get(nodeID) if (fiber) { - const node = yield* store.getNode(dagID, nodeID) yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore) yield* Fiber.interrupt(fiber).pipe(Effect.ignore) } } - entry.fibers.delete(nodeID) + if (confirmed) entry.fibers.delete(nodeID) + if (!confirmed) { + yield* Effect.logDebug("DagLoop dropped stale node terminal event", { dagID, nodeID, expected, dbStatus: node?.status ?? "missing" }) + } const workflow = yield* store.getWorkflow(dagID) entry.runtime.setPaused(workflow?.status === "paused") entry.runtime.setStepMode(workflow?.status === "stepping") // Guard against stale events: a node already cancelled // (markUnsatisfied) or already satisfied must not be flipped // back. Mirrors the NodeFailed handler's isActive guard. - if (entry.runtime.isActive(evt.data.nodeID as string)) { - settle(entry, evt.data.nodeID as string) + if (confirmed && entry.runtime.isActive(nodeID)) { + settle(entry, nodeID) // In stepMode, do NOT auto-advance — wait for the next // explicit step command. checkCompletion still runs so // required-node failure / early completion is detected. @@ -473,19 +483,30 @@ export const layer = Layer.effect( yield* entry.evalLock.withPermits(1)( Effect.gen(function* () { const nid = evt.data.nodeID as string - const fiber = entry.fibers.get(nid) - entry.fibers.delete(nid) + // Generation arbitration via DB status: each projector runs + // INSIDE the durable publish transaction (core/dag/projector.ts), + // so by the time this handler consumes the event the row + // already reflects it. If the row is no longer "failed", a + // later NodeRestarted/replan reset the node — this event + // belongs to a previous generation and must not touch the + // new one (including the fiber map, which may already hold + // the new attempt's fiber). No generation field needed. + const node = yield* store.getNode(dagID, nid) // #3: only markUnsatisfied if the runtime still tracks this // node as non-terminal. A stale NodeFailed event (e.g. from // a replan-ceiling check after the node already completed) // would incorrectly flip a satisfied node to unsatisfied. - if (entry.runtime.isActive(nid)) { - const node = yield* store.getNode(dagID, nid) - yield* abortChild(nid, node?.childSessionId ?? null).pipe(Effect.ignore) + if (node?.status === "failed" && entry.runtime.isActive(nid)) { + const fiber = entry.fibers.get(nid) + entry.fibers.delete(nid) + yield* abortChild(nid, node.childSessionId ?? null).pipe(Effect.ignore) if (fiber) yield* Fiber.interrupt(fiber).pipe(Effect.ignore) entry.runtime.markUnsatisfied(nid) if (!entry.runtime.isStepMode()) yield* spawnReady(dagID) } + if (node?.status !== "failed") { + yield* Effect.logDebug("DagLoop dropped stale NodeFailed", { dagID, nodeID: nid, dbStatus: node?.status ?? "missing" }) + } // In stepMode, checkCompletion (which can trigger autonomous // fail/complete) still runs, but spawnReady is skipped — // stepping must NOT auto-advance after a node fails. @@ -568,6 +589,19 @@ export const layer = Layer.effect( if (wf) entry.config = parseWorkflowConfig(wf.config) const nodes = yield* store.getNodes(dagID) entry.runtime.rebuildGraph(toSchedulingNodes(nodes)) + // Replan resets restarted nodes to pending. Old fibers of nodes + // that are no longer running/queued must be interrupted here: + // nothing else will (there is no NodeRestarted subscriber), and + // a leftover fiber's timeout path publishes NodeFailed — a legal + // pending→failed transition guardNode cannot reject — poisoning + // the new generation. Mirrors the workflow-terminal sweep. + for (const [nodeID, fiber] of [...entry.fibers]) { + const node = nodes.find((n) => n.id === nodeID) + if (node && (node.status === "running" || node.status === "queued")) continue + yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore) + yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + entry.fibers.delete(nodeID) + } yield* spawnReady(dagID) yield* checkCompletion(dagID) }), @@ -700,13 +734,25 @@ export const layer = Layer.effect( for (const dagID of deliveredUnresponsiveDagIDs) { const entry = runtimes.get(dagID) if (!entry) continue - if (entry.runtime.isPaused() || entry.runtime.isStepMode()) continue - // Suppress the net only when current-process execution - // ownership proves that a running node is making progress. - if (entry.runtime.hasRunningMatching((id) => entry.fibers.has(id))) continue - if (entry.runtime.getReadyNodes().length > 0) continue - if (entry.runtime.isComplete()) continue - yield* dag.fail(dagID, "orchestrator_unresponsive").pipe(Effect.ignore) + // Torn-read fix: every runtime/fibers mutation happens as a + // pair inside this entry's evalLock (markRunning+fibers.set + // in spawnReady, settle+spawnReady in the terminal handlers), + // so reading the five conditions under the same lock waits + // out the markRunning→fibers.set window instead of misreading + // it as a stalled orchestrator. dag.fail stays outside the + // lock to avoid holding evalLock across the KeyedMutex. + const shouldFail = yield* entry.evalLock.withPermits(1)( + Effect.sync(() => + !entry.runtime.isPaused() + && !entry.runtime.isStepMode() + // Suppress the net only when current-process execution + // ownership proves that a running node is making progress. + && !entry.runtime.hasRunningMatching((id) => entry.fibers.has(id)) + && entry.runtime.getReadyNodes().length === 0 + && !entry.runtime.isComplete(), + ), + ) + if (shouldFail) yield* dag.fail(dagID, "orchestrator_unresponsive").pipe(Effect.ignore) } } return diff --git a/packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts b/packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts new file mode 100644 index 000000000..293cc7eec --- /dev/null +++ b/packages/opencode/test/dag/dag-replan-stale-nodefailed.test.ts @@ -0,0 +1,317 @@ +import { describe, expect, it } from "bun:test" +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" + +interface PromptGate { + readonly title: string + readonly release: Deferred.Deferred +} + +interface ParentPromptGate { + readonly release: Deferred.Deferred<"success" | "failure"> +} + +function takeWithin(queue: Queue.Queue, message: string) { + return Queue.take(queue).pipe( + Effect.timeoutOption("2 seconds"), + Effect.flatMap(Option.match({ + onNone: () => Effect.fail(new Error(message)), + onSome: Effect.succeed, + })), + ) +} + +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 node(id: string, dependsOn: string[] = [], timeoutMs?: number): NodeConfig { + return { + id, + name: id, + worker_type: "build", + depends_on: dependsOn, + required: true, + prompt_template: { inline: id }, + report_to_parent: true, + ...(timeoutMs ? { worker_config: { timeout_ms: timeoutMs } } : {}), + } +} + +function loopLayer(input: { + readonly childPrompts: Queue.Queue + readonly parentPrompts: 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 childTitles = new Map() + const created: string[] = [] + const session = Layer.mock(Session.Service, { + get: () => Effect.succeed({ id: "ses_parent", permission: [], agent: "build" } as never), + create: (value) => + Effect.sync(() => { + const id = `ses_child_${created.length + 1}` + created.push(id) + childTitles.set(id, (value?.title ?? id).replace(" (DAG node)", "")) + return { id } as never + }), + messages: () => Effect.succeed([]), + }) + const deliver = Effect.fn("test.SessionPrompt.deliver")(function* (value: SessionPrompt.PromptInput) { + const sessionID = value.sessionID as string + if (sessionID === "ses_parent") { + const release = yield* Deferred.make<"success" | "failure">() + yield* Queue.offer(input.parentPrompts, { release }) + const outcome = yield* Deferred.await(release) + if (outcome === "failure") return yield* Effect.die(new Error("provider unavailable")) + return reply(sessionID, "parent handled wake") + } + const release = yield* Deferred.make() + yield* Queue.offer(input.childPrompts, { + title: childTitles.get(sessionID) ?? sessionID, + release, + }) + return reply(sessionID, yield* Deferred.await(release)) + }) + const prompt = Layer.mock(SessionPrompt.Service, { + 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 runLoopTest( + test: (services: { + readonly dag: Dag.Interface + readonly store: DagStore.Interface + readonly childPrompts: Queue.Queue + readonly parentPrompts: Queue.Queue + }) => Effect.Effect, +) { + return Effect.gen(function* () { + const childPrompts = yield* Queue.unbounded() + const parentPrompts = 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() as never, + title: "Parent", + version: "test", + }).run().pipe(Effect.orDie) + yield* loop.init() + return yield* test({ dag, store, childPrompts, parentPrompts }) + }).pipe( + Effect.provide(loopLayer({ childPrompts, parentPrompts })), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: "project-1" }, + } as never), + Effect.scoped, + ) + }) +} + +describe("DagLoop replan vs stale NodeFailed", () => { + it("interrupts the replan-restarted node's old fiber so its timeout cannot poison the new generation", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Replan stale", + config: { name: "replan-stale", nodes: [node("a", [], 500)] }, + }) + const firstA = yield* takeWithin(childPrompts, "a did not start") + expect(firstA.title).toBe("a") + + // Pause so the restarted node is NOT immediately respawned by the + // WorkflowReplanned handler — this is exactly the window where only + // the replan fiber sweep stands between the old fiber's timeout and + // the new-generation pending row (pending→failed is a legal + // projection, so a stale NodeFailed would weld the node to failed). + yield* dag.pause(dagID) + yield* Effect.sleep("150 millis") + + const plan = yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 500), restart: true }] }) + expect(plan.restart).toEqual(["a"]) + expect((yield* store.getNode(dagID, "a"))?.status).toBe("pending") + + // Wait past the old attempt's deadline (admission + 500ms). Had the + // old fiber survived the replan, its timeout path would have + // published NodeFailed and flipped the pending row to failed. + yield* Effect.sleep("800 millis") + const nodeA = yield* store.getNode(dagID, "a") + expect(nodeA?.status).toBe("pending") + expect(nodeA?.errorReason).toBeNull() + + // The node stays schedulable in the new generation. + yield* dag.resume(dagID) + const secondA = yield* takeWithin(childPrompts, "a was not rescheduled after resume") + expect(secondA.title).toBe("a") + yield* Deferred.succeed(secondA.release, "done") + + yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((workflow) => workflow?.status === "completed" ? workflow : undefined), + ), + "workflow did not complete after the restarted node reran", + ) + const parent = yield* takeWithin(parentPrompts, "terminal wake did not reach the parent") + yield* Deferred.succeed(parent.release, "success") + }), + ), + ) + }) + + it("still settles a genuine failure: DB=failed drives markUnsatisfied and the required cascade", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Genuine failure", + config: { name: "genuine-failure", nodes: [node("a", [], 300), node("b", ["a"])] }, + }) + const gate = yield* takeWithin(childPrompts, "a did not start") + expect(gate.title).toBe("a") + // Never release — the node times out for real (DB row becomes + // failed), so the NodeFailed handler's DB cross-check confirms and + // the required-failure cascade must fail the workflow. + yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((workflow) => workflow?.status === "failed" ? workflow : undefined), + ), + "workflow did not fail after the required node timed out", + ) + const nodeA = yield* store.getNode(dagID, "a") + expect(nodeA?.status).toBe("failed") + expect(nodeA?.errorReason).toContain("timeout") + // b never ran: its only required dependency failed, and the + // workflow-fail terminalization skipped it. + expect((yield* store.getNode(dagID, "b"))?.status).toBe("skipped") + expect(Option.isNone(yield* Queue.poll(childPrompts))).toBe(true) + const parent = yield* takeWithin(parentPrompts, "failure wake did not reach the parent") + yield* Deferred.succeed(parent.release, "success") + }), + ), + ) + }) + + it("keeps a mid-flight replan restart schedulable when the node is immediately ready again", async () => { + await Effect.runPromise( + runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Restart ready", + config: { name: "restart-ready", nodes: [node("a", [], 500)] }, + }) + const firstA = yield* takeWithin(childPrompts, "a did not start") + expect(firstA.title).toBe("a") + + // Running workflow: the restarted node is ready again right away, so + // spawnReady replaces the fiber. The old attempt's deadline passing + // must not fail the new attempt (stale NodeFailed dropped by the DB + // status cross-check — the row is running, not failed). + const plan = yield* dag.replan(dagID, { nodes: [{ ...node("a", [], 2000), restart: true }] }) + expect(plan.restart).toEqual(["a"]) + + const secondA = yield* takeWithin(childPrompts, "a was not respawned after restart") + expect(secondA.title).toBe("a") + yield* Effect.sleep("700 millis") + expect((yield* store.getNode(dagID, "a"))?.status).toBe("running") + + yield* Deferred.succeed(secondA.release, "done") + yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((workflow) => workflow?.status === "completed" ? workflow : undefined), + ), + "workflow did not complete after the restarted node reran", + ) + const parent = yield* takeWithin(parentPrompts, "terminal wake did not reach the parent") + yield* Deferred.succeed(parent.release, "success") + }), + ), + ) + }) +}) diff --git a/packages/opencode/test/dag/dag-structured-output.test.ts b/packages/opencode/test/dag/dag-structured-output.test.ts index a97423127..5500a9b7c 100644 --- a/packages/opencode/test/dag/dag-structured-output.test.ts +++ b/packages/opencode/test/dag/dag-structured-output.test.ts @@ -209,6 +209,69 @@ describe("validateAgainstSchema", () => { expect(validateAgainstSchema(5.5, { type: "integer" }).ok).toBe(false) expect(validateAgainstSchema("5", { type: "integer" }).ok).toBe(false) }) + + it("rejects values outside enum and accepts members", () => { + const schema = { type: "string", enum: ["ACCEPT", "REJECT"] } + expect(validateAgainstSchema("ACCEPT", schema).ok).toBe(true) + expect(validateAgainstSchema("REJECT", schema).ok).toBe(true) + const bad = validateAgainstSchema("MAYBE", schema) + expect(bad.ok).toBe(false) + if (!bad.ok) { + expect(bad.error).toContain("ACCEPT") + expect(bad.error).toContain("MAYBE") + } + }) + + it("enum comparison is case-sensitive", () => { + expect(validateAgainstSchema("accept", { enum: ["ACCEPT"] }).ok).toBe(false) + }) + + it("validates mixed-type enum values", () => { + const schema = { enum: [1, "one", true, null] } + expect(validateAgainstSchema(1, schema).ok).toBe(true) + expect(validateAgainstSchema("one", schema).ok).toBe(true) + expect(validateAgainstSchema(true, schema).ok).toBe(true) + expect(validateAgainstSchema(null, schema).ok).toBe(true) + expect(validateAgainstSchema(2, schema).ok).toBe(false) + expect(validateAgainstSchema(false, schema).ok).toBe(false) + }) + + it("validates bare enum without type", () => { + expect(validateAgainstSchema("x", { enum: ["x", "y"] }).ok).toBe(true) + expect(validateAgainstSchema("z", { enum: ["x", "y"] }).ok).toBe(false) + }) + + it("validates const with structural deep equality (key order irrelevant)", () => { + expect(validateAgainstSchema("fixed", { const: "fixed" }).ok).toBe(true) + expect(validateAgainstSchema("other", { const: "fixed" }).ok).toBe(false) + const schema = { const: { a: 1, b: { c: [1, 2] } } } + expect(validateAgainstSchema({ b: { c: [1, 2] }, a: 1 }, schema).ok).toBe(true) + expect(validateAgainstSchema({ a: 1, b: { c: [2, 1] } }, schema).ok).toBe(false) + expect(validateAgainstSchema({ a: 1 }, schema).ok).toBe(false) + }) + + it("validates enum nested inside properties and items", () => { + const schema = { + type: "object", + properties: { + verdict: { type: "string", enum: ["ACCEPT", "REJECT"] }, + tags: { type: "array", items: { enum: ["a", "b"] } }, + }, + } + expect(validateAgainstSchema({ verdict: "ACCEPT", tags: ["a", "b"] }, schema).ok).toBe(true) + const badField = validateAgainstSchema({ verdict: "MAYBE" }, schema) + expect(badField.ok).toBe(false) + if (!badField.ok) expect(badField.error).toContain('field "verdict"') + const badItem = validateAgainstSchema({ tags: ["a", "c"] }, schema) + expect(badItem.ok).toBe(false) + if (!badItem.ok) expect(badItem.error).toContain("item[1]") + }) + + it("reports type mismatch before enum membership", () => { + const result = validateAgainstSchema(1, { type: "string", enum: ["a"] }) + expect(result.ok).toBe(false) + if (!result.ok) expect(result.error).toContain('expected type "string"') + }) }) describe("validatePayload", () => { @@ -250,6 +313,23 @@ describe("spawnNode submit_result capture", () => { expect(completed!.output).toEqual({ status: "ok" }) }) + it("(b2) enum-invalid verdict then valid retry → nodeCompleted with valid payload", async () => { + const { events, dagLayer } = makeEventTracker() + const schema = { + type: "object", + required: ["verdict"], + properties: { verdict: { type: "string", enum: ["ACCEPT", "REJECT"] } }, + } + await runSpawn( + dagLayer, + makePromptLayerWithCapture(reply("text"), [{ verdict: "MAYBE" }, { verdict: "ACCEPT" }], schema), + schema, + ) + const completed = events.find((e) => e.type === "nodeCompleted") + expect(completed).toBeDefined() + expect(completed!.output).toEqual({ verdict: "ACCEPT" }) + }) + it("(c) schema declared, no submit_result call → nodeFailed with verdict_fail", async () => { const { events, dagLayer } = makeEventTracker() const schema = { type: "object" } diff --git a/packages/opencode/test/dag/dag-wake-integration.test.ts b/packages/opencode/test/dag/dag-wake-integration.test.ts index 850bb68ce..7bf513acf 100644 --- a/packages/opencode/test/dag/dag-wake-integration.test.ts +++ b/packages/opencode/test/dag/dag-wake-integration.test.ts @@ -1042,4 +1042,39 @@ describe("DagLoop atomic wake integration", () => { ), ) }) + + it("does not misfire orchestrator_unresponsive while downstream work spawns after a completion", async () => { + await Effect.runPromise( + runWakeTest(({ dag, store, status, childPrompts, parentPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "No unresponsive misfire", + config: { name: "no-unresponsive-misfire", nodes: [node("a"), node("b", ["a"])] }, + }) + const a = yield* takeWithin(childPrompts, "a did not start") + yield* Deferred.succeed(a.release, "A done") + // Extra wake trigger racing b's spawn: the unresponsive check reads + // its five conditions under the entry's evalLock, so it sees either + // the pre-spawn ready set or the post-spawn fiber ownership — never + // the torn markRunning→fibers.set middle that used to read as a + // stalled orchestrator. + yield* status.set("ses_parent" as never, { type: "idle" }) + const b = yield* takeWithin(childPrompts, "b did not start after a completed") + yield* Effect.sleep("100 millis") + expect((yield* store.getWorkflow(dagID))?.status).toBe("running") + yield* Deferred.succeed(b.release, "B done") + yield* pollWithTimeout( + store.getWorkflow(dagID).pipe( + Effect.map((workflow) => workflow?.status === "completed" ? workflow : undefined), + ), + "workflow did not complete", + ) + const parent = yield* takeWithin(parentPrompts, "terminal wake did not reach the parent") + yield* Deferred.succeed(parent.release, "success") + }), + ), + ) + }) })