loopd: write the next epoch from a compaction summary

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
2026-10-03 00:25:04 -07:00
parent c8b0a01c41
commit fe8e3e9354
4 changed files with 284 additions and 2 deletions
@@ -2,12 +2,14 @@
 //! (M5d spec, section 3). Task 07 writes the next files; task 08 calls this after a turn.

 use bxcore::sanitize::neutralise;
-use bxcore::{LogRecord, Usage};
+use bxcore::{Epoch, LogRecord, Timestamp, Usage};

-use crate::baseline::messages_with_carry;
+use crate::baseline::{Baseline, messages_with_carry};
 use crate::config::Config;
 use crate::llama::{ChatMessage, ChatRequest, Client};
+use crate::memory::MemorySource;
 use crate::session::Session;
+use crate::tools::Registry;
 use crate::turn::is_context_full;

 /// Why a summary call or a next-epoch write failed. Every variant's `Display` ends with the
@@ -90,3 +92,44 @@ pub fn summarise(cfg: &Config, client: &Client, session: &Session) -> Result<Str
        }),
    }
 }
+
+/// Append `EpochEnd` (unless it is already last), rebuild the baseline from `memory`, and
+/// switch `session` onto the next epoch. When the current log already ends with `EpochEnd`,
+/// that record's summary is the carry and the argument is ignored.
+pub fn commit(
+    session: &mut Session,
+    cfg: &Config,
+    registry: &Registry,
+    memory: &dyn MemorySource,
+    summary: String,
+) -> Result<(), CompactError> {
+    let (next, summary) = match session.records().last() {
+        Some(LogRecord::EpochEnd { next, summary, .. }) => (*next, summary.clone()),
+        _ => {
+            let next = Epoch(session.epoch().0.saturating_add(1));
+            session
+                .append(LogRecord::EpochEnd {
+                    time: Timestamp::now(),
+                    next,
+                    summary: summary.clone(),
+                })
+                .map_err(|e| CompactError::Failed {
+                    detail: e.to_string(),
+                })?;
+            (next, summary)
+        }
+    };
+    let block = memory
+        .block()
+        .map_err(|e| CompactError::Failed { detail: e })?;
+    let baseline =
+        Baseline::assemble_with(cfg, registry, &block).map_err(|e| CompactError::Failed {
+            detail: e.to_string(),
+        })?;
+    session
+        .begin_epoch(next, baseline, summary)
+        .map_err(|e| CompactError::Failed {
+            detail: e.to_string(),
+        })?;
+    Ok(())
+}
@@ -261,6 +261,43 @@ impl Session {
    fn log_path(&self) -> PathBuf {
        self.dir.join(format!("{}.jsonl", self.epoch.0))
    }
+
+    /// Write `<n>.baseline.json` and `<n>.jsonl` with `SessionStart`, and switch this session
+    /// onto that epoch. The live slot and `next_call` stay. Callers pass the carry that
+    /// `messages_with_carry` will insert.
+    pub(crate) fn begin_epoch(
+        &mut self,
+        next: Epoch,
+        baseline: Baseline,
+        carry: String,
+    ) -> Result<(), SessionError> {
+        let n = next.0;
+        let baseline_path = self.dir.join(format!("{n}.baseline.json"));
+        let json = baseline.to_json().map_err(SessionError::Encode)?;
+        std::fs::write(&baseline_path, json)
+            .map_err(|e| SessionError::Io(baseline_path.clone(), e))?;
+        let log_path = self.dir.join(format!("{n}.jsonl"));
+        let log = OpenOptions::new()
+            .create_new(true)
+            .append(true)
+            .open(&log_path)
+            .map_err(|e| SessionError::Io(log_path.clone(), e))?;
+        self.prior_users = self.prior_users.saturating_add(count_users(&self.records));
+        self.prior_untrusted = self.prior_untrusted || any_untrusted(&self.records);
+        self.epoch = next;
+        self.baseline = baseline;
+        self.carry = Some(carry);
+        self.records = Vec::new();
+        self.log = log;
+        let hash = self.baseline.hash().map_err(SessionError::Baseline)?;
+        self.persist(LogRecord::SessionStart {
+            time: Timestamp::now(),
+            session: self.id.clone(),
+            epoch: next,
+            slot: self.slot,
+            baseline: hash,
+        })
+    }
 }

 /// Epoch numbers that have both `<n>.jsonl` and `<n>.baseline.json`, sorted.
