feat(sp25): cyclone recover-ingest CLI for stranded source files

backend/src/cyclone/cli.py

@@ -1376,5 +1376,88 @@ def ack_orphans_reconcile_cmd(dry_run: bool) -> None:
         click.echo("(dry-run — no rows inserted)")
 
 
+# ---------------------------------------------------------------------------
+# SP25: ``cyclone recover-ingest`` — offline parse + DB-write helper for
+# stranded source files in ingest/. Mirrors the data-writing half of
+# ``submit-batch`` but does NOT SFTP-upload (the files were already
+# uploaded by whatever path got them into ingest/) and does NOT write
+# to audit_log (the submit event is already in history).
+#
+# Supports 837P and 835. 999/TA1/277CA ingestion lives on
+# ``cyclone pull-inbound``; this command is for the larger source files
+# that need a typed BatchRecord + the canonical CycloneStore.add path so
+# live-tail pages light up.
+#
+# Exit codes (mirrors submit-batch):
+#   0  — at least one file ingested OK (or all were duplicates).
+#   2  — a file failed to parse or DB-write.
+#   1  — unexpected exception (DB unreachable, etc.).
+# ---------------------------------------------------------------------------
+
+
+@main.command("recover-ingest")
+@click.option(
+    "--file", "files", multiple=True, required=True,
+    type=click.Path(exists=True, dir_okay=False, path_type=Path),
+    help="One or more X12 files to recover. Repeatable.",
+)
+@click.option(
+    "--sftp-block-name", "sftp_block_name",
+    default="manual-recover", show_default=True,
+    help="Dedup key in processed_inbound_files. Defaults to "
+         "'manual-recover' so it doesn't collide with the real "
+         "dzinesco block.",
+)
+@click.option(
+    "--actor", default="recover-ingest", show_default=True,
+    help="Audit actor tag (reserved — recovery doesn't audit by default).",
+)
+def recover_ingest_cmd(
+    files: tuple[Path, ...],
+    sftp_block_name: str,
+    actor: str,  # noqa: ARG001 — reserved
+) -> None:
+    """Parse X12 files from a local path and write them to the DB.
+
+    Detects the kind (837P / 835) from the first ST segment; supports
+    only those two. Dedupes by ``(sftp_block_name, filename)`` — a
+    second invocation against the same file is a no-op.
+
+    Use this when an 837P / 835 file landed in ``ingest/`` but the
+    cyclic ``parse-837`` / ``parse-835`` CLI only emits JSON (it does
+    NOT write to the DB) and ``submit-batch`` was never called.
+    """
+    from cyclone.submission.recover import recover_file
+
+    click.echo(f"Recovering {len(files)} file(s) under sftp_block={sftp_block_name!r}")
+    any_ok = False
+    failed = 0
+    for f in files:
+        result = recover_file(f, sftp_block_name=sftp_block_name)
+        if result["status"] == "ok":
+            click.echo(
+                f"  ok      {f}  kind={result['kind']}  "
+                f"batch_id={result['batch_id']}"
+            )
+            any_ok = True
+        elif result["status"] == "duplicate":
+            click.echo(f"  skipped {f}  (already processed)")
+        else:  # failed
+            failed += 1
+            click.echo(
+                f"  FAILED  {f}  error={result['error']}", err=True,
+            )
+
+    if failed and not any_ok:
+        click.echo(f"\n{failed} file(s) failed; nothing ingested.", err=True)
+        sys.exit(2)
+    if failed:
+        # Mixed bag — some ok some failed.
+        click.echo(f"\n{failed} file(s) failed; the rest ingested.", err=True)
+        sys.exit(2)
+    if any_ok:
+        click.echo("\nDone.")
+
+
 if __name__ == "__main__":
     main()

backend/src/cyclone/submission/recover.py

