diff --git a/.craftsmanship-baseline.json b/.craftsmanship-baseline.json index b3b63055..a0ebc17f 100644 --- a/.craftsmanship-baseline.json +++ b/.craftsmanship-baseline.json @@ -5517,11 +5517,6 @@ "kind": "method-size", "detail": "classify_write_class" }, - { - "file": "mcp_server/tool_error_handler.py", - "kind": "method-size", - "detail": "_classify_error" - }, { "file": "mcp_server/tool_error_handler.py", "kind": "method-size", diff --git a/docs/module-inventory.md b/docs/module-inventory.md index 64c37a25..002efbce 100644 --- a/docs/module-inventory.md +++ b/docs/module-inventory.md @@ -319,7 +319,7 @@ Run the measurement command in the header to get a current file listing. - `ap_sync_loop.py` — `_SyncLoop`, the cross-loop sync/drain primitive pinning one event loop across an `APBridge` caller's lifetime (issue #258: drains cancelled tasks before stopping the loop, closing the "Task was destroyed but it is pending!" GC-warning race) - `workflow_graph_ast_symbols.py` — AST *symbol* loading: the AP label set, the symbol-type mapping, and the per-label batched query + WHERE-clause construction - `workflow_graph_ast_edges.py` — AST *edge* loading: the ~89 AP rel-table (CALLS/IMPORTS/MEMBER_OF/USES) batched queries -- `workflow_graph_ast_response.py` — `as_list`, normalizes AP's `query_graph` `{columns, rows}` response shape into plain dicts (shared by the symbols + edges modules) +- `workflow_graph_ast_response.py` — `as_list`, normalizes AP's `query_graph` `{columns, rows}` response shape into plain dicts (shared by the symbols + edges modules); `normalize_search_hits`, the same normalization seam applied to `search_codebase` hits (split out of `workflow_graph_source_ast.py::WorkflowGraphASTSource.search_codebase`, over the 300-line cap, PR #449 review round 2) Note: `pg_store.py` persists to PostgreSQL when configured (the `install-plugin.sh --postgres` opt-in or an explicit `DATABASE_URL`); diff --git a/mcp_server/handlers/unified_search.py b/mcp_server/handlers/unified_search.py index 13476ac6..d679eb14 100644 --- a/mcp_server/handlers/unified_search.py +++ b/mcp_server/handlers/unified_search.py @@ -7,6 +7,10 @@ When AP is off, the handler returns Cortex-only results marked ``status: partial, sources: [cortex]`` — never fails. When Cortex returns nothing and AP is on, the response is the AP-only hits. +When AP is on but the per-call attempt itself fails (timeout, transport +error, not installed), ``status`` is also ``partial`` and ``degraded`` +names the source and reason — this is distinct from AP genuinely +returning zero hits, which stays ``status: ok, degraded: null``. The fusion contract: each input list must present unique string ids. - Memories use ``memory:`` (added by this handler). @@ -119,9 +123,11 @@ async def handler(args: dict[str, Any] | None = None) -> dict[str, Any]: sources = ["cortex"] ap_hits: list[dict] = [] + ap_degraded_reason: str | None = None if is_enabled(): ast_source = WorkflowGraphASTSource() ap_hits = ast_source.search_codebase(query, limit=max(top_n * 2, top_n)) + ap_degraded_reason = ast_source.last_search_degraded_reason sources.append("ap") fused = fuse( @@ -129,8 +135,21 @@ async def handler(args: dict[str, Any] | None = None) -> dict[str, Any]: k=k, top_n=top_n, ) + # status/degraded reflect the ACTUAL per-call outcome, not just the + # static is_enabled() config flag: an AP that is enabled but timed out + # or errored on this call must not read the same as "AP found nothing" + # (both would otherwise be counts.ap=0, status=ok, sources=[...,"ap"]). + degraded: dict[str, str] | None = None + if not is_enabled(): + status = "partial" + elif ap_degraded_reason: + status = "partial" + degraded = {"source": "ap", "reason": ap_degraded_reason} + else: + status = "ok" resp = { - "status": "ok" if is_enabled() else "partial", + "status": status, + "degraded": degraded, "query": query, "sources": sources, "counts": { diff --git a/mcp_server/infrastructure/ap_bridge.py b/mcp_server/infrastructure/ap_bridge.py index d793a7bf..9f7f549d 100644 --- a/mcp_server/infrastructure/ap_bridge.py +++ b/mcp_server/infrastructure/ap_bridge.py @@ -30,6 +30,7 @@ from typing import Any from mcp_server.errors import McpConnectionError +from mcp_server.infrastructure.mcp_call_timeout import interactive_call_timeout_s from mcp_server.infrastructure.mcp_client import MCPClient from mcp_server.observability import silent_failure from mcp_server.infrastructure.memory_config import get_memory_settings @@ -288,29 +289,69 @@ async def connect(self) -> bool: ) return False - async def call(self, tool: str, args: dict | None = None) -> Any: - """Call an AP tool. Returns ``None`` if AP is unavailable.""" + def _degrade(self, reason: str, note: str) -> None: + """Record why the last AP call failed and emit the stderr note. + + Shared by both ``call()`` except-branches (timeout, other + exception) — the caller always gets ``None`` back and + ``unavailable_reason`` always names why. + """ + self._unavailable_reason = reason + print(note, file=sys.stderr) + + async def call( + self, tool: str, args: dict | None = None, *, timeout_s: float | None = None + ) -> Any: + """Call an AP tool. Returns ``None`` if AP is unavailable. + + ``timeout_s`` bounds this single call with a wall-clock ceiling. + The client itself runs AP with ``callTimeoutMs=0`` (indexing may + legitimately exceed any fixed bound), so without this the only + backstop is the 600s wedge-silence window — far too slow for an + interactive read/lookup. Interactive wrappers pass + ``interactive_call_timeout_s()``; the indexing wrappers leave it + ``None`` (unbounded). A timeout degrades exactly like any other AP + failure: reason recorded, stderr note, ``None`` returned so callers + fall back to Cortex-only results. + """ if tool not in _AP_TOOLS: raise ValueError(f"AP tool not in allowlist: {tool!r}") + self._unavailable_reason = None # this call's outcome is authoritative if not await self.connect(): return None if self._client is None: # connect() success guarantees a client; defensive return None try: - return await self._client.call(tool, args or {}) + coro = self._client.call(tool, args or {}) + if timeout_s is not None: + return await asyncio.wait_for(coro, timeout=timeout_s) + return await coro + except asyncio.TimeoutError: # interactive ceiling hit — degrade, don't hang + self._degrade( + f"TimeoutError: AP call {tool} exceeded {timeout_s:.0f}s", + f"[cortex] AP call {tool} timed out after {timeout_s:.0f}s " + f"(interactive ceiling); degrading to Cortex-only.", + ) + return None except Exception as exc: # noqa: BLE001 — failure is reported to stderr; execution degrades, never crashes - self._unavailable_reason = f"{type(exc).__name__}: {exc}" - print( + self._degrade( + f"{type(exc).__name__}: {exc}", f"[cortex] AP call {tool} failed: {exc}", - file=sys.stderr, ) return None # ── Convenience wrappers matching AP's MCP schema (src/tool_schemas.rs). # All Stage-3a tools are scoped to a ``graph_path`` returned by # index_codebase; callers pass it through or rely on the cached one. + # ── Interactive read-path tools carry a wall-clock ceiling + # (interactive_call_timeout_s) so a wedged-but-connected AP degrades to + # Cortex-only in seconds instead of stalling for the 600s wedge window. + # The indexing/write tools below (index_codebase, analyze_codebase, + # resolve_graph, cluster_graph, detect_changes) stay unbounded on purpose. async def health_check(self) -> Any: - return await self.call("health_check", {}) + return await self.call( + "health_check", {}, timeout_s=interactive_call_timeout_s() + ) async def index_codebase( self, @@ -331,7 +372,15 @@ async def index_codebase( ) async def query_graph(self, graph_path: str, query: str) -> Any: - """Execute a Cypher ``query`` against the graph at ``graph_path``.""" + """Execute a Cypher ``query`` against the graph at ``graph_path``. + + Left UNBOUNDED: query_graph drives the AST symbol/edge build loop + (iter_symbols / iter_edges, ~21 label + ~89 rel-table queries per + graph), which is part of the ingestion path where a single query + over a large graph may legitimately run long. Only the terminal + interactive lookups (get_symbol / get_context / search_codebase / + …) carry the interactive ceiling. + """ return await self.call( "query_graph", {"graph_path": graph_path, "query": query}, @@ -342,6 +391,7 @@ async def get_symbol(self, graph_path: str, qualified_name: str) -> Any: return await self.call( "get_symbol", {"graph_path": graph_path, "qualified_name": qualified_name}, + timeout_s=interactive_call_timeout_s(), ) async def get_context(self, graph_path: str, qualified_name: str) -> Any: @@ -354,6 +404,7 @@ async def get_context(self, graph_path: str, qualified_name: str) -> Any: return await self.call( "get_context", {"graph_path": graph_path, "qualified_name": qualified_name}, + timeout_s=interactive_call_timeout_s(), ) async def get_processes(self, graph_path: str) -> Any: @@ -362,7 +413,11 @@ async def get_processes(self, graph_path: str) -> Any: Each process: entry_point, entry_kind (main/test/handler/lib_entry), depth, node_count. Requires cluster_graph to have run. """ - return await self.call("get_processes", {"graph_path": graph_path}) + return await self.call( + "get_processes", + {"graph_path": graph_path}, + timeout_s=interactive_call_timeout_s(), + ) async def resolve_graph(self, graph_path: str) -> Any: """Stage 3b — resolve cross-file edges (Imports/Calls/Implements/ @@ -389,6 +444,7 @@ async def search_codebase( return await self.call( "search_codebase", {"graph_path": graph_path, "query": query, "limit": limit}, + timeout_s=interactive_call_timeout_s(), ) async def detect_changes( @@ -425,6 +481,7 @@ async def get_impact(self, graph_path: str, qualified_name: str) -> Any: return await self.call( "get_impact", {"graph_path": graph_path, "qualified_name": qualified_name}, + timeout_s=interactive_call_timeout_s(), ) async def analyze_codebase( diff --git a/mcp_server/infrastructure/mcp_call_timeout.py b/mcp_server/infrastructure/mcp_call_timeout.py index 4d2218a5..5e38d393 100644 --- a/mcp_server/infrastructure/mcp_call_timeout.py +++ b/mcp_server/infrastructure/mcp_call_timeout.py @@ -29,6 +29,23 @@ _DEFAULT_CALL_TIMEOUT_S = 600.0 _ENV_VAR = "CORTEX_MCP_CALL_TIMEOUT_S" +# Wall-clock ceiling (seconds) for INTERACTIVE AP read-path calls +# (search_codebase, get_symbol, get_context, get_impact, get_processes, +# query_graph, health_check). Unlike indexing, a read/lookup is not +# legitimately long-running: an AP that connects but then wedges on such a +# call must degrade to graceful Cortex-only results, not stall the tool. +# The unbounded wedge window above (600s of SILENCE) is reserved for +# ingestion and is far too slow here — unified_search would hang for up to +# 10 minutes before falling back. (get_causal_chain has no AP dependency — +# it is pure knowledge-graph BFS over MemoryStore — and cannot hang on AP; +# do not conflate the two.) +# source: interactive read-path ceiling. AP read tools are documented +# interactive (unified_search target <200ms, docs/mcp-tools.md); 30s is a +# wide margin over that interactive target yet 20x below the 600s indexing +# wedge window, so a wedged AP degrades in seconds instead of minutes. +_DEFAULT_INTERACTIVE_CALL_TIMEOUT_S = 30.0 +_INTERACTIVE_ENV_VAR = "CORTEX_AP_INTERACTIVE_TIMEOUT_S" + def default_call_timeout_s() -> float: """Return the configured wedge silence window in seconds. @@ -48,3 +65,22 @@ def default_call_timeout_s() -> float: except (TypeError, ValueError): pass return _DEFAULT_CALL_TIMEOUT_S + + +def interactive_call_timeout_s() -> float: + """Return the wall-clock ceiling for interactive AP read-path calls. + + Reads ``CORTEX_AP_INTERACTIVE_TIMEOUT_S`` (positive float) when set and + valid; otherwise returns the documented default. A non-positive or + malformed override falls back to the default — an unbounded interactive + call is exactly the hang this ceiling exists to prevent. + """ + raw = os.environ.get(_INTERACTIVE_ENV_VAR) + if raw: + try: + val = float(raw) + if val > 0: + return val + except (TypeError, ValueError): + pass + return _DEFAULT_INTERACTIVE_CALL_TIMEOUT_S diff --git a/mcp_server/infrastructure/workflow_graph_ast_response.py b/mcp_server/infrastructure/workflow_graph_ast_response.py index a2810cfb..ec583d38 100644 --- a/mcp_server/infrastructure/workflow_graph_ast_response.py +++ b/mcp_server/infrastructure/workflow_graph_ast_response.py @@ -72,6 +72,35 @@ def _from_legacy_shape(payload: dict) -> list[dict]: return [r for r in inner if isinstance(r, dict)] +def normalize_search_hits(resp: Any) -> list[dict[str, Any]]: + """Normalize a raw AP ``search_codebase`` response into + ``[{id, qualified_name, file_path, score, snippet, source}, ...]``. + + Split out of ``workflow_graph_source_ast.WorkflowGraphASTSource + .search_codebase`` (over the 300-line file cap) — this module already + owns "normalize an AP response shape", the same seam. Rows with no + ``qualified_name`` are dropped; ``id`` is deterministic so RRF fusion + can dedupe with the same scheme used for SYMBOL graph nodes. + """ + out: list[dict[str, Any]] = [] + for r in as_list(resp): + qname = r.get("qualified_name") or r.get("name") or "" + fpath = r.get("file_path") or r.get("abs_path") or "" + if not qname: + continue + out.append( + { + "id": f"symbol:{fpath}::{qname}", + "qualified_name": str(qname), + "file_path": str(fpath), + "score": float(r.get("score") or 0.0), + "snippet": r.get("snippet") or r.get("signature") or "", + "source": "ap", + } + ) + return out + + def build_path_tails(paths: list[str]) -> set[str]: """Expand each path into itself + every ``/``-boundary suffix ("tail"). @@ -97,4 +126,4 @@ def build_path_tails(paths: list[str]) -> set[str]: return path_tails -__all__ = ["as_list", "build_path_tails"] +__all__ = ["as_list", "build_path_tails", "normalize_search_hits"] diff --git a/mcp_server/infrastructure/workflow_graph_source_ast.py b/mcp_server/infrastructure/workflow_graph_source_ast.py index 3cc00134..d1494eef 100644 --- a/mcp_server/infrastructure/workflow_graph_source_ast.py +++ b/mcp_server/infrastructure/workflow_graph_source_ast.py @@ -40,7 +40,7 @@ _SyncLoop, ) from mcp_server.infrastructure.workflow_graph_ast_edges import edge_batches_async -from mcp_server.infrastructure.workflow_graph_ast_response import as_list +from mcp_server.infrastructure.workflow_graph_ast_response import normalize_search_hits from mcp_server.infrastructure.workflow_graph_ast_symbols import ( _SYMBOL_LABELS, symbol_batches_async, @@ -61,6 +61,13 @@ def __init__(self, bridge: APBridge | None = None) -> None: def enabled(self) -> bool: return is_enabled() + @property + def last_search_degraded_reason(self) -> str | None: + """Why the last ``search_codebase`` call degraded, or ``None`` if + it succeeded or was never attempted. Reads the bridge's own + recorded outcome — no second round-trip.""" + return self._bridge.unavailable_reason + def close(self) -> None: """Close the underlying bridge + pinned loop. Idempotent.""" try: @@ -211,10 +218,10 @@ def search_codebase( """Forward ``search_codebase`` to AP and normalize to a flat list of ``{id, qualified_name, file_path, score, snippet}``. - Phase 3 (ADR-0046). When AP is disabled OR no graph_path is - configured, returns ``[]`` so the unified-search fusion - gracefully falls back to Cortex-only results. - """ + Phase 3 (ADR-0046). Returns ``[]`` when AP is disabled, no + graph_path is configured, or the call itself failed — check + ``last_search_degraded_reason`` afterwards to tell "AP found + nothing" (``None``) apart from "AP call failed" (set).""" if not is_enabled() or not query or not query.strip(): return [] gp = resolve_graph_path() @@ -223,25 +230,7 @@ def search_codebase( resp = self._loop_owner.run( self._bridge.search_codebase(gp, query, limit=int(limit)) ) - out: list[dict[str, Any]] = [] - for r in as_list(resp): - qname = r.get("qualified_name") or r.get("name") or "" - fpath = r.get("file_path") or r.get("abs_path") or "" - if not qname: - continue - out.append( - { - # Deterministic id so RRF fusion can dedupe with - # the same scheme used for SYMBOL graph nodes. - "id": f"symbol:{fpath}::{qname}", - "qualified_name": str(qname), - "file_path": str(fpath), - "score": float(r.get("score") or 0.0), - "snippet": r.get("snippet") or r.get("signature") or "", - "source": "ap", - } - ) - return out + return normalize_search_hits(resp) def verify_symbols(self, qualnames: list[str]) -> dict[str, bool]: """Return ``{qualname: exists_in_ap}`` for each candidate. diff --git a/tests_py/handlers/test_unified_search_degraded_status.py b/tests_py/handlers/test_unified_search_degraded_status.py new file mode 100644 index 00000000..caf33b92 --- /dev/null +++ b/tests_py/handlers/test_unified_search_degraded_status.py @@ -0,0 +1,95 @@ +"""Tests for ``unified_search.handler``'s degraded-status signal (PR #449). + +Bug: ``status`` was derived from the static ``is_enabled()`` config flag +only. An AP that is enabled but times out/errors on the actual +``search_codebase`` call returned ``[]`` for ``ap_hits`` — the response was +``status="ok", sources=["cortex","ap"], counts.ap=0``, indistinguishable +from AP genuinely finding nothing (silent data loss the PR's own body +claims to have fixed for the *timeout* path but the *response* never +surfaced). + +Fix: the handler reads ``WorkflowGraphASTSource.last_search_degraded_reason`` +after the call and sets ``status="partial"`` + a ``degraded`` field naming +the source and reason whenever AP was attempted but failed. +""" + +from __future__ import annotations + +import asyncio + +from mcp_server.handlers import unified_search + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +async def _fake_recall_empty(_args: dict) -> dict: + return {"memories": []} + + +class _WedgedASTSource: + """Stands in for a AST source whose bridge just timed out: no hits, + but a recorded degrade reason — the exact silent-data-loss shape.""" + + def __init__(self) -> None: + self.last_search_degraded_reason: str | None = None + + def search_codebase(self, query: str, *, limit: int = 20) -> list[dict]: + self.last_search_degraded_reason = ( + "TimeoutError: AP call search_codebase exceeded 30s" + ) + return [] + + +class _HealthyASTSource: + """AP genuinely ran and found nothing — the case that must stay + status=ok, degraded=None (never conflated with a call failure).""" + + def __init__(self) -> None: + self.last_search_degraded_reason: str | None = None + + def search_codebase(self, query: str, *, limit: int = 20) -> list[dict]: + return [] + + +def test_wedged_ap_call_surfaces_partial_status_and_degraded_reason( + monkeypatch, +) -> None: + monkeypatch.setattr(unified_search, "recall_handler", _fake_recall_empty) + monkeypatch.setattr(unified_search, "is_enabled", lambda: True) + monkeypatch.setattr(unified_search, "WorkflowGraphASTSource", _WedgedASTSource) + + resp = _run(unified_search.handler({"query": "anything"})) + + # This is the assertion that fails on the pre-fix handler: status="ok" + # and degraded absent/None even though AP never actually returned. + assert resp["status"] == "partial" + assert resp["degraded"] is not None + assert resp["degraded"]["source"] == "ap" + assert "TimeoutError" in resp["degraded"]["reason"] + assert resp["counts"]["ap"] == 0 + assert resp["sources"] == ["cortex", "ap"] + + +def test_ap_genuinely_empty_stays_ok_with_no_degraded_flag(monkeypatch) -> None: + monkeypatch.setattr(unified_search, "recall_handler", _fake_recall_empty) + monkeypatch.setattr(unified_search, "is_enabled", lambda: True) + monkeypatch.setattr(unified_search, "WorkflowGraphASTSource", _HealthyASTSource) + + resp = _run(unified_search.handler({"query": "anything"})) + + assert resp["status"] == "ok" + assert resp["degraded"] is None + assert resp["counts"]["ap"] == 0 + + +def test_ap_disabled_stays_partial_with_no_degraded_flag(monkeypatch) -> None: + monkeypatch.setattr(unified_search, "recall_handler", _fake_recall_empty) + monkeypatch.setattr(unified_search, "is_enabled", lambda: False) + + resp = _run(unified_search.handler({"query": "anything"})) + + assert resp["status"] == "partial" + assert resp["degraded"] is None + assert resp["sources"] == ["cortex"] diff --git a/tests_py/infrastructure/test_ap_bridge_interactive_timeout.py b/tests_py/infrastructure/test_ap_bridge_interactive_timeout.py new file mode 100644 index 00000000..144b2e29 --- /dev/null +++ b/tests_py/infrastructure/test_ap_bridge_interactive_timeout.py @@ -0,0 +1,158 @@ +"""Tests for the interactive read-path timeout on ``APBridge.call``. + +Bug: the AP client runs with ``callTimeoutMs=0`` (indexing may legitimately +exceed any fixed bound), so a connected-but-wedged AP had only the 600s +wedge-silence window as a backstop. Interactive read-path calls +(search_codebase, get_symbol, …) inherited that 600s, so unified_search +would hang for up to 10 minutes instead of degrading to Cortex-only. +(get_causal_chain has no AP dependency and is unaffected.) + +Fix: ``call(tool, args, timeout_s=...)`` wraps the client call in +``asyncio.wait_for`` and degrades a timeout to ``None`` (→ Cortex-only). The +interactive wrappers pass ``interactive_call_timeout_s()``; indexing wrappers +stay unbounded. These tests use a fake client so no subprocess is spawned. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from mcp_server.infrastructure.ap_bridge import APBridge +from mcp_server.infrastructure.mcp_call_timeout import interactive_call_timeout_s + + +class _HangingClient: + """Stands in for a connected AP whose call never returns.""" + + connected = True + + async def call(self, name, args): # noqa: ANN001 — test double + await asyncio.Event().wait() # blocks forever + + +class _SlowClient: + """Returns after ``delay`` seconds — used to prove the ceiling fires + below the delay and passes above it.""" + + connected = True + + def __init__(self, delay: float, payload) -> None: + self._delay = delay + self._payload = payload + + async def call(self, name, args): # noqa: ANN001 — test double + await asyncio.sleep(self._delay) + return self._payload + + +def _bridge_with(client) -> APBridge: + bridge = APBridge() + bridge._client = client + bridge._connected = True + return bridge + + +def test_interactive_timeout_degrades_to_none() -> None: + bridge = _bridge_with(_HangingClient()) + + async def run(): + # A tiny explicit ceiling: the hanging client would otherwise block + # forever; the bridge must return None well within the test. + return await bridge.call("search_codebase", {"query": "x"}, timeout_s=0.05) + + result = asyncio.run(run()) + assert result is None + assert "TimeoutError" in (bridge._unavailable_reason or "") + assert "search_codebase" in (bridge._unavailable_reason or "") + + +def test_call_under_ceiling_returns_payload() -> None: + bridge = _bridge_with(_SlowClient(delay=0.0, payload={"rows": [], "status": "ok"})) + + async def run(): + return await bridge.call("get_symbol", {"qualified_name": "f"}, timeout_s=5.0) + + assert asyncio.run(run()) == {"rows": [], "status": "ok"} + + +def test_no_timeout_argument_is_unbounded() -> None: + # Indexing wrappers pass no timeout_s; the call must NOT be wrapped in + # wait_for. A client that returns promptly proves the unbounded path + # still returns its payload (regression guard on the None-branch). + bridge = _bridge_with(_SlowClient(delay=0.0, payload={"ok": True})) + + async def run(): + return await bridge.call("index_codebase", {"path": "/x"}) + + assert asyncio.run(run()) == {"ok": True} + + +def test_search_codebase_wrapper_passes_interactive_ceiling(monkeypatch) -> None: + # The interactive wrapper must forward a positive, bounded timeout_s. + captured: dict[str, float | None] = {} + + async def fake_call(self, tool, args=None, *, timeout_s=None): # noqa: ANN001 + captured["tool"] = tool + captured["timeout_s"] = timeout_s + return {"rows": [], "status": "ok"} + + monkeypatch.setattr(APBridge, "call", fake_call) + bridge = APBridge() + + asyncio.run(bridge.search_codebase("/graph", "query", limit=5)) + assert captured["tool"] == "search_codebase" + assert captured["timeout_s"] == interactive_call_timeout_s() + assert captured["timeout_s"] > 0 + + +def test_index_codebase_wrapper_stays_unbounded(monkeypatch) -> None: + # The indexing wrapper must NOT forward a timeout (unbounded ingest). + captured: dict[str, float | None] = {} + + async def fake_call(self, tool, args=None, *, timeout_s=None): # noqa: ANN001 + captured["tool"] = tool + captured["timeout_s"] = timeout_s + return {"graph_path": "/g"} + + monkeypatch.setattr(APBridge, "call", fake_call) + bridge = APBridge() + + asyncio.run(bridge.index_codebase("/src", output_dir="/out")) + assert captured["tool"] == "index_codebase" + assert captured["timeout_s"] is None + + +def test_interactive_ceiling_default_is_bounded_and_env_overridable( + monkeypatch, +) -> None: + from mcp_server.infrastructure.mcp_call_timeout import default_call_timeout_s + + monkeypatch.delenv("CORTEX_AP_INTERACTIVE_TIMEOUT_S", raising=False) + default = interactive_call_timeout_s() + # Interactive ceiling must be positive and strictly below the 600s + # indexing wedge-silence window — the whole point of the fix. + assert 0 < default < default_call_timeout_s() + + monkeypatch.setenv("CORTEX_AP_INTERACTIVE_TIMEOUT_S", "12.5") + assert interactive_call_timeout_s() == 12.5 + + # Malformed / non-positive overrides fall back to the default, never + # to an unbounded wait. + monkeypatch.setenv("CORTEX_AP_INTERACTIVE_TIMEOUT_S", "0") + assert interactive_call_timeout_s() == default + monkeypatch.setenv("CORTEX_AP_INTERACTIVE_TIMEOUT_S", "not-a-number") + assert interactive_call_timeout_s() == default + + +@pytest.mark.parametrize("tool", ["search_codebase", "get_symbol", "get_context"]) +def test_named_interactive_tools_degrade_not_hang(tool: str) -> None: + # Sanity: the bridge.call ceiling applies uniformly to the interactive + # read tools by name. + bridge = _bridge_with(_HangingClient()) + + async def run(): + return await bridge.call(tool, {}, timeout_s=0.05) + + assert asyncio.run(run()) is None diff --git a/tests_py/infrastructure/test_ap_bridge_reason_reset_on_reuse.py b/tests_py/infrastructure/test_ap_bridge_reason_reset_on_reuse.py new file mode 100644 index 00000000..3df6a765 --- /dev/null +++ b/tests_py/infrastructure/test_ap_bridge_reason_reset_on_reuse.py @@ -0,0 +1,63 @@ +"""Regression test (PR #449 review round 3): ``APBridge._unavailable_reason`` +must not leak a stale reason from an earlier failed call into a later +successful one on the same, reused instance. + +Bug: the reason was only cleared in ``connect()``'s slow (reconnect) path; +``connect()``'s fast path (already-connected client) skipped the clear, and +``call()``'s success branch never reset it either. A caller that reuses one +``WorkflowGraphASTSource``/``APBridge`` across a loop (the codebase already +does this — ``wiki_verify.py`` reuses one source across its candidate loop) +would see ``last_search_degraded_reason`` report a call as failed when it +had actually just succeeded, contradicting its own docstring ("None if it +succeeded"). + +Fix: ``call()`` resets ``self._unavailable_reason = None`` at its top (after +the allowlist check, before touching ``connect()``/the RPC), so each call's +own outcome is authoritative regardless of prior calls on the same instance. +""" + +from __future__ import annotations + +import asyncio + +from mcp_server.infrastructure.ap_bridge import APBridge + + +class _HangingClient: + """Never completes on its own — only a ceiling can terminate the call.""" + + connected = True + + async def call(self, name, args): # noqa: ANN001 — test double + await asyncio.Event().wait() + + +class _OkClient: + """Always succeeds.""" + + connected = True + + async def call(self, name, args): # noqa: ANN001 — test double + return {"rows": [], "status": "ok"} + + +def test_second_successful_call_clears_a_prior_failure_reason() -> None: + bridge = APBridge() + bridge._client = _HangingClient() + bridge._connected = True + + async def run(): + first = await bridge.call("search_codebase", {"query": "x"}, timeout_s=0.05) + assert first is None + assert bridge.unavailable_reason is not None # first call genuinely failed + + # Same instance, reused (matches wiki_verify.py's loop pattern) — swap + # in a client that succeeds and call again, with NO explicit clear. + bridge._client = _OkClient() + return await bridge.call("search_codebase", {"query": "y"}, timeout_s=5.0) + + second = asyncio.run(run()) + assert second == {"rows": [], "status": "ok"} + # The stale reason from the FIRST call must not survive into the SECOND, + # successful one. + assert bridge.unavailable_reason is None diff --git a/tests_py/infrastructure/test_workflow_graph_ast_search_degraded.py b/tests_py/infrastructure/test_workflow_graph_ast_search_degraded.py new file mode 100644 index 00000000..8b80a4e2 --- /dev/null +++ b/tests_py/infrastructure/test_workflow_graph_ast_search_degraded.py @@ -0,0 +1,88 @@ +"""Tests for ``WorkflowGraphASTSource.last_search_degraded_reason``. + +Bug (PR #449 review): ``unified_search`` derived ``status`` from the static +``is_enabled()`` config flag only. An AP that is enabled but wedged/timed +out on a ``search_codebase`` call returned ``[]`` — byte-for-byte +indistinguishable, at the handler, from "AP genuinely found nothing". + +Fix: the AST-source layer exposes ``last_search_degraded_reason``, reading +the same ``APBridge._unavailable_reason`` the bridge's own timeout/exception +handling already records (see ``ap_bridge.py::APBridge._degrade``) — no +second round-trip, no widened return type. + +Uses a real ``APBridge`` wired to a hanging fake client (never completes on +its own — ``asyncio.Event().wait()``) so the interactive ceiling is what +terminates the call, matching the model in +``test_ap_bridge_interactive_timeout.py``. +""" + +from __future__ import annotations + +import asyncio + +from mcp_server.infrastructure import workflow_graph_source_ast as mod +from mcp_server.infrastructure.ap_bridge import APBridge +from mcp_server.infrastructure.workflow_graph_source_ast import ( + WorkflowGraphASTSource, +) + + +class _HangingClient: + """Stands in for a connected-but-wedged AP: the call never completes + on its own, so only the interactive ceiling can terminate it.""" + + connected = True + + async def call(self, name, args): # noqa: ANN001 — test double + await asyncio.Event().wait() + + +def _wedged_bridge() -> APBridge: + bridge = APBridge() + bridge._client = _HangingClient() + bridge._connected = True + return bridge + + +def test_search_codebase_on_wedged_bridge_returns_empty_and_sets_reason( + monkeypatch, +) -> None: + # Tiny ceiling: the hanging client would otherwise block the test + # forever. The ceiling firing is the behavior under test, not an + # incidental wall-clock assertion — same model as + # test_ap_bridge_interactive_timeout.py. + monkeypatch.setenv("CORTEX_AP_INTERACTIVE_TIMEOUT_S", "0.05") + monkeypatch.setattr(mod, "is_enabled", lambda: True) + monkeypatch.setattr(mod, "resolve_graph_path", lambda: "/fake/graph.kuzu") + + source = WorkflowGraphASTSource(bridge=_wedged_bridge()) + try: + # Before any call: never attempted, so no degrade to report. + assert source.last_search_degraded_reason is None + + hits = source.search_codebase("query", limit=5) + + # The silent-data-loss bug: hits alone cannot distinguish "AP + # found nothing" from "AP call failed" — this is why the caller + # (unified_search) must additionally check the reason below. + assert hits == [] + reason = source.last_search_degraded_reason + assert reason is not None + assert "TimeoutError" in reason + assert "search_codebase" in reason + finally: + source.close() + + +def test_search_codebase_disabled_leaves_reason_none(monkeypatch) -> None: + # AP disabled: search_codebase short-circuits before ever touching the + # bridge. This is the legitimate "AP not in play" contract, not a call + # failure — last_search_degraded_reason must stay None. + monkeypatch.setattr(mod, "is_enabled", lambda: False) + + source = WorkflowGraphASTSource(bridge=_wedged_bridge()) + try: + assert source.search_codebase("query", limit=5) == [] + assert source.last_search_degraded_reason is None + finally: + source.close()