fix(both): fan user's own outbound message to all canvas sessions (#228)

Adds USER_MESSAGE WebSocket broadcast so a user's own outbound message
appears live on all their other sessions without requiring a manual
refresh.

**Server (Go)**
- New `EventUserMessage` constant in `events/types.go`
- New `extractCanvasUserMessage()` in `a2a_proxy_helpers.go` — parses
  A2A JSON-RPC request body, extracts text parts + file attachments
  from user-role message/send calls
- `logA2ASuccess()` now broadcasts `USER_MESSAGE` when canvas sends a
  message (callerID == "" + method == "message/send" + 2xx)

**Canvas (TypeScript)**
- `useChatSocket`: new `onUserMessage` callback + `USER_MESSAGE` handler
  in the socket event listener
- `ChatTab`: wires `onUserMessage` via `appendMessageDeduped`
- `MobileChat`: same wiring for mobile variant
- New test file: 9 tests covering text/file/text+file/wrong-workspace/
  empty/no-callback/other-events/replay/attachment-filtering cases

**Dedup contract**: the originating session's optimistic insert (useChatSend
step 2) and the incoming USER_MESSAGE broadcast share the same
`appendMessageDeduped` path — within the 3-second window the two collapse
into one bubble, so the user sees exactly one message bubble.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

canvas/src/components/mobile/MobileChat.tsx

