From 9d8f83d111cf20dc33ef11e02b265debe71f74e8 Mon Sep 17 00:00:00 2001 From: Nora Date: Mon, 29 Jun 2026 09:59:53 -0600 Subject: [PATCH] docs(plan): implementation plan for SP27 remittances architecture refactor 17 tasks: 1 preflight + 4 handler extracts (999/TA1/277CA/835) + helpers dedup + INBOUND_RE loosen + SFTP timeouts + status surface + atomic 835/reconcile + match invariants + chain endpoint + claim.rejected_after_remit + frontend chain UI + final verify + merge. Each task ends with live-test + autoreview + commit. Spec: docs/superpowers/specs/2026-06-29-cyclone-remittances-architecture-refactor-design.md --- ...clone-remittances-architecture-refactor.md | 2158 +++++++++++++++++ 1 file changed, 2158 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-29-cyclone-remittances-architecture-refactor.md diff --git a/docs/superpowers/plans/2026-06-29-cyclone-remittances-architecture-refactor.md b/docs/superpowers/plans/2026-06-29-cyclone-remittances-architecture-refactor.md new file mode 100644 index 0000000..f6be235 --- /dev/null +++ b/docs/superpowers/plans/2026-06-29-cyclone-remittances-architecture-refactor.md @@ -0,0 +1,2158 @@ +# Remittances Architecture Refactor (SP27) 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:** Refactor `cyclone.scheduler.py`, the ingest helpers, the +filename classifier, the SFTP layer, the 835 ingest+reconciliation +critical section, and the per-claim UI surface into a coherent Tier +1+2 architecture. End state: scheduler.py slim to ~250 LOC, +handlers split into a subpackage, SFTP operations time-bounded with +visible failures, 835 ingest atomic, `GET /api/claims/{id}/chain` +returns the full claim lifecycle in one call. + +**Architecture:** **Handlers as pure functions.** Each file-type +handler is a small module in `cyclone/handlers/` exposing +`handle(text, source_file) -> HandleResult`. The scheduler keeps a +`HANDLERS: dict[str, Callable]` registry; api.py endpoints delegate to +the same handlers. **SFTP timeout guard.** Every paramiko call goes +through `asyncio.wait_for(asyncio.to_thread(...), timeout=N)` with +`N = CYCLONE_SFTP_OP_TIMEOUT_SECONDS` (default 30). **Atomic 835+reconcile.** +`handle_835` opens one DB session, parses + validates + persists +batch/remits/CasAdjustments + runs `reconcile.match` + writes +`matched_remittance_id` + `claim_id` + final `adjustment_amount` +in the same transaction. **Chain view as join, not stored.** A new +endpoint joins Claim + Ack + Two77caAck + Remittance at request time. + +**Tech Stack:** Python 3.11+, SQLAlchemy 2.x, Pydantic v2, paramiko, +asyncio (FastAPI event loop), pytest (backend), vitest (frontend), +TanStack Query (frontend). Existing test infrastructure. No new +dependencies. + +**Spec:** [`docs/superpowers/specs/2026-06-29-cyclone-remittances-architecture-refactor-design.md`](../specs/2026-06-29-cyclone-remittances-architecture-refactor-design.md) + +--- + +## File structure (after SP27) + +``` +backend/src/cyclone/ +├── scheduler.py ← shrinks from 860 to ~250 LOC +├── clearhouse/__init__.py ← paramiko ops wrapped in asyncio.wait_for +├── edi/filenames.py ← parse_inbound_filename accepts suffix-less names +├── store.py ← manual_match / manual_unmatch paired; startup drift check +├── reconcile.py ← `run` left for CLI follow-up; `match` exported +├── api.py ← /api/parse-* delegating to handlers; /api/claims/{id}/chain +├── api_routers/chain.py NEW +├── handlers/ NEW subpackage +│ ├── __init__.py ← exports HANDLERS dict + HandleResult +│ ├── _ack_id.py ← ack_count_summary, ack_synthetic_source_batch_id, *_277ca_* +│ ├── handle_999.py +│ ├── handle_ta1.py +│ ├── handle_277ca.py ← emits claim.rejected_after_remit +│ └── handle_835.py ← atomic 835+reconcile critical section +└── (existing modules unchanged) + +src/ +├── pages/ClaimDrawer.tsx ← adds "Chain" section +├── hooks/useClaimChain.ts NEW +├── hooks/useClaimChain.test.ts NEW +├── lib/api.ts ← exposes api.fetchClaimChain(id) +└── types/index.ts ← adds ClaimChain type + +backend/tests/ +├── test_handlers_999.py NEW +├── test_handlers_ta1.py NEW +├── test_handlers_277ca.py NEW +├── test_handlers_835.py NEW +├── test_inbound_filename_loose.py NEW +├── test_sftp_op_timeout.py NEW +├── test_scheduler_status_errors.py NEW +├── test_handler_835_atomic_reconcile.py NEW +├── test_api_claim_chain.py NEW +├── test_handler_277ca_rejected_after_remit.py NEW +└── test_store_match_invariant.py NEW +``` + +--- + +## Task 0: Pre-flight — baselines + audit + +**Goal:** Snapshot the test baselines so we can prove zero regressions +during the refactor. Audit helper imports so the move is reversible. + +**Files:** +- Read: `backend/src/cyclone/scheduler.py` (target of split) +- Read: `backend/src/cyclone/api.py` (target of dedup) +- Write: `/tmp/sp27-baseline.txt`, `/tmp/sp27-helper-imports.txt` + +- [ ] **Step 1: Capture pytest baseline** + +```bash +cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 \ + | tee /tmp/sp27-baseline.txt | tail -5 +``` + +Expected: a `XXX passed, YY failed, ZZ skipped` line. Record counts. + +- [ ] **Step 2: Capture npm baseline** + +```bash +npm test 2>&1 | tee /tmp/sp27-frontend-baseline.txt | tail -5 +``` + +Expected: a `Tests N passed` line. Record count. + +- [ ] **Step 3: Audit which modules import the helpers we are about to dedupe** + +```bash +grep -rn "_ack_count_summary\|_ack_synthetic_source_batch_id\|_277ca_synthetic_source_batch_id" \ + backend/src/cyclone/ backend/tests/ \ + | tee /tmp/sp27-helper-imports.txt +``` + +Expected: 2 locations — `backend/src/cyclone/scheduler.py:357` +(`_ack_count_summary` def) + `backend/src/cyclone/api.py` (call sites +via local copy). Tests should not import these directly. + +- [ ] **Step 4: Audit `apply_999_rejections` / `apply_277ca_rejections` importers** + +```bash +grep -rn "apply_999_rejections\|apply_277ca_rejections" \ + backend/src/cyclone/ backend/tests/ \ + | tee /tmp/sp27-rejection-importers.txt +``` + +Expected: scheduler.py + api.py + their tests. No surprises. + +- [ ] **Step 5: Verify handlers/ doesn't already exist** + +```bash +ls backend/src/cyclone/handlers/ 2>&1 | head -3 +``` + +Expected: `No such file or directory`. If it exists, abort and +investigate (the package should be new in this SP). + +- [ ] **Step 6: Snapshot current Scheduler.status() shape** + +```bash +grep -A 30 "class SchedulerStatus" backend/src/cyclone/scheduler.py \ + | tee /tmp/sp27-scheduler-status-pre.txt | head -40 +``` + +Expected: dataclass with `running`, `poll_interval_seconds`, +`sftp_block_name`, `last_poll_at`, `poll_count`, `total_processed`, +`total_skipped`, `total_errored`, `last_tick`. After Tasks 9 + 10 we +will add `consecutive_failures`, `last_error_at`, `last_error`, +`last_sftp_attempt_at`. + +No commit. Pre-flight only. + +--- + +## Task 1: Create handlers/ package skeleton + _ack_id.py + +**Goal:** Make the `cyclone/handlers/` package importable. Stand up +`_ack_id.py` with the three helpers that scheduler.py + api.py both +need. Both existing copies (in scheduler.py and api.py) keep working +during the move; they'll be deleted in Tasks 7 and 8. + +**Files:** +- Create: `backend/src/cyclone/handlers/__init__.py` +- Create: `backend/src/cyclone/handlers/_ack_id.py` +- Write: `backend/tests/test_ack_id_helpers.py` (5 cases) + +- [ ] **Step 1: Write the failing test for _ack_id helpers** + +Create `backend/tests/test_ack_id_helpers.py`: + +```python +"""Tests for the dedup-ed ack ID helpers (Task 1). + +Locks the contract for the helpers that scheduler.py + api.py will +both import from one place. +""" +from __future__ import annotations + +from cyclone.handlers._ack_id import ( + ack_count_summary, + ack_synthetic_source_batch_id, + two77ca_synthetic_source_batch_id, +) +from cyclone.parsers.parse_999 import parse_999_text + + +ACCEPTED_999 = ( + "ISA*00* *00* *ZZ*CYCLONE *ZZ*HPE001 " + "*250129*1200*^*00501*000000001*0*P*:~" + "GS*FA*CYCLONE*HPE001*20260129*1200*1*X*005010X231A1~" + "ST*999*0001*005010X231A1~" + "AK1*HC*1*1*1~" + "AK2*837*1~" + "IK5*A~" + "AK9*A*1*1*1~" + "SE*6*0001~" + "GE*1*1~" + "IEA*0*000000001~" +) + + +def test_ack_count_summary_all_accepted(): + result = parse_999_text(ACCEPTED_999, input_file="x.999") + recv, acc, rej, code = ack_count_summary(result) + assert recv == 1 + assert acc == 1 + assert rej == 0 + assert code == "A" + + +def test_ack_count_summary_rejected_only(): + # Same input but flip IK5 to R + rejected = ACCEPTED_999.replace("IK5*A", "IK5*R") + result = parse_999_text(rejected, input_file="x.999") + recv, acc, rej, code = ack_count_summary(result) + assert code == "R" + assert rej == 1 + + +def test_ack_synthetic_with_pcn_and_filename(): + bsid = ack_synthetic_source_batch_id( + "000000001", pcn="PCN-12345", source_filename="tp_999.x12", + ) + assert bsid == "999-PCN-12345-tp_999.x12".replace(".x12", "").replace( + "-tp_999", "-abcd1234", + ) or bsid.startswith("999-PCN-12345-") + # Same filename hashes to the same suffix + bsid2 = ack_synthetic_source_batch_id( + "000000001", pcn="PCN-12345", source_filename="tp_999.x12", + ) + assert bsid == bsid2 + + +def test_ack_synthetic_no_pcn(): + bsid = ack_synthetic_source_batch_id( + "999999999", pcn=None, source_filename="tp.x12", + ) + assert bsid.startswith("999-999999999-") + assert len(bsid) <= 32 + + +def test_two77ca_synthetic_uses_icn(): + assert two77ca_synthetic_source_batch_id("000012345") == "277CA-000012345" + # Empty ICN falls back to default + assert two77ca_synthetic_source_batch_id("") == "277CA-000000001" +``` + +The exact hash format (8 hex chars) isn't asserted beyond prefix +matching — what matters is determinism. + +- [ ] **Step 2: Verify the test fails** + +```bash +cd backend && .venv/bin/pytest tests/test_ack_id_helpers.py -v 2>&1 | tail -10 +``` + +Expected: ImportError (`No module named 'cyclone.handlers'`). + +- [ ] **Step 3: Create the package skeleton** + +Create `backend/src/cyclone/handlers/__init__.py`: + +```python +"""File-type handlers for inbound MFT files. + +Each handler is a pure function that parses + persists + dispatches +events for one file type. The scheduler and the FastAPI endpoints +both delegate here; the ``HANDLERS`` registry maps ``file_type`` → +handler function. + +Public API: + HandleResult — dataclass returned by every handler + HANDLERS — {"999": handle_999, "835": handle_835, ...} + handle_999, handle_ta1, handle_277ca, handle_835 + — call signatures: handle(text: str, source_file: str) -> HandleResult + +The handlers own their own DB session lifecycle. They emit pubsub +events via the optional ``event_bus`` parameter. They never raise +on per-segment problems; per-segment issues are logged and folded +into the result. Whole-document failures (missing ISA, bad encoding) +surface as CycloneParseError, which the caller catches and records +as STATUS_ERROR. +""" +``` + +Empty for now. Modules added in Tasks 2-5. + +- [ ] **Step 4: Create _ack_id.py** + +Create `backend/src/cyclone/handlers/_ack_id.py`: + +```python +"""Ack ID helpers shared between the scheduler and the FastAPI +endpoints. Moved out of scheduler.py in SP27 to dedupe with api.py. + +Helpers +------- +``ack_count_summary(result)`` + Aggregate ``(received, accepted, rejected, ack_code)`` from a + parsed ``ParseResult999``, trusting set-level IK5 over the + functional-group AK9 (Gainwell's MFT ships contradictory AK9 + segments vs the per-set IK5 — see scheduler commit + ``6507a8c`` for the operational context). + +``ack_synthetic_source_batch_id(icn, *, pcn, source_filename)`` + Build a unique-per-file ``batches.id`` for a 999 that ships + without its own source batch. Falls back to a hash suffix of + the filename so daily pulls don't collide on the same ICN. + +``two77ca_synthetic_source_batch_id(icn)`` + Same for a 277CA without its own source batch. +""" +from __future__ import annotations + +import hashlib +from typing import Any, Tuple + + +def ack_count_summary(result: Any) -> Tuple[int, int, int, str]: + """Aggregate (received, accepted, rejected, ack_code) from ParseResult999. + + Counts are derived from the set-level IK5 responses (one per AK2 + in the 999), not the functional-group AK9. Gainwell's MFT ships + contradictory AK9 segments; IK5 is the authoritative signal. + """ + sets = result.set_responses + received = len(sets) + accepted = sum(1 for s in sets if s.set_accept_reject.code == "A") + rejected = received - accepted + if rejected == 0: + code = "A" + elif accepted == 0: + code = "R" + else: + code = "P" + return (received, accepted, rejected, code) + + +def ack_synthetic_source_batch_id( + interchange_control_number: str, + *, + pcn: str | None = None, + source_filename: str | None = None, +) -> str: + """Synthetic batches.id for a received 999 with no source batch.""" + short_hash = "" + if source_filename: + short_hash = hashlib.sha1(source_filename.encode("utf-8")).hexdigest()[:8] + if pcn and pcn.strip(): + return f"999-{pcn.strip()}-{short_hash}" + icn = (interchange_control_number or "").strip() or "000000001" + if short_hash: + return f"999-{icn}-{short_hash}" + return f"999-{short_hash or icn}" + + +def two77ca_synthetic_source_batch_id(interchange_control_number: str) -> str: + """Synthetic batches.id for a received 277CA with no source batch.""" + return f"277CA-{(interchange_control_number or '').strip() or '000000001'}" +``` + +- [ ] **Step 5: Verify the test passes** + +```bash +cd backend && .venv/bin/pytest tests/test_ack_id_helpers.py -v 2>&1 | tail -10 +``` + +Expected: 5 passed. + +- [ ] **Step 6: Verify full backend suite matches baseline** + +```bash +cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5 +``` + +Expected: identical counts to baseline. + +- [ ] **Step 7: Commit** + +```bash +git add backend/src/cyclone/handlers/ backend/tests/test_ack_id_helpers.py +git commit -m "feat(sp27): create handlers/ package skeleton + dedup ack ID helpers" +``` + +--- + +## Task 2: Extract handle_999 from scheduler.py + +**Goal:** Move the 999 handler out of `scheduler.py:146-195` into +`cyclone/handlers/handle_999.py`. Scheduler.py imports it; tests +exercise it directly. + +**Files:** +- Create: `backend/src/cyclone/handlers/handle_999.py` +- Create: `backend/src/cyclone/handlers/__init__.py` (add `HANDLERS`) +- Create: `backend/tests/test_handlers_999.py` +- Modify: `backend/src/cyclone/scheduler.py:146-195` (delete `_handle_999`) + +- [ ] **Step 1: Write failing test** + +Create `backend/tests/test_handlers_999.py`: + +```python +"""Unit tests for the handle_999 handler (SP27 Task 2). + +Drives the handler directly against a prodfiles fixture so we can +pin the contract independent of the scheduler lifecycle. +""" +from __future__ import annotations + +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from cyclone.api import app +from cyclone.handlers.handle_999 import handle +from cyclone.parsers.parse_999 import parse_999_text + + +# Reuse the prodfiles-style 999 fixture copy that already exists in +# tests/fixtures/. If you don't see one, copy from +# docs/prodfiles/FromHPE/.999.txt first. +ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_999.txt" + + +def test_handle_999_persists_ack_row_and_returns_count(): + text = ACCEPTED.read_text() + result = handle(text, source_file="unit-test.999") + assert result.parser_used == "parse_999" + assert result.claim_count >= 1 + + +def test_handle_999_raises_on_bad_x12(): + from cyclone.parsers.exceptions import CycloneParseError + # Construct a 999 that tokenize will accept but parse_999 will + # reject for missing IK5 segment. + bad = ( + "ISA*00* *00* *ZZ*A *ZZ*B " + "*250129*1200*^*00501*000000001*0*P*:~" + "GS*FA*A*B*20260129*1200*1*X*005010X231A1~" + "ST*999*0001*005010X231A1~" + "AK1*HC*1~" + "SE*3*0001~" + "GE*1*1~" + "IEA*0*000000001~" + ) + result = parse_999_text(bad, input_file="bad.999") # may not raise + with pytest.raises(CycloneParseError): + from cyclone.handlers.handle_999 import handle as handle_fn + # The handler raises CycloneParseError if parse_999_text raised; + # otherwise it returns HandleResult. Either is acceptable — + # the contract is "ParseError is propagatable." +``` + +(Test signatures are illustrative — copy the prodfiles fixture +verbatim, then adjust the bad-999 assertion to whatever the parser +actually throws. The point is that the handler is independently +testable from the scheduler.) + +- [ ] **Step 2: Verify test fails (handler doesn't exist yet)** + +```bash +cd backend && .venv/bin/pytest tests/test_handlers_999.py -v 2>&1 | tail -5 +``` + +Expected: ImportError on `cyclone.handlers.handle_999`. + +- [ ] **Step 3: Create handle_999.py** + +```python +"""Handle a 999 Implementation Acknowledgment file. + +The handler opens its own DB session, dispatches to +``parse_999_text``, applies rejections to any matched claims via +``inbox_state.apply_999_rejections``, persists the ack row, and +returns ``HandleResult``. + +Lifted from ``scheduler.py:_handle_999`` in SP27 Task 2. Behaviour is +unchanged. +""" +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import Optional + +from cyclone import db +from cyclone.audit_log import AuditEvent, append_event +from cyclone.handlers._ack_id import ( + ack_count_summary, ack_synthetic_source_batch_id, +) +from cyclone.inbox_state import apply_999_rejections +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.parse_999 import parse_999_text +from cyclone.store import store as cycl_store + +log = logging.getLogger(__name__) + + +@dataclass +class HandleResult: + parser_used: str + claim_count: int + batch_id: Optional[str] = None + + +def handle(text: str, source_file: str, *, event_bus=None) -> HandleResult: + """Parse a 999, apply rejections, persist ack row.""" + try: + result = parse_999_text(text, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"999 parse error: {exc}") from exc + + received, accepted, rejected, ack_code = ack_count_summary(result) + icn = result.envelope.control_number + pcn = ( + result.set_responses[0].set_control_number + if result.set_responses else None + ) + synthetic_id = ack_synthetic_source_batch_id( + icn, pcn=pcn, source_filename=source_file, + ) + + with db.SessionLocal()() as session: + def _lookup(pcn: str): + return ( + session.query(db.Claim) + .filter_by(patient_control_number=pcn) + .first() + ) + rejection_result = apply_999_rejections( + session, result, claim_lookup=_lookup, + ) + if rejection_result.matched: + for cid in rejection_result.matched: + append_event(session, AuditEvent( + event_type="claim.rejected", + entity_type="claim", + entity_id=cid, + payload={"source_batch_id": synthetic_id}, + actor="999-parser-scheduler", + )) + row = cycl_store.add_ack( + source_batch_id=synthetic_id, + accepted_count=accepted, + rejected_count=rejected, + received_count=received, + ack_code=ack_code, + raw_json=json.loads(result.model_dump_json()), + ) + session.commit() + + # Emit pubsub event if requested (api.py injects the FastAPI + # event_bus; scheduler.py passes None). + if event_bus is not None: + try: + event_bus.publish_sync("ack_received", { + "source_batch_id": synthetic_id, + "ack_code": ack_code, + "kind": "999", + }) + except Exception as exc: # noqa: BLE001 + log.warning("event_bus publish failed: %s", exc) + + return HandleResult(parser_used="parse_999", claim_count=received) +``` + +- [ ] **Step 4: Register the handler in handlers/__init__.py** + +Add to `backend/src/cyclone/handlers/__init__.py`: + +```python +from .handle_999 import handle as handle_999 # noqa: F401 +from .handle_ta1 import handle as handle_ta1 # noqa: F401 (added later) +from .handle_277ca import handle as handle_277ca # noqa: F401 (added later) +from .handle_835 import handle as handle_835 # noqa: F401 (added later) +from ._ack_id import ( + ack_count_summary, + ack_synthetic_source_batch_id, + two77ca_synthetic_source_batch_id, +) + +__all__ = [ + "HandleResult", + "HANDLERS", + "handle_999", "handle_ta1", "handle_277ca", "handle_835", + "ack_count_summary", "ack_synthetic_source_batch_id", + "two77ca_synthetic_source_batch_id", +] + +# Handler registry keyed by file_type. The scheduler and the API +# endpoints dispatch to these. +HANDLERS = { + "999": handle_999, + # "TA1": handle_ta1, # filled by Task 3 + # "277CA": handle_277ca, # filled by Task 4 + # "835": handle_835, # filled by Task 5 (and rewritten in Task 11) +} +``` + +(Imports for other handlers added in their respective tasks.) + +- [ ] **Step 5: Run the test** + +```bash +cd backend && .venv/bin/pytest tests/test_handlers_999.py -v 2>&1 | tail -10 +``` + +Expected: tests pass. + +- [ ] **Step 6: Replace scheduler._handle_999 with import + delete** + +In `backend/src/cyclone/scheduler.py`: +- Remove the `_handle_999` function (lines 146-195). +- Add `from cyclone.handlers import handle_999 as _handle_999` (import alias keeps the dict literal `HANDLERS["999"] = _handle_999` unchanged). +- Verify `HANDLERS["999"] = _handle_999` still references the right function. + +- [ ] **Step 7: Verify full suite matches baseline** + +```bash +cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5 +``` + +Expected: identical counts to `/tmp/sp27-baseline.txt`. + +- [ ] **Step 8: Live-test in docker** + +```bash +docker compose restart backend && sleep 5 +docker logs cyclone-backend-1 --since "30s ago" 2>&1 | grep -iE "scheduler|handler" | tail -10 +curl -s -b cookies.txt http://localhost:8080/api/admin/scheduler/status | jq . +``` + +Expected: scheduler starts cleanly, no import errors, status is healthy. + +- [ ] **Step 9: Autoreview** + +```bash +# Spawn the pr-reviewer subagent against the latest commit +``` + +Invoke the `pr-reviewer` subagent (see `~/.grok/skills/review/SKILL.md`). +Review scope: the new handler module, the test file, the scheduler.py +deletion. + +- [ ] **Step 10: Commit** + +```bash +git add backend/src/cyclone/handlers/handle_999.py \ + backend/src/cyclone/handlers/__init__.py \ + backend/src/cyclone/scheduler.py \ + backend/tests/test_handlers_999.py +git commit -m "feat(sp27): extract handle_999 from scheduler.py into handlers/" +``` + +--- + +## Task 3: Extract handle_ta1 + +**Files:** +- Create: `backend/src/cyclone/handlers/handle_ta1.py` +- Create: `backend/tests/test_handlers_ta1.py` +- Modify: `backend/src/cyclone/handlers/__init__.py` (register `handle_ta1`) +- Modify: `backend/src/cyclone/scheduler.py:303-325` (delete `_handle_ta1`, swap import) + +- [ ] **Step 1: Copy `_handle_ta1` from scheduler.py into a new `handle_ta1.py`** + +Same TDD pattern as Task 2: write a failing test against a TA1 fixture, +then move the implementation, then register in the package. + +Test file: `backend/tests/test_handlers_ta1.py` — at minimum: +1. TA1 happy path: persists interchange ack row. +2. TA1 rejected (ack_code 4/5/6/7): handler does not raise, persists row with ack_code. +3. TA1 missing: raises CycloneParseError. + +- [ ] **Step 2: Implement handle_ta1.py** + +Lift the body of `_handle_ta1` from `scheduler.py:303-325` into +`handle_ta1.py`. Replace the parser import with a module-level +import. The audit-event publication path stays internal; the +`event_bus` injection matches `handle_999`. + +- [ ] **Step 3: Register in `handlers/__init__.py`** + +Uncomment the `"TA1": handle_ta1` line in the `HANDLERS` dict. + +- [ ] **Step 4: Update scheduler.py** + +Replace the inline `_handle_ta1` definition with an import: + +```python +from cyclone.handlers import handle_ta1 as _handle_ta1 +``` + +The `HANDLERS["TA1"] = _handle_ta1` line stays unchanged. + +- [ ] **Step 5: Verify tests + suite + live-test + autoreview + commit** + +Same Steps 5-10 as Task 2 (skip the "failing test" preamble since +this is a straight extraction). Commit message: + +```bash +git commit -m "feat(sp27): extract handle_ta1 from scheduler.py into handlers/" +``` + +--- + +## Task 4: Extract handle_277ca + +**Files:** +- Create: `backend/src/cyclone/handlers/handle_277ca.py` +- Create: `backend/tests/test_handlers_277ca.py` +- Modify: `backend/src/cyclone/handlers/__init__.py` (register) +- Modify: `backend/src/cyclone/scheduler.py:243-298` (delete `_handle_277ca`, swap import) + +- [ ] **Step 1: Test file** + +```python +# test_handlers_277ca.py — 3 cases +# 1. 277CA happy path: persists 2 ack rows with classification +# counts (accepted/rejected/paid/pended). Returns HandleResult. +# 2. 277CA rejects a claim already matched to a remit (deferred +# until Task 13 — see Task 13 for the audit-event assertion). +# 3. 277CA accepts a claim: no audit event, no inbox_state change. +``` + +- [ ] **Step 2: Implement handle_277ca.py + register + scheduler.py update** + +Same shape as Task 3. Add `claim.rejected_after_remit` audit +emission **deferred to Task 13** — for now the handler is a +straight move. + +- [ ] **Step 3: Verify + commit** + +Commit message: + +```bash +git commit -m "feat(sp27): extract handle_277ca from scheduler.py into handlers/" +``` + +--- + +## Task 5: Extract handle_835 (without atomic reconcile yet) + +**Goal:** Move the 835 handler out of scheduler.py. Atomic reconcile +unification happens in Task 11; this task is just the move so the +scheduler shrinks incrementally and we have a working regression +baseline at each step. + +**Files:** +- Create: `backend/src/cyclone/handlers/handle_835.py` +- Create: `backend/tests/test_handlers_835.py` +- Modify: `backend/src/cyclone/handlers/__init__.py` (register) +- Modify: `backend/src/cyclone/scheduler.py:198-241` (delete `_handle_835`, swap import) + +- [ ] **Step 1: Test file** + +```python +# test_handlers_835.py — 4 cases +# 1. 835 happy path: persists batch + Remittance rows; adjustment_amount +# initially 0 (reconcile pass runs separately today — updated in Task 11). +# 2. 835 with CAS adjustments: persists CasAdjustment rows; receives +# the right count. +# 3. 835 validation fails: handler raises ValueError, batch marked +# STATUS_ERROR by the scheduler (test the path that's still in +# scheduler.py this task; the atomic path tests come in Task 11). +# 4. 835 for a payer config we don't have: handler raises CycloneParseError. +``` + +- [ ] **Step 2: Implement handle_835.py** + +Lift the body of `_handle_835` from `scheduler.py:198-241`. Use the +`_ack_id` helpers from Task 1. The `adjustment_amount` overwrite by +a separate reconcile pass remains for now (fix in Task 11). + +- [ ] **Step 3: Register in handlers/__init__.py** + +Uncomment the `"835": handle_835` line. + +- [ ] **Step 4: Update scheduler.py** + +Replace inline `_handle_835` with import: + +```python +from cyclone.handlers import handle_835 as _handle_835 +``` + +- [ ] **Step 5: Verify + live-test + autoreview + commit** + +Commit message: + +```bash +git commit -m "feat(sp27): extract handle_835 from scheduler.py into handlers/" +``` + +--- + +## Task 6: Delete scheduler helpers, swap api.py copies + +**Goal:** Now that all 4 handlers + 3 helpers live in `handlers/`, +delete the inline copies in `scheduler.py` and `api.py`. Both +modules import from the new package. + +**Files:** +- Modify: `backend/src/cyclone/scheduler.py:336-413` + (delete `_ack_count_summary`, `_ack_synthetic_source_batch_id`, `_277ca_synthetic_source_batch_id`) +- Modify: `backend/src/cyclone/api.py` (delete local copies of the same 3 helpers) +- Create: `backend/tests/test_api_dedup.py` (verify api.py still works after deleting local helpers) + +- [ ] **Step 1: Grep api.py for the inline helpers** + +```bash +grep -n "^def _ack_count_summary\|^def _ack_synthetic_source_batch_id\|^def _277ca_synthetic_source_batch_id" backend/src/cyclone/api.py +``` + +Expected: a hit in `api.py`. If empty, the dedup is already done +by someone else; skip to Step 5. + +- [ ] **Step 2: Delete the helpers from scheduler.py** + +Delete `scheduler.py:336-413` (lines 336-413 cover +`_ack_count_summary`, `_ack_synthetic_source_batch_id`, and +`_277ca_synthetic_source_batch_id`). Add at the top of scheduler.py: + +```python +from cyclone.handlers._ack_id import ( + ack_count_summary, ack_synthetic_source_batch_id, + two77ca_synthetic_source_batch_id, +) +# Back-compat aliases for any tests / callers still using the old names. +_ack_count_summary = ack_count_summary +_ack_synthetic_source_batch_id = ack_synthetic_source_batch_id +_277ca_synthetic_source_batch_id = two77ca_synthetic_source_batch_id +``` + +- [ ] **Step 3: Delete the helpers from api.py** + +Same deletion. Same import pattern at the top. + +- [ ] **Step 4: Write test_api_dedup.py** + +```python +"""Regression tests: api.py + scheduler.py delegate to handlers/_ack_id.""" +from __future__ import annotations + +import pytest +from fastapi.testclient import TestClient +from cyclone.api import app + + +@pytest.fixture +def client(): + return TestClient(app) + + +def test_ack_endpoints_still_work(client): + """After dedup, the existing 999 ack endpoints behave the same.""" + # Hit each ack endpoint with a no-file payload; we only care that + # no ImportError is raised (the dedup didn't break import paths). + for path in ("/api/acks", "/api/ta1-acks", "/api/277ca-acks"): + resp = client.get(path) + assert resp.status_code in (200, 401), f"{path} broke: {resp.text}" +``` + +- [ ] **Step 5: Verify full suite** + +```bash +cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5 +``` + +Expected: identical counts. If api.py was using its local helpers in +a way the dedup breaks, the test count will drop — investigate +before continuing. + +- [ ] **Step 6: Live-test + autoreview + commit** + +```bash +docker compose restart backend && sleep 5 +curl -s -b cookies.txt http://localhost:8080/api/acks | jq '.[0] // "empty"' | head -3 +``` + +Expected: returns OK (200), no import errors in logs. + +Commit message: + +```bash +git commit -m "feat(sp27): dedup ack ID helpers — one copy in handlers/_ack_id.py" +``` + +--- + +## Task 7: Loosen INBOUND_RE regex + +**Goal:** `parse_inbound_filename` accepts filenames lacking +`_file_type.x12`. Fall back to `orig_tx` for the file type so the +6/15–6/19 835s (which lack the suffix) get parsed correctly. + +**Files:** +- Modify: `backend/src/cyclone/edi/filenames.py:52-60` (loosen `INBOUND_RE` + add fallback in `parse_inbound_filename`) +- Create: `backend/tests/test_inbound_filename_loose.py` +- Modify: `backend/tests/test_filenames.py` (existing tests still pass) + +- [ ] **Step 1: Write failing test** + +Create `backend/tests/test_inbound_filename_loose.py`: + +```python +"""Loosen parse_inbound_filename to accept filenames without +the _file_type.x12 suffix (e.g. the 6/15-6/19 Gainwell 835 batch).""" +from __future__ import annotations + +import pytest +from cyclone.edi.filenames import parse_inbound_filename + + +def test_filename_with_explicit_835_suffix(): + """Existing happy-path filename is unchanged.""" + f = parse_inbound_filename( + "tp11525703-835_M019110219-20260525001606050-1of1_835.x12" + ) + assert f.file_type == "835" + assert f.orig_tx == "835" + + +def test_filename_without_suffix_with_orig_tx_835(): + """New: 6/15-6/19 Gainwell pattern — no _file_type.x12, orig_tx=835.""" + f = parse_inbound_filename( + "tp11525703-835_M019110219-20260525001606050-1of1.x12" + ) + assert f.file_type == "835" + assert f.orig_tx == "835" + assert f.ext == "x12" + + +def test_filename_without_suffix_with_orig_tx_999_falls_back_to_999(): + """orig_tx is the disambiguator when the suffix is missing.""" + f = parse_inbound_filename( + "tp11525703-999_M000000001-20260601000000000-1of1.x12" + ) + assert f.file_type == "999" + + +def test_filename_without_suffix_277ca(): + f = parse_inbound_filename( + "tp11525703-277CA_M000000001-20260601000000000-1of1.x12" + ) + assert f.file_type == "277CA" + + +def test_filename_unknown_orig_tx_rejected(): + """orig_tx=ENCR is in ALLOWED_FILE_TYPES — should still accept.""" + f = parse_inbound_filename( + "tp11525703-ENCR_M000000001-20260601000000000-1of1.x12" + ) + assert f.file_type == "ENCR" + + +def test_filename_invalid_extension_still_rejected(): + """Inbound .txt files are not X12 — rejected even with valid orig_tx.""" + with pytest.raises(ValueError): + parse_inbound_filename( + "tp11525703-835_M000000001-20260601000000000-1of1.txt" + ) +``` + +- [ ] **Step 2: Verify test fails** + +```bash +cd backend && .venv/bin/pytest tests/test_inbound_filename_loose.py -v 2>&1 | tail -10 +``` + +Expected: 4 of the 6 cases fail (the ones without `_file_type.x12`). +The first case (existing happy path) still passes today. + +- [ ] **Step 3: Implement the loosened parse_inbound_filename** + +Replace `parse_inbound_filename` in `backend/src/cyclone/edi/filenames.py` +(lines 122-156) with: + +```python +def parse_inbound_filename(name: str) -> InboundFilename: + """Parse an inbound HCPF filename. + + Accepts both forms (Gainwell ships both): + * With `_file_type.x12` suffix (the historical / spec form): + tp11525703-837P_M019048402-20260520231513488-1of1_999.x12 + * Without the suffix (the Gainwell production 835 path): + tp11525703-835_M019110219-20260525001606050-1of1.x12 + + In the suffix-less form, `file_type` is derived from `orig_tx` + (the trailing 3-5 chars before the `_M` separator). + + Returns: + InboundFilename with tpid, orig_tx, tracking, ts, file_type, ext. + + Raises: + ValueError: If the filename doesn't match either HCPF inbound + form, or if the derived file_type isn't in ALLOWED_FILE_TYPES. + """ + m = INBOUND_RE.match(name) + if m: + file_type = m.group("file_type") + return InboundFilename( + tpid=m.group("tpid"), + orig_tx=m.group("orig_tx"), + tracking=m.group("tracking"), + ts=m.group("ts"), + file_type=file_type, + ext=m.group("ext"), + ) + + # Fall back to the suffix-less form. + m = INBOUND_RE_LOOSE.match(name) + if not m: + raise ValueError(f"Not a valid HCPF inbound filename: {name!r}") + file_type = m.group("file_type") + if file_type not in ALLOWED_FILE_TYPES: + raise ValueError( + f"file_type {file_type!r} not in allowed HCPF set: {sorted(ALLOWED_FILE_TYPES)}" + ) + return InboundFilename( + tpid=m.group("tpid"), + orig_tx=m.group("file_type"), # preserve the historical shape + tracking=m.group("tracking"), + ts=m.group("ts"), + file_type=file_type, + ext=m.group("ext"), + ) +``` + +Add a second regex just above `parse_inbound_filename`: + +```python +# Inbound suffix-less form: tp11525703-835_M019110219-20260525001606050-1of1.x12 +# - prefix: literal "TP" or "tp" (case-insensitive on the first 2 chars) +# - tpid: 1+ digits +# - file_type (used as both orig_tx and file_type): the alphanumeric +# token between "-" and "_M". Gainwell's 6/15-6/19 batch +# uses 3-5 character codes (835 / 999 / 277CA / ENCR). +# - track: M + 1+ uppercase alnum +# - ts: 17 digits +# - seq: literal "1of1" +# - ext: literal "x12" +INBOUND_RE_LOOSE = re.compile( + r"^(?i:TP)(?P\d+)-(?P[A-Z0-9]+)_(?PM[A-Z0-9]+)" + r"-(?P\d{17})-1of1\.(?Px12)$" +) +``` + +- [ ] **Step 4: Verify new tests pass + existing test_filenames.py still passes** + +```bash +cd backend && .venv/bin/pytest tests/test_inbound_filename_loose.py tests/test_filenames.py -v 2>&1 | tail -15 +``` + +Expected: all pass. + +- [ ] **Step 5: Run full suite** + +```bash +cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5 +``` + +Expected: identical counts. + +- [ ] **Step 6: Live-test by re-triggering scheduler against an inbound file lacking the suffix** + +```bash +docker compose restart backend && sleep 5 +# Drop one of the previously-skipped 835s into the staging dir so the +# scheduler picks it up on the next tick. +SUFFIX_LESS="/tmp/cyclone-stage/FromHPE/tp11525703-835_M019110219-20260525001606050-1of1.x12" +ls -la "$SUFFIX_LESS" +# Trigger an admin tick +curl -s -X POST -b cookies.txt http://localhost:8080/api/admin/scheduler/tick | jq . +# Look at the processed-files log +curl -s -b cookies.txt "http://localhost:8080/api/admin/scheduler/processed-files?limit=3" | jq . +``` + +Expected: the suffix-less file is recorded with +`parser_used: "parse_835"`, `status: "ok"`. Before this SP it would +have been `status: "skipped"`. + +- [ ] **Step 7: Autoreview + commit** + +Commit message: + +```bash +git commit -m "feat(sp27): loosen INBOUND_RE to accept suffix-less inbound filenames" +``` + +--- + +## Task 8: Add SFTP operation timeouts + +**Goal:** Every paramiko call inside `clearhouse.SftpClient` runs +under `asyncio.wait_for(asyncio.to_thread(...), timeout=N)` so a +hung `sftp.listdir_attr()` can no longer freeze the scheduler. + +**Files:** +- Modify: `backend/src/cyclone/clearhouse/__init__.py` (paramiko helpers) +- Create: `backend/tests/test_sftp_op_timeout.py` + +- [ ] **Step 1: Write failing test** + +Create `backend/tests/test_sftp_op_timeout.py`: + +```python +"""Timeout guard for SFTP operations (SP27 Task 8). + +paramiko is synchronous; without a timeout, a hung listdir_attr +freezes the worker thread indefinitely. We wrap the call sites in +``asyncio.wait_for(asyncio.to_thread(...), timeout=N)`` so the +event loop can give up after the configured bound. +""" +from __future__ import annotations + +import asyncio +import time +from unittest.mock import MagicMock, patch + +import pytest + +from cyclone.clearhouse import SftpClient +from cyclone.providers import SftpBlock + + +def _block(): + return SftpBlock( + host="mft.example.com", port=22, username="user", + auth={"password_keychain_account": "x"}, + paths={"inbound": "/inbound", "outbound": "/outbound"}, + stub=False, + ) + + +@pytest.mark.asyncio +async def test_list_inbound_times_out_when_paramiko_hangs(monkeypatch): + """A stubbed listdir_attr that sleeps 60s should be cancelled by + the asyncio.wait_for around it once CYCLONE_SFTP_OP_TIMEOUT_SECONDS elapses.""" + monkeypatch.setenv("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "1") + client = _block().__class__(...) # placeholder + + # The simplest implementation is to add an "async_list_inbound" + # method on SftpClient that does the wrap, and assert that + # method raises asyncio.TimeoutError after the configured N seconds. + start = time.monotonic() + with pytest.raises(asyncio.TimeoutError): + # Implement async_list_inbound in clearhouse/__init__.py + await client.async_list_inbound() + elapsed = time.monotonic() - start + assert elapsed < 3, f"timeout fired too late: {elapsed}s" +``` + +(The exact shape of the new async surface is the implementer's +decision — see Step 3 for the recommended shape.) + +- [ ] **Step 2: Verify test fails** + +```bash +cd backend && .venv/bin/pytest tests/test_sftp_op_timeout.py -v 2>&1 | tail -10 +``` + +Expected: ImportError or AttributeError (no `async_list_inbound` yet). + +- [ ] **Step 3: Add async surface to SftpClient** + +Add to `backend/src/cyclone/clearhouse/__init__.py`: + +```python +import asyncio +import os + + +def _op_timeout_seconds() -> float: + """Per-op SFTP timeout from CYCLONE_SFTP_OP_TIMEOUT_SECONDS (default 30).""" + return float(os.environ.get("CYCLONE_SFTP_OP_TIMEOUT_SECONDS", "30")) + + +class SftpClient: + # ...existing methods... + + # ---- Async surface (SP27 Task 8) -------------------------------- + + async def async_list_inbound(self) -> list["InboundFile"]: + """Async-wrapped list_inbound with a per-op asyncio timeout.""" + return await asyncio.wait_for( + asyncio.to_thread(self.list_inbound), + timeout=_op_timeout_seconds(), + ) + + async def async_list_inbound_names(self) -> list["InboundFile"]: + return await asyncio.wait_for( + asyncio.to_thread(self.list_inbound_names), + timeout=_op_timeout_seconds(), + ) + + async def async_download_inbound(self, f: "InboundFile") -> Path: + return await asyncio.wait_for( + asyncio.to_thread(self.download_inbound, f), + timeout=_op_timeout_seconds(), + ) + + async def async_read_file(self, remote_path: str) -> bytes: + return await asyncio.wait_for( + asyncio.to_thread(self.read_file, remote_path), + timeout=_op_timeout_seconds(), + ) + + async def async_write_file(self, remote_path: str, content: bytes) -> Path: + return await asyncio.wait_for( + asyncio.to_thread(self.write_file, remote_path, content), + timeout=_op_timeout_seconds(), + ) +``` + +(Inside `Scheduler._tick_impl` and `Scheduler.process_inbound_files`, +swap the sync `self._list_inbound()` for `await client.async_list_inbound()` +once the wire-up is done. For this task, just add the async surface +and lock it in with the test.) + +- [ ] **Step 4: Wire scheduler.py to the new async surface** + +Replace: + +```python +files = await asyncio.to_thread(self._list_inbound) +``` + +in `scheduler.py:_tick_impl` with: + +```python +try: + files = await self._sftp_client_factory( + self._sftp_block + ).async_list_inbound() +except asyncio.TimeoutError as exc: + log.exception("SFTP list_inbound timed out") + result.errors.append(f"list_inbound: timeout") + result.finished_at = datetime.now(timezone.utc) + return result +``` + +- [ ] **Step 5: Run new test** + +```bash +cd backend && .venv/bin/pytest tests/test_sftp_op_timeout.py -v 2>&1 | tail -10 +``` + +Expected: pass. + +- [ ] **Step 6: Verify full suite** + +```bash +cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5 +``` + +Expected: identical counts. + +- [ ] **Step 7: Live-test + autoreview + commit** + +Commit message: + +```bash +git commit -m "feat(sp27): wrap SFTP operations in asyncio.wait_for with CYCLONE_SFTP_OP_TIMEOUT_SECONDS" +``` + +--- + +## Task 9: Surface SFTP errors in Scheduler.status() + +**Goal:** `Scheduler.status()` exposes `consecutive_failures`, +`last_error_at`, `last_error`, `last_sftp_attempt_at`. After 3 +consecutive failures, the operator UI can show a destructive pill. + +**Files:** +- Modify: `backend/src/cyclone/scheduler.py:115-138` (`SchedulerStatus` + `status()`) +- Create: `backend/tests/test_scheduler_status_errors.py` + +- [ ] **Step 1: Write failing test** + +Create `backend/tests/test_scheduler_status_errors.py`: + +```python +"""SP27 Task 9: scheduler status surfaces SFTP failures.""" +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone + +import pytest + +from cyclone import scheduler as sched_module +from cyclone.providers import SftpBlock + + +def _block(): + return SftpBlock( + host="mft.example.com", port=22, username="user", + auth={"password_keychain_account": "x"}, + paths={"inbound": "/inbound", "outbound": "/outbound"}, + stub=False, + ) + + +@pytest.mark.asyncio +async def test_status_records_last_error_after_failure(monkeypatch): + sched = sched_module.Scheduler(_block(), sftp_client_factory=lambda b: BrokenClient()) + status = sched.status() + # Pre-state: no errors + assert status.consecutive_failures == 0 + assert status.last_error is None + + # Force a tick that fails on the SFTP call. + monkeypatch.setattr(sched, "_list_inbound", lambda: (_ for _ in ()).throw(RuntimeError("boom"))) + await sched.tick() + status = sched.status() + assert status.consecutive_failures == 1 + assert status.last_error is not None + assert "boom" in status.last_error + + +class BrokenClient: + def list_inbound(self): raise RuntimeError("simulated outage") +``` + +- [ ] **Step 2: Verify test fails** + +```bash +cd backend && .venv/bin/pytest tests/test_scheduler_status_errors.py -v 2>&1 | tail -10 +``` + +Expected: AttributeError (no `consecutive_failures` field on status yet). + +- [ ] **Step 3: Extend SchedulerStatus** + +Replace `backend/src/cyclone/scheduler.py:115-138` with: + +```python +@dataclass +class SchedulerStatus: + """Snapshot of the scheduler's runtime state.""" + + running: bool + poll_interval_seconds: int + sftp_block_name: str + last_poll_at: Optional[datetime] + poll_count: int + total_processed: int + total_skipped: int + total_errored: int + last_tick: Optional[TickResult] = None + # SP27 Task 9 additions + consecutive_failures: int = 0 + last_error_at: Optional[datetime] = None + last_error: Optional[str] = None + last_sftp_attempt_at: Optional[datetime] = None + + def as_dict(self) -> dict[str, Any]: + return { + "running": self.running, + "poll_interval_seconds": self.poll_interval_seconds, + "sftp_block_name": self.sftp_block_name, + "last_poll_at": ( + self.last_poll_at.isoformat() if self.last_poll_at else None + ), + "poll_count": self.poll_count, + "total_processed": self.total_processed, + "total_skipped": self.total_skipped, + "total_errored": self.total_errored, + "last_tick": self.last_tick.as_dict() if self.last_tick else None, + "consecutive_failures": self.consecutive_failures, + "last_error_at": ( + self.last_error_at.isoformat() if self.last_error_at else None + ), + "last_error": self.last_error, + "last_sftp_attempt_at": ( + self.last_sftp_attempt_at.isoformat() if self.last_sftp_attempt_at else None + ), + } +``` + +Extend the `Scheduler.__init__` with: + +```python +self._consecutive_failures = 0 +self._last_error_at: Optional[datetime] = None +self._last_error: Optional[str] = None +self._last_sftp_attempt_at: Optional[datetime] = None +``` + +Extend `Scheduler.tick` to bump `_consecutive_failures` on +`TickResult.errors` and clear on a successful tick. Extend +`Scheduler.status()` to expose the new fields. + +Add `_record_sftp_outcome(success, error?)` helper to centralize the +bump/clear logic. + +- [ ] **Step 4: Run test** + +```bash +cd backend && .venv/bin/pytest tests/test_scheduler_status_errors.py -v 2>&1 | tail -10 +``` + +Expected: pass. + +- [ ] **Step 5: Verify suite + live-test + commit** + +```bash +git commit -m "feat(sp27): surface consecutive_failures + last_error in Scheduler.status()" +``` + +--- + +## Task 10: Unify 835 ingest + reconciliation in handle_835 + +**Goal:** `handlers.handle_835` runs parse + validate + persist +batch/remits/CasAdjustments + `reconcile.match` + writes +`matched_remittance_id` + `claim_id` + final `adjustment_amount` +in one critical section. The placeholder `adjustment_amount=0` is +gone. + +**Files:** +- Modify: `backend/src/cyclone/handlers/handle_835.py` +- Create: `backend/tests/test_handler_835_atomic_reconcile.py` +- Modify: `backend/src/cyclone/reconcile.py` (export the pure functions that handlers call) + +- [ ] **Step 1: Write failing test** + +Create `backend/tests/test_handler_835_atomic_reconcile.py`: + +```python +"""SP27 Task 11/10: 835 ingest + reconciliation in one critical section.""" +from __future__ import annotations + +from decimal import Decimal +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from cyclone import db +from cyclone.api import app +from cyclone.handlers.handle_835 import handle +from cyclone.parsers.parse_835 import parse as parse_835 +from cyclone.payers import PAYER_FACTORIES_835 + + +# Use an existing fixture or copy from docs/prodfiles/835fromco/ +EIGHT_THREE_FIVE = ( + Path(__file__).parent / "fixtures" / "minimal_835.txt" +) + + +def test_handle_835_persists_remittance_with_correct_adjustment(): + config = PAYER_FACTORIES_835["co_medicaid_835"]() + text = EIGHT_THREE_FIVE.read_text() + + # Parse standalone to know the expected claim count. + parsed = parse_835(text, config, input_file=EIGHT_THREE_FIVE.name) + n = len(parsed.claims) + + result = handle(text, source_file=EIGHT_THREE_FIVE.name) + assert result.parser_used == "parse_835" + assert result.claim_count == n + + # At least one Remittance should have a non-zero adjustment_amount + # set in the same transaction. + with db.SessionLocal()() as s: + remits = s.query(db.Remittance).filter_by(batch_id=result.batch_id).all() + non_zero = [r for r in remits if r.adjustment_amount > 0] + # Even an 835 with no CAS rows should have adjustment_amount=0, + # so we don't assert non_zero > 0 unconditionally — + # just check the field is correctly populated from CAS rows. + for r in remits: + # Cross-check: sum of CasAdjustment.amount equals Remittance.adjustment_amount + cas_total = sum( + ca.amount for ca in r.cas_adjustments # via relationship + ) if hasattr(r, "cas_adjustments") else Decimal("0") + assert r.adjustment_amount == cas_total, ( + f"remit {r.id} adjustment_amount={r.adjustment_amount} " + f"!= cas_sum={cas_total}" + ) + + +def test_handle_835_matched_remit_pair_synced_in_one_session(): + """If an 835 remit's claim matches a stored claim, both sides of + the matched-pair pointer are set in the same session.""" + # See test_plan in spec — set up a stored Claim, then call handle(), + # then assert Claim.matched_remittance_id == Remittance.claim_id. + pytest.skip("Wired in Task 12 alongside the chain endpoint") +``` + +- [ ] **Step 2: Implement the atomic handle_835** + +Replace `handlers/handle_835.py` with: + +```python +"""Handle an 835 ERA file atomically — parse, validate, persist +batch/remits/CasAdjustments, run reconcile, write matched-pair +pointers, all in one DB session (SP27 Task 10). +""" +from __future__ import annotations + +import json +import logging +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Optional + +from cyclone import db, reconcile +from cyclone.handlers._ack_id import ack_synthetic_source_batch_id +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.parse_835 import parse as parse_835 +from cyclone.parsers.validator_835 import validate as validate_835 +from cyclone.payers import PAYER_FACTORIES_835 +from cyclone.store import store as cycl_store, BatchRecord + +log = logging.getLogger(__name__) + + +@dataclass +class HandleResult: + parser_used: str + claim_count: int + batch_id: Optional[str] = None + matched_count: int = 0 + + +def handle(text: str, source_file: str, *, event_bus=None) -> HandleResult: + """Parse, validate, persist, and reconcile an 835 in one session.""" + config = PAYER_FACTORIES_835["co_medicaid_835"]() + try: + result = parse_835(text, config, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"835 parse error: {exc}") from exc + + # Validation report (mirrors api.py). + report = validate_835(result, config) + n = len(result.claims) + if report.passed: + passed, failed, failed_claim_ids = n, 0, [] + else: + passed, failed, failed_claim_ids = 0, n, [ + c.payer_claim_control_number for c in result.claims + ] + result = result.model_copy(update={ + "validation": report, + "summary": result.summary.model_copy(update={ + "passed": passed, + "failed": failed, + "failed_claim_ids": failed_claim_ids, + }), + }) + + rec = BatchRecord( + id=uuid.uuid4().hex, + kind="835", + input_filename=source_file, + parsed_at=datetime.now(timezone.utc), + result=result, + ) + cycl_store.add(rec, event_bus=event_bus) + matched = reconcile.run_now(rec.id) + return HandleResult( + parser_used="parse_835", + claim_count=n, + batch_id=rec.id, + matched_count=matched, + ) +``` + +(Adjust if `cycl_store.add(rec, event_bus=event_bus)` already passes the bus; +`reconcile.run_now` is a new helper that takes the in-memory batch +record and runs the same aggregation the standalone `reconcile.run` +does, but against the same session. Add it to `reconcile.py` as a +small wrapper.) + +- [ ] **Step 3: Add `reconcile.run_now(batch_id)` to `reconcile.py`** + +The existing `reconcile.run(session, batch_id)` opens its own session. +Add a thin helper that calls it against the scheduler's / api's +session, or restructure so `reconcile.run` accepts an optional +session and uses the caller's. + +- [ ] **Step 4: Verify tests + suite + live-test + commit** + +```bash +git commit -m "feat(sp27): unify 835 ingest + reconciliation in handle_835 (atomic)" +``` + +--- + +## Task 11: Paired write of matched_remittance_id ↔ claim_id in store + +**Goal:** `store.manual_match` and `store.manual_unmatch` write both +`Claim.matched_remittance_id` and `Remittance.claim_id` in one +transaction. Startup drift check logs any pre-existing mismatch +(non-blocking). + +**Files:** +- Modify: `backend/src/cyclone/store.py:1951-2163` (manual_match / manual_unmatch) +- Create: `backend/tests/test_store_match_invariant.py` + +- [ ] **Step 1: Read the current manual_match / manual_unmatch implementations** + +Already exists; shape is "update Claim + insert Match row, both in +one session". Extend it to also write `Remittance.claim_id`. + +- [ ] **Step 2: Write failing test** + +Create `backend/tests/test_store_match_invariant.py`: + +```python +"""SP27 Task 11: matched-pair invariant.""" +from __future__ import annotations + +from cyclone import db +from cyclone.store import manual_match, manual_unmatch, AlreadyMatchedError + + +def test_manual_match_writes_both_sides_in_one_transaction(): + # Set up: create a Claim and an unmatched Remittance in the same + # DB session (use the conftest autouse DB fixture). + cid = "claim-x" + rid = "remit-y" + # ... insert via SQLAlchemy (use cycl_store.add or s.add directly) + # Call manual_match, then assert both sides are set. + assert claim.matched_remittance_id == rid + assert remit.claim_id == cid + + +def test_manual_unmatch_clears_both_sides(): + # Match first, then unmatch; assert both sides return to None. + manual_match(cid, rid) + manual_unmatch(cid) + assert claim.matched_remittance_id is None + assert remit.claim_id is None + + +def test_startup_logs_drift_without_raising(capsys): + # Pre-existing drift: insert mismatched claim/remit directly, + # then re-import cyclone (which runs bootstrap). + # Assert the warning is logged. +``` + +- [ ] **Step 3: Implement the paired write + startup check** + +Add at the bottom of `store.py`: + +```python +def check_match_invariant_at_startup() -> None: + """Log mismatches between Claim.matched_remittance_id and + Remittance.claim_id (SP27 Task 11). + + Non-blocking — drift is informational. The follow-up SP will add + ``cyclone reconcile reindex-matches`` to repair historical drift. + """ + with db.SessionLocal()() as s: + claim_ids = { + row.claim_id for row in s.query(db.Claim) + .filter(db.Claim.matched_remittance_id.isnot(None)) + .all() + } + remit_claim_ids = { + row.id for row in s.query(db.Remittance) + .filter(db.Remittance.claim_id.isnot(None)) + .all() + } + drift = (claim_ids ^ remit_claim_ids) + if drift: + log.warning( + "match invariant drift: %d mismatched claim_ids", len(drift), + extra={"drift_ids": list(drift)[:20]}, + ) +``` + +Wire `check_match_invariant_at_startup()` into `bootstrap.run()`. + +- [ ] **Step 4: Verify tests + live-test + autoreview + commit** + +```bash +git commit -m "feat(sp27): paired write of Claim.matched_remittance_id and Remittance.claim_id" +``` + +--- + +## Task 12: Add GET /api/claims/{id}/chain endpoint + +**Goal:** One endpoint returns a claim's 837 + 999 + 277CA + 835 +(or null for each missing piece, plus a `missing` array). + +**Files:** +- Create: `backend/src/cyclone/api_routers/chain.py` +- Modify: `backend/src/cyclone/api.py` (register the router) +- Create: `backend/tests/test_api_claim_chain.py` + +- [ ] **Step 1: Write the failing test** + +```python +# test_api_claim_chain.py — 5 cases +# 1. Happy path: claim + 999 ack + 277CA ack + 835 remit, all populated. +# 2. Missing 999: claim + 277CA + 835, ack_999=null, missing=["ack_999"]. +# 3. Missing remit: claim + 999, remittance=null, missing=["remittance"]. +# 4. Missing claim: 404. +# 5. No auth: 401. +``` + +- [ ] **Step 2: Implement the chain router** + +Create `backend/src/cyclone/api_routers/chain.py`: + +```python +"""GET /api/claims/{id}/chain — the full lifecycle of one claim.""" +from __future__ import annotations + +from typing import Annotated, Any, Optional + +from fastapi import APIRouter, Depends, HTTPException + +from cyclone import db +from cyclone.auth.deps import matrix_gate + + +router = APIRouter(tags=["claims"], dependencies=[Depends(matrix_gate)]) + + +SLOTS = ("submission", "ack_999", "ack_277ca", "remittance") + + +@router.get("/api/claims/{claim_id}/chain") +def get_claim_chain(claim_id: str) -> dict[str, Any]: + """Return the claim's full chain. Each slot may be null.""" + payload: dict[str, Any] = {"claim_id": claim_id} + missing: list[str] = [] + + with db.SessionLocal()() as session: + claim = session.query(db.Claim).filter_by(id=claim_id).first() + if claim is None: + raise HTTPException(status_code=404, detail="claim not found") + payload["submission"] = { + "batch_id": claim.batch_id, + "patient_control_number": claim.patient_control_number, + "service_date_from": ( + claim.service_date_from.isoformat() + if claim.service_date_from else None + ), + "service_date_to": ( + claim.service_date_to.isoformat() + if claim.service_date_to else None + ), + "charge_amount": str(claim.charge_amount), + "state": claim.state, + "submitted_at": claim.submitted_at.isoformat() if claim.submitted_at else None, + } + + # 999 ack: match by patient_control_number. + ack_999 = ( + session.query(db.Ack) + .filter_by(patient_control_number=claim.patient_control_number) + .order_by(db.Ack.received_at.desc()) + .first() + ) + if ack_999 is None: + payload["ack_999"] = None + missing.append("ack_999") + else: + payload["ack_999"] = { + "source_batch_id": ack_999.source_batch_id, + "ack_code": ack_999.ack_code, + "received_count": ack_999.received_count, + "accepted_count": ack_999.accepted_count, + "rejected_count": ack_999.rejected_count, + "received_at": ( + ack_999.received_at.isoformat() if ack_999.received_at else None + ), + } + + # 277CA ack: similar match. + ack_277ca = ( + session.query(db.Two77caAck) + .filter_by(claim_id=claim_id) + .order_by(db.Two77caAck.received_at.desc()) + .first() + ) + if ack_277ca is None: + payload["ack_277ca"] = None + missing.append("ack_277ca") + else: + payload["ack_277ca"] = { + "source_batch_id": ack_277ca.source_batch_id, + "classification": ack_277ca.classification, + "received_at": ( + ack_277ca.received_at.isoformat() + if ack_277ca.received_at else None + ), + } + + # Remittance. + remit = ( + session.query(db.Remittance) + .filter_by(claim_id=claim_id) + .first() + ) + if remit is None: + payload["remittance"] = None + missing.append("remittance") + else: + payload["remittance"] = { + "id": remit.id, + "payer_claim_control_number": remit.payer_claim_control_number, + "total_paid": str(remit.total_paid), + "patient_responsibility": str(remit.patient_responsibility), + "adjustment_amount": str(remit.adjustment_amount), + "status_label": remit.status_label, + "received_at": ( + remit.received_at.isoformat() if remit.received_at else None + ), + "matched": remit.claim_id == claim_id, + "adjustments": [ + { + "group_code": ca.group_code, + "reason_code": ca.reason_code, + "amount": str(ca.amount), + "label": ca.reason_label, + } + for ca in remit.cas_adjustments + ], + } + + payload["missing"] = missing + return payload +``` + +Add to `backend/src/cyclone/api.py`: + +```python +from cyclone.api_routers import chain +app.include_router(chain.router) +``` + +Adjust the column names if `db.Remittance.cas_adjustments` is a +different relationship attribute name (it's a backref from +`CasAdjustment.remittance`). + +- [ ] **Step 3: Verify tests + suite + live-test + commit** + +```bash +git commit -m "feat(sp27): add GET /api/claims/{id}/chain endpoint" +``` + +--- + +## Task 13: Add claim.rejected_after_remit audit emission in handle_277ca + +**Files:** +- Modify: `backend/src/cyclone/handlers/handle_277ca.py` +- Create: `backend/tests/test_handler_277ca_rejected_after_remit.py` + +- [ ] **Step 1: Write the failing test** + +```python +# 2 cases +# 1. 277CA rejects a claim that is already matched to a remit → +# emits `claim.rejected_after_remit` ActivityEvent with both IDs. +# 2. 277CA rejects an unmatched claim → no such audit event. +``` + +- [ ] **Step 2: Implement the audit emission** + +Inside `handle_277ca.handle`, after `apply_277ca_rejections(...).matched` +loops, check each matched claim: +- If the claim has `matched_remittance_id` set, emit the audit event. + +```python +for cid in apply_result.matched: + claim = ( + session.query(db.Claim) + .filter_by(id=cid).first() + ) + if claim and claim.matched_remittance_id: + append_event(session, AuditEvent( + event_type="claim.rejected_after_remit", + entity_type="claim", + entity_id=cid, + payload={ + "source_batch_id": synthetic_id, + "matched_remittance_id": claim.matched_remittance_id, + }, + actor="277ca-parser", + )) +``` + +- [ ] **Step 3: Verify + live-test + commit** + +```bash +git commit -m "feat(sp27): emit claim.rejected_after_remit from handle_277ca on matched claims" +``` + +--- + +## Task 14: Frontend useClaimChain + ClaimDrawer chain section + +**Files:** +- Create: `src/hooks/useClaimChain.ts` +- Create: `src/hooks/useClaimChain.test.ts` +- Modify: `src/lib/api.ts` (expose `api.fetchClaimChain(id)`) +- Modify: `src/types/index.ts` (add `ClaimChain` type) +- Modify: `src/pages/ClaimDrawer.tsx` (add Chain section) +- Create: `src/pages/ClaimDrawer.test.tsx` (extend with chain-rendering assertions) + +- [ ] **Step 1: Write `useClaimChain.test.ts` first** + +```ts +// @vitest-environment happy-dom +(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true; + +import { afterEach, describe, expect, it, vi } from "vitest"; +import { cleanup, renderHook, waitFor } from "@testing-library/react"; +import { useClaimChain } from "./useClaimChain"; + +vi.mock("@/lib/api", () => ({ + api: { fetchClaimChain: vi.fn() }, +})); + +afterEach(() => { cleanup(); vi.unstubAllGlobals(); }); + +describe("useClaimChain", () => { + it("loads chain data on mount", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ + ok: true, + json: async () => ({ + claim_id: "x", + submission: { batch_id: "b1", charge_amount: "100.00", state: "submitted" }, + ack_999: null, + ack_277ca: null, + remittance: null, + missing: ["ack_999", "ack_277ca", "remittance"], + }), + })); + const { result } = renderHook(() => useClaimChain("x")); + await waitFor(() => expect(result.current.loading).toBe(false)); + expect(result.current.chain?.submission.charge_amount).toBe("100.00"); + }); + + it("surfaces a 404 as an error", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ + ok: false, status: 404, statusText: "Not Found", json: async () => ({}), + })); + const { result } = renderHook(() => useClaimChain("missing")); + await waitFor(() => expect(result.current.error).toBeDefined()); + }); +}); +``` + +- [ ] **Step 2: Implement useClaimChain.ts** + +Standard TanStack Query wrapper, follows the pattern of existing +hooks in `src/hooks/`. + +- [ ] **Step 3: Add to src/lib/api.ts** + +```ts +export const api = { + // ...existing methods... + fetchClaimChain: (id: string) => + fetch(joinUrl(`/api/claims/${encodeURIComponent(id)}/chain`), { + credentials: "include", + }).then(r => { if (!r.ok) throw new Error(`chain ${r.status}`); return r.json(); }), +}; +``` + +- [ ] **Step 4: Add ClaimChain type to src/types/index.ts** + +```ts +export interface ClaimChain { + claim_id: string; + submission: { batch_id: string; charge_amount: string; /* ... */ } | null; + ack_999: { source_batch_id: string; ack_code: string; /* ... */ } | null; + ack_277ca: { /* ... */ } | null; + remittance: { /* ... */ } | null; + missing: ("ack_999" | "ack_277ca" | "remittance")[]; +} +``` + +- [ ] **Step 5: Extend ClaimDrawer.test.tsx** + +Add 2 test cases verifying the chain section renders placeholders +for missing slots and full content when present. + +- [ ] **Step 6: Add Chain section to ClaimDrawer.tsx** + +```tsx +import { useClaimChain } from "@/hooks/useClaimChain"; + +// In the drawer body: +const { chain, loading } = useClaimChain(claim.id); + +
+

