From f05113a20d71706193cfc1b68471ac1caa7642cf Mon Sep 17 00:00:00 2001 From: layered-relay-desktop-agent-host Date: Tue, 10 Nov 2026 10:11:20 +0000 Subject: [PATCH 1/3] fix(relay-node): bind a spawned worker's own session to its spawn registration A spawned Codex or Claude worker's session could be auto-activated (the Codex SessionStart hook or the periodic activation pass) in the window between launch and the spawn's registration being saved. spawned_targets() only excluded saved spawn registrations, so register_session created a second Relaycast agent, and the store race either dropped the SpawnRecord (breaking requester approval) or orphaned an agent. - spawned_targets() also counts in-flight spawn intents (spawning_targets). - register_session waits for the spawn registration of a spawning target and answers with it (name@, SpawnRecord intact); refuses otherwise. - create_registration never replaces a saved spawn registration and deletes the agent it just created if it lost that race. - The spawn save deletes a non-spawn identity it displaces. - codex SessionStart waits for the spawn registration instead of activating. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 1 + crates/relay-node/src/relay/agents.rs | 44 ++- .../src/relay/outbound/hooks/codex_stop.rs | 10 +- crates/relay-node/src/relay/spawn.rs | 64 ++++- crates/relay-node/src/relay/spawn/invoke.rs | 27 +- .../src/relay/spawn/tests/report_back.rs | 2 +- .../src/relay/spawn/tests/spawned_binding.rs | 272 ++++++++++++++++++ 7 files changed, 411 insertions(+), 9 deletions(-) create mode 100644 crates/relay-node/src/relay/spawn/tests/spawned_binding.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 1213a075d..d755ac43f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Authored TypeScript cloud-worker assignments resolve the CLI's pinned Surface from isolated run directories when no dependency is supplied, preserving assignment-provided packages. +- A spawned Codex or Claude worker no longer registers a second Relaycast agent when its own session connects before the spawn finishes: the session binds to the spawn's registration, keeping its `name@` address and the requester who approves it. ## [18.0.0] - 2026-11-09 diff --git a/crates/relay-node/src/relay/agents.rs b/crates/relay-node/src/relay/agents.rs index e3af91eff..adc19fbb8 100644 --- a/crates/relay-node/src/relay/agents.rs +++ b/crates/relay-node/src/relay/agents.rs @@ -211,6 +211,14 @@ fn not_registered(name: &str) -> RpcError { ) } +/// A session a spawn here started, asked to register on its own. +fn spawned_session() -> RpcError { + RpcError::new( + ErrorKind::Refused, + "This session is being spawned by Agent Relay and is registered as its spawned agent once the spawn completes.", + ) +} + fn name_rules() -> RpcError { RpcError::invalid_params("Use 2–48 lowercase letters, digits and single dashes.") } @@ -433,6 +441,16 @@ impl Core { } return self.answer_registered(existing, raw_name, _changing).await; } + // A session a spawn here has started is that spawn's agent: its own + // connection (its SessionStart hook, an activation pass, its own + // registration request) binds to the spawn's registration once the + // spawn saves it, and never gets a second identity. + if self.spawning_targets().contains(target) { + return match self.saved_spawn_registration(target).await { + Some(existing) => self.answer_registered(existing, raw_name, _changing).await, + None => Err(spawned_session()), + }; + } // An already-open Grok agent is pinned to the exact binding it is // registered against; delivery refuses any other. A Grok TUI session // listed from Grok's own registry has no binding yet: adding it binds @@ -539,7 +557,14 @@ impl Core { attempt += 1; candidate = suggested_name_numbered(session, attempt); } - Err(error) => return Err(error), + Err(error) => { + // A spawn saved this session's registration meanwhile: + // it is answered as that agent. + if let Some(existing) = self.saved_spawn_registration_now(target) { + return Ok(already_registered(&existing, &self.machine_prefix)); + } + return Err(error); + } } }; self.sync_agents().await; @@ -708,7 +733,17 @@ impl Core { retired_aliases: None, cleared: None, }; + let mut spawned = false; let saved = self.store.update(|store| { + // A spawn saved its registration for this session meanwhile: + // the session is that agent, so this one is never added. + if store + .registration_for(®istration.target) + .is_some_and(|existing| existing.spawn.is_some()) + { + spawned = true; + return false; + } let added = store.add(registration.clone(), agent.token.clone()); if added { if let Some(owner) = owner.as_ref().filter(|owner| owner.workspace_id.is_some()) { @@ -721,6 +756,11 @@ impl Core { }); match saved { Ok(true) => Ok(registration), + Ok(false) if spawned => { + self.delete_or_retry(&workspace.base_url, &workspace.key, &agent.name) + .await; + Err(spawned_session()) + } Ok(false) => { self.delete_or_retry(&workspace.base_url, &workspace.key, &agent.name) .await; @@ -748,7 +788,7 @@ impl Core { /// Deletes an agent this computer no longer holds the token of; when /// Relaycast cannot be reached, records the deletion to retry (in memory /// at least, when the store cannot be saved either). - async fn delete_or_retry(&self, base_url: &str, key: &Secret, name: &str) { + pub(crate) async fn delete_or_retry(&self, base_url: &str, key: &Secret, name: &str) { let workspace = self.relaycast_scope(base_url, key); let result = client::delete_agent(base_url, key, name).await; self.observe_relaycast_write_result(&workspace, &result); diff --git a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs index 491d95c01..412159704 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -227,7 +227,15 @@ impl Outbound { }) }; let mut identity = registered(); - if identity.is_none() { + if identity.is_none() && self.core.spawning_targets().contains(&session.target) { + // A thread a spawn here started is that spawn's agent: it waits + // for the spawn's registration, never one of its own. + identity = self + .core + .saved_spawn_registration(&session.target) + .await + .and(registered()); + } else if identity.is_none() { if !self.setup.auto_activates(&session).await { return Ok(json!({})); } diff --git a/crates/relay-node/src/relay/spawn.rs b/crates/relay-node/src/relay/spawn.rs index 817f4ae95..ecb8bd4d4 100644 --- a/crates/relay-node/src/relay/spawn.rs +++ b/crates/relay-node/src/relay/spawn.rs @@ -70,6 +70,11 @@ use crate::relay::daemon::Core; use crate::relay::store::{Registration, Target}; use std::collections::HashSet; +/// How long a spawned session's own registration waits for its spawn to +/// save the registration it binds to. +const SPAWN_SAVE_WAIT: std::time::Duration = std::time::Duration::from_secs(5); +const SPAWN_SAVE_POLL: std::time::Duration = std::time::Duration::from_millis(100); + /// The spawn node's name, distinct from a broker node this computer may /// also run: `-desktop`. pub fn node_name(machine_prefix: &str) -> String { @@ -114,18 +119,59 @@ impl Core { } /// The sessions this computer spawned, so discovery never registers - /// them a second time. + /// them a second time: those with a saved spawn registration, and those + /// a spawn in flight has started but not yet saved. A spawned session + /// can start (and Codex run its SessionStart hook) before its spawn's + /// registration is saved; it is never given an identity of its own in + /// that window. pub(crate) fn spawned_targets(&self) -> HashSet { - self.store.read(|store| { + let mut targets: HashSet = self.store.read(|store| { store .registrations .iter() .filter(|r| r.spawn.is_some()) .map(|r| r.target.clone()) .collect() + }); + targets.extend(self.spawning_targets()); + targets + } + + /// The spawn registration saved for `target`, if any. + pub(crate) fn saved_spawn_registration_now(&self, target: &Target) -> Option { + self.store.read(|store| { + store + .registration_for(target) + .filter(|registration| registration.spawn.is_some()) + .cloned() }) } + /// The spawn registration saved for `target`, waiting briefly while a + /// spawn in flight has started the session and not yet saved it. + pub(crate) async fn saved_spawn_registration(&self, target: &Target) -> Option { + let deadline = tokio::time::Instant::now() + SPAWN_SAVE_WAIT; + loop { + if let Some(registration) = self.saved_spawn_registration_now(target) { + return Some(registration); + } + if tokio::time::Instant::now() >= deadline || !self.spawning_targets().contains(target) + { + return None; + } + tokio::time::sleep(SPAWN_SAVE_POLL).await; + } + } + + /// The sessions spawns in flight here have started and not yet saved. + pub(crate) fn spawning_targets(&self) -> HashSet { + intents::load(&self.config.app_data_dir) + .into_iter() + .filter(|intent| !intent.ended && !intent.launch_failed) + .filter_map(|intent| intent.target) + .collect() + } + /// Saves a spawned agent's receipt before it is taken off the relay /// here: its registration may be its invocation's only record (one kept /// after a restart), and a late replay must be answered, not relaunched. @@ -243,6 +289,9 @@ mod tests { pause: Mutex, std::sync::mpsc::Receiver<()>)>>, /// Its agents take native user turns. native: bool, + /// A paused launch starts its session (`THREAD`) before it pauses, + /// as Codex starts a thread before its first turn is confirmed. + starts_before_pause: bool, } /// Answers each agent by name; `Running`/`Idle` for one not set. @@ -299,7 +348,11 @@ mod tests { started_session(&Target::new(SessionKind::Codex, THREAD), None); return Err(LaunchError::Failed("Codex refused the first turn".into())); } + let started_thread = self.starts_before_pause; if let Some((started, proceed)) = self.pause.lock().unwrap().take() { + if started_thread { + started_session(&Target::new(SessionKind::Codex, THREAD), None); + } started.send(()).unwrap(); proceed.recv().unwrap(); } @@ -308,7 +361,11 @@ mod tests { return Err(LaunchError::Failed("Codex refused the first turn".into())); } // A thread of its own for each launch. - let thread = format!("{}{:012x}", &THREAD[..24], launched.len()); + let thread = if started_thread { + THREAD.to_string() + } else { + format!("{}{:012x}", &THREAD[..24], launched.len()) + }; Ok(Launched { target: Target::new(SessionKind::Codex, thread), process: None, @@ -350,6 +407,7 @@ mod tests { mod requester_approvals; mod restart_reasons; mod session_end; + mod spawned_binding; mod unanswered_register; #[tokio::test] diff --git a/crates/relay-node/src/relay/spawn/invoke.rs b/crates/relay-node/src/relay/spawn/invoke.rs index 1ef4ca9d3..8283157ee 100644 --- a/crates/relay-node/src/relay/spawn/invoke.rs +++ b/crates/relay-node/src/relay/spawn/invoke.rs @@ -466,12 +466,29 @@ impl Core { ); // The requester gets the first turn's final answer unless the agent // sends them something itself (`report_back::owed`). + // An identity the session was given of its own before this save + // (one registered in the moment before its spawn recorded it) is + // replaced here, and taken off Relaycast below: the session is the + // spawned agent alone. + let mut displaced = None; let saved = self.store.update(|store| { - report_back::add_spawned( + let earlier = store + .registration_for(®istration.target) + .filter(|earlier| { + earlier.spawn.is_none() + && earlier.agent_id != registration.agent_id + && earlier.name != registration.name + }) + .cloned(); + let added = report_back::add_spawned( store, registration.clone(), Secret::new(agent.token.clone()), - ) + ); + if added { + displaced = earlier; + } + added }); if saved != Ok(true) { if self.abandon(&launched, &mut in_flight).await { @@ -486,6 +503,12 @@ impl Core { } .into()); } + if let Some(earlier) = displaced { + if let Some(key) = earlier.workspace_key.as_ref() { + self.delete_or_retry(&earlier.base_url, key, &earlier.name) + .await; + } + } self.watch_spawned(®istration); in_flight.settled(); self.hub.update(|status| { diff --git a/crates/relay-node/src/relay/spawn/tests/report_back.rs b/crates/relay-node/src/relay/spawn/tests/report_back.rs index 6b7b770cd..91253c2f4 100644 --- a/crates/relay-node/src/relay/spawn/tests/report_back.rs +++ b/crates/relay-node/src/relay/spawn/tests/report_back.rs @@ -19,7 +19,7 @@ pub(super) fn relaycast_rest(invocation: Value) -> (String, Arc, ) -> (String, Arc>>) { use std::io::{Read, Write}; diff --git a/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs b/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs new file mode 100644 index 000000000..d8bc6e0b3 --- /dev/null +++ b/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs @@ -0,0 +1,272 @@ +//! A spawned session's own connection (its SessionStart hook, an activation +//! pass, its own registration request) binds to the registration its spawn +//! saves: one registration, one roster identity, its requester kept. Any +//! other session still gets a registration of its own. + +use super::report_back::relaycast_routes; +use super::*; +use crate::relay::sessions::DiscoveredSession; +use crate::relay::store::SpawnRequester; + +fn session(thread: &str) -> DiscoveredSession { + DiscoveredSession { + target: Target::new(SessionKind::Codex, thread), + // The spawn header is the thread's title, as Codex names it. + title: "[via Agent Relay — spawned as @w1 (w1@test-desk-desktop)] Fix CI".into(), + directory: None, + updated_at_ms: None, + live: true, + host: None, + holder: None, + } +} + +fn with_account(home: &std::path::Path, spawner: Arc, rest: &str) -> Arc { + core_built( + home, + spawner, + Arc::new(crate::relay::cloud_session::SignedOut), + rest, + Some(crate::relay::test_account::TestAccount::new().into_plugin()), + ) +} + +/// The worker's own registration request, started now. +fn worker_registers( + core: &Arc, + thread: &'static str, +) -> tokio::task::JoinHandle< + Result, +> { + let core = core.clone(); + tokio::spawn(async move { + core.register_session(&session(thread), None, None, true) + .await + }) +} + +#[tokio::test] +async fn a_spawned_workers_own_connection_binds_to_its_spawn_registration() { + let home = tempfile::tempdir().unwrap(); + let spawner = Arc::new(FakeSpawner { + starts_before_pause: true, + ..FakeSpawner::default() + }); + let (started_tx, started) = std::sync::mpsc::channel(); + let (proceed, proceed_rx) = std::sync::mpsc::channel(); + *spawner.pause.lock().unwrap() = Some((started_tx, proceed_rx)); + let core = core(home.path(), spawner.clone()); + let (link, sent) = NodeLink::detached(node()); + let frames = relaycast(link.clone(), sent); + let spawning = tokio::spawn(core.clone().answer_spawn(link, request("w1"))); + tokio::task::spawn_blocking(move || started.recv().unwrap()) + .await + .unwrap(); + + // Started, not yet saved: discovery already counts it as spawned. + let target = Target::new(SessionKind::Codex, THREAD); + assert!(core.store.read(|store| store.registrations.is_empty())); + assert!(core.spawned_targets().contains(&target)); + + // The worker's own connection asks to be registered in that window, + // and waits for the spawn instead of registering an agent of its own. + let worker = worker_registers(&core, THREAD); + tokio::time::sleep(Duration::from_millis(300)).await; + assert!( + !worker.is_finished(), + "it waits for the spawn's registration" + ); + proceed.send(()).unwrap(); + spawning.await.unwrap(); + let registered = worker.await.unwrap().unwrap(); + assert_eq!(registered.name, "w1"); + assert!(registered.already_registered); + assert_eq!(registered.address, "w1@test-desk-desktop"); + + // One registration, the spawn's, with its record intact. + let registrations = core.store.read(|store| store.registrations.clone()); + assert_eq!(registrations.len(), 1, "{registrations:?}"); + let registration = ®istrations[0]; + assert_eq!(registration.target, target); + assert_eq!(registration.agent_id, "agent_w1"); + let spawn = registration.spawn.as_ref().unwrap(); + assert_eq!(spawn.invocation_id, "inv_w1"); + assert_eq!(spawn.node_name, "test-desk-desktop"); + assert!(spawn.taken_over_at.is_none()); + + // One roster identity: the node's agent.register, and no other. + let frames = frames.await.unwrap(); + assert_eq!( + frames + .iter() + .filter(|frame| frame["type"] == "agent.register") + .count(), + 1, + "{frames:?}" + ); + + // Asked again once saved, the answer is the same identity. + let again = worker_registers(&core, THREAD).await.unwrap().unwrap(); + assert_eq!( + (again.name.as_str(), again.already_registered), + ("w1", true) + ); + assert_eq!(core.store.read(|store| store.registrations.len()), 1); + assert!(core.spawned_targets().contains(&target)); +} + +#[tokio::test] +async fn the_requester_still_approves_after_the_worker_connected() { + let home = tempfile::tempdir().unwrap(); + let core = with_account( + home.path(), + Arc::new(FakeSpawner::default()), + "http://localhost:9", + ); + core.store + .update(|store| store.extras.spawn_requester_approvals = Some(true)) + .unwrap(); + let target = Target::new(SessionKind::Codex, THREAD); + let requester = SpawnRequester { + name: "grokbot".into(), + agent_id: "agent_grokbot".into(), + }; + // A spawn whose launch started the thread and has not saved it yet. + let mut intent = interrupted("w1", true, Some(target.clone())); + intent.requester = Some(requester.clone()); + intents::put(&core.config.app_data_dir, &intent).unwrap(); + assert!(core.spawned_targets().contains(&target)); + + let worker = worker_registers(&core, THREAD); + tokio::time::sleep(Duration::from_millis(200)).await; + // The spawn saves its registration, as `answer_spawn` does. + let saved = intent.registration( + "agent_w1", + target.clone(), + None, + (&node().node_id, &node().node_name), + ); + assert!(core + .store + .update(|store| { + crate::relay::spawn::report_back::add_spawned( + store, + saved, + crate::relay::workspace::Secret::new("at_live_w1"), + ) + }) + .unwrap()); + let registered = worker.await.unwrap().unwrap(); + assert_eq!( + (registered.name.as_str(), registered.already_registered), + ("w1", true) + ); + + let registration = core + .store + .read(|store| store.registration_for(&target).cloned()) + .unwrap(); + assert_eq!(core.store.read(|store| store.registrations.len()), 1); + let spawn = registration.spawn.as_ref().unwrap(); + assert_eq!(spawn.requester.as_ref(), Some(&requester)); + assert!(spawn.taken_over_at.is_none()); + assert_eq!( + core.approving_requester(®istration), + Some(requester), + "the requester is still found after the worker connected" + ); +} + +#[tokio::test] +async fn a_spawned_session_is_refused_an_identity_of_its_own_while_its_spawn_saves() { + let home = tempfile::tempdir().unwrap(); + let core = core(home.path(), Arc::new(FakeSpawner::default())); + let target = Target::new(SessionKind::Codex, THREAD); + // The launch started it; the spawn never saved (it is being abandoned). + let intent = interrupted("w1", true, Some(target.clone())); + intents::put(&core.config.app_data_dir, &intent).unwrap(); + let refused = worker_registers(&core, THREAD).await.unwrap().unwrap_err(); + assert_eq!( + refused.data.kind, + crate::relay::protocol::ErrorKind::Refused, + "{refused:?}" + ); + assert!(core.store.read(|store| store.registrations.is_empty())); + // Once the spawn has ended, the session is no longer held for it. + intents::put( + &core.config.app_data_dir, + &intents::Intent { + ended: true, + ..intent + }, + ) + .unwrap(); + assert!(!core.spawned_targets().contains(&target)); +} + +#[tokio::test] +async fn a_session_no_spawn_started_is_not_held_for_one() { + const OTHER: &str = "7c9e6679-7425-40de-944b-e07fc1f90ae7"; + let home = tempfile::tempdir().unwrap(); + let core = core(home.path(), Arc::new(FakeSpawner::default())); + // Another thread is being spawned meanwhile. + let spawned = Target::new(SessionKind::Codex, THREAD); + intents::put( + &core.config.app_data_dir, + &interrupted("w1", true, Some(spawned.clone())), + ) + .unwrap(); + let other = Target::new(SessionKind::Codex, OTHER); + assert!(core.spawned_targets().contains(&spawned)); + assert!(!core.spawned_targets().contains(&other)); + // It goes straight to Relaycast for an agent of its own: nothing + // listens here, so the registration fails there, not as a spawn's. + let started = std::time::Instant::now(); + let error = worker_registers(&core, OTHER).await.unwrap().unwrap_err(); + assert_ne!( + error.data.kind, + crate::relay::protocol::ErrorKind::Refused, + "{error:?}" + ); + assert!( + started.elapsed() < Duration::from_secs(4), + "never waits on a spawn" + ); +} + +/// The positive half, against Relaycast's REST API: a session no spawn +/// started is registered as an agent of its own, with no spawn record. +#[tokio::test] +async fn a_session_no_spawn_started_gets_a_registration_of_its_own() { + const OTHER: &str = "7c9e6679-7425-40de-944b-e07fc1f90ae7"; + let home = tempfile::tempdir().unwrap(); + let (rest, seen) = relaycast_routes(vec![( + "POST /v1/agents ", + "200 OK", + json!({"ok": true, "data": { + "id": "agent_solo", "name": "solo", "token": "at_live_solo" + }}), + )]); + let core = core_at(home.path(), Arc::new(FakeSpawner::default()), &rest); + let registered = core + .register_session(&session(OTHER), Some("solo"), None, true) + .await + .unwrap(); + assert_eq!(registered.name, "solo"); + assert!(!registered.already_registered); + assert_ne!(registered.address, "solo@test-desk-desktop"); + assert!( + seen.lock() + .unwrap() + .iter() + .any(|line| line.starts_with("POST /v1/agents ")), + "{seen:?}" + ); + let other = Target::new(SessionKind::Codex, OTHER); + let registration = core + .store + .read(|store| store.registration_for(&other).cloned()) + .unwrap(); + assert_eq!(registration.agent_id, "agent_solo"); + assert!(registration.spawn.is_none()); +} From 9383f18feb8562ed177addcef69ff38b72b4e2c7 Mon Sep 17 00:00:00 2001 From: layered-relay-desktop-agent-host Date: Tue, 10 Nov 2026 10:35:43 +0000 Subject: [PATCH 2/3] spawn binding: fix review threads (displaced agent cleanup survives restart, in-flight read order) - The spawn save records the displaced identity's Relaycast deletion as a PendingCleanup in the same store update that replaces its registration and drops its token. A stop before the deletion leaves it to retry_pending_cleanups after a restart; a deletion that succeeds clears the record (Core::delete_recorded_cleanup). - saved_spawn_registration reads the in-flight state before the saved registration, so a spawn that saves and settles between the two reads is found saved rather than answered with None. Co-Authored-By: Claude Fable 5.1 --- crates/relay-node/src/relay/agents.rs | 21 ++++ crates/relay-node/src/relay/spawn.rs | 7 +- crates/relay-node/src/relay/spawn/invoke.rs | 23 ++-- .../src/relay/spawn/tests/spawned_binding.rs | 117 ++++++++++++++++++ 4 files changed, 159 insertions(+), 9 deletions(-) diff --git a/crates/relay-node/src/relay/agents.rs b/crates/relay-node/src/relay/agents.rs index adc19fbb8..e874b016a 100644 --- a/crates/relay-node/src/relay/agents.rs +++ b/crates/relay-node/src/relay/agents.rs @@ -804,6 +804,27 @@ impl Core { } } + /// Deletes an agent whose removal `cleanup` already records (saved in + /// the same store update that let go of its token), and drops the record + /// once it is gone. Otherwise the record stays for + /// `retry_pending_cleanups`, across a restart too. + pub(crate) async fn delete_recorded_cleanup(&self, cleanup: PendingCleanup) { + let workspace = self.relaycast_scope(&cleanup.base_url, &cleanup.workspace_key); + let result = + client::delete_agent(&cleanup.base_url, &cleanup.workspace_key, &cleanup.name).await; + self.observe_relaycast_write_result(&workspace, &result); + if result.is_ok() { + let _ = self.store.update(|store| { + if let Some(list) = store.pending_cleanups.as_mut() { + list.retain(|c| *c != cleanup); + if list.is_empty() { + store.pending_cleanups = None; + } + } + }); + } + } + /// Deletes a channel this computer created but recorded nowhere, or /// records it to retry. async fn delete_channel_or_retry(&self, base_url: &str, key: &Secret, name: &str) { diff --git a/crates/relay-node/src/relay/spawn.rs b/crates/relay-node/src/relay/spawn.rs index ecb8bd4d4..c44cd3a24 100644 --- a/crates/relay-node/src/relay/spawn.rs +++ b/crates/relay-node/src/relay/spawn.rs @@ -152,11 +152,14 @@ impl Core { pub(crate) async fn saved_spawn_registration(&self, target: &Target) -> Option { let deadline = tokio::time::Instant::now() + SPAWN_SAVE_WAIT; loop { + // In flight first, the registration second: a spawn that saves + // and settles between the two reads is then found saved, never + // taken for one that is neither saved nor in flight. + let spawning = self.spawning_targets().contains(target); if let Some(registration) = self.saved_spawn_registration_now(target) { return Some(registration); } - if tokio::time::Instant::now() >= deadline || !self.spawning_targets().contains(target) - { + if !spawning || tokio::time::Instant::now() >= deadline { return None; } tokio::time::sleep(SPAWN_SAVE_POLL).await; diff --git a/crates/relay-node/src/relay/spawn/invoke.rs b/crates/relay-node/src/relay/spawn/invoke.rs index 8283157ee..833a9e49b 100644 --- a/crates/relay-node/src/relay/spawn/invoke.rs +++ b/crates/relay-node/src/relay/spawn/invoke.rs @@ -14,6 +14,7 @@ use super::{first_text, lifecycle, session_ref, SpawnSettings}; use crate::relay::daemon::Core; use crate::relay::push::PushState; use crate::relay::state::{Level, Listing, SpawnState}; +use crate::relay::store::{CleanupKind, PendingCleanup}; use crate::relay::workspace::Secret; use chrono::Utc; use launching::LaunchFailure; @@ -469,7 +470,9 @@ impl Core { // An identity the session was given of its own before this save // (one registered in the moment before its spawn recorded it) is // replaced here, and taken off Relaycast below: the session is the - // spawned agent alone. + // spawned agent alone. Its removal is recorded in the same update + // that lets go of its token, so a stop before the deletion below + // leaves it to the cleanup retries rather than unreachable. let mut displaced = None; let saved = self.store.update(|store| { let earlier = store @@ -486,7 +489,16 @@ impl Core { Secret::new(agent.token.clone()), ); if added { - displaced = earlier; + displaced = earlier.and_then(|earlier| { + let cleanup = PendingCleanup { + kind: CleanupKind::Agent, + name: earlier.name, + base_url: earlier.base_url, + workspace_key: earlier.workspace_key?, + }; + store.add_pending_cleanup(cleanup.clone()); + Some(cleanup) + }); } added }); @@ -503,11 +515,8 @@ impl Core { } .into()); } - if let Some(earlier) = displaced { - if let Some(key) = earlier.workspace_key.as_ref() { - self.delete_or_retry(&earlier.base_url, key, &earlier.name) - .await; - } + if let Some(cleanup) = displaced { + self.delete_recorded_cleanup(cleanup).await; } self.watch_spawned(®istration); in_flight.settled(); diff --git a/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs b/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs index d8bc6e0b3..8ad1dabe6 100644 --- a/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs +++ b/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs @@ -270,3 +270,120 @@ async fn a_session_no_spawn_started_gets_a_registration_of_its_own() { assert_eq!(registration.agent_id, "agent_solo"); assert!(registration.spawn.is_none()); } + +/// An identity the session was given of its own before its spawn saved. +fn earlier_identity(base_url: &str) -> Registration { + serde_json::from_value(json!({ + "name": "solo", "agentID": "agent_solo", + "target": {"kind": "codex", "id": THREAD}, + "title": "solo", "baseURL": base_url, "registeredAt": "2026-11-10T00:00:00Z", + "workspaceKey": "rk_live_test0000" + })) + .unwrap() +} + +/// Spawns `w1` onto `THREAD`, which already holds `earlier`; the spawn +/// runs on until it has also taken `earlier` off Relaycast. +fn spawn_over_earlier( + core: &Arc, + spawner: &FakeSpawner, + earlier: Registration, +) -> tokio::task::JoinHandle { + assert!(core + .store + .update(|store| store.add( + earlier, + crate::relay::workspace::Secret::new("at_live_solo") + )) + .unwrap()); + let (started_tx, started) = std::sync::mpsc::channel(); + let (proceed, proceed_rx) = std::sync::mpsc::channel(); + proceed.send(()).unwrap(); + *spawner.pause.lock().unwrap() = Some((started_tx, proceed_rx)); + let (link, sent) = NodeLink::detached(node()); + let frames = relaycast(link.clone(), sent); + let core = core.clone(); + tokio::spawn(async move { + let result = core.answer_spawn(link, request("w1")).await; + drop((started, frames)); + result + }) +} + +fn saved_w1(core: &Core) -> bool { + core.store.read(|store| { + store + .registration_for(&Target::new(SessionKind::Codex, THREAD)) + .is_some_and(|registration| registration.name == "w1" && registration.spawn.is_some()) + }) +} + +fn displaced_cleanups(core: &Core) -> Vec { + core.store.read(|store| { + store + .pending_cleanups + .clone() + .unwrap_or_default() + .into_iter() + .filter(|cleanup| cleanup.name == "solo") + .collect() + }) +} + +#[tokio::test] +async fn a_displaced_identity_is_owed_from_the_save_on_across_a_restart() { + let home = tempfile::tempdir().unwrap(); + let spawner = Arc::new(FakeSpawner { + starts_before_pause: true, + ..FakeSpawner::default() + }); + let core = core(home.path(), spawner.clone()); + // Its origin takes the connection and never answers: the deletion after + // the save is still running when the core stops. + let silent = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let origin = format!("http://{}", silent.local_addr().unwrap()); + let spawning = spawn_over_earlier(&core, &spawner, earlier_identity(&origin)); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while !saved_w1(&core) { + assert!(std::time::Instant::now() < deadline, "the spawn saves"); + tokio::time::sleep(Duration::from_millis(20)).await; + } + // Recorded in the update that replaced it, before any deletion. + let owed = displaced_cleanups(&core); + assert_eq!(owed.len(), 1, "{owed:?}"); + assert_eq!(owed[0].kind, crate::relay::store::CleanupKind::Agent); + assert_eq!(owed[0].base_url, origin); + assert!(core.store.read(|store| !store.tokens.contains_key("solo"))); + assert!(!spawning.is_finished(), "the deletion has not finished"); + + // The core stops there. Started again, it still owes the deletion, for + // its cleanup retries. + spawning.abort(); + let _ = spawning.await; + drop(core); + let restarted = super::core(home.path(), Arc::new(FakeSpawner::default())); + assert_eq!(displaced_cleanups(&restarted), owed); +} + +#[tokio::test] +async fn a_displaced_identity_deleted_after_the_save_is_owed_no_more() { + let home = tempfile::tempdir().unwrap(); + let (rest, seen) = relaycast_routes(vec![]); + let spawner = Arc::new(FakeSpawner { + starts_before_pause: true, + ..FakeSpawner::default() + }); + let core = core(home.path(), spawner.clone()); + spawn_over_earlier(&core, &spawner, earlier_identity(&rest)) + .await + .unwrap(); + assert!(saved_w1(&core)); + assert!( + seen.lock() + .unwrap() + .iter() + .any(|line| line.starts_with("DELETE /v1/agents/solo ")), + "{seen:?}" + ); + assert!(displaced_cleanups(&core).is_empty()); +} From 273cde6ab04037eaba138c3067d3c2122430d1ce Mon Sep 17 00:00:00 2001 From: layered-relay-desktop-agent-host Date: Tue, 10 Nov 2026 11:05:11 +0000 Subject: [PATCH 3/3] Spawn binding: delete the displaced agent under its name claim The deletion right after the spawn's save goes through the same helper as the retry pass: the name is claimed first, a registration answering to it again is kept without a DELETE, and the record is dropped only once the removal succeeded. Writes paused or the name held leave the record for retry_pending_cleanups. Co-Authored-By: Claude Fable 5.1 --- crates/relay-node/src/relay/agents.rs | 77 +++++++++++-------- .../src/relay/spawn/tests/spawned_binding.rs | 60 +++++++++++++++ 2 files changed, 105 insertions(+), 32 deletions(-) diff --git a/crates/relay-node/src/relay/agents.rs b/crates/relay-node/src/relay/agents.rs index e874b016a..c255f9979 100644 --- a/crates/relay-node/src/relay/agents.rs +++ b/crates/relay-node/src/relay/agents.rs @@ -806,12 +806,18 @@ impl Core { /// Deletes an agent whose removal `cleanup` already records (saved in /// the same store update that let go of its token), and drops the record - /// once it is gone. Otherwise the record stays for - /// `retry_pending_cleanups`, across a restart too. + /// once it is gone. It goes through `delete_agent_cleanup`, as the retry + /// pass does: when Relaycast writes are paused, the name is held by + /// another removal or registration, or the deletion fails, the record + /// stays for `retry_pending_cleanups`, across a restart too. pub(crate) async fn delete_recorded_cleanup(&self, cleanup: PendingCleanup) { let workspace = self.relaycast_scope(&cleanup.base_url, &cleanup.workspace_key); - let result = - client::delete_agent(&cleanup.base_url, &cleanup.workspace_key, &cleanup.name).await; + if !self.relaycast_write_allowed(&workspace) { + return; + } + let Some(result) = self.delete_agent_cleanup(&cleanup).await else { + return; + }; self.observe_relaycast_write_result(&workspace, &result); if result.is_ok() { let _ = self.store.update(|store| { @@ -825,6 +831,37 @@ impl Core { } } + /// Carries out a recorded agent removal under its name's claim: `None` + /// when the name is held (a registration or another removal in flight; + /// tried again next pass). An agent answering to the name here again + /// (a rename back, a new registration) is kept, and the removal counts + /// as done without a DELETE. + async fn delete_agent_cleanup( + &self, + cleanup: &PendingCleanup, + ) -> Option> { + // The name is held for the removal, as a registration holds it + // until saved: one in flight is waited out, and none starts + // meanwhile. + if !self.claim_cleanup_name(&cleanup.name) { + return None; + } + // Re-read before the removal: the name may have been taken again + // since the cleanup was recorded, and that agent is kept. + let kept = self.store.read(|s| { + s.registrations + .iter() + .any(|r| r.answers_to(&cleanup.name) && r.base_url == cleanup.base_url) + }); + let result = if kept { + Ok(()) + } else { + client::delete_agent(&cleanup.base_url, &cleanup.workspace_key, &cleanup.name).await + }; + self.release_name(&cleanup.name); + Some(result) + } + /// Deletes a channel this computer created but recorded nowhere, or /// records it to retry. async fn delete_channel_or_retry(&self, base_url: &str, key: &Secret, name: &str) { @@ -861,34 +898,10 @@ impl Core { client::delete_channel(&cleanup.base_url, &cleanup.workspace_key, &cleanup.name) .await } - CleanupKind::Agent => { - // The name is held for the removal, as a registration - // holds it until saved: one in flight is waited out (and - // tried next pass), and none starts meanwhile. - if !self.claim_cleanup_name(&cleanup.name) { - continue; - } - // Re-read before the removal: the name may have been - // taken again here (a rename back, a new registration) - // since the cleanup was recorded, and that agent is kept. - let kept = self.store.read(|s| { - s.registrations - .iter() - .any(|r| r.answers_to(&cleanup.name) && r.base_url == cleanup.base_url) - }); - let result = if kept { - Ok(()) - } else { - client::delete_agent( - &cleanup.base_url, - &cleanup.workspace_key, - &cleanup.name, - ) - .await - }; - self.release_name(&cleanup.name); - result - } + CleanupKind::Agent => match self.delete_agent_cleanup(&cleanup).await { + Some(result) => result, + None => continue, + }, }; self.observe_relaycast_write_result(&workspace, &result); if result.is_ok() { diff --git a/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs b/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs index 8ad1dabe6..b34f97846 100644 --- a/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs +++ b/crates/relay-node/src/relay/spawn/tests/spawned_binding.rs @@ -387,3 +387,63 @@ async fn a_displaced_identity_deleted_after_the_save_is_owed_no_more() { ); assert!(displaced_cleanups(&core).is_empty()); } + +/// The removal recorded for `solo` at `base_url`, as the spawn saves it. +fn solo_cleanup(core: &Core, base_url: &str) -> crate::relay::store::PendingCleanup { + let cleanup = crate::relay::store::PendingCleanup { + kind: crate::relay::store::CleanupKind::Agent, + name: "solo".into(), + base_url: base_url.into(), + workspace_key: crate::relay::workspace::Secret::new("rk_live_test0000"), + }; + core.store + .update(|store| store.add_pending_cleanup(cleanup.clone())) + .unwrap(); + cleanup +} + +fn deletes_of_solo(seen: &Mutex>) -> usize { + seen.lock() + .unwrap() + .iter() + .filter(|line| line.starts_with("DELETE /v1/agents/solo ")) + .count() +} + +#[tokio::test] +async fn a_displaced_name_registered_again_before_its_deletion_is_kept() { + let home = tempfile::tempdir().unwrap(); + let (rest, seen) = relaycast_routes(vec![]); + let core = core(home.path(), Arc::new(FakeSpawner::default())); + let cleanup = solo_cleanup(&core, &rest); + // `solo` answers here again (a new registration) before the deletion + // after the save runs: that agent is kept, and nothing more is owed. + let mut again = earlier_identity(&rest); + again.target = Target::new(SessionKind::Codex, "7c9e6679-7425-40de-944b-e07fc1f90ae7"); + assert!(core + .store + .update(|store| store.add(again, crate::relay::workspace::Secret::new("at_live_again"))) + .unwrap()); + + core.delete_recorded_cleanup(cleanup).await; + + assert_eq!(deletes_of_solo(&seen), 0, "{seen:?}"); + assert!(displaced_cleanups(&core).is_empty()); +} + +#[tokio::test] +async fn a_displaced_name_held_by_another_claim_is_left_to_the_retries() { + let home = tempfile::tempdir().unwrap(); + let (rest, seen) = relaycast_routes(vec![]); + let core = core(home.path(), Arc::new(FakeSpawner::default())); + let cleanup = solo_cleanup(&core, &rest); + // A registration (or the retry pass) holds the name: the deletion after + // the save does not race it, and the record stays for the retries. + core.claim_name("solo").unwrap(); + + core.delete_recorded_cleanup(cleanup.clone()).await; + + assert_eq!(deletes_of_solo(&seen), 0, "{seen:?}"); + assert_eq!(displaced_cleanups(&core), vec![cleanup]); + core.release_name("solo"); +}