feat(session): host-agnostic transcript view, legacy-session importer and turn-state store

crates/tinyagents-session/Cargo.toml

@@ -16,7 +16,14 @@ rusqlite = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tinyagents-harness = { path = "../tinyagents-harness", version = "2.1.2", default-features = false, features = ["sqlite"] }
+tinytools-agent = { path = "../../vendor/tinytools/crates/tinytools-agent", version = "0.4.1", default-features = false }
+glob = "0.3"
+hex = "0.4"
+sha2 = "0.11"
+parking_lot = "0.12"
+tempfile = { workspace = true }
 tracing = { workspace = true }
+tokio = { workspace = true, default-features = true, features = ["sync"] }
 
 [features]
 default = []
@@ -26,7 +33,6 @@ default = []
 tracing = ["tinyagents-harness/tracing"]
 
 [dev-dependencies]
-tempfile = { workspace = true }
 tokio = { workspace = true, default-features = true, features = ["macros", "rt-multi-thread"] }
 
 [lints]

crates/tinyagents-session/src/lib.rs

@@ -83,6 +83,7 @@ pub mod run_ledger;
 mod store;
 pub mod testkit;
 pub mod transcript;
+pub mod turn_state;
 pub mod types;
 
 pub use tinyagents_harness::error::{Result, TinyAgentsError};

crates/tinyagents-session/src/transcript.rs

@@ -110,6 +110,7 @@
 
 mod adoption;
 mod history;
+pub mod import;
 mod jsonl;
 mod legacy_md;
 mod markdown;
@@ -119,6 +120,7 @@ mod reader;
 mod session;
 mod thread_lookup;
 mod types;
+pub mod view;
 mod writer;
 
 pub use adoption::{SessionAdoption, adopt_legacy_session_transcripts};

crates/tinyagents-session/src/transcript/import/README.md

@@ -0,0 +1,90 @@
+# transcript::import
+
+One-time migration of legacy session transcripts into TinyAgents
+`Store`/`AppendStore` records ([`ops::run_import`]). Hosts expose it as an
+explicit command, never a boot hook. The module also owns the live
+dual-write and shadow-read paths that keep new turns landing in the same
+store layout (`live.rs`), and [`ops::open_session_stores`], which every other
+TinyAgents-store consumer reuses. The host supplies a
+[`convert::JournalProjector`] so journal records carry its message metadata;
+whether the live paths run is the host's decision.
+
+> The migration was specified in a design doc for issue #4249 that is not
+> checked into this repository; `types.rs` still refers to it when explaining
+> why stream names are dot-separated.
+
+## Sources and destination
+
+- Transcript JSONL under `session_raw/`: the current flat layout
+  (`session_raw/{stem}.jsonl`) and the legacy date-folder layout
+  (`session_raw/{DDMMYYYY}/{stem}.jsonl`).
+- Markdown-only sessions under `sessions/{dir}/{stem}.md` that have no JSONL
+  twin. A Markdown file next to a JSONL source is recorded as a companion
+  pointer only, never read.
+- Precedence per session stem: flat JSONL beats legacy-dir JSONL beats
+  Markdown-only (`scan.rs::discover_sources`); a duplicate stem is a warning,
+  not a second item.
+- `session_db/sessions.db` is opened read-only, when present, to join
+  `agent_runs` ids onto the descriptor and cross-check parent lineage. The
+  stem chain (`a__b` maps to parent `a`) wins over a disagreeing ledger.
+
+All writes land under `{workspace}/tinyagents_store/`:
+
+- `kv/sessions/{sanitized stem}.json`: `SessionDescriptor` compatibility
+  record mapping the OpenHuman session key to TinyAgents identifiers.
+- `kv/migration_items/{sha256(relative source path)}.json`: per-item
+  idempotency ledger (`ItemLedgerRecord`).
+- `kv/migrations/session_import_v1.json`: global run marker.
+- `journal/session.{stem}.messages.jsonl`: message journal, one
+  `StoreRecord` per line via TinyAgents `JsonlAppendStore`.
+
+Source files are never mutated or deleted.
+
+## Idempotency
+
+- A full run (no `only`) that is not a dry run writes the global marker
+  (`MARKER_KEY`). A later full, non-forced, non-dry run sees the marker and
+  returns `already_done` without scanning.
+- Independently, every written source gets a ledger entry keyed by the sha256
+  of its workspace-relative path holding `IMPORT_VERSION`, size, and mtime.
+  A source whose fingerprint still matches is `skipped_unchanged`, even under
+  `only` or after the global marker is gone.
+- `force` ignores both the marker and the ledger.
+- `dry_run` bypasses the marker fast path, reports `would_import` per item,
+  and writes nothing.
+- Re-import is a full rewrite: the journal stream file is deleted and
+  re-appended because `JsonlAppendStore` has no truncate.
+
+## Key files
+
+| File | Role |
+| --- | --- |
+| `mod.rs` | Module docs and re-exports (`ImportOptions`, `ImportSummary`, controller registration). |
+| `types.rs` | Serde types: `ImportOptions`, `ImportSummary`, `ItemReport`, `SessionDescriptor`, `JournalMessage`, `ItemLedgerRecord`, and the store-layout constants (`KV_SUBDIR`, `JOURNAL_SUBDIR`, `NS_*`, `MARKER_KEY`, `IMPORT_VERSION`). |
+| `scan.rs` | `discover_sources` walks `session_raw/` and `sessions/`, deduping by stem per the precedence order above. |
+| `convert.rs` | Pure helpers: `parent_session_key` stem lineage, `sanitize_store_name`, `stream_name`, `effective_thread_id` (synthesizes `imported-{stem}` when `_meta` has none), `build_descriptor`, `journal_messages`. |
+| `ops.rs` | `run_import` scans, reads all, plans/writes per item, then writes the marker; `open_session_stores` opens the shared KV/journal handles over `{workspace}/tinyagents_store/{kv,journal}`. |
+| `live.rs` | Dual-write of each persisted turn into the same layout (`write_live_turn`), gated by `AgentConfig::session_dual_write` (default on) with the `OPENHUMAN_SESSION_DUAL_WRITE` env kill switch; store-backed shadow read of the same session (`shadow_read_compare`), gated by `AgentConfig::session_shadow_reads` (default on) with `OPENHUMAN_SESSION_SHADOW_READS`; `session_kv_store` exposes the KV store on `RunContext.stores` as `TINYAGENTS_SESSION_KV_STORE`. Env vars can only force off, never on. Errors are logged and swallowed by callers; the legacy transcript stays authoritative for both reads and writes. |
+| `schemas.rs` | `session_import.run` `ControllerSchema` and handler; resolves the workspace from config unless `workspace` is passed. |
+| `*_tests.rs` | Sibling test suites for `convert`, `live`, `ops`, `schemas`. |
+
+## RPC
+
+`session_import.run` accepts `dry_run`, `only` (glob over session stems),
+`force`, `verbose`, and an optional `workspace` override, and returns an
+`ImportSummary`. Registered via `all_session_import_registered_controllers`
+in `crates/openhuman-core/src/core/all.rs`.
+
+## Used by
+
+- `crates/openhuman-core/src/core/all.rs`: registers the controller.
+- `tinyagents_session::transcript`: the legacy transcript readers
+  (`read_transcript`, `read_transcript_legacy_md`) the importer converts from.
+- `crates/openhuman-core/src/agent/session_host/turn/session_io/transcript_persist.rs`:
+  calls `live::write_live_turn` after each transcript write and
+  `live::shadow_read_compare` after each transcript load, both on background tasks.
+- `crates/openhuman-core/src/agent/tinyagents/turn_runner.rs`: registers
+  `live::session_kv_store` on the per-turn `RunContext`.
+- `open_session_stores` is reused by `agent/tinyagents/{journal,reaper,todos,replay/ops}.rs`
+  and `threads/goals/migration.rs` so every TinyAgents-store consumer shares
+  one layout.

