diff --git a/src/mcaxl/client.py b/src/mcaxl/client.py index adb7cb4..098be69 100644 --- a/src/mcaxl/client.py +++ b/src/mcaxl/client.py @@ -254,7 +254,19 @@ class AxlClient: if cached is not None: return {**cached, "_cache": "hit"} self._ensure_connected() - resp = self._service.executeSQLQuery(sql=cleaned) + 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 diff --git a/src/mcaxl/route_plan.py b/src/mcaxl/route_plan.py index 2dc6df4..26ddab8 100644 --- a/src/mcaxl/route_plan.py +++ b/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, diff --git a/src/mcaxl/server.py b/src/mcaxl/server.py index 6f03585..19aed01 100644 --- a/src/mcaxl/server.py +++ b/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, diff --git a/tests/test_axl_error_hints.py b/tests/test_axl_error_hints.py new file mode 100644 index 0000000..fa152ca --- /dev/null +++ b/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") diff --git a/tests/test_device_grep.py b/tests/test_device_grep.py new file mode 100644 index 0000000..835ab0b --- /dev/null +++ b/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"]