HawkTalkLive.

Read this before you write a line of code. HawkTalkLive is the tier-03 idea: one presence, many independent lanes — audio, text, tools, thinking, async — advancing at the same time instead of one stream carrying everything serially. The lane mux that implements it, /live/brain, is PREVIEW: loopback only, no auth wired, and not served on api.hawktalk.ai. You cannot call it from production today.

What is shipped is /v1/realtime, and it is a strict subset: voice and tools on one cognition lane. Everything in this page marked describes the lane-mux wire as designed and as it behaves against a local node — it is not a public API. Everything marked [SHIPPED] you can run against wss://api.hawktalk.ai/v1/realtime right now with your sk- key. We tell you which is which; do not ship a roadmap.

There are two /live/ paths and they are not the same thing. /live/ouroboros is an alias of the shipped /v1/realtime socket — same handler, same frames, same sk- key, and either path works. /live/brain is the preview lane mux: a different endpoint, loopback only, not served on api.hawktalk.ai. Everything marked [SHIPPED] below is the first one; everything marked [DESIGN] is the second.

Capability	Where	Status	What that means for you
Voice in + out on one socket	WS /v1/realtime	SHIPPED	base64 pcm16 up, base64 pcm16 down. JSON text frames only — binary frames are rejected by design.
Tool calls mid-conversation	WS /v1/realtime	SHIPPED	Schemas in session.update (once per session, not per turn), results back via conversation.item.create.
Model selection per connection / session / turn	WS /v1/realtime	SHIPPED	Three precedence levels, all accepting auto. Use auto.
Turn telemetry	response.done → x_ouroboros	SHIPPED	Real measurements or null. Never a fabricated 0.
Per-key RPM limiting + usage ledger	gateway	SHIPPED	429 carries Retry-After; rate_limits.updated arrives on the socket.
REST STT / TTS	/v1/audio/transcriptions, /v1/audio/speech	SEAM	Nodes without the seam wired return 501 stt_not_wired / tts_not_wired. Fall back to browser Web Speech — never synthesise audio or invent a transcript.
Lane mux — many lanes, one socket	WS /live/brain	PREVIEW	Loopback only, no auth wired, not publicly served. [DESIGN] in this page.
Affect on the wire	ouroboros.user_affect	PREVIEW	Only on the mux, and it self-marks x_stub when the SER sidecar is not up.
Server-side thinking lane	—	DESIGN	No server frame type exists. You compose it client-side today — see below, it works and it is cheap.
Server-side async lane	—	DESIGN	Same: client-composed today, and the reconciliation policy is yours to own either way.
Monthly token quota in the request path	gateway	ROADMAP	RPM limiting is enforced now; monthly quota is not yet enforced at request time.

Do not hardcode model ids. Ids published in older pages (hawk-oak, hawk-sapling, hawk-feedseed, hawk-sakura) are not what a node serves — copy-pasting them yields 404 model_not_found. Call GET /v1/models, or pass auto and let the router pick per utterance. Pinning a large tier is the single biggest source of overspend on this product.

One socket, OpenAI-Realtime event names, your sk- key. This is the migration target: if you are moving off a mainstream realtime API, the frames below are the ones you already send. Three auth forms are accepted — Authorization: Bearer sk-…, ?api_key=…, or the browser subprotocol openai-insecure-api-key.<key> (browsers cannot set headers on a WebSocket; that is what the subprotocol is for).

What this gives you, and what it does not. You get the audio, text and tools lanes — but they are one cognition lane, scheduled serially by the model. A tool round trip stalls the reply because the same generator is waiting on it. You do not get thinking, async, ouroboros.presence or ouroboros.user_affect here; those are mux lanes and the mux is preview. The last sections show how to build thinking and async client-side on top of exactly this socket, which is what the shipped HawkTalk client itself does.

End of speech is yours. The server does semantic endpointing only and never emits speech_started / speech_stopped. Your VAD decides the utterance is over and sends input_audio_buffer.commit. If you are porting code that waited for a server VAD event, it will hang forever — this is the single most common migration break.

Two frames can carry one finished tool call, so dedupe by call_id. response.output_item.done with an item of type function_call is the frame in the published event list. Some node builds also emit response.function_call_arguments.done for the same call. Every example below accepts both — and keeps a set of handled call_ids, because running your actuator twice for one request is a real bug, not a theoretical one. For the same reason each example sends exactly one response.create after response.done, however many tool calls the turn produced: two response.create frames in flight on one socket is the week you do not get back.

Two more fields live in two places, so read both. Turn telemetry arrives at the root of response.done on some nodes and nested under response on others — read x_ouroboros off the root and off response, or you will log "this node has no telemetry" about a node that is measuring perfectly well. Likewise the base64 on response.audio.delta: some builds put it under delta, some under audio. Read delta ?? audio and you are correct on both. Each is one expression; every example below carries it.

curl — preflight: what does this node actually serve?

# Never hardcode a model id. Ask the registry, every deploy.
curl -sS https://api.hawktalk.ai/v1/models \
  -H "Authorization: Bearer $HAWKTALK_API_KEY"

# /health needs no auth and tells you which seams are wired on THIS node —
# check it before you promise a customer voice. An unwired seam is a 501,
# not silence, so you can branch on it honestly.
curl -sS https://api.hawktalk.ai/health

# The lane mux is not served publicly. This is the check, and it is expected
# to fail — /live/brain is PREVIEW and loopback-only.
curl -sS -i https://api.hawktalk.ai/live/brain \
  -H "Connection: Upgrade" -H "Upgrade: websocket"

# Make the test audio the examples below read. Input default is 24000 Hz,
# mono, signed 16-bit little-endian. Resample on YOUR side — the gateway
# will not guess your rate, and there is no session field to declare one.
ffmpeg -i utterance.wav -ac 1 -ar 24000 -f s16le utterance_24k.pcm

Python

# pip install "websockets>=14"
# >=14 is deliberate: additional_headers= and websockets.InvalidStatus are the
# new asyncio client. On websockets 13.x use extra_headers= and
# InvalidStatusCode(.status_code) instead — same semantics, different names.
# JSON TEXT frames only. Binary frames are rejected by design — if you ship
# raw PCM as a binary frame (as some other providers allow) the socket closes.
import asyncio, base64, json, os
import websockets

KEY = os.environ["HAWKTALK_API_KEY"]
# model=auto runs the router per utterance: short turns land on a small tier,
# hard ones escalate. Pinning "tier:think" here can cost 10x for chit-chat.
URL = "wss://api.hawktalk.ai/v1/realtime?model=auto"

TOOLS = [{
    "type": "function",
    "name": "set_thermostat",
    "description": "Set the target temperature for a room.",
    "parameters": {
        "type": "object",
        "properties": {"room": {"type": "string"},
                       "celsius": {"type": "number"}},
        "required": ["room", "celsius"],
    },
}]

def run_tool(name, args):
    # A tool call off the wire is a REQUEST, not an authorization. Gate it.
    if name == "set_thermostat":
        return {"ok": True, "room": args.get("room"), "celsius": args.get("celsius")}
    return {"ok": False, "error": "unknown tool"}

def tool_call_from(ev):
    # response.output_item.done is the documented carrier. Some builds ALSO
    # emit response.function_call_arguments.done for the same call — accept it,
    # then dedupe on call_id at the call site. Unknown fields are never fatal.
    if ev.get("type") == "response.function_call_arguments.done":
        return ev.get("name"), ev.get("call_id"), ev.get("arguments") or "{}"
    if ev.get("type") == "response.output_item.done":
        it = ev.get("item") or {}
        if it.get("type") == "function_call":
            return it.get("name"), it.get("call_id"), it.get("arguments") or "{}"
    return None, None, None

async def main():
    async with websockets.connect(
        URL,
        additional_headers={"Authorization": f"Bearer {KEY}"},   # websockets<14: extra_headers=
        max_size=16 * 1024 * 1024,        # audio deltas are base64 — raise the frame cap
    ) as ws:
        hello = json.loads(await ws.recv())
        assert hello["type"] == "session.created", hello

        # Tools and instructions go in ONCE, here. Do not re-send them per turn.
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "model": "auto",                        # or a registry id, or "tier:quick"
                "modalities": ["text", "audio"],
                "instructions": "You run a house. Answer in one sentence.",
                "tools": TOOLS,
                "tool_choice": "auto",
            },
        }))

        # ---- CLIENT-side VAD. The server never tells you speech stopped. ----
        pcm = open("utterance_24k.pcm", "rb").read()
        for i in range(0, len(pcm), 4800):        # 100 ms @ 24 kHz mono s16
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(pcm[i:i + 4800]).decode(),
            }))
        await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
        await ws.send(json.dumps({"type": "response.create"}))

        audio, rate = bytearray(), None
        handled = set()              # call_ids already run — the dedupe
        tool_turn_pending = False
        async for raw in ws:
            ev = json.loads(raw)
            t = ev.get("type")

            if t == "conversation.item.input_audio_transcription.completed":
                print("heard:", ev.get("transcript"))
            elif t == "response.text.delta":
                print(ev.get("delta", ""), end="", flush=True)
            elif t == "response.audio.delta":
                # some builds carry the base64 under "audio", not "delta"
                audio += base64.b64decode(ev.get("delta") or ev.get("audio") or "")
                rate = ev.get("x_sample_rate_hz") or rate   # output rate is declared, not assumed
            elif t == "rate_limits.updated":
                print("\nquota:", ev.get("rate_limits"))
            elif t == "error":
                e = ev["error"]
                print("\nerror:", e.get("code"), e.get("message"))
                break
            elif t == "response.done":
                # telemetry sits at the frame root on some nodes and under
                # "response" on others. Read both, always.
                x = (ev.get("x_ouroboros")
                     or (ev.get("response") or {}).get("x_ouroboros") or {})
                # null means NOT MEASURED (e.g. the TTS seam is not wired on this
                # node). Never read null as 0 — you will invent a latency win.
                print(f"\nttft_ms={x.get('ttft_ms')} "
                      f"first_audio_out_ms={x.get('first_audio_out_ms')} "
                      f"end_to_end_ms={x.get('end_to_end_ms')}")
                if tool_turn_pending:
                    # ONE response.create, after the turn closed, no matter how
                    # many tool outputs went in. Single writer, always.
                    tool_turn_pending = False
                    await ws.send(json.dumps({"type": "response.create"}))
                    continue
                break

            name, call_id, args = tool_call_from(ev)
            if name and call_id and call_id not in handled:
                handled.add(call_id)
                out = run_tool(name, json.loads(args))
                await ws.send(json.dumps({
                    "type": "conversation.item.create",
                    "item": {"type": "function_call_output",
                             "call_id": call_id, "output": json.dumps(out)},
                }))
                tool_turn_pending = True      # create the response at response.done

        if not audio:
            print("no audio on this node — fall back to browser Web Speech; "
                  "never synthesise a substitute voice")
        else:
            print(f"pcm16 bytes={len(audio)} at {rate} Hz")

asyncio.run(main())

Dart

// pubspec.yaml:  web_socket_channel: ^3.0.0   (Dart 3 — records are used below)
// JSON TEXT frames only. Binary frames are rejected by design.
import 'dart:convert';
import 'dart:io';
import 'dart:typed_data';
import 'package:web_socket_channel/io.dart';

final _key = Platform.environment['HAWKTALK_API_KEY']!;
final _uri = Uri.parse('wss://api.hawktalk.ai/v1/realtime?model=auto');

const tools = [{
  'type': 'function',
  'name': 'set_thermostat',
  'description': 'Set the target temperature for a room.',
  'parameters': {
    'type': 'object',
    'properties': {
      'room': {'type': 'string'},
      'celsius': {'type': 'number'},
    },
    'required': ['room', 'celsius'],
  },
}];

Map<String, dynamic> runTool(String name, Map<String, dynamic> args) {
  // A tool call off the wire is a REQUEST, not an authorization. Gate it.
  if (name == 'set_thermostat') {
    return {'ok': true, 'room': args['room'], 'celsius': args['celsius']};
  }
  return {'ok': false, 'error': 'unknown tool'};
}

/// Accepts either frame that can carry a finished call.
(String, String, String)? toolCallFrom(Map<String, dynamic> ev) {
  if (ev['type'] == 'response.function_call_arguments.done') {
    final name = ev['name'] as String?;
    final callId = ev['call_id'] as String?;
    if (name == null || callId == null) return null;
    return (name, callId, (ev['arguments'] as String?) ?? '{}');
  }
  final item = ev['item'];
  if (ev['type'] == 'response.output_item.done' &&
      item is Map &&
      item['type'] == 'function_call') {
    final name = item['name'] as String?;
    final callId = item['call_id'] as String?;
    if (name == null || callId == null) return null;
    return (name, callId, (item['arguments'] as String?) ?? '{}');
  }
  return null;
}

Future<void> main() async {
  final channel = IOWebSocketChannel.connect(
    _uri,
    headers: {'Authorization': 'Bearer $_key'},
  );

  void send(Object o) => channel.sink.add(jsonEncode(o));

  final audio = BytesBuilder();
  final handled = <String>{}; // call_ids already run — the dedupe
  int? rate;
  var toolTurnPending = false;
  var helloReceived = false;

  await for (final raw in channel.stream) {
    final ev = jsonDecode(raw as String) as Map<String, dynamic>;
    final t = ev['type'] as String?;

    if (!helloReceived) {
      if (t != 'session.created') {
        throw StateError('expected session.created, got $t');
      }
      helloReceived = true;

      // Tools and instructions go in ONCE, here. Do not re-send them per turn.
      send({
        'type': 'session.update',
        'session': {
          'model': 'auto',
          'modalities': ['text', 'audio'],
          'instructions': 'You run a house. Answer in one sentence.',
          'tools': tools,
          'tool_choice': 'auto',
        },
      });

      // ---- CLIENT-side VAD. The server never tells you speech stopped. ----
      final file = File('utterance_24k.pcm');
      final pcm = file.existsSync() ? await file.readAsBytes() : Uint8List(0);
      for (var i = 0; i < pcm.length; i += 4800) {
        final end = (i + 4800 < pcm.length) ? i + 4800 : pcm.length;
        send({
          'type': 'input_audio_buffer.append',
          'audio': base64Encode(pcm.sublist(i, end)),
        });
      }
      send({'type': 'input_audio_buffer.commit'});
      send({'type': 'response.create'});
      continue;
    }

    switch (t) {
      case 'conversation.item.input_audio_transcription.completed':
        print('heard: ${ev["transcript"]}');

      case 'response.text.delta':
        stdout.write(ev['delta'] ?? '');

      case 'response.audio.delta':
        // some builds carry the base64 under 'audio', not 'delta'
        final b64 = (ev['delta'] ?? ev['audio']) as String?;
        if (b64 != null && b64.isNotEmpty) {
          audio.add(base64Decode(b64));
        }
        rate = (ev['x_sample_rate_hz'] as int?) ?? rate;

      case 'rate_limits.updated':
        print('\nquota: ${ev["rate_limits"]}');

      case 'error':
        final e = ev['error'] as Map<String, dynamic>?;
        stderr.writeln('\nerror: ${e?["code"]} ${e?["message"]}');
        await channel.sink.close();
        break;

      case 'response.done':
        // telemetry sits at the frame root on some nodes and under "response" on others. Read both.
        final response = ev['response'] as Map<String, dynamic>?;
        final x = (ev['x_ouroboros'] ?? response?['x_ouroboros']) as Map<String, dynamic>? ?? const {};
        // null means NOT MEASURED. Never read null as 0.
        print('\nttft_ms=${x["ttft_ms"]} '
            'first_audio_out_ms=${x["first_audio_out_ms"]} '
            'end_to_end_ms=${x["end_to_end_ms"]}');

        if (toolTurnPending) {
          // ONE response.create, after the turn closed, no matter how many tool outputs went in. Single writer, always.
          toolTurnPending = false;
          send({'type': 'response.create'});
          continue;
        }
        await channel.sink.close();
        break;
    }

    final call = toolCallFrom(ev);
    if (call != null) {
      final (name, callId, args) = call;
      if (handled.add(callId)) {
        final parsedArgs = jsonDecode(args) as Map<String, dynamic>;
        final out = runTool(name, parsedArgs);
        send({
          'type': 'conversation.item.create',
          'item': {
            'type': 'function_call_output',
            'call_id': callId,
            'output': jsonEncode(out),
          },
        });
        toolTurnPending = true; // create the response at response.done
      }
    }
  }

  if (audio.isEmpty) {
    print('no audio on this node — fall back to browser Web Speech; never synthesise a substitute voice');
  } else {
    print('pcm16 bytes=${audio.length} at $rate Hz');
  }
}

