Task files, the files they copy in (byte-identical to the reference on m3a-ref), each area's check record, and a README with the per-task table of what each check exposed. The handoff note is done with. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
157 lines
8.3 KiB
Markdown
157 lines
8.3 KiB
Markdown
# 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.
|