Release Copilot idle hooks after native cancellation

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.

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<A> {
     stop_reason: Option<String>,
     tool_name: Option<String>,
     tool_args: Option<A>,
+    timestamp: Option<u64>,
 }
 impl<A> Hook<A> {
     fn fields(&self) -> serde_json::Map<String, Value> {
@@ -87,6 +98,9 @@ impl<A> Hook<A> {
                 );
             }
         }
+        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":{}})
+        ));
     }
 }

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;

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,

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<PathBuf> {
+    crate::relay::store::is_uuid(sid).then(|| config.sessions().join(sid).join("events.jsonl"))
+}
+
+#[cfg(unix)]
+fn identity(metadata: &Metadata) -> Option<Value> {
+    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<Value> {
+    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<Value> {
+    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<File> {
+    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<Vec<u8>> {
+    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<Vec<u8>> {
+    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<bool> {
+    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::<Value>(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;