diff --git a/app/test/playwright/specs/harness-composio-tool-flow.spec.ts b/app/test/playwright/specs/harness-composio-tool-flow.spec.ts index cf95cc635..42937da4c 100644 --- a/app/test/playwright/specs/harness-composio-tool-flow.spec.ts +++ b/app/test/playwright/specs/harness-composio-tool-flow.spec.ts @@ -171,7 +171,7 @@ test.describe('Harness - Composio tool-call prompt flow', () => { await sendMessage(page, 'check my email'); await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 }); - await expect(page.getByText(/Q3 Budget Review/i)).toBeVisible(); + await expect(page.getByText(/Q3 Budget Review/i).first()).toBeVisible(); const log = await requests(); const llmHits = log.filter( diff --git a/app/test/playwright/specs/harness-cron-prompt-flow.spec.ts b/app/test/playwright/specs/harness-cron-prompt-flow.spec.ts index 1a01b38a0..f031f48c8 100644 --- a/app/test/playwright/specs/harness-cron-prompt-flow.spec.ts +++ b/app/test/playwright/specs/harness-cron-prompt-flow.spec.ts @@ -142,7 +142,7 @@ test.describe('Harness - Cron prompt-flow', () => { await setMockBehavior('llmStreamChunkDelayMs', '10'); await sendMessage(page, 'remind me every morning at 9am'); - await expect(page.getByText(CANARY)).toBeVisible({ timeout: 60_000 }); + await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 }); await expect(page.getByText(/Done! I have set up a daily 9am morning reminder/i)).toBeVisible(); const log = await requests(); const llmHits = log.filter( @@ -165,7 +165,7 @@ test.describe('Harness - Cron prompt-flow', () => { await setMockBehavior('llmStreamChunkDelayMs', '10'); await sendMessage(page, 'what are my scheduled tasks'); - await expect(page.getByText(CANARY)).toBeVisible({ timeout: 60_000 }); + await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 }); await expect(page.getByText(/You have 2 scheduled tasks/i)).toBeVisible(); }); @@ -193,7 +193,7 @@ test.describe('Harness - Cron prompt-flow', () => { await setMockBehavior('llmStreamChunkDelayMs', '10'); await sendMessage(page, 'change my morning reminder to 8am'); - await expect(page.getByText(CANARY)).toBeVisible({ timeout: 60_000 }); + await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 }); await expect(page.getByText(/changed your morning reminder to 8am/i)).toBeVisible(); }); @@ -218,7 +218,7 @@ test.describe('Harness - Cron prompt-flow', () => { await setMockBehavior('llmStreamChunkDelayMs', '10'); await sendMessage(page, 'delete the morning reminder'); - await expect(page.getByText(CANARY)).toBeVisible({ timeout: 60_000 }); - await expect(page.getByText(/deleted the morning reminder/i)).toBeVisible(); + await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 }); + await expect(page.getByText(/deleted the morning reminder/i).first()).toBeVisible(); }); }); diff --git a/app/test/playwright/specs/harness-search-tool-flow.spec.ts b/app/test/playwright/specs/harness-search-tool-flow.spec.ts index 018afe7aa..5bf644237 100644 --- a/app/test/playwright/specs/harness-search-tool-flow.spec.ts +++ b/app/test/playwright/specs/harness-search-tool-flow.spec.ts @@ -220,7 +220,7 @@ test.describe('Harness - Search tool-flow', () => { await sendMessage(page, 'read the README'); await expect(page.getByText(CANARY).first()).toBeVisible({ timeout: 60_000 }); - await expect(page.getByText(/OpenHuman is an AI assistant/i)).toBeVisible(); + await expect(page.getByText(/OpenHuman is an AI assistant/i).first()).toBeVisible(); const log = await requests(); const llmHits = log.filter( diff --git a/scripts/test-rust-e2e.sh b/scripts/test-rust-e2e.sh index 742e589e4..9939b24be 100755 --- a/scripts/test-rust-e2e.sh +++ b/scripts/test-rust-e2e.sh @@ -26,10 +26,10 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" -# The full set of `tests/*_e2e.rs` files. Each gets a `--test ` flag -# in the single `cargo test` invocation so cargo compiles them in one -# unit and only the test binaries that exist get run. Tests guarded by -# `#[ignore]` stay skipped unless the caller passes `-- --ignored`. +# The full set of `tests/*_e2e.rs` files. The default runner executes them +# serially so CI does not link several large integration binaries at once. +# Tests guarded by `#[ignore]` stay skipped unless the caller passes +# `-- --ignored`. ALL_E2E_SUITES=( agent_retrieval_e2e autocomplete_memory_e2e @@ -128,19 +128,13 @@ export VITE_BACKEND_URL="$MOCK_API_URL" cd "$REPO_ROOT" source "$HOME/.cargo/env" 2>/dev/null || true -# Assemble the `--test ` flags so a single `cargo test` invocation -# compiles + runs every suite. Cargo will fail fast if any --test binary -# doesn't exist, which is the signal you want when a suite gets renamed. -CARGO_FLAGS=() +echo "[rust-e2e] Running ${#SUITES[@]} suite(s) serially." for suite in "${SUITES[@]}"; do - CARGO_FLAGS+=(--test "$suite") + if [ "${#EXTRA_ARGS[@]}" -gt 0 ]; then + echo "[rust-e2e] cargo test --manifest-path Cargo.toml --test $suite -- ${EXTRA_ARGS[*]}" + cargo test --manifest-path Cargo.toml --test "$suite" -- "${EXTRA_ARGS[@]}" + else + echo "[rust-e2e] cargo test --manifest-path Cargo.toml --test $suite" + cargo test --manifest-path Cargo.toml --test "$suite" + fi done - -echo "[rust-e2e] Running:" -if [ "${#EXTRA_ARGS[@]}" -gt 0 ]; then - echo "[rust-e2e] cargo test --manifest-path Cargo.toml ${CARGO_FLAGS[*]} -- ${EXTRA_ARGS[*]}" - cargo test --manifest-path Cargo.toml "${CARGO_FLAGS[@]}" -- "${EXTRA_ARGS[@]}" -else - echo "[rust-e2e] cargo test --manifest-path Cargo.toml ${CARGO_FLAGS[*]}" - cargo test --manifest-path Cargo.toml "${CARGO_FLAGS[@]}" -fi diff --git a/src/openhuman/memory_sources/reconcile.rs b/src/openhuman/memory_sources/reconcile.rs index 798d2fb8b..89fe4c780 100644 --- a/src/openhuman/memory_sources/reconcile.rs +++ b/src/openhuman/memory_sources/reconcile.rs @@ -140,6 +140,7 @@ fn apply_caps_defaults_to_entries(sources: &mut [MemorySourceEntry]) -> u32 { /// re-run it exactly once). Entries the user has already customised (non-None caps) /// are left untouched. pub async fn apply_composio_source_caps_migration() -> Result<(), String> { + let _guard = registry::memory_sources_write_guard().await; let mut config = config_rpc::load_config_with_timeout().await?; if config.composio_source_caps_migration_version >= CURRENT_CAPS_MIGRATION_VERSION { diff --git a/src/openhuman/memory_sources/registry.rs b/src/openhuman/memory_sources/registry.rs index a362d7021..a2f480661 100644 --- a/src/openhuman/memory_sources/registry.rs +++ b/src/openhuman/memory_sources/registry.rs @@ -6,6 +6,16 @@ use crate::openhuman::config::rpc as config_rpc; use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind}; +use std::sync::OnceLock; + +static MEMORY_SOURCES_WRITE_LOCK: OnceLock> = OnceLock::new(); + +pub(crate) async fn memory_sources_write_guard() -> tokio::sync::MutexGuard<'static, ()> { + MEMORY_SOURCES_WRITE_LOCK + .get_or_init(|| tokio::sync::Mutex::new(())) + .lock() + .await +} /// Conservative default sync caps for a Composio toolkit, keyed by toolkit slug. /// @@ -50,6 +60,7 @@ pub async fn get_source(id: &str) -> Result, String> { pub async fn add_source(entry: MemorySourceEntry) -> Result { entry.validate()?; + let _guard = memory_sources_write_guard().await; let mut config = config_rpc::load_config_with_timeout().await?; if config.memory_sources.iter().any(|s| s.id == entry.id) { @@ -75,6 +86,7 @@ pub async fn update_source( id: &str, patch: MemorySourcePatch, ) -> Result { + let _guard = memory_sources_write_guard().await; let mut config = config_rpc::load_config_with_timeout().await?; let entry = config @@ -159,6 +171,7 @@ pub async fn update_source( } pub async fn remove_source(id: &str) -> Result { + let _guard = memory_sources_write_guard().await; let mut config = config_rpc::load_config_with_timeout().await?; let before = config.memory_sources.len(); config.memory_sources.retain(|s| s.id != id); @@ -182,6 +195,7 @@ pub async fn remove_source(id: &str) -> Result { /// connection-delete flow doesn't have, so this is the connection-keyed /// counterpart. Returns the number of entries removed (0 if none matched). pub async fn remove_composio_source_by_connection_id(connection_id: &str) -> Result { + let _guard = memory_sources_write_guard().await; let mut config = config_rpc::load_config_with_timeout().await?; let before = config.memory_sources.len(); config.memory_sources.retain(|s| { @@ -212,6 +226,7 @@ pub async fn upsert_composio_source( connection_id: &str, label: &str, ) -> Result { + let _guard = memory_sources_write_guard().await; let mut config = config_rpc::load_config_with_timeout().await?; if let Some(existing) = config.memory_sources.iter_mut().find(|s| { @@ -332,6 +347,7 @@ pub struct MemorySourcePatch { /// /// Saves config once after all mutations and returns the updated entries. pub async fn apply_all_in() -> Result, String> { + let _guard = memory_sources_write_guard().await; let mut config = config_rpc::load_config_with_timeout().await?; tracing::info!(