@@ -236,6 +236,8 @@ export function MobileChat({
 
   useChatSocket(agentId, {
     onAgentMessage: appendMessageDeduped,
+    // Fan-out user's own outbound message to all sessions (issue #228).
+    onUserMessage: appendMessageDeduped,
     onSendComplete: releaseSendGuards,
   });
 

canvas/src/components/tabs/ChatTab.tsx

@@ -143,6 +143,12 @@ function MyChatPanel({ workspaceId, data }: Props) {
         releaseSendGuards();
       }
     },
+    // Fan-out of user's own outbound message to all sessions (issue #228).
+    // Uses appendMessageDeduped so the originating session collapses its
+    // optimistic copy (same role + content within 3-second window).
+    onUserMessage: (msg) => {
+      history.setMessages((prev) => appendMessageDeduped(prev, msg));
+    },
     onActivityLog: (entry) => {
       if (!sending) return;
       setActivityLog((prev) => appendActivityLine(prev, entry));

canvas/src/components/tabs/chat/__tests__/useChatSocket.userMessage.test.ts

@@ -0,0 +1,216 @@
+// @vitest-environment jsdom
+/**
+ * Tests for USER_MESSAGE event handling in useChatSocket.
+ *
+ * Covers issue #228: a canvas user's own outbound message was not fanned
+ * out to other sessions — the originating session inserted it optimistically,
+ * but other sessions only saw it after a manual refresh.
+ *
+ * The server now broadcasts USER_MESSAGE on canvas message/send. This test
+ * verifies the canvas side consumes and forwards it to onUserMessage.
+ */
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { renderHook, act } from "@testing-library/react";
+import React from "react";
+import { useChatSocket, type UseChatSocketCallbacks } from "../hooks/useChatSocket";
+import { emitSocketEvent, _resetSocketEventListenersForTests } from "@/store/socket-events";
+import type { WSMessage } from "@/store/socket";
+
+// Silence React StrictMode double-invoke noise — we care about final state.
+const WARN = console.warn;
+beforeEach(() => { console.warn = () => {}; });
+afterEach(() => { console.warn = WARN; });
+
+beforeEach(() => {
+  _resetSocketEventListenersForTests();
+  vi.useFakeTimers();
+  vi.setSystemTime(new Date("2026-05-18T10:00:00Z"));
+});
+afterEach(() => {
+  vi.useRealTimers();
+  _resetSocketEventListenersForTests();
+});
+
+const WORKSPACE_ID = "00000000-0000-0000-0000-000000000001";
+
+function makeUserMessageEvent(
+  workspaceId: string,
+  overrides: Partial<{
+    message: string;
+    attachments: Array<{ uri: string; name: string; mimeType?: string; size?: number }>;
+    messageId: string;
+  }> = {},
+): WSMessage {
+  const { message = "Hello, agent!", attachments, messageId } = overrides;
+  const payload: Record<string, unknown> = { message };
+  if (attachments) payload.attachments = attachments;
+  if (messageId) payload.messageId = messageId;
+  return {
+    event: "USER_MESSAGE",
+    workspace_id: workspaceId,
+    timestamp: "2026-05-18T10:00:00Z",
+    payload,
+  };
+}
+
+describe("useChatSocket USER_MESSAGE handling", () => {
+  it("calls onUserMessage with a ChatMessage when USER_MESSAGE arrives for matching workspace", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    const { result } = renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent(makeUserMessageEvent(WORKSPACE_ID, { message: "Hello!" }));
+    });
+
+    expect(onUserMessage).toHaveBeenCalledTimes(1);
+    const msg = onUserMessage.mock.calls[0][0];
+    expect(msg.role).toBe("user");
+    expect(msg.content).toBe("Hello!");
+    expect(typeof msg.id).toBe("string");
+    expect(msg.timestamp).toBe("2026-05-18T10:00:00.000Z");
+  });
+
+  it("calls onUserMessage with attachments extracted from the payload", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent(
+        makeUserMessageEvent(WORKSPACE_ID, {
+          message: "Here is the file",
+          attachments: [
+            { uri: "workspace:/uploads/report.pdf", name: "report.pdf", mimeType: "application/pdf", size: 4096 },
+          ],
+        }),
+      );
+    });
+
+    expect(onUserMessage).toHaveBeenCalledTimes(1);
+    const msg = onUserMessage.mock.calls[0][0];
+    expect(msg.role).toBe("user");
+    expect(msg.content).toBe("Here is the file");
+    expect(msg.attachments).toHaveLength(1);
+    expect(msg.attachments![0].uri).toBe("workspace:/uploads/report.pdf");
+    expect(msg.attachments![0].name).toBe("report.pdf");
+    expect(msg.attachments![0].mimeType).toBe("application/pdf");
+    expect(msg.attachments![0].size).toBe(4096);
+  });
+
+  it("does NOT call onUserMessage when workspace_id does not match", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent(
+        makeUserMessageEvent("00000000-0000-0000-0000-000000000099", { message: "wrong workspace" }),
+      );
+    });
+
+    expect(onUserMessage).not.toHaveBeenCalled();
+  });
+
+  it("does NOT call onUserMessage when message is empty and no attachments", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent(makeUserMessageEvent(WORKSPACE_ID, { message: "" }));
+    });
+
+    expect(onUserMessage).not.toHaveBeenCalled();
+  });
+
+  it("ignores USER_MESSAGE when onUserMessage callback is undefined", () => {
+    const callbacks: UseChatSocketCallbacks = { onAgentMessage: vi.fn() };
+    // Should not throw — undefined callback is guarded
+    expect(() =>
+      renderHook(() => useChatSocket(WORKSPACE_ID, callbacks)),
+    ).not.toThrow();
+
+    const { result } = renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+    act(() => {
+      emitSocketEvent(makeUserMessageEvent(WORKSPACE_ID, { message: "Hello" }));
+    });
+    // No error thrown even without onUserMessage
+  });
+
+  it("other event types do NOT trigger onUserMessage", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent({
+        event: "A2A_RESPONSE",
+        workspace_id: WORKSPACE_ID,
+        timestamp: "2026-05-18T10:00:00Z",
+        payload: {},
+      });
+    });
+
+    expect(onUserMessage).not.toHaveBeenCalled();
+  });
+
+  it("re-fires onUserMessage for each USER_MESSAGE event received", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent(makeUserMessageEvent(WORKSPACE_ID, { message: "First message" }));
+    });
+    act(() => {
+      emitSocketEvent(makeUserMessageEvent(WORKSPACE_ID, { message: "Second message" }));
+    });
+
+    expect(onUserMessage).toHaveBeenCalledTimes(2);
+    expect(onUserMessage.mock.calls[0][0].content).toBe("First message");
+    expect(onUserMessage.mock.calls[1][0].content).toBe("Second message");
+  });
+
+  it("handles USER_MESSAGE with messageId in payload", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent(
+        makeUserMessageEvent(WORKSPACE_ID, { message: "With ID", messageId: "msg-id-abc" }),
+      );
+    });
+
+    expect(onUserMessage).toHaveBeenCalledTimes(1);
+    const msg = onUserMessage.mock.calls[0][0];
+    expect(msg.content).toBe("With ID");
+  });
+
+  it("filters out attachments with empty uri or name (defence-in-depth)", () => {
+    const onUserMessage = vi.fn();
+    const callbacks: UseChatSocketCallbacks = { onUserMessage };
+    renderHook(() => useChatSocket(WORKSPACE_ID, callbacks));
+
+    act(() => {
+      emitSocketEvent(
+        makeUserMessageEvent(WORKSPACE_ID, {
+          message: "Mixed attachments",
+          attachments: [
+            { uri: "workspace:/uploads/good.pdf", name: "good.pdf" },
+            { uri: "", name: "bad.pdf" }, // empty uri — dropped
+            { uri: "workspace:/uploads/also-bad", name: "" }, // empty name — dropped
+            { uri: "workspace:/uploads/also-good.txt", name: "also-good.txt" },
+          ],
+        }),
+      );
+    });
+
+    expect(onUserMessage).toHaveBeenCalledTimes(1);
+    const msg = onUserMessage.mock.calls[0][0];
+    expect(msg.attachments).toHaveLength(2);
+    expect(msg.attachments![0].name).toBe("good.pdf");
+    expect(msg.attachments![1].name).toBe("also-good.txt");
+  });
+});

