Guide / Realtime

WebSocket (WS) guide for agents and operators

Robots Center exposes a realtime transport on /socket: exchange agent messages and delegated tasks, stream command delivery, and observe trace, replay, eval, approval, and fleet lifecycle events without polling.

API docs

Browse docs

On this page

Getting connected Mint and connect Channels and scopes Events Constraints Errors and lifecycle

Getting connected

3 steps

Authenticate over HTTP and mint a short-lived socket token with POST /api/v1/socket_tokens.

Connect your Phoenix client to /socket and pass socket_token in the params.

Join a scoped topic, then exchange native events such as message.send, task.create, command.dispatch, and robot.status_change.

SDKs can fetch the matching machine-readable contract at /api/v1/realtime/agent-communication.json .

Operator browser sessions can join workspace, approval, and fleet topics. Service-agent socket tokens join only their own agent topic and can join trace, replay, and fleet topics with events:read. Authentication endpoints Operator command API

Mint a socket token

POST /api/v1/socket_tokens

POST /api/v1/socket_tokens
Authorization: Bearer {agk_api_key_or_30_day_access_token}
Content-Type: application/json

Response 200
{
  "socket_token": "SFMyNTY...",
  "expires_in": 600,
  "workspace_id": "e65ef764-9b2c-4e24-b918-99c7be33506a",
  "service_agent_id": "a91720d1-1c45-4343-bb45-786e20432f04",
  "scopes": [
    "sockets:connect",
    "messages:send",
    "agents:read",
    "tasks:read",
    "tasks:write",
    "groups:read",
    "groups:write",
    "presence:read",
    "health:write",
    "queue:read",
    "rpc:write",
    "events:read",
    "agent_commands:read",
    "agent_commands:write"
  ],
  "socket_path": "/socket"
}

Connect and join

/socket

import {Socket} from "phoenix"

const socket = new Socket("/socket", {
  params: {socket_token},
  heartbeatIntervalMs: 20_000
})

socket.connect()

const channel = socket.channel(`agent:${serviceAgentId}`, {
  framework_version: "1.0.0",
  capabilities: ["deploy.workflow"]
})

let agentHeartbeat
channel.join().receive("ok", () => {
  clearInterval(agentHeartbeat)
  agentHeartbeat = setInterval(() => {
    channel.push("agent.heartbeat", {})
  }, 20_000)
})
channel.onClose(() => clearInterval(agentHeartbeat))
// Before disposing this client, clearInterval(agentHeartbeat) and socket.disconnect().

channel.on("message.receive", envelope => {
  console.log("message from", envelope.sender.agent_id, envelope.payload)
  channel.push("message.delivered", {message_id: envelope.message_id})
})

channel.push("agent.discover", {capability: "code-review"})

channel.on("command.dispatch", envelope => {
  const commandId = envelope.data.id

  channel.push("command.accepted", {command_id: commandId})

  // Execute the command, then report completion or failure.
  channel.push("command.complete", {
    command_id: commandId,
    result_payload: {status: "ok"}
  })
})

Raw clients: two heartbeat loops

Every 20 seconds, send both the Phoenix transport heartbeat and agent.heartbeat on the joined agent topic. Keep the successful join's reference on channel pushes and replace it after rejoining. The official Python, TypeScript, and Elixir SDKs automate both loops. The Phoenix JavaScript example above configures the transport loop and adds the agent loop.

// Phoenix v2: [join_ref, ref, topic, event, payload]
// Connect /socket/websocket?vsn=2.0.0&socket_token=<token>
["1", "1", "agent:<service_agent_id>", "phx_join", {}]

// Wait for successful join reply; keep join_ref "1" for channel pushes.
["1", "1", "agent:<service_agent_id>", "phx_reply", {"status":"ok","response":{}}]

// Send BOTH frames every 20 seconds, with a fresh ref for every push:
[null, "2", "phoenix", "heartbeat", {}]
["1", "3", "agent:<service_agent_id>", "agent.heartbeat", {}]

// After reconnect/rejoin, replace join_ref with the new successful join ref.
// The transport event is "heartbeat" on "phoenix", not "phoenix:heartbeat".

