From 6cf845052cdc2c10e56f2de9f19d8e5aaf08bd76 Mon Sep 17 00:00:00 2001 From: Brent G Date: Sat, 17 Oct 2026 19:00:58 +0000 Subject: [PATCH] feat(runtime): log refused session-family requests to runtime.log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every inbound session-v1, terminal-v1 or chat-v1 request answered with an error (or a rejected receipt), plus oversize and full-inbox drops, gets one line: `refused /[ kind=] [ session=][ ()]`. Never content: ops outside the protocol vocabulary show as `?`, decode failures keep only the serde category and field name. At most 20 lines per 10 s per link, then one "โ€ฆ N more refusals suppressed" line. ๐Ÿค– Built with SMT --- AGENTS.md | 2 +- designs/protocol/atem-host-handoff.md | 5 + src/app.rs | 4 +- src/app/hosted.rs | 28 +- src/runtime/host_bridge.rs | 81 ++++- src/runtime/host_bridge_tests.rs | 156 ++++++++ src/runtime/mod.rs | 1 + src/runtime/refusal_log.rs | 496 ++++++++++++++++++++++++++ src/websocket_client.rs | 4 +- 9 files changed, 770 insertions(+), 7 deletions(-) create mode 100644 src/runtime/refusal_log.rs diff --git a/AGENTS.md b/AGENTS.md index 3b5e876..2ebc952 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -240,7 +240,7 @@ Key methods: **Runtime** (`src/runtime/`, `designs/agent-runtime.md`): `atem runtime` (hidden; one per config dir: an exclusive `flock` on `/runtime.lock` holding its pid, the socket `$XDG_RUNTIME_DIR/atem/runtime.sock` (else in the config dir) recorded in `/runtime.socket`, output in `/runtime.log`) is the broker: it serves local frontends over the same-user socket (local protocol v1, `runtime/protocol.rs`: `hello`, `launch`, `attach`, `detach`, `input`, `paste`, `resize`, `reclaim`, `list`, `stop`, `reload`, `astation_op`/`answer` in; `welcome`, `launched`, `attached`, `output`, `exited`, `sessions`, `detached`, `stopped`, `session`, `astation`, `notice`, `print`/`ask`/`done`, `error` out), holds the App with the machine's one Astation link, and owns the session host. Clients (`atem agent โ€ฆ`, bare `atem`, `atem pair`/`pair verify`/`cred unlock|projects`) start it detached when none answers; a runtime of another protocol without live sessions restarts for them. It exits after 10 minutes with no sessions and no non-passive clients (`ATEM_RUNTIME_IDLE_SECS` overrides), on SIGTERM/SIGINT, or when replaced. Every session runs in its own `atem worker --state /runtime --session ` (hidden, spawned detached; socket `$XDG_RUNTIME_DIR/atem/workers/.sock`, else `/runtime/workers/`; spec, record (`session_id`, `process_gen`, pids and OS start times, exit) and log in `/runtime/workers/`), which owns the agent's PTY, the canonical screen, Claude's chat adapter and the session log, and outlives the broker: a new broker adopts each surviving worker only when its `hello` names the record's session id and process generation and the pid's OS start time matches (Linux `/proc//stat` field 22, macOS `proc_pidinfo`); otherwise the session is `recovery_required` (or `interrupted`/`exited`) and nothing is restarted or replayed. The broker is the only writer authority: every write and paste carries the input epoch it granted, and the worker drops one whose epoch is older than its current one at the PTY write boundary (`pty_session.rs` fenced writes). Session ids are `ses-` + 12 base32 characters; each process is a new `process_gen` (UUID). -**Session host** (`runtime/host_bridge.rs`, `session_protocol/`, `app/hosted.rs`; contract `designs/protocol/atem-session-v1.md` r7.1, compatibility mode): only `atem runtime` serves sessions remotely. Its broker owns `SessionHost` over a `FileJournal` in `/runtime/journal/` (per-session receipt logs under `receipts/`, 0700/0600, `O_NOFOLLOW`, fsync per transition, admission halted after an fsync error, compaction, a lease-epoch floors file, `journal.lock`, and a `runtime_epoch` counter incremented at each start) and makes the App's authenticated Astation link one attachment โ€” new on every connect or re-authentication (`App::astation_link_gen`), fenced when the link drops or is re-authenticated, so clients `hello` again and an old `attachment_gen` is `stale_target`. A second `hello` on the same authenticated link (Astation reset its client, e.g. after sleep) replaces the attachment (`Effect::Replaced`): the old one is fenced (its leases end, streams and conversations close), the new one gets a new `attachment_gen`/`controller_session` with the link's unchanged authorization deadline; receipts stay; at most one a second (`busy`). runtime.log gets one `HH:MM:SS` UTC line per `hello` (welcome, replaced, refused ``). Only the paired Astation the link captured, verified in `cred_state.json`, gets one (`view`, `control`, `client_kind` `native`); any other link gets none and its session messages are answered `unauthorized`. `welcome` names the verified `data_account` (`account_id`), this device's verified `device_id` (`atem_device_id`) and `pairing.json`'s `host_trust_gen`; `runtime_gen`/`runtime_epoch` change when the broker restarts or the identity changes (a pairing verified after the broker started rebinds the host over the same journal). The attachment is authorized for 24 h (no freshness proof yet), on the monotonic clock and in wall time, whichever ends first: then it is fenced and the link dropped, and the reconnect authenticates again. Every 5 s the broker re-checks the link's pairing capture against `pairing.json` and drops a link whose pairing changed (unpair works even without `reload`). Session-v1 (`hello`, `list` โ†’ `sessions`/`session_changed` with a per-attachment `seq`, `acquire`/`renew`/`release` โ†’ 15 s leases with explicit `takeover`, `submit` โ†’ receipts, `receipt_query`) is the registry and the only way to get control. Receipts per kind: `prompt` is pasted once `ChatSession::submit` accepts it and completes at the chat's turn end with `detail` `done` or `interrupted` (a session without chat โ€” Codex with `[codex] hooks` off, Claude before its SessionStart or after its chat was withdrawn โ€” rejects it with no effect: "chat is not available for this agent; use the terminal"); `interrupt` writes Esc only while the chat status is `working` or `waiting` and completes when that turn ends, completes at once with nothing written in `ready` or `unknown`, and is `rejected` ("no turn state; use the terminal") whenever the session has no turn state, for any agent (Codex with its hooks off, Claude before its SessionStart or after its chat was withdrawn), never a blind Esc; `stop` completes at the exit; `approval` completes when the agent applied the decision (see Approval hooks). A refused-before-queueing write is `rejected`, a writer error after queueing `outcome_unknown`; executing work is `outcome_unknown` after a broker restart (never retried); duplicate submits return the recorded receipt, the same `request_id` with other content is `request_id_reused`. Terminal-v1 (`subscribe`/`unsubscribe`/`ack`, and `input`/`resize`/`stop` with the lease; terminal id = session id) and chat-v1 (`open`/`close`/`submit` with a binding, `submit` with the lease; `accepted`/`complete`/`error{code}` from its receipt; at most 8 conversations per session) are bound families. Effects reach the workers: remote bytes are checked with `admit_write` before they are sent and tagged with the worker's input epoch, which the broker raises (`set_input_epoch`, at least the lease epoch) whenever the host's epoch moves after remote bytes or while a lease was held; remote writes and pastes also carry `not_after`, the lease's deadline on CLOCK_MONOTONIC (`pty_session::monotonic_ns`), and the worker drops one at the PTY write boundary once it has passed (reported like an epoch drop; local writes carry none); each remote `input` gets a `written` ack (`control changed; not written` when dropped, `the approval was handed to the terminal; not written` when it released a Codex approval hook); a slow terminal transport drops frames and ends streams with `busy`. Local frontend input goes through `local_input` (it revokes any remote lease first). Duplicate session messages (same attachment, `id` and content) within 60 s are handled once (at most 4096 remembered). `advance` runs every 100 ms of the session clock. Adopted workers send a `chat_snapshot` before any `chat` (the whole retained history, 500 messages / 8 MiB, older messages first in `chat_history_part` lines of โ‰ค 1 MiB; approval cards included, since the broker copies each card to its worker with `chat_card`; a card still pending becomes `closed`), so chat goes on after a broker restart. A `chat` with `gap > 0` makes the broker reload the worker's history (`request_chat_history` โ†’ `chat_history`), then open conversations get `error` "the chat history was incomplete โ€ฆ; reopen the chat" and `close`. chat-v1 `open{before}` on an open conversation pages without `opened` or a reset (the page continues its live sequence) and ends with a sequenced `status` carrying `page: {before, count, start}`; cards are filtered for view-only attachments before the limits; a `before` not retained (or named with another `generation`) is `error{code:"stale_target"}`. The reload also carries the worker's chat state (`ChatSession::catch_up`: a missed turn end completes its receipt); a reload lost with its worker, or over 1000 messages, falls back to the same `error`/`close`. Bare `atem` opens no Astation connection, so only the runtime ever sees session-family messages. A runtime whose session host could not start (its journal or a damaged `runtime_epoch`, never reset silently) answers them `unsupported_version`, and `welcome.host_error` (shown by `atem agent list` and in `runtime.log`) says why, e.g. "remote sessions disabled: is damaged; move it aside to reset the runtime epoch". ยง9 features (resume, browser grants, account-wide access, relay confidentiality) are refused. A socket alive is not an authenticated Astation: when Astation sends `auth_required` over an already-authenticated socket (e.g. it alone reconnected to the relay), `app/reauth.rs` shows `Astation โœ—`, fences the attachment and ignores every other Astation request, then answers only the pinned Astation non-interactively (`AstationClient::answer_auth_required`) with the connect path's proof (`local_proof` on loopback, else the saved session); success restores online as a new attachment, while denial, expiry, a different `astation_id` or no answer within 15 s leaves a notice and closes the socket for the normal reconnect (never pairing). The shared fixtures run against the real runtime in `tests/runtime_fixtures_binary.rs` (results: `designs/protocol/atem-host-handoff.md`). +**Session host** (`runtime/host_bridge.rs`, `session_protocol/`, `app/hosted.rs`; contract `designs/protocol/atem-session-v1.md` r7.1, compatibility mode): only `atem runtime` serves sessions remotely. Its broker owns `SessionHost` over a `FileJournal` in `/runtime/journal/` (per-session receipt logs under `receipts/`, 0700/0600, `O_NOFOLLOW`, fsync per transition, admission halted after an fsync error, compaction, a lease-epoch floors file, `journal.lock`, and a `runtime_epoch` counter incremented at each start) and makes the App's authenticated Astation link one attachment โ€” new on every connect or re-authentication (`App::astation_link_gen`), fenced when the link drops or is re-authenticated, so clients `hello` again and an old `attachment_gen` is `stale_target`. A second `hello` on the same authenticated link (Astation reset its client, e.g. after sleep) replaces the attachment (`Effect::Replaced`): the old one is fenced (its leases end, streams and conversations close), the new one gets a new `attachment_gen`/`controller_session` with the link's unchanged authorization deadline; receipts stay; at most one a second (`busy`). runtime.log gets one `HH:MM:SS` UTC line per `hello` (welcome, replaced, refused ``), and one per other refused request (`runtime/refusal_log.rs`: `refused /[ kind=] [ session=][ ()]`, oversize and full-inbox drops included; decode failures keep only the serde category and field name, never content; at most 20 lines per 10 s per link, then "โ€ฆ N more refusals suppressed"). Only the paired Astation the link captured, verified in `cred_state.json`, gets one (`view`, `control`, `client_kind` `native`); any other link gets none and its session messages are answered `unauthorized`. `welcome` names the verified `data_account` (`account_id`), this device's verified `device_id` (`atem_device_id`) and `pairing.json`'s `host_trust_gen`; `runtime_gen`/`runtime_epoch` change when the broker restarts or the identity changes (a pairing verified after the broker started rebinds the host over the same journal). The attachment is authorized for 24 h (no freshness proof yet), on the monotonic clock and in wall time, whichever ends first: then it is fenced and the link dropped, and the reconnect authenticates again. Every 5 s the broker re-checks the link's pairing capture against `pairing.json` and drops a link whose pairing changed (unpair works even without `reload`). Session-v1 (`hello`, `list` โ†’ `sessions`/`session_changed` with a per-attachment `seq`, `acquire`/`renew`/`release` โ†’ 15 s leases with explicit `takeover`, `submit` โ†’ receipts, `receipt_query`) is the registry and the only way to get control. Receipts per kind: `prompt` is pasted once `ChatSession::submit` accepts it and completes at the chat's turn end with `detail` `done` or `interrupted` (a session without chat โ€” Codex with `[codex] hooks` off, Claude before its SessionStart or after its chat was withdrawn โ€” rejects it with no effect: "chat is not available for this agent; use the terminal"); `interrupt` writes Esc only while the chat status is `working` or `waiting` and completes when that turn ends, completes at once with nothing written in `ready` or `unknown`, and is `rejected` ("no turn state; use the terminal") whenever the session has no turn state, for any agent (Codex with its hooks off, Claude before its SessionStart or after its chat was withdrawn), never a blind Esc; `stop` completes at the exit; `approval` completes when the agent applied the decision (see Approval hooks). A refused-before-queueing write is `rejected`, a writer error after queueing `outcome_unknown`; executing work is `outcome_unknown` after a broker restart (never retried); duplicate submits return the recorded receipt, the same `request_id` with other content is `request_id_reused`. Terminal-v1 (`subscribe`/`unsubscribe`/`ack`, and `input`/`resize`/`stop` with the lease; terminal id = session id) and chat-v1 (`open`/`close`/`submit` with a binding, `submit` with the lease; `accepted`/`complete`/`error{code}` from its receipt; at most 8 conversations per session) are bound families. Effects reach the workers: remote bytes are checked with `admit_write` before they are sent and tagged with the worker's input epoch, which the broker raises (`set_input_epoch`, at least the lease epoch) whenever the host's epoch moves after remote bytes or while a lease was held; remote writes and pastes also carry `not_after`, the lease's deadline on CLOCK_MONOTONIC (`pty_session::monotonic_ns`), and the worker drops one at the PTY write boundary once it has passed (reported like an epoch drop; local writes carry none); each remote `input` gets a `written` ack (`control changed; not written` when dropped, `the approval was handed to the terminal; not written` when it released a Codex approval hook); a slow terminal transport drops frames and ends streams with `busy`. Local frontend input goes through `local_input` (it revokes any remote lease first). Duplicate session messages (same attachment, `id` and content) within 60 s are handled once (at most 4096 remembered). `advance` runs every 100 ms of the session clock. Adopted workers send a `chat_snapshot` before any `chat` (the whole retained history, 500 messages / 8 MiB, older messages first in `chat_history_part` lines of โ‰ค 1 MiB; approval cards included, since the broker copies each card to its worker with `chat_card`; a card still pending becomes `closed`), so chat goes on after a broker restart. A `chat` with `gap > 0` makes the broker reload the worker's history (`request_chat_history` โ†’ `chat_history`), then open conversations get `error` "the chat history was incomplete โ€ฆ; reopen the chat" and `close`. chat-v1 `open{before}` on an open conversation pages without `opened` or a reset (the page continues its live sequence) and ends with a sequenced `status` carrying `page: {before, count, start}`; cards are filtered for view-only attachments before the limits; a `before` not retained (or named with another `generation`) is `error{code:"stale_target"}`. The reload also carries the worker's chat state (`ChatSession::catch_up`: a missed turn end completes its receipt); a reload lost with its worker, or over 1000 messages, falls back to the same `error`/`close`. Bare `atem` opens no Astation connection, so only the runtime ever sees session-family messages. A runtime whose session host could not start (its journal or a damaged `runtime_epoch`, never reset silently) answers them `unsupported_version`, and `welcome.host_error` (shown by `atem agent list` and in `runtime.log`) says why, e.g. "remote sessions disabled: is damaged; move it aside to reset the runtime epoch". ยง9 features (resume, browser grants, account-wide access, relay confidentiality) are refused. A socket alive is not an authenticated Astation: when Astation sends `auth_required` over an already-authenticated socket (e.g. it alone reconnected to the relay), `app/reauth.rs` shows `Astation โœ—`, fences the attachment and ignores every other Astation request, then answers only the pinned Astation non-interactively (`AstationClient::answer_auth_required`) with the connect path's proof (`local_proof` on loopback, else the saved session); success restores online as a new attachment, while denial, expiry, a different `astation_id` or no answer within 15 s leaves a notice and closes the socket for the normal reconnect (never pairing). The shared fixtures run against the real runtime in `tests/runtime_fixtures_binary.rs` (results: `designs/protocol/atem-host-handoff.md`). **Astation operations through the runtime** (`runtime/ops.rs`, `runtime/op_client.rs`, `app/op_route.rs`): `atem pair` (first pairing), `atem pair verify`, `atem cred unlock`, `atem cred projects` and the paired-token refresh run inside the broker on its existing Astation link (waiting up to 15 s for it); the CLI is only the user interface (`print`/`ask` out, `answer` in, `done` at the end; answers from `/dev/tty`) and never opens its own connection (a runtime that can't be reached fails the command). One operation at a time (a second is refused); its handshake replies are routed to it only and dropped after it ends; a CLI that goes away ends it as a failure (device verification sends `verifyAbort`); a lost link ends it, never replayed. `atem unpair` and `atem logout` send `reload` to a running runtime (never starting one): it re-reads the config and drops a link whose pairing changed, with its attachments. diff --git a/designs/protocol/atem-host-handoff.md b/designs/protocol/atem-host-handoff.md index eee1b69..1f2e58a 100644 --- a/designs/protocol/atem-host-handoff.md +++ b/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 `. + 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 diff --git a/src/app.rs b/src/app.rs index 9d9c950..929c38b 100644 --- a/src/app.rs +++ b/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. diff --git a/src/app/hosted.rs b/src/app/hosted.rs index 4d715cd..4235648 100644 --- a/src/app/hosted.rs +++ b/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, /// 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, 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, diff --git a/src/runtime/host_bridge.rs b/src/runtime/host_bridge.rs index 097c36b..a8eeb62 100644 --- a/src/runtime/host_bridge.rs +++ b/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, /// 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)>, } /// 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) { + 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 { + 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 { use crate::session_protocol::wire::{ErrorCode, Op, SessionMessage}; diff --git a/src/runtime/host_bridge_tests.rs b/src/runtime/host_bridge_tests.rs index caa0bdc..54de636 100644 --- a/src/runtime/host_bridge_tests.rs +++ b/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 { + 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; + }); +} diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 51dc657..11d0390 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -12,6 +12,7 @@ pub mod op_client; pub mod ops; pub mod protocol; pub mod push_queue; +pub mod refusal_log; pub mod server; pub mod session_log; pub mod test_hooks; diff --git a/src/runtime/refusal_log.rs b/src/runtime/refusal_log.rs new file mode 100644 index 0000000..d726146 --- /dev/null +++ b/src/runtime/refusal_log.rs @@ -0,0 +1,496 @@ +//! One runtime.log line per refused session-family request (session-v1, +//! terminal-v1, chat-v1): its family, op, submit kind, error code and +//! session id, never its content. Decode failures keep only the error's +//! category and the field name (serde errors can echo values). Lines are +//! rate-limited per link: at most `MAX_LINES` per `WINDOW_MS`, then one +//! "โ€ฆ N more refusals suppressed" line. +use crate::session_protocol::wire::{ + self, DecodeError, Envelope, RULE_REVOKE_FIELD, ReceiptKind, is_session_id, +}; +use serde_json::Value; +use std::collections::HashMap; + +pub const MAX_LINES: u32 = 20; +pub const WINDOW_MS: u64 = 10_000; + +/// Op names a refusal line may show; anything else is `?` (an unknown op +/// is the peer's text, not protocol vocabulary). +const OPS: &[&str] = &[ + // session-v1 + "hello", + "challenge", + "resume_proof", + "welcome", + "list", + "sessions", + "session_changed", + "acquire", + "renew", + "release", + "lease", + "lease_released", + "lease_changed", + "submit", + "receipt", + "receipt_query", + "receipts", + "approval_request", + "approval_closed", + "approval_rules", + "error", + // terminal-v1 + "subscribe", + "unsubscribe", + "ack", + "input", + "resize", + "stop", + "control", + "controller", + "attached", + "frame", + "exited", + "written", + // chat-v1 + "open", + "close", + "opened", + "accepted", + "complete", + "message", + "status", + "history", + "heartbeat", + "gap", + "closed", +]; + +const FAMILIES: &[&str] = &["session", "terminal", "agentChat"]; + +/// What a refusal line says about the request. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Request { + pub family: String, + pub op: String, + pub kind: Option, + pub session: Option, +} + +fn known_op(op: Option<&str>) -> String { + op.filter(|op| OPS.contains(op)).unwrap_or("?").to_string() +} + +fn kind_name(kind: ReceiptKind) -> String { + serde_json::to_value(kind) + .ok() + .and_then(|v| v.as_str().map(str::to_string)) + .unwrap_or_default() +} + +fn session_in(v: &Value) -> Option { + [ + &v["binding"]["target"]["session_id"], + &v["terminal_id"], + &v["agentId"], + ] + .into_iter() + .find_map(|s| s.as_str().filter(|s| is_session_id(s)).map(str::to_string)) +} + +fn chat_kind(data: &Value) -> Option { + (data["op"] == "submit").then(|| { + if data.get(RULE_REVOKE_FIELD).is_some() { + kind_name(ReceiptKind::ApprovalRuleRevoke) + } else if data.get("approval").is_some() { + kind_name(ReceiptKind::Approval) + } else { + kind_name(ReceiptKind::Prompt) + } + }) +} + +impl Request { + /// The request as decoded, or (a decode failure) as far as `text` shows + /// it: a known family and op, a valid session id, a receipt kind. + pub fn of(decoded: &Result, text: &str) -> Self { + match decoded { + Ok(Envelope::Session(m)) => { + let v = serde_json::to_value(&m.op).unwrap_or(Value::Null); + Self { + family: "session".into(), + op: known_op(v["op"].as_str()), + kind: match &m.op { + wire::Op::Submit { kind, .. } => Some(kind_name(*kind)), + _ => None, + }, + session: session_in(&v), + } + } + Ok(Envelope::Family(f)) => { + let data = f.data.clone().unwrap_or(Value::Null); + let family = FAMILIES.iter().find(|k| **k == f.family).map_or("?", |k| k); + Self { + family: family.into(), + op: known_op(data["op"].as_str()), + kind: (family == "agentChat").then(|| chat_kind(&data)).flatten(), + session: f + .binding + .as_ref() + .map(|b| b.target.session_id.clone()) + .filter(|s| is_session_id(s)) + .or_else(|| session_in(&data)), + } + } + Err(_) => Self::peek(text), + } + } + + /// A message that failed to decode, from its raw text. + pub fn peek(text: &str) -> Self { + let root: Value = serde_json::from_str(text).unwrap_or(Value::Null); + let family = root["type"] + .as_str() + .and_then(|t| FAMILIES.iter().find(|k| **k == t)) + .map_or("?", |k| k); + let data = &root["data"]; + let kind = match family { + "session" if data["op"] == "submit" => { + serde_json::from_value::(data["kind"].clone()) + .ok() + .map(kind_name) + } + "agentChat" => chat_kind(data), + _ => None, + }; + Self { + family: family.into(), + op: known_op(data["op"].as_str()), + kind, + session: session_in(data), + } + } + + /// An oversize message: only its family is known. + pub fn family_only(family: &str) -> Self { + Self { + family: FAMILIES + .iter() + .find(|k| **k == family) + .map_or("?", |k| k) + .into(), + op: "?".into(), + kind: None, + session: None, + } + } +} + +/// The line's text after the time stamp. +pub fn line(request: &Request, code: &str, reason: Option<&str>) -> String { + let mut line = format!("atem runtime: refused {}/{}", request.family, request.op); + if let Some(kind) = &request.kind { + line.push_str(&format!(" kind={kind}")); + } + line.push_str(&format!(" {code}")); + if let Some(session) = &request.session { + line.push_str(&format!(" session={session}")); + } + if let Some(reason) = reason { + line.push_str(&format!(" ({reason})")); + } + line +} + +/// Why a message failed to decode, without any value it carried. +pub fn decode_reason(error: &DecodeError) -> String { + if error.removed.is_some() { + return "removed in atem-session r6".into(); + } + format!("decode: {}", sanitize(&error.reason)) +} + +/// A field name from a serde message (`` `name` ``), if it is one. +fn field_name(rest: &str) -> Option<&str> { + let name = rest.strip_prefix('`')?.split('`').next()?; + (!name.is_empty() + && name.len() <= 40 + && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')) + .then_some(name) +} + +/// Keeps the category of a decode error (and a field name), never a value. +pub fn sanitize(reason: &str) -> String { + // Contexts this crate prefixes (`binding: `, `agentChat: `, + // `approval_rule_revoke: `): names written in code. + let mut prefix = String::new(); + let mut rest = reason; + while let Some((context, r)) = rest.split_once(": ") + && !context.is_empty() + && context.len() <= 40 + && context + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '_') + { + prefix.push_str(context); + prefix.push_str(": "); + rest = r; + } + // Messages this crate writes itself (no peer text). + const FIXED: &[&str] = &[ + "unsupported v", + "missing or invalid id", + "invalid re", + "missing type", + "missing op", + "message is not an object", + "data must be an object", + "session data must be an object", + ]; + // `canonical::StrictError` variants (`Syntax(offset)`). + const STRICT: &[&str] = &[ + "TooLarge", + "InvalidUtf8", + "Syntax", + "DuplicateKey", + "LoneSurrogate", + "NonInteger", + "UnsafeInteger", + "TooDeep", + ]; + let core = 'core: { + for category in ["unknown field ", "missing field ", "duplicate field "] { + if let Some(after) = rest.strip_prefix(category) { + break 'core match field_name(after) { + Some(name) => format!("{category}`{name}`"), + None => category.trim_end().to_string(), + }; + } + } + for category in [ + "unknown variant", + "invalid type", + "invalid value", + "invalid length", + ] { + if rest.starts_with(category) { + break 'core category.to_string(); + } + } + if FIXED.contains(&rest) { + break 'core rest.to_string(); + } + let head = rest.split('(').next().unwrap_or_default(); + if STRICT.contains(&head) { + break 'core format!("json: {head}"); + } + // `invalid ` from the validators: a name written in code. + if let Some(what) = rest.strip_prefix("invalid ") + && what.len() <= 40 + && what + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | ' ' | '.')) + { + break 'core rest.to_string(); + } + if rest.ends_with("cannot carry a binding") { + break 'core "unexpected binding".to_string(); + } + "invalid".to_string() + }; + format!("{prefix}{core}") +} + +#[derive(Debug)] +struct Window { + start: u64, + lines: u32, + suppressed: u64, +} + +/// The rate limit, per link generation (one link's attachments share it, +/// so a client that says `hello` again can't escape it). +#[derive(Debug, Default)] +pub struct RefusalLog { + windows: HashMap, +} + +fn suppressed_line(n: u64) -> String { + format!("atem runtime: \u{2026} {n} more refusals suppressed") +} + +impl RefusalLog { + /// `line` for `link` at `now` (ms): the lines to write (none while + /// suppressed; the previous window's summary first after one). + pub fn record(&mut self, link: u64, now: u64, line: String) -> Vec { + let mut out = Vec::new(); + let window = self.windows.entry(link).or_insert(Window { + start: now, + lines: 0, + suppressed: 0, + }); + if now.saturating_sub(window.start) >= WINDOW_MS { + if window.suppressed > 0 { + out.push(suppressed_line(window.suppressed)); + } + *window = Window { + start: now, + lines: 0, + suppressed: 0, + }; + } + if window.lines < MAX_LINES { + window.lines += 1; + out.push(line); + } else { + window.suppressed += 1; + } + out + } + + /// Summaries of windows that ended with refusals suppressed; ended + /// windows are forgotten. + pub fn flush(&mut self, now: u64) -> Vec { + let mut out = Vec::new(); + self.windows.retain(|_, w| { + if now.saturating_sub(w.start) < WINDOW_MS { + return true; + } + if w.suppressed > 0 { + out.push(suppressed_line(w.suppressed)); + } + false + }); + out + } +} + +#[cfg(test)] +thread_local! { + /// Lines written on this thread (tests read them). + pub static WRITTEN: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; +} + +/// Writes `lines` to runtime.log (stderr) with the UTC `HH:MM:SS` stamp. +pub fn write(lines: Vec) { + if lines.is_empty() { + return; + } + let secs = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()) + % 86_400; + for line in lines { + eprintln!( + "{:02}:{:02}:{:02} {line}", + secs / 3600, + secs / 60 % 60, + secs % 60 + ); + #[cfg(test)] + WRITTEN.with(|w| w.borrow_mut().push(line)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn serde_values_never_survive_sanitizing() { + let cases = [ + ( + "unknown field `SECRET_x`, expected one of `a`", + "unknown field `SECRET_x`", + ), + ( + "missing field `rule_id` at line 1 column 2", + "missing field `rule_id`", + ), + ( + "invalid value: string \"SECRET\", expected a uuid", + "invalid value", + ), + ( + "invalid type: string \"SECRET\", expected u64", + "invalid type", + ), + ( + "unknown variant `SECRET`, expected `prompt`", + "unknown variant", + ), + ("unknown field `has space SECRET`", "unknown field"), + ( + "binding: missing field `target`", + "binding: missing field `target`", + ), + ( + "agentChat: invalid requestId", + "agentChat: invalid requestId", + ), + ( + "agentChat: approval_rule_revoke: unknown field `rule`, expected `rule_id`", + "agentChat: approval_rule_revoke: unknown field `rule`", + ), + ("invalid value: string \"a: b\"", "invalid value"), + ("Syntax(12)", "json: Syntax"), + ( + "terminal SECRET cannot carry a binding", + "unexpected binding", + ), + ("SECRET free text", "invalid"), + ("invalid \"SECRET\"", "invalid"), + ]; + for (reason, want) in cases { + let got = sanitize(reason); + assert_eq!(got, want, "{reason}"); + if !want.contains("SECRET") { + assert!(!got.contains("SECRET"), "{got}"); + } + } + } + + #[test] + fn at_most_twenty_lines_per_window_then_one_summary() { + let mut log = RefusalLog::default(); + let mut out = Vec::new(); + for i in 0..50 { + out.extend(log.record(1, i * 10, format!("l{i}"))); + } + assert_eq!(out.len(), MAX_LINES as usize); + // Another link has its own budget. + assert_eq!( + log.record(2, 100, "other".into()), + vec!["other".to_string()] + ); + // The next window starts with the summary. + let next = log.record(1, WINDOW_MS + 5, "again".into()); + assert_eq!(next, vec![suppressed_line(30), "again".to_string()]); + // A window that ends quietly is summarized by `flush`. + for i in 0..25 { + log.record(1, WINDOW_MS + 10 + i, "x".into()); + } + assert!(log.flush(WINDOW_MS + 100).is_empty()); + assert_eq!(log.flush(3 * WINDOW_MS), vec![suppressed_line(6)]); + assert!(log.flush(4 * WINDOW_MS).is_empty()); + } + + #[test] + fn a_line_names_family_op_kind_code_and_session() { + let request = Request { + family: "session".into(), + op: "submit".into(), + kind: Some("approval_rule_revoke".into()), + session: Some("ses-abcdefghijkl".into()), + }; + assert_eq!( + line(&request, "stale_target", Some("decode: invalid value")), + "atem runtime: refused session/submit kind=approval_rule_revoke stale_target \ + session=ses-abcdefghijkl (decode: invalid value)" + ); + let unknown = Request::peek(r#"{"type":"session","data":{"op":"SECRET"}}"#); + assert_eq!( + line(&unknown, "unsupported_version", None), + "atem runtime: refused session/? unsupported_version" + ); + } +} diff --git a/src/websocket_client.rs b/src/websocket_client.rs index c7e0907..60ffab9 100644 --- a/src/websocket_client.rs +++ b/src/websocket_client.rs @@ -141,7 +141,7 @@ pub enum AstationMessage { /// (not kept): answered with an `error` (`re` = its session `id`, when /// it had a readable one). #[serde(skip)] - SessionOversize { id: Option }, + SessionOversize { id: Option, family: String }, /// An op removed in atem-session r6 (`REMOVED_OPS`): answered with an /// `error` `unsupported_version` and no effect. @@ -903,7 +903,7 @@ impl AstationClient { Ok(Message::Text(text)) => { let astation_msg = match session_family(&text) { Some(family) if text.len() > if family == "terminal" { 32 * 1024 } else { crate::session_protocol::wire::MAX_MESSAGE_BYTES } => { - Some(AstationMessage::SessionOversize { id: session_message_id(&text) }) + Some(AstationMessage::SessionOversize { id: session_message_id(&text), family: family.to_string() }) } Some(_) => Some(AstationMessage::SessionWire(text)), None if is_removed_op(&text) => Some(AstationMessage::RemovedOp),