fix(dag): preserve coalesced summary updates

packages/opencode/src/dag/runtime/summary-publisher.ts

@@ -1,6 +1,6 @@
 export * as DagSummaryPublisher from "./summary-publisher"
 
-import { Effect, Layer, Scope, Context } from "effect"
+import { Cause, Effect, Exit, Layer, Scope, Context } from "effect"
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { InstanceState } from "@/effect/instance-state"
 import { EventV2Bridge } from "@/event-v2-bridge"
@@ -75,13 +75,46 @@ export const layer = Layer.effect(
         // removed entirely, correctness is unchanged — only more DagStore
         // reads would occur. This satisfies the "stateless derived view"
         // contract: no cached summary is ever served from this map.
-        const pending = new Set<string>()
+        const pending = new Map<string, boolean>()
         // Second coalescing tier keyed by workspace/dagID: node events don't
         // carry a sessionID, and resolving it eagerly meant one getWorkflow
         // query PER EVENT before the debounce window could absorb the burst
         // (P1-4). Resolve the sessionID once after the window, then hand off
         // to the workspace/session debounce.
-        const pendingByDag = new Set<string>()
+        const pendingByDag = new Map<string, boolean>()
+
+        const coalesceLatest = <E, R>(
+          active: Map<string, boolean>,
+          key: string,
+          body: () => Effect.Effect<void, E, R>,
+        ) =>
+          Effect.gen(function* () {
+            if (active.has(key)) {
+              active.set(key, true)
+              return
+            }
+            active.set(key, false)
+            yield* Effect.gen(function* () {
+              for (;;) {
+                yield* Effect.sleep("50 millis")
+                // Events during the debounce window are absorbed by the read
+                // that follows; only events racing the read require a rerun.
+                active.set(key, false)
+                const outcome = yield* body().pipe(Effect.exit)
+                const repeat = yield* Effect.sync(() => {
+                  if (active.get(key)) return true
+                  active.delete(key)
+                  return false
+                })
+                if (Exit.isFailure(outcome) && Cause.hasInterrupts(outcome.cause)) {
+                  return yield* Effect.failCause(outcome.cause)
+                }
+                if (repeat) continue
+                if (Exit.isFailure(outcome)) return yield* Effect.failCause(outcome.cause)
+                return
+              }
+            }).pipe(Effect.ensuring(Effect.sync(() => active.delete(key))))
+          })
 
         const publishForSession = (sessionID: string, workspace: string | undefined) =>
           Effect.gen(function* () {
@@ -98,36 +131,19 @@ export const layer = Layer.effect(
           })
 
         const schedulePublish = (sessionID: string, workspace: string | undefined) =>
-          Effect.gen(function* () {
-            const key = `${workspace ?? ""}\0${sessionID}`
-            // Coalesce: if a recompute is already scheduled for this route,
-            // let it absorb this trigger rather than queueing a second read.
-            // The coalesced early return MUST NOT touch `pending` — only the
-            // owning fiber clears its own slot, otherwise a coalesced caller
-            // would delete the owner's entry and reopen the window.
-            if (pending.has(key)) return
-            pending.add(key)
-            yield* Effect.gen(function* () {
-              yield* Effect.sleep("50 millis")
-              yield* publishForSession(sessionID, workspace)
-            }).pipe(Effect.ensuring(Effect.sync(() => pending.delete(key))))
-          })
+          coalesceLatest(pending, `${workspace ?? ""}\0${sessionID}`, () => publishForSession(sessionID, workspace))
 
         const schedulePublishByDag = (dagID: string, workspace: string | undefined) =>
-          Effect.gen(function* () {
-            const key = `${workspace ?? ""}\0${dagID}`
-            if (pendingByDag.has(key)) return
-            pendingByDag.add(key)
-            yield* Effect.gen(function* () {
-              yield* Effect.sleep("50 millis")
+          coalesceLatest(pendingByDag, `${workspace ?? ""}\0${dagID}`, () =>
+            Effect.gen(function* () {
               const wf = yield* store.getWorkflow(dagID)
               if (wf?.projectId !== ctx.project.id) return
               // Hand off to the session-level debounce (not publishForSession
               // directly) so a concurrent session-keyed window absorbs this
               // trigger instead of producing a duplicate read.
               yield* schedulePublish(wf.sessionId, workspace)
-            }).pipe(Effect.ensuring(Effect.sync(() => pendingByDag.delete(key))))
-          })
+            }),
+          )
 
         const unsubscribe = yield* events.listen((evt) => {
           if (!SUMMARY_TRIGGER_EVENTS.some((def) => def.type === evt.type)) return Effect.void

packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect } from "bun:test"
-import { DateTime, Effect, Layer } from "effect"
+import { DateTime, Deferred, Effect, Layer } from "effect"
 import { DagStore, type WorkflowRow, type WorkflowSummary } from "@opencode-ai/core/dag/store"
 import { DagEvent } from "@opencode-ai/schema/dag-event"
 import { EventV2Bridge } from "@/event-v2-bridge"
@@ -18,6 +18,11 @@ interface SummaryEmission {
 
 interface StoreControl {
   failures: number
+  failuresAfterGate: number
+  readGate?: {
+    started: Deferred.Deferred<void>
+    release: Deferred.Deferred<void>
+  }
   reads: Map<string, number>
   lookups: Map<string, number>
   projects: Map<string, string>
@@ -29,15 +34,16 @@ interface EventControl {
   listener?: (event: never) => Effect.Effect<void>
 }
 
-function control() {
+function control(): StoreControl {
   return {
     failures: 0,
+    failuresAfterGate: 0,
     reads: new Map<string, number>(),
     lookups: new Map<string, number>(),
     projects: new Map<string, string>(),
     sessions: new Map<string, string>(),
     summaries: new Map<string, WorkflowSummary[]>(),
-  } satisfies StoreControl
+  }
 }
 
 function workflow(id: string, sessionId: string, projectId: string): WorkflowRow {
@@ -81,13 +87,24 @@ function runtime(state: StoreControl, bus: EventControl) {
         return sid ? workflow(dagID, sid, state.projects.get(dagID) ?? "global") : undefined
       }),
     getWorkflowSummaries: (sessionID) =>
-      Effect.sync(() => {
+      Effect.gen(function* () {
         state.reads.set(sessionID, (state.reads.get(sessionID) ?? 0) + 1)
         if (state.failures > 0) {
           state.failures -= 1
           throw new Error("simulated summary read failure")
         }
-        return state.summaries.get(sessionID) ?? []
+        const summaries = state.summaries.get(sessionID) ?? []
+        const gate = state.readGate
+        state.readGate = undefined
+        if (gate) {
+          yield* Deferred.succeed(gate.started, undefined)
+          yield* Deferred.await(gate.release)
+        }
+        if (state.failuresAfterGate > 0) {
+          state.failuresAfterGate -= 1
+          throw new Error("simulated summary read failure after gate")
+        }
+        return summaries
       }),
   })
   const events = Layer.mock(EventV2Bridge.Service, {
@@ -232,6 +249,117 @@ describe("DagSummaryPublisher behavior", () => {
     ).pipe(Effect.provide(runtime(state, bus)))
   })
 
+  it.instance("an event arriving during an in-flight read schedules a fresh recompute", () => {
+    const state = control()
+    const bus = {} satisfies EventControl
+    state.sessions.set("dag-inflight", "ses-inflight")
+    state.summaries.set("ses-inflight", [summary("dag-inflight", 1)])
+
+    return withCollector((collector) =>
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        state.readGate = { started, release }
+
+        yield* (yield* DagSummaryPublisher.Service).init()
+        yield* publishNodeEvents(bus, "dag-inflight", 1)
+        yield* Deferred.await(started).pipe(Effect.timeout("1 second"))
+
+        state.summaries.set("ses-inflight", [summary("dag-inflight", 2)])
+        yield* publishNodeEvents(bus, "dag-inflight", 1)
+        yield* Effect.sleep("20 millis")
+        yield* Deferred.succeed(release, undefined)
+
+        yield* pollWithTimeout(
+          Effect.sync(() => (collector.emissions.at(-1)?.summaries[0]?.completedNodes === 2 ? true : undefined)),
+          () =>
+            `event coalesced during an in-flight read was lost (lookups=${state.lookups.get("dag-inflight") ?? 0}, reads=${state.reads.get("ses-inflight") ?? 0}, emissions=${collector.emissions.length})`,
+          "500 millis",
+        )
+
+        expect(state.reads.get("ses-inflight")).toBe(2)
+        expect(collector.emissions.at(-1)?.summaries).toEqual([summary("dag-inflight", 2)])
+      }),
+    ).pipe(Effect.provide(runtime(state, bus)))
+  })
+
+  it.instance("another DAG event arriving during a shared session read schedules a fresh recompute", () => {
+    const state = control()
+    const bus = {} satisfies EventControl
+    state.sessions.set("dag-session-a", "ses-shared")
+    state.sessions.set("dag-session-b", "ses-shared")
+    state.summaries.set("ses-shared", [summary("dag-session-a", 1)])
+
+    return withCollector((collector) =>
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        state.readGate = { started, release }
+
+        yield* (yield* DagSummaryPublisher.Service).init()
+        yield* publishNodeEvents(bus, "dag-session-a", 1)
+        yield* Deferred.await(started).pipe(Effect.timeout("1 second"))
+
+        state.summaries.set("ses-shared", [summary("dag-session-a", 2)])
+        yield* publishNodeEvents(bus, "dag-session-b", 1)
+        yield* Effect.sleep("80 millis")
+        yield* Deferred.succeed(release, undefined)
+
+        yield* pollWithTimeout(
+          Effect.sync(() => (collector.emissions.at(-1)?.summaries[0]?.completedNodes === 2 ? true : undefined)),
+          () =>
+            `shared session event was lost (reads=${state.reads.get("ses-shared") ?? 0}, emissions=${collector.emissions.length})`,
+          "500 millis",
+        )
+
+        expect(state.reads.get("ses-shared")).toBe(2)
+        expect(state.lookups).toEqual(
+          new Map([
+            ["dag-session-a", 1],
+            ["dag-session-b", 1],
+          ]),
+        )
+      }),
+    ).pipe(Effect.provide(runtime(state, bus)))
+  })
+
+  it.instance("an event arriving during a failed in-flight read schedules a fresh recompute", () => {
+    const state = control()
+    const bus = {} satisfies EventControl
+    state.failuresAfterGate = 1
+    state.sessions.set("dag-failed-inflight", "ses-failed-inflight")
+    state.summaries.set("ses-failed-inflight", [summary("dag-failed-inflight", 1)])
+
+    return withCollector((collector) =>
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>()
+        const release = yield* Deferred.make<void>()
+        state.readGate = { started, release }
+
+        yield* (yield* DagSummaryPublisher.Service).init()
+        yield* publishNodeEvents(bus, "dag-failed-inflight", 1)
+        yield* Deferred.await(started).pipe(Effect.timeout("1 second"))
+
+        state.summaries.set("ses-failed-inflight", [summary("dag-failed-inflight", 2)])
+        yield* publishNodeEvents(bus, "dag-failed-inflight", 1)
+        yield* Effect.sleep("20 millis")
+        yield* Deferred.succeed(release, undefined)
+
+        yield* pollWithTimeout(
+          Effect.sync(() => (collector.emissions.at(-1)?.summaries[0]?.completedNodes === 2 ? true : undefined)),
+          () =>
+            `event coalesced during a failed in-flight read was lost (reads=${state.reads.get("ses-failed-inflight") ?? 0}, emissions=${collector.emissions.length})`,
+          "500 millis",
+        )
+
+        expect(state.reads.get("ses-failed-inflight")).toBe(2)
+        expect(collector.emissions).toEqual([
+          { sessionID: "ses-failed-inflight", summaries: [summary("dag-failed-inflight", 2)] },
+        ])
+      }),
+    ).pipe(Effect.provide(runtime(state, bus)))
+  })
+
   it.instance("different sessions coalesce independently", () => {
     const state = control()
     const bus = {} satisfies EventControl

packages/opencode/test/dag/dag-summary-publisher.test.ts

@@ -54,11 +54,11 @@ describe("DagSummaryPublisher contract (stateless derived view)", () => {
       path.resolve("src/dag/runtime/summary-publisher.ts"),
       "utf-8",
     )
-    // No module-level mutable Map/Set. The `pending` Set lives inside
+    // No module-level mutable Map/Set. The `pending` Map lives inside
     // the InstanceState closure, not at module level.
     expect(src).not.toMatch(/^const\s+\w+\s*=\s*new\s+(Map|Set)\b/m)
     expect(src).not.toMatch(/^let\s+\w+\s*=\s*new\s+(Map|Set)\b/m)
-    // The pending Set is declared inside the InstanceState.make closure.
-    expect(src).toMatch(/const pending = new Set<string>/)
+    // The pending Map is declared inside the InstanceState.make closure.
+    expect(src).toMatch(/const pending = new Map<string, boolean>/)
   })
 })