fix(canon): parse the CR-N300 wire format as measured against hardware

First contact with the real camera (129.129.122.151, fw 1.4.0, protocol
5.0.0) on 2026-08-17 produced a probe report that pinpointed why the XC
session never opened.

The ":=" assignment operator
-----------------------------
open.cgi answers, and answers correctly:

    s:=8091-50cdfb31
    s.authority.control:=enabled
    s.duration==0

The operator is ":=", not "=". Splitting on the first "=" yields the key
"s:", which matches none of ("s", "session", "id"), so connect() raised
"open.cgi did not return a session id" against a perfectly good response.
The same bug gave every info.cgi key a trailing colon, so no parameter or
position lookup could ever have worked either.

_parse_kv now matches the operator explicitly, longest first, covering the
three forms the camera actually emits: ":=" for most fields, plain "=" for
timestamp/realtime, and "==" for s.duration.

Identity fields
---------------
get_camera_info() now reads the measured names -- s.hardware ("Canon
CR-N300"), s.firmware ("1.4.0") and s.hardware.address (the MAC, the
closest thing to a serial the camera exposes) -- keeping the previous
guesses as fallbacks.

HTTP 200 is not evidence of existence
-------------------------------------
This server returns 200 for every path, including CGIs it does not
implement, and reports the real outcome in the body: "Unknown Operator"
means absent, "Parameter Missing" means present but wanting arguments.
The probe was counting both as hits, which inverted the conclusion -- it
is what made nphMotionJpeg, preset.cgi, getparam.cgi, camera.cgi,
GetLiveImage, getstill.cgi and snapshot.cgi look present when none of them
exist on this firmware. The sweep now classifies each reply and lists the
unimplemented names separately.

Also: the probe capped reads at 4096 bytes, which truncated info.cgi
mid-response and hid the pan/tilt/zoom fields needed for the axis
encodings (R5). Raised to 256 kB.

Tests: 83 (was 78). The new ones use response bodies copied verbatim from
the camera, so this cannot regress silently.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

csaxs_bec/devices/canon/probe.py

@@ -78,6 +78,12 @@ IDENTITY_PATHS = ["/", "/index.html", "/viewer/", "/admin/"]
 _JPEG_SOI = b"\xff\xd8"
 _TIMEOUT = 5.0
 
+# info.cgi on a CR-N300 is well over 4 kB -- the pan/tilt/zoom fields that resolve the
+# axis encodings (R5) sit past that. Read enough to capture the whole thing.
+_READ_LIMIT = 262144
+# ...but keep only a slice in the JSON report, so it stays readable.
+_TEXT_LIMIT = 65536
+
 
 class Probe:
     """Issues the read-only requests and accumulates a structured report."""
@@ -115,7 +121,7 @@ class Probe:
             req.add_header("Authorization", f"Basic {token}")
         try:
             with self._opener(url).open(req, timeout=_TIMEOUT) as resp:
-                body = resp.read(4096)
+                body = resp.read(_READ_LIMIT)
                 is_jpeg = body.startswith(_JPEG_SOI)
                 return {
                     "url": url,
@@ -127,7 +133,9 @@ class Probe:
                     "head": _describe_body(body),
                     "is_jpeg": is_jpeg,
                     # Kept for session-id parsing; never for image payloads.
-                    "text": "" if is_jpeg else body.decode("utf-8", errors="replace")[:2048],
+                    "text": (
+                        "" if is_jpeg else body.decode("utf-8", errors="replace")[:_TEXT_LIMIT]
+                    ),
                 }
         except urllib.error.HTTPError as exc:
             body = b""
@@ -197,28 +205,44 @@ class Probe:
         print("\n[3] CGI surface sweep -- resolves risk items R1 (prefix) and R2 (endpoints)")
         print("    (401 = endpoint EXISTS but needs auth -- that still confirms the path)")
         found: dict[str, list[dict]] = {}
+        absent: dict[str, list[str]] = {}
         for prefix in CANDIDATE_PREFIXES:
-            hits = []
+            hits, missing = [], []
             for role, names in CANDIDATE_ENDPOINTS.items():
                 for name in names:
                     path = f"/{prefix}/{name}" if prefix else f"/{name}"
                     res = self.get(path)
