agent_engine/extensions/notify_router.rs
1//! Daemon-mode phase 2 (C3): extension notification router.
2//!
3//! One subscriber per loaded extension (`ProcessExtension::subscribe_notifications`),
4//! started ONCE by the host after discovery; forwards `widget.*` frames
5//! ([`super::widgets::is_widget_method`]) to every live session as
6//! `SessionEventWire::ExtensionNotification` via
7//! [`crate::host::EngineHost::broadcast_extension_notification`].
8//!
9//! Routing (phase 3, C2): a frame whose `params.session_id` is a string is
10//! delivered to THAT session only (dropped with a `debug!` if it is not
11//! live); a frame without one is daemon-global and goes to every session
12//! (the pre-phase-3 behaviour — `docs/extensions/session-id.md` §plugin
13//! contract). Delivery is non-blocking: a session whose
14//! command queue is full drops the frame with a `warn!` — widget upserts are
15//! idempotent last-writer-wins UI state (`loop_arms.rs` semantics), and
16//! blocking would backpressure the extension's lossless fan-out and stall
17//! `command.invoke` / `provider.stream` subscribers on the same queue.
18//!
19//! The inline TUI keeps its own watcher (`loop_arms.rs`) today; this router
20//! serves actor-hosted sessions (daemon, `synaps chat` on the actor).
21
22use std::sync::Arc;
23use std::time::Duration;
24
25use crate::host::EngineHost;
26
27/// Resubscribe delay after an extension's notification channel closes
28/// (EOF / restart) — mirrors the TUI watcher.
29const RESUBSCRIBE_DELAY: Duration = Duration::from_millis(500);
30
31/// Start one forwarding task per loaded extension. Idempotent per call
32/// site by construction — the host calls it once after
33/// `extensions_ready()`; call it again only after a genuine re-discovery.
34/// The returned handle completes once every per-extension task has been
35/// spawned (the tasks themselves run for the host's lifetime and end when
36/// their extension is gone and stays gone).
37pub fn spawn_notification_router(host: Arc<EngineHost>) -> tokio::task::JoinHandle<()> {
38 tokio::spawn(async move {
39 let handlers = host.ext_manager().read().await.handlers();
40 for (ext_id, handler) in handlers {
41 let host = Arc::clone(&host);
42 tokio::spawn(async move {
43 forward_extension_notifications(host, ext_id, handler).await;
44 });
45 }
46 })
47}
48
49/// Forward loop for ONE extension. Exits when the extension has been
50/// unloaded from the manager (its channel closed and it is no longer
51/// listed); otherwise resubscribes after [`RESUBSCRIBE_DELAY`].
52pub async fn forward_extension_notifications(
53 host: Arc<EngineHost>,
54 ext_id: String,
55 handler: Arc<dyn super::runtime::ExtensionHandler>,
56) {
57 loop {
58 let (_sub_id, mut rx) = handler.subscribe_notifications().await;
59 while let Some(frame) = rx.recv().await {
60 if !super::widgets::is_widget_method(&frame.method) {
61 continue;
62 }
63 if super::widgets::parse_widget_event(&frame.method, &frame.params).is_err() {
64 tracing::debug!(extension = %ext_id, method = %frame.method,
65 "ignoring malformed widget frame");
66 continue;
67 }
68 let delivered = match route_target(&frame.params) {
69 Some(sid) => {
70 let id = crate::session::SessionId(sid.to_string());
71 match host.attach(&id) {
72 Some(handle) => {
73 let cmd = crate::session::SessionCommand::HostEvent(
74 crate::session::HostEvent::ExtensionNotification {
75 extension_id: ext_id.clone(),
76 method: frame.method.clone(),
77 params: frame.params,
78 },
79 );
80 match handle.send(cmd).await {
81 Ok(()) => 1,
82 Err(err) => {
83 tracing::warn!(session = %id, extension = %ext_id,
84 method = %frame.method, error = %err,
85 "dropping extension notification for session");
86 0
87 }
88 }
89 }
90 None => {
91 tracing::debug!(session = %id, extension = %ext_id,
92 method = %frame.method,
93 "extension notification for a session that is not live; dropped");
94 0
95 }
96 }
97 }
98 None => {
99 host.broadcast_extension_notification(&ext_id, &frame.method, frame.params)
100 .await
101 }
102 };
103 tracing::trace!(extension = %ext_id, method = %frame.method, sessions = delivered,
104 "extension notification routed");
105 }
106 // Channel closed (EOF/restart). Stop if the extension is gone for
107 // good; otherwise wait and resubscribe.
108 tokio::time::sleep(RESUBSCRIBE_DELAY).await;
109 let still_loaded = host
110 .ext_manager()
111 .read()
112 .await
113 .handlers()
114 .iter()
115 .any(|(id, h)| id == &ext_id && Arc::ptr_eq(h, &handler));
116 if !still_loaded {
117 tracing::debug!(extension = %ext_id, "notification router: extension unloaded; stopping");
118 return;
119 }
120 }
121}
122
123/// `params.session_id` when it is a non-empty string — the frame is
124/// session-scoped; otherwise daemon-global (broadcast).
125pub fn route_target(params: &serde_json::Value) -> Option<&str> {
126 params
127 .get("session_id")
128 .and_then(|v| v.as_str())
129 .filter(|s| !s.is_empty())
130}
131
132#[cfg(test)]
133mod tests {
134 use super::route_target;
135
136 #[test]
137 fn route_target_is_session_id_string_or_global() {
138 assert_eq!(
139 route_target(&serde_json::json!({"id": "w", "session_id": "s-1"})),
140 Some("s-1")
141 );
142 assert_eq!(route_target(&serde_json::json!({"id": "w"})), None);
143 assert_eq!(route_target(&serde_json::json!({"session_id": ""})), None);
144 assert_eq!(route_target(&serde_json::json!({"session_id": 7})), None);
145 assert_eq!(route_target(&serde_json::json!(null)), None);
146 }
147}