fix(bus): acquire PubSub subscription eagerly to close /event race (#27959)

packages/opencode/src/bus/index.ts

@@ -37,8 +37,16 @@ export interface Interface {
     properties: BusProperties<D>,
     options?: { id?: string },
   ) => Effect.Effect<void>
-  readonly subscribe: <D extends BusEvent.Definition>(def: D) => Stream.Stream<Payload<D>>
-  readonly subscribeAll: () => Stream.Stream<Payload>
+  // subscribe / subscribeAll are eager: the underlying PubSub subscription is
+  // acquired in the caller's Scope at `yield*` time. Any publish after the
+  // yield is delivered, even if stream consumption starts later. The previous
+  // Stream-returning shape acquired the subscription lazily on first pull,
+  // opening a race window during which publishes were lost — see
+  // test/bus/bus-effect.test.ts RACE tests.
+  readonly subscribe: <D extends BusEvent.Definition>(
+    def: D,
+  ) => Effect.Effect<Stream.Stream<Payload<D>>, never, Scope.Scope>
+  readonly subscribeAll: () => Effect.Effect<Stream.Stream<Payload>, never, Scope.Scope>
   readonly subscribeCallback: <D extends BusEvent.Definition>(
     def: D,
     callback: (event: Payload<D>) => unknown,
@@ -109,26 +117,26 @@ export const layer = Layer.effect(
       })
     }
 
-    function subscribe<D extends BusEvent.Definition>(def: D): Stream.Stream<Payload<D>> {
-      log.info("subscribing", { type: def.type })
-      return Stream.unwrap(
-        Effect.gen(function* () {
-          const s = yield* InstanceState.get(state)
-          const ps = yield* getOrCreate(s, def)
-          return Stream.fromPubSub(ps)
-        }),
-      ).pipe(Stream.ensuring(Effect.sync(() => log.info("unsubscribing", { type: def.type }))))
-    }
+    const subscribe = <D extends BusEvent.Definition>(
+      def: D,
+    ): Effect.Effect<Stream.Stream<Payload<D>>, never, Scope.Scope> =>
+      Effect.gen(function* () {
+        log.info("subscribing", { type: def.type })
+        const s = yield* InstanceState.get(state)
+        const ps = yield* getOrCreate(s, def)
+        const subscription = yield* PubSub.subscribe(ps)
+        yield* Effect.addFinalizer(() => Effect.sync(() => log.info("unsubscribing", { type: def.type })))
+        return Stream.fromSubscription(subscription)
+      })
 
-    function subscribeAll(): Stream.Stream<Payload> {
-      log.info("subscribing", { type: "*" })
-      return Stream.unwrap(
-        Effect.gen(function* () {
-          const s = yield* InstanceState.get(state)
-          return Stream.fromPubSub(s.wildcard)
-        }),
-      ).pipe(Stream.ensuring(Effect.sync(() => log.info("unsubscribing", { type: "*" }))))
-    }
+    const subscribeAll = (): Effect.Effect<Stream.Stream<Payload>, never, Scope.Scope> =>
+      Effect.gen(function* () {
+        log.info("subscribing", { type: "*" })
+        const s = yield* InstanceState.get(state)
+        const subscription = yield* PubSub.subscribe(s.wildcard)
+        yield* Effect.addFinalizer(() => Effect.sync(() => log.info("unsubscribing", { type: "*" })))
+        return Stream.fromSubscription(subscription)
+      })
 
     function on<T>(pubsub: PubSub.PubSub<T>, type: string, callback: (event: T) => unknown) {
       return Effect.gen(function* () {

packages/opencode/src/plugin/index.ts

@@ -243,7 +243,7 @@ export const layer = Layer.effect(
         }
 
         // Subscribe to bus events, fiber interrupted when scope closes
-        yield* bus.subscribeAll().pipe(
+        yield* (yield* bus.subscribeAll()).pipe(
           Stream.runForEach((input) =>
             Effect.sync(() => {
               for (const hook of hooks) {

packages/opencode/src/project/project.ts

@@ -425,7 +425,7 @@ export const layer: Layer.Layer<
 
     const initState = yield* InstanceState.make(
       Effect.fn("Project.initState")(function* (ctx) {
-        yield* bus.subscribe(Command.Event.Executed).pipe(
+        yield* (yield* bus.subscribe(Command.Event.Executed)).pipe(
           Stream.runForEach((payload) =>
             payload.properties.name === Command.Default.INIT ? setInitialized(ctx.project.id) : Effect.void,
           ),

packages/opencode/src/project/vcs.ts

@@ -298,7 +298,7 @@ export const layer: Layer.Layer<Service, never, Git.Service | Bus.Service> = Lay
         const value = { current, root }
         log.info("initialized", { branch: value.current, default_branch: value.root?.name })
 
-        yield* bus.subscribe(FileWatcher.Event.Updated).pipe(
+        yield* (yield* bus.subscribe(FileWatcher.Event.Updated)).pipe(
           Stream.filter((evt) => evt.properties.file.endsWith("HEAD")),
           Stream.runForEach((_evt) =>
             Effect.gen(function* () {

packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts

@@ -20,10 +20,11 @@ function eventData(data: unknown): Sse.Event {
 
 function eventResponse(bus: Bus.Interface) {
   return Effect.gen(function* () {
-    const context = yield* Effect.context()
-
-    const events = bus.subscribeAll().pipe(
-      Stream.provideContext(context),
+    // Subscribe eagerly: the bus subscription is acquired in the request scope
+    // at this yield, so any publish from now on is queued for the body-pump
+    // fiber to drain — closing the race where Stream.concat(server.connected,
+    // lazy-subscribe) used to drop publishes in the prefix-consume window.
+    const events = (yield* bus.subscribeAll()).pipe(
       Stream.takeUntil((event) => event.type === Bus.InstanceDisposed.type),
     )
     const heartbeat = Stream.tick("10 seconds").pipe(

packages/opencode/src/share/share-next.ts

@@ -168,16 +168,20 @@ export const layer = Layer.effect(
           fn: (evt: { properties: any }) => Effect.Effect<void, unknown>,
         ) =>
           bus.subscribe(def as never).pipe(
-            Stream.runForEach((evt) =>
-              fn(evt).pipe(
-                Effect.catchCause((cause) =>
-                  Effect.sync(() => {
-                    log.error("share subscriber failed", { type: def.type, cause })
-                  }),
+            Effect.flatMap((stream) =>
+              stream.pipe(
+                Stream.runForEach((evt) =>
+                  fn(evt).pipe(
+                    Effect.catchCause((cause) =>
+                      Effect.sync(() => {
+                        log.error("share subscriber failed", { type: def.type, cause })
+                      }),
+                    ),
+                  ),
                 ),
+                Effect.forkScoped,
               ),
             ),
-            Effect.forkScoped,
           )
 
         yield* watch(Session.Event.Updated, (evt) =>