Add the channel messages and the usage record to proto

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

crates/proto/src/lib.rs

@@ -15,8 +15,8 @@ pub use frame::{FrameError, MAX_FRAME, read_frame, write_frame};
 pub use grant::{Constraints, Grant, Mode};
 pub use hash::{HashError, Sha256, sha256};
 pub use ids::{CallId, Epoch, Hash32, SessionId, Timestamp, ValueError};
-pub use log::{LogRecord, ToolCall};
+pub use log::{LogRecord, ToolCall, Usage};
 pub use wire::{
-    DenyReason, Envelope, ErrorCode, Message, PROTOCOL_VERSION, ToolRequest, ToolResponse,
-    WireError,
+    DenyReason, Envelope, ErrorCode, Message, PROTOCOL_VERSION, ToolRequest, ToolResponse, Turn,
+    TurnDone, TurnEvent, WireError,
 };

crates/proto/src/log.rs

@@ -12,6 +12,17 @@ pub struct ToolCall {
     pub arguments: String,
 }
 
+/// What one completion cost. The same five numbers as `LogRecord::Usage`, without the time.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Usage {
+    pub cache_n: u64,
+    pub prompt_n: u64,
+    pub predicted_n: u64,
+    pub reasoning_tokens: u64,
+    pub thinking_capped: bool,
+}
+
 // JSON: {"type":"user","time":"…","content":"…"} ; the tag sits beside the fields; types are snake_case
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
@@ -42,6 +53,14 @@ pub enum LogRecord {
         untrusted: bool,
         truncated: bool,
     },
+    Usage {
+        time: Timestamp,
+        cache_n: u64,
+        prompt_n: u64,
+        predicted_n: u64,
+        reasoning_tokens: u64,
+        thinking_capped: bool,
+    },
     CacheLoss {
         time: Timestamp,
         expected: u64,

crates/proto/src/wire.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 
-use crate::{CallId, DataClass, SessionId, Timestamp};
+use crate::{CallId, DataClass, SessionId, Timestamp, Usage};
 
 pub const PROTOCOL_VERSION: u32 = 1;
 
@@ -24,6 +24,9 @@ pub enum Message {
     ToolRequest(ToolRequest),
     ToolResponse(ToolResponse),
     Error(WireError),
+    Turn(Turn),
+    TurnEvent(TurnEvent),
+    TurnDone(TurnDone),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -40,6 +43,12 @@ pub enum ErrorCode {
     BadVersion,
     BadMessage,
     Internal,
+    SessionFull,
+    TurnLimit,
+    SessionBusy,
+    NoSuchSession,
+    SessionExists,
+    Inference,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -82,3 +91,63 @@ pub enum DenyReason {
     ApprovalRefused,
     ApprovalExpired,
 }
+
+// JSON: {"kind":"turn","body":{"session":"…","content":"…","resume":false}}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Turn {
+    pub session: SessionId,
+    pub content: String,
+    pub resume: bool,
+}
+
+// JSON: {"event":"content","text":"…"} — the tag sits beside the fields, snake_case
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "event", deny_unknown_fields, rename_all = "snake_case")]
+pub enum TurnEvent {
+    Queued {
+        ahead: u64,
+    },
+    Waiting {
+        slot_busy: bool,
+    },
+    Progress {
+        total: u64,
+        cache: u64,
+        processed: u64,
+    },
+    Reasoning {
+        text: String,
+    },
+    Content {
+        text: String,
+    },
+    ToolCallStarted {
+        name: String,
+    },
+    ToolResult {
+        name: String,
+        class: DataClass,
+        truncated: bool,
+    },
+    ThinkingCapped {
+        tokens: u64,
+    },
+    Retrying {
+        attempt: u32,
+        after_ms: u64,
+        error: String,
+    },
+    CacheLoss {
+        expected: u64,
+        got: u64,
+    },
+}
+
+// JSON: {"kind":"turn_done","body":{"content":"…","usage":{…}}}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TurnDone {
+    pub content: String,
+    pub usage: Usage,
+}

crates/proto/tests/fixtures/records/session_usage.jsonl