Identical wire traffic, four runtimes. Each one commits with its own VAD, accepts a tool call in either of the two frames that can carry it and dedupes on call_id, sends a single response.create once the turn closes, and reads x_sample_rate_hz off the audio deltas rather than assuming the output rate matches the input rate.

TypeScript / Node

// npm i ws && npm i -D @types/ws typescript   (Node 20+, "type": "module")
// Never ws.send(Buffer) here — binary frames are rejected by design.
import { readFileSync } from "node:fs";
import WebSocket from "ws";

const KEY = process.env.HAWKTALK_API_KEY!;
const ws = new WebSocket("wss://api.hawktalk.ai/v1/realtime?model=auto", {
  headers: { Authorization: `Bearer ${KEY}` },
  maxPayload: 16 * 1024 * 1024,   // base64 audio deltas are large
});

// 401 invalid_api_key / 429 rate_limit_exceeded / 503 land on the HTTP upgrade,
// never as a frame — and an unhandled "error" event kills the Node process.
ws.on("unexpected-response", (_req, res) => {
  console.error("upgrade failed: HTTP", res.statusCode,
                "retry-after=", res.headers["retry-after"] ?? "n/a");
});
ws.on("error", (e) => console.error("socket error:", e.message));

const TOOLS = [{
  type: "function",
  name: "set_thermostat",
  description: "Set the target temperature for a room.",
  parameters: {
    type: "object",
    properties: { room: { type: "string" }, celsius: { type: "number" } },
    required: ["room", "celsius"],
  },
}];

function runTool(name: string, args: any) {
  // A returned call is a request, not an authorization. Gate before you act.
  if (name === "set_thermostat") return { ok: true, ...args };
  return { ok: false, error: "unknown tool" };
}

// Both frames can carry a completed call; accept whichever your node emits.
function toolCallFrom(ev: any): { name: string; callId: string; args: string } | null {
  if (ev.type === "response.function_call_arguments.done" && ev.name && ev.call_id)
    return { name: ev.name, callId: ev.call_id, args: ev.arguments ?? "{}" };
  if (ev.type === "response.output_item.done" && ev.item?.type === "function_call")
    return { name: ev.item.name, callId: ev.item.call_id, args: ev.item.arguments ?? "{}" };
  return null;
}

const send = (o: unknown) => ws.send(JSON.stringify(o));
const audio: Buffer[] = [];
const handled = new Set<string>();      // call_ids already run — the dedupe
let rate: number | null = null;
let toolTurnPending = false;

ws.on("open", () => {
  // Tools and instructions once per session, not once per turn.
  send({ type: "session.update", session: {
    model: "auto",                     // auto = router per utterance = the cheap default
    modalities: ["text", "audio"],
    instructions: "You run a house. Answer in one sentence.",
    tools: TOOLS, tool_choice: "auto",
  }});

  // 24 kHz mono s16le. 100 ms per append; commit is YOUR VAD's decision —
  // the server never emits speech_started / speech_stopped.
  const pcm = readFileSync("utterance_24k.pcm");
  for (let i = 0; i < pcm.length; i += 4800)
    send({ type: "input_audio_buffer.append",
           audio: pcm.subarray(i, i + 4800).toString("base64") });
  send({ type: "input_audio_buffer.commit" });
  send({ type: "response.create" });
});

ws.on("message", (raw) => {
  const ev = JSON.parse(raw.toString());
  switch (ev.type) {
    case "conversation.item.input_audio_transcription.completed":
      console.log("heard:", ev.transcript); break;
    case "response.text.delta":
      process.stdout.write(ev.delta ?? ""); break;
    case "response.audio.delta":
      // some builds carry the base64 under "audio", not "delta"
      audio.push(Buffer.from(ev.delta ?? ev.audio ?? "", "base64"));
      rate = ev.x_sample_rate_hz ?? rate; break;
    case "rate_limits.updated":
      console.log("quota:", ev.rate_limits); break;
    case "error":
      console.error("error:", ev.error.code, ev.error.message); ws.close(); return;
    case "response.done": {
      // root OR nested under response, depending on the node — read both.
      const x = ev.x_ouroboros ?? ev.response?.x_ouroboros ?? {};
      // null = not measured. Do not coerce to 0 in your metrics pipeline.
      console.log("\nttft_ms=", x.ttft_ms, "first_audio_out_ms=", x.first_audio_out_ms,
                  "end_to_end_ms=", x.end_to_end_ms);
      if (toolTurnPending) {
        // One create for the whole turn's tool outputs. Single writer.
        toolTurnPending = false;
        send({ type: "response.create" });
        return;
      }
      const bytes = Buffer.concat(audio).length;
      console.log(bytes ? `pcm16 bytes=${bytes} at ${rate} Hz`
                        : "no audio on this node — fall back to browser Web Speech");
      ws.close(); return;
    }
  }
  const call = toolCallFrom(ev);
  if (call && !handled.has(call.callId)) {
    handled.add(call.callId);
    send({ type: "conversation.item.create", item: {
      type: "function_call_output", call_id: call.callId,
      output: JSON.stringify(runTool(call.name, JSON.parse(call.args))),
    }});
    toolTurnPending = true;      // response.create fires at response.done
  }
});

Dart / Flutter

// pubspec.yaml:  web_socket_channel: ^3.0.0   (Dart 3 — records are used below)
// Flutter is a first-class target here. Note the two connect paths: native
// gets a header, web cannot set headers on a WebSocket and must use the
// subprotocol form instead.
import 'dart:convert';
import 'dart:io';
import 'package:web_socket_channel/io.dart';
import 'package:web_socket_channel/web_socket_channel.dart';   // web path

const tools = [{
  'type': 'function',
  'name': 'set_thermostat',
  'description': 'Set the target temperature for a room.',
  'parameters': {
    'type': 'object',
    'properties': {'room': {'type': 'string'}, 'celsius': {'type': 'number'}},
    'required': ['room', 'celsius'],
  },
}];

Map<String, dynamic> runTool(String name, Map<String, dynamic> args) =>
    name == 'set_thermostat'
        ? {'ok': true, ...args}
        : {'ok': false, 'error': 'unknown tool'};

/// Accepts either frame that can carry a finished call. Null-safe casts: a
/// frame missing a field must not throw — unknown shapes are never fatal.
(String, String, String)? toolCallFrom(Map<String, dynamic> ev) {
  if (ev['type'] == 'response.function_call_arguments.done') {
    final name = ev['name'] as String?;
    final callId = ev['call_id'] as String?;
    if (name == null || callId == null) return null;
    return (name, callId, (ev['arguments'] as String?) ?? '{}');
  }
  final item = ev['item'];
  if (ev['type'] == 'response.output_item.done' &&
      item is Map && item['type'] == 'function_call') {
    final name = item['name'] as String?;
    final callId = item['call_id'] as String?;
    if (name == null || callId == null) return null;
    return (name, callId, (item['arguments'] as String?) ?? '{}');
  }
  return null;
}

Future<void> main() async {
  final key = Platform.environment['HAWKTALK_API_KEY']!;
  final uri = Uri.parse('wss://api.hawktalk.ai/v1/realtime?model=auto');

  // Native / desktop / Android: real header.
  final channel = IOWebSocketChannel.connect(uri,
      headers: {'Authorization': 'Bearer $key'});
  // Flutter web instead:
  //   WebSocketChannel.connect(uri, protocols: ['openai-insecure-api-key.$key']);
  // or append ?api_key=$key — both are accepted by the gateway.

  void send(Object o) => channel.sink.add(jsonEncode(o));
  final audio = BytesBuilder();
  final handled = <String>{};        // call_ids already run — the dedupe
  int? rate;
  var toolTurnPending = false;

  // Once per session. Do not re-send tools on every turn.
  send({'type': 'session.update', 'session': {
    'model': 'auto',
    'modalities': ['text', 'audio'],
    'instructions': 'You run a house. Answer in one sentence.',
    'tools': tools, 'tool_choice': 'auto',
  }});

  // 24 kHz mono s16le, 100 ms per append. commit() is your VAD firing.
  final pcm = await File('utterance_24k.pcm').readAsBytes();
  for (var i = 0; i < pcm.length; i += 4800) {
    final end = (i + 4800 < pcm.length) ? i + 4800 : pcm.length;
    send({'type': 'input_audio_buffer.append',
          'audio': base64Encode(pcm.sublist(i, end))});
  }
  send({'type': 'input_audio_buffer.commit'});
  send({'type': 'response.create'});

  await for (final raw in channel.stream) {
    final ev = jsonDecode(raw as String) as Map<String, dynamic>;
    switch (ev['type']) {
      case 'conversation.item.input_audio_transcription.completed':
        print('heard: ${ev["transcript"]}');
      case 'response.text.delta':
        stdout.write(ev['delta'] ?? '');
      case 'response.audio.delta':
        // some builds carry the base64 under 'audio', not 'delta'
        audio.add(base64Decode((ev['delta'] ?? ev['audio']) as String? ?? ''));
        rate = (ev['x_sample_rate_hz'] as int?) ?? rate;
      case 'error':
        final e = ev['error'] as Map;
        stderr.writeln('error: ${e["code"]} ${e["message"]}');
        await channel.sink.close();
      case 'response.done':
        // root OR nested under 'response', depending on the node — read both.
        final x = (ev['x_ouroboros'] ??
                   (ev['response'] as Map?)?['x_ouroboros'] ?? const {}) as Map;
        // null means not measured on this node. Never render it as 0 ms.
        print('\nttft_ms=${x["ttft_ms"]} e2e_ms=${x["end_to_end_ms"]}');
        if (toolTurnPending) {
          // One create for the whole turn's tool outputs. Single writer.
          toolTurnPending = false;
          send({'type': 'response.create'});
        } else {
          print(audio.length > 0
              ? 'pcm16 bytes=${audio.length} at $rate Hz'
              : 'no audio on this node — fall back to platform speech');
          await channel.sink.close();
        }
    }
    final call = toolCallFrom(ev);
    if (call != null) {
      final (name, callId, args) = call;
      if (handled.add(callId)) {          // false => already handled, skip
        send({'type': 'conversation.item.create', 'item': {
          'type': 'function_call_output', 'call_id': callId,
          'output': jsonEncode(runTool(name, jsonDecode(args) as Map<String, dynamic>)),
        }});
        toolTurnPending = true;
      }
    }
  }
}

Rust

# Cargo.toml
# tokio = { version = "1", features = ["full"] }
# tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] }
# futures-util = "0.3"   serde_json = "1"   base64 = "0.22"   anyhow = "1"
use anyhow::Result;
use base64::{engine::general_purpose::STANDARD as B64, Engine};
use futures_util::{SinkExt, StreamExt};
use serde_json::{json, Value};
use std::collections::HashSet;
use tokio_tungstenite::tungstenite::{client::IntoClientRequest, Message};

/// Either frame may carry the finished call — accept both, dedupe on call_id.
fn tool_call_from(ev: &Value) -> Option<(String, String, String)> {
    let t = ev["type"].as_str()?;
    let src = match t {
        "response.function_call_arguments.done" => ev,
        "response.output_item.done" if ev["item"]["type"] == "function_call" => &ev["item"],
        _ => return None,
    };
    Some((
        src["name"].as_str()?.to_string(),
        src["call_id"].as_str()?.to_string(),
        src["arguments"].as_str().unwrap_or("{}").to_string(),
    ))
}

