From 61d6f42521a9b8a181f75c8810f758e7f802e6c4 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Wed, 1 Jul 2026 14:49:52 -0700 Subject: [PATCH] fix(broker): surface agent-originated fleet deliveries to the dashboard live Confirmed via live debugging: a human's own message sent from Pear's dashboard gets an immediate `relay_inbound` event (ListenApiRequest::Send in runtime/api.rs), so it shows up in the dashboard instantly. But an agent's reply to the same channel arrives back at this broker only through the node control plane (handle_fleet_deliver -> surface_fleet_deliver), which injects it into other agents' PTYs and stops there -- nothing notified the dashboard, so agent replies never appeared live (only via the separate reconciliation/polling fallback). This adds a `relay_inbound` dashboard emission alongside the existing PTY injection in surface_fleet_deliver's Inject branch, via a new pure helper `fleet_dashboard_relay_inbound_event` (and `is_chat_message_delivery`) so the decision logic is unit-testable without a full BrokerRuntime. Two duplicate-message risks (this codebase has a recent history of exactly this class of bug -- the renderer-side canonical-echo-race fixes) are handled broker-side rather than left to renderer dedup: 1. One channel message fans out to one Deliver frame PER local recipient (confirmed: same msg_id, different delivery_id/agent per recipient). The event_id is always deliver.msg_id -- the same value fleet_relay_delivery already uses for the PTY path -- so the renderer's existing exact-id dedup collapses the per-recipient duplicates. 2. A dashboard-originated message round-trips back through node delivery to local workers subscribed to the channel, which would otherwise re-emit it under a different (deliver.msg_id vs http_*) id that exact-id dedup can't catch. Skipped broker-side using the existing sender_is_dashboard_label helper (same one runtime/api.rs's Send handler uses), rather than relying on fuzzy renderer dedup. Also scoped to genuine message-class payload types only: action.completed / action.failed / action.denied are Inject-classified (PTY'd back to the invoking agent as action results) but aren't chat messages and must not show up as a dashboard chat bubble. Added 5 unit tests covering: event shape for a message-class delivery, event_id stability across fanned-out recipients sharing one msg_id, skipping dashboard-originated senders, and skipping action-result deliveries. Co-Authored-By: Claude Sonnet 5 --- crates/broker/src/runtime/fleet.rs | 339 +++++++++++++++++++++++++++++ 1 file changed, 339 insertions(+) diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 1f4e0bf2d..30afcdd0b 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -396,6 +396,35 @@ impl BrokerRuntime { // here, it's never honored anywhere. FleetDeliverySurfacing::Inject => { let fields = fleet_delivery_fields(&deliver.payload, &deliver.agent); + + // Mirror the `relay_inbound` dashboard event that the HTTP + // `Send` handler (`ListenApiRequest::Send` in runtime/api.rs) + // emits at send time, so Pear's dashboard learns about + // agent-originated / remote channel & DM traffic live, the + // same way it already does for messages a human sends from + // Pear's own UI. Without this, a human's own message shows + // up instantly but an agent's reply to the same channel + // never appears live (it's only injected into other agents' + // PTYs here, and eventually reconciled via polling). See + // `fleet_dashboard_relay_inbound_event`'s doc comment for how + // it avoids both the per-recipient-fanout duplicate and the + // dashboard-echoing-its-own-message duplicate. + if let Some(dashboard_event) = fleet_dashboard_relay_inbound_event( + payload_type, + deliver, + &fields, + &self.default_workspace.self_name, + self.default_workspace_id.as_deref(), + self.default_workspace.workspace_alias.as_deref(), + ) { + emit_http_api_event_with_timeout( + &self.sdk_out_tx, + dashboard_event, + http_api_event_emit_timeout(), + ) + .await; + } + let injection_mode = match deliver.mode { DeliveryMode::Wait => MessageInjectionMode::Wait, DeliveryMode::Steer => MessageInjectionMode::Steer, @@ -1222,6 +1251,93 @@ fn classify_fleet_delivery(payload_type: &str) -> FleetDeliverySurfacing { } } +/// Whether a node `deliver` payload `type` represents an actual chat message +/// arriving (channel post, DM, thread reply) as opposed to an action-result +/// fan-out (`action.completed` / `action.failed` / `action.denied`) or an +/// ambient reaction/receipt. Both message-class and action-result types are +/// `FleetDeliverySurfacing::Inject` (both get PTY'd to a worker), but only +/// message-class types are "someone sent a message" for dashboard purposes — +/// mirrors the message-class alias arm of `classify_fleet_delivery` exactly +/// (kept as a separate list rather than folding into that function's return +/// type, since callers of `classify_fleet_delivery` outside the dashboard +/// concern don't need this distinction). +fn is_chat_message_delivery(payload_type: &str) -> bool { + matches!( + payload_type, + "message.created" + | "message.received" + | "message.new" + | "message.sent" + | "message.delivered" + | "thread.reply" + | "thread.message.created" + | "thread.message.sent" + | "dm.received" + | "dm.created" + | "dm.new" + | "dm.sent" + | "dm.message.created" + | "direct_message.received" + | "direct_message.created" + | "direct_message.new" + | "direct_message.sent" + | "group_dm.received" + | "group_dm.created" + | "group_dm.new" + | "group_dm.sent" + | "group_dm.message.created" + | "" + ) +} + +/// Build the `relay_inbound` dashboard event for a node `deliver` frame that +/// is about to be `Inject`-surfaced, or `None` when it shouldn't be surfaced +/// to the dashboard at all. +/// +/// Returns `None` when either: +/// - `payload_type` isn't a genuine chat-message class (e.g. it's an +/// `action.completed`/`action.failed`/`action.denied` result, which is +/// `Inject`-classified for PTY purposes but isn't "someone sent a +/// message"), or +/// - the delivered message's sender is this broker's own dashboard/self +/// identity (`sender_is_dashboard_label`) — that message was already +/// surfaced to the dashboard synchronously at HTTP send time +/// (`ListenApiRequest::Send` in runtime/api.rs) under a different +/// (`http_*`) event id, so re-emitting it here under `deliver.msg_id` +/// would show up as a second, undeduped bubble. +/// +/// When `Some`, the event's `event_id` is always `deliver.msg_id` — the +/// same value the node control plane fans out across every local +/// recipient's own `Deliver` frame for one underlying message (mirrors +/// `fleet_relay_delivery`'s PTY-path `EventId`), so the renderer's +/// exact-id dedup collapses the multiple dashboard events this function +/// will produce (one per local recipient) down to one visible message. +fn fleet_dashboard_relay_inbound_event( + payload_type: &str, + deliver: &Deliver, + fields: &FleetDeliveryFields, + self_name: &str, + workspace_id: Option<&str>, + workspace_alias: Option<&str>, +) -> Option { + if !is_chat_message_delivery(payload_type) { + return None; + } + if sender_is_dashboard_label(&fields.from, self_name) { + return None; + } + Some(json!({ + "kind": "relay_inbound", + "event_id": deliver.msg_id.as_str(), + "from": fields.from.as_str(), + "target": fields.target.as_str(), + "body": fields.body.as_str(), + "thread_id": fields.thread_id.as_ref().map(ThreadId::as_str), + "workspace_id": workspace_id, + "workspace_alias": workspace_alias, + })) +} + /// Resolve the worker name a node `action.invoke` targets: prefer the frame's /// `agent_name`, then the input's `name`/`agent`/`agent_name`/`agent_id` /// fields. Returns `None` when no non-empty identity is present. @@ -1871,4 +1987,227 @@ mod tests { crate::node_control::DeliveryDecision::Deliver { up_to_seq: 1 } ); } + + fn test_deliver(agent: &str, delivery_id: &str, msg_id: &str, payload: Value) -> Deliver { + Deliver { + v: FLEET_WIRE_VERSION, + agent: agent.to_string(), + agent_id: format!("{agent}-id"), + delivery_id: delivery_id.to_string(), + msg_id: msg_id.to_string(), + seq: 1, + mode: DeliveryMode::Wait, + payload, + } + } + + #[test] + fn is_chat_message_delivery_covers_message_classes_but_not_action_results() { + for message_class in [ + "message.created", + "message.received", + "message.new", + "message.sent", + "message.delivered", + "thread.reply", + "thread.message.created", + "thread.message.sent", + "dm.received", + "dm.created", + "dm.new", + "dm.sent", + "dm.message.created", + "direct_message.received", + "direct_message.created", + "direct_message.new", + "direct_message.sent", + "group_dm.received", + "group_dm.created", + "group_dm.new", + "group_dm.sent", + "group_dm.message.created", + "", + ] { + assert!( + is_chat_message_delivery(message_class), + "{message_class} should be a chat-message class" + ); + } + // action-result fan-out is Inject-classified (PTY'd back to the + // caller) but is NOT a chat message and must not be treated as one. + for action_result in ["action.completed", "action.failed", "action.denied"] { + assert!( + !is_chat_message_delivery(action_result), + "{action_result} must not be treated as a chat message" + ); + } + // ack-only / unknown classes are also not chat messages. + for other in ["message.reacted", "message.read", "something.new"] { + assert!(!is_chat_message_delivery(other)); + } + } + + #[test] + fn fleet_dashboard_relay_inbound_event_has_expected_shape_for_message_class_delivery() { + // (a) A message-class delivery to one recipient produces exactly one + // dashboard event, shaped like the HTTP Send handler's `relay_inbound` + // event, with `event_id` stably set to `deliver.msg_id`. + let deliver = test_deliver("claude-1", "delivery-1", "msg-123", json!({})); + let fields = FleetDeliveryFields { + body: "hello #general".to_string(), + from: "codex-1".to_string(), + target: "#general".to_string(), + thread_id: Some(ThreadId::new("thr-1")), + priority: None, + }; + let event = fleet_dashboard_relay_inbound_event( + "message.created", + &deliver, + &fields, + "broker-self", + Some("ws-1"), + Some("alias-1"), + ) + .expect("message-class delivery from a non-dashboard sender should emit"); + + assert_eq!(event["kind"], "relay_inbound"); + assert_eq!(event["event_id"], "msg-123"); + assert_eq!(event["from"], "codex-1"); + assert_eq!(event["target"], "#general"); + assert_eq!(event["body"], "hello #general"); + assert_eq!(event["thread_id"], "thr-1"); + assert_eq!(event["workspace_id"], "ws-1"); + assert_eq!(event["workspace_alias"], "alias-1"); + } + + #[test] + fn fleet_dashboard_relay_inbound_event_id_is_stable_across_fanned_out_recipients() { + // (b) The node control plane fans a single channel message out to one + // `Deliver` frame PER local recipient, each with a distinct + // `delivery_id`/`agent` but the SAME `msg_id`. The dashboard event's + // `event_id` must be that shared `msg_id` in every case, so the + // renderer's exact-id dedup collapses the duplicates instead of + // showing the same message once per recipient. + let fields = FleetDeliveryFields { + body: "hello #general".to_string(), + from: "codex-1".to_string(), + target: "#general".to_string(), + thread_id: None, + priority: None, + }; + let deliver_to_claude = test_deliver("claude-1", "delivery-1", "msg-shared", json!({})); + let deliver_to_gpt = test_deliver("gpt-1", "delivery-2", "msg-shared", json!({})); + + let event_a = fleet_dashboard_relay_inbound_event( + "message.created", + &deliver_to_claude, + &fields, + "broker-self", + None, + None, + ) + .expect("first recipient's delivery should emit"); + let event_b = fleet_dashboard_relay_inbound_event( + "message.created", + &deliver_to_gpt, + &fields, + "broker-self", + None, + None, + ) + .expect("second recipient's delivery should emit"); + + assert_eq!(event_a["event_id"], "msg-shared"); + assert_eq!(event_b["event_id"], "msg-shared"); + assert_eq!( + event_a["event_id"], event_b["event_id"], + "event_id must be identical across every local recipient's Deliver frame for the same underlying message" + ); + } + + #[test] + fn fleet_dashboard_relay_inbound_event_skips_dashboard_originated_messages() { + // (c) A human's own message sent from Pear's dashboard already gets an + // immediate `relay_inbound` emission (under a different, `http_*`, + // event id) at HTTP send time, and then round-trips back through node + // delivery to any local worker subscribed to the channel. Re-emitting + // it here — under `deliver.msg_id` instead of the original `http_*` + // id — would duplicate it under a second id that exact-id dedup can't + // catch, so it must be skipped broker-side using the same + // `sender_is_dashboard_label` check the Send handler uses. + let deliver = test_deliver("claude-1", "delivery-1", "msg-456", json!({})); + for dashboard_label in [ + "Dashboard", + "human:Dashboard", + "human:orchestrator", + "broker-self", + ] { + let fields = FleetDeliveryFields { + body: "hi from dashboard".to_string(), + from: dashboard_label.to_string(), + target: "#general".to_string(), + thread_id: None, + priority: None, + }; + assert!( + fleet_dashboard_relay_inbound_event( + "message.created", + &deliver, + &fields, + "broker-self", + None, + None, + ) + .is_none(), + "sender {dashboard_label} should be recognized as the dashboard/self identity and skipped" + ); + } + // A non-dashboard sender (another agent, a remote human) still emits. + let fields = FleetDeliveryFields { + body: "hi".to_string(), + from: "codex-1".to_string(), + target: "#general".to_string(), + thread_id: None, + priority: None, + }; + assert!(fleet_dashboard_relay_inbound_event( + "message.created", + &deliver, + &fields, + "broker-self", + None, + None, + ) + .is_some()); + } + + #[test] + fn fleet_dashboard_relay_inbound_event_skips_action_result_deliveries() { + // (d) action.completed/action.failed/action.denied are Inject-classified + // (PTY'd back to the invoking agent as an action result) but are not + // chat messages and must not surface as a dashboard `relay_inbound` + // chat bubble. + let deliver = test_deliver("claude-1", "delivery-1", "msg-789", json!({})); + let fields = FleetDeliveryFields { + body: "{\"ok\":true}".to_string(), + from: "codex-1".to_string(), + target: "claude-1".to_string(), + thread_id: None, + priority: None, + }; + for action_result_type in ["action.completed", "action.failed", "action.denied"] { + assert!( + fleet_dashboard_relay_inbound_event( + action_result_type, + &deliver, + &fields, + "broker-self", + None, + None, + ) + .is_none(), + "{action_result_type} must not emit a dashboard relay_inbound event" + ); + } + } }