diff --git a/Cargo.lock b/Cargo.lock index 8ca24cd80..72f8b0246 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1595,6 +1595,7 @@ dependencies = [ "tinyinference-llm", "tinytools 0.4.1", "tokio", + "tracing", ] [[package]] diff --git a/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs b/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs index 19542bf60..36a5d327d 100644 --- a/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs +++ b/crates/tinyagents-integration-tests/tests/feature_session_transcript.rs @@ -11,6 +11,8 @@ use tinyagents_session::transcript::{ fn meta(turn_count: usize, input_tokens: u64, output_tokens: u64) -> TranscriptMeta { TranscriptMeta { + session_id: None, + parent_session_id: None, agent_name: "researcher".into(), agent_id: Some("researcher-v1".into()), agent_type: Some("root".into()), diff --git a/crates/tinyagents-integration-tests/tests/session_conformance.rs b/crates/tinyagents-integration-tests/tests/session_conformance.rs index 923e71749..a2edbf063 100644 --- a/crates/tinyagents-integration-tests/tests/session_conformance.rs +++ b/crates/tinyagents-integration-tests/tests/session_conformance.rs @@ -33,6 +33,8 @@ fn run_ledger_satisfies_the_conformance_suite_on_a_second_independent_workspace( fn contract_meta() -> TranscriptMeta { TranscriptMeta { + session_id: None, + parent_session_id: None, agent_name: "contract-agent".to_string(), agent_id: Some("contract-agent-id".to_string()), agent_type: Some("root".to_string()), diff --git a/crates/tinyagents-runtime/Cargo.toml b/crates/tinyagents-runtime/Cargo.toml index 96dcf9a18..fa820f24c 100644 --- a/crates/tinyagents-runtime/Cargo.toml +++ b/crates/tinyagents-runtime/Cargo.toml @@ -17,6 +17,7 @@ tinyagents-session = { path = "../tinyagents-session", version = "2.1.2" } tinyinference-llm = { path = "../../vendor/tinyinference/crates/tinyinference-llm", version = "0.3.0" } tinytools = { path = "../../vendor/tinytools/crates/tinytools", version = "0.4.1" } tokio = { workspace = true, features = ["macros", "rt", "sync"] } +tracing = { workspace = true } [dev-dependencies] anyhow = { workspace = true } diff --git a/crates/tinyagents-runtime/src/builder.rs b/crates/tinyagents-runtime/src/builder.rs index 955ee686c..f71ff29c5 100644 --- a/crates/tinyagents-runtime/src/builder.rs +++ b/crates/tinyagents-runtime/src/builder.rs @@ -1,6 +1,6 @@ use std::sync::Arc; -use tinyagents_session::transcript::{TranscriptLocator, TranscriptMeta}; +use tinyagents_session::transcript::{SessionRef, TranscriptLocator, TranscriptMeta, session_stem}; use crate::{ NoopSessionHooks, PrefixSnapshot, RuntimeError, Session, SessionDriver, SessionHooks, @@ -20,6 +20,8 @@ pub struct SessionBuilder { struct TranscriptConfig { locator: Arc, stem: String, + session: Option, + resume_agent: Option, meta: TranscriptMeta, } @@ -71,18 +73,57 @@ impl SessionBuilder { self.transcript = Some(TranscriptConfig { locator, stem: stem.into(), + session: None, + resume_agent: None, meta, }); self } + /// Enables persistence addressed by durable session identity. + /// + /// Prefer this over [`Self::transcript`]: the stem it derives is stable + /// across processes and launches, so one conversation stays in one + /// transcript instead of accumulating a file per cold boot. It is also what + /// [`ResumeMode::Session`](crate::ResumeMode::Session) resolves against. + pub fn session( + mut self, + locator: Arc, + session: SessionRef, + mut meta: TranscriptMeta, + ) -> Self { + meta.session_id = Some(session.session_id()); + meta.parent_session_id = session.parent_session_id(); + self.transcript = Some(TranscriptConfig { + locator, + stem: session_stem(&session), + session: Some(session), + resume_agent: None, + meta, + }); + self + } + + /// Uses a distinct agent key for `ResumeMode::LatestForAgent` lookup. + /// + /// Previously reachable only through a hook-supplied `ResumePreparation`, + /// which meant a host that simply wanted a different resume key had to + /// implement a hook to say so. + pub fn resume_agent(mut self, resume_agent: impl Into) -> Self { + if let Some(config) = self.transcript.as_mut() { + config.resume_agent = Some(resume_agent.into()); + } + self + } + /// Builds a session. A codec is required only when transcript persistence /// or transcript resume is configured. pub fn build(self) -> Result, RuntimeError> { let target = self.transcript.map(|config| crate::TranscriptTarget { locator: config.locator, stem: config.stem, - resume_agent: None, + resume_agent: config.resume_agent, + session: config.session, meta: config.meta, }); if target.is_some() && self.codec.is_none() { diff --git a/crates/tinyagents-runtime/src/session.rs b/crates/tinyagents-runtime/src/session.rs index 3500639ce..df82b3364 100644 --- a/crates/tinyagents-runtime/src/session.rs +++ b/crates/tinyagents-runtime/src/session.rs @@ -2,7 +2,7 @@ use std::{future::Future, sync::Arc}; use tinyagents_harness::CancellationToken; use tinyagents_session::transcript::{ - TranscriptHistory, TranscriptMessage, TranscriptPartial, TranscriptTurn, TurnUsage, + SessionRef, TranscriptHistory, TranscriptMessage, TranscriptPartial, TranscriptTurn, TurnUsage, }; use tinyinference_llm::message::Message; @@ -102,6 +102,13 @@ impl Session { history: self.history.clone(), }); }; + // Captured before the scanned transcript's metadata overwrites + // `target.meta` below, so a session-bound target resumed through + // `Thread`/`LatestForAgent` can fall back to its own pre-resume + // metadata if the write destination turns out not to exist yet — + // see the re-derivation block near the end of this method. + let pre_scan_meta = target.meta.clone(); + let mut session_binding: Option = None; let read = match options.resume { ResumeMode::Never => None, ResumeMode::LatestForAgent => target @@ -112,6 +119,45 @@ impl Session { .locator .root_for_thread_scoped(thread, target.meta.agent_id.as_deref()) }), + ResumeMode::Session => { + let Some(session) = options.session.clone().or_else(|| target.session.clone()) + else { + return Ok(SessionResume { + loaded: false, + history: self.history.clone(), + }); + }; + // The head generation, not the session the host named: a + // compaction may have sealed that one and opened a successor, + // and the head is the conversation the model is continuing. + let head = target.locator.head_generation(&session); + let read = target.locator.read_session_transcript(&head); + if read.is_some() { + session_binding = Some(head); + } else if let Some(thread) = options.thread_id.as_deref() { + // Nothing under this identity yet. A conversation written + // before session identity existed is spread over one or + // more timestamped stems; fold them in once so the model + // regains the turns the newest-wins lookup had stranded. + // Adoption is best effort: it recovers history that would + // otherwise be stranded, but failing to recover it must not + // fail the turn the user is waiting on. + if let Err(error) = target.locator.adopt_legacy(&session, thread, &target.meta) + { + tracing::warn!( + "[session] legacy adoption failed session={} thread={thread}: {error}", + session.session_id() + ); + } + session_binding = Some(session.clone()); + } + match read { + Some(read) => Some(read), + None => session_binding + .as_ref() + .and_then(|bound| target.locator.read_session_transcript(bound)), + } + } }; let Some(read) = read else { return Ok(SessionResume { @@ -142,13 +188,98 @@ impl Session { } // A successful explicit resume always rebinds the write handle to the // selected transcript. Builder construction itself remains I/O-free. + // + // For a session resume the handle must address **the file that was just + // read**, not the target's original stem. Binding elsewhere is what + // used to re-materialise a resumed history into a fresh stem and + // orphan the original, leaving two roots claiming one thread. + if let (Some(target), Some(head)) = (self.target.as_mut(), session_binding) { + target.rebind_session(head); + } else if let Some(target) = self.target.as_mut() + && let Some(session) = target.session.clone() + { + // `session_binding` above is set only on the `ResumeMode::Session` + // path, so a session-bound target resumed through `Thread` or + // `LatestForAgent` would otherwise reach the bind below still + // naming generation 0 — even when an earlier compaction already + // sealed it and opened a later head. That write would land in a + // generation the design requires to stay sealed and byte-for-byte + // unchanged. Resolving the head here, for every mode, is what + // `persist`'s own equivalent guard (`self.transcript.is_none()`) + // cannot substitute for: `self.transcript` is bound unconditionally + // a few lines down, so by the time `persist` runs on this turn + // that guard has already been satisfied. + let head = target.locator.head_generation(&session); + if head != session { + target.rebind_session(head); + } + } let target = self.target.as_ref().expect("target checked above"); - self.transcript = Some( - target + self.transcript = Some(match target.session.as_ref() { + Some(session) => target + .locator + .open_session(session, target.meta.clone()) + .map_err(|error| RuntimeError::Persistence(error.to_string()))?, + None => target .locator .open_stem(&target.stem, target.meta.clone()) .map_err(|error| RuntimeError::Persistence(error.to_string()))?, - ); + }); + // For a session-bound target, `target.session`/`target.stem` always + // name the same file (construction and `rebind_session` keep them in + // lockstep) — the bind above is always that file, regardless of + // resume mode. Under `ResumeMode::Session`, `read` was already that + // same file, so `self.persisted` (set above from `transcript`, + // i.e. from `read`) already matches what this turn will append to. + // Under `Thread`/`LatestForAgent`, `read` can legitimately be a + // *different* file — a newest-wins scan recovering history from + // wherever it exists is exactly their contract — while the destination + // this turn writes to is still the session's own, separately-tracked + // file. Using the scan's raw rows as the append-diff baseline for a + // write that lands elsewhere would corrupt whatever is already on + // that other file. Re-derive the baseline from the file this turn + // actually writes to; `self.history` (what the model sees) keeps + // coming from the scanned `read`, which is the intended recovery + // behavior for those modes. + if target.session.is_some() && options.resume != ResumeMode::Session { + // The scanned file's `_meta` (set a few lines up, from `read`) + // is equally wrong as an append baseline when `read` was a + // different file: without this, the destination's next `_meta` + // record would carry over the scanned file's `agent_id`, + // `created`, provider/model, token/cost totals and (unless the + // head changed) session identifiers — none of which describe + // the file actually being appended to. + let destination = self + .transcript + .as_ref() + .expect("bound above") + .read_session() + .map_err(|error| RuntimeError::Persistence(error.to_string()))?; + match destination { + Some(destination_transcript) => { + self.persisted = destination_transcript.messages; + if let Some(target) = self.target.as_mut() { + target.meta = destination_transcript.meta; + } + } + None => { + // Nothing at the destination yet: fall back to this + // target's own pre-resume metadata rather than the + // scanned file's, then reapply the session binding so + // `session_id`/`parent_session_id` stay canonical for + // whatever session this target now names (`resume`'s + // own head-resolution above may have rebound it). + self.persisted = Vec::new(); + if let Some(target) = self.target.as_mut() { + target.meta = pre_scan_meta; + if let Some(session) = target.session.clone() { + target.meta.session_id = Some(session.session_id()); + target.meta.parent_session_id = session.parent_session_id(); + } + } + } + } + } Ok(SessionResume { loaded: true, history, @@ -412,22 +543,81 @@ impl Session { return Ok(None); }; if self.transcript.is_none() { - self.transcript = Some( - target + // A turn can reach the first bind through a resume mode other + // than `ResumeMode::Session` (e.g. `Never`, `LatestForAgent`, + // `Thread`) on a session-bound target — `resume` only rebinds to + // the head generation on its own `Session` path. Without this, + // such a turn binds generation 0 even when a later `.g{n}` + // exists: it appends into a generation the design requires to + // stay sealed, and the next compaction's `begin_generation` then + // fails outright because that later generation already exists. + if let Some(session) = target.session.clone() { + let head = target.locator.head_generation(&session); + if head != session { + target.rebind_session(head); + } + } + self.transcript = Some(match target.session.as_ref() { + Some(session) => target + .locator + .open_session(session, target.meta.clone()) + .map_err(|error| RuntimeError::Persistence(error.to_string()))?, + None => target .locator .open_stem(&target.stem, target.meta.clone()) .map_err(|error| RuntimeError::Persistence(error.to_string()))?, - ); + }); } - let transcript = self.transcript.as_ref().expect("bound above"); + + let previous_len = self.persisted.len(); + let next_len = raw.len(); + let common_len = previous_len.min(next_len); + let extends = next_len >= previous_len && raw[..common_len] == self.persisted[..common_len]; + + // A turn that no longer extends what is persisted is a compaction. For + // a session-bound target that seals the current generation and opens + // the next one rather than appending a replacement record: rewriting + // the logical set in place would make the replaced turns unreadable + // forever, and they are the conversation's own history. + // + // The successor generation and handle are kept in locals, not written + // onto `target`/`self.transcript`, until the append into them below + // actually succeeds. Committing them first — as this used to — left + // `target` pointing at `.g{n+1}` even when the append failed to + // create it: the next turn's `begin_generation` would then find no + // file at `.g{n+1}`, mint `.g{n+2}` instead, and `head_generation` + // would keep resolving the old sealed generation as the head, + // orphaning both the failed generation and the one after it. + let mut prev: &[TranscriptMessage] = &self.persisted; + let empty: [TranscriptMessage; 0] = []; + let mut pending_generation: Option<(SessionRef, Arc)> = None; let mut meta = target.meta.clone(); + if !extends && let Some(session) = target.session.clone() { + let (successor, handle) = target + .locator + .begin_generation(&session, target.meta.clone()) + .map_err(|error| RuntimeError::Persistence(error.to_string()))?; + // The successor starts empty, so the retained set is written + // through the ordinary turn path below and keeps its usage, + // request ids and display partial. + meta.turn_count = 0; + meta.session_id = Some(successor.session_id()); + meta.parent_session_id = successor.parent_session_id(); + pending_generation = Some((successor, handle)); + prev = ∅ + } + + let transcript: &dyn TranscriptHistory = match pending_generation.as_ref() { + Some((_, handle)) => handle.as_ref(), + None => self.transcript.as_deref().expect("bound above"), + }; meta.turn_count += 1; meta.updated = chrono::Utc::now().to_rfc3339(); meta.thread_id = thread_id.map(str::to_owned).or(meta.thread_id); transcript .append_turn_with_partial( TranscriptTurn { - prev: &self.persisted, + prev, next: raw, meta: &meta, turn_usage, @@ -436,12 +626,18 @@ impl Session { partial, ) .map_err(|error| RuntimeError::Persistence(error.to_string()))?; + // Captured before `pending_generation`/`self.transcript` are moved + // from below — `transcript` borrows out of whichever of the two held + // the just-appended handle. + let path = transcript.path().to_path_buf(); + // Only now that the append into the successor generation has + // actually succeeded does the target move onto it. + if let Some((successor, handle)) = pending_generation { + target.rebind_session(successor); + self.transcript = Some(handle); + } target.meta = meta; - let previous_len = self.persisted.len(); - let next_len = raw.len(); - let common_len = previous_len.min(next_len); - let delta = if next_len >= previous_len && raw[..common_len] == self.persisted[..common_len] - { + let delta = if extends { TranscriptDelta::Append { previous_len, appended: previous_len..next_len, @@ -452,10 +648,7 @@ impl Session { next_len, } }; - Ok(Some(TranscriptCommitReceipt { - path: transcript.path().to_path_buf(), - delta, - })) + Ok(Some(TranscriptCommitReceipt { path, delta })) } fn with_prefix(&self, history: Vec) -> Vec { diff --git a/crates/tinyagents-runtime/src/test.rs b/crates/tinyagents-runtime/src/test.rs index 1bea3ee65..2f13ae05a 100644 --- a/crates/tinyagents-runtime/src/test.rs +++ b/crates/tinyagents-runtime/src/test.rs @@ -12,9 +12,9 @@ use tinyagents_harness::{ runtime::AgentHarness, }; use tinyagents_session::transcript::{ - DisplayRecord, FileTranscriptLocator, SessionTranscript, TranscriptHistory, TranscriptLocator, - TranscriptMessage, TranscriptMeta, TranscriptRead, TranscriptTurn, TurnUsage, read_transcript, - read_transcript_display, + DisplayRecord, FileTranscriptLocator, SessionRef, SessionTranscript, TranscriptHistory, + TranscriptLocator, TranscriptMessage, TranscriptMeta, TranscriptRead, TranscriptTurn, + TurnUsage, read_transcript, read_transcript_display, session_stem, }; use tinyinference_llm::message::Message; use tinyinference_llm::providers::MockModel; @@ -88,6 +88,8 @@ fn outcome(history: Vec) -> DriverOutcome { fn meta() -> TranscriptMeta { TranscriptMeta { + session_id: None, + parent_session_id: None, agent_name: "agent".into(), agent_id: Some("agent-id".into()), agent_type: None, @@ -168,6 +170,11 @@ struct Locator { latest_agents: Mutex>, scoped_threads: Mutex)>>, opened_stems: Mutex>, + /// Sessions this double will answer a read for. Empty means "nothing has + /// been written yet", which is how a first-turn session behaves. + known_sessions: Mutex>, + generations: Mutex>, + adopted: Mutex>, } impl TranscriptLocator for Locator { @@ -198,6 +205,34 @@ impl TranscriptLocator for Locator { *self.history.opens.lock().unwrap() += 1; Ok(self.history.clone()) } + fn read_session_transcript(&self, session: &SessionRef) -> Option> { + self.known_sessions + .lock() + .unwrap() + .contains(session) + .then(|| self.history.clone() as Arc) + } + fn adopt_legacy( + &self, + session: &SessionRef, + thread_id: &str, + _: &TranscriptMeta, + ) -> anyhow::Result> { + self.adopted + .lock() + .unwrap() + .push((session.clone(), thread_id.to_string())); + Ok(None) + } + fn begin_generation( + &self, + session: &SessionRef, + _: TranscriptMeta, + ) -> anyhow::Result<(SessionRef, Arc)> { + let successor = session.next_generation(); + self.generations.lock().unwrap().push(successor.clone()); + Ok((successor, self.history.clone())) + } } fn locator(session: Option) -> (Arc, Arc) { @@ -216,6 +251,9 @@ fn locator(session: Option) -> (Arc, Arc TurnOptions { + TurnOptions { + thread_id: Some(thread.into()), + resume, + ..TurnOptions::default() + } +} + +fn session_outcome(history: Vec, output: &str) -> DriverOutcome { + DriverOutcome { + history, + output: Some(output.into()), + partial: None, + interrupted: false, + } +} + +/// The regression this whole design exists for. Two cold sessions on one +/// conversation used to mint two stems, and resume loaded only the newest — +/// so a restart silently dropped the earlier turns. Now the second session +/// reads and appends to the file the first one wrote. +#[tokio::test] +async fn a_restarted_session_continues_the_same_transcript() { + let directory = tempfile::tempdir().unwrap(); + let session_ref = SessionRef::scoped("thread-9fa08", "agent-id"); + + let mut first = SessionBuilder::new(Arc::new(Driver::new(vec![Ok(session_outcome( + vec![ + Message::user("plan a trip to kashmir"), + Message::assistant("when?"), + ], + "when?", + ))]))) + .codec(Arc::new(Codec::default())) + .session( + Arc::new(FileTranscriptLocator::new(directory.path())), + session_ref.clone(), + meta(), + ) + .build() + .unwrap(); + first + .turn( + SessionTurnRequest::new(Message::user("plan a trip to kashmir")), + session_turn_options(ResumeMode::Session, "thread-9fa08"), + ) + .await + .unwrap(); + + // A brand-new Session over the same identity, as a restarted core builds. + let mut second = SessionBuilder::new(Arc::new(Driver::new(vec![Ok(session_outcome( + vec![ + Message::user("plan a trip to kashmir"), + Message::assistant("when?"), + Message::user("what did i ask here?"), + Message::assistant("about kashmir"), + ], + "about kashmir", + ))]))) + .codec(Arc::new(Codec::default())) + .session( + Arc::new(FileTranscriptLocator::new(directory.path())), + session_ref.clone(), + meta(), + ) + .build() + .unwrap(); + let resumed = second + .resume(&session_turn_options(ResumeMode::Session, "thread-9fa08")) + .await + .unwrap(); + + assert!(resumed.loaded, "the restart must find the first session"); + assert_eq!( + resumed.history.first().map(Message::text), + Some("plan a trip to kashmir".to_string()), + "the opening turn must survive the restart" + ); + + second + .turn( + SessionTurnRequest::new(Message::user("what did i ask here?")), + session_turn_options(ResumeMode::Session, "thread-9fa08"), + ) + .await + .unwrap(); + + let roots: Vec<_> = std::fs::read_dir(directory.path().join("session_raw")) + .unwrap() + .flatten() + .map(|entry| entry.file_name().to_string_lossy().to_string()) + .collect(); + assert_eq!( + roots.len(), + 1, + "one conversation, one transcript: {roots:?}" + ); + + let persisted = read_transcript(&directory.path().join("session_raw").join(&roots[0])).unwrap(); + let contents: Vec<&str> = persisted + .messages + .iter() + .map(|message| message.content.as_str()) + .collect(); + assert_eq!( + contents, + [ + "plan a trip to kashmir", + "when?", + "what did i ask here?", + "about kashmir" + ] + ); + assert_eq!( + persisted.meta.session_id.as_deref(), + Some(session_stem(&session_ref).as_str()) + ); +} + +/// A compaction must not rewrite the sealed file: the turns it drops are the +/// conversation's own history. +#[tokio::test] +async fn a_compaction_opens_the_next_generation_and_leaves_the_sealed_one_intact() { + let directory = tempfile::tempdir().unwrap(); + let session_ref = SessionRef::scoped("thread-1", "agent-id"); + let locator = Arc::new(FileTranscriptLocator::new(directory.path())); + + let mut session = SessionBuilder::new(Arc::new(Driver::new(vec![ + Ok(session_outcome( + vec![ + Message::user("one"), + Message::assistant("first"), + Message::user("two"), + Message::assistant("second"), + ], + "second", + )), + // The driver trimmed: the next set is no longer an extension. + Ok(session_outcome( + vec![Message::user("three"), Message::assistant("third")], + "third", + )), + ]))) + .codec(Arc::new(Codec::default())) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + + session + .turn( + SessionTurnRequest::new(Message::user("one")), + session_turn_options(ResumeMode::Session, "thread-1"), + ) + .await + .unwrap(); + let sealed = directory + .path() + .join("session_raw") + .join(format!("{}.jsonl", session_stem(&session_ref))); + let sealed_bytes = std::fs::read(&sealed).unwrap(); + + session + .turn( + SessionTurnRequest::new(Message::user("three")), + session_turn_options(ResumeMode::Never, "thread-1"), + ) + .await + .unwrap(); + + assert_eq!( + std::fs::read(&sealed).unwrap(), + sealed_bytes, + "the sealed generation must be byte-identical after a compaction" + ); + let successor = directory.path().join("session_raw").join(format!( + "{}.jsonl", + session_stem(&session_ref.next_generation()) + )); + let carried = read_transcript(&successor).unwrap(); + assert_eq!( + carried + .messages + .iter() + .map(|message| message.content.as_str()) + .collect::>(), + ["three", "third"] + ); + assert_eq!( + carried.meta.parent_session_id.as_deref(), + Some(session_stem(&session_ref).as_str()) + ); + assert_eq!(locator.head_generation(&session_ref).generation, 1); +} + +/// After a compaction, a restart must land on the head generation rather than +/// replaying the sealed one. +#[tokio::test] +async fn a_restart_after_a_compaction_resumes_the_head_generation() { + let directory = tempfile::tempdir().unwrap(); + let session_ref = SessionRef::scoped("thread-1", "agent-id"); + let locator = Arc::new(FileTranscriptLocator::new(directory.path())); + let (_, handle) = locator.begin_generation(&session_ref, meta()).unwrap(); + locator + .open_session(&session_ref, meta()) + .unwrap() + .append(TranscriptMessage::new("user", "sealed")) + .unwrap(); + handle + .append(TranscriptMessage::new("user", "current")) + .unwrap(); + + let mut session = SessionBuilder::new(Arc::new(Driver::new(Vec::new()))) + .codec(Arc::new(Codec::default())) + .session(locator, session_ref, meta()) + .build() + .unwrap(); + let resumed = session + .resume(&session_turn_options(ResumeMode::Session, "thread-1")) + .await + .unwrap(); + + assert!(resumed.loaded); + assert_eq!( + resumed + .history + .iter() + .map(Message::text) + .collect::>(), + ["current"] + ); +} + +/// A session-bound target's write destination is always its own session +/// file — construction and `rebind_session` keep `target.session`/`stem` in +/// lockstep, regardless of resume mode. `ResumeMode::Thread` can legitimately +/// read a *different* file than that destination (its contract is "find +/// this thread's newest root transcript", not "find the session's own +/// file"). The pre-turn append-diff baseline must still reflect what is +/// actually on the write destination — using the scanned read's rows +/// instead would diff the next append against the wrong file's content and +/// corrupt the destination. +#[tokio::test] +async fn thread_resume_on_a_session_bound_target_does_not_corrupt_its_own_destination() { + let directory = tempfile::tempdir().unwrap(); + let session_ref = SessionRef::scoped("thread-1", "agent-id"); + let locator = Arc::new(FileTranscriptLocator::new(directory.path())); + + // The session's own destination already carries one real message from an + // earlier session-mode turn. + locator + .open_session(&session_ref, meta()) + .unwrap() + .append(TranscriptMessage::new( + "user", + "already on the session file", + )) + .unwrap(); + + // A newer, *different* root transcript for the same thread, with a + // *different* message count (2, not 1) — what a Thread-mode newest-wins + // scan will find and seed `self.history` from instead. + let mut legacy = meta(); + legacy.thread_id = Some("thread-1".into()); + legacy.created = "zzz-later".into(); + tinyagents_session::transcript::write_transcript( + &directory.path().join("session_raw/legacy_other.jsonl"), + &[ + TranscriptMessage::new("user", "legacy one"), + TranscriptMessage::new("user", "legacy two"), + ], + &legacy, + None, + ) + .unwrap(); + + // The driver extends whatever it was handed by exactly one message. + let mut session = SessionBuilder::new(Arc::new(Driver::new(vec![Ok(session_outcome( + vec![ + Message::user("legacy one"), + Message::user("legacy two"), + Message::assistant("brand new turn"), + ], + "brand new turn", + ))]))) + .codec(Arc::new(Codec::default())) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + + session + .turn( + SessionTurnRequest::new(Message::user("legacy two")), + session_turn_options(ResumeMode::Thread, "thread-1"), + ) + .await + .unwrap(); + + // Sensitive invariant: with the append-diff baseline correctly re-derived + // from the destination's own real prior content (1 message), appending + // the driver's 1-message extension must leave the destination with + // exactly as many messages as the model's full candidate history (3) — + // regardless of what the unrelated scanned-from file contained. Using + // the scanned file's row count (2) as the wrong baseline instead makes + // the diff append too few tail rows, silently losing track of one + // message: this assertion catches exactly that class of bug rather than + // merely checking "some content survived", which an append-only writer + // satisfies by construction even when it drops the wrong number of rows. + let destination_path = directory + .path() + .join("session_raw") + .join(format!("{}.jsonl", session_stem(&session_ref))); + let on_disk = read_transcript(&destination_path).unwrap(); + assert_eq!( + on_disk.messages.len(), + 3, + "on-disk message count must match the full candidate history, not the \ + scanned-from file's unrelated row count: {:?}", + on_disk + .messages + .iter() + .map(|message| message.content.as_str()) + .collect::>() + ); + assert_eq!( + on_disk.messages[0].content, "already on the session file", + "the destination's own pre-existing message must never be overwritten" + ); +} + +/// `session_binding` is only ever set on the `ResumeMode::Session` path, so +/// without rebinding for every mode, a session-bound target resumed through +/// `Thread`/`LatestForAgent` after an earlier compaction would still bind +/// A session-bound target resumed through `Thread`/`LatestForAgent` sets +/// `target.meta` from whatever file the scan read — which, per +/// `thread_resume_on_a_session_bound_target_does_not_corrupt_its_own_destination`, +/// can legitimately be a different file than the write destination. Reusing +/// that scanned metadata as the destination's next `_meta` record would +/// carry over the scanned file's `agent_id`/`created`/etc. into a file that +/// has its own, different metadata. +#[tokio::test] +async fn thread_resume_on_a_session_bound_target_reloads_the_destinations_own_metadata() { + let directory = tempfile::tempdir().unwrap(); + let session_ref = SessionRef::scoped("thread-1", "agent-id"); + let locator = Arc::new(FileTranscriptLocator::new(directory.path())); + + // The destination's own metadata: a distinct `created` from whatever the + // scan below will find. `agent_id` has to match the builder's seed + // ("agent-id") on *both* files — `root_for_thread_scoped` filters + // candidates on it, so a mismatch would just make the scan find nothing + // and turn this into a no-op test rather than exercising the reload. + let mut destination_meta = meta(); + destination_meta.created = "destination-created".into(); + locator + .open_session(&session_ref, destination_meta) + .unwrap() + .append(TranscriptMessage::new( + "user", + "already on the session file", + )) + .unwrap(); + + // A newer, different root transcript for the same thread — what + // Thread-mode's newest-wins scan will read instead. + let mut legacy = meta(); + legacy.thread_id = Some("thread-1".into()); + legacy.created = "zzz-scanned-created".into(); + tinyagents_session::transcript::write_transcript( + &directory.path().join("session_raw/legacy_other.jsonl"), + &[TranscriptMessage::new( + "user", + "from a different file entirely", + )], + &legacy, + None, + ) + .unwrap(); + + let mut session = SessionBuilder::new(Arc::new(Driver::new(vec![Ok(session_outcome( + vec![ + Message::user("from a different file entirely"), + Message::assistant("new reply"), + ], + "new reply", + ))]))) + .codec(Arc::new(Codec::default())) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + + session + .turn( + SessionTurnRequest::new(Message::user("from a different file entirely")), + session_turn_options(ResumeMode::Thread, "thread-1"), + ) + .await + .unwrap(); + + let destination_path = directory + .path() + .join("session_raw") + .join(format!("{}.jsonl", session_stem(&session_ref))); + let on_disk = read_transcript(&destination_path).unwrap(); + assert_eq!( + on_disk.meta.created, "destination-created", + "the destination's own created timestamp must survive, not the scanned file's" + ); +} + +/// generation 0 — a generation the design requires to stay sealed and +/// byte-for-byte unchanged — instead of the actual head. +#[tokio::test] +async fn thread_resume_on_a_session_bound_target_writes_the_head_not_a_sealed_generation() { + let directory = tempfile::tempdir().unwrap(); + let session_ref = SessionRef::scoped("thread-1", "agent-id"); + let locator = Arc::new(FileTranscriptLocator::new(directory.path())); + // `ResumeMode::Thread` matches on `_meta.thread_id`, so the seed for both + // generations needs it set — otherwise `root_for_thread_scoped` finds + // neither file and `resume` returns early before ever reaching the bind + // this test is about, making the whole scenario a no-op. + let mut thread_meta = meta(); + thread_meta.thread_id = Some("thread-1".into()); + + // Seal generation 0 and open generation 1, exactly what a prior + // compaction does. + locator + .open_session(&session_ref, thread_meta.clone()) + .unwrap() + .append(TranscriptMessage::new("user", "sealed generation 0")) + .unwrap(); + let (_, head_handle) = locator + .begin_generation(&session_ref, thread_meta.clone()) + .unwrap(); + head_handle + .append(TranscriptMessage::new("user", "head generation 1")) + .unwrap(); + let sealed_path = directory + .path() + .join("session_raw") + .join(format!("{}.jsonl", session_stem(&session_ref))); + let sealed_bytes_before = std::fs::read(&sealed_path).unwrap(); + + let mut session = SessionBuilder::new(Arc::new(Driver::new(vec![Ok(session_outcome( + vec![ + Message::user("head generation 1"), + Message::assistant("new turn"), + ], + "new turn", + ))]))) + .codec(Arc::new(Codec::default())) + .session(locator.clone(), session_ref.clone(), meta()) + .build() + .unwrap(); + + session + .turn( + SessionTurnRequest::new(Message::user("continue")), + session_turn_options(ResumeMode::Thread, "thread-1"), + ) + .await + .unwrap(); + + assert_eq!( + std::fs::read(&sealed_path).unwrap(), + sealed_bytes_before, + "generation 0 must stay sealed and byte-identical" + ); + let head_path = directory.path().join("session_raw").join(format!( + "{}.jsonl", + session_stem(&session_ref.next_generation()) + )); + let head_contents: Vec = read_transcript(&head_path) + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect(); + assert!( + head_contents.contains(&"new turn".to_string()), + "the new turn must land in the head generation, not the sealed one: {head_contents:?}" + ); +} + +/// A conversation written before session identity existed is spread over +/// timestamped stems. The first session resume folds them in, so the model +/// regains the turns newest-wins lookup had stranded. +#[tokio::test] +async fn a_first_session_resume_adopts_a_pre_identity_conversation() { + let directory = tempfile::tempdir().unwrap(); + let mut legacy = meta(); + legacy.thread_id = Some("thread-9fa08".into()); + legacy.created = "2026-09-22T07:30:47Z".into(); + tinyagents_session::transcript::write_transcript( + &tinyagents_session::transcript::resolve_keyed_transcript_path( + directory.path(), + "1790062247_orchestrator", + ) + .unwrap(), + &[TranscriptMessage::new("user", "plan a trip to kashmir")], + &legacy, + None, + ) + .unwrap(); + + let mut session = SessionBuilder::new(Arc::new(Driver::new(Vec::new()))) + .codec(Arc::new(Codec::default())) + .session( + Arc::new(FileTranscriptLocator::new(directory.path())), + SessionRef::scoped("thread-9fa08", "agent-id"), + meta(), + ) + .build() + .unwrap(); + let resumed = session + .resume(&session_turn_options(ResumeMode::Session, "thread-9fa08")) + .await + .unwrap(); + + assert!(resumed.loaded, "the legacy conversation must be adopted"); + assert_eq!( + resumed + .history + .iter() + .map(Message::text) + .collect::>(), + ["plan a trip to kashmir"] + ); +} + +#[tokio::test] +async fn a_session_resume_with_no_history_anywhere_loads_nothing() { + let directory = tempfile::tempdir().unwrap(); + let mut session = SessionBuilder::new(Arc::new(Driver::new(Vec::new()))) + .codec(Arc::new(Codec::default())) + .session( + Arc::new(FileTranscriptLocator::new(directory.path())), + SessionRef::scoped("thread-fresh", "agent-id"), + meta(), + ) + .build() + .unwrap(); + + let resumed = session + .resume(&session_turn_options(ResumeMode::Session, "thread-fresh")) + .await + .unwrap(); + + assert!(!resumed.loaded); +} diff --git a/crates/tinyagents-runtime/src/types.rs b/crates/tinyagents-runtime/src/types.rs index 3aa50698f..b9adb7d67 100644 --- a/crates/tinyagents-runtime/src/types.rs +++ b/crates/tinyagents-runtime/src/types.rs @@ -4,7 +4,9 @@ use tinyagents_harness::{ CancellationToken, context::{RunConfig, RunContext}, }; -use tinyagents_session::transcript::{TranscriptLocator, TranscriptMessage, TranscriptMeta}; +use tinyagents_session::transcript::{ + SessionRef, TranscriptLocator, TranscriptMessage, TranscriptMeta, session_stem, +}; use tinyinference_llm::message::Message; use crate::{PrefixSnapshot, ToolSnapshot}; @@ -19,6 +21,13 @@ pub enum ResumeMode { LatestForAgent, /// Load the most recent root transcript matching `TurnOptions::thread_id`. Thread, + /// Load the head generation of the session bound to this target. + /// + /// Unlike [`Self::Thread`] this is an exact lookup rather than a + /// newest-wins scan, and the file it reads is the file the turn then + /// appends to. That identity between read and write is what keeps one + /// conversation in one transcript across restarts and across processes. + Session, } /// Explicit runtime controls for one session turn. @@ -31,6 +40,9 @@ pub struct TurnOptions { pub stream: bool, /// Transcript resume behavior requested for this turn. pub resume: ResumeMode, + /// Durable session to resume under [`ResumeMode::Session`]. When absent, + /// the bound target's own session is used. + pub session: Option, /// Cooperative cancellation shared with the caller. pub cancellation: CancellationToken, /// Explicit live execution context consumed by the driver. @@ -69,6 +81,10 @@ pub struct TranscriptTarget { /// Optional agent key used only by `ResumeMode::LatestForAgent` lookup. /// When absent, the write stem is also the resume lookup key. pub resume_agent: Option, + /// Durable session identity, when the host binds one. Present means + /// `ResumeMode::Session` can resolve, and that a compaction opens the next + /// generation instead of rewriting this one. + pub session: Option, pub meta: TranscriptMeta, } @@ -82,18 +98,74 @@ impl TranscriptTarget { locator, stem: stem.into(), resume_agent: None, + session: None, + meta, + } + } + + /// A target addressed by durable session identity rather than a raw stem. + /// + /// The stem is derived from the session, so it is stable across processes + /// and launches — the property a `{unix_ts}_{agent}` stem never had. + /// + /// `meta.session_id`/`parent_session_id` are populated from `session` + /// here, the same way [`Self::rebind_session`] keeps them in sync after a + /// compaction. Leaving them as whatever the caller passed in (typically + /// `None`, since a newly bound target usually has no opinion on session + /// identity yet) would otherwise let a session-addressed transcript carry + /// metadata that does not name its own session — metadata-based session + /// discovery would then fail to recognise it. + pub fn for_session( + locator: Arc, + session: SessionRef, + mut meta: TranscriptMeta, + ) -> Self { + meta.session_id = Some(session.session_id()); + meta.parent_session_id = session.parent_session_id(); + Self { + locator, + stem: session_stem(&session), + resume_agent: None, + session: Some(session), meta, } } + /// Rebinds this target onto `session` after a compaction opened it. + pub(crate) fn rebind_session(&mut self, session: SessionRef) { + self.stem = session_stem(&session); + self.meta.session_id = Some(session.session_id()); + self.meta.parent_session_id = session.parent_session_id(); + self.session = Some(session); + } + /// Uses a distinct agent key when looking up the latest transcript. pub fn with_resume_agent(mut self, resume_agent: impl Into) -> Self { self.resume_agent = Some(resume_agent.into()); self } + /// Whether `other` addresses the same durable destination as `self`. + /// + /// For a session-bound target this compares `first_generation()` rather + /// than the `SessionRef`s (or stems) directly: `before_resume` runs on + /// every turn and is expected to keep returning the *same* logical + /// target, but `resume`/`persist` call [`Self::rebind_session`] on it as + /// soon as a later generation is discovered or a compaction opens one. + /// Comparing the raw `session`/`stem` fields would then reject that + /// still-identical target the moment its generation advanced, and + /// `apply_resume_preparation` would fail every subsequent turn with + /// `InvalidSessionState`. Generation 0 is the one identity that never + /// changes across a session's lifetime, so it is what identifies "the + /// same session" here. Non-session targets have no generation to anchor + /// on, so they keep comparing the raw stem. pub(crate) fn same_binding(&self, other: &Self) -> bool { - self.stem == other.stem + let same_destination = match (&self.session, &other.session) { + (Some(a), Some(b)) => a.first_generation() == b.first_generation(), + (None, None) => self.stem == other.stem, + _ => false, + }; + same_destination && self.resume_agent == other.resume_agent && Arc::ptr_eq(&self.locator, &other.locator) } @@ -194,6 +266,7 @@ impl Default for TurnOptions<()> { thread_id: None, stream: false, resume: ResumeMode::Never, + session: None, run_context: RunContext::new(RunConfig::new("session"), ()) .with_cancellation(cancellation.clone()), cancellation, diff --git a/crates/tinyagents-session/src/testkit/conformance.rs b/crates/tinyagents-session/src/testkit/conformance.rs index 612e676f3..d8fba86bb 100644 --- a/crates/tinyagents-session/src/testkit/conformance.rs +++ b/crates/tinyagents-session/src/testkit/conformance.rs @@ -249,6 +249,8 @@ fn content_view(messages: &[TranscriptMessage]) -> Vec<(String, String)> { fn contract_meta() -> TranscriptMeta { TranscriptMeta { + session_id: None, + parent_session_id: None, agent_name: "contract-agent".to_string(), agent_id: Some("contract-agent-id".to_string()), agent_type: Some("root".to_string()), diff --git a/crates/tinyagents-session/src/transcript.rs b/crates/tinyagents-session/src/transcript.rs index a1c3ab709..ed795176b 100644 --- a/crates/tinyagents-session/src/transcript.rs +++ b/crates/tinyagents-session/src/transcript.rs @@ -105,7 +105,10 @@ //! | `markdown` | Human-readable `.md` companion rendering. | //! | `legacy_md` | Legacy HTML-comment `.md` reader. | //! | `migration` | One-shot legacy date-grouped layout conversion. | +//! | `session` | [`SessionRef`] identity and its deterministic stem. | +//! | `adoption` | Folding pre-identity transcripts into a session. | +mod adoption; mod history; mod jsonl; mod legacy_md; @@ -113,10 +116,12 @@ mod markdown; mod migration; mod paths; mod reader; +mod session; mod thread_lookup; mod types; mod writer; +pub use adoption::{SessionAdoption, adopt_legacy_session_transcripts}; pub use history::{ FileTranscriptHistory, FileTranscriptLocator, TranscriptHistory, TranscriptLocator, TranscriptPartial, TranscriptRead, TranscriptTurn, @@ -125,6 +130,7 @@ pub use legacy_md::read_transcript_legacy_md; pub use migration::{TranscriptLayoutMigration, migrate_layout_if_needed}; pub use paths::{find_latest_transcript, resolve_keyed_transcript_path}; pub use reader::{read_transcript, read_transcript_display}; +pub use session::{SessionRef, session_stem}; pub use thread_lookup::{ find_root_transcript_for_thread, find_root_transcript_for_thread_scoped, find_root_transcripts_for_thread, read_thread_usage_summary, @@ -136,7 +142,7 @@ pub use types::{ }; pub use writer::{ append_interrupted_partial, append_transcript_turn, append_transcript_turn_with_partial, - write_transcript, + write_transcript, write_transcript_if_absent, }; // ── Tests ───────────────────────────────────────────────────────────── diff --git a/crates/tinyagents-session/src/transcript/adoption.rs b/crates/tinyagents-session/src/transcript/adoption.rs new file mode 100644 index 000000000..cc1b976a7 --- /dev/null +++ b/crates/tinyagents-session/src/transcript/adoption.rs @@ -0,0 +1,365 @@ +//! Adoption of conversations written before session identity existed. +//! +//! A host that minted a transcript stem per process left one conversation +//! spread across several root transcripts, and resume only ever loaded the +//! newest of them — so the opening turns of a thread became unreachable to the +//! model while the UI, which concatenates every matching root, still displayed +//! them. +//! +//! [`adopt_legacy_session_transcripts`] closes that gap once per conversation. +//! The first time a session is resumed and has no transcript of its own, the +//! legacy roots for its thread are read in `_meta.created` order and written +//! into the session's generation 0. The legacy files are never modified, +//! moved, or deleted: adoption only ever *adds* the file the session layer +//! will use from then on. + +use anyhow::{Context, Result}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, SystemTime}; + +use super::paths::resolve_keyed_transcript_path; +use super::reader::read_transcript; +use super::session::{SessionRef, session_stem}; +use super::thread_lookup::find_root_transcripts_for_thread_reporting_unreadable; +use super::types::{TranscriptMessage, TranscriptMeta}; +use super::writer::write_transcript_if_absent; + +/// What adoption did for one session. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SessionAdoption { + /// The transcript now backing the session. + pub path: PathBuf, + /// The legacy roots folded into it, oldest first. + pub adopted: Vec, + /// Messages carried over. + pub messages: usize, +} + +/// An adoption lock older than this is assumed to belong to a process that +/// crashed mid-adoption rather than one still working, and is reclaimed +/// rather than left blocking every future resume forever. +const STALE_LOCK_AGE: Duration = Duration::from_secs(60); + +/// Fold the legacy root transcripts of `session`'s thread into its own +/// generation 0, if it has none yet and any exist. +/// +/// Returns `Ok(None)` when there is nothing to do — the session already has a +/// transcript, the thread has no legacy roots, or a concurrent adopter is +/// already handling this session — which makes repeat calls harmless. +/// +/// `thread_id` is passed separately from `session` because the legacy files are +/// keyed by `_meta.thread_id`, which is what the host had before it had a +/// session key, and the two are not required to be the same string. +pub fn adopt_legacy_session_transcripts( + workspace_dir: &Path, + session: &SessionRef, + thread_id: &str, + seed_meta: &TranscriptMeta, +) -> Result> { + let stem = session_stem(session); + let destination = resolve_keyed_transcript_path(workspace_dir, &stem)?; + if destination.exists() { + return Ok(None); + } + + // Two processes can otherwise both pass the check above, both scan the + // same legacy roots, and race to write `destination`: whichever full + // rewrite lands last wins and the other's turns (if it observed a + // different, possibly more complete, set of roots) are gone with no way + // to notice, because both callers now see `destination.exists()` and + // return `Ok(None)`. Holding this lock for the whole check-scan-write + // makes one adoption per session the only one that can ever run at a + // time; every other concurrent caller backs off and treats the winner's + // result as its own. + let lock_path = adoption_lock_path(&destination); + let Some(_lock) = AdoptionLock::acquire(&lock_path)? else { + tracing::debug!( + "[transcript-adoption] session={stem} adoption already in progress elsewhere; \ + skipping" + ); + return Ok(None); + }; + // Re-check now that the lock is held: another process may have finished + // adoption (or written the session's own first turn) between the + // unlocked check above and acquiring the lock. + if destination.exists() { + return Ok(None); + } + + // Oldest first, by `_meta.created`. Anything already pointing at this + // session's own file is excluded so a partially-adopted workspace cannot + // fold a file into itself. + // + // The scan itself (not just the fold loop below) can hit a candidate it + // cannot read at all — before it even knows whether that file's + // `_meta.thread_id` would have matched this thread. A silently-dropped + // candidate here would let the fold below complete and finalize on only + // the *other*, readable roots: the destination would then exist as the + // idempotency marker, and the unreadable file's turns would never be + // retried even once it became readable again. `unreadable` makes that + // case visible so it can defer the whole call instead. + let (legacy, unreadable): (Vec, bool) = + find_root_transcripts_for_thread_reporting_unreadable(workspace_dir, thread_id); + let legacy: Vec = legacy + .into_iter() + .filter(|path| path != &destination) + .collect(); + anyhow::ensure!( + !unreadable, + "deferring adoption: at least one root transcript in this workspace could not be read, \ + and it may belong to thread {thread_id}" + ); + if legacy.is_empty() { + return Ok(None); + } + + let mut messages: Vec = Vec::new(); + let mut meta = seed_meta.clone(); + meta.session_id = Some(session.session_id()); + meta.parent_session_id = session.parent_session_id(); + meta.thread_id = Some(thread_id.to_string()); + meta.turn_count = 0; + meta.input_tokens = 0; + meta.output_tokens = 0; + meta.cached_input_tokens = 0; + meta.charged_amount_usd = 0.0; + let mut earliest_created: Option = None; + let mut latest_updated: Option = None; + let mut adopted = Vec::new(); + + for path in legacy { + // A legacy candidate must be read to be filtered, so an unreadable + // one fails the whole call rather than being silently skipped. The + // destination is this call's own idempotency marker: skipping it and + // writing anyway would create that marker over an *incomplete* fold, + // and because `destination.exists()` short-circuits every later + // call, the skipped file's turns would never be retried — even after + // the file became readable again. Failing instead leaves nothing on + // disk, so a later resume simply tries the whole fold again. + let transcript = read_transcript(&path).with_context(|| { + format!( + "legacy root {} is unreadable; deferring adoption rather than finalizing a \ + fold that would permanently drop its turns", + path.display() + ) + })?; + + // Session-identified files are not pre-identity legacy transcripts: + // they are either one of this thread's *other* agents (when + // `session.agent_id` is set, matching the rule + // `find_root_transcript_for_thread_scoped` already applies), or a + // session file — including one already adopted — that happens to + // share this thread id. Folding either in would mix another agent's + // history into this one, or duplicate content that adoption already + // recovered once. + if transcript.meta.session_id.is_some() { + continue; + } + if let Some(expected_agent) = session.agent_id.as_deref() + && transcript.meta.agent_id.as_deref() != Some(expected_agent) + { + continue; + } + + messages.extend(transcript.messages); + meta.turn_count += transcript.meta.turn_count; + meta.input_tokens += transcript.meta.input_tokens; + meta.output_tokens += transcript.meta.output_tokens; + meta.cached_input_tokens += transcript.meta.cached_input_tokens; + meta.charged_amount_usd += transcript.meta.charged_amount_usd; + if !transcript.meta.created.is_empty() + && earliest_created + .as_ref() + .is_none_or(|earliest| transcript.meta.created < *earliest) + { + earliest_created = Some(transcript.meta.created.clone()); + } + if !transcript.meta.updated.is_empty() + && latest_updated + .as_ref() + .is_none_or(|latest| transcript.meta.updated > *latest) + { + latest_updated = Some(transcript.meta.updated.clone()); + } + adopted.push(path); + } + + if messages.is_empty() { + return Ok(None); + } + if let Some(created) = earliest_created { + meta.created = created; + } + if let Some(updated) = latest_updated { + meta.updated = updated; + } + + // `write_transcript_if_absent` never overwrites an existing destination: + // the adoption lock only serializes competing *adopters*, not a normal + // session turn independently creating this same first transcript while + // adoption is still scanning. If that happened, `destination` now holds + // real conversation data that must not be clobbered with an adoption + // fold that started before it existed — so a lost race here discards + // this call's fold and reports `Ok(None)`, the same as "nothing to do". + if !write_transcript_if_absent(&destination, &messages, &meta)? { + tracing::debug!( + "[transcript-adoption] session={stem} lost the race to a concurrent write; \ + discarding this fold" + ); + return Ok(None); + } + tracing::info!( + "[transcript-adoption] session={stem} adopted {} legacy root(s) totalling {} message(s)", + adopted.len(), + messages.len() + ); + Ok(Some(SessionAdoption { + path: destination, + messages: messages.len(), + adopted, + })) +} + +/// The exclusive-create lock path guarding one destination's adoption. +fn adoption_lock_path(destination: &Path) -> PathBuf { + let mut file_name = destination + .file_name() + .map(|name| name.to_os_string()) + .unwrap_or_default(); + file_name.push(".adopting"); + destination.with_file_name(file_name) +} + +/// An exclusive-create file lock held for the duration of one adoption. +/// +/// Backed by [`std::fs::OpenOptions::create_new`] rather than an in-process +/// mutex because concurrent adopters are typically separate processes (two +/// hosts, or a process restarted mid-turn) with no shared memory to +/// synchronize on. +/// +/// The correctness boundary that actually prevents data loss is +/// [`write_transcript_if_absent`]'s atomic publish, not this lock — the lock +/// is an efficiency optimization that lets concurrent *adopters* avoid +/// redundant scanning, not the thing standing between two writers and a +/// corrupted file. That is deliberate: reclaiming a lock purely by file age +/// (below) can never be made fully race-free without OS-level leases this +/// module does not have, so the design accepts an occasional double-scan +/// under reclamation rather than a lock a crashed owner can block forever — +/// knowing that even a full double-scan-and-write race resolves safely +/// through the atomic publish underneath it. +struct AdoptionLock { + path: PathBuf, + /// Written into the lock file's content at creation. [`Drop`] reads the + /// file back and only removes it when the content still matches this + /// token, so a lock this instance *lost* ownership of (reclaimed by + /// another process as stale, see [`Self::acquire`]) is never unlinked + /// out from under its new, legitimate owner. + token: String, +} + +impl AdoptionLock { + /// Acquires the lock at `path`, or returns `Ok(None)` when another + /// process already holds a fresh one. + /// + /// A lock older than [`STALE_LOCK_AGE`] is reclaimed on the assumption + /// that its owner crashed before releasing it — otherwise a single crash + /// mid-adoption would block that session's adoption forever. The + /// remaining reclaim-under-a-live-owner race this cannot fully close + /// (two processes both observe the same stale lock; see [`Self::token`] + /// for how `Drop` avoids compounding it) resolves safely because the + /// eventual writes still go through [`write_transcript_if_absent`]. + fn acquire(path: &Path) -> Result> { + let token = lock_token(); + match Self::create_exclusive(path, &token)? { + true => Ok(Some(Self { + path: path.to_path_buf(), + token, + })), + false if lock_is_stale(path) => { + tracing::warn!( + "[transcript-adoption] reclaiming stale adoption lock {}", + path.display() + ); + let _ = std::fs::remove_file(path); + match Self::create_exclusive(path, &token)? { + true => Ok(Some(Self { + path: path.to_path_buf(), + token, + })), + // Lost the race to reclaim it — the winner will finish + // the adoption (or lose its own race to a normal write, + // safely, via `write_transcript_if_absent`). + false => Ok(None), + } + } + false => Ok(None), + } + } + + /// Attempts to create `path` exclusively with `token` as its content. + /// Returns `Ok(true)` on success, `Ok(false)` when `path` already + /// exists. + fn create_exclusive(path: &Path, token: &str) -> Result { + use std::io::Write; + match std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(path) + { + Ok(mut file) => { + // Best-effort: the exclusive create above is what actually + // establishes ownership. A failed or partial token write + // only widens `Drop`'s safety margin (it would then decline + // to remove a lock it cannot positively confirm as its own), + // it never narrows it. + let _ = file.write_all(token.as_bytes()); + Ok(true) + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Ok(false), + Err(error) => { + Err(error).with_context(|| format!("create adoption lock {}", path.display())) + } + } + } + + /// Whether this instance's token is still what is on disk at `path`. + fn still_owns(&self) -> bool { + std::fs::read_to_string(&self.path).is_ok_and(|contents| contents == self.token) + } +} + +impl Drop for AdoptionLock { + fn drop(&mut self) { + if self.still_owns() { + let _ = std::fs::remove_file(&self.path); + } + } +} + +fn lock_is_stale(path: &Path) -> bool { + std::fs::metadata(path) + .and_then(|metadata| metadata.modified()) + .and_then(|modified| { + SystemTime::now() + .duration_since(modified) + .map_err(|_| std::io::Error::other("clock went backwards")) + }) + .is_ok_and(|age| age > STALE_LOCK_AGE) +} + +/// A per-process-unique token for one lock acquisition: pid plus a +/// monotonically increasing in-process counter. Not a cryptographic nonce — +/// it only has to distinguish this acquisition from acquisitions by other +/// processes and from earlier acquisitions in this one, which pid+counter +/// already guarantees deterministically and without any external +/// dependency. +fn lock_token() -> String { + use std::sync::atomic::{AtomicU64, Ordering}; + static COUNTER: AtomicU64 = AtomicU64::new(0); + let nonce = COUNTER.fetch_add(1, Ordering::Relaxed); + format!("{}-{nonce}", std::process::id()) +} + +#[cfg(test)] +#[path = "adoption_test.rs"] +mod test; diff --git a/crates/tinyagents-session/src/transcript/adoption_test.rs b/crates/tinyagents-session/src/transcript/adoption_test.rs new file mode 100644 index 000000000..711b15d8b --- /dev/null +++ b/crates/tinyagents-session/src/transcript/adoption_test.rs @@ -0,0 +1,784 @@ +use super::*; +use crate::transcript::{ + FileTranscriptLocator, MessageUsage, TranscriptLocator, TranscriptToolCall, TurnUsage, + append_transcript_turn, read_transcript, write_transcript, write_transcript_if_absent, +}; +use tempfile::tempdir; + +fn legacy_meta(created: &str, updated: &str, thread_id: &str) -> TranscriptMeta { + TranscriptMeta { + session_id: None, + parent_session_id: None, + agent_name: "orchestrator".into(), + agent_id: Some("orchestrator".into()), + agent_type: Some("root".into()), + dispatcher: "native".into(), + provider: None, + model: Some("model".into()), + created: created.into(), + updated: updated.into(), + turn_count: 1, + input_tokens: 10, + output_tokens: 5, + cached_input_tokens: 2, + charged_amount_usd: 0.5, + thread_id: Some(thread_id.into()), + task_id: None, + } +} + +fn write_legacy(dir: &Path, stem: &str, created: &str, body: &str, thread_id: &str) { + let path = resolve_keyed_transcript_path(dir, stem).unwrap(); + write_transcript( + &path, + &[TranscriptMessage::new("user", body)], + &legacy_meta(created, created, thread_id), + None, + ) + .unwrap(); +} + +/// The shape that cost a real user their thread: one conversation spread +/// across three timestamped stems, of which resume only ever saw the newest. +#[test] +fn legacy_roots_fold_into_one_session_in_created_order() { + let dir = tempdir().unwrap(); + let thread = "thread-9fa08c44"; + write_legacy( + dir.path(), + "1790062247_orchestrator", + "2026-09-22T07:30:47Z", + "i want to plan a trip to kashmir", + thread, + ); + write_legacy( + dir.path(), + "1790100297_orchestrator", + "2026-09-22T18:05:22Z", + "hello?", + thread, + ); + write_legacy( + dir.path(), + "1790100352_orchestrator", + "2026-09-22T18:06:27Z", + "what did i ask here?", + thread, + ); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .expect("three legacy roots should be adopted"); + + assert_eq!(adoption.adopted.len(), 3); + assert_eq!(adoption.messages, 3); + + let adopted = read_transcript(&adoption.path).unwrap(); + let contents: Vec<&str> = adopted + .messages + .iter() + .map(|message| message.content.as_str()) + .collect(); + assert_eq!( + contents, + [ + "i want to plan a trip to kashmir", + "hello?", + "what did i ask here?" + ] + ); + assert_eq!(adopted.meta.session_id, Some(session.session_id())); + assert_eq!(adopted.meta.thread_id.as_deref(), Some(thread)); +} + +#[test] +fn adoption_sums_usage_and_spans_the_whole_conversation() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy(dir.path(), "1000_a", "2026-01-01T00:00:00Z", "one", thread); + write_legacy(dir.path(), "2000_a", "2026-02-02T00:00:00Z", "two", thread); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let meta = read_transcript(&adoption.path).unwrap().meta; + assert_eq!(meta.turn_count, 2); + assert_eq!(meta.input_tokens, 20); + assert_eq!(meta.output_tokens, 10); + assert_eq!(meta.cached_input_tokens, 4); + assert_eq!(meta.charged_amount_usd, 1.0); + assert_eq!(meta.created, "2026-01-01T00:00:00Z"); + assert_eq!(meta.updated, "2026-02-02T00:00:00Z"); +} + +#[test] +fn adoption_never_touches_the_files_it_reads() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy(dir.path(), "1000_a", "2026-01-01T00:00:00Z", "one", thread); + let legacy_path = resolve_keyed_transcript_path(dir.path(), "1000_a").unwrap(); + let before = std::fs::read(&legacy_path).unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + adopt_legacy_session_transcripts(dir.path(), &session, thread, &legacy_meta("", "", thread)) + .unwrap() + .unwrap(); + + assert_eq!(std::fs::read(&legacy_path).unwrap(), before); +} + +#[test] +fn adoption_is_idempotent_and_never_re_folds() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy(dir.path(), "1000_a", "2026-01-01T00:00:00Z", "one", thread); + let session = SessionRef::scoped(thread, "orchestrator"); + let seed = legacy_meta("", "", thread); + + let first = adopt_legacy_session_transcripts(dir.path(), &session, thread, &seed) + .unwrap() + .unwrap(); + // The adopted transcript is itself a root matching this thread, so a + // second pass must recognise the session as already backed rather than + // folding the conversation into itself. + let second = adopt_legacy_session_transcripts(dir.path(), &session, thread, &seed).unwrap(); + + assert!(second.is_none()); + assert_eq!(read_transcript(&first.path).unwrap().messages.len(), 1); +} + +#[test] +fn a_thread_with_no_legacy_roots_adopts_nothing() { + let dir = tempdir().unwrap(); + let session = SessionRef::scoped("thread-fresh", "orchestrator"); + let result = adopt_legacy_session_transcripts( + dir.path(), + &session, + "thread-fresh", + &legacy_meta("", "", "thread-fresh"), + ) + .unwrap(); + + assert!(result.is_none()); +} + +#[test] +fn an_adopted_session_is_what_the_locator_then_resolves() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy(dir.path(), "1000_a", "2026-01-01T00:00:00Z", "one", thread); + let session = SessionRef::scoped(thread, "orchestrator"); + adopt_legacy_session_transcripts(dir.path(), &session, thread, &legacy_meta("", "", thread)) + .unwrap() + .unwrap(); + + let locator = FileTranscriptLocator::new(dir.path()); + let read = locator + .read_session_transcript(&session) + .expect("the adopted transcript backs the session"); + assert_eq!( + read.read_session().unwrap().unwrap().messages[0].content, + "one" + ); +} + +// ── Legacy OpenHuman layouts ────────────────────────────────────────── +// +// OpenHuman wrote transcripts several ways before session identity existed. +// Adoption has to recover a conversation from each of them, because these are +// exactly the files sitting in users' workspaces today. + +/// A sub-agent transcript shares its parent's `thread_id`. Folding one into the +/// root conversation would splice a delegated worker's private reasoning into +/// the user's chat, so the `parent__child` stems must stay out. +#[test] +fn subagent_siblings_are_never_folded_into_the_root_conversation() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy( + dir.path(), + "1000_orchestrator", + "2026-01-01T00:00:00Z", + "user ask", + thread, + ); + write_legacy( + dir.path(), + "1000_orchestrator__1001_researcher", + "2026-01-01T00:00:01Z", + "worker chatter", + thread, + ); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let contents: Vec = read_transcript(&adoption.path) + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect(); + assert_eq!(contents, ["user ask"]); + assert_eq!(adoption.adopted.len(), 1); +} + +/// OpenHuman's pre-session-key naming was `{agent}_{index}` with no timestamp +/// at all. Those stems carry a `thread_id` in `_meta` just the same, so they +/// must adopt like any other root. +#[test] +fn legacy_indexed_openhuman_stems_are_adopted() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy( + dir.path(), + "orchestrator_1", + "2026-01-01T00:00:00Z", + "first", + thread, + ); + write_legacy( + dir.path(), + "orchestrator_2", + "2026-01-02T00:00:00Z", + "second", + thread, + ); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let contents: Vec = read_transcript(&adoption.path) + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect(); + assert_eq!(contents, ["first", "second"]); +} + +/// The date-grouped `session_raw/DDMMYYYY/` layout is only reachable to the +/// thread scan after the layout migration has flattened it, so the two steps +/// have to compose: migrate, then adopt. +#[test] +fn date_grouped_openhuman_transcripts_adopt_after_the_layout_migration() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + let legacy_dir = dir.path().join("session_raw").join("01012026"); + std::fs::create_dir_all(&legacy_dir).unwrap(); + write_transcript( + &legacy_dir.join("1000_orchestrator.jsonl"), + &[TranscriptMessage::new("user", "from the dated layout")], + &legacy_meta("2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z", thread), + None, + ) + .unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + let seed = legacy_meta("", "", thread); + // Before flattening, the dated file is invisible to the root scan. + assert!( + adopt_legacy_session_transcripts(dir.path(), &session, thread, &seed) + .unwrap() + .is_none() + ); + + crate::transcript::migrate_layout_if_needed(dir.path()).unwrap(); + let adoption = adopt_legacy_session_transcripts(dir.path(), &session, thread, &seed) + .unwrap() + .expect("the flattened transcript is adoptable"); + + assert_eq!( + read_transcript(&adoption.path).unwrap().messages[0].content, + "from the dated layout" + ); +} + +/// Transcripts written before `_meta.session_id` existed deserialize with it +/// absent. Adoption must not require it — it is the whole population being +/// migrated. +#[test] +fn transcripts_without_session_identity_still_adopt() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy( + dir.path(), + "1000_a", + "2026-01-01T00:00:00Z", + "pre-identity", + thread, + ); + let legacy = + read_transcript(&resolve_keyed_transcript_path(dir.path(), "1000_a").unwrap()).unwrap(); + assert_eq!(legacy.meta.session_id, None); + assert_eq!(legacy.meta.parent_session_id, None); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let adopted = read_transcript(&adoption.path).unwrap(); + assert_eq!(adopted.messages[0].content, "pre-identity"); + assert_eq!(adopted.meta.session_id, Some(session.session_id())); +} + +/// Tool calls, tool results and usage are the reason the transcript is the +/// resume source rather than the prose conversation log. Adoption must carry +/// them across intact. +#[test] +fn adoption_preserves_tool_rounds_and_usage_of_legacy_transcripts() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + let mut assistant = TranscriptMessage::assistant("calling a tool"); + assistant.id = Some("call-1".into()); + assistant.turn_usage = Some(TurnUsage { + provider: "openrouter".into(), + model: "model".into(), + usage: MessageUsage { + input: 11, + output: 7, + cached_input: 3, + context_window: 1000, + cost_usd: 0.25, + }, + ts: "2026-01-01T00:00:00Z".into(), + reasoning_content: Some("thinking".into()), + tool_calls: vec![TranscriptToolCall { + id: "call-1".into(), + name: "web_search".into(), + arguments: "{\"q\":\"kashmir\"".into(), + extra_content: None, + }], + iteration: 2, + }); + let mut tool_result = TranscriptMessage::new("tool", "search results"); + tool_result.id = Some("call-1".into()); + tool_result.cache_breakpoints = vec![1]; + + write_transcript( + &resolve_keyed_transcript_path(dir.path(), "1000_a").unwrap(), + &[ + TranscriptMessage::new("user", "find me something"), + assistant, + tool_result, + ], + &legacy_meta("2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z", thread), + None, + ) + .unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let adopted = read_transcript(&adoption.path).unwrap(); + assert_eq!(adopted.messages.len(), 3); + let usage = adopted.messages[1] + .turn_usage + .as_ref() + .expect("assistant usage survives adoption"); + assert_eq!(usage.usage.input, 11); + assert_eq!(usage.reasoning_content.as_deref(), Some("thinking")); + assert_eq!(usage.tool_calls[0].name, "web_search"); + // Raw, unrepaired provider JSON is preserved verbatim. + assert_eq!(usage.tool_calls[0].arguments, "{\"q\":\"kashmir\""); + assert_eq!(adopted.messages[2].role, "tool"); + assert_eq!(adopted.messages[2].cache_breakpoints, vec![1]); +} + +/// A different agent scoped to the same thread id writes its own root +/// transcript. Folding it into this agent's adoption would splice one +/// agent's private history into another's — the same mixing +/// `find_root_transcript_for_thread_scoped` exists to prevent for ordinary +/// resume. +#[test] +fn a_different_agents_root_on_the_same_thread_is_never_folded_in() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy(dir.path(), "1000_a", "2026-01-01T00:00:00Z", "mine", thread); + let mut other_agent = legacy_meta("2026-01-01T00:00:01Z", "2026-01-01T00:00:01Z", thread); + other_agent.agent_id = Some("researcher".into()); + write_transcript( + &resolve_keyed_transcript_path(dir.path(), "1000_researcher").unwrap(), + &[TranscriptMessage::new("user", "not mine")], + &other_agent, + None, + ) + .unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let contents: Vec = read_transcript(&adoption.path) + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect(); + assert_eq!(contents, ["mine"]); + assert_eq!(adoption.adopted.len(), 1); +} + +/// A file that already carries session identity (already adopted, or one of +/// this session's own later generations sharing the thread id) is not +/// pre-identity legacy content. Folding it in would duplicate history that +/// adoption already recovered, or content this call has no business reading. +#[test] +fn a_session_identified_root_on_the_same_thread_is_never_folded_in() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy( + dir.path(), + "1000_a", + "2026-01-01T00:00:00Z", + "legacy", + thread, + ); + let mut already_adopted = legacy_meta("2026-01-01T00:00:01Z", "2026-01-01T00:00:01Z", thread); + already_adopted.session_id = Some("thread-1~deadbeef.orchestrator~deadbeef".into()); + write_transcript( + &resolve_keyed_transcript_path(dir.path(), "2000_a").unwrap(), + &[TranscriptMessage::new("user", "already adopted elsewhere")], + &already_adopted, + None, + ) + .unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let contents: Vec = read_transcript(&adoption.path) + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect(); + assert_eq!(contents, ["legacy"]); + assert_eq!(adoption.adopted.len(), 1); +} + +/// The thread-lookup scan itself silently drops a candidate it cannot read +/// (logs a warning, excludes it) *before* adoption's own fold loop ever sees +/// it — so a fail-fast check inside that loop alone cannot catch this case. +/// The scan must surface "something in this workspace was unreadable" so +/// adoption can defer instead of finalizing on only the readable roots. +#[test] +fn an_unreadable_root_the_scan_itself_drops_defers_adoption_rather_than_finalizing() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + write_legacy( + dir.path(), + "1000_a", + "2026-01-01T00:00:00Z", + "readable", + thread, + ); + // A `.jsonl` file the scan cannot parse at all — `find_root_transcripts_for_thread` + // would silently exclude this from its result and only log a warning. + let corrupt_path = resolve_keyed_transcript_path(dir.path(), "2000_a").unwrap(); + std::fs::write(&corrupt_path, b"not a valid meta line at all\n").unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + let result = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ); + + assert!( + result.is_err(), + "a workspace-wide unreadable candidate must defer adoption, not finalize on a partial fold" + ); + let destination = resolve_keyed_transcript_path(dir.path(), &session_stem(&session)).unwrap(); + assert!( + !destination.exists(), + "a deferred adoption must not create the idempotency marker" + ); +} + +/// Two independent adopters (simulating two racing processes) for the same +/// session must not both fold the same legacy roots: the lock in +/// [`adopt_legacy_session_transcripts`] serializes them, so the second call +/// backs off and sees the first call's result rather than re-folding (which +/// would duplicate messages) or overwriting it with a stale view. +#[test] +fn concurrent_adoption_attempts_do_not_duplicate_or_race() { + use std::sync::Arc; + use std::sync::Barrier; + + let dir = tempdir().unwrap(); + let dir_path: Arc = Arc::new(dir.path().to_path_buf()); + let thread = "thread-1"; + write_legacy( + dir_path.as_path(), + "1000_a", + "2026-01-01T00:00:00Z", + "one", + thread, + ); + write_legacy( + dir_path.as_path(), + "2000_a", + "2026-01-02T00:00:00Z", + "two", + thread, + ); + + let barrier = Arc::new(Barrier::new(4)); + let handles: Vec<_> = (0..4) + .map(|_| { + let dir_path = Arc::clone(&dir_path); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + let session = SessionRef::scoped(thread, "orchestrator"); + barrier.wait(); + adopt_legacy_session_transcripts( + dir_path.as_path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + }) + }) + .collect(); + + let results: Vec<_> = handles + .into_iter() + .map(|handle| handle.join().unwrap().unwrap()) + .collect(); + assert_eq!( + results.iter().filter(|result| result.is_some()).count(), + 1, + "exactly one racing adopter should perform the fold; results: {results:?}" + ); + + let session = SessionRef::scoped(thread, "orchestrator"); + let destination = + resolve_keyed_transcript_path(dir_path.as_path(), &session_stem(&session)).unwrap(); + let adopted = read_transcript(&destination).unwrap(); + assert_eq!(adopted.messages.len(), 2, "no duplication across racers"); +} + +/// The lock above only serializes competing *adopters* — it says nothing +/// about a normal session turn independently creating this same session's +/// first transcript while adoption is mid-scan (the CodeRabbit-reported +/// gap). `write_transcript_if_absent`, not the lock, is what has to make +/// that safe: whichever of the two legitimately wins the race to publish +/// first must never be silently destroyed by the other. +#[test] +fn a_concurrent_normal_write_and_an_adoption_never_destroy_each_other() { + use std::sync::{Arc, Barrier}; + + for _ in 0..20 { + let dir = tempdir().unwrap(); + let dir_path: Arc = Arc::new(dir.path().to_path_buf()); + let thread = "thread-1"; + write_legacy( + dir_path.as_path(), + "1000_a", + "2026-01-01T00:00:00Z", + "legacy", + thread, + ); + + let barrier = Arc::new(Barrier::new(2)); + + let adopt_dir = Arc::clone(&dir_path); + let adopt_barrier = Arc::clone(&barrier); + let adopt_thread = std::thread::spawn(move || { + let session = SessionRef::scoped(thread, "orchestrator"); + adopt_barrier.wait(); + adopt_legacy_session_transcripts( + adopt_dir.as_path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + }); + + let write_dir = Arc::clone(&dir_path); + let write_barrier = Arc::clone(&barrier); + let write_thread = std::thread::spawn(move || { + let session = SessionRef::scoped(thread, "orchestrator"); + let destination = + resolve_keyed_transcript_path(write_dir.as_path(), &session_stem(&session)) + .unwrap(); + let mut turn_meta = legacy_meta("2026-01-01T00:00:05Z", "2026-01-01T00:00:05Z", thread); + turn_meta.session_id = Some(session.session_id()); + write_barrier.wait(); + write_transcript_if_absent( + &destination, + &[TranscriptMessage::new("user", "real first turn")], + &turn_meta, + ) + .unwrap() + }); + + let adopt_result = adopt_thread.join().unwrap(); + let write_result = write_thread.join().unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + let destination = + resolve_keyed_transcript_path(dir_path.as_path(), &session_stem(&session)).unwrap(); + assert!(destination.exists(), "one of the two must have published"); + let contents: Vec = read_transcript(&destination) + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect(); + + // Exactly one side wins, and the file reflects that winner alone — + // never a mix, and the loser's data never touched disk. + if adopt_result.is_some() { + assert_eq!(contents, ["legacy"]); + assert!( + !write_result, + "adoption published first; the concurrent write must have lost" + ); + } else { + assert_eq!(contents, ["real first turn"]); + assert!( + write_result, + "the concurrent write published first; adoption must have lost" + ); + } + } +} + +/// A legacy transcript whose turns were compacted replays as its reduced set. +/// Adoption folds what the model would actually have seen, not the raw lines. +#[test] +fn adoption_folds_the_replayed_context_of_a_compacted_legacy_transcript() { + let dir = tempdir().unwrap(); + let thread = "thread-1"; + let path = resolve_keyed_transcript_path(dir.path(), "1000_a").unwrap(); + let meta = legacy_meta("2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z", thread); + write_transcript( + &path, + &[ + TranscriptMessage::new("user", "one"), + TranscriptMessage::new("user", "two"), + ], + &meta, + None, + ) + .unwrap(); + // A compaction record reduces the logical set, as OpenHuman's trim did. + let retained = [TranscriptMessage::new("user", "two")]; + append_transcript_turn( + &path, + &[ + TranscriptMessage::new("user", "one"), + TranscriptMessage::new("user", "two"), + ], + &retained, + &meta, + None, + None, + ) + .unwrap(); + + let session = SessionRef::scoped(thread, "orchestrator"); + let adoption = adopt_legacy_session_transcripts( + dir.path(), + &session, + thread, + &legacy_meta("", "", thread), + ) + .unwrap() + .unwrap(); + + let contents: Vec = read_transcript(&adoption.path) + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect(); + assert_eq!(contents, ["two"]); +} + +/// A stale-lock reclaim (see `AdoptionLock::acquire`) means two +/// `AdoptionLock` values can briefly both believe they hold the same path. +/// The one that lost that race must never unlink the file out from under +/// the one that actually holds it now — that is exactly what its +/// ownership token (compared in `Drop`) exists to prevent. +#[test] +fn drop_never_removes_a_lock_another_owner_now_holds() { + let dir = tempdir().unwrap(); + let lock_path = dir.path().join("thread-1.jsonl.adopting"); + + let first = AdoptionLock::acquire(&lock_path).unwrap().unwrap(); + // Simulate a concurrent stale-lock reclaim by another process: it + // removes the file and creates its own with a fresh token, exactly + // what `AdoptionLock::acquire`'s reclaim branch does. + std::fs::remove_file(&lock_path).unwrap(); + let second = AdoptionLock::acquire(&lock_path).unwrap().unwrap(); + + drop(first); + assert!( + lock_path.exists(), + "the first (now-stale) lock's Drop must not remove the second owner's live lock" + ); + + drop(second); + assert!( + !lock_path.exists(), + "the still-legitimate second owner's Drop must remove its own lock" + ); +} diff --git a/crates/tinyagents-session/src/transcript/history.rs b/crates/tinyagents-session/src/transcript/history.rs index 89d378579..8d53babf0 100644 --- a/crates/tinyagents-session/src/transcript/history.rs +++ b/crates/tinyagents-session/src/transcript/history.rs @@ -17,17 +17,27 @@ //! rewrite. [`TranscriptHistory::clear`] is therefore an empty compaction. //! +use std::collections::HashMap; use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{Arc, Mutex, OnceLock, Weak}; use crate::transcript::types::TranscriptMessage; use crate::transcript::{ - SessionTranscript, TranscriptMeta, TurnUsage, append_transcript_turn, find_latest_transcript, + SessionAdoption, SessionRef, SessionTranscript, TranscriptMeta, TurnUsage, + adopt_legacy_session_transcripts, append_transcript_turn, find_latest_transcript, find_root_transcript_for_thread, find_root_transcript_for_thread_scoped, read_transcript, - resolve_keyed_transcript_path, + resolve_keyed_transcript_path, session_stem, }; +/// Upper bound on the compaction generations one session may accumulate. +/// +/// Generation resolution probes `{stem}`, `{stem}.g1`, `{stem}.g2` … on disk +/// rather than consulting an index, so it needs a stop condition that holds +/// even if something in the directory is unexpected. A conversation that +/// compacts more than this many times has other problems. +const MAX_GENERATIONS: u32 = 4096; + /// One turn's worth of transcript write, borrowed. /// /// The fields mirror [`append_transcript_turn`]'s argument list one-for-one and @@ -40,7 +50,7 @@ use crate::transcript::{ /// the previously-persisted logical set in memory on `Agent` /// (`persisted_transcript_messages`) precisely so it never has to re-read a /// growing file, and a disk re-read is not a faithful substitute — see -/// `FileTranscriptHistory::write_logical_set`. +/// `FileTranscriptHistory::write_logical_set_locked`. pub struct TranscriptTurn<'a> { /// Logical message set already persisted, for the extension-vs-compaction diff. pub prev: &'a [TranscriptMessage], @@ -193,6 +203,162 @@ pub trait TranscriptLocator: Send + Sync { stem: &str, seed: TranscriptMeta, ) -> anyhow::Result>; + + /// The newest generation of `session` that exists, or `session` itself when + /// none has been written yet. + /// + /// A compaction seals a generation and opens the next + /// ([`Self::begin_generation`]), so the head is the one a resume must load + /// and append to. The default walks the successor chain through + /// [`Self::session_exists`]; an implementor with an index may override it. + fn head_generation(&self, session: &SessionRef) -> SessionRef { + let mut head = session.clone(); + if !self.session_exists(&head) { + return head; + } + while head.generation < MAX_GENERATIONS { + let next = head.next_generation(); + if !self.session_exists(&next) { + break; + } + head = next; + } + head + } + + /// Whether `session` has a transcript on disk. + fn session_exists(&self, session: &SessionRef) -> bool { + self.read_session_transcript(session).is_some() + } + + /// Every generation of `session` that exists, oldest first. + /// + /// A compaction seals a generation and opens the next, so a long + /// conversation is a chain rather than one file. The model reads only the + /// head ([`Self::head_generation`]); a host rendering or exporting the + /// conversation wants the whole chain. Empty when nothing is written yet. + fn session_chain(&self, session: &SessionRef) -> Vec { + let mut chain = Vec::new(); + let mut generation = session.first_generation(); + while generation.generation <= MAX_GENERATIONS && self.session_exists(&generation) { + chain.push(generation.clone()); + generation = generation.next_generation(); + } + chain + } + + /// Reads `session`'s transcript, or `None` when it has none yet. + /// + /// Unlike [`Self::root_for_thread`] this is an exact lookup, not a + /// newest-wins scan: one session resolves to one file, in every process and + /// on every launch. + /// + /// Defaults to opening the stem the session names through + /// [`Self::open_stem`] and reading it back. [`Self::open_stem`] alone is + /// not sufficient — it binds a handle regardless of whether anything has + /// ever been written there, so this default has to perform the read and + /// report `None` unless the transcript actually exists, rather than + /// reporting a handle for a file that was never created. An implementor + /// with a cheaper existence check (a path probe, an index) should still + /// override this. + fn read_session_transcript(&self, session: &SessionRef) -> Option> { + let stem = session_stem(session); + let handle = self + .open_stem(&stem, seed_meta_for_discovered(&stem)) + .ok()?; + match handle.read_session() { + Ok(Some(_)) => Some(handle as Arc), + _ => None, + } + } + + /// Binds `session`'s own transcript for reading **and** appending. + /// + /// This is the method that closes the bug the whole session identity exists + /// for: resume reads and the subsequent append address the same file, so a + /// restart extends the conversation instead of re-materialising it into a + /// fresh stem and orphaning the original. + fn open_session( + &self, + session: &SessionRef, + seed: TranscriptMeta, + ) -> anyhow::Result> { + self.open_stem(&session_stem(session), seed) + } + + /// Folds any pre-identity transcripts of `thread_id` into `session`, once. + /// + /// A conversation written before session identity existed is spread across + /// one or more timestamped stems, of which resume only ever loaded the + /// newest — so its opening turns became unreachable to the model. This + /// recovers them the first time the session is resumed. Returns `Ok(None)` + /// when the session already has a transcript or the thread has no legacy + /// roots, which makes repeat calls harmless. + /// + /// Defaults to doing nothing, for locators that are not file-backed. + fn adopt_legacy( + &self, + session: &SessionRef, + thread_id: &str, + seed: &TranscriptMeta, + ) -> anyhow::Result> { + let _ = (session, thread_id, seed); + Ok(None) + } + + /// Seals `session` and binds its successor generation. + /// + /// Called when a turn's logical message set is no longer an extension of + /// what is persisted — a compaction. Rewriting the sealed file in place + /// would destroy the replaced turns; instead generation `n` is left + /// byte-for-byte as it was and generation `n+1` takes the compacted set as + /// its opening write, recording `n` as its parent. The conversation stays + /// fully recoverable by walking the chain even though the model only sees + /// the head. + /// + /// The returned handle is bound but empty: the caller writes the retained + /// set through the ordinary turn path (`prev: &[]`), so usage, request ids + /// and display partials are recorded exactly as on any other turn. + /// + /// Bounded by [`MAX_GENERATIONS`] — the same limit [`Self::head_generation`] + /// and [`Self::session_chain`] stop probing at. Enforcing it here, at the + /// only place a new generation is minted, is what keeps those two bounded + /// scans complete: without it a chain could grow past what they are + /// willing to walk, leaving its newest generation undiscoverable by resume + /// and its head silently stuck on a stale, capped-off generation that the + /// ordinary append path would then go on writing into. + /// + /// Defaults to sealing through [`Self::open_session`] and the trait's own + /// existence check, which is enough for most implementors; a + /// file-backed locator overrides it only to reuse an already-resolved + /// path. Kept non-defaulted before this comment existed as a required + /// method would have broken every external implementor the moment this + /// method was added — this default is what restores that compatibility. + fn begin_generation( + &self, + session: &SessionRef, + seed: TranscriptMeta, + ) -> anyhow::Result<(SessionRef, Arc)> { + let successor = session.next_generation(); + anyhow::ensure!( + successor.generation <= MAX_GENERATIONS, + "session {} has reached the {MAX_GENERATIONS}-generation compaction limit; \ + refusing to create generation {}", + session.session_id(), + successor.generation + ); + anyhow::ensure!( + !self.session_exists(&successor), + "session generation {} already exists; refusing to overwrite a sealed transcript", + successor.session_id() + ); + + let mut meta = seed; + meta.session_id = Some(successor.session_id()); + meta.parent_session_id = successor.parent_session_id(); + let handle = self.open_session(&successor, meta)?; + Ok((successor, handle)) + } } /// The default [`TranscriptLocator`]: real files under @@ -273,6 +439,83 @@ impl TranscriptLocator for FileTranscriptLocator { seed, )?)) } + + fn session_exists(&self, session: &SessionRef) -> bool { + // A direct path probe, not a read: `head_generation` calls this once + // per generation and only needs to know whether the file is there. + // `is_file()` rather than `exists()`: a directory, FIFO or other + // non-regular entry occupying the canonical path must not be + // reported as an existing generation — reads/appends against it + // would fail (or, for a directory, silently target the wrong thing) + // downstream, and `head_generation`'s chain walk would stop at a + // phantom "generation" that was never actually written. + resolve_keyed_transcript_path(&self.workspace_dir, &session_stem(session)) + .is_ok_and(|path| path.is_file()) + } + + fn read_session_transcript(&self, session: &SessionRef) -> Option> { + let stem = session_stem(session); + let path = resolve_keyed_transcript_path(&self.workspace_dir, &stem).ok()?; + if !path.is_file() { + return None; + } + tracing::debug!( + "[transcript-history] locator read_session session={stem} path={}", + path.display() + ); + Some(Arc::new(FileTranscriptHistory::opened_at( + path, + seed_meta_for_discovered(&stem), + ))) + } + + fn adopt_legacy( + &self, + session: &SessionRef, + thread_id: &str, + seed: &TranscriptMeta, + ) -> anyhow::Result> { + adopt_legacy_session_transcripts(&self.workspace_dir, session, thread_id, seed) + } + + fn begin_generation( + &self, + session: &SessionRef, + seed: TranscriptMeta, + ) -> anyhow::Result<(SessionRef, Arc)> { + let successor = session.next_generation(); + anyhow::ensure!( + successor.generation <= MAX_GENERATIONS, + "session {} has reached the {MAX_GENERATIONS}-generation compaction limit; \ + refusing to create generation {}", + session.session_id(), + successor.generation + ); + let stem = session_stem(&successor); + let path = resolve_keyed_transcript_path(&self.workspace_dir, &stem)?; + anyhow::ensure!( + !path.exists(), + "session generation {stem} already exists; refusing to overwrite a sealed transcript" + ); + + let mut meta = seed; + meta.session_id = Some(successor.session_id()); + meta.parent_session_id = successor.parent_session_id(); + tracing::info!( + "[transcript-history] sealed session={} and opened generation {} at {}", + session.session_id(), + successor.generation, + path.display() + ); + Ok(( + successor, + Arc::new(FileTranscriptHistory::new( + &self.workspace_dir, + &stem, + meta, + )?), + )) + } } /// A placeholder `_meta` for a handle bound to an already-existing transcript. @@ -284,6 +527,8 @@ impl TranscriptLocator for FileTranscriptLocator { /// to unwrap for no benefit. fn seed_meta_for_discovered(agent_name: &str) -> TranscriptMeta { TranscriptMeta { + session_id: None, + parent_session_id: None, agent_name: agent_name.to_string(), agent_id: None, agent_type: None, @@ -395,33 +640,45 @@ impl FileTranscriptHistory { .map(|t| t.meta) .unwrap_or_else(|| self.seed_meta.clone())) } +} - /// Writes `next` as the new logical set, diffing against what is persisted. - /// - /// Routes through [`TranscriptHistory::append_turn`] so every write in this - /// module — trait-driven and turn-path alike — funnels through one call to - /// [`append_transcript_turn`], and the extension-vs-compaction decision - /// stays with the format owner rather than drifting here. - /// - /// The `self.persisted()` disk re-read is what the generic trait path has - /// to do, and is deliberately **not** what the turn path does. - /// [`read_transcript`] reconstructs `TranscriptMessage`s from line records: the - /// `failure` / `failure_detail` fields have been lifted out of - /// `extra_metadata` and turn-usage fields hoisted to top-level line fields. - /// Feeding that back in as `prev` would make `common_prefix_len` mismatch - /// at the first such message, so the writer would emit a full compaction - /// record — re-appending the entire message set — on every single turn. - fn write_logical_set(&self, next: &[TranscriptMessage]) -> anyhow::Result<()> { - let prev = self.persisted()?; - let meta = self.meta_for_write()?; - self.append_turn(TranscriptTurn { - prev: &prev, - next, - meta: &meta, - turn_usage: None, - request_id: None, - }) +/// A process-wide, per-path mutex serializing the read-modify-write sequence +/// [`FileTranscriptHistory::append`]/`replace`/`clear` run against one file. +/// +/// [`SessionRef`]'s own doc names this as a supported shape: two cores in one +/// process sharing a workspace should both see and extend one conversation. +/// Without this, two `FileTranscriptHistory` instances bound to the same +/// path (a legitimate, common way to get there — `open_session` is called +/// fresh per `Session::resume`) can each read the file's current content, +/// compute a diff against that now-stale view, and write. Whichever finishes +/// its own read first computes a `next` that does not extend what the file +/// looks like by the time it *writes* — `append_transcript_turn_with_partial` +/// then reads that mismatch as "the context was reduced" and appends a +/// **compaction record** instead of a plain tail, and a compaction's +/// replacement value is what canonical reads return going forward. The +/// other write's whole contribution becomes unreachable, even though its +/// bytes are still physically on disk as a now-superseded line — a silent +/// lost update, not a crash. +/// +/// Keyed by path rather than by `Arc>` identity because the two +/// racing instances are typically *separate* `FileTranscriptHistory` values, +/// not a shared handle. Entries are [`Weak`] and swept opportunistically so +/// the registry does not grow for the lifetime of a long-running host: once +/// every in-flight critical section for a path finishes, nothing keeps that +/// path's entry alive, and the next unrelated call reclaims the slot. +fn path_lock(path: &Path) -> Arc> { + static REGISTRY: OnceLock>>>> = OnceLock::new(); + let registry = REGISTRY.get_or_init(|| Mutex::new(HashMap::new())); + let mut locks = registry + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + locks.retain(|_, weak| weak.strong_count() > 0); + if let Some(existing) = locks.get(path).and_then(Weak::upgrade) { + return existing; } + let fresh = Arc::new(Mutex::new(())); + locks.insert(path.to_path_buf(), Arc::downgrade(&fresh)); + fresh } impl TranscriptRead for FileTranscriptHistory { @@ -449,11 +706,14 @@ impl TranscriptRead for FileTranscriptHistory { } } -impl TranscriptHistory for FileTranscriptHistory { - /// Pure forwarder: every argument reaches [`append_transcript_turn`] - /// untouched, so the bytes this writes are identical to what the free - /// function would have written at the call site. - fn append_turn(&self, turn: TranscriptTurn<'_>) -> anyhow::Result<()> { +impl FileTranscriptHistory { + /// The actual `append_turn` write. Assumes the caller already holds + /// [`path_lock`] for [`Self::path`] — never call this directly; every + /// public entry point below acquires the lock once and then routes + /// through here (and [`Self::append_turn_with_partial_locked`]) so the + /// lock is taken exactly once per call, never nested (this crate's + /// `Mutex` is not reentrant). + fn append_turn_locked(&self, turn: TranscriptTurn<'_>) -> anyhow::Result<()> { tracing::debug!( "[transcript-history] append_turn prev={} next={} usage={} request_id={:?} path={}", turn.prev.len(), @@ -472,7 +732,9 @@ impl TranscriptHistory for FileTranscriptHistory { ) } - fn append_turn_with_partial( + /// [`Self::append_turn_locked`]'s counterpart for the display-partial + /// variant. Same locking contract. + fn append_turn_with_partial_locked( &self, turn: TranscriptTurn<'_>, partial: Option<&TranscriptPartial>, @@ -494,24 +756,92 @@ impl TranscriptHistory for FileTranscriptHistory { partial, ) } + + /// Writes `next` as the new logical set, diffing against what is + /// persisted. Assumes the caller already holds [`path_lock`] for + /// [`Self::path`] — see [`Self::append_turn_locked`]'s doc for why. + /// + /// Routes through [`Self::append_turn_locked`] so every write in this + /// module — trait-driven and turn-path alike — funnels through one call + /// to [`append_transcript_turn`], and the extension-vs-compaction + /// decision stays with the format owner rather than drifting here. + /// + /// The `self.persisted()` disk re-read is what the generic trait path has + /// to do, and is deliberately **not** what the turn path does. + /// [`read_transcript`] reconstructs `TranscriptMessage`s from line records: the + /// `failure` / `failure_detail` fields have been lifted out of + /// `extra_metadata` and turn-usage fields hoisted to top-level line fields. + /// Feeding that back in as `prev` would make `common_prefix_len` mismatch + /// at the first such message, so the writer would emit a full compaction + /// record — re-appending the entire message set — on every single turn. + fn write_logical_set_locked(&self, next: &[TranscriptMessage]) -> anyhow::Result<()> { + let prev = self.persisted()?; + let meta = self.meta_for_write()?; + self.append_turn_locked(TranscriptTurn { + prev: &prev, + next, + meta: &meta, + turn_usage: None, + request_id: None, + }) + } +} + +impl TranscriptHistory for FileTranscriptHistory { + /// Pure forwarder: every argument reaches [`append_transcript_turn`] + /// untouched, so the bytes this writes are identical to what the free + /// function would have written at the call site. + /// + /// This — not [`TranscriptHistory::append`] — is the turn path's own + /// write call (`Session::persist` in `tinyagents-runtime` calls + /// [`TranscriptHistory::append_turn_with_partial`] directly), so the + /// same [`path_lock`] serialization `append`/`replace`/`clear` need + /// applies here too: two `FileTranscriptHistory` handles bound to the + /// same successor generation (two compactions racing on + /// `TranscriptLocator::begin_generation` for one session) would + /// otherwise both see the file absent and both take the writer's + /// create-fresh path, and whichever `fs::write` lands last would + /// silently discard the other's retained set. + fn append_turn(&self, turn: TranscriptTurn<'_>) -> anyhow::Result<()> { + let lock = path_lock(&self.path); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + self.append_turn_locked(turn) + } + + fn append_turn_with_partial( + &self, + turn: TranscriptTurn<'_>, + partial: Option<&TranscriptPartial>, + ) -> anyhow::Result<()> { + let lock = path_lock(&self.path); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + self.append_turn_with_partial_locked(turn, partial) + } + fn messages(&self) -> anyhow::Result> { self.persisted() } fn append(&self, message: TranscriptMessage) -> anyhow::Result<()> { + let lock = path_lock(&self.path); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let mut next = self.persisted()?; next.push(message); - self.write_logical_set(&next) + self.write_logical_set_locked(&next) } fn replace(&self, messages: &[TranscriptMessage]) -> anyhow::Result<()> { - self.write_logical_set(messages) + let lock = path_lock(&self.path); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + self.write_logical_set_locked(messages) } fn clear(&self) -> anyhow::Result<()> { + let lock = path_lock(&self.path); + let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); if !self.path.exists() { return Ok(()); } - self.write_logical_set(&[]) + self.write_logical_set_locked(&[]) } } diff --git a/crates/tinyagents-session/src/transcript/jsonl.rs b/crates/tinyagents-session/src/transcript/jsonl.rs index 5b93b8b61..a6ddb0956 100644 --- a/crates/tinyagents-session/src/transcript/jsonl.rs +++ b/crates/tinyagents-session/src/transcript/jsonl.rs @@ -55,6 +55,10 @@ pub(super) struct MetaPayload { pub(super) thread_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub(super) task_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) session_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) parent_session_id: Option, } /// One message line in the JSONL — only `role` and `content` are required. @@ -140,6 +144,8 @@ fn meta_payload_from(meta: &TranscriptMeta) -> MetaPayload { version: TRANSCRIPT_SCHEMA_VERSION, agent: meta.agent_name.clone(), agent_id: meta.agent_id.clone(), + session_id: meta.session_id.clone(), + parent_session_id: meta.parent_session_id.clone(), agent_type: meta.agent_type.clone(), dispatcher: meta.dispatcher.clone(), provider: meta.provider.clone(), @@ -276,6 +282,8 @@ pub(super) fn serialise_message_lines( /// Convert a parsed `MetaPayload` into the public [`TranscriptMeta`]. pub(super) fn meta_from_payload(mp: MetaPayload) -> TranscriptMeta { TranscriptMeta { + session_id: mp.session_id, + parent_session_id: mp.parent_session_id, agent_name: mp.agent, agent_id: mp.agent_id, agent_type: mp.agent_type, diff --git a/crates/tinyagents-session/src/transcript/legacy_md.rs b/crates/tinyagents-session/src/transcript/legacy_md.rs index 5a1762fd9..a0d2a8780 100644 --- a/crates/tinyagents-session/src/transcript/legacy_md.rs +++ b/crates/tinyagents-session/src/transcript/legacy_md.rs @@ -57,6 +57,8 @@ fn parse_legacy_meta(raw: &str) -> Result { }; Ok(TranscriptMeta { + session_id: None, + parent_session_id: None, agent_name: get("agent").unwrap_or_else(|| "unknown".into()), dispatcher: get("dispatcher").unwrap_or_else(|| "native".into()), agent_id: None, diff --git a/crates/tinyagents-session/src/transcript/paths.rs b/crates/tinyagents-session/src/transcript/paths.rs index 55833dcfc..a0c146973 100644 --- a/crates/tinyagents-session/src/transcript/paths.rs +++ b/crates/tinyagents-session/src/transcript/paths.rs @@ -37,7 +37,7 @@ pub fn resolve_keyed_transcript_path_in_dir(raw_dir: &Path, stem: &str) -> Resul /// `session_raw/` directory. Allows ASCII alphanumerics plus a small /// punctuation set (`_`, `-`, `.`); every other byte is replaced with /// `_`. Empty inputs fall back to `"session"`. -fn sanitize_stem(stem: &str) -> String { +pub(super) fn sanitize_stem(stem: &str) -> String { let cleaned: String = stem .chars() .map(|c| { diff --git a/crates/tinyagents-session/src/transcript/session.rs b/crates/tinyagents-session/src/transcript/session.rs new file mode 100644 index 000000000..9f6b820c6 --- /dev/null +++ b/crates/tinyagents-session/src/transcript/session.rs @@ -0,0 +1,350 @@ +//! Durable identity for a conversation. +//! +//! Before this module a session was identified only by the caller-supplied +//! *stem*, which hosts minted per process as `{unix_ts}_{agent}`. Two +//! consequences followed, and both cost users their history: +//! +//! 1. Every cold boot produced a new stem, so one conversation accumulated +//! several transcript files. Resume picked the newest by `_meta.created` +//! ([`super::find_root_transcript_for_thread`]), so any file that persisted +//! less than it was seeded with silently shortened the conversation for +//! every session after it. +//! 2. Two processes over one workspace each minted their own stem and neither +//! could see the other's turns. +//! +//! A [`SessionRef`] is that missing identity. It is built from values the host +//! already has — the conversation key (a thread id) and the agent id — and +//! [`session_stem`] maps it to a filename **deterministically, with no +//! timestamp**. The same conversation therefore resolves to the same +//! transcript in every process, on every launch, forever. +//! +//! # Generations +//! +//! A transcript is what the model sees, so a compaction genuinely does shorten +//! it. That must never be done by rewriting history in place: the replaced +//! turns would be gone. Instead a compaction *seals* the current generation and +//! opens the next one ([`SessionRef::next_generation`]), which starts from the +//! compacted set and records the sealed generation as its parent. Generation +//! `n` stays on disk byte-for-byte, so the full conversation remains +//! recoverable by walking the chain even though the model only ever sees the +//! head. + +use super::paths::sanitize_stem; + +/// The separator every root-transcript scan uses to recognise a sub-agent +/// stem. Kept in one place because both [`session_stem`] and the scans in +/// [`super::paths`] / [`super::thread_lookup`] must agree on it exactly. +pub(crate) const SUBAGENT_SEPARATOR: &str = "__"; + +/// Durable identity of one conversation, as seen by the session layer. +/// +/// `session_key` is the host's stable name for the conversation — OpenHuman +/// passes its `thread_id`. `agent_id` scopes it, because one caller-supplied +/// key can legitimately be handed to several independently configured agents +/// and each needs its own transcript (see +/// [`super::find_root_transcript_for_thread_scoped`] for the bug that taught us +/// this). `generation` distinguishes the segments a compaction creates. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct SessionRef { + /// The host's stable name for the conversation. + pub session_key: String, + /// The agent definition this session belongs to, when the host has one. + pub agent_id: Option, + /// Compaction segment. `0` is the original; each compaction adds one. + pub generation: u32, + /// Stem of the parent session for a sub-agent, forming the `parent__child` + /// chain the root scans filter on. `None` for a root session. + parent_stem: Option, +} + +impl SessionRef { + /// A root session for `session_key`, unscoped by agent. + pub fn root(session_key: impl Into) -> Self { + Self { + session_key: session_key.into(), + agent_id: None, + generation: 0, + parent_stem: None, + } + } + + /// A root session for `session_key`, scoped to one agent definition. + pub fn scoped(session_key: impl Into, agent_id: impl Into) -> Self { + Self { + session_key: session_key.into(), + agent_id: Some(agent_id.into()), + generation: 0, + parent_stem: None, + } + } + + /// A sub-agent session beneath `parent`. + /// + /// The resulting stem is `{parent stem}__{child stem}`, which is what keeps + /// a delegated worker out of every root-transcript scan while still + /// recording the delegation path in one flat filename. + /// + /// Unlike [`Self::scoped`], a child has no separate `agent_id` slot: its + /// whole identity beneath `parent` is `child_key`. Two different agent + /// implementations delegated under the same parent with the same + /// `child_key` therefore share one transcript — by design this + /// constructor puts that disambiguation on the caller, the same way + /// [`Self::root`]/[`Self::scoped`] already require `session_key` to be + /// unique per conversation. Choose a `child_key` that is unique per + /// (parent, logical sub-agent) pair — e.g. include the agent's own name + /// or role in it — rather than relying on this type to invent identity + /// it was never given. + pub fn child_of(parent: &SessionRef, child_key: impl Into) -> Self { + Self { + session_key: child_key.into(), + agent_id: None, + generation: 0, + parent_stem: Some(bounded_parent_stem(parent)), + } + } + + /// Generation 0 of this session — the chain's first segment. + pub fn first_generation(&self) -> Self { + Self { + generation: 0, + ..self.clone() + } + } + + /// The successor this session's next compaction writes into. + pub fn next_generation(&self) -> Self { + Self { + generation: self.generation.saturating_add(1), + ..self.clone() + } + } + + /// Whether this session is a delegated sub-agent rather than a root. + pub fn is_subagent(&self) -> bool { + self.parent_stem.is_some() + } + + /// The session id recorded in `_meta.session_id` — the stem, which is the + /// one name that is unique per generation and stable across processes. + pub fn session_id(&self) -> String { + session_stem(self) + } + + /// The session id of the generation this one succeeded, if any. + pub fn parent_session_id(&self) -> Option { + (self.generation > 0).then(|| { + session_stem(&Self { + generation: self.generation - 1, + ..self.clone() + }) + }) + } +} + +/// The transcript stem for `session`: deterministic, filesystem-safe, and +/// **free of any timestamp**. That absence is the point — it is what makes one +/// conversation resolve to one file across restarts and across processes. +/// +/// Shape: `{key}` for a root generation 0, `{key}.{agent}` when scoped, +/// `.g{n}` appended from generation 1, and the whole thing prefixed with +/// `{parent}__` for a sub-agent. +pub fn session_stem(session: &SessionRef) -> String { + let mut stem = sanitize_component(&session.session_key); + if let Some(agent_id) = session.agent_id.as_deref() { + // Emitted whenever `agent_id.is_some()`, blank or not: omitting a + // blank/whitespace-only agent component entirely made + // `SessionRef::scoped(key, "")` encode identically to + // `SessionRef::root(key)`, sharing one transcript between what are, + // by construction (`scoped` vs `root`), two distinct identities. + // `sanitize_component`'s own per-component digest is what actually + // keeps this collision-free even when the sanitized text is empty: + // it digests the raw (possibly blank) value, not the sanitized one. + stem.push('.'); + stem.push_str(&sanitize_component(agent_id)); + } + if session.generation > 0 { + stem.push_str(&format!(".g{}", session.generation)); + } + match session.parent_stem.as_deref() { + Some(parent) => format!("{parent}{SUBAGENT_SEPARATOR}{stem}"), + None => stem, + } +} + +/// Longest human-readable prefix kept before the disambiguating digest. +/// Bounds every component (and therefore the filenames built from it) well +/// under common filesystem name limits (255 bytes), even after a `.g{n}` +/// suffix, an agent id, and a chain of `__`-joined sub-agent ancestors. +const MAX_COMPONENT_PREFIX: usize = 80; + +/// Longest parent-chain prefix a [`SessionRef::child_of`] call keeps +/// verbatim before collapsing it into a digest. +/// +/// [`MAX_COMPONENT_PREFIX`] bounds one component, but `parent_stem` is +/// already the *entire* ancestor chain, and each further `child_of` call +/// concatenates onto it without bound — a delegation several levels deep, +/// each level with a long key, would otherwise grow the final stem past +/// filesystem name limits. Once the chain built so far exceeds this bound, +/// [`bounded_parent_stem`] replaces it with a short digest instead of +/// continuing to grow linearly with depth. +/// +/// The bound has to keep *every* level's own resolved stem — not just the +/// stored `parent_stem` — under common filesystem name limits (255 bytes), +/// because collapsing only the *stored* chain does nothing for the level +/// that is about to be built from it. Worst-case arithmetic, in bytes (every +/// byte in a sanitized component is exactly one ASCII byte, so char counts +/// and byte counts coincide throughout this module): +/// +/// - One component's maximum width is +/// `MAX_COMPONENT_PREFIX` (80) + 1 (`-`) + 32 (hex digest) = 113. +/// - A root's own stem (`session_key`, optionally `.{agent_id}`, optionally +/// `.g{n}`) is at most `113 + 1 + 113 + 6 = 233` — under 255 on its own, +/// with no parent chain to add. (`child_of` never sets `agent_id`, so only +/// the root can carry that second component.) +/// - A non-root level's own contribution (`session_key` plus an optional +/// `.g{n}`) is at most `113 + 6 = 119`. +/// - For a child's *total* resolved stem (`{parent}__{own}`) to stay safely +/// under 255 (240, leaving headroom for the `.jsonl` extension and this +/// arithmetic's own margin), the stored `parent_stem` handed to it must be +/// at most `240 - 2 (__) - 119 = 119`. +/// +/// 110 is chosen comfortably inside that headroom. Any level whose own +/// resolved stem would exceed 110 (which a root with even a moderately long +/// key already does, at 113+) collapses to the ~38-byte `chain-{32 hex}` +/// digest before being handed to its child, so no level's total ever +/// exceeds `233` (an unparented root) or `38 + 2 + 119 = 159` (every +/// subsequent, digest-parented level) — both comfortably under 255. +/// Ordinary shallow delegation with short, human keys (the common case, see +/// `a_shallow_delegation_chain_is_unaffected_by_the_bound`) stays far below +/// 110 at every level and is never collapsed at all. +const MAX_PARENT_CHAIN_PREFIX: usize = 110; + +/// Separator between a component's human-readable prefix and its +/// disambiguating digest. Must be a character [`sanitize_stem`] itself +/// already allows through unchanged (alphanumeric, `_`, `-`, `.`): +/// [`resolve_keyed_transcript_path`](super::paths::resolve_keyed_transcript_path) +/// re-sanitizes the whole stem this function builds before it ever becomes a +/// filename, and a separator outside that set would silently get replaced +/// with `_` at that second pass — reintroducing exactly the alias this +/// function exists to prevent. +const DIGEST_SEPARATOR: char = '-'; + +/// One component of a stem: path-safe, bounded in length, and encoded so +/// that no two *different* raw values can ever collide on the same +/// filename — including collisions introduced by the sanitization itself. +/// +/// Two lossy transforms are needed to keep [`SUBAGENT_SEPARATOR`] and the +/// `.` separators (agent id, `.g{n}` generation suffix) unambiguous: +/// * runs of `_` are collapsed, so a component can never reproduce +/// `__` and be mistaken for the sub-agent separator; and +/// * `.` is replaced with `-`, so a literal `.` in a raw component can +/// never be mistaken for the reserved agent/generation separator. +/// +/// Both are lossy: distinct raw values (`a_b` vs `a__b`, `t.a` vs a `t` root +/// scoped to agent `a`, `thread-1.g1` vs `thread-1`'s next generation) could +/// otherwise sanitize to the *same* text and silently share one transcript. +/// A short deterministic digest of the untouched raw value is appended to +/// rule that out: two components produce the same encoded stem only when +/// their raw values are identical. +/// +/// The digest is [`fnv1a64`], not `std::collections::hash_map::DefaultHasher`: +/// the standard library explicitly documents `DefaultHasher`'s algorithm as +/// unspecified and subject to change between Rust releases. A durable +/// identity that is supposed to "re-derive the same stem forever" cannot be +/// built on a hash the language is free to change out from under it — a +/// toolchain upgrade would silently re-derive different filenames for every +/// existing conversation. FNV-1a's definition is fixed arithmetic with no +/// language- or library-level discretion, so it carries the same forever +/// guarantee the rest of this module's "no timestamp, no randomness" design +/// already relies on. +fn sanitize_component(value: &str) -> String { + let sanitized = sanitize_stem(value); + let mut out = String::with_capacity(sanitized.len().min(MAX_COMPONENT_PREFIX)); + let mut kept = 0usize; + for ch in sanitized.chars() { + if kept >= MAX_COMPONENT_PREFIX { + break; + } + // `.` is reserved for the agent-id and generation separators. + let ch = if ch == '.' { '-' } else { ch }; + if ch == '_' && out.ends_with('_') { + continue; + } + out.push(ch); + kept += 1; + } + + out.push(DIGEST_SEPARATOR); + out.push_str(&format!("{:032x}", fnv1a128(value.as_bytes()))); + out +} + +/// FNV-1a, 64-bit variant: a small, fully-specified, non-cryptographic hash +/// with no algorithmic discretion left to a library or language version — see +/// [`sanitize_component`] for why that fixedness is the point. Operates on +/// bytes rather than `str::hash`, so it does not depend on +/// [`std::hash::Hash`]'s own algorithm-agnostic contract either. +/// +/// `offset_basis` is exposed (rather than hardcoded to FNV's own published +/// constant) so [`fnv1a128`] can run this twice with two different, still +/// fully-specified seeds and combine the results — a single 64-bit digest +/// alone only resists *accidental* collision (host-generated identifiers +/// such as thread ids colliding by chance, astronomically unlikely at 64 +/// bits); it does not resist a party that can choose the raw session key and +/// deliberately search for two values with the same digest, which is +/// tractable against a fast, non-cryptographic 64-bit hash. This module +/// makes no claim of cryptographic collision resistance either way — that +/// would need a real cryptographic hash and a dependency this small crate +/// does not otherwise need — but 128 bits raises the deliberate-search cost +/// enough that it is no longer a practical concern for a value that, per +/// `SessionRef`'s own doc, is "the host's stable name for the conversation" +/// rather than fully attacker-chosen bytes. +fn fnv1a64(bytes: &[u8], offset_basis: u64) -> u64 { + const PRIME: u64 = 0x0000_0100_0000_01b3; + + let mut hash = offset_basis; + for &byte in bytes { + hash ^= u64::from(byte); + hash = hash.wrapping_mul(PRIME); + } + hash +} + +/// 128-bit digest: two independent [`fnv1a64`] passes over the same bytes +/// with two different fixed seeds, concatenated. See [`fnv1a64`]'s doc for +/// why one 64-bit pass alone is not enough. +fn fnv1a128(bytes: &[u8]) -> u128 { + // FNV's own published 64-bit offset basis, and a second, arbitrary but + // fixed 64-bit constant (this crate's version-independent "no discretion + // left to change" requirement only demands that whatever is used is + // fixed forever, not that it hold any particular value). + const OFFSET_BASIS_A: u64 = 0xcbf2_9ce4_8422_2325; + const OFFSET_BASIS_B: u64 = 0x9E37_79B9_7F4A_7C15; + + let high = fnv1a64(bytes, OFFSET_BASIS_A); + let low = fnv1a64(bytes, OFFSET_BASIS_B); + (u128::from(high) << 64) | u128::from(low) +} + +/// `parent`'s stem, bounded to [`MAX_PARENT_CHAIN_PREFIX`]: verbatim when +/// short enough, otherwise collapsed to a fixed-length digest. See +/// [`MAX_PARENT_CHAIN_PREFIX`] for why this exists. +/// +/// Never introduces (or removes) a [`SUBAGENT_SEPARATOR`]: the replacement +/// is a whole new component substituted for the whole prior chain, not a +/// truncation of it — truncating the chain string directly could cut +/// through an existing `__` and either fabricate one at a new position or +/// destroy the one recording a real ancestor boundary. Composed of only +/// alphanumerics and `-`, so it is stable under [`sanitize_component`]'s own +/// second pass same as every other component. +fn bounded_parent_stem(parent: &SessionRef) -> String { + let stem = session_stem(parent); + if stem.len() <= MAX_PARENT_CHAIN_PREFIX { + return stem; + } + format!("chain{DIGEST_SEPARATOR}{:032x}", fnv1a128(stem.as_bytes())) +} + +#[cfg(test)] +#[path = "session_test.rs"] +mod test; diff --git a/crates/tinyagents-session/src/transcript/session_test.rs b/crates/tinyagents-session/src/transcript/session_test.rs new file mode 100644 index 000000000..28beb5d1f --- /dev/null +++ b/crates/tinyagents-session/src/transcript/session_test.rs @@ -0,0 +1,292 @@ +use super::{SUBAGENT_SEPARATOR, SessionRef, session_stem}; + +#[test] +fn a_stem_is_deterministic_and_carries_no_timestamp() { + let session = SessionRef::scoped("thread-9fa08c44", "orchestrator"); + let first = session_stem(&session); + let second = session_stem(&SessionRef::scoped("thread-9fa08c44", "orchestrator")); + + assert_eq!(first, second); + assert!(first.starts_with("thread-9fa08c44-")); + assert!(first.contains(".orchestrator-")); + // The whole point: no `{unix_ts}_` prefix, so nothing varies per launch. + assert!( + !first + .split(['_', '.']) + .next() + .is_some_and(|head| { head.len() >= 10 && head.chars().all(|c| c.is_ascii_digit()) }), + "{first} still looks timestamp-prefixed" + ); +} + +#[test] +fn two_agents_on_one_key_get_distinct_stems() { + let left = session_stem(&SessionRef::scoped("thread-1", "orchestrator")); + let right = session_stem(&SessionRef::scoped("thread-1", "researcher")); + + assert_ne!(left, right); +} + +#[test] +fn an_unscoped_root_is_just_the_key_plus_a_digest() { + let stem = session_stem(&SessionRef::root("thread-1")); + assert!(stem.starts_with("thread-1-")); +} + +#[test] +fn a_scoped_session_with_a_blank_agent_id_is_still_distinct_from_an_unscoped_root() { + // `scoped` and `root` are two different constructors precisely because + // they name two different identities; a blank/whitespace agent id must + // not silently make `scoped` collapse into `root`'s stem (it used to, + // when the agent component was omitted entirely for a blank value). + let scoped_blank = SessionRef::scoped("thread-1", " "); + let root = SessionRef::root("thread-1"); + assert_ne!(session_stem(&scoped_blank), session_stem(&root)); + + // Two distinct blank/empty agent ids must also stay distinct from each + // other, since the digest covers the raw (pre-sanitization) value. + let scoped_empty = SessionRef::scoped("thread-1", ""); + assert_ne!(session_stem(&scoped_blank), session_stem(&scoped_empty)); +} + +#[test] +fn path_traversal_in_a_key_cannot_escape_the_transcript_directory() { + // `.` no longer survives sanitization — it is reserved for the agent and + // generation separators — but the traversal characters it used to leave + // behind must still never produce a path separator. + let stem = session_stem(&SessionRef::root("../../etc/passwd")); + assert!(!stem.contains('/'), "{stem}"); + assert!(!stem.contains('\\'), "{stem}"); + assert!(!stem.contains('.'), "{stem} still contains a literal '.'"); +} + +#[test] +fn generations_are_distinct_and_ordered_by_suffix() { + let first = SessionRef::scoped("thread-1", "orchestrator"); + let second = first.next_generation(); + let third = second.next_generation(); + + let first_stem = session_stem(&first); + let second_stem = session_stem(&second); + let third_stem = session_stem(&third); + + assert!(second_stem.starts_with(&format!("{first_stem}.g1"))); + assert!(third_stem.starts_with(&format!("{first_stem}.g2"))); + assert_ne!(first_stem, second_stem); + assert_ne!(second_stem, third_stem); +} + +#[test] +fn a_generation_knows_the_one_it_succeeded() { + let first = SessionRef::scoped("thread-1", "orchestrator"); + let second = first.next_generation(); + + assert_eq!(first.parent_session_id(), None); + assert_eq!( + second.parent_session_id().as_deref(), + Some(session_stem(&first).as_str()) + ); + assert_eq!(second.session_id(), session_stem(&second)); +} + +#[test] +fn a_subagent_stem_carries_the_separator_every_root_scan_filters_on() { + let parent = SessionRef::scoped("thread-1", "orchestrator"); + let child = SessionRef::child_of(&parent, "worker-7"); + + let stem = session_stem(&child); + assert!(stem.contains(SUBAGENT_SEPARATOR)); + assert!(child.is_subagent()); + assert!(!parent.is_subagent()); +} + +#[test] +fn a_root_stem_never_contains_the_subagent_separator() { + // Root scans treat `__` as "this is a delegated worker", so a root stem + // that contained it would vanish from thread lookup entirely. + for key in ["thread-1", "a__b", "with space", "../escape"] { + let stem = session_stem(&SessionRef::scoped(key, "agent__name")); + assert!( + !stem.contains(SUBAGENT_SEPARATOR), + "root stem {stem:?} from key {key:?} looks like a sub-agent" + ); + } +} + +#[test] +fn nested_delegation_records_the_whole_path_in_one_flat_stem() { + // Short keys: with a 32-hex-char digest on every component, even + // moderately descriptive names ("orchestrator", "researcher") push a + // 2-3 level chain's own resolved stem past `MAX_PARENT_CHAIN_PREFIX` — + // intentionally, since that bound exists to keep the *filename* safe, + // not to guarantee unlimited readability. This test is about the flat + // "__"-joined shape surviving when the chain stays short enough not to + // collapse; `a_deeply_nested_delegation_chain_stays_bounded` covers the + // collapse itself. + let root = SessionRef::scoped("t1", "o"); + let child = SessionRef::child_of(&root, "r1"); + let grandchild = SessionRef::child_of(&child, "r2"); + + let stem = session_stem(&grandchild); + assert_eq!(stem.matches(SUBAGENT_SEPARATOR).count(), 2); + assert!(stem.starts_with(&session_stem(&root))); +} + +// ---- Collision-resistance regressions ----------------------------------- +// +// Every case here is a raw input pair that the pre-digest sanitizer mapped +// to the *same* filename, letting two distinct conversations read and +// overwrite each other's transcript. + +#[test] +fn collapsing_underscore_runs_no_longer_aliases_distinct_keys() { + let a = session_stem(&SessionRef::root("a_b")); + let b = session_stem(&SessionRef::root("a__b")); + assert_ne!(a, b); +} + +#[test] +fn a_literal_dot_in_a_key_no_longer_aliases_the_generation_suffix() { + let literal_dot = session_stem(&SessionRef::root("thread-1.g1")); + let real_generation = session_stem(&SessionRef::root("thread-1").next_generation()); + assert_ne!(literal_dot, real_generation); +} + +#[test] +fn a_literal_dot_in_a_key_no_longer_aliases_the_agent_separator() { + let unscoped_with_dot = session_stem(&SessionRef::root("t.a")); + let scoped = session_stem(&SessionRef::scoped("t", "a")); + assert_ne!(unscoped_with_dot, scoped); +} + +#[test] +fn a_very_long_key_still_produces_a_filesystem_safe_stem() { + let long_key = "k".repeat(400); + let stem = session_stem(&SessionRef::scoped(&long_key, "agent")); + // Comfortably under common filesystem name limits (255 bytes) even after + // `session_raw/{stem}.jsonl` and an agent id/generation suffix. + assert!(stem.len() < 200, "{} bytes: {stem}", stem.len()); +} + +#[test] +fn two_long_keys_that_share_a_bounded_prefix_still_get_distinct_stems() { + let base = "k".repeat(400); + let a = session_stem(&SessionRef::root(&base)); + let b = session_stem(&SessionRef::root(format!("{base}-tail"))); // differs past the bound + assert_ne!(a, b); +} + +/// Pins the exact digest algorithm and its output, not just "some digest". +/// A durable filename must "re-derive the same stem forever" — swapping the +/// hash (or its parameters) is exactly the kind of change that must be +/// caught here rather than silently shipped, because it would re-derive a +/// different filename for every session ever written. +#[test] +fn the_digest_algorithm_is_pinned_to_known_fnv1a64_outputs() { + assert_eq!( + super::fnv1a64(b"", 0xcbf2_9ce4_8422_2325), + 0xcbf2_9ce4_8422_2325 + ); + assert_eq!( + super::fnv1a64(b"a", 0xcbf2_9ce4_8422_2325), + 0xaf63_dc4c_8601_ec8c + ); + assert_eq!( + super::fnv1a64(b"thread-9fa08c44", 0xcbf2_9ce4_8422_2325), + 0xdf74_ac18_4530_bb17 + ); + // fnv1a128 combines two 64-bit passes with two different fixed seeds — + // pin the combined output too, since a future change that widened or + // reseeded only one pass would otherwise slip past the check above. + assert_eq!( + super::fnv1a128(b""), + 0xcbf2_9ce4_8422_2325_9e37_79b9_7f4a_7c15 + ); + assert_eq!( + super::fnv1a128(b"a"), + 0xaf63_dc4c_8601_ec8c_22c0_4a33_4b91_791c + ); + assert_eq!( + super::fnv1a128(b"thread-9fa08c44"), + 0xdf74_ac18_4530_bb17_1e90_1a3d_7200_1847 + ); +} + +/// Pins the exact worst-case arithmetic documented on +/// `MAX_PARENT_CHAIN_PREFIX`: every level of a maximally-sized chain (a +/// maximal `session_key`, a maximal `agent_id` on the root, and an existing +/// compaction at every level) must stay under the 255-byte filesystem name +/// limit, whether or not that level's own resolved stem was itself the +/// trigger for a collapse one level up. +#[test] +fn every_level_of_a_maximal_chain_stays_under_the_filesystem_limit() { + // The worst single (unparented) level: a maximal session_key, a maximal + // agent_id, and an existing compaction (`.g{n}`) — see + // `MAX_PARENT_CHAIN_PREFIX`'s doc for the arithmetic this pins. + let max_key = "k".repeat(200); + let root = SessionRef::scoped(&max_key, &max_key).next_generation(); + let root_stem = session_stem(&root); + assert!( + root_stem.len() < 255, + "an unparented root must fit on its own: {} bytes", + root_stem.len() + ); + + // The worst parented level: a child (never has an agent_id) of that same + // maximal root, itself also compacted. + let child = SessionRef::child_of(&root, max_key.clone()).next_generation(); + let child_stem = session_stem(&child); + assert!( + child_stem.len() < 255, + "a child of a maximal root must still fit: {} bytes", + child_stem.len() + ); + + // And the level after that, to confirm the collapse repeats rather than + // the bound only holding for one transition. + let grandchild = SessionRef::child_of(&child, max_key).next_generation(); + let grandchild_stem = session_stem(&grandchild); + assert!( + grandchild_stem.len() < 255, + "a grandchild must still fit: {} bytes", + grandchild_stem.len() + ); +} + +/// A delegation chain several levels deep, each level with a long key, must +/// not grow the final stem without bound — see `MAX_PARENT_CHAIN_PREFIX`. +/// 50 levels of ~113-byte components would be several KB uncollapsed, +/// comfortably past any sane bound; this pins that the collapse actually +/// keeps growth from compounding rather than merely being "small enough in +/// this one example". +#[test] +fn a_deeply_nested_delegation_chain_stays_bounded() { + let long_key = "k".repeat(80); + let mut current = SessionRef::scoped(&long_key, "orchestrator"); + for level in 0..50 { + current = SessionRef::child_of(¤t, format!("{long_key}-{level}")); + } + let stem = session_stem(¤t); + assert!( + stem.len() < 700, + "50 levels of long keys must not grow the stem past the collapse \ + bound: {} bytes", + stem.len() + ); + // Still never fabricates or destroys the sub-agent separator. + assert!(stem.contains(SUBAGENT_SEPARATOR)); +} + +/// Bounding the chain must never change what a *shallow* delegation's stem +/// looks like — the collapse only kicks in once the accumulated chain +/// actually exceeds the bound. +#[test] +fn a_shallow_delegation_chain_is_unaffected_by_the_bound() { + let root = SessionRef::scoped("t1", "o"); + let child = SessionRef::child_of(&root, "r1"); + let grandchild = SessionRef::child_of(&child, "r2"); + + let stem = session_stem(&grandchild); + assert!(!stem.contains("chain-"), "{stem} collapsed too early"); + assert_eq!(stem.matches(SUBAGENT_SEPARATOR).count(), 2); +} diff --git a/crates/tinyagents-session/src/transcript/test.rs b/crates/tinyagents-session/src/transcript/test.rs index 69e557066..0a0f0d769 100644 --- a/crates/tinyagents-session/src/transcript/test.rs +++ b/crates/tinyagents-session/src/transcript/test.rs @@ -6,10 +6,13 @@ //! Consolidated here per AGENTS.md: one `test.rs` per module directory. use super::*; +use std::sync::{Arc, Barrier}; use tempfile::tempdir; fn meta() -> TranscriptMeta { TranscriptMeta { + session_id: None, + parent_session_id: None, agent_name: "agent".into(), agent_id: Some("agent-id".into()), agent_type: Some("root".into()), @@ -240,3 +243,386 @@ fn discovery_and_legacy_read_replay_the_canonical_format() { "legacy body" ); } + +// ── Session identity ────────────────────────────────────────────────── + +/// One conversation, two cold sessions: the second must find the first's file +/// rather than mint a second stem for the same thread. This is the regression +/// that cost a real user the opening turns of a thread. +#[test] +fn one_session_resolves_to_one_transcript_across_separate_bindings() { + let dir = tempdir().unwrap(); + let session = SessionRef::scoped("thread-9fa08", "orchestrator"); + + let first = FileTranscriptLocator::new(dir.path()) + .open_session(&session, meta()) + .unwrap(); + first + .append(TranscriptMessage::new( + "user", + "i want to plan a trip to kashmir", + )) + .unwrap(); + + // A brand-new locator and handle, as a restarted process would build. + let second = FileTranscriptLocator::new(dir.path()) + .open_session(&session, meta()) + .unwrap(); + second + .append(TranscriptMessage::new("user", "hello?")) + .unwrap(); + + assert_eq!(first.path(), second.path()); + let messages = second.messages().unwrap(); + assert_eq!(messages.len(), 2); + assert_eq!(messages[0].content, "i want to plan a trip to kashmir"); + assert_eq!(messages[1].content, "hello?"); + + let roots: Vec<_> = std::fs::read_dir(dir.path().join("session_raw")) + .unwrap() + .flatten() + .map(|entry| entry.file_name()) + .collect(); + assert_eq!( + roots.len(), + 1, + "one conversation must not sprawl: {roots:?}" + ); +} + +#[test] +fn an_unwritten_session_reads_as_absent_rather_than_erroring() { + let dir = tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + let session = SessionRef::scoped("thread-new", "orchestrator"); + + assert!(locator.read_session_transcript(&session).is_none()); + assert!(!locator.session_exists(&session)); + assert_eq!(locator.head_generation(&session), session); +} + +/// A directory occupying a session's canonical `.jsonl` path must never be +/// treated as an existing generation: reads/appends against it would fail +/// (or silently target the wrong thing), and `head_generation`'s chain walk +/// would otherwise stop at a phantom "generation" that was never written. +#[test] +fn a_directory_at_the_canonical_path_is_never_treated_as_an_existing_session() { + let dir = tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + let session = SessionRef::scoped("thread-1", "orchestrator"); + + let path = resolve_keyed_transcript_path(dir.path(), &session_stem(&session)).unwrap(); + std::fs::create_dir_all(&path).unwrap(); + + assert!(!locator.session_exists(&session)); + assert!(locator.read_session_transcript(&session).is_none()); +} + +#[test] +fn session_identity_round_trips_through_the_jsonl_meta() { + let dir = tempdir().unwrap(); + let path = resolve_keyed_transcript_path(dir.path(), "identity").unwrap(); + let mut written = meta(); + written.session_id = Some("thread-1.orchestrator.g1".into()); + written.parent_session_id = Some("thread-1.orchestrator".into()); + write_transcript( + &path, + &[TranscriptMessage::new("user", "hi")], + &written, + None, + ) + .unwrap(); + + let read = read_transcript(&path).unwrap(); + assert_eq!( + read.meta.session_id.as_deref(), + Some("thread-1.orchestrator.g1") + ); + assert_eq!( + read.meta.parent_session_id.as_deref(), + Some("thread-1.orchestrator") + ); +} + +/// A compaction must never destroy what it replaces. It seals the current +/// generation and opens the next, so the replaced turns stay on disk. +#[test] +fn a_compaction_seals_a_generation_and_leaves_it_untouched() { + let dir = tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + let session = SessionRef::scoped("thread-1", "orchestrator"); + + let first = locator.open_session(&session, meta()).unwrap(); + for turn in ["one", "two", "three"] { + first.append(TranscriptMessage::new("user", turn)).unwrap(); + } + let sealed_path = first.path().to_path_buf(); + let sealed_bytes = std::fs::read(&sealed_path).unwrap(); + + let (successor, handle) = locator.begin_generation(&session, meta()).unwrap(); + // The successor is bound but empty; the retained set is written through the + // ordinary turn path so usage and request ids are recorded as usual. + handle + .replace(&[TranscriptMessage::new("user", "three")]) + .unwrap(); + + assert_eq!(successor.generation, 1); + assert_eq!( + std::fs::read(&sealed_path).unwrap(), + sealed_bytes, + "the sealed generation must be byte-identical afterwards" + ); + assert_ne!(handle.path(), sealed_path); + + let carried = handle.messages().unwrap(); + assert_eq!(carried.len(), 1); + assert_eq!(carried[0].content, "three"); + + let successor_meta = handle.read_session().unwrap().unwrap().meta; + assert_eq!( + successor_meta.session_id.as_deref(), + Some(session_stem(&successor).as_str()) + ); + assert_eq!( + successor_meta.parent_session_id.as_deref(), + Some(session_stem(&session).as_str()) + ); +} + +/// After a compaction, a resume must land on the newest generation — the one +/// the model is actually continuing — not on the sealed original. +#[test] +fn head_generation_follows_the_compaction_chain() { + let dir = tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + let session = SessionRef::scoped("thread-1", "orchestrator"); + + locator + .open_session(&session, meta()) + .unwrap() + .append(TranscriptMessage::new("user", "one")) + .unwrap(); + assert_eq!(locator.head_generation(&session), session); + + let (first_successor, first_handle) = locator.begin_generation(&session, meta()).unwrap(); + first_handle + .append(TranscriptMessage::new("user", "one")) + .unwrap(); + assert_eq!(locator.head_generation(&session), first_successor); + + let (second_successor, second_handle) = + locator.begin_generation(&first_successor, meta()).unwrap(); + second_handle + .append(TranscriptMessage::new("user", "one")) + .unwrap(); + assert_eq!(locator.head_generation(&session).generation, 2); + assert_eq!(locator.head_generation(&session), second_successor); +} + +#[test] +fn opening_a_generation_that_already_exists_is_refused() { + let dir = tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + let session = SessionRef::scoped("thread-1", "orchestrator"); + + let (_, handle) = locator.begin_generation(&session, meta()).unwrap(); + handle + .append(TranscriptMessage::new("user", "one")) + .unwrap(); + let second = locator.begin_generation(&session, meta()); + + assert!( + second.is_err(), + "sealing the same generation twice would overwrite durable history" + ); +} + +/// Two racing `begin_generation` calls for the same session — two cores +/// compacting at once — must not both write their own first append into the +/// generation's still-nonexistent file: `begin_generation` itself performs no +/// I/O (`FileTranscriptHistory::new` only resolves a path), so without the +/// `path_lock` both handles' first `append_turn_with_partial` would +/// otherwise race on the writer's create-fresh branch and whichever `fs::write` +/// lands last would silently discard the other's retained set. +#[test] +fn concurrent_begin_generation_handles_for_one_session_never_lose_either_append() { + let dir = tempdir().unwrap(); + let session = FileTranscriptLocator::new(dir.path()); + let root = SessionRef::scoped("thread-1", "orchestrator"); + // Seal generation 0 so both racers are compacting into the same, + // already-known successor generation 1. + session + .open_session(&root, meta()) + .unwrap() + .append(TranscriptMessage::new("user", "sealed")) + .unwrap(); + + let locator = Arc::new(FileTranscriptLocator::new(dir.path())); + let barrier = Arc::new(Barrier::new(2)); + + let left_locator = Arc::clone(&locator); + let left_root = root.clone(); + let left_barrier = Arc::clone(&barrier); + let left = std::thread::spawn(move || { + left_barrier.wait(); + let (_, handle) = left_locator.begin_generation(&left_root, meta()).unwrap(); + handle + .append(TranscriptMessage::new("user", "from left")) + .unwrap(); + }); + + let right_locator = Arc::clone(&locator); + let right_root = root.clone(); + let right_barrier = Arc::clone(&barrier); + let right = std::thread::spawn(move || { + right_barrier.wait(); + let (_, handle) = right_locator.begin_generation(&right_root, meta()).unwrap(); + handle + .append(TranscriptMessage::new("user", "from right")) + .unwrap(); + }); + + left.join().unwrap(); + right.join().unwrap(); + + let successor = root.next_generation(); + let handle = locator.open_session(&successor, meta()).unwrap(); + let contents: Vec = handle + .messages() + .unwrap() + .into_iter() + .map(|message| message.content) + .collect(); + assert_eq!( + contents.len(), + 2, + "both racing compactions' appends must survive: {contents:?}" + ); +} + +/// Two handles on one session — the shape two cores over one workspace +/// produce — must both land in the same file, with neither losing the other's +/// turns. +#[test] +fn concurrent_handles_on_one_session_both_extend_it() { + let dir = tempdir().unwrap(); + let session = SessionRef::scoped("thread-1", "orchestrator"); + let left = FileTranscriptLocator::new(dir.path()) + .open_session(&session, meta()) + .unwrap(); + let right = FileTranscriptLocator::new(dir.path()) + .open_session(&session, meta()) + .unwrap(); + + // Seed the file first, so both appends below exercise the append-only + // path — `write_logical_set` re-reading `persisted` fresh immediately + // before every write, which is the actual claim under test — rather than + // racing on first-write file creation, a distinct, pre-existing concern + // this test is not about. + left.append(TranscriptMessage::new("user", "seed")).unwrap(); + + // Genuinely overlapping, not merely interleaved: both handles race to + // append from separate OS threads, released together by a barrier so + // neither can start before the other is ready. A handle that cached its + // own view of `persisted` instead of re-reading it fresh before every + // write could lose whichever append the barrier let land second. + let barrier = Arc::new(Barrier::new(2)); + let left_barrier = Arc::clone(&barrier); + let left_thread = std::thread::spawn(move || { + left_barrier.wait(); + left.append(TranscriptMessage::new("user", "from left")) + .unwrap(); + }); + let right_barrier = Arc::clone(&barrier); + let right_thread = std::thread::spawn(move || { + right_barrier.wait(); + right + .append(TranscriptMessage::new("user", "from right")) + .unwrap(); + }); + left_thread.join().unwrap(); + right_thread.join().unwrap(); + + let reread = FileTranscriptLocator::new(dir.path()) + .open_session(&session, meta()) + .unwrap(); + let mut contents: Vec = reread + .messages() + .unwrap() + .into_iter() + .map(|message| message.content) + .collect(); + assert_eq!(contents.remove(0), "seed"); + contents.sort(); + assert_eq!( + contents, + ["from left", "from right"], + "an overlapping append from either handle must not be lost" + ); +} + +/// The model reads only the head generation, but a host rendering or +/// exporting the conversation needs every segment, in order. +#[test] +fn a_session_chain_lists_every_generation_oldest_first() { + let dir = tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + let session = SessionRef::scoped("thread-1", "orchestrator"); + + assert!(locator.session_chain(&session).is_empty()); + + locator + .open_session(&session, meta()) + .unwrap() + .append(TranscriptMessage::new("user", "one")) + .unwrap(); + let (successor, handle) = locator.begin_generation(&session, meta()).unwrap(); + handle + .append(TranscriptMessage::new("user", "two")) + .unwrap(); + + let chain = locator.session_chain(&session); + assert_eq!(chain, vec![session.clone(), successor]); + // Asking from any generation returns the same whole chain. + assert_eq!(locator.session_chain(&chain[1]), chain); +} + +/// [`write_transcript_if_absent`] is what keeps adoption from clobbering a +/// destination a concurrent normal turn created while adoption was still +/// scanning legacy roots (see `adoption::adopt_legacy_session_transcripts`). +/// The guarantee has to hold at the level of this primitive: it must publish +/// when nothing is there, and never overwrite when something already is — +/// regardless of *why* the destination already exists. +#[test] +fn write_transcript_if_absent_publishes_once_and_never_overwrites() { + let dir = tempdir().unwrap(); + let path = resolve_keyed_transcript_path(dir.path(), "identity").unwrap(); + + let published = write_transcript_if_absent( + &path, + &[TranscriptMessage::new("user", "first writer")], + &meta(), + ) + .unwrap(); + assert!(published, "nothing was there yet"); + assert_eq!( + read_transcript(&path).unwrap().messages[0].content, + "first writer" + ); + + let published_again = write_transcript_if_absent( + &path, + &[TranscriptMessage::new( + "user", + "second writer, loses the race", + )], + &meta(), + ) + .unwrap(); + assert!(!published_again, "the destination already exists"); + // The loser's content must never have touched disk. + assert_eq!( + read_transcript(&path).unwrap().messages[0].content, + "first writer" + ); +} diff --git a/crates/tinyagents-session/src/transcript/thread_lookup.rs b/crates/tinyagents-session/src/transcript/thread_lookup.rs index c46d797e9..803fc1d10 100644 --- a/crates/tinyagents-session/src/transcript/thread_lookup.rs +++ b/crates/tinyagents-session/src/transcript/thread_lookup.rs @@ -59,30 +59,55 @@ pub fn find_root_transcript_for_thread_scoped( /// agent-id filtering; exposed directly for callers that need the full /// ordered history rather than just the latest match. pub fn find_root_transcripts_for_thread(workspace_dir: &Path, thread_id: &str) -> Vec { - let mut matches = Vec::new(); - matches.extend(root_transcripts_for_thread_in_dir( - &raw_session_dir(workspace_dir), - thread_id, - )); - matches.sort_by_cached_key(|path| { - let created = read_transcript(path) - .ok() - .map(|transcript| transcript.meta.created) - .unwrap_or_default(); - (created, path.clone()) - }); - matches + root_transcripts_for_thread_in_dir(&raw_session_dir(workspace_dir), thread_id).0 } -fn root_transcripts_for_thread_in_dir(raw_dir: &Path, thread_id: &str) -> Vec { +/// [`find_root_transcripts_for_thread`], additionally reporting whether the +/// scan hit any root `.jsonl` file it could not read at all — matching or +/// not, since a read failure happens *before* the thread-id comparison, so +/// which thread an unreadable file belonged to can never be determined. +/// +/// Exists for [`super::adoption::adopt_legacy_session_transcripts`]: that +/// caller's idempotency marker is the destination file it writes, so folding +/// only the *readable* matches and reporting success would permanently +/// strand an unreadable file's turns — the marker's existence stops every +/// later retry. [`find_root_transcripts_for_thread`] itself is used by +/// callers (thread resume, usage summaries) that already treat "unreadable" +/// as "absent" and are safe to keep doing so; only adoption's +/// once-and-only-once contract needs to know the difference. +pub fn find_root_transcripts_for_thread_reporting_unreadable( + workspace_dir: &Path, + thread_id: &str, +) -> (Vec, bool) { + root_transcripts_for_thread_in_dir(&raw_session_dir(workspace_dir), thread_id) +} + +fn root_transcripts_for_thread_in_dir(raw_dir: &Path, thread_id: &str) -> (Vec, bool) { let thread_id = thread_id.trim(); if thread_id.is_empty() { - return Vec::new(); + return (Vec::new(), false); } - let Ok(entries) = fs::read_dir(raw_dir) else { - return Vec::new(); + let entries = match fs::read_dir(raw_dir) { + Ok(entries) => entries, + // A workspace with no session_raw/ yet has genuinely adopted + // nothing — not an error. Any other failure (permissions, a + // transient I/O error) means the scan could not actually see + // whether a matching root exists, which callers relying on + // `unreadable` (adoption's idempotency contract) must not treat the + // same as "confirmed nothing here". + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return (Vec::new(), false); + } + Err(error) => { + tracing::warn!( + "[transcript] could not scan {} for thread {thread_id}: {error}", + raw_dir.display() + ); + return (Vec::new(), true); + } }; + let mut any_unreadable = false; // Keyed by `meta.created` so the order is chronological rather than // lexicographic. Modern stems are `{unix_ts}_{agent_id}` and sort the same // either way, but a legacy `{agent}_{index}` root encodes no time at all — @@ -90,35 +115,60 @@ fn root_transcripts_for_thread_in_dir(raw_dir: &Path, thread_id: &str) -> Vec = entries - .flatten() - .map(|entry| entry.path()) - .filter(|path| { - path.extension().and_then(|s| s.to_str()) == Some("jsonl") - && path - .file_stem() - .and_then(|s| s.to_str()) - .is_some_and(|stem| !stem.contains("__")) - }) - .filter_map(|path| match read_transcript(&path) { + // + // An explicit loop rather than a filter/filter_map chain: both the + // directory-entry read and the transcript read below can independently + // fail and need to set the same `any_unreadable` flag, and two closures + // cannot each hold a mutable borrow of it at once. + let mut matches: Vec<(String, PathBuf)> = Vec::new(); + for entry in entries { + let path = match entry { + Ok(entry) => entry.path(), + Err(error) => { + // An entry the directory iterator itself could not read + // (e.g. a race with concurrent deletion, a transient I/O + // error) is exactly as invisible to this scan as a file that + // failed `read_transcript` below — the `.flatten()` this + // loop replaced would have hidden it from every caller, + // including adoption's fail-closed contract. + tracing::warn!( + "[transcript] could not read a directory entry in {}: {error}", + raw_dir.display() + ); + any_unreadable = true; + continue; + } + }; + let is_candidate = path.extension().and_then(|s| s.to_str()) == Some("jsonl") + && path + .file_stem() + .and_then(|s| s.to_str()) + .is_some_and(|stem| !stem.contains("__")); + if !is_candidate { + continue; + } + match read_transcript(&path) { Ok(transcript) if transcript.meta.thread_id.as_deref() == Some(thread_id) => { - Some((transcript.meta.created.clone(), path)) + matches.push((transcript.meta.created.clone(), path)); } - Ok(_) => None, + Ok(_) => {} Err(err) => { tracing::warn!( "[transcript] skipping unreadable root transcript candidate {}: {err}", path.display() ); - None + any_unreadable = true; } - }) - .collect(); + } + } // Path is the tiebreak so the order stays total and deterministic when two // transcripts share a `created` stamp. matches.sort_by(|left, right| left.0.cmp(&right.0).then_with(|| left.1.cmp(&right.1))); - matches.into_iter().map(|(_, path)| path).collect() + ( + matches.into_iter().map(|(_, path)| path).collect(), + any_unreadable, + ) } /// Summed token/cost usage for `thread_id` across its root transcripts, or diff --git a/crates/tinyagents-session/src/transcript/types.rs b/crates/tinyagents-session/src/transcript/types.rs index 32c404406..f198f5175 100644 --- a/crates/tinyagents-session/src/transcript/types.rs +++ b/crates/tinyagents-session/src/transcript/types.rs @@ -169,6 +169,19 @@ pub struct TranscriptMeta { /// Sub-agent task id, when this transcript belongs to a spawned worker. #[serde(default, skip_serializing_if = "Option::is_none")] pub task_id: Option, + /// Durable identity of the session this transcript holds, as produced by + /// [`session_stem`](crate::transcript::session_stem). Before this existed + /// a transcript's only identity was its filename, so nothing could tell + /// two files of one conversation apart from two unrelated ones. `None` on + /// transcripts written before session identity landed. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_id: Option, + /// The session this one succeeded when a compaction sealed it. Compaction + /// never rewrites history in place: it opens the next generation and + /// points back here, so the whole conversation stays recoverable by + /// walking the chain even though the model only sees the head. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_session_id: Option, } /// A parsed session transcript: metadata + exact message array. diff --git a/crates/tinyagents-session/src/transcript/writer.rs b/crates/tinyagents-session/src/transcript/writer.rs index 56ac46d9d..9c8e60bdf 100644 --- a/crates/tinyagents-session/src/transcript/writer.rs +++ b/crates/tinyagents-session/src/transcript/writer.rs @@ -14,7 +14,8 @@ use super::types::{TranscriptMeta, TurnUsage}; use anyhow::{Context, Result}; use std::collections::HashMap; use std::fs; -use std::path::Path; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; /// Write JSONL as source of truth **and** re-render the companion `.md`. /// @@ -40,7 +41,7 @@ pub fn write_transcript( jsonl_buf.push('\n'); serialise_message_lines(messages, last_assistant_turn_usage, None, &mut jsonl_buf)?; - fs::write(jsonl_path, jsonl_buf.as_bytes()) + atomic_write(jsonl_path, jsonl_buf.as_bytes()) .with_context(|| format!("write transcript {}", jsonl_path.display()))?; tracing::debug!( @@ -53,6 +54,46 @@ pub fn write_transcript( Ok(()) } +/// Like [`write_transcript`], but never overwrites an existing destination — +/// see [`publish_transcript_if_absent`] for why adoption needs that instead +/// of the full-rewrite semantics every other `write_transcript` caller +/// wants. Returns `Ok(true)` when this call created `jsonl_path`, `Ok(false)` +/// when it already existed (some other write already won the race and this +/// call's `messages`/`meta` were discarded). +pub fn write_transcript_if_absent( + jsonl_path: &Path, + messages: &[TranscriptMessage], + meta: &TranscriptMeta, +) -> Result { + if let Some(parent) = jsonl_path.parent() { + fs::create_dir_all(parent) + .with_context(|| format!("create transcript dir {}", parent.display()))?; + } + + let mut jsonl_buf = String::new(); + jsonl_buf.push_str(&meta_line_json(meta)?); + jsonl_buf.push('\n'); + serialise_message_lines(messages, None, None, &mut jsonl_buf)?; + + let published = publish_transcript_if_absent(jsonl_path, jsonl_buf.as_bytes()) + .with_context(|| format!("publish transcript {}", jsonl_path.display()))?; + + if published { + tracing::debug!( + "[transcript] published {} messages (jsonl, create-if-absent) to {}", + messages.len(), + jsonl_path.display() + ); + render_md_companion(jsonl_path, messages, meta, None); + } else { + tracing::debug!( + "[transcript] create-if-absent lost the race, {} already exists", + jsonl_path.display() + ); + } + Ok(published) +} + /// Append this turn's delta to an **append-only** transcript, never rewriting /// existing lines. /// @@ -277,6 +318,130 @@ fn common_prefix_len(a: &[TranscriptMessage], b: &[TranscriptMessage]) -> usize .count() } +/// A same-directory temp path for `path`, unique per call within this +/// process. Shared by [`atomic_write`] and [`publish_transcript_if_absent`], +/// both of which stage full contents in a temp file before publishing it +/// with one atomic filesystem operation. +fn unique_tmp_path(path: &Path) -> PathBuf { + static NONCE: AtomicU64 = AtomicU64::new(0); + + let dir = path.parent().unwrap_or_else(|| Path::new(".")); + let file_name = path + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("transcript"); + let nonce = NONCE.fetch_add(1, Ordering::Relaxed); + dir.join(format!(".{file_name}.tmp-{}-{nonce}", std::process::id())) +} + +/// Writes `contents` to `path` via a same-directory temp file and an atomic +/// rename, rather than truncating `path` in place. +/// +/// [`write_transcript`] is a full rewrite of the source-of-truth JSONL — used +/// directly by migrations, sub-agent runners, and (through +/// [`publish_transcript_if_absent`]) adoption. A plain `fs::write` truncates +/// the destination before the new bytes land, so a process or filesystem +/// failure partway through leaves a truncated file that nonetheless +/// satisfies `path.exists()`. Writing to a temp file first and renaming it +/// into place means the destination only ever transitions from "absent" +/// straight to "complete"; there is no truncated intermediate state a crash +/// can strand callers on. `fs::rename` always **replaces** an existing +/// destination on both Unix and Windows (`MoveFileExW` with +/// `MOVEFILE_REPLACE_EXISTING`, with a `SetFileInformationByHandle` fallback +/// — see the `std::fs::rename` docs), which is exactly the full-rewrite +/// semantics this function's other callers want. +/// Writes `contents` to a fresh temp file at `tmp_path`, refusing to follow +/// (and so overwrite the target of) any pre-existing filesystem entry — +/// including a symlink — already at that path. +/// +/// `unique_tmp_path` mints a name unique to this process and call, so this +/// should never race with a legitimate temp file of ours; a party able to +/// pre-create an entry at the exact predicted name is exactly the case this +/// guards against. `fs::write` alone would instead follow a pre-planted +/// symlink and write our transcript content through it into whatever the +/// symlink points at — a party with write access to this directory could aim +/// that at a file elsewhere the process can write but should not overwrite. +/// `create_new` fails instead, atomically, without ever opening whatever was +/// really there. +fn write_temp_file(tmp_path: &Path, contents: &[u8]) -> Result<()> { + use std::io::Write; + + let mut file = fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(tmp_path) + .with_context(|| format!("create temp transcript {}", tmp_path.display()))?; + file.write_all(contents) + .with_context(|| format!("write temp transcript {}", tmp_path.display()))?; + Ok(()) +} + +fn atomic_write(path: &Path, contents: &[u8]) -> Result<()> { + let tmp_path = unique_tmp_path(path); + + if let Err(error) = write_temp_file(&tmp_path, contents) { + // A failure after `create_new` succeeded (a full disk, a signal + // interruption partway through `write_all`) leaves the temp file + // behind; nothing else ever looks for or cleans up a name only this + // call ever mints, so it would otherwise orphan forever. A failure + // from `create_new` itself (the guarded case above) means there is + // nothing of ours to clean up. + let _ = fs::remove_file(&tmp_path); + return Err(error); + } + fs::rename(&tmp_path, path).with_context(|| { + let _ = fs::remove_file(&tmp_path); + format!( + "rename temp transcript {} to {}", + tmp_path.display(), + path.display() + ) + })?; + Ok(()) +} + +/// Publishes `contents` at `path` only if nothing is there yet, atomically. +/// +/// Unlike [`atomic_write`] (and [`write_transcript`], which uses it), +/// **never overwrites an existing destination**. `fs::rename` cannot express +/// "fail if the destination exists" — as documented on [`atomic_write`], it +/// always replaces on both platforms — so this uses `fs::hard_link` instead, +/// which fails with `AlreadyExists` without touching whatever is already at +/// `path`. That failure is reported by returning `Ok(false)` rather than an +/// error: it means some other write legitimately won the race, not that +/// anything went wrong. +/// +/// Adoption is this function's one caller and the reason it exists: its +/// destination is the session's very first transcript, and a session's own +/// normal turn persistence can independently create that same file at any +/// point during adoption's scan. Adoption must publish only if it still +/// holds the honor of "first write" when it finishes — never clobber a +/// conversation's genuine first turn with an adoption fold that started +/// scanning before that turn existed. +fn publish_transcript_if_absent(path: &Path, contents: &[u8]) -> Result { + let tmp_path = unique_tmp_path(path); + + if let Err(error) = write_temp_file(&tmp_path, contents) { + // Same orphaned-temp-file / symlink hazard as `atomic_write` — see + // `write_temp_file`'s comment. + let _ = fs::remove_file(&tmp_path); + return Err(error); + } + let published = match fs::hard_link(&tmp_path, path) { + Ok(()) => true, + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => false, + Err(error) => { + let _ = fs::remove_file(&tmp_path); + return Err(error).with_context(|| format!("publish transcript {}", path.display())); + } + }; + // The temp file and its hard-linked destination share one inode; once + // linked (or once we know we lost the race), the temp name itself has + // no further purpose. + let _ = fs::remove_file(&tmp_path); + Ok(published) +} + /// Append raw bytes to a file, opening in append mode (O(1), no read-back). fn append_bytes(path: &Path, bytes: &[u8]) -> Result<()> { use std::io::Write;