From e13c37055777bc25b7b45eb496a6c7e04ff49c5e Mon Sep 17 00:00:00 2001 From: Rasmus Widing Date: Tue, 18 Aug 2026 19:42:32 +0300 Subject: [PATCH 1/2] =?UTF-8?q?feat(daemon):=20mail=20=E2=80=94=20the=20ma?= =?UTF-8?q?ilroom,=20the=20wake=20reactor,=20and=20the=20real-agent=20proo?= =?UTF-8?q?f?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit benchd owns agent-to-agent mail. Three new verbs over the same socket and log: mail/send delivers a front-matter markdown file into mail//inbox (files are the record; a mailbox is created on first delivery, so mail to a handle nobody hosts simply waits and the response says so honestly); mail/list is metadata only with caps reported; mail/read is body plus retirement — inbox moves to read/, nothing is ever deleted. The wake is a reactor, not a mailbox feature: mail/sent to a live session's handle queues a wake, and the reactor pastes 'You have mail from : ' — the retired path, never the body — into the recipient's pty once it has been idle past the gate. The loop cap is a per-recipient token bucket (burst 6, one per minute) in the courier, the only place that knows a wake happened because it caused it (helm #320); a capped wake logs wake/capped and the mail waits safely unread. Sessions claim handles at spawn (--name, default the session id): validated, unique, and 'operator' reserved — addressable by anyone, claimable by no session. The daemon declares BENCH_SESSION, BENCH_HANDLE and BENCH_DIR into every pty, so 'bench mail send' inside an agent needs no flags and lands in the right mailroom. Proven with real agents, first run green (just mail-proof): 100 seeded as mail to a claude, +1 per hop through codex and pi via the production reactor and mailroom, 103 collected back — event trail mail/sent → agent/woken ×4 → mail/read. Conformance grows to 27 tests: the wake observed end to end through the relay (path present, body absent), retirement at delivery, the cap starving wakes but never mail, and the handle reservations. Also completes #341's R3 honestly: that PR typed the daemon's payloads but the CLI half of the patch never applied — the disposition comment overclaimed. The CLI now builds every payload from bench-wire's own types. Gate green: 27 conformance + 4 mail + 6 session + 9 wire tests. --- daemon/AGENTS.md | 2 + daemon/Cargo.lock | 5 + daemon/Cargo.toml | 8 +- daemon/crates/bench-mail/Cargo.toml | 5 + daemon/crates/bench-mail/src/lib.rs | 247 +++++++++++++++ daemon/crates/bench-session/src/lib.rs | 20 ++ daemon/crates/bench-wire/src/lib.rs | 82 ++++- daemon/crates/bench/src/main.rs | 152 ++++++++-- daemon/crates/bench/tests/conformance.rs | 291 +++++++++++++++++- daemon/crates/benchd/Cargo.toml | 1 + daemon/crates/benchd/src/main.rs | 364 ++++++++++++++++++++++- daemon/direction.md | 9 +- daemon/justfile | 71 ++++- daemon/spikes/mail-milestone-proof.md | 22 ++ 14 files changed, 1221 insertions(+), 58 deletions(-) create mode 100644 daemon/crates/bench-mail/Cargo.toml create mode 100644 daemon/crates/bench-mail/src/lib.rs create mode 100644 daemon/spikes/mail-milestone-proof.md diff --git a/daemon/AGENTS.md b/daemon/AGENTS.md index 3a6653b..fd9de51 100644 --- a/daemon/AGENTS.md +++ b/daemon/AGENTS.md @@ -19,6 +19,8 @@ knows nor needs the Rust toolchain, in either direction. - `crates/bench-wire` — every wire type and shared resolution rule, spelled once. If `benchd` and `bench` could disagree about a value, its rule belongs here. +- `crates/bench-mail` — the mailroom: delivery, retirement, listings; no sessions, no + sockets, no wakes — the reactor in `benchd` owns those. - `crates/bench-session` — the pty core: agent allowlist, postures/model/effort/resume argv (one spelling, unit-tested), the ring, the attach relay, drain-then-die close. - `crates/benchd` — the daemon. Foreground, one unix socket, a thread per connection. diff --git a/daemon/Cargo.lock b/daemon/Cargo.lock index b5937b6..507d692 100644 --- a/daemon/Cargo.lock +++ b/daemon/Cargo.lock @@ -16,6 +16,10 @@ dependencies = [ "serde_json", ] +[[package]] +name = "bench-mail" +version = "0.0.1" + [[package]] name = "bench-session" version = "0.0.1" @@ -35,6 +39,7 @@ dependencies = [ name = "benchd" version = "0.0.1" dependencies = [ + "bench-mail", "bench-session", "bench-wire", "serde_json", diff --git a/daemon/Cargo.toml b/daemon/Cargo.toml index c72d63c..f32f88a 100644 --- a/daemon/Cargo.toml +++ b/daemon/Cargo.toml @@ -2,7 +2,13 @@ # `cargo` at the repo root should fail loudly, not half-work (bench-roadmap.md, M0). [workspace] resolver = "2" -members = ["crates/bench-wire", "crates/bench-session", "crates/benchd", "crates/bench"] +members = [ + "crates/bench-wire", + "crates/bench-session", + "crates/bench-mail", + "crates/benchd", + "crates/bench", +] [workspace.package] version = "0.0.1" diff --git a/daemon/crates/bench-mail/Cargo.toml b/daemon/crates/bench-mail/Cargo.toml new file mode 100644 index 0000000..232066e --- /dev/null +++ b/daemon/crates/bench-mail/Cargo.toml @@ -0,0 +1,5 @@ +[package] +name = "bench-mail" +version.workspace = true +edition.workspace = true +description = "The mailroom: files are the record, notices point, retirement never deletes." diff --git a/daemon/crates/bench-mail/src/lib.rs b/daemon/crates/bench-mail/src/lib.rs new file mode 100644 index 0000000..44be37e --- /dev/null +++ b/daemon/crates/bench-mail/src/lib.rs @@ -0,0 +1,247 @@ +//! The mailroom — mail is mail, and nothing else (the operator's decomposition). +//! +//! This crate knows directories and files; it knows nothing about sessions, wakes, +//! sockets, or ptys — the wake is a *reactor* in the daemon, answering `mail/sent` +//! events with a paste, and the loop cap lives there too. The rules carried here are +//! helm's, bought with incidents: +//! +//! - **Files are the record.** A message is a markdown file a plain `cat` reads; the +//! socket is transport, never the only copy. +//! - **The notice carries the path, never the body.** Whatever wakes a recipient says +//! where the mail is; the mail itself costs tokens only when the agent chooses to +//! read it, once, at the moment it matters. +//! - **Retire, never delete.** Reading moves inbox → read. Nothing in the mailroom +//! ever unlinks a message. +//! - **Pull is metadata-only.** A listing returns sender/subject/time/read-state — +//! bodies never — and reports its caps. +//! +//! Layout, under the record root: +//! ```text +//! mail//inbox/.md unread +//! mail//read/.md retired +//! ``` +//! A message file is front-matter plus body, cat-friendly: +//! ```text +//! --- +//! from: post-claude +//! at: 2026-08-18T14:00:00Z +//! subject: the codeword +//! --- +//! +//! ``` + +use std::fs; +use std::path::{Path, PathBuf}; + +/// One line of a listing: everything a triage needs, nothing a body costs. +#[derive(Debug, Clone)] +pub struct MailMeta { + pub id: String, + pub from: String, + pub subject: Option, + pub at: String, + pub unread: bool, +} + +pub fn mail_root(root: &Path) -> PathBuf { + root.join("mail") +} + +fn inbox(root: &Path, handle: &str) -> PathBuf { + mail_root(root).join(handle).join("inbox") +} + +fn read_dir_of(root: &Path, handle: &str) -> PathBuf { + mail_root(root).join(handle).join("read") +} + +/// Deliver a message into a handle's inbox. The mailbox is created on first delivery — +/// a claim is a directory, and mail to a handle nobody has spawned yet simply waits. +/// Returns `(id, path)`; the path is what a notice may carry. +pub fn deliver( + root: &Path, + seq: u64, + from: &str, + to: &str, + subject: Option<&str>, + at_rfc3339: &str, + body: &str, +) -> Result<(String, PathBuf), String> { + let dir = inbox(root, to); + fs::create_dir_all(&dir).map_err(|e| format!("cannot create mailbox for {to:?}: {e}"))?; + let id = format!("m{seq}"); + let path = dir.join(format!("{id}.md")); + let subject_line = subject + .map(|s| format!("subject: {s}\n")) + .unwrap_or_default(); + let content = format!("---\nfrom: {from}\nat: {at_rfc3339}\n{subject_line}---\n{body}\n"); + fs::write(&path, content).map_err(|e| format!("cannot write {}: {e}", path.display()))?; + Ok((id, path)) +} + +/// Retire a message: inbox → read, never delete. Returns the retired path. Retiring an +/// already-retired message is fine and answers with where it lives. +pub fn retire(root: &Path, handle: &str, id: &str) -> Result { + let name = format!("{id}.md"); + let from_path = inbox(root, handle).join(&name); + let to_dir = read_dir_of(root, handle); + let to_path = to_dir.join(&name); + if to_path.exists() { + return Ok(to_path); + } + if !from_path.exists() { + return Err(format!("no message {id:?} in {handle:?}'s mailbox")); + } + fs::create_dir_all(&to_dir).map_err(|e| format!("cannot create read dir: {e}"))?; + fs::rename(&from_path, &to_path).map_err(|e| format!("cannot retire {id}: {e}"))?; + Ok(to_path) +} + +pub fn read_body(path: &Path) -> Result { + fs::read_to_string(path).map_err(|e| format!("cannot read {}: {e}", path.display())) +} + +/// List a mailbox, metadata only, unread first then retired, each side sorted by id. +/// The caller reports any cap it applies; this returns everything. +pub fn list(root: &Path, handle: &str) -> Vec { + let mut out = Vec::new(); + for (dir, unread) in [ + (inbox(root, handle), true), + (read_dir_of(root, handle), false), + ] { + let mut entries: Vec = fs::read_dir(&dir) + .map(|rd| { + rd.filter_map(|e| e.ok().map(|e| e.path())) + .filter(|p| p.extension().is_some_and(|x| x == "md")) + .collect() + }) + .unwrap_or_default(); + entries.sort(); + for path in entries { + let id = path + .file_stem() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_default(); + let head = fs::read_to_string(&path).unwrap_or_default(); + let meta = parse_front_matter(&head); + out.push(MailMeta { + id, + from: meta.0, + subject: meta.2, + at: meta.1, + unread, + }); + } + } + out +} + +/// (from, at, subject) out of the front-matter block. Absent fields come back empty — +/// a listing must render whatever is on disk, not refuse a file a human hand-wrote. +fn parse_front_matter(text: &str) -> (String, String, Option) { + let mut from = String::new(); + let mut at = String::new(); + let mut subject = None; + let mut in_block = false; + for line in text.lines().take(10) { + if line.trim() == "---" { + if in_block { + break; + } + in_block = true; + continue; + } + if !in_block { + continue; + } + if let Some(v) = line.strip_prefix("from: ") { + from = v.trim().to_string(); + } else if let Some(v) = line.strip_prefix("at: ") { + at = v.trim().to_string(); + } else if let Some(v) = line.strip_prefix("subject: ") { + subject = Some(v.trim().to_string()); + } + } + (from, at, subject) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn root() -> PathBuf { + let dir = std::env::temp_dir().join(format!( + "bmail-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .subsec_nanos() + )); + fs::create_dir_all(&dir).unwrap(); + dir + } + + #[test] + fn delivery_creates_the_mailbox_and_the_file_is_cat_friendly() { + let r = root(); + let (id, path) = deliver( + &r, + 7, + "post-claude", + "post-codex", + Some("codeword"), + "2026-08-18T00:00:00Z", + "101", + ) + .unwrap(); + assert_eq!(id, "m7"); + let text = fs::read_to_string(&path).unwrap(); + assert!(text.contains("from: post-claude")); + assert!(text.contains("subject: codeword")); + assert!(text.ends_with("101\n")); + let _ = fs::remove_dir_all(r); + } + + #[test] + fn retire_moves_and_never_deletes_and_is_idempotent() { + let r = root(); + let (id, path) = deliver(&r, 1, "a", "b", None, "t", "hello").unwrap(); + let retired = retire(&r, "b", &id).unwrap(); + assert!(!path.exists(), "inbox copy moved"); + assert!(retired.exists(), "read copy exists — nothing deleted"); + assert_eq!( + retire(&r, "b", &id).unwrap(), + retired, + "retiring twice answers with where it lives" + ); + assert!(retire(&r, "b", "m99").is_err(), "unknown id refuses"); + let _ = fs::remove_dir_all(r); + } + + #[test] + fn listing_is_metadata_only_unread_first() { + let r = root(); + let (id1, _) = deliver(&r, 1, "x", "b", Some("one"), "t1", "SECRET-BODY").unwrap(); + let (_id2, _) = deliver(&r, 2, "y", "b", None, "t2", "another").unwrap(); + retire(&r, "b", &id1).unwrap(); + let listing = list(&r, "b"); + assert_eq!(listing.len(), 2); + assert!(listing[0].unread && listing[0].from == "y"); + assert!(!listing[1].unread && listing[1].subject.as_deref() == Some("one")); + let _ = fs::remove_dir_all(r); + } + + #[test] + fn a_hand_written_file_still_lists() { + let r = root(); + let dir = mail_root(&r).join("b").join("inbox"); + fs::create_dir_all(&dir).unwrap(); + fs::write(dir.join("note.md"), "no front matter at all").unwrap(); + let listing = list(&r, "b"); + assert_eq!(listing.len(), 1); + assert_eq!(listing[0].id, "note"); + assert!(listing[0].from.is_empty()); + let _ = fs::remove_dir_all(r); + } +} diff --git a/daemon/crates/bench-session/src/lib.rs b/daemon/crates/bench-session/src/lib.rs index 44b02c5..bb4ca89 100644 --- a/daemon/crates/bench-session/src/lib.rs +++ b/daemon/crates/bench-session/src/lib.rs @@ -239,6 +239,10 @@ impl Ring { pub struct Session { pub id: String, + /// The mailbox address and human name for this session — `--name` at spawn, else + /// the session id. Uniqueness and the `operator` reservation are the daemon's to + /// enforce; this crate just carries the decided value. + pub handle: String, pub spec: SpawnSpec, pub agent: AgentKind, pub cwd: String, @@ -265,9 +269,11 @@ impl Session { /// exits and forced detaches reach the daemon's log. pub fn spawn( id: String, + handle: String, spec: &SpawnSpec, rows: u16, cols: u16, + extra_env: &[(String, String)], notices: Sender, ) -> Result, String> { let (program, args) = argv(spec)?; @@ -286,6 +292,12 @@ impl Session { } cmd.cwd(&spec.cwd); cmd.env("TERM", "xterm-256color"); + // The session learns its own address and root — what lets an agent inside run + // `bench mail send` with no flags and land in the right mailroom (the same + // declare-don't-derive rule as helm's PaneEnvironment). + for (k, v) in extra_env { + cmd.env(k, v); + } let child = pair .slave .spawn_command(cmd) @@ -303,6 +315,7 @@ impl Session { let session = Arc::new(Session { pid: child.process_id(), + handle, spec: spec.clone(), id: id.clone(), agent: spec.agent, @@ -377,6 +390,13 @@ impl Session { self.ring.lock().unwrap().total } + /// How long the pty has been quiet — the crude idle gate the mail spike proved + /// sufficient for wake delivery. The taps milestone replaces judgement, not + /// plumbing. + pub fn idle_for(&self) -> Duration { + self.ring.lock().unwrap().last_change.elapsed() + } + /// Paste, then submit separately — the launch-line rule, spelled once. pub fn deliver_line(&self, line: &str) -> Result<(), String> { let mut w = self.writer.lock().unwrap(); diff --git a/daemon/crates/bench-wire/src/lib.rs b/daemon/crates/bench-wire/src/lib.rs index 3b9566d..7ec225b 100644 --- a/daemon/crates/bench-wire/src/lib.rs +++ b/daemon/crates/bench-wire/src/lib.rs @@ -123,7 +123,17 @@ impl RequestId { /// compiler forces a verdict when a verb is added, and the justfile's probe list is /// pinned to it by a conformance test that reads the justfile's own source. pub const KNOWN_VERBS: &[&str] = &[ - "status", "events", "stop", "spawn", "sessions", "attach", "close", "resume", + "status", + "events", + "stop", + "spawn", + "sessions", + "attach", + "close", + "resume", + "mail/send", + "mail/list", + "mail/read", ]; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -136,6 +146,9 @@ pub enum Verb { Attach, Close, Resume, + MailSend, + MailList, + MailRead, } impl Verb { @@ -150,6 +163,9 @@ impl Verb { "attach" => Some(Verb::Attach), "close" => Some(Verb::Close), "resume" => Some(Verb::Resume), + "mail/send" => Some(Verb::MailSend), + "mail/list" => Some(Verb::MailList), + "mail/read" => Some(Verb::MailRead), _ => None, } } @@ -176,6 +192,64 @@ pub const READY_WAIT: std::time::Duration = std::time::Duration::from_secs(10); pub const CLIENT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(READY_WAIT.as_secs() + DAEMON_IO_TIMEOUT.as_secs() + 5); +// --------------------------------------------------------------------------- +// Handles and mail payloads +// --------------------------------------------------------------------------- + +/// A mailbox handle. Same shape rule as suites — it decides a directory, so a path can +/// never be one — plus one reservation ported from helm's mailbox verbatim: +/// **`operator` is the operator's**, addressable by anyone, claimable by no session. +pub const OPERATOR_HANDLE: &str = "operator"; + +pub fn validate_handle(raw: &str) -> Result<(), String> { + if raw.is_empty() { + return Err("a handle cannot be empty".into()); + } + if raw.contains('/') || raw.contains('\\') || raw.contains("..") { + return Err(format!("a path is not a handle: {raw:?}")); + } + if raw.len() > 32 + || !raw + .chars() + .next() + .is_some_and(|c| c.is_ascii_lowercase() || c.is_ascii_digit()) + || !raw + .chars() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-') + { + return Err(format!( + "a handle is lowercase ASCII letters, digits and '-', starting alphanumeric, max 32: {raw:?}" + )); + } + Ok(()) +} + +/// `mail/send`'s payload. The body travels IN the request; the notice a recipient gets +/// carries only the path (helm's rule: notice carries path, never body). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MailSendArgs { + pub to: String, + pub from: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subject: Option, + pub body: String, +} + +/// `mail/list`'s payload: whose mailbox. Metadata only comes back — sender, subject, +/// time, read-state — bodies never; pull is on demand, push is minimal. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MailListArgs { + pub handle: String, +} + +/// `mail/read`'s payload: retire-never-delete — reading moves inbox → read, and the +/// response names the new path. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MailReadArgs { + pub handle: String, + pub id: String, +} + // --------------------------------------------------------------------------- // Session verb payloads // --------------------------------------------------------------------------- @@ -188,6 +262,10 @@ pub const CLIENT_READ_TIMEOUT: std::time::Duration = pub struct SpawnArgs { pub agent: String, pub cwd: String, + /// The mailbox address and tab name — defaults to the session id. `operator` is + /// refused: that handle is the operator's, addressable, never claimable. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub name: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub prompt_file: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -417,7 +495,7 @@ mod tests { } assert_eq!( KNOWN_VERBS.len(), - 8, + 11, "a new verb joins KNOWN_VERBS and this count together" ); assert!(Verb::parse("frobnicate").is_none()); diff --git a/daemon/crates/bench/src/main.rs b/daemon/crates/bench/src/main.rs index 77fbf72..0729b9e 100644 --- a/daemon/crates/bench/src/main.rs +++ b/daemon/crates/bench/src/main.rs @@ -13,7 +13,8 @@ //! These are helm's spool codes, kept on purpose. use bench_wire::{ - CLIENT_READ_TIMEOUT, DAEMON_IO_TIMEOUT, EXIT_NO_DAEMON, Request, RequestId, Response, Status, + CLIENT_READ_TIMEOUT, DAEMON_IO_TIMEOUT, EXIT_NO_DAEMON, MailListArgs, MailReadArgs, + MailSendArgs, OPERATOR_HANDLE, Request, RequestId, Response, SessionArgs, SpawnArgs, Status, SuiteName, resolve_root, socket_path, }; use serde_json::{Value, json}; @@ -33,11 +34,15 @@ fn usage() -> &'static str { \x20 events [--since N] read the record back from seq N\n\ \x20 stop log the stop, kill sessions, exit\n\ \x20 spawn --agent --cwd spawn an agent into a bench pty\n\ - \x20 [--prompt-file

] [--model ] [--effort ]\n\ + \x20 [--name ] [--prompt-file

] [--model ] [--effort ]\n\ \x20 sessions list bench sessions\n\ \x20 attach raw relay to a session's pty (Ctrl-\\ detaches)\n\ \x20 close drain-then-die the session\n\ \x20 resume re-enter an exited session's runtime state\n\ + \x20 mail send --to --body deliver mail; a live recipient is woken\n\ + \x20 [--body-file

] [--subject ] [--from ]\n\ + \x20 mail list [--handle ] metadata only, unread first\n\ + \x20 mail read [--handle ] body + retirement (inbox -> read)\n\ env: BENCH_SUITE (flag wins) · BENCH_DIR (root override, wins over suite)\n\ exit: 0 ok · 2 no daemon · 3 refused · 4 daemon failed" } @@ -67,7 +72,8 @@ fn run() -> i32 { None => return refuse("--since needs a sequence number"), }, "--agent" | "--cwd" | "--prompt-file" | "--model" | "--effort" | "--rows" - | "--cols" => { + | "--cols" | "--name" | "--to" | "--from" | "--subject" | "--body" | "--body-file" + | "--handle" => { let key = arg.trim_start_matches("--").replace('-', "_"); match argv.next() { Some(v) => flags.push((key, v)), @@ -86,9 +92,15 @@ fn run() -> i32 { } } - let Some(verb) = verb else { + let Some(mut verb) = verb else { return refuse(usage()); }; + if verb == "mail" { + if positional.is_empty() { + return refuse("mail needs a subcommand: send, list, read"); + } + verb = format!("mail/{}", positional.remove(0)); + } // Suite validated before any socket is touched — a name that cannot isolate must // never resolve to the shared root by accident (#86). One spelling, two edges. @@ -107,46 +119,120 @@ fn run() -> i32 { let bench_dir = std::env::var("BENCH_DIR").ok(); let root = resolve_root(bench_dir.as_deref(), suite.as_ref(), &home); - let mut args = serde_json::Map::new(); - match verb.as_str() { - "events" if since > 0 => { - args.insert("since".into(), json!(since)); - } + let flag = |name: &str| -> Option { + flags + .iter() + .find(|(k, _)| k == name) + .map(|(_, v)| v.clone()) + }; + // The default identity: the session's own declared handle, else the operator — the + // same declare-don't-derive rule as the daemon setting BENCH_HANDLE at spawn. + let own_handle = + || std::env::var("BENCH_HANDLE").unwrap_or_else(|_| OPERATOR_HANDLE.to_string()); + + // Payloads are the wire crate's own types (#341 R3, completed here — the daemon was + // typed in that PR, the CLI half was not): the CLI cannot spell a key the daemon + // does not read. + let args: Value = match verb.as_str() { + "events" if since > 0 => json!({ "since": since }), "spawn" => { - for (k, v) in &flags { - let value: Value = match k.as_str() { - "rows" | "cols" => match v.parse::() { - Ok(n) => json!(n), + let mut spawn = SpawnArgs { + agent: String::new(), + cwd: String::new(), + name: flag("name"), + prompt_file: flag("prompt_file"), + model: flag("model"), + effort: flag("effort"), + rows: None, + cols: None, + }; + if let Some(a) = flag("agent") { + spawn.agent = a; + } else { + return refuse("spawn needs --agent "); + } + if let Some(c) = flag("cwd") { + spawn.cwd = c; + } else { + return refuse("spawn needs --cwd "); + } + for k in ["rows", "cols"] { + if let Some(v) = flag(k) { + match v.parse::() { + Ok(n) => { + if k == "rows" { + spawn.rows = Some(n) + } else { + spawn.cols = Some(n) + } + } Err(_) => return refuse(&format!("--{k} needs a number")), - }, - _ => json!(v), - }; - args.insert(k.clone(), value); + } + } } + json!(spawn) } - "attach" | "close" | "resume" => match positional.first() { - Some(s) => { - args.insert("session".into(), json!(s)); - } - None => { + "attach" | "close" | "resume" => { + let Some(sid) = positional.first() else { return refuse(&format!( "{verb} needs a session id — `bench sessions` lists them" )); - } - }, - _ => {} - } - if verb == "attach" { - // Tell the daemon the viewer's size so the pty matches before replay. - if let Some((rows, cols)) = terminal_size() { - args.insert("rows".into(), json!(rows)); - args.insert("cols".into(), json!(cols)); + }; + let (rows, cols) = if verb == "attach" { + // Tell the daemon the viewer's size so the pty matches before replay. + match terminal_size() { + Some((r, c)) => (Some(r), Some(c)), + None => (None, None), + } + } else { + (None, None) + }; + json!(SessionArgs { + session: sid.clone(), + rows, + cols, + }) } - } + "mail/send" => { + let Some(to) = flag("to") else { + return refuse("mail send needs --to "); + }; + let body = match (flag("body"), flag("body_file")) { + (Some(b), None) => b, + (None, Some(p)) => match std::fs::read_to_string(&p) { + Ok(b) => b, + Err(e) => return refuse(&format!("cannot read --body-file {p:?}: {e}")), + }, + (Some(_), Some(_)) => { + return refuse("--body and --body-file are one or the other"); + } + (None, None) => return refuse("mail send needs --body or --body-file

