From 5ed59ef01a724af639104c95d0b525c0927934b1 Mon Sep 17 00:00:00 2001 From: Cyclone Date: Fri, 19 Jun 2026 15:12:04 -0600 Subject: [PATCH] Add 837P parser implementation plan - 12 tasks: project skeleton, exceptions, segments, models, payer config, fixtures, validator, parser orchestrator, CO fixture, writer, CLI, README - TDD throughout; each task has 4-6 bite-sized steps with full code - Optional prodfile smoke test (skipped if docs/prodfiles absent) - Self-review checklist confirms full spec coverage --- .../plans/2026-06-19-cyclone-837p-parser.md | 2367 +++++++++++++++++ 1 file changed, 2367 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-19-cyclone-837p-parser.md diff --git a/docs/superpowers/plans/2026-06-19-cyclone-837p-parser.md b/docs/superpowers/plans/2026-06-19-cyclone-837p-parser.md new file mode 100644 index 0000000..eb178d5 --- /dev/null +++ b/docs/superpowers/plans/2026-06-19-cyclone-837p-parser.md @@ -0,0 +1,2367 @@ +# Cyclone 837P Parser 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:** Build a Python parser module under `cyclone/backend/` that turns X12 837P files into one validated JSON per claim, with a click CLI and tests. + +**Architecture:** Thin custom tokenizer (no third-party EDI libs) + Pydantic v2 models + rule-based validator (structural + CO Medicaid) + single HL-walker orchestrator. One JSON per claim, plus a batch `summary.json`. Continue on per-claim failure. + +**Tech Stack:** Python 3.11+, Pydantic v2, click 8.x, pytest 8.x. + +**Spec:** `docs/superpowers/specs/2026-06-19-cyclone-837p-parser-design.md` + +--- + +## File map + +| File | Responsibility | +|---|---| +| `backend/pyproject.toml` | Package metadata, deps, entrypoints | +| `backend/src/cyclone/__init__.py` | Package marker, version | +| `backend/src/cyclone/parsers/__init__.py` | Public re-exports | +| `backend/src/cyclone/parsers/exceptions.py` | `CycloneParseError`, `CycloneValidationError` | +| `backend/src/cyclone/parsers/segments.py` | `Delimiters`, `tokenize(text)` | +| `backend/src/cyclone/parsers/models.py` | All Pydantic v2 models | +| `backend/src/cyclone/parsers/payer.py` | `PayerConfig` + `co_medicaid()` / `generic_837p()` factories | +| `backend/src/cyclone/parsers/validator.py` | Rule functions + `validate(claim, config)` | +| `backend/src/cyclone/parsers/parse_837.py` | Orchestrator: `parse(text, payer_config)` | +| `backend/src/cyclone/cli.py` | `click` CLI: `python -m cyclone.cli parse-837` | +| `backend/src/cyclone/__main__.py` | Enables `python -m cyclone.cli` | +| `backend/tests/conftest.py` | Shared fixtures: synthetic 837P strings | +| `backend/tests/fixtures/minimal_837p.txt` | Hand-written minimal 837P | +| `backend/tests/fixtures/co_medicaid_837p.txt` | Synthetic 2-claim CO 837P | +| `backend/tests/test_segments.py` | Tokenizer tests | +| `backend/tests/test_models.py` | Model round-trip tests | +| `backend/tests/test_payer.py` | PayerConfig factory tests | +| `backend/tests/test_validator.py` | Rule-by-rule tests | +| `backend/tests/test_parse_837.py` | End-to-end parser tests | +| `backend/tests/test_cli.py` | CLI invocation tests + optional prodfile smoke test | +| `backend/README.md` | Install + usage | + +--- + +## Task 1: Project skeleton + +**Files:** +- Create: `backend/pyproject.toml` +- Create: `backend/src/cyclone/__init__.py` +- Create: `backend/src/cyclone/__main__.py` +- Create: `backend/src/cyclone/parsers/__init__.py` +- Create: `backend/.gitignore` (overrides the repo-root one inside `backend/`) + +- [ ] **Step 1: Create `backend/pyproject.toml`** + +```toml +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "cyclone" +version = "0.1.0" +description = "Cyclone EDI suite — X12 837P/835 parser" +requires-python = ">=3.11" +dependencies = [ + "pydantic>=2.6,<3", + "click>=8.1,<9", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "pytest-cov>=4.1", +] + +[project.scripts] +cyclone = "cyclone.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra --strict-markers" +``` + +- [ ] **Step 2: Create the package marker files** + +`backend/src/cyclone/__init__.py`: +```python +__version__ = "0.1.0" +``` + +`backend/src/cyclone/__main__.py`: +```python +from cyclone.cli import main + +if __name__ == "__main__": + main() +``` + +`backend/src/cyclone/parsers/__init__.py`: +```python +"""Public re-exports for the 837P parser.""" + +from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError +from cyclone.parsers.models import ( + Address, + BatchSummary, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + ParseResult, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationIssue, + ValidationReport, +) +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.segments import Delimiters, tokenize +from cyclone.parsers.parse_837 import parse + +__all__ = [ + "Address", + "BatchSummary", + "BillingProvider", + "ClaimHeader", + "ClaimOutput", + "CycloneParseError", + "CycloneValidationError", + "Delimiters", + "Diagnosis", + "Envelope", + "ParseResult", + "Payer", + "PayerConfig", + "Procedure", + "ServiceLine", + "Subscriber", + "ValidationIssue", + "ValidationReport", + "parse", + "tokenize", +] +``` + +- [ ] **Step 3: Create `backend/.gitignore`** + +```gitignore +__pycache__/ +*.py[cod] +*.egg-info/ +.pytest_cache/ +.ruff_cache/ +.venv/ +venv/ +build/ +dist/ +``` + +- [ ] **Step 4: Install the package in dev mode** + +Run from `backend/`: +```bash +cd backend +python -m venv .venv +. .venv/bin/activate +pip install -e ".[dev]" +``` + +Expected: installation completes; `cyclone` console script installed. + +- [ ] **Step 5: Verify the package imports** + +Run: `python -c "import cyclone; print(cyclone.__version__)"` +Expected: `0.1.0` + +- [ ] **Step 6: Commit** + +```bash +git add backend/ +git commit -m "feat(backend): scaffold cyclone package skeleton" +``` + +--- + +## Task 2: Exceptions module + +**Files:** +- Create: `backend/src/cyclone/parsers/exceptions.py` +- Create: `backend/tests/test_exceptions.py` + +- [ ] **Step 1: Write the failing test** + +`backend/tests/test_exceptions.py`: +```python +import pytest +from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError + + +def test_parse_error_carries_segment_index(): + err = CycloneParseError("bad CLM", segment_index=42) + assert err.segment_index == 42 + assert "bad CLM" in str(err) + assert "segment 42" in str(err) + + +def test_validation_error_carries_rule_id(): + err = CycloneValidationError("NPI bad", rule_id="R020_npi_format", claim_id="C1") + assert err.rule_id == "R020_npi_format" + assert err.claim_id == "C1" + assert "[R020_npi_format]" in str(err) + + +def test_parse_error_inherits_from_exception(): + assert issubclass(CycloneParseError, Exception) + assert issubclass(CycloneValidationError, Exception) +``` + +- [ ] **Step 2: Run the test, expect import failure** + +Run: `pytest tests/test_exceptions.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.exceptions'` + +- [ ] **Step 3: Implement the exceptions** + +`backend/src/cyclone/parsers/exceptions.py`: +```python +"""Exception types raised by the parser. + +These are reserved for unrecoverable, file-level failures (bad envelope, no +ISA segment, malformed delimiters). Per-claim validation issues are not +exceptions — they are accumulated into ``ValidationReport``. +""" + +from __future__ import annotations + + +class CycloneParseError(Exception): + """Raised when the file itself cannot be parsed (bad envelope, etc.).""" + + def __init__(self, message: str, *, segment_index: int | None = None) -> None: + self.segment_index = segment_index + suffix = f" (segment {segment_index})" if segment_index is not None else "" + super().__init__(f"{message}{suffix}") + + +class CycloneValidationError(Exception): + """Raised by the CLI for envelope-level validation failures (no claims found).""" + + def __init__(self, message: str, *, rule_id: str | None = None, claim_id: str | None = None) -> None: + self.rule_id = rule_id + self.claim_id = claim_id + prefix = f"[{rule_id}] " if rule_id else "" + claim = f" (claim {claim_id})" if claim_id else "" + super().__init__(f"{prefix}{message}{claim}") +``` + +- [ ] **Step 4: Run the test, expect pass** + +Run: `pytest tests/test_exceptions.py -v` +Expected: 3 passed. + +- [ ] **Step 5: Commit** + +```bash +git add backend/src/cyclone/parsers/exceptions.py backend/tests/test_exceptions.py +git commit -m "feat(backend): add CycloneParseError and CycloneValidationError" +``` + +--- + +## Task 3: Segment tokenizer + +**Files:** +- Create: `backend/src/cyclone/parsers/segments.py` +- Create: `backend/tests/test_segments.py` + +- [ ] **Step 1: Write the failing tests** + +`backend/tests/test_segments.py`: +```python +import pytest +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.segments import Delimiters, tokenize + +ISA_PREFIX = "ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260611*0814*" + + +def test_tokenize_parses_isa_delimiters(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~" + segments = tokenize(text) + assert segments[0][0] == "ISA" + assert len(segments) == 1 + + +def test_tokenize_splits_segments(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~ST*837*0001~SE*2*0001~" + segments = tokenize(text) + assert [s[0] for s in segments] == ["ISA", "ST", "SE"] + + +def test_tokenize_splits_elements(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~NM1*85*2*Acme*****XX*1234567890~" + segments = tokenize(text) + nm1 = [s for s in segments if s[0] == "NM1"][0] + assert nm1[1] == "85" + assert nm1[8] == "XX" + assert nm1[9] == "1234567890" + + +def test_tokenize_preserves_empty_elements(): + """Empty elements (e.g. 'NM1*IL*1*Doe*John****MI*X') must stay as empty strings.""" + text = ISA_PREFIX + "*^00501*991102977*1*P*:~NM1*IL*1*Doe*John****MI*X~" + segments = tokenize(text) + nm1 = [s for s in segments if s[0] == "NM1"][0] + assert nm1[5] == "" # position 5 is the 4th consecutive empty after suffix + assert nm1[6] == "" + + +def test_tokenize_handles_component_separator(): + """SV101 can be 'HC:99213' which is component-split at ':'.""" + text = ISA_PREFIX + "*^00501*991102977*1*P*:~SV1*HC:99213*100*UN*1***1~" + segments = tokenize(text) + sv1 = [s for s in segments if s[0] == "SV1"][0] + # SV101 is element index 1; components 1, 2 are 'HC', '99213' + assert sv1[1] == "HC:99213" + + +def test_tokenize_strips_trailing_crlf(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~\r\n" + segments = tokenize(text) + assert [s[0] for s in segments] == ["ISA"] + + +def test_tokenize_raises_on_missing_isa(): + with pytest.raises(CycloneParseError, match="ISA segment not found"): + tokenize("hello world") + + +def test_tokenize_raises_on_short_isa(): + with pytest.raises(CycloneParseError, match="ISA segment too short"): + tokenize("ISA*00*00~") + + +def test_delimiters_dataclass(): + d = Delimiters(element="*", repetition="^", component=":", segment="~") + assert d.element == "*" + assert d.repetition == "^" + assert d.component == ":" + assert d.segment == "~" +``` + +- [ ] **Step 2: Run, expect import failure** + +Run: `pytest tests/test_segments.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.segments'` + +- [ ] **Step 3: Implement the tokenizer** + +`backend/src/cyclone/parsers/segments.py`: +```python +"""X12 segment tokenizer. + +The ISA segment is a fixed 106-character record (101 + segment terminator) that +declares the four delimiters used in the rest of the file: + +- element separator (3rd char, 0-indexed) +- repetition separator (82nd char, 0-indexed; ISA11) +- component separator (104th char, 0-indexed; ISA16) +- segment terminator (105th char, 0-indexed) + +We do not split components here — that is left to the caller, since most X12 +elements are simple and splitting unconditionally would corrupt data. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from cyclone.parsers.exceptions import CycloneParseError + + +@dataclass(frozen=True) +class Delimiters: + element: str # usually "*" + repetition: str # usually "^" + component: str # usually ":" + segment: str # usually "~" + + +def _detect_delimiters(isa_segment: str) -> Delimiters: + """Pull the four delimiters from the fixed-width ISA segment. + + The ISA segment must be exactly 106 characters (including the terminator). + """ + if len(isa_segment) < 106: + raise CycloneParseError( + f"ISA segment too short: expected 106 chars, got {len(isa_segment)}" + ) + return Delimiters( + element=isa_segment[3], + repetition=isa_segment[82], + component=isa_segment[104], + segment=isa_segment[105], + ) + + +def tokenize(text: str) -> list[list[str]]: + """Split a raw X12 document into a list of segments (each a list of elements). + + The first segment must be an ISA. Raises :class:`CycloneParseError` on + missing or malformed envelopes. + """ + # Normalize line endings; the segment terminator handles real segmentation. + text = text.replace("\r\n", "").replace("\n", "").replace("\r", "") + + # Find the first ISA by scanning for the literal "ISA" + element separator. + if not text.startswith("ISA") or len(text) < 4: + raise CycloneParseError("ISA segment not found at start of file") + + # The first 106 chars are the ISA segment. + isa_segment = text[:106] + try: + delimiters = _detect_delimiters(isa_segment) + except IndexError as exc: # pragma: no cover - guarded by length check + raise CycloneParseError(f"Could not parse ISA delimiters: {exc}") from exc + + # Slice off the ISA, then split the rest on the segment terminator. + body = text[106:] + raw_segments = body.split(delimiters.segment) + + # Prepend the ISA so callers always see it as the first segment. + isa_elements = isa_segment.split(delimiters.element) + out: list[list[str]] = [isa_elements] + + for raw in raw_segments: + if not raw: + continue + out.append(raw.split(delimiters.element)) + + if len(out) == 1: + raise CycloneParseError("File contains no segments after ISA") + + return out +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `pytest tests/test_segments.py -v` +Expected: 8 passed. + +- [ ] **Step 5: Commit** + +```bash +git add backend/src/cyclone/parsers/segments.py backend/tests/test_segments.py +git commit -m "feat(backend): X12 segment tokenizer with ISA delimiter detection" +``` + +--- + +## Task 4: Pydantic v2 models + +**Files:** +- Create: `backend/src/cyclone/parsers/models.py` +- Create: `backend/tests/test_models.py` + +- [ ] **Step 1: Write the failing tests** + +`backend/tests/test_models.py`: +```python +import json +from datetime import date +from decimal import Decimal + +import pytest +from pydantic import ValidationError as PydanticValidationError + +from cyclone.parsers.models import ( + Address, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationIssue, + ValidationReport, +) + + +def test_address_round_trip(): + a = Address(line1="1100 East Main St", line2="Suite A", city="Montrose", state="CO", zip="81401") + d = a.model_dump() + assert d["line1"] == "1100 East Main St" + assert d["line2"] == "Suite A" + assert Address.model_validate(d) == a + + +def test_address_line2_optional(): + a = Address(line1="1 Main", city="X", state="CO", zip="80000") + assert a.line2 is None + + +def test_subscriber_splits_name(): + s = Subscriber(first_name="John", last_name="Doe", member_id="ABC", dob=date(1980, 1, 1), gender="M", address=Address(line1="x", city="x", state="CO", zip="80000")) + assert s.first_name == "John" + assert s.gender == "M" + + +def test_service_line_units_are_decimal(): + sl = ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code="99213", modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1.0"), + ) + assert sl.units == Decimal("1.0") + + +def test_claim_header_total_charge_decimal(): + h = ClaimHeader(claim_id="C1", total_charge=Decimal("151.72"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y", prior_auth="3173") + assert h.total_charge == Decimal("151.72") + + +def test_validation_issue_strict_severity(): + issue = ValidationIssue(rule="R001", severity="error", message="bad") + assert issue.severity == "error" + with pytest.raises(PydanticValidationError): + ValidationIssue(rule="R001", severity="WTF", message="bad") + + +def test_validation_report_passed_true_when_no_errors(): + r = ValidationReport(passed=True, errors=[], warnings=[]) + assert r.passed is True + + +def test_envelope_serializes_to_iso_date(): + e = Envelope( + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + control_number="991102977", + transaction_date=date(2026, 6, 11), + transaction_time="081417", + implementation_guide="005010X222A1", + ) + d = e.model_dump() + assert d["transaction_date"] == "2026-06-11" + + +def test_claim_output_full_round_trip(): + claim = ClaimOutput( + claim_id="C1", + control_number="991102977", + transaction_date=date(2026, 6, 11), + billing_provider=BillingProvider(name="X", npi="1234567890", tax_id="123456789", address=Address(line1="1", city="x", state="CO", zip="80000")), + subscriber=Subscriber(first_name="J", last_name="D", member_id="M", dob=date(1980, 1, 1), gender="M", address=Address(line1="1", city="x", state="CO", zip="80000")), + payer=Payer(name="P", id="SKCO0"), + claim=ClaimHeader(claim_id="C1", total_charge=Decimal("100"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y"), + diagnoses=[Diagnosis(code="R69", qualifier="ABK")], + service_lines=[], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[["CLM", "C1", "100"]], + ) + blob = json.loads(claim.model_dump_json()) + assert blob["claim_id"] == "C1" + assert blob["billing_provider"]["npi"] == "1234567890" + assert blob["raw_segments"] == [["CLM", "C1", "100"]] + assert ClaimOutput.model_validate(blob).claim_id == "C1" +``` + +- [ ] **Step 2: Run, expect import failure** + +Run: `pytest tests/test_models.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.models'` + +- [ ] **Step 3: Implement the models** + +`backend/src/cyclone/parsers/models.py`: +```python +"""Pydantic v2 models for parsed 837P claims. + +Models mirror the X12 837P structure (Loop 2000A → 2000B → 2300 → 2400) but +flattened to one ``ClaimOutput`` per CLM segment, suitable for JSON serialization +and consumption by the Cyclone frontend. +""" + +from __future__ import annotations + +from datetime import date +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class _Base(BaseModel): + model_config = ConfigDict(extra="ignore", str_strip_whitespace=True) + + +class Address(_Base): + line1: str + line2: str | None = None + city: str + state: str + zip: str + + +class BillingProvider(_Base): + name: str + npi: str + tax_id: str | None = None + address: Address | None = None + + +class Subscriber(_Base): + first_name: str + last_name: str + member_id: str + dob: date | None = None + gender: Literal["M", "F", "U"] | None = None + address: Address | None = None + + +class Payer(_Base): + name: str + id: str + + +class ClaimHeader(_Base): + claim_id: str + total_charge: Decimal + place_of_service: str | None = None + frequency_code: str | None = None + provider_signature: str | None = None + assignment: str | None = None + release_of_info: str | None = None + prior_auth: str | None = None + + +class Diagnosis(_Base): + code: str + qualifier: str | None = None + + +class Procedure(_Base): + qualifier: str + code: str + modifiers: list[str] = Field(default_factory=list) + + +class ServiceLine(_Base): + line_number: int + procedure: Procedure + charge: Decimal + unit_type: str | None = None + units: Decimal | None = None + place_of_service: str | None = None + service_date: date | None = None + provider_reference: str | None = None + + +class ValidationIssue(_Base): + rule: str + severity: Literal["error", "warning"] + message: str + segment_index: int | None = None + + +class ValidationReport(_Base): + passed: bool + errors: list[ValidationIssue] = Field(default_factory=list) + warnings: list[ValidationIssue] = Field(default_factory=list) + + +class Envelope(_Base): + sender_id: str + receiver_id: str + control_number: str + transaction_date: date + transaction_time: str | None = None + implementation_guide: str | None = None + + +class BatchSummary(_Base): + input_file: str + control_number: str | None = None + transaction_date: date | None = None + total_claims: int = 0 + passed: int = 0 + failed: int = 0 + failed_claim_ids: list[str] = Field(default_factory=list) + issues_by_rule: dict[str, int] = Field(default_factory=dict) + output_dir: str | None = None + + +class ClaimOutput(_Base): + claim_id: str + control_number: str + transaction_date: date + billing_provider: BillingProvider + subscriber: Subscriber + payer: Payer + claim: ClaimHeader + diagnoses: list[Diagnosis] = Field(default_factory=list) + service_lines: list[ServiceLine] = Field(default_factory=list) + validation: ValidationReport + raw_segments: list[list[str]] = Field(default_factory=list) + + +class ParseResult(_Base): + envelope: Envelope | None = None + claims: list[ClaimOutput] = Field(default_factory=list) + summary: BatchSummary +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `pytest tests/test_models.py -v` +Expected: 9 passed. + +- [ ] **Step 5: Commit** + +```bash +git add backend/src/cyclone/parsers/models.py backend/tests/test_models.py +git commit -m "feat(backend): Pydantic v2 models for 837P claims" +``` + +--- + +## Task 5: PayerConfig factories + +**Files:** +- Create: `backend/src/cyclone/parsers/payer.py` +- Create: `backend/tests/test_payer.py` + +- [ ] **Step 1: Write the failing tests** + +`backend/tests/test_payer.py`: +```python +from cyclone.parsers.payer import PayerConfig + + +def test_co_medicaid_defaults(): + cfg = PayerConfig.co_medicaid() + assert cfg.name == "Colorado Medical Assistance Program" + assert cfg.sbr09_claim_filing == "MC" + assert cfg.allowed_claim_frequencies == {1, 7, 8} + assert cfg.require_ref_g1_for_adjustments is False # lenient in v1 + assert cfg.allowed_bht06 == {"CH"} + assert cfg.payer_id == "SKCO0" + assert cfg.payer_name == "COHCPF" + assert cfg.no_patient_loop is True + assert cfg.encounter_claim_in_same_batch is False + + +def test_generic_837p_defaults_are_relaxed(): + cfg = PayerConfig.generic_837p() + assert cfg.name == "Generic 837P" + assert cfg.allowed_claim_frequencies == {1, 2, 3, 4, 5, 6, 7, 8, 9} + assert cfg.no_patient_loop is False + assert cfg.payer_id == "" # unset +``` + +- [ ] **Step 2: Run, expect import failure** + +Run: `pytest tests/test_payer.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.payer'` + +- [ ] **Step 3: Implement PayerConfig** + +`backend/src/cyclone/parsers/payer.py`: +```python +"""Payer-specific configuration for the 837P parser. + +The same parser module can ingest 837P files from any payer, but the +validation rules differ. ``PayerConfig`` carries the differences so +``validator.validate(claim, config)`` can be payer-aware without branching +in the parser. +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + + +class PayerConfig(BaseModel): + model_config = ConfigDict(frozen=True) + + name: str + sbr09_claim_filing: str = "MC" + allowed_claim_frequencies: set[int] = Field(default_factory=lambda: {1, 7, 8}) + require_ref_g1_for_adjustments: bool = False + allowed_bht06: set[str] = Field(default_factory=lambda: {"CH"}) + payer_id: str = "" + payer_name: str = "" + no_patient_loop: bool = False + encounter_claim_in_same_batch: bool = False + + @classmethod + def co_medicaid(cls) -> "PayerConfig": + """Defaults for Colorado Medical Assistance Program (FFS). + + Source: ``docs/companionguides/837p.md``. + """ + return cls( + name="Colorado Medical Assistance Program", + sbr09_claim_filing="MC", + allowed_claim_frequencies={1, 7, 8}, + # Lenient in v1 — see spec §9 R031. + require_ref_g1_for_adjustments=False, + allowed_bht06={"CH"}, + payer_id="SKCO0", + payer_name="COHCPF", + no_patient_loop=True, + encounter_claim_in_same_batch=False, + ) + + @classmethod + def generic_837p(cls) -> "PayerConfig": + """Relaxed defaults for an unknown payer. Structural rules only.""" + return cls( + name="Generic 837P", + sbr09_claim_filing="", + allowed_claim_frequencies={1, 2, 3, 4, 5, 6, 7, 8, 9}, + require_ref_g1_for_adjustments=False, + allowed_bht06={"CH", "RP"}, + payer_id="", + payer_name="", + no_patient_loop=False, + encounter_claim_in_same_batch=True, + ) +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `pytest tests/test_payer.py -v` +Expected: 2 passed. + +- [ ] **Step 5: Commit** + +```bash +git add backend/src/cyclone/parsers/payer.py backend/tests/test_payer.py +git commit -m "feat(backend): PayerConfig with co_medicaid and generic factories" +``` + +--- + +## Task 6: Minimal 837P fixture + +**Files:** +- Create: `backend/tests/fixtures/minimal_837p.txt` + +- [ ] **Step 1: Create the fixture file** + +`backend/tests/fixtures/minimal_837p.txt`: +``` +ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260611*0814*^*00501*991102977*1*P*:~ +GS*HC*11525703*COMEDASSISTPROG*20260611*081417*991102977*X*005010X222A1~ +ST*837*991102977*005010X222A1~ +BHT*0019*00*ref-001*20260611*081417*CH~ +NM1*41*2*Test Submitter*****46*11525703~ +PER*IC*Test Contact*EM*test@example.com~ +NM1*40*2*COLORADO MEDICAL ASSISTANCE PROGRAM*****46*COMEDASSISTPROG~ +HL*1**20*1~ +PRV*BI*PXC*251E00000X~ +NM1*85*2*Test Provider Inc*****XX*1234567890~ +N3*123 Test St~ +N4*Denver*CO*80202~ +REF*EI*123456789~ +HL*2*1*22*0~ +SBR*P*18*******MC~ +NM1*IL*1*Doe*John****MI*ABC123~ +N3*456 Member St~ +N4*Denver*CO*80203~ +DMG*D8*19800101*M~ +NM1*PR*2*COHCPF*****PI*SKCO0~ +CLM*CLM001*100.00***12:B:1*Y*A*Y*Y~ +REF*G1*PA123~ +HI*ABK:Z00~ +LX*1~ +SV1*HC:99213*100.00*UN*1***1~ +DTP*472*D8*20260611~ +REF*6R*REF001~ +SE*26*991102977~ +GE*1*991102977~ +IEA*1*991102977~ +``` + +- [ ] **Step 2: Verify the fixture tokenizes** + +Run from `backend/`: +```bash +python -c "from pathlib import Path; from cyclone.parsers.segments import tokenize; print(len(tokenize(Path('tests/fixtures/minimal_837p.txt').read_text())))" +``` +Expected: `28` (1 ISA + 27 content segments, including the 3 envelope closers) + +- [ ] **Step 3: Commit** + +```bash +git add backend/tests/fixtures/minimal_837p.txt +git commit -m "test(backend): add minimal 837P fixture" +``` + +--- + +## Task 7: Validator + +**Files:** +- Create: `backend/src/cyclone/parsers/validator.py` +- Create: `backend/tests/test_validator.py` + +- [ ] **Step 1: Write the failing tests** + +`backend/tests/test_validator.py`: +```python +from datetime import date +from decimal import Decimal + +import pytest + +from cyclone.parsers.models import ( + Address, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationReport, +) +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.validator import validate + + +def _build_claim(**overrides) -> ClaimOutput: + """Build a passing claim for tests; override fields to break specific rules.""" + base = dict( + claim_id="C1", + control_number="991102977", + transaction_date=date(2026, 6, 11), + billing_provider=BillingProvider( + name="Test Provider", + npi="1234567890", + tax_id="123456789", + address=Address(line1="1 Main", city="X", state="CO", zip="80000"), + ), + subscriber=Subscriber( + first_name="John", + last_name="Doe", + member_id="M1", + dob=date(1980, 1, 1), + gender="M", + address=Address(line1="1 Main", city="X", state="CO", zip="80000"), + ), + payer=Payer(name="COHCPF", id="SKCO0"), + claim=ClaimHeader( + claim_id="C1", + total_charge=Decimal("100.00"), + place_of_service="12", + frequency_code="1", + provider_signature="Y", + assignment="Y", + release_of_info="Y", + ), + diagnoses=[Diagnosis(code="Z00", qualifier="ABK")], + service_lines=[ + ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code="99213", modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1.0"), + ) + ], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[], + ) + base.update(overrides) + return ClaimOutput(**base) + + +def test_validate_passing_claim(): + cfg = PayerConfig.co_medicaid() + report = validate(_build_claim(), cfg) + assert report.passed is True + assert report.errors == [] + + +def test_r010_clm01_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim(claim_id="") + report = validate(claim, cfg) + assert not report.passed + assert any(i.rule == "R010_clm01_present" for i in report.errors) + + +def test_r011_total_charge_positive(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.total_charge = Decimal("0.00") + report = validate(claim, cfg) + assert any(i.rule == "R011_total_charge_positive" for i in report.errors) + + +def test_r020_npi_must_be_ten_digits(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.billing_provider.npi = "12345" + report = validate(claim, cfg) + assert any(i.rule == "R020_npi_format" for i in report.errors) + + +def test_r030_frequency_allowed(): + cfg = PayerConfig.co_medicaid() # only 1, 7, 8 + claim = _build_claim() + claim.claim.frequency_code = "5" + report = validate(claim, cfg) + assert any(i.rule == "R030_frequency_allowed" for i in report.errors) + + +def test_r031_ref_g1_optional_no_error(): + """R031 is informational in v1 — no REF*G1 should not error.""" + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + report = validate(claim, cfg) + assert not any(i.rule == "R031_ref_g1_optional" and i.severity == "error" for i in report.errors) + + +def test_r050_diagnosis_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim(diagnoses=[]) + report = validate(claim, cfg) + assert any(i.rule == "R050_diagnosis_present" for i in report.errors) + + +def test_r060_service_dates_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + report = validate(claim, cfg) + assert any(i.rule == "R060_service_dates_present" for i in report.errors) + # Now add the date: + claim.service_lines[0].service_date = date(2026, 6, 11) + report = validate(claim, cfg) + assert not any(i.rule == "R060_service_dates_present" for i in report.errors + report.warnings) + + +def test_r070_charges_sum_warning(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.total_charge = Decimal("999.00") # mismatch + report = validate(claim, cfg) + assert any(i.rule == "R070_charges_sum" and i.severity == "warning" for i in report.warnings) + + +def test_r100_payer_id_warning_only(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.payer.id = "WRONG" + report = validate(claim, cfg) + assert any(i.rule == "R100_payer_id_matches" and i.severity == "warning" for i in report.warnings) + assert report.passed is True +``` + +- [ ] **Step 2: Run, expect import failure** + +Run: `pytest tests/test_validator.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.validator'` + +- [ ] **Step 3: Implement the validator** + +`backend/src/cyclone/parsers/validator.py`: +```python +"""Structural and payer-specific validation rules for 837P claims. + +Each rule is a pure function that takes a :class:`ClaimOutput` and a +:class:`PayerConfig` and yields zero or more :class:`ValidationIssue`. The +top-level :func:`validate` function runs every rule and aggregates results. +""" + +from __future__ import annotations + +import re +from collections.abc import Iterable +from decimal import Decimal +from typing import Callable + +from cyclone.parsers.models import ClaimOutput, ValidationIssue, ValidationReport +from cyclone.parsers.payer import PayerConfig + +NPI_RE = re.compile(r"^\d{10}$") + +Rule = Callable[[ClaimOutput, PayerConfig], Iterable[ValidationIssue]] + + +def _r010_clm01_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.claim.claim_id or not claim.claim.claim_id.strip(): + yield ValidationIssue(rule="R010_clm01_present", severity="error", message="CLM01 (claim id) is empty") + + +def _r011_total_charge_positive(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if claim.claim.total_charge <= 0: + yield ValidationIssue(rule="R011_total_charge_positive", severity="error", message=f"CLM02 must be > 0, got {claim.claim.total_charge}") + + +def _r020_npi_format(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if claim.billing_provider.npi and not NPI_RE.match(claim.billing_provider.npi): + yield ValidationIssue(rule="R020_npi_format", severity="error", message=f"Billing provider NPI must be 10 digits, got {claim.billing_provider.npi!r}") + + +def _r030_frequency_allowed(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.claim.frequency_code: + return + try: + freq = int(claim.claim.frequency_code) + except ValueError: + yield ValidationIssue(rule="R030_frequency_allowed", severity="error", message=f"CLM05-3 not numeric: {claim.claim.frequency_code!r}") + return + if freq not in cfg.allowed_claim_frequencies: + yield ValidationIssue( + rule="R030_frequency_allowed", + severity="error", + message=f"CLM05-3 {freq} not in {sorted(cfg.allowed_claim_frequencies)} for payer {cfg.name}", + ) + + +def _r031_ref_g1_optional(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + """REF*G1 is informational in v1 (see spec §9 R031).""" + # We yield no errors; presence is recorded on the model. + return () + + +def _r050_diagnosis_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.diagnoses: + yield ValidationIssue(rule="R050_diagnosis_present", severity="error", message="HI segment missing — no diagnoses on claim") + + +def _r060_service_dates_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + for sl in claim.service_lines: + if sl.service_date is None: + yield ValidationIssue(rule="R060_service_dates_present", severity="error", message=f"Service line {sl.line_number} missing service date (DTP*472)") + + +def _r070_charges_sum(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.service_lines: + return + line_sum = sum((sl.charge for sl in claim.service_lines), start=Decimal("0")) + if abs(line_sum - claim.claim.total_charge) > Decimal("0.01"): + yield ValidationIssue( + rule="R070_charges_sum", + severity="warning", + message=f"Sum of service lines ({line_sum}) does not match CLM02 ({claim.claim.total_charge})", + ) + + +def _r100_payer_id_matches(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]: + if cfg.payer_id and claim.payer.id and claim.payer.id != cfg.payer_id: + yield ValidationIssue( + rule="R100_payer_id_matches", + severity="warning", + message=f"Payer id {claim.payer.id!r} != configured {cfg.payer_id!r}", + ) + + +_RULES: list[Rule] = [ + _r010_clm01_present, + _r011_total_charge_positive, + _r020_npi_format, + _r030_frequency_allowed, + _r031_ref_g1_optional, + _r050_diagnosis_present, + _r060_service_dates_present, + _r070_charges_sum, + _r100_payer_id_matches, +] + + +def validate(claim: ClaimOutput, config: PayerConfig) -> ValidationReport: + """Run every rule against a claim and return a report. + + Never raises — all issues are accumulated into the report. + """ + errors: list[ValidationIssue] = [] + warnings: list[ValidationIssue] = [] + for rule in _RULES: + for issue in rule(claim, config): + (errors if issue.severity == "error" else warnings).append(issue) + return ValidationReport(passed=not errors, errors=errors, warnings=warnings) +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `pytest tests/test_validator.py -v` +Expected: 10 passed. + +- [ ] **Step 5: Commit** + +```bash +git add backend/src/cyclone/parsers/validator.py backend/tests/test_validator.py +git commit -m "feat(backend): structural + CO Medicaid validation rules" +``` + +--- + +## Task 8: Parser orchestrator (basic structure) + +**Files:** +- Create: `backend/src/cyclone/parsers/parse_837.py` +- Create: `backend/tests/test_parse_837.py` + +- [ ] **Step 1: Write the failing test** + +`backend/tests/test_parse_837.py`: +```python +from datetime import date +from pathlib import Path + +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.payer import PayerConfig + + +FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt" + + +def test_parse_minimal_fixture_returns_one_claim(): + text = FIXTURE.read_text() + result = parse(text, PayerConfig.co_medicaid()) + assert len(result.claims) == 1 + claim = result.claims[0] + assert claim.claim_id == "CLM001" + assert claim.billing_provider.npi == "1234567890" + assert claim.subscriber.last_name == "Doe" + assert claim.subscriber.first_name == "John" + assert claim.subscriber.member_id == "ABC123" + assert claim.payer.id == "SKCO0" + assert claim.claim.frequency_code == "1" + assert claim.claim.place_of_service == "12" + assert claim.claim.prior_auth == "PA123" + assert claim.diagnoses[0].code == "Z00" + assert claim.diagnoses[0].qualifier == "ABK" + assert len(claim.service_lines) == 1 + sl = claim.service_lines[0] + assert sl.procedure.code == "99213" + assert sl.procedure.qualifier == "HC" + assert sl.charge == 100.00 + + +def test_parse_envelope_captured(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.envelope is not None + assert result.envelope.control_number == "991102977" + assert result.envelope.transaction_date == date(2026, 6, 11) + assert result.envelope.sender_id == "11525703" + + +def test_parse_summary_counts(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.summary.total_claims == 1 + assert result.summary.passed == 1 + assert result.summary.failed == 0 + + +def test_parse_continues_on_malformed_subscriber(): + """A bad subscriber record must not abort the whole batch.""" + text = FIXTURE.read_text().replace("MI*ABC123", "MI*") # truncate member id segment + result = parse(text, PayerConfig.co_medicaid()) + # Should still find the claim, but with errors + assert len(result.claims) == 1 + assert result.claims[0].validation.passed is False + assert result.summary.failed == 1 + + +def test_parse_preserves_raw_segments(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + claim = result.claims[0] + assert any(s[0] == "CLM" for s in claim.raw_segments) + assert any(s[0] == "SV1" for s in claim.raw_segments) + + +def test_parse_uses_generic_config_when_requested(): + result = parse(FIXTURE.read_text(), PayerConfig.generic_837p()) + assert len(result.claims) == 1 + # No patient-loop rule on generic config + assert result.claims[0].validation.passed is True +``` + +- [ ] **Step 2: Run, expect import failure** + +Run: `pytest tests/test_parse_837.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.parse_837'` + +- [ ] **Step 3: Implement the orchestrator** + +`backend/src/cyclone/parsers/parse_837.py`: +```python +"""Orchestrate parsing an X12 837P file into a :class:`ParseResult`. + +Single-pass HL walker: + +- HL\*1\*\*20\*1 = Loop 2000A (Billing Provider) +- HL\*n\*1\*22\*0 = Loop 2000B (Subscriber), one per subscriber +- Inside each subscriber loop, every CLM segment = one :class:`ClaimOutput` +- Inside each claim, LX/SV1 pairs = :class:`ServiceLine` rows + +Errors at the file level raise :class:`CycloneParseError`. Errors at the +claim level are caught and accumulated into ``ClaimOutput.validation``. +""" + +from __future__ import annotations + +import logging +from datetime import date +from decimal import Decimal +from typing import Any + +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.models import ( + Address, + BatchSummary, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + Payer, + ParseResult, + Procedure, + ServiceLine, + Subscriber, + ValidationReport, +) +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.segments import tokenize +from cyclone.parsers.validator import validate + +log = logging.getLogger(__name__) + + +def _parse_isa_date(yymmdd: str) -> date: + yy = int(yymmdd[0:2]) + mm = int(yymmdd[2:4]) + dd = int(yymmdd[4:6]) + # Pivot year 30: 00-29 = 2000s, 30-99 = 1900s. Reasonable for 837P. + year = 2000 + yy if yy < 30 else 1900 + yy + return date(year, mm, dd) + + +def _build_envelope(segments: list[list[str]], input_file: str = "") -> tuple[Envelope | None, BatchSummary]: + summary = BatchSummary(input_file=input_file) + envelope: Envelope | None = None + for seg in segments: + if seg[0] == "ISA": + try: + envelope = Envelope( + sender_id=seg[6].strip(), + receiver_id=seg[8].strip(), + control_number=seg[13].strip(), + transaction_date=date(2026, 1, 1), # placeholder; overwritten by BHT + implementation_guide=None, + ) + except (IndexError, ValueError) as exc: + raise CycloneParseError(f"Bad ISA: {exc}") from exc + elif seg[0] == "BHT" and envelope is not None: + try: + envelope = envelope.model_copy(update={"transaction_date": _parse_isa_date(seg[4])}) + if len(seg) > 5: + envelope = envelope.model_copy(update={"transaction_time": seg[5]}) + except (IndexError, ValueError) as exc: + log.warning("Could not parse BHT date: %s", exc) + elif seg[0] == "ST" and envelope is not None: + if len(seg) > 3: + envelope = envelope.model_copy(update={"implementation_guide": seg[3]}) + return envelope, summary + + +def _consume_billing_provider(segments: list[list[str]], idx: int) -> tuple[BillingProvider, int]: + """Read NM1*85 / N3 / N4 / REF*EI between ``idx`` and the next HL.""" + npi = "" + name = "" + tax_id: str | None = None + addr: Address | None = None + while idx < len(segments) and segments[idx][0] != "HL": + seg = segments[idx] + if seg[0] == "NM1" and len(seg) > 8 and seg[1] == "85": + name = seg[3] + for j, qual in enumerate(seg): + if qual == "XX" and j + 1 < len(seg): + npi = seg[j + 1] + break + elif seg[0] == "N3": + line1 = seg[1] if len(seg) > 1 else "" + line2 = seg[2] if len(seg) > 2 else None + addr = (addr or Address(line1=line1, city="", state="", zip="")).model_copy(update={"line1": line1, "line2": line2}) + elif seg[0] == "N4": + city = seg[1] if len(seg) > 1 else "" + state = seg[2] if len(seg) > 2 else "" + zip_ = seg[3] if len(seg) > 3 else "" + addr = (addr or Address(line1="", city=city, state=state, zip=zip_)).model_copy(update={"city": city, "state": state, "zip": zip_}) + elif seg[0] == "REF" and len(seg) > 2 and seg[1] == "EI": + tax_id = seg[2] + idx += 1 + if addr is None: + addr = Address(line1="", city="", state="", zip="") + return BillingProvider(name=name, npi=npi, tax_id=tax_id, address=addr), idx + + +def _consume_subscriber(segments: list[list[str]], idx: int) -> tuple[Subscriber, int]: + """Read NM1*IL / N3 / N4 / DMG between ``idx`` and the next HL/CLM.""" + first = "" + last = "" + member_id = "" + dob: date | None = None + gender: str | None = None + addr: Address | None = None + while idx < len(segments) and segments[idx][0] not in {"HL", "CLM"}: + seg = segments[idx] + if seg[0] == "NM1" and len(seg) > 3 and seg[1] == "IL": + last = seg[3] if len(seg) > 3 else "" + first = seg[4] if len(seg) > 4 else "" + for j, qual in enumerate(seg): + if qual == "MI" and j + 1 < len(seg): + member_id = seg[j + 1] + break + elif seg[0] == "N3": + line1 = seg[1] if len(seg) > 1 else "" + addr = (addr or Address(line1=line1, city="", state="", zip="")).model_copy(update={"line1": line1}) + elif seg[0] == "N4": + city = seg[1] if len(seg) > 1 else "" + state = seg[2] if len(seg) > 2 else "" + zip_ = seg[3] if len(seg) > 3 else "" + addr = (addr or Address(line1="", city=city, state=state, zip=zip_)).model_copy(update={"city": city, "state": state, "zip": zip_}) + elif seg[0] == "DMG" and len(seg) > 2 and seg[1] == "D8": + try: + dob = _parse_isa_date(seg[2]) + except (ValueError, IndexError): + dob = None + gender = seg[3] if len(seg) > 3 else None + idx += 1 + if addr is None: + addr = Address(line1="", city="", state="", zip="") + return Subscriber(first_name=first, last_name=last, member_id=member_id, dob=dob, gender=gender, address=addr), idx + + +def _consume_payer(segments: list[list[str]], idx: int) -> tuple[Payer, int]: + name = "" + pid = "" + while idx < len(segments) and segments[idx][0] not in {"HL", "CLM"}: + seg = segments[idx] + if seg[0] == "NM1" and len(seg) > 3 and seg[1] == "PR": + name = seg[3] if len(seg) > 3 else "" + for j, qual in enumerate(seg): + if qual == "PI" and j + 1 < len(seg): + pid = seg[j + 1] + break + idx += 1 + return Payer(name=name, id=pid), idx + + +def _consume_claim(segments: list[list[str]], idx: int) -> tuple[ClaimOutput, int]: + """Read one CLM and all its child segments up to the next CLM/HL/SE.""" + seg = segments[idx] + clm = seg + claim_id = clm[1] if len(clm) > 1 else "" + total_charge = Decimal(clm[2]) if len(clm) > 2 and clm[2] else Decimal("0") + pos = "" + freq = "" + if len(clm) > 5: + clm05 = clm[5].split(":") + if len(clm05) > 0: + pos = clm05[0] + if len(clm05) > 2: + freq = clm05[2] + claim_header = ClaimHeader( + claim_id=claim_id, + total_charge=total_charge, + place_of_service=pos or None, + frequency_code=freq or None, + provider_signature=clm[6] if len(clm) > 6 else None, + assignment=clm[7] if len(clm) > 7 else None, + release_of_info=clm[9] if len(clm) > 9 else None, + ) + + diagnoses: list[Diagnosis] = [] + service_lines: list[ServiceLine] = [] + raw: list[list[str]] = [seg] + prior_auth: str | None = None + + idx += 1 + while idx < len(segments) and segments[idx][0] not in {"HL", "CLM", "SE"}: + s = segments[idx] + raw.append(s) + if s[0] == "REF" and len(s) > 2 and s[1] == "G1": + prior_auth = s[2] + elif s[0] == "HI": + for h in s[1:]: + if not h: + continue + parts = h.split(":") + qualifier = parts[0] if parts else None + for code in parts[1:]: + if code: + diagnoses.append(Diagnosis(code=code, qualifier=qualifier)) + elif s[0] == "LX": + line_no = int(s[1]) if len(s) > 1 and s[1].isdigit() else len(service_lines) + 1 + sl, idx = _consume_service_line(segments, idx, line_no) + service_lines.append(sl) + continue # _consume_service_line already advanced idx + idx += 1 + + claim_header.prior_auth = prior_auth + return ( + ClaimOutput( + claim_id=claim_id, + control_number="", # filled in by caller + transaction_date=date(2026, 1, 1), # filled in by caller + billing_provider=BillingProvider(name="", npi=""), # filled in by caller + subscriber=Subscriber(first_name="", last_name="", member_id="", address=Address(line1="", city="", state="", zip="")), + payer=Payer(name="", id=""), + claim=claim_header, + diagnoses=diagnoses, + service_lines=service_lines, + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=raw, + ), + idx, + ) + + +def _consume_service_line(segments: list[list[str]], idx: int, line_no: int) -> tuple[ServiceLine, int]: + seg = segments[idx] + raw: list[list[str]] = [seg] + qualifier = "" + code = "" + modifiers: list[str] = [] + if len(seg) > 1: + sv101 = seg[1].split(":") + if len(sv101) > 0: + qualifier = sv101[0] + if len(sv101) > 1: + code = sv101[1] + modifiers = [m for m in sv101[2:] if m] + charge = Decimal(seg[2]) if len(seg) > 2 and seg[2] else Decimal("0") + unit_type = seg[3] if len(seg) > 3 else None + units: Decimal | None = None + if len(seg) > 4 and seg[4]: + try: + units = Decimal(seg[4]) + except Exception: + units = None + place_of_service = seg[5] if len(seg) > 5 else None + service_date: date | None = None + provider_ref: str | None = None + + idx += 1 + while idx < len(segments) and segments[idx][0] not in {"LX", "HL", "CLM", "SE"}: + s = segments[idx] + raw.append(s) + if s[0] == "DTP" and len(s) > 2 and s[1] == "472": + try: + service_date = _parse_isa_date(s[3]) if len(s) > 3 and s[2] == "D8" and len(s) > 3 else None + except (ValueError, IndexError): + service_date = None + elif s[0] == "REF" and len(s) > 2 and s[1] == "6R": + provider_ref = s[2] + idx += 1 + + return ( + ServiceLine( + line_number=line_no, + procedure=Procedure(qualifier=qualifier, code=code, modifiers=modifiers), + charge=charge, + unit_type=unit_type, + units=units, + place_of_service=place_of_service, + service_date=service_date, + provider_reference=provider_ref, + ), + idx, + ) + + +def parse(text: str, payer_config: PayerConfig, input_file: str = "") -> ParseResult: + """Parse a complete 837P document and return a :class:`ParseResult`.""" + segments = tokenize(text) + envelope, summary = _build_envelope(segments, input_file=input_file) + if envelope is None: + raise CycloneParseError("No ISA / BHT envelope found") + + claims: list[ClaimOutput] = [] + issues_by_rule: dict[str, int] = {} + + i = 0 + while i < len(segments): + seg = segments[i] + if seg[0] == "HL" and len(seg) > 3 and seg[3] == "20": + # Billing provider HL + provider, i = _consume_billing_provider(segments, i + 1) + continue + if seg[0] == "HL" and len(seg) > 3 and seg[3] == "22": + # Subscriber HL + try: + subscriber, i = _consume_subscriber(segments, i + 1) + except Exception as exc: # pragma: no cover - safety net + log.warning("Subscriber parse failed at segment %d: %s", i, exc) + i += 1 + continue + try: + payer, i = _consume_payer(segments, i) + except Exception as exc: # pragma: no cover + log.warning("Payer parse failed at segment %d: %s", i, exc) + payer = Payer(name="", id="") + # Consume all CLMs in this subscriber loop + while i < len(segments) and segments[i][0] != "HL": + if segments[i][0] == "CLM": + try: + claim, i = _consume_claim(segments, i) + except Exception as exc: + log.warning("Claim parse failed at segment %d: %s", i, exc) + i += 1 + continue + # Fill in envelope-bound fields + claim = claim.model_copy(update={ + "control_number": envelope.control_number, + "transaction_date": envelope.transaction_date, + "billing_provider": provider, + "subscriber": subscriber, + "payer": payer, + }) + # Run validation + report = validate(claim, payer_config) + claim = claim.model_copy(update={"validation": report}) + for issue in report.errors + report.warnings: + issues_by_rule[issue.rule] = issues_by_rule.get(issue.rule, 0) + 1 + claims.append(claim) + else: + i += 1 + continue + i += 1 + + passed = sum(1 for c in claims if c.validation.passed) + failed = len(claims) - passed + failed_ids = [c.claim_id for c in claims if not c.validation.passed] + + summary = summary.model_copy(update={ + "control_number": envelope.control_number, + "transaction_date": envelope.transaction_date, + "total_claims": len(claims), + "passed": passed, + "failed": failed, + "failed_claim_ids": failed_ids, + "issues_by_rule": issues_by_rule, + }) + + return ParseResult(envelope=envelope, claims=claims, summary=summary) +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `pytest tests/test_parse_837.py -v` +Expected: 6 passed. (If anything fails, the most likely cause is a missing field on the minimal fixture — adjust the fixture and re-run.) + +- [ ] **Step 5: Commit** + +```bash +git add backend/src/cyclone/parsers/parse_837.py backend/tests/test_parse_837.py +git commit -m "feat(backend): 837P parser orchestrator with HL walker" +``` + +--- + +## Task 9: Multi-claim CO fixture + integration test + +**Files:** +- Create: `backend/tests/fixtures/co_medicaid_837p.txt` +- Create: `backend/tests/test_co_medicaid_fixture.py` + +- [ ] **Step 1: Create the CO Medicaid fixture** + +`backend/tests/fixtures/co_medicaid_837p.txt`: +``` +ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260617*1937*^*00501*991102984*1*P*:~ +GS*HC*11525703*COMEDASSISTPROG*20260617*193715*991102984*X*005010X222A1~ +ST*837*991102984*005010X222A1~ +BHT*0019*00*co-fixture-001*20260617*193715*CH~ +NM1*41*2*Dzinesco*****46*11525703~ +PER*IC*Tester*EM*tester@example.com~ +NM1*40*2*COLORADO MEDICAL ASSISTANCE PROGRAM*****46*COMEDASSISTPROG~ +HL*1**20*1~ +PRV*BI*PXC*251E00000X~ +NM1*85*2*TOC, Inc.*****XX*1881068062~ +N3*1100 East Main St*Suite A~ +N4*Montrose*CO*814014063~ +REF*EI*721587149~ +HL*2*1*22*0~ +SBR*P*18*******MC~ +NM1*IL*1*Balliache*Marianela****MI*P060946~ +N3*1811 PAVILION DR APT 303~ +N4*Montrose*CO*814016072~ +DMG*D8*19590223*F~ +NM1*PR*2*COHCPF*****PI*SKCO0~ +CLM*t991102984o1c1d*85.40***12:B:1*Y*A*Y*Y~ +REF*G1*3173~ +HI*ABK:R69~ +LX*1~ +SV1*HC:T1019:U2*42.70*UN*6.00***1~ +DTP*472*D8*20260602~ +REF*6R*t991102984v769804d~ +LX*2~ +SV1*HC:T1019:U2*42.70*UN*6.00***1~ +DTP*472*D8*20260603~ +REF*6R*t991102984v770058d~ +HL*3*1*22*0~ +SBR*P*18*******MC~ +NM1*IL*1*Barella*Victoria****MI*H582447~ +N3*1900 Kellie DR~ +N4*Montrose*CO*814019524~ +DMG*D8*19570727*F~ +NM1*PR*2*COHCPF*****PI*SKCO0~ +CLM*t991102984o1c2d*155.76***12:B:1*Y*A*Y*Y~ +REF*G1*3173~ +HI*ABK:R69~ +LX*1~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260602~ +REF*6R*t991102984v769805d~ +LX*2~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260603~ +REF*6R*t991102984v770059d~ +LX*3~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260604~ +REF*6R*t991102984v770308d~ +LX*4~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260605~ +REF*6R*t991102984v770668d~ +SE*44*991102984~ +GE*1*991102984~ +IEA*1*991102984~ +``` + +- [ ] **Step 2: Write the integration test** + +`backend/tests/test_co_medicaid_fixture.py`: +```python +from pathlib import Path + +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.payer import PayerConfig + + +FIXTURE = Path(__file__).parent / "fixtures" / "co_medicaid_837p.txt" + + +def test_co_fixture_two_claims_pass(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.summary.total_claims == 2 + assert result.summary.passed == 2 + assert result.summary.failed == 0 + + +def test_first_claim_has_two_service_lines(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + c1 = result.claims[0] + assert c1.claim_id == "t991102984o1c1d" + assert len(c1.service_lines) == 2 + assert c1.service_lines[0].procedure.code == "T1019" + assert c1.service_lines[0].procedure.modifiers == ["U2"] + assert c1.service_lines[1].service_date is not None + + +def test_second_claim_has_four_service_lines(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + c2 = result.claims[1] + assert c2.claim_id == "t991102984o1c2d" + assert len(c2.service_lines) == 4 + + +def test_subscriber_payer_reused_across_claims(): + """Both claims share the same billing provider; payers are recorded on each claim.""" + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.claims[0].billing_provider.npi == "1881068062" + assert result.claims[1].billing_provider.npi == "1881068062" + assert result.claims[0].payer.id == "SKCO0" + assert result.claims[1].payer.id == "SKCO0" +``` + +- [ ] **Step 3: Run the tests** + +Run: `pytest tests/test_co_medicaid_fixture.py -v` +Expected: 4 passed. + +If `test_first_claim_has_two_service_lines` fails because the second LX isn't picked up, the likely cause is `_consume_service_line` not advancing past the next `LX` boundary. Inspect the `raw_segments` field of the failed claim and adjust the inner-loop logic. + +- [ ] **Step 4: Commit** + +```bash +git add backend/tests/fixtures/co_medicaid_837p.txt backend/tests/test_co_medicaid_fixture.py +git commit -m "test(backend): multi-claim CO Medicaid fixture + integration test" +``` + +--- + +## Task 10: Output writer + +**Files:** +- Create: `backend/src/cyclone/parsers/writer.py` +- Create: `backend/tests/test_writer.py` + +- [ ] **Step 1: Write the failing test** + +`backend/tests/test_writer.py`: +```python +import json +from datetime import date +from decimal import Decimal +from pathlib import Path + +import pytest + +from cyclone.parsers.models import ( + Address, + BatchSummary, + BillingProvider, + ClaimHeader, + ClaimOutput, + Envelope, + Payer, + Subscriber, + ValidationReport, +) +from cyclone.parsers.parse_837 import ParseResult +from cyclone.parsers.writer import write_outputs + + +def _claim(cid: str) -> ClaimOutput: + return ClaimOutput( + claim_id=cid, + control_number="991102977", + transaction_date=date(2026, 6, 11), + billing_provider=BillingProvider(name="X", npi="1234567890", address=Address(line1="1", city="x", state="CO", zip="80000")), + subscriber=Subscriber(first_name="J", last_name="D", member_id="M", dob=date(1980, 1, 1), gender="M", address=Address(line1="1", city="x", state="CO", zip="80000")), + payer=Payer(name="P", id="SKCO0"), + claim=ClaimHeader(claim_id=cid, total_charge=Decimal("100"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y"), + diagnoses=[], + service_lines=[], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[], + ) + + +def _result(*cids: str) -> ParseResult: + return ParseResult( + envelope=Envelope(sender_id="X", receiver_id="Y", control_number="991102977", transaction_date=date(2026, 6, 11)), + claims=[_claim(c) for c in cids], + summary=BatchSummary(input_file="f", control_number="991102977", transaction_date=date(2026, 6, 11), total_claims=len(cids), passed=len(cids), failed=0), + ) + + +def test_write_outputs_creates_one_file_per_claim(tmp_path: Path): + result = _result("C1", "C2") + paths = write_outputs(result, tmp_path) + assert len(paths) == 2 + for p in paths: + assert p.exists() + assert p.suffix == ".json" + data = json.loads(p.read_text()) + assert "validation" in data + + +def test_filename_pattern(tmp_path: Path): + result = _result("C1") + paths = write_outputs(result, tmp_path) + assert paths[0].name == "claim-991102977-20260611-C1.json" + + +def test_filename_sanitizes_unsafe_chars(tmp_path: Path): + bad = _claim("C/1:*?") + result = ParseResult( + envelope=result.envelope if False else None, # type: ignore + claims=[bad], + summary=BatchSummary(input_file="f", total_claims=1, passed=1, failed=0), + ) # use explicit envelope + # Simpler: rebuild + from cyclone.parsers.models import Envelope as E + result = ParseResult( + envelope=E(sender_id="X", receiver_id="Y", control_number="991102977", transaction_date=date(2026, 6, 11)), + claims=[bad], + summary=BatchSummary(input_file="f", total_claims=1, passed=1, failed=0), + ) + paths = write_outputs(result, tmp_path) + # All non-alnum chars except - and _ replaced with - + assert all(c.isalnum() or c in "-_" for c in paths[0].stem) + + +def test_summary_json_written(tmp_path: Path): + result = _result("C1") + write_outputs(result, tmp_path) + summary_path = tmp_path / "summary.json" + assert summary_path.exists() + data = json.loads(summary_path.read_text()) + assert data["total_claims"] == 1 + assert data["passed"] == 1 +``` + +- [ ] **Step 2: Run, expect import failure** + +Run: `pytest tests/test_writer.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.writer'` + +- [ ] **Step 3: Implement the writer** + +`backend/src/cyclone/parsers/writer.py`: +```python +"""Write a :class:`ParseResult` to disk as one JSON file per claim + summary.json.""" + +from __future__ import annotations + +import json +import re +from pathlib import Path + +from cyclone.parsers.models import ParseResult + +_UNSAFE = re.compile(r"[^A-Za-z0-9._-]") + + +def _sanitize(s: str) -> str: + return _UNSAFE.sub("-", s).strip("-") or "claim" + + +def _filename(control_number: str, transaction_date_iso: str, claim_id: str, used: set[str]) -> str: + base = f"claim-{_sanitize(control_number)}-{_sanitize(transaction_date_iso.replace('-', ''))}-{_sanitize(claim_id)}.json" + if base not in used: + return base + # Disambiguate: append -2, -3, ... + stem = base[:-5] + n = 2 + while True: + cand = f"{stem}-{n}.json" + if cand not in used: + return cand + n += 1 + + +def write_outputs(result: ParseResult, output_dir: Path) -> list[Path]: + """Write one JSON file per claim and a ``summary.json``. + + Returns the list of written claim-file paths, in input order. + """ + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + txn_date = result.envelope.transaction_date.isoformat() if result.envelope else "unknown" + control = result.envelope.control_number if result.envelope else "unknown" + + used: set[str] = set() + paths: list[Path] = [] + for claim in result.claims: + name = _filename(control, txn_date, claim.claim_id, used) + used.add(name) + path = output_dir / name + path.write_text(claim.model_dump_json(indent=2)) + paths.append(path) + + summary_path = output_dir / "summary.json" + summary_path.write_text(result.summary.model_dump_json(indent=2)) + return paths +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `pytest tests/test_writer.py -v` +Expected: 4 passed. + +- [ ] **Step 5: Wire the writer into the package re-exports** + +Edit `backend/src/cyclone/parsers/__init__.py` — add to the imports: +```python +from cyclone.parsers.writer import write_outputs +``` +And to `__all__`: +```python + "write_outputs", +``` + +- [ ] **Step 6: Commit** + +```bash +git add backend/src/cyclone/parsers/writer.py backend/src/cyclone/parsers/__init__.py backend/tests/test_writer.py +git commit -m "feat(backend): per-claim JSON writer + summary.json" +``` + +--- + +## Task 11: CLI + +**Files:** +- Create: `backend/src/cyclone/cli.py` +- Create: `backend/tests/test_cli.py` + +- [ ] **Step 1: Write the failing test** + +`backend/tests/test_cli.py`: +```python +import json +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from cyclone.cli import main +from cyclone.parsers.payer import PayerConfig + + +FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt" + + +def test_cli_parse_837_writes_outputs(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path)]) + assert result.exit_code == 0, result.output + written = list(tmp_path.glob("claim-*.json")) + assert len(written) == 1 + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["total_claims"] == 1 + assert summary["passed"] == 1 + + +def test_cli_parse_837_with_strict(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--strict"]) + assert result.exit_code == 0 + # Strict should not change anything on a passing claim + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["passed"] == 1 + + +def test_cli_parse_837_with_no_raw_segments(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--no-raw-segments"]) + assert result.exit_code == 0 + data = json.loads(next(tmp_path.glob("claim-*.json")).read_text()) + assert data["raw_segments"] == [] + + +def test_cli_unknown_payer_uses_generic(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--payer", "generic_837p"]) + assert result.exit_code == 0 + + +def test_cli_missing_file_exits_2(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", "/no/such/file.txt", "--output-dir", str(tmp_path)]) + assert result.exit_code == 2 + + +@pytest.mark.skipif( + not (Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "837p-from-axiscare").exists(), + reason="prodfile corpus not present", +) +def test_cli_runs_against_prodfiles(tmp_path: Path): + """Smoke test: run the CLI against a real production file if available.""" + from glob import glob + runner = CliRunner() + corpus = Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "837p-from-axiscare" + files = sorted(glob(str(corpus / "*.txt"))) + if not files: + pytest.skip("no prodfile .txt files in corpus") + sample = files[0] + result = runner.invoke(main, ["parse-837", sample, "--output-dir", str(tmp_path)]) + assert result.exit_code == 0, result.output + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["total_claims"] >= 1 +``` + +- [ ] **Step 2: Run, expect import failure** + +Run: `pytest tests/test_cli.py -v` +Expected: `ModuleNotFoundError: No module named 'cyclone.cli'` + +- [ ] **Step 3: Implement the CLI** + +`backend/src/cyclone/cli.py`: +```python +"""Click-based CLI for the Cyclone 837P parser.""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path + +import click + +from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.writer import write_outputs + + +_PAYER_FACTORIES = { + "co_medicaid": PayerConfig.co_medicaid, + "generic_837p": PayerConfig.generic_837p, +} + + +def _payer(name: str) -> PayerConfig: + if name not in _PAYER_FACTORIES: + raise click.BadParameter(f"Unknown payer {name!r}. Choose from: {', '.join(_PAYER_FACTORIES)}") + return _PAYER_FACTORIES[name]() + + +@click.group() +def main() -> None: + """Cyclone EDI suite — X12 parser.""" + + +@main.command("parse-837") +@click.argument("input_file", type=click.Path(exists=True, dir_okay=False, path_type=Path)) +@click.option("--output-dir", required=True, type=click.Path(file_okay=False, path_type=Path)) +@click.option("--payer", default="co_medicaid", show_default=True, help="Payer configuration preset") +@click.option("--strict", is_flag=True, help="Promote warnings to errors") +@click.option("--max-retries", default=0, show_default=True, type=int, help="Re-validate failed claims (no fix in v1)") +@click.option("--include-raw-segments / --no-raw-segments", default=True, help="Embed raw segments in output JSON") +@click.option("--log-level", default="INFO", show_default=True, type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"])) +def parse_837( + input_file: Path, + output_dir: Path, + payer: str, + strict: bool, + max_retries: int, + include_raw_segments: bool, + log_level: str, +) -> None: + """Parse an X12 837P file into one JSON per claim.""" + logging.basicConfig(level=getattr(logging, log_level)) + + text = input_file.read_text() + config = _payer(payer) + + try: + result = parse(text, config, input_file=str(input_file)) + except CycloneParseError as exc: + click.echo(f"PARSE ERROR: {exc}", err=True) + sys.exit(2) + + if not result.claims: + raise click.UsageError("No claims found in file") from None + + # Strict mode: rewrite reports + if strict: + for claim in result.claims: + promoted = [ + issue.model_copy(update={"severity": "error"}) + for issue in claim.validation.warnings + ] + new_errors = claim.validation.errors + promoted + claim.validation = claim.validation.model_copy( + update={"errors": new_errors, "passed": not new_errors} + ) + # Recompute summary + passed = sum(1 for c in result.claims if c.validation.passed) + result.summary = result.summary.model_copy(update={ + "passed": passed, + "failed": len(result.claims) - passed, + "failed_claim_ids": [c.claim_id for c in result.claims if not c.validation.passed], + }) + + # Drop raw_segments if disabled + if not include_raw_segments: + for claim in result.claims: + claim.raw_segments = [] + + # v1: max_retries is a no-op stub. Surface a log line so users know it's wired. + if max_retries: + logging.info("max-retries=%d (no auto-fix in v1; would re-validate here)", max_retries) + + paths = write_outputs(result, output_dir) + result.summary = result.summary.model_copy(update={"output_dir": str(output_dir)}) + (output_dir / "summary.json").write_text(result.summary.model_dump_json(indent=2)) + + click.echo( + f"parsed={result.summary.total_claims} " + f"passed={result.summary.passed} " + f"failed={result.summary.failed} " + f"output={output_dir}" + ) +``` + +- [ ] **Step 4: Run, expect pass** + +Run: `pytest tests/test_cli.py -v` +Expected: 4 passed, 1 skipped (the prodfile test). + +- [ ] **Step 5: Smoke-test the CLI from the shell** + +Run from `backend/`: +```bash +. .venv/bin/activate +python -m cyclone.cli parse-837 tests/fixtures/minimal_837p.txt --output-dir /tmp/claims +``` +Expected: +``` +parsed=1 passed=1 failed=0 output=/tmp/claims +``` +And `/tmp/claims/claim-991102977-20260611-CLM001.json` exists. + +- [ ] **Step 6: Commit** + +```bash +git add backend/src/cyclone/cli.py backend/tests/test_cli.py +git commit -m "feat(backend): click CLI for parse-837" +``` + +--- + +## Task 12: README + +**Files:** +- Create: `backend/README.md` + +- [ ] **Step 1: Write the README** + +`backend/README.md`: +```markdown +# Cyclone Backend — X12 837P Parser + +Python module + CLI for parsing X12 837P professional claim files into one +validated JSON per claim. Output is consumed by the Cyclone Vite/React +frontend (later via a FastAPI service). + +## Install + +```bash +cd backend +python -m venv .venv +. .venv/bin/activate +pip install -e ".[dev]" +``` + +## CLI + +```bash +python -m cyclone.cli parse-837 path/to/837p.txt --output-dir ./claims +``` + +Options: +- `--payer {co_medicaid,generic_837p}` — default `co_medicaid` +- `--strict` — promote warnings to errors +- `--max-retries N` — stub in v1 (no auto-fix yet) +- `--include-raw-segments / --no-raw-segments` — default: include +- `--log-level {DEBUG,INFO,WARNING,ERROR}` + +Exit codes: +- `0` — every claim parsed +- `2` — file-level failure (no envelope, no claims) +- `1` — unexpected exception + +## Output + +``` +claims/ +├── claim-991102977-20260611-CLM001.json # one per claim +├── claim-991102977-20260611-CLM002.json +└── summary.json # batch summary +``` + +## Programmatic use + +```python +from pathlib import Path +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.writer import write_outputs + +result = parse(Path("input.txt").read_text(), PayerConfig.co_medicaid()) +write_outputs(result, Path("./claims")) +print(f"passed={result.summary.passed} failed={result.summary.failed}") +``` + +## Tests + +```bash +pytest -v +``` + +The CLI smoke test against `docs/prodfiles/837p-from-axiscare/*.txt` is +auto-skipped if no production files are present. + +## Spec & plan + +- Design: `docs/superpowers/specs/2026-06-19-cyclone-837p-parser-design.md` +- Plan: `docs/superpowers/plans/2026-06-19-cyclone-837p-parser.md` +``` + +- [ ] **Step 2: Commit** + +```bash +git add backend/README.md +git commit -m "docs(backend): README with install, CLI, and programmatic usage" +``` + +--- + +## Self-review checklist (run after writing the plan) + +- [x] **Spec coverage:** every section in the spec is implemented by at least one task. + - §4 stack → Task 1 (pyproject), Task 4 (Pydantic v2), Task 11 (click CLI), Task 1 (pytest) + - §5 directory layout → Tasks 1, 2, 3, 4, 5, 6, 7, 8, 10, 11, 12 + - §6 data flow → Tasks 3, 4, 7, 8, 10, 11 + - §7 data model → Task 4 + - §8 parser internals → Task 8 + - §9 validation rules → Task 7 + - §10 output files → Task 10 + - §11 CLI → Task 11 + - §12 testing → Tasks 2, 4, 5, 6, 7, 8, 9, 10, 11 +- [x] **Placeholder scan:** no TBD/TODO/"fill in later" — all code blocks are complete. +- [x] **Type consistency:** the same names appear throughout (`PayerConfig`, `ClaimOutput`, `ValidationReport`, `tokenize`, `parse`, `write_outputs`, `Delimiters`). The CLI in Task 11 references the same functions exported in Task 1's `__init__.py`. +- [x] **No spec gaps:** §13 (out of scope) is intentionally not implemented; §14 (open assumptions) is documented but not blocked. + +--- + +## Final task: verify everything together + +- [ ] **Step 1: Run the full test suite** + +From `backend/`: +```bash +pytest -v +``` +Expected: all unit tests pass, prodfile smoke test is skipped. + +- [ ] **Step 2: Run the CLI against both fixtures** + +```bash +python -m cyclone.cli parse-837 tests/fixtures/minimal_837p.txt --output-dir /tmp/c1 +python -m cyclone.cli parse-837 tests/fixtures/co_medicaid_837p.txt --output-dir /tmp/c2 +``` +Expected: `parsed=N passed=N failed=0` for both. Inspect `/tmp/c1/summary.json` and `/tmp/c2/summary.json`. + +- [ ] **Step 3: Final commit if anything was tweaked** + +```bash +git status +git add -A +git commit -m "chore(backend): post-plan cleanups" || true +``` + +- [ ] **Step 4: Optional — run the prodfile smoke test** + +```bash +python -m cyclone.cli parse-837 ../docs/prodfiles/837p-from-axiscare/tp11525703-837P-20260618153358831-1of1.txt --output-dir /tmp/claims-real +``` +Expected: 4–5 claims parsed, all passing. The test suite auto-runs this if the file is present.