Connections =========== Real-time communication in Agency uses **Django Channels** behind nginx (ASGI). The database is always authoritative; WebSocket frames are **projections** — clients must rehydrate from REST after refresh or reconnect. Routing lives in ``agency-server/framework/server/asgi.py``. Producers: ``agency.channel.protocol`` (``Messenger``, ``Monitor``, ``Catalog``), consumers under ``agency.channel.consumer``. Inside the SPA, transport frames are normalized and routed by the namespaced :doc:`../development/ui/event-bus`; socket paths and bus topics are related but are not the same contract. Addressable UI nodes use ``entity_id`` (``channel`` + ``uuid`` [+ ``name``]) as the bus topic — see :doc:`../development/ui/schema`. Live attrs ride that topic as ``{ attrs: {…} }`` and rehydrate from ``GET /api/ui/overlay/``. Window pose is not an overlay key — it lives on the desktop snapshot. Persisted bot panels are ``GET/POST/PATCH /api/ui/panels/``; open is ``panel.{uuid}`` via ``openAddress`` (REST document, not a WS-only tree). Philosophy ---------- +---------------------------+--------------------------------------------------+ | Rule | Meaning | +===========================+==================================================+ | DB is source of truth | Lists and detail reload via REST. | +---------------------------+--------------------------------------------------+ | WS is a projection | Deltas only; may drop under load. | +---------------------------+--------------------------------------------------+ | Separate concerns | Chat ≠ telemetry ≠ Celery ≠ agent-test chat. | +---------------------------+--------------------------------------------------+ | Auth-scoped groups | HITL reaches ``user-{id}`` even if no UI open. | +---------------------------+--------------------------------------------------+ Socket inventory ---------------- .. list-table:: :header-rows: 1 :widths: 22 16 22 40 * - Path - Consumer - Groups joined - Purpose * - ``/channel/`` - ``UserConsumer`` - ``users``, ``user-{id}`` - Product socket: chat, HITL, observe * - ``/channel/task/`` - ``TaskConsumer`` - ``tasks-all`` - Celery task monitor (System → Tasks) * - ``/channel/log/`` - ``LogConsumer`` - ``logs`` - App log frames (prefer SSE ``/stream/logs/`` ops feed) * - ``/channel/thread/`` - ``ThreadConsumer`` - ``threads`` - Legacy all-thread telemetry * - ``/channel/thread//`` - ``ThreadConsumer`` - ``thread-{uuid}`` - Legacy one-thread telemetry * - ``/channel/agent//`` - ``AgentConsumer`` - ephemeral - Agent test chat (Open chat explorer) * - ``/channel/room//`` - ``RoomConsumer`` - ``room-`` - Ephemeral public chat (no DB history) * - ``/channel/chat//`` - host chat app - chat app - Example app — not mission path Which socket for what? ^^^^^^^^^^^^^^^^^^^^^^ .. list-table:: :header-rows: 1 :widths: 40 60 * - Intent - Socket * - Mission execute + HITL - ``/channel/`` only * - Live execution timeline - Same ``/channel/`` (filter ``thread_id``) * - Persistent secretary conversation - REST ``/api/thread/`` hydrate + same ``/channel/`` (filter ``thread_id``, no ``execution_id``) * - Celery ops - ``/channel/task/`` * - Quick agent chat test - ``/channel/agent//`` (ephemeral; does not create Threads) * - Public room chat - ``/channel/room//`` Groups (publish side) --------------------- .. list-table:: :header-rows: 1 :widths: 20 80 * - Group - Who receives * - ``user-{id}`` - That user's ``/channel/`` connection * - ``users`` - All authenticated user sockets * - ``thread-{id}`` - Legacy thread consumers * - ``tasks-all`` - Task monitor clients * - ``logs`` - Log stream consumers * - ``room-`` - Members of that room socket ``Groups(Run)`` → ``threads`` + ``thread-{id}`` + ``user-{thread.user.id}``. Primary socket: ``/channel/`` ----------------------------- Two domain publishers ^^^^^^^^^^^^^^^^^^^^^ **Messenger** — conversation - Wire ``type``: ``message`` - Payload: serialized LangChain ``BaseMessage`` (``type`` / ``role`` / ``content`` / ``id``) - May include ``ts``, ``interruption``, ``thread_id`` **Monitor** — telemetry - Wire ``type``: ``monitor_event`` - Fields: ``event`` (canonical name), ``data``, optional ``ts`` - Examples: ``run_start``, ``tool_call``, ``run_end``, ``interruption_resolved`` **UserConsumer text** — UI affordances (not Messenger) - Wire ``type``: ``text`` - Welcome, Action buttons, echo, interruption acks - Optional ``data`` bag (``Action``, ``Agent``, ``Image``, …) Inbound frame types (server → client) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Chat and telemetry JSON with top-level ``type``. Canonical bus frames are recognised by ``v`` plus ``topic`` (no ``type`` field): +-------------------+--------------------------------------------------+ | Frame | Use | +===================+==================================================+ | ``message`` | Chat timeline (human / ai / tool) | +-------------------+--------------------------------------------------+ | ``text`` | Plain or rich text + optional ``data.Action`` | +-------------------+--------------------------------------------------+ | ``monitor_event`` | Run telemetry | +-------------------+--------------------------------------------------+ | envelope | Bus (``v`` + ``topic``): presence, catalog | +-------------------+--------------------------------------------------+ | ``ui.*`` | Schema mutations (:doc:`../development/ui/index`)| +-------------------+--------------------------------------------------+ Outbound (client → server) ^^^^^^^^^^^^^^^^^^^^^^^^^^ Chat:: { "message": "hello" } HITL answer:: { "interruption": { "id": "", "answer": { "message": "…", "text": "…" } } } Action click:: { "action": "click", "data": { "id": "…", "kwargs": { … } } } Clients **answer** interruptions; resume runs via Celery on the server — do not call resume from the client on the happy path. Catalog observe (bus envelope) ------------------------------------- Listable rows stay live without polling: 1. ``observe(Model, resource=…, card=…)`` in Django (``agency.channel.observe``) 2. ``Catalog.emit`` → ``publish_catalog_changed`` → group ``users`` as ``agency_frame`` 3. SPA ``dispatchInboundMessage`` → bus → catalog projector → ``applyDelta`` Topic ``..catalog.``, kind ``state``. Payload is the same delta explorers already apply:: { "v": 1, "topic": "execution..catalog.created", "kind": "state", "payload": { "resource": "execution", "action": "created", "id": "…", "record": { }, "scope": { "workspace_id": "…" } }, "subject": { "type": "execution", "id": "…" } } A leftover ``monitor_event`` / ``catalog_changed`` is ignored so it cannot double-apply. Architecture: :doc:`../development/ui/event-bus`. Desktop chrome (bus envelope) ----------------------------- ``open_window`` / ``close_window`` publish on ``user-{id}``:: { "v": 1, "topic": "desktop.displays.visibility.changed", "kind": "state", "payload": { "open": true, "address": "displays" }, "subject": { "type": "window", "id": "displays" } } A Display name uses ``address: "display."`` and ``props.displayId``. The SPA projector calls ``openAddress``; this is not a kiosk assignment. ``show_media`` also publishes ``desktop.media.now_playing.changed`` (image or YouTube) and opens the ``media`` window. YouTube plays only while that window is visible. Registered resources: ``agent``, ``mission``, ``execution``, ``tool``, ``model``, ``provider``, ``document``. Task monitor (``/channel/task/``) --------------------------------- Frames: ``welcome``, ``task_started``, ``task_progress``, ``task_success``, ``task_failure``. REST hydrate: ``GET /api/task/``. Product UI: System → Tasks tab. Agent test socket (``/channel/agent//``) ---------------------------------------------- Separate from mission execution. Client sends: * ``{ "message": "…" }`` — text turn * ``{ "message": "…", "photo_ids": "uuid,uuid" }`` — Photos already in the library (no base64 on the wire; drop from Photos / pins uses this) * ``{ "message": "…", "images": [{ "mime": "image/jpeg", "data": "" }] }`` — PC drops (JPEG/PNG/WebP; ≤1.5 MiB decoded each). Images apply to the current turn only; session history stays text. Requires the agent's model ``vision=True``. Daphne allows 16 MiB frames (default was 1 MiB — two Photos PNGs closed the socket with 1006 before the consumer ran). * ``{ "action": "reset" }`` — clear session history Server replies: ``welcome`` (includes ``vision`` bool), ``processing`` (``has_images`` when applicable), ``event``, ``stream_complete``, ``reset``, ``error``. Boot seeds a **Portero** greeter agent (prefers a vision model) for camera watch. Ephemeral rooms (``/channel/room//``) ------------------------------------------- Explicit exception to the "DB is source of truth" rule: rooms do not persist messages. The WebSocket is the only live state; reconnects lose backlog. Authentication is required (anonymous connections close with code 4401). Slug pattern: ``[-a-z0-9]+``. Wire protocol: * **Client → server:** ``{ "message": "…" }`` * **Server → client message:** ``type=message`` with ``user_id``, ``username``, ``message`` * **Server → client presence:** ``type=presence`` with ``count``, ``user_ids`` Presence tracks unique authenticated users (same user in multiple tabs counts as 1). Broadcast on join and leave. .. note:: Presence is tracked in-process on the consumer class. This works for a single ASGI worker; multi-process deployments would need Redis-backed presence. Nginx already proxies ``/channel/*`` — no host change required. HITL flow --------- 1. Tool ``ask_human`` → ``Interruption`` PENDING + checkpoint 2. ``Messenger`` publishes to user groups 3. SPA ``HitlOverlay``; user sends interruption answer on ``/channel/`` 4. ``UserConsumer`` saves answer → signal → ``resume_mission`` (Celery) Alternate: ``PATCH /api/interruption//`` — same DB outcome. Related code ------------ .. list-table:: :header-rows: 1 :widths: 20 80 * - Area - Path (agency-django unless noted) * - ASGI routes - ``agency-server/framework/server/asgi.py`` * - Protocol - ``agency/channel/protocol.py`` * - User consumer - ``agency/channel/consumer/user.py`` * - Room consumer - ``agency/channel/consumer/room.py`` * - Observe - ``agency/channel/observe.py`` * - Signals - ``agency/signals/wire.py`` * - UI inbound - ``agency-ui/src/services/inbound.ts`` API reference: :doc:`agency/channels` (autodoc).