# Realtime events

> Socket.IO connection auth, Valkey pub/sub fan-out, project and conversation rooms, and domain event payloads including agent monitoring.

- Repository: Paca-AI/paca
- GitHub: https://github.com/Paca-AI/paca
- Human docs: https://grok-wiki.com/public/docs/paca-ai-paca-f238b2ab3d25
- Complete Markdown: https://grok-wiki.com/public/docs/paca-ai-paca-f238b2ab3d25/llms-full.txt

## Source Files

- `services/realtime/README.md`
- `services/realtime/src/server.ts`
- `services/realtime/src/session.ts`
- `services/realtime/src/subscriber.ts`
- `docs/ai-agent/realtime-events.md`
- `apps/web/src/lib/socket-client.ts`

---

---
title: "Realtime events"
description: "Socket.IO connection auth, Valkey pub/sub fan-out, project and conversation rooms, and domain event payloads including agent monitoring."
---

Paca pushes live updates through `services/realtime`: a Bun + Socket.IO server that subscribes to the Valkey Pub/Sub channel `paca.events`, routes messages into permission-scoped rooms, and emits them to authenticated web clients. Producers include `services/api` (tasks, docs, notifications, plugins) and `services/ai-agent` (conversation lifecycle and monitoring signals). The nginx gateway exposes the service at `/ws/socket.io`.

## Architecture

```mermaid
sequenceDiagram
    participant API as services/api
    participant Agent as services/ai-agent
    participant Valkey as Valkey (paca.events)
    participant RT as services/realtime
    participant Web as Web client

    API->>Valkey: PUBLISH {type, payload}
    Agent->>Valkey: PUBLISH {type, payload}
    RT->>Valkey: SUBSCRIBE paca.events
    Valkey-->>RT: message
    RT->>RT: Route to Socket.IO room
    RT-->>Web: emit("event" | "notification")
```

| Component | Role |
|-----------|------|
| `paca.events` | Single Pub/Sub channel for immediate fan-out (constant `events.ChannelRealtime` in the API) |
| `services/realtime` | Auth middleware, room membership, Pub/Sub subscriber, `/healthz` |
| Valkey streams (`paca.task_activities`, `paca:agent:events`, …) | Durable pipelines for DB consumers; **not** read by the realtime service |
| nginx gateway | Proxies `/ws/` → `realtime:3001`, stripping the `/ws` prefix so the service sees `/socket.io/` |

API and ai-agent both publish to `paca.events`. The ai-agent also appends full conversation events to the `paca:agent:events` stream for durability; WebSocket delivery uses the direct Pub/Sub path (`publish_realtime` in `services/ai-agent`).

## Connect and authenticate

Clients connect through the gateway, not directly to port 3001.

<ParamField body="path" type="string" default="/ws/socket.io">
Socket.IO path. The gateway strips `/ws/` before forwarding.
</ParamField>

<ParamField body="withCredentials" type="boolean" default="true">
Required for browser clients so the HttpOnly `access_token` cookie is sent.
</ParamField>

<ParamField body="handshake.auth.token" type="string">
Bearer access JWT for programmatic clients. Checked before the cookie.
</ParamField>

<RequestExample>

```ts title="Browser client"
import { io } from "socket.io-client";

const socket = io(window.location.origin, {
  path: "/ws/socket.io",
  withCredentials: true,
  transports: ["websocket", "polling"],
});
```

</RequestExample>

<RequestExample>

```ts title="Programmatic client"
const socket = io("https://your-paca-host", {
  path: "/ws/socket.io",
  auth: { token: "<access-jwt>" },
});
```

</RequestExample>

### Auth flow

The realtime service **does not verify JWTs locally**. On every connection:

1. Extract token from `handshake.auth.token`, or fall back to the `access_token` cookie.
2. Call `GET /api/v1/users/me/global-permissions` with `Authorization: Bearer <token>`.
3. On HTTP 200, decode identity claims (`sub`, `username`, `role`) from the JWT payload and accept the connection.
4. On any non-200 response, reject with `invalid or expired token`.

A Valkey session record (`realtime:session:<socketId>`, 20-minute TTL) stores user identity and cached project permissions for auditing. The raw token stays in memory only.

Socket.IO connection state recovery is enabled with a 2-minute max disconnection window.

## Room model

