From 5c49a969e564a7d855f7d5ff40638e2995bdd4eb Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 17:14:48 +0530 Subject: [PATCH 1/4] fix(event-bus): raise test-only capacity floors to fix a full-suite flake MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under cfg(test) the crate's ~12.6k unit tests share two process-global bounded resources sized for production: the event bus's 256-slot broadcast channel (src/core/event_bus/bus.rs) and the learning candidate ring's 1024 entries (src/openhuman/learning/candidate.rs, whose drainer does not run in the unit suite). Under the concurrent full-suite publish load a slow subscriber lags and drops events (RecvError::Lagged), and the candidate ring evicts a test's own candidates — intermittently failing bus-delivery tests such as learning::startup::tests::learning_subscriber_fires_with_no_channel_configured. Raise both floors to 65536 for the TEST BINARY ONLY (cfg(test)); production keeps 256 / 1024 and ships byte-identical. No new dependency, no serialization, no weakened assertion — the shared globals simply get enough headroom that neither drops nor evicts under the artificial suite-wide load. --- src/core/event_bus/bus.rs | 9 +++++++++ src/openhuman/learning/candidate.rs | 10 +++++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/src/core/event_bus/bus.rs b/src/core/event_bus/bus.rs index 3faa9d63d0..b608e0de41 100644 --- a/src/core/event_bus/bus.rs +++ b/src/core/event_bus/bus.rs @@ -41,6 +41,15 @@ pub fn init_global(capacity: usize) -> &'static EventBus { // is always safe from anywhere in the process once the bus is up. init_native_registry(); GLOBAL_BUS.get_or_init(|| { + // Under `cfg(test)` the crate's ~12.6k unit tests share this one + // process-global bus and publish concurrently, so the default 256-slot + // broadcast buffer lets a slow subscriber fall behind and drop events + // (`RecvError::Lagged`) — intermittently flaking bus-delivery tests such + // as `learning::startup::tests::learning_subscriber_fires_with_no_channel_configured`. + // Raise the buffer floor for the test binary only; production is + // unaffected (it publishes at human speed and keeps the caller's capacity). + #[cfg(test)] + let capacity = capacity.max(1 << 16); tracing::debug!(capacity, "[event_bus] initializing global singleton"); EventBus::create(capacity) }) diff --git a/src/openhuman/learning/candidate.rs b/src/openhuman/learning/candidate.rs index 95285df258..d063401fae 100644 --- a/src/openhuman/learning/candidate.rs +++ b/src/openhuman/learning/candidate.rs @@ -233,7 +233,15 @@ static GLOBAL_BUFFER: OnceLock = OnceLock::new(); /// Initialised on first call with a default capacity of 1024. All producers /// push into this buffer; the stability detector drains it. pub fn global() -> &'static Buffer { - GLOBAL_BUFFER.get_or_init(|| Buffer::new(1024)) + // Production default: a 1024-entry ring drained by the stability detector. + // Under `cfg(test)` the crate's full parallel suite shares this one global + // ring with no drainer running, so concurrent producers can evict a test's + // own candidates before it asserts on them. Enlarge for the test binary only. + #[cfg(not(test))] + let capacity = 1024; + #[cfg(test)] + let capacity = 1 << 16; + GLOBAL_BUFFER.get_or_init(|| Buffer::new(capacity)) } // ── Tests ──────────────────────────────────────────────────────────────────── From 26b82504790eb30cabd495197f074b4162da96b3 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 17:51:36 +0530 Subject: [PATCH 2/4] fix(ci): acknowledge composio/action_tool.rs in the gate-smoke allowlist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `Rust Feature-Gate Smoke` lane guards that every feature-gated test module is acknowledged in an EXPECTED allowlist. `composio/action_tool.rs` gained a `#[cfg(feature = "flows")]` test in #4853 but was never added to the list, so the guard's ACTUAL-vs-EXPECTED diff fails on it — on every PR since. This is the guard's own sanctioned maintenance (its error message says "Update the EXPECTED allowlist"), not a weakening: the guard still validates every other file. Replicated the guard's grep+diff locally to confirm this is the only drift and the diff is now empty. --- .github/workflows/ci-lite.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci-lite.yml b/.github/workflows/ci-lite.yml index a5162b892c..4ae6ae2029 100644 --- a/.github/workflows/ci-lite.yml +++ b/.github/workflows/ci-lite.yml @@ -451,6 +451,7 @@ jobs: openhuman/agent/harness/session/tests.rs openhuman/agent/harness/subagent_runner/tool_prep.rs openhuman/agent_registry/agents/loader.rs + openhuman/composio/action_tool.rs openhuman/inference/local/mod.rs openhuman/tool_registry/ops_tests.rs openhuman/tool_registry/schemas.rs From d06d58278786372d13590d3d9c0136e8981bb544 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 18:20:48 +0530 Subject: [PATCH 3/4] fix(test): isolate event-bus flakes properly; revert ineffective capacity floor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The earlier test-only capacity floor (5c49a969) did not fix the flake: the two failing tests share the process-global broadcast bus, so under the parallel suite they receive events published by *other* tests — a capacity bump cannot fix cross-test interference. Replace it with real isolation. - learning::startup: rewrite `learning_subscriber_fires_*` to publish a DocumentCanonicalized event on a private `EventBus::create(64)` with a locally-subscribed `EmailSignatureSubscriber`, instead of asserting on the shared global bus. Add a synchronous `register_email_signature_subscriber` guard test for the channel-independent registration path (#5003). - web_chat::event_bus: the automation-halt test reads from the shared web-channel bus; drain foreign events and match on `event == "automation_halt"` rather than taking whatever is first in the buffer. - Revert the `#[cfg(test)]` capacity floors in event_bus/bus.rs and learning/candidate.rs back to origin/main — no longer needed. --- src/core/event_bus/bus.rs | 9 ------ src/openhuman/learning/candidate.rs | 10 +----- src/openhuman/learning/startup.rs | 49 +++++++++++++++++++++++------ src/openhuman/web_chat/event_bus.rs | 29 +++++++++++++---- 4 files changed, 64 insertions(+), 33 deletions(-) diff --git a/src/core/event_bus/bus.rs b/src/core/event_bus/bus.rs index b608e0de41..3faa9d63d0 100644 --- a/src/core/event_bus/bus.rs +++ b/src/core/event_bus/bus.rs @@ -41,15 +41,6 @@ pub fn init_global(capacity: usize) -> &'static EventBus { // is always safe from anywhere in the process once the bus is up. init_native_registry(); GLOBAL_BUS.get_or_init(|| { - // Under `cfg(test)` the crate's ~12.6k unit tests share this one - // process-global bus and publish concurrently, so the default 256-slot - // broadcast buffer lets a slow subscriber fall behind and drop events - // (`RecvError::Lagged`) — intermittently flaking bus-delivery tests such - // as `learning::startup::tests::learning_subscriber_fires_with_no_channel_configured`. - // Raise the buffer floor for the test binary only; production is - // unaffected (it publishes at human speed and keeps the caller's capacity). - #[cfg(test)] - let capacity = capacity.max(1 << 16); tracing::debug!(capacity, "[event_bus] initializing global singleton"); EventBus::create(capacity) }) diff --git a/src/openhuman/learning/candidate.rs b/src/openhuman/learning/candidate.rs index d063401fae..95285df258 100644 --- a/src/openhuman/learning/candidate.rs +++ b/src/openhuman/learning/candidate.rs @@ -233,15 +233,7 @@ static GLOBAL_BUFFER: OnceLock = OnceLock::new(); /// Initialised on first call with a default capacity of 1024. All producers /// push into this buffer; the stability detector drains it. pub fn global() -> &'static Buffer { - // Production default: a 1024-entry ring drained by the stability detector. - // Under `cfg(test)` the crate's full parallel suite shares this one global - // ring with no drainer running, so concurrent producers can evict a test's - // own candidates before it asserts on them. Enlarge for the test binary only. - #[cfg(not(test))] - let capacity = 1024; - #[cfg(test)] - let capacity = 1 << 16; - GLOBAL_BUFFER.get_or_init(|| Buffer::new(capacity)) + GLOBAL_BUFFER.get_or_init(|| Buffer::new(1024)) } // ── Tests ──────────────────────────────────────────────────────────────────── diff --git a/src/openhuman/learning/startup.rs b/src/openhuman/learning/startup.rs index 2da5a5de62..2336bca79b 100644 --- a/src/openhuman/learning/startup.rs +++ b/src/openhuman/learning/startup.rs @@ -270,13 +270,20 @@ mod tests { } #[tokio::test] - async fn learning_subscriber_fires_with_no_channel_configured() { - init_global(DEFAULT_CAPACITY); - let (tmp, _client) = test_client(); - // Make the memory client ready so the full Platform wiring runs — no - // channel runtime is ever constructed in this test. - let _ = crate::openhuman::memory::global::init(tmp.path().join("workspace")); - register_learning_subscribers(tmp.path().to_path_buf()); + async fn learning_subscriber_fires_on_document_event() { + // The email-signature producer reacts to `DocumentCanonicalized` and pushes + // Identity candidates into the global candidate buffer. Driving this through + // the process-global event bus (as it originally did) is flaky under the full + // parallel suite: that singleton is shared by every test, so its receiver + // starves under load and drops this test's event. Exercise the subscriber on + // an ISOLATED bus instead — the repo's canonical test-isolation pattern (see + // `core::event_bus::bus`) — so delivery is deterministic. The #5003 + // regression (the subscriber registers on the channel-independent boot path) + // is guarded separately by + // `email_signature_subscriber_registers_on_channel_independent_path` below. + use crate::core::event_bus::EventBus; + use crate::openhuman::learning::extract::signature::EmailSignatureSubscriber; + use std::sync::Arc; let source_id = unique_source_id("e2e"); let body = signature_body(); @@ -286,12 +293,36 @@ mod tests { "signature body must yield at least one identity candidate" ); - publish_email_doc(&source_id, &body); + let bus = EventBus::create(64); + let _handle = bus.subscribe(Arc::new(EmailSignatureSubscriber)); + bus.publish(DomainEvent::DocumentCanonicalized { + source_id: source_id.clone(), + source_kind: "email".to_string(), + chunks_written: 1, + chunk_ids: vec![format!("{source_id}-c1")], + canonicalized_at: 0.0, + body_preview: Some(body), + }); + let got = wait_for_candidates(&source_id, expected).await; assert_eq!( got, expected, "email-signature subscriber must push the parsed identity candidates \ - with no channel configured anywhere (#5003)" + from a DocumentCanonicalized event" + ); + } + + /// #5003 regression guard (channel-independent boot path): the email-signature + /// subscriber must register even when no memory client / channel is configured. + /// Synchronous (no bus delivery), so it is not subject to the shared-bus flake. + #[tokio::test] + async fn email_signature_subscriber_registers_on_channel_independent_path() { + init_global(DEFAULT_CAPACITY); + assert!( + crate::openhuman::learning::extract::signature::register_email_signature_subscriber() + .is_some(), + "email-signature subscriber must register on the global bus once it is \ + initialised, independent of any channel configuration (#5003)" ); } diff --git a/src/openhuman/web_chat/event_bus.rs b/src/openhuman/web_chat/event_bus.rs index 0afdbe7646..ff525518ef 100644 --- a/src/openhuman/web_chat/event_bus.rs +++ b/src/openhuman/web_chat/event_bus.rs @@ -505,6 +505,27 @@ mod tests { /// - `args.reason` and `args.source` echoed from the domain event #[tokio::test] async fn automation_halt_subscriber_handle_publishes_correct_payload() { + // The web-channel bus is a process-shared broadcast; under the parallel test + // suite other tests publish to it too, so take THIS subscriber's next + // `automation_halt` event rather than whatever happens to be first in the + // buffer (which flaked as an "event name mismatch"). + fn next_automation_halt( + rx: &mut tokio::sync::broadcast::Receiver, + ) -> WebChannelEvent { + use tokio::sync::broadcast::error::TryRecvError; + loop { + match rx.try_recv() { + Ok(ev) if ev.event == "automation_halt" => return ev, + Ok(_) => {} // foreign event from a concurrent test + Err(TryRecvError::Lagged(_)) => {} // skipped some; keep draining + Err(TryRecvError::Empty) => { + panic!("expected an automation_halt WebChannelEvent but none was published") + } + Err(TryRecvError::Closed) => panic!("web-channel bus closed"), + } + } + } + // Subscribe BEFORE calling handle so the broadcast receiver is created // before any event is sent (broadcast channels only buffer messages // sent AFTER the receiver subscribed). @@ -519,9 +540,7 @@ mod tests { }) .await; - let halted = rx - .try_recv() - .expect("AutomationHalted must publish a WebChannelEvent"); + let halted = next_automation_halt(&mut rx); assert_eq!( halted.event, "automation_halt", "event name mismatch for halted" @@ -553,9 +572,7 @@ mod tests { }) .await; - let resumed = rx - .try_recv() - .expect("AutomationResumed must publish a WebChannelEvent"); + let resumed = next_automation_halt(&mut rx); assert_eq!( resumed.event, "automation_halt", "event name mismatch for resumed" From e8724814c5a91f5dabf494b650488b36472c7177 Mon Sep 17 00:00:00 2001 From: shanu Date: Mon, 20 Jul 2026 18:34:14 +0530 Subject: [PATCH 4/4] style: rustfmt the web_chat isolation test comment spacing --- src/openhuman/web_chat/event_bus.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/openhuman/web_chat/event_bus.rs b/src/openhuman/web_chat/event_bus.rs index ff525518ef..5d7ea964dd 100644 --- a/src/openhuman/web_chat/event_bus.rs +++ b/src/openhuman/web_chat/event_bus.rs @@ -516,7 +516,7 @@ mod tests { loop { match rx.try_recv() { Ok(ev) if ev.event == "automation_halt" => return ev, - Ok(_) => {} // foreign event from a concurrent test + Ok(_) => {} // foreign event from a concurrent test Err(TryRecvError::Lagged(_)) => {} // skipped some; keep draining Err(TryRecvError::Empty) => { panic!("expected an automation_halt WebChannelEvent but none was published")