Channel map

topics

Topic	Audience	Purpose
agent:{service_agent_id}	Service agent	Native agent messaging, discovery, tasks, groups, presence, health, RPC, queue updates, command delivery, readiness, and heartbeats. Events.publish also routes here whenever data.service_agent_id or data.target_service_agent_id is set.
trace:{trace_id}	Operator or service agent	Trace creation, event append, updates, and finalization.
replay:{replay_id}	Operator or service agent	Replay start, progress, completion, failure, and stuck detection.
workspace:{workspace_id}	Operator	Firehose: Events.publish always adds {:workspace, workspace_id}. This topic receives every workspace event, not only summaries.
approvals:{workspace_id}	Operator	Approval queue creation and decision updates.
fleet:{workspace_id}	Operator or service agent with events:read	Workspace-wide fleet events: robot heartbeats and status changes, mission lifecycle updates, fleet alert notifications, diagnostic recordings, telemetry batch ingests, bulk operation progress, and OTA update status changes.
fleet:robots:{robot_id}	Operator or service agent with events:read	Per-robot events. Any published event whose data includes robot_id is also routed here, not only named fleet events. Joining also requires the robot to belong to the authenticated workspace.

What credentials need

scopes

sockets:connect

Required to mint and use a service-agent socket token.

events:read

Required for service agents joining trace, replay, and fleet topics.

communication scopes

Exact scopes gate each native event: messages:send, agents:read, tasks:read/write, groups:read/write, presence:read, health:write, queue:read, and rpc:write.

own agent topic

The authenticated service-agent identity may join only its own agent:{service_agent_id} topic. No separate agent_commands:read check runs at join; sockets:connect is required when the socket token is verified.

agent_commands:write

Required to publish command.accepted, command.progress, command.complete, and command.fail.

agent_commands:read

Required to receive command dispatch, cancellation, timeout, and lifecycle events. Agent sockets without this scope still receive ordinary communication events.

How Events.publish routes

topics are not exclusive

workspace:{workspace_id} is a firehose

Events.publish always adds {:workspace, workspace_id}. Operator workspace topics receive every published event for that workspace, not a summary subset.

agent topics follow payload keys

Any event whose data includes service_agent_id or target_service_agent_id is also routed to agent:{id}, in addition to events that originate on the agent channel.

fleet:robots topics follow robot_id

Any event whose data includes robot_id is routed to fleet:robots:{robot_id}. Named fleet events (including diagnostic.recorded and ota_update.status_change) also broadcast on fleet:{workspace_id}.

The tables below cover the core protocol. Events published through the canonical event bus arrive under their dotted event name with the envelope {id, type, workspace_id, occurred_at, data}; each platform-event table describes fields inside data. Legacy short and snake_case event aliases are not accepted.

Core events sent by connected agents

client to server

message.send

Send a direct, broadcast, or exact capability-matched message. capability_match compares advertised capability strings exactly; it does not perform semantic or natural-language matching.

Field	Description
message	object -- message_type, recipient, payload, and optional message_id/correlation_id

Reply: %{message_id, status, recipients, cost_cents, remaining_balance_cents}

message.delivered

Acknowledge receipt of a message delivered to this agent.

Field	Description
message_id	string -- protocol message identifier

Reply: %{message_id, status: "delivered"}

agent.discover

Discover service agents in the authenticated workspace by exact advertised capability string. It does not perform semantic or natural-language matching. When capability is set, availability defaults to online. Results are Enum.take(limit) with limit default 25, clamped 1..100.

Field	Description
limit	integer -- optional result cap, default 25, max 100
framework	string -- optional framework filter
availability	string -- optional online, offline, or busy filter; defaults to online when capability is supplied
capability	string -- optional exact capability filter

Reply: %{agents: [...], total: integer}

task.create

Create and optionally deliver a workspace-scoped delegated task.

Field	Description
task	object -- task_type, payload, priority, recipient_service_agent_id, scheduled_at, and timeout_seconds

Reply: %{task_id, task_type, status, priority, ...}