@@ -0,0 +1,7 @@
+{"type":"session_start","time":"2026-09-18T08:05:00.000Z","session":"chat-1789700000-42","epoch":0,"slot":0,"baseline":"000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f"}
+{"type":"user","time":"2026-09-18T08:05:01.000Z","content":"What time is it?"}
+{"type":"assistant","time":"2026-09-18T08:05:03.000Z","content":null,"reasoning_content":null,"tool_calls":[{"id":"call_a1","name":"clock","arguments":"{}"}]}
+{"type":"usage","time":"2026-09-18T08:05:03.000Z","cache_n":539,"prompt_n":27,"predicted_n":24,"reasoning_tokens":8,"thinking_capped":false}
+{"type":"tool_result","time":"2026-09-18T08:05:03.100Z","call":1,"tool_call_id":"call_a1","content":"2026-09-18T08:05:03.000Z","class":"public","untrusted":false,"truncated":false}
+{"type":"assistant","time":"2026-09-18T08:05:05.000Z","content":"It is five past eight.","reasoning_content":null,"tool_calls":[]}
+{"type":"usage","time":"2026-09-18T08:05:05.000Z","cache_n":589,"prompt_n":40,"predicted_n":64,"reasoning_tokens":27,"thinking_capped":true}

crates/proto/tests/fixtures/wire/error_session_full.json

@@ -0,0 +1 @@
+{"v":1,"id":3,"final":true,"msg":{"kind":"error","body":{"code":"session_full","detail":"this conversation is full; start a new one"}}}

crates/proto/tests/fixtures/wire/turn.json

@@ -0,0 +1 @@
+{"v":1,"id":3,"final":true,"msg":{"kind":"turn","body":{"session":"chat-1789700000-42","content":"What time is it?","resume":false}}}

crates/proto/tests/fixtures/wire/turn_done.json

@@ -0,0 +1 @@
+{"v":1,"id":3,"final":true,"msg":{"kind":"turn_done","body":{"content":"It is noon.","usage":{"cache_n":539,"prompt_n":27,"predicted_n":24,"reasoning_tokens":8,"thinking_capped":false}}}}

crates/proto/tests/fixtures/wire/turn_event_content.json

@@ -0,0 +1 @@
+{"v":1,"id":3,"final":false,"msg":{"kind":"turn_event","body":{"event":"content","text":"It is "}}}

crates/proto/tests/fixtures/wire/turn_event_retrying.json

@@ -0,0 +1 @@
+{"v":1,"id":3,"final":false,"msg":{"kind":"turn_event","body":{"event":"retrying","attempt":2,"after_ms":1500,"error":"the server went silent"}}}

crates/proto/tests/fixtures/wire/turn_event_tool_result.json

@@ -0,0 +1 @@
+{"v":1,"id":3,"final":false,"msg":{"kind":"turn_event","body":{"event":"tool_result","name":"clock","class":"public","truncated":false}}}

crates/proto/tests/strict.rs

@@ -69,11 +69,18 @@ fn envelopes_reject_unknown_keys_at_every_depth() {
         "tool_response_failed.json",
         "tool_response_denied.json",
         "error.json",
+        "turn.json",
+        "turn_event_tool_result.json",
+        "turn_event_content.json",
+        "turn_event_retrying.json",
+        "turn_done.json",
+        "error_session_full.json",
     ] {
-        // Envelope, msg and body: three objects.
+        // Envelope, msg and body: three objects; turn_done also has a usage object.
+        let want = if name == "turn_done.json" { 4 } else { 3 };
         assert_eq!(
             check::<Envelope>(name, &fixture(&format!("wire/{name}"))),
-            3,
+            want,
             "{name}"
         );
     }
@@ -100,6 +107,16 @@ fn log_records_reject_unknown_keys_at_every_depth() {
     assert_eq!(tried, 8);
 }
 
