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

Getting connected

3 steps

Authenticate over HTTP and mint a short-lived socket token with /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.

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}
})

socket.connect()

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

channel.join()
channel.push("agent.ready", {version: "1.0.0"})

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"}
  })
})

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.
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	Workspace-level summaries and shared lifecycle events.
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 including heartbeats, status changes, diagnostics, telemetry batch ingests, and OTA update status changes. 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 and replay 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.

The tables below cover the core protocol. Events published through the canonical event bus arrive as {id, type, workspace_id, occurred_at, data}; each platform-event table describes fields inside data. Fleet topics additionally emit a compatibility fleet_event wrapper.

Core events sent by connected agents

client to server

message.send

Send a direct, broadcast, or capability-matched message.

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.

Field	Description
availability	string -- optional online, offline, or busy filter when capability is supplied
capability	string -- optional exact capability filter
framework	string -- optional framework 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, and retry settings

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

task.complete

Complete a running task as its authenticated recipient.

Field	Description
result	object -- optional result payload
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
task_id	UUID

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

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, ...}

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

health.report

Publish self-reported health telemetry for the authenticated agent.

Field	Description
metrics	object -- cpu_usage, memory_usage, response_time_avg, error_rate, 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}

queue.subscribe

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

Reply: %{subscribed: true}

agent.ready

Signal readiness, publish the supplied metadata, and trigger queued command dispatch. Capabilities are registered in the channel join payload.

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

Reply: %{status: "ready"}

agent.heartbeat

Refresh ConnectionManager and AgentRegistry liveness without changing Phoenix Presence metadata.

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
command_id	string (UUID) -- required, the ID of the dispatched command
result_payload	object -- optional initial result data

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

command.progress

Publish incremental execution status or partial results.

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

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

command.complete

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

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

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

command.fail

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

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

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
message_id	string -- protocol message identifier
payload	object -- application message body
recipient	object -- resolved recipient information
sender	object -- authenticated sender identity

message.delivered

Delivery acknowledgement for a message sent by this agent.

Field	Description
delivered_at	ISO 8601
message_id	string
recipient_service_agent_id	UUID

message.delivery_update

Delivery update for clients subscribed to a specific message lifecycle.

Field	Description
delivered_at	ISO 8601
message_id	string
recipient_service_agent_id	UUID

task.update

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

Field	Description
event_type	string
status	string
task_id	string

group.update

Group metadata or membership update for a subscribed agent.

Field	Description
event_type	string
group_id	string

presence.update

Presence transition for a subscribed service agent.

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

queue.update

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

Field	Description
event_type	queued \| delivered
message_id	string

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

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
cancel_requested_at	ISO 8601 -- when the cancel was requested
correlation_id	string
error_payload	%{code: "cancelled", message: "..."} -- cancellation reason
id	UUID -- command ID
status	current status; cancelled only before acceptance

command.timed_out

Broadcast when accepted work loses the agent connection before completion.

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

trace.created

Emitted when a new trace is ingested.

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

trace.event.appended

Emitted when an event is appended to an existing trace.

Field	Description
event_count	integer -- number of appended events
event_ids	array<UUID> -- IDs of the appended events
external_trace_id	string \| nil
service_agent_id	UUID \| nil
trace_id	UUID -- parent trace identifier

trace.finalized

Emitted when a trace reaches a terminal status.

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

replay.started

Emitted when a replay begins execution.

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

replay.updated

Emitted when a replay reports progress.

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

replay.completed

Emitted when a replay finishes successfully.

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

replay.failed

Emitted when a replay fails.

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

replay.stuck_detected

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

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

approval_request.created

Emitted when a new approval request enters the queue.

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

approval_request.decided

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

Field	Description
action_name	string -- omitted for expiry
approval_request_id	UUID
connector_id	UUID \| nil -- omitted for expiry
decision	approved \| rejected \| expired
reviewed_at	ISO 8601 -- omitted for expiry
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. While a workspace freeze is in force, API authentication and socket connects for the workspace are refused with 403 workspace_frozen.

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

gateway.thawed

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

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

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
approval_request_id	UUID
execution_completed_at	ISO 8601
execution_status	succeeded \| failed
execution_trace_id	UUID \| nil