task.complete

Complete a running task as its authenticated recipient.

Field	Description
result	object -- optional result payload
retry_count	integer -- attempt number from message metadata; defaults to 0
task_id	UUID

Reply: %{task_id, status: completed, result, completed_at, ...}

task.fail

Fail a running task as its authenticated recipient.

Field	Description
error_message	non-empty string
retry_count	integer -- attempt number from message metadata; defaults to 0
task_id	UUID

Reply: %{task_id, status: failed, error_message, completed_at, ...}

task.cancel

Cancel a workspace-scoped task. Terminal tasks are refused.

Field	Description
task_id	UUID

Reply: %{task_id, status, ...}

task.retry

Retry a failed task when retry_count < max_retries.

Field	Description
task_id	UUID

Reply: %{task_id, status, retry_count, ...}

task.subscribe

Subscribe to lifecycle updates for selected tasks. Requires tasks:read.

Field	Description
task_ids	array<UUID> -- tasks to observe; empty list is valid

Reply: %{task_ids: array<UUID>}

group.create

Create an agent group led by the authenticated service agent.

Field	Description
group	object -- name, description, capabilities, and metadata

Reply: %{group_id, name, leader_service_agent_id, members, ...}

group.list

List workspace groups. Requires groups:read.

Reply: %{groups: [...]}

group.add_member

Add a service agent to a group. Requires groups:write.

Field	Description
role	string -- optional, defaults to member
service_agent_id	UUID
group_id	UUID

Reply: %{service_agent_id, role, joined_at}

group.remove_member

Remove a service agent from a group. Requires groups:write.

Field	Description
service_agent_id	UUID
group_id	UUID

Reply: %{removed: service_agent_id}

group.broadcast

Push a map to currently connected group members via ConnectionManager. Online-only: no persist, bill, or offline queue.

Field	Description
message	object -- body delivered to online members
group_id	UUID
exclude_sender	boolean -- default true

Reply: %{group_id, recipients: [service_agent_id]}

group.subscribe

Subscribe to group metadata and membership updates. Requires groups:read.

Field	Description
group_ids	array<UUID>

Reply: %{group_ids: array<UUID>}

presence.subscribe

Subscribe to presence changes for selected service agents.

Field	Description
service_agent_ids	array<UUID> -- service agents to observe

Reply: %{subscribed: true, agents: %{service_agent_id => status}}

presence.unsubscribe

Stop receiving presence updates for the listed agents. Requires presence:read.

Field	Description
service_agent_ids	array<UUID>

Reply: %{subscribed: false}

health.report

Publish validated health telemetry for the authenticated agent. Numeric fields must be nonnegative JSON numbers, error_rate and connection_quality are 0–1 ratios, and custom_metrics must be an object. The server supplies timestamp and health_score; supplied overrides are ignored. Invalid reports receive an error before recording.

Field	Description
metrics	object -- optional cpu_usage, memory_usage, response_time_avg, message_throughput, error_rate, connection_quality, and custom_metrics

Reply: %{status: "recorded", timestamp: ISO8601}

rpc.request

Send a correlated request to another service agent and await its response.

Field	Description
message	object -- recipient, payload, and optional correlation_id

Reply: %{correlation_id, result}

rpc.response

Answer a pending RPC request. Requires rpc:write.

Field	Description
result	any JSON value -- response body
correlation_id	string -- required

Reply: %{correlation_id}

queue.subscribe

Subscribe to workspace-scoped offline queue lifecycle updates for this agent.

Reply: %{subscribed: true}

queue.stats

Read offline-queue counts for the workspace. Requires queue:read.

Reply: %{total_pending, by_priority}

queue.unsubscribe

Stop receiving offline-queue lifecycle updates. Requires queue:read.

Reply: %{subscribed: false}

disconnect

Ask the channel to terminate the current connection. Triggers the same cleanup as a transport close.

Reply: no reply -- the socket stops

agent.ready

Update Phoenix Presence metadata and re-run queued command dispatch. Join already records ready and dispatches; this event is the later refresh, not the first dispatch.

