"""
Agency real-time protocol.
This module defines how Agency communicates execution state and messages
to clients over WebSockets.
Philosophy
----------
- The database is the source of truth.
- Frontends rebuild state from the API and apply real-time deltas via WS.
- WebSocket traffic is a projection of persisted state, not a replacement for it.
- Events and user messages are intentionally separated.
Notes
-----
- ``Monitor`` emits structured runtime and lifecycle events (telemetry).
- ``Messenger`` streams end-user conversation messages.
- ``Catalog`` publishes catalog / schema-style updates for UI surfaces.
- ``Groups`` maps domain objects (Run, User, …) to Channels group names.
- Signals (``agency.signals.wire``) bridge persistence to real-time updates.
- Prefer Monitor/Messenger over legacy ``channel_run`` helpers (deprecated).
- Payloads should be lightweight, idempotent, and replay-safe.
- This module defines protocol and transport concerns, not business logic.
"""
from datetime import datetime
from typing import Any, Optional
from asgiref.sync import async_to_sync
from django.contrib.auth import get_user_model
from channels import DEFAULT_CHANNEL_LAYER
from channels.layers import get_channel_layer, BaseChannelLayer
from agency.utils import serializable
from agency.models import Run
from agency.channel import Publisher
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage, SystemMessage
[docs]
class Groups(list[str]):
"""Resolve domain objects into Django Channels group names.
Groups define how events and messages are routed over WebSockets.
They map Runs, Threads and Users to stable, broadcast-friendly
channel group identifiers.
Notes:
- Group names must be safe for Channels (no colons).
- Groups are derived, never persisted.
- A single event may be routed to multiple groups (e.g. thread + user).
"""
[docs]
def __init__(self, obj: Run | Any | list[str] | None = None):
"""Initialize Groups.
Args:
obj:
- `Run` instance: derives groups for thread and user
- `User` instance: derives groups for that user
- `list[str]`: uses the provided group names as-is
- `None`: creates empty list
"""
if isinstance(obj, get_user_model()):
groups = ["users"]
groups.append(f"user-{obj.id}")
super().__init__(groups)
elif isinstance(obj, Run):
groups = ["threads"]
thread = obj.get_thread()
if thread:
groups.append(f"thread-{thread.id}")
user = thread.user
if user:
groups.append(f"user-{user.id}")
super().__init__(groups)
elif isinstance(obj, list):
super().__init__(obj)
else:
super().__init__()
[docs]
class Messenger(Publisher):
"""Stream end-user conversation messages over WebSockets.
Messenger delivers `BaseMessage` instances (human, assistant, tool, system)
to clients in real time. This is the user-facing message stream.
Notes:
- Messages are streamed for UI consumption.
- Messages reflect execution output but are not the source of truth.
"""
# Icon mapping for message types (used by Monitor.send_message)
ICON_MAP = {
'human': '👤',
'ai': '😑',
'assistant': '😑',
'tool': '🔧',
'system': '💬',
}
[docs]
def __init__(self, user: Any):
"""Initialize Messenger with a user.
The publisher sends messages to the user's channel group.
Args:
user: User instance to send messages to
"""
super().__init__()
self.groups = Groups(user)
self.user = user
[docs]
def stream(self, message: BaseMessage | dict, **kwargs) -> None:
"""Publish a LangChain message to the user's channel.
Automatically determines the icon based on message type and includes
the serialized message data. The message will be sent to the user's
channel group and can be handled by UserConsumer.
Args:
message: LangChain BaseMessage instance or dict with message data
(must have 'type' field if dict)
Example:
publisher = Messenger(user)
publisher.stream(AIMessage(content="Hello!"))
# Or with a dict:
publisher.stream({"type": "ai", "content": "Hello!"})
"""
# Serialize the message
if isinstance(message, BaseMessage):
message_data = message.model_dump()
else:
message_data = serializable(message)
# Send to user channel with type 'message'
self.consumer_handle(
"message",
self.groups,
payload={
"message": message_data,
**kwargs
}
)
[docs]
class Monitor(Publisher):
"""Emit runtime and lifecycle events for a Run.
Monitor produces structured events describing what happens during
execution (state transitions, model and tool activity, interruptions).
These events are intended for telemetry, debugging and admin UIs.
Notes:
- Events describe what happened, not what to do.
- Some events originate from persistence (signals), others from runtime.
- Delivery is best-effort and may be lossy.
"""
[docs]
class Event:
"""Canonical agency event names used over WebSockets.
These constants form the **wire protocol**: JS clients should switch on
these exact strings when handling `"monitor_event"` messages in the
WebSocket consumer.
"""
# Django Signals
CHECKPOINT_CREATED = "checkpoint_created"
"""New Checkpoint created and saved to database."""
RUN_STATUS = "run_status"
"""Run status updated (e.g. from STARTING to RUNNING)."""
# LangGraph Middleware
RUN_START = "run_start"
"""Fired at `RunMiddleware.before_agent` when a run starts."""
MODEL_REQUEST = "model_request"
"""Model invocation requested (before LLM call)."""
MODEL_RESPONSE = "model_response"
"""Model responded (after LLM call, includes token usage and tool_calls)."""
TOOL_CALL = "tool_call"
"""Tool invocation requested by the model."""
TOOL_RESULT = "tool_result"
"""Tool returned successfully."""
TOOL_ERROR = "tool_error"
"""Tool raised an error (non-successful completion)."""
TOOL_INTERRUPT = "tool_interrupt"
"""Tool triggered an explicit interruption (`__interrupt__` channel)."""
TOOL_TIMEOUT = "tool_timeout"
"""Tool did not complete within the expected time window."""
TOOL_CANCEL = "tool_cancel"
"""Tool execution was cancelled by the system or user."""
TOOL_ABORT = "tool_abort"
"""Tool execution was aborted due to a fatal error."""
RUN_INTERRUPT = "run_interrupt"
"""Run was interrupted (e.g. waiting for human-in-the-loop)."""
INTERRUPTION_RESOLVED = "interruption_resolved"
"""Interruption was resolved (answered by user)."""
RUN_END = "run_end"
"""Run finished successfully (terminal state)."""
RUN_ERROR = "run_error"
"""Run ended in an error state (uncaught exception)."""
# Catalog / admin CRUD (DB projection for explorers)
CATALOG_CHANGED = "catalog_changed"
"""A catalog resource was created, updated, or deleted (Agent, Tool, …)."""
[docs]
def __init__(self, instance: Any):
"""Initialize Monitor publisher with an instance.
The publisher uses the `Groups` class to automatically determine
the groups to publish to based on the instance type (e.g., Run).
Args:
instance: The instance to publish events for (typically a Run instance)
"""
super().__init__()
self.groups = Groups(instance)
self.instance = instance
[docs]
def emit(self, event: str, data: Any = None) -> None:
"""Emit an agency event to the appropriate WebSocket groups.
This is the main API used by backend components (e.g. `RunMiddleware`)
to send structured events about a `Run` / `Thread` to the frontend.
It automatically:
- derives target groups via `Groups(instance)` (user/thread-level groups)
- includes the bound `instance` in the payload
- merges any additional `data` dict from the caller
Args:
event: The event type (e.g., Monitor.Event.RUN_START)
data: Optional additional data to include. If None, only the instance
is included in the event data.
Example:
events = Monitor(run)
events.emit(Monitor.Event.RUN_START)
# Or with additional data:
events.emit(Monitor.Event.TOOL_CALL, {"tool_name": "search"})
"""
event_data = {self.instance.__class__.__name__.lower(): self.instance}
if data is not None:
if isinstance(data, dict):
event_data.update(data)
else:
# If data is not a dict, add it as a key-value pair
event_data['extra'] = data
thread_id = None
if hasattr(self.instance, 'thread_id') and self.instance.thread_id:
thread_id = str(self.instance.thread_id)
self.consumer_handle(
"monitor_event",
self.groups,
payload={
"event": event,
"data": event_data,
"thread_id": thread_id,
}
)
[docs]
class Catalog(Publisher):
"""Publish catalog CRUD deltas to the ``users`` group as bus envelopes.
Used by explorers (Agents, …): REST mutates DB; signals emit
``<resource>.<id>.catalog.<action>`` so open UIs stay in sync.
"""
GROUPS = ["users"]
[docs]
def emit(
self,
*,
resource: str,
action: str,
id: str,
record: dict | None = None,
actor_id: int | None = None,
scope: dict | None = None,
) -> None:
"""Emit a catalog change.
Args:
resource: API stem (``agent``, ``tool``, …).
action: ``created`` | ``updated`` | ``deleted``.
id: Primary key as string.
record: Optional card payload (omit/empty on delete).
actor_id: Optional user id that caused the change.
scope: Optional filter hints for clients (``workspace_id``, …).
"""
from agency.bus import publish_catalog_changed
publish_catalog_changed(
resource=resource,
action=action,
id=id,
record=serializable(record) if record is not None else None,
actor_id=actor_id,
scope=serializable(scope) if scope is not None else None,
)