cyclone/.superpowers/skills/cyclone-edi/SKILL.md

name, description

name	description
cyclone-edi	Cyclone EDI parser/validator conventions (837P/835/999/270/271/277CA/TA1). Use when: adding or changing a parser, adding a validator rule (R010/R020/R100/R200-R210/R835_*/NPI Luhn/EIN/CAS), or mapping a new CAS adjustment reason code.

cyclone-edi

Cyclone parses seven X12 EDI transaction types (837P, 835, 999, 270, 271, 277CA, TA1) into typed Pydantic models, then runs per-claim / per-batch validator rules that surface as R-coded ValidationIssue records. This skill codifies the conventions so new parsers and rules stay consistent with the seven that already exist.

As of this writing: 7 parser modules under backend/src/cyclone/parsers/parse_<edi>.py, ~25 per-claim rules numbered R010–R100 and R200–R210 in validator.py, plus a parallel set of 835-specific rules prefixed R835_* in validator_835.py. The next increment is SP22.

When to use

• Adding or changing a parser. You are about to touch backend/src/cyclone/parsers/parse_<edi>.py or its paired models_<edi>.py and need the orchestrator signature, the segment walker convention, and the re-export in parsers/__init__.py.
• Adding a validator rule. You are writing a new _rule_R<n>_<name> (or _r<n>_<name> per the existing snake-case style) and need the rule signature, the R-code numbering scheme, and the ValidationIssue shape.
• Wiring a new CAS / CARC code. The 835 carries Claim Adjustment Reason Codes in CAS segments; the lookup lives in backend/src/cyclone/parsers/cas_codes.py and the UI reads through claim_status_label().
• Debugging a parse failure on a prodfiles sample. You dropped a real EDI file into docs/prodfiles/<source>/ and the parser is choking — load this skill to confirm the tokenizer path, the orchestrator entry point, and which fixture in backend/tests/fixtures/ matches the transaction type.

Conventions

1. Parser signature. Every parser module exports exactly one public entry function. Two flavors coexist in the codebase:
   • parse(text: str, *, input_file: str = "") -> <TypedResult> — used by parse_270.py:337, parse_271.py:356.
   • parse(text: str, payer_config: <PayerConfig>, input_file: str = "") -> <TypedResult> — used by parse_837.py:319 and parse_835.py:459 because both need payer-specific config to validate segments against.
   • parse_<edi>_text(text: str, *, input_file: str = "") -> <TypedResult> — the legacy name-suffixed form, still in use at parse_ta1.py:143, parse_999.py:220, parse_277ca.py:280. The <TypedResult> is always a Pydantic model from models_<edi>.py (or co-located models.py for 837P).
2. Segment walk. Parsers consume backend/src/cyclone/parsers/segments.py — there are exactly three public pieces: Delimiters (frozen dataclass holding the four ISA-derived separators), _detect_delimiters(isa_segment) (private), and tokenize(text) -> list[list[str]] (returns ISA prepended as the first segment). Parsers then index into the list[list[str]] directly — there is no Segment / Loop / next_segment helper class. Whole-document problems (missing ISA, wrong transaction set) raise CycloneParseError; per-segment problems on acks (999/277CA) are surfaced on the result, not raised.
3. Validator rules. Numbered rules live in backend/src/cyclone/parsers/validator.py (for 837P — R010–R100 general + R200–R210 SP9 CO MAP / HCPF naming) and backend/src/cyclone/parsers/validator_835.py (for 835 — names prefixed R835_* because the same numeric space would collide with 837P). Each rule is a function _r<n>_<name>(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue] registered in the module-level _RULES list and run by validate(claim, config). Issues carry the rule name as a stable string (rule="R021_npi_checksum") — the R-code is how the UI surfaces the error, so never invent an unnumbered rule.
4. NPI / EIN / CAS format logic. Identity-format checks live in their own modules — never duplicate them in a parser or validator:
   • backend/src/cyclone/npi.py — is_valid_npi(npi) runs the Luhn checksum with the 80840 NPPES prefix; is_valid_tax_id(ein) enforces XX-XXXXXXX (or 9 raw digits).
   • backend/src/cyclone/parsers/cas_codes.py — reason_label(group, reason) and all_known_codes() for the CARC lookup; snapshot date is exported as LAST_UPDATED.
   • backend/src/cyclone/parsers/models_271.py — SERVICE_TYPE_CODES + service_type_description() for 271 EB benefit codes.
