From 7c704f20a1d9c5b1b7219b5100147ea39c247ca1 Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Wed, 11 Nov 2026 07:49:17 +1100 Subject: [PATCH 1/8] fix(relay-node): register loaded Codex chats without a window owner --- CHANGELOG.md | 2 + crates/relay-node/contract/CoreContract.swift | 9 +- crates/relay-node/src/relay/agents.rs | 87 ++++++- .../src/relay/agents/activation_journal.rs | 18 +- crates/relay-node/src/relay/codex/threads.rs | 31 +++ .../relay-node/src/relay/contract/agents.rs | 4 + crates/relay-node/src/relay/daemon.rs | 3 + crates/relay-node/src/relay/local.rs | 1 + .../src/relay/local/directory_trait.rs | 87 +++++++ .../src/relay/local/session_directory.rs | 12 +- .../relay/local/tests/runtime_registration.rs | 235 ++++++++++++++++++ crates/relay-node/src/relay/local_methods.rs | 7 + .../src/relay/outbound/hooks/codex_stop.rs | 103 ++++++-- .../outbound/hooks/codex_stop/affinity.rs | 23 +- crates/relay-node/src/relay/process.rs | 2 + .../src/relay/process/codex_runtime.rs | 160 ++++++++++++ crates/relay-node/src/relay/sessions.rs | 43 ++++ 17 files changed, 780 insertions(+), 47 deletions(-) create mode 100644 crates/relay-node/src/relay/local/tests/runtime_registration.rs create mode 100644 crates/relay-node/src/relay/process/codex_runtime.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index c20447ea2..7940ce159 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Loaded Codex user chats automatically join and stay registered without a desktop window owner; registration presence no longer implies an idle message-delivery route. + - 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. diff --git a/crates/relay-node/contract/CoreContract.swift b/crates/relay-node/contract/CoreContract.swift index 2d7c9a190..9f5d16223 100644 --- a/crates/relay-node/contract/CoreContract.swift +++ b/crates/relay-node/contract/CoreContract.swift @@ -2312,6 +2312,9 @@ public struct LocalAgent: Codable, Equatable, Hashable, Sendable { /// A session on this computer, with its registration when it has one. public struct LocalSession: Codable, Equatable, Hashable, Sendable { + /// Registration/transport presence is not proof that the model can + /// receive a message. Absent on older cores and when no wait is known. + public let deliveryWait: String? public let directory: String? /// What holds this session's transcript, for a session that can be /// held at all (see [`SessionHolder`]). @@ -2338,7 +2341,8 @@ public struct LocalSession: Codable, Equatable, Hashable, Sendable { public let title: String public let updatedAtMs: Int64? - public init(directory: String? = nil, holder: SessionHolder? = nil, host: String? = nil, live: Bool, project: String? = nil, registration: SessionRegistration? = nil, running: Bool? = nil, suggestedName: String, target: Target, title: String, updatedAtMs: Int64? = nil) { + public init(deliveryWait: String? = nil, directory: String? = nil, holder: SessionHolder? = nil, host: String? = nil, live: Bool, project: String? = nil, registration: SessionRegistration? = nil, running: Bool? = nil, suggestedName: String, target: Target, title: String, updatedAtMs: Int64? = nil) { + self.deliveryWait = deliveryWait self.directory = directory self.holder = holder self.host = host @@ -2353,6 +2357,7 @@ public struct LocalSession: Codable, Equatable, Hashable, Sendable { } enum CodingKeys: String, CodingKey { + case deliveryWait = "delivery_wait" case directory case holder case host @@ -2368,6 +2373,7 @@ public struct LocalSession: Codable, Equatable, Hashable, Sendable { public init(from decoder: Decoder) throws { let c = try decoder.container(keyedBy: CodingKeys.self) + self.deliveryWait = try c.decodeIfPresent(String.self, forKey: .deliveryWait) self.directory = try c.decodeIfPresent(String.self, forKey: .directory) self.holder = try c.decodeIfPresent(SessionHolder.self, forKey: .holder) self.host = try c.decodeIfPresent(String.self, forKey: .host) @@ -2383,6 +2389,7 @@ public struct LocalSession: Codable, Equatable, Hashable, Sendable { public func encode(to encoder: Encoder) throws { var c = encoder.container(keyedBy: CodingKeys.self) + try c.encodeIfPresent(self.deliveryWait, forKey: .deliveryWait) try c.encode(self.directory, forKey: .directory) try c.encodeIfPresent(self.holder, forKey: .holder) try c.encodeIfPresent(self.host, forKey: .host) diff --git a/crates/relay-node/src/relay/agents.rs b/crates/relay-node/src/relay/agents.rs index c255f9979..074fed32f 100644 --- a/crates/relay-node/src/relay/agents.rs +++ b/crates/relay-node/src/relay/agents.rs @@ -19,7 +19,7 @@ use crate::relay::managed_sessions::ManagedAuthority; use crate::relay::people::identity_hash; use crate::relay::protocol::{ErrorKind, RpcError}; use crate::relay::push::{PushState, PushTarget}; -use crate::relay::sessions::DiscoveredSession; +use crate::relay::sessions::{AutomaticCandidate, DiscoveredSession}; use crate::relay::state::{Level, Listing}; use crate::relay::store::{ iso8601_now, random_hex, AgentAlias, CleanupKind, PendingCleanup, Registration, SessionKind, @@ -250,6 +250,25 @@ fn is_name_taken(error: &relaycast::RelayError) -> bool { } impl Core { + /// A trusted host can request a global pass, but cannot name the chat + /// to register. Coalesce signals and leave network work off hook calls. + pub(crate) fn request_codex_discovery(self: &Arc) { + use std::sync::atomic::Ordering; + if self.codex_discovery_refresh.swap(true, Ordering::AcqRel) { + return; + } + let core = self.clone(); + tokio::spawn(async move { + for attempt in 0..3 { + if attempt > 0 { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + core.activate_new_sessions().await; + } + core.codex_discovery_refresh.store(false, Ordering::Release); + }); + } + /// The agents or the sign-in changed: the spawn node is claimed for them, /// the people in the workspace are trusted like the person, and the /// account hears of it. @@ -429,6 +448,26 @@ impl Core { description: Option<&str>, added_by_session: bool, auto_activated: bool, + ) -> Result { + self.register_session_authorized( + session, + raw_name, + description, + added_by_session, + auto_activated, + None, + ) + .await + } + + async fn register_session_authorized( + &self, + session: &DiscoveredSession, + raw_name: Option<&str>, + description: Option<&str>, + added_by_session: bool, + auto_activated: bool, + candidate: Option<&AutomaticCandidate>, ) -> Result { let target = &session.target; let lock = self.target_locks.get(&target.key()); @@ -472,7 +511,16 @@ impl Core { if is_managed(target.kind) { self.verify_managed_registration_authority(target).await?; } - if !session.live { + let runtime_admission = if let Some(candidate) = candidate { + let directory = self.plugins.sessions.clone(); + let candidate = candidate.clone(); + tokio::task::spawn_blocking(move || directory.automatic_candidate_current(&candidate)) + .await + .unwrap_or(false) + } else { + false + }; + if !session.live && !runtime_admission { return Err(RpcError::new( ErrorKind::Refused, "This session is not running.", @@ -514,6 +562,27 @@ impl Core { "No Agent Relay workspace key on this computer.", ) })?; + let still_admitted = if let Some(candidate) = candidate { + let directory = self.plugins.sessions.clone(); + let candidate = candidate.clone(); + tokio::task::spawn_blocking(move || directory.automatic_candidate_current(&candidate)) + .await + .unwrap_or(false) + } else { + runtime_admission + }; + if candidate.is_some() + && (!still_admitted + || !auto_activation::activation_applies( + &self.store.read(auto_activation::ActivationSettings::of), + target.kind, + )) + { + return Err(RpcError::new( + ErrorKind::Refused, + "Automatic admission is disabled or its runtime proof expired.", + )); + } let mut candidate = name; let mut attempt = 1; let registration = loop { @@ -1431,11 +1500,7 @@ impl Core { let discovered = tokio::task::spawn_blocking(move || { sessions.pi_auto_join(pi_auto_join); let all = sessions.sessions(); - let live: HashSet = all - .iter() - .filter(|session| session.live) - .map(|session| session.target.clone()) - .collect(); + let live = sessions.present_targets(); let targets: Vec = registered.iter().map(|(_, t)| t.clone()).collect(); let archived = sessions.archived(&targets); let archived: Vec = registered @@ -1443,7 +1508,7 @@ impl Core { .filter(|(_, target)| archived.contains(target)) .map(|(name, _)| name) .collect(); - let found = sessions.auto_activation_sessions(); + let found = sessions.automatic_candidates(); (found, all, live, archived, sessions.cleared(&targets)) }) .await; @@ -1459,10 +1524,10 @@ impl Core { let spawned_live = self.spawned_live(); live.retain(|target| !spawned.contains(target)); live.extend(spawned_live); - let mut found: HashMap = found + let mut found: HashMap = found .into_iter() - .filter(|session| !spawned.contains(&session.target)) - .map(|session| (session.target.clone(), session)) + .filter(|candidate| !spawned.contains(&candidate.session.target)) + .map(|candidate| (candidate.session.target.clone(), candidate)) .collect(); let moved = self.follow_cleared(&cleared); let now = chrono::Utc::now(); diff --git a/crates/relay-node/src/relay/agents/activation_journal.rs b/crates/relay-node/src/relay/agents/activation_journal.rs index c5a7144af..4b7dc6985 100644 --- a/crates/relay-node/src/relay/agents/activation_journal.rs +++ b/crates/relay-node/src/relay/agents/activation_journal.rs @@ -56,7 +56,7 @@ impl Core { /// admission when the fleet was never reached. pub(super) async fn auto_activate_session( &self, - session: &DiscoveredSession, + candidate: &AutomaticCandidate, now_text: &str, ) -> ActivationAttempt { if self.workspace().is_some_and(|workspace| { @@ -79,7 +79,14 @@ impl Core { )); } match self - .register_session_with_origin(session, None, None, false, true) + .register_session_authorized( + &candidate.session, + None, + None, + false, + true, + Some(candidate), + ) .await { Ok(_) => ActivationAttempt::Attempted(Ok(())), @@ -111,7 +118,7 @@ impl Core { /// journal itself could not be written. pub(super) async fn activate_sessions( &self, - sessions: impl IntoIterator, + sessions: impl IntoIterator, mut remaining: usize, wait_reason: &str, now_text: &str, @@ -119,7 +126,8 @@ impl Core { ) -> (Vec, BTreeMap) { let mut cleared = Vec::new(); let mut blocked = BTreeMap::new(); - for session in sessions { + for candidate in sessions { + let session = &candidate.session; let key = session.target.key(); if self .store @@ -131,7 +139,7 @@ impl Core { blocked.insert(key, wait_reason.to_string()); continue; } - match self.auto_activate_session(&session, now_text).await { + match self.auto_activate_session(&candidate, now_text).await { ActivationAttempt::Paused(message) => { blocked.insert(key.clone(), message.clone()); failures.push(format!("{key}: {message}")); diff --git a/crates/relay-node/src/relay/codex/threads.rs b/crates/relay-node/src/relay/codex/threads.rs index 1a0621a80..3090adeea 100644 --- a/crates/relay-node/src/relay/codex/threads.rs +++ b/crates/relay-node/src/relay/codex/threads.rs @@ -332,6 +332,37 @@ pub fn read_one(config: &CodexConfig, id: &str) -> Option { }) } +/// A root user chat held by a trusted runtime, outside the recent-history +/// page. Internal/prewarmed and exec sessions never become automatic agents. +pub(crate) fn read_runtime_user(config: &CodexConfig, id: &str) -> Option { + if !crate::relay::store::is_uuid(id) { + return None; + } + with_connection(config, Duration::from_secs(2), |connection| { + let columns = columns(connection)?; + // Older schemas keep the existing deliverable-session discovery; + // do not widen it when user-session evidence is unavailable. + if !columns.contains("has_user_event") { + return None; + } + let root = if columns.contains("thread_source") { + " and +thread_source = 'user'" + } else { + "" + }; + let sql = format!( + "select {} from threads where {ID_MATCHES} and +archived = 0 \ + and +source in ('cli', 'vscode') and +has_user_event = 1{root} limit 1", + select_list(&columns)? + ); + let found = connection.query_row(&sql, [id], row).ok()?; + if is_relay_spawned_title(found.1.as_deref().unwrap_or_default()) { + return None; + } + Some(thread(found)) + }) +} + /// Whether this exact thread has a live catalog row, regardless of Codex's /// source label. Used before resuming a registered native spawn; it must /// not broaden the diff --git a/crates/relay-node/src/relay/contract/agents.rs b/crates/relay-node/src/relay/contract/agents.rs index 30b911510..3b64d9d3f 100644 --- a/crates/relay-node/src/relay/contract/agents.rs +++ b/crates/relay-node/src/relay/contract/agents.rs @@ -52,6 +52,10 @@ pub struct LocalSession { /// Whether the session can take a message now; a registered session /// discovery no longer finds is listed with `false`. pub live: bool, + /// Registration/transport presence is not proof that the model can + /// receive a message. Absent on older cores and when no wait is known. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub delivery_wait: Option, /// Whether discovery found the session's process on this computer: /// `false` for a registered session that has ended. A session that runs /// but is not `live` cannot take a message yet (a Grok TUI without its diff --git a/crates/relay-node/src/relay/daemon.rs b/crates/relay-node/src/relay/daemon.rs index 6f371c905..a9403db82 100644 --- a/crates/relay-node/src/relay/daemon.rs +++ b/crates/relay-node/src/relay/daemon.rs @@ -163,6 +163,8 @@ pub struct Core { pub(crate) session_snapshot: crate::relay::local_methods::PresentationSnapshot, /// Native title and mobile-state publications fence older roster snapshots. pub(crate) roster_revision: std::sync::atomic::AtomicU64, + /// Coalesces authenticated startup signals; never selects a chat. + pub(crate) codex_discovery_refresh: std::sync::atomic::AtomicBool, /// One change at a time per session (`target.key()`). pub(crate) target_locks: KeyedLocks, /// One delivery at a time per agent name; a removal waits for it. @@ -294,6 +296,7 @@ impl Core { project_names, session_snapshot: Default::default(), roster_revision: std::sync::atomic::AtomicU64::new(0), + codex_discovery_refresh: std::sync::atomic::AtomicBool::new(false), target_locks: KeyedLocks::default(), agent_locks: KeyedLocks::default(), roster_locks: KeyedLocks::default(), diff --git a/crates/relay-node/src/relay/local.rs b/crates/relay-node/src/relay/local.rs index 7dfdc8663..ac58efbc4 100644 --- a/crates/relay-node/src/relay/local.rs +++ b/crates/relay-node/src/relay/local.rs @@ -1878,6 +1878,7 @@ pub use plugins::{ #[cfg(unix)] mod tests { mod automatic_admission; + mod runtime_registration; mod spawned_delivery; use super::*; diff --git a/crates/relay-node/src/relay/local/directory_trait.rs b/crates/relay-node/src/relay/local/directory_trait.rs index 300015d8a..49dc109e3 100644 --- a/crates/relay-node/src/relay/local/directory_trait.rs +++ b/crates/relay-node/src/relay/local/directory_trait.rs @@ -160,6 +160,64 @@ impl SessionDirectory for LocalSessions { .collect() } + fn automatic_candidates(&self) -> Vec { + use crate::relay::sessions::AutomaticCandidate; + let mut found: HashMap = self + .auto_activation_sessions() + .into_iter() + .filter_map(AutomaticCandidate::live) + .map(|c| (c.session.target.clone(), c)) + .collect(); + for (session, proof) in self.codex_runtime_sessions() { + if self.runtime_admitted(&session) { + found + .entry(session.target.clone()) + .or_insert(AutomaticCandidate { + session, + runtime: Some(proof), + }); + } + } + found.into_values().collect() + } + + fn automatic_candidate_current( + &self, + candidate: &crate::relay::sessions::AutomaticCandidate, + ) -> bool { + let Some(proof) = candidate.runtime.as_ref() else { + return candidate.session.live; + }; + candidate.session.target.kind == SessionKind::Codex + && candidate.session.target.id == proof.thread() + && proof.current(&*self.table, self.config.uid) + && codex::threads::read_runtime_user(&self.config.codex, proof.thread()).is_some_and( + |thread| { + thread.directory == candidate.session.directory + && self.runtime_admitted(&candidate.session) + }, + ) + } + + fn present_targets(&self) -> HashSet { + let mut targets: HashSet<_> = self + .sessions() + .into_iter() + .filter(|s| s.live) + .map(|s| s.target) + .collect(); + targets.extend( + self.codex_runtime_sessions() + .into_iter() + .map(|(s, _)| s.target), + ); + targets + } + + fn trusted_codex_discovery_signal(&self, pid: i32) -> bool { + crate::relay::process::trusted_codex_host(&*self.table, pid, self.config.uid) + } + fn pi_auto_join(&self, auto_join: bool) { self.config.pi.sync_auto_join(auto_join); } @@ -412,6 +470,35 @@ fn fallback_title(kind: SessionKind) -> &'static str { } impl LocalSessions { + pub(super) fn codex_runtime_sessions( + &self, + ) -> Vec<(DiscoveredSession, crate::relay::process::CodexRuntimeProof)> { + crate::relay::process::codex_runtime_proofs(&*self.scans, self.config.uid) + .into_iter() + .filter_map(|proof| { + let thread = codex::threads::read_runtime_user(&self.config.codex, proof.thread())?; + let session = DiscoveredSession { + target: Target::new(SessionKind::Codex, thread.id), + title: thread.title, + directory: thread.directory, + updated_at_ms: thread.updated_at_ms, + host: None, + live: false, + holder: None, + }; + Some((session, proof)) + }) + .collect() + } + + fn runtime_admitted(&self, session: &DiscoveredSession) -> bool { + session + .directory + .as_deref() + .is_some_and(|dir| !is_under(Path::new(dir), &self.config.auto_activate_excluded_dirs)) + && !self.without_fleet_owned(vec![session.clone()]).is_empty() + } + /// Whether automatic activation puts `session` on the relay: one /// predicate for activation itself and for `/setup/status`'s /// `auto_activates`, so the SessionStart hook never promises a join diff --git a/crates/relay-node/src/relay/local/session_directory.rs b/crates/relay-node/src/relay/local/session_directory.rs index 89592db26..e46cca9bd 100644 --- a/crates/relay-node/src/relay/local/session_directory.rs +++ b/crates/relay-node/src/relay/local/session_directory.rs @@ -328,7 +328,17 @@ impl LocalSessions { &*self.scans, self.config.uid, &self.codex_loaded, - ); + ) + .map(|mut rows| { + let known: HashSet<_> = rows.iter().map(|s| s.target.clone()).collect(); + rows.extend( + self.codex_runtime_sessions() + .into_iter() + .map(|(s, _)| s) + .filter(|s| !known.contains(&s.target)), + ); + rows + }); self.codex_threads .lock() .unwrap_or_else(|error| error.into_inner()) diff --git a/crates/relay-node/src/relay/local/tests/runtime_registration.rs b/crates/relay-node/src/relay/local/tests/runtime_registration.rs new file mode 100644 index 000000000..c105ec1aa --- /dev/null +++ b/crates/relay-node/src/relay/local/tests/runtime_registration.rs @@ -0,0 +1,235 @@ +//! Regression: a user chat in the desktop stdio server has no window owner +//! and is absent from the separate managed daemon's loaded-thread list. +use super::*; +use crate::relay::daemon::{Config, Core, Plugins}; +use crate::relay::paths::CorePaths; +use crate::relay::process::{codex_runtime_proofs, file_birth_ns, OpenedFile}; + +#[path = "../../../../tests/support/fake_relaycast.rs"] +mod fake_relaycast; + +const OTHER: &str = "550e8400-e29b-41d4-a716-446655440099"; + +fn seed(w: &World, ids: &[&str]) -> rusqlite::Connection { + let home = &w.sessions.config.codex.home; + std::fs::create_dir_all(home).unwrap(); + let db = rusqlite::Connection::open(home.join("state_5.sqlite")).unwrap(); + db.execute_batch("create table threads (id text primary key, title text, cwd text, + archived integer, source text, updated_at integer, has_user_event integer, thread_source text)").unwrap(); + let mut files = Vec::new(); + let locks = home.join("thread-writer-locks"); + std::fs::create_dir_all(&locks).unwrap(); + for id in ids { + db.execute("insert into threads values (?1, 'User work', '/code/relay', 0, 'vscode', 1, 1, 'user')", [id]).unwrap(); + let lock = locks.join(format!("{id}.lock")); + std::fs::write(&lock, "").unwrap(); + let path = home.join(format!("rollout-test-{id}.jsonl")); + std::fs::write( + &path, + format!( + "{}\n", + serde_json::json!({"type":"session_meta", + "payload":{"id":id,"thread_source":"user"}}) + ), + ) + .unwrap(); + files.extend([lock, path]); + } + w.table.insert( + 500, + FakeProcess { + executable: Some(codex_exe()), + arguments: vec!["codex".into(), "app-server".into()], + open_files: files, + ..process(1, 52) + }, + ); + db +} + +#[test] +fn runtime_candidates_are_independent_of_deliverability_and_recent_history() { + let w = world(); + let db = seed(&w, &[THREAD, OTHER]); + for n in 0..45 { + let id = format!("550e8400-e29b-41d4-a716-{n:012}"); + db.execute( + "insert into threads values (?1, 'History', '/code/relay', 0, 'vscode', 99, 1, 'user')", + [&id], + ) + .unwrap(); + } + assert!(w.sessions.auto_activation_sessions().is_empty()); + let candidates = w.sessions.automatic_candidates(); + assert_eq!(candidates.len(), 2); + assert!(candidates.iter().all(|c| !c.session.live + && c.runtime.is_some() + && w.sessions.automatic_candidate_current(c))); + let rows = w.sessions.sessions(); + assert!(rows.iter().any(|s| s.target.id == THREAD && !s.live)); + // More than five minutes of idle is not absence: fresh kernel proof, + // rather than a hook timestamp, keeps both targets present. + assert!(w + .sessions + .present_targets() + .contains(&Target::new(SessionKind::Codex, THREAD))); + w.table.remove(500); + assert!(candidates + .iter() + .all(|c| !w.sessions.automatic_candidate_current(c))); + assert!(w.sessions.automatic_candidates().is_empty()); +} + +#[test] +fn internal_archived_exec_and_subagent_rows_cannot_join() { + for sql in [ + "has_user_event=0", + "thread_source='internal'", + "source='exec'", + "archived=1", + "source='unknown'", + ] { + let w = world(); + let db = seed(&w, &[THREAD]); + db.execute(&format!("update threads set {sql}"), []) + .unwrap(); + assert!(w.sessions.automatic_candidates().is_empty(), "{sql}"); + } + let w = world(); + let _db = seed(&w, &[THREAD]); + let path = w + .sessions + .config + .codex + .home + .join(format!("rollout-test-{THREAD}.jsonl")); + std::fs::write( + path, + format!( + "{}\n", + serde_json::json!({"type":"session_meta","payload":{ + "id":THREAD,"thread_source":"subagent","parent_thread_id":OTHER, + "source":{"subagent":{"thread_spawn":{"parent_thread_id":OTHER}}}}}) + ), + ) + .unwrap(); + assert!(w.sessions.automatic_candidates().is_empty()); +} + +#[test] +fn proofs_reject_replaced_inodes_wrong_owner_and_process_reuse() { + for change in ["pid", "uid", "untrusted", "rollout", "lock", "held_birth"] { + let w = world(); + let _db = seed(&w, &[THREAD]); + let c = w.sessions.automatic_candidates().pop().unwrap(); + let mut host = w.table.processes.lock().unwrap()[&500].clone(); + match change { + "pid" => host.start += 1, + "uid" => host.uid += 1, + "untrusted" => host.executable = Some("/tmp/fake-codex".into()), + "lock" => host + .open_files + .retain(|p| p.extension().and_then(|s| s.to_str()) != Some("lock")), + "held_birth" => { + host.opened_files = Some( + host.open_files + .iter() + .map(|p| OpenedFile { + path: p.clone(), + birth_ns: file_birth_ns(p).map(|n| n + 1), + }) + .collect(), + ); + } + _ => { + let path = host + .open_files + .iter() + .find(|p| p.extension().and_then(|s| s.to_str()) == Some("jsonl")) + .unwrap(); + let replacement = path.with_extension("new"); + std::fs::write(&replacement, std::fs::read(path).unwrap()).unwrap(); + std::fs::rename(replacement, path).unwrap(); + } + } + w.table.insert(500, host); + assert!(!w.sessions.automatic_candidate_current(&c), "{change}"); + } +} + +#[test] +fn ambiguous_runtime_owners_and_directory_policy_deny_admission() { + let w = world(); + let db = seed(&w, &[THREAD]); + let host = w.table.processes.lock().unwrap()[&500].clone(); + w.table.insert(501, FakeProcess { start: 53, ..host }); + assert!(codex_runtime_proofs(&*w.table, UID).is_empty()); + w.table.remove(501); + db.execute("update threads set cwd='/tmp/private'", []) + .unwrap(); + assert!(w.sessions.automatic_candidates().is_empty()); +} + +#[tokio::test] +async fn loaded_stdio_chats_register_once_and_keep_identity_without_an_idle_route() { + let w = world(); + let db = seed(&w, &[THREAD, OTHER]); + let home = w._temp.path(); + let fake = fake_relaycast::FakeRelaycast::start(Default::default()).await; + let data = home.join("app-data"); + fake_relaycast::sign_in_store(&data, fake_relaycast::WORKSPACE_KEY, &fake.base_url); + let sessions = Arc::new(w.sessions); + let (core, _) = Core::with_plugins( + Config { + paths: CorePaths::for_socket(home.join("core.sock")).unwrap(), + home: home.into(), + app_data_dir: data, + env: Default::default(), + version: "test".into(), + host_name: "Test Desk.local".into(), + rediscover_every: Duration::from_secs(3600), + push_timing: Default::default(), + delivery_timing: Default::default(), + }, + Plugins { + sessions: sessions.clone(), + ..Default::default() + }, + ); + core.rediscover_workspace(); + core.store + .update(|s| s.auto_activate = Some(false)) + .unwrap(); + assert!(core.activate_new_sessions().await.is_empty()); + assert_eq!(fake.calls("POST", "/v1/agents").len(), 0); + core.store.update(|s| s.auto_activate = Some(true)).unwrap(); + let failures = core.activate_new_sessions().await; + assert!( + failures.is_empty(), + "{failures:?}; requests={:?}; log={:?}", + fake.calls("POST", "/v1/agents"), + crate::relay::request_log::snapshot() + ); + assert_eq!(fake.calls("POST", "/v1/agents").len(), 2); + assert!(core.activate_new_sessions().await.is_empty()); + assert_eq!(fake.calls("POST", "/v1/agents").len(), 2); + assert!(core.store.read(|s| s + .registrations + .iter() + .all(|r| r.auto_activated == Some(true) + && r.missing_since.is_none() + && r.added_by_session != Some(true)))); + assert!(sessions + .sessions() + .iter() + .filter(|s| s.target.kind == SessionKind::Codex) + .all(|s| !s.live)); + // Stale proof is denied at the internal registration boundary. + let c = sessions.automatic_candidates().pop().unwrap(); + db.execute( + "update threads set archived=1 where id=?1", + [c.session.target.id.as_str()], + ) + .unwrap(); + assert!(!sessions.automatic_candidate_current(&c)); +} diff --git a/crates/relay-node/src/relay/local_methods.rs b/crates/relay-node/src/relay/local_methods.rs index 9d42942cf..bc4965a12 100644 --- a/crates/relay-node/src/relay/local_methods.rs +++ b/crates/relay-node/src/relay/local_methods.rs @@ -275,6 +275,13 @@ async fn cached_sessions_list(core: &Arc) -> LocalSessions { .and_then(|directory| projects.get(directory).cloned().flatten()), directory: session.directory.clone(), live: session.live, + delivery_wait: (registration.is_some() + && !session.live + && running + && session.target.kind == crate::relay::store::SessionKind::Codex) + .then(|| { + "Registered; waiting for a verified turn hook or desktop owner.".to_string() + }), running: Some(running), holder: session.holder, updated_at_ms: session.updated_at_ms, 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 412159704..21d4a66dd 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -101,26 +101,50 @@ impl Outbound { return Ok(json!({})); }; let target = crate::relay::store::Target::new(SessionKind::Codex, thread.clone()); - let guidance = (event == HookEvent::PostToolUse - && inbox.is_ready() - && self.core.workspace().is_some() - && self.core.allows_self_registration() - && self - .core - .store - .read(|store| store.registration_for(&target).is_none()) - && inbox.needs_registration_guidance(&thread)) - .then(|| { - self.core - .plugins - .relay_send - .as_ref() - .and_then(|send| registration_context(&send.socket)) - }) - .flatten(); + let automatic = self.core.store.read(|s| s.auto_activate == Some(true)); + let identity = self.core.store.read(|s| { + s.registration_for(&target).map(|r| { + ( + r.display_name().to_string(), + r.current_address(&self.core.machine_prefix), + ) + }) + }); + let refresh = automatic && identity.is_none(); + let guidance = if event == HookEvent::PostToolUse + && automatic + && inbox.needs_registration_guidance(&thread) + { + identity.as_ref().and_then(|(name, address)| { + let commands = self.codex_socket_commands()?; + Some(json!({"hookSpecificOutput":{"hookEventName":"PostToolUse", + "additionalContext": crate::relay::hosted_mcp::codex_preamble( + Some((name.as_str(),address.as_str())), &commands)}})) + }) + } else { + (event == HookEvent::PostToolUse + && !automatic + && inbox.is_ready() + && self.core.workspace().is_some() + && self.core.allows_self_registration() + && self + .core + .store + .read(|store| store.registration_for(&target).is_none()) + && inbox.needs_registration_guidance(&thread)) + .then(|| { + self.core + .plugins + .relay_send + .as_ref() + .and_then(|send| registration_context(&send.socket)) + }) + .flatten() + }; if event == HookEvent::PostToolUse && !inbox.turns().has_offered(&thread) && guidance.is_none() + && !refresh { return Ok(json!({})); } @@ -136,17 +160,38 @@ impl Outbound { .to_string(); let directory = self.core.plugins.sessions.clone(); let injector = self.core.plugins.injector.clone(); + let core = self.core.clone(); tokio::task::spawn_blocking(move || { if !directory.codex_hook_turn_matches(pid, &thread, &turn) { return Err(OutboundError::not_a_session()); } + if refresh { + core.request_codex_discovery(); + } if event == HookEvent::PostToolUse { + let mut answer = inbox.turns().answer(&thread, event); if let Some(guidance) = guidance { if inbox.take_registration_guidance(&thread) { - return Ok(guidance); + if let Some(text) = answer + .pointer("/hookSpecificOutput/additionalContext") + .and_then(Value::as_str) + { + let mut combined = guidance; + if let Some(context) = + combined.pointer_mut("/hookSpecificOutput/additionalContext") + { + *context = json!(format!( + "{}\n\n{text}", + context.as_str().unwrap_or_default() + )); + } + answer = combined; + } else { + answer = guidance; + } } } - return Ok(inbox.turns().answer(&thread, event)); + return Ok(answer); } let target = crate::relay::store::Target::new(SessionKind::Codex, thread.clone()); let shown = !turn.is_empty() && !injector.relay_started_turn(&target, &turn); @@ -186,6 +231,14 @@ impl Outbound { thread: String, cwd: Option, ) -> Result { + let directory = self.core.plugins.sessions.clone(); + let trusted_host = + tokio::task::spawn_blocking(move || directory.trusted_codex_discovery_signal(pid)) + .await + .unwrap_or(false); + if trusted_host { + self.core.request_codex_discovery(); + } let directory = self.core.plugins.sessions.clone(); let session = tokio::task::spawn_blocking(move || { if !directory.holds_codex_thread(pid, &thread) { @@ -209,8 +262,16 @@ impl Outbound { }) .await .ok() - .flatten() - .ok_or_else(OutboundError::not_a_session)?; + .flatten(); + // A multiplexing host can request discovery without proving which + // chat will receive this response. Return no identity or context. + let Some(session) = session else { + return if trusted_host { + Ok(json!({})) + } else { + Err(OutboundError::not_a_session()) + }; + }; let Some(commands) = self.codex_socket_commands() else { return Ok(json!({})); }; diff --git a/crates/relay-node/src/relay/outbound/hooks/codex_stop/affinity.rs b/crates/relay-node/src/relay/outbound/hooks/codex_stop/affinity.rs index 78446f77a..abba02526 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop/affinity.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop/affinity.rs @@ -264,10 +264,14 @@ async fn consuming_hook_thread_affinity_preserves_registration_guidance() { .unwrap() .environment .insert("CODEX_THREAD_ID".into(), PARENT.into()); - assert!(outbound - .codex_stop_hook(CHILD_PID, outbound.uid, &fields(PARENT, "SessionStart")) - .await - .is_err()); + assert_eq!( + outbound + .codex_stop_hook(CHILD_PID, outbound.uid, &fields(PARENT, "SessionStart")) + .await + .unwrap(), + json!({}), + "a trusted host can request a global scan, never parent context" + ); assert!(inbox.needs_registration_guidance(PARENT)); assert!(outbound .core @@ -283,10 +287,13 @@ async fn consuming_hook_thread_affinity_preserves_registration_guidance() { .is_err()); assert!(inbox.needs_registration_guidance(PARENT)); for (pid, thread) in [(PARENT_PID, PARENT), (CHILD_PID, CHILD)] { - assert!(outbound - .codex_stop_hook(pid, outbound.uid, &fields(thread, "SessionStart")) - .await - .is_err()); + assert_eq!( + outbound + .codex_stop_hook(pid, outbound.uid, &fields(thread, "SessionStart")) + .await + .unwrap(), + json!({}) + ); assert!(inbox.needs_registration_guidance(thread)); let answer = outbound .codex_stop_hook(pid, outbound.uid, &fields(thread, "PostToolUse")) diff --git a/crates/relay-node/src/relay/process.rs b/crates/relay-node/src/relay/process.rs index 8af786d0b..62b6078dd 100644 --- a/crates/relay-node/src/relay/process.rs +++ b/crates/relay-node/src/relay/process.rs @@ -15,6 +15,7 @@ use std::io::BufRead; use std::path::{Path, PathBuf}; mod codex_hook; +mod codex_runtime; mod command_line; mod fake; #[cfg(any(target_os = "macos", test))] @@ -25,6 +26,7 @@ mod macos_args_bench; mod macos_signing; mod snapshot; pub(crate) use codex_hook::{has_unambiguous_codex_hook_writer, with_codex_hook_rollout}; +pub(crate) use codex_runtime::{codex_runtime_proofs, trusted_codex_host, CodexRuntimeProof}; pub use command_line::CommandLine; pub use fake::{FakeProcess, FakeProcesses}; pub use snapshot::ProcessSnapshot; diff --git a/crates/relay-node/src/relay/process/codex_runtime.rs b/crates/relay-node/src/relay/process/codex_runtime.rs new file mode 100644 index 000000000..b3eb92afc --- /dev/null +++ b/crates/relay-node/src/relay/process/codex_runtime.rs @@ -0,0 +1,160 @@ +//! Runtime presence is sufficient for admission, never for caller identity +//! or deciding which chat receives hook output. +use super::*; + +#[derive(Clone, Debug)] +pub struct CodexRuntimeProof { + process: ProcessInfo, + executable: PathBuf, + thread: String, + rollout: PathBuf, + birth: i128, + lock: PathBuf, + lock_birth: Option, +} + +impl CodexRuntimeProof { + pub(crate) fn thread(&self) -> &str { + &self.thread + } + + pub(crate) fn current(&self, table: &dyn ProcessTable, uid: u32) -> bool { + if self.process.uid != uid + || table.process(self.process.pid) != Some(self.process) + || table.zombie(self.process.pid) + || table.executable(self.process.pid).as_ref() != Some(&self.executable) + || !table.is_trusted_codex(&self.executable) + { + return false; + } + let matches = |files: &[OpenedFile]| { + files + .iter() + .any(|f| f.path == self.lock && f.birth_ns == self.lock_birth) + && files.iter().any(|f| { + f.birth_ns == Some(self.birth) + && f.path.canonicalize().ok().as_ref() == Some(&self.rollout) + }) + }; + let before = table.opened_files(self.process.pid); + if !matches(&before) + || file_birth_ns(&self.rollout) != Some(self.birth) + || rollout_proof(&self.rollout, &self.thread) != Some(RolloutProof::Root) + { + return false; + } + matches(&table.opened_files(self.process.pid)) + && table.process(self.process.pid) == Some(self.process) + && file_birth_ns(&self.rollout) == Some(self.birth) + } +} + +/// Enumerate hosts, not shell descendants bearing inherited thread tags. +pub(crate) fn codex_runtime_proofs(table: &dyn ProcessTable, uid: u32) -> Vec { + let mut found: BTreeMap = BTreeMap::new(); + let mut ambiguous = BTreeSet::new(); + let mut trusted = BTreeMap::new(); + for pid in table.all_pids() { + let Some(process) = table.process(pid).filter(|p| p.uid == uid) else { + continue; + }; + if table.zombie(pid) { + continue; + } + let Some(executable) = table.executable(pid) else { + continue; + }; + if !*trusted + .entry(executable.clone()) + .or_insert_with(|| table.is_trusted_codex(&executable)) + { + continue; + } + let before = table.opened_files(pid); + let locks: BTreeMap<_, _> = before + .iter() + .filter_map(|f| Some((codex_thread_of_lock(&f.path)?, f))) + .collect(); + if locks.is_empty() { + continue; + } + let paths: Vec<_> = before + .iter() + .filter_map(|f| { + let thread = rollout_thread_of_path(&f.path)?; + let birth = f.birth_ns?; + (locks.contains_key(&thread) && file_birth_ns(&f.path) == Some(birth)) + .then(|| f.path.clone()) + }) + .collect(); + let headers = open_rollout_proofs(&paths); + let after = table.opened_files(pid); + if table.process(pid) != Some(process) { + continue; + } + for path in paths { + let Some(thread) = rollout_thread_of_path(&path) else { + continue; + }; + if headers.get(&thread) != Some(&RolloutProof::Root) { + continue; + } + let Some(rollout) = path.canonicalize().ok() else { + continue; + }; + let Some(birth) = file_birth_ns(&rollout) else { + continue; + }; + let lock = locks[&thread]; + if !after + .iter() + .any(|f| f.path == lock.path && f.birth_ns == lock.birth_ns) + || !after + .iter() + .any(|f| f.path == path && f.birth_ns == Some(birth)) + || !before + .iter() + .any(|f| f.path == path && f.birth_ns == Some(birth)) + { + continue; + } + let proof = CodexRuntimeProof { + process, + executable: executable.clone(), + thread: thread.clone(), + rollout, + birth, + lock: lock.path.clone(), + lock_birth: lock.birth_ns, + }; + if found.insert(thread.clone(), proof).is_some() { + ambiguous.insert(thread); + } + } + } + found + .into_iter() + .filter(|(id, _)| !ambiguous.contains(id)) + .map(|(_, p)| p) + .collect() +} + +/// A SessionStart may nudge global discovery without selecting any chat. +pub(crate) fn trusted_codex_host(table: &dyn ProcessTable, pid: i32, uid: u32) -> bool { + let mut chain = Vec::new(); + let mut current = pid; + for _ in 0..MAX_ANCESTRY { + let Some(process) = table.process(current).filter(|p| p.uid == uid && p.pid > 1) else { + return false; + }; + chain.push(process); + if table + .executable(current) + .is_some_and(|exe| table.is_trusted_codex(&exe)) + { + return chain.iter().all(|p| table.process(p.pid) == Some(*p)); + } + current = process.parent; + } + false +} diff --git a/crates/relay-node/src/relay/sessions.rs b/crates/relay-node/src/relay/sessions.rs index c780d1b41..f4ba749bf 100644 --- a/crates/relay-node/src/relay/sessions.rs +++ b/crates/relay-node/src/relay/sessions.rs @@ -12,6 +12,23 @@ use serde::Serialize; use std::collections::HashSet; use std::path::PathBuf; +/// Internal admission evidence. A loaded Codex chat can join without an +/// idle delivery route; this authority can never arrive through an RPC. +#[derive(Clone, Debug)] +pub struct AutomaticCandidate { + pub session: DiscoveredSession, + pub(crate) runtime: Option, +} + +impl AutomaticCandidate { + pub fn live(session: DiscoveredSession) -> Option { + session.live.then_some(Self { + session, + runtime: None, + }) + } +} + /// A session found on this computer. #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct DiscoveredSession { @@ -214,6 +231,32 @@ pub trait SessionDirectory: Send + Sync + 'static { .collect() } + fn automatic_candidates(&self) -> Vec { + self.auto_activation_sessions() + .into_iter() + .filter_map(AutomaticCandidate::live) + .collect() + } + + /// Freshly revalidate internal authority immediately before admission. + fn automatic_candidate_current(&self, candidate: &AutomaticCandidate) -> bool { + candidate.runtime.is_none() && candidate.session.live + } + + /// Runtime presence and deliverability are deliberately distinct. + fn present_targets(&self) -> HashSet { + self.sessions() + .into_iter() + .filter(|s| s.live) + .map(|s| s.target) + .collect() + } + + /// This only schedules global discovery and grants no chat identity. + fn trusted_codex_discovery_signal(&self, _pid: i32) -> bool { + false + } + /// What each of `sessions` (with its working directory, when known) is /// doing, from its own records (`relay::activity`). May block. fn activity( From 87590b2e526582c93d8ba2c45f0dd3964f090c47 Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Wed, 11 Nov 2026 18:51:48 +1100 Subject: [PATCH 2/8] fix(relay-node): recognise native remote user turns and address review findings --- crates/relay-node/src/relay/agents.rs | 13 +- .../src/relay/codex/stop_hook/inbox.rs | 19 +++ crates/relay-node/src/relay/codex/threads.rs | 27 ++-- .../src/relay/local/directory_trait.rs | 27 ++-- .../relay/local/tests/runtime_registration.rs | 130 +++++++++++++++++- crates/relay-node/src/relay/local_methods.rs | 15 +- .../src/relay/outbound/hooks/codex_stop.rs | 42 +++++- .../src/relay/process/codex_runtime.rs | 51 +++++++ crates/relay-node/src/relay/sessions.rs | 8 +- 9 files changed, 299 insertions(+), 33 deletions(-) diff --git a/crates/relay-node/src/relay/agents.rs b/crates/relay-node/src/relay/agents.rs index 074fed32f..efcde3f7d 100644 --- a/crates/relay-node/src/relay/agents.rs +++ b/crates/relay-node/src/relay/agents.rs @@ -258,14 +258,23 @@ impl Core { return; } let core = self.clone(); + struct Reset(Arc); + impl Drop for Reset { + fn drop(&mut self) { + self.0 + .codex_discovery_refresh + .store(false, Ordering::Release); + } + } + let reset = Reset(core.clone()); tokio::spawn(async move { + let _reset = reset; for attempt in 0..3 { if attempt > 0 { tokio::time::sleep(std::time::Duration::from_millis(500)).await; } core.activate_new_sessions().await; } - core.codex_discovery_refresh.store(false, Ordering::Release); }); } @@ -1500,7 +1509,7 @@ impl Core { let discovered = tokio::task::spawn_blocking(move || { sessions.pi_auto_join(pi_auto_join); let all = sessions.sessions(); - let live = sessions.present_targets(); + let live = sessions.present_targets(&all); let targets: Vec = registered.iter().map(|(_, t)| t.clone()).collect(); let archived = sessions.archived(&targets); let archived: Vec = registered diff --git a/crates/relay-node/src/relay/codex/stop_hook/inbox.rs b/crates/relay-node/src/relay/codex/stop_hook/inbox.rs index e6de943fe..9f089d084 100644 --- a/crates/relay-node/src/relay/codex/stop_hook/inbox.rs +++ b/crates/relay-node/src/relay/codex/stop_hook/inbox.rs @@ -38,6 +38,8 @@ pub(super) struct State { /// unregistered chat gets (context, not a reason to keep a turn open). /// Bounded by [`GUIDED_LIMIT`]. registration_guidance: HashSet, + /// Discovery has already been nudged by a verified hook for this thread. + discovery_requested: HashSet, /// By lowercase thread id. held: HashMap>, /// Handed over by the hook, by message id, with when. @@ -60,6 +62,23 @@ pub struct StopHookInbox { } impl StopHookInbox { + /// Whether a verified hook may still request its first discovery pass. + pub(crate) fn needs_discovery_refresh(&self, thread: &str) -> bool { + !self + .state() + .discovery_requested + .contains(&thread.to_lowercase()) + } + + /// Only a kernel-verified hook may consume this bounded one-time nudge. + pub(crate) fn take_discovery_refresh(&self, thread: &str) -> bool { + let mut state = self.state(); + if state.discovery_requested.len() >= GUIDED_LIMIT { + state.discovery_requested.clear(); + } + state.discovery_requested.insert(thread.to_lowercase()) + } + pub(crate) fn needs_registration_guidance(&self, thread: &str) -> bool { !self .state() diff --git a/crates/relay-node/src/relay/codex/threads.rs b/crates/relay-node/src/relay/codex/threads.rs index 3090adeea..148ab492e 100644 --- a/crates/relay-node/src/relay/codex/threads.rs +++ b/crates/relay-node/src/relay/codex/threads.rs @@ -334,28 +334,37 @@ pub fn read_one(config: &CodexConfig, id: &str) -> Option { /// A root user chat held by a trusted runtime, outside the recent-history /// page. Internal/prewarmed and exec sessions never become automatic agents. -pub(crate) fn read_runtime_user(config: &CodexConfig, id: &str) -> Option { +pub(crate) fn read_runtime_user( + config: &CodexConfig, + id: &str, + native_user_turn: impl FnOnce() -> bool, +) -> Option { if !crate::relay::store::is_uuid(id) { return None; } with_connection(config, Duration::from_secs(2), |connection| { let columns = columns(connection)?; - // Older schemas keep the existing deliverable-session discovery; - // do not widen it when user-session evidence is unavailable. - if !columns.contains("has_user_event") { - return None; - } + let user_event = if columns.contains("has_user_event") { + "coalesce(has_user_event, 0)" + } else { + "0" + }; let root = if columns.contains("thread_source") { " and +thread_source = 'user'" } else { "" }; let sql = format!( - "select {} from threads where {ID_MATCHES} and +archived = 0 \ - and +source in ('cli', 'vscode') and +has_user_event = 1{root} limit 1", + "select {}, {user_event} from threads where {ID_MATCHES} and +archived = 0 \ + and +source in ('cli', 'vscode'){root} limit 1", select_list(&columns)? ); - let found = connection.query_row(&sql, [id], row).ok()?; + let (found, has_user_event) = connection + .query_row(&sql, [id], |r| Ok((row(r)?, r.get::<_, i64>(5)? == 1))) + .ok()?; + if !has_user_event && !native_user_turn() { + return None; + } if is_relay_spawned_title(found.1.as_deref().unwrap_or_default()) { return None; } diff --git a/crates/relay-node/src/relay/local/directory_trait.rs b/crates/relay-node/src/relay/local/directory_trait.rs index 49dc109e3..46bc75d9d 100644 --- a/crates/relay-node/src/relay/local/directory_trait.rs +++ b/crates/relay-node/src/relay/local/directory_trait.rs @@ -191,20 +191,20 @@ impl SessionDirectory for LocalSessions { candidate.session.target.kind == SessionKind::Codex && candidate.session.target.id == proof.thread() && proof.current(&*self.table, self.config.uid) - && codex::threads::read_runtime_user(&self.config.codex, proof.thread()).is_some_and( - |thread| { - thread.directory == candidate.session.directory - && self.runtime_admitted(&candidate.session) - }, - ) + && codex::threads::read_runtime_user(&self.config.codex, proof.thread(), || { + proof.has_native_user_turn() + }) + .is_some_and(|thread| { + thread.directory == candidate.session.directory + && self.runtime_admitted(&candidate.session) + }) } - fn present_targets(&self) -> HashSet { - let mut targets: HashSet<_> = self - .sessions() - .into_iter() + fn present_targets(&self, discovered: &[DiscoveredSession]) -> HashSet { + let mut targets: HashSet<_> = discovered + .iter() .filter(|s| s.live) - .map(|s| s.target) + .map(|s| s.target.clone()) .collect(); targets.extend( self.codex_runtime_sessions() @@ -476,7 +476,10 @@ impl LocalSessions { crate::relay::process::codex_runtime_proofs(&*self.scans, self.config.uid) .into_iter() .filter_map(|proof| { - let thread = codex::threads::read_runtime_user(&self.config.codex, proof.thread())?; + let thread = + codex::threads::read_runtime_user(&self.config.codex, proof.thread(), || { + proof.has_native_user_turn() + })?; let session = DiscoveredSession { target: Target::new(SessionKind::Codex, thread.id), title: thread.title, diff --git a/crates/relay-node/src/relay/local/tests/runtime_registration.rs b/crates/relay-node/src/relay/local/tests/runtime_registration.rs index c105ec1aa..be6ccf738 100644 --- a/crates/relay-node/src/relay/local/tests/runtime_registration.rs +++ b/crates/relay-node/src/relay/local/tests/runtime_registration.rs @@ -9,6 +9,29 @@ use crate::relay::process::{codex_runtime_proofs, file_birth_ns, OpenedFile}; mod fake_relaycast; const OTHER: &str = "550e8400-e29b-41d4-a716-446655440099"; +const TURN: &str = "550e8400-e29b-41d4-a716-446655440088"; + +fn native_user_turn(trigger: &str) -> serde_json::Value { + serde_json::json!({"type":"event_msg","payload":{ + "type":"task_started","turn_id":TURN,"root_turn_id":TURN, + "turn_attribution":{"turn_id":TURN,"root_turn_id":TURN, + "turn_trigger":trigger,"parent_turn_id":null,"initiating_agent_path":null}}}) +} + +fn append_record(w: &World, id: &str, value: &serde_json::Value) { + use std::io::Write; + let path = w + .sessions + .config + .codex + .home + .join(format!("rollout-test-{id}.jsonl")); + writeln!( + std::fs::OpenOptions::new().append(true).open(path).unwrap(), + "{value}" + ) + .unwrap(); +} fn seed(w: &World, ids: &[&str]) -> rusqlite::Connection { let home = &w.sessions.config.codex.home; @@ -71,7 +94,7 @@ fn runtime_candidates_are_independent_of_deliverability_and_recent_history() { // rather than a hook timestamp, keeps both targets present. assert!(w .sessions - .present_targets() + .present_targets(&rows) .contains(&Target::new(SessionKind::Codex, THREAD))); w.table.remove(500); assert!(candidates @@ -116,6 +139,93 @@ fn internal_archived_exec_and_subagent_rows_cannot_join() { assert!(w.sessions.automatic_candidates().is_empty()); } +#[test] +fn native_remote_and_composer_turns_admit_an_unset_database_user_flag() { + for trigger in ["remote_android", "composer"] { + for missing_column in [false, true] { + let w = world(); + let db = seed(&w, &[THREAD]); + if missing_column { + db.execute("alter table threads drop column has_user_event", []) + .unwrap(); + } else { + db.execute("update threads set has_user_event=0", []) + .unwrap(); + } + assert!(w.sessions.automatic_candidates().is_empty()); + append_record(&w, THREAD, &native_user_turn(trigger)); + let candidate = w.sessions.automatic_candidates().pop().unwrap(); + assert!(!candidate.session.live); + assert!(w.sessions.automatic_candidate_current(&candidate)); + db.execute("update threads set archived=1", []).unwrap(); + assert!(!w.sessions.automatic_candidate_current(&candidate)); + } + } +} + +#[test] +fn fallback_user_evidence_rejects_internal_turns_and_prompt_text() { + for change in [ + "agent", + "unknown", + "parent", + "initiator", + "root", + "turn", + "missing", + "prompt", + "partial", + ] { + let w = world(); + let db = seed(&w, &[THREAD]); + db.execute("update threads set has_user_event=0", []) + .unwrap(); + let mut event = native_user_turn("remote_android"); + match change { + "agent" => { + event["payload"]["turn_attribution"]["turn_trigger"] = + "app_tool_create_thread".into() + } + "unknown" => event["payload"]["turn_attribution"]["turn_trigger"] = "unknown".into(), + "parent" => event["payload"]["turn_attribution"]["parent_turn_id"] = OTHER.into(), + "initiator" => { + event["payload"]["turn_attribution"]["initiating_agent_path"] = "/agent".into() + } + "root" => event["payload"]["turn_attribution"]["root_turn_id"] = OTHER.into(), + "turn" => event["payload"]["turn_attribution"]["turn_id"] = OTHER.into(), + "missing" => { + event["payload"]["turn_attribution"] + .as_object_mut() + .unwrap() + .remove("parent_turn_id"); + } + "prompt" => { + event = serde_json::json!({"type":"response_item","payload":{ + "type":"message","role":"user","content":[{"type":"input_text","text":event.to_string()}]}}) + } + "partial" => { + use std::io::Write; + let path = w + .sessions + .config + .codex + .home + .join(format!("rollout-test-{THREAD}.jsonl")); + write!( + std::fs::OpenOptions::new().append(true).open(path).unwrap(), + "{event}" + ) + .unwrap(); + assert!(w.sessions.automatic_candidates().is_empty()); + continue; + } + _ => unreachable!(), + } + append_record(&w, THREAD, &event); + assert!(w.sessions.automatic_candidates().is_empty(), "{change}"); + } +} + #[test] fn proofs_reject_replaced_inodes_wrong_owner_and_process_reuse() { for change in ["pid", "uid", "untrusted", "rollout", "lock", "held_birth"] { @@ -174,6 +284,13 @@ fn ambiguous_runtime_owners_and_directory_policy_deny_admission() { async fn loaded_stdio_chats_register_once_and_keep_identity_without_an_idle_route() { let w = world(); let db = seed(&w, &[THREAD, OTHER]); + // Real Remote Control desktop rollouts can record user turns while + // leaving this catalog flag at zero for the whole conversation. + db.execute("update threads set has_user_event=0", []) + .unwrap(); + for id in [THREAD, OTHER] { + append_record(&w, id, &native_user_turn("remote_android")); + } let home = w._temp.path(); let fake = fake_relaycast::FakeRelaycast::start(Default::default()).await; let data = home.join("app-data"); @@ -224,6 +341,14 @@ async fn loaded_stdio_chats_register_once_and_keep_identity_without_an_idle_rout .iter() .filter(|s| s.target.kind == SessionKind::Codex) .all(|s| !s.live)); + let listed = sessions.sessions(); + *core.session_snapshot.lock().unwrap() = (1, listed); + let rows = crate::relay::local_methods::sessions_list(&core).await; + assert!(rows + .sessions + .iter() + .filter(|s| s.target.kind == SessionKind::Codex) + .all(|s| s.delivery_wait.is_some())); // Stale proof is denied at the internal registration boundary. let c = sessions.automatic_candidates().pop().unwrap(); db.execute( @@ -232,4 +357,7 @@ async fn loaded_stdio_chats_register_once_and_keep_identity_without_an_idle_rout ) .unwrap(); assert!(!sessions.automatic_candidate_current(&c)); + w.table.remove(500); + let rows = crate::relay::local_methods::sessions_list(&core).await; + assert!(rows.sessions.iter().all(|s| s.delivery_wait.is_none())); } diff --git a/crates/relay-node/src/relay/local_methods.rs b/crates/relay-node/src/relay/local_methods.rs index bc4965a12..a843a2b40 100644 --- a/crates/relay-node/src/relay/local_methods.rs +++ b/crates/relay-node/src/relay/local_methods.rs @@ -243,6 +243,19 @@ async fn cached_sessions_list(core: &Arc) -> LocalSessions { .1 .clone(); let store = core.store.snapshot(); + let present = if found.iter().any(|s| { + !s.live + && s.target.kind == crate::relay::store::SessionKind::Codex + && store.registration_for(&s.target).is_some() + }) { + let sessions = core.plugins.sessions.clone(); + let rows = found.clone(); + tokio::task::spawn_blocking(move || sessions.present_targets(&rows)) + .await + .unwrap_or_default() + } else { + Default::default() + }; let status = core.status(); let machine_prefix = core.machine_prefix.clone(); let stalls = core.inbox_stalls(); @@ -277,7 +290,7 @@ async fn cached_sessions_list(core: &Arc) -> LocalSessions { live: session.live, delivery_wait: (registration.is_some() && !session.live - && running + && present.contains(&session.target) && session.target.kind == crate::relay::store::SessionKind::Codex) .then(|| { "Registered; waiting for a verified turn hook or desktop owner.".to_string() 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 21d4a66dd..8dd78fb4c 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -110,7 +110,7 @@ impl Outbound { ) }) }); - let refresh = automatic && identity.is_none(); + let refresh = automatic && identity.is_none() && inbox.needs_discovery_refresh(&thread); let guidance = if event == HookEvent::PostToolUse && automatic && inbox.needs_registration_guidance(&thread) @@ -122,8 +122,11 @@ impl Outbound { Some((name.as_str(),address.as_str())), &commands)}})) }) } else { + None + }; + let guidance = guidance.or_else(|| { (event == HookEvent::PostToolUse - && !automatic + && identity.is_none() && inbox.is_ready() && self.core.workspace().is_some() && self.core.allows_self_registration() @@ -140,7 +143,7 @@ impl Outbound { .and_then(|send| registration_context(&send.socket)) }) .flatten() - }; + }); if event == HookEvent::PostToolUse && !inbox.turns().has_offered(&thread) && guidance.is_none() @@ -165,7 +168,7 @@ impl Outbound { if !directory.codex_hook_turn_matches(pid, &thread, &turn) { return Err(OutboundError::not_a_session()); } - if refresh { + if refresh && inbox.take_discovery_refresh(&thread) { core.request_codex_discovery(); } if event == HookEvent::PostToolUse { @@ -454,6 +457,37 @@ mod tests { ); } + #[tokio::test] + async fn automatic_unregistered_chats_keep_manual_guidance_and_one_discovery_nudge() { + let (_temp, outbound, inbox) = fixture(); + outbound + .core + .store + .update(|s| s.auto_activate = Some(true)) + .unwrap(); + assert!(outbound + .codex_stop_hook(99, outbound.uid, &fields()) + .await + .is_err()); + assert!(inbox.needs_discovery_refresh(THREAD)); + let answer = outbound + .codex_stop_hook(42, outbound.uid, &fields()) + .await + .unwrap(); + assert!(answer.to_string().contains(" register --socket ")); + assert!(!inbox.needs_discovery_refresh(THREAD)); + assert!(!inbox.needs_registration_guidance(THREAD)); + // Once there is no pending delivery, later hooks take the fast path + // rather than repeatedly probing affinity and scheduling discovery. + assert_eq!( + outbound + .codex_stop_hook(99, outbound.uid, &fields()) + .await + .unwrap(), + json!({}) + ); + } + #[tokio::test] async fn forged_thread_cannot_get_or_consume_registration_guidance() { let (_temp, outbound, inbox) = fixture(); diff --git a/crates/relay-node/src/relay/process/codex_runtime.rs b/crates/relay-node/src/relay/process/codex_runtime.rs index b3eb92afc..a9c05f301 100644 --- a/crates/relay-node/src/relay/process/codex_runtime.rs +++ b/crates/relay-node/src/relay/process/codex_runtime.rs @@ -18,6 +18,57 @@ impl CodexRuntimeProof { &self.thread } + /// New desktop builds can leave `has_user_event` unset for a real + /// composer or Android Remote Control turn. Trust native attribution + /// from this held root rollout, never a prompt's text or role alone. + pub(crate) fn has_native_user_turn(&self) -> bool { + const MAX_SCAN: u64 = 512 * 1024; + if file_birth_ns(&self.rollout) != Some(self.birth) + || rollout_proof(&self.rollout, &self.thread) != Some(RolloutProof::Root) + { + return false; + } + let Ok(file) = std::fs::File::open(&self.rollout) else { + return false; + }; + let mut reader = std::io::BufReader::new(std::io::Read::take(file, MAX_SCAN)); + let mut line = String::new(); + for _ in 0..64 { + line.clear(); + if reader.read_line(&mut line).ok().is_none_or(|n| n == 0) || !line.ends_with('\n') { + return false; + } + let Ok(value) = serde_json::from_str::(&line) else { + return false; + }; + if value["type"] != "event_msg" || value["payload"]["type"] != "task_started" { + continue; + } + let payload = &value["payload"]; + let attribution = &payload["turn_attribution"]; + let Some(turn) = payload["turn_id"].as_str() else { + continue; + }; + if crate::relay::store::is_uuid(turn) + && matches!( + attribution["turn_trigger"].as_str(), + Some("composer" | "remote_android") + ) + && attribution["turn_id"].as_str() == Some(turn) + && attribution["root_turn_id"].as_str() == Some(turn) + && attribution + .get("parent_turn_id") + .is_some_and(serde_json::Value::is_null) + && attribution + .get("initiating_agent_path") + .is_some_and(serde_json::Value::is_null) + { + return file_birth_ns(&self.rollout) == Some(self.birth); + } + } + false + } + pub(crate) fn current(&self, table: &dyn ProcessTable, uid: u32) -> bool { if self.process.uid != uid || table.process(self.process.pid) != Some(self.process) diff --git a/crates/relay-node/src/relay/sessions.rs b/crates/relay-node/src/relay/sessions.rs index f4ba749bf..d92632d61 100644 --- a/crates/relay-node/src/relay/sessions.rs +++ b/crates/relay-node/src/relay/sessions.rs @@ -244,11 +244,11 @@ pub trait SessionDirectory: Send + Sync + 'static { } /// Runtime presence and deliverability are deliberately distinct. - fn present_targets(&self) -> HashSet { - self.sessions() - .into_iter() + fn present_targets(&self, discovered: &[DiscoveredSession]) -> HashSet { + discovered + .iter() .filter(|s| s.live) - .map(|s| s.target) + .map(|s| s.target.clone()) .collect() } From 8e21627cd02962a9962466fc72c6c6b82a76150f Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Wed, 11 Nov 2026 19:39:14 +1100 Subject: [PATCH 3/8] fix(relay-node): separate identity guidance and cache runtime presence --- crates/relay-node/src/relay/agents.rs | 10 +++ .../src/relay/codex/stop_hook/inbox.rs | 28 +++++++ crates/relay-node/src/relay/daemon.rs | 9 ++ .../src/relay/local/directory_trait.rs | 17 +++- .../relay/local/tests/runtime_registration.rs | 54 ++++++++++++ crates/relay-node/src/relay/local_methods.rs | 23 +++-- .../src/relay/outbound/hooks/codex_stop.rs | 84 +++++++++++++++++-- .../src/relay/process/codex_runtime.rs | 6 +- crates/relay-node/src/relay/sessions.rs | 2 + 9 files changed, 207 insertions(+), 26 deletions(-) diff --git a/crates/relay-node/src/relay/agents.rs b/crates/relay-node/src/relay/agents.rs index f50a35313..814ec6b75 100644 --- a/crates/relay-node/src/relay/agents.rs +++ b/crates/relay-node/src/relay/agents.rs @@ -1523,8 +1523,18 @@ impl Core { }) .await; if let Ok((_, _, live, _, _)) = &discovered { + *self + .runtime_presence + .lock() + .unwrap_or_else(|e| e.into_inner()) = + Some((std::time::Instant::now(), live.clone())); // Relay Connect credentials follow the same running sessions. self.connect.lock().unwrap().observe_live_targets(live); + } else { + *self + .runtime_presence + .lock() + .unwrap_or_else(|e| e.into_inner()) = None; } let (found, all, mut live, archived, cleared) = discovered.unwrap_or_default(); self.mark_sessions_archived(&archived); diff --git a/crates/relay-node/src/relay/codex/stop_hook/inbox.rs b/crates/relay-node/src/relay/codex/stop_hook/inbox.rs index 9f089d084..cd6e43130 100644 --- a/crates/relay-node/src/relay/codex/stop_hook/inbox.rs +++ b/crates/relay-node/src/relay/codex/stop_hook/inbox.rs @@ -38,6 +38,9 @@ pub(super) struct State { /// unregistered chat gets (context, not a reason to keep a turn open). /// Bounded by [`GUIDED_LIMIT`]. registration_guidance: HashSet, + /// Registered address last announced to each thread; manual guidance + /// cannot spend this marker, and a new registration can be announced. + registered_identity: HashMap, /// Discovery has already been nudged by a verified hook for this thread. discovery_requested: HashSet, /// By lowercase thread id. @@ -62,6 +65,31 @@ pub struct StopHookInbox { } impl StopHookInbox { + /// Whether this actual registered address still needs its preamble. + pub(crate) fn needs_registered_identity(&self, thread: &str, address: &str) -> bool { + self.state() + .registered_identity + .get(&thread.to_lowercase()) + .map(String::as_str) + != Some(address) + } + + /// Consume identity context only after the receiving turn is verified. + pub(crate) fn take_registered_identity(&self, thread: &str, address: &str) -> bool { + let mut state = self.state(); + let thread = thread.to_lowercase(); + if state.registered_identity.get(&thread).map(String::as_str) == Some(address) { + return false; + } + if state.registered_identity.len() >= GUIDED_LIMIT { + state.registered_identity.clear(); + } + state + .registered_identity + .insert(thread, address.to_string()); + true + } + /// Whether a verified hook may still request its first discovery pass. pub(crate) fn needs_discovery_refresh(&self, thread: &str) -> bool { !self diff --git a/crates/relay-node/src/relay/daemon.rs b/crates/relay-node/src/relay/daemon.rs index 165b32da3..999c37dfb 100644 --- a/crates/relay-node/src/relay/daemon.rs +++ b/crates/relay-node/src/relay/daemon.rs @@ -161,6 +161,14 @@ pub struct Core { pub project_names: crate::relay::project::Names, /// Completed provider snapshots. Bridge reads never scan providers. pub(crate) session_snapshot: crate::relay::local_methods::PresentationSnapshot, + /// Presence from completed background discovery; listings never query + /// provider databases to derive a delivery-wait explanation. + pub(crate) runtime_presence: Mutex< + Option<( + std::time::Instant, + std::collections::HashSet, + )>, + >, /// Native title and mobile-state publications fence older roster snapshots. pub(crate) roster_revision: std::sync::atomic::AtomicU64, /// Coalesces authenticated startup signals; never selects a chat. @@ -295,6 +303,7 @@ impl Core { plugins, project_names, session_snapshot: Default::default(), + runtime_presence: Default::default(), roster_revision: std::sync::atomic::AtomicU64::new(0), codex_discovery_refresh: std::sync::atomic::AtomicBool::new(false), target_locks: KeyedLocks::default(), diff --git a/crates/relay-node/src/relay/local/directory_trait.rs b/crates/relay-node/src/relay/local/directory_trait.rs index 46bc75d9d..73565d2b9 100644 --- a/crates/relay-node/src/relay/local/directory_trait.rs +++ b/crates/relay-node/src/relay/local/directory_trait.rs @@ -160,6 +160,7 @@ impl SessionDirectory for LocalSessions { .collect() } + /// Combine live routes with internal evidence of loaded user chats. fn automatic_candidates(&self) -> Vec { use crate::relay::sessions::AutomaticCandidate; let mut found: HashMap = self @@ -181,6 +182,7 @@ impl SessionDirectory for LocalSessions { found.into_values().collect() } + /// A discovery result must still prove its authority at admission. fn automatic_candidate_current( &self, candidate: &crate::relay::sessions::AutomaticCandidate, @@ -200,20 +202,27 @@ impl SessionDirectory for LocalSessions { }) } + /// Fleet ownership excludes lifetime presence; directory admission does not. fn present_targets(&self, discovered: &[DiscoveredSession]) -> HashSet { let mut targets: HashSet<_> = discovered .iter() .filter(|s| s.live) .map(|s| s.target.clone()) .collect(); + let runtime = self + .codex_runtime_sessions() + .into_iter() + .map(|(s, _)| s) + .collect(); targets.extend( - self.codex_runtime_sessions() + self.without_fleet_owned(runtime) .into_iter() - .map(|(s, _)| s.target), + .map(|s| s.target), ); targets } + /// A trusted host ancestry grants only a global discovery nudge. fn trusted_codex_discovery_signal(&self, pid: i32) -> bool { crate::relay::process::trusted_codex_host(&*self.table, pid, self.config.uid) } @@ -470,6 +479,7 @@ fn fallback_title(kind: SessionKind) -> &'static str { } impl LocalSessions { + /// Current root user chats held by a runtime, independent of routing. pub(super) fn codex_runtime_sessions( &self, ) -> Vec<(DiscoveredSession, crate::relay::process::CodexRuntimeProof)> { @@ -481,7 +491,7 @@ impl LocalSessions { proof.has_native_user_turn() })?; let session = DiscoveredSession { - target: Target::new(SessionKind::Codex, thread.id), + target: Target::new(SessionKind::Codex, proof.thread().to_string()), title: thread.title, directory: thread.directory, updated_at_ms: thread.updated_at_ms, @@ -494,6 +504,7 @@ impl LocalSessions { .collect() } + /// Apply directory and fleet policy before automatic registration. fn runtime_admitted(&self, session: &DiscoveredSession) -> bool { session .directory diff --git a/crates/relay-node/src/relay/local/tests/runtime_registration.rs b/crates/relay-node/src/relay/local/tests/runtime_registration.rs index be6ccf738..4e245f8ee 100644 --- a/crates/relay-node/src/relay/local/tests/runtime_registration.rs +++ b/crates/relay-node/src/relay/local/tests/runtime_registration.rs @@ -349,6 +349,21 @@ async fn loaded_stdio_chats_register_once_and_keep_identity_without_an_idle_rout .iter() .filter(|s| s.target.kind == SessionKind::Codex) .all(|s| s.delivery_wait.is_some())); + // Listing must stay off the provider/SQLite path, even while that + // provider's database is exclusively locked by its own writer. + db.execute_batch("begin exclusive").unwrap(); + tokio::time::timeout( + Duration::from_millis(250), + crate::relay::local_methods::sessions_list(&core), + ) + .await + .expect("cached listing must not wait for the Codex database"); + db.execute_batch("rollback").unwrap(); + core.runtime_presence.lock().unwrap().as_mut().unwrap().0 = + std::time::Instant::now() - Duration::from_secs(61); + let rows = crate::relay::local_methods::sessions_list(&core).await; + assert!(rows.sessions.iter().all(|s| s.delivery_wait.is_none())); + assert!(core.activate_new_sessions().await.is_empty()); // Stale proof is denied at the internal registration boundary. let c = sessions.automatic_candidates().pop().unwrap(); db.execute( @@ -358,6 +373,45 @@ async fn loaded_stdio_chats_register_once_and_keep_identity_without_an_idle_rout .unwrap(); assert!(!sessions.automatic_candidate_current(&c)); w.table.remove(500); + core.activate_new_sessions().await; let rows = crate::relay::local_methods::sessions_list(&core).await; assert!(rows.sessions.iter().all(|s| s.delivery_wait.is_none())); } + +#[test] +fn duplicate_descriptors_do_not_create_a_second_runtime_owner() { + let w = world(); + let _db = seed(&w, &[THREAD]); + let mut host = w.table.processes.lock().unwrap()[&500].clone(); + host.open_files + .push(host.open_files.last().unwrap().clone()); + w.table.insert(500, host); + assert_eq!(w.sessions.automatic_candidates().len(), 1); +} + +#[test] +fn uppercase_catalog_ids_bind_to_the_normalized_runtime_proof() { + let w = world(); + let db = seed(&w, &[THREAD]); + db.execute("update threads set id=upper(id)", []).unwrap(); + let candidate = w.sessions.automatic_candidates().pop().unwrap(); + assert_eq!(candidate.session.target.id, THREAD); + assert!(w.sessions.automatic_candidate_current(&candidate)); +} + +#[test] +fn runtime_presence_filters_fleet_ownership_but_keeps_manual_excluded_chats() { + let w = world(); + let db = seed(&w, &[THREAD]); + db.execute("update threads set cwd='/tmp/private'", []) + .unwrap(); + let target = Target::new(SessionKind::Codex, THREAD); + assert!(w.sessions.automatic_candidates().is_empty()); + assert!(w.sessions.present_targets(&[]).contains(&target)); + w.sessions + .fleet_owned + .lock() + .unwrap() + .insert(target.clone()); + assert!(!w.sessions.present_targets(&[]).contains(&target)); +} diff --git a/crates/relay-node/src/relay/local_methods.rs b/crates/relay-node/src/relay/local_methods.rs index a843a2b40..f3727328d 100644 --- a/crates/relay-node/src/relay/local_methods.rs +++ b/crates/relay-node/src/relay/local_methods.rs @@ -243,19 +243,16 @@ async fn cached_sessions_list(core: &Arc) -> LocalSessions { .1 .clone(); let store = core.store.snapshot(); - let present = if found.iter().any(|s| { - !s.live - && s.target.kind == crate::relay::store::SessionKind::Codex - && store.registration_for(&s.target).is_some() - }) { - let sessions = core.plugins.sessions.clone(); - let rows = found.clone(); - tokio::task::spawn_blocking(move || sessions.present_targets(&rows)) - .await - .unwrap_or_default() - } else { - Default::default() - }; + // Runtime checks belong to background discovery, not an RPC read. If + // discovery stalls, an expired snapshot must not promise a waiting route. + let present = core + .runtime_presence + .lock() + .unwrap_or_else(|e| e.into_inner()) + .as_ref() + .filter(|(at, _)| at.elapsed() <= std::time::Duration::from_secs(60)) + .map(|(_, targets)| targets.clone()) + .unwrap_or_default(); let status = core.status(); let machine_prefix = core.machine_prefix.clone(); let stalls = core.inbox_stalls(); 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 b82d2df12..6c0b40952 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -111,16 +111,16 @@ impl Outbound { }) }); let refresh = automatic && identity.is_none() && inbox.needs_discovery_refresh(&thread); - let guidance = if event == HookEvent::PostToolUse - && automatic - && inbox.needs_registration_guidance(&thread) - { - identity.as_ref().and_then(|(name, address)| { - let commands = self.codex_socket_commands()?; - Some(json!({"hookSpecificOutput":{"hookEventName":"PostToolUse", + let guidance = if event == HookEvent::PostToolUse && automatic { + identity + .as_ref() + .filter(|(_, address)| inbox.needs_registered_identity(&thread, address)) + .and_then(|(name, address)| { + let commands = self.codex_socket_commands()?; + Some(json!({"hookSpecificOutput":{"hookEventName":"PostToolUse", "additionalContext": crate::relay::hosted_mcp::codex_preamble( Some((name.as_str(),address.as_str())), &commands)}})) - }) + }) } else { None }; @@ -174,7 +174,11 @@ impl Outbound { if event == HookEvent::PostToolUse { let mut answer = inbox.turns().answer(&thread, event); if let Some(guidance) = guidance { - if inbox.take_registration_guidance(&thread) { + let take = match identity.as_ref() { + Some((_, address)) => inbox.take_registered_identity(&thread, address), + None => inbox.take_registration_guidance(&thread), + }; + if take { if let Some(text) = answer .pointer("/hookSpecificOutput/additionalContext") .and_then(Value::as_str) @@ -489,6 +493,68 @@ mod tests { ); } + #[tokio::test] + async fn manual_guidance_cannot_spend_later_registered_identity_context() { + let (_temp, outbound, inbox) = fixture(); + outbound + .core + .store + .update(|s| s.auto_activate = Some(true)) + .unwrap(); + let first = outbound + .codex_stop_hook(42, outbound.uid, &fields()) + .await + .unwrap(); + assert!(first.to_string().contains(" register --socket ")); + assert!(!inbox.needs_registration_guidance(THREAD)); + let registration = serde_json::from_value(json!({ + "name":"codex-api", "agentID":"agent-1", + "target":{"kind":"codex", "id":THREAD}, "title":"API", + "baseURL":"https://cast.test", "registeredAt":"2026-11-02T00:00:00Z" + })) + .unwrap(); + outbound + .core + .store + .update(|s| { + assert!(s.add(registration, Secret::new("fixture-only"))); + }) + .unwrap(); + let address = outbound.core.store.read(|s| { + s.registration_for(&Target::new(SessionKind::Codex, THREAD)) + .unwrap() + .current_address(&outbound.core.machine_prefix) + }); + assert!(inbox.needs_registered_identity(THREAD, &address)); + assert!(outbound + .codex_stop_hook(99, outbound.uid, &fields()) + .await + .is_err()); + assert!(inbox.needs_registered_identity(THREAD, &address)); + inbox + .turns() + .offer(THREAD, "late-message", "late delivery fixture") + .unwrap(); + let answer = outbound + .codex_stop_hook(42, outbound.uid, &fields()) + .await + .unwrap(); + let context = answer["hookSpecificOutput"]["additionalContext"] + .as_str() + .unwrap(); + assert!(context.contains("codex-api")); + assert!(context.contains("late delivery fixture")); + assert!(!context.contains("This Codex chat is not registered")); + assert!(!inbox.needs_registered_identity(THREAD, &address)); + assert_eq!( + outbound + .codex_stop_hook(99, outbound.uid, &fields()) + .await + .unwrap(), + json!({}) + ); + } + #[tokio::test] async fn forged_thread_cannot_get_or_consume_registration_guidance() { let (_temp, outbound, inbox) = fixture(); diff --git a/crates/relay-node/src/relay/process/codex_runtime.rs b/crates/relay-node/src/relay/process/codex_runtime.rs index a9c05f301..d5af19d64 100644 --- a/crates/relay-node/src/relay/process/codex_runtime.rs +++ b/crates/relay-node/src/relay/process/codex_runtime.rs @@ -14,6 +14,7 @@ pub struct CodexRuntimeProof { } impl CodexRuntimeProof { + /// Canonical thread id bound to the held rollout and writer lock. pub(crate) fn thread(&self) -> &str { &self.thread } @@ -69,6 +70,7 @@ impl CodexRuntimeProof { false } + /// Revalidate process incarnation, executable, held lock and rollout. pub(crate) fn current(&self, table: &dyn ProcessTable, uid: u32) -> bool { if self.process.uid != uid || table.process(self.process.pid) != Some(self.process) @@ -129,7 +131,7 @@ pub(crate) fn codex_runtime_proofs(table: &dyn ProcessTable, uid: u32) -> Vec = before + let mut paths: Vec<_> = before .iter() .filter_map(|f| { let thread = rollout_thread_of_path(&f.path)?; @@ -138,6 +140,8 @@ pub(crate) fn codex_runtime_proofs(table: &dyn ProcessTable, uid: u32) -> Vec Option { session.live.then_some(Self { session, @@ -231,6 +232,7 @@ pub trait SessionDirectory: Send + Sync + 'static { .collect() } + /// Internal admission evidence; public discovery is not caller authority. fn automatic_candidates(&self) -> Vec { self.auto_activation_sessions() .into_iter() From 10ca12ff33bf1772cd5901ae7b47d4a646740b06 Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Wed, 11 Nov 2026 19:51:49 +1100 Subject: [PATCH 4/8] fix(relay-node): bind Unix runtime proofs to exact files --- .../relay/local/tests/runtime_registration.rs | 19 ++++++++-- .../src/relay/outbound/hooks/codex_stop.rs | 7 +++- .../src/relay/process/codex_runtime.rs | 36 +++++++++++++++++-- 3 files changed, 57 insertions(+), 5 deletions(-) diff --git a/crates/relay-node/src/relay/local/tests/runtime_registration.rs b/crates/relay-node/src/relay/local/tests/runtime_registration.rs index 4e245f8ee..005a6d878 100644 --- a/crates/relay-node/src/relay/local/tests/runtime_registration.rs +++ b/crates/relay-node/src/relay/local/tests/runtime_registration.rs @@ -228,7 +228,15 @@ fn fallback_user_evidence_rejects_internal_turns_and_prompt_text() { #[test] fn proofs_reject_replaced_inodes_wrong_owner_and_process_reuse() { - for change in ["pid", "uid", "untrusted", "rollout", "lock", "held_birth"] { + for change in [ + "pid", + "uid", + "untrusted", + "rollout", + "lock", + "lock_replaced", + "held_birth", + ] { let w = world(); let _db = seed(&w, &[THREAD]); let c = w.sessions.automatic_candidates().pop().unwrap(); @@ -255,7 +263,14 @@ fn proofs_reject_replaced_inodes_wrong_owner_and_process_reuse() { let path = host .open_files .iter() - .find(|p| p.extension().and_then(|s| s.to_str()) == Some("jsonl")) + .find(|p| { + p.extension().and_then(|s| s.to_str()) + == Some(if change == "lock_replaced" { + "lock" + } else { + "jsonl" + }) + }) .unwrap(); let replacement = path.with_extension("new"); std::fs::write(&replacement, std::fs::read(path).unwrap()).unwrap(); 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 6c0b40952..07d604423 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -398,7 +398,9 @@ mod tests { push_timing: PushTiming::default(), delivery_timing: DeliveryTiming::default(), }; - let inbox = Arc::new(StopHookInbox::default()); + let inbox = Arc::new(StopHookInbox::with_turns( + crate::relay::codex::turn_hook::TurnInbox::open(home.join("turns.json")), + )); inbox.set_ready(true); let direct = Arc::new(Direct); let plugins = Plugins { @@ -493,9 +495,12 @@ mod tests { ); } + /// Registration may complete between two hooks after manual guidance. #[tokio::test] async fn manual_guidance_cannot_spend_later_registered_identity_context() { let (_temp, outbound, inbox) = fixture(); + let send = outbound.core.plugins.relay_send.as_ref().unwrap(); + send.serving.set(Some(send.socket.clone())); outbound .core .store diff --git a/crates/relay-node/src/relay/process/codex_runtime.rs b/crates/relay-node/src/relay/process/codex_runtime.rs index d5af19d64..0bd14c4f6 100644 --- a/crates/relay-node/src/relay/process/codex_runtime.rs +++ b/crates/relay-node/src/relay/process/codex_runtime.rs @@ -9,8 +9,10 @@ pub struct CodexRuntimeProof { thread: String, rollout: PathBuf, birth: i128, + rollout_key: Option<(u64, u64)>, lock: PathBuf, lock_birth: Option, + lock_key: Option<(u64, u64)>, } impl CodexRuntimeProof { @@ -25,6 +27,7 @@ impl CodexRuntimeProof { pub(crate) fn has_native_user_turn(&self) -> bool { const MAX_SCAN: u64 = 512 * 1024; if file_birth_ns(&self.rollout) != Some(self.birth) + || file_key(&self.rollout) != self.rollout_key || rollout_proof(&self.rollout, &self.thread) != Some(RolloutProof::Root) { return false; @@ -64,7 +67,8 @@ impl CodexRuntimeProof { .get("initiating_agent_path") .is_some_and(serde_json::Value::is_null) { - return file_birth_ns(&self.rollout) == Some(self.birth); + return file_birth_ns(&self.rollout) == Some(self.birth) + && file_key(&self.rollout) == self.rollout_key; } } false @@ -72,7 +76,9 @@ impl CodexRuntimeProof { /// Revalidate process incarnation, executable, held lock and rollout. pub(crate) fn current(&self, table: &dyn ProcessTable, uid: u32) -> bool { - if self.process.uid != uid + if file_key(&self.rollout) != self.rollout_key + || file_key(&self.lock) != self.lock_key + || self.process.uid != uid || table.process(self.process.pid) != Some(self.process) || table.zombie(self.process.pid) || table.executable(self.process.pid).as_ref() != Some(&self.executable) @@ -99,6 +105,8 @@ impl CodexRuntimeProof { matches(&table.opened_files(self.process.pid)) && table.process(self.process.pid) == Some(self.process) && file_birth_ns(&self.rollout) == Some(self.birth) + && file_key(&self.rollout) == self.rollout_key + && file_key(&self.lock) == self.lock_key } } @@ -173,14 +181,22 @@ pub(crate) fn codex_runtime_proofs(table: &dyn ProcessTable, uid: u32) -> Vec Vec Option<(u64, u64)> { + use std::os::unix::fs::MetadataExt; + let metadata = std::fs::metadata(path).ok()?; + Some((metadata.dev(), metadata.ino())) +} + +#[cfg(not(unix))] +/// Other platforms retain their existing held-file creation-time proof. +fn file_key(_: &Path) -> Option<(u64, u64)> { + None +} + /// A SessionStart may nudge global discovery without selecting any chat. pub(crate) fn trusted_codex_host(table: &dyn ProcessTable, pid: i32, uid: u32) -> bool { let mut chain = Vec::new(); From 03cdf2b4d4de6a5008bd61e5a0fc72d3b1b3de71 Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Wed, 11 Nov 2026 20:22:51 +1100 Subject: [PATCH 5/8] fix(relay-node): bind runtime admission to the configured Codex home --- .../src/relay/local/directory_trait.rs | 44 ++++++----- .../relay/local/tests/runtime_registration.rs | 77 ++++++++++++++++++- crates/relay-node/src/relay/local_methods.rs | 1 + .../src/relay/outbound/hooks/codex_stop.rs | 21 +++-- .../src/relay/process/codex_runtime.rs | 32 ++++++-- 5 files changed, 144 insertions(+), 31 deletions(-) diff --git a/crates/relay-node/src/relay/local/directory_trait.rs b/crates/relay-node/src/relay/local/directory_trait.rs index 73565d2b9..1b62bbf4c 100644 --- a/crates/relay-node/src/relay/local/directory_trait.rs +++ b/crates/relay-node/src/relay/local/directory_trait.rs @@ -192,7 +192,7 @@ impl SessionDirectory for LocalSessions { }; candidate.session.target.kind == SessionKind::Codex && candidate.session.target.id == proof.thread() - && proof.current(&*self.table, self.config.uid) + && proof.current(&*self.table, self.config.uid, &self.config.codex.home) && codex::threads::read_runtime_user(&self.config.codex, proof.thread(), || { proof.has_native_user_turn() }) @@ -483,25 +483,29 @@ impl LocalSessions { pub(super) fn codex_runtime_sessions( &self, ) -> Vec<(DiscoveredSession, crate::relay::process::CodexRuntimeProof)> { - crate::relay::process::codex_runtime_proofs(&*self.scans, self.config.uid) - .into_iter() - .filter_map(|proof| { - let thread = - codex::threads::read_runtime_user(&self.config.codex, proof.thread(), || { - proof.has_native_user_turn() - })?; - let session = DiscoveredSession { - target: Target::new(SessionKind::Codex, proof.thread().to_string()), - title: thread.title, - directory: thread.directory, - updated_at_ms: thread.updated_at_ms, - host: None, - live: false, - holder: None, - }; - Some((session, proof)) - }) - .collect() + crate::relay::process::codex_runtime_proofs( + &*self.scans, + self.config.uid, + &self.config.codex.home, + ) + .into_iter() + .filter_map(|proof| { + let thread = + codex::threads::read_runtime_user(&self.config.codex, proof.thread(), || { + proof.has_native_user_turn() + })?; + let session = DiscoveredSession { + target: Target::new(SessionKind::Codex, proof.thread().to_string()), + title: thread.title, + directory: thread.directory, + updated_at_ms: thread.updated_at_ms, + host: None, + live: false, + holder: None, + }; + Some((session, proof)) + }) + .collect() } /// Apply directory and fleet policy before automatic registration. diff --git a/crates/relay-node/src/relay/local/tests/runtime_registration.rs b/crates/relay-node/src/relay/local/tests/runtime_registration.rs index 005a6d878..e71162077 100644 --- a/crates/relay-node/src/relay/local/tests/runtime_registration.rs +++ b/crates/relay-node/src/relay/local/tests/runtime_registration.rs @@ -33,6 +33,7 @@ fn append_record(w: &World, id: &str, value: &serde_json::Value) { .unwrap(); } +/// Model a desktop writer holding root rollouts while IPC has no window owner. fn seed(w: &World, ids: &[&str]) -> rusqlite::Connection { let home = &w.sessions.config.codex.home; std::fs::create_dir_all(home).unwrap(); @@ -71,6 +72,7 @@ fn seed(w: &World, ids: &[&str]) -> rusqlite::Connection { } #[test] +/// History limits and missing delivery routes must not hide loaded user chats. fn runtime_candidates_are_independent_of_deliverability_and_recent_history() { let w = world(); let db = seed(&w, &[THREAD, OTHER]); @@ -104,6 +106,7 @@ fn runtime_candidates_are_independent_of_deliverability_and_recent_history() { } #[test] +/// Runtime ownership alone cannot admit internal or archived catalog records. fn internal_archived_exec_and_subagent_rows_cannot_join() { for sql in [ "has_user_event=0", @@ -140,6 +143,7 @@ fn internal_archived_exec_and_subagent_rows_cannot_join() { } #[test] +/// Native user attribution compensates for absent or stale catalog flags. fn native_remote_and_composer_turns_admit_an_unset_database_user_flag() { for trigger in ["remote_android", "composer"] { for missing_column in [false, true] { @@ -164,6 +168,7 @@ fn native_remote_and_composer_turns_admit_an_unset_database_user_flag() { } #[test] +/// Prompt text, incomplete records and agent-triggered turns confer no authority. fn fallback_user_evidence_rejects_internal_turns_and_prompt_text() { for change in [ "agent", @@ -227,6 +232,7 @@ fn fallback_user_evidence_rejects_internal_turns_and_prompt_text() { } #[test] +/// A proof must fail once its process incarnation or either held file changes. fn proofs_reject_replaced_inodes_wrong_owner_and_process_reuse() { for change in [ "pid", @@ -283,12 +289,13 @@ fn proofs_reject_replaced_inodes_wrong_owner_and_process_reuse() { } #[test] +/// Distinct owners remain ambiguous and directory admission remains enforced. fn ambiguous_runtime_owners_and_directory_policy_deny_admission() { let w = world(); let db = seed(&w, &[THREAD]); let host = w.table.processes.lock().unwrap()[&500].clone(); w.table.insert(501, FakeProcess { start: 53, ..host }); - assert!(codex_runtime_proofs(&*w.table, UID).is_empty()); + assert!(codex_runtime_proofs(&*w.table, UID, &w.sessions.config.codex.home).is_empty()); w.table.remove(501); db.execute("update threads set cwd='/tmp/private'", []) .unwrap(); @@ -296,6 +303,7 @@ fn ambiguous_runtime_owners_and_directory_policy_deny_admission() { } #[tokio::test] +/// Admission, retention and cached presentation must agree across a lifecycle. async fn loaded_stdio_chats_register_once_and_keep_identity_without_an_idle_route() { let w = world(); let db = seed(&w, &[THREAD, OTHER]); @@ -404,6 +412,73 @@ fn duplicate_descriptors_do_not_create_a_second_runtime_owner() { assert_eq!(w.sessions.automatic_candidates().len(), 1); } +/// Reproduce a copied profile with identical thread identifiers and contents. +fn copied_runtime_files(w: &World, host: &FakeProcess) -> Vec { + let copy = w._temp.path().join("copied-codex-home"); + host.open_files + .iter() + .map(|path| { + let destination = copy.join(path.strip_prefix(&w.sessions.config.codex.home).unwrap()); + std::fs::create_dir_all(destination.parent().unwrap()).unwrap(); + std::fs::copy(path, &destination).unwrap(); + destination + }) + .collect() +} + +#[test] +/// Both held files must belong to the catalog's configured runtime home. +fn copied_or_mixed_profile_files_cannot_authorize_the_configured_catalog() { + for moved in ["both", "lock", "rollout"] { + let w = world(); + let _db = seed(&w, &[THREAD]); + let mut host = w.table.processes.lock().unwrap()[&500].clone(); + let copies = copied_runtime_files(&w, &host); + for (path, copy) in host.open_files.iter_mut().zip(copies) { + let lock = path.extension().and_then(|s| s.to_str()) == Some("lock"); + if moved == "both" || (moved == "lock") == lock { + *path = copy; + } + } + w.table.insert(500, host); + assert!(w.sessions.automatic_candidates().is_empty(), "{moved}"); + assert!(w.sessions.present_targets(&[]).is_empty(), "{moved}"); + } +} + +#[test] +/// A foreign profile cannot suppress a valid owner or replace its authority. +fn another_profiles_same_id_does_not_ambiguate_the_configured_runtime() { + let mut w = world(); + let _db = seed(&w, &[THREAD]); + let host = w.table.processes.lock().unwrap()[&500].clone(); + let files = copied_runtime_files(&w, &host); + w.table.insert( + 501, + FakeProcess { + start: 53, + open_files: files, + ..host + }, + ); + let candidate = w.sessions.automatic_candidates().pop().unwrap(); + assert!(w.sessions.automatic_candidate_current(&candidate)); + w.sessions.config.codex.home = w._temp.path().join("copied-codex-home"); + assert!(!w.sessions.automatic_candidate_current(&candidate)); +} + +#[cfg(unix)] +#[test] +fn a_symlink_to_the_configured_home_preserves_the_same_runtime_authority() { + let mut w = world(); + let _db = seed(&w, &[THREAD]); + let alias = w._temp.path().join("codex-home-alias"); + std::os::unix::fs::symlink(&w.sessions.config.codex.home, &alias).unwrap(); + w.sessions.config.codex.home = alias; + let candidate = w.sessions.automatic_candidates().pop().unwrap(); + assert!(w.sessions.automatic_candidate_current(&candidate)); +} + #[test] fn uppercase_catalog_ids_bind_to_the_normalized_runtime_proof() { let w = world(); diff --git a/crates/relay-node/src/relay/local_methods.rs b/crates/relay-node/src/relay/local_methods.rs index f3727328d..a87b8345f 100644 --- a/crates/relay-node/src/relay/local_methods.rs +++ b/crates/relay-node/src/relay/local_methods.rs @@ -235,6 +235,7 @@ pub fn discovered_sessions(core: &Core) -> Vec { .clone() } +/// Present completed discovery and unexpired presence without provider I/O. async fn cached_sessions_list(core: &Arc) -> LocalSessions { let found = core .session_snapshot 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 07d604423..0a14cbd1f 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -31,6 +31,7 @@ const JOIN_POLL: Duration = Duration::from_millis(250); #[path = "codex_stop/affinity.rs"] mod affinity; +/// Explain consent-based self-registration using this core's actual socket. fn registration_context(socket: &std::path::Path) -> Option { let command = super::super::hints::probe_session_command( socket, @@ -111,7 +112,7 @@ impl Outbound { }) }); let refresh = automatic && identity.is_none() && inbox.needs_discovery_refresh(&thread); - let guidance = if event == HookEvent::PostToolUse && automatic { + let guidance = if event == HookEvent::PostToolUse { identity .as_ref() .filter(|(_, address)| inbox.needs_registered_identity(&thread, address)) @@ -495,16 +496,15 @@ mod tests { ); } - /// Registration may complete between two hooks after manual guidance. - #[tokio::test] - async fn manual_guidance_cannot_spend_later_registered_identity_context() { + /// A late identity must reach only its verified turn, with queued messages. + async fn registered_identity_after_manual_guidance(automatic: bool) { let (_temp, outbound, inbox) = fixture(); let send = outbound.core.plugins.relay_send.as_ref().unwrap(); send.serving.set(Some(send.socket.clone())); outbound .core .store - .update(|s| s.auto_activate = Some(true)) + .update(|s| s.auto_activate = Some(automatic)) .unwrap(); let first = outbound .codex_stop_hook(42, outbound.uid, &fields()) @@ -560,6 +560,17 @@ mod tests { ); } + /// Registration may complete between two hooks after manual guidance. + #[tokio::test] + async fn manual_guidance_cannot_spend_later_registered_identity_context() { + registered_identity_after_manual_guidance(true).await; + } + + #[tokio::test] + async fn manual_registration_gets_verified_identity_when_automatic_activation_is_off() { + registered_identity_after_manual_guidance(false).await; + } + #[tokio::test] async fn forged_thread_cannot_get_or_consume_registration_guidance() { let (_temp, outbound, inbox) = fixture(); diff --git a/crates/relay-node/src/relay/process/codex_runtime.rs b/crates/relay-node/src/relay/process/codex_runtime.rs index 0bd14c4f6..74b68fe11 100644 --- a/crates/relay-node/src/relay/process/codex_runtime.rs +++ b/crates/relay-node/src/relay/process/codex_runtime.rs @@ -75,8 +75,13 @@ impl CodexRuntimeProof { } /// Revalidate process incarnation, executable, held lock and rollout. - pub(crate) fn current(&self, table: &dyn ProcessTable, uid: u32) -> bool { - if file_key(&self.rollout) != self.rollout_key + pub(crate) fn current(&self, table: &dyn ProcessTable, uid: u32, home: &Path) -> bool { + let Ok(home) = home.canonicalize() else { + return false; + }; + if !self.rollout.starts_with(&home) + || !lock_in_home(&self.lock, &home) + || file_key(&self.rollout) != self.rollout_key || file_key(&self.lock) != self.lock_key || self.process.uid != uid || table.process(self.process.pid) != Some(self.process) @@ -111,7 +116,14 @@ impl CodexRuntimeProof { } /// Enumerate hosts, not shell descendants bearing inherited thread tags. -pub(crate) fn codex_runtime_proofs(table: &dyn ProcessTable, uid: u32) -> Vec { +pub(crate) fn codex_runtime_proofs( + table: &dyn ProcessTable, + uid: u32, + home: &Path, +) -> Vec { + let Ok(home) = home.canonicalize() else { + return Vec::new(); + }; let mut found: BTreeMap = BTreeMap::new(); let mut ambiguous = BTreeSet::new(); let mut trusted = BTreeMap::new(); @@ -134,6 +146,7 @@ pub(crate) fn codex_runtime_proofs(table: &dyn ProcessTable, uid: u32) -> Vec = before .iter() + .filter(|f| lock_in_home(&f.path, &home)) .filter_map(|f| Some((codex_thread_of_lock(&f.path)?, f))) .collect(); if locks.is_empty() { @@ -144,8 +157,10 @@ pub(crate) fn codex_runtime_proofs(table: &dyn ProcessTable, uid: u32) -> Vec Vec bool { + path.canonicalize() + .ok() + .is_some_and(|lock| lock.parent() == Some(home.join("thread-writer-locks").as_path())) +} + /// Unix replacements can share a creation timestamp when they happen /// rapidly; device and inode bind the exact named file as well. Other /// platforms retain the existing kernel-held creation-time evidence. From bf0aa8d4fa18e9cd62dbfc43f49c836d98b3c2e6 Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Wed, 11 Nov 2026 20:34:44 +1100 Subject: [PATCH 6/8] fix(relay-node): remember identity context supplied at session start --- crates/relay-node/src/relay/outbound/hooks/codex_stop.rs | 9 +++++++++ 1 file changed, 9 insertions(+) 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 0a14cbd1f..5a6cba4b7 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -324,6 +324,15 @@ impl Outbound { .map(|(name, address)| (name.as_str(), address.as_str())), &commands, ); + // This startup response already supplies the verified identity. + // The late tool fallback must not repeat it; a later resume or + // compaction still receives its own unconditional startup context. + if let (Some((_, address)), Some(inbox)) = ( + identity.as_ref(), + self.core.plugins.codex_stop_hook.as_ref(), + ) { + inbox.take_registered_identity(&session.target.id, address); + } Ok(json!({"hookSpecificOutput": { "hookEventName": SESSION_START, "additionalContext": text, From 4fe750b1a074279eadcc3581c66b7eeb3758f189 Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Wed, 11 Nov 2026 20:56:18 +1100 Subject: [PATCH 7/8] fix(relay-node): acknowledge startup identity after helper output --- crates/relay-node/src/relay/cli.rs | 84 +++++++++++++++-- .../src/relay/codex/stop_hook/inbox.rs | 89 +++++++++++++++++++ .../src/relay/outbound/hooks/codex_stop.rs | 45 ++++++++-- crates/relay-node/tests/relay_outbound.rs | 78 +++++++++++++++- 4 files changed, 282 insertions(+), 14 deletions(-) diff --git a/crates/relay-node/src/relay/cli.rs b/crates/relay-node/src/relay/cli.rs index 078648b59..fb5234c4f 100644 --- a/crates/relay-node/src/relay/cli.rs +++ b/crates/relay-node/src/relay/cli.rs @@ -264,7 +264,7 @@ const SESSION_START_TIMEOUT: Duration = Duration::from_secs(8); pub fn codex_stop_hook(socket: Option<&Path>, build: &Build) -> Result<()> { let mut input = Vec::new(); let _ = std::io::stdin().take(4 << 20).read_to_end(&mut input); - let answer = (|| -> Result { + let (answer, receipt) = (|| -> Result<(Value, Option<(std::path::PathBuf, Vec)>)> { let hook: Value = serde_json::from_slice(&input)?; let body = serde_json::to_vec(&json!({ "session_id": hook["session_id"], @@ -286,14 +286,48 @@ pub fn codex_stop_hook(socket: Option<&Path>, build: &Build) -> Result<()> { socket_response(&socket, "POST", "/codex/stop-hook", &body, timeout)?; anyhow::ensure!((200..300).contains(&status), "refused"); let response: Value = serde_json::from_slice(&response)?; - Ok(response["data"].clone()) + let mut answer = response["data"].clone(); + let receipt = answer + .as_object_mut() + .and_then(|object| object.remove("relayIdentityReceipt")) + .and_then(|value| value.as_str().map(str::to_string)) + .map(|receipt| { + let body = serde_json::to_vec(&json!({"session_id":hook["session_id"], + "hook_event_name":"RelayIdentityReceipt", "receipt":receipt}))?; + Ok::<_, anyhow::Error>((socket, body)) + }) + .transpose()?; + Ok((answer, receipt)) })() .ok() - .filter(Value::is_object) - .unwrap_or_else(|| json!({})); + .filter(|(answer, _)| answer.is_object()) + .unwrap_or_else(|| (json!({}), None)); let mut stdout = std::io::stdout(); - stdout.write_all(&serde_json::to_vec(&answer)?)?; + write_codex_hook_answer(&mut stdout, &answer, || { + if let Some((socket, body)) = receipt { + // Output is already flushed; acknowledgement cannot replace it + // with {} or extend the normal hook budget substantially. + let _ = socket_response( + &socket, + "POST", + "/codex/stop-hook", + &body, + Duration::from_millis(250), + ); + } + }) +} + +/// Flush context to the host before acknowledging the startup response. +fn write_codex_hook_answer( + stdout: &mut impl Write, + answer: &Value, + acknowledge: impl FnOnce(), +) -> Result<()> { + stdout.write_all(&serde_json::to_vec(answer)?)?; stdout.write_all(b"\n")?; + stdout.flush()?; + acknowledge(); Ok(()) } @@ -656,6 +690,46 @@ fn socket_response_with( mod tests { use super::*; + /// Failed writes/flushes must leave startup fallback available. + #[test] + fn codex_startup_ack_follows_successful_output_flush() { + struct Output { + bytes: Vec, + fail_write: bool, + fail_flush: bool, + } + impl Write for Output { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + if self.fail_write { + return Err(std::io::ErrorKind::BrokenPipe.into()); + } + self.bytes.extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + if self.fail_flush { + Err(std::io::ErrorKind::BrokenPipe.into()) + } else { + Ok(()) + } + } + } + for (fail_write, fail_flush) in [(false, false), (true, false), (false, true)] { + let mut output = Output { + bytes: Vec::new(), + fail_write, + fail_flush, + }; + let mut acknowledged = false; + let result = + write_codex_hook_answer(&mut output, &json!({"hookSpecificOutput":{}}), || { + acknowledged = true + }); + assert_eq!(result.is_ok(), !fail_write && !fail_flush); + assert_eq!(acknowledged, result.is_ok()); + } + } + use crate::relay::transport::{CoreListener, CoreStream}; use std::future::Future; use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; diff --git a/crates/relay-node/src/relay/codex/stop_hook/inbox.rs b/crates/relay-node/src/relay/codex/stop_hook/inbox.rs index cd6e43130..9ee493673 100644 --- a/crates/relay-node/src/relay/codex/stop_hook/inbox.rs +++ b/crates/relay-node/src/relay/codex/stop_hook/inbox.rs @@ -41,6 +41,9 @@ pub(super) struct State { /// Registered address last announced to each thread; manual guidance /// cannot spend this marker, and a new registration can be announced. registered_identity: HashMap, + /// Startup replies awaiting the same helper's acknowledgement: thread, + /// address, kernel peer pid and issue time, keyed by an opaque receipt. + startup_receipts: HashMap, /// Discovery has already been nudged by a verified hook for this thread. discovery_requested: HashSet, /// By lowercase thread id. @@ -65,6 +68,57 @@ pub struct StopHookInbox { } impl StopHookInbox { + /// Offering startup context does not prove the helper received it. + pub(crate) fn startup_identity_receipt(&self, thread: &str, address: &str, pid: i32) -> String { + let mut state = self.state(); + let thread = thread.to_lowercase(); + // Resume/compaction starts a new context lifetime. An old receipt + // must not acknowledge a later startup whose output was lost. + state.registered_identity.remove(&thread); + state.startup_receipts.retain(|_, (owner, _, _, at)| { + owner != &thread && at.elapsed() < Duration::from_secs(30) + }); + if state.startup_receipts.len() >= GUIDED_LIMIT { + state.startup_receipts.clear(); + } + let receipt = crate::relay::store::random_hex(16); + state.startup_receipts.insert( + receipt.clone(), + (thread, address.to_string(), pid, Instant::now()), + ); + receipt + } + + /// Only the original helper can confirm its unexpired startup response. + pub(crate) fn acknowledge_startup_identity( + &self, + thread: &str, + pid: i32, + receipt: &str, + ) -> bool { + let mut state = self.state(); + let Some((owner, address, peer, at)) = state.startup_receipts.get(receipt) else { + return false; + }; + if owner != &thread.to_lowercase() + || *peer != pid + || at.elapsed() >= Duration::from_secs(30) + { + return false; + } + let address = address.clone(); + state.startup_receipts.remove(receipt); + // Confirm atomically with receipt validation, so a concurrent + // startup cannot clear the marker between validation and storage. + if state.registered_identity.len() >= GUIDED_LIMIT { + state.registered_identity.clear(); + } + state + .registered_identity + .insert(thread.to_lowercase(), address); + true + } + /// Whether this actual registered address still needs its preamble. pub(crate) fn needs_registered_identity(&self, thread: &str, address: &str) -> bool { self.state() @@ -354,3 +408,38 @@ pub(super) fn recap(answered: &[(Unseen, String)]) -> String { } text } + +#[cfg(test)] +mod receipt_tests { + use super::*; + + /// Missing, forged, expired and replayed receipts cannot spend context. + #[test] + fn startup_receipts_are_private_bounded_and_replay_safe() { + let inbox = StopHookInbox::default(); + let receipt = inbox.startup_identity_receipt("THREAD", "agent@host", 42); + assert!(inbox.needs_registered_identity("thread", "agent@host")); + assert!(!inbox.acknowledge_startup_identity("other", 42, &receipt)); + assert!(!inbox.acknowledge_startup_identity("thread", 99, &receipt)); + assert!(!inbox.acknowledge_startup_identity("thread", 42, "forged")); + assert!(inbox.needs_registered_identity("thread", "agent@host")); + assert!(inbox.acknowledge_startup_identity("thread", 42, &receipt)); + assert!(!inbox.needs_registered_identity("thread", "agent@host")); + assert!(!inbox.acknowledge_startup_identity("thread", 42, &receipt)); + let old = inbox.startup_identity_receipt("thread", "agent@host", 42); + let resumed = inbox.startup_identity_receipt("thread", "agent@host", 42); + assert!(inbox.needs_registered_identity("thread", "agent@host")); + assert!(!inbox.acknowledge_startup_identity("thread", 42, &old)); + assert!(inbox.acknowledge_startup_identity("thread", 42, &resumed)); + assert!(!inbox.needs_registered_identity("thread", "agent@host")); + let expired = inbox.startup_identity_receipt("other", "new@host", 42); + inbox.state().startup_receipts.get_mut(&expired).unwrap().3 = + Instant::now() - Duration::from_secs(31); + assert!(!inbox.acknowledge_startup_identity("other", 42, &expired)); + assert!(inbox.needs_registered_identity("other", "new@host")); + for n in 0..=GUIDED_LIMIT { + inbox.startup_identity_receipt(&n.to_string(), "new@host", 42); + } + assert!(inbox.state().startup_receipts.len() <= GUIDED_LIMIT); + } +} 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 5a6cba4b7..2a9e0121d 100644 --- a/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs +++ b/crates/relay-node/src/relay/outbound/hooks/codex_stop.rs @@ -90,6 +90,17 @@ impl Outbound { ) })?; let event_name = fields.get("hook_event_name").and_then(Value::as_str); + if event_name == Some("RelayIdentityReceipt") { + let confirmed = fields + .get("receipt") + .and_then(Value::as_str) + .is_some_and(|receipt| inbox.acknowledge_startup_identity(&thread, pid, receipt)); + return if confirmed { + Ok(json!({})) + } else { + Err(OutboundError::not_a_session()) + }; + } if event_name == Some(SESSION_START) { let cwd = fields .get("cwd") @@ -324,19 +335,21 @@ impl Outbound { .map(|(name, address)| (name.as_str(), address.as_str())), &commands, ); - // This startup response already supplies the verified identity. - // The late tool fallback must not repeat it; a later resume or - // compaction still receives its own unconditional startup context. + let mut answer = json!({"hookSpecificOutput": { + "hookEventName": SESSION_START, + "additionalContext": text, + }}); + // Only the helper's receipt acknowledgement spends the fallback. + // A timeout or unreadable reply must leave later tool hooks able + // to supply identity. Resume/compaction still gets fresh context. if let (Some((_, address)), Some(inbox)) = ( identity.as_ref(), self.core.plugins.codex_stop_hook.as_ref(), ) { - inbox.take_registered_identity(&session.target.id, address); + answer["relayIdentityReceipt"] = + json!(inbox.startup_identity_receipt(&session.target.id, address, pid)); } - Ok(json!({"hookSpecificOutput": { - "hookEventName": SESSION_START, - "additionalContext": text, - }})) + Ok(answer) } /// The probe commands on this core's serving session socket, naming the @@ -539,6 +552,22 @@ mod tests { .unwrap() .current_address(&outbound.core.machine_prefix) }); + if !automatic { + // Simulate a startup reply that the helper never receives. + // Merely constructing it must not suppress the late fallback. + let start = json!({"session_id":THREAD,"hook_event_name":"SessionStart"}); + let offered = outbound + .codex_stop_hook(42, outbound.uid, start.as_object().unwrap()) + .await + .unwrap(); + let receipt = offered["relayIdentityReceipt"].as_str().unwrap(); + let forged = json!({"session_id":THREAD,"hook_event_name":"RelayIdentityReceipt", + "receipt":receipt}); + assert!(outbound + .codex_stop_hook(99, outbound.uid, forged.as_object().unwrap()) + .await + .is_err()); + } assert!(inbox.needs_registered_identity(THREAD, &address)); assert!(outbound .codex_stop_hook(99, outbound.uid, &fields()) diff --git a/crates/relay-node/tests/relay_outbound.rs b/crates/relay-node/tests/relay_outbound.rs index 24bfbd184..cb2eb8a81 100644 --- a/crates/relay-node/tests/relay_outbound.rs +++ b/crates/relay-node/tests/relay_outbound.rs @@ -1936,7 +1936,21 @@ async fn a_starting_codex_session_is_told_who_it_is_and_its_local_route() { let other = json!({"session_id": "11111111-2222-3333-4444-555555555555", "hook_event_name": "SessionStart"}); let (status, _) = code(h.call(IN_CODEX, "POST", "/codex/stop-hook", other).await); assert_eq!(status, 403); - // The other events still carry only relayed messages. + // The helper acknowledges only after flushing its startup output. + let receipt = renamed["relayIdentityReceipt"].as_str().unwrap(); + let ack = + json!({"session_id":CODEX,"hook_event_name":"RelayIdentityReceipt", "receipt":receipt}); + let (status, _) = code( + h.call(OUTSIDE, "POST", "/codex/stop-hook", ack.clone()) + .await, + ); + assert_eq!(status, 403); + h.call(IN_CODEX, "POST", "/codex/stop-hook", ack.clone()) + .await + .unwrap(); + let (status, _) = code(h.call(IN_CODEX, "POST", "/codex/stop-hook", ack).await); + assert_eq!(status, 403, "a startup receipt is consumed only once"); + // A confirmed startup suppresses redundant tool context. let tool = json!({"session_id": CODEX, "hook_event_name": "PostToolUse"}); assert_eq!( h.call(IN_CODEX, "POST", "/codex/stop-hook", tool) @@ -1947,6 +1961,68 @@ async fn a_starting_codex_session_is_told_who_it_is_and_its_local_route() { h.stop().await; } +/// The real helper strips private metadata and acknowledges after output. +#[tokio::test] +async fn a_codex_startup_helper_confirms_its_received_identity() { + use std::io::Write; + use std::process::{Command, Stdio}; + let h = Harness::start().await; + h.register(IN_CODEX, json!({"name":"codex-api"})).await; + let mut child = Command::new(env!("CARGO_BIN_EXE_agent-relay-node")) + .arg("codex-stop-hook") + .arg("--socket") + .arg(&h.relay_socket) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + h.sessions.call_from( + child.id() as i32, + session(SessionKind::Codex, CODEX, "/Users/me/Projects/relay"), + ); + child + .stdin + .take() + .unwrap() + .write_all( + json!({"session_id":CODEX,"hook_event_name":"SessionStart"}) + .to_string() + .as_bytes(), + ) + .unwrap(); + let output = tokio::task::spawn_blocking(move || child.wait_with_output().unwrap()) + .await + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let answer: Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!( + answer["hookSpecificOutput"]["hookEventName"], + "SessionStart" + ); + assert!(answer["hookSpecificOutput"]["additionalContext"] + .as_str() + .unwrap() + .contains("codex-api")); + assert!(answer.get("relayIdentityReceipt").is_none()); + assert_eq!( + h.call( + IN_CODEX, + "POST", + "/codex/stop-hook", + json!({"session_id":CODEX,"hook_event_name":"PostToolUse"}) + ) + .await + .unwrap(), + json!({}) + ); + h.stop().await; +} + // Review on #2029: a Codex session starts before activation has put it on // the relay. The hook runs activation then, says the session is joining // while its registration is held, and names it once it is registered; the From 793f712f69a549bbd24a4f72b634b07dbadc34ab Mon Sep 17 00:00:00 2001 From: "Marian-Andrei RIZOIU (poseidon)" Date: Thu, 12 Nov 2026 21:02:33 +1100 Subject: [PATCH 8/8] fix(relay-node): keep session hooks working through private runtime links --- CHANGELOG.md | 1 + crates/relay-node/src/relay/outbound.rs | 81 +++++++++++++++++++++++++ 2 files changed, 82 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 614c10d40..385d292e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Session hooks keep receiving messages when the session-socket pointer directory is a user-owned symlink to private runtime storage; inaccessible or shared symlink targets remain refused. - `agent-relay-probe connect create` starts a new room when Cloud has already closed the saved one, even before its saved expiry, instead of refusing with "already has a pending or active room". `connect end` clears a room that has expired, ended or no longer exists instead of failing; `end` and `extend` let Cloud, not the local clock, decide whether a room is over. - Loaded Codex user chats automatically join and stay registered without a desktop window owner; registration presence no longer implies an idle message-delivery route. diff --git a/crates/relay-node/src/relay/outbound.rs b/crates/relay-node/src/relay/outbound.rs index e806ea785..0d341bcaf 100644 --- a/crates/relay-node/src/relay/outbound.rs +++ b/crates/relay-node/src/relay/outbound.rs @@ -1857,6 +1857,23 @@ fn private_directory(path: &Path) -> Result<(), String> { .create(path) .map_err(|e| e.to_string())?; let metadata = fs::symlink_metadata(path).map_err(|e| e.to_string())?; + if metadata.file_type().is_symlink() { + // A private runtime directory may be reached through a user-owned + // link (for example when volatile state lives on tmpfs). Never + // chmod through it, or accept a target another user can access. + let target = fs::metadata(path).map_err(|e| e.to_string())?; + if metadata.uid() == current_uid() + && target.is_dir() + && target.uid() == current_uid() + && target.mode() & 0o077 == 0 + { + return Ok(()); + } + return Err(format!( + "{} must link to a private directory this user owns.", + path.display() + )); + } if !metadata.is_dir() || metadata.uid() != current_uid() { return Err(format!( "{} must be a directory this user owns.", @@ -2336,6 +2353,70 @@ mod tests { assert!(Build::named(false, Some("dev")).is_err()); } + #[cfg(unix)] + #[test] + fn a_socket_pointer_works_through_private_runtime_symlinks() { + use std::os::unix::fs::symlink; + + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let runtime = tempfile::tempdir().unwrap(); + symlink(state.path(), home.path().join(".agentworkforce")).unwrap(); + symlink(runtime.path(), state.path().join("desktop")).unwrap(); + let pointer = pointer_directory(home.path()).join("relay-socket"); + let socket = runtime.path().join("relay.sock"); + let mode = fs::metadata(runtime.path()).unwrap().mode(); + + let identity = write_pointer(&pointer, &socket).unwrap(); + assert_eq!( + crate::relay::cli::pointed_socket(home.path(), &Build::Production).unwrap(), + socket, + "unqualified session hooks must find the serving socket" + ); + assert_eq!(fs::metadata(&pointer).unwrap().mode() & 0o777, 0o600); + assert_eq!(fs::metadata(runtime.path()).unwrap().mode(), mode); + assert!(fs::symlink_metadata(pointer.parent().unwrap()) + .unwrap() + .file_type() + .is_symlink()); + + let serving = ServingPath::default(); + serving.start(&socket, &pointer, Some(identity)); + fs::remove_file(&pointer).unwrap(); + assert_eq!(serving.reclaim_pointer(), Reclaimed::Rewritten); + assert_eq!(serving.reclaim_pointer(), Reclaimed::Current); + remove_pointer(&pointer, &socket, serving.pointer_identity().unwrap()); + assert!(!pointer.exists()); + assert!(runtime.path().is_dir()); + } + + #[cfg(unix)] + #[test] + fn socket_pointer_refuses_accessible_symlink_targets_without_chmod() { + use std::os::unix::fs::symlink; + + let home = tempfile::tempdir().unwrap(); + for mode in [0o755, 0o770, 0o777] { + let target = tempfile::tempdir().unwrap(); + fs::set_permissions(target.path(), fs::Permissions::from_mode(mode)).unwrap(); + let link = home.path().join(format!("runtime-{mode:o}")); + symlink(target.path(), &link).unwrap(); + let pointer = link.join("relay-socket"); + assert!(write_pointer(&pointer, Path::new("/private/relay.sock")).is_err()); + assert!(!pointer.exists()); + assert_eq!(fs::metadata(target.path()).unwrap().mode() & 0o777, mode); + } + let file = home.path().join("ordinary-file"); + fs::write(&file, b"keep").unwrap(); + let file_link = home.path().join("file-link"); + symlink(&file, &file_link).unwrap(); + assert!(private_directory(&file_link).is_err()); + assert_eq!(fs::read(&file).unwrap(), b"keep"); + let broken = home.path().join("broken-link"); + symlink(home.path().join("absent"), &broken).unwrap(); + assert!(private_directory(&broken).is_err()); + } + #[cfg(unix)] #[test] fn a_build_reclaims_only_its_own_pointer() {