[fix] chat: wire live message delivery end-to-end (MESSAGE_SENT)

send_message persisted messages but never broadcast them, there was no
MESSAGE_SENT event type or bridge forwarder, and the panel session view
had no websocket subscription — the live chat path was dead end-to-end.

- add EventType.MESSAGE_SENT and publish it best-effort on every persisted
  send (a bus outage logs, never rolls back the durable row)
- bridge _handle_message_event forwards to /ws/sessions/{id} and
  /ws/channels/{id}; subscribe it in register_websocket_bridge_handlers
- panel useSessionStream subscribes the session view; the page invalidates
  the transcript + session-detail queries on each message.new so the held
  (staleTime Infinity) views refresh live without the manual Refresh

panel/src/app/(dashboard)/communications/[sessionId]/page.tsx

@@ -3,7 +3,13 @@
 import { useEffect, useRef } from "react";
 import { useParams, useSearchParams } from "next/navigation";
 import { useMutation, useQueryClient } from "@tanstack/react-query";
-import { useSession, useSessionMessages } from "@/hooks/use-channels";
+import {
+  useSession,
+  useSessionMessages,
+  messageKeys,
+  sessionKeys,
+} from "@/hooks/use-channels";
+import { useSessionStream } from "@/hooks/use-websocket";
 import { messagesApi } from "@/lib/api/messages";
 import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
 import { Badge } from "@/components/ui/badge";
@@ -64,6 +70,17 @@ function SessionDetailContent() {
     refetch: refetchMessages,
   } = useSessionMessages(sessionId);
 
