Log in Create workspace
Skip to content

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
Start here First connection Getting connected Mint and connect Channels and scopes Events Constraints Errors and lifecycle Troubleshooting

Start here

Use a WebSocket to receive messages and state changes without polling. Each connection subscribes to named topics. HTTP still creates credentials and reads saved resource state.

Before you begin

  • A machine credential with sockets:connect and the scopes required by your chosen events.
  • A Phoenix Channels client, or a raw client implementing Phoenix v2 frames and heartbeats.
  • The service-agent ID returned with the socket token; it is not a user or robot ID.

Put it into practice

  1. Mint a token

    POST /api/v1/socket_tokens returns socket_token and service_agent_id. The token expires after 600 seconds; mint a fresh one for later reconnects.

  2. Connect and join

    Phoenix clients use /socket; raw WebSockets use /socket/websocket with Phoenix v2 parameters. Join agent:{service_agent_id} and wait for the ok reply.

  3. Keep both heartbeats running

    Send transport heartbeat and agent.heartbeat every 20 seconds. Phoenix handles transport; your integration sends agent.heartbeat. The browser example runs two heartbeat loops. Clean up timers when disconnecting.

What success looks like

A successful join and heartbeat replies confirm connectivity. Subscribe before triggering work; acknowledge message.receive with message.delivered so the durable inbox records delivery.

Getting connected

3 steps

01

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

02

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

03

Join a scoped topic. Agents send native events such as message.send and task.create; the server emits command.dispatch and robot.status_change to authorized subscribers.

SDKs can fetch the matching machine-readable contract at /api/v1/realtime/agent-communication.json .
Operator browser sessions can join workspace, approval, fleet, trace, and replay 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

Run this server-side with curl and a credential that has sockets:connect. HTTP 200 returns socket_token, service_agent_id, workspace_id, granted scopes, and expires_in: 600. Keep the returned token private.

bash 
# Load AGENT_API_KEY from your secret manager first.
curl --fail-with-body --silent --show-error \
  --request POST \
  --header "Authorization: Bearer $AGENT_API_KEY" \
  --header "Accept: application/json" \
  https://robotscenter.net/api/v1/socket_tokens

Illustrative response; the token is redacted and your granted scopes may differ:

json 
{
  "scopes": [
    "sockets:connect"
  ],
  "socket_path": "/socket",
  "expires_in": 600,
  "service_agent_id": "a91720d1-1c45-4343-bb45-786e20432f04",
  "workspace_id": "e65ef764-9b2c-4e24-b918-99c7be33506a",
  "socket_token": "<short-lived-socket-token>"
}

Connect and join

/socket

Run this helper in a trusted server process with the service_agent_id returned above and a getSocketToken callback that mints a new 600-second token using a server-held credential on every connection attempt. Phoenix params callbacks are synchronous, so token minting happens before constructing each socket. Never put a long-lived machine key in browser JavaScript. A successful join logs “Agent channel joined”; add your event handlers after verifying this first connection.

javascript 
import {Socket} from "phoenix"

