From 47970b14682698b3d0e2564e6dda5d9336a87801 Mon Sep 17 00:00:00 2001 From: XU Ke Date: Thu, 15 Oct 2026 00:20:30 +0800 Subject: [PATCH 01/11] fix: allow switching harnesses during draft startup --- .../canvas/canvas-conversation-surface.tsx | 2 +- src/components/chat/agent-selector.tsx | 1 + .../conversation-detail-panel.tsx | 8 +- src/contexts/acp-connections-context.test.tsx | 97 +++++++++++++++++++ src/contexts/acp-connections-context.tsx | 8 ++ 5 files changed, 112 insertions(+), 4 deletions(-) diff --git a/src/components/canvas/canvas-conversation-surface.tsx b/src/components/canvas/canvas-conversation-surface.tsx index 8341d0fc71..1f63f3c310 100644 --- a/src/components/canvas/canvas-conversation-surface.tsx +++ b/src/components/canvas/canvas-conversation-surface.tsx @@ -799,7 +799,7 @@ export function CanvasConversationSurface({ defaultAgentType={agentType} onSelect={onAgentTypeChange} onFallback={onAgentTypeChange} - disabled={conn.status === "connecting"} + // Draft cards can change agents during connection startup. /> )} diff --git a/src/components/chat/agent-selector.tsx b/src/components/chat/agent-selector.tsx index f80f1b63ad..9dc0fb3303 100644 --- a/src/components/chat/agent-selector.tsx +++ b/src/components/chat/agent-selector.tsx @@ -51,6 +51,7 @@ interface AgentSelectorProps { onFallback?: (agentType: AgentType) => void onAgentsLoaded?: (agents: AcpAgentInfo[]) => void onOpenAgentsSettings?: () => void + /** Lock a bound conversation; draft pickers stay usable during startup. */ disabled?: boolean /** * Where the pill sits inside the row it is given. The selector now spans the diff --git a/src/components/conversations/conversation-detail-panel.tsx b/src/components/conversations/conversation-detail-panel.tsx index e20b673a4e..6e59d200b5 100644 --- a/src/components/conversations/conversation-detail-panel.tsx +++ b/src/components/conversations/conversation-detail-panel.tsx @@ -718,7 +718,6 @@ const ConversationTabView = memo(function ConversationTabView({ useEffect(() => { isViewerRef.current = conn.isViewer }, [conn.isViewer]) - const isConnecting = connStatus === "connecting" // The tab's connection is keyed by a stable tabId, but agent switching is // async — and for a not-installed target, connect()'s preflight throws BEFORE // it tears down the old connection. So `conn` can still describe the PREVIOUS @@ -2426,7 +2425,9 @@ const ConversationTabView = memo(function ConversationTabView({ ) }} onOpenAgentsSettings={handleOpenAgentsSettings} - disabled={isConnecting || dbConversationId != null} + // Drafts remain switchable while connecting; saved + // conversations retain the agent they were created with. + disabled={dbConversationId != null} /> {composerBlockedMessage ? ( @@ -2506,7 +2507,8 @@ const ConversationTabView = memo(function ConversationTabView({ ) }} onOpenAgentsSettings={handleOpenAgentsSettings} - disabled={isConnecting || dbConversationId != null} + // Same draft/saved rule as the welcome-mode picker above. + disabled={dbConversationId != null} /> {composerBlockedMessage ? (
diff --git a/src/contexts/acp-connections-context.test.tsx b/src/contexts/acp-connections-context.test.tsx index 3efe44ae95..6846488fe8 100644 --- a/src/contexts/acp-connections-context.test.tsx +++ b/src/contexts/acp-connections-context.test.tsx @@ -6324,6 +6324,103 @@ describe("connect() is observable while it is still in flight", () => { }) }) +describe("a switch that lands while another agent is still connecting", () => { + // The composer's agent row stays clickable for the whole `connecting` + // window, so a second pick can arrive while the first agent is still + // spawning. The user asked for the NEWEST one: `connect()` must park that + // request, kill what the abandoned attempt spawned, and never let it reach + // the store — otherwise a stale agent claims the tab and the switch the user + // watched happen silently reverts. + it("kills the abandoned spawn and runs the agent picked last", async () => { + await mountProvider() + const spawns: Array<() => void> = [] + h.acpConnect.mockImplementation( + (agentType: string) => + new Promise((res) => { + spawns.push(() => res(`${agentType}-conn`)) + }) + ) + + let first: Promise | undefined + let second: Promise | undefined + await act(async () => { + first = h.actions!.connect(TAB, "claude_code", "/tmp/x") + await h.actions!.connect(TAB, "codex", "/tmp/x") + second = h.actions!.connect(TAB, "gemini", "/tmp/x") + }) + + // Parked, not racing: exactly one agent is being spawned. + expect(h.acpConnect).toHaveBeenCalledTimes(1) + expect(h.acpConnect.mock.calls[0]![0]).toBe("claude_code") + expect(h.store!.getConnectPending(TAB)?.agentType).toBe("gemini") + + await act(async () => { + spawns[0]!() + await first + }) + await act(async () => {}) + + // The abandoned attempt is torn down and never registered... + expect(h.acpDisconnect).toHaveBeenCalledWith("claude_code-conn") + expect(h.store!.getConnection(TAB)).toBeUndefined() + // ...and the newest request is the one actually connecting now. + expect(h.acpConnect).toHaveBeenCalledTimes(2) + expect(h.acpConnect.mock.calls[1]![0]).toBe("gemini") + + await act(async () => { + spawns[1]!() + await second + }) + + expect(h.store!.getConnection(TAB)?.agentType).toBe("gemini") + }) + + it("replaces a registered connection before its startup finishes", async () => { + await mountProvider() + h.acpConnect + .mockResolvedValueOnce("starting-claude") + .mockResolvedValueOnce("starting-codex") + await act(async () => { + await h.actions!.connect(TAB, "claude_code", "/tmp/x") + }) + // No connected event has arrived from the first agent's handshake. + expect(h.store!.getConnection(TAB)?.status).toBe("connecting") + + await act(async () => { + await h.actions!.connect(TAB, "codex", "/tmp/x") + }) + expect(h.acpDisconnect).toHaveBeenCalledWith("starting-claude") + expect(h.acpConnect).toHaveBeenCalledTimes(2) + expect(h.store!.getConnection(TAB)?.agentType).toBe("codex") + expect(h.store!.getConnection(TAB)?.status).toBe("connecting") + }) + + it("does not notify a failed preflight for a choice the user replaced", async () => { + await mountProvider() + let rejectStatus!: (error: Error) => void + h.acpGetAgentStatus.mockImplementationOnce( + () => + new Promise((_resolve, reject) => { + rejectStatus = reject + }) + ) + let first: Promise | undefined + await act(async () => { + first = h.actions!.connect(TAB, "claude_code", "/tmp/x") + await h.actions!.connect(TAB, "codex", "/tmp/x") + }) + await act(async () => { + rejectStatus(new Error("old agent configuration failed")) + await first + }) + + expect(h.store!.getConnection(TAB)?.agentType).toBe("codex") + expect(h.store!.getConnectError(TAB)).toBeUndefined() + expect(h.recordAlert).not.toHaveBeenCalled() + expect(h.toastError).not.toHaveBeenCalled() + }) +}) + describe("connect failures land on the surface that asked", () => { // A failed connect never produces a store entry, so its reason used to go // only to the status-bar alerts — for a saved conversation the ONLY place it diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index a7622c5c34..bd8793a6e2 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -6574,6 +6574,11 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { lastConnectParamsRef.current.set(contextKey, request) if (connectingKeysRef.current.has(contextKey)) { pendingConnectRequestsRef.current.set(contextKey, request) + setConnectPending(contextKey, { + agentType, + workingDir: workingDir ?? null, + }) + setConnectError(contextKey, null) return } connectingKeysRef.current.add(contextKey) @@ -6597,6 +6602,9 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // rejection must not report a failure over whatever it does next. const publishConnectError = (info: ConnectErrorInfo) => { if (abandonedKeysRef.current.has(contextKey)) return + const pendingRequest = pendingConnectRequestsRef.current.get(contextKey) + if (pendingRequest && !sameConnectRequest(pendingRequest, request)) + return setConnectError(contextKey, info) notifyConnectError(contextKey, info) } From 3b722ddd722c912c33cc9eb117f4a2e547e145e9 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 07:40:50 +0800 Subject: [PATCH 02/11] fix(acp): name the agent actually connecting in the status-bar row A draft can now switch agents while it connects, and the status reads `connecting` straight through the switch, so the row minted for the first agent kept naming it. The row is keyed on the leg and the agent coming up (the entry's own, else the one asked for), and is retired and re-minted when either changes. --- .../use-connection-lifecycle.opening.test.ts | 58 +++++++++++++++++-- src/hooks/use-connection-lifecycle.ts | 35 ++++++++--- 2 files changed, 81 insertions(+), 12 deletions(-) diff --git a/src/hooks/use-connection-lifecycle.opening.test.ts b/src/hooks/use-connection-lifecycle.opening.test.ts index d2961de2c8..f642aef7f9 100644 --- a/src/hooks/use-connection-lifecycle.opening.test.ts +++ b/src/hooks/use-connection-lifecycle.opening.test.ts @@ -48,18 +48,24 @@ const tasks = vi.hoisted(() => ({ }, })) -vi.mock("@/contexts/task-context", () => ({ - useTaskContext: () => ({ +vi.mock("@/contexts/task-context", () => { + // One identity for the hook's whole life, like the real provider's + // `useCallback`s. A fresh `removeTask` per render re-runs the hook's + // unmount cleanup on every rerender, which retires the very row under test. + const context = { addTask: (id: string, label: string, description?: string) => tasks.added.push({ id, label, description }), updateTask: (id: string, update: { status?: string }) => tasks.updated.push({ id, ...update }), removeTask: (id: string) => tasks.removed.push(id), - }), -})) + } + return { useTaskContext: () => context } +}) const conn = vi.hoisted(() => ({ status: null as string | null, + /** The agent of the store entry at this key; null while there is none. */ + agentType: null as string | null, selectorsReady: false, hasCachedSelectors: false, })) @@ -67,6 +73,7 @@ const conn = vi.hoisted(() => ({ vi.mock("@/hooks/use-connection", () => ({ useConnection: () => ({ status: conn.status, + agentType: conn.agentType, selectorsReady: conn.selectorsReady, hasCachedSelectors: conn.hasCachedSelectors, connect: vi.fn().mockResolvedValue(undefined), @@ -84,6 +91,7 @@ vi.mock("@/hooks/use-connection", () => ({ })) import { useConnectionLifecycle } from "@/hooks/use-connection-lifecycle" +import type { AgentType } from "@/lib/types" function renderLifecycle(preparing: boolean) { return renderHook( @@ -103,6 +111,7 @@ describe("useConnectionLifecycle opening legs", () => { beforeEach(() => { tasks.reset() conn.status = null + conn.agentType = null conn.selectorsReady = false conn.hasCachedSelectors = false }) @@ -147,6 +156,47 @@ describe("useConnectionLifecycle opening legs", () => { expect(tasks.live()).toEqual([]) }) + it("renames the connect row when a draft switches agents before it connects", () => { + // A draft's agent picker stays usable while it connects, and the status + // reads `connecting` straight through the switch: the old agent's entry, + // then the new one's in-flight marker, then its entry. The row has to + // follow the agent coming up rather than the status, or it goes on naming + // the agent the user just left. + conn.status = "connecting" + conn.agentType = "claude_code" + const { rerender } = renderHook( + (props: { agentType: AgentType }) => + useConnectionLifecycle({ + contextKey: "ctx-1", + agentType: props.agentType, + isActive: false, + }), + { initialProps: { agentType: "claude_code" as AgentType } } + ) + const first = tasks.live()[0] + expect(first.label).toBe('tasks.connectingTitle:{"agent":"Claude Code"}') + + // Picked, but the old agent's entry is still the one coming up: the + // switch has not torn it down yet (or Codex's preflight failed and left + // it running). The row keeps naming what is actually connecting. + act(() => rerender({ agentType: "codex" })) + expect(tasks.live()).toEqual([first]) + + // The old entry is gone; Codex's in-flight marker carries the status. + conn.agentType = null + act(() => rerender({ agentType: "codex" })) + expect(tasks.removed).toContain(first.id) + const second = tasks.live()[0] + expect(tasks.live().map((t) => t.label)).toEqual([ + 'tasks.connectingTitle:{"agent":"Codex"}', + ]) + + // Codex's own entry lands, still connecting: same agent, same row. + conn.agentType = "codex" + act(() => rerender({ agentType: "codex" })) + expect(tasks.live()).toEqual([second]) + }) + it("adds nothing while idle — a settled, disconnected tab is not opening", () => { const { result } = renderLifecycle(false) expect(tasks.live()).toEqual([]) diff --git a/src/hooks/use-connection-lifecycle.ts b/src/hooks/use-connection-lifecycle.ts index 5cf1d40b91..296441fe50 100644 --- a/src/hooks/use-connection-lifecycle.ts +++ b/src/hooks/use-connection-lifecycle.ts @@ -273,15 +273,23 @@ export function useConnectionLifecycle({ // "connecting" — `connect()` is in flight: agent spawn, ACP `initialize`, // then `session/resume|load|new` // (The third leg, session init after `connected`, has its own task below.) - // A task's label is fixed when it is added, so a leg change retires the old - // row and mints a new one rather than leaving stale wording on screen. + // A task's label is fixed when it is added, so a change to what it names + // retires the old row and mints a new one rather than leaving stale wording + // on screen. That is the leg, and also the agent: a draft can switch agents + // while it connects, and the status reads `connecting` straight through it. + // The agent named is the one actually coming up — the entry's own while one + // is there (the switch has yet to tear it down, or a failed preflight left + // it running), else the one asked for. + const connAgentType = conn.agentType const taskIdRef = useRef(null) - const taskPhaseRef = useRef<"preparing" | "connecting" | null>(null) + const taskLabelKeyRef = useRef(null) useEffect(() => { const phase = status === "connecting" ? "connecting" : preparing ? "preparing" : null - if (phase !== null && phase === taskPhaseRef.current) return - taskPhaseRef.current = phase + const taskAgentType = connAgentType ?? agentType + const labelKey = phase === null ? null : `${phase}:${taskAgentType}` + if (labelKey !== null && labelKey === taskLabelKeyRef.current) return + taskLabelKeyRef.current = labelKey if (taskIdRef.current) { const finished = taskIdRef.current taskIdRef.current = null @@ -300,9 +308,11 @@ export function useConnectionLifecycle({ } } if (phase === null) return - const id = `acp-${phase}-${Date.now()}` + // The agent is in the id too: a switch can retire a row and mint its + // replacement within the same millisecond. + const id = `acp-${phase}-${taskAgentType}-${Date.now()}` taskIdRef.current = id - const agent = getAgentLabel(agentType) + const agent = getAgentLabel(taskAgentType) addTask( id, phase === "connecting" @@ -313,7 +323,16 @@ export function useConnectionLifecycle({ : t("tasks.preparingDescription") ) updateTask(id, { status: "running" }) - }, [status, preparing, addTask, updateTask, removeTask, agentType, t]) + }, [ + status, + preparing, + addTask, + updateTask, + removeTask, + agentType, + connAgentType, + t, + ]) const clearSelectorTask = useCallback(() => { if (selectorTaskIdRef.current) { From 2329bf14195d34582702492294e27edd06b4d54d Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 07:40:50 +0800 Subject: [PATCH 03/11] docs(acp): describe parked requests in the pending-connect contract --- src/contexts/acp-connections-context.tsx | 23 +++++++++++++++++------ 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index bd8793a6e2..b9158c2cd3 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -3044,6 +3044,10 @@ function connectionsReducer( * that map would have to learn about a half-entry. It is a separate, reactive * side table read only by `useConnection` (which reports it as `connecting`) * and the composer's status chip. + * + * It names what the surface is waiting FOR: when a newer `connect()` is parked + * behind the one in flight (a draft switching agents while it connects), it is + * that newest request, which the in-flight call only runs ahead of. */ export interface ConnectPendingInfo { agentType: AgentType @@ -3074,7 +3078,8 @@ export interface ConnectErrorInfo { interface InternalStore { connections: ConnectionsMap - /** contextKey → the in-flight `connect()` for it (see ConnectPendingInfo). */ + /** contextKey → the in-flight `connect()` for it, or the newest request + * parked behind it (see ConnectPendingInfo). */ connectPending: Map /** contextKey → why its last `connect()` failed (see ConnectErrorInfo). */ connectErrors: Map @@ -3087,9 +3092,10 @@ interface InternalStore { export interface ConnectionStoreApi { getConnection(key: string): ConnectionState | undefined - /** The in-flight `connect()` for this key, or undefined when none is. The - * returned object is reference-stable for the lifetime of that connect, so - * it is safe as a `useSyncExternalStore` snapshot. */ + /** The in-flight `connect()` for this key (or the newest request parked + * behind it), or undefined when none is. The returned object is + * reference-stable until the next change, so it is safe as a + * `useSyncExternalStore` snapshot. */ getConnectPending(key: string): ConnectPendingInfo | undefined /** Why the last `connect()` for this key failed, or undefined. Reference- * stable until the next change, like `getConnectPending`. */ @@ -6573,6 +6579,10 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // `reconnect()` has nothing else to go on. lastConnectParamsRef.current.set(contextKey, request) if (connectingKeysRef.current.has(contextKey)) { + // Parked: the call in flight runs this next, unless a newer request + // replaces it first. The surface is waiting on THIS request from now + // on, so it is what the pending marker names — a draft that switched + // agents mid-connect shows the agent picked last, not the one it left. pendingConnectRequestsRef.current.set(contextKey, request) setConnectPending(contextKey, { agentType, @@ -6598,8 +6608,9 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { // too. setConnectError(contextKey, null) // Publish THIS attempt's failure — unless the surface let go of the - // attempt meanwhile (`disconnect()` marks it abandoned): a late - // rejection must not report a failure over whatever it does next. + // attempt meanwhile (`disconnect()` marks it abandoned) or parked a + // different request behind it: a late rejection must not report a + // failure over whatever it does next. const publishConnectError = (info: ConnectErrorInfo) => { if (abandonedKeysRef.current.has(contextKey)) return const pendingRequest = pendingConnectRequestsRef.current.get(contextKey) From b56b41f86c5a25293d59fa491314c484ce5f6fba Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 07:55:14 +0800 Subject: [PATCH 04/11] fix(canvas): keep a draft card off a connection bound to another agent A draft card can switch agents, and connect() tears the previous agent's connection down only once the new pick passes its preflight; a pick that is not installed never does. Until then the card presented that connection as the selected agent's: its status, mode and config chips and commands, a mode change saved its mode list under the new agent, and a send created the conversation for the new agent while delivering the prompt to the old one. The card now reconciles against its selected agent like the tab view does: the composer reads connecting (or disconnected when the agent is not installed), selectors come from the selected agent's cache, and sends are refused. --- .../canvas-conversation-surface.test.tsx | 277 ++++++++++++++++++ .../canvas/canvas-conversation-surface.tsx | 74 ++++- 2 files changed, 337 insertions(+), 14 deletions(-) create mode 100644 src/components/canvas/canvas-conversation-surface.test.tsx diff --git a/src/components/canvas/canvas-conversation-surface.test.tsx b/src/components/canvas/canvas-conversation-surface.test.tsx new file mode 100644 index 0000000000..3a07fd3894 --- /dev/null +++ b/src/components/canvas/canvas-conversation-surface.test.tsx @@ -0,0 +1,277 @@ +/** + * A draft card's agent picker stays usable while the card connects, and + * `connect()` tears the previous agent's connection down only once the new + * pick gets past its preflight. Until then — and for good when the pick can't + * connect at all (not installed) — the connection under the card's key still + * belongs to the agent the user just left, often already `connected`. + * + * These pin what the card shows and does in that state: the composer must not + * offer the old agent's session as the picked one's (its status, its mode and + * config chips, its commands), a send must not reach it, and a mode change must + * not file the old agent's mode list under the new agent's saved preference. + * The heavy children (shell, transcript, picker) are stubbed; the card's own + * wiring is real. + */ +import { act, cleanup, render } from "@testing-library/react" +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" +import type { AcpAgentInfo, PromptDraft } from "@/lib/types" + +vi.mock("next-intl", () => ({ + useTranslations: () => (key: string) => key, +})) + +const shell = vi.hoisted(() => ({ + props: null as Record | null, +})) +vi.mock("@/components/chat/conversation-shell", () => ({ + ConversationShell: (props: Record) => { + shell.props = props + return <>{props.children as React.ReactNode} + }, +})) + +const picker = vi.hoisted(() => ({ + props: null as Record | null, +})) +vi.mock("@/components/chat/agent-selector", () => ({ + AgentSelector: (props: Record) => { + picker.props = props + return null + }, +})) + +vi.mock("@/components/message/message-list-view", () => ({ + MessageListView: () => null, +})) + +const cachedSelectors = vi.hoisted( + () => + new Map< + string, + { modes: unknown; configOptions: unknown; availableCommands?: unknown } + >() +) +vi.mock("@/contexts/acp-connections-context", () => { + // One identity for the card's whole life, like the real provider's actions: + // the live-message sink effect depends on it. + const actions = { + registerLiveMessageSink: () => () => {}, + answerQuestion: vi.fn(), + answerPlanApproval: vi.fn(), + dismissSessionFailures: vi.fn(), + } + return { + useAcpActions: () => actions, + getCachedSelectors: (agentType: string) => + cachedSelectors.get(agentType) ?? null, + } +}) + +const lifecycle = vi.hoisted(() => ({ + conn: {} as Record, + handleSend: vi.fn(), +})) +vi.mock("@/hooks/use-connection-lifecycle", () => ({ + useConnectionLifecycle: () => ({ + conn: lifecycle.conn, + modeLoading: false, + configOptionsLoading: false, + selectorsLoading: false, + autoConnectError: null, + handleFocus: vi.fn(), + handleSend: lifecycle.handleSend, + handleSetConfigOption: vi.fn(), + handleCancel: vi.fn(), + handleRespondPermission: vi.fn(), + }), +})) + +vi.mock("@/hooks/use-conversation-detail", () => ({ + useConversationDetail: () => ({ + detail: null, + loading: false, + error: null, + acpLoadError: null, + }), +})) + +const registry = vi.hoisted(() => ({ agents: [] as AcpAgentInfo[] })) +vi.mock("@/hooks/use-acp-agents", () => ({ + useAcpAgents: () => ({ + agents: registry.agents, + fresh: true, + refresh: vi.fn(), + }), +})) + +const api = vi.hoisted(() => ({ + acpStopAsyncTask: vi.fn(), + createChatConversation: vi.fn(), + createConversation: vi.fn(), + createChatDir: vi.fn(), +})) +vi.mock("@/lib/api", () => api) + +vi.mock("@/lib/notify", () => ({ notify: vi.fn() })) + +const prefs = vi.hoisted(() => ({ + getSavedModeId: vi.fn((): string | null => null), + saveModePreference: vi.fn(), +})) +vi.mock("@/lib/selector-prefs-storage", () => prefs) + +import { CanvasConversationSurface } from "@/components/canvas/canvas-conversation-surface" + +const CLAUDE_MODES = { + current_mode_id: "default", + available_modes: [ + { id: "default", name: "Default" }, + { id: "plan", name: "Plan" }, + ], +} +const CLAUDE_CONFIG = [ + { + id: "model", + name: "Model", + type: "select", + current_value: "sonnet", + options: [{ value: "sonnet", name: "Sonnet" }], + }, +] +const CODEX_MODES = { + current_mode_id: "auto", + available_modes: [{ id: "auto", name: "Auto" }], +} + +function agent( + agentType: string, + installedVersion: string | null +): AcpAgentInfo { + return { + agent_type: agentType, + enabled: true, + available: true, + installed_version: installedVersion, + is_acp_adapter: false, + } as unknown as AcpAgentInfo +} + +function connection(agentType: string, status: string) { + return { + connectionId: `${agentType}-conn`, + agentType, + status, + sessionId: `${agentType}-session`, + modes: agentType === "claude_code" ? CLAUDE_MODES : null, + configOptions: agentType === "claude_code" ? CLAUDE_CONFIG : null, + availableCommands: + agentType === "claude_code" ? [{ name: "review", description: "" }] : [], + } +} + +function renderDraftCard() { + return render( + + ) +} + +const draft: PromptDraft = { + blocks: [{ type: "text", text: "hello" }], + displayText: "hello", +} + +describe("a draft card whose connection still belongs to the previous agent", () => { + beforeEach(() => { + shell.props = null + picker.props = null + cachedSelectors.clear() + registry.agents = [agent("claude_code", "1.0.0"), agent("codex", "2.0.0")] + lifecycle.handleSend.mockReset() + api.createConversation.mockReset() + api.createChatConversation.mockReset() + prefs.saveModePreference.mockReset() + }) + afterEach(() => cleanup()) + + it("keeps the picker usable while the draft is still connecting", () => { + lifecycle.conn = connection("codex", "connecting") + renderDraftCard() + expect(picker.props).not.toBeNull() + expect(picker.props?.disabled).toBeFalsy() + }) + + it("shows the picked agent's own session once it is the one connected", () => { + lifecycle.conn = { ...connection("codex", "connected"), modes: CODEX_MODES } + renderDraftCard() + expect(shell.props?.status).toBe("connected") + expect(shell.props?.modes).toEqual(CODEX_MODES.available_modes) + }) + + it("presents the switch as still connecting, with the picked agent's chips", () => { + // Codex is installed, so its connect is on its way and will replace this. + cachedSelectors.set("codex", { modes: CODEX_MODES, configOptions: null }) + lifecycle.conn = connection("claude_code", "connected") + renderDraftCard() + + expect(shell.props?.status).toBe("connecting") + expect(shell.props?.modes).toEqual(CODEX_MODES.available_modes) + expect(shell.props?.configOptions).toEqual([]) + expect(shell.props?.selectedModeId).toBe("auto") + expect(shell.props?.availableCommands).toEqual([]) + }) + + it("reads as disconnected when the picked agent is not installed", () => { + // The preflight refused Codex before tearing Claude's connection down, so + // that connection stays here, live, for as long as Codex is selected. + registry.agents = [agent("claude_code", "1.0.0"), agent("codex", null)] + lifecycle.conn = connection("claude_code", "connected") + renderDraftCard() + expect(shell.props?.status).toBe("disconnected") + }) + + it("does not send to the previous agent or file a conversation for it", async () => { + registry.agents = [agent("claude_code", "1.0.0"), agent("codex", null)] + lifecycle.conn = connection("claude_code", "connected") + renderDraftCard() + + const onSend = shell.props?.onSend as (draft: PromptDraft) => void + await act(async () => { + onSend(draft) + }) + + expect(api.createConversation).not.toHaveBeenCalled() + expect(api.createChatConversation).not.toHaveBeenCalled() + expect(lifecycle.handleSend).not.toHaveBeenCalled() + }) + + it("does not save the previous agent's modes as the picked agent's preference", () => { + lifecycle.conn = connection("claude_code", "connected") + renderDraftCard() + + const onModeChange = shell.props?.onModeChange as (id: string) => void + act(() => onModeChange("plan")) + + expect(prefs.saveModePreference).not.toHaveBeenCalled() + }) + + it("still sends once the connection is the picked agent's", async () => { + api.createConversation.mockResolvedValue(42) + lifecycle.conn = connection("codex", "connected") + renderDraftCard() + + const onSend = shell.props?.onSend as (draft: PromptDraft) => void + await act(async () => { + onSend(draft) + }) + + expect(api.createConversation).toHaveBeenCalledWith(7, "codex", "hello") + expect(lifecycle.handleSend).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/components/canvas/canvas-conversation-surface.tsx b/src/components/canvas/canvas-conversation-surface.tsx index 1f63f3c310..595386483b 100644 --- a/src/components/canvas/canvas-conversation-surface.tsx +++ b/src/components/canvas/canvas-conversation-surface.tsx @@ -13,7 +13,11 @@ import { AgentSelector } from "@/components/chat/agent-selector" import { ConversationShell } from "@/components/chat/conversation-shell" import type { ConversationFolderPickerOverride } from "@/components/chat/conversation-context-bar" import { MessageListView } from "@/components/message/message-list-view" -import { useAcpActions } from "@/contexts/acp-connections-context" +import { + getCachedSelectors, + useAcpActions, +} from "@/contexts/acp-connections-context" +import { useAcpAgents } from "@/hooks/use-acp-agents" import { useConnectionLifecycle } from "@/hooks/use-connection-lifecycle" import { useConversationDetail } from "@/hooks/use-conversation-detail" import { @@ -388,6 +392,42 @@ export function CanvasConversationSurface({ const connStatus = conn.status const connSessionId = conn.sessionId + // A draft card can switch agents, and `connect()` tears the previous agent's + // connection down only once the new pick gets past its preflight. Until then + // — and for good when the pick can't connect at all — the connection under + // this key still belongs to the agent the user just left, often already + // `connected`. Never present it as the selected agent's: its mode and config + // chips are not this agent's, and a send would reach the agent the user left + // while the conversation row is filed under the one they picked. Same rule as + // the tab view's `connIsForOtherAgent` (conversation-detail-panel.tsx). + const connIsForOtherAgent = + conn.agentType != null && conn.agentType !== agentType + // Enabled and platform-available, yet nothing installed: the preflight + // refuses such an agent before it tears anything down, so a switch to one + // never gets past the old connection. + const { agents: acpAgents } = useAcpAgents() + const selectedAgentNotInstalled = useMemo(() => { + const info = acpAgents.find((a) => a.agent_type === agentType) + return ( + info != null && info.enabled && info.available && !info.installed_version + ) + }, [acpAgents, agentType]) + // What the composer shows. Only ever differs from `connStatus` while the + // connection belongs to another agent: still `connecting` while the switch + // can complete, `disconnected` when the selected agent can't connect at all. + const composerStatus = connIsForOtherAgent + ? selectedAgentNotInstalled + ? "disconnected" + : "connecting" + : connStatus + const effectiveModes = connIsForOtherAgent + ? (getCachedSelectors(agentType)?.modes ?? null) + : conn.modes + const effectiveConfigOptions = connIsForOtherAgent + ? (getCachedSelectors(agentType)?.configOptions ?? null) + : conn.configOptions + const connectionCommands = connIsForOtherAgent ? null : conn.availableCommands + /** * Re-entry repair: let the database have the last word on a session this * card is coming back to. @@ -506,15 +546,17 @@ export function CanvasConversationSurface({ (nextModeId: string) => { setModeId(nextModeId) // The preference stores the whole mode SHAPE, not just the id, so it can - // be shipped to the backend at connect time. - if (conn.modes) { + // be shipped to the backend at connect time. Reconciled to the selected + // agent, so a change made mid-switch can't save the previous agent's + // mode list under this one. + if (effectiveModes) { saveModePreference(agentType, { - ...conn.modes, + ...effectiveModes, current_mode_id: nextModeId, }) } }, - [agentType, conn.modes] + [agentType, effectiveModes] ) const handleSend = useCallback( @@ -526,6 +568,9 @@ export function CanvasConversationSurface({ // never rolled back. const alreadyPersisted = dbConversationIdRef.current != null if (!alreadyPersisted && (creatingRef.current || !draftTarget)) return + // The composer already reads as not ready while the connection belongs + // to another agent; this catches a submit that races the switch. + if (connIsForOtherAgent) return const optimisticTurn = buildOptimisticUserTurn( draft, @@ -640,6 +685,7 @@ export function CanvasConversationSurface({ agentType, appendOptimisticTurn, boundFolderId, + connIsForOtherAgent, connSessionId, contextKey, draftTarget, @@ -717,20 +763,20 @@ export function CanvasConversationSurface({ ) const connectionModes = useMemo( - () => conn.modes?.available_modes ?? [], - [conn.modes] + () => effectiveModes?.available_modes ?? [], + [effectiveModes] ) const connectionConfigOptions = useMemo( - () => conn.configOptions ?? [], - [conn.configOptions] + () => effectiveConfigOptions ?? [], + [effectiveConfigOptions] ) const selectedModeId = useMemo(() => { if (connectionModes.length === 0) return null if (modeId && connectionModes.some((mode) => mode.id === modeId)) { return modeId } - return conn.modes?.current_mode_id ?? connectionModes[0]?.id ?? null - }, [conn.modes, connectionModes, modeId]) + return effectiveModes?.current_mode_id ?? connectionModes[0]?.id ?? null + }, [effectiveModes, connectionModes, modeId]) // Arrow-key history source, read lazily on the first Up/Down (see // `MessageInput.getSentHistory`) so streaming tokens cost nothing here. @@ -753,7 +799,7 @@ export function CanvasConversationSurface({ added here later. */}
Date: Thu, 15 Oct 2026 07:55:14 +0800 Subject: [PATCH 05/11] fix(acp): show a parked connect's failure in the composer banner A connect request parked behind another in-flight connect resolves at once and runs later, so its failure never reached the lifecycle hook's auto-connect error. useConnection now exposes the connect error the provider publishes for the key, and the banner falls back to it when it names the selected agent. --- .../chat/sub-agent-overlay.test.tsx | 1 + .../message/delegated-sub-thread.test.tsx | 1 + .../message/resumed-delegation-card.test.tsx | 1 + .../message/sub-agent-session-dialog.test.tsx | 1 + ...connection-lifecycle.connect-error.test.ts | 129 ++++++++++++++++++ src/hooks/use-connection-lifecycle.ts | 13 +- src/hooks/use-connection.test.tsx | 1 + src/hooks/use-connection.ts | 19 +++ 8 files changed, 165 insertions(+), 1 deletion(-) create mode 100644 src/hooks/use-connection-lifecycle.connect-error.test.ts diff --git a/src/components/chat/sub-agent-overlay.test.tsx b/src/components/chat/sub-agent-overlay.test.tsx index 8844024467..0a001aadb3 100644 --- a/src/components/chat/sub-agent-overlay.test.tsx +++ b/src/components/chat/sub-agent-overlay.test.tsx @@ -24,6 +24,7 @@ vi.mock("@/contexts/acp-connections-context", async () => { subscribeKey: () => () => {}, getConnection: () => undefined, getConnectPending: () => undefined, + getConnectError: () => undefined, getActiveKey: () => null, subscribeActiveKey: () => () => {}, }), diff --git a/src/components/message/delegated-sub-thread.test.tsx b/src/components/message/delegated-sub-thread.test.tsx index 58fcc9c953..6b6b8e6b60 100644 --- a/src/components/message/delegated-sub-thread.test.tsx +++ b/src/components/message/delegated-sub-thread.test.tsx @@ -26,6 +26,7 @@ vi.mock("@/contexts/acp-connections-context", async () => { subscribeKey: () => () => {}, getConnection: () => mockChildConnection, getConnectPending: () => undefined, + getConnectError: () => undefined, getActiveKey: () => null, subscribeActiveKey: () => () => {}, }), diff --git a/src/components/message/resumed-delegation-card.test.tsx b/src/components/message/resumed-delegation-card.test.tsx index 130069a742..0366025d00 100644 --- a/src/components/message/resumed-delegation-card.test.tsx +++ b/src/components/message/resumed-delegation-card.test.tsx @@ -25,6 +25,7 @@ vi.mock("@/contexts/acp-connections-context", async () => { subscribeKey: () => () => {}, getConnection: () => undefined, getConnectPending: () => undefined, + getConnectError: () => undefined, getActiveKey: () => null, subscribeActiveKey: () => () => {}, }), diff --git a/src/components/message/sub-agent-session-dialog.test.tsx b/src/components/message/sub-agent-session-dialog.test.tsx index 33663cec37..4d8f0d6357 100644 --- a/src/components/message/sub-agent-session-dialog.test.tsx +++ b/src/components/message/sub-agent-session-dialog.test.tsx @@ -86,6 +86,7 @@ vi.mock("@/contexts/acp-connections-context", async () => { }, getConnection: () => mockChildConnection, getConnectPending: () => undefined, + getConnectError: () => undefined, getActiveKey: () => null, subscribeActiveKey: () => () => {}, }), diff --git a/src/hooks/use-connection-lifecycle.connect-error.test.ts b/src/hooks/use-connection-lifecycle.connect-error.test.ts new file mode 100644 index 0000000000..8a56eb3a1f --- /dev/null +++ b/src/hooks/use-connection-lifecycle.connect-error.test.ts @@ -0,0 +1,129 @@ +/** + * The composer's inline banner reads `autoConnectError`, which the hook used to + * learn only from the promise its own auto-connect got back. A connect request + * that lands while another is still in flight for the same key is PARKED: its + * promise resolves at once, and the provider runs it later from the in-flight + * call's cleanup, swallowing the rejection there. So when a draft switches + * agents mid-connect and the agent picked last fails, the hook saw a success. + * The provider still publishes that failure for the key (it is what the toast + * and the status chip show); the banner has to read it too. + */ + +import { renderHook } from "@testing-library/react" +import { beforeEach, describe, expect, it, vi } from "vitest" +import type { ConnectErrorInfo } from "@/contexts/acp-connections-context" + +vi.mock("next-intl", () => ({ + useTranslations: () => (key: string) => key, +})) + +vi.mock("@/contexts/acp-connections-context", () => { + const actions = { setActiveKey: vi.fn(), touchActivity: vi.fn() } + return { useAcpActions: () => actions } +}) + +vi.mock("@/contexts/task-context", () => { + const context = { addTask: vi.fn(), updateTask: vi.fn(), removeTask: vi.fn() } + return { useTaskContext: () => context } +}) + +const conn = vi.hoisted(() => ({ + status: null as string | null, + connectError: null as ConnectErrorInfo | null, +})) + +vi.mock("@/hooks/use-connection", () => { + // A parked request: the call resolves without saying how the request did. + const connect = vi.fn().mockResolvedValue(undefined) + const noop = vi.fn().mockResolvedValue(undefined) + return { + useConnection: () => ({ + status: conn.status, + agentType: null, + connectError: conn.connectError, + selectorsReady: false, + hasCachedSelectors: false, + connect, + disconnect: noop, + sendPrompt: noop, + setMode: noop, + setConfigOption: noop, + cancel: noop, + respondPermission: noop, + modes: null, + configOptions: null, + isViewer: false, + backgroundOutstanding: 0, + }), + } +}) + +import { useConnectionLifecycle } from "@/hooks/use-connection-lifecycle" + +function renderDraft() { + return renderHook(() => + useConnectionLifecycle({ + contextKey: "draft-tab", + agentType: "codex", + isActive: true, + workingDir: "/tmp/proj", + }) + ) +} + +describe("useConnectionLifecycle autoConnectError", () => { + beforeEach(() => { + conn.status = null + conn.connectError = null + }) + + it("reports the published failure of a request that was parked", async () => { + conn.connectError = { + agentType: "codex", + title: "connectFailedTitle", + detail: "spawn failed: exit code 1", + opensAgentSettings: false, + } + const { result } = renderDraft() + // Let the auto-connect promise resolve, as a parked request's does. + await vi.waitFor(() => { + expect(result.current.autoConnectError).toBe("spawn failed: exit code 1") + }) + }) + + it("falls back to the title when the failure has nothing more to say", () => { + conn.connectError = { + agentType: "codex", + title: "Codex's ACP adapter is not installed", + detail: null, + opensAgentSettings: true, + } + const { result } = renderDraft() + expect(result.current.autoConnectError).toBe( + "Codex's ACP adapter is not installed" + ) + }) + + it("ignores a failure published for another agent", () => { + conn.connectError = { + agentType: "claude_code", + title: "connectFailedTitle", + detail: "old agent failed", + opensAgentSettings: false, + } + const { result } = renderDraft() + expect(result.current.autoConnectError).toBeNull() + }) + + it("stays quiet once the session is up", () => { + conn.status = "connected" + conn.connectError = { + agentType: "codex", + title: "connectFailedTitle", + detail: "stale", + opensAgentSettings: false, + } + const { result } = renderDraft() + expect(result.current.autoConnectError).toBeNull() + }) +}) diff --git a/src/hooks/use-connection-lifecycle.ts b/src/hooks/use-connection-lifecycle.ts index 296441fe50..6a5c9874d0 100644 --- a/src/hooks/use-connection-lifecycle.ts +++ b/src/hooks/use-connection-lifecycle.ts @@ -451,13 +451,24 @@ export function useConnectionLifecycle({ touchActivity, ]) + // Why connecting failed, for the composer's inline banner. The auto-connect's + // own rejection says it first; a request parked behind another in-flight + // connect never rejects to it (its promise resolves at once, and the run + // that fails comes later), so fall back to what the provider published for + // this key — the record the toast and the status chip show — as long as it + // names the agent this surface wants. `detail ?? title` is the text that + // request would have rejected with. + const publishedConnectError = + conn.connectError?.agentType === agentType ? conn.connectError : null const autoConnectError = status === "connected" || status === "prompting" ? null : lastAutoConnectError?.contextKey === contextKey && lastAutoConnectError.agentType === agentType ? lastAutoConnectError.message - : null + : publishedConnectError + ? (publishedConnectError.detail ?? publishedConnectError.title) + : null // sendPrompt, connCancel, connRespondPermission are stable (depend // only on actions + contextKey), so these callbacks are effectively stable. diff --git a/src/hooks/use-connection.test.tsx b/src/hooks/use-connection.test.tsx index 4f4b06cd24..4949d84580 100644 --- a/src/hooks/use-connection.test.tsx +++ b/src/hooks/use-connection.test.tsx @@ -27,6 +27,7 @@ const fake = vi.hoisted(() => { store: { getConnection: () => conn, getConnectPending: () => pending, + getConnectError: () => undefined, getActiveKey: () => null, subscribeKey: (_key: string, cb: () => void) => { listeners.add(cb) diff --git a/src/hooks/use-connection.ts b/src/hooks/use-connection.ts index 7957aa8d7e..acbb3a030d 100644 --- a/src/hooks/use-connection.ts +++ b/src/hooks/use-connection.ts @@ -6,6 +6,7 @@ import { useConnectionStore, getCachedSelectors, type ClaudeApiRetryState, + type ConnectErrorInfo, type ConnectionState, type PendingPermission, type PendingUserMessage, @@ -56,6 +57,12 @@ export interface UseConnectionReturn { */ isViewer: boolean status: ConnectionStatus | null + /** Why the last `connect()` for this contextKey failed, as the provider + * published it (the toast and the status chip read the same record), or + * null. Covers requests whose own promise never reported it: a request + * parked behind another in-flight connect resolves at once and is run, and + * can fail, later. */ + connectError: ConnectErrorInfo | null promptCapabilities: PromptCapabilitiesInfo supportsFork: boolean selectorsReady: boolean @@ -240,6 +247,16 @@ export function useConnection(contextKey: string): UseConnectionReturn { // placeholder, no loading cue and no status-bar task. A real entry always // wins: once it exists, its own status is the more specific truth. const status = connection?.status ?? (connectPending ? "connecting" : null) + const getConnectErrorSnapshot = useCallback( + () => store.getConnectError(contextKey), + [store, contextKey] + ) + const connectError = + useSyncExternalStore( + subscribe, + getConnectErrorSnapshot, + getConnectErrorSnapshot + ) ?? null const promptCapabilities = connection?.promptCapabilities ?? DEFAULT_PROMPT_CAPABILITIES const supportsFork = connection?.supportsFork ?? false @@ -353,6 +370,7 @@ export function useConnection(contextKey: string): UseConnectionReturn { agentType, isViewer, status, + connectError, promptCapabilities, supportsFork, selectorsReady, @@ -395,6 +413,7 @@ export function useConnection(contextKey: string): UseConnectionReturn { agentType, isViewer, status, + connectError, promptCapabilities, supportsFork, selectorsReady, From 7ca5ee65a402b31db06459b565d547182d7de672 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 07:55:14 +0800 Subject: [PATCH 06/11] test(chat): pin the draft agent pickers' lock rule --- .../draft-agent-picker-lock.test.ts | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) create mode 100644 src/components/conversations/draft-agent-picker-lock.test.ts diff --git a/src/components/conversations/draft-agent-picker-lock.test.ts b/src/components/conversations/draft-agent-picker-lock.test.ts new file mode 100644 index 0000000000..9ea465a088 --- /dev/null +++ b/src/components/conversations/draft-agent-picker-lock.test.ts @@ -0,0 +1,42 @@ +/** + * The tab view's agent pickers lock once the tab is bound to a conversation + * row — a saved conversation keeps the agent it was created with — and never + * because the agent is still connecting: a draft can be switched while its + * first agent spawns, and `connect()` replaces the abandoned attempt with the + * agent picked last. + * + * `ConversationTabView` is too heavy to render here (see + * tab-first-show-auto-connect.test.tsx), so this pins the wiring in its source. + * The canvas card's picker is covered by a render test of its own + * (canvas-conversation-surface.test.tsx). + */ +import { readFileSync } from "node:fs" +import { resolve } from "node:path" +import { describe, expect, it } from "vitest" + +describe("ConversationTabView agent pickers", () => { + const panel = readFileSync( + resolve( + process.cwd(), + "src/components/conversations/conversation-detail-panel.tsx" + ), + "utf8" + ) + const pickers = panel.match(//g) ?? [] + + it("locks both pickers on the conversation row alone", () => { + // The welcome-mode picker and the docked composer's. + expect(pickers).toHaveLength(2) + for (const picker of pickers) { + expect(picker.match(/\bdisabled=\{([^}]*)\}/)?.[1]).toBe( + "dbConversationId != null" + ) + } + }) + + it("ignores a pick once the tab is bound, whatever the picker shows", () => { + expect(panel).toMatch( + /const handleAgentSelect = useCallback\(\s+\(nextAgentType: AgentType\) => \{\s+if \(nextAgentType === selectedAgentRef\.current\) return\s+if \(dbConvIdRef\.current\) return/ + ) + }) +}) From 809efc7a10447b0ffdac54a19e1c91e01343eea4 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 08:04:02 +0800 Subject: [PATCH 07/11] fix(acp): keep config edits off a connection bound to another agent While a draft's connection still belongs to the agent it switched away from, the chat tab and the canvas card show the selected agent's cached config options. An edit went to the other agent's connection with a value from a list that is not its own, and was saved as that agent's preference. Such an edit is now dropped, as it is when no connection exists for the surface. --- .../canvas-conversation-surface.test.tsx | 27 +++++- ...-connection-lifecycle.agent-switch.test.ts | 92 +++++++++++++++++++ src/hooks/use-connection-lifecycle.ts | 9 +- 3 files changed, 124 insertions(+), 4 deletions(-) create mode 100644 src/hooks/use-connection-lifecycle.agent-switch.test.ts diff --git a/src/components/canvas/canvas-conversation-surface.test.tsx b/src/components/canvas/canvas-conversation-surface.test.tsx index 3a07fd3894..7c8ff6c76d 100644 --- a/src/components/canvas/canvas-conversation-surface.test.tsx +++ b/src/components/canvas/canvas-conversation-surface.test.tsx @@ -70,6 +70,7 @@ vi.mock("@/contexts/acp-connections-context", () => { const lifecycle = vi.hoisted(() => ({ conn: {} as Record, handleSend: vi.fn(), + handleSetConfigOption: vi.fn(), })) vi.mock("@/hooks/use-connection-lifecycle", () => ({ useConnectionLifecycle: () => ({ @@ -80,7 +81,7 @@ vi.mock("@/hooks/use-connection-lifecycle", () => ({ autoConnectError: null, handleFocus: vi.fn(), handleSend: lifecycle.handleSend, - handleSetConfigOption: vi.fn(), + handleSetConfigOption: lifecycle.handleSetConfigOption, handleCancel: vi.fn(), handleRespondPermission: vi.fn(), }), @@ -142,6 +143,18 @@ const CODEX_MODES = { current_mode_id: "auto", available_modes: [{ id: "auto", name: "Auto" }], } +const CODEX_CONFIG = [ + { + id: "model", + name: "Model", + type: "select", + current_value: "gpt-5.5", + options: [ + { value: "gpt-5.5", name: "GPT-5.5" }, + { value: "gpt-5.4", name: "GPT-5.4" }, + ], + }, +] function agent( agentType: string, @@ -216,13 +229,21 @@ describe("a draft card whose connection still belongs to the previous agent", () it("presents the switch as still connecting, with the picked agent's chips", () => { // Codex is installed, so its connect is on its way and will replace this. - cachedSelectors.set("codex", { modes: CODEX_MODES, configOptions: null }) + cachedSelectors.set("codex", { + modes: CODEX_MODES, + configOptions: CODEX_CONFIG, + }) lifecycle.conn = connection("claude_code", "connected") renderDraftCard() expect(shell.props?.status).toBe("connecting") expect(shell.props?.modes).toEqual(CODEX_MODES.available_modes) - expect(shell.props?.configOptions).toEqual([]) + // Codex's own list, not Claude's. An edit to it is kept off Claude's + // connection by the lifecycle hook (use-connection-lifecycle.agent-switch). + expect(shell.props?.configOptions).toEqual(CODEX_CONFIG) + expect(shell.props?.onConfigOptionChange).toBe( + lifecycle.handleSetConfigOption + ) expect(shell.props?.selectedModeId).toBe("auto") expect(shell.props?.availableCommands).toEqual([]) }) diff --git a/src/hooks/use-connection-lifecycle.agent-switch.test.ts b/src/hooks/use-connection-lifecycle.agent-switch.test.ts new file mode 100644 index 0000000000..501ce1fdbf --- /dev/null +++ b/src/hooks/use-connection-lifecycle.agent-switch.test.ts @@ -0,0 +1,92 @@ +/** + * While a draft switches agents, the connection under its key can still belong + * to the agent the user just left — for a moment, or for good when the new pick + * can't connect (its preflight refuses it before anything is torn down). The + * surfaces then show the SELECTED agent's cached config options. An edit to one + * must not reach the other agent's connection: it would hand that agent a value + * from a list that isn't its own and file it under that agent's saved + * preferences, which its next connect ships back. + */ + +import { renderHook } from "@testing-library/react" +import { beforeEach, describe, expect, it, vi } from "vitest" + +vi.mock("next-intl", () => ({ + useTranslations: () => (key: string) => key, +})) + +vi.mock("@/contexts/acp-connections-context", () => { + const actions = { setActiveKey: vi.fn(), touchActivity: vi.fn() } + return { useAcpActions: () => actions } +}) + +vi.mock("@/contexts/task-context", () => { + const context = { addTask: vi.fn(), updateTask: vi.fn(), removeTask: vi.fn() } + return { useTaskContext: () => context } +}) + +const conn = vi.hoisted(() => ({ + agentType: null as string | null, + setConfigOption: vi.fn(() => Promise.resolve()), +})) + +vi.mock("@/hooks/use-connection", () => { + const noop = vi.fn().mockResolvedValue(undefined) + return { + useConnection: () => ({ + status: "connected", + agentType: conn.agentType, + connectError: null, + selectorsReady: true, + hasCachedSelectors: true, + connect: noop, + disconnect: noop, + sendPrompt: noop, + setMode: noop, + setConfigOption: conn.setConfigOption, + cancel: noop, + respondPermission: noop, + modes: null, + configOptions: null, + isViewer: false, + backgroundOutstanding: 0, + }), + } +}) + +import { useConnectionLifecycle } from "@/hooks/use-connection-lifecycle" + +function renderDraft() { + return renderHook(() => + useConnectionLifecycle({ + contextKey: "draft-tab", + agentType: "codex", + isActive: false, + }) + ) +} + +describe("useConnectionLifecycle config edits across an agent switch", () => { + beforeEach(() => { + conn.agentType = null + conn.setConfigOption.mockClear() + }) + + it("keeps an edit off a connection that belongs to the previous agent", () => { + conn.agentType = "claude_code" + const { result } = renderDraft() + + result.current.handleSetConfigOption("model", "gpt-5.4") + + expect(conn.setConfigOption).not.toHaveBeenCalled() + }) + + it("applies the edit once the connection is the selected agent's", () => { + conn.agentType = "codex" + const { result } = renderDraft() + + result.current.handleSetConfigOption("model", "gpt-5.4") + + expect(conn.setConfigOption).toHaveBeenCalledWith("model", "gpt-5.4") + }) +}) diff --git a/src/hooks/use-connection-lifecycle.ts b/src/hooks/use-connection-lifecycle.ts index 6a5c9874d0..c19588c57c 100644 --- a/src/hooks/use-connection-lifecycle.ts +++ b/src/hooks/use-connection-lifecycle.ts @@ -538,12 +538,19 @@ export function useConnectionLifecycle({ const handleSetConfigOption = useCallback( (configId: string, valueId: string) => { + // The connection here still belongs to another agent (a draft mid-switch, + // or one whose new pick could not connect), so the options on screen are + // the selected agent's cached ones. Applying the edit would hand the + // other agent a value from a list that isn't its own, and file it under + // that agent's saved preferences. Dropped, as an edit is when there is + // no connection here at all. + if (connAgentType != null && connAgentType !== agentType) return touchActivity(contextKey) connSetConfigOption(configId, valueId).catch((e: unknown) => console.error("[ConnLifecycle] setConfigOption:", e) ) }, - [connSetConfigOption, contextKey, touchActivity] + [agentType, connAgentType, connSetConfigOption, contextKey, touchActivity] ) const handleRespondPermission = useCallback( From 6cd52fd0bf382e481aca461b8ea87e37b48d56e3 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 08:13:42 +0800 Subject: [PATCH 08/11] test(chat): pin the canvas card's transcript commands like the tab view's The canvas card now passes the transcript `connectionCommands`, null while its connection still belongs to another agent, the same shape the tab view is already pinned to. --- .../use-transcript-known-invocations.test.tsx | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/src/components/message/use-transcript-known-invocations.test.tsx b/src/components/message/use-transcript-known-invocations.test.tsx index 2cd276b7e5..1f51f59595 100644 --- a/src/components/message/use-transcript-known-invocations.test.tsx +++ b/src/components/message/use-transcript-known-invocations.test.tsx @@ -286,19 +286,21 @@ describe("MessageListView", () => { const tag = source.slice(start, source.indexOf("/>", start)) return /\bavailableCommands=\{([^}]*)\}/.exec(tag)?.[1].trim() } - expect( - commandsPassedBy("src/components/canvas/canvas-conversation-surface.tsx") - ).toBe("conn.availableCommands") // Except with no agent known at all: then nothing it remembered applies, // and `[]` is the deliberate answer. expect( commandsPassedBy("src/components/message/live-transcript-view.tsx") ).toBe("transcriptAgent ? conn?.availableCommands : []") - const detailPanel = - "src/components/conversations/conversation-detail-panel.tsx" - expect(commandsPassedBy(detailPanel)).toBe("connectionCommands") - expect(read(detailPanel)).toMatch( - /const connectionCommands = connIsForOtherAgent \? null : conn\.availableCommands\n/ - ) + // A connection still bound to the agent a draft switched away from has + // advertised nothing for the selected one: unknown again, never `[]`. + for (const surface of [ + "src/components/conversations/conversation-detail-panel.tsx", + "src/components/canvas/canvas-conversation-surface.tsx", + ]) { + expect(commandsPassedBy(surface)).toBe("connectionCommands") + expect(read(surface)).toMatch( + /const connectionCommands = connIsForOtherAgent \? null : conn\.availableCommands\n/ + ) + } }) }) From 6bf1aedfbba15c51189bb88003ef66d729f8024c Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 09:14:13 +0800 Subject: [PATCH 09/11] fix(acp): stop a connection's handshake when it is disconnected Only the conversation loop reads a connection's command channel, and the loop starts once the session is established. A connection disconnected during its handshake therefore finished Initialize (up to 60 s) and the session setup first, keeping its agent process running for nobody - as with the agent a draft switches away from while it is still connecting. Every manager teardown path now cancels a per-connection token before it sends Disconnect, and the handshake's waits on the agent (Initialize, session/resume, session/load, session/new, the preference replay) give way to it. The connection then ends quietly and its agent process is stopped. The command channel is left alone: prompts sent before the session is up still wait there for the loop. --- src-tauri/src/acp/connection.rs | 489 +++++++++++++++++++++++++++----- src-tauri/src/acp/lifecycle.rs | 1 + src-tauri/src/acp/manager.rs | 138 ++++++++- 3 files changed, 558 insertions(+), 70 deletions(-) diff --git a/src-tauri/src/acp/connection.rs b/src-tauri/src/acp/connection.rs index 54897c758f..1d0dd1586d 100644 --- a/src-tauri/src/acp/connection.rs +++ b/src-tauri/src/acp/connection.rs @@ -30,13 +30,14 @@ use agent_client_protocol::schema::v1::{ use agent_client_protocol::schema::ProtocolVersion; use agent_client_protocol::util::MatchDispatch; use agent_client_protocol::{ - on_receive_notification, on_receive_request, Agent, Client, ConnectionTo, Dispatch, + on_receive_notification, on_receive_request, Agent, Client, ConnectTo, ConnectionTo, Dispatch, HandleDispatchFrom, Handled, JsonRpcRequest, RequestCancellation, Responder, Role, UntypedMessage, }; use crate::acp::agent_process::AcpAgent; use crate::acp::agent_session::AgentSession; use tokio::sync::{mpsc, oneshot, RwLock}; +use tokio_util::sync::CancellationToken; use crate::acp::agent_mentions::append_agent_routes; use crate::acp::background_watch; @@ -1252,6 +1253,25 @@ pub enum ConnectionCommand { /// by the outer `.map_err(...)` in `run_connection`. const INIT_TIMEOUT_SENTINEL: &str = "__codeg_init_timeout__"; +/// Sentinel a step of the handshake fails with once the manager has cancelled +/// it ([`AgentConnection::handshake`]), so the outer match in `run_connection` +/// can end the connection quietly: the connection was let go, it did not fail. +/// Same trick as [`INIT_TIMEOUT_SENTINEL`]. +const HANDSHAKE_CANCELLED_SENTINEL: &str = "__codeg_handshake_cancelled__"; + +/// Wait on the agent for `step` of the session's establishment, unless the +/// handshake is cancelled first. Then the step is dropped and this fails with +/// [`HANDSHAKE_CANCELLED_SENTINEL`], which `?` carries out of the handshake. +async fn unless_cancelled( + handshake: &CancellationToken, + step: impl std::future::Future, +) -> Result { + handshake + .run_until_cancelled(step) + .await + .ok_or_else(|| agent_client_protocol::util::internal_error(HANDSHAKE_CANCELLED_SENTINEL)) +} + /// Sentinel appended to a `session/new` failure when codeg had just forwarded /// MCP servers to a *custom* agent, so the outer `.map_err(...)` can raise /// `AcpError::McpRejectedByAgent` and point the user at the `supports_mcp` @@ -1381,6 +1401,16 @@ pub struct AgentConnection { pub status: ConnectionStatus, pub owner_window_label: String, pub cmd_tx: mpsc::Sender, + /// Cancelled by every manager path that ends this connection, before it + /// sends `Disconnect` on `cmd_tx`. Only the conversation loop reads that + /// channel, and the loop starts once the session is established — so a + /// connection let go during its handshake (Initialize, which may wait 60 s, + /// then opening or restoring the session) would otherwise finish the + /// handshake first, its agent process running for nobody all the while, as + /// with the agent a draft switches away from mid-connect. The handshake + /// cannot read `cmd_tx` itself: prompts sent before the session is up wait + /// there for the loop. + pub handshake: CancellationToken, /// 后端权威的会话状态。所有 `emit_with_state` 写入此状态并自增 seq。 /// 使用 `Arc>` 让 spawn 出的连接 task 与外部 snapshot 读取共享。 pub state: Arc>, @@ -2476,6 +2506,7 @@ pub async fn spawn_agent_connection( prepend_officecli_path(&mut terminal_base_env); let (cmd_tx, cmd_rx) = mpsc::channel::(32); + let handshake = CancellationToken::new(); let conn_id = connection_id.clone(); let emitter_clone = emitter.clone(); let cleanup_connections = connections.clone(); @@ -2500,6 +2531,7 @@ pub async fn spawn_agent_connection( status: ConnectionStatus::Connecting, owner_window_label, cmd_tx, + handshake: handshake.clone(), state: Arc::clone(&session_state), emitter: emitter.clone(), prompt_lock: Arc::new(tokio::sync::Mutex::new(())), @@ -2541,6 +2573,7 @@ pub async fn spawn_agent_connection( working_dir, session_id, cmd_rx, + handshake, emitter_clone.clone(), Arc::clone(&state_clone), terminal_base_env, @@ -5868,12 +5901,16 @@ fn canonical_spec_to_mcp_server(name: &str, spec: &serde_json::Value) -> Result< ) )] async fn run_connection( - agent: AcpAgent, + // The launched agent process; any transport to an agent in tests. + agent: impl ConnectTo, connection_id: String, agent_type: AgentType, working_dir: Option, session_id: Option, mut cmd_rx: mpsc::Receiver, + // `AgentConnection::handshake`: once cancelled, every wait on the agent + // that establishes the session gives up. + handshake: CancellationToken, emitter: EventEmitter, state: Arc>, terminal_base_env: BTreeMap, @@ -5990,7 +6027,7 @@ async fn run_connection( Arc::clone(&prompt_ledger), ); - Client + let ended = Client .builder() .name("codeg") // First in the chain on purpose: it has to claim a null-`sessionId` @@ -6353,11 +6390,14 @@ async fn run_connection( ProtocolVersion::V1 ); let init_started = std::time::Instant::now(); - let init_resp = match tokio::time::timeout( - std::time::Duration::from_secs(60), - cx.send_request_to(Agent, init_request).block_task(), + let init_resp = match unless_cancelled( + &handshake, + tokio::time::timeout( + std::time::Duration::from_secs(60), + cx.send_request_to(Agent, init_request).block_task(), + ), ) - .await + .await? { Ok(Ok(resp)) => { tracing::info!( @@ -6630,7 +6670,7 @@ async fn run_connection( &cwd, mcp_servers.clone(), ); - match send_resume_session(&cx, resume_req).await { + match unless_cancelled(&handshake, send_resume_session(&cx, resume_req)).await? { Ok((resume_resp, grok_models_raw)) => { let initial_config_options = resume_resp.config_options.clone(); let new_resp = NewSessionResponse::new(SessionId::new(sid.clone())) @@ -6663,19 +6703,22 @@ async fn run_connection( ) .await; emit_session_modes(&state, &emitter_clone, session.modes()).await; - apply_and_emit_session_config_options( - &cx, - &mut session, - &state, - &emitter_clone, - agent_type, - grok_meta.as_ref(), - grok_model_specs.as_ref(), - preferred_mode_id.as_deref(), - &preferred_config_values, - initial_config_options.unwrap_or_default(), + unless_cancelled( + &handshake, + apply_and_emit_session_config_options( + &cx, + &mut session, + &state, + &emitter_clone, + agent_type, + grok_meta.as_ref(), + grok_model_specs.as_ref(), + preferred_mode_id.as_deref(), + &preferred_config_values, + initial_config_options.unwrap_or_default(), + ), ) - .await; + .await?; emit_selectors_ready(&state, &emitter_clone).await; let loop_result = run_conversation_loop( @@ -6764,7 +6807,8 @@ async fn run_connection( &cwd, mcp_servers.clone(), ); - cx.send_request_to(Agent, load_req).block_task().await + unless_cancelled(&handshake, cx.send_request_to(Agent, load_req).block_task()) + .await? } else { Err(agent_client_protocol::Error::method_not_found() .data("agent does not advertise the loadSession capability")) @@ -6813,6 +6857,13 @@ async fn run_connection( // usable until the replay is consumed) but records // nothing more, so the transcript ends at a line // boundary instead of growing holes. + // + // Unlike the requests around it, the drain does not give + // way to a cancelled handshake: the agent sends the + // replay before it answers the load, so there is little + // left to wait for, and stopping partway would leave a + // hydrated transcript cut short for good — once it has + // history, the next load never hydrates it again. let mut recording = hydrate_from_replay; while let Ok(Ok(dispatch)) = tokio::time::timeout( std::time::Duration::from_millis(100), @@ -6930,21 +6981,24 @@ async fn run_connection( ) .await; emit_session_modes(&state, &emitter_clone, session.modes()).await; - apply_and_emit_session_config_options( - &cx, - &mut session, - &state, - &emitter_clone, - agent_type, - grok_meta.as_ref(), - // `session/load` is a typed send with no raw `models` - // capture, so effort stays on the flat fallback. - None, - preferred_mode_id.as_deref(), - &preferred_config_values, - initial_config_options.unwrap_or_default(), + unless_cancelled( + &handshake, + apply_and_emit_session_config_options( + &cx, + &mut session, + &state, + &emitter_clone, + agent_type, + grok_meta.as_ref(), + // `session/load` is a typed send with no raw `models` + // capture, so effort stays on the flat fallback. + None, + preferred_mode_id.as_deref(), + &preferred_config_values, + initial_config_options.unwrap_or_default(), + ), ) - .await; + .await?; emit_selectors_ready(&state, &emitter_clone).await; let loop_result = run_conversation_loop( @@ -7085,11 +7139,14 @@ async fn run_connection( ) .await; } - let (new_resp, grok_models_raw) = send_new_session_capturing_models( - &cx, - build_new_session_request(agent_type, &cwd, mcp_servers.clone()), + let (new_resp, grok_models_raw) = unless_cancelled( + &handshake, + send_new_session_capturing_models( + &cx, + build_new_session_request(agent_type, &cwd, mcp_servers.clone()), + ), ) - .await + .await? .map_err(|e| tag_new_session_failure(e, agent_type, &mcp_servers))?; let fallback_sid = new_resp.session_id.0.to_string(); let initial_config_options = new_resp.config_options.clone(); @@ -7127,19 +7184,22 @@ async fn run_connection( ) .await; emit_session_modes(&state, &emitter_clone, session.modes()).await; - apply_and_emit_session_config_options( - &cx, - &mut session, - &state, - &emitter_clone, - agent_type, - grok_meta.as_ref(), - grok_model_specs.as_ref(), - preferred_mode_id.as_deref(), - &preferred_config_values, - initial_config_options.unwrap_or_default(), + unless_cancelled( + &handshake, + apply_and_emit_session_config_options( + &cx, + &mut session, + &state, + &emitter_clone, + agent_type, + grok_meta.as_ref(), + grok_model_specs.as_ref(), + preferred_mode_id.as_deref(), + &preferred_config_values, + initial_config_options.unwrap_or_default(), + ), ) - .await; + .await?; emit_selectors_ready(&state, &emitter_clone).await; let loop_result = run_conversation_loop( @@ -7185,11 +7245,14 @@ async fn run_connection( } } else { // Create new session - let (new_resp, grok_models_raw) = send_new_session_capturing_models( - &cx, - build_new_session_request(agent_type, &cwd, mcp_servers.clone()), + let (new_resp, grok_models_raw) = unless_cancelled( + &handshake, + send_new_session_capturing_models( + &cx, + build_new_session_request(agent_type, &cwd, mcp_servers.clone()), + ), ) - .await + .await? .map_err(|e| tag_new_session_failure(e, agent_type, &mcp_servers))?; let sid = new_resp.session_id.0.to_string(); let initial_config_options = new_resp.config_options.clone(); @@ -7214,19 +7277,22 @@ async fn run_connection( ) .await; emit_session_modes(&state, &emitter_clone, session.modes()).await; - apply_and_emit_session_config_options( - &cx, - &mut session, - &state, - &emitter_clone, - agent_type, - grok_meta.as_ref(), - grok_model_specs.as_ref(), - preferred_mode_id.as_deref(), - &preferred_config_values, - initial_config_options.unwrap_or_default(), + unless_cancelled( + &handshake, + apply_and_emit_session_config_options( + &cx, + &mut session, + &state, + &emitter_clone, + agent_type, + grok_meta.as_ref(), + grok_model_specs.as_ref(), + preferred_mode_id.as_deref(), + &preferred_config_values, + initial_config_options.unwrap_or_default(), + ), ) - .await; + .await?; emit_selectors_ready(&state, &emitter_clone).await; let loop_result = run_conversation_loop( @@ -7268,8 +7334,17 @@ async fn run_connection( .await } }) - .await - .map_err(connection_failure) + .await; + match ended { + // Let go before its session was up. Leaving the closure above has + // already shut the connection down, and dropping its transport stops + // the agent process. + Err(err) if err.to_string().contains(HANDSHAKE_CANCELLED_SENTINEL) => { + tracing::info!("[ACP] disconnected during the handshake; stopped it there"); + Ok(()) + } + ended => ended.map_err(connection_failure), + } } /// The codeg error a failed connection ends with: the sentinel an inner step @@ -25530,6 +25605,282 @@ mod tests { agent.await.unwrap(); } + /// An in-process agent for `run_connection`'s handshake. It reports each + /// request it receives (a prompt with its text), answers Initialize unless + /// told to stall there, holds session/new until `open_session` is notified, + /// and ends every prompt at once. `closed` fires when codeg's end of the + /// connection goes away. + struct HandshakeAgent { + seen: mpsc::UnboundedReceiver, + open_session: Arc, + closed: oneshot::Receiver<()>, + task: tokio::task::JoinHandle<()>, + } + + impl HandshakeAgent { + fn start(agent_end: agent_client_protocol::Channel, stall_initialize: bool) -> Self { + use agent_client_protocol::schema::v1::{InitializeResponse, PromptResponse}; + + let (seen_tx, seen) = mpsc::unbounded_channel(); + let (init_seen, new_seen, prompt_seen) = (seen_tx.clone(), seen_tx.clone(), seen_tx); + let open_session = Arc::new(tokio::sync::Notify::new()); + let release = Arc::clone(&open_session); + let (closed_tx, closed) = oneshot::channel(); + let task = tokio::spawn(async move { + let _ = Agent + .builder() + .on_receive_request( + async move |_req: InitializeRequest, + responder: Responder, + _cx: ConnectionTo| { + let _ = init_seen.send("initialize".to_string()); + if stall_initialize { + // Kept, never answered: dropping it would answer. + tokio::spawn(async move { + std::future::pending::<()>().await; + drop(responder); + }); + return Ok(()); + } + responder.respond(InitializeResponse::new(ProtocolVersion::V1)) + }, + on_receive_request!(), + ) + .on_receive_request( + async move |_req: NewSessionRequest, + responder: Responder, + _cx: ConnectionTo| { + let _ = new_seen.send("session/new".to_string()); + let release = Arc::clone(&release); + // Off the handler, so the held answer doesn't hold + // up the agent's other incoming messages. + tokio::spawn(async move { + release.notified().await; + responder.respond(NewSessionResponse::new(SessionId::new("s1"))) + }); + Ok(()) + }, + on_receive_request!(), + ) + .on_receive_request( + async move |req: PromptRequest, + responder: Responder, + _cx: ConnectionTo| { + let text: String = req + .prompt + .iter() + .filter_map(|block| match block { + ContentBlock::Text(text) => Some(text.text.as_str()), + _ => None, + }) + .collect(); + let _ = prompt_seen.send(format!("session/prompt: {text}")); + responder.respond(PromptResponse::new(StopReason::EndTurn)) + }, + on_receive_request!(), + ) + .connect_with(agent_end, async move |cx: ConnectionTo| { + cx.incoming_closed().await; + let _ = closed_tx.send(()); + Ok(()) + }) + .await; + }); + Self { + seen, + open_session, + closed, + task, + } + } + + async fn next_request(&mut self) -> String { + tokio::time::timeout(std::time::Duration::from_secs(10), self.seen.recv()) + .await + .expect("codeg sends the agent its next request") + .expect("the agent is running") + } + + /// Wait for codeg's end of the connection to go away. + async fn closed(self) { + tokio::time::timeout(std::time::Duration::from_secs(5), self.closed) + .await + .expect("codeg closes its end of the connection") + .expect("the agent saw it close"); + self.task.abort(); + } + } + + /// `run_connection` on the other end of a [`HandshakeAgent`], driven the way + /// `spawn_agent_connection` drives it: on a thread of its own with the + /// connection's stack, since a debug build's frame for it does not fit a + /// test thread's. + struct HandshakeRun { + cmd_tx: mpsc::Sender, + handshake: CancellationToken, + state: Arc>, + events: tokio::sync::broadcast::Receiver>, + ended: oneshot::Receiver>, + _dir: tempfile::TempDir, + } + + impl HandshakeRun { + async fn start(client_end: agent_client_protocol::Channel) -> Self { + let dir = tempfile::tempdir().expect("tempdir"); + let cwd = dir.path().to_path_buf(); + // An agent codeg forwards no MCP servers to, so the handshake reads + // no MCP config from the machine running the test. + let agent_type = AgentType::OpenClaw; + let state = Arc::new(RwLock::new(SessionState::new( + "conn-handshake".to_string(), + agent_type, + Some(cwd.clone()), + "test-window".to_string(), + None, + ))); + let events = state.read().await.event_stream().subscribe(); + let (cmd_tx, cmd_rx) = mpsc::channel(32); + let handshake = CancellationToken::new(); + let (ended_tx, ended) = oneshot::channel(); + let runtime = tokio::runtime::Handle::current(); + let (run_state, run_handshake) = (Arc::clone(&state), handshake.clone()); + std::thread::Builder::new() + .name("acp-conn-handshake-test".to_string()) + .stack_size(ACP_CONNECTION_STACK_SIZE) + .spawn(move || { + let ended = runtime.block_on(run_connection( + client_end, + "conn-handshake".to_string(), + agent_type, + Some(cwd.to_string_lossy().into_owned()), + None, + cmd_rx, + run_handshake, + EventEmitter::Noop, + run_state, + BTreeMap::new(), + TerminalShellRuntimeConfig::new(), + None, + BTreeMap::new(), + None, + FsAccessPolicy::from_env(&cwd, agent_type, &BTreeMap::new()), + HostToolsPolicy::from_env(&BTreeMap::new()), + Arc::new(StderrTail::new()), + )); + let _ = ended_tx.send(ended); + }) + .expect("connection thread"); + Self { + cmd_tx, + handshake, + state, + events, + ended, + _dir: dir, + } + } + + /// Admit a prompt the way `send_prompt_inner` does, then hand it over. + async fn prompt(&self, text: &str) { + self.state.write().await.turn_in_flight = true; + self.cmd_tx + .send(ConnectionCommand::Prompt { + blocks: vec![PromptInputBlock::Text { text: text.into() }], + user_message: None, + }) + .await + .expect("the connection is running"); + } + + /// How the connection ended — which has to be soon — and what it + /// emitted on the way. + async fn ended(mut self) -> (Result<(), AcpError>, Vec) { + let ended = tokio::time::timeout(std::time::Duration::from_secs(5), &mut self.ended) + .await + .expect("the connection ends promptly") + .expect("the connection thread reports how it ended"); + let mut emitted = Vec::new(); + loop { + match self.events.try_recv() { + Ok(envelope) => emitted.push(envelope.payload.clone()), + Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => continue, + Err(_) => break, + } + } + (ended, emitted) + } + } + + /// A connection let go while its agent has not answered Initialize stops + /// there. The `Disconnect` command goes unread until the conversation loop + /// starts and Initialize waits up to 60 s, so it used to sit that wait out + /// — with the agent process kept running for nobody. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_cancelled_handshake_stops_waiting_for_initialize() { + let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); + let mut agent = HandshakeAgent::start(agent_end, true); + let run = HandshakeRun::start(client_end).await; + assert_eq!(agent.next_request().await, "initialize"); + + run.handshake.cancel(); + + let (ended, emitted) = run.ended().await; + assert!(ended.is_ok(), "being let go is not a failure: {ended:?}"); + assert!( + !emitted.iter().any(|e| matches!(e, AcpEvent::Error { .. })), + "{emitted:#?}" + ); + agent.closed().await; + } + + /// The same once Initialize is answered and the agent is opening the + /// session — the longer wait for an agent that starts its CLI and its MCP + /// servers there. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_cancelled_handshake_stops_waiting_for_the_session() { + let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); + let mut agent = HandshakeAgent::start(agent_end, false); + let run = HandshakeRun::start(client_end).await; + assert_eq!(agent.next_request().await, "initialize"); + assert_eq!(agent.next_request().await, "session/new"); + + run.handshake.cancel(); + + let (ended, emitted) = run.ended().await; + assert!(ended.is_ok(), "being let go is not a failure: {ended:?}"); + assert!( + !emitted + .iter() + .any(|e| matches!(e, AcpEvent::SessionStarted { .. } | AcpEvent::Error { .. })), + "{emitted:#?}" + ); + agent.closed().await; + } + + /// Why the handshake has a signal of its own rather than reading the + /// command channel: a prompt sent before the session is up waits in that + /// channel, and has to reach the agent once the session is. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_prompt_sent_during_the_handshake_reaches_the_agent() { + let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); + let mut agent = HandshakeAgent::start(agent_end, false); + let run = HandshakeRun::start(client_end).await; + assert_eq!(agent.next_request().await, "initialize"); + assert_eq!(agent.next_request().await, "session/new"); + + run.prompt("hello").await; + agent.open_session.notify_one(); + + assert_eq!(agent.next_request().await, "session/prompt: hello"); + run.cmd_tx + .send(ConnectionCommand::Disconnect) + .await + .expect("the connection is running"); + let (ended, _) = run.ended().await; + assert!(ended.is_ok(), "{ended:?}"); + agent.closed().await; + } + #[test] fn the_model_selector_is_the_model_recorded_on_a_turn() { let select = |id: &str, category: &str, current: &str| SessionConfigOptionInfo { diff --git a/src-tauri/src/acp/lifecycle.rs b/src-tauri/src/acp/lifecycle.rs index fc4bbad4d3..604a45fd8a 100644 --- a/src-tauri/src/acp/lifecycle.rs +++ b/src-tauri/src/acp/lifecycle.rs @@ -1794,6 +1794,7 @@ mod tests { status: crate::acp::types::ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(RwLock::new(state)), emitter: EventEmitter::Noop, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), diff --git a/src-tauri/src/acp/manager.rs b/src-tauri/src/acp/manager.rs index c38c665e03..2b0edd621e 100644 --- a/src-tauri/src/acp/manager.rs +++ b/src-tauri/src/acp/manager.rs @@ -622,6 +622,7 @@ impl ConnectionManager { status: ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(tokio::sync::RwLock::new(state)), emitter, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), @@ -665,6 +666,7 @@ impl ConnectionManager { status: ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(tokio::sync::RwLock::new(state)), emitter, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), @@ -2360,6 +2362,10 @@ impl ConnectionManager { }; if let Some(conn) = removed { tracing::info!("[ACP] disconnect connection={}", conn_id); + // First: a connection still in its handshake is not reading + // `cmd_tx`, and a full channel would hold the send below until it + // did. Cancelled, it stops and drops the receiver instead. + conn.handshake.cancel(); let _ = conn.cmd_tx.send(ConnectionCommand::Disconnect).await; Ok(()) } else { @@ -2624,6 +2630,7 @@ impl ConnectionManager { // Same handoff as `disconnect`: closing a window leaves // the agents exiting, not exited. self.park_draining(&conn).await; + conn.handshake.cancel(); txs.push(conn.cmd_tx); } } @@ -2710,6 +2717,7 @@ impl ConnectionManager { // immediately, so the child is parked as draining to stay // visible while it exits — and, here, to be swept below. self.park_draining(&conn).await; + conn.handshake.cancel(); txs.push(conn.cmd_tx); } } @@ -2817,7 +2825,10 @@ impl ConnectionManager { let mut connections = self.connections.lock().await; connections .drain() - .map(|(_, conn)| (conn.cmd_tx, conn.child_pid)) + .map(|(_, conn)| { + conn.handshake.cancel(); + (conn.cmd_tx, conn.child_pid) + }) .collect() }; let disconnected = handles.len(); @@ -4502,6 +4513,7 @@ mod tests { status: ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(RwLock::new(state)), emitter: EventEmitter::Noop, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), @@ -4759,6 +4771,126 @@ mod tests { ); } + /// Every path that ends a connection also cancels its handshake. The + /// `Disconnect` each sends goes unread until the session is established, so + /// a path that skipped the cancel would leave a connection still starting + /// up to finish its handshake — and its agent running — before it noticed. + #[tokio::test] + async fn every_disconnect_path_cancels_the_handshake() { + let mgr = ConnectionManager::new(); + async fn connect( + mgr: &ConnectionManager, + id: &str, + agent_type: AgentType, + ) -> tokio_util::sync::CancellationToken { + mgr.insert_test_connection(id, agent_type, None, EventEmitter::Noop) + .await; + mgr.connections.lock().await[id].handshake.clone() + } + + let closed_tab = connect(&mgr, "closed-tab", AgentType::Codex).await; + mgr.disconnect("closed-tab").await.unwrap(); + assert!(closed_tab.is_cancelled(), "disconnect"); + + let closed_window = connect(&mgr, "closed-window", AgentType::Codex).await; + mgr.disconnect_by_owner_window("test-window").await; + assert!(closed_window.is_cancelled(), "disconnect_by_owner_window"); + + let signed_out = connect(&mgr, "signed-out", AgentType::Antigravity).await; + mgr.disconnect_by_agent_type(AgentType::Antigravity).await; + assert!(signed_out.is_cancelled(), "disconnect_by_agent_type"); + + let quit = connect(&mgr, "quit", AgentType::Codex).await; + mgr.disconnect_all().await; + assert!(quit.is_cancelled(), "disconnect_all"); + } + + /// The case that made it matter: a draft switching agents disconnects the + /// connection it is leaving, whose agent may not have answered Initialize + /// yet. The `Disconnect` goes unread until the conversation loop starts, + /// and Initialize waits up to 60 s, so the agent process outlived the + /// disconnect by up to a minute. + /// + /// The agent is `sleep`, launched as a custom binary agent: it never + /// answers. Unix-only (relies on `sleep` / `kill(2)`). + #[cfg(unix)] + #[allow(clippy::await_holding_lock)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn disconnect_stops_an_agent_that_has_not_answered_initialize() { + use crate::acp::custom_registry::{ + hydrate, hydrate_test_guard, BinaryPlatformSpec, CustomAgentDef, CustomAgentSpec, + CustomDistributionKind, + }; + + // `hydrate` replaces a process-global map; hold its lock throughout. + let _registry = hydrate_test_guard(); + let def = CustomAgentDef { + registry_id: "never-answers".into(), + name: "Never answers".into(), + description: String::new(), + version: "1.0.0".into(), + distribution_kind: CustomDistributionKind::Binary, + spec: CustomAgentSpec { + binary: BTreeMap::from([( + crate::acp::registry::current_platform().to_string(), + BinaryPlatformSpec { + archive: "https://example.invalid/never-answers.tar.gz".into(), + // Nothing cached under this id, so the launch falls + // back to `sleep` on PATH. + cmd: "sleep".into(), + args: vec!["30".into()], + env: BTreeMap::new(), + sha256: None, + }, + )]), + ..Default::default() + }, + icon_url: None, + skills_shared_store: false, + skills_dir: None, + source: Default::default(), + version_probe: None, + supports_mcp: true, + }; + assert!(hydrate(std::slice::from_ref(&def)).is_empty()); + let agent_type = AgentType::custom("never-answers").unwrap(); + + let mgr = ConnectionManager::new(); + let dir = tempfile::tempdir().expect("tempdir"); + let id = mgr + .spawn_agent( + agent_type, + Some(dir.path().to_string_lossy().into_owned()), + None, + BTreeMap::new(), + "test-window".to_string(), + EventEmitter::Noop, + None, + BTreeMap::new(), + ) + .await + .expect("the agent launches"); + let pid_cell = Arc::clone(&mgr.connections.lock().await[&id].child_pid); + let mut pid = 0; + for _ in 0..500 { + pid = pid_cell.load(std::sync::atomic::Ordering::SeqCst); + if pid != 0 { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_ne!(pid, 0, "the agent process never started"); + + mgr.disconnect(&id).await.unwrap(); + + let gone = wait_until_dead(pid as i32).await; + hydrate(&[]); + assert!( + gone, + "agent {pid} outlived its disconnect — the handshake kept it running" + ); + } + /// Like [`spawn_process_tree`], but both levels ignore `SIGTERM`: the shell /// sets it to `SIG_IGN`, which `sleep` then inherits across `exec`. Stands /// in for an agent that traps it and takes its time — or never leaves. @@ -4957,6 +5089,7 @@ mod tests { status: ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(RwLock::new(state)), emitter: EventEmitter::Noop, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), @@ -5850,6 +5983,7 @@ mod tests { status: ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(RwLock::new(state)), emitter: EventEmitter::Noop, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), @@ -7533,6 +7667,7 @@ mod tests { status: ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(RwLock::new(state)), emitter: EventEmitter::Noop, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), @@ -8257,6 +8392,7 @@ mod tests { status: ConnectionStatus::Connected, owner_window_label: "test-window".to_string(), cmd_tx: tx, + handshake: tokio_util::sync::CancellationToken::new(), state: Arc::new(RwLock::new(state)), emitter: EventEmitter::Noop, prompt_lock: Arc::new(tokio::sync::Mutex::new(())), From 68c13c930ddbc75ddd2b99efc33aeee0dc9f840b Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 09:30:47 +0800 Subject: [PATCH 10/11] fix(acp): end a connection's terminals when its handshake stops early Terminals an agent starts through terminal/create are codeg's own children, so stopping the agent does not stop them. Only the conversation loop released them, per session on its way out, and a handshake that stops before the loop - cancelled by a disconnect, or failed - never got there. run_connection now ends every terminal its connection still holds, however the connection ends. A handshake error now counts as the cancel's own only when the token was in fact cancelled, so an agent error carrying the same text still takes the failure path. --- src-tauri/src/acp/connection.rs | 175 +++++++++++++++++++++++--- src-tauri/src/acp/manager.rs | 16 ++- src-tauri/src/acp/terminal_runtime.rs | 79 ++++++++++-- 3 files changed, 243 insertions(+), 27 deletions(-) diff --git a/src-tauri/src/acp/connection.rs b/src-tauri/src/acp/connection.rs index 1d0dd1586d..e1856c777c 100644 --- a/src-tauri/src/acp/connection.rs +++ b/src-tauri/src/acp/connection.rs @@ -6027,6 +6027,9 @@ async fn run_connection( Arc::clone(&prompt_ledger), ); + // For after the connection ends; the closure below takes the originals. + let terminals = Arc::clone(&terminal_runtime); + let handshake_token = handshake.clone(); let ended = Client .builder() .name("codeg") @@ -6859,11 +6862,11 @@ async fn run_connection( // boundary instead of growing holes. // // Unlike the requests around it, the drain does not give - // way to a cancelled handshake: the agent sends the - // replay before it answers the load, so there is little - // left to wait for, and stopping partway would leave a - // hydrated transcript cut short for good — once it has - // history, the next load never hydrates it again. + // way to a cancelled handshake: the agent sends the whole + // replay before it answers the load, so the drain reads + // what has already arrived, and stopping partway would + // leave a hydrated transcript cut short for good — once + // it has history, the next load never hydrates it again. let mut recording = hydrate_from_replay; while let Ok(Ok(dispatch)) = tokio::time::timeout( std::time::Duration::from_millis(100), @@ -7335,11 +7338,19 @@ async fn run_connection( } }) .await; + // Nothing the connection started may outlive it. The conversation loop + // releases its session's terminals on the way out, but a handshake that + // stopped early — cancelled, or failed — never reaches the loop. + terminals.release_all().await; match ended { // Let go before its session was up. Leaving the closure above has // already shut the connection down, and dropping its transport stops - // the agent process. - Err(err) if err.to_string().contains(HANDSHAKE_CANCELLED_SENTINEL) => { + // the agent process. The token has to agree: an error the agent sent + // could carry the same text. + Err(err) + if handshake_token.is_cancelled() + && err.to_string().contains(HANDSHAKE_CANCELLED_SENTINEL) => + { tracing::info!("[ACP] disconnected during the handshake; stopped it there"); Ok(()) } @@ -25605,11 +25616,23 @@ mod tests { agent.await.unwrap(); } + /// What a [`HandshakeAgent`] does with the requests that establish a session. + #[derive(Clone, Default)] + struct HandshakeScript { + /// Never answer Initialize. + stall_initialize: bool, + /// While opening the session, have codeg run this command line in a + /// terminal it hosts, and report "terminal started" once it has. + terminal: Option, + /// Refuse session/new, once released, instead of opening a session. + fail_session_new: bool, + } + /// An in-process agent for `run_connection`'s handshake. It reports each /// request it receives (a prompt with its text), answers Initialize unless - /// told to stall there, holds session/new until `open_session` is notified, - /// and ends every prompt at once. `closed` fires when codeg's end of the - /// connection goes away. + /// its script stalls there, holds session/new until `open_session` is + /// notified, and ends every prompt at once. `closed` fires when codeg's end + /// of the connection goes away. struct HandshakeAgent { seen: mpsc::UnboundedReceiver, open_session: Arc, @@ -25618,9 +25641,14 @@ mod tests { } impl HandshakeAgent { - fn start(agent_end: agent_client_protocol::Channel, stall_initialize: bool) -> Self { + fn start(agent_end: agent_client_protocol::Channel, script: HandshakeScript) -> Self { use agent_client_protocol::schema::v1::{InitializeResponse, PromptResponse}; + let HandshakeScript { + stall_initialize, + terminal, + fail_session_new, + } = script; let (seen_tx, seen) = mpsc::unbounded_channel(); let (init_seen, new_seen, prompt_seen) = (seen_tx.clone(), seen_tx.clone(), seen_tx); let open_session = Arc::new(tokio::sync::Notify::new()); @@ -25649,13 +25677,29 @@ mod tests { .on_receive_request( async move |_req: NewSessionRequest, responder: Responder, - _cx: ConnectionTo| { + cx: ConnectionTo| { let _ = new_seen.send("session/new".to_string()); - let release = Arc::clone(&release); + let (seen, release) = (new_seen.clone(), Arc::clone(&release)); + let terminal = terminal.clone(); // Off the handler, so the held answer doesn't hold // up the agent's other incoming messages. tokio::spawn(async move { + if let Some(command) = terminal { + cx.send_request_to( + Client, + CreateTerminalRequest::new(SessionId::new("s1"), command), + ) + .block_task() + .await?; + let _ = seen.send("terminal started".to_string()); + } release.notified().await; + if fail_session_new { + return responder.respond_with_error( + agent_client_protocol::Error::internal_error() + .data("scripted session/new failure"), + ); + } responder.respond(NewSessionResponse::new(SessionId::new("s1"))) }); Ok(()) @@ -25818,7 +25862,13 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_cancelled_handshake_stops_waiting_for_initialize() { let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); - let mut agent = HandshakeAgent::start(agent_end, true); + let mut agent = HandshakeAgent::start( + agent_end, + HandshakeScript { + stall_initialize: true, + ..Default::default() + }, + ); let run = HandshakeRun::start(client_end).await; assert_eq!(agent.next_request().await, "initialize"); @@ -25839,7 +25889,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_cancelled_handshake_stops_waiting_for_the_session() { let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); - let mut agent = HandshakeAgent::start(agent_end, false); + let mut agent = HandshakeAgent::start(agent_end, HandshakeScript::default()); let run = HandshakeRun::start(client_end).await; assert_eq!(agent.next_request().await, "initialize"); assert_eq!(agent.next_request().await, "session/new"); @@ -25863,7 +25913,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_prompt_sent_during_the_handshake_reaches_the_agent() { let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); - let mut agent = HandshakeAgent::start(agent_end, false); + let mut agent = HandshakeAgent::start(agent_end, HandshakeScript::default()); let run = HandshakeRun::start(client_end).await; assert_eq!(agent.next_request().await, "initialize"); assert_eq!(agent.next_request().await, "session/new"); @@ -25881,6 +25931,99 @@ mod tests { agent.closed().await; } + /// A [`HandshakeAgent`] that starts a terminal while opening the session, + /// and the pid the terminal's command reports once it runs. + #[cfg(unix)] + async fn agent_with_a_terminal( + agent_end: agent_client_protocol::Channel, + pid_file: &std::path::Path, + fail_session_new: bool, + ) -> (HandshakeAgent, i32) { + let mut agent = HandshakeAgent::start( + agent_end, + HandshakeScript { + terminal: Some(format!( + "sh -c 'echo $$ > {}; exec sleep 60'", + pid_file.display() + )), + fail_session_new, + ..Default::default() + }, + ); + assert_eq!(agent.next_request().await, "initialize"); + assert_eq!(agent.next_request().await, "session/new"); + assert_eq!(agent.next_request().await, "terminal started"); + for _ in 0..200 { + if let Some(pid) = std::fs::read_to_string(pid_file) + .ok() + .and_then(|raw| raw.trim().parse::().ok()) + { + return (agent, pid); + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + panic!("the terminal never reported its pid"); + } + + /// Whether `pid` is gone within a few seconds. + #[cfg(unix)] + async fn process_ends(pid: i32) -> bool { + for _ in 0..200 { + // SAFETY: signal 0 only probes for existence; it sends no signal. + if unsafe { libc::kill(pid, 0) } != 0 { + return true; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + false + } + + /// A terminal the agent starts while the session is being opened is + /// codeg's own child, not the agent's: stopping the agent leaves it running. + /// The conversation loop releases a session's terminals on its way out, but + /// a cancelled handshake never reaches the loop. + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_cancelled_handshake_ends_the_terminals_it_started() { + let dir = tempfile::tempdir().expect("tempdir"); + let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); + let run = HandshakeRun::start(client_end).await; + let (agent, pid) = + agent_with_a_terminal(agent_end, &dir.path().join("terminal.pid"), false).await; + + run.handshake.cancel(); + + let (ended, _) = run.ended().await; + assert!(ended.is_ok(), "{ended:?}"); + assert!( + process_ends(pid).await, + "terminal process {pid} outlived its connection" + ); + agent.closed().await; + } + + /// The same when the handshake fails instead: the agent refuses the session + /// after starting a terminal for it. + #[cfg(unix)] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_failed_handshake_ends_the_terminals_it_started() { + let dir = tempfile::tempdir().expect("tempdir"); + let (client_end, agent_end) = agent_client_protocol::Channel::duplex(); + let run = HandshakeRun::start(client_end).await; + let (agent, pid) = + agent_with_a_terminal(agent_end, &dir.path().join("terminal.pid"), true).await; + + agent.open_session.notify_one(); + + let (ended, _) = run.ended().await; + assert!(ended.is_err(), "the agent refused the session"); + assert!( + process_ends(pid).await, + "terminal process {pid} outlived its connection" + ); + agent.closed().await; + } + #[test] fn the_model_selector_is_the_model_recorded_on_a_turn() { let select = |id: &str, category: &str, current: &str| SessionConfigOptionInfo { diff --git a/src-tauri/src/acp/manager.rs b/src-tauri/src/acp/manager.rs index 2b0edd621e..bbfcd74293 100644 --- a/src-tauri/src/acp/manager.rs +++ b/src-tauri/src/acp/manager.rs @@ -4838,7 +4838,7 @@ mod tests { // Nothing cached under this id, so the launch falls // back to `sleep` on PATH. cmd: "sleep".into(), - args: vec!["30".into()], + args: vec!["120".into()], env: BTreeMap::new(), sha256: None, }, @@ -4880,10 +4880,22 @@ mod tests { tokio::time::sleep(Duration::from_millis(20)).await; } assert_ne!(pid, 0, "the agent process never started"); + // Up and still waiting to be answered, so whatever ends it below is + // the disconnect and not the agent leaving on its own. + assert!(is_alive(pid as i32), "the agent exited on its own"); mgr.disconnect(&id).await.unwrap(); - let gone = wait_until_dead(pid as i32).await; + // Longer than `wait_until_dead` allows, for a loaded CI machine; the + // handshake used to hold on for up to 60 s. + let mut gone = false; + for _ in 0..500 { + if !is_alive(pid as i32) { + gone = true; + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } hydrate(&[]); assert!( gone, diff --git a/src-tauri/src/acp/terminal_runtime.rs b/src-tauri/src/acp/terminal_runtime.rs index 93374e3145..45299c0f9d 100644 --- a/src-tauri/src/acp/terminal_runtime.rs +++ b/src-tauri/src/acp/terminal_runtime.rs @@ -839,15 +839,22 @@ impl TerminalRuntime { removed }; - // Kill concurrently. Sequentially, a session holding N terminals would - // cost up to N * KILL_REPORT_BUDGET to tear down, and this runs on the - // cancel path where the user is waiting. - futures::future::join_all(removed.into_iter().map(|terminal| async move { - if let Err(err) = terminal.kill_command().await { - tracing::error!("[ACP] Failed to release terminal during cleanup: {err:?}"); - } - })) - .await; + kill_released(removed).await; + } + + /// Kill every terminal still held here, whatever session it was started + /// for. For a connection that is ending: these are codeg's own children, + /// so ending the agent does not end them, and nothing else will once the + /// connection is gone. + pub async fn release_all(&self) { + let removed: Vec<_> = self + .terminals + .lock() + .await + .drain() + .map(|(_, terminal)| terminal) + .collect(); + kill_released(removed).await; } async fn find_terminal( @@ -873,6 +880,18 @@ impl TerminalRuntime { } } +/// Kill terminals taken out of the map, concurrently. Sequentially, N terminals +/// would cost up to N * KILL_REPORT_BUDGET to tear down, and this runs on the +/// cancel path where the user is waiting. +async fn kill_released(removed: Vec>) { + futures::future::join_all(removed.into_iter().map(|terminal| async move { + if let Err(err) = terminal.kill_command().await { + tracing::error!("[ACP] Failed to release terminal during cleanup: {err:?}"); + } + })) + .await; +} + async fn read_stream(mut reader: R, terminal: Arc) where R: AsyncRead + Unpin, @@ -2038,6 +2057,48 @@ mod tests { ); } + /// An ending connection ends every terminal it still holds, whatever + /// session each was started for — including the ones a handshake that + /// stopped early never got to release by session. + #[tokio::test] + async fn release_all_ends_every_terminal_whatever_its_session() { + let dir = tempfile::tempdir().expect("temp dir"); + let runtime = Arc::new(TerminalRuntime::with_base_env(BTreeMap::new())); + let mut pids = Vec::new(); + for session in ["first-session", "second-session"] { + let pid_file = dir.path().join(format!("{session}.pid")); + let command = format!("sh -c 'echo $$ > {}; exec sleep 30'", pid_file.display()); + runtime + .create_terminal(CreateTerminalRequest::new( + SessionId::new(session.to_string()), + command, + )) + .await + .expect("create terminal"); + let mut pid = None; + for _ in 0..200 { + pid = std::fs::read_to_string(&pid_file) + .ok() + .and_then(|raw| raw.trim().parse::().ok()); + if pid.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + pids.push(pid.expect("the terminal announced its pid")); + } + + runtime.release_all().await; + + for pid in pids { + assert!( + wait_until_pid_gone(pid, Duration::from_secs(5)).await, + "terminal process {pid} outlived release_all" + ); + } + assert!(runtime.terminals.lock().await.is_empty()); + } + /// Output is readable while the command is still running — the snapshot is /// not gated on the process finishing. #[tokio::test] From 8214e7a60b93d1dc781ced933da59856d24f5af4 Mon Sep 17 00:00:00 2001 From: xintaofei Date: Thu, 15 Oct 2026 09:49:05 +0800 Subject: [PATCH 11/11] fix(acp): end a terminal whose creation is dropped before registering it create_terminal spawns the command, then awaited the reader-handle lock and the terminal map before the terminal was registered. Dropped at either await - its connection shutting down mid-request - the child was dropped unkilled (tokio does not kill on drop), or kept running by its owner task with no map entry, where no release could reach it. The reader handles are now registered without awaiting, so nothing between the spawn and the owner task can yield, and a guard has the owner kill the terminal if the creation is dropped before the map insert lands. The terminal tests now also check the command is still running right before the teardown they test. --- src-tauri/src/acp/connection.rs | 9 ++- src-tauri/src/acp/terminal_runtime.rs | 88 ++++++++++++++++++++++++++- 2 files changed, 94 insertions(+), 3 deletions(-) diff --git a/src-tauri/src/acp/connection.rs b/src-tauri/src/acp/connection.rs index e1856c777c..c3918324e7 100644 --- a/src-tauri/src/acp/connection.rs +++ b/src-tauri/src/acp/connection.rs @@ -25943,7 +25943,7 @@ mod tests { agent_end, HandshakeScript { terminal: Some(format!( - "sh -c 'echo $$ > {}; exec sleep 60'", + "sh -c 'echo $$ > {}; exec sleep 120'", pid_file.display() )), fail_session_new, @@ -25958,6 +25958,13 @@ mod tests { .ok() .and_then(|raw| raw.trim().parse::().ok()) { + // Running, so whatever ends it later is the teardown under test. + // SAFETY: signal 0 only probes for existence; it sends no signal. + assert_eq!( + unsafe { libc::kill(pid, 0) }, + 0, + "the terminal exited on its own" + ); return (agent, pid); } tokio::time::sleep(std::time::Duration::from_millis(25)).await; diff --git a/src-tauri/src/acp/terminal_runtime.rs b/src-tauri/src/acp/terminal_runtime.rs index 45299c0f9d..afe50fe688 100644 --- a/src-tauri/src/acp/terminal_runtime.rs +++ b/src-tauri/src/acp/terminal_runtime.rs @@ -703,7 +703,16 @@ impl TerminalRuntime { } if !handles.is_empty() { - terminal.reader_handles.lock().await.extend(handles); + // Without awaiting: from the spawn above to the owner below, this + // future must not yield — dropped there (its connection shutting + // down mid-request), it would drop `child`, which tokio does not + // kill. Nothing else holds this lock yet; its only other user is the + // owner task. + terminal + .reader_handles + .try_lock() + .expect("a new terminal's reader handles are not shared yet") + .extend(handles); } // Hand the child to its owner task only AFTER the reader handles are @@ -712,10 +721,12 @@ impl TerminalRuntime { // list and leave the readers running past the published completion. tokio::spawn(own_terminal_process(terminal.clone(), child)); + let unregistered = KillUnlessRegistered(Some(Arc::clone(&terminal))); self.terminals .lock() .await .insert(terminal_id.clone(), terminal); + unregistered.registered(); Ok(CreateTerminalResponse::new(terminal_id)) } @@ -880,6 +891,26 @@ impl TerminalRuntime { } } +/// Kills a new terminal if `create_terminal` is dropped before the terminal is +/// in the map, which is where every later release looks for it — dropped, say, +/// while waiting for the map as its connection shuts down mid-request. Its +/// owner task does the killing, as for any other kill. +struct KillUnlessRegistered(Option>); + +impl KillUnlessRegistered { + fn registered(mut self) { + self.0 = None; + } +} + +impl Drop for KillUnlessRegistered { + fn drop(&mut self) { + if let Some(terminal) = self.0.take() { + terminal.kill.notify_one(); + } + } +} + /// Kill terminals taken out of the map, concurrently. Sequentially, N terminals /// would cost up to N * KILL_REPORT_BUDGET to tear down, and this runs on the /// cancel path where the user is waiting. @@ -2067,7 +2098,7 @@ mod tests { let mut pids = Vec::new(); for session in ["first-session", "second-session"] { let pid_file = dir.path().join(format!("{session}.pid")); - let command = format!("sh -c 'echo $$ > {}; exec sleep 30'", pid_file.display()); + let command = format!("sh -c 'echo $$ > {}; exec sleep 120'", pid_file.display()); runtime .create_terminal(CreateTerminalRequest::new( SessionId::new(session.to_string()), @@ -2087,6 +2118,12 @@ mod tests { } pids.push(pid.expect("the terminal announced its pid")); } + for &pid in &pids { + assert!( + pid_is_alive(pid), + "terminal process {pid} exited on its own" + ); + } runtime.release_all().await; @@ -2099,6 +2136,53 @@ mod tests { assert!(runtime.terminals.lock().await.is_empty()); } + /// Until a new terminal is in the map, nothing that releases terminals can + /// find it. A `create_terminal` dropped while it waits to put it there — its + /// connection shutting down mid-request — must not leave the command + /// running for good. + #[tokio::test] + async fn a_terminal_whose_creation_is_dropped_before_registering_is_ended() { + let dir = tempfile::tempdir().expect("temp dir"); + let pid_file = dir.path().join("terminal.pid"); + let runtime = Arc::new(TerminalRuntime::with_base_env(BTreeMap::new())); + // Holding the map makes the creation wait right where it registers. + let map = runtime.terminals.lock().await; + let creating = tokio::spawn({ + let runtime = Arc::clone(&runtime); + let command = format!("sh -c 'echo $$ > {}; exec sleep 120'", pid_file.display()); + async move { + runtime + .create_terminal(CreateTerminalRequest::new( + SessionId::new("dropped-creation".to_string()), + command, + )) + .await + } + }); + let mut pid = None; + for _ in 0..200 { + pid = std::fs::read_to_string(&pid_file) + .ok() + .and_then(|raw| raw.trim().parse::().ok()); + if pid.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + let pid = pid.expect("the terminal announced its pid"); + assert!(pid_is_alive(pid), "the command exited on its own"); + + creating.abort(); + assert!(creating.await.is_err_and(|e| e.is_cancelled())); + drop(map); + + assert!( + wait_until_pid_gone(pid, Duration::from_secs(10)).await, + "terminal process {pid} outlived its dropped creation" + ); + assert!(runtime.terminals.lock().await.is_empty()); + } + /// Output is readable while the command is still running — the snapshot is /// not gated on the process finishing. #[tokio::test]