diff --git a/CONTEXT.md b/CONTEXT.md index 2a0bc8df920..ae25c055ad3 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -61,6 +61,7 @@ The point immediately before a provider call, after durable input promotion and - Nested project instruction discovery after successful reads remains a follow-up; when implemented, discovered instructions must be admitted durably at the next **Safe Provider-Turn Boundary**. - Location-scoped services naturally re-resolve effective context when a moved session next runs in its destination location. - Moving a Session clears its active **Context Epoch**, so the destination must initialize a complete baseline before another prompt can promote. +- Context Epoch initialization is fenced against the authoritative Session Location, so an old-Location runner cannot recreate source context after a concurrent move. - Instruction discovery, source identity, persistence, and file loading belong to the instruction service; the **System Context** abstraction only composes effectful producers and renders loaded values. - The first instruction-service slice observes global and upward project `AGENTS.md` files as one ordered aggregate **Context Source** at each **Safe Provider-Turn Boundary**. - Built-in and instruction context producers register through the **System Context Registry** with stable contribution keys. Plugin-defined context registration and hot-reload lifecycle remain a follow-up built on the same scoped registry seam. diff --git a/packages/core/src/session/context-epoch.ts b/packages/core/src/session/context-epoch.ts index 6781bd533d5..22ce8d5d40f 100644 --- a/packages/core/src/session/context-epoch.ts +++ b/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 = (attempt: () => Effect.Effect): Effect.Effect => attempt().pipe( @@ -34,8 +36,9 @@ export function initialize( db: DatabaseService, context: SystemContextRegistry.Interface, sessionID: SessionSchema.ID, + location: Location.Ref, ): Effect.Effect { - 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 { - 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) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index b628861033a..0f1880205de 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/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() 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, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index fe60aece822..160884402ed 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -631,6 +631,39 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("does not create a source Location epoch after a concurrent Session move", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + const { db } = yield* Database.Service + let moved = false + systemLoadHook = Effect.suspend(() => { + if (moved) return Effect.void + moved = true + return events + .publish(SessionEvent.Moved, { + sessionID, + timestamp: DateTime.makeUnsafe(1), + location: { directory: AbsolutePath.make("/moved") }, + }) + .pipe(Effect.asVoid) + }) + yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false }) + + expect(Exit.isFailure(yield* session.resume(sessionID).pipe(Effect.exit))).toBe(true) + expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true) + expect( + yield* db + .select() + .from(SessionContextEpochTable) + .where(eq(SessionContextEpochTable.session_id, sessionID)) + .get(), + ).toBeUndefined() + expect((yield* session.get(sessionID)).location.directory).toBe(AbsolutePath.make("/moved")) + }), + ) + it.effect("reuses one durable baseline after the context producer changes", () => Effect.gen(function* () { yield* setup diff --git a/packages/llm/src/protocols/anthropic-messages.ts b/packages/llm/src/protocols/anthropic-messages.ts index 38f1f19babb..d543e659ca6 100644 --- a/packages/llm/src/protocols/anthropic-messages.ts +++ b/packages/llm/src/protocols/anthropic-messages.ts @@ -363,6 +363,18 @@ const canUseNativeSystemUpdate = (messages: LLMRequest["messages"], index: numbe ) } +const splitsLocalToolResults = (messages: LLMRequest["messages"], index: number) => { + const pending = new Set() + for (const message of messages.slice(0, index)) { + for (const part of message.content) { + if (message.role === "assistant" && part.type === "tool-call" && part.providerExecuted !== true) + pending.add(part.id) + if (message.role === "tool" && part.type === "tool-result") pending.delete(part.id) + } + } + return pending.size > 0 +} + const lowerNativeSystemUpdate = Effect.fn("AnthropicMessages.lowerNativeSystemUpdate")(function* ( message: LLMRequest["messages"][number], breakpoints: Cache.Breakpoints, @@ -386,6 +398,8 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* ( for (const [index, message] of request.messages.entries()) { if (message.role === "system") { + if (splitsLocalToolResults(request.messages, index)) + return yield* invalid("Anthropic Messages system updates cannot split a local tool call from its tool result") if (supportsNativeSystemUpdates(request) && canUseNativeSystemUpdate(request.messages, index)) { messages.push(yield* lowerNativeSystemUpdate(message, breakpoints)) continue diff --git a/packages/llm/test/provider/anthropic-messages.test.ts b/packages/llm/test/provider/anthropic-messages.test.ts index 3fb3cdb3b91..1cd8f4dd9e1 100644 --- a/packages/llm/test/provider/anthropic-messages.test.ts +++ b/packages/llm/test/provider/anthropic-messages.test.ts @@ -168,6 +168,25 @@ describe("Anthropic Messages route", () => { }), ) + it.effect("rejects a system update between a local tool call and its result", () => + Effect.gen(function* () { + const error = yield* LLMClient.prepare( + LLM.request({ + model: opus48, + messages: [ + Message.user("Use the tool."), + Message.assistant([ToolCallPart.make({ id: "call_1", name: "lookup", input: {} })]), + Message.system("Too early."), + Message.tool({ id: "call_1", name: "lookup", result: "Done." }), + ], + cache: "none", + }), + ).pipe(Effect.flip) + + expect(error.message).toContain("system updates cannot split a local tool call from its tool result") + }), + ) + it.effect("prepares tool call and tool result messages", () => Effect.gen(function* () { const prepared = yield* LLMClient.prepare( diff --git a/specs/v2/schema-changelog.md b/specs/v2/schema-changelog.md index 4f19607d49b..5c5980a2ede 100644 --- a/specs/v2/schema-changelog.md +++ b/specs/v2/schema-changelog.md @@ -762,6 +762,7 @@ Change: - Preserve admitted instructions across transient scan/read failures and block first-epoch initialization while any context source is unavailable. - Retry Context Epoch preparation until stable after optimistic revision mismatches. - Clear the active Context Epoch when a Session moves so the destination initializes a complete baseline before promoting more input. +- Fence Context Epoch initialization against the authoritative Session Location so a concurrent old-Location runner cannot recreate stale privileged context after a move. - Canonicalize ambient instruction traversal boundaries, honor `OPENCODE_DISABLE_PROJECT_CONFIG`, and make non-empty aggregate updates explicitly supersede previously loaded instructions. Compatibility: