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;