-                    status = res.get("status")
-                    if status in (200, 401, 403):
-                        hits.append({"role": role, "name": name, "path": path, **res})
+                    if res.get("status") not in (200, 401, 403):
+                        continue
+                    verdict = _livescope_verdict(res)
+                    if verdict == "absent":
+                        missing.append(name)
+                        continue
+                    hits.append(
+                        {"role": role, "name": name, "path": path, "verdict": verdict, **res}
+                    )
             if hits:
                 label = f"/{prefix}/" if prefix else "/ (root)"
                 print(f"\n    prefix {label}")
                 for hit in hits:
-                    flag = " <-- JPEG!" if hit.get("is_jpeg") else ""
+                    if hit.get("is_jpeg"):
+                        note = " <-- JPEG!"
+                    elif hit["verdict"] == "needs-params":
+                        note = " (exists, wants arguments)"
+                    else:
+                        note = ""
                     print(
                         f"      {hit['status']}  {hit['role']:<8} {hit['name']:<16}"
-                        f" {hit.get('content_type', '')[:24]}{flag}"
+                        f" {hit.get('content_type', '')[:24]}{note}"
                     )
                 found[prefix] = hits
+            if missing:
+                absent[prefix] = missing
+                print(f"      not implemented here: {', '.join(missing)}")
         if not found:
             print("    (no candidate path answered -- see step 4, the web UI is the spec)")
         self.report["steps"]["endpoints"] = found
+        self.report["steps"]["endpoints_absent"] = absent
 
     def step_still(self) -> None:
         """Did anything hand us a real JPEG? That de-risks both default modes."""
@@ -348,6 +372,35 @@ class Probe:
         print("=" * 72)
 
 
+def _livescope_verdict(res: dict) -> str:
+    """Classify a WebView Livescope reply beyond its HTTP status.
+
+    MEASURED (CR-N300 fw 1.4.0): this server answers **HTTP 200 for everything**,
+    including CGIs it does not implement, and puts the real outcome in the body::
+
+        --- WebView Livescope Http Server Error ---
+        Unknown Operator          <- the CGI does NOT exist
+        operator=nphMotionJpeg
+
+        --- WebView Livescope Http Server Error ---
+        Parameter Missing         <- the CGI DOES exist; it wants arguments
+        1 parameter(s)
+
+    Taking 200 at face value would report every probed name as present, which is
+    exactly backwards. Returns ``absent``, ``needs-params`` or ``ok``.
+    """
+    if res.get("is_jpeg"):
+        return "ok"
+    text = res.get("text", "")
+    if "Unknown Operator" in text:
+        return "absent"
+    if "Parameter Missing" in text:
+        return "needs-params"
+    if "Livescope Http Server Error" in text:
+        return "absent"
+    return "ok"
+
+
 def _import_xctransport():
     """Locate ``XCTransport``, trying progressively less-demanding routes.
 
@@ -399,10 +452,18 @@ def _parse_session_id(text: str, set_cookie: str) -> str | None:
     Mirrors ``XCTransport.connect()`` -- risk item R6b. Kept self-contained so this
     file works when copied on its own.
     """
+    # MEASURED (CR-N300 fw 1.4.0): open.cgi answers "s:=8091-50cdfb31" -- the
+    # operator is ":=", so splitting on "=" alone yields the key "s:" and never
+    # matches. Longest separator first.
     for line in text.replace("\r", "\n").split("\n"):
-        key, _, value = line.strip().partition("=")
-        if key.strip() in ("s", "session", "id") and value.strip():
-            return value.strip()
+        line = line.strip()
+        for sep in (":=", "==", "="):
+            if sep not in line:
+                continue
+            key, _, value = line.partition(sep)
+            if key.strip() in ("s", "session", "id") and value.strip():
+                return value.strip()
+            break
     if "=" in set_cookie:
         candidate = set_cookie.split(";", 1)[0].split("=", 1)[1].strip()
         return candidate or None

csaxs_bec/devices/canon/transport/xc.py

@@ -262,16 +262,30 @@ class XCTransport(CameraTransport):
         assert last_exc is not None
         raise last_exc
 
