Return chunked body data as soon as it is available

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/loopd/src/http.rs

@@ -292,9 +292,6 @@ impl<'a> Body<'a> {
                     }
                 },
                 ChunkPhase::Data => {
-                    if buf.len() - written == 0 {
-                        return Ok(written);
-                    }
                     if self.buf.is_empty() {
                         let k = self.stream.read(&mut tmp)?;
                         if k == 0 {
@@ -310,9 +307,8 @@ impl<'a> Body<'a> {
                     self.chunk_remaining -= take;
                     if self.chunk_remaining == 0 {
                         self.phase = ChunkPhase::Crlf;
-                    } else if buf.len() - written == 0 {
-                        return Ok(written);
                     }
+                    return Ok(written);
                 }
                 ChunkPhase::Crlf => {
                     if self.buf.len() >= 2 {

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
+    );
+}

docs/implementer-log.md

@@ -29,6 +29,7 @@ reviewer adds findings under "Reviews" once per milestone.
 | M2a/12-selftest | 2026-09-18 | done | 1 | pass | Ornith-1.5-35B-A3B |
 | M2a/13-verify-device | 2026-09-18 | done | 1 | pass | none |
 | M2a/14-inferproxy-close | 2026-09-18 | done | 2 | fail | none | Made the proxy close towards the client as soon as the server-to-client copy ends, for any reason. `forward` now joins only the s2c thread and returns the c2s JoinHandle, so it returns when the server stops sending instead of waiting for the client to stop sending too; `handle` drops the OpenGuard inside a block scope, then shuts down the client (Both) and server (Both) so the client's read returns EOF at once and the c2s thread ends, then joins c2s. This is rule 4 of task 02 (drop the open-place before closing the client). The half-close when the client stops sending first is unchanged. The copied `forward.rs` is byte-identical to the plan. 7 passed ten runs in a row; `make gate` prints `gate: ok`. | ? | Wrote no library code. Put the two given files in place (`crates/loopd/tests/device.rs`, and a `Makefile` whose only difference from the old one is the new `verify-device` target), confirmed `make gate` prints `gate: ok` with `device` at `0 passed; 0 failed; 4 ignored`, and `curl http://straylight:11434/health` returned `{"status":"ok"}`. `make verify-device` ran the four `#[ignore]` checks one at a time against the real server on slot 0: self-test, capped thinking block, a request surviving its own proxy being killed and restarted, and a second turn reusing the cache of a first turn that contained thinking. All 4 passed in ~21s (well under two minutes). | ? | Wrote crates/loopd/src/selftest.rs (SelfTestError with Display/std::error::Error/From<InferError>, and `run` doing the three checks in order) and registered `pub mod selftest;` in lib.rs; rewrote main.rs into `loopd selftest --config <path>`. Check 1 calls props() and compares chat-template sha256, then n_ctx, then total_slots, a sha256 error being Hash so a wrong server is sent no prompt; check 2 posts one read_file tool with chat_with_retry and requires finish_reason ToolCalls, a first call named read_file whose arguments parse as JSON carrying a string path, wrapping an InferError as Infer; check 3 runs turn 1 then an extension of it and maps a cache Loss to CacheMiss. The copied test's cache_outcome/CacheOutcome live at crate::llama::info, so the import follows that. main.rs parses args as &[&str] via a two-step String->&str collect; unknown/missing args are exit 2 and a config load failure is `loopd: <error>` exit 1. Real server via inferproxy against straylight: minimal.toml gave three step lines and `selftest: ok` exit 0; setting slots=3 gave `selftest: FAILED: slot count: expected 3, got 2` exit 1. | Ornith-1.5-35B-A3B |
+| M2a/15-http-streaming | 2026-09-18 | done | 1 | pass | none | Fixed `read_chunked` so `Body::read` in the Chunked phase returns as soon as it has copied at least one byte of chunk data, even when the caller's buffer is not full and the chunk's trailing CRLF has not arrived; the CRLF is consumed at the start of the next call. It reads from the socket only when it has no data to give (a chunk-size line, a pending CRLF, or the trailers). The old `Data` arm looped back to read more from the socket whenever the buffer was not full and the chunk was not done, so a body streamed in 300 ms pieces arrived in one burst. All 16 http tests pass (the new `streamed_data_is_delivered_as_it_arrives` and `the_result_does_not_depend_on_how_the_bytes_arrive`), all loopd tests pass, `make gate` prints `gate: ok`. `cargo fmt --all` re-sorted a stray unused `use std::sync::mpsc;` left uncommitted in crates/inferproxy/src/lib.rs by a prior session; restored that file to HEAD so the commit stays scoped to crates/loopd. | Ornith-1.5-35B-A3B |
 
 
 ## Reviews