6Real-Time Observability

Every step streams observations, reasoning, actions, and metrics via WebSocket. Researchers see what the agent sees, read what it thinks, and can pause/step/resume.

Design doc: Execution Logs

1. Architecture

Two broadcast paths coexist (TODO #38 refactor, 2026-04-10):

GraphExecutor
  │
  ├─ (A) Viewer sinks self-emit per-node viewer_data events
  │         (imageViewer, textViewer, textScroll, actionLog, metrics)
  │           │
  │           └─ _SinkBase.forward() → serialize_for_display() →
  │                broadcast({type: "viewer_data", node_id, step, fields})
  │
  ├─ (B) graphOut sinks buffer _last_inputs → IterOut fires →
  │        _broadcast_step() consolidates into one nav_step event
  │
  ├─ Per-node firing → ExecutionLogger → exec_log event (ADR-observability-003)
  ├─ Status changes → nav_status; loop end → nav_complete
  └─ Batch eval → eval_progress + eval_episode_done (ADR-eval-002)
           │
           └─ broadcast() → WebSocket → all connected clients
                                         │
                                  ws.ts (WSManager)
                                         │
                           ┌──────────────┼──────────────┐
                           ▼              ▼              ▼
                      store.ts       per-node hook     UI panels
                    (navSteps)      useNodeOutput      (render)
                                   (execution_id
                                    + node_id routing)

The old single-path consolidation (every viewer type was special-cased in _broadcast_step) is gone — viewer sinks now own their own serialization and broadcast directly, and the executor only consolidates graphOut for the global NavigatePage / EvalPage store. The _OUTPUT_TYPES registry in the executor was deleted with the refactor.

2. WebSocket Protocol

Endpoint: ws://[host]/ws

2.1 Message Envelope

{
  type: string,             // Event type
  data: unknown,            // Type-specific payload
  timestamp: string,        // ISO 8601
  execution_id?: string     // Canvas execution routing
}

2.2 Event Types

2.3 nav_step Payload

{
  step: number,
  action: number,                      // 0-3
  action_name: string,                 // "FORWARD", "LEFT", "RIGHT", "STOP"
  position: [number, number, number],
  rgb_base64: string,                  // base64 PNG
  depth_base64: string,                // base64 PNG
  response: string,                    // LLM response text
  done: boolean,
  metrics?: Record<string, number>,    // only when done=true
  containers?: Record<string, {        // state container snapshots
    label: string,
    states: Record<string, { type, value_type, size?, value?, preview? }>
  }>
}

3. Output Viewer Nodes

Output viewer nodes are canvas nodes that display data to the researcher. They don't produce outputs — they're sinks. All subclass _SinkBase in agent_loop/builtin_nodes.py, which walks each node's display_fields, runs serialize_for_display() on the wired inputs, and emits one viewer_data WS event per firing.

Viewer sinks serialize and emit their own viewer_data events inside forward(). Only graphOut still buffers inputs in node.state["_last_inputs"] for the iteration-boundary nav_step consolidation. The legacy _OUTPUT_TYPES registry that special-cased every viewer in the executor has been deleted.

4. Broadcast Paths

4.1 Per-node viewer_data (primary path)

Each viewer sink self-emits a viewer_data event at every firing. The frontend's useNodeOutput(nodeId) hook subscribes by node_id, so panels render independently of whether the upstream graph is still mid-iteration.

# _SinkBase.forward() — serialize wired inputs, accumulate if needed, broadcast
fields[key] = serialize_for_display(value)          # IMAGE/DEPTH → base64, etc.
if field.accumulate:                                 # textScroll, actionLog
    history = getattr(ctx, f"vacc_{key}", []) + [fields[key]]
    setattr(ctx, f"vacc_{key}", history)
    fields[key] = list(history)
await broadcast(ctx.session._ws("viewer_data",
    {"node_id": self.node_id, "step": ctx.step, "fields": fields}))

4.2 Consolidated nav_step (graphOut path)

_broadcast_step() still fires at each iterOut boundary (two-pivot model, ADR-dataflow-008) but only scans graphOut sinks — it merges their _last_inputs, converts numpy arrays to base64, and emits one nav_step event for the global navigation store. Graphs without an graphOut sink produce no nav_step — they rely on viewer_data alone.

4.3 Execution logs (exec_log)

The ExecutionLogger (ADR-observability-003) wraps every node firing: exterior I/O captured automatically, interior detail added via self._self_log(key, value) inside forward(). Each firing emits a short exec_log summary; full records stream to JSONL. Canvas runs persist to outputs/runs/{execution_id}/log.jsonl; batch eval persists per episode to outputs/eval_runs/{run_id}/episodes/ep{idx:04d}/log.jsonl and suppresses the WS event (ExecutionPrinciples.suppress_nav_events). See ADR-eval-004 for the per-episode layout rationale.

4.4 Batch eval events

BatchEvalRunner (ADR-eval-002) emits eval_progress + eval_episode_done tagged with worker_id so the Eval page can render per-worker lanes. A run_id groups events into one batch.

4.5 Error reporting (error_event)

The ErrorBus (ADR-observability-004, agentcanvas/backend/app/errors.py) is the unified event channel for everything the user needs to see about a run going wrong — distinct from the structured per-firing capture that exec_log provides. Producers covered today:

• Per-node executor catch (graph_executor.py · _fire_node) — wraps _fire_node(); envelope carries scope = {node_id, node_type, origin, step, execution_id}.
• Graph-level executor catch (graph_executor.py · run) — top-level try/except around the run loop; envelope carries scope = {step, execution_id}.
• FastAPI global Exception handler (main.py) — uncaught HTTP route exceptions; envelope carries scope = {endpoint, method} and is also returned as the 500 body so the frontend HTTP wrapper can recover the same id.
• LogBridgeHandler — installed on the root logger at INFO+; every log.info / warning / error / exception becomes an envelope with source="log". Skips agentcanvas.errors + uvicorn / httpx / httpcore / watchfiles / asyncio prefixes to avoid feedback loops + noise.

Bus dispatch is async over a 500-deep queue: publish() does one ring-buffer append (lock-protected) + loop.call_soon_threadsafe(queue.put_nowait, env), so producers stay loop-agnostic and thread-safe. A single dispatcher coroutine started in the FastAPI lifespan drains the queue and fans out to subscribers; the WS bridge subscriber emits each envelope as {type: "error_event", data: envelope.to_dict()} over /ws. The 200-entry ring buffer is exposed at GET /api/errors for backfill — the frontend calls it on every WS (re)connect to recover envelopes published while the socket was down.

5. Execution ID Routing

Multiple agent graphs can run concurrently. The execution_id mechanism ensures events reach the correct output panels:

1. Frontend generates an exec_<timestamp>_<random> id when the user clicks Play (not a UUID)
2. The id is sent as execution_id in POST /run
3. Backend tags every WebSocket event with the same execution_id
4. Frontend filters events by execution_id and routes to the correct canvas nodes

Graph A (exec_id: "aaa") → Output nodes see only "aaa" events
Graph B (exec_id: "bbb") → Output nodes see only "bbb" events

6. Frontend Wiring

6.1 WSManager (ws.ts)

class WSManager {
  connect()                          // Auto-reconnect with exponential backoff
  on(type: string, handler): unsub   // Subscribe to event type
  on('*', handler): unsub            // Wildcard: all events
}

6.2 Zustand Store (store.ts)

subscribeWS() wires WebSocket events to dataflow store updates:

6.3 Playback Controls

The frontend toolbar sends control commands:

7. Key Files

8. Node Schema API

The frontend fetches node rendering metadata at design time via a REST endpoint. This is distinct from the runtime WebSocket events above — it's a one-shot query used when the canvas loads to know how to render each node type.

8.1 Endpoint

GET /api/components/node-schemas

Returns all registered node types (built-in + loaded nodesets) with port metadata and ui_config.

8.2 Response Shape

[
  {
    "type": "env_habitat__observe",
    "display_name": "Observe",
    "description": "Get current RGB + depth observation",
    "category": "env",
    "icon": "Eye",
    "kind": "block",
    "ports_mode": "mirror",
    "config_schema": {},
    "default_config": {},
    "ui_config": {
      "color": "emerald",
      "layout": "block",
      "width": "",
      "min_height": "",
      "rounding": "",
      "min_width": "",
      "max_width": "",
      "config_fields": [],
      "display_fields": []
    },
    "input_ports": [
      { "name": "trigger", "wire_type": "BOOL", "description": "...", "optional": false }
    ],
    "output_ports": [
      { "name": "rgb", "wire_type": "IMAGE", "description": "...", "optional": false }
    ]
  }
]

8.3 ui_config Fields

config_fields and display_fields are serialized from ConfigField and DisplayField dataclasses defined on each BaseCanvasNode subclass's ui_config ClassVar (ADR-components-004).

8.4 Dynamic Option Injection

Config fields with a __DYNAMIC_PROFILES__ sentinel in their options list are replaced at response time with the live LLM profile list from ProfileStore. This enables the LLMCallNode's model selector dropdown to reflect the user's configured profiles without hardcoding.

8.5 Key Files

Status