Streaming de sesiuni

Ambele canale livrează aceleași frame-uri. Fiecare frame poartă v: 1 și un tip t ∈ event | presence | lease | control. La conectare serverul redă întâi evenimentele stocate cu seq > after (până la 1000), apoi trimite un frame lease, apoi trece pe live. Frame-urile event duplicate sunt suprimate după seq, așa că un mesaj pe care îl ai deja nu sosește niciodată de două ori.

SSE

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"

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

Un frame lease cu "holder_device_id": null, "expires_at": null înseamnă că lease-ul este liber. Un frame control sosește de două ori per control: o dată la acceptare (applied: false, body complet) și o dată când executorul îl marchează ca aplicat (applied: true, body scurt).

Un cititor minimal bazat pe fetch, care urmărește seq și se reconectează:

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; // tăierea de 30 min sau cădere de rețea → reconectare cu ultimul 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; // comentariile ": ping"
        const d = JSON.parse(raw);
        if (t === 'event') {
          if (d.seq <= after) continue;
          after = d.seq;
        }
        onFrame(t, d);
      }
    }
  }
}

WebSocket

const ws = new WebSocket(`wss://ai.codai.ro/v1/sessions/${id}/ws?after=${after}&device=${device}&platform=web`, {
  headers: { Authorization: `Bearer ${apiKey}` }, // Node `ws`; browserele nu pot seta headere pe WebSocket
});

Coduri de închidere: 4401 cheie lipsă/invalidă (sau o cheie în query string) · 4403 nu ești membru · 4404 sesiune negăsită · 4400 altă eroare de client (de ex. dispozitiv lipsă) · 1011 eroare internă.

Frame-urile down sunt obiectele SSE cu data imbricat:

{ "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 } }

Mesajele up (frame-uri text, JSON):

O eroare la un mesaj up primește răspunsul { "v": 1, "t": "error", "ref"?: <control id>, "error": { "message", "type", "code", "details" } } și socket-ul rămâne deschis. Frame-urile binare sunt ignorate.

Reluarea

Există exact un mecanism de reluare: ?after=<seq>. Frame-urile nu poartă o linie id: și Last-Event-ID nu este citit. Păstrează cel mai mare seq pe care l-ai procesat; la orice deconectare — rețea, tăierea SSE de 30 de minute, un redeploy — reconectează-te cu el și serverul redă ce ai pierdut.

Două lucruri de făcut corect:

1. Replay-ul este limitat la 1000 de evenimente. Dacă last_seq − after poate depăși asta (o sesiune lungă pe care nu ai urmărit-o de ceva vreme), paginează întâi GET /v1/sessions/:id/events?after=&limit=1000 până ajungi la last_seq, apoi deschide stream-ul de acolo. Altfel evenimentele dintre sfârșitul replay-ului și primul frame live nu sunt livrate niciodată.
2. Doar frame-urile event au seq. Frame-urile lease, presence și control sunt stare, nu intrări de jurnal — ia-l pe cel mai recent, nu încerca să le ordonezi față de evenimente. (Frame-ul control poartă totuși seq-ul evenimentului control pe care îl face ecou — așa le corelezi pe cele două.)

Scrierea idempotentă

Retry-urile sunt sigure când fiecare scriere poartă propria cheie:

Pentru a te feri să scrii pe o vedere învechită, adaugă expected_last_seq la un batch de evenimente: dacă sesiunea a avansat, întregul batch este refuzat cu 409 seq_conflict și nu se scrie nimic — reîncarcă, apoi reîncearcă.

Bucla executorului

Executorul de referință (telefonul) se comportă astfel; ceilalți executori ar trebui să facă la fel.

1. La pornire sau la o trezire prin dispatch, pentru fiecare sesiune deschisă: POST …/lease; la succes pornește rularea agentului.
2. Scrie fiecare linie de trace local și trimite-o în batch la POST …/events (sau WS events) — flush la fiecare ≤ 200 ms sau 20 de evenimente, mereu cu client_event_id.
3. Consumă GET …/controls?applied=false (opțional &target=me), apoi abonează-te la frame-urile control live.
4. Aplică fiecare control; când ai terminat, POST …/control/:cid/applied. Sari controls al căror target_device_id este setat și nu este propriul tău id.
5. Heartbeat PUT …/lease la fiecare 10 s. La 409: oprește execuția, devino viewer, afișează „condus de <device>”.
6. Dacă un turn are nevoie de ecran, emite { "kind": "screen_wait", "payload": { "position" } } ca viewerii să înțeleagă de ce sesiunea stă.