fix(core): route session events to location subscribers (#45411)

Co-authored-by: thdxr <826656+thdxr@users.noreply.github.com>

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<D extends Event.Definition = Event.Definition> = (event: Event.Payload<D>) => Effect.Effect<void>
 export type Unsubscribe = Effect.Effect<void>
@@ -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<Event.Payload>
   <D extends Event.Definition>(definition: D): Stream.Stream<Event.Payload<D>>
@@ -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<Event.Payload>(),
           durable: new Map<string, Set<PubSub.PubSub<void>>>(),
@@ -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<SessionID, Location.Ref>()
+        // 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<Event.Payload, readonly Location.Ref[]>()
+
+        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<SessionID, Location.Ref | undefined>()
+          const resolved = new Map<Event.Payload, readonly Location.Ref[]>()
+          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<D>
                   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<Event.Payload>()
+                          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<typeof EventTable.$inferInsert>()
                           const ids = new Set<Event.ID>()
                           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<I>
+                  yield* Effect.forEach(committed.events, (event) => notify(event, true), { discard: true })
+                  return committed.events as PublishResult<I>
                 }),
               ),
             )
@@ -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)
+                      }),
+                    )
+                  },
                 }),
               ),
             ),

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<void>) => {
+  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<void>()
+      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<void>()
+      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")
+    }),
+  )
+})

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