agent_block_core/bus/
event.rs

1//! Event type flowing through the [`EventBus`](crate::bus::EventBus).
2//!
3//! Each [`Event`] carries a `kind` string used for handler dispatch, an `id`
4//! for correlation/logging, a `payload` (source-defined), a `meta` map
5//! (source-defined), and an optional `ack_tx` one-shot channel used to return
6//! the Lua handler's return value back to the source that produced the
7//! request (e.g. a mesh request/response round-trip).
8//!
9//! The `ack_tx` is `Option` because some sources are fire-and-forget
10//! (e.g. a future webhook broadcast) and do not need a response.
11
12use serde_json::Value;
13use tokio::sync::oneshot;
14
15use agent_block_types::error::BlockError;
16
17/// Result carried back to the originating source via [`Event::ack_tx`].
18pub type AckResult = Result<Value, BlockError>;
19
20/// Sender half of the ack channel. Carried inside [`Event`].
21pub type AckSender = oneshot::Sender<AckResult>;
22
23/// Receiver half of the ack channel. Held by whatever source produced the
24/// event and awaits the handler's return value.
25///
26/// Used by `Event::with_ack` callers — an adapter that pushes events in and
27/// waits for what the handler returned (webhook / WSS / timer). Kept exported
28/// for downstream consumers; no adapter in this crate uses it, because the
29/// mesh adapter in `host.rs` drives the ack loop directly.
30#[allow(dead_code)]
31pub type AckReceiver = oneshot::Receiver<AckResult>;
32
33/// A normalized event flowing through the bus.
34///
35/// Ownership: produced by a [`Source`](crate::bus::Source), moved through a
36/// bounded `mpsc::Sender<Event>` into the single dispatcher loop. The
37/// dispatcher consumes the `ack_tx` (via `Option::take`) to send the
38/// handler's return value back to the source.
39#[derive(Debug)]
40pub struct Event {
41    /// Dispatch key. Matched against `bus.on(kind, fn)` registrations.
42    pub kind: String,
43    /// Correlation id (source-assigned). Used in tracing/logging.
44    pub id: String,
45    /// Source-defined payload. Converted to Lua table at dispatch time.
46    pub payload: Value,
47    /// Source-defined metadata (e.g. mesh `from`, timestamps). Converted to
48    /// Lua table at dispatch time.
49    pub meta: Value,
50    /// Optional one-shot channel used to return the Lua handler's result.
51    /// `None` for fire-and-forget sources.
52    pub ack_tx: Option<AckSender>,
53}
54
55impl Event {
56    /// Construct a new event without an ack channel (fire-and-forget).
57    ///
58    /// Intended for an adapter that broadcasts without waiting (webhook
59    /// broadcast / timer). Not used by the mesh path, which needs the ack
60    /// round-trip.
61    #[allow(dead_code)]
62    pub fn fire_and_forget(kind: impl Into<String>, id: impl Into<String>, payload: Value) -> Self {
63        Self {
64            kind: kind.into(),
65            id: id.into(),
66            payload,
67            meta: Value::Null,
68            ack_tx: None,
69        }
70    }
71
72    /// Construct a new event paired with a fresh ack channel. Returns the
73    /// event (to be pushed to the bus) and the receiver half (to be awaited
74    /// by the source).
75    ///
76    /// Used by the dispatcher's in-crate tests, and by an adapter that wants
77    /// the ack without assembling the channel itself. The mesh adapter
78    /// constructs `Event` directly instead, to keep control over the `meta`
79    /// map and the ack sender's lifetime.
80    #[allow(dead_code)]
81    pub fn with_ack(
82        kind: impl Into<String>,
83        id: impl Into<String>,
84        payload: Value,
85        meta: Value,
86    ) -> (Self, AckReceiver) {
87        let (tx, rx) = oneshot::channel();
88        let evt = Self {
89            kind: kind.into(),
90            id: id.into(),
91            payload,
92            meta,
93            ack_tx: Some(tx),
94        };
95        (evt, rx)
96    }
97
98    /// Send `result` on `ack_tx` if it is still present. Logs a warning when
99    /// the receiver has been dropped (tracing-missing-on-err policy).
100    ///
101    /// Returns `Ok(())` when the ack was delivered or the event was
102    /// fire-and-forget. Returns `Err(BlockError::Bus)` only when the
103    /// receiver had been dropped — the caller can decide whether to treat
104    /// that as fatal.
105    pub fn deliver_ack(&mut self, result: AckResult) -> Result<(), BlockError> {
106        let Some(tx) = self.ack_tx.take() else {
107            return Ok(());
108        };
109        if let Err(dropped) = tx.send(result) {
110            tracing::warn!(
111                kind = %self.kind,
112                id = %self.id,
113                "ack receiver dropped; handler result discarded: {:?}",
114                dropped.as_ref().map(|_| "ok").unwrap_or_else(|e| match e {
115                    BlockError::Bus(_) => "bus-err",
116                    _ => "other-err",
117                })
118            );
119            return Err(BlockError::Bus(format!(
120                "ack receiver dropped (kind={}, id={})",
121                self.kind, self.id
122            )));
123        }
124        Ok(())
125    }
126}
agent_block_core/bus/event.rs

agent_block_core/bus/
event.rs
