Session stream (planned)

import WebSocket from "ws";

function followSession(session: { id: string; stream_url: string }, currentToken: () => string, stopOutput: () => void) {
  const ws = new WebSocket(session.stream_url, ["tlp.session.v1", `tlp.session.${currentToken()}`]);
  let pings: NodeJS.Timeout | undefined;

  ws.on("open", () => {
    pings = setInterval(() => ws.send(JSON.stringify({ type: "ping", id: String(Date.now()) })), 25_000);
  });

  ws.on("message", (raw) => {
    const message = JSON.parse(raw.toString());
    switch (message.type) {
      case "session.terminated":
        stopOutput(); // Within one second.
        ws.close();
        break;
      case "session.warning":
        console.warn(`${message.code}: act by ${message.act_by}`);
        break;
      case "sample.decided":
        if (message.decision === "deny") stopOutput(); // session.terminated follows
        break;
    }
  });

  ws.on("error", (err) => console.error("session stream:", err.message)); // "close" follows

  ws.on("close", async (code) => {
    clearInterval(pings);
    if (code === 1000) return stopOutput(); // The session ended or was terminated.
    try {
      const res = await fetch(`https://app.reflections.ai/v1/sessions/${session.id}`, {
        headers: {
          Authorization: `Bearer ${process.env.REFLECTIONS_KEY}`,
          "Reflections-Version": "2026-10-15",
        },
      });
      if (!res.ok) throw new Error(`Retrieve session failed: ${res.status}`);
      const current = await res.json();
      if (current.status === "active") {
        setTimeout(() => followSession(session, currentToken, stopOutput), 1_000);
        return;
      }
    } catch (err) {
      console.error(err);
    }
    stopOutput(); // Ended, terminated or unknown.
  });
}

import asyncio, json, os, time
import requests
import websockets

async def follow_session(session, current_token, stop_output):
    while True:
        try:
            async with websockets.connect(
                session["stream_url"], subprotocols=["tlp.session.v1", f"tlp.session.{current_token()}"]
            ) as ws:
                pinger = asyncio.create_task(send_pings(ws))
                try:
                    async for raw in ws:
                        message = json.loads(raw)
                        if message["type"] == "session.terminated":
                            stop_output()  # Within one second.
                            return
                        if message["type"] == "session.warning":
                            print(f"{message['code']}: act by {message['act_by']}")
                        if message["type"] == "sample.decided" and message["decision"] == "deny":
                            stop_output()  # session.terminated follows
                finally:
                    pinger.cancel()
        except websockets.ConnectionClosed as closed:
            if closed.rcvd is not None and closed.rcvd.code == 1000:
                stop_output()  # The session ended or was terminated.
                return
        except (websockets.InvalidHandshake, OSError):
            pass  # Read the session before reconnecting.

        try:
            res = await asyncio.to_thread(
                requests.get,
                f"https://app.reflections.ai/v1/sessions/{session['id']}",
                headers={
                    "Authorization": f"Bearer {os.environ['REFLECTIONS_KEY']}",
                    "Reflections-Version": "2026-10-15",
                },
                timeout=5,
            )
            res.raise_for_status()
            active = res.json()["status"] == "active"
        except (requests.RequestException, ValueError, KeyError):
            active = False
        if not active:
            stop_output()  # Ended, terminated or unknown.
            return
        await asyncio.sleep(1)

async def send_pings(ws):
    while True:
        await asyncio.sleep(25)
        await ws.send(json.dumps({"type": "ping", "id": str(int(time.time()))}))