From 18ba57a7b037676640f80ce87c9af5a5f52ba387 Mon Sep 17 00:00:00 2001 From: Timmy Date: Wed, 15 Jul 2026 16:22:11 +0100 Subject: [PATCH] Add `upgrade all` chat command with drain check and convergence verify MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - `upgrade all` sweeps every registered sled in sequence, streaming per-sled phase markers and reporting a summary. - Binary source is now the gateway's own artifact store via host.docker.internal (was: unresolvable `gateway` hostname serving the gateway's macOS binary to Linux sleds — would have bricked them). - Sleds with active claude processes are skipped, never killed. - After reconnect, /api/version git_hash is compared against the published artifact's .hash sidecar; divergence is reported loudly. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_019fHdm92yjvguPi2LiXfLB9 --- .../matrix/bot/messages/on_room_message.rs | 51 +++- .../chat/transport/matrix/project_rebuild.rs | 2 +- .../src/chat/transport/matrix/sled_upgrade.rs | 260 +++++++++++++----- server/src/http/mod.rs | 7 + 4 files changed, 250 insertions(+), 70 deletions(-) diff --git a/server/src/chat/transport/matrix/bot/messages/on_room_message.rs b/server/src/chat/transport/matrix/bot/messages/on_room_message.rs index ff5628da..da87f7d6 100644 --- a/server/src/chat/transport/matrix/bot/messages/on_room_message.rs +++ b/server/src/chat/transport/matrix/bot/messages/on_room_message.rs @@ -537,7 +537,7 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( let bot_sent = Arc::clone(&ctx.bot_sent_event_ids); let room = room_id_str.clone(); - let response = super::super::super::sled_upgrade::handle_sled_upgrade( + let outcome = super::super::super::sled_upgrade::handle_sled_upgrade( &project, store, ctx.gateway_port, @@ -557,6 +557,53 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( }, ) .await; + let response = format!("{project}: {outcome}"); + + let html = markdown_to_html(&response); + if let Ok(msg_id) = ctx + .transport + .send_message(&room_id_str, &response, &html) + .await + && let Ok(event_id) = msg_id.parse() + { + ctx.bot_sent_event_ids.lock().await.insert(event_id); + } + } else { + let msg = "Gateway projects store unavailable — cannot upgrade sled."; + let html = markdown_to_html(msg); + if let Ok(msg_id) = ctx.transport.send_message(&room_id_str, msg, &html).await + && let Ok(event_id) = msg_id.parse() + { + ctx.bot_sent_event_ids.lock().await.insert(event_id); + } + } + } + super::super::super::sled_upgrade::UpgradeCommand::UpgradeAll => { + slog!("[matrix-bot] Handling 'upgrade all' from {sender}"); + if let Some(ref store) = ctx.gateway_projects_store { + let transport = Arc::clone(&ctx.transport); + let bot_sent = Arc::clone(&ctx.bot_sent_event_ids); + let room = room_id_str.clone(); + + let response = super::super::super::sled_upgrade::handle_upgrade_all( + store, + ctx.gateway_port, + |phase_msg| { + let transport = Arc::clone(&transport); + let bot_sent = Arc::clone(&bot_sent); + let room = room.clone(); + async move { + let html = markdown_to_html(&phase_msg); + if let Ok(msg_id) = + transport.send_message(&room, &phase_msg, &html).await + && let Ok(event_id) = msg_id.parse() + { + bot_sent.lock().await.insert(event_id); + } + } + }, + ) + .await; let html = markdown_to_html(&response); if let Ok(msg_id) = ctx @@ -568,7 +615,7 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( ctx.bot_sent_event_ids.lock().await.insert(event_id); } } else { - let msg = "Gateway projects store unavailable — cannot upgrade sled."; + let msg = "Gateway projects store unavailable — cannot upgrade sleds."; let html = markdown_to_html(msg); if let Ok(msg_id) = ctx.transport.send_message(&room_id_str, msg, &html).await && let Ok(event_id) = msg_id.parse() diff --git a/server/src/chat/transport/matrix/project_rebuild.rs b/server/src/chat/transport/matrix/project_rebuild.rs index 0e583f05..2673eea1 100644 --- a/server/src/chat/transport/matrix/project_rebuild.rs +++ b/server/src/chat/transport/matrix/project_rebuild.rs @@ -343,7 +343,7 @@ async fn wait_for_drain(container_name: &str, timeout_secs: u64) -> Option pgrep -f claude` — exits 0 with PID list when found, /// exits 1 when no matches (treated as 0 active processes). -async fn count_active_claude_processes(container_name: &str) -> Result { +pub(crate) async fn count_active_claude_processes(container_name: &str) -> Result { let out = tokio::process::Command::new("docker") .args(["exec", container_name, "pgrep", "-f", "claude"]) .output() diff --git a/server/src/chat/transport/matrix/sled_upgrade.rs b/server/src/chat/transport/matrix/sled_upgrade.rs index aaccf05e..3288c969 100644 --- a/server/src/chat/transport/matrix/sled_upgrade.rs +++ b/server/src/chat/transport/matrix/sled_upgrade.rs @@ -1,16 +1,26 @@ -//! `upgrade []` gateway chat command — streaming sled binary upgrade. +//! `upgrade [|all]` gateway chat command — streaming sled binary upgrade. //! //! Usage (gateway mode only): //! - `{bot} upgrade ` — upgrade the named sled's binary in-container. +//! - `{bot} upgrade all` — upgrade every registered sled in sequence. //! - `{bot} upgrade` — list registered projects (shows what can be targeted). //! -//! The gateway orchestrates the upgrade in four phases, streaming a marker to +//! The binary comes from the gateway's own artifact store +//! (`~/.huskies/artifacts/`, published by the `release` command) — sleds never +//! download from anywhere but their gateway. Sleds with active agent +//! processes are skipped so an upgrade never kills in-flight work. +//! +//! The gateway orchestrates each upgrade in four phases, streaming a marker to //! the chat room at each step: //! 1. `[1/4] downloading` — POSTs to `{sled_url}/api/upgrade`; sled starts download. //! 2. `[2/4] swapping binary` — gateway received 202; sled atomically renamed the binary. -//! 3. `[3/4] restarting sled` — sled re-execs with the new binary; HTTP goes dark briefly. +//! 3. `[3/4] restarting sled` — sled exits cleanly; Docker restarts it with the new binary. //! 4. `[4/4] reconnected to gateway` — sled's `/health` probe is responding again. //! +//! After reconnection the gateway polls `/api/version` and verifies the sled's +//! reported git hash matches the published artifact's hash (when a `.hash` +//! sidecar file exists). +//! //! Concurrent `upgrade` invocations are serialised via a global async mutex so //! that two simultaneous upgrades cannot interleave their phase markers or race //! on the sled restart. @@ -40,6 +50,8 @@ pub enum UpgradeCommand { /// The project/sled name to upgrade. project: String, }, + /// `upgrade all` — upgrade every registered sled in sequence. + UpgradeAll, /// `upgrade` with no argument — list available projects. ListProjects, } @@ -68,10 +80,14 @@ pub fn extract_upgrade_command( } if rest.is_empty() { - Some(UpgradeCommand::ListProjects) + return Some(UpgradeCommand::ListProjects); + } + let target = rest.split_whitespace().next().unwrap_or(rest); + if target.eq_ignore_ascii_case("all") { + Some(UpgradeCommand::UpgradeAll) } else { Some(UpgradeCommand::Upgrade { - project: rest.split_whitespace().next().unwrap_or(rest).to_string(), + project: target.to_string(), }) } } @@ -98,6 +114,84 @@ pub async fn handle_upgrade_list_projects( format!("Registered projects (use `upgrade ` to upgrade one):\n{list}") } +/// Resolve the artifact source URL and expected git hash for an upgrade. +/// +/// The URL points at the gateway's own artifact endpoint via +/// `host.docker.internal` (resolvable from inside sled containers). The +/// expected hash comes from the `.hash` sidecar written by `release`, when +/// present. `HUSKIES_GATEWAY_BINARY_URL` overrides the URL (no hash check). +/// +/// Returns `Err` with a user-facing message when no artifact has been +/// published yet. +fn resolve_artifact_source(gateway_port: Option) -> Result<(String, Option), String> { + if let Ok(url) = std::env::var("HUSKIES_GATEWAY_BINARY_URL") { + return Ok((url, None)); + } + + let artifact_name = crate::http::SLED_ARTIFACT_NAME; + let artifact_path = crate::http::artifacts_dir().join(artifact_name); + if !artifact_path.exists() { + return Err(format!( + "No published artifact at `{}`. Run `release` first to build and publish one.", + artifact_path.display() + )); + } + + let expected_hash = + std::fs::read_to_string(crate::http::artifacts_dir().join(format!("{artifact_name}.hash"))) + .ok() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()); + + let url = format!( + "http://host.docker.internal:{}/api/artifacts/{artifact_name}", + gateway_port.unwrap_or(3000) + ); + Ok((url, expected_hash)) +} + +/// Upgrade every registered sled in sequence, streaming per-sled phase markers. +/// +/// Sleds with active agent processes are skipped (reported, not failed). +/// Returns a summary listing the outcome for each sled. +pub async fn handle_upgrade_all( + projects_store: &Arc>>, + gateway_port: Option, + send_phase: F, +) -> String +where + F: Fn(String) -> Fut, + Fut: Future, +{ + let names: Vec = { + let projects = projects_store.read().await; + projects.keys().cloned().collect() + }; + if names.is_empty() { + return "No projects are currently registered with the gateway.".to_string(); + } + + // Fail fast before touching any sled if there is nothing to distribute. + if let Err(e) = resolve_artifact_source(gateway_port) { + return e; + } + + let mut results: Vec = Vec::with_capacity(names.len()); + for name in &names { + let outcome = handle_sled_upgrade(name, projects_store, gateway_port, |msg| { + send_phase(format!("**{name}** {msg}")) + }) + .await; + results.push(format!("- {name}: {outcome}")); + } + + format!( + "Upgrade sweep over {} sled(s) complete:\n{}", + names.len(), + results.join("\n") + ) +} + /// Upgrade a named sled by streaming phase markers to the chat room. /// /// Acquires the global upgrade lock to serialise concurrent invocations. Each @@ -105,9 +199,8 @@ pub async fn handle_upgrade_list_projects( /// begins. On any failure, an error message is returned and the previous /// binary remains active on the sled. /// -/// `gateway_port` is used to derive the default binary source URL -/// (`http://gateway:/api/huskies-binary`) when neither -/// `HUSKIES_GATEWAY_BINARY_URL` nor `--source` is set. +/// Sleds with active agent processes are skipped so an upgrade never kills +/// in-flight work. pub async fn handle_sled_upgrade( project: &str, projects_store: &Arc>>, @@ -137,13 +230,41 @@ where } }; - // ── Resolve binary source URL ──────────────────────────────────────────── - let source_url = std::env::var("HUSKIES_GATEWAY_BINARY_URL").unwrap_or_else(|_| { - format!( - "http://gateway:{}/api/huskies-binary", - gateway_port.unwrap_or(3000) - ) - }); + // ── Resolve binary source ──────────────────────────────────────────────── + let (source_url, expected_hash) = match resolve_artifact_source(gateway_port) { + Ok(v) => v, + Err(e) => return e, + }; + + // ── Drain check ────────────────────────────────────────────────────────── + // Never kill in-flight agent work; the caller can retry once idle. + let container_name = format!("huskies-{project}"); + if let Ok(n) = super::project_rebuild::count_active_claude_processes(&container_name).await + && n > 0 + { + return format!( + "skipped — {n} active agent process(es) in `{container_name}`. Retry when idle." + ); + } + + run_sled_upgrade(project, &sled_url, &source_url, expected_hash, send_phase).await +} + +/// Run the four-phase upgrade against a sled whose source URL is already +/// resolved. Split from [`handle_sled_upgrade`] so tests can drive the wire +/// behaviour without a published artifact on the host. +async fn run_sled_upgrade( + project: &str, + sled_url: &str, + source_url: &str, + expected_hash: Option, + send_phase: F, +) -> String +where + F: Fn(String) -> Fut, + Fut: Future, +{ + let container_name = format!("huskies-{project}"); // ── Acquire serial lock ────────────────────────────────────────────────── let _lock = upgrade_lock().lock().await; @@ -206,9 +327,25 @@ where // ── Phase 4: reconnected ───────────────────────────────────────────────── send_phase("[4/4] reconnected to gateway".to_string()).await; - // ── Report new version ─────────────────────────────────────────────────── - let version = fetch_sled_version(&client, &sled_url).await; - format!("{project} upgraded to version {version}") + // ── Verify convergence ─────────────────────────────────────────────────── + match fetch_sled_version(&client, sled_url).await { + Some((version, git_hash)) => match expected_hash { + Some(expected) if git_hash == expected => { + format!("upgraded to v{version} ({git_hash}) — matches published artifact") + } + Some(expected) => format!( + "**upgrade did not converge**: sled reports {git_hash}, published artifact \ + is {expected}. The sled is healthy but running the wrong binary — check \ + `docker logs {container_name}`." + ), + None => format!("upgraded to v{version} ({git_hash})"), + }, + None => format!( + "upgraded and healthy, but `/api/version` is unavailable — the sled is \ + probably still on a pre-version-endpoint binary. Check `docker logs \ + {container_name}`." + ), + } } // ── Helpers ─────────────────────────────────────────────────────────────────── @@ -231,41 +368,16 @@ async fn wait_for_health(client: &reqwest::Client, health_url: &str, timeout_sec } } -/// Fetch the running version from the sled's `get_version` MCP tool. +/// Fetch `(version, git_hash)` from the sled's `/api/version` endpoint. /// -/// Returns the version string on success, or `"unknown"` on any error so the -/// final chat reply is still meaningful. -async fn fetch_sled_version(client: &reqwest::Client, sled_url: &str) -> String { - let mcp_url = format!("{}/mcp", sled_url.trim_end_matches('/')); - let body = serde_json::json!({ - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": { - "name": "get_version", - "arguments": {} - } - }); - let resp = match client.post(&mcp_url).json(&body).send().await { - Ok(r) => r, - Err(_) => return "unknown".to_string(), - }; - let val: serde_json::Value = match resp.json().await { - Ok(v) => v, - Err(_) => return "unknown".to_string(), - }; - // MCP tools/call response: result.content[0].text is a JSON string. - let text = val - .pointer("/result/content/0/text") - .and_then(|v| v.as_str()) - .unwrap_or(""); - if text.is_empty() { - return "unknown".to_string(); - } - serde_json::from_str::(text) - .ok() - .and_then(|v| v.get("version").and_then(|v| v.as_str()).map(String::from)) - .unwrap_or_else(|| "unknown".to_string()) +/// Returns `None` when the endpoint is unreachable or malformed — e.g. a sled +/// still running a binary that predates the endpoint. +async fn fetch_sled_version(client: &reqwest::Client, sled_url: &str) -> Option<(String, String)> { + let url = format!("{}/api/version", sled_url.trim_end_matches('/')); + let val: serde_json::Value = client.get(&url).send().await.ok()?.json().await.ok()?; + let version = val.get("version").and_then(|v| v.as_str())?.to_string(); + let git_hash = val.get("git_hash").and_then(|v| v.as_str())?.to_string(); + Some((version, git_hash)) } // ── Tests ────────────────────────────────────────────────────────────────────── @@ -405,24 +517,38 @@ mod tests { ); } + #[test] + fn extract_upgrade_all() { + let cmd = extract_upgrade_command("Timmy upgrade all", "Timmy", "@timmy:home"); + assert_eq!(cmd, Some(UpgradeCommand::UpgradeAll)); + let cmd = extract_upgrade_command("@timmy upgrade ALL", "Timmy", "@timmy:home"); + assert_eq!(cmd, Some(UpgradeCommand::UpgradeAll)); + } + + #[tokio::test] + async fn upgrade_all_empty_store_reports_no_projects() { + let store: Arc>> = + Arc::new(RwLock::new(BTreeMap::new())); + let msg = handle_upgrade_all(&store, Some(3000), |_msg| async {}).await; + assert!( + msg.contains("No projects"), + "empty store should say no projects: {msg}" + ); + } + #[tokio::test] async fn upgrade_unreachable_sled_reports_failure() { - let mut map = BTreeMap::new(); - map.insert( - "myapp".to_string(), - ProjectEntry { - url: Some("http://127.0.0.1:1".into()), // port 1 is never listening - auth_token: None, - ssh_port: None, - host_path: None, - }, - ); - let store = Arc::new(RwLock::new(map)); let phases: std::sync::Mutex> = std::sync::Mutex::new(vec![]); - let result = handle_sled_upgrade("myapp", &store, Some(3000), |msg| { - phases.lock().unwrap().push(msg); - async {} - }) + let result = run_sled_upgrade( + "myapp", + "http://127.0.0.1:1", // port 1 is never listening + "http://127.0.0.1:1/api/artifacts/huskies-linux-arm64", + None, + |msg| { + phases.lock().unwrap().push(msg); + async {} + }, + ) .await; // Phase 1 marker must have been sent before the failed request. let sent = phases.lock().unwrap().clone(); diff --git a/server/src/http/mod.rs b/server/src/http/mod.rs index cf89a6ed..88903fee 100644 --- a/server/src/http/mod.rs +++ b/server/src/http/mod.rs @@ -298,6 +298,13 @@ pub async fn serve_binary_handler() -> poem::Response { } } +/// Canonical artifact filename for sled binaries on this deployment's platform. +/// +/// Sleds run linux/arm64 under OrbStack on Apple Silicon. When amd64 hosts +/// arrive, `release` publishes one artifact per platform and this becomes a +/// lookup instead of a constant. +pub const SLED_ARTIFACT_NAME: &str = "huskies-linux-arm64"; + /// Directory where the gateway stores distributable sled binaries. /// /// Host-global (`~/.huskies/artifacts/`), not per-project: one artifact serves