# M3a task 17: `BrokerPort`, `[broker]` config and the runbook pointers

**Branch:** `m3a` (run `git switch m3a`; `git status --short` must be empty, otherwise stop)
**Commit subject:** `Add BrokerPort: loopd asks brokerd for every tool call`

## Goal

`loopd serve` sends every tool call to `brokerd` over `broker.sock`. Whatever goes wrong there, the
model gets a plain failure and the turn goes on. Every fail-closed message `loopd` prints ends with
its entry in `docs/runbook.md`.

## Files

- Copy: `crates/loopd/tests/broker_port.rs`, `broker_port_bad.rs`, `pointers.rs`, `config.rs`,
  `device.rs`, and `crates/loopd/tests/support/broker.rs`
- Create: `crates/loopd/src/broker_port.rs` (add `pub mod broker_port;` to `lib.rs`)
- Modify: `crates/loopd/src/config.rs`, `main.rs`, `session.rs`, `baseline.rs`,
  `docs/implementer-log.md`

## Interfaces

```rust
// config.rs: same style as `Loop`. `Config` gains `#[serde(default)] pub broker: Broker`.
#[serde(deny_unknown_fields, default)]
pub struct Broker { pub socket: Option<std::path::PathBuf>, pub timeout_ms: u64 }
// Default: socket None, timeout_ms 120_000. No default path: absent means no broker.

// broker_port.rs
pub const UNAVAILABLE: &str = "the tool broker is unavailable";
pub const NOT_CONFIGURED: &str = "no tool broker is configured";
pub const TOO_LARGE: &str = "the request is too large for the tool broker";
pub const POINTER: &str = "see docs/runbook.md#broker-unavailable";
/// "loopd: {UNAVAILABLE}: {why}; {POINTER}"
pub fn unavailable_line(why: &str) -> String;
/// "loopd: {NOT_CONFIGURED}: every tool call except the core ones will fail; {POINTER}"
pub fn not_configured_line() -> String;

pub struct BrokerPort { /* socket: PathBuf, timeout: Duration, log: Box<dyn Fn(&str) + Send + Sync> */ }
impl BrokerPort {
    pub fn new(socket: PathBuf, timeout: Duration) -> BrokerPort;   // log = |l| eprintln!("{l}")
    pub fn with_log(socket: PathBuf, timeout: Duration, log: Box<dyn Fn(&str) + Send + Sync>) -> BrokerPort;
}
impl ToolPort for BrokerPort { /* below */ }
pub struct NoBroker;       // ToolPort: always Failed { NOT_CONFIGURED }, prints nothing
```

## The protocol

One connection per call. `loopd` sends one frame: `v: PROTOCOL_VERSION`, `id` = `request.call.0`,
`final: true`, `Message::ToolRequest(request.clone())`. `brokerd` answers with frames carrying the
same `id`: optionally one `ToolResponse::PendingApproval` with `final: false`, then exactly one
`Result`, `Failed` or `Denied` with `final: true`, and closes.

## The timeout is a deadline, not a per-read timeout

A read timeout set once lets a peer that sends one byte now and then hold the turn for ever. Read
through this, which allows each `read` only what is left:

```rust
struct Deadline<'a> { stream: &'a UnixStream, until: Instant }
impl std::io::Read for Deadline<'_> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let left = self.until.saturating_duration_since(Instant::now());
        if left.is_zero() {
            return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
        }
        self.stream.set_read_timeout(Some(left))?;
        let mut stream = self.stream;       // `Read` is implemented for `&UnixStream`
        stream.read(buf)
    }
}
```

Verified in the std docs of Rust 1.98.1: `set_read_timeout` and `set_write_timeout` "An Err is
returned if the zero Duration is passed to this method", hence the `is_zero` check first. A read
that times out fails with kind `WouldBlock` on Linux (`TimedOut` elsewhere); treat both as a
timeout, as `http.rs` does. `Instant::checked_add(Duration) -> Option<Instant>`.

- First deadline: `Instant::now().checked_add(timeout)`, taken before connecting.
- After a pending frame: `wait = expires.unix_millis().saturating_sub(Timestamp::now().unix_millis())`,
  then `Instant::now().checked_add(Duration::from_millis(wait))` and `.checked_add(timeout)` on
  that. An `expires` in the past leaves `timeout`. `expires` comes from a peer: never add without
  `checked_add`.

## What `call` does. Every exit is listed; there are no others

Exits marked **U** do the same three things: call `log` exactly once with
`unavailable_line(why)`, return `ToolResponse::Failed { message: UNAVAILABLE.to_string() }`, and
drop the stream. Write one helper for it and use it at every **U**. Never panic, never return
`PendingApproval`.

1. First deadline is `None` → **U**, why `the timeout is too large`.
2. `UnixStream::connect(&socket)` fails → **U**, `cannot connect to {socket}: {e}`.
3. `set_write_timeout(Some(timeout))` fails (it does for a zero timeout) → **U**,
   `cannot set a timeout: {e}`.
4. `write_frame` fails with `FrameError::TooLarge(_)` → return `Failed { TOO_LARGE }` and do
   **not** call `log`: the broker is fine. Any other `write_frame` error → **U**,
   `cannot send the request: {e}`.
5. `read_frame(&mut Deadline { .. })` fails. `FrameError::Closed` → **U**,
   `the connection closed before the final answer`. `FrameError::Io(e)` with kind `WouldBlock` or
   `TimedOut` → **U**, `no answer in time`. Any other error (including `TooLarge` here) → **U**,
   `bad frame: {e}`.
6. The frame's `id` is not the request's → **U**, `an answer for request {got}, not {want}`.
7. `msg` is `Message::Error(e)` → **U**, `the broker reported an error: {detail}`. Any other
   message that is not `Message::ToolResponse` → **U**, `an unexpected message`.
8. `PendingApproval` with `final: true` → **U**, `a pending frame marked final`.
9. `Result`, `Failed` or `Denied` with `final: false` → **U**, `an answer not marked final`.
10. `PendingApproval` with `final: false`: if one was already seen → **U**,
    `a second pending frame`. Otherwise compute the new deadline (`None` → **U**,
    `an expiry too far away`), call `on_pending(&Pending { approval, expires })` once, and go
    back to step 5 with the new deadline.
11. `Result`, `Failed` or `Denied` with `final: true` → return it unchanged. Nothing is logged: a
    denial is not an outage.

Steps 5 to 10 apply to the frame after a pending frame exactly as to the first.

## Wiring and pointers

- `main.rs`, `run_serve`: `Some(path)` → `BrokerPort::new(path.clone(),
  Duration::from_millis(cfg.broker.timeout_ms))`; `None` → `eprintln!("{}",
  not_configured_line())` once, then `NoBroker`. `FakeTools` is no longer used by `main.rs`. A
  socket that does not exist yet is not a startup error: `loopd` connects per call.
- `main.rs`, `run_selftest_check`: the failure line becomes
  `selftest: FAILED: {e}; see docs/runbook.md#loopd-selftest-failed`. It serves both commands.
