fix(core): fence v2 context placement

packages/core/src/session/context-epoch.ts

@@ -4,17 +4,19 @@ import { and, eq, isNull, lt, or, sql } from "drizzle-orm"
 import { DateTime, Effect, Schema } from "effect"
 import type { Database } from "../database/database"
 import { EventV2 } from "../event"
+import { Location } from "../location"
 import { SystemContext } from "../system-context"
 import { SystemContextRegistry } from "../system-context-registry"
 import { SessionEvent } from "./event"
 import { SessionInput } from "./input"
 import { SessionMessageID } from "./message-id"
 import { SessionSchema } from "./schema"
-import { SessionContextEpochTable } from "./sql"
+import { SessionContextEpochTable, SessionTable } from "./sql"
 
 type DatabaseService = Database.Interface["db"]
 
 class RevisionMismatch extends Error {}
+class LocationMismatch extends Error {}
 
 const retryRevisionMismatch = <A, E>(attempt: () => Effect.Effect<A, E>): Effect.Effect<A, E> =>
   attempt().pipe(
@@ -34,8 +36,9 @@ export function initialize(
   db: DatabaseService,
   context: SystemContextRegistry.Interface,
   sessionID: SessionSchema.ID,
+  location: Location.Ref,
 ): Effect.Effect<Prepared | undefined, SystemContext.InitializationBlocked> {
-  return retryRevisionMismatch(() => initializeOnce(db, context, sessionID)).pipe(
+  return retryRevisionMismatch(() => initializeOnce(db, context, sessionID, location)).pipe(
     Effect.withSpan("SessionContextEpoch.initialize"),
   )
 }
@@ -45,8 +48,9 @@ export function prepare(
   events: EventV2.Interface,
   context: SystemContextRegistry.Interface,
   sessionID: SessionSchema.ID,
+  location: Location.Ref,
 ): Effect.Effect<Prepared, SystemContext.InitializationBlocked> {
-  return retryRevisionMismatch(() => prepareOnce(db, events, context, sessionID)).pipe(
+  return retryRevisionMismatch(() => prepareOnce(db, events, context, sessionID, location)).pipe(
     Effect.withSpan("SessionContextEpoch.prepare"),
   )
 }
@@ -56,11 +60,12 @@ const prepareOnce = Effect.fnUntraced(function* (
   events: EventV2.Interface,
   context: SystemContextRegistry.Interface,
   sessionID: SessionSchema.ID,
+  location: Location.Ref,
 ) {
   const [value, stored] = yield* Effect.all([context.load(), find(db, sessionID)], { concurrency: "unbounded" })
   if (!stored) {
     const generation = yield* SystemContext.initialize(value)
-    const baselineSeq = yield* insert(db, sessionID, generation)
+    const baselineSeq = yield* insert(db, sessionID, location, generation)
     return { baseline: generation.baseline, baselineSeq }
   }
 
@@ -89,10 +94,11 @@ const initializeOnce = Effect.fnUntraced(function* (
   db: DatabaseService,
   context: SystemContextRegistry.Interface,
   sessionID: SessionSchema.ID,
+  location: Location.Ref,
 ) {
   if (yield* exists(db, sessionID)) return
   const generation = yield* context.load().pipe(Effect.flatMap(SystemContext.initialize))
-  const baselineSeq = yield* insert(db, sessionID, generation)
+  const baselineSeq = yield* insert(db, sessionID, location, generation)
   return { baseline: generation.baseline, baselineSeq }
 })
 
@@ -142,12 +148,28 @@ export const reset = Effect.fn("SessionContextEpoch.reset")(function* (db: Datab
 const insert = Effect.fnUntraced(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
+  location: Location.Ref,
   generation: SystemContext.Generation,
 ) {
   return yield* db
     .transaction(
       () =>
         Effect.gen(function* () {
+          const placed = yield* db
+            .select({ sessionID: SessionTable.id })
+            .from(SessionTable)
+            .where(
+              and(
+                eq(SessionTable.id, sessionID),
+                eq(SessionTable.directory, location.directory),
+                location.workspaceID === undefined
+                  ? isNull(SessionTable.workspace_id)
+                  : eq(SessionTable.workspace_id, location.workspaceID),
+              ),
+            )
+            .get()
+            .pipe(Effect.orDie)
+          if (!placed) return yield* Effect.die(new LocationMismatch())
           const baselineSeq = yield* SessionInput.latestSeq(db, sessionID)
           yield* db
             .insert(SessionContextEpochTable)

packages/core/src/session/runner/llm.ts

@@ -132,7 +132,7 @@ export const layer = Layer.effect(
       promotion: "steer" | "queue" | undefined,
     ) {
       const session = yield* getSession(sessionID)
-      const initialized = yield* SessionContextEpoch.initialize(db, systemContext, session.id)
+      const initialized = yield* SessionContextEpoch.initialize(db, systemContext, session.id, session.location)
       const model = yield* models.resolve(session)
       const toolFibers = yield* FiberSet.make<void, never>()
       let needsContinuation = false
@@ -144,7 +144,7 @@ export const layer = Layer.effect(
           yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
         }
       }
-      const system = initialized ?? (yield* SessionContextEpoch.prepare(db, events, systemContext, session.id))
+      const system = initialized ?? (yield* SessionContextEpoch.prepare(db, events, systemContext, session.id, session.location))
       const context = yield* store.runnerContext(session.id, system.baselineSeq)
       const request = LLM.request({
         model,