+  // Live updates: subscribe to the session stream. On a new persisted message
+  // (MESSAGE_SENT → bridge → /ws/sessions/{id}) invalidate the transcript +
+  // session-detail queries so the held (staleTime Infinity) views refresh
+  // without the manual Refresh button.
+  const { lastMessage } = useSessionStream(sessionId);
+  useEffect(() => {
+    if (lastMessage?.type !== "message.new") return;
+    queryClient.invalidateQueries({ queryKey: messageKeys.list(sessionId) });
+    queryClient.invalidateQueries({ queryKey: sessionKeys.detail(sessionId) });
+  }, [lastMessage, queryClient, sessionId]);
+
   // Sort messages chronologically (oldest first for chat UI)
   const messages = [...(messagesData?.items || [])].sort(
     (a, b) => new Date(a.timestamp).getTime() - new Date(b.timestamp).getTime(),

panel/src/hooks/__tests__/use-session-stream.test.tsx

@@ -0,0 +1,114 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { render, act } from "@testing-library/react";
+import { useEffect } from "react";
+import type {
+  ConnectionState,
+  WebSocketOptions,
+} from "@/lib/websocket/connection";
+
+// Bundle A: the session detail view had no live subscription — send_message now
+// publishes MESSAGE_SENT → bridge → /ws/sessions/{id}, and useSessionStream is
+// the panel half that subscribes so an open transcript updates without a manual
+// Refresh. Mock the connection so the test can drive onMessage frames.
+const hoisted = vi.hoisted(() => {
+  const instances: MockConnection[] = [];
+  class MockConnection {
+    url: string;
+    onMessage?: (data: unknown) => void;
+    onStateChange?: (state: ConnectionState) => void;
+    didConnect = false;
+    didDisconnect = false;
+    constructor(opts: WebSocketOptions) {
+      this.url = opts.url;
+      this.onMessage = opts.onMessage;
+      this.onStateChange = opts.onStateChange;
+      instances.push(this);
+    }
+    connect() {
+      this.didConnect = true;
+      this.onStateChange?.("connecting");
+      this.onStateChange?.("connected");
+    }
+    disconnect() {
+      this.didDisconnect = true;
+      this.onStateChange?.("disconnected");
+    }
+  }
+  return { instances, MockConnection };
+});
+
+vi.mock("@/lib/websocket/connection", () => ({
+  getWebSocketUrl: () => "ws://test/ws",
+  WebSocketConnection: hoisted.MockConnection,
+}));
+
+vi.mock("@/lib/constants", () => ({
+  CEO_AGENT_ID: "00000000-0000-0000-0000-000000000001",
+  STREAM_MAX_MESSAGES: 100,
+}));
+
+import { useSessionStream } from "../use-websocket";
+
+const resultRef: {
+  current: ReturnType<typeof useSessionStream> | null;
+} = { current: null };
+
+function Harness({ sessionId }: { sessionId: string | null }) {
+  const ws = useSessionStream(sessionId);
+  useEffect(() => {
+    resultRef.current = ws;
+  });
+  return null;
+}
+
+describe("useSessionStream", () => {
+  beforeEach(() => {
+    hoisted.instances.length = 0;
+    resultRef.current = null;
+  });
+  afterEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("connects to the session endpoint with the CEO agent_id", () => {
+    render(<Harness sessionId="sess-1" />);
+    expect(hoisted.instances).toHaveLength(1);
+    expect(hoisted.instances[0].url).toContain("/sessions/sess-1");
+    expect(hoisted.instances[0].url).toContain(
+      "agent_id=00000000-0000-0000-0000-000000000001",
+    );
+    expect(resultRef.current?.isConnected).toBe(true);
+  });
+
+  it("does not connect when sessionId is null", () => {
+    render(<Harness sessionId={null} />);
+    expect(hoisted.instances).toHaveLength(0);
+  });
+
+  it("surfaces a message.new frame in sessionMessages and lastMessage", () => {
+    render(<Harness sessionId="sess-1" />);
+    const conn = hoisted.instances[0];
+    act(() => {
+      conn.onMessage?.({
+        type: "message.new",
+        message_id: "m1",
+        session_id: "sess-1",
+        agent_id: "a1",
+        content: "hello",
+        message_type: "dialogue",
+      });
+    });
+    expect(resultRef.current?.sessionMessages).toHaveLength(1);
+    expect(resultRef.current?.sessionMessages[0].message_id).toBe("m1");
+    expect(resultRef.current?.lastMessage?.type).toBe("message.new");
+  });
+
+  it("ignores the initial connected frame (not a real message)", () => {
+    render(<Harness sessionId="sess-1" />);
+    const conn = hoisted.instances[0];
+    act(() => {
+      conn.onMessage?.({ type: "connected", session_id: "sess-1" });
+    });
+    expect(resultRef.current?.sessionMessages).toHaveLength(0);
+  });
+});

panel/src/hooks/use-websocket.ts

@@ -44,6 +44,19 @@ export interface NotificationMessage {
   timestamp?: string;
 }
 
+export interface SessionMessage {
+  type: "connected" | "message.new";
+  message_id?: string;
+  session_id?: string;
+  channel_id?: string;
+  agent_id?: string;
+  content?: string;
+  message_type?: string;
+  is_reply?: boolean;
+  reply_to?: string | null;
+  timestamp?: string;
+}
+
 // =============================================================================
 // Generic WebSocket Hook
 // =============================================================================
@@ -197,6 +210,42 @@ export function useChannelStream(channelId: string | null) {
   };
 }
 
+/**
+ * Subscribe to a session's live message stream (`/ws/sessions/{id}`).
+ *
+ * The backend publishes MESSAGE_SENT on every persisted send; the websocket
+ * bridge fans it to this stream as a `message.new` frame. The session detail
+ * view consumes `lastMessage` to refresh its transcript live instead of
+ * relying on the manual Refresh button.
+ */
+export function useSessionStream(sessionId: string | null) {
+  const {
+    state,
+    lastMessage,
+    messages,
+    clearMessages,
+    isConnected,
+    isConnecting,
+  } = useWebSocket<SessionMessage>(
+    sessionId ? "/sessions/" + sessionId : "",
+    { agent_id: CEO_AGENT_ID },
+    !!sessionId,
+  );
+
+  // Filter to only actual messages (drop the initial `connected` frame).
+  const sessionMessages = messages.filter((m) => m.type === "message.new");
+
+  return {
+    state,
+    lastMessage,
+    sessionMessages,
+    allMessages: messages,
+    clearMessages,
+    isConnected,
+    isConnecting,
+  };
+}
+
 /**
  * Subscribe to notifications for the CEO
  */