// Run in a trusted server process. getSocketToken must mint a fresh token
// through your backend using a server-held credential, never a browser key.
// Install the phoenix client package in your integration first.
export function connectAgent({serviceAgentId, getSocketToken}) {
  const retryDelays = [1_000, 2_000, 5_000, 10_000, 30_000]
  let socket
  let channel
  let agentHeartbeat
  let retryTimer
  let attempts = 0
  let stopped = false

  const stopHeartbeat = () => {
    clearInterval(agentHeartbeat)
    agentHeartbeat = undefined
  }

  const retire = () => {
    const oldChannel = channel
    const oldSocket = socket
    // Clear first: leave/disconnect may synchronously fire close callbacks.
    channel = undefined
    socket = undefined
    stopHeartbeat()
    if (oldChannel && !oldChannel.isClosed?.()) oldChannel.leave()
    if (oldSocket) oldSocket.disconnect() // Cancels Phoenix's own reconnect timer.
  }

  const retry = () => {
    if (stopped || retryTimer) return
    const delay = retryDelays[Math.min(attempts++, retryDelays.length - 1)]
    retryTimer = setTimeout(() => {
      retryTimer = undefined
      void connect()
    }, delay)
  }

  const connect = async () => {
    try {
      // Await outside Phoenix params: its callback is synchronous.
      const socketToken = await getSocketToken()
      if (stopped) return
      socket = new Socket("https://robotscenter.net/socket", {
        params: {socket_token: socketToken},
        heartbeatIntervalMs: 20_000
      })
      const currentSocket = socket
      channel = socket.channel(`agent:${serviceAgentId}`, {})
      const currentChannel = channel
      socket.onClose(() => {
        if (stopped || socket !== currentSocket) return
        retire()
        retry()
      })
      channel.onClose(() => {
        if (stopped || channel !== currentChannel) return
        // The server may stop only this channel (e.g. liveness timeout).
        retire()
        retry()
      })
      channel.onError(stopHeartbeat)
      socket.connect()
      channel.join()
        .receive("ok", () => {
          if (stopped || channel !== currentChannel) return
          attempts = 0
          console.log("Agent channel joined")
          stopHeartbeat()
          agentHeartbeat = setInterval(() => {
            channel.push("agent.heartbeat", {})
              .receive("error", reason => console.error("Heartbeat refused", reason))
          }, 20_000)
        })
        .receive("error", reason => {
          if (stopped || channel !== currentChannel) return
          console.error("Join refused", reason)
          retire()
          retry()
        })
        .receive("timeout", () => {
          if (stopped || channel !== currentChannel) return
          console.error("Join timed out")
          retire()
          retry()
        })
    } catch (error) {
      if (!stopped) {
        console.error("Socket token mint failed", error)
        retry()
      }
    }
  }

  void connect()
  // Call this before replacing the client; pending token mints are ignored.
  return () => {
    stopped = true
    clearTimeout(retryTimer)
    retryTimer = undefined
    retire()
  }
}

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", {"load":0.7}]

// After reconnect/rejoin, replace join_ref with the new successful join ref.
// The transport event is "heartbeat" on "phoenix", not "phoenix:heartbeat".
// The raw {"load":0.7} payload becomes data.metadata, not data.metadata.metadata.

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 using data.service_agent_id first, or data.target_service_agent_id only when service_agent_id is absent.
trace:{trace_id} Operator or service agent with events:read Trace creation, event append, updates, and finalization. Any published event whose data includes trace_id (or generated_trace_id when trace_id is absent) is also routed here, such as approval_request.created, failure_group.created and replay events for the generated trace.
replay:{replay_id} Operator or service agent with events:read 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, decision, and execution updates.
fleet:{workspace_id} Operator or service agent with events:read Workspace-wide fleet events: robot heartbeats and status changes, mission lifecycle updates, maintenance work-order transitions, fleet alert notifications, geofence breach/clear transitions, 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

JSON values keep their types

Nested null and boolean values are sent as JSON null/true/false, not the strings "nil", "true" or "false". Enum atoms are strings, timestamps use ISO 8601, and absent optional fields (including an unknown OTA previous_version) can be omitted rather than sent as null. Existing clients that depended on the accidental string sentinels must adjust.

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

For each event, service_agent_id takes precedence over target_service_agent_id. Events.publish routes to agent:{service_agent_id}, or to agent:{target_service_agent_id} when the first key is absent, not both.

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. Requires messages:send.

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. Requires messages:send.

Field Description
message_id string -- protocol message identifier

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

agent.discover

Discover agents in the authenticated workspace. Channel filter precedence is capability > framework > list: when capability is a string, match exact advertised capability (not semantic matching) and apply availability (default online); otherwise, when framework is a string, filter by framework; otherwise list agents. Availability applies only to the capability branch. Unlike REST discovery, these filters are not combined. Results are Enum.take(limit) with limit default 25, clamped 1..100. Requires agents:read.

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. Requires tasks:write.

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. Requires tasks:write.

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. Requires tasks:write.

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. Requires tasks:write.

