From f3b6b9adf12b512b89452b9909d48915ce5c2754 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 07:40:00 +0300 Subject: [PATCH 1/8] chore(session): remove legacy transcript migration code The legacy module and its one-shot transcript migration driver are no longer needed now that sessions are read through the port layer, so the dead code paths have been dropped. Auto-committed-on: dragonfly --- crates/tinyagents-session/src/legacy/mod.rs | 7 ++ .../src/legacy/transcripts_once.rs | 111 ++++++++++++++++++ .../src/port/drivers/mod.rs | 19 +++ 3 files changed, 137 insertions(+) create mode 100644 crates/tinyagents-session/src/legacy/transcripts_once.rs diff --git a/crates/tinyagents-session/src/legacy/mod.rs b/crates/tinyagents-session/src/legacy/mod.rs index 93cd8d4b9..67e175b1f 100644 --- a/crates/tinyagents-session/src/legacy/mod.rs +++ b/crates/tinyagents-session/src/legacy/mod.rs @@ -31,15 +31,22 @@ pub(crate) mod rows; pub mod source; pub mod transcripts; +mod transcripts_once; pub mod turn_states; 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, OnceOutcome, TRANSCRIPT_MARKER_ID, import_transcripts_once, +}; pub use turn_states::{TurnStateImportReport, import_turn_states}; pub use workspace::{LegacyWorkspaceReport, WorkspaceLayout, import_workspace}; +#[cfg(test)] +#[path = "transcripts_once_tests.rs"] +mod transcripts_once_tests; #[cfg(test)] #[path = "files_tests.rs"] mod files_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..3afb240ba --- /dev/null +++ b/crates/tinyagents-session/src/legacy/transcripts_once.rs @@ -0,0 +1,111 @@ +//! 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 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]"; + +/// 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, + }, +} + +/// 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/port/drivers/mod.rs b/crates/tinyagents-session/src/port/drivers/mod.rs index 550d00808..3a1f40b84 100644 --- a/crates/tinyagents-session/src/port/drivers/mod.rs +++ b/crates/tinyagents-session/src/port/drivers/mod.rs @@ -202,6 +202,25 @@ 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. + 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. + 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) From a2cd8588feafb15eb04c47b9d14712a6fd7eb8d2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 07:40:14 +0300 Subject: [PATCH 2/8] refactor(session): move workspace port into its own module Split the workspace port out of the port module root into a dedicated workspace submodule so the port definitions stay organised as more session ports are added. Auto-committed-on: dragonfly --- crates/tinyagents-session/src/port/mod.rs | 4 + .../tinyagents-session/src/port/workspace.rs | 113 ++++++++++++++++++ 2 files changed, 117 insertions(+) create mode 100644 crates/tinyagents-session/src/port/workspace.rs diff --git a/crates/tinyagents-session/src/port/mod.rs b/crates/tinyagents-session/src/port/mod.rs index 9d50cdbd3..42848ca43 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,8 @@ pub use drivers::{ }; pub use memory::{InMemorySessionStores, InMemoryTranscriptLocator, InMemoryTurnStates}; pub use types::AgentStores; +#[cfg(feature = "legacy-import")] +pub use workspace::{STORE_DIR, WorkspaceStores}; pub use tinyagents_harness::store::{AppendStore, Store}; diff --git a/crates/tinyagents-session/src/port/workspace.rs b/crates/tinyagents-session/src/port/workspace.rs new file mode 100644 index 000000000..054beca65 --- /dev/null +++ b/crates/tinyagents-session/src/port/workspace.rs @@ -0,0 +1,113 @@ +//! 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}; + +use tinystoragedrivers_core::StorageBackend; +use tinystoragedrivers_sqlite::SqliteStorage; + +use super::{AgentStores, DriverSessionStores}; +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"; + +/// An agent's driver-backed stores for one workspace. +#[derive(Clone)] +pub struct WorkspaceStores { + provider: Arc, + stores: AgentStores, + outcome: Arc, +} + +impl std::fmt::Debug for WorkspaceStores { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WorkspaceStores") + .field("outcome", &self.outcome) + .finish_non_exhaustive() + } +} + +type Cache = Mutex>; + +fn cache() -> &'static Cache { + static CACHE: OnceLock = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(HashMap::new())) +} + +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) = cache() + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(&key) + { + return Ok(found.clone()); + } + 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" + ); + let opened = Self { + provider, + stores, + outcome: Arc::new(outcome), + }; + Ok(cache() + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry(key) + .or_insert(opened) + .clone()) + } + + /// The transcript port. + pub fn transcripts(&self) -> Arc { + Arc::clone(&self.stores.transcripts) + } + + /// Every store of the agent. + pub fn stores(&self) -> &AgentStores { + &self.stores + } + + /// The provider the stores come from. + pub fn provider(&self) -> &Arc { + &self.provider + } + + /// What the first open's legacy import did. + pub fn import_outcome(&self) -> &OnceOutcome { + &self.outcome + } +} + +#[cfg(test)] +#[path = "workspace_tests.rs"] +mod tests; From dde3c7ccce5b77b608a6ec0c37809f3c7ed2f1df Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 07:40:42 +0300 Subject: [PATCH 3/8] style(session): reformat legacy transcript marker code Reformatted the marker helper functions in the legacy transcript importer to satisfy rustfmt line width limits, and reordered the test module declaration in the legacy module. No behaviour changes. Auto-committed-on: dragonfly --- crates/tinyagents-session/src/legacy/mod.rs | 6 +- .../src/legacy/transcripts_once.rs | 16 ++- .../src/legacy/transcripts_once_tests.rs | 113 ++++++++++++++++++ .../src/port/workspace_tests.rs | 105 ++++++++++++++++ 4 files changed, 234 insertions(+), 6 deletions(-) create mode 100644 crates/tinyagents-session/src/legacy/transcripts_once_tests.rs create mode 100644 crates/tinyagents-session/src/port/workspace_tests.rs diff --git a/crates/tinyagents-session/src/legacy/mod.rs b/crates/tinyagents-session/src/legacy/mod.rs index 67e175b1f..3f5fdce95 100644 --- a/crates/tinyagents-session/src/legacy/mod.rs +++ b/crates/tinyagents-session/src/legacy/mod.rs @@ -44,9 +44,6 @@ pub use transcripts_once::{ pub use turn_states::{TurnStateImportReport, import_turn_states}; pub use workspace::{LegacyWorkspaceReport, WorkspaceLayout, import_workspace}; -#[cfg(test)] -#[path = "transcripts_once_tests.rs"] -mod transcripts_once_tests; #[cfg(test)] #[path = "files_tests.rs"] mod files_tests; @@ -55,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 index 3afb240ba..42d33087d 100644 --- a/crates/tinyagents-session/src/legacy/transcripts_once.rs +++ b/crates/tinyagents-session/src/legacy/transcripts_once.rs @@ -83,10 +83,15 @@ pub fn import_transcripts_once( Ok(OnceOutcome::Imported { report, marked }) } -async fn marker_present(docs: &dyn DocumentStore) -> Result { +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()) + Ok(docs + .get(MARKER_COLLECTION, TRANSCRIPT_MARKER_ID) + .await? + .is_some()) } async fn write_marker( @@ -101,7 +106,12 @@ async fn write_marker( "messages": report.messages, }); match docs - .put(MARKER_COLLECTION, TRANSCRIPT_MARKER_ID, doc, Precondition::Absent) + .put( + MARKER_COLLECTION, + TRANSCRIPT_MARKER_ID, + doc, + Precondition::Absent, + ) .await { // A concurrent launch marked it first; the marker is idempotent. 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..314bf95b5 --- /dev/null +++ b/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs @@ -0,0 +1,113 @@ +//! The one-time transcript import: marker, idempotence, untouched sources. + +use super::*; +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/port/workspace_tests.rs b/crates/tinyagents-session/src/port/workspace_tests.rs new file mode 100644 index 000000000..8d8e0464e --- /dev/null +++ b/crates/tinyagents-session/src/port/workspace_tests.rs @@ -0,0 +1,105 @@ +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()); +} From 5669b828b3952ce1e4c3da89856815ccc125726c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 07:44:56 +0300 Subject: [PATCH 4/8] feat(session): serve workspace transcripts through driver ports Adds `port::WorkspaceStores::open`, which serves transcripts from a SQLite backend under the workspace and hands callers a `TranscriptLocator` port instead of a file path, with handles cached per workspace and agent. A one-time legacy import moves classic `session_raw/*.jsonl` files into the store, recording completion in a marker document so unreadable files are retried and the originals are left untouched. The two driver accessors used Auto-committed-on: dragonfly --- CHANGELOG.md | 16 ++++++++++++++++ .../tinyagents-session/src/port/drivers/mod.rs | 2 ++ 2 files changed, 18 insertions(+) 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/port/drivers/mod.rs b/crates/tinyagents-session/src/port/drivers/mod.rs index 3a1f40b84..961e50b11 100644 --- a/crates/tinyagents-session/src/port/drivers/mod.rs +++ b/crates/tinyagents-session/src/port/drivers/mod.rs @@ -205,6 +205,7 @@ 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, @@ -217,6 +218,7 @@ impl DriverSessionStores { } /// The bridge this provider runs synchronous seams on. + #[cfg(feature = "legacy-import")] pub(crate) fn bridge(&self) -> &Blocking { &self.bridge } From e7f68b175330c2eba7a738c5495a21398260e704 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 08:00:02 +0300 Subject: [PATCH 5/8] refactor(session): split legacy and port session types into modules Moved the session type definitions out of the monolithic legacy module into dedicated legacy and port submodules, with the one-shot transcript logic and its tests now living in their own files. This is a pure restructuring to make the legacy and port boundaries explicit without changing behaviour. Auto-committed-on: dragonfly --- crates/tinyagents-session/src/legacy/mod.rs | 6 ++--- .../src/legacy/transcripts_once.rs | 16 +------------ .../src/legacy/transcripts_once_tests.rs | 4 ++-- crates/tinyagents-session/src/legacy/types.rs | 18 +++++++++++++++ crates/tinyagents-session/src/port/mod.rs | 2 +- crates/tinyagents-session/src/port/types.rs | 23 +++++++++++++++++++ 6 files changed, 48 insertions(+), 21 deletions(-) create mode 100644 crates/tinyagents-session/src/legacy/types.rs diff --git a/crates/tinyagents-session/src/legacy/mod.rs b/crates/tinyagents-session/src/legacy/mod.rs index 3f5fdce95..1124b960c 100644 --- a/crates/tinyagents-session/src/legacy/mod.rs +++ b/crates/tinyagents-session/src/legacy/mod.rs @@ -32,15 +32,15 @@ pub(crate) mod rows; pub mod source; pub mod transcripts; mod transcripts_once; +mod types; pub mod turn_states; 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, OnceOutcome, TRANSCRIPT_MARKER_ID, import_transcripts_once, -}; +pub use transcripts_once::{MARKER_COLLECTION, TRANSCRIPT_MARKER_ID, import_transcripts_once}; +pub use types::OnceOutcome; pub use turn_states::{TurnStateImportReport, import_turn_states}; pub use workspace::{LegacyWorkspaceReport, WorkspaceLayout, import_workspace}; diff --git a/crates/tinyagents-session/src/legacy/transcripts_once.rs b/crates/tinyagents-session/src/legacy/transcripts_once.rs index 42d33087d..284ae6262 100644 --- a/crates/tinyagents-session/src/legacy/transcripts_once.rs +++ b/crates/tinyagents-session/src/legacy/transcripts_once.rs @@ -14,6 +14,7 @@ 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. @@ -23,21 +24,6 @@ pub const TRANSCRIPT_MARKER_ID: &str = "session_raw_jsonl_v1"; const LOG_PREFIX: &str = "[session_db:import:transcripts:once]"; -/// 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, - }, -} - /// Imports `/session_raw/` into `agent_id`'s transcripts once. /// /// A marker document in the agent's scope records completion. A provider diff --git a/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs b/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs index 314bf95b5..e55bc6831 100644 --- a/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs +++ b/crates/tinyagents-session/src/legacy/transcripts_once_tests.rs @@ -1,6 +1,6 @@ -//! The one-time transcript import: marker, idempotence, untouched sources. - 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; 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/mod.rs b/crates/tinyagents-session/src/port/mod.rs index 42848ca43..e778d46dc 100644 --- a/crates/tinyagents-session/src/port/mod.rs +++ b/crates/tinyagents-session/src/port/mod.rs @@ -59,7 +59,7 @@ pub use drivers::{ DriverSessionStores, DriverTranscriptHistory, DriverTranscriptLocator, DriverTurnStates, }; pub use memory::{InMemorySessionStores, InMemoryTranscriptLocator, InMemoryTurnStates}; -pub use types::AgentStores; +pub use types::{AgentStores, WorkspaceStores}; #[cfg(feature = "legacy-import")] pub use workspace::{STORE_DIR, WorkspaceStores}; diff --git a/crates/tinyagents-session/src/port/types.rs b/crates/tinyagents-session/src/port/types.rs index f02bf6ea3..e6e55fa21 100644 --- a/crates/tinyagents-session/src/port/types.rs +++ b/crates/tinyagents-session/src/port/types.rs @@ -32,3 +32,26 @@ 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)] +pub struct WorkspaceStores { + pub(super) inner: Arc, +} + +pub(super) struct WorkspaceInner { + pub(super) provider: Arc, + pub(super) stores: AgentStores, + pub(super) outcome: crate::legacy::OnceOutcome, +} + +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() + } +} From 7d77fb6e95ad6423e6c620770f64a600e351a927 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 08:00:27 +0300 Subject: [PATCH 6/8] refactor(session): split workspace port into its own module Moved the workspace port definitions out of the port module root into a dedicated workspace submodule so the port surface is easier to navigate as it grows. No behaviour change. Auto-committed-on: dragonfly --- crates/tinyagents-session/src/port/mod.rs | 2 +- .../tinyagents-session/src/port/workspace.rs | 88 +++++++++++-------- 2 files changed, 51 insertions(+), 39 deletions(-) diff --git a/crates/tinyagents-session/src/port/mod.rs b/crates/tinyagents-session/src/port/mod.rs index e778d46dc..3faa7da37 100644 --- a/crates/tinyagents-session/src/port/mod.rs +++ b/crates/tinyagents-session/src/port/mod.rs @@ -61,7 +61,7 @@ pub use drivers::{ pub use memory::{InMemorySessionStores, InMemoryTranscriptLocator, InMemoryTurnStates}; pub use types::{AgentStores, WorkspaceStores}; #[cfg(feature = "legacy-import")] -pub use workspace::{STORE_DIR, WorkspaceStores}; +pub use workspace::STORE_DIR; pub use tinyagents_harness::store::{AppendStore, Store}; diff --git a/crates/tinyagents-session/src/port/workspace.rs b/crates/tinyagents-session/src/port/workspace.rs index 054beca65..51a528534 100644 --- a/crates/tinyagents-session/src/port/workspace.rs +++ b/crates/tinyagents-session/src/port/workspace.rs @@ -12,41 +12,45 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex, OnceLock, PoisonError}; +use std::sync::{Arc, Mutex, OnceLock, PoisonError, Weak}; use tinystoragedrivers_core::StorageBackend; use tinystoragedrivers_sqlite::SqliteStorage; -use super::{AgentStores, DriverSessionStores}; +use super::types::WorkspaceInner; +use super::AgentStores; +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"; -/// An agent's driver-backed stores for one workspace. -#[derive(Clone)] -pub struct WorkspaceStores { - provider: Arc, - stores: AgentStores, - outcome: Arc, -} - -impl std::fmt::Debug for WorkspaceStores { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("WorkspaceStores") - .field("outcome", &self.outcome) - .finish_non_exhaustive() - } -} - -type Cache = Mutex>; +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. @@ -57,12 +61,13 @@ impl WorkspaceStores { /// 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) = cache() - .lock() - .unwrap_or_else(PoisonError::into_inner) - .get(&key) - { - return Ok(found.clone()); + 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))?); @@ -74,37 +79,44 @@ impl WorkspaceStores { agent_id, "[session-store] workspace stores opened" ); - let opened = Self { + // 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: Arc::new(outcome), - }; - Ok(cache() - .lock() - .unwrap_or_else(PoisonError::into_inner) - .entry(key) - .or_insert(opened) - .clone()) + 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.stores.transcripts) + Arc::clone(&self.inner.stores.transcripts) } /// Every store of the agent. pub fn stores(&self) -> &AgentStores { - &self.stores + &self.inner.stores } /// The provider the stores come from. pub fn provider(&self) -> &Arc { - &self.provider + &self.inner.provider } /// What the first open's legacy import did. pub fn import_outcome(&self) -> &OnceOutcome { - &self.outcome + &self.inner.outcome } } From d9f83cfd5b6d5fa4c2523b4f46eff1cbdfeea77c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 08:00:50 +0300 Subject: [PATCH 7/8] refactor(session): move workspace port into its own module The workspace port trait and its tests were relocated into a dedicated port module to keep the session crate's public surface organised as it grows. No behaviour changes. Auto-committed-on: dragonfly --- crates/tinyagents-session/src/legacy/mod.rs | 4 +-- .../tinyagents-session/src/port/workspace.rs | 2 +- .../src/port/workspace_tests.rs | 28 +++++++++++++++++++ 3 files changed, 31 insertions(+), 3 deletions(-) diff --git a/crates/tinyagents-session/src/legacy/mod.rs b/crates/tinyagents-session/src/legacy/mod.rs index 1124b960c..70b603851 100644 --- a/crates/tinyagents-session/src/legacy/mod.rs +++ b/crates/tinyagents-session/src/legacy/mod.rs @@ -32,16 +32,16 @@ pub(crate) mod rows; pub mod source; pub mod transcripts; mod transcripts_once; -mod types; 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 types::OnceOutcome; pub use turn_states::{TurnStateImportReport, import_turn_states}; +pub use types::OnceOutcome; pub use workspace::{LegacyWorkspaceReport, WorkspaceLayout, import_workspace}; #[cfg(test)] diff --git a/crates/tinyagents-session/src/port/workspace.rs b/crates/tinyagents-session/src/port/workspace.rs index 51a528534..c460f1a7c 100644 --- a/crates/tinyagents-session/src/port/workspace.rs +++ b/crates/tinyagents-session/src/port/workspace.rs @@ -17,8 +17,8 @@ use std::sync::{Arc, Mutex, OnceLock, PoisonError, Weak}; use tinystoragedrivers_core::StorageBackend; use tinystoragedrivers_sqlite::SqliteStorage; -use super::types::WorkspaceInner; use super::AgentStores; +use super::types::WorkspaceInner; use super::{DriverSessionStores, WorkspaceStores}; use crate::legacy::{OnceOutcome, import_transcripts_once}; use crate::transcript::TranscriptLocator; diff --git a/crates/tinyagents-session/src/port/workspace_tests.rs b/crates/tinyagents-session/src/port/workspace_tests.rs index 8d8e0464e..01561fe92 100644 --- a/crates/tinyagents-session/src/port/workspace_tests.rs +++ b/crates/tinyagents-session/src/port/workspace_tests.rs @@ -103,3 +103,31 @@ fn a_workspace_with_no_legacy_files_opens_empty() { 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())); +} From 83754eeb09907e74cc64fca0747c3d62ea968690 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 15 Oct 2026 08:15:18 +0300 Subject: [PATCH 8/8] refactor(session): move port types into a dedicated module Split the session port types out of the port module root into their own types submodule so the trait definitions and their supporting data structures can evolve independently. No behaviour changes. Auto-committed-on: dragonfly --- crates/tinyagents-session/src/port/mod.rs | 4 +++- crates/tinyagents-session/src/port/types.rs | 3 +++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/tinyagents-session/src/port/mod.rs b/crates/tinyagents-session/src/port/mod.rs index 3faa7da37..d2395ead2 100644 --- a/crates/tinyagents-session/src/port/mod.rs +++ b/crates/tinyagents-session/src/port/mod.rs @@ -59,7 +59,9 @@ pub use drivers::{ DriverSessionStores, DriverTranscriptHistory, DriverTranscriptLocator, DriverTurnStates, }; pub use memory::{InMemorySessionStores, InMemoryTranscriptLocator, InMemoryTurnStates}; -pub use types::{AgentStores, WorkspaceStores}; +pub use types::AgentStores; +#[cfg(feature = "legacy-import")] +pub use types::WorkspaceStores; #[cfg(feature = "legacy-import")] pub use workspace::STORE_DIR; diff --git a/crates/tinyagents-session/src/port/types.rs b/crates/tinyagents-session/src/port/types.rs index e6e55fa21..889b06c9d 100644 --- a/crates/tinyagents-session/src/port/types.rs +++ b/crates/tinyagents-session/src/port/types.rs @@ -38,16 +38,19 @@ impl std::fmt::Debug for AgentStores { /// 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")