diff --git a/apps/staged/src-tauri/src/pikchr_mcp.rs b/apps/staged/src-tauri/src/pikchr_mcp.rs
index 99115aa57..aa3ebfe71 100644
--- a/apps/staged/src-tauri/src/pikchr_mcp.rs
+++ b/apps/staged/src-tauri/src/pikchr_mcp.rs
@@ -1713,6 +1713,7 @@ arrow from COLL.e to SNOW.w"#;
             cancel_token: &CancellationToken,
             _agent_session_id: Option<&str>,
             _config_options: &[acp_client::AcpSessionConfigOptionSelection],
+            _auth_selection: Option<&acp_client::AcpAuthenticationSelection>,
         ) -> Result<acp_client::AgentRunOutcome, String> {
             assert!(
                 self.registry.cancel(session_id),
@@ -1804,6 +1805,7 @@ arrow from COLL.e to SNOW.w"#;
             _cancel_token: &CancellationToken,
             _agent_session_id: Option<&str>,
             _config_options: &[acp_client::AcpSessionConfigOptionSelection],
+            _auth_selection: Option<&acp_client::AcpAuthenticationSelection>,
         ) -> Result<acp_client::AgentRunOutcome, String> {
             self.ran.set(true);
             Ok(acp_client::AgentRunOutcome::Completed)
diff --git a/apps/staged/src-tauri/src/pikchr_subsession.rs b/apps/staged/src-tauri/src/pikchr_subsession.rs
index 5d58fab2a..23cde89cf 100644
--- a/apps/staged/src-tauri/src/pikchr_subsession.rs
+++ b/apps/staged/src-tauri/src/pikchr_subsession.rs
@@ -328,6 +328,7 @@ async fn generate_pikchr_source_inner<D: AgentDriver + ?Sized>(
                 cancel_token,
                 agent_session_id.as_deref(),
                 config_options,
+                None,
             )
             .await;
         writer_dyn.finalize().await;
@@ -559,6 +560,7 @@ mod tests {
             _cancel_token: &CancellationToken,
             agent_session_id: Option<&str>,
             config_options: &[acp_client::AcpSessionConfigOptionSelection],
+            _auth_selection: Option<&acp_client::AcpAuthenticationSelection>,
         ) -> Result<acp_client::AgentRunOutcome, String> {
             *self.calls.lock().unwrap() += 1;
             self.seen_session_ids
@@ -1162,6 +1164,7 @@ agent: http=false, sse=false). Select a provider that supports MCP over HTTP/SSE
             cancel_token: &CancellationToken,
             agent_session_id: Option<&str>,
             config_options: &[acp_client::AcpSessionConfigOptionSelection],
+            _auth_selection: Option<&acp_client::AcpAuthenticationSelection>,
         ) -> Result<acp_client::AgentRunOutcome, String> {
             self.store
                 .update_session_status(
@@ -1182,6 +1185,7 @@ agent: http=false, sse=false). Select a provider that supports MCP over HTTP/SSE
                     cancel_token,
                     agent_session_id,
                     config_options,
+                    None,
                 )
                 .await
         }
