feat(test-hooks): PR-1 — event-based-waits foundation

crates/actor/src/broker.rs

@@ -29,6 +29,20 @@ impl<T: Message<Result = ()>> Message for BrokerSubscribe<T> {
     type Result = SubscriptionId;
 }
 
+/// Fire-and-forget subscribe. Behaves like [`BrokerSubscribe`] but returns
+/// `()` so callers in synchronous contexts can use [`crate::Addr::do_send`]
+/// to enqueue the subscription without awaiting confirmation.
+///
+/// Because the broker's mailbox is FIFO, any [`Publish`] enqueued after
+/// this call is processed after the subscription is registered — no
+/// events are lost as long as no publish was enqueued before this call
+/// returns.
+pub struct BrokerAttach<T: Message<Result = ()>>(pub Recipient<T>);
+
+impl<T: Message<Result = ()>> Message for BrokerAttach<T> {
+    type Result = ();
+}
+
 /// Unsubscribe by ID.
 pub struct BrokerUnsubscribe(pub SubscriptionId);
 
@@ -90,6 +104,19 @@ impl<T: Message<Result = ()> + Clone> Handler<BrokerSubscribe<T>> for Broker<T>
     }
 }
 
+impl<T: Message<Result = ()> + Clone> Handler<BrokerAttach<T>> for Broker<T> {
+    fn handle(
+        &mut self,
+        msg: BrokerAttach<T>,
+        _ctx: &mut Context<Self>,
+    ) -> impl Future<Output = ()> + Send {
+        let id = SubscriptionId(self.next_id);
+        self.next_id += 1;
+        self.subscribers.push((id, msg.0));
+        async {}
+    }
+}
+
 impl<T: Message<Result = ()> + Clone> Handler<BrokerUnsubscribe> for Broker<T> {
     fn handle(
         &mut self,

crates/actor/src/lib.rs

@@ -42,7 +42,9 @@ pub mod system;
 
 pub use actor::{Actor, Handler, Message, StreamHandler};
 pub use addr::{Addr, AnyAddr, Recipient};
-pub use broker::{Broker, BrokerSubscribe, BrokerUnsubscribe, Publish, SubscriptionId};
+pub use broker::{
+    Broker, BrokerAttach, BrokerSubscribe, BrokerUnsubscribe, Publish, SubscriptionId,
+};
 pub use context::{Context, IntervalHandle, TimerHandle};
 pub use debounce::{Debounce, Enqueue, Throttle};
 pub use derived::{derived, DeriveSource, DerivedActor};

crates/client/Cargo.toml

@@ -7,6 +7,11 @@ description = "UI-agnostic client library for the Willow P2P chat network"
 
 [features]
 test-utils = ["willow-network/test-utils"]
+# Read-only test instrumentation hooks. Distinct from `test-utils`:
+# `test-hooks` exposes narrow accessors for the web crate's
+# `WillowTestHooks` (Promise-returning JS pull API) without enabling
+# `MemNetwork`. See `docs/specs/2026-04-27-event-based-waits-design.md`.
+test-hooks = []
 
 [dependencies]
 willow-actor = { path = "../actor" }

crates/client/src/accessors.rs

@@ -208,3 +208,31 @@ impl<N: willow_network::Network> ClientHandle<N> {
         .await
     }
 }
+
+// ── Test-only address getters (test-hooks feature) ────────────────────────
+//
+// Gated behind `test-hooks` so non-test consumers (`willow-agent`,
+// `willow-replay`, etc.) never see them. The address itself doesn't grant
+// write access without an active mutator — these are a read-only handle
+// for `WillowTestHooks` in the web crate, which cannot hold a generic
+// `ClientHandle<N>` across the wasm_bindgen boundary.
+
+#[cfg(feature = "test-hooks")]
+impl<N: willow_network::Network> ClientHandle<N> {
+    /// Clone the per-author Merkle-DAG actor address. Test-only.
+    pub fn dag_addr_clone(
+        &self,
+    ) -> willow_actor::Addr<willow_actor::StateActor<crate::state_actors::DagState>> {
+        self.dag_addr.clone()
+    }
+
+    /// Clone the materialised `ServerState` actor address. Test-only.
+    ///
+    /// Used by the snapshot builder in `WillowTestHooks` to read the
+    /// channels view for assertion-style polling.
+    pub fn event_state_addr_clone(
+        &self,
+    ) -> willow_actor::Addr<willow_actor::StateActor<willow_state::ServerState>> {
+        self.event_state_addr.clone()
+    }
+}

crates/client/src/lib.rs

@@ -123,7 +123,7 @@ pub enum ClientError {
 /// Helper to bridge `Broker<ClientEvent>` into an async stream receiver.
 pub mod event_receiver {
     use crate::events::ClientEvent;
-    use willow_actor::{Actor, Addr, Broker, BrokerSubscribe, Context, Handler};
+    use willow_actor::{Actor, Addr, Broker, BrokerAttach, BrokerSubscribe, Context, Handler};
 
     /// Async receiver for [`ClientEvent`]s from a [`Broker`].
     ///
@@ -147,6 +147,31 @@ pub mod event_receiver {
             Self { rx }
         }
 
+        /// Subscribe synchronously without awaiting confirmation.
+        ///
+        /// Queues `BrokerSubscribe` via `do_send`, returning the receiver
+        /// immediately. The broker's mailbox is FIFO, so any `Publish`
+        /// queued AFTER this call is processed AFTER the subscription
+        /// is registered — no events are lost as long as no publish has
+        /// been queued before this call returns.
+        ///
+        /// Use from synchronous contexts (mount blocks, `Drop`, etc.)
+        /// where awaiting the async [`subscribe`] would create a window
+        /// in which events emitted between scheduling and confirmation
+        /// would be missed.
+        ///
+        /// [`subscribe`]: Self::subscribe
+        pub fn subscribe_now(
+            broker: &Addr<Broker<ClientEvent>>,
+            system: &willow_actor::SystemHandle,
+        ) -> Self {
+            let (tx, rx) =
+                willow_actor::runtime::channel(willow_actor::runtime::DEFAULT_MAILBOX_CAPACITY);
+            let addr = system.spawn(ForwarderActor { tx });
+            broker.do_send(BrokerAttach(addr.into())).ok();
+            Self { rx }
+        }
+
         /// Await the next event. Returns `None` if the broker is closed.
         pub async fn recv(&mut self) -> Option<ClientEvent> {
             self.rx.recv().await
@@ -2127,4 +2152,41 @@ mod tests {
             "unmute must remove the channel from the set"
         );
     }
+
+    /// `subscribe_now` registers the recipient before any subsequent
+    /// `Publish` is processed by the broker, eliminating the race where
+    /// the async `subscribe` would miss events emitted between scheduling
+    /// and confirmation.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn subscribe_now_captures_event_published_synchronously_after_subscribe() {
+        use crate::events::ClientEvent;
+        use willow_actor::{Broker, Publish, System};
+
+        let sys = System::new();
+        let broker = sys.spawn(Broker::<ClientEvent>::default());
+
+        // Subscribe + publish without any await between them. The async
+        // `EventReceiver::subscribe` would create a window here in which
+        // the publish could overtake the BrokerSubscribe registration.
+        let mut rx = EventReceiver::subscribe_now(&broker, &sys.handle());
+        broker
+            .do_send(Publish(ClientEvent::SyncCompleted { ops_applied: 7 }))
+            .unwrap();
+
+        let received = tokio::time::timeout(std::time::Duration::from_secs(1), async {
+            loop {
+                if let Some(event) = rx.try_recv() {
+                    return event;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("subscribe_now must capture events published synchronously after subscription");
+
+        match received {
+            ClientEvent::SyncCompleted { ops_applied } => assert_eq!(ops_applied, 7),
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
 }

crates/web/Cargo.toml

@@ -52,7 +52,12 @@ futures = { workspace = true }
 gloo-timers = { version = "0.4", features = ["futures"] }
 serde = { workspace = true }
 serde_json = "1"
+serde-wasm-bindgen = { version = "0.6", optional = true }
 blake3 = { workspace = true }
 
 [dev-dependencies]
 wasm-bindgen-test = "0.3"
+
+[features]
+default = []
+test-hooks = ["dep:serde-wasm-bindgen", "willow-client/test-hooks"]

crates/web/src/app.rs

@@ -153,6 +153,31 @@ pub fn App() -> impl IntoView {
     let handle_inner = (*handle).clone().with_trust_store(trust_store.clone());
     let handle: WebClientHandle = SendWrapper::new(handle_inner);
 
+    #[cfg(feature = "test-hooks")]
+    {
+        let inner_for_hooks = (*handle).clone();
+        let hooks = crate::test_hooks::WillowTestHooks::new(&inner_for_hooks);
+        if let Some(window) = web_sys::window() {
+            let _ = js_sys::Reflect::set(
+                &window,
+                &"__willow".into(),
+                &wasm_bindgen::JsValue::from(hooks),
+            );
+        }
+        // Subscribe synchronously: any ClientEvent published after this
+        // call is guaranteed to land in the dispatcher (broker mailbox is
+        // FIFO). An async subscribe would create a window between mount
+        // and confirmation in which boot-time events would be lost.
+        let rx = willow_client::event_receiver::EventReceiver::subscribe_now(
+            inner_for_hooks.event_broker(),
+            inner_for_hooks.system(),
+        );
+        let dispatcher = crate::test_hooks::install_push_dispatcher(rx);
+        // Leak: dispatcher must live for app lifetime; in wasm32 the
+        // process IS the app, so leaking is fine.
+        std::mem::forget(dispatcher);
+    }
+
     // Provide context so child components can access the handle and state.
     provide_context(handle.clone());
     provide_context(app_state);

crates/web/src/lib.rs

@@ -20,6 +20,8 @@ pub mod profile;
 pub mod service_worker_bridge;
 pub mod state;
 pub mod state_bridge;
+#[cfg(feature = "test-hooks")]
+pub mod test_hooks;
 pub mod trust_store;
 pub mod util;
 pub mod voice;