fix(ap): bound interactive AP read calls so a wedged pipeline degrades instead of hanging

.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",

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`);

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:<memory_id>`` (added by this handler).
@@ -119,18 +123,33 @@ 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(
         [("cortex", memories), ("ap", ap_hits)],
         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": {

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(

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

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"]

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.