feat(sp37): POST /api/submit-batch endpoint

Thin HTTP wrapper around `cyclone.submission.submit_file` — the
HTTP twin of the `cyclone submit-batch` CLI (SP37 Task 5). Same
walker pattern, same `._*` AppleDouble skip, same `limit`
semantics, same per-file outcomes; both surfaces now agree on DB +
SFTP state byte-for-byte.

Body: `{ingest_dir, validate?, actor?, limit?}`. Response:
`{submitted, skipped, failed, results:[{file, outcome, batch_id,
error}]}`. Status codes:
  - 200 on completed runs (per-file failures live in the body)
  - 401 unauthenticated (matrix_gate)
  - 404 no clearhouse seeded
  - 409 clearhouse SFTP block in stub mode
  - 422 ingest_dir missing on disk or Pydantic body error

Does NOT inject `sftp_client_factory` — submit_file uses its
default paramiko factory so SKIPPED is reachable in production.
Tests monkey-patch submit_file itself.

Permissions entry: `POST /api/submit-batch` → WRITE_ROLES (admin +
user), matching `POST /api/resubmit` posture.

Tests (11): happy path, AppleDouble skip, limit, empty ingest_dir
422, mixed-outcome 200, unexpected-exception-as-failed,
Pydantic-missing-422, on-disk-missing-422, no-clearhouse-404,
stub-mode-409, auth-gate-401.

backend/src/cyclone/api_routers/__init__.py

@@ -28,6 +28,7 @@ from cyclone.api_routers import (
     providers,
     reconciliation,
     remittances,
+    submission,
     ta1_acks,
 )
 
@@ -49,6 +50,7 @@ routers: list[APIRouter] = [
     providers.router,   # gated
     reconciliation.router, # gated
     remittances.router, # gated
+    submission.router,  # gated
     ta1_acks.router,    # gated
 ]
 

backend/src/cyclone/api_routers/submission.py