Chain

+ {loading ? : ( + <> +
Submission: {chain?.submission?.batch_id ?? "—"}
+
999 Ack: {chain?.ack_999?.ack_code ?? "(pending)"}
+
277CA: {chain?.ack_277ca?.classification ?? "(pending)"}
+
Remittance: {chain?.remittance ? `${chain.remittance.total_paid}` : "(pending)"}
+ {chain && chain.missing.length > 0 && ( + Awaiting: {chain.missing.join(", ")} + )} + + )} +
+``` + +- [ ] **Step 7: Verify frontend tests + suite + commit** + +```bash +npm test +npm run typecheck +git commit -m "feat(sp27): add useClaimChain hook and ClaimDrawer chain section" +``` + +--- + +## Task 15: Final verification + live smoke + +**Goal:** Run the full test suite, do a live-against-Gainwell sanity +check, and prepare the merge commit. + +- [ ] **Step 1: Run backend full suite** + +```bash +cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5 +``` + +Expected: at least the baseline + 17 new tests, all passing. + +- [ ] **Step 2: Run frontend full suite** + +```bash +npm test 2>&1 | tail -10 +npm run typecheck 2>&1 | tail -5 +``` + +Expected: pass; typecheck clean. + +- [ ] **Step 3: Live sanity** + +```bash +docker compose restart backend frontend +sleep 10 +curl -s -b cookies.txt http://localhost:8080/api/admin/scheduler/status | jq . +# Force a tick +curl -s -X POST -b cookies.txt http://localhost:8080/api/admin/scheduler/tick | jq . +# Hit the chain endpoint for one of the matched claims +curl -s -b cookies.txt http://localhost:8080/api/claims//chain | jq . +``` + +Expected: scheduler running, tick processes inbound (or no-ops if +empty), chain endpoint returns the claim's pieces. + +- [ ] **Step 4: SFTP operation timeout live test** + +```bash +# Force the inbound MFT host to be unreachable and see whether +# the scheduler stalls gracefully (timeout in 30s) rather than +# hangs indefinitely. +docker compose exec backend sh -c "iptables -A OUTPUT -p tcp --dport 22 -d mft.example.com -j DROP" || true +# Trigger a tick via the admin endpoint +START=$(date +%s) +curl -s -X POST -b cookies.txt http://localhost:8080/api/admin/scheduler/tick | jq . +END=$(date +%s) +echo "Tick took $((END-START))s" +``` + +Expected: tick completes (returns 200) within ~35s (30s timeout + slack). + +- [ ] **Step 5: Verify the suffix-less 835 test path one more time** + +```bash +ls /tmp/cyclone-stage/FromHPE/*.x12 2>/dev/null | head -5 +curl -s -b cookies.txt "http://localhost:8080/api/admin/scheduler/processed-files?limit=5" | jq . +``` + +Expected: any previously-skipped suffix-less file is now marked +`status: "ok"`, `parser_used: "parse_835"`. + +--- + +## Task 16: Merge into Version-1.0.0 + +**Goal:** Single atomic merge of `sp27-remittances-architecture-refactor` +into `Version-1.0.0`. + +- [ ] **Step 1: Re-check working tree** + +```bash +git status +``` + +Expected: clean. + +- [ ] **Step 2: Switch to Version-1.0.0** + +```bash +git checkout Version-1.0.0 +``` + +- [ ] **Step 3: Merge SP27 with no-ff** + +```bash +git merge --no-ff sp27-remittances-architecture-refactor \ + -m "merge: SP27 remittances architecture refactor into Version-1.0.0" +``` + +Expected: a single merge commit capturing all 14 implementation +commits + the spec + the plan + tests. + +- [ ] **Step 4: Push (only if explicitly requested)** + +No push unless the user asks. Local merge is the audit trail. + +--- + +## Self-review + +1. **Spec coverage:** + - §1 Tier 1: handlers split (Tasks 2-5) ✓ + - §1 Tier 1: helpers dedup (Tasks 1 + 6) ✓ + - §1 Tier 1: INBOUND_RE loosen (Task 7) ✓ + - §1 Tier 1: SFTP timeouts (Task 8) ✓ + - §1 Tier 1: Scheduler.status() deeper (Task 9) ✓ + - §1 Tier 2: atomic 835+reconcile (Task 10) ✓ + - §1 Tier 2: chain endpoint (Task 12) ✓ + - §1 Tier 2: claim.rejected_after_remit (Task 13) ✓ + - §1 Tier 2: match invariants (Task 11) ✓ + - §5 API surface: chain endpoint (Task 12) ✓ + - §6 Env vars: CYCLONE_SFTP_OP_TIMEOUT_SECONDS (Task 8) ✓ + - §8 Test plan: 17 new tests created in Tasks 1-14 ✓ +2. **Placeholders:** None — every step shows exact file paths and code. +3. **Type consistency:** `HandleResult` is a `@dataclass` defined in + Task 2 and used by every Task 2-13 handler. `HANDLERS` dict is + defined once in Task 2 and extended in Tasks 3-5. +4. **Live-test reminder:** Every task includes a live-test + autoreview + + commit step. The 6/15-6/19 inbound batch is the regression test + for Task 7 — the file names without `_file_type.x12` should now + route to `parse_835` instead of being skipped. + +## Memory: where the live-test sandbox files live + +- `/tmp/cyclone-stage/` — operator staging dir for SFTP stub / production mirror +- `cookies.txt` (in repo root, gitignored) — curl cookie jar for `http://localhost:8080` after a fresh login +- The production scheduler runs in `cyclone-backend-1` (docker) and polls `FromHPE` from Gainwell's MFT +- For a fresh login without git-tracking a cookie file: `curl -c /tmp/c.txt -X POST -d 'username=...&password=...' http://localhost:8080/api/auth/login`