diff --git a/bun.lock b/bun.lock index d507ddf7d..4ec2cc838 100644 --- a/bun.lock +++ b/bun.lock @@ -56,6 +56,7 @@ "omp": "src/cli.ts", }, "dependencies": { + "@agentclientprotocol/sdk": "0.16.1", "@mozilla/readability": "^0.6", "@oh-my-pi/omp-stats": "workspace:*", "@oh-my-pi/pi-agent-core": "workspace:*", @@ -71,6 +72,7 @@ "handlebars": "^4.7", "linkedom": "^0.18", "puppeteer": "^24.37", + "zod": "4.3.6", }, "devDependencies": { "@types/bun": "^1.3", @@ -175,6 +177,8 @@ }, }, "packages": { + "@agentclientprotocol/sdk": ["@agentclientprotocol/sdk@0.16.1", "", { "peerDependencies": { "zod": "^3.25.0 || ^4.0.0" } }, "sha512-1ad+Sc/0sCtZGHthxxvgEUo5Wsbw16I+aF+YwdiLnPwkZG8KAGUEAPK6LM6Pf69lCyJPt1Aomk1d+8oE3C4ZEw=="], + "@anthropic-ai/sdk": ["@anthropic-ai/sdk@0.78.0", "", { "dependencies": { "json-schema-to-ts": "^3.1.1" }, "peerDependencies": { "zod": "^3.25.0 || ^4.0.0" }, "optionalPeers": ["zod"], "bin": { "anthropic-ai-sdk": "bin/cli" } }, "sha512-PzQhR715td/m1UaaN5hHXjYB8Gl2lF9UVhrrGrZeysiF6Rb74Wc9GCB8hzLdzmQtBd1qe89F9OptgB9Za1Ib5w=="], "@aws-crypto/crc32": ["@aws-crypto/crc32@5.2.0", "", { "dependencies": { "@aws-crypto/util": "^5.2.0", "@aws-sdk/types": "^3.222.0", "tslib": "^2.6.2" } }, "sha512-nLbCWqQNgUiwwtFsen1AdzAtvuLRsQS8rYgMuxCrdKf9kOssamGLuPwyTY9wyYblNr9+1XM8v6zoDTPPSIeANg=="], diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index c9a4f6e37..d52f924bc 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -1,6 +1,16 @@ # Changelog ## [Unreleased] +### Added + +- Added ACP (Agent Client Protocol) mode for headless agent operation via `--mode acp` +- Added support for Agent Client Protocol SDK integration with session management, MCP server configuration, and streaming communication +- Added `ensureOnDisk()` method to SessionManager to persist sessions immediately for ACP discovery + +### Changed + +- Changed session collection to include sessions with zero messages, enabling ACP mode to create discoverable sessions immediately +- Changed session persistence logic to use atomic file rewrite when flushing unflushed sessions to prevent duplication ## [13.14.0] - 2026-03-20 diff --git a/packages/coding-agent/package.json b/packages/coding-agent/package.json index db8d41f98..8c01dd0c1 100644 --- a/packages/coding-agent/package.json +++ b/packages/coding-agent/package.json @@ -40,6 +40,7 @@ "test": "bun test" }, "dependencies": { + "@agentclientprotocol/sdk": "0.16.1", "@mozilla/readability": "^0.6", "@oh-my-pi/omp-stats": "workspace:*", "@oh-my-pi/pi-agent-core": "workspace:*", @@ -54,7 +55,8 @@ "diff": "^8.0", "handlebars": "^4.7", "linkedom": "^0.18", - "puppeteer": "^24.37" + "puppeteer": "^24.37", + "zod": "4.3.6" }, "devDependencies": { "@types/bun": "^1.3" diff --git a/packages/coding-agent/src/cli/args.ts b/packages/coding-agent/src/cli/args.ts index 4a3a8d56f..3b5b1392e 100644 --- a/packages/coding-agent/src/cli/args.ts +++ b/packages/coding-agent/src/cli/args.ts @@ -7,7 +7,7 @@ import chalk from "chalk"; import { parseEffort } from "../thinking"; import { BUILTIN_TOOLS } from "../tools"; -export type Mode = "text" | "json" | "rpc"; +export type Mode = "text" | "json" | "rpc" | "acp"; export interface Args { cwd?: string; @@ -67,7 +67,7 @@ export function parseArgs(args: string[], extensionFlags?: Map void, lspServers: Array<{ name: string; status: "ready" | "error"; fileTypes: string[]; error?: string }> | undefined, - mcpManager: import("./mcp").MCPManager | undefined, + mcpManager: MCPManager | undefined, initialMessage?: string, initialImages?: ImageContent[], ): Promise { @@ -717,6 +718,8 @@ export async function runRootCommand(parsed: Args, rawArgs: string[]): Promise 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 new file mode 100644 index 000000000..1b5303ca8 --- /dev/null +++ b/packages/coding-agent/src/modes/acp/acp-agent.ts @@ -0,0 +1,957 @@ +import * as path from "node:path"; +import { + type Agent, + type AgentSideConnection, + type AuthenticateRequest, + type AuthenticateResponse, + type AvailableCommand, + type InitializeRequest, + type InitializeResponse, + type ListSessionsRequest, + type ListSessionsResponse, + type LoadSessionRequest, + type LoadSessionResponse, + type McpServer, + type NewSessionRequest, + type NewSessionResponse, + PROTOCOL_VERSION, + type PromptRequest, + type PromptResponse, + type SessionConfigOption, + type SessionInfo, + type SessionModeState, + type SessionNotification, + type SessionUpdate, + type SetSessionConfigOptionRequest, + type SetSessionConfigOptionResponse, + type SetSessionModeRequest, + type SetSessionModeResponse, +} from "@agentclientprotocol/sdk"; +import type { Model } from "@oh-my-pi/pi-ai"; +import { logger, VERSION } from "@oh-my-pi/pi-utils"; +import type { ExtensionUIContext } from "../../extensibility/extensions"; +import { loadSlashCommands } from "../../extensibility/slash-commands"; +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 { parseThinkingLevel } from "../../thinking"; +import { mapAgentSessionEventToAcpSessionUpdates, mapToolKind } from "./acp-event-mapper"; + +const ACP_MODE_ID = "default"; +const MODE_CONFIG_ID = "mode"; +const MODEL_CONFIG_ID = "model"; +const THINKING_CONFIG_ID = "thinking"; +const THINKING_OFF = "off"; +const SESSION_PAGE_SIZE = 50; + +type AgentImageContent = { + type: "image"; + data: string; + mimeType: string; +}; + +type PromptTurnState = { + messageId: string | null; + cancelRequested: boolean; + settled: boolean; + unsubscribe: (() => void) | undefined; + resolve: (value: PromptResponse) => void; + reject: (reason?: unknown) => void; +}; + +type ReplayableMessage = { + role: string; + content?: unknown; + errorMessage?: string; + toolCallId?: string; + toolName?: string; + details?: unknown; + isError?: boolean; +}; + +type MCPConfigMap = { + [name: string]: MCPServerConfig; +}; + +type MCPSource = { + provider: string; + providerName: string; + path: string; + level: "project"; +}; + +type MCPSourceMap = { + [name: string]: MCPSource; +}; + +const acpExtensionUiContext: ExtensionUIContext = { + select: async () => undefined, + confirm: async () => false, + input: async () => undefined, + notify: (message, type) => { + logger.debug("ACP extension notification", { message, type }); + }, + onTerminalInput: () => () => {}, + setStatus: () => {}, + setWorkingMessage: () => {}, + setWidget: () => {}, + setFooter: () => {}, + setHeader: () => {}, + setTitle: () => {}, + custom: async () => undefined as never, + pasteToEditor: () => {}, + setEditorText: () => {}, + getEditorText: () => "", + editor: async () => undefined, + setEditorComponent: () => {}, + get theme() { + return theme; + }, + getAllThemes: async () => [], + getTheme: async () => undefined, + setTheme: async () => ({ success: false, error: "Theme changes are unavailable in ACP mode" }), + getToolsExpanded: () => false, + setToolsExpanded: () => {}, +}; + +export class AcpAgent implements Agent { + #connection: AgentSideConnection; + #session: AgentSession; + #mcpManager: MCPManager | undefined; + #promptTurn: PromptTurnState | undefined; + #hasOpenedSession = false; + + constructor(connection: AgentSideConnection, session: AgentSession) { + this.#connection = connection; + this.#session = session; + } + + async initialize(_params: InitializeRequest): Promise { + return { + protocolVersion: PROTOCOL_VERSION, + agentInfo: { + name: "oh-my-pi", + title: "Oh My Pi", + version: VERSION, + }, + authMethods: [ + { + id: "agent", + name: "Agent-managed authentication", + description: "Oh My Pi uses its existing local authentication and provider configuration.", + }, + ], + agentCapabilities: { + loadSession: true, + mcpCapabilities: { + http: true, + sse: true, + }, + promptCapabilities: { + embeddedContext: true, + image: true, + }, + sessionCapabilities: { + list: {}, + }, + }, + }; + } + + async authenticate(_params: AuthenticateRequest): Promise { + return {}; + } + + 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 response: NewSessionResponse = { + sessionId: this.#sessionId, + configOptions: this.#buildConfigOptions(), + modes: this.#buildModeState(), + }; + this.#scheduleBootstrapUpdates(this.#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 response: LoadSessionResponse = { + configOptions: this.#buildConfigOptions(), + modes: this.#buildModeState(), + }; + this.#scheduleBootstrapUpdates(this.#sessionId); + return response; + } + + async listSessions(params: ListSessionsRequest): Promise { + if (params.cwd) { + this.#assertAbsoluteCwd(params.cwd); + } + await this.#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); + const nextOffset = offset + paged.length; + return { + sessions: paged.map(session => this.#toSessionInfo(session)), + nextCursor: nextOffset < sessions.length ? String(nextOffset) : undefined, + }; + } + + async setSessionMode(params: SetSessionModeRequest): Promise { + this.#assertSameSession(params.sessionId); + if (params.modeId !== ACP_MODE_ID) { + throw new Error(`Unsupported ACP mode: ${params.modeId}`); + } + await this.#connection.sessionUpdate({ + sessionId: this.#sessionId, + update: this.#buildCurrentModeUpdate(), + }); + return {}; + } + + async setSessionConfigOption(params: SetSessionConfigOptionRequest): Promise { + this.#assertSameSession(params.sessionId); + if (typeof params.value === "boolean") { + throw new Error(`Unsupported boolean ACP config option: ${params.configId}`); + } + + switch (params.configId) { + case MODE_CONFIG_ID: + if (params.value !== ACP_MODE_ID) { + throw new Error(`Unsupported ACP mode config value: ${params.value}`); + } + break; + case MODEL_CONFIG_ID: + await this.#setModelById(params.value); + break; + case THINKING_CONFIG_ID: + this.#setThinkingLevelById(params.value); + break; + default: + throw new Error(`Unknown ACP config option: ${params.configId}`); + } + + const configOptions = this.#buildConfigOptions(); + await this.#connection.sessionUpdate({ + sessionId: this.#sessionId, + update: { + sessionUpdate: "config_option_update", + configOptions, + }, + }); + return { configOptions }; + } + + async prompt(params: PromptRequest): Promise { + this.#assertSameSession(params.sessionId); + if (this.#promptTurn && !this.#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, + cancelRequested: false, + settled: false, + unsubscribe: undefined, + resolve: pendingPrompt.resolve, + reject: pendingPrompt.reject, + }; + + this.#promptTurn.unsubscribe = this.#session.subscribe(event => { + void this.#handlePromptEvent(event); + }); + + this.#session.prompt(converted.text, { images: converted.images }).catch((error: unknown) => { + this.#finishPrompt(undefined, error); + }); + + return await pendingPrompt.promise; + } + + async cancel(params: { sessionId: string }): Promise { + this.#assertSameSession(params.sessionId); + const promptTurn = this.#promptTurn; + if (!promptTurn || promptTurn.settled) { + return; + } + promptTurn.cancelRequested = true; + try { + await this.#session.abort(); + this.#finishPrompt({ + stopReason: "cancelled", + userMessageId: promptTurn.messageId, + }); + } catch (error: unknown) { + this.#finishPrompt(undefined, error); + } + } + + async extMethod(_method: string, _params: { [key: string]: unknown }): Promise<{ [key: string]: unknown }> { + throw new Error("ACP extension methods are not implemented"); + } + + async extNotification(_method: string, _params: { [key: string]: unknown }): Promise {} + + get signal(): AbortSignal { + return this.#connection.signal; + } + + get closed(): Promise { + return this.#connection.closed; + } + + get #sessionId(): string { + return this.#session.sessionId; + } + + async #handlePromptEvent(event: AgentSessionEvent): Promise { + const promptTurn = this.#promptTurn; + if (!promptTurn || promptTurn.settled) { + return; + } + + for (const notification of mapAgentSessionEventToAcpSessionUpdates(event, this.#sessionId)) { + await this.#connection.sessionUpdate(notification); + } + + if (event.type === "agent_end") { + await this.#emitEndOfTurnUpdates(); + this.#finishPrompt({ + stopReason: promptTurn.cancelRequested ? "cancelled" : "end_turn", + userMessageId: promptTurn.messageId, + }); + } + } + + #finishPrompt(response?: PromptResponse, error?: unknown): void { + const promptTurn = this.#promptTurn; + if (!promptTurn || promptTurn.settled) { + return; + } + promptTurn.settled = true; + promptTurn.unsubscribe?.(); + this.#promptTurn = undefined; + if (error !== undefined) { + promptTurn.reject(error); + return; + } + 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}`); + } + } + + #convertPromptBlocks(blocks: PromptRequest["prompt"]): { text: string; images: AgentImageContent[] } { + const textParts: string[] = []; + const images: AgentImageContent[] = []; + for (const block of blocks) { + switch (block.type) { + case "text": + textParts.push(block.text); + break; + case "image": + images.push({ type: "image", data: block.data, mimeType: block.mimeType }); + break; + case "resource": + if ("text" in block.resource) { + textParts.push(block.resource.text); + } else { + textParts.push(`[embedded resource: ${block.resource.uri}]`); + } + break; + case "resource_link": + textParts.push(block.title ?? block.name ?? block.uri); + break; + case "audio": + textParts.push("[audio omitted]"); + break; + } + } + return { + text: textParts.join("\n\n").trim(), + images, + }; + } + + #buildConfigOptions(): SessionConfigOption[] { + const configOptions: SessionConfigOption[] = [ + { + id: MODE_CONFIG_ID, + name: "Mode", + category: "mode", + type: "select", + currentValue: ACP_MODE_ID, + options: [{ value: ACP_MODE_ID, name: "Default", description: "Standard ACP headless mode" }], + }, + ]; + + const models = this.#session.getAvailableModels(); + const currentModel = this.#session.model; + if (models.length > 0) { + configOptions.push({ + id: MODEL_CONFIG_ID, + name: "Model", + category: "model", + type: "select", + currentValue: currentModel ? this.#toModelId(currentModel) : this.#toModelId(models[0]), + options: models.map(model => ({ + value: this.#toModelId(model), + name: model.name, + description: `${model.provider}/${model.id}`, + })), + }); + } + + configOptions.push({ + id: THINKING_CONFIG_ID, + name: "Thinking", + category: "thought_level", + type: "select", + currentValue: this.#toThinkingConfigValue(this.#session.thinkingLevel), + options: this.#buildThinkingOptions(), + }); + return configOptions; + } + + #buildThinkingOptions(): Array<{ value: string; name: string; description?: string }> { + return [ + { value: THINKING_OFF, name: "Off" }, + ...this.#session.getAvailableThinkingLevels().map(level => ({ + value: level, + name: level, + })), + ]; + } + + #toThinkingConfigValue(value: string | undefined): string { + return value && value !== "inherit" ? value : THINKING_OFF; + } + + async #setModelById(modelId: string): Promise { + const model = this.#session.getAvailableModels().find(candidate => this.#toModelId(candidate) === modelId); + if (!model) { + throw new Error(`Unknown ACP model: ${modelId}`); + } + await this.#session.setModel(model); + } + + #setThinkingLevelById(value: string): void { + const thinkingLevel = parseThinkingLevel(value); + if (!thinkingLevel) { + throw new Error(`Unknown ACP thinking level: ${value}`); + } + this.#session.setThinkingLevel(thinkingLevel); + } + + #toModelId(model: Model): string { + return `${model.provider}/${model.id}`; + } + + #buildModeState(): SessionModeState { + return { + availableModes: [{ id: ACP_MODE_ID, name: "Default", description: "Standard ACP headless mode" }], + currentModeId: ACP_MODE_ID, + }; + } + + #buildCurrentModeUpdate(): SessionUpdate { + return { + sessionUpdate: "current_mode_update", + currentModeId: ACP_MODE_ID, + }; + } + + async #buildAvailableCommands(): Promise { + const commands: AvailableCommand[] = []; + const seenNames = new Set(); + const appendCommand = (command: AvailableCommand): void => { + if (seenNames.has(command.name)) { + return; + } + seenNames.add(command.name); + commands.push(command); + }; + + for (const command of this.#session.customCommands) { + appendCommand({ + name: command.command.name, + description: command.command.description, + input: { hint: "arguments" }, + }); + } + + for (const command of await loadSlashCommands({ cwd: this.#session.sessionManager.getCwd() })) { + appendCommand({ + name: command.name, + description: command.description, + }); + } + + return commands; + } + + #toSessionInfo(session: StoredSessionInfo): SessionInfo { + return { + sessionId: session.id, + cwd: session.cwd, + title: session.title, + updatedAt: session.modified.toISOString(), + }; + } + + #scheduleBootstrapUpdates(sessionId: string): void { + setTimeout(() => { + if (sessionId !== this.#sessionId || this.#connection.signal.aborted) { + return; + } + void this.#emitBootstrapUpdates(sessionId); + }, 0); + } + + async #emitBootstrapUpdates(sessionId: string): Promise { + if (sessionId !== this.#sessionId) { + return; + } + await this.#connection.sessionUpdate({ + sessionId, + update: { + sessionUpdate: "available_commands_update", + availableCommands: await this.#buildAvailableCommands(), + }, + }); + await this.#connection.sessionUpdate({ + sessionId, + update: { + sessionUpdate: "session_info_update", + title: this.#session.sessionName, + updatedAt: this.#session.sessionManager.getHeader()?.timestamp, + }, + }); + } + + async #emitEndOfTurnUpdates(): Promise { + const sessionId = this.#sessionId; + + // Emit usage update with context token counts + const contextUsage = this.#session.getContextUsage(); + if (contextUsage) { + const usageStats = this.#session.sessionManager.getUsageStatistics(); + await this.#connection.sessionUpdate({ + sessionId, + update: { + sessionUpdate: "usage_update", + size: contextUsage.contextWindow, + used: contextUsage.tokens ?? 0, + cost: usageStats.cost > 0 ? { amount: usageStats.cost, currency: "USD" } : undefined, + }, + }); + } + + // Push latest session title + await this.#connection.sessionUpdate({ + sessionId, + update: { + sessionUpdate: "session_info_update", + title: this.#session.sessionName, + updatedAt: new Date().toISOString(), + }, + }); + } + + 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()); + } + + async #findStoredSession(sessionId: string, cwd: string): Promise { + const sessions = await this.#listStoredSessions(cwd); + return sessions.find(session => session.id === sessionId); + } + + #parseCursor(cursor: string | undefined): number { + if (!cursor) { + return 0; + } + const parsed = Number.parseInt(cursor, 10); + if (!Number.isFinite(parsed) || parsed < 0) { + throw new Error(`Invalid ACP session cursor: ${cursor}`); + } + return parsed; + } + + async #replaySessionHistory(): Promise { + for (const message of this.#session.sessionManager.buildSessionContext().messages as ReplayableMessage[]) { + for (const notification of this.#messageToReplayNotifications(message)) { + await this.#connection.sessionUpdate(notification); + } + } + } + + #messageToReplayNotifications(message: ReplayableMessage): SessionNotification[] { + if (message.role === "assistant") { + return this.#replayAssistantMessage(message); + } + if ( + message.role === "user" || + message.role === "developer" || + message.role === "custom" || + message.role === "hookMessage" + ) { + return this.#wrapReplayContent(this.#extractReplayContent(message.content, undefined), "user_message_chunk"); + } + if ( + message.role === "toolResult" && + typeof message.toolCallId === "string" && + typeof message.toolName === "string" + ) { + return this.#replayToolResult({ ...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 []; + } + + #replayAssistantMessage(message: ReplayableMessage): SessionNotification[] { + const notifications: SessionNotification[] = []; + const sessionId = this.#sessionId; + if (Array.isArray(message.content)) { + for (const item of message.content) { + if (typeof item !== "object" || item === null || !("type" in item)) { + continue; + } + 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 } }, + }); + continue; + } + if ( + item.type === "thinking" && + "thinking" in item && + typeof item.thinking === "string" && + item.thinking.length > 0 + ) { + notifications.push({ + sessionId, + update: { sessionUpdate: "agent_thought_chunk", content: { type: "text", text: item.thinking } }, + }); + continue; + } + if ( + (item.type === "toolCall" || item.type === "tool_use") && + "id" in item && + typeof item.id === "string" && + "name" in item && + typeof item.name === "string" + ) { + const update: SessionUpdate = { + sessionUpdate: "tool_call", + toolCallId: item.id, + title: item.name, + kind: mapToolKind(item.name), + status: "completed", + }; + if ("arguments" in item && typeof item.arguments === "string") { + update.rawInput = item.arguments; + } + notifications.push({ sessionId, update }); + } + } + } + if (notifications.length === 0 && message.errorMessage) { + notifications.push({ + sessionId, + update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: message.errorMessage } }, + }); + } + return notifications; + } + + #replayToolResult( + message: Required> & ReplayableMessage, + ): SessionNotification[] { + const args = this.#buildReplayToolArgs(message.details); + const startEvent: AgentSessionEvent = { + type: "tool_execution_start", + toolCallId: message.toolCallId, + toolName: message.toolName, + args, + }; + const endEvent: AgentSessionEvent = { + type: "tool_execution_end", + toolCallId: message.toolCallId, + toolName: message.toolName, + isError: message.isError === true, + result: { + content: message.content, + details: message.details, + errorMessage: message.errorMessage, + }, + }; + return [ + ...mapAgentSessionEventToAcpSessionUpdates(startEvent, this.#sessionId), + ...mapAgentSessionEventToAcpSessionUpdates(endEvent, this.#sessionId), + ]; + } + + #buildReplayToolArgs(details: unknown): { path?: string } { + if (typeof details !== "object" || details === null || !("path" in details)) { + return {}; + } + const value = (details as { path?: unknown }).path; + return typeof value === "string" && value.length > 0 ? { path: value } : {}; + } + + #wrapReplayContent( + content: PromptRequest["prompt"], + kind: "agent_message_chunk" | "user_message_chunk", + ): SessionNotification[] { + return content.map(block => ({ + sessionId: this.#sessionId, + update: { + sessionUpdate: kind, + content: block, + }, + })); + } + + #extractReplayContent(content: unknown, errorMessage: string | undefined): PromptRequest["prompt"] { + const replay: PromptRequest["prompt"] = []; + if (Array.isArray(content)) { + for (const item of content) { + if (typeof item !== "object" || item === null || !("type" in item)) { + continue; + } + if (item.type === "text" && "text" in item && typeof item.text === "string" && item.text.length > 0) { + replay.push({ type: "text", text: item.text }); + continue; + } + if ( + item.type === "image" && + "data" in item && + "mimeType" in item && + typeof item.data === "string" && + typeof item.mimeType === "string" + ) { + replay.push({ type: "image", data: item.data, mimeType: item.mimeType }); + } + } + } + if (replay.length === 0 && errorMessage) { + replay.push({ type: "text", text: errorMessage }); + } + return replay; + } + + async #configureExtensions(): Promise { + const extensionRunner = this.#session.extensionRunner; + if (!extensionRunner) { + return; + } + + extensionRunner.initialize( + { + sendMessage: (message, options) => { + this.#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) => { + logger.warn("ACP extension sendUserMessage failed", { error }); + }); + }, + appendEntry: (customType, data) => { + this.#session.sessionManager.appendCustomEntry(customType, data); + }, + setLabel: (targetId, label) => { + this.#session.sessionManager.appendLabelChange(targetId, label); + }, + getActiveTools: () => this.#session.getActiveToolNames(), + getAllTools: () => this.#session.getAllToolNames(), + setActiveTools: toolNames => this.#session.setActiveToolsByName(toolNames), + getCommands: () => [], + setModel: async model => { + const apiKey = await this.#session.modelRegistry.getApiKey(model); + if (!apiKey) { + return false; + } + await this.#session.setModel(model); + return true; + }, + getThinkingLevel: () => this.#session.thinkingLevel, + setThinkingLevel: level => this.#session.setThinkingLevel(level), + }, + { + getModel: () => this.#session.model, + isIdle: () => !this.#session.isStreaming, + abort: () => { + void this.#session.abort(); + }, + hasPendingMessages: () => this.#session.queuedMessageCount > 0, + shutdown: () => {}, + getContextUsage: () => this.#session.getContextUsage(), + getSystemPrompt: () => this.#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); + }, + }, + { + getContextUsage: () => this.#session.getContextUsage(), + waitForIdle: () => this.#session.agent.waitForIdle(), + newSession: async options => { + const success = await this.#session.newSession({ parentSession: options?.parentSession }); + if (success && options?.setup) { + await options.setup(this.#session.sessionManager); + } + return { cancelled: !success }; + }, + branch: async entryId => { + const result = await this.#session.branch(entryId); + return { cancelled: result.cancelled }; + }, + navigateTree: async (targetId, options) => { + const result = await this.#session.navigateTree(targetId, { summarize: options?.summarize }); + return { cancelled: result.cancelled }; + }, + switchSession: async sessionPath => { + const success = await this.#session.switchSession(sessionPath); + return { cancelled: !success }; + }, + reload: async () => { + await this.#session.reload(); + }, + compact: async instructionsOrOptions => { + const instructions = typeof instructionsOrOptions === "string" ? instructionsOrOptions : undefined; + const options = + instructionsOrOptions && typeof instructionsOrOptions === "object" + ? instructionsOrOptions + : undefined; + await this.#session.compact(instructions, options); + }, + }, + acpExtensionUiContext, + ); + await extensionRunner.emit({ type: "session_start" }); + } + + async #configureMcpServers(servers: McpServer[]): Promise { + if (this.#mcpManager) { + await this.#mcpManager.disconnectAll(); + } + if (servers.length === 0) { + this.#mcpManager = undefined; + await this.#session.refreshMCPTools([]); + return; + } + + const manager = new MCPManager(this.#session.sessionManager.getCwd()); + const configs: MCPConfigMap = {}; + const sources: MCPSourceMap = {}; + for (const server of servers) { + configs[server.name] = this.#toMcpConfig(server); + sources[server.name] = { + provider: "acp", + providerName: "ACP Client", + path: `acp://${server.name}`, + level: "project", + }; + } + + const result = await manager.connectServers(configs, sources); + if (result.errors.size > 0) { + throw new Error( + Array.from(result.errors.entries()) + .map(([name, message]) => `${name}: ${message}`) + .join("; "), + ); + } + + this.#mcpManager = manager; + await this.#session.refreshMCPTools(result.tools); + } + + #toMcpConfig(server: McpServer): MCPServerConfig { + if ("command" in server) { + return { + type: "stdio", + command: server.command, + args: server.args, + env: this.#toNameValueMap(server.env), + }; + } + if (server.type === "http") { + return { + type: "http", + url: server.url, + headers: this.#toNameValueMap(server.headers), + }; + } + return { + type: "sse", + url: server.url, + headers: this.#toNameValueMap(server.headers), + }; + } + + #toNameValueMap(values: Array<{ name: string; value: string }>): { [name: string]: string } { + const mapped: { [name: string]: string } = {}; + for (const value of values) { + mapped[value.name] = value.value; + } + return mapped; + } +} diff --git a/packages/coding-agent/src/modes/acp/acp-event-mapper.ts b/packages/coding-agent/src/modes/acp/acp-event-mapper.ts new file mode 100644 index 000000000..de3806d4c --- /dev/null +++ b/packages/coding-agent/src/modes/acp/acp-event-mapper.ts @@ -0,0 +1,531 @@ +import type { + SessionNotification, + SessionUpdate, + ToolCallContent, + ToolCallLocation, + ToolKind, +} from "@agentclientprotocol/sdk"; +import type { AgentSessionEvent } from "../../session/agent-session"; +import type { TodoStatus } from "../../tools/todo-write"; + +interface ContentArrayContainer { + content?: unknown; +} + +interface TypedValue { + type?: unknown; +} + +interface TextLikeContent extends TypedValue { + text?: unknown; +} + +interface BinaryLikeContent extends TypedValue { + data?: unknown; + mimeType?: unknown; +} + +interface PathContainer { + path?: unknown; +} + +interface OldPathContainer { + oldPath?: unknown; +} + +interface NewPathContainer { + newPath?: unknown; +} + +interface CommandContainer { + command?: unknown; +} + +interface PatternContainer { + pattern?: unknown; +} + +interface QueryContainer { + query?: unknown; +} + +interface ErrorMessageContainer { + errorMessage?: unknown; +} + +interface MessageContainer { + message?: unknown; +} + +interface ResourceLinkLikeContent extends TypedValue { + uri?: unknown; + name?: unknown; + title?: unknown; + description?: unknown; + mimeType?: unknown; + size?: unknown; +} + +interface BlobResourceLike { + uri?: unknown; + blob?: unknown; + mimeType?: unknown; +} + +interface TextResourceLike { + uri?: unknown; + text?: unknown; + mimeType?: unknown; +} + +interface EmbeddedResourceLikeContent extends TypedValue { + resource?: unknown; +} + +interface TextMessageLike { + role?: unknown; +} + +const ACP_TEXT_LIMIT = 4_000; + +export function mapToolKind(toolName: string): ToolKind { + switch (toolName) { + case "read": + return "read"; + case "write": + case "edit": + return "edit"; + case "delete": + return "delete"; + case "move": + return "move"; + case "bash": + case "python": + return "execute"; + case "grep": + case "find": + case "ast_grep": + return "search"; + case "fetch": + case "web_search": + return "fetch"; + case "todo_write": + return "think"; + default: + return "other"; + } +} + +export function mapAgentSessionEventToAcpSessionUpdates( + event: AgentSessionEvent, + sessionId: string, +): SessionNotification[] { + switch (event.type) { + case "message_update": + return mapAssistantMessageUpdate(event, sessionId); + case "tool_execution_start": { + const update: SessionUpdate = { + sessionUpdate: "tool_call", + toolCallId: event.toolCallId, + title: buildToolTitle(event.toolName, event.args, event.intent), + kind: mapToolKind(event.toolName), + status: "pending", + rawInput: event.args, + }; + const locations = extractToolLocations(event.args); + if (locations.length > 0) { + update.locations = locations; + } + return [toSessionNotification(sessionId, update)]; + } + case "tool_execution_update": { + const content = extractToolCallContent(event.partialResult); + const update: SessionUpdate = { + sessionUpdate: "tool_call_update", + toolCallId: event.toolCallId, + status: "in_progress", + rawOutput: event.partialResult, + }; + if (content.length > 0) { + update.content = content; + } + return [toSessionNotification(sessionId, update)]; + } + case "tool_execution_end": { + const content = extractToolCallContent(event.result); + const update: SessionUpdate = { + sessionUpdate: "tool_call_update", + toolCallId: event.toolCallId, + status: event.isError ? "failed" : "completed", + rawOutput: event.result, + }; + if (content.length > 0) { + update.content = content; + } + return [toSessionNotification(sessionId, update)]; + } + case "todo_reminder": { + const entries = event.todos.map(todo => ({ + content: todo.content, + priority: "medium" as const, + status: mapTodoStatus(todo.status), + })); + return [toSessionNotification(sessionId, { sessionUpdate: "plan", entries })]; + } + case "todo_auto_clear": + return [toSessionNotification(sessionId, { sessionUpdate: "plan", entries: [] })]; + default: + return []; + } +} + +function mapAssistantMessageUpdate( + event: Extract, + sessionId: string, +): SessionNotification[] { + if (!isAssistantMessage(event.message)) { + return []; + } + + let sessionUpdate: "agent_message_chunk" | "agent_thought_chunk"; + let text: string; + switch (event.assistantMessageEvent.type) { + case "text_delta": + sessionUpdate = "agent_message_chunk"; + text = event.assistantMessageEvent.delta; + break; + case "thinking_delta": + sessionUpdate = "agent_thought_chunk"; + text = event.assistantMessageEvent.delta; + break; + case "error": + sessionUpdate = "agent_message_chunk"; + text = event.assistantMessageEvent.error.errorMessage ?? "Unknown error"; + break; + default: + return []; + } + if (text.length === 0) { + return []; + } + + return [ + toSessionNotification(sessionId, { + sessionUpdate, + content: { type: "text", text }, + }), + ]; +} + +function toSessionNotification(sessionId: string, update: SessionUpdate): SessionNotification { + return { sessionId, update }; +} + +const todoStatusMap: Record = { + pending: "pending", + in_progress: "in_progress", + completed: "completed", + abandoned: "completed", +}; + +function mapTodoStatus(status: TodoStatus): "pending" | "in_progress" | "completed" { + return todoStatusMap[status]; +} + +function buildToolTitle(toolName: string, args: unknown, intent: string | undefined): string { + const trimmedIntent = intent?.trim(); + if (trimmedIntent) { + return trimmedIntent; + } + + const subject = + extractStringProperty(args, "path") ?? + extractStringProperty(args, "command") ?? + extractStringProperty(args, "pattern") ?? + extractStringProperty(args, "query"); + if (subject) { + return `${toolName}: ${subject}`; + } + + return toolName; +} + +function extractToolLocations(args: unknown): ToolCallLocation[] { + const locations: ToolCallLocation[] = []; + const path = extractStringProperty(args, "path"); + if (path) { + locations.push({ path }); + } + + const oldPath = extractStringProperty(args, "oldPath"); + if (oldPath && oldPath !== path) { + locations.push({ path: oldPath }); + } + + const newPath = extractStringProperty(args, "newPath"); + if (newPath && newPath !== path && newPath !== oldPath) { + locations.push({ path: newPath }); + } + + return locations; +} + +function extractToolCallContent(value: unknown): ToolCallContent[] { + const richContent = extractStructuredToolCallContent(value); + const fallbackText = extractReadableText(value); + if (!fallbackText) { + return richContent; + } + if (hasEquivalentTextContent(richContent, fallbackText)) { + return richContent; + } + return [...richContent, textToolCallContent(fallbackText)]; +} + +function extractStructuredToolCallContent(value: unknown): ToolCallContent[] { + const blocks = getContentBlocks(value); + if (!blocks) { + return []; + } + + const content: ToolCallContent[] = []; + for (const block of blocks) { + const toolCallContent = toToolCallContent(block); + if (toolCallContent) { + content.push(toolCallContent); + } + } + return content; +} + +function getContentBlocks(value: unknown): unknown[] | undefined { + if (Array.isArray(value)) { + return value; + } + if (typeof value !== "object" || value === null || !("content" in value)) { + return undefined; + } + const content = (value as ContentArrayContainer).content; + return Array.isArray(content) ? content : undefined; +} + +function toToolCallContent(value: unknown): ToolCallContent | undefined { + const type = getContentType(value); + if (!type) { + return undefined; + } + + switch (type) { + case "text": { + const text = extractStructuredText(value); + return text ? textToolCallContent(text) : undefined; + } + case "image": + case "audio": { + const data = extractStringProperty(value, "data"); + const mimeType = extractStringProperty(value, "mimeType"); + if (!data || !mimeType) { + return undefined; + } + return { + type: "content", + content: { + type, + data, + mimeType, + }, + }; + } + case "resource_link": { + const uri = extractStringProperty(value, "uri"); + const name = extractStringProperty(value, "name"); + if (!uri || !name) { + return undefined; + } + const resourceLinkContent: { + type: "resource_link"; + uri: string; + name: string; + title?: string; + description?: string; + mimeType?: string; + size?: number; + } = { + type: "resource_link", + uri, + name, + }; + const title = extractStringProperty(value, "title"); + if (title) { + resourceLinkContent.title = title; + } + const description = extractStringProperty(value, "description"); + if (description) { + resourceLinkContent.description = description; + } + const mimeType = extractStringProperty(value, "mimeType"); + if (mimeType) { + resourceLinkContent.mimeType = mimeType; + } + const size = extractNumberProperty(value, "size"); + if (size !== undefined) { + resourceLinkContent.size = size; + } + return { + type: "content", + content: resourceLinkContent, + }; + } + case "resource": { + const resource = extractEmbeddedResource(value); + return resource + ? { + type: "content", + content: { + type: "resource", + resource, + }, + } + : undefined; + } + default: + return undefined; + } +} + +function extractEmbeddedResource( + value: unknown, +): { uri: string; text: string; mimeType?: string } | { uri: string; blob: string; mimeType?: string } | undefined { + if (typeof value !== "object" || value === null || !("resource" in value)) { + return undefined; + } + + const resource = (value as EmbeddedResourceLikeContent).resource; + if (typeof resource !== "object" || resource === null) { + return undefined; + } + + const uri = extractStringProperty(resource, "uri"); + if (!uri) { + return undefined; + } + + const text = extractStringProperty(resource, "text"); + if (text) { + const mimeType = extractStringProperty(resource, "mimeType"); + return mimeType ? { uri, text, mimeType } : { uri, text }; + } + + const blob = extractStringProperty(resource, "blob"); + if (!blob) { + return undefined; + } + const mimeType = extractStringProperty(resource, "mimeType"); + return mimeType ? { uri, blob, mimeType } : { uri, blob }; +} + +function textToolCallContent(text: string): ToolCallContent { + return { + type: "content", + content: { + type: "text", + text, + }, + }; +} + +function hasEquivalentTextContent(content: ToolCallContent[], text: string): boolean { + return content.some(item => item.type === "content" && item.content.type === "text" && item.content.text === text); +} + +function extractReadableText(value: unknown): string | undefined { + if (typeof value === "string") { + return normalizeText(value); + } + if (value instanceof Error) { + return normalizeText(value.message); + } + if (typeof value !== "object" || value === null) { + return undefined; + } + + const directText = + extractStringProperty(value, "text") ?? + extractStringProperty(value, "errorMessage") ?? + extractStringProperty(value, "message"); + if (directText) { + return normalizeText(directText); + } + + const contentBlocks = getContentBlocks(value); + if (contentBlocks) { + const text = contentBlocks + .map(block => extractStructuredText(block)) + .filter((chunk): chunk is string => typeof chunk === "string" && chunk.length > 0) + .join("\n"); + if (text.length > 0) { + return normalizeText(text); + } + } + + const serialized = safeJsonStringify(value); + return normalizeText(serialized); +} + +function extractStructuredText(value: unknown): string | undefined { + const text = extractStringProperty(value, "text"); + if (!text) { + return undefined; + } + return limitText(text); +} + +function getContentType(value: unknown): string | undefined { + if (typeof value !== "object" || value === null || !("type" in value)) { + return undefined; + } + const type = (value as TypedValue).type; + return typeof type === "string" ? type : undefined; +} + +function extractStringProperty(value: unknown, key: keyof T): string | undefined { + if (typeof value !== "object" || value === null || !(key in value)) { + return undefined; + } + const property = (value as T)[key]; + return typeof property === "string" && property.length > 0 ? property : undefined; +} + +function extractNumberProperty(value: unknown, key: keyof T): number | undefined { + if (typeof value !== "object" || value === null || !(key in value)) { + return undefined; + } + const property = (value as T)[key]; + return typeof property === "number" && Number.isFinite(property) ? property : undefined; +} + +function isAssistantMessage(value: unknown): boolean { + return ( + typeof value === "object" && value !== null && "role" in value && (value as TextMessageLike).role === "assistant" + ); +} + +function normalizeText(text: string | undefined): string | undefined { + if (!text) { + return undefined; + } + const normalized = text.trim(); + return normalized.length > 0 ? limitText(normalized) : undefined; +} + +function limitText(text: string): string { + return text.length > ACP_TEXT_LIMIT ? `${text.slice(0, ACP_TEXT_LIMIT - 1)}…` : text; +} + +function safeJsonStringify(value: unknown): string | undefined { + try { + return JSON.stringify(value); + } catch { + return undefined; + } +} diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts new file mode 100644 index 000000000..e4bc9d54d --- /dev/null +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -0,0 +1,13 @@ +import * as stream from "node:stream"; +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 { + 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); + await connection.closed; + process.exit(0); +} diff --git a/packages/coding-agent/src/modes/acp/index.ts b/packages/coding-agent/src/modes/acp/index.ts new file mode 100644 index 000000000..1b0d975b0 --- /dev/null +++ b/packages/coding-agent/src/modes/acp/index.ts @@ -0,0 +1,2 @@ +export * from "./acp-agent"; +export * from "./acp-mode"; diff --git a/packages/coding-agent/src/modes/index.ts b/packages/coding-agent/src/modes/index.ts index 677b2cab3..143261ff1 100644 --- a/packages/coding-agent/src/modes/index.ts +++ b/packages/coding-agent/src/modes/index.ts @@ -4,6 +4,7 @@ import { postmortem } from "@oh-my-pi/pi-utils"; /** * Run modes for the coding agent. */ +export { runAcpMode } from "./acp"; export { InteractiveMode, type InteractiveModeOptions } from "./interactive-mode"; export { type PrintModeOptions, runPrintMode } from "./print-mode"; export { type ModelInfo, RpcClient, type RpcClientOptions, type RpcEventListener } from "./rpc/rpc-client"; diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 297b77c7d..edcb16b1f 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -46,6 +46,7 @@ import { calculateRateLimitBackoffMs, getSupportedEfforts, isContextOverflow, + isUsageLimitError, modelsAreEqual, parseRateLimitReason, } from "@oh-my-pi/pi-ai"; @@ -4476,8 +4477,9 @@ export class AgentSession { // ========================================================================= /** - * Check if an error is retryable (overloaded, rate limit, server errors). + * Check if an error is retryable (transient errors or usage limits). * Context overflow errors are NOT retryable (handled by compaction instead). + * Usage-limit errors are retryable because the retry handler performs credential switching. */ #isRetryableError(message: AssistantMessage): boolean { if (message.stopReason !== "error" || !message.errorMessage) return false; @@ -4487,20 +4489,15 @@ export class AgentSession { if (isContextOverflow(message, contextWindow)) return false; const err = message.errorMessage; - return this.#isRetryableErrorMessage(err); + return this.#isTransientErrorMessage(err) || isUsageLimitError(err); } - #isRetryableErrorMessage(errorMessage: string): boolean { - // Match: overloaded_error, rate limit, usage limit, 429, 500, 502, 503, 504, service unavailable, connection error, fetch failed, retry delay exceeded, stream stall - return /overloaded|rate.?limit|usage.?limit|too many requests|429|500|502|503|504|service.?unavailable|server error|internal error|connection.?error|unable to connect|fetch failed|retry delay|stream stall/i.test( + #isTransientErrorMessage(errorMessage: string): boolean { + return /overloaded|rate.?limit|too many requests|429|500|502|503|504|service.?unavailable|server error|internal error|connection.?error|unable to connect|fetch failed|retry delay|stream stall/i.test( errorMessage, ); } - #isUsageLimitErrorMessage(errorMessage: string): boolean { - return /usage.?limit|usage_limit_reached|limit_reached|quota.?exceeded|resource.?exhausted/i.test(errorMessage); - } - #parseRetryAfterMsFromError(errorMessage: string): number | undefined { const now = Date.now(); const retryAfterMsMatch = /retry-after-ms\s*[:=]\s*(\d+)/i.exec(errorMessage); diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index 99f932504..36302e806 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -1302,21 +1302,19 @@ async function collectSessionsFromFiles(files: string[], storage: SessionStorage } } - if (messageCount) { - const stats = storage.statSync(file); - sessions.push({ - path: file, - id: header.id, - cwd: typeof header.cwd === "string" ? header.cwd : "", - title: header.title ?? shortSummary, - parentSessionPath: (header as SessionHeader).parentSession, - created: new Date(header.timestamp), - modified: stats.mtime, - messageCount, - firstMessage: firstMessage || "(no messages)", - allMessagesText: allMessages.join(" "), - }); - } + const stats = storage.statSync(file); + sessions.push({ + path: file, + id: header.id, + cwd: typeof header.cwd === "string" ? header.cwd : "", + title: header.title ?? shortSummary, + parentSessionPath: (header as SessionHeader).parentSession, + created: new Date(header.timestamp), + modified: stats.mtime, + messageCount, + firstMessage: firstMessage || "(no messages)", + allMessagesText: allMessages.join(" "), + }); } catch {} }), ); @@ -1782,6 +1780,16 @@ export class SessionManager { return this.persist; } + /** + * Force-persist all current entries to disk, even when no assistant message exists yet. + * Used by ACP mode where session/new must create a discoverable session immediately. + */ + async ensureOnDisk(): Promise { + if (!this.persist || !this.#sessionFile) return; + if (this.#flushed) return; + await this.#rewriteFile(); + } + /** Flush pending writes to disk. Call before switching sessions or on shutdown. */ async flush(): Promise { await this.#queuePersistTask(async () => { @@ -1917,17 +1925,9 @@ export class SessionManager { } if (!this.#flushed) { - this.#flushed = true; - void this.#queuePersistTask(async () => { - const writer = this.#ensurePersistWriter(); - if (!writer) return; - const entries = await Promise.all( - this.#fileEntries.map(e => prepareEntryForPersistence(e, this.#blobStore)), - ); - for (const persistedEntry of entries) { - await writer.write(persistedEntry); - } - }); + // Full flush: rewrite the entire file atomically to avoid + // duplicating entries if the file already exists (e.g. from ensureOnDisk). + void this.#rewriteFile(); } else { void this.#queuePersistTask(async () => { const writer = this.#ensurePersistWriter();