diff --git a/crates/core/src/agent.rs b/crates/core/src/agent.rs index 95a399b3..209f7bef 100644 --- a/crates/core/src/agent.rs +++ b/crates/core/src/agent.rs @@ -10038,6 +10038,9 @@ mod tests { assert!(sessions::save_messages(&core, &id, &messages, &mut persisted, true)); } + // Released first, so the hydration below claims the session over the + // damaged index: a claim that saw it indexed saves without asking. + sessions::detach(&core, &id).unwrap(); let index = core.data_dir.join("sessions").join("index.json"); let intact = std::fs::read(&index).unwrap(); std::fs::write(&index, "{ not json").unwrap(); @@ -10118,6 +10121,9 @@ mod tests { sessions::record_resume_point(&core, &id, 3); sessions::record_resume_point(&core, &id, 4); + // Released first, so the hydration below claims the session over the + // damaged index: a claim that saw it indexed saves without asking. + sessions::detach(&core, &id).unwrap(); let index = core.data_dir.join("sessions").join("index.json"); let intact = std::fs::read(&index).unwrap(); std::fs::write(&index, "{ not json").unwrap(); @@ -10192,6 +10198,9 @@ mod tests { // This sitting begins after everything on file. sessions::record_resume_point(&core, &id, messages.len() as u64); + // Released first, so the hydration below claims the session over the + // damaged index: a claim that saw it indexed saves without asking. + sessions::detach(&core, &id).unwrap(); let index = core.data_dir.join("sessions").join("index.json"); let intact = std::fs::read(&index).unwrap(); std::fs::write(&index, "{ not json").unwrap(); @@ -10250,6 +10259,9 @@ mod tests { sessions::record_resume_point(&core, &id, 3); sessions::record_resume_point(&core, &id, 4); + // Released first, so the hydration below claims the session over the + // damaged index: a claim that saw it indexed saves without asking. + sessions::detach(&core, &id).unwrap(); let index = core.data_dir.join("sessions").join("index.json"); let intact = std::fs::read(&index).unwrap(); std::fs::write(&index, "{ not json").unwrap(); diff --git a/crates/core/src/doctor.rs b/crates/core/src/doctor.rs index 69110a73..b03671ca 100644 --- a/crates/core/src/doctor.rs +++ b/crates/core/src/doctor.rs @@ -779,13 +779,16 @@ pub(crate) fn check_at(project_root: &Path, data_dir: &Path) -> Vec { // `--continue`, in every project, until it is moved aside. Only damage // is a finding: a healthy index is state, not configuration. Moving it // is the whole repair (a missing index is an empty store), but only - // with openmax closed: a running session whose entry leaves with the - // file reads as deleted, and its later saves are dropped silently. The - // target is a name not yet taken, so a second incident never overwrites - // the copy an earlier one moved aside. That name is only free when this - // prints; `-n` keeps it so when the command runs later (rerun from shell - // history, or copied from an older `--check`), and a skipped move leaves - // the index damaged, so the next `--check` names a free target again. + // with openmax closed: a running session's entry leaves with the file. + // One claimed while the index was healthy keeps saving, so its later + // history is written but listed nowhere (invisible to `--continue`, + // `/resume`, and `--recall`); one claimed over the damage, or claimed + // again later, has its saves dropped. The target is a name not yet + // taken, so a second incident never overwrites the copy an earlier one + // moved aside. That name is only free when this prints; `-n` keeps it + // so when the command runs later (rerun from shell history, or copied + // from an older `--check`), and a skipped move leaves the index + // damaged, so the next `--check` names a free target again. // `-n` skips without a word and exits 0, so the command checks that the // index left and, when it did not, says so and exits nonzero: a silent // skip, or a zero status a script checks, reads as a done repair while diff --git a/crates/core/src/sessions.rs b/crates/core/src/sessions.rs index f9e5441e..96e1581b 100644 --- a/crates/core/src/sessions.rs +++ b/crates/core/src/sessions.rs @@ -21,15 +21,19 @@ //! the user their work. The one thing that is not best-effort is ordering //! against deletion. Every writer re-checks under `sessions_lock` that the //! session is still indexed, so a session deleted mid-turn cannot be -//! resurrected by an append that was already in flight. +//! resurrected by an append that was already in flight. The answer comes from +//! what this process knows (`StoreMemo`), not from reading the index on every +//! write. //! //! The index itself is shared wider than one process: it is one file per data //! dir, and parallel openmax processes are normal usage. Its read-modify-write //! therefore also holds an exclusive flock (`index.lock`), and a damaged index //! is refused rather than defaulted to empty - either failure mode would end -//! with sessions silently dropped from the index, which the still-indexed gate -//! then converts into silently dropping their transcripts. +//! with sessions silently dropped from the index. No lookup finds such a +//! session again, and the still-indexed gate silently drops the transcript +//! writes of any claim that did not see it indexed. +use std::collections::{HashMap, HashSet}; use std::io::Write; use std::path::{Path, PathBuf}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -89,13 +93,13 @@ fn claim_session(core: &Core, id: &str, validate: bool, reactivate: bool) -> Res let mut owners = core.session_owners.lock().unwrap(); if let Some(owner) = owners.get_mut(id) { if reactivate { - load_messages(core, id)?; + revalidate_if_changed(core, id)?; owner.detach = false; } return Ok(()); } // See `SessionOwner`: the open and the lock go under the store lock. - let _store = lock_store(core)?; + let (mut memo, _flock) = lock_store(core)?; let path = lock_path(core, id); let file = std::fs::OpenOptions::new().create(true).write(true).truncate(false) .open(&path).map_err(|e| format!("cannot open session lock {}: {e}", path.display()))?; @@ -104,11 +108,87 @@ fn claim_session(core: &Core, id: &str, validate: bool, reactivate: bool) -> Res format!("session {id} is already open in another process; close it there or start a new session"), std::fs::TryLockError::Error(e) => format!("cannot lock session {id}: {e}"), })?; - if validate { load_messages_locked(core, id)?; } + if validate { + let seen = stamp(&messages_path(core, id)); + load_messages_locked(core, id)?; + remember_transcript(&mut memo, id, seen); + } + // The one index read the session's writes cost while this claim lasts. + if matches!(read_index(core), IndexRead::Loaded(metas) if metas.iter().any(|m| m.id == id)) { + memo.listed.insert(id.to_string()); + } owners.insert(id.to_string(), SessionOwner { _file: file, detach: false }); Ok(()) } +/// What this process knows about the session store, so that a steady-state +/// write reads nothing back to learn it. The index lists every session of +/// every project and is never pruned: a model request or a save that read it +/// would cost more with every session ever recorded. Guarded by +/// `Core::sessions_lock`, the lock every session write and removal holds, so +/// a write never acts on an answer a removal has already changed. +#[derive(Default)] +pub(crate) struct StoreMemo { + /// Sessions this process holds and saw indexed when it claimed them. No + /// other process can remove one meanwhile, since every removal claims the + /// session first, so only this process's own removal or its release of + /// the claim can make an entry stale, and both take the entry out first. + listed: HashSet, + /// Sessions this process removed. A write still in flight when its + /// session was removed (cancellation is cooperative) is dropped on this + /// record alone. + removed: HashSet, + /// Each owned transcript as this process last validated or wrote it. + transcripts: HashMap, +} + +/// A file as a stat sees it: its length, modification time, and status-change +/// time. The modification time alone can be put back, as a copy or sync that +/// preserves timestamps does, while the change time moves on every write and +/// every timestamp change and no caller can set it. Only a write that keeps +/// the length within one tick of the file system's clock goes unseen. +type Stamp = (u64, SystemTime, (i64, i64)); + +/// None when the file is missing or cannot be stat'ed, which matches no +/// remembered stamp, so whatever is there gets read. +fn stamp(path: &Path) -> Option { + let meta = std::fs::metadata(path).ok()?; + #[cfg(unix)] + let changed = { + use std::os::unix::fs::MetadataExt; + (meta.ctime(), meta.ctime_nsec()) + }; + #[cfg(not(unix))] + let changed = (0, 0); + Some((meta.len(), meta.modified().ok()?, changed)) +} + +/// Record the transcript this process just validated or wrote, as stamped +/// before it was read or after it was written. +fn remember_transcript(memo: &mut StoreMemo, id: &str, seen: Option) { + match seen { + Some(seen) => memo.transcripts.insert(id.to_string(), seen), + None => memo.transcripts.remove(id), + }; +} + +/// Validate an owned transcript again only when it changed since this +/// process last validated or wrote it. Every turn start claims its session +/// again, and reparsing the whole history there made each turn cost more +/// than the one before. Between turns an owned transcript changes only when +/// something outside openmax writes it, which moves its stamp, so damage +/// written then is still refused before the turn runs. +fn revalidate_if_changed(core: &Core, id: &str) -> Result<(), String> { + let (mut memo, _flock) = lock_store(core)?; + let seen = stamp(&messages_path(core, id)); + if seen.is_some() && memo.transcripts.get(id) == seen.as_ref() { + return Ok(()); + } + load_messages_locked(core, id)?; + remember_transcript(&mut memo, id, seen); + Ok(()) +} + /// Release idle state, or defer release until an in-flight turn has settled. /// The caller may switch frontends immediately after requesting cancellation. pub fn detach(core: &Core, id: &str) -> Result<(), String> { @@ -119,20 +199,31 @@ pub fn detach(core: &Core, id: &str) -> Result<(), String> { if let Some(owner) = owners.get_mut(id) { owner.detach = true; } } else { state.remove(id); - owners.remove(id); + release(core, &mut owners, id); } Ok(()) } /// Called under both the session map and running locks, after turn cleanup. -pub(crate) fn finish_detach(core: &Core, id: &str, state: &mut std::collections::HashMap) { +pub(crate) fn finish_detach(core: &Core, id: &str, state: &mut HashMap) { let mut owners = core.session_owners.lock().unwrap(); if owners.get(id).is_some_and(|owner| owner.detach) { state.remove(id); - owners.remove(id); + release(core, &mut owners, id); } } +/// Let go of a claim. What it let this process know goes first: once the +/// flock is released another process can claim and remove the session, and +/// from then on only the index can say whether it still exists. +fn release(core: &Core, owners: &mut HashMap, id: &str) { + let mut memo = core.sessions_lock.lock().unwrap(); + memo.listed.remove(id); + memo.transcripts.remove(id); + drop(memo); + owners.remove(id); +} + pub const UNTITLED: &str = "New session"; fn now() -> u64 { @@ -184,8 +275,8 @@ pub fn unix_now() -> u64 { /// Append a compaction event. Best-effort: failures surface as an agent warning. #[cfg(test)] pub fn append_compaction(core: &Core, id: &str, record: &CompactionRecord) { - let _guard = core.sessions_lock.lock().unwrap(); - if !still_indexed_locked(core, id) { + let memo = core.sessions_lock.lock().unwrap(); + if !still_indexed_locked(&memo, core, id) { return; } let path = compaction_path(core, id); @@ -238,8 +329,8 @@ pub fn append_usage(core: &Core, id: &str, record: &TokenUsage) { core.send_agent(id, AgentEvent::Error { message }); return; } - let _guard = core.sessions_lock.lock().unwrap(); - if !still_indexed_locked(core, id) { + let memo = core.sessions_lock.lock().unwrap(); + if !still_indexed_locked(&memo, core, id) { return; } let Ok(line) = serde_json::to_string(record) else { return }; @@ -330,8 +421,8 @@ pub fn append_archive(core: &Core, id: &str, messages: &[ChatMessage]) -> bool { if messages.is_empty() { return true; } - let _guard = core.sessions_lock.lock().unwrap(); - if !still_indexed_locked(core, id) { + let memo = core.sessions_lock.lock().unwrap(); + if !still_indexed_locked(&memo, core, id) { return true; } let mut lines = String::new(); @@ -385,8 +476,8 @@ pub fn save_manifest(core: &Core, id: &str, manifest: &crate::registry::Registry // The last of the five session files to take the rule. A refreeze can be // in flight when the session is deleted, and cancellation is cooperative, // so without this the manifest outlives everything it described. - let _guard = core.sessions_lock.lock().unwrap(); - if !still_indexed_locked(core, id) { + let memo = core.sessions_lock.lock().unwrap(); + if !still_indexed_locked(&memo, core, id) { return; } if let Err(e) = write_atomic(&manifest_path(core, id), json) { @@ -419,9 +510,10 @@ pub fn index_diagnostic(core: &Core) -> Option { /// session, pointing at `--check` when the refusal is the index's damage. /// Only frontends add the pointer, never the shared reason: `--check` gives /// the repair with the step that makes it safe (close every openmax), while -/// a bare path acted on under a running session turns that session's later -/// saves into silent no-ops. Any other refusal (a lock or write failure) -/// passes through as is. +/// a bare path acted on under a running session leaves that session listed +/// nowhere: its later saves land in history no lookup finds, or are dropped +/// without a word. Any other refusal (a lock or write failure) passes +/// through as is. pub fn refusal_with_repair(core: &Core, reason: String) -> String { match read_index(core) { IndexRead::Damaged(damage) if damage == reason => { @@ -458,10 +550,13 @@ fn read_index(core: &Core) -> IndexRead { } /// The reason names the file and the problem, never a repair. It also -/// reaches live sessions (the save warning, the compaction refusal, recall), -/// and moving the index aside under one makes the still-indexed gate read -/// its session as deleted, so every later save is dropped without a word. -/// The repair is `--check`'s to give (see `doctor::check_at`). +/// reaches live sessions (the save warning of one claimed over the damage, +/// the compaction refusal, recall), and moving the index aside under one +/// leaves its session listed nowhere. A session claimed while the index was +/// healthy keeps saving history that `--continue`, `/resume`, and `--recall` +/// never find; one claimed over the damage, or claimed again later, has +/// every later save dropped without a word. The repair is `--check`'s to +/// give (see `doctor::check_at`). fn read_index_at(path: &Path) -> IndexRead { let text = match std::fs::read_to_string(path) { Ok(text) => text, @@ -473,6 +568,8 @@ fn read_index_at(path: &Path) -> IndexRead { )) } }; + #[cfg(test)] + INDEX_BYTES_READ.with(|read| read.set(read.get() + text.len())); match serde_json::from_str(&text) { Ok(metas) => IndexRead::Loaded(metas), Err(e) => IndexRead::Damaged(format!( @@ -500,24 +597,39 @@ fn load_index_checked(core: &Core) -> Result, String> { } /// Whether the session still exists, i.e. whether writing a sidecar for it is -/// still meaningful. +/// still meaningful: Ok(false) once it is gone, Err when only a damaged index +/// could say. /// /// Every sidecar here is opened with `create`, so a write that lands after /// `delete` recreates the file it just removed. Cancellation narrows that /// window but cannot close it: a request already on the wire settles when it -/// settles, and its usage record arrives afterwards. Checking the index makes -/// the write a no-op instead, which is what "deleted" has to mean if it is to +/// settles, and its usage record arrives afterwards. Answering no makes the +/// write a no-op instead, which is what "deleted" has to mean if it is to /// mean anything. /// -/// One small read per append. Compaction and archive appends happen at prune -/// time, and a usage append happens once per request, next to a network round -/// trip that costs several orders of magnitude more. -/// Callers must already hold `sessions_lock`: an unlocked check is a -/// time-of-check/time-of-use bug, because `delete` can remove the entry and -/// the file between the check passing and the write landing, which recreates -/// exactly the file that was deleted. -fn still_indexed_locked(core: &Core, id: &str) -> bool { - matches!(read_index(core), IndexRead::Loaded(metas) if metas.iter().any(|m| m.id == id)) +/// `memo` answers for every session this process holds or removed, so a +/// steady-state write reads nothing. The index answers only for the rest, +/// such as a session claimed while the index was damaged or did not list it. +/// Callers must already hold `sessions_lock` and pass what it guards: an +/// unlocked check is a time-of-check/time-of-use bug, because `delete` can +/// remove the entry and the file between the check passing and the write +/// landing, which recreates exactly the file that was deleted. +fn indexed_locked(memo: &StoreMemo, core: &Core, id: &str) -> Result { + if memo.removed.contains(id) { + return Ok(false); + } + if memo.listed.contains(id) { + return Ok(true); + } + match read_index(core) { + IndexRead::Loaded(metas) => Ok(metas.iter().any(|m| m.id == id)), + IndexRead::Missing => Ok(false), + IndexRead::Damaged(reason) => Err(reason), + } +} + +fn still_indexed_locked(memo: &StoreMemo, core: &Core, id: &str) -> bool { + matches!(indexed_locked(memo, core, id), Ok(true)) } fn save_index(core: &Core, metas: &[SessionMeta]) -> Result<(), String> { @@ -529,12 +641,13 @@ fn save_index(core: &Core, metas: &[SessionMeta]) -> Result<(), String> { /// covers turns within one process; the index is one file per data dir, and /// two openmax processes are normal usage. Without this, their load -> save /// cycles interleave, the loser's `create` entry vanishes from the index, and -/// the still-indexed gate then silently drops every write that session makes -/// for the rest of its life - transcript included. Same flock discipline as -/// the ledger and trust stores. std's `File::lock` and `try_lock` are flock(2) -/// on Linux and macOS, the lock every released binary takes; where std has no -/// file lock (Android, DragonFly) they return Unsupported, so every store -/// refuses to write rather than write unlocked. +/// with it the session: no lookup finds it again, and the still-indexed gate +/// silently drops the writes, transcript included, of any claim that did not +/// see it indexed. Same flock discipline as the ledger and trust stores. +/// std's `File::lock` and `try_lock` are flock(2) on Linux and macOS, the +/// lock every released binary takes; where std has no file lock (Android, +/// DragonFly) they return Unsupported, so every store refuses to write rather +/// than write unlocked. /// /// Callers must already hold `sessions_lock`: flock is per open file /// description, so that is what keeps one process from contending with @@ -606,8 +719,9 @@ pub(crate) fn raw_flock(path: &std::path::Path) -> Option { /// Read-modify-write the index under the state lock (concurrent turns in /// this process) and the index flock (concurrent processes). A damaged index /// is refused, not defaulted: saving the empty fallback over it would erase -/// every session's metadata and turn the still-indexed gate off for all of -/// them, converting one bad read into permanent, silent data loss. +/// every session's metadata, hiding all of them from every lookup and closing +/// the still-indexed gate to every later claim of one, converting one bad +/// read into permanent, silent data loss. fn with_index(core: &Core, f: impl FnOnce(&mut Vec) -> R) -> Result { let _store = lock_store(core)?; let mut metas = match read_index(core) { @@ -620,7 +734,7 @@ fn with_index(core: &Core, f: impl FnOnce(&mut Vec) -> R) -> Res Ok(result) } -fn lock_store(core: &Core) -> Result<(std::sync::MutexGuard<'_, ()>, FileLock), String> { +fn lock_store(core: &Core) -> Result<(std::sync::MutexGuard<'_, StoreMemo>, FileLock), String> { let guard = core.sessions_lock.lock().unwrap(); let flock = lock_index(core)?; Ok((guard, flock)) @@ -654,6 +768,10 @@ thread_local! { static SYNCED: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; /// The OS error every directory sync on this thread fails with, if any. static FAIL_DIR_SYNC: std::cell::Cell> = const { std::cell::Cell::new(None) }; + /// Bytes this thread read from session indexes. + static INDEX_BYTES_READ: std::cell::Cell = const { std::cell::Cell::new(0) }; + /// Transcripts this thread read whole to load or validate them. + static TRANSCRIPT_READS: std::cell::Cell = const { std::cell::Cell::new(0) }; } /// Commit archive, digest record, replay boundaries, and transcript under one @@ -683,7 +801,7 @@ pub(crate) fn commit_compaction( head: usize, ) -> Result<(), String> { ensure_owned(core, id)?; - let _store = lock_store(core)?; + let (mut memo, _flock) = lock_store(core)?; let mut metas = match read_index(core) { IndexRead::Loaded(metas) => metas, _ => return Err("compaction requires a readable session index".into()), @@ -723,6 +841,7 @@ pub(crate) fn commit_compaction( Err(restore) => format!("{error}; the replay boundaries could not be restored either: {restore}"), }); } + remember_transcript(&mut memo, id, stamp(&messages_path(core, id))); Ok(()) } @@ -970,11 +1089,24 @@ fn records_end(bytes: &[u8]) -> usize { /// Sessions for one project, most recently updated first. Err names a /// damaged index rather than listing nothing. +/// +/// A session belongs to `project` when its stored path is the requested one +/// as spelled or as resolved (`state::canonical_root`). Frontends store the +/// working directory as the OS reports it, already resolved, which is the +/// form trust keys a project by, so a lookup spelled through a symlink +/// (`/tmp` for `/private/tmp`) still finds them. A stored path is never +/// resolved again: it names the directory the session was recorded in, and +/// once a symlink takes that path's place (the directory moved), the path +/// leads to another project, and resolving it would hand that project this +/// one's history through `--continue`, `/resume`, and `--recall`. A relative +/// stored path (the "." a frontend stores when it cannot read its working +/// directory) therefore matches only its own spelling, never every project. pub fn list(core: &Core, project: &str) -> Result, String> { + let wanted = crate::state::canonical_root(Path::new(project)); let mut metas: Vec<(usize, SessionMeta)> = load_index_checked(core)? .into_iter() .enumerate() - .filter(|(_, m)| m.project == project) + .filter(|(_, m)| m.project == project || Path::new(&m.project) == wanted.as_path()) .collect(); // updated_at is whole seconds and the index is append-ordered, so two // sessions touched in the same second tie on the timestamp alone; the @@ -1143,7 +1275,7 @@ fn remove(core: &Core, id: &str, only_if_empty: bool) -> Result { // The index entry and the files go under one lock. Dropping the entry // first and the files second would let an append pass its check against // the stale index and recreate what this call is removing. - let _store = lock_store(core)?; + let (mut memo, flock) = lock_store(core)?; // A turn of this process can save between the caller's look and this // lock. Every transcript write holds the lock, so the answer here is // final: a session that gained history in that window stays. @@ -1159,6 +1291,11 @@ fn remove(core: &Core, id: &str, only_if_empty: bool) -> Result { }; metas.retain(|m| m.id != id); save_index(core, &metas)?; + // Gone from the index, so gone for every write still in flight here, + // which learns it from this record rather than from the index. + memo.listed.remove(id); + memo.transcripts.remove(id); + memo.removed.insert(id.to_string()); let _ = std::fs::remove_file(messages_path(core, id)); let _ = std::fs::remove_file(manifest_path(core, id)); let _ = std::fs::remove_file(compaction_path(core, id)); @@ -1167,7 +1304,7 @@ fn remove(core: &Core, id: &str, only_if_empty: bool) -> Result { // Still held by the claim above, and unlinked under the store lock every // claim opens it under (see `SessionOwner`). let _ = std::fs::remove_file(lock_path(core, id)); - drop(_store); + drop((memo, flock)); if only_if_empty { core.cancel(id); } @@ -1247,6 +1384,8 @@ fn load_messages_locked(core: &Core, id: &str) -> Result Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None), Err(e) => return Err(format!("transcript {} is unreadable: {e}; original file preserved", path.display())), }; + #[cfg(test)] + TRANSCRIPT_READS.with(|reads| reads.set(reads.get() + 1)); let text = std::str::from_utf8(&bytes[..records_end(&bytes)]) .map_err(|e| format!("transcript {} is unreadable: {e}; original file preserved", path.display()))?; let mut parsed = Vec::new(); @@ -1289,7 +1428,7 @@ pub fn save_messages(core: &Core, id: &str, messages: &[ChatMessage], persisted: return false; } let path = messages_path(core, id); - let _store = match lock_store(core) { + let (mut memo, _flock) = match lock_store(core) { Ok(store) => store, Err(reason) => { core.send_agent(id, AgentEvent::Error { message: format!("failed to persist session: {reason}") }); @@ -1301,11 +1440,14 @@ pub fn save_messages(core: &Core, id: &str, messages: &[ChatMessage], persisted: // ends with an unconditional save. Without this the transcript of a // deleted session comes back, which is the one file that made the // deletion visible in the first place. - if !still_indexed_locked(core, id) { - // "Deleted" stays a silent no-op. A damaged index is a different - // state: the write is being dropped for a reason the user can fix, - // so say so instead of losing the transcript quietly. - if let IndexRead::Damaged(reason) = read_index(core) { + match indexed_locked(&memo, core, id) { + Ok(true) => {} + // "Deleted" stays a silent no-op. + Ok(false) => return true, + // A damaged index is a different state: the write is being dropped + // for a reason the user can fix, so say so instead of losing the + // transcript quietly. + Err(reason) => { core.send_agent( id, AgentEvent::Error { @@ -1314,7 +1456,6 @@ pub fn save_messages(core: &Core, id: &str, messages: &[ChatMessage], persisted: ); return false; } - return true; } let needs_rewrite = rewrite || messages.len() < *persisted; // Validate before replacing existing bytes. A failed append below may @@ -1337,7 +1478,14 @@ pub fn save_messages(core: &Core, id: &str, messages: &[ChatMessage], persisted: // or a power cut interrupts leaves a fragment or an empty file, with no // complete record to resume from, so the session could never be // resumed; an interrupted rename leaves no transcript, a fresh session. - let result = if needs_rewrite || (*persisted == 0 && !messages.is_empty()) { + let whole = needs_rewrite || (*persisted == 0 && !messages.is_empty()); + let wrote = whole || messages.len() > *persisted; + // Whether the file this save leaves holds only what this process + // validated or wrote, so the next turn start need not read it again + // (`revalidate_if_changed`): a whole write does, and an append does when + // nothing else wrote the file since this process last did. + let known = whole || stamp(&path).is_some_and(|seen| memo.transcripts.get(id) == Some(&seen)); + let result = if whole { write_jsonl(&path, messages) } else if messages.len() > *persisted { // Append is best-effort for the common path. On any failure (including @@ -1355,10 +1503,14 @@ pub fn save_messages(core: &Core, id: &str, messages: &[ChatMessage], persisted: match result { Ok(()) => { + if wrote { + remember_transcript(&mut memo, id, known.then(|| stamp(&path)).flatten()); + } *persisted = messages.len(); true } Err(e) => { + memo.transcripts.remove(id); core.send_agent( id, AgentEvent::Error { @@ -1849,7 +2001,8 @@ mod tests { /// file, removals included (a frontend discards the empty session it /// leaves), and `--check` sees the same damage without a Core. The shared /// reason carries no repair: it also reaches live sessions, where moving - /// the index aside turns every later save into a silent no-op. A + /// the index aside leaves the session listed nowhere, its later saves + /// landing where no lookup finds them or dropped without a word. A /// frontend's refusal points at `--check`, and only for the damage: a /// lock or write failure is not something `--check` repairs. #[test] @@ -2570,6 +2723,257 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + /// The index lists every session of every project and is never pruned, + /// so a write that reads it costs more with every session ever recorded. + /// Once a claim has seen its session indexed, a model request's usage + /// record, a transcript save, and a manifest save read none of it: no + /// other process can remove a session this one holds, since a removal + /// claims first. A removal by this process mid-turn is remembered, so the + /// turn's late writes are still dropped, without reading the index either. + #[test] + fn a_steady_state_request_and_save_read_no_index_bytes() { + let dir = std::env::temp_dir().join(format!("openmax-index-hot-path-{}", uuid::Uuid::new_v4())); + let (core, _rx) = Core::new(dir.clone()).unwrap(); + for _ in 0..20 { + create(&core, "/tmp/elsewhere".into()).unwrap(); + } + let id = create(&core, "/tmp/p".into()).unwrap().id; + attach(&core, &id).unwrap(); + let read = || INDEX_BYTES_READ.with(|read| read.replace(0)); + read(); + + let usage = |ts| TokenUsage { ts, prompt_tokens: 10, completion_tokens: 1, cached_tokens: None }; + let mut messages = vec![ChatMessage::system("rules"), ChatMessage::user("first")]; + let mut persisted = 0usize; + assert!(save_messages(&core, &id, &messages, &mut persisted, false)); + for turn in 0..3 { + append_usage(&core, &id, &usage(turn)); + messages.push(ChatMessage::user(format!("message {turn}"))); + assert!(save_messages(&core, &id, &messages, &mut persisted, false)); + } + assert!(save_messages(&core, &id, &messages, &mut persisted, true), "a rewrite too"); + save_manifest(&core, &id, &crate::registry::Registry::builtin_only().to_manifest()); + assert_eq!(read(), 0, "a steady-state request or save read the session index"); + assert_eq!(load_usage(&core, &id).len(), 3); + assert_eq!(load_messages(&core, &id).unwrap().unwrap().len(), messages.len()); + + // Deleted while its turn runs, the session keeps its claim until the + // turn settles, and what the turn still writes is dropped. + core.running.lock().unwrap().insert(id.clone()); + delete(&core, &id).unwrap(); + read(); + append_usage(&core, &id, &usage(9)); + messages.push(ChatMessage::user("late")); + assert!(save_messages(&core, &id, &messages, &mut persisted, false), "a deleted session's save is a silent no-op"); + save_manifest(&core, &id, &crate::registry::Registry::builtin_only().to_manifest()); + assert_eq!(read(), 0, "the deleted-session guard read the session index"); + for suffix in ["messages.json", "manifest.json", "usage.jsonl"] { + assert!(!sessions_dir(&core).join(format!("{id}.{suffix}")).exists(), "{suffix} came back after delete"); + } + core.running.lock().unwrap().remove(&id); + let _ = std::fs::remove_dir_all(dir); + } + + /// Every turn start claims its session again (`attach`). Reparsing the + /// whole transcript there made each turn cost more than the one before, + /// though between turns an owned transcript changes only when something + /// outside openmax writes it. So a turn start reads it again only once + /// its stat stamp moved past what this process last validated or wrote, + /// and damage written in between is still refused. + #[test] + fn a_turn_start_rereads_an_owned_transcript_only_after_it_changed() { + let dir = std::env::temp_dir().join(format!("openmax-turn-start-{}", uuid::Uuid::new_v4())); + let (core, _rx) = Core::new(dir.clone()).unwrap(); + let id = create(&core, "/tmp/p".into()).unwrap().id; + attach(&core, &id).unwrap(); + let mut messages = vec![ChatMessage::system("rules"), ChatMessage::user("first")]; + let mut persisted = 0usize; + assert!(save_messages(&core, &id, &messages, &mut persisted, false)); + messages.push(ChatMessage::user("second")); + assert!(save_messages(&core, &id, &messages, &mut persisted, false)); + let reads = || TRANSCRIPT_READS.with(|reads| reads.replace(0)); + reads(); + + attach(&core, &id).unwrap(); + attach(&core, &id).unwrap(); + assert_eq!(reads(), 0, "a turn start reparsed a transcript nothing had changed"); + + let path = messages_path(&core, &id); + let intact = std::fs::read(&path).unwrap(); + std::fs::write(&path, [intact.as_slice(), b"{damaged}\n"].concat()).unwrap(); + assert!(attach(&core, &id).unwrap_err().contains("damaged at line 4")); + assert_eq!(reads(), 1); + let edited = [intact.as_slice(), b"{\"role\":\"user\",\"content\":\"added outside\"}\n"].concat(); + std::fs::write(&path, edited).unwrap(); + attach(&core, &id).unwrap(); + attach(&core, &id).unwrap(); + assert_eq!(reads(), 1, "a repaired transcript is read once, then trusted again"); + let _ = std::fs::remove_dir_all(dir); + } + + /// An append does not read the history it extends, so it vouches for the + /// file it leaves only when nothing else wrote it since this process last + /// did. Damage written from outside during a turn, then appended past by + /// that turn's own save, would otherwise be trusted by the next turn + /// start and go unreported until a later sitting failed to load it. + #[test] + fn a_turn_start_rereads_a_transcript_written_from_outside_before_an_append() { + let dir = std::env::temp_dir().join(format!("openmax-append-over-outside-{}", uuid::Uuid::new_v4())); + let (core, _rx) = Core::new(dir.clone()).unwrap(); + let id = create(&core, "/tmp/p".into()).unwrap().id; + attach(&core, &id).unwrap(); + let mut messages = vec![ChatMessage::system("rules"), ChatMessage::user("first"), ChatMessage::user("second")]; + let mut persisted = 0usize; + assert!(save_messages(&core, &id, &messages, &mut persisted, false)); + let path = messages_path(&core, &id); + let intact = std::fs::read(&path).unwrap(); + std::fs::write(&path, [intact.as_slice(), b"{damaged}\n"].concat()).unwrap(); + let reads = || TRANSCRIPT_READS.with(|reads| reads.replace(0)); + reads(); + + messages.push(ChatMessage::user("third")); + assert!(save_messages(&core, &id, &messages, &mut persisted, false)); + assert_eq!(reads(), 0, "the save appended rather than rewrote"); + let reason = attach(&core, &id).unwrap_err(); + assert!(reason.contains("damaged at line 4"), "{reason}"); + assert_eq!(reads(), 1); + let _ = std::fs::remove_dir_all(dir); + } + + /// A stamp of length and modification time alone misses an outside edit + /// that keeps the length and puts the modification time back, as a copy + /// or sync that preserves timestamps does. The status-change time moves + /// on every write and every timestamp change, and no caller can set it, + /// so the next turn start still reads such an edit and refuses its damage. + #[cfg(unix)] + #[test] + fn a_turn_start_rereads_a_transcript_edited_with_its_modification_time_restored() { + use std::os::unix::fs::MetadataExt; + let dir = std::env::temp_dir().join(format!("openmax-restored-mtime-{}", uuid::Uuid::new_v4())); + let (core, _rx) = Core::new(dir.clone()).unwrap(); + let id = create(&core, "/tmp/p".into()).unwrap().id; + attach(&core, &id).unwrap(); + let messages = vec![ChatMessage::system("rules"), ChatMessage::user("first"), ChatMessage::user("second")]; + assert!(save_messages(&core, &id, &messages, &mut 0, false)); + let path = messages_path(&core, &id); + let saved = std::fs::metadata(&path).unwrap(); + // The edit lands on a later tick of the file system's clock than the + // save, as an edit by hand or by another program does: within one + // tick no stat field tells two writes of the same length apart. + let probe = dir.join("clock-probe"); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + for tick in 0u64.. { + std::fs::write(&probe, tick.to_string()).unwrap(); + let probed = std::fs::metadata(&probe).unwrap(); + if (probed.ctime(), probed.ctime_nsec()) > (saved.ctime(), saved.ctime_nsec()) { + break; + } + assert!(std::time::Instant::now() < deadline, "the file system clock never moved"); + std::thread::sleep(std::time::Duration::from_millis(1)); + } + + let mut damaged = std::fs::read(&path).unwrap(); + let second_line = damaged.iter().position(|&b| b == b'\n').unwrap() + 1; + damaged[second_line] = b'x'; + std::fs::write(&path, &damaged).unwrap(); + std::fs::File::options().write(true).open(&path).unwrap().set_modified(saved.modified().unwrap()).unwrap(); + let edited = std::fs::metadata(&path).unwrap(); + assert_eq!( + (edited.len(), edited.modified().unwrap()), + (saved.len(), saved.modified().unwrap()), + "the edit kept the length and modification time" + ); + + let reason = attach(&core, &id).unwrap_err(); + assert!(reason.contains("damaged at line 2"), "{reason}"); + let _ = std::fs::remove_dir_all(dir); + } + + /// Frontends record a session under the working directory as the OS + /// reports it, already resolved, which is the form trust keys a project + /// by. A lookup spelled through a symlink (`/tmp` for `/private/tmp`) + /// still lists them, and `--continue` takes the latest. + #[cfg(unix)] + #[test] + fn a_project_lists_its_sessions_from_any_spelling_of_its_path() { + let dir = std::env::temp_dir().join(format!("openmax-spelling-{}", uuid::Uuid::new_v4())); + let real = dir.join("real"); + let link = dir.join("link"); + std::fs::create_dir_all(&real).unwrap(); + std::fs::create_dir_all(dir.join("other")).unwrap(); + std::os::unix::fs::symlink(&real, &link).unwrap(); + let (core, _rx) = Core::new(dir.join("data")).unwrap(); + let spell = |path: &Path| path.display().to_string(); + let resolved = std::fs::canonicalize(&real).unwrap(); + let older = create(&core, spell(&resolved)).unwrap().id; + let newer = create(&core, spell(&resolved)).unwrap().id; + create(&core, spell(&std::fs::canonicalize(dir.join("other")).unwrap())).unwrap(); + touch_at(&core, &older, 1_000); + touch_at(&core, &newer, 2_000); + + for spelling in [&real, &link, &resolved] { + let listed: Vec = list(&core, &spell(spelling)).unwrap().into_iter().map(|m| m.id).collect(); + assert_eq!(listed, vec![newer.clone(), older.clone()], "{}", spelling.display()); + } + touch_at(&core, &older, 3_000); + assert_eq!(latest(&core, &spell(&link)).unwrap().unwrap().id, older); + let _ = std::fs::remove_dir_all(dir); + } + + /// A stored path is the directory a session was recorded in, resolved + /// when it was recorded. Once that directory moves and a symlink to + /// another project takes its place, the path leads to the other project, + /// as it does for trust. Resolving the stored path again at lookup would + /// hand the other project this one's history through `--continue`, + /// `/resume`, and `--recall`. + #[cfg(unix)] + #[test] + fn a_symlink_that_takes_a_projects_place_does_not_carry_its_history() { + let dir = std::env::temp_dir().join(format!("openmax-retargeted-{}", uuid::Uuid::new_v4())); + let first = dir.join("first"); + let second = dir.join("second"); + std::fs::create_dir_all(&first).unwrap(); + std::fs::create_dir_all(&second).unwrap(); + let (core, _rx) = Core::new(dir.join("data")).unwrap(); + let spell = |path: &Path| path.display().to_string(); + let recorded = std::fs::canonicalize(&first).unwrap(); + let theirs = create(&core, spell(&recorded)).unwrap().id; + let ours = create(&core, spell(&std::fs::canonicalize(&second).unwrap())).unwrap().id; + touch_at(&core, &ours, 1_000); + touch_at(&core, &theirs, 2_000); + + std::fs::rename(&recorded, dir.join("moved")).unwrap(); + std::os::unix::fs::symlink(&second, &recorded).unwrap(); + for spelling in [second.clone(), std::fs::canonicalize(&second).unwrap()] { + let listed: Vec = list(&core, &spell(&spelling)).unwrap().into_iter().map(|m| m.id).collect(); + assert_eq!(listed, vec![ours.clone()], "{}", spelling.display()); + } + assert_eq!(latest(&core, &spell(&second)).unwrap().unwrap().id, ours); + let _ = std::fs::remove_dir_all(dir); + } + + /// A frontend whose working directory cannot be read stores the project + /// as ".", which names whatever directory the process that wrote it was + /// in. Resolved by a later lookup it would name the lookup's own + /// directory, so one stray entry would be every project's history and + /// `--continue` anywhere would resume a stranger's session. A stored path + /// is never resolved, so a relative one matches only its own spelling. + #[test] + fn a_relative_project_path_is_no_other_projects_history() { + let dir = std::env::temp_dir().join(format!("openmax-relative-{}", uuid::Uuid::new_v4())); + let (core, _rx) = Core::new(dir.join("data")).unwrap(); + let stray = create(&core, ".".into()).unwrap().id; + let here = std::env::current_dir().unwrap().display().to_string(); + + assert!(list(&core, &here).unwrap().is_empty()); + assert!(latest(&core, &here).unwrap().is_none()); + let own = create(&core, here.clone()).unwrap().id; + assert_eq!(latest(&core, &here).unwrap().unwrap().id, own); + let listed: Vec = list(&core, ".").unwrap().into_iter().map(|m| m.id).collect(); + assert!(listed.contains(&stray), "a relative path still matches its own spelling"); + let _ = std::fs::remove_dir_all(dir); + } + /// The manifest must reconstruct the exact frozen registry with no config /// on disk at all: the fixture tool files are deleted before reload. #[test] diff --git a/crates/core/src/state.rs b/crates/core/src/state.rs index 14f97aaf..ccd66803 100644 --- a/crates/core/src/state.rs +++ b/crates/core/src/state.rs @@ -149,8 +149,10 @@ pub struct Core { pub cancel_flags: Mutex>>, /// Pending tool-approval prompts awaiting a user decision. pub approvals: Mutex>>, - /// Serializes read-modify-write cycles on the session index file. - pub sessions_lock: Mutex<()>, + /// Serializes this process's session store writes, the index's + /// read-modify-write cycles included, and guards what it knows about the + /// store without reading it (`sessions::StoreMemo`). + pub(crate) sessions_lock: Mutex, /// Process ownership lasts as long as writable session state is attached. pub(crate) session_owners: Mutex>, /// Panic restorations must finish before the terminal receipt releases a turn. diff --git a/crates/tui/src/main.rs b/crates/tui/src/main.rs index a72e6d3a..1bfdc234 100644 --- a/crates/tui/src/main.rs +++ b/crates/tui/src/main.rs @@ -407,8 +407,9 @@ async fn main() -> std::io::Result<()> { if let Some(query) = &cli.recall { // Read-only introspection, like --ledger: no session, no endpoint, // no trust gate. Recall only ever surfaces this project's own history - // (the session index is keyed by project), and the project key is the - // same raw current_dir form session creation stores. + // (the session index is keyed by project), and the index matches + // this path, as given or resolved, against the resolved path each + // session was recorded in. let project = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")); // Settings are how a turn reaches a provider and what it may spend; // recall reads neither, so a settings file this process will never act diff --git a/crates/tui/tests/cli.rs b/crates/tui/tests/cli.rs index ccaed20c..1ed94ef1 100644 --- a/crates/tui/tests/cli.rs +++ b/crates/tui/tests/cli.rs @@ -106,6 +106,47 @@ fn damaged_continuation_preserves_bytes_before_hooks_or_provider_requests() { let _ = std::fs::remove_dir_all(project.parent().unwrap()); } +/// A session is recorded under its project's resolved path. When that +/// directory moves and a symlink to another project takes its place, the +/// path leads to the other project, which inherits none of the history +/// recorded there: run from it, `--recall` finds nothing and `--continue` +/// has no session to resume. +#[cfg(unix)] +#[test] +fn a_symlink_that_takes_a_projects_place_does_not_carry_its_history() { + let (project, home) = fresh_dirs("retargeted-project"); + let server = TcpListener::bind("127.0.0.1:0").unwrap(); + server.set_nonblocking(true).unwrap(); + write_settings(&home, &format!("http://{}/v1", server.local_addr().unwrap())); + let earlier = project.parent().unwrap().join("earlier"); + std::fs::create_dir(&earlier).unwrap(); + let recorded = std::fs::canonicalize(&earlier).unwrap(); + let (core, _) = open_max_core::state::Core::new(home.join(".openmax")).unwrap(); + let id = open_max_core::sessions::create(&core, recorded.display().to_string()).unwrap().id; + let messages = [ + open_max_core::types::ChatMessage::system("rules"), + open_max_core::types::ChatMessage::user("the quokkaberry rollout plan"), + ]; + assert!(open_max_core::sessions::save_messages(&core, &id, &messages, &mut 0, false)); + drop(core); + std::fs::rename(&recorded, project.parent().unwrap().join("moved")).unwrap(); + std::os::unix::fs::symlink(&project, &recorded).unwrap(); + + let recalled = cmd(&project, &home).args(["--recall", "quokkaberry", "--json"]).output().unwrap(); + assert!(recalled.status.success(), "{}", String::from_utf8_lossy(&recalled.stderr)); + let report: serde_json::Value = serde_json::from_slice(&recalled.stdout).unwrap(); + assert_eq!(report["sessions_scanned"], 0, "{report}"); + assert!(report["hits"].as_array().unwrap().is_empty(), "{report}"); + + let resumed = finish_with_deadline(cmd(&project, &home).args(["--trust-project", "--continue", "--stdio"]) + .stdin(Stdio::null()).stdout(Stdio::piped()).stderr(Stdio::piped()).spawn().unwrap()); + let stderr = String::from_utf8_lossy(&resumed.stderr); + assert_eq!(resumed.status.code(), Some(2), "{stderr}"); + assert!(stderr.contains("no prior session in this directory"), "{stderr}"); + assert_eq!(server.accept().unwrap_err().kind(), std::io::ErrorKind::WouldBlock); + let _ = std::fs::remove_dir_all(project.parent().unwrap()); +} + /// A damaged session index is history openmax cannot read, not an empty past: /// `--continue` names the file instead of reporting that no prior session /// exists, a new session is refused the same way, and `--check` lists it as