feat(pid): ACADMCP_PID payload encoding with 255-char chunking

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd

engineering/pid/xdata.py

@@ -0,0 +1,93 @@
+"""``ACADMCP_PID`` XDATA payloads.
+
+XDATA strings are capped at 255 characters, so a JSON payload is stored as
+``["json", chunk, chunk, ...]`` and re-joined on read. Every symbol INSERT and
+every P&ID line carries one, so a reader without the catalogue (another tool,
+a later catalogue version) still sees ports, classes and numbers.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from backends.base import AutoCADBackend
+
+APP_NAME = "ACADMCP_PID"
+MARKER = "json"
+CHUNK = 255
+PAYLOAD_VERSION = 1
+
+
+def encode_payload(payload: dict) -> list[str]:
+    text = json.dumps(payload, separators=(",", ":"), sort_keys=True, ensure_ascii=True)
+    return [MARKER] + [text[i : i + CHUNK] for i in range(0, len(text), CHUNK)]
+
+
+def decode_payload(values: list) -> dict | None:
+    if not values or values[0] != MARKER:
+        return None
+    parts = [v for v in values[1:] if isinstance(v, str)]
+    try:
+        decoded = json.loads("".join(parts))
+    except json.JSONDecodeError:
+        return None
+    return decoded if isinstance(decoded, dict) else None
+
+
+async def write_payload(backend: AutoCADBackend, handle: str, payload: dict) -> dict:
+    return await backend.entity_set_xdata(handle, APP_NAME, encode_payload(payload))
+
+
+async def read_payload(backend: AutoCADBackend, handle: str) -> dict | None:
+    result = await backend.entity_get_xdata(handle, APP_NAME)
+    return decode_payload(result.get("xdata", {}).get(APP_NAME, []))
+
+
+def symbol_payload(spec: Any, params: dict | None = None) -> dict:
+    """``spec`` is an ``engineering.pid.symbols.SymbolSpec`` (typed loosely to
+    avoid an import cycle). Ports are symbol-local ``[x, y, direction, kind, radius]``."""
+    from .symbols import CATALOG_VERSION
+
+    return {
+        "v": PAYLOAD_VERSION,
+        "kind": "symbol",
+        "catalog": CATALOG_VERSION,
+        "name": spec.name,
+        "family": spec.family,
+        "symbol": spec.symbol,
+        "variant": spec.variant,
+        "params": dict(params or spec.params),
+        "ports": {p.name: [p.x, p.y, p.direction_deg, p.kind, p.radius] for p in spec.ports},
+    }
+
+
+def line_payload(
+    line_class: str,
+    number: str | None,
+    size: str | None,
+    service: str | None,
+    spec: str | None,
+    insulation: str | None,
+    seq: int | None,
+    from_ref: dict | None,
+    to_ref: dict | None,
+) -> dict:
+    return {
+        "v": PAYLOAD_VERSION,
+        "kind": "line",
+        "class": line_class,
+        "number": number,
+        "size": size,
+        "service": service,
+        "spec": spec,
+        "insulation": insulation,
+        "seq": seq,
+        "from": from_ref,
+        "to": to_ref,
+    }
+
+
+def marker_payload(line_handle: str) -> dict:
+    return {"v": PAYLOAD_VERSION, "kind": "marker", "line": line_handle}

tests/test_pid_xdata.py

@@ -0,0 +1,69 @@
+"""ACADMCP_PID payloads survive the 255-character XDATA string limit."""
+
+from __future__ import annotations
+
+import pytest
+
+from engineering.pid.xdata import (
+    APP_NAME,
+    MARKER,
+    decode_payload,
+    encode_payload,
+    line_payload,
+    marker_payload,
+    read_payload,
+    write_payload,
+)
+
+pytestmark = pytest.mark.asyncio
+
+
+def test_short_payload_is_marker_plus_one_chunk():
+    values = encode_payload({"v": 1, "kind": "marker", "line": "3A"})
+    assert values[0] == MARKER and len(values) == 2
+    assert decode_payload(values) == {"kind": "marker", "line": "3A", "v": 1}
+
+
+def test_long_payload_is_chunked_at_255_and_rejoined():
+    payload = {"v": 1, "kind": "symbol", "ports": {f"N{i}": [i, i, 90] for i in range(60)}}
+    values = encode_payload(payload)
+    assert all(len(v) <= 255 for v in values)
+    assert len(values) > 2
+    assert decode_payload(values) == payload
+
+
+def test_decode_rejects_foreign_or_broken_values():
+    assert decode_payload([]) is None
+    assert decode_payload(["not-json", "{}"]) is None
+    assert decode_payload([MARKER, "{not json"]) is None
+
+
+def test_builders_carry_the_documented_keys():
+    line = line_payload(
+        "process_major",
+        "100-P-1001",
+        "100",
+        "P",
+        "CS1",
+        "IH",
+        1001,
+        {"handle": "2F", "port": "discharge"},
+        {"handle": "31", "port": "N1"},
+    )
+    assert line["kind"] == "line" and line["class"] == "process_major"
+    assert line["from"] == {"handle": "2F", "port": "discharge"} and line["seq"] == 1001
+    assert marker_payload("3A") == {"v": 1, "kind": "marker", "line": "3A"}
+
+
+async def test_roundtrip_through_a_backend(backend):
+    line = await backend.entity_create_line(0, 0, 10, 0)
+    payload = {"v": 1, "kind": "line", "class": "electric", "number": None}
+    await write_payload(backend, line.handle, payload)
+    assert await read_payload(backend, line.handle) == payload
+    got = await backend.entity_get_xdata(line.handle, APP_NAME)
+    assert got["xdata"][APP_NAME][0] == MARKER
+
+
+async def test_read_payload_is_none_without_xdata(backend):
+    line = await backend.entity_create_line(0, 0, 10, 0)
+    assert await read_payload(backend, line.handle) is None