fix(broker): surface agent-originated fleet deliveries to the dashboard live

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<Value> {
+    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"
+            );
+        }
+    }
 }