diff --git a/packages/core/src/bus.ts b/packages/core/src/bus.ts index dcb14cf2817..6ca5da6bb4d 100644 --- a/packages/core/src/bus.ts +++ b/packages/core/src/bus.ts @@ -11,6 +11,9 @@ import { KeyedMutex } from "./effect/keyed-mutex.js" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { isDeepStrictEqual } from "node:util" import { Durable } from "@opencode-ai/schema/durable-event-manifest" +import { SessionEvent } from "@opencode-ai/schema/session-event" +import type { SessionID } from "@opencode-ai/schema/session-id" +import { AbsolutePath } from "@opencode-ai/schema/schema" export type Subscriber = (event: Event.Payload) => Effect.Effect export type Unsubscribe = Effect.Effect @@ -119,6 +122,9 @@ 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`. + * With an ambient Location, delivery is restricted to that Location and global + * events. Unlocated Session events use the Session's owner at publication time. + * Session moves reach both the old and new Location, without changing the event. */ (): Stream.Stream (definition: D): Stream.Stream> @@ -183,6 +189,7 @@ export function configured(options?: Options) { // Deferred import: a static one would close the module cycle // bus → location → project → bus and hit the node bindings in TDZ. const { Location } = yield* Effect.promise(() => import("./location.js")) + const { SessionTable } = yield* Effect.promise(() => import("./session/sql.js")) const pubsub = { live: yield* PubSub.unbounded(), durable: new Map>>(), @@ -194,6 +201,64 @@ export function configured(options?: Options) { const { db } = yield* Database.Service const logReadPageSize = options?.logReadPageSize ?? 512 const persist = options?.persist ?? false + const sessions = new Map() + // Keep routing separate from the public event, and retain its snapshot + // while a slow subscriber drains events queued before a move or deletion. + const routes = new WeakMap() + + const isSessionEvent = (event: Event.Payload): event is SessionEvent.Event => + Object.hasOwn(SessionEvent.All.cases, event.type) + + const prepareRoutes = Effect.fnUntraced(function* (events: readonly Event.Payload[]) { + const updates = new Map() + const resolved = new Map() + for (const event of events) { + if (!isSessionEvent(event)) continue + const id = event.data.sessionID + if (event.type === "session.created") { + updates.set(id, event.data.location) + resolved.set(event, [event.location ?? event.data.location]) + continue + } + if (event.location && event.type !== "session.forked" && event.type !== "session.moved") { + if (event.type === "session.deleted") updates.set(id, undefined) + continue + } + const owner = event.type === "session.forked" ? event.data.parentID : id + let ref = updates.has(owner) ? updates.get(owner) : sessions.get(owner) + if (!ref && !updates.has(owner)) { + const row = yield* db + .select({ directory: SessionTable.directory, workspaceID: SessionTable.workspace_id }) + .from(SessionTable) + .where(eq(SessionTable.id, owner)) + .get() + .pipe(Effect.orDie) + ref = row + ? { directory: AbsolutePath.make(row.directory), workspaceID: row.workspaceID ?? undefined } + : undefined + updates.set(owner, ref) + } + if (event.type === "session.moved") { + // Both owners need the transition, even if the producer supplied + // an envelope location. Later events use only the destination. + updates.set(id, event.data.location) + resolved.set(event, ref ? [ref, event.data.location] : [event.data.location]) + continue + } + if (event.type === "session.forked") updates.set(id, ref) + resolved.set(event, event.location ? [event.location] : ref ? [ref] : []) + if (event.type === "session.deleted") updates.set(id, undefined) + } + // Apply only after the projection transaction commits. A failed move + // must not redirect events away from the Session's actual location. + return () => { + for (const [id, ref] of updates) { + if (ref) sessions.set(id, ref) + else sessions.delete(id) + } + for (const [event, ref] of resolved) routes.set(event, ref) + } + }) const getOrCreate = (definition: Event.Definition) => Effect.gen(function* () { @@ -335,6 +400,7 @@ export function configured(options?: Options) { ...event, durable: { aggregateID, seq, version: durable.version }, } as Event.Payload + const route = yield* prepareRoutes([committed]) for (const projector of list) { yield* projector(committed) } @@ -366,12 +432,13 @@ export function configured(options?: Options) { ]) .run() .pipe(Effect.orDie) - return { aggregateID, seq } + return { aggregateID, seq, event: committed, route } }), { behavior: "immediate" }, ) .pipe(Effect.orDie) if (committed) { + committed.route() yield* Effect.forEach( pubsub.durable.get(committed.aggregateID) ?? [], (wake) => PubSub.publish(wake, undefined), @@ -409,15 +476,14 @@ export function configured(options?: Options) { Effect.gen(function* () { const committed = yield* commitDurableEvent(definition, event as Event.Payload, undefined, commit) if (!committed) return event - event = { - ...event, - durable: envelope(committed.aggregateID, committed.seq, definition.durable.version), - } + event = committed.event as Event.Payload yield* notify(event as Event.Payload, true) return event }), ) } + const route = yield* prepareRoutes([event as Event.Payload]) + route() yield* notify(event as Event.Payload, false) return event }) @@ -527,7 +593,11 @@ export function configured(options?: Options) { .pipe(Effect.orDie) const firstSeq = (row?.seq ?? -1) + 1 const finalSeq = firstSeq + payloads.length - 1 - const result = new Array() + const queued = payloads.map((item, index) => ({ + ...item.event, + durable: envelope(aggregateID, firstSeq + index, item.definition.durable.version), + })) + const route = yield* prepareRoutes(queued) const rows = new Array() const ids = new Set() for (const [index, item] of payloads.entries()) { @@ -559,10 +629,7 @@ export function configured(options?: Options) { }), ) } - const event = { - ...item.event, - durable: envelope(aggregateID, seq, item.definition.durable.version), - } as Event.Payload + const event = queued[index] for (const projector of projectors.get( versionedType(item.definition.type, item.definition.durable.version), ) ?? []) { @@ -578,7 +645,6 @@ export function configured(options?: Options) { type: versionedType(item.definition.type, item.definition.durable.version), data: encoded, }) - result.push(event) } yield* db .insert(EventSequenceTable) @@ -587,11 +653,12 @@ export function configured(options?: Options) { .run() .pipe(Effect.orDie) if (persist) yield* db.insert(EventTable).values(rows).run().pipe(Effect.orDie) - return result + return { events: queued, route } }), { behavior: "immediate" }, ) .pipe(Effect.orDie) + committed.route() yield* Effect.forEach( pubsub.durable.get(aggregateID) ?? [], (wake) => PubSub.publish(wake, undefined), @@ -599,8 +666,8 @@ export function configured(options?: Options) { discard: true, }, ) - yield* Effect.forEach(committed, (event) => notify(event, true), { discard: true }) - return committed as PublishResult + yield* Effect.forEach(committed.events, (event) => notify(event, true), { discard: true }) + return committed.events as PublishResult }), ), ) @@ -632,13 +699,7 @@ export function configured(options?: Options) { strictOwner: options?.strictOwner, }) if (committed && options?.publish) { - yield* notify( - { - ...payload, - durable: envelope(committed.aggregateID, committed.seq, definition.durable.version), - }, - true, - ) + yield* notify(committed.event, true) } }), ) @@ -653,7 +714,10 @@ export function configured(options?: Options) { yield* db.delete(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).run() }), ) - .pipe(Effect.orDie) + .pipe( + Effect.tap(() => Effect.sync(() => sessions.delete(aggregateID as SessionID))), + Effect.orDie, + ) } function claim(aggregateID: string, ownerID: string) { @@ -671,15 +735,17 @@ export function configured(options?: Options) { Effect.map((location) => Option.match(location, { onNone: () => stream, - onSome: (location) => - stream.pipe( - Stream.filter( - (event) => - !event.location || - (event.location.directory === location.directory && - event.location.workspaceID === location.workspaceID), - ), - ), + onSome: (location) => { + const matches = (ref: Location.Ref) => + ref.directory === location.directory && ref.workspaceID === location.workspaceID + return stream.pipe( + Stream.filter((event) => { + const refs = routes.get(event) + if (refs) return refs.some(matches) + return !event.location || matches(event.location) + }), + ) + }, }), ), ), diff --git a/packages/core/test/bus-session-routing.test.ts b/packages/core/test/bus-session-routing.test.ts new file mode 100644 index 00000000000..c542d96f5a4 --- /dev/null +++ b/packages/core/test/bus-session-routing.test.ts @@ -0,0 +1,336 @@ +import { describe, expect } from "bun:test" +import { Deferred, Effect, Exit, Fiber, Stream } from "effect" +import { eq } from "drizzle-orm" +import { Bus } from "@opencode-ai/core/bus" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { Location } from "@opencode-ai/core/location" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { Event } from "@opencode-ai/schema/event" +import { Project } from "@opencode-ai/schema/project" +import { AbsolutePath } from "@opencode-ai/schema/schema" +import { SessionID } from "@opencode-ai/schema/session-id" +import { SessionEvent } from "@opencode-ai/schema/session-event" +import { SessionMessage } from "@opencode-ai/schema/session-message" +import { WorkspaceID } from "@opencode-ai/schema/workspace-id" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { location } from "./fixture/location" +import { testEffect } from "./lib/effect" + +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node]), [ + [Bus.node, Bus.configured({ persist: true })], + ]), +) +const a = Location.Ref.make({ directory: AbsolutePath.make("/a") }) +const b = Location.Ref.make({ directory: AbsolutePath.make("/b") }) +const otherWorkspace = Location.Ref.make({ directory: a.directory, workspaceID: WorkspaceID.make("wrk_other") }) +const id = SessionID.make("ses_routing") +const Done = Bus.ephemeral({ type: "test.routing.done", schema: {} }) +const Global = Bus.ephemeral({ type: "test.routing.global", schema: { sessionID: SessionID } }) + +const seed = Effect.fn(function* (ref: Location.Ref = a) { + const database = yield* Database.Service + yield* database.db.insert(ProjectTable).values({ id: Project.ID.global, worktree: a.directory, sandboxes: [] }).run() + yield* database.db + .insert(SessionTable) + .values({ + id, + project_id: Project.ID.global, + directory: ref.directory, + workspace_id: ref.workspaceID, + slug: "routing", + version: "test", + }) + .run() +}) + +const watch = (bus: Bus.Interface, ref?: Location.Ref, gate?: Deferred.Deferred) => { + const collect = bus.subscribe().pipe( + Stream.takeUntil((event) => event.type === Done.type), + Stream.mapEffect((event) => (gate ? Deferred.await(gate).pipe(Effect.as(event)) : Effect.succeed(event))), + Stream.runCollect, + ) + return (ref ? collect.pipe(Effect.provideService(Location.Service, location(ref))) : collect).pipe( + Effect.forkScoped({ startImmediately: true }), + ) +} + +const delta = (bus: Bus.Interface) => + bus.publish(SessionEvent.Text.Delta, { + sessionID: id, + assistantMessageID: SessionMessage.ID.make("msg_routing"), + ordinal: 0, + delta: "text", + }) + +describe("Bus Session routing", () => { + it.effect("delivers workspace-only moves to both owners without duplicating same-location moves", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + const first = yield* watch(bus, a) + const second = yield* watch(bus, otherWorkspace) + const moved = yield* bus.publish( + SessionEvent.Moved, + { sessionID: id, location: otherWorkspace, projectID: Project.ID.global }, + { location: a }, + ) + const after = yield* delta(bus) + const same = yield* bus.publish(SessionEvent.Moved, { + sessionID: id, + location: otherWorkspace, + projectID: Project.ID.global, + }) + const done = yield* bus.publish(Done, {}) + expect(Array.from(yield* Fiber.join(first))).toEqual([moved, done]) + expect(Array.from(yield* Fiber.join(second))).toEqual([moved, after, same, done]) + expect(moved.location).toEqual(a) + }), + ) + + it.effect("routes forks through their parent before the child exists", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + const database = yield* Database.Service + yield* bus.publish(SessionEvent.Synthetic, { sessionID: id, text: "Fork boundary" }) + const boundary = yield* database.db + .select({ id: SessionMessageTable.id }) + .from(SessionMessageTable) + .where(eq(SessionMessageTable.session_id, id)) + .get() + if (!boundary) return yield* Effect.die("Missing fork boundary") + + yield* Effect.forEach(["publish", "batch", "replay"] as const, (mode) => + Effect.gen(function* () { + const child = SessionID.create() + const first = yield* watch(bus, a) + const second = yield* watch(bus, b) + const payload = { + sessionID: child, + parentID: id, + boundary: { type: "before" as const, messageID: boundary.id }, + } + const eventID = Event.ID.create() + if (mode === "publish") yield* bus.publish(SessionEvent.Forked, payload, { id: eventID }) + if (mode === "batch") yield* bus.publishAll([[SessionEvent.Forked, payload, { id: eventID }]]) + if (mode === "replay") + yield* bus.replay( + { + id: eventID, + type: Bus.versionedType(SessionEvent.Forked.type, 2), + seq: 0, + aggregateID: child, + data: payload, + }, + { publish: true }, + ) + const after = yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID: child }) + const done = yield* bus.publish(Done, {}) + expect(Array.from(yield* Fiber.join(first)).map((event) => event.id)).toEqual([eventID, after.id, done.id]) + expect(Array.from(yield* Fiber.join(second))).toEqual([done]) + }), + ) + }), + ) + + it.effect("routes existing Sessions without changing public events or global delivery", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + const first = yield* watch(bus, a) + const second = yield* watch(bus, b) + const workspace = yield* watch(bus, otherWorkspace) + const global = yield* watch(bus) + const listened: Event.Payload[] = [] + yield* bus.listen((event) => + Effect.sync(() => { + listened.push(event) + }), + ) + + const renamed = yield* bus.publish(SessionEvent.Renamed, { sessionID: id, title: "first" }) + const text = yield* delta(bus) + const broadcast = yield* bus.publish(Global, { sessionID: id }) + const explicit = yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID: id }, { location: b }) + const done = yield* bus.publish(Done, {}) + + expect(Array.from(yield* Fiber.join(first))).toEqual([renamed, text, broadcast, done]) + expect(Array.from(yield* Fiber.join(second))).toEqual([broadcast, explicit, done]) + expect(Array.from(yield* Fiber.join(workspace))).toEqual([broadcast, done]) + expect(Array.from(yield* Fiber.join(global))).toEqual([renamed, text, broadcast, explicit, done]) + expect(listened).toEqual([renamed, text, broadcast, explicit, done]) + expect(renamed).not.toHaveProperty("location") + expect(text).not.toHaveProperty("location") + expect(JSON.parse(JSON.stringify(renamed))).not.toHaveProperty("location") + const history = yield* bus.log({ aggregateID: id }).pipe(Stream.runCollect) + expect( + Array.from(history) + .filter((event): event is Event.Payload => !Bus.isSynced(event)) + .every((event) => !event.location), + ).toBe(true) + }), + ) + + it.effect("applies the same routing to typed and multi-type subscriptions", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + const typed = yield* bus + .subscribe(SessionEvent.Renamed) + .pipe( + Stream.take(1), + Stream.runCollect, + Effect.provideService(Location.Service, location(b)), + Effect.forkScoped({ startImmediately: true }), + ) + const multiple = yield* bus.subscribe([SessionEvent.Renamed, Done]).pipe( + Stream.takeUntil((event) => event.type === Done.type), + Stream.runCollect, + Effect.provideService(Location.Service, location(b)), + Effect.forkScoped({ startImmediately: true }), + ) + yield* bus.publish(SessionEvent.Renamed, { sessionID: id, title: "wrong location" }) + yield* bus.publish(SessionEvent.Moved, { sessionID: id, location: b, projectID: Project.ID.global }) + const expected = yield* bus.publish(SessionEvent.Renamed, { sessionID: id, title: "destination" }) + const done = yield* bus.publish(Done, {}) + expect(Array.from(yield* Fiber.join(typed))).toEqual([expected]) + expect(Array.from(yield* Fiber.join(multiple))).toEqual([expected, done]) + }), + ) + + it.effect("snapshots routing across creation and moves for slow subscribers", () => + Effect.gen(function* () { + const database = yield* Database.Service + yield* database.db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: a.directory, sandboxes: [] }) + .run() + const bus = yield* Bus.Service + const gate = yield* Deferred.make() + const first = yield* watch(bus, a, gate) + const second = yield* watch(bus, b, gate) + const created = yield* bus.publish(SessionEvent.Created, { + sessionID: id, + location: a, + projectID: Project.ID.global, + slug: "routing", + version: "test", + }) + const before = yield* bus.publish(SessionEvent.Renamed, { sessionID: id, title: "before" }) + const moved = yield* bus.publish(SessionEvent.Moved, { sessionID: id, location: b, projectID: Project.ID.global }) + const after = yield* delta(bus) + const done = yield* bus.publish(Done, {}) + yield* Deferred.succeed(gate, undefined) + + expect(Array.from(yield* Fiber.join(first))).toEqual([created, before, moved, done]) + expect(Array.from(yield* Fiber.join(second))).toEqual([moved, after, done]) + expect(moved).not.toHaveProperty("location") + }), + ) + + it.effect("routes a cold Session deletion before its projector removes ownership", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + const first = yield* watch(bus, a) + const second = yield* watch(bus, b) + const global = yield* watch(bus) + const deleted = yield* bus.publish(SessionEvent.Deleted, { sessionID: id }) + const missing = yield* delta(bus) + const done = yield* bus.publish(Done, {}) + + const database = yield* Database.Service + expect(yield* database.db.select().from(SessionTable).where(eq(SessionTable.id, id)).get()).toBeUndefined() + expect(Array.from(yield* Fiber.join(first))).toEqual([deleted, done]) + expect(Array.from(yield* Fiber.join(second))).toEqual([done]) + expect(Array.from(yield* Fiber.join(global))).toEqual([deleted, missing, done]) + }), + ) + + it.effect("preserves routing through a batch that moves and deletes a Session", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + const gate = yield* Deferred.make() + const first = yield* watch(bus, a, gate) + const second = yield* watch(bus, b, gate) + const events = yield* bus.publishAll([ + [SessionEvent.Renamed, { sessionID: id, title: "before" }], + [SessionEvent.Moved, { sessionID: id, location: b, projectID: Project.ID.global }], + [SessionEvent.Renamed, { sessionID: id, title: "after" }], + [SessionEvent.Deleted, { sessionID: id }], + ]) + const done = yield* bus.publish(Done, {}) + yield* Deferred.succeed(gate, undefined) + expect(Array.from(yield* Fiber.join(first))).toEqual([events[0], events[1], done]) + expect(Array.from(yield* Fiber.join(second))).toEqual([events[1], events[2], events[3], done]) + }), + ) + + it.effect("does not change ownership when single or batched moves roll back", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + const first = yield* watch(bus, a) + const second = yield* watch(bus, b) + const before = yield* delta(bus) + const single = yield* bus + .publish( + SessionEvent.Moved, + { sessionID: id, location: b, projectID: Project.ID.global }, + { commit: () => Effect.die("rollback") }, + ) + .pipe(Effect.exit) + const batch = yield* bus + .publishAll([ + [SessionEvent.Moved, { sessionID: id, location: b, projectID: Project.ID.global }], + [SessionEvent.Renamed, { sessionID: id, title: "rollback" }, { commit: () => Effect.die("rollback") }], + ]) + .pipe(Effect.exit) + const after = yield* delta(bus) + const done = yield* bus.publish(Done, {}) + expect(Exit.isFailure(single)).toBe(true) + expect(Exit.isFailure(batch)).toBe(true) + expect(Array.from(yield* Fiber.join(first))).toEqual([before, after, done]) + expect(Array.from(yield* Fiber.join(second))).toEqual([done]) + }), + ) + + it.effect("updates cached ownership on silent replay and filters published replay", () => + Effect.gen(function* () { + yield* seed() + const bus = yield* Bus.Service + yield* delta(bus) + const first = yield* watch(bus, a) + const second = yield* watch(bus, b) + yield* bus.replay({ + id: Event.ID.create(), + type: Bus.versionedType(SessionEvent.Moved.type, 1), + seq: 0, + aggregateID: id, + data: { sessionID: id, location: b, projectID: Project.ID.global }, + }) + const after = yield* delta(bus) + const replayID = Event.ID.create() + yield* bus.replay( + { + id: replayID, + type: Bus.versionedType(SessionEvent.Renamed.type, 1), + seq: 1, + aggregateID: id, + data: { sessionID: id, title: "replayed" }, + }, + { publish: true }, + ) + const done = yield* bus.publish(Done, {}) + expect(Array.from(yield* Fiber.join(first))).toEqual([done]) + const received = Array.from(yield* Fiber.join(second)) + expect(received.map((event) => event.id)).toEqual([after.id, replayID, done.id]) + expect(received[1]).not.toHaveProperty("location") + }), + ) +}) diff --git a/packages/core/test/location-layer.test.ts b/packages/core/test/location-layer.test.ts index 27eda53455e..fdaadb05426 100644 --- a/packages/core/test/location-layer.test.ts +++ b/packages/core/test/location-layer.test.ts @@ -81,6 +81,28 @@ const itWithActivity = testEffect( ) describe("LocationServiceMap", () => { + itWithActivity.effect("does not refresh lifetime from inferred Session routing", () => + Effect.gen(function* () { + const locations = yield* LocationServiceMap.Service + const bus = yield* Bus.Service + const ref = Location.Ref.make({ directory: AbsolutePath.make("/project") }) + const sessionID = Session.ID.make("ses_routing_activity") + yield* Location.Service.pipe(Effect.provide(locations.get(ref)), Effect.scoped) + yield* bus.publish(SessionEvent.Created, { + sessionID, + location: ref, + projectID: Project.ID.global, + slug: "routing", + version: "test", + }) + yield* TestClock.adjust("59 minutes") + const event = yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID }) + expect(event).not.toHaveProperty("location") + yield* TestClock.adjust("2 minutes") + expect(Array.from(yield* RcMap.keys(locations.rcMap))).toEqual([]) + }), + ) + itWithActivity.effect("refreshes lifetime from Session events only", () => Effect.gen(function* () { const locations = yield* LocationServiceMap.Service