From 96ca7e544d15c6655d8663c08e3eb9b11656e45a Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:18:40 -0700 Subject: [PATCH] revert(proxy): drop overfit Caido reconnect/HTTPQL band-aids, keep serialization lock (#799) Co-authored-by: Ahmed Allam --- strix/agents/prompts/system_prompt.jinja | 1 + strix/core/runner.py | 4 - strix/runtime/caido_bootstrap.py | 64 ++------ strix/runtime/session_manager.py | 21 +-- strix/tools/proxy/caido_api.py | 106 +----------- strix/tools/proxy/tools.py | 125 ++++++-------- tests/test_proxy_client.py | 199 ++--------------------- 7 files changed, 88 insertions(+), 432 deletions(-) diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index fd3b66d6..0ebd7c79 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -434,6 +434,7 @@ SPECIALIZED TOOLS: PROXY & INTERCEPTION: - Caido CLI - Modern web proxy (already running). Use the proxy tools directly, or import `caido_api` from sandbox Python scripts. +- HTTPQL filters (for `list_requests`): quote string values, leave integers unquoted (`resp.code.eq:200`, not `"200"`); combine terms with `AND`/`OR` (there is no `NOT` — use the negated operator `ne`/`ncont`/`nregex`). Numeric fields (`resp.code`, `req.port`) use `eq`/`ne`/`gt`/`gte`/`lt`/`lte`; text fields (`req.host`, `req.path`, `req.method`, `req.raw`) use `cont`/`ncont`/`eq`/`regex`. Example: `resp.code.gte:200 AND resp.code.lt:300 AND req.host.cont:"api"`. - NOTE: If you are seeing proxy errors when sending requests, it usually means you are not sending requests to a correct url/host/port. - Ignore Caido proxy-generated 50x HTML error pages; these are proxy issues (might happen when requesting a wrong host or SSL/TLS issues, etc). diff --git a/strix/core/runner.py b/strix/core/runner.py index 1e5d66b4..e6a15dc3 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -282,10 +282,6 @@ async def run_strix_scan( context: dict[str, Any] = { "coordinator": coordinator, "sandbox_session": bundle["session"], - # One ``SharedCaidoClient`` is reused by every agent in the scan - # (child contexts are shallow copies via ``dict(parent_ctx)``). It - # serializes access to the non-concurrency-safe GraphQL transport - # and rebuilds it if it dies mid-scan. "caido_client": bundle["caido_client"], "agent_id": root_id, "parent_id": None, diff --git a/strix/runtime/caido_bootstrap.py b/strix/runtime/caido_bootstrap.py index ae81f7f3..0b9ad5b1 100644 --- a/strix/runtime/caido_bootstrap.py +++ b/strix/runtime/caido_bootstrap.py @@ -80,72 +80,30 @@ async def _login_as_guest( raise RuntimeError(f"loginAsGuest failed after {attempts} attempts: {last_err}") -async def _aclose_quietly(client: Client) -> None: - """Best-effort close of a client whose setup failed; never raises.""" - with contextlib.suppress(Exception): - await client.aclose() - - -async def _connect_client( - session: BaseSandboxSession, - *, - host_url: str, - container_url: str, -) -> Client: - access_token = await _login_as_guest(session, container_url=container_url) - client = Client(host_url, auth=TokenAuthOptions(token=access_token)) - await client.connect() - return client - - async def bootstrap_caido( session: BaseSandboxSession, *, host_url: str, container_url: str, -) -> tuple[Client, str]: - """Connect to the in-container Caido sidecar and select a fresh project. - - Returns the connected client and the id of the temporary project it - selected. The project id lets :func:`reconnect_caido` rebuild a dead - transport while staying on the *same* project (and its captured traffic) - instead of creating a new empty one. - """ +) -> Client: + """Connect to the in-container Caido sidecar and select a fresh project.""" logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url) - client = await _connect_client(session, host_url=host_url, container_url=container_url) + access_token = await _login_as_guest(session, container_url=container_url) + + client = Client(host_url, auth=TokenAuthOptions(token=access_token)) + await client.connect() + try: project = await client.project.create( CreateProjectOptions(name="sandbox", temporary=True), ) await client.project.select(project.id) except BaseException: - # Don't leak the connected transport if project setup fails. - await _aclose_quietly(client) + # The connected client never reaches the session bundle if project + # setup fails, so close it here to avoid leaking the transport. + with contextlib.suppress(Exception): + await client.aclose() raise logger.info("Caido project selected: %s", project.id) - return client, str(project.id) - - -async def reconnect_caido( - session: BaseSandboxSession, - *, - host_url: str, - container_url: str, - project_id: str, -) -> Client: - """Rebuild a Caido client after its transport died, keeping the project. - - Re-authenticates, reconnects, and re-selects the existing project so the - caller keeps access to the traffic captured before the disconnect. - """ - logger.info("Reconnecting Caido client (host=%s, project=%s)", host_url, project_id) - client = await _connect_client(session, host_url=host_url, container_url=container_url) - try: - await client.project.select(project_id) - except BaseException: - # A missing/unavailable project must not leave the freshly-connected - # transport dangling — otherwise every retry leaks another one. - await _aclose_quietly(client) - raise return client diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py index 84950666..cdc6d955 100644 --- a/strix/runtime/session_manager.py +++ b/strix/runtime/session_manager.py @@ -5,20 +5,15 @@ from __future__ import annotations import logging import shutil from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import Any from agents.sandbox.entries import BaseEntry, LocalDir from agents.sandbox.manifest import Environment, Manifest from strix.config import load_settings from strix.runtime.backends import get_backend -from strix.runtime.caido_bootstrap import bootstrap_caido, reconnect_caido +from strix.runtime.caido_bootstrap import bootstrap_caido from strix.runtime.local_dir_staging import stage_symlink_safe_dir -from strix.tools.proxy.caido_api import SharedCaidoClient - - -if TYPE_CHECKING: - from caido_sdk_client import Client as CaidoClient logger = logging.getLogger(__name__) @@ -136,24 +131,16 @@ async def create_or_reuse( host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}" logger.debug("Caido host endpoint resolved: %s", host_caido_url) - caido_client, caido_project_id = await bootstrap_caido( + caido_client = await bootstrap_caido( session, host_url=host_caido_url, container_url=container_caido_url, ) - async def _reconnect_caido() -> CaidoClient: - return await reconnect_caido( - session, - host_url=host_caido_url, - container_url=container_caido_url, - project_id=caido_project_id, - ) - bundle = { "client": client, "session": session, - "caido_client": SharedCaidoClient(caido_client, _reconnect_caido), + "caido_client": caido_client, } _SESSION_CACHE[scan_id] = bundle logger.info("Sandbox session for scan %s ready and cached", scan_id) diff --git a/strix/tools/proxy/caido_api.py b/strix/tools/proxy/caido_api.py index 11e26859..277d5c48 100644 --- a/strix/tools/proxy/caido_api.py +++ b/strix/tools/proxy/caido_api.py @@ -3,9 +3,7 @@ from __future__ import annotations import asyncio -import contextlib import json -import logging import os import time import urllib.request @@ -28,9 +26,6 @@ if TYPE_CHECKING: from caido_sdk_client import Client as CaidoClient -logger = logging.getLogger(__name__) - - RequestPart = Literal["request", "response"] SortBy = Literal[ "timestamp", @@ -50,18 +45,6 @@ _SITEMAP_PAGE_SIZE = 30 _DEFAULT_CAIDO_URL = "http://127.0.0.1:48080" _CLIENT_CACHE: dict[str, Client] = {} _CLIENT_LOCK = asyncio.Lock() - -# Substrings that mean the shared client's transport has died or is being used -# concurrently — recoverable by rebuilding the client and retrying once. -_CONNECTION_ERROR_MARKERS = ( - "transport is already connected", - "connector is closed", - "server disconnected", - "session is closed", - "cannot write to closing transport", - "connection reset", - "connection closed", -) _REQ_FIELD_MAP: dict[SortBy, tuple[str, str]] = { "timestamp": ("req", "created_at"), "host": ("req", "host"), @@ -108,22 +91,6 @@ async def _new_client() -> Client: return client -async def _safe_aclose(client: Client | None) -> None: - """Close a (possibly dead) client without letting teardown errors escape.""" - if client is None: - return - with contextlib.suppress(Exception): - await client.aclose() - - -def _is_connection_error(exc: BaseException) -> bool: - message = str(exc).lower() - if any(marker in message for marker in _CONNECTION_ERROR_MARKERS): - return True - cause = exc.__cause__ or exc.__context__ - return cause is not None and cause is not exc and _is_connection_error(cause) - - async def get_client() -> Client: """Return the shared Caido client, creating it under a lock if needed. @@ -139,73 +106,19 @@ async def get_client() -> Client: return client -async def call_with_client[T]( - fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True -) -> T: - """Run ``fn`` against the shared client, serialized and reconnect-safe. +async def call_with_client[T](fn: Callable[[Client], Awaitable[T]]) -> T: + """Run ``fn`` against the shared client, serialized through ``_CLIENT_LOCK``. The Caido GraphQL transport is not safe for concurrent use: two in-flight - requests race and raise "Transport is already connected". All proxy calls - are therefore serialized through ``_CLIENT_LOCK``. If the cached client's - transport has since died ("Connector is closed" / "Server disconnected"), - the stale client is closed and rebuilt so subsequent calls stop failing - against a dead client. - - ``fn`` is only re-run automatically when ``idempotent`` is true. For - mutations (replay, scope create/update/delete) a connection error may - arrive *after* Caido applied the change, so we heal the client for future - calls but re-raise instead of risking a double-apply. + requests race and raise "Transport is already connected". Serializing every + proxy call through the lock prevents that. """ async with _CLIENT_LOCK: client = _CLIENT_CACHE.get("default") if client is None: client = await _new_client() _CLIENT_CACHE["default"] = client - try: - return await fn(client) - except Exception as exc: - if not _is_connection_error(exc): - raise - new_client = await _new_client() - _CLIENT_CACHE["default"] = new_client - await _safe_aclose(client) - if not idempotent: - raise - return await fn(new_client) - - -class SharedCaidoClient: - """Serialized, reconnect-safe wrapper around one host-side Caido client. - - Every agent in a scan shares a single instance (propagated through the - shallow-copied run context). ``call`` serializes access — the SDK transport - is not concurrency-safe — and, when the transport dies, rebuilds the client - via ``reconnect`` (which preserves the Caido project) and closes the dead - one, so a transient Caido restart no longer disables proxy tools for the - rest of the scan. - """ - - def __init__(self, client: Client, reconnect: Callable[[], Awaitable[Client]]) -> None: - self._client = client - self._reconnect = reconnect - self._lock = asyncio.Lock() - - async def call[T](self, fn: Callable[[Client], Awaitable[T]], *, idempotent: bool = True) -> T: - async with self._lock: - try: - return await fn(self._client) - except Exception as exc: - if not _is_connection_error(exc): - raise - dead, self._client = self._client, await self._reconnect() - await _safe_aclose(dead) - if not idempotent: - raise - return await fn(self._client) - - async def aclose(self) -> None: - async with self._lock: - await _safe_aclose(self._client) + return await fn(client) async def close_client() -> None: @@ -546,9 +459,7 @@ async def repeat_request( ) return await replay_send_raw(client, raw=raw, connection=connection) - # A replay mutates server state; don't auto-retry if the transport dies - # mid-send (the request may already have been sent). - return await call_with_client(_run, idempotent=False) + return await call_with_client(_run) async def scope_rules( @@ -569,8 +480,7 @@ async def scope_rules( scope_name=scope_name, ) - # get/list are read-only and safe to retry; create/update/delete mutate. - return await call_with_client(_run, idempotent=action in {"get", "list"}) + return await call_with_client(_run) async def _scope_rules_with_client( @@ -819,11 +729,9 @@ async def view_sitemap_entry(entry_id: str) -> dict[str, Any]: __all__ = [ "RequestPart", "ScopeAction", - "SharedCaidoClient", "SitemapDepth", "SortBy", "SortOrder", - "call_with_client", "close_client", "get_client", "list_requests", diff --git a/strix/tools/proxy/tools.py b/strix/tools/proxy/tools.py index 4fdb850a..091c489f 100644 --- a/strix/tools/proxy/tools.py +++ b/strix/tools/proxy/tools.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import dataclasses import json import logging @@ -13,13 +14,14 @@ from typing import TYPE_CHECKING, Any, Literal from agents import RunContextWrapper, function_tool from strix.tools.proxy import caido_api -from strix.tools.proxy.caido_api import SharedCaidoClient logger = logging.getLogger(__name__) if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + from caido_sdk_client import Client from strix.tools.proxy.caido_api import ( @@ -29,7 +31,7 @@ if TYPE_CHECKING: SortOrder, ) else: - from strix.tools.proxy.caido_api import ( + from strix.tools.proxy.caido_api import ( # noqa: TC001 RequestPart, SitemapDepth, SortBy, @@ -39,19 +41,21 @@ else: ScopeAction = Literal["get", "list", "create", "update", "delete"] +# All agents in a scan share one host-side Caido client whose GraphQL transport +# is not concurrency-safe (parallel calls raise "Transport is already +# connected"). Serialize every host-side proxy call through this lock. +_CAIDO_CALL_LOCK = asyncio.Lock() -def _ctx_proxy(ctx: RunContextWrapper) -> SharedCaidoClient | None: - """Return the scan-wide serialized, reconnect-safe Caido client holder. - All agents in a scan share one :class:`SharedCaidoClient` whose GraphQL - transport is not concurrency-safe (parallel calls raise "Transport is - already connected"). ``SharedCaidoClient.call`` serializes access and - rebuilds the transport if it dies mid-scan. Returns ``None`` when no holder - is present (e.g. standalone tool invocation outside a scan run). - """ +def _ctx_client(ctx: RunContextWrapper) -> Client | None: inner = ctx.context if isinstance(ctx.context, dict) else {} - proxy = inner.get("caido_client") - return proxy if isinstance(proxy, SharedCaidoClient) else None + return inner.get("caido_client") + + +async def _call[T](client: Client, fn: Callable[[Client], Awaitable[T]]) -> T: + """Run ``fn`` against the shared client, serialized under ``_CAIDO_CALL_LOCK``.""" + async with _CAIDO_CALL_LOCK: + return await fn(client) def _to_tool_json(value: Any) -> Any: @@ -93,39 +97,6 @@ def _err(name: str, exc: Exception) -> str: ) -_HTTPQL_HINT = ( - "HTTPQL syntax: quote string values and leave integers unquoted; combine " - "terms with AND / OR (there is no NOT). Numeric fields (resp.code, req.port, " - "id, roundtrip) use eq/ne/gt/gte/lt/lte; text/byte fields (req.host, req.path, " - "req.method, req.raw, resp.raw) use cont/ncont/eq/ne/like/nlike/regex/nregex. " - "Example: 'resp.code.gte:200 AND resp.code.lt:300 AND req.host.cont:\"api\"'." -) - - -def _is_httpql_error(exc: Exception) -> bool: - message = str(exc).lower() - return "httpql" in message or ("filter" in message and "pars" in message) - - -def _httpql_error(exc: Exception, httpql_filter: str | None) -> str: - """Return an actionable error for a rejected HTTPQL filter. - - Preserves Caido's exact parser message and echoes the offending query so - the agent can self-correct instead of retrying the same broken filter. - """ - logger.info("list_requests rejected HTTPQL filter %r: %s", httpql_filter, exc) - return json.dumps( - { - "success": False, - "error": f"Invalid HTTPQL filter: {exc}", - "httpql_filter": httpql_filter, - "hint": _HTTPQL_HINT, - }, - ensure_ascii=False, - default=str, - ) - - @function_tool(timeout=120) async def list_requests( ctx: RunContextWrapper, @@ -184,12 +155,13 @@ async def list_requests( sort_order: ``asc`` or ``desc``. scope_id: Restrict to a Caido scope (managed via ``scope_rules``). """ - proxy = _ctx_proxy(ctx) - if proxy is None: + client = _ctx_client(ctx) + if client is None: return _no_client() try: - connection = await proxy.call( + connection = await _call( + client, lambda client: caido_api.list_requests_with_client( client, httpql_filter=httpql_filter, @@ -198,7 +170,7 @@ async def list_requests( sort_by=sort_by, sort_order=sort_order, scope_id=scope_id, - ) + ), ) entries = [] @@ -252,8 +224,6 @@ async def list_requests( default=str, ) except Exception as exc: # noqa: BLE001 - if httpql_filter and _is_httpql_error(exc): - return _httpql_error(exc, httpql_filter) return _err("list_requests", exc) @@ -291,13 +261,14 @@ async def view_request( page: 1-indexed page number (only when no ``search_pattern``). page_size: Lines per page. """ - proxy = _ctx_proxy(ctx) - if proxy is None: + client = _ctx_client(ctx) + if client is None: return _no_client() try: - result = await proxy.call( - lambda client: caido_api.get_request_with_client(client, request_id, part=part) + result = await _call( + client, + lambda client: caido_api.get_request_with_client(client, request_id, part=part), ) if result is None: return json.dumps( @@ -408,8 +379,8 @@ async def repeat_request( - ``body`` — replace the body string entirely. - ``cookies`` — dict of cookies to add/update. """ - proxy = _ctx_proxy(ctx) - if proxy is None: + client = _ctx_client(ctx) + if client is None: return _no_client() mods = modifications or {} @@ -431,9 +402,7 @@ async def repeat_request( return await caido_api.replay_send_raw(client, raw=raw, connection=connection) try: - # A replay mutates target state, so don't auto-retry on a mid-send - # transport failure (the request may already have been sent). - replay = await proxy.call(_do, idempotent=False) + replay = await _call(client, _do) if replay is None: return json.dumps( {"success": False, "error": f"Request {request_id} not found"}, @@ -492,18 +461,19 @@ async def list_sitemap( (recursive subtree). Only meaningful with ``parent_id``. page: 1-indexed page (30 entries per page). """ - proxy = _ctx_proxy(ctx) - if proxy is None: + client = _ctx_client(ctx) + if client is None: return _no_client() try: - payload = await proxy.call( + payload = await _call( + client, lambda client: caido_api.list_sitemap_with_client( client, scope_id=scope_id, parent_id=parent_id, depth=depth, page=page, - ) + ), ) return json.dumps(payload, ensure_ascii=False, default=str) except Exception as exc: # noqa: BLE001 @@ -525,12 +495,13 @@ async def view_sitemap_entry( Args: entry_id: ID from ``list_sitemap`` (or any nested entry). """ - proxy = _ctx_proxy(ctx) - if proxy is None: + client = _ctx_client(ctx) + if client is None: return _no_client() try: - payload = await proxy.call( - lambda client: caido_api.view_sitemap_entry_with_client(client, entry_id) + payload = await _call( + client, + lambda client: caido_api.view_sitemap_entry_with_client(client, entry_id), ) return json.dumps(payload, ensure_ascii=False, default=str) except Exception as exc: # noqa: BLE001 @@ -583,13 +554,13 @@ async def scope_rules( scope_id: Required for ``get`` / ``update`` / ``delete``. scope_name: Required for ``create`` / ``update``. """ - proxy = _ctx_proxy(ctx) - if proxy is None: + client = _ctx_client(ctx) + if client is None: return _no_client() try: if action == "list": - scopes = await proxy.call(caido_api.scope_list) + scopes = await _call(client, caido_api.scope_list) return json.dumps( {"success": True, "scopes": [_to_tool_json(s) for s in scopes]}, ensure_ascii=False, @@ -602,7 +573,7 @@ async def scope_rules( ensure_ascii=False, default=str, ) - scope = await proxy.call(lambda client: caido_api.scope_get(client, scope_id)) + scope = await _call(client, lambda client: caido_api.scope_get(client, scope_id)) return json.dumps( {"success": True, "scope": _to_tool_json(scope)}, ensure_ascii=False, @@ -615,11 +586,11 @@ async def scope_rules( ensure_ascii=False, default=str, ) - scope = await proxy.call( + scope = await _call( + client, lambda client: caido_api.scope_create( client, name=scope_name, allowlist=allowlist, denylist=denylist ), - idempotent=False, ) return json.dumps( {"success": True, "scope": _to_tool_json(scope)}, @@ -636,11 +607,11 @@ async def scope_rules( ensure_ascii=False, default=str, ) - scope = await proxy.call( + scope = await _call( + client, lambda client: caido_api.scope_update( client, scope_id, name=scope_name, allowlist=allowlist, denylist=denylist ), - idempotent=False, ) return json.dumps( {"success": True, "scope": _to_tool_json(scope)}, @@ -653,7 +624,7 @@ async def scope_rules( ensure_ascii=False, default=str, ) - await proxy.call(lambda client: caido_api.scope_delete(client, scope_id), idempotent=False) + await _call(client, lambda client: caido_api.scope_delete(client, scope_id)) return json.dumps( { "success": True, diff --git a/tests/test_proxy_client.py b/tests/test_proxy_client.py index 64436583..3b589459 100644 --- a/tests/test_proxy_client.py +++ b/tests/test_proxy_client.py @@ -1,20 +1,19 @@ -"""Tests for the shared Caido client lifecycle and proxy error handling. +"""Tests for the shared Caido client lifecycle and proxy call serialization. -Covers the concurrency/reconnect guarantees of ``caido_api.call_with_client`` -(the sandbox-imported path) and ``caido_api.SharedCaidoClient`` (the host-side -holder), plus the actionable HTTPQL errors in ``proxy.tools``. +Covers the caching + serialization guarantees of ``caido_api.call_with_client`` +(the sandbox-imported path) and ``proxy.tools._call`` (the host-side path). The +Caido GraphQL transport is not concurrency-safe, so both paths must run one +call at a time against the shared client. """ from __future__ import annotations import asyncio -import json from typing import TYPE_CHECKING, Any, cast import pytest from strix.tools.proxy import caido_api, tools -from strix.tools.proxy.caido_api import SharedCaidoClient if TYPE_CHECKING: @@ -91,83 +90,15 @@ async def test_failed_init_does_not_poison_cache(monkeypatch: pytest.MonkeyPatch assert "default" not in caido_api._CLIENT_CACHE -async def test_call_with_client_reconnects_and_closes_dead_transport( - monkeypatch: pytest.MonkeyPatch, -) -> None: - dead = _FakeClient("dead") - fresh = _FakeClient("fresh") - caido_api._CLIENT_CACHE["default"] = cast("Any", dead) - - new_calls = {"n": 0} - - async def _new() -> Any: - new_calls["n"] += 1 - return fresh - - monkeypatch.setattr(caido_api, "_new_client", _new) - - attempts: list[Any] = [] - - async def fn(client: Any) -> str: - attempts.append(client) - if len(attempts) == 1: - raise RuntimeError("Transport is already connected") - return "ok" - - assert await caido_api.call_with_client(fn) == "ok" - assert attempts == [dead, fresh] - assert new_calls["n"] == 1 - assert caido_api._CLIENT_CACHE["default"] is fresh - assert dead.closed is True # stale transport is not leaked - - -async def test_call_with_client_non_idempotent_rebuilds_but_reraises( - monkeypatch: pytest.MonkeyPatch, -) -> None: - dead = _FakeClient("dead") - fresh = _FakeClient("fresh") - caido_api._CLIENT_CACHE["default"] = cast("Any", dead) - - async def _new() -> Any: - return fresh - - monkeypatch.setattr(caido_api, "_new_client", _new) - - calls = {"n": 0} - - async def fn(_client: Any) -> str: - calls["n"] += 1 - raise RuntimeError("Server disconnected") - - # A mutation must not be auto-retried (it may already have applied), but the - # dead client is still healed so later calls succeed. - with pytest.raises(RuntimeError, match="Server disconnected"): - await caido_api.call_with_client(fn, idempotent=False) - assert calls["n"] == 1 - assert caido_api._CLIENT_CACHE["default"] is fresh - assert dead.closed is True - - -async def test_call_with_client_does_not_retry_application_errors( - monkeypatch: pytest.MonkeyPatch, -) -> None: +async def test_call_with_client_propagates_errors() -> None: cached = _FakeClient("cached") caido_api._CLIENT_CACHE["default"] = cast("Any", cached) - async def _new() -> Any: - raise AssertionError("deterministic errors must not trigger a reconnect") - - monkeypatch.setattr(caido_api, "_new_client", _new) - - calls = {"n": 0} - async def fn(_client: Any) -> str: - calls["n"] += 1 raise ValueError("Invalid HTTPQL filter") with pytest.raises(ValueError, match="Invalid HTTPQL"): await caido_api.call_with_client(fn) - assert calls["n"] == 1 assert caido_api._CLIENT_CACHE["default"] is cached @@ -177,7 +108,7 @@ async def test_call_with_client_serializes_concurrent_calls( caido_api._CLIENT_CACHE["default"] = cast("Any", _FakeClient("shared")) async def _new() -> Any: - raise AssertionError("no reconnect expected") + raise AssertionError("no new client expected") monkeypatch.setattr(caido_api, "_new_client", _new) @@ -194,61 +125,8 @@ async def test_call_with_client_serializes_concurrent_calls( assert state["max"] == 1 -async def test_shared_client_reconnects_and_closes_dead_transport() -> None: - dead = _FakeClient("dead") - fresh = _FakeClient("fresh") - - async def _reconnect() -> Any: - return fresh - - holder = SharedCaidoClient(cast("Any", dead), _reconnect) - - attempts: list[Any] = [] - - async def fn(client: Any) -> str: - attempts.append(client) - if len(attempts) == 1: - raise RuntimeError("Connector is closed") - return "ok" - - assert await holder.call(fn) == "ok" - assert attempts == [dead, fresh] - assert dead.closed is True - - -async def test_shared_client_non_idempotent_rebuilds_but_reraises() -> None: - dead = _FakeClient("dead") - fresh = _FakeClient("fresh") - - async def _reconnect() -> Any: - return fresh - - holder = SharedCaidoClient(cast("Any", dead), _reconnect) - - calls = {"n": 0} - - async def fn(_client: Any) -> str: - calls["n"] += 1 - raise RuntimeError("Server disconnected") - - with pytest.raises(RuntimeError, match="Server disconnected"): - await holder.call(fn, idempotent=False) - assert calls["n"] == 1 - assert dead.closed is True - # The healthy client remains for the next call. - assert await holder.call(lambda _c: _ok()) == "ok" - - -async def _ok() -> str: - return "ok" - - -async def test_shared_client_serializes_concurrent_calls() -> None: - async def _reconnect() -> Any: - raise AssertionError("no reconnect expected") - - holder = SharedCaidoClient(cast("Any", _FakeClient("shared")), _reconnect) - +async def test_host_call_serializes_concurrent_calls() -> None: + client = _FakeClient("host") state = {"active": 0, "max": 0} async def fn(_client: Any) -> str: @@ -258,64 +136,21 @@ async def test_shared_client_serializes_concurrent_calls() -> None: state["active"] -= 1 return "ok" - await asyncio.gather(*(holder.call(fn) for _ in range(6))) + await asyncio.gather(*(tools._call(cast("Any", client), fn) for _ in range(6))) assert state["max"] == 1 -async def test_shared_client_passes_through_application_errors() -> None: - async def _reconnect() -> Any: - raise AssertionError("deterministic errors must not trigger a reconnect") - - holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect) - - async def fn(_client: Any) -> str: - raise ValueError("Invalid HTTPQL filter") - - with pytest.raises(ValueError, match="Invalid HTTPQL"): - await holder.call(fn) - - -def test_is_connection_error_matches_markers_and_causes() -> None: - assert caido_api._is_connection_error(RuntimeError("Transport is already connected")) - assert caido_api._is_connection_error(RuntimeError("Connector is closed")) - assert caido_api._is_connection_error(RuntimeError("Server disconnected")) - assert not caido_api._is_connection_error(ValueError("Invalid HTTPQL filter")) - - nested = RuntimeError("wrapper") - nested.__cause__ = RuntimeError("connection reset by peer") - assert caido_api._is_connection_error(nested) - - class _Ctx: def __init__(self, context: Any) -> None: self.context = context -def test_ctx_proxy_returns_holder_when_present() -> None: - async def _reconnect() -> Any: - raise AssertionError("unused") - - holder = SharedCaidoClient(cast("Any", _FakeClient("c")), _reconnect) - got = tools._ctx_proxy(cast("Any", _Ctx({"caido_client": holder}))) - assert got is holder +def test_ctx_client_returns_client_when_present() -> None: + client = _FakeClient("host") + got = tools._ctx_client(cast("Any", _Ctx({"caido_client": client}))) + assert got is client -def test_ctx_proxy_returns_none_without_holder() -> None: - assert tools._ctx_proxy(cast("Any", _Ctx({}))) is None - assert tools._ctx_proxy(cast("Any", _Ctx(None))) is None - assert tools._ctx_proxy(cast("Any", _Ctx({"caido_client": object()}))) is None - - -def test_is_httpql_error_detection() -> None: - assert tools._is_httpql_error(RuntimeError("HTTPQL parse error at column 4")) - assert tools._is_httpql_error(RuntimeError("failed to parse filter")) - assert not tools._is_httpql_error(RuntimeError("Transport is already connected")) - - -def test_httpql_error_preserves_message_and_query() -> None: - exc = RuntimeError("HTTPQL parse error: unexpected token at column 12") - payload = json.loads(tools._httpql_error(exc, 'resp.code.eq:"200"')) - assert payload["success"] is False - assert "unexpected token at column 12" in payload["error"] - assert payload["httpql_filter"] == 'resp.code.eq:"200"' - assert "AND / OR" in payload["hint"] +def test_ctx_client_returns_none_without_client() -> None: + assert tools._ctx_client(cast("Any", _Ctx({}))) is None + assert tools._ctx_client(cast("Any", _Ctx(None))) is None