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

# Session events and listeners

> blobhub.types.workflow.Session: read a session's event stream, follow it, and switch its listeners

`Workflow.session()` returns a `Session`, a handle on one session of a workflow revision. It reads the session's [event
stream](https://docs.blobhub.io/blob-types/workflow/workflows/session-events.md), lists its [listeners](https://docs.blobhub.io/blob-types/workflow/workflows/listeners.md)
and switches them, and lists the executions a listener started. Everything goes through the revision's data channel, so
the SDK opens no WebSocket: following the stream is polling.

```python theme={null}
import blobhub
from blobhub.types.workflow import Workflow

hub = blobhub.connect()
workflow = Workflow(hub.blob("acme", "support-bot").revision())

session = workflow.session("sdk-demo")     # reads the session now; never creates one
for event in session.events({"types": ["thread.item_posted"]}, limit=20):
    print(event["created_at"], event["type"], event["object"]["alias"])
```

Events are plain `dict` records, as everywhere in the SDK. The SDK checks no event type: the taxonomy is the
platform's, so a new type needs no SDK release.

## The session handle

`workflow.session(session, *, verify=True)` takes a session id or alias.

| `verify` | What happens |
| :- | :- |
| `True`, the default | Reads the session, so one that is absent or unreachable fails here with `PermissionDenied`. The handle keeps the record. |
| `False` | Sends nothing and holds only the name given. It is how to read the stream of a session that was deleted: that stream is still served by id while it lasts. |

It never creates a session, unlike `execute(session=...)`.

| Attribute | Holds |
| :- | :- |
| `id` | The record's id, or the name given when unverified. |
| `alias`, `status`, `record` | From the record read; `None` when none was read. |

`session.refresh()` reads the record again. A handle that holds a record sends its **id** on every read, so one opened
by alias keeps reading the same session even if that session is deleted and the alias is reused. An unverified handle
sends the name it was given.

A finished run leads back to its session: `run.session` is the `Session` for the execution's `session_id`, and
`run.source` says what started it.

## Reading events

`session.events(filter=None, *, after=None, before=None, ascending=True, limit=None, page_size=100, follow=False,
interval=2.0, timeout=None)` returns a lazy iterator over the stream.

| Argument | Meaning |
| :- | :- |
| `filter` | `{types?, aliases?, executions?, where?}`, the [event filter](https://docs.blobhub.io/blob-types/workflow/workflows/event-filter.md). The server checks it; a refusal is `CommandError` (`invalid_filter`) naming the clause. |
| `after`, `before` | Exclusive event ids that bound the range. Build one from a time with [`boundary_id`](#boundary-ids). |
| `ascending` | Oldest first, the default. The platform's own default is newest first, so the SDK reverses it; pass `False` to read newest first. |
| `limit` | The total number of events to yield, then stop. |
| `page_size` | Events per request, 1 to 1000. |
| `follow`, `interval`, `timeout` | See [Following](#following). |

Each request is one `list_session_events`; the iterator follows the platform's cursor until it is `null`, and never
stops on a short page, because the filter runs on the server and a page can hold fewer matches than `page_size` while
more remain. Arguments the SDK can check without a request are checked before the first one: a bad `page_size`,
`limit`, `after`, `before` or `filter` is `ValueError`. The filter's own grammar is the server's.

```python theme={null}
# the five newest events that touched a worker object
newest = list(session.events({"aliases": ["worker/**"]}, ascending=False, limit=5))
```

`session.latest()` returns the newest event, or `None` for an empty stream, in one request. `session.event(event_id)`
returns one event and raises `NotFound` when there is none.

### Following

`follow=True` keeps the iterator open. When the range is exhausted it sleeps `interval` seconds and reads again, until
the caller closes the iterator, `limit` events were yielded, or `timeout` seconds have passed. The deadline ends the
follow whether or not anything remains to read. Following needs `ascending=True` and no `before`; either combination is
`ValueError`.

```python theme={null}
start = session.latest()
after = start["id"] if start else None
for event in session.events({"types": ["object.written"]}, after=after, follow=True, timeout=300):
    print(event["object"]["alias"])
```

**A late commit is not missed.** An event is visible to a read at most 60 seconds after the time in its id, so
`after: <last id>` alone can skip a row that commits late. Each pass of a follow therefore re-reads from a bound a
minute behind the newest id it knew before the previous pass began, and drops the ids it already yielded. See
[Settle time](https://docs.blobhub.io/blob-types/workflow/workflows/session-events.md#settle-time). Two consequences:

* Events arrive in id order except a late one, which arrives when it is found, after events with larger ids. No event
  is yielded twice.
* A quiet stream re-reads the same last minute every `interval`: one small listing a poll. Raise `interval` for a
  slow reader. A clock skew between the platform's producers larger than the window can still hide a row.

A follow ends with the error when the session goes away: the next read raises `PermissionDenied`.

### Boundary ids

`boundary_id(moment)` returns the id that sorts before every event of `moment`'s microsecond, so `after=boundary_id(t)`
starts at `t`. It takes epoch seconds or an aware `datetime`; a naive `datetime` is `ValueError`, because the SDK never
guesses a zone.

```python theme={null}
from datetime import datetime, timezone
from blobhub.types.workflow import boundary_id

since = boundary_id(datetime(2026, 11, 9, 9, 0, tzinfo=timezone.utc))
events = list(session.events(after=since))
```

### execution.ended

`run.wait()` ends on the run's status, not on this event. To read the event itself:

```python theme={null}
ended = next(session.events({"types": ["execution.ended"], "executions": [run.id]}), None)
```

It carries the run's `status`, and its `route` and `error` when there are any. The platform writes it best-effort, so it
is usually there but not guaranteed; the run's `status` is the authority.

## Listeners

`session.listeners(*, binding=None)` returns the session's listeners as the platform reports them: every declared
listener of the revision, then the connection listeners open on the session. `binding="declared"` or `"connection"`
keeps only that kind, and anything else is `ValueError`.

```python theme={null}
for entry in session.listeners(binding="declared"):
    print(entry["alias"], entry["enabled"], entry["counts"], entry["in_flight"])
```

A declared entry carries `definition_id`, `alias`, `description`, `enabled` (the effective switch), `session_enabled`
(present only when the session overrides the definition), `definition`, `counts`, `last_outcome`, `outcomes` (the
newest 20), `in_flight` and, when a batch is collecting, `pending`. A connection entry carries `listener_id`,
`connection_id`, `filter`, `batch` and `created_at`. See [Listeners](https://docs.blobhub.io/blob-types/workflow/workflows/listeners.md#outcomes)
for what the outcomes mean.

`session.set_listener(listener, enabled)` switches a declared listener on or off for this session and returns its entry.
`listener` is an alias or definition id and `enabled` must be a `bool`; anything else is `TypeError` or `ValueError`
before any request. An unknown listener is `NotFound`.

```python theme={null}
entry = session.set_listener("orchestrate", False)
print(entry["enabled"], entry.get("session_enabled"))
```

The switch works on a session of any status and raises no event. Setting a value equal to the definition's own
clears the override. It is a write, so the SDK repeats it only when the failure proves it never ran
(see [Errors and retries](https://docs.blobhub.io/sdk/errors.md#why-a-write-is-not-retried)); after a failure that may have reached the platform,
read `listeners()` to learn the state, or set it again, which is idempotent.

`session.executions(listener=None)` yields the session's executions as `Execution` objects, newest first, following the
platform's cursor to its end. With `listener` (an alias or definition id) it yields only the runs that listener
started; their `run.source["kind"]` is `listener`.

```python theme={null}
for run in session.executions(listener="orchestrate"):
    print(run.id, run.status, run.source["listener"]["alias"])
```

A listener alias that resolves to no listener of the revision is `PermissionDenied`, unless it is shaped like an id: a
deleted listener's id still lists the executions it started.

Creating, uploading and editing a listener definition is [blobhub-cli](https://docs.blobhub.io/cli/workflow/manifest.md#listener-entries)'s job,
as for workflow definitions. A connection listener needs a WebSocket, which the SDK does not open; use the
[WebSocket listener protocol](https://docs.blobhub.io/web-socket/server/listener.md).

## Errors

| Error | When |
| :- | :- |
| `ValueError` | A bad `page_size`, `limit`, `after`, `before`, `filter`, `interval`, `binding`, `listener` or `event_id`; a naive `datetime` for `boundary_id`; `ascending=False` or `before` with `follow`. |
| `TypeError` | `enabled` is not a `bool`, or `listener` is not a string. |
| `PermissionDenied` | The session id or alias is absent or not reachable with this key, or a listener alias for `executions` resolves to nothing. |
| `NotFound` | `event()` finds no such event, or `set_listener` names no declared listener. |
| `CommandError` `invalid_filter` | The filter breaks the grammar or its bounds. |
| `CommandError` `invalid_cursor` | The platform refused a cursor. |
| `ServerError` `cursor_loop` | The platform returned a cursor the SDK had already sent. |

## See also

* [Session Events](https://docs.blobhub.io/blob-types/workflow/workflows/session-events.md) — the record, its source, the types and the settle
  time.
* [Event filter](https://docs.blobhub.io/blob-types/workflow/workflows/event-filter.md) — the clauses `filter` takes.
* [Listeners](https://docs.blobhub.io/blob-types/workflow/workflows/listeners.md) — the body, outcomes and the switch.
* [Workflow](https://docs.blobhub.io/sdk/workflow.md) — `execute`, `wait` and the execution log.


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