feat(runtime): log refused session, terminal and chat requests

designs/protocol/atem-host-handoff.md

@@ -101,6 +101,11 @@ Attachments and identity:
   `hello` meant the client already had an attachment, never that the device
   needs verifying). runtime.log has one `HH:MM:SS` UTC line per `hello`:
   `welcome`, `replaced attachment … with …; welcome`, or `refused <code>`.
+  Every other refused request (an `error` reply, a `rejected` receipt, an
+  oversize drop) also gets one line with its family, op, code and session,
+  never its content: `refused agentChat/submit kind=approval_rule_revoke
+  unsupported_version session=ses-… (decode: agentChat: approval_rule_revoke:
+  unknown field `rule`)`; at most 20 per 10 s per link.
 - `welcome`: `account_id` = the verified `data_account` from
   `cred_state.json`; `atem_device_id` = this device's verified `device_id`;
   `host_trust_gen` = `pairing.json`'s generation (moves on every

src/app.rs

@@ -660,8 +660,8 @@ impl App {
             AstationMessage::SessionWire(text) => {
                 self.session_wire(text);
             }
-            AstationMessage::SessionOversize { id } => {
-                self.session_refusal(id, crate::session_protocol::wire::ErrorCode::UnsupportedVersion, None);
+            AstationMessage::SessionOversize { id, family } => {
+                self.session_oversize(id, &family);
             }
             AstationMessage::RemovedOp => {
                 // r6 §0: refused, with no effect.

src/app/hosted.rs

@@ -8,6 +8,7 @@
 //! The TUI without a runtime (bare `atem`) opens no Astation connection at
 //! all, so it never sees a session-family message.
 use super::App;
+use crate::runtime::refusal_log;
 use crate::session::{Provider, SessionId};
 use crate::session_protocol::wire::{self, Envelope, ErrorCode, Op, SessionMessage};
 use std::collections::VecDeque;
@@ -38,6 +39,9 @@ pub struct Hosted {
     pub sessions: Vec<HostedSession>,
     /// Mark-task prompts to write: (task id, session, text).
     pub writes: Vec<(String, SessionId, String)>,
+    /// Refusals answered here (oversize, a full inbox): (link generation,
+    /// runtime.log line), written by the broker under its rate limit.
+    pub refused: Vec<(u64, String)>,
 }
 
 impl Hosted {
@@ -77,7 +81,13 @@ impl App {
             hosted.inbox.push_back((link, text));
             return;
         }
-        let re = match wire::decode(text.as_bytes()) {
+        let decoded = wire::decode(text.as_bytes());
+        let request = refusal_log::Request::of(&decoded, &text);
+        let line = refusal_log::line(&request, "busy", Some("inbox full"));
+        if hosted.refused.len() < INBOX_LIMIT {
+            hosted.refused.push((link, line));
+        }
+        let re = match decoded {
             Ok(Envelope::Session(m)) => Some(m.id),
             Ok(Envelope::Family(_)) => None,
             Err(e) => e.id,
@@ -86,6 +96,22 @@ impl App {
         self.session_refusal(re, ErrorCode::Busy, None);
     }
 
+    /// A session-family message over the size bound: dropped unread and
+    /// refused (`unsupported_version`), with a runtime.log line.
+    pub(super) fn session_oversize(&mut self, id: Option<String>, family: &str) {
+        if self.astation_connected && !self.astation_client.is_ws_closed() {
+            let link = self.astation_link_gen;
+            if let Some(hosted) = self.hosted.as_mut()
+                && hosted.refused.len() < INBOX_LIMIT
+            {
+                let request = refusal_log::Request::family_only(family);
+                let line = refusal_log::line(&request, "unsupported_version", Some("oversize"));
+                hosted.refused.push((link, line));
+            }
+        }
+        self.session_refusal(id, ErrorCode::UnsupportedVersion, None);
+    }
+
     /// A session `error` with `code` (answering `re`) on an authenticated link.
     pub(super) fn session_refusal(
         &mut self,

src/runtime/host_bridge.rs

@@ -64,6 +64,7 @@
 //! trust (`confirm_approval_rules`; the worker drops the rest, and allows by
 //! none until then), and a rebind or trust change drops them all.
 use super::broker::{BrokerSession, SessionState as BrokerState, WriteOutcome, WriteTag};
+use super::refusal_log::{self, RefusalLog, Request};
 use super::worker_approvals::Event as ApprovalEvent;
 use super::worker_protocol::{Activity as WorkerActivity, ApplyOutcome, HeldRule, RuleGrant};
 use crate::app::App;
@@ -389,6 +390,11 @@ pub struct HostBridge {
     revokes: HashMap<String, (String, String)>,
     /// Push requests for the relay (`push_queue`).
     pushes: super::push_queue::PushQueue,
+    /// runtime.log lines for refused requests, rate-limited per link.
+    refusal_log: RefusalLog,
+    /// The request being handled (its link and what a refusal line says
+    /// about it): error replies while it is set are logged.
+    refusing: Option<(u64, Request, Option<String>)>,
 }
 
 /// Where effects go: the link's client and the broker's sessions.
@@ -437,6 +443,8 @@ impl HostBridge {
             approvals: HashMap::new(),
             revokes: HashMap::new(),
             pushes: Default::default(),
+            refusal_log: RefusalLog::default(),
+            refusing: None,
         })
     }
 
@@ -593,6 +601,7 @@ impl HostBridge {
             }
         }
         self.worker_reports(&mut cx);
+        self.app_refusals(&mut app.hosted);
         let now = self.clock.now();
         if now >= self.last_advance + ADVANCE_EVERY {
             self.last_advance = now;
@@ -804,6 +813,10 @@ impl HostBridge {
                 let _ = cx.client.try_send_wire(unauthorized(re), false);
                 if hello.is_some() {
                     log_handshake("refused unauthorized (no attachment on this link)");
+                } else {
+                    let request = Request::of(&decoded, text);
+                    let line = refusal_log::line(&request, "unauthorized", Some("no attachment"));
+                    self.log_refusal(link_gen, line);
                 }
             }
             return;
@@ -820,11 +833,46 @@ impl HostBridge {
                 return;
             }
         }
+        // `hello` outcomes have their own line (`log_hello`).
+        if hello.is_none() {
+            let reason = decoded.as_ref().err().map(refusal_log::decode_reason);
+            self.refusing = Some((link_gen, Request::of(&decoded, text), reason));
+        }
         let effects = self.host.receive_decoded(&attachment, decoded, now);
         if let Some(id) = &hello {
             log_hello(id, &effects);
         }
         self.exec(effects, cx);
+        self.refusing = None;
+    }
+
+    /// Writes `line` unless the link's refusal lines are suppressed.
+    fn log_refusal(&mut self, link_gen: u64, line: String) {
+        let now = self.clock.now();
+        refusal_log::write(self.refusal_log.record(link_gen, now, line));
+    }
+
+    /// An error reply (`code`) to the request being handled: logged once
+    /// per reply, never with its content.
+    fn refused(&mut self, code: &str) {
+        let Some((link_gen, request, reason)) = self.refusing.clone() else {
+            return;
+        };
+        let line = refusal_log::line(&request, code, reason.as_deref());
+        self.log_refusal(link_gen, line);
+    }
+
+    /// Refusals the App answered itself (oversize, a full inbox).
+    fn app_refusals(&mut self, hosted: &mut Option<crate::app::hosted::Hosted>) {
+        let refused = hosted
+            .as_mut()
+            .map(|h| std::mem::take(&mut h.refused))
+            .unwrap_or_default();
+        for (link_gen, line) in refused {
+            self.log_refusal(link_gen, line);
+        }
+        let now = self.clock.now();
+        refusal_log::write(self.refusal_log.flush(now));
     }
 
     /// The same session message (attachment, `id`, content) was handled
@@ -859,6 +907,11 @@ impl HostBridge {
         if !self.current(attachment) {
             return;
         }
+        if self.refusing.is_some()
+            && let Some(code) = error_code(message)
+        {
+            self.refused(&code);
+        }
         let text = String::from_utf8(wire::encode(message)).expect("wire JSON is UTF-8");
         let terminal = matches!(message, Envelope::Family(f) if f.family == "terminal");
         if terminal {
@@ -879,7 +932,12 @@ impl HostBridge {
         }
     }
 
-    fn send_chat(&self, env: ChatEnvelope, cx: &mut Cx) {
+    fn send_chat(&mut self, env: ChatEnvelope, cx: &mut Cx) {
+        if env.op == "error"
+            && let Some(code) = env.body.get("code").and_then(Value::as_str)
+        {
+            self.refused(code);
+        }
         let text = serde_json::to_string(&AstationMessage::AgentChat(env))
             .expect("chat envelopes serialize");
         let _ = cx.client.try_send_wire(text, false);
@@ -2022,6 +2080,7 @@ impl HostBridge {
         message: &str,
         cx: &mut Cx,
     ) {
+        self.refused(code);
         let mut data = json!({
             "v": 1, "op": "error", "terminal_id": sid, "code": code, "message": message,
         });
@@ -2660,6 +2719,26 @@ pub(super) fn rule_revoked_line(sid: &str, rule_id: &str, ok: bool) -> String {
     }
 }
 
+/// The code of an error reply: a session `error`, a terminal-v1 `error`, or
+/// a `rejected` receipt.
+fn error_code(message: &Envelope) -> Option<String> {
+    use crate::session_protocol::wire::{Op, ReceiptState};
+    match message {
+        Envelope::Session(m) => match &m.op {
+            Op::Error { code, .. } => serde_json::to_value(code)
+                .ok()
+                .and_then(|v| v.as_str().map(str::to_string)),
+            Op::Receipt(r) if r.state == ReceiptState::Rejected => Some("rejected".into()),
+            _ => None,
+        },
+        Envelope::Family(f) if f.family == "terminal" => {
+            let data = f.data.as_ref()?;
+            (data["op"] == "error").then(|| data["code"].as_str().unwrap_or("?").to_string())
+        }
+        Envelope::Family(_) => None,
+    }
+}
+
 /// A session `error{code:"unauthorized"}` for a link without an attachment.
 fn unauthorized(re: Option<String>) -> String {
     use crate::session_protocol::wire::{ErrorCode, Op, SessionMessage};

src/runtime/host_bridge_tests.rs

@@ -3400,3 +3400,159 @@ async fn host_rebinds_drop_the_workers_session_rules() {
         let _ = session.stop();
     }
 }
+
+// ── Refusal lines in runtime.log ──
+
+const MARKER: &str = "SECRET-MARKER-7f3a";
+
+/// The refusal lines written on this thread so far.
+fn refusal_lines() -> Vec<String> {
+    super::refusal_log::WRITTEN.with(|w| w.borrow().clone())
+}
+
+/// Waits until a refusal line `pick` accepts was written; that line.
+async fn refusal_line(pick: impl Fn(&str) -> bool) -> String {
+    tokio::time::timeout(WAIT, async {
+        loop {
+            if let Some(line) = refusal_lines().into_iter().find(|l| pick(l)) {
+                return line;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("no such refusal line; written: {:#?}", refusal_lines()))
+}
+
+#[test]
+fn refused_requests_are_logged_by_family_op_and_code_without_their_content() {
+    local(async {
+        super::refusal_log::WRITTEN.with(|w| w.borrow_mut().clear());
+        let mut rt = Runtime::start(true).await;
+        let mut client = rt.client().await;
+        let agent = echo_agent(rt.dir.path());
+        let session = rt.launch(&mut client, "codex", &agent, None).await;
+        let (sid, pg) = (session.id.clone(), session.generation.clone());
+        let peer = &mut rt.peer;
+        peer.hello().await;
+
+        // A chat-v1 revoke of the wrong shape: a decode failure whose serde
+        // error names the field (and would echo values).
+        peer.chat(
+            "submit",
+            "conv-1",
+            &sid,
+            &pg,
+            json!({"generation": pg, "requestId": uuid::Uuid::new_v4().to_string(),
+                   "approval_rule_revoke": {"rule": MARKER}}),
+        );
+        let line = refusal_line(|l| l.contains("refused agentChat/submit")).await;
+        assert_eq!(
+            line,
+            format!(
+                "atem runtime: refused agentChat/submit kind=approval_rule_revoke \
+                 unsupported_version session={sid} \
+                 (decode: agentChat: approval_rule_revoke: unknown field `rule`)"
+            )
+        );
+
+        // A session-v1 submit whose kind is a value serde would echo.
+        let id = peer.request(
+            "submit",
+            json!({"binding": peer.binding(&sid, &pg, None),
+                   "request_id": uuid::Uuid::new_v4().to_string(),
+                   "kind": MARKER, "payload": {"text": MARKER}}),
+        );
+        assert_eq!(peer.reply(&id).await["code"], "unsupported_version");
+        let line = refusal_line(|l| l.contains("refused session/submit")).await;
+        assert_eq!(
+            line,
+            format!(
+                "atem runtime: refused session/submit unsupported_version session={sid} \
+                 (decode: unknown variant)"
+            )
+        );
+
+        // A session-v1 revoke whose payload has the wrong shape.
+        let id = peer.request(
+            "submit",
+            json!({"binding": peer.binding(&sid, &pg, None),
+                   "request_id": uuid::Uuid::new_v4().to_string(),
+                   "kind": "approval_rule_revoke", "payload": {"rule": MARKER}}),
+        );
+        let reply = peer.reply(&id).await;
+        let code = match reply["op"].as_str() {
+            Some("error") => reply["code"].as_str().unwrap().to_string(),
+            _ => reply["state"].as_str().unwrap().to_string(),
+        };
+        let line = refusal_line(|l| {
+            l.contains("kind=approval_rule_revoke") && l.contains("session/submit")
+        })
+        .await;
+        assert!(
+            line.starts_with(&format!(
+                "atem runtime: refused session/submit kind=approval_rule_revoke {code} session={sid}"
+            )),
+            "{line}"
+        );
+
+        // A terminal-v1 subscribe to another process generation.
+        let other = uuid::Uuid::new_v4().to_string();
+        peer.terminal("subscribe", &sid, &other, json!({}), None);
+        let line = refusal_line(|l| l.contains("refused terminal/subscribe")).await;
+        assert_eq!(
+            line,
+            format!("atem runtime: refused terminal/subscribe stale_target session={sid}")
+        );
+
+        for line in refusal_lines() {
+            assert!(!line.contains(MARKER), "{line}");
+            assert!(!line.contains('{'), "{line}");
+        }
+        rt.shutdown().await;
+    });
+}
+
+#[test]
+fn a_link_without_an_attachment_logs_its_refusals() {
+    local(async {
+        super::refusal_log::WRITTEN.with(|w| w.borrow_mut().clear());
+        let mut rt = Runtime::start(false).await;
+        let peer = &mut rt.peer;
+        let id = uuid::Uuid::new_v4().to_string();
+        peer.send(json!({"type":"session","data":{"v":1,"op":"list","id":id,"x":MARKER}}));
+        assert_eq!(peer.reply(&id).await["code"], "unauthorized");
+        let line = refusal_line(|l| l.contains("refused session/")).await;
+        assert_eq!(
+            line,
+            "atem runtime: refused session/list unauthorized (no attachment)"
+        );
+        assert!(refusal_lines().iter().all(|l| !l.contains(MARKER)));
+        rt.shutdown().await;
+    });
+}
+
+#[test]
+fn a_looping_client_gets_twenty_refusal_lines_per_window_and_a_summary() {
+    local(async {
+        super::refusal_log::WRITTEN.with(|w| w.borrow_mut().clear());
+        let mut rt = Runtime::start(true).await;
+        let peer = &mut rt.peer;
+        peer.hello().await;
+        let mut last = String::new();
+        for _ in 0..35 {
+            last = uuid::Uuid::new_v4().to_string();
+            peer.send(json!({"type":"session","data":{"v":2,"op":"list","id":last,"x":MARKER}}));
+        }
+        peer.reply(&last).await;
+        let refused = |lines: &[String]| lines.iter().filter(|l| l.contains("refused ")).count();
+        assert_eq!(refused(&refusal_lines()), 20, "{:#?}", refusal_lines());
+        assert!(!refusal_lines().iter().any(|l| l.contains("suppressed")));
+        // The window ends: one summary line for the other 15.
+        rt.clock.skip(super::refusal_log::WINDOW_MS);
+        let line = refusal_line(|l| l.contains("suppressed")).await;
+        assert_eq!(line, "atem runtime: \u{2026} 15 more refusals suppressed");
+        assert_eq!(refused(&refusal_lines()), 20);
+        rt.shutdown().await;
+    });
+}