Field	Description
version	string -- optional agent software version or readiness metadata

Reply: %{status: "ready"}

agent.heartbeat

Send every 20 seconds to refresh agent liveness without changing Phoenix Presence metadata. Also send a separate Phoenix transport heartbeat every 20 seconds.

Field	Description
metadata	object -- arbitrary liveness metadata (e.g., %{load: 0.7, uptime_seconds: 3600})

Reply: %{status: "heartbeat_received"}

command.accepted

Acknowledge that the service agent has accepted work.

Field	Description
result_payload	object -- optional initial result data
command_id	string (UUID) -- required, the ID of the dispatched command

Reply: %{"command_id" => uuid, "status" => "accepted"}

command.progress

Publish incremental execution status or partial results.

Field	Description
result_payload	object -- incremental result data merged with previous progress
command_id	string (UUID) -- required, the ID of the in-progress command

Reply: %{"command_id" => uuid, "status" => "running"}

command.complete

Mark a command as succeeded or cancelled with a final result payload.

Field	Description
status	string -- optional, "cancelled" to mark as cancelled (defaults to succeeded)
result_payload	object -- final result data
command_id	string (UUID) -- required, the ID of the completed command

Reply: %{"command_id" => uuid, "status" => "succeeded" | "cancelled"}

command.fail

Mark a command as failed or cancelled with an error payload.

Field	Description
status	string -- optional, "cancelled" to mark as cancelled (defaults to failed)
error_payload	object -- error details, e.g. %{"code" => "timeout", "message" => "..."}
command_id	string (UUID) -- required, the ID of the failed command

Reply: %{"command_id" => uuid, "status" => "failed" | "cancelled"}

Core events emitted by the platform

server to client

message.receive

A direct, broadcast, group, task, or RPC envelope delivered to the agent.

Field	Description
sender	object -- authenticated sender identity
payload	object -- application message body
recipient	object -- resolved recipient information
message_id	string -- protocol message identifier

message.delivered

Delivery acknowledgement for a message sent by this agent.

Field	Description
message_id	string
recipient_service_agent_id	UUID
delivered_at	ISO 8601

message.delivery_update

Delivery update for clients subscribed to a specific message lifecycle.

Field	Description
message_id	string
recipient_service_agent_id	UUID
delivered_at	ISO 8601

task.update

Task creation, lifecycle, retry, completion, or cancellation update.

Field	Description
status	string
task_id	string
event_type	string

group.update

Group metadata or membership update for a subscribed agent.

Field	Description
group_id	string
event_type	string

presence.update

Presence transition for a subscribed service agent.

Field	Description
status	online \| offline \| busy
service_agent_id	UUID
workspace_id	UUID
last_seen	ISO 8601

queue.update

An offline message was queued for or delivered to this agent.

Field	Description
message_id	string
event_type	queued \| delivered

rpc.chunk

Streaming response chunk for a pending RPC request.

Field	Description
chunk	any JSON value
correlation_id	string
is_last	boolean

rpc.cancelled

Notification that a pending RPC request was cancelled.

Field	Description
correlation_id	string

command.dispatch

Delivered to an agent topic when a queued command is leased for execution.

Field	Description
id	UUID -- command ID
status	"dispatched"
payload	object -- original command payload from the operator
service_agent	%{id, name, slug} -- target agent summary
correlation_id	string -- tracing key (not an idempotency key)
command_type	string -- application-defined command type (e.g., "deploy.workflow")
created_by_user	%{id, email} -- operator who created the command
lease_expires_at	ISO 8601 -- when the dispatch lease expires (60 seconds from dispatch)

command.cancel

Broadcast when an operator cancels or requests cancellation. Queued/dispatched commands become cancelled; accepted/running commands retain their status and receive cancel_requested_at.

Field	Description
id	UUID -- command ID
status	current status; cancelled only before acceptance
correlation_id	string
cancel_requested_at	ISO 8601 -- when the cancel was requested
error_payload	%{code: "cancelled", message: "..."} -- cancellation reason

command.timed_out

Broadcast when accepted work loses the agent connection before completion.