+#[test]
+fn usage_log_records_reject_unknown_keys_at_every_depth() {
+    let mut tried = 0;
+    for (i, line) in fixture("records/session_usage.jsonl").lines().enumerate() {
+        tried += check::<LogRecord>(&format!("session_usage.jsonl:{}", i + 1), line);
+    }
+    // Seven records, plus the one tool call inside the first assistant record.
+    assert_eq!(tried, 8);
+}
+
 #[test]
 fn grants_reject_unknown_keys_at_every_depth() {
     let grant: Grant = toml::from_str(&fixture("grant/full.toml")).unwrap();

crates/proto/tests/turn_wire.rs

@@ -0,0 +1,251 @@
+//! Tests for the channel messages and the usage record, against byte-exact fixtures. Do not edit.
+
+use proto::{
+    CallId, DataClass, Envelope, Epoch, ErrorCode, Hash32, LogRecord, Message, SessionId,
+    Timestamp, ToolCall, Turn, TurnDone, TurnEvent, Usage, WireError,
+};
+
+fn fixture(kind: &str, name: &str) -> String {
+    let path = format!(
+        "{}/tests/fixtures/{kind}/{name}",
+        env!("CARGO_MANIFEST_DIR")
+    );
+    std::fs::read_to_string(&path)
+        .unwrap_or_else(|e| panic!("{path}: {e}"))
+        .trim_end_matches('\n')
+        .to_string()
+}
+
+fn check(name: &str, want: Envelope) {
+    let text = fixture("wire", name);
+    let got: Envelope = serde_json::from_str(&text).unwrap_or_else(|e| panic!("{name}: {e}"));
+    assert_eq!(got, want, "{name}: decoded value");
+    assert_eq!(
+        serde_json::to_string(&want).unwrap(),
+        text,
+        "{name}: encoded bytes"
+    );
+}
+
+fn env(id: u64, r#final: bool, msg: Message) -> Envelope {
+    Envelope {
+        v: 1,
+        id,
+        r#final,
+        msg,
+    }
+}
+
+fn usage() -> Usage {
+    Usage {
+        cache_n: 539,
+        prompt_n: 27,
+        predicted_n: 24,
+        reasoning_tokens: 8,
+        thinking_capped: false,
+    }
+}
+
+#[test]
+fn turn() {
+    let body = Turn {
+        session: SessionId::new("chat-1789700000-42").unwrap(),
+        content: "What time is it?".to_string(),
+        resume: false,
+    };
+    check("turn.json", env(3, true, Message::Turn(body)));
+}
+
+#[test]
+fn turn_events() {
+    check(
+        "turn_event_tool_result.json",
+        env(
+            3,
+            false,
+            Message::TurnEvent(TurnEvent::ToolResult {
+                name: "clock".to_string(),
+                class: DataClass::Public,
+                truncated: false,
+            }),
+        ),
+    );
+    check(
+        "turn_event_content.json",
+        env(
+            3,
+            false,
+            Message::TurnEvent(TurnEvent::Content {
+                text: "It is ".to_string(),
+            }),
+        ),
+    );
+    check(
+        "turn_event_retrying.json",
+        env(
+            3,
+            false,
+            Message::TurnEvent(TurnEvent::Retrying {
+                attempt: 2,
+                after_ms: 1500,
+                error: "the server went silent".to_string(),
+            }),
+        ),
+    );
+}
+
+#[test]
+fn turn_done_and_the_new_error_codes() {
+    check(
+        "turn_done.json",
+        env(
+            3,
+            true,
+            Message::TurnDone(TurnDone {
+                content: "It is noon.".to_string(),
+                usage: usage(),
+            }),
+        ),
+    );
+    let body = WireError {
+        code: ErrorCode::SessionFull,
+        detail: "this conversation is full; start a new one".to_string(),
+    };
+    check(
+        "error_session_full.json",
+        env(3, true, Message::Error(body)),
+    );
+    let codes = [
+        (ErrorCode::SessionFull, "session_full"),
+        (ErrorCode::TurnLimit, "turn_limit"),
+        (ErrorCode::SessionBusy, "session_busy"),
+        (ErrorCode::NoSuchSession, "no_such_session"),
+        (ErrorCode::SessionExists, "session_exists"),
+        (ErrorCode::Inference, "inference"),
+    ];
+    for (value, text) in codes {
+        assert_eq!(
+            serde_json::to_string(&value).unwrap(),
+            format!("\"{text}\"")
+        );
+    }
+}
+
+#[test]
+fn every_turn_event_kind_round_trips() {
+    let all = vec![
+        TurnEvent::Queued { ahead: 1 },
+        TurnEvent::Waiting { slot_busy: true },
+        TurnEvent::Progress {
+            total: 100,
+            cache: 50,
+            processed: 75,
+        },
+        TurnEvent::Reasoning {
+            text: "hm".to_string(),
+        },
+        TurnEvent::Content {
+            text: "hi".to_string(),
+        },
+        TurnEvent::ToolCallStarted {
+            name: "clock".to_string(),
+        },
+        TurnEvent::ToolResult {
+            name: "clock".to_string(),
+            class: DataClass::Secret,
+            truncated: true,
+        },
+        TurnEvent::ThinkingCapped { tokens: 4096 },
+        TurnEvent::Retrying {
+            attempt: 2,
+            after_ms: 10,
+            error: "x".to_string(),
+        },
+        TurnEvent::CacheLoss {
+            expected: 500,
+            got: 20,
+        },
+    ];
+    for event in all {
+        let text = serde_json::to_string(&event).unwrap();
+        assert!(text.starts_with("{\"event\":\""), "{text}");
+        assert_eq!(serde_json::from_str::<TurnEvent>(&text).unwrap(), event);
+    }
+    assert!(serde_json::from_str::<TurnEvent>(r#"{"event":"content","text":"x","zz":1}"#).is_err());
+    assert!(serde_json::from_str::<TurnEvent>(r#"{"event":"dance"}"#).is_err());
+}
+
+#[test]
+fn usage_records() {
+    let text = fixture("records", "session_usage.jsonl");
+    let lines: Vec<&str> = text.lines().collect();
+    let want = [
+        LogRecord::SessionStart {
+            time: Timestamp::parse("2026-09-18T08:05:00.000Z").unwrap(),
+            session: SessionId::new("chat-1789700000-42").unwrap(),
+            epoch: Epoch(0),
+            slot: 0,
+            baseline: Hash32::from_hex(
+                "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f",
+            )
+            .unwrap(),
+        },
+        LogRecord::User {
+            time: Timestamp::parse("2026-09-18T08:05:01.000Z").unwrap(),
+            content: "What time is it?".to_string(),
+        },
+        LogRecord::Assistant {
+            time: Timestamp::parse("2026-09-18T08:05:03.000Z").unwrap(),
+            content: None,
+            reasoning_content: None,
+            tool_calls: vec![ToolCall {
+                id: "call_a1".to_string(),
+                name: "clock".to_string(),
+                arguments: "{}".to_string(),
+            }],
+        },
+        LogRecord::Usage {
+            time: Timestamp::parse("2026-09-18T08:05:03.000Z").unwrap(),
+            cache_n: 539,
+            prompt_n: 27,
+            predicted_n: 24,
+            reasoning_tokens: 8,
+            thinking_capped: false,
+        },
+        LogRecord::ToolResult {
+            time: Timestamp::parse("2026-09-18T08:05:03.100Z").unwrap(),
+            call: CallId(1),
+            tool_call_id: "call_a1".to_string(),
+            content: "2026-09-18T08:05:03.000Z".to_string(),
+            class: DataClass::Public,
+            untrusted: false,
+            truncated: false,
+        },
+        LogRecord::Assistant {
+            time: Timestamp::parse("2026-09-18T08:05:05.000Z").unwrap(),
+            content: Some("It is five past eight.".to_string()),
+            reasoning_content: None,
+            tool_calls: vec![],
+        },
+        LogRecord::Usage {
+            time: Timestamp::parse("2026-09-18T08:05:05.000Z").unwrap(),
+            cache_n: 589,
+            prompt_n: 40,
+            predicted_n: 64,
+            reasoning_tokens: 27,
+            thinking_capped: true,
+        },
+    ];
+    assert_eq!(lines.len(), want.len());
+    for (i, (line, want)) in lines.iter().zip(&want).enumerate() {
+        let got: LogRecord =
+            serde_json::from_str(line).unwrap_or_else(|e| panic!("line {}: {e}", i + 1));
+        assert_eq!(&got, want, "line {}", i + 1);
+        assert_eq!(
+            &serde_json::to_string(want).unwrap(),
+            line,
+            "line {}: bytes",
+            i + 1
+        );
+    }
+}

docs/implementer-log.md

@@ -30,6 +30,7 @@ reviewer adds findings under "Reviews" once per milestone.
 | 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 |
+| M2b/01-proto-channel-types | 2026-09-18 | done | 1 | pass | none | Added `Usage` struct and a `Usage` variant (between ToolResult and CacheLoss) in log.rs, and `Turn`, `TurnEvent`, `TurnDone` plus six `ErrorCode` variants (SessionFull..Inference) and three `Message` variants (after Error) in wire.rs; re-exported Usage, Turn, TurnEvent, TurnDone from lib.rs. All four new types carry `deny_unknown_fields`; field order matches the byte-exact fixtures (attempt/after_ms/error, name/class/truncated). 55 proto tests pass (turn_wire 5, strict 5, wire 9, ids 12, frame 13, grant 4, hash 4, records 3) and `make gate` prints `gate: ok`; the old fixtures stay byte-identical. One duplicate block of the three wire types left by an interrupted edit had to be removed mid-task. | Ornith-1.5-35B-A3B |
 
 
 ## Reviews