From 9ba3a54d83c0fc4f8317ae605bd6a6a528213396 Mon Sep 17 00:00:00 2001 From: Kameron Smith <112618179+nullStack65@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:21:36 -0400 Subject: [PATCH] fix(acp): support root session replacement in V2 An agent can replace the root session behind the same ACP connection: omp's /fresh starts a new provider session and publishes every later session/update under a new id while the original id stays the one session/load replays. The V2 runtime treated those updates as a foreign child session, so the thread went silent for the rest of the turn. Add a generic, default-off runtime capability (AcpSessionRuntimeOptions.adoptRootSessionReplacement) that adopts a new live root session id first seen while a root prompt is in flight, at most once per prompt, and projects adopted notifications back onto the durable setup id. Prompts, cancellation, session loading, and item identity keep using the durable session; foreign ids seen while idle stay rejected. The ACP Registry adapter exposes the opt-in per instance through AcpRegistrySettings. The frozen bespoke reference is PR #11973; this slice carries only the provider-neutral adoption mechanism. --- apps/server/scripts/acp-mock-agent.ts | 76 +++++ .../Adapters/AcpRegistryAdapterV2.test.ts | 145 +++++++++- .../Adapters/AcpRegistryAdapterV2.ts | 4 + .../provider/acp/AcpJsonRpcConnection.test.ts | 270 ++++++++++++++++++ .../src/provider/acp/AcpSessionRuntime.ts | 108 ++++++- packages/contracts/src/settings.ts | 11 +- 6 files changed, 609 insertions(+), 5 deletions(-) diff --git a/apps/server/scripts/acp-mock-agent.ts b/apps/server/scripts/acp-mock-agent.ts index 569ad7278c37..1bb73eb39a4b 100644 --- a/apps/server/scripts/acp-mock-agent.ts +++ b/apps/server/scripts/acp-mock-agent.ts @@ -46,6 +46,19 @@ const emitActiveToolThenHang = process.env.T3_ACP_EMIT_ACTIVE_TOOL_THEN_HANG === const emitGrokMonitorPostTurnPoll = process.env.T3_ACP_EMIT_GROK_MONITOR_POST_TURN_POLL === "1"; const emitGrokBackgroundTaskStarted = process.env.T3_ACP_EMIT_GROK_BACKGROUND_TASK_STARTED === "1"; const emitForeignSessionUpdates = process.env.T3_ACP_EMIT_FOREIGN_SESSION_UPDATES === "1"; +// Root-session replacement: emit the first chunk under the requested session id +// and then publish every later update (including completion) under a fresh id, +// mirroring an agent-side `/fresh` provider session swap. The fresh id is not +// addressable: prompts and cancellation keep targeting the requested id. +const emitRootSessionReplacement = process.env.T3_ACP_ROOT_SESSION_REPLACEMENT === "1"; +const emitStaleRootAfterReplacement = process.env.T3_ACP_STALE_ROOT_AFTER_REPLACEMENT === "1"; +const emitIdleForeignSession = process.env.T3_ACP_IDLE_FOREIGN_SESSION === "1"; +const hangRootSessionReplacement = process.env.T3_ACP_ROOT_SESSION_REPLACEMENT_HANG === "1"; +// Completes the prompt on the requested session id while content published +// under the replacement id. Lets a strict-mode client settle the turn without +// the replacement completion. +const completeDurableAfterRootReplacement = + process.env.T3_ACP_ROOT_SESSION_REPLACEMENT_COMPLETE_DURABLE === "1"; const waitForResumeRelease = process.env.T3_ACP_WAIT_FOR_RESUME_RELEASE === "1"; const completeFirstPromptOnCancel = process.env.T3_ACP_COMPLETE_FIRST_PROMPT_ON_CANCEL === "1"; const floodStderr = process.env.T3_ACP_FLOOD_STDERR === "1"; @@ -122,6 +135,8 @@ let currentFast = false; let authenticated = !requiresAuthentication; let promptCount = 0; let overlappingFirstPromptId: string | undefined; +let liveReplacementSessionId: string | undefined; +let previousReplacementSessionId: string | undefined; const cancelledSessions = new Set(); let configuredProvider: AcpSchema.ProviderCurrentConfig | null = null; @@ -981,6 +996,67 @@ const program = Effect.gen(function* () { return yield* finishPrompt(requestedSessionId, "end_turn"); } + if (emitRootSessionReplacement) { + previousReplacementSessionId = liveReplacementSessionId; + liveReplacementSessionId = `mock-session-fresh-${promptCount}`; + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "agent_message_chunk", + messageId: "mock-agent-message", + content: { type: "text", text: "root before fresh" }, + }, + }); + yield* agent.client.sessionUpdate({ + sessionId: liveReplacementSessionId, + update: { + sessionUpdate: "agent_message_chunk", + messageId: "mock-agent-message", + content: { type: "text", text: "replaced live root" }, + }, + }); + if (emitStaleRootAfterReplacement) { + // Stragglers from replaced identities: the durable id stays accepted + // as root; a previously adopted live id is stale. + yield* agent.client.sessionUpdate({ + sessionId: requestedSessionId, + update: { + sessionUpdate: "agent_message_chunk", + messageId: "mock-agent-message", + content: { type: "text", text: "durable root straggler" }, + }, + }); + if (previousReplacementSessionId !== undefined) { + yield* agent.client.sessionUpdate({ + sessionId: previousReplacementSessionId, + update: { + sessionUpdate: "agent_message_chunk", + messageId: "mock-agent-message", + content: { type: "text", text: "stale replaced root" }, + }, + }); + } + } + if (hangRootSessionReplacement) { + return yield* Effect.never; + } + yield* finishPrompt( + completeDurableAfterRootReplacement ? requestedSessionId : liveReplacementSessionId, + "end_turn", + ); + if (emitIdleForeignSession) { + yield* agent.client.sessionUpdate({ + sessionId: "mock-session-idle-foreign", + update: { + sessionUpdate: "agent_message_chunk", + messageId: "mock-agent-message", + content: { type: "text", text: "foreign while idle" }, + }, + }); + } + return {}; + } + if (residualCallbackTriggerPath !== undefined) { yield* Effect.gen(function* () { while (!(yield* Effect.sync(() => NodeFS.existsSync(residualCallbackTriggerPath)))) { diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index 067a2de922cd..af5259ff44a5 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -1,6 +1,16 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, it } from "@effect/vitest"; -import { ProviderInstanceId, ProviderSessionId, ThreadId } from "@t3tools/contracts"; +import { + MessageId, + NodeId, + ProjectId, + ProviderInstanceId, + ProviderSessionId, + RunAttemptId, + RunId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Crypto from "effect/Crypto"; @@ -9,6 +19,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import { HttpClient, HttpClientResponse } from "effect/unstable/http"; import { ChildProcessSpawner } from "effect/unstable/process"; @@ -57,6 +68,11 @@ const registryLayer = Layer.succeed( cmd: "fixture-agent", args: [], }, + "darwin-x86_64": { + archive: "https://registry.test/unused", + cmd: "fixture-agent", + args: [], + }, "linux-x86_64": { archive: "https://registry.test/unused", cmd: "fixture-agent", @@ -90,6 +106,7 @@ describe("AcpRegistryAdapterV2", () => { authMethodId: "", distribution: "auto", customModels: [], + rootSessionReplacement: false, }); }); @@ -236,4 +253,130 @@ describe("AcpRegistryAdapterV2", () => { }); }).pipe(Effect.provide(testLayer), Effect.scoped), ); + + it.effect("adopts a replaced root session only when the instance opts in", () => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fileSystem = yield* FileSystem.FileSystem; + const idAllocator = yield* IdAllocatorV2; + const path = yield* Path.Path; + const serverConfig = yield* ServerConfig; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const resolver = yield* makeAcpRegistryCatalog({ + cacheDir: serverConfig.providerStatusCacheDir, + registryUrl, + }); + const settings = yield* decodeAcpRegistryAdapterSettings({ + agentId: "fixture-agent", + commandPath: process.execPath, + authMethodId: "test", + rootSessionReplacement: true, + }); + const instanceId = ProviderInstanceId.make("acp-registry-root-replacement"); + const adapter = makeAcpRegistryAdapterV2({ + crypto: yield* Crypto.Crypto, + instanceId, + settings, + environment: { + T3_ACP_SESSION_LIFECYCLE: "1", + T3_ACP_ROOT_SESSION_REPLACEMENT: "1", + }, + childProcessSpawner, + fileSystem, + idAllocator, + resolver: { + resolve: (configuredSettings, cwd, environment) => + resolver.resolve(configuredSettings, cwd, environment).pipe( + Effect.map((resolved) => ({ + ...resolved, + spawn: { ...resolved.spawn, args: [mockAgentPath] }, + })), + ), + }, + serverConfig, + }); + const threadId = ThreadId.make("thread-acp-registry-root-replacement"); + const runtimePolicy = ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-acp-root-replacement"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + const runId = RunId.make(`run:${threadId}:1`); + yield* runtime.startTurn({ + appThread: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make(`project:${threadId}`), + title: "ACP registry root replacement", + providerInstanceId: instanceId, + modelSelection, + runtimeMode: "approval-required", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: providerThread.id, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: RunAttemptId.make(`attempt:${threadId}:1`), + rootNodeId: NodeId.make(`node:${threadId}:1`), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: MessageId.make(`message:${threadId}:1`), + text: "hi", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + const events = Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ); + const assistantText = events + .flatMap((event) => (event.type === "turn_item.updated" ? [event.turnItem] : [])) + .flatMap((item) => + item.type === "assistant_message" && item.threadId === threadId ? [item.text] : [], + ) + .join(""); + assert.include(assistantText, "replaced live root"); + // The durable native thread id still addresses session/load. + assert.equal(providerThread.nativeThreadRef?.nativeId, "mock-session-1"); + }).pipe(Effect.provide(testLayer), Effect.scoped), + ); }); diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.ts index 422073222820..437abf5b8a23 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.ts @@ -102,6 +102,10 @@ function makeAcpRegistryRuntime(options: AcpRegistryAdapterV2Options) { ...resolved.spawn, env: { ...resolved.spawn.env, ...processEnvironment }, }, + // Generic per-instance opt-in: the operator declares that this agent + // replaces its root session on the same connection (see + // AcpSessionRuntimeOptions.adoptRootSessionReplacement). + ...(options.settings.rootSessionReplacement ? { adoptRootSessionReplacement: true } : {}), ...(options.settings.authMethodId ? { authMethodId: options.settings.authMethodId } : {}), }).pipe( Layer.provide( diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index 3a28294e8092..729ae361a3a7 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -940,6 +940,276 @@ describe("AcpSessionRuntime", () => { ), ); + describe("root session replacement", () => { + const contentDeltaText = (events: ReadonlyArray) => + events + .filter((event) => event._tag === "ContentDelta") + .map((event) => event.text) + .join(""); + + const collectEvents = (runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]) => + Effect.gen(function* () { + const events: Array = []; + yield* runtime.getEvents().pipe( + Stream.runForEach((event) => { + if (event._tag === "EventStreamBarrier") { + return Deferred.succeed(event.acknowledge, undefined); + } + events.push(event); + return Effect.void; + }), + Effect.forkChild, + ); + return events; + }); + + const replacementLayer = (input?: { + readonly env?: Readonly>; + readonly options?: Partial; + }) => + AcpSessionRuntime.layer({ + spawn: { + command: mockAgentCommand, + args: mockAgentArgs, + env: { T3_ACP_ROOT_SESSION_REPLACEMENT: "1", ...input?.env }, + }, + cwd: process.cwd(), + clientInfo: { name: "t3-test", version: "0.0.0" }, + authMethodId: "test", + ...input?.options, + }); + + it.effect("keeps a replaced root session rejected without the opt-in", () => + Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; + yield* runtime.start(); + const events = yield* collectEvents(runtime); + + const promptResult = yield* runtime.prompt({ + prompt: [{ type: "text", text: "hi" }], + }); + yield* runtime.drainEvents; + + expect(promptResult).toMatchObject({ stopReason: "end_turn" }); + expect(contentDeltaText(events)).toBe("root before fresh"); + }).pipe( + Effect.provide( + replacementLayer({ + env: { T3_ACP_ROOT_SESSION_REPLACEMENT_COMPLETE_DURABLE: "1" }, + }), + ), + Effect.scoped, + Effect.provide(NodeServices.layer), + ), + ); + + it.effect("adopts a replacement first seen during a root prompt", () => { + const replacements: Array<{ previousSessionId: string; sessionId: string }> = []; + return Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; + yield* runtime.start(); + const events = yield* collectEvents(runtime); + + const promptResult = yield* runtime.prompt({ + prompt: [{ type: "text", text: "hi" }], + }); + yield* runtime.drainEvents; + + expect(promptResult).toMatchObject({ stopReason: "end_turn" }); + expect(contentDeltaText(events)).toBe("root before freshreplaced live root"); + expect(replacements).toEqual([ + { previousSessionId: "mock-session-1", sessionId: "mock-session-fresh-1" }, + ]); + // Adopted updates project onto the durable identity, so assistant item + // ids keep one root session namespace. + const started = events.find((event) => event._tag === "AssistantItemStarted"); + expect(started?._tag === "AssistantItemStarted" ? started.itemId : "").toContain( + "assistant:mock-session-1:", + ); + }).pipe( + Effect.provide( + replacementLayer({ + options: { + adoptRootSessionReplacement: true, + onRootSessionReplaced: (change) => { + replacements.push(change); + }, + }, + }), + ), + Effect.scoped, + Effect.provide(NodeServices.layer), + ); + }); + + it.effect("adopts successive replacements deterministically", () => { + const replacements: Array<{ previousSessionId: string; sessionId: string }> = []; + return Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; + yield* runtime.start(); + const events = yield* collectEvents(runtime); + + yield* runtime.prompt({ prompt: [{ type: "text", text: "first" }] }); + yield* runtime.drainEvents; + yield* runtime.prompt({ prompt: [{ type: "text", text: "second" }] }); + yield* runtime.drainEvents; + + expect(replacements).toEqual([ + { previousSessionId: "mock-session-1", sessionId: "mock-session-fresh-1" }, + { previousSessionId: "mock-session-fresh-1", sessionId: "mock-session-fresh-2" }, + ]); + expect(contentDeltaText(events)).toBe( + "root before freshreplaced live rootroot before freshreplaced live root", + ); + }).pipe( + Effect.provide( + replacementLayer({ + options: { + adoptRootSessionReplacement: true, + onRootSessionReplaced: (change) => { + replacements.push(change); + }, + }, + }), + ), + Effect.scoped, + Effect.provide(NodeServices.layer), + ); + }); + + it.effect("keeps prompt and cancel requests on the durable session id", () => { + const promptSessionIds: Array = []; + const protocolEvents: Array = []; + return Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; + yield* runtime.start(); + + const replacedDelta = yield* runtime.getEvents().pipe( + Stream.filter( + (event) => event._tag === "ContentDelta" && event.text === "replaced live root", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild({ startImmediately: true }), + ); + const hangingPrompt = yield* runtime + .prompt({ prompt: [{ type: "text", text: "hi" }] }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Fiber.join(replacedDelta); + yield* runtime.cancel; + const promptResult = yield* Fiber.join(hangingPrompt); + + expect(promptResult).toMatchObject({ stopReason: "cancelled" }); + expect([...new Set(promptSessionIds)]).toEqual(["mock-session-1"]); + expect( + protocolEvents.some( + (event) => + event.direction === "outgoing" && + event.stage === "raw" && + typeof event.payload === "string" && + event.payload.includes('"method":"session/cancel"') && + event.payload.includes('"sessionId":"mock-session-1"'), + ), + ).toBe(true); + expect( + protocolEvents.some( + (event) => + typeof event.payload === "string" && + event.payload.includes('"method":"session/cancel"') && + event.payload.includes("mock-session-fresh-1"), + ), + ).toBe(false); + }).pipe( + Effect.provide( + replacementLayer({ + env: { T3_ACP_ROOT_SESSION_REPLACEMENT_HANG: "1" }, + options: { + adoptRootSessionReplacement: true, + requestLogger: (event) => { + if (event.method === "session/prompt") { + const payload = event.payload as { sessionId?: string }; + if (typeof payload.sessionId === "string") { + promptSessionIds.push(payload.sessionId); + } + } + return Effect.void; + }, + protocolLogging: { + logOutgoing: true, + logger: (event) => + Effect.sync(() => { + protocolEvents.push(event); + }), + }, + }, + }), + ), + Effect.scoped, + Effect.provide(NodeServices.layer), + ); + }); + + it.effect("keeps durable stragglers and drops stale replaced-live ids", () => + Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; + yield* runtime.start(); + const events = yield* collectEvents(runtime); + + yield* runtime.prompt({ prompt: [{ type: "text", text: "first" }] }); + yield* runtime.drainEvents; + yield* runtime.prompt({ prompt: [{ type: "text", text: "second" }] }); + yield* runtime.drainEvents; + + const text = contentDeltaText(events); + expect(text).toBe( + "root before freshreplaced live root" + + "durable root straggler" + + "root before freshreplaced live root" + + "durable root straggler", + ); + expect(text).not.toContain("stale replaced root"); + }).pipe( + Effect.provide( + replacementLayer({ + env: { T3_ACP_STALE_ROOT_AFTER_REPLACEMENT: "1" }, + options: { adoptRootSessionReplacement: true }, + }), + ), + Effect.scoped, + Effect.provide(NodeServices.layer), + ), + ); + + it.effect("never adopts a foreign session while no root prompt is in flight", () => { + const replacements: Array<{ previousSessionId: string; sessionId: string }> = []; + return Effect.gen(function* () { + const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; + yield* runtime.start(); + const events = yield* collectEvents(runtime); + + yield* runtime.prompt({ prompt: [{ type: "text", text: "hi" }] }); + yield* runtime.drainEvents; + + expect(replacements).toHaveLength(1); + expect(contentDeltaText(events)).not.toContain("foreign while idle"); + }).pipe( + Effect.provide( + replacementLayer({ + env: { T3_ACP_IDLE_FOREIGN_SESSION: "1" }, + options: { + adoptRootSessionReplacement: true, + onRootSessionReplaced: (change) => { + replacements.push(change); + }, + }, + }), + ), + Effect.scoped, + Effect.provide(NodeServices.layer), + ); + }); + }); + it.effect("supports successive standard ACP prompts", () => Effect.gen(function* () { const runtime = yield* AcpSessionRuntime.AcpSessionRuntime; diff --git a/apps/server/src/provider/acp/AcpSessionRuntime.ts b/apps/server/src/provider/acp/AcpSessionRuntime.ts index 99343641a875..83f08864c1a1 100644 --- a/apps/server/src/provider/acp/AcpSessionRuntime.ts +++ b/apps/server/src/provider/acp/AcpSessionRuntime.ts @@ -128,6 +128,24 @@ export interface AcpSessionRuntimeOptions { readonly transformSessionUpdate?: ( notification: EffectAcpSchema.SessionNotification, ) => EffectAcpSchema.SessionNotification; + /** + * Adopts a new live root session id when an agent replaces the root session + * behind the same connection — for example an agent-side command that starts a + * fresh provider session while the original id stays the one `session/load` + * replays. Adopted notifications are projected back onto the durable setup id, + * so prompts, cancellation, session loading, and item identity keep one root + * identity. Only a new id first seen while a root prompt is in flight can + * replace the live root, at most once per prompt; foreign ids seen while idle + * stay rejected. Defaults to `false`. Enable only for agents whose connection + * publishes a single live root session id and never child/subagent session + * ids; child traffic must be normalized before it reaches the runtime. + */ + readonly adoptRootSessionReplacement?: boolean; + /** Observes an adopted root session replacement for diagnostics. */ + readonly onRootSessionReplaced?: (change: { + readonly previousSessionId: string; + readonly sessionId: string; + }) => void; /** Receives bounded stderr chunks. Redact secrets before logging. A failure closes the runtime. */ readonly onStderr?: (text: string) => Effect.Effect; /** Disable only for non-interactive discovery that must surface auth-required immediately. */ @@ -1367,6 +1385,77 @@ interface AcpActivePrompt { readonly completed: Deferred.Deferred; } +interface AcpRootSessionReplacement { + readonly reset: (sessionId: string) => void; + readonly project: ( + notification: EffectAcpSchema.SessionNotification, + ) => EffectAcpSchema.SessionNotification; + readonly beginPrompt: () => void; + readonly endPrompt: () => void; +} + +/** + * Root-session replacement keeps one durable ACP session id while an agent + * publishes live updates under a replacement id on the same connection (see + * {@link AcpSessionRuntimeOptions.adoptRootSessionReplacement}). State is a + * plain object because the projection runs synchronously inside the client's + * notification decode path, before either the runtime or the adapter sees the + * notification. + */ +function makeAcpRootSessionReplacement( + options: Pick, +): AcpRootSessionReplacement | undefined { + if (options.adoptRootSessionReplacement !== true) { + return undefined; + } + let durableSessionId: string | undefined; + let liveSessionId: string | undefined; + let promptInFlight = false; + let adoptedDuringPrompt = false; + const adoptedSessionIds = new Set(); + const reset = (sessionId: string): void => { + durableSessionId = sessionId; + liveSessionId = sessionId; + adoptedDuringPrompt = false; + adoptedSessionIds.clear(); + }; + const project = ( + notification: EffectAcpSchema.SessionNotification, + ): EffectAcpSchema.SessionNotification => { + if (durableSessionId === undefined || notification.sessionId === durableSessionId) { + return notification; + } + // A replaced root keeps streaming under its adopted id; project it back so + // downstream session id checks cannot reject the live root. + if (notification.sessionId === liveSessionId) { + return { ...notification, sessionId: durableSessionId }; + } + // Unrelated foreign sessions (child/subagent traffic, stale ids) must never + // be flattened into the root: only a new id first seen while a root prompt + // is in flight replaces the live root, and only once per prompt. + if (!promptInFlight || adoptedDuringPrompt || adoptedSessionIds.has(notification.sessionId)) { + return notification; + } + const previousSessionId = liveSessionId ?? durableSessionId; + liveSessionId = notification.sessionId; + adoptedDuringPrompt = true; + adoptedSessionIds.add(notification.sessionId); + options.onRootSessionReplaced?.({ previousSessionId, sessionId: notification.sessionId }); + return { ...notification, sessionId: durableSessionId }; + }; + return { + reset, + project, + beginPrompt: () => { + promptInFlight = true; + adoptedDuringPrompt = false; + }, + endPrompt: () => { + promptInFlight = false; + }, + }; +} + export const make = ( options: AcpSessionRuntimeOptions, ): Effect.Effect< @@ -1410,6 +1499,14 @@ export const make = ( const activePromptRef = yield* Ref.make>(Option.none()); const assistantUpdatesOpenRef = yield* Ref.make(true); const sessionLoadGateRef = yield* Ref.make>(Option.none()); + const rootSessionReplacement = makeAcpRootSessionReplacement(options); + const transformSessionUpdate = + options.transformSessionUpdate === undefined && rootSessionReplacement === undefined + ? undefined + : (notification: EffectAcpSchema.SessionNotification) => { + const normalized = options.transformSessionUpdate?.(notification) ?? notification; + return rootSessionReplacement?.project(normalized) ?? normalized; + }; const ensureConnected = Effect.gen(function* () { const error = yield* Ref.get(terminationErrorRef); @@ -1729,9 +1826,7 @@ export const make = ( const acpContext = yield* Layer.build( EffectAcpClient.layerChildProcess(child, { ...(options.transformStdout ? { transformStdout: options.transformStdout } : {}), - ...(options.transformSessionUpdate - ? { transformSessionUpdate: options.transformSessionUpdate } - : {}), + ...(transformSessionUpdate === undefined ? {} : { transformSessionUpdate }), ...(options.protocolLogging?.logIncoming !== undefined ? { logIncoming: options.protocolLogging.logIncoming } : {}), @@ -1987,6 +2082,7 @@ export const make = ( yield* Ref.set(toolCallsRef, new Map()); yield* Ref.set(assistantSegmentRef, { nextSegmentIndex: 0 }); yield* Ref.set(startStateRef, { _tag: "Started", result: nextState }); + rootSessionReplacement?.reset(sessionId); return nextState; }); @@ -2332,6 +2428,7 @@ export const make = ( return yield* error.value; } yield* Ref.set(startStateRef, { _tag: "Started", result }); + rootSessionReplacement?.reset(result.sessionId); const metadata = yield* Ref.getAndSet(startupMetadataRef, []); for (const notification of metadata) { if (notification.sessionId === result.sessionId) { @@ -2637,6 +2734,10 @@ export const make = ( ...payload, } satisfies EffectAcpSchema.PromptRequest; const completed = yield* Deferred.make(); + // A root replacement can only arrive while its prompt turn is + // running; see makeAcpRootSessionReplacement. Mark the turn + // before the request can reach the agent. + rootSessionReplacement?.beginPrompt(); const fiber = yield* runLoggedRequest( "session/prompt", requestPayload, @@ -2665,6 +2766,7 @@ export const make = ( ), (activePrompt, result) => Effect.gen(function* () { + rootSessionReplacement?.endPrompt(); if ( options.cancelBehavior === "wait-for-prompt" && Exit.isFailure(result) && diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index dff5b92cbc6d..58bac7bdd161 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -885,9 +885,18 @@ export const AcpRegistrySettings = makeProviderSettingsSchema( Schema.withDecodingDefault(Effect.succeed([])), Schema.annotateKey({ providerSettingsForm: { hidden: true } }), ), + rootSessionReplacement: Schema.Boolean.pipe( + Schema.withDecodingDefault(Effect.succeed(false)), + Schema.annotateKey({ + title: "Root session replacement", + description: + "Follow a new live session id when the agent replaces its root session on the same connection, such as after a /fresh-style command. Leave this off for agents that publish child or subagent session ids.", + providerSettingsForm: { control: "switch" }, + }), + ), }, { - order: ["agentId", "commandPath", "authMethodId"], + order: ["agentId", "commandPath", "authMethodId", "rootSessionReplacement"], }, ); export type AcpRegistrySettings = typeof AcpRegistrySettings.Type;