-    @staticmethod
-    def _parse_kv(text: str) -> dict[str, str]:
-        """Parse a newline-delimited ``key=value`` CGI response body."""
+    # Assignment operators used by the WebView Livescope CGI responses, longest
+    # first so ":=" wins over the "=" inside it. MEASURED against a CR-N300
+    # (fw 1.4.0, protocol 5.0.0) on 2026-08-17: most fields use ":=", a few
+    # ("timestamp", "realtime") use plain "=", and "s.duration==0" uses "==".
+    _KV_SEPARATORS = (":=", "==", "=")
+
+    @classmethod
+    def _parse_kv(cls, text: str) -> dict[str, str]:
+        """Parse a newline-delimited ``key<op>value`` CGI response body.
+
+        Splitting naively on ``"="`` would yield keys with a trailing colon
+        (``"s:"`` instead of ``"s"``) and break every lookup, so the operator is
+        matched explicitly.
+        """
         result: dict[str, str] = {}
         for line in text.replace("\r", "\n").split("\n"):
             line = line.strip()
-            if not line or "=" not in line:
+            if not line:
                 continue
-            key, _, value = line.partition("=")
-            result[key.strip()] = value.strip()
+            for sep in cls._KV_SEPARATORS:
+                if sep in line:
+                    key, _, value = line.partition(sep)
+                    result[key.strip()] = value.strip()
+                    break
         return result
 
     # -- session lifecycle -------------------------------------------------
@@ -281,7 +295,9 @@ class XCTransport(CameraTransport):
             return
         resp = self._request("open", timeout=timeout)
         kv = self._parse_kv(resp.text)
-        # SPEC: the session-id field name. "s" / "session" / "id" are plausible.
+        # MEASURED (CR-N300 fw 1.4.0): open.cgi returns "s:=8091-50cdfb31" plus
+        # s.authority.* / s.origin / s.priority. The other names are kept as
+        # fallbacks for firmware that differs.
         session_id = kv.get("s") or kv.get("session") or kv.get("id")
         if not session_id:
             # Some firmware returns the session id only in a Set-Cookie header.
@@ -335,10 +351,14 @@ class XCTransport(CameraTransport):
     def get_camera_info(self) -> CameraInfo:
         resp = self._request("info")
         kv = self._parse_kv(resp.text)
+        # MEASURED (CR-N300 fw 1.4.0): info.cgi reports "s.hardware:=Canon CR-N300",
+        # "s.firmware:=1.4.0" and "s.hardware.address:=7438B7FF507C" (the MAC, which
+        # is the closest thing the camera exposes to a serial number). Older field
+        # names are kept as fallbacks.
         return CameraInfo(
-            model=kv.get("model", ""),
-            firmware=kv.get("firmware", kv.get("version", "")),
-            serial=kv.get("serial", ""),
+            model=kv.get("s.hardware", kv.get("model", "")),
+            firmware=kv.get("s.firmware", kv.get("firmware", kv.get("version", ""))),
+            serial=kv.get("s.hardware.address", kv.get("serial", "")),
         )
 
     # -- parameters --------------------------------------------------------

tests/tests_devices/test_canon_probe.py

@@ -227,6 +227,39 @@ def test_builtin_session_step_uses_the_discovered_prefix_not_the_default():
         probe.summarise()
 
 
+def test_session_id_parsing_handles_the_real_camera_operator():
+    """MEASURED: the CR-N300 writes 's:=8091-50cdfb31', not 's=...'."""
+    from csaxs_bec.devices.canon.probe import _parse_session_id
+
+    body = (
+        "s:=8091-50cdfb31\n"
+        "s.authority.control:=enabled\n"
+        "s.origin:=129.129.122.151:80\n"
+        "s.duration==0\n"
+    )
+    assert _parse_session_id(body, "") == "8091-50cdfb31"
+    # 's.authority.control' must not be mistaken for the session field.
+    assert _parse_session_id("s.authority.control:=enabled\n", "") is None
+
+
+def test_livescope_verdict_distinguishes_absent_from_needs_arguments():
+    """HTTP 200 means nothing here -- the body carries the real outcome."""
+    from csaxs_bec.devices.canon.probe import _livescope_verdict
+
+    absent = {
+        "text": "--- WebView Livescope Http Server Error ---\n"
+        "Unknown Operator\noperator=nphMotionJpeg\n"
+    }
+    exists = {
+        "text": "--- WebView Livescope Http Server Error ---\n"
+        "Parameter Missing\n1 parameter(s)\n"
+    }
+    assert _livescope_verdict(absent) == "absent"
+    assert _livescope_verdict(exists) == "needs-params"
+    assert _livescope_verdict({"text": "s:=8091-50cdfb31\n"}) == "ok"
+    assert _livescope_verdict({"is_jpeg": True, "text": ""}) == "ok"
+
+
 def test_session_id_parsing_covers_body_and_set_cookie():
     from csaxs_bec.devices.canon.probe import _parse_session_id
 

