feat(session): workspace transcripts on driver ports with one-time legacy JSONL import

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 `<workspace>/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`

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;

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 `<workspace_dir>/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<OnceOutcome> {
+    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<bool, tinystoragedrivers_core::StorageError> {
+    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(|_| ()),
+    }
+}

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<dyn StorageBackend>).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, .. }));
+}

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,
+    },
+}

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<dyn DocumentStore>, 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)

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};
 

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<WorkspaceInner>,
+}
+
+#[cfg(feature = "legacy-import")]
+pub(super) struct WorkspaceInner {
+    pub(super) provider: Arc<super::DriverSessionStores>,
+    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()
+    }
+}