From 4010743eb5e6e77ca46eb775eb3892adff461ce6 Mon Sep 17 00:00:00 2001 From: Nikolay Miroshnichenko Date: Sat, 12 Sep 2026 21:15:55 +0200 Subject: [PATCH] tools: stable ordering + cursor paging for list_group_addresses and get_devices - sort by the address itself (GA main/middle/sub numerically, device area/line/device), so the result no longer depends on parser dict order and a retry returns the same page - return {items, total_matched, returned, next_cursor} instead of a bare list truncated at limit: truncation was silent and a caller could not distinguish a full answer from a cut one - tests/test_paging.py (in CI): numeric vs lexical order, full walk at page sizes 1..100 covers every row exactly once, same page after a reshuffled re-parse, filters reflected in total_matched - verified on a real 3646-GA project; README EN/RU + CHANGELOG updated - prompted by feedback on the r/mcp post (2026-09-12) --- .github/workflows/ci.yml | 3 ++ nickol_knx_mcp/server.py | 100 +++++++++++++++++++++++++++++---------- tests/test_paging.py | 95 +++++++++++++++++++++++++++++++++++++ 3 files changed, 174 insertions(+), 24 deletions(-) create mode 100644 tests/test_paging.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5d24e83..badcf5f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,6 +52,9 @@ jobs: - name: Structure-first entity suggestions (HA SuggestionProvider prototype) run: python tests/test_suggest.py + - name: Stable ordering + cursor paging for large tool results + run: python tests/test_paging.py + - name: Console script is installed run: | python -c "import importlib.metadata as m; print('entry points:', [e.name for e in m.entry_points(group='console_scripts') if e.name == 'nickol-knx-mcp'])" diff --git a/nickol_knx_mcp/server.py b/nickol_knx_mcp/server.py index ecfce51..1b685d0 100644 --- a/nickol_knx_mcp/server.py +++ b/nickol_knx_mcp/server.py @@ -55,6 +55,29 @@ def _project() -> LoadedProject: return p +def _ga_sort_key(address: str) -> tuple: + """Stable ordering for group addresses across styles. + + ThreeLevel "1/2/3" and TwoLevel "1/2" sort numerically per part; free-style "1234" + sorts numerically; anything unparseable sorts last, lexically. Two group addresses + never compare equal unless they are the same address, so a cursor is unambiguous. + """ + parts = (address or "").split("/") + try: + nums = tuple(int(p) for p in parts) + return (0, len(nums), nums, address or "") + except ValueError: + return (1, 0, (), address or "") + + +def _ia_sort_key(address: str) -> tuple: + """Stable ordering for individual addresses "area.line.device".""" + try: + return (0, tuple(int(p) for p in (address or "").split(".")), address or "") + except ValueError: + return (1, (), address or "") + + def _safe_write(rel_or_abs_path: str, content: str) -> str: """Write inside the workspace only. Returns the absolute path written.""" _WORKSPACE.mkdir(parents=True, exist_ok=True) @@ -107,43 +130,72 @@ def load_project(path: str, password: Optional[str] = None, def list_group_addresses(category: Optional[str] = None, kind: Optional[str] = None, missing_dpt_only: bool = False, - limit: int = 500) -> list[dict[str, Any]]: - """List parsed group addresses with classification. + limit: int = 500, + cursor: Optional[str] = None) -> dict[str, Any]: + """List parsed group addresses with classification, in a **stable order**. Filters: category (lighting/shutter/hvac/sensor/scene/energy/diagnostics), kind (command/status/sensor), missing_dpt_only. + + Paging: results are always sorted by the group address itself (main/middle/sub + numerically, free-style addresses numerically, anything else lexically), so the + order does not depend on how the project happened to parse and a retry returns + the same page. Pass the returned `next_cursor` back as `cursor` for the next + page; `next_cursor` is null on the last page. `total_matched` reports how many + addresses match the filters, so a truncated answer is never silent. """ proj = _project() - out = [] - for ga in proj.gas.values(): - if category and ga.category != category: - continue - if kind and ga.kind != kind: - continue - if missing_dpt_only and ga.dpt_main is not None: - continue - out.append({ + rows = [ga for ga in proj.gas.values() + if not (category and ga.category != category) + and not (kind and ga.kind != kind) + and not (missing_dpt_only and ga.dpt_main is not None)] + rows.sort(key=lambda ga: _ga_sort_key(ga.address)) + total = len(rows) + start = 0 + if cursor: + ck = _ga_sort_key(cursor) + start = next((i for i, ga in enumerate(rows) if _ga_sort_key(ga.address) > ck), total) + page = rows[start:start + max(1, limit)] + return { + "group_addresses": [{ "address": ga.address, "name": ga.name, "dpt": ga.dpt, "category": ga.category, "kind": ga.kind, "intent": ga.intent, "ha_platform": ga.ha_platform, "secure": ga.data_secure, "description": ga.description, - }) - if len(out) >= limit: - break - return out + } for ga in page], + "total_matched": total, + "returned": len(page), + "next_cursor": page[-1].address if start + len(page) < total else None, + } @mcp.tool() -def get_devices() -> list[dict[str, Any]]: - """List devices: individual address, name, order number, manufacturer.""" +def get_devices(limit: int = 500, cursor: Optional[str] = None) -> dict[str, Any]: + """List devices (individual address, name, order number, manufacturer), sorted by + individual address (area/line/device numerically). Same paging contract as + `list_group_addresses`: `next_cursor` / `total_matched` / `returned`.""" proj = _project() - return [{ - "individual_address": d.get("individual_address"), - "name": d.get("name"), - "order_number": d.get("order_number"), - "manufacturer": d.get("manufacturer_name"), - "communication_objects": len(d.get("communication_object_ids", []) or []), - } for d in proj.devices.values()] + devs = sorted(proj.devices.values(), + key=lambda d: _ia_sort_key(d.get("individual_address") or "")) + total = len(devs) + start = 0 + if cursor: + ck = _ia_sort_key(cursor) + start = next((i for i, d in enumerate(devs) + if _ia_sort_key(d.get("individual_address") or "") > ck), total) + page = devs[start:start + max(1, limit)] + return { + "devices": [{ + "individual_address": d.get("individual_address"), + "name": d.get("name"), + "order_number": d.get("order_number"), + "manufacturer": d.get("manufacturer_name"), + "communication_objects": len(d.get("communication_object_ids", []) or []), + } for d in page], + "total_matched": total, + "returned": len(page), + "next_cursor": (page[-1].get("individual_address") if start + len(page) < total else None), + } @mcp.tool() diff --git a/tests/test_paging.py b/tests/test_paging.py new file mode 100644 index 0000000..0bb997f --- /dev/null +++ b/tests/test_paging.py @@ -0,0 +1,95 @@ +"""Stable ordering and cursor paging for the two tools whose natural result is large +(`list_group_addresses`, `get_devices`). + +Why this exists: the old implementation returned a bare list truncated at `limit`, in +whatever order the parser happened to produce, with no signal that anything was cut. A +retry could therefore return a different page, and a caller could not tell a complete +answer from a truncated one (raised in r/mcp, 2026-09-12). +""" +import random + +from nickol_knx_mcp.project import build_loaded_from_raw +from nickol_knx_mcp import server as srv + + +def _ga(addr, name, dm=1, ds=1): + return {"name": name, "identifier": f"G{addr}", "raw_address": 0, "address": addr, + "project_uid": None, "dpt": {"main": dm, "sub": ds}, "data_secure": False, + "communication_object_ids": [], "description": "", "comment": ""} + + +def _project(shuffled: bool) -> dict: + addrs = [f"{m}/{mid}/{sub}" for m in (1, 2, 10) for mid in (0, 3) for sub in (1, 2, 12)] + if shuffled: + random.Random(7).shuffle(addrs) # parser order is not address order + gas = {a: _ga(a, f"GA {a}") for a in addrs} # bare 1.001, no domain word -> category unknown + devs = {ia: {"individual_address": ia, "name": f"dev {ia}", "order_number": "X", + "manufacturer_name": "M", "communication_object_ids": [], "channels": {}} + for ia in (["1.1.10", "1.1.2", "1.2.1", "10.1.1", "1.1.1"] if shuffled + else ["1.1.1", "1.1.2", "1.1.10", "1.2.1", "10.1.1"])} + raw = {"info": {"group_address_style": "ThreeLevel", "schema_version": "21"}, + "group_addresses": gas, "communication_objects": {}, "devices": devs, + "functions": {}, "topology": {}, "group_ranges": {}} + return raw + + +def _all_pages(fn, key, page_size): + out, cursor, guard = [], None, 0 + while True: + res = fn(limit=page_size, cursor=cursor) + out.extend(res[key]) + cursor = res["next_cursor"] + guard += 1 + assert guard < 50, "cursor never terminated" + if cursor is None: + return out, res["total_matched"] + + +def main(): + # the same project, once in address order and once shuffled by the "parser" + for shuffled in (False, True): + srv._STATE["project"] = build_loaded_from_raw(_project(shuffled), "paging.knxproj") + + res = srv.list_group_addresses(limit=5) + addrs = [g["address"] for g in res["group_addresses"]] + # 1. numeric, not lexical: 1/0/2 before 1/0/12, and main 2 before main 10 + assert addrs == ["1/0/1", "1/0/2", "1/0/12", "1/3/1", "1/3/2"], addrs + # 2. truncation is never silent + assert res["total_matched"] == 18 and res["returned"] == 5, res + assert res["next_cursor"] == "1/3/2", res["next_cursor"] + + # 3. paging covers everything exactly once, in order, whatever the page size + for size in (1, 5, 7, 18, 100): + got, total = _all_pages(lambda **kw: srv.list_group_addresses(**kw), + "group_addresses", size) + seen = [g["address"] for g in got] + assert total == 18 and len(seen) == 18 and len(set(seen)) == 18, (size, len(seen)) + assert seen == sorted(seen, key=srv._ga_sort_key), size + + # 4. devices: individual addresses sort numerically, same contract + d = srv.get_devices(limit=3) + assert [x["individual_address"] for x in d["devices"]] == ["1.1.1", "1.1.2", "1.1.10"], d + assert d["total_matched"] == 5 and d["next_cursor"] == "1.1.10" + devs, total = _all_pages(lambda **kw: srv.get_devices(**kw), "devices", 2) + assert total == 5 and [x["individual_address"] for x in devs] == \ + ["1.1.1", "1.1.2", "1.1.10", "1.2.1", "10.1.1"], devs + + # 5. filters still work and are reflected in total_matched + f = srv.list_group_addresses(kind="command", limit=100) + assert f["total_matched"] == f["returned"] == 18 and f["next_cursor"] is None, f + + # 6. the last page of one run equals the last page of a re-parsed, reshuffled project + srv._STATE["project"] = build_loaded_from_raw(_project(False), "paging.knxproj") + a = srv.list_group_addresses(limit=4, cursor="1/3/2") + srv._STATE["project"] = build_loaded_from_raw(_project(True), "paging.knxproj") + b = srv.list_group_addresses(limit=4, cursor="1/3/2") + assert [g["address"] for g in a["group_addresses"]] == [g["address"] for g in b["group_addresses"]], (a, b) + + srv._STATE["project"] = None + print("test_paging: OK — group addresses and devices sort numerically and identically " + "regardless of parser order; cursor paging covers every row exactly once; " + "total_matched makes truncation explicit.") + + +if __name__ == "__main__": + main()