refactor(core): remove unused session watermarks

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/src/session.ts

@@ -142,8 +142,6 @@ export type Error =
 export interface Interface {
   readonly list: (input?: ListInput) => Effect.Effect<{
     readonly data: SessionSchema.Info[]
-    /** Per-session durable log watermark, read in the same transaction as the snapshot. Sessions without events are absent. */
-    readonly watermarks: ReadonlyMap<string, EventV2.Seq>
   }>
   readonly create: (input: CreateInput) => Effect.Effect<SessionSchema.Info, NotFoundError>
   readonly fork: (input: ForkInput) => Effect.Effect<SessionSchema.Info, NotFoundError | MessageNotFoundError>
@@ -177,8 +175,6 @@ export interface Interface {
     after?: number
     follow?: boolean
   }) => Stream.Stream<SessionEvent.DurableEvent | EventLog.Synced, NotFoundError>
-  /** Latest durable log seq per session. Sessions without events are absent. */
-  readonly watermarks: (sessionIDs: ReadonlyArray<SessionSchema.ID>) => Effect.Effect<ReadonlyMap<string, EventV2.Seq>>
   readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect<void, NotFoundError>
   readonly switchModel: (input: {
     sessionID: SessionSchema.ID
@@ -394,21 +390,10 @@ const layer = Layer.effect(
             order === "asc" ? asc(sortColumn) : desc(sortColumn),
             order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id),
           )
-        // Watermarks must pair with the snapshot exactly, so both reads share a transaction:
-        // a higher watermark would let an attached tail skip events missing from the snapshot.
-        const snapshot = yield* db
-          .transaction(() =>
-            Effect.gen(function* () {
-              const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
-                Effect.orDie,
-              )
-              const watermarks = yield* events.sequences(rows.map((row) => row.id))
-              return { rows, watermarks }
-            }),
-          )
-          .pipe(Effect.orDie)
-        const rows = direction === "previous" ? snapshot.rows.toReversed() : snapshot.rows
-        return { data: rows.map((row) => fromRow(row)), watermarks: snapshot.watermarks }
+        const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe(
+          Effect.orDie,
+        )
+        return { data: (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row)) }
       }),
       messages: Effect.fn("V2Session.messages")(function* (input) {
         yield* result.get(input.sessionID)
@@ -463,9 +448,6 @@ const layer = Layer.effect(
               EventV2.isSynced(item) || isDurableSessionEvent(item),
           ),
         ),
-      watermarks: Effect.fn("V2Session.watermarks")(function* (sessionIDs) {
-        return yield* events.sequences(sessionIDs)
-      }),
       prompt: Effect.fn("V2Session.prompt")((input) =>
         Effect.uninterruptible(
           Effect.gen(function* () {
@@ -540,8 +522,7 @@ const layer = Layer.effect(
             const started = yield* Effect.gen(function* () {
               const shell = yield* Shell.Service
               return yield* shell.create({ command: input.command, cwd: session.location.directory })
-            })
-              .pipe(Effect.provide(locations.get(session.location)))
+            }).pipe(Effect.provide(locations.get(session.location)))
             yield* events.publish(
               SessionEvent.Shell.Started,
               {

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-log.test.ts

@@ -130,33 +130,3 @@ describe("SessionV2.log", () => {
     }),
   )
 })
-
-describe("SessionV2 watermarks", () => {
-  it.effect("list pairs each session snapshot with its durable log watermark", () =>
-    Effect.gen(function* () {
-      const session = yield* SessionV2.Service
-      const events = yield* EventV2.Service
-      const first = yield* session.create({ location })
-      const second = yield* session.create({ location })
-      yield* session.rename({ sessionID: first.id, title: "session.renamed" })
-
-      const page = yield* session.list()
-      const sequences = yield* events.sequences([first.id, second.id])
-
-      expect(page.data.map((info) => info.id).toSorted()).toEqual([first.id, second.id].toSorted())
-      expect(page.watermarks).toEqual(sequences)
-      expect(page.watermarks.get(first.id)).toBeGreaterThan(page.watermarks.get(second.id)!)
-    }),
-  )
-
-  it.effect("watermarks omits sessions without durable events", () =>
-    Effect.gen(function* () {
-      const session = yield* SessionV2.Service
-      const created = yield* session.create({ location })
-
-      const watermarks = yield* session.watermarks([created.id, SessionV2.ID.create()])
-
-      expect(Array.from(watermarks.keys())).toEqual([created.id])
-    }),
-  )
-})

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),