Agent Events (SSE)

Real-time push-based agent event streaming. Replaces polling-based status updates with Postgres LISTEN/NOTIFY piped through Server-Sent Events (SSE). Refs #5024

---

Overview

Agent events flow through Postgres NOTIFY channels and are exposed as SSE streams. Each agent gets its own channel. A global channel aggregates all agent events for admin dashboards.

Service calls notify()
    |
    v
Postgres NOTIFY on agent_events_{agent_id}
    |
    v
Dedicated asyncpg LISTEN connection (not from pool)
    |
    v
SSE endpoint streams events to client

Key design decisions:

• Dedicated connection: LISTEN uses a standalone asyncpg connection, not the SQLAlchemy pool. Long-lived SSE connections don't consume pool slots.
• Auto-truncation: Payloads exceeding 7.5 KB are replaced with a slim notification containing a fetch_url for the full payload.
• 30-second keepalive: Clients receive keepalive pings every 30 seconds to prevent connection timeouts.

---

Endpoints

Method	Path	Description
GET	/api/v1/cloud/agents/{agent_id}/events/stream	SSE stream for a specific agent
GET	/api/v1/cloud/agents/events/stream	SSE stream for all agents (admin)
POST	/api/v1/cloud/agents/{agent_id}/events/emit	Manually emit an event (testing)

---

Consuming Events

Per-Agent Stream

const source = new EventSource(
  'https://api.ainative.studio/api/v1/cloud/agents/agent-abc-123/events/stream',
  { headers: { 'Authorization': 'Bearer <token>' } }
);

source.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data.type, data);
};

Admin Stream (All Agents)

const source = new EventSource(
  'https://api.ainative.studio/api/v1/cloud/agents/events/stream'
);

source.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(`[${data.agent_id}]`, data.type, data);
};

---

Event Format

Standard Event

{
  "agent_id": "agent-abc-123",
  "type": "status_changed",
  "timestamp": "2026-07-13T10:30:00+00:00",
  "data": {
    "status": "RUNNING",
    "task_id": "task-456"
  }
}

Connected Event (first event on stream open)

{
  "type": "connected",
  "agent_id": "agent-abc-123",
  "channel": "agent_events_agent_abc_123",
  "timestamp": "2026-07-13T10:30:00+00:00"
}

Keepalive Ping

{
  "type": "keepalive",
  "timestamp": "2026-07-13T10:30:30+00:00"
}

Truncated Event (payload > 7.5 KB)

When a payload exceeds the Postgres NOTIFY size limit, the event is replaced with a pointer:

{
  "agent_id": "agent-abc-123",
  "type": "large_output",
  "timestamp": "2026-07-13T10:30:00+00:00",
  "truncated": true,
  "fetch_url": "/api/v1/cloud/agents/agent-abc-123/events/latest"
}

---

Emitting Events (Server-Side)

Any backend service can emit events using the AgentEventBus:

from app.services.agent_event_bus import AgentEventBus

bus = AgentEventBus(db=session)
bus.notify(
    agent_id="agent-abc-123",
    event_type="status_changed",
    payload={"status": "RUNNING"}
)

Manual Emit (Testing)

POST /api/v1/cloud/agents/agent-abc-123/events/emit
Authorization: Bearer <token>
Content-Type: application/json

{
  "event_type": "status_changed",
  "payload": {
    "status": "COMPLETED",
    "result": "Task finished successfully"
  }
}

---

Common Event Types

Type	Description
connected	Client connected to SSE stream
keepalive	Periodic ping (every 30s)
status_changed	Agent status transition (IDLE, RUNNING, PAUSED, ERROR)
task_started	Agent began executing a task
task_completed	Agent finished a task
task_failed	Agent task failed with error
heartbeat	Agent heartbeat signal
log	Agent log output

---

Channel Naming

Postgres NOTIFY channels follow the pattern agent_events_{sanitized_id} where dashes in the agent ID are replaced with underscores:

Agent ID	Channel
agent-abc-123	agent_events_agent_abc_123
my-agent	agent_events_my_agent
bot42	agent_events_bot42

The global admin channel is simply agent_events.

---

Key Source Files

File	Purpose
src/backend/app/services/agent_event_bus.py	Core event bus: notify() + listen() + listen_all()
src/backend/app/api/v1/endpoints/agent_events_sse.py	SSE endpoints for per-agent and admin streams
src/backend/tests/test_agent_event_bus.py	Unit tests for notify and payload handling