feat(sp41): POST/GET /api/admin/rebill-from-835 (auth-gated)

backend/src/cyclone/api_routers/__init__.py

@@ -26,6 +26,7 @@ from cyclone.api_routers import (
     parse,
     payers,
     providers,
+    rebill,
     reconciliation,
     remittances,
     submission,
@@ -48,6 +49,7 @@ routers: list[APIRouter] = [
     parse.router,       # gated
     payers.router,      # gated
     providers.router,   # gated
+    rebill.router,      # gated (SP41)
     reconciliation.router, # gated
     remittances.router, # gated
     submission.router,  # gated

backend/src/cyclone/api_routers/rebill.py

@@ -0,0 +1,222 @@
+"""SP41 — rebill admin endpoints.
+
+POST /api/admin/rebill-from-835
+  body: {"window": "YYYY-MM-DD..YYYY-MM-DD",
+         "override_filing": bool,
+         "visits_csv_path": str (optional),
+         "ingest_dir": str (optional),
+         "out_dir": str (optional)}
+  Returns: {"summary_path": str, "counts": {...},
+            "pipeline_a_files": [...], "pipeline_b_files": [...]}
+
+GET /api/admin/rebill-from-835/status
+  Returns: {"recent_runs": [{"as_of": ..., "summary_path": ...,
+                              "counts": {...}}, ...]}
+
+Status-code contract (per the cyclone-api-router / cyclone-cli skills):
+  - 200: completed run (POST) or tally returned (GET).
+  - 401: not authenticated (matrix_gate).
+  - 403: authenticated but not authorized for /api/admin/*.
+  - 422: window is malformed or Pydantic body validation failed.
+
+The POST handler delegates to ``cyclone.rebill.run.run_rebill`` (the same
+orchestrator the ``cyclone rebill-from-835`` CLI uses). The GET handler
+is a filesystem scan under ``dev/rebills/*/summary.csv`` — no DB table
+for "recent runs" exists today, and Task 12 deliberately doesn't add
+one (per its design notes). Sorted by directory mtime descending;
+truncated to the most recent 5.
+"""
+from __future__ import annotations
+
+import csv
+import logging
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from cyclone.auth.deps import matrix_gate
+
+log = logging.getLogger(__name__)
+
+router = APIRouter(
+    prefix="/api/admin/rebill-from-835",
+    tags=["rebill"],
+    dependencies=[Depends(matrix_gate)],
+)
+
+
+# Filesystem base for /status scans. Module-level so tests can monkeypatch
+# it to a tmp_path-relative dir without chdir-ing the whole test process.
+REBILLS_DIR: Path = Path("dev/rebills")
+
+# How many recent runs /status surfaces. 5 matches the CLI's default; the
+# body schema below mirrors it as a query parameter so callers can ask for
+# more (or fewer) when they want.
+DEFAULT_RECENT_LIMIT = 5
+
+
+class RebillRequest(BaseModel):
+    """Body schema for ``POST /api/admin/rebill-from-835``.
+
+    ``window`` is parsed as ``YYYY-MM-DD..YYYY-MM-DD`` (inclusive both
+    ends); the validator rejects anything else with a 422. The other
+    path fields are optional — ``run_rebill`` falls back to its own
+    defaults (``data/source/apr-jun27.csv`` for visits, ``ingest/`` for
+    835s, ``dev/rebills/<today>/`` for output) when unset.
+    """
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    window: str = Field(
+        default="2026-01-01..2026-06-27",
+        description="DOS window as YYYY-MM-DD..YYYY-MM-DD (inclusive both ends).",
+    )
+    override_filing: bool = Field(
+        default=False,
+        description="Relax the 120-day timely-filing gate for past-window visits.",
+    )
+    visits_csv_path: str | None = Field(
+        default=None,
+        description="Path to the AxisCare visits CSV; defaults to "
+                    "data/source/apr-jun27.csv when unset.",
+    )
+    ingest_dir: str | None = Field(
+        default=None,
+        description="Directory containing *.835 / *.x12 835 files; defaults to ./ingest.",
+    )
+    out_dir: str | None = Field(
+        default=None,
+        description="Output directory for summary.csv + pipeline-a/b files; "
+                    "defaults to dev/rebills/<today>/.",
+    )
+
+    @field_validator("window")
+    @classmethod
+    def _validate_window(cls, v: str) -> str:
+        # Manual split (Click can't help here). Two-date range is the
+        # only accepted shape; ``..`` is the giveaway separator so the
+        # input is unambiguous.
+        try:
+            start_str, end_str = v.split("..", 1)
+        except ValueError as exc:
+            raise ValueError(
+                f"window must be 'YYYY-MM-DD..YYYY-MM-DD', got {v!r}"
+            ) from exc
+        try:
+            start = date.fromisoformat(start_str)
+            end = date.fromisoformat(end_str)
+        except ValueError as exc:
+            raise ValueError(
+                f"window must be 'YYYY-MM-DD..YYYY-MM-DD', got {v!r}: {exc}"
+            ) from exc
+        if start > end:
+            raise ValueError(
+                f"window start {start.isoformat()} is after end {end.isoformat()}"
+            )
+        return v
+
+
+@router.post("")
+def post_rebill(req: RebillRequest) -> dict[str, Any]:
+    """Run the rebill pipeline for the given DOS window.
+
+    Delegates to :func:`cyclone.rebill.run.run_rebill` and returns the
+    resulting ``RunResult`` as a plain JSON dict (the underlying dataclass
+    has no ``to_dict`` method — fields are mapped here so callers don't
+    have to import the dataclass shape).
+
+    The handler does NOT swallow exceptions; the app-level
+    :func:`_unhandled_exception_handler` renders them as a 500 JSON
+    envelope with CORS headers. Per-file failures land in the summary
+    CSV (and the ``counts`` dict) rather than raising here.
+    """
+    from cyclone.rebill.run import run_rebill  # local import — keeps --help fast
+
+    start_str, end_str = req.window.split("..", 1)
+    # Validator already proved these parse cleanly.
+    window_start = date.fromisoformat(start_str)
+    window_end = date.fromisoformat(end_str)
+
+    log.info(
+        "rebill-from-835 starting: window=%s override_filing=%s",
+        req.window, req.override_filing,
+    )
+    result = run_rebill(
+        window_start=window_start,
+        window_end=window_end,
+        override_filing=req.override_filing,
+        visits_csv_path=req.visits_csv_path,
+        ingest_dir=req.ingest_dir,
+        out_dir=req.out_dir,
+    )
+    log.info(
+        "rebill-from-835 done: summary=%s counts=%s",
+        result.summary_path, result.counts,
+    )
+
+    return {
+        "summary_path": str(result.summary_path),
+        "counts": result.counts,
+        "pipeline_a_files": [str(p) for p in result.pipeline_a_files],
+        "pipeline_b_files": [str(p) for p in result.pipeline_b_files],
+    }
+
+
+@router.get("/status")
+def get_rebill_status(
+    limit: int = DEFAULT_RECENT_LIMIT,
+) -> dict[str, Any]:
+    """Return the most recent rebill runs, newest first.
+
+    Scans :data:`REBILLS_DIR` (``dev/rebills/`` by default) for any
+    subdirectory whose name is a date in YYYY-MM-DD format and that
+    contains a ``summary.csv``. Sorted by directory mtime descending so
+    the most recently-run batch wins ties on equal-named dirs (rare in
+    practice — operators tend to pick a fresh date per run).
+
+    Each entry's ``counts`` dict is a tally of the summary.csv's
+    ``disposition`` column (the same per-category counters the operator
+    sees on the CLI ``--status`` view). Missing disposition values
+    surface as ``UNKNOWN`` so a hand-edited CSV can't silently drop a
+    bucket.
+
+    ``limit`` defaults to :data:`DEFAULT_RECENT_LIMIT` (5) and is
+    clamped to ``[1, 50]`` to keep the response bounded.
+    """
+    if limit < 1 or limit > 50:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="limit must be between 1 and 50",
+        )
+
+    recent: list[dict[str, Any]] = []
+    rebills_dir = REBILLS_DIR
+    if rebills_dir.exists():
+        candidates = sorted(
+            (
+                d for d in rebills_dir.iterdir()
+                if d.is_dir()
+                and (d / "summary.csv").exists()
+                # Filter to date-named dirs (YYYY-MM-DD) so a stray
+                # ``lost+found`` or tempdir doesn't sneak in.
+                and len(d.name) == 10 and d.name[4] == "-" and d.name[7] == "-"
+            ),
+            key=lambda d: d.stat().st_mtime,
+            reverse=True,
+        )[:limit]
+        for d in candidates:
+            summary_path = d / "summary.csv"
+            counts: dict[str, int] = {}
+            with summary_path.open(newline="") as f:
+                for row in csv.DictReader(f):
+                    disp = (row.get("disposition") or "UNKNOWN").strip() or "UNKNOWN"
+                    counts[disp] = counts.get(disp, 0) + 1
+            recent.append({
+                "as_of": d.name,
+                "summary_path": str(summary_path),
+                "counts": counts,
+            })
+    return {"recent_runs": recent}

backend/src/cyclone/auth/permissions.py

@@ -69,6 +69,8 @@ PERMISSIONS: dict[tuple[str, str], set[Role]] = {
     ("POST", "/api/admin/reload-config"): ADMIN_ONLY,
     ("GET", "/api/admin/validate-provider"): ADMIN_ONLY,
     ("POST", "/api/admin/validate-837"): ADMIN_ONLY,  # SP40: Edifabric validation probe
+    ("POST", "/api/admin/rebill-from-835"): ADMIN_ONLY,  # SP41: run the in-window rebill pipeline
+    ("GET", "/api/admin/rebill-from-835"): ADMIN_ONLY,   # SP41: covers /status (prefix match)
 
     # Write endpoints (admin + user, no viewer).
     ("POST", "/api/parse-837"): WRITE_ROLES,

backend/tests/test_api_rebill.py

@@ -0,0 +1,193 @@
+"""SP41 Task 11 — HTTP endpoints for the in-window rebill pipeline.
+
+Three tests:
+
+  1. ``test_post_rebill_requires_auth_or_returns_404_or_422_when_no_input`` —
+     bare POST with no body. With ``AUTH_DISABLED = True`` (the autouse
+     conftest default) the request reaches the handler; without a body,
+     Pydantic returns 422. If AUTH_DISABLED is ever flipped off, the
+     matrix_gate returns 401/403 first. Either outcome proves the route
+     is wired.
+
+  2. ``test_post_rebill_runs_and_returns_summary_path`` — happy path.
+     Builds a 1-row visits CSV + zero-SVC 835 fixture under tmp_path,
+     POSTs with the window pinned + paths pointed at the fixtures,
+     asserts 200 + ``summary_path`` + a ``counts`` dict that includes
+     the REBILLED_B key (the in-window visit lands as NOT_IN_835 →
+     REBILLED_B because the 835 has no SVCs and the visit is fresh
+     enough to clear the 120-day gate).
+
+  3. ``test_get_rebill_status_returns_recent_runs`` — seeds a stub
+     ``summary.csv`` under ``tmp_path / "rebills_root"`` and
+     monkey-patches :data:`cyclone.api_routers.rebill.REBILLS_DIR` to
+     point there, so the GET handler's filesystem scan finds it
+     without the test having to chdir the whole process.
+"""
+from __future__ import annotations
+
+import csv
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+# --------------------------------------------------------------------------- #
+# Fixtures
+# --------------------------------------------------------------------------- #
+
+
+@pytest.fixture
+def client() -> TestClient:
+    """Standard TestClient; conftest.py autouse handles DB + auth gate.
+
+    AUTH_DISABLED is True so the matrix_gate short-circuits and the
+    request reaches the handler (no login dance required). The seeded
+    clearhouse isn't needed — the rebill pipeline doesn't touch SFTP.
+    """
+    from cyclone.api import app
+    return TestClient(app)
+
+
+def _write_visits_csv(path: Path, rows: list[tuple[str, str, str, str]]) -> Path:
+    """rows: list of (dos_mmddyyyy, member, procedure, billed_str)."""
+    with path.open("w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow(["Visit Date", "Member ID", "Procedure Code", "Billable Amount"])
+        for r in rows:
+            w.writerow(r)
+    return path
+
+
+def _stub_835(ingest_dir: Path, name: str = "x.835") -> Path:
+    """An *.835 file parseable enough to not crash the walker.
+
+    Content is just an empty 835 envelope; the rebill pipeline's
+    ``parse_835_svc`` reparser emits zero SVCs, which is what we want
+    for the NOT_IN_835 → REBILLED_B classification.
+    """
+    ingest_dir.mkdir(exist_ok=True)
+    p = ingest_dir / name
+    p.write_text("ST*835*0001~SE*0*0001~")
+    return p
+
+
+# --------------------------------------------------------------------------- #
+# Tests
+# --------------------------------------------------------------------------- #
+
+
+def test_post_rebill_requires_auth_or_returns_404_or_422_when_no_input(client):
+    """Bare POST with no body — should NOT 500.
+
+    With AUTH_DISABLED the matrix_gate short-circuits and Pydantic
+    validation fires on the empty body (422). With auth enabled the
+    matrix_gate returns 401/403 first. Either outcome proves the route
+    is wired and the request reached the handler.
+    """
+    resp = client.post("/api/admin/rebill-from-835")
+    assert resp.status_code in (401, 403, 422), (
+        f"unexpected status {resp.status_code}: {resp.text}"
+    )
+
+
+def test_post_rebill_runs_and_returns_summary_path(client, tmp_path):
+    """Happy path: 1 in-window visit, 0 SVCs → REBILLED_B + summary.csv."""
+    visits_path = _write_visits_csv(tmp_path / "visits.csv", [
+        ("06/27/2026", "J813715", "T1019", "$2.32"),
+    ])
+    ingest_dir = _stub_835(tmp_path / "ingest")
+    out_dir = tmp_path / "out"
+
+    resp = client.post(
+        "/api/admin/rebill-from-835",
+        json={
+            "window": "2026-01-01..2026-06-27",
+            "override_filing": False,
+            "visits_csv_path": str(visits_path),
+            "ingest_dir": str(ingest_dir),
+            "out_dir": str(out_dir),
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert "summary_path" in body, body
+    assert "counts" in body, body
+    assert "pipeline_a_files" in body, body
+    assert "pipeline_b_files" in body, body
+
+    # summary_path is real and points at the out_dir we passed in.
+    assert Path(body["summary_path"]).exists(), body["summary_path"]
+    assert str(out_dir) in body["summary_path"]
+
+    # The counts dict surfaces REBILLED_B for the in-window visit (the
+    # 835 has no SVCs so the visit falls into NOT_IN_835 → REBILLED_B
+    # — it's well within the 120-day gate).
+    counts = body["counts"]
+    assert "REBILLED_B" in counts, counts
+    assert counts["REBILLED_B"] >= 1, counts
+
+    # Pipeline B emitted exactly one file for the one REBILLED_B visit.
+    assert len(body["pipeline_b_files"]) == 1, body["pipeline_b_files"]
+
+    # The summary CSV header matches the writer's contract so the GET
+    # /status handler can parse it.
+    with open(body["summary_path"], newline="") as f:
+        reader = csv.DictReader(f)
+        rows = list(reader)
+    assert len(rows) == 1, rows
+    assert rows[0]["disposition"] == "REBILLED_B"
+    assert rows[0]["member_id"] == "J813715"
+
+
+def test_get_rebill_status_returns_recent_runs(client, tmp_path, monkeypatch):
+    """GET /status scans the configured REBILLS_DIR and tallies per-disposition counts.
+
+    The handler reads :data:`cyclone.api_routers.rebill.REBILLS_DIR`
+    (module-level so tests can monkeypatch it). We point that at a
+    tmp_path-relative tree, seed one dated subdir with a 2-row
+    summary.csv, and assert the response surfaces that directory's
+    name + per-disposition tally.
+    """
+    from cyclone.api_routers import rebill
+
+    # Build a date-named subdir with a real summary.csv shape so the
+    # handler's csv.DictReader finds a `disposition` column.
+    rebills_root = tmp_path / "rebills_root"
+    dated_dir = rebills_root / "2026-07-07"
+    dated_dir.mkdir(parents=True)
+    summary = dated_dir / "summary.csv"
+    with summary.open("w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow([
+            "dos", "member_id", "procedure", "billed",
+            "disposition", "unpaid", "cas_reasons", "file_path",
+        ])
+        w.writerow([
+            "2026-06-27", "MEM-A", "T1019", "2.32",
+            "REBILLED_B", "2.32", "", "pipeline-b/",
+        ])
+        w.writerow([
+            "2020-01-01", "MEM-B", "T1019", "2.32",
+            "EXCLUDED_TIMELY_FILING", "2.32", "", "",
+        ])
+
+    # Repoint the handler's filesystem root at our tmp_path tree.
+    monkeypatch.setattr(rebill, "REBILLS_DIR", rebills_root)
+
+    resp = client.get("/api/admin/rebill-from-835/status")
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert "recent_runs" in body, body
+    assert len(body["recent_runs"]) == 1, body["recent_runs"]
+
+    run = body["recent_runs"][0]
+    assert run["as_of"] == "2026-07-07", run
+    assert "summary_path" in run, run
+    assert Path(run["summary_path"]).exists(), run["summary_path"]
+    counts = run["counts"]
+    # Each disposition surfaced exactly once — matches the 2-row CSV.
+    assert counts.get("REBILLED_B") == 1, counts
+    assert counts.get("EXCLUDED_TIMELY_FILING") == 1, counts