# M5c-1 task 05: `brokerd` answers `memoryd`'s question about a session's state on `state.sock` **Branch:** `m5c1` (run `git switch m5c1`; `git status --short` must be empty, otherwise stop) **Commit subject:** `brokerd: state.sock answers memoryd's question about a session's state` **Log row:** the Task cell is `M5c1/05-brokerd-state-socket`. ## Goal When a turn ends, `memoryd` will decide where the turn goes in Honcho, and for that it needs the session's state as `brokerd` knows it: the session's class (its taint) and whether it has read untrusted content (M5c spec, sections 5 and 7). Task 01 added the two messages: `bxcore::SessionState { session }`, the question, and `bxcore::SessionStateReply { class, untrusted }`, the answer. This task makes `brokerd` answer the question on a new socket. `memoryd` connects to it in task 09. - **The socket.** `brokerd` serves a third socket, beside `broker.sock` (tools, from `loopd`) and `admin.sock` (approvals, from `bxctl`): `state.sock`, by default `/run/memory-broker/state.sock`. It is configured exactly as the other two are: a key in the `[sockets]` table of `brokerd.toml`, `state = ""`, where an empty or missing value means the default under `home`. It is bound by the same `listen` function as the other two, so its directory is made (and kept) at mode 0700, a stale socket file is removed before the bind, and the socket is 0600. - **One question per connection**, as on the other sockets: `memoryd` connects, sends one frame, reads one final frame with the same `id`, and the connection is closed. A second frame on the same connection is never read. - **The answer** comes from the state store, `StateStore::read`, the same files the ledger writes (`/broker/sessions/.json`): `class` is the stored taint and `untrusted` the stored flag. A session with no file has made no tool call, and the store answers its default, `policy::SessionState::default()`, which is `private` and trusted (`untrusted: false`): exactly what the spec says such a session answers. Asking changes nothing: no file, no directory, no audit record. - **Every error is an error frame, never an answer**, so `memoryd` leaves the turn undecided: | What arrives, or happens | The final frame | Logged | |---|---|---| | `session_state`, the state file reads | `session_state_reply { class, untrusted }` | nothing | | `session_state`, the file exists but cannot be read (damaged, a directory, …) | `error`, code `internal`, the store's own text (it names the file and ends with `see docs/runbook.md#broker-state-damaged`) | `brokerd: ` and the same text | | any other kind (a tool request, an approval, a `session_state_reply`, …) | `error`, code `forbidden`, ` is not accepted on state.sock` | `brokerd: refused the message kind on state.sock` and the `socket-forbidden` pointer | | a malformed frame: not JSON, a session id that is not one, an unknown key, another version | `error`, `bad_message` or `bad_version` (or `bad_frame`), with id 0 | nothing | The last two rows are what `broker.sock` and `admin.sock` already do, through the same functions (`forbid`, `read_request`); nothing new is written for them. ## Every place that changes This is the complete list. | Place | What happens there | |---|---| | `crates/brokerd/src/config.rs`, `Sockets` | new field `state`, and a doc comment (edit C1) | | `crates/brokerd/src/config.rs`, `impl Config` | new method `state_socket`, after `admin_socket` (edit C2) | | `crates/brokerd/src/lib.rs` | `pub mod state_socket;` (edit L1) | | `crates/brokerd/src/state_socket.rs` | new file: `SOCKET` and `handle` (step 4 gives it whole) | | `crates/brokerd/src/serve.rs`, `Started`, `start`, `Which`, `run`, `accept_loop`, `serve_one` | the third socket bound, accepted and handled (edits V1 to V11) | | `crates/brokerd/src/main.rs` | one more line in the serving notice, and the module comment (edits M1 to M3) | | `docs/runbook.md`, `brokerd-listener-lost` and `socket-forbidden` | they name the third socket (edits R1 to R3) | | `crates/brokerd/src/broker.rs`, `admin.rs`, `state.rs`, `ledger/` | **no change**: `forbid`, `read_request`, `send` and `StateStore::read` are used as they are | ## Files - Copy, from `docs/plans/M5c1/files/` (the same path under it): - new: `crates/brokerd/tests/state_socket.rs`, `crates/brokerd/tests/serve_state.rs` - **new versions** of (they replace the files in the tree): `crates/brokerd/tests/config.rs` and `crates/brokerd/tests/fixtures/config/full.toml` (both now set and check `state`), and `crates/brokerd/tests/serve.rs` (it waits for an answer before it reads the notices; below). - Create: `crates/brokerd/src/state_socket.rs` (step 4). - Modify: `crates/brokerd/src/config.rs`, `crates/brokerd/src/lib.rs`, `crates/brokerd/src/serve.rs`, `crates/brokerd/src/main.rs`, `docs/runbook.md`, `docs/implementer-log.md`. - Do not touch: any other source file (`broker.rs`, `admin.rs`, `state.rs`, `ledger/` and `policy/` in particular), any file of another crate, any `Cargo.toml`, `crates/brokerd/tests/support/`, any other test. ## Facts you need - **Two types named `SessionState`.** `brokerd::policy::SessionState { taint, untrusted }` is what the state store returns; `bxcore::SessionState { session }` is the question on the wire. The new file names **neither** in a `use` line: it matches the question as `Message::SessionState(question)`, which needs no import, and reads the store's answer through its fields (`state.taint`, `state.untrusted`), which needs none either. It imports `bxcore::SessionStateReply` only. Do not add `SessionState` to any `use` line in `brokerd`'s sources. (The given `state_socket.rs` test needs both, and imports the question as `bxcore::SessionState as StateQuestion`.) - **The store's default.** `StateStore::read` returns `Ok(SessionState::default())` when the file is missing, and `policy::SessionState::default()` is `taint: DataClass::Private, untrusted: false`. A file that exists and cannot be read, or holds something that is not a state (bad JSON, an unknown key, a `public` taint), is `Err(StateError)`. `read` never writes: the handler calls nothing else on the store. - **Why a new `StateStore` and not the ledger.** The ledger holds the store behind its lock, for the steps that read and then raise the state. The question only reads, and the store writes each file atomically (a rename), so a read sees the old state or the new one, whole. The handler makes its own `StateStore::new(&broker.cfg().state_dir())`, the same directory the ledger's store is made with in `serve::start`; it never takes the ledger's lock, and so a question never waits behind a tool call. - **Threads.** As for the other two sockets: one accept thread for the socket (named `accept-state`), and one thread per accepted connection (`serve_one`), which runs `handle` once and ends. A connection that sends nothing holds only its own thread; a given test checks that the next question is still answered. - **The order of the binds.** `start` binds `state.sock` **first**, then `broker.sock` and `admin.sock`, so whoever waits for the tool and admin sockets finds the third bound too. - **Why a new `tests/serve.rs`.** Its tests started `brokerd`, waited until the two sockets accepted a connection, then stopped it and looked for the startup notices. A socket accepts as soon as it is bound, and `brokerd` prints its notices only after that, so under load the notice was at times not yet printed (`a_torn_tail_is_recovered_and_it_serves` and `it_makes_its_directories_0700_and_its_sockets_0600` failed about one run in ten, before this task too). The new version also waits for the admin socket to answer an `approvals` request, which a handler thread does only after the notices are printed. Nothing else in it changes. - **The serving notice.** `main` prints one more line after `brokerd: serving tools on … and approvals on …`: `brokerd: serving session state on `. The existing line keeps its words: the runbook quotes it. - **The error for a damaged state.** `StateError`'s `Display` already names the file and ends with `see docs/runbook.md#broker-state-damaged`, an entry that exists. The handler logs `brokerd: {e}`, as the ledger does, and sends `e.to_string()` as the error's `detail`. No new runbook entry: `memoryd`'s pointer for an unanswered question, `broker-state-unavailable`, comes with its entry in task 09. - **The given tests.** `state_socket.rs` calls `state_socket::handle` over a socket pair (the support file `client.rs`, as `broker.rs` does); every read has a 10-second limit, and each test's home is a `TempDir` removed on drop. `serve_state.rs` starts `brokerd serve` as a process: every wait has a 10-second deadline, and a guard kills and waits for the process before the home is removed. On Linux, a Unix socket closed with a frame still unread in it resets the connection, so the test for "one question per connection" accepts a reset as well as a clean close. - The calls the edits use: | Call | Signature, and what it gives | |---|---| | `read_request` | `pub fn read_request(stream: &mut UnixStream) -> Option` (`broker.rs`): the frame, or `None` after answering a malformed one | | `forbid` | `pub fn forbid(broker: &Broker, stream: &mut UnixStream, id: u64, msg: &Message, socket: &str)` (`broker.rs`): logs and sends the `forbidden` error | | `send` | `pub fn send(stream: &mut UnixStream, id: u64, last: bool, msg: Message) -> bool` (`broker.rs`) | | `StateStore::new` | `pub fn new(dir: &Path) -> StateStore` (`state.rs`) | | `StateStore::read` | `pub fn read(&self, session: &SessionId) -> Result` (`state.rs`) | | `Broker::cfg`, `Broker::log` | `pub fn cfg(&self) -> &Config`; `pub fn log(&self, line: &str)` | | `Config::state_dir` | `pub fn state_dir(&self) -> PathBuf`: `/broker/sessions` | | `Config::state_socket` | `pub fn state_socket(&self) -> PathBuf` (new, C2) | | `state_socket::handle` | `pub fn handle(mut stream: UnixStream, broker: &Broker)` (new, step 4) | | `listen` | `fn listen(socket: &Path) -> Result` (`serve.rs`, unchanged) | ## The edits, exactly Nothing else in these files changes. Each quoted "before" block occurs exactly once in its file. ### `crates/brokerd/src/config.rs` **C1.** The lines ```rust /// 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, } ``` become ```rust /// 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, } ``` **C2.** The line ```rust /// `gateway.sock`, when `[gateway]` is set. ``` becomes ```rust /// `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. ``` ### `crates/brokerd/src/lib.rs` **L1.** The line ```rust pub mod state; ``` becomes ```rust pub mod state; pub mod state_socket; ``` ### `crates/brokerd/src/serve.rs` **V1.** The line ```rust //! `brokerd serve`: bring the broker up, listen on both sockets, and keep it running. ``` becomes ```rust //! `brokerd serve`: bring the broker up, listen on its three sockets, and keep it running. ``` **V2.** The line ```rust use crate::state::StateStore; ``` becomes ```rust use crate::state::StateStore; use crate::state_socket; ``` **V3.** The lines ```rust /// The result of `start`: a running broker with both sockets open. pub struct Started { pub broker: Arc, pub recovered: bool, pub accepted: Option, tools: UnixListener, admin: UnixListener, } /// Start the broker: open the audit chain, bind both sockets, and build the ledger and broker. ``` become ```rust /// The result of `start`: a running broker with its three sockets open. pub struct Started { pub broker: Arc, pub recovered: bool, pub accepted: Option, tools: UnixListener, admin: UnixListener, state: UnixListener, } /// Start the broker: open the audit chain, bind the three sockets, and build the ledger and broker. ``` **V4.** In `start`, the lines ```rust // Both sockets, in the order the task gives. let tools = listen(&cfg.broker_socket())?; ``` become ```rust // 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())?; ``` **V5.** At the end of `start`, the lines ```rust tools, admin, }) } ``` become ```rust tools, admin, state, }) } ``` **V6.** The lines ```rust enum Which { Tools, Admin, } ``` become ```rust enum Which { Tools, Admin, State, } ``` **V7.** In `run`, the lines ```rust tools, admin, .. } = self; ``` become ```rust tools, admin, state, .. } = self; ``` **V8.** In `run`, the line ```rust accept_loop(admin, Arc::clone(&broker), tx, Which::Admin)?; ``` becomes ```rust accept_loop(admin, Arc::clone(&broker), tx.clone(), Which::Admin)?; accept_loop(state, Arc::clone(&broker), tx, Which::State)?; ``` **V9.** Above `accept_loop`, the line ```rust /// socket with it. Any other error stops the daemon through `tx`. ``` becomes ```rust /// sockets with it. Any other error stops the daemon through `tx`. ``` **V10.** In `accept_loop`, the lines ```rust Which::Admin => "accept-admin", }; ``` become ```rust Which::Admin => "accept-admin", Which::State => "accept-state", }; ``` **V11.** In `serve_one`, the lines ```rust Which::Admin => admin::handle(stream, &for_thread), }); ``` become ```rust Which::Admin => admin::handle(stream, &for_thread), Which::State => state_socket::handle(stream, &for_thread), }); ``` ### `crates/brokerd/src/main.rs` **M1.** The line ```rust //! `brokerd`: the broker role. `serve` takes the audit lock, binds both sockets, and serves. ``` becomes ```rust //! `brokerd`: the broker role. `serve` takes the audit lock, binds its three sockets, and serves. ``` **M2.** The line ```rust let admin_path = cfg.admin_socket(); ``` becomes ```rust let admin_path = cfg.admin_socket(); let state_path = cfg.state_socket(); ``` **M3.** The line ```rust eprintln!("brokerd: {runtime_notice}"); ``` becomes ```rust eprintln!("brokerd: serving session state on {}", state_path.display()); eprintln!("brokerd: {runtime_notice}"); ``` ### `docs/runbook.md` **R1.** In `## brokerd-listener-lost`, the lines ```text - `brokerd: cannot accept on accept-broker (or accept-admin) for now, retrying: `. It keeps running and tries every 200 ms; the line is printed once per episode. ``` become ```text - `brokerd: cannot accept on accept-broker (or accept-admin, accept-state) for now, retrying: `. It keeps running and tries every 200 ms; the line is printed once per episode. ``` **R2.** In `## socket-forbidden`, the line ```text requests from `bxctl`. Nothing in the harness sends a wrong kind, so this is a bug or a component ``` becomes ```text 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 ``` **R3.** In `## socket-forbidden`, the lines ```text **Confirm.** Note the time, kind and socket. Check what was running: `pgrep -a 'loopd|bxctl'`. Look at the session logs active at that time. ``` become ```text **Confirm.** Note the time, kind and socket. Check what was running: `pgrep -a 'loopd|bxctl|memoryd'`. Look at the session logs active at that time. ``` ## Steps - [ ] **1. Copy the given files.** From the repository root, five commands: ```sh cp docs/plans/M5c1/files/crates/brokerd/tests/state_socket.rs crates/brokerd/tests/ cp docs/plans/M5c1/files/crates/brokerd/tests/serve_state.rs crates/brokerd/tests/ cp docs/plans/M5c1/files/crates/brokerd/tests/config.rs crates/brokerd/tests/ cp docs/plans/M5c1/files/crates/brokerd/tests/serve.rs crates/brokerd/tests/ cp docs/plans/M5c1/files/crates/brokerd/tests/fixtures/config/full.toml \ crates/brokerd/tests/fixtures/config/ ``` Then `git status --short` lists 3 modified files and 2 new ones, all under `crates/brokerd/tests/`. - [ ] **2. See them fail.** Two commands, each with a shell timeout of 600000 ms: - `cargo build --tests --keep-going -p brokerd`: two test targets do not compile. `config`, with four `E0599` (``no method named `state_socket` found for struct `Config` ``), and `state_socket`, with one `E0432` (``unresolved import `brokerd::state_socket` ``). The new `serve.rs` compiles. - `cargo test -p brokerd --test serve_state`: it compiles, and its 5 tests fail, after about 10 s (they run side by side). Four fail with `brokerd never listened on …/run/memory-broker/state.sock`; `sockets_state_in_the_config_moves_it` fails with `brokerd never listened on …/run/loop-broker/broker.sock`, because `brokerd` refuses a config with the unknown key `state` and exits. - [ ] **3. The configuration.** Make edits C1 and C2. `cargo test -p brokerd --test config`: 7 passed. - [ ] **4. The handler.** Make edit L1, then create `crates/brokerd/src/state_socket.rs` with exactly this content: ```rust //! 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); } ``` `cargo check -p brokerd`: no error and no warning. Then `cargo test -p brokerd --test state_socket`: 8 passed. - [ ] **5. Serve it.** Make edits V1 to V11 in `serve.rs`, then M1 to M3 in `main.rs`. `cargo check -p brokerd`: no error and no warning. Then `cargo test -p brokerd --test serve_state`: 5 passed. Then `cargo test -p brokerd --test serve --test instance --test serve_pointers`: every test passes (8, 2 and 10). The new tests: | Test | What it checks | |---|---| | `the_answer_is_the_stored_class_and_flag` | step 4: class and flag, all four pairs, from states written by the store's own `raise` | | `a_state_file_the_owner_wrote_is_answered_as_it_says` | step 4: a file written by hand | | `a_session_brokerd_has_not_seen_is_private_and_trusted` | step 4: the store's default; no file or directory made | | `asking_changes_nothing` | step 4: state files, audit log and log lines unchanged | | `a_damaged_state_is_an_error_never_an_answer` | step 4: `internal`, the store's text, logged once | | `every_other_kind_is_forbidden_and_logged` | step 4: `forbid` with `SOCKET` | | `a_malformed_frame_is_answered_with_an_error` | step 4: `read_request`'s answers, a path as a session id | | `a_second_question_on_one_connection_is_not_answered` | step 4: one question per connection | | `it_answers_on_its_default_path_from_the_state_files` | C2, V3 to V11, M2, M3, end to end | | `its_directory_is_0700_its_socket_0600_and_a_stale_one_is_replaced` | V4: bound by `listen` | | `sockets_state_in_the_config_moves_it` | C1, C2: `[sockets] state` | | `a_silent_connection_does_not_hold_up_the_next` | V11: a thread per connection | | `the_other_sockets_refuse_the_question` | the tool and admin sockets forbid it (no change) | The new versions of `config.rs` and `full.toml` check `state_socket` by default, when set, and beside the other two. - [ ] **6. The runbook.** Make edits R1, R2 and R3. - [ ] **7. Check.** Four commands: - `grep -n 'listen(&cfg' crates/brokerd/src/serve.rs` prints three lines, `state_socket` first. - `grep -n '^use' crates/brokerd/src/state_socket.rs` prints four lines; the one from `bxcore` imports `ErrorCode`, `Message`, `SessionStateReply` and `WireError`, and no `SessionState`. - `git diff --stat -- crates/brokerd/src` lists exactly `config.rs`, `lib.rs`, `main.rs` and `serve.rs` (the new `state_socket.rs` is untracked until you stage it). - `git diff --stat -- crates/bxcore crates/loopd crates/memoryd crates/gatewayd crates/bxctl` prints nothing. - [ ] **8. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms. Expected last line: `gate: ok`, with 1171 tests: 1158 before this task, plus 13 new ones (eight in `state_socket.rs`, five in `serve_state.rs`; the new `config.rs` and `serve.rs` keep their counts, 7 and 8). - [ ] **9. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is `M5c1/05-brokerd-state-socket`. `git status --short` lists the four sources of "Modify" under `crates/brokerd/src/`, the three replaced test files, `docs/runbook.md` and your log as modified (ten files), and `crates/brokerd/src/state_socket.rs` and the two new test files as new. Stage with `git add crates/brokerd docs/runbook.md docs/implementer-log.md Cargo.lock`, then commit as `AGENTS.md` says. `Cargo.lock` does not change in this task; the `git add` is harmless. ## Done when - `make gate` prints `gate: ok` with 1171 tests. - `git diff HEAD~1 --stat` lists only the four modified sources, `state_socket.rs`, the five copied files, `docs/runbook.md` and `docs/implementer-log.md`. ## Stop and report if - A "before" block of any edit is not in its file exactly as shown, or is there more than once. - Step 2 shows a different result: other test targets, errors of another kind, or other tests failing in `serve_state`. - `cargo check` reports an error or a warning after step 4 or 5. - After step 3, 4 or 5 a test fails. Do not change a test, a fixture or the support files: stop, and give the failing tests' names and their messages. - The fix seems to need a change anywhere else: `broker.rs`, `state.rs`, the ledger, another crate, a default answer when the state cannot be read, a second question on one connection, a new runbook entry, or `SessionState` in a `use` line. It does not; stop and say why you think so. - A test in `inferproxy`, `toolkit`, `loopd` or `memoryd` fails in the gate: check once whether it fails without your change, then stop and report both results. - The gate passes with a count other than 1171. - A comment in an edit or in the new file does not fit the code under it. Quote it and stop.