diff --git a/Packages/Sources/RxCodeCore/Backend/AgentBackend.swift b/Packages/Sources/RxCodeCore/Backend/AgentBackend.swift new file mode 100644 index 00000000..2a95614f --- /dev/null +++ b/Packages/Sources/RxCodeCore/Backend/AgentBackend.swift @@ -0,0 +1,90 @@ +import Foundation + +/// Unified send envelope passed to any `AgentBackend`. Carries the superset +/// of parameters the three current backends need; each adapter consumes only +/// the fields it cares about. +public struct BackendSendRequest: Sendable { + public let streamId: UUID + public let prompt: String + public let cwd: String + /// CLI session id (Claude) / thread id (Codex) / ACP session id, depending + /// on the backend. `nil` means "start a new session". + public let sessionId: String? + public let model: String? + public let effort: String? + public let permissionMode: PermissionMode + public let planMode: Bool + /// Path to the Claude hook settings JSON written for this turn. Claude-only. + public let hookSettingsPath: String? + /// Path to the Claude MCP config JSON written for this turn. Claude-only. + public let mcpClaudeConfigPath: String? + /// `-c` overrides handed to the Codex app-server child. Codex-only. + public let mcpCodexOverrides: [String] + /// JSON-RPC payload for ACP's `session/new` `mcpServers` parameter. + /// ACP-only. Each element is an object with `name`/`command`/`args`/`env`. + public let acpMCPServers: [JSONValue] + /// Resolved client spec when `provider == .acp`. Ignored otherwise. + public let acpSpec: ACPClientSpec? + /// AppState's internal session key for the pooled ACP entry. + public let clientSessionKey: String + + public init( + streamId: UUID, + prompt: String, + cwd: String, + sessionId: String?, + model: String?, + effort: String? = nil, + permissionMode: PermissionMode, + planMode: Bool = false, + hookSettingsPath: String? = nil, + mcpClaudeConfigPath: String? = nil, + mcpCodexOverrides: [String] = [], + acpMCPServers: [JSONValue] = [], + acpSpec: ACPClientSpec? = nil, + clientSessionKey: String + ) { + self.streamId = streamId + self.prompt = prompt + self.cwd = cwd + self.sessionId = sessionId + self.model = model + self.effort = effort + self.permissionMode = permissionMode + self.planMode = planMode + self.hookSettingsPath = hookSettingsPath + self.mcpClaudeConfigPath = mcpClaudeConfigPath + self.mcpCodexOverrides = mcpCodexOverrides + self.acpMCPServers = acpMCPServers + self.acpSpec = acpSpec + self.clientSessionKey = clientSessionKey + } +} + +/// Common surface for every agent transport (Claude CLI, Codex app-server, +/// ACP). AppState dispatches via this protocol instead of switching on +/// `AgentProvider` directly. +/// +/// Lifecycle per turn: +/// 1. AppState builds a `BackendSendRequest` and calls `send(_:)`. +/// 2. The returned `AsyncStream` carries unified events. +/// 3. On natural completion AppState calls `finalize(streamId:)`. +/// 4. On user-initiated stop AppState calls `cancel(streamId:)`. +public protocol AgentBackend: Actor { + nonisolated var provider: AgentProvider { get } + nonisolated var staticCapabilities: CapabilitySet { get } + + /// Runtime override hook. Phase 1 returns `staticCapabilities`; Phase 2+ + /// can refine after the agent's handshake reports its real toolset. + func capabilities(for sessionKey: String) async -> CapabilitySet + + func send(_ request: BackendSendRequest) -> AsyncStream + func cancel(streamId: UUID) + func finalize(streamId: UUID) +} + +public extension AgentBackend { + func capabilities(for sessionKey: String) async -> CapabilitySet { + staticCapabilities + } +} diff --git a/Packages/Sources/RxCodeCore/Backend/BackendCapability.swift b/Packages/Sources/RxCodeCore/Backend/BackendCapability.swift new file mode 100644 index 00000000..636518f7 --- /dev/null +++ b/Packages/Sources/RxCodeCore/Backend/BackendCapability.swift @@ -0,0 +1,42 @@ +import Foundation + +/// Editor-facing features an agent backend may natively support. A backend +/// that lacks a capability gets it polyfilled via the IDE-side MCP server +/// (Phase 2+) so the user-visible behavior is uniform across providers. +public enum BackendCapability: String, Sendable, Hashable, CaseIterable, Codable { + case askUserQuestion + case todos + case planMode + case fileEdit + case getUsage + case customSlashCommands + case attachments + case hooks + case mcpServers +} + +public typealias CapabilitySet = Set + +public extension AgentProvider { + /// Compile-time defaults. Phase 1 ships with these; backends may later + /// override at runtime via `AgentBackend.capabilities(for:)` once handshake + /// negotiation lands (Phase 2+). + var staticCapabilities: CapabilitySet { + switch self { + case .claudeCode: + return [ + .askUserQuestion, .todos, .planMode, .fileEdit, .hooks, + .mcpServers, .attachments, .customSlashCommands, .getUsage, + ] + case .codex: + return [ + .askUserQuestion, .todos, .planMode, .fileEdit, + .mcpServers, .attachments, .getUsage, + ] + case .acp: + return [ + .planMode, .fileEdit, .mcpServers, .attachments, .getUsage, + ] + } + } +} diff --git a/Packages/Sources/RxCodeCore/Backend/IDEToolHandling.swift b/Packages/Sources/RxCodeCore/Backend/IDEToolHandling.swift new file mode 100644 index 00000000..f52fef7c --- /dev/null +++ b/Packages/Sources/RxCodeCore/Backend/IDEToolHandling.swift @@ -0,0 +1,30 @@ +import Foundation + +/// Anything that can serve IDE-side MCP tool calls. AppState conforms in +/// the host target so the bridge listener stays decoupled from UI state. +/// +/// All entry points are isolated to the conforming actor; the bridge hops +/// to the actor before invoking. Handlers may freely touch UI state. +public protocol IDEToolHandling: AnyObject, Sendable { + /// Tools exposed to the agent for `sessionKey`. The bridge calls this + /// at `tools/list` time. Implementations typically derive the list from + /// `IDEToolRegistry.tools(for:)` minus any per-session opt-outs. + func ideAvailableTools(forSession sessionKey: String) async -> [IDETool] + + /// Dispatch a `tools/call`. Returns the tool result as a `JSONValue` + /// (typically `.object(["content": .array([...]) ])` per the MCP spec) + /// or throws an `IDEToolError`. The bridge translates throws into MCP + /// error responses. + func ideHandleToolCall( + name: String, + arguments: JSONValue, + sessionKey: String + ) async throws -> JSONValue +} + +public enum IDEToolError: Error, Sendable { + case unknownTool(String) + case invalidArguments(String) + case notSupported(String) + case handlerFailed(String) +} diff --git a/Packages/Sources/RxCodeCore/Backend/IDEToolRegistry.swift b/Packages/Sources/RxCodeCore/Backend/IDEToolRegistry.swift new file mode 100644 index 00000000..185899c1 --- /dev/null +++ b/Packages/Sources/RxCodeCore/Backend/IDEToolRegistry.swift @@ -0,0 +1,161 @@ +import Foundation + +/// Catalog of IDE-side MCP tools the editor can expose to agent backends. +/// Tools are gated by the *complement* of the backend's capability set: +/// if a backend natively supports a feature, the corresponding polyfill +/// tool is NOT registered for that session. Tools tagged `.alwaysIDEOnly` +/// are exposed to every backend regardless of capabilities — they expose +/// IDE-only state (running jobs, thread history, usage) that has no +/// equivalent in any agent's native toolset. +public struct IDETool: Sendable { + public enum Visibility: Sendable { + /// Exposed only when the backend lacks the matched capability. + case polyfill(BackendCapability) + /// Exposed unconditionally — agent-native equivalents don't exist. + case alwaysIDEOnly + } + + public let name: String + public let description: String + public let visibility: Visibility + /// JSON Schema for the tool's input parameters, serialized as a + /// `JSONValue` object. Consumed by the MCP server's `tools/list` handler. + public let inputSchema: JSONValue + + public init(name: String, description: String, visibility: Visibility, inputSchema: JSONValue) { + self.name = name + self.description = description + self.visibility = visibility + self.inputSchema = inputSchema + } +} + +public enum IDEToolRegistry { + public static let allTools: [IDETool] = [ + IDETool( + name: "ide__ask_user", + description: "Ask the user a question and wait for their selection. Use when you need clarification or a decision the user must make.", + visibility: .polyfill(.askUserQuestion), + inputSchema: .object([ + "type": .string("object"), + "properties": .object([ + "question": .object([ + "type": .string("string"), + "description": .string("The question to show the user."), + ]), + "options": .object([ + "type": .string("array"), + "items": .object(["type": .string("string")]), + "description": .string("2–4 mutually-exclusive options the user can pick from."), + ]), + "allow_multiple": .object([ + "type": .string("boolean"), + "description": .string("Whether more than one option may be selected."), + ]), + ]), + "required": .array([.string("question"), .string("options")]), + ]) + ), + IDETool( + name: "ide__set_todos", + description: "Replace the current todo list shown in the IDE sidebar. Use to track multi-step work.", + visibility: .polyfill(.todos), + inputSchema: .object([ + "type": .string("object"), + "properties": .object([ + "todos": .object([ + "type": .string("array"), + "items": .object([ + "type": .string("object"), + "properties": .object([ + "content": .object(["type": .string("string")]), + "status": .object([ + "type": .string("string"), + "enum": .array([ + .string("pending"), + .string("in_progress"), + .string("completed"), + ]), + ]), + ]), + "required": .array([.string("content"), .string("status")]), + ]), + ]), + ]), + "required": .array([.string("todos")]), + ]) + ), + IDETool( + name: "ide__get_running_jobs", + description: "List run-profile jobs (dev servers, scripts) currently executing in the IDE.", + visibility: .alwaysIDEOnly, + inputSchema: .object([ + "type": .string("object"), + "properties": .object([:]), + ]) + ), + IDETool( + name: "ide__get_job_output", + description: "Fetch the tail of a running job's output buffer.", + visibility: .alwaysIDEOnly, + inputSchema: .object([ + "type": .string("object"), + "properties": .object([ + "job_id": .object(["type": .string("string")]), + "limit": .object([ + "type": .string("integer"), + "description": .string("Maximum lines to return from the tail. Defaults to 200."), + ]), + ]), + "required": .array([.string("job_id")]), + ]) + ), + IDETool( + name: "ide__get_threads", + description: "List chat threads in the current project (or all projects if none specified).", + visibility: .alwaysIDEOnly, + inputSchema: .object([ + "type": .string("object"), + "properties": .object([ + "project_id": .object(["type": .string("string")]), + ]), + ]) + ), + IDETool( + name: "ide__get_thread_detail", + description: "Fetch the message history of a specific thread by id.", + visibility: .alwaysIDEOnly, + inputSchema: .object([ + "type": .string("object"), + "properties": .object([ + "thread_id": .object(["type": .string("string")]), + ]), + "required": .array([.string("thread_id")]), + ]) + ), + IDETool( + name: "ide__get_usage", + description: "Get current rate-limit / token usage stats reported by the active provider.", + visibility: .alwaysIDEOnly, + inputSchema: .object([ + "type": .string("object"), + "properties": .object([:]), + ]) + ), + ] + + /// Returns the tools that should be exposed to an agent whose declared + /// capabilities are `capabilities`. Polyfill tools whose capability is + /// already covered natively are filtered out; `alwaysIDEOnly` tools + /// always pass through. + public static func tools(for capabilities: CapabilitySet) -> [IDETool] { + allTools.filter { tool in + switch tool.visibility { + case .polyfill(let cap): + return !capabilities.contains(cap) + case .alwaysIDEOnly: + return true + } + } + } +} diff --git a/RxCode.xcodeproj/project.pbxproj b/RxCode.xcodeproj/project.pbxproj index c420486b..4a182baa 100644 --- a/RxCode.xcodeproj/project.pbxproj +++ b/RxCode.xcodeproj/project.pbxproj @@ -14,6 +14,7 @@ DFA0CCD12FB4CC01005991E1 /* PlanDecisionTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = DFA0CCC02FB4CC01005991E1 /* PlanDecisionTests.swift */; }; DFA0CCD22FB4CC01005991E1 /* PlanCardViewTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = DFA0CCC12FB4CC01005991E1 /* PlanCardViewTests.swift */; }; DFA0CCD42FB4CC01005991E1 /* RxCodeChatKit in Frameworks */ = {isa = PBXBuildFile; productRef = DFA0CCC32FB4CC01005991E1 /* RxCodeChatKit */; }; + DFA0CCE12FB4CC02005991E1 /* HistoryListArchiveFilterTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = DFA0CCD52FB4CC02005991E1 /* HistoryListArchiveFilterTests.swift */; }; E6821AC12F7CEE7200829FC9 /* SwiftTerm in Frameworks */ = {isa = PBXBuildFile; productRef = E6A001012F8A000100000001 /* SwiftTerm */; }; E6C001022F9B000100000001 /* Sparkle in Frameworks */ = {isa = PBXBuildFile; productRef = E6C001012F9B000100000001 /* Sparkle */; }; E6D001032FA0000100000001 /* RxCodeCore in Frameworks */ = {isa = PBXBuildFile; productRef = E6D001012FA0000100000001 /* RxCodeCore */; }; @@ -36,6 +37,7 @@ A9993BB72A5307039A88B729 /* Cocoa.framework */ = {isa = PBXFileReference; lastKnownFileType = wrapper.framework; name = Cocoa.framework; path = Platforms/MacOSX.platform/Developer/SDKs/MacOSX15.0.sdk/System/Library/Frameworks/Cocoa.framework; sourceTree = DEVELOPER_DIR; }; DFA0CCC02FB4CC01005991E1 /* PlanDecisionTests.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = PlanDecisionTests.swift; sourceTree = ""; }; DFA0CCC12FB4CC01005991E1 /* PlanCardViewTests.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = PlanCardViewTests.swift; sourceTree = ""; }; + DFA0CCD52FB4CC02005991E1 /* HistoryListArchiveFilterTests.swift */ = {isa = PBXFileReference; includeInIndex = 1; lastKnownFileType = sourcecode.swift; path = HistoryListArchiveFilterTests.swift; sourceTree = ""; }; E67335382F7356F600FD26C7 /* RxCode.app */ = {isa = PBXFileReference; explicitFileType = wrapper.application; includeInIndex = 0; path = RxCode.app; sourceTree = BUILT_PRODUCTS_DIR; }; /* End PBXFileReference section */ @@ -79,6 +81,7 @@ 4381E755142272EB2DAA9C96 /* AppStateProjectSwitchTests.swift */, DFA0CCC02FB4CC01005991E1 /* PlanDecisionTests.swift */, DFA0CCC12FB4CC01005991E1 /* PlanCardViewTests.swift */, + DFA0CCD52FB4CC02005991E1 /* HistoryListArchiveFilterTests.swift */, ); path = RxCodeTests; sourceTree = ""; @@ -237,6 +240,7 @@ 33993F0F87CF4DB09F2813A8 /* AppStateProjectSwitchTests.swift in Sources */, DFA0CCD12FB4CC01005991E1 /* PlanDecisionTests.swift in Sources */, DFA0CCD22FB4CC01005991E1 /* PlanCardViewTests.swift in Sources */, + DFA0CCE12FB4CC02005991E1 /* HistoryListArchiveFilterTests.swift in Sources */, ); runOnlyForDeploymentPostprocessing = 0; }; diff --git a/RxCode/App/AppState.swift b/RxCode/App/AppState.swift index a6f30343..03e2d121 100644 --- a/RxCode/App/AppState.swift +++ b/RxCode/App/AppState.swift @@ -894,6 +894,7 @@ final class AppState { let mcp: MCPService let threadStore: ThreadStore let runService = RunService() + let ideMCPServer = IDEMCPServer() // MARK: - Run Profiles @@ -960,9 +961,29 @@ final class AppState { self.mcp = MCPService(claudeService: claude) self.threadStore = ThreadStore.make() - // Bridge ACP `session/request_permission` into the existing PermissionServer. + // Bridge ACP `session/request_permission` and Codex in-band permission + // requests into the existing PermissionServer. let permission = self.permission - Task { await acp.setPermissionServer(permission) } + let codex = self.codex + let ideMCPServer = self.ideMCPServer + Task { + await acp.setPermissionServer(permission) + await codex.setPermissionServer(permission) + await ideMCPServer.setHandler(self) + } + } + + // MARK: - Agent Backends + + /// Looks up the `AgentBackend` for the given provider. Used by + /// `processStream`/`cancel`/`finalize` to dispatch via the unified + /// protocol instead of switching on the enum directly. + func backend(for provider: AgentProvider) -> any AgentBackend { + switch provider { + case .claudeCode: return claude + case .codex: return codex + case .acp: return acp + } } // MARK: - ACP Actions @@ -2201,45 +2222,49 @@ final class AppState { var sessionKey = internalSessionKey - let stream: AsyncStream + // Resolve per-backend send-request fields (MCP injection, ACP client + // spec, model split) before dispatching through the unified protocol. + var mcpClaudeConfigPath: String? = nil + var mcpCodexOverrides: [String] = [] + var acpMCPServers: [JSONValue] = [] + var acpSpec: ACPClientSpec? = nil + var resolvedModel: String? = model + var resolvedSendMode: PermissionMode = permissionMode + var earlyStream: AsyncStream? = nil + switch agentProvider { case .claudeCode: - let mcpConfigPath = await mcp.writeClaudeConfig(projectPath: cwd) - stream = await claude.send( - streamId: streamId, - prompt: prompt, - cwd: cwd, - sessionId: cliSessionId, - model: model, - effort: effort, - hookSettingsPath: hookSettingsPath, - mcpConfigPath: mcpConfigPath, - permissionMode: permissionMode - ) + mcpClaudeConfigPath = await mcp.writeClaudeConfig(projectPath: cwd) case .codex: - let mcpConfigOverrides = await mcp.codexConfigOverrides(projectPath: cwd) - stream = await codex.send( - streamId: streamId, - prompt: prompt, - cwd: cwd, - threadId: cliSessionId, - model: model, - permissionMode: registerMode, - planMode: permissionMode == .plan, - mcpConfigOverrides: mcpConfigOverrides, - permissionServer: permission - ) + mcpCodexOverrides = await mcp.codexConfigOverrides(projectPath: cwd) + resolvedSendMode = registerMode case .acp: + // Allocate a per-session IDE-MCP port so the ACP agent can call + // polyfill / introspection tools. The agent's MCP child is a + // perl one-liner that bridges its stdio to our TCP listener; + // the listener stays bound to this session for its lifetime. + let idePort = await ideMCPServer.allocate( + sessionKey: sessionKey, + capabilities: AgentProvider.acp.staticCapabilities + ) + let bridge = idePort.map { IDEMCPServer.bridgeCommand(forPort: $0) } + acpMCPServers = await mcp.acpMCPServers( + projectPath: cwd, + bridgeCommand: bridge + ) // `model` may be a composite `::` key (from the picker) // or a bare model id (from a per-session override). let split = acpSelectionParts(for: model) let resolvedClientId = split?.clientId ?? sessionStates[sessionKey]?.acpClientId ?? selectedACPClientId - let resolvedModel = split?.model ?? model - guard let spec = acpClients.first(where: { $0.id == resolvedClientId && $0.enabled }) else { + resolvedModel = split?.model ?? model + resolvedSendMode = registerMode + if let spec = acpClients.first(where: { $0.id == resolvedClientId && $0.enabled }) { + acpSpec = spec + } else { logger.error("[ACP] no enabled client for id=\(resolvedClientId, privacy: .public)") - let placeholder = AsyncStream { c in + earlyStream = AsyncStream { c in c.yield(.user(UserMessage( toolUseId: nil, content: "No ACP client configured. Add one in Settings → ACP Clients.", @@ -2252,19 +2277,30 @@ final class AppState { ))) c.finish() } - stream = placeholder - break } - stream = await acp.send( + } + + let stream: AsyncStream + if let earlyStream { + stream = earlyStream + } else { + let request = BackendSendRequest( streamId: streamId, prompt: prompt, cwd: cwd, sessionId: cliSessionId, model: resolvedModel, - spec: spec, - permissionMode: registerMode, + effort: effort, + permissionMode: resolvedSendMode, + planMode: permissionMode == .plan, + hookSettingsPath: hookSettingsPath, + mcpClaudeConfigPath: mcpClaudeConfigPath, + mcpCodexOverrides: mcpCodexOverrides, + acpMCPServers: acpMCPServers, + acpSpec: acpSpec, clientSessionKey: sessionKey ) + stream = await backend(for: agentProvider).send(request) } startFlushTimer(for: sessionKey) @@ -2483,18 +2519,70 @@ final class AppState { logger.info("[Stream:UI] Codex usage applied messageId=\(assistantMessage.id ?? "", privacy: .public) output=\(liveOutput) total=\(total)") } } - // Extract text only when no text_delta has been received in the current turn. - // Normally content_block_delta(text_delta) is the primary path, so this branch rarely executes. + // ACP-style providers deliver fully-formed tool_use blocks inside .assistant + // events (no content_block_start raw stream). Commit any buffered text first + // so tool bubbles appear after — and not in the middle of — the prior text. + let hasToolUse = assistantMessage.content.contains { + if case .toolUse = $0 { return true } + return false + } + if hasToolUse { + flushPendingUpdates(for: sessionKey) + } + updateState(sessionKey) { state in - guard state.textDeltaBuffer.isEmpty else { return } - let afterLastUser = (state.messages.lastIndex(where: { $0.role == .user }).map { $0 + 1 }) ?? 0 - let hasStreamedText = state.messages.suffix(from: afterLastUser).contains { - $0.role == .assistant && $0.blocks.contains(where: \.isText) - } - guard !hasStreamedText else { return } + // Text fallback: only buffer text when no text_delta has been received in + // this turn. Normally content_block_delta(text_delta) is the primary path. + let canBufferText: Bool = { + guard state.textDeltaBuffer.isEmpty else { return false } + let afterLastUser = (state.messages.lastIndex(where: { $0.role == .user }).map { $0 + 1 }) ?? 0 + return !state.messages.suffix(from: afterLastUser).contains { + $0.role == .assistant && $0.blocks.contains(where: \.isText) + } + }() + for block in assistantMessage.content { - if case .text(let text) = block, !text.isEmpty { - state.textDeltaBuffer += text + switch block { + case .text(let text): + if canBufferText, !text.isEmpty { + state.textDeltaBuffer += text + } + case .toolUse(let id, let name, let input): + state.isThinking = false + // Merge updates by id: ACP agents may re-emit the same toolUse + // with additional input (e.g. diff content arriving via a + // follow-up tool_call_update). Patch the existing block in + // place so the live edit info reaches `flushPendingUpdates` + // when the result lands. + if let existingMsgIdx = state.messages.indices.reversed().first(where: { + state.messages[$0].toolCallIndex(id: id) != nil + }), + let existingBlockIdx = state.messages[existingMsgIdx].toolCallIndex(id: id) { + var merged = state.messages[existingMsgIdx].blocks[existingBlockIdx].toolCall?.input ?? [:] + for (key, value) in input { merged[key] = value } + state.messages[existingMsgIdx].blocks[existingBlockIdx].toolCall?.input = merged + } else { + if state.needsNewMessage { + if let idx = state.messages.indices.reversed().first(where: { + state.messages[$0].role == .assistant && state.messages[$0].isStreaming + }) { + state.messages[idx].isStreaming = false + state.messages[idx].finalizeToolCalls() + Self.stripNoOpText(at: idx, in: &state.messages) + } + state.messages.append(ChatMessage(role: .assistant, isStreaming: true)) + state.needsNewMessage = false + } else if state.messages.last?.role != .assistant + || !(state.messages.last?.isStreaming ?? false) { + state.messages.append(ChatMessage(role: .assistant, isStreaming: true)) + } + if let lastIndex = state.messages.indices.last, + state.messages[lastIndex].role == .assistant { + state.messages[lastIndex].appendToolCall(ToolCall(id: id, name: name, input: input)) + } + } + case .thinking: + state.isThinking = true } } } @@ -2684,15 +2772,18 @@ final class AppState { } private func finalizeAgentStream(agentProvider: AgentProvider, streamId: UUID) async { - switch agentProvider { - case .claudeCode: + // Claude needs an explicit stdin close before finalize so the CLI + // sees EOF; other backends manage stdin internally. + if agentProvider == .claudeCode { await claude.closeStdin(streamId: streamId) - await claude.finalize(streamId: streamId) - case .codex: - await codex.finalize(streamId: streamId) - case .acp: - await acp.finalize(streamId: streamId) } + await backend(for: agentProvider).finalize(streamId: streamId) + } + + /// Release the per-session IDE-MCP listener allocated for ACP turns. + /// Safe to call for non-ACP providers (no-op). + private func releaseIDESession(sessionKey: String) async { + await ideMCPServer.release(sessionKey: sessionKey) } private func consumeAgentStderr(agentProvider: AgentProvider, streamId: UUID) async -> String? { @@ -2956,14 +3047,7 @@ final class AppState { if let streamToCancel { let provider = sessionStates[key]?.agentProvider ?? effectiveModelSelection(in: window).provider - switch provider { - case .claudeCode: - await claude.cancel(streamId: streamToCancel) - case .codex: - await codex.cancel(streamId: streamToCancel) - case .acp: - await acp.cancel(streamId: streamToCancel) - } + await backend(for: provider).cancel(streamId: streamToCancel) } flushPendingUpdates(for: key) diff --git a/RxCode/Services/ACPService.swift b/RxCode/Services/ACPService.swift index 8b712a05..ad6a9224 100644 --- a/RxCode/Services/ACPService.swift +++ b/RxCode/Services/ACPService.swift @@ -7,43 +7,60 @@ import os // Speaks the Agent Client Protocol (https://agentclientprotocol.com) over // newline-delimited JSON-RPC on a child process's stdio. // -// One ACPService instance hosts many concurrent streams. Each stream owns a -// dedicated agent subprocess for its turn — the simplest mapping that mirrors -// `ClaudeService`'s one-process-per-prompt model. A future optimization could -// pool processes per (cwd, clientSpec) since ACP supports `session/load`. +// One ACPService instance hosts many concurrent sessions. Each session pools +// a single agent subprocess across turns of the same RxCode thread so the +// agent retains conversation memory — ACP itself doesn't pass history in +// `session/prompt`, so a fresh process per turn would treat every prompt as +// the start of a new conversation. The pool is keyed by a canonical session +// key (initially the AppState `clientSessionKey`, re-keyed to the agent's +// `session/new` sessionId once known) so subsequent turns whose key has been +// renamed in AppState still find the same process. actor ACPService { private let logger = Logger(subsystem: "com.claudework", category: "ACPService") - private struct StreamEntry { + /// Per-pool entry. Persistent fields outlive a single turn; per-turn + /// fields are reset before each `session/prompt`. + private struct SessionEntry { + // Persistent (process-level) let process: Process let stdin: FileHandle + let spec: ACPClientSpec + let cwd: String + var canonicalKey: String + var agentSessionId: String? + var modelConfigId: String? var nextId: Int = 1 var pending: [Int: CheckedContinuation] = [:] - var agentSessionId: String? - var continuation: AsyncStream.Continuation? - var spec: ACPClientSpec - var cwd: String - var clientSessionKey: String var stderr: String = "" - /// `tool_call.toolCallId` → synthesized RxCode ToolCall id (same string). + var stdoutReaderTask: Task? + + // Per-turn (reset before `session/prompt`) + var currentStreamId: UUID? + var continuation: AsyncStream.Continuation? var liveToolCalls: Set = [] - /// Plan card uses a stable synthetic id per stream so successive `plan` updates replace the same card. - let planSyntheticId: String = "acp-plan-\(UUID().uuidString)" + var planSyntheticId: String = "acp-plan-\(UUID().uuidString)" var planEmitted: Bool = false - /// `SessionConfigId` of the model selector advertised by the agent's - /// `session/new` response, if any. Cached so the turn can issue - /// `session/set_config_option` to switch models. - var modelConfigId: String? - /// Toggled on just before `session/prompt` is sent. Agents like - /// OpenCode replay the historical conversation as `session/update` - /// notifications during `session/load`; we discard those so they don't - /// get rendered as live messages. + /// Toggled on just before `session/prompt` is sent. Some agents replay + /// the historical conversation as `session/update` notifications during + /// `session/load`; we discard those so they don't get rendered as live + /// messages. var acceptingUpdates: Bool = false + /// Probe entries get torn down in `probeModels` instead of pooled + /// across turns. + let isEphemeral: Bool } - private var streams: [UUID: StreamEntry] = [:] + private var sessions: [String: SessionEntry] = [:] + /// Maps an externally-issued streamId to its canonical pool key. + private var streamToKey: [UUID: String] = [:] + /// Maps a previously-seen pool key (e.g. AppState's "pending-…" key) to + /// the canonical key (the agent's `session/new` sessionId). Used to + /// resolve subsequent turns whose `clientSessionKey` was renamed by + /// AppState's session-id reconciliation. + private var aliasToCanonical: [String: String] = [:] + /// Reference to the permission server for bridging `session/request_permission`. private weak var permissionServer: PermissionServer? @@ -68,9 +85,10 @@ actor ACPService { model: String?, spec: ACPClientSpec, permissionMode: PermissionMode, - clientSessionKey: String + clientSessionKey: String, + mcpServers: [JSONValue] = [] ) -> AsyncStream { - logger.info("[ACP] send streamId=\(streamId.uuidString, privacy: .public) client=\(spec.displayName, privacy: .public) launch=\(spec.launch.displayKind, privacy: .public) model=\(model ?? "", privacy: .public) sessionId=\(sessionId ?? "", privacy: .public) mode=\(String(describing: permissionMode), privacy: .public) cwd=\(cwd, privacy: .public) promptLen=\(prompt.count)") + logger.info("[ACP] send streamId=\(streamId.uuidString, privacy: .public) client=\(spec.displayName, privacy: .public) launch=\(spec.launch.displayKind, privacy: .public) model=\(model ?? "", privacy: .public) sessionId=\(sessionId ?? "", privacy: .public) mode=\(String(describing: permissionMode), privacy: .public) cwd=\(cwd, privacy: .public) clientKey=\(clientSessionKey, privacy: .public) mcpServers=\(mcpServers.count) promptLen=\(prompt.count)") return AsyncStream { continuation in let task = Task { await self.runTurn( @@ -82,6 +100,7 @@ actor ACPService { spec: spec, permissionMode: permissionMode, clientSessionKey: clientSessionKey, + mcpServers: mcpServers, continuation: continuation ) } @@ -92,17 +111,30 @@ actor ACPService { } } + /// Releases per-stream bookkeeping for a turn that ran to completion. + /// The pooled agent process stays alive so the next turn for the same + /// session inherits its conversation memory. func finalize(streamId: UUID) { - guard var entry = streams.removeValue(forKey: streamId) else { return } - logger.info("[ACP] finalize streamId=\(streamId.uuidString, privacy: .public) pid=\(entry.process.processIdentifier) pending=\(entry.pending.count) running=\(entry.process.isRunning)") - try? entry.stdin.close() - if entry.process.isRunning { - entry.process.terminate() + guard let key = streamToKey.removeValue(forKey: streamId) else { return } + guard var entry = sessions[key] else { return } + if entry.currentStreamId == streamId { + entry.currentStreamId = nil + entry.continuation = nil + entry.acceptingUpdates = false + entry.liveToolCalls.removeAll() + entry.planEmitted = false + // pending should be empty after a clean turn; if any are left, + // resume them with streamClosed so the turn surfaces the error. + for (_, cont) in entry.pending { + cont.resume(throwing: ACPError.streamClosed) + } + entry.pending.removeAll() } - for (_, cont) in entry.pending { - cont.resume(throwing: ACPError.streamClosed) + sessions[key] = entry + if entry.isEphemeral { + killSession(key: key) } - entry.pending.removeAll() + logger.info("[ACP] finalize streamId=\(streamId.uuidString, privacy: .public) key=\(key, privacy: .public) keepingProcess=\(!entry.isEphemeral) running=\(entry.process.isRunning)") } func handleStreamTermination(streamId: UUID) { @@ -110,32 +142,75 @@ actor ACPService { finalize(streamId: streamId) } + /// Cancels the in-flight turn for `streamId` via `session/cancel` but + /// keeps the agent process alive so the user can immediately send a new + /// prompt without losing conversation state. func cancel(streamId: UUID) { - // Attempt graceful cancel via session/cancel notification, then kill the process. - if let entry = streams[streamId], let sid = entry.agentSessionId { + guard let key = streamToKey[streamId], let entry = sessions[key] else { + return + } + if let agentSid = entry.agentSessionId { let frame: [String: Any] = [ "jsonrpc": "2.0", "method": "session/cancel", - "params": ["sessionId": sid] + "params": ["sessionId": agentSid] ] if let data = try? JSONSerialization.data(withJSONObject: frame), let line = String(data: data, encoding: .utf8) { _ = try? entry.stdin.write(contentsOf: Data((line + "\n").utf8)) } } + // Resume any pending requests with .streamClosed so the runTurn await + // unblocks and we hit the per-turn cleanup path in finalize. + mutateSession(key) { e in + for (_, cont) in e.pending { + cont.resume(throwing: ACPError.streamClosed) + } + e.pending.removeAll() + e.continuation?.finish() + } finalize(streamId: streamId) } func consumeStderr(for streamId: UUID) -> String? { - guard let entry = streams[streamId] else { return nil } + guard let key = streamToKey[streamId], let entry = sessions[key] else { + return nil + } let trimmed = entry.stderr.trimmingCharacters(in: .whitespacesAndNewlines) return trimmed.isEmpty ? nil : trimmed } func cleanup() { - for (id, _) in streams { - finalize(streamId: id) + for (key, _) in sessions { + killSession(key: key) } + sessions.removeAll() + streamToKey.removeAll() + aliasToCanonical.removeAll() + } + + /// Tear down a pooled session — terminate the process and forget all + /// references. Used by `cleanup()`, ephemeral probe sessions, and as a + /// recovery path when a pooled process dies between turns. + private func killSession(key: String) { + guard var entry = sessions.removeValue(forKey: key) else { return } + try? entry.stdin.close() + if entry.process.isRunning { + entry.process.terminate() + } + entry.stdoutReaderTask?.cancel() + for (_, cont) in entry.pending { + cont.resume(throwing: ACPError.streamClosed) + } + entry.pending.removeAll() + entry.continuation?.finish() + if let sid = entry.currentStreamId { + streamToKey.removeValue(forKey: sid) + } + // Drop any aliases that pointed at this canonical key so future + // lookups don't follow a dangling entry. + let aliases = aliasToCanonical.filter { $0.value == key }.map(\.key) + for alias in aliases { aliasToCanonical.removeValue(forKey: alias) } } /// One-shot probe that spawns the agent, runs `initialize` + `session/new`, @@ -151,30 +226,35 @@ actor ACPService { timeout: Duration = .seconds(20) ) async throws -> ACPModelConfig? { let streamId = UUID() + let key = "probe-\(streamId.uuidString)" logger.info("[ACP] probe start: \(spec.displayName, privacy: .public) (launch=\(spec.launch.displayKind, privacy: .public)) cwd=\(cwd, privacy: .public)") let (process, stdin, stdout, stderr) = try await spawn(spec: spec, model: nil, cwd: cwd) logger.info("[ACP] probe spawned pid=\(process.processIdentifier) for \(spec.displayName, privacy: .public)") - let entry = StreamEntry( + var entry = SessionEntry( process: process, stdin: stdin, spec: spec, cwd: cwd, - clientSessionKey: "" + canonicalKey: key, + isEphemeral: true ) - streams[streamId] = entry + entry.currentStreamId = streamId + sessions[key] = entry + streamToKey[streamId] = key - startStderrReader(streamId: streamId, stderr: stderr) - let readerTask = Self.spawnStdoutReader(streamId: streamId, stdout: stdout, service: self) + startStderrReader(key: key, stderr: stderr) + let readerTask = Self.spawnStdoutReader(key: key, stdout: stdout, service: self) + mutateSession(key) { $0.stdoutReaderTask = readerTask } process.terminationHandler = { [weak self] _ in - Task.detached { await self?.handleProcessExit(streamId: streamId) } + Task.detached { await self?.handleProcessExit(key: key) } } do { let result = try await withThrowingTaskGroup(of: ACPModelConfig?.self) { group in group.addTask { - try await self.runProbeSequence(streamId: streamId, spec: spec, cwd: cwd) + try await self.runProbeSequence(key: key, spec: spec, cwd: cwd) } group.addTask { try await Task.sleep(for: timeout) @@ -185,27 +265,29 @@ actor ACPService { return value } readerTask.cancel() - finalize(streamId: streamId) + killSession(key: key) + streamToKey.removeValue(forKey: streamId) return result } catch { - let stderrSnapshot = streams[streamId]?.stderr.trimmingCharacters(in: .whitespacesAndNewlines) ?? "" + let stderrSnapshot = sessions[key]?.stderr.trimmingCharacters(in: .whitespacesAndNewlines) ?? "" if !stderrSnapshot.isEmpty { logger.error("[ACP] probe stderr for \(spec.displayName, privacy: .public): \(stderrSnapshot, privacy: .public)") } readerTask.cancel() - finalize(streamId: streamId) + killSession(key: key) + streamToKey.removeValue(forKey: streamId) throw error } } private func runProbeSequence( - streamId: UUID, + key: String, spec: ACPClientSpec, cwd: String ) async throws -> ACPModelConfig? { logger.info("[ACP] probe → initialize \(spec.displayName, privacy: .public)") _ = try await sendRequest( - streamId: streamId, + key: key, method: "initialize", params: [ "protocolVersion": .number(1), @@ -221,7 +303,7 @@ actor ACPService { logger.info("[ACP] probe → session/new \(spec.displayName, privacy: .public)") let newResult = try await sendRequest( - streamId: streamId, + key: key, method: "session/new", params: [ "cwd": .string(cwd), @@ -251,176 +333,333 @@ actor ACPService { spec: ACPClientSpec, permissionMode: PermissionMode, clientSessionKey: String, + mcpServers: [JSONValue] = [], continuation: AsyncStream.Continuation ) async { do { - let (process, stdin, stdout, stderr) = try await spawn(spec: spec, model: model, cwd: cwd) - var entry = StreamEntry( - process: process, - stdin: stdin, - spec: spec, - cwd: cwd, - clientSessionKey: clientSessionKey - ) - entry.continuation = continuation - streams[streamId] = entry - - // Start background readers detached so they don't fight the writer - // for ACPService actor reentrance — they only hop onto the actor - // when delivering a line. - startStderrReader(streamId: streamId, stderr: stderr) - let readerTask = Self.spawnStdoutReader(streamId: streamId, stdout: stdout, service: self) - process.terminationHandler = { [weak self] _ in - Task.detached { await self?.handleProcessExit(streamId: streamId) } + // Resolve the pool key — AppState may pass the original + // `pending-…` key OR the agent's later sessionId. Either should + // map to the same pooled entry. + let resolvedKey = aliasToCanonical[clientSessionKey] ?? clientSessionKey + let canReuse: Bool = { + guard let existing = sessions[resolvedKey] else { return false } + guard !existing.isEphemeral else { return false } + guard existing.spec.id == spec.id else { return false } + guard existing.cwd == cwd else { return false } + guard existing.process.isRunning else { return false } + guard existing.agentSessionId != nil else { return false } + return true + }() + + if canReuse { + try await runReusedTurn( + streamId: streamId, + poolKey: resolvedKey, + prompt: prompt, + model: model, + continuation: continuation + ) + return } - let heartbeat = Task.detached { [weak self, displayName = spec.displayName, launchKind = spec.launch.displayKind] in - let started = Date() - while !Task.isCancelled { - try? await Task.sleep(for: .seconds(5)) - if Task.isCancelled { break } - let elapsed = Int(Date().timeIntervalSince(started)) - await self?.heartbeatTick(displayName: displayName, elapsed: elapsed, launchKind: launchKind) - } + + // Pool has a stale entry (different spec/cwd or process dead). + if sessions[resolvedKey] != nil { + logger.info("[ACP] dropping stale pooled session for key=\(resolvedKey, privacy: .public)") + killSession(key: resolvedKey) } - // 1. initialize - let initResult = try await sendRequest( + try await runFreshTurn( streamId: streamId, - method: "initialize", - params: [ - "protocolVersion": .number(1), - "clientCapabilities": .object([ - "fs": .object([ - "readTextFile": .bool(true), - "writeTextFile": .bool(true) - ]) - ]) - ] - ) - heartbeat.cancel() - logger.info("[ACP] initialize ok for \(spec.displayName, privacy: .public): \(initResult.shortDescription, privacy: .public)") - - // Surface the agent identity as a system event so the chat header - // matches the Claude path. - continuation.yield(.system(SystemEvent( - subtype: "init", - sessionId: incomingSessionId, - tools: nil, + bootstrapKey: clientSessionKey, + prompt: prompt, + cwd: cwd, + incomingSessionId: incomingSessionId, model: model, - claudeCodeVersion: nil + spec: spec, + permissionMode: permissionMode, + mcpServers: mcpServers, + continuation: continuation + ) + } catch { + logger.error("[ACP] turn failed: \(error.localizedDescription, privacy: .public)") + continuation.yield(.user(UserMessage( + toolUseId: nil, + content: "ACP error: \(error.localizedDescription)", + isError: true ))) + continuation.finish() + finalize(streamId: streamId) + } + } - // 2. Always `session/new`. We could `session/load` here if the agent - // advertises `loadSession`, but ACP's load-replay is only useful - // when the same agent *process* is being reattached. Our current - // architecture spawns a fresh subprocess per turn, so `session/load` - // on agents that persist sessions to disk (Gemini CLI) makes them - // re-emit the persisted conversation during the next `session/prompt`, - // which surfaces in the UI as prior answers being prepended to new - // ones. Each turn therefore starts a fresh ACP session; conversation - // continuity is preserved by RxCode's chat history, not by the agent. - let agentSessionId: String - var modelConfig: ACPModelConfig? - let sessionMethod = "session/new" - let newParams: [String: JSONValue] = [ - "cwd": .string(cwd), - "mcpServers": .array([]) + private func runFreshTurn( + streamId: UUID, + bootstrapKey: String, + prompt: String, + cwd: String, + incomingSessionId: String?, + model: String?, + spec: ACPClientSpec, + permissionMode: PermissionMode, + mcpServers: [JSONValue] = [], + continuation: AsyncStream.Continuation + ) async throws { + let (process, stdin, stdout, stderr) = try await spawn(spec: spec, model: model, cwd: cwd) + var entry = SessionEntry( + process: process, + stdin: stdin, + spec: spec, + cwd: cwd, + canonicalKey: bootstrapKey, + isEphemeral: false + ) + entry.continuation = continuation + entry.currentStreamId = streamId + sessions[bootstrapKey] = entry + streamToKey[streamId] = bootstrapKey + + // Reader closures capture the canonical key, not the streamId, so + // the readers keep delivering correctly across turns when the pool + // entry is later re-keyed to the agent's sessionId. + startStderrReader(key: bootstrapKey, stderr: stderr) + let readerTask = Self.spawnStdoutReader(key: bootstrapKey, stdout: stdout, service: self) + mutateSession(bootstrapKey) { $0.stdoutReaderTask = readerTask } + process.terminationHandler = { [weak self] _ in + // Use the canonical key at exit time — `handleProcessExit` looks + // up via the alias map, so this works even if the pool entry has + // since been re-keyed to the agent's sessionId. + Task.detached { await self?.handleProcessExit(key: bootstrapKey) } + } + + let heartbeat = Task.detached { [weak self, displayName = spec.displayName, launchKind = spec.launch.displayKind] in + let started = Date() + while !Task.isCancelled { + try? await Task.sleep(for: .seconds(5)) + if Task.isCancelled { break } + let elapsed = Int(Date().timeIntervalSince(started)) + await self?.heartbeatTick(displayName: displayName, elapsed: elapsed, launchKind: launchKind) + } + } + + // 1. initialize + let initResult = try await sendRequest( + key: bootstrapKey, + method: "initialize", + params: [ + "protocolVersion": .number(1), + "clientCapabilities": .object([ + "fs": .object([ + "readTextFile": .bool(true), + "writeTextFile": .bool(true) + ]) + ]) ] - let newResult = try await sendRequest(streamId: streamId, method: "session/new", params: newParams) - guard let sid = newResult.objectValue?["sessionId"]?.stringValue else { - throw ACPError.protocolMismatch("session/new returned no sessionId") + ) + heartbeat.cancel() + logger.info("[ACP] initialize ok for \(spec.displayName, privacy: .public): \(initResult.shortDescription, privacy: .public)") + + continuation.yield(.system(SystemEvent( + subtype: "init", + sessionId: incomingSessionId, + tools: nil, + model: model, + claudeCodeVersion: nil + ))) + + // 2. session/new — fresh ACP session backed by the new process. The + // agent's history will accumulate within this process across pooled + // turns; we don't use `session/load` because that re-emits the + // persisted conversation as updates and would duplicate messages. + let newParams: [String: JSONValue] = [ + "cwd": .string(cwd), + "mcpServers": .array(mcpServers) + ] + let newResult = try await sendRequest(key: bootstrapKey, method: "session/new", params: newParams) + guard let agentSessionId = newResult.objectValue?["sessionId"]?.stringValue else { + throw ACPError.protocolMismatch("session/new returned no sessionId") + } + let modelConfig = Self.parseModelConfig(from: newResult) + _ = incomingSessionId + + // Re-key the pool entry to the agent's sessionId so subsequent turns + // (where AppState's sessionKey is the agent sid) hit the cache. + let canonicalKey = agentSessionId + if canonicalKey != bootstrapKey { + if var moved = sessions.removeValue(forKey: bootstrapKey) { + moved.canonicalKey = canonicalKey + moved.agentSessionId = agentSessionId + moved.modelConfigId = modelConfig?.configId + sessions[canonicalKey] = moved } - modelConfig = Self.parseModelConfig(from: newResult) - agentSessionId = sid - _ = incomingSessionId // intentionally unused — see comment above - mutateStream(streamId) { + streamToKey[streamId] = canonicalKey + aliasToCanonical[bootstrapKey] = canonicalKey + } else { + mutateSession(canonicalKey) { $0.agentSessionId = agentSessionId $0.modelConfigId = modelConfig?.configId } + } - // Tell AppState about the discovered model list so it can refresh - // the picker and persist it to disk. - if let modelConfig { - logger.info("[ACP] discovered model selector for \(spec.displayName, privacy: .public) via \(sessionMethod, privacy: .public) configId=\(modelConfig.configId, privacy: .public) current=\(modelConfig.currentValue ?? "nil", privacy: .public) models=\(modelConfig.options.count) [\(Self.modelListDescription(modelConfig.options), privacy: .public)]") - continuation.yield(.acpModelsDiscovered(ACPModelsDiscoveredEvent( - clientId: spec.id, - config: modelConfig - ))) - } else { - logger.info("[ACP] no model selector discovered for \(spec.displayName, privacy: .public) via \(sessionMethod, privacy: .public)") - } + if let modelConfig { + logger.info("[ACP] discovered model selector for \(spec.displayName, privacy: .public) configId=\(modelConfig.configId, privacy: .public) current=\(modelConfig.currentValue ?? "nil", privacy: .public) models=\(modelConfig.options.count) [\(Self.modelListDescription(modelConfig.options), privacy: .public)]") + continuation.yield(.acpModelsDiscovered(ACPModelsDiscoveredEvent( + clientId: spec.id, + config: modelConfig + ))) + } else { + logger.info("[ACP] no model selector discovered for \(spec.displayName, privacy: .public)") + } - // Apply the user's model choice via the spec-compliant route when - // the agent advertised a selector. Env-var-at-spawn covers the - // fallback case for older agents. - if let modelConfig, - let model, - !model.isEmpty, - modelConfig.options.contains(where: { $0.value == model }), - modelConfig.currentValue != model { - do { - _ = try await sendRequest( - streamId: streamId, - method: "session/set_config_option", - params: [ - "sessionId": .string(agentSessionId), - "configId": .string(modelConfig.configId), - "value": .string(model) - ] - ) - } catch { - logger.warning("[ACP] session/set_config_option failed for model \(model, privacy: .public): \(error.localizedDescription, privacy: .public)") - } - } + try await applyModelSelection( + key: canonicalKey, + agentSessionId: agentSessionId, + modelConfig: modelConfig, + model: model + ) - // Emit the agent's session id so the chat UI replaces its pending placeholder. - continuation.yield(.system(SystemEvent( - subtype: "session_started", - sessionId: agentSessionId, - tools: nil, - model: model, - claudeCodeVersion: nil - ))) + continuation.yield(.system(SystemEvent( + subtype: "session_started", + sessionId: agentSessionId, + tools: nil, + model: model, + claudeCodeVersion: nil + ))) - // 3. session/prompt — blocks until the turn completes. - // Flip the gate AFTER the optional `session/set_config_option` - // exchange so we don't accept stray updates before the prompt is - // actually in flight. - mutateStream(streamId) { $0.acceptingUpdates = true } - let promptResult = try await sendRequest( - streamId: streamId, - method: "session/prompt", + try await runPrompt( + key: canonicalKey, + agentSessionId: agentSessionId, + prompt: prompt, + continuation: continuation + ) + } + + private func runReusedTurn( + streamId: UUID, + poolKey: String, + prompt: String, + model: String?, + continuation: AsyncStream.Continuation + ) async throws { + guard let agentSessionId = sessions[poolKey]?.agentSessionId else { + throw ACPError.protocolMismatch("pooled session missing agentSessionId") + } + let modelConfigId = sessions[poolKey]?.modelConfigId + let spec = sessions[poolKey]!.spec + logger.info("[ACP] reusing pooled session key=\(poolKey, privacy: .public) agentSid=\(agentSessionId, privacy: .public) client=\(spec.displayName, privacy: .public)") + + // Reset per-turn state and adopt the new streamId/continuation. + mutateSession(poolKey) { e in + e.currentStreamId = streamId + e.continuation = continuation + e.liveToolCalls.removeAll() + e.planEmitted = false + e.planSyntheticId = "acp-plan-\(UUID().uuidString)" + e.acceptingUpdates = false + } + streamToKey[streamId] = poolKey + + continuation.yield(.system(SystemEvent( + subtype: "init", + sessionId: agentSessionId, + tools: nil, + model: model, + claudeCodeVersion: nil + ))) + + // Apply model change if the user picked a different one for this turn. + if let modelConfigId, let model, !model.isEmpty { + let cfg = ACPModelConfig(configId: modelConfigId, currentValue: nil, options: [ + ACPModelOption(value: model, name: model, description: nil) + ]) + try await applyModelSelection( + key: poolKey, + agentSessionId: agentSessionId, + modelConfig: cfg, + model: model + ) + } + + continuation.yield(.system(SystemEvent( + subtype: "session_started", + sessionId: agentSessionId, + tools: nil, + model: model, + claudeCodeVersion: nil + ))) + + try await runPrompt( + key: poolKey, + agentSessionId: agentSessionId, + prompt: prompt, + continuation: continuation + ) + } + + /// Issues `session/set_config_option` to switch the agent's active model + /// when the user picked a different value than the one currently selected. + /// Best-effort: failures are logged but don't abort the turn. + private func applyModelSelection( + key: String, + agentSessionId: String, + modelConfig: ACPModelConfig?, + model: String? + ) async throws { + guard let modelConfig, + let model, + !model.isEmpty, + modelConfig.options.contains(where: { $0.value == model }), + modelConfig.currentValue != model + else { return } + do { + _ = try await sendRequest( + key: key, + method: "session/set_config_option", params: [ "sessionId": .string(agentSessionId), - "prompt": .array([.object([ - "type": .string("text"), - "text": .string(prompt) - ])]) + "configId": .string(modelConfig.configId), + "value": .string(model) ] ) + } catch { + logger.warning("[ACP] session/set_config_option failed for model \(model, privacy: .public): \(error.localizedDescription, privacy: .public)") + } + } - let stopReason = promptResult.objectValue?["stopReason"]?.stringValue ?? "end_turn" - continuation.yield(.result(ResultEvent( - durationMs: nil, - totalCostUsd: nil, - sessionId: agentSessionId, - isError: stopReason == "refusal", - totalTurns: nil, - usage: nil, - contextWindow: nil - ))) + private func runPrompt( + key: String, + agentSessionId: String, + prompt: String, + continuation: AsyncStream.Continuation + ) async throws { + // Flip the gate AFTER any optional `session/set_config_option` + // exchange so we don't accept stray updates before the prompt is + // actually in flight. + mutateSession(key) { $0.acceptingUpdates = true } + let promptResult = try await sendRequest( + key: key, + method: "session/prompt", + params: [ + "sessionId": .string(agentSessionId), + "prompt": .array([.object([ + "type": .string("text"), + "text": .string(prompt) + ])]) + ] + ) - readerTask.cancel() - continuation.finish() - finalize(streamId: streamId) - } catch { - logger.error("[ACP] turn failed: \(error.localizedDescription, privacy: .public)") - continuation.yield(.user(UserMessage( - toolUseId: nil, - content: "ACP error: \(error.localizedDescription)", - isError: true - ))) - continuation.finish() + let stopReason = promptResult.objectValue?["stopReason"]?.stringValue ?? "end_turn" + continuation.yield(.result(ResultEvent( + durationMs: nil, + totalCostUsd: nil, + sessionId: agentSessionId, + isError: stopReason == "refusal", + totalTurns: nil, + usage: nil, + contextWindow: nil + ))) + + continuation.finish() + if let streamId = sessions[key]?.currentStreamId { finalize(streamId: streamId) } } @@ -517,7 +756,6 @@ actor ACPService { { switch launch { case .npx(let package, let args, let env): - // `npx -y ` so it runs without an install prompt. return ("/usr/bin/env", ["npx", "-y", package] + args, env) case .uvx(let package, let args, let env): return ("/usr/bin/env", ["uvx", package] + args, env) @@ -530,36 +768,36 @@ actor ACPService { // MARK: - JSON-RPC Framing - private func sendRequest(streamId: UUID, method: String, params: [String: JSONValue]) + private func sendRequest(key: String, method: String, params: [String: JSONValue]) async throws -> JSONValue { - guard streams[streamId] != nil else { throw ACPError.streamClosed } + guard sessions[key] != nil else { throw ACPError.streamClosed } - let id = nextRequestId(streamId: streamId) + let id = nextRequestId(key: key) let frame: [String: JSONValue] = [ "jsonrpc": .string("2.0"), "id": .number(Double(id)), "method": .string(method), "params": .object(params) ] - try writeFrame(streamId: streamId, frame: .object(frame)) - logger.info("[ACP] → \(method, privacy: .public) id=\(id) stream=\(streamId.uuidString.prefix(8), privacy: .public)") + try writeFrame(key: key, frame: .object(frame)) + logger.info("[ACP] → \(method, privacy: .public) id=\(id) key=\(key, privacy: .public)") return try await withCheckedThrowingContinuation { cont in - mutateStream(streamId) { $0.pending[id] = cont } + mutateSession(key) { $0.pending[id] = cont } } } - private func sendResult(streamId: UUID, id: JSONValue, result: JSONValue) { + private func sendResult(key: String, id: JSONValue, result: JSONValue) { let frame: [String: JSONValue] = [ "jsonrpc": .string("2.0"), "id": id, "result": result ] - try? writeFrame(streamId: streamId, frame: .object(frame)) + try? writeFrame(key: key, frame: .object(frame)) } - private func sendError(streamId: UUID, id: JSONValue, code: Int, message: String) { + private func sendError(key: String, id: JSONValue, code: Int, message: String) { let frame: [String: JSONValue] = [ "jsonrpc": .string("2.0"), "id": id, @@ -568,30 +806,30 @@ actor ACPService { "message": .string(message) ]) ] - try? writeFrame(streamId: streamId, frame: .object(frame)) + try? writeFrame(key: key, frame: .object(frame)) } - private func writeFrame(streamId: UUID, frame: JSONValue) throws { - guard let entry = streams[streamId] else { throw ACPError.streamClosed } + private func writeFrame(key: String, frame: JSONValue) throws { + guard let entry = sessions[key] else { throw ACPError.streamClosed } let data = try JSONEncoder().encode(frame) var line = data - line.append(0x0A) // newline + line.append(0x0A) try entry.stdin.write(contentsOf: line) } - private func nextRequestId(streamId: UUID) -> Int { + private func nextRequestId(key: String) -> Int { var id = 0 - mutateStream(streamId) { entry in + mutateSession(key) { entry in id = entry.nextId entry.nextId += 1 } return id } - private func mutateStream(_ streamId: UUID, _ mutate: (inout StreamEntry) -> Void) { - guard var entry = streams[streamId] else { return } + private func mutateSession(_ key: String, _ mutate: (inout SessionEntry) -> Void) { + guard var entry = sessions[key] else { return } mutate(&entry) - streams[streamId] = entry + sessions[key] = entry } // MARK: - Read Loop @@ -603,7 +841,7 @@ actor ACPService { // arrives, and split into newline-delimited frames ourselves. private static func spawnStdoutReader( - streamId: UUID, + key: String, stdout: FileHandle, service: ACPService ) -> Task { @@ -623,12 +861,11 @@ actor ACPService { } return Task.detached { [weak service] in - let shortId = String(streamId.uuidString.prefix(8)) - await service?.logReaderStarted(shortId: shortId) + await service?.logReaderStarted(key: key) var buffer = Data() for await chunk in chunkStream { if Task.isCancelled { - await service?.logReaderCancelled(shortId: shortId) + await service?.logReaderCancelled(key: key) stdout.readabilityHandler = nil return } @@ -638,32 +875,35 @@ actor ACPService { buffer.removeSubrange(buffer.startIndex...nlIdx) guard !lineData.isEmpty, let line = String(data: lineData, encoding: .utf8) else { continue } - await service?.deliverStdoutLine(streamId: streamId, line: line, data: lineData) + await service?.deliverStdoutLine(key: key, line: line, data: lineData) } } - await service?.logReaderEOF(shortId: shortId) + await service?.logReaderEOF(key: key) } } - fileprivate func logReaderStarted(shortId: String) { - logger.info("[ACP] read loop started for stream=\(shortId, privacy: .public)") + fileprivate func logReaderStarted(key: String) { + logger.info("[ACP] read loop started for key=\(key, privacy: .public)") } - fileprivate func logReaderCancelled(shortId: String) { - logger.info("[ACP] read loop cancelled for stream=\(shortId, privacy: .public)") + fileprivate func logReaderCancelled(key: String) { + logger.info("[ACP] read loop cancelled for key=\(key, privacy: .public)") } - fileprivate func logReaderEOF(shortId: String) { - logger.info("[ACP] read loop EOF for stream=\(shortId, privacy: .public)") + fileprivate func logReaderEOF(key: String) { + logger.info("[ACP] read loop EOF for key=\(key, privacy: .public)") } fileprivate func logReaderError(error: Error) { logger.warning("[ACP] read loop error: \(error.localizedDescription, privacy: .public)") } - fileprivate func deliverStdoutLine(streamId: UUID, line: String, data: Data) async { + fileprivate func deliverStdoutLine(key: String, line: String, data: Data) async { + // Resolve the latest canonical key in case the entry has been + // re-keyed (bootstrap key → agent sessionId). + let resolved = aliasToCanonical[key] ?? key logger.info("[ACP][stdout] \(line.prefix(400), privacy: .public)") - await handleIncoming(streamId: streamId, data: data) + await handleIncoming(key: resolved, data: data) } - private func handleIncoming(streamId: UUID, data: Data) async { + private func handleIncoming(key: String, data: Data) async { let value: JSONValue do { value = try JSONDecoder().decode(JSONValue.self, from: data) @@ -675,7 +915,7 @@ actor ACPService { // Response (has "id" and "result"/"error", no "method"). if obj["method"] == nil, let idVal = obj["id"], let idInt = idVal.intValue { - resolveResponse(streamId: streamId, id: idInt, body: obj) + resolveResponse(key: key, id: idInt, body: obj) return } @@ -684,21 +924,19 @@ actor ACPService { let params = obj["params"] ?? .null if let idVal = obj["id"] { - // Server-initiated request: respond. - await handleAgentRequest(streamId: streamId, id: idVal, method: method, params: params) + await handleAgentRequest(key: key, id: idVal, method: method, params: params) } else { - // Notification. - handleAgentNotification(streamId: streamId, method: method, params: params) + handleAgentNotification(key: key, method: method, params: params) } } - private func resolveResponse(streamId: UUID, id: Int, body: [String: JSONValue]) { + private func resolveResponse(key: String, id: Int, body: [String: JSONValue]) { var cont: CheckedContinuation? - mutateStream(streamId) { entry in + mutateSession(key) { entry in cont = entry.pending.removeValue(forKey: id) } guard let cont else { - logger.warning("[ACP] ← response id=\(id) had no pending continuation stream=\(streamId.uuidString.prefix(8), privacy: .public)") + logger.warning("[ACP] ← response id=\(id) had no pending continuation key=\(key, privacy: .public)") return } if let err = body["error"]?.objectValue { @@ -715,29 +953,26 @@ actor ACPService { // MARK: - Agent Notifications - private func handleAgentNotification(streamId: UUID, method: String, params: JSONValue) { + private func handleAgentNotification(key: String, method: String, params: JSONValue) { switch method { case "session/update": - // Drop updates that arrive before `session/prompt` is sent — those - // are the agent's replay of historical content from `session/load` - // and would duplicate prior messages in the UI. - guard streams[streamId]?.acceptingUpdates == true else { + guard sessions[key]?.acceptingUpdates == true else { let kind = params.objectValue?["update"]?.objectValue?["sessionUpdate"]?.stringValue ?? "" logger.info("[ACP] ⟵ pre-prompt session/update dropped kind=\(kind, privacy: .public)") return } - handleSessionUpdate(streamId: streamId, params: params) + handleSessionUpdate(key: key, params: params) default: logger.warning("[ACP] ⟵ unknown notification: \(method, privacy: .public)") } } - private func handleSessionUpdate(streamId: UUID, params: JSONValue) { + private func handleSessionUpdate(key: String, params: JSONValue) { guard let p = params.objectValue, let update = p["update"]?.objectValue, let kind = update["sessionUpdate"]?.stringValue, - let continuation = streams[streamId]?.continuation else { - logger.warning("[ACP] session/update missing fields or no continuation stream=\(streamId.uuidString.prefix(8), privacy: .public)") + let continuation = sessions[key]?.continuation else { + logger.warning("[ACP] session/update missing fields or no continuation key=\(key, privacy: .public)") return } @@ -745,12 +980,6 @@ actor ACPService { case "agent_message_chunk": if let text = update["content"]?.objectValue?["text"]?.stringValue { logger.info("[ACP] ⟵ agent_message_chunk len=\(text.count)") - // Route through the Claude `content_block_delta` / `text_delta` - // pipeline so consecutive chunks accumulate into the same - // streaming assistant bubble via `state.textDeltaBuffer`. The - // `.assistant(.text(_))` path only flushes the first chunk per - // turn (dedup guard in `processStream`), so it was dropping - // every chunk after the first. continuation.yield(.unknown(Self.textDeltaFrame(text))) } else { logger.warning("[ACP] agent_message_chunk had no text content") @@ -759,33 +988,29 @@ actor ACPService { case "agent_thought_chunk": if let text = update["content"]?.objectValue?["text"]?.stringValue { logger.info("[ACP] ⟵ agent_thought_chunk len=\(text.count)") - // Only the `isThinking` flag flip in `handlePartialEvent` is - // wired up for thinking text; emitting a thinking_delta is - // enough to surface the "thinking…" indicator in the UI. continuation.yield(.unknown(Self.thinkingDeltaFrame(text))) } case "plan": let entries = update["entries"]?.arrayValue?.count ?? 0 logger.info("[ACP] ⟵ plan entries=\(entries)") - handlePlanUpdate(streamId: streamId, update: update, continuation: continuation) + handlePlanUpdate(key: key, update: update, continuation: continuation) case "tool_call": - handleToolCall(streamId: streamId, update: update, continuation: continuation) + handleToolCall(key: key, update: update, continuation: continuation) case "tool_call_update": - handleToolCallUpdate(streamId: streamId, update: update, continuation: continuation) + handleToolCallUpdate(key: key, update: update, continuation: continuation) default: logger.warning("[ACP] ⟵ unhandled sessionUpdate kind: \(kind, privacy: .public)") } } - private func handlePlanUpdate(streamId: UUID, update: [String: JSONValue], + private func handlePlanUpdate(key: String, update: [String: JSONValue], continuation: AsyncStream.Continuation) { guard let entries = update["entries"]?.arrayValue else { return } - // Render a markdown checklist so the existing PlanCardView renders it. var markdown = "# Plan\n\n" for entry in entries { guard let obj = entry.objectValue else { continue } @@ -800,7 +1025,7 @@ actor ACPService { markdown += "\(mark)\(content)\n" } - let planId = streams[streamId]?.planSyntheticId ?? "acp-plan" + let planId = sessions[key]?.planSyntheticId ?? "acp-plan" continuation.yield(.assistant(AssistantMessage( role: "assistant", content: [.toolUse( @@ -809,28 +1034,30 @@ actor ACPService { input: ["plan": .string(markdown)] )] ))) - mutateStream(streamId) { $0.planEmitted = true } + mutateSession(key) { $0.planEmitted = true } } - private func handleToolCall(streamId: UUID, update: [String: JSONValue], + private func handleToolCall(key: String, update: [String: JSONValue], continuation: AsyncStream.Continuation) { guard let toolCallId = update["toolCallId"]?.stringValue else { logger.warning("[ACP] tool_call missing toolCallId") return } let title = update["title"]?.stringValue ?? update["kind"]?.stringValue ?? "tool" + let kind = update["kind"]?.stringValue let rawInput = update["rawInput"]?.objectValue ?? [:] - logger.info("[ACP] ⟵ tool_call id=\(toolCallId, privacy: .public) name=\(title, privacy: .public) inputKeys=[\(rawInput.keys.sorted().joined(separator: ","), privacy: .public)]") + let normalized = Self.normalizeToolCall(kind: kind, title: title, update: update, rawInput: rawInput) + logger.info("[ACP] ⟵ tool_call id=\(toolCallId, privacy: .public) name=\(normalized.name, privacy: .public) kind=\(kind ?? "", privacy: .public) inputKeys=[\(normalized.input.keys.sorted().joined(separator: ","), privacy: .public)]") - mutateStream(streamId) { $0.liveToolCalls.insert(toolCallId) } + mutateSession(key) { $0.liveToolCalls.insert(toolCallId) } continuation.yield(.assistant(AssistantMessage( role: "assistant", - content: [.toolUse(id: toolCallId, name: title, input: rawInput)] + content: [.toolUse(id: toolCallId, name: normalized.name, input: normalized.input)] ))) } - private func handleToolCallUpdate(streamId: UUID, update: [String: JSONValue], + private func handleToolCallUpdate(key: String, update: [String: JSONValue], continuation: AsyncStream.Continuation) { guard let toolCallId = update["toolCallId"]?.stringValue else { logger.warning("[ACP] tool_call_update missing toolCallId") @@ -839,6 +1066,24 @@ actor ACPService { let status = update["status"]?.stringValue ?? "completed" logger.info("[ACP] ⟵ tool_call_update id=\(toolCallId, privacy: .public) status=\(status, privacy: .public)") + // If the update carries diff content (ACP allows the agent to attach + // diffs on either tool_call or tool_call_update), re-emit a toolUse + // with the merged input so AppState can persist the file edit when the + // result lands. AppState merges by toolCallId, so this patches the + // existing block in place rather than appending a duplicate. + let diffEntries = Self.diffEntries(in: update) + if !diffEntries.isEmpty { + let kind = update["kind"]?.stringValue + let title = update["title"]?.stringValue ?? kind ?? "tool" + let rawInput = update["rawInput"]?.objectValue ?? [:] + let normalized = Self.normalizeToolCall(kind: kind ?? "edit", title: title, update: update, rawInput: rawInput) + logger.info("[ACP] ⟵ tool_call_update id=\(toolCallId, privacy: .public) carrying diffs=\(diffEntries.count) → patching toolUse name=\(normalized.name, privacy: .public)") + continuation.yield(.assistant(AssistantMessage( + role: "assistant", + content: [.toolUse(id: toolCallId, name: normalized.name, input: normalized.input)] + ))) + } + // Compose tool result text from rawOutput or content[] var resultText = "" if let raw = update["rawOutput"] { @@ -867,54 +1112,54 @@ actor ACPService { // MARK: - Agent Requests (server-initiated) - private func handleAgentRequest(streamId: UUID, id: JSONValue, method: String, params: JSONValue) async { - logger.info("[ACP] ⟵ agent-request \(method, privacy: .public) stream=\(streamId.uuidString.prefix(8), privacy: .public)") + private func handleAgentRequest(key: String, id: JSONValue, method: String, params: JSONValue) async { + logger.info("[ACP] ⟵ agent-request \(method, privacy: .public) key=\(key, privacy: .public)") switch method { case "fs/read_text_file": - await handleFsReadTextFile(streamId: streamId, id: id, params: params) + await handleFsReadTextFile(key: key, id: id, params: params) case "fs/write_text_file": - await handleFsWriteTextFile(streamId: streamId, id: id, params: params) + await handleFsWriteTextFile(key: key, id: id, params: params) case "session/request_permission": - await handleSessionRequestPermission(streamId: streamId, id: id, params: params) + await handleSessionRequestPermission(key: key, id: id, params: params) default: logger.warning("[ACP] unsupported agent-request: \(method, privacy: .public)") - sendError(streamId: streamId, id: id, code: -32601, message: "Method not supported: \(method)") + sendError(key: key, id: id, code: -32601, message: "Method not supported: \(method)") } } - private func handleFsReadTextFile(streamId: UUID, id: JSONValue, params: JSONValue) async { + private func handleFsReadTextFile(key: String, id: JSONValue, params: JSONValue) async { guard let path = params.objectValue?["path"]?.stringValue else { - sendError(streamId: streamId, id: id, code: -32602, message: "Missing path") + sendError(key: key, id: id, code: -32602, message: "Missing path") return } do { let line = params.objectValue?["line"]?.intValue let limit = params.objectValue?["limit"]?.intValue let content = try Self.readTextFile(path: path, line: line, limit: limit) - sendResult(streamId: streamId, id: id, result: .object(["content": .string(content)])) + sendResult(key: key, id: id, result: .object(["content": .string(content)])) } catch { - sendError(streamId: streamId, id: id, code: -32000, message: error.localizedDescription) + sendError(key: key, id: id, code: -32000, message: error.localizedDescription) } } - private func handleFsWriteTextFile(streamId: UUID, id: JSONValue, params: JSONValue) async { + private func handleFsWriteTextFile(key: String, id: JSONValue, params: JSONValue) async { guard let path = params.objectValue?["path"]?.stringValue, let content = params.objectValue?["content"]?.stringValue else { - sendError(streamId: streamId, id: id, code: -32602, message: "Missing path or content") + sendError(key: key, id: id, code: -32602, message: "Missing path or content") return } do { try Self.writeTextFile(path: path, content: content) - sendResult(streamId: streamId, id: id, result: .object([:])) + sendResult(key: key, id: id, result: .object([:])) } catch { - sendError(streamId: streamId, id: id, code: -32000, message: error.localizedDescription) + sendError(key: key, id: id, code: -32000, message: error.localizedDescription) } } - private func handleSessionRequestPermission(streamId: UUID, id: JSONValue, params: JSONValue) async { - guard let entry = streams[streamId] else { - logger.warning("[ACP] permission request arrived for closed stream") - sendError(streamId: streamId, id: id, code: -32000, message: "Stream closed") + private func handleSessionRequestPermission(key: String, id: JSONValue, params: JSONValue) async { + guard let entry = sessions[key] else { + logger.warning("[ACP] permission request arrived for closed session") + sendError(key: key, id: id, code: -32000, message: "Session closed") return } let toolCall = params.objectValue?["toolCall"]?.objectValue ?? [:] @@ -924,12 +1169,11 @@ actor ACPService { logger.info("[ACP] permission request tool=\(toolName, privacy: .public) id=\(toolCallId, privacy: .public)") guard let server = permissionServer else { - // No permission server registered — auto-allow so the agent doesn't hang. let optionId = params.objectValue?["options"]?.arrayValue? .first(where: { $0.objectValue?["kind"]?.stringValue == "allow_once" })? .objectValue?["optionId"]?.stringValue ?? "allow" logger.warning("[ACP] no permission server — auto-allowing \(toolName, privacy: .public) via optionId=\(optionId, privacy: .public)") - sendResult(streamId: streamId, id: id, result: .object([ + sendResult(key: key, id: id, result: .object([ "outcome": .object([ "outcome": .string("selected"), "optionId": .string(optionId) @@ -947,7 +1191,6 @@ actor ACPService { ) logger.info("[ACP] permission decision tool=\(toolName, privacy: .public) decision=\(String(describing: decision), privacy: .public)") - // Map RxCode decision to an ACP option from the options[] array, defaulting to allow_once / reject_once. let options = params.objectValue?["options"]?.arrayValue ?? [] let wantKind: String switch decision { @@ -961,7 +1204,7 @@ actor ACPService { let optionId = chosen?.objectValue?["optionId"]?.stringValue ?? wantKind logger.info("[ACP] permission reply wantKind=\(wantKind, privacy: .public) optionId=\(optionId, privacy: .public)") - sendResult(streamId: streamId, id: id, result: .object([ + sendResult(key: key, id: id, result: .object([ "outcome": .object([ "outcome": .string("selected"), "optionId": .string(optionId) @@ -1012,7 +1255,6 @@ actor ACPService { for entry in opts { guard let entryObj = entry.objectValue else { continue } if let groupOpts = entryObj["options"]?.arrayValue { - // SessionConfigSelectGroup — flatten its options. for groupEntry in groupOpts { if let parsed = parseSelectOption(groupEntry) { flattened.append(parsed) @@ -1032,6 +1274,102 @@ actor ACPService { return nil } + // MARK: - Tool Call Normalization + // + // ACP `tool_call` notifications expose two views of a tool invocation: + // a machine-readable `kind` ("edit", "execute", "read", …) and a + // human-readable `title`. The agent's `rawInput` is opaque (each agent + // chooses its own schema), but the optional `content` array surfaces + // structured `diff` entries (`{path, oldText, newText}`) that we can + // translate into the Claude-shaped `Edit`/`Write`/`MultiEdit` input keys + // the rest of RxCode (sidebar persistence, diff renderer, plan card) + // already understands. + + private struct NormalizedToolCall { + let name: String + let input: [String: JSONValue] + } + + /// Extracts `{type: "diff", path, oldText, newText}` entries from a + /// session/update payload's `content` array. Returns `[]` for non-edit + /// kinds or absent content. + private static func diffEntries(in update: [String: JSONValue]) -> [(path: String, oldText: String?, newText: String)] { + guard let content = update["content"]?.arrayValue else { return [] } + var out: [(path: String, oldText: String?, newText: String)] = [] + for entry in content { + guard let obj = entry.objectValue, + obj["type"]?.stringValue == "diff", + let path = obj["path"]?.stringValue, + let newText = obj["newText"]?.stringValue + else { continue } + let oldText = obj["oldText"]?.stringValue + out.append((path: path, oldText: oldText, newText: newText)) + } + return out + } + + /// Translates an ACP tool_call payload into a Claude-shaped (name, input). + /// Edit-kind calls with diff content become `Edit`/`Write`/`MultiEdit` so + /// `ChatMessage.fileEditHunks` and `editedFilePath` can extract the path + /// and hunks for sidebar persistence. Other kinds fall back to the agent's + /// title + rawInput unchanged. + private static func normalizeToolCall( + kind: String?, + title: String, + update: [String: JSONValue], + rawInput: [String: JSONValue] + ) -> NormalizedToolCall { + let normalizedKind = (kind ?? "").lowercased() + let diffs = diffEntries(in: update) + if normalizedKind == "edit" || !diffs.isEmpty { + if !diffs.isEmpty { + if diffs.count == 1 { + let only = diffs[0] + let oldText = only.oldText ?? "" + if oldText.isEmpty { + return NormalizedToolCall( + name: "Write", + input: [ + "file_path": .string(only.path), + "content": .string(only.newText) + ] + ) + } + return NormalizedToolCall( + name: "Edit", + input: [ + "file_path": .string(only.path), + "old_string": .string(oldText), + "new_string": .string(only.newText) + ] + ) + } + let primaryPath = diffs.first!.path + let edits: [JSONValue] = diffs.filter { $0.path == primaryPath }.map { d in + .object([ + "old_string": .string(d.oldText ?? ""), + "new_string": .string(d.newText) + ]) + } + return NormalizedToolCall( + name: "MultiEdit", + input: [ + "file_path": .string(primaryPath), + "edits": .array(edits) + ] + ) + } + var input = rawInput + if input["file_path"] == nil, + let path = (update["locations"]?.arrayValue?.first?.objectValue?["path"]?.stringValue) { + input["file_path"] = .string(path) + } + return NormalizedToolCall(name: "Edit", input: input) + } + + return NormalizedToolCall(name: title, input: rawInput) + } + /// Wraps a text chunk in the same raw-event shape Claude's CLI emits, so /// `AppState.handlePartialEvent` accumulates it into `state.textDeltaBuffer`. private static func textDeltaFrame(_ text: String) -> String { @@ -1074,9 +1412,7 @@ actor ACPService { // MARK: - Process Lifecycle - private func startStderrReader(streamId: UUID, stderr: FileHandle) { - // Same rationale as `spawnStdoutReader`: avoid `FileHandle.bytes.lines` - // and pump from `readabilityHandler` instead. + private func startStderrReader(key: String, stderr: FileHandle) { let chunkStream = AsyncStream { continuation in stderr.readabilityHandler = { handle in let data = handle.availableData @@ -1104,29 +1440,34 @@ actor ACPService { let lineData = buffer.subdata(in: buffer.startIndex.. AsyncStream { + guard let spec = request.acpSpec else { + return AsyncStream { c in + c.yield(.user(UserMessage( + toolUseId: nil, + content: "No ACP client configured. Add one in Settings → ACP Clients.", + isError: true + ))) + c.yield(.result(ResultEvent( + durationMs: nil, totalCostUsd: nil, + sessionId: request.sessionId ?? request.clientSessionKey, + isError: true, totalTurns: nil, usage: nil, contextWindow: nil + ))) + c.finish() + } + } + return send( + streamId: request.streamId, + prompt: request.prompt, + cwd: request.cwd, + sessionId: request.sessionId, + model: request.model, + spec: spec, + permissionMode: request.permissionMode, + clientSessionKey: request.clientSessionKey, + mcpServers: request.acpMCPServers + ) + } +} diff --git a/RxCode/Services/ClaudeService.swift b/RxCode/Services/ClaudeService.swift index 85dcc086..76528739 100644 --- a/RxCode/Services/ClaudeService.swift +++ b/RxCode/Services/ClaudeService.swift @@ -1038,3 +1038,24 @@ actor ClaudeCodeServer { } typealias ClaudeService = ClaudeCodeServer + +// MARK: - AgentBackend Conformance + +extension ClaudeCodeServer: AgentBackend { + nonisolated var provider: AgentProvider { .claudeCode } + nonisolated var staticCapabilities: CapabilitySet { AgentProvider.claudeCode.staticCapabilities } + + func send(_ request: BackendSendRequest) -> AsyncStream { + send( + streamId: request.streamId, + prompt: request.prompt, + cwd: request.cwd, + sessionId: request.sessionId, + model: request.model, + effort: request.effort, + hookSettingsPath: request.hookSettingsPath, + mcpConfigPath: request.mcpClaudeConfigPath, + permissionMode: request.permissionMode + ) + } +} diff --git a/RxCode/Services/CodexAppServer.swift b/RxCode/Services/CodexAppServer.swift index 6cfc41b8..614970b3 100644 --- a/RxCode/Services/CodexAppServer.swift +++ b/RxCode/Services/CodexAppServer.swift @@ -37,6 +37,14 @@ actor CodexAppServer { private var rateLimitsFetchTask: Task? private let rateLimitsCacheTTL: TimeInterval = 300 private var notificationMethodCounts: [String: Int] = [:] + /// Reference to the permission server for in-band permission requests. + /// Wired once at app init; the `AgentBackend.send(_:)` adapter pulls it + /// from here so callers don't have to pass it on every turn. + private weak var permissionServer: PermissionServer? + + func setPermissionServer(_ server: PermissionServer) { + self.permissionServer = server + } private static var candidatePaths: [String] { let home = FileManager.default.homeDirectoryForCurrentUser.path @@ -1196,3 +1204,40 @@ private enum AppStateModelFormatter { .joined(separator: " ") } } + +// MARK: - AgentBackend Conformance + +extension CodexAppServer: AgentBackend { + nonisolated var provider: AgentProvider { .codex } + nonisolated var staticCapabilities: CapabilitySet { AgentProvider.codex.staticCapabilities } + + func send(_ request: BackendSendRequest) -> AsyncStream { + guard let permissionServer else { + logger.error("[Codex] send called before setPermissionServer wired") + return AsyncStream { c in + c.yield(.user(UserMessage( + toolUseId: nil, + content: "Codex backend not initialized (missing permission server).", + isError: true + ))) + c.yield(.result(ResultEvent( + durationMs: nil, totalCostUsd: nil, + sessionId: request.sessionId ?? request.clientSessionKey, + isError: true, totalTurns: nil, usage: nil, contextWindow: nil + ))) + c.finish() + } + } + return send( + streamId: request.streamId, + prompt: request.prompt, + cwd: request.cwd, + threadId: request.sessionId, + model: request.model, + permissionMode: request.permissionMode, + planMode: request.planMode, + mcpConfigOverrides: request.mcpCodexOverrides, + permissionServer: permissionServer + ) + } +} diff --git a/RxCode/Services/IDEServer/AppState+IDEToolHandling.swift b/RxCode/Services/IDEServer/AppState+IDEToolHandling.swift new file mode 100644 index 00000000..0aeb783f --- /dev/null +++ b/RxCode/Services/IDEServer/AppState+IDEToolHandling.swift @@ -0,0 +1,191 @@ +import Foundation +import RxCodeCore + +// MARK: - IDEToolHandling Conformance +// +// AppState exposes IDE-side tools to MCP-capable agents through the bridge +// `IDEMCPServer`. The conformance lives in its own file so AppState.swift +// stays focused on chat/session state. +// +// Per-session capability gating: +// • Polyfill tools (`ide__set_todos`, `ide__ask_user`) are filtered out +// for backends that natively support the feature — see +// `IDEToolRegistry.tools(for:)`. +// • IDE-only tools (`ide__get_running_jobs` etc.) always appear so any +// agent can introspect editor state. +// +// `sessionKey` is the AppState session key the agent is bound to. The +// listener allocates one port per session, so this is unambiguous. + +extension AppState: IDEToolHandling { + public func ideAvailableTools(forSession sessionKey: String) async -> [IDETool] { + let provider = await MainActor.run { sessionStates[sessionKey]?.agentProvider } ?? .acp + let caps = await backend(for: provider).capabilities(for: sessionKey) + return IDEToolRegistry.tools(for: caps) + } + + public func ideHandleToolCall( + name: String, + arguments: JSONValue, + sessionKey: String + ) async throws -> JSONValue { + switch name { + case "ide__set_todos": + return try await handleSetTodos(arguments: arguments, sessionKey: sessionKey) + case "ide__get_running_jobs": + return await handleGetRunningJobs() + case "ide__get_job_output": + throw IDEToolError.notSupported("ide__get_job_output is not yet implemented") + case "ide__get_threads": + return await handleGetThreads(arguments: arguments) + case "ide__get_thread_detail": + return try await handleGetThreadDetail(arguments: arguments) + case "ide__get_usage": + return await handleGetUsage() + case "ide__ask_user": + throw IDEToolError.notSupported("ide__ask_user polyfill not implemented yet — surface the question as plain assistant text instead.") + default: + throw IDEToolError.unknownTool(name) + } + } + + // MARK: - Handlers + + @MainActor + private func handleSetTodos(arguments: JSONValue, sessionKey: String) throws -> JSONValue { + guard let todosArray = arguments["todos"]?.arrayValue else { + throw IDEToolError.invalidArguments("missing 'todos' array") + } + let parsed: [TodoItem] = todosArray.enumerated().compactMap { idx, entry -> TodoItem? in + guard + let dict = entry.objectValue, + let content = dict["content"]?.stringValue, + let statusRaw = dict["status"]?.stringValue, + let status = TodoItem.Status(rawValue: statusRaw) + else { return nil } + let activeForm = dict["activeForm"]?.stringValue ?? content + return TodoItem(id: idx, content: content, activeForm: activeForm, status: status) + } + threadStore.upsertTodoSnapshot(sessionId: sessionKey, items: parsed) + return textResult("Recorded \(parsed.count) todo(s).") + } + + @MainActor + private func handleGetRunningJobs() -> JSONValue { + let entries: [JSONValue] = runService.activeTasks.map { task in + .object([ + "id": .string(task.id.uuidString), + "profile_name": .string(task.profile.name), + "project_id": .string(task.project.id.uuidString), + "started_at": .string(ISO8601DateFormatter().string(from: task.startedAt)), + "status": .string(String(describing: task.status)), + ]) + } + return jsonTextResult(.array(entries)) + } + + @MainActor + private func handleGetThreads(arguments: JSONValue) -> JSONValue { + let projectFilter: UUID? = { + if let s = arguments["project_id"]?.stringValue { return UUID(uuidString: s) } + return nil + }() + let summaries = threadStore.loadAllSummaries() + let filtered = summaries + .filter { projectFilter == nil || $0.projectId == projectFilter } + .filter { !$0.isArchived } + .sorted { $0.updatedAt > $1.updatedAt } + .prefix(50) + let entries: [JSONValue] = filtered.map { s in + .object([ + "id": .string(s.id), + "title": .string(s.title), + "project_id": .string(s.projectId.uuidString), + "updated_at": .string(ISO8601DateFormatter().string(from: s.updatedAt)), + "agent_provider": .string(s.agentProvider.rawValue), + ]) + } + return jsonTextResult(.array(entries)) + } + + @MainActor + private func handleGetThreadDetail(arguments: JSONValue) throws -> JSONValue { + guard let id = arguments["thread_id"]?.stringValue else { + throw IDEToolError.invalidArguments("missing 'thread_id'") + } + guard let thread = threadStore.fetch(id: id) else { + throw IDEToolError.handlerFailed("No thread with id \(id)") + } + // ChatThread is the SwiftData summary record; the full message + // body is persisted by ChatSession on disk. Return the metadata + // here — a future revision can hydrate ChatSession via + // PersistenceService.loadSession for richer detail. + return jsonTextResult(.object([ + "id": .string(thread.id), + "title": .string(thread.title), + "project_id": .string(thread.projectId.uuidString), + "created_at": .string(ISO8601DateFormatter().string(from: thread.createdAt)), + "updated_at": .string(ISO8601DateFormatter().string(from: thread.updatedAt)), + "model": thread.model.map { .string($0) } ?? .null, + "agent_provider": thread.agentProviderRaw.map { .string($0) } ?? .null, + ])) + } + + private func handleGetUsage() async -> JSONValue { + let provider = await MainActor.run { selectedAgentProvider } + let usage = await rateLimitUsage(for: provider, forceRefresh: false) + guard let usage else { + return jsonTextResult(.object(["available": .bool(false)])) + } + return jsonTextResult(.object([ + "available": .bool(true), + "provider": .string(provider.rawValue), + "five_hour_percent": .number(usage.fiveHourPercent), + "seven_day_percent": .number(usage.sevenDayPercent), + "twenty_four_hour_percent": usage.twentyFourHourPercent.map { .number($0) } ?? .null, + "five_hour_resets_at": usage.fiveHourResetsAt.map { .string(ISO8601DateFormatter().string(from: $0)) } ?? .null, + "seven_day_resets_at": usage.sevenDayResetsAt.map { .string(ISO8601DateFormatter().string(from: $0)) } ?? .null, + ])) + } + + // MARK: - Formatting helpers + + fileprivate func textResult(_ text: String) -> JSONValue { + .object([ + "content": .array([ + .object([ + "type": .string("text"), + "text": .string(text), + ]) + ]) + ]) + } + + fileprivate func jsonTextResult(_ value: JSONValue) -> JSONValue { + textResult(prettyJSON(value)) + } + + fileprivate func prettyJSON(_ value: JSONValue) -> String { + if let any = jsonValueToAny(value), + (JSONSerialization.isValidJSONObject(any) || any is [Any]), + let data = try? JSONSerialization.data(withJSONObject: any, options: [.prettyPrinted, .sortedKeys]), + let s = String(data: data, encoding: .utf8) { + return s + } + return value.description + } + + fileprivate func jsonValueToAny(_ value: JSONValue) -> Any? { + switch value { + case .null: return NSNull() + case .bool(let b): return b + case .number(let n): return n + case .string(let s): return s + case .array(let arr): return arr.map { jsonValueToAny($0) ?? NSNull() } + case .object(let dict): + var out: [String: Any] = [:] + for (k, v) in dict { out[k] = jsonValueToAny(v) ?? NSNull() } + return out + } + } +} diff --git a/RxCode/Services/IDEServer/IDEMCPServer.swift b/RxCode/Services/IDEServer/IDEMCPServer.swift new file mode 100644 index 00000000..f1f60576 --- /dev/null +++ b/RxCode/Services/IDEServer/IDEMCPServer.swift @@ -0,0 +1,429 @@ +import Foundation +import Network +import RxCodeCore +import os + +/// Hosts a local MCP server that polyfills missing editor capabilities for +/// agents whose native toolset lacks them (currently: ACP's `ask_user` / +/// `todos`), and exposes always-IDE-only introspection (running jobs, +/// thread history, usage) to every backend. +/// +/// Agents speak to the server by running `nc 127.0.0.1 ` as their MCP +/// stdio command. The actor owns one parent `NWListener` per allocated +/// session (each on its own port) so the same agent process can't cross +/// session boundaries; releasing the session shuts that listener down. +actor IDEMCPServer { + + // MARK: - Constants + + private static let basePort: UInt16 = 19847 + private static let maxPort: UInt16 = 19946 + private static let messageMaxLength = 1 << 20 // 1 MiB + private static let logger = Logger(subsystem: "com.claudework", category: "IDEMCPServer") + + // MARK: - Session State + + /// One allocated listener per RxCode session. Created on + /// `allocate(sessionKey:capabilities:)`, torn down on `release`. + private struct Allocation { + let port: UInt16 + let listener: NWListener + let sessionKey: String + let capabilities: CapabilitySet + var connections: Set = [] + } + private var allocations: [String: Allocation] = [:] + /// Port → sessionKey, so connection accept can resolve back. + private var portIndex: [UInt16: String] = [:] + private var nextPort: UInt16 = IDEMCPServer.basePort + + private weak var handler: (any IDEToolHandling)? + + init() {} + + func setHandler(_ handler: any IDEToolHandling) { + self.handler = handler + } + + // MARK: - Bridge Command + + /// Build the `(command, args)` an ACP agent should run as its MCP + /// server child. The child is a perl one-liner that pipes its stdin to + /// our TCP listener and our reply stream back to its stdout — perl is + /// always at `/usr/bin/perl` on macOS and is far more reliable than + /// BSD `nc` for bidirectional line-buffered forwarding (some `nc` + /// builds close the write side on stdin EOF detection, which kills + /// long-lived MCP sessions). + static func bridgeCommand(forPort port: UInt16) -> (command: String, args: [String]) { + let script = """ + use IO::Socket::INET;use IO::Select;$|=1;\ + my($h,$p)=@ARGV;my $s=IO::Socket::INET->new(PeerAddr=>$h,PeerPort=>$p,Proto=>"tcp")or die $!;\ + $s->autoflush(1);binmode STDIN;binmode STDOUT;binmode $s;\ + my $sel=IO::Select->new($s,\\*STDIN);\ + while(my @r=$sel->can_read){for my $fh(@r){my $buf;my $n=sysread($fh,$buf,4096);exit 0 unless $n;\ + if($fh==$s){syswrite(STDOUT,$buf);}else{syswrite($s,$buf);}}} + """ + return ("/usr/bin/perl", ["-e", script, "127.0.0.1", String(port)]) + } + + // MARK: - Allocate / Release + + /// Reserve a port + listener for `sessionKey`. Returns the port that + /// should be embedded in the MCP server `args` (`["127.0.0.1", ""]`). + /// If the session is already allocated, returns the existing port. + func allocate(sessionKey: String, capabilities: CapabilitySet) async -> UInt16? { + if let existing = allocations[sessionKey] { + return existing.port + } + for _ in 0..<(Self.maxPort - Self.basePort + 1) { + let candidate = nextPort + nextPort = nextPort == Self.maxPort ? Self.basePort : nextPort + 1 + if portIndex[candidate] != nil { continue } + do { + let listener = try makeListener(port: candidate) + var alloc = Allocation( + port: candidate, + listener: listener, + sessionKey: sessionKey, + capabilities: capabilities + ) + _ = alloc + allocations[sessionKey] = Allocation( + port: candidate, + listener: listener, + sessionKey: sessionKey, + capabilities: capabilities + ) + portIndex[candidate] = sessionKey + listener.newConnectionHandler = { [weak self] connection in + guard let self else { return } + Task { await self.accept(connection: connection, port: candidate) } + } + listener.start(queue: .global(qos: .userInitiated)) + Self.logger.info("[IDE] allocated port \(candidate) for session=\(sessionKey, privacy: .public)") + return candidate + } catch { + Self.logger.warning("[IDE] port \(candidate) unavailable: \(error.localizedDescription, privacy: .public)") + continue + } + } + Self.logger.error("[IDE] no available port for session=\(sessionKey, privacy: .public)") + return nil + } + + func release(sessionKey: String) async { + guard let alloc = allocations.removeValue(forKey: sessionKey) else { return } + portIndex.removeValue(forKey: alloc.port) + alloc.listener.cancel() + Self.logger.info("[IDE] released port \(alloc.port) for session=\(sessionKey, privacy: .public)") + } + + // MARK: - Listener + + private func makeListener(port: UInt16) throws -> NWListener { + let params = NWParameters.tcp + params.allowLocalEndpointReuse = true + params.requiredInterfaceType = .loopback + return try NWListener(using: params, on: NWEndpoint.Port(rawValue: port)!) + } + + private func accept(connection: NWConnection, port: UInt16) async { + guard let sessionKey = portIndex[port] else { + connection.cancel() + return + } + guard let capabilities = allocations[sessionKey]?.capabilities else { + connection.cancel() + return + } + let id = UUID() + allocations[sessionKey]?.connections.insert(id) + connection.start(queue: .global(qos: .userInitiated)) + Self.logger.info("[IDE] accepted connection id=\(id.uuidString, privacy: .public) on port \(port) session=\(sessionKey, privacy: .public)") + + await runMCP( + connection: connection, + connectionId: id, + sessionKey: sessionKey, + capabilities: capabilities + ) + + connection.cancel() + allocations[sessionKey]?.connections.remove(id) + Self.logger.info("[IDE] closed connection id=\(id.uuidString, privacy: .public)") + } + + // MARK: - MCP Protocol Loop + + private func runMCP( + connection: NWConnection, + connectionId: UUID, + sessionKey: String, + capabilities: CapabilitySet + ) async { + var pendingBuffer = Data() + while true { + // Drain any complete lines already buffered. + while let newline = pendingBuffer.firstIndex(of: 0x0A) { + let lineData = pendingBuffer[pendingBuffer.startIndex.. Self.messageMaxLength { + Self.logger.error("[IDE] message exceeded limit, closing connection id=\(connectionId.uuidString, privacy: .public)") + return + } + } catch { + Self.logger.info("[IDE] connection read ended: \(error.localizedDescription, privacy: .public)") + return + } + } + } + + private func readChunk(connection: NWConnection) async throws -> Data { + try await withCheckedThrowingContinuation { continuation in + connection.receive(minimumIncompleteLength: 1, maximumLength: 65_536) { data, _, isComplete, error in + if let error { + continuation.resume(throwing: error) + } else if let data, !data.isEmpty { + continuation.resume(returning: data) + } else if isComplete { + continuation.resume(returning: Data()) + } else { + continuation.resume(returning: Data()) + } + } + } + } + + // MARK: - Per-Message Dispatch + + private func processLine( + data: Data.SubSequence, + connection: NWConnection, + sessionKey: String, + capabilities: CapabilitySet + ) async { + let bytes = Data(data) + guard + let raw = try? JSONSerialization.jsonObject(with: bytes, options: [.fragmentsAllowed]), + let dict = raw as? [String: Any] + else { + Self.logger.warning("[IDE] malformed JSON-RPC line, dropping") + return + } + let id = dict["id"] + let method = dict["method"] as? String + let params = dict["params"] as? [String: Any] ?? [:] + + // Notifications (no id) — we only care about `notifications/initialized` + if id == nil { + return + } + + switch method { + case "initialize": + await reply( + id: id!, + result: [ + "protocolVersion": "2024-11-05", + "capabilities": [ + "tools": [String: Any]() + ], + "serverInfo": [ + "name": "rxcode-ide", + "version": "1" + ] + ], + on: connection + ) + + case "tools/list": + let tools = await currentTools(sessionKey: sessionKey, capabilities: capabilities) + await reply( + id: id!, + result: ["tools": tools.map(Self.toolDescriptor)], + on: connection + ) + + case "tools/call": + let name = params["name"] as? String ?? "" + let arguments = params["arguments"] as? [String: Any] ?? [:] + let argsValue = JSONValue.fromAny(arguments) + do { + let handler = self.handler + guard let handler else { + throw IDEToolError.handlerFailed("IDE handler not attached") + } + let result = try await handler.ideHandleToolCall( + name: name, + arguments: argsValue, + sessionKey: sessionKey + ) + let payload = Self.wrapToolResult(result) + await reply(id: id!, result: payload, on: connection) + } catch let error as IDEToolError { + await replyError(id: id!, code: Self.errorCode(for: error), message: Self.errorMessage(for: error), on: connection) + } catch { + await replyError(id: id!, code: -32603, message: error.localizedDescription, on: connection) + } + + default: + await replyError(id: id!, code: -32601, message: "Method not found: \(method ?? "")", on: connection) + } + } + + private func currentTools(sessionKey: String, capabilities: CapabilitySet) async -> [IDETool] { + if let handler { + return await handler.ideAvailableTools(forSession: sessionKey) + } + return IDEToolRegistry.tools(for: capabilities) + } + + // MARK: - Reply + + private func reply(id: Any, result: Any, on connection: NWConnection) async { + var payload: [String: Any] = [ + "jsonrpc": "2.0", + "id": id, + "result": result + ] + _ = payload + let body: [String: Any] = [ + "jsonrpc": "2.0", + "id": id, + "result": result + ] + await send(json: body, on: connection) + } + + private func replyError(id: Any, code: Int, message: String, on connection: NWConnection) async { + let body: [String: Any] = [ + "jsonrpc": "2.0", + "id": id, + "error": ["code": code, "message": message] + ] + await send(json: body, on: connection) + } + + private func send(json: [String: Any], on connection: NWConnection) async { + guard + JSONSerialization.isValidJSONObject(json), + var bytes = try? JSONSerialization.data(withJSONObject: json, options: []) + else { + Self.logger.error("[IDE] failed to encode response, dropping") + return + } + bytes.append(0x0A) + await withCheckedContinuation { (continuation: CheckedContinuation) in + connection.send(content: bytes, completion: .contentProcessed { _ in + continuation.resume() + }) + } + } + + // MARK: - Helpers + + private static func toolDescriptor(_ tool: IDETool) -> [String: Any] { + [ + "name": tool.name, + "description": tool.description, + "inputSchema": tool.inputSchema.toAny() ?? [:] + ] + } + + /// Wrap a handler's `JSONValue` result in the MCP tool-call response + /// envelope: a `content` array with `text` blocks. Handlers may either + /// pass a string scalar (auto-wrapped) or the full object form. + private static func wrapToolResult(_ value: JSONValue) -> [String: Any] { + if case .object(let dict) = value, dict["content"] != nil { + return value.toAny() as? [String: Any] ?? [:] + } + let text: String + switch value { + case .string(let s): text = s + case .null: text = "" + default: + if + let data = try? JSONSerialization.data( + withJSONObject: value.toAny() ?? [:], + options: [.prettyPrinted, .sortedKeys] + ), + let s = String(data: data, encoding: .utf8) + { + text = s + } else { + text = "\(value)" + } + } + return [ + "content": [ + ["type": "text", "text": text] + ] + ] + } + + private static func errorCode(for error: IDEToolError) -> Int { + switch error { + case .unknownTool: return -32601 + case .invalidArguments: return -32602 + case .notSupported: return -32004 + case .handlerFailed: return -32603 + } + } + + private static func errorMessage(for error: IDEToolError) -> String { + switch error { + case .unknownTool(let n): return "Unknown tool: \(n)" + case .invalidArguments(let m): return "Invalid arguments: \(m)" + case .notSupported(let m): return "Not supported: \(m)" + case .handlerFailed(let m): return m + } + } +} + +// MARK: - JSONValue <-> Any helpers + +private extension JSONValue { + static func fromAny(_ any: Any) -> JSONValue { + if let s = any as? String { return .string(s) } + if let b = any as? Bool { return .bool(b) } + if let n = any as? NSNumber { + return .number(n.doubleValue) + } + if let arr = any as? [Any] { + return .array(arr.map { Self.fromAny($0) }) + } + if let dict = any as? [String: Any] { + var out: [String: JSONValue] = [:] + for (k, v) in dict { out[k] = Self.fromAny(v) } + return .object(out) + } + return .null + } + + func toAny() -> Any? { + switch self { + case .null: return NSNull() + case .bool(let b): return b + case .number(let n): return n + case .string(let s): return s + case .array(let arr): return arr.map { $0.toAny() ?? NSNull() } + case .object(let dict): + var out: [String: Any] = [:] + for (k, v) in dict { out[k] = v.toAny() ?? NSNull() } + return out + } + } +} diff --git a/RxCode/Services/MCPService.swift b/RxCode/Services/MCPService.swift index a0c2fcb3..f3d7315b 100644 --- a/RxCode/Services/MCPService.swift +++ b/RxCode/Services/MCPService.swift @@ -144,6 +144,58 @@ actor MCPService { } } + /// Build the `mcpServers` array passed to ACP's `session/new` RPC. + /// ACP v1 only specifies stdio transport; http/sse servers are silently + /// dropped. Disabled servers are filtered out. If `bridgeCommand` is + /// non-nil an additional entry `rxcode-ide` is prepended — that's the + /// local MCP polyfill server. + func acpMCPServers( + projectPath: String?, + bridgeCommand: (command: String, args: [String])? = nil + ) async -> [JSONValue] { + // IMPORTANT: do NOT add a `"type": "stdio"` field. The ACP spec's + // anyOf union uses presence of `type` as the stdio-vs-http/sse + // discriminator, and known agent implementations (OpenCode) bail + // into the HTTP parsing path if `type` is present at all. + var entries: [JSONValue] = [] + if let bridge = bridgeCommand { + entries.append(.object([ + "name": .string("rxcode-ide"), + "command": .string(bridge.command), + "args": .array(bridge.args.map { .string($0) }), + "env": .array([]) + ])) + } + do { + let records = try enabledRecords(projectPath: projectPath) + let userEntries: [JSONValue] = records + .filter { $0.transport == .stdio } + .sorted { $0.name < $1.name } + .map { record -> JSONValue in + let envPairs: [JSONValue] = record.env + .sorted { $0.key < $1.key } + .map { key, value in + .object([ + "name": .string(key), + "value": .string(value), + ]) + } + return .object([ + "name": .string(record.name), + "command": .string(record.command ?? ""), + "args": .array(record.args.map { .string($0) }), + "env": .array(envPairs), + ]) + } + entries.append(contentsOf: userEntries) + } catch { + logger.warning("Failed to build ACP MCP servers: \(error.localizedDescription)") + } + let names = entries.compactMap { $0["name"]?.stringValue }.joined(separator: ", ") + logger.info("[ACP-MCP] built \(entries.count, privacy: .public) mcpServers entries for project=\(projectPath ?? "", privacy: .public) [\(names, privacy: .public)]") + return entries + } + func codexConfigOverrides(projectPath: String?) async -> [String] { do { let config = try loadConfig() diff --git a/RxCode/Views/Sidebar/HistoryListView.swift b/RxCode/Views/Sidebar/HistoryListView.swift index fac90ec8..002a066a 100644 --- a/RxCode/Views/Sidebar/HistoryListView.swift +++ b/RxCode/Views/Sidebar/HistoryListView.swift @@ -261,7 +261,7 @@ struct HistoryListView: View { // MARK: - Display Model - private struct DisplaySession: Identifiable, Equatable { + struct DisplaySession: Identifiable, Equatable { let id: String let projectId: UUID let title: String @@ -279,30 +279,55 @@ struct HistoryListView: View { } } - private static func sessionOrder( + static func sessionOrder( _ a: ChatSession.Summary, _ b: ChatSession.Summary ) -> Bool { if a.isPinned != b.isPinned { return a.isPinned } return a.updatedAt > b.updatedAt } + /// Filter + sort the summary feed for the History sidebar. + /// + /// `projectId == nil` returns the all-projects feed (deduplicated by id). + /// `projectId != nil` returns sessions scoped to that project. In either + /// mode the result includes ONLY summaries whose `isArchived` matches + /// `showArchived`, so archived chats never leak into the active list and + /// active chats never leak into the Archived view. + static func filteredSummaries( + from summaries: [ChatSession.Summary], + projectId: UUID?, + showArchived: Bool + ) -> [ChatSession.Summary] { + if let projectId { + return summaries + .filter { $0.projectId == projectId && $0.isArchived == showArchived } + .sorted { sessionOrder($0, $1) } + } + var seen = Set() + return summaries + .filter { $0.isArchived == showArchived } + .sorted { sessionOrder($0, $1) } + .filter { seen.insert($0.id).inserted } + } + private var currentProjectSessions: [DisplaySession] { guard let projectId = windowState.selectedProject?.id else { return [] } let streamingIds = appState.backgroundStreamingSessionIds(in: windowState) - return appState.allSessionSummaries - .filter { $0.projectId == projectId && $0.isArchived == showArchived } - .sorted { Self.sessionOrder($0, $1) } - .map { summary in - DisplaySession( - id: summary.id, - projectId: summary.projectId, - title: summary.title, - updatedAt: summary.updatedAt, - isPinned: summary.isPinned, - isBackgroundStreaming: streamingIds.contains(summary.id), - projectName: nil - ) - } + return Self.filteredSummaries( + from: appState.allSessionSummaries, + projectId: projectId, + showArchived: showArchived + ).map { summary in + DisplaySession( + id: summary.id, + projectId: summary.projectId, + title: summary.title, + updatedAt: summary.updatedAt, + isPinned: summary.isPinned, + isBackgroundStreaming: streamingIds.contains(summary.id), + projectName: nil + ) + } } private var allProjectSessions: [DisplaySession] { @@ -310,22 +335,21 @@ struct HistoryListView: View { uniqueKeysWithValues: appState.projects.map { ($0.id, $0.name) } ) let streamingIds = appState.backgroundStreamingSessionIds(in: windowState) - var seen = Set() - return appState.allSessionSummaries - .filter { $0.isArchived == showArchived } - .sorted { Self.sessionOrder($0, $1) } - .filter { seen.insert($0.id).inserted } - .map { summary in - DisplaySession( - id: summary.id, - projectId: summary.projectId, - title: summary.title, - updatedAt: summary.updatedAt, - isPinned: summary.isPinned, - isBackgroundStreaming: streamingIds.contains(summary.id), - projectName: projectNames[summary.projectId] - ) - } + return Self.filteredSummaries( + from: appState.allSessionSummaries, + projectId: nil, + showArchived: showArchived + ).map { summary in + DisplaySession( + id: summary.id, + projectId: summary.projectId, + title: summary.title, + updatedAt: summary.updatedAt, + isPinned: summary.isPinned, + isBackgroundStreaming: streamingIds.contains(summary.id), + projectName: projectNames[summary.projectId] + ) + } } // MARK: - Helpers diff --git a/RxCode/Views/Sidebar/ProjectTreeView.swift b/RxCode/Views/Sidebar/ProjectTreeView.swift index 30940b6c..ba4ea00d 100644 --- a/RxCode/Views/Sidebar/ProjectTreeView.swift +++ b/RxCode/Views/Sidebar/ProjectTreeView.swift @@ -376,12 +376,11 @@ private struct ProjectChatsList: View { @State private var showsAllThreads = false private var sessions: [ChatSession.Summary] { - appState.allSessionSummaries - .filter { $0.projectId == project.id } - .sorted { a, b in - if a.isPinned != b.isPinned { return a.isPinned } - return a.updatedAt > b.updatedAt - } + HistoryListView.filteredSummaries( + from: appState.allSessionSummaries, + projectId: project.id, + showArchived: false + ) } private var visibleSessions: [ChatSession.Summary] { diff --git a/RxCodeTests/HistoryListArchiveFilterTests.swift b/RxCodeTests/HistoryListArchiveFilterTests.swift new file mode 100644 index 00000000..711a7d65 --- /dev/null +++ b/RxCodeTests/HistoryListArchiveFilterTests.swift @@ -0,0 +1,311 @@ +import XCTest +import SwiftUI +import ViewInspector +import RxCodeCore +@testable import RxCode + +/// Guards the contract that the History sidebar hides sessions whose +/// `isArchived` flag does not match the current `showArchived` toggle. +/// +/// Two layers of coverage: +/// +/// 1. Direct unit tests of `HistoryListView.filteredSummaries(...)` — the pure +/// function that drives every row the sidebar renders. If the filter +/// excludes archived summaries, the view physically cannot render a row +/// for them. +/// +/// 2. ViewInspector tests against `ArchiveFilterPreview`, a small helper view +/// that pipes the same filter through a `List` of `TypewriterTitleText`. +/// These verify that the filtered slice is what actually reaches the +/// rendered SwiftUI tree. +/// +/// The reason we don't `.inspect()` `HistoryListView` directly: ViewInspector +/// 0.10.3 cannot resolve `@Environment(MyObservable.self)` values through +/// `.environment(_:)` modifiers during body evaluation, which crashes with +/// "No Observable object of type WindowState found." The preview below has +/// no environment dependencies, so the inspector can walk its body cleanly. +@MainActor +final class HistoryListArchiveFilterTests: XCTestCase { + + // MARK: - Pure filter behavior + + func testFilter_currentProjectScope_excludesArchivedWhenShowArchivedFalse() { + let projectId = UUID() + let active = summary(id: "a", projectId: projectId, title: "Active", isArchived: false) + let archived = summary(id: "b", projectId: projectId, title: "Archived", isArchived: true) + + let result = HistoryListView.filteredSummaries( + from: [active, archived], + projectId: projectId, + showArchived: false + ) + + XCTAssertEqual( + result.map(\.id), + ["a"], + "Active History list must drop archived summaries for the current project" + ) + } + + func testFilter_currentProjectScope_includesOnlyArchivedWhenShowArchivedTrue() { + let projectId = UUID() + let active = summary(id: "a", projectId: projectId, title: "Active", isArchived: false) + let archived = summary(id: "b", projectId: projectId, title: "Archived", isArchived: true) + + let result = HistoryListView.filteredSummaries( + from: [active, archived], + projectId: projectId, + showArchived: true + ) + + XCTAssertEqual( + result.map(\.id), + ["b"], + "Archived view must drop active summaries for the current project" + ) + } + + func testFilter_currentProjectScope_excludesOtherProjects() { + let projectId = UUID() + let otherProjectId = UUID() + let inScope = summary(id: "a", projectId: projectId, title: "Mine", isArchived: false) + let otherProject = summary(id: "b", projectId: otherProjectId, title: "Theirs", isArchived: false) + + let result = HistoryListView.filteredSummaries( + from: [inScope, otherProject], + projectId: projectId, + showArchived: false + ) + + XCTAssertEqual(result.map(\.id), ["a"]) + } + + func testFilter_allProjectsScope_dropsArchivedAcrossProjects() { + let projectA = UUID() + let projectB = UUID() + let summaries = [ + summary(id: "a-active", projectId: projectA, title: "A", isArchived: false), + summary(id: "a-archived", projectId: projectA, title: "Aa", isArchived: true), + summary(id: "b-active", projectId: projectB, title: "B", isArchived: false), + summary(id: "b-archived", projectId: projectB, title: "Bb", isArchived: true), + ] + + let result = HistoryListView.filteredSummaries( + from: summaries, + projectId: nil, + showArchived: false + ) + + XCTAssertEqual(Set(result.map(\.id)), ["a-active", "b-active"]) + } + + func testFilter_allProjectsScope_dropsActiveWhenShowArchivedTrue() { + let projectA = UUID() + let projectB = UUID() + let summaries = [ + summary(id: "a-active", projectId: projectA, title: "A", isArchived: false), + summary(id: "a-archived", projectId: projectA, title: "Aa", isArchived: true), + summary(id: "b-active", projectId: projectB, title: "B", isArchived: false), + summary(id: "b-archived", projectId: projectB, title: "Bb", isArchived: true), + ] + + let result = HistoryListView.filteredSummaries( + from: summaries, + projectId: nil, + showArchived: true + ) + + XCTAssertEqual(Set(result.map(\.id)), ["a-archived", "b-archived"]) + } + + func testFilter_pinnedSummariesAlwaysSortFirst_withinSameArchiveBucket() { + let projectId = UUID() + let now = Date() + let pinnedOld = summary( + id: "pinned", projectId: projectId, title: "Pinned", + isArchived: false, isPinned: true, + updatedAt: now.addingTimeInterval(-1000) + ) + let recentActive = summary( + id: "recent", projectId: projectId, title: "Recent", + isArchived: false, isPinned: false, + updatedAt: now + ) + + let result = HistoryListView.filteredSummaries( + from: [recentActive, pinnedOld], + projectId: projectId, + showArchived: false + ) + + XCTAssertEqual( + result.map(\.id), + ["pinned", "recent"], + "Within the active bucket, pinned must sort before unpinned regardless of recency" + ) + } + + // MARK: - ViewInspector — rendered list reflects the filter + + func testInspector_activeView_doesNotRenderArchivedTitle() throws { + let projectId = UUID() + let preview = ArchiveFilterPreview( + summaries: [ + summary(id: "a", projectId: projectId, title: "Active Chat", isArchived: false), + summary(id: "b", projectId: projectId, title: "Archived Chat", isArchived: true), + ], + projectId: projectId, + showArchived: false + ) + + let titles = try renderedTitles(in: preview) + + XCTAssertTrue( + titles.contains("Active Chat"), + "Active sessions must reach the rendered list. Titles: \(titles)" + ) + XCTAssertFalse( + titles.contains("Archived Chat"), + "Archived sessions must NOT render in the sidebar when showArchived=false. Titles: \(titles)" + ) + } + + func testInspector_archivedView_doesNotRenderActiveTitle() throws { + let projectId = UUID() + let preview = ArchiveFilterPreview( + summaries: [ + summary(id: "a", projectId: projectId, title: "Active Chat", isArchived: false), + summary(id: "b", projectId: projectId, title: "Archived Chat", isArchived: true), + ], + projectId: projectId, + showArchived: true + ) + + let titles = try renderedTitles(in: preview) + + XCTAssertTrue( + titles.contains("Archived Chat"), + "Archived sessions must reach the rendered list when toggled on. Titles: \(titles)" + ) + XCTAssertFalse( + titles.contains("Active Chat"), + "Active sessions must NOT render in the Archived view. Titles: \(titles)" + ) + } + + func testInspector_emptyState_renderedWhenAllSummariesAreArchived() throws { + let projectId = UUID() + let preview = ArchiveFilterPreview( + summaries: [ + summary(id: "a", projectId: projectId, title: "Archived 1", isArchived: true), + summary(id: "b", projectId: projectId, title: "Archived 2", isArchived: true), + ], + projectId: projectId, + showArchived: false + ) + + let titles = try renderedTitles(in: preview) + XCTAssertTrue( + titles.isEmpty, + "When every summary is archived, the active list must render nothing. Titles: \(titles)" + ) + + let texts = try preview.inspect() + .findAll(ViewType.Text.self) + .compactMap { try? $0.string() } + XCTAssertTrue( + texts.contains("No chat history"), + "Empty state copy should match HistoryListView's. Texts: \(texts)" + ) + } + + func testInspector_allProjectsScope_dropsArchivedAcrossProjects() throws { + let projectA = UUID() + let projectB = UUID() + let preview = ArchiveFilterPreview( + summaries: [ + summary(id: "a-active", projectId: projectA, title: "Alpha Active", isArchived: false), + summary(id: "a-archived", projectId: projectA, title: "Alpha Archived", isArchived: true), + summary(id: "b-active", projectId: projectB, title: "Beta Active", isArchived: false), + summary(id: "b-archived", projectId: projectB, title: "Beta Archived", isArchived: true), + ], + projectId: nil, + showArchived: false + ) + + let titles = try renderedTitles(in: preview) + + XCTAssertTrue(titles.contains("Alpha Active")) + XCTAssertTrue(titles.contains("Beta Active")) + XCTAssertFalse( + titles.contains("Alpha Archived"), + "All-projects active list must not contain archived rows. Titles: \(titles)" + ) + XCTAssertFalse( + titles.contains("Beta Archived"), + "All-projects active list must not contain archived rows. Titles: \(titles)" + ) + } + + // MARK: - Helpers + + private func renderedTitles(in preview: ArchiveFilterPreview) throws -> [String] { + try preview.inspect() + .findAll(TypewriterTitleText.self) + .map { try $0.actualView().title } + } + + private func summary( + id: String, + projectId: UUID, + title: String, + isArchived: Bool, + isPinned: Bool = false, + updatedAt: Date = Date() + ) -> ChatSession.Summary { + ChatSession.Summary( + id: id, + projectId: projectId, + title: title, + createdAt: updatedAt, + updatedAt: updatedAt, + isPinned: isPinned, + isArchived: isArchived, + archivedAt: isArchived ? updatedAt : nil + ) + } +} + +// MARK: - Test helper view + +/// Minimal stand-in for `HistoryListView`'s session list. Uses the same +/// `filteredSummaries` function and renders titles through `TypewriterTitleText` +/// just like the real row, but takes inputs directly so ViewInspector can walk +/// its body without needing `AppState` / `WindowState` in the environment. +private struct ArchiveFilterPreview: View { + let summaries: [ChatSession.Summary] + let projectId: UUID? + let showArchived: Bool + + private var filtered: [ChatSession.Summary] { + HistoryListView.filteredSummaries( + from: summaries, + projectId: projectId, + showArchived: showArchived + ) + } + + var body: some View { + if filtered.isEmpty { + // Matches HistoryListView's active-list empty copy so the empty + // state test asserts against the production string verbatim. + Text(showArchived ? "No archived chats" : "No chat history") + } else { + List(filtered, id: \.id) { summary in + TypewriterTitleText( + title: summary.title.prefix(1).uppercased() + summary.title.dropFirst() + ) + } + } + } +}