diff --git a/crates/agent-vm/src/intercept_hook.rs b/crates/agent-vm/src/intercept_hook.rs index 83d1433..cbf775e 100644 --- a/crates/agent-vm/src/intercept_hook.rs +++ b/crates/agent-vm/src/intercept_hook.rs @@ -692,7 +692,15 @@ fn write_response(bytes: &[u8]) -> Result<()> { /// testable without a live `claude` session or a real expiring token /// (PLAN.md A1). fn refresh_anthropic(state_dir: &Path) -> Result> { - trigger_host_refresh("claude", &["-p", "hi", "--model", "sonnet"])?; + // Single-flight: serialize host-side rotations for this provider so + // two racing in-guest refreshes don't each spawn `claude -p`. The + // first waiter rotates; the second, on acquiring the lock, finds the + // token file freshly rewritten and skips its own host CLI. The lock + // is Anthropic-specific so a concurrent OpenAI refresh isn't blocked. + let _flight = RefreshLock::acquire(state_dir, secrets::REFRESH_LOCK_ANTHROPIC)?; + if !token_recently_rotated(&secrets::anthropic_token_path(state_dir)) { + trigger_host_refresh("claude", &["-p", "hi", "--model", "sonnet"])?; + } let host_path = host_claude_creds_path().context("HOME not set")?; let raw = std::fs::read_to_string(&host_path) @@ -753,15 +761,22 @@ fn rotate_anthropic(state_dir: &Path, host_creds_json: &str) -> Result> /// hand the contents to [`rotate_openai`] for the testable rotation /// logic. fn refresh_openai(state_dir: &Path) -> Result> { - trigger_host_refresh( - "codex", - &[ - "exec", - "--skip-git-repo-check", - "--dangerously-bypass-approvals-and-sandbox", - "Reply with OK", - ], - )?; + // Single-flight (see `refresh_anthropic`): serialize host rotations + // and skip the `codex exec` if the token file was just rewritten by + // the launcher that held the lock before us. OpenAI-specific lock so + // an in-flight Anthropic refresh doesn't serialize against this one. + let _flight = RefreshLock::acquire(state_dir, secrets::REFRESH_LOCK_OPENAI)?; + if !token_recently_rotated(&secrets::openai_token_path(state_dir)) { + trigger_host_refresh( + "codex", + &[ + "exec", + "--skip-git-repo-check", + "--dangerously-bypass-approvals-and-sandbox", + "Reply with OK", + ], + )?; + } let host_path = host_codex_auth_path().context("HOME not set")?; let raw = std::fs::read_to_string(&host_path) @@ -803,13 +818,301 @@ fn rotate_openai(state_dir: &Path, host_auth_json: &str) -> Result> { Ok(http_200_json(&serde_json::to_vec(&body)?)) } +/// How recently a token file must have been rewritten for the +/// single-flight waiter to trust it and skip its own host rotation. +/// +/// Tied to [`HOST_REFRESH_TIMEOUT`]: a host `claude -p` / `codex exec` +/// is *allowed* to take up to that long, so a fixed 10 s window would +/// silently no-op in exactly the slow-rotation case the optimization is +/// meant to help — the holder finishes after, say, 25 s, and the waiter +/// then sees an mtime older than 10 s and redundantly re-runs the host +/// CLI even though the token it would read is current. Matching the +/// window to the rotation budget means a just-completed slow rotation is +/// still recognized as fresh. This is safe: the file's mtime is only +/// bumped by an actual successful rotation write, and host access tokens +/// live far longer than 90 s, so we never serve a stale token. A small +/// slack is added so a waiter that wakes slightly after the holder +/// returns still counts the rotation as fresh. +const REFRESH_FRESHNESS_WINDOW: std::time::Duration = + HOST_REFRESH_TIMEOUT.saturating_add(std::time::Duration::from_secs(5)); + +/// True if `path` exists and was modified within +/// [`REFRESH_FRESHNESS_WINDOW`]. Used by the second single-flight +/// waiter to decide it can re-read the just-rotated token file instead +/// of spawning another host CLI. Any error (missing file, clock skew +/// making mtime appear in the future) conservatively returns `false` +/// so we fall back to actually refreshing. +fn token_recently_rotated(path: &Path) -> bool { + let Ok(meta) = std::fs::metadata(path) else { + return false; + }; + let Ok(modified) = meta.modified() else { + return false; + }; + match modified.elapsed() { + Ok(age) => age <= REFRESH_FRESHNESS_WINDOW, + Err(_) => false, + } +} + +/// Advisory cross-process lock serializing host-side OAuth refreshes +/// for one provider within a single project. Held for the duration of +/// one `refresh_anthropic` / `refresh_openai` call so two in-guest +/// agents (or two launchers) racing the *same* provider's token +/// rotation don't each spawn a host `claude -p` / `codex exec`. +/// +/// The lock is keyed per provider (see [`secrets::refresh_lock_path_for`]): +/// an Anthropic rotation and an OpenAI rotation touch independent host +/// artifacts, so they hold different lock files and may run +/// concurrently — only same-provider refreshes serialize. +/// +/// Uses a non-blocking `flock(LOCK_EX|LOCK_NB)` polled with a deadline +/// — already available via the `libc` dependency, so no new crate. The +/// lock is associated with the open file description and released +/// automatically when the fd is closed on `Drop` (or if the process +/// dies), so a crashed refresh can't wedge future rotations. +struct RefreshLock { + /// `Some` when we actually hold the flock; `None` when [`acquire`] + /// timed out waiting for a wedged-but-live holder and we degraded + /// to proceeding without serialization (see [`acquire`]). The fd is + /// still kept open in that case so `Drop` is uniform, but no + /// `LOCK_UN` is issued. + file: Option, +} + +impl RefreshLock { + /// Acquire the per-provider refresh lock named `lock_name` (e.g. + /// [`secrets::REFRESH_LOCK_ANTHROPIC`]). + /// + /// Bounded wait: a live-but-wedged holder must not block a waiter + /// indefinitely, which would reintroduce the unbounded stall the + /// [`HOST_REFRESH_TIMEOUT`] cap was added to prevent (review #8). + /// We poll `LOCK_EX|LOCK_NB` until the ceiling, then degrade to + /// proceeding *without* the lock — i.e. the pre-feature behavior of + /// just refreshing. That can cost a redundant host CLI spawn in the + /// rare wedged-holder case, but keeps the whole refresh path time + /// bounded, which matters more. + fn acquire(state_dir: &Path, lock_name: &str) -> Result { + Self::acquire_with_ceiling(state_dir, lock_name, HOST_REFRESH_TIMEOUT) + } + + fn acquire_with_ceiling( + state_dir: &Path, + lock_name: &str, + ceiling: std::time::Duration, + ) -> Result { + let path = secrets::refresh_lock_path_for(state_dir, lock_name); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("creating secrets dir {}", parent.display()))?; + } + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .with_context(|| format!("opening refresh lock {}", path.display()))?; + use std::os::unix::io::AsRawFd as _; + use std::time::Instant; + let fd = file.as_raw_fd(); + let start = Instant::now(); + // Poll interval is small relative to a host rotation (seconds); + // the extra wakeups over a ~90 s ceiling are negligible. + let poll = std::time::Duration::from_millis(50); + loop { + let rc = unsafe { libc::flock(fd, libc::LOCK_EX | libc::LOCK_NB) }; + if rc == 0 { + return Ok(Self { file: Some(file) }); + } + let err = std::io::Error::last_os_error(); + match err.raw_os_error() { + Some(libc::EINTR) => continue, + // Held by someone else — wait and retry until the ceiling. + Some(libc::EWOULDBLOCK) => { + if start.elapsed() >= ceiling { + // Degrade to no-lock rather than block forever on + // a wedged-but-live holder. `file: None` so Drop + // issues no LOCK_UN we never took. + tracing::warn!( + lock = %path.display(), + "refresh lock contended past {}s; proceeding without single-flight", + ceiling.as_secs(), + ); + return Ok(Self { file: None }); + } + std::thread::sleep(poll); + continue; + } + _ => { + return Err(anyhow::Error::new(err) + .context(format!("flock(LOCK_EX|LOCK_NB) on {}", path.display()))); + } + } + } + } +} + +impl Drop for RefreshLock { + fn drop(&mut self) { + use std::os::unix::io::AsRawFd as _; + // Only unlock if we actually acquired it; a timed-out acquire + // never took the lock, so issuing LOCK_UN would be wrong (and + // could release a lock another fd in this process holds, though + // that doesn't happen here). + if let Some(file) = &self.file { + unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_UN) }; + } + } +} + +#[cfg(test)] +mod refresh_lock_tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[test] + fn token_recently_rotated_window() { + let dir = tempfile::tempdir().unwrap(); + let f = dir.path().join("anthropic"); + // Missing file -> not fresh. + assert!(!token_recently_rotated(&f)); + // Just-written file -> fresh. + std::fs::write(&f, b"tok").unwrap(); + assert!(token_recently_rotated(&f)); + } + + /// Two threads contending the same project lock must run their + /// critical sections one at a time. We track the number of holders + /// inside the locked region and assert it never exceeds one. + #[test] + fn contending_threads_serialize() { + let dir = tempfile::tempdir().unwrap(); + let state = dir.path().join("proj"); + std::fs::create_dir_all(&state).unwrap(); + + let in_section = Arc::new(AtomicUsize::new(0)); + let max_concurrent = Arc::new(AtomicUsize::new(0)); + let acquisitions = Arc::new(AtomicUsize::new(0)); + + let spawn = |state: std::path::PathBuf, + in_section: Arc, + max_concurrent: Arc, + acquisitions: Arc| { + std::thread::spawn(move || { + for _ in 0..20 { + let _guard = RefreshLock::acquire(&state, secrets::REFRESH_LOCK_ANTHROPIC) + .expect("acquire"); + acquisitions.fetch_add(1, Ordering::SeqCst); + let now = in_section.fetch_add(1, Ordering::SeqCst) + 1; + // Record the peak observed concurrency. + max_concurrent.fetch_max(now, Ordering::SeqCst); + // Hold briefly to widen the race window. + std::thread::sleep(std::time::Duration::from_millis(1)); + in_section.fetch_sub(1, Ordering::SeqCst); + } + }) + }; + + let t1 = spawn( + state.clone(), + in_section.clone(), + max_concurrent.clone(), + acquisitions.clone(), + ); + let t2 = spawn( + state.clone(), + in_section.clone(), + max_concurrent.clone(), + acquisitions.clone(), + ); + t1.join().unwrap(); + t2.join().unwrap(); + + assert_eq!(acquisitions.load(Ordering::SeqCst), 40); + assert_eq!( + max_concurrent.load(Ordering::SeqCst), + 1, + "flock failed to serialize: two holders entered the critical section at once" + ); + } + + /// Different providers use different lock files, so a holder of the + /// Anthropic lock must not block an OpenAI acquire (and vice versa). + #[test] + fn distinct_providers_do_not_contend() { + let dir = tempfile::tempdir().unwrap(); + let state = dir.path().join("proj"); + std::fs::create_dir_all(&state).unwrap(); + + let anthropic = RefreshLock::acquire(&state, secrets::REFRESH_LOCK_ANTHROPIC) + .expect("anthropic acquire"); + // Holding the Anthropic lock, an OpenAI acquire must succeed + // immediately (not time out, not degrade) — it's a different file. + let openai = RefreshLock::acquire_with_ceiling( + &state, + secrets::REFRESH_LOCK_OPENAI, + std::time::Duration::from_millis(50), + ) + .expect("openai acquire"); + // Both genuinely hold their locks. + assert!(openai.file.is_some(), "openai should hold its own lock"); + drop(anthropic); + drop(openai); + } + + /// A live-but-wedged holder must not block a waiter past the + /// ceiling: the second acquire returns within the bound and degrades + /// to the no-lock state instead of hanging forever. + #[test] + fn bounded_wait_degrades_when_holder_wedged() { + let dir = tempfile::tempdir().unwrap(); + let state = dir.path().join("proj"); + std::fs::create_dir_all(&state).unwrap(); + + // First holder keeps the lock for the whole test. + let _held = RefreshLock::acquire(&state, secrets::REFRESH_LOCK_ANTHROPIC) + .expect("first acquire"); + + let ceiling = std::time::Duration::from_millis(200); + let start = std::time::Instant::now(); + let second = RefreshLock::acquire_with_ceiling( + &state, + secrets::REFRESH_LOCK_ANTHROPIC, + ceiling, + ) + .expect("second acquire returns Ok (degraded)"); + let waited = start.elapsed(); + + // Returned within a small multiple of the ceiling (not blocked + // indefinitely), and degraded to no-lock. + assert!( + waited < ceiling * 4, + "acquire blocked {waited:?}, expected ~{ceiling:?}" + ); + assert!( + second.file.is_none(), + "contended acquire past ceiling must degrade to the no-lock state" + ); + } +} + +/// Bound on how long we'll wait for a host `claude -p` / `codex exec` +/// to drive a token rotation. A hung host CLI must not keep the in-VM +/// agent's OAuth refresh waiting indefinitely (review #8). 90 s is +/// enough for normal claude/codex round-trips and small enough to +/// surface a problem before the guest agent's own timeout fires. +/// +/// Shared so the single-flight lock-wait ceiling +/// ([`RefreshLock::acquire`]) and the freshness window +/// ([`REFRESH_FRESHNESS_WINDOW`]) stay tied to the actual rotation +/// budget rather than drifting from it. +const HOST_REFRESH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(90); + fn trigger_host_refresh(cmd: &str, args: &[&str]) -> Result<()> { - // Bounded wait so a hung host CLI doesn't keep the in-VM agent's - // OAuth refresh waiting indefinitely (review #8). 90 s is enough - // for normal claude/codex round-trips and small enough to surface - // a problem before the guest agent's own timeout fires. use std::time::{Duration, Instant}; - const TIMEOUT: Duration = Duration::from_secs(90); + const TIMEOUT: Duration = HOST_REFRESH_TIMEOUT; let mut child = Command::new(cmd) .args(args) diff --git a/crates/agent-vm/src/secrets.rs b/crates/agent-vm/src/secrets.rs index a57502d..2c759d5 100644 --- a/crates/agent-vm/src/secrets.rs +++ b/crates/agent-vm/src/secrets.rs @@ -174,6 +174,64 @@ pub fn gh_token_path(state_dir: &Path) -> PathBuf { host_secret_dir(state_dir).join("gh") } +/// Per-provider advisory lock file serializing host-side OAuth refreshes +/// for one provider within a single project. The in-VM agent's +/// `_intercept-hook` acquires an exclusive `flock` on this path before +/// spawning a host `claude -p` / `codex exec` to drive a token rotation, +/// so two racing in-guest refreshes of the *same* provider don't each +/// launch their own host CLI. +/// +/// Keyed per provider (`name` is e.g. [`REFRESH_LOCK_ANTHROPIC`]) so a +/// concurrent Anthropic and OpenAI in-guest refresh don't serialize +/// against each other — they rotate independent host credential files +/// and write distinct token files, so there is no shared state to guard +/// across providers. Lives in the host-only [`host_secret_dir`] (never +/// bind-mounted into the guest), alongside the token files the refresh +/// rewrites. +/// +/// Note: the launcher's [`refresh`] uses a *different*, single shared +/// lock ([`ProjectRefreshLock`]) because it does read-modify-write on +/// per-project state files shared across all providers (`claude.json`, +/// `claude/settings.json`, `opencode-config/opencode.json`), so its +/// critical section genuinely spans every provider. +pub fn refresh_lock_path_for(state_dir: &Path, name: &str) -> PathBuf { + host_secret_dir(state_dir).join(name) +} + +/// Lock basename for the Anthropic in-guest refresh single-flight. +pub const REFRESH_LOCK_ANTHROPIC: &str = ".refresh.anthropic.lock"; +/// Lock basename for the OpenAI in-guest refresh single-flight. +pub const REFRESH_LOCK_OPENAI: &str = ".refresh.openai.lock"; + +#[cfg(test)] +mod refresh_lock_tests { + use super::*; + + #[test] + fn per_provider_lock_paths_are_distinct_and_in_secret_dir() { + let state = Path::new("/home/u/.cache/agent-vm/abc123"); + let anthropic = refresh_lock_path_for(state, REFRESH_LOCK_ANTHROPIC); + let openai = refresh_lock_path_for(state, REFRESH_LOCK_OPENAI); + // Two providers must not share a lock file. + assert_ne!(anthropic, openai); + // Expected concrete paths in the sibling `.secrets/` dir. + assert_eq!( + anthropic, + Path::new("/home/u/.cache/agent-vm/abc123.secrets/.refresh.anthropic.lock") + ); + assert_eq!( + openai, + Path::new("/home/u/.cache/agent-vm/abc123.secrets/.refresh.openai.lock") + ); + // Both live in the host-only secrets dir, never under state_dir + // (which is bind-mounted into the guest). + assert_eq!(anthropic.parent(), anthropic_token_path(state).parent()); + assert_eq!(openai.parent(), anthropic_token_path(state).parent()); + assert!(!anthropic.starts_with(state)); + assert!(!openai.starts_with(state)); + } +} + /// OpenCode reuses the same OpenAI access token file: both Codex and /// OpenCode hit api.openai.com / chatgpt.com and the proxy substitutes /// each provider's distinct placeholder string for the same real