From 4b1012056c3c6ba6c4369d9593a2a6764e6d14f3 Mon Sep 17 00:00:00 2001 From: iris Date: Mon, 9 Nov 2026 22:31:51 +0100 Subject: [PATCH] hive-agent: order a stored login by when the store wrote it, not by its number A delete removes the leaf's metadata, so the KV number restarts at 1. A login made right after a delete, before the next poll, was version 1 again and looked no newer than the recorded 1, so it never installed. The version file now records the store's write time next to the number and a later write installs whatever its number. A record from before the time was kept falls back to the number and gets its time when the same number is read again. --- docs/2-handbook/swarm/claude-login.md | 18 +- hive-agent/src/claude_credentials.rs | 287 +++++++++++++++++++++----- swarm-secret-client/src/client.rs | 33 ++- 3 files changed, 274 insertions(+), 64 deletions(-) diff --git a/docs/2-handbook/swarm/claude-login.md b/docs/2-handbook/swarm/claude-login.md index b40745daf..90176f444 100644 --- a/docs/2-handbook/swarm/claude-login.md +++ b/docs/2-handbook/swarm/claude-login.md @@ -54,20 +54,24 @@ reads the credential. and every version of it, not just the newest: an older version is still a session. An agent that already holds a session logs out too, the next time it reads the store: within 5 minutes, or when a paused agent resumes (see -[the harness's rule](#why-the-harness-takes-only-a-newer-version)). +[the harness's rule](#why-the-harness-takes-only-a-newer-login)). -## Why the harness takes only a newer version +## Why the harness takes only a newer login `claude` rotates the refresh token in `.credentials.json` as it uses the session, and the agent can't write the store. After the first refresh the stored copy is stale. A harness that installed it whenever the bytes differed would replace the agent's live session with a stale copy and log the agent -out. The harness therefore compares the store's KV version with the last -version it installed (`claude-credentials-version` under the harness dir): +out. The harness therefore compares the store's version with the last one it +installed (`claude-credentials-version` under the harness dir), by the time +the store wrote each. A delete restarts the KV number but not the time, so a +login right after a delete, version 1 again, still counts as newer. A record +from before the harness kept the time falls back to the number, and gets its +time the first time it reads the same number. -- it installs a newer version, that is, a new login; -- it leaves an equal or older version alone, even when the agent needs a - login again, so it never puts back a session that has gone bad; +- it installs a newer write, that is, a new login; +- it leaves an equal or older one alone, even when the agent needs a login + again, so it never puts back a session that has gone bad; - it logs the agent out when the store holds nothing although a version was installed: the operator deleted the stored login. A read that failed never does it, since an unreachable store says nothing about the login. The diff --git a/hive-agent/src/claude_credentials.rs b/hive-agent/src/claude_credentials.rs index 376e7aa43..2ae83269b 100644 --- a/hive-agent/src/claude_credentials.rs +++ b/hive-agent/src/claude_credentials.rs @@ -3,7 +3,7 @@ //! certificate while it waits for a login ([`crate::login::wait_for_login`]), //! and every few minutes while it runs ([`crate::login_watch`]). //! -//! ⚠️ **Installed only when the store's version is newer than the last one +//! ⚠️ **Installed only when the store's write is newer than the last one //! installed here**, never because the bytes differ. `claude` rotates the //! refresh token in `~/.claude/.credentials.json`, so after its first refresh //! the stored copy is stale; putting it back would end the rotated session. @@ -38,11 +38,80 @@ const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10); /// The file `claude` reads its session from. const CREDENTIALS_FILE: &str = ".credentials.json"; -/// Whether a stored `version` replaces what is installed: only a version -/// newer than the last one installed, or any when none has been. +/// A stored version's identity: its KV number, and when the store wrote it, in +/// ms since the epoch. The number restarts when the operator deletes the stored +/// login, so the time is what orders a new login after the one before the +/// delete. A record from before the time was kept has none. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Stamp { + pub version: u64, + pub written_ms: Option, +} + +impl Stamp { + /// What the version file holds: `N`, or `N T` once the time is known. + fn encode(self) -> String { + match self.written_ms { + Some(ms) => format!("{} {ms}", self.version), + None => self.version.to_string(), + } + } + + fn parse(s: &str) -> Option { + let mut parts = s.split_whitespace(); + let version = parts.next()?.parse().ok()?; + let written_ms = match parts.next() { + Some(ms) => Some(ms.parse().ok()?), + None => None, + }; + parts.next().is_none().then_some(Self { + version, + written_ms, + }) + } + + /// From the store's version: its number and RFC 3339 time. A time that + /// does not parse is no time. + fn of(version: &swarm_secret_client::client::Version) -> Self { + Self { + version: version.number, + written_ms: chrono::DateTime::parse_from_rfc3339(&version.created_time) + .ok() + .map(|t| t.timestamp_millis()), + } + } +} + +/// What a stored version is to what is installed. +#[derive(Debug, PartialEq, Eq)] +pub enum Verdict { + /// Replace the installed login with it. + Install, + /// Leave the installed login alone. + Leave, + /// The same login the record names, which has no time yet: leave the login + /// and record the time. + Upgrade, +} + +/// Whether a stored version replaces what is installed. Ordered by the time +/// the store wrote it when both have one, so a login made after a delete +/// installs though its number restarted; by number otherwise, where only a +/// larger one is newer. A record without a time next to the same number is +/// ambiguous (the same login, or the first after a delete), so it is left and +/// gets its time. #[must_use] -pub fn is_newer(stored: u64, installed: Option) -> bool { - installed.is_none_or(|last| stored > last) +pub fn judge(stored: Stamp, installed: Option) -> Verdict { + let Some(last) = installed else { + return Verdict::Install; + }; + match (stored.written_ms, last.written_ms) { + (Some(s), Some(l)) if s > l => Verdict::Install, + (Some(_), None) if stored.version == last.version => Verdict::Upgrade, + (Some(_), None) => Verdict::Install, + (None, _) if stored.version > last.version => Verdict::Install, + _ => Verdict::Leave, + } } /// What one [`Pickup::check`] did. @@ -65,7 +134,7 @@ pub enum Outcome { pub trait Source: Send + Sync { /// The stored credentials and their version, or `None` when nothing is /// stored. - fn fetch(&self) -> impl Future>> + Send; + fn fetch(&self) -> impl Future>> + Send; } /// The agent's own leaf, read under its own certificate. @@ -76,15 +145,16 @@ pub struct Store { } impl Source for Store { - async fn fetch(&self) -> anyhow::Result> { + async fn fetch(&self) -> anyhow::Result> { let read = async { let store = SecretStore::connect(&self.settings, &self.role, DEFAULT_CERT_MOUNT) .await .context("logging in to the swarm secret store as this agent")?; - store + let read = store .read_versioned(&self.path) .await - .with_context(|| format!("reading {}", self.path)) + .with_context(|| format!("reading {}", self.path))?; + Ok::<_, anyhow::Error>(read.map(|(value, version)| (value, Stamp::of(&version)))) }; tokio::time::timeout(STORE_READ_TIMEOUT, read) .await @@ -129,26 +199,37 @@ impl Pickup { /// The store read failed, or writing the credentials or the version did. pub async fn check(&self) -> anyhow::Result { let Some((credentials, stored)) = self.source.fetch().await? else { - return Ok(match read_version(&self.version_file)? { - Some(installed) => Outcome::Deleted { installed }, + return Ok(match read_stamp(&self.version_file)? { + Some(installed) => Outcome::Deleted { + installed: installed.version, + }, None => Outcome::NothingStored, }); }; - let installed = read_version(&self.version_file)?; - if !is_newer(stored, installed) { - return Ok(Outcome::NotNewer { - stored, - installed: installed.unwrap_or_default(), - }); + let installed = read_stamp(&self.version_file)?; + match judge(stored, installed) { + Verdict::Install => {} + verdict => { + if verdict == Verdict::Upgrade { + write_private(&self.version_file, stored.encode().as_bytes()) + .context("recording when the installed credentials were stored")?; + } + return Ok(Outcome::NotNewer { + stored: stored.version, + installed: installed.map_or(0, |s| s.version), + }); + } } write_private( &self.claude_dir.join(CREDENTIALS_FILE), credentials.value.as_bytes(), ) .context("installing the stored claude credentials")?; - write_private(&self.version_file, stored.to_string().as_bytes()) + write_private(&self.version_file, stored.encode().as_bytes()) .context("recording the installed credentials version")?; - Ok(Outcome::Installed { version: stored }) + Ok(Outcome::Installed { + version: stored.version, + }) } /// Forget the installed version, once the deletion it showed has been @@ -166,11 +247,9 @@ impl Pickup { } /// The version recorded in `file`, or `None` when there is no file. -fn read_version(file: &Path) -> anyhow::Result> { +fn read_stamp(file: &Path) -> anyhow::Result> { match std::fs::read_to_string(file) { - Ok(s) => s - .trim() - .parse() + Ok(s) => Stamp::parse(&s) .map(Some) .with_context(|| format!("{} does not hold a version", file.display())), Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), @@ -204,16 +283,16 @@ mod tests { use super::*; - struct Fake(Mutex>); + struct Fake(Mutex>); impl Source for Fake { - async fn fetch(&self) -> anyhow::Result> { + async fn fetch(&self) -> anyhow::Result> { Ok(self .0 .lock() .expect("lock") .clone() - .map(|(value, v)| (Credentials { value }, v))) + .map(|(value, stamp)| (Credentials { value }, stamp))) } } @@ -235,27 +314,104 @@ mod tests { Dir(d) } + fn stamp(version: u64, written_ms: Option) -> Stamp { + Stamp { + version, + written_ms, + } + } + + /// A pickup over a stored login that carries no write time. fn pickup(d: &Dir, stored: Option<(&str, u64)>) -> Pickup { + pickup_stamped(d, stored.map(|(v, n)| (v, stamp(n, None)))) + } + + fn pickup_stamped(d: &Dir, stored: Option<(&str, Stamp)>) -> Pickup { Pickup::new( - Fake(Mutex::new(stored.map(|(v, n)| (v.to_owned(), n)))), + Fake(Mutex::new(stored.map(|(v, s)| (v.to_owned(), s)))), d.0.join("claude"), d.0.join(VERSION_FILE), ) } + fn set(p: &Pickup, stored: Option<(&str, Stamp)>) { + *p.source.0.lock().expect("lock") = stored.map(|(v, s)| (v.to_owned(), s)); + } + fn installed(d: &Dir) -> Option { std::fs::read_to_string(d.0.join("claude").join(CREDENTIALS_FILE)).ok() } + fn recorded(d: &Dir) -> Option { + read_stamp(&d.0.join(VERSION_FILE)).expect("read") + } + #[test] - fn only_a_newer_version_replaces_the_installed_one() { - assert!(is_newer(1, None)); - assert!(is_newer(3, Some(2))); - assert!( - !is_newer(2, Some(2)), - "an equal version was installed again" + fn without_times_only_a_larger_number_replaces_the_installed_one() { + assert_eq!(judge(stamp(1, None), None), Verdict::Install); + assert_eq!( + judge(stamp(3, None), Some(stamp(2, None))), + Verdict::Install ); - assert!(!is_newer(1, Some(2)), "an older version was installed"); + assert_eq!(judge(stamp(2, None), Some(stamp(2, None))), Verdict::Leave); + assert_eq!(judge(stamp(1, None), Some(stamp(2, None))), Verdict::Leave); + } + + #[test] + fn with_times_the_later_write_wins_whatever_its_number() { + let before = Some(stamp(1, Some(1_000))); + assert_eq!( + judge(stamp(1, Some(2_000)), before), + Verdict::Install, + "a login after a delete restarts at 1 and is still newer" + ); + assert_eq!(judge(stamp(1, Some(1_000)), before), Verdict::Leave); + assert_eq!( + judge(stamp(5, Some(900)), before), + Verdict::Leave, + "an older write is not newer for its larger number" + ); + } + + #[test] + fn a_record_without_a_time_gets_one_when_the_number_matches() { + let legacy = Some(stamp(1, None)); + assert_eq!(judge(stamp(1, Some(5)), legacy), Verdict::Upgrade); + assert_eq!(judge(stamp(2, Some(5)), legacy), Verdict::Install); + assert_eq!( + judge(stamp(1, Some(5)), Some(stamp(3, None))), + Verdict::Install, + "a smaller number than recorded can only be a restarted counter" + ); + } + + #[test] + fn the_version_file_round_trips_and_reads_the_old_form() { + for s in [stamp(4, None), stamp(1, Some(1_731_000_000_123))] { + assert_eq!(Stamp::parse(&s.encode()), Some(s)); + } + assert_eq!(Stamp::parse("7\n"), Some(stamp(7, None))); + assert_eq!(Stamp::parse("7 x"), None); + assert_eq!(Stamp::parse("7 8 9"), None); + assert_eq!(Stamp::parse(""), None); + } + + #[test] + fn the_stores_rfc3339_time_is_read_to_ms() { + let v = swarm_secret_client::client::Version { + number: 3, + created_time: "2026-11-09T20:00:00.123456789Z".to_owned(), + }; + assert_eq!( + Stamp::of(&v), + stamp(3, Some(1_794_254_400_123)), + "nanoseconds are cut to ms" + ); + let v = swarm_secret_client::client::Version { + number: 3, + created_time: "not a time".to_owned(), + }; + assert_eq!(Stamp::of(&v), stamp(3, None)); } #[tokio::test] @@ -274,16 +430,13 @@ mod tests { .permissions() .mode(); assert_eq!(mode & 0o777, 0o600); - assert_eq!( - read_version(&d.0.join(VERSION_FILE)).expect("read"), - Some(1) - ); + assert_eq!(recorded(&d), Some(stamp(1, None))); } #[tokio::test] async fn an_equal_or_older_version_leaves_the_rotated_session_alone() { let d = dir("stale"); - let p = pickup(&d, Some(("{\"v\":2}", 2))); + let p = pickup_stamped(&d, Some(("{\"v\":2}", stamp(2, Some(2_000))))); p.check().await.expect("first install"); // claude refreshes the session in place. std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("refresh"); @@ -294,7 +447,7 @@ mod tests { installed: 2 } ); - *p.source.0.lock().expect("lock") = Some(("{\"v\":1}".to_owned(), 1)); + set(&p, Some(("{\"v\":1}", stamp(1, Some(1_000))))); assert_eq!( p.check().await.expect("check"), Outcome::NotNewer { @@ -308,10 +461,10 @@ mod tests { #[tokio::test] async fn a_relogin_is_installed_over_the_rotated_session() { let d = dir("relogin"); - let p = pickup(&d, Some(("{\"v\":1}", 1))); + let p = pickup_stamped(&d, Some(("{\"v\":1}", stamp(1, Some(1_000))))); p.check().await.expect("first install"); std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("refresh"); - *p.source.0.lock().expect("lock") = Some(("{\"v\":2}".to_owned(), 2)); + set(&p, Some(("{\"v\":2}", stamp(2, Some(2_000))))); assert_eq!( p.check().await.expect("check"), Outcome::Installed { version: 2 } @@ -325,15 +478,15 @@ mod tests { let p = pickup(&d, None); assert_eq!(p.check().await.expect("check"), Outcome::NothingStored); assert_eq!(installed(&d), None); - assert_eq!(read_version(&d.0.join(VERSION_FILE)).expect("read"), None); + assert_eq!(recorded(&d), None); } #[tokio::test] async fn a_login_deleted_after_an_install_is_reported_and_forgotten_on_request() { let d = dir("deleted"); - let p = pickup(&d, Some(("{\"v\":1}", 1))); + let p = pickup_stamped(&d, Some(("{\"v\":1}", stamp(1, Some(1_000))))); p.check().await.expect("install"); - *p.source.0.lock().expect("lock") = None; + set(&p, None); assert_eq!( p.check().await.expect("check"), Outcome::Deleted { installed: 1 } @@ -346,7 +499,7 @@ mod tests { p.forget_installed().expect("forget"); p.forget_installed().expect("forgetting twice is fine"); assert_eq!(p.check().await.expect("check"), Outcome::NothingStored); - *p.source.0.lock().expect("lock") = Some(("{\"v\":1}".to_owned(), 1)); + set(&p, Some(("{\"v\":1}", stamp(1, Some(3_000))))); assert_eq!( p.check().await.expect("a new login"), Outcome::Installed { version: 1 }, @@ -354,10 +507,45 @@ mod tests { ); } + #[tokio::test] + async fn a_login_made_between_two_polls_after_a_delete_still_installs() { + // The agent installed version 1. The operator deletes it and logs in + // again before the next poll, so the store never answers "nothing + // stored": it answers version 1 again, written later. + let d = dir("between"); + let p = pickup_stamped(&d, Some(("{\"old\":1}", stamp(1, Some(1_000))))); + p.check().await.expect("install"); + std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("refresh"); + set(&p, Some(("{\"new\":1}", stamp(1, Some(9_000))))); + assert_eq!( + p.check().await.expect("check"), + Outcome::Installed { version: 1 } + ); + assert_eq!(installed(&d).as_deref(), Some("{\"new\":1}")); + assert_eq!(recorded(&d), Some(stamp(1, Some(9_000)))); + } + + #[tokio::test] + async fn a_record_from_before_the_time_was_kept_gets_its_time_and_keeps_the_session() { + let d = dir("upgrade"); + std::fs::write(d.0.join(VERSION_FILE), "1").expect("an old record"); + std::fs::write(d.0.join("claude").join(CREDENTIALS_FILE), "rotated").expect("session"); + let p = pickup_stamped(&d, Some(("{\"v\":1}", stamp(1, Some(1_000))))); + assert_eq!( + p.check().await.expect("check"), + Outcome::NotNewer { + stored: 1, + installed: 1 + } + ); + assert_eq!(installed(&d).as_deref(), Some("rotated")); + assert_eq!(recorded(&d), Some(stamp(1, Some(1_000)))); + } + struct Failing; impl Source for Failing { - async fn fetch(&self) -> anyhow::Result> { + async fn fetch(&self) -> anyhow::Result> { anyhow::bail!("the store did not answer") } } @@ -368,9 +556,6 @@ mod tests { std::fs::write(d.0.join(VERSION_FILE), "4").expect("an installed version"); let p = Pickup::new(Failing, d.0.join("claude"), d.0.join(VERSION_FILE)); assert!(p.check().await.is_err()); - assert_eq!( - read_version(&d.0.join(VERSION_FILE)).expect("read"), - Some(4) - ); + assert_eq!(recorded(&d), Some(stamp(4, None))); } } diff --git a/swarm-secret-client/src/client.rs b/swarm-secret-client/src/client.rs index 23ab2899b..f7761177b 100644 --- a/swarm-secret-client/src/client.rs +++ b/swarm-secret-client/src/client.rs @@ -48,6 +48,19 @@ pub const CERT_ROLE_TOKEN_TTL_SECS: u64 = 3600; /// this bound the call fails with [`Error::Vault`] instead. const VAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); +/// The identity of one stored version of an object: its KV v2 number, and the +/// time the store wrote it. +/// +/// The number restarts when the object is deleted outright; the time does not, +/// so a caller that must tell "a newer write" from "the same one again" across +/// a delete compares the time. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Version { + pub number: u64, + /// RFC 3339, as the store spells it. + pub created_time: String, +} + /// Where the store is and which identity we present to it. /// /// Separate from the connect so the environment can be checked without one: @@ -234,18 +247,20 @@ impl SecretStore { } } - /// The newest version of the object at `path` and its KV v2 version - /// number, or `None` when nothing is stored there. + /// The newest version of the object at `path` and its KV v2 [`Version`], or + /// `None` when nothing is stored there. /// - /// One read, so the number is the one the value was stored as. Every - /// [`write`][Self::write] makes a larger one. + /// One read, so the version is the one the value was stored as. Every + /// [`write`][Self::write] makes a larger number, until the path is deleted + /// outright and its numbering starts again; the time it was written, which + /// the store stamps, keeps ordering across that. /// /// # Errors /// As [`read_optional`][Self::read_optional]. pub async fn read_versioned( &self, path: &str, - ) -> Result, Error> { + ) -> Result, Error> { let endpoint = vaultrs::api::kv2::requests::ReadSecretRequest::builder() .mount(MOUNT) .path(path) @@ -255,7 +270,13 @@ impl SecretStore { Ok(res) => { let value = serde_json::from_value(res.data) .map_err(|source| vaultrs::error::ClientError::JsonParseError { source })?; - Ok(Some((value, res.metadata.version))) + Ok(Some(( + value, + Version { + number: res.metadata.version, + created_time: res.metadata.created_time, + }, + ))) } Err(vaultrs::error::ClientError::APIError { code: 404, .. }) => Ok(None), Err(e) => Err(e.into()),