From 6ca595202e42e9a9a09a271ecfae0e023f6e5d6f Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 4 Sep 2026 15:02:10 +0000 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=A4=96=20refactor:=20bind=20subscript?= =?UTF-8?q?ions=20to=20runtime=20context=20(Wave=204=20PR=204)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep the no-context path unchanged; strip build-fiber artifacts before exporting the context and pin virtual-clock cadence and teardown. --- _Generated with [`mux`](https://github.com/coder/mux) • Model: `coder:openai/gpt-6-astra` • Thinking: `high`_ --- src/node/orpc/router.ts | 4 +- src/node/orpc/routerSubscriptions.ts | 45 +++++----- src/node/orpc/streamBridge.test.ts | 107 +++++++++++++++++------- src/node/orpc/streamBridge.ts | 10 ++- src/node/services/di/appRuntime.test.ts | 13 ++- src/node/services/di/appRuntime.ts | 26 ++++-- 6 files changed, 147 insertions(+), 58 deletions(-) diff --git a/src/node/orpc/router.ts b/src/node/orpc/router.ts index ebbc2f51dc..37c2c13fa7 100644 --- a/src/node/orpc/router.ts +++ b/src/node/orpc/router.ts @@ -908,7 +908,9 @@ export const router = (authToken?: string) => { subscribeLogs: t .input(schemas.general.subscribeLogs.input) .output(schemas.general.subscribeLogs.output) - .handler(({ input, signal }) => subscribeLogs(input.level ?? "info", signal)), + .handler(({ context, input, signal }) => + subscribeLogs(context, input.level ?? "info", signal) + ), restartApp: t .input(schemas.general.restartApp.input) .output(schemas.general.restartApp.output) diff --git a/src/node/orpc/routerSubscriptions.ts b/src/node/orpc/routerSubscriptions.ts index 9c3b6ba5a0..8f0a115245 100644 --- a/src/node/orpc/routerSubscriptions.ts +++ b/src/node/orpc/routerSubscriptions.ts @@ -16,7 +16,7 @@ import type { DevToolsEvent } from "@/common/types/devtools"; import { createCoalescedReader } from "@/common/utils/coalescedReader"; import { getErrorMessage } from "@/common/utils/errors"; import type { ORPCContext } from "./context"; -import { subscriptionIterable } from "./streamBridge"; +import { subscriptionIterable, type SubscriptionStreamOptions } from "./streamBridge"; import { createReplayBufferedStreamMessageRelay } from "@/node/services/replayBufferedStreamMessageRelay"; import { TIMELINE_DEFAULT_PAGE_LIMIT } from "@/node/services/timelineService"; import type { LogEntry } from "@/node/services/logBuffer"; @@ -70,6 +70,10 @@ const LOG_LEVEL_PRIORITY: Record = { debug: 3, }; +function runtimeSubscription(context: ORPCContext, options: SubscriptionStreamOptions) { + return subscriptionIterable({ ...options, context: context["effect/context"] }); +} + function shouldIncludeLogEntry( entryLevel: LogEntry["level"], minLevel: LogEntry["level"] @@ -84,7 +88,7 @@ export function subscribeConfigChanges( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, buffer: "latest", subscribe: (emit) => context.config.onConfigChanged(() => emit.push(undefined)), @@ -97,7 +101,7 @@ export function subscribeDevTools( signal?: AbortSignal ): AsyncGenerator { const service = context.devToolsService; - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => { const eventName = "update:" + workspaceId; @@ -112,7 +116,7 @@ export function subscribeProviderConfig( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, buffer: "latest", subscribe: (emit) => context.providerService.onConfigChanged(() => emit.push(undefined)), @@ -123,7 +127,7 @@ export function subscribePolicyChanges( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, buffer: "latest", subscribe: (emit) => context.policyService.onPolicyChanged(() => emit.push(undefined)), @@ -148,11 +152,12 @@ export function createTickIterable( } export function subscribeLogs( + context: ORPCContext, minLevel: LogEntry["level"], signal?: AbortSignal ): AsyncGenerator { let snapshot: ReturnType["snapshot"]; - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => { const subscription = subscribeLogFeed((event) => { @@ -185,7 +190,7 @@ export function subscribeMemoryChanges( validate?.(); const metadata = workspaceId ? await context.workspaceService.getInfo(workspaceId) : null; const projectPath = metadata ? resolveMemoryProjectIdentity(metadata) : null; - yield* subscriptionIterable({ + yield* runtimeSubscription(context, { signal, subscribe: (emit) => { const onChange = (event: MemoryChangeEvent) => { @@ -214,7 +219,7 @@ export function subscribeTimeline( const pendingEvents: TimelineSubscriptionEvent["events"] = []; let snapshotSequence: number | undefined; let pushEvent: ((event: TimelineSubscriptionEvent) => void) | undefined; - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => { pushEvent = emit.push; @@ -266,7 +271,7 @@ export function subscribeWorkspaceChat( } let replayRelay: ReturnType; // Subscribe before replay so the relay can buffer overlapping live deltas. - return subscriptionIterable({ + return runtimeSubscription(context, { signal, heartbeat: { value: { type: "heartbeat" as const } }, subscribe: (emit) => { @@ -285,7 +290,7 @@ export function subscribeMetadata( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => { context.workspaceService.on("metadata", emit.push); @@ -298,7 +303,7 @@ export function subscribeWorkspaceActivity( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, heartbeat: { value: { type: "heartbeat" } }, subscribe: (emit) => { @@ -326,7 +331,7 @@ export function subscribeBackgroundBashes( let reader: ReturnType | undefined; // Full snapshots coalesce ("latest") because replaying stale intermediate // state only grows memory. - return subscriptionIterable>>({ + return runtimeSubscription>>(context, { signal, buffer: "latest", subscribe: (emit) => { @@ -408,7 +413,7 @@ export function subscribeWorkspaceStats( }, remaining); pendingTimer.unref?.(); }; - return subscriptionIterable({ + return runtimeSubscription(context, { signal, buffer: "latest", subscribe: (emit) => { @@ -443,7 +448,7 @@ export function subscribeTerminalOutput( sessionId: string, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => context.terminalService.onOutput(sessionId, emit.push), }); @@ -455,7 +460,7 @@ export function attachTerminal( signal?: AbortSignal ): AsyncGenerator { // Output subscribes before screen capture so attach cannot lose bytes in the handshake. - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => context.terminalService.onOutput(sessionId, (data) => emit.push({ type: "output", data })), @@ -471,7 +476,7 @@ export function subscribeTerminalExit( sessionId: string, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => context.terminalService.onExit(sessionId, emit.push), take: 1, @@ -482,7 +487,7 @@ export function subscribeTerminalActivity( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, heartbeat: { value: { type: "heartbeat" } }, subscribe: (emit) => @@ -504,7 +509,7 @@ export function subscribeUpdateStatus( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => context.updateService.onStatus(emit.push), }); @@ -514,7 +519,7 @@ export function subscribeOpenSettings( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => context.menuEventService.onOpenSettings(() => emit.push(undefined)), }); @@ -524,7 +529,7 @@ export function subscribeSshPrompts( context: ORPCContext, signal?: AbortSignal ): AsyncGenerator { - return subscriptionIterable({ + return runtimeSubscription(context, { signal, subscribe: (emit) => { const releaseResponder = context.sshPromptService.registerInteractiveResponder(); diff --git a/src/node/orpc/streamBridge.test.ts b/src/node/orpc/streamBridge.test.ts index b9d18eb1f2..a1231b3499 100644 --- a/src/node/orpc/streamBridge.test.ts +++ b/src/node/orpc/streamBridge.test.ts @@ -9,6 +9,9 @@ * error propagation), not implementation literals. */ import { describe, expect, test } from "bun:test"; +import { Context, Effect } from "effect"; +import { TestClock } from "effect/testing"; +import { disposeAppRuntime, makeAppRuntime } from "@/node/services/di/appRuntime"; import { EventEmitter } from "node:events"; import { subscriptionIterable, type SubscriptionEmit } from "./streamBridge"; @@ -30,11 +33,12 @@ async function collect(iterable: AsyncGenerator, count: number): Promise { +describe.each([undefined, Context.empty()])("subscriptionIterable teardown (%p)", (context) => { test("listener count returns to baseline after client abort", async () => { const emitter = new EventEmitter(); const controller = new AbortController(); const iterable = subscriptionIterable({ + context, signal: controller.signal, subscribe: (emit) => { emitter.on("value", emit.push); @@ -54,12 +58,15 @@ describe("subscriptionIterable teardown", () => { // Abort completes the generator normally (no throw) and detaches. await consumed; + await iterable.return(undefined); + await iterable.return(undefined); expect(emitter.listenerCount("value")).toBe(0); }); test("consumer break (generator return) detaches the listener", async () => { const emitter = new EventEmitter(); const iterable = subscriptionIterable({ + context, subscribe: (emit) => { emitter.on("value", emit.push); return () => emitter.off("value", emit.push); @@ -80,6 +87,7 @@ describe("subscriptionIterable teardown", () => { const emitter = new EventEmitter(); const boom = new Error("bootstrap failed"); const iterable = subscriptionIterable({ + context, subscribe: (emit) => { emitter.on("value", emit.push); return () => emitter.off("value", emit.push); @@ -100,6 +108,7 @@ describe("subscriptionIterable teardown", () => { const emitter = new EventEmitter(); const controller = new AbortController(); const iterable = subscriptionIterable({ + context, signal: controller.signal, subscribe: (emit) => { emitter.on("value", emit.push); @@ -118,6 +127,7 @@ describe("subscriptionIterable teardown", () => { test("take completes the stream and detaches immediately", async () => { const emitter = new EventEmitter(); const iterable = subscriptionIterable({ + context, subscribe: (emit) => { emitter.on("exit", emit.push); return () => emitter.off("exit", emit.push); @@ -138,6 +148,7 @@ describe("subscriptionIterable teardown", () => { controller.abort(); let subscribed = false; const iterable = subscriptionIterable({ + context, signal: controller.signal, subscribe: (emit) => { subscribed = true; @@ -194,22 +205,30 @@ describe("subscriptionIterable ordering and buffering", () => { }); 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"; - }, - }); + const app = makeAppRuntime(TestClock.layer()); + try { + let emitHandle: SubscriptionEmit | undefined; + const iterable = subscriptionIterable({ + context: app.context, + 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 app.managed.runPromise(Effect.sleep(1)); + return "snapshot"; + }, + }); - expect(await collect(iterable, 2)).toEqual(["snapshot", "during-initial"]); + const consumed = collect(iterable, 2); + await app.managed.runPromise(TestClock.adjust(1)); + expect(await consumed).toEqual(["snapshot", "during-initial"]); + } finally { + await disposeAppRuntime(app.managed); + } }); test("emit.end drains buffered values, then onEnd error surfaces", async () => { @@ -237,23 +256,53 @@ describe("subscriptionIterable ordering and buffering", () => { }); test("heartbeat values are injected while the subscription is idle", async () => { + const app = makeAppRuntime(TestClock.layer()); + const controller = new AbortController(); + let detached = 0; + const values: string[] = []; + const intervalMs = 1_000; const iterable = subscriptionIterable({ - heartbeat: { value: "heartbeat", intervalMs: 10 }, - subscribe: () => () => undefined, + context: app.context, + signal: controller.signal, + heartbeat: { value: "heartbeat", intervalMs }, + subscribe: () => () => { + detached += 1; + }, }); - expect(await collect(iterable, 2)).toEqual(["heartbeat", "heartbeat"]); + const consumed = (async () => { + for await (const value of iterable) values.push(value); + })(); + try { + const startedAt = performance.now(); + await app.managed.runPromise(TestClock.adjust(3 * intervalMs)); + expect(values).toEqual(["heartbeat", "heartbeat", "heartbeat"]); + expect(performance.now() - startedAt).toBeLessThan(50); + } finally { + controller.abort(); + await consumed; + await iterable.return(undefined); + await iterable.return(undefined); + await disposeAppRuntime(app.managed); + } + expect(detached).toBe(1); }); 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); + const app = makeAppRuntime(TestClock.layer()); + try { + let subscribed = false; + const iterable = subscriptionIterable({ + context: app.context, + subscribe: () => { + subscribed = true; + return () => undefined; + }, + }); + await app.managed.runPromise(TestClock.adjust(10)); + expect(subscribed).toBe(false); + await iterable.return(undefined); + } finally { + await disposeAppRuntime(app.managed); + } }); }); diff --git a/src/node/orpc/streamBridge.ts b/src/node/orpc/streamBridge.ts index 3e9ef35909..9b36770b36 100644 --- a/src/node/orpc/streamBridge.ts +++ b/src/node/orpc/streamBridge.ts @@ -36,7 +36,7 @@ * - **Laziness**: nothing (not even `validate`) runs until the consumer's * first `next()` call, matching async-generator semantics. */ -import type { Cause } from "effect"; +import type { Cause, Context } from "effect"; import { Effect, Queue, Stream } from "effect"; import { SUBSCRIPTION_HEARTBEAT_INTERVAL_MS } from "@/common/utils/withQueueHeartbeat"; @@ -55,6 +55,8 @@ export interface SubscriptionEmit { } export interface SubscriptionStreamOptions { + /** Runtime references (notably Clock); omitted for the original global-runtime path. */ + context?: Context.Context; signal?: AbortSignal; /** Runs first; a throw rejects the subscription before any resource is acquired. */ validate?: () => void; @@ -177,7 +179,11 @@ export function subscriptionIterable(options: SubscriptionStreamOptions): return (async function* () { if (options.signal?.aborted) return; - const iterator = Stream.toAsyncIterable(subscriptionStream(options))[Symbol.asyncIterator](); + const stream = subscriptionStream(options); + const iterable = options.context + ? Stream.toAsyncIterableWith(stream, options.context) + : Stream.toAsyncIterable(stream); + const iterator = iterable[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?.(); diff --git a/src/node/services/di/appRuntime.test.ts b/src/node/services/di/appRuntime.test.ts index 5d27867a39..cfceead0eb 100644 --- a/src/node/services/di/appRuntime.test.ts +++ b/src/node/services/di/appRuntime.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, spyOn } from "bun:test"; -import { Context, Effect, Layer } from "effect"; +import { Context, Effect, Layer, Scheduler, Scope } from "effect"; import { log } from "@/node/services/log"; import { disposeAppRuntime, makeAppRuntime } from "./appRuntime"; @@ -15,6 +15,17 @@ describe("makeAppRuntime", () => { expect(Context.get(app.context, ProbeA)).toBe(app.get(ProbeA)); }); + it("exports a context without build-fiber artifacts for subscription pulls", async () => { + const app = makeAppRuntime(Layer.succeed(ProbeA)({ name: "a" })); + try { + for (const key of [Scope.Scope, Layer.CurrentMemoMap, Scheduler.Scheduler]) { + expect(app.context.mapUnsafe.has(key.key)).toBe(false); + } + } finally { + await disposeAppRuntime(app.managed); + } + }); + it("throws synchronously when a layer body suspends", () => { const asyncLayer = Layer.effect( ProbeA, diff --git a/src/node/services/di/appRuntime.ts b/src/node/services/di/appRuntime.ts index a51fc39741..6571c7c3b6 100644 --- a/src/node/services/di/appRuntime.ts +++ b/src/node/services/di/appRuntime.ts @@ -8,7 +8,9 @@ * `xum workflow`, via `coreServicesRoot.ts`). It owns the app-lifetime * `Scope`, its built `Context` is what oRPC Effect-native handlers * receive as `"effect/context"`, and it is the source of the two runtime seams - * described below. Service classes were not rewritten for this: Layers are + * described below. Subscription streams run on this context, so heartbeat + * sleeps use the runtime's Clock without inheriting build-fiber artifacts. + * Service classes were not rewritten for this: Layers are * thin adapters around the existing constructors, and the former composition * roots' setter/listener wiring lives in `CoreWiringLive`/`DesktopWiringLive` * in the original statement order. @@ -170,7 +172,7 @@ * * Startup as a Layer (would break I1's failure semantics; it became a * runtime-run effect instead, see "Startup"), layer finalizers for the existing - * `dispose()` steps (I5), `streamBridge` on the runtime, per-service optional + * `dispose()` steps (I5), per-service optional * tags (optional cross-cutting services stay optional via `CoreOptionsTag`), * per-step timeouts for `runStartupHousekeeping()` (policy, see "Startup"). The * streamManager engine core became the `AppFiberScope` occupant in Wave 4 PR 1; @@ -178,8 +180,17 @@ * unsupervised (nothing durable exists for it yet). */ import assert from "@/common/utils/assert"; -import { Context, Duration, Effect, Exit, Fiber, ManagedRuntime, Scope } from "effect"; -import type { Layer } from "effect"; +import { + Context, + Duration, + Effect, + Exit, + Fiber, + Layer, + ManagedRuntime, + Scheduler, + Scope, +} from "effect"; import { APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS, APP_RUNTIME_DISPOSE_TIMEOUT_MS, @@ -203,7 +214,12 @@ export interface AppRuntime { export function makeAppRuntime(layer: Layer.Layer): AppRuntime { const startedAt = performance.now(); const managed = ManagedRuntime.make(layer); - const context = managed.runSync(Effect.context()); + // Subscription pulls must not inherit the eager build's scope, memo map or sync scheduler. + const context = Context.omit( + Scope.Scope, + Layer.CurrentMemoMap, + Scheduler.Scheduler + )(managed.runSync(Effect.context())); assert( managed.cachedContext !== undefined, "AppRuntime layer graph must build synchronously (see DI contract in di/appRuntime.ts)" From 3469a7a80184fbb4fabbdbad1ac1de49e6f630ba Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Fri, 4 Sep 2026 15:22:07 +0000 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=A4=96=20tests:=20pin=20handler=20clo?= =?UTF-8?q?ck=20propagation=20and=20type=20the=20context=20projection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- _Generated with [`mux`](https://github.com/coder/mux) • Model: `coder:openai/gpt-6-astra` • Thinking: `high`_ --- src/node/orpc/routerSubscriptions.test.ts | 29 +++++++++++++++++++++++ src/node/services/di/appRuntime.ts | 3 ++- 2 files changed, 31 insertions(+), 1 deletion(-) create mode 100644 src/node/orpc/routerSubscriptions.test.ts diff --git a/src/node/orpc/routerSubscriptions.test.ts b/src/node/orpc/routerSubscriptions.test.ts new file mode 100644 index 0000000000..ee8b7329b5 --- /dev/null +++ b/src/node/orpc/routerSubscriptions.test.ts @@ -0,0 +1,29 @@ +import { expect, test } from "bun:test"; +import { EventEmitter } from "node:events"; +import { TestClock } from "effect/testing"; +import { SUBSCRIPTION_HEARTBEAT_INTERVAL_MS } from "@/common/utils/withQueueHeartbeat"; +import { disposeAppRuntime, makeAppRuntime } from "@/node/services/di/appRuntime"; +import type { ORPCContext } from "./context"; +import { subscribeWorkspaceActivity } from "./routerSubscriptions"; + +test("subscription handlers forward the oRPC runtime Clock", async () => { + const app = makeAppRuntime(TestClock.layer()); + const workspaceService = new EventEmitter(); + const controller = new AbortController(); + const context = { "effect/context": app.context, workspaceService } as unknown as ORPCContext; + const events: unknown[] = []; + const consumed = (async () => { + for await (const event of subscribeWorkspaceActivity(context, controller.signal)) { + events.push(event); + } + })(); + try { + await app.managed.runPromise(TestClock.adjust(SUBSCRIPTION_HEARTBEAT_INTERVAL_MS)); + expect(events).toEqual([{ type: "heartbeat" }]); + } finally { + controller.abort(); + await consumed; + await disposeAppRuntime(app.managed); + } + expect(workspaceService.listenerCount("activity")).toBe(0); +}); diff --git a/src/node/services/di/appRuntime.ts b/src/node/services/di/appRuntime.ts index 6571c7c3b6..7fb7ceebde 100644 --- a/src/node/services/di/appRuntime.ts +++ b/src/node/services/di/appRuntime.ts @@ -215,11 +215,12 @@ export function makeAppRuntime(layer: Layer.Layer): AppRunti const startedAt = performance.now(); const managed = ManagedRuntime.make(layer); // Subscription pulls must not inherit the eager build's scope, memo map or sync scheduler. + // These artifacts are not declared app services in R (see the DI contract). const context = Context.omit( Scope.Scope, Layer.CurrentMemoMap, Scheduler.Scheduler - )(managed.runSync(Effect.context())); + )(managed.runSync(Effect.context())) as Context.Context; assert( managed.cachedContext !== undefined, "AppRuntime layer graph must build synchronously (see DI contract in di/appRuntime.ts)"