feat(sp38): CycloneStore.find_ack_orphan_st02_summary + reconcile_orphan_st02s

Implements the store-helper half of the SP38 spec (tasks 1-3 of
docs/superpowers/plans/2026-07-07-cyclone-orphan-ack-housekeeping.md).

Two new CycloneStore methods + one extraction helper in claim_acks.py:

  - find_ack_orphan_st02_summary() -> list[dict]
    Per-ST02 breakdown of orphan 999 acks. Returns one row per
    distinct (st02, ack_count, has_batch, batch_id). Sorted by
    ack_count DESC so the heaviest backlog surfaces first in CLI
    output. 999-only — 277ca/ta1 orphans have different ST02
    semantics (interchange control number vs source 837 ST02)
    and aggregating them would conflate unrelated identifiers.

  - reconcile_orphan_st02s(*, dry_run=False) -> dict
    One-shot synthetic-batch seeder. For every orphan ST02 without
    a batches row, insert one with kind='837p',
    input_filename='<synthetic:orphan-reconcile>' (the sentinel),
    transaction_set_control_number=<st02>, and totals_json +
    validation_json documenting the orphan-reconcile provenance.
    Idempotent: re-running after a successful pass is a no-op.
    dry_run=True returns the plan shape without writing.

  - _iter_orphan_999_st02s() -> Iterator[(st02, count)]
    Extracted walk helper. The existing find_ack_orphans('999')
    refactored to consume it for the link check; the new summary
    uses it for the per-ST02 aggregation. Accepts both ORM dicts
    and raw sqlite3 strings for raw_json (the SQLAlchemy JSON
    type hands back Python dicts, but tests can hand strings).

11 new tests (test_ack_orphan_summary.py) cover:
  - per-ST02 counts with duplicates
  - has_batch=True when a batches row covers the ST02
  - sort order (heaviest first)
  - exclusion of 999s with a claim_acks link
  - skip of malformed/empty raw_json
  - empty-DB case
  - reconcile creates synthetic rows for missing ST02s only
  - reconcile skips ST02s with existing batches
  - idempotency
  - --dry-run flag
  - ack_count preserved in totals_json

Live-verified against ~/.local/share/cyclone/cyclone.db: 4 distinct
orphan ST02s, 804 total orphan rows. After reconcile, all 4 marked
has_batch=yes.

backend/src/cyclone/store/__init__.py

@@ -87,6 +87,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 +321,153 @@ 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>'``
+          * ``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)
+
+        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.
+        """
+        import json
+        import uuid
+        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": []}
+
+        from datetime import datetime, timezone
+        now = datetime.now(timezone.utc)
+        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,77 +277,29 @@ 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
-            # one AK2 that resolved to a claim is never an orphan).
-            # We treat a 999 as orphan when it has zero ClaimAck
-            # rows tied to its id.
-            all_acks = s.query(Ack).order_by(Ack.id.desc()).all()
-            for ack_row in all_acks:
             count = (
                 s.query(ClaimAck)
-                    .filter(ClaimAck.ack_kind == "999",
+                .filter(ClaimAck.ack_kind == kind,
                         ClaimAck.ack_id == ack_row.id)
                 .count()
             )
-                if count == 0:
+            if count > 0:
+                continue
             out.append({
-                        "kind": "999",
+                "kind": kind,
                 "ack_id": ack_row.id,
-                        "control_number": _ack_control_number(ack_row, "999"),
+                "control_number": _ack_control_number(ack_row, kind),
-                        "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"
@@ -356,6 +309,81 @@ def find_ack_orphans(kind: str) -> list[dict]:
     return out
 
 
+def _iter_orphan_999_st02s() -> Iterator[tuple[str, int]]:
+    """Yield ``(set_control_number, orphan_count)`` for each distinct orphan 999.
+
+    An orphan 999 has zero ``claim_acks`` rows tied to its id (per
+    :func:`find_ack_orphans`). The ST02 is the source 837's
+    ``transaction_set_control_number``, extracted from the 999's
+    ``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).
+    """
+    with db.SessionLocal()() as s:
+        # LEFT OUTER JOIN keeps all acks; orphan when no claim_acks row
+        # 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
+    yield from sorted(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: Ack, kind: str) -> str:
     """Best-effort control-number lookup for a 999 ack row.
 

backend/tests/test_ack_orphan_summary.py

@@ -0,0 +1,320 @@
+"""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