diff --git a/CHANGELOG.md b/CHANGELOG.md index 604ded6ec..385d292e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,9 @@ 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. ## [18.1.0] - 2026-11-11 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 41546b780..814ec6b75 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,34 @@ 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(); + 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; + } + }); + } + /// 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 +457,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 +520,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 +571,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 { @@ -1432,11 +1510,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(&all); let targets: Vec = registered.iter().map(|(_, t)| t.clone()).collect(); let archived = sessions.archived(&targets); let archived: Vec = registered @@ -1444,13 +1518,23 @@ 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; 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); @@ -1460,10 +1544,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/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 e6de943fe..9ee493673 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,14 @@ 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, + /// 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. held: HashMap>, /// Handed over by the hook, by message id, with when. @@ -60,6 +68,99 @@ 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() + .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 + .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() @@ -307,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/codex/threads.rs b/crates/relay-node/src/relay/codex/threads.rs index 1a0621a80..148ab492e 100644 --- a/crates/relay-node/src/relay/codex/threads.rs +++ b/crates/relay-node/src/relay/codex/threads.rs @@ -332,6 +332,46 @@ 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, + 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)?; + 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 {}, {user_event} from threads where {ID_MATCHES} and +archived = 0 \ + and +source in ('cli', 'vscode'){root} limit 1", + select_list(&columns)? + ); + 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; + } + 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 363ebbfbd..999c37dfb 100644 --- a/crates/relay-node/src/relay/daemon.rs +++ b/crates/relay-node/src/relay/daemon.rs @@ -161,8 +161,18 @@ 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. + 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. @@ -293,7 +303,9 @@ 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(), 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..1b62bbf4c 100644 --- a/crates/relay-node/src/relay/local/directory_trait.rs +++ b/crates/relay-node/src/relay/local/directory_trait.rs @@ -160,6 +160,73 @@ 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 + .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() + } + + /// A discovery result must still prove its authority at admission. + 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, &self.config.codex.home) + && 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) + }) + } + + /// 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.without_fleet_owned(runtime) + .into_iter() + .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) + } + fn pi_auto_join(&self, auto_join: bool) { self.config.pi.sync_auto_join(auto_join); } @@ -412,6 +479,44 @@ 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)> { + 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. + 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..e71162077 --- /dev/null +++ b/crates/relay-node/src/relay/local/tests/runtime_registration.rs @@ -0,0 +1,507 @@ +//! 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"; +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(); +} + +/// 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(); + 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] +/// 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]); + 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(&rows) + .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] +/// 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", + "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] +/// 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] { + 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] +/// 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", + "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] +/// 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", + "uid", + "untrusted", + "rollout", + "lock", + "lock_replaced", + "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(if change == "lock_replaced" { + "lock" + } else { + "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] +/// 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, &w.sessions.config.codex.home).is_empty()); + w.table.remove(501); + db.execute("update threads set cwd='/tmp/private'", []) + .unwrap(); + assert!(w.sessions.automatic_candidates().is_empty()); +} + +#[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]); + // 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"); + 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)); + 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())); + // 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( + "update threads set archived=1 where id=?1", + [c.session.target.id.as_str()], + ) + .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); +} + +/// 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(); + 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 9d42942cf..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 @@ -243,6 +244,16 @@ async fn cached_sessions_list(core: &Arc) -> LocalSessions { .1 .clone(); let store = core.store.snapshot(); + // 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(); @@ -275,6 +286,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 + && 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() + }), running: Some(running), holder: session.holder, updated_at_ms: session.updated_at_ms, 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() { 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 713bbfe06..2a9e0121d 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, @@ -89,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") @@ -101,26 +113,53 @@ 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 + 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() && inbox.needs_discovery_refresh(&thread); + let guidance = if event == HookEvent::PostToolUse { + identity .as_ref() - .and_then(|send| registration_context(&send.socket)) - }) - .flatten(); + .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 + }; + let guidance = guidance.or_else(|| { + (event == HookEvent::PostToolUse + && identity.is_none() + && 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 +175,42 @@ 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 && inbox.take_discovery_refresh(&thread) { + 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); + 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) + { + 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 +250,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 +281,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!({})); }; @@ -255,10 +335,21 @@ impl Outbound { .map(|(name, address)| (name.as_str(), address.as_str())), &commands, ); - Ok(json!({"hookSpecificOutput": { + 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(), + ) { + answer["relayIdentityReceipt"] = + json!(inbox.startup_identity_receipt(&session.target.id, address, pid)); + } + Ok(answer) } /// The probe commands on this core's serving session socket, naming the @@ -330,7 +421,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 { @@ -394,6 +487,128 @@ 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!({}) + ); + } + + /// 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(automatic)) + .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) + }); + 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()) + .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!({}) + ); + } + + /// 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/outbound/hooks/codex_stop/affinity.rs b/crates/relay-node/src/relay/outbound/hooks/codex_stop/affinity.rs index 27ad6c73b..a7ff84744 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 @@ -265,10 +265,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 @@ -284,10 +288,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..74b68fe11 --- /dev/null +++ b/crates/relay-node/src/relay/process/codex_runtime.rs @@ -0,0 +1,269 @@ +//! 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, + rollout_key: Option<(u64, u64)>, + lock: PathBuf, + lock_birth: Option, + lock_key: Option<(u64, u64)>, +} + +impl CodexRuntimeProof { + /// Canonical thread id bound to the held rollout and writer lock. + pub(crate) fn thread(&self) -> &str { + &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) + || file_key(&self.rollout) != self.rollout_key + || 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) + && file_key(&self.rollout) == self.rollout_key; + } + } + false + } + + /// Revalidate process incarnation, executable, held lock and rollout. + 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) + || 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) + && file_key(&self.rollout) == self.rollout_key + && file_key(&self.lock) == self.lock_key + } +} + +/// Enumerate hosts, not shell descendants bearing inherited thread tags. +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(); + 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(|f| lock_in_home(&f.path, &home)) + .filter_map(|f| Some((codex_thread_of_lock(&f.path)?, f))) + .collect(); + if locks.is_empty() { + continue; + } + let mut paths: Vec<_> = before + .iter() + .filter_map(|f| { + let thread = rollout_thread_of_path(&f.path)?; + let birth = f.birth_ns?; + (locks.contains_key(&thread) + && f.path.canonicalize().ok()?.starts_with(&home) + && file_birth_ns(&f.path) == Some(birth)) + .then(|| f.path.clone()) + }) + .collect(); + paths.sort(); + paths.dedup(); + 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 rollout_key = file_key(&rollout); + let lock_key = file_key(&lock.path); + #[cfg(unix)] + if rollout_key.is_none() || lock_key.is_none() { + continue; + } + let proof = CodexRuntimeProof { + process, + executable: executable.clone(), + thread: thread.clone(), + rollout_key, + rollout, + birth, + lock: lock.path.clone(), + lock_birth: lock.birth_ns, + lock_key, + }; + if found.insert(thread.clone(), proof).is_some() { + ambiguous.insert(thread); + } + } + } + found + .into_iter() + .filter(|(id, _)| !ambiguous.contains(id)) + .map(|(_, p)| p) + .collect() +} + +/// A copied profile's lock cannot confer authority on this catalog. +fn lock_in_home(path: &Path, home: &Path) -> 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. +#[cfg(unix)] +fn file_key(path: &Path) -> 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(); + 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..b90126af9 100644 --- a/crates/relay-node/src/relay/sessions.rs +++ b/crates/relay-node/src/relay/sessions.rs @@ -12,6 +12,24 @@ 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 { + /// Existing providers admit only sessions with an established live route. + 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 +232,33 @@ 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() + .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, discovered: &[DiscoveredSession]) -> HashSet { + discovered + .iter() + .filter(|s| s.live) + .map(|s| s.target.clone()) + .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( diff --git a/crates/relay-node/tests/relay_outbound.rs b/crates/relay-node/tests/relay_outbound.rs index f40977f2e..3c7b70f51 100644 --- a/crates/relay-node/tests/relay_outbound.rs +++ b/crates/relay-node/tests/relay_outbound.rs @@ -2038,7 +2038,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) @@ -2049,6 +2063,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