fix(sp38): restore per-kind control_number in find_ack_orphans + review cleanups

Three pr-reviewer followups from the 2026-07-07 review of commit ad14b56:

1. BUG: find_ack_orphans refactor routed 277ca/ta1 through
   _ack_control_number which only knew 999 — restored per-kind source
   (999 reads raw_json.envelope.control_number, 277ca/ta1 read the ORM
   control_number column). Added regression test pinning all three
   kinds.

2. DOCSTRING DRIFT: reconcile_orphan_st02s said 'remaining columns
   take their defaults' but explicitly passes parsed_at and
   transaction_set_control_number — added both to the explicit list
   and clarified the rest take schema defaults.

3. WASTED SORT: _iter_orphan_999_st02s yielded sorted() but the
   caller re-sorts by (-ack_count, st02) — yielding unsorted now.

Plus three cleanups the reviewer flagged:

* Hoisted 'json' / 'uuid' / 'datetime' imports to module top of
  store/__init__.py (replaced in-method imports).
* Added two missing tests: sentinel grep-discoverability via
  LIKE '<synthetic:%>' + 999-walk tolerance for non-dict raw_json
  (None / list shapes — bytes is unreachable through the ORM).
* Aligned spec/plan exit codes to the cyclone-cli convention
  (exit 1 on DB error, not 2 — matches the existing CLI
  sys.exit(1) and the cyclone-cli skill documentation).

36/36 SP38 tests pass.

backend/src/cyclone/store/__init__.py

@@ -38,7 +38,9 @@ Backward-compat shims for tests:
 
 from __future__ import annotations
 
+import json
 import threading
+import uuid
 from datetime import datetime, timezone
 
 
@@ -383,6 +385,8 @@ class CycloneStore:
 
           * ``kind = '837p'``
           * ``input_filename = '<synthetic:orphan-reconcile>'``
