multica/docs/message-paths.md

Message Paths — Desktop / Web / Channel

Three independent paths deliver messages to and from the Hub's agent. All three share the same AsyncAgent instance — they are just different I/O surfaces.

---

Overview

Desktop (Electron IPC)          Web (WebSocket via Gateway)        Channel (Bot API, e.g. Telegram)
        │                                │                                  │
        ▼                                ▼                                  ▼
  localChat:send IPC            client.send → Gateway WS           plugin.gateway (polling/webhook)
        │                                │                                  │
        ▼                                ▼                                  ▼
  hub.ts / ipc/hub.ts           hub.ts / onMessage                  manager.ts / routeIncoming
  clearLastRoute()              clearLastRoute()                    set lastRoute
        │                                │                                  │
        └────────────────► agent.write(text) ◄──────────────────────────────┘
                                     │
                                     ▼
                              AsyncAgent.run()
                                     │
                        ┌────────────┴────────────────┐
                        ▼                              ▼
                  agent.subscribe()              agent.read()
                  (multi-consumer)            (single-consumer iterable)
                        │                              │
               ┌────────┴────────┐                     ▼
               ▼                 ▼              hub.ts / consumeAgent()
        Desktop IPC        Channel Manager             │
     (ipc/hub.ts)       (manager.ts)                   ▼
               │                 │              Gateway WS → Web client
               ▼                 ▼
         localChat:event   Bot API reply
         → renderer        (via lastRoute)

---

Path 1: Desktop (Electron IPC)

Send (User → Agent)

Renderer: sendMessage(text)
  → IPC: localChat:send
    → ipc/hub.ts handler
      → hub.channelManager.clearLastRoute()   // reply stays in desktop
      → agent.write(text)

File: apps/desktop/electron/ipc/hub.ts — localChat:send handler (line ~373)

Receive (Agent → User)

Agent runs LLM
  → pi-agent-core fires AgentEvent
    → Agent.subscribeAll() → AsyncAgent channel + subscribers
      → agent.subscribe() callback in ipc/hub.ts
        → Filter: assistant messages + tool_execution + passthrough (compaction, agent_error)
        → IPC: mainWindow.webContents.send('localChat:event', { agentId, streamId, event })
          → Renderer: use-local-chat.ts onEvent callback
            → chat.handleStream(payload)

Files:

• apps/desktop/electron/ipc/hub.ts — localChat:subscribe handler (line ~248)
• apps/desktop/src/hooks/use-local-chat.ts — onEvent listener (line ~54)
• packages/hooks/src/use-chat.ts — handleStream() (line ~133)

Error Handling

Agent.run() throws / returns error
  → AsyncAgent.write() catch block
    → channel.send(legacy Message)           // for read() consumers (Web)
    → agent.emitMulticaEvent({ type: "agent_error", error })  // for subscribe() consumers
      → ipc/hub.ts subscriber → passthrough event → localChat:event
        → use-local-chat.ts → chat.setError() + setIsLoading(false)

---

Path 2: Web (WebSocket via Gateway)

Send (User → Agent)

Web app: sendMessage(text)
  → GatewayClient.send(hubId, "message", { agentId, content })
    → Socket.io → Gateway server → routes to Hub device
      → hub.ts / onMessage handler
        → channelManager.clearLastRoute()    // reply stays in gateway
        → agentSenders.set(agentId, deviceId)
        → agent.write(content)

File: src/hub/hub.ts — onMessage handler (line ~154)

Receive (Agent → User)

Agent runs LLM
  → pi-agent-core fires AgentEvent
    → Agent.subscribeAll() → AsyncAgent channel + subscribers
      → agent.read() consumed by hub.ts / consumeAgent()
        → Filter: assistant messages + tool_execution + passthrough (compaction, agent_error)
        → client.send(targetDeviceId, StreamAction, { streamId, agentId, event })
          → Socket.io → Gateway → routes to Web client device
            → GatewayClient.onMessage callback
              → use-gateway-chat.ts → chat.handleStream(payload)

Files:

• src/hub/hub.ts — consumeAgent() (line ~314)
• packages/hooks/src/use-gateway-chat.ts — onMessage listener (line ~50)
• packages/hooks/src/use-chat.ts — handleStream() (line ~133)

Error Handling

