> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blobhub.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Listener

> Add a listener to a session over the WebSocket and receive its matching events as frames.

## Overview

A **listener** is a [filter](https://docs.blobhub.io/blob-types/workflow/workflows/event-filter.md) on one session's
[event stream](https://docs.blobhub.io/blob-types/workflow/workflows/session-events.md) that pushes each match to the connection that added it.
It lasts as long as that connection and is gone when the connection closes. Declared listeners, which act inside the
session instead of pushing, are on [Listeners](https://docs.blobhub.io/blob-types/workflow/workflows/listeners.md).

```mermaid theme={null}
sequenceDiagram
    participant Client
    participant Server

    Client->>Server: "status/initialize" request
    Server-->>Client: "status/initialize" response

    Client->>Server: "listener/add" request
    Server-->>Client: "listener/add" response with listener_id

    loop while the session emits matching events
        Server->>Client: "listener/event" message
    end

    Client->>Server: "status/ping" request
    Server-->>Client: "status/ping" response

    Client->>Server: "listener/remove" request
    Server-->>Client: "listener/remove" response
    Client->>Server: close the connection
```

## Add a listener

`listener/add` starts a listener on a session of a revision. The caller needs READ on the revision's blob.

```json theme={null}
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "listener/add",
  "params": {
    "revision_id": "0A06AB3A-DD72-4D42-939A-61DC4EF69418",
    "session": "28E70055-DBB4-4CA8-88BC-133D6CEA68E6",
    "filter": {"types": ["thread.item_posted"], "aliases": ["missions/*/meta"]},
    "batch": {"window_seconds": 5, "max_events": 100}
  }
}
```

| Param | Required | Meaning |
| - | - | - |
| `revision_id` | yes | The revision the session belongs to. |
| `session` | yes | The session's id or alias; it must belong to `revision_id`. |
| `filter` | no | An [event filter](https://docs.blobhub.io/blob-types/workflow/workflows/event-filter.md); without one, every event matches. |
| `batch` | no | Collect events and push them together; see [Batch](#batch). Without it, one event per frame. |

The server answers with the id of the new listener:

```json theme={null}
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {"listener_id": "6C1B0A5E-3F52-4E0B-9B38-0D3B5E2F7A14"}
}
```

A connection holds at most 32 listeners (`workflow.listeners_per_connection`). Frames name their listener by
`listener_id`, which is how one connection tells its listeners apart.

## Remove a listener

`listener/remove` ends a listener of this connection.

```json theme={null}
{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "listener/remove",
  "params": {"listener_id": "6C1B0A5E-3F52-4E0B-9B38-0D3B5E2F7A14"}
}
```

The server answers `{}` in `result`. A `listener_id` this connection does not hold, whether it never existed, was
already removed or belongs to another connection, is answered with a 400 `invalid_params` error and the connection
stays open.

## Receive events

Without `batch`, each matching event arrives in its own frame. The frame is a notification, so it has no `id`.

```json theme={null}
{"jsonrpc": "2.0", "method": "listener/event", "params": {"listener_id": "${listener_id}", "event": {...}}}
```

`event` is the [event record](https://docs.blobhub.io/blob-types/workflow/workflows/session-events.md#record) that `get_session_event` returns.

### Batch

With `batch: {window_seconds?, max_events?, distinct_by?}` events are collected and arrive together:

```json theme={null}
{"jsonrpc": "2.0", "method": "listener/event", "params": {"listener_id": "${listener_id}",
 "batch": {"id": "...", "count": 3, "first_id": "...", "last_id": "..."}, "events": [...]}}
```

* `window_seconds` (default 0) opens at the first pending event's own time.
* `max_events` (default 100 with a `batch`) counts entries after `distinct_by`.
* `distinct_by` is up to 8 dotted paths. Events that agree on every path are one entry, which keeps the newest event.
* `count` is the number of entries in `events`; `first_id` and `last_id` are the ids of the first and last of them.
* A frame is cut before it passes 120 KiB. An event that alone passes 120 KiB is never pushed; read it with
  `get_session_event`.

An invalid `batch` is answered `invalid_params` and no listener is added.

### Resume

Pushes are at least once and can be lost, and a connection lasts at most two hours, so a client that must not miss an
event pairs the listener with the listing:

1. Connect, send `status/initialize`, and `listener/add` first, so nothing that happens next is missed.
2. Read what came before with `list_session_events` in ascending order, with `after: <last id held>` when there is one.
3. Read again from the boundary id of a timestamp a minute older than the last event seen, because a row can commit
   after a row with a larger id. This is the
   [settle time](https://docs.blobhub.io/blob-types/workflow/workflows/session-events.md#settle-time) recipe.
4. Drop every id already held, from pushes and listing alike. An id is the event's identity.
5. After a reconnect, repeat from step 1 with `after` set to the last id held. Listeners do not survive a connection,
   so add them again.

The [worker](https://docs.blobhub.io/worker/session-agent-harness/overview.md) is a client of this kind: it follows a session's events and
reacts to the thread items posted in it.

## Keep the connection

`status/initialize` must be the first request on a connection; any other method but `status/ping` before it is a 405
and the server closes the connection. `status/ping` answers `{}` in every phase and writes nothing. API Gateway closes
a connection idle for 10 minutes, so ping at least that often. It also closes every connection after 2 hours, whatever
its traffic. See [Lifecycle](https://docs.blobhub.io/web-socket/protocol/lifecycle.md).

## Errors

An error answers the request that caused it, in the [error shape](https://docs.blobhub.io/web-socket/protocol/overview.md#responses).

| Code | `message` | `data` | Cause | Connection |
| - | - | - | - | - |
| 400 | the reason, by clause | `{"error": "invalid_params"}` | A filter, a `batch` or a `listener_id` that is not valid. | stays open |
| 400 | the limit's message | `{"error": "too_many_listeners", "limit": {"id": "workflow.listeners_per_connection", "value": 32}}` | The connection already holds 32 listeners. | stays open |
| 400 | `Invalid syntax` | none | The request does not match the method's parameters. | stays open |
| 403 | `Access forbidden` | none | No READ on the blob, or the revision or session is not found. | stays open |
| 405 | `Method not supported` | none | Unknown method, or one not allowed in the current phase. | closed |
| 500 | `Unexpected failure` | none | The server failed; retry. | stays open |

A message that is not valid JSON-RPC, or that names an internal method, closes the connection without a reply. A 403
does not say whether the revision or session exists.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.