5. Prodfiles reuse. When adding a parser for a new transaction type, ship at least one fixture in backend/tests/fixtures/<edi>/<sample>.txt (the existing 13 fixtures are flat at the top level of fixtures/ — no per-test subdirectories). Copy from docs/prodfiles/<source>/<file>.txt; never reach into docs/prodfiles/ from a test. The matching test should declare the path as a module-level Path constant.

Patterns

Minimal parse_<edi>.py — using segments.py, exporting parse_ta1_text

Taken from backend/src/cyclone/parsers/parse_ta1.py:1-29 (the smallest parser — TA1 is just ISA + TA1 + IEA). The same skeleton scales to every other EDI type by adding _consume_<segment> helpers.

"""Parse an X12 TA1 (Interchange Acknowledgment) file.

Whole-document problems (missing ISA, no TA1) raise CycloneParseError.
"""
from __future__ import annotations

import logging
from datetime import date

from cyclone.parsers.exceptions import CycloneParseError
from cyclone.parsers.models import BatchSummary, Envelope
from cyclone.parsers.models_ta1 import ParseResultTa1, Ta1Ack
from cyclone.parsers.segments import tokenize

log = logging.getLogger(__name__)


def _parse_yyyymmdd(s: str) -> date | None:
    """Parse an 8-digit CCYYMMDD string. Returns None on bad input."""
    ...


def _build_envelope(segments: list[list[str]], input_file: str) -> Envelope:
    """Build the envelope from ISA. TA1 has no GS/ST — just ISA → TA1 → IEA."""
    ...


def _consume_ta1(segments: list[list[str]], idx: int) -> tuple[Ta1Ack, int]:
    """Read a TA1 segment and return a Ta1Ack. Returns (model, next_idx)."""
    ...


def parse_ta1_text(text: str, *, input_file: str = "") -> ParseResultTa1:
    """Parse a complete TA1 document and return a ParseResultTa1."""
    segments = tokenize(text)
    envelope = _build_envelope(segments, input_file=input_file)
    ta1_idx = next(
        (i for i, seg in enumerate(segments) if seg[0] == "TA1"), None,
    )
    if ta1_idx is None:
        raise CycloneParseError("No TA1 segment found")
    ta1, _ = _consume_ta1(segments, ta1_idx)
    ...
    return ParseResultTa1(envelope=envelope, ta1=ta1, summary=summary, ...)


__all__ = ["parse_ta1_text"]

The orchestrator pattern is the same in every parser: tokenize → _build_envelope → segment consumers in order → wrap into a ParseResult<EDI> model. The Pydantic result is what the API / store layer consumes.

A validator rule — _r<n>_<name> registered in _RULES

Taken from backend/src/cyclone/parsers/validator.py:23-66 (the canonical R010–R100 block).

from collections.abc import Iterable
from cyclone.parsers.models import ClaimOutput, ValidationIssue
from cyclone.parsers.payer import PayerConfig

NPI_RE = re.compile(r"^\d{10}$")
Rule = Callable[[ClaimOutput, PayerConfig], Iterable[ValidationIssue]]


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 _r021_npi_checksum(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]:
    """SP20: validate the billing-provider NPI's Luhn check digit."""
    npi = claim.billing_provider.npi
    if not npi or not NPI_RE.match(npi):
        return  # R020 already flagged the format — skip silently.
    try:
        from cyclone.npi import is_valid_npi
    except ImportError:
        return
    if not is_valid_npi(npi):
        yield ValidationIssue(
            rule="R021_npi_checksum",
            severity="warning",
            message=f"Billing provider NPI {npi!r} fails Luhn checksum (likely typo)",
        )