fn run_tool(name: &str, args: &Value) -> Value {
    // A tool call off the wire is a request, not an authorization.
    match name {
        "set_thermostat" => json!({"ok": true, "room": args["room"], "celsius": args["celsius"]}),
        _ => json!({"ok": false, "error": "unknown tool"}),
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let key = std::env::var("HAWKTALK_API_KEY")?;
    // model=auto: the router picks a tier per utterance. A pinned "tier:think"
    // on a smalltalk-heavy workload is the classic 10x bill.
    let mut req = "wss://api.hawktalk.ai/v1/realtime?model=auto".into_client_request()?;
    req.headers_mut()
        .insert("Authorization", format!("Bearer {key}").parse()?);
    // 401 / 429 / 503 come back on the HTTP upgrade, as an Err here.
    let (ws, _) = tokio_tungstenite::connect_async(req).await?;
    let (mut tx, mut rx) = ws.split();

    // Text frames only — Message::Binary is rejected by design.
    // Tools go in once, with the session.
    tx.send(Message::Text(json!({
        "type": "session.update",
        "session": {
            "model": "auto",
            "modalities": ["text", "audio"],
            "instructions": "You run a house. Answer in one sentence.",
            "tools": [{
                "type": "function", "name": "set_thermostat",
                "description": "Set the target temperature for a room.",
                "parameters": {"type": "object", "properties": {
                    "room": {"type": "string"}, "celsius": {"type": "number"}},
                    "required": ["room", "celsius"]}
            }],
            "tool_choice": "auto"
        }
    }).to_string().into())).await?;

    // 24 kHz mono s16le, 100 ms per append. The commit is YOUR VAD.
    let pcm = std::fs::read("utterance_24k.pcm")?;
    for chunk in pcm.chunks(4800) {
        tx.send(Message::Text(json!({
            "type": "input_audio_buffer.append", "audio": B64.encode(chunk)
        }).to_string().into())).await?;
    }
    tx.send(Message::Text(json!({"type": "input_audio_buffer.commit"}).to_string().into())).await?;
    tx.send(Message::Text(json!({"type": "response.create"}).to_string().into())).await?;

    let mut audio: Vec<u8> = Vec::new();
    let mut rate: Option<i64> = None;
    let mut handled: HashSet<String> = HashSet::new();
    let mut tool_turn_pending = false;

    while let Some(msg) = rx.next().await {
        let raw = match msg? {
            Message::Text(t) => t,
            Message::Close(_) => break,
            _ => continue,
        };
        let ev: Value = serde_json::from_str(raw.as_str())?;
        match ev["type"].as_str().unwrap_or("") {
            "conversation.item.input_audio_transcription.completed" => {
                println!("heard: {}", ev["transcript"].as_str().unwrap_or(""));
            }
            "response.text.delta" => print!("{}", ev["delta"].as_str().unwrap_or("")),
            "response.audio.delta" => {
                // some builds carry the base64 under "audio", not "delta"
                let b64 = ev["delta"].as_str()
                    .or_else(|| ev["audio"].as_str())
                    .unwrap_or("");
                audio.extend(B64.decode(b64)?);
                rate = ev["x_sample_rate_hz"].as_i64().or(rate);
            }
            "error" => {
                eprintln!("error: {} {}", ev["error"]["code"], ev["error"]["message"]);
                break;
            }
            "response.done" => {
                // Telemetry is at the frame root on some nodes and under
                // "response" on others. Read both.
                let x = if ev["x_ouroboros"].is_object() {
                    &ev["x_ouroboros"]
                } else {
                    &ev["response"]["x_ouroboros"]
                };
                // x_ouroboros fields are null when not measured — serde gives you
                // Value::Null, and you must not unwrap_or(0) it into a fake metric.
                println!("\nttft_ms={} first_audio_out_ms={} end_to_end_ms={}",
                         x["ttft_ms"], x["first_audio_out_ms"], x["end_to_end_ms"]);
                if tool_turn_pending {
                    // One create for the whole turn's outputs. Single writer.
                    tool_turn_pending = false;
                    tx.send(Message::Text(
                        json!({"type": "response.create"}).to_string().into())).await?;
                    continue;
                }
                break;
            }
            _ => {}
        }
        if let Some((name, call_id, args)) = tool_call_from(&ev) {
            if handled.insert(call_id.clone()) {   // false => duplicate frame
                let parsed: Value = serde_json::from_str(&args)?;
                tx.send(Message::Text(json!({
                    "type": "conversation.item.create",
                    "item": {"type": "function_call_output", "call_id": call_id,
                             "output": run_tool(&name, &parsed).to_string()}
                }).to_string().into())).await?;
                tool_turn_pending = true;
            }
        }
    }

    if audio.is_empty() {
        eprintln!("no audio on this node — fall back to platform speech, never fake it");
    } else {
        println!("pcm16 bytes={} at {:?} Hz", audio.len(), rate);
    }
    Ok(())
}

Go

// go get github.com/gorilla/websocket
package main

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"

	"github.com/gorilla/websocket"
)

type ev map[string]any

// Accepts either frame that can carry a finished tool call.
func toolCallFrom(e ev) (name, callID, args string, ok bool) {
	switch e["type"] {
	case "response.function_call_arguments.done":
		name, _ = e["name"].(string)
		callID, _ = e["call_id"].(string)
		args, _ = e["arguments"].(string)
	case "response.output_item.done":
		it, _ := e["item"].(map[string]any)
		if it == nil || it["type"] != "function_call" {
			return "", "", "", false
		}
		name, _ = it["name"].(string)
		callID, _ = it["call_id"].(string)
		args, _ = it["arguments"].(string)
	default:
		return "", "", "", false
	}
	if args == "" {
		args = "{}"
	}
	return name, callID, args, name != "" && callID != ""
}

func runTool(name string, args map[string]any) map[string]any {
	// A returned call is a request. Authorize it on your side, every time.
	if name == "set_thermostat" {
		return map[string]any{"ok": true, "room": args["room"], "celsius": args["celsius"]}
	}
	return map[string]any{"ok": false, "error": "unknown tool"}
}

func main() {
	key := os.Getenv("HAWKTALK_API_KEY")
	hdr := http.Header{"Authorization": []string{"Bearer " + key}}
	// auto = per-utterance routing. Pin a tier only with a measured reason.
	c, resp, err := websocket.DefaultDialer.Dial(
		"wss://api.hawktalk.ai/v1/realtime?model=auto", hdr)
	if err != nil {
		// 401 invalid_api_key and 429 rate_limit_exceeded surface here, on the
		// HTTP upgrade — not as a frame. Read resp.StatusCode before retrying.
		if resp != nil {
			log.Fatalf("dial: %v (http %d, retry-after %q)",
				err, resp.StatusCode, resp.Header.Get("Retry-After"))
		}
		log.Fatalf("dial: %v", err)
	}
	defer c.Close()

	send := func(v any) {
		b, err := json.Marshal(v)
		if err != nil {
			log.Fatalf("marshal: %v", err)
		}
		// TextMessage always. BinaryMessage is rejected by design.
		if err := c.WriteMessage(websocket.TextMessage, b); err != nil {
			log.Fatalf("write: %v", err)
		}
	}

	// Once per session — never re-send tools on each turn.
	send(ev{"type": "session.update", "session": ev{
		"model":        "auto",
		"modalities":   []string{"text", "audio"},
		"instructions": "You run a house. Answer in one sentence.",
		"tool_choice":  "auto",
		"tools": []ev{{
			"type": "function", "name": "set_thermostat",
			"description": "Set the target temperature for a room.",
			"parameters": ev{"type": "object", "properties": ev{
				"room": ev{"type": "string"}, "celsius": ev{"type": "number"}},
				"required": []string{"room", "celsius"}},
		}},
	}})

	pcm, err := os.ReadFile("utterance_24k.pcm") // 24 kHz mono s16le
	if err != nil {
		log.Fatalf("read pcm: %v", err)
	}
	for i := 0; i < len(pcm); i += 4800 { // 100 ms per append
		end := i + 4800
		if end > len(pcm) {
			end = len(pcm)
		}
		send(ev{"type": "input_audio_buffer.append",
			"audio": base64.StdEncoding.EncodeToString(pcm[i:end])})
	}
	// Your VAD said the utterance ended. The server will never say it for you.
	send(ev{"type": "input_audio_buffer.commit"})
	send(ev{"type": "response.create"})

	var audio []byte
	var rate any
	handled := map[string]bool{} // call_ids already run — the dedupe
	toolTurnPending := false

	for {
		_, raw, err := c.ReadMessage()
		if err != nil {
			log.Fatalf("read: %v", err)
		}
		var e ev
		if err := json.Unmarshal(raw, &e); err != nil {
			continue // forward-compatible: an unparseable frame is never fatal
		}
		switch e["type"] {
		case "conversation.item.input_audio_transcription.completed":
			fmt.Println("heard:", e["transcript"])
		case "response.text.delta":
			if d, ok := e["delta"].(string); ok {
				fmt.Print(d)
			}
		case "response.audio.delta":
			d, ok := e["delta"].(string)
			if !ok {
				// some builds carry the base64 under "audio", not "delta"
				d, _ = e["audio"].(string)
			}
			b, err := base64.StdEncoding.DecodeString(d)
			if err != nil {
				log.Fatalf("audio decode: %v", err)
			}
			audio = append(audio, b...)
			if r, ok := e["x_sample_rate_hz"]; ok {
				rate = r
			}
		case "error":
			er, _ := e["error"].(map[string]any)
			log.Fatalf("error: %v %v", er["code"], er["message"])
		case "response.done":
			x, _ := e["x_ouroboros"].(map[string]any)
			if x == nil {
				// nested under "response" on some nodes — read both.
				if r, _ := e["response"].(map[string]any); r != nil {
					x, _ = r["x_ouroboros"].(map[string]any)
				}
			}
			// nil here means NOT MEASURED. Do not default it to 0.
			fmt.Printf("\nttft_ms=%v first_audio_out_ms=%v end_to_end_ms=%v\n",
				x["ttft_ms"], x["first_audio_out_ms"], x["end_to_end_ms"])
			if toolTurnPending {
				// One create for the whole turn's outputs. Single writer.
				toolTurnPending = false
				send(ev{"type": "response.create"})
				continue
			}
			if len(audio) == 0 {
				fmt.Println("no audio on this node — fall back to platform speech")
			} else {
				fmt.Printf("pcm16 bytes=%d at %v Hz\n", len(audio), rate)
			}
			return
		}
		if name, callID, args, ok := toolCallFrom(e); ok && !handled[callID] {
			handled[callID] = true
			var parsed map[string]any
			if err := json.Unmarshal([]byte(args), &parsed); err != nil {
				log.Fatalf("tool args: %v", err)
			}
			out, err := json.Marshal(runTool(name, parsed))
			if err != nil {
				log.Fatalf("tool out: %v", err)
			}
			send(ev{"type": "conversation.item.create", "item": ev{
				"type": "function_call_output", "call_id": callID, "output": string(out)}})
			toolTurnPending = true
		}
	}
}

The frames you actually see

# One voice turn with a tool call. This is a GROUPING, not a strict sequence:
# text and audio deltas INTERLEAVE, and telemetry lands between anything.
# Every frame is JSON text.
session.created
session.updated
conversation.created
conversation.item.created                  # your committed audio, as an item
conversation.item.input_audio_transcription.completed
response.created
response.output_item.added
response.content_part.added
response.text.delta        × N          # interleaved with the audio deltas
response.audio.delta       × N          # base64 pcm16 + x_sample_rate_hz
response.viseme.delta      × N          # mouth shapes, if the node renders them.
                                          # The EVENT NAME is published; its payload keys
                                          # are NOT. Tested nodes put the shape under
                                          # "viseme" — skip a frame with no key you
                                          # recognise instead of throwing on it.
ouroboros.telemetry                      # every 8 tokens, plus a 5 s idle heartbeat
ouroboros.endpoint                       # semantic endpointing — NOT a VAD event
response.text.done / response.audio.done
response.content_part.done / response.output_item.done
response.done                             # carries x_ouroboros
rate_limits.updated

# A real response.done from a node whose TTS seam is NOT wired. Here the
# telemetry is at the frame ROOT; other nodes nest the same object under
# "response". Read both places — never just the one you saw first.
{
  "type": "response.done",
  "response": {
    "id": "resp_9f2c1a",
    "status": "completed",
    "usage": {"input_tokens": 312, "output_tokens": 48, "total_tokens": 360}
  },
  "x_ouroboros": {
    "stt_ms": 412,
    "ttft_ms": 286,
    "first_audio_out_ms": null,     # not measured — no audio was produced
    "end_to_end_ms": 1180,
    "end_to_end_audio_ms": null      # null is the honest answer, 0 would be a lie
  }
}

Everything in this section describes /live/brain, which is PREVIEW: loopback only, no auth wired, not served on api.hawktalk.ai. It is documented because the shape is stable and because you should design your client for it now — but you cannot call it in production today, and any page that implies otherwise is lying to you. The x_ fields and gen stamps below are mux-local: they are not part of the published /v1/realtime contract.

Why lanes instead of one stream. On /v1/realtime one generator produces everything in order: it cannot emit the next audio chunk while it is waiting on your tool to return, because the tool result is an input to the next token. That is head-of-line blocking, and it is why single-stream voice agents go silent for the duration of every API call they make. Lanes break the ordering constraint: the audio lane keeps speaking a filled pause, the tools lane blocks on your actuator, the thinking lane checks the claim, and the async lane is still off doing a six-second retrieval. They share state and a generation counter; they do not share a queue.

Physically it is still one socket. Independence is scheduling, not a second connection: the mux drains an internal priority queue so a telemetry burst can never delay an audio frame. The published order is fixed:

# lane priority — lower drains first, audio always wins
AUDIO > VISEME > PRESENCE > TEXT > USER_AFFECT > CONDUCT > AGENT > TELEMETRY

# under backpressure the queue sheds from the BOTTOM: telemetry dies first,
# audio never drops. Design your client to tolerate missing telemetry frames.

Lane	Carries	Frames	Deadline	Status
audio	what the person hears, plus mouth shapes	response.audio.delta, response.audio.done, response.viseme.delta — the viseme event name is published, its payload keys are not: tested nodes carry the shape under viseme, so skip an unrecognised payload rather than throwing on it	hard realtime	SHIPPED on /v1/realtime
text	the words of the reply, for captions and logs	response.text.delta, response.text.done	soft realtime	SHIPPED
tools	function calls out, results back in	response.output_item.done (published) or response.function_call_arguments.done (some builds) — dedupe on call_id, then conversation.item.create	as fast as your actuator	SHIPPED
thinking	private reasoning the user never hears	no server frame type exists	may miss the turn	DESIGN — client-composed today
async	slow work that lands after the turn closed	no server frame type exists	seconds to minutes	DESIGN — client-composed today
presence	backchannels and filled pauses while a slow tier composes	ouroboros.presence	sub-second or worthless	PREVIEW (mux only)
affect	how the person sounds, not what they said	ouroboros.user_affect	per utterance	PREVIEW, self-marks x_stub
conduct	the baton: who is allowed to speak right now	hawk.conduct	immediate	PREVIEW (mux only)
telemetry	token pace, tier, timings, liveness	ouroboros.telemetry	best effort, sheds first	SHIPPED on /v1/realtime

The first frame is a capability declaration, and you must read it. The mux opens by telling you which lanes are real on this node and which are standing in. A lane that is not backed by a live upstream says so — it does not go quiet and it does not fake a value:

# first frame on /live/brain, from a node with no SER sidecar and no reactive model
{
  "type": "hawk.brain.hello",
  "lanes": {
    "partial_transcript": "live (gateway)",
    "endpoint":           "live (gateway)",
    "text":               "live (gateway)",
    "audio":              "live (gateway)",
    "viseme":             "live (gateway)",
    "telemetry":          "live (gateway)",
    "user_affect":        "stub (SER not up)",
    "presence":           "heuristic (reactive model not up)",
    "conduct":            "live (lane_mux)"
  },
  "priority": "AUDIO>VISEME>PRESENCE>TEXT>USER_AFFECT>CONDUCT>AGENT>TELEMETRY"
}

# the three lanes the mux adds on top of the gateway's realtime frames.
# every one is gen-stamped: `gen` is the turn generation it belongs to.

{"type": "hawk.conduct", "op": "route", "target": "reactive", "gen": 3,
 "reason": "utterance committed — reading affect", "x_lane_mux": true}