- `session.rs`: only `SessionError::Torn` changes, to
  `{path}:{line}: {why}; see docs/runbook.md#session-log-damaged`.
- `baseline.rs`: add `BaselineError::Core(PathBuf, std::io::Error)`, displayed as
  `{path}: {err}; see docs/runbook.md#core-memory-unreadable`, and return it (not `Read`) for an
  unreadable `memory/core.md`. `Read` stays as it is for `system.md`.

## Steps

- [ ] **1. Copy.** `git switch m3a`, then
  `cp docs/plans/M3a/files/crates/loopd/tests/{broker_port,broker_port_bad,pointers,config,device}.rs crates/loopd/tests/`
  and `cp docs/plans/M3a/files/crates/loopd/tests/support/broker.rs crates/loopd/tests/support/`
- [ ] **2. See the tests fail.** `cargo test -p loopd --test broker_port`. Expected: no compile.
- [ ] **3. Write** `config.rs`, then `broker_port.rs`, then the wiring and pointers. `cargo fmt --all`.
- [ ] **4. See the tests pass.**
  `cargo test -p loopd --test broker_port --test broker_port_bad --test pointers --test config`,
  five times in a row. Expected every time: `15 passed`, `2 passed`, `7 passed`, `11 passed`.
  `broker_port` takes a few seconds: it waits for real timeouts.
- [ ] **5. Run the gate.** `make gate`. Expected last line: `gate: ok`.
- [ ] **6. Log and commit.** `git add crates/loopd docs/implementer-log.md && git commit`

## Done when

- Step 4's counts, five runs in a row, and `make gate` prints `gate: ok`.
- You have checked each of the eleven exits above against your code, one by one, and the log's
  Notes say so.

## Stop and report if

- A test needs the port to keep a connection open between calls, or to retry.
- `the_timeout_is_a_deadline_for_the_frame_not_for_each_read` fails although you read through
  `Deadline`. Do not loosen it.
- `device.rs` does not compile. Do not run it: it needs straylight and is the owner's to run.