tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp38): CycloneStore.find_ack_orphan_st02_summary + reconcile_orphan_st02s

Browse Source 
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.
This commit is contained in:
Nora
2026-07-07 12:53:40 -06:00
parent 8890627014
commit ad14b56732
3 changed files with 568 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/store/__init__.py
+148
View File
@@ -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
+100 -72
View File
@@ -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.
backend/tests/test_ack_orphan_summary.py
+320
View File
@@ -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