brokerd: state.sock answers memoryd's question about a session's state

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

crates/brokerd/src/config.rs

@@ -25,12 +25,14 @@ impl Default for Paths {
     }
 }
 
-/// An empty path means "the default under `home`".
+/// The sockets `brokerd` serves. An empty path means "the default under `home`".
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Default)]
 #[serde(deny_unknown_fields, default)]
 pub struct Sockets {
     pub broker: PathBuf,
     pub admin: PathBuf,
+    /// `memoryd`'s session-state questions (M5c spec, section 7).
+    pub state: PathBuf,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
@@ -284,6 +286,14 @@ impl Config {
             self.sockets.admin.clone()
         }
     }
+    /// `state.sock`, where `memoryd` asks for a session's state.
+    pub fn state_socket(&self) -> PathBuf {
+        if self.sockets.state.as_os_str().is_empty() {
+            self.paths.home.join("run/memory-broker/state.sock")
+        } else {
+            self.sockets.state.clone()
+        }
+    }
     /// `gateway.sock`, when `[gateway]` is set.
     pub fn gateway_socket(&self) -> Option<PathBuf> {
         let gateway = self.gateway.as_ref()?;

crates/brokerd/src/lib.rs

@@ -18,3 +18,4 @@ pub mod policy;
 pub mod runner;
 pub mod serve;
 pub mod state;
+pub mod state_socket;

crates/brokerd/src/main.rs

@@ -1,4 +1,4 @@
-//! `brokerd`: the broker role. `serve` takes the audit lock, binds both sockets, and serves.
+//! `brokerd`: the broker role. `serve` takes the audit lock, binds its three sockets, and serves.
 
 use std::ffi::OsString;
 use std::path::PathBuf;
@@ -58,6 +58,7 @@ fn main() -> ExitCode {
     // Keep the socket paths for the serving notice, before `cfg` is moved into `start`.
     let broker_path = cfg.broker_socket();
     let admin_path = cfg.admin_socket();
+    let state_path = cfg.state_socket();
 
     // The runtime and the line that says which. A `[runner]` runs calls in containers; without it,
     // every call is refused, as in M3a.
@@ -117,6 +118,7 @@ fn main() -> ExitCode {
         broker_path.display(),
         admin_path.display()
     );
+    eprintln!("brokerd: serving session state on {}", state_path.display());
     eprintln!("brokerd: {runtime_notice}");
 
     match started.run() {

crates/brokerd/src/serve.rs

@@ -1,4 +1,4 @@
-//! `brokerd serve`: bring the broker up, listen on both sockets, and keep it running.
+//! `brokerd serve`: bring the broker up, listen on its three sockets, and keep it running.
 
 use std::fs::{DirBuilder, Permissions};
 use std::io;
@@ -16,6 +16,7 @@ use crate::config::Config;
 use crate::ledger::Ledger;
 use crate::runner::Runtime;
 use crate::state::StateStore;
+use crate::state_socket;
 use bxcore::{ChainFailure, Timestamp};
 
 /// A failure while `brokerd serve` is starting up.
@@ -48,16 +49,17 @@ impl std::error::Error for ServeError {
     }
 }
 
-/// The result of `start`: a running broker with both sockets open.
+/// The result of `start`: a running broker with its three sockets open.
 pub struct Started {
     pub broker: Arc<Broker>,
     pub recovered: bool,
     pub accepted: Option<ChainFailure>,
     tools: UnixListener,
     admin: UnixListener,
+    state: UnixListener,
 }
 
-/// Start the broker: open the audit chain, bind both sockets, and build the ledger and broker.
+/// Start the broker: open the audit chain, bind the three sockets, and build the ledger and broker.
 pub fn start(
     cfg: Config,
     accept_break: bool,
@@ -67,7 +69,9 @@ pub fn start(
     // The audit lock comes first: if the chain will not open, no directory or socket is made.
     let opened = Writer::open(&cfg.audit_dir(), accept_break).map_err(ServeError::Audit)?;
 
-    // Both sockets, in the order the task gives.
+    // The three sockets. `state.sock` comes first: whoever waits for the tool and admin sockets
+    // then finds it bound too.
+    let state = listen(&cfg.state_socket())?;
     let tools = listen(&cfg.broker_socket())?;
     let admin = listen(&cfg.admin_socket())?;
 
@@ -93,6 +97,7 @@ pub fn start(
         accepted: opened.accepted.map(|b| *b),
         tools,
         admin,
+        state,
     })
 }
 
@@ -156,6 +161,7 @@ fn listen(socket: &Path) -> Result<UnixListener, ServeError> {
 enum Which {
     Tools,
     Admin,
+    State,
 }
 
 /// Run the broker: start the expiry thread, one accept thread per socket, then wait for the
@@ -166,6 +172,7 @@ impl Started {
             broker,
             tools,
             admin,
+            state,
             ..
         } = self;
 
@@ -187,7 +194,8 @@ impl Started {
         let (tx, rx) = mpsc::channel::<io::Error>();
 
         accept_loop(tools, Arc::clone(&broker), tx.clone(), Which::Tools)?;
-        accept_loop(admin, Arc::clone(&broker), tx, Which::Admin)?;
+        accept_loop(admin, Arc::clone(&broker), tx.clone(), Which::Admin)?;
+        accept_loop(state, Arc::clone(&broker), tx, Which::State)?;
 
         // The first listener error stops the daemon.
         match rx.recv() {
@@ -200,7 +208,7 @@ impl Started {
 /// Accept streams from one listener forever, starting a handler thread per stream. When the
 /// system is out of file descriptors or memory, the listener pauses and tries again instead of
 /// stopping: otherwise a peer holding many idle connections could stop the daemon, and the other
-/// socket with it. Any other error stops the daemon through `tx`.
+/// sockets with it. Any other error stops the daemon through `tx`.
 fn accept_loop(
     listener: UnixListener,
     broker: Arc<Broker>,
@@ -210,6 +218,7 @@ fn accept_loop(
     let name = match which {
         Which::Tools => "accept-broker",
         Which::Admin => "accept-admin",
+        Which::State => "accept-state",
     };
     std::thread::Builder::new()
         .name(name.to_string())
@@ -248,6 +257,7 @@ fn serve_one(stream: UnixStream, broker: &Arc<Broker>, which: Which) {
     let started = std::thread::Builder::new().spawn(move || match which {
         Which::Tools => broker::handle(stream, &for_thread),
         Which::Admin => admin::handle(stream, &for_thread),
+        Which::State => state_socket::handle(stream, &for_thread),
     });
     if let Err(e) = started {
         broker.log(&format!(

crates/brokerd/src/state_socket.rs

@@ -0,0 +1,51 @@
+//! One `state.sock` connection: `memoryd` asks for a session's state when a turn ends, and the
+//! answer is what the state store holds (M5c spec, section 7). It reads, and never changes
+//! anything: no state, no audit record.
+
+use std::os::unix::net::UnixStream;
+
+use bxcore::{ErrorCode, Message, SessionStateReply, WireError};
+
+use crate::broker::{Broker, forbid, read_request, send};
+use crate::state::StateStore;
+
+/// The socket's name in the lines `brokerd` logs and the errors it sends.
+pub const SOCKET: &str = "state.sock";
+
+/// Serve one `state.sock` connection from `memoryd`: one question, one final answer, then the
+/// connection is closed.
+pub fn handle(mut stream: UnixStream, broker: &Broker) {
+    // 1. The peer is gone before it speaks, or its frame is malformed: answered, then return.
+    let Some(envelope) = read_request(&mut stream) else {
+        return;
+    };
+    let id = envelope.id;
+    let msg = envelope.msg;
+
+    // 2. Only the question belongs here. `bxcore::SessionState` is matched through `Message`, so
+    //    this file names neither it nor `policy::SessionState`, which the store returns.
+    let Message::SessionState(question) = msg else {
+        forbid(broker, &mut stream, id, &msg, SOCKET);
+        return;
+    };
+
+    // 3. The store the ledger writes. A session without a file is the store's default, `private`
+    //    and trusted; a file that cannot be read is an error, never that default.
+    let store = StateStore::new(&broker.cfg().state_dir());
+    let response = match store.read(&question.session) {
+        Ok(state) => Message::SessionStateReply(SessionStateReply {
+            class: state.taint,
+            untrusted: state.untrusted,
+        }),
+        Err(e) => {
+            broker.log(&format!("brokerd: {e}"));
+            Message::Error(WireError {
+                code: ErrorCode::Internal,
+                detail: e.to_string(),
+            })
+        }
+    };
+
+    // 4. The final frame; a failed send is ignored, as the peer is gone.
+    let _ = send(&mut stream, id, true, response);
+}

crates/brokerd/tests/config.rs

@@ -41,6 +41,10 @@ fn sockets_and_directories_default_to_places_under_home() {
         c.admin_socket(),
         PathBuf::from("/srv/boxmaker/run/owner-broker/admin.sock")
     );
+    assert_eq!(
+        c.state_socket(),
+        PathBuf::from("/srv/boxmaker/run/memory-broker/state.sock")
+    );
     assert_eq!(c.audit_dir(), PathBuf::from("/srv/boxmaker/audit"));
     assert_eq!(
         c.state_dir(),
@@ -57,6 +61,7 @@ fn every_key_can_be_set() {
     assert_eq!(c.paths.grants, PathBuf::from("/srv/boxmaker-grants"));
     assert_eq!(c.broker_socket(), PathBuf::from("/run/bx/broker.sock"));
     assert_eq!(c.admin_socket(), PathBuf::from("/run/bx/admin.sock"));
+    assert_eq!(c.state_socket(), PathBuf::from("/run/bx/state.sock"));
     assert_eq!(c.approvals.ttl_ms, 60_000);
     // The two directories always follow home.
     assert_eq!(c.audit_dir(), PathBuf::from("/srv/boxmaker/audit"));
@@ -75,6 +80,21 @@ fn one_socket_set_leaves_the_other_at_its_default() {
         c.broker_socket(),
         PathBuf::from("/h/run/loop-broker/broker.sock")
     );
+    assert_eq!(
+        c.state_socket(),
+        PathBuf::from("/h/run/memory-broker/state.sock")
+    );
+    let c =
+        Config::parse("[paths]\nhome = \"/h\"\n[sockets]\nstate = \"/x/state.sock\"\n").unwrap();
+    assert_eq!(c.state_socket(), PathBuf::from("/x/state.sock"));
+    assert_eq!(
+        c.broker_socket(),
+        PathBuf::from("/h/run/loop-broker/broker.sock")
+    );
+    assert_eq!(
+        c.admin_socket(),
+        PathBuf::from("/h/run/owner-broker/admin.sock")
+    );
 }
 
 #[test]

crates/brokerd/tests/fixtures/config/full.toml

@@ -6,6 +6,7 @@ grants = "/srv/boxmaker-grants"
 [sockets]
 broker = "/run/bx/broker.sock"
 admin = "/run/bx/admin.sock"
+state = "/run/bx/state.sock"
 
 [approvals]
 ttl_ms = 60000

crates/brokerd/tests/serve.rs

@@ -63,7 +63,9 @@ impl Home {
         command
     }
 
-    /// Starts `brokerd serve` and waits until both sockets answer.
+    /// Starts `brokerd serve` and waits until both sockets answer. A socket accepts connections
+    /// as soon as it is bound, before `brokerd` has printed its startup notices; the admin
+    /// socket's answer comes from a handler thread, which runs only after they are printed.
     fn serve(&self) -> Running {
         let child = self
             .command(&[])
@@ -79,6 +81,11 @@ impl Home {
             assert!(Instant::now() < until, "brokerd never listened");
             std::thread::sleep(Duration::from_millis(20));
         }
+        let answer = exchange(&self.admin(), 1, Message::Approvals(Empty {}));
+        assert!(
+            matches!(&answer[0].msg, Message::ApprovalList(_)),
+            "{answer:?}"
+        );
         running
     }
 

crates/brokerd/tests/serve_state.rs

@@ -0,0 +1,269 @@
+//! `brokerd serve` answers `memoryd`'s session-state question on its third socket,
+//! `run/memory-broker/state.sock` or `[sockets] state` (M5c spec, section 7). Do not edit.
+
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::os::unix::fs::PermissionsExt;
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command, Stdio};
+use std::time::{Duration, Instant};
+
+use bxcore::{
+    DataClass, Envelope, ErrorCode, Message, PROTOCOL_VERSION, SessionId,
+    SessionState as StateQuestion, SessionStateReply,
+};
+use tmp::TempDir;
+
+struct Home {
+    dir: TempDir,
+    config: PathBuf,
+}
+
+impl Home {
+    /// A home whose config has `sockets` as the body of its `[sockets]` table.
+    fn new(tag: &str, sockets: &str) -> Home {
+        let dir = TempDir::new(tag);
+        std::fs::create_dir_all(dir.path().join("grants")).unwrap();
+        let text = format!(
+            "[paths]\nhome = \"{home}\"\ngrants = \"{home}/grants\"\n[sockets]\n{sockets}\n",
+            home = dir.path().display()
+        );
+        let config = dir.write("brokerd.toml", &text);
+        Home { dir, config }
+    }
+
+    fn path(&self, relative: &str) -> PathBuf {
+        self.dir.path().join(relative)
+    }
+
+    fn state(&self) -> PathBuf {
+        self.path("run/memory-broker/state.sock")
+    }
+
+    /// Writes the session's state file, as the ledger would.
+    fn session(&self, id: &str, text: &str) {
+        std::fs::create_dir_all(self.path("broker/sessions")).unwrap();
+        std::fs::write(self.path(&format!("broker/sessions/{id}.json")), text).unwrap();
+    }
+
+    /// Starts `brokerd serve` and waits until the tool socket, the admin socket and `state`
+    /// accept a connection.
+    fn serve(&self, state: &Path) -> Running {
+        let child = Command::new(env!("CARGO_BIN_EXE_brokerd"))
+            .args(["serve", "--config"])
+            .arg(&self.config)
+            .stdout(Stdio::null())
+            .stderr(Stdio::piped())
+            .spawn()
+            .unwrap();
+        let running = Running(Some(child));
+        let until = Instant::now() + Duration::from_secs(10);
+        for socket in [
+            self.path("run/loop-broker/broker.sock"),
+            self.path("run/owner-broker/admin.sock"),
+            state.to_path_buf(),
+        ] {
+            while UnixStream::connect(&socket).is_err() {
+                assert!(
+                    Instant::now() < until,
+                    "brokerd never listened on {}",
+                    socket.display()
+                );
+                std::thread::sleep(Duration::from_millis(20));
+            }
+        }
+        running
+    }
+}
+
+/// Kills the daemon when dropped, and waits for it; `stop` returns what it printed.
+struct Running(Option<Child>);
+
+impl Running {
+    fn stop(mut self) -> String {
+        let mut child = self.0.take().unwrap();
+        child.kill().unwrap();
+        let output = child.wait_with_output().unwrap();
+        String::from_utf8_lossy(&output.stderr).to_string()
+    }
+}
+
+impl Drop for Running {
+    fn drop(&mut self) {
+        if let Some(child) = &mut self.0 {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+    }
+}
+
+fn mode(path: &Path) -> u32 {
+    std::fs::metadata(path).unwrap().permissions().mode() & 0o777
+}
+
+/// A connection with a ten-second read limit, so a daemon that never answers fails the test.
+fn connect(socket: &Path) -> UnixStream {
+    let stream = UnixStream::connect(socket).unwrap();
+    stream
+        .set_read_timeout(Some(Duration::from_secs(10)))
+        .unwrap();
+    stream
+}
+
+/// Sends one message under `id` and returns the one final frame that answers it.
+fn exchange(socket: &Path, id: u64, msg: Message) -> Envelope {
+    let mut stream = connect(socket);
+    bxcore::write_frame(
+        &mut stream,
+        &Envelope {
+            v: PROTOCOL_VERSION,
+            id,
+            r#final: true,
+            msg,
+        },
+    )
+    .unwrap();
+    let env = bxcore::read_frame(&mut stream).unwrap();
+    assert_eq!((env.id, env.r#final), (id, true), "{env:?}");
+    env
+}
+
+fn question(session: &str) -> Message {
+    Message::SessionState(StateQuestion {
+        session: SessionId::new(session).unwrap(),
+    })
+}
+
+fn reply(env: &Envelope) -> SessionStateReply {
+    match &env.msg {
+        Message::SessionStateReply(reply) => reply.clone(),
+        other => panic!("wanted a session_state_reply, got {other:?}"),
+    }
+}
+
+#[test]
+fn it_answers_on_its_default_path_from_the_state_files() {
+    let home = Home::new("serve-state-answers", "");
+    home.session("s1", "{\"taint\":\"secret\",\"untrusted\":true}\n");
+    home.session("s2", "{\"taint\":\"private\",\"untrusted\":true}\n");
+    let running = home.serve(&home.state());
+    for (id, session, class, untrusted) in [
+        (1, "s1", DataClass::Secret, true),
+        (2, "s2", DataClass::Private, true),
+        (3, "unseen", DataClass::Private, false),
+    ] {
+        assert_eq!(
+            reply(&exchange(&home.state(), id, question(session))),
+            SessionStateReply { class, untrusted },
+            "{session}"
+        );
+    }
+    assert!(!home.path("broker/sessions/unseen.json").exists());
+    let printed = running.stop();
+    assert!(
+        printed.contains(&format!(
+            "brokerd: serving session state on {}\n",
+            home.state().display()
+        )),
+        "{printed}"
+    );
+}
+
+/// As for the other two sockets: the directory is made private even when it was there at 0755,
+/// the socket is 0600, and a stale socket left by a killed `brokerd` is replaced.
+#[test]
+fn its_directory_is_0700_its_socket_0600_and_a_stale_one_is_replaced() {
+    let home = Home::new("serve-state-modes", "");
+    std::fs::create_dir_all(home.path("run/memory-broker")).unwrap();
+    std::fs::set_permissions(
+        home.path("run/memory-broker"),
+        std::fs::Permissions::from_mode(0o755),
+    )
+    .unwrap();
+    drop(UnixListener::bind(home.state()).unwrap());
+    assert!(home.state().exists(), "the stale socket file is there");
+    let _running = home.serve(&home.state());
+    assert_eq!(mode(&home.path("run/memory-broker")), 0o700);
+    assert_eq!(mode(&home.state()), 0o600);
+    assert_eq!(
+        reply(&exchange(&home.state(), 9, question("s9"))),
+        SessionStateReply {
+            class: DataClass::Private,
+            untrusted: false,
+        }
+    );
+}
+
+#[test]
+fn sockets_state_in_the_config_moves_it() {
+    let dir = TempDir::new("serve-state-moved-sock");
+    let moved = dir.path().join("st/state.sock");
+    let home = Home::new(
+        "serve-state-moved",
+        &format!("state = \"{}\"", moved.display()),
+    );
+    home.session("s1", "{\"taint\":\"secret\",\"untrusted\":false}\n");
+    let running = home.serve(&moved);
+    assert_eq!(
+        reply(&exchange(&moved, 1, question("s1"))),
+        SessionStateReply {
+            class: DataClass::Secret,
+            untrusted: false,
+        }
+    );
+    assert_eq!(mode(&dir.path().join("st")), 0o700);
+    assert!(
+        !home.path("run/memory-broker").exists(),
+        "nothing at the default"
+    );
+    let printed = running.stop();
+    assert!(
+        printed.contains(&format!(
+            "brokerd: serving session state on {}\n",
+            moved.display()
+        )),
+        "{printed}"
+    );
+}
+
+/// A connection that sends nothing holds only its own thread: the next question is answered.
+#[test]
+fn a_silent_connection_does_not_hold_up_the_next() {
+    let home = Home::new("serve-state-silent", "");
+    let _running = home.serve(&home.state());
+    let _silent = connect(&home.state());
+    assert_eq!(
+        reply(&exchange(&home.state(), 2, question("s2"))),
+        SessionStateReply {
+            class: DataClass::Private,
+            untrusted: false,
+        }
+    );
+}
+
+/// The question belongs on `state.sock` only: the tool and admin sockets refuse it.
+#[test]
+fn the_other_sockets_refuse_the_question() {
+    let home = Home::new("serve-state-others", "");
+    let running = home.serve(&home.state());
+    for (id, socket) in [
+        (4, home.path("run/loop-broker/broker.sock")),
+        (5, home.path("run/owner-broker/admin.sock")),
+    ] {
+        match exchange(&socket, id, question("s1")).msg {
+            Message::Error(e) => assert_eq!(e.code, ErrorCode::Forbidden, "{e:?}"),
+            other => panic!("{other:?}"),
+        }
+    }
+    let printed = running.stop();
+    assert!(
+        printed.contains("session_state on broker.sock\nsee docs/runbook.md#socket-forbidden"),
+        "{printed}"
+    );
+    assert!(
+        printed.contains("session_state on admin.sock\nsee docs/runbook.md#socket-forbidden"),
+        "{printed}"
+    );
+}

crates/brokerd/tests/state_socket.rs

@@ -0,0 +1,371 @@
+//! One `state.sock` connection: `memoryd` asks for a session's state, and `brokerd` answers from
+//! its state store, changing nothing (M5c spec, section 7). Every error is an error frame, never
+//! an answer. Do not edit.
+
+#[path = "support/client.rs"]
+mod client;
+#[path = "support/rig.rs"]
+mod rig;
+#[path = "support/runtime.rs"]
+mod runtime;
+#[path = "support/sink.rs"]
+mod sink;
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::io::Write;
+use std::os::unix::net::UnixStream;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+
+use brokerd::broker::Broker;
+use brokerd::policy::{Label, SessionState};
+use brokerd::state_socket;
+use bxcore::{
+    CallId, DataClass, Empty, Envelope, ErrorCode, FrameError, Message, PROTOCOL_VERSION,
+    SessionId, SessionState as StateQuestion, SessionStateReply, SessionUpdated, ToolRequest,
+    WireError,
+};
+use client::{Serve, next, open};
+use rig::Rig;
+use runtime::Recording;
+
+fn broker(rig: &Rig) -> Arc<Broker> {
+    rig.broker(&Recording::answering("x"))
+}
+
+fn question(session: &str) -> Message {
+    Message::SessionState(StateQuestion {
+        session: SessionId::new(session).unwrap(),
+    })
+}
+
+/// Asks once, under `id`, and returns the one frame that comes back.
+fn ask(broker: &Arc<Broker>, id: u64, msg: Message) -> Envelope {
+    let mut stream = open(broker, state_socket::handle, id, msg);
+    let env = next(&mut stream);
+    assert_eq!((env.id, env.r#final), (id, true), "{env:?}");
+    env
+}
+
+fn reply(env: &Envelope) -> SessionStateReply {
+    match &env.msg {
+        Message::SessionStateReply(reply) => reply.clone(),
+        other => panic!("wanted a session_state_reply, got {other:?}"),
+    }
+}
+
+fn error(env: &Envelope) -> WireError {
+    match &env.msg {
+        Message::Error(e) => e.clone(),
+        other => panic!("wanted an error, got {other:?}"),
+    }
+}
+
+/// Raises a session's state the way the ledger does, through the store's own `raise`.
+fn raise(rig: &Rig, session: &str, class: DataClass, untrusted: bool) {
+    rig.state()
+        .raise(
+            &SessionId::new(session).unwrap(),
+            SessionState::default(),
+            Label { class, untrusted },
+        )
+        .unwrap();
+}
+
+/// Every file under `dir`, with its bytes, sorted; empty when `dir` does not exist.
+fn files(dir: &Path) -> Vec<(String, Vec<u8>)> {
+    let mut all = Vec::new();
+    if let Ok(entries) = std::fs::read_dir(dir) {
+        for entry in entries {
+            let entry = entry.unwrap();
+            all.push((
+                entry.file_name().into_string().unwrap(),
+                std::fs::read(entry.path()).unwrap_or_default(),
+            ));
+        }
+    }
+    all.sort();
+    all
+}
+
+#[test]
+fn the_answer_is_the_stored_class_and_flag() {
+    let rig = Rig::new("state-sock-stored");
+    raise(&rig, "a", DataClass::Private, false);
+    raise(&rig, "b", DataClass::Private, true);
+    raise(&rig, "c", DataClass::Secret, false);
+    raise(&rig, "d", DataClass::Secret, true);
+    // A public result leaves a session at private: a session is never below private.
+    raise(&rig, "e", DataClass::Public, true);
+    let broker = broker(&rig);
+    for (i, (session, class, untrusted)) in [
+        ("a", DataClass::Private, false),
+        ("b", DataClass::Private, true),
+        ("c", DataClass::Secret, false),
+        ("d", DataClass::Secret, true),
+        ("e", DataClass::Private, true),
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let id = 40 + i as u64;
+        assert_eq!(
+            reply(&ask(&broker, id, question(session))),
+            SessionStateReply { class, untrusted },
+            "session {session}"
+        );
+    }
+    assert!(rig.lines.all().is_empty(), "{:?}", rig.lines.all());
+}
+
+/// A file written by hand, as the runbook tells the owner to write one, is read the same way.
+#[test]
+fn a_state_file_the_owner_wrote_is_answered_as_it_says() {
+    let rig = Rig::new("state-sock-owner");
+    std::fs::create_dir_all(rig.cfg.state_dir()).unwrap();
+    std::fs::write(
+        rig.state_file("s1"),
+        "{\"taint\":\"secret\",\"untrusted\":true}\n",
+    )
+    .unwrap();
+    let env = ask(&broker(&rig), 7, question("s1"));
+    assert_eq!(
+        reply(&env),
+        SessionStateReply {
+            class: DataClass::Secret,
+            untrusted: true,
+        }
+    );
+}
+
+/// A session `brokerd` has not seen has made no tool call: `private`, trusted (M5c spec, section
+/// 7). It is answered without making its file, or the directory.
+#[test]
+fn a_session_brokerd_has_not_seen_is_private_and_trusted() {
+    let rig = Rig::new("state-sock-unseen");
+    let broker = broker(&rig);
+    let env = ask(&broker, 3, question("never-seen"));
+    assert_eq!(
+        reply(&env),
+        SessionStateReply {
+            class: DataClass::Private,
+            untrusted: false,
+        }
+    );
+    assert!(!rig.cfg.state_dir().exists(), "asking makes no directory");
+    // With the directory there and another session in it, still no file for this one.
+    raise(&rig, "other", DataClass::Secret, true);
+    let env = ask(&broker, 4, question("never-seen"));
+    assert_eq!(
+        reply(&env),
+        SessionStateReply {
+            class: DataClass::Private,
+            untrusted: false,
+        }
+    );
+    assert!(!rig.state_file("never-seen").exists());
+}
+
+#[test]
+fn asking_changes_nothing() {
+    let rig = Rig::new("state-sock-nothing");
+    raise(&rig, "s1", DataClass::Secret, true);
+    raise(&rig, "s2", DataClass::Private, false);
+    let before = files(&rig.cfg.state_dir());
+    let broker = broker(&rig);
+    for id in 0..3 {
+        for session in ["s1", "s2", "s3"] {
+            ask(&broker, id, question(session));
+        }
+    }
+    assert_eq!(files(&rig.cfg.state_dir()), before);
+    assert!(
+        rig.events().is_empty(),
+        "nothing is recorded for a question"
+    );
+    assert!(rig.lines.all().is_empty(), "{:?}", rig.lines.all());
+}
+
+/// A state file that cannot be read is an `internal` error with the store's reason and runbook
+/// pointer, logged once, never an answer: `memoryd` then leaves the turn undecided.
+#[test]
+fn a_damaged_state_is_an_error_never_an_answer() {
+    let cases: [(&str, Option<&str>); 5] = [
+        ("not json", Some("{\"taint\":")),
+        (
+            "public",
+            Some("{\"taint\":\"public\",\"untrusted\":false}\n"),
+        ),
+        (
+            "unknown key",
+            Some("{\"taint\":\"private\",\"untrusted\":false,\"x\":1}\n"),
+        ),
+        (
+            "unknown class",
+            Some("{\"taint\":\"loud\",\"untrusted\":false}\n"),
+        ),
+        ("a directory", None),
+    ];
+    for (i, (name, text)) in cases.into_iter().enumerate() {
+        let rig = Rig::new("state-sock-damaged");
+        std::fs::create_dir_all(rig.cfg.state_dir()).unwrap();
+        match text {
+            Some(text) => std::fs::write(rig.state_file("s1"), text).unwrap(),
+            None => std::fs::create_dir(rig.state_file("s1")).unwrap(),
+        }
+        let before = files(&rig.cfg.state_dir());
+        let env = ask(&broker(&rig), 60 + i as u64, question("s1"));
+        let e = error(&env);
+        assert_eq!(e.code, ErrorCode::Internal, "{name}: {e:?}");
+        assert!(e.detail.contains("s1.json"), "{name}: {}", e.detail);
+        assert!(
+            e.detail
+                .ends_with("see docs/runbook.md#broker-state-damaged"),
+            "{name}: {}",
+            e.detail
+        );
+        let lines = rig.lines.all();
+        assert_eq!(lines.len(), 1, "{name}: {lines:?}");
+        assert!(lines[0].starts_with("brokerd: "), "{name}: {lines:?}");
+        assert!(lines[0].contains("s1.json"), "{name}: {lines:?}");
+        assert!(
+            lines[0].ends_with("see docs/runbook.md#broker-state-damaged"),
+            "{name}: {lines:?}"
+        );
+        assert_eq!(files(&rig.cfg.state_dir()), before, "{name}");
+    }
+}
+
+#[test]
+fn every_other_kind_is_forbidden_and_logged() {
+    let rig = Rig::new("state-sock-forbidden");
+    raise(&rig, "s1", DataClass::Secret, true);
+    let broker = broker(&rig);
+    let session = SessionId::new("s1").unwrap();
+    let kinds = vec![
+        Message::ToolRequest(ToolRequest {
+            session: session.clone(),
+            call: CallId(1),
+            tool: "read_file".to_string(),
+            arguments: "{}".to_string(),
+        }),
+        Message::Approvals(Empty {}),
+        Message::CheckGrants(Empty {}),
+        Message::Ok(Empty {}),
+        Message::SessionUpdated(SessionUpdated {
+            session: session.clone(),
+            turn: 1,
+        }),
+        // An answer sent as a question is not a question.
+        Message::SessionStateReply(SessionStateReply {
+            class: DataClass::Private,
+            untrusted: false,
+        }),
+    ];
+    let count = kinds.len();
+    for (i, msg) in kinds.into_iter().enumerate() {
+        let id = 100 + i as u64;
+        let e = error(&ask(&broker, id, msg));
+        assert_eq!(e.code, ErrorCode::Forbidden, "{e:?}");
+        assert!(e.detail.ends_with("is not accepted on state.sock"), "{e:?}");
+    }
+    let lines = rig.lines.with("on state.sock");
+    assert_eq!(lines.len(), count, "{:?}", rig.lines.all());
+    for line in &lines {
+        assert!(
+            line.ends_with("\nsee docs/runbook.md#socket-forbidden"),
+            "{line}"
+        );
+    }
+    assert_eq!(
+        rig.lines
+            .with("refused the message kind session_state_reply on state.sock")
+            .len(),
+        1
+    );
+    assert!(rig.events().is_empty(), "nothing is recorded");
+}
+
+/// A frame that is not a well-formed question is answered with the frame error, id 0, as on the
+/// other sockets. A session id that is not one (a path, say) is refused when the frame is read.
+#[test]
+fn a_malformed_frame_is_answered_with_an_error() {
+    let rig = Rig::new("state-sock-malformed");
+    raise(&rig, "s1", DataClass::Secret, true);
+    let broker = broker(&rig);
+    let body = |session: &str, v: u32, extra: &str| {
+        format!(
+            "{{\"v\":{v},\"id\":5,\"final\":true,\"msg\":{{\"kind\":\"session_state\",\
+             \"body\":{{\"session\":\"{session}\"{extra}}}}}}}"
+        )
+    };
+    let cases = [
+        ("not json", "{{{".to_string(), ErrorCode::BadMessage),
+        ("a path", body("../s1", 1, ""), ErrorCode::BadMessage),
+        ("upper case", body("S1", 1, ""), ErrorCode::BadMessage),
+        (
+            "unknown key",
+            body("s1", 1, ",\"x\":1"),
+            ErrorCode::BadMessage,
+        ),
+        ("version 2", body("s1", 2, ""), ErrorCode::BadVersion),
+    ];
+    for (name, text, code) in cases {
+        let (mut client, server) = UnixStream::pair().unwrap();
+        let b = Arc::clone(&broker);
+        std::thread::spawn(move || state_socket::handle(server, &b));
+        client
+            .write_all(&(text.len() as u32).to_be_bytes())
+            .unwrap();
+        client.write_all(text.as_bytes()).unwrap();
+        let env = next(&mut client);
+        assert_eq!((env.id, env.r#final), (0, true), "{name}");
+        assert_eq!(error(&env).code, code, "{name}");
+    }
+    // The well-formed question, for contrast.
+    let env = ask(&broker, 5, question("s1"));
+    assert_eq!(
+        reply(&env),
+        SessionStateReply {
+            class: DataClass::Secret,
+            untrusted: true,
+        }
+    );
+}
+
+/// One question per connection: after its answer the connection is closed, and a second question
+/// on it is never answered.
+#[test]
+fn a_second_question_on_one_connection_is_not_answered() {
+    let rig = Rig::new("state-sock-once");
+    raise(&rig, "s1", DataClass::Secret, true);
+    let broker = broker(&rig);
+    let mut stream = open(&broker, state_socket::handle, 1, question("s1"));
+    assert!(matches!(
+        next(&mut stream).msg,
+        Message::SessionStateReply(_)
+    ));
+    // The handler may have closed its end already, so the second frame may not be written.
+    let _ = bxcore::write_frame(
+        &mut stream,
+        &Envelope {
+            v: PROTOCOL_VERSION,
+            id: 2,
+            r#final: true,
+            msg: question("s1"),
+        },
+    );
+    // Once the peer has closed, macOS refuses the timeout with EINVAL (22); the read cannot block
+    // then.
+    if let Err(e) = stream.set_read_timeout(Some(Duration::from_secs(10))) {
+        assert_eq!(e.raw_os_error(), Some(22), "set_read_timeout: {e}");
+    }
+    match bxcore::read_frame(&mut stream) {
+        Err(FrameError::Closed) => {}
+        // Linux resets a Unix socket closed with unread data in it: closed as well.
+        Err(FrameError::Io(e)) if e.kind() == std::io::ErrorKind::ConnectionReset => {}
+        other => panic!("wanted the connection closed, got {other:?}"),
+    }
+}

docs/implementer-log.md

@@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone.
 
 | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model |
 |---|---|---|---|---|---|---|---|
+| M5c1/05-brokerd-state-socket | 2026-09-28 | done | 2 | pass | none | Copied five files from `docs/plans/M5c1/files/`: `state_socket.rs` and `serve_state.rs` (new) and new versions of `config.rs`, `serve.rs` and `tests/fixtures/config/full.toml` (all now set and check `state`). Step 2 showed exactly the named failures: `config` with four E0599 (`no method named state_socket`) and `state_socket` with one E0432 (`unresolved import brokerd::state_socket`); the new `serve.rs` compiled. `cargo build --tests` reported those two targets only. Made the edits exactly as the task gives: C1 `Sockets` gains `state: PathBuf` with the doc comment, C2 `Config::state_socket` after `admin_socket` returning the default `run/memory-broker/state.sock` when empty; L1 `pub mod state_socket;` after `pub mod state;`. Created `crates/brokerd/src/state_socket.rs` with the given body: `read_request`, match `Message::SessionState(question)` (no `SessionState` import), `StateStore::new(&broker.cfg().state_dir())`, `Ok` -> `SessionStateReply { class: state.taint, untrusted: state.untrusted }`, `Err` -> log `brokerd: {e}` and send `internal` with `e.to_string()`, `forbid` for any other kind; `send` the final frame. V1-V11 in `serve.rs`: doc comment, `use crate::state_socket`, `Started` gains `state: UnixListener`, `state.sock` bound first then tool and admin, `Which::State`, the `run` destructure, the second `accept_loop` with `tx` moved, the `accept-state` name, and `state_socket::handle` in `serve_one`; M1-M3 in `main.rs`: doc comment, `state_path = cfg.state_socket()`, the `serving session state on <path>` notice. R1-R3 in `docs/runbook.md`: `accept-broker (or accept-admin, accept-state)`, the `state.sock` line in `socket-forbidden`, and `pgrep -a 'loopd|bxctl|memoryd'`. After the edits `cargo check -p brokerd` is clean with no warning; `config` 7, `state_socket` 8, `serve_state` 5, `instance` 2, `serve` 8, `serve_pointers` 10 pass. Step 7: `grep -n 'listen(&cfg'` prints three lines with `state_socket` first, `grep -n '^use' state_socket.rs` prints four and the `bxcore` import carries `ErrorCode`, `Message`, `SessionStateReply`, `WireError` with no `SessionState`, `git diff --stat -- crates/brokerd/src` lists exactly config.rs/lib.rs/main.rs/serve.rs, the other-crates diff is empty. `make gate` printed `gate: ok` with 1171 tests (1158 before this task plus the 13 new: eight in `state_socket.rs`, five in `serve_state.rs`; the new `config.rs` and `serve.rs` kept their counts, 7 and 8). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b |
 | M5c1/04-bxctl-speaker | 2026-09-28 | done | 1 | pass | none | Copied four files from `docs/plans/M5c1/files/`: `chat_speaker.rs` (new) and new versions of `chat.rs`, `chat_approvals.rs` and `cli.rs` (did not copy `chat.rs-task01`). Step 2 showed exactly the named failures: four non-compiling test targets (`chat`, `chat_approvals`, `chat_speaker`, `cli`) with E0061 for `run_turn` (5 args) and `stream_turn` (6 args), E0282 on the closure in `chat.rs`, and E0560/E0609 for the missing `speaker` field. Made the edits exactly as the task gives: C1 the `--speaker NAME` flag added to the chat line of `USAGE` with one explaining line (six-space indent); C2 `ChatOptions` gains `speaker: Option<String>`; C3 `default` sets `speaker: None`; C4 the `--speaker` arm in `parse_chat` refuses a second one and falls through to `i += 2`; C5 `parse_speaker` (empty, or any whitespace/control char -> `UsageError`, else the value as typed). R1 `run_turn` gains `speaker: Option<&str>` after `resume` with the doc comment, R2 `speaker.map(str::to_string)` on the `Turn`, R3 `stream_turn` gains the parameter and passes it. M1-M3 all three `stream_turn` calls in `main.rs` pass `opts.speaker.as_deref()`. `main` unchanged (a `UsageError` already prints the usage text and exits 2). After the edits `cargo check -p bxctl` is clean with no warning; `cli` 15 passed, `chat_speaker` 7 passed, every bxctl suite passes. Step 7: `grep -rn 'speaker: None' crates/bxctl/src` prints one line (the default), `git diff --stat -- crates/bxctl/src` lists exactly chat.rs/cli.rs/main.rs, the other-crates diff is empty. `make gate` printed `gate: ok` with 1158 tests (1148 before plus the 10 new: three in `cli.rs`, seven in `chat_speaker.rs`; `chat.rs` and `chat_approvals.rs` kept their counts). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b |
 | M5c1/03-gatewayd-speaker | 2026-09-28 | done | 2 | pass | none | Copied the seven given files from `docs/plans/M5c1/files/`: `deliver_speaker.rs` and `serve_speaker.rs` (new) and new versions of `sessions.rs`, `deliver.rs`, `deliver_links.rs`, `deliver_notice.rs` and `support/fake_mm.rs` (it gains `refuse_users`). Step 2 showed exactly the named failures: `deliver`, `deliver_links`, `deliver_notice`, `deliver_speaker` and `sessions` did not compile (E0560 for `Batch.senders` / `Message.user`, E0061 for `deliver` taking 5 args, E0282 on the closure), and `serve_speaker` compiled with its four tests failing (three found `None` where a speaker was expected, the fourth had no `"no username for"` log line). Made the edits exactly as the task gives: S1 `Message` gains `user` (the post's `user_id`); S2/S3 set `user` on both `Route::Queue` paths; S4 `Batch` gains `senders` plus `Batch::sender` (`first()?` then `all(|u| u == first)`, `None` for two or more people or for none); S5 `Pending` gains `senders`; S6 `push` pushes `message.user` to `pending.senders` on the `Waiting` path; S7 the `Start` path is `senders: vec![message.user.clone()]` with the new `Pending`'s `senders: Vec::new()`; S8 `finish` takes `senders` with `std::mem::take`. D1 `one_turn` gains `speaker: Option<&str>`; D2 `speaker.map(str::to_string)` on the `Turn` (its comment rewritten); D3 `run_turn` gains `speaker` threaded through both `one_turn` calls; D4 `deliver` gains `speaker`; D5 `deliver` passes `speaker` to `run_turn`. G1 `Gateway` gains `usernames: HashMap<String,String>`; G2 `new()` inits it. H1 `start` computes `speaker = self.speaker(&batch)` and passes `speaker.as_deref()` to `deliver`. H2 added `Gateway::speaker` (`batch.sender()?`; a cache hit returns it; else `client.username`, `Ok` inserts and returns, `Err` logs `"gatewayd: no username for {user}: {e}; the turn has no speaker"` and returns `None`). `approvals.rs` and `mm/rest.rs` untouched, the text still joins as before and the speaker never enters it. `cargo check -p gatewayd` clean after sessions; after deliver it fails only in `serve/handle.rs` at the `deliver` call (E0061/E0277) as the task predicts, then clean after serve. `sessions` 11 passed, `deliver_speaker` 3, `serve_speaker` 4, every gatewayd suite passes. Step 7: the grep for `speaker: None` prints nothing, `git diff --stat -- crates/gatewayd/src` lists exactly `deliver.rs`, `serve/handle.rs`, `serve/mod.rs`, `sessions.rs`, the other-crates diff is empty. `make gate` printed `gate: ok` with 1148 tests (1137 in `cargo test --workspace`, plus the 1 ignored `end_to_end` the gate runs separately with `BOXMAKER_BROKERD`; 1138 before this task plus the 10 new: 3 in `sessions.rs`, 3 in `deliver_speaker.rs`, 4 in `serve_speaker.rs`). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b |
 | M5c1/02-loopd-speaker-turn | 2026-09-28 | done | 2 | pass | none | Copied five files from `docs/plans/M5c1/files/`: `speaker_turn.rs` (new) and new versions of `memory.rs`, `turn_broker.rs`, `end_to_end.rs` and `support/turn.rs` (skipped the older `memory.rs-task01`). Step 2 showed exactly the named failures: eight non-compiling test targets (end_to_end, limits, memory, neutralise, neutralise_replay, speaker_turn, turn, turn_broker) with E0061 for `run_turn` and `updated`, E0282 on the closure, E0050 and E0599 for `turns`. Made the edits exactly as the task gives: S1 `Session::turns` counting `User` records in `records` via `matches!` and `u64::try_from(...).unwrap_or(u64::MAX)`, placed before `last_usage`; M1 the trait `updated` gains `turn: u64`, M2 `NoMemory` and M3 `MemoryPort` pass it (the `turn: 0` placeholder and its comment dropped for `turn`); T1 `run_turn` gains `speaker: Option<&str>` after `content` with the doc comment, T2 logs `speaker: speaker.map(str::to_string)` (as it came, not checked, not neutralised, `None` not written); C1 passes `turn.speaker.as_deref()` to `run_turn`, C2 passes `session.turns()` to `updated` (the comment now says the number is this turn's, a failed turn not told but its logged message still counts). `baseline::messages` untouched, the request is the same with and without a speaker. Step 7 all pass: the grep for `turn: 0`/`speaker: None` prints nothing, `git diff --stat -- crates/loopd/src` lists exactly channel.rs/memory.rs/session.rs/turn.rs (baseline.rs absent), the other-crates diff is empty. After the edits `cargo check -p loopd` is clean, `speaker_turn` 8 passed, every loopd suite passes. `make gate` printed `gate: ok` with 1138 tests (1130 before plus the 8 of `speaker_turn.rs`). No new dependency, so `Cargo.lock` is unchanged. | ornith-1.5-35b-a3b |

docs/runbook.md

@@ -200,8 +200,8 @@ socket, stop it; `brokerd` removes a stale socket file by itself.
 
 **What you see.** One of three lines, then this entry:
 
-- `brokerd: cannot accept on accept-broker (or accept-admin) for now, retrying: <error>`. It keeps
+- `brokerd: cannot accept on accept-broker (or accept-admin, accept-state) for now, retrying:
-  running and tries every 200 ms; the line is printed once per episode.
+  <error>`. It keeps running and tries every 200 ms; the line is printed once per episode.
 - `brokerd: cannot start a thread for a connection, so it was closed`. It keeps running.
 - `brokerd: stopped serving: <error>`, and it exits 1: any other failure of `accept`.
  In both cases `loopd` reports
@@ -293,11 +293,12 @@ be removed.
 this entry.
 
 **Why.** `broker.sock` accepts only tool requests from `loopd`; `admin.sock` accepts only admin
-requests from `bxctl`. Nothing in the harness sends a wrong kind, so this is a bug or a component
+requests from `bxctl`; `state.sock` accepts only `memoryd`'s questions about a session's state.
+Nothing in the harness sends a wrong kind, so this is a bug or a component
 doing what it should not, such as something trying to approve through the tool socket.
 
-**Confirm.** Note the time, kind and socket. Check what was running: `pgrep -a 'loopd|bxctl'`.
+**Confirm.** Note the time, kind and socket. Check what was running:
-Look at the session logs active at that time.
+`pgrep -a 'loopd|bxctl|memoryd'`. Look at the session logs active at that time.
 
 **Fix.** If a harness version mismatch explains it (a new `bxctl` against an old `brokerd`),
 rebuild and restart both. Otherwise keep the logs, stop `loopd`, and investigate before running it