boxmaker/docs/plans/M5d1/02-loopd-session-epochs.md

M5d-1 task 02: Session::open loads the current epoch

Branch: m5d1. Check it: git branch --show-current prints m5d1 and git status --short is empty; otherwise stop. Never switch branch. Commit subject: loopd: open the highest epoch and carry its summary Log row: the Task cell is M5d1/02-loopd-session-epochs.

Goal

A session directory may have 0.jsonl and, after compaction, 1.baseline.json / 1.jsonl and so on (spec 3.1). Session::open loads the highest epoch that has both a baseline file and a log. For epoch n > 0 the previous log must end with EpochEnd { next: n, summary }; that summary is the carry. turns() and untrusted() span every epoch. next_call is one more than the highest ToolResult call in any epoch. append writes the current epoch's log.

records() stays the current epoch only: later tasks replay that file (plus the carry) as the request.

Every place that changes

This is the complete list.

Place	What happens there
crates/loopd/src/session.rs	NoCarry; fields epoch, slot, carry, prior_users, prior_untrusted; open; turns; untrusted; log_path; accessors; helpers
crates/loopd/tests/session_epochs.rs	new, given
everything else	no change: not session.rs's existing tests, not turn.rs, not Honcho

Files

• Copy, from docs/plans/M5d1/files/ (the same path under it): the new test crates/loopd/tests/session_epochs.rs.
• Modify: crates/loopd/src/session.rs, docs/implementer-log.md.
• Do not touch: any other file.

Facts you need

• Epoch is pub struct Epoch(pub u32) in bxcore. Format a file name with format!("{}.jsonl", epoch.0) and format!("{}.baseline.json", epoch.0).
• A complete epoch is both <n>.baseline.json and <n>.jsonl in the session directory. A baseline without a log, or a log without a baseline, does not count. The highest complete n is the current epoch. If none is complete, open is NotFound, as today when 0.jsonl is missing.
• Carry. For current epoch n > 0, read (n-1).jsonl. Among its records, the last EpochEnd whose next is Epoch(n) gives summary. If that file is missing, or has no such record, SessionError::NoCarry { path } naming (n-1).jsonl. Epoch 0 has carry: None.
• records on the live Session is only the current epoch's log, so messages in task 03 can replay it. prior_users / prior_untrusted are counted from every earlier complete epoch while open walks them. turns() is prior_users plus the User records in records. untrusted() is prior_untrusted or any untrusted ToolResult in records.
• next_call is one more than the highest ToolResult.call.0 in every complete epoch (current included). A call id of 4 in epoch 0 and 2 in epoch 1 continues at 5.
• slot is the slot on the current epoch's SessionStart. create stores the slot argument. If the current log has no SessionStart, use 0.
• append / persist already write self.log_path(). Change log_path to the current epoch's file; create still starts at 0.jsonl.
• A last EpochEnd on epoch 0 with no epoch-1 files means compact stopped halfway: open stays on epoch 0, carry is None, and the EpochEnd is in records. Task 07 resumes that.
• The torn-log checks stay as they are, per file, with the line number of that file.
• session.rs must stay at or under 500 lines (wc -l). After these edits it is about 430.
• The items the edits use. This table is complete.

  Item	What it does
  std::fs::read_dir	list <n>.jsonl names
  str::strip_suffix(".jsonl"), str::parse::<u32>	the epoch number
  Path::exists	the matching baseline
  u64::try_from(count).unwrap_or(u64::MAX)	turns, as today
  CallId(highest + 1)	next_call, as today, over every epoch
  OpenOptions::new().append(true).open	the current log

The edits, exactly

Nothing else in this file changes. Each quoted "before" block occurs exactly once. Keep struct fields in the order shown.

crates/loopd/src/session.rs

S1. The module doc comment, the lines

//! 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.

become

//! 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.

S2. After Encode(serde_json::Error), in SessionError, add a variant (the lines)

    Encode(serde_json::Error),
}

become

    Encode(serde_json::Error),
    NoCarry {
        path: PathBuf,
    },
}

S3. In Display, the lines

            SessionError::Encode(err) => write!(f, "{err}"),
        }

become

            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()
            ),
        }

S4. The struct, the lines

pub struct Session {
    id: SessionId,
    dir: PathBuf,
    baseline: Baseline,
    records: Vec<LogRecord>,
    log: File,
    next_call: CallId,
}

become

pub struct Session {
    id: SessionId,
    dir: PathBuf,
    baseline: Baseline,
    records: Vec<LogRecord>,
    log: File,
    next_call: CallId,
    epoch: Epoch,
    slot: u32,
    carry: Option<String>,
    prior_users: u64,
    prior_untrusted: bool,
}

