diff --git a/Cargo.lock b/Cargo.lock index 00911d8..ca7005d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -325,6 +325,7 @@ dependencies = [ "serde", "serde_json", "toml", + "uuid", ] [[package]] @@ -2051,6 +2052,17 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "valuable" version = "0.1.1" diff --git a/Documentation.md b/Documentation.md index 1520ab5..d589802 100644 --- a/Documentation.md +++ b/Documentation.md @@ -152,6 +152,7 @@ installed_at = "2026-01-01T00:00:00Z" ## Debugging tips - Run `bread events` to see live normalized events. +- Run `bread events --tree` *(Since: v1.5)* to render events as a causality tree instead of a flat stream — events that a Lua handler emitted via `bread.emit()` in reaction to another event are nested underneath it, following the `caused_by` chain (see [Dictionary: Event reference](#dictionary-event-reference)). Useful for untangling "why did this event fire" when several modules chain-react to each other. - Run `bread state` to see full runtime state as JSON. - Run `bread doctor` to check adapter and module health. - Log event payloads with `bread.log(tostring(event.data))`. @@ -213,7 +214,7 @@ end, { Unsubscribe an event handler or state watch by ID. #### `bread.emit(event, data)` -Emit a custom event into the system pipeline. Useful for cross-module communication. +Emit a custom event into the system pipeline. Useful for cross-module communication. If called synchronously from inside a `bread.on` subscriber callback (i.e. in reaction to a matched event), the emitted event's `caused_by` *(Since: v1.5)* is set to the id of the event that triggered the callback, threading causality across chains of modules that react to each other — see [Dictionary: Event reference](#dictionary-event-reference). #### `bread.wait(pattern, opts) -> event | nil` Coroutine-only helper that suspends until a matching event arrives. @@ -855,10 +856,15 @@ Events are delivered as a `BreadEvent`: "event": "bread.device.dock.connected", "timestamp": 1710000000000, "source": "Udev", - "data": {} + "data": {}, + "id": "b3f2c9a0-4e6d-4b8a-9c1e-7a2f5d8e0c11", + "caused_by": null } ``` +- **`id`** *(Since: v1.5)* — a unique id assigned to this specific event instance at construction. Every `BreadEvent`, regardless of origin (adapter-normalized, IPC `emit`, Lua `bread.emit()`, or a daemon-internal send like `bread.system.startup`), gets one. +- **`caused_by`** *(Since: v1.5)* — the `id` of the event whose Lua subscriber handler emitted this event via `bread.emit()`, or `null` if this event did not originate from inside a running handler (adapter events, IPC `emit`, daemon-internal sends). This lets you reconstruct causality chains across modules that react to each other's events: if module A's handler for event X calls `bread.emit("Y", ...)`, then Y's `caused_by` is X's `id`. See `bread events --tree` below for a rendering of these chains. + ### Pattern matching | Pattern | Matches | diff --git a/bread-cli/src/main.rs b/bread-cli/src/main.rs index 843e758..92cc0cf 100644 --- a/bread-cli/src/main.rs +++ b/bread-cli/src/main.rs @@ -6,6 +6,7 @@ use anyhow::Result; use clap::{Parser, Subcommand}; use notify::{RecommendedWatcher, RecursiveMode, Watcher}; use serde_json::{json, Value}; +use std::collections::HashMap; use std::env; use std::io::{self, Write as IoWrite}; use std::path::{Path, PathBuf}; @@ -54,6 +55,12 @@ enum Commands { /// Replay events from the last N seconds #[arg(long)] since: Option, + /// Render events as a causality tree via `caused_by` instead of a + /// flat stream (events emitted from inside a `bread.emit()` call + /// made by another event's Lua handler are nested under it). + /// Overrides `--json` — tree rendering always uses the formatted view. + #[arg(long)] + tree: bool, }, /// Manage installed Lua modules Modules { @@ -158,8 +165,9 @@ async fn main() -> Result<()> { json, fields, since, + tree, } => { - stream_events(&socket, pattern, json, fields, since).await?; + stream_events(&socket, pattern, json, fields, since, tree).await?; } Commands::Modules { subcommand } => { handle_modules_cmd(subcommand, &socket).await?; @@ -378,7 +386,13 @@ async fn stream_events( raw_json: bool, fields: Option, since: Option, + tree: bool, ) -> Result<()> { + // Tree rendering needs `id`/`caused_by` visible in a consistent shape, + // so it always uses the formatted view — a live-streaming-friendly + // indent-as-you-go tree rather than buffering the whole stream. + let mut causality = CausalityTracker::default(); + if let Some(seconds) = since { let replay = send_request( socket, @@ -388,7 +402,9 @@ async fn stream_events( .await?; if let Some(list) = replay.as_array() { for item in list { - if raw_json { + if tree { + causality.print(item); + } else if raw_json { println!("{}", serde_json::to_string_pretty(item)?); } else { print_event(item, fields.as_deref()); @@ -426,7 +442,9 @@ async fn stream_events( while let Some(line) = lines.next_line().await? { let value: Value = serde_json::from_str(&line)?; - if raw_json { + if tree { + causality.print(&value); + } else if raw_json { println!("{}", serde_json::to_string_pretty(&value)?); } else { print_event(&value, fields.as_deref()); @@ -436,6 +454,63 @@ async fn stream_events( Ok(()) } +/// Tracks `id` -> `caused_by` for events seen so far in this stream/replay +/// batch, so each new event can be indented under its parent as it arrives +/// — no buffering, no waiting for the stream to end. An event whose parent +/// hasn't been seen yet (e.g. the parent predates a replay window, or the +/// causing event is filtered out by the subscription pattern) is rendered +/// as its own root rather than blocking on a parent that may never show up. +#[derive(Default)] +struct CausalityTracker { + parents: HashMap>, +} + +impl CausalityTracker { + /// Depth = number of ancestors reachable by following `caused_by`. + /// Guards against cycles/self-loops (shouldn't happen, but a rendering + /// bug here must never hang the CLI) with both a seen-set and a hard cap. + fn depth(&self, id: &str) -> usize { + let mut depth = 0; + let mut current = id.to_string(); + let mut seen = std::collections::HashSet::new(); + while let Some(Some(parent)) = self.parents.get(¤t) { + if depth >= 64 || !seen.insert(current.clone()) { + break; + } + depth += 1; + current = parent.clone(); + } + depth + } + + fn print(&mut self, event: &Value) { + let id = event.get("id").and_then(Value::as_str).map(str::to_string); + let caused_by = event + .get("caused_by") + .and_then(Value::as_str) + .map(str::to_string); + + let depth = if let Some(id) = &id { + self.parents.insert(id.clone(), caused_by.clone()); + self.depth(id) + } else { + 0 + }; + + let ts = event.get("timestamp").and_then(Value::as_u64).unwrap_or(0); + let event_name = event.get("event").and_then(Value::as_str).unwrap_or("?"); + let source = event.get("source").and_then(Value::as_str).unwrap_or("?"); + let time = format_timestamp(ts); + let indent = " ".repeat(depth); + let connector = if depth > 0 { "\u{2514}\u{2500} " } else { "" }; + let id_display = id.as_deref().unwrap_or("?"); + println!("{indent}{connector}{time} {event_name} source={source} id={id_display}"); + if let Some(data) = event.get("data") { + println!("{indent} data: {data}"); + } + } +} + fn print_json(value: &Value) -> Result<()> { println!("{}", serde_json::to_string_pretty(value)?); Ok(()) diff --git a/bread-shared/Cargo.toml b/bread-shared/Cargo.toml index bf6ce66..7be3f15 100644 --- a/bread-shared/Cargo.toml +++ b/bread-shared/Cargo.toml @@ -8,3 +8,4 @@ serde.workspace = true serde_json.workspace = true dirs.workspace = true toml = "0.8" +uuid = { version = "1", features = ["v4"] } diff --git a/bread-shared/src/lib.rs b/bread-shared/src/lib.rs index 50a2f25..a84d88d 100644 --- a/bread-shared/src/lib.rs +++ b/bread-shared/src/lib.rs @@ -90,20 +90,65 @@ pub struct BreadEvent { pub source: AdapterSource, /// Structured event data. The shape depends on the event family. pub data: serde_json::Value, + /// Unique id for this specific event instance, assigned at construction. + /// + /// *Since: v1.5* — enables causality tracking (`caused_by`) across chains + /// of Lua modules that re-emit events from inside `bread.on` handlers. + pub id: String, + /// The `id` of the event whose Lua handler emitted this event via + /// `bread.emit()`, if any. + /// + /// `None` for events that originate outside any Lua handler invocation + /// (adapter-normalized events, IPC `emit`, daemon-internal sends like + /// `bread.system.startup` / `bread.profile.activated`). Populated only + /// when the event was constructed by `bread.emit()` while a subscriber + /// callback was synchronously running — see the "current dispatch id" + /// mechanism on `breadd`'s Lua engine. + /// + /// *Since: v1.5* + pub caused_by: Option, } impl BreadEvent { - /// Construct a new event with `timestamp` set to the current wall-clock. + /// Construct a new event with `timestamp` set to the current wall-clock, + /// a freshly generated `id`, and `caused_by` unset. pub fn new(event: impl Into, source: AdapterSource, data: serde_json::Value) -> Self { + Self::with_timestamp(event, now_unix_ms(), source, data) + } + + /// Construct a new event with an explicit `timestamp`, preserving the + /// originating signal's observed time instead of "now". Used by the + /// normalizer, which carries `RawEvent::timestamp` through unchanged. + /// + /// Like [`BreadEvent::new`], this always assigns a fresh `id` and leaves + /// `caused_by` unset — callers that need to thread causality set + /// `caused_by` on the returned value themselves. + pub fn with_timestamp( + event: impl Into, + timestamp: u64, + source: AdapterSource, + data: serde_json::Value, + ) -> Self { Self { event: event.into(), - timestamp: now_unix_ms(), + timestamp, source, data, + id: new_event_id(), + caused_by: None, } } } +/// Generate a fresh unique id for a [`BreadEvent`]. +/// +/// Every construction path (`BreadEvent::new`, `BreadEvent::with_timestamp`, +/// and any remaining struct-literal construction) calls this so every event +/// gets a stable identity to hang `caused_by` chains off of. +pub fn new_event_id() -> String { + uuid::Uuid::new_v4().to_string() +} + /// Current Unix epoch in milliseconds. /// /// Falls back to `0` if the system clock is before the epoch, which keeps @@ -316,6 +361,8 @@ mod tests { timestamp: 1_700_000_000_000, source: AdapterSource::Udev, data: json!({ "id": "usb-1-1.4", "name": "Logitech" }), + id: "test-id-1".to_string(), + caused_by: Some("test-id-0".to_string()), }; let raw = serde_json::to_string(&original).unwrap(); let decoded: BreadEvent = serde_json::from_str(&raw).unwrap(); @@ -324,6 +371,30 @@ mod tests { assert_eq!(decoded.timestamp, original.timestamp); assert_eq!(decoded.source, original.source); assert_eq!(decoded.data, original.data); + assert_eq!(decoded.id, original.id); + assert_eq!(decoded.caused_by, original.caused_by); + } + + #[test] + fn bread_event_new_assigns_unique_id_and_no_cause() { + let a = BreadEvent::new("bread.test.a", AdapterSource::System, json!({})); + let b = BreadEvent::new("bread.test.b", AdapterSource::System, json!({})); + assert!(!a.id.is_empty()); + assert_ne!(a.id, b.id, "each constructed event should get a unique id"); + assert_eq!(a.caused_by, None); + } + + #[test] + fn bread_event_with_timestamp_preserves_timestamp_and_assigns_id() { + let event = BreadEvent::with_timestamp( + "bread.test.c", + 42, + AdapterSource::Udev, + json!({ "x": 1 }), + ); + assert_eq!(event.timestamp, 42); + assert!(!event.id.is_empty()); + assert_eq!(event.caused_by, None); } #[test] diff --git a/breadd/src/core/normalizer.rs b/breadd/src/core/normalizer.rs index 96e40c8..942d3c0 100644 --- a/breadd/src/core/normalizer.rs +++ b/breadd/src/core/normalizer.rs @@ -43,6 +43,8 @@ impl EventNormalizer { event: raw.kind.clone(), timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }], }; @@ -143,6 +145,8 @@ impl EventNormalizer { event: format!("bread.device.{}", verb), timestamp: raw.timestamp, source: AdapterSource::Udev, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "id": id, "device": "unknown", @@ -173,36 +177,48 @@ impl EventNormalizer { event: "bread.workspace.changed".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }], "createworkspace" => vec![BreadEvent { event: "bread.workspace.created".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "workspace": data }), }], "destroyworkspace" => vec![BreadEvent { event: "bread.workspace.destroyed".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "workspace": data }), }], "monitoradded" => vec![BreadEvent { event: "bread.monitor.connected".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "name": data }), }], "monitorremoved" => vec![BreadEvent { event: "bread.monitor.disconnected".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "name": data }), }], "activewindow" => vec![BreadEvent { event: "bread.window.focus.changed".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }], "activewindowv2" => { @@ -211,6 +227,8 @@ impl EventNormalizer { event: "bread.window.focused".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "address": fields.first().unwrap_or(&"") }), @@ -222,6 +240,8 @@ impl EventNormalizer { event: "bread.window.opened".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "address": fields.first().unwrap_or(&""), "workspace": fields.get(1).unwrap_or(&""), @@ -236,6 +256,8 @@ impl EventNormalizer { event: "bread.window.closed".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "address": fields.first().unwrap_or(&"") }), }] } @@ -245,6 +267,8 @@ impl EventNormalizer { event: "bread.window.moved".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "address": fields.first().unwrap_or(&""), "workspace": fields.get(1).unwrap_or(&""), @@ -255,6 +279,8 @@ impl EventNormalizer { event: "bread.hyprland.event".to_string(), timestamp: raw.timestamp, source: AdapterSource::Hyprland, + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }], } @@ -272,6 +298,8 @@ impl EventNormalizer { }, timestamp: raw.timestamp, source: AdapterSource::Power, + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }); } @@ -294,6 +322,8 @@ impl EventNormalizer { event: event.to_string(), timestamp: raw.timestamp, source: AdapterSource::Power, + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }); } @@ -304,6 +334,8 @@ impl EventNormalizer { event: "bread.power.changed".to_string(), timestamp: raw.timestamp, source: AdapterSource::Power, + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }); } @@ -339,6 +371,8 @@ impl EventNormalizer { event: "bread.device.connected".to_string(), timestamp: raw.timestamp, source: AdapterSource::Bluetooth, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "id": path, "device": "unknown", @@ -352,6 +386,8 @@ impl EventNormalizer { event: "bread.device.disconnected".to_string(), timestamp: raw.timestamp, source: AdapterSource::Bluetooth, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "id": path, "device": "unknown", @@ -365,6 +401,8 @@ impl EventNormalizer { event: "bread.bluetooth.device.paired".to_string(), timestamp: raw.timestamp, source: AdapterSource::Bluetooth, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "id": path, "name": name, @@ -377,6 +415,8 @@ impl EventNormalizer { event: "bread.bluetooth.device.unpaired".to_string(), timestamp: raw.timestamp, source: AdapterSource::Bluetooth, + id: bread_shared::new_event_id(), + caused_by: None, data: json!({ "id": path, "address": address, @@ -421,6 +461,8 @@ impl EventNormalizer { event: name.to_string(), timestamp: raw.timestamp, source: AdapterSource::Network, + id: bread_shared::new_event_id(), + caused_by: None, data, }] } @@ -435,6 +477,8 @@ impl EventNormalizer { event: format!("bread.terminal.{}", raw.kind), timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }] } @@ -444,6 +488,8 @@ impl EventNormalizer { event: format!("bread.remote.{}", raw.kind), timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }] } @@ -453,6 +499,8 @@ impl EventNormalizer { event: format!("bread.git.{}", raw.kind), timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }] } @@ -462,6 +510,8 @@ impl EventNormalizer { event: format!("bread.project.{}", raw.kind), timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }] } @@ -474,6 +524,8 @@ impl EventNormalizer { event: format!("bread.service.{suffix}"), timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }] } @@ -490,6 +542,8 @@ impl EventNormalizer { event, timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }] } @@ -512,6 +566,8 @@ impl EventNormalizer { event: raw.kind.clone(), timestamp: raw.timestamp, source: raw.source.clone(), + id: bread_shared::new_event_id(), + caused_by: None, data: raw.payload.clone(), }] } diff --git a/breadd/src/core/state_engine.rs b/breadd/src/core/state_engine.rs index a88fd70..538eaf9 100644 --- a/breadd/src/core/state_engine.rs +++ b/breadd/src/core/state_engine.rs @@ -656,6 +656,8 @@ mod tests { timestamp: 0, source: AdapterSource::System, data, + id: bread_shared::new_event_id(), + caused_by: None, } } diff --git a/breadd/src/ipc/mod.rs b/breadd/src/ipc/mod.rs index e132770..10cdfdf 100644 --- a/breadd/src/ipc/mod.rs +++ b/breadd/src/ipc/mod.rs @@ -27,7 +27,7 @@ use crate::lua::RuntimeHandle; /// something new-but-additive (a binding, an event, an IPC param); bump the /// major version only for a breaking change, which should not happen inside /// this daemon's v1 lifetime per that section's stated policy. -const API_VERSION: &str = "1.4.0"; +const API_VERSION: &str = "1.5.0"; #[derive(Clone)] pub struct Server { diff --git a/breadd/src/lua/mod.rs b/breadd/src/lua/mod.rs index edbf437..fa0d3fc 100644 --- a/breadd/src/lua/mod.rs +++ b/breadd/src/lua/mod.rs @@ -211,6 +211,19 @@ struct LuaEngine { next_sub_id: Arc, next_timer_id: Arc, current_module: Arc>>, + /// The `id` of the `BreadEvent` whose subscriber callback is currently + /// executing, if any. Set immediately before invoking a handler in + /// `handle_event` and restored (not merely cleared) immediately after, + /// so nested/reentrant dispatch and a single event fan-out to multiple + /// subscriptions both see the correct parent id. Read synchronously by + /// `bread.emit()`'s binding to populate the outgoing event's + /// `caused_by`. The Lua engine runs single-threaded/cooperatively (one + /// engine instance processes `LuaMessage`s serially on a dedicated + /// thread), so a callback invocation and any `bread.emit()` inside it + /// happen synchronously within one `handle_event` call — no additional + /// synchronization beyond the existing `Mutex` (mirroring + /// `current_module` above) is needed. + current_dispatch_id: Arc>>, modules: Arc>>, module_decls: Arc>>, module_order: Arc>>, @@ -240,6 +253,7 @@ impl LuaEngine { next_sub_id: Arc::new(AtomicU64::new(1)), next_timer_id: Arc::new(AtomicU64::new(1)), current_module: Arc::new(Mutex::new(None)), + current_dispatch_id: Arc::new(Mutex::new(None)), modules: Arc::new(Mutex::new(HashMap::new())), module_decls: Arc::new(Mutex::new(HashMap::new())), module_order: Arc::new(Mutex::new(Vec::new())), @@ -438,6 +452,7 @@ impl LuaEngine { bread.set("off", off_fn)?; let emit_tx = self.emit_tx.clone(); + let current_dispatch_id = self.current_dispatch_id.clone(); let emit_fn = self.lua .create_function(move |lua, (event_name, payload): (String, Value)| { @@ -447,8 +462,19 @@ impl LuaEngine { .from_value::(other) .unwrap_or_else(|_| serde_json::json!({})), }; + // Daemon-internal emit — same trusted path as adapter/IPC + // construction, tagged System. If this runs synchronously + // inside a subscriber callback (see `handle_event`), thread + // the currently-dispatching event's id through as + // `caused_by` so the causality chain can be reconstructed. + let caused_by = current_dispatch_id + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + let mut event = BreadEvent::new(event_name, AdapterSource::System, data); + event.caused_by = caused_by; emit_tx - .send(BreadEvent::new(event_name, AdapterSource::System, data)) + .send(event) .map_err(|_| LuaError::external("event channel closed"))?; Ok(()) })?; @@ -1489,6 +1515,11 @@ impl LuaEngine { } self.set_current_module(module.clone()); + // Every subscriber invocation triggered by this event should see the + // same parent id, and a `bread.emit()` call made synchronously + // inside the callback should attribute its `caused_by` to *this* + // event, not whatever was dispatching (if anything) before it. + let previous_dispatch_id = self.set_current_dispatch_id(Some(event.id.clone())); let result = match kind { HandlerKind::Event => { let event_value = json_to_lua(&self.lua, &event)?; @@ -1502,6 +1533,9 @@ impl LuaEngine { callback.call::<_, ()>((new_lua, old_lua)) } }; + // Restore rather than clear, so this correctly unwinds if dispatch + // is ever reentrant (e.g. a callback that pumps messages itself). + self.set_current_dispatch_id(previous_dispatch_id); self.set_current_module(None); if let Err(err) = result { @@ -1678,6 +1712,18 @@ impl LuaEngine { } } + /// Set the "currently dispatching" event id, returning whatever id was + /// there before. Callers restore the previous value (rather than + /// clearing to `None`) when the handler invocation finishes, so nested/ + /// reentrant dispatch unwinds correctly — see `current_dispatch_id`'s + /// doc comment on the struct definition. + fn set_current_dispatch_id(&self, id: Option) -> Option { + match self.current_dispatch_id.lock() { + Ok(mut guard) => std::mem::replace(&mut *guard, id), + Err(poisoned) => std::mem::replace(&mut *poisoned.into_inner(), id), + } + } + fn cancel_all_timers(&self) { if let Ok(mut map) = self.timers.lock() { for (_, entry) in map.drain() { diff --git a/breadd/tests/ipc_integration.rs b/breadd/tests/ipc_integration.rs index 7e7129b..513e6e1 100644 --- a/breadd/tests/ipc_integration.rs +++ b/breadd/tests/ipc_integration.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::fs; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; @@ -521,6 +522,147 @@ async fn events_stream_receives_emitted_events() -> Result<()> { Ok(()) } +/// A chain of 3 handlers, each reacting to the previous one's emitted event +/// entirely inside the Lua runtime (no test-side pumping between links): +/// module A handles the external trigger and emits X, module B subscribes +/// to X and emits Y, module C subscribes to Y and emits Z. This exercises +/// the "current dispatch id" mechanism in `breadd::lua::LuaEngine` end to +/// end: `caused_by` must thread through every hop of the chain, not just a +/// single emit. +#[tokio::test] +async fn event_causality_chain_threads_caused_by_across_handlers() -> Result<()> { + let harness = TestHarness::spawn_with_init( + r#" + -- module A: reacts to the external trigger, emits X + bread.on("bread.chain.trigger", function(event) + bread.emit("bread.chain.x", {}) + end) + + -- module B: reacts to X, emits Y + bread.on("bread.chain.x", function(event) + bread.emit("bread.chain.y", {}) + end) + + -- module C: reacts to Y, emits Z + bread.on("bread.chain.y", function(event) + bread.emit("bread.chain.z", {}) + end) + "#, + )?; + harness.wait_until_ready().await?; + + let stream = UnixStream::connect(harness.socket_path()).await?; + let (read_half, mut write_half) = stream.into_split(); + let subscribe = json!({ + "id": "sub-chain", + "method": "events.subscribe", + "params": { "filter": "bread.chain.*" } + }); + write_half + .write_all(format!("{}\n", serde_json::to_string(&subscribe)?).as_bytes()) + .await?; + + let mut reader = BufReader::new(read_half).lines(); + let ack = reader + .next_line() + .await? + .ok_or_else(|| anyhow!("missing subscribe ack"))?; + let ack_json: Value = serde_json::from_str(&ack)?; + assert_eq!( + ack_json + .get("result") + .and_then(|v| v.get("subscribed")) + .and_then(Value::as_bool), + Some(true) + ); + + // The trigger itself comes in over IPC's unsourced `emit`, outside any + // Lua handler — its `caused_by` must be None. Everything downstream + // (X, Y, Z) is emitted by `bread.emit()` from inside a running handler. + harness + .send_request("emit", json!({ "event": "bread.chain.trigger", "data": {} })) + .await?; + + let mut events: HashMap = HashMap::new(); + let deadline = Instant::now() + Duration::from_secs(5); + while events.len() < 4 && Instant::now() < deadline { + let Some(line) = reader.next_line().await? else { + break; + }; + let event: Value = serde_json::from_str(&line)?; + let name = event + .get("event") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + if name.starts_with("bread.chain.") { + events.insert(name, event); + } + } + + let trigger = events + .get("bread.chain.trigger") + .expect("missing trigger event"); + let x = events.get("bread.chain.x").expect("missing X event"); + let y = events.get("bread.chain.y").expect("missing Y event"); + let z = events.get("bread.chain.z").expect("missing Z event"); + + let trigger_id = trigger + .get("id") + .and_then(Value::as_str) + .expect("trigger event missing id") + .to_string(); + let x_id = x + .get("id") + .and_then(Value::as_str) + .expect("X event missing id") + .to_string(); + let y_id = y + .get("id") + .and_then(Value::as_str) + .expect("Y event missing id") + .to_string(); + let z_id = z + .get("id") + .and_then(Value::as_str) + .expect("Z event missing id") + .to_string(); + + assert_eq!( + trigger.get("caused_by").and_then(Value::as_str), + None, + "IPC-originated trigger event should have no caused_by" + ); + assert_eq!( + x.get("caused_by").and_then(Value::as_str), + Some(trigger_id.as_str()), + "X should be caused_by the trigger event's id" + ); + assert_eq!( + y.get("caused_by").and_then(Value::as_str), + Some(x_id.as_str()), + "Y should be caused_by X's id, not the original trigger" + ); + assert_eq!( + z.get("caused_by").and_then(Value::as_str), + Some(y_id.as_str()), + "Z should be caused_by Y's id" + ); + + let ids: std::collections::HashSet<&str> = + [trigger_id.as_str(), x_id.as_str(), y_id.as_str(), z_id.as_str()] + .into_iter() + .collect(); + assert_eq!( + ids.len(), + 4, + "every event in the chain should have a distinct id" + ); + + harness.shutdown(); + Ok(()) +} + #[tokio::test] async fn workflow_reaches_done_via_wait_any_happy_path() -> Result<()> { let harness = TestHarness::spawn_with_init(