From ad14b56732dbc547e87a5df9ef5a091ad41e142e Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 12:53:40 -0600 Subject: [PATCH] feat(sp38): CycloneStore.find_ack_orphan_st02_summary + reconcile_orphan_st02s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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='' (the sentinel), transaction_set_control_number=, 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 | 148 +++++++++++ backend/src/cyclone/store/claim_acks.py | 172 +++++++----- backend/tests/test_ack_orphan_summary.py | 320 +++++++++++++++++++++++ 3 files changed, 568 insertions(+), 72 deletions(-) create mode 100644 backend/tests/test_ack_orphan_summary.py diff --git a/backend/src/cyclone/store/__init__.py b/backend/src/cyclone/store/__init__.py index 5c6c772..aaa3b05 100644 --- a/backend/src/cyclone/store/__init__.py +++ b/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 = ''`` + * ``totals_json = {"orphan_reconcile": true, + "ack_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 ```` 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="", + 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) diff --git a/backend/src/cyclone/store/claim_acks.py b/backend/src/cyclone/store/claim_acks.py index 2ba9107..6e8716b 100644 --- a/backend/src/cyclone/store/claim_acks.py +++ b/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,86 +277,113 @@ 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 - # any claim_acks link; orphan when NO link was created. - if kind == "999": - # For 999, the "ack has no link" means no ClaimAck row - # was emitted at all (the auto-linker emits one per AK2 - # 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", - ClaimAck.ack_id == ack_row.id) - .count() - ) - if count == 0: - out.append({ - "kind": "999", - "ack_id": ack_row.id, - "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 - ), - }) + # Every ack row of the given kind; orphan when NO claim_acks + # link was created for it. The auto-linker emits one claim_acks + # row per AK2 even when the AK2 is rejected, so a 999 with at + # least one linked AK2 is never an orphan. + for ack_row in s.query(ack_table).order_by(ack_table.id.desc()).all(): + count = ( + s.query(ClaimAck) + .filter(ClaimAck.ack_kind == kind, + ClaimAck.ack_id == ack_row.id) + .count() + ) + if count > 0: + continue + out.append({ + "kind": kind, + "ack_id": ack_row.id, + "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 + ), + }) 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. diff --git a/backend/tests/test_ack_orphan_summary.py b/backend/tests/test_ack_orphan_summary.py new file mode 100644 index 0000000..a443bf3 --- /dev/null +++ b/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 ```` + 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 == "") + .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 == "") + .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 == "") + .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 == "") + .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 \ No newline at end of file