agentmux_srv/server/identity_auth_spawn.rs
1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Subprocess spawn/drain engine for the pre-launch OAuth flow.
5//!
6//! Split out of `identity_handlers.rs` (module-organization pass, see
7//! `docs/specs/PLAN_LOGIN_SINGLE_PATH_CONSOLIDATION_2026_07_20.md`) —
8//! owns `spawn_auth_cli` (plain piped-stdio spawn), `spawn_auth_cli_pty`
9//! (PTY-backed spawn for providers whose auth subcommand requires an
10//! interactive TTY), and `confirm_authenticated` (the shared
11//! authCheckCommand probe both spawn paths use). Both spawn paths call
12//! into `identity_auth_persist::persist_oauth_success` on the OAuth-
13//! success path.
14
15#[cfg(windows)]
16use agentmux_common::win32::NoWindow;
17use std::sync::atomic::{AtomicBool, Ordering};
18use std::sync::Arc;
19
20use crate::backend::storage::store::Store;
21use crate::backend::mps::Broker;
22
23use super::identity_auth_persist::persist_oauth_success;
24
25/// Spawn the provider's auth-login CLI and drive the session through
26/// to a terminal state. Background-only — returns immediately. The
27/// drain task feeds stdout+stderr lines into
28/// `AuthSessionManager::record_line`; on a login-success pattern OR
29/// child exit, runs the provider's authCheckCommand to confirm and
30/// transitions to Success or Failed.
31///
32/// On the success path (CLI exited cleanly + authCheckCommand passed),
33/// persists a standalone `IdentityAccount` (`persist_oauth_direct_account`)
34/// with `SecretRef::OAuthConfigDir` pointing at `bundle_dir` (the
35/// account's own isolation dir — see `compute_and_ensure_account_dir`),
36/// status `valid`. `account_id` is the id minted/reused for that account.
37/// After this point, future launches of any agent linked to the account
38/// resolve through `inject_identity_env`'s oauth-class dispatch and reuse
39/// the same CLI-managed tokens.
40///
41/// `into_bundle_id`/`direct_account` are vestigial parameters kept only
42/// so the wire request shape and call sites don't need touching — bundle
43/// mode (binding into `db_identity_bundles`) was retired in Phase 4c of
44/// SPEC_PRESET_TO_BUNDLE_REFACTOR_2026_07_02.md; `persist_oauth_success`
45/// always takes the direct-account path now.
46#[allow(clippy::too_many_arguments)]
47pub(crate) fn spawn_auth_cli(
48 mgr: Arc<crate::identity::auth_session::AuthSessionManager>,
49 mstore: Arc<Store>,
50 identity_store: Arc<Store>,
51 broker: Arc<Broker>,
52 session_id: String,
53 provider_id: String,
54 into_bundle_id: Option<String>,
55 bundle_dir: Option<String>,
56 direct_account: bool,
57 account_id: String,
58 cli_path: String,
59 auth_login_args: Vec<String>,
60 auth_check_args: Vec<String>,
61 auth_env: std::collections::HashMap<String, String>,
62 requires_tty: bool,
63) {
64 use std::process::Stdio;
65 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
66 use tokio::process::Command;
67 use tokio::sync::mpsc;
68
69 // Channel for SubmitAuthCallback → CLI stdin forwarding.
70 // Buffer of 4 is enough — only one URL per session in normal use.
71 let (stdin_tx, mut stdin_rx) = mpsc::channel::<String>(4);
72
73 // PTY branch: providers like OpenClaw whose auth subcommand
74 // refuses to run when `isatty()==0`. Spawn via portable_pty so the
75 // child sees an interactive terminal, feed PTY output through the
76 // same record_line matcher the pipes path uses. Stdout/stderr are
77 // merged into the single PTY stream — record_line handles both as
78 // identical input.
79 if requires_tty {
80 spawn_auth_cli_pty(
81 mgr,
82 mstore,
83 identity_store,
84 broker,
85 session_id,
86 provider_id,
87 into_bundle_id,
88 bundle_dir,
89 direct_account,
90 account_id,
91 cli_path,
92 auth_login_args,
93 auth_check_args,
94 auth_env,
95 stdin_tx,
96 stdin_rx,
97 );
98 return;
99 }
100
101 let mgr_for_task = mgr.clone();
102 let session_id_for_task = session_id.clone();
103 let cli_path_for_check = cli_path.clone();
104 let auth_check_args_for_check = auth_check_args.clone();
105 let auth_env_for_check = auth_env.clone();
106 let mstore_for_task = mstore.clone();
107 let identity_store_for_task = identity_store.clone();
108 let broker_for_task = broker.clone();
109 let into_bundle_id_for_task = into_bundle_id.clone();
110 let bundle_dir_for_task = bundle_dir.clone();
111 let provider_id_for_task = provider_id.clone();
112 let account_id_for_task = account_id.clone();
113
114 let handle = tokio::spawn(async move {
115 tracing::info!(
116 session_id = %session_id_for_task,
117 provider_id = %provider_id,
118 cli_path = %cli_path,
119 auth_login_args = ?auth_login_args,
120 "auth.spawn: launching provider CLI"
121 );
122
123 // Spawn the CLI. kill_on_drop guarantees cleanup if our
124 // task is aborted (cancel path).
125 let mut cmd = Command::new(&cli_path);
126 // A provider OAuth CLI is a third-party, network-connected binary.
127 // Without this it inherits srv's whole AGENTMUX_* environment — channel,
128 // data dir, cache dir, and this instance's API endpoint. Stricter than
129 // the pane policy: not even the in-pane helper keep-set applies here.
130 // (ReAgent P0, round 2, on PR #3326.)
131 crate::backend::pane_env::sanitize_external_command(&mut cmd);
132 cmd.args(&auth_login_args)
133 .envs(&auth_env)
134 .stdin(Stdio::piped())
135 .stdout(Stdio::piped())
136 .stderr(Stdio::piped())
137 .kill_on_drop(true);
138 // CREATE_NO_WINDOW: console-flash suppression, see crates/common/src/cli.rs
139 #[cfg(windows)]
140 {
141 cmd.no_window();
142 }
143 let mut child = match cmd.spawn()
144 {
145 Ok(c) => {
146 tracing::info!(
147 session_id = %session_id_for_task,
148 pid = ?c.id(),
149 "auth.spawn: child started"
150 );
151 c
152 }
153 Err(e) => {
154 tracing::error!(
155 session_id = %session_id_for_task,
156 error = %e,
157 "auth.spawn: Command::spawn failed"
158 );
159 mgr_for_task.finish_failure(
160 &session_id_for_task,
161 format!("spawn `{cli_path}` failed: {e}"),
162 );
163 mgr_for_task.detach_process(&session_id_for_task);
164 return;
165 }
166 };
167
168 let stdout = child
169 .stdout
170 .take()
171 .expect("piped stdout");
172 let stderr = child
173 .stderr
174 .take()
175 .expect("piped stderr");
176 let mut stdin = child
177 .stdin
178 .take()
179 .expect("piped stdin");
180
181 // Stdin writer — forwards callback URLs (pasted back by the
182 // user when browser auto-open failed) into the CLI process.
183 let stdin_writer = tokio::spawn(async move {
184 while let Some(line) = stdin_rx.recv().await {
185 if stdin.write_all(line.as_bytes()).await.is_err() {
186 break;
187 }
188 if stdin.write_all(b"\n").await.is_err() {
189 break;
190 }
191 let _ = stdin.flush().await;
192 }
193 });
194
195 // Stdout drain — line-by-line into the session manager. When
196 // we see a login-success pattern, confirm via authCheckCommand
197 // and transition to Success. Don't break the loop — some CLIs
198 // continue printing after the success line.
199 let mgr_stdout = mgr_for_task.clone();
200 let sid_stdout = session_id_for_task.clone();
201 let cli_path_stdout = cli_path_for_check.clone();
202 let check_args_stdout = auth_check_args_for_check.clone();
203 let check_env_stdout = auth_env_for_check.clone();
204 let mstore_stdout = mstore_for_task.clone();
205 let identity_store_stdout = identity_store_for_task.clone();
206 let broker_stdout = broker_for_task.clone();
207 let into_bundle_id_stdout = into_bundle_id_for_task.clone();
208 let bundle_dir_stdout = bundle_dir_for_task.clone();
209 let provider_id_stdout = provider_id_for_task.clone();
210 let account_id_stdout = account_id_for_task.clone();
211 // Shared between drain + post-exit. The drain sets it after
212 // persisting on a LoginSuccess match; the post-exit transition
213 // block (below) checks it and skips its entire success path if
214 // already transitioned — without this guard the drain's
215 // persist + post-exit's persist both ran on every successful
216 // OAuth, producing orphan IdentityAccount rows (each `Uuid::new_v4`)
217 // and duplicate `identityaccounts:changed` publishes.
218 // Reagent P1 on #981.
219 let success_transitioned = Arc::new(AtomicBool::new(false));
220 let success_transitioned_drain = Arc::clone(&success_transitioned);
221 let stdout_drain = tokio::spawn(async move {
222 let mut lines = BufReader::new(stdout).lines();
223 while let Ok(Some(line)) = lines.next_line().await {
224 // NOT logging the raw line — OAuth providers print
225 // auth URLs / callback codes / device codes on stdout
226 // and even debug-level logs would persist those into
227 // ~/.agentmux/logs/. Length-only is enough for "the
228 // CLI emitted something" diagnostics. Reagent P2 on #847.
229 tracing::debug!(session_id = %sid_stdout, bytes = line.len(), "auth.spawn: stdout line");
230 let m = mgr_stdout.record_line(&sid_stdout, &line);
231 if !success_transitioned_drain.load(Ordering::Acquire)
232 && matches!(
233 m,
234 Some(crate::identity::auth_patterns::AuthPatternMatch::LoginSuccess { .. })
235 )
236 {
237 if confirm_authenticated(
238 &cli_path_stdout,
239 &check_args_stdout,
240 &check_env_stdout,
241 )
242 .await
243 {
244 // Persist the standalone OAuthConfigDir account
245 // (always direct-account mode now). If `bundle_dir`
246 // (the account's own isolation dir) failed to
247 // resolve at spawn, persist_oauth_success skips
248 // persistence and the session still succeeds.
249 // Read the email BEFORE finish_success consumes the
250 // session — see AuthSessionManager::captured_email.
251 // Ended meanwhile (replaced, cancelled, timed out): a newer sign-in
252 // may own the directory, so this login is not saved.
253 if mgr_stdout.ended(&sid_stdout) {
254 tracing::info!(session_id = %sid_stdout, "auth.spawn: session ended before its login was saved; not saving it");
255 return;
256 }
257 let login_email = mgr_stdout.captured_email(&sid_stdout);
258 let (bundle_id, account_id) = persist_oauth_success(
259 &mstore_stdout,
260 &identity_store_stdout,
261 &broker_stdout,
262 direct_account,
263 &account_id_stdout,
264 into_bundle_id_stdout.as_deref(),
265 &provider_id_stdout,
266 bundle_dir_stdout.as_deref(),
267 &sid_stdout,
268 login_email.as_deref(),
269 );
270 mgr_stdout.finish_success(&sid_stdout, bundle_id, account_id);
271 success_transitioned_drain.store(true, Ordering::Release);
272 }
273 }
274 }
275 });
276
277 // Stderr drain — same matcher path. Some CLIs emit the OAuth
278 // URL on stderr (verbose logging style).
279 let mgr_stderr = mgr_for_task.clone();
280 let sid_stderr = session_id_for_task.clone();
281 let stderr_drain = tokio::spawn(async move {
282 let mut lines = BufReader::new(stderr).lines();
283 while let Ok(Some(line)) = lines.next_line().await {
284 // Same redaction rationale as stdout above.
285 tracing::debug!(session_id = %sid_stderr, bytes = line.len(), "auth.spawn: stderr line");
286 let _ = mgr_stderr.record_line(&sid_stderr, &line);
287 }
288 });
289
290 // Wait for the child to exit (or for our task to be aborted
291 // by cancel_session — in which case kill_on_drop handles the
292 // child).
293 let exit = child.wait().await;
294 tracing::info!(session_id = %session_id_for_task, exit = ?exit, "auth.spawn: child exited");
295
296 // Let stdout/stderr drains catch any final lines.
297 let _ = stdout_drain.await;
298 let _ = stderr_drain.await;
299 // Stdin writer exits when the tx side is dropped (after this
300 // task ends) — abort defensively in case it's blocked on a
301 // pending receive.
302 stdin_writer.abort();
303
304 // Final transition if we haven't already hit Success on the
305 // login-pattern path. Re-run authCheck because some CLIs exit
306 // cleanly without printing a "logged in as" line.
307 //
308 // Skip the entire block when the drain has already transitioned
309 // (LoginSuccess pattern matched + confirm_authenticated + persist
310 // ran). Without this guard, persist_oauth_success would fire a
311 // second time on every successful OAuth — a fresh IdentityAccount
312 // UUID would be upserted and the broker would re-publish the
313 // accounts-changed event. Reagent P1 on #981.
314 if !success_transitioned.load(Ordering::Acquire) {
315 match exit {
316 Ok(s) if s.success() => {
317 if confirm_authenticated(
318 &cli_path_for_check,
319 &auth_check_args_for_check,
320 &auth_env_for_check,
321 )
322 .await
323 {
324 // Read the email BEFORE finish_success consumes the session
325 // — see AuthSessionManager::captured_email.
326 // Ended meanwhile (replaced, cancelled, timed out): a newer sign-in
327 // may own the directory, so this login is not saved.
328 if mgr_for_task.ended(&session_id_for_task) {
329 tracing::info!(session_id = %session_id_for_task, "auth.spawn: session ended before its login was saved; not saving it");
330 return;
331 }
332 let login_email = mgr_for_task.captured_email(&session_id_for_task);
333 let (bundle_id, account_id) = persist_oauth_success(
334 &mstore_for_task,
335 &identity_store_for_task,
336 &broker_for_task,
337 direct_account,
338 &account_id_for_task,
339 into_bundle_id_for_task.as_deref(),
340 &provider_id_for_task,
341 bundle_dir_for_task.as_deref(),
342 &session_id_for_task,
343 login_email.as_deref(),
344 );
345 mgr_for_task.finish_success(&session_id_for_task, bundle_id, account_id);
346 } else {
347 mgr_for_task.finish_failure(
348 &session_id_for_task,
349 "CLI exited 0 but authentication check failed".to_string(),
350 );
351 }
352 }
353 Ok(s) => {
354 mgr_for_task.finish_failure(
355 &session_id_for_task,
356 format!("CLI exited with status {s}"),
357 );
358 }
359 Err(e) => {
360 mgr_for_task.finish_failure(
361 &session_id_for_task,
362 format!("wait error: {e}"),
363 );
364 }
365 }
366 }
367
368 mgr_for_task.detach_process(&session_id_for_task);
369 });
370
371 mgr.attach_process(&session_id, handle, stdin_tx);
372}
373
374/// PTY-backed variant of spawn_auth_cli. Used for providers whose auth
375/// subcommand bails on `isatty()==0` (currently OpenClaw). Mirrors the
376/// pipes path's lifecycle: spawn → drain → confirm → finish_success or
377/// finish_failure → detach. Stdout+stderr are merged on the PTY side;
378/// `record_line` doesn't care which stream a line came from.
379///
380/// Same OAuth-success invariant as the pipes path — once `bundle_dir`
381/// (the account's own isolation dir) resolves and `confirm_authenticated`
382/// returns true, persists a standalone `SecretRef::OAuthConfigDir`
383/// account before `finish_success` — see `spawn_auth_cli`'s doc comment.
384#[allow(clippy::too_many_arguments)]
385fn spawn_auth_cli_pty(
386 mgr: Arc<crate::identity::auth_session::AuthSessionManager>,
387 mstore: Arc<Store>,
388 identity_store: Arc<Store>,
389 broker: Arc<Broker>,
390 session_id: String,
391 provider_id: String,
392 into_bundle_id: Option<String>,
393 bundle_dir: Option<String>,
394 direct_account: bool,
395 account_id: String,
396 cli_path: String,
397 auth_login_args: Vec<String>,
398 auth_check_args: Vec<String>,
399 auth_env: std::collections::HashMap<String, String>,
400 stdin_tx: tokio::sync::mpsc::Sender<String>,
401 mut stdin_rx: tokio::sync::mpsc::Receiver<String>,
402) {
403 use portable_pty::{native_pty_system, CommandBuilder, PtySize};
404
405 let mgr_for_task = mgr.clone();
406 let session_id_for_task = session_id.clone();
407 let cli_path_for_check = cli_path.clone();
408 let auth_check_args_for_check = auth_check_args.clone();
409 let auth_env_for_check = auth_env.clone();
410 let mstore_for_task = mstore.clone();
411 let identity_store_for_task = identity_store.clone();
412 let broker_for_task = broker.clone();
413 let into_bundle_id_for_task = into_bundle_id.clone();
414 let bundle_dir_for_task = bundle_dir.clone();
415 let provider_id_for_task = provider_id.clone();
416 let account_id_for_task = account_id.clone();
417
418 tracing::info!(
419 session_id = %session_id,
420 provider_id = %provider_id,
421 cli_path = %cli_path,
422 auth_login_args = ?auth_login_args,
423 "auth.spawn (PTY): launching provider CLI"
424 );
425
426 // Allocate the PTY, build the command, spawn the child, and
427 // register the PID — all synchronously, BEFORE the async drain/
428 // wait task and BEFORE `attach_process`. That way no
429 // `cancel_session` racing with the spawn can find a registered
430 // handle but a missing PID; either the PID is registered first
431 // (cancel can kill), or the early-return path fired and there
432 // is no child to kill.
433 let pty_system = native_pty_system();
434 let pair = match pty_system.openpty(PtySize {
435 rows: 24,
436 // Wide enough that the CLI never wraps its OAuth URL (`LOGIN_PTY_COLS`).
437 cols: agentmux_common::login_pty::LOGIN_PTY_COLS,
438 pixel_width: 0,
439 pixel_height: 0,
440 }) {
441 Ok(p) => p,
442 Err(e) => {
443 // PTY allocation failure means the provider's auth
444 // subprocess can't get the interactive TTY it requires
445 // (openclaw `models auth login`). Surface as the typed
446 // `AMX-AUTH-001` so the FailedBanner shows the friendly
447 // recovery hint instead of "openpty failed: <e>".
448 let mux = agentmux_common::AgentMuxError::AuthRequiresTty {
449 provider: provider_id.clone(),
450 };
451 tracing::error!(
452 target: "amx::error",
453 session_id = %session_id,
454 amx_code = %mux.code(),
455 error = %e,
456 "auth.spawn (PTY): openpty failed"
457 );
458 mgr.finish_failure(&session_id, mux.to_wire().to_string());
459 mgr.detach_process(&session_id);
460 return;
461 }
462 };
463
464 // On Windows, resolve past the npm `.cmd`/`.bat` shim: CommandBuilder on the
465 // raw shim path goes through `cmd.exe /c`, which hangs under a real ConPTY
466 // (the desktop host's login PTY found this first). `None` means the shim
467 // matched no known npm shape: fail rather than fall back to the same hang.
468 #[cfg(windows)]
469 let (spawn_program, spawn_prefix_args) = match agentmux_common::resolve_cli_spawn_target(&cli_path) {
470 Some(target) => target,
471 None => {
472 mgr.finish_failure(&session_id, format!("could not resolve .cmd/.bat shim for spawn: {cli_path}"));
473 mgr.detach_process(&session_id);
474 return;
475 }
476 };
477 #[cfg(not(windows))]
478 let (spawn_program, spawn_prefix_args) = (cli_path.clone(), Vec::<String>::new());
479 let mut cmd = CommandBuilder::new(&spawn_program);
480 // Same as the non-PTY path above: CommandBuilder seeds from vars_os(), so
481 // without this the provider CLI inherits the full instance identity.
482 crate::backend::pane_env::sanitize_external_pty_command(&mut cmd);
483 for a in &spawn_prefix_args {
484 cmd.arg(a);
485 }
486 for a in &auth_login_args {
487 cmd.arg(a);
488 }
489 for (k, v) in &auth_env {
490 cmd.env(k, v);
491 }
492 if let Ok(cwd) = std::env::current_dir() {
493 cmd.cwd(cwd);
494 }
495
496 let child = match pair.slave.spawn_command(cmd) {
497 Ok(c) => c,
498 Err(e) => {
499 tracing::error!(
500 session_id = %session_id,
501 error = %e,
502 "auth.spawn (PTY): spawn_command failed"
503 );
504 mgr.finish_failure(&session_id, format!("PTY spawn `{cli_path}` failed: {e}"));
505 mgr.detach_process(&session_id);
506 return;
507 }
508 };
509
510 if let Some(pid) = child.process_id() {
511 mgr.attach_pty_pid(&session_id, pid);
512 }
513
514 let reader = match pair.master.try_clone_reader() {
515 Ok(r) => r,
516 Err(e) => {
517 tracing::error!(session_id = %session_id, error = %e, "auth.spawn (PTY): try_clone_reader failed");
518 mgr.finish_failure(&session_id, format!("PTY reader: {e}"));
519 mgr.detach_process(&session_id);
520 return;
521 }
522 };
523 let writer = match pair.master.take_writer() {
524 Ok(w) => Arc::new(std::sync::Mutex::new(w)),
525 Err(e) => {
526 tracing::error!(session_id = %session_id, error = %e, "auth.spawn (PTY): take_writer failed");
527 mgr.finish_failure(&session_id, format!("PTY writer: {e}"));
528 mgr.detach_process(&session_id);
529 return;
530 }
531 };
532
533 let handle = tokio::spawn(async move {
534
535 // Stdin writer: forward callback URLs from the manager into
536 // the child's PTY input. portable_pty's master writer is sync;
537 // wrap each write in `block_in_place` so the blocking IO
538 // doesn't starve the tokio reactor when the PTY input buffer
539 // is full.
540 // Answers the cursor-position query (`ESC[6n`) Claude blocks on (issue
541 // #2429) through the same writer, from the drain's thread.
542 let dsr_writer = Arc::clone(&writer);
543 let answer_dsr = move || {
544 use std::io::Write;
545 if let Ok(mut w) = dsr_writer.lock() {
546 let _ = w.write_all(agentmux_common::login_pty::DSR_CURSOR_POSITION_REPLY);
547 let _ = w.flush();
548 }
549 };
550 let stdin_writer_handle = tokio::spawn(async move {
551 while let Some(line) = stdin_rx.recv().await {
552 let res = tokio::task::block_in_place(|| {
553 use std::io::Write;
554 let Ok(mut writer) = writer.lock() else {
555 return Err(std::io::Error::other("PTY writer lock poisoned"));
556 };
557 writer
558 .write_all(line.as_bytes())
559 .and_then(|_| writer.write_all(b"\n"))
560 .and_then(|_| writer.flush())
561 });
562 if res.is_err() {
563 break;
564 }
565 }
566 });
567
568 // Drain: synchronous line reader from PTY master, feeds into
569 // the same record_line matcher the pipes path uses. Runs in a
570 // blocking thread; sends parsed events through a oneshot when
571 // we hit a login-success pattern (so the async task can run
572 // confirm_authenticated without crossing the blocking thread).
573 let mgr_drain = mgr_for_task.clone();
574 let sid_drain = session_id_for_task.clone();
575 let cli_path_drain = cli_path_for_check.clone();
576 let check_args_drain = auth_check_args_for_check.clone();
577 let check_env_drain = auth_env_for_check.clone();
578 let mstore_drain = mstore_for_task.clone();
579 let identity_store_drain = identity_store_for_task.clone();
580 let broker_drain = broker_for_task.clone();
581 let into_bundle_id_drain = into_bundle_id_for_task.clone();
582 let bundle_dir_drain = bundle_dir_for_task.clone();
583 let provider_id_drain = provider_id_for_task.clone();
584 let account_id_drain = account_id_for_task.clone();
585 // Shared with the post-exit fallback block below — same pattern
586 // as the pipes path. Drain's detached `Handle::current().spawn`
587 // task does the persist + finish_success; the drain sets this
588 // atomic BEFORE returning so the outer's check sees it. The
589 // detached task may still be in flight when the outer checks
590 // (PTY drain returns on EOF, the persist task runs async) — that
591 // is fine because the OUTER would only re-persist if the atomic
592 // is false, which it never is once the drain matched the
593 // LoginSuccess pattern. Reagent P1 on #981.
594 let success_transitioned = Arc::new(AtomicBool::new(false));
595 let success_transitioned_drain = Arc::clone(&success_transitioned);
596 let drain_handle = tokio::task::spawn_blocking(move || {
597 use std::io::BufRead;
598 let mut reader = std::io::BufReader::new(agentmux_common::login_pty::DsrRespondingReader::new(
599 reader, answer_dsr,
600 ));
601 let mut line = String::new();
602 // The detached confirm/persist task's JoinHandle, populated
603 // when the drain matches LoginSuccess. Returned to outer so
604 // it can await completion before its post-exit check —
605 // without that wait, outer would race the detached and
606 // either skip too soon (if it sets the atomic eagerly) or
607 // double-persist. Reagent P1 follow-up on #981.
608 let mut maybe_detached: Option<tokio::task::JoinHandle<()>> = None;
609 loop {
610 line.clear();
611 match reader.read_line(&mut line) {
612 Ok(0) => break, // EOF
613 Ok(_) => {
614 // Same redaction policy as the pipes path —
615 // OAuth URLs / codes can be in here.
616 tracing::debug!(session_id = %sid_drain, bytes = line.len(), "auth.spawn (PTY): line");
617 let m = mgr_drain.record_line(&sid_drain, &line);
618 if maybe_detached.is_none()
619 && matches!(
620 m,
621 Some(crate::identity::auth_patterns::AuthPatternMatch::LoginSuccess { .. })
622 )
623 {
624 // Hand off to async to call confirm_authenticated.
625 // The detached task OWNS the transition —
626 // it calls finish_success on confirm-OK and
627 // finish_failure on confirm-NOT-OK, AND
628 // sets the shared atomic in BOTH cases. The
629 // drain only captures the JoinHandle here
630 // and returns it; outer awaits both drain
631 // and the inner handle before its check, so
632 // the atomic is correctly settled by then.
633 let cli = cli_path_drain.clone();
634 let args = check_args_drain.clone();
635 let env = check_env_drain.clone();
636 let mgr2 = mgr_drain.clone();
637 let sid2 = sid_drain.clone();
638 let mstore2 = mstore_drain.clone();
639 let identity_store2 = identity_store_drain.clone();
640 let broker2 = broker_drain.clone();
641 let into_bundle_id2 = into_bundle_id_drain.clone();
642 let bundle_dir2 = bundle_dir_drain.clone();
643 let provider_id2 = provider_id_drain.clone();
644 let account_id2 = account_id_drain.clone();
645 let success_for_detached = Arc::clone(&success_transitioned_drain);
646 let handle = tokio::runtime::Handle::current().spawn(async move {
647 if confirm_authenticated(&cli, &args, &env).await {
648 // Read the email BEFORE finish_success consumes the session
649 // — see AuthSessionManager::captured_email.
650 // Ended meanwhile (replaced, cancelled, timed out): a newer sign-in
651 // may own the directory, so this login is not saved.
652 if mgr2.ended(&sid2) {
653 tracing::info!(session_id = %sid2, "auth.spawn: session ended before its login was saved; not saving it");
654 return;
655 }
656 let login_email = mgr2.captured_email(&sid2);
657 let (bundle_id, account_id) = persist_oauth_success(
658 &mstore2,
659 &identity_store2,
660 &broker2,
661 direct_account,
662 &account_id2,
663 into_bundle_id2.as_deref(),
664 &provider_id2,
665 bundle_dir2.as_deref(),
666 &sid2,
667 login_email.as_deref(),
668 );
669 mgr2.finish_success(&sid2, bundle_id, account_id);
670 // Atomic set ONLY on confirm-success
671 // — mirrors the pipes path. On
672 // confirm-miss the detached does
673 // nothing (atomic stays false),
674 // outer's post-exit fallback gets
675 // another shot at confirm by the
676 // time creds are fully written.
677 // Outer awaits this handle first,
678 // so the atomic is correctly
679 // reflected by the time it checks.
680 // codex P1 follow-up on #981.
681 success_for_detached.store(true, Ordering::Release);
682 }
683 });
684 maybe_detached = Some(handle);
685 }
686 }
687 Err(_) => break,
688 }
689 }
690 maybe_detached
691 });
692
693 // Wait for the child in a blocking task. pair (master + slave)
694 // moves into the closure so its destructor runs AFTER
695 // child.wait() — ConPTY contract on Windows.
696 let wait_handle = tokio::task::spawn_blocking(move || {
697 let mut child = child;
698 let exit = child.wait();
699 drop(pair);
700 exit
701 });
702
703 let exit = wait_handle.await;
704 tracing::info!(session_id = %session_id_for_task, exit = ?exit, "auth.spawn (PTY): child exited");
705
706 // Await the drain itself, THEN its optional detached
707 // confirm/persist task. Both have to complete before the
708 // post-exit check — otherwise outer might see atomic=false
709 // (detached still confirming), run its fallback persist, AND
710 // the detached's persist also runs → the original double-
711 // persist race re-opens. Reagent P1 follow-up on #981.
712 let detached = drain_handle.await.ok().flatten();
713 if let Some(h) = detached {
714 let _ = h.await;
715 }
716 stdin_writer_handle.abort();
717
718 // Final transition fallback — some CLIs exit cleanly without
719 // emitting a login-success line that record_line recognizes.
720 //
721 // Skip the whole block when the drain already transitioned —
722 // same double-persist guard as the pipes path. Reagent P1 on
723 // #981.
724 if !success_transitioned.load(Ordering::Acquire) {
725 match exit {
726 Ok(Ok(s)) if s.success() => {
727 if confirm_authenticated(
728 &cli_path_for_check,
729 &auth_check_args_for_check,
730 &auth_env_for_check,
731 )
732 .await
733 {
734 // Read the email BEFORE finish_success consumes the session
735 // — see AuthSessionManager::captured_email.
736 // Ended meanwhile (replaced, cancelled, timed out): a newer sign-in
737 // may own the directory, so this login is not saved.
738 if mgr_for_task.ended(&session_id_for_task) {
739 tracing::info!(session_id = %session_id_for_task, "auth.spawn: session ended before its login was saved; not saving it");
740 return;
741 }
742 let login_email = mgr_for_task.captured_email(&session_id_for_task);
743 let (bundle_id, account_id) = persist_oauth_success(
744 &mstore_for_task,
745 &identity_store_for_task,
746 &broker_for_task,
747 direct_account,
748 &account_id_for_task,
749 into_bundle_id_for_task.as_deref(),
750 &provider_id_for_task,
751 bundle_dir_for_task.as_deref(),
752 &session_id_for_task,
753 login_email.as_deref(),
754 );
755 mgr_for_task.finish_success(&session_id_for_task, bundle_id, account_id);
756 } else {
757 mgr_for_task.finish_failure(
758 &session_id_for_task,
759 "CLI exited cleanly but auth-check still failed".to_string(),
760 );
761 }
762 }
763 Ok(Ok(s)) => {
764 mgr_for_task.finish_failure(
765 &session_id_for_task,
766 format!("CLI exited with code {:?}", s.exit_code()),
767 );
768 }
769 Ok(Err(e)) => {
770 mgr_for_task.finish_failure(
771 &session_id_for_task,
772 format!("PTY wait error: {e}"),
773 );
774 }
775 Err(e) => {
776 mgr_for_task.finish_failure(
777 &session_id_for_task,
778 format!("PTY wait task join error: {e}"),
779 );
780 }
781 }
782 }
783
784 mgr_for_task.detach_process(&session_id_for_task);
785 });
786
787 mgr.attach_process(&session_id, handle, stdin_tx);
788}
789
790/// How long a provider's auth-status check may take (`confirm_authenticated`).
791const AUTH_CHECK_TIMEOUT_SECS: u64 = 30;
792
793/// Run the provider's auth-check subcommand and return true if it
794/// exits 0. Failure modes (binary missing, network error, etc.) are
795/// all treated as "not authenticated" — the caller will then either
796/// keep waiting (drain task loop) or transition to Failed (exit
797/// fallback).
798async fn confirm_authenticated(
799 cli_path: &str,
800 args: &[String],
801 env: &std::collections::HashMap<String, String>,
802) -> bool {
803 use std::process::Stdio;
804 // Through the shim parser (`make_cli_cmd`): a raw npm `.cmd` shim spawned
805 // with piped stdio can drop its arguments or never run the CLI, which
806 // would report a good login as failed. It also sets CREATE_NO_WINDOW, so
807 // the probe, polled during an auth flow, flashes no console.
808 let mut c = agentmux_common::make_cli_cmd(cli_path);
809 // The third spawn of this same provider CLI in this file — the auth-check
810 // poll driven from the OAuth drain loop. Same strict policy as the other
811 // two. (ReAgent P0, round 4, on #3326.)
812 crate::backend::pane_env::sanitize_external_command(&mut c);
813 c.args(args)
814 .envs(env)
815 .stdin(Stdio::null())
816 .stdout(Stdio::null())
817 .stderr(Stdio::null())
818 .kill_on_drop(true);
819 // A status check that hangs counts as not logged in (and is killed on drop).
820 match tokio::time::timeout(std::time::Duration::from_secs(AUTH_CHECK_TIMEOUT_SECS), c.status()).await {
821 Ok(Ok(s)) => s.success(),
822 _ => false,
823 }
824}