crates/tinyagents-session/src/transcript/import/convert.rs

@@ -0,0 +1,139 @@
+//! Pure conversion helpers: stem lineage, stream naming, descriptor
+//! assembly, and message-record projection.
+
+use crate::transcript::{SessionTranscript, TranscriptMessage};
+use sha2::{Digest, Sha256};
+
+use super::types::{
+    DescriptorImport, DescriptorSource, DescriptorUsage, IMPORT_VERSION, JournalMessage,
+    SessionDescriptor,
+};
+
+/// Parent session key from the `__` stem chain.
+///
+/// Stems are `{parent_chain}__{unix_ts}_{agent_id}`; the parent key is
+/// everything before the **last** `__`. Roots (no `__`) have no parent.
+pub fn parent_session_key(stem: &str) -> Option<String> {
+    stem.rfind("__").map(|idx| stem[..idx].to_string())
+}
+
+/// Sanitize a string into the TinyAgents store-name alphabet (ASCII
+/// alphanumerics, `-`, `_`, `.`); anything else becomes `_`. Empty input
+/// becomes `"session"` to match the transcript layer's fallback.
+pub fn sanitize_store_name(name: &str) -> String {
+    let cleaned: String = name
+        .chars()
+        .map(|c| {
+            if c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.' {
+                c
+            } else {
+                '_'
+            }
+        })
+        .collect();
+    if cleaned.is_empty() {
+        "session".to_string()
+    } else if cleaned.bytes().all(|b| b == b'.') {
+        let digest = Sha256::digest(name.as_bytes());
+        format!("session_{}", hex::encode(&digest[..6]))
+    } else if cleaned != name {
+        let digest = Sha256::digest(name.as_bytes());
+        format!("{cleaned}_{}", hex::encode(&digest[..6]))
+    } else {
+        cleaned
+    }
+}
+
+/// Journal stream name for a session.
+///
+/// Per-session streams (`session.{stem}.messages`) rather than per-thread:
+/// multiple transcript files can share one `_meta.thread_id`, and appending
+/// them into a shared stream would interleave sessions. The descriptor
+/// carries `thread_id` so thread-level views can still be projected.
+pub fn stream_name(session_key: &str) -> String {
+    format!("session.{}.messages", sanitize_store_name(session_key))
+}
+
+/// Effective thread id for a transcript: `_meta.thread_id` when present,
+/// otherwise a synthesized stable id. Returns `(thread_id, synthesized)`.
+pub fn effective_thread_id(session_key: &str, meta_thread_id: Option<&str>) -> (String, bool) {
+    match meta_thread_id {
+        Some(t) if !t.is_empty() => (t.to_string(), false),
+        _ => (
+            format!("imported-{}", sanitize_store_name(session_key)),
+            true,
+        ),
+    }
+}
+
+/// Build the `sessions/{session_key}` descriptor from a parsed transcript.
+#[allow(clippy::too_many_arguments)]
+pub fn build_descriptor(
+    session_key: &str,
+    transcript: &SessionTranscript,
+    thread_id: String,
+    thread_id_synthesized: bool,
+    run_ids: Vec<String>,
+    source: DescriptorSource,
+    imported_at: String,
+    warnings: usize,
+) -> SessionDescriptor {
+    let meta = &transcript.meta;
+    SessionDescriptor {
+        session_key: session_key.to_string(),
+        parent_session_key: parent_session_key(session_key),
+        thread_id,
+        thread_id_synthesized,
+        task_id: meta.task_id.clone(),
+        run_ids,
+        stream: stream_name(session_key),
+        dispatcher: meta.dispatcher.clone(),
+        agent_name: meta.agent_name.clone(),
+        agent_id: meta.agent_id.clone(),
+        agent_type: meta.agent_type.clone(),
+        provider: meta.provider.clone(),
+        model: meta.model.clone(),
+        created: meta.created.clone(),
+        updated: meta.updated.clone(),
+        turn_count: meta.turn_count,
+        usage: DescriptorUsage {
+            input: meta.input_tokens,
+            output: meta.output_tokens,
+            cached_input: meta.cached_input_tokens,
+            cost_usd: meta.charged_amount_usd,
+        },
+        source,
+        import: DescriptorImport {
+            version: IMPORT_VERSION,
+            imported_at,
+            warnings,
+        },
+    }
+}
+
+/// Host-supplied projection of one durable transcript row into its journal
+/// record. The journal must carry whatever host metadata the host folds into a
+/// row when it reads a transcript back (OpenHuman reconstructs
+/// `openhuman_turn_usage` and friends), so the importer takes the projection as
+/// a parameter instead of assuming a message type.
+pub type JournalProjector = fn(TranscriptMessage) -> JournalMessage;
+
+/// Neutral projector: copies `id`, `role`, `content` and `extra_metadata`
+/// verbatim and adds nothing. For hosts with no message-shape sidecars, and for
+/// tests.
+pub fn plain_journal_message(message: TranscriptMessage) -> JournalMessage {
+    JournalMessage {
+        id: message.id,
+        role: message.role,
+        content: message.content,
+        extra_metadata: message.extra_metadata,
+    }
+}
+
+/// Project a transcript's messages into journal records.
+pub fn journal_messages(
+    transcript: &SessionTranscript,
+    project: JournalProjector,
+) -> Vec<JournalMessage> {
+    transcript.messages.iter().cloned().map(project).collect()
+}