Field Description
task_id UUID

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

task.retry

Retry a failed task when retry_count < max_retries. Requires tasks:write.

Field Description
task_id UUID

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

task.subscribe

Subscribe to lifecycle updates for selected workspace tasks. Requires tasks:read. At most 100 ids per call; an unknown id refuses the whole call with task_not_found. An agent holds at most 1,000 task subscriptions (subscription_limit_reached); a task's subscriptions are released once it completes or is cancelled.

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

Reply: %{task_ids: array<UUID> -- every task this agent is subscribed to}

task.unsubscribe

Stop receiving updates for the listed tasks. Requires tasks:read.

Field Description
task_ids array<UUID> -- at most 100

Reply: %{task_ids: array<UUID> -- remaining subscriptions}

group.create

Create an agent group. The leader defaults to the authenticated service agent; an optional leader_service_agent_id may name another same-workspace service agent. Requires groups:write.

Field Description
group object -- name, description, capabilities, metadata, and optional same-workspace leader_service_agent_id

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. Requires groups:write.

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. At most 100 ids per call; an unknown id refuses the whole call with group_not_found. An agent holds at most 1,000 group subscriptions (subscription_limit_reached); a deleted group's subscriptions are released.

Field Description
group_ids array<UUID>

Reply: %{group_ids: array<UUID> -- every group this agent is subscribed to}

group.unsubscribe

Stop receiving updates for the listed groups. Requires groups:read.

Field Description
group_ids array<UUID> -- at most 100

Reply: %{group_ids: array<UUID> -- remaining subscriptions}

presence.subscribe

Subscribe to presence changes for selected service agents. At most 100 distinct ids per call (too_many_service_agent_ids); an id outside the workspace refuses the call with forbidden. Requires presence:read.

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> -- at most 100 distinct ids

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 of at most 4,096 bytes in the server's internal encoding (roughly 3 KB of JSON). The server supplies timestamp and health_score; supplied overrides are ignored. Invalid reports receive an error before recording. Reports count toward the credential's shared socket rate limit and can return rate_limited. The server keeps at most the newest 720 raw reports per agent and ages raw samples out after 2 hours; only hourly aggregates persist longer (90 days). Requires health:write.

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. The recipient must be direct (recipient.agent_id, a UUID); broadcast and capability_match discovery are refused with rpc_requires_direct_recipient, a recipient without agent_id with missing_agent_id, and any other malformed recipient with invalid_recipient. The channel keeps serving other events while the request is pending. payload.timeout_seconds defaults to 30 and is capped at 300. Requires rpc:write.

Field Description
message object -- recipient (direct agent_id), 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. Requires queue:read.

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 agent channel to stop and run its cleanup; this does not guarantee the underlying socket transport closes. A timeout or replacement is a separate connection lifecycle path.

Reply: no reply -- the agent channel 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. Any raw payload fields become data.metadata on the emitted event; do not wrap them in a metadata key. Also send a separate Phoenix transport heartbeat every 20 seconds.

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

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
event member_joined | member_left | group_updated | group_deleted
group_id 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. Generic queued/dispatched commands become cancelled; accepted/running commands retain their status and receive cancel_requested_at. Batch-owned commands use cooperative batch cancellation and may not be terminally cancelled after possible device delivery.

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 | absent for an unclaimed pending robot
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, origin}; origin is "partner" or "runtime" on marketplace engagement events (the credential that wrote it) and null otherwise
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.updated

Emitted for a correction to a terminal trace that preserves its terminal identity (status and failure signature), not a new finalization. The event carries the current trace snapshot.

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

trace.finalized

Emitted when a trace reaches a terminal status.

Field Description
status "ok" | "error" | "partial"
service_agent_id UUID | nil
trace_id UUID
duration_ms integer | nil
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
trace_id UUID | nil
connector_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 five-minute expiry sweep.