{"type": "ouroboros.presence", "kind": "filled_pause", "text": "let me check",
 "gen": 3, "x_reactive": "heuristic", "x_stub": true}

# affect with no SER sidecar behind it. NOT an emotion of "neutral" —
# an explicit stub with the reason attached. Render nothing, or render unknown.
{"type": "ouroboros.user_affect", "gen": 3, "x_stub": true,
 "x_reason": "SER sidecar (:8896) not reachable"}

# and with the sidecar up
{"type": "ouroboros.user_affect", "gen": 4, "emotion": "frustrated",
 "arousal": 0.71, "valence": -0.42, "confidence": 0.63, "x_ser_ms": 88}

gen is the whole reconciliation story in one integer. The mux bumps it on every committed utterance, on VAD onset, and on response.cancel — and it drops superseded presence and text frames at the mux rather than shipping them to you. Your client must apply the same rule to its own lanes: a result computed for gen 3 is dead once gen 4 has begun. Gate on gen before you filter by lane, or a generation bump announced on a lane you ignore will let stale frames through on a lane you keep. Skipping this is how a superharness becomes four race conditions in a trenchcoat.

You cannot run this against api.hawktalk.ai. /live/brain is served by a local node only — ws://localhost:8891/live/brain — with no auth wired in the preview build. Send the Authorization header anyway: when the route is promoted, your code should not change. Treat everything here as a client you are pre-building, and keep /v1/realtime as the path you actually ship on.

There is no subscribe frame, and you must not invent one. The mux sends every lane it has and declares them in hawk.brain.hello. "Subscribing to a lane" means registering a handler for that lane's frame types and ignoring the rest — the filtering is yours. What the hello frame buys you is the right to refuse: if your product sells affect and lanes.user_affect comes back "stub (SER not up)", fail loudly at startup instead of shipping a neutral-emotion placeholder to a customer.

There is also no session field for the input sample rate. Input is 24000 Hz mono s16le by default on both the mux and /v1/realtime; resample on your side. Do not send a rate-declaring key in session.update — none exists, and a node that validates session fields will answer 400 invalid_request.

Two rules that make the difference between a mux client and a mess: drop anything stamped with an old gen, and never render a frame carrying x_stub: true as if it were measured. The Rust and Go clients follow the same three steps as the two below — read the hello, gate on gen, route by frame type — with the connect code from the shipped section above.

curl — is it even there?

# The preview mux binds loopback. This is the only place it answers.
curl -sS -i --http1.1 \
  -H "Connection: Upgrade" -H "Upgrade: websocket" \
  -H "Sec-WebSocket-Version: 13" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  http://localhost:8891/live/brain
# 101 Switching Protocols => the preview mux is up on this node.
# anything else => it is not, and there is no public fallback. Use /v1/realtime.

# The lanes the SHIPPED endpoint gives you are visible without any of this:
curl -sS http://localhost:8890/health      # no auth; lists models + voice seams

Python

# pip install "websockets>=14"   (13.x: extra_headers= instead)
# PREVIEW: /live/brain is loopback-only and has no auth wired. Do not point
# this at a production host — there is nothing there.
import asyncio, base64, json, os
import websockets

MUX = os.environ.get("HAWKTALK_MUX", "ws://localhost:8891/live/brain")
KEY = os.environ.get("HAWKTALK_API_KEY", "")

# "Subscribing" is client-side routing. Name the lanes you want; everything
# else is ignored without error, which is also your forward-compat story.
LANES = {
    "audio":     {"response.audio.delta", "response.audio.done", "response.viseme.delta"},
    "text":      {"response.text.delta", "response.text.done"},
    "presence":  {"ouroboros.presence"},
    "affect":    {"ouroboros.user_affect"},
    "conduct":   {"hawk.conduct"},
    "telemetry": {"ouroboros.telemetry"},
}
WANT = {"audio", "text", "presence", "conduct"}   # affect/telemetry: not this product

def lane_of(frame_type):
    for lane, types in LANES.items():
        if frame_type in types:
            return lane
    return None

class MuxClient:
    def __init__(self):
        self.gen = 0
        self.lanes = {}          # from hawk.brain.hello — the capability contract

    def accept(self, ev):
        # RULE 1: a frame stamped with an older generation is dead. The user has
        # moved on; speaking it is worse than dropping it. This runs on EVERY
        # frame, including lanes you ignore — a gen bump announced on telemetry
        # still has to retire stale frames on the lanes you keep.
        g = ev.get("gen")
        if g is not None and g < self.gen:
            return False
        if g is not None:
            self.gen = g
        # RULE 2: a stub is not a measurement. Never render it as one.
        if ev.get("x_stub"):
            print(f"[stub] {ev.get('type')}: {ev.get('x_reason', 'lane standing in')}")
            return False
        return True

    async def run(self):
        async with websockets.connect(
            MUX,
            additional_headers=({"Authorization": f"Bearer {KEY}"} if KEY else None),
            max_size=16 * 1024 * 1024,
        ) as ws:
            hello = json.loads(await ws.recv())
            if hello.get("type") != "hawk.brain.hello":
                raise RuntimeError(f"not the mux: {hello.get('type')}")
            self.lanes = hello.get("lanes") or {}
            print("lane declaration:", json.dumps(self.lanes, indent=2))
            print("priority:", hello.get("priority"))

            # Refuse to run degraded if a lane you SELL is standing in.
            for lane in ("audio", "text"):
                state = str(self.lanes.get(lane, "absent"))
                if not state.startswith("live"):
                    raise RuntimeError(f"lane {lane} is '{state}' — refusing to start")

            # The mux forwards your frames to the gateway verbatim: the send
            # vocabulary is exactly the /v1/realtime one you already wrote.
            # Input is 24000 Hz mono s16le by default — resample on your side.
            # There is NO session field for the input rate. Do not invent one.
            await ws.send(json.dumps({"type": "session.update",
                                      "session": {"model": "auto"}}))
            pcm = open("utterance_24k.pcm", "rb").read()
            for i in range(0, len(pcm), 4800):
                await ws.send(json.dumps({"type": "input_audio_buffer.append",
                                          "audio": base64.b64encode(pcm[i:i+4800]).decode()}))
            # commit bumps the mux generation and kicks the affect + presence lanes
            await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
            await ws.send(json.dumps({"type": "response.create"}))

            async for raw in ws:
                ev = json.loads(raw)
                if not self.accept(ev):        # gen gate FIRST, lane filter second
                    continue
                lane = lane_of(ev.get("type", ""))
                if lane not in WANT:
                    continue
                if lane == "text" and ev["type"].endswith(".delta"):
                    print(ev.get("delta", ""), end="", flush=True)
                elif lane == "presence":
                    # a backchannel while a slow tier composes — speak it IMMEDIATELY
                    # or not at all; a late "mm-hm" is worse than silence.
                    print(f"\n[presence/{ev.get('kind')}] {ev.get('text', '')}")
                elif lane == "conduct":
                    print(f"\n[baton] {ev.get('op')} -> {ev.get('target')}: {ev.get('reason')}")
                elif lane == "audio" and ev["type"] == "response.audio.delta":
                    # hand straight to the player; x_sample_rate_hz is authoritative
                    pass

asyncio.run(MuxClient().run())

TypeScript / Node [DESIGN]

// npm i ws   —   PREVIEW endpoint: loopback only, no auth wired.
import WebSocket from "ws";
import { readFileSync } from "node:fs";

const MUX = process.env.HAWKTALK_MUX ?? "ws://localhost:8891/live/brain";
const KEY = process.env.HAWKTALK_API_KEY ?? "";

type Lane = "audio" | "text" | "presence" | "affect" | "conduct" | "telemetry";

const LANE_OF: Record<string, Lane> = {
  "response.audio.delta": "audio",
  "response.audio.done": "audio",
  "response.viseme.delta": "audio",
  "response.text.delta": "text",
  "response.text.done": "text",
  "ouroboros.presence": "presence",
  "ouroboros.user_affect": "affect",
  "hawk.conduct": "conduct",
  "ouroboros.telemetry": "telemetry",
};
// Client-side subscription. There is no server-side lane filter to ask for.
const WANT = new Set<Lane>(["audio", "text", "presence", "conduct"]);

const ws = new WebSocket(MUX, {
  headers: KEY ? { Authorization: `Bearer ${KEY}` } : {},   // ignored in preview
  maxPayload: 16 * 1024 * 1024,
});

let gen = 0;
let lanes: Record<string, string> = {};

ws.on("message", (raw) => {
  const ev = JSON.parse(raw.toString());

  if (ev.type === "hawk.brain.hello") {
    lanes = ev.lanes ?? {};
    console.log("lanes:", lanes, "\npriority:", ev.priority);
    // Fail loudly rather than shipping a degraded lane as if it were real.
    for (const required of ["audio", "text"]) {
      if (!lanes[required]?.startsWith("live")) {
        console.error(`lane ${required} is "${lanes[required] ?? "absent"}" — refusing`);
        ws.close(); return;
      }
    }
    // Same send vocabulary as /v1/realtime: the mux forwards it verbatim.
    // 24 kHz mono s16le in, resampled by YOU — there is no session field
    // for the input rate, so do not send one.
    ws.send(JSON.stringify({ type: "session.update", session: { model: "auto" } }));
    const pcm = readFileSync("utterance_24k.pcm");
    for (let i = 0; i < pcm.length; i += 4800)
      ws.send(JSON.stringify({ type: "input_audio_buffer.append",
        audio: pcm.subarray(i, i + 4800).toString("base64") }));
    ws.send(JSON.stringify({ type: "input_audio_buffer.commit" }));  // bumps gen
    ws.send(JSON.stringify({ type: "response.create" }));
    return;
  }

  // RULE 1 — generation gate, BEFORE the lane filter. Old gen = user moved on.
  if (typeof ev.gen === "number") {
    if (ev.gen < gen) return;
    gen = ev.gen;
  }
  // RULE 2 — a stub is not data. Log it, never render it.
  if (ev.x_stub) {
    console.warn(`[stub] ${ev.type}: ${ev.x_reason ?? "lane standing in"}`);
    return;
  }

  const lane = LANE_OF[ev.type];
  if (!lane || !WANT.has(lane)) return;      // unknown frames are never fatal

  switch (lane) {
    case "text":
      if (ev.type === "response.text.delta") process.stdout.write(ev.delta ?? "");
      break;
    case "presence":
      // Speak it now or drop it. A filled pause that lands after the real
      // answer is a bug the user hears.
      console.log(`\n[presence/${ev.kind}] ${ev.text ?? ""}`);
      break;
    case "conduct":
      console.log(`\n[baton] ${ev.op} -> ${ev.target}: ${ev.reason ?? ""}`);
      break;
    case "audio":
      // straight to the player; x_sample_rate_hz is authoritative
      break;
  }
});

ws.on("error", (e) => {
  console.error("mux unreachable — this endpoint is PREVIEW and loopback-only:", e.message);
});

Dart / Flutter [DESIGN]

// pubspec.yaml:  web_socket_channel: ^3.0.0
// PREVIEW: loopback only. On Android an emulator reaches the host at
// 10.0.2.2, and cleartext ws:// needs a debug network-security config.
import 'dart:convert';
import 'dart:io';
import 'package:web_socket_channel/io.dart';

enum Lane { audio, text, presence, affect, conduct, telemetry }

const laneOf = <String, Lane>{
  'response.audio.delta': Lane.audio,
  'response.audio.done': Lane.audio,
  'response.viseme.delta': Lane.audio,
  'response.text.delta': Lane.text,
  'response.text.done': Lane.text,
  'ouroboros.presence': Lane.presence,
  'ouroboros.user_affect': Lane.affect,
  'hawk.conduct': Lane.conduct,
  'ouroboros.telemetry': Lane.telemetry,
};

/// Subscription is a client-side filter — the mux has no subscribe verb.
const want = {Lane.audio, Lane.text, Lane.presence, Lane.conduct};

Future<void> main() async {
  final mux = Platform.environment['HAWKTALK_MUX'] ?? 'ws://localhost:8891/live/brain';
  final key = Platform.environment['HAWKTALK_API_KEY'] ?? '';

  final channel = IOWebSocketChannel.connect(
    Uri.parse(mux),
    headers: key.isEmpty ? null : {'Authorization': 'Bearer $key'},
  );
  void send(Object o) => channel.sink.add(jsonEncode(o));

  var gen = 0;
  Map<String, dynamic> lanes = {};

  await for (final raw in channel.stream) {
    final ev = jsonDecode(raw as String) as Map<String, dynamic>;

    if (ev['type'] == 'hawk.brain.hello') {
      lanes = ((ev['lanes'] ?? const {}) as Map).cast<String, dynamic>();
      stdout.writeln('lanes: $lanes');
      stdout.writeln('priority: ${ev["priority"]}');
      for (final required in ['audio', 'text']) {
        final state = (lanes[required] ?? 'absent').toString();
        if (!state.startsWith('live')) {
          // Honest frontier: refuse, do not degrade silently.
          throw StateError('lane $required is "$state" — refusing to start');
        }
      }
      // No input-rate session field exists. 24 kHz mono s16le, resampled
      // by you, is the contract.
      send({'type': 'session.update', 'session': {'model': 'auto'}});
      final pcm = await File('utterance_24k.pcm').readAsBytes();
      for (var i = 0; i < pcm.length; i += 4800) {
        final end = (i + 4800 < pcm.length) ? i + 4800 : pcm.length;
        send({'type': 'input_audio_buffer.append',
              'audio': base64Encode(pcm.sublist(i, end))});
      }
      send({'type': 'input_audio_buffer.commit'});   // bumps the mux generation
      send({'type': 'response.create'});
      continue;
    }

    // RULE 1 — generation gate, before the lane filter.
    final g = ev['gen'];
    if (g is int) {
      if (g < gen) continue;
      gen = g;
    }
    // RULE 2 — never render a stub as a measurement.
    if (ev['x_stub'] == true) {
      stderr.writeln('[stub] ${ev["type"]}: ${ev["x_reason"] ?? "lane standing in"}');
      continue;
    }

    final lane = laneOf[ev['type']];
    if (lane == null || !want.contains(lane)) continue;

    switch (lane) {
      case Lane.text:
        if (ev['type'] == 'response.text.delta') stdout.write(ev['delta'] ?? '');
      case Lane.presence:
        stdout.writeln('\n[presence/${ev["kind"]}] ${ev["text"] ?? ""}');
      case Lane.conduct:
        stdout.writeln('\n[baton] ${ev["op"]} -> ${ev["target"]}: ${ev["reason"] ?? ""}');
      case Lane.audio:
        // feed the player; read x_sample_rate_hz off the frame
        break;
      default:
        break;
    }
  }
}

