Function _running_server

tests/server/test_agent_server_e2e.py:178–194

(tmp_path, provider_cls, registry, agent_config=None)

@contextlib.asynccontextmanager
async def _running_server(tmp_path, provider_cls, registry, agent_config=None):
    config = ServerConfig(host="127.0.0.1", port=_free_port(), workspace=str(tmp_path))
    manager = SessionManager(workspace=str(tmp_path), index_path=tmp_path / "idx.json")
    spawn = make_spawn_agent(agent_config or AgentServerConfig())
    server = DirectConnectServer(config=config, manager=manager, spawn_agent=spawn)
    await server.start()
    serve_task = asyncio.get_running_loop().create_task(server.serve_forever())
    with contextlib.ExitStack() as stack:
        for p in _patches(provider_cls, registry):
            stack.enter_context(p)
        try:
            yield config
        finally:
            await server.stop()
            serve_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await serve_task


async def _wait_for(predicate, timeout=10.0, interval=0.05):