telos_agent/integrations/event_channel/
mod.rs

1//! Bidirectional HTTP event channel for the agent session.
2//!
3//! The channel runs an embedded HTTP server that provides two endpoints:
4//!
5//! - `POST /inject` — External callers push events into the agent context.
6//!   The event payload is injected as a system message into the conversation.
7//! - `GET /events` — External callers subscribe to the agent's TurnEvent
8//!   stream via Server-Sent Events (SSE).
9//!
10//! Topics use glob patterns for subscription matching (e.g. `"github.*"`,
11//! `"ci.*"`, `"*"`).
12
13use std::net::SocketAddr;
14use std::sync::Arc;
15
16use axum::extract::State;
17use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
18use axum::routing::{get, post};
19use axum::{Json, Router};
20use futures_core::stream::Stream;
21use serde::{Deserialize, Serialize};
22use tokio::sync::{broadcast, mpsc, oneshot};
23use tracing;
24
25use crate::agent::turn::TurnEvent;
26use crate::error::AgentError;
27use crate::model::message::Message;
28
29// ── configuration ───────────────────────────────────────────────────────────
30
31/// Top-level configuration for the embedded HTTP event channel.
32///
33/// Add this to [`AgentConfig`](crate::AgentConfig) via the
34/// [`event_channel`](crate::AgentConfig::event_channel) field.
35#[derive(Debug, Clone, Serialize, Deserialize)]
36pub struct EventChannelConfig {
37    /// When `false` (default) the HTTP server is not started.
38    pub enabled: bool,
39    /// The socket address the HTTP server listens on.
40    #[serde(default = "default_listen_addr")]
41    pub listen: SocketAddr,
42    /// Glob patterns matching topics the agent should accept. An incoming
43    /// event whose `topic` matches at least one pattern is injected into the
44    /// conversation context.
45    #[serde(default)]
46    pub subscriptions: Vec<Subscription>,
47}
48
49/// A single topic subscription pattern.
50#[derive(Debug, Clone, Serialize, Deserialize)]
51pub struct Subscription {
52    /// Glob pattern (e.g. `"github.*"`, `"ci.alerts"`). An empty string or
53    /// `"*"` matches every topic.
54    pub topic: String,
55}
56
57fn default_listen_addr() -> SocketAddr {
58    SocketAddr::from(([127, 0, 0, 1], 9090))
59}
60
61impl Default for EventChannelConfig {
62    fn default() -> Self {
63        Self { enabled: false, listen: default_listen_addr(), subscriptions: Vec::new() }
64    }
65}
66
67// ── external event ──────────────────────────────────────────────────────────
68
69/// An event pushed from outside into the agent session.
70#[derive(Debug, Clone, Serialize, Deserialize)]
71pub struct ExternalEvent {
72    /// Topic used for subscription matching.
73    pub topic: String,
74    /// Free-form payload text injected into the agent's context.
75    pub payload: String,
76}
77
78// ── internal types ──────────────────────────────────────────────────────────
79
80/// Shared state held by the axum router.
81#[derive(Clone)]
82struct AppState {
83    inject_tx: mpsc::Sender<ExternalEvent>,
84    broadcast_tx: broadcast::Sender<TurnEvent>,
85}
86
87#[derive(Deserialize)]
88struct InjectBody {
89    topic: String,
90    payload: String,
91}
92
93#[derive(Serialize)]
94struct InjectResponse {
95    accepted: bool,
96    topic: String,
97    message: String,
98}
99
100// ── EventChannel ────────────────────────────────────────────────────────────
101
102/// A running bidirectional HTTP event channel.
103///
104/// Created via [`EventChannel::start`] and stored in
105/// [`AgentSession`](crate::AgentSession). The channel is torn down when
106/// `EventChannel` is dropped (the server graceful-shutdown completes).
107pub struct EventChannel {
108    inject_rx: mpsc::Receiver<ExternalEvent>,
109    broadcast_tx: broadcast::Sender<TurnEvent>,
110    subscriptions: Vec<Subscription>,
111    _server: tokio::task::JoinHandle<()>,
112    _shutdown: oneshot::Sender<()>,
113}
114
115impl EventChannel {
116    /// Bind the HTTP server on [`EventChannelConfig::listen`] and spawn the
117    /// event-loop in a background tokio task. Returns `Ok(None)` when
118    /// `config.enabled` is `false`, or an error if binding fails.
119    ///
120    /// This is a **synchronous** method — it uses a std `TcpListener` to bind,
121    /// then converts the socket to a tokio listener.
122    pub fn start(config: EventChannelConfig) -> Result<Option<Self>, AgentError> {
123        if !config.enabled {
124            return Ok(None);
125        }
126
127        let addr = config.listen;
128
129        let std_listener = std::net::TcpListener::bind(addr)
130            .map_err(|e| AgentError::Config(format!("EventChannel: cannot bind to {addr}: {e}")))?;
131        std_listener
132            .set_nonblocking(true)
133            .map_err(|e| AgentError::Config(format!("EventChannel: set_nonblocking: {e}")))?;
134        let listener = tokio::net::TcpListener::from_std(std_listener)
135            .map_err(|e| AgentError::Config(format!("EventChannel: from_std: {e}")))?;
136
137        let (inject_tx, inject_rx) = mpsc::channel(256);
138        let (broadcast_tx, _) = broadcast::channel(256);
139        let (shutdown_tx, shutdown_rx) = oneshot::channel();
140
141        let app_state = Arc::new(AppState { inject_tx, broadcast_tx: broadcast_tx.clone() });
142
143        let app = Router::new()
144            .route("/health", get(handle_health))
145            .route("/inject", post(handle_inject))
146            .route("/events", get(handle_events))
147            .with_state(app_state);
148
149        let server = tokio::spawn(async move {
150            tracing::info!("EventChannel HTTP server listening on {addr}");
151            axum::serve(listener, app)
152                .with_graceful_shutdown(async {
153                    let _ = shutdown_rx.await;
154                })
155                .await
156                .ok();
157        });
158
159        tracing::info!("EventChannel started on {addr}");
160
161        Ok(Some(Self {
162            inject_rx,
163            broadcast_tx,
164            subscriptions: config.subscriptions,
165            _server: server,
166            _shutdown: shutdown_tx,
167        }))
168    }
169
170    /// Publish a [`TurnEvent`] to all connected SSE clients.
171    pub fn publish(&self, event: &TurnEvent) {
172        let _ = self.broadcast_tx.send(event.clone());
173    }
174
175    /// Non-blocking drain of all external events whose topic matches a
176    /// configured subscription pattern.
177    pub fn try_drain_incoming(&mut self) -> Vec<ExternalEvent> {
178        let mut events = Vec::new();
179        while let Ok(event) = self.inject_rx.try_recv() {
180            if self.topic_matches(&event.topic) {
181                events.push(event);
182            }
183        }
184        events
185    }
186
187    /// Convert a drained [`ExternalEvent`] into a system message ready for
188    /// injection into the conversation.
189    pub fn to_system_message(event: &ExternalEvent) -> Message {
190        Message::system(format!(
191            "<external-event topic=\"{}\">\n{}\n</external-event>",
192            event.topic, event.payload
193        ))
194    }
195
196    fn topic_matches(&self, topic: &str) -> bool {
197        if self.subscriptions.is_empty() {
198            return false;
199        }
200        self.subscriptions.iter().any(|sub| {
201            // Treat "*" as catch-all.
202            if sub.topic == "*" || sub.topic.is_empty() {
203                return true;
204            }
205            glob::Pattern::new(&sub.topic).map(|pat| pat.matches(topic)).unwrap_or(false)
206        })
207    }
208}
209
210// ── HTTP handlers ───────────────────────────────────────────────────────────
211
212async fn handle_health() -> &'static str {
213    "ok"
214}
215
216async fn handle_inject(
217    State(state): State<Arc<AppState>>,
218    Json(body): Json<InjectBody>,
219) -> Json<InjectResponse> {
220    let event = ExternalEvent { topic: body.topic.clone(), payload: body.payload };
221
222    match state.inject_tx.send(event).await {
223        Ok(()) => Json(InjectResponse {
224            accepted: true,
225            topic: body.topic,
226            message: "event injected into agent context".into(),
227        }),
228        Err(_) => Json(InjectResponse {
229            accepted: false,
230            topic: body.topic,
231            message: "agent channel closed".into(),
232        }),
233    }
234}
235
236fn make_sse_stream(
237    mut rx: broadcast::Receiver<TurnEvent>,
238) -> impl Stream<Item = Result<SseEvent, std::convert::Infallible>> {
239    async_stream::stream! {
240        loop {
241            match rx.recv().await {
242                Ok(event) => {
243                    let json = serde_json::to_string(&event).unwrap_or_default();
244                    yield Ok(SseEvent::default().data(json));
245                }
246                Err(broadcast::error::RecvError::Lagged(n)) => {
247                    tracing::warn!(n, "SSE client lagged; skipping events");
248                    rx = rx.resubscribe();
249                    continue;
250                }
251                Err(broadcast::error::RecvError::Closed) => break,
252            }
253        }
254    }
255}
256
257async fn handle_events(
258    State(state): State<Arc<AppState>>,
259) -> Sse<impl Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
260    let rx = state.broadcast_tx.subscribe();
261    Sse::new(make_sse_stream(rx)).keep_alive(KeepAlive::default())
262}
telos_agent/integrations/event_channel/mod.rs

telos_agent/integrations/event_channel/
mod.rs