Agent.run() throws / returns error
  → AsyncAgent.write() catch block
    → channel.send(legacy Message)           // consumed by consumeAgent() → sent as "message" action
    → agent.emitMulticaEvent({ type: "agent_error", error })
      → read() → consumeAgent() → passthrough event → StreamAction
        → GatewayClient → use-gateway-chat.ts → chat.setError() + setIsLoading(false)

Note: Legacy error Messages also reach the Web client as "message" action (a plain text fallback). The agent_error event provides structured error info for proper UI rendering.

---

Path 3: Channel (Bot API, e.g. Telegram)

Send (User → Agent)

User sends message in Telegram
  → grammy long-polling receives Update
    → plugin.gateway.start() callback: onMessage(channelMessage)
      → ChannelManager.routeIncoming()
        → Set lastRoute = { plugin, deliveryCtx }   // reply goes back to Telegram
        → agent.write(text)                          // same as desktop/web

File: src/channels/manager.ts — routeIncoming() (line ~233)

Receive (Agent → User)

Agent runs LLM
  → pi-agent-core fires AgentEvent
    → Agent.subscribeAll() → AsyncAgent channel + subscribers
      → agent.subscribe() callback in ChannelManager.subscribeToAgent()
        → Check: if (!lastRoute) return         // no active channel route, skip
        → Filter: only assistant messages
        → message_start → createAggregator()    // MessageAggregator buffers/chunks text
        → message_update → aggregator.handleEvent()
        → message_end → aggregator.handleEvent() → null aggregator
          → Aggregator emits text blocks
            → Block 0: plugin.outbound.replyText(deliveryCtx, text)   // Telegram reply
            → Block N: plugin.outbound.sendText(deliveryCtx, text)    // follow-up messages

Files:

• src/channels/manager.ts — subscribeToAgent() (line ~151), createAggregator() (line ~205)
• src/hub/message-aggregator.ts — text chunking/buffering logic

Error Handling

Agent.run() throws / returns error
  → AsyncAgent.write() catch block
    → agent.emitMulticaEvent({ type: "agent_error", error })
      → subscribe() → ChannelManager subscriber
        → if lastRoute exists:
          → plugin.outbound.sendText(deliveryCtx, "[Error] ${errorMsg}")

---

Comparison Table

Aspect	Desktop (IPC)	Web (WebSocket)	Channel (Bot API)
Transport	Electron IPC	Socket.io via Gateway	Bot API (HTTP)
Send entry	localChat:send	client.send → Gateway	routeIncoming
Receive method	agent.subscribe()	agent.read() (iterable)	agent.subscribe()
Consumer	ipc/hub.ts subscriber	hub.ts consumeAgent()	manager.ts subscriber
Frontend hook	use-local-chat.ts	use-gateway-chat.ts	N/A (Bot API)
State hook	use-chat.ts	use-chat.ts	N/A
Reply routing	Always (IPC channel)	agentSenders Map	lastRoute pattern
clearLastRoute	Yes (on send)	Yes (on send)	No (sets lastRoute)
Error display	agent_error → UI	agent_error → UI	agent_error → Bot text
Tool results	Rendered in UI	Rendered in UI	Skipped (text only)
Text chunking	No (full stream)	No (full stream)	Yes (MessageAggregator)

---

lastRoute Pattern

The lastRoute tracks which channel last sent a message. When the agent replies:

• If lastRoute is set → reply goes to that channel (e.g. Telegram)
• If lastRoute is null → reply goes to Desktop/Web only (via their own mechanisms)

Clearing: Desktop and Web both call channelManager.clearLastRoute() before agent.write(), so channel replies stop when the user switches to desktop/web.

Setting: routeIncoming() sets lastRoute when a channel message arrives.

Desktop and Web always receive agent events regardless of lastRoute — they use their own independent delivery mechanisms (IPC subscribe / Gateway read).

---

Event Filtering

All three paths filter raw agent events. Only these are forwarded to consumers:

Event Type	Desktop	Web	Channel
message_start	assistant only	assistant only	assistant only
message_update	assistant only	assistant only	assistant only
message_end	assistant only	assistant only	assistant only
tool_execution_start	Yes	Yes	No
tool_execution_end	Yes	Yes	No
compaction_start	Yes (passthrough)	Yes (passthrough)	No
compaction_end	Yes (passthrough)	Yes (passthrough)	No
agent_error	Yes (passthrough)	Yes (passthrough)	Yes (→ text)
User message events	Filtered out	Filtered out	Filtered out