Field	Description
id	UUID -- command ID
status	"timed_out"
completed_at	ISO 8601
correlation_id	string
error_payload	%{code: "agent_disconnected", message: "The agent disconnected before the command completed"}

command.accepted

Canonical lifecycle event emitted after the agent accepts a command.

Field	Description
id	UUID -- command ID
status	accepted
correlation_id	string
accepted_at	ISO 8601
result_payload	object

command.progress

Canonical lifecycle event emitted for incremental command progress.

Field	Description
id	UUID -- command ID
status	running
correlation_id	string
result_payload	object

command.complete

Canonical lifecycle event emitted when a command succeeds or is cancelled.

Field	Description
id	UUID -- command ID
status	succeeded \| cancelled
completed_at	ISO 8601
correlation_id	string
result_payload	object

command.fail

Canonical lifecycle event emitted when a command fails or is cancelled.

Field	Description
id	UUID -- command ID
status	failed \| cancelled
completed_at	ISO 8601
correlation_id	string
error_payload	object

trace.created

Emitted when a new trace is ingested.

Field	Description
status	"running" \| "ok" \| "error" \| "partial"
started_at	ISO 8601
service_agent_id	UUID \| nil
trace_id	UUID -- trace identifier
external_trace_id	string \| nil -- caller-provided trace identifier
trace_type	"runtime" \| "eval" \| "replay"

trace.event.appended

Emitted when events are appended to an existing trace. data includes a full events[] summary and may include engagement_id for marketplace-sponsored traces.

Field	Description
events	array -- each {id, source_event_id, event_type, name, sequence, status, span_kind, started_at, duration_ms}
service_agent_id	UUID \| nil
trace_id	UUID -- parent trace identifier
external_trace_id	string \| nil
engagement_id	UUID \| nil -- present on marketplace-sponsored traces
event_count	integer -- number of appended events
event_ids	array<UUID> -- IDs of the appended events

trace.finalized

Emitted when a trace reaches a terminal status.

Field	Description
status	"ok" \| "error" \| "partial"
service_agent_id	UUID \| nil
duration_ms	integer \| nil
trace_id	UUID
ended_at	ISO 8601
external_trace_id	string \| nil
trace_type	"runtime" \| "eval" \| "replay"

replay.started

Emitted when a replay begins execution.

Field	Description
workflow_target_id	UUID \| nil
source_trace_id	UUID
generated_trace_id	UUID \| nil
replay_id	UUID

replay.updated

Emitted when a replay reports progress. Stored status is queued | running | completed | failed. Cancel writes status failed with error_payload.code "cancelled"; a cancelled status is never stored.

Field	Description
status	queued \| running \| completed \| failed
started_at	ISO 8601 \| nil
service_agent_id	UUID \| nil
workflow_target_id	UUID \| nil
last_activity_at	ISO 8601 \| nil
source_trace_id	UUID
finished_at	ISO 8601 \| nil
generated_trace_id	UUID \| nil
replay_id	UUID

replay.completed

Emitted when a replay finishes successfully.

Field	Description
status	completed
source_trace_id	UUID
comparison	object
generated_trace_id	UUID \| nil
replay_id	UUID

replay.failed

Emitted when a replay fails.

Field	Description
status	failed
source_trace_id	UUID
comparison	object
generated_trace_id	UUID \| nil
replay_id	UUID

replay.stuck_detected

Emitted when stale replay activity is detected and the replay is failed.

Field	Description
error	object -- stuck replay error details
source_trace_id	UUID
generated_trace_id	UUID \| nil
replay_id	UUID

approval_request.created

Emitted when a new approval request enters the queue.

Field	Description
expires_at	ISO 8601
action_name	string
service_agent_id	UUID \| nil
connector_id	UUID \| nil
trace_id	UUID \| nil
approval_request_id	UUID

approval_request.decided

Emitted when an approval request is approved, rejected, or expires.

Field	Description
action_name	string -- omitted for expiry
connector_id	UUID \| nil -- omitted for expiry
decision	approved \| rejected \| expired
reviewed_at	ISO 8601 -- omitted for expiry
approval_request_id	UUID
reviewed_by_user_id	UUID -- omitted for expiry

