WebSocket

// Browser
const ws = new WebSocket('ws://localhost:8080/ws?token=<jwt>');

ws.onopen = () => console.log('connected');
ws.onmessage = (e) => {
  const frame = JSON.parse(e.data);
  console.log(frame.type, frame);
};
ws.onclose = (e) => {
  if (e.code === 4001) console.error('Missing token');
  if (e.code === 4003) console.error('Invalid or expired token');
};

// Server sends immediately after successful auth
{
  "type": "connected",
  "timestamp": "2026-02-27T12:00:00.000Z"
}

// Internal connection storage
// One uid can have N concurrent sockets (multi-device)
const connections = new Map<string, Set<AuthenticatedSocket>>();

// Each AuthenticatedSocket carries:
// - socket: WebSocket
// - uid: string
// - connectedAt: Date
// - inFlight: { abortController: AbortController, id: string } | null

// Server → Client every 30 seconds
{ type: "ping" }

// Client must respond with
{ type: "pong" }

// If no pong received before next cycle, socket is terminated

// Emitted on the typed event bus for each connection lifecycle event
bus.emit('ws.connected',    { uid, socketCount });
bus.emit('ws.disconnected', { uid, socketCount, code, reason });

// type: "message" — send a message to an agent
{
  "type": "message",
  "id": "req_abc123",               // string, required — client-generated request ID
  "agentId": "chad",                // string, default "chad"
  "message": "What is the weather?", // string, 1–10 000 chars
  "surface": "chat",                // "chat" | "prop" | "group", default "chat"
  "surfaceId": "main"               // string, default "main"
}

// type: "ping" — keep-alive from client
{
  "type": "ping"
}

import { z } from 'zod';

const messageFrame = z.object({
  type:      z.literal('message'),
  id:        z.string(),
  agentId:   z.string().default('chad'),
  message:   z.string().min(1).max(10000),
  surface:   z.enum(['chat', 'prop', 'group']).default('chat'),
  surfaceId: z.string().default('main'),
});

const pingFrame = z.object({
  type: z.literal('ping'),
});

const inboundFrame = z.discriminatedUnion('type', [messageFrame, pingFrame]);

// Invalid JSON
{ "type": "error", "message": "Invalid JSON" }

// Unknown frame type
{ "type": "error", "message": "Unknown frame type: subscribe" }

// Zod validation failure
{ "type": "error", "message": "message: String must contain at least 1 character(s)" }

// Sent after session lookup / creation
{
  "type": "session_resolved",
  "id": "req_abc123",
  "sessionId": "ses_xyz",
  "isNew": true
}

// Incremental text delta — one per token batch
{
  "type": "stream_chunk",
  "id": "req_abc123",
  "content": "The weather in"
}

// Tool call started
{
  "type": "tool_call",
  "id": "req_abc123",
  "index": 0,
  "toolCallId": "tc_1",
  "name": "web_search"
}

// Tool call complete (includes parsed arguments)
{
  "type": "tool_call",
  "id": "req_abc123",
  "index": 0,
  "toolCallId": "tc_1",
  "name": "web_search",
  "complete": true,
  "arguments": "{\"query\":\"weather today\"}"
}

// After tool execution finishes
{
  "type": "tool_result",
  "id": "req_abc123",
  "name": "web_search",
  "durationMs": 320,
  "error": null
}

// Final frame — full response + metadata
{
  "type": "stream_done",
  "id": "req_abc123",
  "content": "The weather in San Francisco is 62°F and sunny.",
  "sessionId": "ses_xyz",
  "usage": {
    "promptTokens": 1240,
    "completionTokens": 38,
    "totalTokens": 1278
  },
  "toolsUsed": ["web_search"]
}

// Error tied to a specific request
{
  "type": "error",
  "id": "req_abc123",
  "message": "Rate limit exceeded"
}

// In-memory sliding window per uid
const WINDOW_MS   = 60_000;   // 60 seconds
const MAX_MESSAGES = 30;       // per window

// Timestamps older than WINDOW_MS are pruned every 60 seconds
// Exceeding the limit returns:
{
  "type": "error",
  "id": "req_abc123",
  "message": "Rate limit exceeded"
}

// Max 1 in-flight agent request PER SOCKET (not per uid)
// A user with 3 devices can have 3 parallel requests

// Second message while one is in progress:
{
  "type": "error",
  "id": "req_def456",
  "message": "A request is already in progress on this connection"
}

// Each in-flight request carries an AbortController
// Closing the socket calls abortController.abort()
// → the inference provider cancels the request upstream

// Before sending each chunk, the gateway checks:
if (socket.bufferedAmount > 65_536) {   // 64 KB threshold
  // Pause streaming — poll every 50ms
  while (socket.bufferedAmount > 65_536) {
    await sleep(50);
  }
}
// Then resume sending chunks

starting → streaming → tool_calling → completed
                                    ↘ error

// Pushed to ALL of the user's WS connections on every state change
{
  "type": "presence",
  "agentId": "chad",
  "sessionId": "ses_xyz",
  "status": "streaming",
  "currentTool": null,
  "timestamp": "2026-02-27T12:00:01.000Z"
}

// In-memory presence storage
const presence = new Map<string, Map<string, AgentPresenceState>>();
//                       uid           sessionId

interface AgentPresenceState {
  agentId: string;
  sessionId: string;
  status: 'starting' | 'streaming' | 'tool_calling' | 'completed' | 'error';
  currentTool: string | null;
  updatedAt: Date;
}

// Push a frame to all of a user's connected devices
const sentCount = pushToUser(uid, {
  type: 'briefing',
  content: 'Good morning — here is your daily briefing.'
});
// sentCount = number of sockets that received the frame

// Broadcast to every connected user
const totalSent = broadcast({
  type: 'line_alert',
  message: 'Scheduled maintenance in 15 minutes'
});
// totalSent = total frames sent across all users

// Check how many unique users are online
const online = getConnectedUserCount();