From b274fa7d6248a80b42d5f664cf7b9eea53af6e3a Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 3 Oct 2026 00:38:35 -0700 Subject: [PATCH] loopd: compact after a full turn and retry SessionFull once Co-Authored-By: ornith-1.5-35b-a3b --- crates/loopd/src/channel.rs | 35 ++-- crates/loopd/src/compact.rs | 51 ++++++ crates/loopd/tests/compact_turn.rs | 277 +++++++++++++++++++++++++++++ docs/implementer-log.md | 1 + 4 files changed, 353 insertions(+), 11 deletions(-) create mode 100644 crates/loopd/tests/compact_turn.rs diff --git a/crates/loopd/src/channel.rs b/crates/loopd/src/channel.rs index 5eb43b5..6c39b3c 100644 --- a/crates/loopd/src/channel.rs +++ b/crates/loopd/src/channel.rs @@ -178,25 +178,38 @@ pub fn handle(stream: UnixStream, ctx: Arc) { } }; + let mut emit = |event: &bxcore::TurnEvent| { + let _ = write_frame( + &mut stream, + &Envelope { + v: PROTOCOL_VERSION, + id: request.id, + r#final: false, + msg: Message::TurnEvent(event.clone()), + }, + ); + }; + let outcome = run_turn( &mut session, &runtime, &turn.content, turn.speaker.as_deref(), - &mut |event| { - let _ = write_frame( - &mut stream, - &Envelope { - v: PROTOCOL_VERSION, - id: request.id, - r#final: false, - msg: Message::TurnEvent(event.clone()), - }, - ); - }, + &mut emit, + ); + let outcome = crate::compact::finish_turn( + &mut session, + &runtime, + &*ctx.memory, + &turn.content, + turn.speaker.as_deref(), + &mut emit, + outcome, ); // 5. Release the session before the final frame, so the next turn does not find it busy. + // Compaction ran inside the hold: a second turn on this session waits until the new + // epoch is on disk (or compact failed and the epoch is unchanged). drop(held); // A finished turn's records are for `memoryd` to index; best effort, since `memoryd` catches diff --git a/crates/loopd/src/compact.rs b/crates/loopd/src/compact.rs index f6e8ca3..b1ffe83 100644 --- a/crates/loopd/src/compact.rs +++ b/crates/loopd/src/compact.rs @@ -11,6 +11,8 @@ use crate::memory::MemorySource; use crate::session::Session; use crate::tools::Registry; use crate::turn::is_context_full; +use crate::turn::{Runtime, TurnError, TurnOutcome, run_turn}; +use bxcore::TurnEvent; /// Why a summary call or a next-epoch write failed. Every variant's `Display` ends with the /// runbook pointer `session-compact-failed`. @@ -133,3 +135,52 @@ pub fn commit( })?; Ok(()) } + +/// After `run_turn`: compact when the last `Usage` is due, or on `SessionFull` (then retry the +/// user message once). Compact failure on a successful turn keeps the success; on `SessionFull` +/// keeps `SessionFull`. Other errors skip compact. +pub fn finish_turn( + session: &mut Session, + rt: &Runtime<'_>, + memory: &dyn MemorySource, + text: &str, + speaker: Option<&str>, + on_event: &mut dyn FnMut(&TurnEvent), + outcome: Result, +) -> Result { + match outcome { + Err(TurnError::SessionFull) => match compact_session(session, rt, memory) { + Ok(()) => run_turn(session, rt, text, speaker, on_event), + Err(e) => { + log_compact(session, &e); + Err(TurnError::SessionFull) + } + }, + Ok(done) => { + if let Some(usage) = session.last_usage() + && due(&usage, rt.cfg.compact.threshold, rt.cfg.expect.n_ctx) + && let Err(e) = compact_session(session, rt, memory) + { + log_compact(session, &e); + } + Ok(done) + } + Err(e) => Err(e), + } +} + +fn compact_session( + session: &mut Session, + rt: &Runtime<'_>, + memory: &dyn MemorySource, +) -> Result<(), CompactError> { + let summary = summarise(rt.cfg, rt.client, session)?; + commit(session, rt.cfg, rt.registry, memory, summary) +} + +fn log_compact(session: &Session, e: &CompactError) { + eprintln!( + "loopd: could not compact session {}: {e}", + session.id().as_str() + ); +} diff --git a/crates/loopd/tests/compact_turn.rs b/crates/loopd/tests/compact_turn.rs new file mode 100644 index 0000000..1612189 --- /dev/null +++ b/crates/loopd/tests/compact_turn.rs @@ -0,0 +1,277 @@ +//! Tests for compacting after a turn and retrying a full context. Do not edit. + +mod support; + +use bxcore::{Epoch, LogRecord, SessionId, TurnEvent}; +use loopd::baseline::{Baseline, CARRY_HEADING}; +use loopd::compact; +use loopd::llama::Client; +use loopd::memory::NoMemory; +use loopd::session::Session; +use loopd::tools::Registry; +use loopd::turn::{Runtime, TurnError, run_turn}; +use support::{FakeServer, Home, Reply}; + +const CHAT: &str = "/v1/chat/completions"; +const PROMPT: &str = "Summarise this conversation for the next epoch.\n"; + +fn sse(content: &str, cache_n: u64, prompt_n: u64) -> Reply { + let content_json = serde_json::to_string(content).unwrap(); + let data1 = format!( + r#"data: {{"choices":[{{"finish_reason":null,"index":0,"delta":{{"content":{content_json}}}}}],"id":"t","timings":{{"cache_n":{cache_n},"prompt_n":{prompt_n},"predicted_n":1}}}}"# + ); + let data2 = format!( + r#"data: {{"choices":[{{"finish_reason":"stop","index":0,"delta":{{}}}}],"id":"t","timings":{{"cache_n":{cache_n},"prompt_n":{prompt_n},"predicted_n":1}}}}"# + ); + let body = format!("{data1}\n\n{data2}\n\ndata: [DONE]\n\n"); + Reply::raw(format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + )) +} + +struct World { + home: Home, + server: FakeServer, + cfg: loopd::config::Config, + client: Client, + port: support::ScriptedPort, + registry: Registry, +} + +impl World { + fn new() -> World { + let home = Home::new(); + home.write("compact.md", PROMPT); + let server = FakeServer::start(); + let mut cfg = home.config(&server.socket); + cfg.expect.n_ctx = 1_000; + cfg.compact.threshold = 0.85; + let client = Client::new(cfg.clone()); + World { + home, + server, + cfg, + client, + port: support::ScriptedPort::new(vec![]), + registry: Registry::m2b(), + } + } + + fn session(&self) -> Session { + let baseline = Baseline::assemble(&self.cfg, &self.registry).unwrap(); + Session::create( + &self.home.dir, + SessionId::new("a").unwrap(), + baseline, + self.cfg.slots.main, + ) + .unwrap() + } + + fn runtime(&self) -> Runtime<'_> { + Runtime { + cfg: &self.cfg, + client: &self.client, + port: &self.port, + registry: &self.registry, + } + } + + fn finish( + &self, + session: &mut Session, + text: &str, + outcome: Result, + ) -> Result { + let mut events = Vec::new(); + compact::finish_turn( + session, + &self.runtime(), + &NoMemory, + text, + None, + &mut |e: &TurnEvent| events.push(e.clone()), + outcome, + ) + } +} + +#[test] +fn usage_under_the_threshold_leaves_the_epoch() { + let w = World::new(); + w.server.route(CHAT, vec![sse("ok", 400, 449)]); + let mut s = w.session(); + let mut events = Vec::new(); + let outcome = run_turn(&mut s, &w.runtime(), "hi", None, &mut |e| { + events.push(e.clone()) + }); + let outcome = w.finish(&mut s, "hi", outcome).unwrap(); + assert_eq!(outcome.content, "ok"); + assert_eq!(s.epoch(), Epoch(0)); + assert_eq!(w.server.requests_to(CHAT).len(), 1); +} + +#[test] +fn usage_at_the_threshold_compacts_before_the_turn_returns() { + let w = World::new(); + w.server + .route(CHAT, vec![sse("ok", 400, 450), sse("the sum", 0, 10)]); + let mut s = w.session(); + let mut events = Vec::new(); + let outcome = run_turn(&mut s, &w.runtime(), "hi", None, &mut |e| { + events.push(e.clone()) + }); + w.finish(&mut s, "hi", outcome).unwrap(); + assert_eq!(s.epoch(), Epoch(1)); + assert_eq!(s.carry(), Some("the sum")); + w.server.route(CHAT, vec![sse("next", 0, 20)]); + let mut events = Vec::new(); + run_turn(&mut s, &w.runtime(), "again", None, &mut |e| { + events.push(e.clone()) + }) + .unwrap(); + let sent = w.server.requests_to(CHAT); + assert_eq!(sent.len(), 3); + assert_eq!(sent[0].json()["id_slot"], 0); + assert_eq!(sent[1].json()["id_slot"], 1); + assert!(sent[1].json().get("tools").is_none()); + let live = sent[2].json(); + assert_eq!(live["id_slot"], 0); + let contents: Vec<&str> = live["messages"] + .as_array() + .unwrap() + .iter() + .map(|m| m["content"].as_str().unwrap_or("")) + .collect(); + assert!( + contents + .iter() + .any(|c| c.contains(CARRY_HEADING) && c.contains("the sum")), + "{contents:?}" + ); + let old = w.home.records("a"); + assert!(matches!( + old.last(), + Some(LogRecord::EpochEnd { next, .. }) if *next == Epoch(1) + )); +} + +#[test] +fn session_full_compacts_and_retries_the_user_message_once() { + let w = World::new(); + w.server.route( + CHAT, + vec![ + Reply::fixture("context_full"), + sse("the sum", 0, 10), + sse("retried", 0, 20), + ], + ); + let mut s = w.session(); + let mut events = Vec::new(); + let outcome = run_turn(&mut s, &w.runtime(), "please fit", None, &mut |e| { + events.push(e.clone()) + }); + assert!(matches!(outcome, Err(TurnError::SessionFull))); + let done = w.finish(&mut s, "please fit", outcome).unwrap(); + assert_eq!(done.content, "retried"); + assert_eq!(s.epoch(), Epoch(1)); + let users: Vec<_> = s + .records() + .iter() + .filter_map(|r| match r { + LogRecord::User { content, .. } => Some(content.as_str()), + _ => None, + }) + .collect(); + assert_eq!(users, ["please fit"]); + let old_users = w + .home + .records("a") + .iter() + .filter(|r| matches!(r, LogRecord::User { .. })) + .count(); + assert_eq!(old_users, 1); + assert_eq!(s.turns(), 2); + let sent = w.server.requests_to(CHAT); + assert_eq!(sent.len(), 3); + assert_eq!(sent[1].json()["id_slot"], 1); + assert_eq!(sent[2].json()["id_slot"], 0); +} + +#[test] +fn session_full_stays_session_full_when_compact_fails() { + let w = World::new(); + w.server.route( + CHAT, + vec![ + Reply::fixture("context_full"), + Reply::fixture("bad_request"), + ], + ); + let mut s = w.session(); + let mut events = Vec::new(); + let outcome = run_turn(&mut s, &w.runtime(), "x", None, &mut |e| { + events.push(e.clone()) + }); + let err = w.finish(&mut s, "x", outcome).unwrap_err(); + assert!(matches!(err, TurnError::SessionFull), "{err:?}"); + assert_eq!(s.epoch(), Epoch(0)); +} + +#[test] +fn a_successful_turn_stays_successful_when_compact_fails() { + let home = Home::new(); + let server = FakeServer::start(); + let mut cfg = home.config(&server.socket); + cfg.expect.n_ctx = 1_000; + cfg.compact.threshold = 0.85; + let client = Client::new(cfg.clone()); + server.route(CHAT, vec![sse("ok", 400, 450)]); + let baseline = Baseline::assemble(&cfg, &Registry::m2b()).unwrap(); + let mut s = Session::create( + &home.dir, + SessionId::new("a").unwrap(), + baseline, + cfg.slots.main, + ) + .unwrap(); + let port = support::ScriptedPort::new(vec![]); + let registry = Registry::m2b(); + let rt = Runtime { + cfg: &cfg, + client: &client, + port: &port, + registry: ®istry, + }; + let mut events = Vec::new(); + let outcome = run_turn(&mut s, &rt, "hi", None, &mut |e| events.push(e.clone())); + let done = compact::finish_turn( + &mut s, + &rt, + &NoMemory, + "hi", + None, + &mut |e| events.push(e.clone()), + outcome, + ) + .unwrap(); + assert_eq!(done.content, "ok"); + assert_eq!(s.epoch(), Epoch(0), "compact.md is missing, so no epoch"); +} + +#[test] +fn finish_turn_does_not_compact_an_error_other_than_session_full() { + let w = World::new(); + w.server.route(CHAT, vec![Reply::fixture("bad_request")]); + let mut s = w.session(); + let mut events = Vec::new(); + let outcome = run_turn(&mut s, &w.runtime(), "x", None, &mut |e| { + events.push(e.clone()) + }); + let err = w.finish(&mut s, "x", outcome).unwrap_err(); + assert!(matches!(err, TurnError::Infer(_)), "{err:?}"); + assert_eq!(s.epoch(), Epoch(0)); + assert_eq!(w.server.requests_to(CHAT).len(), 1); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index 47ac34d..7b057a2 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone. | Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model | |---|---|---|---|---|---|---|---| +| M5d1/08-loopd-compact-turn | 2026-10-03 | done | 3 | pass | none | Copied the given test `crates/loopd/tests/compact_turn.rs` (new). Step 2 (`cargo build --tests --keep-going -p loopd`) showed exactly the named failure: `compact_turn` not compiling with E0425 (`cannot find function finish_turn in module compact`) — the task named E0599, this is E0425 because it is a free function not a method; every other target compiled, including `turn` and `channel`. Made F1 (`compact.rs`: extended the `use crate::` group with `use crate::turn::{Runtime, TurnError, TurnOutcome, run_turn};` and added `use bxcore::TurnEvent;`, then appended `finish_turn`/`compact_session`/`log_compact` after `commit`; the `Ok(done)` arm uses `&& let Err(e) = compact_session(...)` so a failure is logged with `log_compact` and the original `done` is returned, the `SessionFull` arm compacts then `run_turn`s `text`/`speaker` once in the new epoch and keeps `SessionFull` if compact fails, other errors pass through unchanged). Made H1 (`channel.rs`: extracted the inline event closure to a `mut emit` binding, called `run_turn` then `crate::compact::finish_turn(&mut session, &runtime, &*ctx.memory, &turn.content, turn.speaker.as_deref(), &mut emit, outcome)` before `drop(held)`; `Runtime` gains no `MemorySource`, `turn_code` untouched). `cargo check -p loopd` clean, no error or warning. `wc -l` prints 186 for `compact.rs` and 284 for `channel.rs`, each ≤ 500. `cargo test -p loopd --test compact_turn --test turn --test channel --test compact_advance` passes: 6 in each. `git diff --stat` lists only `compact.rs` and `channel.rs`; `git status --short` shows `compact_turn.rs` as new. `cargo fmt --all` changed nothing after the edits. `make gate` printed `gate: ok` with 1508 tests (1502 before, plus the 6 of `compact_turn.rs`). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | | M5d1/07-loopd-compact-advance | 2026-10-03 | done | 2 | pass | none | Copied the given test `crates/loopd/tests/compact_advance.rs` (new). Step 2 (`cargo build --tests --keep-going -p loopd`) showed exactly the named failure: `compact_advance` with E0425 (`cannot find function commit in module compact`); every other target compiled. Made S1 (`session.rs`: `begin_epoch` after `log_path`, in that order — writes `.baseline.json` with `to_json`, `create_new` the `.jsonl`, accumulates `prior_users`/`prior_untrusted` from the old `records` before clearing them, sets `epoch`/`baseline`/`carry`/empty `records`/the new `File`, then `persist` `SessionStart` with the live `slot`; does not reset `next_call` or `slot`). Made C1 (`compact.rs`: extended the `use crate::` group to `Baseline`/`MemorySource`/`Registry` and the bxcore group to `Epoch`/`Timestamp`, then added `commit` after `summarise` — skips the append and takes `next`/summary from a trailing `EpochEnd`, else appends `EpochEnd` with `Epoch(epoch.0.saturating_add(1))`, reads `memory.block()` as `Failed`, `Baseline::assemble_with(cfg, registry, &block)` as `Failed`, then `begin_epoch`). `cargo check -p loopd` clean, no error or warning. `wc -l crates/loopd/src/session.rs` prints 438 and `crates/loopd/src/compact.rs` prints 134, each ≤ 500. `cargo test -p loopd --test compact_advance --test session_epochs --test compact_summary` passes: 6 in `compact_advance`, 9 in `session_epochs`, 6 in `compact_summary`. `git diff --stat` lists only `session.rs` and `compact.rs`; `git status --short` shows `compact_advance.rs` as new. `make gate` printed `gate: ok` with 1502 tests (1496 before, plus the 6 of `compact_advance.rs`). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | | M5d1/05-loopd-compact-due | 2026-10-03 | done | 1 | pass | none | Copied the given test `crates/loopd/tests/compact_due.rs` and the starter `config/compact.md` from `docs/plans/M5d1/files/`. Created `crates/loopd/src/compact.rs`: `due` does `cache_n.saturating_add(prompt_n) as f64 >= threshold * (n_ctx as f64)`; `fallback_summary` `filter_map`s `LogRecord::User { content, .. }` to `&str`, reverses, joins `"\n\n"`, then `cap_chars`; `cap_chars` is the given `chars().take(n)`. Added `pub mod compact;` after `pub mod channel;` in `lib.rs` (rustfmt order, accepted). `cargo check -p loopd` clean, no error or warning. `cargo test -p loopd --test compact_due` passes all 6. `git diff --stat` lists only `lib.rs` and `compact.rs`; `git status --short` shows `compact_due.rs` and `compact.md` as new. `make gate` printed `gate: ok` with 1490 tests (1484 before, plus the 6 of `compact_due.rs`). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b | | M5d1/04-memoryd-honcho-epochs | 2026-10-02 | done | 1 | pass | none | Copied the given test `crates/memoryd/tests/honcho_epochs.rs` (new). Step 2 (`cargo build --tests --keep-going -p memoryd`) showed exactly the named failure: `honcho_epochs` with one E0432 (`unresolved import memoryd::honcho::turns::read_session`); every other target compiled. Made H1 (`turns.rs`: `log_path` doc now names the sibling epoch files and that `read_session` concatenates them), H1b (inserted `epoch_logs` after `read_records`: lists `.jsonl` under `sessions//`, `n` parsed as `u32`, sorted by `n`, a missing directory empty; then `read_session` concatenates `read_records` over each), H2 (`resume` reads `read_session` instead of `read_log(&log_path(...))`), H3 (`records` matches `read_session(...).map_err(|e| e.to_string())`, keeping the logged line shape); W1 (`worker.rs`: added `use ...decide::{final_reply, turn_records, user_text}` and `use ...turns::{LogError, TurnSync, log_path, read_session}`), W2 (`send_turn` loads `read_session`, `self.readable(&path)` on Ok, `LogError::Damaged(_)` -> `Unsent::Local("log")`, other error -> `Unsent::Unreadable(e.to_string())`, then `turn_records`/`user_text`/`final_reply` from `decide`), W3 (`batch` called with `text`/`reply` by value, no borrow). `read_turn` no longer used in the worker body but stays re-exported. `cargo check -p memoryd` clean, no error or warning. `wc -l` prints 369 for `turns.rs` and 453 for `worker.rs`, each ≤ 500. `cargo test -p memoryd --test honcho_epochs --test honcho_turns --test honcho_worker` passes: 6 in `honcho_epochs`, 16 in `honcho_turns`, 12 in `honcho_worker`. `git diff --stat` lists only `turns.rs` and `worker.rs`; `cargo fmt --all` changed nothing after the edits. `make gate` printed `gate: ok` with 1484 tests (1478 before, plus the 6 of `honcho_epochs.rs`). No new dependency, so `Cargo.lock` unchanged. | ornith-1.5-35b-a3b |