Status, plainly. There is no server-side thinking frame. Not on /v1/realtime, not on the preview mux. Any client that claims a server-side thinking lane today is showing you its own output with a different label. is a lane the mux carries; what you build now is the client-composed substitute below — and it is genuinely useful, because the discipline it forces (bounded deadline, confidence out, hedge instead of stall) is the same discipline the server lane will need.

The shape. The fast mind answers on the live socket. In parallel — not after — a review mind runs a cheap REST completion on a small tier, sees the same question and the fast mind's claim, and returns a confidence and an optional correction. That call never touches TTS, is never spoken, and is never allowed to make the person wait.

Yes, this pins a tier — on purpose. This is the "measured reason" the migration guide's step 2 carves out: the reviewer and the slow job are pinned deliberately (tier:quick here, tier:think in the next section) because their latency and cost budgets are fixed and known. Your user-facing turn stays on auto. Pinning a background lane is a decision; pinning the spoken turn is the overspend.

The rule that matters: low confidence hedges delivery, it does not stall the conversation. A voice agent that goes silent for 900 ms while it double-checks itself sounds broken; one that says "I think it's Tuesday — let me confirm" sounds careful. Give the review mind a hard deadline shorter than your first-audio budget. If it misses, do not wait — it becomes an async correction (next section), and the reconciliation policy decides whether it is worth interrupting for.

confidence	delivery	why
≥ 0.75	speak as written	Hedging a correct answer trains the user to distrust every answer.
0.40 – 0.75	speak with a hedge marker	The person can act on it and knows to verify. Costs nothing in latency.
< 0.40	speak the hedge and offer to check	Still no stall — you convert a wrong answer into an offer of work.
deadline missed	speak unhedged, reconcile later	Silence is the worst option. Late verdicts route to the async lane.

curl — the review mind is one cheap REST call

# tier:quick, not tier:think. The reviewer runs on EVERY turn — putting a large
# tier here roughly doubles your token bill for a checkbox.
curl -sS https://api.hawktalk.ai/v1/chat/completions \
  -H "Authorization: Bearer $HAWKTALK_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "tier:quick",
    "max_tokens": 120,
    "temperature": 0,
    "messages": [
      {"role": "system",
       "content": "You are a reviewer. Reply ONLY with JSON: {\"confidence\":0..1,\"correction\":string|null}. Never address the user."},
      {"role": "user",
       "content": "Question: when does the Fulton order ship?\nProposed answer: It ships Tuesday."}
    ]
  }'

# the response — ordinary OpenAI chat shape, plus HawkTalk telemetry
{
  "id": "chatcmpl-7a41e2",
  "object": "chat.completion",
  "model": "hawkalphaquick",
  "choices": [{
    "index": 0,
    "message": {"role": "assistant",
                "content": "{\"confidence\":0.42,\"correction\":\"Ship date is unverified; the order is still in packing.\"}"},
    "finish_reason": "stop"
  }],
  "usage": {"prompt_tokens": 96, "completion_tokens": 28, "total_tokens": 124}
}
# plus "x_timing" and "x_compute" objects: node-specific telemetry. Log them,
# do not branch on their contents — the field set varies by node.
# Note "model" in the response is the id that ACTUALLY ran. That is the id to
# put in your logs — never the alias you asked for. The id you see here came
# from this node's registry; do not hardcode it, read GET /v1/models.

Python

# pip install httpx  —  runs CONCURRENTLY with the live turn, never before it.
import asyncio, json, os
import httpx

BASE = "https://api.hawktalk.ai"
KEY = os.environ["HAWKTALK_API_KEY"]
DEADLINE_S = 0.40    # shorter than your first-audio budget. Miss = do not wait.

REVIEWER = ("You are a reviewer. Reply ONLY with JSON: "
            '{"confidence":0..1,"correction":string|null}. Never address the user.')

async def review(client, question, claim):
    """The thinking lane. Cheap tier, hard deadline, never spoken aloud."""
    try:
        r = await client.post(
            f"{BASE}/v1/chat/completions",
            headers={"Authorization": f"Bearer {KEY}"},
            json={"model": "tier:quick", "max_tokens": 120, "temperature": 0,
                  "messages": [
                      {"role": "system", "content": REVIEWER},
                      {"role": "user",
                       "content": f"Question: {question}\nProposed answer: {claim}"}]},
            timeout=DEADLINE_S)
        r.raise_for_status()
        body = r.json()
        verdict = json.loads(body["choices"][0]["message"]["content"])
        # unknown is null, never a fabricated number — if the reviewer did not
        # produce a confidence, say so instead of defaulting to 1.0 or 0.0.
        conf = verdict.get("confidence")
        return {"confidence": float(conf) if conf is not None else None,
                "correction": verdict.get("correction"),
                "ran_on": body.get("model")}
    except httpx.TimeoutException:
        return None                     # missed the turn -> async lane, not a stall
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 429:
            # the reviewer is the FIRST thing to shed under rate limiting.
            # Retry-After is authoritative; do NOT retry inside a live turn.
            print("reviewer rate-limited; Retry-After =",
                  e.response.headers.get("Retry-After"))
            return None
        if e.response.status_code in (503,):
            return None                 # model_unavailable: skip the check, do not stall
        raise
    except (KeyError, ValueError):
        return None                     # malformed verdict is not a verdict

def deliver(answer, verdict):
    """Hedge the DELIVERY. Never delay it."""
    if verdict is None or verdict["confidence"] is None:
        return answer                                   # unknown != low
    c = verdict["confidence"]
    if c >= 0.75:
        return answer
    if c >= 0.40:
        return f"I think {answer[0].lower()}{answer[1:]}"
    return (f"I think {answer[0].lower()}{answer[1:]} — I'm not certain, "
            "want me to check it properly?")

async def main():
    question = "When does the Fulton order ship?"
    claim = "It ships Tuesday."
    async with httpx.AsyncClient() as client:
        # In production this runs in parallel with response.create on the live
        # socket — the person is already hearing the fast mind's first words.
        verdict = await review(client, question, claim)
    print(deliver(claim, verdict))
    if verdict:
        print("[thinking, never spoken]", verdict)

asyncio.run(main())

TypeScript / Node

// Node 20+ (global fetch), ESM ("type": "module") for the top-level await.
// Fires alongside response.create — never after it.
const BASE = "https://api.hawktalk.ai";
const KEY = process.env.HAWKTALK_API_KEY!;
const DEADLINE_MS = 400;   // under your first-audio budget. Missing it is fine.

type Verdict = { confidence: number | null; correction: string | null; ranOn?: string };

const REVIEWER =
  'You are a reviewer. Reply ONLY with JSON: ' +
  '{"confidence":0..1,"correction":string|null}. Never address the user.';

async function review(question: string, claim: string): Promise<Verdict | null> {
  // AbortController is the deadline. There is no "wait a bit longer" branch:
  // a late verdict becomes an async correction, never a pause in the audio.
  const ac = new AbortController();
  const timer = setTimeout(() => ac.abort(), DEADLINE_MS);
  try {
    const r = await fetch(`${BASE}/v1/chat/completions`, {
      method: "POST",
      headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
      signal: ac.signal,
      body: JSON.stringify({
        model: "tier:quick",     // cheap tier: this runs on every single turn
        max_tokens: 120,
        temperature: 0,
        messages: [
          { role: "system", content: REVIEWER },
          { role: "user", content: `Question: ${question}\nProposed answer: ${claim}` },
        ],
      }),
    });
    if (r.status === 429) {
      // Shed the reviewer first — never retry inside a live turn.
      console.warn("reviewer rate-limited; retry-after =", r.headers.get("Retry-After"));
      return null;
    }
    if (!r.ok) {
      // 401 / 404 model_not_found / 503 model_unavailable land here. Log the
      // body — silently swallowing it is how a dead reviewer goes unnoticed.
      console.warn("review failed:", r.status, await r.text());
      return null;
    }
    const body = await r.json();
    const v = JSON.parse(body.choices[0].message.content);
    return {
      confidence: typeof v.confidence === "number" ? v.confidence : null,
      correction: v.correction ?? null,
      ranOn: body.model,          // the id that ACTUALLY ran — log this one
    };
  } catch (e) {
    // AbortError (deadline) or bad JSON: no verdict, no stall.
    return null;
  } finally {
    clearTimeout(timer);
  }
}

function deliver(answer: string, v: Verdict | null): string {
  if (!v || v.confidence === null) return answer;          // unknown is not low
  const lower = answer[0].toLowerCase() + answer.slice(1);
  if (v.confidence >= 0.75) return answer;
  if (v.confidence >= 0.40) return `I think ${lower}`;
  return `I think ${lower} — I'm not certain, want me to check it properly?`;
}

const question = "When does the Fulton order ship?";
const claim = "It ships Tuesday.";
const verdict = await review(question, claim);
console.log(deliver(claim, verdict));
if (verdict) console.log("[thinking, never spoken]", verdict);

Dart / Flutter

// pubspec.yaml:  http: ^1.2.0
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:http/http.dart' as http;

const base = 'https://api.hawktalk.ai';
const deadline = Duration(milliseconds: 400);
const reviewer =
    'You are a reviewer. Reply ONLY with JSON: '
    '{"confidence":0..1,"correction":string|null}. Never address the user.';

class Verdict {
  final double? confidence;   // null = the reviewer did not say. Not 0.0.
  final String? correction;
  final String? ranOn;
  Verdict(this.confidence, this.correction, this.ranOn);
}

Future<Verdict?> review(String question, String claim) async {
  final key = Platform.environment['HAWKTALK_API_KEY']!;
  try {
    final r = await http
        .post(Uri.parse('$base/v1/chat/completions'),
            headers: {
              'Authorization': 'Bearer $key',
              'Content-Type': 'application/json',
            },
            body: jsonEncode({
              'model': 'tier:quick',      // runs every turn — keep it small
              'max_tokens': 120,
              'temperature': 0,
              'messages': [
                {'role': 'system', 'content': reviewer},
                {'role': 'user',
                 'content': 'Question: $question\nProposed answer: $claim'},
              ],
            }))
        .timeout(deadline);          // hard deadline; a miss is not an error
    if (r.statusCode == 429) {
      stderr.writeln('reviewer rate-limited; Retry-After=${r.headers["retry-after"]}');
      return null;                   // shed the thinking lane before the voice lane
    }
    if (r.statusCode != 200) {
      stderr.writeln('review failed ${r.statusCode}: ${r.body}');
      return null;
    }
    final body = jsonDecode(r.body) as Map<String, dynamic>;
    final content = body['choices'][0]['message']['content'] as String;
    final v = jsonDecode(content) as Map<String, dynamic>;
    final c = v['confidence'];
    return Verdict(c is num ? c.toDouble() : null,
                   v['correction'] as String?, body['model'] as String?);
  } on TimeoutException {
    return null;                     // missed the turn: route to the async lane
  } on FormatException {
    return null;                     // a malformed verdict is not a verdict
  }
}

String deliver(String answer, Verdict? v) {
  if (v == null || v.confidence == null) return answer;
  final lower = answer[0].toLowerCase() + answer.substring(1);
  if (v.confidence! >= 0.75) return answer;
  if (v.confidence! >= 0.40) return 'I think $lower';
  return "I think $lower — I'm not certain, want me to check it properly?";
}

Future<void> main() async {
  const question = 'When does the Fulton order ship?';
  const claim = 'It ships Tuesday.';
  final v = await review(question, claim);   // concurrent with the spoken turn
  stdout.writeln(deliver(claim, v));
  if (v != null) stdout.writeln('[thinking, never spoken] conf=${v.confidence} '
      'correction=${v.correction} ran_on=${v.ranOn}');
}

for the server lane: there is no async frame type on /v1/realtime or on the preview mux. What exists today, and what the examples below build, is the client-side async lane — a background job with a generation stamp, a materiality gate, and a way back into the conversation. When the server lane ships, the job moves; the gate and the stamp do not.

The point of an async lane is that the turn does not wait for it. The person asked a question, got an answer in 700 ms, and moved on. Six seconds later the supplier feed comes back and says the answer was wrong. A single-lane agent has no way to express that — the turn is closed, the socket is idle, and the fact just sits in a variable. An async lane's whole job is to decide whether that fact is worth walking back into the room for, and then to walk in properly.

Interjection is expensive — gate it. Getting back into a conversation costs a full extra turn: input tokens for the whole context, output tokens, TTS, and the person's attention, which is the expensive one. Four things must all be true before you interject:

Gate	Test	If it fails
Fresh	The job's gen is still the current gen.	Drop it. The user has moved to another subject; a correction to a dead turn is noise.
Material	It changes a number, a date, or an action the person will take.	Stash it and fold it into the next natural turn. No wire traffic, no cost.
Non-duplicative	The audio lane did not already say this.	Drop it. Repeating yourself with new confidence reads as a malfunction.
Turn closed	No response is open on the socket.	Queue it until response.done. Two response.create frames in flight on one socket is the single-writer bug.

The mechanism. You do not speak on the async lane. You inject the finding as a conversation item and let the one voice that is already talking say it — otherwise two lanes narrate at once and the person hears an argument. Concretely: conversation.item.create with the finding, then response.create. That is it: two frames, both already in the shipped vocabulary — and in a real product both of them go through the reconciler in the next section, which is the only component allowed to send them.

curl — the slow job is just a bigger REST call

# The async lane's WORK is an ordinary completion (or your own database, or a
# supplier API). tier:think is where a slow, careful answer belongs — and why
# it must never be in the path of the spoken turn.
curl -sS https://api.hawktalk.ai/v1/chat/completions \
  -H "Authorization: Bearer $HAWKTALK_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "tier:think",
    "max_tokens": 400,
    "messages": [
      {"role": "system", "content": "Verify against the supplier feed. Answer in one sentence."},
      {"role": "user", "content": "Does the Fulton order ship Tuesday?"}
    ]
  }'

# seconds later — again, "model" is whatever this node's router actually ran:
{
  "id": "chatcmpl-b31f70",
  "object": "chat.completion",
  "model": "hawkalphaquick-hexagon",
  "choices": [{"index": 0, "finish_reason": "stop",
    "message": {"role": "assistant",
                "content": "No — the Fulton order is held in packing and now ships Thursday."}}],
  "usage": {"prompt_tokens": 141, "completion_tokens": 22, "total_tokens": 163}
}

# That result cannot be spoken from here. It has to go back through the live
# socket as a conversation item, because the socket owns the voice — and only
# once the current turn has closed:
#   {"type":"conversation.item.create","item":{...}}   then
#   {"type":"response.create","response":{"model":"auto"}}

Python

# pip install "websockets>=14" httpx
# One live socket + a background job. The job never touches the socket
# directly; it hands its finding to the interjector, which owns the turn.
import asyncio, json, os, time
import httpx, websockets

KEY = os.environ["HAWKTALK_API_KEY"]
BASE = "https://api.hawktalk.ai"
WS = "wss://api.hawktalk.ai/v1/realtime?model=auto"