@@ -0,0 +1,201 @@
+//! Tests for writing the next epoch. Do not edit.
+
+mod support;
+
+use bxcore::{Epoch, LogRecord, SessionId};
+use loopd::baseline::Baseline;
+use loopd::compact;
+use loopd::memory::{MemorySource, NoMemory};
+use loopd::session::Session;
+use loopd::tools::Registry;
+use support::Home;
+
+struct Block(&'static str);
+
+impl MemorySource for Block {
+    fn block(&self) -> Result<String, String> {
+        Ok(self.0.to_string())
+    }
+    fn updated(&self, _session: &SessionId, _turn: u64) -> Result<(), String> {
+        Ok(())
+    }
+}
+
+fn id(s: &str) -> SessionId {
+    SessionId::new(s).unwrap()
+}
+
+fn session(home: &Home) -> (loopd::config::Config, Session) {
+    let cfg = home.config(std::path::Path::new("/tmp/unused.sock"));
+    let baseline = Baseline::assemble(&cfg, &Registry::m2b()).unwrap();
+    let s = Session::create(&home.dir, id("a"), baseline, cfg.slots.main).unwrap();
+    (cfg, s)
+}
+
+#[test]
+fn commit_writes_epoch_end_a_new_baseline_and_a_new_log() {
+    let home = Home::new();
+    let (cfg, mut s) = session(&home);
+    s.append(LogRecord::User {
+        time: bxcore::Timestamp::parse("2026-09-18T08:00:00.000Z").unwrap(),
+        content: "hi".to_string(),
+        speaker: None,
+    })
+    .unwrap();
+    compact::commit(
+        &mut s,
+        &cfg,
+        &Registry::m2b(),
+        &NoMemory,
+        "the sum".to_string(),
+    )
+    .unwrap();
+
+    assert_eq!(s.epoch(), Epoch(1));
+    assert_eq!(s.carry(), Some("the sum"));
+    assert_eq!(s.slot(), cfg.slots.main);
+    assert_eq!(s.turns(), 1, "the old user still counts");
+    assert_eq!(
+        s.records()
+            .iter()
+            .filter(|r| matches!(r, LogRecord::User { .. }))
+            .count(),
+        0
+    );
+
+    let old = home.records("a");
+    assert!(
+        matches!(old.last(), Some(LogRecord::EpochEnd { next, summary, .. }) if *next == Epoch(1) && summary == "the sum")
+    );
+    assert!(home.dir.join("sessions/a/1.baseline.json").is_file());
+    let new_log = home.read("sessions/a/1.jsonl");
+    assert!(new_log.contains("\"epoch\":1"), "{new_log}");
+}
+
+#[test]
+fn the_new_baseline_takes_a_fresh_memory_block() {
+    let home = Home::new();
+    let (cfg, mut s) = session(&home);
+    compact::commit(
+        &mut s,
+        &cfg,
+        &Registry::m2b(),
+        &Block("I take my tea strong."),
+        "sum".to_string(),
+    )
+    .unwrap();
+    assert!(
+        s.baseline().system.contains("I take my tea strong."),
+        "{}",
+        s.baseline().system
+    );
+    let on_disk = home.read("sessions/a/1.baseline.json");
+    assert!(on_disk.contains("I take my tea strong."), "{on_disk}");
+}
+
+#[test]
+fn reopen_loads_the_new_epoch() {
+    let home = Home::new();
+    let (cfg, mut s) = session(&home);
+    compact::commit(
+        &mut s,
+        &cfg,
+        &Registry::m2b(),
+        &NoMemory,
+        "carry".to_string(),
+    )
+    .unwrap();
+    drop(s);
+    let s = Session::open(&home.dir, id("a")).unwrap();
+    assert_eq!(s.epoch(), Epoch(1));
+    assert_eq!(s.carry(), Some("carry"));
+}
+
+#[test]
+fn a_second_commit_does_not_write_another_epoch_end_on_the_old_log() {
+    let home = Home::new();
+    let (cfg, mut s) = session(&home);
+    compact::commit(
+        &mut s,
+        &cfg,
+        &Registry::m2b(),
+        &NoMemory,
+        "first".to_string(),
+    )
+    .unwrap();
+    let ends = home
+        .records("a")
+        .iter()
+        .filter(|r| matches!(r, LogRecord::EpochEnd { .. }))
+        .count();
+    assert_eq!(ends, 1);
+    compact::commit(
+        &mut s,
+        &cfg,
+        &Registry::m2b(),
+        &NoMemory,
+        "second".to_string(),
+    )
+    .unwrap();
+    assert_eq!(s.epoch(), Epoch(2));
+    let ends = home
+        .records("a")
+        .iter()
+        .filter(|r| matches!(r, LogRecord::EpochEnd { .. }))
+        .count();
+    assert_eq!(ends, 1, "epoch 0 still has one EpochEnd");
+    let mid = home.read("sessions/a/1.jsonl");
+    assert!(mid.contains("\"type\":\"epoch_end\""), "{mid}");
+}
+
+#[test]
+fn resume_after_epoch_end_without_next_files_does_not_write_a_second_end() {
+    let home = Home::new();
+    let (cfg, mut s) = session(&home);
+    s.append(LogRecord::EpochEnd {
+        time: bxcore::Timestamp::now(),
+        next: Epoch(1),
+        summary: "already".to_string(),
+    })
+    .unwrap();
+    drop(s);
+    let mut s = Session::open(&home.dir, id("a")).unwrap();
+    assert_eq!(s.epoch(), Epoch(0));
+    compact::commit(
+        &mut s,
+        &cfg,
+        &Registry::m2b(),
+        &NoMemory,
+        "ignored".to_string(),
+    )
+    .unwrap();
+    let ends = home
+        .records("a")
+        .iter()
+        .filter(|r| matches!(r, LogRecord::EpochEnd { .. }))
+        .count();
+    assert_eq!(ends, 1);
+    assert_eq!(s.epoch(), Epoch(1));
+    assert_eq!(
+        s.carry(),
+        Some("already"),
+        "the summary already in EpochEnd is the carry"
+    );
+}
+
+#[test]
+fn the_new_session_start_keeps_the_live_slot() {
+    let home = Home::new();
+    let cfg = home.config(std::path::Path::new("/tmp/unused.sock"));
+    let baseline = Baseline::assemble(&cfg, &Registry::m2b()).unwrap();
+    let mut s = Session::create(&home.dir, id("a"), baseline, 7).unwrap();
+    compact::commit(&mut s, &cfg, &Registry::m2b(), &NoMemory, "s".to_string()).unwrap();
+    assert_eq!(s.slot(), 7);
+    match &s.records()[0] {
+        LogRecord::SessionStart { slot, epoch, .. } => {
+            assert_eq!(*slot, 7);
+            assert_eq!(*epoch, Epoch(1));
+        }
+        other => panic!("{other:?}"),
+    }
+}