@@ -0,0 +1,184 @@
+"""SP37 Task 6: HTTP endpoint for the canonical submit-batch flow.
+
+Thin wrapper around ``cyclone.submission.submit_file`` — same logic as
+the ``cyclone submit-batch`` CLI (SP37 Task 5), just framed as JSON in
+/ JSON out and gated by ``matrix_gate``. The walker pattern, ``._*``
+AppleDouble skip, ``limit`` semantics, and per-file outcomes all match
+the CLI byte-for-byte so a batch run via the CLI and the same batch
+run via this endpoint produce identical DB + SFTP state.
+
+The endpoint deliberately does NOT inject an ``sftp_client_factory``:
+``submit_file`` defaults to its paramiko-based factory so SKIPPED is
+reachable in production (the ``SftpClient`` wrapper has no ``stat()``).
+Tests monkey-patch ``cyclone.api_routers.submission.submit_file``
+itself; that avoids the paramiko factory entirely without touching
+the helper's contract.
+
+Status code contract (per Task 6 spec §4):
+  - 200: completed run. Per-file failures live in the JSON body.
+  - 401: not authenticated (matrix_gate).
+  - 404: no clearhouse seeded (config-level "missing" → 4xx, not 5xx).
+  - 409: clearhouse SFTP block is in stub mode (refuses to upload).
+  - 422: ``ingest_dir`` missing on disk OR Pydantic body validation
+    failed (missing fields, wrong types).
+  - 5xx: truly unexpected exceptions propagate (do not swallow).
+"""
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, ConfigDict, Field
+
+from cyclone.auth.deps import matrix_gate
+from cyclone.submission.core import submit_file
+from cyclone.submission.result import SubmitOutcome, SubmitResult
+
+log = logging.getLogger(__name__)
+
+router = APIRouter(
+    prefix="/api",
+    tags=["submission"],
+    dependencies=[Depends(matrix_gate)],
+)
+
+
+class SubmitBatchRequest(BaseModel):
+    """Body schema for ``POST /api/submit-batch``.
+
+    ``ingest_dir`` has no default so Pydantic raises 422 when it's
+    missing — better UX than letting the walker crash on a missing
+    path. ``validate_files`` and ``actor`` default so a minimal client
+    can skip them. ``limit`` truncates the file list after the walker
+    collects it (mirrors the CLI's post-collection ``if i > limit:
+    break`` semantics, but applied as a slice since the HTTP body
+    model is type-checked up-front).
+
+    The ``validate_files`` field is aliased to ``validate`` in the JSON
+    body to mirror the CLI's ``--validate`` flag and avoid the
+    hardcoded Pydantic warning about ``validate`` shadowing
+    ``BaseModel.validate``. ``populate_by_name=True`` lets tests
+    construct the model with either key.
+    """
+    model_config = ConfigDict(populate_by_name=True, protected_namespaces=())
+
+    ingest_dir: str
+    validate_files: bool = Field(default=True, alias="validate")
+    actor: str = "api-submit-batch"
+    limit: int | None = None
+
+
+@router.post("/submit-batch")
+def post_submit_batch(body: SubmitBatchRequest):
+    """Submit every ``batch-*-claims/*.x12`` under ``ingest_dir``.
+
+    Walks ``ingest_dir`` for any directory matching ``batch-*-claims``,
+    collects each one's ``*.x12`` files (sorted, with ``._*``
+    AppleDouble files skipped), truncates to ``limit`` if set, then
+    calls :func:`cyclone.submission.submit_file` per file with the
+    seeded clearhouse's ``sftp_block`` and ``actor`` from the body.
+
+    Returns counts (``submitted`` / ``skipped`` / ``failed``) plus a
+    per-file ``results`` array. Per-file failures NEVER change the
+    HTTP status code — the response is 200 whenever the run itself
+    completed.
+    """
+    from cyclone.store import store as cycl_store
+
+    # 1. Config-level guards. Order matters: a missing clearhouse is a
+    #    404 (config-level "missing"), but if it IS present and in stub
+    #    mode the operator's request is a 409 (configured-but-wrong).
+    clearhouse = cycl_store.get_clearhouse()
+    if clearhouse is None:
+        raise HTTPException(
+            status_code=404, detail="no clearhouse seeded",
+        )
+    sftp_block = clearhouse.sftp_block
+    if sftp_block.stub:
+        raise HTTPException(
+            status_code=409, detail="clearhouse SFTP block is in stub mode",
+        )
+
+    # 2. Resolve + validate the ingest dir. Use ``resolve()`` so a
+    #    symlink-relative path still produces a stable error message.
+    root = Path(body.ingest_dir).resolve()
+    if not root.exists():
+        raise HTTPException(
+            status_code=422,
+            detail=f"ingest_dir does not exist: {root}",
+        )
+
+    # 3. Walker — must match the CLI EXACTLY. Same sort, same ``._*``
+    #    AppleDouble skip. Any drift here is a quiet split between the
+    #    two surfaces and silently produces different batch outcomes.
+    files: list[Path] = []
+    for batch_dir in sorted(root.glob("batch-*-claims")):
+        files.extend(sorted(
+            p for p in batch_dir.glob("*.x12")
+            if not p.name.startswith("._")
+        ))
+
+    # 4. ``limit`` truncates after collection. The CLI uses an inline
+    #    ``if i > limit: break``; we slice instead because the HTTP
+    #    body model validates ``limit`` up-front (Pydantic-level int
+    #    check) and slicing keeps the walker branchless.
+    if body.limit is not None:
+        files = files[: body.limit]
+
+    if not files:
+        raise HTTPException(
+            status_code=422,
+            detail=f"no batch-*-claims/*.x12 files found under {root}",
+        )
+
+    # 5. Per-file submit. Wrap the helper call in try/except so an
+    #    unexpected exception in submit_file surfaces as a per-file
+    #    failure (outcome="unexpected") instead of crashing the whole
+    #    run. The helper's own SubmitOutcome enum covers every typed
+    #    failure path; an uncaught exception here is a true
+    #    surprise (bug or service outage mid-loop).
+    results: list[dict] = []
+    submitted = skipped = failed = 0
+    for src in files:
+        try:
+            r = submit_file(
+                src,
+                sftp_block=sftp_block,
+                actor=body.actor,
+                validate=body.validate_files,
+                # No ``sftp_client_factory`` — submit_file's default
+                # paramiko factory opens the real MFT. Tests
+                # monkey-patch submit_file itself instead of wiring a
+                # factory here.
+            )
+        except Exception as exc:  # noqa: BLE001
+            log.exception(
+                "submit-batch unexpected error on %s", src.name,
+            )
+            r = SubmitResult(
+                file=src.name,
+                outcome=SubmitOutcome.SFTP_FAILED,
+                error=f"{exc.__class__.__name__}: {exc}",
+            )
+
+        if r.outcome == SubmitOutcome.SUBMITTED:
+            submitted += 1
+        elif r.outcome == SubmitOutcome.SKIPPED:
+            skipped += 1
+        else:
+            failed += 1
+
+        results.append({
+            "file": r.file,
+            "outcome": r.outcome.value,
+            "batch_id": r.batch_id,
+            "error": r.error,
+        })
+
+    return {
+        "submitted": submitted,
+        "skipped": skipped,
+        "failed": failed,
+        "results": results,
+    }

