merge: SP38 orphan-ack housekeeping into main

Captures the historical 804-orphan 999 situation as a known
artifact (RUNBOOK entry + status command) and provides a
one-shot idempotent synthetic-batch seeder so future acks for
the orphan ST02s can resolve against batch_envelope_index.

6 commits merged atomically (no squash — the SP-N merge commit
is the audit record):
  d776a4a docs(spec): design for SP38 orphan-ack housekeeping
  8890627 docs(plan): SP38 orphan-ack housekeeping plan
  ad14b56 feat(sp38): CycloneStore.find_ack_orphan_st02_summary + reconcile_orphan_st02s
  9ef749c feat(sp38): 'cyclone ack-orphans {status,reconcile}' CLI subcommands
  76923a7 docs(sp38): RUNBOOK entry for orphan-ack housekeeping
  07ea7ca fix(sp38): restore per-kind control_number in find_ack_orphans + review cleanups

Test delta: +36 passed (12 store helper + 7 CLI + 17 claim_acks
regression including the pr-reviewer 277ca/ta1 control_number
fix). No new failures.

Live verified: cyclone ack-orphans status exits 0 with 4 distinct
orphan ST02s (419+226+106+53=804) matching the SQLite investigation.

backend/src/cyclone/cli.py

@@ -1284,5 +1284,97 @@ def pull_inbound(
             click.echo(f"  {e}", err=True)
 
 
+# ---------------------------------------------------------------------------
+# SP38: ack-orphans status + reconcile
+# ---------------------------------------------------------------------------
+
+
+@main.group("ack-orphans")
+def ack_orphans_group() -> None:
+    """Inspect and reconcile 999 acks with no resolvable source claim.
+
+    "Orphans" are 999 acks whose source 837 batch is not present in
+    the ``batches`` table — typically because the source 837 was
+    submitted to HPE before the current ``cyclone.db`` snapshot was
+    created. They are real production data (valid audit history) but
+    cannot be auto-linked to claims.
+
+    See ``docs/superpowers/specs/2026-07-07-cyclone-orphan-ack-housekeeping-design.md``
+    for the full design and operator triage workflow.
+    """
+
+
+@ack_orphans_group.command("status")
+def ack_orphans_status_cmd() -> None:
+    """Print a per-ST02 summary of orphan 999 acks.
+
+    One row per distinct orphan ``set_control_number`` (ST02), ranked
+    by ``ack_count DESC`` so the heaviest backlog surfaces first.
+    Plus a TOTAL row at the bottom.
+
+    Exit codes: 0 on success, 1 on DB error.
+    """
+    from cyclone import db as db_mod
+    from cyclone.store import store as cycl_store
+
+    try:
+        db_mod.init_db()
+        summary = cycl_store.find_ack_orphan_st02_summary()
+    except Exception as exc:  # noqa: BLE001
+        click.echo(f"ack-orphans status failed: {type(exc).__name__}: {exc}", err=True)
+        sys.exit(1)
+
+    if not summary:
+        click.echo("no orphans (acks is empty or all are linked)")
+        return
+
+    # Table layout: ST02 | ACK COUNT | HAS BATCH
+    click.echo(f"{'ST02':<15} {'ACK COUNT':>10}   HAS BATCH")
+    click.echo(f"{'----':-<15} {'---------':->10}   ---------")
+    total = 0
+    for row in summary:
+        marker = "yes" if row["has_batch"] else "no"
+        click.echo(
+            f"{row['st02']:<15} {row['ack_count']:>10}   {marker}"
+        )
+        total += row["ack_count"]
+    click.echo(f"{'TOTAL':<15} {total:>10}")
+
+
+@ack_orphans_group.command("reconcile")
+@click.option("--dry-run", is_flag=True,
+              help="Print the plan but do not insert any rows.")
+def ack_orphans_reconcile_cmd(dry_run: bool) -> None:
+    """Insert synthetic batches rows for orphan ST02s that lack one.
+
+    For every orphan ST02 where no ``batches`` row exists, insert a
+    synthetic row marked with ``input_filename =
+    '<synthetic:orphan-reconcile>'`` so future 999 acks for the same
+    ST02s can resolve against the ``batch_envelope_index`` (though
+    they still won't link to claims — the source 837s were never
+    ingested).
+
+    Idempotent: re-running after a successful pass is a no-op.
+
+    Exit codes: 0 on success, 1 on DB error.
+    """
+    from cyclone import db as db_mod
+    from cyclone.store import store as cycl_store
+
+    try:
+        db_mod.init_db()
+        plan = cycl_store.reconcile_orphan_st02s(dry_run=dry_run)
+    except Exception as exc:  # noqa: BLE001
+        click.echo(f"ack-orphans reconcile failed: {type(exc).__name__}: {exc}", err=True)
+        sys.exit(1)
+
+    click.echo(
+        f"Created: {plan['created']} synthetic batch rows. "
+        f"Skipped: {plan['skipped']} (already had a batch row)."
+    )
+    if dry_run:
+        click.echo("(dry-run — no rows inserted)")
+
+
 if __name__ == "__main__":
     main()

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
 
 
@@ -87,6 +89,7 @@ from .claim_acks import (
     list_acks_for_claim as _list_acks_for_claim,
     list_claims_for_ack as _list_claims_for_ack,
     remove_claim_ack as _remove_claim_ack,
+    _iter_orphan_999_st02s as _iter_orphan_999_st02s,
 )
 from .backups import add_backup_pending
 from .exceptions import AlreadyMatchedError, InvalidStateError, NotMatchedError
@@ -320,6 +323,155 @@ class CycloneStore:
         """Return acks with no resolvable link (Inbox ack-orphans lane)."""
         return _find_ack_orphans(kind)
 
+    def find_ack_orphan_st02_summary(self):
+        """Return per-ST02 summary of orphan 999 acks (sp38).
+
+        Output is a list of dicts, sorted by ``ack_count DESC`` so
+        the heaviest orphan ST02s surface first in CLI output:
+
+          ``{"st02": str, "ack_count": int, "has_batch": bool,
+             "batch_id": str | None}``
+
+        ``has_batch`` is True if a row exists in ``batches`` with
+        ``transaction_set_control_number == st02``. ``batch_id`` is
+        that row's ``id`` (or ``None``).
+
+        999-only. 277ca / ta1 orphans are tracked separately; their
+        "ST02" semantics differ (it's the interchange control number,
+        not the source 837's ST02) and aggregating them with 999
+        ST02s would conflate unrelated identifiers.
+
+        Used by the ``cyclone ack-orphans status`` and ``reconcile``
+        CLI subcommands (sp38). Sibling to ``find_ack_orphans(kind)``
+        which returns one row per orphan ack; the summary is the
+        aggregated form.
+        """
+        from cyclone import db
+        from cyclone.db import Batch
+
+        # (st02, ack_count) from the 999-orphan walk.
+        counts = dict(_iter_orphan_999_st02s())
+
+        if not counts:
+            return []
+
+        # Map ST02 -> existing batch row (if any). One query.
+        with db.SessionLocal()() as s:
+            existing = {
+                row.transaction_set_control_number: row.id
+                for row in s.query(Batch)
+                .filter(Batch.transaction_set_control_number.in_(counts.keys()))
+                .all()
+            }
+
+        out = []
+        for st02, ack_count in counts.items():
+            batch_id = existing.get(st02)
+            out.append({
+                "st02": st02,
+                "ack_count": ack_count,
+                "has_batch": batch_id is not None,
+                "batch_id": batch_id,
+            })
+        # Heaviest first; tie-break by st02 ascending for determinism.
+        out.sort(key=lambda r: (-r["ack_count"], r["st02"]))
+        return out
+
+    def reconcile_orphan_st02s(self, *, dry_run: bool = False) -> dict:
+        """One-shot synthetic-batch seeder for orphan ST02s (sp38).
+
+        Inserts a ``batches`` row for every orphan ST02 that does
+        NOT already have one. The synthetic row is marked with:
+
+          * ``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,
+                                 "note": "sp38 synthetic batch row;
+                                 source 837 was never ingested into
+                                 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.
+
+        Future 999 acks referencing these ST02s will resolve
+        against the batch envelope index (so the operator can see
+        "this 999 is for a known orphan source") but will not link
+        to claims (because no claim rows exist for the synthetic
+        batch).
+
+        Args:
+            dry_run: If True, returns the plan but does not insert
+                any rows. Used by the CLI ``--dry-run`` flag so
+                operators can preview the reconcile.
+
+        Returns:
+            ``{"created": int, "skipped": int,
+               "synthetic_batch_ids": list[str]}``. ``created`` is
+            the count of rows inserted (or that WOULD be inserted
+            under dry_run). ``skipped`` is the count of orphan ST02s
+            that already had a batches row.
+
+        Idempotent: re-running after a successful pass is a no-op.
+        """
+        from cyclone import db
+        from cyclone.db import Batch
+
+        summary = self.find_ack_orphan_st02_summary()
+        if not summary:
+            return {"created": 0, "skipped": 0, "synthetic_batch_ids": []}
+
+        created = 0
+        skipped = 0
+        synthetic_ids: list[str] = []
+
+        if dry_run:
+            for row in summary:
+                if row["has_batch"]:
+                    skipped += 1
+                else:
+                    created += 1  # would-create count
+            return {"created": created, "skipped": skipped, "synthetic_batch_ids": []}
+
+        now = utcnow()
+        with db.SessionLocal()() as s:
+            for row in summary:
+                if row["has_batch"]:
+                    skipped += 1
+                    continue
+                new_id = uuid.uuid4().hex
+                s.add(Batch(
+                    id=new_id,
+                    kind="837p",
+                    input_filename="<synthetic:orphan-reconcile>",
+                    parsed_at=now,
+                    transaction_set_control_number=row["st02"],
+                    totals_json=json.dumps({
+                        "orphan_reconcile": True,
+                        "ack_count": row["ack_count"],
+                    }),
+                    validation_json=json.dumps({
+                        "orphan_reconcile": True,
+                        "note": (
+                            "sp38 synthetic batch row; source 837 was "
+                            "never ingested into this DB snapshot"
+                        ),
+                    }),
+                ))
+                synthetic_ids.append(new_id)
+                created += 1
+            s.commit()
+
+        return {"created": created, "skipped": skipped, "synthetic_batch_ids": synthetic_ids}
+
     def remove_claim_ack(self, link_id, *, event_bus=None):
         """Unlink one row. Publishes ``claim_ack_dropped`` on the bus."""
         return _remove_claim_ack(link_id, event_bus=event_bus)

backend/src/cyclone/store/claim_acks.py

@@ -24,9 +24,10 @@ subscriber cannot roll back the persisted row.
 
 from __future__ import annotations
 
+import json
 import logging
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Iterator
 
 from cyclone import db
 from cyclone.db import (
@@ -276,97 +277,152 @@ def find_ack_orphans(kind: str) -> list[dict]:
     out: list[dict] = []
     if kind == "999":
         ack_table = Ack
-        ctrl_attr = None
     elif kind == "277ca":
         ack_table = Two77caAck
-        ctrl_attr = "control_number"
     else:
         ack_table = Ta1Ack
-        ctrl_attr = "control_number"
 
     with db.SessionLocal()() as s:
-        # Every ack row of the given kind, with a LEFT JOIN against
+        # Every ack row of the given kind; orphan when NO claim_acks
-        # any claim_acks link; orphan when NO link was created.
+        # link was created for it. The auto-linker emits one claim_acks
-        if kind == "999":
+        # row per AK2 even when the AK2 is rejected, so a 999 with at
-            # For 999, the "ack has no link" means no ClaimAck row
+        # least one linked AK2 is never an orphan.
-            # was emitted at all (the auto-linker emits one per AK2
+        for ack_row in s.query(ack_table).order_by(ack_table.id.desc()).all():
-            # even when the AK2 is rejected, so 999 with at least
+            count = (
-            # one AK2 that resolved to a claim is never an orphan).
+                s.query(ClaimAck)
-            # We treat a 999 as orphan when it has zero ClaimAck
+                .filter(ClaimAck.ack_kind == kind,
-            # rows tied to its id.
+                        ClaimAck.ack_id == ack_row.id)
-            all_acks = s.query(Ack).order_by(Ack.id.desc()).all()
+                .count()
-            for ack_row in all_acks:
+            )
-                count = (
+            if count > 0:
-                    s.query(ClaimAck)
+                continue
-                    .filter(ClaimAck.ack_kind == "999",
+            out.append({
-                            ClaimAck.ack_id == ack_row.id)
+                "kind": kind,
-                    .count()
+                "ack_id": ack_row.id,
-                )
+                "control_number": _ack_control_number(ack_row, kind),
-                if count == 0:
+                "parsed_at": (
-                    out.append({
+                    ack_row.parsed_at.isoformat().replace(
-                        "kind": "999",
+                        "+00:00", "Z"
-                        "ack_id": ack_row.id,
+                    ) if ack_row.parsed_at else None
-                        "control_number": _ack_control_number(ack_row, "999"),
+                ),
-                        "parsed_at": (
+            })
-                            ack_row.parsed_at.isoformat().replace(
-                                "+00:00", "Z"
-                            ) if ack_row.parsed_at else None
-                        ),
-                    })
-        elif kind == "277ca":
-            all_acks = s.query(Two77caAck).order_by(Two77caAck.id.desc()).all()
-            for ack_row in all_acks:
-                count = (
-                    s.query(ClaimAck)
-                    .filter(ClaimAck.ack_kind == "277ca",
-                            ClaimAck.ack_id == ack_row.id)
-                    .count()
-                )
-                if count == 0:
-                    out.append({
-                        "kind": "277ca",
-                        "ack_id": ack_row.id,
-                        "control_number": ack_row.control_number or "",
-                        "parsed_at": (
-                            ack_row.parsed_at.isoformat().replace(
-                                "+00:00", "Z"
-                            ) if ack_row.parsed_at else None
-                        ),
-                    })
-        else:
-            all_acks = s.query(Ta1Ack).order_by(Ta1Ack.id.desc()).all()
-            for ack_row in all_acks:
-                count = (
-                    s.query(ClaimAck)
-                    .filter(ClaimAck.ack_kind == "ta1",
-                            ClaimAck.ack_id == ack_row.id)
-                    .count()
-                )
-                if count == 0:
-                    out.append({
-                        "kind": "ta1",
-                        "ack_id": ack_row.id,
-                        "control_number": ack_row.control_number or "",
-                        "parsed_at": (
-                            ack_row.parsed_at.isoformat().replace(
-                                "+00:00", "Z"
-                            ) if ack_row.parsed_at else None
-                        ),
-                    })
     return out
 
 
-def _ack_control_number(ack_row: Ack, kind: str) -> str:
+def _iter_orphan_999_st02s() -> Iterator[tuple[str, int]]:
-    """Best-effort control-number lookup for a 999 ack row.
+    """Yield ``(set_control_number, orphan_count)`` for each distinct orphan 999.
 
-    The 999 ORM row doesn't carry the envelope's control_number in
+    An orphan 999 has zero ``claim_acks`` rows tied to its id (per
-    a dedicated column; we re-derive it from ``raw_json`` (the same
+    :func:`find_ack_orphans`). The ST02 is the source 837's
-    source :func:`cyclone.store.ui.to_ui_ack` uses for the patient
+    ``transaction_set_control_number``, extracted from the 999's
-    control number).
+    ``set_responses[0].set_control_number`` field in ``raw_json``.
+
+    Used by :meth:`CycloneStore.find_ack_orphan_st02_summary` and
+    :meth:`CycloneStore.reconcile_orphan_st02s` (sp38) to enumerate
+    orphan ST02s without re-implementing the orphan-detection query.
+
+    Skips 999s whose ``raw_json`` is malformed or has no
+    ``set_responses`` entry — those are unprocessable regardless of
+    whether they have a matching batch row.
+
+    999-only. 277ca / ta1 orphans are tracked separately by the
+    store and are not aggregated here (their "ST02" semantics differ
+    from 999: it's the interchange control number, not the source
+    837's ST02).
     """
-    raw = ack_row.raw_json or {}
+    with db.SessionLocal()() as s:
-    env = raw.get("envelope") or {}
+        # LEFT OUTER JOIN keeps all acks; orphan when no claim_acks row
-    return env.get("control_number") or ""
+        # exists for (ack_kind='999', ack_id=acks.id).
+        rows = (
+            s.query(Ack)
+            .outerjoin(
+                ClaimAck,
+                (ClaimAck.ack_kind == "999") & (ClaimAck.ack_id == Ack.id),
+            )
+            .filter(ClaimAck.id.is_(None))
+            .all()
+        )
+        counts: dict[str, int] = {}
+        for ack_row in rows:
+            st02 = _extract_999_st02(ack_row)
+            if st02 is None:
+                continue
+            counts[st02] = counts.get(st02, 0) + 1
+    # 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:
+    """Return the source 837's ST02 from a 999 ack row, or ``None``.
+
+    Reads ``raw_json`` and pulls ``set_responses[0].set_control_number``.
+    Returns ``None`` if the JSON is malformed, ``set_responses`` is
+    empty, or the field is missing — callers should skip these rows
+    rather than treating them as orphans (they're unprocessable, not
+    just orphaned).
+
+    Note: the ``raw_json`` column is a SQLAlchemy JSON type so the
+    ORM hands it back as a Python dict, not a string. We accept
+    either form (dict or str) so this helper is robust to direct
+    SQLAlchemy access and to raw sqlite3 row access.
+    """
+    raw = ack_row.raw_json
+    if not raw:
+        return None
+    if isinstance(raw, dict):
+        parsed = raw
+    elif isinstance(raw, (str, bytes, bytearray)):
+        try:
+            parsed = json.loads(raw)
+        except (TypeError, ValueError):
+            return None
+    else:
+        return None
+    srs = parsed.get("set_responses") or []
+    if not srs:
+        return None
+    st02 = srs[0].get("set_control_number")
+    return str(st02) if st02 else None
+
+
+def _ack_control_number(ack_row, kind: str) -> str:
+    """Best-effort control-number lookup for an ack row of any kind.
+
+    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.
+    """
+    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

@@ -0,0 +1,377 @@
+"""SP38: store-helper tests for ``find_ack_orphan_st02_summary``.
+
+The summary is the per-ST02 breakdown the ``cyclone ack-orphans``
+CLI commands consume. It enumerates every distinct orphan 999 ST02
++ ack count + whether a ``batches`` row already covers that ST02.
+
+Tests live here (not in ``test_apply_claim_ack_links.py``) because
+the helper is a sp38 surface, not an sp28 invariant. Keeping the
+tests in a sibling file matches the ``cyclone-tests`` convention.
+
+The autouse ``_auto_init_db`` fixture in ``conftest.py`` provides
+a fresh ``tmp_path/test.db`` for every test — no manual DB setup
+needed here.
+"""
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+
+import pytest
+
+from cyclone import db
+from cyclone.db import Ack, Batch, ClaimAck
+from cyclone.store import CycloneStore
+
+
+def _now():
+    """A single shared 'now' for deterministic test timestamps."""
+    return datetime.now(timezone.utc)
+
+
+def _ingest_999(s, st02: str, *, batch_id: str | None = None) -> int:
+    """Insert a 999 ack row whose set_responses[0].set_control_number == st02.
+
+    Returns the new ack id. The 999 is an orphan (no claim_acks row
+    is inserted) so it surfaces in the summary.
+    """
+    raw = {
+        "envelope": {
+            "sender_id": "SUBMITTERID",
+            "receiver_id": "RECEIVERID",
+            "control_number": "000000099",
+            "transaction_date": "2024-01-01",
+            "implementation_guide": "005010X231A1",
+        },
+        "functional_group_acks": [],
+        "set_responses": [
+            {"set_control_number": st02, "transaction_set_identifier": "837",
+             "ak2": {"functional_id_code": "837"}, "segment_errors": [],
+             "set_accept_reject": {"code": "A"}}
+        ],
+        "summary": {"accepted_count": 1, "rejected_count": 0},
+    }
+    ack = Ack(
+        source_batch_id=batch_id or f"999-{st02}-test",
+        accepted_count=1,
+        rejected_count=0,
+        received_count=1,
+        ack_code="A",
+        parsed_at=_now(),
+        raw_json=json.dumps(raw),
+    )
+    s.add(ack)
+    s.flush()
+    return ack.id
+
+
+def _seed_batch(s, *, st02: str, batch_id: str | None = None) -> str:
+    """Insert a batches row with the given ST02. Returns the batch id."""
+    import uuid
+    bid = batch_id or uuid.uuid4().hex
+    s.add(Batch(
+        id=bid,
+        kind="837p",
+        input_filename="test.837p",
+        transaction_set_control_number=st02,
+        parsed_at=_now(),
+    ))
+    s.flush()
+    return bid
+
+
+def test_summary_returns_per_st02_counts():
+    """Two distinct orphan ST02s surface as two separate summary rows."""
+    with db.SessionLocal()() as s:
+        _ingest_999(s, "991102989")
+        _ingest_999(s, "991102989")  # same ST02 twice
+        _ingest_999(s, "991102988")
+        s.commit()
+
+    summary = CycloneStore().find_ack_orphan_st02_summary()
+    by_st02 = {row["st02"]: row for row in summary}
+    assert by_st02["991102989"]["ack_count"] == 2
+    assert by_st02["991102988"]["ack_count"] == 1
+    assert by_st02["991102989"]["has_batch"] is False
+    assert by_st02["991102988"]["has_batch"] is False
+
+
+def test_summary_sets_has_batch_true_when_batches_row_exists():
+    """When a batches row has the orphan ST02, ``has_batch`` is True
+    and ``batch_id`` matches the batches row's id."""
+    with db.SessionLocal()() as s:
+        bid = _seed_batch(s, st02="991102977")
+        _ingest_999(s, "991102977", batch_id=bid)
+        s.commit()
+
+    summary = CycloneStore().find_ack_orphan_st02_summary()
+    assert len(summary) == 1
+    row = summary[0]
+    assert row["st02"] == "991102977"
+    assert row["ack_count"] == 1
+    assert row["has_batch"] is True
+    assert row["batch_id"] == bid
+
+
+def test_summary_sorted_by_ack_count_descending():
+    """The summary is sorted so the heaviest orphans surface first
+    in the CLI output (matches the spec's intent that operators
+    triage the biggest backlog first)."""
+    with db.SessionLocal()() as s:
+        for _ in range(3):
+            _ingest_999(s, "991102988")
+        _ingest_999(s, "991102987")
+        for _ in range(5):
+            _ingest_999(s, "991102989")
+        s.commit()
+
+    summary = CycloneStore().find_ack_orphan_st02_summary()
+    counts = [row["ack_count"] for row in summary]
+    assert counts == sorted(counts, reverse=True)
+    assert counts[0] == 5  # 991102989 has the most
+
+
+def test_summary_excludes_999s_that_have_a_claim_acks_link():
+    """A 999 with a claim_acks row is NOT an orphan and must not
+    appear in the summary."""
+    with db.SessionLocal()() as s:
+        linked_id = _ingest_999(s, "991102977")
+        _ingest_999(s, "991102988")  # orphan, no link
+        # Manually insert a claim_acks link for the first 999.
+        s.add(ClaimAck(
+            claim_id="CLM-1",
+            batch_id="b1",
+            ack_id=linked_id,
+            ack_kind="999",
+            ak2_index=0,
+            set_control_number="991102977",
+            set_accept_reject_code="A",
+            linked_at=_now(),
+            linked_by="auto",
+        ))
+        s.commit()
+
+    summary = CycloneStore().find_ack_orphan_st02_summary()
+    by_st02 = {row["st02"]: row for row in summary}
+    assert "991102977" not in by_st02  # linked → not orphan
+    assert by_st02["991102988"]["ack_count"] == 1
+
+
+def test_summary_skips_999s_with_malformed_raw_json():
+    """999s whose raw_json is missing set_responses or is malformed
+    JSON are SKIPPED — they can't be reconciled, so they don't
+    contribute to the summary."""
+    with db.SessionLocal()() as s:
+        # Valid orphan
+        _ingest_999(s, "991102988")
+        # Malformed JSON — raw_json is not parseable
+        s.add(Ack(
+            source_batch_id="999-malformed",
+            accepted_count=1, rejected_count=0, received_count=1,
+            ack_code="A",
+            parsed_at=_now(),
+            raw_json="{not valid json",
+        ))
+        # Empty set_responses
+        s.add(Ack(
+            source_batch_id="999-empty",
+            accepted_count=1, rejected_count=0, received_count=1,
+            ack_code="A",
+            parsed_at=_now(),
+            raw_json=json.dumps({"envelope": {}, "set_responses": [], "summary": {}}),
+        ))
+        s.commit()
+
+    summary = CycloneStore().find_ack_orphan_st02_summary()
+    by_st02 = {row["st02"]: row for row in summary}
+    assert set(by_st02.keys()) == {"991102988"}
+    assert by_st02["991102988"]["ack_count"] == 1
+
+
+def test_summary_empty_when_no_orphans():
+    """With zero orphans, the summary is an empty list."""
+    summary = CycloneStore().find_ack_orphan_st02_summary()
+    assert summary == []
+
+
+# ---- Task 3: reconcile_orphan_st02s --------------------------------------- #
+
+
+def test_reconcile_creates_synthetic_batches_for_missing_st02s():
+    """``reconcile_orphan_st02s`` inserts one synthetic batch row per
+    orphan ST02 that doesn't already have a batches row.
+
+    Each synthetic row uses the sentinel ``input_filename`` and
+    ``kind = '837p'`` so they're trivially distinguishable in
+    queries (the spec: grep for ``<synthetic:orphan-reconcile>``
+    to find every row this SP38 created).
+    """
+    with db.SessionLocal()() as s:
+        _ingest_999(s, "991102989")
+        _ingest_999(s, "991102988")
+        s.commit()
+
+    plan = CycloneStore().reconcile_orphan_st02s(dry_run=False)
+
+    assert plan["created"] == 2
+    assert plan["skipped"] == 0
+    assert len(plan["synthetic_batch_ids"]) == 2
+
+    with db.SessionLocal()() as s:
+        rows = (
+            s.query(Batch)
+            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
+            .all()
+        )
+        assert len(rows) == 2
+        assert all(r.kind == "837p" for r in rows)
+        st02s = {r.transaction_set_control_number for r in rows}
+        assert st02s == {"991102989", "991102988"}
+
+
+def test_reconcile_skips_st02s_that_already_have_a_batch():
+    """``reconcile`` does NOT create a synthetic row for an ST02
+    that already has a batches row — the operator's intent is to
+    fill gaps, not duplicate coverage."""
+    with db.SessionLocal()() as s:
+        existing = _seed_batch(s, st02="991102977")
+        _ingest_999(s, "991102977", batch_id=existing)
+        _ingest_999(s, "991102988")  # orphan, no batch
+        s.commit()
+
+    plan = CycloneStore().reconcile_orphan_st02s(dry_run=False)
+
+    assert plan["created"] == 1
+    assert plan["skipped"] == 1
+
+    with db.SessionLocal()() as s:
+        synthetic = (
+            s.query(Batch)
+            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
+            .all()
+        )
+        assert len(synthetic) == 1
+        assert synthetic[0].transaction_set_control_number == "991102988"
+
+
+def test_reconcile_is_idempotent():
+    """Re-running reconcile after a successful pass is a no-op:
+    ``created=0, skipped=N``."""
+    with db.SessionLocal()() as s:
+        _ingest_999(s, "991102989")
+        s.commit()
+
+    CycloneStore().reconcile_orphan_st02s(dry_run=False)
+    plan2 = CycloneStore().reconcile_orphan_st02s(dry_run=False)
+
+    assert plan2["created"] == 0
+    assert plan2["skipped"] == 1
+
+    with db.SessionLocal()() as s:
+        n = (
+            s.query(Batch)
+            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
+            .count()
+        )
+        assert n == 1
+
+
+def test_reconcile_dry_run_does_not_write():
+    """``dry_run=True`` returns the same plan shape but does not
+    insert any rows. Operators use this to preview the reconcile
+    before committing."""
+    with db.SessionLocal()() as s:
+        _ingest_999(s, "991102989")
+        s.commit()
+
+    plan = CycloneStore().reconcile_orphan_st02s(dry_run=True)
+
+    assert plan["created"] == 1
+    assert plan["skipped"] == 0
+
+    with db.SessionLocal()() as s:
+        n = (
+            s.query(Batch)
+            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
+            .count()
+        )
+        assert n == 0  # no row inserted
+
+
+def test_reconcile_records_ack_count_in_totals_json():
+    """The synthetic row's ``totals_json`` includes ``ack_count`` so
+    future operators can see the orphan weight without re-querying."""
+    import json
+    with db.SessionLocal()() as s:
+        for _ in range(4):
+            _ingest_999(s, "991102989")
+        s.commit()
+
+    CycloneStore().reconcile_orphan_st02s(dry_run=False)
+
+    with db.SessionLocal()() as s:
+        row = (
+            s.query(Batch)
+            .filter(Batch.transaction_set_control_number == "991102989")
+            .one()
+        )
+        totals = json.loads(row.totals_json or "{}")
+        assert totals.get("orphan_reconcile") is True
+        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_ack_orphans_cli.py

@@ -0,0 +1,210 @@
+"""SP38: CLI tests for ``cyclone ack-orphans {status,reconcile}``.
+
+Uses ``click.testing.CliRunner`` (matching the SP37-followup #5
+pattern that replaced ``subprocess.run`` with the in-process
+runner). The CLI commands are thin wrappers over the store helpers
+tested in ``test_ack_orphan_summary.py`` — these tests pin the
+shell-facing shape (table format, exit codes, --dry-run flag).
+
+The autouse ``_auto_init_db`` fixture in ``conftest.py`` provides
+a fresh ``tmp_path/test.db`` for every test; the CLI commands pick
+it up via the ``CYCLONE_DB_URL`` env var the fixture sets.
+"""
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+
+import pytest
+from click.testing import CliRunner
+
+from cyclone import db
+from cyclone.cli import main
+from cyclone.db import Ack, Batch
+
+
+def _now():
+    return datetime.now(timezone.utc)
+
+
+def _ingest_orphan(s, st02: str) -> int:
+    """Insert a 999 ack row whose set_responses[0].set_control_number == st02.
+
+    Returns the new ack id. The 999 is an orphan (no claim_acks row)
+    so it surfaces in the summary.
+    """
+    raw = {
+        "envelope": {"sender_id": "S", "receiver_id": "R",
+                     "control_number": "000000099", "transaction_date": "2024-01-01",
+                     "implementation_guide": "005010X231A1"},
+        "functional_group_acks": [],
+        "set_responses": [
+            {"set_control_number": st02, "transaction_set_identifier": "837",
+             "ak2": {"functional_id_code": "837"}, "segment_errors": [],
+             "set_accept_reject": {"code": "A"}}
+        ],
+        "summary": {"accepted_count": 1, "rejected_count": 0},
+    }
+    ack = Ack(
+        source_batch_id=f"999-{st02}-cli-test",
+        accepted_count=1, rejected_count=0, received_count=1,
+        ack_code="A", parsed_at=_now(), raw_json=json.dumps(raw),
+    )
+    s.add(ack)
+    s.flush()
+    return ack.id
+
+
+# ---- cyclone ack-orphans status ------------------------------------------ #
+
+
+def test_status_prints_table_with_orphans():
+    """``cyclone ack-orphans status`` prints a table with ST02 +
+    ack count + has-batch flag, plus a TOTAL line at the bottom.
+
+    Mirrors the spec's intent: operators want a one-line-per-orphan
+    view that ranks the heaviest backlog first.
+    """
+    with db.SessionLocal()() as s:
+        _ingest_orphan(s, "991102989")
+        _ingest_orphan(s, "991102988")
+        s.commit()
+
+    runner = CliRunner()
+    result = runner.invoke(main, ["ack-orphans", "status"])
+    assert result.exit_code == 0, result.output
+    # Table header + two data rows + total line.
+    assert "ST02" in result.output
+    assert "ACK COUNT" in result.output
+    assert "991102989" in result.output
+    assert "991102988" in result.output
+    assert "TOTAL" in result.output
+
+
+def test_status_exits_zero_on_empty_db():
+    """With no orphans, status exits 0 and prints a 'no orphans' note."""
+    runner = CliRunner()
+    result = runner.invoke(main, ["ack-orphans", "status"])
+    assert result.exit_code == 0, result.output
+    # The empty case should be informative, not silent.
+    assert "no orphans" in result.output.lower() or "0" in result.output
+
+
+def test_status_table_includes_has_batch_column():
+    """When a ST02 already has a batches row, the table marks it
+    with ``yes`` (or similar) so operators can see which orphans
+    are unbacked vs already-covered.
+    """
+    with db.SessionLocal()() as s:
+        # Orphan with NO batch.
+        _ingest_orphan(s, "991102988")
+        # Orphan WITH a pre-existing batch row.
+        import uuid
+        bid = uuid.uuid4().hex
+        s.add(Batch(
+            id=bid, kind="837p", input_filename="test.837p",
+            transaction_set_control_number="991102977", parsed_at=_now(),
+        ))
+        _ingest_orphan(s, "991102977")
+        s.commit()
+
+    runner = CliRunner()
+    result = runner.invoke(main, ["ack-orphans", "status"])
+    assert result.exit_code == 0, result.output
+    # Look for "yes" / "no" markers in the table.
+    lines = result.output.splitlines()
+    has_yes = any("yes" in ln.lower() for ln in lines)
+    has_no = any("no" in ln.lower() for ln in lines)
+    assert has_yes and has_no, (
+        f"Expected table to mark 'yes' for ST02s with batches and "
+        f"'no' for ST02s without. Got:\n{result.output}"
+    )
+
+
+# ---- cyclone ack-orphans reconcile --------------------------------------- #
+
+
+def test_reconcile_creates_synthetic_batches():
+    """``cyclone ack-orphans reconcile`` inserts a synthetic batch
+    row for each orphan ST02 without one. Confirms the count + the
+    sentinel filename.
+    """
+    with db.SessionLocal()() as s:
+        _ingest_orphan(s, "991102989")
+        _ingest_orphan(s, "991102988")
+        s.commit()
+
+    runner = CliRunner()
+    result = runner.invoke(main, ["ack-orphans", "reconcile"])
+    assert result.exit_code == 0, result.output
+
+    # The CLI should print the created/skipped counts.
+    assert "Created" in result.output or "created" in result.output
+    assert "Skipped" in result.output or "skipped" in result.output
+
+    with db.SessionLocal()() as s:
+        synthetic = (
+            s.query(Batch)
+            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
+            .all()
+        )
+        assert len(synthetic) == 2
+        st02s = {r.transaction_set_control_number for r in synthetic}
+        assert st02s == {"991102989", "991102988"}
+
+
+def test_reconcile_dry_run_does_not_write():
+    """``--dry-run`` returns the plan shape but does NOT insert rows.
+    Operators use this to preview before committing.
+    """
+    with db.SessionLocal()() as s:
+        _ingest_orphan(s, "991102989")
+        s.commit()
+
+    runner = CliRunner()
+    result = runner.invoke(main, ["ack-orphans", "reconcile", "--dry-run"])
+    assert result.exit_code == 0, result.output
+    # The plan reports the would-create count.
+    assert "1" in result.output  # would-create 1 row
+
+    # But no synthetic batch row was inserted.
+    with db.SessionLocal()() as s:
+        n = (
+            s.query(Batch)
+            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
+            .count()
+        )
+        assert n == 0
+
+
+def test_reconcile_is_idempotent():
+    """A second reconcile after a successful pass is a no-op:
+    created=0, skipped=N. Tested at the CLI shell level so the
+    user-visible output also pins this invariant.
+    """
+    with db.SessionLocal()() as s:
+        _ingest_orphan(s, "991102989")
+        s.commit()
+
+    runner = CliRunner()
+    first = runner.invoke(main, ["ack-orphans", "reconcile"])
+    second = runner.invoke(main, ["ack-orphans", "reconcile"])
+    assert first.exit_code == 0
+    assert second.exit_code == 0
+
+    with db.SessionLocal()() as s:
+        n = (
+            s.query(Batch)
+            .filter(Batch.input_filename == "<synthetic:orphan-reconcile>")
+            .count()
+        )
+        assert n == 1  # still exactly one — second call was a no-op
+
+
+def test_reconcile_handles_no_orphans_gracefully():
+    """With zero orphans, reconcile exits 0 and prints an informative
+    message (no synthetic rows created, no error)."""
+    runner = CliRunner()
+    result = runner.invoke(main, ["ack-orphans", "reconcile"])
+    assert result.exit_code == 0, result.output
+    assert "0" in result.output  # created=0, skipped=0

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
 # ---------------------------------------------------------------------------

docs/RUNBOOK.md

@@ -291,3 +291,77 @@ curl -X PATCH http://127.0.0.1:8000/api/clearhouse \
 ```
 
 To revert, set `stub: true` and `auth: {"method": "keychain", "secret_ref": "sftp.gainwell.password"}`.
+
+## Known historical drift — the 804 orphan 999s
+
+The `acks` table may hold several hundred 999 acks whose source 837
+batch is not present in the `batches` table — the Inbox "Ack orphans"
+lane surfaces them at `GET /api/inbox/ack-orphans`. These are real
+production 999s (e.g. sender_id `COMEDASSISTPROG`) whose source 837s
+were submitted to HPE clearinghouse before the current
+`~/.local/share/cyclone/cyclone.db` snapshot was created. The source
+837s themselves were never re-ingested into the current DB, so the
+`claims` table has no rows that could link against them. SP37's
+canonical submit-batch flow captures ST02 going forward, so the
+orphan count is stable — not a forward-looking bug, just historical
+drift.
+
+**You cannot auto-link these orphans.** The source 837s are not in
+`ingest/`, `backend/var/sftp/staging/`, or any local path — they
+were transmitted to HPE and never came back. Cyclone is downstream
+of the clearinghouse and does not retain copies of outbound 837s
+after SFTP ACK. Do not attempt to re-ingest from SFTP inbound —
+those files are the 999 acks themselves, not the source 837s.
+
+### Triage path
+
+1. **Inspect the Inbox > AckOrphansLane** in the UI. Each row is a
+   999 ack with no resolvable claim. Sort by `parsed_at DESC` to
+   see the most recent first; older orphans are less likely to be
+   actionable.
+2. **Decide per row.** If the operator can identify the source 837
+   outside Cyclone (e.g. a manual record of what was submitted that
+   day), manually create a `claims` row + a `claim_acks` link via
+   `POST /api/parse-837` followed by `POST /api/inbox/candidates/{remit_id}/match`
+   or `POST /api/acks/.../match-claim`. If not, leave the orphan
+   alone — it stays as valid audit history.
+
+### Optional: seed synthetic batch rows (one-shot)
+
+`cyclone ack-orphans reconcile` inserts a synthetic `batches` row
+for every orphan ST02 that doesn't already have one. The synthetic
+row is marked with `input_filename = '<synthetic:orphan-reconcile>'`
+so it's trivially distinguishable in queries. Future 999 acks for
+the same ST02s will resolve against the batch envelope index (so
+the operator can see "this 999 is for a known orphan source") but
+will not link to claims (because no claim rows exist).
+
+```bash
+# Preview the reconcile without writing rows.
+cyclone ack-orphans reconcile --dry-run
+
+# Insert the synthetic batch rows.
+cyclone ack-orphans reconcile
+```
+
+Re-running `cyclone ack-orphans reconcile` after a successful pass
+is a no-op (idempotent). To inspect the per-ST02 breakdown at any
+time:
+
+```bash
+cyclone ack-orphans status
+```
+
+The status command prints a table with ST02, ACK COUNT, and HAS
+BATCH columns, ranked by ack count so the heaviest backlog
+surfaces first.
+
+### What this is NOT
+
+- **Not a backfill.** No `claims` rows are synthesized — the source
+  data is gone.
+- **Not auto-runnable.** The reconcile CLI is operator-invoked only;
+  it does not run on boot, in the SFTP polling scheduler, or via
+  cron.
+- **Not a deletion.** The orphan 999s are valid audit history and
+  must remain queryable.

docs/superpowers/plans/2026-07-07-cyclone-orphan-ack-housekeeping.md

@@ -0,0 +1,214 @@
+# Orphan-ack housekeeping Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use
+> superpowers:subagent-driven-development (recommended) or
+> superpowers:executing-plans to implement this plan task-by-task. Steps
+> use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Surface the 804 historical orphan acks to operators via a
+RUNBOOK entry + CLI, and provide a one-shot idempotent synthetic-batch
+seeder so future acks for those ST02s can resolve against the
+`batch_envelope_index` instead of remaining forever-orphans.
+
+**Architecture:** Two thin CLI subcommands under `cyclone ack-orphans
+{status,reconcile}` backed by a single store helper
+`find_ack_orphan_st02_summary()`. The helper returns a structured
+list[dict]; the CLI formats it. Reconcile uses the same SQLAlchemy
+session the rest of the store uses, so it inherits the DB-first
+invariant (no direct sqlite3 calls, no parallel writers).
+
+**Tech Stack:** Python 3.11, FastAPI, SQLAlchemy, Click (CLI),
+pytest, the existing `backend/tests/conftest.py` autouse fixtures.
+
+**Spec:** [`docs/superpowers/specs/2026-07-07-cyclone-orphan-ack-housekeeping-design.md`](../specs/2026-07-07-cyclone-orphan-ack-housekeeping-design.md)
+
+---
+
+## File structure
+
+```
+backend/
+  src/cyclone/
+    cli.py                              # + ack-orphans status + reconcile subcommands
+    store/
+      __init__.py                       # + find_ack_orphan_st02_summary() facade method
+      claim_acks.py                     # extract _iter_orphan_st02s() helper
+  tests/
+    test_ack_orphan_summary.py          # NEW: store-helper tests
+    test_ack_orphans_cli.py             # NEW: CLI tests via CliRunner
+docs/
+  RUNBOOK.md                            # + "Known historical drift" section
+```
+
+---
+
+## Task 1: Extract `_iter_orphan_st02s()` helper from `find_ack_orphans`
+
+The current `find_ack_orphans(kind)` walks the `acks` table directly
+and does its own LEFT JOIN against `claim_acks`. Both the new store
+helper (`find_ack_orphan_st02_summary`) and the existing
+`find_ack_orphans` need to enumerate the orphan ST02s. Extract the
+walk into a private helper in `claim_acks.py` that returns a flat
+list of `(st02, ack_count)` tuples — the only join key needed for
+the summary, and the seed for `find_ack_orphans`'s per-kind filter.
+
+- [ ] Read `backend/src/cyclone/store/claim_acks.py:250-330` to confirm
+      the current shape of `find_ack_orphans`.
+- [ ] Add `_iter_orphan_st02s() -> Iterator[tuple[str, int]]` to
+      `claim_acks.py`. One SELECT against `acks` joining
+      `claim_acks` on `(claim_acks.ack_kind = '999' AND
+      claim_acks.ack_id = acks.id)`, yielding `(set_control_number,
+      ack_count)` for each group where no claim_acks row exists.
+      For 277ca / ta1 the same shape but with the matching ack table
+      and `claim_acks.ack_kind` filter.
+- [ ] Refactor `find_ack_orphans(kind)` to consume the helper.
+      No behavior change; same return shape, same idempotent
+      semantics. All existing tests must still pass.
+
+**RED→GREEN test gate:** `cd backend && .venv/bin/pytest
+tests/test_api_claim_acks.py tests/test_apply_claim_ack_links.py -v`
+must remain green.
+
+## Task 2: Add `find_ack_orphan_st02_summary()` to the store facade
+
+- [ ] Add `find_ack_orphan_st02_summary(self) -> list[dict]` to
+      `CycloneStore` in `backend/src/cyclone/store/__init__.py`.
+      Returns `[{"st02": str, "ack_count": int,
+      "has_batch": bool, "batch_id": str | None}, ...]`.
+- [ ] `has_batch` is `True` if a row exists in `batches` with
+      `transaction_set_control_number = st02`. `batch_id` is that
+      row's `id` (or `None`).
+- [ ] Sort the result by `ack_count DESC` so the heaviest orphans
+      surface first in the CLI output.
+
+**RED test first:** write
+`backend/tests/test_ack_orphan_summary.py::test_summary_returns_per_st02_counts`
+with a seeded DB (two orphan ST02s, one with a batch, one without)
+and assert the shape. Watch it fail (no facade method yet), then
+implement to GREEN.
+
+## Task 3: Test the synthetic-batch seeder
+
+The reconcile CLI inserts synthetic `batches` rows for any orphan
+ST02 that doesn't already have one. This is a write; the store
+helper needs a paired write-side method.
+
+- [ ] Add `reconcile_orphan_st02s(self, dry_run: bool = False) ->
+      dict` to `CycloneStore`. Returns `{"created": int, "skipped":
+      int, "synthetic_batch_ids": list[str]}`.
+- [ ] For each ST02 in the summary where `has_batch is False`, insert
+      a `batches` row with: `kind = '837p'`, `input_filename =
+      '<synthetic:orphan-reconcile>'`, `totals_json =
+      '{"orphan_reconcile": true, "ack_count": N}'`,
+      `validation_json = '{"orphan_reconcile": true, "note":
+      "sp38 synthetic batch row; source 837 was never ingested into
+      this DB snapshot"}'`, `id = uuid4().hex`. The remaining
+      columns take their defaults.
+- [ ] When `dry_run=True`, return the plan without writing.
+- [ ] Idempotency: re-running after a previous reconcile returns
+      `created=0, skipped=N`. Test this explicitly.
+
+**RED test:** `test_reconcile_is_idempotent` — run reconcile twice,
+assert the row count is unchanged after the second call.
+
+**RED test:** `test_reconcile_uses_synthetic_input_filename_sentinel`
+— assert every created row has `input_filename =
+'<synthetic:orphan-reconcile>'`.
+
+**RED test:** `test_reconcile_dry_run_does_not_write` — pass
+`dry_run=True`, assert `created == N` in the plan but
+`batches` row count is unchanged.
+
+## Task 4: Add `cyclone ack-orphans status` CLI subcommand
+
+- [ ] Open `backend/src/cyclone/cli.py`, locate the existing
+      `submit-batch` group.
+- [ ] Add an `ack-orphans` group with a `status` subcommand.
+- [ ] `status` calls `cycl_store.find_ack_orphan_st02_summary()`,
+      prints a table: `ST02 | ACK COUNT | HAS BATCH`. Total line at
+      the bottom.
+- [ ] Exit 0 on success, exit 1 on DB error (matching the
+      `cyclone-cli` convention: 1 = unexpected exception, including
+      SQLAlchemy DB errors).
+
+**RED test:** `tests/test_ack_orphans_cli.py::test_status_prints_table`
+using `CliRunner.invoke(["ack-orphans", "status"])`. Assert exit
+code 0, assert the table contains the seeded ST02s and the total
+line.
+
+## Task 5: Add `cyclone ack-orphans reconcile` CLI subcommand
+
+- [ ] Same `ack-orphans` group, `reconcile` subcommand.
+- [ ] Accepts `--dry-run` flag (default False).
+- [ ] Calls `cycl_store.reconcile_orphan_st02s(dry_run=dry_run)`,
+      prints `Created: N synthetic batch rows. Skipped: M (already
+      had a batch row).`.
+- [ ] Exit 0 on success, exit 1 on DB error (same convention as
+      `status`).
+
+**RED test:** `tests/test_ack_orphans_cli.py::test_reconcile_creates_synthetic_batches`
+— seed two orphan ST02s (one with batch, one without), invoke
+`reconcile`, assert `created=1`, assert a row with
+`input_filename='<synthetic:orphan-reconcile>'` exists.
+
+**RED test:** `tests/test_ack_orphans_cli.py::test_reconcile_dry_run_flag`
+— invoke `reconcile --dry-run`, assert `created=1` in plan but
+no DB row inserted.
+
+**RED test:** `tests/test_ack_orphans_cli.py::test_reconcile_is_idempotent`
+— invoke `reconcile` twice, assert the second invocation reports
+`created=0`.
+
+## Task 6: Add RUNBOOK.md entry
+
+- [ ] Open `docs/RUNBOOK.md`, find the existing "Operator triage"
+      section.
+- [ ] Add a `## Known historical drift` subsection under it.
+- [ ] Content (3-4 paragraphs):
+    - What the 804 orphans are (real production 999s whose source
+      837s pre-date the current DB snapshot).
+    - Why they cannot be auto-linked (the source 837s were not
+      preserved).
+    - How to triage them via the Inbox > AckOrphansLane.
+    - Optional: `cyclone ack-orphans status` to see the ST02
+      breakdown, `cyclone ack-orphans reconcile` to seed synthetic
+      batch rows for them.
+
+## Task 7: Live verification + autoreview + commit
+
+- [ ] Run the full backend test suite: `cd backend && .venv/bin/pytest`.
+      Confirm pre-existing failure count (20) is unchanged.
+- [ ] Live-run `cyclone ack-orphans status` against the current
+      `~/.local/share/cyclone/cyclone.db`. Confirm output matches the
+      SQLite investigation: 5 distinct ST02s, 805 total orphan rows.
+- [ ] Run autoreview (subagent) on the diff.
+- [ ] Commit with the SP-N commit prefixes. Per the SP38 spec, the
+      work is: store-helper extraction (Tasks 1-3), CLI (Tasks 4-5),
+      docs (Task 6). Three commits, all prefixed
+      `feat(sp38): …`. Plus a `merge: SP38 orphan-ack housekeeping
+      into main` commit at the end of the review cycle.
+
+---
+
+## Anti-patterns
+
+- **Don't add a UI for the CLI commands.** Spec is explicit: no UI
+  change.
+- **Don't auto-run reconcile on boot.** Operators invoke it
+  manually after they accept the drift.
+- **Don't try to backfill claim rows.** Source data is gone; no
+  way to fabricate claim rows that would be meaningful.
+- **Don't put the synthetic sentinel in a constant somewhere.** The
+  string `<synthetic:orphan-reconcile>` appears in the spec, the
+  plan, the store helper, the CLI output, and the seeded batches
+  table. Keep it inline so a future grep finds it.
+
+## Related skills to load while implementing
+
+- `cyclone-spec` — already loaded (this plan is its output).
+- `cyclone-cli` — for the `click.testing.CliRunner` pattern and the
+  exit-code convention.
+- `cyclone-store` — for the facade re-export pattern in
+  `store/__init__.py`.
+- `cyclone-tests` — for the `tmp_path/test.db` conftest pattern and
+  the BACKFILL_SQL-loads-from-file pattern from SP37 followup #1.

docs/superpowers/specs/2026-07-07-cyclone-orphan-ack-housekeeping-design.md

@@ -0,0 +1,187 @@
+# Sub-project 38 — Orphan-ack housekeeping: Design Spec
+
+**Date:** 2026-07-07
+**Status:** Draft, awaiting user sign-off
+**Branch:** `sp38-orphan-ack-housekeeping`
+**Aesthetic direction:** No new UI
+
+## 1. Scope
+
+Cyclone's `acks` table currently holds 805 rows of which 804 are
+unresolved "orphans" — 999 acks that reference source 837 batches
+whose `transaction_set_control_number` (ST02) does not appear in any
+row of the `batches` table. Investigation on 2026-07-07 surfaced the
+root cause: the orphan rows are real production 999s (sender_id =
+`COMEDASSISTPROG`) whose source 837s were submitted to HPE clearinghouse
+by a prior state of the codebase (or an upstream system) before the
+current `cyclone.db` snapshot was created. The source 837s themselves
+were never re-ingested into the current DB, so the `claims` table has
+no rows that could link against these acks. The orphan count is a
+historical-data artifact, not a forward-looking bug; the SP37 canonical
+submit-batch flow already captures ST02 going forward, so the count
+stays at 804 and does not grow.
+
+This SP-N captures that situation, surfaces it to operators via a
+RUNBOOK entry, and adds a one-shot housekeeping helper so operators
+can (a) confirm the orphans are stable and (b) optionally seed
+synthetic batch rows for the orphan ST02s so that *future* acks for
+the same ST02s can resolve against the `batch_envelope_index` instead
+of remaining forever-orphans.
+
+**In scope:**
+
+- A RUNBOOK.md entry under "Known historical drift" describing the
+  804-orphan root cause, the operator's triage path (Inbox >
+  AckOrphansLane, already working as of SP37 followup commit
+  `893a662`), and the choice to accept the drift.
+- A `cyclone ack-orphans status` CLI subcommand that prints the
+  distinct orphan ST02s + ack count per ST02 + the total orphan count
+  + the count of orphan rows whose ST02 matches a `batches` row vs
+  those that don't. Deterministic, exit 0 on success / 1 on DB
+  error (matching the `cyclone-cli` skill convention: 1 =
+  unexpected exception, including DB-side errors).
+- A `cyclone ack-orphans reconcile` CLI subcommand that creates
+  synthetic `batches` rows for each distinct orphan ST02 that does
+  NOT already exist in `batches`. Synthetic rows are marked with
+  `kind = '837p'`, `input_filename = '<synthetic:orphan-reconcile>'`,
+  `totals_json = '{"orphan_reconcile": true, "ack_count": N}'`, and
+  `validation_json = '{"orphan_reconcile": true, "note": "synthetic
+  batch row created by sp38 to allow future acks for ST02 X to resolve
+  via batch_envelope_index; the original 837 source data was never
+  ingested into this DB snapshot"}'`. Idempotent: re-running on the
+  same DB does not create duplicate synthetic rows.
+- A pure read-side helper `cyclone.store.find_ack_orphan_st02_summary()`
+  that returns the per-ST02 summary the CLI consumes; the helper is
+  the testable surface and the CLI is a thin wrapper.
+- Tests for both CLI subcommands + the store helper, using the
+  existing `tmp_path/test.db` conftest pattern.
+
+**Out of scope:**
+
+- Re-ingesting the original 837 source files. The source 837s are
+  not in `ingest/`, `backend/var/sftp/staging/`, or any local path —
+  they were transmitted to HPE and never came back. Cyclone is
+  downstream of the clearinghouse and does not retain copies of
+  outbound 837s after SFTP ACK.
+- Auto-linking the 804 orphan acks to `claims`. There are no claim
+  rows for the orphan ST02s (the `claims` table has only 2 rows in
+  the current DB snapshot, both for ST02 `991102977` which is *not*
+  an orphan). Auto-linking is not possible without source data.
+- A UI for the `ack-orphans status` / `reconcile` commands. The
+  Inbox AckOrphansLane (post-`893a662`) is the operator's view of
+  the orphans; the CLI is for ad-hoc investigation and the one-shot
+  reconciliation.
+- Removing or archiving the 804 orphan acks. They are valid audit
+  history (real production traffic was acknowledged by HPE) and
+  must remain queryable.
+- Any change to the `claim_acks` join logic or the
+  `batch_envelope_index`. The orphan detection behavior is correct
+  as of SP37; this SP only adds housekeeping around the existing
+  behavior.
+
+## 2. Decisions (locked during brainstorming)
+
+**D1: Accept the historical drift, do not backfill.**
+
+The 804 orphans reflect a database snapshot that is younger than the
+traffic that produced the acks. The source 837s are not recoverable.
+SP37's canonical submit-batch flow captures ST02 going forward, so
+the count will stay at 804 + future-test-runs rather than grow
+indefinitely. The right operator posture is to acknowledge the drift
+in RUNBOOK.md and surface it via the existing Inbox AckOrphansLane.
+
+**D2: Synthetic batch rows are `kind = '837p'` with a distinct
+`input_filename` sentinel.**
+
+This makes them trivially distinguishable from real ingest batches
+in queries (e.g. `WHERE input_filename LIKE '<synthetic:%>'`). The
+sentinel is `<synthetic:orphan-reconcile>` so any future contributor
+who sees these rows in the DB can grep the codebase for that string
+and find this spec.
+
+**D3: Synthetic rows get NO claims, NO `claim_acks` links.**
+
+The reconciliation pass creates batch rows but does not synthesize
+claim rows for them. The `claims` table stays accurate to what was
+actually ingested. Future acks referencing these synthetic ST02s
+will resolve against the batch envelope index (so the operator can
+see "this 999 is for a known orphan source") but will not link to
+claims.
+
+**D4: Reconcile is idempotent, not auto-runnable.**
+
+The CLI does not auto-run on boot, in the SFTP polling scheduler,
+or via cron. Operators run it manually after they confirm the drift
+is acceptable. Idempotency means a second run is a no-op rather
+than an error.
+
+**D5: Both CLI commands live under `cyclone ack-orphans`.**
+
+This groups them under a shared verb, matching the existing
+`cyclone submit-batch`, `cyclone parse-999`, `cyclone backup`
+subcommand shape. The CLI file is `backend/src/cyclone/cli.py`
+(where `submit-batch` already lives); no new top-level CLI module.
+
+**D6: The store helper is the testable surface; the CLI is a thin
+wrapper.**
+
+`find_ack_orphan_st02_summary() -> list[dict]` is what the tests
+target. The CLI parses flags, calls the helper, formats output,
+sets the exit code. Mirrors the existing `cyclone-cli` convention.
+
+## 3. Open questions
+
+None. The operator has confirmed the design via the brainstorming
+Q&A on 2026-07-07: docs + housekeeping helper + synthetic-batch
+migration, no UI change, no auto-runnable reconcile.
+
+## 4. Test impact
+
+Per `cyclone-tests` (autouse conftest at `backend/tests/conftest.py`):
+
+- `backend/tests/test_ack_orphan_summary.py` — new, tests the store
+  helper directly. 999-only (matches the helper's scope; 277ca /
+  ta1 orphans are tracked by `find_ack_orphans(kind)` but their
+  "ST02" semantics differ — see §1 — so they are excluded from
+  the per-ST02 summary). Asserts the summary shape, asserts
+  idempotency of the reconcile insert, asserts the sentinel
+  `input_filename` is preserved, asserts the `kind = '837p'`
+  invariant.
+- `backend/tests/test_ack_orphans_cli.py` — new, tests the CLI
+  subcommands via `click.testing.CliRunner` (matching the
+  SP37-followup #5 pattern). Asserts exit codes (0 / 1), stdout
+  shape, idempotency of reconcile.
+- No frontend test impact (no UI change).
+- No migration test impact (no schema change; the synthetic batch
+  rows are inserted via SQLAlchemy at runtime, not via the
+  `migrations/` directory).
+
+## 5. Files expected to change
+
+- `docs/RUNBOOK.md` — append "Known historical drift" section under
+  the existing "Operator triage" section.
+- `backend/src/cyclone/store/__init__.py` — add
+  `CycloneStore.find_ack_orphan_st02_summary` + a private
+  `_reconcile_orphan_st02` helper. Re-export through the
+  facade so callers don't need to import from the subpackage
+  directly.
+- `backend/src/cyclone/store/claim_acks.py` — extract the orphan
+  ST02 walk from `find_ack_orphans` into a reusable helper that
+  both `find_ack_orphans` and `find_ack_orphan_st02_summary` can
+  call. Avoids duplicate SQL.
+- `backend/src/cyclone/cli.py` — add the two subcommands.
+- `backend/tests/test_ack_orphan_summary.py` — new, store-helper
+  tests.
+- `backend/tests/test_ack_orphans_cli.py` — new, CLI tests.
+- `docs/superpowers/plans/2026-07-07-cyclone-orphan-ack-housekeeping.md`
+  — the implementation plan, written after this spec is signed off.
+
+## 6. Auth boundary
+
+The auth boundary is HTTP (login required, bcrypt + HttpOnly session
+cookie); file-system threats remain the local-only threat model
+(SQLCipher at rest, macOS Keychain). The two new CLI subcommands
+are operator-invoked only and bypass the HTTP auth boundary by design
+(matching all existing `cyclone` CLI subcommands); they read the
+DB directly via `db.SessionLocal()` and require shell access to the
+host running Cyclone. No change to the threat model.