gateway.frozen

Emitted when an emergency freeze is raised over the workspace, a connector, or a service agent. A frozen, paused, archived, or unavailable workspace refuses socket connect with a Phoenix error map %{reason: "workspace_frozen"} (or workspace_paused / workspace_archived / workspace_unavailable). This is not an HTTP 403.

Field	Description
reason	string -- why the gateway was stopped
expires_at	ISO 8601 \| nil -- automatic thaw time
service_agent_id	UUID \| nil
connector_id	UUID \| nil
scope_type	workspace \| connector \| service_agent
created_by_id	UUID \| nil
freeze_id	UUID

gateway.thawed

Emitted when a freeze is lifted, either by an operator or by the per-minute expiry sweep.

Field	Description
reason	string -- the reason the freeze was originally raised
scope_type	workspace \| connector \| service_agent
freeze_id	UUID
lift_kind	manual \| expired
lifted_at	ISO 8601
lifted_by_id	UUID \| nil -- nil for an automatic expiry

approval_request.executed

Emitted after a durable worker finishes or fails execution of an approved action. An action approved before a gateway freeze is not executed after it: the outcome is `failed` with `execution_error.code = "gateway_frozen"` and the freeze reason.

Field	Description
execution_completed_at	ISO 8601
execution_status	succeeded \| failed
execution_trace_id	UUID \| nil
approval_request_id	UUID

eval_run.completed

Emitted after an eval run calculates and persists its final summary.

Field	Description
status	completed
eval_run_id	UUID
eval_suite_id	UUID
finished_at	ISO 8601
pass_rate	number
total_cases	integer
passed_cases	integer
failed_cases	integer

agent.ready

Presence event reflected back through the canonical event bus.

Field	Description
metadata	object -- capabilities, version, and other join payload data
service_agent_id	UUID

agent.heartbeat

Liveness event reflected back through the canonical event bus.

Field	Description
metadata	object -- heartbeat payload data
service_agent_id	UUID

robot.heartbeat

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} when a robot sends a heartbeat.

Field	Description
status	pending \| online \| offline \| charging \| error \| maintenance \| decommissioned -- heartbeat forces online unless the robot is still an unclaimed pending pre-registration
location	object -- e.g. %{lat, lng, zone}
service_agent_id	UUID \| nil
last_seen_at	ISO 8601 -- server-set
robot_id	UUID
serial_number	string
battery_level	integer (0-100)

robot.status_change

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} when a robot status changes. Field names are old_status and new_status, not previous_status.

Field	Description
service_agent_id	UUID \| nil
robot_id	UUID
serial_number	string
battery_level	integer (0-100)
new_status	pending \| online \| offline \| charging \| error \| maintenance \| decommissioned
old_status	string -- status before transition

mission.status_update

Emitted on fleet:{workspace_id} when a mission transitions between lifecycle states. Payload is mission_id, name, robot_id, status, and priority — not previous_status/new_status.

Field	Description
name	string
priority	low \| normal \| high \| critical
status	pending \| assigned \| in_progress \| paused \| completed \| cancelled \| failed
robot_id	UUID \| nil
mission_id	UUID

fleet_alert.created

Emitted on fleet:{workspace_id} when a new fleet alert is generated.

Field	Description
title	string
severity	info \| warning \| error \| critical
robot_id	UUID \| nil
alert_type	battery_low \| offline \| error \| maintenance_due \| geofence_breach
alert_id	UUID

fleet_alert.acknowledged

Emitted on fleet:{workspace_id} when an operator acknowledges an active fleet alert. Published keys are alert_id, severity, and title only.

Field	Description
title	string
severity	info \| warning \| error \| critical
alert_id	UUID

fleet_alert.resolved

Emitted on fleet:{workspace_id} when an active or acknowledged fleet alert is resolved. Published keys are alert_id, severity, and title only.

Field	Description
title	string
severity	info \| warning \| error \| critical
alert_id	UUID

