mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive-agent: order a stored login by when the store wrote it, not by its number

Browse source 
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.
This commit is contained in:
iris 2026-11-09 22:31:51 +01:00 • committed by mara
commit 4b1012056c
3 changed files with 275 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

18
docs/2-handbook/swarm/claude-login.md
View file

@ -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

287
hive-agent/src/claude_credentials.rs
View file

@ -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<i64>,
}
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<Self> {
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<u64>) -> bool {
installed.is_none_or(|last| stored > last)
pub fn judge(stored: Stamp, installed: Option<Stamp>) -> 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<Output = anyhow::Result<Option<(Credentials, u64)>>> + Send;
fn fetch(&self) -> impl Future<Output = anyhow::Result<Option<(Credentials, Stamp)>>> + 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<Option<(Credentials, u64)>> {
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, Stamp)>> {
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<S: Source> Pickup<S> {
/// The store read failed, or writing the credentials or the version did.
pub async fn check(&self) -> anyhow::Result<Outcome> {
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<S: Source> Pickup<S> {
}
/// The version recorded in `file`, or `None` when there is no file.
fn read_version(file: &Path) -> anyhow::Result<Option<u64>> {
fn read_stamp(file: &Path) -> anyhow::Result<Option<Stamp>> {
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<Option<(String, u64)>>);
struct Fake(Mutex<Option<(String, Stamp)>>);
impl Source for Fake {
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, u64)>> {
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, Stamp)>> {
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<i64>) -> 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<Fake> {
pickup_stamped(d, stored.map(|(v, n)| (v, stamp(n, None))))
}
fn pickup_stamped(d: &Dir, stored: Option<(&str, Stamp)>) -> Pickup<Fake> {
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<Fake>, stored: Option<(&str, Stamp)>) {
*p.source.0.lock().expect("lock") = stored.map(|(v, s)| (v.to_owned(), s));
}
fn installed(d: &Dir) -> Option<String> {
std::fs::read_to_string(d.0.join("claude").join(CREDENTIALS_FILE)).ok()
}
fn recorded(d: &Dir) -> Option<Stamp> {
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<Option<(Credentials, u64)>> {
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, Stamp)>> {
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)));
}
}

33
swarm-secret-client/src/client.rs
View file

@ -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<T: DeserializeOwned>(
&self,
path: &str,
) -> Result<Option<(T, u64)>, Error> {
) -> Result<Option<(T, Version)>, 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()),