# Streaming sessions (/docs/sessions/streaming)

Follow a session live over SSE or WebSocket, resume from any seq, and write idempotently so a retry never duplicates.

Both channels deliver the same **frames**. Every frame carries `v: 1` and a type `t` ∈ `event | presence | lease | control`. On connect the server first **replays** stored events with `seq > after` (up to 1000), then sends one `lease` frame, then switches to live. Duplicate `event` frames are suppressed by `seq`, so a message you already have never arrives twice.

| | SSE | WebSocket |
| --- | --- | --- |
| Path | `GET /v1/sessions/:id/stream?after=<seq>` | `GET /v1/sessions/:id/ws?after=<seq>` (upgrade) |
| Direction | Down only | Down + up (controls, events, presence) |
| Auth | `Authorization` header | `Authorization` header — a key in the query string is rejected |
| Device | `x-codai-device` header, required | Header, or `?device=<uuid>&platform=<p>` |
| Link share | `x-codai-share-token` or `?share=` | Same |
| Frame shape | `event: <t>` + `data: {"v":1,"seq"?,…data}` — data **flattened** | `{ "v":1, "t", "seq"?, "data": {…} }` — data **nested** |
| Keep-alive | `: ping <ms>` comment every 15 s of silence | — |
| Max connection | 30 min, then the server closes | — |

## SSE [#sse]

```bash
curl -N "https://ai.codai.ro/v1/sessions/5b3e…/stream?after=6" \
  -H "Authorization: Bearer $CODAI_API_KEY" \
  -H "x-codai-device: 9a0b…" -H "x-codai-device-platform: web"
```

```text
event: event
data: {"v":1,"seq":7,"kind":"tool_call","ts":1757757852512,"sender_device_id":"1c2f…","turn_id":"t-1","client_event_id":"c9f2…","payload":{"name":"open_app"}}

event: lease
data: {"v":1,"holder_device_id":"1c2f…","expires_at":"2026-09-13T10:04:40.000Z"}

event: presence
data: {"v":1,"device_id":"9a0b…","user_id":"…","role":"editor","executor":false,"last_seen":1757757852600,"driving":false,"online":true}

event: control
data: {"v":1,"id":"ctl-8d1a…","kind":"steer","text":"Use the second result.","turn_id":"t-1","ask_id":null,"from_device_id":"9a0b…","seq":8,"applied":false}

event: control
data: {"v":1,"id":"ctl-8d1a…","kind":"steer","seq":8,"applied":true}

: ping 1757757870000
```

A `lease` frame with `"holder_device_id": null, "expires_at": null` means the lease is free. A `control` frame arrives twice per control: once when accepted (`applied: false`, full body) and once when the executor marks it applied (`applied: true`, short body).

<Callout type="warn">
The browser's `EventSource` cannot set `Authorization` or `x-codai-device`, so use a fetch-based SSE reader (`fetch` + `ReadableStream`) or the WebSocket. Never pass the API key in the URL — the protocol forbids it and the WebSocket enforces it.
</Callout>

A minimal fetch-based reader that tracks `seq` and reconnects:

