feat(session): locator-based thread view, spend, background append and legacy rewrite

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<ThreadTranscripts>` 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

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)]

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
+/// `<workspace_dir>/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<PathBuf> {
+    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<std::fs::DirEntry>,
+    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;