+          * ``parsed_at = utcnow()``
+          * ``transaction_set_control_number = <orphan ST02>``
           * ``totals_json = {"orphan_reconcile": true,
                               "ack_count": <orphan count>}``
           * ``validation_json = {"orphan_reconcile": true,
@@ -391,6 +395,9 @@ class CycloneStore:
                                  this DB snapshot"}``
           * ``id = uuid4().hex`` (no claim rows, no claim_acks links)
 
+        Remaining columns (``claim_count``, ``received_count``, …) take
+        their schema defaults.
+
         The sentinel ``<synthetic:orphan-reconcile>`` makes these
         rows trivially distinguishable in queries — grep for that
         string to find every row this method has created.
@@ -415,8 +422,6 @@ class CycloneStore:
 
         Idempotent: re-running after a successful pass is a no-op.
         """
-        import json
-        import uuid
         from cyclone import db
         from cyclone.db import Batch
 
@@ -436,8 +441,7 @@ class CycloneStore:
                     created += 1  # would-create count
             return {"created": created, "skipped": skipped, "synthetic_batch_ids": []}
 
-        from datetime import datetime, timezone
-        now = datetime.now(timezone.utc)
+        now = utcnow()
         with db.SessionLocal()() as s:
             for row in summary:
                 if row["has_batch"]:

backend/src/cyclone/store/claim_acks.py

@@ -348,7 +348,11 @@ def _iter_orphan_999_st02s() -> Iterator[tuple[str, int]]:
             if st02 is None:
                 continue
             counts[st02] = counts.get(st02, 0) + 1
-    yield from sorted(counts.items())
+    # Unsorted on purpose — the caller re-sorts by (-ack_count, st02)
+    # so the heaviest backlog surfaces first in CLI output. Sorting
+    # here would just be wasted CPU (and risk a different order if
+    # the caller's key changes).
+    yield from counts.items()
 
 
 def _extract_999_st02(ack_row: Ack) -> str | None:
@@ -384,17 +388,41 @@ def _extract_999_st02(ack_row: Ack) -> str | None:
     return str(st02) if st02 else None
 
 
-def _ack_control_number(ack_row: Ack, kind: str) -> str:
-    """Best-effort control-number lookup for a 999 ack row.
+def _ack_control_number(ack_row, kind: str) -> str:
+    """Best-effort control-number lookup for an ack row of any kind.
 
-    The 999 ORM row doesn't carry the envelope's control_number in
-    a dedicated column; we re-derive it from ``raw_json`` (the same
-    source :func:`cyclone.store.ui.to_ui_ack` uses for the patient
-    control number).
+    Per-kind sources (preserved across the SP38 refactor of
+    :func:`find_ack_orphans`):
+
+      * ``999`` — 999 ORM row has no dedicated control_number column;
+        we re-derive it from ``raw_json.envelope.control_number``
+        (the same source :func:`cyclone.store.ui.to_ui_ack` uses).
+        Accepts both dict (post-ORM hydration) and str (raw sqlite3
+        row or freshly inserted JSON string).
+      * ``277ca`` / ``ta1`` — both carry the control number in a
+        dedicated ORM column (``ack_row.control_number``). Reading
+        from ``raw_json`` would yield empty strings.
+
+    Returns ``""`` (empty string) when the source field is missing
+    so the JSON renderer still emits a stable shape.
     """
-    raw = ack_row.raw_json or {}
-    env = raw.get("envelope") or {}
-    return env.get("control_number") or ""
+    if kind == "999":
+        raw = ack_row.raw_json
+        if raw is None or raw == "":
+            return ""
+        if isinstance(raw, dict):
+            env = raw.get("envelope") or {}
+            return env.get("control_number") or ""
+        if isinstance(raw, (str, bytes, bytearray)):
+            try:
+                parsed = json.loads(raw)
+            except (TypeError, ValueError):
+                return ""
+            env = parsed.get("envelope") or {}
+            return env.get("control_number") or ""
+        return ""
+    # 277ca / ta1 — control_number is an ORM column on both tables.
+    return getattr(ack_row, "control_number", "") or ""
 
 
 __all__ = [

backend/tests/test_ack_orphan_summary.py

@@ -317,4 +317,61 @@ def test_reconcile_records_ack_count_in_totals_json():
         )
         totals = json.loads(row.totals_json or "{}")
         assert totals.get("orphan_reconcile") is True
-        assert totals.get("ack_count") == 4
+        assert totals.get("ack_count") == 4
+
+
+def test_reconcile_sentinel_is_grep_discoverable():
+    """``input_filename = '<synthetic:orphan-reconcile>'`` must be
+    discoverable via a plain ``LIKE '<synthetic:%>'`` query so
+    operators can find every sp38-created row without knowing the
+    exact sentinel string. Pins the spec's D2 grep-discoverability
+    invariant."""
+    with db.SessionLocal()() as s:
+        _ingest_999(s, "991102988")
+        s.commit()
+
+    CycloneStore().reconcile_orphan_st02s(dry_run=False)
+
+    with db.SessionLocal()() as s:
+        n = (
+            s.query(Batch)
+            .filter(Batch.input_filename.like("<synthetic:%>"))
+            .count()
+        )
+        assert n == 1
+
+
+def test_summary_skips_999s_with_non_dict_raw_json():
+    """``_extract_999_st02`` must tolerate the un-dict shapes a JSON
+    column can take (None, list) and never raise. Rows with
+    non-dict raw_json are skipped — they have no ST02 to contribute
+    to the summary regardless.
+
+    Note: the SQLAlchemy JSON column rejects bytes at insert time
+    (it must be JSON-serializable), so we don't exercise the
+    ``bytes`` branch here — that's a defensive-programming guard for
+    raw sqlite3 reads, not a normal ORM codepath.
+    """
+    with db.SessionLocal()() as s:
+        # Valid orphan — contributes 1 row
+        _ingest_999(s, "991102988")
+        # None — should be skipped, not raise
+        s.add(Ack(
+            source_batch_id="999-none-raw",
+            accepted_count=1, rejected_count=0, received_count=1,
+            ack_code="A", parsed_at=_now(),
+            raw_json=None,
+        ))
+        # list — not a dict; should be skipped (not raise)
+        s.add(Ack(
+            source_batch_id="999-list-raw",
+            accepted_count=1, rejected_count=0, received_count=1,
+            ack_code="A", parsed_at=_now(),
+            raw_json=[1, 2, 3],
+        ))
+        s.commit()
+
+    summary = CycloneStore().find_ack_orphan_st02_summary()
+    by_st02 = {row["st02"]: row for row in summary}
+    assert by_st02 == {"991102988": {"st02": "991102988", "ack_count": 1,
+                                     "has_batch": False, "batch_id": None}}

backend/tests/test_apply_claim_ack_links.py

@@ -34,6 +34,7 @@ from cyclone.claim_acks import (
     lookup_claims_for_ack_set_response,
 )
 from cyclone.db import (
+    Ack,
     Batch,
     Claim,
     ClaimState,
@@ -744,6 +745,75 @@ def test_store_facade_exposes_claim_ack_methods():
         assert hasattr(store, name), f"missing CycloneStore.{name}"
 
 
+# ---------------------------------------------------------------------------
+# SP38 follow-up: regression test for the 277ca / ta1 control_number
+# behavior preserved across the find_ack_orphans refactor.
+# ---------------------------------------------------------------------------
+
+
+def test_find_ack_orphans_returns_control_number_for_all_kinds():
+    """For each ack kind, ``find_ack_orphans(kind)`` must return the
+    control_number from the per-kind source:
+
+      * 999 — from raw_json.envelope.control_number
+      * 277ca — from the ORM column ``control_number``
+      * ta1 — from the ORM column ``control_number``
+
+    Regression for sp38 commit ad14b56 which initially routed all
+    three kinds through _ack_control_number and broke 277ca/ta1
+    (the helper only knew 999). Found by pr-reviewer 2026-07-07.
+    """
+    import json
+    import cyclone.db as _db_mod
+    Two77caAck = _db_mod.Two77caAck
+    Ta1Ack = _db_mod.Ta1Ack
+    parsed_at = datetime.now(timezone.utc)
+
+    with db.SessionLocal()() as s:
+        # 999 orphan with envelope.control_number in raw_json
+        s.add(Ack(
+            source_batch_id="999-orph-test",
+            accepted_count=1, rejected_count=0, received_count=1,
+            ack_code="A",
+            parsed_at=parsed_at,
+            raw_json=json.dumps({
+                "envelope": {"control_number": "999-ctrl-num",
+                             "sender_id": "S", "receiver_id": "R",
+                             "transaction_date": "2024-01-01",
+                             "implementation_guide": "005010X231A1"},
+                "set_responses": [],
+                "summary": {},
+            }),
+        ))
+        # 277ca orphan with control_number in ORM column
+        s.add(Two77caAck(
+            source_batch_id="277ca-orph-test",
+            accepted_count=1, rejected_count=0,
+            control_number="277CA-CTRL-NUM",
+            parsed_at=parsed_at,
+        ))
+        # ta1 orphan with control_number in ORM column
+        s.add(Ta1Ack(
+            source_batch_id="ta1-orph-test",
+            ack_code="A",
+            control_number="TA1-CTRL-NUM",
+            parsed_at=parsed_at,
+        ))
+        s.commit()
+
+    orphans_999 = store.find_ack_orphans("999")
+    orphans_277ca = store.find_ack_orphans("277ca")
+    orphans_ta1 = store.find_ack_orphans("ta1")
+
+    ctrl_999 = [o["control_number"] for o in orphans_999 if "999-ctrl-num" in o["control_number"]]
+    ctrl_277ca = [o["control_number"] for o in orphans_277ca]
+    ctrl_ta1 = [o["control_number"] for o in orphans_ta1]
+
+    assert "999-ctrl-num" in ctrl_999, f"999 control_number not found: {ctrl_999}"
+    assert "277CA-CTRL-NUM" in ctrl_277ca, f"277ca control_number empty: {ctrl_277ca}"
+    assert "TA1-CTRL-NUM" in ctrl_ta1, f"ta1 control_number empty: {ctrl_ta1}"
+
+
 # ---------------------------------------------------------------------------
 # Step 2.1 — pure walk-through unit test (no DB) for the orphan tracking
 # ---------------------------------------------------------------------------