diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 988daad01..786304013 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -1,6 +1,7 @@ # Changelog ## [Unreleased] + ### Breaking Changes - Simplified chunk edit operations: removed `append_child`, `prepend_child`, `append_sibling`, `prepend_sibling`, and `replace_body` ops in favor of unified `replace`, `before`, `after`, `prepend`, and `append` with region targeting (`@container`, `@prologue`, `@body`, `@epilogue`) @@ -9,6 +10,16 @@ ### Added +- Multi-session support in ACP mode: agents can now manage multiple concurrent sessions with independent state, models, and configurations +- Session forking in ACP mode: `unstable_forkSession` creates a new session from an existing one's history +- Session resumption in ACP mode: `unstable_resumeSession` reloads a previously saved session +- Session closure in ACP mode: `unstable_closeSession` cleanly shuts down a session and releases resources +- Model state reporting in ACP mode: `SessionModelState` with available models and current selection in session responses +- Direct model setting in ACP mode: `unstable_setSessionModel` RPC command for changing the active model +- Turn-level usage tracking in ACP mode: prompt responses now include `usage` with input/output/cached token counts +- Message ID tracking in ACP mode: stable message IDs for assistant chunks enabling client-side message correlation +- Settings cloning: `Settings.cloneForCwd()` method to create isolated settings instances for different working directories +- Extension flag value retrieval: `ExtensionRunner.getFlagValues()` to inspect current flag state - Exported autoresearch module and submodules via `./autoresearch` and `./autoresearch/*` package paths - Exported autoresearch tools via `./autoresearch/tools/*` package path - Exported CLI commands via `./cli/commands/*` package path @@ -55,6 +66,10 @@ ### Changed +- ACP agent now manages multiple sessions instead of a single session; session lifecycle and configuration are now per-session +- ACP session creation now uses a factory function to support creating new sessions for different working directories +- ACP event mapping now accepts optional `getMessageId` callback for stable message ID assignment to assistant chunks +- ACP session initialization now registers connection cleanup handlers to dispose all sessions on disconnect - Reorganized package.json exports: moved `./edit` exports before `./plan-mode` for better logical grouping - Notebook conversion logic now checks for raw read mode or non-chunk mode before converting via markit, allowing chunk-mode reads of `.ipynb` files to use chunk parsing instead of conversion - Go receiver methods now render as top-level siblings instead of nested under their receiver type in chunk read output @@ -109,6 +124,7 @@ ### Fixed +- ACP session cleanup now properly cancels in-flight prompts and disposes resources when sessions are closed or connection aborts - Removed unused `_createErrorToolResult` helper function from RPC host-tools module - Fixed Go receiver method indentation in append operations to preserve relative indentation from the anchor chunk - Fixed Go type chunk line counts to report only the type body lines instead of including grouped receiver methods diff --git a/packages/coding-agent/src/config/settings.ts b/packages/coding-agent/src/config/settings.ts index b62aad5de..63ef361eb 100644 --- a/packages/coding-agent/src/config/settings.ts +++ b/packages/coding-agent/src/config/settings.ts @@ -269,6 +269,21 @@ export class Settings { } } + async cloneForCwd(cwd: string): Promise { + const cloned = new Settings({ + cwd, + agentDir: this.#agentDir, + inMemory: !this.#persist, + }); + cloned.#storage = this.#storage; + cloned.#global = structuredClone(this.#global); + cloned.#project = this.#persist ? await cloned.#loadProjectSettings() : structuredClone(this.#project); + cloned.#overrides = structuredClone(this.#overrides); + cloned.#rebuildMerged(); + cloned.#fireAllHooks(); + return cloned; + } + // ───────────────────────────────────────────────────────────────────────── // Accessors // ───────────────────────────────────────────────────────────────────────── diff --git a/packages/coding-agent/src/extensibility/extensions/runner.ts b/packages/coding-agent/src/extensibility/extensions/runner.ts index 9758a3b8d..840ebf456 100644 --- a/packages/coding-agent/src/extensibility/extensions/runner.ts +++ b/packages/coding-agent/src/extensibility/extensions/runner.ts @@ -262,6 +262,10 @@ export class ExtensionRunner { return allFlags; } + getFlagValues(): Map { + return new Map(this.runtime.flagValues); + } + setFlagValue(name: string, value: boolean | string): void { this.runtime.flagValues.set(name, value); } diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 933c80784..0c75acb27 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -822,10 +822,32 @@ export async function runRootCommand(parsed: Args, rawArgs: string[]): Promise(); + const createAcpSession = async (cwd: string) => { + const nextSettings = await session.settings.cloneForCwd(cwd); + const nextSessionManager = SessionManager.create(cwd, parsedArgs.sessionDir); + const { session: nextSession } = await createAgentSession({ + ...sessionOptions, + cwd, + sessionManager: nextSessionManager, + settings: nextSettings, + authStorage, + modelRegistry, + searchDb: session.searchDb, + hasUI: false, + }); + if (nextSession.extensionRunner) { + for (const [flagName, value] of extensionFlagValues) { + nextSession.extensionRunner.setFlagValue(flagName, value); + } + } + return nextSession; + }; + if (mode === "rpc") { await runRpcMode(session); } else if (mode === "acp") { - await runAcpMode(session); + await runAcpMode(session, createAcpSession); } else if (isInteractive) { const versionCheckPromise = checkForNewVersion(VERSION).catch(() => undefined); const changelogMarkdown = await getChangelogForDisplay(parsedArgs); diff --git a/packages/coding-agent/src/modes/acp/acp-agent.ts b/packages/coding-agent/src/modes/acp/acp-agent.ts index 4dfd21547..f6a0c06ef 100644 --- a/packages/coding-agent/src/modes/acp/acp-agent.ts +++ b/packages/coding-agent/src/modes/acp/acp-agent.ts @@ -5,6 +5,10 @@ import { type AuthenticateRequest, type AuthenticateResponse, type AvailableCommand, + type CloseSessionRequest, + type CloseSessionResponse, + type ForkSessionRequest, + type ForkSessionResponse, type InitializeRequest, type InitializeResponse, type ListSessionsRequest, @@ -17,15 +21,21 @@ import { PROTOCOL_VERSION, type PromptRequest, type PromptResponse, + type ResumeSessionRequest, + type ResumeSessionResponse, type SessionConfigOption, type SessionInfo, + type SessionModelState, type SessionModeState, type SessionNotification, type SessionUpdate, type SetSessionConfigOptionRequest, type SetSessionConfigOptionResponse, + type SetSessionModelRequest, + type SetSessionModelResponse, type SetSessionModeRequest, type SetSessionModeResponse, + type Usage, } from "@agentclientprotocol/sdk"; import type { Model } from "@oh-my-pi/pi-ai"; import { logger, VERSION } from "@oh-my-pi/pi-utils"; @@ -35,7 +45,11 @@ import { MCPManager } from "../../mcp/manager"; import type { MCPServerConfig } from "../../mcp/types"; import { theme } from "../../modes/theme/theme"; import type { AgentSession, AgentSessionEvent } from "../../session/agent-session"; -import { SessionManager, type SessionInfo as StoredSessionInfo } from "../../session/session-manager"; +import { + SessionManager, + type SessionInfo as StoredSessionInfo, + type UsageStatistics, +} from "../../session/session-manager"; import { parseThinkingLevel } from "../../thinking"; import { mapAgentSessionEventToAcpSessionUpdates, mapToolKind } from "./acp-event-mapper"; @@ -53,14 +67,23 @@ type AgentImageContent = { }; type PromptTurnState = { - messageId: string | null; + userMessageId: string; cancelRequested: boolean; settled: boolean; + usageBaseline: UsageStatistics; unsubscribe: (() => void) | undefined; resolve: (value: PromptResponse) => void; reject: (reason?: unknown) => void; }; +type ManagedSessionRecord = { + session: AgentSession; + mcpManager: MCPManager | undefined; + promptTurn: PromptTurnState | undefined; + liveMessageIds: WeakMap; + extensionsConfigured: boolean; +}; + type ReplayableMessage = { role: string; content?: unknown; @@ -86,6 +109,8 @@ type MCPSourceMap = { [name: string]: MCPSource; }; +type CreateAcpSession = (cwd: string) => Promise; + const acpExtensionUiContext: ExtensionUIContext = { select: async () => undefined, confirm: async () => false, @@ -118,17 +143,20 @@ const acpExtensionUiContext: ExtensionUIContext = { export class AcpAgent implements Agent { #connection: AgentSideConnection; - #session: AgentSession; - #mcpManager: MCPManager | undefined; - #promptTurn: PromptTurnState | undefined; - #hasOpenedSession = false; + #initialSession: AgentSession | undefined; + #createSession: CreateAcpSession; + #sessions = new Map(); + #disposePromise: Promise | undefined; + #cleanupRegistered = false; - constructor(connection: AgentSideConnection, session: AgentSession) { + constructor(connection: AgentSideConnection, initialSession: AgentSession, createSession: CreateAcpSession) { this.#connection = connection; - this.#session = session; + this.#initialSession = initialSession; + this.#createSession = createSession; } async initialize(_params: InitializeRequest): Promise { + this.#registerConnectionCleanup(); return { protocolVersion: PROTOCOL_VERSION, agentInfo: { @@ -155,6 +183,9 @@ export class AcpAgent implements Agent { }, sessionCapabilities: { list: {}, + fork: {}, + resume: {}, + close: {}, }, }, }; @@ -166,50 +197,27 @@ export class AcpAgent implements Agent { async newSession(params: NewSessionRequest): Promise { this.#assertAbsoluteCwd(params.cwd); - await this.#session.sessionManager.flush(); - await this.#session.sessionManager.moveTo(params.cwd); - if (this.#hasOpenedSession) { - const success = await this.#session.newSession(); - if (!success) { - throw new Error("ACP session creation was cancelled"); - } - } - this.#hasOpenedSession = true; - await this.#session.sessionManager.ensureOnDisk(); - await this.#configureExtensions(); - await this.#configureMcpServers(params.mcpServers); + const record = await this.#createNewSessionRecord(params.cwd, params.mcpServers); const response: NewSessionResponse = { - sessionId: this.#sessionId, - configOptions: this.#buildConfigOptions(), + sessionId: record.session.sessionId, + configOptions: this.#buildConfigOptions(record.session), + models: this.#buildModelState(record.session), modes: this.#buildModeState(), }; - this.#scheduleBootstrapUpdates(this.#sessionId); + this.#scheduleBootstrapUpdates(record.session.sessionId); return response; } async loadSession(params: LoadSessionRequest): Promise { this.#assertAbsoluteCwd(params.cwd); - await this.#session.sessionManager.flush(); - const storedSession = await this.#findStoredSession(params.sessionId, params.cwd); - if (!storedSession) { - throw new Error(`ACP session not found: ${params.sessionId}`); - } - const currentSessionFile = this.#session.sessionManager.getSessionFile(); - if (currentSessionFile !== storedSession.path) { - const success = await this.#session.switchSession(storedSession.path); - if (!success) { - throw new Error(`ACP session load was cancelled: ${params.sessionId}`); - } - } - this.#hasOpenedSession = true; - await this.#configureExtensions(); - await this.#configureMcpServers(params.mcpServers); - await this.#replaySessionHistory(); + const record = await this.#loadManagedSession(params.sessionId, params.cwd, params.mcpServers); + await this.#replaySessionHistory(record); const response: LoadSessionResponse = { - configOptions: this.#buildConfigOptions(), + configOptions: this.#buildConfigOptions(record.session), + models: this.#buildModelState(record.session), modes: this.#buildModeState(), }; - this.#scheduleBootstrapUpdates(this.#sessionId); + this.#scheduleBootstrapUpdates(record.session.sessionId); return response; } @@ -217,7 +225,9 @@ export class AcpAgent implements Agent { if (params.cwd) { this.#assertAbsoluteCwd(params.cwd); } - await this.#session.sessionManager.flush(); + for (const record of this.#sessions.values()) { + await record.session.sessionManager.flush(); + } const sessions = await this.#listStoredSessions(params.cwd ?? undefined); const offset = this.#parseCursor(params.cursor ?? undefined); const paged = sessions.slice(offset, offset + SESSION_PAGE_SIZE); @@ -228,20 +238,54 @@ export class AcpAgent implements Agent { }; } + async unstable_resumeSession(params: ResumeSessionRequest): Promise { + this.#assertAbsoluteCwd(params.cwd); + const record = await this.#resumeManagedSession(params.sessionId, params.cwd, params.mcpServers ?? []); + const response: ResumeSessionResponse = { + configOptions: this.#buildConfigOptions(record.session), + models: this.#buildModelState(record.session), + modes: this.#buildModeState(), + }; + this.#scheduleBootstrapUpdates(record.session.sessionId); + return response; + } + + async unstable_forkSession(params: ForkSessionRequest): Promise { + this.#assertAbsoluteCwd(params.cwd); + const record = await this.#forkManagedSession(params); + const response: ForkSessionResponse = { + sessionId: record.session.sessionId, + configOptions: this.#buildConfigOptions(record.session), + models: this.#buildModelState(record.session), + modes: this.#buildModeState(), + }; + this.#scheduleBootstrapUpdates(record.session.sessionId); + return response; + } + + async unstable_closeSession(params: CloseSessionRequest): Promise { + const record = this.#sessions.get(params.sessionId); + if (!record) { + return {}; + } + await this.#closeManagedSession(params.sessionId, record); + return {}; + } + async setSessionMode(params: SetSessionModeRequest): Promise { - this.#assertSameSession(params.sessionId); + const record = this.#getSessionRecord(params.sessionId); if (params.modeId !== ACP_MODE_ID) { throw new Error(`Unsupported ACP mode: ${params.modeId}`); } await this.#connection.sessionUpdate({ - sessionId: this.#sessionId, + sessionId: record.session.sessionId, update: this.#buildCurrentModeUpdate(), }); return {}; } async setSessionConfigOption(params: SetSessionConfigOptionRequest): Promise { - this.#assertSameSession(params.sessionId); + const record = this.#getSessionRecord(params.sessionId); if (typeof params.value === "boolean") { throw new Error(`Unsupported boolean ACP config option: ${params.configId}`); } @@ -253,18 +297,18 @@ export class AcpAgent implements Agent { } break; case MODEL_CONFIG_ID: - await this.#setModelById(params.value); + await this.#setModelById(record.session, params.value); break; case THINKING_CONFIG_ID: - this.#setThinkingLevelById(params.value); + this.#setThinkingLevelById(record.session, params.value); break; default: throw new Error(`Unknown ACP config option: ${params.configId}`); } - const configOptions = this.#buildConfigOptions(); + const configOptions = this.#buildConfigOptions(record.session); await this.#connection.sessionUpdate({ - sessionId: this.#sessionId, + sessionId: record.session.sessionId, update: { sessionUpdate: "config_option_update", configOptions, @@ -273,49 +317,64 @@ export class AcpAgent implements Agent { return { configOptions }; } + async unstable_setSessionModel(params: SetSessionModelRequest): Promise { + const record = this.#getSessionRecord(params.sessionId); + await this.#setModelById(record.session, params.modelId); + await this.#connection.sessionUpdate({ + sessionId: record.session.sessionId, + update: { + sessionUpdate: "config_option_update", + configOptions: this.#buildConfigOptions(record.session), + }, + }); + return {}; + } + async prompt(params: PromptRequest): Promise { - this.#assertSameSession(params.sessionId); - if (this.#promptTurn && !this.#promptTurn.settled) { + const record = this.#getSessionRecord(params.sessionId); + if (record.promptTurn && !record.promptTurn.settled) { throw new Error("ACP prompt already in progress for this session"); } const converted = this.#convertPromptBlocks(params.prompt); const pendingPrompt = Promise.withResolvers(); - this.#promptTurn = { - messageId: params.messageId ?? null, + record.promptTurn = { + userMessageId: params.messageId ?? crypto.randomUUID(), cancelRequested: false, settled: false, + usageBaseline: this.#cloneUsageStatistics(record.session.sessionManager.getUsageStatistics()), unsubscribe: undefined, resolve: pendingPrompt.resolve, reject: pendingPrompt.reject, }; - this.#promptTurn.unsubscribe = this.#session.subscribe(event => { - void this.#handlePromptEvent(event); + record.promptTurn.unsubscribe = record.session.subscribe(event => { + void this.#handlePromptEvent(record, event); }); - this.#session.prompt(converted.text, { images: converted.images }).catch((error: unknown) => { - this.#finishPrompt(undefined, error); + record.session.prompt(converted.text, { images: converted.images }).catch((error: unknown) => { + this.#finishPrompt(record, undefined, error); }); return await pendingPrompt.promise; } async cancel(params: { sessionId: string }): Promise { - this.#assertSameSession(params.sessionId); - const promptTurn = this.#promptTurn; + const record = this.#getSessionRecord(params.sessionId); + const promptTurn = record.promptTurn; if (!promptTurn || promptTurn.settled) { return; } promptTurn.cancelRequested = true; try { - await this.#session.abort(); - this.#finishPrompt({ + await record.session.abort(); + this.#finishPrompt(record, { stopReason: "cancelled", - userMessageId: promptTurn.messageId, + usage: this.#buildTurnUsage(promptTurn.usageBaseline, record.session.sessionManager.getUsageStatistics()), + userMessageId: promptTurn.userMessageId, }); } catch (error: unknown) { - this.#finishPrompt(undefined, error); + this.#finishPrompt(record, undefined, error); } } @@ -333,37 +392,206 @@ export class AcpAgent implements Agent { return this.#connection.closed; } - get #sessionId(): string { - return this.#session.sessionId; + #registerConnectionCleanup(): void { + if (this.#cleanupRegistered) { + return; + } + this.#cleanupRegistered = true; + this.#connection.signal.addEventListener( + "abort", + () => { + void this.#disposeAllSessions(); + }, + { once: true }, + ); } - async #handlePromptEvent(event: AgentSessionEvent): Promise { - const promptTurn = this.#promptTurn; + async #createNewSessionRecord(cwd: string, mcpServers: McpServer[]): Promise { + const session = await this.#createSession(path.resolve(cwd)); + try { + await session.sessionManager.ensureOnDisk(); + } catch (error) { + await this.#disposeStandaloneSession(session); + throw error; + } + return await this.#registerPreparedSession(session, mcpServers); + } + + async #loadManagedSession(sessionId: string, cwd: string, mcpServers: McpServer[]): Promise { + const existing = this.#sessions.get(sessionId); + if (existing) { + this.#assertMatchingCwd(existing.session, cwd); + await this.#configureMcpServers(existing, mcpServers); + return existing; + } + + const storedSession = await this.#findStoredSession(sessionId, cwd); + if (!storedSession) { + throw new Error(`ACP session not found: ${sessionId}`); + } + return await this.#openStoredSession(storedSession.path, cwd, mcpServers, sessionId); + } + + async #resumeManagedSession(sessionId: string, cwd: string, mcpServers: McpServer[]): Promise { + const existing = this.#sessions.get(sessionId); + if (existing) { + this.#assertMatchingCwd(existing.session, cwd); + await this.#configureMcpServers(existing, mcpServers); + return existing; + } + + const storedSession = await this.#findStoredSession(sessionId, cwd); + if (!storedSession) { + throw new Error(`ACP session not found: ${sessionId}`); + } + return await this.#openStoredSession(storedSession.path, cwd, mcpServers, sessionId); + } + + async #forkManagedSession(params: ForkSessionRequest): Promise { + const sourcePath = await this.#resolveForkSourceSessionPath(params.sessionId); + const session = await this.#createSession(path.resolve(params.cwd)); + try { + const success = await session.switchSession(sourcePath); + if (!success) { + throw new Error(`ACP session fork was cancelled: ${params.sessionId}`); + } + await session.sessionManager.flush(); + const forked = await session.sessionManager.fork(); + if (!forked) { + throw new Error(`ACP session fork failed: ${params.sessionId}`); + } + session.agent.sessionId = session.sessionManager.getSessionId(); + await session.sessionManager.ensureOnDisk(); + } catch (error) { + await this.#disposeStandaloneSession(session); + throw error; + } + return await this.#registerPreparedSession(session, params.mcpServers ?? []); + } + + async #openStoredSession( + sessionPath: string, + cwd: string, + mcpServers: McpServer[], + sessionId: string, + ): Promise { + const session = await this.#createSession(path.resolve(cwd)); + try { + const success = await session.switchSession(sessionPath); + if (!success) { + throw new Error(`ACP session load was cancelled: ${sessionId}`); + } + } catch (error) { + await this.#disposeStandaloneSession(session); + throw error; + } + return await this.#registerPreparedSession(session, mcpServers); + } + + async #registerPreparedSession(session: AgentSession, mcpServers: McpServer[]): Promise { + const record = this.#createManagedSessionRecord(session); + try { + await this.#configureExtensions(record); + await this.#configureMcpServers(record, mcpServers); + this.#sessions.set(session.sessionId, record); + return record; + } catch (error) { + await this.#disposeSessionRecord(record); + throw error; + } + } + + #createManagedSessionRecord(session: AgentSession): ManagedSessionRecord { + return { + session, + mcpManager: undefined, + promptTurn: undefined, + liveMessageIds: new WeakMap(), + extensionsConfigured: false, + }; + } + + #getSessionRecord(sessionId: string): ManagedSessionRecord { + const record = this.#sessions.get(sessionId); + if (!record) { + throw new Error(`Unsupported ACP session: ${sessionId}`); + } + return record; + } + + #assertMatchingCwd(session: AgentSession, cwd: string): void { + const expected = path.resolve(cwd); + const actual = path.resolve(session.sessionManager.getCwd()); + if (actual !== expected) { + throw new Error(`ACP session ${session.sessionId} is already loaded for ${actual}, not ${expected}`); + } + } + + async #resolveForkSourceSessionPath(sessionId: string): Promise { + const loaded = this.#sessions.get(sessionId); + if (loaded) { + const promptTurn = loaded.promptTurn; + if (promptTurn && !promptTurn.settled) { + throw new Error(`ACP session fork is unavailable while a prompt is in progress: ${sessionId}`); + } + await loaded.session.sessionManager.flush(); + const sessionPath = loaded.session.sessionManager.getSessionFile(); + if (!sessionPath) { + throw new Error(`ACP session cannot be forked before it is persisted: ${sessionId}`); + } + return sessionPath; + } + + const storedSession = await this.#findStoredSessionById(sessionId); + if (!storedSession) { + throw new Error(`ACP session not found: ${sessionId}`); + } + return storedSession.path; + } + + async #handlePromptEvent(record: ManagedSessionRecord, event: AgentSessionEvent): Promise { + const promptTurn = record.promptTurn; if (!promptTurn || promptTurn.settled) { return; } - for (const notification of mapAgentSessionEventToAcpSessionUpdates(event, this.#sessionId)) { + for (const notification of mapAgentSessionEventToAcpSessionUpdates(event, record.session.sessionId, { + getMessageId: message => this.#getLiveMessageId(record, message), + })) { await this.#connection.sessionUpdate(notification); } if (event.type === "agent_end") { - await this.#emitEndOfTurnUpdates(); - this.#finishPrompt({ + await this.#emitEndOfTurnUpdates(record); + this.#finishPrompt(record, { stopReason: promptTurn.cancelRequested ? "cancelled" : "end_turn", - userMessageId: promptTurn.messageId, + usage: this.#buildTurnUsage(promptTurn.usageBaseline, record.session.sessionManager.getUsageStatistics()), + userMessageId: promptTurn.userMessageId, }); } } - #finishPrompt(response?: PromptResponse, error?: unknown): void { - const promptTurn = this.#promptTurn; + #getLiveMessageId(record: ManagedSessionRecord, message: unknown): string | undefined { + if (typeof message !== "object" || message === null) { + return undefined; + } + const existing = record.liveMessageIds.get(message); + if (existing) { + return existing; + } + const nextMessageId = crypto.randomUUID(); + record.liveMessageIds.set(message, nextMessageId); + return nextMessageId; + } + + #finishPrompt(record: ManagedSessionRecord, response?: PromptResponse, error?: unknown): void { + const promptTurn = record.promptTurn; if (!promptTurn || promptTurn.settled) { return; } promptTurn.settled = true; promptTurn.unsubscribe?.(); - this.#promptTurn = undefined; + record.promptTurn = undefined; if (error !== undefined) { promptTurn.reject(error); return; @@ -371,12 +599,6 @@ export class AcpAgent implements Agent { promptTurn.resolve(response ?? { stopReason: "end_turn" }); } - #assertSameSession(sessionId: string): void { - if (sessionId !== this.#sessionId) { - throw new Error(`Unsupported ACP session: ${sessionId}`); - } - } - #assertAbsoluteCwd(cwd: string): void { if (!path.isAbsolute(cwd)) { throw new Error(`ACP cwd must be absolute: ${cwd}`); @@ -415,7 +637,7 @@ export class AcpAgent implements Agent { }; } - #buildConfigOptions(): SessionConfigOption[] { + #buildConfigOptions(session: AgentSession): SessionConfigOption[] { const configOptions: SessionConfigOption[] = [ { id: MODE_CONFIG_ID, @@ -427,8 +649,8 @@ export class AcpAgent implements Agent { }, ]; - const models = this.#session.getAvailableModels(); - const currentModel = this.#session.model; + const models = session.getAvailableModels(); + const currentModel = session.model; if (models.length > 0) { configOptions.push({ id: MODEL_CONFIG_ID, @@ -449,16 +671,38 @@ export class AcpAgent implements Agent { name: "Thinking", category: "thought_level", type: "select", - currentValue: this.#toThinkingConfigValue(this.#session.thinkingLevel), - options: this.#buildThinkingOptions(), + currentValue: this.#toThinkingConfigValue(session.thinkingLevel), + options: this.#buildThinkingOptions(session), }); return configOptions; } - #buildThinkingOptions(): Array<{ value: string; name: string; description?: string }> { + #buildModelState(session: AgentSession): SessionModelState | undefined { + const models = session.getAvailableModels(); + if (models.length === 0) { + return undefined; + } + + const availableModels = models.map(model => ({ + modelId: this.#toModelId(model), + name: model.name, + description: `${model.provider}/${model.id}`, + })); + const currentModelId = session.model ? this.#toModelId(session.model) : availableModels[0]?.modelId; + if (!currentModelId) { + return undefined; + } + + return { + availableModels, + currentModelId, + }; + } + + #buildThinkingOptions(session: AgentSession): Array<{ value: string; name: string; description?: string }> { return [ { value: THINKING_OFF, name: "Off" }, - ...this.#session.getAvailableThinkingLevels().map(level => ({ + ...session.getAvailableThinkingLevels().map(level => ({ value: level, name: level, })), @@ -469,20 +713,20 @@ export class AcpAgent implements Agent { return value && value !== "inherit" ? value : THINKING_OFF; } - async #setModelById(modelId: string): Promise { - const model = this.#session.getAvailableModels().find(candidate => this.#toModelId(candidate) === modelId); + async #setModelById(session: AgentSession, modelId: string): Promise { + const model = session.getAvailableModels().find(candidate => this.#toModelId(candidate) === modelId); if (!model) { throw new Error(`Unknown ACP model: ${modelId}`); } - await this.#session.setModel(model); + await session.setModel(model); } - #setThinkingLevelById(value: string): void { + #setThinkingLevelById(session: AgentSession, value: string): void { const thinkingLevel = parseThinkingLevel(value); if (!thinkingLevel) { throw new Error(`Unknown ACP thinking level: ${value}`); } - this.#session.setThinkingLevel(thinkingLevel); + session.setThinkingLevel(thinkingLevel); } #toModelId(model: Model): string { @@ -503,7 +747,7 @@ export class AcpAgent implements Agent { }; } - async #buildAvailableCommands(): Promise { + async #buildAvailableCommands(session: AgentSession): Promise { const commands: AvailableCommand[] = []; const seenNames = new Set(); const appendCommand = (command: AvailableCommand): void => { @@ -514,7 +758,7 @@ export class AcpAgent implements Agent { commands.push(command); }; - for (const command of this.#session.customCommands) { + for (const command of session.customCommands) { appendCommand({ name: command.command.name, description: command.command.description, @@ -522,7 +766,7 @@ export class AcpAgent implements Agent { }); } - for (const command of await loadSlashCommands({ cwd: this.#session.sessionManager.getCwd() })) { + for (const command of await loadSlashCommands({ cwd: session.sessionManager.getCwd() })) { appendCommand({ name: command.name, description: command.description, @@ -543,41 +787,44 @@ export class AcpAgent implements Agent { #scheduleBootstrapUpdates(sessionId: string): void { setTimeout(() => { - if (sessionId !== this.#sessionId || this.#connection.signal.aborted) { + if (this.#connection.signal.aborted) { return; } - void this.#emitBootstrapUpdates(sessionId); + const record = this.#sessions.get(sessionId); + if (!record) { + return; + } + void this.#emitBootstrapUpdates(sessionId, record); }, 0); } - async #emitBootstrapUpdates(sessionId: string): Promise { - if (sessionId !== this.#sessionId) { + async #emitBootstrapUpdates(sessionId: string, record: ManagedSessionRecord): Promise { + if (this.#sessions.get(sessionId) !== record) { return; } await this.#connection.sessionUpdate({ sessionId, update: { sessionUpdate: "available_commands_update", - availableCommands: await this.#buildAvailableCommands(), + availableCommands: await this.#buildAvailableCommands(record.session), }, }); await this.#connection.sessionUpdate({ sessionId, update: { sessionUpdate: "session_info_update", - title: this.#session.sessionName, - updatedAt: this.#session.sessionManager.getHeader()?.timestamp, + title: record.session.sessionName, + updatedAt: record.session.sessionManager.getHeader()?.timestamp, }, }); } - async #emitEndOfTurnUpdates(): Promise { - const sessionId = this.#sessionId; + async #emitEndOfTurnUpdates(record: ManagedSessionRecord): Promise { + const sessionId = record.session.sessionId; - // Emit usage update with context token counts - const contextUsage = this.#session.getContextUsage(); + const contextUsage = record.session.getContextUsage(); if (contextUsage) { - const usageStats = this.#session.sessionManager.getUsageStatistics(); + const usageStats = record.session.sessionManager.getUsageStatistics(); await this.#connection.sessionUpdate({ sessionId, update: { @@ -589,17 +836,52 @@ export class AcpAgent implements Agent { }); } - // Push latest session title await this.#connection.sessionUpdate({ sessionId, update: { sessionUpdate: "session_info_update", - title: this.#session.sessionName, + title: record.session.sessionName, updatedAt: new Date().toISOString(), }, }); } + #cloneUsageStatistics(usage: UsageStatistics): UsageStatistics { + return { + input: usage.input, + output: usage.output, + cacheRead: usage.cacheRead, + cacheWrite: usage.cacheWrite, + premiumRequests: usage.premiumRequests, + cost: usage.cost, + }; + } + + #buildTurnUsage(previous: UsageStatistics, current: UsageStatistics): Usage | undefined { + const inputTokens = Math.max(0, current.input - previous.input); + const outputTokens = Math.max(0, current.output - previous.output); + const cachedReadTokens = Math.max(0, current.cacheRead - previous.cacheRead); + const cachedWriteTokens = Math.max(0, current.cacheWrite - previous.cacheWrite); + const totalTokens = inputTokens + outputTokens + cachedReadTokens + cachedWriteTokens; + + if (totalTokens === 0) { + return undefined; + } + + const usage: Usage = { + inputTokens, + outputTokens, + totalTokens, + }; + if (cachedReadTokens > 0) { + usage.cachedReadTokens = cachedReadTokens; + } + if (cachedWriteTokens > 0) { + usage.cachedWriteTokens = cachedWriteTokens; + } + return usage; + } + async #listStoredSessions(cwd?: string): Promise { const sessions = cwd ? await SessionManager.list(cwd) : await SessionManager.listAll(); return sessions.sort((left, right) => right.modified.getTime() - left.modified.getTime()); @@ -610,6 +892,11 @@ export class AcpAgent implements Agent { return sessions.find(session => session.id === sessionId); } + async #findStoredSessionById(sessionId: string): Promise { + const sessions = await this.#listStoredSessions(); + return sessions.find(session => session.id === sessionId); + } + #parseCursor(cursor: string | undefined): number { if (!cursor) { return 0; @@ -621,17 +908,17 @@ export class AcpAgent implements Agent { return parsed; } - async #replaySessionHistory(): Promise { - for (const message of this.#session.sessionManager.buildSessionContext().messages as ReplayableMessage[]) { - for (const notification of this.#messageToReplayNotifications(message)) { + async #replaySessionHistory(record: ManagedSessionRecord): Promise { + for (const message of record.session.sessionManager.buildSessionContext().messages as ReplayableMessage[]) { + for (const notification of this.#messageToReplayNotifications(record.session.sessionId, message)) { await this.#connection.sessionUpdate(notification); } } } - #messageToReplayNotifications(message: ReplayableMessage): SessionNotification[] { + #messageToReplayNotifications(sessionId: string, message: ReplayableMessage): SessionNotification[] { if (message.role === "assistant") { - return this.#replayAssistantMessage(message); + return this.#replayAssistantMessage(sessionId, message); } if ( message.role === "user" || @@ -639,28 +926,42 @@ export class AcpAgent implements Agent { message.role === "custom" || message.role === "hookMessage" ) { - return this.#wrapReplayContent(this.#extractReplayContent(message.content, undefined), "user_message_chunk"); + return this.#wrapReplayContent( + sessionId, + this.#extractReplayContent(message.content, undefined), + "user_message_chunk", + crypto.randomUUID(), + ); } if ( message.role === "toolResult" && typeof message.toolCallId === "string" && typeof message.toolName === "string" ) { - return this.#replayToolResult({ ...message, toolCallId: message.toolCallId, toolName: message.toolName }); + return this.#replayToolResult(sessionId, { + ...message, + toolCallId: message.toolCallId, + toolName: message.toolName, + }); } if ( message.role === "bashExecution" || message.role === "pythonExecution" || message.role === "compactionSummary" ) { - return this.#wrapReplayContent(this.#extractReplayContent(message.content, undefined), "user_message_chunk"); + return this.#wrapReplayContent( + sessionId, + this.#extractReplayContent(message.content, undefined), + "user_message_chunk", + crypto.randomUUID(), + ); } return []; } - #replayAssistantMessage(message: ReplayableMessage): SessionNotification[] { + #replayAssistantMessage(sessionId: string, message: ReplayableMessage): SessionNotification[] { const notifications: SessionNotification[] = []; - const sessionId = this.#sessionId; + const messageId = crypto.randomUUID(); if (Array.isArray(message.content)) { for (const item of message.content) { if (typeof item !== "object" || item === null || !("type" in item)) { @@ -669,7 +970,11 @@ export class AcpAgent implements Agent { if (item.type === "text" && "text" in item && typeof item.text === "string" && item.text.length > 0) { notifications.push({ sessionId, - update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: item.text } }, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: item.text }, + messageId, + }, }); continue; } @@ -681,7 +986,11 @@ export class AcpAgent implements Agent { ) { notifications.push({ sessionId, - update: { sessionUpdate: "agent_thought_chunk", content: { type: "text", text: item.thinking } }, + update: { + sessionUpdate: "agent_thought_chunk", + content: { type: "text", text: item.thinking }, + messageId, + }, }); continue; } @@ -709,13 +1018,18 @@ export class AcpAgent implements Agent { if (notifications.length === 0 && message.errorMessage) { notifications.push({ sessionId, - update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: message.errorMessage } }, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: message.errorMessage }, + messageId, + }, }); } return notifications; } #replayToolResult( + sessionId: string, message: Required> & ReplayableMessage, ): SessionNotification[] { const args = this.#buildReplayToolArgs(message.details); @@ -737,8 +1051,8 @@ export class AcpAgent implements Agent { }, }; return [ - ...mapAgentSessionEventToAcpSessionUpdates(startEvent, this.#sessionId), - ...mapAgentSessionEventToAcpSessionUpdates(endEvent, this.#sessionId), + ...mapAgentSessionEventToAcpSessionUpdates(startEvent, sessionId), + ...mapAgentSessionEventToAcpSessionUpdates(endEvent, sessionId), ]; } @@ -751,14 +1065,17 @@ export class AcpAgent implements Agent { } #wrapReplayContent( + sessionId: string, content: PromptRequest["prompt"], kind: "agent_message_chunk" | "user_message_chunk", + messageId: string, ): SessionNotification[] { return content.map(block => ({ - sessionId: this.#sessionId, + sessionId, update: { sessionUpdate: kind, content: block, + messageId, }, })); } @@ -791,89 +1108,94 @@ export class AcpAgent implements Agent { return replay; } - async #configureExtensions(): Promise { - const extensionRunner = this.#session.extensionRunner; + async #configureExtensions(record: ManagedSessionRecord): Promise { + if (record.extensionsConfigured) { + return; + } + + const extensionRunner = record.session.extensionRunner; if (!extensionRunner) { + record.extensionsConfigured = true; return; } extensionRunner.initialize( { sendMessage: (message, options) => { - this.#session.sendCustomMessage(message, options).catch((error: unknown) => { + record.session.sendCustomMessage(message, options).catch((error: unknown) => { logger.warn("ACP extension sendMessage failed", { error }); }); }, sendUserMessage: (content, options) => { - this.#session.sendUserMessage(content, options).catch((error: unknown) => { + record.session.sendUserMessage(content, options).catch((error: unknown) => { logger.warn("ACP extension sendUserMessage failed", { error }); }); }, appendEntry: (customType, data) => { - this.#session.sessionManager.appendCustomEntry(customType, data); + record.session.sessionManager.appendCustomEntry(customType, data); }, setLabel: (targetId, label) => { - this.#session.sessionManager.appendLabelChange(targetId, label); + record.session.sessionManager.appendLabelChange(targetId, label); }, - getActiveTools: () => this.#session.getActiveToolNames(), - getAllTools: () => this.#session.getAllToolNames(), - setActiveTools: toolNames => this.#session.setActiveToolsByName(toolNames), + getActiveTools: () => record.session.getActiveToolNames(), + getAllTools: () => record.session.getAllToolNames(), + setActiveTools: toolNames => record.session.setActiveToolsByName(toolNames), getCommands: () => [], setModel: async model => { - const apiKey = await this.#session.modelRegistry.getApiKey(model); + const apiKey = await record.session.modelRegistry.getApiKey(model); if (!apiKey) { return false; } - await this.#session.setModel(model); + await record.session.setModel(model); return true; }, - getThinkingLevel: () => this.#session.thinkingLevel, - setThinkingLevel: level => this.#session.setThinkingLevel(level), + getThinkingLevel: () => record.session.thinkingLevel, + setThinkingLevel: level => record.session.setThinkingLevel(level), }, { - getModel: () => this.#session.model, - getSearchDb: () => this.#session.searchDb, - isIdle: () => !this.#session.isStreaming, + getModel: () => record.session.model, + getSearchDb: () => record.session.searchDb, + isIdle: () => !record.session.isStreaming, abort: () => { - void this.#session.abort(); + void record.session.abort(); }, - hasPendingMessages: () => this.#session.queuedMessageCount > 0, + hasPendingMessages: () => record.session.queuedMessageCount > 0, shutdown: () => {}, - getContextUsage: () => this.#session.getContextUsage(), - getSystemPrompt: () => this.#session.systemPrompt, + getContextUsage: () => record.session.getContextUsage(), + getSystemPrompt: () => record.session.systemPrompt, compact: async instructionsOrOptions => { const instructions = typeof instructionsOrOptions === "string" ? instructionsOrOptions : undefined; const options = instructionsOrOptions && typeof instructionsOrOptions === "object" ? instructionsOrOptions : undefined; - await this.#session.compact(instructions, options); + await record.session.compact(instructions, options); }, }, { - getContextUsage: () => this.#session.getContextUsage(), - waitForIdle: () => this.#session.agent.waitForIdle(), + getContextUsage: () => record.session.getContextUsage(), + waitForIdle: () => record.session.agent.waitForIdle(), newSession: async options => { - const success = await this.#session.newSession({ parentSession: options?.parentSession }); + const success = await record.session.newSession({ parentSession: options?.parentSession }); if (success && options?.setup) { - await options.setup(this.#session.sessionManager); + await options.setup(record.session.sessionManager); } return { cancelled: !success }; }, branch: async entryId => { - const result = await this.#session.branch(entryId); + const result = await record.session.branch(entryId); return { cancelled: result.cancelled }; }, navigateTree: async (targetId, options) => { - const result = await this.#session.navigateTree(targetId, { summarize: options?.summarize }); + const result = await record.session.navigateTree(targetId, { summarize: options?.summarize }); return { cancelled: result.cancelled }; }, switchSession: async sessionPath => { - const success = await this.#session.switchSession(sessionPath); + const success = await record.session.switchSession(sessionPath); return { cancelled: !success }; }, reload: async () => { - await this.#session.reload(); + await record.session.reload(); }, compact: async instructionsOrOptions => { const instructions = typeof instructionsOrOptions === "string" ? instructionsOrOptions : undefined; @@ -881,25 +1203,26 @@ export class AcpAgent implements Agent { instructionsOrOptions && typeof instructionsOrOptions === "object" ? instructionsOrOptions : undefined; - await this.#session.compact(instructions, options); + await record.session.compact(instructions, options); }, }, acpExtensionUiContext, ); await extensionRunner.emit({ type: "session_start" }); + record.extensionsConfigured = true; } - async #configureMcpServers(servers: McpServer[]): Promise { - if (this.#mcpManager) { - await this.#mcpManager.disconnectAll(); + async #configureMcpServers(record: ManagedSessionRecord, servers: McpServer[]): Promise { + if (record.mcpManager) { + await record.mcpManager.disconnectAll(); } if (servers.length === 0) { - this.#mcpManager = undefined; - await this.#session.refreshMCPTools([]); + record.mcpManager = undefined; + await record.session.refreshMCPTools([]); return; } - const manager = new MCPManager(this.#session.sessionManager.getCwd()); + const manager = new MCPManager(record.session.sessionManager.getCwd()); const configs: MCPConfigMap = {}; const sources: MCPSourceMap = {}; for (const server of servers) { @@ -921,8 +1244,8 @@ export class AcpAgent implements Agent { ); } - this.#mcpManager = manager; - await this.#session.refreshMCPTools(result.tools); + record.mcpManager = manager; + await record.session.refreshMCPTools(result.tools); } #toMcpConfig(server: McpServer): MCPServerConfig { @@ -955,4 +1278,84 @@ export class AcpAgent implements Agent { } return mapped; } + + async #closeManagedSession(sessionId: string, record: ManagedSessionRecord): Promise { + this.#sessions.delete(sessionId); + await this.#cancelPromptForClose(record); + await this.#disposeSessionRecord(record); + } + + async #cancelPromptForClose(record: ManagedSessionRecord): Promise { + const promptTurn = record.promptTurn; + if (!promptTurn || promptTurn.settled) { + return; + } + + promptTurn.cancelRequested = true; + promptTurn.unsubscribe?.(); + try { + await record.session.abort(); + } catch (error) { + logger.warn("Failed to abort ACP prompt during session close", { error }); + } + this.#finishPrompt(record, { + stopReason: "cancelled", + usage: this.#buildTurnUsage(promptTurn.usageBaseline, record.session.sessionManager.getUsageStatistics()), + userMessageId: promptTurn.userMessageId, + }); + } + + async #disposeSessionRecord(record: ManagedSessionRecord): Promise { + if (record.mcpManager) { + try { + await record.mcpManager.disconnectAll(); + } catch (error) { + logger.warn("Failed to disconnect ACP MCP servers", { error }); + } + record.mcpManager = undefined; + } + try { + await record.session.dispose(); + } catch (error) { + logger.warn("Failed to dispose ACP session", { error }); + } + } + + async #disposeStandaloneSession(session: AgentSession): Promise { + try { + await session.dispose(); + } catch (error) { + logger.warn("Failed to dispose ACP session", { error }); + } + } + + async #disposeAllSessions(): Promise { + if (this.#disposePromise) { + await this.#disposePromise; + return; + } + + this.#disposePromise = (async () => { + const records = Array.from(this.#sessions.entries()); + this.#sessions.clear(); + await Promise.all( + records.map(async ([sessionId, record]) => { + try { + await this.#cancelPromptForClose(record); + await this.#disposeSessionRecord(record); + } catch (error) { + logger.warn("Failed to clean up ACP session", { sessionId, error }); + } + }), + ); + + const initialSession = this.#initialSession; + this.#initialSession = undefined; + if (initialSession) { + await this.#disposeStandaloneSession(initialSession); + } + })(); + + await this.#disposePromise; + } } diff --git a/packages/coding-agent/src/modes/acp/acp-event-mapper.ts b/packages/coding-agent/src/modes/acp/acp-event-mapper.ts index 9ca2633a3..43a784acf 100644 --- a/packages/coding-agent/src/modes/acp/acp-event-mapper.ts +++ b/packages/coding-agent/src/modes/acp/acp-event-mapper.ts @@ -8,6 +8,10 @@ import type { import type { AgentSessionEvent } from "../../session/agent-session"; import type { TodoStatus } from "../../tools/todo-write"; +interface AcpEventMapperOptions { + getMessageId?: (message: unknown) => string | undefined; +} + interface ContentArrayContainer { content?: unknown; } @@ -118,10 +122,11 @@ export function mapToolKind(toolName: string): ToolKind { export function mapAgentSessionEventToAcpSessionUpdates( event: AgentSessionEvent, sessionId: string, + options: AcpEventMapperOptions = {}, ): SessionNotification[] { switch (event.type) { case "message_update": - return mapAssistantMessageUpdate(event, sessionId); + return mapAssistantMessageUpdate(event, sessionId, options); case "tool_execution_start": { const update: SessionUpdate = { sessionUpdate: "tool_call", @@ -181,6 +186,7 @@ export function mapAgentSessionEventToAcpSessionUpdates( function mapAssistantMessageUpdate( event: Extract, sessionId: string, + options: AcpEventMapperOptions, ): SessionNotification[] { if (!isAssistantMessage(event.message)) { return []; @@ -208,10 +214,12 @@ function mapAssistantMessageUpdate( return []; } + const messageId = options.getMessageId?.(event.message); return [ toSessionNotification(sessionId, { sessionUpdate, content: { type: "text", text }, + messageId, }), ]; } diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index e4bc9d54d..543e19ce8 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -3,11 +3,13 @@ import { AgentSideConnection, ndJsonStream } from "@agentclientprotocol/sdk"; import type { AgentSession } from "../../session/agent-session"; import { AcpAgent } from "./acp-agent"; -export async function runAcpMode(session: AgentSession): Promise { +export type AcpSessionFactory = (cwd: string) => Promise; + +export async function runAcpMode(session: AgentSession, createSession: AcpSessionFactory): Promise { const input = stream.Writable.toWeb(process.stdout); const output = stream.Readable.toWeb(process.stdin); const transport = ndJsonStream(input, output); - const connection = new AgentSideConnection(conn => new AcpAgent(conn, session), transport); + const connection = new AgentSideConnection(conn => new AcpAgent(conn, session, createSession), transport); await connection.closed; process.exit(0); } diff --git a/packages/coding-agent/test/acp-agent.test.ts b/packages/coding-agent/test/acp-agent.test.ts new file mode 100644 index 000000000..f95dc88ce --- /dev/null +++ b/packages/coding-agent/test/acp-agent.test.ts @@ -0,0 +1,382 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import type { AgentSideConnection, PromptRequest, SessionNotification } from "@agentclientprotocol/sdk"; +import type { Model } from "@oh-my-pi/pi-ai"; +import { getConfigRootDir, setAgentDir } from "@oh-my-pi/pi-utils"; +import { AcpAgent } from "../src/modes/acp/acp-agent"; +import type { AgentSession, AgentSessionEvent } from "../src/session/agent-session"; +import { SessionManager } from "../src/session/session-manager"; + +const TEST_MODELS: Model[] = [ + { + id: "claude-sonnet-4-20250514", + name: "Claude Sonnet", + api: "anthropic-messages", + provider: "anthropic", + baseUrl: "https://example.invalid", + reasoning: true, + input: ["text", "image"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8_192, + }, + { + id: "gpt-5.4", + name: "GPT-5.4", + api: "openai-responses", + provider: "openai", + baseUrl: "https://example.invalid", + reasoning: true, + input: ["text", "image"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 200_000, + maxTokens: 8_192, + }, +]; + +function makeAssistantMessage(text: string, thinking?: string) { + const content: Array<{ type: "text"; text: string } | { type: "thinking"; thinking: string }> = [ + { type: "text", text }, + ]; + if (thinking) { + content.push({ type: "thinking" as const, thinking }); + } + return { + role: "assistant" as const, + content, + api: "anthropic-messages" as const, + provider: "anthropic" as const, + model: TEST_MODELS[0].id, + usage: { + input: 10, + output: 5, + cacheRead: 2, + cacheWrite: 1, + totalTokens: 18, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop" as const, + timestamp: Date.now(), + }; +} + +class FakeAgentSession { + sessionManager: SessionManager; + sessionId: string; + agent: { sessionId: string; waitForIdle: () => Promise }; + model: Model | undefined; + thinkingLevel: string | undefined; + customCommands: [] = []; + extensionRunner = undefined; + searchDb = undefined; + isStreaming = false; + queuedMessageCount = 0; + systemPrompt = "system"; + disposed = false; + #listeners = new Set<(event: AgentSessionEvent) => void>(); + + constructor( + cwd: string, + private readonly models: Model[] = TEST_MODELS, + ) { + this.sessionManager = SessionManager.create(cwd); + this.sessionId = this.sessionManager.getSessionId(); + this.agent = { + sessionId: this.sessionId, + waitForIdle: async () => {}, + }; + this.model = models[0]; + } + + get sessionName(): string { + return this.sessionManager.getHeader()?.title ?? `Session ${this.sessionId}`; + } + + get modelRegistry(): { getApiKey: (model: Model) => Promise } { + return { + getApiKey: async (_model: Model) => "test-key", + }; + } + + getAvailableModels(): Model[] { + return this.models; + } + + getAvailableThinkingLevels(): ReadonlyArray { + return ["low", "medium", "high"]; + } + + setThinkingLevel(level: string | undefined): void { + this.thinkingLevel = level; + } + + async setModel(model: Model): Promise { + this.model = model; + } + + subscribe(listener: (event: AgentSessionEvent) => void): () => void { + this.#listeners.add(listener); + return () => { + this.#listeners.delete(listener); + }; + } + + async prompt(text: string): Promise { + this.isStreaming = true; + this.sessionManager.appendMessage({ role: "user", content: text, timestamp: Date.now() }); + const assistantMessage = makeAssistantMessage("pong"); + for (const listener of this.#listeners) { + listener({ + type: "message_update", + message: assistantMessage, + assistantMessageEvent: { type: "text_delta", delta: "pong" }, + } as AgentSessionEvent); + } + this.sessionManager.appendMessage(assistantMessage); + for (const listener of this.#listeners) { + listener({ + type: "agent_end", + messages: [assistantMessage], + } as AgentSessionEvent); + } + this.isStreaming = false; + } + + async abort(): Promise { + this.isStreaming = false; + } + + async refreshMCPTools(_tools: unknown[]): Promise {} + + getContextUsage(): undefined { + return undefined; + } + + async switchSession(sessionPath: string): Promise { + await this.sessionManager.setSessionFile(sessionPath); + this.sessionId = this.sessionManager.getSessionId(); + this.agent.sessionId = this.sessionId; + return true; + } + + async dispose(): Promise { + this.disposed = true; + await this.sessionManager.close(); + } + + async reload(): Promise {} + + async newSession(): Promise { + await this.sessionManager.newSession(); + this.sessionId = this.sessionManager.getSessionId(); + this.agent.sessionId = this.sessionId; + return true; + } + + async branch(_entryId: string): Promise<{ cancelled: boolean }> { + return { cancelled: false }; + } + + async navigateTree(_targetId: string): Promise<{ cancelled: boolean }> { + return { cancelled: false }; + } + + getActiveToolNames(): string[] { + return []; + } + + getAllToolNames(): string[] { + return []; + } + + setActiveToolsByName(_toolNames: string[]): void {} + + async sendCustomMessage(_message: string, _options?: unknown): Promise {} + + async sendUserMessage(_content: string, _options?: unknown): Promise {} + + async compact(_instructions?: string, _options?: unknown): Promise {} +} + +interface AgentHarness { + agent: AcpAgent; + updates: SessionNotification[]; + abortController: AbortController; + sessions: FakeAgentSession[]; + cwdA: string; + cwdB: string; + findSession(sessionId: string): FakeAgentSession | undefined; +} + +function getChunkMessageId(notification: SessionNotification): string | undefined { + const update = notification.update as { messageId?: string | null }; + return typeof update.messageId === "string" ? update.messageId : undefined; +} + +const cleanupRoots: string[] = []; +const originalAgentDir = process.env.PI_CODING_AGENT_DIR; +const fallbackAgentDir = path.join(getConfigRootDir(), "agent"); + +afterEach(async () => { + if (originalAgentDir) { + setAgentDir(originalAgentDir); + } else { + setAgentDir(fallbackAgentDir); + delete process.env.PI_CODING_AGENT_DIR; + } + + for (const root of cleanupRoots.splice(0)) { + await fs.promises.rm(root, { recursive: true, force: true }); + } +}); + +async function createHarness(): Promise { + const root = await fs.promises.mkdtemp(path.join(os.tmpdir(), "omp-acp-test-")); + cleanupRoots.push(root); + const agentDir = path.join(root, "agent"); + const cwdA = path.join(root, "cwd-a"); + const cwdB = path.join(root, "cwd-b"); + await fs.promises.mkdir(agentDir, { recursive: true }); + await fs.promises.mkdir(cwdA, { recursive: true }); + await fs.promises.mkdir(cwdB, { recursive: true }); + setAgentDir(agentDir); + + const updates: SessionNotification[] = []; + const abortController = new AbortController(); + const sessions: FakeAgentSession[] = []; + const connection = { + sessionUpdate: async (notification: SessionNotification) => { + updates.push(notification); + }, + signal: abortController.signal, + closed: Promise.withResolvers().promise, + } as unknown as AgentSideConnection; + + const initialSession = new FakeAgentSession(cwdA); + sessions.push(initialSession); + const factory = async (cwd: string): Promise => { + const session = new FakeAgentSession(cwd); + sessions.push(session); + return session as unknown as AgentSession; + }; + + return { + agent: new AcpAgent(connection, initialSession as unknown as AgentSession, factory), + updates, + abortController, + sessions, + cwdA, + cwdB, + findSession: (sessionId: string) => sessions.find(session => session.sessionId === sessionId), + }; +} + +describe("ACP agent", () => { + it("supports multiple live ACP sessions with model and lifecycle handlers", async () => { + const harness = await createHarness(); + const first = await harness.agent.newSession({ cwd: harness.cwdA, mcpServers: [] }); + const second = await harness.agent.newSession({ cwd: harness.cwdB, mcpServers: [] }); + + expect(first.models?.availableModels.map(model => model.modelId)).toEqual( + TEST_MODELS.map(model => `${model.provider}/${model.id}`), + ); + + await harness.agent.unstable_setSessionModel({ + sessionId: first.sessionId, + modelId: `${TEST_MODELS[1]!.provider}/${TEST_MODELS[1]!.id}`, + }); + await harness.agent.setSessionConfigOption({ + sessionId: first.sessionId, + configId: "thinking", + value: "high", + }); + + const firstSession = harness.findSession(first.sessionId); + const secondSession = harness.findSession(second.sessionId); + expect(firstSession?.model?.id).toBe(TEST_MODELS[1]!.id); + expect(firstSession?.thinkingLevel).toBe("high"); + expect(secondSession?.model?.id).toBe(TEST_MODELS[0]!.id); + expect(secondSession?.thinkingLevel).toBeUndefined(); + + firstSession?.sessionManager.appendMessage({ role: "user", content: "fork me", timestamp: Date.now() }); + await firstSession?.sessionManager.flush(); + + const forked = await harness.agent.unstable_forkSession({ + sessionId: first.sessionId, + cwd: harness.cwdA, + mcpServers: [], + }); + const forkedSession = harness.findSession(forked.sessionId); + const forkedMessages = forkedSession?.sessionManager.buildSessionContext().messages ?? []; + expect(forked.sessionId).not.toBe(first.sessionId); + expect(forkedMessages.some(message => message.role === "user" && message.content === "fork me")).toBe(true); + + await harness.agent.unstable_closeSession({ sessionId: forked.sessionId }); + await expect(harness.agent.setSessionMode({ sessionId: forked.sessionId, modeId: "default" })).rejects.toThrow( + "Unsupported ACP session", + ); + + harness.abortController.abort(); + await Bun.sleep(0); + }); + + it("replays messageIds and returns turn usage for prompts", async () => { + const harness = await createHarness(); + const stored = new FakeAgentSession(harness.cwdA); + harness.sessions.push(stored); + stored.sessionManager.appendMessage({ role: "user", content: "hello", timestamp: Date.now() }); + stored.sessionManager.appendMessage(makeAssistantMessage("reply", "reasoning")); + await stored.sessionManager.ensureOnDisk(); + await stored.sessionManager.flush(); + + await harness.agent.loadSession({ sessionId: stored.sessionId, cwd: harness.cwdA, mcpServers: [] }); + const replayChunks = harness.updates.filter( + update => + update.sessionId === stored.sessionId && + (update.update.sessionUpdate === "user_message_chunk" || + update.update.sessionUpdate === "agent_message_chunk" || + update.update.sessionUpdate === "agent_thought_chunk"), + ); + const replayAssistantChunks = replayChunks.filter( + update => + update.update.sessionUpdate === "agent_message_chunk" || + update.update.sessionUpdate === "agent_thought_chunk", + ); + + expect( + replayChunks.every( + update => typeof getChunkMessageId(update) === "string" && getChunkMessageId(update)!.length > 0, + ), + ).toBe(true); + expect(new Set(replayAssistantChunks.map(update => getChunkMessageId(update))).size).toBe(1); + + const live = await harness.agent.newSession({ cwd: harness.cwdB, mcpServers: [] }); + const response = await harness.agent.prompt({ + sessionId: live.sessionId, + messageId: "05b17a6f-b310-4be7-b767-6b4f3a84eb63", + prompt: [{ type: "text", text: "ping" }], + } as PromptRequest); + + const liveChunks = harness.updates.filter( + update => update.sessionId === live.sessionId && update.update.sessionUpdate === "agent_message_chunk", + ); + expect(response.userMessageId).toBe("05b17a6f-b310-4be7-b767-6b4f3a84eb63"); + expect(response.usage).toEqual({ + inputTokens: 10, + outputTokens: 5, + cachedReadTokens: 2, + cachedWriteTokens: 1, + totalTokens: 18, + }); + expect( + liveChunks.some( + update => typeof getChunkMessageId(update) === "string" && getChunkMessageId(update)!.length > 0, + ), + ).toBe(true); + + harness.abortController.abort(); + await Bun.sleep(0); + }); +}); diff --git a/packages/coding-agent/test/acp-event-mapper.test.ts b/packages/coding-agent/test/acp-event-mapper.test.ts new file mode 100644 index 000000000..a95cc0293 --- /dev/null +++ b/packages/coding-agent/test/acp-event-mapper.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from "bun:test"; +import { mapAgentSessionEventToAcpSessionUpdates } from "../src/modes/acp/acp-event-mapper"; +import type { AgentSessionEvent } from "../src/session/agent-session"; + +function makeAssistantMessage(text: string) { + return { + role: "assistant" as const, + content: [{ type: "text" as const, text }], + api: "anthropic-messages" as const, + provider: "anthropic" as const, + model: "claude-sonnet-4-20250514", + usage: { + input: 10, + output: 5, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 15, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop" as const, + timestamp: Date.now(), + }; +} + +function getChunkMessageId(event: { update: object }): string | undefined { + const update = event.update as { messageId?: string | null }; + return typeof update.messageId === "string" ? update.messageId : undefined; +} + +describe("ACP event mapper", () => { + it("attaches a stable messageId to live assistant chunks", () => { + const assistantMessage = makeAssistantMessage("chunk"); + const getMessageId = (message: unknown): string | undefined => + message === assistantMessage ? "a80f1ff7-4f0a-4e6b-9f09-c94857b62a4a" : undefined; + + const textUpdates = mapAgentSessionEventToAcpSessionUpdates( + { + type: "message_update", + message: assistantMessage, + assistantMessageEvent: { type: "text_delta", delta: "chunk" }, + } as AgentSessionEvent, + "session-1", + { getMessageId }, + ); + const thoughtUpdates = mapAgentSessionEventToAcpSessionUpdates( + { + type: "message_update", + message: assistantMessage, + assistantMessageEvent: { type: "thinking_delta", delta: "plan" }, + } as AgentSessionEvent, + "session-1", + { getMessageId }, + ); + + expect(textUpdates).toHaveLength(1); + expect(thoughtUpdates).toHaveLength(1); + expect(textUpdates[0] ? getChunkMessageId(textUpdates[0]) : undefined).toBe( + "a80f1ff7-4f0a-4e6b-9f09-c94857b62a4a", + ); + expect(thoughtUpdates[0] ? getChunkMessageId(thoughtUpdates[0]) : undefined).toBe( + "a80f1ff7-4f0a-4e6b-9f09-c94857b62a4a", + ); + }); +});