From 2a1fd15ea213b80a7c21ce8523c91551ea93d058 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Mon, 12 Oct 2026 14:40:48 -0700 Subject: [PATCH 01/14] fix(broker): reconcile owned identity cleanup on a jittered backoff until it resolves MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A failed owned cleanup retried five times over ~25s and then left the worker name in "pending owned cleanup" until an explicit generation-matched release retry — which is how a name became unusable for the rest of a sustained outage (e.g. Relaycast answering d1_pressure/workspace_busy on every release attempt). Custody now resolves itself: - Failed attempts retry on an exponential backoff (5s doubling to a 5m cap) with ±25% jitter, so a fleet of stuck cleanups cannot pile onto an already-pressured Relaycast. Fleet action callers still get their unconfirmed result after the first five quick attempts; API callers keep the immediate-error contract. - A durable release conflict (agent_release_generation_conflict / agent_identity_mismatch / expected_token_hash_mismatch) proves the remote identity is not this generation's to delete, so the cleanup completes instead of retrying a conflict that can never resolve. - A cleanup that captured no credential hash can never run its guarded delete; it releases custody outright, matching the journal's restart behavior of not persisting un-capturable entries. Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- CHANGELOG.md | 1 + crates/broker/src/relaycast/ws.rs | 46 ++++ crates/broker/src/runtime/identity_cleanup.rs | 106 ++++++-- crates/broker/src/runtime/tests.rs | 257 ++++++++++++++++-- 4 files changed, 367 insertions(+), 43 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a159f0fb..8e53b2a95 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Owned worker-name cleanup no longer gives up after five tries and blocks the name until manual recovery. The broker keeps reconciling the release on a jittered backoff (up to five minutes between attempts), and completes the cleanup outright when Relaycast reports the remote identity is already gone or belongs to a newer generation — so a name stuck in "pending owned cleanup" frees itself once the remote state is reachable or resolved. - `npx @agent-relay/connect join` no longer waits 60 seconds and rolls back when the local probe is running but cannot match the command to a live Claude Code or Codex session (for example, when it was run from a separate terminal). It stops within about a second and says to run the command from the agent's own shell. A timeout that does happen now names the probe's last answer. `connect install` no longer reports such a probe as ready when its version cannot be checked. ## [13.4.0] - 2026-10-11 diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index b3dab880c..a061e519a 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -72,6 +72,31 @@ pub type RelaycastRegistrationError = AgentRegistrationError; pub type RegRetryOutcome = AgentRegistrationRetryOutcome; pub(crate) use relaycast::registration_retry_after_secs; +/// Terminal verdict for a guarded owned-identity release: the remote +/// identity is definitively not this generation's — it is already gone or +/// was re-registered by a replacement. The guarded delete must not run (that +/// is exactly what the token-hash guard prevents), and there is nothing +/// left for the cleanup to delete, so callers treat this as a completed +/// cleanup rather than a retryable failure. +#[derive(Debug)] +pub(crate) struct ReleaseIdentityConflict { + agent_name: String, + status: u16, + code: String, +} + +impl std::fmt::Display for ReleaseIdentityConflict { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "remote identity for '{}' is not this generation's (HTTP {} {})", + self.agent_name, self.status, self.code + ) + } +} + +impl std::error::Error for ReleaseIdentityConflict {} + /// Why the broker is asking `register_agent_token` for a token. /// /// The two intents diverge on how a name-collision-with-a-live-agent is @@ -1391,6 +1416,27 @@ impl RelaycastHttpClient { break 'cleanup (status, result); }; if !status.is_success() { + // A durable release conflict means the remote identity is + // not this generation's to delete — gone already or held + // by a replacement — so the cleanup's goal is met without + // another write. Transient failures (429/503/d1_pressure, + // timeouts) keep the generic retryable bail. + let error_code = result["error"]["code"].as_str().unwrap_or_default(); + let not_ours = (status == reqwest::StatusCode::CONFLICT + && matches!( + error_code, + "agent_release_generation_conflict" | "agent_identity_mismatch" + )) + || (status == reqwest::StatusCode::FORBIDDEN + && error_code == "expected_token_hash_mismatch"); + if not_ours { + return Err(ReleaseIdentityConflict { + agent_name: agent_name.to_string(), + status: status.as_u16(), + code: error_code.to_string(), + } + .into()); + } anyhow::bail!("owned identity cleanup rejected (HTTP {status}); retry/reconcile without deleting a replacement"); } if result["data"]["status"] != "completed" { diff --git a/crates/broker/src/runtime/identity_cleanup.rs b/crates/broker/src/runtime/identity_cleanup.rs index df6343cd3..96742a06f 100644 --- a/crates/broker/src/runtime/identity_cleanup.rs +++ b/crates/broker/src/runtime/identity_cleanup.rs @@ -9,8 +9,39 @@ use std::{collections::HashMap, path::Path}; use tokio::{sync::oneshot, task::JoinHandle}; const CLEANUP_RETRY_DELAY: Duration = Duration::from_secs(5); +const CLEANUP_RETRY_MAX_DELAY: Duration = Duration::from_secs(300); const CLEANUP_ACK_TIMEOUT: Duration = Duration::from_secs(30); +/// The retry delay doubles per failed attempt up to the cap, with ±25% +/// jitter, so a fleet of stuck cleanups cannot pile onto an already +/// pressured Relaycast (e.g. sustained `d1_pressure`/`workspace_busy`). +fn cleanup_retry_delay(failed_attempts: u32) -> Duration { + use rand::Rng; + let shift = failed_attempts.saturating_sub(1).min(6); + let base = CLEANUP_RETRY_DELAY + .saturating_mul(1u32 << shift) + .min(CLEANUP_RETRY_MAX_DELAY); + let base_ms = base.as_millis() as u64; + let jitter = base_ms / 4; + Duration::from_millis(rand::thread_rng().gen_range(base_ms - jitter..=base_ms + jitter)) +} + +/// What an owned-cleanup attempt proved about the remote identity. +pub(super) enum CleanupOutcome { + /// The guarded delete (or binding-only release) was confirmed. + Confirmed, + /// The remote identity is definitively not this generation's — gone or + /// re-registered by a replacement. Nothing is left for this cleanup to + /// delete, so custody can release instead of retrying a conflict that + /// can never resolve. + IdentityNotOurs, + /// No captured credential hash exists for this generation, so the + /// guarded delete can never run — retrying can never succeed. Releasing + /// custody matches the journal's restart behavior (un-capturable + /// cleanups are never persisted). + Unverifiable, +} + pub(super) enum CleanupCompletion { Api( oneshot::Sender>, @@ -80,8 +111,8 @@ pub(crate) struct PendingIdentityCleanup { pub(super) agent_id: Option, pub(super) expected_token_hash: Result, deregistered: Arc, - pub(super) attempts: u8, - task: Option>>, + pub(super) attempts: u32, + task: Option>>, pub(super) retry_at: Instant, pub(super) completions: Vec, } @@ -99,7 +130,7 @@ fn start_attempt( delete_identity: bool, expected_token_hash: Result, deregistered: Arc, -) -> JoinHandle> { +) -> JoinHandle> { inventory.remove(&name); let acknowledgement = (|| { if deregistered.load(Ordering::Acquire) { @@ -139,16 +170,31 @@ fn start_attempt( // Only delete an identity created by this generation. A supplied token // can request binding teardown, but never grants deletion ownership. if delete_identity { - http.release_agent_identity_guarded( - &name, - Some("owned worker cleanup"), - true, - Some(&expected_token_hash?), - ) - .await - .map_err(|error| error.to_string())?; + let Ok(expected_token_hash) = expected_token_hash else { + return Ok(CleanupOutcome::Unverifiable); + }; + match http + .release_agent_identity_guarded( + &name, + Some("owned worker cleanup"), + true, + Some(&expected_token_hash), + ) + .await + { + Ok(()) => Ok(CleanupOutcome::Confirmed), + Err(error) + if error + .downcast_ref::() + .is_some() => + { + Ok(CleanupOutcome::IdentityNotOurs) + } + Err(error) => Err(error.to_string()), + } + } else { + Ok(CleanupOutcome::Confirmed) } - Ok(()) }) } @@ -393,8 +439,8 @@ impl BrokerRuntime { continue; } if pending.task.is_none() { - if pending.attempts < 5 && Instant::now() >= pending.retry_at { - pending.attempts += 1; + if Instant::now() >= pending.retry_at { + pending.attempts = pending.attempts.saturating_add(1); pending.task = Some(start_attempt( &self.fleet_control_tx, pending.agent_id.clone(), @@ -418,10 +464,11 @@ impl BrokerRuntime { // A fleet action retry retained only in this sandbox cannot // outlive its action result: Cloud tears the sandbox down as soon // as it receives a failed spawn. Keep fleet callers waiting while - // bounded retries remain, and report an unconfirmed cleanup only - // after the final attempt. API release callers retain their - // existing immediate-error contract because the broker remains - // alive for an explicit retry. + // the quick retries run, and report an unconfirmed cleanup once + // those are spent — even though jittered background retries keep + // running so the name cannot stay blocked forever. API release + // callers retain their existing immediate-error contract because + // the broker remains alive for an explicit retry. let completions = if result.is_err() && pending.attempts < 5 { let (retained, ready): (Vec<_>, Vec<_>) = std::mem::take(&mut pending.completions) .into_iter() @@ -432,14 +479,29 @@ impl BrokerRuntime { std::mem::take(&mut pending.completions) }; if let Err(error) = &result { - pending.retry_at = Instant::now() + CLEANUP_RETRY_DELAY; - if pending.attempts >= 5 { + pending.retry_at = Instant::now() + cleanup_retry_delay(pending.attempts); + if pending.attempts == 5 { + // Quick retries did not confirm the remote release — the + // waiters are answered above — but custody is retained + // and jittered background retries continue, so a + // sustained outage cannot leave the name blocked + // forever. tracing::error!(worker = %name, %generation, agent_id = ?pending.agent_id, %error, - "owned identity cleanup retries exhausted; name stays reserved until an explicit generation-matched release retry"); + "owned identity cleanup still unconfirmed after quick retries; continuing on a jittered backoff"); } else { - tracing::warn!(worker = %name, %generation, %error, "owned identity cleanup unconfirmed; name reserved for retry"); + tracing::warn!(worker = %name, %generation, %error, attempts = pending.attempts, + "owned identity cleanup unconfirmed; name reserved for retry"); } } else { + match &result { + Ok(CleanupOutcome::IdentityNotOurs) => { + tracing::info!(worker = %name, %generation, + "remote identity is no longer this generation's; releasing owned cleanup custody") + } + Ok(CleanupOutcome::Unverifiable) => tracing::warn!(worker = %name, %generation, + "owned cleanup has no captured credential hash to verify with; releasing custody"), + _ => {} + } let delete_identity = pending.delete_identity; let agent_id = pending.agent_id.clone(); self.workers.identity_cleanups.remove(&name); diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 8f80d356e..5e47d94ae 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -212,7 +212,7 @@ struct WorkerEventRuntimeFixture { } #[tokio::test] -async fn owned_cleanup_exhaustion_signals_once_and_explicit_release_restarts() { +async fn owned_cleanup_keeps_retrying_with_backoff_and_explicit_release_restarts() { use crate::listen_api::ListenApiRequest; use std::io::Write; use std::sync::Arc; @@ -274,7 +274,8 @@ async fn owned_cleanup_exhaustion_signals_once_and_explicit_release_restarts() { reply, }) .await; - for attempt in 1..=5 { + let mut expected_delay = Duration::from_secs(5); + for attempt in 1..=7u32 { loop { if let FleetControlCommand::DeregisterAgent { reply, .. } = fixture.fleet_control_rx.recv().await.unwrap() @@ -299,25 +300,29 @@ async fn owned_cleanup_exhaustion_signals_once_and_explicit_release_restarts() { fixture.runtime.workers.identity_cleanups[&name].attempts, attempt ); - fixture - .runtime - .workers - .identity_cleanups - .get_mut(&name) - .unwrap() - .retry_at = Instant::now(); - if attempt < 5 { - fixture.runtime.reconcile_identity_cleanups().await; - } - } - for _ in 0..6 { - fixture.runtime.reconcile_identity_cleanups().await; - } - while let Ok(command) = fixture.fleet_control_rx.try_recv() { + // Backoff doubles per failed attempt up to the 300s cap, jittered + // ±25% — retries must keep going past the old five-attempt cap so + // a sustained outage cannot leave the name blocked forever. + let delay = fixture.runtime.workers.identity_cleanups[&name] + .retry_at + .saturating_duration_since(Instant::now()); + let low = expected_delay.mul_f64(0.7); + let high = expected_delay.mul_f64(1.3); assert!( - !matches!(command, FleetControlCommand::DeregisterAgent { .. }), - "exhaustion must stop automatic retries" + delay >= low && delay <= high, + "attempt {attempt} retry delay {delay:?} outside [{low:?}, {high:?}]" ); + expected_delay = (expected_delay * 2).min(Duration::from_secs(300)); + if attempt < 7 { + fixture + .runtime + .workers + .identity_cleanups + .get_mut(&name) + .unwrap() + .retry_at = Instant::now(); + fixture.runtime.reconcile_identity_cleanups().await; + } } assert!(fixture .runtime @@ -351,19 +356,229 @@ async fn owned_cleanup_exhaustion_signals_once_and_explicit_release_restarts() { assert_eq!(fixture.runtime.workers.identity_cleanups[&name].attempts, 1); let text = String::from_utf8(logs.lock().unwrap().clone()).unwrap(); assert_eq!( - text.matches("owned identity cleanup retries exhausted") + text.matches("still unconfirmed after quick retries") .count(), 1 ); assert!(text.contains("ERROR")); assert!(text.contains("original-agent-id")); assert!(text.contains(&generation.to_string())); - assert!(text.contains("explicit generation-matched release retry")); } .with_subscriber(subscriber) .await; } +#[tokio::test] +async fn owned_cleanup_recovers_on_later_retry_and_frees_the_name() { + use crate::listen_api::ListenApiRequest; + use httpmock::{Method::POST, MockServer}; + use tokio::sync::oneshot; + let server = MockServer::start(); + let release = server.mock(|when, then| { + when.method(POST) + .path("/v1/agents/release") + .json_body_partial(json!({"name":"recovered", "delete_agent":true}).to_string()); + then.status(200) + .json_body(json!({"ok":true,"data":{"status":"completed"}})); + }); + let registry = make_worker_registry_with_worker("unrelated").await; + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + let name = WorkerName::from("recovered"); + let generation = Uuid::new_v4(); + let http = RelaycastHttpClient::new(Some(server.base_url()), "rk_live_test", "broker", "codex"); + http.seed_agent_token(&name, "owned-token"); + fixture + .runtime + .workers + .owned_spawn_generations + .insert(name.clone(), (generation, http)); + fixture + .runtime + .fleet_delivery_book + .bind_authoritative_identity(name.to_string(), "recovered-id".to_string()); + let (reply, released) = oneshot::channel(); + fixture + .runtime + .handle_api_request(ListenApiRequest::Release { + name: name.clone(), + reason: None, + expected_generation: Some(generation.to_string()), + delete_identity: true, + reply, + }) + .await; + // First attempt loses its fleet acknowledgement; the cleanup must retain + // custody and retry rather than blocking the name. + loop { + if let FleetControlCommand::DeregisterAgent { reply, .. } = + fixture.fleet_control_rx.recv().await.unwrap() + { + reply.send(Err("fixture rejection".into())).unwrap(); + break; + } + } + let before = Instant::now(); + tokio::time::timeout(Duration::from_secs(2), async { + loop { + fixture.runtime.reconcile_identity_cleanups().await; + if fixture.runtime.workers.identity_cleanups[&name].retry_at > before { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("cleanup rejection should be reconciled"); + fixture + .runtime + .workers + .identity_cleanups + .get_mut(&name) + .unwrap() + .retry_at = Instant::now(); + fixture.runtime.reconcile_identity_cleanups().await; + // Second attempt: acknowledgement succeeds, the guarded release + // completes, and the name is freed for reuse. + loop { + if let FleetControlCommand::DeregisterAgent { reply, .. } = + fixture.fleet_control_rx.recv().await.unwrap() + { + reply.send(Ok(())).unwrap(); + break; + } + } + tokio::time::timeout(Duration::from_secs(2), async { + loop { + fixture.runtime.reconcile_identity_cleanups().await; + if !fixture + .runtime + .workers + .identity_cleanups + .contains_key(&name) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("cleanup should complete on the retried attempt"); + release.assert_hits(1); + assert!(!fixture + .runtime + .workers + .owned_spawn_generations + .contains_key(&name)); + // The API caller keeps its immediate-error contract on the first failed + // attempt even though the retained retry later confirms the cleanup. + assert!(released.await.unwrap().is_err()); + let mut respawn = fixture.runtime.workers.workers["unrelated"].spec.clone(); + respawn.name = name.clone(); + let spawn_error = fixture + .runtime + .workers + .spawn(respawn, None, None, None, false, None, None, None, None) + .await + .err() + .map(|error| error.to_string()) + .unwrap_or_default(); + assert!( + !spawn_error.contains("pending owned cleanup"), + "freed name must not be rejected: {spawn_error}" + ); +} + +#[tokio::test] +async fn owned_cleanup_releases_custody_when_remote_identity_is_not_ours() { + use crate::listen_api::ListenApiRequest; + use httpmock::{Method::POST, MockServer}; + use tokio::sync::oneshot; + let server = MockServer::start(); + // The remote identity was already released or re-registered by a + // replacement: the guarded delete is rejected durably, so the cleanup + // completes instead of retrying a conflict that can never resolve. + let release = server.mock(|when, then| { + when.method(POST) + .path("/v1/agents/release") + .json_body_partial(json!({"name":"conflicted", "delete_agent":true}).to_string()); + then.status(409).json_body(json!({ + "ok": false, + "error": { "code": "agent_release_generation_conflict", "message": "generation changed" } + })); + }); + let registry = make_worker_registry_with_worker("unrelated").await; + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + let name = WorkerName::from("conflicted"); + let generation = Uuid::new_v4(); + let http = RelaycastHttpClient::new(Some(server.base_url()), "rk_live_test", "broker", "codex"); + http.seed_agent_token(&name, "owned-token"); + fixture + .runtime + .workers + .owned_spawn_generations + .insert(name.clone(), (generation, http)); + fixture + .runtime + .fleet_delivery_book + .bind_authoritative_identity(name.to_string(), "conflicted-id".to_string()); + let (reply, released) = oneshot::channel(); + fixture + .runtime + .handle_api_request(ListenApiRequest::Release { + name: name.clone(), + reason: None, + expected_generation: Some(generation.to_string()), + delete_identity: true, + reply, + }) + .await; + loop { + if let FleetControlCommand::DeregisterAgent { reply, .. } = + fixture.fleet_control_rx.recv().await.unwrap() + { + reply.send(Ok(())).unwrap(); + break; + } + } + tokio::time::timeout(Duration::from_secs(2), async { + loop { + fixture.runtime.reconcile_identity_cleanups().await; + if !fixture + .runtime + .workers + .identity_cleanups + .contains_key(&name) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("terminal release conflict should complete the cleanup"); + release.assert_hits(1); + assert!(!fixture + .runtime + .workers + .owned_spawn_generations + .contains_key(&name)); + assert!(released.await.unwrap().is_ok()); + let mut respawn = fixture.runtime.workers.workers["unrelated"].spec.clone(); + respawn.name = name.clone(); + let spawn_error = fixture + .runtime + .workers + .spawn(respawn, None, None, None, false, None, None, None, None) + .await + .err() + .map(|error| error.to_string()) + .unwrap_or_default(); + assert!( + !spawn_error.contains("pending owned cleanup"), + "freed name must not be rejected: {spawn_error}" + ); +} + #[tokio::test] async fn owned_cleanup_waits_off_actor_and_retains_custody_until_confirmed() { use crate::listen_api::ListenApiRequest; From 7ce7ba5e9bbb77019e40e2e25965307070b9ee3c Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Mon, 12 Oct 2026 15:28:52 -0700 Subject: [PATCH 02/14] fix(broker): bound owned-cleanup jitter, end persistent 404s, and report unverifiable releases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback on the owned identity cleanup retry: - Clamp the jittered retry delay at the five-minute cap instead of letting ±25% jitter push a capped base past it. - Treat a persistent 404 agent_not_found (after the bounded visibility-lag retries) as a terminal ownership conflict so deleted identities stop retrying forever. - Invalidate the cached registration credential when the remote identity resolves to a different generation, so a stale token can't be replayed against the new owner's row. - Free custody for unverifiable cleanups (no captured token hash) but report the release as an error and skip the completed-release tombstone, rather than claiming `identity: deleted` for a row that may still exist. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- CHANGELOG.md | 2 +- crates/broker/src/relaycast/ws.rs | 49 ++++++++++- crates/broker/src/runtime/identity_cleanup.rs | 49 ++++++++++- crates/broker/src/runtime/tests.rs | 88 ++++++++++++++++++- 4 files changed, 181 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e53b2a95..d78f05e91 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Owned worker-name cleanup no longer gives up after five tries and blocks the name until manual recovery. The broker keeps reconciling the release on a jittered backoff (up to five minutes between attempts), and completes the cleanup outright when Relaycast reports the remote identity is already gone or belongs to a newer generation — so a name stuck in "pending owned cleanup" frees itself once the remote state is reachable or resolved. +- Owned worker-name cleanup no longer gives up after five tries and blocks the name until manual recovery. The broker keeps reconciling the release on a jittered backoff (up to five minutes between attempts), and completes the cleanup outright when Relaycast reports the remote identity is already gone or belongs to a newer generation — so a name stuck in "pending owned cleanup" frees itself once the remote state is reachable or resolved. When the remote deletion cannot be verified (no captured registration token), the release reports the cleanup as unverifiable instead of claiming the identity was deleted, and a confirmed ownership conflict drops the cached registration credential so it cannot be replayed. - `npx @agent-relay/connect join` no longer waits 60 seconds and rolls back when the local probe is running but cannot match the command to a live Claude Code or Codex session (for example, when it was run from a separate terminal). It stops within about a second and says to run the command from the agent's own shell. A timeout that does happen now names the probe's last answer. `connect install` no longer reports such a probe as ready when its version cannot be checked. ## [13.4.0] - 2026-10-11 diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index a061e519a..c1730c0e0 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -451,7 +451,7 @@ impl RelaycastHttpClient { } } - fn invalidate_cached_registration(&self, agent_name: &str) { + pub(crate) fn invalidate_cached_registration(&self, agent_name: &str) { if let Some(registration) = self.registration.as_ref() { registration.invalidate_cached_registration(agent_name); } @@ -1428,7 +1428,13 @@ impl RelaycastHttpClient { "agent_release_generation_conflict" | "agent_identity_mismatch" )) || (status == reqwest::StatusCode::FORBIDDEN - && error_code == "expected_token_hash_mismatch"); + && error_code == "expected_token_hash_mismatch") + // The bounded visibility retries above already covered + // read-after-write lag; a 404 that outlives them means + // the identity is genuinely gone — the cleanup's goal + // is met and retrying can never change the answer. + || (status == reqwest::StatusCode::NOT_FOUND + && error_code == "agent_not_found"); if not_ours { return Err(ReleaseIdentityConflict { agent_name: agent_name.to_string(), @@ -3214,7 +3220,7 @@ mod tests { retry_workspace_busy_reconcile, with_registration_attempts, workspace_busy_reconcile_delay, workspace_busy_retry_allowed, ImpersonationAwareRegistrationError, MessageInjectionMode, RecipientReachability, RegRetryOutcome, RegisterIntent, RelaycastHttpClient, - RelaycastRegistrationError, MAX_AGENT_REGISTRATION_ELAPSED, + RelaycastRegistrationError, ReleaseIdentityConflict, MAX_AGENT_REGISTRATION_ELAPSED, MAX_AGENT_REGISTRATION_OUTER_TIMEOUT, MAX_AGENT_REGISTRATION_RETRY_DELAY, WORKSPACE_BUSY_ACTION_SAFETY_CAP, WORKSPACE_BUSY_CREATE_ONLY_SAFETY_CAP, WORKSPACE_BUSY_RECONCILE_BUDGET, WORKSPACE_BUSY_RECONCILE_SAFETY_CAP, @@ -5345,6 +5351,43 @@ mod tests { assert!(client.owned_identity_token_hash("owned-worker").is_err()); } + #[tokio::test] + async fn owned_identity_cleanup_treats_persistent_agent_not_found_as_terminal() { + use sha2::{Digest, Sha256}; + let server = MockServer::start(); + // The bounded visibility retries already covered read-after-write + // lag; a 404 that outlives them means the identity is genuinely gone + // and the cleanup's goal is met — not retried forever. + let release = server.mock(|when, then| { + when.method(POST).path("/v1/agents/release"); + then.status(404).json_body(json!({"ok":false,"error":{ + "code":"agent_not_found","message":"not present on this Relaycast target" + }})); + }); + let client = seeded_http_client(&server.base_url()); + client.seed_agent_token("owned-worker", "owned-token"); + let expected_hash = format!("{:x}", Sha256::digest(b"owned-token")); + + let error = client + .release_agent_identity_guarded_with_budget( + "owned-worker", + None, + true, + Some(&expected_hash), + &[1, 1], + Duration::from_secs(30), + ) + .await + .expect_err("a persistent agent_not_found resolves the cleanup, not a retry"); + + assert!( + error.downcast_ref::().is_some(), + "expected a durable identity conflict, got: {error:#}" + ); + // Two visibility retries plus the terminal verdict: three requests. + release.assert_hits(3); + } + #[tokio::test] async fn owned_identity_cleanup_has_one_overall_deadline() { use sha2::{Digest, Sha256}; diff --git a/crates/broker/src/runtime/identity_cleanup.rs b/crates/broker/src/runtime/identity_cleanup.rs index 96742a06f..ad786a539 100644 --- a/crates/broker/src/runtime/identity_cleanup.rs +++ b/crates/broker/src/runtime/identity_cleanup.rs @@ -15,6 +15,8 @@ const CLEANUP_ACK_TIMEOUT: Duration = Duration::from_secs(30); /// The retry delay doubles per failed attempt up to the cap, with ±25% /// jitter, so a fleet of stuck cleanups cannot pile onto an already /// pressured Relaycast (e.g. sustained `d1_pressure`/`workspace_busy`). +/// Jitter is applied inside the cap so the five-minute maximum is a real +/// bound, not a base the jitter can exceed. fn cleanup_retry_delay(failed_attempts: u32) -> Duration { use rand::Rng; let shift = failed_attempts.saturating_sub(1).min(6); @@ -24,6 +26,7 @@ fn cleanup_retry_delay(failed_attempts: u32) -> Duration { let base_ms = base.as_millis() as u64; let jitter = base_ms / 4; Duration::from_millis(rand::thread_rng().gen_range(base_ms - jitter..=base_ms + jitter)) + .min(CLEANUP_RETRY_MAX_DELAY) } /// What an owned-cleanup attempt proved about the remote identity. @@ -493,8 +496,15 @@ impl BrokerRuntime { "owned identity cleanup unconfirmed; name reserved for retry"); } } else { + let unverifiable = matches!(&result, Ok(CleanupOutcome::Unverifiable)); match &result { Ok(CleanupOutcome::IdentityNotOurs) => { + // A replacement holding the name means this + // generation's token is dead — and it may still sit + // in the shared registration cache, where the next + // spawn's 409 takeover would be handed it back. + // Drop it the way a confirmed delete does. + pending.http.invalidate_cached_registration(name.as_str()); tracing::info!(worker = %name, %generation, "remote identity is no longer this generation's; releasing owned cleanup custody") } @@ -514,7 +524,12 @@ impl BrokerRuntime { .is_some_and(|(owned, _)| *owned == generation) { self.workers.owned_spawn_generations.remove(&name); - if delete_identity { + // Only a remotely-proven release earns the completion + // tombstone: an unverifiable cleanup freed custody without + // proving the remote identity is gone, and the tombstone's + // `{"identity":"deleted"}` fast-path must not repeat a + // claim the cleanup could not confirm. + if delete_identity && !unverifiable { self.workers .completed_owned_releases .push_back((name.clone(), generation)); @@ -529,7 +544,16 @@ impl BrokerRuntime { persist_journal(&self.workers); } for completion in completions { - let cleanup_error = result.as_ref().err().map(|error| format!("owned identity cleanup unconfirmed for {name} generation {generation}; retry retained: {error}")); + let cleanup_error = match &result { + Err(error) => Some(format!("owned identity cleanup unconfirmed for {name} generation {generation}; retry retained: {error}")), + // Custody freed, but nothing remote was proven — callers + // must hear that, not `identity: deleted` for a row that + // may still exist. + Ok(CleanupOutcome::Unverifiable) => Some(format!( + "owned identity cleanup unverifiable for {name} generation {generation}; name released without remote confirmation" + )), + _ => None, + }; if let Some(delivery_ack) = send_cleanup_completion(&self.fleet_completion_tx, completion, cleanup_error) { @@ -539,3 +563,24 @@ impl BrokerRuntime { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cleanup_retry_delay_never_exceeds_the_cap() { + // The ±25% jitter is applied inside the five-minute cap — a capped + // base must not be pushed past the stated maximum. + for attempt in [6_u32, 7, 8, 20, u32::MAX] { + for _ in 0..200 { + assert!( + cleanup_retry_delay(attempt) <= CLEANUP_RETRY_MAX_DELAY, + "delay exceeded the cap at attempt {attempt}" + ); + } + } + // Lower attempts keep their jittered range below the cap anyway. + assert!(cleanup_retry_delay(1) <= CLEANUP_RETRY_MAX_DELAY); + } +} diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 5e47d94ae..82e3d5096 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -516,7 +516,7 @@ async fn owned_cleanup_releases_custody_when_remote_identity_is_not_ours() { .runtime .workers .owned_spawn_generations - .insert(name.clone(), (generation, http)); + .insert(name.clone(), (generation, http.clone())); fixture .runtime .fleet_delivery_book @@ -563,6 +563,92 @@ async fn owned_cleanup_releases_custody_when_remote_identity_is_not_ours() { .owned_spawn_generations .contains_key(&name)); assert!(released.await.unwrap().is_ok()); + // The conflict proves a replacement holds the name, so this generation's + // cached token is dead — a later spawn's takeover must not reuse it. + assert!(http.owned_identity_token_hash(&name).is_err()); + let mut respawn = fixture.runtime.workers.workers["unrelated"].spec.clone(); + respawn.name = name.clone(); + let spawn_error = fixture + .runtime + .workers + .spawn(respawn, None, None, None, false, None, None, None, None) + .await + .err() + .map(|error| error.to_string()) + .unwrap_or_default(); + assert!( + !spawn_error.contains("pending owned cleanup"), + "freed name must not be rejected: {spawn_error}" + ); +} + +#[tokio::test] +async fn owned_cleanup_reports_unverifiable_release_without_claiming_delete() { + use crate::listen_api::ListenApiRequest; + use tokio::sync::oneshot; + let registry = make_worker_registry_with_worker("unrelated").await; + let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); + let name = WorkerName::from("unverifiable"); + let generation = Uuid::new_v4(); + // No seeded credential: the guarded delete can never run, so the cleanup + // frees custody rather than retrying forever — but the caller must hear + // "unverifiable", not `identity: deleted` for a row that may still exist. + let http = RelaycastHttpClient::new( + Some("http://127.0.0.1:1".into()), + "rk_live_fixture", + "broker", + "codex", + ); + fixture + .runtime + .workers + .owned_spawn_generations + .insert(name.clone(), (generation, http)); + let (reply, released) = oneshot::channel(); + fixture + .runtime + .handle_api_request(ListenApiRequest::Release { + name: name.clone(), + reason: None, + expected_generation: Some(generation.to_string()), + delete_identity: true, + reply, + }) + .await; + tokio::time::timeout(Duration::from_secs(2), async { + loop { + fixture.runtime.reconcile_identity_cleanups().await; + if !fixture + .runtime + .workers + .identity_cleanups + .contains_key(&name) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("an unverifiable cleanup must not retry forever"); + let error = released + .await + .unwrap() + .expect_err("unverified deletion must not report success"); + assert!(error.contains("unverifiable"), "{error}"); + assert!(!fixture + .runtime + .workers + .owned_spawn_generations + .contains_key(&name)); + // Nothing remote was proven, so the cleanup earns no completion + // tombstone — a replayed release must not answer `identity: deleted`. + assert!(fixture + .runtime + .workers + .completed_owned_releases + .iter() + .all(|(n, _)| *n != name)); let mut respawn = fixture.runtime.workers.workers["unrelated"].spec.clone(); respawn.name = name.clone(); let spawn_error = fixture From 2e44f9e3c305dcff0db72d82909e19fe61eb3b2f Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 03:08:28 -0700 Subject: [PATCH 03/14] fix(broker): verify owned cleanup terminal outcomes --- .../active/traj_jumh13krdh7q/trajectory.json | 70 ++++++ CHANGELOG.md | 2 +- crates/broker/src/relaycast/ws.rs | 140 ++++++++++- crates/broker/src/runtime/identity_cleanup.rs | 57 ++--- crates/broker/src/runtime/tests.rs | 56 ++--- .../1969-owned-cleanup-conflict/case.json | 21 ++ .../cases/1969-owned-cleanup-conflict/run.mjs | 221 ++++++++++++++++++ 7 files changed, 493 insertions(+), 74 deletions(-) create mode 100644 .agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json create mode 100644 tests/relayflows/cases/1969-owned-cleanup-conflict/case.json create mode 100644 tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json new file mode 100644 index 000000000..c4a06c7b5 --- /dev/null +++ b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json @@ -0,0 +1,70 @@ +{ + "id": "traj_jumh13krdh7q", + "version": 1, + "task": { + "title": "Finish relay PR 1969 checks and review threads without bypassing required approval", + "source": { + "system": "plain", + "id": "relay#1969" + } + }, + "status": "active", + "startedAt": "2026-10-13T09:52:37.132Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-10-13T09:58:24.852Z" + } + ], + "chapters": [ + { + "id": "chap_wo294hy0g2za", + "title": "Work", + "agentName": "default", + "startedAt": "2026-10-13T09:58:24.852Z", + "events": [ + { + "ts": 1791885504852, + "type": "decision", + "content": "Require exact-token evidence before treating a persistent release 404 as terminal: Require exact-token evidence before treating a persistent release 404 as terminal", + "raw": { + "question": "Require exact-token evidence before treating a persistent release 404 as terminal", + "chosen": "Require exact-token evidence before treating a persistent release 404 as terminal", + "alternatives": [], + "reasoning": "The by-name release index is eventually consistent. A successful exact-token read proves the identity is still live and retains custody; an invalid/not-found exact token or a locally proven replacement generation permits safe completion. Missing ownership proof remains retryable and keeps the worker name reserved." + }, + "significance": "high" + }, + { + "ts": 1791886108683, + "type": "reflection", + "content": "All five open review findings are covered by code or tests; the new RelayFlow proves base versus head generation-conflict behavior, while the full broker suite and build pass.", + "raw": { + "focalPoints": [ + "review-safety", + "proof", + "verification" + ], + "confidence": 0.9 + }, + "significance": "high", + "tags": [ + "focal:review-safety", + "focal:proof", + "focal:verification", + "confidence:0.9" + ] + } + ] + } + ], + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "7ce7ba5e9bbb77019e40e2e25965307070b9ee3c", + "endRef": "7ce7ba5e9bbb77019e40e2e25965307070b9ee3c" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index d78f05e91..470b242c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Owned worker-name cleanup no longer gives up after five tries and blocks the name until manual recovery. The broker keeps reconciling the release on a jittered backoff (up to five minutes between attempts), and completes the cleanup outright when Relaycast reports the remote identity is already gone or belongs to a newer generation — so a name stuck in "pending owned cleanup" frees itself once the remote state is reachable or resolved. When the remote deletion cannot be verified (no captured registration token), the release reports the cleanup as unverifiable instead of claiming the identity was deleted, and a confirmed ownership conflict drops the cached registration credential so it cannot be replayed. +- Owned worker-name cleanup no longer gives up after five tries and blocks the name until manual recovery. The broker keeps reconciling the release on a jittered backoff (up to five minutes between attempts), confirms persistent by-name absence through the captured agent token, and preserves a replacement generation's cached credential. Cleanup without captured ownership proof fails closed and keeps the name reserved instead of claiming an unverified deletion. - `npx @agent-relay/connect join` no longer waits 60 seconds and rolls back when the local probe is running but cannot match the command to a live Claude Code or Codex session (for example, when it was run from a separate terminal). It stops within about a second and says to run the command from the agent's own shell. A timeout that does happen now names the probe's last answer. `connect install` no longer reports such a probe as ready when its version cannot be checked. ## [13.4.0] - 2026-10-11 diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index c1730c0e0..6b3dbe4dd 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -1276,7 +1276,10 @@ impl RelaycastHttpClient { let total_budget = identity_visibility_total_budget( visibility_backoffs_ms, NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT, - ); + ) + // A persistent by-name 404 gets one independent exact-token probe + // before it can be considered terminal. + .saturating_add(NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT); self.release_agent_identity_guarded_with_budget( agent_name, reason, @@ -1422,19 +1425,76 @@ impl RelaycastHttpClient { // another write. Transient failures (429/503/d1_pressure, // timeouts) keep the generic retryable bail. let error_code = result["error"]["code"].as_str().unwrap_or_default(); + if status == reqwest::StatusCode::NOT_FOUND && error_code == "agent_not_found" { + let cached_token = self.cached_agent_token(agent_name).with_context(|| { + format!( + "owned identity cleanup for '{agent_name}' cannot verify a persistent by-name 404 because its captured token is no longer cached" + ) + })?; + let cached_hash = format!("{:x}", Sha256::digest(cached_token.as_bytes())); + if cached_hash != expected_token_hash { + return Err(ReleaseIdentityConflict { + agent_name: agent_name.to_string(), + status: reqwest::StatusCode::CONFLICT.as_u16(), + code: "cached_token_generation_changed".to_string(), + } + .into()); + } + let remaining = + deadline.saturating_duration_since(tokio::time::Instant::now()); + anyhow::ensure!( + !remaining.is_zero(), + "owned identity cleanup for '{agent_name}' exceeded its {total_budget:?} total budget before exact-token verification" + ); + let request_budget = + NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT.min(remaining); + let relay = self.relay_client().context( + "SDK relay client not initialized for exact-token cleanup verification", + )?; + match tokio::time::timeout( + request_budget, + relay.get_current_agent(cached_token), + ) + .await + { + Err(_) => anyhow::bail!( + "owned identity cleanup exact-token verification for '{agent_name}' timed out after {request_budget:?}" + ), + Ok(Ok(agent)) => anyhow::bail!( + "owned identity cleanup release lookup returned agent_not_found, but the captured token still authenticates as {}/{}; retaining custody for retry", + agent.name, + agent.id + ), + Ok(Err(error)) + if super::auth::is_agent_token_invalid(&error) + || matches!( + &error, + RelayError::Api { + status: 404, + code, + .. + } if code == "agent_not_found" + ) => + { + return Err(ReleaseIdentityConflict { + agent_name: agent_name.to_string(), + status: status.as_u16(), + code: error_code.to_string(), + } + .into()); + } + Ok(Err(error)) => anyhow::bail!( + "owned identity cleanup exact-token verification for '{agent_name}' failed: {error}" + ), + } + } let not_ours = (status == reqwest::StatusCode::CONFLICT && matches!( error_code, "agent_release_generation_conflict" | "agent_identity_mismatch" )) || (status == reqwest::StatusCode::FORBIDDEN - && error_code == "expected_token_hash_mismatch") - // The bounded visibility retries above already covered - // read-after-write lag; a 404 that outlives them means - // the identity is genuinely gone — the cleanup's goal - // is met and retrying can never change the answer. - || (status == reqwest::StatusCode::NOT_FOUND - && error_code == "agent_not_found"); + && error_code == "expected_token_hash_mismatch"); if not_ours { return Err(ReleaseIdentityConflict { agent_name: agent_name.to_string(), @@ -5305,7 +5365,7 @@ mod tests { .release_agent_identity("owned-worker", None, true) .await .expect_err("ownership failure must remain terminal"); - assert!(error.to_string().contains("HTTP 403")); + assert!(error.downcast_ref::().is_some()); cleanup.assert_hits(1); assert!(client.owned_identity_token_hash("owned-worker").is_ok()); } @@ -5355,15 +5415,22 @@ mod tests { async fn owned_identity_cleanup_treats_persistent_agent_not_found_as_terminal() { use sha2::{Digest, Sha256}; let server = MockServer::start(); - // The bounded visibility retries already covered read-after-write - // lag; a 404 that outlives them means the identity is genuinely gone - // and the cleanup's goal is met — not retried forever. + // A persistent by-name 404 is terminal only after the captured token's + // independent identity lookup also proves the row is gone. let release = server.mock(|when, then| { when.method(POST).path("/v1/agents/release"); then.status(404).json_body(json!({"ok":false,"error":{ "code":"agent_not_found","message":"not present on this Relaycast target" }})); }); + let token_probe = server.mock(|when, then| { + when.method(GET) + .path("/v1/agent") + .header("authorization", "Bearer owned-token"); + then.status(404).json_body(json!({"ok":false,"error":{ + "code":"agent_not_found","message":"token identity is gone" + }})); + }); let client = seeded_http_client(&server.base_url()); client.seed_agent_token("owned-worker", "owned-token"); let expected_hash = format!("{:x}", Sha256::digest(b"owned-token")); @@ -5386,6 +5453,55 @@ mod tests { ); // Two visibility retries plus the terminal verdict: three requests. release.assert_hits(3); + token_probe.assert_hits(1); + } + + #[tokio::test] + async fn owned_identity_cleanup_retains_custody_when_by_name_404_is_stale() { + use sha2::{Digest, Sha256}; + let server = MockServer::start(); + let release = server.mock(|when, then| { + when.method(POST).path("/v1/agents/release"); + then.status(404).json_body(json!({"ok":false,"error":{ + "code":"agent_not_found","message":"not visible by name" + }})); + }); + let token_probe = server.mock(|when, then| { + when.method(GET) + .path("/v1/agent") + .header("authorization", "Bearer owned-token"); + then.status(200).json_body(json!({"ok":true,"data":{ + "id":"agent-owned","workspace_id":"ws-test","name":"owned-worker", + "type":"agent","status":"online","persona":null,"metadata":{},"channels":[] + }})); + }); + let client = seeded_http_client(&server.base_url()); + client.seed_agent_token("owned-worker", "owned-token"); + let expected_hash = format!("{:x}", Sha256::digest(b"owned-token")); + + let error = client + .release_agent_identity_guarded_with_budget( + "owned-worker", + None, + true, + Some(&expected_hash), + &[1, 1], + Duration::from_secs(30), + ) + .await + .expect_err("a live exact-token identity must keep cleanup retryable"); + + assert!( + error.downcast_ref::().is_none(), + "a stale by-name 404 must not be treated as completed cleanup: {error:#}" + ); + assert!( + error.to_string().contains("still authenticates"), + "{error:#}" + ); + assert!(client.owned_identity_token_hash("owned-worker").is_ok()); + release.assert_hits(3); + token_probe.assert_hits(1); } #[tokio::test] diff --git a/crates/broker/src/runtime/identity_cleanup.rs b/crates/broker/src/runtime/identity_cleanup.rs index ad786a539..a6da3c3dc 100644 --- a/crates/broker/src/runtime/identity_cleanup.rs +++ b/crates/broker/src/runtime/identity_cleanup.rs @@ -38,11 +38,6 @@ pub(super) enum CleanupOutcome { /// delete, so custody can release instead of retrying a conflict that /// can never resolve. IdentityNotOurs, - /// No captured credential hash exists for this generation, so the - /// guarded delete can never run — retrying can never succeed. Releasing - /// custody matches the journal's restart behavior (un-capturable - /// cleanups are never persisted). - Unverifiable, } pub(super) enum CleanupCompletion { @@ -173,9 +168,10 @@ fn start_attempt( // Only delete an identity created by this generation. A supplied token // can request binding teardown, but never grants deletion ownership. if delete_identity { - let Ok(expected_token_hash) = expected_token_hash else { - return Ok(CleanupOutcome::Unverifiable); - }; + // Missing ownership proof is a cleanup failure, not evidence that + // the remote row is gone. Keep custody so a same-name spawn cannot + // collide with an identity we never verified or deleted. + let expected_token_hash = expected_token_hash?; match http .release_agent_identity_guarded( &name, @@ -496,21 +492,27 @@ impl BrokerRuntime { "owned identity cleanup unconfirmed; name reserved for retry"); } } else { - let unverifiable = matches!(&result, Ok(CleanupOutcome::Unverifiable)); - match &result { - Ok(CleanupOutcome::IdentityNotOurs) => { - // A replacement holding the name means this - // generation's token is dead — and it may still sit - // in the shared registration cache, where the next - // spawn's 409 takeover would be handed it back. - // Drop it the way a confirmed delete does. + if let Ok(CleanupOutcome::IdentityNotOurs) = &result { + // Drop only this cleanup generation's credential. A + // same-name replacement may have seeded a newer token + // while the guarded request was in flight. + if pending + .expected_token_hash + .as_ref() + .ok() + .is_some_and(|expected| { + pending + .http + .owned_identity_token_hash(name.as_str()) + .ok() + .as_deref() + == Some(expected.as_str()) + }) + { pending.http.invalidate_cached_registration(name.as_str()); - tracing::info!(worker = %name, %generation, - "remote identity is no longer this generation's; releasing owned cleanup custody") } - Ok(CleanupOutcome::Unverifiable) => tracing::warn!(worker = %name, %generation, - "owned cleanup has no captured credential hash to verify with; releasing custody"), - _ => {} + tracing::info!(worker = %name, %generation, + "remote identity is no longer this generation's; releasing owned cleanup custody") } let delete_identity = pending.delete_identity; let agent_id = pending.agent_id.clone(); @@ -524,12 +526,7 @@ impl BrokerRuntime { .is_some_and(|(owned, _)| *owned == generation) { self.workers.owned_spawn_generations.remove(&name); - // Only a remotely-proven release earns the completion - // tombstone: an unverifiable cleanup freed custody without - // proving the remote identity is gone, and the tombstone's - // `{"identity":"deleted"}` fast-path must not repeat a - // claim the cleanup could not confirm. - if delete_identity && !unverifiable { + if delete_identity { self.workers .completed_owned_releases .push_back((name.clone(), generation)); @@ -546,12 +543,6 @@ impl BrokerRuntime { for completion in completions { let cleanup_error = match &result { Err(error) => Some(format!("owned identity cleanup unconfirmed for {name} generation {generation}; retry retained: {error}")), - // Custody freed, but nothing remote was proven — callers - // must hear that, not `identity: deleted` for a row that - // may still exist. - Ok(CleanupOutcome::Unverifiable) => Some(format!( - "owned identity cleanup unverifiable for {name} generation {generation}; name released without remote confirmation" - )), _ => None, }; if let Some(delivery_ack) = diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 82e3d5096..2b7912a58 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -489,9 +489,10 @@ async fn owned_cleanup_recovers_on_later_retry_and_frees_the_name() { } #[tokio::test] -async fn owned_cleanup_releases_custody_when_remote_identity_is_not_ours() { +async fn owned_cleanup_releases_custody_without_clearing_replacement_token() { use crate::listen_api::ListenApiRequest; use httpmock::{Method::POST, MockServer}; + use sha2::Digest; use tokio::sync::oneshot; let server = MockServer::start(); // The remote identity was already released or re-registered by a @@ -532,6 +533,10 @@ async fn owned_cleanup_releases_custody_when_remote_identity_is_not_ours() { reply, }) .await; + // A same-name replacement may arrive while the guarded release is in + // flight. The conflict belongs to the captured generation and must not + // invalidate this newer credential. + http.seed_agent_token(&name, "replacement-token"); loop { if let FleetControlCommand::DeregisterAgent { reply, .. } = fixture.fleet_control_rx.recv().await.unwrap() @@ -563,9 +568,10 @@ async fn owned_cleanup_releases_custody_when_remote_identity_is_not_ours() { .owned_spawn_generations .contains_key(&name)); assert!(released.await.unwrap().is_ok()); - // The conflict proves a replacement holds the name, so this generation's - // cached token is dead — a later spawn's takeover must not reuse it. - assert!(http.owned_identity_token_hash(&name).is_err()); + assert_eq!( + http.owned_identity_token_hash(&name).unwrap(), + format!("{:x}", sha2::Sha256::digest(b"replacement-token")) + ); let mut respawn = fixture.runtime.workers.workers["unrelated"].spec.clone(); respawn.name = name.clone(); let spawn_error = fixture @@ -583,16 +589,16 @@ async fn owned_cleanup_releases_custody_when_remote_identity_is_not_ours() { } #[tokio::test] -async fn owned_cleanup_reports_unverifiable_release_without_claiming_delete() { +async fn owned_cleanup_without_ownership_proof_retains_custody() { use crate::listen_api::ListenApiRequest; use tokio::sync::oneshot; let registry = make_worker_registry_with_worker("unrelated").await; let mut fixture = worker_event_runtime_fixture(registry, HashMap::new()); let name = WorkerName::from("unverifiable"); let generation = Uuid::new_v4(); - // No seeded credential: the guarded delete can never run, so the cleanup - // frees custody rather than retrying forever — but the caller must hear - // "unverifiable", not `identity: deleted` for a row that may still exist. + // No seeded credential: the guarded delete cannot run. This must fail + // closed and retain custody rather than free a name whose remote identity + // was never verified or deleted. let http = RelaycastHttpClient::new( Some("http://127.0.0.1:1".into()), "rk_live_fixture", @@ -604,7 +610,7 @@ async fn owned_cleanup_reports_unverifiable_release_without_claiming_delete() { .workers .owned_spawn_generations .insert(name.clone(), (generation, http)); - let (reply, released) = oneshot::channel(); + let (reply, mut released) = oneshot::channel(); fixture .runtime .handle_api_request(ListenApiRequest::Release { @@ -615,34 +621,28 @@ async fn owned_cleanup_reports_unverifiable_release_without_claiming_delete() { reply, }) .await; - tokio::time::timeout(Duration::from_secs(2), async { + let error = tokio::time::timeout(Duration::from_secs(2), async { loop { fixture.runtime.reconcile_identity_cleanups().await; - if !fixture - .runtime - .workers - .identity_cleanups - .contains_key(&name) - { - break; + if let Ok(result) = released.try_recv() { + break result.expect_err("unverified deletion must not report success"); } tokio::task::yield_now().await; } }) .await - .expect("an unverifiable cleanup must not retry forever"); - let error = released - .await - .unwrap() - .expect_err("unverified deletion must not report success"); - assert!(error.contains("unverifiable"), "{error}"); - assert!(!fixture + .expect("an unverifiable cleanup must report its failure"); + assert!(error.contains("cached credential"), "{error}"); + assert!(fixture + .runtime + .workers + .identity_cleanups + .contains_key(&name)); + assert!(fixture .runtime .workers .owned_spawn_generations .contains_key(&name)); - // Nothing remote was proven, so the cleanup earns no completion - // tombstone — a replayed release must not answer `identity: deleted`. assert!(fixture .runtime .workers @@ -660,8 +660,8 @@ async fn owned_cleanup_reports_unverifiable_release_without_claiming_delete() { .map(|error| error.to_string()) .unwrap_or_default(); assert!( - !spawn_error.contains("pending owned cleanup"), - "freed name must not be rejected: {spawn_error}" + spawn_error.contains("pending owned cleanup"), + "unverified name must remain reserved: {spawn_error}" ); } diff --git a/tests/relayflows/cases/1969-owned-cleanup-conflict/case.json b/tests/relayflows/cases/1969-owned-cleanup-conflict/case.json new file mode 100644 index 000000000..1d68d8d02 --- /dev/null +++ b/tests/relayflows/cases/1969-owned-cleanup-conflict/case.json @@ -0,0 +1,21 @@ +{ + "version": 1, + "id": "1969-owned-cleanup-conflict", + "kind": "bugfix", + "title": "Release owned cleanup custody after a generation conflict", + "runner": { + "command": ["node", "tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs"] + }, + "requirements": ["broker-linux-x64"], + "timeoutSeconds": 900, + "expected": { + "base": { + "outcome": "bug", + "signature": "owned_cleanup_conflict_retains_name" + }, + "head": { + "outcome": "fixed", + "signature": "owned_cleanup_conflict_releases_name" + } + } +} diff --git a/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs new file mode 100644 index 000000000..edaec9d45 --- /dev/null +++ b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs @@ -0,0 +1,221 @@ +import assert from 'node:assert/strict'; +import { execFileSync, spawn } from 'node:child_process'; +import { access, mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import http from 'node:http'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const CASE_ID = '1969-owned-cleanup-conflict'; +const NAME = 'owned-conflict-probe'; +const GENERATION = '11111111-1111-4111-8111-111111111111'; +const TOKEN_HASH = 'bec092bff160b23541205064ab9f4485d6c2089760b1bb4e5f5ce19f0274aad3'; +const API_KEY = 'br_owned_conflict_probe'; + +const targetDir = required('RELAY_PR_PROOF_TARGET_DIR'); +const harnessDir = required('RELAY_PR_PROOF_HARNESS_DIR'); +const binary = required('RELAY_PR_PROOF_BROKER_BINARY'); +const resultPath = required('RELAY_PR_PROOF_RESULT_PATH'); +const arm = required('RELAY_PR_PROOF_ARM'); +assert.ok(arm === 'base' || arm === 'head'); +assert.equal( + execFileSync('git', ['-C', targetDir, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim(), + required(arm === 'base' ? 'RELAY_PR_PROOF_BASE_SHA' : 'RELAY_PR_PROOF_HEAD_SHA') +); +const relativeRunner = path.relative(path.resolve(harnessDir), fileURLToPath(import.meta.url)); +assert.ok(relativeRunner && !relativeRunner.startsWith('..') && !path.isAbsolute(relativeRunner)); +await access(binary, 1); + +const probeDir = await mkdtemp(path.join(tmpdir(), 'relayflow-owned-conflict-')); +const sockets = new Set(); +let broker; +let stdout = ''; +let stderr = ''; +let guardedReleaseCount = 0; + +const server = http.createServer(async (request, response) => { + const chunks = []; + for await (const chunk of request) chunks.push(chunk); + const body = chunks.length ? JSON.parse(Buffer.concat(chunks)) : {}; + const pathname = new URL(request.url, 'http://fixture.invalid').pathname; + const send = (status, value, ok = true) => { + response.writeHead(status, { 'content-type': 'application/json' }); + response.end(JSON.stringify(ok ? { ok: true, data: value } : { ok: false, error: value })); + }; + + if (request.method === 'POST' && pathname === '/v1/agents') { + send(201, { + id: `id_${body.name}`, + workspace_id: 'ws_owned_conflict_probe', + name: body.name, + token: `at_fixture_${body.name}`, + status: 'online', + created_at: '2026-01-01T00:00:00Z', + }); + return; + } + if (request.method === 'POST' && pathname === '/v1/agents/release') { + assert.equal(body.name, NAME); + assert.equal(body.delete_agent, true); + assert.equal(body.expected_token_hash, TOKEN_HASH); + guardedReleaseCount += 1; + send( + 409, + { + code: 'agent_release_generation_conflict', + message: 'the worker name belongs to a replacement generation', + }, + false + ); + return; + } + send(404, { code: 'not_found', message: 'unsupported fixture route' }, false); +}); +server.on('connection', (socket) => { + sockets.add(socket); + socket.once('close', () => sockets.delete(socket)); +}); + +try { + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const stateDir = path.join(probeDir, 'state'); + await mkdir(stateDir); + await writeFile( + path.join(stateDir, 'owned-cleanups.json'), + JSON.stringify({ + [NAME]: { generation: GENERATION, expected_token_hash: TOKEN_HASH, agent_id: null }, + }) + ); + + const cleanEnv = Object.fromEntries( + Object.entries(process.env).filter(([key]) => !/^(RELAY|AGENT_RELAY)/.test(key)) + ); + broker = spawn( + binary, + [ + 'init', + '--instance-name', + 'owned-conflict-probe-node', + '--workspace-key', + 'rk_owned_conflict_probe', + '--state-dir', + stateDir, + '--api-port', + '0', + '--channels', + '', + ], + { + cwd: probeDir, + env: { + ...cleanEnv, + RELAYCAST_BASE_URL: `http://127.0.0.1:${server.address().port}`, + RELAY_BROKER_API_KEY: API_KEY, + RELAY_NODE_ID: 'node_owned_conflict_probe', + RELAY_NODE_TOKEN: 'nt_owned_conflict_probe', + AGENT_RELAY_NO_DEBUG_FILES: '1', + AGENT_RELAY_TELEMETRY_DISABLED: '1', + }, + stdio: ['ignore', 'pipe', 'pipe'], + } + ); + broker.stdout.on('data', (chunk) => { + stdout = (stdout + chunk).slice(-12000); + }); + broker.stderr.on('data', (chunk) => { + stderr = (stderr + chunk).slice(-12000); + }); + + const apiPort = await waitForApiPort(); + const api = async (route, options = {}) => { + const response = await fetch(`http://127.0.0.1:${apiPort}${route}`, { + ...options, + headers: { 'content-type': 'application/json', 'x-api-key': API_KEY }, + signal: AbortSignal.timeout(45_000), + redirect: 'error', + }); + const raw = await response.text(); + let body; + try { + body = JSON.parse(raw); + } catch { + body = { raw }; + } + return { status: response.status, body }; + }; + await waitFor(async () => (await api('/api/session')).status === 200, 'broker API readiness'); + + const released = await api(`/api/spawned/${NAME}`, { + method: 'DELETE', + body: JSON.stringify({ expected_generation: GENERATION, delete_identity: true }), + }); + assert.ok(guardedReleaseCount >= 1, 'The restored cleanup never reached the guarded release endpoint.'); + + let outcome; + let signature; + let details; + if ( + released.status === 200 && + released.body?.process === 'stopped' && + released.body?.identity === 'deleted' + ) { + outcome = 'fixed'; + signature = 'owned_cleanup_conflict_releases_name'; + details = + 'The exact head broker treated the token-hash generation conflict as terminal for the retired generation, released custody, and completed the generation-matched API request.'; + } else { + assert.ok(released.status >= 400, JSON.stringify(released)); + assert.match(JSON.stringify(released.body), /cleanup unconfirmed|generation conflict/i); + outcome = 'bug'; + signature = 'owned_cleanup_conflict_retains_name'; + details = + 'The exact base broker treated the durable generation conflict as retryable, returned an unconfirmed cleanup error, and retained worker-name custody.'; + } + assert.equal(outcome, arm === 'head' ? 'fixed' : 'bug'); + + await mkdir(path.dirname(resultPath), { recursive: true }); + await writeFile( + resultPath, + `${JSON.stringify({ version: 1, caseId: CASE_ID, arm, outcome, signature, details })}\n`, + 'utf8' + ); +} finally { + await stopProcess(broker); + for (const socket of sockets) socket.destroy(); + await new Promise((resolve) => server.close(resolve)); + await rm(probeDir, { recursive: true, force: true }); +} + +function required(name) { + const value = process.env[name]?.trim(); + if (!value) throw new Error(`Missing ${name}.`); + return value; +} + +async function waitForApiPort() { + for (let attempt = 0; attempt < 200; attempt += 1) { + if (broker.exitCode !== null) throw new Error(`Broker exited before API startup: ${stderr}`); + const match = stdout.match(/API listening on http:\/\/127\.0\.0\.1:([1-9]\d{0,4})(?:\s|$)/); + if (match) return Number(match[1]); + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error(`Broker did not announce its API port: ${stderr}`); +} + +async function waitFor(predicate, label) { + for (let attempt = 0; attempt < 200; attempt += 1) { + if (broker.exitCode !== null) throw new Error(`Broker exited before ${label}: ${stderr}`); + if (await predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error(`Timed out waiting for ${label}: ${stderr}`); +} + +async function stopProcess(child) { + if (!child || child.exitCode !== null || child.signalCode !== null) return; + const exited = new Promise((resolve) => child.once('exit', resolve)); + child.kill('SIGTERM'); + const timer = setTimeout(() => child.kill('SIGKILL'), 2_000); + await exited; + clearTimeout(timer); +} From 20b6771c5d9742b3d8cb65dc312024b80dd3b0c5 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 03:28:32 -0700 Subject: [PATCH 04/14] fix(broker): recover durable owned cleanups --- .../active/traj_jumh13krdh7q/trajectory.json | 12 ++ crates/broker/src/relaycast/ws.rs | 192 ++++++++++++++++-- crates/broker/src/runtime/identity_cleanup.rs | 4 +- 3 files changed, 191 insertions(+), 17 deletions(-) diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json index c4a06c7b5..8de4803e1 100644 --- a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json +++ b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json @@ -55,6 +55,18 @@ "focal:verification", "confidence:0.9" ] + }, + { + "ts": 1791886828636, + "type": "decision", + "content": "Use durable agent-id workspace verification for cleanup recovery: Use durable agent-id workspace verification for cleanup recovery", + "raw": { + "question": "Use durable agent-id workspace verification for cleanup recovery", + "chosen": "Use durable agent-id workspace verification for cleanup recovery", + "alternatives": [], + "reasoning": "A restarted broker no longer has the exact bearer token, and persisting that reusable secret would violate journal hygiene. The existing journal agent_id plus an independent workspace lookup distinguishes confirmed absence, replacement, and a still-live original generation without weakening custody." + }, + "significance": "high" } ] } diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 6b3dbe4dd..db8ada61d 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -1254,12 +1254,35 @@ impl RelaycastHttpClient { reason: Option<&str>, delete_identity: bool, expected_token_hash: Option<&str>, + ) -> Result<()> { + self.release_agent_identity_guarded_for_agent( + agent_name, + reason, + delete_identity, + expected_token_hash, + None, + ) + .await + } + + /// The durable cleanup journal records the immutable agent id but not the + /// bearer token. Supplying that id lets a restored cleanup independently + /// distinguish a deleted row from a stale release-path name lookup without + /// persisting a reusable credential on disk. + pub(crate) async fn release_agent_identity_guarded_for_agent( + &self, + agent_name: &str, + reason: Option<&str>, + delete_identity: bool, + expected_token_hash: Option<&str>, + expected_agent_id: Option<&str>, ) -> Result<()> { self.release_agent_identity_guarded_with_backoffs( agent_name, reason, delete_identity, expected_token_hash, + expected_agent_id, &NODE_REGISTER_VISIBILITY_BACKOFFS_MS, ) .await @@ -1271,32 +1294,36 @@ impl RelaycastHttpClient { reason: Option<&str>, delete_identity: bool, expected_token_hash: Option<&str>, + expected_agent_id: Option<&str>, visibility_backoffs_ms: &[u64], ) -> Result<()> { let total_budget = identity_visibility_total_budget( visibility_backoffs_ms, NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT, ) - // A persistent by-name 404 gets one independent exact-token probe - // before it can be considered terminal. + // A persistent by-name 404 gets one independent exact-token or + // immutable-id probe before it can be considered terminal. .saturating_add(NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT); self.release_agent_identity_guarded_with_budget( agent_name, reason, delete_identity, expected_token_hash, + expected_agent_id, visibility_backoffs_ms, total_budget, ) .await } + #[allow(clippy::too_many_arguments)] async fn release_agent_identity_guarded_with_budget( &self, agent_name: &str, reason: Option<&str>, delete_identity: bool, expected_token_hash: Option<&str>, + expected_agent_id: Option<&str>, visibility_backoffs_ms: &[u64], total_budget: Duration, ) -> Result<()> { @@ -1426,20 +1453,6 @@ impl RelaycastHttpClient { // timeouts) keep the generic retryable bail. let error_code = result["error"]["code"].as_str().unwrap_or_default(); if status == reqwest::StatusCode::NOT_FOUND && error_code == "agent_not_found" { - let cached_token = self.cached_agent_token(agent_name).with_context(|| { - format!( - "owned identity cleanup for '{agent_name}' cannot verify a persistent by-name 404 because its captured token is no longer cached" - ) - })?; - let cached_hash = format!("{:x}", Sha256::digest(cached_token.as_bytes())); - if cached_hash != expected_token_hash { - return Err(ReleaseIdentityConflict { - agent_name: agent_name.to_string(), - status: reqwest::StatusCode::CONFLICT.as_u16(), - code: "cached_token_generation_changed".to_string(), - } - .into()); - } let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); anyhow::ensure!( @@ -1451,6 +1464,64 @@ impl RelaycastHttpClient { let relay = self.relay_client().context( "SDK relay client not initialized for exact-token cleanup verification", )?; + let Some(cached_token) = self.cached_agent_token(agent_name) else { + match tokio::time::timeout( + request_budget, + relay.get_agent(agent_name), + ) + .await + { + Err(_) => anyhow::bail!( + "owned identity cleanup workspace verification for '{agent_name}' timed out after {request_budget:?}" + ), + Ok(Ok(agent)) + if expected_agent_id + .is_none_or(|expected| agent.id == expected) => + { + anyhow::bail!( + "owned identity cleanup release lookup returned agent_not_found, but workspace verification still sees {}/{}; retaining custody for retry", + agent.name, + agent.id + ) + } + Ok(Ok(agent)) => { + return Err(ReleaseIdentityConflict { + agent_name: agent_name.to_string(), + status: reqwest::StatusCode::CONFLICT.as_u16(), + code: format!( + "agent_id_changed_from_{}_to_{}", + expected_agent_id.unwrap_or("unknown"), + agent.id + ), + } + .into()); + } + Ok(Err(RelayError::Api { + status: 404, + code, + .. + })) if code == "agent_not_found" => { + return Err(ReleaseIdentityConflict { + agent_name: agent_name.to_string(), + status: status.as_u16(), + code: error_code.to_string(), + } + .into()); + } + Ok(Err(error)) => anyhow::bail!( + "owned identity cleanup workspace verification for '{agent_name}' failed: {error}" + ), + } + }; + let cached_hash = format!("{:x}", Sha256::digest(cached_token.as_bytes())); + if cached_hash != expected_token_hash { + return Err(ReleaseIdentityConflict { + agent_name: agent_name.to_string(), + status: reqwest::StatusCode::CONFLICT.as_u16(), + code: "cached_token_generation_changed".to_string(), + } + .into()); + } match tokio::time::timeout( request_budget, relay.get_current_agent(cached_token), @@ -5441,6 +5512,7 @@ mod tests { None, true, Some(&expected_hash), + None, &[1, 1], Duration::from_secs(30), ) @@ -5485,6 +5557,7 @@ mod tests { None, true, Some(&expected_hash), + None, &[1, 1], Duration::from_secs(30), ) @@ -5504,6 +5577,91 @@ mod tests { token_probe.assert_hits(1); } + #[tokio::test] + async fn restored_owned_identity_cleanup_confirms_absence_without_plaintext_token() { + use sha2::{Digest, Sha256}; + let server = MockServer::start(); + let release = server.mock(|when, then| { + when.method(POST).path("/v1/agents/release"); + then.status(404).json_body(json!({"ok":false,"error":{ + "code":"agent_not_found","message":"not present on this Relaycast target" + }})); + }); + let identity_probe = server.mock(|when, then| { + when.method(GET) + .path("/v1/agents/owned-worker") + .header("authorization", "Bearer rk_live_test"); + then.status(404).json_body(json!({"ok":false,"error":{ + "code":"agent_not_found","message":"identity is gone" + }})); + }); + let client = seeded_http_client(&server.base_url()); + let expected_hash = format!("{:x}", Sha256::digest(b"owned-token")); + + let error = client + .release_agent_identity_guarded_with_budget( + "owned-worker", + None, + true, + Some(&expected_hash), + None, + &[1, 1], + Duration::from_secs(30), + ) + .await + .expect_err("a restored cleanup should complete after workspace absence proof"); + + assert!(error.downcast_ref::().is_some()); + release.assert_hits(3); + identity_probe.assert_hits(1); + } + + #[tokio::test] + async fn restored_owned_identity_cleanup_retains_custody_when_id_still_exists() { + use sha2::{Digest, Sha256}; + let server = MockServer::start(); + let release = server.mock(|when, then| { + when.method(POST).path("/v1/agents/release"); + then.status(404).json_body(json!({"ok":false,"error":{ + "code":"agent_not_found","message":"not visible by release path" + }})); + }); + let identity_probe = server.mock(|when, then| { + when.method(GET) + .path("/v1/agents/owned-worker") + .header("authorization", "Bearer rk_live_test"); + then.status(200).json_body(json!({"ok":true,"data":{ + "id":"agent-owned","workspace_id":"ws-test","name":"owned-worker", + "type":"agent","status":"online","persona":null,"metadata":{},"channels":[] + }})); + }); + let client = seeded_http_client(&server.base_url()); + let expected_hash = format!("{:x}", Sha256::digest(b"owned-token")); + + let error = client + .release_agent_identity_guarded_with_budget( + "owned-worker", + None, + true, + Some(&expected_hash), + Some("agent-owned"), + &[1, 1], + Duration::from_secs(30), + ) + .await + .expect_err("a still-live immutable id must keep cleanup retryable"); + + assert!(error.downcast_ref::().is_none()); + assert!( + error + .to_string() + .contains("workspace verification still sees"), + "{error:#}" + ); + release.assert_hits(3); + identity_probe.assert_hits(1); + } + #[tokio::test] async fn owned_identity_cleanup_has_one_overall_deadline() { use sha2::{Digest, Sha256}; @@ -5525,6 +5683,7 @@ mod tests { None, true, Some(&expected_hash), + None, &[5_000], Duration::from_millis(50), ) @@ -5580,6 +5739,7 @@ mod tests { None, true, Some(&expected_hash), + None, &[0; 7], ) .await diff --git a/crates/broker/src/runtime/identity_cleanup.rs b/crates/broker/src/runtime/identity_cleanup.rs index a6da3c3dc..8ac402c48 100644 --- a/crates/broker/src/runtime/identity_cleanup.rs +++ b/crates/broker/src/runtime/identity_cleanup.rs @@ -130,6 +130,7 @@ fn start_attempt( deregistered: Arc, ) -> JoinHandle> { inventory.remove(&name); + let cleanup_agent_id = agent_id.clone(); let acknowledgement = (|| { if deregistered.load(Ordering::Acquire) { return Ok(None); @@ -173,11 +174,12 @@ fn start_attempt( // collide with an identity we never verified or deleted. let expected_token_hash = expected_token_hash?; match http - .release_agent_identity_guarded( + .release_agent_identity_guarded_for_agent( &name, Some("owned worker cleanup"), true, Some(&expected_token_hash), + cleanup_agent_id.as_deref(), ) .await { From d7b5395e8e37e9319d7d4fac05dd02a8e27b6bfc Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 03:49:31 -0700 Subject: [PATCH 05/14] fix(broker): use exact release for restored cleanup --- .../active/traj_jumh13krdh7q/trajectory.json | 12 ++ crates/broker/src/relaycast/ws.rs | 193 ++++++++++-------- crates/broker/src/runtime/tests.rs | 36 ++-- .../cases/1969-owned-cleanup-conflict/run.mjs | 23 ++- 4 files changed, 155 insertions(+), 109 deletions(-) diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json index 8de4803e1..9d8c89dba 100644 --- a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json +++ b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json @@ -67,6 +67,18 @@ "reasoning": "A restarted broker no longer has the exact bearer token, and persisting that reusable secret would violate journal hygiene. The existing journal agent_id plus an independent workspace lookup distinguishes confirmed absence, replacement, and a still-live original generation without weakening custody." }, "significance": "high" + }, + { + "ts": 1791888565057, + "type": "decision", + "content": "Use Relaycast atomic exact-release for journaled cleanups: Use Relaycast atomic exact-release for journaled cleanups", + "raw": { + "question": "Use Relaycast atomic exact-release for journaled cleanups", + "chosen": "Use Relaycast atomic exact-release for journaled cleanups", + "alternatives": [], + "reasoning": "Relaycast 10.3 exposes a crash-safe immutable-ID and idempotency-key release endpoint. Using it avoids stale by-name absence proofs; legacy journals without agent_id remain retryable and fail closed." + }, + "significance": "high" } ] } diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index db8ada61d..ae1131e00 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -1266,9 +1266,9 @@ impl RelaycastHttpClient { } /// The durable cleanup journal records the immutable agent id but not the - /// bearer token. Supplying that id lets a restored cleanup independently - /// distinguish a deleted row from a stale release-path name lookup without - /// persisting a reusable credential on disk. + /// bearer token. Supplying that id selects Relaycast's atomic exact-release + /// contract, so restored cleanup never depends on a stale by-name lookup or + /// persists a reusable credential on disk. pub(crate) async fn release_agent_identity_guarded_for_agent( &self, agent_name: &str, @@ -1301,8 +1301,9 @@ impl RelaycastHttpClient { visibility_backoffs_ms, NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT, ) - // A persistent by-name 404 gets one independent exact-token or - // immutable-id probe before it can be considered terminal. + // A persistent legacy by-name 404 gets one independent exact-token + // probe before it can be considered terminal. Journaled immutable ids + // use the atomic exact-release route instead. .saturating_add(NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT); self.release_agent_identity_guarded_with_budget( agent_name, @@ -1346,6 +1347,10 @@ impl RelaycastHttpClient { expected_token_hash.context("owned cleanup has no captured credential hash")?; let mut body = serde_json::to_value(&request)?; body["expected_token_hash"] = Value::String(expected_token_hash.to_string()); + let exact_release = expected_agent_id.is_some(); + if let Some(expected_agent_id) = expected_agent_id { + body["expected_agent_id"] = Value::String(expected_agent_id.to_string()); + } let started = Instant::now(); let deadline = tokio::time::Instant::now() + total_budget; let client = reqwest::Client::new(); @@ -1359,17 +1364,32 @@ impl RelaycastHttpClient { ); let request_budget = NODE_REGISTER_VISIBILITY_REQUEST_TIMEOUT.min(remaining); let request = async { - let response = client + let path = if exact_release { + "/v1/agents/release-exact" + } else { + "/v1/agents/release" + }; + let mut request = client .post(format!( - "{}/v1/agents/release", + "{}{}", self.base_url .as_deref() .unwrap_or("https://cast.agentrelay.com") - .trim_end_matches('/') + .trim_end_matches('/'), + path, )) .bearer_auth(&self.api_key) .json(&body) - .timeout(request_budget) + .timeout(request_budget); + if let Some(expected_agent_id) = expected_agent_id { + request = request.header( + "Idempotency-Key", + format!( + "relay-owned-cleanup-{expected_agent_id}-{expected_token_hash}" + ), + ); + } + let response = request .send() .await .context("owned identity cleanup request failed")?; @@ -1403,7 +1423,10 @@ impl RelaycastHttpClient { } let agent_not_found = status == reqwest::StatusCode::NOT_FOUND && result["error"]["code"] == "agent_not_found"; - if agent_not_found && agent_not_found_retries < visibility_backoffs_ms.len() { + if !exact_release + && agent_not_found + && agent_not_found_retries < visibility_backoffs_ms.len() + { let delay_ms = visibility_backoffs_ms[agent_not_found_retries]; agent_not_found_retries += 1; tracing::warn!( @@ -1452,6 +1475,19 @@ impl RelaycastHttpClient { // another write. Transient failures (429/503/d1_pressure, // timeouts) keep the generic retryable bail. let error_code = result["error"]["code"].as_str().unwrap_or_default(); + if exact_release + && ((status == reqwest::StatusCode::NOT_FOUND + && error_code == "agent_not_found") + || (status == reqwest::StatusCode::CONFLICT + && error_code == "agent_identity_mismatch")) + { + return Err(ReleaseIdentityConflict { + agent_name: agent_name.to_string(), + status: status.as_u16(), + code: error_code.to_string(), + } + .into()); + } if status == reqwest::StatusCode::NOT_FOUND && error_code == "agent_not_found" { let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); @@ -1464,55 +1500,11 @@ impl RelaycastHttpClient { let relay = self.relay_client().context( "SDK relay client not initialized for exact-token cleanup verification", )?; - let Some(cached_token) = self.cached_agent_token(agent_name) else { - match tokio::time::timeout( - request_budget, - relay.get_agent(agent_name), + let cached_token = self.cached_agent_token(agent_name).with_context(|| { + format!( + "owned identity cleanup for '{agent_name}' cannot verify a persistent by-name 404 because its captured token is no longer cached and no immutable agent id was journaled" ) - .await - { - Err(_) => anyhow::bail!( - "owned identity cleanup workspace verification for '{agent_name}' timed out after {request_budget:?}" - ), - Ok(Ok(agent)) - if expected_agent_id - .is_none_or(|expected| agent.id == expected) => - { - anyhow::bail!( - "owned identity cleanup release lookup returned agent_not_found, but workspace verification still sees {}/{}; retaining custody for retry", - agent.name, - agent.id - ) - } - Ok(Ok(agent)) => { - return Err(ReleaseIdentityConflict { - agent_name: agent_name.to_string(), - status: reqwest::StatusCode::CONFLICT.as_u16(), - code: format!( - "agent_id_changed_from_{}_to_{}", - expected_agent_id.unwrap_or("unknown"), - agent.id - ), - } - .into()); - } - Ok(Err(RelayError::Api { - status: 404, - code, - .. - })) if code == "agent_not_found" => { - return Err(ReleaseIdentityConflict { - agent_name: agent_name.to_string(), - status: status.as_u16(), - code: error_code.to_string(), - } - .into()); - } - Ok(Err(error)) => anyhow::bail!( - "owned identity cleanup workspace verification for '{agent_name}' failed: {error}" - ), - } - }; + })?; let cached_hash = format!("{:x}", Sha256::digest(cached_token.as_bytes())); if cached_hash != expected_token_hash { return Err(ReleaseIdentityConflict { @@ -5578,7 +5570,7 @@ mod tests { } #[tokio::test] - async fn restored_owned_identity_cleanup_confirms_absence_without_plaintext_token() { + async fn restored_owned_identity_cleanup_without_agent_id_retains_custody() { use sha2::{Digest, Sha256}; let server = MockServer::start(); let release = server.mock(|when, then| { @@ -5587,14 +5579,6 @@ mod tests { "code":"agent_not_found","message":"not present on this Relaycast target" }})); }); - let identity_probe = server.mock(|when, then| { - when.method(GET) - .path("/v1/agents/owned-worker") - .header("authorization", "Bearer rk_live_test"); - then.status(404).json_body(json!({"ok":false,"error":{ - "code":"agent_not_found","message":"identity is gone" - }})); - }); let client = seeded_http_client(&server.base_url()); let expected_hash = format!("{:x}", Sha256::digest(b"owned-token")); @@ -5609,30 +5593,66 @@ mod tests { Duration::from_secs(30), ) .await - .expect_err("a restored cleanup should complete after workspace absence proof"); + .expect_err("a legacy journal without immutable identity proof must retain custody"); - assert!(error.downcast_ref::().is_some()); + assert!(error.downcast_ref::().is_none()); + assert!( + error.to_string().contains("no immutable agent id"), + "{error:#}" + ); release.assert_hits(3); - identity_probe.assert_hits(1); } #[tokio::test] - async fn restored_owned_identity_cleanup_retains_custody_when_id_still_exists() { + async fn restored_owned_identity_cleanup_confirms_absence_via_exact_release() { use sha2::{Digest, Sha256}; let server = MockServer::start(); let release = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); + when.method(POST) + .path("/v1/agents/release-exact") + .header_exists("idempotency-key") + .json_body_partial( + json!({ + "name":"owned-worker", + "expected_agent_id":"agent-owned", + "delete_agent":true + }) + .to_string(), + ); then.status(404).json_body(json!({"ok":false,"error":{ - "code":"agent_not_found","message":"not visible by release path" + "code":"agent_not_found","message":"exact identity is gone" }})); }); - let identity_probe = server.mock(|when, then| { - when.method(GET) - .path("/v1/agents/owned-worker") - .header("authorization", "Bearer rk_live_test"); - then.status(200).json_body(json!({"ok":true,"data":{ - "id":"agent-owned","workspace_id":"ws-test","name":"owned-worker", - "type":"agent","status":"online","persona":null,"metadata":{},"channels":[] + let client = seeded_http_client(&server.base_url()); + let expected_hash = format!("{:x}", Sha256::digest(b"owned-token")); + + let error = client + .release_agent_identity_guarded_with_budget( + "owned-worker", + None, + true, + Some(&expected_hash), + Some("agent-owned"), + &[1, 1], + Duration::from_secs(30), + ) + .await + .expect_err("atomic exact-release absence should complete restored cleanup"); + + assert!(error.downcast_ref::().is_some()); + release.assert_hits(1); + } + + #[tokio::test] + async fn restored_owned_identity_cleanup_rejects_replacement_via_exact_release() { + use sha2::{Digest, Sha256}; + let server = MockServer::start(); + let release = server.mock(|when, then| { + when.method(POST) + .path("/v1/agents/release-exact") + .header_exists("idempotency-key"); + then.status(409).json_body(json!({"ok":false,"error":{ + "code":"agent_identity_mismatch","message":"name belongs to a replacement" }})); }); let client = seeded_http_client(&server.base_url()); @@ -5649,17 +5669,10 @@ mod tests { Duration::from_secs(30), ) .await - .expect_err("a still-live immutable id must keep cleanup retryable"); + .expect_err("an exact immutable-id mismatch should complete retired cleanup"); - assert!(error.downcast_ref::().is_none()); - assert!( - error - .to_string() - .contains("workspace verification still sees"), - "{error:#}" - ); - release.assert_hits(3); - identity_probe.assert_hits(1); + assert!(error.downcast_ref::().is_some()); + release.assert_hits(1); } #[tokio::test] diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 2b7912a58..1606cedac 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -376,9 +376,9 @@ async fn owned_cleanup_recovers_on_later_retry_and_frees_the_name() { let server = MockServer::start(); let release = server.mock(|when, then| { when.method(POST) - .path("/v1/agents/release") + .path("/v1/agents/release-exact") .json_body_partial(json!({"name":"recovered", "delete_agent":true}).to_string()); - then.status(200) + then.status(201) .json_body(json!({"ok":true,"data":{"status":"completed"}})); }); let registry = make_worker_registry_with_worker("unrelated").await; @@ -500,11 +500,11 @@ async fn owned_cleanup_releases_custody_without_clearing_replacement_token() { // completes instead of retrying a conflict that can never resolve. let release = server.mock(|when, then| { when.method(POST) - .path("/v1/agents/release") + .path("/v1/agents/release-exact") .json_body_partial(json!({"name":"conflicted", "delete_agent":true}).to_string()); then.status(409).json_body(json!({ "ok": false, - "error": { "code": "agent_release_generation_conflict", "message": "generation changed" } + "error": { "code": "agent_identity_mismatch", "message": "generation changed" } })); }); let registry = make_worker_registry_with_worker("unrelated").await; @@ -673,9 +673,9 @@ async fn owned_cleanup_waits_off_actor_and_retains_custody_until_confirmed() { let server = MockServer::start(); let release = server.mock(|when, then| { when.method(POST) - .path("/v1/agents/release") + .path("/v1/agents/release-exact") .json_body_partial(json!({"name":"retired", "delete_agent":true}).to_string()); - then.status(200) + then.status(201) .json_body(json!({"ok":true,"data":{"status":"completed"}})); }); let registry = make_worker_registry_with_worker("unrelated").await; @@ -8807,7 +8807,7 @@ async fn node_owned_identity_create_reconcile_teardown_create_again() { }})); }); let mut delayed_cleanup = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); + when.method(POST).path("/v1/agents/release-exact"); then.status(503).json_body(json!({"ok":false,"error":{ "code":"cleanup_unavailable","message":"release service is temporarily unavailable" }})); @@ -8919,8 +8919,8 @@ async fn node_owned_identity_create_reconcile_teardown_create_again() { delayed_cleanup.delete(); let cleanup = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); - then.status(200) + when.method(POST).path("/v1/agents/release-exact"); + then.status(201) .json_body(json!({"ok":true,"data":{"status":"completed"}})); }); fixture @@ -9201,7 +9201,7 @@ async fn owned_cleanup_retries_delete_without_repeating_acknowledged_deregistrat use tokio::sync::oneshot; let server = MockServer::start(); let mut failed = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); + when.method(POST).path("/v1/agents/release-exact"); then.status(503); }); let registry = make_worker_registry_with_worker("unrelated").await; @@ -9253,8 +9253,8 @@ async fn owned_cleanup_retries_delete_without_repeating_acknowledged_deregistrat failed.assert_hits(1); failed.delete(); let success = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); - then.status(200) + when.method(POST).path("/v1/agents/release-exact"); + then.status(201) .json_body(json!({"ok":true,"data":{"status":"completed"}})); }); fixture @@ -9300,8 +9300,8 @@ async fn name_only_release_of_retired_owned_worker_deletes_directly_and_is_idemp let server = MockServer::start(); let release = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); - then.status(200) + when.method(POST).path("/v1/agents/release-exact"); + then.status(201) .json_body(json!({"ok":true,"data":{"status":"completed"}})); }); let registry = make_worker_registry_with_worker("unrelated").await; @@ -9393,8 +9393,8 @@ async fn completed_owned_release_history_keeps_prior_generation_idempotent_after let server = MockServer::start(); let release = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); - then.status(200) + when.method(POST).path("/v1/agents/release-exact"); + then.status(201) .json_body(json!({"ok":true,"data":{"status":"completed"}})); }); let name = WorkerName::from("retired-name-only"); @@ -9516,8 +9516,8 @@ async fn completed_owned_release_history_is_bounded_and_evictions_drop_old_gener let server = MockServer::start(); let release = server.mock(|when, then| { - when.method(POST).path("/v1/agents/release"); - then.status(200) + when.method(POST).path("/v1/agents/release-exact"); + then.status(201) .json_body(json!({"ok":true,"data":{"status":"completed"}})); }); let registry = make_worker_registry_with_worker("unrelated").await; diff --git a/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs index edaec9d45..ae1586f33 100644 --- a/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs +++ b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs @@ -69,6 +69,23 @@ const server = http.createServer(async (request, response) => { ); return; } + if (request.method === 'POST' && pathname === '/v1/agents/release-exact') { + assert.equal(body.name, NAME); + assert.equal(body.expected_agent_id, `id_${NAME}`); + assert.equal(body.delete_agent, true); + assert.equal(body.expected_token_hash, TOKEN_HASH); + assert.ok(request.headers['idempotency-key']); + guardedReleaseCount += 1; + send( + 409, + { + code: 'agent_identity_mismatch', + message: 'the immutable identity was replaced', + }, + false + ); + return; + } send(404, { code: 'not_found', message: 'unsupported fixture route' }, false); }); server.on('connection', (socket) => { @@ -83,7 +100,11 @@ try { await writeFile( path.join(stateDir, 'owned-cleanups.json'), JSON.stringify({ - [NAME]: { generation: GENERATION, expected_token_hash: TOKEN_HASH, agent_id: null }, + [NAME]: { + generation: GENERATION, + expected_token_hash: TOKEN_HASH, + agent_id: `id_${NAME}`, + }, }) ); From 6d99b6cdcd36916beb6875517c838297b010dc4c Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 03:53:00 -0700 Subject: [PATCH 06/14] test(flows): exercise exact cleanup release --- .../cases/1969-owned-cleanup-conflict/run.mjs | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs index ae1586f33..7fa54f327 100644 --- a/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs +++ b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs @@ -1,5 +1,6 @@ import assert from 'node:assert/strict'; import { execFileSync, spawn } from 'node:child_process'; +import crypto from 'node:crypto'; import { access, mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; import http from 'node:http'; import { tmpdir } from 'node:os'; @@ -11,6 +12,7 @@ const NAME = 'owned-conflict-probe'; const GENERATION = '11111111-1111-4111-8111-111111111111'; const TOKEN_HASH = 'bec092bff160b23541205064ab9f4485d6c2089760b1bb4e5f5ce19f0274aad3'; const API_KEY = 'br_owned_conflict_probe'; +const WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'; const targetDir = required('RELAY_PR_PROOF_TARGET_DIR'); const harnessDir = required('RELAY_PR_PROOF_HARNESS_DIR'); @@ -92,6 +94,27 @@ server.on('connection', (socket) => { sockets.add(socket); socket.once('close', () => sockets.delete(socket)); }); +server.on('upgrade', (request, socket) => { + const key = request.headers['sec-websocket-key']; + if (!key) { + socket.destroy(); + return; + } + const accept = crypto.createHash('sha1').update(key + WS_GUID).digest('base64'); + socket.write( + `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n` + ); + socket.on( + 'data', + createFrameReader((frame) => { + if (frame.type === 'node.register' || frame.type === 'inventory.sync' || frame.id) { + socket.write( + encodeTextFrame(JSON.stringify({ v: 1, type: 'reply', id: frame.id, ok: true, data: {} })) + ); + } + }) + ); +}); try { await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); @@ -240,3 +263,47 @@ async function stopProcess(child) { await exited; clearTimeout(timer); } + +function encodeTextFrame(text) { + const payload = Buffer.from(text, 'utf8'); + if (payload.length < 126) return Buffer.concat([Buffer.from([0x81, payload.length]), payload]); + const header = Buffer.alloc(4); + header[0] = 0x81; + header[1] = 126; + header.writeUInt16BE(payload.length, 2); + return Buffer.concat([header, payload]); +} + +function createFrameReader(onFrame) { + let buffer = Buffer.alloc(0); + return (chunk) => { + buffer = Buffer.concat([buffer, chunk]); + for (;;) { + if (buffer.length < 2) return; + const opcode = buffer[0] & 0x0f; + const masked = (buffer[1] & 0x80) !== 0; + let length = buffer[1] & 0x7f; + let offset = 2; + if (length === 126) { + if (buffer.length < offset + 2) return; + length = buffer.readUInt16BE(offset); + offset += 2; + } else if (length === 127) { + if (buffer.length < offset + 8) return; + length = Number(buffer.readBigUInt64BE(offset)); + offset += 8; + } + let mask; + if (masked) { + if (buffer.length < offset + 4) return; + mask = buffer.subarray(offset, offset + 4); + offset += 4; + } + if (buffer.length < offset + length) return; + const payload = Buffer.from(buffer.subarray(offset, offset + length)); + buffer = buffer.subarray(offset + length); + if (mask) for (let index = 0; index < payload.length; index += 1) payload[index] ^= mask[index % 4]; + if (opcode === 0x1) onFrame(JSON.parse(payload.toString('utf8'))); + } + }; +} From eef2c14a5192c4d43db0fb0cb55ef712d3c8b73b Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 03:53:11 -0700 Subject: [PATCH 07/14] style(flows): format cleanup proof --- tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs index 7fa54f327..3e4b6ebe6 100644 --- a/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs +++ b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs @@ -100,7 +100,10 @@ server.on('upgrade', (request, socket) => { socket.destroy(); return; } - const accept = crypto.createHash('sha1').update(key + WS_GUID).digest('base64'); + const accept = crypto + .createHash('sha1') + .update(key + WS_GUID) + .digest('base64'); socket.write( `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n` ); From 8bfe7d6533ed3da80f59778fe7d36fbd8a391258 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 03:57:59 -0700 Subject: [PATCH 08/14] fix(broker): retry exact release visibility --- crates/broker/src/relaycast/ws.rs | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index ae1131e00..70b99ddb7 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -1423,10 +1423,7 @@ impl RelaycastHttpClient { } let agent_not_found = status == reqwest::StatusCode::NOT_FOUND && result["error"]["code"] == "agent_not_found"; - if !exact_release - && agent_not_found - && agent_not_found_retries < visibility_backoffs_ms.len() - { + if agent_not_found && agent_not_found_retries < visibility_backoffs_ms.len() { let delay_ms = visibility_backoffs_ms[agent_not_found_retries]; agent_not_found_retries += 1; tracing::warn!( @@ -4911,7 +4908,7 @@ mod tests { .await .expect("explicit release should succeed"); - release.assert_hits(1); + release.assert_hits(3); } #[tokio::test] @@ -5640,7 +5637,9 @@ mod tests { .expect_err("atomic exact-release absence should complete restored cleanup"); assert!(error.downcast_ref::().is_some()); - release.assert_hits(1); + // Exact release still spends the bounded visibility window before + // accepting absence as terminal. + release.assert_hits(3); } #[tokio::test] From c7d37dfd4dd930e2a16b183866e093ca86b3beab Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 04:03:21 -0700 Subject: [PATCH 09/14] test(broker): restore explicit release count --- crates/broker/src/relaycast/ws.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 70b99ddb7..ee90cb51c 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -4908,7 +4908,7 @@ mod tests { .await .expect("explicit release should succeed"); - release.assert_hits(3); + release.assert_hits(1); } #[tokio::test] From 41f37e5b1e59ce922cef87cfb4bb2dca948ec8cb Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 04:12:26 -0700 Subject: [PATCH 10/14] test(flows): model exact agent release --- .../active/traj_jumh13krdh7q/trajectory.json | 12 ++++++++ .../fake-relaycast.mjs | 30 ++++++++++++++++--- 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json index 9d8c89dba..626413941 100644 --- a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json +++ b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json @@ -79,6 +79,18 @@ "reasoning": "Relaycast 10.3 exposes a crash-safe immutable-ID and idempotency-key release endpoint. Using it avoids stale by-name absence proofs; legacy journals without agent_id remain retryable and fail closed." }, "significance": "high" + }, + { + "ts": 1791889916352, + "type": "decision", + "content": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract: Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", + "raw": { + "question": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", + "chosen": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", + "alternatives": [], + "reasoning": "The production cleanup path now uses immutable agent-id and token-hash proof. The existing routing proof must accept and validate that exact release during teardown while preserving legacy release for its base arm." + }, + "significance": "high" } ] } diff --git a/tests/relayflows/cases/1615-api-send-recipient-reachability/fake-relaycast.mjs b/tests/relayflows/cases/1615-api-send-recipient-reachability/fake-relaycast.mjs index 04d2fed23..479793d87 100644 --- a/tests/relayflows/cases/1615-api-send-recipient-reachability/fake-relaycast.mjs +++ b/tests/relayflows/cases/1615-api-send-recipient-reachability/fake-relaycast.mjs @@ -19,6 +19,7 @@ export async function startFakeRelaycast({ directMessages: [], channelMessages: [], agentReads: [], + exactReleaseRequests: [], nodeFrames: [], nodeSocket: undefined, }; @@ -165,8 +166,30 @@ export async function startFakeRelaycast({ return; } + if (request.method === 'POST' && pathname === '/v1/agents/release-exact') { + state.exactReleaseRequests.push(body); + const expectedTokenHash = createHash('sha256').update('at_relayflow_recipient').digest('hex'); + if (body?.expected_agent_id !== state.agentId || body?.expected_token_hash !== expectedTokenHash) { + sendJson(response, 409, { + ok: false, + error: { + code: 'agent_identity_mismatch', + message: 'The requested agent generation does not match.', + }, + }); + return; + } + sendJson(response, 200, releaseResponse(body?.name ?? recipientName)); + return; + } + if (request.method === 'POST' && pathname === '/v1/agents/release') { - sendJson(response, 200, { + sendJson(response, 200, releaseResponse(body?.name ?? recipientName)); + return; + } + + function releaseResponse(name) { + return { ok: true, data: { invocation_id: 'invocation_release_relayflow_1615', @@ -174,12 +197,11 @@ export async function startFakeRelaycast({ handler_agent_id: null, handler_node_id: null, dispatched_node_id: null, - input: { name: body?.name ?? recipientName }, + input: { name }, status: 'completed', created_at: '2026-09-02T00:00:00.000Z', }, - }); - return; + }; } if (request.method === 'POST' && /^\/v1\/nodes\/[^/]+\/agents$/.test(pathname)) { From 2d403e082955c85c4d98bd6e74e433ea86ba70d2 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 04:25:14 -0700 Subject: [PATCH 11/14] test(fleet): pin exact-release engine --- .../active/traj_jumh13krdh7q/trajectory.json | 12 ++++++++++++ .github/workflows/fleet-e2e.yml | 8 ++++---- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json index 626413941..7e03d28dd 100644 --- a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json +++ b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json @@ -91,6 +91,18 @@ "reasoning": "The production cleanup path now uses immutable agent-id and token-hash proof. The existing routing proof must accept and validate that exact release during teardown while preserving legacy release for its base arm." }, "significance": "high" + }, + { + "ts": 1791890710219, + "type": "decision", + "content": "Pinned Fleet E2E to the released Relaycast v10.3 engine: Pinned Fleet E2E to the released Relaycast v10.3 engine", + "raw": { + "question": "Pinned Fleet E2E to the released Relaycast v10.3 engine", + "chosen": "Pinned Fleet E2E to the released Relaycast v10.3 engine", + "alternatives": [], + "reasoning": "The prior fixture predates /v1/agents/release-exact, so failed-spawn cleanup could not settle. The release workflow was completed first specifically to make the atomic contract available; the targeted unsupported-resume scenario passes against the v10.3 source tree." + }, + "significance": "high" } ] } diff --git a/.github/workflows/fleet-e2e.yml b/.github/workflows/fleet-e2e.yml index a636fb510..e28c26a80 100644 --- a/.github/workflows/fleet-e2e.yml +++ b/.github/workflows/fleet-e2e.yml @@ -65,15 +65,15 @@ jobs: # The relaycast engine lives in a separate repo. This E2E drives the # node-provider model and heartbeats whose load may be unreported for - # unbounded providers. This companion PR candidate supplies isolated - # registration, acknowledged cleanup and node-owned spawn status reads. - # It is a test pin, not a released or deployed product dependency. + # unbounded providers. This released engine supplies isolated + # registration, atomic generation-guarded cleanup and node-owned spawn + # status reads. It is a test pin, not a deployed product dependency. - name: Checkout relaycast engine (node-provider model) uses: actions/checkout@v4 with: persist-credentials: false repository: AgentWorkforce/relaycast - ref: a5c7ac66f64a650274fbe2c60e3cd5d3c9b87dcd # relaycast#387: verified served-node spawn completion + ref: aabc862dd0ef9ea77bbddb1590907530e30ac070 # relaycast v10.3.0: atomic exact-agent release path: relaycast-engine - name: Setup Node.js From e9053516ce87846df6d340a9a7548b3f77ebc913 Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 04:39:44 -0700 Subject: [PATCH 12/14] fix(broker): preserve release completion ordering --- .../active/traj_jumh13krdh7q/trajectory.json | 12 +++++ crates/broker/src/runtime/fleet.rs | 44 ++++++------------- tests/e2e/fleet/fleet-e2e.test.ts | 2 +- 3 files changed, 26 insertions(+), 32 deletions(-) diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json index 7e03d28dd..832c7f9f8 100644 --- a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json +++ b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json @@ -103,6 +103,18 @@ "reasoning": "The prior fixture predates /v1/agents/release-exact, so failed-spawn cleanup could not settle. The release workflow was completed first specifically to make the atomic contract available; the targeted unsupported-resume scenario passes against the v10.3 source tree." }, "significance": "high" + }, + { + "ts": 1791891580056, + "type": "decision", + "content": "Queued fleet release completion before inventory removal: Queued fleet release completion before inventory removal", + "raw": { + "question": "Queued fleet release completion before inventory removal", + "chosen": "Queued fleet release completion before inventory removal", + "alternatives": [], + "reasoning": "Relaycast v10.3 atomically authorizes a built-in release while the exact agent binding is still active. Sending agent.deregister first creates a deterministic generation conflict; FIFO action.result first lets the engine own the atomic lifecycle transition, after which local inventory can be pruned." + }, + "significance": "high" } ] } diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 1286e0de3..a7d45bd07 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -2090,32 +2090,6 @@ impl BrokerRuntime { self.pty_observability.remove(&name); - let mut deregistration_failed = false; - if outcome == super::relaycast_events::ReleaseOutcome::Released { - match deregister_fleet_agent(&self.fleet_control_tx, &self.fleet_delivery_book, &name) - .await - { - Ok(_) => { - prune_fleet_agent_state( - &self.fleet_control_tx, - &mut self.fleet_inventory, - &mut self.fleet_delivery_book, - &name, - ) - .await; - } - Err(error) => { - tracing::warn!(worker = %name, %error, "retaining fleet identity after release cleanup"); - deregistration_failed = true; - prune_fleet_inventory_entry( - &self.fleet_control_tx, - &mut self.fleet_inventory, - &name, - ) - .await; - } - } - } if let Some(pending) = self.pending_verified_spawns.remove(&name) { self.send_fleet_action_result(verified_spawn_failed_result( pending.invocation_id, @@ -2123,24 +2097,32 @@ impl BrokerRuntime { )) .await; } - self.publish_fleet_load(true).await; match outcome { - super::relaycast_events::ReleaseOutcome::Released if deregistration_failed => { - self.reply_action_error(&invoke.invocation_id, "release_deregistration_failed") - .await; - } super::relaycast_events::ReleaseOutcome::Released => { + // The built-in release action's result is the engine's atomic + // authorization point. Keep its binding active until this + // frame is first on the FIFO control channel; an earlier + // agent.deregister or inventory removal makes the completion + // fail closed with agent_release_generation_conflict. self.reply_action_output( &invoke.invocation_id, json!({ "released": true, "name": name.as_str() }), ) .await; + prune_fleet_agent_state( + &self.fleet_control_tx, + &mut self.fleet_inventory, + &mut self.fleet_delivery_book, + &name, + ) + .await; } super::relaycast_events::ReleaseOutcome::Failed => { self.reply_action_error(&invoke.invocation_id, "release_failed") .await; } } + self.publish_fleet_load(true).await; } async fn reply_action_output(&self, invocation_id: &str, output: Value) { diff --git a/tests/e2e/fleet/fleet-e2e.test.ts b/tests/e2e/fleet/fleet-e2e.test.ts index 04acf9574..eeabea83e 100644 --- a/tests/e2e/fleet/fleet-e2e.test.ts +++ b/tests/e2e/fleet/fleet-e2e.test.ts @@ -603,7 +603,7 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { }, { label: 'origin worker release confirmed', timeoutMs: 30_000 } ); - expect(released.status).toBe('completed'); + expect(released.status, JSON.stringify(released)).toBe('completed'); // The legacy broker release action stops the PTY and detaches its binding; // the fixture owner separately removes its retained engine identity. expect(await releaseAgent(engine, workspaceKey, name)).toBeLessThan(300); From 52757f16d261f24e9561b79276ebbca22571230d Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 05:39:57 -0700 Subject: [PATCH 13/14] fix(broker): atomically invalidate owned cache entries Identity cleanup read the cached token hash and then invalidated the cache under separate SDK locks, so a same-name replacement registration landing in between could be deleted. Both the synchronous release path and the background retry now use relaycast 9.0.0's invalidate_cached_registration_if_token_hash, which compares and removes under the cache lock. CreateNodeRequest's new optional machine_id is left unset. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../active/traj_jumh13krdh7q/trajectory.json | 130 ------------------ Cargo.lock | 5 +- crates/broker/Cargo.toml | 2 +- crates/broker/src/node_control.rs | 1 + crates/broker/src/relaycast/ws.rs | 20 ++- crates/broker/src/runtime/identity_cleanup.rs | 18 +-- 6 files changed, 24 insertions(+), 152 deletions(-) delete mode 100644 .agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json deleted file mode 100644 index 832c7f9f8..000000000 --- a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json +++ /dev/null @@ -1,130 +0,0 @@ -{ - "id": "traj_jumh13krdh7q", - "version": 1, - "task": { - "title": "Finish relay PR 1969 checks and review threads without bypassing required approval", - "source": { - "system": "plain", - "id": "relay#1969" - } - }, - "status": "active", - "startedAt": "2026-10-13T09:52:37.132Z", - "agents": [ - { - "name": "default", - "role": "lead", - "joinedAt": "2026-10-13T09:58:24.852Z" - } - ], - "chapters": [ - { - "id": "chap_wo294hy0g2za", - "title": "Work", - "agentName": "default", - "startedAt": "2026-10-13T09:58:24.852Z", - "events": [ - { - "ts": 1791885504852, - "type": "decision", - "content": "Require exact-token evidence before treating a persistent release 404 as terminal: Require exact-token evidence before treating a persistent release 404 as terminal", - "raw": { - "question": "Require exact-token evidence before treating a persistent release 404 as terminal", - "chosen": "Require exact-token evidence before treating a persistent release 404 as terminal", - "alternatives": [], - "reasoning": "The by-name release index is eventually consistent. A successful exact-token read proves the identity is still live and retains custody; an invalid/not-found exact token or a locally proven replacement generation permits safe completion. Missing ownership proof remains retryable and keeps the worker name reserved." - }, - "significance": "high" - }, - { - "ts": 1791886108683, - "type": "reflection", - "content": "All five open review findings are covered by code or tests; the new RelayFlow proves base versus head generation-conflict behavior, while the full broker suite and build pass.", - "raw": { - "focalPoints": [ - "review-safety", - "proof", - "verification" - ], - "confidence": 0.9 - }, - "significance": "high", - "tags": [ - "focal:review-safety", - "focal:proof", - "focal:verification", - "confidence:0.9" - ] - }, - { - "ts": 1791886828636, - "type": "decision", - "content": "Use durable agent-id workspace verification for cleanup recovery: Use durable agent-id workspace verification for cleanup recovery", - "raw": { - "question": "Use durable agent-id workspace verification for cleanup recovery", - "chosen": "Use durable agent-id workspace verification for cleanup recovery", - "alternatives": [], - "reasoning": "A restarted broker no longer has the exact bearer token, and persisting that reusable secret would violate journal hygiene. The existing journal agent_id plus an independent workspace lookup distinguishes confirmed absence, replacement, and a still-live original generation without weakening custody." - }, - "significance": "high" - }, - { - "ts": 1791888565057, - "type": "decision", - "content": "Use Relaycast atomic exact-release for journaled cleanups: Use Relaycast atomic exact-release for journaled cleanups", - "raw": { - "question": "Use Relaycast atomic exact-release for journaled cleanups", - "chosen": "Use Relaycast atomic exact-release for journaled cleanups", - "alternatives": [], - "reasoning": "Relaycast 10.3 exposes a crash-safe immutable-ID and idempotency-key release endpoint. Using it avoids stale by-name absence proofs; legacy journals without agent_id remain retryable and fail closed." - }, - "significance": "high" - }, - { - "ts": 1791889916352, - "type": "decision", - "content": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract: Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", - "raw": { - "question": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", - "chosen": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", - "alternatives": [], - "reasoning": "The production cleanup path now uses immutable agent-id and token-hash proof. The existing routing proof must accept and validate that exact release during teardown while preserving legacy release for its base arm." - }, - "significance": "high" - }, - { - "ts": 1791890710219, - "type": "decision", - "content": "Pinned Fleet E2E to the released Relaycast v10.3 engine: Pinned Fleet E2E to the released Relaycast v10.3 engine", - "raw": { - "question": "Pinned Fleet E2E to the released Relaycast v10.3 engine", - "chosen": "Pinned Fleet E2E to the released Relaycast v10.3 engine", - "alternatives": [], - "reasoning": "The prior fixture predates /v1/agents/release-exact, so failed-spawn cleanup could not settle. The release workflow was completed first specifically to make the atomic contract available; the targeted unsupported-resume scenario passes against the v10.3 source tree." - }, - "significance": "high" - }, - { - "ts": 1791891580056, - "type": "decision", - "content": "Queued fleet release completion before inventory removal: Queued fleet release completion before inventory removal", - "raw": { - "question": "Queued fleet release completion before inventory removal", - "chosen": "Queued fleet release completion before inventory removal", - "alternatives": [], - "reasoning": "Relaycast v10.3 atomically authorizes a built-in release while the exact agent binding is still active. Sending agent.deregister first creates a deterministic generation conflict; FIFO action.result first lets the engine own the atomic lifecycle transition, after which local inventory can be pruned." - }, - "significance": "high" - } - ] - } - ], - "commits": [], - "filesChanged": [], - "projectId": "AgentWorkforce/relay", - "tags": [], - "_trace": { - "startRef": "7ce7ba5e9bbb77019e40e2e25965307070b9ee3c", - "endRef": "7ce7ba5e9bbb77019e40e2e25965307070b9ee3c" - } -} \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 9e6333a0e..4f18feb0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2073,14 +2073,15 @@ dependencies = [ [[package]] name = "relaycast" -version = "8.0.1" +version = "9.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53adee6048a65af443c258f6a3f360bae1b523507141c14228698fd0b6ef9407" +checksum = "583647f5a1c61a8ec747b1099bd79b87f6e36b6fd10790e4d3b136d95bef8ed3" dependencies = [ "futures-util", "reqwest", "serde", "serde_json", + "sha2", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.24.0", diff --git a/crates/broker/Cargo.toml b/crates/broker/Cargo.toml index 767ac131b..22da73aa1 100644 --- a/crates/broker/Cargo.toml +++ b/crates/broker/Cargo.toml @@ -29,7 +29,7 @@ json5 = "0.4" sha2 = "0.10" shlex = "1.3" thiserror = "2.0" -relaycast = "=8.0.1" +relaycast = "9.0.0" tokio = { version = "1.44", features = ["full"] } tracing = "0.1" tracing-appender = "0.2" diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index 314cb3294..7bd58aba1 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -386,6 +386,7 @@ pub(crate) fn create_node_request( max_agents: None, tags: None, version: Some(broker_version.to_string()), + machine_id: None, } } diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index ee90cb51c..17fca3716 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -457,6 +457,20 @@ impl RelaycastHttpClient { } } + pub(crate) fn invalidate_cached_registration_if_token_hash( + &self, + agent_name: &str, + expected_token_hash: &str, + ) -> bool { + self.registration + .as_ref() + .as_ref() + .is_some_and(|registration| { + registration + .invalidate_cached_registration_if_token_hash(agent_name, expected_token_hash) + }) + } + pub(crate) fn relay_client(&self) -> Option<&RelayCast> { self.relay.as_ref().as_ref() } @@ -1568,11 +1582,7 @@ impl RelaycastHttpClient { if result["data"]["status"] != "completed" { anyhow::bail!("owned identity cleanup queued but unconfirmed; reconcile lifecycle invocation {}", result["data"]["invocation_id"]); } - if self.owned_identity_token_hash(agent_name).ok().as_deref() - == Some(expected_token_hash) - { - self.invalidate_cached_registration(agent_name); - } + self.invalidate_cached_registration_if_token_hash(agent_name, expected_token_hash); return Ok(()); } // Invalidate the cached token before the call so an ambiguous diff --git a/crates/broker/src/runtime/identity_cleanup.rs b/crates/broker/src/runtime/identity_cleanup.rs index 8ac402c48..588a9e3d2 100644 --- a/crates/broker/src/runtime/identity_cleanup.rs +++ b/crates/broker/src/runtime/identity_cleanup.rs @@ -498,20 +498,10 @@ impl BrokerRuntime { // Drop only this cleanup generation's credential. A // same-name replacement may have seeded a newer token // while the guarded request was in flight. - if pending - .expected_token_hash - .as_ref() - .ok() - .is_some_and(|expected| { - pending - .http - .owned_identity_token_hash(name.as_str()) - .ok() - .as_deref() - == Some(expected.as_str()) - }) - { - pending.http.invalidate_cached_registration(name.as_str()); + if let Ok(expected) = pending.expected_token_hash.as_ref() { + pending + .http + .invalidate_cached_registration_if_token_hash(name.as_str(), expected); } tracing::info!(worker = %name, %generation, "remote identity is no longer this generation's; releasing owned cleanup custody") From 47f8e20f26d187e40d4f8891e3798a2c5fd6689b Mon Sep 17 00:00:00 2001 From: khaliqgant Date: Tue, 13 Oct 2026 13:50:32 -0700 Subject: [PATCH 14/14] chore: restore the cleanup-fix trajectory record Co-Authored-By: Claude Opus 5.5 (1M context) --- .../active/traj_jumh13krdh7q/trajectory.json | 162 ++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 .agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json diff --git a/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json new file mode 100644 index 000000000..d60408945 --- /dev/null +++ b/.agentworkforce/trajectories/rs-relaycast-d1/active/traj_jumh13krdh7q/trajectory.json @@ -0,0 +1,162 @@ +{ + "id": "traj_jumh13krdh7q", + "version": 1, + "task": { + "title": "Finish relay PR 1969 checks and review threads without bypassing required approval", + "source": { + "system": "plain", + "id": "relay#1969" + } + }, + "status": "active", + "startedAt": "2026-10-13T09:52:37.132Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-10-13T09:58:24.852Z" + } + ], + "chapters": [ + { + "id": "chap_wo294hy0g2za", + "title": "Work", + "agentName": "default", + "startedAt": "2026-10-13T09:58:24.852Z", + "events": [ + { + "ts": 1791885504852, + "type": "decision", + "content": "Require exact-token evidence before treating a persistent release 404 as terminal: Require exact-token evidence before treating a persistent release 404 as terminal", + "raw": { + "question": "Require exact-token evidence before treating a persistent release 404 as terminal", + "chosen": "Require exact-token evidence before treating a persistent release 404 as terminal", + "alternatives": [], + "reasoning": "The by-name release index is eventually consistent. A successful exact-token read proves the identity is still live and retains custody; an invalid/not-found exact token or a locally proven replacement generation permits safe completion. Missing ownership proof remains retryable and keeps the worker name reserved." + }, + "significance": "high" + }, + { + "ts": 1791886108683, + "type": "reflection", + "content": "All five open review findings are covered by code or tests; the new RelayFlow proves base versus head generation-conflict behavior, while the full broker suite and build pass.", + "raw": { + "focalPoints": [ + "review-safety", + "proof", + "verification" + ], + "confidence": 0.9 + }, + "significance": "high", + "tags": [ + "focal:review-safety", + "focal:proof", + "focal:verification", + "confidence:0.9" + ] + }, + { + "ts": 1791886828636, + "type": "decision", + "content": "Use durable agent-id workspace verification for cleanup recovery: Use durable agent-id workspace verification for cleanup recovery", + "raw": { + "question": "Use durable agent-id workspace verification for cleanup recovery", + "chosen": "Use durable agent-id workspace verification for cleanup recovery", + "alternatives": [], + "reasoning": "A restarted broker no longer has the exact bearer token, and persisting that reusable secret would violate journal hygiene. The existing journal agent_id plus an independent workspace lookup distinguishes confirmed absence, replacement, and a still-live original generation without weakening custody." + }, + "significance": "high" + }, + { + "ts": 1791888565057, + "type": "decision", + "content": "Use Relaycast atomic exact-release for journaled cleanups: Use Relaycast atomic exact-release for journaled cleanups", + "raw": { + "question": "Use Relaycast atomic exact-release for journaled cleanups", + "chosen": "Use Relaycast atomic exact-release for journaled cleanups", + "alternatives": [], + "reasoning": "Relaycast 10.3 exposes a crash-safe immutable-ID and idempotency-key release endpoint. Using it avoids stale by-name absence proofs; legacy journals without agent_id remain retryable and fail closed." + }, + "significance": "high" + }, + { + "ts": 1791889916352, + "type": "decision", + "content": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract: Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", + "raw": { + "question": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", + "chosen": "Extended the 1615 RelayFlow fake Relaycast with the 10.3 release-exact contract", + "alternatives": [], + "reasoning": "The production cleanup path now uses immutable agent-id and token-hash proof. The existing routing proof must accept and validate that exact release during teardown while preserving legacy release for its base arm." + }, + "significance": "high" + }, + { + "ts": 1791890710219, + "type": "decision", + "content": "Pinned Fleet E2E to the released Relaycast v10.3 engine: Pinned Fleet E2E to the released Relaycast v10.3 engine", + "raw": { + "question": "Pinned Fleet E2E to the released Relaycast v10.3 engine", + "chosen": "Pinned Fleet E2E to the released Relaycast v10.3 engine", + "alternatives": [], + "reasoning": "The prior fixture predates /v1/agents/release-exact, so failed-spawn cleanup could not settle. The release workflow was completed first specifically to make the atomic contract available; the targeted unsupported-resume scenario passes against the v10.3 source tree." + }, + "significance": "high" + }, + { + "ts": 1791891580056, + "type": "decision", + "content": "Queued fleet release completion before inventory removal: Queued fleet release completion before inventory removal", + "raw": { + "question": "Queued fleet release completion before inventory removal", + "chosen": "Queued fleet release completion before inventory removal", + "alternatives": [], + "reasoning": "Relaycast v10.3 atomically authorizes a built-in release while the exact agent binding is still active. Sending agent.deregister first creates a deterministic generation conflict; FIFO action.result first lets the engine own the atomic lifecycle transition, after which local inventory can be pruned." + }, + "significance": "high" + }, + { + "ts": 1791895058900, + "type": "decision", + "content": "Pin the draft Relay integration to Relaycast SDK PR commit 3fa31906: Pin the draft Relay integration to Relaycast SDK PR commit 3fa31906", + "raw": { + "question": "Pin the draft Relay integration to Relaycast SDK PR commit 3fa31906", + "chosen": "Pin the draft Relay integration to Relaycast SDK PR commit 3fa31906", + "alternatives": [], + "reasoning": "The atomic helper is not published and publishing is explicitly out of scope; an exact Git revision makes the preparatory commit reproducible and must be replaced by a released SDK version before merge." + }, + "significance": "high" + }, + { + "ts": 1791895058953, + "type": "reflection", + "content": "The cache race predates PR 1969 in the synchronous release path; PR 1969 adds a second background-cleanup occurrence. Both now use the SDK atomic helper in the draft integration.", + "raw": { + "focalPoints": [ + "scope", + "atomicity", + "integration" + ], + "confidence": 0.95 + }, + "significance": "high", + "tags": [ + "focal:scope", + "focal:atomicity", + "focal:integration", + "confidence:0.95" + ] + } + ] + } + ], + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relay", + "tags": [], + "_trace": { + "startRef": "7ce7ba5e9bbb77019e40e2e25965307070b9ee3c", + "endRef": "7ce7ba5e9bbb77019e40e2e25965307070b9ee3c" + } +} \ No newline at end of file