S5. In create, the lines

        let mut session = Session {
            id,
            dir,
            baseline,
            records: Vec::new(),
            log,
            next_call: CallId(1),
        };

become

        let mut session = Session {
            id,
            dir,
            baseline,
            records: Vec::new(),
            log,
            next_call: CallId(1),
            epoch: Epoch(0),
            slot,
            carry: None,
            prior_users: 0,
            prior_untrusted: false,
        };

S6. Replace the whole open function. The lines from pub fn open(home: &Path, id: SessionId) through the Ok(Session { block that ends next_call: CallId(highest + 1), and }) } become the function below. Fill any todo!() from its comment, then delete the comment. Each comment is at most six lines.

    pub fn open(home: &Path, id: SessionId) -> Result<Session, SessionError> {
        let dir = Session::dir_for(home, &id);
        let epochs = complete_epochs(&dir)?;
        let Some(&n) = epochs.last() else {
            return Err(SessionError::NotFound(id));
        };
        let epoch = Epoch(n);
        let baseline_path = dir.join(format!("{n}.baseline.json"));
        let baseline = Baseline::load(&baseline_path).map_err(SessionError::Baseline)?;
        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 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,
            baseline,
            records,
            log,
            next_call: CallId(highest + 1),
            epoch,
            slot,
            carry,
            prior_users,
            prior_untrusted,
        })
    }

S7. After pub fn records(&self), add accessors (the lines)

    pub fn records(&self) -> &[LogRecord] {
        &self.records
    }

become

    pub fn records(&self) -> &[LogRecord] {
        &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()
    }

S8. untrusted and turns, the lines

    pub fn untrusted(&self) -> bool {
        self.records.iter().any(|record| {
            matches!(
                record,
                LogRecord::ToolResult {
                    untrusted: true,
                    ..
                }
            )
        })
    }

    /// 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.
    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)
    }

become

    pub fn untrusted(&self) -> bool {
        self.prior_untrusted || any_untrusted(&self.records)
    }

    /// 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 {
        self.prior_users
            .saturating_add(count_users(&self.records))
    }

S9. log_path, the lines

    fn log_path(&self) -> PathBuf {
        self.dir.join("0.jsonl")
    }
}

become

    fn log_path(&self) -> PathBuf {
        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)
}

The open body in S6 is given whole: copy it, do not leave a todo!() in open. The helpers in S9 are given whole too.

Steps

• 1. Copy the given test. From the repository root:

  cp docs/plans/M5d1/files/crates/loopd/tests/session_epochs.rs crates/loopd/tests/
  

• 2. See it fail. cargo build --tests --keep-going -p loopd, timeout 600000 ms: test target session_epochs does not compile (E0599 no method epoch / carry / slot, and/or E0433 NoCarry). Every other target compiles. Existing session tests still compile.

• 3. The edits. Make S1 to S9 in order. cargo check -p loopd prints no error and no warning. wc -l crates/loopd/src/session.rs prints a number ≤ 500.

• 4. The tests. cargo test -p loopd --test session_epochs --test session: every test passes, 9 in session_epochs and the others as before.

• 5. Check. git diff --stat lists exactly crates/loopd/src/session.rs.

• 6. Run the gate. cargo fmt --all, then make gate, timeout 600000 ms. Last line gate: ok. Test count 1473 (1464 before, plus 9).

• 7. Log and commit. Task cell M5d1/02-loopd-session-epochs. Stage with git add crates/loopd docs/implementer-log.md Cargo.lock, then commit as AGENTS.md says.

Done when

• make gate prints gate: ok with 1473 tests.
• git diff HEAD~1 --stat lists only crates/loopd/src/session.rs, crates/loopd/tests/session_epochs.rs and docs/implementer-log.md.

Stop and report if

• The branch check fails.
• A "before" block is not found exactly once.
• Step 2 shows a different result.
• cargo check warns or errors, or session.rs exceeds 500 lines.
• A test fails in step 4. Do not change the test.
• An existing session test fails: stop and report. Those tests still create and open epoch 0 only.
• The gate count is not 1473.
• A comment does not fit the code under it.

What the given tests pin

Opening a directory with epochs 0 and 1 loads 1 and the carry from 0's EpochEnd; a missing EpochEnd on epoch 0 when epoch 1 files exist is NoCarry naming 0.jsonl with the damaged-log runbook pointer; epoch 0 has no carry and keeps its SessionStart slot; two Users in epoch 0 and one in epoch 1 is turns() == 3 while records() has one User; an untrusted ToolResult in epoch 0 taints after compact; next_call is one more than the highest call in any epoch; append after opening epoch 1 writes 1.jsonl; a 1.baseline.json without 1.jsonl is ignored; an EpochEnd on epoch 0 without the next files stays on epoch 0.