Adds three live-tail streaming endpoints that emit an NDJSON snapshot
then forward new event-bus events as they arrive, with a 15s idle
heartbeat (overridable via CYCLONE_TAIL_HEARTBEAT_S for tests).
Each endpoint:
1. yields a snapshot of existing rows as {"type":"item","data":<row>}
2. terminates the snapshot with {"type":"snapshot_end","data":{"count":N}}
3. subscribes to its event kind and forwards each new event as an
{"type":"item","data":<event>} line
4. emits a {"type":"heartbeat","data":{"ts":<iso>}} line every
CYCLONE_TAIL_HEARTBEAT_S seconds when idle
5. checks request.is_disconnected() before each yield and unsubscribes
from the bus on cleanup so a closed stream releases its queue
The shared tail loop lives in api._tail_events, which polls
bus.subscribe_raw()'s queue directly instead of using the bus's
async-iterator wrapper — wait_for on an async generator cancels the
inner future on timeout, which poisons subsequent __anext__ calls with
StopAsyncIteration. Queue.get() is idempotent under cancellation, so
heartbeats don't break the subscription.
EventBus gains an unsubscribe(queue, kinds) method (idempotent) so
the tail loop can release its queue in a try/finally. The disconnect
test asserts the subscriber list is empty after the body iterator is
closed, validating no queue leak per open stream.
Tests in test_api_stream_live.py: 8 tests covering snapshot shape,
post-snapshot publish, heartbeat timing, multi-item snapshots, and
client disconnect cleanup. Plus 2 tests in test_pubsub.py for the
new unsubscribe method.
Cyclone Backend — X12 837P / 835 ERA Parser
Python module + CLI for parsing X12 837P professional claim files and X12 835 ERA (Health Care Claim Payment/Advice) files into one validated JSON per claim / payout. Output is consumed by the Cyclone Vite/React frontend via a FastAPI service (see REST API below).
Install
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}— defaultco_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 parsed2— 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.
REST API
The Cyclone Vite/React frontend (default dev origin http://localhost:5173)
talks to the parser over HTTP. The API module is cyclone.api:app.
Run the server
uvicorn cyclone.api:app --reload --port 8000
# or, equivalently:
python -m cyclone serve
Endpoints
| Method | Path | Purpose |
|---|---|---|
| GET | /api/health |
Liveness probe: {"status": "ok", "version": ...} |
| POST | /api/parse-837 |
Upload an X12 837P file, get parsed claims back |
| POST | /api/parse-835 |
Upload an X12 835 ERA file, get parsed payouts back |
POST /api/parse-837 accepts multipart/form-data with a single file
field. Optional query parameters:
payer—co_medicaid(default) orgeneric_837pinclude_raw_segments—true(default) orfalsestrict—false(default) ortrue(promote warnings to errors)
The response shape depends on the request's Accept header:
Accept: application/json→ a singleParseResult-shaped JSON object (envelope, claims, summary).- No
Acceptheader (or anything that isn'tapplication/json) →application/x-ndjsonstreaming, one JSON object per line in this order:{"type": "envelope", "data": {...}}, one{"type": "claim", "data": {...}}per claim, then{"type": "summary", "data": {...}}.
Error responses are JSON of the form {"error": "...", "detail": "..."}:
400— file-level parse error (CycloneParseError) or unknownpayer422— at least one claim failed validation (body still includes the fullParseResultso the client can render the errors)500— unexpected internal error
CORS is configured to allow http://localhost:5173 with GET and POST
and any headers.
Example: upload + JSON response
curl -X POST http://localhost:8000/api/parse-837 \
-F "file=@./samples/co_837p.txt" \
-H "Accept: application/json"
{
"envelope": { "sender_id": "...", "receiver_id": "...", ... },
"claims": [ { "claim_id": "C1", ... }, { "claim_id": "C2", ... } ],
"summary": { "total_claims": 2, "passed": 2, "failed": 0, ... }
}
Example: stream NDJSON from the browser
const resp = await fetch("http://localhost:8000/api/parse-837", {
method: "POST",
body: formData, // FormData with a "file" entry
});
const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
for (const line of buffer.split("\n")) {
if (!line) continue;
const evt = JSON.parse(line);
if (evt.type === "claim") renderClaim(evt.data);
else if (evt.type === "summary") showSummary(evt.data);
}
buffer = buffer.slice(buffer.lastIndexOf("\n") + 1);
}
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
835 ERA parser
The 835 ERA parser mirrors the 837P architecture but is a separate
module (cyclone.parsers.parse_835) with its own models
(cyclone.parsers.models_835), writer (cyclone.parsers.writer_835),
validator (cyclone.parsers.validator_835), and PayerConfig835 preset.
It parses one 835 batch into:
envelope— ISA / GS / ST control segmentsfinancial_info— BPR (total paid, handling code, payer tax id, date)trace— TRN (reassociation trace number + originating company id)payer— Loop 1000A (NM1*PR + N3/N4/PER contact URL)payee— Loop 1000B (NM1*PE + N3/N4)claims— oneClaimPaymentper Loop 2100 CLP, with embeddedServicePaymentrows for each Loop 2110 SVCsummary— sameBatchSummaryused by 837P
835 CLI
python -m cyclone.cli parse-835 path/to/835.txt --output-dir ./payouts
Options:
--payer {co_medicaid_835,generic_835}— defaultco_medicaid_835--strict— promote warnings to errors--include-raw-segments / --no-raw-segments— default: include--log-level {DEBUG,INFO,WARNING,ERROR}
Output:
payouts/
├── payout-991102984-20260617-CLM001.json # one per claim, with envelope/financial/trace/payer/payee embedded
├── payout-991102984-20260617-CLM002.json
└── summary.json
835 programmatic use
from pathlib import Path
from cyclone.parsers.parse_835 import parse
from cyclone.parsers.payer import PayerConfig835
from cyclone.parsers.validator_835 import validate
from cyclone.parsers.writer_835 import write_outputs_835
result = parse(Path("input.txt").read_text(), PayerConfig835.co_medicaid_835())
report = validate(result, PayerConfig835.co_medicaid_835())
print(f"passed={report.passed} errors={len(report.errors)}")
write_outputs_835(result, Path("./payouts"))
835 payout balancing
The 835 validator enforces two monetary-balancing rules:
R835_BAL_BPR_vs_CLP04— the BPR02 total paid must equal the sum ofCLP04(total claim paid) across all claims in the batch. Tolerance is $0.01.R835_BAL_CLP04_vs_SVC03— eachCLP04must equal the sum of its embeddedSVC03service payment amounts. Tolerance is $0.01.
These rules catch the common "BPR doesn't match the sum of the claims" mismatch (often caused by a missing or duplicated claim in the batch).
835 REST API
POST /api/parse-835 mirrors POST /api/parse-837 exactly:
multipart/form-datawith a singlefilefield- Query params:
payer(defaultco_medicaid_835),include_raw_segments(defaulttrue),strict(defaultfalse) - Content negotiation:
Accept: application/json→ singleParseResult835object; default → NDJSON stream withenvelope → financial_info → trace → payer → payee → claim_payment (×N) → summarylines 400onCycloneParseErroror unknownpayer422whenvalidate()reports errors (body still includes the fullParseResult835so the UI can show the issues)
curl -X POST http://localhost:8000/api/parse-835 \
-F "file=@./samples/co_835.txt" \
-H "Accept: application/json"