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 0000000000..d604089454 --- /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 diff --git a/.github/workflows/fleet-e2e.yml b/.github/workflows/fleet-e2e.yml index a636fb510c..e28c26a806 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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a159f0fbe..470b242c11 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), 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/Cargo.lock b/Cargo.lock index 9e6333a0eb..4f18feb0f2 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 767ac131b6..22da73aa19 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 314cb32941..7bd58aba1e 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 b3dab880c5..17fca37165 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 @@ -426,12 +451,26 @@ 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); } } + 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() } @@ -1229,12 +1268,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 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, + 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 @@ -1246,29 +1308,37 @@ 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 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, 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<()> { @@ -1291,6 +1361,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(); @@ -1304,17 +1378,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")?; @@ -1391,16 +1480,109 @@ 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(); + 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()); + 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", + )?; + 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" + ) + })?; + 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), + ) + .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"); + 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" { 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 @@ -3168,7 +3350,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, @@ -5253,7 +5435,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()); } @@ -5299,6 +5481,209 @@ 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(); + // 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")); + + 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 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); + 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), + None, + &[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] + 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| { + 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()); + 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 legacy journal without immutable identity proof must retain custody"); + + assert!(error.downcast_ref::().is_none()); + assert!( + error.to_string().contains("no immutable agent id"), + "{error:#}" + ); + release.assert_hits(3); + } + + #[tokio::test] + 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-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":"exact 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), + 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()); + // Exact release still spends the bounded visibility window before + // accepting absence as terminal. + release.assert_hits(3); + } + + #[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()); + 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("an exact immutable-id mismatch should complete retired cleanup"); + + assert!(error.downcast_ref::().is_some()); + release.assert_hits(1); + } + #[tokio::test] async fn owned_identity_cleanup_has_one_overall_deadline() { use sha2::{Digest, Sha256}; @@ -5320,6 +5705,7 @@ mod tests { None, true, Some(&expected_hash), + None, &[5_000], Duration::from_millis(50), ) @@ -5375,6 +5761,7 @@ mod tests { None, true, Some(&expected_hash), + None, &[0; 7], ) .await diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 1286e0de36..a7d45bd07d 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/crates/broker/src/runtime/identity_cleanup.rs b/crates/broker/src/runtime/identity_cleanup.rs index df6343cd31..588a9e3d25 100644 --- a/crates/broker/src/runtime/identity_cleanup.rs +++ b/crates/broker/src/runtime/identity_cleanup.rs @@ -9,8 +9,37 @@ 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`). +/// 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); + 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)) + .min(CLEANUP_RETRY_MAX_DELAY) +} + +/// 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, +} + pub(super) enum CleanupCompletion { Api( oneshot::Sender>, @@ -80,8 +109,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,8 +128,9 @@ fn start_attempt( delete_identity: bool, expected_token_hash: Result, deregistered: Arc, -) -> JoinHandle> { +) -> JoinHandle> { inventory.remove(&name); + let cleanup_agent_id = agent_id.clone(); let acknowledgement = (|| { if deregistered.load(Ordering::Acquire) { return Ok(None); @@ -139,16 +169,33 @@ 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())?; + // 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_for_agent( + &name, + Some("owned worker cleanup"), + true, + Some(&expected_token_hash), + cleanup_agent_id.as_deref(), + ) + .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 +440,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 +465,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 +480,32 @@ 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 { + 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 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") + } let delete_identity = pending.delete_identity; let agent_id = pending.agent_id.clone(); self.workers.identity_cleanups.remove(&name); @@ -467,7 +533,10 @@ 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}")), + _ => None, + }; if let Some(delivery_ack) = send_cleanup_completion(&self.fleet_completion_tx, completion, cleanup_error) { @@ -477,3 +546,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 8f80d356e3..1606cedac4 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,315 @@ 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-exact") + .json_body_partial(json!({"name":"recovered", "delete_agent":true}).to_string()); + then.status(201) + .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_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 + // 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-exact") + .json_body_partial(json!({"name":"conflicted", "delete_agent":true}).to_string()); + then.status(409).json_body(json!({ + "ok": false, + "error": { "code": "agent_identity_mismatch", "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.clone())); + 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; + // 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() + { + 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()); + 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 + .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_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 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", + "broker", + "codex", + ); + fixture + .runtime + .workers + .owned_spawn_generations + .insert(name.clone(), (generation, http)); + let (reply, mut 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; + let error = tokio::time::timeout(Duration::from_secs(2), async { + loop { + fixture.runtime.reconcile_identity_cleanups().await; + 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 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)); + 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 + .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"), + "unverified name must remain reserved: {spawn_error}" + ); +} + #[tokio::test] async fn owned_cleanup_waits_off_actor_and_retains_custody_until_confirmed() { use crate::listen_api::ListenApiRequest; @@ -372,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; @@ -8506,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" }})); @@ -8618,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 @@ -8900,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; @@ -8952,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 @@ -8999,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; @@ -9092,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"); @@ -9215,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/e2e/fleet/fleet-e2e.test.ts b/tests/e2e/fleet/fleet-e2e.test.ts index 04acf9574d..eeabea83ed 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); 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 04d2fed230..479793d87a 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)) { 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 0000000000..1d68d8d02b --- /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 0000000000..3e4b6ebe67 --- /dev/null +++ b/tests/relayflows/cases/1969-owned-cleanup-conflict/run.mjs @@ -0,0 +1,312 @@ +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'; +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 WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'; + +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; + } + 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) => { + 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)); + 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: `id_${NAME}`, + }, + }) + ); + + 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); +} + +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'))); + } + }; +}