diff --git a/crates/loopd/src/lib.rs b/crates/loopd/src/lib.rs index 52391fb..b8f7bc0 100644 --- a/crates/loopd/src/lib.rs +++ b/crates/loopd/src/lib.rs @@ -5,5 +5,6 @@ pub mod config; pub mod http; pub mod llama; pub mod selftest; +pub mod session; pub mod sse; pub mod tools; diff --git a/crates/loopd/src/session.rs b/crates/loopd/src/session.rs new file mode 100644 index 0000000..af5dfcb --- /dev/null +++ b/crates/loopd/src/session.rs @@ -0,0 +1,236 @@ +//! A session on disk: `sessions//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. + +use std::fs::{File, OpenOptions}; +use std::io::Write; +use std::path::{Path, PathBuf}; + +use crate::baseline::{Baseline, BaselineError}; +use proto::{CallId, Epoch, LogRecord, SessionId, Timestamp}; + +/// Creating, opening, appending and resuming a session, and the errors each can raise. +#[derive(Debug)] +pub enum SessionError { + Exists(SessionId), + NotFound(SessionId), + Io(PathBuf, std::io::Error), + Torn { + path: PathBuf, + line: usize, + why: String, + }, + Baseline(BaselineError), + Encode(serde_json::Error), +} + +impl std::fmt::Display for SessionError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SessionError::Exists(id) => write!(f, "session already exists: {}", id.as_str()), + SessionError::NotFound(id) => write!(f, "session not found: {}", id.as_str()), + SessionError::Io(path, err) => write!(f, "{}: {err}", path.display()), + SessionError::Torn { path, line, why } => write!(f, "{}:{line}: {why}", path.display()), + SessionError::Baseline(err) => write!(f, "{err}"), + SessionError::Encode(err) => write!(f, "{err}"), + } + } +} + +impl std::error::Error for SessionError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + SessionError::Io(_, err) => Some(err), + SessionError::Baseline(err) => Some(err), + SessionError::Encode(err) => Some(err), + _ => None, + } + } +} + +/// A live session: its id, the directory holding the baseline and log, the replayed records, the +/// appended log file, and the next call id. +#[derive(Debug)] +pub struct Session { + id: SessionId, + dir: PathBuf, + baseline: Baseline, + records: Vec, + log: File, + next_call: CallId, +} + +impl Session { + pub fn dir_for(home: &Path, id: &SessionId) -> PathBuf { + home.join("sessions").join(id.as_str()) + } + + pub fn create( + home: &Path, + id: SessionId, + baseline: Baseline, + slot: u32, + ) -> Result { + let dir = Session::dir_for(home, &id); + if dir.exists() { + return Err(SessionError::Exists(id)); + } + std::fs::create_dir_all(&dir).map_err(|e| SessionError::Io(dir.clone(), e))?; + + let baseline_path = dir.join("0.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 = dir.join("0.jsonl"); + let log = OpenOptions::new() + .create_new(true) + .append(true) + .open(&log_path) + .map_err(|e| SessionError::Io(log_path.clone(), e))?; + + let mut session = Session { + id, + dir, + baseline, + records: Vec::new(), + log, + next_call: CallId(1), + }; + session.persist(LogRecord::SessionStart { + time: Timestamp::now(), + session: session.id.clone(), + epoch: Epoch(0), + slot, + baseline: session.baseline.hash().map_err(SessionError::Baseline)?, + })?; + Ok(session) + } + + pub fn open(home: &Path, id: SessionId) -> Result { + let dir = Session::dir_for(home, &id); + let log_path = dir.join("0.jsonl"); + if !log_path.exists() { + return Err(SessionError::NotFound(id)); + } + let baseline_path = dir.join("0.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::(line) { + Ok(record) => records.push(record), + Err(why) => { + return Err(SessionError::Torn { + path: log_path.clone(), + line: line_no, + why: why.to_string(), + }); + } + } + } + } + + let mut highest = 0u64; + for record in &records { + if let LogRecord::ToolResult { call, .. } = record { + highest = highest.max(call.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), + }) + } + + pub fn id(&self) -> &SessionId { + &self.id + } + + pub fn dir(&self) -> &Path { + &self.dir + } + + pub fn baseline(&self) -> &Baseline { + &self.baseline + } + + pub fn records(&self) -> &[LogRecord] { + &self.records + } + + pub fn append(&mut self, record: LogRecord) -> Result<(), SessionError> { + self.persist(record) + } + + pub fn next_call(&mut self) -> CallId { + let id = self.next_call; + self.next_call = CallId(self.next_call.0 + 1); + id + } + + pub fn last_usage(&self) -> Option { + for record in self.records.iter().rev() { + if let LogRecord::Usage { + cache_n, + prompt_n, + predicted_n, + reasoning_tokens, + thinking_capped, + .. + } = record + { + return Some(proto::Usage { + cache_n: *cache_n, + prompt_n: *prompt_n, + predicted_n: *predicted_n, + reasoning_tokens: *reasoning_tokens, + thinking_capped: *thinking_capped, + }); + } + } + None + } + + /// Encode one record, write it and its newline, sync the data to disk, and only then keep it in + /// memory — so a record is never in memory without being durable. + fn persist(&mut self, record: LogRecord) -> Result<(), SessionError> { + let line = serde_json::to_string(&record).map_err(SessionError::Encode)?; + let path = self.log_path(); + self.log + .write_all(line.as_bytes()) + .map_err(|e| SessionError::Io(path.clone(), e))?; + self.log + .write_all(b"\n") + .map_err(|e| SessionError::Io(path.clone(), e))?; + self.log + .sync_data() + .map_err(|e| SessionError::Io(path.clone(), e))?; + self.records.push(record); + Ok(()) + } + + fn log_path(&self) -> PathBuf { + self.dir.join("0.jsonl") + } +} diff --git a/crates/loopd/tests/session.rs b/crates/loopd/tests/session.rs new file mode 100644 index 0000000..70ee75f --- /dev/null +++ b/crates/loopd/tests/session.rs @@ -0,0 +1,198 @@ +//! Tests for the session store. Do not edit. + +mod support; + +use loopd::baseline::Baseline; +use loopd::session::{Session, SessionError}; +use proto::{CallId, DataClass, Epoch, LogRecord, SessionId, Timestamp}; +use support::Home; + +fn baseline() -> Baseline { + Baseline { + system: "sys".to_string(), + tools: vec![], + } +} + +fn id(s: &str) -> SessionId { + SessionId::new(s).unwrap() +} + +fn ts() -> Timestamp { + Timestamp::parse("2026-09-18T08:00:00.000Z").unwrap() +} + +#[test] +fn create_writes_the_baseline_file_and_the_start_record() { + let home = Home::new(); + let s = Session::create(&home.dir, id("a"), baseline(), 3).unwrap(); + assert_eq!(s.dir(), home.dir.join("sessions").join("a")); + assert_eq!( + Baseline::from_json(&home.read("sessions/a/0.baseline.json")).unwrap(), + baseline() + ); + let records = home.records("a"); + assert_eq!(records.len(), 1); + match &records[0] { + LogRecord::SessionStart { + session, + epoch, + slot, + baseline: hash, + .. + } => { + assert_eq!(session, &id("a")); + assert_eq!(*epoch, Epoch(0)); + assert_eq!(*slot, 3); + assert_eq!(*hash, baseline().hash().unwrap()); + } + other => panic!("{other:?}"), + } + assert_eq!(s.records(), &records[..]); + assert!(matches!( + Session::create(&home.dir, id("a"), baseline(), 0), + Err(SessionError::Exists(_)) + )); +} + +#[test] +fn append_is_visible_on_disk_at_once_and_after_reopen() { + let home = Home::new(); + let mut s = Session::create(&home.dir, id("a"), baseline(), 0).unwrap(); + s.append(LogRecord::User { + time: ts(), + content: "hi\nthere \"quoted\" caf\u{e9}".to_string(), + }) + .unwrap(); + assert_eq!( + home.records("a").len(), + 2, + "written and synced before append returns" + ); + drop(s); + let s = Session::open(&home.dir, id("a")).unwrap(); + assert_eq!(s.records().len(), 2); + assert_eq!(s.baseline(), &baseline()); + assert!( + matches!(s.records()[1], LogRecord::User { ref content, .. } if content.contains("caf\u{e9}")) + ); +} + +#[test] +fn resume_uses_the_baseline_file_not_system_md() { + let home = Home::new(); + let s = Session::create(&home.dir, id("a"), baseline(), 0).unwrap(); + drop(s); + home.write("system.md", "A different prompt.\n"); + let s = Session::open(&home.dir, id("a")).unwrap(); + assert_eq!(s.baseline().system, "sys"); +} + +#[test] +fn opening_a_missing_session_is_not_found() { + let home = Home::new(); + assert!(matches!( + Session::open(&home.dir, id("nope")), + Err(SessionError::NotFound(_)) + )); +} + +#[test] +fn a_torn_log_is_refused_with_the_line_number() { + let home = Home::new(); + let mut s = Session::create(&home.dir, id("a"), baseline(), 0).unwrap(); + s.append(LogRecord::User { + time: ts(), + content: "hi".to_string(), + }) + .unwrap(); + drop(s); + let path = "sessions/a/0.jsonl"; + let good = home.read(path); + // A last line cut in the middle. + home.write(path, &good[..good.len() - 5]); + match Session::open(&home.dir, id("a")).map(|_| ()) { + Err(SessionError::Torn { line, .. }) => assert_eq!(line, 2), + other => panic!("{other:?}"), + } + // A complete line that is not a record. + home.write(path, &format!("{good}{{\"type\":\"zz\"}}\n")); + match Session::open(&home.dir, id("a")).map(|_| ()) { + Err(SessionError::Torn { line, path, .. }) => { + assert_eq!(line, 3); + assert!(path.ends_with("0.jsonl")); + } + other => panic!("{other:?}"), + } + let e: Box = + Box::new(Session::open(&home.dir, id("a")).map(|_| ()).unwrap_err()); + assert!(e.to_string().contains("0.jsonl:3"), "{e}"); +} + +#[test] +fn call_ids_continue_across_a_reopen() { + let home = Home::new(); + let mut s = Session::create(&home.dir, id("a"), baseline(), 0).unwrap(); + assert_eq!(s.next_call(), CallId(1)); + assert_eq!(s.next_call(), CallId(2)); + s.append(LogRecord::ToolResult { + time: ts(), + call: CallId(2), + tool_call_id: "x".to_string(), + content: String::new(), + class: DataClass::Public, + untrusted: false, + truncated: false, + }) + .unwrap(); + drop(s); + let mut s = Session::open(&home.dir, id("a")).unwrap(); + assert_eq!( + s.next_call(), + CallId(3), + "one more than the highest call in the log" + ); + let mut fresh = Session::create(&home.dir, id("b"), baseline(), 0).unwrap(); + assert_eq!(fresh.next_call(), CallId(1)); +} + +#[test] +fn last_usage_is_the_latest_usage_record() { + let home = Home::new(); + let mut s = Session::create(&home.dir, id("a"), baseline(), 0).unwrap(); + assert_eq!(s.last_usage(), None); + s.append(LogRecord::Usage { + time: ts(), + cache_n: 1, + prompt_n: 2, + predicted_n: 3, + reasoning_tokens: 0, + thinking_capped: false, + }) + .unwrap(); + s.append(LogRecord::User { + time: ts(), + content: "x".to_string(), + }) + .unwrap(); + s.append(LogRecord::Usage { + time: ts(), + cache_n: 6, + prompt_n: 7, + predicted_n: 8, + reasoning_tokens: 4, + thinking_capped: true, + }) + .unwrap(); + let u = s.last_usage().unwrap(); + assert_eq!( + ( + u.cache_n, + u.prompt_n, + u.predicted_n, + u.reasoning_tokens, + u.thinking_capped + ), + (6, 7, 8, 4, true) + ); +} diff --git a/docs/implementer-log.md b/docs/implementer-log.md index b16c864..5483d03 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -34,6 +34,7 @@ reviewer adds findings under "Reviews" once per milestone. | M2b/02-loopd-config | 2026-09-18 | done | 2 | fail | none | Added Paths/Channel/Loop/Baseline structs to config.rs with `#[serde(deny_unknown_fields, default)]` and Debug/Clone/PartialEq/Eq/Deserialize (Default derived for Channel, hand-written for the other three: home is $BOXMAKER_HOME else /var/lib/boxmaker, loop is 8/true/16384, baseline system is system.md); Config gained the four `#[serde(default)]` fields and `channel_socket()` fills the default `/run/loop/loop.sock` when the socket is empty. `load` joins `baseline.system` to the config file's directory via `parent.join` (which replaces an already-absolute path); `parse` leaves it. 9 config tests pass, `deny_unknown_fields` count is 10. Two clippy fixes on the first (failing) gate run: the nested if in `load` collapsed by relying on Path::join replacing absolute paths instead of a 1.98 let-chain, and `Path::is_empty` (stable 1.98) replaced with `as_os_str().is_empty()`. | ? | | M2b/03-loopd-tools | 2026-09-18 | done | 1 | pass | none | Added `pub mod tools;` to lib.rs and `serde::Serialize`/`serde::Deserialize`/`deny_unknown_fields` to ToolSchema; wrote crates/loopd/src/tools.rs with ToolPort, Entry, Registry (m2b/config), core_schemas/get/find, FIND_TOOL/CALL_TOOL constants, clock_schema/echo_schema, Dispatch with dispatch (find_tool/call_tool/local dispatch rows), cap_result via floor_char_boundary, and FakeTools recording calls and answering clock/echo/denying others with `unwrap_or_else(|p| p.into_inner())` on Mutex::lock. 7 tools tests pass, `make gate` prints `gate: ok`, no `unwrap()` in tools.rs. | ? | | M2b/04-loopd-baseline | 2026-09-18 | done | 2 | fail | none | Wrote crates/loopd/src/baseline.rs: `Baseline` (system prompt + core tool schemas, `deny_unknown_fields`), `BaselineError` (Read/Parse name the file, plus Hash) with Display/std::error::Error, `assemble` (system prompt trimmed of trailing whitespace, core memory appended with a blank line when its trimmed content is non-empty), `to_json`/`from_json`, `load`, and `hash` (sha256 of the canonical JSON). `messages` prepends the system message and replays every `LogRecord` variant explicitly named, so a new one is a compile error. The `\n\n` separator between system prompt and core memory had to be two newlines (a blank line), not one. All 6 baseline tests pass and all loopd tests pass with the new support module; first gate run failed on a rustfmt import-order diff, fixed with `cargo fmt`. | ? | +| M2b/05-loopd-session | 2026-09-18 | done | 2 | fail | none | Copied the given test byte-identical and wrote crates/loopd/src/session.rs: `Session` (id, dir, baseline, records, appended log file, next_call) and `SessionError` (Exists/NotFound/Io/Torn/Baseline/Encode) with derived Debug, Display and std::error::Error::source. `create` refuses an existing dir, writes `0.baseline.json`, opens `0.jsonl` with `create_new`+`append`, and appends a `SessionStart` (`Timestamp::now()`, epoch 0, the slot, `baseline.hash()`). `open` reads the baseline from the file (not `system.md`), requires every log line to end in `\n` and parse as a `LogRecord` else `Torn` with the 1-based line and reason, and sets `next_call` to one past the highest `ToolResult` call. `append` encodes, writes, `sync_data()`, then pushes to memory. All 7 session tests pass. First gate failed on clippy: split the `source()` arm that bound three different error types into three arms, removed the redundant `.write(true)` (implied by `append`), and used `path.display()` for the `Torn` path. | ? | ## Reviews