Field Description
reason string -- the reason the freeze was originally raised
scope_type workspace | connector | service_agent
lift_kind manual | expired
lifted_at ISO 8601
freeze_id UUID
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_status succeeded | failed
execution_completed_at ISO 8601
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
failed_cases integer
pass_rate number
passed_cases integer
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 -- raw heartbeat payload fields
service_agent_id UUID

robot.heartbeat

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} when a robot sends a heartbeat. Canonical event normalization preserves nested JSON null, true and false rather than the old string sentinels; enum atoms still become strings. This is an externally visible wire correction, not proof that every external consumer was updated.

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.updated

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} after a committed roster-relevant robot edit (name, manufacturer, model, firmware, capabilities, or tags). The payload names changed fields but does not include arbitrary robot metadata; fetch the scoped robot for current state.

Field Description
robot_id UUID
changed_fields array of field names

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} for mission execution/control requests and robot ACK outcomes. Payload status is the current persisted mission status, not a requested target; a queued execute stays assigned, and a pending pause/resume/cancel leaves status unchanged. Fetch mission detail for execution_status, command_id, execution_result, control_action/status/command_id/result and trace. Only a known_unsent queued execute cancels locally. Legacy unknown or possibly_delivered queued executes get a pending device cancel and are fenced from execute redelivery; their late bound ACK remains valid. An accepted control that loses transport becomes interrupted (outcome unknown), retaining its exact command link and cancel fence; it is not called failed or retried automatically. An uncertain prior control returns control_pending without replacement until its authentic late ACK. Each fleet.mission.pause/resume/cancel command payload is {"mission_id": "mission UUID", "execute_command_id": "current execute command UUID"}; execute_command_id binds the control to that fleet.mission.execute attempt.

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

maintenance_work_order.created

Emitted after a work order and its audit/event evidence commit. Open work is planning only and does not reserve a robot.

Field Description
status open
robot_id UUID
alert_id UUID | nil
work_order_id UUID

maintenance_work_order.updated

Emitted after a scheduled, assignment, start, cancellation, or editable-field change commits. Only in_progress reserves the robot. Delivery may be retried; hydrate current state from the work-order API.

Field Description
status open | scheduled | in_progress | cancelled
robot_id UUID
alert_id UUID | nil
work_order_id UUID

maintenance_work_order.completed

Emitted after completion and audit evidence commit. The completed summary and outcome are available from the work-order API; a separate manual hold can remain.

Field Description
status completed
robot_id UUID
alert_id UUID | nil
work_order_id UUID

geofence.breach

Emitted after a trusted fix enters an unsafe side of an active fence. First unsafe fix alerts immediately; repeated same-side fixes are deduped. The event is also sent to the robot topic and workspace webhooks. Delivery order is not guaranteed: key state by workspace_id, robot_id and geofence_id. An API-hydrated transition_version is an inclusive floor: ignore event versions less than or equal to that snapshot version, even when the event id was not seen. Thereafter accept only strictly larger versions; ignore equal-version retries (stable envelope ids aid duplicate detection). Reject events from an older geometry_revision than the hydrated state.

Field Description
direction enter | exit -- physical side reached
robot_id UUID
alert_id UUID -- fence-specific fleet alert
coordinates [longitude, latitude]
geofence_id UUID
observed_at ISO 8601 -- trusted device observation time or server receipt time
safety_state breached -- string safety snapshot after this transition
transition_version integer -- trusted observed_at as Unix microseconds

geofence.clear

Emitted only after a trusted fix establishes the robot on a fence's safe side. Geometry edits and reactivation while still unsafe retain the incident and emit no false clear. Disabling or deleting a rule never certifies safe return. Apply the same max transition_version rule as geofence.breach; hydrate active state from the robot location-history API before newer events.

