🤖 refactor: bridge oRPC subscription procedures through Effect Stream

[ThomasK33]

Summary

Phase 9 of the progressive Effect migration: adds a single reusable Effect Stream bridge (src/node/orpc/streamBridge.ts) for oRPC subscription procedures and converts all 20 event-subscription handlers in routerSubscriptions.ts to it — including the two deferred by name from earlier phases (memory.onChange from #4025 and config onConfigChanged from #4036). Wire behavior (payload shapes, ordering, completion/error semantics) is unchanged; all existing tests pass unmodified.

Background

Wave 2 (#4033, #4034, #4035, #4036, #4038) put every service-backed unary mutation/query on handlerGen/Effect pipelines. What remained on the router was the subscription surface: ~20 procedures consuming EventEmitter-based sources via on/off + the asyncIterableFromSubscription push seam. That seam hand-rolls lifecycle (abort listeners, queue end, unsubscribe-in-finally) per call. This phase replaces it on the router with a Scope-managed Effect Stream pipeline so listener teardown becomes structural instead of conventional.

Bridge design (streamBridge.ts)

subscriptionIterable(options) builds a scoped Stream and adapts it back to the AsyncGenerator wire shape oRPC event-iterator procedures expect:

• Buffering: an Effect Queue (Queue.unbounded for FIFO, Queue.sliding(1) for latest-value coalescing — exact replacement for createLatestValueQueue snapshot semantics).
• 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 (e.g. memory consolidation emits statusChange synchronously post-write; subscribers observe events in exactly that order).
• Guaranteed teardown: listener attach/detach is wrapped in Effect.acquireRelease (one acquireRelease per resource), tied to the stream's scope. The scope closes on client disconnect (AbortSignal → iterator close → fiber interruption), consumer return(), stream failure, and natural completion — off()/removeListener() runs on every exit path.
• Completion: emit.end() maps to Queue.endUnsafe, which drains buffered values before signalling Done (same drain-then-complete contract the old queues had); onEnd then runs and may fail the stream (bootstrap-error surfacing for subscribeBackgroundBashes).
• Heartbeats: a scope-tied forked fiber offers the heartbeat value on an interval, started after initialize so heartbeats never interleave into history replay (replaces withQueueHeartbeat on the router).
• Ordering of initial: evaluated after attach, delivered before any buffered events — events firing while a snapshot is computed are neither lost nor reordered ahead of it.
• Interruption posture: subscription streams are interruptible by design. Aborts interrupt the pull fiber at its next suspension point; there is no in-flight mutation to protect, only listener handles, which scope finalizers release. (Deterministic-winner funnels from 🤖 refactor: convert memoryConsolidationService and workspaceStatusGenerator internals to Effect; make in-flight run-lock reservation deterministic #4038 are unaffected; none live on this surface.)
• Defect folding: validate/subscribe throws and initialize/initial/onEnd rejections fail the pull and surface as a rejection of the consumer's next() — the same observable contract as the old seam; nothing escapes as an unhandled rejection.
• Laziness: nothing (not even validate) runs until the consumer's first next(), matching async-generator semantics.

The wire adapter (Stream.toAsyncIterable + an abort-aware generator wrapper) closes the stream iterator on abort, which interrupts a pending pull immediately — slightly prompter teardown than the old seam (which detached only when the generator resumed), and wire-invisible.

Conversion table

Subscription	Bridge features used	Notes
subscribeConfigChanges	buffer: "latest"	deferred by name from #4036
subscribeMemoryChanges (memory.onChange)	default FIFO; outer wrapper keeps validate + workspace-identity prelude	deferred by name from #4025
subscribeProviderConfig, subscribePolicyChanges	buffer: "latest"
subscribeDevTools	initial (async snapshot)
subscribeLogs	initial (snapshot captured at attach)
subscribeTimeline	initial + pre-snapshot event buffering	catch-up/dedup logic unchanged
subscribeWorkspaceChat	heartbeat, initialize (history replay)	replay relay unchanged; FIFO chunks preserve batch delivery
subscribeMetadata, subscribeWorkspaceActivity	default / heartbeat
subscribeBackgroundBashes	buffer: "latest", initialize, emit.end + onEnd	coalesced reader now constructed at attach (was eager); bootstrap-error contract unchanged
subscribeWorkspaceStats	buffer: "latest", initialize	throttle/serialization closure unchanged
subscribeTerminalOutput, attachTerminal	default / initial (screen state)	subscribe-before-capture handshake preserved
subscribeTerminalExit	take: 1
subscribeTerminalActivity	heartbeat, initial
subscribeUpdateStatus, subscribeOpenSettings, subscribeSshPrompts	default	ssh prompt responder release stays paired with unsubscribe in one detach thunk

Deferred (with reasons):

Item	Reason
createTickIterable (general.tick)	Pure timed generator: no event source to attach, no resource to release — the bridge's acquireRelease lifecycle adds machinery without value. Documented at the definition.
WorkflowService internal use of asyncIterableFromSubscription	Service-internal consumption, not a router subscription; out of Phase 9's router-surface scope. The common seam (asyncEventIterator.ts etc.) stays intact for it and for browser-side consumers.

Validation

• 12 new behavioral tests in streamBridge.test.ts pin the genuinely new invariants: listener-count-returns-to-baseline after abort (leak test), teardown on consumer break / stream error / hung-initialize abort / take completion (previously unpinned), synchronous-burst emit ordering, latest-value coalescing, initial-before-buffered ordering, drain-then-onEnd-error completion, heartbeat injection, and laziness.
• make static-check green; full bun test src green.
• tests/ipc run compared against a baseline worktree at the parent commit in the same environment: failure sets are identical (3 suites, all AI-gateway Forbidden/PTY environment failures); websocketHistoryReplay and all other subscription-exercising suites pass with the change.

Risks

Medium-surface change: every UI subscription (chat, metadata, terminals, stats, logs, timeline, prompts) now flows through the new bridge. The main regression classes would be ordering (mitigated: synchronous offerUnsafe, FIFO queues, tests), teardown leaks (mitigated: per-resource acquireRelease + leak tests), and completion/error semantics (mitigated: drain-then-Done queue closing, onEnd contract test, unchanged integration suites). Severity if wrong: stale UI panes or leaked listeners on long-lived sessions.

Lessons for Phase 10 (streamManager seams 1–4)

• Temp-dir Scope (seam 1): Effect.acquireRelease per resource composes cleanly under Stream.unwrap/scoped effects; the temp-dir create/cleanup pair should become one acquireRelease rather than try/finally, and interruption during acquisition is already safe (release only registers after acquire succeeds).
• Partial-write debounce fiber (seam 2): the scope-tied Effect.forkScoped heartbeat ticker here is the template — a debounce fiber owned by the stream/pipeline scope gets interrupted with its owner, removing manual timer bookkeeping. Start such fibers after replay/bootstrap phases if their output must not interleave (same reasoning as heartbeat-after-initialize).
• error/lostResponseIds Refs (seam 3): this phase kept producer-side mutable closures (throttle state, bootstrap flags) as plain variables because producers run in non-Effect contexts. Same rule applies to streamManager: state mutated from AI-SDK callbacks should stay in plain mutables or be bridged via unsafe APIs (Queue.offerUnsafe pattern); Ref is only worth it where fibers are the mutators.
• Usage accounting (seam 4): AI-SDK recordUsage-style callbacks are foreign callsites — the SubscriptionEmit pattern (stable function identities wrapping unsafe queue ops, no suspension between callback and buffer) transfers directly.
• Effect v4 specifics that will recur: Queue.sliding(1) is an exact latest-value-queue replacement; Queue.endUnsafe drains before Done (state Closing); Stream.toAsyncIterable.return() memoizes its close promise, so double-close is safe; type the queue's error channel as Cause.Done at creation (Queue.unbounded<T, Cause.Done>()) or end won't typecheck.

---

Generated with xum • Model: anthropic:claude-fable-5 • Thinking: xhigh • Cost: $15.70

[ThomasK33]

@codex review

[ThomasK33]

@codex security review