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 Agency 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 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

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/<uuid>/	ThreadConsumer	thread-{uuid}	Legacy one-thread telemetry
/channel/agent/<uuid>/	AgentConsumer	ephemeral	Agent test chat (Open chat explorer)
/channel/room/<slug>/	RoomConsumer	room-<slug>	Ephemeral public chat (no DB history)
/channel/chat/<id>/	host chat app	chat app	Example app — not mission path

Which socket for what?

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/<uuid>/ (ephemeral; does not create Threads)
Public room chat	/channel/room/<slug>/

Groups (publish side)

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-<slug>	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 (UI)

Outbound (client → server)

Chat:

{ "message": "hello" }

HITL answer:

{
  "interruption": {
    "id": "<uuid>",
    "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 <resource>.<id>.catalog.<action>, kind state. Payload is the same delta explorers already apply:

{
  "v": 1,
  "topic": "execution.<id>.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: Agency 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.<uuid>" 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/<uuid>/)

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": "<base64>" }] } — 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/<slug>/)

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/<id>/ — same DB outcome.

Related code

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: Channels (autodoc).