route_plan + client: device_grep + AXL anti-pattern error hints

Two complementary additions from cucx-docs's prompt-suggestions handoff
(see axl/agent-threads/cucx-prompt-suggestions/ for the source thread).

device_grep(pattern, classes=None) — fuzzy device discovery by name OR
description, optionally filtered by tkclass.name. Surfaces "wait, there
are TWO of these?" findings (parallel fax servers, duplicate CUBEs,
vestigial conference bridges) by grouping matches by class so the
structure of what matched is visible at a glance. CUCM-style % wildcards
work; case-insensitive matching via UPPER(); single quotes properly
escaped via _esc.

axl_sql error hints — when AXL returns an error AND the query contains
the trigger phrase, append a path-correction hint to the error message.
Two patterns shipped:

  - "Column (fkdevice) not found" + numplan in query → suggest the
    devicenumplanmap M:N join (the literal multi-attempt schema-discovery
    experience cucx-docs hit at Bingham — numplan has no direct fkdevice)

  - "not in database" + sipdestination in query → suggest sipdestinationgroup
    + sipprofile + axl_list_tables(pattern='sip%') for discovery (the
    `sipdestination` table is reasonable-sounding but doesn't exist)

Hints are surgical (both error fragment AND query trigger must match)
to keep false-positive risk near zero. Validator behavior unchanged —
this is post-execution error augmentation, not gate enhancement. Failing
queries now raise RuntimeError(augmented) when a hint applies; otherwise
the original exception passes through unchanged.

Tests: +19 (8 device_grep + 11 error-hints with end-to-end mock through
execute_sql_query). Full suite 219 → 238 passing.

Live-cluster smoke test still pending (TLS handshake intermittent
this session). Sequencing nit from cucx-docs's msg 003 (move error-hint
earlier) honored — bundled with device_grep in this single commit.

src/mcaxl/client.py

@@ -254,7 +254,19 @@ class AxlClient:
         if cached is not None:
             return {**cached, "_cache": "hit"}
         self._ensure_connected()
+        try:
             resp = self._service.executeSQLQuery(sql=cleaned)
+        except Exception as e:
+            # Anti-pattern hints — if the AXL error matches a known operator
+            # mistake (numplan.fkdevice, FROM sipdestination, etc.), append
+            # the right-path hint to the error message. Saves the 3-5 turns
+            # an LLM would otherwise spend rediscovering the schema. See
+            # axl/agent-threads/cucx-prompt-suggestions/ for the source
+            # observations that motivated each hint.
+            augmented = _augment_axl_error(str(e), cleaned)
+            if augmented != str(e):
+                raise RuntimeError(augmented) from e
+            raise
         rows = _parse_sql_rows(resp)
         result = {"row_count": len(rows), "rows": rows, "query": cleaned}
         self._response_cache.set("executeSQLQuery", {"sql": cleaned}, result)
@@ -431,3 +443,52 @@ def _stringify(v: Any) -> Any:
     if v is None or isinstance(v, (str, int, float, bool)):
         return v
     return str(v)
+
+
+# ─── AXL anti-pattern error hints ──────────────────────────────────────
+#
+# When AXL returns a specific class of error AND the query contains the
+# trigger phrase, append a hint to the error explaining the right path.
+# These are surgical (not fuzzy) because targeted hints have lower false-
+# positive risk than generic "did you mean ..." suggestions.
+#
+# Source: cucx-docs handoff (msg 003 in
+# axl/agent-threads/cucx-prompt-suggestions/) — three patterns recurring
+# enough during cucx-docs CUCM 15 audits to warrant a structural fix.
+
+_AXL_ERROR_HINTS: list[dict[str, str]] = [
+    {
+        "error_fragment": "Column (fkdevice) not found",
+        "query_must_contain": "numplan",
+        "hint": (
+            "Hint: numplan has no fkdevice column. The numplan-to-device "
+            "link is M:N through devicenumplanmap. Try:\n"
+            "  JOIN devicenumplanmap m ON m.fknumplan = numplan.pkid\n"
+            "  JOIN device d ON m.fkdevice = d.pkid"
+        ),
+    },
+    {
+        "error_fragment": "not in database",
+        "query_must_contain": "sipdestination",
+        "hint": (
+            "Hint: `sipdestination` does not exist as a table. SIP trunk "
+            "destinations live on `device` joined with `sipdestinationgroup` "
+            "and `sipprofile`. Try `axl_list_tables(pattern='sip%')` to see "
+            "the actual SIP-related tables."
+        ),
+    },
+]
+
+
+def _augment_axl_error(error_text: str, query: str) -> str:
+    """Append a hint to the AXL error message if any anti-pattern matches.
+
+    Returns `error_text` unchanged when no hint applies — caller can compare
+    identity to decide whether to wrap the original exception.
+    """
+    upper_query = query.upper()
+    for entry in _AXL_ERROR_HINTS:
+        if (entry["error_fragment"] in error_text
+                and entry["query_must_contain"].upper() in upper_query):
+            return f"{error_text}\n\n{entry['hint']}"
+    return error_text