roboco/api/websocket_bridge.py

@@ -104,6 +104,46 @@ async def _handle_session_event(event: Event) -> None:
     )
 
 
+async def _handle_message_event(event: Event) -> None:
+    """Forward a MESSAGE_SENT event to the channel + session WebSocket streams.
+
+    The service publishes MESSAGE_SENT on every persisted send; this builds the
+    ``message.new`` payload the panel's channel/session stream filters on and
+    fans it out to both subscribers so a live chat view updates without a
+    manual refresh. Either id missing/unparseable → no-op (defensive).
+    """
+    data = event.data
+    session_id_str = data.get("session_id")
+    channel_id_str = data.get("channel_id")
+    if not session_id_str or not channel_id_str:
+        return
+    try:
+        session_id = UUID(session_id_str)
+        channel_id = UUID(channel_id_str)
+    except ValueError:
+        return
+
+    payload = {
+        "type": "message.new",
+        "message_id": data.get("message_id"),
+        "session_id": session_id_str,
+        "channel_id": channel_id_str,
+        "agent_id": data.get("agent_id"),
+        "content": data.get("content", ""),
+        "message_type": data.get("message_type", "dialogue"),
+        "is_reply": bool(data.get("is_reply")),
+        "reply_to": data.get("reply_to"),
+        "timestamp": data.get("timestamp"),
+    }
+    await manager.broadcast_to_session(session_id, payload)
+    await manager.broadcast_to_channel(channel_id, payload)
+    logger.debug(
+        "Message event forwarded to WebSocket",
+        message_id=data.get("message_id"),
+        session_id=session_id_str,
+    )
+
+
 async def _handle_agent_event(event: Event) -> None:
     """Handle agent lifecycle events and forward to WebSocket."""
     data = event.data
@@ -197,6 +237,9 @@ def register_websocket_bridge_handlers() -> None:
     # Usage events -> system WebSocket (panel dashboard)
     bus.subscribe(EventType.USAGE_SNAPSHOT, _handle_usage_event)
 
+    # Message delivery -> channel + session WebSocket streams (live chat)
+    bus.subscribe(EventType.MESSAGE_SENT, _handle_message_event)
+
     logger.info("WebSocket bridge handlers registered")
 
 

roboco/models/events.py

@@ -46,6 +46,10 @@ class EventType(StrEnum):
     SESSION_CLOSED = "session.closed"
     SESSION_TIMEOUT = "session.timeout"
 
+    # Message events — a chat message was persisted and should be pushed live to
+    # /ws/channels/{id} and /ws/sessions/{id} subscribers via the bridge.
+    MESSAGE_SENT = "message.sent"
+
     # Handoff events
     HANDOFF_CREATED = "handoff.created"
     HANDOFF_ACCEPTED = "handoff.accepted"

roboco/services/messaging.py

@@ -1597,6 +1597,39 @@ class MessagingService(BaseService):
         self._update_message_stats(session, group, channel, content_length)
         await self.session.flush()
 
+        # Publish MESSAGE_SENT so the websocket bridge fans the new message out
+        # live to /ws/channels/{id} and /ws/sessions/{id} subscribers. Best-effort
+        # — a bus outage logs and never rolls back the persisted message (live
+        # delivery is supplementary; the row is already durable).
+        try:
+            bus = get_event_bus()
+            if bus.is_connected():
+                await bus.publish(
+                    Event(
+                        type=EventType.MESSAGE_SENT,
+                        data={
+                            "message_id": str(message.id),
+                            "session_id": str(session.id),
+                            "channel_id": str(channel.id),
+                            "group_id": str(group.id),
+                            "agent_id": str(req.agent_id),
+                            "content": req.content,
+                            "message_type": (
+                                req.message_type.value
+                                if req.message_type
+                                else "dialogue"
+                            ),
+                            "is_reply": req.reply_to is not None,
+                            "reply_to": str(req.reply_to) if req.reply_to else None,
+                            "timestamp": datetime.now(UTC).isoformat(),
+                        },
+                    )
+                )
+        except Exception as e:
+            self.log.warning(
+                "Failed to publish message event", error=str(e)
+            )
+
         # Notify mentioned agents via Redis Streams
         await self._notify_mentions(message, req.agent_id, channel.slug)
 