class AsyncLane:
    """Slow work, stamped with the generation it was started for."""

    def __init__(self, ws):
        self.ws = ws
        self.gen = 0             # bumped by YOUR VAD on every commit
        self.turn_open = False
        self.spoken = []         # what the audio lane has already said
        self.stash = []          # immaterial findings, for the next natural turn
        self.pending = []        # material findings waiting for the turn to close

    async def job(self, question, gen):
        # Runs off the hot path. tier:think is slow and pricey — that is exactly
        # why it is here and not in the spoken turn.
        async with httpx.AsyncClient(timeout=30.0) as c:
            r = await c.post(f"{BASE}/v1/chat/completions",
                headers={"Authorization": f"Bearer {KEY}"},
                json={"model": "tier:think", "max_tokens": 400, "messages": [
                    {"role": "system",
                     "content": "Verify against the supplier feed. One sentence."},
                    {"role": "user", "content": question}]})
            r.raise_for_status()
            finding = r.json()["choices"][0]["message"]["content"]
        await self.land(finding, gen)

    def material(self, finding):
        # Your rule, your domain. "Changes a date, a number, or an action."
        return any(k in finding.lower() for k in ("no —", "thursday", "cancel", "held"))

    async def land(self, finding, gen):
        # GATE 1 — fresh? A finding for a superseded generation is dead.
        if gen != self.gen:
            print(f"[async] dropped: computed for gen {gen}, now on {self.gen}")
            return
        # GATE 2 — material?
        if not self.material(finding):
            self.stash.append(finding)
            print("[async] stashed for the next natural turn (no interjection cost)")
            return
        # GATE 3 — did the audio lane already say it?
        if any(finding[:24].lower() in s.lower() for s in self.spoken):
            print("[async] dropped: already spoken")
            return
        # GATE 4 — SINGLE WRITER. Never create a second response while one is
        # open. Queue it; the read loop drains at response.done.
        if self.turn_open:
            self.pending.append(finding)
            print("[async] queued until the turn closes")
            return
        await self.interject(finding)

    async def interject(self, finding):
        # Never speak from this lane. Hand the fact to the voice that is already
        # in the room. Cost: one full extra turn (context in, tokens out, TTS).
        await self.ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {"type": "message", "role": "system", "content": [{
                "type": "input_text",
                "text": (f"A background check finished: {finding} "
                         "Correct what you told the user, in one short sentence.")}]}}))
        await self.ws.send(json.dumps({"type": "response.create",
                                       "response": {"model": "auto"}}))
        self.turn_open = True
        print("[async] interjected")

async def main():
    async with websockets.connect(
            WS, additional_headers={"Authorization": f"Bearer {KEY}"},
            max_size=16 * 1024 * 1024) as ws:
        assert json.loads(await ws.recv())["type"] == "session.created"
        lane = AsyncLane(ws)

        # A text turn, so this example runs with no microphone attached.
        lane.gen += 1
        await ws.send(json.dumps({"type": "conversation.item.create",
            "item": {"type": "message", "role": "user",
                     "content": [{"type": "input_text",
                                  "text": "When does the Fulton order ship?"}]}}))
        await ws.send(json.dumps({"type": "response.create"}))
        lane.turn_open = True

        # Fire the slow job NOW, concurrently. It is stamped with this gen.
        job = asyncio.create_task(lane.job("Does the Fulton order ship Tuesday?", lane.gen))

        started = time.monotonic()
        buf = ""
        async for raw in ws:
            ev = json.loads(raw)
            if ev["type"] == "response.text.delta":
                buf += ev.get("delta", "")
                print(ev.get("delta", ""), end="", flush=True)
            elif ev["type"] == "response.done":
                lane.spoken.append(buf); buf = ""
                lane.turn_open = False
                print("\n[turn closed at", round(time.monotonic() - started, 2), "s]")
                if lane.pending:              # now it is safe to be the writer
                    await lane.interject(lane.pending.pop(0))
                    continue
                if job.done():
                    break
            elif ev["type"] == "error":
                print("\nerror:", ev["error"]["code"], ev["error"]["message"])
                break
        await job

asyncio.run(main())

TypeScript / Node

// npm i ws   —   Node 20+, global fetch for the slow job.
import WebSocket from "ws";

const KEY = process.env.HAWKTALK_API_KEY!;
const ws = new WebSocket("wss://api.hawktalk.ai/v1/realtime?model=auto", {
  headers: { Authorization: `Bearer ${KEY}` }, maxPayload: 16 * 1024 * 1024,
});
// 401 / 429 / 503 arrive on the upgrade; an unhandled "error" kills the process.
ws.on("unexpected-response", (_req, res) =>
  console.error("upgrade failed: HTTP", res.statusCode,
                "retry-after=", res.headers["retry-after"] ?? "n/a"));
ws.on("error", (e) => console.error("socket error:", e.message));

const send = (o: unknown) => ws.send(JSON.stringify(o));

let gen = 0;                 // bumped by YOUR VAD on every commit
let turnOpen = false;        // SINGLE WRITER guard
const spoken: string[] = []; // what the audio lane has already said
const stash: string[] = [];  // immaterial findings — folded in later, free
const pending: string[] = [];// material findings waiting for response.done

// "Changes a date, a number, or an action." Your domain, your rule.
const material = (f: string) =>
  ["no —", "thursday", "cancel", "held"].some((k) => f.toLowerCase().includes(k));

async function slowJob(question: string, startedAtGen: number) {
  // tier:think off the hot path. It is slow and pricey; that is WHY it is here.
  const r = await fetch("https://api.hawktalk.ai/v1/chat/completions", {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({ model: "tier:think", max_tokens: 400, messages: [
      { role: "system", content: "Verify against the supplier feed. One sentence." },
      { role: "user", content: question }] }),
  });
  if (!r.ok) throw new Error(`async job failed: ${r.status} ${await r.text()}`);
  const finding = (await r.json()).choices[0].message.content as string;
  land(finding, startedAtGen);
}

function land(finding: string, startedAtGen: number) {
  if (startedAtGen !== gen)                       // GATE 1 — fresh?
    return console.log(`[async] dropped: gen ${startedAtGen}, now ${gen}`);
  if (!material(finding)) {                       // GATE 2 — material?
    stash.push(finding);
    return console.log("[async] stashed — costs nothing, said next turn");
  }
  if (spoken.some((s) => s.toLowerCase().includes(finding.slice(0, 24).toLowerCase())))
    return console.log("[async] dropped: already spoken");   // GATE 3
  if (turnOpen) {                                 // GATE 4 — single writer
    pending.push(finding);
    return console.log("[async] queued until the turn closes");
  }
  interject(finding);
}

function interject(finding: string) {
  // Hand the fact to the voice already in the room. One extra full turn.
  send({ type: "conversation.item.create", item: {
    type: "message", role: "system",
    content: [{ type: "input_text",
      text: `A background check finished: ${finding} Correct what you told the user, in one short sentence.` }],
  }});
  send({ type: "response.create", response: { model: "auto" } });
  turnOpen = true;
  console.log("[async] interjected");
}

let buf = "";
ws.on("open", () => {
  gen += 1;
  send({ type: "conversation.item.create", item: { type: "message", role: "user",
    content: [{ type: "input_text", text: "When does the Fulton order ship?" }] }});
  send({ type: "response.create" });
  turnOpen = true;
  // Concurrent from the first millisecond — never awaited by the turn.
  slowJob("Does the Fulton order ship Tuesday?", gen).catch((e) => console.error(e));
});

ws.on("message", (raw) => {
  const ev = JSON.parse(raw.toString());
  if (ev.type === "response.text.delta") { buf += ev.delta ?? ""; process.stdout.write(ev.delta ?? ""); }
  else if (ev.type === "response.done") {
    spoken.push(buf); buf = ""; turnOpen = false;
    console.log("\n[turn closed]");
    const next = pending.shift();
    if (next) interject(next);        // safe now: no response is open
  }
  else if (ev.type === "error") { console.error(ev.error.code, ev.error.message); ws.close(); }
});

Dart

// pubspec.yaml: web_socket_channel: ^3.0.0, http: ^1.2.0
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:http/http.dart' as http;
import 'package:web_socket_channel/io.dart';

final _key = Platform.environment['HAWKTALK_API_KEY']!;
final _base = Platform.environment['HAWKTALK_BASE'] ?? 'https://api.hawktalk.ai';
final _wsUri = Uri.parse('wss://api.hawktalk.ai/v1/realtime?model=auto');

final class AsyncLane {
  final void Function(Map<String, dynamic>) send;
  final http.Client httpClient;
  int gen = 0; // bumped by YOUR VAD on every commit
  bool turnOpen = false;
  final List<String> spoken = []; // what the audio lane has already said
  final List<String> stash = []; // immaterial findings, for the next natural turn
  final List<String> pending = []; // material findings waiting for the turn to close

  AsyncLane({required this.send, required this.httpClient});

  // "Changes a date, a number, or an action." Your domain, your rule.
  bool material(String finding) {
    final lower = finding.toLowerCase();
    return const ['no —', 'thursday', 'cancel', 'held']
        .any((k) => lower.contains(k));
  }

  Future<void> job(String question, int startedAtGen) async {
    // tier:think off the hot path. It is slow and pricey; that is WHY it is here.
    final res = await httpClient.post(
      Uri.parse('$_base/v1/chat/completions'),
      headers: {
        'Authorization': 'Bearer $_key',
        'Content-Type': 'application/json',
      },
      body: jsonEncode({
        'model': 'tier:think',
        'max_tokens': 400,
        'messages': [
          {
            'role': 'system',
            'content': 'Verify against the supplier feed. One sentence.',
          },
          {'role': 'user', 'content': question},
        ],
      }),
    );

    if (res.statusCode != 200) {
      throw HttpException('async job failed: ${res.statusCode} ${res.body}');
    }

    final body = jsonDecode(utf8.decode(res.bodyBytes)) as Map<String, dynamic>;
    final choices = body['choices'] as List<dynamic>;
    final msg = (choices.first as Map<String, dynamic>)['message'] as Map<String, dynamic>;
    final finding = msg['content'] as String;
    await land(finding, startedAtGen);
  }

  Future<void> land(String finding, int startedAtGen) async {
    // GATE 1 — fresh? A finding for a superseded generation is dead.
    if (startedAtGen != gen) {
      print('[async] dropped: computed for gen $startedAtGen, now on $gen');
      return;
    }
    // GATE 2 — material?
    if (!material(finding)) {
      stash.add(finding);
      print('[async] stashed for the next natural turn (no interjection cost)');
      return;
    }
    // GATE 3 — did the audio lane already say it?
    final prefix = finding.length > 24 ? finding.substring(0, 24).toLowerCase() : finding.toLowerCase();
    if (spoken.any((s) => s.toLowerCase().contains(prefix))) {
      print('[async] dropped: already spoken');
      return;
    }
    // GATE 4 — SINGLE WRITER. Never create a second response while one is open.
    if (turnOpen) {
      pending.add(finding);
      print('[async] queued until the turn closes');
      return;
    }
    await interject(finding);
  }

  Future<void> interject(String finding) async {
    // Never speak from this lane. Hand the fact to the voice that is already
    // in the room. Cost: one full extra turn (context in, tokens out, TTS).
    send({
      'type': 'conversation.item.create',
      'item': {
        'type': 'message',
        'role': 'system',
        'content': [
          {
            'type': 'input_text',
            'text': 'A background check finished: $finding '
                'Correct what you told the user, in one short sentence.',
          },
        ],
      },
    });
    send({
      'type': 'response.create',
      'response': {'model': 'auto'},
    });
    turnOpen = true;
    print('[async] interjected');
  }
}

Future<void> main() async {
  final channel = IOWebSocketChannel.connect(
    _wsUri,
    headers: {'Authorization': 'Bearer $_key'},
  );
  final httpClient = http.Client();

  void send(Map<String, dynamic> o) => channel.sink.add(jsonEncode(o));
  final lane = AsyncLane(send: send, httpClient: httpClient);

  var buf = '';
  Future<void>? bgJob;

  await for (final raw in channel.stream) {
    final ev = jsonDecode(raw as String) as Map<String, dynamic>;
    final t = ev['type'] as String?;

    if (t == 'session.created') {
      // A text turn, so this example runs with no microphone attached.
      lane.gen += 1;
      send({
        'type': 'conversation.item.create',
        'item': {
          'type': 'message',
          'role': 'user',
          'content': [
            {'type': 'input_text', 'text': 'When does the Fulton order ship?'},
          ],
        },
      });
      send({'type': 'response.create'});
      lane.turnOpen = true;

      // Fire the slow job NOW, concurrently. It is stamped with this gen.
      bgJob = lane.job('Does the Fulton order ship Tuesday?', lane.gen);
    } else if (t == 'response.text.delta') {
      final delta = ev['delta'] as String? ?? '';
      buf += delta;
      stdout.write(delta);
    } else if (t == 'response.done') {
      lane.spoken.add(buf);
      buf = '';
      lane.turnOpen = false;
      print('\n[turn closed]');

      if (lane.pending.isNotEmpty) {
        // now it is safe to be the writer
        await lane.interject(lane.pending.removeAt(0));
        continue;
      }
      break;
    } else if (t == 'error') {
      final err = ev['error'] as Map<String, dynamic>?;
      stderr.writeln('\nerror: ${err?["code"]} ${err?["message"]}');
      break;
    }
  }

  if (bgJob != null) await bgJob;
  await channel.sink.close();
  httpClient.close();
}

This is the part everyone skips, and it is the part that decides whether you have a superharness or four race conditions in a trenchcoat. The moment more than one lane can produce a claim about the world, two of them will disagree, and the disagreement will arrive while the first one is still being spoken aloud. If you have not decided the precedence up front, in code, the winner is whichever coroutine happened to be scheduled first — which means your product's behaviour is not designed, it is emergent.

Three primitives make this tractable, and all three are already on the wire:

Primitive	What it gives you
gen (mux) / your own counter	Freshness. Every claim is stamped with the turn it belongs to; an old stamp is dead work.
response.cancel	A clean undo. It rolls conversation history back to the turn boundary, so after a cancel the model's memory matches what the person actually heard. This is what makes "cancel and re-answer" safe rather than a way to desynchronise the model from reality.
response.done	The turn boundary itself. Before it, the turn is yours to withdraw; after it, the words are history.

The one rule you cannot break: spoken audio is never retracted. Once bytes have gone to the speaker the person has heard them, and no protocol event un-hears them. You correct; you do not pretend. That single asymmetry — the audio lane is authoritative about the past, the async lane may be authoritative about the facts — is the whole policy in one sentence.

# THE DECISION MATRIX. Decide this once, write it down, put it in code.

                                | turn OPEN,        | turn OPEN,        | turn
                                | nothing spoken    | audio flowing     | CLOSED