crates/tinyagents-session/src/transcript/import/convert_test.rs

@@ -0,0 +1,47 @@
+use super::convert::*;
+
+#[test]
+fn parent_key_follows_last_double_underscore() {
+    assert_eq!(parent_session_key("1719_orchestrator"), None);
+    assert_eq!(
+        parent_session_key("1719_orchestrator__1720_researcher"),
+        Some("1719_orchestrator".to_string())
+    );
+    assert_eq!(
+        parent_session_key("a__b__c"),
+        Some("a__b".to_string()),
+        "two-level chains keep the full parent chain as the parent key"
+    );
+}
+
+#[test]
+fn sanitize_maps_unsafe_bytes_and_guards_dots() {
+    assert_eq!(sanitize_store_name("1719_agent"), "1719_agent");
+    assert!(sanitize_store_name("a/b:c d").starts_with("a_b_c_d_"));
+    assert_eq!(sanitize_store_name(""), "session");
+    assert!(sanitize_store_name("..").starts_with("session_"));
+}
+
+#[test]
+fn thread_id_synthesized_only_when_absent() {
+    assert_eq!(
+        effective_thread_id("s1", Some("t-1")),
+        ("t-1".to_string(), false)
+    );
+    assert_eq!(
+        effective_thread_id("s1", None),
+        ("imported-s1".to_string(), true)
+    );
+    assert_eq!(
+        effective_thread_id("s1", Some("")),
+        ("imported-s1".to_string(), true)
+    );
+}
+
+#[test]
+fn stream_name_is_store_safe() {
+    assert_eq!(
+        stream_name("1719_a__1720_b"),
+        "session.1719_a__1720_b.messages"
+    );
+}