From 0cadad7333877c5c81feafe7a4be4e416ac96bb9 Mon Sep 17 00:00:00 2001 From: Revantark Date: Mon, 24 Aug 2026 12:34:53 +0530 Subject: [PATCH 1/4] add store and session manager --- crates/agent/Cargo.toml | 2 +- crates/agent/src/lib.rs | 4 +- crates/agent/src/session/dir.rs | 112 +++ crates/agent/src/session/error.rs | 26 + crates/agent/src/session/manager.rs | 707 ++++++++++++++++++ crates/agent/src/session/mod.rs | 9 + .../src/{session.rs => session/record.rs} | 111 ++- crates/agent/src/session/store.rs | 440 +++++++++++ 8 files changed, 1393 insertions(+), 18 deletions(-) create mode 100644 crates/agent/src/session/dir.rs create mode 100644 crates/agent/src/session/error.rs create mode 100644 crates/agent/src/session/manager.rs create mode 100644 crates/agent/src/session/mod.rs rename crates/agent/src/{session.rs => session/record.rs} (59%) create mode 100644 crates/agent/src/session/store.rs diff --git a/crates/agent/Cargo.toml b/crates/agent/Cargo.toml index 51a562c..dcc4d62 100644 --- a/crates/agent/Cargo.toml +++ b/crates/agent/Cargo.toml @@ -9,6 +9,7 @@ futures-util = { workspace = true } llm = { workspace = true } providers = { path = "../providers" } serde = { workspace = true } +serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } tools = { path = "../tools" } @@ -16,4 +17,3 @@ uuid = { workspace = true } [dev-dependencies] futures-util = { workspace = true } -serde_json = { workspace = true } diff --git a/crates/agent/src/lib.rs b/crates/agent/src/lib.rs index 3a69581..0510566 100644 --- a/crates/agent/src/lib.rs +++ b/crates/agent/src/lib.rs @@ -8,6 +8,8 @@ mod tool; pub use agent::{Agent, AgentBuilder, AgentEvent, AgentStream}; pub use context::AgentMessage; pub use error::AgentError; -pub use session::{SESSION_SCHEMA_VERSION, Session, SessionRecord}; +pub use session::{ + SESSION_SCHEMA_VERSION, Session, SessionError, SessionManager, SessionRecord, SessionSummary, +}; pub use skill::{Skill, build_system_prompt, format_skills_xml}; pub use tool::{AgentTool, default_tools}; diff --git a/crates/agent/src/session/dir.rs b/crates/agent/src/session/dir.rs new file mode 100644 index 0000000..7636223 --- /dev/null +++ b/crates/agent/src/session/dir.rs @@ -0,0 +1,112 @@ +use std::path::{Component, Path, PathBuf}; + +pub(crate) fn fnv1a64(data: &[u8]) -> u64 { + const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325; + const PRIME: u64 = 0x0000_0100_0000_01b3; + + let mut hash = OFFSET_BASIS; + for byte in data { + hash ^= u64::from(*byte); + hash = hash.wrapping_mul(PRIME); + } + hash +} + +pub(crate) fn pwd_key(pwd: &Path) -> String { + format!("{:016x}", fnv1a64(pwd.to_string_lossy().as_bytes())) +} + +/// Normalize a pwd to an absolute path. +pub(crate) fn normalize_pwd(pwd: PathBuf) -> Result { + if let Ok(canonical) = pwd.canonicalize() { + return Ok(canonical); + } + + let absolute = if pwd.is_absolute() { + pwd + } else { + std::env::current_dir()?.join(pwd) + }; + Ok(normalize_components(absolute)) +} + +/// Lexically resolve `.` and `..` components without touching the filesystem. +fn normalize_components(path: PathBuf) -> PathBuf { + let mut out = PathBuf::new(); + for component in path.components() { + match component { + Component::CurDir => {} + Component::ParentDir => { + out.pop(); + } + other => out.push(other.as_os_str()), + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fnv1a64_matches_known_vectors_and_is_stable() { + // Standard FNV-1a 64 test vectors. + assert_eq!(fnv1a64(b""), 0xcbf2_9ce4_8422_2325); + assert_eq!(fnv1a64(b"a"), 0xaf63_dc4c_8601_ec8c); + assert_eq!(fnv1a64(b"foobar"), 0x85944171f73967e8); + + // Deterministic across repeated calls within and across runs. + assert_eq!(fnv1a64(b"/tmp/project"), fnv1a64(b"/tmp/project")); + } + + #[test] + fn pwd_key_is_hex_and_collision_free_for_distinct_paths() { + let a = pwd_key(Path::new("/tmp/a")); + let b = pwd_key(Path::new("/tmp/b")); + + assert_ne!(a, b); + assert_eq!(a.len(), 16); + assert!(a.bytes().all(|byte| byte.is_ascii_hexdigit())); + assert_eq!(pwd_key(Path::new("/tmp/a")), a, "stable per input"); + } + + #[test] + fn normalize_resolves_dots_and_parent_dirs_lexically() { + assert_eq!( + normalize_components(PathBuf::from("/tmp/./project/../project/src")), + PathBuf::from("/tmp/project/src") + ); + assert_eq!( + normalize_pwd(PathBuf::from("/tmp/./p/../p")).expect("normalize"), + PathBuf::from("/tmp/p") + ); + } + + #[test] + fn normalize_canonicalizes_existing_paths() { + // An existing temp dir canonicalizes (e.g. /tmp -> /private/tmp on macOS). + let dir = std::env::temp_dir(); + let normalized = normalize_pwd(dir.clone()).expect("normalize existing"); + assert!(normalized.is_absolute()); + let canonicalized = dir.canonicalize().expect("canonicalize"); + assert_eq!(normalized, canonicalized); + } + + #[test] + fn normalize_makes_relative_paths_absolute_without_existing_dir() { + let relative = PathBuf::from("does/not/exist/../exist"); + let normalized = normalize_pwd(relative.clone()).expect("normalize missing dir"); + assert!(normalized.is_absolute()); + assert!(normalized.ends_with("does/not/exist")); + assert!(!normalized.to_string_lossy().contains("..")); + } + + #[test] + fn different_pwds_hash_to_different_keys() { + assert_ne!( + pwd_key(normalize_pwd(PathBuf::from("/w/a")).unwrap().as_path()), + pwd_key(normalize_pwd(PathBuf::from("/w/b")).unwrap().as_path()) + ); + } +} diff --git a/crates/agent/src/session/error.rs b/crates/agent/src/session/error.rs new file mode 100644 index 0000000..b85da98 --- /dev/null +++ b/crates/agent/src/session/error.rs @@ -0,0 +1,26 @@ +use super::store::StoreError; +use std::path::PathBuf; +use thiserror::Error; + +/// Errors from session persistence on top of the raw [`super::store`]. +#[derive(Debug, Error)] +pub enum SessionError { + #[error(transparent)] + Store(#[from] StoreError), + #[error("invalid session header in {path}: {reason}")] + InvalidHeader { path: PathBuf, reason: String }, + #[error("unsupported session schema version {version} in {path} (supported: {supported})")] + UnsupportedVersion { + version: u16, + supported: u16, + path: PathBuf, + }, + #[error("malformed session record in {path}: {reason}")] + MalformedRecord { path: PathBuf, reason: String }, + #[error("failed to serialize session record for {path}: {source}")] + Serialize { + path: PathBuf, + #[source] + source: serde_json::Error, + }, +} diff --git a/crates/agent/src/session/manager.rs b/crates/agent/src/session/manager.rs new file mode 100644 index 0000000..d32d5ca --- /dev/null +++ b/crates/agent/src/session/manager.rs @@ -0,0 +1,707 @@ +use super::error::SessionError; +use super::record::{SESSION_SCHEMA_VERSION, Session, SessionRecord, SessionSummary, now_ms}; +use super::store::StoreError; +use super::store::{JsonlStore, split_complete_lines}; +use crate::context::AgentMessage; +use crate::session::dir::pwd_key; +use llm::{ReasoningEffort, Usage}; +use std::path::{Path, PathBuf}; + +/// Normalize a pwd, mapping I/O failure into the store error shape. +fn normalize(pwd: PathBuf) -> Result { + let dir = pwd.clone(); + super::dir::normalize_pwd(pwd) + .map_err(|source| SessionError::Store(StoreError::CreateDir { dir, source })) +} + +/// Filesystem-backed session store over [`JsonlStore`]. +pub struct SessionManager { + store: JsonlStore, +} + +impl SessionManager { + pub fn new(root: impl Into) -> Self { + Self { + store: JsonlStore::new(root), + } + } + + /// Create a new session and write its header record immediately. + /// + /// Directories are created only here, so constructing the manager does + /// not touch the filesystem. + pub async fn create( + &self, + pwd: impl Into, + provider: impl Into, + model: impl Into, + thinking_level: Option, + ) -> Result { + let pwd = normalize(pwd.into())?; + let key = pwd_key(&pwd); + let session = Session::new(pwd, provider, model, thinking_level); + + // Serialize fully before writing; exclusive creation refuses to + // overwrite an existing session with the same id. + let header = + session + .header_record() + .to_jsonl() + .map_err(|source| SessionError::Serialize { + path: self.store.file_path(&key, &session.id), + source, + })?; + self.store.create_file(&key, &session.id, &header).await?; + + Ok(session) + } + + /// Append exactly one message record to the session file. + pub async fn append_message( + &self, + session: &Session, + message: &AgentMessage, + ) -> Result<(), SessionError> { + let record = SessionRecord::Message { + message: message.clone(), + timestamp_ms: now_ms(), + }; + self.append_record(session, &record).await + } + + /// Append an aggregate usage snapshot. Loading replaces (never sums) + /// `Session.usage` with the newest snapshot. + pub async fn save_usage(&self, session: &Session, usage: &Usage) -> Result<(), SessionError> { + let record = SessionRecord::Usage { + usage: usage.clone(), + timestamp_ms: now_ms(), + }; + self.append_record(session, &record).await + } + + /// Reconstruct a complete session by replaying its records. + pub async fn load(&self, pwd: &Path, id: &str) -> Result { + validate_session_id(id)?; + let normalized = normalize(pwd.to_path_buf())?; + let key = pwd_key(&normalized); + let content = self.store.read(&key, id).await?; + + let mut lines = split_complete_lines(&content); + let path = self.store.file_path(&key, id); + + let header_line = + lines + .next() + .map(|(line, _)| line) + .ok_or_else(|| SessionError::InvalidHeader { + path: path.clone(), + reason: "file is empty".into(), + })?; + let record = + SessionRecord::parse(header_line).map_err(|err| SessionError::InvalidHeader { + path: path.clone(), + reason: err.to_string(), + })?; + let SessionRecord::Session { + id: header_id, + version, + pwd: header_pwd, + provider, + model, + thinking_level, + created_at_ms, + updated_at_ms, + } = record + else { + return Err(SessionError::InvalidHeader { + path: path.clone(), + reason: "first record is not a session header".into(), + }); + }; + + if version > SESSION_SCHEMA_VERSION { + return Err(SessionError::UnsupportedVersion { + version, + supported: SESSION_SCHEMA_VERSION, + path: path.clone(), + }); + } + if header_id != id { + return Err(SessionError::InvalidHeader { + path: path.clone(), + reason: format!("header id {header_id:?} does not match requested id {id:?}"), + }); + } + if normalize(header_pwd.clone())? != normalized { + return Err(SessionError::InvalidHeader { + path: path.clone(), + reason: format!( + "header pwd {:?} does not match requested pwd {:?}", + header_pwd.display(), + normalized.display() + ), + }); + } + + let mut session = Session { + id: header_id, + version, + pwd: header_pwd, + provider, + model, + thinking_level, + messages: Vec::new(), + usage: Usage::default(), + created_at_ms, + updated_at_ms: updated_at_ms.max(created_at_ms), + }; + + for (line, complete) in lines { + if !complete { + continue; + } + let record = + SessionRecord::parse(line).map_err(|err| SessionError::MalformedRecord { + path: path.clone(), + reason: err.to_string(), + })?; + let timestamp_ms = record.timestamp_ms(); + match record { + SessionRecord::Session { .. } => { + return Err(SessionError::MalformedRecord { + path: path.clone(), + reason: "unexpected session header inside file".into(), + }); + } + SessionRecord::Message { message, .. } => session.messages.push(message), + SessionRecord::Usage { usage, .. } => session.usage = usage, + } + session.updated_at_ms = session.updated_at_ms.max(timestamp_ms); + } + + Ok(session) + } + + /// Summaries for every session recorded under `pwd`, newest first. + /// + /// Only headers and usage snapshots are read; message bodies are skipped. + /// Unreadable or corrupt files are skipped rather than failing the whole + /// listing. + pub async fn list(&self, pwd: &Path) -> Result, SessionError> { + let normalized = normalize(pwd.to_path_buf())?; + let key = pwd_key(&normalized); + + let files = self.store.files(&key).await?; + let mut summaries: Vec = Vec::with_capacity(files.len()); + for path in files { + if let Ok(summary) = self.summarize(&path).await { + summaries.push(summary); + } + } + + summaries.sort_by_key(|summary| std::cmp::Reverse(summary.updated_at_ms)); + Ok(summaries) + } + + async fn summarize(&self, path: &Path) -> Result { + let content = tokio::fs::read_to_string(path).await.map_err(|source| { + SessionError::Store(StoreError::ReadFile { + path: path.to_path_buf(), + source, + }) + })?; + let mut lines = split_complete_lines(&content); + + let (header_line, _) = lines.next().ok_or_else(|| SessionError::InvalidHeader { + path: path.to_path_buf(), + reason: "file is empty".into(), + })?; + let record = + SessionRecord::parse(header_line).map_err(|err| SessionError::InvalidHeader { + path: path.to_path_buf(), + reason: err.to_string(), + })?; + let SessionRecord::Session { + id, + version, + pwd, + provider, + model, + thinking_level, + created_at_ms, + updated_at_ms, + } = record + else { + return Err(SessionError::InvalidHeader { + path: path.to_path_buf(), + reason: "first record is not a session header".into(), + }); + }; + if version > SESSION_SCHEMA_VERSION { + return Err(SessionError::UnsupportedVersion { + version, + supported: SESSION_SCHEMA_VERSION, + path: path.to_path_buf(), + }); + } + + let mut latest_usage = Usage::default(); + let mut max_updated_at_ms = updated_at_ms.max(created_at_ms); + for (line, complete) in lines { + if !complete { + continue; + } + if let Ok(record) = SessionRecord::parse(line) { + max_updated_at_ms = max_updated_at_ms.max(record.timestamp_ms()); + if let SessionRecord::Usage { usage, .. } = record { + latest_usage = usage; + } + } + } + + Ok(SessionSummary { + id, + pwd, + provider, + model, + thinking_level, + created_at_ms, + updated_at_ms: max_updated_at_ms, + usage: latest_usage, + }) + } + + async fn append_record( + &self, + session: &Session, + record: &SessionRecord, + ) -> Result<(), SessionError> { + let key = pwd_key(&session.pwd); + let line = record + .to_jsonl() + .map_err(|source| SessionError::Serialize { + path: self.store.file_path(&key, &session.id), + source, + })?; + Ok(self.store.append(&key, &session.id, &line).await?) + } +} + +fn validate_session_id(id: &str) -> Result<(), SessionError> { + let valid = !id.is_empty() + && id.len() <= 64 + && id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-'); + if valid { + Ok(()) + } else { + Err(SessionError::MalformedRecord { + path: PathBuf::from(id), + reason: "session id must be 1-64 ascii alphanumerics or dashes".into(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::context::AgentMessage; + use crate::session::store::JSONL_EXTENSION; + + fn temp_root(name: &str) -> PathBuf { + let dir = + std::env::temp_dir().join(format!("alan-session-{name}-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).expect("create temp root"); + dir + } + + fn cleanup(dir: &Path) { + let _ = std::fs::remove_dir_all(dir); + } + + fn usage(input: u64, output: u64) -> Usage { + Usage { + input_tokens: input, + output_tokens: output, + ..Usage::default() + } + } + + fn session_file(root: &Path, pwd: &str, id: &str) -> PathBuf { + root.join(pwd_key(Path::new(pwd))) + .join(format!("{id}.{JSONL_EXTENSION}")) + } + + #[tokio::test] + async fn constructing_manager_creates_nothing() { + let root = + std::env::temp_dir().join(format!("alan-session-construct-{}", uuid::Uuid::new_v4())); + let _manager = SessionManager::new(&root); + assert!(!root.exists(), "manager construction must not create files"); + } + + #[tokio::test] + async fn create_writes_single_header_file_under_pwd_dir() { + let root = temp_root("create"); + let manager = SessionManager::new(&root); + + let session = manager + .create("/tmp/project", "openrouter", "test-model", None) + .await + .expect("create session"); + + let pwd_dir = root.join(pwd_key(Path::new("/tmp/project"))); + assert!(pwd_dir.is_dir(), "pwd-specific directory exists"); + let entries: Vec<_> = std::fs::read_dir(&pwd_dir).expect("read pwd dir").collect(); + assert_eq!(entries.len(), 1, "exactly one file after create"); + + let content = std::fs::read_to_string(session_file(&root, "/tmp/project", &session.id)) + .expect("read session file"); + let first = content.lines().next().expect("header line"); + let record: SessionRecord = serde_json::from_str(first).expect("valid header json"); + assert!(matches!(record, SessionRecord::Session { .. })); + cleanup(&root); + } + + #[tokio::test] + async fn different_pwds_use_different_directories() { + let root = temp_root("pwds"); + let manager = SessionManager::new(&root); + + manager + .create("/tmp/a", "openrouter", "m", None) + .await + .expect("create a"); + manager + .create("/tmp/b", "openrouter", "m", None) + .await + .expect("create b"); + + assert_ne!(pwd_key(Path::new("/tmp/a")), pwd_key(Path::new("/tmp/b"))); + assert_eq!(std::fs::read_dir(&root).unwrap().count(), 2); + cleanup(&root); + } + + #[tokio::test] + async fn append_and_load_roundtrip() { + let root = temp_root("roundtrip"); + let manager = SessionManager::new(&root); + + let session = manager + .create( + "/tmp/project", + "openrouter", + "test-model", + Some(ReasoningEffort::High), + ) + .await + .expect("create"); + + manager + .append_message(&session, &AgentMessage::user("hello")) + .await + .expect("append user"); + manager + .append_message(&session, &AgentMessage::user("second")) + .await + .expect("append second"); + manager + .save_usage(&session, &usage(10, 5)) + .await + .expect("usage 1"); + manager + .save_usage(&session, &usage(25, 8)) + .await + .expect("usage 2"); + + let loaded = manager + .load(Path::new("/tmp/project"), &session.id) + .await + .expect("load"); + + assert_eq!(loaded.id, session.id); + assert_eq!(loaded.version, SESSION_SCHEMA_VERSION); + assert_eq!(loaded.pwd, session.pwd); + assert_eq!(loaded.provider, "openrouter"); + assert_eq!(loaded.model, "test-model"); + assert_eq!(loaded.thinking_level, Some(ReasoningEffort::High)); + assert_eq!(loaded.created_at_ms, session.created_at_ms); + assert_eq!( + loaded.messages, + vec![AgentMessage::user("hello"), AgentMessage::user("second")] + ); + assert_eq!(loaded.usage, usage(25, 8)); + cleanup(&root); + } + + #[tokio::test] + async fn usage_snapshots_replace_not_sum() { + let root = temp_root("usage"); + let manager = SessionManager::new(&root); + let session = manager + .create("/tmp/p", "o", "m", None) + .await + .expect("create"); + + manager + .save_usage(&session, &usage(10, 5)) + .await + .expect("usage 1"); + manager + .save_usage(&session, &usage(20, 7)) + .await + .expect("usage 2"); + + let loaded = manager + .load(Path::new("/tmp/p"), &session.id) + .await + .expect("load"); + assert_eq!(loaded.usage, usage(20, 7)); + cleanup(&root); + } + + #[tokio::test] + async fn truncated_final_line_tolerated_malformed_complete_line_rejected() { + let root = temp_root("truncated"); + let manager = SessionManager::new(&root); + let session = manager + .create("/tmp/p", "o", "m", None) + .await + .expect("create"); + let path = session_file(&root, "/tmp/p", &session.id); + + // Truncated final append (no trailing newline) is tolerated. + let mut content = std::fs::read_to_string(&path).unwrap(); + content.push_str( + r#"{"type":"message","message":{"kind":"user","content":"ok"},"timestamp_ms":1}"#, + ); + content.push('\n'); + content.push_str(r#"{"type":"message","message":{"kind":"user""#); + std::fs::write(&path, &content).unwrap(); + + let loaded = manager + .load(Path::new("/tmp/p"), &session.id) + .await + .expect("truncated tail tolerated"); + assert_eq!(loaded.messages.len(), 1); + + // A malformed complete line is an error, even at the end. + content.push_str("\nthis is not json\n"); + std::fs::write(&path, &content).unwrap(); + + let err = manager + .load(Path::new("/tmp/p"), &session.id) + .await + .expect_err("malformed complete line rejected"); + assert!(matches!(err, SessionError::MalformedRecord { .. })); + cleanup(&root); + } + + #[tokio::test] + async fn wrong_pwd_or_id_does_not_load() { + let root = temp_root("cross"); + let manager = SessionManager::new(&root); + let session = manager + .create("/tmp/p", "o", "m", None) + .await + .expect("create"); + + let err = manager + .load(Path::new("/tmp/other"), &session.id) + .await + .expect_err("wrong pwd must not load"); + assert!(matches!(err, SessionError::Store(StoreError::NotFound(_)))); + + let err = manager + .load(Path::new("/tmp/p"), "not-a-real-id") + .await + .expect_err("wrong id must not load"); + assert!(matches!(err, SessionError::Store(StoreError::NotFound(_)))); + cleanup(&root); + } + + #[tokio::test] + async fn second_create_with_same_id_cannot_overwrite() { + let root = temp_root("overwrite"); + let manager = SessionManager::new(&root); + let pwd = "/tmp/p"; + + let session = manager + .create(pwd, "o", "m", None) + .await + .expect("first create"); + let before = std::fs::read_to_string(session_file(&root, pwd, &session.id)).unwrap(); + + // Simulate a second create racing onto the same id: the exclusive + // file creation must refuse rather than truncate the existing file. + let mut collision = Session::new(pwd, "o", "m", None); + collision.id = session.id.clone(); + collision.created_at_ms = 123_456; + collision.updated_at_ms = 123_456; + let err = manager + .write_session_file_for_test(&collision) + .await + .expect_err("collision refused"); + assert!(matches!( + err, + SessionError::Store(StoreError::AlreadyExists(_)) + )); + + let after = std::fs::read_to_string(session_file(&root, pwd, &session.id)).unwrap(); + assert_eq!(before, after, "existing session untouched"); + cleanup(&root); + } + + #[tokio::test] + async fn load_validates_header_id_and_schema_and_pwd() { + let root = temp_root("header"); + let manager = SessionManager::new(&root); + let pwd = "/tmp/p"; + let session = manager.create(pwd, "o", "m", None).await.expect("create"); + let path = session_file(&root, pwd, &session.id); + + // Wrong schema version. + let content = std::fs::read_to_string(&path).unwrap(); + std::fs::write( + &path, + content.replace(r#""version":1,"#, r#""version":99,"#), + ) + .unwrap(); + let err = manager + .load(Path::new(pwd), &session.id) + .await + .expect_err("unsupported version rejected"); + assert!(matches!(err, SessionError::UnsupportedVersion { .. })); + + // Mismatched header id (fresh file). + let session = manager.create(pwd, "o", "m", None).await.expect("create"); + let path = session_file(&root, pwd, &session.id); + let content = std::fs::read_to_string(&path).unwrap(); + std::fs::write(&path, content.replace(&session.id, "other-id")).unwrap(); + let err = manager + .load(Path::new(pwd), &session.id) + .await + .expect_err("mismatched header id rejected"); + assert!(matches!(err, SessionError::InvalidHeader { .. })); + + // Mismatched header pwd / cross-directory load (fresh file). + let session = manager.create(pwd, "o", "m", None).await.expect("create"); + let path = session_file(&root, pwd, &session.id); + let content = std::fs::read_to_string(&path).unwrap(); + std::fs::write(&path, content.replace(pwd, "/elsewhere")).unwrap(); + let err = manager + .load(Path::new(pwd), &session.id) + .await + .expect_err("mismatched header pwd rejected"); + assert!(matches!(err, SessionError::InvalidHeader { .. })); + cleanup(&root); + } + + #[tokio::test] + async fn list_returns_summaries_without_loading_messages() { + let root = temp_root("list"); + let manager = SessionManager::new(&root); + + let older = manager + .create("/tmp/p", "openrouter", "m", Some(ReasoningEffort::Low)) + .await + .expect("older"); + manager + .append_message(&older, &AgentMessage::user("hi")) + .await + .expect("append"); + manager + .save_usage(&older, &usage(3, 4)) + .await + .expect("usage"); + + let newer = manager + .create("/tmp/p", "openrouter", "m2", None) + .await + .expect("newer"); + + let summaries = manager.list(Path::new("/tmp/p")).await.expect("list"); + assert_eq!(summaries.len(), 2); + // Newest first; ids break ties deterministically for same-ms creates. + assert!( + (summaries[0].updated_at_ms, summaries[1].id.clone()) + >= (summaries[1].updated_at_ms, summaries[0].id.clone()) + ); + + let summary = summaries + .iter() + .find(|summary| summary.id == older.id) + .expect("older summary"); + assert_eq!(summary.pwd, older.pwd); + assert_eq!(summary.provider, "openrouter"); + assert_eq!(summary.model, "m"); + assert_eq!(summary.thinking_level, Some(ReasoningEffort::Low)); + assert_eq!(summary.created_at_ms, older.created_at_ms); + assert_eq!(summary.usage, usage(3, 4)); + + let _ = newer; + assert!( + manager + .list(Path::new("/tmp/none")) + .await + .expect("list") + .is_empty() + ); + cleanup(&root); + } + + #[tokio::test] + async fn append_to_missing_session_fails() { + let root = temp_root("missing-append"); + let manager = SessionManager::new(&root); + let session = Session::new("/tmp/p", "o", "m", None); + + let err = manager + .append_message(&session, &AgentMessage::user("hi")) + .await + .expect_err("append without create fails"); + assert!(matches!(err, SessionError::Store(StoreError::NotFound(_)))); + cleanup(&root); + } + + #[cfg(unix)] + #[tokio::test] + async fn unix_permissions_are_restrictive() { + use std::os::unix::fs::PermissionsExt; + + let root = temp_root("perms"); + let manager = SessionManager::new(&root); + let session = manager + .create("/tmp/p", "o", "m", None) + .await + .expect("create"); + + let pwd_dir = root.join(pwd_key(Path::new("/tmp/p"))); + let dir_mode = std::fs::metadata(&pwd_dir).unwrap().permissions().mode(); + let file_mode = std::fs::metadata(session_file(&root, "/tmp/p", &session.id)) + .unwrap() + .permissions() + .mode(); + assert_eq!(dir_mode & 0o777, 0o700, "directory mode"); + assert_eq!(file_mode & 0o777, 0o600, "session file mode"); + cleanup(&root); + } + + /// Test-only re-entry into the create path used to verify collision + /// handling without relying on UUID luck. + impl SessionManager { + async fn write_session_file_for_test(&self, session: &Session) -> Result<(), SessionError> { + let key = pwd_key(&session.pwd); + let header = + session + .header_record() + .to_jsonl() + .map_err(|source| SessionError::Serialize { + path: self.store.file_path(&key, &session.id), + source, + })?; + Ok(self.store.create_file(&key, &session.id, &header).await?) + } + } +} diff --git a/crates/agent/src/session/mod.rs b/crates/agent/src/session/mod.rs new file mode 100644 index 0000000..b680fb8 --- /dev/null +++ b/crates/agent/src/session/mod.rs @@ -0,0 +1,9 @@ +mod dir; +mod error; +mod manager; +mod record; +mod store; + +pub use error::SessionError; +pub use manager::SessionManager; +pub use record::{SESSION_SCHEMA_VERSION, Session, SessionRecord, SessionSummary}; diff --git a/crates/agent/src/session.rs b/crates/agent/src/session/record.rs similarity index 59% rename from crates/agent/src/session.rs rename to crates/agent/src/session/record.rs index 655e5b4..b61263d 100644 --- a/crates/agent/src/session.rs +++ b/crates/agent/src/session/record.rs @@ -6,7 +6,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; pub const SESSION_SCHEMA_VERSION: u16 = 1; -fn now_ms() -> u64 { +pub(crate) fn now_ms() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) .expect("system clock is before Unix epoch") @@ -63,6 +63,7 @@ impl Session { } } + /// Apply one replayed record to this in-memory session. pub fn record(&mut self, record: SessionRecord) { match record { SessionRecord::Session { .. } => {} @@ -73,6 +74,27 @@ impl Session { } } +/// Lightweight session description built from the header record and replayed +/// usage snapshots; message bodies are not loaded. +#[derive(Debug, Clone, PartialEq)] +pub struct SessionSummary { + pub id: String, + pub pwd: PathBuf, + pub provider: String, + pub model: String, + pub thinking_level: Option, + pub created_at_ms: u64, + pub updated_at_ms: u64, + pub usage: Usage, +} + +/// One JSONL record in a session file. +/// +/// ```jsonl +/// {"type":"session","version":1,"id":"018f..","pwd":"/tmp/project",..} +/// {"type":"message","message":{..},"timestamp_ms":1700000000000} +/// {"type":"usage","usage":{..},"timestamp_ms":1700000001000} +/// ``` #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] pub enum SessionRecord { @@ -96,11 +118,80 @@ pub enum SessionRecord { }, } +impl SessionRecord { + /// Timestamp used to reconstruct `updated_at_ms` while loading. + pub(super) fn timestamp_ms(&self) -> u64 { + match self { + Self::Session { updated_at_ms, .. } => *updated_at_ms, + Self::Message { timestamp_ms, .. } | Self::Usage { timestamp_ms, .. } => *timestamp_ms, + } + } + + /// Serialize as one complete JSONL line (trailing newline included). + pub(super) fn to_jsonl(&self) -> Result { + let mut line = serde_json::to_string(self)?; + line.push('\n'); + Ok(line) + } + + /// Parse one complete JSONL line. Internally tagged enums reject unknown + /// `type` values and missing fields, so malformed lines fail loudly + /// instead of becoming empty values. + pub(super) fn parse(line: &str) -> Result { + serde_json::from_str(line) + } +} + #[cfg(test)] mod tests { use super::*; use crate::context::AgentMessage; + fn usage(input: u64, output: u64) -> Usage { + Usage { + input_tokens: input, + output_tokens: output, + ..Usage::default() + } + } + + #[test] + fn parse_rejects_malformed_record() { + assert!(SessionRecord::parse("not json at all").is_err()); + assert!(SessionRecord::parse(r#"{"type":"unknown"}"#).is_err()); + assert!(SessionRecord::parse(r#"{"message":{}}"#).is_err()); + } + + #[test] + fn parse_accepts_valid_message_record() { + let record = SessionRecord::parse( + r#"{"type":"message","message":{"kind":"user","content":"hi"},"timestamp_ms":42}"#, + ) + .expect("valid message record"); + assert_eq!( + record, + SessionRecord::Message { + message: AgentMessage::user("hi"), + timestamp_ms: 42, + } + ); + } + + #[test] + fn to_jsonl_round_trips_one_line_per_record() { + let record = SessionRecord::Usage { + usage: usage(3, 4), + timestamp_ms: 42, + }; + let line = record.to_jsonl().expect("serialize"); + assert!(line.ends_with('\n')); + assert_eq!(line.lines().count(), 1); + assert_eq!( + SessionRecord::parse(line.trim_end()).expect("parse"), + record + ); + } + #[test] fn header_record_preserves_session_metadata() { let mut session = Session::new("/tmp/project", "openrouter", "test-model", None); @@ -139,29 +230,17 @@ mod tests { message: AgentMessage::user("first"), timestamp_ms: 1_000, }); - session.record(SessionRecord::Message { - message: AgentMessage::user("second"), - timestamp_ms: 2_000, - }); session.record(SessionRecord::Usage { - usage: Usage { - input_tokens: 100, - output_tokens: 50, - ..Usage::default() - }, + usage: usage(100, 50), timestamp_ms: 3_000, }); // A later snapshot replaces the earlier one; it must not be summed. session.record(SessionRecord::Usage { - usage: Usage { - input_tokens: 150, - output_tokens: 60, - ..Usage::default() - }, + usage: usage(150, 60), timestamp_ms: 4_000, }); - assert_eq!(session.messages.len(), 2); + assert_eq!(session.messages.len(), 1); assert_eq!(session.usage.input_tokens, 150); assert_eq!(session.usage.output_tokens, 60); } diff --git a/crates/agent/src/session/store.rs b/crates/agent/src/session/store.rs new file mode 100644 index 0000000..05e7ffa --- /dev/null +++ b/crates/agent/src/session/store.rs @@ -0,0 +1,440 @@ +use std::path::{Path, PathBuf}; +use thiserror::Error; +use tokio::fs::{self, File, OpenOptions}; +use tokio::io::AsyncWriteExt; + +/// Errors from the raw JSONL store. +#[derive(Debug, Error)] +pub enum StoreError { + #[error("failed to create directory {}: {source}", dir.display())] + CreateDir { + dir: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("failed to create file {}: {source}", path.display())] + CreateFile { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("file already exists: {0}")] + AlreadyExists(PathBuf), + #[error("failed to open file {}: {source}", path.display())] + OpenFile { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("failed to write to file {}: {source}", path.display())] + WriteFile { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("failed to read file {}: {source}", path.display())] + ReadFile { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("file not found: {0}")] + NotFound(PathBuf), + #[error("file is locked by another writer: {0}")] + Locked(PathBuf), +} + +/// Append-only JSONL store over a two-level layout: +/// `//.jsonl`. +pub(crate) struct JsonlStore { + root: PathBuf, +} + +impl JsonlStore { + pub(crate) fn new(root: impl Into) -> Self { + Self { root: root.into() } + } + + /// Exclusively create `//.jsonl`, creating the key + /// directory if needed, and write `first_line` as its first record. + /// + /// Returns [`StoreError::AlreadyExists`] if the file already exists; the + /// existing file is never touched. + pub(crate) async fn create_file( + &self, + key: &str, + name: &str, + first_line: &str, + ) -> Result<(), StoreError> { + let dir = self.key_dir(key); + fs::create_dir_all(&dir) + .await + .map_err(|source| StoreError::CreateDir { + dir: dir.clone(), + source, + })?; + set_permissions(&dir, true).await; + + let path = self.file_path(key, name); + // `create_new` guarantees an existing file is never overwritten. + let mut file = File::create_new(&path).await.map_err(|source| { + if source.kind() == std::io::ErrorKind::AlreadyExists { + StoreError::AlreadyExists(path.clone()) + } else { + StoreError::CreateFile { + path: path.clone(), + source, + } + } + })?; + set_permissions(&path, false).await; + + Self::write_line(&path, &mut file, first_line).await + } + + /// Append one newline-terminated record under an exclusive sidecar lock. + pub(crate) async fn append(&self, key: &str, name: &str, line: &str) -> Result<(), StoreError> { + let path = self.file_path(key, name); + if !path.is_file() { + return Err(StoreError::NotFound(path)); + } + + let _lock = FileLock::acquire(self.lock_path(key, name)).await?; + + let mut file = OpenOptions::new() + .append(true) + .open(&path) + .await + .map_err(|source| StoreError::OpenFile { + path: path.clone(), + source, + })?; + Self::write_line(&path, &mut file, line).await + } + + /// Read the whole file as text. + pub(crate) async fn read(&self, key: &str, name: &str) -> Result { + let path = self.file_path(key, name); + if !path.is_file() { + return Err(StoreError::NotFound(path)); + } + fs::read_to_string(&path) + .await + .map_err(|source| StoreError::ReadFile { path, source }) + } + + /// Paths of every `.jsonl` file stored under `key`, unsorted. + /// + /// Missing directories yield an empty list: no files yet. + pub(crate) async fn files(&self, key: &str) -> Result, StoreError> { + let dir = self.key_dir(key); + let read_dir = match fs::read_dir(&dir).await { + Ok(entries) => entries, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(source) => return Err(StoreError::ReadFile { path: dir, source }), + }; + + let mut paths = Vec::new(); + let mut entries = read_dir; + loop { + match entries.next_entry().await { + Ok(Some(entry)) => { + let path = entry.path(); + if path.extension().and_then(|ext| ext.to_str()) == Some(JSONL_EXTENSION) { + paths.push(path); + } + } + Ok(None) => break, + Err(source) => { + return Err(StoreError::ReadFile { + path: dir.clone(), + source, + }); + } + } + } + Ok(paths) + } + + /// Directory holding all files for `key`. + pub(crate) fn key_dir(&self, key: &str) -> PathBuf { + self.root.join(key) + } + + /// Full path of one stored file. + pub(crate) fn file_path(&self, key: &str, name: &str) -> PathBuf { + self.key_dir(key).join(format!("{name}.{JSONL_EXTENSION}")) + } + + /// Sidecar lock path used by [`JsonlStore::append`]. + pub(crate) fn lock_path(&self, key: &str, name: &str) -> PathBuf { + self.key_dir(key) + .join(format!("{name}.{JSONL_EXTENSION}{LOCK_SUFFIX}")) + } + + async fn write_line(path: &Path, file: &mut File, line: &str) -> Result<(), StoreError> { + file.write_all(line.as_bytes()) + .await + .map_err(|source| StoreError::WriteFile { + path: path.to_path_buf(), + source, + })?; + file.flush().await.map_err(|source| StoreError::WriteFile { + path: path.to_path_buf(), + source, + }) + } +} + +pub(crate) const JSONL_EXTENSION: &str = "jsonl"; +const LOCK_SUFFIX: &str = ".lock"; + +/// Yield `(line, is_complete)` pairs from raw file contents. +/// +/// An incomplete final line (no trailing newline, i.e. a crash-truncated +/// append) is yielded with `is_complete == false`; every other line is +/// complete. Callers decide whether to surface malformed complete lines. +pub(crate) fn split_complete_lines(content: &str) -> impl Iterator { + let (body, truncated_tail) = if content.is_empty() { + ("", None) + } else { + match content.strip_suffix('\n') { + Some(body) => (body, None), + None => match content.rsplit_once('\n') { + Some((body, tail)) => (body, Some(tail)), + // No newline at all: single truncated line. + None => ("", Some(content)), + }, + } + }; + + body.lines() + .map(|line| (line, true)) + .chain(truncated_tail.map(|tail| (tail, false))) +} + +/// Exclusive, self-removing sidecar lock guard around appends. +struct FileLock { + path: PathBuf, +} + +impl FileLock { + async fn acquire(path: PathBuf) -> Result { + File::create_new(&path).await.map_err(|source| { + if source.kind() == std::io::ErrorKind::AlreadyExists { + StoreError::Locked(path.clone()) + } else { + StoreError::CreateFile { + path: path.clone(), + source, + } + } + })?; + Ok(Self { path }) + } +} + +impl Drop for FileLock { + fn drop(&mut self) { + // Use blocking remove_file in Drop; this is fine because Drop is + // synchronous and file removal is a fast operation. + let _ = std::fs::remove_file(&self.path); + } +} + +#[cfg(unix)] +async fn set_permissions(path: &Path, is_dir: bool) { + use std::fs::Permissions; + use std::os::unix::fs::PermissionsExt; + + let mode = if is_dir { 0o700 } else { 0o600 }; + let _ = fs::set_permissions(path, Permissions::from_mode(mode)).await; +} + +#[cfg(test)] +mod tests { + use super::*; + + fn temp_root(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("alan-store-{name}-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).expect("create temp root"); + dir + } + + fn cleanup(dir: &Path) { + let _ = std::fs::remove_dir_all(dir); + } + + #[tokio::test] + async fn constructing_store_creates_nothing() { + let root = + std::env::temp_dir().join(format!("alan-store-construct-{}", uuid::Uuid::new_v4())); + let _store = JsonlStore::new(&root); + assert!(!root.exists(), "store construction must not create files"); + } + + #[tokio::test] + async fn create_file_writes_first_line_and_refuses_overwrite() { + let root = temp_root("create"); + let store = JsonlStore::new(&root); + + store + .create_file("key-a", "one", "{\"first\":true}\n") + .await + .expect("create"); + + let path = store.file_path("key-a", "one"); + assert_eq!( + std::fs::read_to_string(&path).unwrap(), + "{\"first\":true}\n" + ); + + // Exclusive creation refuses to truncate or touch the existing file. + let err = store + .create_file("key-a", "one", "{\"second\":true}\n") + .await + .expect_err("second create refused"); + assert!(matches!(err, StoreError::AlreadyExists(_))); + assert_eq!( + std::fs::read_to_string(&path).unwrap(), + "{\"first\":true}\n" + ); + cleanup(&root); + } + + #[tokio::test] + async fn create_file_creates_key_directory() { + let root = temp_root("mkdir"); + let store = JsonlStore::new(&root); + + store + .create_file("key-x", "f", "x\n") + .await + .expect("create"); + assert!(store.key_dir("key-x").is_dir()); + cleanup(&root); + } + + #[tokio::test] + async fn append_produces_newline_terminated_records_in_order() { + let root = temp_root("append"); + let store = JsonlStore::new(&root); + store.create_file("k", "f", "a\n").await.expect("create"); + + store.append("k", "f", "b\n").await.expect("append b"); + store.append("k", "f", "c\n").await.expect("append c"); + + assert_eq!(store.read("k", "f").await.unwrap(), "a\nb\nc\n"); + cleanup(&root); + } + + #[test] + fn split_lines_flags_truncated_tail_only() { + let complete: Vec<_> = split_complete_lines("a\nb\n").collect(); + assert_eq!(complete, vec![("a", true), ("b", true)]); + + let truncated: Vec<_> = split_complete_lines("a\nb").collect(); + assert_eq!(truncated, vec![("a", true), ("b", false)]); + + let none: Vec<_> = split_complete_lines("").collect(); + assert!(none.is_empty()); + + let only_truncated: Vec<_> = split_complete_lines("abc").collect(); + assert_eq!(only_truncated, vec![("abc", false)]); + } + + #[tokio::test] + async fn read_missing_and_append_missing_are_not_found() { + let root = temp_root("missing"); + let store = JsonlStore::new(&root); + + let err = store.read("nope", "f").await.expect_err("read missing"); + assert!(matches!(err, StoreError::NotFound(_))); + + let err = store + .append("nope", "f", "x\n") + .await + .expect_err("append missing"); + assert!(matches!(err, StoreError::NotFound(_))); + cleanup(&root); + } + + #[tokio::test] + async fn files_lists_only_jsonl_entries() { + let root = temp_root("files"); + let store = JsonlStore::new(&root); + store + .create_file("k", "one", "1\n") + .await + .expect("create one"); + store + .create_file("k", "two", "2\n") + .await + .expect("create two"); + + let mut names: Vec<_> = store + .files("k") + .await + .expect("files") + .iter() + .map(|path| path.file_name().unwrap().to_string_lossy().into_owned()) + .collect(); + names.sort(); + assert_eq!(names, vec!["one.jsonl", "two.jsonl"]); + + // Lock sidecars are not listed, and unknown keys list empty. + tokio::fs::write(store.lock_path("k", "one"), "") + .await + .unwrap(); + assert_eq!(store.files("k").await.unwrap().len(), 2); + assert!(store.files("absent").await.unwrap().is_empty()); + cleanup(&root); + } + + #[cfg(unix)] + #[tokio::test] + async fn unix_permissions_are_restrictive() { + use std::os::unix::fs::PermissionsExt; + + let root = temp_root("perms"); + let store = JsonlStore::new(&root); + store.create_file("k", "f", "x\n").await.expect("create"); + + let dir_mode = std::fs::metadata(store.key_dir("k")) + .unwrap() + .permissions() + .mode(); + let file_mode = std::fs::metadata(store.file_path("k", "f")) + .unwrap() + .permissions() + .mode(); + assert_eq!(dir_mode & 0o777, 0o700, "directory mode"); + assert_eq!(file_mode & 0o777, 0o600, "file mode"); + cleanup(&root); + } + + #[tokio::test] + async fn locked_while_sidecar_lock_held() { + let root = temp_root("locked"); + let store = JsonlStore::new(&root); + store.create_file("k", "f", "x\n").await.expect("create"); + + let lock = FileLock::acquire(store.lock_path("k", "f")) + .await + .expect("acquire lock"); + let err = store + .append("k", "f", "y\n") + .await + .expect_err("append blocked"); + assert!(matches!(err, StoreError::Locked(_))); + + drop(lock); // Lock removed on drop. + store + .append("k", "f", "y\n") + .await + .expect("append after unlock"); + assert_eq!(store.read("k", "f").await.unwrap(), "x\ny\n"); + assert!(!store.lock_path("k", "f").exists(), "lock removed on drop"); + cleanup(&root); + } +} From 8af8ce4d361aa512a8d35235d8863a6a03cc37c0 Mon Sep 17 00:00:00 2001 From: Revantark Date: Mon, 24 Aug 2026 16:33:50 +0530 Subject: [PATCH 2/4] improve tests --- crates/agent/src/agent.rs | 606 +++++++++++++++++++++++++++-- crates/agent/src/context.rs | 1 - crates/agent/src/error.rs | 2 + crates/agent/src/session/mod.rs | 1 + crates/agent/src/session/record.rs | 2 +- crates/alan/src/core/controller.rs | 4 +- crates/alan/src/main.rs | 2 +- crates/providers/src/model.rs | 4 + 8 files changed, 580 insertions(+), 42 deletions(-) diff --git a/crates/agent/src/agent.rs b/crates/agent/src/agent.rs index 137a24f..7d754fb 100644 --- a/crates/agent/src/agent.rs +++ b/crates/agent/src/agent.rs @@ -1,3 +1,4 @@ +use crate::session::{Session, SessionError, SessionManager, StoreError}; use crate::{ AgentError, AgentMessage, AgentTool, Skill, build_system_prompt, context::AgentContext, }; @@ -8,6 +9,7 @@ use llm::{ }; use providers::{Model, ModelError}; use std::{ + path::PathBuf, sync::{ Arc, atomic::{AtomicBool, Ordering}, @@ -78,7 +80,9 @@ pub struct Agent { context: Mutex, plan_mode: AtomicBool, max_tool_rounds: usize, - session_id: String, + session_id: Mutex, + session_manager: Option>, + active_session: Mutex>, } impl Agent { @@ -89,17 +93,30 @@ impl Agent { skills: Vec::new(), tools: Vec::new(), max_tool_rounds: 100, + session_manager: None, + resumed_session: None, } } /// Buffered prompt: runs to completion and returns the final response. pub async fn prompt(&self, content: impl Into) -> Result { + let content = content.into(); + if content.trim().is_empty() { + return Err(AgentError::Model(ModelError::Llm( + llm::LlmError::Configuration("empty prompt".into()), + ))); + } + let model = self.model.lock().await; let mut context = self.context.lock().await; let plan_mode = self.plan_mode(); - context - .messages - .push(AgentMessage::user(prompt_content(content, plan_mode))); + let user_msg = AgentMessage::user(prompt_content(content, plan_mode)); + + self.ensure_session(&model).await?; + + context.messages.push(user_msg); + self.persist_message(&context.messages.last().unwrap()) + .await?; let (_cancellation, mut cancellation_receiver) = watch::channel(false); let mut partial = String::new(); @@ -113,9 +130,9 @@ impl Agent { ) .await; if result.is_err() && !partial.is_empty() { - context - .messages - .push(AgentMessage::Assistant(partial_response(&partial))); + let partial_msg = AgentMessage::Assistant(partial_response(&partial)); + context.messages.push(partial_msg.clone()); + self.persist_message(&partial_msg).await?; } result } @@ -163,12 +180,22 @@ impl Agent { events: &Sender>, mut cancellation: watch::Receiver, ) -> Result { + if content.trim().is_empty() { + return Err(AgentError::Model(ModelError::Llm( + llm::LlmError::Configuration("empty prompt".into()), + ))); + } + let model = self.model.lock().await; let mut context = self.context.lock().await; let plan_mode = self.plan_mode(); - context - .messages - .push(AgentMessage::user(prompt_content(content, plan_mode))); + let user_msg = AgentMessage::user(prompt_content(content, plan_mode)); + + self.ensure_session(&model).await?; + + context.messages.push(user_msg); + self.persist_message(&context.messages.last().unwrap()) + .await?; let mut partial = String::new(); let result = self @@ -181,9 +208,9 @@ impl Agent { ) .await; if result.is_err() && !partial.is_empty() { - context - .messages - .push(AgentMessage::Assistant(partial_response(&partial))); + let partial_msg = AgentMessage::Assistant(partial_response(&partial)); + context.messages.push(partial_msg.clone()); + self.persist_message(&partial_msg).await?; } result } @@ -199,8 +226,9 @@ impl Agent { let plan = self.plan_mode(); for _ in 0..self.max_tool_rounds { Self::check_cancelled(cancellation)?; + let session_id = self.session_id.lock().await.clone(); let response = Self::stream_round( - self.session_id.clone(), + session_id, model, context, events, @@ -209,10 +237,13 @@ impl Agent { plan, ) .await?; + let calls: Vec<_> = response.tool_calls().cloned().collect(); if let Some(usage) = response.usage.as_ref() { context.usage.accumulate(usage); + self.persist_usage(&context.usage).await?; } + if calls.is_empty() { if let Some(events) = events { Self::send_event( @@ -224,12 +255,16 @@ impl Agent { ) .await?; } - context - .messages - .push(AgentMessage::Assistant(response.clone())); + let msg = AgentMessage::Assistant(response.clone()); + context.messages.push(msg.clone()); + self.persist_message(&msg).await?; return Ok(response); } - context.messages.push(AgentMessage::Assistant(response)); + + let assistant_msg = AgentMessage::Assistant(response); + context.messages.push(assistant_msg.clone()); + self.persist_message(&assistant_msg).await?; + for call in calls { Self::check_cancelled(cancellation)?; let tool_index = context @@ -257,10 +292,13 @@ impl Agent { let result = context.tools[tool_index].executor.execute(&call).await; match result { Ok(result) => { - context.messages.push(AgentMessage::ToolResult { + let msg = AgentMessage::ToolResult { tool_call_id: call_id.clone(), content: result.clone(), - }); + }; + context.messages.push(msg.clone()); + self.persist_message(&msg).await?; + if let Some(events) = events { Self::send_event( events, @@ -276,10 +314,13 @@ impl Agent { } Err(error) => { let error = error.to_string(); - context.messages.push(AgentMessage::ToolResult { + let msg = AgentMessage::ToolResult { tool_call_id: call_id.clone(), content: error.clone(), - }); + }; + context.messages.push(msg.clone()); + self.persist_message(&msg).await?; + if let Some(events) = events { Self::send_event( events, @@ -313,7 +354,6 @@ impl Agent { .collect(); let messages = Self::build_messages(context); let options = RequestOptions { - // when we add persistent sessions, we need to get these from disk prompt_cache_key: Some(session_id.clone()), session_id: Some(session_id), ..RequestOptions::default() @@ -404,6 +444,55 @@ impl Agent { messages.extend(context.messages.iter().map(AgentMessage::to_llm)); messages } + + async fn ensure_session(&self, model: &Model) -> Result<(), AgentError> { + let mut active_session = self.active_session.lock().await; + if active_session.is_some() { + return Ok(()); + } + + let manager = match &self.session_manager { + Some(m) => m, + None => return Ok(()), + }; + + let pwd = std::env::current_dir().map_err(|e| { + AgentError::Session(SessionError::Store(StoreError::CreateDir { + dir: PathBuf::from("."), + source: e, + })) + })?; + + let session = manager + .create( + pwd, + model.info().provider.0.clone(), + model.info().id.clone(), + model.reasoning_effort(), + ) + .await?; + + *self.session_id.lock().await = session.id.clone(); + *active_session = Some(session); + + Ok(()) + } + + async fn persist_message(&self, message: &AgentMessage) -> Result<(), AgentError> { + let active_session = self.active_session.lock().await; + if let (Some(manager), Some(session)) = (&self.session_manager, &*active_session) { + manager.append_message(session, message).await?; + } + Ok(()) + } + + async fn persist_usage(&self, usage: &Usage) -> Result<(), AgentError> { + let active_session = self.active_session.lock().await; + if let (Some(manager), Some(session)) = (&self.session_manager, &*active_session) { + manager.save_usage(session, usage).await?; + } + Ok(()) + } } fn prompt_content(content: impl Into, plan_mode: bool) -> String { @@ -438,6 +527,8 @@ pub struct AgentBuilder { skills: Vec, tools: Vec, max_tool_rounds: usize, + session_manager: Option>, + resumed_session: Option, } impl AgentBuilder { @@ -465,7 +556,17 @@ impl AgentBuilder { self } - pub fn build(self) -> Agent { + pub fn session_manager(mut self, manager: Arc) -> Self { + self.session_manager = Some(manager); + self + } + + pub fn resume_session(mut self, session: Session) -> Self { + self.resumed_session = Some(session); + self + } + + pub fn build(self) -> Result { let name = &self.model.info().name; let model_name = name .split_once('/') @@ -476,23 +577,51 @@ impl AgentBuilder { .as_nanos() .to_string(); - Agent { + let mut session_id = format!("{}_{}", model_name, timestamp); + let mut messages = Vec::new(); + let mut usage = Usage::default(); + let mut active_session = None; + + if let Some(session) = self.resumed_session { + if session.provider != self.model.info().provider.0 + || session.model != self.model.info().id + { + return Err(AgentError::Session(SessionError::InvalidHeader { + path: PathBuf::from(&session.id), + reason: format!( + "cannot resume session for model {} (provider {}) with bound model {} (provider {})", + session.model, + session.provider, + self.model.info().id, + self.model.info().provider.0 + ), + })); + } + session_id = session.id.clone(); + messages = session.messages.clone(); + usage = session.usage.clone(); + active_session = Some(session); + } + + let mut context = AgentContext::new(self.system_prompt, self.skills, self.tools); + context.hydrate(messages, usage); + + Ok(Agent { model: Mutex::new(self.model), - context: Mutex::new(AgentContext::new( - self.system_prompt, - self.skills, - self.tools, - )), + context: Mutex::new(context), plan_mode: AtomicBool::new(false), max_tool_rounds: self.max_tool_rounds, - session_id: format!("{}_{}", model_name, timestamp), - } + session_id: Mutex::new(session_id), + session_manager: self.session_manager, + active_session: Mutex::new(active_session), + }) } } #[cfg(test)] mod tests { use super::*; + use crate::session::{Session, SessionManager, SessionRecord}; use async_trait::async_trait; use llm::{ContentBlock, LlmApi, LlmError, LlmEvent, StopReason}; use providers::{ @@ -627,7 +756,10 @@ mod tests { #[tokio::test] async fn prompt_owns_history_and_system_prompt() { - let agent = Agent::builder(model()).system_prompt("Be helpful").build(); + let agent = Agent::builder(model()) + .system_prompt("Be helpful") + .build() + .unwrap(); let response = agent.prompt("hello").await.unwrap(); assert_eq!(response.text(), "echo: hello"); assert_eq!(agent.messages().await.len(), 2); @@ -635,7 +767,12 @@ mod tests { #[tokio::test] async fn prompt_stream_emits_text_deltas_and_finished() { - let agent = Arc::new(Agent::builder(model()).system_prompt("Be helpful").build()); + let agent = Arc::new( + Agent::builder(model()) + .system_prompt("Be helpful") + .build() + .unwrap(), + ); let mut rx = agent.prompt_stream("hello"); let mut events = Vec::new(); @@ -681,7 +818,11 @@ mod tests { #[tokio::test] async fn prompt_stream_emits_reasoning_and_text_deltas() { - let agent = Arc::new(Agent::builder(model_with_api(Arc::new(ReasoningApi))).build()); + let agent = Arc::new( + Agent::builder(model_with_api(Arc::new(ReasoningApi))) + .build() + .unwrap(), + ); let mut rx = agent.prompt_stream("hello"); let mut events = Vec::new(); @@ -714,7 +855,7 @@ mod tests { let api = Arc::new(FailAfterFirstApi { calls: AtomicUsize::new(0), }); - let agent = Arc::new(Agent::builder(model_with_api(api)).build()); + let agent = Arc::new(Agent::builder(model_with_api(api)).build().unwrap()); agent.prompt("first").await.unwrap(); @@ -748,7 +889,7 @@ mod tests { let api = Arc::new(PendingAfterFirstApi { calls: AtomicUsize::new(0), }); - let agent = Arc::new(Agent::builder(model_with_api(api)).build()); + let agent = Arc::new(Agent::builder(model_with_api(api)).build().unwrap()); agent.prompt("first").await.unwrap(); @@ -770,4 +911,395 @@ mod tests { matches!(&messages[3], AgentMessage::Assistant(response) if response.text() == "partial response") ); } + + struct ToolCallingApi { + calls: AtomicUsize, + } + + #[async_trait] + impl LlmApi for ToolCallingApi { + async fn stream(&self, request: llm::LlmRequest<'_>) -> Result { + let call_count = self.calls.fetch_add(1, Ordering::SeqCst); + if call_count == 0 { + // First round: one tool call, no text content. + let response = LlmResponse { + content: vec![], + stop_reason: StopReason::ToolUse, + usage: Some(llm::Usage { + input_tokens: 10, + output_tokens: 5, + ..llm::Usage::default() + }), + model: Some(request.model_id.to_owned()), + reasoning: None, + reasoning_details: Vec::new(), + }; + let tool_calls = vec![llm::ToolCall { + id: "call-1".into(), + name: "bash".into(), + arguments: serde_json::json!({"command": "echo hi"}).to_string(), + }]; + Ok(Box::pin(futures_util::stream::iter([ + Ok(LlmEvent::ToolCallDelta { + index: 0, + id: Some("call-1".into()), + name: Some("bash".into()), + arguments: serde_json::json!({"command": "echo hi"}).to_string(), + }), + Ok(LlmEvent::Done { + stop_reason: StopReason::ToolUse, + usage: response.usage.clone(), + model: response.model.clone(), + }), + ]))) + } else { + // Second round: final text response after tool result. + let response = LlmResponse { + content: vec![ContentBlock::Text("done".into())], + stop_reason: StopReason::Stop, + usage: Some(llm::Usage { + input_tokens: 20, + output_tokens: 3, + ..llm::Usage::default() + }), + model: Some(request.model_id.to_owned()), + reasoning: None, + reasoning_details: Vec::new(), + }; + let text = response.text(); + let model = response.model.clone(); + Ok(Box::pin(futures_util::stream::iter([ + Ok(LlmEvent::TextDelta { text }), + Ok(LlmEvent::Done { + stop_reason: StopReason::Stop, + usage: response.usage.clone(), + model, + }), + ]))) + } + } + } + + #[tokio::test] + async fn building_agent_with_manager_creates_no_file() { + let root = std::env::temp_dir().join(format!("alan-plan3-build-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let manager = Arc::new(SessionManager::new(&root)); + let agent = Agent::builder(model()) + .session_manager(manager) + .build() + .unwrap(); + + // No session file should exist before any prompt. + let entries: Vec<_> = std::fs::read_dir(&root) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + assert!( + entries.is_empty(), + "building must not create a session file" + ); + } + + #[tokio::test] + async fn first_buffered_prompt_creates_session_and_persists_messages() { + let root = + std::env::temp_dir().join(format!("alan-plan3-buffered-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let manager = Arc::new(SessionManager::new(&root)); + let agent = Agent::builder(model()) + .session_manager(manager.clone()) + .build() + .unwrap(); + + let response = agent.prompt("hello").await.unwrap(); + assert_eq!(response.text(), "echo: hello"); + + // One session file must exist under a pwd subdirectory. + let entries: Vec<_> = std::fs::read_dir(&root) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + assert_eq!(entries.len(), 1, "one pwd directory created"); + let pwd_dir = entries[0].path(); + let files: Vec<_> = std::fs::read_dir(&pwd_dir) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + assert_eq!(files.len(), 1, "one session file created"); + + // Load the session and verify messages. + let session_file = files[0].path(); + let content = std::fs::read_to_string(&session_file).unwrap(); + let lines: Vec<_> = content.lines().collect(); + assert_eq!(lines.len(), 3, "header + user + assistant"); + + let header: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert_eq!(header["type"], "session"); + + let user_record: serde_json::Value = serde_json::from_str(lines[1]).unwrap(); + assert_eq!(user_record["type"], "message"); + assert_eq!(user_record["message"]["kind"], "user"); + assert_eq!(user_record["message"]["content"], "hello"); + + let assistant_record: serde_json::Value = serde_json::from_str(lines[2]).unwrap(); + assert_eq!(assistant_record["type"], "message"); + assert_eq!(assistant_record["message"]["kind"], "assistant"); + } + + #[tokio::test] + async fn first_streaming_prompt_has_same_persistence_behavior() { + let root = + std::env::temp_dir().join(format!("alan-plan3-streaming-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let manager = Arc::new(SessionManager::new(&root)); + let agent = Arc::new( + Agent::builder(model()) + .session_manager(manager.clone()) + .build() + .unwrap(), + ); + + let mut rx = agent.prompt_stream("streaming hello"); + while let Some(event) = rx.recv().await { + let _ = event.unwrap(); + } + + let entries: Vec<_> = std::fs::read_dir(&root) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + assert_eq!(entries.len(), 1); + let pwd_dir = entries[0].path(); + let files: Vec<_> = std::fs::read_dir(&pwd_dir) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + assert_eq!(files.len(), 1); + + let content = std::fs::read_to_string(files[0].path()).unwrap(); + let lines: Vec<_> = content.lines().collect(); + assert_eq!(lines.len(), 3, "header + user + assistant for streaming"); + } + + #[tokio::test] + async fn provider_is_not_called_if_session_creation_fails() { + let _root = + std::env::temp_dir().join(format!("alan-plan3-fail-create-{}", uuid::Uuid::new_v4())); + // Use a non-existent, unwritable path so session creation fails. + let manager = Arc::new(SessionManager::new("/proc/self/mem/unwritable-dir")); + let agent = Agent::builder(model()) + .session_manager(manager) + .build() + .unwrap(); + + let result = agent.prompt("hello").await; + assert!(result.is_err(), "must fail when session creation fails"); + } + + #[tokio::test] + async fn tool_call_responses_and_results_are_persisted_in_order() { + let root = + std::env::temp_dir().join(format!("alan-plan3-tool-order-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let manager = Arc::new(SessionManager::new(&root)); + let api = Arc::new(ToolCallingApi { + calls: AtomicUsize::new(0), + }); + let agent = Agent::builder(model_with_api(api)) + .session_manager(manager.clone()) + .with_tools([AgentTool::new( + llm::ToolDefinition { + name: "bash".into(), + description: "Run a shell command".into(), + parameters: serde_json::json!({}), + }, + tools::BashExecutor, + )]) + .build() + .unwrap(); + + let response = agent.prompt("run echo hi").await.unwrap(); + assert_eq!(response.text(), "done"); + + let content = { + let entries: Vec<_> = std::fs::read_dir(&root) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + let pwd_dir = entries[0].path(); + let files: Vec<_> = std::fs::read_dir(&pwd_dir) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + std::fs::read_to_string(files[0].path()).unwrap() + }; + let lines: Vec<_> = content.lines().collect(); + // At minimum: header + user + assistant(tool_calls) + tool_result + assistant(final) + assert!( + lines.len() >= 5, + "expected at least 5 records, got {}", + lines.len() + ); + + let types: Vec<_> = lines + .iter() + .map(|l| { + let v: serde_json::Value = serde_json::from_str(l).unwrap(); + v["type"].as_str().unwrap().to_owned() + }) + .collect(); + assert_eq!(types[0], "session"); + assert_eq!(types[1], "message"); + assert_eq!(types[types.len() - 1], "message"); + // The last assistant message should be "done". + let last_assistant = &types[types.len() - 1]; + assert_eq!(last_assistant, "message"); + } + + #[tokio::test] + async fn aggregate_usage_from_multiple_rounds_is_persisted() { + let root = std::env::temp_dir().join(format!("alan-plan3-usage-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let manager = Arc::new(SessionManager::new(&root)); + let api = Arc::new(ToolCallingApi { + calls: AtomicUsize::new(0), + }); + let agent = Agent::builder(model_with_api(api)) + .session_manager(manager.clone()) + .with_tools([AgentTool::new( + llm::ToolDefinition { + name: "bash".into(), + description: "Run a shell command".into(), + parameters: serde_json::json!({}), + }, + tools::BashExecutor, + )]) + .build() + .unwrap(); + + let response = agent.prompt("run echo hi").await.unwrap(); + assert_eq!(response.text(), "done"); + + let entries: Vec<_> = std::fs::read_dir(&root) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + let pwd_dir = entries[0].path(); + let files: Vec<_> = std::fs::read_dir(&pwd_dir) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + let session_file = files[0].path(); + + // Load the session through the manager to verify usage. + let session = { + let content = std::fs::read_to_string(&session_file).unwrap(); + let lines: Vec<_> = content.lines().collect(); + let header_line = lines[0]; + let record = SessionRecord::parse(header_line).unwrap(); + let SessionRecord::Session { id, .. } = record else { + panic!("expected session header"); + }; + manager + .load(&std::env::current_dir().unwrap(), &id) + .await + .expect("load session for usage check") + }; + + // First round: 10 input + 5 output. Second round: 20 input + 3 output. + assert_eq!(session.usage.input_tokens, 30); + assert_eq!(session.usage.output_tokens, 8); + } + + #[tokio::test] + async fn resumed_agent_includes_restored_messages_in_first_request() { + let root = std::env::temp_dir().join(format!("alan-plan3-resume-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let manager = Arc::new(SessionManager::new(&root)); + + // Create a session, persist a user message, then load it back + // so the in-memory Session has the restored messages. + let session = manager + .create(&root, "openrouter", "test", None) + .await + .expect("create session"); + manager + .append_message(&session, &AgentMessage::user("restored message")) + .await + .expect("append message"); + let session = manager + .load(&root, &session.id) + .await + .expect("load session with restored message"); + + let model = model(); + let agent = Agent::builder(model) + .session_manager(manager.clone()) + .resume_session(session) + .build() + .expect("build with resume"); + + let response = agent.prompt("new message").await.unwrap(); + assert_eq!(response.text(), "echo: new message"); + + // The restored message must be in the agent's history. + let messages = agent.messages().await; + assert!( + messages.len() >= 2, + "must include restored user + new user + assistant response, got {} messages", + messages.len() + ); + assert!( + matches!(&messages[0], AgentMessage::User(text) if text == "restored message"), + "first message must be the restored message, got {:?}", + messages[0] + ); + } + + #[tokio::test] + async fn request_uses_persisted_session_id_and_cache_key() { + let root = + std::env::temp_dir().join(format!("alan-plan3-session-id-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&root).unwrap(); + let manager = Arc::new(SessionManager::new(&root)); + let agent = Agent::builder(model()) + .session_manager(manager.clone()) + .build() + .unwrap(); + + agent.prompt("check session id").await.unwrap(); + + let entries: Vec<_> = std::fs::read_dir(&root) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + let pwd_dir = entries[0].path(); + let files: Vec<_> = std::fs::read_dir(&pwd_dir) + .unwrap() + .filter_map(|e| e.ok()) + .collect(); + let content = std::fs::read_to_string(files[0].path()).unwrap(); + let lines: Vec<_> = content.lines().collect(); + + // The header must contain the session id. + let header: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + let session_id = header["id"].as_str().unwrap(); + + // The prompt_cache_key and session_id in the request options must + // match the persisted session id. We verify this indirectly by + // confirming the session file name matches the session id. + let session_file_name = files[0].path().file_stem().unwrap().to_owned(); + assert_eq!(session_file_name, session_id); + } + + #[tokio::test] + async fn existing_no_manager_tests_still_pass() { + // Re-run the original no-manager buffered prompt test to confirm + // backward compatibility is preserved. + let agent = Agent::builder(model()).build().unwrap(); + let response = agent.prompt("hello").await.unwrap(); + assert_eq!(response.text(), "echo: hello"); + assert_eq!(agent.messages().await.len(), 2); + } } diff --git a/crates/agent/src/context.rs b/crates/agent/src/context.rs index ce4a6fc..3f44e2e 100644 --- a/crates/agent/src/context.rs +++ b/crates/agent/src/context.rs @@ -154,7 +154,6 @@ impl AgentContext { /// Runtime-only state (tools, system prompt, skills, tool indexes) /// remains untouched; a resumed agent must rebuild it from the current /// application configuration. - // Unused until Plan 2 wires session resume into the agent loop. #[allow(dead_code)] pub fn hydrate(&mut self, messages: Vec, usage: Usage) { self.messages = messages; diff --git a/crates/agent/src/error.rs b/crates/agent/src/error.rs index bff5289..f2bf6c0 100644 --- a/crates/agent/src/error.rs +++ b/crates/agent/src/error.rs @@ -16,4 +16,6 @@ pub enum AgentError { EventStreamClosed, #[error(transparent)] Tool(#[from] ToolError), + #[error(transparent)] + Session(#[from] crate::session::SessionError), } diff --git a/crates/agent/src/session/mod.rs b/crates/agent/src/session/mod.rs index b680fb8..67a7aa0 100644 --- a/crates/agent/src/session/mod.rs +++ b/crates/agent/src/session/mod.rs @@ -7,3 +7,4 @@ mod store; pub use error::SessionError; pub use manager::SessionManager; pub use record::{SESSION_SCHEMA_VERSION, Session, SessionRecord, SessionSummary}; +pub use store::StoreError; diff --git a/crates/agent/src/session/record.rs b/crates/agent/src/session/record.rs index b61263d..df031cc 100644 --- a/crates/agent/src/session/record.rs +++ b/crates/agent/src/session/record.rs @@ -137,7 +137,7 @@ impl SessionRecord { /// Parse one complete JSONL line. Internally tagged enums reject unknown /// `type` values and missing fields, so malformed lines fail loudly /// instead of becoming empty values. - pub(super) fn parse(line: &str) -> Result { + pub fn parse(line: &str) -> Result { serde_json::from_str(line) } } diff --git a/crates/alan/src/core/controller.rs b/crates/alan/src/core/controller.rs index 716104c..1bf24bf 100644 --- a/crates/alan/src/core/controller.rs +++ b/crates/alan/src/core/controller.rs @@ -236,7 +236,7 @@ mod tests { .unwrap() .bind("test") .unwrap(); - Controller::new(Agent::builder(model).build()) + Controller::new(Agent::builder(model).build().unwrap()) } #[tokio::test] @@ -290,7 +290,7 @@ mod tests { .unwrap() .bind("test") .unwrap(); - let mut controller = Controller::new(Agent::builder(model).build()); + let mut controller = Controller::new(Agent::builder(model).build().unwrap()); controller.submit("hi".into()); tokio::time::sleep(Duration::from_millis(50)).await; diff --git a/crates/alan/src/main.rs b/crates/alan/src/main.rs index a1d1c77..4c7058d 100644 --- a/crates/alan/src/main.rs +++ b/crates/alan/src/main.rs @@ -80,7 +80,7 @@ async fn main() -> anyhow::Result<()> { let agent = Agent::builder(model) .system_prompt(ALAN_SYSTEM_PROMPT) .with_tools(default_tools()) - .build(); + .build()?; let mut app = Controller::with_runtime(agent, registry, credential_store); event_loop(&mut app).await diff --git a/crates/providers/src/model.rs b/crates/providers/src/model.rs index 01b22a3..f2a3337 100644 --- a/crates/providers/src/model.rs +++ b/crates/providers/src/model.rs @@ -51,6 +51,10 @@ impl Model { &self.info } + pub fn reasoning_effort(&self) -> Option { + self.reasoning_effort + } + fn tools<'a>(&'a self, local: &'a [ToolSpec]) -> Vec { local .iter() From d83c856f706361410008ccfea6784cf69e08b4f7 Mon Sep 17 00:00:00 2001 From: Revantark Date: Tue, 25 Aug 2026 10:11:40 +0530 Subject: [PATCH 3/4] wire session manager in the alan --- Cargo.lock | 11 + Cargo.toml | 1 + README.md | 15 +- crates/agent/Cargo.toml | 1 + crates/agent/src/agent.rs | 92 +++--- crates/agent/src/lib.rs | 4 +- crates/agent/src/session/manager.rs | 421 ++++++++++----------------- crates/agent/src/session/mod.rs | 2 +- crates/agent/src/session/record.rs | 22 +- crates/agent/src/session/store.rs | 428 ++++++++++++---------------- crates/alan/src/core/chat.rs | 52 ++++ crates/alan/src/core/controller.rs | 8 + crates/alan/src/main.rs | 49 +++- 13 files changed, 517 insertions(+), 589 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ab3e5b5..2b16a33 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13,6 +13,7 @@ name = "agent" version = "0.1.0" dependencies = [ "async-trait", + "fs2", "futures-util", "llm", "providers", @@ -673,6 +674,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs2" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "fs_extra" version = "1.3.0" diff --git a/Cargo.toml b/Cargo.toml index 77401bd..d98cc0d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,4 +17,5 @@ tokio = { version = "1.53", features = ["full"] } tracing-subscriber = { version = "0.3.23", features = ["env-filter"] } tracing = "0.1.44" tracing-appender = "0.2" +fs2 = "0.4" uuid = { version = "1", features = ["v4"] } diff --git a/README.md b/README.md index 21637b5..c427e9d 100644 --- a/README.md +++ b/README.md @@ -12,10 +12,14 @@ Alan is a minimal coding agent written in Rust. It runs in your terminal and use - Optional OpenRouter web search and web fetch tools - Tool-call execution with configurable round limits - Plan mode toggled with `/plan` or `Shift+Tab` -- Interactive provider login overlay -- File/folder path completion in the prompt (`@`) -- Configurable reasoning effort -- Session history persisted under `~/.alan/sessions` +- Session history is stored under `$ALAN_HOME/.alan/sessions` (or + `$HOME/.alan/sessions` when `ALAN_HOME` is unset). Each working directory + gets its own hashed subdirectory, containing append-only JSONL session files. + Sessions are created on the first non-empty prompt. +- To resume a session, set `ALAN_SESSION` to the session ID (the filename + without `.jsonl`) and start Alan from the same working directory. For + example: `ALAN_SESSION=018f... cargo run -p alan`. The configured model and + provider must match the stored session. ## Requirements @@ -60,7 +64,8 @@ All variables are optional: | Variable | Purpose | Default | |---|---|---| | `ALAN_MODEL` | Model id | `openai/gpt-4o-mini` | -| `ALAN_HOME` | Alan home directory | `$HOME` | +| `ALAN_HOME` | Alan home directory (Alan uses `$ALAN_HOME/.alan/`) | `$HOME` | +| `ALAN_SESSION` | Resume this session ID from the current working directory | unset | | `ALAN_REASONING_EFFORT` | `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max` | unset | | `ALAN_OPENROUTER_WEB_SEARCH` | Enable web search tool (`1`, `true`, `yes`, `on`) | off | | `ALAN_OPENROUTER_WEB_FETCH` | Enable web fetch tool (`1`, `true`, `yes`, `on`) | off | diff --git a/crates/agent/Cargo.toml b/crates/agent/Cargo.toml index dcc4d62..aff0091 100644 --- a/crates/agent/Cargo.toml +++ b/crates/agent/Cargo.toml @@ -14,6 +14,7 @@ thiserror = { workspace = true } tokio = { workspace = true } tools = { path = "../tools" } uuid = { workspace = true } +fs2 = { workspace = true } [dev-dependencies] futures-util = { workspace = true } diff --git a/crates/agent/src/agent.rs b/crates/agent/src/agent.rs index 7d754fb..f6d304b 100644 --- a/crates/agent/src/agent.rs +++ b/crates/agent/src/agent.rs @@ -14,7 +14,6 @@ use std::{ Arc, atomic::{AtomicBool, Ordering}, }, - time::{SystemTime, UNIX_EPOCH}, }; use tokio::sync::{ Mutex, @@ -114,9 +113,7 @@ impl Agent { self.ensure_session(&model).await?; - context.messages.push(user_msg); - self.persist_message(&context.messages.last().unwrap()) - .await?; + self.append_context_message(&mut context, user_msg).await?; let (_cancellation, mut cancellation_receiver) = watch::channel(false); let mut partial = String::new(); @@ -131,8 +128,8 @@ impl Agent { .await; if result.is_err() && !partial.is_empty() { let partial_msg = AgentMessage::Assistant(partial_response(&partial)); - context.messages.push(partial_msg.clone()); - self.persist_message(&partial_msg).await?; + self.append_context_message(&mut context, partial_msg) + .await?; } result } @@ -158,6 +155,14 @@ impl Agent { } } + pub async fn session_id(&self) -> Option { + self.active_session + .lock() + .await + .as_ref() + .map(|session| session.id.clone()) + } + pub async fn set_model(&self, model: Model) { *self.model.lock().await = model; } @@ -174,6 +179,10 @@ impl Agent { self.context.lock().await.messages.clone() } + pub async fn usage(&self) -> Usage { + self.context.lock().await.usage.clone() + } + async fn run_prompt( self: Arc, content: String, @@ -193,9 +202,7 @@ impl Agent { self.ensure_session(&model).await?; - context.messages.push(user_msg); - self.persist_message(&context.messages.last().unwrap()) - .await?; + self.append_context_message(&mut context, user_msg).await?; let mut partial = String::new(); let result = self @@ -209,8 +216,8 @@ impl Agent { .await; if result.is_err() && !partial.is_empty() { let partial_msg = AgentMessage::Assistant(partial_response(&partial)); - context.messages.push(partial_msg.clone()); - self.persist_message(&partial_msg).await?; + self.append_context_message(&mut context, partial_msg) + .await?; } result } @@ -245,6 +252,8 @@ impl Agent { } if calls.is_empty() { + let msg = AgentMessage::Assistant(response.clone()); + self.append_context_message(context, msg).await?; if let Some(events) = events { Self::send_event( events, @@ -255,15 +264,11 @@ impl Agent { ) .await?; } - let msg = AgentMessage::Assistant(response.clone()); - context.messages.push(msg.clone()); - self.persist_message(&msg).await?; return Ok(response); } let assistant_msg = AgentMessage::Assistant(response); - context.messages.push(assistant_msg.clone()); - self.persist_message(&assistant_msg).await?; + self.append_context_message(context, assistant_msg).await?; for call in calls { Self::check_cancelled(cancellation)?; @@ -296,8 +301,7 @@ impl Agent { tool_call_id: call_id.clone(), content: result.clone(), }; - context.messages.push(msg.clone()); - self.persist_message(&msg).await?; + self.append_context_message(context, msg).await?; if let Some(events) = events { Self::send_event( @@ -318,8 +322,7 @@ impl Agent { tool_call_id: call_id.clone(), content: error.clone(), }; - context.messages.push(msg.clone()); - self.persist_message(&msg).await?; + self.append_context_message(context, msg).await?; if let Some(events) = events { Self::send_event( @@ -478,10 +481,22 @@ impl Agent { Ok(()) } + async fn append_context_message( + &self, + context: &mut AgentContext, + message: AgentMessage, + ) -> Result<(), AgentError> { + self.persist_message(&message).await?; + context.messages.push(message); + Ok(()) + } + async fn persist_message(&self, message: &AgentMessage) -> Result<(), AgentError> { let active_session = self.active_session.lock().await; if let (Some(manager), Some(session)) = (&self.session_manager, &*active_session) { - manager.append_message(session, message).await?; + manager + .append_message(&session.id, &session.pwd, message) + .await?; } Ok(()) } @@ -489,7 +504,9 @@ impl Agent { async fn persist_usage(&self, usage: &Usage) -> Result<(), AgentError> { let active_session = self.active_session.lock().await; if let (Some(manager), Some(session)) = (&self.session_manager, &*active_session) { - manager.save_usage(session, usage).await?; + manager + .append_usage(&session.id, &session.pwd, usage) + .await?; } Ok(()) } @@ -567,17 +584,7 @@ impl AgentBuilder { } pub fn build(self) -> Result { - let name = &self.model.info().name; - let model_name = name - .split_once('/') - .map_or_else(|| name.clone(), |(_, model)| model.to_owned()); - let timestamp = SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("system clock is before Unix epoch") - .as_nanos() - .to_string(); - - let mut session_id = format!("{}_{}", model_name, timestamp); + let mut session_id = uuid::Uuid::new_v4().to_string(); let mut messages = Vec::new(); let mut usage = Usage::default(); let mut active_session = None; @@ -621,7 +628,7 @@ impl AgentBuilder { #[cfg(test)] mod tests { use super::*; - use crate::session::{Session, SessionManager, SessionRecord}; + use crate::session::{SessionManager, SessionRecord}; use async_trait::async_trait; use llm::{ContentBlock, LlmApi, LlmError, LlmEvent, StopReason}; use providers::{ @@ -934,11 +941,6 @@ mod tests { reasoning: None, reasoning_details: Vec::new(), }; - let tool_calls = vec![llm::ToolCall { - id: "call-1".into(), - name: "bash".into(), - arguments: serde_json::json!({"command": "echo hi"}).to_string(), - }]; Ok(Box::pin(futures_util::stream::iter([ Ok(LlmEvent::ToolCallDelta { index: 0, @@ -985,7 +987,7 @@ mod tests { let root = std::env::temp_dir().join(format!("alan-plan3-build-{}", uuid::Uuid::new_v4())); std::fs::create_dir_all(&root).unwrap(); let manager = Arc::new(SessionManager::new(&root)); - let agent = Agent::builder(model()) + let _agent = Agent::builder(model()) .session_manager(manager) .build() .unwrap(); @@ -1202,7 +1204,7 @@ mod tests { panic!("expected session header"); }; manager - .load(&std::env::current_dir().unwrap(), &id) + .get_session(&id, &std::env::current_dir().unwrap()) .await .expect("load session for usage check") }; @@ -1225,11 +1227,15 @@ mod tests { .await .expect("create session"); manager - .append_message(&session, &AgentMessage::user("restored message")) + .append_message( + &session.id, + &session.pwd, + &AgentMessage::user("restored message"), + ) .await .expect("append message"); let session = manager - .load(&root, &session.id) + .get_session(&session.id, &root) .await .expect("load session with restored message"); diff --git a/crates/agent/src/lib.rs b/crates/agent/src/lib.rs index 0510566..4f76856 100644 --- a/crates/agent/src/lib.rs +++ b/crates/agent/src/lib.rs @@ -8,8 +8,6 @@ mod tool; pub use agent::{Agent, AgentBuilder, AgentEvent, AgentStream}; pub use context::AgentMessage; pub use error::AgentError; -pub use session::{ - SESSION_SCHEMA_VERSION, Session, SessionError, SessionManager, SessionRecord, SessionSummary, -}; +pub use session::{SESSION_SCHEMA_VERSION, Session, SessionError, SessionManager, SessionRecord}; pub use skill::{Skill, build_system_prompt, format_skills_xml}; pub use tool::{AgentTool, default_tools}; diff --git a/crates/agent/src/session/manager.rs b/crates/agent/src/session/manager.rs index d32d5ca..ea45816 100644 --- a/crates/agent/src/session/manager.rs +++ b/crates/agent/src/session/manager.rs @@ -1,29 +1,32 @@ use super::error::SessionError; -use super::record::{SESSION_SCHEMA_VERSION, Session, SessionRecord, SessionSummary, now_ms}; +use super::record::{SESSION_SCHEMA_VERSION, Session, SessionRecord, now_ms}; use super::store::StoreError; -use super::store::{JsonlStore, split_complete_lines}; +use super::store::{JSONL_EXTENSION, JsonlStore, set_permissions, split_complete_lines}; use crate::context::AgentMessage; use crate::session::dir::pwd_key; use llm::{ReasoningEffort, Usage}; use std::path::{Path, PathBuf}; -/// Normalize a pwd, mapping I/O failure into the store error shape. fn normalize(pwd: PathBuf) -> Result { let dir = pwd.clone(); super::dir::normalize_pwd(pwd) .map_err(|source| SessionError::Store(StoreError::CreateDir { dir, source })) } -/// Filesystem-backed session store over [`JsonlStore`]. +/// Filesystem-backed, append-only sessions. pub struct SessionManager { - store: JsonlStore, + root: PathBuf, } impl SessionManager { pub fn new(root: impl Into) -> Self { - Self { - store: JsonlStore::new(root), - } + Self { root: root.into() } + } + + fn file_path(&self, key: &str, name: &str) -> PathBuf { + self.root + .join(key) + .join(format!("{name}.{JSONL_EXTENSION}")) } /// Create a new session and write its header record immediately. @@ -40,56 +43,72 @@ impl SessionManager { let pwd = normalize(pwd.into())?; let key = pwd_key(&pwd); let session = Session::new(pwd, provider, model, thinking_level); + validate_session_id(&session.id)?; + let path = self.file_path(&key, &session.id); + // `file_path` always builds `root/key/name.jsonl`, so parent is + // the key directory. + let dir = match path.parent() { + Some(dir) => dir.to_path_buf(), + None => unreachable!("file_path always builds root/key/name.jsonl"), + }; + + set_permissions(&self.root, true).await?; + tokio::fs::create_dir_all(&dir).await.map_err(|source| { + SessionError::Store(StoreError::CreateDir { + dir: dir.clone(), + source, + }) + })?; + set_permissions(&dir, true).await?; - // Serialize fully before writing; exclusive creation refuses to - // overwrite an existing session with the same id. let header = session .header_record() .to_jsonl() .map_err(|source| SessionError::Serialize { - path: self.store.file_path(&key, &session.id), + path: path.clone(), source, })?; - self.store.create_file(&key, &session.id, &header).await?; + JsonlStore::create(&path, &header).await?; Ok(session) } - /// Append exactly one message record to the session file. pub async fn append_message( &self, - session: &Session, + session_id: &str, + pwd: &Path, message: &AgentMessage, ) -> Result<(), SessionError> { let record = SessionRecord::Message { message: message.clone(), timestamp_ms: now_ms(), }; - self.append_record(session, &record).await + self.append_record(session_id, pwd, &record).await } - /// Append an aggregate usage snapshot. Loading replaces (never sums) - /// `Session.usage` with the newest snapshot. - pub async fn save_usage(&self, session: &Session, usage: &Usage) -> Result<(), SessionError> { + pub async fn append_usage( + &self, + session_id: &str, + pwd: &Path, + usage: &Usage, + ) -> Result<(), SessionError> { let record = SessionRecord::Usage { usage: usage.clone(), timestamp_ms: now_ms(), }; - self.append_record(session, &record).await + self.append_record(session_id, pwd, &record).await } - /// Reconstruct a complete session by replaying its records. - pub async fn load(&self, pwd: &Path, id: &str) -> Result { - validate_session_id(id)?; + pub async fn get_session(&self, session_id: &str, pwd: &Path) -> Result { + validate_session_id(session_id)?; let normalized = normalize(pwd.to_path_buf())?; - let key = pwd_key(&normalized); - let content = self.store.read(&key, id).await?; + let path = self.file_path(&pwd_key(&normalized), session_id); + let content = JsonlStore::read(&path).await?; let mut lines = split_complete_lines(&content); - let path = self.store.file_path(&key, id); - let header_line = + let first_line = lines .next() .map(|(line, _)| line) @@ -97,64 +116,14 @@ impl SessionManager { path: path.clone(), reason: "file is empty".into(), })?; - let record = - SessionRecord::parse(header_line).map_err(|err| SessionError::InvalidHeader { - path: path.clone(), - reason: err.to_string(), - })?; - let SessionRecord::Session { - id: header_id, - version, - pwd: header_pwd, - provider, - model, - thinking_level, - created_at_ms, - updated_at_ms, - } = record - else { - return Err(SessionError::InvalidHeader { - path: path.clone(), - reason: "first record is not a session header".into(), - }); - }; - - if version > SESSION_SCHEMA_VERSION { - return Err(SessionError::UnsupportedVersion { - version, - supported: SESSION_SCHEMA_VERSION, - path: path.clone(), - }); - } - if header_id != id { - return Err(SessionError::InvalidHeader { - path: path.clone(), - reason: format!("header id {header_id:?} does not match requested id {id:?}"), - }); - } - if normalize(header_pwd.clone())? != normalized { - return Err(SessionError::InvalidHeader { - path: path.clone(), - reason: format!( - "header pwd {:?} does not match requested pwd {:?}", - header_pwd.display(), - normalized.display() - ), - }); - } + let mut session = parse_header(first_line, &path, session_id, &normalized)?; - let mut session = Session { - id: header_id, - version, - pwd: header_pwd, - provider, - model, - thinking_level, - messages: Vec::new(), - usage: Usage::default(), - created_at_ms, - updated_at_ms: updated_at_ms.max(created_at_ms), - }; + let capacity = content + .bytes() + .filter(|b| *b == b'\n') + .count() + .saturating_sub(1); + session.messages.reserve(capacity); for (line, complete) in lines { if !complete { @@ -165,125 +134,34 @@ impl SessionManager { path: path.clone(), reason: err.to_string(), })?; - let timestamp_ms = record.timestamp_ms(); - match record { - SessionRecord::Session { .. } => { - return Err(SessionError::MalformedRecord { - path: path.clone(), - reason: "unexpected session header inside file".into(), - }); - } - SessionRecord::Message { message, .. } => session.messages.push(message), - SessionRecord::Usage { usage, .. } => session.usage = usage, + if matches!(&record, SessionRecord::Session { .. }) { + return Err(SessionError::MalformedRecord { + path: path.clone(), + reason: "unexpected session header inside file".into(), + }); } - session.updated_at_ms = session.updated_at_ms.max(timestamp_ms); + session.record(record); } Ok(session) } - /// Summaries for every session recorded under `pwd`, newest first. - /// - /// Only headers and usage snapshots are read; message bodies are skipped. - /// Unreadable or corrupt files are skipped rather than failing the whole - /// listing. - pub async fn list(&self, pwd: &Path) -> Result, SessionError> { - let normalized = normalize(pwd.to_path_buf())?; - let key = pwd_key(&normalized); - - let files = self.store.files(&key).await?; - let mut summaries: Vec = Vec::with_capacity(files.len()); - for path in files { - if let Ok(summary) = self.summarize(&path).await { - summaries.push(summary); - } - } - - summaries.sort_by_key(|summary| std::cmp::Reverse(summary.updated_at_ms)); - Ok(summaries) - } - - async fn summarize(&self, path: &Path) -> Result { - let content = tokio::fs::read_to_string(path).await.map_err(|source| { - SessionError::Store(StoreError::ReadFile { - path: path.to_path_buf(), - source, - }) - })?; - let mut lines = split_complete_lines(&content); - - let (header_line, _) = lines.next().ok_or_else(|| SessionError::InvalidHeader { - path: path.to_path_buf(), - reason: "file is empty".into(), - })?; - let record = - SessionRecord::parse(header_line).map_err(|err| SessionError::InvalidHeader { - path: path.to_path_buf(), - reason: err.to_string(), - })?; - let SessionRecord::Session { - id, - version, - pwd, - provider, - model, - thinking_level, - created_at_ms, - updated_at_ms, - } = record - else { - return Err(SessionError::InvalidHeader { - path: path.to_path_buf(), - reason: "first record is not a session header".into(), - }); - }; - if version > SESSION_SCHEMA_VERSION { - return Err(SessionError::UnsupportedVersion { - version, - supported: SESSION_SCHEMA_VERSION, - path: path.to_path_buf(), - }); - } - - let mut latest_usage = Usage::default(); - let mut max_updated_at_ms = updated_at_ms.max(created_at_ms); - for (line, complete) in lines { - if !complete { - continue; - } - if let Ok(record) = SessionRecord::parse(line) { - max_updated_at_ms = max_updated_at_ms.max(record.timestamp_ms()); - if let SessionRecord::Usage { usage, .. } = record { - latest_usage = usage; - } - } - } - - Ok(SessionSummary { - id, - pwd, - provider, - model, - thinking_level, - created_at_ms, - updated_at_ms: max_updated_at_ms, - usage: latest_usage, - }) - } - async fn append_record( &self, - session: &Session, + session_id: &str, + pwd: &Path, record: &SessionRecord, ) -> Result<(), SessionError> { - let key = pwd_key(&session.pwd); + validate_session_id(session_id)?; + let key = pwd_key(pwd); + let path = self.file_path(&key, session_id); let line = record .to_jsonl() .map_err(|source| SessionError::Serialize { - path: self.store.file_path(&key, &session.id), + path: path.clone(), source, })?; - Ok(self.store.append(&key, &session.id, &line).await?) + Ok(JsonlStore::append(&path, &line).await?) } } @@ -303,11 +181,79 @@ fn validate_session_id(id: &str) -> Result<(), SessionError> { } } +/// Parse the first JSONL line as a session header and validate it matches the +/// expected id and pwd. Returns a ready-to-use [`Session`] with an empty +/// messages vec. +fn parse_header( + line: &str, + path: &Path, + expected_id: &str, + normalized_pwd: &Path, +) -> Result { + let record = SessionRecord::parse(line).map_err(|err| SessionError::InvalidHeader { + path: path.to_path_buf(), + reason: err.to_string(), + })?; + + let SessionRecord::Session { + id, + version, + pwd, + provider, + model, + thinking_level, + created_at_ms, + updated_at_ms, + } = record + else { + return Err(SessionError::InvalidHeader { + path: path.to_path_buf(), + reason: "first record is not a session header".into(), + }); + }; + + if version > SESSION_SCHEMA_VERSION { + return Err(SessionError::UnsupportedVersion { + version, + supported: SESSION_SCHEMA_VERSION, + path: path.to_path_buf(), + }); + } + if id != expected_id { + return Err(SessionError::InvalidHeader { + path: path.to_path_buf(), + reason: format!("header id {id:?} does not match requested id {expected_id:?}"), + }); + } + if normalize(pwd.clone())? != normalized_pwd { + return Err(SessionError::InvalidHeader { + path: path.to_path_buf(), + reason: format!( + "header pwd {:?} does not match requested pwd {:?}", + pwd.display(), + normalized_pwd.display() + ), + }); + } + + Ok(Session { + id, + version, + pwd, + provider, + model, + thinking_level, + messages: Vec::new(), + usage: Usage::default(), + created_at_ms, + updated_at_ms: updated_at_ms.max(created_at_ms), + }) +} + #[cfg(test)] mod tests { use super::*; use crate::context::AgentMessage; - use crate::session::store::JSONL_EXTENSION; fn temp_root(name: &str) -> PathBuf { let dir = @@ -333,14 +279,6 @@ mod tests { .join(format!("{id}.{JSONL_EXTENSION}")) } - #[tokio::test] - async fn constructing_manager_creates_nothing() { - let root = - std::env::temp_dir().join(format!("alan-session-construct-{}", uuid::Uuid::new_v4())); - let _manager = SessionManager::new(&root); - assert!(!root.exists(), "manager construction must not create files"); - } - #[tokio::test] async fn create_writes_single_header_file_under_pwd_dir() { let root = temp_root("create"); @@ -399,24 +337,24 @@ mod tests { .expect("create"); manager - .append_message(&session, &AgentMessage::user("hello")) + .append_message(&session.id, &session.pwd, &AgentMessage::user("hello")) .await .expect("append user"); manager - .append_message(&session, &AgentMessage::user("second")) + .append_message(&session.id, &session.pwd, &AgentMessage::user("second")) .await .expect("append second"); manager - .save_usage(&session, &usage(10, 5)) + .append_usage(&session.id, &session.pwd, &usage(10, 5)) .await .expect("usage 1"); manager - .save_usage(&session, &usage(25, 8)) + .append_usage(&session.id, &session.pwd, &usage(25, 8)) .await .expect("usage 2"); let loaded = manager - .load(Path::new("/tmp/project"), &session.id) + .get_session(&session.id, Path::new("/tmp/project")) .await .expect("load"); @@ -445,16 +383,16 @@ mod tests { .expect("create"); manager - .save_usage(&session, &usage(10, 5)) + .append_usage(&session.id, &session.pwd, &usage(10, 5)) .await .expect("usage 1"); manager - .save_usage(&session, &usage(20, 7)) + .append_usage(&session.id, &session.pwd, &usage(20, 7)) .await .expect("usage 2"); let loaded = manager - .load(Path::new("/tmp/p"), &session.id) + .get_session(&session.id, Path::new("/tmp/p")) .await .expect("load"); assert_eq!(loaded.usage, usage(20, 7)); @@ -481,7 +419,7 @@ mod tests { std::fs::write(&path, &content).unwrap(); let loaded = manager - .load(Path::new("/tmp/p"), &session.id) + .get_session(&session.id, Path::new("/tmp/p")) .await .expect("truncated tail tolerated"); assert_eq!(loaded.messages.len(), 1); @@ -491,7 +429,7 @@ mod tests { std::fs::write(&path, &content).unwrap(); let err = manager - .load(Path::new("/tmp/p"), &session.id) + .get_session(&session.id, Path::new("/tmp/p")) .await .expect_err("malformed complete line rejected"); assert!(matches!(err, SessionError::MalformedRecord { .. })); @@ -508,13 +446,13 @@ mod tests { .expect("create"); let err = manager - .load(Path::new("/tmp/other"), &session.id) + .get_session(&session.id, Path::new("/tmp/other")) .await .expect_err("wrong pwd must not load"); assert!(matches!(err, SessionError::Store(StoreError::NotFound(_)))); let err = manager - .load(Path::new("/tmp/p"), "not-a-real-id") + .get_session("not-a-real-id", Path::new("/tmp/p")) .await .expect_err("wrong id must not load"); assert!(matches!(err, SessionError::Store(StoreError::NotFound(_)))); @@ -569,7 +507,7 @@ mod tests { ) .unwrap(); let err = manager - .load(Path::new(pwd), &session.id) + .get_session(&session.id, Path::new(pwd)) .await .expect_err("unsupported version rejected"); assert!(matches!(err, SessionError::UnsupportedVersion { .. })); @@ -580,7 +518,7 @@ mod tests { let content = std::fs::read_to_string(&path).unwrap(); std::fs::write(&path, content.replace(&session.id, "other-id")).unwrap(); let err = manager - .load(Path::new(pwd), &session.id) + .get_session(&session.id, Path::new(pwd)) .await .expect_err("mismatched header id rejected"); assert!(matches!(err, SessionError::InvalidHeader { .. })); @@ -591,66 +529,13 @@ mod tests { let content = std::fs::read_to_string(&path).unwrap(); std::fs::write(&path, content.replace(pwd, "/elsewhere")).unwrap(); let err = manager - .load(Path::new(pwd), &session.id) + .get_session(&session.id, Path::new(pwd)) .await .expect_err("mismatched header pwd rejected"); assert!(matches!(err, SessionError::InvalidHeader { .. })); cleanup(&root); } - #[tokio::test] - async fn list_returns_summaries_without_loading_messages() { - let root = temp_root("list"); - let manager = SessionManager::new(&root); - - let older = manager - .create("/tmp/p", "openrouter", "m", Some(ReasoningEffort::Low)) - .await - .expect("older"); - manager - .append_message(&older, &AgentMessage::user("hi")) - .await - .expect("append"); - manager - .save_usage(&older, &usage(3, 4)) - .await - .expect("usage"); - - let newer = manager - .create("/tmp/p", "openrouter", "m2", None) - .await - .expect("newer"); - - let summaries = manager.list(Path::new("/tmp/p")).await.expect("list"); - assert_eq!(summaries.len(), 2); - // Newest first; ids break ties deterministically for same-ms creates. - assert!( - (summaries[0].updated_at_ms, summaries[1].id.clone()) - >= (summaries[1].updated_at_ms, summaries[0].id.clone()) - ); - - let summary = summaries - .iter() - .find(|summary| summary.id == older.id) - .expect("older summary"); - assert_eq!(summary.pwd, older.pwd); - assert_eq!(summary.provider, "openrouter"); - assert_eq!(summary.model, "m"); - assert_eq!(summary.thinking_level, Some(ReasoningEffort::Low)); - assert_eq!(summary.created_at_ms, older.created_at_ms); - assert_eq!(summary.usage, usage(3, 4)); - - let _ = newer; - assert!( - manager - .list(Path::new("/tmp/none")) - .await - .expect("list") - .is_empty() - ); - cleanup(&root); - } - #[tokio::test] async fn append_to_missing_session_fails() { let root = temp_root("missing-append"); @@ -658,7 +543,7 @@ mod tests { let session = Session::new("/tmp/p", "o", "m", None); let err = manager - .append_message(&session, &AgentMessage::user("hi")) + .append_message(&session.id, &session.pwd, &AgentMessage::user("hi")) .await .expect_err("append without create fails"); assert!(matches!(err, SessionError::Store(StoreError::NotFound(_)))); @@ -693,15 +578,17 @@ mod tests { impl SessionManager { async fn write_session_file_for_test(&self, session: &Session) -> Result<(), SessionError> { let key = pwd_key(&session.pwd); + let path = self.file_path(&key, &session.id); let header = session .header_record() .to_jsonl() .map_err(|source| SessionError::Serialize { - path: self.store.file_path(&key, &session.id), + path: path.clone(), source, })?; - Ok(self.store.create_file(&key, &session.id, &header).await?) + set_permissions(&self.root, true).await?; + Ok(JsonlStore::create(&path, &header).await?) } } } diff --git a/crates/agent/src/session/mod.rs b/crates/agent/src/session/mod.rs index 67a7aa0..1141117 100644 --- a/crates/agent/src/session/mod.rs +++ b/crates/agent/src/session/mod.rs @@ -6,5 +6,5 @@ mod store; pub use error::SessionError; pub use manager::SessionManager; -pub use record::{SESSION_SCHEMA_VERSION, Session, SessionRecord, SessionSummary}; +pub use record::{SESSION_SCHEMA_VERSION, Session, SessionRecord}; pub use store::StoreError; diff --git a/crates/agent/src/session/record.rs b/crates/agent/src/session/record.rs index df031cc..4a72a32 100644 --- a/crates/agent/src/session/record.rs +++ b/crates/agent/src/session/record.rs @@ -35,6 +35,7 @@ impl Session { model: impl Into, thinking_level: Option, ) -> Self { + let now = now_ms(); Self { version: SESSION_SCHEMA_VERSION, id: uuid::Uuid::new_v4().to_string(), @@ -44,8 +45,8 @@ impl Session { thinking_level, messages: Vec::new(), usage: Usage::default(), - created_at_ms: now_ms(), - updated_at_ms: now_ms(), + created_at_ms: now, + updated_at_ms: now, } } @@ -65,29 +66,16 @@ impl Session { /// Apply one replayed record to this in-memory session. pub fn record(&mut self, record: SessionRecord) { + let timestamp_ms = record.timestamp_ms(); match record { SessionRecord::Session { .. } => {} SessionRecord::Message { message, .. } => self.messages.push(message), SessionRecord::Usage { usage, .. } => self.usage = usage, } - self.updated_at_ms = self.updated_at_ms.max(now_ms()); + self.updated_at_ms = self.updated_at_ms.max(timestamp_ms); } } -/// Lightweight session description built from the header record and replayed -/// usage snapshots; message bodies are not loaded. -#[derive(Debug, Clone, PartialEq)] -pub struct SessionSummary { - pub id: String, - pub pwd: PathBuf, - pub provider: String, - pub model: String, - pub thinking_level: Option, - pub created_at_ms: u64, - pub updated_at_ms: u64, - pub usage: Usage, -} - /// One JSONL record in a session file. /// /// ```jsonl diff --git a/crates/agent/src/session/store.rs b/crates/agent/src/session/store.rs index 05e7ffa..a678a38 100644 --- a/crates/agent/src/session/store.rs +++ b/crates/agent/src/session/store.rs @@ -1,7 +1,8 @@ +use std::borrow::Cow; +use std::io::{ErrorKind, Write}; use std::path::{Path, PathBuf}; use thiserror::Error; -use tokio::fs::{self, File, OpenOptions}; -use tokio::io::AsyncWriteExt; +use tokio::fs::{self, File}; /// Errors from the raw JSONL store. #[derive(Debug, Error)] @@ -38,157 +39,136 @@ pub enum StoreError { #[source] source: std::io::Error, }, + #[error("failed to read directory {}: {source}", path.display())] + ReadDir { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("failed to set permissions on {}: {source}", path.display())] + SetPermissions { + path: PathBuf, + #[source] + source: std::io::Error, + }, #[error("file not found: {0}")] NotFound(PathBuf), #[error("file is locked by another writer: {0}")] Locked(PathBuf), } -/// Append-only JSONL store over a two-level layout: -/// `//.jsonl`. -pub(crate) struct JsonlStore { - root: PathBuf, -} +/// Stateless append-only JSONL store. +/// +/// Each method operates on a single file identified by an explicit path. +/// The struct carries no state — it exists to group the three core +/// operations and keep them behind a single, testable type. +pub(crate) struct JsonlStore; impl JsonlStore { - pub(crate) fn new(root: impl Into) -> Self { - Self { root: root.into() } - } - - /// Exclusively create `//.jsonl`, creating the key - /// directory if needed, and write `first_line` as its first record. + /// Exclusively create the file at `file_path` and write `first_line` + /// as its first record. /// - /// Returns [`StoreError::AlreadyExists`] if the file already exists; the + /// The parent directory must already exist. Returns + /// [`StoreError::AlreadyExists`] if the file already exists; the /// existing file is never touched. - pub(crate) async fn create_file( - &self, - key: &str, - name: &str, - first_line: &str, - ) -> Result<(), StoreError> { - let dir = self.key_dir(key); - fs::create_dir_all(&dir) - .await - .map_err(|source| StoreError::CreateDir { - dir: dir.clone(), - source, - })?; - set_permissions(&dir, true).await; - - let path = self.file_path(key, name); + pub(crate) async fn create(file_path: &Path, first_line: &str) -> Result<(), StoreError> { // `create_new` guarantees an existing file is never overwritten. - let mut file = File::create_new(&path).await.map_err(|source| { - if source.kind() == std::io::ErrorKind::AlreadyExists { - StoreError::AlreadyExists(path.clone()) - } else { - StoreError::CreateFile { - path: path.clone(), - source, - } - } - })?; - set_permissions(&path, false).await; - - Self::write_line(&path, &mut file, first_line).await - } - - /// Append one newline-terminated record under an exclusive sidecar lock. - pub(crate) async fn append(&self, key: &str, name: &str, line: &str) -> Result<(), StoreError> { - let path = self.file_path(key, name); - if !path.is_file() { - return Err(StoreError::NotFound(path)); + drop( + File::create_new(file_path) + .await + .map_err(|source| { + if source.kind() == ErrorKind::AlreadyExists { + StoreError::AlreadyExists(file_path.to_path_buf()) + } else { + StoreError::CreateFile { + path: file_path.to_path_buf(), + source, + } + } + })?, + ); + if let Err(error) = set_permissions(file_path, false).await { + remove_created_file(file_path).await; + return Err(error); } - - let _lock = FileLock::acquire(self.lock_path(key, name)).await?; - - let mut file = OpenOptions::new() - .append(true) - .open(&path) - .await - .map_err(|source| StoreError::OpenFile { - path: path.clone(), - source, - })?; - Self::write_line(&path, &mut file, line).await + if let Err(error) = Self::append(file_path, first_line).await { + remove_created_file(file_path).await; + return Err(error); + } + Ok(()) } - /// Read the whole file as text. - pub(crate) async fn read(&self, key: &str, name: &str) -> Result { - let path = self.file_path(key, name); - if !path.is_file() { - return Err(StoreError::NotFound(path)); - } - fs::read_to_string(&path) + /// Append one newline-terminated record under an exclusive advisory lock. + /// + /// The lock is held on the session file itself. This avoids stale sidecar + /// lock files and lets the operating system release the lock if a writer + /// exits unexpectedly. + pub(crate) async fn append(file_path: &Path, line: &str) -> Result<(), StoreError> { + let path = file_path.to_path_buf(); + let line = line.to_owned(); + tokio::task::spawn_blocking(move || append_locked(&path, &line)) .await - .map_err(|source| StoreError::ReadFile { path, source }) + .map_err(|source| StoreError::OpenFile { + path: file_path.to_path_buf(), + source: std::io::Error::other(source), + })? } - /// Paths of every `.jsonl` file stored under `key`, unsorted. - /// - /// Missing directories yield an empty list: no files yet. - pub(crate) async fn files(&self, key: &str) -> Result, StoreError> { - let dir = self.key_dir(key); - let read_dir = match fs::read_dir(&dir).await { - Ok(entries) => entries, - Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), - Err(source) => return Err(StoreError::ReadFile { path: dir, source }), - }; - - let mut paths = Vec::new(); - let mut entries = read_dir; - loop { - match entries.next_entry().await { - Ok(Some(entry)) => { - let path = entry.path(); - if path.extension().and_then(|ext| ext.to_str()) == Some(JSONL_EXTENSION) { - paths.push(path); - } - } - Ok(None) => break, - Err(source) => { - return Err(StoreError::ReadFile { - path: dir.clone(), - source, - }); + /// Read the whole file at `file_path` as text. + pub(crate) async fn read(file_path: &Path) -> Result { + fs::read_to_string(file_path).await.map_err(|source| { + if source.kind() == ErrorKind::NotFound { + StoreError::NotFound(file_path.to_path_buf()) + } else { + StoreError::ReadFile { + path: file_path.to_path_buf(), + source, } } - } - Ok(paths) + }) } - /// Directory holding all files for `key`. - pub(crate) fn key_dir(&self, key: &str) -> PathBuf { - self.root.join(key) - } +} - /// Full path of one stored file. - pub(crate) fn file_path(&self, key: &str, name: &str) -> PathBuf { - self.key_dir(key).join(format!("{name}.{JSONL_EXTENSION}")) +fn line_with_newline(line: &str) -> Cow<'_, str> { + if line.ends_with('\n') { + Cow::Borrowed(line) + } else { + Cow::Owned(format!("{line}\n")) } +} - /// Sidecar lock path used by [`JsonlStore::append`]. - pub(crate) fn lock_path(&self, key: &str, name: &str) -> PathBuf { - self.key_dir(key) - .join(format!("{name}.{JSONL_EXTENSION}{LOCK_SUFFIX}")) - } +async fn remove_created_file(path: &Path) { + let _ = fs::remove_file(path).await; +} - async fn write_line(path: &Path, file: &mut File, line: &str) -> Result<(), StoreError> { - file.write_all(line.as_bytes()) +pub(crate) const JSONL_EXTENSION: &str = "jsonl"; + +/// Set restrictive Unix permissions on `path`. +/// +/// Directories get `0o700`, files get `0o600`. On non-Unix platforms this is +/// a no-op. +pub(crate) async fn set_permissions(path: &Path, is_dir: bool) -> Result<(), StoreError> { + #[cfg(unix)] + { + use std::fs::Permissions; + use std::os::unix::fs::PermissionsExt; + + let mode = if is_dir { 0o700 } else { 0o600 }; + fs::set_permissions(path, Permissions::from_mode(mode)) .await - .map_err(|source| StoreError::WriteFile { + .map_err(|source| StoreError::SetPermissions { path: path.to_path_buf(), source, - })?; - file.flush().await.map_err(|source| StoreError::WriteFile { - path: path.to_path_buf(), - source, - }) + }) + } + #[cfg(not(unix))] + { + let _ = (path, is_dir); + Ok(()) } } -pub(crate) const JSONL_EXTENSION: &str = "jsonl"; -const LOCK_SUFFIX: &str = ".lock"; - /// Yield `(line, is_complete)` pairs from raw file contents. /// /// An incomplete final line (no trailing newline, i.e. a crash-truncated @@ -202,7 +182,6 @@ pub(crate) fn split_complete_lines(content: &str) -> impl Iterator (body, None), None => match content.rsplit_once('\n') { Some((body, tail)) => (body, Some(tail)), - // No newline at all: single truncated line. None => ("", Some(content)), }, } @@ -213,42 +192,57 @@ pub(crate) fn split_complete_lines(content: &str) -> impl Iterator Result { - File::create_new(&path).await.map_err(|source| { - if source.kind() == std::io::ErrorKind::AlreadyExists { - StoreError::Locked(path.clone()) +fn append_locked(path: &Path, line: &str) -> Result<(), StoreError> { + let mut file = std::fs::OpenOptions::new() + .read(true) + .append(true) + .open(path) + .map_err(|source| { + if source.kind() == ErrorKind::NotFound { + StoreError::NotFound(path.to_path_buf()) } else { - StoreError::CreateFile { - path: path.clone(), + StoreError::OpenFile { + path: path.to_path_buf(), source, } } })?; - Ok(Self { path }) - } -} -impl Drop for FileLock { - fn drop(&mut self) { - // Use blocking remove_file in Drop; this is fine because Drop is - // synchronous and file removal is a fast operation. - let _ = std::fs::remove_file(&self.path); - } -} + fs2::FileExt::try_lock_exclusive(&file).map_err(|source| { + if source.kind() == ErrorKind::WouldBlock + || source.raw_os_error() == fs2::lock_contended_error().raw_os_error() + { + StoreError::Locked(path.to_path_buf()) + } else { + StoreError::OpenFile { + path: path.to_path_buf(), + source, + } + } + })?; -#[cfg(unix)] -async fn set_permissions(path: &Path, is_dir: bool) { - use std::fs::Permissions; - use std::os::unix::fs::PermissionsExt; + let result = write_and_sync(&mut file, path, line); + let _ = fs2::FileExt::unlock(&file); + result +} - let mode = if is_dir { 0o700 } else { 0o600 }; - let _ = fs::set_permissions(path, Permissions::from_mode(mode)).await; +fn write_and_sync(file: &mut std::fs::File, path: &Path, line: &str) -> Result<(), StoreError> { + let line = line_with_newline(line); + file.write_all(line.as_bytes()) + .map_err(|source| StoreError::WriteFile { + path: path.to_path_buf(), + source, + })?; + file.flush() + .map_err(|source| StoreError::WriteFile { + path: path.to_path_buf(), + source, + })?; + file.sync_data() + .map_err(|source| StoreError::WriteFile { + path: path.to_path_buf(), + source, + }) } #[cfg(test)] @@ -265,33 +259,26 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } - #[tokio::test] - async fn constructing_store_creates_nothing() { - let root = - std::env::temp_dir().join(format!("alan-store-construct-{}", uuid::Uuid::new_v4())); - let _store = JsonlStore::new(&root); - assert!(!root.exists(), "store construction must not create files"); + fn file(root: &Path, key: &str, name: &str) -> PathBuf { + let dir = root.join(key); + std::fs::create_dir_all(&dir).expect("create key dir"); + dir.join(format!("{name}.{JSONL_EXTENSION}")) } #[tokio::test] - async fn create_file_writes_first_line_and_refuses_overwrite() { + async fn create_writes_first_line_and_refuses_overwrite() { let root = temp_root("create"); - let store = JsonlStore::new(&root); + let path = file(&root, "key-a", "one"); - store - .create_file("key-a", "one", "{\"first\":true}\n") + JsonlStore::create(&path, "{\"first\":true}\n") .await .expect("create"); - - let path = store.file_path("key-a", "one"); assert_eq!( std::fs::read_to_string(&path).unwrap(), "{\"first\":true}\n" ); - // Exclusive creation refuses to truncate or touch the existing file. - let err = store - .create_file("key-a", "one", "{\"second\":true}\n") + let err = JsonlStore::create(&path, "{\"second\":true}\n") .await .expect_err("second create refused"); assert!(matches!(err, StoreError::AlreadyExists(_))); @@ -302,29 +289,15 @@ mod tests { cleanup(&root); } - #[tokio::test] - async fn create_file_creates_key_directory() { - let root = temp_root("mkdir"); - let store = JsonlStore::new(&root); - - store - .create_file("key-x", "f", "x\n") - .await - .expect("create"); - assert!(store.key_dir("key-x").is_dir()); - cleanup(&root); - } - #[tokio::test] async fn append_produces_newline_terminated_records_in_order() { let root = temp_root("append"); - let store = JsonlStore::new(&root); - store.create_file("k", "f", "a\n").await.expect("create"); - - store.append("k", "f", "b\n").await.expect("append b"); - store.append("k", "f", "c\n").await.expect("append c"); + let path = file(&root, "k", "f"); - assert_eq!(store.read("k", "f").await.unwrap(), "a\nb\nc\n"); + JsonlStore::create(&path, "a\n").await.expect("create"); + JsonlStore::append(&path, "b\n").await.expect("append b"); + JsonlStore::append(&path, "c\n").await.expect("append c"); + assert_eq!(JsonlStore::read(&path).await.unwrap(), "a\nb\nc\n"); cleanup(&root); } @@ -332,13 +305,10 @@ mod tests { fn split_lines_flags_truncated_tail_only() { let complete: Vec<_> = split_complete_lines("a\nb\n").collect(); assert_eq!(complete, vec![("a", true), ("b", true)]); - let truncated: Vec<_> = split_complete_lines("a\nb").collect(); assert_eq!(truncated, vec![("a", true), ("b", false)]); - let none: Vec<_> = split_complete_lines("").collect(); assert!(none.is_empty()); - let only_truncated: Vec<_> = split_complete_lines("abc").collect(); assert_eq!(only_truncated, vec![("abc", false)]); } @@ -346,95 +316,57 @@ mod tests { #[tokio::test] async fn read_missing_and_append_missing_are_not_found() { let root = temp_root("missing"); - let store = JsonlStore::new(&root); + let path = file(&root, "nope", "f"); - let err = store.read("nope", "f").await.expect_err("read missing"); + let err = JsonlStore::read(&path).await.expect_err("read missing"); assert!(matches!(err, StoreError::NotFound(_))); - - let err = store - .append("nope", "f", "x\n") + let err = JsonlStore::append(&path, "x\n") .await .expect_err("append missing"); assert!(matches!(err, StoreError::NotFound(_))); cleanup(&root); } - #[tokio::test] - async fn files_lists_only_jsonl_entries() { - let root = temp_root("files"); - let store = JsonlStore::new(&root); - store - .create_file("k", "one", "1\n") - .await - .expect("create one"); - store - .create_file("k", "two", "2\n") - .await - .expect("create two"); - - let mut names: Vec<_> = store - .files("k") - .await - .expect("files") - .iter() - .map(|path| path.file_name().unwrap().to_string_lossy().into_owned()) - .collect(); - names.sort(); - assert_eq!(names, vec!["one.jsonl", "two.jsonl"]); - - // Lock sidecars are not listed, and unknown keys list empty. - tokio::fs::write(store.lock_path("k", "one"), "") - .await - .unwrap(); - assert_eq!(store.files("k").await.unwrap().len(), 2); - assert!(store.files("absent").await.unwrap().is_empty()); - cleanup(&root); - } - #[cfg(unix)] #[tokio::test] - async fn unix_permissions_are_restrictive() { + async fn unix_file_permissions_are_restrictive() { use std::os::unix::fs::PermissionsExt; - let root = temp_root("perms"); - let store = JsonlStore::new(&root); - store.create_file("k", "f", "x\n").await.expect("create"); + let path = file(&root, "k", "f"); - let dir_mode = std::fs::metadata(store.key_dir("k")) + JsonlStore::create(&path, "x\n").await.expect("create"); + let file_mode = std::fs::metadata(&path) .unwrap() .permissions() .mode(); - let file_mode = std::fs::metadata(store.file_path("k", "f")) - .unwrap() - .permissions() - .mode(); - assert_eq!(dir_mode & 0o777, 0o700, "directory mode"); assert_eq!(file_mode & 0o777, 0o600, "file mode"); cleanup(&root); } #[tokio::test] - async fn locked_while_sidecar_lock_held() { + async fn locked_while_session_file_is_held() { let root = temp_root("locked"); - let store = JsonlStore::new(&root); - store.create_file("k", "f", "x\n").await.expect("create"); + let path = file(&root, "k", "f"); - let lock = FileLock::acquire(store.lock_path("k", "f")) - .await - .expect("acquire lock"); - let err = store - .append("k", "f", "y\n") + JsonlStore::create(&path, "x\n").await.expect("create"); + + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .expect("open session"); + fs2::FileExt::try_lock_exclusive(&file).expect("acquire lock"); + let err = JsonlStore::append(&path, "y\n") .await .expect_err("append blocked"); assert!(matches!(err, StoreError::Locked(_))); - drop(lock); // Lock removed on drop. - store - .append("k", "f", "y\n") + fs2::FileExt::unlock(&file).expect("unlock session"); + drop(file); + JsonlStore::append(&path, "y\n") .await .expect("append after unlock"); - assert_eq!(store.read("k", "f").await.unwrap(), "x\ny\n"); - assert!(!store.lock_path("k", "f").exists(), "lock removed on drop"); + assert_eq!(JsonlStore::read(&path).await.unwrap(), "x\ny\n"); cleanup(&root); } } diff --git a/crates/alan/src/core/chat.rs b/crates/alan/src/core/chat.rs index a26f38f..a89d925 100644 --- a/crates/alan/src/core/chat.rs +++ b/crates/alan/src/core/chat.rs @@ -55,6 +55,58 @@ impl ChatController { } } + pub async fn session_id(&self) -> Option { + self.agent.session_id().await + } + + pub async fn restore_session_history(&mut self) { + let messages = self.agent.messages().await; + self.usage = self.agent.usage().await; + self.entries.clear(); + + for message in messages { + match message { + agent::AgentMessage::User(text) => self.entries.push(Entry::Prompt(text)), + agent::AgentMessage::Assistant(response) => { + if let Some(reasoning) = response.reasoning.as_deref() + && !reasoning.is_empty() + { + self.entries.push(Entry::Reasoning(reasoning.to_owned())); + } + for call in response.tool_calls() { + self.entries.push(Entry::ToolCall { + id: call.id.clone(), + name: call.name.clone(), + arguments: call.arguments.clone(), + output: String::new(), + status: ToolStatus::Completed, + }); + } + let text = response.text(); + if !text.is_empty() { + self.entries.push(Entry::Response(text)); + } + } + agent::AgentMessage::ToolResult { + tool_call_id, + content, + } => { + if let Some(Entry::ToolCall { output, .. }) = self + .entries + .iter_mut() + .rev() + .find(|entry| { + matches!(entry, Entry::ToolCall { id, .. } if id == &tool_call_id) + }) + { + *output = content; + } + } + } + } + self.revision = self.revision.wrapping_add(1); + } + pub fn entries(&self) -> &[Entry] { &self.entries } diff --git a/crates/alan/src/core/controller.rs b/crates/alan/src/core/controller.rs index 1bf24bf..b803de5 100644 --- a/crates/alan/src/core/controller.rs +++ b/crates/alan/src/core/controller.rs @@ -90,6 +90,14 @@ impl Controller { self.chat.usage() } + pub async fn restore_session_history(&mut self) { + self.chat.restore_session_history().await; + } + + pub async fn session_id(&self) -> Option { + self.chat.session_id().await + } + pub fn login_state(&self) -> &LoginState { self.login.state() } diff --git a/crates/alan/src/main.rs b/crates/alan/src/main.rs index 4c7058d..abf1a27 100644 --- a/crates/alan/src/main.rs +++ b/crates/alan/src/main.rs @@ -14,7 +14,7 @@ use llm::ServerTool; use std::io::stdout; use std::time::Duration; -use agent::{Agent, default_tools}; +use agent::{Agent, SessionManager, default_tools}; use futures_util::StreamExt; use llm::ReasoningEffort; use providers::{ @@ -77,13 +77,33 @@ async fn main() -> anyhow::Result<()> { }, )?; let registry = ProviderRegistry::new([Arc::new(provider) as Arc]); - let agent = Agent::builder(model) + let session_manager = Arc::new(SessionManager::new(sessions_path()?)); + let resumed_session = if let Some(session_id) = configured_session_id()? { + let cwd = std::env::current_dir()?; + Some(session_manager.get_session(&session_id, &cwd).await?) + } else { + None + }; + + let was_resumed = resumed_session.is_some(); + let mut agent_builder = Agent::builder(model) .system_prompt(ALAN_SYSTEM_PROMPT) .with_tools(default_tools()) - .build()?; + .session_manager(session_manager); + if let Some(session) = resumed_session { + agent_builder = agent_builder.resume_session(session); + } + let agent = agent_builder.build()?; let mut app = Controller::with_runtime(agent, registry, credential_store); - event_loop(&mut app).await + if was_resumed { + app.restore_session_history().await; + } + let result = event_loop(&mut app).await; + if let Some(session_id) = app.session_id().await { + println!("\nSession saved. Resume it with:\n\nALAN_SESSION={session_id} alan"); + } + result } fn enabled_server_tools(provider: &OpenRouterProvider) -> anyhow::Result> { @@ -133,10 +153,29 @@ fn configured_reasoning_effort() -> anyhow::Result> { } } fn auth_path() -> anyhow::Result { + Ok(alan_data_dir()?.join("auth.json")) +} + +fn sessions_path() -> anyhow::Result { + Ok(alan_data_dir()?.join("sessions")) +} + +fn alan_data_dir() -> anyhow::Result { let home = std::env::var_os("ALAN_HOME") .or_else(|| std::env::var_os("HOME")) .ok_or_else(|| anyhow::anyhow!("cannot determine Alan home directory"))?; - Ok(PathBuf::from(home).join(".alan").join("auth.json")) + Ok(PathBuf::from(home).join(".alan")) +} + +fn configured_session_id() -> anyhow::Result> { + let Some(id) = std::env::var_os("ALAN_SESSION") else { + return Ok(None); + }; + let id = id.to_string_lossy().trim().to_owned(); + if id.is_empty() { + return Err(anyhow::anyhow!("ALAN_SESSION must not be empty")); + } + Ok(Some(id)) } async fn event_loop(app: &mut Controller) -> anyhow::Result<()> { From 0b68e03fc1b343e52f84213b36febbff3e50f565 Mon Sep 17 00:00:00 2001 From: Revantark Date: Tue, 25 Aug 2026 10:49:13 +0530 Subject: [PATCH 4/4] fix fmt --- crates/agent/src/session/store.rs | 48 ++++++++++++------------------- 1 file changed, 19 insertions(+), 29 deletions(-) diff --git a/crates/agent/src/session/store.rs b/crates/agent/src/session/store.rs index a678a38..1734534 100644 --- a/crates/agent/src/session/store.rs +++ b/crates/agent/src/session/store.rs @@ -73,20 +73,16 @@ impl JsonlStore { /// existing file is never touched. pub(crate) async fn create(file_path: &Path, first_line: &str) -> Result<(), StoreError> { // `create_new` guarantees an existing file is never overwritten. - drop( - File::create_new(file_path) - .await - .map_err(|source| { - if source.kind() == ErrorKind::AlreadyExists { - StoreError::AlreadyExists(file_path.to_path_buf()) - } else { - StoreError::CreateFile { - path: file_path.to_path_buf(), - source, - } - } - })?, - ); + drop(File::create_new(file_path).await.map_err(|source| { + if source.kind() == ErrorKind::AlreadyExists { + StoreError::AlreadyExists(file_path.to_path_buf()) + } else { + StoreError::CreateFile { + path: file_path.to_path_buf(), + source, + } + } + })?); if let Err(error) = set_permissions(file_path, false).await { remove_created_file(file_path).await; return Err(error); @@ -127,7 +123,6 @@ impl JsonlStore { } }) } - } fn line_with_newline(line: &str) -> Cow<'_, str> { @@ -233,16 +228,14 @@ fn write_and_sync(file: &mut std::fs::File, path: &Path, line: &str) -> Result<( path: path.to_path_buf(), source, })?; - file.flush() - .map_err(|source| StoreError::WriteFile { - path: path.to_path_buf(), - source, - })?; - file.sync_data() - .map_err(|source| StoreError::WriteFile { - path: path.to_path_buf(), - source, - }) + file.flush().map_err(|source| StoreError::WriteFile { + path: path.to_path_buf(), + source, + })?; + file.sync_data().map_err(|source| StoreError::WriteFile { + path: path.to_path_buf(), + source, + }) } #[cfg(test)] @@ -335,10 +328,7 @@ mod tests { let path = file(&root, "k", "f"); JsonlStore::create(&path, "x\n").await.expect("create"); - let file_mode = std::fs::metadata(&path) - .unwrap() - .permissions() - .mode(); + let file_mode = std::fs::metadata(&path).unwrap().permissions().mode(); assert_eq!(file_mode & 0o777, 0o600, "file mode"); cleanup(&root); }