refactor(core): remove unused bus sequences (#40681)

packages/core/src/bus.ts

@@ -3,7 +3,7 @@ export * as Bus from "./bus"
 import { Cause, Context, DateTime, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
 import type { EventLog } from "@opencode-ai/schema/event-log"
-import { and, asc, eq, gt, inArray, lte, sql } from "drizzle-orm"
+import { and, asc, eq, gt, lte, sql } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
 import { Location } from "./location"
@@ -134,8 +134,6 @@ export interface Interface {
     readonly after?: number
     readonly follow?: boolean
   }) => Stream.Stream<LogItem>
-  /** Latest committed seq per aggregate. Aggregates without events are absent. */
-  readonly sequences: (aggregateIDs: ReadonlyArray<string>) => Effect.Effect<ReadonlyMap<string, Event.Seq>>
   /** @deprecated Use `subscribe()` and consume the returned stream. */
   readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
   readonly project: <D extends Event.Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
@@ -657,19 +655,6 @@ export const layerWith = (options?: LayerOptions) =>
           }),
         )
 
-      const sequences = (aggregateIDs: ReadonlyArray<string>): Effect.Effect<ReadonlyMap<string, Event.Seq>> => {
-        if (aggregateIDs.length === 0) return Effect.succeed(new Map())
-        return db
-          .select({ aggregateID: EventSequenceTable.aggregate_id, seq: EventSequenceTable.seq })
-          .from(EventSequenceTable)
-          .where(inArray(EventSequenceTable.aggregate_id, Array.from(aggregateIDs)))
-          .all()
-          .pipe(
-            Effect.orDie,
-            Effect.map((rows) => new Map(rows.map((row) => [row.aggregateID, Event.Seq.make(row.seq)]))),
-          )
-      }
-
       const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
         Effect.sync(() => {
           listeners.push(listener)
@@ -691,7 +676,6 @@ export const layerWith = (options?: LayerOptions) =>
         publish,
         subscribe,
         log,
-        sequences,
         listen,
         project,
         replay,

packages/core/test/bus.test.ts

@@ -1298,24 +1298,4 @@ describe("Bus", () => {
     }),
   )
 
-  it.effect("sequences returns the latest committed seq per aggregate and omits unknown aggregates", () =>
-    Effect.gen(function* () {
-      const bus = yield* Bus.Service
-      const first = Session.ID.create()
-      const second = Session.ID.create()
-      yield* bus.publish(DurableMessage, durableData(first, "zero"))
-      yield* bus.publish(DurableMessage, durableData(first, "one"))
-      yield* bus.publish(DurableMessage, durableData(second, "zero"))
-
-      const sequences = yield* bus.sequences([first, second, Session.ID.create()])
-
-      expect(sequences).toEqual(
-        new Map([
-          [first, Event.Seq.make(1)],
-          [second, Event.Seq.make(0)],
-        ]),
-      )
-      expect(yield* bus.sequences([])).toEqual(new Map())
-    }),
-  )
 })

packages/core/test/session-log.test.ts

@@ -41,17 +41,15 @@ describe("Session.log", () => {
   it.effect("replays public session events and marks synced at the aggregate watermark", () =>
     Effect.gen(function* () {
       const session = yield* Session.Service
-      const bus = yield* Bus.Service
       const created = yield* session.create({ location })
       yield* session.rename({ sessionID: created.id, title: "session.renamed" })
 
       const items = Array.from(yield* Stream.runCollect(session.log({ sessionID: created.id })))
-      const watermark = (yield* bus.sequences([created.id])).get(created.id)
 
       // Session creation commits a non-public durable event, so the marker's
       // seq covers more of the aggregate than the public events emitted.
       expect(items.map((item) => item.type)).toEqual(["session.renamed", "log.synced"])
-      expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: watermark })
+      expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: Event.Seq.make(1) })
     }),
   )