hive-agent: a logged-in agent polls the swarm login, and logs out when it is deleted
wait_for_login read the store leaf only while the agent waited for a login. A logged-in agent now reads it every 5 minutes, and a paused one once before it resumes, so a login made or deleted in the swarm UI reaches an agent that never failed. A newer version is installed; a store that answers nothing stored after a version was installed means the stored login was deleted, and the agent is logged out as /api/logout does it, with the recorded version forgotten so the next login installs. A read that fails changes nothing.
This commit is contained in:
parent
71a4d1d255
commit
a2c9e9a8d4
6 changed files with 188 additions and 23 deletions
|
|
@ -1,6 +1,7 @@
|
|||
//! The claude credentials the swarm's login minter stored for this agent, at
|
||||
//! `swarm/agents/<agent>/claude-credentials`, read under the agent's own
|
||||
//! certificate while it waits for a login ([`crate::login::wait_for_login`]).
|
||||
//! 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 here**, never because the bytes differ. `claude` rotates the
|
||||
|
|
@ -9,6 +10,10 @@
|
|||
//! The last installed version is kept in [`VERSION_FILE`] under the harness
|
||||
//! dir. A credentials file `wait_for_login` sees land resumes the agent as a
|
||||
//! web-UI login does.
|
||||
//!
|
||||
//! A store that answers "nothing stored" after a version was installed here
|
||||
//! means the operator deleted the stored login ([`Outcome::Deleted`]); a store
|
||||
//! that does not answer is an error and means nothing.
|
||||
|
||||
use std::future::Future;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
|
@ -43,8 +48,12 @@ pub fn is_newer(stored: u64, installed: Option<u64>) -> bool {
|
|||
/// What one [`Pickup::check`] did.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum Outcome {
|
||||
/// Nothing is stored for this agent.
|
||||
/// Nothing is stored for this agent, and no version was ever installed
|
||||
/// here: no swarm login has reached this agent.
|
||||
NothingStored,
|
||||
/// The store answered that nothing is stored, though version `installed`
|
||||
/// was installed here: the operator deleted the stored login.
|
||||
Deleted { installed: u64 },
|
||||
/// The stored version is not newer than the installed one.
|
||||
NotNewer { stored: u64, installed: u64 },
|
||||
/// The stored version is installed now.
|
||||
|
|
@ -120,7 +129,10 @@ 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(Outcome::NothingStored);
|
||||
return Ok(match read_version(&self.version_file)? {
|
||||
Some(installed) => Outcome::Deleted { installed },
|
||||
None => Outcome::NothingStored,
|
||||
});
|
||||
};
|
||||
let installed = read_version(&self.version_file)?;
|
||||
if !is_newer(stored, installed) {
|
||||
|
|
@ -138,6 +150,19 @@ impl<S: Source> Pickup<S> {
|
|||
.context("recording the installed credentials version")?;
|
||||
Ok(Outcome::Installed { version: stored })
|
||||
}
|
||||
|
||||
/// Forget the installed version, once the deletion it showed has been
|
||||
/// acted on, so the next login installs whatever version it gets.
|
||||
///
|
||||
/// # Errors
|
||||
/// Removing the version file failed for a reason other than its absence.
|
||||
pub fn forget_installed(&self) -> anyhow::Result<()> {
|
||||
match std::fs::remove_file(&self.version_file) {
|
||||
Ok(()) => Ok(()),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
|
||||
Err(e) => Err(e).with_context(|| format!("removing {}", self.version_file.display())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The version recorded in `file`, or `None` when there is no file.
|
||||
|
|
@ -302,4 +327,50 @@ mod tests {
|
|||
assert_eq!(installed(&d), None);
|
||||
assert_eq!(read_version(&d.0.join(VERSION_FILE)).expect("read"), 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)));
|
||||
p.check().await.expect("install");
|
||||
*p.source.0.lock().expect("lock") = None;
|
||||
assert_eq!(
|
||||
p.check().await.expect("check"),
|
||||
Outcome::Deleted { installed: 1 }
|
||||
);
|
||||
// Reporting it changes nothing: the caller acts, then forgets.
|
||||
assert_eq!(
|
||||
p.check().await.expect("check again"),
|
||||
Outcome::Deleted { installed: 1 }
|
||||
);
|
||||
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));
|
||||
assert_eq!(
|
||||
p.check().await.expect("a new login"),
|
||||
Outcome::Installed { version: 1 },
|
||||
"the counter restarts after a delete, and its first login still installs"
|
||||
);
|
||||
}
|
||||
|
||||
struct Failing;
|
||||
|
||||
impl Source for Failing {
|
||||
async fn fetch(&self) -> anyhow::Result<Option<(Credentials, u64)>> {
|
||||
anyhow::bail!("the store did not answer")
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failed_read_is_an_error_never_a_deletion() {
|
||||
let d = dir("failing");
|
||||
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)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -191,6 +191,14 @@ pub async fn wait_for_login(
|
|||
version,
|
||||
"installed the claude credentials the swarm login stored"
|
||||
),
|
||||
Ok(claude_credentials::Outcome::Deleted { installed }) => {
|
||||
// Nothing to log out of, but the recorded version would
|
||||
// keep a restarted counter's first login from installing.
|
||||
tracing::info!(installed, "the swarm's stored claude login was deleted");
|
||||
if let Err(e) = pickup.forget_installed() {
|
||||
tracing::warn!(error = %format!("{e:#}"), "forgetting the installed login version failed");
|
||||
}
|
||||
}
|
||||
Ok(outcome) => tracing::debug!(?outcome, "no newer swarm claude credentials"),
|
||||
Err(e) => tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
|
|
@ -214,7 +222,7 @@ pub async fn wait_for_login(
|
|||
|
||||
/// The swarm-login pickup for `claude_dir`, or `None` when this process has
|
||||
/// no store to read from.
|
||||
fn swarm_pickup(
|
||||
pub(crate) fn swarm_pickup(
|
||||
claude_dir: &Path,
|
||||
) -> Option<claude_credentials::Pickup<claude_credentials::Store>> {
|
||||
match claude_credentials::Store::from_env() {
|
||||
|
|
|
|||
72
hive-agent/src/login_watch.rs
Normal file
72
hive-agent/src/login_watch.rs
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
//! The swarm login, watched while the agent runs. [`crate::login::wait_for_login`]
|
||||
//! reads the agent's store leaf while the agent waits for a login; this reads
|
||||
//! it every [`POLL_INTERVAL`] while the agent is logged in, and once more just
|
||||
//! before a paused agent resumes, so a login the operator made or deleted in
|
||||
//! the swarm UI reaches an agent that never failed.
|
||||
//!
|
||||
//! - A newer stored version is installed over the session, as after a login
|
||||
//! while waiting ([`crate::claude_credentials`]).
|
||||
//! - A store that answers "nothing stored" after a version was installed here
|
||||
//! means the operator deleted the stored login: the agent is logged out as
|
||||
//! `/api/logout` does it, and the recorded version is forgotten so the next
|
||||
//! login installs.
|
||||
//! - A read that fails does nothing. A store that cannot be reached says
|
||||
//! nothing about the login, and logging out on it would turn a store restart
|
||||
//! into an outage of every agent.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::claude_credentials::Outcome;
|
||||
use crate::events::{Bus, LiveEvent};
|
||||
use crate::login::{self, LoginState};
|
||||
use crate::web_ui::{LoginStateCell, force_logout};
|
||||
|
||||
/// How often a logged-in agent asks the store.
|
||||
pub const POLL_INTERVAL: Duration = Duration::from_mins(5);
|
||||
|
||||
/// Check the store every [`POLL_INTERVAL`], for the life of the process.
|
||||
pub async fn run(login: LoginStateCell, bus: Bus) {
|
||||
let mut tick = tokio::time::interval(POLL_INTERVAL);
|
||||
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||||
// The first tick is immediate; the boot path has read the store already.
|
||||
tick.tick().await;
|
||||
loop {
|
||||
tick.tick().await;
|
||||
check_once(&login, &bus).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Read the store once for a logged-in agent and act on what it holds. Does
|
||||
/// nothing for an agent waiting for a login, which polls for itself.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if the login-state lock is poisoned.
|
||||
pub async fn check_once(login: &LoginStateCell, bus: &Bus) {
|
||||
if *login.lock().unwrap() != LoginState::Online {
|
||||
return;
|
||||
}
|
||||
let Some(pickup) = login::swarm_pickup(&crate::paths::claude_dir()) else {
|
||||
return;
|
||||
};
|
||||
match pickup.check().await {
|
||||
Ok(Outcome::Installed { version }) => {
|
||||
tracing::info!(version, "installed a newer swarm claude login");
|
||||
bus.emit(LiveEvent::Note {
|
||||
text: format!("swarm login: installed version {version} over the running session"),
|
||||
});
|
||||
}
|
||||
Ok(Outcome::Deleted { installed }) => {
|
||||
tracing::info!(installed, "the swarm's stored claude login was deleted");
|
||||
force_logout(login, bus, "swarm login deleted").await;
|
||||
if let Err(e) = pickup.forget_installed() {
|
||||
tracing::warn!(error = %format!("{e:#}"), "forgetting the installed login version failed");
|
||||
}
|
||||
}
|
||||
Ok(outcome) => tracing::debug!(?outcome, "no change in the swarm claude login"),
|
||||
Err(e) => tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
"checking the swarm store for a claude login failed; leaving the session alone"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
|
@ -20,6 +20,7 @@ mod events;
|
|||
mod harness_state;
|
||||
mod identity;
|
||||
mod login;
|
||||
mod login_watch;
|
||||
mod mcp_config;
|
||||
mod otel_turn_metrics;
|
||||
mod paths;
|
||||
|
|
@ -540,6 +541,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
|
|||
// hive-c0re, because the files are agent-owned — see `vacuum` module docs.
|
||||
tokio::spawn(crate::vacuum::run());
|
||||
tokio::spawn(claude_usage_watch::run());
|
||||
tokio::spawn(login_watch::run(login_state.clone(), bus.clone()));
|
||||
tokio::spawn(subagent_role_watch::run());
|
||||
// Log web_ui::serve's error instead of dropping it. A bare
|
||||
// `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so
|
||||
|
|
@ -780,6 +782,9 @@ async fn serve_loop<S: Surface>(
|
|||
});
|
||||
was_paused = false;
|
||||
todo_miss_streak = 0;
|
||||
// A login the operator deleted or renewed while the agent was parked
|
||||
// counts before its first turn, not five minutes into it.
|
||||
login_watch::check_once(&login_state, &bus).await;
|
||||
}
|
||||
let next = match next_outcome(
|
||||
S::recv_next(socket),
|
||||
|
|
|
|||
|
|
@ -6,9 +6,10 @@ use axum::{
|
|||
response::{IntoResponse, Response},
|
||||
};
|
||||
|
||||
use crate::events::Bus;
|
||||
use crate::login::LoginState;
|
||||
|
||||
use super::AppState;
|
||||
use super::{AppState, LoginStateCell};
|
||||
|
||||
/// Operator-driven `/logout`: SIGINT claude, delete the credential
|
||||
/// files (via [`crate::login::clear_session`]), flip `LoginState::NeedsLogin`.
|
||||
|
|
@ -20,6 +21,13 @@ use super::AppState;
|
|||
/// (the `/api/logout` bullet) for the three-step rationale +
|
||||
/// preservation invariants.
|
||||
pub(super) async fn post_logout(State(state): State<AppState>) -> Response {
|
||||
let said = force_logout(&state.login, &state.bus, "operator: /logout").await;
|
||||
(axum::http::StatusCode::OK, format!("ok: {said}")).into_response()
|
||||
}
|
||||
|
||||
/// The three steps of a logout, shared by `/api/logout` and the swarm login
|
||||
/// watch: `by` heads the Note. Returns what was wiped, for the response.
|
||||
pub(crate) async fn force_logout(login: &LoginStateCell, bus: &Bus, by: &str) -> String {
|
||||
// Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`).
|
||||
let _ = super::sigint_claude();
|
||||
// Step 2: delete OAuth credential files only — login::clear_session owns
|
||||
|
|
@ -34,19 +42,13 @@ pub(super) async fn post_logout(State(state): State<AppState>) -> Response {
|
|||
};
|
||||
// Step 3: flip LoginState + emit Note. Turn loop sees the flip on
|
||||
// its next iteration and parks into wait_for_login.
|
||||
*state.login.lock().unwrap() = LoginState::NeedsLogin;
|
||||
state.bus.emit(crate::events::LiveEvent::Note {
|
||||
text: format!(
|
||||
"operator: /logout — {wipe_summary} in {}{warn_suffix}",
|
||||
dir.display()
|
||||
),
|
||||
*login.lock().unwrap() = LoginState::NeedsLogin;
|
||||
let said = format!("{wipe_summary} in {}{warn_suffix}", dir.display());
|
||||
bus.emit(crate::events::LiveEvent::Note {
|
||||
text: format!("{by} — {said}"),
|
||||
});
|
||||
state.bus.emit_status("needs_login_idle");
|
||||
(
|
||||
axum::http::StatusCode::OK,
|
||||
format!("ok: {wipe_summary} in {}{warn_suffix}", dir.display()),
|
||||
)
|
||||
.into_response()
|
||||
bus.emit_status("needs_login_idle");
|
||||
said
|
||||
}
|
||||
|
||||
/// Human-readable summary of a [`crate::login::ClearedSession`] outcome for
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@
|
|||
|
||||
mod actions;
|
||||
mod auth;
|
||||
pub(crate) use auth::force_logout;
|
||||
mod proxy;
|
||||
mod screen;
|
||||
mod state;
|
||||
|
|
|
|||
Loading…
Reference in a new issue