Review M2a: accept with two follow-up tasks; record models and lessons

The branch passes every check, including make verify-device on
straylight and repeated timing runs under load. Reading and probing
found that inferproxy does not pass an upstream close on to a client
that is still sending, and that the chunked body reader delivers a
stream only when the caller's buffer fills or the stream ends. Both
were also gaps in the tasks and tests, so tasks 14 and 15 carry the
fixes with new tests checked against the reference.

The Model column is corrected: tasks 04 to 06 and 08 to 13 were Ornith.
Lessons gain four implementer tips and five task-writing tips; three
rules are promoted to AGENTS.md.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

docs/plans/M2a/14-inferproxy-close.md

@@ -0,0 +1,52 @@
+# M2a task 14: pass the upstream's close on to the client (review follow-up)
+
+**Branch:** `m2a` (run `git switch m2a`; `git status --short` must be empty, otherwise stop)
+**Commit subject:** `Close the client side when the upstream closes, in inferproxy`
+
+## Goal
+
+The M2a review found that `inferproxy` only closes towards the client after the client has stopped
+sending. `loopd`'s HTTP client never stops sending on its own: it keeps its side open until it
+drops the connection. So when `llama-server` closes, or dies in the middle of an answer, the client
+is not told; it sits until its own liveness timeout and reports `Stalled` instead of
+`StreamClosedEarly`, thirty seconds late. A body that runs to the close would hang the same way.
+
+Task 02, rule 3, said: "When the server stops sending, the exchange is over: close both." The
+current code waits for both copy threads to finish instead. This was partly the plan's fault: every
+test client in `forward.rs` half-closed after sending, so the gap could not show. The new test
+sends without half-closing.
+
+## Files
+
+- Copy (replacing the old one): `crates/inferproxy/tests/forward.rs`
+- Modify: `crates/inferproxy/src/lib.rs`, `docs/implementer-log.md`
+
+## Required behaviour
+
+- When the server-to-client copy ends, for any reason, shut down the client connection (both
+  directions) and the server connection, so that the client's read returns 0 at once and the
+  client-to-server copy thread ends. Do this **after** releasing the open-connection place, as
+  rule 4 of task 02 requires.
+- Everything else from task 02 still holds, including the half-close when the client stops
+  sending first.
+
+## Steps
+
+- [ ] **1. Copy.** `git switch m2a`, then
+  `cp docs/plans/M2a/files/crates/inferproxy/tests/forward.rs crates/inferproxy/tests/`
+- [ ] **2. See it fail.** `cargo test -p inferproxy --test forward`. Expected: 1 of 7 fails,
+  `the_upstreams_close_reaches_a_client_that_is_still_sending`, with a timeout.
+- [ ] **3. Fix `lib.rs`.** Run `cargo fmt --all`.
+- [ ] **4. See it pass.** `cargo test -p inferproxy --test forward`, ten times in a row. Expected:
+  `7 passed` every time.
+- [ ] **5. Run the gate.** `make gate`. Expected last line: `gate: ok`.
+- [ ] **6. Log and commit.** `git add crates/inferproxy docs/implementer-log.md && git commit`
+
+## Done when
+
+- `cargo test -p inferproxy --test forward` reports 7 passed, ten runs in a row; `make gate` prints
+  `gate: ok`; `cmp crates/inferproxy/tests/forward.rs docs/plans/M2a/files/crates/inferproxy/tests/forward.rs` prints nothing.
+
+## Stop and report if
+
+- Passing the new test breaks `forwards_both_ways_and_passes_the_half_close_on`.

docs/plans/M2a/15-http-streaming.md

