refactor(core): unify live event subscriptions (#35578)

packages/core/src/event.ts

@@ -118,6 +118,11 @@ export type SubscribePayload<D extends readonly Definition[]> = D[number] extend
   : never
 
 export interface Subscribe {
+  /**
+   * Volatile live channel: every event published from now on, nothing before or
+   * across a disconnect. Consumers that need reliability combine it with `log`.
+   */
+  (): Stream.Stream<Payload>
   <D extends Definition>(definition: D): Stream.Stream<Payload<D>>
   <const D extends readonly [Definition, ...Definition[]]>(definitions: D): Stream.Stream<SubscribePayload<D>>
 }
@@ -131,12 +136,6 @@ export interface Interface {
     options?: PublishOptions,
   ) => Effect.Effect<Payload<D>>
   readonly subscribe: Subscribe
-  /**
-   * Volatile live channel: every event published from now on, nothing before,
-   * nothing across a disconnect. The only channel that carries non-durable
-   * events; consumers that need reliability combine `changes` with `log`.
-   */
-  readonly live: () => Stream.Stream<Payload>
   /**
    * Durable, ordered, gap-free per-aggregate log read. `follow: false`
    * completes at the end of the log; `follow: true` replays then transitions
@@ -150,7 +149,7 @@ export interface Interface {
   }) => Stream.Stream<LogItem>
   /** Latest committed seq per aggregate. Aggregates without events are absent. */
   readonly sequences: (aggregateIDs: ReadonlyArray<string>) => Effect.Effect<ReadonlyMap<string, Seq>>
-  /** @deprecated Use `all()` and consume the returned stream. */
+  /** @deprecated Use `subscribe()` and consume the returned stream. */
   readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
   readonly project: <D extends Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
   readonly replay: (
@@ -575,17 +574,16 @@ export const layerWith = (options?: LayerOptions) =>
           ),
         )
 
-      const subscribeOne = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>
-        local(Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub))))).pipe(
-          Stream.map((event) => event as Payload<D>),
-        )
-
+      function subscribe(): Stream.Stream<Payload>
       function subscribe<D extends Definition>(definition: D): Stream.Stream<Payload<D>>
       function subscribe<const D extends readonly [Definition, ...Definition[]]>(
         definitions: D,
       ): Stream.Stream<SubscribePayload<D>>
-      function subscribe(input: Definition | readonly Definition[]): Stream.Stream<Payload> {
-        if (isDefinition(input)) return subscribeOne(input)
+      function subscribe(input?: Definition | readonly Definition[]): Stream.Stream<Payload> {
+        if (input === undefined) return streamLive()
+        if (isDefinition(input)) {
+          return local(Stream.unwrap(getOrCreate(input).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))))
+        }
         const types = new Set(input.map((definition) => definition.type))
         return streamLive().pipe(Stream.filter((event) => types.has(event.type)))
       }
@@ -737,7 +735,6 @@ export const layerWith = (options?: LayerOptions) =>
       return Service.of({
         publish,
         subscribe,
-        live: streamLive,
         log,
         sequences,
         listen,

packages/core/src/plugin/host.ts

@@ -159,7 +159,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: PluginV2.Int
         }),
     },
     event: {
-      subscribe: () => events.live().pipe(Stream.filter(EventManifest.isServer)),
+      subscribe: () => events.subscribe().pipe(Stream.filter(EventManifest.isServer)),
     },
     integration: {
       list: () => response(integration.list()),

packages/core/test/event.test.ts

@@ -177,7 +177,7 @@ describe("EventV2", () => {
     Effect.gen(function* () {
       const events = yield* EventV2.Service
       const typed = yield* events.subscribe(Message).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
-      const wildcard = yield* events.live().pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
+      const wildcard = yield* events.subscribe().pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
       yield* Effect.yieldNow
       const event = yield* events.publish(Message, { text: "hello" })
 
@@ -258,7 +258,7 @@ describe("EventV2", () => {
     Effect.gen(function* () {
       const events = yield* EventV2.Service
       const received = new Array<string>()
-      const fiber = yield* events.live().pipe(
+      const fiber = yield* events.subscribe().pipe(
         Stream.take(1),
         Stream.runForEach(() => Effect.sync(() => received.push("stream"))),
         Effect.forkScoped,

packages/core/test/session-runner-tool-events.test.ts

@@ -27,7 +27,6 @@ const capture = () => {
         return event
       }),
     subscribe: () => Stream.empty,
-    live: () => Stream.empty,
     log: () => Stream.empty,
     sequences: () => Effect.succeed(new Map()),
     listen: () => Effect.succeed(Effect.void),