From 5ac0f7d74d1aa0e70254f3f5abbd3de4b6482c27 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 10:23:03 +0200 Subject: [PATCH] feat(broker): log agent-name reclaims with their audit id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A takeover currently leaves no trace. relay#1596 made the broker reclaim a colliding agent name through an audited takeover instead of the register-or-rotate path that 401'd, but nothing logs that it happened — so from the outside a reclaim is only visible as the *absence* of a `401 Agent token required`, which is indistinguishable from the name never having collided at all. That gap is load-bearing where it matters most. Cloud retains no per-agent broker logs, so proving a scheduled workflow reuses stable agent names across runs currently rests on inference (same workspace, two green runs) rather than evidence. Log the reclaim at the three decision points: - takeover succeeded info, with agent, agent_id and audit_id - recover succeeded info, same fields (crash reclaim, auth.rs) - legacy rotate fallback warn, on engines older than 8.2.0 The audit id is the point: it is the handle onto the engine's own workspace-readable audit record, so the reclaim becomes checkable after the fact instead of inferred. The fallback is a `warn` because that path predates the audit surface and therefore produces no record — an operator should be able to tell a silently-unaudited reclaim from an audited one. Tokens are never logged, and the existing hashed `session_ref` treatment is untouched. `takeover_logs_the_audit_id` captures the subscriber output and asserts the audit id and agent name reach the log and the token does not. Verified to fail with the log line removed ("takeover must be logged, got: "). 1025 lib tests pass, clippy and fmt clean. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 4 + crates/broker/src/relaycast/auth.rs | 36 +++++-- crates/broker/src/relaycast/ws.rs | 139 +++++++++++++++++++++++++++- 3 files changed, 169 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ea489df8..15e761275 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Agent-name reclaims are now logged. A takeover records the agent, its id and the engine's audit id at `info`; a crash recovery through the recover route does the same; and both log a `warn` when they fall back to the legacy workspace-key rotate on an engine older than 8.2.0, which produces no audit record. Previously a reclaim was only visible as the absence of a `401`, which is indistinguishable from the name never having collided. Tokens are never logged. + ## [11.8.2] - 2026-08-22 ### Fixed diff --git a/crates/broker/src/relaycast/auth.rs b/crates/broker/src/relaycast/auth.rs index c37d825b8..f0e793d72 100644 --- a/crates/broker/src/relaycast/auth.rs +++ b/crates/broker/src/relaycast/auth.rs @@ -1087,19 +1087,37 @@ async fn admit_agent_registration( // agent-specific 404 (`agent_not_found`) is a real failure and is // deliberately excluded. let token_response = match token_response { - Ok(response) => response.token, + Ok(response) => { + // Same reasoning as the takeover path in ws.rs: the audit id + // ties this recovery to the engine's audit record, so a + // crash reclaim is checkable rather than inferred. + tracing::info!( + agent = %existing.name, + agent_id = %existing.id, + audit_id = %response.audit_id, + "recovered agent identity via the recover route" + ); + response.token + } Err(RelayError::Api { status: 404, ref code, .. - }) if code != "agent_not_found" => relay - .rotate_agent_token(&existing.name, workspace_key) - .await - .map_err(relay_error_to_anyhow) - .context( - "recover unavailable on this engine and the legacy rotate fallback failed", - )? - .token, + }) if code != "agent_not_found" => { + tracing::warn!( + agent = %existing.name, + agent_id = %existing.id, + "recover route absent on this engine; fell back to the legacy workspace-key rotate (unaudited)" + ); + relay + .rotate_agent_token(&existing.name, workspace_key) + .await + .map_err(relay_error_to_anyhow) + .context( + "recover unavailable on this engine and the legacy rotate fallback failed", + )? + .token + } Err(error) => return Err(relay_error_to_anyhow(error)), }; Ok(( diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 86d92a6ba..d82ef11ed 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -288,6 +288,10 @@ impl RelaycastHttpClient { // taking over again would invalidate it. if let Some(registration) = self.registration.as_ref().as_ref() { if let Some(cached) = registration.cached_agent_token(agent_name) { + tracing::debug!( + agent = %agent_name, + "another caller completed the takeover while we waited; reusing its token" + ); return Ok(cached); } } @@ -330,7 +334,19 @@ impl RelaycastHttpClient { ) .await { - Ok(response) => response, + Ok(response) => { + // The audit id is the whole point of logging here: it ties this + // reclaim to the engine's workspace-readable audit record, so + // "the name was taken over" is checkable after the fact rather + // than inferred from the absence of a 401. + tracing::info!( + agent = %agent_name, + agent_id = %existing_id, + audit_id = %response.audit_id, + "reclaimed agent name via takeover" + ); + response + } // Engines before 8.2.0 have no `/takeover` route — the whole // identity-recovery surface arrived with it. Those engines still // allow the workspace key to rotate an agent's token, which is what @@ -356,6 +372,14 @@ impl RelaycastHttpClient { "takeover unavailable on this engine and the legacy rotate fallback failed: {error}" ), })?; + // No audit id exists on this path: the legacy rotate route + // predates the audit surface. Warn so an operator can tell a + // silently-unaudited reclaim from an audited one. + tracing::warn!( + agent = %agent_name, + agent_id = %existing_id, + "takeover route absent on this engine; fell back to the legacy workspace-key rotate (unaudited)" + ); AgentIdentityRecoveryResponse { agent_id: existing_id, name: agent_name.to_string(), @@ -1424,6 +1448,7 @@ mod tests { }; use relaycast::AgentRegistrationError; use serde_json::json; + use std::sync::{Arc, Mutex as StdMutex}; use std::time::Duration; use crate::{fleet_wire::AgentRegistrationMetadata, ids::ChannelName}; @@ -2315,6 +2340,118 @@ mod tests { takeover.assert_hits(1); } + /// The takeover log line is the only operator-visible evidence that a + /// colliding name was *reclaimed* rather than freshly registered. Cloud + /// retains no per-agent broker logs, so without this an operator can only + /// infer the reclaim from the absence of a 401 — which is indistinguishable + /// from the name never having collided at all. Assert the audit id, the + /// handle onto the engine's own audit record, actually reaches the log. + #[tokio::test] + async fn takeover_logs_the_audit_id() { + #[derive(Clone)] + struct Capture(Arc>>); + + impl std::io::Write for Capture { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0 + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Capture { + type Writer = Capture; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + let server = MockServer::start(); + server.mock(|when, then| { + when.method(POST).path("/v1/agents"); + then.status(409).json_body(json!({ + "ok": false, + "error": { "code": "agent_already_exists", "message": "exists" } + })); + }); + server.mock(|when, then| { + when.method(GET).path("/v1/agents/worker-a"); + then.status(200).json_body(json!({ + "ok": true, + "data": { + "id": "a_worker-a", + "name": "worker-a", + "type": "agent", + "status": "offline", + "persona": null, + "metadata": {}, + "last_seen": "2026-08-16T20:00:00.000Z" + } + })); + }); + server.mock(|when, then| { + when.method(POST).path("/v1/agents/worker-a/takeover"); + then.status(200).json_body(json!({ + "ok": true, + "data": { "agent_id": "a_worker-a", "name": "worker-a", "token": "at_live_taken", "audit_id": "aud_42" } + })); + }); + + let buffer = Arc::new(StdMutex::new(Vec::new())); + let subscriber = tracing_subscriber::fmt() + .with_writer(Capture(Arc::clone(&buffer))) + .with_max_level(tracing::Level::INFO) + .with_ansi(false) + .finish(); + + { + let _guard = tracing::subscriber::set_default(subscriber); + let client = RelaycastHttpClient::new( + Some(server.base_url()), + "rk_live_test", + "broker", + "codex", + ); + let token = client + .register_agent_token("worker-a", Some("codex")) + .await + .expect("takeover succeeds"); + assert_eq!(token, "at_live_taken"); + } + + let logged = String::from_utf8( + buffer + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone(), + ) + .expect("log output is utf-8"); + + assert!( + logged.contains("reclaimed agent name via takeover"), + "takeover must be logged, got: {logged}" + ); + assert!( + logged.contains("aud_42"), + "the audit id must reach the log so the reclaim is checkable, got: {logged}" + ); + assert!( + logged.contains("worker-a"), + "the agent name must reach the log, got: {logged}" + ); + assert!( + !logged.contains("at_live_taken"), + "the token must never be logged, got: {logged}" + ); + } + /// Must-not-fire: the existing `register_agent_token` API keeps its /// spawn-time behaviour (register or rotate on collision) so /// supervisor-driven worker restart at maintenance.rs:569 continues to