"), + }; + json!(MailSendArgs { + to, + from: flag("from").unwrap_or_else(own_handle), + subject: flag("subject"), + body, + }) + } + "mail/list" => json!(MailListArgs { + handle: flag("handle").unwrap_or_else(own_handle), + }), + "mail/read" => { + let Some(id) = positional.first() else { + return refuse("mail read needs a message id — `bench mail list` shows them"); + }; + json!(MailReadArgs { + handle: flag("handle").unwrap_or_else(own_handle), + id: id.clone(), + }) + } + _ => Value::Null, + }; let cli = Cli { verb: verb.clone(), - args: Value::Object(args), + args, root, }; if verb == "attach" { diff --git a/daemon/crates/bench/tests/conformance.rs b/daemon/crates/bench/tests/conformance.rs index 96cfa7e..39d2572 100644 --- a/daemon/crates/bench/tests/conformance.rs +++ b/daemon/crates/bench/tests/conformance.rs @@ -456,12 +456,20 @@ fn the_justfile_probes_every_known_verb() { // read the literal out of the source and compare (R5). let justfile = fs::read_to_string(Path::new(env!("CARGO_MANIFEST_DIR")).join("../../justfile")) .expect("daemon/justfile readable from the bench crate"); - let probe_line: String = justfile - .lines() - .skip_while(|l| !l.contains("for verb in")) - .take_while(|l| !l.contains("; do")) - .collect::>() - .join(" "); + let mut probe_lines: Vec<&str> = Vec::new(); + let mut in_probe = false; + for line in justfile.lines() { + if line.contains("for verb in") { + in_probe = true; + } + if in_probe { + probe_lines.push(line); + if line.contains("; do") { + break; + } + } + } + let probe_line = probe_lines.join(" "); for verb in bench_wire::KNOWN_VERBS { assert!( probe_line.contains(verb), @@ -840,3 +848,274 @@ fn libc_kill(pid: i32) { fn libc_alive(pid: i32) -> bool { unsafe { kill(pid, 0) == 0 } } + +// --------------------------------------------------------------------------- +// Mail: the mailroom, the notice discipline, the wake reactor, the cap +// --------------------------------------------------------------------------- + +#[test] +fn mail_to_a_handle_nobody_hosts_waits_in_the_record() { + let home = TestHome::claim("ghostmail"); + let _daemon = DaemonGuard::start(&home.dir, None); + let send = bench( + &home.dir, + &[ + "mail", + "send", + "--to", + "ghost", + "--body", + "hello there", + "--subject", + "hi", + ], + ); + assert_eq!(send.code, 0, "stderr: {}", send.stderr); + let sent: serde_json::Value = serde_json::from_str(&send.stdout).unwrap(); + assert_eq!( + sent["wake"], "no-live-session", + "honest: nothing will wake a ghost" + ); + let id = sent["id"].as_str().unwrap().to_string(); + + // Pull is metadata-only: the listing must never carry the body. + let list = bench(&home.dir, &["mail", "list", "--handle", "ghost"]); + assert_eq!(list.code, 0); + assert!(list.stdout.contains("\"unread\": true")); + assert!( + !list.stdout.contains("hello there"), + "bodies never ride a listing" + ); + + // Read = body + retirement; retire-never-delete. + let read = bench(&home.dir, &["mail", "read", &id, "--handle", "ghost"]); + assert_eq!(read.code, 0, "stderr: {}", read.stderr); + assert!(read.stdout.contains("hello there")); + assert!( + read.stdout.contains("/read/"), + "the response names where it lives now" + ); + let mailbox = home.dir.join(".bench/mail/ghost"); + assert!(mailbox.join("read").join(format!("{id}.md")).exists()); + assert_eq!( + fs::read_dir(mailbox.join("inbox")).unwrap().count(), + 0, + "moved, not copied — and never deleted" + ); + + let relist = bench(&home.dir, &["mail", "list", "--handle", "ghost"]); + assert!(relist.stdout.contains("\"unread\": false")); +} + +#[test] +fn a_send_to_a_live_session_wakes_it_with_a_path_never_the_body() { + let home = TestHome::claim("wake"); + let daemon = DaemonGuard::start(&home.dir, None); + let spawn = bench( + &home.dir, + &[ + "spawn", + "--agent", + "test-echo", + "--cwd", + "/tmp", + "--name", + "echo1", + ], + ); + assert_eq!(spawn.code, 0, "stderr: {}", spawn.stderr); + + let send = bench( + &home.dir, + &[ + "mail", + "send", + "--to", + "echo1", + "--body", + "SECRET-BODY-99", + "--subject", + "ping", + ], + ); + assert_eq!(send.code, 0, "stderr: {}", send.stderr); + let sent: serde_json::Value = serde_json::from_str(&send.stdout).unwrap(); + assert_eq!(sent["wake"], "queued"); + let id = sent["id"].as_str().unwrap().to_string(); + + // The reactor pastes the notice into the pty; cat echoes it into the ring, which an + // attach replays — the production wake path observed end to end. + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let log = fs::read_to_string(home.dir.join(".bench/events.jsonl")).unwrap_or_default(); + if log.contains("agent/woken") { + break; + } + assert!(Instant::now() < deadline, "the wake never happened: {log}"); + std::thread::sleep(Duration::from_millis(200)); + } + let (resp, stream) = raw_request( + &daemon.socket, + "attach", + serde_json::json!({"session": "s1"}), + ); + assert_eq!(resp["status"], "ok"); + let _ = stream.set_read_timeout(Some(Duration::from_millis(300))); + let mut seen = Vec::new(); + let mut chunk = [0u8; 4096]; + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + match (&stream).read(&mut chunk) { + Ok(0) => break, + Ok(n) => seen.extend_from_slice(&chunk[..n]), + Err(_) => {} + } + if String::from_utf8_lossy(&seen).contains("You have mail") { + break; + } + } + let text = String::from_utf8_lossy(&seen); + assert!( + text.contains("You have mail from operator"), + "the notice names the sender: {text}" + ); + assert!( + text.contains("/read/"), + "the notice carries the retired path: {text}" + ); + assert!( + !text.contains("SECRET-BODY-99"), + "the notice must NEVER carry the body: {text}" + ); + + // Retired at delivery: the notice's path is where the file already lives. + let mailbox = home.dir.join(".bench/mail/echo1"); + assert!(mailbox.join("read").join(format!("{id}.md")).exists()); +} + +#[test] +fn the_wake_cap_starves_wakes_never_mail() { + let home = TestHome::claim("cap"); + let _daemon = DaemonGuard::start(&home.dir, None); + let spawn = bench( + &home.dir, + &[ + "spawn", + "--agent", + "test-echo", + "--cwd", + "/tmp", + "--name", + "echo2", + ], + ); + assert_eq!(spawn.code, 0, "stderr: {}", spawn.stderr); + for i in 0..9 { + let send = bench( + &home.dir, + &[ + "mail", + "send", + "--to", + "echo2", + "--body", + &format!("msg {i}"), + ], + ); + assert_eq!(send.code, 0, "send {i} failed: {}", send.stderr); + } + // Six tokens of burst; the seventh-plus wake must be capped and say so. + let deadline = Instant::now() + Duration::from_secs(60); + let (mut woken, mut capped); + loop { + let log = fs::read_to_string(home.dir.join(".bench/events.jsonl")).unwrap_or_default(); + woken = log.matches("agent/woken").count(); + capped = log.matches("wake/capped").count(); + if capped >= 1 && woken >= 6 { + break; + } + assert!( + Instant::now() < deadline, + "cap never engaged: woken={woken} capped={capped}" + ); + std::thread::sleep(Duration::from_millis(300)); + } + assert!( + woken <= 6, + "the burst budget is six; {woken} wakes happened" + ); + // The starved mail is safe in the inbox, unread — the cap brakes wakes, never mail. + let listing = bench(&home.dir, &["mail", "list", "--handle", "echo2"]); + assert!( + listing.stdout.contains("\"unread\": true"), + "capped mail waits unread: {}", + listing.stdout + ); +} + +#[test] +fn the_operator_handle_is_addressable_but_never_claimable() { + let home = TestHome::claim("oper"); + let _daemon = DaemonGuard::start(&home.dir, None); + let claim = bench( + &home.dir, + &[ + "spawn", + "--agent", + "test-echo", + "--cwd", + "/tmp", + "--name", + "operator", + ], + ); + assert_eq!(claim.code, 3, "stderr: {}", claim.stderr); + assert!( + claim.stderr.contains("operator"), + "the refusal names the reservation: {}", + claim.stderr + ); + let send = bench( + &home.dir, + &["mail", "send", "--to", "operator", "--body", "for you"], + ); + assert_eq!(send.code, 0, "addressable: {}", send.stderr); +} + +#[test] +fn a_claimed_handle_refuses_a_second_claim_and_a_path_is_not_a_handle() { + let home = TestHome::claim("dupes"); + let _daemon = DaemonGuard::start(&home.dir, None); + let a = bench( + &home.dir, + &[ + "spawn", + "--agent", + "test-echo", + "--cwd", + "/tmp", + "--name", + "worker", + ], + ); + assert_eq!(a.code, 0); + let b = bench( + &home.dir, + &[ + "spawn", + "--agent", + "test-echo", + "--cwd", + "/tmp", + "--name", + "worker", + ], + ); + assert_eq!(b.code, 3, "stderr: {}", b.stderr); + assert!(b.stderr.contains("already claimed")); + let c = bench( + &home.dir, + &["mail", "send", "--to", "../escape", "--body", "x"], + ); + assert_eq!(c.code, 3, "stderr: {}", c.stderr); +} diff --git a/daemon/crates/benchd/Cargo.toml b/daemon/crates/benchd/Cargo.toml index 1b22a80..d785641 100644 --- a/daemon/crates/benchd/Cargo.toml +++ b/daemon/crates/benchd/Cargo.toml @@ -7,5 +7,6 @@ description = "The bench daemon: owns the record, answers on the socket." [dependencies] bench-wire = { path = "../bench-wire" } bench-session = { path = "../bench-session" } +bench-mail = { path = "../bench-mail" } serde_json.workspace = true time.workspace = true diff --git a/daemon/crates/benchd/src/main.rs b/daemon/crates/benchd/src/main.rs index e35d89f..d6262b9 100644 --- a/daemon/crates/benchd/src/main.rs +++ b/daemon/crates/benchd/src/main.rs @@ -24,8 +24,9 @@ use bench_session::{AgentKind, Notice, Session, SpawnSpec, TEST_AGENT_ENV, mint_session_id}; use bench_wire::{ DAEMON_IO_TIMEOUT, EVENTS_LOG_FORMAT, EVENTS_LOG_VERSION, Event, KNOWN_VERBS, - MAX_REQUEST_BYTES, READY_WAIT, Request, Response, SessionArgs, SpawnArgs, Status, SuiteName, - Verb, check_socket_path, events_path, resolve_root, socket_path, + MAX_REQUEST_BYTES, MailListArgs, MailReadArgs, MailSendArgs, OPERATOR_HANDLE, READY_WAIT, + Request, Response, SessionArgs, SpawnArgs, Status, SuiteName, Verb, check_socket_path, + events_path, resolve_root, socket_path, validate_handle, }; use serde_json::{Value, json}; use std::collections::HashMap; @@ -113,6 +114,29 @@ struct RepairNote { dropped_bytes: usize, } +/// Mail waiting to wake its recipient. The reactor drains this — mail/sent events in, +/// pastes out — and the loop cap lives HERE, in the courier, because only the thing +/// that causes a wake can count wakes (helm #320's measurement: a hook cannot). +struct PendingWake { + handle: String, + mail_id: String, + from: String, + capped_logged: bool, +} + +/// A token bucket per recipient: burst of WAKE_BURST, refilling one per minute. A +/// two-agent ping-pong self-throttles instead of burning until the money runs out. +struct WakeBucket { + tokens: f64, + last: Instant, +} + +const WAKE_BURST: f64 = 6.0; +const WAKE_REFILL_PER_SEC: f64 = 1.0 / 60.0; +/// The idle gate: the pty must have been quiet this long before a paste. The crude +/// form the mail spike proved; taps refine the judgement later, not the plumbing. +const WAKE_IDLE_GATE: Duration = Duration::from_secs(2); + /// Read the log with byte offsets. A clean log returns the next seq. An unreadable /// line refuses — unless it is the LAST non-empty line, which is an interrupted append: /// quarantine the tail, truncate back to the last good byte, and report the repair (R1). @@ -190,6 +214,9 @@ struct Core { booted: Instant, sessions: HashMap>, next_session: u64, + next_mail: u64, + pending_wakes: Vec, + wake_tokens: HashMap, notices: mpsc::Sender, } @@ -268,6 +295,9 @@ fn boot(root: PathBuf, suite: Option) -> Result { booted: Instant::now(), sessions: HashMap::new(), next_session: 1, + next_mail: 1, + pending_wakes: Vec::new(), + wake_tokens: HashMap::new(), notices: notice_tx, })); @@ -328,6 +358,12 @@ fn boot(root: PathBuf, suite: Option) -> Result { }); } + // The wake reactor: mail/sent facts become pastes into idle recipient ptys. + { + let core = Arc::clone(&core); + std::thread::spawn(move || wake_reactor(core)); + } + for stream in listener.incoming() { let Ok(stream) = stream else { continue }; let core = Arc::clone(&core); @@ -336,6 +372,120 @@ fn boot(root: PathBuf, suite: Option) -> Result { Ok(0) } +/// Answer `mail/sent` with `agent/woken` — the composition the mail spike proved, as a +/// reactor over daemon state. Per pending wake: recipient must be a live session, its +/// pty quiet past the idle gate, and its token bucket willing; then the message is +/// retired (the notice carries the path it will KEEP), the notice is pasted and +/// submitted, and the wake is logged. A capped wake logs once and waits for refill — +/// the mail itself sits safely in the mailbox either way. +fn wake_reactor(core: Arc>) { + loop { + std::thread::sleep(Duration::from_millis(400)); + // Snapshot under the lock; judge and paste outside it. + let candidates: Vec<(String, String, String, Arc)> = { + let c = core.lock().unwrap(); + c.pending_wakes + .iter() + .filter_map(|p| { + c.sessions + .values() + .find(|s| s.handle == p.handle && s.is_live()) + .map(|s| { + ( + p.handle.clone(), + p.mail_id.clone(), + p.from.clone(), + Arc::clone(s), + ) + }) + }) + .collect() + }; + // Drop pendings whose recipient session is gone for good. + { + let mut c = core.lock().unwrap(); + let known: std::collections::HashSet = + c.sessions.values().map(|s| s.handle.clone()).collect(); + let mut dropped: Vec<(String, String)> = Vec::new(); + c.pending_wakes.retain(|p| { + let has_session = known.contains(&p.handle); + if !has_session { + dropped.push((p.handle.clone(), p.mail_id.clone())); + } + has_session + }); + for (handle, mail_id) in dropped { + let _ = c.append( + "wake/dropped", + json!({ "handle": handle, "mail": mail_id, "why": "recipient session gone; mail stays in the mailbox" }), + ); + } + } + for (handle, mail_id, from, session) in candidates { + if session.idle_for() < WAKE_IDLE_GATE { + continue; + } + // Token, event, and pending-list mutation under the lock; the paste outside. + let (go, root) = { + let mut c = core.lock().unwrap(); + let now = Instant::now(); + let bucket = c.wake_tokens.entry(handle.clone()).or_insert(WakeBucket { + tokens: WAKE_BURST, + last: now, + }); + let refill = now.duration_since(bucket.last).as_secs_f64() * WAKE_REFILL_PER_SEC; + bucket.tokens = (bucket.tokens + refill).min(WAKE_BURST); + bucket.last = now; + if bucket.tokens < 1.0 { + if let Some(p) = c + .pending_wakes + .iter_mut() + .find(|p| p.mail_id == mail_id && !p.capped_logged) + { + p.capped_logged = true; + let _ = + c.append("wake/capped", json!({ "handle": handle, "mail": mail_id })); + } + (false, c.root.clone()) + } else { + bucket.tokens -= 1.0; + (true, c.root.clone()) + } + }; + if !go { + continue; + } + let retired = match bench_mail::retire(&root, &handle, &mail_id) { + Ok(p) => p, + Err(why) => { + let mut c = core.lock().unwrap(); + c.pending_wakes.retain(|p| p.mail_id != mail_id); + let _ = c.append( + "wake/dropped", + json!({ "handle": handle, "mail": mail_id, "why": why }), + ); + continue; + } + }; + let notice = format!("You have mail from {from}: {}", retired.display()); + let delivered = session.deliver_line(¬ice).is_ok(); + let mut c = core.lock().unwrap(); + c.pending_wakes.retain(|p| p.mail_id != mail_id); + if delivered { + let _ = c.append( + "agent/woken", + json!({ "session": session.id, "handle": handle, "mail": mail_id }), + ); + } else { + let _ = c.append( + "wake/dropped", + json!({ "handle": handle, "mail": mail_id, "why": "paste failed" }), + ); + } + } + } +} + enum AfterResponse { Done, /// The connection upgrades to an attach relay AFTER the response line: replay @@ -574,13 +724,48 @@ fn dispatch( runtime_session: agent.mints_session_id().then(mint_session_id), resume: false, }; - let (id, notices) = { + let (id, handle, root, notices) = { let mut c = core.lock().unwrap(); let id = format!("s{}", c.next_session); + let handle = parsed.name.clone().unwrap_or_else(|| id.clone()); + if let Err(why) = validate_handle(&handle) { + return (refused(why), AfterResponse::Done); + } + if handle == OPERATOR_HANDLE { + return ( + refused(format!( + "{OPERATOR_HANDLE:?} is the operator's handle — addressable by anyone, claimable by no session" + )), + AfterResponse::Done, + ); + } + if c.sessions.values().any(|s| s.handle == handle) { + return ( + refused(format!( + "handle {handle:?} is already claimed — `bench sessions` lists them" + )), + AfterResponse::Done, + ); + } c.next_session += 1; - (id, c.notices.clone()) + (id, handle, c.root.clone(), c.notices.clone()) }; - let session = match Session::spawn(id.clone(), &spec, rows, cols, notices) { + // The session learns its address and root, so `bench mail send` inside it + // needs no flags and lands in the right mailroom. + let extra_env = [ + ("BENCH_SESSION".to_string(), id.clone()), + ("BENCH_HANDLE".to_string(), handle.clone()), + ("BENCH_DIR".to_string(), root.display().to_string()), + ]; + let session = match Session::spawn( + id.clone(), + handle.clone(), + &spec, + rows, + cols, + &extra_env, + notices, + ) { Ok(s) => s, Err(why) => return (errored(why), AfterResponse::Done), }; @@ -591,6 +776,7 @@ fn dispatch( "session/spawned", json!({ "session": id, + "handle": session.handle, "agent": agent.name(), "cwd": spec.cwd, "pid": session.pid, @@ -617,6 +803,7 @@ fn dispatch( ( ok(json!({ "session": session.id, + "handle": session.handle, "pid": session.pid, "agent": agent.name(), "runtime_session": session.runtime_session, @@ -635,6 +822,7 @@ fn dispatch( .map(|s| { json!({ "session": s.id, + "handle": s.handle, "agent": s.agent.name(), "cwd": s.cwd, "pid": s.pid, @@ -759,13 +947,26 @@ fn dispatch( } let mut spec = old.spec.clone(); spec.resume = true; - let (id, notices) = { + let (id, root, notices) = { let mut c = core.lock().unwrap(); let id = format!("s{}", c.next_session); c.next_session += 1; - (id, c.notices.clone()) + (id, c.root.clone(), c.notices.clone()) }; - let session = match Session::spawn(id.clone(), &spec, 40, 140, notices) { + let extra_env = [ + ("BENCH_SESSION".to_string(), id.clone()), + ("BENCH_HANDLE".to_string(), old.handle.clone()), + ("BENCH_DIR".to_string(), root.display().to_string()), + ]; + let session = match Session::spawn( + id.clone(), + old.handle.clone(), + &spec, + 40, + 140, + &extra_env, + notices, + ) { Ok(s) => s, Err(why) => return (refused(why), AfterResponse::Done), }; @@ -792,6 +993,153 @@ fn dispatch( ) } + Some(Verb::MailSend) => { + let parsed: MailSendArgs = match serde_json::from_value(req.args.clone()) { + Ok(a) => a, + Err(e) => return (refused(format!("mail/send args: {e}")), AfterResponse::Done), + }; + for (role, h) in [("to", &parsed.to), ("from", &parsed.from)] { + if let Err(why) = validate_handle(h) { + return (refused(format!("{role}: {why}")), AfterResponse::Done); + } + } + let (seq, root) = { + let mut c = core.lock().unwrap(); + let seq = c.next_mail; + c.next_mail += 1; + (seq, c.root.clone()) + }; + let (id, path) = match bench_mail::deliver( + &root, + seq, + &parsed.from, + &parsed.to, + parsed.subject.as_deref(), + &now_rfc3339(), + &parsed.body, + ) { + Ok(pair) => pair, + Err(why) => return (errored(why), AfterResponse::Done), + }; + let wake = { + let mut c = core.lock().unwrap(); + if let Err(why) = c.append( + "mail/sent", + json!({ + "id": id, + "from": parsed.from, + "to": parsed.to, + "subject": parsed.subject, + "path": path.display().to_string(), + }), + ) { + return (errored(why), AfterResponse::Done); + } + let live = c + .sessions + .values() + .any(|s| s.handle == parsed.to && s.is_live()); + if live { + c.pending_wakes.push(PendingWake { + handle: parsed.to.clone(), + mail_id: id.clone(), + from: parsed.from.clone(), + capped_logged: false, + }); + "queued" + } else { + // Honest: the mail is delivered and waits; nothing will wake a + // recipient this daemon does not host. + "no-live-session" + } + }; + ( + ok(json!({ + "id": id, + "to": parsed.to, + "path": path.display().to_string(), + "wake": wake, + })), + AfterResponse::Done, + ) + } + + Some(Verb::MailList) => { + let parsed: MailListArgs = match serde_json::from_value(req.args.clone()) { + Ok(a) => a, + Err(e) => return (refused(format!("mail/list args: {e}")), AfterResponse::Done), + }; + if let Err(why) = validate_handle(&parsed.handle) { + return (refused(why), AfterResponse::Done); + } + const MAX_RETURNED: usize = 200; + let root = core.lock().unwrap().root.clone(); + let all = bench_mail::list(&root, &parsed.handle); + let total = all.len(); + let mail: Vec = all + .iter() + .take(MAX_RETURNED) + .map(|m| { + json!({ + "id": m.id, + "from": m.from, + "subject": m.subject, + "at": m.at, + "unread": m.unread, + }) + }) + .collect(); + let returned = mail.len(); + ( + ok(json!({ + "handle": parsed.handle, + "mail": mail, + "total": total, + "returned": returned, + "truncated": returned < total, + })), + AfterResponse::Done, + ) + } + + Some(Verb::MailRead) => { + let parsed: MailReadArgs = match serde_json::from_value(req.args.clone()) { + Ok(a) => a, + Err(e) => return (refused(format!("mail/read args: {e}")), AfterResponse::Done), + }; + if let Err(why) = validate_handle(&parsed.handle) { + return (refused(why), AfterResponse::Done); + } + let root = core.lock().unwrap().root.clone(); + // Retire-never-delete: reading moves inbox -> read; reading again answers + // from where it lives. + let path = match bench_mail::retire(&root, &parsed.handle, &parsed.id) { + Ok(p) => p, + Err(why) => return (refused(why), AfterResponse::Done), + }; + let body = match bench_mail::read_body(&path) { + Ok(b) => b, + Err(why) => return (errored(why), AfterResponse::Done), + }; + { + let mut c = core.lock().unwrap(); + // Reading is a mutation here (the retirement), so it is logged. + let _ = c.append( + "mail/read", + json!({ "handle": parsed.handle, "id": parsed.id }), + ); + c.pending_wakes.retain(|p| p.mail_id != parsed.id); + } + ( + ok(json!({ + "id": parsed.id, + "path": path.display().to_string(), + "body": body, + })), + AfterResponse::Done, + ) + } + None => ( refused(format!( "unknown verb {:?} — this daemon answers: {}", diff --git a/daemon/direction.md b/daemon/direction.md index 286c31a..5abad9a 100644 --- a/daemon/direction.md +++ b/daemon/direction.md @@ -14,7 +14,14 @@ operator and the agents are equal owners; every verb exists in an addressed, non form; both parties go through the same socket. Migration is strangler-style inside this repo: one vertical at a time, old code unwired only when the new is proven. -**Where it stands: M0 + M5a.** A suite-aware record root, an append-only event log, one +**Where it stands: M0 + M5a + mail.** The daemon owns the mailroom (`bench-mail`: +files are the record, notices carry the path never the body, retire-never-delete, +metadata-only listings) and the wake reactor (`mail/sent ⇒ agent/woken` by pasting into +an idle pty the daemon owns, with the loop cap as a per-recipient token bucket in the +courier — where helm #320 proved it must live). Proven end to end by `just mail-proof`: +a number passed as mail around a real claude→codex→pi ring, +1 per hop. + +**Where it stood before mail: M0 + M5a.** A suite-aware record root, an append-only event log, one unix socket, eight verbs, a CLI speaking helm's exit-code discipline, and a conformance gate that runs the real binaries. M5a is the pty core: `spawn` puts a real interactive agent (claude, codex, pi — the allowlist) into a daemon-owned pty with posture, model diff --git a/daemon/justfile b/daemon/justfile index adaa7e6..e1c60fb 100644 --- a/daemon/justfile +++ b/daemon/justfile @@ -58,13 +58,17 @@ attn-list: watch handle: {{BENCH}} watch --handle {{handle}} -# M2 — mail through benchd's authority. -mail-send to subject body-file: - {{BENCH}} mail/send --to {{to}} --subject "{{subject}}" --body-file {{body-file}} +# Mail — deliver; a live recipient is woken by the reactor. +mail-send to body subject="": + {{BENCH}} mail send --to {{to}} --body "{{body}}" {{ if subject != "" { "--subject " + subject } else { "" } }} -# M2 — the mailbox listing. -mail-list: - {{BENCH}} mail/list +# Mail — metadata only, unread first. +mail-list handle="operator": + {{BENCH}} mail list --handle {{handle}} + +# Mail — body plus retirement (inbox -> read). +mail-read id handle="operator": + {{BENCH}} mail read {{id}} --handle {{handle}} # M5a — spawn an interactive agent into a bench pty. Prompt by FILE, never argv (#93). spawn agent="claude" cwd=justfile_directory() prompt-file="": @@ -104,8 +108,11 @@ spec: # probed last for real; listed here so the row exists printf ' ✓ %-12s (M0)\n' "stop"; continue fi - if cargo run -q -p bench -- "$v" >/dev/null 2>&1; then + err=$(cargo run -q -p bench -- "$v" 2>&1 >/dev/null); code=$? + if [ $code -eq 0 ]; then printf ' ✓ %-12s\n' "$v" + elif [ $code -eq 3 ] && ! printf '%s' "$err" | grep -q "unknown verb"; then + printf ' ✓ %-12s (answers; needs args)\n' "$v" else printf ' ✗ %-12s not yet\n' "$v" fi @@ -148,3 +155,53 @@ live-smoke: ./target/debug/bench stop >/dev/null rm -rf "$SMOKE" echo "live-smoke: clean" + +# The mail proof, real agents end to end: claude, codex and pi in daemon ptys, a number +# passed as MAIL around the ring through the production mailroom and wake reactor — +# each hop +1, so echoes cannot fake it. Costs three small agent turns. +mail-proof: + #!/usr/bin/env bash + set -euo pipefail + DIR="{{justfile_directory()}}" + SMOKE=$(mktemp -d "${TMPDIR:-/tmp}/bmp.XXXX") + export BENCH_DIR="$SMOKE/r" + cargo build -q --workspace + BENCH="$DIR/target/debug/bench" + (timeout 600 "$DIR/target/debug/benchd" &) + sleep 1 + ring=(post-claude post-codex post-pi) + agents=(claude codex pi) + for i in 0 1 2; do + next=${ring[$(( (i + 1) % 3 ))]} + cat > "$SMOKE/brief-$i.txt" <' arrives, run: cat . The last line is an integer N. Compute N+1. Then run: $BENCH mail send --to $next --body "" (with the actual number). Then wait silently for more mail. Never send otherwise. Say READY now and wait. + BRIEF + tr '\n' ' ' < "$SMOKE/brief-$i.txt" > "$SMOKE/brief-$i.line" + "$BENCH" spawn --agent "${agents[$i]}" --cwd "$DIR" --name "${ring[$i]}" --prompt-file "$SMOKE/brief-$i.line" >/dev/null + done + sleep 5 + "$BENCH" mail send --to post-claude --from operator --subject ring --body "100" >/dev/null + echo "mail-proof: seeded 100 -> post-claude; waiting for 103 to come back around..." + deadline=$(( $(date +%s) + 300 )) + while true; do + if [ "$(date +%s)" -gt "$deadline" ]; then + echo "mail-proof: TIMEOUT — events tail:"; "$BENCH" events | tail -c 2000; exit 1 + fi + listing=$("$BENCH" mail list --handle post-claude 2>/dev/null || true) + id=$(printf '%s' "$listing" | python3 -c "import json,sys; d=json.load(sys.stdin); m=[x for x in d['mail'] if x['from']=='post-pi']; print(m[0]['id'] if m else '')" 2>/dev/null || true) + if [ -n "$id" ]; then + body=$("$BENCH" mail read "$id" --handle post-claude | python3 -c "import json,sys; print(json.load(sys.stdin)['body'].strip().splitlines()[-1])") + if [ "$body" = "103" ]; then + echo "mail-proof: 103 arrived from post-pi — full ring through the real mailroom" + break + else + echo "mail-proof: FAILED — final body was $body, expected 103"; exit 1 + fi + fi + sleep 2 + done + "$BENCH" events | python3 -c "import json,sys; d=json.load(sys.stdin); ks=[e['kind'] for e in d['events']]; print('mail-proof events:', ' '.join(k for k in ks if k.startswith(('mail/','agent/woken','wake/'))))" + for h in post-claude post-codex post-pi; do :; done + "$BENCH" stop >/dev/null + rm -rf "$SMOKE" + echo "mail-proof: clean" diff --git a/daemon/spikes/mail-milestone-proof.md b/daemon/spikes/mail-milestone-proof.md new file mode 100644 index 0000000..cd3db48 --- /dev/null +++ b/daemon/spikes/mail-milestone-proof.md @@ -0,0 +1,22 @@ +# Mail milestone — the acceptance proof + +**2026-08-18, `just mail-proof`, first run green.** Three real agents (claude, codex, +pi) in daemon ptys, handles `post-claude`/`post-codex`/`post-pi`, briefed by file. The +operator seeded `100` as mail; each agent was WOKEN by the production reactor (notice: +path + sender, never the body), `cat`-ed the real message file, computed N+1, and sent +onward through the real `bench mail send` — from-identity resolved from `BENCH_HANDLE`, +which the daemon declared into its pty. `103` arrived back from `post-pi` and was +collected with `bench mail read`. + +Event trail, verbatim shape: `mail/sent → agent/woken` ×4, then `mail/read`. The fourth +wake is the honest observation: the ring protocol is unbounded (claude, woken by the +returning 103, would have sent 104), and the run stayed sane because the wake cap is a +per-recipient token bucket in the courier. **Stop conditions live in briefs; the brake +lives in the reactor** — the group-room spike's lesson, now load-bearing in production. + +What the conformance suite pins beyond this run (27 tests, real binaries): +notice-carries-path-never-body observed through the relay; retirement at delivery; +retire-never-delete; metadata-only listings; the cap starving wakes but never mail; +`operator` addressable-never-claimable; duplicate and path-shaped handles refused. + +Rerun anytime: `just mail-proof` (three small agent turns). From 46ee16ffcb63b3a0ed131877cb2dc9e398d63059 Mon Sep 17 00:00:00 2001 From: Rasmus Widing Date: Tue, 18 Aug 2026 21:32:02 +0300 Subject: [PATCH 2/2] =?UTF-8?q?feat(daemon):=20bench-mail=20skill=20?= =?UTF-8?q?=E2=80=94=20the=20capability=20surface,=20snippets=20executed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The intelligence stays with the agents: the skill states what the mailroom mechanically does — your address is BENCH_HANDLE, a wake is one line carrying a path never a body, wakes are idle-gated and capped while mail waits unread, listings are metadata, read retires, files are the record — and prescribes nothing. Etiquette and protocols get added the day a real failure earns them. Per the executed-docs rule, the snippets are not prose that can drift: a conformance test extracts every bash fence from SKILL.md and runs it in order against a real daemon, then asserts the sequence did what the skill says (the read snippet retired the sent message). --- .claude/skills/bench-mail/SKILL.md | 72 ++++++++++++++++++++++++ daemon/crates/bench/tests/conformance.rs | 62 ++++++++++++++++++++ 2 files changed, 134 insertions(+) create mode 100644 .claude/skills/bench-mail/SKILL.md diff --git a/.claude/skills/bench-mail/SKILL.md b/.claude/skills/bench-mail/SKILL.md new file mode 100644 index 0000000..d3bd496 --- /dev/null +++ b/.claude/skills/bench-mail/SKILL.md @@ -0,0 +1,72 @@ +--- +name: bench-mail +description: Send and receive mail between agents on the bench (benchd). Use when you have a BENCH_HANDLE, when a "You have mail" notice appears in your session, or when you need to message another bench agent or the operator. +--- + +# bench mail + +Mail between agents hosted by `benchd`. This is the capability surface — what the +mailroom does, mechanically. What to say, when to reply, and how to structure a +conversation are yours. + +**Your address is `$BENCH_HANDLE`**, set by the daemon that spawned you. If it is +unset, you are not a bench session; you can still send and list as `operator`. +`$BENCH_DIR` (also set for you) is the record root; every command below resolves it +automatically. The `bench` CLI is on your PATH or named by `$BENCH`. + +## Receiving + +A wake is one line, pasted into your session by the daemon: + +```text +You have mail from : +``` + +The path IS the message — a markdown file with `from:`/`at:`/`subject:` front-matter and +the body below it. `cat` it. It is already retired (moved to your `read/` directory), so +the path in the notice stays valid. + +Facts with edges: + +- **The notice never contains the body.** Reading the file is how you get the message. +- **Wakes are delivered only while your pty is idle** (quiet ≥2s), and they are + **capped**: burst of 6 per recipient, refilling one per minute. Capped or undeliverable + mail waits **unread in your inbox** — nothing is lost, but nothing further will nudge + you. `bench mail list` is how you find what accumulated. +- Nothing else wakes you. No polling loop exists to arm. + +## Sending + +```bash +BENCH="${BENCH:-bench}" +$BENCH mail send --to operator --subject demo --body "one line is fine" +``` + +- `--body ` or `--body-file ` (multi-line goes by file). +- Your identity is `$BENCH_HANDLE` automatically; `--from` overrides. +- `operator` is always addressable and belongs to the operator. +- A recipient the daemon does not host is not woken; the response says + `"wake": "no-live-session"` and the mail waits in their box. + +## Listing and reading + +```bash +BENCH="${BENCH:-bench}" +$BENCH mail list --handle operator +``` + +Metadata only — id, sender, subject, time, read-state — bodies never, caps reported. + +```bash +BENCH="${BENCH:-bench}" +ID=$($BENCH mail list --handle operator | python3 -c "import json,sys; m=json.load(sys.stdin)['mail']; print(m[0]['id'] if m else '')") +[ -n "$ID" ] && $BENCH mail read "$ID" --handle operator +``` + +`read` returns the body and retires the message (inbox → `read/`). Nothing in the +mailroom ever deletes; the files under `$BENCH_DIR/mail//` are the record and +plain `cat` reads them. + +## Exit codes + +`0` ok · `2` no daemon · `3` refused (the reason names the rule) · `4` daemon failed. diff --git a/daemon/crates/bench/tests/conformance.rs b/daemon/crates/bench/tests/conformance.rs index 39d2572..98adc6e 100644 --- a/daemon/crates/bench/tests/conformance.rs +++ b/daemon/crates/bench/tests/conformance.rs @@ -1119,3 +1119,65 @@ fn a_claimed_handle_refuses_a_second_claim_and_a_path_is_not_a_handle() { ); assert_eq!(c.code, 3, "stderr: {}", c.stderr); } + +#[test] +fn the_bench_mail_skills_snippets_execute_against_a_real_daemon() { + // The house rule: a documented snippet is executed, never restated — a test that + // retypes it is a second copy that drifts (helm's mail-skill gate, ported). Every + // ```bash fence in SKILL.md runs in order, as operator, against a throwaway root. + let skill = fs::read_to_string( + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../.claude/skills/bench-mail/SKILL.md"), + ) + .expect("bench-mail SKILL.md readable"); + let mut snippets: Vec = Vec::new(); + let mut current: Option = None; + for line in skill.lines() { + match (&mut current, line.trim()) { + (None, "```bash") => current = Some(String::new()), + (Some(buf), "```") => { + snippets.push(std::mem::take(buf)); + current = None; + } + (Some(buf), _) => { + buf.push_str(line); + buf.push('\n'); + } + _ => {} + } + } + assert!( + snippets.len() >= 3, + "the skill's send/list/read snippets exist" + ); + + let home = TestHome::claim("skill"); + let _daemon = DaemonGuard::start(&home.dir, None); + let root = home.dir.join(".bench"); + for (i, snippet) in snippets.iter().enumerate() { + let out = Command::new("bash") + .args(["-euo", "pipefail", "-c", snippet]) + .env_remove("BENCH_SUITE") + .env_remove("BENCH_HANDLE") + .env("HOME", &home.dir) + .env("BENCH_DIR", &root) + .env("BENCH", bench_bin()) + .output() + .expect("run snippet"); + assert!( + out.status.success(), + "SKILL.md snippet {} failed (exit {:?}):\n{}\n--- stderr:\n{}", + i + 1, + out.status.code(), + snippet, + String::from_utf8_lossy(&out.stderr) + ); + } + // The sequence is the story the skill tells: a send exists, the listing shows it + // or its retirement, and the read snippet retired it. + let listing = bench(&home.dir, &["mail", "list", "--handle", "operator"]); + assert!( + listing.stdout.contains("\"unread\": false"), + "the read snippet retired the sent message: {}", + listing.stdout + ); +}