feat(sp39): CycloneStore.record_resubmission + find_resubmission_status

backend/src/cyclone/store/__init__.py

@@ -93,6 +93,11 @@ from .claim_acks import (
 )
 from .backups import add_backup_pending
 from .exceptions import AlreadyMatchedError, InvalidStateError, NotMatchedError
+from .resubmissions import (
+    ResubmissionStatus,
+    find_resubmission_status,
+    record_resubmission,
+)
 from .inbox import list_unmatched, manual_match, manual_unmatch
 from .kpis import dashboard_kpis
 from .orm_builders import (
@@ -323,6 +328,27 @@ class CycloneStore:
         """Return acks with no resolvable link (Inbox ack-orphans lane)."""
         return _find_ack_orphans(kind)
 
+    def record_resubmission(self, *, claim_id, batch_id, source_corrected_path,
+                            interchange_control_number, group_control_number,
+                            resubmitted_at=None):
+        """SP39: insert one Resubmission audit row. Idempotent on
+        (claim_id, interchange_control_number). Returns True if
+        inserted, False if a duplicate was suppressed."""
+        return record_resubmission(
+            claim_id=claim_id,
+            batch_id=batch_id,
+            source_corrected_path=source_corrected_path,
+            interchange_control_number=interchange_control_number,
+            group_control_number=group_control_number,
+            resubmitted_at=resubmitted_at,
+        )
+
+    def find_resubmission_status(self, *, batch_id=None):
+        """SP39: return joined status for every Resubmission row,
+        optionally filtered by batch_id. Statuses are derived at
+        read-time from claim_acks (SP28/31 auto-link) + remittances."""
+        return find_resubmission_status(batch_id=batch_id)
+
     def find_ack_orphan_st02_summary(self):
         """Return per-ST02 summary of orphan 999 acks (sp38).
 

backend/src/cyclone/store/resubmissions.py

@@ -0,0 +1,164 @@
+"""SP39: read/write helpers for the ``resubmissions`` audit table.
+
+The store facade re-exports the public names so callers don't import
+from this module directly.
+
+- ``record_resubmission(...)`` — insert one row per (claim, ICN).
+  Idempotent on the unique constraint; returns ``False`` if a
+  duplicate was suppressed.
+- ``find_resubmission_status(...)`` — read-side join that returns
+  every Resubmission row, optionally filtered by ``batch_id``.
+  Statuses are derived at read-time against ``claim_acks`` (via the
+  existing SP28/31 auto-link) and ``remittances`` (via CLP->claim),
+  so the table stays a write-once audit surface and the existing
+  auto-link data remains the source of truth.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from cyclone import db as cycl_db
+from cyclone.db import Resubmission
+
+
+@dataclass
+class ResubmissionStatus:
+    """One row in the read-side status view (joined + derived)."""
+
+    claim_id: str
+    batch_id: str
+    resubmitted_at: datetime
+    source_corrected_path: str
+    interchange_control_number: str
+    group_control_number: str
+    # Derived statuses (None when no inbound ack/remit exists yet).
+    ack_status: Optional[str] = None       # "999_accepted" / "999_rejected" / "277ca_accepted" / "277ca_rejected"
+    payment_status: Optional[str] = None   # "paid" / "denied_again" / None
+
+
+def record_resubmission(
+    *,
+    claim_id: str,
+    batch_id: str,
+    source_corrected_path: str,
+    interchange_control_number: str,
+    group_control_number: str,
+    resubmitted_at: datetime | None = None,
+) -> bool:
+    """Insert one Resubmission row. Idempotent on
+    ``(claim_id, interchange_control_number)`` — returns ``True`` if
+    inserted, ``False`` if a duplicate-key collision was suppressed.
+    """
+    resubmitted_at = resubmitted_at or datetime.now(timezone.utc)
+    with cycl_db.SessionLocal()() as session:
+        try:
+            session.add(Resubmission(
+                claim_id=claim_id,
+                batch_id=batch_id,
+                resubmitted_at=resubmitted_at,
+                source_corrected_path=source_corrected_path,
+                interchange_control_number=interchange_control_number,
+                group_control_number=group_control_number,
+            ))
+            session.commit()
+            return True
+        except IntegrityError:
+            session.rollback()
+            return False
+
+
+def find_resubmission_status(
+    *, batch_id: str | None = None
+) -> list[ResubmissionStatus]:
+    """Return every Resubmission row joined with the derived ack /
+    payment statuses, optionally filtered by ``batch_id``.
+
+    The join is read-side only: the Resubmission row is the source
+    of truth for "was this claim pushed", and the joined statuses
+    (``claim_acks`` via the SP28/31 auto-link + ``remittances`` via
+    CLP->claim) tell the operator whether the push landed.
+    """
+    with cycl_db.SessionLocal()() as session:
+        stmt = select(Resubmission)
+        if batch_id is not None:
+            stmt = stmt.where(Resubmission.batch_id == batch_id)
+        rows = session.execute(
+            stmt.order_by(Resubmission.batch_id, Resubmission.claim_id)
+        ).scalars().all()
+
+        if not rows:
+            return []
+
+        claim_ids = [r.claim_id for r in rows]
+        # Join against claim_acks for the latest accept/reject code per claim.
+        from cyclone.db import ClaimAck
+        ack_rows = session.execute(
+            select(
+                ClaimAck.claim_id,
+                ClaimAck.ack_kind,
+                ClaimAck.set_accept_reject_code,
+            ).where(ClaimAck.claim_id.in_(claim_ids))
+        ).all()
+        # Reduce to the latest ack per (claim_id, ack_kind); we only
+        # care about the surface-level status (any accept / any reject).
+        ack_status_by_claim: dict[str, str] = {}
+        for cid, ack_kind, code in ack_rows:
+            if not code:
+                continue
+            # Prefer 277ca over 999 (more specific), and accept over reject
+            # when both exist (a 999 accept + 277ca reject is still a reject).
+            existing = ack_status_by_claim.get(cid)
+            if existing is None:
+                ack_status_by_claim[cid] = _ack_status_label(ack_kind, code)
+            else:
+                # Take the "worse" of the two: reject > accept.
+                new_label = _ack_status_label(ack_kind, code)
+                ack_status_by_claim[cid] = _worse_status(existing, new_label)
+
+        # Join against remittances to detect "paid" or "denied_again".
+        from cyclone.db import Remittance
+        paid_claim_ids = set(session.execute(
+            select(Remittance.claim_id).where(Remittance.claim_id.in_(claim_ids))
+        ).scalars().all())
+
+        return [
+            ResubmissionStatus(
+                claim_id=r.claim_id,
+                batch_id=r.batch_id,
+                resubmitted_at=r.resubmitted_at,
+                source_corrected_path=r.source_corrected_path,
+                interchange_control_number=r.interchange_control_number,
+                group_control_number=r.group_control_number,
+                ack_status=ack_status_by_claim.get(r.claim_id),
+                payment_status="paid" if r.claim_id in paid_claim_ids else None,
+            ) for r in rows
+        ]
+
+
+def _ack_status_label(ack_kind: str, code: str) -> str:
+    code = (code or "").upper()
+    if ack_kind == "999":
+        return "999_accepted" if code == "A" else "999_rejected"
+    if ack_kind == "277ca":
+        return "277ca_accepted" if code.startswith("A") else "277ca_rejected"
+    return f"{ack_kind}_{code.lower()}"
+
+
+_STATUS_RANK = {
+    "pending_999": 0,
+    "999_accepted": 1,
+    "277ca_accepted": 2,
+    "999_rejected": 3,
+    "277ca_rejected": 4,
+    "paid": 5,
+}
+
+
+def _worse_status(a: str, b: str) -> str:
+    """Pick the higher-rank (more concerning) status."""
+    return a if _STATUS_RANK.get(a, 0) >= _STATUS_RANK.get(b, 0) else b