diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c519e93e..39c6f8088 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Copilot stop hooks release their idle wait when the person cancels, starts another turn or encounters a session error, without claiming a queued Relay message. +- `agent-relay-node` built for its own site (`AGENT_RELAY_SITE_URL`) no longer falls back to Agent Relay's public gateways: a sign-in whose site named no valid Relaycast URL, or a CLI workspace key with no recorded server or `RELAY_BASE_URL`, is refused with a reason naming the site, and `RELAYCAST_URL` is ignored in such a build. +- Copilot stop hooks release their idle wait when the person cancels, starts another turn or encounters a session error, without claiming a queued Relay message. - `POST /copilot/mcp` no longer returns an unknown-route error during native Copilot initialization: it uses Desktop's managed transport when available and otherwise reports the transport as unavailable. - `agent-relay-node` built for its own site (`AGENT_RELAY_SITE_URL`) no longer falls back to Agent Relay's public gateways: a sign-in whose site named no valid Relaycast URL, or a CLI workspace key with no recorded server or `RELAY_BASE_URL`, is refused with a reason naming the site, and `RELAYCAST_URL` is ignored in such a build. - `agent-relay-node` lists Codex sessions and delivers messages to them promptly when Codex's thread database is large. diff --git a/crates/relay-node/src/relay/cli/copilot_hook.rs b/crates/relay-node/src/relay/cli/copilot_hook.rs index fd4f7a95f..1a9c7a08a 100644 --- a/crates/relay-node/src/relay/cli/copilot_hook.rs +++ b/crates/relay-node/src/relay/cli/copilot_hook.rs @@ -1,4 +1,5 @@ use super::{session_hook_socket, socket_response}; +use crate::relay::copilot::{stop_wait, CopilotConfig}; use anyhow::Result; use serde::{ de::{DeserializeOwned, IgnoredAny}, @@ -12,10 +13,19 @@ use std::time::Duration; /// waits without model calls, and returns when the person cancels the hook, /// setup is removed, the runtime exits, or the configured timeout approaches. pub fn copilot_hook(development: bool, event: &str) -> Result<()> { - let Some(value) = input(std::io::stdin().lock(), event) else { + let Some(mut value) = input(std::io::stdin().lock(), event) else { std::io::stdout().write_all(b"{}\n")?; return Ok(()); }; + if value["relayEvent"] == "stop" && value["stopReason"] == "end_turn" { + let stamp = super::paths::home().ok().and_then(|home| { + let config = CopilotConfig::from_env(&|key| std::env::var(key).ok(), &home); + stop_wait::snapshot(&config, value["sessionId"].as_str()?) + }); + if let Some(stamp) = stamp { + value["relayJournal"] = stamp; + } + } let input = serde_json::to_vec(&value)?; let Some(socket) = session_hook_socket(None, development) else { return Ok(()); @@ -70,6 +80,7 @@ struct Hook { stop_reason: Option, tool_name: Option, tool_args: Option, + timestamp: Option, } impl Hook { fn fields(&self) -> serde_json::Map { @@ -87,6 +98,9 @@ impl Hook { ); } } + if let Some(timestamp) = self.timestamp { + fields.insert("timestamp".into(), json!(timestamp)); + } fields } } @@ -185,12 +199,14 @@ mod tests { } #[test] fn hook_forwards_identity_without_large_tool_results_or_client_binding() { - let payload = json!({"sessionId":"native-id","cwd":"/fixture","initialPrompt":"x".repeat(1000),"toolResult":"x".repeat(5 * 1024 * 1024),"toolArgs":{"patch":"x".repeat(5 * 1024 * 1024)},"relayBinding":"forged"}); + let payload = json!({"sessionId":"native-id","cwd":"/fixture","timestamp":1794029389427_u64,"relayJournal":{"identity":"forged"},"initialPrompt":"x".repeat(1000),"toolResult":"x".repeat(5 * 1024 * 1024),"toolArgs":{"patch":"x".repeat(5 * 1024 * 1024)},"relayBinding":"forged"}); let value = input(serde_json::to_vec(&payload).unwrap().as_slice(), "tool").unwrap(); assert_eq!(value["sessionId"], "native-id"); assert_eq!(value["initialPrompt"].as_str().unwrap().len(), 96); assert!(value.get("toolResult").is_none()); assert!(value.get("relayBinding").is_none()); + assert!(value.get("relayJournal").is_none()); + assert_eq!(value["timestamp"], 1794029389427_u64); assert!(serde_json::to_vec(&value).unwrap().len() < 1024); } #[test] @@ -214,5 +230,8 @@ mod tests { &json!({"relayEvent":"tool","stopReason":"end_turn"}) )); assert!(waits(&json!({"relayEvent":"stop","stopReason":"end_turn"}))); + assert!(waits( + &json!({"relayEvent":"stop","stopReason":"end_turn", "timestamp":123, "relayJournal":{}}) + )); } } diff --git a/crates/relay-node/src/relay/copilot.rs b/crates/relay-node/src/relay/copilot.rs index c6ad72104..2a67dad10 100644 --- a/crates/relay-node/src/relay/copilot.rs +++ b/crates/relay-node/src/relay/copilot.rs @@ -9,6 +9,7 @@ pub use settings::CopilotSettings; mod sessions; pub use sessions::{discover, hook_answer, session}; mod receipts; +pub(crate) mod stop_wait; pub use receipts::{context_accepted, final_answer}; mod delivery; pub use delivery::send; diff --git a/crates/relay-node/src/relay/copilot/sessions.rs b/crates/relay-node/src/relay/copilot/sessions.rs index 336db58cd..fa5ef7292 100644 --- a/crates/relay-node/src/relay/copilot/sessions.rs +++ b/crates/relay-node/src/relay/copilot/sessions.rs @@ -147,6 +147,9 @@ pub fn hook_answer( if !super::settings::caller_is_hook(config, table, pid, event) { return None; } + if cancelled_stop(config, payload, sid, event) { + return None; + } // A previously verified runtime's incarnation is enough to stop orphaned // empty polls. A descriptor/signature scan still precedes every data claim. if idle_stop(config, payload, sid, event) { @@ -175,6 +178,9 @@ fn take_answer(config: &CopilotConfig, payload: &Value, sid: &str, event: &str) if event == "stop" && payload["stopReason"].as_str() != Some("end_turn") { return Some(json!({})); } + if cancelled_stop(config, payload, sid, event) { + return None; + } let kind = if event == "stop" { crate::relay::codex::turn_hook::HookEvent::Stop } else { @@ -194,6 +200,12 @@ fn take_answer(config: &CopilotConfig, payload: &Value, sid: &str, event: &str) } Some(answer) } +fn cancelled_stop(config: &CopilotConfig, payload: &Value, sid: &str, event: &str) -> bool { + event == "stop" + && payload["stopReason"] == "end_turn" + && payload.get("timestamp").is_some() + && super::stop_wait::cancelled(config, sid, payload) +} fn observe( config: &CopilotConfig, table: &dyn ProcessTable, diff --git a/crates/relay-node/src/relay/copilot/stop_wait.rs b/crates/relay-node/src/relay/copilot/stop_wait.rs new file mode 100644 index 000000000..4b5ac949d --- /dev/null +++ b/crates/relay-node/src/relay/copilot/stop_wait.rs @@ -0,0 +1,142 @@ +//! Bounded native control records release a waiting stop hook after cancellation. +//! This never ingests history or claims a delivery. The native input timestamp +//! anchors this invocation, so an earlier abort cannot cancel a later wait. +use super::CopilotConfig; +use serde_json::{json, Value}; +use std::fs::{File, Metadata, OpenOptions}; +use std::io::{Read, Seek, SeekFrom}; +use std::path::PathBuf; + +const LIMIT: u64 = 64 << 10; + +fn path(config: &CopilotConfig, sid: &str) -> Option { + crate::relay::store::is_uuid(sid).then(|| config.sessions().join(sid).join("events.jsonl")) +} + +#[cfg(unix)] +fn identity(metadata: &Metadata) -> Option { + use std::os::unix::fs::MetadataExt; + Some(json!([metadata.dev(), metadata.ino()])) +} + +// Messaging is unavailable on Windows today. Keep the control reader portable; +// creation time is its file incarnation, as for the Codex rollout reader. +#[cfg(windows)] +fn identity(metadata: &Metadata) -> Option { + let created = metadata + .created() + .ok()? + .duration_since(std::time::UNIX_EPOCH) + .ok()?; + Some(json!([created.as_secs(), created.subsec_nanos()])) +} + +pub(crate) fn snapshot(config: &CopilotConfig, sid: &str) -> Option { + let metadata = std::fs::symlink_metadata(path(config, sid)?).ok()?; + metadata.is_file().then_some(())?; + Some(json!({"identity":identity(&metadata)?,"length":metadata.len()})) +} + +fn open(path: &std::path::Path) -> Option { + std::fs::symlink_metadata(path) + .ok()? + .is_file() + .then_some(())?; + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK); + } + options.open(path).ok() +} + +fn bytes(config: &CopilotConfig, sid: &str, stamp: &Value) -> Option> { + let path = path(config, sid)?; + let mut file = open(&path)?; + let metadata = file.metadata().ok()?; + if !metadata.is_file() + || identity(&metadata)? != stamp["identity"] + || metadata.len() < stamp["length"].as_u64()? + { + return None; + } + let bytes = read_tail_at(&mut file, metadata.len())?; + let after = std::fs::symlink_metadata(path).ok()?; + if !after.is_file() || identity(&after)? != stamp["identity"] || after.len() < metadata.len() { + return None; + } + Some(bytes) +} + +fn read_tail_at(file: &mut File, length: u64) -> Option> { + let start = length.saturating_sub(LIMIT); + let trim = if start > 0 { + file.seek(SeekFrom::Start(start - 1)).ok()?; + let mut previous = [0]; + file.read_exact(&mut previous).ok()?; + previous[0] != b'\n' + } else { + false + }; + file.seek(SeekFrom::Start(start)).ok()?; + let mut bytes = Vec::new(); + // Appends after the metadata snapshot belong to the next poll. Reading + // past that bound must not turn an ordinary append into cancellation. + file.take(length - start).read_to_end(&mut bytes).ok()?; + if trim { + let first = bytes.iter().position(|b| *b == b'\n')?; + bytes.drain(..=first); + } + Some(bytes) +} + +fn anchor(event: &Value, sid: &str, timestamp: u64) -> bool { + event["type"] == "hook.start" + && event["data"]["hookType"] == "agentStop" + && event["data"]["input"]["sessionId"] == sid + && event["data"]["input"]["timestamp"].as_u64() == Some(timestamp) + && event["data"]["input"]["stopReason"] == "end_turn" +} + +fn scan(bytes: &[u8], sid: &str, timestamp: u64) -> Option { + let mut anchored = false; + // An incomplete last record can be an ordinary concurrent journal append. + for line in bytes + .split_inclusive(|b| *b == b'\n') + .filter(|l| l.ends_with(b"\n")) + { + let event: Value = match serde_json::from_slice::(line) { + Ok(value) if value.is_object() && value["type"].is_string() => value, + _ if anchored => return Some(true), + _ => continue, + }; + if !anchored { + anchored = anchor(&event, sid, timestamp); + continue; + } + if matches!( + event["type"].as_str(), + Some("abort" | "user.message" | "session.error") + ) || (event["type"] == "hook.start" && event["data"]["hookType"] == "agentStop") + { + return Some(true); + } + } + anchored.then_some(false) +} + +/// Unverifiable/replaced/truncated journals release the waiter quietly. Returning +/// true is only a refusal to claim a message, never authority to send or read one. +pub(crate) fn cancelled(config: &CopilotConfig, sid: &str, payload: &Value) -> bool { + (|| { + let timestamp = payload["timestamp"].as_u64()?; + let bytes = bytes(config, sid, &payload["relayJournal"])?; + scan(&bytes, sid, timestamp) + })() + .unwrap_or(true) +} + +#[cfg(test)] +mod tests; diff --git a/crates/relay-node/src/relay/copilot/stop_wait/tests.rs b/crates/relay-node/src/relay/copilot/stop_wait/tests.rs new file mode 100644 index 000000000..820634e19 --- /dev/null +++ b/crates/relay-node/src/relay/copilot/stop_wait/tests.rs @@ -0,0 +1,262 @@ +use super::*; +use std::io::Write; +const SID: &str = "1069773c-0824-4fad-b7f9-d3dc3c4ed148"; + +fn start(timestamp: u64) -> Value { + json!({"type":"hook.start","data":{"hookInvocationId":format!("stop-{timestamp}"), + "hookType":"agentStop","input":{"sessionId":SID,"timestamp":timestamp,"stopReason":"end_turn"}}}) +} +fn append(path: &std::path::Path, value: Value) { + writeln!( + std::fs::OpenOptions::new().append(true).open(path).unwrap(), + "{value}" + ) + .unwrap(); +} +fn fixture() -> (tempfile::TempDir, CopilotConfig, PathBuf, Value) { + let home = tempfile::tempdir().unwrap(); + let config = CopilotConfig::from_env(&|_| None, home.path()); + let path = path(&config, SID).unwrap(); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write( + &path, + format!( + "{}\n{}\n{}\n", + start(100), + json!({"type":"abort"}), + start(200) + ), + ) + .unwrap(); + let payload = json!({"timestamp":200,"relayJournal":snapshot(&config,SID).unwrap()}); + (home, config, path, payload) +} + +#[test] +fn an_abort_after_this_stop_releases_the_wait_but_old_aborts_do_not() { + let (_home, config, path, payload) = fixture(); + assert!(!cancelled(&config, SID, &payload)); + append(&path, json!({"type":"abort"})); + assert!(cancelled(&config, SID, &payload)); + append(&path, start(300)); + let next = json!({"timestamp":300,"relayJournal":snapshot(&config,SID).unwrap()}); + assert!(!cancelled(&config, SID, &next)); + assert!(cancelled(&config, SID, &payload)); +} + +#[test] +fn a_human_turn_or_native_error_also_releases_the_wait() { + for kind in ["user.message", "session.error"] { + let (_home, config, path, payload) = fixture(); + append(&path, json!({"type":kind,"data":{}})); + assert!(cancelled(&config, SID, &payload)); + } +} + +#[test] +fn partial_appends_wait_for_the_complete_record_and_corruption_releases() { + let (_home, config, path, payload) = fixture(); + let mut file = std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap(); + file.write_all(b"{\"type\":\"abor").unwrap(); + assert!(!cancelled(&config, SID, &payload)); + file.write_all(b"t\"}\n").unwrap(); + assert!(cancelled(&config, SID, &payload)); + let (_home, config, path, payload) = fixture(); + std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap() + .write_all(b"invalid\n") + .unwrap(); + assert!(cancelled(&config, SID, &payload)); +} + +#[test] +fn non_object_or_untyped_complete_records_are_unverifiable_after_the_anchor() { + for value in [ + json!(null), + json!([]), + json!(true), + json!(42), + json!("bad"), + json!({}), + json!({"type":7}), + ] { + let (_home, config, path, payload) = fixture(); + append(&path, value); + assert!(cancelled(&config, SID, &payload)); + } +} + +#[test] +fn an_append_after_the_read_snapshot_is_not_a_cancellation() { + let (_home, config, path, _payload) = fixture(); + let history = format!("{}\n", json!({"type":"fixture","data":"x".repeat(4096)})); + std::fs::write(&path, format!("{}{}\n", history.repeat(20), start(200))).unwrap(); + let payload = json!({"timestamp":200,"relayJournal":snapshot(&config,SID).unwrap()}); + let mut file = open(&path).unwrap(); + let length = file.metadata().unwrap().len(); + assert!(length > LIMIT); + append(&path, json!({"type":"session.usage_record","data":{}})); + let bytes = read_tail_at(&mut file, length).unwrap(); + assert!(bytes.len() as u64 <= LIMIT); + assert_eq!(scan(&bytes, SID, 200), Some(false)); + assert!(!cancelled(&config, SID, &payload)); + append(&path, json!({"type":"abort"})); + assert!(cancelled(&config, SID, &payload)); +} + +#[test] +fn a_complete_anchor_at_the_tail_boundary_is_preserved() { + let (_home, config, path, _payload) = fixture(); + let anchor = format!("{}\n", start(200)); + let overhead = format!("{}\n", json!({"type":"fixture","data":""})).len(); + let filler = format!( + "{}\n", + json!({"type":"fixture","data":"x".repeat(LIMIT as usize-anchor.len()-overhead)}) + ); + let tail = format!("{anchor}{filler}"); + assert_eq!(tail.len() as u64, LIMIT); + let prefix = format!("{}\n", json!({"type":"fixture"})); + std::fs::write(&path, format!("{prefix}{tail}")).unwrap(); + let payload = json!({"timestamp":200,"relayJournal":snapshot(&config,SID).unwrap()}); + assert!(!cancelled(&config, SID, &payload)); +} + +#[test] +fn replacement_truncation_and_missing_anchors_never_claim_a_delivery() { + let (_home, config, path, payload) = fixture(); + let copy = path.with_extension("replacement"); + std::fs::copy(&path, ©).unwrap(); + std::fs::rename(copy, &path).unwrap(); + assert!(cancelled(&config, SID, &payload)); + let (_home, config, path, payload) = fixture(); + std::fs::write(&path, format!("{}\n", start(200))).unwrap(); + assert!(cancelled(&config, SID, &payload)); + let (_home, config, _path, mut payload) = fixture(); + payload["timestamp"] = json!(999); + assert!(cancelled(&config, SID, &payload)); + assert!(cancelled(&config, "../../outside", &payload)); +} + +#[test] +fn read_is_bounded_and_does_not_reuse_an_unreachable_anchor() { + let (_home, config, path, payload) = fixture(); + append( + &path, + json!({"type":"fixture","data":"x".repeat(LIMIT as usize)}), + ); + assert!(cancelled(&config, SID, &payload)); +} + +#[cfg(unix)] +#[test] +fn a_symlink_or_named_pipe_is_unavailable_without_waiting_for_io() { + use std::os::unix::fs::symlink; + let (_home, config, path, payload) = fixture(); + let moved = path.with_extension("original"); + std::fs::rename(&path, &moved).unwrap(); + symlink(&moved, &path).unwrap(); + assert!(snapshot(&config, SID).is_none()); + assert!(cancelled(&config, SID, &payload)); + std::fs::remove_file(&path).unwrap(); + nix::unistd::mkfifo( + &path, + nix::sys::stat::Mode::S_IRUSR | nix::sys::stat::Mode::S_IWUSR, + ) + .unwrap(); + assert!(snapshot(&config, SID).is_none()); + assert!(cancelled(&config, SID, &payload)); +} + +#[cfg(target_os = "macos")] +#[test] +fn cancellation_exits_idle_wait_and_leaves_an_offered_message_for_the_next_hook() { + use crate::relay::codex::turn_hook::Claimed; + use crate::relay::copilot::{hook_answer, proof, CopilotSettings}; + let (_home, config, table, _) = proof::tests::fixture(); + CopilotSettings::new(&config, "/opt/agent-relay-desktop-probe".into()) + .set_enabled(true) + .unwrap(); + { + let mut processes = table.processes.lock().unwrap(); + let hook = processes.get_mut(&103).unwrap(); + hook.executable = Some("/opt/agent-relay-desktop-probe".into()); + hook.arguments = [ + "/opt/agent-relay-desktop-probe", + "copilot-hook", + "--event", + "start", + ] + .into_iter() + .map(str::to_owned) + .collect(); + } + let mut payload = json!({"sessionId":SID,"relayEvent":"start","relayBinding":"binding"}); + assert_eq!( + hook_answer(&config, &table, 501, 103, &payload), + Some(json!({})) + ); + *table + .processes + .lock() + .unwrap() + .get_mut(&103) + .unwrap() + .arguments + .last_mut() + .unwrap() = "stop".into(); + let path = path(&config, SID).unwrap(); + std::fs::write(&path, format!("{}\n", start(200))).unwrap(); + payload["relayEvent"] = json!("stop"); + payload["stopReason"] = json!("end_turn"); + payload["timestamp"] = json!(200); + payload["relayJournal"] = snapshot(&config, SID).unwrap(); + assert_eq!( + hook_answer(&config, &table, 501, 103, &payload), + Some(json!({})) + ); + append(&path, json!({"type":"abort"})); + assert!(hook_answer(&config, &table, 501, 103, &payload).is_none()); + let thread = format!("{SID}:binding"); + config + .inbox + .offer(&thread, "cancelled-wait-message", "A real offered message") + .unwrap(); + assert!(hook_answer(&config, &table, 501, 103, &payload).is_none()); + assert!(config.inbox.has_offered(&thread)); + assert!(matches!( + config.inbox.claimed("cancelled-wait-message"), + Claimed::Offered + )); + append(&path, start(300)); + payload["timestamp"] = json!(300); + payload["relayJournal"] = snapshot(&config, SID).unwrap(); + let answer = hook_answer(&config, &table, 501, 103, &payload).unwrap(); + assert_eq!(answer["decision"], "block"); + assert!(answer["reason"] + .as_str() + .unwrap() + .contains("A real offered message")); + // Older native stop inputs still wait for a later offer. They cannot use + // timestamp-based cancellation, but must not lose the existing idle route. + payload.as_object_mut().unwrap().remove("timestamp"); + payload.as_object_mut().unwrap().remove("relayJournal"); + assert_eq!( + hook_answer(&config, &table, 501, 103, &payload), + Some(json!({})) + ); + config + .inbox + .offer(&thread, "legacy-stop-message", "A later legacy offer") + .unwrap(); + let legacy = hook_answer(&config, &table, 501, 103, &payload).unwrap(); + assert_eq!(legacy["decision"], "block"); + assert!(legacy["reason"] + .as_str() + .unwrap() + .contains("A later legacy offer")); +} diff --git a/packages/sdk-py/src/agent_relay/bin/agent-relay-broker b/packages/sdk-py/src/agent_relay/bin/agent-relay-broker new file mode 100755 index 000000000..e69de29bb