eval_run.completed

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

Field	Description
eval_run_id	UUID
eval_suite_id	UUID
failed_cases	integer
finished_at	ISO 8601
pass_rate	number
passed_cases	integer
status	completed
total_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 with updated battery, location, or status.

Field	Description
battery_level	integer (0-100) -- current battery percentage
location	object -- %{lat, lng, zone}
robot_id	UUID
status	"online" \| "offline" \| "charging" \| "error" \| "maintenance"
workspace_id	UUID

robot.status_change

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} when a robot transitions between online, offline, charging, error, or maintenance status.

Field	Description
new_status	"online" \| "offline" \| "charging" \| "error" \| "maintenance"
previous_status	string -- status before transition
robot_id	UUID
workspace_id	UUID

mission.status_update

Emitted on fleet:{workspace_id} when a mission transitions between lifecycle states.

Field	Description
mission_id	UUID
new_status	"pending" \| "assigned" \| "in_progress" \| "paused" \| "completed" \| "cancelled" \| "failed"
previous_status	string -- status before transition
robot_id	UUID \| nil
workspace_id	UUID

fleet_alert.created

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

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

fleet_alert.acknowledged

Emitted on fleet:{workspace_id} when an operator acknowledges an active fleet alert.

Field	Description
acknowledged_at	ISO 8601
acknowledged_by	UUID -- operator user ID
alert_id	UUID
workspace_id	UUID

fleet_alert.resolved

Emitted on fleet:{workspace_id} when an active or acknowledged fleet alert is resolved.

Field	Description
alert_id	UUID
resolved_at	ISO 8601
resolved_by	UUID -- operator user ID
workspace_id	UUID

diagnostic.recorded

Emitted on fleet:robots:{robot_id} when a new diagnostic metric reading is recorded for a robot.

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

ota_update.status_change

Emitted on fleet:robots:{robot_id} when an OTA update transitions between status values.

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

alert.created

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

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

alert.acknowledged

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

Field	Description
alert_id	UUID
operator_status	"acknowledged"
workspace_id	UUID

alert.snoozed

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

Field	Description
alert_id	UUID
operator_status	"snoozed"
workspace_id	UUID

alert.resolved

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

Field	Description
alert_id	UUID
operator_status	"resolved"
workspace_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
batch_operation_id	UUID
failed_count	integer
kind	"set_status" \| "apply_tags" \| "remove_tags" \| "add_to_cohort" \| "remove_from_cohort" \| "dispatch_command" \| "acknowledge_alerts"
skipped_count	integer
status	"queued"
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
batch_operation_id	UUID
failed_count	integer
kind	string -- the operation kind
skipped_count	integer
status	"running"
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
batch_operation_id	UUID
failed_count	integer
kind	string -- the operation kind
skipped_count	integer
status	"completed" \| "completed_with_errors" \| "cancelled"
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
failure_group_id	UUID
severity	string
signature	string -- the clustered failure signature
trace_id	UUID

security.violation.created

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

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

security.violation.resolved

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

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

security.report.generated

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

Field	Description
framework	string -- e.g., "soc2", "gdpr"
generated_at	ISO 8601
period_end	ISO 8601
period_start	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
error_payload	object \| nil
id	UUID -- the command
result_payload	object \| nil
status	"accepted" \| "running" \| "succeeded" \| "failed" \| "cancelled"

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.

Presence tracking

On join, the agent channel tracks presence via Presence.track/3. A subsequent agent.ready call updates Phoenix Presence metadata and dispatches queued commands. agent.heartbeat refreshes runtime liveness without changing that Presence metadata.

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

Queued messages are marked delivered when the server pushes them after reconnect. Agents should still send message.delivered as an application-level receipt acknowledgement.

Fleet channel compatibility push

Fleet topics deliver each event twice, in two shapes. Once under its own event name with the canonical envelope (id, type, workspace_id, occurred_at, data), and once as a fleet_event message carrying %{type, data, timestamp}, where timestamp is the same value as occurred_at. New clients should subscribe by event name; fleet_event exists for existing consumers.

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.