cyclone/docs/superpowers/plans/2026-06-29-cyclone-remittances-architecture-refactor.md

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

---

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

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

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

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

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

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

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:

"""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

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:

"""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:

"""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

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

cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5

Expected: identical counts to baseline.

• Step 7: Commit

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:

"""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/<sample>.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)

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

"""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:

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

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

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

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

# 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

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:

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:

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

# 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:

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

# 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:

from cyclone.handlers import handle_835 as _handle_835

• Step 5: Verify + live-test + autoreview + commit

Commit message:

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

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:

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

"""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

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

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:

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:

"""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

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:

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<tracking>` 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:

# 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<tpid>\d+)-(?P<file_type>[A-Z0-9]+)_(?P<tracking>M[A-Z0-9]+)"
    r"-(?P<ts>\d{17})-1of1\.(?P<ext>x12)$"
)

• Step 4: Verify new tests pass + existing test_filenames.py still passes

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

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

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:

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:

"""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

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:

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:

files = await asyncio.to_thread(self._list_inbound)

in scheduler.py:_tick_impl with:

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

cd backend && .venv/bin/pytest tests/test_sftp_op_timeout.py -v 2>&1 | tail -10

Expected: pass.

• Step 6: Verify full suite

cd backend && .venv/bin/pytest tests/ --tb=line -q 2>&1 | tail -5

Expected: identical counts.

• Step 7: Live-test + autoreview + commit

Commit message:

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:

"""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

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:

@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:

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

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

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:

"""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:

"""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

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:

"""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:

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

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

# 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:

"""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:

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

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

# 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.

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

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

// @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

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

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

import { useClaimChain } from "@/hooks/useClaimChain";

// In the drawer body:
const { chain, loading } = useClaimChain(claim.id);

<section>
  <h3>Chain</h3>
  {loading ? <Skeleton /> : (
    <>
      <div>Submission: {chain?.submission?.batch_id ?? "—"}</div>
      <div>999 Ack: {chain?.ack_999?.ack_code ?? "(pending)"}</div>
      <div>277CA: {chain?.ack_277ca?.classification ?? "(pending)"}</div>
      <div>Remittance: {chain?.remittance ? `${chain.remittance.total_paid}` : "(pending)"}</div>
      {chain && chain.missing.length > 0 && (
        <em>Awaiting: {chain.missing.join(", ")}</em>
      )}
    </>
  )}
</section>

• Step 7: Verify frontend tests + suite + commit

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

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

npm test 2>&1 | tail -10
npm run typecheck 2>&1 | tail -5

Expected: pass; typecheck clean.

• Step 3: Live sanity

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/<some-id>/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

# 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

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

git status

Expected: clean.

• Step 2: Switch to Version-1.0.0

git checkout Version-1.0.0

• Step 3: Merge SP27 with no-ff

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