Add the session store: baseline file and append-only log
Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
This commit is contained in:
@@ -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;
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
//! 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.
|
||||
|
||||
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<LogRecord>,
|
||||
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<Session, SessionError> {
|
||||
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<Session, SessionError> {
|
||||
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::<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 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<proto::Usage> {
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -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<dyn std::error::Error> =
|
||||
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)
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user