feat(control-plane)!: /ws/events says when it is live, over an ordered in-process event bus

.github/workflows/test.yml

@@ -79,7 +79,14 @@ jobs:
       # and cannot live in the proc-macro crate's own lib tests.
       # streamlib-engine's lib tests (host-body tier-1 + engine twin) are a
       # tracked follow-up, pending a fix to a parallel-run test flake.
+      # streamlib-api-server carries the control plane's route-surface, auth-gate
+      # and WebSocket frame locks, and ran nowhere in CI before.
+      # core::pubsub is named explicitly for the same reason python-wheel.yml
+      # names core::signals: the engine lib tests are excluded wholesale, and
+      # without this the subscription-live locks run nowhere.
       - name: Run unit tests
         run: |
           cargo test --locked -p streamlib -p streamlib-macros --lib
           cargo test --locked -p streamlib-engine --test attribute_macro_test
+          cargo test --locked -p streamlib-api-server --lib
+          cargo test --locked -p streamlib-engine --lib -- core::pubsub

runtime/streamlib-api-server/Cargo.toml

@@ -62,3 +62,8 @@ streamlib = { path = "../../sdk/streamlib-sdk", version = "0.17.0", features = [
 tempfile = "3"
 tower = {version = "0.5", features = ["util"]}
 serial_test = "3.2"
+# Real WebSocket client for the /ws/events frame-ordering test: `tower::oneshot`
+# drives a router but never completes an upgrade, so it cannot observe frames.
+# Pinned to the generation axum itself speaks, so the test client and the server
+# under test share one tungstenite rather than forking a second into the lock.
+tokio-tungstenite = "0.29"

runtime/streamlib-api-server/src/control_plane_stub_support.rs

@@ -87,3 +87,21 @@ macro_rules! graph_mutation_ops_are_unreachable {
 }
 
 pub(crate) use graph_mutation_ops_are_unreachable;
+
+/// Initialize the process-global `PUBSUB` for this test binary, once.
+///
+/// `PUBSUB` is process-global and initialized through a `OnceLock`, so a test
+/// binary cannot hold both a live bus and a dead one — whichever a test sees
+/// depends on what ran before it. Every test whose behaviour depends on the bus
+/// calls this and carries `#[serial]`: the bus is always live, and a publish in
+/// one test can never land inside another's sample window.
+pub(crate) fn initialize_process_global_pubsub_for_tests() {
+    use std::sync::Once;
+    static INITIALIZED: Once = Once::new();
+
+    INITIALIZED.call_once(|| {
+        let node = ::streamlib::sdk::iceoryx2::Iceoryx2Node::new()
+            .expect("iceoryx2 node for the control-plane test bus");
+        ::streamlib::sdk::pubsub::PUBSUB.init("test-api-server-control-plane", node);
+    });
+}

runtime/streamlib-api-server/src/mcp.rs

@@ -37,7 +37,9 @@ use parking_lot::Mutex;
 use serde::Deserialize;
 use serde_json::{Value, json};
 use streamlib::sdk::error::Result;
-use streamlib::sdk::pubsub::{Event, EventListener, PUBSUB, topics};
+use streamlib::sdk::pubsub::{
+    DEFAULT_SUBSCRIPTION_LIVE_BUDGET, Event, EventListener, PUBSUB, topics,
+};
 use streamlib::sdk::runtime::RuntimeOperations;
 
 use crate::state::{AppState, RuntimeShutdownRequest};
@@ -342,7 +344,16 @@ async fn call_logs(runtime: &Arc<dyn RuntimeOperations>, arguments: Value) -> Va
 
     let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Event>();
     let listener = Arc::new(Mutex::new(McpEventForwarder { tx }));
-    PUBSUB.subscribe(topics::ALL, listener.clone());
+    let subscription_live_signal = PUBSUB.subscribe(topics::ALL, listener.clone());
+
+    // The sample window starts once the subscription can actually receive, so
+    // the window this tool reports back is the window it actually sampled.
+    if let Err(e) = subscription_live_signal
+        .wait_until_subscription_is_live_async(DEFAULT_SUBSCRIPTION_LIVE_BUDGET)
+        .await
+    {
+        return tool_error(format!("event subscription never went live: {e}"));
+    }
 
     let mut events: Vec<Value> = Vec::with_capacity(sample);
     let deadline = tokio::time::Instant::now() + LOGS_SAMPLE_WINDOW;
@@ -884,11 +895,20 @@ mod tests {
     }
 
     #[tokio::test]
+    #[serial_test::serial]
     async fn tools_call_logs_returns_bounded_window_sample() {
-        // Hermetic: PUBSUB is uninitialized here, so no event is delivered and
-        // the collection is bounded by the monotonic sample window, returning an
-        // empty sample rather than hanging. Live event delivery rides iceoryx2
-        // and is exercised by the engine's pubsub integration tests, not here.
+        // A live bus that nobody publishes to: the tool waits for its
+        // subscription to go live, then collects nothing, and the monotonic
+        // sample window bounds the wait rather than letting it hang. Live event
+        // delivery rides iceoryx2 and is exercised by the engine's pubsub
+        // integration tests, not here.
+        //
+        // The bus must be live for the empty sample to mean "the node was
+        // quiet" — against an absent bus the tool reports an error instead, so
+        // the two would be indistinguishable. `#[serial]` keeps another test's
+        // publish out of this window.
+        crate::control_plane_stub_support::initialize_process_global_pubsub_for_tests();
+
         let started = tokio::time::Instant::now();
         let (status, body) = mcp_call(
             Arc::new(ControlPlaneMcpDispatchStubRuntime::new()),
@@ -910,8 +930,12 @@ mod tests {
             sample["window_ms"].as_u64().unwrap(),
             LOGS_SAMPLE_WINDOW.as_millis() as u64
         );
+        // The budget names the subscription wait because the measured span now
+        // contains it: the tool waits for its subscription before the window
+        // starts, so a slow iceoryx2 open is time this assertion must allow
+        // rather than a hang it should catch.
         assert!(
-            elapsed < LOGS_SAMPLE_WINDOW * 4,
+            elapsed < LOGS_SAMPLE_WINDOW * 4 + DEFAULT_SUBSCRIPTION_LIVE_BUDGET,
             "logs must return within its sample window, not hang; took {elapsed:?}"
         );
     }