tests/tests_devices/test_canon_transport.py

@@ -238,6 +238,75 @@ def test_xc_focus_mode_roundtrip_wire_tokens():
     assert "focusmode=auto" in set_url
 
 
+# --------------------------------------------------------------------------- #
+# Real CR-N300 wire format (measured 2026-08-17, fw 1.4.0, protocol 5.0.0)     #
+# --------------------------------------------------------------------------- #
+
+# Verbatim from the camera. The assignment operator is ":=", not "=" -- parsing
+# on "=" alone yields the key "s:" and the session id is never found.
+REAL_OPEN_BODY = (
+    "s:=8091-50cdfb31\n"
+    "s.authority.audio:=disabled\n"
+    "s.authority.control:=enabled\n"
+    "s.authority.video:=enabled\n"
+    "s.origin:=129.129.122.151:80\n"
+    "s.duration==0\n"
+    "s.priority:=0\n"
+    "v:=jpg:1280x720:6:12500\n"
+)
+
+REAL_INFO_BODY = (
+    "timestamp=15931.405\n"
+    "realtime=1786986136.725\n"
+    "s.system.status:=idle\n"
+    "s.epoch:=Mon, 17 Aug 2026 13:37:37 +0100\n"
+    "s.hardware:=Canon CR-N300\n"
+    "s.hardware.id:=111\n"
+    "s.hardware.address:=7438B7FF507C\n"
+    "s.hardware.maxsize:=3840x2160\n"
+    "s.firmware:=1.4.0\n"
+    "s.protocol:=5.0.0\n"
+    "c.1.exp:=auto\n"
+    "c.1.exp.list:=auto,av,tv,manual\n"
+    "c.1.ae.brightness.min:=-8\n"
+)
+
+
+def test_xc_parses_the_walrus_assignment_operator():
+    """The camera uses ':=' for most fields, '=' for a few, '==' for one."""
+    kv = XCTransport._parse_kv(REAL_OPEN_BODY)
+    assert kv["s"] == "8091-50cdfb31"
+    assert kv["s.authority.control"] == "enabled"
+    assert kv["s.origin"] == "129.129.122.151:80"
+    assert kv["s.duration"] == "0"  # written "s.duration==0"
+    assert "s:" not in kv, "keys must not keep a trailing colon"
+
+    info = XCTransport._parse_kv(REAL_INFO_BODY)
+    assert info["timestamp"] == "15931.405"  # plain '=' still works
+    assert info["s.hardware"] == "Canon CR-N300"
+    assert info["c.1.exp.list"] == "auto,av,tv,manual"
+    assert info["c.1.ae.brightness.min"] == "-8"  # negative values survive
+
+
+def test_xc_connect_finds_the_session_id_in_a_real_open_response():
+    """Regression for the bug that blocked first contact on 2026-08-17."""
+    opener = RecordingOpener(bodies={"open.cgi": REAL_OPEN_BODY})
+    t = make_xc(opener)
+    t.connect()
+    assert t.connected
+    assert t._session_id == "8091-50cdfb31"
+
+
+def test_xc_camera_info_uses_the_measured_field_names():
+    opener = RecordingOpener(bodies={"open.cgi": REAL_OPEN_BODY, "info.cgi": REAL_INFO_BODY})
+    t = make_xc(opener)
+    t.connect()
+    info = t.get_camera_info()
+    assert info.model == "Canon CR-N300"
+    assert info.firmware == "1.4.0"
+    assert info.serial == "7438B7FF507C"
+
+
 # --------------------------------------------------------------------------- #
 # Single-frame capture (backs the jpeg_on_demand acquisition mode)            #
 # --------------------------------------------------------------------------- #