diff --git a/packages/opencode/src/dag/runtime/summary-publisher.ts b/packages/opencode/src/dag/runtime/summary-publisher.ts index e3487172c7..091d3d7672 100644 --- a/packages/opencode/src/dag/runtime/summary-publisher.ts +++ b/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() + const pending = new Map() // 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() + const pendingByDag = new Map() + + const coalesceLatest = ( + active: Map, + key: string, + body: () => Effect.Effect, + ) => + 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 diff --git a/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts b/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts index b26d0af2e3..e585761945 100644 --- a/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts +++ b/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 + release: Deferred.Deferred + } reads: Map lookups: Map projects: Map @@ -29,15 +34,16 @@ interface EventControl { listener?: (event: never) => Effect.Effect } -function control() { +function control(): StoreControl { return { failures: 0, + failuresAfterGate: 0, reads: new Map(), lookups: new Map(), projects: new Map(), sessions: new Map(), summaries: new Map(), - } 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() + const release = yield* Deferred.make() + 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() + const release = yield* Deferred.make() + 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() + const release = yield* Deferred.make() + 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 diff --git a/packages/opencode/test/dag/dag-summary-publisher.test.ts b/packages/opencode/test/dag/dag-summary-publisher.test.ts index 101d3ccc34..b84e74638e 100644 --- a/packages/opencode/test/dag/dag-summary-publisher.test.ts +++ b/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/) + // The pending Map is declared inside the InstanceState.make closure. + expect(src).toMatch(/const pending = new Map/) }) })