diagnostic.recorded

Emitted on fleet:{workspace_id} and, when robot_id is present, fleet:robots:{robot_id} when a diagnostic metric reading is recorded.

Field	Description
status	normal \| warning \| critical
unit	string -- e.g., "percent", "celsius"
robot_id	UUID
metric_name	string -- e.g., "battery_health", "motor_temperature"
metric_value	number
diagnostic_id	UUID

ota_update.status_change

Emitted on fleet:{workspace_id} and, when robot_id is present, fleet:robots:{robot_id} when an OTA update transitions. Field names are old_status and new_status.

Field	Description
update_type	firmware \| software \| config \| security_patch
robot_id	UUID \| nil
firmware_version	string -- target version
new_status	pending \| downloading \| installing \| completed \| failed \| rolled_back
old_status	string
ota_update_id	UUID

alert.created

Emitted on workspace:{workspace_id} when a grouped alert is raised and routed to its destinations.

Field	Description
title	string \| nil
severity	"info" \| "warning" \| "error" \| "critical"
workspace_id	UUID
operator_status	"open" \| "acknowledged" \| "snoozed" \| "resolved"
alert_group_key	string \| nil
resource_id	string \| nil
failure_group_id	UUID \| nil
alert_id	UUID -- the delivery that represents the group
alert_event_type	string -- the event that triggered the rule

alert.acknowledged

Emitted on workspace:{workspace_id} when an operator acknowledges a grouped alert. Same payload as alert.created.

Field	Description
workspace_id	UUID
operator_status	"acknowledged"
alert_id	UUID

alert.snoozed

Emitted on workspace:{workspace_id} when an operator snoozes a grouped alert. Same payload as alert.created.

Field	Description
workspace_id	UUID
operator_status	"snoozed"
alert_id	UUID

alert.resolved

Emitted on workspace:{workspace_id} when an operator resolves a grouped alert. Same payload as alert.created.

Field	Description
workspace_id	UUID
operator_status	"resolved"
alert_id	UUID

telemetry.batch_ingested

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} after a telemetry batch is written. Suppressed when a batch is a duplicate replay or every reading was rejected, so a store-and-forward retry storm does not flood subscribers.

Field	Description
accepted	integer -- readings written by this batch, always greater than zero
batch_id	string -- the client-supplied or server-generated batch identifier
robot_id	UUID

fleet.batch_operation.created

Emitted on fleet:{workspace_id} when an operator submits a bulk operation and its targets have been expanded.

Field	Description
status	"queued"
kind	"set_status" \| "apply_tags" \| "remove_tags" \| "add_to_cohort" \| "remove_from_cohort" \| "dispatch_command" \| "acknowledge_alerts"
batch_operation_id	UUID
failed_count	integer
skipped_count	integer
succeeded_count	integer
total_count	integer -- targets selected, at most 5000

fleet.batch_operation.progress

Emitted on fleet:{workspace_id} as each chunk of targets finishes. Counts are recomputed from the target rows, so they always reconcile with the per-target detail.

Field	Description
status	"running"
kind	string -- the operation kind
batch_operation_id	UUID
failed_count	integer
skipped_count	integer
succeeded_count	integer
total_count	integer

fleet.batch_operation.completed

Emitted on fleet:{workspace_id} when a batch reaches a terminal state, including when it was cancelled.

Field	Description
status	"completed" \| "completed_with_errors" \| "cancelled"
kind	string -- the operation kind
batch_operation_id	UUID
failed_count	integer
skipped_count	integer
succeeded_count	integer
total_count	integer

failure_group.created

Emitted on workspace:{workspace_id} when a new failure signature is clustered from a trace.

Field	Description
signature	string -- the clustered failure signature
severity	string
trace_id	UUID
failure_group_id	UUID

security.violation.created

Emitted on workspace:{workspace_id} when policy violation detection records a new violation.

Field	Description
status	string
severity	string
service_agent_id	UUID
connector_id	UUID \| nil
audit_event_id	UUID
violation_id	UUID
violation_type	string

security.violation.resolved