canvas/src/components/tabs/chat/hooks/useChatSocket.ts

@@ -7,6 +7,10 @@ import { createMessage, type ChatMessage } from "../types";
 
 export interface UseChatSocketCallbacks {
   onAgentMessage?: (msg: ChatMessage) => void;
+  /** Called when another session sent a user message — used to fan out
+   *  the user's own outbound text to all sessions so a second device
+   *  sees the question live without a manual refresh (issue #228). */
+  onUserMessage?: (msg: ChatMessage) => void;
   onActivityLog?: (entry: string) => void;
   onSendComplete?: () => void;
   onSendError?: (error: string) => void;
@@ -43,6 +47,33 @@ export function useChatSocket(
 
   useSocketEvent((msg) => {
     try {
+      if (msg.event === "USER_MESSAGE" && msg.workspace_id === workspaceId) {
+        const p = msg.payload || {};
+        const message = typeof p.message === "string" ? p.message : "";
+        const rawAttachments = p.attachments;
+        const attachments =
+          Array.isArray(rawAttachments)
+            ? (rawAttachments as Array<{ uri?: unknown; name?: unknown; mimeType?: unknown; size?: unknown }>)
+                .filter(
+                  (a) =>
+                    typeof a?.uri === "string" && a.uri.length > 0 &&
+                    typeof a?.name === "string" && a.name.length > 0,
+                )
+                .map((a) => ({
+                  uri: a.uri as string,
+                  name: a.name as string,
+                  mimeType: typeof a.mimeType === "string" ? a.mimeType : undefined,
+                  size: typeof a.size === "number" ? a.size : undefined,
+                }))
+            : undefined;
+        if (message || (attachments && attachments.length > 0)) {
+          callbacksRef.current.onUserMessage?.(
+            createMessage("user", message, attachments),
+          );
+        }
+        return;
+      }
+
       if (msg.event === "ACTIVITY_LOGGED") {
         if (msg.workspace_id !== workspaceId) return;
 

workspace-server/internal/events/types.go

@@ -41,8 +41,9 @@ type EventType string
 // scan-friendly as it grows.
 const (
 	// Chat / agent messaging — surfaces in canvas chat panels.
-	EventAgentMessage EventType = "AGENT_MESSAGE"
-	EventA2AResponse  EventType = "A2A_RESPONSE"
+	EventAgentMessage   EventType = "AGENT_MESSAGE"
+	EventA2AResponse   EventType = "A2A_RESPONSE"
+	EventUserMessage    EventType = "USER_MESSAGE"
 	EventActivityLogged EventType = "ACTIVITY_LOGGED"
 	EventChannelMessage EventType = "CHANNEL_MESSAGE"
 
@@ -95,6 +96,7 @@ const (
 var AllEventTypes = []EventType{
 	EventA2AResponse,
 	EventActivityLogged,
+	EventUserMessage,
 	EventAgentAssigned,
 	EventAgentCardUpdated,
 	EventAgentMessage,

workspace-server/internal/events/types_test.go

@@ -41,6 +41,7 @@ func TestAllEventTypes_IsSnapshot(t *testing.T) {
 		"DELEGATION_STATUS",
 		"EXTERNAL_CREDENTIALS_ROTATED",
 		"TASK_UPDATED",
+		"USER_MESSAGE",
 		"WORKSPACE_AWAITING_AGENT",
 		"WORKSPACE_DEGRADED",
 		"WORKSPACE_HEARTBEAT",

workspace-server/internal/handlers/a2a_proxy_helpers.go

@@ -11,6 +11,7 @@ import (
 	"log"
 	"net/http"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
@@ -344,6 +345,19 @@ func (h *WorkspaceHandler) logA2ASuccess(ctx context.Context, workspaceID, calle
 			"duration_ms":   durationMs,
 		})
 	}
+
+	// #228: fan user's own outbound message to all sessions of the workspace.
+	// When a canvas user sends a message (callerID == "" and method == "message/send"),
+	// the originating session already inserted it optimistically in useChatSend.
+	// Other sessions see nothing until a manual refresh — this broadcast closes
+	// that gap. The originating session collapses its optimistic copy via the
+	// 3-second appendMessageDeduped window (same role + content = deduped).
+	if callerID == "" && a2aMethod == "message/send" && statusCode < 400 {
+		userPayload := extractCanvasUserMessage(body)
+		if userPayload != nil {
+			h.broadcaster.BroadcastOnly(workspaceID, string(events.EventUserMessage), userPayload)
+		}
+	}
 }
 
 func nilIfEmpty(s string) *string {
@@ -393,6 +407,119 @@ func validateCallerToken(ctx context.Context, c *gin.Context, callerID string) e
 // matching (the wsauth errors are typed for the invalid case).
 var errInvalidCallerToken = errors.New("missing caller auth token")
 
+// canvasUserMessage holds the extracted user message extracted from an
+// A2A canvas request body for broadcasting to other sessions.
+type canvasUserMessage struct {
+	Message   string                               `json:"message,omitempty"`
+	Parts     []map[string]interface{}              `json:"parts,omitempty"`
+	MessageID string                               `json:"messageId,omitempty"`
+	Attachments []map[string]interface{}            `json:"attachments,omitempty"`
+}
+
+// extractCanvasUserMessage parses an A2A JSON-RPC request body and extracts
+// the user-authored text and attachments from a canvas-initiated message/send.
+// Returns nil when the body is not a canvas user message (empty, malformed,
+// or not a message/send from canvas). The returned payload is safe to pass
+// directly to BroadcastOnly — nil fields are omitted from JSON.
+func extractCanvasUserMessage(body []byte) map[string]interface{} {
+	if len(body) == 0 {
+		return nil
+	}
+	var top map[string]json.RawMessage
+	if err := json.Unmarshal(body, &top); err != nil {
+		return nil
+	}
+	// Only handle message/send from canvas
+	var method string
+	if err := json.Unmarshal(top["method"], &method); err != nil || method != "message/send" {
+		return nil
+	}
+	params, ok := top["params"]
+	if !ok {
+		return nil
+	}
+	var paramsMap map[string]json.RawMessage
+	if err := json.Unmarshal(params, &paramsMap); err != nil {
+		return nil
+	}
+	msgRaw, ok := paramsMap["message"]
+	if !ok {
+		return nil
+	}
+	var msg map[string]json.RawMessage
+	if err := json.Unmarshal(msgRaw, &msg); err != nil {
+		return nil
+	}
+
+	// role field: only broadcast user-role messages (canvas users)
+	var role string
+	if err := json.Unmarshal(msg["role"], &role); err != nil || role != "user" {
+		return nil
+	}
+
+	result := make(map[string]interface{})
+
+	// Extract messageId if present
+	var mid string
+	if err := json.Unmarshal(msg["messageId"], &mid); err == nil && mid != "" {
+		result["messageId"] = mid
+	}
+
+	// Extract text from parts — accumulate all text parts into a single string
+	var parts []json.RawMessage
+	if err := json.Unmarshal(msg["parts"], &parts); err == nil {
+		var texts []string
+		var fileAttachments []map[string]interface{}
+		for _, pRaw := range parts {
+			var p map[string]json.RawMessage
+			if err := json.Unmarshal(pRaw, &p); err != nil {
+				continue
+			}
+			var t string
+			if err := json.Unmarshal(p["text"], &t); err == nil && t != "" {
+				texts = append(texts, t)
+			}
+			var fileRaw json.RawMessage
+			if err := json.Unmarshal(p["file"], &fileRaw); err == nil && fileRaw != nil {
+				var f map[string]json.RawMessage
+				if err := json.Unmarshal(fileRaw, &f); err == nil {
+					att := make(map[string]interface{})
+					var s string
+					if err := json.Unmarshal(f["uri"], &s); err == nil {
+						att["uri"] = s
+					}
+					if err := json.Unmarshal(f["name"], &s); err == nil {
+						att["name"] = s
+					}
+					if err := json.Unmarshal(f["mimeType"], &s); err == nil {
+						att["mimeType"] = s
+					}
+					var n float64
+					if err := json.Unmarshal(f["size"], &n); err == nil {
+						att["size"] = n
+					}
+					if len(att) > 0 {
+						fileAttachments = append(fileAttachments, att)
+					}
+				}
+			}
+		}
+		if len(texts) > 0 {
+			// Join with newlines — user may have sent multiple text parts
+			result["message"] = strings.Join(texts, "\n")
+		}
+		if len(fileAttachments) > 0 {
+			result["attachments"] = fileAttachments
+		}
+	}
+
+	// Drop empty payloads
+	if len(result) == 0 {
+		return nil
+	}
+	return result
+}
+
 // extractToolTrace pulls metadata.tool_trace from an A2A JSON-RPC response.
 // Returns nil when absent or malformed — callers can pass it straight through.
 func extractToolTrace(respBody []byte) json.RawMessage {

workspace-server/internal/handlers/a2a_proxy_helpers_user_message_test.go

@@ -0,0 +1,262 @@
+package handlers
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+// TestExtractCanvasUserMessage_TextOnly covers the primary path: a canvas user
+// sends a plain text message with no attachments.
+func TestExtractCanvasUserMessage_TextOnly(t *testing.T) {
+	body := []byte(`{
+		"method": "message/send",
+		"params": {
+			"message": {
+				"role": "user",
+				"messageId": "msg-abc-123",
+				"parts": [
+					{"kind": "text", "text": "Hello, agent!"}
+				]
+			}
+		}
+	}`)
+	got := extractCanvasUserMessage(body)
+	if got == nil {
+		t.Fatal("expected non-nil payload for text message")
+	}
+	if got["message"] != "Hello, agent!" {
+		t.Errorf("message = %v, want %q", got["message"], "Hello, agent!")
+	}
+	mid, ok := got["messageId"].(string)
+	if !ok || mid != "msg-abc-123" {
+		t.Errorf("messageId = %v, want %q", got["messageId"], "msg-abc-123")
+	}
+	_, hasAttachments := got["attachments"]
+	if hasAttachments {
+		t.Errorf("unexpected attachments: %v", got["attachments"])
+	}
+}
+
+// TestExtractCanvasUserMessage_FileOnly covers a user message with a file but no text.
+func TestExtractCanvasUserMessage_FileOnly(t *testing.T) {
+	body := []byte(`{
+		"method": "message/send",
+		"params": {
+			"message": {
+				"role": "user",
+				"messageId": "msg-file-456",
+				"parts": [
+					{
+						"kind": "file",
+						"file": {
+							"name": "report.pdf",
+							"uri": "workspace:/uploads/report.pdf",
+							"mimeType": "application/pdf",
+							"size": 4096
+						}
+					}
+				]
+			}
+		}
+	}`)
+	got := extractCanvasUserMessage(body)
+	if got == nil {
+		t.Fatal("expected non-nil payload for file-only message")
+	}
+	if got["message"] != nil {
+		t.Errorf("unexpected message text: %v", got["message"])
+	}
+	attachments, ok := got["attachments"].([]map[string]interface{})
+	if !ok || len(attachments) != 1 {
+		t.Fatalf("attachments = %v, want 1-element array", got["attachments"])
+	}
+	att := attachments[0]
+	if att["uri"] != "workspace:/uploads/report.pdf" {
+		t.Errorf("uri = %v, want %q", att["uri"], "workspace:/uploads/report.pdf")
+	}
+	if att["name"] != "report.pdf" {
+		t.Errorf("name = %v, want %q", att["name"], "report.pdf")
+	}
+	if att["mimeType"] != "application/pdf" {
+		t.Errorf("mimeType = %v, want %q", att["mimeType"], "application/pdf")
+	}
+}
+
+// TestExtractCanvasUserMessage_TextAndFile covers a user message with both text and a file.
+func TestExtractCanvasUserMessage_TextAndFile(t *testing.T) {
+	body := []byte(`{
+		"method": "message/send",
+		"params": {
+			"message": {
+				"role": "user",
+				"parts": [
+					{"kind": "text", "text": "Here is the file:"},
+					{"kind": "text", "text": "see below"},
+					{
+						"kind": "file",
+						"file": {
+							"name": "data.csv",
+							"uri": "workspace:/exports/data.csv",
+							"mimeType": "text/csv",
+							"size": 8192
+						}
+					}
+				]
+			}
+		}
+	}`)
+	got := extractCanvasUserMessage(body)
+	if got == nil {
+		t.Fatal("expected non-nil payload")
+	}
+	// Two text parts are joined with newline
+	if got["message"] != "Here is the file:\nsee below" {
+		t.Errorf("message = %v, want %q", got["message"], "Here is the file:\nsee below")
+	}
+	attachments, ok := got["attachments"].([]map[string]interface{})
+	if !ok || len(attachments) != 1 {
+		t.Fatalf("attachments = %v, want 1-element array", got["attachments"])
+	}
+}
+
+// TestExtractCanvasUserMessage_Malformed covers malformed JSON.
+func TestExtractCanvasUserMessage_Malformed(t *testing.T) {
+	cases := []struct {
+		name string
+		body []byte
+	}{
+		{"not JSON", []byte(`{not valid`)},
+		{"wrong type top-level", []byte(`123`)},
+		{"missing params", []byte(`{"method":"message/send"}`)},
+		{"params not object", []byte(`{"method":"message/send","params":123}`)},
+		{"missing message", []byte(`{"method":"message/send","params":{}}`)},
+		{"message not object", []byte(`{"method":"message/send","params":{"message":123}}`)},
+		{"role missing", []byte(`{"method":"message/send","params":{"message":{"parts":[]}}}`)},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := extractCanvasUserMessage(tc.body); got != nil {
+				t.Errorf("expected nil for %s, got %v", tc.name, got)
+			}
+		})
+	}
+}
+
+// TestExtractCanvasUserMessage_NotUserRole covers agent/workspace callers
+// whose role is not "user" — these should not be broadcast as USER_MESSAGE.
+func TestExtractCanvasUserMessage_NotUserRole(t *testing.T) {
+	cases := []struct {
+		name string
+		body []byte
+	}{
+		{
+			"agent role",
+			[]byte(`{"method":"message/send","params":{"message":{"role":"agent","parts":[{"kind":"text","text":"hello"}]}}}`),
+		},
+		{
+			"assistant role",
+			[]byte(`{"method":"message/send","params":{"message":{"role":"assistant","parts":[{"kind":"text","text":"hello"}]}}}`),
+		},
+		{
+			"empty role",
+			[]byte(`{"method":"message/send","params":{"message":{"role":"","parts":[{"kind":"text","text":"hello"}]}}}`),
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := extractCanvasUserMessage(tc.body); got != nil {
+				t.Errorf("expected nil for role=%s, got %v", tc.name, got)
+			}
+		})
+	}
+}
+
+// TestExtractCanvasUserMessage_NotMessageSend covers non-message/send methods.
+func TestExtractCanvasUserMessage_NotMessageSend(t *testing.T) {
+	cases := []struct {
+		name   string
+		method string
+	}{
+		{"tasks/send", "tasks/send"},
+		{"initialize", "initialize"},
+		{"ping", "ping"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			body, _ := json.Marshal(map[string]interface{}{
+				"method": tc.method,
+				"params": map[string]interface{}{
+					"message": map[string]interface{}{
+						"role":  "user",
+						"parts": []map[string]interface{}{{"kind": "text", "text": "hello"}},
+					},
+				},
+			})
+			if got := extractCanvasUserMessage(body); got != nil {
+				t.Errorf("expected nil for method=%q, got %v", tc.method, got)
+			}
+		})
+	}
+}
+
+// TestExtractCanvasUserMessage_BlankOrEmpty covers text with only whitespace
+// and empty parts arrays.
+func TestExtractCanvasUserMessage_BlankOrEmpty(t *testing.T) {
+	cases := []struct {
+		name string
+		body []byte
+	}{
+		{
+			"empty text part",
+			[]byte(`{"method":"message/send","params":{"message":{"role":"user","parts":[{"kind":"text","text":""}]}}}`),
+		},
+		{
+			"empty parts array",
+			[]byte(`{"method":"message/send","params":{"message":{"role":"user","parts":[]}}}`),
+		},
+		{
+			"whitespace-only text — still included as valid content",
+			[]byte(`{"method":"message/send","params":{"message":{"role":"user","parts":[{"kind":"text","text":"   "}]}}}`),
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := extractCanvasUserMessage(tc.body)
+			if tc.name == "whitespace-only text — still included as valid content" {
+				// Whitespace-only text is valid content — preserve it as-is.
+				// Canvas dedup collapses identical copies; whitespace is not stripped.
+				if got == nil {
+					t.Error("expected non-nil for whitespace-only text")
+				} else if got["message"] != "   " {
+					t.Errorf("message = %q, want %q", got["message"], "   ")
+				}
+				return
+			}
+			if got != nil {
+				t.Errorf("expected nil for %s, got %v", tc.name, got)
+			}
+		})
+	}
+}
+
+// TestExtractCanvasUserMessage_Unicode covers non-ASCII text.
+func TestExtractCanvasUserMessage_Unicode(t *testing.T) {
+	body := []byte(`{
+		"method": "message/send",
+		"params": {
+			"message": {
+				"role": "user",
+				"parts": [
+					{"kind": "text", "text": "こんにちは世界 🌍 日本語"}
+				]
+			}
+		}
+	}`)
+	got := extractCanvasUserMessage(body)
+	if got == nil {
+		t.Fatal("expected non-nil payload for unicode message")
+	}
+	if got["message"] != "こんにちは世界 🌍 日本語" {
+		t.Errorf("message = %v, want %q", got["message"], "こんにちは世界 🌍 日本語")
+	}
+}