--------------------------------+-------------------+-------------------+---------------
async AGREES with what was said | drop              | drop              | drop
async CONTRADICTS, material     | response.cancel   | queue, then       | interject
                                | + re-answer       | interject after   | now
                                |                   | response.done     |
async CONTRADICTS, immaterial   | let it finish     | let it finish     | stash for the
                                |                   |                   | next turn
stamped with an older gen       | drop              | drop              | drop
lower-ranked source than the    | drop              | drop              | drop
claim already on the record     |                   |                   |

# SOURCE RANK — declare it before you need it. Highest wins within one gen.
system_of_record  >  retrieval_with_citation  >  tool_result  >  model_opinion

# SINGLE WRITER — exactly one component may call response.create. Two lanes
# both creating a response on one socket is the bug you will spend a week on.

Python

# The reconciler owns the socket. Lanes SUBMIT claims; they never send.
# Drop this next to the async lane from the previous section.
import json
from dataclasses import dataclass, field
from enum import IntEnum

class Rank(IntEnum):
    MODEL_OPINION = 0
    TOOL_RESULT = 1
    RETRIEVAL = 2
    SYSTEM_OF_RECORD = 3

@dataclass
class Claim:
    text: str
    gen: int
    rank: Rank
    material: bool
    contradicts: bool     # your comparator decided this against `spoken`

@dataclass
class Reconciler:
    ws: object
    gen: int = 0
    turn_open: bool = False
    audio_started: bool = False       # flips on the first response.audio.delta
    spoken: list = field(default_factory=list)
    record_rank: Rank = Rank.MODEL_OPINION
    queued: list = field(default_factory=list)
    stash: list = field(default_factory=list)
    log: list = field(default_factory=list)

    # ---- the only two places that touch the socket ----
    async def _send(self, obj):
        await self.ws.send(json.dumps(obj))

    async def _speak_correction(self, claim: Claim):
        await self._send({"type": "conversation.item.create",
            "item": {"type": "message", "role": "system", "content": [{
                "type": "input_text",
                "text": (f"Correction, from a more authoritative source: {claim.text} "
                         "Say the correction in one short sentence. Do not apologise twice.")}]}})
        # One extra turn of tokens + TTS. The materiality gate already paid for it.
        await self._send({"type": "response.create", "response": {"model": "auto"}})
        self.turn_open = True
        self.record_rank = claim.rank

    # ---- the policy ----
    async def submit(self, claim: Claim):
        d = await self._decide(claim)
        self.log.append({"gen": claim.gen, "rank": int(claim.rank), "decision": d})
        print(f"[reconcile] gen={claim.gen} rank={claim.rank.name} -> {d}")

    async def _decide(self, claim: Claim) -> str:
        if claim.gen != self.gen:
            return "drop: stale generation"
        if not claim.contradicts:
            return "drop: agrees with the record"
        if claim.rank <= self.record_rank:
            # A model opinion never overturns a system of record. Declaring this
            # up front is what stops two lanes ping-ponging corrections.
            return "drop: source outranked"
        if not claim.material:
            self.stash.append(claim)
            return "stash: immaterial, fold into the next natural turn"

        if self.turn_open and not self.audio_started:
            # Nothing has been heard yet: withdraw cleanly. response.cancel rolls
            # conversation history back to the turn boundary, so the model does not
            # believe it said words the person never heard.
            await self._send({"type": "response.cancel"})
            self.turn_open = False
            await self._speak_correction(claim)
            return "cancel + re-answer"

        if self.turn_open and self.audio_started:
            # Audio is in the air. NEVER retract it — the person heard it.
            self.queued.append(claim)
            return "queued: interject after response.done"

        await self._speak_correction(claim)
        return "interject now"

    # ---- wire it to the socket's events ----
    async def on_event(self, ev):
        t = ev.get("type")
        if t == "response.created":
            self.turn_open, self.audio_started = True, False
        elif t == "response.audio.delta":
            self.audio_started = True          # the point of no return
        elif t == "response.text.done":
            self.spoken.append(ev.get("text", ""))
        elif t == "response.done":
            self.turn_open, self.audio_started = False, False
            # Drain ONE claim per turn boundary: _speak_correction opens a new
            # response, and only one may be open at a time.
            while self.queued:
                claim = self.queued.pop(0)
                if claim.gen == self.gen:      # re-check: the user may have spoken
                    await self._speak_correction(claim)
                    break

    def new_utterance(self):
        # Call this where you send input_audio_buffer.commit. Everything computed
        # for the previous generation is now dead, including queued corrections.
        self.gen += 1
        self.queued.clear()
        self.record_rank = Rank.MODEL_OPINION
        return self.gen

TypeScript / Node

// The reconciler is the SINGLE WRITER. Lanes submit; only this sends.
export enum Rank { ModelOpinion = 0, ToolResult = 1, Retrieval = 2, SystemOfRecord = 3 }

export type Claim = {
  text: string;
  gen: number;
  rank: Rank;
  material: boolean;      // changes a date, a number, or an action
  contradicts: boolean;   // your comparator, against what was spoken
};

export class Reconciler {
  gen = 0;
  private turnOpen = false;
  private audioStarted = false;     // true from the first response.audio.delta
  private recordRank = Rank.ModelOpinion;
  private queued: Claim[] = [];
  readonly spoken: string[] = [];
  readonly stash: Claim[] = [];
  readonly log: { gen: number; rank: Rank; decision: string }[] = [];

  constructor(private send: (o: unknown) => void) {}

  /// Call where you send input_audio_buffer.commit.
  newUtterance(): number {
    this.gen += 1;
    this.queued = [];            // corrections for a dead turn are dead
    this.recordRank = Rank.ModelOpinion;
    return this.gen;
  }

  submit(c: Claim) {
    const decision = this.decide(c);
    this.log.push({ gen: c.gen, rank: c.rank, decision });
    console.log(`[reconcile] gen=${c.gen} rank=${Rank[c.rank]} -> ${decision}`);
  }

  private decide(c: Claim): string {
    if (c.gen !== this.gen) return "drop: stale generation";
    if (!c.contradicts) return "drop: agrees with the record";
    if (c.rank <= this.recordRank) return "drop: source outranked";
    if (!c.material) { this.stash.push(c); return "stash: immaterial"; }

    if (this.turnOpen && !this.audioStarted) {
      // Nothing heard yet. response.cancel rolls conversation history back to
      // the turn boundary — the model will not think it said the old answer.
      this.send({ type: "response.cancel" });
      this.turnOpen = false;
      this.correct(c);
      return "cancel + re-answer";
    }
    if (this.turnOpen && this.audioStarted) {
      // The person is hearing it right now. Spoken audio is never retracted.
      this.queued.push(c);
      return "queued: interject after response.done";
    }
    this.correct(c);
    return "interject now";
  }

  private correct(c: Claim) {
    this.send({ type: "conversation.item.create", item: {
      type: "message", role: "system",
      content: [{ type: "input_text",
        text: `Correction, from a more authoritative source: ${c.text} ` +
              `Say the correction in one short sentence. Do not apologise twice.` }],
    }});
    // Costs a full extra turn — the materiality gate is what pays for it.
    this.send({ type: "response.create", response: { model: "auto" } });
    this.turnOpen = true;
    this.recordRank = c.rank;
  }

  onEvent(ev: any) {
    switch (ev.type) {
      case "response.created": this.turnOpen = true; this.audioStarted = false; break;
      case "response.audio.delta": this.audioStarted = true; break;   // no going back
      case "response.text.done": this.spoken.push(ev.text ?? ""); break;
      case "response.done": {
        this.turnOpen = false; this.audioStarted = false;
        // One correction per turn boundary — correct() opens a new response.
        while (this.queued.length) {
          const c = this.queued.shift()!;
          if (c.gen === this.gen) { this.correct(c); break; }   // re-check gen
        }
        break;
      }
    }
  }
}

Dart

// pubspec.yaml: web_socket_channel: ^3.0.0
// The reconciler owns the socket. Lanes SUBMIT claims; they never send.
import 'dart:convert';

enum Rank implements Comparable<Rank> {
  modelOpinion(0),
  toolResult(1),
  retrieval(2),
  systemOfRecord(3);

  final int value;
  const Rank(this.value);

  @override
  int compareTo(Rank other) => value.compareTo(other.value);
}

final class Claim {
  final String text;
  final int gen;
  final Rank rank;
  final bool material;
  final bool contradicts; // your comparator decided this against `spoken`

  const Claim({
    required this.text,
    required this.gen,
    required this.rank,
    required this.material,
    required this.contradicts,
  });
}

final class Reconciler {
  final void Function(Map<String, dynamic>) send;
  int gen = 0;
  bool turnOpen = false;
  bool audioStarted = false; // flips on the first response.audio.delta
  Rank recordRank = Rank.modelOpinion;
  final List<Claim> queued = [];
  final List<String> spoken = [];
  final List<Claim> stash = [];
  final List<Map<String, dynamic>> log = [];

  Reconciler(this.send);

  /// Call this where you send input_audio_buffer.commit. Everything computed
  /// for the previous generation is now dead, including queued corrections.
  int newUtterance() {
    gen += 1;
    queued.clear();
    recordRank = Rank.modelOpinion;
    return gen;
  }

  Future<void> submit(Claim claim) async {
    final decision = await _decide(claim);
    log.add({'gen': claim.gen, 'rank': claim.rank.value, 'decision': decision});
    print('[reconcile] gen=${claim.gen} rank=${claim.rank.name} -> $decision');
  }

  Future<String> _decide(Claim claim) async {
    if (claim.gen != gen) return 'drop: stale generation';
    if (!claim.contradicts) return 'drop: agrees with the record';
    if (claim.rank.value <= recordRank.value) {
      // A model opinion never overturns a system of record. Declaring this
      // up front is what stops two lanes ping-ponging corrections.
      return 'drop: source outranked';
    }
    if (!claim.material) {
      stash.add(claim);
      return 'stash: immaterial, fold into the next natural turn';
    }

    if (turnOpen && !audioStarted) {
      // Nothing has been heard yet: withdraw cleanly. response.cancel rolls
      // conversation history back to the turn boundary, so the model does not
      // believe it said words the person never heard.
      send({'type': 'response.cancel'});
      turnOpen = false;
      await _speakCorrection(claim);
      return 'cancel + re-answer';
    }

    if (turnOpen && audioStarted) {
      // Audio is in the air. NEVER retract it — the person heard it.
      queued.add(claim);
      return 'queued: interject after response.done';
    }

    await _speakCorrection(claim);
    return 'interject now';
  }

  Future<void> _speakCorrection(Claim claim) async {
    send({
      'type': 'conversation.item.create',
      'item': {
        'type': 'message',
        'role': 'system',
        'content': [
          {
            'type': 'input_text',
            'text':
                'Correction, from a more authoritative source: ${claim.text} '
                'Say the correction in one short sentence. Do not apologise twice.',
          },
        ],
      },
    });
    // One extra turn of tokens + TTS. The materiality gate already paid for it.
    send({
      'type': 'response.create',
      'response': {'model': 'auto'},
    });
    turnOpen = true;
    recordRank = claim.rank;
  }

  Future<void> onEvent(Map<String, dynamic> ev) async {
    final t = ev['type'] as String?;
    switch (t) {
      case 'response.created':
        turnOpen = true;
        audioStarted = false;
      case 'response.audio.delta':
        audioStarted = true; // the point of no return
      case 'response.text.done':
        spoken.add((ev['text'] as String?) ?? '');
      case 'response.done':
        turnOpen = false;
        audioStarted = false;
        // Drain ONE claim per turn boundary: _speakCorrection opens a new
        // response, and only one may be open at a time.
        while (queued.isNotEmpty) {
          final claim = queued.removeAt(0);
          if (claim.gen == gen) {
            // re-check: the user may have spoken
            await _speakCorrection(claim);
            break;
          }
        }
    }
  }
}

A session has a lifecycle: it is provisioned for a task, warmed, worked, drained, closed, and settled. Treating it as "the WebSocket I opened at boot and hope stays up" is the second most expensive mistake on this product after pinning a large tier — a standing session is metered while it is open, and every session contributes a line to the usage ledger.

note: on the preview mux each connection is one billable session and the lane state (gen, buffered PCM, presence, affect) lives and dies with it — there is no lane-level reattach. On the shipped /v1/realtime the same is true of conversation history: the session is the history. Which is also why you must never hand one session to two users.

Phase	What happens	What you must do
provision	Pick the tier for the task.	GET /v1/models once per deploy, not per session. Then pass auto unless you measured a reason not to.
open	The upgrade succeeds or fails with an HTTP status.	401, 429 (with Retry-After) and 503 arrive here, not as frames. Back off on 429; degrade tier on 503.
warm	session.created, then your session.update.	Send instructions and tools once. The first turn pays the cold cost — warm before the person speaks, not after.
work	Turns. Metered while open.	Watch ouroboros.telemetry: it arrives every 8 tokens and as a 5 s idle heartbeat, so a gap of >15 s means the session is gone even if TCP has not noticed. The check must run on a timer, not on frame arrival — a dead socket delivers no frames.
drain	The last turn finishes.	If a response is still open, send response.cancel first — it rolls history back to the turn boundary and leaves nothing half-said.
close	WebSocket close, code 1000.	Close explicitly. An abandoned socket is still an open session.
settle	You reconcile what it cost.	Record the last rate_limits.updated and each turn's x_ouroboros. Nulls stay null in your metrics — a not-measured latency is not a fast one.

curl — provision once, not per session

# Cache this per deploy. Hitting the registry on every session start adds a
# round trip to the very latency you are trying to protect.
curl -sS https://api.hawktalk.ai/v1/models \
  -H "Authorization: Bearer $HAWKTALK_API_KEY"

# The registry carries availability + routing info per model. Use it to decide
# WHETHER a pin is possible; use "auto" to decide WHICH tier per utterance.
# Tier vocabulary: ouro | quick | dank | think | cloud
#   (aliases: self, route, ouromega, live, specialist, t0-t3)

# Before you promise a customer voice, ask the node what it can actually do.
# /health needs no auth and reports auth state, models, voice seams, telemetry.
curl -sS https://api.hawktalk.ai/health

# If a seam is not wired, the REST voice endpoints say so honestly:
curl -sS -o /dev/null -w "%{http_code}\n" \
  https://api.hawktalk.ai/v1/audio/speech \
  -H "Authorization: Bearer $HAWKTALK_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"auto","input":"lifecycle check"}'
# 200 -> real audio/wav bytes.  501 -> tts_not_wired on this node: fall back to
# browser Web Speech. Never synthesise a substitute and never invent a transcript.
# Voice ids are node-specific — read them off GET /health rather than guessing
# a name; an unknown voice is a 400 invalid_request, not a fallback.

Python