src/mcaxl/route_plan.py

@@ -261,6 +261,77 @@ def inspect_pattern(
     }
 
 
+def device_grep(
+    client: "AxlClient",
+    pattern: str,
+    classes: list[str] | None = None,
+) -> dict:
+    """Fuzzy device discovery — match `pattern` against name OR description,
+    optionally filtered by device class.
+
+    Surfaces the kind of "wait, there are TWO of these?" findings that
+    cucx-docs hit at Bingham (parallel ZetaFax + RightFax fax servers, etc.):
+    devices whose name or description contains a vendor/role keyword,
+    grouped by class so trunks-vs-phones-vs-route-lists is visible at a
+    glance.
+
+    Args:
+      pattern: substring to match (case-insensitive) against `device.name`
+        OR `device.description`. CUCM-style `%` wildcards work — pass
+        `"FAX"` for substring, `"FAX%"` for prefix-only, etc.
+      classes: optional `tkclass.name` filter — e.g.
+        `["Phone", "Trunk"]`, `["Route List"]`, `["Gateway", "H323 Gateway"]`.
+        If None, all classes returned.
+
+    Returns:
+      `{pattern, match_count, groups: {class_name: [{name, description, type, pool}, ...]}}`
+    """
+    if not pattern or not pattern.strip():
+        raise ValueError("pattern must be a non-empty string")
+
+    safe_pat = _esc(pattern)
+    class_filter = ""
+    if classes:
+        escaped = ", ".join(f"'{_esc(c)}'" for c in classes)
+        class_filter = f"AND tc.name IN ({escaped})"
+
+    sql = f"""
+    SELECT
+      d.name,
+      d.description,
+      tc.name AS class_name,
+      dt.name AS device_type,
+      dp.name AS pool_name
+    FROM device d
+    LEFT OUTER JOIN typeclass tc ON d.tkclass = tc.enum
+    LEFT OUTER JOIN typemodel dt ON d.tkmodel = dt.enum
+    LEFT OUTER JOIN devicepool dp ON d.fkdevicepool = dp.pkid
+    WHERE (UPPER(d.name) LIKE UPPER('%{safe_pat}%')
+        OR UPPER(d.description) LIKE UPPER('%{safe_pat}%'))
+      {class_filter}
+    ORDER BY tc.name, d.name
+    """
+    result = client.execute_sql_query(sql)
+    rows = result["rows"]
+
+    groups: dict[str, list] = {}
+    for row in rows:
+        cls = row.get("class_name") or "Unknown"
+        groups.setdefault(cls, []).append({
+            "name": row.get("name"),
+            "description": row.get("description"),
+            "type": row.get("device_type"),
+            "pool": row.get("pool_name"),
+        })
+
+    return {
+        "pattern": pattern,
+        "classes_filter": classes,
+        "match_count": len(rows),
+        "groups": groups,
+    }
+
+
 def patterns_targeting_device(
     client: "AxlClient",
     device_name: str,

src/mcaxl/server.py

@@ -239,6 +239,29 @@ def route_inspect_pattern(pattern: str, partition: str | None = None) -> dict:
     return route_plan.inspect_pattern(_client(), pattern, partition)
 
 
+@mcp.tool
+def device_grep(
+    pattern: str,
+    classes: list[str] | None = None,
+) -> dict:
+    """Fuzzy device discovery — match `pattern` against name OR description,
+    optionally filtered by device class.
+
+    Surfaces "wait, there are TWO of these?" findings — vendor systems
+    deployed twice and forgotten once (parallel fax servers, duplicate
+    CUBEs, vestigial conference bridges, etc.). Grouped by class so the
+    structure of what matched is visible at a glance.
+
+    Args:
+        pattern: substring (case-insensitive) matched against device name
+            OR description. CUCM-style `%` wildcards work — `"FAX"` is
+            substring, `"FAX%"` is prefix-only.
+        classes: optional list of `tkclass.name` values to filter — e.g.
+            `["Phone", "Trunk"]`, `["Route List"]`, `["Gateway", "H323 Gateway"]`.
+    """
+    return route_plan.device_grep(_client(), pattern, classes)
+
+
 @mcp.tool
 def route_patterns_targeting(
     device_name: str,

tests/test_axl_error_hints.py

@@ -0,0 +1,126 @@
+"""Tests for the AXL anti-pattern error-hint enhancement.
+
+Source: cucx-docs handoff (msg 003 in
+axl/agent-threads/cucx-prompt-suggestions/) — three recurring operator
+mistakes with cluster-side error messages that don't suggest the right
+path. Hints are surgical: only fire when both the error fragment AND
+the query trigger phrase match.
+"""
+
+import pytest
+
+from mcaxl.client import _augment_axl_error
+
+
+class TestNumplanFkdeviceHint:
+    def test_fires_on_matching_error_and_query(self):
+        err = "Column (fkdevice) not found in any table in the query (or SLV is undefined)."
+        query = "SELECT name FROM numplan WHERE fkdevice = 'foo'"
+        out = _augment_axl_error(err, query)
+        assert err in out
+        assert "devicenumplanmap" in out
+        assert "M:N" in out
+
+    def test_no_fire_on_error_alone(self):
+        # Same error fragment but query doesn't mention numplan
+        err = "Column (fkdevice) not found in any table in the query."
+        query = "SELECT name FROM device WHERE fkdevice = 'foo'"
+        assert _augment_axl_error(err, query) == err
+
+    def test_no_fire_on_unrelated_error(self):
+        err = "Some completely different error about a different column."
+        query = "SELECT name FROM numplan"
+        assert _augment_axl_error(err, query) == err
+
+    def test_case_insensitive_query_match(self):
+        # User's query in mixed case should still trigger
+        err = "Column (fkdevice) not found in any table in the query."
+        query = "SELECT name FROM NumPlan np WHERE np.fkdevice IS NOT NULL"
+        out = _augment_axl_error(err, query)
+        assert "devicenumplanmap" in out
+
+
+class TestSipDestinationHint:
+    def test_fires_on_matching_error_and_query(self):
+        err = "Table (sipdestination) is not in database."
+        query = "SELECT * FROM sipdestination"
+        out = _augment_axl_error(err, query)
+        assert "sipdestinationgroup" in out
+        assert "axl_list_tables" in out
+
+    def test_no_fire_when_query_doesnt_reference_table(self):
+        err = "Table (xyz) is not in database."
+        query = "SELECT * FROM xyz"
+        assert _augment_axl_error(err, query) == err
+
+
+class TestNoMatch:
+    def test_unrelated_error_returns_unchanged(self):
+        err = "Permission denied for user 'CCMSysUser'."
+        query = "SELECT * FROM device"
+        assert _augment_axl_error(err, query) == err
+
+    def test_empty_error_returns_empty(self):
+        assert _augment_axl_error("", "SELECT 1") == ""
+
+    def test_identity_check_for_no_hint(self):
+        """Caller compares identity to decide whether to wrap the original
+        exception. Make sure no-hint path returns the literal input."""
+        err = "Random AXL error"
+        query = "SELECT 1"
+        out = _augment_axl_error(err, query)
+        assert out is err  # not just equal — same object
+
+
+class TestEndToEnd:
+    """Confirm the augmentation propagates through execute_sql_query.
+
+    Uses a mock that raises the AXL exception from inside zeep's call;
+    the augmenter should wrap it as a RuntimeError with the hint
+    appended.
+    """
+
+    def test_execute_sql_query_wraps_with_hint(self):
+        from unittest.mock import MagicMock
+        from mcaxl.client import AxlClient
+        from mcaxl.cache import AxlCache
+
+        # Real cache, but in a temp location so tests don't pollute
+        import tempfile
+        from pathlib import Path
+        with tempfile.TemporaryDirectory() as td:
+            cache = AxlCache(Path(td) / "test.sqlite", default_ttl=0, cluster_id="test")
+            client = AxlClient(cache)
+            # Bypass _ensure_connected by setting service directly
+            mock_service = MagicMock()
+            mock_service.executeSQLQuery.side_effect = RuntimeError(
+                "Server raised fault: Column (fkdevice) not found in any table"
+            )
+            client._service = mock_service
+            client._connected_at = 0.0
+
+            with pytest.raises(RuntimeError, match="devicenumplanmap"):
+                client.execute_sql_query(
+                    "SELECT name FROM numplan WHERE fkdevice IS NOT NULL"
+                )
+
+    def test_execute_sql_query_passes_through_unrelated_error(self):
+        from unittest.mock import MagicMock
+        from mcaxl.client import AxlClient
+        from mcaxl.cache import AxlCache
+
+        import tempfile
+        from pathlib import Path
+        with tempfile.TemporaryDirectory() as td:
+            cache = AxlCache(Path(td) / "test.sqlite", default_ttl=0, cluster_id="test")
+            client = AxlClient(cache)
+            mock_service = MagicMock()
+            mock_service.executeSQLQuery.side_effect = RuntimeError(
+                "Some unrelated cluster error"
+            )
+            client._service = mock_service
+            client._connected_at = 0.0
+
+            # Original exception type + message preserved when no hint applies
+            with pytest.raises(RuntimeError, match="Some unrelated cluster error"):
+                client.execute_sql_query("SELECT * FROM device")

tests/test_device_grep.py

@@ -0,0 +1,94 @@
+"""Tests for device_grep — fuzzy device discovery by name/description.
+
+The "wait there are TWO of these?" finding shape from cucx-docs's
+ZetaFax-vs-RightFax discovery (msg 001 in
+axl/agent-threads/cucx-prompt-suggestions/).
+"""
+
+import pytest
+
+from mcaxl.route_plan import device_grep
+
+
+class FakeAxlClient:
+    def __init__(self, rows):
+        self._rows = rows
+        self.queries = []
+
+    def execute_sql_query(self, sql):
+        self.queries.append(sql)
+        return {"row_count": len(self._rows), "rows": self._rows}
+
+
+def _make_row(name, description, class_name, device_type="SIP Trunk", pool="DP-1"):
+    return {
+        "name": name,
+        "description": description,
+        "class_name": class_name,
+        "device_type": device_type,
+        "pool_name": pool,
+    }
+
+
+class TestDeviceGrepBasics:
+    def test_groups_by_class(self):
+        rows = [
+            _make_row("RightFax-Trunk", "RightFax inbound", "Trunk"),
+            _make_row("ZetaFax-Trunk", "ZetaFax internal", "Trunk"),
+            _make_row("SEPABCDFAXX01", "RightFax desk test", "Phone", "Cisco 8841"),
+            _make_row("RightFax-RL", "RightFax route list", "Route List"),
+        ]
+        client = FakeAxlClient(rows)
+        result = device_grep(client, "FAX")
+        assert result["match_count"] == 4
+        assert set(result["groups"].keys()) == {"Trunk", "Phone", "Route List"}
+        assert len(result["groups"]["Trunk"]) == 2
+
+    def test_class_filter_passed_to_sql(self):
+        client = FakeAxlClient([])
+        device_grep(client, "FAX", classes=["Trunk", "Route List"])
+        # Both class names appear escaped in the SQL IN clause
+        sql = client.queries[0]
+        assert "tc.name IN" in sql
+        assert "'Trunk'" in sql
+        assert "'Route List'" in sql
+
+    def test_no_class_filter_omits_in_clause(self):
+        client = FakeAxlClient([])
+        device_grep(client, "FAX")
+        assert "tc.name IN" not in client.queries[0]
+
+    def test_empty_pattern_raises(self):
+        client = FakeAxlClient([])
+        with pytest.raises(ValueError, match="non-empty"):
+            device_grep(client, "")
+        with pytest.raises(ValueError, match="non-empty"):
+            device_grep(client, "   ")
+
+    def test_pattern_quote_escaped(self):
+        client = FakeAxlClient([])
+        device_grep(client, "fake'; DROP TABLE device --")
+        # SQL injection via pattern is escaped (doubled single quotes)
+        assert "fake''" in client.queries[0]
+
+    def test_class_filter_quote_escaped(self):
+        client = FakeAxlClient([])
+        device_grep(client, "FAX", classes=["Phone'; DROP TABLE device --"])
+        assert "Phone''" in client.queries[0]
+
+    def test_unknown_class_grouped_separately(self):
+        # If tkclass enum doesn't resolve (LEFT JOIN miss), class_name is NULL
+        rows = [
+            {"name": "WeirdDevice", "description": "?", "class_name": None,
+             "device_type": None, "pool_name": None},
+        ]
+        client = FakeAxlClient(rows)
+        result = device_grep(client, "weird")
+        assert "Unknown" in result["groups"]
+        assert result["groups"]["Unknown"][0]["name"] == "WeirdDevice"
+
+    def test_response_includes_filter_metadata(self):
+        client = FakeAxlClient([])
+        result = device_grep(client, "FAX", classes=["Trunk"])
+        assert result["pattern"] == "FAX"
+        assert result["classes_filter"] == ["Trunk"]