@@ -0,0 +1,60 @@
+# M2a task 15: deliver streamed data as it arrives (review follow-up)
+
+**Branch:** `m2a` (run `git switch m2a`; `git status --short` must be empty, otherwise stop)
+**Commit subject:** `Return chunked body data as soon as it is available`
+
+## Goal
+
+The M2a review found that the chunked body reader in `http.rs` keeps reading until the caller's
+buffer is full or the stream ends. With an 8 KiB buffer that means a streamed completion reaches
+`loopd` in bursts of about twenty events, and a short completion arrives all at once at its end.
+The reviewer measured it: with events sent 300 ms apart, every event was delivered at the moment
+the last one arrived. Three things follow: the thinking cap fires up to twenty chunks late, the
+text a user sees in M2b would come in lumps, and bytes already copied into the caller's buffer are
+lost when a later read in the same call times out.
+
+Task 04 did not state the contract of `std::io::Read`, which is the plan's fault. The contract is:
+**a `read` returns as soon as it has any data to give. It blocks only when it has none.** The test
+you copy in sends a recording in pieces 300 ms apart and requires the first data within 200 ms and
+the rest spread out over the run.
+
+## Files
+
+- Copy (replacing the old one): `crates/loopd/tests/http.rs`
+- Modify: `crates/loopd/src/http.rs`, `docs/implementer-log.md`
+
+## Required behaviour
+
+In the chunked framing, `Body::read`:
+
+1. Returns as soon as it has copied at least one byte of chunk data into the caller's buffer, even
+   if the buffer is not full and even if the chunk's trailing `\r\n` has not arrived yet. Consume
+   that `\r\n` at the start of the **next** call.
+2. Reads from the socket only when it has no data to give: at a chunk-size line, at a pending
+   `\r\n`, or at the trailers.
+3. Everything else from task 04 still holds: the same bytes come out however they arrive, errors
+   and limits are unchanged, and the 15 earlier tests still pass.
+
+The length and close framings already behave this way.
+
+## Steps
+
+- [ ] **1. Copy.** `git switch m2a`, then
+  `cp docs/plans/M2a/files/crates/loopd/tests/http.rs crates/loopd/tests/`
+- [ ] **2. See it fail.** `cargo test -p loopd --test http`. Expected: 1 of 16 fails,
+  `streamed_data_is_delivered_as_it_arrives`, saying the reader is buffering the stream.
+- [ ] **3. Fix `read_chunked`.** Run `cargo fmt --all`.
+- [ ] **4. See it pass.** `cargo test -p loopd`. Expected: `http` 16 passed and every other file as
+  before (the `chat`, `cap`, `sse` and `selftest` tests read through this code).
+- [ ] **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
+
+- `cargo test -p loopd --test http` reports 16 passed; `make gate` prints `gate: ok`;
+  `cmp crates/loopd/tests/http.rs docs/plans/M2a/files/crates/loopd/tests/http.rs` prints nothing.
+
+## Stop and report if
+
+- `the_result_does_not_depend_on_how_the_bytes_arrive` fails after the change: the fix has broken
+  the framing somewhere.

docs/plans/M2a/README.md

@@ -1,7 +1,7 @@
 # M2a implementation plan: the inference path
 
 > **For the implementing model:** do not work from this file. The owner gives you one task file at
-> a time (`01-…` to `13-…`). This file is the index for the owner and the reviewer.
+> a time (`01-…` to `15-…`). This file is the index for the owner and the reviewer.
 
 **Goal:** `loopd` can hold a correct, robust conversation with `llama-server` through a Unix
 socket: requests built from typed input, streams reassembled exactly, every kind of silence and
@@ -28,7 +28,7 @@ fake server replaying responses recorded from straylight.
   use and newer builds add more, so structs that parse the server's responses must not use
   `deny_unknown_fields`. Each task says which kind it is dealing with.
 - Branch `m2a`. One task, one fresh OpenCode session, one commit. Run `cargo fmt --all` before the
-  gate. Review happens once, after task 13.
+  gate. Review happened once, after task 13; tasks 14 and 15 are its follow-ups.
 
 ## Tasks
 
@@ -47,6 +47,8 @@ fake server replaying responses recorded from straylight.
 | 11 | `11-llama-gate-retry.md` | `SlotGate`, `chat_with_retry` | `loopd/tests/retry.rs` |
 | 12 | `12-selftest.md` | `loopd::selftest`, `loopd selftest --config` | `loopd/tests/selftest.rs` |
 | 13 | `13-verify-device.md` | `make verify-device` against straylight | `loopd/tests/device.rs` |