# pip install "websockets>=14"
# An ephemeral session as a context manager: it cannot outlive its task.
# NOTE: on websockets<14 the failed-handshake exception is InvalidStatusCode
# with .status_code, and the header kwarg is extra_headers=. Same semantics.
import asyncio, contextlib, json, os, time
import websockets

KEY = os.environ["HAWKTALK_API_KEY"]
URL = "wss://api.hawktalk.ai/v1/realtime?model=auto"
HEARTBEAT_GAP_S = 15.0     # telemetry idles at 5 s; 3 misses = gone
POLL_S = 2.0               # how often the watchdog gets to run

class EphemeralSession:
    """Provision -> open -> warm -> work -> drain -> close -> settle."""

    def __init__(self, instructions, tools=None):
        self.instructions = instructions
        self.tools = tools or []
        self.ws = None
        self.turn_open = False
        self.last_telemetry = None
        self.rate_limits = None
        self.turns = []            # one x_ouroboros per turn, for settle()

    async def __aenter__(self):
        # OPEN. 401/429/503 land on the HTTP upgrade, not as frames.
        try:
            self.ws = await websockets.connect(
                URL, additional_headers={"Authorization": f"Bearer {KEY}"},
                max_size=16 * 1024 * 1024, open_timeout=10)
        except websockets.InvalidStatus as e:
            code = e.response.status_code
            if code == 429:
                # Retry-After is authoritative. Do not invent your own backoff.
                raise RuntimeError(
                    f"rate_limit_exceeded; retry after {e.response.headers.get('Retry-After')}s")
            if code == 503:
                raise RuntimeError("model_unavailable — degrade a tier and retry")
            raise RuntimeError(f"open failed: HTTP {code}")

        hello = json.loads(await self.ws.recv())
        assert hello["type"] == "session.created", hello

        # WARM. Do this before the person speaks — the first turn pays the
        # cold cost, and paying it during a silence is free. Tools go in ONCE.
        await self.ws.send(json.dumps({"type": "session.update", "session": {
            "model": "auto", "modalities": ["text", "audio"],
            "instructions": self.instructions,
            "tools": self.tools, "tool_choice": "auto"}}))
        self.last_telemetry = time.monotonic()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        # DRAIN. A half-open response is rolled back to the turn boundary, so
        # nothing is left half-said in the model's memory.
        if self.turn_open:
            with contextlib.suppress(Exception):
                await self.ws.send(json.dumps({"type": "response.cancel"}))
        # CLOSE. Explicitly — an abandoned socket is still a metered session.
        with contextlib.suppress(Exception):
            await self.ws.close(code=1000, reason="task complete")
        self.settle()

    def settle(self):
        # SETTLE. Nulls stay null: a latency that was not measured is not 0.
        print("[settle] turns=", len(self.turns))
        for i, x in enumerate(self.turns):
            print(f"  turn {i}: ttft_ms={x.get('ttft_ms')} "
                  f"e2e_ms={x.get('end_to_end_ms')} "
                  f"e2e_audio_ms={x.get('end_to_end_audio_ms')}")
        print("[settle] last rate_limits =", self.rate_limits)

    async def ask(self, text):
        """One turn. Returns the assistant text."""
        await self.ws.send(json.dumps({"type": "conversation.item.create",
            "item": {"type": "message", "role": "user",
                     "content": [{"type": "input_text", "text": text}]}}))
        await self.ws.send(json.dumps({"type": "response.create"}))
        self.turn_open = True
        out = ""
        while True:
            # The watchdog has to be able to fire when NOTHING arrives — that is
            # precisely the case it exists for. Bound the recv; never park in a
            # plain `async for` and check liveness at the bottom of the body.
            try:
                raw = await asyncio.wait_for(self.ws.recv(), timeout=POLL_S)
            except asyncio.TimeoutError:
                if time.monotonic() - self.last_telemetry > HEARTBEAT_GAP_S:
                    raise RuntimeError("no telemetry for 15 s — session is dead, reconnect")
                continue
            ev = json.loads(raw)
            t = ev.get("type")
            if t == "ouroboros.telemetry":
                # every 8 tokens, plus a 5 s idle heartbeat = your liveness signal
                self.last_telemetry = time.monotonic()
            elif t == "rate_limits.updated":
                self.rate_limits = ev.get("rate_limits")
            elif t == "response.text.delta":
                out += ev.get("delta", "")
            elif t == "error":
                raise RuntimeError(f"{ev['error']['code']}: {ev['error']['message']}")
            elif t == "response.done":
                self.turn_open = False
                # root OR under "response", depending on the node — read both
                self.turns.append(ev.get("x_ouroboros")
                                  or (ev.get("response") or {}).get("x_ouroboros")
                                  or {})
                return out

async def main():
    # Spun up for ONE task, torn down after. Never shared between people:
    # the session IS the conversation history.
    async with EphemeralSession("Summarise in one sentence.") as s:
        print(await s.ask("What is on the Fulton order?"))

asyncio.run(main())

TypeScript / Node

// npm i ws — ESM ("type": "module") for the top-level await at the bottom.
// An ephemeral session with an explicit lifecycle and a
// finally-block teardown, so it cannot be leaked by an early throw.
import WebSocket from "ws";

const KEY = process.env.HAWKTALK_API_KEY!;
const HEARTBEAT_GAP_MS = 15_000;   // telemetry idles at 5 s; 3 misses = dead

export class EphemeralSession {
  private ws!: WebSocket;
  private turnOpen = false;
  private lastTelemetry = 0;
  private rateLimits: unknown = null;
  private readonly turns: any[] = [];

  constructor(private instructions: string, private tools: unknown[] = []) {}

  async open(): Promise<void> {
    this.ws = new WebSocket("wss://api.hawktalk.ai/v1/realtime?model=auto", {
      headers: { Authorization: `Bearer ${KEY}` }, maxPayload: 16 * 1024 * 1024,
    });
    await new Promise<void>((resolve, reject) => {
      // 401 / 429 / 503 arrive on the HTTP upgrade, before any frame exists.
      this.ws.once("unexpected-response", (_req, res) => {
        const hint = res.statusCode === 429
          ? `rate_limit_exceeded; Retry-After=${res.headers["retry-after"]}`
          : res.statusCode === 503 ? "model_unavailable — degrade a tier" : "";
        reject(new Error(`open failed: HTTP ${res.statusCode} ${hint}`));
      });
      this.ws.once("error", reject);
      this.ws.once("open", () => resolve());
    });
    // A socket with no "error" listener throws on the process. Keep one for
    // the whole life of the session, not just the handshake.
    this.ws.on("error", (e) => console.error("socket error:", e.message));

    // WARM before the person speaks: the first turn pays the cold cost.
    // Instructions and tools once per session, never per turn.
    this.send({ type: "session.update", session: {
      model: "auto", modalities: ["text", "audio"],
      instructions: this.instructions, tools: this.tools, tool_choice: "auto" }});
    this.lastTelemetry = Date.now();
  }

  private send(o: unknown) { this.ws.send(JSON.stringify(o)); }

  async ask(text: string): Promise<string> {
    this.send({ type: "conversation.item.create", item: { type: "message",
      role: "user", content: [{ type: "input_text", text }] }});
    this.send({ type: "response.create" });
    this.turnOpen = true;

    return await new Promise<string>((resolve, reject) => {
      let out = "";
      let watchdog: ReturnType<typeof setInterval>;
      const onMessage = (raw: WebSocket.RawData) => {
        const ev = JSON.parse(raw.toString());
        switch (ev.type) {
          case "ouroboros.telemetry": this.lastTelemetry = Date.now(); break;
          case "rate_limits.updated": this.rateLimits = ev.rate_limits; break;
          case "response.text.delta": out += ev.delta ?? ""; break;
          case "error":
            cleanup();
            reject(new Error(`${ev.error.code}: ${ev.error.message}`)); break;
          case "response.done":
            this.turnOpen = false;
            // root OR under response, depending on the node — read both
            this.turns.push(ev.x_ouroboros ?? ev.response?.x_ouroboros ?? {});
            cleanup();
            resolve(out); break;
        }
      };
      // Every exit path clears the interval AND unhooks the listener —
      // otherwise the timer keeps the process alive forever and each turn
      // leaks another handler onto the socket.
      const cleanup = () => { clearInterval(watchdog); this.ws.off("message", onMessage); };
      watchdog = setInterval(() => {
        if (Date.now() - this.lastTelemetry > HEARTBEAT_GAP_MS) {
          cleanup();
          reject(new Error("no telemetry for 15 s — session is dead, reconnect"));
        }
      }, 2000);
      this.ws.on("message", onMessage);
    });
  }

  async close() {
    // DRAIN: roll a half-finished response back to the turn boundary.
    if (this.turnOpen) this.send({ type: "response.cancel" });
    this.ws.close(1000, "task complete");        // metered until this happens
    // SETTLE: nulls stay null. An unmeasured latency is not a fast one.
    for (const [i, x] of this.turns.entries())
      console.log(`turn ${i}: ttft_ms=${x.ttft_ms} e2e_ms=${x.end_to_end_ms}`);
    console.log("last rate_limits =", this.rateLimits);
  }
}

const s = new EphemeralSession("Summarise in one sentence.");
await s.open();
try {
  console.log(await s.ask("What is on the Fulton order?"));
} finally {
  await s.close();   // one task, one session. Never shared between people.
}

Dart

// pubspec.yaml: web_socket_channel: ^3.0.0
// An ephemeral session with an explicit lifecycle and a
// try/finally teardown, so it cannot be leaked by an unhandled exception.
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:web_socket_channel/io.dart';

final _key = Platform.environment['HAWKTALK_API_KEY']!;
final _url = Uri.parse('wss://api.hawktalk.ai/v1/realtime?model=auto');

const heartbeatGap = Duration(seconds: 15); // telemetry idles at 5 s; 3 misses = dead
const pollInterval = Duration(seconds: 2);

final class EphemeralSession {
  final String instructions;
  final List<dynamic> tools;
  late final IOWebSocketChannel _channel;
  late final StreamIterator<dynamic> _iterator;
  bool _turnOpen = false;
  DateTime _lastTelemetry = DateTime.now();
  dynamic rateLimits;
  final List<Map<String, dynamic>> turns = [];

  EphemeralSession(this.instructions, [this.tools = const []]);

  void _send(Object o) => _channel.sink.add(jsonEncode(o));

  Future<void> open() async {
    // OPEN. 401 / 429 / 503 arrive on the HTTP upgrade, not as frames.
    try {
      _channel = IOWebSocketChannel.connect(
        _url,
        headers: {'Authorization': 'Bearer $_key'},
      );
    } catch (e) {
      throw HttpException('open failed: $e');
    }

    _iterator = StreamIterator<dynamic>(_channel.stream);

    if (!await _iterator.moveNext()) {
      throw const HttpException('socket closed before session.created');
    }
    final hello = jsonDecode(_iterator.current as String) as Map<String, dynamic>;
    if (hello['type'] != 'session.created') {
      throw StateError('expected session.created, got ${hello["type"]}');
    }

    // WARM before the person speaks: the first turn pays the cold cost.
    // Instructions and tools once per session, never per turn.
    _send({
      'type': 'session.update',
      'session': {
        'model': 'auto',
        'modalities': ['text', 'audio'],
        'instructions': instructions,
        'tools': tools,
        'tool_choice': 'auto',
      },
    });
    _lastTelemetry = DateTime.now();
  }

  Future<String> ask(String text) async {
    _send({
      'type': 'conversation.item.create',
      'item': {
        'type': 'message',
        'role': 'user',
        'content': [
          {'type': 'input_text', 'text': text},
        ],
      },
    });
    _send({'type': 'response.create'});
    _turnOpen = true;

    final out = StringBuffer();
    while (true) {
      // The watchdog has to be able to fire when NOTHING arrives — that is
      // precisely the case it exists for. Bound the wait; never block indefinitely.
      final hasNext = await _iterator.moveNext().timeout(
        pollInterval,
        onTimeout: () {
          if (DateTime.now().difference(_lastTelemetry) > heartbeatGap) {
            throw TimeoutException('no telemetry for 15 s — session is dead, reconnect');
          }
          return true; // continue waiting
        },
      );

      if (!hasNext) {
        throw const HttpException('socket closed unexpectedly during turn');
      }

      final ev = jsonDecode(_iterator.current as String) as Map<String, dynamic>;
      final t = ev['type'] as String?;

      switch (t) {
        case 'ouroboros.telemetry':
          // every 8 tokens, plus a 5 s idle heartbeat = your liveness signal
          _lastTelemetry = DateTime.now();

        case 'rate_limits.updated':
          rateLimits = ev['rate_limits'];

        case 'response.text.delta':
          out.write(ev['delta'] ?? '');

        case 'error':
          final err = ev['error'] as Map<String, dynamic>?;
          throw HttpException('${err?["code"]}: ${err?["message"]}');

        case 'response.done':
          _turnOpen = false;
          // root OR under "response", depending on the node — read both
          final response = ev['response'] as Map<String, dynamic>?;
          final x = (ev['x_ouroboros'] ?? response?['x_ouroboros']) as Map<String, dynamic>? ?? const {};
          turns.add(x);
          return out.toString();
      }
    }
  }

  Future<void> close() async {
    // DRAIN: roll a half-finished response back to the turn boundary.
    if (_turnOpen) {
      try {
        _send({'type': 'response.cancel'});
      } catch (_) {}
    }
    // CLOSE: explicitly — an abandoned socket is still a metered session.
    try {
      await _channel.sink.close(1000, 'task complete');
    } catch (_) {}
    settle();
  }

  void settle() {
    // SETTLE: nulls stay null. An unmeasured latency is not a fast one.
    print('[settle] turns=${turns.length}');
    for (var i = 0; i < turns.length; i++) {
      final x = turns[i];
      print('  turn $i: ttft_ms=${x["ttft_ms"]} '
          'e2e_ms=${x["end_to_end_ms"]} '
          'e2e_audio_ms=${x["end_to_end_audio_ms"]}');
    }
    print('[settle] last rate_limits=$rateLimits');
  }
}

Future<void> main() async {
  // Spun up for ONE task, torn down after. Never shared between people:
  // the session IS the conversation history.
  final session = EphemeralSession('Summarise in one sentence.');
  await session.open();
  try {
    final reply = await session.ask('What is on the Fulton order?');
    print(reply);
  } finally {
    await session.close();
  }
}

Where this leaves you. Ship on /v1/realtime today: it is real, it is metered, it has voice and tools, and the thinking, async and reconciliation patterns above run on top of it with nothing but a second REST call and a policy object. Build the lane-mux client if you like — the frames are stable — but do not put /live/brain on a production path, and do not tell your users you have five lanes when the deployed endpoint gives you one. When the mux is promoted past PREVIEW, the send vocabulary you already wrote does not change.