From 05084b1d057c8813ceb7d4a46dc84115140047be Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:02:34 +0300 Subject: [PATCH 01/17] chore(vendor): bump tinyagents submodule Auto-committed-on: dragonfly --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index 869503c389e..80460be5f8e 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 869503c389ed7ca915e58f15e1d7d6b459251b77 +Subproject commit 80460be5f8e0daed81479dc0921c5040ddfd3bea From cc8c5784fd32323ef3312645f0747245be2828df Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:03:39 +0300 Subject: [PATCH 02/17] fix(storage): mark legacy imports with documents instead of renaming files Legacy session and Composio state files are now imported once and recorded with a marker document, so the original files are only read and stay byte-identical rather than being renamed to `.migrated`. The marker also short-circuits repeat imports, which previously relied on the rename to avoid re-reading the file. Auto-committed-on: dragonfly --- .../orchestration/subagent_sessions/store.rs | 16 ++++---- .../subagent_sessions/store_documents.rs | 36 ++++++++++++++++- .../src/integrations/composio/file_store.rs | 19 +++++---- .../composio/file_store_documents.rs | 39 +++++++++++++++---- 4 files changed, 88 insertions(+), 22 deletions(-) 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..d83e4ec8355 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,14 @@ 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().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 +31,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(&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..5a0058a2957 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,14 @@ 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"; 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 +89,35 @@ impl Docs { .collect() } + /// Whether the one-time legacy file import finished in this scope. + pub(super) fn legacy_imported(&self) -> Result { + let found = self + .0 + .run(|docs| async move { docs.get(IMPORTED, LEGACY_MARKER).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, sessions: &[DurableSubagentSession]) -> Result<()> { + self.save(sessions)?; + self.0.run(|docs| async move { + match docs + .put( + IMPORTED, + LEGACY_MARKER, + 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/integrations/composio/file_store.rs b/crates/openhuman-core/src/integrations/composio/file_store.rs index f3ce082391f..31b1f476753 100644 --- a/crates/openhuman-core/src/integrations/composio/file_store.rs +++ b/crates/openhuman-core/src/integrations/composio/file_store.rs @@ -20,14 +20,23 @@ 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 +58,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..7ac262a8476 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,40 @@ 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) }) } From 090834ae5a9c9e9fa03ff9d232bb0b899e9c6893 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:03:56 +0300 Subject: [PATCH 03/17] test: assert legacy files are left byte-identical after import Update the legacy-import tests for subagent sessions and Composio scopes to expect the original file to remain in place with unchanged bytes and no `.migrated` sibling, and to confirm the import marker alone prevents a second import. Auto-committed-on: dragonfly --- .../store_documents_tests.rs | 25 +++++++++++++------ .../integrations/composio/file_store_tests.rs | 14 ++++++++--- 2 files changed, 28 insertions(+), 11 deletions(-) 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..99f7752b4e7 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,30 @@ 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 session saved after + // the first import and then removed from the store is not resurrected. + store.save(&[]).unwrap(); + assert_eq!(store.load().unwrap().len(), 1); + assert_eq!( + std::fs::read_to_string(legacy.join("subagent_sessions.json")).unwrap(), + original + ); } 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}"# + ); } From 54215f92a7909360546af0896d8a991812e2acab Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:04:04 +0300 Subject: [PATCH 04/17] test(subagent_sessions): assert marker blocks reimport of changed legacy file The legacy import test now rewrites the legacy file after the first import and checks that loading still returns the originally imported session, proving the migration marker rather than the file contents prevents a second import. Auto-committed-on: dragonfly --- .../subagent_sessions/store_documents_tests.rs | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) 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 99f7752b4e7..fc25962d730 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 @@ -161,12 +161,11 @@ fn a_legacy_session_file_is_imported_once_and_left_byte_identical() { ); 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 session saved after - // the first import and then removed from the store is not resurrected. - store.save(&[]).unwrap(); - assert_eq!(store.load().unwrap().len(), 1); - assert_eq!( - std::fs::read_to_string(legacy.join("subagent_sessions.json")).unwrap(), - original - ); + // 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"); } From 0045ddb6deb83ad9f8c3e82aad90a3b3dbe292e4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:05:02 +0300 Subject: [PATCH 05/17] chore(threads): add store backend module Introduce a backend module for the threads store to hold the storage implementation behind a dedicated file. Auto-committed-on: dragonfly --- .../src/threads/store/backend.rs | 331 ++++++++++++++++++ 1 file changed, 331 insertions(+) create mode 100644 crates/openhuman-core/src/threads/store/backend.rs 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..e3cd12772d2 --- /dev/null +++ b/crates/openhuman-core/src/threads/store/backend.rs @@ -0,0 +1,331 @@ +//! 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; + +use tinyagents_session::threads::{ + ConversationPersistence, ConversationResult, ConversationThread, CreateConversationThread, + CrossThreadHit, DriverConversationStore, ThreadMessage, ThreadMessagePatch, +}; +pub use tinyagents_session::threads::ConversationPurgeStats; + +use crate::storage::documents::Repo; +use crate::storage::{self, Scope, StorageError}; + +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()) +} + +fn no_collections() -> Vec { + Vec::new() +} + +/// 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, no_collections) + .map_err(|e| format!("{LOG_PREFIX} {e:#}"))?; + 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:#}"))?; + let store = DriverConversationStore::new(scoped); + 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, 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}"); + e.to_string() + }) + } + + /// Creates the thread or returns the existing one. + pub fn ensure_thread( + &self, + request: CreateConversationThread, + ) -> Result { + self.run("ensure_thread", |s| async move { + s.ensure_thread(request).await + }) + } + + /// Every live thread, newest first. + pub fn list_threads(&self) -> Result, String> { + self.run("list_threads", |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", |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", |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", |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", |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", |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", |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", |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", |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", |s| async move { + s.delete_thread(&id, &at).await + }) + } + + /// Removes every thread. + pub fn purge_threads(&self) -> Result { + self.run("purge_threads", |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") +} From 9ec047286a19e9e9125c5c815f332b3243d6ea76 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:05:42 +0300 Subject: [PATCH 06/17] refactor(threads): source collection specs from legacy import module The conversation store now takes its collection specs from the legacy import module instead of a local empty helper, and the driver store is constructed before the schema declaration so it can be reused. This removes the duplicated no-op helper and keeps collection definitions in one place. Auto-committed-on: dragonfly --- .../src/threads/store/backend.rs | 8 +- .../src/threads/store/legacy_import.rs | 221 ++++++++++++++++++ 2 files changed, 223 insertions(+), 6 deletions(-) create mode 100644 crates/openhuman-core/src/threads/store/legacy_import.rs diff --git a/crates/openhuman-core/src/threads/store/backend.rs b/crates/openhuman-core/src/threads/store/backend.rs index e3cd12772d2..002b6ab0129 100644 --- a/crates/openhuman-core/src/threads/store/backend.rs +++ b/crates/openhuman-core/src/threads/store/backend.rs @@ -42,10 +42,6 @@ fn conversation_scope() -> Result { .map_err(|e| e.to_string()) } -fn no_collections() -> Vec { - Vec::new() -} - /// A conversation store bound to one workspace's storage backend. #[derive(Clone)] pub struct ConversationStore { @@ -67,8 +63,9 @@ impl ConversationStore { let scoped = backend .for_scope(&scope) .map_err(|e| format!("{LOG_PREFIX} open the storage scope: {e}"))?; - let repo = Repo::on(&backend, &scope, DOMAIN, no_collections) + 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 { @@ -78,7 +75,6 @@ impl ConversationStore { .map_err(|e| anyhow::anyhow!("declare the conversation schema: {e}")) }) .map_err(|e| format!("{LOG_PREFIX} {e:#}"))?; - let store = DriverConversationStore::new(scoped); super::legacy_import::import_once(&self.workspace_dir, &repo, &store) .map_err(|e| format!("{LOG_PREFIX} legacy import: {e:#}"))?; Ok(store) 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..a9f16d0a75a --- /dev/null +++ b/crates/openhuman-core/src/threads/store/legacy_import.rs @@ -0,0 +1,221 @@ +//! 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_ID: &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); + repo.once_per_backend("legacy-import", || { + if !root.join("threads.jsonl").is_file() { + return Ok(()); + } + let marked = repo.run(|docs| 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)) +} From d66e12de9f48678f06f0db782f0d2d98d8b058c7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:07:42 +0300 Subject: [PATCH 07/17] refactor(threads): move conversation storage onto the storage backend The thread store now resolves conversations through the process storage backend instead of the retired path-bound JSONL store, keeping the existing synchronous facade for callers. A one-time read-only import of the legacy `memory/conversations` tree preserves previously persisted transcripts. Auto-committed-on: dragonfly --- .../src/threads/store/legacy_import_tests.rs | 1 + .../openhuman-core/src/threads/store/mod.rs | 61 +++++++++++-------- 2 files changed, 37 insertions(+), 25 deletions(-) create mode 100644 crates/openhuman-core/src/threads/store/legacy_import_tests.rs 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..4563e55b7b1 --- /dev/null +++ b/crates/openhuman-core/src/threads/store/legacy_import_tests.rs @@ -0,0 +1 @@ +use super::*; 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, }; From ac49a90ccbc4a615df853ae32265066daf81addb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:17:30 +0300 Subject: [PATCH 08/17] test(threads): cover legacy jsonl import and backend store round trip Add tests for importing legacy jsonl conversations, verifying the import runs only once per workspace and leaves the original files byte identical, and that a workspace without legacy files starts empty. Also cover the backend store round trip for threads, messages, search, and deletion. Auto-committed-on: dragonfly --- .../src/threads/store/legacy_import_tests.rs | 150 ++++++++++++++++++ 1 file changed, 150 insertions(+) diff --git a/crates/openhuman-core/src/threads/store/legacy_import_tests.rs b/crates/openhuman-core/src/threads/store/legacy_import_tests.rs index 4563e55b7b1..9f31d23b69b 100644 --- a/crates/openhuman-core/src/threads/store/legacy_import_tests.rs +++ b/crates/openhuman-core/src/threads/store/legacy_import_tests.rs @@ -1 +1,151 @@ 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()); +} From 097b3b020da8ace02d7da9579906fa87df59aede Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:17:57 +0300 Subject: [PATCH 09/17] fix(storage): forward append_batch_if_offset through fenced streams The fenced stream store now delegates append_batch_if_offset to the inner backend instead of falling back to the trait default, which reported the operation as unsupported and rejected every conversation append even on drivers that implement it. Auto-committed-on: dragonfly --- .../openhuman-core/src/storage/fenced_backend.rs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) 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 } From 08e41562118ee654fda28d1c1ab9a3e5b327b442 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:20:34 +0300 Subject: [PATCH 10/17] test(storage): cover conditional append pass-through in fenced backend Add tests asserting that the fenced backend wrapper forwards append_batch_if_offset to the underlying driver instead of falling back to the trait default, which returned Unsupported and rejected every conversation append on the default SQLite backend. The shared helper is exercised against both the in-memory and SQLite backends, checking offset assignment, conflict reporting on a stale offset, and that the stream length stays unchanged. Auto-committed-on: dragonfly --- .../src/storage/fenced_backend_tests.rs | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) 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; +} From b645cfc48ff322d77ad0907a3f53693bd246ce70 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:23:01 +0300 Subject: [PATCH 11/17] fix(threads): restore not-found wording in conversation store errors The store now maps driver not-found errors back to the "thread not found" and "message not found in thread " text the file store produced, so callers that match on that wording keep working. Each operation passes the thread or message it addresses so the error can name the right subject. Auto-committed-on: dragonfly --- .../src/threads/store/backend.rs | 82 +++++++++++++++---- 1 file changed, 67 insertions(+), 15 deletions(-) diff --git a/crates/openhuman-core/src/threads/store/backend.rs b/crates/openhuman-core/src/threads/store/backend.rs index 002b6ab0129..6d0eb24958a 100644 --- a/crates/openhuman-core/src/threads/store/backend.rs +++ b/crates/openhuman-core/src/threads/store/backend.rs @@ -19,13 +19,14 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use tinyagents_session::threads::{ - ConversationPersistence, ConversationResult, ConversationThread, CreateConversationThread, + ConversationError, ConversationPersistence, ConversationResult, ConversationThread, CreateConversationThread, CrossThreadHit, DriverConversationStore, ThreadMessage, ThreadMessagePatch, }; pub use tinyagents_session::threads::ConversationPurgeStats; use crate::storage::documents::Repo; use crate::storage::{self, Scope, StorageError}; +use tinystoragedrivers::ErrorKind; const DOMAIN: &str = "threads::store"; const LOG_PREFIX: &str = "[conversations]"; @@ -42,6 +43,52 @@ fn conversation_scope() -> Result { .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 { @@ -80,7 +127,12 @@ impl ConversationStore { Ok(store) } - fn run(&self, operation: &'static str, f: F) -> Result + fn run( + &self, + operation: &'static str, + subject: Subject<'_>, + f: F, + ) -> Result where F: FnOnce(DriverConversationStore) -> Fut, Fut: std::future::Future> + Send + 'static, @@ -92,7 +144,7 @@ impl ConversationStore { .map_err(|e: StorageError| format!("{LOG_PREFIX} {operation}: {e}"))?; outcome.map_err(|e| { log::debug!("{LOG_PREFIX} {operation} failed: {e}"); - e.to_string() + describe(&e, subject) }) } @@ -101,20 +153,20 @@ impl ConversationStore { &self, request: CreateConversationThread, ) -> Result { - self.run("ensure_thread", |s| async move { + 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", |s| async move { s.list_threads().await }) + 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", |s| async move { + self.run("get_messages", thread(thread_id), |s| async move { if s.get_thread(&id).await?.is_none() { return Ok(Vec::new()); } @@ -131,7 +183,7 @@ impl ConversationStore { ) -> Result, String> { let query = query.to_string(); let exclude = exclude_thread_id.map(str::to_string); - self.run("search_cross_thread_messages", |s| async move { + self.run("search_cross_thread_messages", NONE, |s| async move { s.search_cross_thread_messages(&query, limit, exclude.as_deref()) .await }) @@ -144,7 +196,7 @@ impl ConversationStore { message: ThreadMessage, ) -> Result { let id = thread_id.to_string(); - self.run("append_message", |s| async move { + self.run("append_message", thread(thread_id), |s| async move { s.append_message(&id, message).await }) } @@ -157,7 +209,7 @@ impl ConversationStore { updated_at: &str, ) -> Result { let (id, title, at) = (thread_id.to_string(), title.to_string(), updated_at.to_string()); - self.run("update_thread_title", |s| async move { + self.run("update_thread_title", thread(thread_id), |s| async move { s.update_thread_title(&id, &title, &at).await }) } @@ -170,7 +222,7 @@ impl ConversationStore { updated_at: &str, ) -> Result { let (id, at) = (thread_id.to_string(), updated_at.to_string()); - self.run("update_thread_working_dir", |s| async move { + self.run("update_thread_working_dir", thread(thread_id), |s| async move { s.update_thread_working_dir(&id, working_dir, &at).await }) } @@ -183,7 +235,7 @@ impl ConversationStore { updated_at: &str, ) -> Result { let (id, at) = (thread_id.to_string(), updated_at.to_string()); - self.run("update_thread_labels", |s| async move { + self.run("update_thread_labels", thread(thread_id), |s| async move { s.update_thread_labels(&id, labels, &at).await }) } @@ -196,7 +248,7 @@ impl ConversationStore { patch: ThreadMessagePatch, ) -> Result { let (id, mid) = (thread_id.to_string(), message_id.to_string()); - self.run("update_message", |s| async move { + self.run("update_message", message(thread_id, message_id), |s| async move { s.update_message(&id, &mid, patch).await }) } @@ -208,7 +260,7 @@ impl ConversationStore { message_id: &str, ) -> Result, String> { let (id, mid) = (thread_id.to_string(), message_id.to_string()); - self.run("delete_messages_from", |s| async move { + self.run("delete_messages_from", message(thread_id, message_id), |s| async move { s.delete_messages_from(&id, &mid).await }) } @@ -216,14 +268,14 @@ impl ConversationStore { /// 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", |s| async move { + 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", |s| async move { s.purge_threads().await }) + self.run("purge_threads", NONE, |s| async move { s.purge_threads().await }) } } From 724970c73a7f7469ec2e657f93da69d7d4fcd74d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:28:31 +0300 Subject: [PATCH 12/17] docs(core): clarify storage backend comments Update the rusqlite re-export note to record that both consumers are still present after re-checking #7768, so the re-export stays. Also reword the threads module doc to describe storage as the storage backend rather than JSONL files. Auto-committed-on: dragonfly --- crates/openhuman-core/src/storage/mod.rs | 3 ++- crates/openhuman-core/src/threads/mod.rs | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs index e9c572f504d..6de7ad9b534 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, 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; From 16121ae43c29b1035e10304280127530dd68aa20 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:28:51 +0300 Subject: [PATCH 13/17] docs(session_host): tidy driver module comment Reword the module-level doc comment on the session driver adapter to read more concisely, clarifying that the runtime session owns transcript and history state and supplies the immutable model history and tool declarations per call. Auto-committed-on: dragonfly --- crates/openhuman-core/src/agent/session_host/driver.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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; From 52d391d7638fa076da9fb02e41a186cf7a9b067f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:33:33 +0300 Subject: [PATCH 14/17] chore(vendor): bump tinyagents submodule Update the vendored tinyagents submodule to a newer commit. Auto-committed-on: dragonfly --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index 80460be5f8e..fed928f7d7e 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 80460be5f8e0daed81479dc0921c5040ddfd3bea +Subproject commit fed928f7d7e3f26f6b525a872bfef1026f7a0678 From e9332f48cb601581302aaaf8dea472c636f33bc3 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 03:48:10 +0300 Subject: [PATCH 15/17] chore(threads): reformat store and composio call sites Reformatted several call sites in the conversation store, legacy import, and composio file store to satisfy rustfmt line-width rules. No behaviour changes. Auto-committed-on: dragonfly --- .../src/integrations/composio/file_store.rs | 11 +++-- .../composio/file_store_documents.rs | 7 ++- .../src/threads/store/backend.rs | 49 +++++++++++++------ .../src/threads/store/legacy_import.rs | 12 +++-- 4 files changed, 54 insertions(+), 25 deletions(-) diff --git a/crates/openhuman-core/src/integrations/composio/file_store.rs b/crates/openhuman-core/src/integrations/composio/file_store.rs index 31b1f476753..7fbe4cc54c0 100644 --- a/crates/openhuman-core/src/integrations/composio/file_store.rs +++ b/crates/openhuman-core/src/integrations/composio/file_store.rs @@ -30,10 +30,13 @@ async fn import_legacy( 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:#}"))?; + 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(()); } 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 7ac262a8476..d8564938d1e 100644 --- a/crates/openhuman-core/src/integrations/composio/file_store_documents.rs +++ b/crates/openhuman-core/src/integrations/composio/file_store_documents.rs @@ -148,7 +148,12 @@ impl Docs { Err(error) => return Err(error), }; match docs - .put(IMPORTED, &id, json!({ "imported": true }), Precondition::Absent) + .put( + IMPORTED, + &id, + json!({ "imported": true }), + Precondition::Absent, + ) .await { Ok(_) => {} diff --git a/crates/openhuman-core/src/threads/store/backend.rs b/crates/openhuman-core/src/threads/store/backend.rs index 6d0eb24958a..0391254ea81 100644 --- a/crates/openhuman-core/src/threads/store/backend.rs +++ b/crates/openhuman-core/src/threads/store/backend.rs @@ -18,11 +18,12 @@ 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, + ConversationError, ConversationPersistence, ConversationResult, ConversationThread, + CreateConversationThread, CrossThreadHit, DriverConversationStore, ThreadMessage, + ThreadMessagePatch, }; -pub use tinyagents_session::threads::ConversationPurgeStats; use crate::storage::documents::Repo; use crate::storage::{self, Scope, StorageError}; @@ -160,7 +161,11 @@ impl ConversationStore { /// Every live thread, newest first. pub fn list_threads(&self) -> Result, String> { - self.run("list_threads", NONE, |s| async move { s.list_threads().await }) + self.run( + "list_threads", + NONE, + |s| async move { s.list_threads().await }, + ) } /// A thread's messages in append order; empty for an unknown thread. @@ -208,7 +213,11 @@ impl ConversationStore { title: &str, updated_at: &str, ) -> Result { - let (id, title, at) = (thread_id.to_string(), title.to_string(), updated_at.to_string()); + 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 }) @@ -222,9 +231,11 @@ impl ConversationStore { 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 - }) + 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. @@ -248,9 +259,11 @@ impl ConversationStore { 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 - }) + 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. @@ -260,9 +273,11 @@ impl ConversationStore { 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 - }) + 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. @@ -275,7 +290,11 @@ impl ConversationStore { /// Removes every thread. pub fn purge_threads(&self) -> Result { - self.run("purge_threads", NONE, |s| async move { s.purge_threads().await }) + self.run( + "purge_threads", + NONE, + |s| async move { s.purge_threads().await }, + ) } } diff --git a/crates/openhuman-core/src/threads/store/legacy_import.rs b/crates/openhuman-core/src/threads/store/legacy_import.rs index a9f16d0a75a..7f940283251 100644 --- a/crates/openhuman-core/src/threads/store/legacy_import.rs +++ b/crates/openhuman-core/src/threads/store/legacy_import.rs @@ -100,7 +100,8 @@ fn fold_threads(root: &Path) -> Result> { } => 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.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 { @@ -158,12 +159,13 @@ pub(super) fn import_once( 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))?); + 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 - })?; + 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( From a6726984224ca4582002ef90d9a169a71e40e3fb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 04:01:00 +0300 Subject: [PATCH 16/17] refactor(storage): route subagent and thread stores through shared storage module The subagent session store and thread store now go through the shared storage module instead of their own paths, and the legacy import path was updated to match. This consolidates storage access so the stores share one implementation rather than duplicating it. Auto-committed-on: dragonfly --- .../orchestration/subagent_sessions/store.rs | 4 ++-- .../subagent_sessions/store_documents.rs | 22 ++++++++++++++----- crates/openhuman-core/src/storage/mod.rs | 10 +++++++++ .../src/threads/store/blocking.rs | 2 +- .../src/threads/store/legacy_import.rs | 22 ++++++++++++++----- 5 files changed, 46 insertions(+), 14 deletions(-) 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 d83e4ec8355..1c628aa36ea 100644 --- a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs +++ b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs @@ -20,7 +20,7 @@ impl SubagentSessionStore { /// 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().map_err(|err| format!("{err:#}"))? { + if docs.legacy_imported(&self.workspace_dir).map_err(|err| format!("{err:#}"))? { return Ok(()); } let path = self.legacy_path(); @@ -31,7 +31,7 @@ impl SubagentSessionStore { }; let sessions: Vec = serde_json::from_str(&raw) .map_err(|err| format!("failed to parse legacy subagent sessions: {err}"))?; - docs.import_legacy(&sessions) + docs.import_legacy(&self.workspace_dir, &sessions) .map_err(|err| format!("{err:#}"))?; log::debug!( "[subagent_sessions] imported legacy sessions count={}", 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 5a0058a2957..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,7 +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) | +//! | `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 @@ -32,6 +32,12 @@ const SESSIONS: &str = "subagent_sessions"; /// 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 { @@ -90,22 +96,28 @@ impl Docs { } /// Whether the one-time legacy file import finished in this scope. - pub(super) fn legacy_imported(&self) -> Result { + 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, LEGACY_MARKER).await })?; + .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, sessions: &[DurableSubagentSession]) -> Result<()> { + 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, - LEGACY_MARKER, + &id, serde_json::json!({ "imported": true }), Precondition::Absent, ) diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs index 6de7ad9b534..ce186221106 100644 --- a/crates/openhuman-core/src/storage/mod.rs +++ b/crates/openhuman-core/src/storage/mod.rs @@ -339,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/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 index 7f940283251..f6181efd91a 100644 --- a/crates/openhuman-core/src/threads/store/legacy_import.rs +++ b/crates/openhuman-core/src/threads/store/legacy_import.rs @@ -22,7 +22,7 @@ use tinystoragedrivers::{CollectionSpec, ErrorKind, Precondition}; use crate::storage::documents::Repo; const MARKERS: &str = "threads_import"; -const MARKER_ID: &str = "legacy-conversations"; +const MARKER_PREFIX: &str = "legacy-conversations"; const LOG_PREFIX: &str = "[conversations:import]"; /// The collection that holds the completion marker. @@ -147,11 +147,21 @@ pub(super) fn import_once( store: &DriverConversationStore, ) -> Result<()> { let root = super::backend::legacy_root(workspace_dir); - repo.once_per_backend("legacy-import", || { - if !root.join("threads.jsonl").is_file() { - return Ok(()); + 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| async move { docs.get(MARKERS, MARKER_ID).await })?; + 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(()); @@ -170,7 +180,7 @@ pub(super) fn import_once( match docs .put( MARKERS, - MARKER_ID, + &marker_id, serde_json::json!({ "threads": threads_n, "messages": messages_n }), Precondition::Absent, ) From 8121dcc243720d5de2dfb8efb875f4c318db3d3b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Wed, 14 Oct 2026 04:06:32 +0300 Subject: [PATCH 17/17] test(subagent_sessions): cover per-workspace legacy import markers Add tests asserting that workspaces sharing a backend each import their own legacy file, and that the legacy import marker is keyed per source workspace. Also cover that an unreadable legacy log surfaces as an error rather than an empty store, so a later retry can still import. Auto-committed-on: dragonfly --- .../orchestration/subagent_sessions/store.rs | 5 +++- .../store_documents_tests.rs | 18 ++++++++++++ .../src/threads/store/legacy_import_tests.rs | 29 +++++++++++++++++++ 3 files changed, 51 insertions(+), 1 deletion(-) 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 1c628aa36ea..3daca14f75b 100644 --- a/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs +++ b/crates/openhuman-core/src/agent/orchestration/subagent_sessions/store.rs @@ -20,7 +20,10 @@ impl SubagentSessionStore { /// 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:#}"))? { + if docs + .legacy_imported(&self.workspace_dir) + .map_err(|err| format!("{err:#}"))? + { return Ok(()); } let path = self.legacy_path(); 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 fc25962d730..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 @@ -169,3 +169,21 @@ fn a_legacy_session_file_is_imported_once_and_left_byte_identical() { 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/threads/store/legacy_import_tests.rs b/crates/openhuman-core/src/threads/store/legacy_import_tests.rs index 9f31d23b69b..6a580ade1f8 100644 --- a/crates/openhuman-core/src/threads/store/legacy_import_tests.rs +++ b/crates/openhuman-core/src/threads/store/legacy_import_tests.rs @@ -149,3 +149,32 @@ fn threads_round_trip_through_the_backend_store() { 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" + ); + } +}