loopd: open the highest epoch and carry its summary

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
2026-10-02 23:38:10 -07:00
parent 518d6ec416
commit 18039fa2e4
3 changed files with 453 additions and 60 deletions
@@ -1,5 +1,6 @@
-//! A session on disk: `sessions/<id>/0.baseline.json` and `0.jsonl`. The log is the state —
-//! create, open, append and resume all live here, and nothing is durable until `append` syncs it.
+//! A session on disk: `sessions/<id>/<epoch>.baseline.json` and `<epoch>.jsonl`. The log is the
+//! state — create, open, append and resume all live here, and nothing is durable until `append`
+//! syncs it. Epoch 0 is the first; later epochs appear after compaction.

 use std::fs::{File, OpenOptions};
 use std::io::Write;
@@ -21,6 +22,9 @@ pub enum SessionError {
    },
    Baseline(BaselineError),
    Encode(serde_json::Error),
+    NoCarry {
+        path: PathBuf,
+    },
 }

 impl std::fmt::Display for SessionError {
@@ -36,6 +40,11 @@ impl std::fmt::Display for SessionError {
            ),
            SessionError::Baseline(err) => write!(f, "{err}"),
            SessionError::Encode(err) => write!(f, "{err}"),
+            SessionError::NoCarry { path } => write!(
+                f,
+                "{}: compacted epoch is missing its EpochEnd; see docs/runbook.md#session-log-damaged",
+                path.display()
+            ),
        }
    }
 }