@@ -0,0 +1,247 @@
+"""SP25 Task 5: ``cyclone recover-ingest`` parse + DB-write helper.
+
+What this module is for
+------------------------
+On 2026-07-07 the dashboard was empty even though four 837P files
+and one 835 file were sitting in ``ingest/``. The 837Ps had been
+SFTP-shipped to Gainwell (the operator's SFTP client, manually),
+but cyclone never recorded them in the DB — ``parse-837`` only
+emits JSON, and ``submit-batch`` was never called for that run.
+The 835 came back from the payer and was dropped into ``ingest/``,
+but ``pull-inbound`` was never run for it.
+
+This helper is the recovery path: parse the local file, build a
+typed :class:`BatchRecord`, call :meth:`CycloneStore.add` (which
+publishes events to the live-tail bus), record the file in
+``processed_inbound_files`` for dedup, and stop — no SFTP, no audit
+log because the file was already uploaded (or already came back).
+
+Public surface
+--------------
+- :func:`recover_file` — one file. Returns a structured result dict.
+- :func:`detect_kind` — ST01 helper, exposed for tests.
+
+Why this is its own module (not a new branch in submit_file)
+------------------------------------------------------------
+``submit_file`` is the SFTP-write path; its invariants are
+"DB write → SFTP upload → audit event" and removing the SFTP half
+would mutate that contract for every caller. Recovery is a
+deliberately offline path and a separate surface keeps the two
+flows readable. ``recover_file`` shares the same parse + store
+write code path (``CycloneStore.add``) so the live-tail pages
+light up exactly the way they would for a fresh submission.
+"""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from cyclone import db as db_mod
+from cyclone.store import store as cycl_store
+from cyclone.db import ProcessedInboundFile
+from cyclone.parsers.parse_837 import parse as parse_837_text
+from cyclone.parsers.parse_835 import parse as parse_835_text
+from cyclone.parsers.payer import PayerConfig, PayerConfig835
+from cyclone.parsers.segments import tokenize as _tokenize_segments
+from cyclone.store.records import BatchRecord835, BatchRecord837
+
+log = logging.getLogger(__name__)
+
+# Default dedup key for the sftp_block_name column. Distinct from
+# ``dzinesco`` so a real remote-pull can't accidentally mask a
+# manual recovery (or vice versa).
+DEFAULT_SFTP_BLOCK_NAME = "manual-recover"
+
+
+def detect_kind(text: str) -> str:
+    """Return the ST01 of the *first* ST segment, or ``"unknown"``.
+
+    X12 envelopes can contain multiple transactions; the first ST is
+    the canonical kind for the whole file under our parsing rules.
+
+    Mirrors the SP35 envelope check in ``api_routers/parse.py``
+    (auto-detect layer A; this is layer B for offline recovery).
+    """
+    try:
+        segments = _tokenize_segments(text)
+    except Exception:  # noqa: BLE001 — recover-context, log only
+        log.warning("recover-ingest: tokenize failed", exc_info=True)
+        return "unknown"
+    for seg in segments:
+        if seg and seg[0] == "ST" and len(seg) > 1:
+            return seg[1]
+    return "unknown"
+
+
+def _normalize_kind(st01: str) -> str:
+    """Map ``"837"`` → ``"837p"`` (parser-level kind); pass others through.
+
+    The parser distinguishes ``837p`` (professional) from the
+    837I/D variants; only 837P is supported in the recovery path
+    today. Anything else is rejected with a clear error message.
+    """
+    if st01 == "837":
+        return "837p"
+    if st01 in {"835", "837p"}:
+        return st01
+    return st01
+
+
+def _already_processed(name: str, sftp_block_name: str) -> bool:
+    """Return True if ``(sftp_block_name, name)`` was already recovered.
+
+    Mirrors the dedup index ``ux_processed_inbound_files_block_name``
+    (UNIQUE on sftp_block_name, name).
+    """
+    s = db_mod.SessionLocal()()
+    try:
+        return s.query(ProcessedInboundFile).filter_by(
+            sftp_block_name=sftp_block_name, name=name
+        ).first() is not None
+    finally:
+        s.close()
+
+
+def _mark_processed(
+    *,
+    sftp_block_name: str,
+    path: Path,
+    file_type: str,
+    parser_used: str,
+    claim_count: int,
+) -> None:
+    """Insert a ``processed_inbound_files`` row for the recovered file.
+
+    Status hardcoded to ``"ok"`` because by the time we reach this
+    function the parse + store add have both succeeded.
+    """
+    stat = path.stat()
+    now = datetime.now(tz=timezone.utc)
+    s = db_mod.SessionLocal()()
+    try:
+        s.add(ProcessedInboundFile(
+            sftp_block_name=sftp_block_name,
+            name=path.name,
+            size=stat.st_size,
+            modified_at=datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc),
+            file_type=file_type,
+            processed_at=now,
+            parser_used=parser_used,
+            claim_count=claim_count,
+            status="ok",
+            error_message=None,
+        ))
+        s.commit()
+    finally:
+        s.close()
+
+
+def recover_file(
+    path: str | Path,
+    *,
+    sftp_block_name: str = DEFAULT_SFTP_BLOCK_NAME,
+    actor: str = "recover-ingest",
+) -> dict:
+    """Recover one X12 file from a local path: parse → DB write (no SFTP).
+
+    Args:
+        path: local file. 837P / 835 only.
+        sftp_block_name: dedup key in ``processed_inbound_files``.
+            Defaults to ``"manual-recover"`` so the recovery doesn't
+            collide with the real ``dzinesco`` SFTP block.
+        actor: reserved for audit. Recovery doesn't audit-log per
+            design (the file was already uploaded by whatever path
+            got it into ingest/), but the parameter is kept so a
+            future audit-enabled variant doesn't break callers.
+
+    Returns:
+        dict with keys:
+          - ``file``: path as passed in (str)
+          - ``kind``: ``"837p"`` / ``"835"`` / ``None``
+          - ``status``: ``"ok"`` / ``"duplicate"`` / ``"failed"``
+          - ``batch_id``: hex uuid4 when status=="ok", else None
+          - ``error``: str when status=="failed", else None
+    """
+    p = Path(path)
+    if not p.exists():
+        return {"file": str(p), "kind": None, "status": "failed",
+                "batch_id": None, "error": f"file does not exist: {p}"}
+    if not p.is_file():
+        return {"file": str(p), "kind": None, "status": "failed",
+                "batch_id": None, "error": f"not a regular file: {p}"}
+
+    # 1. Dedup — short-circuit if this exact name was already recovered.
+    if _already_processed(p.name, sftp_block_name):
+        return {"file": str(p), "kind": None, "status": "duplicate",
+                "batch_id": None, "error": None}
+
+    # 2. Read + detect kind.
+    try:
+        text = p.read_text(encoding="utf-8")
+    except UnicodeDecodeError as exc:
+        return {"file": str(p), "kind": None, "status": "failed",
+                "batch_id": None, "error": f"encoding: {exc!r}"}
+
+    raw_kind = detect_kind(text)
+    kind = _normalize_kind(raw_kind)
+    if kind not in {"837p", "835"}:
+        return {"file": str(p), "kind": raw_kind, "status": "failed",
+                "batch_id": None,
+                "error": f"unsupported kind {raw_kind!r} (recovery supports 837p + 835)"}
+
+    # 3. Parse through the canonical parser for the detected kind.
+    parsed: Any = None
+    if kind == "837p":
+        try:
+            parsed = parse_837_text(text, PayerConfig.co_medicaid(), input_file=p.name)
+        except Exception as exc:  # noqa: BLE001
+            return {"file": str(p), "kind": kind, "status": "failed",
+                    "batch_id": None, "error": f"parse_837: {exc!r}"}
+    else:  # "835"
+        try:
+            parsed = parse_835_text(text, PayerConfig835.co_medicaid_835(), input_file=p.name)
+        except Exception as exc:  # noqa: BLE001
+            return {"file": str(p), "kind": kind, "status": "failed",
+                    "batch_id": None, "error": f"parse_835: {exc!r}"}
+
+    # 4. Build a typed BatchRecord (mirrors api.py parse-837 happy path).
+    batch_id = uuid.uuid4().hex
+    now = datetime.now(tz=timezone.utc)
+    if kind == "837p":
+        record = BatchRecord837(
+            id=batch_id, input_filename=p.name, parsed_at=now, result=parsed,
+        )
+    else:
+        record = BatchRecord835(
+            id=batch_id, input_filename=p.name, parsed_at=now, result=parsed,
+        )
+
+    # 5. DB write — canonical path. CycloneStore.add publishes events on
+    # the bundled bus so live-tail subscribers see new claims/remits.
+    try:
+        cycl_store.add(record)
+    except Exception as exc:  # noqa: BLE001
+        return {"file": str(p), "kind": kind, "status": "failed",
+                "batch_id": None, "error": f"db_write: {exc!r}"}
+
+    # 6. Record success for dedup.
+    claim_count = len(getattr(parsed, "claims", []) or [])
+    try:
+        _mark_processed(
+            sftp_block_name=sftp_block_name,
+            path=p,
+            file_type=kind,
+            parser_used=f"parse_{kind}",
+            claim_count=claim_count,
+        )
+    except Exception as exc:  # noqa: BLE001
+        # The DB write succeeded but dedup-record failed — log and
+        # still report ok (re-running will hit dedup once it's
+        # recorded; until then a re-run is a real duplicate insert).
+        log.warning("recover-ingest: dedup-write failed for %s: %r", p, exc)
+
+    return {"file": str(p), "kind": kind, "status": "ok",
+            "batch_id": batch_id, "error": None}

