Merge feature/event-causality (Workstream B)

This commit is contained in:
Breadway 2026-08-04 22:33:29 +08:00
commit 0384ea1354
9 changed files with 419 additions and 8 deletions

12
Cargo.lock generated
View file

@ -325,6 +325,7 @@ dependencies = [
"serde", "serde",
"serde_json", "serde_json",
"toml", "toml",
"uuid",
] ]
[[package]] [[package]]
@ -2051,6 +2052,17 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" 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]] [[package]]
name = "valuable" name = "valuable"
version = "0.1.1" version = "0.1.1"

View file

@ -152,6 +152,7 @@ installed_at = "2026-01-01T00:00:00Z"
## Debugging tips ## Debugging tips
- Run `bread events` to see live normalized events. - 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 state` to see full runtime state as JSON.
- Run `bread doctor` to check adapter and module health. - Run `bread doctor` to check adapter and module health.
- Log event payloads with `bread.log(tostring(event.data))`. - Log event payloads with `bread.log(tostring(event.data))`.
@ -213,7 +214,7 @@ end, {
Unsubscribe an event handler or state watch by ID. Unsubscribe an event handler or state watch by ID.
#### `bread.emit(event, data)` #### `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` #### `bread.wait(pattern, opts) -> event | nil`
Coroutine-only helper that suspends until a matching event arrives. 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", "event": "bread.device.dock.connected",
"timestamp": 1710000000000, "timestamp": 1710000000000,
"source": "Udev", "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 matching
| Pattern | Matches | | Pattern | Matches |

View file

@ -6,6 +6,7 @@ use anyhow::Result;
use clap::{Parser, Subcommand}; use clap::{Parser, Subcommand};
use notify::{RecommendedWatcher, RecursiveMode, Watcher}; use notify::{RecommendedWatcher, RecursiveMode, Watcher};
use serde_json::{json, Value}; use serde_json::{json, Value};
use std::collections::HashMap;
use std::env; use std::env;
use std::io::{self, Write as IoWrite}; use std::io::{self, Write as IoWrite};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
@ -54,6 +55,12 @@ enum Commands {
/// Replay events from the last N seconds /// Replay events from the last N seconds
#[arg(long)] #[arg(long)]
since: Option<u64>, since: Option<u64>,
/// 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 /// Manage installed Lua modules
Modules { Modules {
@ -158,8 +165,9 @@ async fn main() -> Result<()> {
json, json,
fields, fields,
since, since,
tree,
} => { } => {
stream_events(&socket, pattern, json, fields, since).await?; stream_events(&socket, pattern, json, fields, since, tree).await?;
} }
Commands::Modules { subcommand } => { Commands::Modules { subcommand } => {
handle_modules_cmd(subcommand, &socket).await?; handle_modules_cmd(subcommand, &socket).await?;
@ -378,7 +386,13 @@ async fn stream_events(
raw_json: bool, raw_json: bool,
fields: Option<String>, fields: Option<String>,
since: Option<u64>, since: Option<u64>,
tree: bool,
) -> Result<()> { ) -> 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 { if let Some(seconds) = since {
let replay = send_request( let replay = send_request(
socket, socket,
@ -388,7 +402,9 @@ async fn stream_events(
.await?; .await?;
if let Some(list) = replay.as_array() { if let Some(list) = replay.as_array() {
for item in list { for item in list {
if raw_json { if tree {
causality.print(item);
} else if raw_json {
println!("{}", serde_json::to_string_pretty(item)?); println!("{}", serde_json::to_string_pretty(item)?);
} else { } else {
print_event(item, fields.as_deref()); print_event(item, fields.as_deref());
@ -426,7 +442,9 @@ async fn stream_events(
while let Some(line) = lines.next_line().await? { while let Some(line) = lines.next_line().await? {
let value: Value = serde_json::from_str(&line)?; 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)?); println!("{}", serde_json::to_string_pretty(&value)?);
} else { } else {
print_event(&value, fields.as_deref()); print_event(&value, fields.as_deref());
@ -436,6 +454,63 @@ async fn stream_events(
Ok(()) 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<String, Option<String>>,
}
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(&current) {
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<()> { fn print_json(value: &Value) -> Result<()> {
println!("{}", serde_json::to_string_pretty(value)?); println!("{}", serde_json::to_string_pretty(value)?);
Ok(()) Ok(())

View file

@ -8,3 +8,4 @@ serde.workspace = true
serde_json.workspace = true serde_json.workspace = true
dirs.workspace = true dirs.workspace = true
toml = "0.8" toml = "0.8"
uuid = { version = "1", features = ["v4"] }

View file

@ -106,20 +106,65 @@ pub struct BreadEvent {
pub source: AdapterSource, pub source: AdapterSource,
/// Structured event data. The shape depends on the event family. /// Structured event data. The shape depends on the event family.
pub data: serde_json::Value, 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<String>,
} }
impl BreadEvent { 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<String>, source: AdapterSource, data: serde_json::Value) -> Self { pub fn new(event: impl Into<String>, 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<String>,
timestamp: u64,
source: AdapterSource,
data: serde_json::Value,
) -> Self {
Self { Self {
event: event.into(), event: event.into(),
timestamp: now_unix_ms(), timestamp,
source, source,
data, 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. /// Current Unix epoch in milliseconds.
/// ///
/// Falls back to `0` if the system clock is before the epoch, which keeps /// Falls back to `0` if the system clock is before the epoch, which keeps
@ -337,6 +382,8 @@ mod tests {
timestamp: 1_700_000_000_000, timestamp: 1_700_000_000_000,
source: AdapterSource::Udev, source: AdapterSource::Udev,
data: json!({ "id": "usb-1-1.4", "name": "Logitech" }), 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 raw = serde_json::to_string(&original).unwrap();
let decoded: BreadEvent = serde_json::from_str(&raw).unwrap(); let decoded: BreadEvent = serde_json::from_str(&raw).unwrap();
@ -345,6 +392,30 @@ mod tests {
assert_eq!(decoded.timestamp, original.timestamp); assert_eq!(decoded.timestamp, original.timestamp);
assert_eq!(decoded.source, original.source); assert_eq!(decoded.source, original.source);
assert_eq!(decoded.data, original.data); 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] #[test]

View file

@ -43,6 +43,8 @@ impl EventNormalizer {
event: raw.kind.clone(), event: raw.kind.clone(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}], }],
// `Manual` is never constructed as a `RawEvent::source` in this // `Manual` is never constructed as a `RawEvent::source` in this
@ -156,6 +158,8 @@ impl EventNormalizer {
event: format!("bread.device.{}", verb), event: format!("bread.device.{}", verb),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Udev, source: AdapterSource::Udev,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"id": id, "id": id,
"device": "unknown", "device": "unknown",
@ -186,36 +190,48 @@ impl EventNormalizer {
event: "bread.workspace.changed".to_string(), event: "bread.workspace.changed".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}], }],
"createworkspace" => vec![BreadEvent { "createworkspace" => vec![BreadEvent {
event: "bread.workspace.created".to_string(), event: "bread.workspace.created".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "workspace": data }), data: json!({ "workspace": data }),
}], }],
"destroyworkspace" => vec![BreadEvent { "destroyworkspace" => vec![BreadEvent {
event: "bread.workspace.destroyed".to_string(), event: "bread.workspace.destroyed".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "workspace": data }), data: json!({ "workspace": data }),
}], }],
"monitoradded" => vec![BreadEvent { "monitoradded" => vec![BreadEvent {
event: "bread.monitor.connected".to_string(), event: "bread.monitor.connected".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "name": data }), data: json!({ "name": data }),
}], }],
"monitorremoved" => vec![BreadEvent { "monitorremoved" => vec![BreadEvent {
event: "bread.monitor.disconnected".to_string(), event: "bread.monitor.disconnected".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "name": data }), data: json!({ "name": data }),
}], }],
"activewindow" => vec![BreadEvent { "activewindow" => vec![BreadEvent {
event: "bread.window.focus.changed".to_string(), event: "bread.window.focus.changed".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}], }],
"activewindowv2" => { "activewindowv2" => {
@ -224,6 +240,8 @@ impl EventNormalizer {
event: "bread.window.focused".to_string(), event: "bread.window.focused".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"address": fields.first().unwrap_or(&"") "address": fields.first().unwrap_or(&"")
}), }),
@ -235,6 +253,8 @@ impl EventNormalizer {
event: "bread.window.opened".to_string(), event: "bread.window.opened".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"address": fields.first().unwrap_or(&""), "address": fields.first().unwrap_or(&""),
"workspace": fields.get(1).unwrap_or(&""), "workspace": fields.get(1).unwrap_or(&""),
@ -249,6 +269,8 @@ impl EventNormalizer {
event: "bread.window.closed".to_string(), event: "bread.window.closed".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ "address": fields.first().unwrap_or(&"") }), data: json!({ "address": fields.first().unwrap_or(&"") }),
}] }]
} }
@ -258,6 +280,8 @@ impl EventNormalizer {
event: "bread.window.moved".to_string(), event: "bread.window.moved".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"address": fields.first().unwrap_or(&""), "address": fields.first().unwrap_or(&""),
"workspace": fields.get(1).unwrap_or(&""), "workspace": fields.get(1).unwrap_or(&""),
@ -268,6 +292,8 @@ impl EventNormalizer {
event: "bread.hyprland.event".to_string(), event: "bread.hyprland.event".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Hyprland, source: AdapterSource::Hyprland,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}], }],
} }
@ -285,6 +311,8 @@ impl EventNormalizer {
}, },
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Power, source: AdapterSource::Power,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}); });
} }
@ -307,6 +335,8 @@ impl EventNormalizer {
event: event.to_string(), event: event.to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Power, source: AdapterSource::Power,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}); });
} }
@ -317,6 +347,8 @@ impl EventNormalizer {
event: "bread.power.changed".to_string(), event: "bread.power.changed".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Power, source: AdapterSource::Power,
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}); });
} }
@ -352,6 +384,8 @@ impl EventNormalizer {
event: "bread.device.connected".to_string(), event: "bread.device.connected".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Bluetooth, source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"id": path, "id": path,
"device": "unknown", "device": "unknown",
@ -365,6 +399,8 @@ impl EventNormalizer {
event: "bread.device.disconnected".to_string(), event: "bread.device.disconnected".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Bluetooth, source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"id": path, "id": path,
"device": "unknown", "device": "unknown",
@ -378,6 +414,8 @@ impl EventNormalizer {
event: "bread.bluetooth.device.paired".to_string(), event: "bread.bluetooth.device.paired".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Bluetooth, source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"id": path, "id": path,
"name": name, "name": name,
@ -390,6 +428,8 @@ impl EventNormalizer {
event: "bread.bluetooth.device.unpaired".to_string(), event: "bread.bluetooth.device.unpaired".to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Bluetooth, source: AdapterSource::Bluetooth,
id: bread_shared::new_event_id(),
caused_by: None,
data: json!({ data: json!({
"id": path, "id": path,
"address": address, "address": address,
@ -434,6 +474,8 @@ impl EventNormalizer {
event: name.to_string(), event: name.to_string(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: AdapterSource::Network, source: AdapterSource::Network,
id: bread_shared::new_event_id(),
caused_by: None,
data, data,
}] }]
} }
@ -448,6 +490,8 @@ impl EventNormalizer {
event: format!("bread.terminal.{}", raw.kind), event: format!("bread.terminal.{}", raw.kind),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}] }]
} }
@ -457,6 +501,8 @@ impl EventNormalizer {
event: format!("bread.remote.{}", raw.kind), event: format!("bread.remote.{}", raw.kind),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}] }]
} }
@ -466,6 +512,8 @@ impl EventNormalizer {
event: format!("bread.git.{}", raw.kind), event: format!("bread.git.{}", raw.kind),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}] }]
} }
@ -475,6 +523,8 @@ impl EventNormalizer {
event: format!("bread.project.{}", raw.kind), event: format!("bread.project.{}", raw.kind),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}] }]
} }
@ -487,6 +537,8 @@ impl EventNormalizer {
event: format!("bread.service.{suffix}"), event: format!("bread.service.{suffix}"),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}] }]
} }
@ -503,6 +555,8 @@ impl EventNormalizer {
event, event,
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}] }]
} }
@ -525,6 +579,8 @@ impl EventNormalizer {
event: raw.kind.clone(), event: raw.kind.clone(),
timestamp: raw.timestamp, timestamp: raw.timestamp,
source: raw.source.clone(), source: raw.source.clone(),
id: bread_shared::new_event_id(),
caused_by: None,
data: raw.payload.clone(), data: raw.payload.clone(),
}] }]
} }

View file

@ -656,6 +656,8 @@ mod tests {
timestamp: 0, timestamp: 0,
source: AdapterSource::System, source: AdapterSource::System,
data, data,
id: bread_shared::new_event_id(),
caused_by: None,
} }
} }

View file

@ -211,6 +211,19 @@ struct LuaEngine {
next_sub_id: Arc<AtomicU64>, next_sub_id: Arc<AtomicU64>,
next_timer_id: Arc<AtomicU64>, next_timer_id: Arc<AtomicU64>,
current_module: Arc<Mutex<Option<String>>>, current_module: Arc<Mutex<Option<String>>>,
/// 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<Mutex<Option<String>>>,
modules: Arc<Mutex<HashMap<String, ModuleInfo>>>, modules: Arc<Mutex<HashMap<String, ModuleInfo>>>,
module_decls: Arc<Mutex<HashMap<String, ModuleDecl>>>, module_decls: Arc<Mutex<HashMap<String, ModuleDecl>>>,
module_order: Arc<Mutex<Vec<String>>>, module_order: Arc<Mutex<Vec<String>>>,
@ -240,6 +253,7 @@ impl LuaEngine {
next_sub_id: Arc::new(AtomicU64::new(1)), next_sub_id: Arc::new(AtomicU64::new(1)),
next_timer_id: Arc::new(AtomicU64::new(1)), next_timer_id: Arc::new(AtomicU64::new(1)),
current_module: Arc::new(Mutex::new(None)), current_module: Arc::new(Mutex::new(None)),
current_dispatch_id: Arc::new(Mutex::new(None)),
modules: Arc::new(Mutex::new(HashMap::new())), modules: Arc::new(Mutex::new(HashMap::new())),
module_decls: Arc::new(Mutex::new(HashMap::new())), module_decls: Arc::new(Mutex::new(HashMap::new())),
module_order: Arc::new(Mutex::new(Vec::new())), module_order: Arc::new(Mutex::new(Vec::new())),
@ -438,6 +452,7 @@ impl LuaEngine {
bread.set("off", off_fn)?; bread.set("off", off_fn)?;
let emit_tx = self.emit_tx.clone(); let emit_tx = self.emit_tx.clone();
let current_dispatch_id = self.current_dispatch_id.clone();
let emit_fn = let emit_fn =
self.lua self.lua
.create_function(move |lua, (event_name, payload): (String, Value)| { .create_function(move |lua, (event_name, payload): (String, Value)| {
@ -447,8 +462,19 @@ impl LuaEngine {
.from_value::<serde_json::Value>(other) .from_value::<serde_json::Value>(other)
.unwrap_or_else(|_| serde_json::json!({})), .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 emit_tx
.send(BreadEvent::new(event_name, AdapterSource::System, data)) .send(event)
.map_err(|_| LuaError::external("event channel closed"))?; .map_err(|_| LuaError::external("event channel closed"))?;
Ok(()) Ok(())
})?; })?;
@ -1489,6 +1515,11 @@ impl LuaEngine {
} }
self.set_current_module(module.clone()); 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 { let result = match kind {
HandlerKind::Event => { HandlerKind::Event => {
let event_value = json_to_lua(&self.lua, &event)?; let event_value = json_to_lua(&self.lua, &event)?;
@ -1502,6 +1533,9 @@ impl LuaEngine {
callback.call::<_, ()>((new_lua, old_lua)) 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); self.set_current_module(None);
if let Err(err) = result { 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<String>) -> Option<String> {
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) { fn cancel_all_timers(&self) {
if let Ok(mut map) = self.timers.lock() { if let Ok(mut map) = self.timers.lock() {
for (_, entry) in map.drain() { for (_, entry) in map.drain() {

View file

@ -1,3 +1,4 @@
use std::collections::HashMap;
use std::fs; use std::fs;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio}; use std::process::{Child, Command, Stdio};
@ -675,6 +676,147 @@ async fn events_stream_receives_emitted_events() -> Result<()> {
Ok(()) 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<String, Value> = 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] #[tokio::test]
async fn workflow_reaches_done_via_wait_any_happy_path() -> Result<()> { async fn workflow_reaches_done_via_wait_any_happy_path() -> Result<()> {
let harness = TestHarness::spawn_with_init( let harness = TestHarness::spawn_with_init(