Rooms gate **which** events a socket receives. Permission checks happen once at join time; the Pub/Sub subscriber performs no per-message authorization.

### Auto-joined rooms

On connect, every authenticated socket joins:

```
user:<userId>:notifications
```

This room receives `notification.*` events without an explicit client `join`.

### Project rooms

When a client emits `join` with `{ projectId }`, the server:

1. Calls `GET /api/v1/projects/:projectId/members/me/permissions`.
2. Silently ignores the request if the user is not a project member (404).
3. Joins only the namespace rooms the user is permitted to see.

| Room | Required permission | Event prefixes |
|------|---------------------|----------------|
| `project:<projectId>:tasks` | `tasks.read` (or `tasks.*`, `*`) | `task.*`, `github.*`, `agent.*` |
| `project:<projectId>:docs` | `docs.read` (or `docs.*`, `*`) | `doc.*` |

Wildcard permissions are honored: `*` grants everything; `tasks.*` covers `tasks.read`.

<Steps>
<Step title="Join a project">

```ts
socket.emit("join", { projectId: "<uuid>" });
```

The server fetches permissions and places the socket into zero, one, or both namespace rooms.

</Step>
<Step title="Leave a project">

```ts
socket.emit("leave", { projectId: "<uuid>" });
```

Removes the socket from both `project:<id>:tasks` and `project:<id>:docs`.

</Step>
<Step title="Verify delivery">

