Mode C — full-duplex agent

wss://<host>/ws

{
  "type": "hello",
  "protocol": 1,
  "mac": "AA:BB:CC:DD:EE:FF",
  "session": "b7f3a1c2",
  "property_id": 4,
  "table": 5,
  "firmware_version": "2.1.4",
  "ota_version": "2.1.4",
  "audio":        { "encoding": "pcm_s16le", "sample_rate": 16000,
                    "channels": 1, "frame_ms": 20 },
  "capabilities": { "aec": false, "mic_mute_on_playback": true,
                    "button_interrupt": true, "playback_buffer_ms": 400 }
}

{
  "type": "ready",
  "session_id": "srv-9f2c",
  "property_id": 4,
  "table": 5,
  "currency": "PKR",
  "tts": { "encoding": "pcm_s16le", "sample_rate": 16000,
           "channels": 1, "frame_ms": 40 },
  "tail_ms": 300,
  "heartbeat_interval_s": 20
}

import asyncio, json, websockets

URL  = "wss://api.yourcompany.com/ws"
HELLO = {
    "type": "hello", "protocol": 1,
    "mac": "AA:BB:CC:DD:EE:FF", "session": "b7f3a1c2",
    "property_id": 4, "table": 5,
    "firmware_version": "2.1.4", "ota_version": "2.1.4",
    "audio": {"encoding": "pcm_s16le", "sample_rate": 16000,
              "channels": 1, "frame_ms": 20},
    "capabilities": {"aec": False, "mic_mute_on_playback": True,
                     "button_interrupt": True, "playback_buffer_ms": 400},
}
PERMANENT = {"unknown_device", "unknown_property", "protocol_unsupported"}

async def session(mic, speaker, button):
    async with websockets.connect(URL, ping_interval=None,      # server pings us
                                  max_queue=8) as ws:           # bounded: backpressure
        await ws.send(json.dumps(HELLO))
        ready = json.loads(await ws.recv())
        if ready["type"] == "error":
            if ready["code"] in PERMANENT:
                raise SystemExit(f"permanent: {ready['code']}")  # never retry
            raise ConnectionError(ready["code"])                 # retry w/ backoff

        tail_ms = ready.get("tail_ms", 300)
        muted   = asyncio.Event()

        async def uplink():
            async for frame in mic:                 # 640-byte s16le buffers
                if muted.is_set():
                    continue                        # drop, do NOT send silence
                await ws.send(frame)

        async def downlink():
            async for msg in ws:
                if isinstance(msg, bytes):
                    # Awaiting playback is the pacing mechanism: the receive
                    # window closes behind us and TCP throttles the server.
                    await speaker.play(msg)
                    continue
                evt = json.loads(msg)
                if evt["type"] == "speaking_start":
                    muted.set()
                elif evt["type"] == "speaking_end":
                    await asyncio.sleep(tail_ms / 1000)
                    muted.clear()
                elif evt["type"] == "close":
                    return

        async def barge_in():
            async for press in button:
                if press.long:
                    await ws.send(json.dumps({"type": "bye", "reason": "user_ended"}))
                    return
                if speaker.active:
                    speaker.stop(); speaker.flush()   # locally, first
                    muted.clear()
                    await ws.send(json.dumps({"type": "interrupt"}))

        await asyncio.gather(uplink(), downlink(), barge_in())

// A site's access point reboots and thirty devices reconnect in the same
// second. Each reconnect costs a TLS handshake, a device lookup, a context
// fetch and a recognition socket. Thirty at once can wedge a process.
let attempt = 0;

function nextDelay() {
  const base = Math.min(2 ** attempt, 60) * 1000;   // 1,2,4,8,16,32,60 s cap
  attempt += 1;
  return base * (0.75 + Math.random() * 0.5);       // ±25 % jitter
}

function onConnected() {
  // Reset only after the connection has survived a while — a socket that
  // opens and dies immediately is still a failing connection.
  setTimeout(() => { attempt = 0; }, 60_000);
}