diff --git a/CHANGELOG.md b/CHANGELOG.md index cc00ff7ed..f75606e13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,54 @@ # Changelog +## Unreleased: locator-based transcript reads and writes (tinyagents#453) + +Hosts can now do through a `TranscriptLocator` everything they used the +path-taking transcript API for, so the same code serves JSONL files and a +driver store: + +- `TranscriptRead::read_display()` (default: from `read_session`; file + handles read the JSONL display log; driver handles replay their entry log + with compactions and interrupted partials in place). +- `TranscriptLocator::thread_transcripts(thread_id, agent_id) -> + Option` lists a thread's roots in projection order and + its sub-agent transcripts (`ThreadTranscript { stem, transcript }`). +- `view::get_page_in(locator, thread_id, agent_id, cursor, limit)` and + `view::project_transcripts(...)`; `spend::thread_spend_in(locator, + thread_id)`. +- `append_background_message_in(locator, session, message, options)`; for a + non-file store the provenance rides in `extra_metadata` under + `BACKGROUND_METADATA_KEY`. `TranscriptLocator::file_backing()` names the + file locator behind a file-backed one. +- `DriverTranscriptLocator::adopt_legacy` folds a thread's pre-identity + roots (for example the imported `session_raw/` timestamped stems of a + `WorkspaceStores` workspace) into the session once, with the file adopter's + semantics; the write reserves the session stem, so a concurrent first turn + wins and the call reports `Ok(None)`. +- `legacy::rewrite_legacy_transcripts(workspace_dir, rewrite) -> + LegacyRewriteReport` for pre-import fixups of classic `session_raw/` files. + +The path API is unchanged; `project_from_files` and `thread_spend` now share +the locator code path. + +Review follow-ups: `rewrite_legacy_transcripts` keeps a transcript's recorded +tool declarations, counts unreadable directories and entries in `errors` +(a missing `session_raw/` is still an empty report) and skips symlinks. +Thread spend counts a compacted sub-agent's `.gN` chain as one run with its +carried rows counted once (`thread_spend` and `thread_spend_in`). +`AgentTranscriptFiles` adopts a shared session on a background append and +lists shared sub-agents beside its own. A background append through a +locator with no destination key is serialised by that locator's identity. +Driver-backed stores repair the index on a display read, order pre-identity +roots by `_meta.created`, drop adopted roots beside an unscoped session root, +bound the sub-agent listing query to the thread, and find roots missing from +the index before adoption. +A model-context `TranscriptMessage` now keeps a JSONL row's background +provenance in `extra_metadata` (under `BACKGROUND_METADATA_KEY`, as row +stores do) and the line codec writes it back as the line field, so legacy +rewrites and compactions keep the idempotency key on both backends. +`InMemoryTranscriptLocator` lists a thread's generations and sub-agents, and +`InMemoryTranscriptHistory::read_display` shows recorded partials. + ## Unreleased: unapplied steers are back in the run queue when a turn ends `SteeringForwarderGuard` now drains and delivers under its shutdown gate and diff --git a/crates/tinyagents-session/src/legacy/mod.rs b/crates/tinyagents-session/src/legacy/mod.rs index 70b603851..d44d07562 100644 --- a/crates/tinyagents-session/src/legacy/mod.rs +++ b/crates/tinyagents-session/src/legacy/mod.rs @@ -28,6 +28,7 @@ //! - [`transcripts::import_transcripts`] and [`turn_states::import_turn_states`]: //! the file stores alone. +mod rewrite; pub(crate) mod rows; pub mod source; pub mod transcripts; @@ -37,11 +38,12 @@ mod types; mod workspace; pub use crate::run_ledger::drivers::{ImportCounts, LegacyDatabaseReport}; +pub use rewrite::rewrite_legacy_transcripts; pub use source::{LegacySourceSnapshot, QuiescentSource, SourceOwner, open_source}; pub use transcripts::{TranscriptImportReport, import_transcripts}; pub use transcripts_once::{MARKER_COLLECTION, TRANSCRIPT_MARKER_ID, import_transcripts_once}; pub use turn_states::{TurnStateImportReport, import_turn_states}; -pub use types::OnceOutcome; +pub use types::{LegacyRewriteReport, OnceOutcome}; pub use workspace::{LegacyWorkspaceReport, WorkspaceLayout, import_workspace}; #[cfg(test)] diff --git a/crates/tinyagents-session/src/legacy/rewrite.rs b/crates/tinyagents-session/src/legacy/rewrite.rs new file mode 100644 index 000000000..a99bf2d23 --- /dev/null +++ b/crates/tinyagents-session/src/legacy/rewrite.rs @@ -0,0 +1,166 @@ +//! Pre-import maintenance of classic JSONL transcripts. +//! +//! A host that must fix legacy content before it is imported (for example a +//! migration that strips a retired prompt section from every persisted +//! system message) used to walk `session_raw/` and call the path-taking +//! reader and writer itself. [`rewrite_legacy_transcripts`] owns that walk +//! here, beside the importer that owns the format, so the host names no file. + +use std::path::{Path, PathBuf}; + +use super::transcripts::raw_dir; +use super::types::LegacyRewriteReport; +use crate::transcript::{SessionTranscript, read_transcript, write_transcript_with_tools}; + +const LOG_PREFIX: &str = "[session:legacy:rewrite]"; + +/// Applies `rewrite` to every classic transcript under +/// `/session_raw/` (flat and `DDMMYYYY/` date-grouped +/// `*.jsonl`), writing a transcript back in full when `rewrite` returns +/// `true`. A rewritten file is a full rewrite of its reduced context (its +/// pre-compaction history is not kept), exactly as the classic writer's +/// `write_transcript` produces. +/// +/// Run it before the workspace is imported: the import is one-time, so a +/// file changed afterwards is not re-read. Per-file read or write failures +/// are counted in [`LegacyRewriteReport::errors`] and do not stop the walk; a +/// missing `session_raw/` is an empty report. A directory or directory entry +/// that cannot be read (anything but a missing `session_raw/`) is counted in +/// `errors` too, so a report with `errors == 0` means every transcript was +/// seen. Symlinks *inside* `session_raw/` are skipped, never followed, so +/// the walk cannot leave the transcript directory through an entry. +/// `session_raw` itself is resolved like every other transcript reader and +/// writer resolves it: a workspace may share its `session_raw` through a +/// symlink (the turn lock keys on the resolved directory for exactly that), +/// and that directory is the one whose transcripts the import will read. A +/// rewritten transcript keeps its latest recorded tool declarations +/// ([`SessionTranscript::tools`]) and each row's background-delivery +/// provenance. +pub fn rewrite_legacy_transcripts( + workspace_dir: &Path, + mut rewrite: impl FnMut(&mut SessionTranscript) -> bool, +) -> LegacyRewriteReport { + let mut report = LegacyRewriteReport::default(); + for path in jsonl_files(&raw_dir(workspace_dir), &mut report.errors) { + report.scanned += 1; + let mut session = match read_transcript(&path) { + Ok(session) => session, + Err(error) => { + report.errors += 1; + tracing::warn!("{LOG_PREFIX} unreadable path={}: {error:#}", path.display()); + continue; + } + }; + if !rewrite(&mut session) { + report.unchanged += 1; + continue; + } + match write_transcript_with_tools( + &path, + &session.messages, + &session.meta, + None, + session.tools.as_ref(), + ) { + Ok(()) => { + report.rewritten += 1; + tracing::debug!("{LOG_PREFIX} rewrote path={}", path.display()); + } + Err(error) => { + report.errors += 1; + tracing::warn!( + "{LOG_PREFIX} write failed path={}: {error:#}", + path.display() + ); + } + } + } + tracing::debug!( + "{LOG_PREFIX} scanned={} rewritten={} unchanged={} errors={}", + report.scanned, + report.rewritten, + report.unchanged, + report.errors + ); + report +} + +/// Every `*.jsonl` directly under `raw` or one level down, sorted. +/// +/// A missing `raw` is no files. Any other failure to read a directory or one +/// of its entries adds one to `errors` (and is logged) instead of silently +/// shrinking the scan. Symlinks are skipped without being followed. +fn jsonl_files(raw: &Path, errors: &mut usize) -> Vec { + let mut found = Vec::new(); + let entries = match std::fs::read_dir(raw) { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return found, + Err(error) => { + *errors += 1; + tracing::warn!("{LOG_PREFIX} unreadable dir={}: {error}", raw.display()); + return found; + } + }; + for entry in entries { + let Some((path, kind)) = scanned_entry(raw, entry, errors) else { + continue; + }; + if kind.is_dir() { + let inner = match std::fs::read_dir(&path) { + Ok(inner) => inner, + Err(error) => { + *errors += 1; + tracing::warn!("{LOG_PREFIX} unreadable dir={}: {error}", path.display()); + continue; + } + }; + for entry in inner { + if let Some((path, kind)) = scanned_entry(&path, entry, errors) + && kind.is_file() + && is_jsonl(&path) + { + found.push(path); + } + } + } else if kind.is_file() && is_jsonl(&path) { + found.push(path); + } + } + found.sort(); + found +} + +/// One directory entry with its own (not followed) file type; `None` for a +/// symlink, or for an entry that could not be read (counted in `errors`). +fn scanned_entry( + dir: &Path, + entry: std::io::Result, + errors: &mut usize, +) -> Option<(PathBuf, std::fs::FileType)> { + // `DirEntry::file_type` does not traverse symlinks: a link reports as a + // link, whatever it points at. + let read = entry.and_then(|entry| Ok((entry.path(), entry.file_type()?))); + match read { + Ok((path, kind)) if kind.is_symlink() => { + tracing::debug!("{LOG_PREFIX} skipping symlink path={}", path.display()); + None + } + Ok(found) => Some(found), + Err(error) => { + *errors += 1; + tracing::warn!( + "{LOG_PREFIX} unreadable entry in dir={}: {error}", + dir.display() + ); + None + } + } +} + +fn is_jsonl(path: &Path) -> bool { + path.extension().and_then(|ext| ext.to_str()) == Some("jsonl") +} + +#[cfg(test)] +#[path = "rewrite_tests.rs"] +mod tests; diff --git a/crates/tinyagents-session/src/legacy/rewrite_tests.rs b/crates/tinyagents-session/src/legacy/rewrite_tests.rs new file mode 100644 index 000000000..4ab113512 --- /dev/null +++ b/crates/tinyagents-session/src/legacy/rewrite_tests.rs @@ -0,0 +1,232 @@ +use super::*; +use crate::transcript::{TranscriptMessage, TranscriptMeta, write_transcript}; + +fn meta() -> TranscriptMeta { + TranscriptMeta { + agent_name: "orchestrator".into(), + agent_id: Some("orchestrator".into()), + agent_type: None, + dispatcher: "native".into(), + provider: None, + model: None, + created: "2026-01-01T00:00:00Z".into(), + updated: "2026-01-01T00:00:00Z".into(), + turn_count: 1, + prefix_message_count: None, + input_tokens: 0, + output_tokens: 0, + cached_input_tokens: 0, + charged_amount_usd: 0.0, + thread_id: None, + task_id: None, + session_id: None, + parent_session_id: None, + } +} + +fn write(path: &Path, system: &str) { + write_transcript( + path, + &[ + TranscriptMessage::system(system), + TranscriptMessage::user("hi"), + ], + &meta(), + None, + ) + .unwrap(); +} + +#[test] +fn rewrites_flat_and_dated_transcripts_and_counts_the_rest() { + let dir = tempfile::tempdir().unwrap(); + let raw = raw_dir(dir.path()); + write(&raw.join("flat.jsonl"), "keep ### SECRET"); + write(&raw.join("01012026").join("dated.jsonl"), "clean"); + std::fs::write(raw.join("broken.jsonl"), b"not json\n").unwrap(); + std::fs::write(raw.join("notes.md"), b"ignored").unwrap(); + + let report = rewrite_legacy_transcripts(dir.path(), |session| { + let first = &mut session.messages[0]; + if !first.content.contains("### SECRET") { + return false; + } + first.content = first.content.replace(" ### SECRET", ""); + true + }); + + assert_eq!( + report, + LegacyRewriteReport { + scanned: 3, + rewritten: 1, + unchanged: 1, + errors: 1, + } + ); + let flat = read_transcript(&raw.join("flat.jsonl")).unwrap(); + assert_eq!(flat.messages[0].content, "keep"); + assert_eq!(flat.messages[1].content, "hi"); + let dated = read_transcript(&raw.join("01012026").join("dated.jsonl")).unwrap(); + assert_eq!(dated.messages[0].content, "clean"); +} + +#[test] +fn a_workspace_without_transcripts_is_an_empty_report() { + let dir = tempfile::tempdir().unwrap(); + assert_eq!( + rewrite_legacy_transcripts(dir.path(), |_| true), + LegacyRewriteReport::default() + ); +} + +#[test] +fn a_rewrite_keeps_the_recorded_tool_declarations() { + let dir = tempfile::tempdir().unwrap(); + let path = raw_dir(dir.path()).join("tools.jsonl"); + write(&path, "keep ### SECRET"); + let tools = serde_json::json!([{"name": "search"}]); + crate::transcript::append_tools_record(&path, &tools).unwrap(); + + let report = rewrite_legacy_transcripts(dir.path(), |session| { + session.messages[0].content = "keep".into(); + true + }); + + assert_eq!(report.rewritten, 1); + let rewritten = read_transcript(&path).unwrap(); + assert_eq!(rewritten.messages[0].content, "keep"); + assert_eq!(rewritten.tools, Some(tools)); +} + +#[test] +fn an_unreadable_session_raw_is_an_error_not_an_empty_scan() { + let dir = tempfile::tempdir().unwrap(); + // `session_raw` exists but is not a directory: `read_dir` fails with + // something other than `NotFound`. + std::fs::write(raw_dir(dir.path()), b"not a directory").unwrap(); + + let report = rewrite_legacy_transcripts(dir.path(), |_| true); + + assert_eq!(report.scanned, 0); + assert_eq!(report.errors, 1, "the failed scan is reported"); +} + +#[cfg(unix)] +#[test] +fn an_unreadable_dated_directory_is_counted() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().unwrap(); + let raw = raw_dir(dir.path()); + write(&raw.join("flat.jsonl"), "flat"); + let dated = raw.join("01012026"); + write(&dated.join("dated.jsonl"), "dated"); + std::fs::set_permissions(&dated, std::fs::Permissions::from_mode(0o000)).unwrap(); + // A privileged runner (root) can read it anyway; nothing to observe then. + let readable = std::fs::read_dir(&dated).is_ok(); + + let report = rewrite_legacy_transcripts(dir.path(), |_| false); + std::fs::set_permissions(&dated, std::fs::Permissions::from_mode(0o755)).unwrap(); + + if !readable { + assert_eq!(report.scanned, 1); + assert_eq!( + report.errors, 1, + "the unreadable child directory is reported" + ); + } +} + +#[cfg(unix)] +#[test] +fn symlinks_are_neither_followed_nor_rewritten() { + use std::os::unix::fs::symlink; + + let dir = tempfile::tempdir().unwrap(); + let outside = tempfile::tempdir().unwrap(); + let raw = raw_dir(dir.path()); + write(&raw.join("inside.jsonl"), "inside"); + let target_file = outside.path().join("outside.jsonl"); + write(&target_file, "outside"); + let target_dir = outside.path().join("01012026"); + write(&target_dir.join("nested.jsonl"), "nested"); + symlink(&target_file, raw.join("linked.jsonl")).unwrap(); + symlink(&target_dir, raw.join("02012026")).unwrap(); + + let report = rewrite_legacy_transcripts(dir.path(), |session| { + session.messages[0].content = "rewritten".into(); + true + }); + + assert_eq!(report.scanned, 1, "only the real file is scanned"); + assert_eq!(report.rewritten, 1); + assert_eq!( + read_transcript(&target_file).unwrap().messages[0].content, + "outside" + ); + assert_eq!( + read_transcript(&target_dir.join("nested.jsonl")) + .unwrap() + .messages[0] + .content, + "nested" + ); +} + +#[tokio::test] +async fn a_rewrite_keeps_background_delivery_provenance() { + use crate::transcript::{ + BackgroundAppend, BackgroundAppendOutcome, DisplayRecord, FileTranscriptLocator, + SessionRef, TranscriptLocator, append_background_message, read_transcript_display, + resolve_keyed_transcript_path, session_stem, + }; + + let dir = tempfile::tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + let session = SessionRef::scoped("thread-1", "orchestrator"); + locator + .open_session(&session, meta()) + .unwrap() + .replace(&[ + TranscriptMessage::system("keep ### SECRET"), + TranscriptMessage::user("hi"), + ]) + .unwrap(); + let options = || BackgroundAppend::new("run-1", serde_json::json!({"kind": "cron"})); + let deliver = || { + append_background_message( + &locator, + &session, + TranscriptMessage::assistant("Time to stretch!"), + options(), + ) + }; + assert_eq!( + deliver().await.unwrap(), + BackgroundAppendOutcome::Appended { generation: 0 } + ); + + let report = rewrite_legacy_transcripts(dir.path(), |session| { + session.messages[0].content = "keep".into(); + true + }); + assert_eq!(report.rewritten, 1); + + let path = resolve_keyed_transcript_path(dir.path(), &session_stem(&session)).unwrap(); + let display = read_transcript_display(&path).unwrap(); + let origin = display + .records + .iter() + .find_map(|record| match record { + DisplayRecord::Message(message) => message.background.clone(), + DisplayRecord::Compaction(_) => None, + }) + .expect("the rewritten row keeps its provenance"); + assert_eq!(origin.idempotency_key, "run-1"); + assert_eq!( + deliver().await.unwrap(), + BackgroundAppendOutcome::Duplicate { generation: 0 }, + "the idempotency key survives the rewrite" + ); +} diff --git a/crates/tinyagents-session/src/legacy/types.rs b/crates/tinyagents-session/src/legacy/types.rs index 154bfecf6..4fa2f07a5 100644 --- a/crates/tinyagents-session/src/legacy/types.rs +++ b/crates/tinyagents-session/src/legacy/types.rs @@ -16,3 +16,16 @@ pub enum OnceOutcome { marked: bool, }, } + +/// What [`rewrite_legacy_transcripts`](super::rewrite_legacy_transcripts) did. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct LegacyRewriteReport { + /// JSONL transcripts found. + pub scanned: usize, + /// Transcripts the rewrite changed and that were written back. + pub rewritten: usize, + /// Transcripts the rewrite left as they were. + pub unchanged: usize, + /// Transcripts that could not be read or written back. + pub errors: usize, +} diff --git a/crates/tinyagents-session/src/port/drivers/transcripts.rs b/crates/tinyagents-session/src/port/drivers/transcripts.rs index 7e3b0308f..131da838c 100644 --- a/crates/tinyagents-session/src/port/drivers/transcripts.rs +++ b/crates/tinyagents-session/src/port/drivers/transcripts.rs @@ -615,6 +615,10 @@ impl TranscriptRead for DriverTranscriptHistory { })) })?) } + + fn read_display(&self) -> anyhow::Result> { + self.display() + } } /// The entry recording a turn whose logical messages are now `next`. @@ -1011,6 +1015,25 @@ impl TranscriptLocator for DriverTranscriptLocator { Ok(Arc::new(self.handle(stem, seed))) } + fn thread_transcripts( + &self, + thread_id: &str, + agent_id: Option<&str>, + ) -> Option { + self.list_thread(thread_id, agent_id) + } + + /// Folds the thread's pre-identity roots into `session` once; see + /// `transcripts_adoption.rs`. + fn adopt_legacy( + &self, + session: &SessionRef, + thread_id: &str, + seed: &TranscriptMeta, + ) -> anyhow::Result> { + self.adopt(session, thread_id, seed) + } + fn session_exists(&self, session: &SessionRef) -> bool { let stem = session_stem(session); let written = self.run(|docs, declared| async move { @@ -1153,6 +1176,11 @@ impl DriverTranscriptLocator { } } +#[path = "transcripts_adoption.rs"] +mod adoption; +#[path = "transcripts_display.rs"] +mod display; + #[cfg(test)] #[path = "transcripts_tests.rs"] pub(super) mod tests; diff --git a/crates/tinyagents-session/src/port/drivers/transcripts_adoption.rs b/crates/tinyagents-session/src/port/drivers/transcripts_adoption.rs new file mode 100644 index 000000000..e6a0c1bbb --- /dev/null +++ b/crates/tinyagents-session/src/port/drivers/transcripts_adoption.rs @@ -0,0 +1,246 @@ +//! Adoption of pre-identity roots into a session, for driver-backed +//! transcripts. +//! +//! A workspace imported from classic `session_raw/` files +//! ([`legacy::import_transcripts_once`](crate::legacy::import_transcripts_once)) +//! keeps its timestamped roots under their old stems. Resume addresses a +//! conversation by [`SessionRef`], so without this those roots are never read +//! again. [`TranscriptLocator::adopt_legacy`] folds them into the session's +//! generation 0 once, with the semantics of the file adopter +//! ([`adopt_legacy_session_transcripts`](crate::transcript::adopt_legacy_session_transcripts)): +//! +//! - nothing happens when the session already has a transcript; +//! - candidates are the thread's written roots other than the session's own +//! stem, skipping session-identified ones and, when the session names an +//! agent, another agent's; an unreadable candidate defers the whole call; +//! - candidates come from the index **and** from the entry log, so a root +//! whose durable write outran its index refresh is still folded (and its +//! index repaired by the read) instead of being left out of a fold that +//! never runs again; a failed log scan defers the call too; +//! - they are folded oldest first by `_meta.created` (messages, last recorded +//! tools, summed totals, earliest `created`, latest `updated`, the first +//! non-empty source's `prefix_message_count`); +//! - the legacy transcripts are never modified. +//! +//! The write is race-safe: the session's stem is reserved first (as a new +//! generation is) and written through the reserved handle, so a concurrent +//! first turn or adopter makes this call report `Ok(None)` instead of +//! overwriting it. + +use std::collections::BTreeSet; + +use serde_json::Value; +use tinystoragedrivers_core::{DocumentStoreExt, Filter, Query, StorageError}; + +use super::{DriverTranscriptLocator, ENTRIES, discovered_seed, is_subagent, run_on}; +use crate::transcript::{ + SessionAdoption, SessionRef, TranscriptHistory, TranscriptLocator, TranscriptMessage, + TranscriptMeta, TranscriptRead, TranscriptTurn, session_stem, +}; + +/// One legacy root read for folding. +struct Candidate { + path: std::path::PathBuf, + meta: TranscriptMeta, + messages: Vec, + tools: Option, +} + +impl DriverTranscriptLocator { + /// The legacy roots of `thread_id` that `session` adopts, oldest first. + fn adoption_candidates( + &self, + session: &SessionRef, + thread_id: &str, + own_stem: &str, + ) -> anyhow::Result> { + let filter = Filter::eq("thread_id", thread_id).and(Filter::eq("subagent", false)); + let mut stems: Vec = self + .indexed(filter, "adopt_legacy") + .into_iter() + .map(|indexed| indexed.stem) + .collect(); + for logged in self.logged_roots(thread_id)? { + if !stems.contains(&logged) { + tracing::debug!( + target: "tinyagents_session::port::drivers", + stem = %logged, + "[session-store] adoption found a root missing from the index" + ); + stems.push(logged); + } + } + let mut found = Vec::new(); + for stem in stems { + if stem == own_stem { + continue; + } + let handle = self.handle(&stem, discovered_seed(&stem)); + let Some(transcript) = handle.read_session().map_err(|error| { + error.context(format!( + "legacy root {stem} is unreadable; deferring adoption rather than finalizing a \ + fold that would permanently drop its turns" + )) + })? + else { + continue; + }; + if transcript.meta.session_id.is_some() + || transcript.meta.thread_id.as_deref() != Some(thread_id) + { + continue; + } + if let Some(expected) = session.agent_id.as_deref() + && transcript.meta.agent_id.as_deref() != Some(expected) + { + continue; + } + found.push(Candidate { + path: handle.path().to_path_buf(), + meta: transcript.meta, + messages: transcript.messages, + tools: transcript.tools, + }); + } + // Stable: equal `created` keeps the index's creation order. + found.sort_by(|a, b| a.meta.created.cmp(&b.meta.created)); + Ok(found) + } + + /// Root stems whose entry log names `thread_id` in a `_meta`, whether or + /// not the index has caught up with them. An error when the log cannot be + /// scanned: adoption must not finalize without it. + fn logged_roots(&self, thread_id: &str) -> anyhow::Result> { + let thread_id = thread_id.to_string(); + let entries = self.run(|docs, declared| async move { + declared.ensure(&docs).await?; + let query = Query::filter(Filter::eq("meta.thread_id", thread_id)); + docs.query_all(ENTRIES, &query).await + }); + let entries = entries.map_err(|error| { + anyhow::Error::new(error) + .context("could not scan the transcript log for legacy roots; deferring adoption") + })?; + Ok(entries + .iter() + .filter_map(|entry| entry.doc.get("stem").and_then(Value::as_str)) + .filter(|stem| !is_subagent(stem)) + .map(str::to_string) + .collect()) + } + + pub(super) fn adopt( + &self, + session: &SessionRef, + thread_id: &str, + seed: &TranscriptMeta, + ) -> anyhow::Result> { + if self.session_exists(session) { + return Ok(None); + } + let stem = session_stem(session); + let candidates = self.adoption_candidates(session, thread_id, &stem)?; + + let mut meta = seed.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.prefix_message_count = None; + meta.input_tokens = 0; + meta.output_tokens = 0; + meta.cached_input_tokens = 0; + meta.charged_amount_usd = 0.0; + let mut messages: Vec = Vec::new(); + let mut tools = None; + let mut earliest_created: Option = None; + let mut latest_updated: Option = None; + let mut adopted = Vec::new(); + for candidate in candidates { + if messages.is_empty() && !candidate.messages.is_empty() { + meta.prefix_message_count = candidate.meta.prefix_message_count; + } + messages.extend(candidate.messages); + if candidate.tools.is_some() { + tools = candidate.tools; + } + meta.turn_count += candidate.meta.turn_count; + meta.input_tokens += candidate.meta.input_tokens; + meta.output_tokens += candidate.meta.output_tokens; + meta.cached_input_tokens += candidate.meta.cached_input_tokens; + meta.charged_amount_usd += candidate.meta.charged_amount_usd; + let created = &candidate.meta.created; + if !created.is_empty() && earliest_created.as_ref().is_none_or(|e| created < e) { + earliest_created = Some(created.clone()); + } + let updated = &candidate.meta.updated; + if !updated.is_empty() && latest_updated.as_ref().is_none_or(|l| updated > l) { + latest_updated = Some(updated.clone()); + } + adopted.push(candidate.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; + } + + let reservation = { + let stem = stem.clone(); + let docs = std::sync::Arc::clone(&self.docs); + let declared = std::sync::Arc::clone(&self.declared); + run_on(&self.bridge, async move { + declared.ensure(&docs).await?; + Ok(Self::reserve(&docs, &stem).await) + })? + }; + let reservation = match reservation { + Ok(reservation) => reservation, + Err(error) if error.downcast_ref::().is_some() => return Err(error), + Err(_) => { + tracing::debug!( + target: "tinyagents_session::port::drivers", + session = %stem, + "[session-store] adoption lost the session stem to a concurrent write" + ); + return Ok(None); + } + }; + let handle = self.reserved_handle(&stem, meta.clone(), Some(reservation)); + let written = handle.append_turn(TranscriptTurn { + prev: &[], + next: &messages, + meta: &meta, + turn_usage: None, + request_id: None, + tools: tools.as_ref(), + }); + if let Err(error) = written { + if self.session_exists(session) { + return Ok(None); + } + self.release(&stem, reservation); + return Err(error); + } + tracing::info!( + target: "tinyagents_session::port::drivers", + session = %stem, + adopted = adopted.len(), + messages = messages.len(), + "[session-store] adopted legacy roots into the session" + ); + Ok(Some(SessionAdoption { + path: handle.path().to_path_buf(), + adopted, + messages: messages.len(), + })) + } +} + +#[cfg(test)] +#[path = "transcripts_adoption_tests.rs"] +mod tests; diff --git a/crates/tinyagents-session/src/port/drivers/transcripts_adoption_tests.rs b/crates/tinyagents-session/src/port/drivers/transcripts_adoption_tests.rs new file mode 100644 index 000000000..21e9c0be6 --- /dev/null +++ b/crates/tinyagents-session/src/port/drivers/transcripts_adoption_tests.rs @@ -0,0 +1,325 @@ +use super::*; +use std::sync::Arc; + +use crate::port::drivers::transcripts::tests::meta; +use crate::transcript::TranscriptMessage; +use tinystoragedrivers_core::{Blocking, DocumentStore, MemoryStorage, Scope, StorageBackend}; + +const THREAD: &str = "t"; + +fn locator() -> DriverTranscriptLocator { + let storage = MemoryStorage::new(); + let docs: Arc = + Arc::clone(storage.for_scope(&Scope::local()).unwrap().documents()); + DriverTranscriptLocator::new(docs, Blocking::new().unwrap(), "memory://adoption") +} + +/// A root of `THREAD` under `stem`, written with `meta` (as an import does). +fn write_root(locator: &DriverTranscriptLocator, stem: &str, meta: &TranscriptMeta, rows: &[&str]) { + let rows: Vec = rows + .iter() + .map(|row| TranscriptMessage::user(*row)) + .collect(); + locator + .open_stem(stem, meta.clone()) + .unwrap() + .append_turn(TranscriptTurn { + prev: &[], + next: &rows, + meta, + turn_usage: None, + request_id: None, + tools: None, + }) + .unwrap(); +} + +fn legacy(agent: &str, created: &str, turns: usize, input: u64) -> TranscriptMeta { + let mut meta = meta(THREAD); + meta.agent_name = agent.into(); + meta.agent_id = Some(agent.into()); + meta.created = created.into(); + meta.updated = created.into(); + meta.turn_count = turns; + meta.input_tokens = input; + meta +} + +fn contents(locator: &dyn TranscriptLocator, session: &SessionRef) -> Vec { + locator + .read_session_transcript(session) + .expect("the session has a transcript") + .read_session() + .unwrap() + .unwrap() + .messages + .into_iter() + .map(|message| message.content) + .collect() +} + +#[test] +fn legacy_roots_are_folded_oldest_first_once() { + let locator = locator(); + // Written newest first: the fold orders by `_meta.created`, not by write. + write_root( + &locator, + "1700000200_planner", + &legacy("planner", "2026-01-02", 1, 7), + &["two"], + ); + write_root( + &locator, + "1700000100_planner", + &legacy("planner", "2026-01-01", 2, 5), + &["one"], + ); + let session = SessionRef::scoped(THREAD, "planner"); + + let adoption = locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .expect("two legacy roots to adopt"); + assert_eq!(adoption.messages, 2); + assert_eq!(adoption.adopted.len(), 2); + assert!(adoption.adopted[0].ends_with("1700000100_planner")); + assert!(adoption.path.ends_with(session_stem(&session))); + assert_eq!(contents(&locator, &session), vec!["one", "two"]); + + let folded = locator + .read_session_transcript(&session) + .unwrap() + .read_session() + .unwrap() + .unwrap() + .meta; + assert_eq!(folded.session_id, Some(session.session_id())); + assert_eq!(folded.thread_id.as_deref(), Some(THREAD)); + assert_eq!((folded.turn_count, folded.input_tokens), (3, 12)); + assert_eq!( + (folded.created.as_str(), folded.updated.as_str()), + ("2026-01-01", "2026-01-02") + ); + + assert!( + locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .is_none(), + "a second call finds the session already written" + ); + assert_eq!(contents(&locator, &session), vec!["one", "two"]); + // The legacy roots are left as they were. + let untouched = locator + .open_stem("1700000100_planner", meta(THREAD)) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert_eq!(untouched.messages.len(), 1); + assert!(untouched.meta.session_id.is_none()); +} + +#[test] +fn session_identified_roots_and_other_agents_are_not_folded() { + let locator = locator(); + write_root( + &locator, + "1700000100_planner", + &legacy("planner", "2026-01-01", 1, 0), + &["mine"], + ); + write_root( + &locator, + "1700000150_writer", + &legacy("writer", "2026-01-01", 1, 0), + &["writer's"], + ); + let other = SessionRef::scoped("another-key", "planner"); + let mut identified = legacy("planner", "2026-01-01", 1, 0); + identified.session_id = Some(other.session_id()); + write_root( + &locator, + &session_stem(&other), + &identified, + &["already a session"], + ); + + let session = SessionRef::scoped(THREAD, "planner"); + let adoption = locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .unwrap(); + assert_eq!(adoption.adopted.len(), 1); + assert_eq!(contents(&locator, &session), vec!["mine"]); +} + +#[test] +fn nothing_to_fold_writes_nothing() { + let locator = locator(); + let session = SessionRef::scoped(THREAD, "planner"); + assert!( + locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .is_none() + ); + write_root( + &locator, + "1700000100_writer", + &legacy("writer", "2026-01-01", 1, 0), + &["x"], + ); + assert!( + locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .is_none() + ); + assert!(!locator.session_exists(&session)); +} + +#[test] +fn an_existing_session_is_left_alone() { + let locator = locator(); + write_root( + &locator, + "1700000100_planner", + &legacy("planner", "2026-01-01", 1, 0), + &["old"], + ); + let session = SessionRef::scoped(THREAD, "planner"); + let mut own = meta(THREAD); + own.session_id = Some(session.session_id()); + write_root(&locator, &session_stem(&session), &own, &["new"]); + assert!( + locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .is_none() + ); + assert_eq!(contents(&locator, &session), vec!["new"]); +} + +#[test] +fn a_reserved_stem_reports_nothing_adopted() { + let locator = locator(); + write_root( + &locator, + "1700000100_planner", + &legacy("planner", "2026-01-01", 1, 0), + &["old"], + ); + let session = SessionRef::scoped(THREAD, "planner"); + // A concurrent writer holds the session's stem. + let stem = session_stem(&session); + let docs = Arc::clone(&locator.docs); + let declared = Arc::clone(&locator.declared); + run_on(&locator.bridge, async move { + declared.ensure(&docs).await?; + Ok(DriverTranscriptLocator::reserve(&docs, &stem).await) + }) + .unwrap() + .unwrap(); + assert!( + locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .is_none() + ); +} + +#[cfg(feature = "legacy-import")] +#[test] +fn an_imported_file_workspace_is_adopted_and_resumes_the_folded_history() { + use crate::port::WorkspaceStores; + use crate::transcript::write_transcript; + + let dir = tempfile::tempdir().unwrap(); + let raw = dir.path().join("session_raw"); + for (stem, created, row) in [ + ("1700000100_orchestrator", "2026-01-01", "first"), + ("1700000200_orchestrator", "2026-01-02", "second"), + ] { + let mut legacy = legacy("orchestrator", created, 1, 0); + legacy.dispatcher = "native".into(); + write_transcript( + &raw.join(format!("{stem}.jsonl")), + &[TranscriptMessage::user(row)], + &legacy, + None, + ) + .unwrap(); + } + let session = SessionRef::scoped(THREAD, "orchestrator"); + let seed = legacy("orchestrator", "2026-02-01", 0, 0); + + let stores = WorkspaceStores::open(dir.path(), "orchestrator").unwrap(); + let transcripts = stores.transcripts(); + let adoption = transcripts + .adopt_legacy(&session, THREAD, &seed) + .unwrap() + .expect("the imported legacy roots are adopted"); + assert_eq!(adoption.messages, 2); + assert!( + transcripts + .adopt_legacy(&session, THREAD, &seed) + .unwrap() + .is_none() + ); + drop((transcripts, stores)); + + // The next process resumes the session in place. + let stores = WorkspaceStores::open(dir.path(), "orchestrator").unwrap(); + let transcripts = stores.transcripts(); + assert!( + transcripts + .adopt_legacy(&session, THREAD, &seed) + .unwrap() + .is_none() + ); + assert_eq!( + contents(transcripts.as_ref(), &session), + vec!["first", "second"] + ); +} + +#[test] +fn a_root_missing_from_the_index_is_still_folded() { + let locator = locator(); + write_root( + &locator, + "1700000100_planner", + &legacy("planner", "2026-01-01", 1, 0), + &["one"], + ); + write_root( + &locator, + "1700000200_planner", + &legacy("planner", "2026-01-02", 1, 0), + &["two"], + ); + // The second write was durable but every index refresh after it failed. + let docs = Arc::clone(&locator.docs); + Blocking::new() + .unwrap() + .run(async move { + docs.delete( + super::super::INDEX, + &super::super::doc_key(&["1700000200_planner"]), + tinystoragedrivers_core::Precondition::None, + ) + .await + }) + .unwrap() + .unwrap(); + let session = SessionRef::scoped(THREAD, "planner"); + + let adoption = locator + .adopt_legacy(&session, THREAD, &meta(THREAD)) + .unwrap() + .expect("both legacy roots to adopt"); + + assert_eq!(adoption.adopted.len(), 2); + assert_eq!(contents(&locator, &session), vec!["one", "two"]); +} diff --git a/crates/tinyagents-session/src/port/drivers/transcripts_display.rs b/crates/tinyagents-session/src/port/drivers/transcripts_display.rs new file mode 100644 index 000000000..afbcb0a43 --- /dev/null +++ b/crates/tinyagents-session/src/port/drivers/transcripts_display.rs @@ -0,0 +1,340 @@ +//! The display read and the thread listing of driver-backed transcripts. +//! +//! - **Display.** [`TranscriptRead::read_display`] replays a stem's whole +//! entry log, not the reduced context: `extend` rows are message records, a +//! `set` after the first write is a compaction marker carrying the new +//! context (the JSONL writer records exactly that), a `clear` is an empty +//! compaction, and a `partial` is an interrupted record at the position it +//! was written. Entries keep no wall clock, so a marker's `ts` is `None` and +//! intermediate tool-call rows carry no per-step timestamp. +//! - **Thread listing.** [`TranscriptLocator::thread_transcripts`] reads the +//! index: the thread's written roots, ordered like the file view (by +//! `_meta.created`, session chains oldest first, pre-identity roots dropped +//! once a session root exists), and the sub-agent transcripts that descend +//! from a root stem or (unscoped) name the thread. + +use std::collections::HashSet; +use std::sync::Arc; + +use serde_json::Value; +use tinystoragedrivers_core::{DocumentStoreExt, Filter, Query, Sort, StorageError}; + +use super::{ + DriverTranscriptHistory, DriverTranscriptLocator, ENTRIES, Entry, HistoryInner, INDEX, + discovered_seed, normalized, run_on, +}; +use crate::transcript::{ + CompactionMarker, DisplayRecord, DisplaySessionTranscript, SessionRef, ThreadTranscript, + ThreadTranscripts, TranscriptLocator, TranscriptMeta, TranscriptPartial, TranscriptRead, + display_message_of, partial_display_message, session_stem, +}; + +impl HistoryInner { + /// The display projection of this stem's whole log; `None` when nothing + /// was ever written to it. + async fn display(&self) -> Result, StorageError> { + self.declared.ensure(&self.docs).await?; + { + // As the model-context read does: a write whose index refresh + // failed is repaired here, so a transcript read only for display + // does not stay invisible to the thread listing and spend. + let mut replay = self.replay.lock().await; + self.refresh(&mut replay).await?; + self.repair_index(&mut replay).await; + } + let query = Query::filter(Filter::eq("stem", self.stem.as_str())).sort(Sort::asc("seq")); + let mut meta = None; + let mut records: Vec = Vec::new(); + let mut written = false; + for stored in self.docs.query_all(ENTRIES, &query).await? { + let entry: Entry = serde_json::from_value(stored.doc).map_err(|error| { + StorageError::serialization(format!( + "transcript entry {} is unreadable: {error}", + stored.id + )) + })?; + let was_written = written; + written |= entry.written; + if let Some(entry_meta) = entry.meta { + meta = Some(entry_meta); + } + if let Some(set) = entry.set { + let rows: Vec<_> = normalized(set).iter().map(display_message_of).collect(); + if was_written { + records.push(DisplayRecord::Compaction(CompactionMarker { + replacement: rows, + ts: None, + request_id: entry.request_id.clone(), + })); + } else { + records.extend( + rows.into_iter() + .map(|row| DisplayRecord::Message(Box::new(row))), + ); + } + } + if let Some(extend) = entry.extend { + records.extend( + normalized(extend) + .iter() + .map(|row| DisplayRecord::Message(Box::new(display_message_of(row)))), + ); + } + if let Some(partial) = entry.partial { + let partial = TranscriptPartial::from(partial); + records.push(DisplayRecord::Message(Box::new(partial_display_message( + &partial, + entry.request_id.as_deref(), + )))); + } + if entry.clear { + records.push(DisplayRecord::Compaction(CompactionMarker { + replacement: Vec::new(), + ts: None, + request_id: None, + })); + } + } + Ok(written.then(|| DisplaySessionTranscript { + meta: meta.unwrap_or_else(|| self.seed.clone()), + records, + })) + } +} + +impl DriverTranscriptHistory { + pub(super) fn display(&self) -> anyhow::Result> { + let inner = Arc::clone(&self.inner); + Ok(run_on(&self.bridge, async move { inner.display().await })?) + } +} + +/// One index document's lookup fields. +pub(super) struct Indexed { + pub(super) stem: String, + agent_id: Option, + thread_id: Option, +} + +impl Indexed { + fn of(doc: &Value) -> Option { + let field = |key: &str| doc.get(key).and_then(Value::as_str).map(str::to_string); + Some(Self { + stem: field("stem")?, + agent_id: field("agent_id"), + thread_id: field("thread_id"), + }) + } +} + +impl DriverTranscriptLocator { + /// The written index documents matching `filter`, oldest first. + pub(super) fn indexed(&self, filter: Filter, what: &'static str) -> Vec { + let found = self.run(|docs, declared| async move { + declared.ensure(&docs).await?; + let query = Query::filter(filter.and(Filter::eq("written", true))) + .sort(Sort::asc("created_at")); + docs.query_all(INDEX, &query).await + }); + match found { + Ok(items) => items + .iter() + .filter_map(|item| Indexed::of(&item.doc)) + .collect(), + Err(error) => { + tracing::warn!( + target: "tinyagents_session::port::drivers", + locator = %self.label, + lookup = what, + %error, + "[session-store] transcript listing failed" + ); + Vec::new() + } + } + } + + fn listed(&self, stem: String) -> ThreadTranscript { + let transcript: Arc = + Arc::new(self.handle(&stem, discovered_seed(&stem))); + ThreadTranscript { stem, transcript } + } + + /// The `_meta` of the transcript at `stem`, when it is readable. + fn root_meta(&self, stem: &str) -> Option { + self.handle(stem, discovered_seed(stem)) + .read_session() + .ok() + .flatten() + .map(|session| session.meta) + } + + /// The thread's roots in projection order; see the module docs. + /// + /// Like the file view, roots are first ordered by `_meta.created` (then + /// generation, then stem), not by when their index document was written: + /// an import writes files in name order, which need not be chronological. + /// Once any root records a session identity, pre-identity roots are + /// dropped — they were folded into that session — whether or not the + /// session is agent-scoped. + fn ordered_roots(&self, thread_id: &str, roots: Vec) -> Vec { + let mut roots: Vec<(Indexed, Option)> = roots + .into_iter() + .map(|root| { + let meta = self.root_meta(&root.stem); + (root, meta) + }) + .collect(); + roots.sort_by(|(a, a_meta), (b, b_meta)| { + let created = |meta: &Option| { + meta.as_ref() + .map(|meta| meta.created.clone()) + .unwrap_or_default() + }; + created(a_meta) + .cmp(&created(b_meta)) + .then_with(|| stem_generation(&a.stem).cmp(&stem_generation(&b.stem))) + .then_with(|| a.stem.cmp(&b.stem)) + }); + let has_session_id = |meta: &Option| { + meta.as_ref().is_some_and(|meta| meta.session_id.is_some()) + }; + if !roots.iter().any(|(_, meta)| has_session_id(meta)) { + return roots.into_iter().map(|(root, _)| root.stem).collect(); + } + let mut chains: Vec> = Vec::new(); + let mut agents: HashSet = HashSet::new(); + for (root, _) in &roots { + if let Some(agent_id) = &root.agent_id + && agents.insert(agent_id.clone()) + { + let chain: Vec = self + .session_chain(&SessionRef::scoped(thread_id, agent_id.as_str())) + .iter() + .map(session_stem) + .collect(); + if !chain.is_empty() { + chains.push(chain); + } + } + } + let mut seen: HashSet = HashSet::new(); + let mut ordered = Vec::new(); + let mut dropped_legacy = 0usize; + for (root, meta) in roots { + if seen.contains(&root.stem) { + continue; + } + if let Some(chain) = chains.iter().find(|chain| chain.contains(&root.stem)) { + for link in chain { + if seen.insert(link.clone()) { + ordered.push(link.clone()); + } + } + } else if has_session_id(&meta) { + seen.insert(root.stem.clone()); + ordered.push(root.stem); + } else { + dropped_legacy += 1; + } + } + if dropped_legacy > 0 { + tracing::debug!( + target: "tinyagents_session::port::drivers", + thread_id, + dropped_legacy, + "[session-store] dropped adopted legacy roots in favour of the session chain" + ); + } + ordered + } + + pub(super) fn list_thread( + &self, + thread_id: &str, + agent_id: Option<&str>, + ) -> Option { + let thread_id = thread_id.trim(); + if thread_id.is_empty() { + return None; + } + let mut filter = Filter::eq("thread_id", thread_id).and(Filter::eq("subagent", false)); + if let Some(agent_id) = agent_id { + filter = filter.and(Filter::eq("agent_id", agent_id)); + } + let roots = self.indexed(filter, "thread_roots"); + if roots.is_empty() { + return None; + } + let prefixes: Vec = roots + .iter() + .map(|root| format!("{}__", root.stem)) + .collect(); + let roots = self.ordered_roots(thread_id, roots); + // Bounded to this thread: each root's `{stem}__` range, plus + // (unscoped) the sub-agents naming the thread. The in-memory filter + // below stays as the exact rule. + let by_prefix = prefixes + .iter() + .map(|prefix| prefix_range("stem", prefix)) + .reduce(Filter::or); + let by_thread = Filter::eq("thread_id", thread_id); + let wanted = match (by_prefix, agent_id) { + (Some(ranges), None) => ranges.or(by_thread), + (Some(ranges), Some(_)) => ranges, + (None, _) => by_thread, + }; + let mut subagents: Vec = self + .indexed(Filter::eq("subagent", true).and(wanted), "thread_subagents") + .into_iter() + .filter(|sub| { + prefixes.iter().any(|prefix| sub.stem.starts_with(prefix)) + || (agent_id.is_none() && sub.thread_id.as_deref() == Some(thread_id)) + }) + .map(|sub| sub.stem) + .collect(); + subagents.sort(); + subagents.dedup(); + tracing::debug!( + target: "tinyagents_session::port::drivers", + thread_id, + roots = roots.len(), + subagents = subagents.len(), + "[session-store] listed thread transcripts" + ); + Some(ThreadTranscripts { + roots: roots.into_iter().map(|stem| self.listed(stem)).collect(), + subagents: subagents + .into_iter() + .map(|stem| self.listed(stem)) + .collect(), + }) + } +} + +/// `field` starts with `prefix`, as a range: `prefix <= field < upper`, +/// where `upper` is `prefix` with its last character incremented. A stem +/// prefix here always ends in `__`, so the bump is `_` to `` ` ``. +fn prefix_range(field: &str, prefix: &str) -> Filter { + let mut upper: Vec = prefix.chars().collect(); + let filter = Filter::gte(field, prefix); + match upper.pop().and_then(|last| char::from_u32(last as u32 + 1)) { + Some(next) => { + upper.push(next); + filter.and(Filter::lt(field, upper.into_iter().collect::())) + } + None => filter, + } +} + +/// The compaction generation a stem encodes: `n` for `{base}.g{n}`, else 0. +fn stem_generation(stem: &str) -> u32 { + stem.rsplit_once(".g") + .filter(|(_, digits)| !digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit())) + .and_then(|(_, digits)| digits.parse().ok()) + .unwrap_or(0) +} + +#[cfg(test)] +#[path = "transcripts_display_tests.rs"] +mod tests; diff --git a/crates/tinyagents-session/src/port/drivers/transcripts_display_tests.rs b/crates/tinyagents-session/src/port/drivers/transcripts_display_tests.rs new file mode 100644 index 000000000..3d7a93769 --- /dev/null +++ b/crates/tinyagents-session/src/port/drivers/transcripts_display_tests.rs @@ -0,0 +1,175 @@ +use super::*; +use crate::port::drivers::transcripts::tests::meta; +use crate::transcript::{TranscriptMessage, TranscriptTurn}; +use tinystoragedrivers_core::{Blocking, DocumentStore, MemoryStorage, Scope, StorageBackend}; + +fn locator() -> DriverTranscriptLocator { + let storage = MemoryStorage::new(); + let docs: Arc = + Arc::clone(storage.for_scope(&Scope::local()).unwrap().documents()); + DriverTranscriptLocator::new(docs, Blocking::new().unwrap(), "memory://display") +} + +fn write(locator: &DriverTranscriptLocator, stem: &str, thread: &str, rows: &[&str]) { + let rows: Vec = rows.iter().map(|r| TranscriptMessage::user(*r)).collect(); + let meta = meta(thread); + locator + .open_stem(stem, meta.clone()) + .unwrap() + .append_turn(TranscriptTurn { + prev: &[], + next: &rows, + meta: &meta, + turn_usage: None, + request_id: None, + tools: None, + }) + .unwrap(); +} + +#[test] +fn a_replace_after_the_first_write_is_a_compaction_and_a_clear_an_empty_one() { + let locator = locator(); + let history = locator.open_stem("s", meta("t")).unwrap(); + history.replace(&[TranscriptMessage::user("a")]).unwrap(); + history.replace(&[TranscriptMessage::user("b")]).unwrap(); + history.clear().unwrap(); + let records = history.read_display().unwrap().unwrap().records; + assert!(matches!(&records[0], DisplayRecord::Message(m) if m.message.content == "a")); + assert!(matches!(&records[1], DisplayRecord::Compaction(c) if c.replacement.len() == 1)); + assert!(matches!(&records[2], DisplayRecord::Compaction(c) if c.replacement.is_empty())); +} + +#[test] +fn an_unwritten_stem_has_no_display() { + let locator = locator(); + let history = locator.open_stem("never", meta("t")).unwrap(); + assert!(history.read_display().unwrap().is_none()); +} + +#[test] +fn legacy_roots_are_dropped_once_a_session_root_exists() { + let locator = locator(); + // A pre-identity root of the thread (no session id in its meta). + write(&locator, "1700000000_planner", "t", &["old"]); + let listed = locator.thread_transcripts("t", None).unwrap(); + assert_eq!(listed.roots.len(), 1, "alone, a legacy root is listed"); + + let session = SessionRef::scoped("t", "planner"); + let mut seed = meta("t"); + seed.session_id = Some(session.session_id()); + locator + .open_session(&session, seed.clone()) + .unwrap() + .append_turn(TranscriptTurn { + prev: &[], + next: &[TranscriptMessage::user("new")], + meta: &seed, + turn_usage: None, + request_id: None, + tools: None, + }) + .unwrap(); + let stems: Vec = locator + .thread_transcripts("t", None) + .unwrap() + .roots + .into_iter() + .map(|root| root.stem) + .collect(); + assert_eq!(stems, vec![session_stem(&session)]); +} + +#[test] +fn subagents_are_found_by_stem_prefix_and_unscoped_by_thread() { + let locator = locator(); + write(&locator, "root", "t", &["q"]); + write(&locator, "root__1_child", "other-thread", &["by prefix"]); + write(&locator, "elsewhere__2_child", "t", &["by thread"]); + write( + &locator, + "elsewhere__3_child", + "other-thread", + &["unrelated"], + ); + let subs = |agent: Option<&str>| -> Vec { + locator + .thread_transcripts("t", agent) + .unwrap() + .subagents + .into_iter() + .map(|sub| sub.stem) + .collect() + }; + assert_eq!(subs(None), vec!["elsewhere__2_child", "root__1_child"]); + assert_eq!(subs(Some("planner")), vec!["root__1_child"]); + assert!(locator.thread_transcripts(" ", None).is_none()); +} + +fn write_with(locator: &DriverTranscriptLocator, stem: &str, meta: &TranscriptMeta, rows: &[&str]) { + let rows: Vec = rows.iter().map(|r| TranscriptMessage::user(*r)).collect(); + locator + .open_stem(stem, meta.clone()) + .unwrap() + .append_turn(TranscriptTurn { + prev: &[], + next: &rows, + meta, + turn_usage: None, + request_id: None, + tools: None, + }) + .unwrap(); +} + +fn root_stems(locator: &DriverTranscriptLocator, thread: &str) -> Vec { + locator + .thread_transcripts(thread, None) + .unwrap() + .roots + .into_iter() + .map(|root| root.stem) + .collect() +} + +#[test] +fn legacy_roots_are_dropped_once_an_unscoped_session_root_exists() { + let locator = locator(); + let mut legacy = meta("t"); + legacy.agent_id = None; + write_with(&locator, "1700000000_planner", &legacy, &["old"]); + + let session = SessionRef::root("t"); + let mut seed = legacy.clone(); + seed.session_id = Some(session.session_id()); + write_with(&locator, &session_stem(&session), &seed, &["old", "new"]); + + assert_eq!(root_stems(&locator, "t"), vec![session_stem(&session)]); +} + +#[test] +fn pre_identity_roots_are_ordered_by_their_creation_time() { + let locator = locator(); + // Imported in name order, which is not their chronology. + let mut later = meta("t"); + later.created = "2026-01-02T00:00:00Z".into(); + write_with(&locator, "a_planner", &later, &["second"]); + let mut earlier = meta("t"); + earlier.created = "2026-01-01T00:00:00Z".into(); + write_with(&locator, "b_planner", &earlier, &["first"]); + + assert_eq!(root_stems(&locator, "t"), vec!["b_planner", "a_planner"]); +} + +#[test] +fn prefix_ranges_match_exactly_the_prefixed_stems() { + let filter = prefix_range("stem", "root__"); + let matches = |stem: &str| filter.matches("id", &serde_json::json!({ "stem": stem })); + assert!(matches("root__")); + assert!(matches("root__1_child")); + assert!(matches("root__1_child__2_grandchild.g3")); + assert!(!matches("root_")); + assert!(!matches("root_x__1_child")); + assert!(!matches("root")); + assert!(!matches("rootz__1_child")); +} diff --git a/crates/tinyagents-session/src/port/drivers/transcripts_tests.rs b/crates/tinyagents-session/src/port/drivers/transcripts_tests.rs index 69e762cfd..97dbf8129 100644 --- a/crates/tinyagents-session/src/port/drivers/transcripts_tests.rs +++ b/crates/tinyagents-session/src/port/drivers/transcripts_tests.rs @@ -615,6 +615,32 @@ fn a_read_repairs_an_index_a_write_could_not_refresh() { ); } +#[test] +fn a_display_read_repairs_an_index_a_write_could_not_refresh() { + let docs = docs(); + let locator = locator(&docs); + let history = locator.handle("s", meta("t")); + history.append(message("user", "a")).unwrap(); + // The index refresh after that write was lost. + let raw = Arc::clone(&docs); + on_bridge(async move { + raw.delete(INDEX, &doc_key(&["s"]), Precondition::None) + .await + }); + assert!(locator.thread_transcripts("t", None).is_none()); + // A handle whose replay has not indexed the write (as after a refresh + // that failed every attempt). + locator + .handle("s", meta("t")) + .read_display() + .unwrap() + .unwrap(); + let listed = locator + .thread_transcripts("t", None) + .expect("the display read re-indexed it"); + assert_eq!(listed.roots[0].stem, "s"); +} + #[test] fn clearing_an_unwritten_successor_frees_the_generation() { let docs = docs(); diff --git a/crates/tinyagents-session/src/port/memory/mod.rs b/crates/tinyagents-session/src/port/memory/mod.rs index 3bb9ec0bd..1eac58c20 100644 --- a/crates/tinyagents-session/src/port/memory/mod.rs +++ b/crates/tinyagents-session/src/port/memory/mod.rs @@ -16,8 +16,8 @@ use tinyagents_harness::store::{InMemoryAppendStore, InMemoryStore}; use super::{AgentStores, SessionStoreProvider, TurnStates}; use crate::testkit::InMemoryTranscriptHistory; use crate::transcript::{ - SessionRef, TranscriptHistory, TranscriptLocator, TranscriptMeta, TranscriptPartial, - TranscriptRead, session_stem, + SessionRef, ThreadTranscript, ThreadTranscripts, TranscriptHistory, TranscriptLocator, + TranscriptMeta, TranscriptPartial, TranscriptRead, session_stem, }; use crate::turn_state::{TurnLifecycle, TurnState}; @@ -245,6 +245,74 @@ impl TranscriptLocator for InMemoryTranscriptLocator { self.root_for_thread_scoped(thread_id, None) } + /// Every written root of the thread and its sub-agents, with the file + /// view's rules: roots by `_meta.created` (then generation, then stem), + /// pre-identity roots dropped once a session root exists; sub-agents that + /// extend a matched root's stem or (unscoped) name the thread, by stem. + fn thread_transcripts( + &self, + thread_id: &str, + agent_id: Option<&str>, + ) -> Option { + let thread_id = thread_id.trim(); + if thread_id.is_empty() { + return None; + } + let written: Vec<(String, bool, TranscriptMeta, Arc)> = self + .stems + .lock() + .unwrap_or_else(PoisonError::into_inner) + .iter() + .filter_map(|(stem, is_subagent, history)| { + let session = history.read_session().ok().flatten()?; + Some((stem.clone(), *is_subagent, session.meta, history.clone())) + }) + .collect(); + let mut roots: Vec<_> = written + .iter() + .filter(|(_, is_subagent, meta, _)| { + !is_subagent + && meta.thread_id.as_deref() == Some(thread_id) + && agent_id.is_none_or(|agent| meta.agent_id.as_deref() == Some(agent)) + }) + .collect(); + if roots.is_empty() { + return None; + } + roots.sort_by(|a, b| { + a.2.created + .cmp(&b.2.created) + .then_with(|| stem_generation(&a.0).cmp(&stem_generation(&b.0))) + .then_with(|| a.0.cmp(&b.0)) + }); + let prefixes: Vec = roots.iter().map(|root| format!("{}__", root.0)).collect(); + if roots.iter().any(|root| root.2.session_id.is_some()) { + roots.retain(|root| root.2.session_id.is_some()); + } + let mut subagents: Vec<_> = written + .iter() + .filter(|(stem, is_subagent, meta, _)| { + *is_subagent + && (prefixes.iter().any(|prefix| stem.starts_with(prefix)) + || (agent_id.is_none() && meta.thread_id.as_deref() == Some(thread_id))) + }) + .collect(); + subagents.sort_by(|a, b| a.0.cmp(&b.0)); + let listed = |(stem, _, _, history): &&( + String, + bool, + TranscriptMeta, + Arc, + )| ThreadTranscript { + stem: stem.clone(), + transcript: history.clone() as Arc, + }; + Some(ThreadTranscripts { + roots: roots.iter().map(listed).collect(), + subagents: subagents.iter().map(listed).collect(), + }) + } + fn root_for_thread_scoped( &self, thread_id: &str, @@ -518,6 +586,14 @@ impl TurnStates for InMemoryTurnStates { } } +/// The compaction generation a stem encodes: `n` for `{base}.g{n}`, else 0. +fn stem_generation(stem: &str) -> u32 { + stem.rsplit_once(".g") + .filter(|(_, digits)| !digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit())) + .and_then(|(_, digits)| digits.parse().ok()) + .unwrap_or(0) +} + #[cfg(test)] #[path = "mod_tests.rs"] mod tests; diff --git a/crates/tinyagents-session/src/port/memory/mod_tests.rs b/crates/tinyagents-session/src/port/memory/mod_tests.rs index de4969c29..a786abbaa 100644 --- a/crates/tinyagents-session/src/port/memory/mod_tests.rs +++ b/crates/tinyagents-session/src/port/memory/mod_tests.rs @@ -267,3 +267,109 @@ fn completed_retention_uses_completion_time() { assert_eq!(listed[0].request_id, "new-start"); assert_eq!(listed[1].request_id, "old-start"); } + +fn write_rows(history: &dyn TranscriptHistory, meta: &TranscriptMeta, rows: &[&str]) { + let rows: Vec = rows + .iter() + .map(|row| TranscriptMessage::user(*row)) + .collect(); + history + .append_turn(TranscriptTurn { + prev: &[], + next: &rows, + meta, + turn_usage: None, + request_id: None, + tools: None, + }) + .unwrap(); +} + +#[test] +fn a_thread_lists_every_generation_and_sub_agent() { + let locator = InMemoryTranscriptLocator::new("listing"); + let session = SessionRef::scoped("t", "planner"); + let mut seed = test_meta("t"); + seed.session_id = Some(session.session_id()); + // A pre-identity root, folded into the session and dropped from the view. + write_rows( + locator + .open_stem("1700000000_planner", test_meta("t")) + .unwrap() + .as_ref(), + &test_meta("t"), + &["legacy"], + ); + write_rows( + locator + .open_session(&session, seed.clone()) + .unwrap() + .as_ref(), + &seed, + &["one"], + ); + let (successor, head) = locator.begin_generation(&session, seed.clone()).unwrap(); + write_rows(head.as_ref(), &seed, &["two"]); + let root = crate::transcript::session_stem(&session); + write_rows( + locator + .open_stem(&format!("{root}__1_child"), test_meta("worker")) + .unwrap() + .as_ref(), + &test_meta("worker"), + &["by prefix"], + ); + write_rows( + locator + .open_stem("elsewhere__2_child", test_meta("t")) + .unwrap() + .as_ref(), + &test_meta("t"), + &["by thread"], + ); + + let listed = locator.thread_transcripts("t", None).expect("listed"); + let stems = |items: &[ThreadTranscript]| -> Vec { + items.iter().map(|item| item.stem.clone()).collect() + }; + assert_eq!( + stems(&listed.roots), + vec![root.clone(), crate::transcript::session_stem(&successor)] + ); + assert_eq!( + stems(&listed.subagents), + vec!["elsewhere__2_child".to_string(), format!("{root}__1_child")] + ); + let scoped = locator.thread_transcripts("t", Some("planner")).unwrap(); + assert_eq!(stems(&scoped.subagents), vec![format!("{root}__1_child")]); + assert!(locator.thread_transcripts("missing", None).is_none()); +} + +#[test] +fn the_display_read_shows_recorded_partials_where_they_were_written() { + let history = InMemoryTranscriptHistory::new("display", test_meta("display")); + write_rows(&history, &test_meta("display"), &["question"]); + assert!(history.record_partial( + TranscriptPartial::new("half an answer"), + Some("req-1".into()) + )); + + let display = history.read_display().unwrap().expect("written"); + let rows: Vec<(String, bool)> = display + .records + .iter() + .map(|record| match record { + crate::transcript::DisplayRecord::Message(message) => { + (message.message.content.clone(), message.interrupted) + } + crate::transcript::DisplayRecord::Compaction(_) => panic!("no compaction here"), + }) + .collect(); + assert_eq!( + rows, + vec![ + ("question".to_string(), false), + ("half an answer".to_string(), true) + ] + ); +} diff --git a/crates/tinyagents-session/src/testkit/in_memory_transcript.rs b/crates/tinyagents-session/src/testkit/in_memory_transcript.rs index d1e5980aa..4ab124ab9 100644 --- a/crates/tinyagents-session/src/testkit/in_memory_transcript.rs +++ b/crates/tinyagents-session/src/testkit/in_memory_transcript.rs @@ -11,8 +11,9 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use crate::transcript::{ - SessionTranscript, TranscriptHistory, TranscriptMessage, TranscriptMeta, TranscriptPartial, - TranscriptRead, TranscriptTurn, + DisplayRecord, DisplaySessionTranscript, SessionTranscript, TranscriptHistory, + TranscriptMessage, TranscriptMeta, TranscriptPartial, TranscriptRead, TranscriptTurn, + display_message_of, partial_display_message, }; /// A [`TranscriptHistory`] backed by a `Vec` behind a @@ -43,9 +44,10 @@ struct InMemoryTranscriptState { tools: Option, /// `false` until the first write, mirroring a file that does not exist yet. written: bool, - /// Display-only partials of interrupted turns, with their request ids. - /// Never part of [`TranscriptHistory::messages`]. - partials: Vec<(TranscriptPartial, Option)>, + /// Display-only partials of interrupted turns, with their request ids + /// and how many logical rows preceded them when recorded (where the + /// display read places them). Never part of [`TranscriptHistory::messages`]. + partials: Vec<(TranscriptPartial, Option, usize)>, sealed: bool, } @@ -116,7 +118,8 @@ impl InMemoryTranscriptHistory { if state.sealed || partial.content.is_empty() { return false; } - state.partials.push((partial, request_id)); + let position = state.messages.len(); + state.partials.push((partial, request_id, position)); state.written = true; true } @@ -128,7 +131,9 @@ impl InMemoryTranscriptHistory { .lock() .unwrap_or_else(|e| e.into_inner()) .partials - .clone() + .iter() + .map(|(partial, request_id, _)| (partial.clone(), request_id.clone())) + .collect() } pub(crate) fn seal(&self) { @@ -165,6 +170,39 @@ impl TranscriptRead for InMemoryTranscriptHistory { tools: state.tools.clone(), })) } + + /// The logical rows with the recorded interrupted partials placed where + /// they were written, as the file and driver display reads show them. + /// There is no append log here, so earlier compactions are not replayed. + fn read_display(&self) -> anyhow::Result> { + let state = self.state.lock().unwrap_or_else(|e| e.into_inner()); + if !state.written { + return Ok(None); + } + let mut partials: Vec<_> = state.partials.iter().collect(); + partials.sort_by_key(|(_, _, position)| *position); + let mut partials = partials.into_iter().peekable(); + let mut records = Vec::new(); + for index in 0..=state.messages.len() { + while let Some((partial, request_id, _)) = + partials.next_if(|(_, _, position)| (*position).min(state.messages.len()) == index) + { + records.push(DisplayRecord::Message(Box::new(partial_display_message( + partial, + request_id.as_deref(), + )))); + } + if let Some(message) = state.messages.get(index) { + records.push(DisplayRecord::Message(Box::new(display_message_of( + message, + )))); + } + } + Ok(Some(DisplaySessionTranscript { + meta: state.meta.clone(), + records, + })) + } } impl TranscriptHistory for InMemoryTranscriptHistory { @@ -206,7 +244,8 @@ impl TranscriptHistory for InMemoryTranscriptHistory { state.tools = Some(tools.clone()); } if let Some(partial) = partial.filter(|partial| !partial.content.is_empty()) { - state.partials.push((partial.clone(), request_id)); + let position = state.messages.len(); + state.partials.push((partial.clone(), request_id, position)); } state.written = true; Ok(()) diff --git a/crates/tinyagents-session/src/transcript.rs b/crates/tinyagents-session/src/transcript.rs index c7df8c3d5..16f67bcc4 100644 --- a/crates/tinyagents-session/src/transcript.rs +++ b/crates/tinyagents-session/src/transcript.rs @@ -134,6 +134,7 @@ mod adoption; mod agent_files; mod background; +mod display; mod history; mod jsonl; mod legacy_md; @@ -151,7 +152,12 @@ mod writer; pub use adoption::{SessionAdoption, adopt_legacy_session_transcripts}; pub use agent_files::{AgentTranscriptFiles, agent_transcript_root}; -pub use background::{append_background_message, append_background_message_with_lock_notification}; +pub use background::{ + append_background_message, append_background_message_in, + append_background_message_with_lock_notification, +}; +pub use display::BACKGROUND_METADATA_KEY; +pub(crate) use display::{display_from_session, display_message_of, partial_display_message}; pub(crate) use history::same_transcript_messages; pub use history::{ FileTranscriptHistory, FileTranscriptLocator, TranscriptHistory, TranscriptLocator, @@ -171,9 +177,10 @@ pub use turn_lock::{SessionTurnGuard, lock_session_turn}; pub use types::{ BackgroundAppend, BackgroundAppendOutcome, BackgroundOrigin, CompactionMarker, DisplayMessage, DisplayRecord, DisplaySessionTranscript, LegacyText, MessageUsage, SessionTranscript, - ToolFailure, TranscriptMediaRef, TranscriptMessage, TranscriptMeta, TranscriptPart, - TranscriptToolCall, TurnUsage, UsageCostSource, + ThreadTranscript, ThreadTranscripts, ToolFailure, TranscriptMediaRef, TranscriptMessage, + TranscriptMeta, TranscriptPart, TranscriptToolCall, TurnUsage, UsageCostSource, }; +pub(crate) use writer::write_transcript_with_tools; pub use writer::{ append_interrupted_partial, append_tools_record, append_transcript_turn, append_transcript_turn_with_partial, write_transcript, write_transcript_if_absent, @@ -182,6 +189,9 @@ pub use writer::{ // ── Tests ───────────────────────────────────────────────────────────── +#[cfg(test)] +#[path = "transcript/locator_parity_tests.rs"] +mod locator_parity_tests; #[cfg(test)] #[path = "transcript/transcript_tests.rs"] mod test; diff --git a/crates/tinyagents-session/src/transcript/README.md b/crates/tinyagents-session/src/transcript/README.md index 1d34cedd1..77e419f78 100644 --- a/crates/tinyagents-session/src/transcript/README.md +++ b/crates/tinyagents-session/src/transcript/README.md @@ -72,6 +72,11 @@ Re-exported from `transcript::` (see `../transcript.rs`); reachable as `"background": {"idempotency_key", "provenance"}` field, so the model-context reader resumes it as an assistant row and the display reader exposes the origin as `DisplayMessage::background`. +- `append_background_message_in(&dyn TranscriptLocator, …)` — the same + contract through any locator. A file-backed one (`file_backing()`) writes + the identical JSONL line; any other store appends through its history + handle and keeps the provenance in the row's `extra_metadata` under + `BACKGROUND_METADATA_KEY`, which the display read lifts back out. - `lock_session_turn(&dyn TranscriptLocator, &SessionRef) -> Option` (async) — the per-session, in-process turn lock, keyed by `destination_key()` + the generation-0 stem. A live turn holds it @@ -87,12 +92,22 @@ Re-exported from `transcript::` (see `../transcript.rs`); reachable as `.md` reader when no `.jsonl` exists. - `read_transcript_display` — display projection: every record in file order, including pre-compaction history and interrupted partials. +- `TranscriptRead::read_display()` — the same projection through any handle + (a driver store replays its entry log: a later `set` is a compaction + marker, a `clear` an empty one, a partial an interrupted record; entries + keep no clock, so markers carry no `ts`). +- `TranscriptLocator::thread_transcripts(thread_id, agent_id)` — the roots + (session chains oldest first, pre-identity roots dropped once a session + root exists) and sub-agent transcripts backing a thread, as + `ThreadTranscripts`. `view::get_page_in` and `spend::thread_spend_in` take a + locator and use it, so a host never names a transcript file. ### Spend and cost provenance (`spend.rs`) - `transcript_spend(&SessionTranscript)` sums usage in the logical context. Compaction removes older rows from that view; use - `thread_spend(workspace_dir, thread_id)` for the full append history. + `thread_spend(workspace_dir, thread_id)` (or `thread_spend_in(locator, + thread_id)` for any store) for the full append history. - `thread_spend` returns `ThreadSpend`: `root` contains the orchestrator's own spend, and `subagents` groups descendant runs by archetype, including children whose headers name a different worker thread. Retained root rows diff --git a/crates/tinyagents-session/src/transcript/agent_files.rs b/crates/tinyagents-session/src/transcript/agent_files.rs index 301420718..0e04e7f52 100644 --- a/crates/tinyagents-session/src/transcript/agent_files.rs +++ b/crates/tinyagents-session/src/transcript/agent_files.rs @@ -11,8 +11,8 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use super::{ - FileTranscriptLocator, SessionAdoption, SessionRef, TranscriptHistory, TranscriptLocator, - TranscriptMessage, TranscriptMeta, TranscriptPartial, TranscriptRead, + FileTranscriptLocator, SessionAdoption, SessionRef, ThreadTranscripts, TranscriptHistory, + TranscriptLocator, TranscriptMessage, TranscriptMeta, TranscriptPartial, TranscriptRead, find_root_transcript_for_thread_scoped, resolve_keyed_transcript_path, session_stem, }; @@ -97,6 +97,44 @@ impl AgentTranscriptFiles { } impl TranscriptLocator for AgentTranscriptFiles { + fn file_backing(&self) -> Option<&FileTranscriptLocator> { + Some(&self.own) + } + + fn thread_transcripts( + &self, + thread_id: &str, + agent_id: Option<&str>, + ) -> Option { + let own = self.own.thread_transcripts(thread_id, agent_id); + let legacy = self.legacy.thread_transcripts(thread_id, agent_id); + match (own, legacy) { + (Some(mut own), Some(legacy)) => { + // The agent's roots supersede the shared ones (an adopted + // root is a copy), but sub-agents spawned before the move + // exist only in the shared directory: list both, the agent's + // copy winning a stem both hold. + let held: std::collections::HashSet = + own.subagents.iter().map(|sub| sub.stem.clone()).collect(); + let shared: Vec<_> = legacy + .subagents + .into_iter() + .filter(|sub| !held.contains(&sub.stem)) + .collect(); + if !shared.is_empty() { + tracing::debug!( + "[session_store] listing {} shared sub-agent transcript(s) beside the agent's own", + shared.len() + ); + own.subagents.extend(shared); + own.subagents.sort_by(|a, b| a.stem.cmp(&b.stem)); + } + Some(own) + } + (own, legacy) => own.or(legacy), + } + } + fn destination_key(&self) -> Option { self.own.destination_key() } diff --git a/crates/tinyagents-session/src/transcript/agent_files_tests.rs b/crates/tinyagents-session/src/transcript/agent_files_tests.rs index ecf388632..5fa5f07dd 100644 --- a/crates/tinyagents-session/src/transcript/agent_files_tests.rs +++ b/crates/tinyagents-session/src/transcript/agent_files_tests.rs @@ -149,3 +149,70 @@ fn fixture_workspace_resumes_by_stem_without_readoption_or_renaming() { let again = AgentTranscriptFiles::new(workspace.path(), "alpha"); assert!(again.session_exists(&session)); } + +#[test] +fn shared_sub_agents_stay_listed_after_the_root_is_adopted() { + let workspace = tempfile::tempdir().expect("tempdir"); + let session = SessionRef::scoped("thread-1", "alpha"); + let shared = FileTranscriptLocator::new(workspace.path()); + shared + .open_session(&session, meta()) + .expect("bind shared") + .replace(&[TranscriptMessage::user("before")]) + .expect("write shared"); + // Sub-agents spawned before the agent kept its own directory, one of + // which the agent's directory also holds by the same stem. + for stem in ["1700000000_alpha__1_old", "1700000000_alpha__2_both"] { + shared + .open_stem(stem, meta()) + .expect("bind shared sub-agent") + .replace(&[TranscriptMessage::user(stem)]) + .expect("write shared sub-agent"); + } + + let files = AgentTranscriptFiles::new(workspace.path(), "alpha"); + files + .open_session(&session, meta()) + .expect("bind own") + .replace(&[ + TranscriptMessage::user("before"), + TranscriptMessage::user("after"), + ]) + .expect("continue"); + for stem in ["1700000000_alpha__2_both", "1700000000_alpha__3_new"] { + files + .open_stem(stem, meta()) + .expect("bind own sub-agent") + .replace(&[TranscriptMessage::user(stem)]) + .expect("write own sub-agent"); + } + + let listed = files + .thread_transcripts("thread-1", None) + .expect("the thread is listed"); + let roots: Vec<_> = listed.roots.iter().map(|root| root.stem.clone()).collect(); + assert_eq!(roots, [session_stem(&session)]); + assert_eq!( + messages(listed.roots[0].transcript.as_ref()), + ["before", "after"], + "the agent's own root is listed" + ); + let subs: Vec<_> = listed + .subagents + .iter() + .map(|sub| sub.stem.clone()) + .collect(); + assert_eq!( + subs, + [ + "1700000000_alpha__1_old", + "1700000000_alpha__2_both", + "1700000000_alpha__3_new" + ] + ); + let own_root = agent_transcript_root(workspace.path(), "alpha"); + assert!( + listed.subagents[1].transcript.path().starts_with(&own_root), + "the agent's copy wins a stem both directories hold" + ); +} diff --git a/crates/tinyagents-session/src/transcript/background.rs b/crates/tinyagents-session/src/transcript/background.rs index 6b65bae79..30d3f994a 100644 --- a/crates/tinyagents-session/src/transcript/background.rs +++ b/crates/tinyagents-session/src/transcript/background.rs @@ -14,7 +14,7 @@ //! line written between that read and the persist makes the baseline stale and //! the user's turn fails to persist. So: //! -//! - **A live turn must hold [`lock_session_turn`] from its resume read +//! - **A live turn must hold [`lock_session_turn`](super::lock_session_turn) from its resume read //! through its persist.** `tinyagents-runtime`'s `Session::turn` does this //! for every session-bound target whose locator names its destination. //! A host driving the transcript some other way must take the lock itself. @@ -33,15 +33,16 @@ //! corrupted), and the append itself takes the same cross-process file locks //! as every other writer, so bytes never interleave. +use super::display::with_background; use super::history::seed_meta_for_discovered; use super::history::{FileTranscriptHistory, FileTranscriptLocator, TranscriptLocator}; use super::jsonl::{LineKind, build_message_line, classify_line}; use super::paths::resolve_keyed_transcript_path; use super::reader::read_jsonl_lines; use super::session::{SessionRef, session_stem}; -use super::turn_lock::lock_session_turn_with_notification; +use super::turn_lock::{lock_session_turn_or_identity, lock_session_turn_with_notification}; use super::types::{ - BackgroundAppend, BackgroundAppendOutcome, BackgroundOrigin, TranscriptMessage, + BackgroundAppend, BackgroundAppendOutcome, BackgroundOrigin, DisplayRecord, TranscriptMessage, }; use super::writer::append_bytes; use anyhow::Context; @@ -63,8 +64,11 @@ const MAX_HEAD_MOVES: usize = 3; /// - **Idempotent.** A head generation that already holds a line with the same /// [`BackgroundAppend::idempotency_key`] returns /// [`BackgroundAppendOutcome::Duplicate`] and writes nothing. The key is -/// checked within the head generation only: once a compaction has carried -/// the message into a successor, that successor holds it as an ordinary row. +/// checked within the head generation only. A compaction that carries the +/// message into a successor carries its provenance too (the model-context +/// row keeps it in `extra_metadata`, and the writer puts it back on the +/// line), so a retry against that successor is still a duplicate; one the +/// compaction dropped is appended again. /// - **Never creates a session.** No transcript yet is /// [`BackgroundAppendOutcome::NoSession`]. /// @@ -92,7 +96,7 @@ pub async fn append_background_message( message: TranscriptMessage, options: BackgroundAppend, ) -> anyhow::Result { - append_background_message_impl(locator, session, message, options, None).await + append_background_message_impl(locator, locator, session, message, options, None).await } /// Test-oriented variant that signals immediately before waiting for the @@ -105,11 +109,24 @@ pub async fn append_background_message_with_lock_notification( options: BackgroundAppend, lock_attempted: &Notify, ) -> anyhow::Result { - append_background_message_impl(locator, session, message, options, Some(lock_attempted)).await + append_background_message_impl( + locator, + locator, + session, + message, + options, + Some(lock_attempted), + ) + .await } +/// `files` writes the JSONL line; `resolver` resolves the head and decides +/// whether the session exists. They differ for a wrapper over a file locator +/// (such as [`AgentTranscriptFiles`](super::AgentTranscriptFiles)) whose +/// existence checks adopt a transcript into `files` before it is written. async fn append_background_message_impl( - locator: &FileTranscriptLocator, + files: &FileTranscriptLocator, + resolver: &dyn TranscriptLocator, session: &SessionRef, message: TranscriptMessage, options: BackgroundAppend, @@ -121,10 +138,10 @@ async fn append_background_message_impl( let key = options.idempotency_key.as_str(); tracing::debug!(session = %stem, idempotency_key = %key, expected_generation = ?options.expected_generation, "[transcript-background] append requested"); - let _turn = lock_session_turn_with_notification(locator, &root, lock_attempted).await; + let _turn = lock_session_turn_with_notification(resolver, &root, lock_attempted).await; for _ in 0..MAX_HEAD_MOVES { - let head = locator.head_generation(&root); - if !locator.session_exists(&head) { + let head = resolver.head_generation(&root); + if !resolver.session_exists(&head) { tracing::debug!(session = %stem, idempotency_key = %key, "[transcript-background] no transcript; nothing written"); return Ok(BackgroundAppendOutcome::NoSession); } @@ -132,7 +149,7 @@ async fn append_background_message_impl( tracing::debug!(session = %stem, idempotency_key = %key, ?stale, "[transcript-background] head moved on; nothing written"); return Ok(stale); } - match append_to_head(locator, &head, &message, &options)? { + match append_to_head(files, &head, &message, &options)? { Attempt::Done(outcome) => { tracing::debug!(session = %stem, idempotency_key = %key, ?outcome, "[transcript-background] append finished"); return Ok(outcome); @@ -147,6 +164,121 @@ async fn append_background_message_impl( ) } +/// [`append_background_message`] through any [`TranscriptLocator`]. +/// +/// A file-backed locator ([`TranscriptLocator::file_backing`]) goes through +/// the file writer, so the JSONL line is exactly what +/// [`append_background_message`] writes. Any other store gets the same +/// contract — the turn lock, head-only, idempotent within the head +/// generation, never creating a session — with the message appended through +/// its [`TranscriptHistory`](super::TranscriptHistory) handle and the +/// provenance recorded in the row's `extra_metadata` under +/// [`BACKGROUND_METADATA_KEY`](super::BACKGROUND_METADATA_KEY), which the +/// display read lifts back into [`DisplayMessage::background`](super::DisplayMessage::background). +/// +/// For such a store the turn lock is the only serialisation: the duplicate +/// check and the append are two calls, so two processes appending the same key +/// to one store at once could both write it. A locator with no +/// [`destination_key`](TranscriptLocator::destination_key) is locked by its +/// own identity instead, so appends through that one locator still serialise. +/// +/// # Errors +/// +/// As [`append_background_message`], plus any store error. +pub async fn append_background_message_in( + locator: &dyn TranscriptLocator, + session: &SessionRef, + message: TranscriptMessage, + options: BackgroundAppend, +) -> anyhow::Result { + if let Some(files) = locator.file_backing() { + // Existence and head resolution go through `locator` itself, so a + // wrapper that adopts transcripts on lookup (an agent's own files over + // a shared workspace) still finds a session only its fallback holds. + return append_background_message_impl(files, locator, session, message, options, None) + .await; + } + let message = validate(message, &options)?; + let root = session.first_generation(); + let stem = session_stem(&root); + let key = options.idempotency_key.as_str(); + tracing::debug!(session = %stem, idempotency_key = %key, expected_generation = ?options.expected_generation, "[transcript-background] store append requested"); + + // Falls back to the locator's own identity when it names no destination, + // so two appends through one such locator still serialise their + // duplicate check and append. + let _turn = lock_session_turn_or_identity(locator, &root).await; + for _ in 0..MAX_HEAD_MOVES { + let head = locator.head_generation(&root); + if !locator.session_exists(&head) { + tracing::debug!(session = %stem, idempotency_key = %key, "[transcript-background] no transcript in the store; nothing written"); + return Ok(BackgroundAppendOutcome::NoSession); + } + if let Some(stale) = stale(&options, head.generation) { + return Ok(stale); + } + match append_to_stored_head(locator, &head, &message, &options)? { + Attempt::Done(outcome) => { + tracing::debug!(session = %stem, idempotency_key = %key, ?outcome, "[transcript-background] store append finished"); + return Ok(outcome); + } + Attempt::HeadMoved => { + tracing::debug!(session = %stem, idempotency_key = %key, generation = head.generation, "[transcript-background] store head sealed during append; re-resolving"); + } + } + } + anyhow::bail!( + "session {stem} head kept moving during a background append ({MAX_HEAD_MOVES} attempts)" + ) +} + +/// One attempt against a store's resolved `head`. +fn append_to_stored_head( + locator: &dyn TranscriptLocator, + head: &SessionRef, + message: &TranscriptMessage, + options: &BackgroundAppend, +) -> anyhow::Result { + if locator.session_exists(&head.next_generation()) { + return Ok(Attempt::HeadMoved); + } + let generation = head.generation; + let handle = locator.open_session(head, seed_meta_for_discovered(""))?; + let key = options.idempotency_key.as_str(); + let held = handle.read_display()?.is_some_and(|display| { + display.records.iter().any(|record| match record { + DisplayRecord::Message(message) => carries_key(message.background.as_ref(), key), + DisplayRecord::Compaction(compaction) => compaction + .replacement + .iter() + .any(|message| carries_key(message.background.as_ref(), key)), + }) + }); + if held { + return Ok(Attempt::Done(BackgroundAppendOutcome::Duplicate { + generation, + })); + } + let origin = BackgroundOrigin { + idempotency_key: options.idempotency_key.clone(), + provenance: options.provenance.clone(), + }; + if let Err(error) = handle.append(with_background(message.clone(), &origin)) { + // A compaction sealed the head between the check and the write. + if locator.session_exists(&head.next_generation()) { + return Ok(Attempt::HeadMoved); + } + return Err(error); + } + Ok(Attempt::Done(BackgroundAppendOutcome::Appended { + generation, + })) +} + +fn carries_key(origin: Option<&BackgroundOrigin>, key: &str) -> bool { + origin.is_some_and(|origin| origin.idempotency_key == key) +} + /// The result of one locked attempt against a resolved head. enum Attempt { Done(BackgroundAppendOutcome), diff --git a/crates/tinyagents-session/src/transcript/background_tests.rs b/crates/tinyagents-session/src/transcript/background_tests.rs index 564f5c505..1a02737da 100644 --- a/crates/tinyagents-session/src/transcript/background_tests.rs +++ b/crates/tinyagents-session/src/transcript/background_tests.rs @@ -429,3 +429,147 @@ fn an_attempt_against_a_sealed_head_writes_nothing_and_asks_to_re_resolve() { assert!(matches!(attempt, Attempt::HeadMoved)); assert_eq!(std::fs::read(&sealed).unwrap(), before); } + +#[tokio::test] +async fn an_agents_files_adopt_a_shared_session_before_the_append() { + let dir = tempdir().unwrap(); + // The conversation exists only in the shared workspace, as written + // before the agent kept its own directory. + let shared = FileTranscriptLocator::new(dir.path()); + live_turn(&shared, &session(), &[], &first_turn()); + let shared_path = head_path(dir.path(), &session()); + let before = std::fs::read(&shared_path).unwrap(); + + let agent = super::super::AgentTranscriptFiles::new(dir.path(), "orchestrator"); + let outcome = append_background_message_in( + &agent, + &session(), + TranscriptMessage::assistant("Time to stretch!"), + options("run-1"), + ) + .await + .unwrap(); + + assert_eq!(outcome, BackgroundAppendOutcome::Appended { generation: 0 }); + let own_root = super::super::agent_transcript_root(dir.path(), "orchestrator"); + let own = read_transcript(&head_path(&own_root, &session())).unwrap(); + assert_eq!( + own.messages.last().unwrap().content, + "Time to stretch!", + "the adopted copy receives the message" + ); + assert_eq!( + std::fs::read(&shared_path).unwrap(), + before, + "the shared file is never modified" + ); +} + +/// A store-like locator that names no destination (and is not file backed), +/// over real transcript files. +struct Unnamed(FileTranscriptLocator); + +impl TranscriptLocator for Unnamed { + fn latest_for_agent( + &self, + agent_name: &str, + ) -> Option> { + self.0.latest_for_agent(agent_name) + } + fn root_for_thread( + &self, + thread_id: &str, + ) -> Option> { + self.0.root_for_thread(thread_id) + } + fn open_stem( + &self, + stem: &str, + seed: TranscriptMeta, + ) -> anyhow::Result> { + self.0.open_stem(stem, seed) + } +} + +#[tokio::test] +async fn appends_through_one_unnamed_locator_are_serialised() { + let dir = tempdir().unwrap(); + let files = FileTranscriptLocator::new(dir.path()); + live_turn(&files, &session(), &[], &first_turn()); + let locator = std::sync::Arc::new(Unnamed(FileTranscriptLocator::new(dir.path()))); + assert!(locator.destination_key().is_none()); + + let held = super::super::turn_lock::lock_session_turn_or_identity( + locator.as_ref() as &dyn TranscriptLocator, + &session(), + ) + .await; + let delivery = { + let locator = std::sync::Arc::clone(&locator); + tokio::spawn(async move { + append_background_message_in( + locator.as_ref(), + &session(), + TranscriptMessage::assistant("Time to stretch!"), + options("run-1"), + ) + .await + }) + }; + for _ in 0..50 { + tokio::task::yield_now().await; + } + assert!( + !delivery.is_finished(), + "an append through the same unnamed locator waits for its lock" + ); + drop(held); + assert_eq!( + delivery.await.unwrap().unwrap(), + BackgroundAppendOutcome::Appended { generation: 0 } + ); +} + +/// A compaction that keeps the delivered message carries its provenance into +/// the successor, so a retry is a duplicate there too — as it is for a store +/// that keeps rows. +#[tokio::test] +async fn a_delivery_carried_through_a_compaction_is_still_a_duplicate() { + let dir = tempdir().unwrap(); + let locator = FileTranscriptLocator::new(dir.path()); + live_turn(&locator, &session(), &[], &first_turn()); + let bound = session(); + let deliver = || { + append_background_message( + &locator, + &bound, + TranscriptMessage::assistant("Time to stretch!"), + options("run-1"), + ) + }; + assert_eq!( + deliver().await.unwrap(), + BackgroundAppendOutcome::Appended { generation: 0 } + ); + + // The runtime compacts from the context it read back. + let retained = read_transcript(&head_path(dir.path(), &session())) + .unwrap() + .messages; + compact(&locator, &retained); + + assert_eq!( + deliver().await.unwrap(), + BackgroundAppendOutcome::Duplicate { generation: 1 } + ); + let head = read_transcript_display(&head_path(dir.path(), &session())).unwrap(); + let carried = head + .records + .iter() + .filter_map(|record| match record { + DisplayRecord::Message(message) => message.background.as_ref(), + DisplayRecord::Compaction(_) => None, + }) + .count(); + assert_eq!(carried, 1, "the successor's row keeps its provenance"); +} diff --git a/crates/tinyagents-session/src/transcript/display.rs b/crates/tinyagents-session/src/transcript/display.rs new file mode 100644 index 000000000..1d589fed8 --- /dev/null +++ b/crates/tinyagents-session/src/transcript/display.rs @@ -0,0 +1,129 @@ +//! Display records built from the neutral transcript types. +//! +//! The JSONL reader turns each file line into a [`DisplayMessage`]. A store +//! that does not keep lines (the driver-backed one) holds [`TranscriptMessage`] +//! rows instead; these helpers give it the same display shape by running each +//! row through the same line codec the file writer uses, so a projection reads +//! the two identically. +//! +//! Out-of-band delivery provenance ([`BackgroundOrigin`]) is a line field in +//! JSONL. A [`TranscriptMessage`] carries it in its `extra_metadata` under +//! [`BACKGROUND_METADATA_KEY`] — the JSONL reader lifts the line field in +//! there and the line codec writes it back out as the line field — so a row +//! keeps its provenance through every read and rewrite on either backend; +//! [`display_message_of`] lifts it back out for display. + +use super::history::TranscriptPartial; +use super::jsonl::{build_message_line, display_message_from_line, partial_line}; +use super::types::{ + BackgroundOrigin, DisplayMessage, DisplayRecord, DisplaySessionTranscript, SessionTranscript, + TranscriptMessage, +}; + +/// The `extra_metadata` key a row-keeping store records a background +/// delivery's provenance under. +pub const BACKGROUND_METADATA_KEY: &str = "tinyagents_background"; + +/// The display form of one stored row, exactly as the file reader would +/// produce it for the line the file writer makes of that row. +pub(crate) fn display_message_of(message: &TranscriptMessage) -> DisplayMessage { + let mut message = message.clone(); + let background = take_background(&mut message); + let mut line = build_message_line( + &message, + message.turn_usage.as_ref(), + message.request_id.as_deref(), + message.interrupted, + ); + line.background = background; + display_message_from_line(line) +} + +/// The display form of an interrupted turn's partial answer. +pub(crate) fn partial_display_message( + partial: &TranscriptPartial, + request_id: Option<&str>, +) -> DisplayMessage { + display_message_from_line(partial_line(partial, request_id)) +} + +/// A display projection of `session`'s model context: one message record per +/// row. For a store that keeps no more than the reduced context. +pub(crate) fn display_from_session(session: SessionTranscript) -> DisplaySessionTranscript { + DisplaySessionTranscript { + records: session + .messages + .iter() + .map(|message| DisplayRecord::Message(Box::new(display_message_of(message)))) + .collect(), + meta: session.meta, + } +} + +/// `message` marked as delivered out of band by `origin`, for a store that +/// keeps rows rather than lines. +pub(crate) fn with_background( + mut message: TranscriptMessage, + origin: &BackgroundOrigin, +) -> TranscriptMessage { + message.extra_metadata = metadata_with_background(message.extra_metadata.take(), origin); + message +} + +/// `metadata` with `origin` recorded under [`BACKGROUND_METADATA_KEY`]. +pub(crate) fn metadata_with_background( + metadata: Option, + origin: &BackgroundOrigin, +) -> Option { + let mut metadata = match metadata { + Some(serde_json::Value::Object(map)) => map, + Some(other) => { + let mut map = serde_json::Map::new(); + map.insert("value".into(), other); + map + } + None => serde_json::Map::new(), + }; + metadata.insert( + BACKGROUND_METADATA_KEY.into(), + serde_json::to_value(origin).unwrap_or(serde_json::Value::Null), + ); + Some(serde_json::Value::Object(metadata)) +} + +/// Splits the background provenance recorded under +/// [`BACKGROUND_METADATA_KEY`] out of `metadata`: the line codec writes it as +/// the line's own `background` field instead. +pub(crate) fn split_background( + metadata: Option, +) -> (Option, Option) { + let mut message = TranscriptMessage::assistant(""); + message.extra_metadata = metadata; + let origin = take_background(&mut message); + (message.extra_metadata, origin) +} + +/// The background provenance recorded on a stored row, if any. +fn background_of(message: &TranscriptMessage) -> Option { + message + .extra_metadata + .as_ref() + .and_then(|metadata| metadata.get(BACKGROUND_METADATA_KEY)) + .and_then(|value| serde_json::from_value(value.clone()).ok()) +} + +/// Removes the background provenance from `message`'s metadata, returning it. +fn take_background(message: &mut TranscriptMessage) -> Option { + let origin = background_of(message)?; + if let Some(serde_json::Value::Object(map)) = message.extra_metadata.as_mut() { + map.remove(BACKGROUND_METADATA_KEY); + if map.is_empty() { + message.extra_metadata = None; + } + } + Some(origin) +} + +#[cfg(test)] +#[path = "display_tests.rs"] +mod tests; diff --git a/crates/tinyagents-session/src/transcript/display_tests.rs b/crates/tinyagents-session/src/transcript/display_tests.rs new file mode 100644 index 000000000..611be7915 --- /dev/null +++ b/crates/tinyagents-session/src/transcript/display_tests.rs @@ -0,0 +1,76 @@ +use super::*; + +fn origin() -> BackgroundOrigin { + BackgroundOrigin { + idempotency_key: "run-1".into(), + provenance: serde_json::json!({"kind": "cron"}), + } +} + +#[test] +fn background_provenance_round_trips_through_row_metadata() { + let row = with_background(TranscriptMessage::assistant("hi"), &origin()); + assert_eq!(background_of(&row), Some(origin())); + let display = display_message_of(&row); + assert_eq!(display.background, Some(origin())); + assert!(display.message.extra_metadata.is_none()); +} + +#[test] +fn other_metadata_survives_the_lift() { + let mut row = TranscriptMessage::assistant("hi"); + row.extra_metadata = Some(serde_json::json!({"keep": 1})); + let display = display_message_of(&with_background(row, &origin())); + assert_eq!( + display.message.extra_metadata, + Some(serde_json::json!({"keep": 1})) + ); + assert_eq!(display.background, Some(origin())); +} + +#[test] +fn non_object_metadata_is_kept_under_value() { + let mut row = TranscriptMessage::assistant("hi"); + row.extra_metadata = Some(serde_json::json!("raw")); + let display = display_message_of(&with_background(row, &origin())); + assert_eq!( + display.message.extra_metadata, + Some(serde_json::json!({"value": "raw"})) + ); +} + +#[test] +fn a_plain_row_has_no_background() { + let display = display_message_of(&TranscriptMessage::user("hello")); + assert!(display.background.is_none()); + assert_eq!(display.message.content, "hello"); + assert!(!display.interrupted); +} + +#[test] +fn a_partial_is_an_interrupted_assistant_record() { + let mut partial = TranscriptPartial::new("half"); + partial.iteration = Some(2); + partial.reasoning_content = Some(" thinking ".into()); + let display = partial_display_message(&partial, Some("r1")); + assert!(display.interrupted); + assert_eq!(display.message.role, "assistant"); + assert_eq!(display.request_id.as_deref(), Some("r1")); + assert_eq!(display.iteration, Some(2)); + assert_eq!(display.reasoning_content.as_deref(), Some("thinking")); +} + +#[test] +fn a_session_projects_one_record_per_message() { + let session = SessionTranscript { + meta: crate::transcript::history::seed_meta_for_discovered("a"), + messages: vec![ + TranscriptMessage::user("q"), + TranscriptMessage::assistant("a"), + ], + tools: None, + }; + let display = display_from_session(session); + assert_eq!(display.records.len(), 2); + assert_eq!(display.meta.agent_name, "a"); +} diff --git a/crates/tinyagents-session/src/transcript/history.rs b/crates/tinyagents-session/src/transcript/history.rs index 67312b486..6e3e17b16 100644 --- a/crates/tinyagents-session/src/transcript/history.rs +++ b/crates/tinyagents-session/src/transcript/history.rs @@ -26,12 +26,26 @@ use crate::transcript::types::TranscriptMessage; use fs2::FileExt; use crate::transcript::{ - SessionAdoption, SessionRef, SessionTranscript, TranscriptMeta, TurnUsage, - adopt_legacy_session_transcripts, find_latest_transcript, find_root_transcript_for_thread, + DisplaySessionTranscript, SessionAdoption, SessionRef, SessionTranscript, ThreadTranscript, + ThreadTranscripts, TranscriptMeta, TurnUsage, adopt_legacy_session_transcripts, + find_latest_transcript, find_root_transcript_for_thread, find_root_transcript_for_thread_scoped, read_transcript, resolve_keyed_transcript_path, session_stem, }; +/// The stem a read handle is bound to: its path's last component, without the +/// `.jsonl` extension a file handle carries. (A stem may itself contain dots, +/// so this never uses `file_stem`.) +pub(super) fn stem_of_handle(path: &Path) -> String { + let name = path + .file_name() + .map(|name| name.to_string_lossy().into_owned()) + .unwrap_or_default(); + name.strip_suffix(".jsonl") + .map(str::to_string) + .unwrap_or(name) +} + /// Upper bound on the compaction generations one session may accumulate. /// /// Generation resolution probes `{stem}`, `{stem}.g1`, `{stem}.g2` … on disk @@ -163,6 +177,17 @@ pub trait TranscriptRead: Send + Sync { /// [`SessionTranscript`] rather than messages alone is what lets the shadow /// read keep working through this seam. fn read_session(&self) -> anyhow::Result>; + + /// The display projection of this transcript — every record, including + /// pre-compaction history, compaction markers and interrupted partials — + /// or `Ok(None)` when it does not exist. + /// + /// The default builds it from [`Self::read_session`], which holds only + /// the reduced model context: an implementor that keeps more (the file + /// log, a driver's entry log) overrides it. + fn read_display(&self) -> anyhow::Result> { + Ok(self.read_session()?.map(super::display_from_session)) + } } /// Resolves transcripts by the two keys the turn path actually has, and binds @@ -217,6 +242,37 @@ pub trait TranscriptLocator: Send + Sync { self.root_for_thread(thread_id) } + /// Every transcript backing `thread_id`'s view: its root generations in + /// projection order and its sub-agent transcripts, or `None` when the + /// thread has no root transcript yet. Scoped to the roots `agent_id` owns + /// when given (sub-agents are then found by stem only). + /// + /// Defaults to [`Self::root_for_thread_scoped`] alone, with no + /// sub-agents, for a locator that cannot list. + fn thread_transcripts( + &self, + thread_id: &str, + agent_id: Option<&str>, + ) -> Option { + let root = self.root_for_thread_scoped(thread_id, agent_id)?; + Some(ThreadTranscripts { + roots: vec![ThreadTranscript { + stem: stem_of_handle(root.path()), + transcript: root, + }], + subagents: Vec::new(), + }) + } + + /// The file locator this one writes through, when it is file-backed. + /// + /// Lets a generic writer keep the JSONL line format exact (for example + /// [`append_background_message_in`](super::append_background_message_in) + /// records provenance as a line field there). `None` for any other store. + fn file_backing(&self) -> Option<&FileTranscriptLocator> { + None + } + /// Binds (creating on first write) this session's own write handle for /// `stem`, with `seed` used only when no file exists yet. fn open_stem( @@ -557,6 +613,40 @@ impl FileTranscriptLocator { } impl TranscriptLocator for FileTranscriptLocator { + fn file_backing(&self) -> Option<&FileTranscriptLocator> { + Some(self) + } + + /// The same file set the path-based view projects + /// ([`resolve_files_scoped`](crate::transcript::view::resolve_files_scoped)). + fn thread_transcripts( + &self, + thread_id: &str, + agent_id: Option<&str>, + ) -> Option { + let (roots, subagents) = crate::transcript::view::resolve_files_scoped( + &self.workspace_dir, + thread_id, + agent_id, + )?; + let listed = |paths: Vec| -> Vec { + paths + .into_iter() + .map(|path| { + let stem = stem_of_handle(&path); + let transcript: Arc = Arc::new( + FileTranscriptHistory::opened_at(path, seed_meta_for_discovered(&stem)), + ); + ThreadTranscript { stem, transcript } + }) + .collect() + }; + Some(ThreadTranscripts { + roots: listed(roots), + subagents: listed(subagents), + }) + } + /// The workspace root every lookup and bind resolves under, which is /// this locator's only field and therefore its whole identity. fn destination_key(&self) -> Option { @@ -1280,6 +1370,19 @@ impl TranscriptRead for FileTranscriptHistory { ); Ok(Some(session)) } + + /// [`read_transcript_display`](crate::transcript::read_transcript_display) + /// on this handle's file. A legacy `.md` transcript has no display log, so + /// it projects its model context instead. + fn read_display(&self) -> anyhow::Result> { + if !self.path.exists() { + return Ok(None); + } + if self.path.extension().and_then(|ext| ext.to_str()) != Some("jsonl") { + return Ok(self.read_session()?.map(super::display_from_session)); + } + crate::transcript::read_transcript_display(&self.path).map(Some) + } } impl FileTranscriptHistory { diff --git a/crates/tinyagents-session/src/transcript/jsonl.rs b/crates/tinyagents-session/src/transcript/jsonl.rs index 9cef1ba73..0a8e3108e 100644 --- a/crates/tinyagents-session/src/transcript/jsonl.rs +++ b/crates/tinyagents-session/src/transcript/jsonl.rs @@ -2,6 +2,7 @@ //! conversions between them and the public [`TranscriptMessage`] / //! [`TranscriptMeta`] / [`DisplayMessage`] types. +use super::display::{metadata_with_background, split_background}; use super::types::{ BackgroundOrigin, DisplayMessage, MessageUsage, TRANSCRIPT_SCHEMA_VERSION, TranscriptMeta, TurnUsage, @@ -317,7 +318,8 @@ pub(super) fn build_message_line( } else { None }; - let extra_metadata = msg.extra_metadata.clone(); + // Provenance a row carries in its metadata is a line field on disk. + let (extra_metadata, background) = split_background(msg.extra_metadata.clone()); let failure = msg .tool_failure .as_ref() @@ -391,7 +393,7 @@ pub(super) fn build_message_line( shape: typed.as_ref().map(|typed| typed.shape.as_str().to_string()), tool_call_id: typed.as_ref().and_then(|typed| typed.tool_call_id.clone()), parts: typed.and_then(|typed| typed.parts), - background: None, + background, _extra: HashMap::new(), } } @@ -773,6 +775,12 @@ pub(super) fn message_from_line(ml: MessageLine) -> TranscriptMessage { let turn_usage = turn_usage_from_line(&ml); let failure_detail = ml.failure.then(|| ml.failure_detail.clone()); let typed = row_from_line(&ml); + // The row keeps a background delivery's provenance, so a rewrite or a + // compaction carrying it writes the line field back. + let extra_metadata = match &ml.background { + Some(origin) => metadata_with_background(ml.extra_metadata, origin), + None => ml.extra_metadata, + }; TranscriptMessage { id: typed.id, role: typed.role, @@ -781,7 +789,7 @@ pub(super) fn message_from_line(ml: MessageLine) -> TranscriptMessage { tool_call_id: typed.tool_call_id, parts: typed.parts, legacy: typed.legacy, - extra_metadata: ml.extra_metadata, + extra_metadata, cache_breakpoints: ml.cache_breakpoints, turn_usage: turn_usage.clone(), request_id: ml.request_id, @@ -865,3 +873,25 @@ pub(super) fn display_message_from_line(ml: MessageLine) -> DisplayMessage { }, } } + +/// The display-only line of an interrupted `partial`, without a timestamp +/// (the file writer stamps one; a store that kept none leaves it unset). +pub(super) fn partial_line( + partial: &super::history::TranscriptPartial, + request_id: Option<&str>, +) -> MessageLine { + let mut line = build_message_line( + &TranscriptMessage::assistant(&partial.content), + None, + request_id, + true, + ); + line.iteration = partial.iteration; + line.reasoning_content = partial + .reasoning_content + .as_deref() + .map(str::trim) + .filter(|content| !content.is_empty()) + .map(str::to_owned); + line +} diff --git a/crates/tinyagents-session/src/transcript/locator_parity_tests.rs b/crates/tinyagents-session/src/transcript/locator_parity_tests.rs new file mode 100644 index 000000000..3358d5c67 --- /dev/null +++ b/crates/tinyagents-session/src/transcript/locator_parity_tests.rs @@ -0,0 +1,403 @@ +//! The locator-generic reads and writes (`read_display`, +//! `thread_transcripts`, `view::get_page_in`, `spend::thread_spend_in`, +//! `append_background_message_in`) give the same answer over the JSONL files +//! and over a driver store, for one scenario written through the port alone. + +use std::sync::Arc; + +use serde_json::Value; +use tinystoragedrivers_core::MemoryStorage; + +use crate::port::{DriverSessionStores, SessionStoreProvider}; +use crate::transcript::spend::{ThreadSpend, thread_spend, thread_spend_in}; +use crate::transcript::view::{get_page, get_page_in}; +use crate::transcript::{ + BackgroundAppend, BackgroundAppendOutcome, DisplayRecord, FileTranscriptLocator, MessageUsage, + SessionRef, TranscriptLocator, TranscriptMessage, TranscriptMeta, TranscriptPartial, + TranscriptTurn, TurnUsage, append_background_message_in, session_stem, +}; + +const THREAD: &str = "thread-1"; +const AGENT: &str = "orchestrator"; +const CHILD_SUFFIX: &str = "1767225600_researcher"; + +fn meta(agent: &str, session: Option<&SessionRef>) -> TranscriptMeta { + TranscriptMeta { + session_id: session.map(SessionRef::session_id), + parent_session_id: session.and_then(SessionRef::parent_session_id), + agent_name: agent.into(), + agent_id: Some(agent.into()), + agent_type: Some( + if session.is_some() { + "root" + } else { + "subagent" + } + .into(), + ), + dispatcher: "native".into(), + provider: None, + model: Some("model-a".into()), + created: "2026-01-01T00:00:00Z".into(), + updated: "2026-01-01T00:00:00Z".into(), + turn_count: 1, + prefix_message_count: None, + input_tokens: 0, + output_tokens: 0, + cached_input_tokens: 0, + charged_amount_usd: 0.0, + thread_id: Some(THREAD.into()), + task_id: None, + } +} + +fn usage(input: u64, output: u64, cost: f64) -> TurnUsage { + TurnUsage { + provider: "provider".into(), + model: "model-a".into(), + usage: MessageUsage { + input, + output, + cost_usd: cost, + ..MessageUsage::default() + }, + ts: "2026-01-01T00:00:01Z".into(), + reasoning_content: None, + tool_calls: Vec::new(), + iteration: 1, + } +} + +fn root() -> SessionRef { + SessionRef::scoped(THREAD, AGENT) +} + +fn child_stem() -> String { + format!("{}__{CHILD_SUFFIX}", session_stem(&root())) +} + +/// Two turns (the second interrupted after a partial), a compaction into a +/// second generation with one more turn, and one sub-agent run. +fn write_scenario(locator: &dyn TranscriptLocator) { + let session = root(); + let seed = meta(AGENT, Some(&session)); + let handle = locator.open_session(&session, seed.clone()).unwrap(); + let first = vec![ + TranscriptMessage::system("system prompt"), + TranscriptMessage::user("question one"), + TranscriptMessage::assistant("answer one"), + ]; + handle + .append_turn(TranscriptTurn { + prev: &[], + next: &first, + meta: &seed, + turn_usage: Some(&usage(10, 5, 0.01)), + request_id: Some("r1"), + tools: None, + }) + .unwrap(); + let mut second = first.clone(); + second.push(TranscriptMessage::user("question two")); + second.push(TranscriptMessage::assistant("answer two")); + handle + .append_turn_with_partial( + TranscriptTurn { + prev: &handle.messages().unwrap(), + next: &second, + meta: &seed, + turn_usage: Some(&usage(20, 7, 0.02)), + request_id: Some("r2"), + tools: None, + }, + Some(&TranscriptPartial::new("half an answer")), + ) + .unwrap(); + + let (successor, next) = locator.begin_generation(&session, seed.clone()).unwrap(); + let successor_meta = meta(AGENT, Some(&successor)); + let compacted = vec![ + TranscriptMessage::system("system prompt"), + TranscriptMessage::assistant("summary of turns one and two"), + TranscriptMessage::user("question three"), + TranscriptMessage::assistant("answer three"), + ]; + next.append_turn(TranscriptTurn { + prev: &[], + next: &compacted, + meta: &successor_meta, + turn_usage: Some(&usage(30, 9, 0.03)), + request_id: Some("r3"), + tools: None, + }) + .unwrap(); + + let child_meta = meta("researcher", None); + let child = locator + .open_stem(&child_stem(), child_meta.clone()) + .unwrap(); + child + .append_turn(TranscriptTurn { + prev: &[], + next: &[ + TranscriptMessage::user("research this"), + TranscriptMessage::assistant("researched"), + ], + meta: &child_meta, + turn_usage: Some(&usage(3, 2, 0.005)), + request_id: None, + tools: None, + }) + .unwrap(); +} + +fn driver() -> Arc { + DriverSessionStores::new(Arc::new(MemoryStorage::new())) + .unwrap() + .for_agent(AGENT) + .transcripts +} + +/// `value` without wall-clock fields: the file writer stamps compaction +/// markers and partials with the time of the write, a driver store keeps none. +fn without_clock(mut value: Value) -> Value { + fn strip(value: &mut Value) { + match value { + Value::Object(map) => { + map.remove("ts"); + map.values_mut().for_each(strip); + } + Value::Array(items) => items.iter_mut().for_each(strip), + _ => {} + } + } + strip(&mut value); + value +} + +fn spend_summary(spend: &ThreadSpend) -> Value { + serde_json::json!({ + "found": spend.found_transcript, + "root": [spend.root.input_tokens, spend.root.output_tokens, spend.root.turns], + "root_cost": format!("{:.4}", spend.root.cost_usd), + "subagents": spend.subagents.iter().map(|(agent, (spend, runs))| { + serde_json::json!([agent, spend.input_tokens, spend.output_tokens, spend.turns, runs]) + }).collect::>(), + }) +} + +#[test] +fn the_thread_listing_matches_across_stores() { + let dir = tempfile::tempdir().unwrap(); + let files = FileTranscriptLocator::new(dir.path()); + let store = driver(); + write_scenario(&files); + write_scenario(store.as_ref()); + + let stems = |locator: &dyn TranscriptLocator| { + let listed = locator + .thread_transcripts(THREAD, None) + .expect("thread has roots"); + ( + listed + .roots + .iter() + .map(|t| t.stem.clone()) + .collect::>(), + listed + .subagents + .iter() + .map(|t| t.stem.clone()) + .collect::>(), + ) + }; + let (roots, subagents) = stems(&files); + assert_eq!( + roots, + vec![ + session_stem(&root()), + session_stem(&root().next_generation()) + ] + ); + assert_eq!(subagents, vec![child_stem()]); + assert_eq!(stems(store.as_ref()), (roots, subagents)); + assert!(files.thread_transcripts("no-such-thread", None).is_none()); + assert!(store.thread_transcripts("no-such-thread", None).is_none()); +} + +#[test] +fn the_display_read_matches_across_stores() { + let dir = tempfile::tempdir().unwrap(); + let files = FileTranscriptLocator::new(dir.path()); + let store = driver(); + write_scenario(&files); + write_scenario(store.as_ref()); + + let shape = |locator: &dyn TranscriptLocator| -> Vec { + let head = locator.read_session_transcript(&root()).unwrap(); + head.read_display() + .unwrap() + .unwrap() + .records + .iter() + .map(|record| match record { + DisplayRecord::Message(message) => format!( + "{}:{}:{}:{:?}", + message.message.role, + message.message.content, + message.interrupted, + message.request_id + ), + DisplayRecord::Compaction(marker) => { + format!("compaction:{}", marker.replacement.len()) + } + }) + .collect() + }; + let file_shape = shape(&files); + assert!( + file_shape + .iter() + .any(|row| row == "assistant:half an answer:true:Some(\"r2\")"), + "{file_shape:?}" + ); + assert_eq!(shape(store.as_ref()), file_shape); +} + +#[test] +fn the_projected_page_matches_across_stores_and_the_path_view() { + let dir = tempfile::tempdir().unwrap(); + let files = FileTranscriptLocator::new(dir.path()); + let store = driver(); + write_scenario(&files); + write_scenario(store.as_ref()); + + let page = |page: crate::transcript::view::TranscriptPage| { + without_clock(serde_json::to_value(page).unwrap()) + }; + let by_path = page(get_page(dir.path(), THREAD, None, Some(200))); + let file_page = page(get_page_in(&files, THREAD, None, None, Some(200))); + assert_eq!(file_page, by_path, "the locator view is the path view"); + assert_eq!(file_page["hasTranscript"], Value::Bool(true)); + let text = file_page.to_string(); + assert!(text.contains("answer three") && text.contains("researched")); + assert_eq!( + page(get_page_in(store.as_ref(), THREAD, None, None, Some(200))), + file_page + ); + + let empty = get_page_in(store.as_ref(), "no-such-thread", None, None, None); + assert!(!empty.has_transcript && empty.items.is_empty()); +} + +#[test] +fn spend_matches_across_stores_and_the_path_walk() { + let dir = tempfile::tempdir().unwrap(); + let files = FileTranscriptLocator::new(dir.path()); + let store = driver(); + write_scenario(&files); + write_scenario(store.as_ref()); + + let by_path = spend_summary(&thread_spend(dir.path(), THREAD)); + let file_spend = spend_summary(&thread_spend_in(&files, THREAD)); + assert_eq!(file_spend, by_path); + assert_eq!( + file_spend["subagents"], + serde_json::json!([["researcher", 3, 2, 1, 1]]), + "the sub-agent run is counted once, under its archetype" + ); + assert_eq!( + spend_summary(&thread_spend_in(store.as_ref(), THREAD)), + file_spend + ); + assert!(!thread_spend_in(store.as_ref(), "no-such-thread").found_transcript); +} + +#[test] +fn a_subagent_written_through_a_driver_keeps_its_usage() { + let store = driver(); + write_scenario(store.as_ref()); + let child = store + .thread_transcripts(THREAD, None) + .unwrap() + .subagents + .remove(0); + let session = child.transcript.read_session().unwrap().unwrap(); + let last = session.messages.last().unwrap(); + assert_eq!(last.role, "assistant"); + let recorded = last + .turn_usage + .as_ref() + .expect("usage on the last assistant row"); + assert_eq!((recorded.usage.input, recorded.usage.output), (3, 2)); +} + +async fn background_outcomes(locator: &dyn TranscriptLocator) -> Vec { + let message = || TranscriptMessage::assistant("reminder delivered"); + let options = + |key: &str| BackgroundAppend::new(key, serde_json::json!({"kind": "cron", "run_id": key})); + let mut outcomes = Vec::new(); + outcomes.push( + append_background_message_in(locator, &root(), message(), options("run-1")) + .await + .unwrap(), + ); + write_scenario(locator); + for key in ["run-1", "run-1"] { + outcomes.push( + append_background_message_in(locator, &root(), message(), options(key)) + .await + .unwrap(), + ); + } + let mut pinned = options("run-2"); + pinned.expected_generation = Some(0); + outcomes.push( + append_background_message_in(locator, &root(), message(), pinned) + .await + .unwrap(), + ); + outcomes +} + +#[tokio::test(flavor = "multi_thread")] +async fn background_append_matches_across_stores() { + let dir = tempfile::tempdir().unwrap(); + let files = FileTranscriptLocator::new(dir.path()); + let store = driver(); + let expected = vec![ + BackgroundAppendOutcome::NoSession, + BackgroundAppendOutcome::Appended { generation: 1 }, + BackgroundAppendOutcome::Duplicate { generation: 1 }, + BackgroundAppendOutcome::StaleGeneration { + expected: 0, + head: 1, + }, + ]; + assert_eq!(background_outcomes(&files).await, expected); + assert_eq!(background_outcomes(store.as_ref()).await, expected); + + for locator in [&files as &dyn TranscriptLocator, store.as_ref()] { + let head = locator + .read_session_transcript(&root().next_generation()) + .unwrap(); + let display = head.read_display().unwrap().unwrap(); + let Some(DisplayRecord::Message(last)) = display.records.last() else { + panic!("the delivery is the last record"); + }; + assert_eq!(last.message.content, "reminder delivered"); + let origin = last.background.as_ref().expect("provenance recorded"); + assert_eq!(origin.idempotency_key, "run-1"); + assert_eq!(origin.provenance["kind"], "cron"); + assert!( + last.message.extra_metadata.is_none(), + "the provenance is lifted out of the row's metadata" + ); + // The model context replays the delivery as an ordinary row. + let session = head.read_session().unwrap().unwrap(); + assert_eq!( + session.messages.last().unwrap().content, + "reminder delivered" + ); + } +} diff --git a/crates/tinyagents-session/src/transcript/spend.rs b/crates/tinyagents-session/src/transcript/spend.rs index ee2fb1fd6..4204e6d97 100644 --- a/crates/tinyagents-session/src/transcript/spend.rs +++ b/crates/tinyagents-session/src/transcript/spend.rs @@ -9,8 +9,9 @@ use std::collections::{BTreeMap, HashSet}; use std::path::{Path, PathBuf}; use super::{ - DisplayRecord, SessionTranscript, TranscriptMessage, TranscriptMeta, TurnUsage, - UsageCostSource, find_root_transcripts_for_thread, read_transcript_display, + DisplayRecord, DisplaySessionTranscript, SessionTranscript, ThreadTranscript, + TranscriptLocator, TranscriptMessage, TranscriptMeta, TurnUsage, UsageCostSource, + find_root_transcripts_for_thread, read_transcript_display, }; /// One transcript's own spend, summed from the per-turn `turn_usage` records @@ -251,6 +252,14 @@ fn file_spend( return None; } }; + Some(display_spend(display, earlier_requests)) +} + +/// [`file_spend`] of an already-read display projection. +fn display_spend( + display: DisplaySessionTranscript, + earlier_requests: &mut HashSet, +) -> (TranscriptSpend, TranscriptMeta) { let mut usages: Vec<&TurnUsage> = Vec::new(); let mut counted_requests: HashSet<&str> = HashSet::new(); // The logical context as of the previous record: what a compaction's kept @@ -316,7 +325,7 @@ fn file_spend( } let spend = spend_from_usages(usages, display.meta.model.as_ref()); earlier_requests.extend(counted_requests.into_iter().map(str::to_owned)); - Some((spend, display.meta)) + (spend, display.meta) } /// Whether `row` in a compaction replacement is `prior` carried over: same @@ -393,57 +402,165 @@ pub fn thread_spend(workspace_dir: &Path, thread_id: &str) -> ThreadSpend { for root in &roots { out.found_transcript = true; if let Some((spend, meta)) = file_spend(root, &mut root_requests) { - out.root.input_tokens = out.root.input_tokens.saturating_add(spend.input_tokens); - out.root.output_tokens = out.root.output_tokens.saturating_add(spend.output_tokens); - out.root.cached_input_tokens = out - .root - .cached_input_tokens - .saturating_add(spend.cached_input_tokens); - out.root.cost_usd += spend.cost_usd; - out.root.cost_split.merge(&spend.cost_split); - out.root.turns += spend.turns; - // `find_root_transcripts_for_thread` returns oldest first, so the - // last root to report a turn owns the last-turn view. - if spend.turns > 0 { - out.root.last_input_tokens = spend.last_input_tokens; - out.root.last_output_tokens = spend.last_output_tokens; - out.root.last_context_tokens = spend.last_context_tokens; - out.root.last_context_input_tokens = spend.last_context_input_tokens; - out.root.last_context_output_tokens = spend.last_context_output_tokens; - } - if spend.context_window > 0 { - out.root.context_window = spend.context_window; + out.add_root(spend, meta); + } + let children = descendant_transcripts(root) + .into_iter() + .map(|path| (stem_of(&path), path)); + for chain in subagent_chains(children) { + // One run per chain: a compacted sub-agent's `.gN` successors open + // with rows carried from the earlier generation, which keep their + // `request_id` and must not be counted again. + let mut requests = HashSet::new(); + let mut new_run = true; + for child in chain { + if let Some((spend, child_meta)) = file_spend(&child, &mut requests) { + out.add_subagent(spend, &child_meta, std::mem::take(&mut new_run)); + } } - if spend.model.is_some() { - out.root.model = spend.model; + } + } + out +} + +/// [`thread_spend`] over any [`TranscriptLocator`]: the thread's roots and +/// sub-agent transcripts come from +/// [`TranscriptLocator::thread_transcripts`], each counted once — the roots +/// in projection order with one shared set of counted requests, every listed +/// sub-agent into its archetype. Unlike the file walk, a sub-agent the +/// listing finds by its `_meta.thread_id` (not only by stem) is counted too. +pub fn thread_spend_in(locator: &dyn TranscriptLocator, thread_id: &str) -> ThreadSpend { + let mut out = ThreadSpend::default(); + let Some(listed) = locator.thread_transcripts(thread_id, None) else { + return out; + }; + let mut root_requests = HashSet::new(); + for root in &listed.roots { + out.found_transcript = true; + if let Some(display) = read_listed(root) { + let (spend, meta) = display_spend(display, &mut root_requests); + out.add_root(spend, meta); + } + } + let children = listed + .subagents + .iter() + .map(|child| (child.stem.clone(), child)); + for chain in subagent_chains(children) { + // As in the file walk: one run and one set of counted requests per + // compaction chain. + let mut requests = HashSet::new(); + let mut new_run = true; + for child in chain { + if let Some(display) = read_listed(child) { + let (spend, meta) = display_spend(display, &mut requests); + out.add_subagent(spend, &meta, std::mem::take(&mut new_run)); } - out.updated = Some(meta.updated); } - for child in descendant_transcripts(root) { - let Some((spend, child_meta)) = file_spend(&child, &mut HashSet::new()) else { - continue; - }; - out.found_transcript = true; - let entry = out - .subagents - .entry(child_meta.agent_name.clone()) - .or_default(); - entry.0.input_tokens = entry.0.input_tokens.saturating_add(spend.input_tokens); - entry.0.output_tokens = entry.0.output_tokens.saturating_add(spend.output_tokens); - entry.0.cached_input_tokens = entry - .0 - .cached_input_tokens - .saturating_add(spend.cached_input_tokens); - entry.0.cost_usd += spend.cost_usd; - entry.0.cost_split.merge(&spend.cost_split); - entry.0.turns += spend.turns; - if entry.0.model.is_none() { - entry.0.model = spend.model; + } + out +} + +/// Groups sub-agent transcripts into compaction chains: a compaction seals +/// `{stem}` and opens `{stem}.g1`, `{stem}.g2`, … Chains come out ordered by +/// their base stem, each one's generations oldest first. +fn subagent_chains(items: impl IntoIterator) -> Vec> { + let mut chains: BTreeMap> = BTreeMap::new(); + for (stem, item) in items { + let (base, generation) = match stem.rsplit_once(".g") { + Some((base, digits)) + if !digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit()) => + { + (base.to_string(), digits.parse().unwrap_or(u32::MAX)) } + _ => (stem.clone(), 0), + }; + chains + .entry(base) + .or_default() + .push((generation, stem, item)); + } + chains + .into_values() + .map(|mut chain| { + chain.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1))); + chain.into_iter().map(|(_, _, item)| item).collect() + }) + .collect() +} + +fn stem_of(path: &Path) -> String { + path.file_stem() + .map(|stem| stem.to_string_lossy().into_owned()) + .unwrap_or_default() +} + +/// A listed transcript's display projection; unreadable or missing ones are +/// skipped with a warning, as the file walk skips an unreadable file. +fn read_listed(listed: &ThreadTranscript) -> Option { + match listed.transcript.read_display() { + Ok(found) => found, + Err(err) => { + tracing::warn!( + "[transcript:spend] skipping unreadable transcript {}: {err}", + listed.stem + ); + None + } + } +} + +impl ThreadSpend { + /// Adds one root transcript's own spend; roots arrive oldest first, so + /// the last one to report a turn owns the last-turn view. + fn add_root(&mut self, spend: TranscriptSpend, meta: TranscriptMeta) { + self.root.input_tokens = self.root.input_tokens.saturating_add(spend.input_tokens); + self.root.output_tokens = self.root.output_tokens.saturating_add(spend.output_tokens); + self.root.cached_input_tokens = self + .root + .cached_input_tokens + .saturating_add(spend.cached_input_tokens); + self.root.cost_usd += spend.cost_usd; + self.root.cost_split.merge(&spend.cost_split); + self.root.turns += spend.turns; + if spend.turns > 0 { + self.root.last_input_tokens = spend.last_input_tokens; + self.root.last_output_tokens = spend.last_output_tokens; + self.root.last_context_tokens = spend.last_context_tokens; + self.root.last_context_input_tokens = spend.last_context_input_tokens; + self.root.last_context_output_tokens = spend.last_context_output_tokens; + } + if spend.context_window > 0 { + self.root.context_window = spend.context_window; + } + if spend.model.is_some() { + self.root.model = spend.model; + } + self.updated = Some(meta.updated); + } + + /// Adds one sub-agent transcript's own spend to its archetype + /// (`_meta.agent`); `new_run` is false for a later generation of a run + /// already counted. + fn add_subagent(&mut self, spend: TranscriptSpend, meta: &TranscriptMeta, new_run: bool) { + self.found_transcript = true; + let entry = self.subagents.entry(meta.agent_name.clone()).or_default(); + entry.0.input_tokens = entry.0.input_tokens.saturating_add(spend.input_tokens); + entry.0.output_tokens = entry.0.output_tokens.saturating_add(spend.output_tokens); + entry.0.cached_input_tokens = entry + .0 + .cached_input_tokens + .saturating_add(spend.cached_input_tokens); + entry.0.cost_usd += spend.cost_usd; + entry.0.cost_split.merge(&spend.cost_split); + entry.0.turns += spend.turns; + if entry.0.model.is_none() { + entry.0.model = spend.model; + } + if new_run { entry.1 += 1; } } - out } #[cfg(test)] diff --git a/crates/tinyagents-session/src/transcript/spend_tests.rs b/crates/tinyagents-session/src/transcript/spend_tests.rs index 225cbdf03..db69a5cc3 100644 --- a/crates/tinyagents-session/src/transcript/spend_tests.rs +++ b/crates/tinyagents-session/src/transcript/spend_tests.rs @@ -698,3 +698,65 @@ fn last_call_fields_round_trip_and_default_on_old_lines() { assert_eq!(old.last_call_input, 0); assert_eq!(old.last_call_output, 0); } + +/// A compacted sub-agent leaves `{stem}` and `{stem}.g1` behind; the successor +/// opens with the rows it carried over (same `request_id`). The pair is one +/// run, and the carried turn is counted once, by the file walk and the +/// locator walk alike. +#[test] +fn a_compacted_subagent_is_one_run_counted_once() { + use crate::transcript::FileTranscriptLocator; + let tmp = tempfile::tempdir().expect("tempdir"); + let thread = "thread-subagent-generations"; + let root_stem = "1790000002_orchestrator_generations"; + write_transcript_turns( + tmp.path(), + root_stem, + &meta("orchestrator", "root", Some(thread)), + &[(1_000, 10, 0)], + ); + let child = format!("{root_stem}__1790000003_researcher_sub"); + write_transcript_turns( + tmp.path(), + &child, + &meta("researcher", "subagent", Some(thread)), + &[(5_000, 50, 0)], + ); + // The successor re-writes `req-0` (carried) and adds `req-1`. + write_transcript_turns( + tmp.path(), + &format!("{child}.g1"), + &meta("researcher", "subagent", Some(thread)), + &[(5_000, 50, 0), (7_000, 70, 0)], + ); + + for (walk, spend) in [ + ("files", thread_spend(tmp.path(), thread)), + ( + "locator", + thread_spend_in(&FileTranscriptLocator::new(tmp.path()), thread), + ), + ] { + let (researcher, runs) = spend.subagents.get("researcher").expect("child counted"); + assert_eq!(*runs, 1, "{walk}: one delegation across its generations"); + assert_eq!( + researcher.input_tokens, 12_000, + "{walk}: carried turn counted once" + ); + assert_eq!(researcher.output_tokens, 120, "{walk}"); + assert_eq!(researcher.turns, 2, "{walk}"); + } +} + +#[test] +fn subagent_chains_order_generations_numerically() { + let chains = subagent_chains( + ["a.g10", "b", "a.g2", "a", "a.gx"] + .into_iter() + .map(|stem| (stem.to_string(), stem)), + ); + assert_eq!( + chains, + vec![vec!["a", "a.g2", "a.g10"], vec!["a.gx"], vec!["b"]] + ); +} diff --git a/crates/tinyagents-session/src/transcript/turn_lock.rs b/crates/tinyagents-session/src/transcript/turn_lock.rs index fc6c41213..8f4c836ec 100644 --- a/crates/tinyagents-session/src/transcript/turn_lock.rs +++ b/crates/tinyagents-session/src/transcript/turn_lock.rs @@ -54,6 +54,26 @@ pub async fn lock_session_turn( Some(SessionTurnGuard { _guard: guard }) } +/// [`lock_session_turn`], but never `None`: a locator that names no +/// destination is keyed by its own address instead, so callers sharing that +/// one locator instance still serialise. (Another instance over the same store +/// is not found; nothing could tell the two apart.) +pub(crate) async fn lock_session_turn_or_identity( + locator: &dyn TranscriptLocator, + session: &SessionRef, +) -> SessionTurnGuard { + let destination = locator.destination_key().unwrap_or_else(|| { + let address = std::ptr::from_ref(locator).cast::<()>() as usize; + format!("anonymous-locator:{address:#x}") + }); + let stem = session_stem(&session.first_generation()); + let lock = registry_lock(destination, stem.clone()); + tracing::debug!(session = %stem, "[transcript-turn-lock] acquiring"); + let guard = lock.lock_owned().await; + tracing::debug!(session = %stem, "[transcript-turn-lock] acquired"); + SessionTurnGuard { _guard: guard } +} + #[doc(hidden)] pub(crate) async fn lock_session_turn_with_notification( locator: &dyn TranscriptLocator, diff --git a/crates/tinyagents-session/src/transcript/types.rs b/crates/tinyagents-session/src/transcript/types.rs index 421d17f04..1f8d16304 100644 --- a/crates/tinyagents-session/src/transcript/types.rs +++ b/crates/tinyagents-session/src/transcript/types.rs @@ -896,3 +896,34 @@ pub enum BackgroundAppendOutcome { /// transcript was created). NoSession, } + +/// One transcript of a thread, as a locator lists it. +#[derive(Clone)] +pub struct ThreadTranscript { + /// The transcript's stem (its session key; a sub-agent's chains its + /// parent's with `__`). + pub stem: String, + /// A read handle on it. + pub transcript: std::sync::Arc, +} + +impl std::fmt::Debug for ThreadTranscript { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ThreadTranscript") + .field("stem", &self.stem) + .finish_non_exhaustive() + } +} + +/// Every transcript backing one thread's view, as +/// [`TranscriptLocator::thread_transcripts`](super::TranscriptLocator::thread_transcripts) +/// lists them. +#[derive(Debug, Clone, Default)] +pub struct ThreadTranscripts { + /// The thread's root transcripts in projection order: each session's + /// generations oldest first, pre-identity roots dropped once a session + /// root exists. + pub roots: Vec, + /// Its sub-agent transcripts, sorted by stem. + pub subagents: Vec, +} diff --git a/crates/tinyagents-session/src/transcript/view/mod.rs b/crates/tinyagents-session/src/transcript/view/mod.rs index f49762f2c..b75a385fb 100644 --- a/crates/tinyagents-session/src/transcript/view/mod.rs +++ b/crates/tinyagents-session/src/transcript/view/mod.rs @@ -18,8 +18,10 @@ use serde::Serialize; pub use project::{ project_records, project_thread, project_thread_scoped, project_thread_scoped_with, - resolve_files_scoped, + project_transcripts, resolve_files_scoped, }; + +use crate::transcript::TranscriptLocator; #[allow(deprecated)] pub use types::SubagentStatus; pub use types::{DisplayItem, ProjectedTranscript, ToolCallStatus, TranscriptSubagentStatus}; @@ -101,15 +103,44 @@ pub fn get_page_scoped( cursor: Option<&str>, limit: Option, ) -> TranscriptPage { - let limit = limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT); - let upper_cursor = parse_cursor(cursor); - let projected = if let Some(agent_id) = agent_id { project::project_thread_scoped(workspace_dir, thread_id, Some(agent_id)) .map(std::sync::Arc::new) } else { cache::global().get_or_project(workspace_dir, thread_id) }; + page_of(thread_id, projected.as_deref(), cursor, limit) +} + +/// [`get_page_scoped`] over any [`TranscriptLocator`]: the thread's +/// transcripts come from +/// [`TranscriptLocator::thread_transcripts`] and are projected exactly as the +/// file view projects them. Not cached. +pub fn get_page_in( + locator: &dyn TranscriptLocator, + thread_id: &str, + agent_id: Option<&str>, + cursor: Option<&str>, + limit: Option, +) -> TranscriptPage { + let projected = locator + .thread_transcripts(thread_id, agent_id) + .map(|listed| { + project::project_transcripts(thread_id, &listed.roots, &listed.subagents, None) + }); + page_of(thread_id, projected.as_ref(), cursor, limit) +} + +/// One newest-first page of `projected` (an empty, transcript-less page when +/// the thread has none). +fn page_of( + thread_id: &str, + projected: Option<&ProjectedTranscript>, + cursor: Option<&str>, + limit: Option, +) -> TranscriptPage { + let limit = limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT); + let upper_cursor = parse_cursor(cursor); let Some(projected) = projected else { tracing::debug!("{LOG_PREFIX} get_page thread={thread_id}: no transcript"); return TranscriptPage { diff --git a/crates/tinyagents-session/src/transcript/view/project.rs b/crates/tinyagents-session/src/transcript/view/project.rs index 7cf7ac9c0..5b3fe3f18 100644 --- a/crates/tinyagents-session/src/transcript/view/project.rs +++ b/crates/tinyagents-session/src/transcript/view/project.rs @@ -8,8 +8,13 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::path::{Path, PathBuf}; +use std::sync::Arc; -use crate::transcript::{self, CompactionMarker, DisplayMessage, DisplayRecord}; +use crate::transcript::history::seed_meta_for_discovered; +use crate::transcript::{ + CompactionMarker, DisplayMessage, DisplayRecord, DisplaySessionTranscript, + FileTranscriptHistory, ThreadTranscript, TranscriptRead, +}; use super::{SpawnCallLookup, TOOL_RESULT_FAILURES_METADATA_KEY}; @@ -102,31 +107,65 @@ pub fn project_from_files( root_paths: &[PathBuf], sub_paths: &[PathBuf], lookup: Option<&dyn SpawnCallLookup>, +) -> ProjectedTranscript { + let handles = |paths: &[PathBuf]| -> Vec { + paths + .iter() + .map(|path| { + // The same stem rule as `FileTranscriptLocator::thread_transcripts` + // (strips only `.jsonl`), so both paths name a file alike. + let stem = crate::transcript::history::stem_of_handle(path); + let transcript: Arc = Arc::new( + FileTranscriptHistory::opened_at(path.clone(), seed_meta_for_discovered(&stem)), + ); + ThreadTranscript { stem, transcript } + }) + .collect() + }; + project_transcripts(thread_id, &handles(root_paths), &handles(sub_paths), lookup) +} + +/// `project_from_files` over a listed transcript set (see +/// [`TranscriptLocator::thread_transcripts`](crate::transcript::TranscriptLocator::thread_transcripts)), +/// so any store projects the same way. Unreadable transcripts degrade to +/// empty rather than failing. +pub fn project_transcripts( + thread_id: &str, + roots: &[ThreadTranscript], + subagents: &[ThreadTranscript], + lookup: Option<&dyn SpawnCallLookup>, ) -> ProjectedTranscript { tracing::debug!( "{LOG_PREFIX} projecting thread={thread_id} roots={} subagent_files={}", - root_paths.len(), - sub_paths.len() + roots.len(), + subagents.len() ); + let read = |root: &ThreadTranscript| match root.transcript.read_display() { + Ok(Some(display)) => Some(display), + Ok(None) => { + tracing::warn!("{LOG_PREFIX} root transcript {} is missing", root.stem); + None + } + Err(err) => { + tracing::warn!( + "{LOG_PREFIX} failed to read root transcript {}: {err}", + root.stem + ); + None + } + }; + let displays: Vec<(String, DisplaySessionTranscript)> = roots + .iter() + .filter_map(|root| read(root).map(|display| (root.stem.clone(), display))) + .collect(); + let mut records: Vec = Vec::new(); - let root_thread_id = root_paths.iter().find_map(|path| { - transcript::read_transcript_display(path) - .ok() - .and_then(|display| display.meta.thread_id) - }); + let root_thread_id = displays + .iter() + .find_map(|(_, display)| display.meta.thread_id.clone()); let mut previous: Option<(Option, Vec)> = None; - for root_path in root_paths { - let display = match transcript::read_transcript_display(root_path) { - Ok(display) => display, - Err(err) => { - tracing::warn!( - "{LOG_PREFIX} failed to read root transcript {}: {err}", - root_path.display() - ); - continue; - } - }; + for (root_stem, display) in displays { let successor_of_previous = matches!( (&previous, display.meta.parent_session_id.as_deref()), (Some((Some(prev_id), _)), Some(parent)) if prev_id == parent @@ -138,8 +177,7 @@ pub fn project_from_files( .unwrap_or_default(); let (kept, retained) = resolve::drop_retained_rows(&display.records, predecessor); tracing::debug!( - "{LOG_PREFIX} generation {} retained={} new_records={}", - root_path.display(), + "{LOG_PREFIX} generation {root_stem} retained={} new_records={}", retained.len(), kept.len() ); @@ -167,7 +205,7 @@ pub fn project_from_files( let top_level = items.len(); subagents::attach( &mut items, - sub_paths, + subagents, &subagents::turn_segments(&records), root_thread_id.as_deref(), lookup, diff --git a/crates/tinyagents-session/src/transcript/view/subagents.rs b/crates/tinyagents-session/src/transcript/view/subagents.rs index 2695006e5..91a2939e2 100644 --- a/crates/tinyagents-session/src/transcript/view/subagents.rs +++ b/crates/tinyagents-session/src/transcript/view/subagents.rs @@ -25,11 +25,9 @@ //! there are no turns) instead of after every root item, which is where all //! sub-agents used to go. -use std::path::{Path, PathBuf}; - use super::SpawnCallLookup; -use crate::transcript::{self, DisplayRecord}; +use crate::transcript::{DisplayRecord, ThreadTranscript}; use super::project::{native_tool_round, project_records}; use super::types::{DisplayItem, ToolCallStatus, TranscriptSubagentStatus}; @@ -80,19 +78,19 @@ enum OwnState { /// `segments` are the root turns' `(request_id, commit unix)` pairs. pub(super) fn attach( items: &mut Vec, - sub_paths: &[PathBuf], + subs: &[ThreadTranscript], segments: &[(String, i64)], root_thread_id: Option<&str>, lookup: Option<&dyn SpawnCallLookup>, ) { - let children = build_children(sub_paths, None, root_thread_id, 0, lookup); + let children = build_children(subs, None, root_thread_id, 0, lookup); place(items, children, segments, lookup); } /// Project the direct children of `parent_stem` (or of the roots, when /// `None`), recursing into their own children. fn build_children( - sub_paths: &[PathBuf], + subs: &[ThreadTranscript], parent_stem: Option<&str>, parent_thread_id: Option<&str>, depth: usize, @@ -102,19 +100,16 @@ fn build_children( return Vec::new(); } let mut children = Vec::new(); - for path in sub_paths { - let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else { - continue; - }; + for sub in subs { + let stem = sub.stem.as_str(); // A compacted child transcript is a generation chain. The newest // generation contains the retained prefix plus its subsequent rows; // project that head once instead of rendering every generation as a // separate child card. if let Some((base, generation)) = child_generation(stem) - && sub_paths + && subs .iter() - .filter_map(|candidate| candidate.file_stem().and_then(|name| name.to_str())) - .filter_map(|candidate| child_generation(candidate)) + .filter_map(|candidate| child_generation(&candidate.stem)) .any(|(candidate_base, candidate_generation)| { candidate_base == base && candidate_generation > generation }) @@ -137,11 +132,10 @@ fn build_children( .then(|| stem.split_once("__").map(|(parent, _)| parent)) .flatten(); if let Some(child) = build_child( - path, - stem, + sub, suffix, parent_thread_id.or(stem_parent_thread_id), - sub_paths, + subs, depth, lookup, ) { @@ -158,28 +152,29 @@ fn child_generation(stem: &str) -> Option<(&str, u32)> { } fn build_child( - path: &Path, - stem: &str, + sub: &ThreadTranscript, suffix: &str, parent_thread_id: Option<&str>, - sub_paths: &[PathBuf], + subs: &[ThreadTranscript], depth: usize, lookup: Option<&dyn SpawnCallLookup>, ) -> Option { - let display = match transcript::read_transcript_display(path) { - Ok(display) => display, + let stem = sub.stem.as_str(); + let display = match sub.transcript.read_display() { + Ok(Some(display)) => display, + Ok(None) => { + tracing::warn!("{LOG_PREFIX} sub-agent transcript {stem} is missing"); + return None; + } Err(err) => { - tracing::warn!( - "{LOG_PREFIX} failed to read sub-agent transcript {}: {err}", - path.display() - ); + tracing::warn!("{LOG_PREFIX} failed to read sub-agent transcript {stem}: {err}"); return None; } }; let own_state = own_state(&display.records); let mut items = project_records(&display.records); let grandchildren = build_children( - sub_paths, + subs, Some(stem), display.meta.thread_id.as_deref(), depth + 1, diff --git a/crates/tinyagents-session/src/transcript/writer.rs b/crates/tinyagents-session/src/transcript/writer.rs index 621a0d603..4d9c9eb60 100644 --- a/crates/tinyagents-session/src/transcript/writer.rs +++ b/crates/tinyagents-session/src/transcript/writer.rs @@ -4,7 +4,7 @@ use super::history::TranscriptPartial; use super::jsonl::{ - COMPACTION_KIND, CompactionLine, MessageLine, build_message_line, meta_line_json, + COMPACTION_KIND, CompactionLine, MessageLine, build_message_line, meta_line_json, partial_line, serialise_message_lines, tools_line_json, }; use super::markdown::render_markdown; @@ -29,6 +29,19 @@ pub fn write_transcript( messages: &[TranscriptMessage], meta: &TranscriptMeta, last_assistant_turn_usage: Option<&TurnUsage>, +) -> Result<()> { + write_transcript_with_tools(jsonl_path, messages, meta, last_assistant_turn_usage, None) +} + +/// [`write_transcript`], additionally recording `tools` as the transcript's +/// latest `{"kind":"tools"}` record, so a full rewrite of a transcript that +/// recorded its tool declarations keeps them. +pub(crate) fn write_transcript_with_tools( + jsonl_path: &Path, + messages: &[TranscriptMessage], + meta: &TranscriptMeta, + last_assistant_turn_usage: Option<&TurnUsage>, + tools: Option<&serde_json::Value>, ) -> Result<()> { if let Some(parent) = jsonl_path.parent() { fs::create_dir_all(parent) @@ -40,6 +53,10 @@ pub fn write_transcript( jsonl_buf.push_str(&meta_line_json(meta)?); jsonl_buf.push('\n'); serialise_message_lines(messages, last_assistant_turn_usage, None, &mut jsonl_buf)?; + if let Some(tools) = tools { + jsonl_buf.push_str(&tools_line_json(tools)?); + jsonl_buf.push('\n'); + } atomic_write(jsonl_path, jsonl_buf.as_bytes()) .with_context(|| format!("write transcript {}", jsonl_path.display()))?; @@ -331,19 +348,7 @@ fn serialise_interrupted_partial( let Some(partial) = partial.filter(|partial| !partial.content.is_empty()) else { return Ok(()); }; - let mut line = build_message_line( - &TranscriptMessage::assistant(&partial.content), - None, - request_id, - true, - ); - line.iteration = partial.iteration; - line.reasoning_content = partial - .reasoning_content - .as_deref() - .map(str::trim) - .filter(|content| !content.is_empty()) - .map(str::to_owned); + let mut line = partial_line(partial, request_id); line.ts = Some(chrono::Utc::now().to_rfc3339()); buf.push_str(&serde_json::to_string(&line).context("serialise interrupted partial line")?); buf.push('\n');