@@ -61,6 +70,11 @@ pub struct Session {
    records: Vec<LogRecord>,
    log: File,
    next_call: CallId,
+    epoch: Epoch,
+    slot: u32,
+    carry: Option<String>,
+    prior_users: u64,
+    prior_untrusted: bool,
 }

 impl Session {
@@ -99,6 +113,11 @@ impl Session {
            records: Vec::new(),
            log,
            next_call: CallId(1),
+            epoch: Epoch(0),
+            slot,
+            carry: None,
+            prior_users: 0,
+            prior_untrusted: false,
        };
        session.persist(LogRecord::SessionStart {
            time: Timestamp::now(),
@@ -112,51 +131,26 @@ impl Session {

    pub fn open(home: &Path, id: SessionId) -> Result<Session, SessionError> {
        let dir = Session::dir_for(home, &id);
-        let log_path = dir.join("0.jsonl");
-        if !log_path.exists() {
+        let epochs = complete_epochs(&dir)?;
+        let Some(&n) = epochs.last() else {
            return Err(SessionError::NotFound(id));
-        }
-        let baseline_path = dir.join("0.baseline.json");
+        };
+        let epoch = Epoch(n);
+        let baseline_path = dir.join(format!("{n}.baseline.json"));
        let baseline = Baseline::load(&baseline_path).map_err(SessionError::Baseline)?;
-
-        let text = std::fs::read_to_string(&log_path)
-            .map_err(|e| SessionError::Io(log_path.clone(), e))?;
-        let mut records = Vec::new();
-        if !text.is_empty() {
-            for (idx, line) in text.split_inclusive('\n').enumerate() {
-                let line_no = idx + 1;
-                if !line.ends_with('\n') {
-                    return Err(SessionError::Torn {
-                        path: log_path.clone(),
-                        line: line_no,
-                        why: "last line does not end in a newline".to_string(),
-                    });
-                }
-                match serde_json::from_str::<LogRecord>(line) {
-                    Ok(record) => records.push(record),
-                    Err(why) => {
-                        return Err(SessionError::Torn {
-                            path: log_path.clone(),
-                            line: line_no,
-                            why: why.to_string(),
-                        });
-                    }
-                }
-            }
+        let log_path = dir.join(format!("{n}.jsonl"));
+        let records = load_records(&log_path)?;
+        let (carry, prior_users, prior_untrusted) = priors(&dir, n)?;
+        let mut highest = max_call(&records);
+        for earlier in &epochs[..epochs.len() - 1] {
+            let path = dir.join(format!("{earlier}.jsonl"));
+            highest = highest.max(max_call(&load_records(&path)?));
        }
-
-        let mut highest = 0u64;
-        for record in &records {
-            if let LogRecord::ToolResult { call, .. } = record {
-                highest = highest.max(call.0);
-            }
-        }
-
+        let slot = live_slot(&records, 0);
        let log = OpenOptions::new()
            .append(true)
            .open(&log_path)
            .map_err(|e| SessionError::Io(log_path.clone(), e))?;
-
        Ok(Session {
            id,
            dir,
@@ -164,6 +158,11 @@ impl Session {
            records,
            log,
            next_call: CallId(highest + 1),
+            epoch,
+            slot,
+            carry,
+            prior_users,
+            prior_untrusted,
        })
    }

@@ -183,6 +182,18 @@ impl Session {
        &self.records
    }

+    pub fn epoch(&self) -> Epoch {
+        self.epoch
+    }
+
+    pub fn slot(&self) -> u32 {
+        self.slot
+    }
+
+    pub fn carry(&self) -> Option<&str> {
+        self.carry.as_deref()
+    }
+
    pub fn append(&mut self, record: LogRecord) -> Result<(), SessionError> {
        self.persist(record)
    }
@@ -197,28 +208,13 @@ impl Session {
    /// grant, or a recalled memory-tainted note, which `brokerd` labels so. The labels are in
    /// the log, so this holds after the session is reopened, and it never becomes false again.
    pub fn untrusted(&self) -> bool {
-        self.records.iter().any(|record| {
-            matches!(
-                record,
-                LogRecord::ToolResult {
-                    untrusted: true,
-                    ..
-                }
-            )
-        })
+        self.prior_untrusted || any_untrusted(&self.records)
    }

-    /// How many turns the session has had: its `User` records, one per turn that reached the
-    /// log, a failed turn's too. Counted from `records`, which `open` reads back whole, so the
-    /// count goes on after a reopen or a restart; right after a turn it is that turn's number,
-    /// the first being 1. Epoch 0 is a session's only log until compaction is built.
+    /// How many turns the session has had: its `User` records across every epoch, a failed
+    /// turn's too. The first `User` of the session is turn 1 for its life.
    pub fn turns(&self) -> u64 {
-        let users = self
-            .records
-            .iter()
-            .filter(|record| matches!(record, LogRecord::User { .. }))
-            .count();
-        u64::try_from(users).unwrap_or(u64::MAX)
+        self.prior_users.saturating_add(count_users(&self.records))
    }

    pub fn last_usage(&self) -> Option<bxcore::Usage> {
@@ -263,6 +259,143 @@ impl Session {
    }

    fn log_path(&self) -> PathBuf {
-        self.dir.join("0.jsonl")
+        self.dir.join(format!("{}.jsonl", self.epoch.0))
    }
 }
+
+/// Epoch numbers that have both `<n>.jsonl` and `<n>.baseline.json`, sorted.
+fn complete_epochs(dir: &Path) -> Result<Vec<u32>, SessionError> {
+    let entries = match std::fs::read_dir(dir) {
+        Ok(e) => e,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(e) => return Err(SessionError::Io(dir.to_path_buf(), e)),
+    };
+    let mut out = Vec::new();
+    for entry in entries {
+        let entry = entry.map_err(|e| SessionError::Io(dir.to_path_buf(), e))?;
+        let name = entry.file_name();
+        let Some(name) = name.to_str() else { continue };
+        let Some(stem) = name.strip_suffix(".jsonl") else {
+            continue;
+        };
+        let Ok(n) = stem.parse::<u32>() else { continue };
+        if dir.join(format!("{n}.baseline.json")).exists() {
+            out.push(n);
+        }
+    }
+    out.sort_unstable();
+    Ok(out)
+}
+
+fn load_records(path: &Path) -> Result<Vec<LogRecord>, SessionError> {
+    let text = match std::fs::read_to_string(path) {
+        Ok(t) => t,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+            return Err(SessionError::Io(path.to_path_buf(), e));
+        }
+        Err(e) => return Err(SessionError::Io(path.to_path_buf(), e)),
+    };
+    let mut records = Vec::new();
+    if text.is_empty() {
+        return Ok(records);
+    }
+    for (idx, line) in text.split_inclusive('\n').enumerate() {
+        let line_no = idx + 1;
+        if !line.ends_with('\n') {
+            return Err(SessionError::Torn {
+                path: path.to_path_buf(),
+                line: line_no,
+                why: "last line does not end in a newline".to_string(),
+            });
+        }
+        match serde_json::from_str::<LogRecord>(line) {
+            Ok(record) => records.push(record),
+            Err(why) => {
+                return Err(SessionError::Torn {
+                    path: path.to_path_buf(),
+                    line: line_no,
+                    why: why.to_string(),
+                });
+            }
+        }
+    }
+    Ok(records)
+}
+
+/// Carry from epoch `n-1`, and user / untrusted counts from every epoch before `n`.
+fn priors(dir: &Path, n: u32) -> Result<(Option<String>, u64, bool), SessionError> {
+    let mut prior_users = 0u64;
+    let mut prior_untrusted = false;
+    let mut carry = None;
+    if n == 0 {
+        return Ok((None, 0, false));
+    }
+    for earlier in 0..n {
+        let path = dir.join(format!("{earlier}.jsonl"));
+        let records = match load_records(&path) {
+            Ok(r) => r,
+            Err(SessionError::Io(path, e)) if e.kind() == std::io::ErrorKind::NotFound => {
+                return Err(SessionError::NoCarry { path });
+            }
+            Err(e) => return Err(e),
+        };
+        prior_users = prior_users.saturating_add(count_users(&records));
+        prior_untrusted = prior_untrusted || any_untrusted(&records);
+        if earlier + 1 == n {
+            carry = carry_from(&records, Epoch(n)).map(str::to_string);
+            if carry.is_none() {
+                return Err(SessionError::NoCarry { path });
+            }
+        }
+    }
+    Ok((carry, prior_users, prior_untrusted))
+}
+
+fn carry_from(records: &[LogRecord], next: Epoch) -> Option<&str> {
+    records.iter().rev().find_map(|record| match record {
+        LogRecord::EpochEnd {
+            next: n, summary, ..
+        } if *n == next => Some(summary.as_str()),
+        _ => None,
+    })
+}
+
+fn count_users(records: &[LogRecord]) -> u64 {
+    let users = records
+        .iter()
+        .filter(|record| matches!(record, LogRecord::User { .. }))
+        .count();
+    u64::try_from(users).unwrap_or(u64::MAX)
+}
+
+fn any_untrusted(records: &[LogRecord]) -> bool {
+    records.iter().any(|record| {
+        matches!(
+            record,
+            LogRecord::ToolResult {
+                untrusted: true,
+                ..
+            }
+        )
+    })
+}
+
+fn max_call(records: &[LogRecord]) -> u64 {
+    let mut highest = 0u64;
+    for record in records {
+        if let LogRecord::ToolResult { call, .. } = record {
+            highest = highest.max(call.0);
+        }
+    }
+    highest
+}
+
+fn live_slot(records: &[LogRecord], fallback: u32) -> u32 {
+    records
+        .iter()
+        .find_map(|record| match record {
+            LogRecord::SessionStart { slot, .. } => Some(*slot),
+            _ => None,
+        })
+        .unwrap_or(fallback)
+}