M5c-2 plan: task 03, brokerd's wait for ask_memory

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

docs/plans/M5c2/03-brokerd-ask-memory-wait.md

@@ -0,0 +1,267 @@
+# M5c-2 task 03: `brokerd` waits 170 s for `ask_memory`
+
+**Branch:** `m5c2`. Check it: `git branch --show-current` prints `m5c2` and `git status --short`
+is empty; otherwise stop. Never switch branch.
+**Commit subject:** `brokerd: a wait of its own for ask_memory, 170 s`
+**Log row:** the Task cell is `M5c2/03-brokerd-ask-memory-wait`.
+
+## Goal
+
+`brokerd` hands each decided memory call to `memoryd` through a `MemoryPort`
+(`crates/brokerd/src/memory_port.rs`), and waits for the answer: today 30 s (`MEMORY_WAIT`), set
+as the socket's read and write timeouts. Task 02 made `ask_memory` a memory tool. Its answer can
+take much longer: `memoryd` asks Honcho's chat about each named workspace, both at once, with
+150 s each (M5c spec, section 9). So `ask_memory` gets a wait of its own:
+
+- `ASK_MEMORY_WAIT` = 170 s, a constant, not configuration. `MEMORY_WAIT` stays 30 s, for
+  `recall` and `memory`.
+- The port chooses the wait by the call's tool (`wait_for`), and sets it as the read and write
+  timeouts of that call's connection only.
+- The tests need short waits, so the wait for `ask_memory` can be set on a port
+  (`with_ask_wait`), as `new` already takes the wait for the other two. `MemoryPort::new` keeps
+  its signature, and `brokerd`'s `main.rs` does not change: `new` gives every port
+  `ASK_MEMORY_WAIT`.
+
+`loopd` waits 120 s for `brokerd` today, for any tool; task 04 gives `ask_memory` 180 s there.
+
+## Every place that changes
+
+This is the complete list.
+
+| Place | What happens there |
+|---|---|
+| `crates/brokerd/src/memory_port.rs` | the module's doc comment; a `use` line; `ASK_MEMORY_WAIT`; `MemoryPort`'s new field `ask_wait`; `new`, `with_ask_wait`, `wait_for`; `ask` uses `wait_for` (edits W1 to W5) |
+| `crates/brokerd/tests/memory_wait.rs` | new, given |
+| everything else | **no change**: not `main.rs`, not `args.rs`, not the other tests, no `Cargo.toml` |
+
+## Files
+
+- Copy, from `docs/plans/M5c2/files/` (the same path under it): the new test
+  `crates/brokerd/tests/memory_wait.rs`.
+- Modify: `crates/brokerd/src/memory_port.rs`, `docs/implementer-log.md`.
+- Do not touch: any other file.
+
+## Facts you need
+
+- **Struct update syntax.** `MemoryPort { ask_wait: wait, ..self }` builds a new `MemoryPort`
+  that takes `ask_wait` from `wait` and every other field from `self` (moved: `with_ask_wait`
+  takes `self` by value). It is what edit W3 writes.
+- **`ToolName`** is `crate::args::ToolName`, `Copy`, with the variant `AskMemory` since task 02.
+  `wait_for` matches it with a `_` arm for every other tool: only `ask_memory` has its own wait.
+- **The given test** `memory_wait.rs` (3 tests):
+  - the two constants; which tool gets which wait, with `new`'s own wait, with the default for
+    `ask_memory`, and with `with_ask_wait`;
+  - a fake `memoryd` that answers after 1.5 s: with a 300 ms wait for `recall` and `memory` and
+    20 s for `ask_memory`, the `ask_memory` call gets the answer and the other two get the
+    `UNREACHABLE` failure; and the other way round for `recall` and `ask_memory`. A failed
+    answer comes back within the short wait, and a good one as soon as it is written, so the
+    file takes about 3 s.
+- The items the edits use. This table is complete; nothing from a crate's source (the Cargo
+  registry, `~/.cargo/registry`) is needed or allowed.
+
+  | Item | Where |
+  |---|---|
+  | `std::time::Duration::from_secs(u64) -> Duration` (a `const fn`) | W3 |
+  | `crate::args::ToolName` | W2, W3 |
+  | `UnixStream::set_read_timeout(&self, Option<Duration>)`, `set_write_timeout` | W5, as before |
+
+## The edits, exactly
+
+Nothing else in this file changes. Each quoted "before" block occurs exactly once in the file when
+its turn comes (make them in order). Keep the fields in the order shown.
+
+### `crates/brokerd/src/memory_port.rs`
+
+**W1.** The module's doc comment, the lines
+
+```rust
+//! Where decided `recall` and `memory` calls go: to `memoryd` over its `brokerd` socket, with the
+//! session's state, instead of a container (M5a spec, section 8). `Routed` sends every other call
+//! to the configured runtime.
+```
+
+become
+
+```rust
+//! Where decided `recall`, `memory` and `ask_memory` calls go: to `memoryd` over its `brokerd`
+//! socket, with the session's state, instead of a container (M5a spec, section 8). `Routed` sends
+//! every other call to the configured runtime. `ask_memory` has a wait of its own (M5c spec,
+//! section 9).
+```
+
+**W2.** Add one line. The line
+
+```rust
+use crate::policy::Label;
+```
+
+becomes
+
+```rust
+use crate::args::ToolName;
+use crate::policy::Label;
+```
+
+**W3.** The lines
+
+```rust
+/// How long `brokerd` waits for `memoryd` to answer one call.
+pub const MEMORY_WAIT: Duration = Duration::from_secs(30);
+
+/// A client of `memoryd`'s `brokerd` socket. One connection per call.
+pub struct MemoryPort {
+    socket: PathBuf,
+    wait: Duration,
+}
+
+impl MemoryPort {
+    pub fn new(socket: PathBuf, wait: Duration) -> MemoryPort {
+        MemoryPort { socket, wait }
+    }
+```
+
+become
+
+```rust
+/// How long `brokerd` waits for `memoryd` to answer one `recall` or `memory` call.
+pub const MEMORY_WAIT: Duration = Duration::from_secs(30);
+/// How long it waits for one `ask_memory` call: `memoryd` asks Honcho's chat about each
+/// workspace, both at once, with 150 s each (M5c spec, section 9).
+pub const ASK_MEMORY_WAIT: Duration = Duration::from_secs(170);
+
+/// A client of `memoryd`'s `brokerd` socket. One connection per call.
+pub struct MemoryPort {
+    socket: PathBuf,
+    wait: Duration,
+    ask_wait: Duration,
+}
+
+impl MemoryPort {
+    /// A port that waits `wait` for `recall` and `memory`, and `ASK_MEMORY_WAIT` for `ask_memory`.
+    pub fn new(socket: PathBuf, wait: Duration) -> MemoryPort {
+        MemoryPort {
+            socket,
+            wait,
+            ask_wait: ASK_MEMORY_WAIT,
+        }
+    }
+
+    /// The same port, waiting `wait` for `ask_memory`. The tests use a short one.
+    pub fn with_ask_wait(self, wait: Duration) -> MemoryPort {
+        MemoryPort {
+            ask_wait: wait,
+            ..self
+        }
+    }
+
+    /// How long one call of `tool` waits for `memoryd`, to write the call and to read the answer.
+    pub fn wait_for(&self, tool: ToolName) -> Duration {
+        match tool {
+            ToolName::AskMemory => self.ask_wait,
+            _ => self.wait,
+        }
+    }
+```
+
+**W4.** In `ask`, the comment's lines
+
+```rust
+        // untrusted from `spec.state()`. 2. `UnixStream::connect(&self.socket)`, then read and
+        // write timeouts of `self.wait`. 3. `write_frame` an Envelope { v: PROTOCOL_VERSION, id:
+        // spec.call().0, final: true, msg: Message::MemoryCall(call) }. 4. `read_frame`; a
+        // Message::ToolResponse is Some(it), any other message None.
+```
+
+become
+
+```rust
+        // untrusted from `spec.state()`. 2. `UnixStream::connect(&self.socket)`, then read and
+        // write timeouts of `self.wait_for(spec.tool())`. 3. `write_frame` an Envelope { v:
+        // PROTOCOL_VERSION, id: spec.call().0, final: true, msg: Message::MemoryCall(call) }. 4.
+        // `read_frame`; a Message::ToolResponse is Some(it), any other message None.
+```
+
+**W5.** In `ask`, the lines
+
+```rust
+        let mut stream = UnixStream::connect(&self.socket).ok()?;
+        stream.set_read_timeout(Some(self.wait)).ok()?;
+        stream.set_write_timeout(Some(self.wait)).ok()?;
+```
+
+become
+
+```rust
+        let wait = self.wait_for(spec.tool());
+        let mut stream = UnixStream::connect(&self.socket).ok()?;
+        stream.set_read_timeout(Some(wait)).ok()?;
+        stream.set_write_timeout(Some(wait)).ok()?;
+```
+
+## Steps
+
+- [ ] **1. Copy the given test.** From the repository root, one command:
+
+  ```sh
+  cp docs/plans/M5c2/files/crates/brokerd/tests/memory_wait.rs crates/brokerd/tests/
+  ```
+
+  Then `git status --short` lists one new file, `crates/brokerd/tests/memory_wait.rs`.
+- [ ] **2. See it fail.** `cargo build --tests --keep-going -p brokerd`, with a shell timeout of
+  600000 ms: one test target, `memory_wait`, does not compile, with 10 errors: one `E0432`
+  (``unresolved import `brokerd::memory_port::ASK_MEMORY_WAIT` ``), six `E0599` (``no method
+  named `wait_for` found for struct `MemoryPort` ``) and three `E0599` (``no method named
+  `with_ask_wait` found for struct `MemoryPort` ``). Every other target compiles.
+- [ ] **3. The edits.** Make edits W1 to W5, in order. `cargo check -p brokerd` prints no error
+  and no warning.
+- [ ] **4. The tests.** With a shell timeout of 600000 ms, every test passes: 3 in `memory_wait`,
+  5 in `memory_port`, 9 in `ask_memory_tool`:
+
+  ```sh
+  cargo test -p brokerd --test memory_wait --test memory_port --test ask_memory_tool
+  ```
+- [ ] **5. Check.** Two commands:
+  - `git diff --stat` lists exactly `crates/brokerd/src/memory_port.rs`.
+  - `grep -n 'MEMORY_WAIT' crates/brokerd/src/main.rs` prints the two lines it printed before
+    this task: the `use` line and `MemoryPort::new(cfg.memory_socket(), MEMORY_WAIT)`.
+- [ ] **6. Run the gate.** `cargo fmt --all`, then `make gate`, with a shell timeout of 600000 ms.
+  Expected last line: `gate: ok`. The test count is the passed total of `cargo test --workspace`
+  plus one, for the ignored `end_to_end` test that the gate runs on its own: **1384** after this
+  task, 1381 before it; the 3 new ones are the tests of `memory_wait.rs`. To count, add up the
+  `N passed` of every `test result:` line the gate's `cargo test --workspace` printed, and add 1.
+  Never run the gate again only to count.
+- [ ] **7. Log and commit.** Add your row to `docs/implementer-log.md`; its Task cell is
+  `M5c2/03-brokerd-ask-memory-wait`. `git status --short` lists `memory_port.rs` and your log as
+  modified, and `memory_wait.rs` as new. Stage with
+  `git add crates/brokerd 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 1384 tests.
+- `git diff HEAD~1 --stat` lists only `crates/brokerd/src/memory_port.rs`,
+  `crates/brokerd/tests/memory_wait.rs` and `docs/implementer-log.md`.
+
+## Stop and report if
+
+- The branch check at the top fails.
+- A "before" block is not in the file exactly as shown when its turn comes, or is there more
+  than once.
+- Step 2 shows a different result: another target failing, or other errors.
+- `cargo check` in step 3 prints an error or a warning. Quote it, and stop.
+- A test fails in step 4. Do not change the test: stop, and give the failing tests' names and
+  their messages.
+- The fix seems to need a change anywhere else: `main.rs`, a configuration key, `MemoryPort::new`'s
+  signature, another test. It does not; stop and say why you think so.
+- A test outside `memory_wait.rs` fails in the gate: check once whether it fails without your
+  change (`git stash -u`, run it, `git stash pop`), then stop and report both results.
+- The gate passes with a count other than 1384.
+- A comment in an edit does not fit the code under it. Quote it and stop.
+
+## What the given tests pin
+
+Each of these changes to the finished code makes at least one given test fail (10 tried, all
+caught): either constant changed; `new` giving `ask_memory` its own wait instead of
+`ASK_MEMORY_WAIT`; `with_ask_wait` doing nothing, or setting the other tools' wait too;
+`wait_for` giving `ask_memory` the other tools' wait, or `recall` `ask_memory`'s; `ask` using
+the other tools' wait for every call, or `ask_memory`'s for every call; the read timeout not set.

docs/plans/M5c2/files/crates/brokerd/tests/memory_wait.rs

@@ -0,0 +1,131 @@
+//! How long `brokerd` waits for `memoryd`, per tool: 30 s for `recall` and `memory`, 170 s for
+//! `ask_memory`, whose two Honcho chats have 150 s each (M5c spec, sections 9 and 11). The waits
+//! are shortened here with `with_ask_wait` and `new`'s own wait, so no test waits long. Do not
+//! edit.
+
+#[path = "support/build.rs"]
+mod build;
+#[path = "support/tmp.rs"]
+mod tmp;
+
+use std::os::unix::net::UnixListener;
+use std::path::Path;
+use std::time::Duration;
+
+use brokerd::args::ToolName;
+use brokerd::memory_port::{ASK_MEMORY_WAIT, MEMORY_WAIT, MemoryPort, UNREACHABLE};
+use brokerd::policy::decide;
+use brokerd::runner::run;
+use build::{allowed, now, private, request, set};
+use bxcore::{DataClass, Envelope, Message, ToolRequest, ToolResponse, read_frame, write_frame};
+use tmp::TempDir;
+
+/// How long the slow fake `memoryd` takes to answer.
+const SLOW: Duration = Duration::from_millis(1_500);
+/// A wait shorter than `SLOW`, and one far longer.
+const SHORT: Duration = Duration::from_millis(300);
+const LONG: Duration = Duration::from_secs(20);
+
+fn found() -> ToolResponse {
+    ToolResponse::Result {
+        content: "own: memory · The owner prefers UTC.".to_string(),
+        class: DataClass::Private,
+        untrusted: false,
+        truncated: false,
+    }
+}
+
+/// A fake `memoryd` that answers one call after `SLOW`. A failed write is ignored: the port has
+/// stopped waiting, which is what some tests expect.
+fn slow_memoryd(socket: &Path) {
+    let listener = UnixListener::bind(socket).unwrap();
+    std::thread::spawn(move || {
+        let Ok((mut stream, _)) = listener.accept() else {
+            return;
+        };
+        let Ok(env) = read_frame(&mut stream) else {
+            return;
+        };
+        std::thread::sleep(SLOW);
+        let reply = Envelope {
+            v: 1,
+            id: env.id,
+            r#final: true,
+            msg: Message::ToolResponse(found()),
+        };
+        let _ = write_frame(&mut stream, &reply);
+    });
+}
+
+/// One allowed call of `req` through `port`, to a slow fake `memoryd` of its own.
+fn ask(
+    dir: &TempDir,
+    name: &str,
+    port: impl Fn(&Path) -> MemoryPort,
+    req: ToolRequest,
+) -> ToolResponse {
+    let socket = dir.path().join(name);
+    slow_memoryd(&socket);
+    let decision = allowed(decide(req, &set(vec![]), private(), now()));
+    run(decision, &port(&socket))
+}
+
+fn unreachable() -> ToolResponse {
+    ToolResponse::Failed {
+        message: UNREACHABLE.to_string(),
+    }
+}
+
+const QUESTION: &str = r#"{"question":"What does the owner prefer?"}"#;
+const QUERY: &str = r#"{"query":"owner"}"#;
+
+#[test]
+fn the_waits_and_which_tool_gets_which() {
+    assert_eq!(MEMORY_WAIT, Duration::from_secs(30));
+    assert_eq!(ASK_MEMORY_WAIT, Duration::from_secs(170));
+    let port = MemoryPort::new("memory.sock".into(), MEMORY_WAIT);
+    assert_eq!(port.wait_for(ToolName::Recall), MEMORY_WAIT);
+    assert_eq!(port.wait_for(ToolName::Memory), MEMORY_WAIT);
+    assert_eq!(port.wait_for(ToolName::AskMemory), ASK_MEMORY_WAIT);
+    let port = MemoryPort::new("memory.sock".into(), SHORT);
+    assert_eq!(port.wait_for(ToolName::Recall), SHORT);
+    assert_eq!(port.wait_for(ToolName::Memory), SHORT);
+    assert_eq!(port.wait_for(ToolName::AskMemory), ASK_MEMORY_WAIT);
+    let port = MemoryPort::new("memory.sock".into(), MEMORY_WAIT).with_ask_wait(LONG);
+    assert_eq!(port.wait_for(ToolName::Recall), MEMORY_WAIT);
+    assert_eq!(port.wait_for(ToolName::Memory), MEMORY_WAIT);
+    assert_eq!(port.wait_for(ToolName::AskMemory), LONG);
+}
+
+/// A slow answer reaches `ask_memory`, whose wait is long, and not `recall` or `memory`, whose
+/// wait is short: the wait is chosen by the call's tool.
+#[test]
+fn a_slow_answer_reaches_ask_memory_only_when_its_wait_is_longer() {
+    let dir = TempDir::new("wait-ask");
+    let port = |socket: &Path| MemoryPort::new(socket.to_path_buf(), SHORT).with_ask_wait(LONG);
+    assert_eq!(
+        ask(&dir, "a.sock", port, request("ask_memory", QUESTION)),
+        found()
+    );
+    assert_eq!(
+        ask(&dir, "r.sock", port, request("recall", QUERY)),
+        unreachable()
+    );
+    let note = r#"{"action":"note","text":"x"}"#;
+    assert_eq!(
+        ask(&dir, "m.sock", port, request("memory", note)),
+        unreachable()
+    );
+}
+
+/// And the other way round: a short wait for `ask_memory` gives up, while `recall` waits.
+#[test]
+fn a_slow_answer_reaches_recall_only_when_its_wait_is_longer() {
+    let dir = TempDir::new("wait-recall");
+    let port = |socket: &Path| MemoryPort::new(socket.to_path_buf(), LONG).with_ask_wait(SHORT);
+    assert_eq!(ask(&dir, "r.sock", port, request("recall", QUERY)), found());
+    assert_eq!(
+        ask(&dir, "a.sock", port, request("ask_memory", QUESTION)),
+        unreachable()
+    );
+}