diff --git a/src/node/orpc/router.ts b/src/node/orpc/router.ts index 6faeaee27f..ebbc2f51dc 100644 --- a/src/node/orpc/router.ts +++ b/src/node/orpc/router.ts @@ -1400,8 +1400,8 @@ export const router = (authToken?: string) => { }) ), // Subscription, not a unary call: the handler returns an event iterator, - // which handlerGen cannot produce. Streams stay Promise/AsyncGenerator - // until an Effect Stream bridge exists (later migration phase). + // which handlerGen cannot produce. Rides the Effect Stream bridge + // (streamBridge.ts) via subscribeMemoryChanges. onChange: t .input(schemas.memory.onChange.input) .output(schemas.memory.onChange.output) diff --git a/src/node/orpc/routerSubscriptions.ts b/src/node/orpc/routerSubscriptions.ts index 6bb39d0524..9c3b6ba5a0 100644 --- a/src/node/orpc/routerSubscriptions.ts +++ b/src/node/orpc/routerSubscriptions.ts @@ -13,16 +13,10 @@ import type { import type { SshPromptEvent, SshPromptRequest } from "@/common/orpc/schemas/ssh"; import type { TimelineSubscriptionEvent } from "@/common/orpc/schemas/timeline"; import type { DevToolsEvent } from "@/common/types/devtools"; -import { - asyncIterableFromSubscription, - createAsyncEventQueue, - createLatestValueQueue, -} from "@/common/utils/asyncEventIterator"; -import { createAsyncMessageQueue } from "@/common/utils/asyncMessageQueue"; import { createCoalescedReader } from "@/common/utils/coalescedReader"; import { getErrorMessage } from "@/common/utils/errors"; -import { withQueueHeartbeat } from "@/common/utils/withQueueHeartbeat"; import type { ORPCContext } from "./context"; +import { subscriptionIterable } from "./streamBridge"; import { createReplayBufferedStreamMessageRelay } from "@/node/services/replayBufferedStreamMessageRelay"; import { TIMELINE_DEFAULT_PAGE_LIMIT } from "@/node/services/timelineService"; import type { LogEntry } from "@/node/services/logBuffer"; @@ -90,10 +84,10 @@ export function subscribeConfigChanges( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - queue: createLatestValueQueue(), - subscribe: (push) => context.config.onConfigChanged(() => push(undefined)), + buffer: "latest", + subscribe: (emit) => context.config.onConfigChanged(() => emit.push(undefined)), }); } @@ -103,12 +97,12 @@ export function subscribeDevTools( signal?: AbortSignal ): AsyncGenerator { const service = context.devToolsService; - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => { + subscribe: (emit) => { const eventName = "update:" + workspaceId; - service.on(eventName, push); - return () => service.off(eventName, push); + service.on(eventName, emit.push); + return () => service.off(eventName, emit.push); }, initial: async () => ({ type: "snapshot" as const, runs: await service.getRuns(workspaceId) }), }); @@ -118,10 +112,10 @@ export function subscribeProviderConfig( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - queue: createLatestValueQueue(), - subscribe: (push) => context.providerService.onConfigChanged(() => push(undefined)), + buffer: "latest", + subscribe: (emit) => context.providerService.onConfigChanged(() => emit.push(undefined)), }); } @@ -129,13 +123,18 @@ export function subscribePolicyChanges( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - queue: createLatestValueQueue(), - subscribe: (push) => context.policyService.onPolicyChanged(() => push(undefined)), + buffer: "latest", + subscribe: (emit) => context.policyService.onPolicyChanged(() => emit.push(undefined)), }); } +/** + * Deliberately NOT on the Effect Stream bridge: this is a pure timed + * generator with no event source to attach and no resource to release, so the + * bridge's acquireRelease lifecycle would add machinery without value. + */ export function createTickIterable( count: number, intervalMs: number @@ -153,17 +152,17 @@ export function subscribeLogs( signal?: AbortSignal ): AsyncGenerator { let snapshot: ReturnType["snapshot"]; - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => { + subscribe: (emit) => { const subscription = subscribeLogFeed((event) => { if (event.type === "append") { if (shouldIncludeLogEntry(event.entry.level, minLevel)) { - push({ type: "append", epoch: event.epoch, entries: [event.entry] }); + emit.push({ type: "append", epoch: event.epoch, entries: [event.entry] }); } return; } - push({ type: "reset", epoch: event.epoch }); + emit.push({ type: "reset", epoch: event.epoch }); }, minLevel); snapshot = subscription.snapshot; return subscription.unsubscribe; @@ -186,15 +185,16 @@ export function subscribeMemoryChanges( validate?.(); const metadata = workspaceId ? await context.workspaceService.getInfo(workspaceId) : null; const projectPath = metadata ? resolveMemoryProjectIdentity(metadata) : null; - yield* asyncIterableFromSubscription({ + yield* subscriptionIterable({ signal, - subscribe: (push) => { + subscribe: (emit) => { const onChange = (event: MemoryChangeEvent) => { if (event.scope === "workspace" && event.workspaceId !== workspaceId) return; if (event.scope === "project" && event.projectPath !== projectPath) return; - push(event); + emit.push(event); }; - const onStatusChange = (event: MemoryConsolidationStatusChangeEventPayload) => push(event); + const onStatusChange = (event: MemoryConsolidationStatusChangeEventPayload) => + emit.push(event); context.memoryService.on("change", onChange); context.memoryConsolidationService.on("statusChange", onStatusChange); return () => { @@ -214,10 +214,10 @@ export function subscribeTimeline( const pendingEvents: TimelineSubscriptionEvent["events"] = []; let snapshotSequence: number | undefined; let pushEvent: ((event: TimelineSubscriptionEvent) => void) | undefined; - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => { - pushEvent = push; + subscribe: (emit) => { + pushEvent = emit.push; const onAppended = (event: { workspaceId: string; events: TimelineSubscriptionEvent["events"]; @@ -230,7 +230,7 @@ export function subscribeTimeline( return; } const events = event.events.filter((item) => item.seq > sequence); - if (events.length > 0) push({ type: "appended", events }); + if (events.length > 0) emit.push({ type: "appended", events }); }; context.timelineService.on("appended", onAppended); return () => context.timelineService.off("appended", onAppended); @@ -264,20 +264,17 @@ export function subscribeWorkspaceChat( if (typeof input.legacyAutoRetryEnabled === "boolean") { session.setLegacyAutoRetryEnabledHint(input.legacyAutoRetryEnabled); } - const queue = withQueueHeartbeat(createAsyncMessageQueue(), { - type: "heartbeat" as const, - }); let replayRelay: ReturnType; // Subscribe before replay so the relay can buffer overlapping live deltas. - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - queue, - subscribe: (push) => { - replayRelay = createReplayBufferedStreamMessageRelay(push); + heartbeat: { value: { type: "heartbeat" as const } }, + subscribe: (emit) => { + replayRelay = createReplayBufferedStreamMessageRelay(emit.push); return session.onChatEvent(({ message }) => replayRelay.handleSessionMessage(message)); }, - initialize: async (push) => { - await session.replayHistory(({ message }) => push(message), input.mode); + initialize: async (emit) => { + await session.replayHistory(({ message }) => emit.push(message), input.mode); replayRelay.finishReplay(); session.scheduleStartupRecovery(); }, @@ -288,11 +285,11 @@ export function subscribeMetadata( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => { - context.workspaceService.on("metadata", push); - return () => context.workspaceService.off("metadata", push); + subscribe: (emit) => { + context.workspaceService.on("metadata", emit.push); + return () => context.workspaceService.off("metadata", emit.push); }, }); } @@ -301,17 +298,14 @@ export function subscribeWorkspaceActivity( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - const queue = withQueueHeartbeat(createAsyncEventQueue(), { - type: "heartbeat", - }); - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - queue, - subscribe: (push) => { + heartbeat: { value: { type: "heartbeat" } }, + subscribe: (emit) => { const onActivity = (event: { workspaceId: string; activity: WorkspaceActivitySnapshot | null; - }) => push({ type: "activity", ...event }); + }) => emit.push({ type: "activity", ...event }); context.workspaceService.on("activity", onActivity); return () => context.workspaceService.off("activity", onActivity); }, @@ -328,39 +322,41 @@ export function subscribeBackgroundBashes( processes: await service.listBackgroundProcesses(workspaceId), foregroundToolCallIds: service.getForegroundToolCallIds(workspaceId), }); - // Full snapshots coalesce because replaying stale intermediate state only grows memory. - const queue = createLatestValueQueue>>(); const bootstrap = { delivered: false, error: null as Error | null }; - const reader = createCoalescedReader({ - read: async () => { - try { - queue.push(await getState()); - bootstrap.delivered = true; - } catch (error) { - if (!bootstrap.delivered) { - bootstrap.error = error instanceof Error ? error : new Error(getErrorMessage(error)); - queue.end(); - return; - } - throw error; - } - }, - retryDelayMs: 1_000, - }); - return asyncIterableFromSubscription({ + let reader: ReturnType | undefined; + // Full snapshots coalesce ("latest") because replaying stale intermediate + // state only grows memory. + return subscriptionIterable>>({ signal, - queue, - subscribe: () => { + buffer: "latest", + subscribe: (emit) => { + reader = createCoalescedReader({ + read: async () => { + try { + emit.push(await getState()); + bootstrap.delivered = true; + } catch (error) { + if (!bootstrap.delivered) { + bootstrap.error = error instanceof Error ? error : new Error(getErrorMessage(error)); + emit.end(); + return; + } + throw error; + } + }, + retryDelayMs: 1_000, + }); const onChange = (changedWorkspaceId: string) => { - if (changedWorkspaceId === workspaceId) reader.trigger(); + if (changedWorkspaceId === workspaceId) reader?.trigger(); }; service.onBackgroundBashChange(onChange); return () => { - reader.stop(); + reader?.stop(); service.offBackgroundBashChange(onChange); }; }, - initialize: () => reader.trigger(), + // subscribe runs before initialize, so the reader is always set here. + initialize: () => reader?.trigger(), onEnd: () => { if (bootstrap.error != null) throw bootstrap.error; }, @@ -372,13 +368,14 @@ export function subscribeWorkspaceStats( workspaceId: string, signal?: AbortSignal ): AsyncGenerator { - const queue = createLatestValueQueue(); const throttleMs = 100; let lastPushedAtMs = 0; let inFlight = true; let pendingTimer: ReturnType | undefined; let pendingSnapshot = false; let closed = false; + // Assigned in subscribe, which runs before any onChange event or initialize. + let push: (snapshot: WorkspaceStatsSnapshot) => void = () => undefined; // Snapshot reads are serialized and throttled so token deltas cannot build a backlog. const pushSnapshot = async () => { if (closed || inFlight || !pendingSnapshot) return; @@ -388,7 +385,7 @@ export function subscribeWorkspaceStats( const snapshot = await context.sessionTimingService.getSnapshot(workspaceId); if (closed) return; lastPushedAtMs = snapshot.generatedAt; - queue.push(snapshot); + push(snapshot); } finally { inFlight = false; if (!closed && pendingSnapshot) scheduleSnapshot(); @@ -411,10 +408,11 @@ export function subscribeWorkspaceStats( }, remaining); pendingTimer.unref?.(); }; - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - queue, - subscribe: () => { + buffer: "latest", + subscribe: (emit) => { + push = emit.push; const onChange = (changedWorkspaceId: string) => { if (changedWorkspaceId === workspaceId) scheduleSnapshot(); }; @@ -427,11 +425,11 @@ export function subscribeWorkspaceStats( context.sessionTimingService.removeSubscriber(workspaceId); }; }, - initialize: async () => { + initialize: async (emit) => { try { const initial = await context.sessionTimingService.getSnapshot(workspaceId); lastPushedAtMs = initial.generatedAt; - queue.push(initial); + emit.push(initial); } finally { inFlight = false; if (!closed && pendingSnapshot) scheduleSnapshot(); @@ -445,9 +443,9 @@ export function subscribeTerminalOutput( sessionId: string, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => context.terminalService.onOutput(sessionId, push), + subscribe: (emit) => context.terminalService.onOutput(sessionId, emit.push), }); } @@ -457,10 +455,10 @@ export function attachTerminal( signal?: AbortSignal ): AsyncGenerator { // Output subscribes before screen capture so attach cannot lose bytes in the handshake. - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => - context.terminalService.onOutput(sessionId, (data) => push({ type: "output", data })), + subscribe: (emit) => + context.terminalService.onOutput(sessionId, (data) => emit.push({ type: "output", data })), initial: () => ({ type: "screenState" as const, data: context.terminalService.getScreenState(sessionId), @@ -473,9 +471,9 @@ export function subscribeTerminalExit( sessionId: string, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => context.terminalService.onExit(sessionId, push), + subscribe: (emit) => context.terminalService.onExit(sessionId, emit.push), take: 1, }); } @@ -484,15 +482,12 @@ export function subscribeTerminalActivity( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - const queue = withQueueHeartbeat(createAsyncEventQueue(), { - type: "heartbeat", - }); - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - queue, - subscribe: (push) => + heartbeat: { value: { type: "heartbeat" } }, + subscribe: (emit) => context.terminalService.onActivityChange((workspaceId) => - push({ + emit.push({ type: "update", workspaceId, activity: context.terminalService.getWorkspaceActivity(workspaceId), @@ -509,9 +504,9 @@ export function subscribeUpdateStatus( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => context.updateService.onStatus(push), + subscribe: (emit) => context.updateService.onStatus(emit.push), }); } @@ -519,9 +514,9 @@ export function subscribeOpenSettings( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => context.menuEventService.onOpenSettings(() => push(undefined)), + subscribe: (emit) => context.menuEventService.onOpenSettings(() => emit.push(undefined)), }); } @@ -529,16 +524,16 @@ export function subscribeSshPrompts( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return asyncIterableFromSubscription({ + return subscriptionIterable({ signal, - subscribe: (push) => { + subscribe: (emit) => { const releaseResponder = context.sshPromptService.registerInteractiveResponder(); // The service returns snapshot plus listeners atomically, preventing a request gap. const { snapshot, unsubscribe } = context.sshPromptService.subscribeRequests( - (request: SshPromptRequest) => push({ type: "request", ...request }), - (requestId: string) => push({ type: "removed", requestId }) + (request: SshPromptRequest) => emit.push({ type: "request", ...request }), + (requestId: string) => emit.push({ type: "removed", requestId }) ); - for (const request of snapshot) push({ type: "request", ...request }); + for (const request of snapshot) emit.push({ type: "request", ...request }); return () => { releaseResponder(); unsubscribe(); diff --git a/src/node/orpc/streamBridge.test.ts b/src/node/orpc/streamBridge.test.ts new file mode 100644 index 0000000000..b9d18eb1f2 --- /dev/null +++ b/src/node/orpc/streamBridge.test.ts @@ -0,0 +1,259 @@ +/** + * Behavioral tests for the Effect Stream subscription bridge (streamBridge.ts). + * + * These pin the lifecycle invariants the bridge introduces — guaranteed + * listener teardown on every exit path (disconnect, consumer break, stream + * error, mid-initialize abort) — plus the ordering/coalescing wire semantics + * the oRPC subscription handlers rely on. Teardown-on-error was previously + * unpinned; the remaining tests assert behavior (ordering, completion, + * error propagation), not implementation literals. + */ +import { describe, expect, test } from "bun:test"; +import { EventEmitter } from "node:events"; +import { subscriptionIterable, type SubscriptionEmit } from "./streamBridge"; + +async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise { + const start = Date.now(); + while (!predicate()) { + if (Date.now() - start > timeoutMs) throw new Error("waitFor timed out"); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +/** Collect up to `count` values, resolving early if the iterator completes. */ +async function collect(iterable: AsyncGenerator, count: number): Promise { + const values: T[] = []; + for await (const value of iterable) { + values.push(value); + if (values.length >= count) break; + } + return values; +} + +describe("subscriptionIterable teardown", () => { + test("listener count returns to baseline after client abort", async () => { + const emitter = new EventEmitter(); + const controller = new AbortController(); + const iterable = subscriptionIterable({ + signal: controller.signal, + subscribe: (emit) => { + emitter.on("value", emit.push); + return () => emitter.off("value", emit.push); + }, + }); + + const consumed = (async () => { + const values: number[] = []; + for await (const value of iterable) values.push(value); + return values; + })(); + + await waitFor(() => emitter.listenerCount("value") === 1); + emitter.emit("value", 1); + controller.abort(); + + // Abort completes the generator normally (no throw) and detaches. + await consumed; + expect(emitter.listenerCount("value")).toBe(0); + }); + + test("consumer break (generator return) detaches the listener", async () => { + const emitter = new EventEmitter(); + const iterable = subscriptionIterable({ + subscribe: (emit) => { + emitter.on("value", emit.push); + return () => emitter.off("value", emit.push); + }, + }); + + const first = (async () => { + for await (const value of iterable) return value; + throw new Error("iterator ended without a value"); + })(); + await waitFor(() => emitter.listenerCount("value") === 1); + emitter.emit("value", 42); + expect(await first).toBe(42); + await waitFor(() => emitter.listenerCount("value") === 0); + }); + + test("initialize failure detaches the listener and surfaces the error", async () => { + const emitter = new EventEmitter(); + const boom = new Error("bootstrap failed"); + const iterable = subscriptionIterable({ + subscribe: (emit) => { + emitter.on("value", emit.push); + return () => emitter.off("value", emit.push); + }, + initialize: () => Promise.reject(boom), + }); + + try { + await iterable.next(); + expect.unreachable("initialize failure must reject the subscription"); + } catch (error) { + expect(error).toBe(boom); + } + expect(emitter.listenerCount("value")).toBe(0); + }); + + test("abort during a hung initialize still detaches the listener", async () => { + const emitter = new EventEmitter(); + const controller = new AbortController(); + const iterable = subscriptionIterable({ + signal: controller.signal, + subscribe: (emit) => { + emitter.on("value", emit.push); + return () => emitter.off("value", emit.push); + }, + initialize: () => new Promise(() => undefined), + }); + + const consumed = collect(iterable, 1); + await waitFor(() => emitter.listenerCount("value") === 1); + controller.abort(); + expect(await consumed).toEqual([]); + expect(emitter.listenerCount("value")).toBe(0); + }); + + test("take completes the stream and detaches immediately", async () => { + const emitter = new EventEmitter(); + const iterable = subscriptionIterable({ + subscribe: (emit) => { + emitter.on("exit", emit.push); + return () => emitter.off("exit", emit.push); + }, + take: 1, + }); + + const consumed = collect(iterable, 2); + await waitFor(() => emitter.listenerCount("exit") === 1); + emitter.emit("exit", 7); + expect(await consumed).toEqual([7]); + expect(emitter.listenerCount("exit")).toBe(0); + }); + + test("pre-aborted signal never attaches the listener", async () => { + const emitter = new EventEmitter(); + const controller = new AbortController(); + controller.abort(); + let subscribed = false; + const iterable = subscriptionIterable({ + signal: controller.signal, + subscribe: (emit) => { + subscribed = true; + emitter.on("value", emit.push); + return () => emitter.off("value", emit.push); + }, + }); + expect(await collect(iterable, 1)).toEqual([]); + expect(subscribed).toBe(false); + }); +}); + +describe("subscriptionIterable ordering and buffering", () => { + test("synchronous burst from a foreign callsite is delivered in emit order", async () => { + let emitHandle: SubscriptionEmit | undefined; + const iterable = subscriptionIterable({ + subscribe: (emit) => { + emitHandle = emit; + return () => undefined; + }, + }); + + const consumed = collect(iterable, 3); + await waitFor(() => emitHandle !== undefined); + // Producer emits synchronously (EventEmitter-style): the values must land + // in the buffer before push returns, preserving emit-after-write order. + emitHandle?.push(1); + emitHandle?.push(2); + emitHandle?.push(3); + expect(await consumed).toEqual([1, 2, 3]); + }); + + test("latest buffer coalesces values while the consumer is slow", async () => { + let emitHandle: SubscriptionEmit | undefined; + const iterable = subscriptionIterable({ + buffer: "latest", + subscribe: (emit) => { + emitHandle = emit; + return () => undefined; + }, + }); + + // Force attach without consuming further, then burst before the next read. + const first = iterable.next(); + await waitFor(() => emitHandle !== undefined); + emitHandle?.push(1); + expect((await first).value).toBe(1); + emitHandle?.push(2); + emitHandle?.push(3); + emitHandle?.push(4); + // An unconsumed snapshot is replaced, never queued: only the newest survives. + expect((await iterable.next()).value).toBe(4); + await iterable.return(undefined); + }); + + test("initial value is delivered before events buffered while it was computed", async () => { + let emitHandle: SubscriptionEmit | undefined; + const iterable = subscriptionIterable({ + subscribe: (emit) => { + emitHandle = emit; + return () => undefined; + }, + initial: async () => { + // Event fires between attach and snapshot completion — it must not be + // lost, and it must arrive after the snapshot. + emitHandle?.push("during-initial"); + await new Promise((resolve) => setTimeout(resolve, 1)); + return "snapshot"; + }, + }); + + expect(await collect(iterable, 2)).toEqual(["snapshot", "during-initial"]); + }); + + test("emit.end drains buffered values, then onEnd error surfaces", async () => { + const iterable = subscriptionIterable({ + subscribe: (emit) => { + emit.push(1); + emit.push(2); + emit.end(); + return () => undefined; + }, + onEnd: () => { + throw new Error("bootstrap error"); + }, + }); + + const values: number[] = []; + try { + for await (const value of iterable) values.push(value); + expect.unreachable("onEnd error must reject the subscription"); + } catch (error) { + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toBe("bootstrap error"); + } + expect(values).toEqual([1, 2]); + }); + + test("heartbeat values are injected while the subscription is idle", async () => { + const iterable = subscriptionIterable({ + heartbeat: { value: "heartbeat", intervalMs: 10 }, + subscribe: () => () => undefined, + }); + expect(await collect(iterable, 2)).toEqual(["heartbeat", "heartbeat"]); + }); + + test("nothing runs until the consumer starts pulling", async () => { + let subscribed = false; + const iterable = subscriptionIterable({ + subscribe: () => { + subscribed = true; + return () => undefined; + }, + }); + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(subscribed).toBe(false); + await iterable.return(undefined); + }); +}); diff --git a/src/node/orpc/streamBridge.ts b/src/node/orpc/streamBridge.ts new file mode 100644 index 0000000000..3e9ef35909 --- /dev/null +++ b/src/node/orpc/streamBridge.ts @@ -0,0 +1,198 @@ +/** + * Effect Stream bridge for oRPC subscription procedures (Effect migration + * Phase 9). + * + * Adapts the EventEmitter-style subscription seam (attach listener → push + * events → detach) that `routerSubscriptions.ts` handlers use into an Effect + * Stream pipeline, then re-exposes it as the `AsyncGenerator` wire shape oRPC + * event-iterator procedures expect. Wire behavior (payload shapes, ordering, + * completion/error semantics) is intentionally identical to the previous + * `asyncIterableFromSubscription` seam. + * + * Design invariants: + * + * - **Foreign-callsite emissions**: producers fire from non-Effect contexts + * (EventEmitter callbacks, AI-SDK callbacks). `emit.push` is a synchronous + * `Queue.offerUnsafe` — the value lands in the buffer before `push` returns, + * with no fiber suspension between the producer's emit and the queue offer. + * This preserves emit-after-durable-write ordering guarantees (e.g. memory + * consolidation emits `statusChange` synchronously after its durable write; + * subscribers must observe events in exactly that order). + * - **Guaranteed teardown**: listener detach runs via `Effect.acquireRelease` + * (one acquireRelease per resource), tied to the stream's scope. The scope + * closes on client disconnect (AbortSignal → iterator close → fiber + * interruption), on consumer `return()`, on stream failure, and on natural + * completion — so `off()`/`removeListener()` is guaranteed on every exit + * path. + * - **Interruption posture**: subscription streams are interruptible by + * design. Client aborts interrupt the pull fiber at the next suspension + * point; there is no in-flight mutation to protect, only listener handles, + * which the scope finalizers release. + * - **Defect folding**: bridge-internal failures (validate/subscribe throws, + * initialize/initial rejections, onEnd errors) fail the pull, which + * surfaces as a rejection of the consumer's `next()` — the same observable + * contract as the previous async-generator seam. Nothing escapes as an + * unhandled rejection. + * - **Laziness**: nothing (not even `validate`) runs until the consumer's + * first `next()` call, matching async-generator semantics. + */ +import type { Cause } from "effect"; +import { Effect, Queue, Stream } from "effect"; +import { SUBSCRIPTION_HEARTBEAT_INTERVAL_MS } from "@/common/utils/withQueueHeartbeat"; + +/** Producer-facing handle. Safe to call from any non-Effect callsite. */ +export interface SubscriptionEmit { + /** + * Enqueue a value synchronously. No-op after the subscription ended. + * Never throws and never suspends. + */ + push: (value: T) => void; + /** + * Gracefully complete the subscription: already-buffered values are still + * delivered, then the stream ends (and `onEnd` runs, if provided). + */ + end: () => void; +} + +export interface SubscriptionStreamOptions { + signal?: AbortSignal; + /** Runs first; a throw rejects the subscription before any resource is acquired. */ + validate?: () => void; + /** + * Attach the underlying listener(s); returns the detach thunk. Attach and + * detach are wrapped in `Effect.acquireRelease`, so detach is guaranteed on + * disconnect, error, interruption, and natural completion. Values pushed + * synchronously during attach are buffered and delivered after `initial`. + */ + subscribe: (emit: SubscriptionEmit) => () => void; + /** + * Buffering strategy. `"all"` (default) is an unbounded FIFO. `"latest"` + * coalesces: an unconsumed value is replaced by the newest one, so a slow + * consumer never accumulates a backlog and never replays stale snapshots + * (mirrors `createLatestValueQueue`). + */ + buffer?: "all" | "latest"; + /** + * Inject `value` into the queue every `intervalMs` (default + * SUBSCRIPTION_HEARTBEAT_INTERVAL_MS) while the subscription is live. + * The ticker starts after `initialize` completes — heartbeats cannot + * interleave into a history replay — and stops with the stream's scope. + */ + heartbeat?: { value: T; intervalMs?: number }; + /** Runs after attach, before any value is delivered. Pushes are buffered. */ + initialize?: (emit: SubscriptionEmit) => void | Promise; + /** + * Produce a value delivered before any buffered events (evaluated after + * attach + `initialize`, so subscriptions cannot lose events that fire + * while the initial snapshot is computed). + */ + initial?: () => T | Promise; + /** Complete after delivering this many values (counting `initial`). */ + take?: number; + /** + * Runs when the queue completes gracefully via `emit.end`. A throw fails + * the subscription (used to surface bootstrap errors to the client). + */ + onEnd?: () => void | Promise; +} + +/** + * Build the scoped Effect Stream for a subscription. The returned stream owns + * the listener lifecycle: acquisition happens when the stream starts and every + * finalizer runs when the stream ends, fails, or is interrupted. + */ +function subscriptionStream(options: SubscriptionStreamOptions): Stream.Stream { + return Stream.unwrap( + Effect.gen(function* () { + options.validate?.(); + + const queue = yield* options.buffer === "latest" + ? Queue.sliding(1) + : Queue.unbounded(); + const emit: SubscriptionEmit = { + push: (value) => void Queue.offerUnsafe(queue, value), + end: () => void Queue.endUnsafe(queue), + }; + + // Close the queue when the scope closes so late producer pushes become + // no-ops (offerUnsafe returns false on a non-open queue). + yield* Effect.acquireRelease(Effect.void, () => Effect.sync(() => Queue.endUnsafe(queue))); + + // Per-resource acquireRelease: listener detach is guaranteed on every + // exit path (disconnect, error, interruption, completion). + yield* Effect.acquireRelease( + Effect.sync(() => options.subscribe(emit)), + (unsubscribe) => Effect.sync(unsubscribe) + ); + + if (options.initialize) { + const initialize = options.initialize; + // Async thunk so synchronous throws follow the same rejection path. + yield* Effect.promise(async () => initialize(emit)); + } + + let head: Stream.Stream = Stream.empty; + if (options.initial) { + const initial = options.initial; + const value = yield* Effect.promise(async () => initial()); + head = Stream.make(value); + } + + if (options.heartbeat) { + const heartbeat = options.heartbeat; + // Scope-tied ticker fiber: interrupted with the stream. Started after + // `initialize` so heartbeats never interleave into replayed history. + yield* Effect.forkScoped( + Effect.forever( + Effect.flatMap( + Effect.sleep(heartbeat.intervalMs ?? SUBSCRIPTION_HEARTBEAT_INTERVAL_MS), + () => Queue.offer(queue, heartbeat.value) + ) + ) + ); + } + + const onEnd = options.onEnd; + let stream: Stream.Stream = Stream.concat(head, Stream.fromQueue(queue)); + if (onEnd) { + stream = Stream.concat(stream, Stream.fromEffectDrain(Effect.promise(async () => onEnd()))); + } + if (options.take != null) { + stream = Stream.take(stream, options.take); + } + return stream; + }) + ); +} + +/** + * Adapt a subscription to the AsyncGenerator wire shape oRPC expects, backed + * by the Effect Stream above. + * + * Abort handling mirrors the previous seam: when `signal` aborts, the stream + * iterator is closed, which interrupts the pull fiber and closes the scope + * (running all release finalizers); the generator then completes normally. + */ +export function subscriptionIterable(options: SubscriptionStreamOptions): AsyncGenerator { + return (async function* () { + if (options.signal?.aborted) return; + + const iterator = Stream.toAsyncIterable(subscriptionStream(options))[Symbol.asyncIterator](); + // `return()` memoizes its close promise, so the extra call in `finally` + // awaits the same teardown instead of re-running it. + const onAbort = () => void iterator.return?.(); + options.signal?.addEventListener("abort", onAbort, { once: true }); + if (options.signal?.aborted) onAbort(); + + try { + while (true) { + const result = await iterator.next(); + if (result.done) return; + yield result.value; + } + } finally { + options.signal?.removeEventListener("abort", onAbort); + await iterator.return?.(); + } + })(); +}