diff --git a/apps/staged/src-tauri/src/session_runner.rs b/apps/staged/src-tauri/src/session_runner.rs
index 1948675cf..bd87e4bae 100644
--- a/apps/staged/src-tauri/src/session_runner.rs
+++ b/apps/staged/src-tauri/src/session_runner.rs
@@ -1403,6 +1403,7 @@ pub fn start_session(
                         &cancel_token,
                         agent_session_id.as_deref(),
                         &selected_acp_config_options,
+                        None,
                     )
                     .await
                 {
diff --git a/apps/staged/src/lib/features/doctor/agentLogin.svelte.ts b/apps/staged/src/lib/features/doctor/agentLogin.svelte.ts
index e3551fdb4..4d15939be 100644
--- a/apps/staged/src/lib/features/doctor/agentLogin.svelte.ts
+++ b/apps/staged/src/lib/features/doctor/agentLogin.svelte.ts
@@ -54,6 +54,7 @@ import { listenToEvent, type UnlistenFn } from '../../transport';
 
 /** Output lines kept for display, oldest first. */
 const MAX_OUTPUT_LINES = 40;
+const LOGIN_OUTPUT_SUBSCRIPTION_ERROR = 'Could not subscribe to login output, try again';
 
 /** How a login ended, short of failing. */
 export type AgentLoginOutcome = 'completed' | 'cancelled';
@@ -300,6 +301,16 @@ function fail(attempt: Attempt, error: string) {
   attempt.reject(new Error(error));
 }
 
+function failRegistration(attempt: Attempt) {
+  if (!live(attempt)) return;
+  if (attempt.probing) {
+    finish(attempt);
+    attempt.reject(new Error(LOGIN_OUTPUT_SUBSCRIPTION_ERROR));
+    return;
+  }
+  fail(attempt, LOGIN_OUTPUT_SUBSCRIPTION_ERROR);
+}
+
 /** A probe that found nothing — or was superseded — leaves the record alone. */
 function abandon(attempt: Attempt) {
   finish(attempt);
@@ -472,6 +483,7 @@ export function startAgentLogin(checkId: string): Promise<AgentLoginOutcome> {
     'doctor-login-output',
     (output) => handleEvent(attempt, output),
     {
+      onRegistrationFailed: () => failRegistration(attempt),
       onEstablished: () => {
         if (!live(attempt)) return;
         if (started) {
@@ -555,6 +567,7 @@ export function attachAgentLogin(checkId: string): Promise<AgentLoginOutcome | n
     'doctor-login-output',
     (output) => handleEvent(attempt, output),
     {
+      onRegistrationFailed: () => failRegistration(attempt),
       onEstablished: () => {
         if (!live(attempt)) return;
         if (asked) {
diff --git a/apps/staged/src/lib/features/doctor/agentLogin.test.ts b/apps/staged/src/lib/features/doctor/agentLogin.test.ts
index 2ebe5c21d..74b673d29 100644
--- a/apps/staged/src/lib/features/doctor/agentLogin.test.ts
+++ b/apps/staged/src/lib/features/doctor/agentLogin.test.ts
@@ -6,6 +6,7 @@ interface Registration {
   event: string;
   callback: (payload: DoctorLoginOutput) => void;
   onEstablished?: () => void;
+  onRegistrationFailed?: (error: unknown) => void;
   unlisten: ReturnType<typeof vi.fn>;
 }
 
@@ -46,10 +47,16 @@ describe('agentLogin', () => {
       listenToEvent: (
         event: string,
         callback: (payload: DoctorLoginOutput) => void,
-        opts?: { onEstablished?: () => void }
+        opts?: { onEstablished?: () => void; onRegistrationFailed?: (error: unknown) => void }
       ) => {
         const unlisten = vi.fn();
-        registrations.push({ event, callback, onEstablished: opts?.onEstablished, unlisten });
+        registrations.push({
+          event,
+          callback,
+          onEstablished: opts?.onEstablished,
+          onRegistrationFailed: opts?.onRegistrationFailed,
+          unlisten,
+        });
         return unlisten;
       },
     }));
@@ -277,6 +284,33 @@ describe('agentLogin', () => {
     expect(agentLogin.running).toBe(false);
   });
 
+  it('reports listener registration failure instead of leaving a start running forever', async () => {
+    const { agentLogin, startAgentLogin } = await load();
+
+    const settled = startAgentLogin('ai-agent-claude');
+    const registration = only();
+    registration.onRegistrationFailed?.(new Error('ipc down'));
+
+    await expect(settled).rejects.toThrow('Could not subscribe to login output, try again');
+    expect(startDoctorLogin).not.toHaveBeenCalled();
+    expect(agentLogin.running).toBe(false);
+    expect(agentLogin.error).toBe('Could not subscribe to login output, try again');
+    expect(registration.unlisten).toHaveBeenCalledTimes(1);
+  });
+
+  it('reports listener registration failure instead of leaving an attach probe pending', async () => {
+    const { agentLogin, attachAgentLogin } = await load();
+
+    const attached = attachAgentLogin('ai-agent-claude');
+    const registration = only();
+    registration.onRegistrationFailed?.(new Error('ipc down'));
+
+    await expect(attached).rejects.toThrow('Could not subscribe to login output, try again');
+    expect(doctorLoginStatus).not.toHaveBeenCalled();
+    expect(agentLogin.running).toBe(false);
+    expect(registration.unlisten).toHaveBeenCalledTimes(1);
+  });
+
   it('keeps the code box open after a send, so a rejected code can be retried', async () => {
     const { agentLogin, startAgentLogin, submitAgentLoginCode } = await load();
 
diff --git a/apps/staged/src/lib/features/sessions/SessionChatPane.svelte b/apps/staged/src/lib/features/sessions/SessionChatPane.svelte
index 742def8ec..6612226f4 100644
--- a/apps/staged/src/lib/features/sessions/SessionChatPane.svelte
+++ b/apps/staged/src/lib/features/sessions/SessionChatPane.svelte
@@ -81,10 +81,9 @@
   } from '../../api/commands';
   import { listenToEvent, type UnlistenFn } from '../../transport';
   import { openSettings } from '../layout/navigation.svelte';
-  import { doctorState, runChecks } from '../doctor/doctor.svelte';
-  import { agentLogin, attachAgentLogin, startAgentLogin } from '../doctor/agentLogin.svelte';
   import AgentLoginPrompt from '../doctor/AgentLoginPrompt.svelte';
-  import { canOfferLogin, doctorCheckForProvider, isAuthenticationError } from './authRecovery';
+  import { isAuthenticationError } from './authRecovery';
+  import { createSessionLoginController } from './sessionLogin.svelte';
   import AcpFixedConfigPicker from '../agents/AcpFixedConfigPicker.svelte';
   import { agentState } from '../agents/agent.svelte';
   import {
@@ -244,64 +243,10 @@
    * `noteTaskStopOutcome`.
    */
   let taskStopNotices = $state<Map<string, string>>(new Map());
-  /** Doctor check id for this session's agent — the login's identity. */
-  let loginCheckId = $derived(session?.provider ? `ai-agent-${session.provider}` : null);
-  let loginCheck = $derived(doctorCheckForProvider(session?.provider, doctorState.report));
-  let canLogin = $derived(canOfferLogin(loginCheck));
-  let loginRunning = $derived(agentLogin.running && agentLogin.checkId === loginCheckId);
-  /**
-   * Sessions whose authentication failure has already asked for a report, so a
-   * scan that fails (leaving `report` null) isn't retried on every flush.
-   */
-  let authReportRequestedFor: string | null = null;
-  /**
-   * `Log in` is the primary action on an authentication failure, but it depends
-   * on doctor's auth probe — and `doctorState.report` is otherwise filled in
-   * only by opening the Doctor settings panel. On a fresh launch that left
-   * every auth-failed session showing `Fix` alone until the user had visited
-   * that panel and come back, so run the checks the first time such a failure
-   * is displayed.
-   */
-  $effect(() => {
-    const id = sessionId;
-    const failed = session?.status === 'error' || session?.status === 'cancelled';
-    if (!active || !id || !failed || !isAuthenticationError(session?.errorMessage)) return;
-    if (doctorState.report || doctorState.loading || authReportRequestedFor === id) return;
-    authReportRequestedFor = id;
-    void runChecks();
-  });
-  /**
-   * Sessions whose authentication failure has already asked the backend about a
-   * running login, per open — see below.
-   */
-  let loginAttachRequestedFor: string | null = null;
-  /**
-   * A login for this agent may already be running on the backend — started from
-   * the Doctor panel, from another client, or before this webview reloaded — with
-   * the shared record here knowing nothing of it. Ask once per open when the
-   * alert shows, so its URL and code box come back instead of a `Log in` the
-   * backend would answer "already running". Not gated on `canLogin`: that needs
-   * the doctor report, and the login exists whether or not it has arrived.
-   */
-  $effect(() => {
-    const id = sessionId;
-    const checkId = loginCheckId;
-    if (!active) {
-      loginAttachRequestedFor = null;
-      return;
-    }
-    const failed = session?.status === 'error' || session?.status === 'cancelled';
-    if (!id || !checkId || !failed || !isAuthenticationError(session?.errorMessage)) return;
-    if (agentLogin.running || loginAttachRequestedFor === id) return;
-    loginAttachRequestedFor = id;
-    void attachAgentLogin(checkId)
-      .then((outcome) => {
-        // A signed-in agent changes the check the "Log in" button depends on.
-        if (outcome === 'completed') void runChecks();
-      })
-      .catch(() => {
-        // The failure is on the shared login record, which the alert renders.
-      });
+  const login = createSessionLoginController({
+    getActive: () => active,
+    getSessionId: () => sessionId,
+    getSession: () => session,
   });
 
   let inputText = $state('');
@@ -678,9 +623,6 @@
     stopPolling();
     unlistenStatus?.();
     unlistenBackgroundHold?.();
-    // A login in flight is deliberately not torn down here: the subprocess
-    // outlives this pane, and its shared record is what the Doctor panel — or
-    // this pane on its next open — needs to keep feeding it a code.
   });
 
   // This pane can be mounted once and reused across opens (the `active` prop toggles
@@ -843,18 +785,6 @@
     if (taskStopNotices.size > 0) taskStopNotices = new Map();
   }
 
-  async function startLogin() {
-    if (!loginCheckId || !canLogin || agentLogin.running) return;
-    try {
-      const outcome = await startAgentLogin(loginCheckId);
-      // A signed-in agent changes the check the "Log in" button depends on; a
-      // cancelled login changes nothing.
-      if (outcome === 'completed') void runChecks();
-    } catch {
-      // The failure is on the shared login record, which the alert renders.
-    }
-  }
-
   function isComposerFocused(): boolean {
     return document.activeElement === inputEl;
   }
@@ -2398,9 +2328,9 @@
                  check passing (its probe can't see an expired token), so `Fix`
                  is the fallback, not the answer. -->
             <div class="auth-actions">
-              {#if canLogin}
-                <Button variant="default" size="xs" disabled={loginRunning} onclick={startLogin}>
-                  {loginRunning ? 'Logging in…' : 'Log in'}
+              {#if login.canLogin}
+                <Button variant="default" size="xs" disabled={login.running} onclick={login.start}>
+                  {login.running ? 'Logging in…' : 'Log in'}
                 </Button>
               {/if}
               <Button variant="outline" size="xs" onclick={() => openSettings('doctor')}>
@@ -2410,7 +2340,7 @@
           </Alert.Action>
         {/if}
       </Alert.Root>
-      <AgentLoginPrompt checkId={loginCheckId} />
+      <AgentLoginPrompt checkId={login.checkId} />
     {:else if session && session.status !== 'running' && session.status !== 'queued'}
       {#if isResumableReason(session.completionReason)}
         {@const isWarning =
diff --git a/apps/staged/src/lib/features/sessions/authRecovery.test.ts b/apps/staged/src/lib/features/sessions/authRecovery.test.ts
index 26de9fd5c..f34b1ee13 100644
--- a/apps/staged/src/lib/features/sessions/authRecovery.test.ts
+++ b/apps/staged/src/lib/features/sessions/authRecovery.test.ts
@@ -72,6 +72,52 @@ describe('authentication recovery helpers', () => {
     expect(isAuthenticationError('npm install failed with exit code 1')).toBe(false);
   });
 
+  // The exact strings the ACP driver produces for an ACP `-32000 auth_required`
+  // and the session runner stores verbatim as `session.errorMessage`. The inner
+  // sentence is `AcpAuthenticationRequired::describe` and the outer wrapper is
+  // `run`, both in `crates/acp-client/src/driver.rs`, whose own tests assert
+  // the same bytes for each of the three stages (`session_setup_returns_auth_
+  // required_without_guessing_a_method`, `full_session_setup_surfaces_auth_
+  // required_at_session_load`, `prompt_auth_required_reads_like_the_session_
+  // setup_error`). Pinned here so a change to either cannot silently drop the
+  // Log in action. A bridge that says more on its `-32000` gets that appended
+  // after this sentence, so these prefixes still hold.
+  it.each([
+    {
+      stage: 'session/new',
+      message:
+        'ACP protocol failed: Error { code: -32603: Internal error, message: "Internal error", data: Some(String("ACP authentication is required to create ACP session. Sign this agent in, then retry.")) }',
+    },
+    {
+      stage: 'session/load',
+      message:
+        'ACP protocol failed: Error { code: -32603: Internal error, message: "Internal error", data: Some(String("ACP authentication is required to load ACP session. Sign this agent in, then retry.")) }',
+    },
+    {
+      stage: 'session/prompt',
+      message:
+        'ACP protocol failed: Error { code: -32603: Internal error, message: "Internal error", data: Some(String("ACP authentication is required to run the prompt. Sign this agent in, then retry.")) }',
+    },
+    {
+      // What session/prompt produced before the driver described it: the raw
+      // Debug rendering of the ACP error. Kept so recognition never comes to
+      // depend on that rewrite.
+      stage: 'session/prompt (pre-describe Debug shape)',
+      message:
+        'ACP protocol failed: Error { code: -32603: Internal error, message: "Internal error", data: Some(String("Prompt failed: Error { code: -32000: Authentication required, message: \\"Authentication required\\", data: None }")) }',
+    },
+  ])('offers login for the driver error shape at $stage', ({ message }) => {
+    expect(isAuthenticationError(message)).toBe(true);
+  });
+
+  it("leaves goose's unconfigured native provider to the Fix action", () => {
+    // goose reports a missing native provider as an internal error carrying
+    // this phrase, not as `-32000`. Doctor has no login command for goose, so
+    // canOfferLogin is false for it regardless and the alert can only offer
+    // Fix; matching here would promise a Log in that cannot run.
+    expect(isAuthenticationError('ACP protocol failed: Provider is not configured')).toBe(false);
+  });
+
   describe('canOfferLogin', () => {
     it('offers login for a positively signed-out agent', () => {
       expect(canOfferLogin(check())).toBe(true);
diff --git a/apps/staged/src/lib/features/sessions/sessionLogin.svelte.ts b/apps/staged/src/lib/features/sessions/sessionLogin.svelte.ts
new file mode 100644
index 000000000..4da4a864e
--- /dev/null
+++ b/apps/staged/src/lib/features/sessions/sessionLogin.svelte.ts
@@ -0,0 +1,106 @@
+/**
+ * Pane-local orchestration over the shared agent login. Create once during
+ * component initialization; the effects follow the pane's reactive context.
+ *
+ * The shared Doctor controller owns run identity, output, code submission,
+ * cancellation and reconnect recovery. A login deliberately outlives the pane:
+ * closing or switching sessions must not tear it down, and its completion must
+ * still refresh the global doctor report. authRecovery.ts stays pure.
+ */
+import type { Session } from '../../types';
+import {
+  agentLogin,
+  attachAgentLogin,
+  startAgentLogin,
+  type AgentLoginOutcome,
+} from '../doctor/agentLogin.svelte';
+import { doctorState, runChecks } from '../doctor/doctor.svelte';
+import { canOfferLogin, doctorCheckForProvider, isAuthenticationError } from './authRecovery';
+
+export function createSessionLoginController(context: {
+  getActive: () => boolean;
+  getSessionId: () => string;
+  getSession: () => Pick<Session, 'provider' | 'status' | 'errorMessage'> | null;
+}) {
+  // A failed scan must not be retried on every flush or reopen. Unlike the
+  // report request, an attach probe is allowed again on the next open.
+  let reportRequestedFor: string | null = null;
+  let attachRequestedFor: string | null = null;
+
+  function checkId(): string | null {
+    const provider = context.getSession()?.provider;
+    return provider ? `ai-agent-${provider}` : null;
+  }
+
+  function canLogin(): boolean {
+    return canOfferLogin(
+      doctorCheckForProvider(context.getSession()?.provider, doctorState.report)
+    );
+  }
+
+  function hasAuthFailure(): boolean {
+    const session = context.getSession();
+    return (
+      (session?.status === 'error' || session?.status === 'cancelled') &&
+      isAuthenticationError(session.errorMessage)
+    );
+  }
+
+  function refreshAfterLogin(outcome: AgentLoginOutcome | null) {
+    // "Completed" can also mean a reconnect found the run had ended, not
+    // necessarily that sign-in succeeded. Doctor's probe decides that.
+    if (outcome === 'completed') void runChecks();
+  }
+
+  $effect(() => {
+    const id = context.getSessionId();
+    if (!context.getActive() || !id || !hasAuthFailure()) return;
+    // On a fresh launch the Doctor panel may never have populated the report
+    // that determines whether this provider can offer Log in.
+    if (doctorState.report || doctorState.loading || reportRequestedFor === id) return;
+    reportRequestedFor = id;
+    void runChecks();
+  });
+
+  $effect(() => {
+    const id = context.getSessionId();
+    const check = checkId();
+    if (!context.getActive()) {
+      attachRequestedFor = null;
+      return;
+    }
+    if (!id || !check || !hasAuthFailure()) return;
+    if (agentLogin.running || attachRequestedFor === id) return;
+    attachRequestedFor = id;
+    // Probe even without a report or an eligible login command: another pane,
+    // client or webview may already have started a login we can follow.
+    void attachAgentLogin(check)
+      .then(refreshAfterLogin)
+      .catch(() => {
+        // Login failures are rendered by AgentLoginPrompt from the shared record.
+      });
+  });
+
+  return {
+    get checkId() {
+      return checkId();
+    },
+    get canLogin() {
+      return canLogin();
+    },
+    get running() {
+      return agentLogin.running && agentLogin.checkId === checkId();
+    },
+    async start(): Promise<void> {
+      const check = checkId();
+      // Presentation is provider-local, but any running login owns the one
+      // shared record and prevents a new start for another provider.
+      if (!check || !canLogin() || agentLogin.running) return;
+      try {
+        refreshAfterLogin(await startAgentLogin(check));
+      } catch {
+        // Login failures are rendered by AgentLoginPrompt from the shared record.
+      }
+    },
+  };
+}
diff --git a/apps/staged/src/lib/features/sessions/sessionLogin.test.ts b/apps/staged/src/lib/features/sessions/sessionLogin.test.ts
new file mode 100644
index 000000000..a1a30ce17
--- /dev/null
+++ b/apps/staged/src/lib/features/sessions/sessionLogin.test.ts
@@ -0,0 +1,373 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import type { DoctorCheck, DoctorReport } from '../../api/commands';
+import type { Session } from '../../types';
+import type { AgentLoginOutcome } from '../doctor/agentLogin.svelte';
+
+type LoginSession = Pick<Session, 'provider' | 'status' | 'errorMessage'>;
+
+function session(overrides: Partial<LoginSession> = {}): LoginSession {
+  return {
+    provider: 'claude',
+    status: 'error',
+    errorMessage: 'Authentication required',
+    ...overrides,
+  };
+}
+
+function report(overrides: Partial<DoctorCheck> = {}): DoctorReport {
+  return {
+    checks: [
+      {
+        id: 'ai-agent-claude',
+        label: 'Claude Code',
+        status: 'warn',
+        message: 'Not authenticated',
+        fixUrl: null,
+        fixCommand: null,
+        fixType: null,
+        path: '/usr/local/bin/claude-agent-acp',
+        bridgePath: null,
+        rawOutput: null,
+        authStatus: 'notAuthenticated',
+        loginCommand: 'claude-agent-acp --cli auth login',
+        installedVersion: null,
+        latestVersion: null,
+        updateAvailable: null,
+        installSource: null,
+        selfUpdating: null,
+        main: null,
+        bridge: null,
+        ...overrides,
+      },
+    ],
+  };
+}
+
+function deferred<T>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
+let input: { active: boolean; sessionId: string; session: LoginSession | null };
+let doctorState: { report: DoctorReport | null; loading: boolean };
+let agentLogin: { checkId: string | null; running: boolean };
+let runChecks: ReturnType<typeof vi.fn<() => Promise<void>>>;
+let attachAgentLogin: ReturnType<typeof vi.fn<(id: string) => Promise<AgentLoginOutcome | null>>>;
+let startAgentLogin: ReturnType<typeof vi.fn<(id: string) => Promise<AgentLoginOutcome>>>;
+let effects: Array<() => void>;
+
+beforeEach(() => {
+  vi.resetModules();
+  input = { active: true, sessionId: 'session-1', session: session() };
+  doctorState = { report: null, loading: false };
+  agentLogin = { checkId: null, running: false };
+  runChecks = vi.fn().mockResolvedValue(undefined);
+  attachAgentLogin = vi.fn().mockResolvedValue(null);
+  startAgentLogin = vi.fn().mockResolvedValue('cancelled');
+  effects = [];
+  // Vitest has no Svelte plugin: capture effects and explicitly rerun them
+  // after input changes. Getters must read live data without rune transforms.
+  vi.stubGlobal('$effect', (effect: () => void) => effects.push(effect));
+  vi.doMock('../doctor/doctor.svelte', () => ({ doctorState, runChecks }));
+  vi.doMock('../doctor/agentLogin.svelte', () => ({
+    agentLogin,
+    attachAgentLogin,
+    startAgentLogin,
+  }));
+});
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+  vi.doUnmock('../doctor/doctor.svelte');
+  vi.doUnmock('../doctor/agentLogin.svelte');
+});
+
+async function create() {
+  const { createSessionLoginController } = await import('./sessionLogin.svelte');
+  return createSessionLoginController({
+    getActive: () => input.active,
+    getSessionId: () => input.sessionId,
+    getSession: () => input.session,
+  });
+}
+
+function runEffects() {
+  effects.forEach((effect) => effect());
+}
+
+describe('session login controller', () => {
+  it('reads provider, report and shared running changes through live getters', async () => {
+    const controller = await create();
+    expect(controller.checkId).toBe('ai-agent-claude');
+    expect(controller.canLogin).toBe(false);
+    expect(controller.running).toBe(false);
+
+    doctorState.report = report();
+    expect(controller.canLogin).toBe(true);
+    doctorState.report = report({ id: 'ai-agent-codex' });
+    expect(controller.canLogin).toBe(false);
+    input.session = session({ provider: 'codex' });
+    expect(controller.checkId).toBe('ai-agent-codex');
+    expect(controller.canLogin).toBe(true);
+
+    agentLogin.running = true;
+    agentLogin.checkId = 'ai-agent-claude';
+    expect(controller.running).toBe(false);
+    agentLogin.checkId = 'ai-agent-codex';
+    expect(controller.running).toBe(true);
+    agentLogin.running = false;
+    expect(controller.running).toBe(false);
+    doctorState.report = null;
+    expect(controller.canLogin).toBe(false);
+
+    input.session = session({ provider: null });
+    expect(controller.checkId).toBeNull();
+    input.session = null;
+    expect(controller.checkId).toBeNull();
+    expect(controller.canLogin).toBe(false);
+    expect(controller.running).toBe(false);
+    expect(runChecks).not.toHaveBeenCalled();
+    expect(attachAgentLogin).not.toHaveBeenCalled();
+  });
+
+  it.each([
+    { name: 'inactive pane', active: false },
+    { name: 'missing session ID', sessionId: '' },
+    { name: 'missing session', session: null },
+    { name: 'queued session', session: session({ status: 'queued' }) },
+    { name: 'running session', session: session({ status: 'running' }) },
+    { name: 'completed session', session: session({ status: 'completed' }) },
+    { name: 'unrelated error', session: session({ errorMessage: 'Connection refused' }) },
+    { name: 'missing error', session: session({ errorMessage: null }) },
+    {
+      name: 'unrelated cancellation',
+      session: session({ status: 'cancelled', errorMessage: 'User stopped the session' }),
+    },
+  ])(
+    'gates both effects for $name without consuming the attempt',
+    async ({ name: _, ...patch }) => {
+      Object.assign(input, patch);
+      await create();
+      runEffects();
+      expect(runChecks).not.toHaveBeenCalled();
+      expect(attachAgentLogin).not.toHaveBeenCalled();
+
+      input = { active: true, sessionId: 'session-1', session: session() };
+      runEffects();
+      expect(runChecks).toHaveBeenCalledTimes(1);
+      expect(attachAgentLogin).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+    }
+  );
+
+  it.each(['error', 'cancelled'] as const)(
+    'recovers an authentication failure in %s',
+    async (status) => {
+      input.session = session({ status });
+      await create();
+      runEffects();
+      runEffects();
+      expect(runChecks).toHaveBeenCalledTimes(1);
+      expect(attachAgentLogin).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+    }
+  );
+
+  it('can request a report without a provider, but cannot attach', async () => {
+    input.session = session({ provider: null });
+    await create();
+    runEffects();
+    expect(runChecks).toHaveBeenCalledTimes(1);
+    expect(attachAgentLogin).not.toHaveBeenCalled();
+  });
+
+  it.each(['loading', 'existing report'] as const)(
+    'defers the initial report for %s without deferring attach',
+    async (gate) => {
+      doctorState.loading = gate === 'loading';
+      doctorState.report = gate === 'existing report' ? report() : null;
+      await create();
+      runEffects();
+      expect(runChecks).not.toHaveBeenCalled();
+      expect(attachAgentLogin).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+
+      doctorState.loading = false;
+      doctorState.report = null;
+      runEffects();
+      expect(runChecks).toHaveBeenCalledTimes(1);
+      expect(attachAgentLogin).toHaveBeenCalledTimes(1);
+    }
+  );
+
+  it('does not spin or retry a failed scan on reopen, but scans a new session', async () => {
+    const scan = deferred<void>();
+    runChecks.mockImplementationOnce(async () => {
+      doctorState.loading = true;
+      await scan.promise;
+      // runChecks swallows scan failures, leaving no report.
+      doctorState.loading = false;
+    });
+    await create();
+    runEffects();
+    runEffects();
+    expect(runChecks).toHaveBeenCalledTimes(1);
+    scan.resolve(undefined);
+    await runChecks.mock.results[0].value;
+    runEffects();
+    runEffects();
+    input.active = false;
+    runEffects();
+    input.active = true;
+    runEffects();
+    expect(runChecks).toHaveBeenCalledTimes(1);
+    expect(attachAgentLogin).toHaveBeenCalledTimes(2);
+
+    input.sessionId = 'session-2';
+    runEffects();
+    expect(runChecks).toHaveBeenCalledTimes(2);
+  });
+
+  it.each([
+    { name: 'no matching check', overrides: { id: 'ai-agent-codex' } },
+    { name: 'unknown authentication', overrides: { authStatus: 'unknown' as const } },
+    { name: 'no login command', overrides: { loginCommand: null } },
+  ])('attaches independently of eligibility: $name', async ({ overrides }) => {
+    doctorState.report = report(overrides);
+    const controller = await create();
+    expect(controller.canLogin).toBe(false);
+    runEffects();
+    expect(attachAgentLogin).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+    expect(runChecks).not.toHaveBeenCalled();
+  });
+
+  it('probes once per open and again on reopening or switching sessions', async () => {
+    doctorState.report = report();
+    await create();
+    runEffects();
+    await Promise.resolve(); // Let the null attach result settle before another flush.
+    runEffects();
+    expect(attachAgentLogin).toHaveBeenCalledTimes(1);
+    input.active = false;
+    runEffects();
+    expect(attachAgentLogin).toHaveBeenCalledTimes(1);
+    input.active = true;
+    runEffects();
+    input.sessionId = 'session-2';
+    input.session = session({ provider: 'codex' });
+    runEffects();
+    runEffects();
+    expect(attachAgentLogin.mock.calls).toEqual([
+      ['ai-agent-claude'],
+      ['ai-agent-claude'],
+      ['ai-agent-codex'],
+    ]);
+  });
+
+  it('defers attach while any shared login runs without consuming the attempt', async () => {
+    doctorState.report = report();
+    agentLogin = { checkId: 'ai-agent-codex', running: true };
+    await create();
+    runEffects();
+    expect(attachAgentLogin).not.toHaveBeenCalled();
+    agentLogin.running = false;
+    runEffects();
+    runEffects();
+    expect(attachAgentLogin).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+  });
+
+  it('starts only with a provider and an eligible report', async () => {
+    const controller = await create();
+    await controller.start();
+    doctorState.report = report();
+    input.session = null;
+    await controller.start();
+    input.session = session({ provider: null });
+    await controller.start();
+    input.session = session();
+    doctorState.report = report({ loginCommand: null });
+    await controller.start();
+    expect(startAgentLogin).not.toHaveBeenCalled();
+    doctorState.report = report();
+    await controller.start();
+    expect(startAgentLogin).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+  });
+
+  it.each(['claude', 'codex'])(
+    'blocks starts globally while %s runs, but presents running per provider',
+    async (provider) => {
+      doctorState.report = report();
+      agentLogin = { checkId: `ai-agent-${provider}`, running: true };
+      const controller = await create();
+      expect(controller.canLogin).toBe(true);
+      expect(controller.running).toBe(provider === 'claude');
+      await controller.start();
+      expect(startAgentLogin).not.toHaveBeenCalled();
+      agentLogin.running = false;
+      await controller.start();
+      expect(startAgentLogin).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+    }
+  );
+
+  it('does not refresh when attach finds no login', async () => {
+    doctorState.report = report();
+    await create();
+    runEffects();
+    await Promise.resolve();
+    expect(attachAgentLogin).toHaveBeenCalledTimes(1);
+    expect(runChecks).not.toHaveBeenCalled();
+  });
+
+  describe.each(['start', 'attach'] as const)('%s completion', (entry) => {
+    it.each(['completed', 'cancelled'] as const)(
+      'refreshes only for completed: %s',
+      async (outcome) => {
+        doctorState.report = report();
+        const controller = await create();
+        const action = entry === 'start' ? startAgentLogin : attachAgentLogin;
+        action.mockResolvedValueOnce(outcome);
+        await (entry === 'start' ? controller.start() : runEffects());
+        expect(action).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+        expect(runChecks).toHaveBeenCalledTimes(outcome === 'completed' ? 1 : 0);
+      }
+    );
+
+    it('swallows rejection without refreshing', async () => {
+      doctorState.report = report();
+      const controller = await create();
+      const action = entry === 'start' ? startAgentLogin : attachAgentLogin;
+      action.mockRejectedValueOnce(new Error('Login failed'));
+      await (entry === 'start' ? controller.start() : runEffects());
+      expect(action).toHaveBeenCalledTimes(1);
+      expect(runChecks).not.toHaveBeenCalled();
+    });
+
+    it.each(['close', 'switch session'] as const)(
+      'refreshes the global report even after %s while completion was pending',
+      async (change) => {
+        doctorState.report = report();
+        const controller = await create();
+        const pending = deferred<AgentLoginOutcome>();
+        const action = entry === 'start' ? startAgentLogin : attachAgentLogin;
+        action.mockReturnValueOnce(pending.promise);
+        const settled = entry === 'start' ? controller.start() : runEffects();
+        agentLogin.running = true;
+        agentLogin.checkId = 'ai-agent-claude';
+
+        if (change === 'close') input.active = false;
+        else {
+          input.sessionId = 'session-2';
+          input.session = session({ provider: 'codex' });
+        }
+        runEffects();
+        expect(agentLogin.running).toBe(true);
+        expect(runChecks).not.toHaveBeenCalled();
+        pending.resolve('completed');
+        await pending.promise;
+        await settled;
+        expect(runChecks).toHaveBeenCalledTimes(1);
+        expect(action).toHaveBeenCalledExactlyOnceWith('ai-agent-claude');
+      }
+    );
+  });
+});
diff --git a/apps/staged/src/lib/transport.test.ts b/apps/staged/src/lib/transport.test.ts
index b71b384dd..9c959572a 100644
--- a/apps/staged/src/lib/transport.test.ts
+++ b/apps/staged/src/lib/transport.test.ts
@@ -287,28 +287,192 @@ describe('web transport', () => {
 
     unlisten();
   });
+
+  it('announces establishment again when the server reports dropped events', async () => {
+    vi.useFakeTimers();
+    vi.stubGlobal('WebSocket', MockWebSocket);
+    const { listenToEvent } = await import('./transport');
+    const callback = vi.fn();
+    const onEstablished = vi.fn();
+    const unlisten = listenToEvent('doctor-login-output', callback, { onEstablished });
+    await vi.waitFor(() => expect(sockets).toHaveLength(1));
+
+    sockets[0].open();
+    expect(onEstablished).toHaveBeenCalledTimes(1);
+
+    // Events the server shed under load are as gone as ones emitted while the
+    // socket was down, so a consumer that catches up on `onEstablished` must
+    // hear about this gap the same way — there is no reconnect to prompt it.
+    sockets[0].emit({ event: 'transport:event-gap', payload: null });
+
+    expect(onEstablished).toHaveBeenCalledTimes(2);
+    expect(callback).not.toHaveBeenCalled();
+    expect(sockets).toHaveLength(1);
+
+    unlisten();
+  });
+
+  it('reports WebSocket setup failure to every listener waiting on the same connection', async () => {
+    vi.doMock('./services/prPollingService', () => ({
+      getPrPollClientId: () => {
+        throw new Error('client id unavailable');
+      },
+      replayPrPollInterestHints: vi.fn().mockResolvedValue(undefined),
+    }));
+    vi.stubGlobal('WebSocket', MockWebSocket);
+    const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
+
+    const { listenToEvent } = await import('./transport');
+    const firstFailed = vi.fn();
+    const secondFailed = vi.fn();
+    const firstEstablished = vi.fn();
+    const secondEstablished = vi.fn();
+
+    const unlistenFirst = listenToEvent('doctor-login-output', vi.fn(), {
+      onEstablished: firstEstablished,
+      onRegistrationFailed: firstFailed,
+    });
+    const unlistenSecond = listenToEvent('session-background-hold', vi.fn(), {
+      onEstablished: secondEstablished,
+      onRegistrationFailed: secondFailed,
+    });
+
+    await vi.waitFor(() => expect(firstFailed).toHaveBeenCalledTimes(1));
+    expect(secondFailed).toHaveBeenCalledTimes(1);
+    expect(firstFailed.mock.calls[0][0]).toBeInstanceOf(Error);
+    expect(secondFailed.mock.calls[0][0]).toBe(firstFailed.mock.calls[0][0]);
+    expect(firstEstablished).not.toHaveBeenCalled();
+    expect(secondEstablished).not.toHaveBeenCalled();
+    expect(sockets).toHaveLength(0);
+
+    unlistenFirst();
+    unlistenSecond();
+    consoleError.mockRestore();
+  });
+
+  it('continues after WebSocket setup failure and does not report listeners that left', async () => {
+    let fail = true;
+    vi.doMock('./services/prPollingService', () => ({
+      getPrPollClientId: () => {
+        if (fail) throw new Error('client id unavailable');
+        return 'web-client-1';
+      },
+      replayPrPollInterestHints: vi.fn().mockResolvedValue(undefined),
+    }));
+    vi.useFakeTimers();
+    vi.stubGlobal('WebSocket', MockWebSocket);
+    const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
+
+    const { listenToEvent } = await import('./transport');
+    const staleFailed = vi.fn();
+    const currentFailed = vi.fn();
+    const onEstablished = vi.fn();
+
+    const unlistenStale = listenToEvent('doctor-login-output', vi.fn(), {
+      onRegistrationFailed: staleFailed,
+    });
+    unlistenStale();
+    const unlistenCurrent = listenToEvent('doctor-login-output', vi.fn(), {
+      onEstablished,
+      onRegistrationFailed: currentFailed,
+    });
+
+    await vi.waitFor(() => expect(currentFailed).toHaveBeenCalledTimes(1));
+    expect(staleFailed).not.toHaveBeenCalled();
+    expect(sockets).toHaveLength(0);
+
+    fail = false;
+    const onSecondEstablished = vi.fn();
+    const unlistenSecond = listenToEvent('session-background-hold', vi.fn(), {
+      onEstablished: onSecondEstablished,
+    });
+    await vi.waitFor(() => expect(sockets).toHaveLength(1));
+    sockets[0].open();
+    await vi.waitFor(() => expect(onSecondEstablished).toHaveBeenCalledTimes(1));
+    expect(onEstablished).not.toHaveBeenCalled();
+
+    unlistenCurrent();
+    unlistenSecond();
+    consoleError.mockRestore();
+  });
+
+  it('keeps listeners without onRegistrationFailed registered across a WebSocket setup failure', async () => {
+    let fail = true;
+    vi.doMock('./services/prPollingService', () => ({
+      getPrPollClientId: () => {
+        if (fail) throw new Error('client id unavailable');
+        return 'web-client-1';
+      },
+      replayPrPollInterestHints: vi.fn().mockResolvedValue(undefined),
+    }));
+    vi.useFakeTimers();
+    vi.stubGlobal('WebSocket', MockWebSocket);
+    const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
+
+    const { listenToEvent } = await import('./transport');
+    const plainCallback = vi.fn();
+    const loginCallback = vi.fn();
+    const loginFailed = vi.fn();
+
+    // A plain listener (session status, change feed, ...) has no way to be
+    // told about the failure and nothing to do about it, so it must survive
+    // for the next attempt. The login listener asked to hear about it and is
+    // dropped: its owner has already given the run up.
+    const unlistenPlain = listenToEvent('session-status-changed', plainCallback);
+    const unlistenLogin = listenToEvent('doctor-login-output', loginCallback, {
+      onRegistrationFailed: loginFailed,
+    });
+
+    await vi.waitFor(() => expect(loginFailed).toHaveBeenCalledTimes(1));
+    expect(sockets).toHaveLength(0);
+
+    fail = false;
+    const onEstablished = vi.fn();
+    const unlistenNext = listenToEvent('session-background-hold', vi.fn(), { onEstablished });
+    await vi.waitFor(() => expect(sockets).toHaveLength(1));
+    sockets[0].open();
+    await vi.waitFor(() => expect(onEstablished).toHaveBeenCalledTimes(1));
+
+    sockets[0].emit({ event: 'session-status-changed', payload: { sessionId: 's1' } });
+    expect(plainCallback).toHaveBeenCalledWith({ sessionId: 's1' });
+
+    sockets[0].emit({ event: 'doctor-login-output', payload: { line: 'late' } });
+    expect(loginCallback).not.toHaveBeenCalled();
+
+    unlistenPlain();
+    unlistenLogin();
+    unlistenNext();
+    expect(sockets[0].closed).toBe(true);
+    consoleError.mockRestore();
+  });
 });
 
 describe('tauri listener establishment', () => {
   let resolveListen: ((unlisten: () => void) => void) | undefined;
+  let rejectListen: ((error: unknown) => void) | undefined;
   let listen: ReturnType<typeof vi.fn>;
   let tauriUnlisten: ReturnType<typeof vi.fn<() => void>>;
+  let consoleError: MockInstance<typeof console.error>;
 
   beforeEach(() => {
     vi.resetModules();
     vi.stubGlobal('__TAURI__', {});
     resolveListen = undefined;
+    rejectListen = undefined;
     tauriUnlisten = vi.fn<() => void>();
     listen = vi.fn(
       () =>
-        new Promise<() => void>((resolve) => {
+        new Promise<() => void>((resolve, reject) => {
           resolveListen = resolve;
+          rejectListen = reject;
         })
     );
     vi.doMock('@tauri-apps/api/event', () => ({ listen }));
+    consoleError = vi.spyOn(console, 'error').mockImplementation(() => {});
   });
 
   afterEach(() => {
+    consoleError.mockRestore();
     vi.doUnmock('@tauri-apps/api/event');
     vi.unstubAllGlobals();
   });
@@ -345,6 +509,38 @@ describe('tauri listener establishment', () => {
     await vi.waitFor(() => expect(tauriUnlisten).toHaveBeenCalledTimes(1));
     expect(onEstablished).not.toHaveBeenCalled();
   });
+
+  it('reports a registration that fails instead of leaving the consumer waiting', async () => {
+    const { listenToEvent } = await import('./transport');
+    const onEstablished = vi.fn();
+    const onRegistrationFailed = vi.fn();
+    listenToEvent('doctor-login-output', vi.fn(), { onEstablished, onRegistrationFailed });
+
+    await vi.waitFor(() => expect(listen).toHaveBeenCalledTimes(1));
+    const failure = new Error('ipc down');
+    rejectListen?.(failure);
+
+    // A consumer that waits for `onEstablished` before acting has no other way
+    // to learn that the wait will never end.
+    await vi.waitFor(() => expect(onRegistrationFailed).toHaveBeenCalledWith(failure));
+    expect(onEstablished).not.toHaveBeenCalled();
+    expect(consoleError).toHaveBeenCalledTimes(1);
+  });
+
+  it('does not report a failed registration the consumer already walked away from', async () => {
+    const { listenToEvent } = await import('./transport');
+    const onRegistrationFailed = vi.fn();
+    const unlisten = listenToEvent('doctor-login-output', vi.fn(), { onRegistrationFailed });
+
+    await vi.waitFor(() => expect(listen).toHaveBeenCalledTimes(1));
+    unlisten();
+    rejectListen?.(new Error('ipc down'));
+
+    // Still logged — the failure is real — but the hook stays silent, matching
+    // `onEstablished` after an early unlisten.
+    await vi.waitFor(() => expect(consoleError).toHaveBeenCalledTimes(1));
+    expect(onRegistrationFailed).not.toHaveBeenCalled();
+  });
 });
 
 describe('tauri window label', () => {
diff --git a/apps/staged/src/lib/transport.ts b/apps/staged/src/lib/transport.ts
index 7ea63cad4..ddef2b2b7 100644
--- a/apps/staged/src/lib/transport.ts
+++ b/apps/staged/src/lib/transport.ts
@@ -136,10 +136,24 @@ export interface ListenOptions {
    * returned unlisten. Fires once in Tauri mode (its in-process bus loses
    * nothing after registration) and once per web-socket connect, including
    * every reconnect: events emitted while the socket was down are gone for
-   * good. Not called when the unlisten precedes establishment, or when Tauri
-   * registration fails.
+   * good. Fires again, too, each time the server reports it dropped events for
+   * this socket (a client that lags the bounded broadcast channel is shed
+   * from, not queued for) — the same loss without a reconnect to mark it. Not
+   * called when the unlisten precedes establishment, or when registration
+   * fails — see `onRegistrationFailed` for that.
    */
   onEstablished?: () => void;
+  /**
+   * Called when registration failed outright, so this listener will never go
+   * live and `onEstablished` will never fire. Lets a consumer that waits on
+   * `onEstablished` before acting (rather than merely tolerating a missed
+   * event) give up instead of waiting forever. In Tauri mode this is the
+   * `listen()` roundtrip rejecting; in web mode it is the socket failing to be
+   * set up at all — a connect attempt that merely fails is retried and is not
+   * reported. Always asynchronous, like `onEstablished`. Also logged, so a
+   * consumer only needs this to react, not to make the failure visible.
+   */
+  onRegistrationFailed?: (error: unknown) => void;
 }
 
 /**
@@ -162,7 +176,7 @@ export function listenToEvent<T>(
   opts?: ListenOptions
 ): UnlistenFn {
   if (!isTauri) {
-    return webSocketListen<T>(event, callback, opts?.onEstablished);
+    return webSocketListen<T>(event, callback, opts);
   }
 
   let cancelled = false;
@@ -179,6 +193,7 @@ export function listenToEvent<T>(
     opts?.onEstablished?.();
   })().catch((e) => {
     console.error(`[transport] Failed to register listener for event "${event}":`, e);
+    if (!cancelled) opts?.onRegistrationFailed?.(e);
   });
 
   return () => {
@@ -197,7 +212,7 @@ export function listenToEvent<T>(
  */
 export function listenToWindowEvent<T>(event: string, callback: (payload: T) => void): UnlistenFn {
   if (!isTauri) {
-    return webSocketListen<T>(event, callback);
+    return webSocketListen<T>(event, callback, undefined);
   }
 
   let cancelled = false;
@@ -226,6 +241,7 @@ interface WebSocketListener {
   event: string;
   callback: (payload: unknown) => void;
   onEstablished?: () => void;
+  onRegistrationFailed?: (error: unknown) => void;
 }
 
 const WEB_SOCKET_HEARTBEAT_MS = 30_000;
@@ -332,6 +348,24 @@ function notifyListenersEstablished(): void {
   }
 }
 
+/**
+ * Report a failed socket set-up to the listeners that asked to hear about it,
+ * and drop only those: a consumer told its registration failed has given up on
+ * it, and keeping the entry would let a later connect deliver events to a
+ * listener whose owner believes it dead. Every other listener stays registered
+ * and is picked up by the next `ensureWebSocket`, exactly as before this
+ * attempt — in Tauri mode a failed `listen()` costs only that one listener, and
+ * the shared socket should not widen that to everyone who happened to be
+ * waiting on the same connection.
+ */
+function notifyListenersRegistrationFailed(error: unknown): void {
+  const failed = wsListeners.filter((l) => l.onRegistrationFailed);
+  wsListeners = wsListeners.filter((l) => !l.onRegistrationFailed);
+  for (const listener of failed) {
+    listener.onRegistrationFailed?.(error);
+  }
+}
+
 async function ensureWebSocket(): Promise<void> {
   if (ws && (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING)) {
     return;
@@ -339,20 +373,25 @@ async function ensureWebSocket(): Promise<void> {
   if (wsConnecting) return;
 
   wsConnecting = true;
-  const url = await getWsUrl();
-  if (wsListeners.length === 0) {
+  let socket: WebSocket;
+  let isReconnect: boolean;
+  try {
+    const url = await getWsUrl();
+    if (wsListeners.length === 0) return;
+
+    isReconnect = wsHadSocket;
+    wsHadSocket = true;
+
+    socket = new WebSocket(url);
+    ws = socket;
+  } catch (e) {
+    notifyListenersRegistrationFailed(e);
+    throw e;
+  } finally {
     wsConnecting = false;
-    return;
   }
 
-  const isReconnect = wsHadSocket;
-  wsHadSocket = true;
-
-  const socket = new WebSocket(url);
-  ws = socket;
-
   socket.onopen = () => {
-    wsConnecting = false;
     startHeartbeat();
     replayCurrentPrPollInterestHints();
     if (isReconnect) recoverAfterEventGap();
@@ -369,6 +408,10 @@ async function ensureWebSocket(): Promise<void> {
       const data = JSON.parse(messageEvent.data) as { event: string; payload: unknown };
       if (data.event === WEB_SOCKET_EVENT_GAP) {
         recoverAfterEventGap();
+        // The events the server shed are as gone as ones emitted during a
+        // reconnect, and a listener that pairs this stream with a snapshot
+        // has the same catching up to do — see `ListenOptions.onEstablished`.
+        notifyListenersEstablished();
         return;
       }
       for (const listener of wsListeners) {
@@ -382,7 +425,6 @@ async function ensureWebSocket(): Promise<void> {
   };
 
   socket.onclose = () => {
-    wsConnecting = false;
     stopHeartbeat();
     if (ws === socket) {
       ws = null;
@@ -399,25 +441,32 @@ async function ensureWebSocket(): Promise<void> {
     }
   };
 
-  socket.onerror = () => {
-    wsConnecting = false;
-  };
+  socket.onerror = () => {};
 }
 
 function webSocketListen<T>(
   event: string,
   callback: (payload: T) => void,
-  onEstablished?: () => void
+  opts: ListenOptions | undefined
 ): UnlistenFn {
+  const onEstablished = opts?.onEstablished;
   const listener: WebSocketListener = {
     event,
     callback: callback as (payload: unknown) => void,
     onEstablished,
+    onRegistrationFailed: opts?.onRegistrationFailed,
   };
 
   wsListeners.push(listener);
   void ensureWebSocket().catch((e) => {
     console.error('[transport] Failed to connect WebSocket:', e);
+    // Only the socket set-up itself can reject here — a socket that opens and
+    // then drops is retried from `onclose`, never surfaced this way — so this is
+    // the one web-mode case where the listener provably never goes live. The
+    // setup attempt notifies every still-registered listener that opted into
+    // `onRegistrationFailed` before rejecting, including any that joined while
+    // the attempt was in flight; listeners without the hook stay registered for
+    // the next attempt.
   });
 
   // A socket that is still connecting will notify this listener along with the
diff --git a/crates/acp-client/src/driver.rs b/crates/acp-client/src/driver.rs
index d76263ea4..5136b52a6 100644
--- a/crates/acp-client/src/driver.rs
+++ b/crates/acp-client/src/driver.rs
@@ -21,9 +21,9 @@ use agent_client_protocol::{
         v1::{
             AgentCapabilities, AgentNotification, AuthMethod, AuthenticateRequest,
             CancelNotification, ClientCapabilities, ContentBlock as AcpContentBlock, ContentChunk,
-            ExtNotification, ImageContent, Implementation, InitializeRequest, InitializeResponse,
-            LoadSessionRequest, McpCapabilities, McpServer, Meta, NewSessionRequest,
-            PermissionOption as SchemaPermissionOption, PermissionOptionId,
+            ErrorCode, ExtNotification, ImageContent, Implementation, InitializeRequest,
+            InitializeResponse, LoadSessionRequest, McpCapabilities, McpServer, Meta,
+            NewSessionRequest, PermissionOption as SchemaPermissionOption, PermissionOptionId,
             PermissionOptionKind as SchemaPermissionOptionKind, PromptRequest, PromptResponse,
             RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
             SelectedPermissionOutcome, SessionConfigKind, SessionConfigOption,
@@ -33,7 +33,7 @@ use agent_client_protocol::{
         },
         ProtocolVersion,
     },
-    Agent, ByteStreams, Client, ConnectionTo, JsonRpcMessage, UntypedMessage,
+    Agent, ByteStreams, Client, ConnectionTo, JsonRpcMessage, JsonRpcRequest, UntypedMessage,
 };
 use async_trait::async_trait;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
@@ -319,6 +319,193 @@ pub struct AcpSessionConfigOptionSelection {
     pub value_id: String,
 }
 
+/// Provider-specific opt-in to authenticate with a known ACP method.
+///
+/// Generic ACP setup never derives this from advertised method IDs, display
+/// names, or ordering. Callers may populate it only after their integration has
+/// explicit knowledge that the method is safe and currently usable.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AcpAuthenticationSelection {
+    pub method_id: String,
+}
+
+impl AcpAuthenticationSelection {
+    pub fn new(method_id: impl Into<String>) -> Self {
+        Self {
+            method_id: method_id.into(),
+        }
+    }
+}
+
+/// What kind of credential an advertised ACP auth method needs, which is what
+/// decides whether this client can act on it at all.
+///
+/// Deliberately not `Serialize`: nothing is threaded to the session layer yet,
+/// and a derive would advertise a wire shape no consumer has agreed to.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AcpAuthenticationMethodCategory {
+    AgentManaged,
+    EnvironmentBacked,
+    Terminal,
+    Unsupported,
+}
+
+impl AcpAuthenticationMethodCategory {
+    fn label(self) -> &'static str {
+        match self {
+            Self::AgentManaged => "agent-managed",
+            Self::EnvironmentBacked => "environment-backed",
+            Self::Terminal => "terminal",
+            Self::Unsupported => "unsupported",
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AcpAuthenticationMethod {
+    pub id: String,
+    pub display_name: String,
+    pub description: Option<String>,
+    pub category: AcpAuthenticationMethodCategory,
+    pub can_handle: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AcpAuthenticationRequired {
+    pub methods: Vec<AcpAuthenticationMethod>,
+    pub attempted_method_id: Option<String>,
+    /// What the agent itself said on its `-32000`, beyond the stock text; see
+    /// [`Self::with_agent_detail`].
+    pub agent_detail: Option<String>,
+}
+
+impl AcpAuthenticationRequired {
+    fn from_auth_methods(auth_methods: &[AuthMethod]) -> Self {
+        Self {
+            methods: auth_methods.iter().map(auth_method_details).collect(),
+            attempted_method_id: None,
+            agent_detail: None,
+        }
+    }
+
+    fn after_authentication_attempt(mut self, method_id: impl Into<String>) -> Self {
+        self.attempted_method_id = Some(method_id.into());
+        self
+    }
+
+    /// Keep whatever the agent added to the `-32000` that made this error final.
+    ///
+    /// Most bridges send the schema's bare `auth_required` — stock "Authentication
+    /// required" and no data — and for those [`Self::describe`] is unchanged. pi-acp
+    /// answers `session/new` with "Configure an API key or log in with an OAuth
+    /// provider.", which names the remedy, so a `message` or string `data` that says
+    /// more than the stock text is appended to the sentence the user reads.
+    /// Replaces any earlier detail: after an authenticate-and-retry the error that
+    /// matters is the second one.
+    fn with_agent_detail(mut self, error: &agent_client_protocol::Error) -> Self {
+        self.agent_detail = agent_auth_required_detail(error);
+        self
+    }
+
+    /// The sentence the user reads. This string lands in `session.errorMessage`
+    /// and is rendered verbatim in Staged's session alert, so the method
+    /// inventory behind the decision is not part of it. Pure: the call site
+    /// that decides the error is final also calls [`Self::log_methods`], so the
+    /// inventory is recorded exactly once however often this is formatted.
+    ///
+    /// Always opens with "ACP authentication is required", whatever the agent
+    /// added: the frontend recognises the error, and offers Log in, by that word.
+    fn describe(&self, operation: &str) -> String {
+        let mut text = match self.attempted_method_id.as_deref() {
+            Some(method_id) => format!(
+                "ACP authentication is required to {operation}. Signing in with '{method_id}' did not clear it — sign this agent in again, then retry."
+            ),
+            None => format!(
+                "ACP authentication is required to {operation}. Sign this agent in, then retry."
+            ),
+        };
+        if let Some(detail) = &self.agent_detail {
+            text.push_str(" The agent said: ");
+            text.push_str(detail);
+        }
+        text
+    }
+
+    /// Record the full advertised-method inventory, which is the diagnostic that
+    /// explains why the driver refused to pick one for the user.
+    fn log_methods(&self, context: &str) {
+        log::debug!(
+            "ACP authentication {context}; not retrying automatically without an explicit supported authentication method. Advertised methods: {}",
+            self.method_inventory()
+        );
+    }
+
+    fn method_inventory(&self) -> String {
+        if self.methods.is_empty() {
+            return "no advertised authentication methods".to_string();
+        }
+        self.methods
+            .iter()
+            .map(|method| {
+                let support = if method.can_handle {
+                    "client-supported"
+                } else {
+                    "unsupported by this client"
+                };
+                let description = method
+                    .description
+                    .as_deref()
+                    .filter(|description| !description.trim().is_empty())
+                    .map(|description| format!(", description: {description}"))
+                    .unwrap_or_default();
+                format!(
+                    "{} (id: {}, category: {}, {support}{description})",
+                    method.display_name,
+                    method.id,
+                    method.category.label(),
+                )
+            })
+            .collect::<Vec<_>>()
+            .join("; ")
+    }
+}
+
+fn auth_method_details(method: &AuthMethod) -> AcpAuthenticationMethod {
+    let category = auth_method_category(method);
+    AcpAuthenticationMethod {
+        id: method.id().to_string(),
+        display_name: method.name().to_string(),
+        description: method.description().map(str::to_string),
+        category,
+        can_handle: auth_method_can_be_handled(category),
+    }
+}
+
+fn auth_method_category(method: &AuthMethod) -> AcpAuthenticationMethodCategory {
+    match method {
+        AuthMethod::Agent(_) => AcpAuthenticationMethodCategory::AgentManaged,
+        AuthMethod::EnvVar(_) => AcpAuthenticationMethodCategory::EnvironmentBacked,
+        AuthMethod::Terminal(_) => AcpAuthenticationMethodCategory::Terminal,
+        _ => AcpAuthenticationMethodCategory::Unsupported,
+    }
+}
+
+fn auth_method_can_be_handled(category: AcpAuthenticationMethodCategory) -> bool {
+    match category {
+        // Staged can pass an explicitly selected agent-managed method ID to
+        // `authenticate`, but the generic driver still must not choose one by
+        // guessing from provider-defined IDs, names, or list order.
+        AcpAuthenticationMethodCategory::AgentManaged => true,
+        // Environment-backed methods need provider-specific confirmation that
+        // credentials are available in the agent process environment.
+        AcpAuthenticationMethodCategory::EnvironmentBacked => false,
+        // Terminal authentication requires a complete interactive terminal
+        // flow. Until that exists, never send it through `authenticate`.
+        AcpAuthenticationMethodCategory::Terminal => false,
+        AcpAuthenticationMethodCategory::Unsupported => false,
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ReplayBoundary {
     pub role: String,
@@ -835,6 +1022,7 @@ pub trait AgentDriver {
         cancel_token: &CancellationToken,
         agent_session_id: Option<&str>,
         config_options: &[AcpSessionConfigOptionSelection],
+        auth_selection: Option<&AcpAuthenticationSelection>,
     ) -> Result<AgentRunOutcome, String>;
 }
 
@@ -1407,6 +1595,7 @@ impl AcpDriver {
         cancel_token: &CancellationToken,
         agent_session_id: Option<&str>,
         config_options: &[AcpSessionConfigOptionSelection],
+        auth_selection: Option<&AcpAuthenticationSelection>,
     ) -> Result<SessionConnection, String> {
         let spawn_working_dir = resolve_spawn_working_dir(working_dir, self.is_remote);
         let acp_working_dir = resolve_acp_working_dir(
@@ -1677,6 +1866,7 @@ impl AcpDriver {
         let our_session_id = session_id.to_string();
         let acp_session_id = agent_session_id.map(str::to_string);
         let config_options = config_options.to_vec();
+        let auth_selection = auth_selection.cloned();
         let mcp_servers = self.mcp_servers.clone();
         let agent_label = self.agent_label.clone();
         let is_remote = self.is_remote;
@@ -1766,6 +1956,7 @@ impl AcpDriver {
                         &mcp_servers,
                         &agent_label,
                         &cancel_token,
+                        auth_selection.as_ref(),
                         background_hold.as_ref(),
                         background_hold_observer.as_ref(),
                         &child_exited,
@@ -1838,6 +2029,7 @@ impl AgentDriver for AcpDriver {
         cancel_token: &CancellationToken,
         agent_session_id: Option<&str>,
         config_options: &[AcpSessionConfigOptionSelection],
+        auth_selection: Option<&AcpAuthenticationSelection>,
     ) -> Result<AgentRunOutcome, String> {
         let mut connection = self
             .connect(
@@ -1848,6 +2040,7 @@ impl AgentDriver for AcpDriver {
                 cancel_token,
                 agent_session_id,
                 config_options,
+                auth_selection,
             )
             .await?;
         let settled_rx = connection.take_settled_receiver();
@@ -4732,6 +4925,7 @@ async fn run_acp_session(
     mcp_servers: &[McpServer],
     agent_label: &str,
     cancel_token: &CancellationToken,
+    auth_selection: Option<&AcpAuthenticationSelection>,
     background_hold: Option<&BackgroundHoldConfig>,
     background_hold_observer: Option<&BackgroundHoldObserver>,
     child_exited: &CancellationToken,
@@ -4752,6 +4946,7 @@ async fn run_acp_session(
             config_options,
             mcp_servers,
             agent_label,
+            auth_selection,
         }),
     );
     let setup = tokio::select! {
@@ -5026,7 +5221,7 @@ async fn run_prompt_turn(
 
     let prompt_response = tokio::select! {
         result = &mut prompt_task => {
-            result.map_err(|e| format!("Prompt failed: {e:?}"))?
+            result.map_err(|error| describe_prompt_error(error, &setup.auth_methods))?
         }
         _ = cancel_token.cancelled() => {
             handler.cancel_pending_permissions();
@@ -5055,6 +5250,66 @@ async fn run_prompt_turn(
     ))
 }
 
+/// The text a failed `session/prompt` surfaces to the user.
+///
+/// amp, claude, and goose with a native provider accept `session/new` while
+/// signed out and raise ACP `auth_required` only here, so this is the
+/// `session/prompt` counterpart of the `auth_required` arm in
+/// [`send_session_setup_request`]: the same one sentence from
+/// [`AcpAuthenticationRequired::describe`] instead of a nested Debug dump, and
+/// the same inventory in the log. There is no authenticate-and-retry: a prompt
+/// may already have streamed output, and the user's remedy is the same sign-in
+/// either way. Every other error keeps its Debug rendering.
+fn describe_prompt_error(
+    error: agent_client_protocol::Error,
+    auth_methods: &[AuthMethod],
+) -> String {
+    if error.code == ErrorCode::AuthRequired {
+        let operation = "run the prompt";
+        let required =
+            AcpAuthenticationRequired::from_auth_methods(auth_methods).with_agent_detail(&error);
+        required.log_methods(&format!("required to {operation}"));
+        return required.describe(operation);
+    }
+    format!("Prompt failed: {error:?}")
+}
+
+/// What an agent said on a `-32000` beyond the schema's stock text, if anything.
+///
+/// `message` and a string `data` are both kept, once each and in that order,
+/// skipping either when it is empty or just the stock "Authentication required"
+/// (that comparison ignores case and a trailing period). Each part is made to
+/// end a sentence and internal whitespace is collapsed, so two parts read as
+/// two sentences and a multi-line `data` does not break the alert's one
+/// paragraph. Structured `data` is not rendered; it was never meant for a
+/// person to read.
+fn agent_auth_required_detail(error: &agent_client_protocol::Error) -> Option<String> {
+    let stock = ErrorCode::AuthRequired.to_string();
+    let is_stock = |text: &str| text.trim_end_matches('.').eq_ignore_ascii_case(&stock);
+    let mut parts: Vec<String> = Vec::new();
+    for candidate in [
+        Some(error.message.as_str()),
+        error.data.as_ref().and_then(serde_json::Value::as_str),
+    ]
+    .into_iter()
+    .flatten()
+    {
+        let text = candidate.split_whitespace().collect::<Vec<_>>().join(" ");
+        if text.is_empty() || is_stock(&text) {
+            continue;
+        }
+        let sentence = if text.ends_with(['.', '!', '?']) {
+            text
+        } else {
+            format!("{text}.")
+        };
+        if !parts.contains(&sentence) {
+            parts.push(sentence);
+        }
+    }
+    (!parts.is_empty()).then(|| parts.join(" "))
+}
+
 /// Whether the agent advertises support for the transport an MCP server needs.
 /// Stdio is always supported per the ACP spec.
 fn mcp_server_transport_supported(server: &McpServer, caps: &McpCapabilities) -> bool {
@@ -5482,6 +5737,10 @@ struct AcpSessionSetup {
     /// How this connection tracks background tasks, negotiated at initialize
     /// (see [`task_tracking_mode_from_initialize`]).
     task_tracking_mode: TaskTrackingMode,
+    /// The authentication methods the agent advertised at initialize, kept so
+    /// an `auth_required` raised at `session/prompt` can log the same
+    /// inventory the setup path logs (see [`describe_prompt_error`]).
+    auth_methods: Vec<AuthMethod>,
 }
 
 struct AcpSessionSetupContext<'a> {
@@ -5497,6 +5756,7 @@ struct AcpSessionSetupContext<'a> {
     config_options: &'a [AcpSessionConfigOptionSelection],
     mcp_servers: &'a [McpServer],
     agent_label: &'a str,
+    auth_selection: Option<&'a AcpAuthenticationSelection>,
 }
 
 /// `_meta` for `session/new` and `session/load` asking the Claude bridge to
@@ -5557,6 +5817,7 @@ async fn setup_acp_session(context: AcpSessionSetupContext<'_>) -> Result<AcpSes
         config_options,
         mcp_servers,
         agent_label,
+        auth_selection,
     } = context;
     let writer = &handler.writer;
 
@@ -5588,7 +5849,9 @@ async fn setup_acp_session(context: AcpSessionSetupContext<'_>) -> Result<AcpSes
         ));
     }
 
-    authenticate_if_advertised(connection, &init_response.auth_methods).await?;
+    // Authentication methods are capabilities, not proof that authentication
+    // is currently required. Existing provider credentials (for example a
+    // persisted ChatGPT login) may already make session setup succeed.
 
     // Required servers must all have a supported transport, or the session
     // fails. Route the decision through mcp_server_transport_supported so the
@@ -5638,19 +5901,23 @@ async fn setup_acp_session(context: AcpSessionSetupContext<'_>) -> Result<AcpSes
                 "Resuming ACP session {existing_id} via load_session for session {our_session_id}"
             );
 
-            let load_response = connection
-                .send_request(
-                    LoadSessionRequest::new(existing_id.to_string(), working_dir.to_path_buf())
-                        .mcp_servers(mcp_servers.to_vec())
-                        // The Claude bridge recreates the session on load and
-                        // reads `_meta.claudeCode.emitRawSDKMessages` there
-                        // too, so resumed sessions keep the raw
-                        // background-task frames flowing.
-                        .meta(background_task_tracking_meta(task_tracking_mode)),
-                )
-                .block_task()
-                .await
-                .map_err(|e| format!("Failed to load ACP session: {e:?}"))?;
+            let load_request = || {
+                LoadSessionRequest::new(existing_id.to_string(), working_dir.to_path_buf())
+                    .mcp_servers(mcp_servers.to_vec())
+                    // The Claude bridge recreates the session on load and
+                    // reads `_meta.claudeCode.emitRawSDKMessages` there
+                    // too, so resumed sessions keep the raw
+                    // background-task frames flowing.
+                    .meta(background_task_tracking_meta(task_tracking_mode))
+            };
+            let load_response = send_session_setup_request(
+                connection,
+                load_request,
+                &init_response.auth_methods,
+                auth_selection,
+                "load ACP session",
+            )
+            .await?;
 
             if let Some(ref modes) = load_response.modes {
                 writer.on_model_state_update(modes).await;
@@ -5669,17 +5936,23 @@ async fn setup_acp_session(context: AcpSessionSetupContext<'_>) -> Result<AcpSes
                 agent_session_id: existing_id.to_string(),
                 agent_capabilities,
                 task_tracking_mode,
+                auth_methods: init_response.auth_methods.clone(),
             })
         }
         None => {
-            let new_session_request = NewSessionRequest::new(working_dir.to_path_buf())
-                .mcp_servers(mcp_servers.to_vec())
-                .meta(background_task_tracking_meta(task_tracking_mode));
-            let session_response = connection
-                .send_request(new_session_request)
-                .block_task()
-                .await
-                .map_err(|e| format!("Failed to create ACP session: {e:?}"))?;
+            let new_session_request = || {
+                NewSessionRequest::new(working_dir.to_path_buf())
+                    .mcp_servers(mcp_servers.to_vec())
+                    .meta(background_task_tracking_meta(task_tracking_mode))
+            };
+            let session_response = send_session_setup_request(
+                connection,
+                new_session_request,
+                &init_response.auth_methods,
+                auth_selection,
+                "create ACP session",
+            )
+            .await?;
 
             let new_id = session_response.session_id.to_string();
             store
@@ -5703,6 +5976,7 @@ async fn setup_acp_session(context: AcpSessionSetupContext<'_>) -> Result<AcpSes
                 agent_session_id: new_id,
                 agent_capabilities,
                 task_tracking_mode,
+                auth_methods: init_response.auth_methods.clone(),
             })
         }
     }
@@ -5751,16 +6025,84 @@ fn describe_auth_methods(auth_methods: &[AuthMethod]) -> String {
         .join(", ")
 }
 
-async fn authenticate_if_advertised(
+async fn send_session_setup_request<Req, MakeRequest>(
     connection: &ConnectionTo<Agent>,
+    make_request: MakeRequest,
     auth_methods: &[AuthMethod],
+    auth_selection: Option<&AcpAuthenticationSelection>,
+    operation: &str,
+) -> Result<Req::Response, String>
+where
+    Req: JsonRpcRequest,
+    MakeRequest: Fn() -> Req,
+{
+    match connection.send_request(make_request()).block_task().await {
+        Ok(response) => Ok(response),
+        Err(error) if error.code == ErrorCode::AuthRequired => {
+            let required = AcpAuthenticationRequired::from_auth_methods(auth_methods)
+                .with_agent_detail(&error);
+            let Some(selection) = auth_selection else {
+                required.log_methods(&format!("required to {operation}"));
+                return Err(required.describe(operation));
+            };
+
+            authenticate_with_explicit_method(connection, auth_methods, selection).await?;
+            let attempted = selection.method_id.clone();
+            match connection.send_request(make_request()).block_task().await {
+                Ok(response) => Ok(response),
+                Err(error) if error.code == ErrorCode::AuthRequired => {
+                    let required = required
+                        .after_authentication_attempt(attempted)
+                        .with_agent_detail(&error);
+                    required.log_methods(&format!(
+                        "still required to {operation} after authenticating with '{}'",
+                        selection.method_id
+                    ));
+                    Err(required.describe(operation))
+                }
+                Err(error) => Err(format!(
+                    "Failed to {operation} after authentication with explicitly selected method '{}': {error:?}",
+                    selection.method_id
+                )),
+            }
+        }
+        Err(error) => Err(format!("Failed to {operation}: {error:?}")),
+    }
+}
+
+async fn authenticate_with_explicit_method(
+    connection: &ConnectionTo<Agent>,
+    auth_methods: &[AuthMethod],
+    selection: &AcpAuthenticationSelection,
 ) -> Result<(), String> {
-    let Some(method) = auth_methods.first() else {
-        return Ok(());
-    };
+    let method = auth_methods
+        .iter()
+        .find(|method| method.id().to_string() == selection.method_id)
+        .ok_or_else(|| {
+            // Both of these are integration bugs rather than something the user
+            // can fix, so the inventory that proves it goes to the log and the
+            // error stays the one sentence that reaches the session alert.
+            AcpAuthenticationRequired::from_auth_methods(auth_methods)
+                .log_methods("method selected explicitly but not advertised");
+            format!(
+                "ACP authentication method '{}' was selected explicitly, but the agent did not advertise it.",
+                selection.method_id,
+            )
+        })?;
+    let details = auth_method_details(method);
+    if !details.can_handle {
+        AcpAuthenticationRequired::from_auth_methods(auth_methods)
+            .log_methods("selected method cannot be handled by this client");
+        return Err(format!(
+            "ACP authentication method '{}' ({}) cannot be handled by this client because it is {}.",
+            details.display_name,
+            details.id,
+            details.category.label(),
+        ));
+    }
 
     log::debug!(
-        "ACP agent advertised authentication methods; selecting {} ({})",
+        "ACP authentication required; using explicitly selected method {} ({})",
         method.name(),
         method.id()
     );
@@ -5769,9 +6111,9 @@ async fn authenticate_if_advertised(
         .send_request(AuthenticateRequest::new(method.id().clone()))
         .block_task()
         .await
-        .map_err(|e| {
+        .map_err(|error| {
             format!(
-                "ACP authentication failed with method {} ({}): {e:?}",
+                "ACP authentication failed with explicitly selected method {} ({}): {error:?}",
                 method.name(),
                 method.id()
             )
@@ -5932,36 +6274,41 @@ mod tests {
     use std::collections::BTreeSet;
 
     use super::{
-        acp_spawn_command, air_client_capabilities_meta, apply_or_record_session_config_options,
-        async_task_stop_message, async_task_stop_outcome, autoapprove_permission_decision,
-        background_continuation_origin, background_task_tracking_meta, build_prompt_content_blocks,
-        consume_remote_acp_line, decode_remote_acp_line, defensive_permission_decision,
+        acp_spawn_command, agent_auth_required_detail, air_client_capabilities_meta,
+        apply_or_record_session_config_options, async_task_stop_message, async_task_stop_outcome,
+        autoapprove_permission_decision, background_continuation_origin,
+        background_task_tracking_meta, build_prompt_content_blocks, consume_remote_acp_line,
+        decode_remote_acp_line, defensive_permission_decision, describe_prompt_error,
         hold_for_background_quiescence, is_config_selection_unavailable_error,
         is_missing_mcp_transport_error, labeled_background_continuation_origin,
         mcp_server_transport_supported, origin_task_name_label, permission_response_for_decision,
         permission_response_for_options, reject_queued_stop_requests, remote_acp_segments,
         resolve_acp_working_dir, resolve_session_config_option_selection,
-        resolve_spawn_working_dir, sanitize_remote_acp_chunk, sdk_message_mentions_task,
-        sdk_message_origin_kind, sdk_message_session_state, sdk_message_settles_task,
-        shell_exec_line, shell_quote, task_tracking_mode_from_initialize, AcpDriver,
-        AcpEventMetadata, AcpNotificationHandler, AcpPermissionDecision, AcpPermissionOption,
+        resolve_spawn_working_dir, run_prompt_turn, sanitize_remote_acp_chunk,
+        sdk_message_mentions_task, sdk_message_origin_kind, sdk_message_session_state,
+        sdk_message_settles_task, send_session_setup_request, setup_acp_session, shell_exec_line,
+        shell_quote, task_tracking_mode_from_initialize, AcpAuthenticationMethodCategory,
+        AcpAuthenticationRequired, AcpAuthenticationSelection, AcpDriver, AcpEventMetadata,
+        AcpNotificationHandler, AcpPermissionDecision, AcpPermissionOption,
         AcpPermissionOptionKind, AcpPermissionRequest, AcpSessionConfigOptionSelection,
-        AcpToolCallMetadata, AgentRunOutcome, AsyncTaskNotification, AsyncTaskState,
-        AsyncTaskStopHandle, AsyncTaskUpdate, BackgroundActivity, BackgroundHoldConfig,
-        BackgroundHoldObserver, BackgroundHoldStatus, BackgroundHoldTask, BackgroundTaskSet,
-        BasicMessageWriter, HoldOutcome, HoldSettle, HoldingState, IncomingSessionUpdate,
-        MessageWriter, OutOfTurnPermissionPolicy, QueuedSessionTurn, RemoteLineOutcome,
-        ReplayBoundary, ReplayBuffer, ReplayEvent, SdkSessionState, SessionLifetime,
-        SessionSettleReason, SessionSettled, StopAsyncTaskRequest, TaskTrackingMode,
-        TypedAsyncTaskSet, ASYNC_TASK_STOP_METHOD, AVAILABILITY_PROBE_SUBTYPE,
-        BACKGROUND_CONTINUATION_ORIGIN, BACKGROUND_TASK_SUBTYPES, CLAUDE_SDK_MESSAGE_METHOD,
-        CONTINUATION_MESSAGE_ID_PREFIX, ORIGIN_TASK_NAME_MAX_CHARS, PERMISSION_ANNOUNCEMENT_GRACE,
-        SESSION_STATE_SUBTYPE, TASK_NOTIFICATION_ORIGIN,
+        AcpSessionSetup, AcpSessionSetupContext, AcpToolCallMetadata, AgentRunOutcome,
+        AsyncTaskNotification, AsyncTaskState, AsyncTaskStopHandle, AsyncTaskUpdate,
+        BackgroundActivity, BackgroundHoldConfig, BackgroundHoldObserver, BackgroundHoldStatus,
+        BackgroundHoldTask, BackgroundTaskSet, BasicMessageWriter, HoldOutcome, HoldSettle,
+        HoldingState, IncomingSessionUpdate, MessageWriter, OutOfTurnPermissionPolicy,
+        QueuedSessionTurn, RemoteLineOutcome, ReplayBoundary, ReplayBuffer, ReplayEvent,
+        SdkSessionState, SessionLifetime, SessionSettleReason, SessionSettled,
+        StopAsyncTaskRequest, Store, TaskTrackingMode, TypedAsyncTaskSet, ASYNC_TASK_STOP_METHOD,
+        AVAILABILITY_PROBE_SUBTYPE, BACKGROUND_CONTINUATION_ORIGIN, BACKGROUND_TASK_SUBTYPES,
+        CLAUDE_SDK_MESSAGE_METHOD, CONTINUATION_MESSAGE_ID_PREFIX, ORIGIN_TASK_NAME_MAX_CHARS,
+        PERMISSION_ANNOUNCEMENT_GRACE, SESSION_STATE_SUBTYPE, TASK_NOTIFICATION_ORIGIN,
     };
     use agent_client_protocol::schema::v1::{
-        ContentBlock as AcpContentBlock, ContentChunk, ExtNotification, McpCapabilities, McpServer,
-        McpServerHttp, McpServerSse, McpServerStdio, PermissionOption, PermissionOptionKind, Plan,
-        PlanEntry, PlanEntryPriority, PlanEntryStatus, RequestPermissionOutcome,
+        AgentCapabilities, AuthMethod, AuthMethodAgent, AuthenticateRequest, AuthenticateResponse,
+        ContentBlock as AcpContentBlock, ContentChunk, ExtNotification, LoadSessionRequest,
+        McpCapabilities, McpServer, McpServerHttp, McpServerSse, McpServerStdio, NewSessionRequest,
+        NewSessionResponse, PermissionOption, PermissionOptionKind, Plan, PlanEntry,
+        PlanEntryPriority, PlanEntryStatus, PromptRequest, RequestPermissionOutcome,
         RequestPermissionRequest, SessionConfigOption, SessionConfigOptionCategory,
         SessionConfigSelectOption, SessionNotification, SessionUpdate,
         SetSessionConfigOptionRequest, SetSessionConfigOptionResponse, StopReason, TextContent,
@@ -5976,6 +6323,631 @@ mod tests {
     use tokio::sync::{mpsc, oneshot};
     use tokio_util::sync::CancellationToken;
 
+    #[derive(Default)]
+    struct RecordingStore {
+        agent_session_ids: Mutex<Vec<(String, String)>>,
+    }
+
+    #[async_trait::async_trait]
+    impl Store for RecordingStore {
+        fn set_agent_session_id(
+            &self,
+            session_id: &str,
+            agent_session_id: &str,
+        ) -> Result<(), String> {
+            self.agent_session_ids
+                .lock()
+                .unwrap()
+                .push((session_id.to_string(), agent_session_id.to_string()));
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn auth_method_metadata_does_not_guess_from_provider_names() {
+        let methods = vec![
+            AuthMethod::Agent(AuthMethodAgent::new("api-key", "API Key")),
+            AuthMethod::Agent(AuthMethodAgent::new("chat-gpt", "ChatGPT")),
+            AuthMethod::Agent(AuthMethodAgent::new("arbitrary", "Arbitrary")),
+        ];
+
+        let required = AcpAuthenticationRequired::from_auth_methods(&methods);
+
+        assert_eq!(required.methods.len(), 3);
+        assert!(required.methods.iter().all(|method| method.can_handle));
+        assert!(required
+            .methods
+            .iter()
+            .all(|method| method.category == AcpAuthenticationMethodCategory::AgentManaged));
+        assert_eq!(required.methods[0].id, "api-key");
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn session_setup_returns_auth_required_without_guessing_a_method() {
+        let calls = Arc::new(Mutex::new(Vec::<String>::new()));
+        let calls_for_auth = Arc::clone(&calls);
+        let calls_for_session = Arc::clone(&calls);
+        let agent = agent_client_protocol::Agent
+            .builder()
+            .on_receive_request(
+                async move |request: AuthenticateRequest, responder, _cx| {
+                    calls_for_auth
+                        .lock()
+                        .unwrap()
+                        .push(format!("authenticate:{}", request.method_id));
+                    responder.respond(AuthenticateResponse::new())
+                },
+                agent_client_protocol::on_receive_request!(),
+            )
+            .on_receive_request(
+                async move |_request: NewSessionRequest, responder, _cx| {
+                    calls_for_session.lock().unwrap().push("session/new".into());
+                    responder.respond_with_error(agent_client_protocol::Error::auth_required())
+                },
+                agent_client_protocol::on_receive_request!(),
+            );
+        let methods = vec![
+            AuthMethod::Agent(AuthMethodAgent::new("api-key", "API Key")),
+            AuthMethod::Agent(AuthMethodAgent::new("chat-gpt", "ChatGPT")),
+        ];
+
+        let error = agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                send_session_setup_request(
+                    &connection,
+                    || NewSessionRequest::new(PathBuf::from("/tmp")),
+                    &methods,
+                    None,
+                    "create ACP session",
+                )
+                .await
+                .map(|_| ())
+                .map_err(agent_client_protocol::util::internal_error)
+            })
+            .await
+            .expect_err("auth_required should be surfaced without authenticate");
+
+        assert!(format!("{error:?}").contains("ACP authentication is required"));
+        // The exact string the session alert shows once `run` has wrapped it;
+        // `authRecovery.test.ts` pins the same bytes on the frontend side.
+        assert_eq!(
+            format!("ACP protocol failed: {error:?}"),
+            r#"ACP protocol failed: Error { code: -32603: Internal error, message: "Internal error", data: Some(String("ACP authentication is required to create ACP session. Sign this agent in, then retry.")) }"#
+        );
+        assert_eq!(calls.lock().unwrap().as_slice(), &["session/new"]);
+    }
+
+    /// The bridges that accept `session/new` signed out (amp, claude, goose
+    /// with a native provider) raise `-32000` at `session/prompt` instead. The
+    /// user must read the same one sentence the setup path produces, not the
+    /// Debug rendering of the ACP error.
+    #[tokio::test(flavor = "current_thread")]
+    async fn prompt_auth_required_reads_like_the_session_setup_error() {
+        let auth_required = agent_client_protocol::Error::auth_required();
+        assert_eq!(i32::from(auth_required.code), -32000);
+        let agent = agent_client_protocol::Agent.builder().on_receive_request(
+            async move |_request: PromptRequest, responder, _cx| {
+                responder.respond_with_error(auth_required.clone())
+            },
+            agent_client_protocol::on_receive_request!(),
+        );
+        let setup = AcpSessionSetup {
+            agent_session_id: "session-1".to_string(),
+            agent_capabilities: AgentCapabilities::default(),
+            task_tracking_mode: TaskTrackingMode::Raw,
+            auth_methods: vec![AuthMethod::Agent(AuthMethodAgent::new(
+                "api-key", "API Key",
+            ))],
+        };
+        let handler = Arc::new(AcpNotificationHandler::new(
+            Arc::new(BasicMessageWriter::new()),
+            false,
+            vec![],
+            CancellationToken::new(),
+        ));
+
+        let outcome = agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                Ok(run_prompt_turn(
+                    &connection,
+                    &setup,
+                    &handler,
+                    "local-session",
+                    &CancellationToken::new(),
+                    "hello",
+                    &[],
+                )
+                .await)
+            })
+            .await
+            .expect("the connection itself stays healthy");
+
+        let error = outcome.expect_err("auth_required should fail the prompt");
+        assert_eq!(
+            error,
+            "ACP authentication is required to run the prompt. Sign this agent in, then retry."
+        );
+        assert!(error.contains("authentication"));
+
+        // `run` wraps every protocol failure the same way before it reaches
+        // the session alert; `authRecovery.test.ts` pins this exact string on
+        // the frontend side, so the two must agree byte for byte.
+        let wrapped = format!(
+            "ACP protocol failed: {:?}",
+            agent_client_protocol::util::internal_error(error)
+        );
+        assert_eq!(
+            wrapped,
+            r#"ACP protocol failed: Error { code: -32603: Internal error, message: "Internal error", data: Some(String("ACP authentication is required to run the prompt. Sign this agent in, then retry.")) }"#
+        );
+    }
+
+    #[test]
+    fn prompt_errors_other_than_auth_required_keep_their_debug_shape() {
+        let methods = vec![AuthMethod::Agent(AuthMethodAgent::new(
+            "api-key", "API Key",
+        ))];
+
+        let error = describe_prompt_error(agent_client_protocol::Error::internal_error(), &methods);
+
+        assert_eq!(
+            error,
+            r#"Prompt failed: Error { code: -32603: Internal error, message: "Internal error", data: None }"#
+        );
+    }
+
+    /// A `-32000` with the bridge's own sentence on it, as pi-acp sends at
+    /// `session/new`; the stock `auth_required` is what every other surveyed
+    /// bridge sends.
+    fn auth_required_saying(message: &str) -> agent_client_protocol::Error {
+        let mut error = agent_client_protocol::Error::auth_required();
+        error.message = message.to_string();
+        error
+    }
+
+    /// pi-acp answers `session/new` signed out with "Configure an API key or log
+    /// in with an OAuth provider." on the `-32000`, which names the remedy
+    /// where the stock text does not. It is appended after the one sentence the
+    /// user already read, so the frontend's match on "authentication" holds.
+    #[tokio::test(flavor = "current_thread")]
+    async fn session_setup_auth_required_appends_the_agents_own_message() {
+        let agent = agent_client_protocol::Agent.builder().on_receive_request(
+            async move |_request: NewSessionRequest, responder, _cx| {
+                responder.respond_with_error(auth_required_saying(
+                    "Configure an API key or log in with an OAuth provider.",
+                ))
+            },
+            agent_client_protocol::on_receive_request!(),
+        );
+        let methods = vec![AuthMethod::Agent(AuthMethodAgent::new(
+            "api-key", "API Key",
+        ))];
+
+        let error = agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                Ok(send_session_setup_request(
+                    &connection,
+                    || NewSessionRequest::new(PathBuf::from("/tmp")),
+                    &methods,
+                    None,
+                    "create ACP session",
+                )
+                .await
+                .map(|_| ()))
+            })
+            .await
+            .expect("the connection itself stays healthy")
+            .expect_err("auth_required should fail the setup");
+
+        assert_eq!(
+            error,
+            "ACP authentication is required to create ACP session. Sign this agent in, then retry. The agent said: Configure an API key or log in with an OAuth provider."
+        );
+        assert!(error.to_lowercase().contains("authentication"));
+    }
+
+    /// The prompt path appends the same detail the same way, so a bridge that
+    /// says more at `session/prompt` reads like one that says it at `session/new`.
+    #[test]
+    fn prompt_auth_required_appends_the_agents_own_message() {
+        let methods = vec![AuthMethod::Agent(AuthMethodAgent::new(
+            "api-key", "API Key",
+        ))];
+
+        let error = describe_prompt_error(
+            auth_required_saying("Configure an API key or log in with an OAuth provider."),
+            &methods,
+        );
+
+        assert_eq!(
+            error,
+            "ACP authentication is required to run the prompt. Sign this agent in, then retry. The agent said: Configure an API key or log in with an OAuth provider."
+        );
+    }
+
+    /// The stock `auth_required` — which is what amp, claude, codex and goose
+    /// send — must keep producing exactly the strings pinned in
+    /// `authRecovery.test.ts`; only text the agent added is appended.
+    #[test]
+    fn auth_required_detail_keeps_only_what_the_agent_added() {
+        let stock = agent_client_protocol::Error::auth_required();
+        assert_eq!(stock.message, "Authentication required");
+        assert_eq!(agent_auth_required_detail(&stock), None);
+
+        // The stock text restated in `data`, or with different case and a
+        // period, is still nothing the user has not already been told.
+        let mut restated = auth_required_saying("authentication required.");
+        restated.data = Some(serde_json::Value::String("Authentication required".into()));
+        assert_eq!(agent_auth_required_detail(&restated), None);
+
+        // Blank and structured data are not rendered.
+        let mut blank = agent_client_protocol::Error::auth_required();
+        blank.data = Some(serde_json::json!({ "reason": "expired" }));
+        assert_eq!(agent_auth_required_detail(&blank), None);
+        blank.data = Some(serde_json::Value::String("   \n".into()));
+        assert_eq!(agent_auth_required_detail(&blank), None);
+
+        // A remedy in `data` alone is kept.
+        let mut in_data = agent_client_protocol::Error::auth_required();
+        in_data.data = Some(serde_json::Value::String("Run `amp login` first".into()));
+        assert_eq!(
+            agent_auth_required_detail(&in_data).as_deref(),
+            Some("Run `amp login` first.")
+        );
+
+        // `message` and `data` both kept, in order, once each, as sentences;
+        // a multi-line `data` collapses onto the alert's one paragraph.
+        let mut both = auth_required_saying("Not signed in");
+        both.data = Some(serde_json::Value::String(
+            "Run `amp login`\n  first.".into(),
+        ));
+        assert_eq!(
+            agent_auth_required_detail(&both).as_deref(),
+            Some("Not signed in. Run `amp login` first.")
+        );
+        let mut same = auth_required_saying("Not signed in.");
+        same.data = Some(serde_json::Value::String("Not signed in".into()));
+        assert_eq!(
+            agent_auth_required_detail(&same).as_deref(),
+            Some("Not signed in.")
+        );
+
+        // And the sentence the user reads is unchanged for the stock error.
+        let required = AcpAuthenticationRequired::from_auth_methods(&[]).with_agent_detail(&stock);
+        assert_eq!(
+            required.describe("create ACP session"),
+            "ACP authentication is required to create ACP session. Sign this agent in, then retry."
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn session_setup_authenticates_explicit_method_and_retries_once() {
+        let calls = Arc::new(Mutex::new(Vec::<String>::new()));
+        let calls_for_auth = Arc::clone(&calls);
+        let calls_for_session = Arc::clone(&calls);
+        let session_attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let session_attempts_for_handler = Arc::clone(&session_attempts);
+        let agent = agent_client_protocol::Agent
+            .builder()
+            .on_receive_request(
+                async move |request: AuthenticateRequest, responder, _cx| {
+                    calls_for_auth
+                        .lock()
+                        .unwrap()
+                        .push(format!("authenticate:{}", request.method_id));
+                    responder.respond(AuthenticateResponse::new())
+                },
+                agent_client_protocol::on_receive_request!(),
+            )
+            .on_receive_request(
+                async move |_request: NewSessionRequest, responder, _cx| {
+                    calls_for_session.lock().unwrap().push("session/new".into());
+                    if session_attempts_for_handler
+                        .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
+                        == 0
+                    {
+                        responder.respond_with_error(agent_client_protocol::Error::auth_required())
+                    } else {
+                        responder.respond(NewSessionResponse::new("session-1"))
+                    }
+                },
+                agent_client_protocol::on_receive_request!(),
+            );
+        let methods = vec![
+            AuthMethod::Agent(AuthMethodAgent::new("api-key", "API Key")),
+            AuthMethod::Agent(AuthMethodAgent::new("chat-gpt", "ChatGPT")),
+        ];
+        let selection = AcpAuthenticationSelection::new("chat-gpt");
+
+        agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                send_session_setup_request(
+                    &connection,
+                    || NewSessionRequest::new(PathBuf::from("/tmp")),
+                    &methods,
+                    Some(&selection),
+                    "create ACP session",
+                )
+                .await
+                .map(|_| ())
+                .map_err(agent_client_protocol::util::internal_error)
+            })
+            .await
+            .expect("protocol should succeed");
+
+        assert_eq!(
+            calls.lock().unwrap().as_slice(),
+            &["session/new", "authenticate:chat-gpt", "session/new"]
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn session_setup_stops_after_second_auth_required() {
+        let calls = Arc::new(Mutex::new(Vec::<String>::new()));
+        let calls_for_auth = Arc::clone(&calls);
+        let calls_for_session = Arc::clone(&calls);
+        let agent = agent_client_protocol::Agent
+            .builder()
+            .on_receive_request(
+                async move |request: AuthenticateRequest, responder, _cx| {
+                    calls_for_auth
+                        .lock()
+                        .unwrap()
+                        .push(format!("authenticate:{}", request.method_id));
+                    responder.respond(AuthenticateResponse::new())
+                },
+                agent_client_protocol::on_receive_request!(),
+            )
+            .on_receive_request(
+                async move |_request: NewSessionRequest, responder, _cx| {
+                    calls_for_session.lock().unwrap().push("session/new".into());
+                    responder.respond_with_error(agent_client_protocol::Error::auth_required())
+                },
+                agent_client_protocol::on_receive_request!(),
+            );
+        let methods = vec![AuthMethod::Agent(AuthMethodAgent::new(
+            "chat-gpt", "ChatGPT",
+        ))];
+        let selection = AcpAuthenticationSelection::new("chat-gpt");
+
+        let error = agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                send_session_setup_request(
+                    &connection,
+                    || NewSessionRequest::new(PathBuf::from("/tmp")),
+                    &methods,
+                    Some(&selection),
+                    "create ACP session",
+                )
+                .await
+                .map(|_| ())
+                .map_err(agent_client_protocol::util::internal_error)
+            })
+            .await
+            .expect_err("a second auth_required should terminate the setup");
+
+        let error = format!("{error:?}");
+        // The user-facing sentence has to name the method that was tried, so
+        // "sign in again" doesn't read as a suggestion to repeat what just
+        // failed. The full advertised-method inventory stays in the debug log.
+        assert!(
+            error.contains("Signing in with 'chat-gpt' did not clear it"),
+            "error should say the selected method was already tried; got {error}",
+        );
+        assert_eq!(
+            calls.lock().unwrap().as_slice(),
+            &["session/new", "authenticate:chat-gpt", "session/new"]
+        );
+    }
+
+    #[test]
+    fn terminal_methods_are_reported_unsupported() {
+        use agent_client_protocol::schema::v1::AuthMethodTerminal;
+
+        let methods = vec![AuthMethod::Terminal(AuthMethodTerminal::new(
+            "terminal-login",
+            "Terminal Login",
+        ))];
+        let required = AcpAuthenticationRequired::from_auth_methods(&methods);
+
+        assert_eq!(
+            required.methods[0].category,
+            AcpAuthenticationMethodCategory::Terminal
+        );
+        assert!(!required.methods[0].can_handle);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn terminal_methods_are_never_sent_to_authenticate() {
+        use agent_client_protocol::schema::v1::AuthMethodTerminal;
+
+        let auth_called = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let auth_called_for_handler = Arc::clone(&auth_called);
+        let agent = agent_client_protocol::Agent
+            .builder()
+            .on_receive_request(
+                async move |_request: AuthenticateRequest, responder, _cx| {
+                    auth_called_for_handler.store(true, std::sync::atomic::Ordering::SeqCst);
+                    responder.respond(AuthenticateResponse::new())
+                },
+                agent_client_protocol::on_receive_request!(),
+            )
+            .on_receive_request(
+                async move |_request: NewSessionRequest, responder, _cx| {
+                    responder.respond_with_error(agent_client_protocol::Error::auth_required())
+                },
+                agent_client_protocol::on_receive_request!(),
+            );
+        let methods = vec![AuthMethod::Terminal(AuthMethodTerminal::new(
+            "terminal-login",
+            "Terminal Login",
+        ))];
+        let selection = AcpAuthenticationSelection::new("terminal-login");
+
+        let error = agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                send_session_setup_request(
+                    &connection,
+                    || NewSessionRequest::new(PathBuf::from("/tmp")),
+                    &methods,
+                    Some(&selection),
+                    "create ACP session",
+                )
+                .await
+                .map(|_| ())
+                .map_err(agent_client_protocol::util::internal_error)
+            })
+            .await
+            .expect_err("terminal auth is unsupported");
+
+        assert!(format!("{error:?}").contains("terminal"));
+        assert!(!auth_called.load(std::sync::atomic::Ordering::SeqCst));
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn full_session_setup_uses_existing_login_without_eager_authentication() {
+        use agent_client_protocol::schema::v1::{InitializeRequest, InitializeResponse};
+
+        let calls = Arc::new(Mutex::new(Vec::<String>::new()));
+        let calls_for_auth = Arc::clone(&calls);
+        let calls_for_session = Arc::clone(&calls);
+        let agent = agent_client_protocol::Agent
+            .builder()
+            .on_receive_request(
+                async |request: InitializeRequest, responder, _cx| {
+                    responder.respond(
+                        InitializeResponse::new(request.protocol_version).auth_methods(vec![
+                            AuthMethod::Agent(AuthMethodAgent::new("api-key", "API Key")),
+                            AuthMethod::Agent(AuthMethodAgent::new("chat-gpt", "ChatGPT")),
+                        ]),
+                    )
+                },
+                agent_client_protocol::on_receive_request!(),
+            )
+            .on_receive_request(
+                async move |request: AuthenticateRequest, responder, _cx| {
+                    calls_for_auth
+                        .lock()
+                        .unwrap()
+                        .push(format!("authenticate:{}", request.method_id));
+                    responder.respond_with_error(agent_client_protocol::Error::internal_error())
+                },
+                agent_client_protocol::on_receive_request!(),
+            )
+            .on_receive_request(
+                async move |_request: NewSessionRequest, responder, _cx| {
+                    calls_for_session.lock().unwrap().push("session/new".into());
+                    responder.respond(NewSessionResponse::new("session-1"))
+                },
+                agent_client_protocol::on_receive_request!(),
+            );
+        let store: Arc<dyn Store> = Arc::new(RecordingStore::default());
+        let handler = Arc::new(AcpNotificationHandler::new(
+            Arc::new(BasicMessageWriter::new()),
+            false,
+            vec![],
+            CancellationToken::new(),
+        ));
+
+        agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                setup_acp_session(AcpSessionSetupContext {
+                    connection: &connection,
+                    working_dir: Path::new("/tmp"),
+                    store: &store,
+                    handler: &handler,
+                    our_session_id: "local-session",
+                    acp_session_id: None,
+                    config_options: &[],
+                    mcp_servers: &[],
+                    agent_label: "Codex",
+                    auth_selection: None,
+                })
+                .await
+                .map(|_| ())
+                .map_err(agent_client_protocol::util::internal_error)
+            })
+            .await
+            .expect("existing login should create a session without authenticate");
+
+        assert_eq!(calls.lock().unwrap().as_slice(), &["session/new"]);
+    }
+
+    /// The `session/load` counterpart of the `session/new` and `session/prompt`
+    /// assertions: a resumed session on a bridge that raises `-32000` at load
+    /// reads the same sentence with its own operation. `authRecovery.test.ts`
+    /// pins this exact wrapped string on the frontend side, so the two must
+    /// agree byte for byte.
+    #[tokio::test(flavor = "current_thread")]
+    async fn full_session_setup_surfaces_auth_required_at_session_load() {
+        use agent_client_protocol::schema::v1::{InitializeRequest, InitializeResponse};
+
+        let calls = Arc::new(Mutex::new(Vec::<String>::new()));
+        let calls_for_load = Arc::clone(&calls);
+        let agent = agent_client_protocol::Agent
+            .builder()
+            .on_receive_request(
+                async |request: InitializeRequest, responder, _cx| {
+                    responder.respond(
+                        InitializeResponse::new(request.protocol_version)
+                            .agent_capabilities(AgentCapabilities::default().load_session(true))
+                            .auth_methods(vec![AuthMethod::Agent(AuthMethodAgent::new(
+                                "claude-login",
+                                "Log in with Claude Code",
+                            ))]),
+                    )
+                },
+                agent_client_protocol::on_receive_request!(),
+            )
+            .on_receive_request(
+                async move |request: LoadSessionRequest, responder, _cx| {
+                    calls_for_load
+                        .lock()
+                        .unwrap()
+                        .push(format!("session/load:{}", request.session_id));
+                    responder.respond_with_error(agent_client_protocol::Error::auth_required())
+                },
+                agent_client_protocol::on_receive_request!(),
+            );
+        let store: Arc<dyn Store> = Arc::new(RecordingStore::default());
+        let handler = Arc::new(AcpNotificationHandler::new(
+            Arc::new(BasicMessageWriter::new()),
+            false,
+            vec![],
+            CancellationToken::new(),
+        ));
+
+        let error = agent_client_protocol::Client
+            .connect_with(agent, async |connection| {
+                setup_acp_session(AcpSessionSetupContext {
+                    connection: &connection,
+                    working_dir: Path::new("/tmp"),
+                    store: &store,
+                    handler: &handler,
+                    our_session_id: "local-session",
+                    acp_session_id: Some("session-1"),
+                    config_options: &[],
+                    mcp_servers: &[],
+                    agent_label: "Claude",
+                    auth_selection: None,
+                })
+                .await
+                .map(|_| ())
+                .map_err(agent_client_protocol::util::internal_error)
+            })
+            .await
+            .expect_err("auth_required at session/load should fail the setup");
+
+        assert_eq!(
+            format!("ACP protocol failed: {error:?}"),
+            r#"ACP protocol failed: Error { code: -32603: Internal error, message: "Internal error", data: Some(String("ACP authentication is required to load ACP session. Sign this agent in, then retry.")) }"#
+        );
+        assert_eq!(
+            calls.lock().unwrap().as_slice(),
+            &["session/load:session-1"]
+        );
+    }
+
     fn unique_test_dir(prefix: &str) -> PathBuf {
         let nonce = SystemTime::now()
             .duration_since(UNIX_EPOCH)
diff --git a/crates/acp-client/src/lib.rs b/crates/acp-client/src/lib.rs
index c7fc1e6e7..1bcb4c898 100644
--- a/crates/acp-client/src/lib.rs
+++ b/crates/acp-client/src/lib.rs
@@ -29,12 +29,12 @@ pub use agent_client_protocol::schema::v1::{
 pub use driver::{
     autoapprove_permission_decision, background_continuation_origin,
     is_config_selection_unavailable_error, is_missing_mcp_transport_error,
-    labeled_background_continuation_origin, strip_code_fences, AcpDriver, AcpEventMetadata,
-    AcpInitializeMetadata, AcpPermissionDecision, AcpPermissionOption, AcpPermissionOptionKind,
-    AcpPermissionRequest, AcpSessionConfigOptionSelection, AcpToolCallMetadata, AgentDriver,
-    AgentRunOutcome, AsyncTaskStopHandle, BackgroundHoldConfig, BackgroundHoldObserver,
-    BackgroundHoldStatus, BackgroundHoldTask, BasicMessageWriter, MessageWriter,
-    OutOfTurnPermissionPolicy, ReplayBoundary, SessionConnection, SessionLifetime,
+    labeled_background_continuation_origin, strip_code_fences, AcpAuthenticationSelection,
+    AcpDriver, AcpEventMetadata, AcpInitializeMetadata, AcpPermissionDecision, AcpPermissionOption,
+    AcpPermissionOptionKind, AcpPermissionRequest, AcpSessionConfigOptionSelection,
+    AcpToolCallMetadata, AgentDriver, AgentRunOutcome, AsyncTaskStopHandle, BackgroundHoldConfig,
+    BackgroundHoldObserver, BackgroundHoldStatus, BackgroundHoldTask, BasicMessageWriter,
+    MessageWriter, OutOfTurnPermissionPolicy, ReplayBoundary, SessionConnection, SessionLifetime,
     SessionSettleReason, SessionSettled, Store, BACKGROUND_CONTINUATION_ORIGIN,
 };
 pub use simple::{run_acp_prompt, run_acp_prompt_with_interpreter_env_snapshot};
diff --git a/crates/acp-client/src/simple.rs b/crates/acp-client/src/simple.rs
index 901a02e99..17675f75f 100644
--- a/crates/acp-client/src/simple.rs
+++ b/crates/acp-client/src/simple.rs
@@ -58,6 +58,7 @@ impl AgentDriver for SimpleDriverWrapper {
         cancel_token: &CancellationToken,
         agent_session_id: Option<&str>,
         config_options: &[crate::driver::AcpSessionConfigOptionSelection],
+        auth_selection: Option<&crate::driver::AcpAuthenticationSelection>,
     ) -> Result<AgentRunOutcome, String> {
         if !images.is_empty() {
             log::debug!(
@@ -77,6 +78,7 @@ impl AgentDriver for SimpleDriverWrapper {
                 cancel_token,
                 agent_session_id,
                 config_options,
+                auth_selection,
             )
             .await
     }
@@ -160,6 +162,7 @@ async fn run_acp_prompt_with_options(
                     &cancel_token,
                     None,
                     &[],
+                    None,
                 )
                 .await
                 .map_err(|e| anyhow::anyhow!("ACP driver error: {e}"))?;
diff --git a/crates/doctor/src/command.rs b/crates/doctor/src/command.rs
index 0c9099b07..7276b16a4 100644
--- a/crates/doctor/src/command.rs
+++ b/crates/doctor/src/command.rs
@@ -218,6 +218,8 @@ pub(crate) fn kill_child_process_group_or_child(child: &mut Child) -> KillReach
     KillReach::ChildOnly
 }
 
+/// `SIGKILL` the child's process group: `kill(2)` treats a negative pid as a
+/// process group ID.
 #[cfg(unix)]
 fn kill_child_process_group(child: &Child) -> bool {
     let Ok(pid) = i32::try_from(child.id()) else {