+| 14 | `14-inferproxy-close.md` | Review follow-up: `inferproxy` closes towards the client when the upstream closes | updated `inferproxy/tests/forward.rs` |
+| 15 | `15-http-streaming.md` | Review follow-up: chunked body data is returned as soon as it is available | updated `loopd/tests/http.rs` |
 
 `files/` holds everything the tasks copy into place: tests, the fake server
 (`loopd/tests/support/mod.rs`), recordings (`fixtures/http/*.http`), expected results derived from

docs/plans/M2a/files/crates/inferproxy/tests/forward.rs

@@ -154,3 +154,44 @@ fn an_unreachable_upstream_closes_the_client() {
     let path = start("127.0.0.1:1".to_string(), Limits::default());
     assert!(was_refused(&path));
 }
+
+/// An upstream that answers at once and closes, whether or not the client has finished sending.
+/// This is what `llama-server` does, and what a server that dies mid-answer looks like.
+fn answer_and_close_upstream() -> String {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap().to_string();
+    thread::spawn(move || {
+        for stream in listener.incoming() {
+            let mut stream = stream.unwrap();
+            let mut first = [0u8; 1];
+            let _ = stream.read(&mut first);
+            stream.write_all(b"answer").unwrap();
+            // dropping `stream` closes it
+        }
+    });
+    addr
+}
+
+#[test]
+fn the_upstreams_close_reaches_a_client_that_is_still_sending() {
+    // The HTTP client in loopd never half-closes: it keeps its sending side open until it drops
+    // the connection. When the server closes, the proxy must close towards the client at once,
+    // or the client only learns of a dead server from its own timeout.
+    let path = start(answer_and_close_upstream(), Limits::default());
+    let mut s = UnixStream::connect(&path).unwrap();
+    s.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+    s.write_all(b"request").unwrap();
+    let started = std::time::Instant::now();
+    let mut got = Vec::new();
+    let result = s.read_to_end(&mut got);
+    assert!(
+        result.is_ok(),
+        "the read must end with EOF, not a timeout: {result:?}"
+    );
+    assert_eq!(got, b"answer");
+    assert!(
+        started.elapsed() < Duration::from_millis(1000),
+        "EOF took {:?}",
+        started.elapsed()
+    );
+}

docs/plans/M2a/files/crates/loopd/tests/http.rs

@@ -308,3 +308,46 @@ fn a_missing_socket_is_a_connect_error() {
     let e: Box<dyn std::error::Error> = Box::new(HttpError::Timeout);
     assert!(!e.to_string().is_empty());
 }
+
+/// A streamed body is read while it arrives. A `read` that already has data must not block
+/// waiting for more, or every event is delivered one buffer late and the last ones only when
+/// the stream ends.
+#[test]
+fn streamed_data_is_delivered_as_it_arrives() {
+    let server = FakeServer::start();
+    let reply = Reply::fixture("turn1");
+    let piece = reply.offset_after_events(1);
+    // About one event per piece, 300 ms apart: the whole stream takes over a second.
+    server.route("/v1/chat/completions", vec![reply.trickle(piece, 300)]);
+    let mut conn = Connection::open(&server.socket).unwrap();
+    conn.set_read_timeout(Duration::from_secs(5)).unwrap();
+    conn.send(&Request {
+        method: "POST",
+        path: "/v1/chat/completions",
+        body: Some(b"{}"),
+    })
+    .unwrap();
+    let head = conn.read_head().unwrap();
+    let started = std::time::Instant::now();
+    let mut body = conn.body(&head).unwrap();
+    let mut arrivals = Vec::new();
+    let mut buf = [0u8; 8192];
+    loop {
+        let n = std::io::Read::read(&mut body, &mut buf).unwrap();
+        if n == 0 {
+            break;
+        }
+        arrivals.push(started.elapsed());
+    }
+    let first = arrivals.first().copied().unwrap_or_default();
+    let last = arrivals.last().copied().unwrap_or_default();
+    assert!(
+        first < Duration::from_millis(200),
+        "the first data came only after {first:?}"
+    );
+    assert!(
+        last - first >= Duration::from_millis(600),
+        "everything arrived within {:?} of the first read: the reader is buffering the stream",
+        last - first
+    );
+}