Files

T

Cyclone 5ed59ef01a 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

2026-06-19 15:12:04 -06:00

74 KiB

Raw Permalink Blame History

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

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

__version__ = "0.1.0"

backend/src/cyclone/__main__.py:

from cyclone.cli import main

if __name__ == "__main__":
    main()

backend/src/cyclone/parsers/__init__.py:

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

__pycache__/
*.py[cod]
*.egg-info/
.pytest_cache/
.ruff_cache/
.venv/
venv/
build/
dist/

Step 4: Install the package in dev mode

Run from backend/:

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

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:

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:

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

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:

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:

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

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:

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:

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

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:

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:

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

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

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

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:

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:

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

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:

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:

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

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:

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

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:

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:

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

from cyclone.parsers.writer import write_outputs

And to __all__:

    "write_outputs",

Step 6: Commit

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:

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:

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

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

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:

# 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

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

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

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)

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
Placeholder scan: no TBD/TODO/"fill in later" — all code blocks are complete.
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.
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/:

pytest -v

Expected: all unit tests pass, prodfile smoke test is skipped.

Step 2: Run the CLI against both fixtures

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

git status
git add -A
git commit -m "chore(backend): post-plan cleanups" || true

Step 4: Optional — run the prodfile smoke test

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.

74 KiB Raw Permalink Blame History Unescape Escape

Cyclone 837P Parser Implementation Plan

File map

Task 1: Project skeleton

Task 2: Exceptions module

Task 3: Segment tokenizer

Task 4: Pydantic v2 models

Task 5: PayerConfig factories

Task 6: Minimal 837P fixture

Task 7: Validator

Task 8: Parser orchestrator (basic structure)

Task 9: Multi-claim CO fixture + integration test

Task 10: Output writer

Task 11: CLI

Task 12: README

CLI

Output

Programmatic use

Tests

Spec & plan

Self-review checklist (run after writing the plan)

Final task: verify everything together

74 KiB

Raw Permalink Blame History