Field Description
direction enter | exit -- physical side reached
robot_id UUID
alert_id UUID -- resolved fence-specific fleet alert
coordinates [longitude, latitude]
geofence_id UUID
observed_at ISO 8601
safety_state safe -- string safety snapshot after this transition
transition_version integer -- trusted observed_at as Unix microseconds

fleet_alert.acknowledged

Emitted on fleet:{workspace_id} and, when robot_id is present, fleet:robots:{robot_id} when an operator acknowledges an active fleet alert.

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.resolved

Emitted on fleet:{workspace_id} and, when robot_id is present, fleet:robots:{robot_id} when an active or acknowledged fleet alert is resolved.

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

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"
operator_status "open" | "acknowledged" | "snoozed" | "resolved"
alert_group_key string | nil
resource_id string | nil
alert_id UUID -- the delivery that represents the group
failure_group_id UUID | nil
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
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
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
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"
failed_count integer
skipped_count integer
succeeded_count integer
total_count integer -- targets selected, at most 5000
batch_operation_id UUID

fleet.batch_operation.progress

Emitted on fleet:{workspace_id} after a committed target change on a best-effort post-commit PubSub path. A connection can miss this event; it is not a durable outbox guarantee. Poll the stored batch and paginated targets after reconnect. Counts are recomputed from rows; dispatch_command targets remain queued/running through accepted delivery and become succeeded only on exact terminal device result.

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

fleet.batch_operation.cancel_requested

Emitted on fleet:{workspace_id} when an operator stops new command-batch work. Queued targets and commands not dispatched to a device are skipped/cancelled locally; linked work that may have reached a device remains running until its terminal report. This event is a request acknowledgement, not proof of physical stop.

Field Description
status "running" until all device outcomes are terminal
kind "dispatch_command"
cancel_requested_at ISO 8601 timestamp -- request time
cancelled_by_id UUID -- requesting operator
failed_count integer
skipped_count integer
succeeded_count integer
total_count integer
batch_operation_id UUID
cancel_reason bounded scrubbed request reason, distinct from batch creation reason

fleet.batch_operation.completed

Emitted on fleet:{workspace_id} only when every target is terminal. A cancelled command batch may still record a device success or failure that arrived after the stop-new-work request; cancelled does not prove physical device stop.

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

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 when a credential exceeds its socket budget for the event; includes retry_after_ms. Requests (message.send, agent.discover, rpc.request, health.report, and the task.*, group.*, presence.* and queue.* events) share 1,000 per 60 seconds. Acknowledgements (message.delivered, rpc.response, and command.accepted/progress/complete/fail) share 10,000 per 60 seconds. Joins of agent:{service_agent_id}, agent.ready and agent.heartbeat share 120 per 60 seconds.

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/3, 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 closes the stale agent connection without taking the agent offline; a missing agent heartbeat times out its connection after 60 seconds. The explicit disconnect push stops the agent channel and runs cleanup, but does not guarantee the underlying socket transport closes. Only the last live connection performs offline cleanup and command-disconnect handling. A machine credential is re-validated on every topic join; if it has been revoked, archived, or has expired, the join is refused and that credential's existing sockets are closed. On the agent topic, credential expiry is checked during agent.heartbeat, at most once a minute. Trace, replay and fleet machine joins also schedule a disconnect at expires_at when the remaining lifetime is within the timer limit of about 49.7 days. Mint a new socket token with a live credential before reconnecting.

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.

Troubleshooting

A topic refuses the join
Check identity and scopes. Machine trace, replay, and fleet topics require events:read; workspace topics require an operator session.
Disconnects after a minute
Check agent.heartbeat, not just the transport heartbeat. Restart its timer after rejoining.
Recover a dropped connection
Check expiry, revocation, and workspace status before reconnecting. Reconcile saved state through HTTP; not every stream replays missed events.

Where to go next

see also
Authentication Obtain and scope the machine credential behind your connection. Agent communication Connect messages, tasks, and discovery to your HTTP workflow. Fleet telemetry Understand the saved data behind live robot updates.