Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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@<node>` address and the requester who approves it.

## [18.0.0] - 2026-11-09

Expand Down
134 changes: 104 additions & 30 deletions crates/relay-node/src/relay/agents.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.")
}
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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(&registration.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()) {
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -764,6 +804,64 @@ 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. 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);
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| {
if let Some(list) = store.pending_cleanups.as_mut() {
list.retain(|c| *c != cleanup);
if list.is_empty() {
store.pending_cleanups = None;
}
}
});
}
}

/// 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<relaycast::Result<()>> {
// 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) {
Expand Down Expand Up @@ -800,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() {
Expand Down
10 changes: 9 additions & 1 deletion crates/relay-node/src/relay/outbound/hooks/codex_stop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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!({}));
}
Expand Down
67 changes: 64 additions & 3 deletions crates/relay-node/src/relay/spawn.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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: `<machine prefix>-desktop`.
pub fn node_name(machine_prefix: &str) -> String {
Expand Down Expand Up @@ -114,18 +119,62 @@ 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<Target> {
self.store.read(|store| {
let mut targets: HashSet<Target> = 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<Registration> {
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<Registration> {
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 !spawning || tokio::time::Instant::now() >= deadline {
return None;
}
Comment thread
agent-relay-code[bot] marked this conversation as resolved.
Resolved
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<Target> {
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.
Expand Down Expand Up @@ -243,6 +292,9 @@ mod tests {
pause: Mutex<Option<(std::sync::mpsc::Sender<()>, 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.
Expand Down Expand Up @@ -299,7 +351,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();
}
Expand All @@ -308,7 +364,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,
Expand Down Expand Up @@ -350,6 +410,7 @@ mod tests {
mod requester_approvals;
mod restart_reasons;
mod session_end;
mod spawned_binding;
mod unanswered_register;

#[tokio::test]
Expand Down
36 changes: 34 additions & 2 deletions crates/relay-node/src/relay/spawn/invoke.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -466,12 +467,40 @@ 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. 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| {
report_back::add_spawned(
let earlier = store
.registration_for(&registration.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.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
});
if saved != Ok(true) {
if self.abandon(&launched, &mut in_flight).await {
Expand All @@ -486,6 +515,9 @@ impl Core {
}
.into());
}
if let Some(cleanup) = displaced {
self.delete_recorded_cleanup(cleanup).await;
}
self.watch_spawned(&registration);
in_flight.settled();
self.hub.update(|status| {
Expand Down
Loading
Loading