backend/tests/test_cli_recover_ingest.py

@@ -0,0 +1,161 @@
+"""SP25: integration tests for ``cyclone recover-ingest`` (parse + DB-write recovery).
+
+Fixtures used (all real production EDI from the test fixture corpus):
+  - ``co_medicaid_837p.txt`` — 1-claim 837P fixture (small, fast).
+  - ``co_medicaid_835.txt`` — minimal 835 fixture (real CO Medicaid).
+
+These tests do NOT cover the operator's full ingest/ batch — that's
+Task 6 in the plan and runs against the live DB after these unit
+tests are green.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+
+from cyclone import db, store
+from cyclone.cli import main as cli_main
+from cyclone.db import Batch, Claim, ProcessedInboundFile, Remittance
+
+FIX_837P = Path(__file__).parent / "fixtures" / "co_medicaid_837p.txt"
+FIX_835 = Path(__file__).parent / "fixtures" / "co_medicaid_835.txt"
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+# --- helpers --------------------------------------------------------------
+
+def _count_claims() -> int:
+    s = db.SessionLocal()()
+    try:
+        return s.query(Claim).count()
+    finally:
+        s.close()
+
+
+def _count_remittances() -> int:
+    s = db.SessionLocal()()
+    try:
+        return s.query(Remittance).count()
+    finally:
+        s.close()
+
+
+def _count_batches() -> int:
+    s = db.SessionLocal()()
+    try:
+        return s.query(Batch).count()
+    finally:
+        s.close()
+
+
+# --- tests ----------------------------------------------------------------
+
+def test_recover_ingest_837p_persists_claims(runner):
+    """One 837P ingest via the CLI lands a Batch + ≥1 Claim row + a dedup row."""
+    pre_claims = _count_claims()
+    pre_batches = _count_batches()
+    result = runner.invoke(
+        cli_main,
+        ["recover-ingest", "--file", str(FIX_837P),
+         "--sftp-block-name", "test-recover-837p"],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0, result.stderr or result.stdout
+    out = (result.stdout or "") + (result.stderr or "")
+    assert "ok" in out
+    assert FIX_837P.name in out
+
+    assert _count_claims() > pre_claims, "expected new Claim rows"
+    assert _count_batches() > pre_batches, "expected new Batch row"
+
+    # Dedup row created.
+    s = db.SessionLocal()()
+    try:
+        row = s.query(ProcessedInboundFile).filter_by(
+            sftp_block_name="test-recover-837p", name=FIX_837P.name,
+        ).first()
+    finally:
+        s.close()
+    assert row is not None, "expected processed_inbound_files row"
+    assert row.status == "ok"
+    assert row.file_type == "837p"
+
+
+def test_recover_ingest_835_persists_remittance(runner):
+    """One 835 ingest via the CLI lands a Batch + Remittance row + dedup row."""
+    pre_batches = _count_batches()
+    pre_remits = _count_remittances()
+    result = runner.invoke(
+        cli_main,
+        ["recover-ingest", "--file", str(FIX_835),
+         "--sftp-block-name", "test-recover-835"],
+        catch_exceptions=False,
+    )
+    assert result.exit_code == 0, result.stderr or result.stdout
+    assert "ok" in (result.stdout or "") + (result.stderr or "")
+
+    assert _count_batches() > pre_batches
+    assert _count_remittances() > pre_remits
+
+    s = db.SessionLocal()()
+    try:
+        row = s.query(ProcessedInboundFile).filter_by(
+            sftp_block_name="test-recover-835", name=FIX_835.name,
+        ).first()
+    finally:
+        s.close()
+    assert row is not None
+    assert row.status == "ok"
+    assert row.file_type == "835"
+    assert row.claim_count >= 1
+
+
+def test_recover_ingest_is_idempotent(runner):
+    """Second invocation with the same (block, file) is a no-op (duplicate)."""
+    # First call — must ingest.
+    r1 = runner.invoke(
+        cli_main,
+        ["recover-ingest", "--file", str(FIX_837P),
+         "--sftp-block-name", "test-recover-idem"],
+        catch_exceptions=False,
+    )
+    assert r1.exit_code == 0, r1.stderr or r1.stdout
+
+    pre_claims = _count_claims()
+    pre_batches = _count_batches()
+    # Second call — must skip.
+    r2 = runner.invoke(
+        cli_main,
+        ["recover-ingest", "--file", str(FIX_837P),
+         "--sftp-block-name", "test-recover-idem"],
+        catch_exceptions=False,
+    )
+    assert r2.exit_code == 0, r2.stderr or r2.stdout
+    assert "skipped" in (r2.stdout or "") + (r2.stderr or "")
+
+    # No new rows on second call.
+    assert _count_claims() == pre_claims
+    assert _count_batches() == pre_batches
+
+
+def test_recover_ingest_missing_file_returns_failed(runner, tmp_path):
+    """Missing file → Click rejects with usage error (the CLI uses exists=True)."""
+    missing = tmp_path / "does-not-exist.837"
+    pre_claims = _count_claims()
+    result = runner.invoke(
+        cli_main,
+        ["recover-ingest", "--file", str(missing),
+         "--sftp-block-name", "test-recover-missing"],
+        catch_exceptions=False,
+    )
+    # Click rejects up-front; usage error exits with code 2.
+    assert result.exit_code == 2
+    out = (result.stdout or "") + (result.stderr or "")
+    assert "does not exist" in out or "Invalid value" in out
+    assert _count_claims() == pre_claims