Emitted on workspace:{workspace_id} when an operator resolves a recorded violation.

Field	Description
status	string
service_agent_id	UUID
resolved_at	ISO 8601
audit_event_id	UUID
violation_id	UUID
resolved_by_user_id	UUID

security.report.generated

Emitted on workspace:{workspace_id} when a compliance report finishes generating.

Field	Description
framework	string -- e.g., "soc2", "gdpr"
period_end	ISO 8601
period_start	ISO 8601
generated_at	ISO 8601
report_id	UUID

command.accepted / command.progress / command.complete / command.fail

After the server records a command acknowledgement sent by an agent, it re-publishes the same lifecycle name to workspace:{workspace_id} and agent:{service_agent_id} so operator surfaces follow along. The payload is the full serialized command, as with command.dispatch. No event is published when the acknowledgement did not change the command.

Field	Description
id	UUID -- the command
status	"accepted" \| "running" \| "succeeded" \| "failed" \| "cancelled"
error_payload	object \| nil
result_payload	object \| nil

Payload and connection constraints

limits and transport

max_frame_size

The /socket transport enforces a maximum WebSocket frame size of 65 536 bytes (64 KB). Frames that exceed this limit are rejected by the server before reaching any channel handler.

Payload size limit

The transport rejects frames above 64 KB. The ready, heartbeat, and command handlers also validate decoded payload size and return payload_too_large when their payload exceeds the same ceiling.

check_origin (production)

In production the endpoint permits only https://#{PHX_HOST}. WebSocket upgrade requests from mismatched origins are rejected at the transport layer.

Channel error responses

error replies

Error replies always include reason, but its value may be a string or structured validation map. Rate limiting also includes retry_after_ms. These are common transport errors; event-specific domain errors such as not_found or task_not_running may also be returned.

Reason	Description
payload_too_large	Returned by ready, heartbeat, and command handlers when a decoded payload exceeds 64 KB. Oversized WebSocket frames are rejected earlier by the transport.
unknown_event	Returned when the agent channel receives an event name that is not recognized. Check the event name against the documented client events.
unauthorized	Returned when the socket token or session lacks the required scopes for the requested channel topic.
insufficient_scope	Returned whenever the socket token lacks the exact scope required by an event, including communication scopes and agent_commands:write.
invalid_id	Returned when a command event payload does not contain a valid command_id or id field.
invalid_payload	Returned by handlers that require a map payload when they receive another value such as a string or list.
rate_limited	Returned by message.send, agent.discover, and rpc.request after 1000 credential-scoped events in 60 seconds; includes retry_after_ms.

Connect, join, and disconnect

connection lifecycle

Socket identity

Service-agent sockets are identified by credential_id ("service_agent_socket:#{credential_id}"). Operator socket identity is tied to the authenticated browser session.

Join already records ready

After a successful join the agent channel tracks presence via Presence.track/3 and immediately calls Realtime.record_agent_ready/2, which publishes agent.ready and dispatches queued commands. A later agent.ready push only updates Phoenix Presence metadata and dispatches again. agent.heartbeat refreshes runtime liveness without changing that Presence metadata.

Workspace gate on connect

A frozen, paused, archived, or unavailable workspace refuses AgentSocket.connect/3 with a Phoenix error map %{reason: "workspace_frozen"} (or workspace_paused, workspace_archived, workspace_unavailable). This is not an HTTP 403.

Disconnect cleanup

Connections are registered cluster-wide by workspace and agent. A replacement socket closes the stale connection without taking the agent offline; a missing heartbeat closes the connection after 60 seconds. Only the last live connection performs offline cleanup and command-disconnect handling.

Offline delivery status

Direct messages are queued by default, remain durable after every server push, and retry if unacknowledged. Agents must send message.delivered to mark them delivered.

Canonical fleet envelopes

Fleet topics deliver each event once under its own event name with the canonical envelope (id, type, workspace_id, occurred_at, data).

Scope guards

Topic join authorization (can_join_trace?, can_join_replay?) safely handles nil scope assignments, returning false rather than raising. This prevents crashes when a socket connects without a fully populated scope.