backend/src/cyclone/auth/permissions.py

@@ -86,6 +86,7 @@ PERMISSIONS: dict[tuple[str, str], set[Role]] = {
     ("POST", "/api/acks"): WRITE_ROLES,
     ("POST", "/api/batches"): WRITE_ROLES,  # /export-837 regenerates X12 from DB rows
     ("POST", "/api/eligibility"): WRITE_ROLES,
+    ("POST", "/api/submit-batch"): WRITE_ROLES,  # SP37: canonical outbound path (mirrors CLI)
 
     # CSV export — read-only.
     ("GET", "/api/export.csv"): ALL_ROLES,

backend/tests/test_api_submit_batch.py

@@ -0,0 +1,355 @@
+"""SP37 Task 6: POST /api/submit-batch endpoint.
+
+Thin HTTP wrapper around ``cyclone.submission.submit_file``. Mirrors the
+``cyclone submit-batch`` CLI walker exactly: each ``batch-*-claims/*.x12``
+under ``ingest_dir`` is submitted in order, ``._*`` AppleDouble files are
+skipped, ``limit`` truncates after collection.
+
+Tests monkey-patch ``cyclone.api_routers.submission.submit_file`` so we
+never touch the real paramiko factory or a live MFT.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cyclone.submission.result import SubmitOutcome, SubmitResult
+
+_FIXTURE = Path(__file__).parent / "fixtures" / "submit-batch" / "single-claim.x12"
+
+
+@pytest.fixture
+def client(tmp_path) -> TestClient:
+    """Standard TestClient; conftest.py autouse handles DB + auth gate.
+
+    Does NOT touch env vars (per the plan-vs-reality correction in
+    Task 6's instructions). Seeds the clearhouse then flips
+    ``sftp_block.stub`` to ``False`` so the file-walking tests reach
+    the walker (otherwise the default-seeded stub=True trips the 409
+    guard before any per-file work runs). The ``stub_mode`` test
+    re-seeds stub=True explicitly so its 409 still fires.
+    """
+    from cyclone import db as db_mod
+    from cyclone.store import store as cycl_store
+
+    db_mod._reset_for_tests()
+    db_mod.init_db()
+    cycl_store.ensure_clearhouse_seeded()
+    ch = cycl_store.get_clearhouse()
+    # Flip stub=False so the walker actually runs (SFTP calls are
+    # intercepted via the submit_file monkey-patch in each test).
+    cycl_store.update_clearhouse(
+        ch.model_copy(update={"sftp_block": ch.sftp_block.model_copy(update={"stub": False})}),
+    )
+
+    from cyclone.api import app
+
+    return TestClient(app)
+
+
+def _stage_batch(tmp_path: Path, n: int) -> Path:
+    """Copy N copies of the single-claim fixture into ``batch-test-claims/``."""
+    batch_dir = tmp_path / "batch-test-claims"
+    batch_dir.mkdir()
+    payload = _FIXTURE.read_bytes()
+    for i in range(n):
+        (batch_dir / f"claim-{i}.x12").write_bytes(payload)
+    return batch_dir
+
+
+# --------------------------------------------------------------------------- #
+# Happy path + walker semantics
+# --------------------------------------------------------------------------- #
+
+
+def test_submit_batch_happy_path(client: TestClient, tmp_path, monkeypatch):
+    """3 valid 837 files → 200 + body shape with submitted=3 (or skipped=3 on re-run).
+
+    Monkey-patches ``cyclone.api_routers.submission.submit_file`` to
+    return a synthetic SUBMITTED SubmitResult per file. Skipped stays
+    possible because SubmitResult's batch_id is what matters and the
+    fake returns it consistently — but a fresh DB should always land
+    in the SUBMITTED branch.
+    """
+    _stage_batch(tmp_path, 3)
+
+    def _fake_submit(path, **_kw):
+        return SubmitResult(
+            file=path.name, outcome=SubmitOutcome.SUBMITTED, batch_id="b1",
+        )
+
+    monkeypatch.setattr(
+        "cyclone.api_routers.submission.submit_file", _fake_submit,
+    )
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    # Tolerate either counter — counter totals are deterministic with a
+    # fresh DB, but the assertion below focuses on the structural shape
+    # the operator cares about: every file is accounted for.
+    assert body["submitted"] + body["skipped"] + body["failed"] == 3
+    assert isinstance(body["results"], list)
+    assert len(body["results"]) == 3
+    for row in body["results"]:
+        assert "file" in row
+        assert "outcome" in row
+        assert "batch_id" in row
+        assert "error" in row
+
+
+def test_submit_batch_skips_apple_double_files(client: TestClient, tmp_path, monkeypatch):
+    """``._foo.x12`` AppleDouble files must be skipped (mirrors the CLI walker)."""
+    batch_dir = tmp_path / "batch-test-claims"
+    batch_dir.mkdir()
+    (batch_dir / "real.x12").write_bytes(_FIXTURE.read_bytes())
+    (batch_dir / "._real.x12").write_bytes(b"\x00\x01\x02")  # macOS AppleDouble noise
+
+    def _fake_submit(path, **_kw):
+        return SubmitResult(
+            file=path.name, outcome=SubmitOutcome.SUBMITTED, batch_id="b1",
+        )
+
+    monkeypatch.setattr(
+        "cyclone.api_routers.submission.submit_file", _fake_submit,
+    )
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    # Only real.x12 was submitted (one call); AppleDouble was skipped.
+    assert len(body["results"]) == 1
+    assert body["results"][0]["file"] == "real.x12"
+
+
+def test_submit_batch_limit_truncates(client: TestClient, tmp_path, monkeypatch):
+    """``limit=1`` against 3 files → body has exactly 1 result."""
+    _stage_batch(tmp_path, 3)
+
+    submitted_files: list[str] = []
+
+    def _fake_submit(path, **_kw):
+        submitted_files.append(path.name)
+        return SubmitResult(
+            file=path.name, outcome=SubmitOutcome.SUBMITTED, batch_id="b1",
+        )
+
+    monkeypatch.setattr(
+        "cyclone.api_routers.submission.submit_file", _fake_submit,
+    )
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False, "limit": 1},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert len(body["results"]) == 1
+    assert body["submitted"] + body["skipped"] + body["failed"] == 1
+    assert len(submitted_files) == 1
+
+
+def test_submit_batch_empty_ingest_dir_returns_422(client: TestClient, tmp_path):
+    """No batch-*-claims dir under ingest_dir → 422 (matches the CLI's exit 2 path)."""
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 422, resp.text
+
+
+def test_submit_batch_per_file_failure_keeps_status_200(
+    client: TestClient, tmp_path, monkeypatch,
+):
+    """A SUBMITTED + SFTP_FAILED mix → 200 with failed=1, submitted=1.
+
+    Per-file failures live in the JSON body, NOT in the status code.
+    The endpoint always returns 200 on a completed run.
+    """
+    _stage_batch(tmp_path, 2)
+
+    outcomes = iter([
+        SubmitResult(file="claim-0.x12", outcome=SubmitOutcome.SUBMITTED, batch_id="b1"),
+        SubmitResult(
+            file="claim-1.x12", outcome=SubmitOutcome.SFTP_FAILED,
+            batch_id="b2", error="sftp down",
+        ),
+    ])
+
+    def _fake_submit(path, **_kw):
+        return next(outcomes)
+
+    monkeypatch.setattr(
+        "cyclone.api_routers.submission.submit_file", _fake_submit,
+    )
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["submitted"] == 1
+    assert body["failed"] == 1
+    assert body["skipped"] == 0
+    assert len(body["results"]) == 2
+    # The failure row carries the error string so the operator can see
+    # what went wrong without a separate API call.
+    assert body["results"][1]["outcome"] == "sftp_failed"
+    assert body["results"][1]["error"] == "sftp down"
+
+
+def test_submit_batch_unexpected_exception_recorded_in_results(
+    client: TestClient, tmp_path, monkeypatch,
+):
+    """If submit_file raises, the file lands in results[] as a failed row.
+
+    The router catches the exception, builds a SubmitResult with
+    ``outcome=SubmitOutcome.SFTP_FAILED`` (per Task 6's instruction
+    that "any enum value works as a marker; choose whichever is least
+    misleading"), and counts it as ``failed``. The full error string
+    (including exception class name) is in ``error`` so the operator
+    can tell it was a real exception, not a typed failure path.
+    """
+    _stage_batch(tmp_path, 1)
+
+    def _explode(path, **_kw):
+        raise RuntimeError("kaboom")
+
+    monkeypatch.setattr(
+        "cyclone.api_routers.submission.submit_file", _explode,
+    )
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["failed"] == 1
+    # The marker is the SFTP_FAILED enum value (per Task 6 design
+    # choice). The full exception class+message is in ``error`` so the
+    # operator can distinguish a real exception from a typed failure.
+    assert body["results"][0]["outcome"] == "sftp_failed"
+    assert "kaboom" in body["results"][0]["error"]
+    assert "RuntimeError" in body["results"][0]["error"]
+
+
+# --------------------------------------------------------------------------- #
+# Request validation
+# --------------------------------------------------------------------------- #
+
+
+def test_submit_batch_missing_ingest_dir_returns_422(client: TestClient):
+    """Body without ``ingest_dir`` → 422 from Pydantic (NOT your manual check)."""
+    resp = client.post("/api/submit-batch", json={"validate": False})
+    assert resp.status_code == 422, resp.text
+
+
+def test_submit_batch_ingest_dir_missing_on_disk_returns_422(
+    client: TestClient, tmp_path,
+):
+    """Body has a path that doesn't exist on disk → 422."""
+    resp = client.post(
+        "/api/submit-batch",
+        json={
+            "ingest_dir": str(tmp_path / "does-not-exist"),
+            "validate": False,
+        },
+    )
+    assert resp.status_code == 422, resp.text
+    assert "does not exist" in resp.text or "does-not-exist" in resp.text
+
+
+# --------------------------------------------------------------------------- #
+# Clearhouse posture (404 / 409)
+# --------------------------------------------------------------------------- #
+
+
+def test_submit_batch_no_clearhouse_returns_404(client: TestClient, tmp_path, monkeypatch):
+    """Delete the clearhouse row → endpoint must refuse with 404, not 500.
+
+    404 per the Task 6 status code contract: a missing clearhouse is a
+    config-level "missing" (4xx), not an unexpected exception.
+    """
+    from cyclone import db as db_mod
+    from cyclone.db import ClearhouseORM
+
+    # Wipe the seeded clearhouse row.
+    with db_mod.SessionLocal()() as s:
+        row = s.get(ClearhouseORM, 1)
+        if row is not None:
+            s.delete(row)
+            s.commit()
+
+    _stage_batch(tmp_path, 1)
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 404, resp.text
+    assert "clearhouse" in resp.text.lower()
+
+
+def test_submit_batch_stub_mode_returns_409(client: TestClient, tmp_path, monkeypatch):
+    """When the seeded clearhouse's ``sftp_block.stub`` is True → 409.
+
+    The default client fixture flips stub=False so file-walking tests
+    reach the walker. This test flips it back to True so the
+    stub-mode 409 guard fires before any per-file work runs.
+    """
+    from cyclone.store import store as cycl_store
+
+    ch = cycl_store.get_clearhouse()
+    assert ch is not None
+    # Re-flip stub=True (the default client fixture set it to False).
+    cycl_store.update_clearhouse(
+        ch.model_copy(update={
+            "sftp_block": ch.sftp_block.model_copy(update={"stub": True}),
+        }),
+    )
+
+    _stage_batch(tmp_path, 1)
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 409, resp.text
+    assert "stub" in resp.text.lower()
+
+
+# --------------------------------------------------------------------------- #
+# Auth gate
+# --------------------------------------------------------------------------- #
+
+
+def test_submit_batch_auth_gate(client: TestClient, tmp_path, monkeypatch):
+    """No cookie + AUTH_DISABLED=False → 401 from matrix_gate.
+
+    The conftest autouse fixture sets ``AUTH_DISABLED = True`` at
+    fixture setup and resets to False at teardown. Inside the test we
+    flip the module attribute so the gate fires — but to avoid leaking
+    the flip into a sibling test we use monkeypatch.setattr (which
+    restores at test teardown, before conftest's finally block runs).
+    """
+    import cyclone.auth.deps as auth_deps
+
+    monkeypatch.setattr(auth_deps, "AUTH_DISABLED", False)
+
+    _stage_batch(tmp_path, 1)
+
+    resp = client.post(
+        "/api/submit-batch",
+        json={"ingest_dir": str(tmp_path), "validate": False},
+    )
+    assert resp.status_code == 401, resp.text