Add authenticated WebSocket realtime event stream

[Justinabox]

Motivation

The roadmap calls for a realtime feed so integrations can react to SMS, delivery reports, call state, signal quality, and USSD responses without polling. Today the HTTP adapter exposes polling-style routes and best-effort webhooks, while the core already has a typed EventBus. A narrow /ws adapter would make Callstack feel like a modern PBX/SMS gateway while preserving the existing service boundaries.

This decomposes the authenticated WebSocket slice from #9 into one implementable PR. It should not invent a new event system or expose raw modem lines; it should adapt sanitized typed events from modem.bus.

User journey

1. A local integration opens an authenticated WebSocket connection to GET /ws.
2. Callstack sends a small hello envelope with protocol/version information.
3. As events occur, the client receives JSON envelopes for completed inbound SMS, SMS delivery reports, call state changes, signal quality updates, modem reconnect/disconnect, and USSD responses.
4. If the client disconnects or falls behind, Callstack cleans up the subscription without affecting modem operation.
5. Operators can rely on the same auth/PII rules used by the HTTP API.

API / UX sketch

Connection:

GET /ws
Authorization: Bearer <api-key>

Initial server message:

{"type":"hello","version":1,"events":["sms.received","sms.delivery_report","call.state","signal.quality","modem.state","ussd.response"]}

Representative event envelope:

{
  "type": "sms.received",
  "timestamp": "2026-06-24T00:00:00Z",
  "data": {
    "sender": "+155****0100",
    "body": "example message"
  }
}

Optional filters can be deferred unless they are trivial. If implemented, keep them explicit and low-risk, for example:

GET /ws?events=sms.received,call.state

Technical approach

1. Add an aiohttp WebSocket route in create_app(modem, ...), either directly in server.py or in a small helper if the module starts getting crowded.
2. Reuse the same API-key auth middleware/policy as HTTP write endpoints. Do not allow unauthenticated LAN access to message/call metadata.
3. For each connected client, subscribe to the relevant EventBus event types and place serialized event envelopes onto a per-client bounded asyncio.Queue.
4. Serialize only public typed events, not raw AT command lines:
   • IncomingSMSEvent -> sms.received
   • SMSDeliveryReportEvent -> sms.delivery_report
   • CallStateEvent -> call.state
   • SignalQualityEvent -> signal.quality
   • ModemDisconnectedEvent / ModemReconnectedEvent -> modem.state
   • USSDResponseEvent -> ussd.response
5. Use bounded queues and a clear overflow policy. For the first PR, dropping the oldest event and sending an overflow notice is preferable to unbounded memory growth.
6. Ensure cleanup unsubscribes handlers and cancels sender tasks when the WebSocket closes.
7. Avoid logging phone numbers, SMS bodies, raw USSD text, API keys, or full close payloads. Debug logs should identify connection ids only.
8. Add tests with pytest-aiohttp that inject fake events and assert received JSON envelopes; no modem hardware should be required.

Affected modules and tests

Likely files:

• Modify: server.py — register /ws, auth behavior, event-to-WebSocket adapter.
• Create (optional): callstack/realtime.py — connection manager / serializer if that keeps server.py readable.
• Create: tests/test_websocket.py — auth rejection, successful connection, event serialization, disconnect cleanup, overflow behavior.
• Modify: tests/test_api_auth.py only if middleware needs WebSocket-specific coverage.
• Docs follow-up: README HTTP/realtime section after implementation.

Existing context:

• server.py already creates an aiohttp application and receives a Modem instance.
• EventBus.subscribe() / unsubscribe() provide imperative subscription hooks.
• EventBus.stream() currently streams a single event type; a WebSocket adapter may be simpler with explicit subscribe/unsubscribe handlers per event type.
• tests/test_api_auth.py already uses pytest-aiohttp patterns that can be mirrored.

Hardware / modem caveats

• This feature must be testable with fake events only; no real serial ports, SIM cards, phone numbers, or carrier network access.
• Do not stream raw +CMT, +CDSI, RING, or other AT lines. Typed event adapters keep modem-family quirks out of clients.
• Multipart SMS reassembly (Reassemble inbound multipart SMS before public delivery #10) should define when sms.received represents a complete logical message. Until then, document that /ws mirrors the current public incoming-message event semantics.
• Realtime clients can miss events while disconnected; durable replay is out of scope for this first PR.

Acceptance criteria

• GET /ws requires the same API-key policy as protected HTTP routes.
• A successful WebSocket connection receives a hello envelope with protocol version and supported event types.
• Emitting representative typed events on modem.bus sends sanitized JSON envelopes to connected clients.
• Multiple connected clients each receive events without one slow client blocking the others.
• Disconnect cleanup unsubscribes event handlers and cancels per-client tasks.
• Per-client queues are bounded and overflow behavior is deterministic/tested.
• Logs do not include API keys, full phone numbers, SMS bodies, raw USSD text, SIM identifiers, or modem serial/IMEI values.
• Tests require no real modem hardware.

Exact verification gates

git diff --check
PYTHONPATH=. uv run --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/test_websocket.py tests/test_api_auth.py tests/test_events.py -q
PYTHONPATH=. uv run --no-project --with pytest --with pytest-asyncio --with pytest-aiohttp --with pyserial-asyncio --with aiosqlite pytest tests/ -q

Non-goals

• Browser dashboard UI.
• Durable event replay or cursor/resume semantics.
• Replacing webhooks (Add reliable signed SMS webhooks with retry/backoff #21) or HTTP polling routes.
• Implementing multipart SMS reassembly (Reassemble inbound multipart SMS before public delivery #10).
• Streaming raw modem/AT traffic to clients.