Listen for the unified `event` channel (project-scoped) or `notification` (user-scoped). See [Listen for events](#listen-for-events).

</Step>
</Steps>

### Conversation granularity

There is **no** `conversation:<conversationId>` Socket.IO room. Agent monitoring events are delivered to `project:<projectId>:tasks` like other `agent.*` types. Clients filter on `payload.conversation_id` when they need conversation-level updates.

## Listen for events

Paca uses two Socket.IO event names on the wire:

| Socket.IO event | When emitted | Message shape |
|-----------------|--------------|---------------|
| `event` | Project-scoped domain events (`task.*`, `doc.*`, `agent.*`, `github.*`, plugin topics) | `{ type: string, payload: object }` |
| `notification` | User-scoped `notification.*` events | `{ type: string, payload: object }` |

<ResponseExample>

```json title="Project event envelope"
{
  "type": "task.comment.added",
  "payload": {
    "id": "uuid",
    "task_id": "uuid",
    "project_id": "uuid",
    "activity_type": "comment",
    "content": "...",
    "actor_id": "uuid",
    "created_at": "2026-06-13T10:00:00Z",
    "updated_at": "2026-06-13T10:00:00Z"
  }
}
```

</ResponseExample>

<ResponseExample>

```json title="Notification envelope"
{
  "type": "notification.created",
  "payload": {
    "id": "uuid",
    "recipient_user_id": "uuid",
    "actor_member_id": "uuid",
    "type": "assigned",
    "task_id": "uuid",
    "project_id": "uuid",
    "created_at": "2026-06-13T10:00:00Z"
  }
}
```

</ResponseExample>

The Pub/Sub subscriber drops messages that lack a routable payload, lack `project_id` (for project events), or use an unrecognized type prefix. Malformed JSON is logged and skipped.

### Web app integration pattern

The web app uses a singleton socket (`connectSocket` / `disconnectSocket` in `apps/web/src/lib/socket-client.ts`):

- **Authenticated layout** — connects on login, listens for `notification`, disconnects on logout.
- **Project pages** — `useProjectRealtime(projectId)` emits `join` on mount and maps incoming `event` types to React Query invalidations.

```ts
socket.on("event", ({ type, payload }) => {
  if (type.startsWith("task.")) {
    invalidate(["projects", projectId, "tasks"]);
  }
  if (type.startsWith("agent.")) {
    invalidate(["projects", projectId, "conversations"]);
    if (payload.conversation_id) {
      invalidate(["projects", projectId, "conversations", payload.conversation_id, "events"]);
    }
  }
});
```

## Event producers and type catalog

### Task events (`task.*`)

Published by `services/api` when task activities are created or comments mutate.

| Type | Origin | Notes |
|------|--------|-------|
| `task.created` | Activity stream + Pub/Sub | System activity; persisted via `paca.task_activities` consumer |
| `task.updated` | Activity stream + Pub/Sub | `content` holds field change JSON |
| `task.deleted` | Activity stream + Pub/Sub | |
| `task.attachment.added` | Activity stream + Pub/Sub | |
| `task.attachment.removed` | Activity stream + Pub/Sub | |
| `task.comment.added` | Pub/Sub only | Writes DB directly; not appended to activity stream |
| `task.comment.updated` | Pub/Sub only | |
| `task.comment.deleted` | Pub/Sub only | |
| `agent.session.started` | Activity stream + Pub/Sub | Recorded as a task activity when an agent session begins (e.g. description-write trigger) |

Standard activity payload fields:

<ResponseField name="id" type="uuid">Activity entry ID.</ResponseField>
<ResponseField name="task_id" type="uuid">Affected task.</ResponseField>
<ResponseField name="project_id" type="uuid">Required for routing.</ResponseField>
<ResponseField name="activity_type" type="string">Semantic activity type (may match `type`).</ResponseField>
<ResponseField name="content" type="string">JSON-encoded activity body.</ResponseField>
<ResponseField name="actor_id" type="uuid">Project member ID of the actor, when present.</ResponseField>
<ResponseField name="created_at" type="string">RFC 3339 timestamp.</ResponseField>
<ResponseField name="updated_at" type="string">RFC 3339 timestamp.</ResponseField>

### Doc events (`doc.*`)

Published by `services/api` for document and folder activities.

| Type | Notes |
|------|-------|
| `doc.created`, `doc.updated`, `doc.deleted`, `doc.moved` | Stream + Pub/Sub via `paca.doc_activities` |
| `doc.folder.created`, `doc.folder.updated`, `doc.folder.deleted` | Stream + Pub/Sub |
| `doc.comment.added`, `doc.comment.updated`, `doc.comment.deleted` | Pub/Sub only |

Payload shape mirrors task activities with `project_id` for routing.

### GitHub plugin events (`github.*`)

Plugin-emitted events (for example `github.branch.linked`, `github.pr.linked`) route to `project:<id>:tasks` because they are task-scoped. Payloads include `task_id` and `project_id`.

### Notification events (`notification.*`)

| Type | Routing |
|------|---------|
| `notification.created` | `user:<recipient_user_id>:notifications` via separate `notification` Socket.IO event |

Notification `type` values in the payload: `assigned`, `mentioned`.

### Plugin-emitted events

WASM plugins call `paca.event_emit(topic, payload)` which publishes directly to `paca.events`:

```json
{ "type": "<plugin-topic>", "source": "<plugin-name>", "payload": { ... } }
```

The subscriber reads `payload` (or legacy `data`) for routing. Plugin topics must include `project_id` in the payload and use a recognized prefix (`task.`, `doc.`, `github.`, `agent.`) to reach clients.

## Agent monitoring events

`services/ai-agent` publishes lightweight realtime signals to `paca.events` via `publish_realtime`, while persisting full event bodies to PostgreSQL and the `paca:agent:events` stream.

### Lifecycle events

| Type | When published |
|------|----------------|
| `agent.conversation.finished` | OpenHands run completes successfully |
| `agent.conversation.failed` | Run errors or unhandled exception |

Both carry at minimum:

<ResponseField name="project_id" type="uuid">Routes to `project:<id>:tasks`.</ResponseField>
<ResponseField name="conversation_id" type="uuid">Filter key for monitoring UIs.</ResponseField>

Constants for `agent.conversation.started`, `agent.conversation.paused`, `agent.conversation.resumed`, and `agent.conversation.stopped` exist in the API event catalog; the ai-agent executor currently publishes `finished` and `failed` for terminal states.

### Per-event monitoring signals

During a conversation, each persisted OpenHands SDK event triggers a realtime notification:

```
agent.<sdk_event_type_lowercase>
```

Examples: `agent.cmddrunaction`, `agent.filewriteaction`, `agent.messageevent`.

The Pub/Sub payload is intentionally minimal:

```json
{
  "type": "agent.cmddrunaction",
  "payload": {
    "project_id": "uuid",
    "conversation_id": "uuid"
  }
}
```

Full event bodies (command text, file paths, message content) live in the REST conversation-events API. The monitoring panel (`conversation-view.tsx`) loads events via HTTP and refreshes when realtime invalidates the `["projects", projectId, "conversations", conversationId, "events"]` query key.

SDK events skipped from both DB and realtime include `StreamingDeltaEvent`, `ConversationStateUpdateEvent`, `SystemPromptEvent`, and `ConversationErrorEvent`. Agent `MessageEvent` entries from the event callback are skipped because streaming token callbacks produce richer `agent.messageevent` rows.

### Agent trigger pipeline (not realtime)

Agent **triggers** (`agent.task_assigned`, `agent.comment_mention`, `agent.chat_message`, `agent.description_write`, `agent.stop`) flow through the `paca:agent:triggers` Valkey stream consumed by ai-agent workers. These are control-plane messages, not Socket.IO events.

<AccordionGroup>
<Accordion title="Monitoring a single conversation">

1. Ensure the client has joined the project (`socket.emit("join", { projectId })`).
2. Listen on `event` and filter `type.startsWith("agent.")`.
3. Match `payload.conversation_id` to the active conversation.
4. Re-fetch `GET /api/v1/projects/:projectId/conversations/:conversationId/events` (paginated) on each matching event.

There is no separate conversation room subscription.

</Accordion>
<Accordion title="Common OpenHands event_type values in stored events">

| Stored `event_type` | Typical source |
|---------------------|----------------|
| `MessageEvent` | User or agent conversational message |
| `CmdRunAction` | Shell command |
| `CmdOutputObservation` | Command output |
| `FileReadAction`, `FileWriteAction`, `FileEditAction` | File operations |
| `BrowseURLAction` | Web fetch |
| `AgentThinkAction` | Internal reasoning |
| `AgentFinishAction` | Run complete |
| `ErrorObservation` | Tool error |

Realtime `type` values lowercase the SDK class name (e.g. `CmdRunAction` → `agent.cmddrunaction`).

</Accordion>
</AccordionGroup>

## Configuration

Environment variables for `services/realtime`:

| Variable | Required | Default | Description |
|----------|----------|---------|-------------|
| `PORT` | No | `3001` | HTTP + Socket.IO listen port |
| `REDIS_URL` | **Yes** | — | Same Valkey instance as the API |
| `API_URL` | **Yes** | — | Internal API base URL for auth and permissions (not the public gateway) |
| `CORS_ORIGINS` | No | `http://localhost:3000` | Comma-separated browser origins |
| `LOG_LEVEL` | No | `info` | Pino log level |
| `NODE_ENV` | No | `development` | `production` enables JSON logging |

Gateway WebSocket proxy (`deploy/nginx/gateway.conf`): `/ws/` → `realtime:3001` with 3600s read timeout for long-lived connections.

## Troubleshooting

| Symptom | Likely cause | Check |
|---------|--------------|-------|
| Connection refused immediately | Missing or invalid token | Confirm `access_token` cookie or `auth.token`; verify API returns 200 on `/users/me/global-permissions` |
| Connected but no project events | Not joined or missing permission | Emit `join` with valid `projectId`; confirm `tasks.read` / `docs.read` |
| Agent events missing | Not in tasks room | Requires `tasks.read`; join project first |
| Notifications missing | Wrong Socket.IO listener | Listen on `notification`, not `event` |
| Events stop after ~15 min | Access token expired | Re-authenticate; reconnect socket with fresh token |
| No events at all | Valkey disconnect | Check realtime logs for subscriber errors; confirm `REDIS_URL` matches API |
| Malformed events in logs | Bad Pub/Sub JSON | Inspect publisher; subscriber skips unparseable messages |

Health check: `GET /healthz` on the realtime service returns `{"status":"ok"}`.

## Related pages

<Card href="/platform-architecture" title="Platform architecture" icon="layers">
Monorepo service boundaries, Valkey decoupling, and nginx routing between web, API, realtime, and storage.
</Card>

<Card href="/ai-agents" title="AI agents" icon="bot">
Agent triggers, OpenHands conversation lifecycle, Valkey streams, and how agents participate as project teammates.
</Card>

<Card href="/task-activity-api" title="Task activity API" icon="list">
Activity types, JSONB content shapes, and REST endpoints that complement realtime task events.
</Card>

<Card href="/configuration-reference" title="Configuration reference" icon="settings">
Environment variables across API, web gateway, realtime, and ai-agent services.
</Card>

<Card href="/troubleshooting" title="Troubleshooting" icon="life-buoy">
Stack-wide failures including Valkey connectivity, auth cookies, and health checks.
</Card>