diff --git a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs index ee5039f8bc8..3daca14f75b 100644 --- a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs +++ b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs @@ -15,10 +15,17 @@ impl SubagentSessionStore { } /// Imports the pre-documents session file once: its sessions are written - /// (an already stored newer copy wins) and the file is renamed to - /// `subagent_sessions.json.migrated`, so it is never imported again and - /// stays recoverable. A missing file is the common case and a no-op. + /// (an already stored newer copy wins) and a marker document records that + /// the import finished. The file is only read, never renamed or rewritten, + /// so it stays byte-identical. A missing file is the common case and a + /// no-op. fn import_legacy(&self, docs: &super::store_documents::Docs) -> Result<(), String> { + if docs + .legacy_imported(&self.workspace_dir) + .map_err(|err| format!("{err:#}"))? + { + return Ok(()); + } let path = self.legacy_path(); let raw = match std::fs::read_to_string(&path) { Ok(raw) => raw, @@ -27,14 +34,12 @@ impl SubagentSessionStore { }; let sessions: Vec = serde_json::from_str(&raw) .map_err(|err| format!("failed to parse legacy subagent sessions: {err}"))?; - docs.save(&sessions).map_err(|err| format!("{err:#}"))?; + docs.import_legacy(&self.workspace_dir, &sessions) + .map_err(|err| format!("{err:#}"))?; log::debug!( "[subagent_sessions] imported legacy sessions count={}", sessions.len() ); - if let Err(err) = std::fs::rename(&path, path.with_extension("json.migrated")) { - log::warn!("[subagent_sessions] could not retire legacy file: {err}"); - } Ok(()) } diff --git a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents.rs b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents.rs index 4e6719d6998..e17a48f96bd 100644 --- a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents.rs +++ b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents.rs @@ -10,6 +10,7 @@ //! | Collection | Document id | Holds | //! | --- | --- | --- | //! | `subagent_sessions` | the session id | one [`DurableSubagentSession`] plus `created_at` for ordering | +//! | `subagent_sessions_import` | `subagent_sessions.json:` | marker: the legacy file was imported (it is left byte-identical) | //! //! The callers in `ops.rs` load and save the whole list. `save` writes each session as //! its own document, so a session's history stays well under any document-size @@ -27,10 +28,20 @@ use crate::storage::documents::Repo; use crate::storage::DocumentStoreExt; const SESSIONS: &str = "subagent_sessions"; +/// Holds the marker document that records the one-time legacy file import, in +/// a collection of its own so `load` never sees it as a session. +const IMPORTED: &str = "subagent_sessions_import"; +const LEGACY_MARKER: &str = "subagent_sessions.json"; + +/// The marker id for one workspace's legacy file, so workspaces sharing a +/// backend each import their own. +fn marker_id(source: &std::path::Path) -> String { + format!("{LEGACY_MARKER}:{}", crate::storage::source_key(source)) +} const DOMAIN: &str = "subagent_sessions::store"; fn collections() -> Vec { - vec![CollectionSpec::new(SESSIONS)] + vec![CollectionSpec::new(SESSIONS), CollectionSpec::new(IMPORTED)] } /// The document store for this call: the installed backend, else the @@ -84,6 +95,41 @@ impl Docs { .collect() } + /// Whether the one-time legacy file import finished in this scope. + pub(super) fn legacy_imported(&self, source: &std::path::Path) -> Result { + let id = marker_id(source); + let found = self + .0 + .run(|docs| async move { docs.get(IMPORTED, &id).await })?; + Ok(found.is_some()) + } + + /// Imports `sessions` read from the legacy file (a stored newer copy + /// wins), then records the import in a marker document. + pub(super) fn import_legacy( + &self, + source: &std::path::Path, + sessions: &[DurableSubagentSession], + ) -> Result<()> { + self.save(sessions)?; + let id = marker_id(source); + self.0.run(|docs| async move { + match docs + .put( + IMPORTED, + &id, + serde_json::json!({ "imported": true }), + Precondition::Absent, + ) + .await + { + Ok(_) => Ok(()), + Err(error) if error.kind() == tinystoragedrivers::ErrorKind::Conflict => Ok(()), + Err(error) => Err(error), + } + }) + } + pub(super) fn save(&self, sessions: &[DurableSubagentSession]) -> Result<()> { let mut writes = Vec::with_capacity(sessions.len()); for session in sessions { diff --git a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents_tests.rs b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents_tests.rs index 72ad6b5f1e1..f6b5ab215a6 100644 --- a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents_tests.rs +++ b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store_documents_tests.rs @@ -143,21 +143,47 @@ fn a_stale_snapshot_does_not_revert_a_newer_session() { } #[test] -fn a_legacy_session_file_is_imported_once_and_retired() { +fn a_legacy_session_file_is_imported_once_and_left_byte_identical() { let dir = tempfile::tempdir().unwrap(); let legacy = dir.path().join(".openhuman"); std::fs::create_dir_all(&legacy).unwrap(); - std::fs::write( - legacy.join("subagent_sessions.json"), - serde_json::to_string(&vec![session("legacy-1")]).unwrap(), - ) - .unwrap(); + let original = serde_json::to_string(&vec![session("legacy-1")]).unwrap(); + std::fs::write(legacy.join("subagent_sessions.json"), &original).unwrap(); let store = SubagentSessionStore::new(dir.path().to_path_buf()); let loaded = store.load().unwrap(); assert_eq!(loaded.len(), 1); assert_eq!(loaded[0].subagent_session_id, "legacy-1"); - assert!(!legacy.join("subagent_sessions.json").exists()); - assert!(legacy.join("subagent_sessions.json.migrated").exists()); + // The legacy file is read only: same name, same bytes, no `.migrated`. + assert_eq!( + std::fs::read_to_string(legacy.join("subagent_sessions.json")).unwrap(), + original + ); + assert!(!legacy.join("subagent_sessions.json.migrated").exists()); assert_eq!(store.load().unwrap().len(), 1); + // The marker, not the file, stops a second import: a file that changed + // afterwards is not imported again. + let changed = serde_json::to_string(&vec![session("legacy-2")]).unwrap(); + std::fs::write(legacy.join("subagent_sessions.json"), &changed).unwrap(); + let again = store.load().unwrap(); + assert_eq!(again.len(), 1); + assert_eq!(again[0].subagent_session_id, "legacy-1"); +} + +#[test] +fn workspaces_sharing_a_backend_each_import_their_own_legacy_file() { + let docs = docs_in(&MemoryStorage::new(), "local"); + let (one, two) = ( + std::path::Path::new("/ws/one"), + std::path::Path::new("/ws/two"), + ); + assert!(!docs.legacy_imported(one).unwrap()); + docs.import_legacy(one, &[session("from-one")]).unwrap(); + assert!(docs.legacy_imported(one).unwrap()); + assert!( + !docs.legacy_imported(two).unwrap(), + "importing one workspace must not suppress another's import" + ); + docs.import_legacy(two, &[session("from-two")]).unwrap(); + assert_eq!(docs.load().unwrap().len(), 2); } diff --git a/crates/openhuman-core/src/agent/session_host/driver.rs b/crates/openhuman-core/src/agent/session_host/driver.rs index 7c6686db1a5..204fd94c520 100644 --- a/crates/openhuman-core/src/agent/session_host/driver.rs +++ b/crates/openhuman-core/src/agent/session_host/driver.rs @@ -1,9 +1,8 @@ //! OpenHuman's model-and-harness implementation of the neutral session driver. //! //! This adapter deliberately has no transcript, prefix, or generic history -//! state. `tinyagents_runtime::Session` owns those concerns and gives this -//! type the complete immutable model history and tool declaration snapshot for -//! one invocation. +//! state. `tinyagents_runtime::Session` owns those and hands this type the +//! complete immutable model history and tool declaration snapshot per call. mod grounded_close; diff --git a/crates/openhuman-core/src/integrations/composio/file_store.rs b/crates/openhuman-core/src/integrations/composio/file_store.rs index f3ce082391f..7fbe4cc54c0 100644 --- a/crates/openhuman-core/src/integrations/composio/file_store.rs +++ b/crates/openhuman-core/src/integrations/composio/file_store.rs @@ -20,14 +20,26 @@ pub(crate) const IDENTITIES_FILE: &str = "composio_identities.json"; pub(crate) const USER_SCOPES_FILE: &str = "composio_user_scopes.json"; /// Moves a pre-documents `/integrations/` file into the -/// document store, once: the value is stored only if none exists yet, then the -/// file is renamed to `.migrated` so it is never imported again and the -/// old data stays recoverable. A missing file is the common case and a no-op. +/// document store, once: the value is stored only if none exists yet, then a +/// marker document records that the import finished. The legacy file is only +/// ever read, never renamed or rewritten, so the old data stays byte-identical. +/// A missing file is the common case and a no-op. async fn import_legacy( config: &Config, docs: &super::file_store_documents::Docs, name: &'static str, ) -> Result<(), String> { + let marker_docs = docs.clone(); + let done = + crate::core::runtime::spawn_blocking_scoped(move || marker_docs.legacy_imported(name)) + .await + .map_err(|error| format!("[composio:store] checking legacy {name} failed: {error}"))? + .map_err(|error| { + format!("[composio:store] checking legacy {name} failed: {error:#}") + })?; + if done { + return Ok(()); + } let path = config.workspace_dir.join("integrations").join(name); let bytes = match tokio::fs::read(&path).await { Ok(bytes) => bytes, @@ -49,10 +61,6 @@ async fn import_legacy( format!("[composio:store] importing legacy {name} failed: {error:#}") })?; log::debug!("[composio:store] legacy import name={name} wrote={wrote}"); - let migrated = path.with_file_name(format!("{name}.migrated")); - if let Err(error) = tokio::fs::rename(&path, &migrated).await { - log::warn!("[composio:store] could not retire legacy {name}: {error}"); - } Ok(()) } diff --git a/crates/openhuman-core/src/integrations/composio/file_store_documents.rs b/crates/openhuman-core/src/integrations/composio/file_store_documents.rs index 4c89aa3aaf0..d8564938d1e 100644 --- a/crates/openhuman-core/src/integrations/composio/file_store_documents.rs +++ b/crates/openhuman-core/src/integrations/composio/file_store_documents.rs @@ -11,6 +11,7 @@ //! | Collection | Document id | Holds | //! | --- | --- | --- | //! | `composio_state` | the value name (`composio_identities.json`, ...) | `value`: the serde value | +//! | `composio_state_import` | the value name | marker: the legacy `integrations/` file was imported (the file is left byte-identical) | use std::path::Path; @@ -23,10 +24,13 @@ use tinystoragedrivers::{CollectionSpec, ErrorKind, Precondition, StorageError}; use crate::storage::documents::Repo; const STATE: &str = "composio_state"; +/// One marker document per legacy file whose one-time import finished, so the +/// file itself is never touched. +const IMPORTED: &str = "composio_state_import"; const DOMAIN: &str = "composio::file_store"; fn collections() -> Vec { - vec![CollectionSpec::new(STATE)] + vec![CollectionSpec::new(STATE), CollectionSpec::new(IMPORTED)] } /// The document store for this call: the installed backend, else the @@ -119,19 +123,45 @@ impl Docs { }) } + /// Whether the one-time import of the legacy `integrations/` file + /// finished in this scope. + pub(super) fn legacy_imported(&self, name: &str) -> Result { + let id = name.to_string(); + let found = self + .0 + .run(|docs| async move { docs.get(IMPORTED, &id).await })?; + Ok(found.is_some()) + } + /// Stores `value` under `name` only when no value is stored there yet - /// (the one-time import of a pre-documents state file). Returns whether it - /// wrote. + /// (the one-time import of a pre-documents state file), then records the + /// import in a marker document so it never runs again. The legacy file is + /// not modified. Returns whether the value was written. pub(super) fn import_if_absent(&self, name: &str, value: serde_json::Value) -> Result { let id = name.to_string(); let doc = json!({ "value": value }); log::debug!("[composio:store] document import id={id}"); self.0.run(|docs| async move { - match docs.put(STATE, &id, doc, Precondition::Absent).await { - Ok(_) => Ok(true), - Err(error) if error.kind() == ErrorKind::Conflict => Ok(false), - Err(error) => Err(error), + let wrote = match docs.put(STATE, &id, doc, Precondition::Absent).await { + Ok(_) => true, + Err(error) if error.kind() == ErrorKind::Conflict => false, + Err(error) => return Err(error), + }; + match docs + .put( + IMPORTED, + &id, + json!({ "imported": true }), + Precondition::Absent, + ) + .await + { + Ok(_) => {} + // Another core finished first. + Err(error) if error.kind() == ErrorKind::Conflict => {} + Err(error) => return Err(error), } + Ok(wrote) }) } diff --git a/crates/openhuman-core/src/integrations/composio/file_store_tests.rs b/crates/openhuman-core/src/integrations/composio/file_store_tests.rs index 3e521bf3732..b72e50dda8d 100644 --- a/crates/openhuman-core/src/integrations/composio/file_store_tests.rs +++ b/crates/openhuman-core/src/integrations/composio/file_store_tests.rs @@ -46,7 +46,7 @@ async fn update_applies_and_persists_a_change() { } #[tokio::test] -async fn a_legacy_state_file_is_imported_once_and_retired() { +async fn a_legacy_state_file_is_imported_once_and_left_byte_identical() { let dir = tempfile::tempdir().unwrap(); let config = config_in(&dir); let legacy_dir = dir.path().join("integrations"); @@ -55,8 +55,12 @@ async fn a_legacy_state_file_is_imported_once_and_retired() { let loaded: BTreeMap = load(&config, USER_SCOPES_FILE).await.unwrap(); assert_eq!(loaded["gmail"], 7); - assert!(!legacy_dir.join(USER_SCOPES_FILE).exists()); - assert!(legacy_dir + // The legacy file is read only: same name, same bytes, no `.migrated`. + assert_eq!( + std::fs::read(legacy_dir.join(USER_SCOPES_FILE)).unwrap(), + br#"{"gmail":7}"# + ); + assert!(!legacy_dir .join(format!("{USER_SCOPES_FILE}.migrated")) .exists()); @@ -73,4 +77,8 @@ async fn a_legacy_state_file_is_imported_once_and_retired() { .unwrap(); let again: BTreeMap = load(&config, USER_SCOPES_FILE).await.unwrap(); assert_eq!(again["gmail"], 9); + assert_eq!( + std::fs::read(legacy_dir.join(USER_SCOPES_FILE)).unwrap(), + br#"{"gmail":7}"# + ); } diff --git a/crates/openhuman-core/src/storage/fenced_backend.rs b/crates/openhuman-core/src/storage/fenced_backend.rs index 66be4fcac44..0cdc8055476 100644 --- a/crates/openhuman-core/src/storage/fenced_backend.rs +++ b/crates/openhuman-core/src/storage/fenced_backend.rs @@ -358,6 +358,22 @@ impl StreamStore for FencedStreams { ) } + /// Forwarded (and fenced): without it the trait default answers + /// `Unsupported(ConditionalStreamAppend)` even on a driver that has it, + /// which refuses every conversation append. + async fn append_batch_if_offset( + &self, + stream: &str, + expected_next_offset: u64, + values: Vec, + ) -> Result { + fenced_write!( + self, + "streams.append_batch_if_offset", + streams.append_batch_if_offset(stream, expected_next_offset, values) + ) + } + async fn read_window(&self, stream: &str, from: u64, limit: usize) -> Result> { self.inner.read_window(stream, from, limit).await } diff --git a/crates/openhuman-core/src/storage/fenced_backend_tests.rs b/crates/openhuman-core/src/storage/fenced_backend_tests.rs index 150c22930e0..12a1a28bb01 100644 --- a/crates/openhuman-core/src/storage/fenced_backend_tests.rs +++ b/crates/openhuman-core/src/storage/fenced_backend_tests.rs @@ -544,3 +544,42 @@ async fn expiring_blob_writes_are_fenced() { .unwrap() .is_none()); } + +/// The wrapper must pass a conditional append through to a driver that has it: +/// the trait default answers `Unsupported(ConditionalStreamAppend)`, which +/// refused every conversation append on the default (SQLite) backend. +async fn conditional_append_passes_through(inner: Arc) { + let wrapped = wrap(inner); + let scoped = wrapped.for_scope(&Scope::local()).unwrap(); + let streams = scoped.streams(); + assert_eq!( + streams + .append_batch_if_offset("log", 0, vec![json!(1)]) + .await + .unwrap(), + 0 + ); + let stale = streams + .append_batch_if_offset("log", 0, vec![json!(2)]) + .await + .unwrap_err(); + assert_eq!(stale.kind(), tinystoragedrivers::ErrorKind::Conflict); + assert_eq!(streams.len("log").await.unwrap(), 1); +} + +#[tokio::test] +async fn fenced_memory_backend_passes_conditional_appends_through() { + conditional_append_passes_through(shared()).await; +} + +#[cfg(feature = "storage-sqlite")] +#[tokio::test] +async fn fenced_sqlite_backend_passes_conditional_appends_through() { + let dir = tempfile::tempdir().unwrap(); + let url = format!("sqlite:{}", dir.path().join("cond.db").display()); + let backend = + tinystoragedrivers::open(&tinystoragedrivers::StorageConfig::parse(&url).unwrap()) + .await + .unwrap(); + conditional_append_passes_through(backend).await; +} diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs index e9c572f504d..ce186221106 100644 --- a/crates/openhuman-core/src/storage/mod.rs +++ b/crates/openhuman-core/src/storage/mod.rs @@ -50,7 +50,8 @@ use std::sync::{Arc, LazyLock, Mutex, OnceLock, RwLock}; /// foreign file, and its `run_blocking` hands the closure a `rusqlite` /// connection anyway. The core has no `rusqlite` dependency of its own, /// tinystoragedrivers owns the one copy; the re-export goes when the importers -/// do (the one-time legacy import window closes). +/// do (the one-time legacy import window closes). Re-checked for #7768: both +/// consumers are still present, so it stays. pub use tinystoragedrivers::sqlite::rusqlite; pub use tinystoragedrivers::{ Blocking, CollectionSpec, DocumentStore, DocumentStoreExt, MemoryStorage, Precondition, Scope, @@ -338,6 +339,16 @@ pub fn scope_for_profile(profile_id: &str) -> Scope { .unwrap_or_else(|_| unreachable!("a sha256 hex scope is always valid")) } +/// A stable key for a legacy import source (a workspace directory), for the +/// marker document and once-per-process key of an importer. Two workspaces +/// that share one explicit backend and scope must each import their own legacy +/// files, so a marker names its source. +pub fn source_key(source: &std::path::Path) -> String { + use sha2::Digest; + let digest = sha2::Sha256::digest(source.to_string_lossy().as_bytes()); + hex::encode(&digest[..8]) +} + #[cfg(test)] #[path = "mod_tests.rs"] mod tests; diff --git a/crates/openhuman-core/src/threads/mod.rs b/crates/openhuman-core/src/threads/mod.rs index 34b62dd9292..78fb738641f 100644 --- a/crates/openhuman-core/src/threads/mod.rs +++ b/crates/openhuman-core/src/threads/mod.rs @@ -1,7 +1,7 @@ //! Conversation thread and message management. //! //! Thread lifecycle (create, list, delete, purge) and per-thread message -//! CRUD. Storage is [`store`] (JSONL files over `tinyagents_session::threads`); +//! CRUD. Storage is [`store`] (the storage backend, through `tinyagents_session::threads`); //! this module owns the RPC surface and controller registry. pub mod error; diff --git a/crates/openhuman-core/src/threads/store/backend.rs b/crates/openhuman-core/src/threads/store/backend.rs new file mode 100644 index 00000000000..0391254ea81 --- /dev/null +++ b/crates/openhuman-core/src/threads/store/backend.rs @@ -0,0 +1,398 @@ +//! Chat threads on the storage backend: the workspace-keyed, synchronous +//! facade the host's callers use, over TinyAgents' +//! [`DriverConversationStore`] (the JSONL file store is gone upstream). +//! +//! Every call resolves the process's storage backend for the workspace +//! ([`crate::storage::local::backend_for`]: the installed backend, else the +//! workspace default) and binds it to the **conversation scope**: the calling +//! tenant's profile scope when there is one, else `local`. Conversations span +//! agents, so the acting agent's scope is deliberately not used: a thread a +//! sub-agent wrote must stay visible to the UI that lists it. +//! +//! The API is the one the file store had (`String` errors, owned arguments, +//! free functions beside [`ConversationStore`]) so the ~50 host callers did not +//! change. Each call blocks on the storage bridge ([`crate::storage::block_on`]); +//! request paths already go through [`super::blocking`], which keeps that off +//! the async workers. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +pub use tinyagents_session::threads::ConversationPurgeStats; +use tinyagents_session::threads::{ + ConversationError, ConversationPersistence, ConversationResult, ConversationThread, + CreateConversationThread, CrossThreadHit, DriverConversationStore, ThreadMessage, + ThreadMessagePatch, +}; + +use crate::storage::documents::Repo; +use crate::storage::{self, Scope, StorageError}; +use tinystoragedrivers::ErrorKind; + +const DOMAIN: &str = "threads::store"; +const LOG_PREFIX: &str = "[conversations]"; + +/// The storage scope conversations live under: the calling tenant's profile, +/// else `local`; refused in SaaS mode with no profile. +fn conversation_scope() -> Result { + let tenant = crate::core::runtime::current_tenant().map_err(|e| e.to_string())?; + storage::scope_from( + tenant.profile.as_deref(), + None, + crate::core::runtime::mode::is_saas(), + ) + .map_err(|e| e.to_string()) +} + +/// What an operation addresses, to word a not-found error as the file store did. +#[derive(Clone, Copy)] +struct Subject<'a> { + thread: Option<&'a str>, + message: Option<&'a str>, +} + +const NONE: Subject<'static> = Subject { + thread: None, + message: None, +}; + +const fn thread(id: &str) -> Subject<'_> { + Subject { + thread: Some(id), + message: None, + } +} + +const fn message<'a>(thread: &'a str, message: &'a str) -> Subject<'a> { + Subject { + thread: Some(thread), + message: Some(message), + } +} + +/// The error text callers match on: a missing thread or message reads +/// `thread not found` / `message not found in thread ` (the +/// wording the file store used and `threads::ops` maps to a typed error); +/// anything else is the driver's own message. +fn describe(error: &ConversationError, subject: Subject<'_>) -> String { + if let ConversationError::Driver(inner) = error { + if inner.kind() == ErrorKind::NotFound { + let text = inner.to_string(); + match (subject.thread, subject.message) { + (Some(t), Some(m)) if text.contains("message") => { + return format!("message {m} not found in thread {t}") + } + (Some(t), _) => return format!("thread {t} not found"), + _ => {} + } + } + } + error.to_string() +} + +/// A conversation store bound to one workspace's storage backend. +#[derive(Clone)] +pub struct ConversationStore { + workspace_dir: PathBuf, +} + +impl ConversationStore { + /// A store for `workspace_dir`. Nothing is opened until the first call. + pub fn new(workspace_dir: PathBuf) -> Self { + Self { workspace_dir } + } + + /// The driver store for this call, its schema declared once per backend + /// and scope, and the legacy JSONL conversations imported once. + fn open(&self) -> Result { + let scope = conversation_scope()?; + let backend = storage::local::backend_for(&self.workspace_dir) + .map_err(|e| format!("{LOG_PREFIX} open the storage backend: {e:#}"))?; + let scoped = backend + .for_scope(&scope) + .map_err(|e| format!("{LOG_PREFIX} open the storage scope: {e}"))?; + let repo = Repo::on(&backend, &scope, DOMAIN, super::legacy_import::collections) + .map_err(|e| format!("{LOG_PREFIX} {e:#}"))?; + let store = DriverConversationStore::new(scoped.clone()); + repo.once_per_backend("schema", || { + let docs = Arc::clone(scoped.documents()); + storage::block_on(async move { + DriverConversationStore::prepare_schema(docs.as_ref()).await?; + DriverConversationStore::prepare_append_schema(docs.as_ref()).await + }) + .map_err(|e| anyhow::anyhow!("declare the conversation schema: {e}")) + }) + .map_err(|e| format!("{LOG_PREFIX} {e:#}"))?; + super::legacy_import::import_once(&self.workspace_dir, &repo, &store) + .map_err(|e| format!("{LOG_PREFIX} legacy import: {e:#}"))?; + Ok(store) + } + + fn run( + &self, + operation: &'static str, + subject: Subject<'_>, + f: F, + ) -> Result + where + F: FnOnce(DriverConversationStore) -> Fut, + Fut: std::future::Future> + Send + 'static, + T: Send + 'static, + { + let store = self.open()?; + let future = f(store); + let outcome = storage::block_on(async move { Ok(future.await) }) + .map_err(|e: StorageError| format!("{LOG_PREFIX} {operation}: {e}"))?; + outcome.map_err(|e| { + log::debug!("{LOG_PREFIX} {operation} failed: {e}"); + describe(&e, subject) + }) + } + + /// Creates the thread or returns the existing one. + pub fn ensure_thread( + &self, + request: CreateConversationThread, + ) -> Result { + self.run("ensure_thread", NONE, |s| async move { + s.ensure_thread(request).await + }) + } + + /// Every live thread, newest first. + pub fn list_threads(&self) -> Result, String> { + self.run( + "list_threads", + NONE, + |s| async move { s.list_threads().await }, + ) + } + + /// A thread's messages in append order; empty for an unknown thread. + pub fn get_messages(&self, thread_id: &str) -> Result, String> { + let id = thread_id.to_string(); + self.run("get_messages", thread(thread_id), |s| async move { + if s.get_thread(&id).await?.is_none() { + return Ok(Vec::new()); + } + s.get_messages(&id).await + }) + } + + /// Cross-thread substring search over message content. + pub fn search_cross_thread_messages( + &self, + query: &str, + limit: usize, + exclude_thread_id: Option<&str>, + ) -> Result, String> { + let query = query.to_string(); + let exclude = exclude_thread_id.map(str::to_string); + self.run("search_cross_thread_messages", NONE, |s| async move { + s.search_cross_thread_messages(&query, limit, exclude.as_deref()) + .await + }) + } + + /// Appends a message to a thread. + pub fn append_message( + &self, + thread_id: &str, + message: ThreadMessage, + ) -> Result { + let id = thread_id.to_string(); + self.run("append_message", thread(thread_id), |s| async move { + s.append_message(&id, message).await + }) + } + + /// Renames a thread. + pub fn update_thread_title( + &self, + thread_id: &str, + title: &str, + updated_at: &str, + ) -> Result { + let (id, title, at) = ( + thread_id.to_string(), + title.to_string(), + updated_at.to_string(), + ); + self.run("update_thread_title", thread(thread_id), |s| async move { + s.update_thread_title(&id, &title, &at).await + }) + } + + /// Sets a thread's working directory. + pub fn update_thread_working_dir( + &self, + thread_id: &str, + working_dir: Option, + updated_at: &str, + ) -> Result { + let (id, at) = (thread_id.to_string(), updated_at.to_string()); + self.run( + "update_thread_working_dir", + thread(thread_id), + |s| async move { s.update_thread_working_dir(&id, working_dir, &at).await }, + ) + } + + /// Replaces a thread's labels. + pub fn update_thread_labels( + &self, + thread_id: &str, + labels: Vec, + updated_at: &str, + ) -> Result { + let (id, at) = (thread_id.to_string(), updated_at.to_string()); + self.run("update_thread_labels", thread(thread_id), |s| async move { + s.update_thread_labels(&id, labels, &at).await + }) + } + + /// Patches one message. + pub fn update_message( + &self, + thread_id: &str, + message_id: &str, + patch: ThreadMessagePatch, + ) -> Result { + let (id, mid) = (thread_id.to_string(), message_id.to_string()); + self.run( + "update_message", + message(thread_id, message_id), + |s| async move { s.update_message(&id, &mid, patch).await }, + ) + } + + /// Removes a message and everything after it; `None` for an unknown id. + pub fn delete_messages_from( + &self, + thread_id: &str, + message_id: &str, + ) -> Result, String> { + let (id, mid) = (thread_id.to_string(), message_id.to_string()); + self.run( + "delete_messages_from", + message(thread_id, message_id), + |s| async move { s.delete_messages_from(&id, &mid).await }, + ) + } + + /// Soft-deletes a thread; whether it existed. + pub fn delete_thread(&self, thread_id: &str, deleted_at: &str) -> Result { + let (id, at) = (thread_id.to_string(), deleted_at.to_string()); + self.run("delete_thread", thread(thread_id), |s| async move { + s.delete_thread(&id, &at).await + }) + } + + /// Removes every thread. + pub fn purge_threads(&self) -> Result { + self.run( + "purge_threads", + NONE, + |s| async move { s.purge_threads().await }, + ) + } +} + +fn store(workspace_dir: PathBuf) -> ConversationStore { + ConversationStore::new(workspace_dir) +} + +/// Free-function shim around [`ConversationStore::ensure_thread`]. +pub fn ensure_thread( + workspace_dir: PathBuf, + request: CreateConversationThread, +) -> Result { + store(workspace_dir).ensure_thread(request) +} + +/// Free-function shim around [`ConversationStore::list_threads`]. +pub fn list_threads(workspace_dir: PathBuf) -> Result, String> { + store(workspace_dir).list_threads() +} + +/// Free-function shim around [`ConversationStore::get_messages`]. +pub fn get_messages(workspace_dir: PathBuf, thread_id: &str) -> Result, String> { + store(workspace_dir).get_messages(thread_id) +} + +/// Free-function shim around [`ConversationStore::append_message`]. +pub fn append_message( + workspace_dir: PathBuf, + thread_id: &str, + message: ThreadMessage, +) -> Result { + store(workspace_dir).append_message(thread_id, message) +} + +/// Free-function shim around [`ConversationStore::update_thread_title`]. +pub fn update_thread_title( + workspace_dir: PathBuf, + thread_id: &str, + title: &str, + updated_at: &str, +) -> Result { + store(workspace_dir).update_thread_title(thread_id, title, updated_at) +} + +/// Free-function shim around [`ConversationStore::update_thread_labels`]. +pub fn update_thread_labels( + workspace_dir: PathBuf, + thread_id: &str, + labels: Vec, + updated_at: &str, +) -> Result { + store(workspace_dir).update_thread_labels(thread_id, labels, updated_at) +} + +/// Free-function shim around [`ConversationStore::update_thread_working_dir`]. +pub fn update_thread_working_dir( + workspace_dir: PathBuf, + thread_id: &str, + working_dir: Option, + updated_at: &str, +) -> Result { + store(workspace_dir).update_thread_working_dir(thread_id, working_dir, updated_at) +} + +/// Free-function shim around [`ConversationStore::update_message`]. +pub fn update_message( + workspace_dir: PathBuf, + thread_id: &str, + message_id: &str, + patch: ThreadMessagePatch, +) -> Result { + store(workspace_dir).update_message(thread_id, message_id, patch) +} + +/// Free-function shim around [`ConversationStore::delete_messages_from`]. +pub fn delete_messages_from( + workspace_dir: PathBuf, + thread_id: &str, + message_id: &str, +) -> Result, String> { + store(workspace_dir).delete_messages_from(thread_id, message_id) +} + +/// Free-function shim around [`ConversationStore::purge_threads`]. +pub fn purge_threads(workspace_dir: PathBuf) -> Result { + store(workspace_dir).purge_threads() +} + +/// Free-function shim around [`ConversationStore::delete_thread`]. +pub fn delete_thread( + workspace_dir: PathBuf, + thread_id: &str, + deleted_at: &str, +) -> Result { + store(workspace_dir).delete_thread(thread_id, deleted_at) +} + +/// Whether `path` is the legacy conversation root of `workspace_dir`. +pub(super) fn legacy_root(workspace_dir: &Path) -> PathBuf { + workspace_dir.join("memory").join("conversations") +} diff --git a/crates/openhuman-core/src/threads/store/blocking.rs b/crates/openhuman-core/src/threads/store/blocking.rs index f1a798694c6..673f6674cef 100644 --- a/crates/openhuman-core/src/threads/store/blocking.rs +++ b/crates/openhuman-core/src/threads/store/blocking.rs @@ -69,7 +69,7 @@ where F: FnOnce() -> Result + Send + 'static, T: Send + 'static, { - match tokio::task::spawn_blocking(f).await { + match crate::core::runtime::spawn_blocking_scoped(f).await { Ok(result) => result, Err(error) => { tracing::warn!( diff --git a/crates/openhuman-core/src/threads/store/legacy_import.rs b/crates/openhuman-core/src/threads/store/legacy_import.rs new file mode 100644 index 00000000000..f6181efd91a --- /dev/null +++ b/crates/openhuman-core/src/threads/store/legacy_import.rs @@ -0,0 +1,233 @@ +//! One-time, read-only import of the retired JSONL conversation files. +//! +//! Older installs kept chat threads as `threads.jsonl` (an append-only +//! upsert/delete log) plus one `threads/.jsonl` per thread +//! under `/memory/conversations/`. The file store is gone, so the +//! first use of a workspace's backend folds those files into it. The files are +//! only ever **read**: a marker document in the backend records completion, and +//! the legacy tree stays byte-identical (`storage::import` follows the same +//! rule for the per-domain databases). + +use std::collections::BTreeMap; +use std::path::Path; + +use anyhow::{anyhow, Context, Result}; +use serde::Deserialize; +use sha2::{Digest, Sha256}; +use tinyagents_session::threads::{ + ConversationPersistence, CreateConversationThread, DriverConversationStore, ThreadMessage, +}; +use tinystoragedrivers::{CollectionSpec, ErrorKind, Precondition}; + +use crate::storage::documents::Repo; + +const MARKERS: &str = "threads_import"; +const MARKER_PREFIX: &str = "legacy-conversations"; +const LOG_PREFIX: &str = "[conversations:import]"; + +/// The collection that holds the completion marker. +pub(super) fn collections() -> Vec { + vec![CollectionSpec::new(MARKERS)] +} + +/// The `threads.jsonl` records the importer needs; the stats records the old +/// store derived (`message_appended`, `stats`) are recomputed from messages. +#[derive(Deserialize)] +#[serde(tag = "op", rename_all = "snake_case")] +enum LogEntry { + Upsert { + thread_id: String, + title: String, + created_at: String, + #[serde(default)] + parent_thread_id: Option, + #[serde(default)] + labels: Option>, + #[serde(default)] + personality_id: Option, + #[serde(default)] + working_dir: Option, + }, + Delete { + thread_id: String, + }, + #[serde(other)] + Derived, +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// `threads/.jsonl` for a thread: hex of short ids, a SHA-256 digest for +/// longer ones, as the file store named them. +fn messages_path(root: &Path, thread_id: &str) -> std::path::PathBuf { + let stem = if thread_id.len() <= 124 { + hex(thread_id.as_bytes()) + } else { + format!("sha256-{}", hex(&Sha256::digest(thread_id.as_bytes()))) + }; + root.join("threads").join(format!("{stem}.jsonl")) +} + +/// Parseable lines of a JSONL file; a torn or foreign line is skipped, as the +/// file store skipped it. A missing file is empty. +fn read_jsonl Deserialize<'de>>(path: &Path) -> Result> { + let text = match std::fs::read_to_string(path) { + Ok(text) => text, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(anyhow!("read {}: {error}", path.display())), + }; + Ok(text + .lines() + .filter_map(|line| serde_json::from_str::(line.trim()).ok()) + .collect()) +} + +/// Folds the upsert/delete log into the live threads, in first-seen order. +fn fold_threads(root: &Path) -> Result> { + let mut live: BTreeMap = BTreeMap::new(); + for entry in read_jsonl::(&root.join("threads.jsonl"))? { + match entry { + LogEntry::Upsert { + thread_id, + title, + created_at, + parent_thread_id, + labels, + personality_id, + working_dir, + } => match live.get_mut(&thread_id) { + Some(existing) => { + existing.title = title; + existing.parent_thread_id = + parent_thread_id.or(existing.parent_thread_id.take()); + existing.labels = labels.or(existing.labels.take()); + existing.personality_id = personality_id.or(existing.personality_id.take()); + existing.working_dir = match working_dir { + Some(dir) if dir.is_empty() => None, + Some(dir) => Some(dir), + None => existing.working_dir.take(), + }; + } + None => { + live.insert( + thread_id.clone(), + CreateConversationThread { + id: thread_id, + title, + created_at, + parent_thread_id, + labels, + personality_id, + working_dir: working_dir.filter(|dir| !dir.is_empty()), + }, + ); + } + }, + LogEntry::Delete { thread_id } => { + live.remove(&thread_id); + } + LogEntry::Derived => {} + } + } + Ok(live.into_values().collect()) +} + +/// Imports the workspace's legacy conversations into `store` once per backend +/// and scope. `repo` carries the marker collection ([`collections`]). +/// +/// # Errors +/// +/// A legacy file that cannot be read, or a storage failure; the next call +/// retries (a thread or message already imported is skipped). +pub(super) fn import_once( + workspace_dir: &Path, + repo: &Repo, + store: &DriverConversationStore, +) -> Result<()> { + let root = super::backend::legacy_root(workspace_dir); + let source = crate::storage::source_key(&root); + let marker_id = format!("{MARKER_PREFIX}:{source}"); + repo.once_per_backend(&format!("legacy-import:{source}"), || { + match std::fs::metadata(root.join("threads.jsonl")) { + Ok(meta) if meta.is_file() => {} + Ok(_) => return Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(error) => { + return Err(anyhow::Error::new(error).context("stat the legacy threads log")) + } + } + let marked = repo.run(|docs| { + let marker_id = marker_id.clone(); + async move { docs.get(MARKERS, &marker_id).await } + })?; + if marked.is_some() { + log::debug!("{LOG_PREFIX} already imported"); + return Ok(()); + } + let threads = fold_threads(&root)?; + let mut messages = Vec::with_capacity(threads.len()); + for thread in &threads { + messages.push(read_jsonl::(&messages_path( + &root, &thread.id, + ))?); + } + let store = store.clone(); + let (threads_n, messages_n) = + crate::storage::block_on_anyhow(async move { copy(&store, threads, messages).await })?; + repo.run(|docs| async move { + match docs + .put( + MARKERS, + &marker_id, + serde_json::json!({ "threads": threads_n, "messages": messages_n }), + Precondition::Absent, + ) + .await + { + Ok(_) => Ok(()), + Err(error) if error.kind() == ErrorKind::Conflict => Ok(()), + Err(error) => Err(error), + } + })?; + log::info!("{LOG_PREFIX} imported threads={threads_n} messages={messages_n}"); + Ok(()) + }) +} + +/// Writes the folded threads and their messages; one already present (an +/// earlier interrupted import) is not duplicated. +async fn copy( + store: &DriverConversationStore, + threads: Vec, + messages: Vec>, +) -> Result<(usize, usize)> { + let mut message_total = 0; + for (thread, thread_messages) in threads.into_iter().zip(messages) { + let id = thread.id.clone(); + store + .ensure_thread(thread) + .await + .with_context(|| format!("create thread {id}"))?; + let present: std::collections::HashSet = store + .get_messages(&id) + .await + .with_context(|| format!("read thread {id}"))? + .into_iter() + .map(|m| m.id) + .collect(); + for message in thread_messages { + if present.contains(&message.id) { + continue; + } + store + .append_message(&id, message) + .await + .with_context(|| format!("append to thread {id}"))?; + message_total += 1; + } + } + let thread_total = store.list_threads().await?.len(); + Ok((thread_total, message_total)) +} diff --git a/crates/openhuman-core/src/threads/store/legacy_import_tests.rs b/crates/openhuman-core/src/threads/store/legacy_import_tests.rs new file mode 100644 index 00000000000..6a580ade1f8 --- /dev/null +++ b/crates/openhuman-core/src/threads/store/legacy_import_tests.rs @@ -0,0 +1,180 @@ +use super::*; +use std::path::Path; + +fn hex(id: &str) -> String { + id.bytes().map(|b| format!("{b:02x}")).collect() +} + +fn write_legacy(root: &Path, log: &[&str], threads: &[(&str, &[&str])]) { + std::fs::create_dir_all(root.join("threads")).unwrap(); + std::fs::write(root.join("threads.jsonl"), log.join("\n") + "\n").unwrap(); + for (id, lines) in threads { + std::fs::write( + root.join("threads").join(format!("{}.jsonl", hex(id))), + lines.join("\n") + "\n", + ) + .unwrap(); + } +} + +fn message(id: &str, content: &str, at: &str) -> String { + format!( + r#"{{"id":"{id}","content":"{content}","type":"text","extraMetadata":null,"sender":"user","createdAt":"{at}"}}"# + ) +} + +fn upsert(id: &str, title: &str) -> String { + format!( + r#"{{"op":"upsert","thread_id":"{id}","title":"{title}","created_at":"2026-01-01T00:00:00Z","updated_at":"2026-01-01T00:00:00Z"}}"# + ) +} + +fn snapshot(root: &Path) -> Vec<(std::path::PathBuf, Vec)> { + let mut files = Vec::new(); + let mut stack = vec![root.to_path_buf()]; + while let Some(dir) = stack.pop() { + for entry in std::fs::read_dir(dir).unwrap() { + let path = entry.unwrap().path(); + if path.is_dir() { + stack.push(path); + } else { + files.push((path.clone(), std::fs::read(&path).unwrap())); + } + } + } + files.sort(); + files +} + +#[test] +fn legacy_jsonl_conversations_are_imported_and_the_files_stay_byte_identical() { + let dir = tempfile::tempdir().unwrap(); + let root = backend::legacy_root(dir.path()); + let (m1, m2) = ( + message("m1", "hello", "2026-01-01T00:00:01Z"), + message("m2", "world", "2026-01-01T00:00:02Z"), + ); + let log = [ + upsert("t1", "First"), + upsert("t2", "Doomed"), + r#"{"op":"delete","thread_id":"t2","deleted_at":"2026-01-02T00:00:00Z"}"#.to_string(), + upsert("t1", "Renamed"), + r#"{"op":"message_appended","thread_id":"t1","last_message_at":"2026-01-01T00:00:02Z"}"# + .to_string(), + ]; + let log_refs: Vec<&str> = log.iter().map(String::as_str).collect(); + write_legacy(&root, &log_refs, &[("t1", &[m1.as_str(), m2.as_str()])]); + let before = snapshot(&root); + + let store = ConversationStore::new(dir.path().to_path_buf()); + let threads = store.list_threads().unwrap(); + assert_eq!(threads.len(), 1, "the deleted thread is not imported"); + assert_eq!(threads[0].id, "t1"); + assert_eq!(threads[0].title, "Renamed"); + let messages = store.get_messages("t1").unwrap(); + let ids: Vec<&str> = messages.iter().map(|m| m.id.as_str()).collect(); + assert_eq!(ids, ["m1", "m2"]); + assert_eq!(messages[1].content, "world"); + + assert_eq!(snapshot(&root), before, "legacy files must not change"); +} + +#[test] +fn legacy_import_runs_once_even_when_the_files_change_afterwards() { + let dir = tempfile::tempdir().unwrap(); + let root = backend::legacy_root(dir.path()); + let m1 = message("m1", "hello", "2026-01-01T00:00:01Z"); + write_legacy(&root, &[&upsert("t1", "First")], &[("t1", &[m1.as_str()])]); + + let store = ConversationStore::new(dir.path().to_path_buf()); + assert_eq!(store.list_threads().unwrap().len(), 1); + + // A new process on the same workspace: the marker, not the process cache, + // stops the second import. + crate::storage::local::forget(dir.path()); + write_legacy( + &root, + &[&upsert("t1", "First"), &upsert("t9", "Late")], + &[("t1", &[m1.as_str()])], + ); + let after = ConversationStore::new(dir.path().to_path_buf()); + let threads = after.list_threads().unwrap(); + assert_eq!(threads.len(), 1); + assert_eq!(after.get_messages("t1").unwrap().len(), 1); +} + +#[test] +fn a_workspace_without_legacy_files_starts_empty() { + let dir = tempfile::tempdir().unwrap(); + let store = ConversationStore::new(dir.path().to_path_buf()); + assert!(store.list_threads().unwrap().is_empty()); + assert!(store.get_messages("nope").unwrap().is_empty()); + assert!(!backend::legacy_root(dir.path()).exists()); +} + +#[test] +fn threads_round_trip_through_the_backend_store() { + let dir = tempfile::tempdir().unwrap(); + let store = ConversationStore::new(dir.path().to_path_buf()); + store + .ensure_thread(CreateConversationThread { + id: "t".into(), + title: "T".into(), + created_at: "2026-01-01T00:00:00Z".into(), + parent_thread_id: None, + labels: None, + personality_id: None, + working_dir: None, + }) + .unwrap(); + let appended = store + .append_message( + "t", + ConversationMessage { + id: "a".into(), + content: "findable needle".into(), + message_type: "text".into(), + extra_metadata: serde_json::Value::Null, + sender: "user".into(), + created_at: "2026-01-01T00:00:01Z".into(), + }, + ) + .unwrap(); + assert_eq!(appended.id, "a"); + assert_eq!(store.get_messages("t").unwrap().len(), 1); + assert!(!store + .search_cross_thread_messages("needle", 5, None) + .unwrap() + .is_empty()); + assert!(store.delete_thread("t", "2026-01-02T00:00:00Z").unwrap()); + assert!(store.list_threads().unwrap().is_empty()); +} + +#[test] +fn legacy_import_marker_is_per_source_workspace() { + let a = crate::storage::source_key(Path::new("/ws/a")); + let b = crate::storage::source_key(Path::new("/ws/b")); + assert_ne!(a, b); + assert_eq!(a, crate::storage::source_key(Path::new("/ws/a"))); +} + +#[cfg(unix)] +#[test] +fn an_unreadable_legacy_log_is_an_error_not_an_empty_store() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().unwrap(); + let root = backend::legacy_root(dir.path()); + std::fs::create_dir_all(&root).unwrap(); + std::fs::write(root.join("threads.jsonl"), "").unwrap(); + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o000)).unwrap(); + // Root bypasses permission bits; only assert when the stat truly fails. + let blocked = std::fs::metadata(root.join("threads.jsonl")).is_err(); + let result = ConversationStore::new(dir.path().to_path_buf()).list_threads(); + std::fs::set_permissions(&root, std::fs::Permissions::from_mode(0o755)).unwrap(); + if blocked { + assert!( + result.is_err(), + "a stat failure must surface so a retry can import" + ); + } +} diff --git a/crates/openhuman-core/src/threads/store/mod.rs b/crates/openhuman-core/src/threads/store/mod.rs index 1f11bf9cc08..a122b604c27 100644 --- a/crates/openhuman-core/src/threads/store/mod.rs +++ b/crates/openhuman-core/src/threads/store/mod.rs @@ -1,45 +1,56 @@ -//! Workspace-backed chat thread/message storage (`threads::store`). +//! Backend-backed chat thread/message storage (`threads::store`). //! -//! Conversations are JSONL files under `/memory/conversations/`: -//! thread metadata as an append-only upsert/delete log in `threads.jsonl`, and -//! each thread's messages in a dedicated file under -//! `threads/.jsonl`. This is **transcript persistence** — the -//! raw records plus a trigram/CJK-bigram index for cross-thread substring -//! search over them. It is not memory: memory v2 (`crate::memory`) only -//! *ingests* committed turns through its own bus subscriber. +//! Conversations live in the process's storage backend +//! ([`crate::storage`]), through TinyAgents' `DriverConversationStore`: thread +//! metadata as one document per thread and each thread's messages as an +//! append stream. This is **transcript persistence**: the raw records plus +//! cross-thread substring search over them. It is not memory: memory v2 +//! (`crate::memory`) only *ingests* committed turns through its own bus +//! subscriber. //! -//! The store itself (on-disk format, locking, warm index cache, CRUD and -//! search) is `tinyagents_session::threads`, re-exported below so callers -//! name `crate::threads::store::{...}`. The two message types keep their -//! historical host spelling (`ConversationMessage`, `ConversationMessagePatch`) -//! as aliases of `ThreadMessage` / `ThreadMessagePatch`; type names never -//! reach the disk. What stays here is host wiring: +//! TinyAgents removed its path-bound JSONL store, so the host keeps only: //! +//! - [`backend`] - the workspace-keyed synchronous facade +//! (`ConversationStore::new(workspace_dir)` and the free functions) the ~50 +//! callers already use, resolving the backend and the conversation scope. +//! - `legacy_import` - the one-time, read-only import of the retired +//! `/memory/conversations` JSONL tree. A marker document records +//! completion; the files are never modified. //! - [`blocking`] - `spawn_blocking` wrappers. Every store entry point is -//! synchronous and can take `parking_lot` locks across fsync'd file IO, so an -//! `async fn` that calls one directly parks a tokio **worker** thread for the -//! whole wait. Request paths must use these (#5156). +//! synchronous and blocks on the storage bridge, so an `async fn` that calls +//! one directly parks a tokio **worker** thread. Request paths must use these +//! (#5156). //! - `bus` - the `core::bus` subscriber that mirrors inbound and processed //! channel turns into the store, so Slack/Telegram/... persist alongside the //! UI's own threads. //! -//! The on-disk format is unchanged: the root is still -//! `/memory/conversations`. +//! The two message types keep their historical host spelling +//! (`ConversationMessage`, `ConversationMessagePatch`) as aliases of +//! `ThreadMessage` / `ThreadMessagePatch`. +pub mod backend; pub mod blocking; mod bus; -use tinyagents_session::threads as store; +mod legacy_import; +#[cfg(test)] +#[path = "legacy_import_tests.rs"] +mod legacy_import_tests; + +use backend as store; +use tinyagents_session::threads as types; // Only the hosted-channel relay (`channels`) claims turns. #[cfg(feature = "channels")] pub(crate) use bus::claim_channel_turn; pub use bus::register_conversation_persistence_subscriber; pub use store::{ - append_message, delete_messages_from, delete_thread, ensure_thread, get_messages, - is_deterministic_message_id, list_threads, purge_threads, reply_run_id, run_reply_message_id, - update_message, update_thread_labels, update_thread_title, update_thread_working_dir, - ConversationPurgeStats, ConversationStore, ConversationThread, CreateConversationThread, - CrossThreadHit, ThreadMessage as ConversationMessage, + append_message, delete_messages_from, delete_thread, ensure_thread, get_messages, list_threads, + purge_threads, update_message, update_thread_labels, update_thread_title, + update_thread_working_dir, ConversationPurgeStats, ConversationStore, +}; +pub use types::{ + is_deterministic_message_id, reply_run_id, run_reply_message_id, ConversationThread, + CreateConversationThread, CrossThreadHit, ThreadMessage as ConversationMessage, ThreadMessagePatch as ConversationMessagePatch, }; diff --git a/vendor/tinyagents b/vendor/tinyagents index 869503c389e..fed928f7d7e 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 869503c389ed7ca915e58f15e1d7d6b459251b77 +Subproject commit fed928f7d7e3f26f6b525a872bfef1026f7a0678