_RULES: list[Rule] = [
    _r010_clm01_present,
    _r011_total_charge_positive,
    _r020_npi_format,
    _r021_npi_checksum,
    # ... R030, R031, R032-R035, R050, R060, R070, R100, R200-R210
]

For 835 rules, prefix the rule string with R835_ (e.g. R835_BPR01_handling_code_allowed) and target the ParseResult835 model instead of ClaimOutput — see backend/src/cyclone/parsers/validator_835.py:38-79.

A test that uses a prodfiles fixture

Taken from backend/tests/test_api_999.py:53-72. The autouse conftest.py already provides a per-test SQLite DB; most tests just add a client fixture and reference the fixture path as a module-level constant.

"""Tests for the FastAPI surface in cyclone.api for the 999 endpoint."""
from __future__ import annotations

from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from cyclone.api import app

# Fixture reference — flat, module-level Path constant. NEVER reach into
# docs/prodfiles/ from a test; the fixtures/ dir is the test-consumed surface.
ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_999.txt"
REJECTED = Path(__file__).parent / "fixtures" / "minimal_999_rejected.txt"


@pytest.fixture
def client() -> TestClient:
    return TestClient(app)


def test_parse_999_endpoint_happy_path(client: TestClient):
    text = ACCEPTED.read_text()
    resp = client.post(
        "/api/parse-999",
        files={"file": ("minimal_999.txt", text, "text/plain")},
        headers={"Accept": "application/json"},
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["ack"]["ack_code"] == "A"

For pure-unit parser tests (no API), the same path is reused — see backend/tests/test_parse_837.py:8 (FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt").

Anti-patterns

• Don't re-parse raw X12 strings inside validators. Always parse first into the typed ParseResult<EDI> / ClaimOutput, then validate against that. Validators index into the model fields (or claim.raw_segments for spot-checks of specific segment presence) — they never call tokenize again. R034's REF*G1 presence check (validator.py:104-107) is the only place that legitimately touches raw_segments, and it does so to confirm a single segment exists.
• Don't bake payer-specific logic into the generic parser. Payer variations live in backend/src/cyclone/parsers/payer.py (PayerConfig, PayerConfig835) and backend/src/cyclone/payers.py (the YAML loader from config/payers.yaml). Parsers accept the config as an argument; rules read it from the cfg parameter. A new payer never requires a new parser file — extend the config and add / adjust an R-code rule.
• Don't add a validator rule without an R-code. The rule="R<n>_<name>" string is the stable identifier the UI greys out, the API returns in errors[].rule, and tests assert against. Inventing a rule without an R-code (or reusing an R-code with new semantics) breaks the operator workflow. New SP-N increments reserve their R-code range up front (SP9 reserved R200–R210, SP20 added R021) and document it in the spec.

Related skills

• cyclone-store — load when the increment changes how a parsed ClaimOutput / ParseResult<EDI> is persisted (store.py write path, <entity>_written events).
• cyclone-api-router — load when the increment adds or changes an HTTP endpoint that surfaces a parsed result (e.g. /api/parse-999, /api/parse-837, /api/parse-835).
• cyclone-tests — every parser addition ships a fixture in backend/tests/fixtures/ and a pytest case; load this skill for the fixture-drop-in and autouse-conftest rules.
• cyclone-cli — load when the increment adds a CLI subcommand. The cyclone parse-837 <file> and cyclone parse-835 <file> smoke commands at backend/src/cyclone/cli.py:77,151 are the parser-level smoke tests; the validate-npi and validate-tax-id commands exercise the format helpers.
• cyclone-spec — load when the SP-N spec for the increment introduces a new R-code range or a new transaction type; the spec's ## Decisions section is where the R-code reservation gets locked in.