diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 1e228af47d..a961d5ab50 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -57,6 +57,7 @@ import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; +import * as Scheduler from "effect/Scheduler"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -6917,6 +6918,34 @@ class RuntimeReaper extends Context.Service()( describe("agent browser access", () => { const projectId = ProjectId.make("project-browser-access"); + it.effect("subscribes to adapter events before exposing the provider service", () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const events = yield* PubSub.unbounded(); + let subscribed = false; + const adapter: ProviderAdapterShape = { + ...codex.adapter, + streamEvents: Stream.unwrap( + Effect.map(PubSub.subscribe(events), (subscription) => { + // This runs only after PubSub has registered the subscription, not merely when + // the stream wrapper is constructed. + subscribed = true; + return Stream.fromSubscription(subscription); + }), + ), + }; + yield* Effect.gen(function* () { + yield* ProviderService.ProviderService; + assert.isTrue(subscribed); + }).pipe(Effect.provide(makeAgentBrowserProviderLayer(false, { ...codex, adapter }, {}))); + }).pipe( + // Keep this one fiber from yielding before the assertion. The old deferred fork then + // reproducibly leaves `subscribed` false; immediate startup passes without a timer. + Effect.provideService(Scheduler.PreventSchedulerYield, true), + Effect.provide(NodeServices.layer), + ), + ); + const makeBrowserAccessProjectionLayer = ( threadId: ThreadId, status: "found" | "missing" | "failed" = "found", diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 2fc810c8a3..e999423390 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -1758,6 +1758,8 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( // Publish the new adapter set before any replacement stream can emit. yield* Ref.set(subscribedAdapters, next); for (const [id, adapter] of newSubscriptions) { + // A provider can emit a receipt synchronously during an API call. Subscribe before + // returning from reconciliation so that first receipt cannot outrun a deferred fork. yield* Stream.runForEach(adapter.streamEvents, (event) => processRuntimeEvent( { @@ -1767,7 +1769,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }, event, ), - ).pipe(Effect.forkScoped); + ).pipe(Effect.forkScoped({ startImmediately: true })); } });