diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index a725cdd563..938d27604a 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -74,16 +74,15 @@ const serviceLayer = Layer.effect( const recovering = new Set() const wakeInFlight = new Set() const wakePending = new Set() - // GOAL-FP-01-14: per-session record of the last wake summary whose - // transcript part was written. The durable mark runs AFTER the write - // (at-least-once delivery: a mark failure keeps the batch unreported - // for a retry), so a retry of an already-written summary would - // re-inject the same digest into the transcript. The retry dedupes on - // this map and only re-marks. In-process only — a crash between write - // and mark still duplicates on the restart sweep (a durable - // delivering-marker would need a schema change; registered, see - // GOAL-FP-01-14). Capped: evicting entries degrades to the pre-fix - // duplicate visibility, never to a lost wake. + // Per-session record of the last wake summary whose transcript part was + // written. Admit success IS the delivery (issue #321): the durable + // wake_reported mark lands immediately after the write, not after the + // wake-driven turn completes, so the in-memory map only needs to dedupe + // the narrow mark-retry path — if the leased mark returns None the rows + // stay unreported and a later trigger re-marks WITHOUT re-prompting. + // In-process only; restart durability now comes from wake_reported being + // persisted at admit time, not from this map. Capped: evicting entries + // degrades to a redundant re-prompt, never to a lost wake. const deliveredWakeSummaries = new Map() // Seed the commented global dag.jsonc once per instance init — the @@ -1315,24 +1314,30 @@ const serviceLayer = Layer.effect( ) if (!wakeLease) return - // Persist wake_reported AFTER successful delivery only. - // A failure stays durable for a later idle event or restart scan; - // it must not spin synchronously on the same row. // The part is marked synthetic: model-visible (the orchestrator // receives the node result and can act) but NOT rendered as a user // message in the TUI chat — DAG data surfaces via the sidebar panel // and Inspector, keeping the chat conversation clean. // - // GOAL-FP-01-14: the transcript part is written BEFORE the - // durable mark. A mark failure (or a crash between the two) - // leaves the batch unreported and the retry would re-inject the - // SAME summary. When this session already had this exact summary - // written, skip the prompt and only re-mark — the write is - // idempotent in effect because an identical digest adds no - // information. A differing summary (new results committed - // between attempts) always prompts. + // ADMIT SUCCESS == DELIVERED (issue #321). The previous contract + // (GOAL-FP-01-14) persisted wake_reported only AFTER the whole + // wake-driven parent turn completed. A restart or mid-turn + // interruption therefore left wake_reported=false while the + // synthetic part was already durable in transcript, so the startup + // sweep re-injected a byte-identical wake (real incident: the same + // 4391-char wake injected twice, ~10 min apart, after a TUI + // restart). Redelivery adds duplicates, never information. The + // mark (and the terminal-workflow unregisters) now land right + // after admitIfIdle admits the part, BEFORE awaiting the turn. + // + // The in-memory dedup map still guards the retry path: if the + // leased mark below returns None (generation/owner changed), the + // rows stay unreported and a later trigger re-marks without + // re-prompting. A differing summary (new results committed between + // attempts) always prompts. if (deliveredWakeSummaries.size > 1024) deliveredWakeSummaries.clear() const didDeliver = yield* Effect.gen(function* () { + let wakeTurn: Effect.Effect | undefined if (deliveredWakeSummaries.get(sessionID) !== summary) { const delivered = yield* SessionPrompt.admitIfIdle(promptSvc, automation, wakeLease, { sessionID: SessionID.make(sessionID), @@ -1340,28 +1345,63 @@ const serviceLayer = Layer.effect( }) if (Option.isNone(delivered)) return false deliveredWakeSummaries.set(sessionID, summary) - yield* delivered.value.pipe( - Effect.onError(() => - Effect.sync(() => { - if (deliveredWakeSummaries.get(sessionID) === summary) { - deliveredWakeSummaries.delete(sessionID) - } - }), - ), - ) + wakeTurn = delivered.value } - const markLease = yield* automation.claim(SessionID.make(sessionID), { kind: "dag" }) - if (Option.isNone(markLease)) return false - const marked = yield* automation.use(markLease.value, store.markWakeBatchReported(batch)) - if (Option.isNone(marked)) return false - plan.unresponsiveDagIDs.forEach((workflowID) => deliveredUnresponsiveDagIDs.add(workflowID)) - yield* Effect.forEach( - batch.workflows.filter((workflow) => isWorkflowTerminalStatus(workflow.status as never)), - (workflow) => automation.unregister(SessionID.make(sessionID), { kind: "dag", id: workflow.id }), - { discard: true }, + // Admit success == delivered (issue #321): persist wake_reported + // at admit time. A leased mark returning None (generation/owner + // raced) is treated as retry-later — the rows stay unreported and + // a later trigger re-marks — but the admitted turn still runs + // below (its end-of-turn idle is what re-arms that retry). + const markLease = Option.getOrUndefined( + yield* automation.claim(SessionID.make(sessionID), { kind: "dag" }), ) - return true + // Any mark failure (lease lost, generation raced, or the store + // write dying) degrades to retry-later instead of propagating: + // the rows stay unreported and a later trigger re-marks, while + // the admitted turn below still runs (its end-of-turn idle is + // what re-arms that retry). Only interruption propagates. + const markSucceeded = markLease + ? Option.isSome( + yield* automation.use(markLease, store.markWakeBatchReported(batch)).pipe( + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.failCause(cause) + : Effect.logWarning("DAG wake batch mark failed; rows stay unreported for retry", { + sessionID, + cause: Cause.pretty(cause), + }).pipe(Effect.as(Option.none())), + ), + ), + ) + : false + if (markSucceeded) { + plan.unresponsiveDagIDs.forEach((workflowID) => deliveredUnresponsiveDagIDs.add(workflowID)) + yield* Effect.forEach( + batch.workflows.filter((workflow) => isWorkflowTerminalStatus(workflow.status as never)), + (workflow) => automation.unregister(SessionID.make(sessionID), { kind: "dag", id: workflow.id }), + { discard: true }, + ) + } + + // Pacing — keep one wake turn at a time. The turn runs AFTER the + // durable mark, so its failure can no longer lose the report; + // swallow non-interrupt failures instead of surfacing them as a + // delivery failure. It also runs when the mark raced, so its + // end-of-turn idle re-arms the mark retry. + if (wakeTurn) { + yield* wakeTurn.pipe( + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.failCause(cause) + : Effect.logInfo("DAG wake turn did not complete; wake already reported", { + sessionID, + cause: Cause.pretty(cause), + }), + ), + ) + } + return markSucceeded }).pipe( Effect.catchCause((cause) => Effect.logWarning("DAG wake delivery failed", { sessionID, cause: Cause.pretty(cause) }).pipe( diff --git a/packages/opencode/src/session/automation-lease.ts b/packages/opencode/src/session/automation-lease.ts index 1613608c39..ac848c2cb6 100644 --- a/packages/opencode/src/session/automation-lease.ts +++ b/packages/opencode/src/session/automation-lease.ts @@ -101,10 +101,13 @@ export const layer = Layer.effect( // GOAL-FP-01-02: when the dag ownership actually disappears (owner // transitions dag → goal/none), re-trigger the goal evaluation through // the EXISTING idle status event mechanism so a goal that yielded to the - // dag on the last idle event gets a fresh evaluation. The final dag - // unregister of a wake delivery (U2 in dag/runtime/loop.ts) lands AFTER - // the wake turn's idle event — without this re-trigger the goal silently - // stalls until the next external idle. This is also the GOAL-FP-01-11 + // dag on the last idle event gets a fresh evaluation. Since issue #321 + // the final dag unregister of a wake delivery lands MID-TURN at admit + // time (the mark moves with it), so the session is busy when the owner + // flips: the idle gate below skips the re-emit here, and the in-flight + // wake turn re-emits idle on completion, which re-drives the goal. + // Without this re-trigger path the goal silently stalls until the next + // external idle. This is also the GOAL-FP-01-11 // mitigation surface: a claim that lost the ownership race gets another // chance once the owner actually transfers. // diff --git a/packages/opencode/test/dag/dag-timeout-escalation.test.ts b/packages/opencode/test/dag/dag-timeout-escalation.test.ts index f8c88f46a8..0167ab318d 100644 --- a/packages/opencode/test/dag/dag-timeout-escalation.test.ts +++ b/packages/opencode/test/dag/dag-timeout-escalation.test.ts @@ -195,6 +195,7 @@ function runLoopTest( test: (services: { readonly dag: Dag.Interface readonly store: DagStore.Interface + readonly status: SessionStatus.Interface readonly childPrompts: Queue.Queue readonly parentPrompts: Queue.Queue readonly getCancelCount: () => number @@ -212,6 +213,7 @@ function runLoopTest( const dag = yield* Dag.Service const loop = yield* DagLoop.Service const store = yield* DagStore.Service + const status = yield* SessionStatus.Service const database = yield* Database.Service yield* database.db.insert(ProjectTable).values({ id: Project.ID.make("project-1"), @@ -230,6 +232,7 @@ function runLoopTest( return yield* test({ dag, store, + status, childPrompts, parentPrompts, getCancelCount: harness.getCancelCount, @@ -1044,7 +1047,7 @@ describe("DagLoop timeout escalation", () => { // P5 escalationPending ∧ wakeReported → proceed [recovery — test below + L880/L567] it("blocks re-time while the escalation wake is undelivered (Q2: escalationPending ∧ ¬wakeReported ⇒ skip)", async () => { await Effect.runPromise( - runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + runLoopTest(({ dag, store, status, childPrompts, parentPrompts }) => Effect.gen(function* () { const dagID = yield* dag.create({ projectID: "project-1", @@ -1054,10 +1057,16 @@ describe("DagLoop timeout escalation", () => { }) yield* takeWithin(childPrompts, "a did not start") - // Acceptance #2: the deadline-driven INITIAL escalation (watchdog → - // nodeTimeoutEscalated → first wake) is NOT touched by the gate — - // the gate only governs the replan re-time path. It fires and its - // wake reaches the parent. + // issue #321: wake_reported now lands at ADMIT time, so an in-flight + // turn no longer holds wake_reported=false (the wake is reported as + // soon as it is admitted). The only window that keeps the escalation + // wake UNDELIVERED now is a BUSY parent session — the idle-gate never + // admits it. Hold the wake there. + yield* status.set(SessionID.make("ses_parent"), { type: "busy" }) + + // Initial escalation fires; while the parent is busy its wake is held + // UNDELIVERED (never admitted). The node sits at the public-path state + // [escalationPending ∧ ¬wakeReported ∧ deadline≤now]. const escalated = yield* pollWithTimeout( store.getNode(dagID, "a").pipe( Effect.map((current) => current?.timeoutExtensions === 1 ? current : undefined), @@ -1066,17 +1075,10 @@ describe("DagLoop timeout escalation", () => { ) expect(escalated.status).toBe("running") expect(escalated.escalationPending).toBe(true) + expect(escalated.wakeReported).toBe(false) const baselineDeadline = escalated.deadlineMs - const timeoutWake = yield* takeWithin(parentPrompts, "initial escalation wake did not reach the parent") - expect(timeoutWake.text).toContain("[DAG Node Timeout]") - - // Hold the wake UNDELIVERED: the harness blocks delivery on the - // release Deferred, and the loop persists wake_reported=true only - // AFTER successful delivery (loop.ts:1125). The node sits at the - // public-path state [escalationPending ∧ ¬wakeReported ∧ deadline≤now]. - const undelivered = yield* store.getNode(dagID, "a") - expect(undelivered?.escalationPending).toBe(true) - expect(undelivered?.wakeReported).toBe(false) + // The wake is held by the busy gate — nothing reached the parent yet. + expect(Option.isNone(yield* Queue.poll(parentPrompts))).toBe(true) // Main agent replans with a NEW timeout. Q2 must SKIP the re-time: // adjudication cannot land before the escalation wake was delivered. @@ -1106,7 +1108,11 @@ describe("DagLoop timeout escalation", () => { expect(reEscalated.wakeReported).toBe(false) expect(reEscalated.deadlineMs).toBe(baselineDeadline) - // Release the held wake so the loop marks delivery before teardown. + // Release the held wake: the parent goes idle, the wake is admitted + // (wake_reported lands at admit, issue #321), and the turn settles. + yield* status.set(SessionID.make("ses_parent"), { type: "idle" }) + const timeoutWake = yield* takeWithin(parentPrompts, "escalation wake did not reach the parent once idle") + expect(timeoutWake.text).toContain("[DAG Node Timeout]") yield* Deferred.succeed(timeoutWake.release, "success") }), ), @@ -1179,7 +1185,7 @@ describe("DagLoop timeout escalation", () => { // terminal-cleanup path. it("C1: nodeExtendTimeout distinguishes Q2 rejection (-2) from terminal rejection (0) — three-valued contract", async () => { await Effect.runPromise( - runLoopTest(({ dag, store, childPrompts, parentPrompts }) => + runLoopTest(({ dag, store, status, childPrompts, parentPrompts }) => Effect.gen(function* () { const dagID = yield* dag.create({ projectID: "project-1", @@ -1189,12 +1195,15 @@ describe("DagLoop timeout escalation", () => { }) const gate = yield* takeWithin(childPrompts, "a did not start") - // Escalation fires; the wake is held UNDELIVERED so the node sits at - // the Q2 state [escalationPending ∧ ¬wakeReported ∧ running]. + // issue #321: wake_reported lands at ADMIT time, so the Q2 state + // [escalationPending ∧ ¬wakeReported] can no longer be held by an + // in-flight turn — hold the escalation wake UNDELIVERED via the busy + // idle-gate instead, which keeps it never admitted. + yield* status.set(SessionID.make("ses_parent"), { type: "busy" }) const escalated = yield* pollWithTimeout( store.getNode(dagID, "a").pipe( Effect.map((current) => - current?.timeoutExtensions === 1 && current.escalationPending && !current.wakeReported + current && current.status === "running" && current.escalationPending && !current.wakeReported ? current : undefined, ), @@ -1213,7 +1222,9 @@ describe("DagLoop timeout escalation", () => { expect(afterQ2?.status).toBe("running") expect(afterQ2?.deadlineMs).toBe(frozenDeadline) - // Deliver the wake and let the child finish — the node terminalizes. + // Release the held wake (idle admits it) and let the child finish — + // the node terminalizes. + yield* status.set(SessionID.make("ses_parent"), { type: "idle" }) const wake = yield* takeWithin(parentPrompts, "escalation wake did not reach the parent") yield* Deferred.succeed(wake.release, "success") yield* Deferred.succeed(gate.release, "done") diff --git a/packages/opencode/test/dag/dag-wake-integration.test.ts b/packages/opencode/test/dag/dag-wake-integration.test.ts index 76039be7da..560e48e457 100644 --- a/packages/opencode/test/dag/dag-wake-integration.test.ts +++ b/packages/opencode/test/dag/dag-wake-integration.test.ts @@ -9,8 +9,12 @@ import { DagProjector } from "@opencode-ai/core/dag/projector" import { WorkflowNodeTable, WorkflowTable } from "@opencode-ai/core/dag/sql" import { DagStore } from "@opencode-ai/core/dag/store" import { EventV2 } from "@opencode-ai/core/event" +import { Project } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionTable } from "@opencode-ai/core/session/sql" +import { Model } from "@opencode-ai/schema/model" +import { Provider } from "@opencode-ai/schema/provider" import { Agent } from "@/agent/agent" import { fingerprintBrief } from "@/dag/admission" import { Dag, type NodeConfig } from "@/dag/dag" @@ -1028,7 +1032,11 @@ describe("DagLoop atomic wake integration", () => { ) }) - it("leaves the whole batch unreported when parent delivery fails", async () => { + // FLIPPED for issue #321: previously a failed parent turn left the whole + // batch unreported (for later redelivery). Admit success now IS the + // delivery — the mark lands at admit time, so a failed/interrupted turn + // still leaves the batch reported. + it("reports the wake batch at admit time even when the parent turn then fails (issue #321)", async () => { await Effect.runPromise( runWakeTest(({ dag, store, childPrompts, parentPrompts, parentSettled }) => Effect.gen(function* () { @@ -1044,14 +1052,20 @@ describe("DagLoop atomic wake integration", () => { yield* Deferred.succeed(parent.release, "failure") yield* takeWithin(parentSettled, "failed parent prompt did not settle") - expect(yield* store.getUnreportedWakeNodes("ses_parent")).toHaveLength(1) - expect(yield* store.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(1) + // The synthetic part is durable in transcript either way; marking at + // admit time means a restart or mid-turn interruption has nothing to + // re-inject. + expect(yield* store.getUnreportedWakeNodes("ses_parent")).toHaveLength(0) + expect(yield* store.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(0) }), ), ) }) - it("retries the parent prompt after a provider failure instead of silently marking the wake", async () => { + // FLIPPED for issue #321: previously a failed parent turn was retried — a + // fresh idle event re-injected the SAME wake. Admit success is now the + // delivery, so a later trigger must inject NO duplicate prompt. + it("does not redeliver a wake whose parent turn failed (issue #321)", async () => { await Effect.runPromise( runWakeTest(({ dag, store, status, childPrompts, parentPrompts, parentSettled }) => Effect.gen(function* () { @@ -1066,14 +1080,16 @@ describe("DagLoop atomic wake integration", () => { const first = yield* takeWithin(parentPrompts, "retryable batch did not wake the parent") yield* Deferred.succeed(first.release, "failure") yield* takeWithin(parentSettled, "failed parent prompt did not settle") - yield* status.set(SessionID.make("ses_parent"), { type: "idle" }) - const second = yield* takeWithin(parentPrompts, "failed provider wake was not prompted again") - expect(promptText(second.input)).toContain('Node "retryable-node" completed: retry me') - yield* Deferred.succeed(second.release, "success") - yield* takeWithin(parentSettled, "successful retry did not settle") + // The batch is reported at admit time even though the turn failed. expect(yield* store.getUnreportedWakeNodes("ses_parent")).toHaveLength(0) expect(yield* store.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(0) + + // Re-trigger the delivery path: the idle gate must NOT inject a + // duplicate prompt (pre-fix this re-delivered the identical wake). + yield* status.set(SessionID.make("ses_parent"), { type: "idle" }) + yield* Effect.sleep("500 millis") + expect(Option.isNone(yield* Queue.poll(parentPrompts))).toBe(true) }), ), ) @@ -1123,6 +1139,190 @@ describe("DagLoop atomic wake integration", () => { ) }) + // NEW for issue #321: simulates the production restart. A wake is admitted + // but its parent turn NEVER finishes (the process dies mid-turn). Pre-fix the + // mark only landed after the turn completed, so the restart sweep saw + // wake_reported=false and re-injected the byte-identical wake. Post-fix the + // mark lands at admit time, so a fresh loop's startup sweep delivers nothing. + it("does not redeliver an already-admitted wake across a restart (issue #321)", async () => { + await Effect.runPromise( + Effect.gen(function* () { + const childTitles = new Map() + const created: string[] = [] + const session = Layer.mock(Session.Service, { + get: () => + Effect.succeed({ + id: SessionID.make("ses_parent"), + slug: "parent", + projectID: Project.ID.make("project-1"), + directory: process.cwd(), + title: "Parent", + version: "test", + time: { created: 0, updated: 0 }, + permission: [], + agent: "build", + }), + 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: SessionID.make(id), + slug: "child", + projectID: Project.ID.make("project-1"), + directory: process.cwd(), + title: value?.title ?? id, + version: "test", + time: { created: 0, updated: 0 }, + } + }), + messages: () => Effect.succeed([]), + }) + const agent = Layer.mock(Agent.Service, { + get: () => + Effect.succeed({ + name: "build", + mode: "all", + permission: [], + options: {}, + description: "", + prompt: "", + model: { providerID: Provider.ID.make("test"), modelID: Model.ID.make("test-model") }, + tools: {}, + hooks: {}, + }), + }) + const deliver = (queues: { + readonly childPrompts: Queue.Queue + readonly parentPrompts: Queue.Queue + readonly parentSettled: Queue.Queue + }) => + 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(queues.parentPrompts, { input: value, release }) + const outcome = yield* Deferred.await(release).pipe( + Effect.ensuring(Queue.offer(queues.parentSettled, undefined)), + ) + 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(queues.childPrompts, { + title: childTitles.get(sessionID) ?? sessionID, + input: value, + release, + }) + return reply(sessionID, yield* Deferred.await(release)) + }) + const promptLayer = (queues: { + readonly childPrompts: Queue.Queue + readonly parentPrompts: Queue.Queue + readonly parentSettled: Queue.Queue + }) => Layer.mock(SessionPrompt.Service, withIdleAdmission({ + cancel: () => Effect.void, + prompt: deliver(queues), + promptIfIdle: (value) => deliver(queues)(value).pipe(Effect.map(Option.some)), + })) + + 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) + + yield* Effect.gen(function* () { + const storeSvc = yield* DagStore.Service + const databaseSvc = yield* Database.Service + yield* databaseSvc.db.insert(ProjectTable).values({ + id: Project.ID.make("project-1"), + worktree: AbsolutePath.make(process.cwd()), + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* databaseSvc.db.insert(SessionTable).values({ + id: SessionID.make("ses_parent"), + project_id: Project.ID.make("project-1"), + slug: "parent", + directory: AbsolutePath.make(process.cwd()), + title: "Parent", + version: "test", + }).run().pipe(Effect.orDie) + + // Phase 1: admit the wake, then "die" before the parent turn finishes. + const q1 = { + childPrompts: yield* Queue.unbounded(), + parentPrompts: yield* Queue.unbounded(), + parentSettled: yield* Queue.unbounded(), + } + yield* Effect.scoped( + Effect.gen(function* () { + const dagSvc = yield* Dag.Service + const loopSvc = yield* DagLoop.Service + yield* loopSvc.init() + yield* dagSvc.create({ + projectID: "project-1", + sessionID: "ses_parent", + title: "Restart wake", + config: { name: "restart-wake", nodes: [node("restart-node")] }, + }) + const child = yield* takeWithin(q1.childPrompts, "restart node did not start") + yield* Deferred.succeed(child.release, "done") + // The wake was admitted (mark landed at admit). Do NOT release the + // parent turn — disposing the scope simulates a restart mid-turn. + yield* takeWithin(q1.parentPrompts, "terminal workflow did not wake the parent") + }).pipe(Effect.provide(DagLoop.layer.pipe( + Layer.provide(session), + Layer.provide(promptLayer(q1)), + Layer.provide(agent), + ))), + ) + + // The durable rows are already reported even though the turn never ran. + expect(yield* storeSvc.getUnreportedWakeNodes("ses_parent")).toHaveLength(0) + expect(yield* storeSvc.getUnreportedWakeWorkflows("ses_parent")).toHaveLength(0) + expect(yield* storeSvc.getSessionsWithUnreportedWakes()).toHaveLength(0) + + // Phase 2: a fresh loop over the SAME store runs the startup sweep. + const q2 = { + childPrompts: yield* Queue.unbounded(), + parentPrompts: yield* Queue.unbounded(), + parentSettled: yield* Queue.unbounded(), + } + yield* Effect.scoped( + Effect.gen(function* () { + const loopSvc = yield* DagLoop.Service + yield* loopSvc.init() + // Bound the window in which a (now-forbidden) redelivery could appear. + yield* Effect.sleep("500 millis") + expect(Option.isNone(yield* Queue.poll(q2.parentPrompts))).toBe(true) + }).pipe(Effect.provide(DagLoop.layer.pipe( + Layer.provide(session), + Layer.provide(promptLayer(q2)), + Layer.provide(agent), + ))), + ) + }).pipe( + Effect.provide(base), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { + id: Project.ID.make("project-1"), + worktree: process.cwd(), + time: { created: 0, updated: 0 }, + sandboxes: [], + }, + }), + ) + }).pipe(Effect.scoped), + ) + }) + it("keeps a wake unreported while the parent is busy and delivers it on idle", async () => { await Effect.runPromise( runWakeTest(({ dag, store, status, childPrompts, parentPrompts }) =>