storage(#7768): importers leave legacy files untouched; tinyagents back to main with backend-backed threads

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<DurableSubagentSession> = 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(())
     }
 

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:<workspace key>` | 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<CollectionSpec> {
-    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<bool> {
+        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 {

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

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;
 

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 `<workspace>/integrations/<name>` file into the
-/// document store, once: the value is stored only if none exists yet, then the
-/// file is renamed to `<name>.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(())
 }
 

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/<name>` 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<CollectionSpec> {
-    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/<name>` file
+    /// finished in this scope.
+    pub(super) fn legacy_imported(&self, name: &str) -> Result<bool> {
+        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<bool> {
         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)
         })
     }
 

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<String, u32> = 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<String, u32> = 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}"#
+    );
 }

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<Value>,
+    ) -> Result<u64> {
+        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<Vec<StreamEntry>> {
         self.inner.read_window(stream, from, limit).await
     }

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