loopd: compact after a full turn and retry SessionFull once

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>

crates/loopd/src/channel.rs

@@ -178,25 +178,38 @@ pub fn handle(stream: UnixStream, ctx: Arc<Context>) {
         }
     };
 
+    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

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<TurnOutcome, TurnError>,
+) -> Result<TurnOutcome, TurnError> {
+    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()
+    );
+}

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<loopd::turn::TurnOutcome, TurnError>,
+    ) -> Result<loopd::turn::TurnOutcome, TurnError> {
+        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: &registry,
+    };
+    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);
+}

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 `<n>.baseline.json` with `to_json`, `create_new` the `<n>.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 `<n>.jsonl` under `sessions/<id>/`, `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 |