```ts
async function follow(sessionId: string, apiKey: string, device: string, onFrame: (t: string, d: any) => void) {
  let after = 0;
  for (;;) {
    const res = await fetch(`https://ai.codai.ro/v1/sessions/${sessionId}/stream?after=${after}`, {
      headers: { Authorization: `Bearer ${apiKey}`, 'x-codai-device': device, 'x-codai-device-platform': 'web' },
    });
    const reader = res.body!.pipeThrough(new TextDecoderStream()).getReader();
    let buf = '';
    for (;;) {
      const { value, done } = await reader.read();
      if (done) break; // 30-min cut or network drop → reconnect with the last seq
      buf += value;
      let i: number;
      while ((i = buf.indexOf('\n\n')) >= 0) {
        const block = buf.slice(0, i);
        buf = buf.slice(i + 2);
        const t = /^event: (.+)$/m.exec(block)?.[1];
        const raw = /^data: (.+)$/m.exec(block)?.[1];
        if (!t || !raw) continue; // ": ping" comments
        const d = JSON.parse(raw);
        if (t === 'event') {
          if (d.seq <= after) continue;
          after = d.seq;
        }
        onFrame(t, d);
      }
    }
  }
}
```

## WebSocket [#websocket]

```ts
const ws = new WebSocket(`wss://ai.codai.ro/v1/sessions/${id}/ws?after=${after}&device=${device}&platform=web`, {
  headers: { Authorization: `Bearer ${apiKey}` }, // Node `ws`; browsers cannot set headers on WebSocket
});
```

Close codes: `4401` missing/invalid key (or a key in the query string) · `4403` not a member · `4404` session not found · `4400` other client error (e.g. missing device) · `1011` internal.

**Down frames** are the SSE objects with `data` nested:

```json
{ "v": 1, "t": "event", "seq": 7, "data": { "kind": "tool_call", "ts": 1757757852512, "sender_device_id": "1c2f…", "turn_id": "t-1", "client_event_id": "c9f2…", "payload": { "name": "open_app" } } }
{ "v": 1, "t": "lease", "data": { "holder_device_id": "1c2f…", "expires_at": "2026-09-13T10:04:40.000Z" } }
{ "v": 1, "t": "presence", "data": { "device_id": "9a0b…", "role": "viewer", "executor": false, "driving": false, "last_seen": 1757757852600, "online": true } }
{ "v": 1, "t": "control", "data": { "id": "ctl-8d1a…", "kind": "send", "text": "…", "from_device_id": "9a0b…", "seq": 9, "applied": false } }
```

**Up messages** (text frames, JSON):

| `t` | Body | Min role | Ack |
| --- | --- | --- | --- |
| `control` | `{ "t": "control", "id", "kind", "text"?, "turn_id"?, "ask_id"? }` | editor | `{ "v": 1, "t": "ack", "ref": <id>, "accepted": true, "seq", "duplicate" }` |
| `events` | `{ "t": "events", "events": IncomingEvent[1..200], "expected_last_seq"? }` | lease holder | `{ "v": 1, "t": "ack", "last_seq", "events": [{ "client_event_id", "seq" }] }` |
| `presence` | `{ "t": "presence", "driving": bool }` | viewer | A `presence` frame to everyone |

An error on an up message is answered with `{ "v": 1, "t": "error", "ref"?: <control id>, "error": { "message", "type", "code", "details" } }` and the socket **stays open**. Binary frames are ignored.

## Resuming [#resuming]

There is exactly one resume mechanism: **`?after=<seq>`**. Frames carry no `id:` line and `Last-Event-ID` is not read. Keep the highest `seq` you have processed; on any disconnect — network, the 30-minute SSE cut, a redeploy — reconnect with it and the server replays what you missed.

Two things to get right:

1. **Replay is capped at 1000 events.** If `last_seq − after` may exceed that (a long session you have not watched for a while), page `GET /v1/sessions/:id/events?after=&limit=1000` first until you reach `last_seq`, then open the stream from there. Otherwise the events between the end of the replay and the first live frame are never delivered.
2. **Only `event` frames have a `seq`.** `lease`, `presence` and `control` frames are state, not log entries — take the latest, do not try to order them against events. (The `control` frame does carry the `seq` of its echoed `control` *event*, which is how you correlate the two.)

## Writing idempotently [#writing-idempotently]

Retries are safe when every write carries its own key:

| Write | Your key | On retry |
| --- | --- | --- |
| `POST …/events` / WS `events` | `client_event_id` per event | The duplicate is not stored; you get its **original** `seq` in `events[]` and it is excluded from `accepted`. Send one on **every** event — an event without it is never deduplicated. The key is scoped to your device: the same `client_event_id` from another device is a different event. |
| `POST …/control` / WS `control` | control `id` | `200 { accepted: true, seq: <original>, duplicate: true }` — no new event, nothing re-echoed. |
| `POST …/dispatch` | `control_id` | `200 duplicate: true`, **no push re-sent**. Always supply your own `control_id` when you may retry. |
| `POST …/lease` | your device | Re-claiming refreshes `expires_at`. |

To guard against writing on a stale view, add `expected_last_seq` to an events batch: if the session moved on, the whole batch is refused with `409 seq_conflict` and nothing is written — reload, then retry.

## The executor loop [#the-executor-loop]

The reference executor (the phone) behaves as follows; other executors should match.

1. On start or dispatch wake, for each open session: `POST …/lease`; on success start the agent run.
2. Write every trace line locally **and** batch it to `POST …/events` (or WS `events`) — flush every ≤ 200 ms or 20 events, always with `client_event_id`.
3. Drain `GET …/controls?applied=false` (optionally `&target=me`), then subscribe for live `control` frames.
4. Apply each control; when done, `POST …/control/:cid/applied`. Skip controls whose `target_device_id` is set and is not your own id.
5. Heartbeat `PUT …/lease` every 10 s. On `409`: stop executing, become a viewer, show "driven by `<device>`".
6. If a turn needs the screen, emit `{ "kind": "screen_wait", "payload": { "position" } }` so viewers understand why the session is idle.