diff --git a/daemon/AGENTS.md b/daemon/AGENTS.md index 4aaac51..3a6653b 100644 --- a/daemon/AGENTS.md +++ b/daemon/AGENTS.md @@ -19,8 +19,10 @@ 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/benchd` — the daemon. Foreground, one unix socket, serial request handling. -- `crates/bench` — the CLI, the one agent-facing surface (and the future skill surface). +- `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. +- `crates/bench` — the CLI, the one agent-facing surface, and the attach client. - There is deliberately **no root `Cargo.toml`** in the repo: `cargo` at the repo root fails loudly instead of half-working. diff --git a/daemon/Cargo.lock b/daemon/Cargo.lock index 99c4aa2..b5937b6 100644 --- a/daemon/Cargo.lock +++ b/daemon/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + [[package]] name = "bench" version = "0.0.1" @@ -10,6 +16,13 @@ dependencies = [ "serde_json", ] +[[package]] +name = "bench-session" +version = "0.0.1" +dependencies = [ + "portable-pty", +] + [[package]] name = "bench-wire" version = "0.0.1" @@ -22,35 +35,128 @@ dependencies = [ name = "benchd" version = "0.0.1" dependencies = [ + "bench-session", "bench-wire", "serde_json", "time", ] +[[package]] +name = "bitflags" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" + +[[package]] +name = "bitflags" +version = "2.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "cfg_aliases" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd16c4719339c4530435d38e511904438d07cce7950afa3718a84ac36c10e89e" + [[package]] name = "deranged" version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +[[package]] +name = "downcast-rs" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" + +[[package]] +name = "filedescriptor" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e40758ed24c9b2eeb76c35fb0aebc66c626084edd827e07e1552279814c6682d" +dependencies = [ + "libc", + "thiserror", + "winapi", +] + [[package]] name = "itoa" version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "log" +version = "0.4.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" + [[package]] name = "memchr" version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "nix" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab2156c4fce2f8df6c499cc1c763e4394b7482525bf2a9701c9d79d215f519e4" +dependencies = [ + "bitflags 2.13.1", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "num-conv" version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "portable-pty" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4a596a2b3d2752d94f51fac2d4a96737b8705dddd311a32b9af47211f08671e" +dependencies = [ + "anyhow", + "bitflags 1.3.2", + "downcast-rs", + "filedescriptor", + "lazy_static", + "libc", + "log", + "nix", + "serial2", + "shared_library", + "shell-words", + "winapi", + "winreg", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -102,7 +208,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.3", ] [[package]] @@ -118,6 +224,44 @@ dependencies = [ "zmij", ] +[[package]] +name = "serial2" +version = "0.2.38" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b16809bc35793b19ce4e0c53924bc0dce3937f15487997cfdaed936004180730" +dependencies = [ + "cfg-if", + "libc", + "windows-sys", +] + +[[package]] +name = "shared_library" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a9e7e0f2bfae24d8a5b5a66c5b257a83c7412304311512a0c054cd5e619da11" +dependencies = [ + "lazy_static", + "libc", +] + +[[package]] +name = "shell-words" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc6fe69c597f9c37bfeeeeeb33da3530379845f10be461a66d16d03eca2ded77" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "3.0.3" @@ -129,6 +273,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "time" version = "0.3.55" @@ -165,6 +329,52 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "winreg" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80d0f4e272c85def139476380b12f9ac60926689dd2e01d4923222f40580869d" +dependencies = [ + "winapi", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/daemon/Cargo.toml b/daemon/Cargo.toml index 72e6da2..c72d63c 100644 --- a/daemon/Cargo.toml +++ b/daemon/Cargo.toml @@ -2,7 +2,7 @@ # `cargo` at the repo root should fail loudly, not half-work (bench-roadmap.md, M0). [workspace] resolver = "2" -members = ["crates/bench-wire", "crates/benchd", "crates/bench"] +members = ["crates/bench-wire", "crates/bench-session", "crates/benchd", "crates/bench"] [workspace.package] version = "0.0.1" diff --git a/daemon/crates/bench-session/Cargo.toml b/daemon/crates/bench-session/Cargo.toml new file mode 100644 index 0000000..fd0810b --- /dev/null +++ b/daemon/crates/bench-session/Cargo.toml @@ -0,0 +1,8 @@ +[package] +name = "bench-session" +version.workspace = true +edition.workspace = true +description = "The pty core: daemon-owned sessions hosting full interactive agent TUIs." + +[dependencies] +portable-pty = "0.9" diff --git a/daemon/crates/bench-session/src/lib.rs b/daemon/crates/bench-session/src/lib.rs new file mode 100644 index 0000000..44b02c5 --- /dev/null +++ b/daemon/crates/bench-session/src/lib.rs @@ -0,0 +1,633 @@ +//! The pty core — M5a's substance, built directly on what the spikes proved +//! (`daemon/spikes/`): a headless process can own ptys hosting full interactive agent +//! TUIs, deliver prompts paste-then-submit, and re-enter session state from outside. +//! +//! This crate knows runtimes and ptys; it knows nothing about sockets, verbs, or the +//! event log — the daemon composes those. Everything an incident already paid for is +//! carried as code: +//! +//! - **Postures are helm's `SpoolUnattendedPolicy`, plus the model/effort columns the +//! model-selection spike proved.** A posture removes a prompt; it never withholds +//! capability (helm #179). +//! - **The agent allowlist is the security line** (helm `SpoolPolicy`): a spawn request +//! arrives over a socket, and `sh` in a login shell is what an ungated spawn would be. +//! - **Prompts travel by file, never argv** (helm #93) — and are pasted, then submitted +//! separately (helm's launch-line rule; bracketed-paste measurement). +//! - **Runtime session ids are minted at spawn, never inferred later** — the +//! session-state spike grabbed a live session that was not ours by inferring; minting +//! is what makes `resume` a lookup instead of a guess. +//! - **Close is drain-then-die**: SIGKILL races the transcript write (session-state +//! spike), so a close is a term, a grace, then the kill. +//! - **The reader never stops draining the master** — an undrained pty blocks the agent +//! on write (pty spike). + +use portable_pty::{Child, CommandBuilder, MasterPty, PtySize, native_pty_system}; +use std::collections::VecDeque; +use std::io::{Read, Write}; +use std::os::unix::net::UnixStream; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::Sender; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +/// Ring capacity per session: enough scrollback for an attach to land mid-thought, +/// small enough that fifty sessions are a footnote. The cap is reported to an attacher +/// via `replayed`, never silent. +pub const RING_CAPACITY: usize = 256 * 1024; + +/// The environment gate for the conformance-test agent. Real deployments never set it; +/// CI has no claude/codex/pi, and the relay's byte-fidelity still has to be proven +/// against a REAL spawned process — `/bin/cat` echoes what it is sent, which is exactly +/// the oracle a relay test needs. +pub const TEST_AGENT_ENV: &str = "BENCH_SESSION_TEST_AGENT"; + +// --------------------------------------------------------------------------- +// Agents and their argv +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AgentKind { + Claude, + Codex, + Pi, + /// `/bin/cat`, admitted only when `BENCH_SESSION_TEST_AGENT=1` — the conformance + /// suite's echo oracle, refused everywhere else. + TestEcho, +} + +impl AgentKind { + pub fn parse(raw: &str, test_agent_allowed: bool) -> Result { + match raw { + "claude" => Ok(AgentKind::Claude), + "codex" => Ok(AgentKind::Codex), + "pi" => Ok(AgentKind::Pi), + "test-echo" if test_agent_allowed => Ok(AgentKind::TestEcho), + other => Err(format!( + "agent {other:?} is not on the allowlist — this bench spawns: claude, codex, pi" + )), + } + } + + pub fn name(&self) -> &'static str { + match self { + AgentKind::Claude => "claude", + AgentKind::Codex => "codex", + AgentKind::Pi => "pi", + AgentKind::TestEcho => "test-echo", + } + } + + /// Whether this runtime can mint its session identity at spawn — the property that + /// makes `resume` a lookup. codex names its own sessions after the fact, so resume + /// for it is refused with the reason rather than guessed at (session-state spike). + pub fn mints_session_id(&self) -> bool { + matches!(self, AgentKind::Claude | AgentKind::Pi) + } +} + +/// What a spawn (or resume — `resume_from` set) wants. Pure data; `argv()` is the one +/// spelling of every posture/model/effort/resume flag, unit-tested per runtime. +#[derive(Debug, Clone)] +pub struct SpawnSpec { + pub agent: AgentKind, + pub cwd: String, + pub model: Option, + pub effort: Option, + /// The runtime session id this bench minted (claude, pi) — present on spawn for + /// minting runtimes, and on resume naming what to re-enter. + pub runtime_session: Option, + pub resume: bool, +} + +/// The single spelling of how each runtime is started unattended. Postures verbatim +/// from helm's `SpoolUnattendedPolicy`; model/effort flags verbatim from the +/// model-selection spike; resume flags from the session-state spike. +pub fn argv(spec: &SpawnSpec) -> Result<(String, Vec), String> { + let mut args: Vec = Vec::new(); + let program = match spec.agent { + AgentKind::Claude => { + args.push("--dangerously-skip-permissions".into()); + if let Some(m) = &spec.model { + args.extend(["--model".into(), m.clone()]); + } + if let Some(e) = &spec.effort { + args.extend(["--effort".into(), e.clone()]); + } + match (&spec.runtime_session, spec.resume) { + (Some(id), false) => args.extend(["--session-id".into(), id.clone()]), + (Some(id), true) => args.extend(["--resume".into(), id.clone()]), + (None, false) => {} + (None, true) => return Err("claude resume needs the minted session id".into()), + } + "claude" + } + AgentKind::Codex => { + if spec.resume { + return Err( + "codex names its own sessions after the fact; resume is not supported for it yet — spawn fresh, or use claude/pi where the bench mints the id" + .into(), + ); + } + args.push("--dangerously-bypass-approvals-and-sandbox".into()); + if let Some(m) = &spec.model { + args.extend(["-m".into(), m.clone()]); + } + if let Some(e) = &spec.effort { + args.extend(["-c".into(), format!("model_reasoning_effort={e}")]); + } + "codex" + } + AgentKind::Pi => { + args.push("--approve".into()); + if let Some(m) = &spec.model { + // pi carries thinking as a `:` suffix on the model — one flag, + // measured working in the model-selection spike. + let model = match &spec.effort { + Some(e) => format!("{m}:{e}"), + None => m.clone(), + }; + args.extend(["--model".into(), model]); + } else if let Some(e) = &spec.effort { + args.extend(["--thinking".into(), e.clone()]); + } + // --session-id creates when missing and re-enters when present, so spawn + // and resume are the same flag (session-state spike). + if let Some(id) = &spec.runtime_session { + args.extend(["--session-id".into(), id.clone()]); + } else if spec.resume { + return Err("pi resume needs the minted session id".into()); + } + "pi" + } + AgentKind::TestEcho => { + if spec.resume { + return Err("the test agent has no sessions to resume".into()); + } + "/bin/cat" + } + }; + Ok((program.to_string(), args)) +} + +/// Mint a runtime session id. `uuidgen` where present; a /dev/urandom-derived v4 shape +/// otherwise. Never inferred after the fact — that is the spike's hard-won rule. +pub fn mint_session_id() -> String { + if let Ok(out) = std::process::Command::new("uuidgen").output() { + let s = String::from_utf8_lossy(&out.stdout).trim().to_lowercase(); + if s.len() == 36 { + return s; + } + } + let mut bytes = [0u8; 16]; + if let Ok(mut f) = std::fs::File::open("/dev/urandom") { + let _ = f.read_exact(&mut bytes); + } + bytes[6] = (bytes[6] & 0x0f) | 0x40; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + let h: Vec = bytes.iter().map(|b| format!("{b:02x}")).collect(); + format!( + "{}{}{}{}-{}{}-{}{}-{}{}-{}{}{}{}{}{}", + h[0], + h[1], + h[2], + h[3], + h[4], + h[5], + h[6], + h[7], + h[8], + h[9], + h[10], + h[11], + h[12], + h[13], + h[14], + h[15] + ) +} + +// --------------------------------------------------------------------------- +// The live session +// --------------------------------------------------------------------------- + +/// What the reader thread reports upward. The daemon logs these — bench-visible means +/// logged, and the exit of a session is exactly the kind of fact the log exists for. +#[derive(Debug)] +pub enum Notice { + Exited { session: String }, + Detached { session: String }, +} + +struct Ring { + bytes: VecDeque, + total: u64, + last_change: Instant, +} + +impl Ring { + fn push(&mut self, chunk: &[u8]) { + self.total += chunk.len() as u64; + self.last_change = Instant::now(); + for &b in chunk { + if self.bytes.len() == RING_CAPACITY { + self.bytes.pop_front(); + } + self.bytes.push_back(b); + } + } +} + +pub struct Session { + pub id: String, + pub spec: SpawnSpec, + pub agent: AgentKind, + pub cwd: String, + pub pid: Option, + pub runtime_session: Option, + pub spawned_at: Instant, + master: Mutex>, + writer: Mutex>, + child: Arc>>, + ring: Arc>, + /// dtach-grade: at most one attached client. A new attach REPLACES the old one — + /// reconnect-after-drop is the common case, and "already attached" refusals would + /// strand every dropped connection until a timeout nothing owns. The generation is + /// what makes replacement safe: a replaced connection's pump thread wakes on the + /// shutdown and must clear ONLY the attachment it owned — clearing blindly tears + /// down the newcomer (found by the takeover conformance test, not by review). + attached: Arc>>, + attach_gen: AtomicU64, + exited: Arc, +} + +impl Session { + /// Spawn the agent into a fresh pty and start the drain thread. `notices` is how + /// exits and forced detaches reach the daemon's log. + pub fn spawn( + id: String, + spec: &SpawnSpec, + rows: u16, + cols: u16, + notices: Sender, + ) -> Result, String> { + let (program, args) = argv(spec)?; + let pty = native_pty_system(); + let pair = pty + .openpty(PtySize { + rows, + cols, + pixel_width: 0, + pixel_height: 0, + }) + .map_err(|e| format!("openpty: {e}"))?; + let mut cmd = CommandBuilder::new(&program); + for a in &args { + cmd.arg(a); + } + cmd.cwd(&spec.cwd); + cmd.env("TERM", "xterm-256color"); + let child = pair + .slave + .spawn_command(cmd) + .map_err(|e| format!("spawn {program}: {e}"))?; + drop(pair.slave); + + let mut reader = pair + .master + .try_clone_reader() + .map_err(|e| format!("clone reader: {e}"))?; + let writer = pair + .master + .take_writer() + .map_err(|e| format!("take writer: {e}"))?; + + let session = Arc::new(Session { + pid: child.process_id(), + spec: spec.clone(), + id: id.clone(), + agent: spec.agent, + cwd: spec.cwd.clone(), + runtime_session: spec.runtime_session.clone(), + spawned_at: Instant::now(), + master: Mutex::new(pair.master), + writer: Mutex::new(writer), + child: Arc::new(Mutex::new(child)), + ring: Arc::new(Mutex::new(Ring { + bytes: VecDeque::with_capacity(8192), + total: 0, + last_change: Instant::now(), + })), + attached: Arc::new(Mutex::new(None)), + attach_gen: AtomicU64::new(0), + exited: Arc::new(AtomicBool::new(false)), + }); + + // The drain thread: the pty owner's first duty. It also carries live output to + // the attached client, byte-for-byte — escape sequences included, which is what + // lets an OSC ride the relay into whatever terminal hosts `bench attach`. + { + let ring = Arc::clone(&session.ring); + let attached = Arc::clone(&session.attached); + let exited = Arc::clone(&session.exited); + let child = Arc::clone(&session.child); + std::thread::spawn(move || { + let mut chunk = [0u8; 8192]; + loop { + match reader.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => { + ring.lock().unwrap().push(&chunk[..n]); + let mut guard = attached.lock().unwrap(); + if let Some((_, stream)) = guard.as_mut() + && stream.write_all(&chunk[..n]).is_err() + { + let _ = stream.shutdown(std::net::Shutdown::Both); + *guard = None; + let _ = notices.send(Notice::Detached { + session: id.clone(), + }); + } + } + } + } + // Reap at the moment of exit (PR #341 review, R2): EOF on the master + // means the child is gone or going; wait() here ends its lifetime with + // its bytes, so no session leaves a zombie for `close` to find — and + // `resume`'s removal of the old session needs no second job. close()'s + // own wait after this is an ignored ECHILD, never a hang. + let _ = child.lock().unwrap().wait(); + exited.store(true, Ordering::SeqCst); + let _ = notices.send(Notice::Exited { + session: id.clone(), + }); + }); + } + Ok(session) + } + + pub fn is_live(&self) -> bool { + !self.exited.load(Ordering::SeqCst) + } + + pub fn is_attached(&self) -> bool { + self.attached.lock().unwrap().is_some() + } + + pub fn output_bytes(&self) -> u64 { + self.ring.lock().unwrap().total + } + + /// 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(); + w.write_all(line.as_bytes()) + .map_err(|e| format!("paste: {e}"))?; + w.flush().ok(); + drop(w); + std::thread::sleep(Duration::from_millis(300)); + let mut w = self.writer.lock().unwrap(); + w.write_all(b"\r").map_err(|e| format!("submit: {e}"))?; + w.flush().ok(); + Ok(()) + } + + pub fn write_input(&self, bytes: &[u8]) -> Result<(), String> { + let mut w = self.writer.lock().unwrap(); + w.write_all(bytes).map_err(|e| format!("input: {e}"))?; + w.flush().ok(); + Ok(()) + } + + /// Wait until the TUI is ready for its first paste. claude has a content marker + /// (the yolo footer — settle heuristics alone raced history redraws, measured); + /// the others settle on quiet output. The test agent is ready by construction. + pub fn wait_ready(&self, cap: Duration) -> bool { + if self.agent == AgentKind::TestEcho { + return true; + } + let start = Instant::now(); + loop { + { + let ring = self.ring.lock().unwrap(); + match self.agent { + AgentKind::Claude => { + let text: String = ring.bytes.iter().map(|&b| b as char).collect(); + if text.contains("bypass permissions") { + drop(ring); + std::thread::sleep(Duration::from_secs(1)); + return true; + } + } + _ => { + if ring.total > 500 && ring.last_change.elapsed() > Duration::from_secs(2) { + return true; + } + } + } + } + if start.elapsed() > cap { + return false; + } + std::thread::sleep(Duration::from_millis(200)); + } + } + + /// Attach: resize to the viewer, replay the ring, then hand live output to this + /// stream. Replaces any previous attachment — the old stream is shut down, which + /// its client sees as EOF. Returns the generation this attachment owns; the pump + /// hands it back to `detach_generation` so a replaced pump cannot clear its + /// replacement. + pub fn attach(&self, stream: UnixStream, rows: u16, cols: u16) -> Result { + if rows > 0 && cols > 0 { + let master = self.master.lock().unwrap(); + let _ = master.resize(PtySize { + rows, + cols, + pixel_width: 0, + pixel_height: 0, + }); + } + let replay: Vec = { + let ring = self.ring.lock().unwrap(); + ring.bytes.iter().copied().collect() + }; + let mut s = stream; + s.write_all(&replay).map_err(|e| format!("replay: {e}"))?; + let generation = self.attach_gen.fetch_add(1, Ordering::SeqCst) + 1; + let mut guard = self.attached.lock().unwrap(); + if let Some((_, old)) = guard.take() { + let _ = old.shutdown(std::net::Shutdown::Both); + } + *guard = Some((generation, s)); + Ok(generation) + } + + /// Unconditional — for close/stop, where whatever is attached goes. + pub fn detach(&self) { + let mut guard = self.attached.lock().unwrap(); + if let Some((_, old)) = guard.take() { + let _ = old.shutdown(std::net::Shutdown::Both); + } + } + + /// Clear the attachment only if `generation` still owns it. Returns whether it did + /// — a pump whose attachment was taken over reports nothing, because the detach it + /// noticed was the takeover, already logged from the other side. + pub fn detach_generation(&self, generation: u64) -> bool { + let mut guard = self.attached.lock().unwrap(); + match guard.as_ref() { + Some((g, _)) if *g == generation => { + if let Some((_, old)) = guard.take() { + let _ = old.shutdown(std::net::Shutdown::Both); + } + true + } + _ => false, + } + } + + /// Drain-then-die (session-state spike): a grace for the runtime to flush its + /// transcript, a term, another grace, then the kill. Returns whether it was still + /// live when asked. + pub fn close(&self, grace: Duration) -> bool { + let was_live = self.is_live(); + if was_live { + std::thread::sleep(grace); + } + self.detach(); + let mut child = self.child.lock().unwrap(); + let _ = child.kill(); + let _ = child.wait(); + was_live + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn spec(agent: AgentKind) -> SpawnSpec { + SpawnSpec { + agent, + cwd: "/tmp".into(), + model: None, + effort: None, + runtime_session: None, + resume: false, + } + } + + #[test] + fn postures_are_helms_table_verbatim() { + let (p, a) = argv(&spec(AgentKind::Claude)).unwrap(); + assert_eq!(p, "claude"); + assert_eq!(a, vec!["--dangerously-skip-permissions"]); + let (p, a) = argv(&spec(AgentKind::Codex)).unwrap(); + assert_eq!(p, "codex"); + assert_eq!(a, vec!["--dangerously-bypass-approvals-and-sandbox"]); + let (p, a) = argv(&spec(AgentKind::Pi)).unwrap(); + assert_eq!(p, "pi"); + assert_eq!(a, vec!["--approve"]); + } + + #[test] + fn model_and_effort_flags_match_the_spike() { + let mut s = spec(AgentKind::Claude); + s.model = Some("opus".into()); + s.effort = Some("high".into()); + let (_, a) = argv(&s).unwrap(); + assert_eq!( + a, + vec![ + "--dangerously-skip-permissions", + "--model", + "opus", + "--effort", + "high" + ] + ); + + let mut s = spec(AgentKind::Codex); + s.model = Some("gpt-5.3-codex".into()); + s.effort = Some("high".into()); + let (_, a) = argv(&s).unwrap(); + assert_eq!( + a, + vec![ + "--dangerously-bypass-approvals-and-sandbox", + "-m", + "gpt-5.3-codex", + "-c", + "model_reasoning_effort=high" + ] + ); + + let mut s = spec(AgentKind::Pi); + s.model = Some("anthropic/claude-opus-4-5".into()); + s.effort = Some("high".into()); + let (_, a) = argv(&s).unwrap(); + assert_eq!( + a, + vec!["--approve", "--model", "anthropic/claude-opus-4-5:high"] + ); + } + + #[test] + fn resume_needs_a_minted_id_and_codex_refuses_with_the_reason() { + let mut s = spec(AgentKind::Claude); + s.resume = true; + assert!(argv(&s).is_err(), "resume without an id must refuse"); + s.runtime_session = Some("abc-123".into()); + let (_, a) = argv(&s).unwrap(); + assert!(a.contains(&"--resume".to_string()) && a.contains(&"abc-123".to_string())); + + let mut s = spec(AgentKind::Pi); + s.runtime_session = Some("sess-9".into()); + let (_, a) = argv(&s).unwrap(); + assert!(a.contains(&"--session-id".to_string())); + s.resume = true; + let (_, a) = argv(&s).unwrap(); + assert!( + a.contains(&"--session-id".to_string()), + "pi resume is the same flag" + ); + + let mut s = spec(AgentKind::Codex); + s.resume = true; + let err = argv(&s).unwrap_err(); + assert!( + err.contains("resume is not supported"), + "the refusal names the reason: {err}" + ); + } + + #[test] + fn the_allowlist_refuses_arbitrary_commands_and_gates_the_test_agent() { + assert!(AgentKind::parse("sh", false).is_err()); + assert!( + AgentKind::parse("test-echo", false).is_err(), + "test agent needs the env gate" + ); + assert!(AgentKind::parse("test-echo", true).is_ok()); + assert!(AgentKind::parse("claude", false).is_ok()); + } + + #[test] + fn minted_ids_are_uuid_shaped() { + let id = mint_session_id(); + assert_eq!(id.len(), 36, "{id}"); + assert_eq!(id.chars().filter(|&c| c == '-').count(), 4); + } + + #[test] + fn the_ring_caps_and_reports_totals() { + let mut ring = Ring { + bytes: VecDeque::new(), + total: 0, + last_change: Instant::now(), + }; + ring.push(&vec![b'x'; RING_CAPACITY + 100]); + assert_eq!(ring.bytes.len(), RING_CAPACITY); + assert_eq!(ring.total, (RING_CAPACITY + 100) as u64); + } +} diff --git a/daemon/crates/bench-wire/src/lib.rs b/daemon/crates/bench-wire/src/lib.rs index 0885500..3b9566d 100644 --- a/daemon/crates/bench-wire/src/lib.rs +++ b/daemon/crates/bench-wire/src/lib.rs @@ -122,13 +122,20 @@ impl RequestId { /// string derives from this list, the dispatcher matches on the parsed enum so the /// 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"]; +pub const KNOWN_VERBS: &[&str] = &[ + "status", "events", "stop", "spawn", "sessions", "attach", "close", "resume", +]; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Verb { Status, Events, Stop, + Spawn, + Sessions, + Attach, + Close, + Resume, } impl Verb { @@ -138,6 +145,11 @@ impl Verb { "status" => Some(Verb::Status), "events" => Some(Verb::Events), "stop" => Some(Verb::Stop), + "spawn" => Some(Verb::Spawn), + "sessions" => Some(Verb::Sessions), + "attach" => Some(Verb::Attach), + "close" => Some(Verb::Close), + "resume" => Some(Verb::Resume), _ => None, } } @@ -151,11 +163,53 @@ impl Verb { /// client that connects and never finishes its line gets a refusal, not the daemon. pub const DAEMON_IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); -/// A caller waits at most this long for an answer — strictly longer than the daemon's -/// own bound, so a daemon-side refusal always outruns the client giving up. A timeout -/// maps to `EXIT_NO_DAEMON`: no exit code at all is the one failure an unattended -/// agent cannot act on. -pub const CLIENT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15); +/// How long the daemon waits for a spawned TUI to look ready before giving up on +/// prompt delivery. Lives HERE, not in the daemon, because the client's patience is +/// derived from it below — PR #341's R1 was these two numbers spelled apart (30 vs 15), +/// so a slow spawn exited 2 "no daemon" while the daemon was mid-success. +pub const READY_WAIT: std::time::Duration = std::time::Duration::from_secs(10); + +/// A caller waits at most this long for an answer — strictly longer than every +/// daemon-side wait, **true by construction**: the sum of the waits plus slack, so the +/// two sides cannot drift apart again. A timeout maps to `EXIT_NO_DAEMON`: no exit +/// code at all is the one failure an unattended agent cannot act on. +pub const CLIENT_READ_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(READY_WAIT.as_secs() + DAEMON_IO_TIMEOUT.as_secs() + 5); + +// --------------------------------------------------------------------------- +// Session verb payloads +// --------------------------------------------------------------------------- + +/// `spawn`'s payload, typed once (PR #341 review, R3): both binaries serialize and +/// decode this struct, so a one-sided rename is a compile error or a refusal naming +/// the missing field — never a silently-defaulted option or an allowlist refusal that +/// misdescribes a missing key. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SpawnArgs { + pub agent: String, + pub cwd: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prompt_file: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub effort: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rows: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cols: Option, +} + +/// The payload shared by `attach`, `close` and `resume`: a session id, plus the +/// viewer's size where the verb has a viewer. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SessionArgs { + pub session: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub rows: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cols: Option, +} // --------------------------------------------------------------------------- // The envelope @@ -363,12 +417,31 @@ mod tests { } assert_eq!( KNOWN_VERBS.len(), - 3, + 8, "a new verb joins KNOWN_VERBS and this count together" ); assert!(Verb::parse("frobnicate").is_none()); } + #[test] + fn the_clients_patience_outlasts_every_daemon_wait_by_construction() { + assert!( + CLIENT_READ_TIMEOUT > READY_WAIT + DAEMON_IO_TIMEOUT, + "R1's invariant: a daemon-side outcome always outruns the client giving up" + ); + } + + #[test] + fn spawn_args_refuse_a_missing_required_key_naming_the_field() { + let err = serde_json::from_value::(serde_json::json!({"cwd": "/tmp"})) + .unwrap_err() + .to_string(); + assert!(err.contains("agent"), "the refusal names the field: {err}"); + let ok: SpawnArgs = + serde_json::from_value(serde_json::json!({"agent": "claude", "cwd": "/tmp"})).unwrap(); + assert!(ok.model.is_none() && ok.rows.is_none()); + } + #[test] fn status_maps_to_helm_exit_codes() { assert_eq!(Status::Ok.exit_code(), 0); diff --git a/daemon/crates/bench/src/main.rs b/daemon/crates/bench/src/main.rs index 0ce9dc8..77fbf72 100644 --- a/daemon/crates/bench/src/main.rs +++ b/daemon/crates/bench/src/main.rs @@ -1,22 +1,23 @@ -//! bench — the CLI, which is also the future agent skill surface (bench-roadmap M0/M3). +//! bench — the CLI, the one agent-facing surface over the benchd socket (and, at M5a, +//! the operator's attach client). //! -//! One connection, one JSON request line, one JSON response line. The exit code IS the -//! outcome — an agent reads `$?`, not prose: +//! One connection, one JSON request line, one JSON response line — except `attach`, +//! which upgrades the same connection into a raw byte relay after the response: pty +//! output down, keystrokes up, Ctrl-\ to detach. The exit code IS the outcome: //! -//! 0 ok (the verb's data, pretty JSON, on stdout) -//! 2 no daemon (the socket could not be reached — transport, not a daemon answer) +//! 0 ok (the verb's data, pretty JSON, on stdout; attach: a clean detach) +//! 2 no daemon (the socket could not be reached, or never answered — transport) //! 3 refused (the daemon said no and named why, on stderr) //! 4 daemon failed (the daemon tried and could not, named why, on stderr) //! -//! These are helm's spool codes, kept on purpose: every agent skill in this repo already -//! knows them, and a code that changes meaning across tools is worse than no code. +//! 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, SuiteName, resolve_root, socket_path, }; use serde_json::{Value, json}; -use std::io::{BufRead, BufReader, Write}; +use std::io::{IsTerminal, Read, Write}; use std::os::unix::net::UnixStream; use std::path::PathBuf; use std::process; @@ -27,35 +28,60 @@ fn main() { } fn usage() -> &'static str { - "usage: bench [--suite ] \n\ - verbs: status daemon identity, root, uptime, event count\n\ - \x20 events [--since N] read the record back from seq N\n\ - \x20 stop ask the daemon to log its stop and exit\n\ + "usage: bench [--suite ] [args]\n\ + verbs: status daemon identity, root, uptime, counts\n\ + \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 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\ env: BENCH_SUITE (flag wins) · BENCH_DIR (root override, wins over suite)\n\ exit: 0 ok · 2 no daemon · 3 refused · 4 daemon failed" } +struct Cli { + verb: String, + args: Value, + root: PathBuf, +} + fn run() -> i32 { - let mut args = std::env::args().skip(1).peekable(); + let mut argv = std::env::args().skip(1).peekable(); let mut suite_flag: Option = None; let mut verb: Option = None; + let mut positional: Vec = Vec::new(); + let mut flags: Vec<(String, String)> = Vec::new(); let mut since: u64 = 0; - while let Some(arg) = args.next() { + while let Some(arg) = argv.next() { match arg.as_str() { - "--suite" => match args.next() { + "--suite" => match argv.next() { Some(v) => suite_flag = Some(v), None => return refuse("--suite needs a name"), }, - "--since" => match args.next().and_then(|v| v.parse::().ok()) { + "--since" => match argv.next().and_then(|v| v.parse::().ok()) { Some(n) => since = n, None => return refuse("--since needs a sequence number"), }, + "--agent" | "--cwd" | "--prompt-file" | "--model" | "--effort" | "--rows" + | "--cols" => { + let key = arg.trim_start_matches("--").replace('-', "_"); + match argv.next() { + Some(v) => flags.push((key, v)), + None => return refuse(&format!("{arg} needs a value")), + } + } "--help" | "-h" => { println!("{}", usage()); return 0; } other if verb.is_none() && !other.starts_with('-') => verb = Some(other.to_string()), + other if verb.is_some() && !other.starts_with('-') => { + positional.push(other.to_string()) + } other => return refuse(&format!("unknown argument {other:?}\n{}", usage())), } } @@ -64,9 +90,8 @@ fn run() -> i32 { return refuse(usage()); }; - // The suite is validated here, before any socket is touched: a name that cannot - // isolate must never resolve to the shared root by accident (#86 semantics). The - // daemon applies the identical rule from the same crate — one spelling, two edges. + // 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. let suite_raw = suite_flag.or_else(|| std::env::var("BENCH_SUITE").ok()); let suite = match suite_raw.as_deref() { Some(raw) => match SuiteName::validate(raw) { @@ -81,57 +106,77 @@ 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 sock = socket_path(&root); - - let args_value = match verb.as_str() { - "events" if since > 0 => json!({ "since": since }), - _ => Value::Null, - }; - let request = Request { - id: request_id(), - verb, - args: args_value, - }; - - let stream = match UnixStream::connect(&sock) { - Ok(s) => s, - Err(e) => { - eprintln!("bench: no daemon at {} ({e})", sock.display()); - return EXIT_NO_DAEMON; + let mut args = serde_json::Map::new(); + match verb.as_str() { + "events" if since > 0 => { + args.insert("since".into(), json!(since)); } - }; - // Bounded on both directions (R2): a caller that hangs produces no exit code, - // which is the one failure an unattended agent cannot act on. A timeout is exit 2 - // — from the caller's seat, a daemon that never answers IS no daemon. - let _ = stream.set_write_timeout(Some(DAEMON_IO_TIMEOUT)); - let _ = stream.set_read_timeout(Some(CLIENT_READ_TIMEOUT)); + "spawn" => { + for (k, v) in &flags { + let value: Value = match k.as_str() { + "rows" | "cols" => match v.parse::() { + Ok(n) => json!(n), + Err(_) => return refuse(&format!("--{k} needs a number")), + }, + _ => json!(v), + }; + args.insert(k.clone(), value); + } + } + "attach" | "close" | "resume" => match positional.first() { + Some(s) => { + args.insert("session".into(), json!(s)); + } + None => { + 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 mut line = match serde_json::to_string(&request) { - Ok(l) => l, - Err(e) => return fail(&format!("cannot encode request: {e}")), + let cli = Cli { + verb: verb.clone(), + args: Value::Object(args), + root, }; - line.push('\n'); - if let Err(e) = (&stream).write_all(line.as_bytes()) { - eprintln!("bench: write to {} failed ({e})", sock.display()); - return EXIT_NO_DAEMON; + if verb == "attach" { + attach(cli) + } else { + simple(cli) } +} - let mut reply = String::new(); - if BufReader::new(&stream).read_line(&mut reply).is_err() || reply.is_empty() { - eprintln!( - "bench: no answer from {} within {}s", - sock.display(), - CLIENT_READ_TIMEOUT.as_secs() - ); +/// The ordinary one-line-in, one-line-out path. +fn simple(cli: Cli) -> i32 { + let (stream, request_line) = match open(&cli) { + Ok(pair) => pair, + Err(code) => return code, + }; + if let Err(e) = (&stream).write_all(request_line.as_bytes()) { + eprintln!("bench: write failed ({e})"); return EXIT_NO_DAEMON; } - + let reply = match read_response_line(&stream) { + Some(r) => r, + None => { + eprintln!("bench: no answer within {}s", CLIENT_READ_TIMEOUT.as_secs()); + return EXIT_NO_DAEMON; + } + }; let response: Response = match serde_json::from_str(&reply) { Ok(r) => r, Err(e) => return fail(&format!("unreadable response ({e}): {}", reply.trim())), }; - if let Some(reason) = &response.reason { eprintln!("bench: {reason}"); } @@ -144,6 +189,181 @@ fn run() -> i32 { response.status.exit_code() } +/// The attach path: response line, then the connection is a raw relay. Local terminal +/// goes raw (keystrokes reach the agent unmangled, Ctrl-C included); Ctrl-\ detaches. +fn attach(cli: Cli) -> i32 { + let (stream, request_line) = match open(&cli) { + Ok(pair) => pair, + Err(code) => return code, + }; + if let Err(e) = (&stream).write_all(request_line.as_bytes()) { + eprintln!("bench: write failed ({e})"); + return EXIT_NO_DAEMON; + } + let reply = match read_response_line(&stream) { + Some(r) => r, + None => { + eprintln!("bench: no answer within {}s", CLIENT_READ_TIMEOUT.as_secs()); + return EXIT_NO_DAEMON; + } + }; + let response: Response = match serde_json::from_str(&reply) { + Ok(r) => r, + Err(e) => return fail(&format!("unreadable response ({e}): {}", reply.trim())), + }; + if response.status != Status::Ok { + if let Some(reason) = &response.reason { + eprintln!("bench: {reason}"); + } + return response.status.exit_code(); + } + eprintln!("bench: attached — Ctrl-\\ detaches"); + let _ = stream.set_read_timeout(None); + + // Raw local terminal for the duration; restored on the way out. `stty -g` gives a + // restore token, so whatever the mode was is what comes back. + let stdin_is_tty = std::io::stdin().is_terminal(); + let saved = if stdin_is_tty { + let saved = std::process::Command::new("stty") + .arg("-g") + .stdin(std::process::Stdio::inherit()) + .output() + .ok() + .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string()); + let _ = std::process::Command::new("stty") + .args(["raw", "-echo"]) + .stdin(std::process::Stdio::inherit()) + .status(); + saved + } else { + None + }; + + // Down: socket → stdout, byte-for-byte — escape sequences ride through, which is + // what lets an OSC from the agent reach whatever terminal hosts this client. + let down = { + let mut sock = match stream.try_clone() { + Ok(s) => s, + Err(_) => return fail("cannot clone stream"), + }; + std::thread::spawn(move || { + let mut out = std::io::stdout(); + let mut chunk = [0u8; 8192]; + loop { + match sock.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => { + if out.write_all(&chunk[..n]).is_err() { + break; + } + let _ = out.flush(); + } + } + } + }) + }; + + // Up: stdin → socket, until Ctrl-\ (0x1C) or EOF. + let mut stdin = std::io::stdin(); + let up_sock = stream; + let mut chunk = [0u8; 1024]; + let code = loop { + match stdin.read(&mut chunk) { + Ok(0) | Err(_) => break 0, + Ok(n) => { + if let Some(pos) = chunk[..n].iter().position(|&b| b == 0x1c) { + if pos > 0 { + let _ = (&up_sock).write_all(&chunk[..pos]); + } + break 0; + } + if (&up_sock).write_all(&chunk[..n]).is_err() { + break 0; + } + } + } + }; + let _ = up_sock.shutdown(std::net::Shutdown::Both); + let _ = down.join(); + if let Some(token) = saved { + let _ = std::process::Command::new("stty") + .arg(token) + .stdin(std::process::Stdio::inherit()) + .status(); + } + eprintln!("\nbench: detached"); + code +} + +fn open(cli: &Cli) -> Result<(UnixStream, String), i32> { + let sock = socket_path(&cli.root); + let stream = match UnixStream::connect(&sock) { + Ok(s) => s, + Err(e) => { + eprintln!("bench: no daemon at {} ({e})", sock.display()); + return Err(EXIT_NO_DAEMON); + } + }; + // Bounded on both directions (R2): a hung caller has no exit code, which is the one + // failure an unattended agent cannot act on. + let _ = stream.set_write_timeout(Some(DAEMON_IO_TIMEOUT)); + let _ = stream.set_read_timeout(Some(CLIENT_READ_TIMEOUT)); + let request = Request { + id: request_id(), + verb: cli.verb.clone(), + args: cli.args.clone(), + }; + let mut line = match serde_json::to_string(&request) { + Ok(l) => l, + Err(_) => return Err(4), + }; + line.push('\n'); + Ok((stream, line)) +} + +/// Read exactly the response line, byte by byte — a BufReader would swallow the raw +/// replay bytes that follow it on an attach connection. +fn read_response_line(mut stream: &UnixStream) -> Option { + let mut line = Vec::new(); + let mut byte = [0u8; 1]; + loop { + match stream.read(&mut byte) { + Ok(0) | Err(_) => { + return if line.is_empty() { + None + } else { + Some(String::from_utf8_lossy(&line).into_owned()) + }; + } + Ok(_) => { + if byte[0] == b'\n' { + return Some(String::from_utf8_lossy(&line).into_owned()); + } + line.push(byte[0]); + if line.len() > 1_000_000 { + return None; + } + } + } + } +} + +fn terminal_size() -> Option<(u16, u16)> { + if !std::io::stdout().is_terminal() { + return None; + } + let out = std::process::Command::new("stty") + .arg("size") + .stdin(std::process::Stdio::inherit()) + .output() + .ok()?; + let text = String::from_utf8_lossy(&out.stdout); + let mut parts = text.split_whitespace(); + let rows = parts.next()?.parse().ok()?; + let cols = parts.next()?.parse().ok()?; + Some((rows, cols)) +} + // Pre-socket exits derive from the same enum as socket-answered ones (R3). fn refuse(why: &str) -> i32 { eprintln!("bench: {why}"); @@ -155,8 +375,7 @@ fn fail(why: &str) -> i32 { Status::Error.exit_code() } -/// A fresh id per invocation, inside `RequestId`'s own pattern — validated, not assumed, -/// so the client can never send an id the daemon-side rule would refuse. +/// A fresh id per invocation, inside `RequestId`'s own pattern — validated, not assumed. fn request_id() -> String { let nanos = SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/daemon/crates/bench/tests/conformance.rs b/daemon/crates/bench/tests/conformance.rs index 0769a9b..96cfa7e 100644 --- a/daemon/crates/bench/tests/conformance.rs +++ b/daemon/crates/bench/tests/conformance.rs @@ -13,7 +13,7 @@ //! binary is located beside our own CARGO_BIN_EXE path. use std::fs; -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; @@ -72,6 +72,7 @@ impl DaemonGuard { let mut cmd = Command::new(benchd_bin()); cmd.env_remove("BENCH_DIR") .env_remove("BENCH_SUITE") + .env("BENCH_SESSION_TEST_AGENT", "1") .env("HOME", home) .stdout(Stdio::null()) .stderr(Stdio::null()); @@ -564,3 +565,278 @@ fn a_second_daemon_on_a_claimed_root_is_refused_loudly() { "the refusal says who holds the root: {stderr}" ); } + +// --------------------------------------------------------------------------- +// M5a: the pty core — sessions, the relay, and the OSC passthrough proof +// --------------------------------------------------------------------------- + +/// Send one request on a raw socket and return (response_line, open_stream). +fn raw_request( + socket: &Path, + verb: &str, + args: serde_json::Value, +) -> (serde_json::Value, UnixStream) { + let stream = UnixStream::connect(socket).expect("connect"); + let req = format!( + "{{\"id\":\"t-{}\",\"verb\":\"{verb}\",\"args\":{args}}}\n", + std::process::id() + ); + (&stream).write_all(req.as_bytes()).unwrap(); + let mut line = Vec::new(); + let mut b = [0u8; 1]; + loop { + match (&stream).read(&mut b) { + Ok(0) | Err(_) => break, + Ok(_) => { + if b[0] == b'\n' { + break; + } + line.push(b[0]); + } + } + } + ( + serde_json::from_slice(&line).expect("response json"), + stream, + ) +} + +#[test] +fn spawn_refuses_an_agent_off_the_allowlist() { + let home = TestHome::claim("allow"); + let _daemon = DaemonGuard::start(&home.dir, None); + let run = bench(&home.dir, &["spawn", "--agent", "sh", "--cwd", "/tmp"]); + assert_eq!(run.code, 3, "stderr: {}", run.stderr); + assert!( + run.stderr.contains("allowlist"), + "the refusal names the rule: {}", + run.stderr + ); +} + +#[test] +fn spawn_refuses_a_cwd_that_is_not_an_absolute_directory() { + let home = TestHome::claim("cwd"); + let _daemon = DaemonGuard::start(&home.dir, None); + let run = bench( + &home.dir, + &["spawn", "--agent", "test-echo", "--cwd", "relative/x"], + ); + assert_eq!(run.code, 3, "stderr: {}", run.stderr); +} + +#[test] +fn a_session_relays_bytes_faithfully_including_an_osc_sequence() { + // THE canvas-passthrough proof (M5a's named assumption): an OSC 777 written by the + // agent must cross the relay byte-for-byte, because whatever terminal hosts + // `bench attach` is what parses it — helm included. + let home = TestHome::claim("relay"); + let daemon = DaemonGuard::start(&home.dir, None); + let run = bench( + &home.dir, + &["spawn", "--agent", "test-echo", "--cwd", "/tmp"], + ); + assert_eq!(run.code, 0, "stderr: {}", run.stderr); + let spawned: serde_json::Value = serde_json::from_str(&run.stdout).unwrap(); + let sid = spawned["session"].as_str().unwrap().to_string(); + + let (resp, stream) = raw_request( + &daemon.socket, + "attach", + serde_json::json!({"session": sid}), + ); + assert_eq!(resp["status"], "ok", "{resp}"); + + // cat echoes what the pty carries; the OSC must come back intact. + let osc = "\u{1b}]777;notify;helm.canvas;/tmp/proof.html\u{7}"; + let payload = format!("before {osc} after\n"); + (&stream).write_all(payload.as_bytes()).unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + let mut seen = Vec::new(); + let mut chunk = [0u8; 4096]; + let _ = stream.set_read_timeout(Some(Duration::from_millis(300))); + while Instant::now() < deadline { + match (&stream).read(&mut chunk) { + Ok(0) => break, + Ok(n) => seen.extend_from_slice(&chunk[..n]), + Err(_) => {} + } + if seen.windows(osc.len()).any(|w| w == osc.as_bytes()) { + break; + } + } + assert!( + seen.windows(osc.len()).any(|w| w == osc.as_bytes()), + "the OSC must survive the relay byte-for-byte; got {} bytes: {:?}", + seen.len(), + String::from_utf8_lossy(&seen) + ); + + let close = bench(&home.dir, &["close", &sid]); + assert_eq!(close.code, 0, "stderr: {}", close.stderr); + + // Bench-visible means logged: the session's whole life is in the record. + let log = fs::read_to_string(home.dir.join(".bench/events.jsonl")).unwrap(); + for kind in ["session/spawned", "session/attached", "session/closed"] { + assert!(log.contains(kind), "{kind} missing from the log"); + } +} + +#[test] +fn a_second_attach_takes_over_and_the_first_sees_eof() { + let home = TestHome::claim("takeover"); + let daemon = DaemonGuard::start(&home.dir, None); + let run = bench( + &home.dir, + &["spawn", "--agent", "test-echo", "--cwd", "/tmp"], + ); + let sid = serde_json::from_str::(&run.stdout).unwrap()["session"] + .as_str() + .unwrap() + .to_string(); + + let (r1, s1) = raw_request( + &daemon.socket, + "attach", + serde_json::json!({"session": sid}), + ); + assert_eq!(r1["status"], "ok"); + let (r2, _s2) = raw_request( + &daemon.socket, + "attach", + serde_json::json!({"session": sid}), + ); + assert_eq!(r2["status"], "ok"); + + // The first stream is shut down by the takeover — its next read is EOF, not a hang. + let _ = s1.set_read_timeout(Some(Duration::from_secs(5))); + let mut chunk = [0u8; 64]; + let deadline = Instant::now() + Duration::from_secs(5); + let mut eof = false; + while Instant::now() < deadline { + match (&s1).read(&mut chunk) { + Ok(0) => { + eof = true; + break; + } + Ok(_) => {} + Err(_) => { + eof = true; + break; + } + } + } + assert!(eof, "the replaced attach must be closed, not left dangling"); + + let sessions = bench(&home.dir, &["sessions"]); + let data: serde_json::Value = serde_json::from_str(&sessions.stdout).unwrap(); + assert_eq!( + data["sessions"][0]["live"], true, + "takeover must not kill the session" + ); +} + +#[test] +fn an_exited_session_refuses_attach_and_the_exit_is_logged() { + let home = TestHome::claim("exited"); + let _daemon = DaemonGuard::start(&home.dir, None); + let run = bench( + &home.dir, + &["spawn", "--agent", "test-echo", "--cwd", "/tmp"], + ); + let spawned: serde_json::Value = serde_json::from_str(&run.stdout).unwrap(); + let sid = spawned["session"].as_str().unwrap().to_string(); + let pid = spawned["pid"].as_u64().unwrap(); + + // Kill the agent out from under the daemon — the reader thread must notice and log. + libc_kill(pid as i32); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let log = fs::read_to_string(home.dir.join(".bench/events.jsonl")).unwrap_or_default(); + if log.contains("session/exited") { + break; + } + assert!( + Instant::now() < deadline, + "session/exited never reached the log" + ); + std::thread::sleep(Duration::from_millis(100)); + } + + // R2: the exit must also REAP — a `` child is a lifetime the daemon owns + // and dropped. `ps -o stat=` on a zombie prints a state containing 'Z'; a reaped + // pid prints nothing. + let reap_deadline = Instant::now() + Duration::from_secs(5); + loop { + let out = Command::new("ps") + .args(["-o", "stat=", "-p", &pid.to_string()]) + .output() + .unwrap(); + let stat = String::from_utf8_lossy(&out.stdout).trim().to_string(); + if !stat.contains('Z') { + break; + } + assert!( + Instant::now() < reap_deadline, + "the exited child stayed a zombie (stat {stat:?}) — the drain thread must reap" + ); + std::thread::sleep(Duration::from_millis(100)); + } + + let attach = bench(&home.dir, &["attach", &sid]); + assert_eq!( + attach.code, 3, + "attach to an exited session is a refusal: {}", + attach.stderr + ); + assert!( + attach.stderr.contains("resume"), + "the refusal names the route: {}", + attach.stderr + ); + + // The test agent has nothing to resume — the refusal says why. + let resume = bench(&home.dir, &["resume", &sid]); + assert_eq!(resume.code, 3, "stderr: {}", resume.stderr); +} + +#[test] +fn stop_takes_the_sessions_with_it() { + let home = TestHome::claim("stopall"); + let _daemon = DaemonGuard::start(&home.dir, None); + let run = bench( + &home.dir, + &["spawn", "--agent", "test-echo", "--cwd", "/tmp"], + ); + let pid = serde_json::from_str::(&run.stdout).unwrap()["pid"] + .as_u64() + .unwrap(); + let stop = bench(&home.dir, &["stop"]); + assert_eq!(stop.code, 0); + let deadline = Instant::now() + Duration::from_secs(8); + loop { + let alive = libc_alive(pid as i32); + if !alive { + break; + } + assert!( + Instant::now() < deadline, + "the spawned agent outlived the daemon's stop" + ); + std::thread::sleep(Duration::from_millis(100)); + } +} + +// Minimal libc shims — kill(2) with SIGKILL / signal 0 liveness — to avoid a dependency +// for two calls. +unsafe extern "C" { + fn kill(pid: i32, sig: i32) -> i32; +} +fn libc_kill(pid: i32) { + unsafe { + kill(pid, 9); + } +} +fn libc_alive(pid: i32) -> bool { + unsafe { kill(pid, 0) == 0 } +} diff --git a/daemon/crates/benchd/Cargo.toml b/daemon/crates/benchd/Cargo.toml index 0a8fce9..1b22a80 100644 --- a/daemon/crates/benchd/Cargo.toml +++ b/daemon/crates/benchd/Cargo.toml @@ -6,5 +6,6 @@ description = "The bench daemon: owns the record, answers on the socket." [dependencies] bench-wire = { path = "../bench-wire" } +bench-session = { path = "../bench-session" } serde_json.workspace = true time.workspace = true diff --git a/daemon/crates/benchd/src/main.rs b/daemon/crates/benchd/src/main.rs index 4f1427d..e35d89f 100644 --- a/daemon/crates/benchd/src/main.rs +++ b/daemon/crates/benchd/src/main.rs @@ -1,37 +1,43 @@ -//! benchd — the bench daemon. M0: skeleton and isolation. +//! benchd — the bench daemon. M0 (skeleton and isolation) + M5a (the pty core). //! -//! What exists at this milestone: a suite-aware record root, an append-only event log -//! that is the single source of truth, and one unix socket answering three verbs -//! (`status`, `events`, `stop`). What deliberately does not exist yet: panes, mail, -//! attention, taps — those are M1+ (docs/future-planning/bench-roadmap.md) and each -//! arrives as new event kinds plus new verbs over this same spine, never as a second -//! channel beside it. +//! What exists: a suite-aware record root, an append-only event log that is the single +//! source of truth, and one unix socket answering eight verbs — status/events/stop from +//! M0, and spawn/sessions/attach/close/resume from M5a: daemon-owned ptys hosting full +//! interactive agent TUIs, viewed through a dtach-grade raw relay (`bench attach`). +//! What deliberately does not exist yet: mail, attention, taps, the painter — those are +//! next milestones, and each arrives as new event kinds plus new verbs over this same +//! spine, never as a second channel beside it. //! //! Design rules this file carries (argued in ../../direction.md): -//! - **Bench-visible means logged.** Every mutation appends an event before the response -//! that reports it; readers project from the log, never from daemon memory alone. -//! - **One door.** The socket is the only way in; the CLI, the face, and every agent use -//! the same verbs. There is no privileged in-process path to grow attached to. +//! - **Bench-visible means logged.** Every mutation appends an event before the +//! response that reports it; readers project from the log, never from daemon memory +//! alone. Sessions exiting, attaching, detaching — all events. +//! - **One door.** The socket is the only way in. `attach` upgrades a connection to a +//! raw byte relay AFTER an ordinary response line — same door, one more room. //! - **Refuse loudly.** Unknown verbs, malformed requests, oversized lines, a corrupt -//! log, an already-claimed socket: each is a named refusal, never a silent default. +//! log, an off-allowlist agent: each is a named refusal, never a silent default. //! -//! The daemon runs in the foreground and takes one request per connection, serially. -//! That is enough for M0's callers by construction, and a bounded, inspectable behavior -//! beats a concurrency story nothing needs yet. +//! Connections are handled on a thread each; the shared core (log + session registry) +//! sits behind one mutex held only for map and log operations — never across a ready +//! wait, a prompt delivery, or an attach pump. +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, Request, Response, Status, SuiteName, Verb, check_socket_path, events_path, - resolve_root, socket_path, + MAX_REQUEST_BYTES, READY_WAIT, Request, Response, SessionArgs, SpawnArgs, Status, SuiteName, + Verb, check_socket_path, events_path, resolve_root, socket_path, }; use serde_json::{Value, json}; +use std::collections::HashMap; use std::fs::{self, File, OpenOptions}; use std::io::{BufRead, BufReader, Read, Write}; use std::os::unix::fs::{DirBuilderExt, PermissionsExt}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::PathBuf; use std::process; -use std::time::Instant; +use std::sync::mpsc; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; fn main() { process::exit(run()); @@ -45,8 +51,6 @@ fn usage() -> &'static str { } fn run() -> i32 { - // Flag wins over environment — the caller's explicit word over the inherited one, - // helm's PaneEnvironment convention. let mut args = std::env::args().skip(1); let mut suite_flag: Option = None; while let Some(arg) = args.next() { @@ -67,8 +71,8 @@ fn run() -> i32 { let suite = match suite_raw.as_deref() { Some(raw) => match SuiteName::validate(raw) { Ok(s) => Some(s), - // The whole point of the suite is isolation; a name that cannot isolate - // must stop the launch, never fall back to the operator's live root (#86). + // A name that cannot isolate must stop the launch, never fall back to the + // operator's live root (#86). Err(why) => return refuse_start(&why), }, None => None, @@ -81,15 +85,14 @@ fn run() -> i32 { let bench_dir = std::env::var("BENCH_DIR").ok(); let root = resolve_root(bench_dir.as_deref(), suite.as_ref(), &home); - match Daemon::start(root, suite) { - Ok(mut daemon) => daemon.serve(), + match boot(root, suite) { + Ok(code) => code, Err(StartError::Refused(why)) => refuse_start(&why), Err(StartError::Failed(why)) => fail_start(&why), } } -// Pre-socket exits derive from the same enum as socket-answered ones (R3): one -// spelling of the contract, no hand-typed twin to drift. +// Pre-socket exits derive from the same enum as socket-answered ones (R3). fn refuse_start(why: &str) -> i32 { eprintln!("benchd: refusing to start: {why}"); Status::Refused.exit_code() @@ -112,8 +115,7 @@ struct RepairNote { /// 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 to a named sibling, truncate the log back to its last good byte, -/// and report the repair so the caller can log it (R1). +/// quarantine the tail, truncate back to the last good byte, and report the repair (R1). fn scan_log(events: &PathBuf) -> Result<(u64, Option), StartError> { let bytes = match fs::read(events) { Ok(b) => b, @@ -143,7 +145,6 @@ fn scan_log(events: &PathBuf) -> Result<(u64, Option), StartError> { i + 1 ))); } - // Torn tail: quarantine, truncate, and say so loudly. let epoch = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs()) @@ -178,339 +179,671 @@ fn scan_log(events: &PathBuf) -> Result<(u64, Option), StartError> { Ok((seq, None)) } -struct Daemon { +/// The shared core: the log and the session registry, behind one mutex held only for +/// map and log operations. +struct Core { root: PathBuf, suite: Option, - listener: UnixListener, log: File, next_seq: u64, started_at: String, booted: Instant, + sessions: HashMap>, + next_session: u64, + notices: mpsc::Sender, } -impl Daemon { - fn start(root: PathBuf, suite: Option) -> Result { - // 0700, like every helm record directory: single-user machine, but the record - // is still nobody else's to read. - let mut builder = fs::DirBuilder::new(); - builder.recursive(true).mode(0o700); - builder.create(&root).map_err(|e| { - StartError::Failed(format!("cannot create record root {}: {e}", root.display())) - })?; - - let sock = socket_path(&root); - check_socket_path(&sock).map_err(StartError::Refused)?; +impl Core { + fn append(&mut self, kind: &str, data: Value) -> Result<(), String> { + let event = Event { + seq: self.next_seq, + at: now_rfc3339(), + kind: kind.to_string(), + data, + }; + let mut line = + serde_json::to_string(&event).map_err(|e| format!("cannot encode event: {e}"))?; + line.push('\n'); + self.log + .write_all(line.as_bytes()) + .and_then(|()| self.log.flush()) + .map_err(|e| format!("cannot append to event log: {e}"))?; + // Best-effort durability: the record is the point of this process. + let _ = self.log.sync_data(); + self.next_seq += 1; + Ok(()) + } +} - // A socket file can outlive its daemon (SIGKILL leaves it behind). Connectable - // means a live daemon owns this root — refuse, because two writers on one log is - // corruption with extra steps. Dead means stale — say so and reclaim. - if sock.exists() { - match UnixStream::connect(&sock) { - Ok(_) => { - return Err(StartError::Refused(format!( - "a live benchd already answers at {} — one daemon per root", - sock.display() - ))); - } - Err(_) => { - eprintln!("benchd: removing stale socket {}", sock.display()); - fs::remove_file(&sock).map_err(|e| { - StartError::Failed(format!("cannot remove stale socket: {e}")) - })?; - } +fn boot(root: PathBuf, suite: Option) -> Result { + let mut builder = fs::DirBuilder::new(); + builder.recursive(true).mode(0o700); + builder.create(&root).map_err(|e| { + StartError::Failed(format!("cannot create record root {}: {e}", root.display())) + })?; + + let sock = socket_path(&root); + check_socket_path(&sock).map_err(StartError::Refused)?; + + // A connectable socket means a live daemon owns this root — refuse; a dead one is + // stale — reclaim, saying so. + if sock.exists() { + match UnixStream::connect(&sock) { + Ok(_) => { + return Err(StartError::Refused(format!( + "a live benchd already answers at {} — one daemon per root", + sock.display() + ))); + } + Err(_) => { + eprintln!("benchd: removing stale socket {}", sock.display()); + fs::remove_file(&sock) + .map_err(|e| StartError::Failed(format!("cannot remove stale socket: {e}")))?; } } + } - // Boot-time integrity scan. The log is the record; a daemon that appends after - // history it cannot read would be writing history it does not understand. One - // exception, from PR #340's review (R1): a torn LAST line is an interrupted - // append — the daemon's own crash mid-write, or ENOSPC part-way through a line - // — and refusing it forever bricks the root with no route out. The tail is - // quarantined beside the log and the repair is itself logged. A bad line in - // the MIDDLE stays a refusal naming the line: that one is unexplained. - let events = events_path(&root); - let (next_seq, repair) = scan_log(&events)?; - - let log = OpenOptions::new() - .create(true) - .append(true) - .open(&events) - .map_err(|e| { - StartError::Failed(format!("cannot open event log {}: {e}", events.display())) - })?; - let _ = fs::set_permissions(&events, fs::Permissions::from_mode(0o600)); - - let listener = UnixListener::bind(&sock) - .map_err(|e| StartError::Failed(format!("cannot bind {}: {e}", sock.display())))?; - - let mut daemon = Daemon { - root, - suite, - listener, - log, - next_seq, - started_at: now_rfc3339(), - booted: Instant::now(), - }; - // A fresh log opens with its format marker (R4): the file is read outside the - // process, and a reader that predates a change must fail on the marker rather - // than misread history. - if daemon.next_seq == 0 { - daemon - .append( - "log/format", - json!({ "format": EVENTS_LOG_FORMAT, "version": EVENTS_LOG_VERSION }), - ) - .map_err(StartError::Failed)?; + let events = events_path(&root); + let (next_seq, repair) = scan_log(&events)?; + + let log = OpenOptions::new() + .create(true) + .append(true) + .open(&events) + .map_err(|e| { + StartError::Failed(format!("cannot open event log {}: {e}", events.display())) + })?; + let _ = fs::set_permissions(&events, fs::Permissions::from_mode(0o600)); + + let listener = UnixListener::bind(&sock) + .map_err(|e| StartError::Failed(format!("cannot bind {}: {e}", sock.display())))?; + + let (notice_tx, notice_rx) = mpsc::channel::(); + let core = Arc::new(Mutex::new(Core { + root, + suite, + log, + next_seq, + started_at: now_rfc3339(), + booted: Instant::now(), + sessions: HashMap::new(), + next_session: 1, + notices: notice_tx, + })); + + { + let mut c = core.lock().unwrap(); + if c.next_seq == 0 { + c.append( + "log/format", + json!({ "format": EVENTS_LOG_FORMAT, "version": EVENTS_LOG_VERSION }), + ) + .map_err(StartError::Failed)?; } if let Some(note) = repair { - daemon - .append( - "log/repaired", - json!({ - "quarantine": note.quarantine.display().to_string(), - "dropped_bytes": note.dropped_bytes, - }), - ) - .map_err(StartError::Failed)?; - } - daemon - .append( - "daemon/started", + c.append( + "log/repaired", json!({ - "pid": process::id(), - "version": env!("CARGO_PKG_VERSION"), - "suite": daemon.suite.as_ref().map(|s| s.as_str()), + "quarantine": note.quarantine.display().to_string(), + "dropped_bytes": note.dropped_bytes, }), ) .map_err(StartError::Failed)?; + } + let suite_name = c.suite.as_ref().map(|s| s.as_str().to_string()); + c.append( + "daemon/started", + json!({ + "pid": process::id(), + "version": env!("CARGO_PKG_VERSION"), + "suite": suite_name, + }), + ) + .map_err(StartError::Failed)?; eprintln!( "benchd {} listening at {} (root {})", env!("CARGO_PKG_VERSION"), - socket_path(&daemon.root).display(), - daemon.root.display() + socket_path(&c.root).display(), + c.root.display() ); - Ok(daemon) } - fn serve(&mut self) -> i32 { - loop { - let stream = match self.listener.accept() { - Ok((s, _)) => s, - Err(e) => { - eprintln!("benchd: accept failed: {e}"); - continue; + // Session notices — exits and forced detaches — become events. The reader threads + // send; this thread logs. Bench-visible means logged, including facts nobody asked + // a verb for. + { + let core = Arc::clone(&core); + std::thread::spawn(move || { + while let Ok(notice) = notice_rx.recv() { + let mut c = core.lock().unwrap(); + match notice { + Notice::Exited { session } => { + let _ = c.append("session/exited", json!({ "session": session })); + } + Notice::Detached { session } => { + let _ = c.append("session/detached", json!({ "session": session })); + } } - }; - match self.handle(stream) { - Handled::Continue => {} - Handled::Stop => break, } - } - // The stop event was appended before the response that promised it; all that is - // left is to stop answering. - let _ = fs::remove_file(socket_path(&self.root)); - 0 + }); } - fn handle(&mut self, stream: UnixStream) -> Handled { - // Bounded in TIME as well as bytes (R2): this is the daemon's only loop, and a - // client that connects and never finishes its line must get the refusal, not - // the daemon. Timeouts are set before the reader clone so both share them. - let _ = stream.set_read_timeout(Some(DAEMON_IO_TIMEOUT)); - let _ = stream.set_write_timeout(Some(DAEMON_IO_TIMEOUT)); - let mut reader = BufReader::new(match stream.try_clone() { - Ok(s) => s, - Err(_) => return Handled::Continue, - }); - let mut line = String::new(); - // Bounded read: a line that never ends must not become memory nobody asked for. - let mut limited = (&mut reader).take(MAX_REQUEST_BYTES as u64 + 1); - if limited.read_line(&mut line).is_err() { - respond( - &stream, - &Response { - id: "timed-out".into(), - status: Status::Refused, - reason: Some(format!( - "request not completed within {}s — one line, newline-terminated", - DAEMON_IO_TIMEOUT.as_secs() - )), - data: None, - }, - ); - return Handled::Continue; - } - if line.len() > MAX_REQUEST_BYTES { + for stream in listener.incoming() { + let Ok(stream) = stream else { continue }; + let core = Arc::clone(&core); + std::thread::spawn(move || handle(core, stream)); + } + Ok(0) +} + +enum AfterResponse { + Done, + /// The connection upgrades to an attach relay AFTER the response line: replay + /// happens then, so the protocol stays "one response line first" even with a full + /// ring. This thread then pumps client bytes into the session's pty until EOF. + Pump { + session: Arc, + stream: UnixStream, + rows: u16, + cols: u16, + }, + Stop, +} + +fn handle(core: Arc>, stream: UnixStream) { + // Bounded in time as well as bytes (R2): this connection gets DAEMON_IO_TIMEOUT to + // deliver its line; an attach upgrade lifts the bound after the response. + let _ = stream.set_read_timeout(Some(DAEMON_IO_TIMEOUT)); + let _ = stream.set_write_timeout(Some(DAEMON_IO_TIMEOUT)); + let mut reader = BufReader::new(match stream.try_clone() { + Ok(s) => s, + Err(_) => return, + }); + let mut line = String::new(); + let mut limited = (&mut reader).take(MAX_REQUEST_BYTES as u64 + 1); + if limited.read_line(&mut line).is_err() { + respond( + &stream, + &Response { + id: "timed-out".into(), + status: Status::Refused, + reason: Some(format!( + "request not completed within {}s — one line, newline-terminated", + DAEMON_IO_TIMEOUT.as_secs() + )), + data: None, + }, + ); + return; + } + if line.len() > MAX_REQUEST_BYTES { + respond( + &stream, + &Response { + id: "oversized".into(), + status: Status::Refused, + reason: Some(format!("request exceeds {MAX_REQUEST_BYTES} bytes")), + data: None, + }, + ); + return; + } + let request: Request = match serde_json::from_str(&line) { + Ok(r) => r, + Err(e) => { respond( &stream, &Response { - id: "oversized".into(), + id: "unparseable".into(), status: Status::Refused, - reason: Some(format!("request exceeds {MAX_REQUEST_BYTES} bytes")), + reason: Some(format!("not a request: {e}")), data: None, }, ); - return Handled::Continue; + return; } + }; - // Permissive in shape, strict in judgment: a body that is not a Request still - // gets a refusal naming the parse failure, under the only id we have. - let request: Request = match serde_json::from_str(&line) { - Ok(r) => r, - Err(e) => { - respond( - &stream, - &Response { - id: "unparseable".into(), - status: Status::Refused, - reason: Some(format!("not a request: {e}")), - data: None, - }, - ); - return Handled::Continue; + let (response, after) = dispatch(&core, &request, &stream); + match after { + AfterResponse::Done => respond(&stream, &response), + AfterResponse::Stop => { + respond(&stream, &response); + // Drain-then-die for every session, then leave. process::exit is deliberate: + // the accept loop has no other owner to unblock. + let sessions: Vec> = { + let c = core.lock().unwrap(); + c.sessions.values().cloned().collect() + }; + for s in sessions { + let _ = s.close(Duration::from_secs(1)); } - }; - - let (response, outcome) = self.dispatch(&request); - respond(&stream, &response); - outcome + let root = core.lock().unwrap().root.clone(); + let _ = fs::remove_file(socket_path(&root)); + process::exit(0); + } + AfterResponse::Pump { + session, + stream: raw, + rows, + cols, + } => { + // Response first (an ordinary line), then the replay, then this connection + // is a byte relay — the protocol stays "one response line first" even with + // a full ring. + respond_keep_open(&raw, &response); + let _ = raw.set_read_timeout(None); + let relay = match raw.try_clone() { + Ok(s) => s, + Err(_) => return, + }; + let generation = match session.attach(relay, rows, cols) { + Ok(g) => g, + Err(_) => { + let _ = raw.shutdown(std::net::Shutdown::Both); + return; + } + }; + let mut input = raw; + let mut chunk = [0u8; 8192]; + loop { + match input.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => { + if session.write_input(&chunk[..n]).is_err() { + break; + } + } + } + } + // Only the attachment this pump owns is cleared; a taken-over pump's + // detach was the takeover itself, logged from the other side. + if session.detach_generation(generation) { + let mut c = core.lock().unwrap(); + let _ = c.append("session/detached", json!({ "session": session.id })); + } + } } +} + +fn dispatch( + core: &Arc>, + req: &Request, + stream: &UnixStream, +) -> (Response, AfterResponse) { + let ok = |data: Value| Response { + id: req.id.clone(), + status: Status::Ok, + reason: None, + data: Some(data), + }; + let refused = |why: String| Response { + id: req.id.clone(), + status: Status::Refused, + reason: Some(why), + data: None, + }; + let errored = |why: String| Response { + id: req.id.clone(), + status: Status::Error, + reason: Some(why), + data: None, + }; + + match Verb::parse(&req.verb) { + Some(Verb::Status) => { + let c = core.lock().unwrap(); + let live = c.sessions.values().filter(|s| s.is_live()).count(); + ( + ok(json!({ + "pid": process::id(), + "version": env!("CARGO_PKG_VERSION"), + "suite": c.suite.as_ref().map(|s| s.as_str().to_string()), + "root": c.root.display().to_string(), + "socket": socket_path(&c.root).display().to_string(), + "started_at": c.started_at, + "uptime_secs": c.booted.elapsed().as_secs(), + "events": c.next_seq, + "sessions": { "total": c.sessions.len(), "live": live }, + })), + AfterResponse::Done, + ) + } + Some(Verb::Events) => { + let since = req.args.get("since").and_then(Value::as_u64).unwrap_or(0); + let path = { + let c = core.lock().unwrap(); + events_path(&c.root) + }; + match read_events(&path, since) { + Ok(data) => (ok(data), AfterResponse::Done), + Err(why) => (errored(why), AfterResponse::Done), + } + } + Some(Verb::Stop) => { + let mut c = core.lock().unwrap(); + match c.append("daemon/stopped", json!({ "pid": process::id() })) { + Ok(()) => (ok(json!({ "stopping": true })), AfterResponse::Stop), + Err(why) => (errored(why), AfterResponse::Done), + } + } - fn dispatch(&mut self, req: &Request) -> (Response, Handled) { - // The verb is parsed, not string-matched (R5): the enum makes a new verb a - // compile-forced decision here, and the refusal derives its list from the same - // spelling the parser uses. - match Verb::parse(&req.verb) { - Some(Verb::Status) => (self.ok(req, self.status_data()), Handled::Continue), - Some(Verb::Events) => { - let since = req.args.get("since").and_then(Value::as_u64).unwrap_or(0); - match self.read_events(since) { - Ok(data) => (self.ok(req, data), Handled::Continue), - Err(why) => (self.error(req, why), Handled::Continue), + Some(Verb::Spawn) => { + // Typed decode first (R3: one spelling, both sides), judged strictly after + // — a missing required key refuses naming the FIELD, never a rule that did + // not actually fire. + let parsed: SpawnArgs = match serde_json::from_value(req.args.clone()) { + Ok(a) => a, + Err(e) => return (refused(format!("spawn args: {e}")), AfterResponse::Done), + }; + let test_ok = std::env::var(TEST_AGENT_ENV).is_ok_and(|v| v == "1"); + let agent = match AgentKind::parse(&parsed.agent, test_ok) { + Ok(a) => a, + Err(why) => return (refused(why), AfterResponse::Done), + }; + let cwd = parsed.cwd.as_str(); + if !cwd.starts_with('/') || !PathBuf::from(cwd).is_dir() { + return ( + refused(format!( + "cwd must be an absolute path to an existing directory, got {cwd:?}" + )), + AfterResponse::Done, + ); + } + let prompt = match parsed.prompt_file.as_deref() { + None => None, + Some(p) => match fs::read_to_string(p) { + // The file must outlive the spawn (helm #93) — read it now, refuse + // loudly if it is not there, never pass it through argv. + Ok(text) => Some(text), + Err(e) => { + return ( + refused(format!("cannot read prompt_file {p:?}: {e}")), + AfterResponse::Done, + ); + } + }, + }; + let rows = parsed.rows.unwrap_or(40); + let cols = parsed.cols.unwrap_or(140); + let spec = SpawnSpec { + agent, + cwd: parsed.cwd.clone(), + model: parsed.model.clone(), + effort: parsed.effort.clone(), + runtime_session: agent.mints_session_id().then(mint_session_id), + resume: false, + }; + let (id, notices) = { + let mut c = core.lock().unwrap(); + let id = format!("s{}", c.next_session); + c.next_session += 1; + (id, c.notices.clone()) + }; + let session = match Session::spawn(id.clone(), &spec, rows, cols, notices) { + Ok(s) => s, + Err(why) => return (errored(why), AfterResponse::Done), + }; + { + let mut c = core.lock().unwrap(); + c.sessions.insert(id.clone(), Arc::clone(&session)); + if let Err(why) = c.append( + "session/spawned", + json!({ + "session": id, + "agent": agent.name(), + "cwd": spec.cwd, + "pid": session.pid, + "runtime_session": spec.runtime_session, + "model": spec.model, + "effort": spec.effort, + }), + ) { + return (errored(why), AfterResponse::Done); } } - Some(Verb::Stop) => { - // Logged before answered: the record must already say "stopped" when the - // caller is told it worked (bench-visible means logged). - match self.append("daemon/stopped", json!({ "pid": process::id() })) { - Ok(()) => (self.ok(req, json!({ "stopping": true })), Handled::Stop), - Err(why) => (self.error(req, why), Handled::Continue), + // Ready wait and prompt delivery happen WITHOUT the core lock. + let mut ready = true; + let mut prompt_delivered = false; + if let Some(text) = prompt { + ready = session.wait_ready(READY_WAIT); + if ready { + let one_line = text.replace('\n', " "); + prompt_delivered = session.deliver_line(one_line.trim()).is_ok(); + let mut c = core.lock().unwrap(); + let _ = c.append("session/prompted", json!({ "session": session.id })); } } - None => ( - Response { - id: req.id.clone(), - status: Status::Refused, - reason: Some(format!( - "unknown verb {:?} — this daemon answers: {}", - req.verb, - KNOWN_VERBS.join(", ") + ( + ok(json!({ + "session": session.id, + "pid": session.pid, + "agent": agent.name(), + "runtime_session": session.runtime_session, + "ready": ready, + "prompt_delivered": prompt_delivered, + })), + AfterResponse::Done, + ) + } + + Some(Verb::Sessions) => { + let c = core.lock().unwrap(); + let list: Vec = c + .sessions + .values() + .map(|s| { + json!({ + "session": s.id, + "agent": s.agent.name(), + "cwd": s.cwd, + "pid": s.pid, + "live": s.is_live(), + "attached": s.is_attached(), + "output_bytes": s.output_bytes(), + "runtime_session": s.runtime_session, + "uptime_secs": s.spawned_at.elapsed().as_secs(), + }) + }) + .collect(); + (ok(json!({ "sessions": list })), AfterResponse::Done) + } + + Some(Verb::Attach) => { + let parsed: SessionArgs = match serde_json::from_value(req.args.clone()) { + Ok(a) => a, + Err(e) => return (refused(format!("attach args: {e}")), AfterResponse::Done), + }; + let sid = parsed.session.as_str(); + let rows = parsed.rows.unwrap_or(0); + let cols = parsed.cols.unwrap_or(0); + let session = { + let c = core.lock().unwrap(); + c.sessions.get(sid).cloned() + }; + let Some(session) = session else { + return ( + refused(format!("no session {sid:?} — `bench sessions` lists them")), + AfterResponse::Done, + ); + }; + if !session.is_live() { + return ( + refused(format!( + "session {sid} has exited — `bench resume {sid}` re-enters it where the runtime supports that" )), - data: None, + AfterResponse::Done, + ); + } + let raw = match stream.try_clone() { + Ok(s) => s, + Err(e) => { + return ( + errored(format!("cannot clone stream: {e}")), + AfterResponse::Done, + ); + } + }; + { + let mut c = core.lock().unwrap(); + let _ = c.append("session/attached", json!({ "session": session.id })); + } + ( + ok(json!({ + "session": session.id, + "detach": "Ctrl-\\", + })), + AfterResponse::Pump { + session, + stream: raw, + rows, + cols, }, - Handled::Continue, - ), + ) } - } - fn status_data(&self) -> Value { - json!({ - "pid": process::id(), - "version": env!("CARGO_PKG_VERSION"), - "suite": self.suite.as_ref().map(|s| s.as_str()), - "root": self.root.display().to_string(), - "socket": socket_path(&self.root).display().to_string(), - "started_at": self.started_at, - "uptime_secs": self.booted.elapsed().as_secs(), - "events": self.next_seq, - }) - } - - /// Read back the log — from the file, not from memory, because the file is the - /// record and this verb is how a reader checks that claim. Caps are reported, never - /// silent: `returned < total` plus `truncated` says exactly what was left out. - fn read_events(&self, since: u64) -> Result { - const MAX_RETURNED: usize = 1000; - let path = events_path(&self.root); - let file = File::open(&path).map_err(|e| format!("cannot open {}: {e}", path.display()))?; - let mut events: Vec = Vec::new(); - let mut total = 0u64; - for line in BufReader::new(file).lines() { - let line = line.map_err(|e| format!("cannot read {}: {e}", path.display()))?; - if line.trim().is_empty() { - continue; + Some(Verb::Close) => { + let parsed: SessionArgs = match serde_json::from_value(req.args.clone()) { + Ok(a) => a, + Err(e) => return (refused(format!("close args: {e}")), AfterResponse::Done), + }; + let sid = parsed.session.as_str(); + let session = { + let mut c = core.lock().unwrap(); + c.sessions.remove(sid) + }; + let Some(session) = session else { + return ( + refused(format!("no session {sid:?} — `bench sessions` lists them")), + AfterResponse::Done, + ); + }; + { + let mut c = core.lock().unwrap(); + if let Err(why) = c.append("session/closed", json!({ "session": sid })) { + return (errored(why), AfterResponse::Done); + } } - let ev: Event = serde_json::from_str(&line) - .map_err(|e| format!("corrupt event in {}: {e}", path.display()))?; - if ev.seq < since { - continue; + let was_live = session.close(Duration::from_secs(2)); + ( + ok(json!({ "session": sid, "was_live": was_live })), + AfterResponse::Done, + ) + } + + Some(Verb::Resume) => { + let parsed: SessionArgs = match serde_json::from_value(req.args.clone()) { + Ok(a) => a, + Err(e) => return (refused(format!("resume args: {e}")), AfterResponse::Done), + }; + let sid = parsed.session.as_str(); + let old = { + let c = core.lock().unwrap(); + c.sessions.get(sid).cloned() + }; + let Some(old) = old else { + return ( + refused(format!( + "no session {sid:?} in this daemon's lifetime — resume across a daemon restart is not built yet" + )), + AfterResponse::Done, + ); + }; + if old.is_live() { + return ( + refused(format!( + "session {sid} is still live — `bench attach {sid}` instead" + )), + AfterResponse::Done, + ); } - total += 1; - if events.len() < MAX_RETURNED { - events.push(ev); + let mut spec = old.spec.clone(); + spec.resume = true; + let (id, notices) = { + let mut c = core.lock().unwrap(); + let id = format!("s{}", c.next_session); + c.next_session += 1; + (id, c.notices.clone()) + }; + let session = match Session::spawn(id.clone(), &spec, 40, 140, notices) { + Ok(s) => s, + Err(why) => return (refused(why), AfterResponse::Done), + }; + { + let mut c = core.lock().unwrap(); + c.sessions.remove(sid); + c.sessions.insert(id.clone(), Arc::clone(&session)); + if let Err(why) = c.append( + "session/resumed", + json!({ "session": id, "from": sid, "runtime_session": session.runtime_session }), + ) { + return (errored(why), AfterResponse::Done); + } } + let ready = session.wait_ready(READY_WAIT); + ( + ok(json!({ + "session": session.id, + "from": sid, + "pid": session.pid, + "ready": ready, + })), + AfterResponse::Done, + ) } - let returned = events.len(); - Ok(json!({ - "events": events, - "total": total, - "returned": returned, - "truncated": (returned as u64) < total, - })) - } - fn append(&mut self, kind: &str, data: Value) -> Result<(), String> { - let event = Event { - seq: self.next_seq, - at: now_rfc3339(), - kind: kind.to_string(), - data, - }; - let mut line = - serde_json::to_string(&event).map_err(|e| format!("cannot encode event: {e}"))?; - line.push('\n'); - self.log - .write_all(line.as_bytes()) - .and_then(|()| self.log.flush()) - .map_err(|e| format!("cannot append to event log: {e}"))?; - // Best-effort durability: the record is the point of this process. A failed - // sync is not a failed append — the bytes are handed off either way. - let _ = self.log.sync_data(); - self.next_seq += 1; - Ok(()) + None => ( + refused(format!( + "unknown verb {:?} — this daemon answers: {}", + req.verb, + KNOWN_VERBS.join(", ") + )), + AfterResponse::Done, + ), } +} - fn ok(&self, req: &Request, data: Value) -> Response { - Response { - id: req.id.clone(), - status: Status::Ok, - reason: None, - data: Some(data), +/// Report caps, never hide them: `returned < total` plus `truncated` says exactly what +/// was left out. +fn read_events(path: &PathBuf, since: u64) -> Result { + const MAX_RETURNED: usize = 1000; + let file = File::open(path).map_err(|e| format!("cannot open {}: {e}", path.display()))?; + let mut events: Vec = Vec::new(); + let mut total = 0u64; + for line in BufReader::new(file).lines() { + let line = line.map_err(|e| format!("cannot read {}: {e}", path.display()))?; + if line.trim().is_empty() { + continue; } - } - - fn error(&self, req: &Request, why: String) -> Response { - Response { - id: req.id.clone(), - status: Status::Error, - reason: Some(why), - data: None, + let ev: Event = serde_json::from_str(&line) + .map_err(|e| format!("corrupt event in {}: {e}", path.display()))?; + if ev.seq < since { + continue; + } + total += 1; + if events.len() < MAX_RETURNED { + events.push(ev); } } + let returned = events.len(); + Ok(json!({ + "events": events, + "total": total, + "returned": returned, + "truncated": (returned as u64) < total, + })) } -enum Handled { - Continue, - Stop, +fn respond(stream: &UnixStream, response: &Response) { + respond_keep_open(stream, response); + let _ = stream.shutdown(std::net::Shutdown::Both); } -fn respond(mut stream: &UnixStream, response: &Response) { +fn respond_keep_open(mut stream: &UnixStream, response: &Response) { if let Ok(mut line) = serde_json::to_string(response) { line.push('\n'); let _ = stream.write_all(line.as_bytes()); } - let _ = stream.shutdown(std::net::Shutdown::Both); } fn now_rfc3339() -> String { diff --git a/daemon/direction.md b/daemon/direction.md index c02d9af..286c31a 100644 --- a/daemon/direction.md +++ b/daemon/direction.md @@ -14,10 +14,15 @@ 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.** A suite-aware record root, an append-only event log, one unix -socket, three verbs (`status`, `events`, `stop`), a CLI speaking helm's exit-code -discipline, and a conformance gate that runs the real binaries. Nothing helm does today -is owned here yet — M1 (attention queue and taps) is the first capability helm never had. +**Where it stands: 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 +and effort flags spelled once in `bench-session`, prompt by file, runtime session id +minted at spawn; `attach` is a dtach-grade raw relay with ring replay and Ctrl-\ detach; +`close` is drain-then-die; `resume` re-enters an exited session where the runtime mints +its id (claude, pi — codex refuses with the reason). Nothing helm does today is owned +here yet; mail is next, waking agents by pasting into ptys this daemon now owns. ## The spine diff --git a/daemon/justfile b/daemon/justfile index bb7ed76..adaa7e6 100644 --- a/daemon/justfile +++ b/daemon/justfile @@ -66,17 +66,25 @@ mail-send to subject body-file: mail-list: {{BENCH}} mail/list -# M3 — spawn an interactive agent into a bench pty. Prompt by FILE, never argv (#93). -spawn agent="claude" cwd="." prompt-file="": - {{BENCH}} spawn --agent {{agent}} --cwd {{cwd}} --prompt-file {{prompt-file}} +# M5a — spawn an interactive agent into a bench pty. Prompt by FILE, never argv (#93). +spawn agent="claude" cwd=justfile_directory() prompt-file="": + {{BENCH}} spawn --agent {{agent}} --cwd {{cwd}} {{ if prompt-file != "" { "--prompt-file " + prompt-file } else { "" } }} -# M3 — close a pane; the busy refusal and no-force-over-keyboard rules ride along. -close pane: - {{BENCH}} close --pane {{pane}} +# M5a — list bench sessions. +sessions: + {{BENCH}} sessions -# M5 — attach this terminal to a bench pty. -attach pane: - {{BENCH}} attach --pane {{pane}} +# M5a — attach this terminal to a session's pty (Ctrl-\ detaches). +attach session: + {{BENCH}} attach {{session}} + +# M5a — drain-then-die a session. +close session: + {{BENCH}} close {{session}} + +# M5a — re-enter an exited session's runtime state (claude and pi). +resume session: + {{BENCH}} resume {{session}} # ---- the scoreboard --------------------------------------------------------- @@ -88,8 +96,9 @@ spec: (timeout 60 cargo run -q -p benchd &) sleep 1 echo "bench surface, $(date -u +%Y-%m-%dT%H:%MZ):" - for verb in status events stop:probe spawn close select name capture cmd \ - attn/post attn/list attn/ack watch mail/send mail/list mail/read attach; do + for verb in status events stop:probe spawn sessions attach close resume \ + select name capture cmd \ + attn/post attn/list attn/ack watch mail/send mail/list mail/read; do v=${verb%:probe} if [ "$verb" = "stop:probe" ]; then # probed last for real; listed here so the row exists @@ -102,3 +111,40 @@ spec: fi done cargo run -q -p bench -- stop >/dev/null 2>&1 || true + +# The live proof, repeatable: spawn a real claude through the daemon, attach over the +# socket, demand arithmetic back through the relay. Costs one small claude turn — not +# for CI, for whoever asks "did we actually run this?". +live-smoke: + #!/usr/bin/env bash + set -euo pipefail + SMOKE=$(mktemp -d "${TMPDIR:-/tmp}/blive.XXXX") + export BENCH_DIR="$SMOKE/r" + cargo build -q --workspace + (timeout 180 ./target/debug/benchd &) + sleep 1 + printf 'Reply with exactly BENCHD--OK where is 6*7. Use no tools, output nothing else.' > "$SMOKE/p.txt" + ./target/debug/bench spawn --agent claude --cwd "$PWD" --prompt-file "$SMOKE/p.txt" >/dev/null + python3 - "$BENCH_DIR/benchd.sock" <<'PY' + import socket, sys, re, time, json + s = socket.socket(socket.AF_UNIX); s.connect(sys.argv[1]) + s.sendall(b'{"id":"live-1","verb":"attach","args":{"session":"s1"}}\n') + line = b"" + while not line.endswith(b"\n"): line += s.recv(1) + assert json.loads(line)["status"] == "ok", line + s.settimeout(0.5); buf = b""; deadline = time.time() + 60 + while time.time() < deadline: + try: + c = s.recv(8192) + if not c: break + buf += c + except socket.timeout: pass + t = re.sub(rb'\x1b\][^\x07\x1b]*(\x07|\x1b\\)|\x1b\[[0-9;?]*[a-zA-Z]', b'', buf) + if b'BENCHD-42-OK' in re.sub(rb'\s', b'', t): + print("live-smoke: BENCHD-42-OK through the relay —", len(buf), "bytes"); sys.exit(0) + sys.exit("live-smoke: computed reply never arrived") + PY + ./target/debug/bench close s1 >/dev/null + ./target/debug/bench stop >/dev/null + rm -rf "$SMOKE" + echo "live-smoke: clean"