Class DirectConnectServer

src/server/server.py:161–490

Direct Connect server — HTTP ``/sessions`` + WS ``/ws/ ``. Lifecycle: srv = DirectConnectServer(config, manager, spawn_agent) await srv.start() await srv.serve_forever() # or use srv.stop() to shut down await srv.stop() Two listeners: HTTP on ``config.p

@dataclass
class DirectConnectServer:
    """Direct Connect server — HTTP ``/sessions`` + WS ``/ws/<sid>``.

    Lifecycle:
        srv = DirectConnectServer(config, manager, spawn_agent)
        await srv.start()
        await srv.serve_forever()  # or use srv.stop() to shut down
        await srv.stop()

    Two listeners: HTTP on ``config.port`` (or ephemeral) and WS on a
    separate ephemeral port. ``ws_url`` in the create-session response
    points at the WS port. Auth is per-session: ``POST /sessions``
    returns ``auth_token``; the WS upgrade requires it via
    ``Authorization: Bearer <token>`` or ``?token=<token>``.
    """

    config: ServerConfig
    manager: SessionManager
    spawn_agent: SpawnAgent
    _http_server: asyncio.AbstractServer | None = None
    _ws_server: asyncio.Server | None = None
    _session_tokens: dict[str, str] = field(default_factory=dict)
    _ws_port: int | None = None

    # ─── Public lifecycle ────────────────────────────────────────────

    async def start(self) -> None:
        """Bind both listeners; ``serve_forever`` to actually run."""
        # WS listener first so we know its port for the ``ws_url`` we
        # hand out from the HTTP route.
        self._ws_server = await ws_serve(
            self._handle_ws_connection,
            host=self.config.host or '127.0.0.1',
            port=0,  # ephemeral
        )
        ws_sockets = list(self._ws_server.sockets or [])
        if not ws_sockets:
            raise RuntimeError('DirectConnectServer: WS listener has no socket')
        self._ws_port = ws_sockets[0].getsockname()[1]

        # HTTP listener.
        if self.config.unix:
            self._http_server = await asyncio.start_unix_server(
                self._handle_http_connection,
                path=self.config.unix,
            )
        else:
            self._http_server = await asyncio.start_server(
                self._handle_http_connection,
                host=self.config.host or '127.0.0.1',
                port=self.config.port,
            )

    async def serve_forever(self) -> None:
        if self._http_server is None or self._ws_server is None:
            raise RuntimeError('start() must be called before serve_forever()')
        # Both servers run concurrently; cancellation of either tears
        # down both atomically via the gather.