tests/integration/test_messaging_service.py

@@ -17,6 +17,7 @@ from roboco.db.tables import AgentTable, MessageTable, ProjectTable, TaskTable
 from roboco.db.tables import AgentTable as _AgentTable
 from roboco.enforcement.channel_access import ChannelAccessDeniedError
 from roboco.models import AgentRole, AgentStatus, MessageType, Team
+from roboco.models.events import EventType
 from roboco.models.base import (
     ChannelType,
     SessionStatus,
@@ -644,6 +645,59 @@ async def test_send_message_to_session(msg_setup: dict) -> None:
     assert msg.content == "hello world"
 
 
+@pytest.mark.asyncio
+async def test_send_message_publishes_message_sent_when_bus_connected(
+    msg_setup: dict,
+) -> None:
+    """Bus connected → a MESSAGE_SENT event is published carrying the message
+    payload so the websocket bridge can fan the new message out to
+    /ws/channels/{id} and /ws/sessions/{id} subscribers. Without this publish
+    the live chat path is dead — the panel never sees incoming messages."""
+    svc = msg_setup["svc"]
+    aid = msg_setup["agent_id"]
+    ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
+    grp = await svc.create_group(GroupCreateRequest(name="g1", channel_id=ch.id))
+    sess = await svc.create_session(SessionCreateRequest(group_id=grp.id))
+    mock_bus = AsyncMock()
+    mock_bus.is_connected = lambda: True
+    mock_bus.publish = AsyncMock(return_value=None)
+    with patch("roboco.services.messaging.get_event_bus", return_value=mock_bus):
+        msg = await svc.send_message(
+            MessageCreateRequest(agent_id=aid, session_id=sess.id, content="hi")
+        )
+    mock_bus.publish.assert_awaited()
+    published = mock_bus.publish.await_args.args[0]
+    assert published.type is EventType.MESSAGE_SENT
+    data = published.data
+    assert data["message_id"] == str(msg.id)
+    assert data["session_id"] == str(sess.id)
+    assert data["channel_id"] == str(ch.id)
+    assert data["agent_id"] == str(aid)
+    assert data["content"] == "hi"
+
+
+@pytest.mark.asyncio
+async def test_send_message_bus_failure_does_not_break_send(
+    msg_setup: dict,
+) -> None:
+    """A bus outage during the MESSAGE_SENT publish is logged but never rolls
+    back the persisted message — live delivery is best-effort."""
+    svc = msg_setup["svc"]
+    aid = msg_setup["agent_id"]
+    ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
+    grp = await svc.create_group(GroupCreateRequest(name="g1", channel_id=ch.id))
+    sess = await svc.create_session(SessionCreateRequest(group_id=grp.id))
+    with patch(
+        "roboco.services.messaging.get_event_bus",
+        side_effect=RuntimeError("bus down"),
+    ):
+        msg = await svc.send_message(
+            MessageCreateRequest(agent_id=aid, session_id=sess.id, content="hi")
+        )
+    assert msg.id is not None
+    assert msg.content == "hi"
+
+
 @pytest.mark.asyncio
 async def test_send_message_rejects_empty(msg_setup: dict) -> None:
     svc = msg_setup["svc"]

tests/unit/api/test_websocket_bridge.py

@@ -14,6 +14,7 @@ from uuid import uuid4
 import pytest
 from roboco.api.websocket_bridge import (
     _handle_agent_event,
+    _handle_message_event,
     _handle_notification_sent,
     _handle_rate_limit_event,
     _handle_session_event,
@@ -181,6 +182,102 @@ async def test_handle_session_event_broadcasts() -> None:
         assert call_args.args[1]["type"] == "session.closed"
 
 
+# ---------------------------------------------------------------------------
+# _handle_message_event
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_handle_message_event_skips_missing_ids() -> None:
+    """A MESSAGE_SENT event with no session_id/message_id → no broadcast."""
+    event = _evt(EventType.MESSAGE_SENT, {"content": "x"})
+    with patch("roboco.api.websocket_bridge.manager") as mgr:
+        mgr.broadcast_to_session = AsyncMock()
+        mgr.broadcast_to_channel = AsyncMock()
+        await _handle_message_event(event)
+        mgr.broadcast_to_session.assert_not_called()
+        mgr.broadcast_to_channel.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_handle_message_event_skips_invalid_uuid() -> None:
+    event = _evt(
+        EventType.MESSAGE_SENT,
+        {"session_id": "bad", "channel_id": "bad", "message_id": "bad"},
+    )
+    with patch("roboco.api.websocket_bridge.manager") as mgr:
+        mgr.broadcast_to_session = AsyncMock()
+        mgr.broadcast_to_channel = AsyncMock()
+        await _handle_message_event(event)
+        mgr.broadcast_to_session.assert_not_called()
+        mgr.broadcast_to_channel.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_handle_message_event_broadcasts_to_session_and_channel() -> None:
+    """A MESSAGE_SENT event fans out a `message.new` payload to both the
+    session stream and the channel stream — the payload the panel's
+    useChannelStream/useSessionStream filters on."""
+    sid = uuid4()
+    cid = uuid4()
+    mid = uuid4()
+    aid = uuid4()
+    event = _evt(
+        EventType.MESSAGE_SENT,
+        {
+            "session_id": str(sid),
+            "channel_id": str(cid),
+            "message_id": str(mid),
+            "agent_id": str(aid),
+            "content": "hello",
+            "message_type": "dialogue",
+            "timestamp": "2026-06-30T00:00:00+00:00",
+        },
+    )
+    with patch("roboco.api.websocket_bridge.manager") as mgr:
+        mgr.broadcast_to_session = AsyncMock()
+        mgr.broadcast_to_channel = AsyncMock()
+        await _handle_message_event(event)
+        mgr.broadcast_to_session.assert_awaited_once()
+        mgr.broadcast_to_channel.assert_awaited_once()
+        s_payload = mgr.broadcast_to_session.await_args.args[1]
+        c_payload = mgr.broadcast_to_channel.await_args.args[1]
+        assert s_payload["type"] == "message.new"
+        assert c_payload["type"] == "message.new"
+        assert s_payload["message_id"] == str(mid)
+        assert s_payload["session_id"] == str(sid)
+        assert s_payload["channel_id"] == str(cid)
+        assert s_payload["content"] == "hello"
+        assert mgr.broadcast_to_session.await_args.args[0] == sid
+        assert mgr.broadcast_to_channel.await_args.args[0] == cid
+
+
+@pytest.mark.asyncio
+async def test_handle_message_event_skips_when_no_connections() -> None:
+    """No subscribers on either stream → broadcast helpers still called (they
+    no-op internally), but the UUIDs must resolve without error."""
+    sid = uuid4()
+    cid = uuid4()
+    event = _evt(
+        EventType.MESSAGE_SENT,
+        {
+            "session_id": str(sid),
+            "channel_id": str(cid),
+            "message_id": str(uuid4()),
+            "agent_id": str(uuid4()),
+            "content": "x",
+            "message_type": "dialogue",
+        },
+    )
+    with patch("roboco.api.websocket_bridge.manager") as mgr:
+        mgr.broadcast_to_session = AsyncMock()
+        mgr.broadcast_to_channel = AsyncMock()
+        await _handle_message_event(event)
+        # Forwarder always dispatches; the manager no-ops on empty sets.
+        mgr.broadcast_to_session.assert_awaited_once()
+        mgr.broadcast_to_channel.assert_awaited_once()
+
+
 # ---------------------------------------------------------------------------
 # _handle_agent_event
 # ---------------------------------------------------------------------------
@@ -366,6 +463,8 @@ def test_register_websocket_bridge_handlers_subscribes_all_event_types() -> None
     assert EventType.RATE_LIMIT_LIFTED in types
     # Usage events forwarded to /ws/system.
     assert EventType.USAGE_SNAPSHOT in types
+    # Message delivery forwarded to /ws/channels + /ws/sessions.
+    assert EventType.MESSAGE_SENT in types
 
 
 @pytest.mark.asyncio