diff --git a/CHANGELOG.md b/CHANGELOG.md index 22ac50711..be4b95f39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # Changelog +## Unreleased: workspace transcripts on driver ports (tinyagents#453, transcript half) + +- `port::WorkspaceStores::open(workspace, agent_id)` (feature `legacy-import`) + replaces `FileTranscriptLocator::new(workspace)`: transcripts are served by + `DriverSessionStores` over a SQLite backend kept at `/session_store/` + (an internal driver location), and the caller holds a `TranscriptLocator` + port. Handles are cached per workspace and agent. +- `legacy::import_transcripts_once` imports the classic `session_raw/*.jsonl` + files a single time. A marker document in `session_import_markers` records + completion (not written while a file is unreadable, so it is retried); legacy + files are only read, never modified, renamed or deleted. +- Not yet moved: `FileTranscriptLocator`/`FileTranscriptHistory` and the + path-taking free functions (`write_transcript`, `append_transcript_turn`, + `read_transcript`, `spend`, `view`, `adopt_legacy_session_transcripts`) + remain until hosts switch to `WorkspaceStores`. + ## Unreleased: tinystoragedrivers by git tag The workspace takes `tinystoragedrivers-core` / `-sqlite` by git tag `v0.11.0` diff --git a/crates/tinyagents-session/src/legacy/mod.rs b/crates/tinyagents-session/src/legacy/mod.rs index 93cd8d4b9..70b603851 100644 --- a/crates/tinyagents-session/src/legacy/mod.rs +++ b/crates/tinyagents-session/src/legacy/mod.rs @@ -31,13 +31,17 @@ pub(crate) mod rows; pub mod source; pub mod transcripts; +mod transcripts_once; pub mod turn_states; +mod types; mod workspace; pub use crate::run_ledger::drivers::{ImportCounts, LegacyDatabaseReport}; 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 workspace::{LegacyWorkspaceReport, WorkspaceLayout, import_workspace}; #[cfg(test)] @@ -48,3 +52,6 @@ pub(crate) mod fixtures; #[cfg(test)] #[path = "mod_tests.rs"] mod tests; +#[cfg(test)] +#[path = "transcripts_once_tests.rs"] +mod transcripts_once_tests; diff --git a/crates/tinyagents-session/src/legacy/transcripts_once.rs b/crates/tinyagents-session/src/legacy/transcripts_once.rs new file mode 100644 index 000000000..284ae6262 --- /dev/null +++ b/crates/tinyagents-session/src/legacy/transcripts_once.rs @@ -0,0 +1,107 @@ +//! The one-time form of the classic JSONL transcript import. +//! +//! [`import_transcripts`] is already read-only, id-preserving and +//! only-if-absent, so running it again is harmless but rescans and re-parses +//! every file in `session_raw/`. A host that imports on every launch records a +//! marker document beside the session collections instead: once a pass has +//! read every file, later launches skip the scan and never touch the legacy +//! directory again. The legacy files themselves are never modified, renamed or +//! deleted. + +use std::path::Path; + +use serde_json::{Value, json}; +use tinystoragedrivers_core::{CollectionSpec, DocumentStore, ErrorKind, Precondition}; + +use super::transcripts::{TranscriptImportReport, import_transcripts}; +use super::types::OnceOutcome; +use crate::port::DriverSessionStores; + +/// Collection holding the import markers. +pub const MARKER_COLLECTION: &str = "session_import_markers"; +/// Id of the transcript-import marker document. +pub const TRANSCRIPT_MARKER_ID: &str = "session_raw_jsonl_v1"; + +const LOG_PREFIX: &str = "[session_db:import:transcripts:once]"; + +/// Imports `/session_raw/` into `agent_id`'s transcripts once. +/// +/// A marker document in the agent's scope records completion. A provider +/// without schema authority (`ValidateOnly`) cannot declare the marker +/// collection, so it runs the import every time (still idempotent). +/// +/// # Errors +/// +/// The destination refuses a read or write, or the raw directory cannot be +/// listed. No marker is written then, so a later call resumes the work. +pub fn import_transcripts_once( + provider: &DriverSessionStores, + agent_id: &str, + workspace_dir: &Path, +) -> anyhow::Result { + let stores = provider.try_for_agent(agent_id)?; + let (docs, may_declare) = provider.documents_for(agent_id)?; + if may_declare { + let docs = docs.clone(); + let marked = provider + .bridge() + .run(async move { marker_present(docs.as_ref()).await })??; + if marked { + tracing::debug!("{LOG_PREFIX} marker present; skipping scan agent_id={agent_id}"); + return Ok(OnceOutcome::AlreadyImported); + } + } + let report = import_transcripts(workspace_dir, stores.transcripts.as_ref())?; + let mut marked = false; + if may_declare && report.unreadable == 0 { + let docs = docs.clone(); + provider + .bridge() + .run(async move { write_marker(docs.as_ref(), &report).await })??; + marked = true; + } + tracing::info!( + "{LOG_PREFIX} agent_id={agent_id} imported={} already_present={} unreadable={} marked={marked}", + report.imported, + report.already_present, + report.unreadable + ); + Ok(OnceOutcome::Imported { report, marked }) +} + +async fn marker_present( + docs: &dyn DocumentStore, +) -> Result { + docs.ensure_collection(&CollectionSpec::new(MARKER_COLLECTION)) + .await?; + Ok(docs + .get(MARKER_COLLECTION, TRANSCRIPT_MARKER_ID) + .await? + .is_some()) +} + +async fn write_marker( + docs: &dyn DocumentStore, + report: &TranscriptImportReport, +) -> Result<(), tinystoragedrivers_core::StorageError> { + let doc: Value = json!({ + "imported_at": chrono::Utc::now().to_rfc3339(), + "read": report.read, + "imported": report.imported, + "already_present": report.already_present, + "messages": report.messages, + }); + match docs + .put( + MARKER_COLLECTION, + TRANSCRIPT_MARKER_ID, + doc, + Precondition::Absent, + ) + .await + { + // A concurrent launch marked it first; the marker is idempotent. + Err(error) if error.kind() == ErrorKind::Conflict => Ok(()), + other => other.map(|_| ()), + } +} diff --git a/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs b/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs new file mode 100644 index 000000000..e55bc6831 --- /dev/null +++ b/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs @@ -0,0 +1,113 @@ +use super::*; + +// The one-time transcript import: marker, idempotence, untouched sources. +use crate::port::DriverSessionStores; +use crate::transcript::{TranscriptMessage, TranscriptMeta, write_transcript}; +use std::path::Path; +use std::sync::Arc; +use tinystoragedrivers_core::{MemoryStorage, StorageBackend}; + +fn meta() -> TranscriptMeta { + TranscriptMeta { + session_id: None, + parent_session_id: None, + agent_name: "agent".into(), + agent_id: Some("agent-id".into()), + agent_type: Some("root".into()), + 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: Some("thread-once".into()), + task_id: None, + } +} + +fn provider() -> DriverSessionStores { + DriverSessionStores::new(Arc::new(MemoryStorage::new()) as Arc).unwrap() +} + +fn write(workspace: &Path, stem: &str, text: &str) -> std::path::PathBuf { + let raw = workspace.join("session_raw"); + std::fs::create_dir_all(&raw).unwrap(); + let path = raw.join(format!("{stem}.jsonl")); + write_transcript( + &path, + &[ + TranscriptMessage::user(text), + TranscriptMessage::assistant("ok"), + ], + &meta(), + None, + ) + .unwrap(); + path +} + +#[test] +fn a_second_launch_skips_the_legacy_scan_and_never_touches_the_files() { + let dir = tempfile::tempdir().unwrap(); + let path = write(dir.path(), "1700000000_agent", "hello"); + let before = std::fs::read(&path).unwrap(); + let provider = provider(); + + let first = import_transcripts_once(&provider, "agent", dir.path()).unwrap(); + let OnceOutcome::Imported { report, marked } = first else { + panic!("first launch imports"); + }; + assert_eq!((report.imported, report.already_present), (1, 0)); + assert!(marked); + + // A file an old binary drops in after the marker is not picked up: the + // import is one-time. + write(dir.path(), "1700000001_agent", "late"); + let second = import_transcripts_once(&provider, "agent", dir.path()).unwrap(); + assert_eq!(second, OnceOutcome::AlreadyImported); + let stores = provider.try_for_agent("agent").unwrap(); + let late = stores + .transcripts + .open_stem("1700000001_agent", meta()) + .unwrap(); + assert!(late.read_session().unwrap().is_none()); + + assert_eq!(std::fs::read(&path).unwrap(), before, "source untouched"); +} + +#[test] +fn an_unreadable_file_leaves_the_import_unmarked_so_it_is_retried() { + let dir = tempfile::tempdir().unwrap(); + write(dir.path(), "good", "hi"); + std::fs::write(dir.path().join("session_raw").join("bad.jsonl"), "").unwrap(); + let provider = provider(); + + let first = import_transcripts_once(&provider, "agent", dir.path()).unwrap(); + let OnceOutcome::Imported { report, marked } = first else { + panic!("imports"); + }; + assert_eq!(report.unreadable, 1); + assert!(!marked); + + // Retry is idempotent: the good transcript is already there. + let again = import_transcripts_once(&provider, "agent", dir.path()).unwrap(); + let OnceOutcome::Imported { report, .. } = again else { + panic!("retries"); + }; + assert_eq!((report.imported, report.already_present), (0, 1)); +} + +#[test] +fn markers_are_per_agent_scope() { + let dir = tempfile::tempdir().unwrap(); + write(dir.path(), "s", "x"); + let provider = provider(); + import_transcripts_once(&provider, "a", dir.path()).unwrap(); + let other = import_transcripts_once(&provider, "b", dir.path()).unwrap(); + assert!(matches!(other, OnceOutcome::Imported { marked: true, .. })); +} diff --git a/crates/tinyagents-session/src/legacy/types.rs b/crates/tinyagents-session/src/legacy/types.rs new file mode 100644 index 000000000..154bfecf6 --- /dev/null +++ b/crates/tinyagents-session/src/legacy/types.rs @@ -0,0 +1,18 @@ +//! Result types of the legacy imports. + +use super::transcripts::TranscriptImportReport; + +/// How a one-time import ended. +#[derive(Debug, Clone, PartialEq)] +pub enum OnceOutcome { + /// A marker from an earlier launch says the import already ran. + AlreadyImported, + /// This pass ran. `marked` is false when some file did not parse, so the + /// next launch tries again instead of treating the import as complete. + Imported { + /// What the pass did. + report: TranscriptImportReport, + /// Whether the completion marker was written. + marked: bool, + }, +} diff --git a/crates/tinyagents-session/src/port/drivers/mod.rs b/crates/tinyagents-session/src/port/drivers/mod.rs index 550d00808..961e50b11 100644 --- a/crates/tinyagents-session/src/port/drivers/mod.rs +++ b/crates/tinyagents-session/src/port/drivers/mod.rs @@ -202,6 +202,27 @@ impl DriverSessionStores { }) } + /// The document store of `agent_id`'s scope, and whether this provider may + /// declare collections in it. Used by host-side one-time bookkeeping (the + /// legacy import marker) that sits beside the session collections. + #[cfg(feature = "legacy-import")] + pub(crate) fn documents_for( + &self, + agent_id: &str, + ) -> Result<(Arc, bool), StorageError> { + let scoped = self.backend.for_scope(&Self::scope_for(agent_id))?; + Ok(( + Arc::clone(scoped.documents()), + self.schema_admission == SchemaAdmission::Declare, + )) + } + + /// The bridge this provider runs synchronous seams on. + #[cfg(feature = "legacy-import")] + pub(crate) fn bridge(&self) -> &Blocking { + &self.bridge + } + /// The scope agent `agent_id` is stored under. pub fn scope_for(agent_id: &str) -> Scope { if !agent_id.starts_with(HASHED_SCOPE) diff --git a/crates/tinyagents-session/src/port/mod.rs b/crates/tinyagents-session/src/port/mod.rs index 9d50cdbd3..d2395ead2 100644 --- a/crates/tinyagents-session/src/port/mod.rs +++ b/crates/tinyagents-session/src/port/mod.rs @@ -50,6 +50,8 @@ mod drivers; mod memory; mod types; +#[cfg(feature = "legacy-import")] +mod workspace; use std::sync::Arc; @@ -58,6 +60,10 @@ pub use drivers::{ }; pub use memory::{InMemorySessionStores, InMemoryTranscriptLocator, InMemoryTurnStates}; pub use types::AgentStores; +#[cfg(feature = "legacy-import")] +pub use types::WorkspaceStores; +#[cfg(feature = "legacy-import")] +pub use workspace::STORE_DIR; pub use tinyagents_harness::store::{AppendStore, Store}; diff --git a/crates/tinyagents-session/src/port/types.rs b/crates/tinyagents-session/src/port/types.rs index f02bf6ea3..889b06c9d 100644 --- a/crates/tinyagents-session/src/port/types.rs +++ b/crates/tinyagents-session/src/port/types.rs @@ -32,3 +32,29 @@ impl std::fmt::Debug for AgentStores { .finish_non_exhaustive() } } + +/// An agent's driver-backed stores for one workspace. +/// +/// Cloning shares one open backend; the backend closes when the last clone +/// drops. +#[derive(Clone)] +#[cfg(feature = "legacy-import")] +pub struct WorkspaceStores { + pub(super) inner: Arc, +} + +#[cfg(feature = "legacy-import")] +pub(super) struct WorkspaceInner { + pub(super) provider: Arc, + pub(super) stores: AgentStores, + pub(super) outcome: crate::legacy::OnceOutcome, +} + +#[cfg(feature = "legacy-import")] +impl std::fmt::Debug for WorkspaceStores { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WorkspaceStores") + .field("outcome", &self.inner.outcome) + .finish_non_exhaustive() + } +} diff --git a/crates/tinyagents-session/src/port/workspace.rs b/crates/tinyagents-session/src/port/workspace.rs new file mode 100644 index 000000000..c460f1a7c --- /dev/null +++ b/crates/tinyagents-session/src/port/workspace.rs @@ -0,0 +1,125 @@ +//! Transcripts for a host that only has a workspace directory. +//! +//! [`WorkspaceStores::open`] replaces `FileTranscriptLocator::new(workspace)`: +//! it opens a SQLite storage backend at `/session_store/` (an +//! internal driver location, not part of the public surface), serves the +//! agent's transcripts through [`DriverSessionStores`], and imports the +//! workspace's classic `session_raw/*.jsonl` files once. Callers receive a +//! `TranscriptLocator` port; they never name a transcript file. +//! +//! Handles are cached per `(workspace, agent)` so a call site that used to +//! construct a file locator on every use does not reopen the database. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, OnceLock, PoisonError, Weak}; + +use tinystoragedrivers_core::StorageBackend; +use tinystoragedrivers_sqlite::SqliteStorage; + +use super::AgentStores; +use super::types::WorkspaceInner; +use super::{DriverSessionStores, WorkspaceStores}; +use crate::legacy::{OnceOutcome, import_transcripts_once}; +use crate::transcript::TranscriptLocator; + +/// Directory under the workspace holding the internal SQLite driver files. +pub const STORE_DIR: &str = "session_store"; + +type Cache = Mutex>>; + +/// Weak entries: a backend is released once every handle drops, while live +/// handles are still shared. +fn cache() -> &'static Cache { + static CACHE: OnceLock = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// Serialises first opens so two threads never run the legacy import for the +/// same workspace concurrently (one could see the other's half-written +/// transcript as complete). +static OPEN_LOCK: Mutex<()> = Mutex::new(()); + +fn cached(key: &(PathBuf, String)) -> Option { + let mut map = cache().lock().unwrap_or_else(PoisonError::into_inner); + match map.get(key).and_then(Weak::upgrade) { + Some(inner) => Some(WorkspaceStores { inner }), + None => { + map.remove(key); + None + } + } +} + +impl WorkspaceStores { + /// Opens (or reuses) the stores of `agent_id` for `workspace_dir`, running + /// the one-time legacy JSONL import on first open. + /// + /// # Errors + /// + /// The SQLite store cannot be opened, or the import is refused. Nothing is + /// cached then, so the next call retries. + pub fn open(workspace_dir: &Path, agent_id: &str) -> anyhow::Result { + let key = (workspace_dir.to_path_buf(), agent_id.to_string()); + if let Some(found) = cached(&key) { + return Ok(found); + } + let _opening = OPEN_LOCK.lock().unwrap_or_else(PoisonError::into_inner); + // Another thread may have finished while this one waited. + if let Some(found) = cached(&key) { + return Ok(found); + } + let backend: Arc = + Arc::new(SqliteStorage::open(workspace_dir.join(STORE_DIR))?); + let provider = Arc::new(DriverSessionStores::new(backend)?); + let stores = provider.try_for_agent(agent_id)?; + let outcome = import_transcripts_once(&provider, agent_id, workspace_dir)?; + tracing::debug!( + target: "tinyagents_session::port::workspace", + agent_id, + "[session-store] workspace stores opened" + ); + // An import that skipped an unreadable file is not complete: do not + // cache it, so the next open retries once the file is fixed. + let complete = matches!( + outcome, + OnceOutcome::AlreadyImported | OnceOutcome::Imported { marked: true, .. } + ); + let inner = Arc::new(WorkspaceInner { + provider, + stores, + outcome, + }); + if complete { + cache() + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(key, Arc::downgrade(&inner)); + } + Ok(Self { inner }) + } + + /// The transcript port. + pub fn transcripts(&self) -> Arc { + Arc::clone(&self.inner.stores.transcripts) + } + + /// Every store of the agent. + pub fn stores(&self) -> &AgentStores { + &self.inner.stores + } + + /// The provider the stores come from. + pub fn provider(&self) -> &Arc { + &self.inner.provider + } + + /// What the first open's legacy import did. + pub fn import_outcome(&self) -> &OnceOutcome { + &self.inner.outcome + } +} + +#[cfg(test)] +#[path = "workspace_tests.rs"] +mod tests; diff --git a/crates/tinyagents-session/src/port/workspace_tests.rs b/crates/tinyagents-session/src/port/workspace_tests.rs new file mode 100644 index 000000000..01561fe92 --- /dev/null +++ b/crates/tinyagents-session/src/port/workspace_tests.rs @@ -0,0 +1,133 @@ +use super::*; +use crate::legacy::OnceOutcome; +use crate::transcript::{ + SessionRef, TranscriptMessage, TranscriptMeta, TranscriptTurn, session_stem, write_transcript, +}; + +fn meta() -> TranscriptMeta { + TranscriptMeta { + session_id: None, + parent_session_id: None, + agent_name: "agent".into(), + agent_id: Some("agent".into()), + agent_type: Some("root".into()), + 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: Some("thread-ws".into()), + task_id: None, + } +} + +#[test] +fn legacy_generations_resume_through_the_port_and_a_new_generation_chains() { + let dir = tempfile::tempdir().unwrap(); + let raw = dir.path().join("session_raw"); + std::fs::create_dir_all(&raw).unwrap(); + let session = SessionRef::root("thread-ws"); + let g1 = session.next_generation(); + let sealed = raw.join(format!("{}.jsonl", session_stem(&session))); + let head = raw.join(format!("{}.jsonl", session_stem(&g1))); + let rows = |t: &str| { + vec![ + TranscriptMessage::user(t), + TranscriptMessage::assistant("ok"), + ] + }; + write_transcript(&sealed, &rows("first"), &meta(), None).unwrap(); + write_transcript(&head, &rows("second"), &meta(), None).unwrap(); + let (sealed_bytes, head_bytes) = ( + std::fs::read(&sealed).unwrap(), + std::fs::read(&head).unwrap(), + ); + + let opened = WorkspaceStores::open(dir.path(), "agent").unwrap(); + let OnceOutcome::Imported { report, marked } = opened.import_outcome() else { + panic!("first open imports"); + }; + assert_eq!(report.imported, 2); + assert!(marked); + + let locator = opened.transcripts(); + assert_eq!( + locator.session_chain(&session), + vec![session.clone(), g1.clone()] + ); + assert_eq!(locator.head_generation(&session), g1); + let resumed = locator + .read_session_transcript(&g1) + .unwrap() + .read_session() + .unwrap() + .unwrap(); + assert_eq!(resumed.messages[0].content, "second"); + + // Compaction never erases: a new generation chains, both predecessors stay. + let (g2, handle) = locator.begin_generation(&g1, meta()).unwrap(); + let next = rows("summary"); + handle + .append_turn(TranscriptTurn { + prev: &[], + next: &next, + meta: &meta(), + turn_usage: None, + request_id: None, + tools: None, + }) + .unwrap(); + assert_eq!(locator.session_chain(&session).len(), 3); + assert_eq!(locator.head_generation(&session), g2); + assert!(locator.session_exists(&session)); + + // Legacy files were only read. + assert_eq!(std::fs::read(&sealed).unwrap(), sealed_bytes); + assert_eq!(std::fs::read(&head).unwrap(), head_bytes); + assert!(dir.path().join(STORE_DIR).is_dir()); + + // The handle is cached per workspace and agent. + let again = WorkspaceStores::open(dir.path(), "agent").unwrap(); + assert!(Arc::ptr_eq(again.provider(), opened.provider())); +} + +#[test] +fn a_workspace_with_no_legacy_files_opens_empty() { + let dir = tempfile::tempdir().unwrap(); + let opened = WorkspaceStores::open(dir.path(), "agent").unwrap(); + assert!(opened.transcripts().latest_for_agent("agent").is_none()); +} + +#[test] +fn a_dropped_handle_releases_its_backend_and_reopens_fresh() { + let dir = tempfile::tempdir().unwrap(); + let first = WorkspaceStores::open(dir.path(), "agent").unwrap(); + let weak = Arc::downgrade(first.provider()); + drop(first); + assert!( + weak.upgrade().is_none(), + "cache must not keep the backend alive" + ); + WorkspaceStores::open(dir.path(), "agent").unwrap(); +} + +#[test] +fn an_incomplete_import_is_retried_on_the_next_open() { + let dir = tempfile::tempdir().unwrap(); + let raw = dir.path().join("session_raw"); + std::fs::create_dir_all(&raw).unwrap(); + std::fs::write(raw.join("broken.jsonl"), "not json at all\n").unwrap(); + let first = WorkspaceStores::open(dir.path(), "agent").unwrap(); + assert!(matches!( + first.import_outcome(), + OnceOutcome::Imported { marked: false, .. } + )); + let second = WorkspaceStores::open(dir.path(), "agent").unwrap(); + assert!(!Arc::ptr_eq(first.provider(), second.provider())); +}