brokerd serves a third socket, [sockets] state (default <home>/run/memory-broker/state.sock), bound by the same listen (dir 0700, stale socket removed, socket 0600), one thread per connection, one question per connection. The answer is StateStore::read: the stored taint and flag, the store's default (private, trusted) for a session with no file. A damaged state is an internal error with the store's text, never an answer; other kinds are forbidden; malformed frames get read_request's error. A new tests/serve.rs waits for an answer before reading the startup notices (a race that failed about one run in ten under load, before this task too). Checks: every before block found once at ref 04 (20 edits); the new file in the task is byte-identical to the reference; the given tests fail before as the task says (config and state_socket do not compile; serve_state 5 of 5 fail); gate ok with 1171 tests on the reference and on a replay from the task text in a fresh ref-04 worktree with its own build dir (byte-identical to the reference; removed); /tmp unchanged around the gate; 25 mutations, 24 caught, the one missed (state bound last) changes no tested behaviour; the socket tests 25 runs under load, 0 failures; no task-text line over 100 characters outside tables and code. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
25 KiB
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.
brokerdserves a third socket, besidebroker.sock(tools, fromloopd) andadmin.sock(approvals, frombxctl):state.sock, by default<home>/run/memory-broker/state.sock. It is configured exactly as the other two are: a key in the[sockets]table ofbrokerd.toml,state = "<path>", where an empty or missing value means the default underhome. It is bound by the samelistenfunction 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:
memorydconnects, sends one frame, reads one final frame with the sameid, 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 (<home>/broker/sessions/<id>.json):classis the stored taint anduntrustedthe stored flag. A session with no file has made no tool call, and the store answers its default,policy::SessionState::default(), which isprivateand 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
memorydleaves the turn undecided:What arrives, or happens The final frame Logged session_state, the state file readssession_state_reply { class, untrusted }nothing session_state, the file exists but cannot be read (damaged, a directory, …)error, codeinternal, the store's own text (it names the file and ends withsee docs/runbook.md#broker-state-damaged)brokerd:and the same textany other kind (a tool request, an approval, a session_state_reply, …)error, codeforbidden,<kind> is not accepted on state.sockbrokerd: refused the message kind <kind> on state.sockand thesocket-forbiddenpointera malformed frame: not JSON, a session id that is not one, an unknown key, another version error,bad_messageorbad_version(orbad_frame), with id 0nothing The last two rows are what
broker.sockandadmin.sockalready 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.rsandcrates/brokerd/tests/fixtures/config/full.toml(both now set and checkstate), andcrates/brokerd/tests/serve.rs(it waits for an answer before it reads the notices; below).
- new:
- 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/andpolicy/in particular), any file of another crate, anyCargo.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 auseline: it matches the question asMessage::SessionState(question), which needs no import, and reads the store's answer through its fields (state.taint,state.untrusted), which needs none either. It importsbxcore::SessionStateReplyonly. Do not addSessionStateto anyuseline inbrokerd's sources. (The givenstate_socket.rstest needs both, and imports the question asbxcore::SessionState as StateQuestion.) - The store's default.
StateStore::readreturnsOk(SessionState::default())when the file is missing, andpolicy::SessionState::default()istaint: 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, apublictaint), isErr(StateError).readnever writes: the handler calls nothing else on the store. - Why a new
StateStoreand 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 ownStateStore::new(&broker.cfg().state_dir()), the same directory the ledger's store is made with inserve::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 runshandleonce 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.
startbindsstate.sockfirst, thenbroker.sockandadmin.sock, so whoever waits for the tool and admin sockets finds the third bound too. - Why a new
tests/serve.rs. Its tests startedbrokerd, 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, andbrokerdprints its notices only after that, so under load the notice was at times not yet printed (a_torn_tail_is_recovered_and_it_servesandit_makes_its_directories_0700_and_its_sockets_0600failed about one run in ten, before this task too). The new version also waits for the admin socket to answer anapprovalsrequest, which a handler thread does only after the notices are printed. Nothing else in it changes. - The serving notice.
mainprints one more line afterbrokerd: serving tools on … and approvals on …:brokerd: serving session state on <path>. The existing line keeps its words: the runbook quotes it. - The error for a damaged state.
StateError'sDisplayalready names the file and ends withsee docs/runbook.md#broker-state-damaged, an entry that exists. The handler logsbrokerd: {e}, as the ledger does, and sendse.to_string()as the error'sdetail. 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.rscallsstate_socket::handleover a socket pair (the support fileclient.rs, asbroker.rsdoes); every read has a 10-second limit, and each test's home is aTempDirremoved on drop.serve_state.rsstartsbrokerd serveas 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_requestpub fn read_request(stream: &mut UnixStream) -> Option<Envelope>(broker.rs): the frame, orNoneafter answering a malformed oneforbidpub fn forbid(broker: &Broker, stream: &mut UnixStream, id: u64, msg: &Message, socket: &str)(broker.rs): logs and sends theforbiddenerrorsendpub fn send(stream: &mut UnixStream, id: u64, last: bool, msg: Message) -> bool(broker.rs)StateStore::newpub fn new(dir: &Path) -> StateStore(state.rs)StateStore::readpub fn read(&self, session: &SessionId) -> Result<SessionState, StateError>(state.rs)Broker::cfg,Broker::logpub fn cfg(&self) -> &Config;pub fn log(&self, line: &str)Config::state_dirpub fn state_dir(&self) -> PathBuf:<home>/broker/sessionsConfig::state_socketpub fn state_socket(&self) -> PathBuf(new, C2)state_socket::handlepub fn handle(mut stream: UnixStream, broker: &Broker)(new, step 4)listenfn listen(socket: &Path) -> Result<UnixListener, ServeError>(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
/// 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
/// 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
/// `gateway.sock`, when `[gateway]` is set.
becomes
/// `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
pub mod state;
becomes
pub mod state;
pub mod state_socket;
crates/brokerd/src/serve.rs
V1. The line
//! `brokerd serve`: bring the broker up, listen on both sockets, and keep it running.
becomes
//! `brokerd serve`: bring the broker up, listen on its three sockets, and keep it running.
V2. The line
use crate::state::StateStore;
becomes
use crate::state::StateStore;
use crate::state_socket;
V3. The lines
/// The result of `start`: a running broker with both sockets open.
pub struct Started {
pub broker: Arc<Broker>,
pub recovered: bool,
pub accepted: Option<ChainFailure>,
tools: UnixListener,
admin: UnixListener,
}
/// Start the broker: open the audit chain, bind both sockets, and build the ledger and broker.
become
/// 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 the three sockets, and build the ledger and broker.
V4. In start, the lines
// Both sockets, in the order the task gives.
let tools = listen(&cfg.broker_socket())?;
become
// 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
tools,
admin,
})
}
become
tools,
admin,
state,
})
}
V6. The lines
enum Which {
Tools,
Admin,
}
become
enum Which {
Tools,
Admin,
State,
}
V7. In run, the lines
tools,
admin,
..
} = self;
become
tools,
admin,
state,
..
} = self;
V8. In run, the line
accept_loop(admin, Arc::clone(&broker), tx, Which::Admin)?;
becomes
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
/// socket with it. Any other error stops the daemon through `tx`.
becomes
/// sockets with it. Any other error stops the daemon through `tx`.
V10. In accept_loop, the lines
Which::Admin => "accept-admin",
};
become
Which::Admin => "accept-admin",
Which::State => "accept-state",
};
V11. In serve_one, the lines
Which::Admin => admin::handle(stream, &for_thread),
});
become
Which::Admin => admin::handle(stream, &for_thread),
Which::State => state_socket::handle(stream, &for_thread),
});
crates/brokerd/src/main.rs
M1. The line
//! `brokerd`: the broker role. `serve` takes the audit lock, binds both sockets, and serves.
becomes
//! `brokerd`: the broker role. `serve` takes the audit lock, binds its three sockets, and serves.
M2. The line
let admin_path = cfg.admin_socket();
becomes
let admin_path = cfg.admin_socket();
let state_path = cfg.state_socket();
M3. The line
eprintln!("brokerd: {runtime_notice}");
becomes
eprintln!("brokerd: serving session state on {}", state_path.display());
eprintln!("brokerd: {runtime_notice}");
docs/runbook.md
R1. In ## brokerd-listener-lost, the lines
- `brokerd: cannot accept on accept-broker (or accept-admin) for now, retrying: <error>`. It keeps
running and tries every 200 ms; the line is printed once per episode.
become
- `brokerd: cannot accept on accept-broker (or accept-admin, accept-state) for now, retrying:
<error>`. It keeps running and tries every 200 ms; the line is printed once per episode.
R2. In ## socket-forbidden, the line
requests from `bxctl`. Nothing in the harness sends a wrong kind, so this is a bug or a component
becomes
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
**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
**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:
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 --shortlists 3 modified files and 2 new ones, all undercrates/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 fourE0599(no method named `state_socket` found for struct `Config`), andstate_socket, with oneE0432(unresolved import `brokerd::state_socket`). The newserve.rscompiles.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 withbrokerd never listened on …/run/memory-broker/state.sock;sockets_state_in_the_config_moves_itfails withbrokerd never listened on …/run/loop-broker/broker.sock, becausebrokerdrefuses a config with the unknown keystateand 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.rswith exactly this content://! 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. Thencargo test -p brokerd --test state_socket: 8 passed. -
5. Serve it. Make edits V1 to V11 in
serve.rs, then M1 to M3 inmain.rs.cargo check -p brokerd: no error and no warning. Thencargo test -p brokerd --test serve_state: 5 passed. Thencargo 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_flagstep 4: class and flag, all four pairs, from states written by the store's own raisea_state_file_the_owner_wrote_is_answered_as_it_saysstep 4: a file written by hand a_session_brokerd_has_not_seen_is_private_and_trustedstep 4: the store's default; no file or directory made asking_changes_nothingstep 4: state files, audit log and log lines unchanged a_damaged_state_is_an_error_never_an_answerstep 4: internal, the store's text, logged onceevery_other_kind_is_forbidden_and_loggedstep 4: forbidwithSOCKETa_malformed_frame_is_answered_with_an_errorstep 4: read_request's answers, a path as a session ida_second_question_on_one_connection_is_not_answeredstep 4: one question per connection it_answers_on_its_default_path_from_the_state_filesC2, V3 to V11, M2, M3, end to end its_directory_is_0700_its_socket_0600_and_a_stale_one_is_replacedV4: bound by listensockets_state_in_the_config_moves_itC1, C2: [sockets] statea_silent_connection_does_not_hold_up_the_nextV11: a thread per connection the_other_sockets_refuse_the_questionthe tool and admin sockets forbid it (no change) The new versions of
config.rsandfull.tomlcheckstate_socketby 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.rsprints three lines,state_socketfirst.grep -n '^use' crates/brokerd/src/state_socket.rsprints four lines; the one frombxcoreimportsErrorCode,Message,SessionStateReplyandWireError, and noSessionState.git diff --stat -- crates/brokerd/srclists exactlyconfig.rs,lib.rs,main.rsandserve.rs(the newstate_socket.rsis untracked until you stage it).git diff --stat -- crates/bxcore crates/loopd crates/memoryd crates/gatewayd crates/bxctlprints nothing.
-
8. Run the gate.
cargo fmt --all, thenmake 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 instate_socket.rs, five inserve_state.rs; the newconfig.rsandserve.rskeep their counts, 7 and 8). -
9. Log and commit. Add your row to
docs/implementer-log.md; its Task cell isM5c1/05-brokerd-state-socket.git status --shortlists the four sources of "Modify" undercrates/brokerd/src/, the three replaced test files,docs/runbook.mdand your log as modified (ten files), andcrates/brokerd/src/state_socket.rsand the two new test files as new. Stage withgit add crates/brokerd docs/runbook.md docs/implementer-log.md Cargo.lock, then commit asAGENTS.mdsays.Cargo.lockdoes not change in this task; thegit addis harmless.
Done when
make gateprintsgate: okwith 1171 tests.git diff HEAD~1 --statlists only the four modified sources,state_socket.rs, the five copied files,docs/runbook.mdanddocs/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 checkreports 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, orSessionStatein auseline. It does not; stop and say why you think so. - A test in
inferproxy,toolkit,loopdormemorydfails 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.