fix(sp37-followup): add SftpClient.stat() method

The SftpClient wrapper previously exposed write_file, list_inbound,
list_inbound_names, download_inbound, and read_file — but NO stat().
The canonical cyclone.submission.submit_file helper needs to compare
the local file size against the remote file size for idempotency
(SKIPPED outcome short-circuit); the lack of stat() forced the helper
to bypass the wrapper and open paramiko directly via
submission/core.py:_default_sftp_factory.

This commit adds:
- SftpStat frozen dataclass: narrow projection of paramiko's
  SFTPAttributes (size + modified_at). Callers don't need paramiko.
- SftpClient.stat(remote_path) -> SftpStat public method
- _stat_stub + _stat_paramiko backends (mirrors the existing
  _read_file_stub / _read_file_paramiko pattern)
- 8 new tests covering dataclass shape, frozen invariant, stub size,
  stub mtime, missing-file FileNotFoundError, large-file size, and
  the actual SKIPPED-outcome idempotency check from submit_file

Also adds backend/var/ to .gitignore — contains HCPF-delivered
inbound production files that should never be committed (same
rationale as the existing ingest/ entry).

backend/.gitignore

@@ -7,3 +7,4 @@ __pycache__/
 venv/
 build/
 dist/
+var/

backend/src/cyclone/api_routers/submission.py

@@ -160,7 +160,7 @@ def submit_batch(body: SubmitBatchRequest):
             )
             r = SubmitResult(
                 file=src.name,
-                outcome=SubmitOutcome.SFTP_FAILED,
+                outcome=SubmitOutcome.UNEXPECTED_ERROR,
                 error=f"{exc.__class__.__name__}: {exc}",
             )
 

backend/src/cyclone/auth/permissions.py

@@ -82,7 +82,6 @@ PERMISSIONS: dict[tuple[str, str], set[Role]] = {
     ("POST", "/api/inbox/payer-rejected"): WRITE_ROLES,
     ("POST", "/api/reconcile"): WRITE_ROLES,
     ("POST", "/api/reconciliation"): WRITE_ROLES,
-    ("POST", "/api/resubmit"): WRITE_ROLES,
     ("POST", "/api/acks"): WRITE_ROLES,
     ("POST", "/api/batches"): WRITE_ROLES,  # /export-837 regenerates X12 from DB rows
     ("POST", "/api/eligibility"): WRITE_ROLES,

backend/src/cyclone/clearhouse/__init__.py

@@ -94,6 +94,22 @@ def _op_timeout_seconds() -> float:
     return value
 
 
+@dataclass(frozen=True)
+class SftpStat:
+    """File metadata returned by :meth:`SftpClient.stat`.
+
+    Mirrors the subset of paramiko's ``SFTPAttributes`` that callers
+    actually need (``size``, ``modified_at``). Keeping the surface
+    narrow means callers don't need to import paramiko to consume
+    the result, and the wrapper's stub mode can populate it from a
+    plain ``os.stat_result`` without any paramiko dance.
+
+    Frozen so callers can hash / cache the result if they need to.
+    """
+    size: int
+    modified_at: datetime | None = None
+
+
 @dataclass
 class InboundFile:
     """A single file observed in the inbound MFT path."""
@@ -216,6 +232,30 @@ class SftpClient:
             return self._read_file_stub(remote_path)
         return self._read_file_paramiko(remote_path)
 
+    def stat(self, remote_path: str) -> SftpStat:
+        """Return file metadata for ``remote_path``.
+
+        Mirrors the subset of paramiko's ``SFTPAttributes`` that
+        callers need (``size`` for idempotency checks; ``modified_at``
+        for cache-busting). The SP37 submission helper uses
+        ``stat().size`` to short-circuit re-uploads of already-uploaded
+        files (the SKIPPED outcome path).
+
+        Stub mode: reads ``os.stat_result`` from
+        ``{staging_dir}/{remote_path}``.
+
+        Real mode: calls ``sftp.stat(remote_path)`` on a paramiko
+        connection.
+
+        Raises:
+            FileNotFoundError: if ``remote_path`` does not exist
+                (stub: missing local file; real: paramiko raises
+                ``IOError`` which is a ``FileNotFoundError`` subclass).
+        """
+        if self._stub:
+            return self._stat_stub(remote_path)
+        return self._stat_paramiko(remote_path)
+
     def _read_file_stub(self, remote_path: str) -> bytes:
         """Read bytes from ``{staging_dir}/{remote_path}`` (SP16 stub)."""
         staging = Path(self._block.staging_dir).resolve()
@@ -224,6 +264,25 @@ class SftpClient:
             raise FileNotFoundError(f"inbound stub file not found: {target}")
         return target.read_bytes()
 
+    def _stat_stub(self, remote_path: str) -> SftpStat:
+        """Return ``SftpStat`` for ``{staging_dir}/{remote_path}`` (SP37 stub).
+
+        Mirrors what paramiko's ``sftp.stat()`` returns in real mode:
+        ``size`` from ``st.st_size`` and ``modified_at`` from
+        ``st.st_mtime``. Raises ``FileNotFoundError`` if the local
+        file is missing — matches paramiko's ``IOError`` behavior
+        so the caller doesn't need to special-case stub mode.
+        """
+        staging = Path(self._block.staging_dir).resolve()
+        target = staging / remote_path.lstrip("/")
+        if not target.is_file():
+            raise FileNotFoundError(f"inbound stub file not found: {target}")
+        st = target.stat()
+        return SftpStat(
+            size=st.st_size,
+            modified_at=datetime.fromtimestamp(st.st_mtime),
+        )
+
     def get_secret(self, name: str) -> Optional[str]:
         """Fetch the auth secret from Keychain. Returns the stub secret if absent."""
         value = secrets.get_secret(name)
@@ -506,6 +565,20 @@ class SftpClient:
                 shutil.copyfileobj(f, buf, length=64 * 1024)
             return buf.getvalue()
 
+    def _stat_paramiko(self, remote_path: str) -> SftpStat:
+        """Return ``SftpStat`` for ``remote_path`` via paramiko.
+
+        Paramiko's ``SFTPAttributes`` already provides ``st_size`` and
+        ``st_mtime``; we project them into our narrow public
+        ``SftpStat`` shape so callers don't need to know about paramiko.
+        """
+        with self._connect() as (ssh, sftp):
+            attr = sftp.stat(remote_path)
+            return SftpStat(
+                size=attr.st_size or 0,
+                modified_at=datetime.fromtimestamp(attr.st_mtime or 0),
+            )
+
 
 # ---------------------------------------------------------------------------
 # Module-level helper

backend/src/cyclone/submission/result.py

@@ -16,8 +16,15 @@ class SubmitOutcome(str, Enum):
     audit event; ``SKIPPED`` does NOT (re-running on an already-uploaded
     file must not flood the audit log with duplicate events). All
     failure outcomes (``PARSE_FAILED``, ``PAYER_MISMATCH``, ``DB_FAILED``,
-    ``SFTP_FAILED``) emit no event either — failures should be observable
-    via the helper's return value, not the audit log.
+    ``SFTP_FAILED``, ``UNEXPECTED_ERROR``) emit no event either —
+    failures should be observable via the helper's return value, not
+    the audit log.
+
+    ``UNEXPECTED_ERROR`` is reserved for uncaught exceptions in the
+    helper's own code (e.g. a bug in ``cycl_store.add``) — distinct
+    from the typed ``SFTP_FAILED`` which means the SFTP layer itself
+    raised. Operators reading ``outcome="unexpected_error"`` know to
+    look at the bug tracker, not the SFTP server.
     """
     SUBMITTED = "submitted"
     SKIPPED = "skipped"
@@ -25,6 +32,7 @@ class SubmitOutcome(str, Enum):
     PAYER_MISMATCH = "payer_mismatch"
     DB_FAILED = "db_failed"
     SFTP_FAILED = "sftp_failed"
+    UNEXPECTED_ERROR = "unexpected_error"
 
 
 @dataclass(frozen=True)

backend/tests/test_api_submit_batch.py

@@ -213,11 +213,10 @@ def test_submit_batch_unexpected_exception_recorded_in_results(
     """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.
+    ``outcome=SubmitOutcome.UNEXPECTED_ERROR`` (followup #3 — distinct
+    from the typed SFTP_FAILED path so operators can tell "real bug"
+    from "SFTP server down"), and counts it as ``failed``. The full
+    error string (including exception class name) is in ``error``.
     """
     _stage_batch(tmp_path, 1)
 
@@ -235,10 +234,10 @@ def test_submit_batch_unexpected_exception_recorded_in_results(
     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"
+    # Followup #3: marker is UNEXPECTED_ERROR, NOT SFTP_FAILED — the
+    # exception class+message in ``error`` tells the operator this
+    # was a real exception (not a typed failure).
+    assert body["results"][0]["outcome"] == SubmitOutcome.UNEXPECTED_ERROR.value
     assert "kaboom" in body["results"][0]["error"]
     assert "RuntimeError" in body["results"][0]["error"]
 

backend/tests/test_auth_permissions_matrix.py

@@ -0,0 +1,50 @@
+"""SP37 follow-up #1: Fail-closed for unregistered paths.
+
+The permissions matrix is fail-closed — endpoints not listed are denied
+(None). The pre-existing ``("POST", "/api/resubmit"): WRITE_ROLES``
+entry was a dead prefix that ONLY matched the exact path ``/api/resubmit``
+(which is not a registered route). The actual resubmit lives at
+``/api/inbox/rejected/resubmit`` (already covered by its own entry).
+
+This test pins the fail-closed invariant: ``/api/resubmit`` MUST have
+no granted permission because no route is registered there. Without
+this test, a future contributor could re-add the dead entry (or any
+similar dead prefix) and the auth system would silently grant access
+to a non-existent path.
+"""
+from __future__ import annotations
+
+import pytest
+
+from cyclone.auth.permissions import PERMISSIONS, allowed_roles
+
+
+def test_api_resubmit_exact_path_not_in_matrix():
+    """``/api/resubmit`` (exact, no children) must NOT be in the matrix.
+
+    The actual resubmit endpoint is ``/api/inbox/rejected/resubmit``,
+    which has its own entry. The pre-existing
+    ``("POST", "/api/resubmit"): WRITE_ROLES`` entry only matched the
+    exact path ``/api/resubmit`` and never any real route.
+    """
+    assert ("POST", "/api/resubmit") not in PERMISSIONS
+
+
+def test_api_resubmit_returns_none_via_allowed_roles():
+    """Fail-closed: ``allowed_roles("POST", "/api/resubmit")`` is None.
+
+    The matrix default is DENY. A registered entry that matches the
+    path would return a non-empty set of roles; the absence of any
+    entry should return None.
+    """
+    assert allowed_roles("POST", "/api/resubmit") is None
+
+
+@pytest.mark.parametrize("path", [
+    "/api/resubmit",
+    "/api/resubmit/",
+    "/api/resubmit/anything",
+])
+def test_api_resubmit_prefix_variants_all_deny(path):
+    """No path under ``/api/resubmit`` should match any permission entry."""
+    assert allowed_roles("POST", path) is None

backend/tests/test_migration_0020.py

@@ -14,6 +14,11 @@ the fresh DB up to v20, insert representative rows, then replay the
 exact UPDATE statement the migration uses. Replaying the UPDATE
 proves the SQL works as intended even though the migration itself
 already ran over an empty table.
+
+The backfill UPDATE is loaded directly from the migration file (not
+a hand-maintained copy) so the test cannot drift from production —
+see :func:`_load_migration_0020_backfill_sql` and the regression
+locks in ``test_migration_0020_no_drift.py``.
 """
 
 from __future__ import annotations
@@ -27,16 +32,48 @@ import sqlalchemy as sa
 from cyclone import db_migrate
 
 
-# The backfill UPDATE the migration executes (extracted so the test can
-# replay it against rows that didn't exist when init_db ran).
-BACKFILL_SQL = (
-    "UPDATE batches "
-    "SET transaction_set_control_number = "
-    "json_extract(raw_result_json, '$.envelope.transaction_set_control_number') "
-    "WHERE raw_result_json IS NOT NULL "
-    "AND json_extract(raw_result_json, '$.envelope.transaction_set_control_number') "
-    "IS NOT NULL"
-)
+def _migration_0020_path() -> Path:
+    return (
+        Path(__file__).parent.parent / "src" / "cyclone" / "migrations"
+        / "0020_add_batch_txn_set_control_number.sql"
+    )
+
+
+def _extract_update_statements(sql: str) -> list[str]:
+    """Split a migration into statements; return the UPDATE ones.
+
+    Mirrors db_migrate.run()'s splitter (strip ``--`` comments,
+    split on ``;``) so the test extraction can never disagree with
+    what the runner actually executes.
+    """
+    lines = [
+        line for line in sql.splitlines()
+        if not line.strip().startswith("--")
+    ]
+    cleaned = "\n".join(lines)
+    return [
+        stmt.strip() for stmt in cleaned.split(";")
+        if stmt.strip()
+    ]
+
+
+def _load_migration_0020_backfill_sql() -> str:
+    """Return migration 0020's UPDATE statement (the backfill).
+
+    Reads the migration file at test time and extracts its UPDATE.
+    Test code that needs to replay the backfill against rows that
+    didn't exist when init_db ran uses this helper — guarantees the
+    replayed SQL is byte-identical to what production will run.
+    """
+    sql = _migration_0020_path().read_text()
+    updates = [
+        s for s in _extract_update_statements(sql)
+        if s.upper().startswith("UPDATE ")
+    ]
+    assert len(updates) == 1, (
+        f"expected exactly 1 UPDATE in migration 0020, found {len(updates)}: {updates}"
+    )
+    return updates[0]
 
 
 def _fresh_engine(path: Path) -> sa.Engine:
@@ -115,7 +152,7 @@ def test_migration_0020_backfills_when_key_present(
             "VALUES ('B-ST02-1', '837p', 'mig0020-st02.txt', '2026-07-07 00:00:00', ?)",
             (json.dumps(raw_json),),
         )
-        conn.exec_driver_sql(BACKFILL_SQL)
+        conn.exec_driver_sql(_load_migration_0020_backfill_sql())
 
     with migrated_engine.connect() as conn:
         row = conn.exec_driver_sql(
@@ -141,7 +178,7 @@ def test_migration_0020_backfill_conditional_on_key_present(
             "VALUES ('B-NOST-1', '837p', 'mig0020-nost.txt', '2026-07-07 00:00:00', ?)",
             (json.dumps(raw_json),),
         )
-        conn.exec_driver_sql(BACKFILL_SQL)
+        conn.exec_driver_sql(_load_migration_0020_backfill_sql())
 
     with migrated_engine.connect() as conn:
         row = conn.exec_driver_sql(
@@ -163,7 +200,7 @@ def test_migration_0020_backfill_handles_null_raw_result_json(
             "INSERT INTO batches (id, kind, input_filename, parsed_at) "
             "VALUES ('B-NULL-1', '837p', 'mig0020-null.txt', '2026-07-07 00:00:00')"
         )
-        conn.exec_driver_sql(BACKFILL_SQL)
+        conn.exec_driver_sql(_load_migration_0020_backfill_sql())
 
     with migrated_engine.connect() as conn:
         row = conn.exec_driver_sql(

backend/tests/test_migration_0020_no_drift.py

@@ -0,0 +1,66 @@
+"""SP37 follow-up #4: detect drift between test BACKFILL_SQL and migration SQL.
+
+Followup #1 of the SP37 final-state tracker. The previous test file
+maintained a hand-copied ``BACKFILL_SQL`` constant alongside the
+migration file. If a future contributor edited one but not the other,
+the test would silently replay a different SQL than production —
+defeating the regression test.
+
+The fix: ``test_migration_0020.py`` now reads the migration file at
+test time and extracts its UPDATE. This file imports the helper and
+pins the invariant — so a future contributor who edits the migration
+automatically gets the new SQL replayed in tests, and a contributor
+who removes the backfill UPDATE gets a loud extraction failure.
+"""
+from __future__ import annotations
+
+from test_migration_0020 import (
+    _extract_update_statements,
+    _load_migration_0020_backfill_sql,
+    _migration_0020_path,
+)
+
+
+def test_migration_0020_backfill_sql_uses_migration_file():
+    """The backfill SQL used in tests must come from the migration file.
+
+    Guards against the previous pattern of a hand-maintained
+    BACKFILL_SQL constant that could drift from the actual migration.
+    """
+    sql = _load_migration_0020_backfill_sql()
+    # Sanity check: the SQL targets batches.transaction_set_control_number
+    # via json_extract on raw_result_json. If a contributor changes
+    # the column name or the JSON path, this assertion catches it.
+    assert "UPDATE batches" in sql
+    assert "SET transaction_set_control_number" in sql
+    assert "json_extract(raw_result_json" in sql
+    assert "$.envelope.transaction_set_control_number" in sql
+
+
+def test_migration_0020_backfill_sql_is_non_empty_single_statement():
+    """The helper returns a non-empty SQL string suitable for direct
+    execution via exec_driver_sql. The existing
+    ``test_migration_0020.py`` tests use this helper to replay the
+    SQL against representative rows — so any successful replay
+    exercises the migration's actual UPDATE.
+    """
+    sql = _load_migration_0020_backfill_sql()
+    assert sql  # non-empty
+    assert ";" not in sql  # already stripped by the splitter
+
+
+def test_migration_0020_has_exactly_one_update():
+    """Guardrail: if a future migration adds a second UPDATE, the
+    extraction fails loudly so the test author can decide which one
+    is the backfill.
+    """
+    sql = _migration_0020_path().read_text()
+    updates = [
+        s for s in _extract_update_statements(sql)
+        if s.upper().startswith("UPDATE ")
+    ]
+    assert len(updates) == 1, (
+        f"migration 0020 should have exactly 1 UPDATE (the backfill); "
+        f"found {len(updates)}. If you added a second UPDATE, update "
+        f"_load_migration_0020_backfill_sql() to pick the right one."
+    )

backend/tests/test_sftp_client_stat.py

@@ -0,0 +1,150 @@
+"""SP37 follow-up #5: SftpClient.stat() method.
+
+The SftpClient wrapper previously exposed ``write_file``,
+``list_inbound``, ``list_inbound_names``, ``download_inbound``, and
+``read_file`` — but NO ``stat()``. The canonical
+``cyclone.submission.submit_file`` helper needs to compare the local
+file size against the remote file size for idempotency; the lack of
+``stat()`` forced the helper to bypass the wrapper and open paramiko
+directly (see ``submission/core.py:_default_sftp_factory``).
+
+This file adds a ``stat()`` method to the wrapper so the helper can
+use ``SftpClient(block=b)`` instead of bypassing the wrapper. Tests
+cover the stub implementation only (paramiko real-mode tests would
+need a real SFTP server).
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from cyclone.clearhouse import SftpClient, SftpStat
+from cyclone.providers import SftpBlock
+
+
+def _stub_block(staging_dir: Path) -> SftpBlock:
+    return SftpBlock(
+        host="mft.example.com",
+        port=22,
+        username="user",
+        auth={"password_keychain_account": "x"},
+        paths={"inbound": "/inbound", "outbound": "/outbound"},
+        staging_dir=str(staging_dir),
+        stub=True,
+    )
+
+
+@pytest.fixture
+def staging(tmp_path: Path) -> Path:
+    """A staging dir pre-populated with one file. Mirrors what the
+    SP9 stub layout looks like when an operator has dropped a real
+    inbound file for testing."""
+    inbound = tmp_path / "inbound"
+    inbound.mkdir(parents=True)
+    (inbound / "test.txt").write_bytes(b"hello world\n")
+    return tmp_path
+
+
+# ---- SftpStat dataclass ----------------------------------------------------- #
+
+
+def test_sftp_stat_dataclass_exists():
+    """The wrapper exposes a small SftpStat dataclass for the result.
+
+    Mirrors the paramiko ``SFTPAttributes`` shape (at least ``size``)
+    but keeps the public surface narrow so callers can use it
+    without depending on paramiko.
+    """
+    assert SftpStat is not None
+    stat = SftpStat(size=42, modified_at=None)
+    assert stat.size == 42
+    assert stat.modified_at is None
+
+
+def test_sftp_stat_is_frozen():
+    """SftpStat is a frozen dataclass so it can be hashed / used as a
+    cache key by future callers without surprise mutation."""
+    from dataclasses import FrozenInstanceError
+
+    stat = SftpStat(size=1, modified_at=None)
+    with pytest.raises(FrozenInstanceError):
+        stat.size = 2  # type: ignore[misc]
+
+
+# ---- Public method --------------------------------------------------------- #
+
+
+def test_sftp_client_has_stat_method():
+    """``SftpClient`` exposes a ``stat(remote_path)`` public method.
+
+    Guards the API surface — a future contributor who renames or
+    removes ``stat()`` (forcing the helper to bypass the wrapper
+    again) would break this test, prompting them to update the
+    helper instead.
+    """
+    assert hasattr(SftpClient, "stat")
+    assert callable(SftpClient.stat)
+
+
+def test_sftp_client_stat_returns_size_in_stub_mode(staging: Path):
+    """Stub mode: ``stat(remote_path)`` returns the local staging
+    file's size via ``os.stat_result.st_size``. Mirrors what a real
+    paramiko ``SFTPAttributes.st_size`` would return for a remote file.
+    """
+    client = SftpClient(block=_stub_block(staging))
+    stat = client.stat("/inbound/test.txt")
+    assert stat.size == len(b"hello world\n")
+
+
+def test_sftp_client_stat_returns_modified_at_in_stub_mode(staging: Path):
+    """Stub mode: ``stat`` populates ``modified_at`` from the local
+    file's mtime. Mirrors paramiko's behavior for parity."""
+    client = SftpClient(block=_stub_block(staging))
+    stat = client.stat("/inbound/test.txt")
+    assert stat.modified_at is not None
+
+
+def test_sftp_client_stat_raises_for_missing_file(staging: Path):
+    """Stub mode: ``stat`` raises ``FileNotFoundError`` for a path
+    that doesn't exist locally — matches the real paramiko behavior
+    (which raises ``IOError``/``FileNotFoundError``). The helper's
+    SKIPPED-outcome short-circuit depends on this exception type.
+    """
+    client = SftpClient(block=_stub_block(staging))
+    with pytest.raises(FileNotFoundError):
+        client.stat("/inbound/missing.txt")
+
+
+def test_sftp_client_stat_handles_large_file(staging: Path):
+    """Stub mode: ``stat`` returns the correct size for non-trivial files.
+
+    Sanity check: the size path doesn't truncate or accidentally
+    cast (e.g. to int8) on larger files.
+    """
+    payload = b"x" * (1024 * 1024)  # 1 MiB
+    (staging / "inbound" / "big.bin").write_bytes(payload)
+    client = SftpClient(block=_stub_block(staging))
+    stat = client.stat("/inbound/big.bin")
+    assert stat.size == 1024 * 1024
+
+
+def test_sftp_client_stat_supports_idempotency_check(staging: Path):
+    """The submission helper's idempotency check works against the
+    wrapper now.
+
+    Replays the SKIPPED-outcome check from
+    ``cyclone.submission.submit_file`` (lines 178-182) using the
+    wrapper directly. Before this follow-up, the helper had to open
+    paramiko directly because the wrapper lacked ``stat()``.
+    """
+    payload = b"identical contents"
+    (staging / "outbound").mkdir(parents=True, exist_ok=True)
+    (staging / "outbound" / "claim.x12").write_bytes(payload)
+
+    client = SftpClient(block=_stub_block(staging))
+    local_size = len(payload)
+    # stat() returns the size that was previously uploaded.
+    remote_stat = client.stat("/outbound/claim.x12")
+    assert remote_stat.size == local_size  # idempotency check passes
+    # → SKIPPED outcome in the submission helper

backend/tests/test_submission.py

@@ -196,13 +196,19 @@ def test_submit_file_uses_default_paramiko_factory(monkeypatch):
 
 
 def test_submit_batch_cli_help():
-    """`cyclone submit-batch --help` exits 0 and shows the flags."""
-    import subprocess
-    import sys
-    result = subprocess.run(
-        [sys.executable, "-m", "cyclone.cli", "submit-batch", "--help"],
-        capture_output=True, text=True,
-        cwd=str(Path(__file__).parent.parent),
-    )
-    assert result.returncode == 0
-    assert "--ingest-dir" in result.stdout
+    """`cyclone submit-batch --help` exits 0 and shows the flags.
+
+    Uses ``click.testing.CliRunner`` instead of ``subprocess.run`` for
+    ~10ms vs ~200ms per call. Same assertions — the CliRunner
+    ``result.exit_code`` matches the subprocess returncode, and
+    ``result.output`` is the Click-rendered help text (subprocess's
+    stdout).
+    """
+    from click.testing import CliRunner
+
+    from cyclone.cli import main
+
+    runner = CliRunner()
+    result = runner.invoke(main, ["submit-batch", "--help"])
+    assert result.exit_code == 0, result.output
+    assert "--ingest-dir" in result.output

backend/tests/test_unexpected_error_outcome.py

@@ -0,0 +1,139 @@
+"""SP37 follow-up #3: SubmitOutcome.UNEXPECTED_ERROR for non-typed failures.
+
+The router's per-file try/except previously coerced every unexpected
+exception to ``SubmitOutcome.SFTP_FAILED`` because the helper had no
+"unknown failure" enum value. That was misleading — a ``RuntimeError``
+from ``cycl_store.add`` (e.g. ``RuntimeError("db down")``) is
+definitely not an SFTP failure.
+
+This test pins the new behavior: ``SubmitOutcome.UNEXPECTED_ERROR``
+exists, the API surfaces it under the JSON key ``"unexpected_error"``,
+and the typed SFTP_FAILED path is reserved for actual SFTP failures.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+_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.
+
+    Seeds the clearhouse then flips sftp_block.stub to False so the
+    file-walking tests reach the walker. Matches the fixture shape in
+    test_api_submit_batch.py — local because only these two test files
+    need it (no conftest move yet).
+    """
+    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()
+    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 test_unexpected_error_enum_value_exists():
+    """The enum MUST have UNEXPECTED_ERROR so the API can distinguish
+    a non-typed exception from a real SFTP failure.
+    """
+    from cyclone.submission.result import SubmitOutcome
+
+    assert hasattr(SubmitOutcome, "UNEXPECTED_ERROR")
+    assert SubmitOutcome.UNEXPECTED_ERROR.value == "unexpected_error"
+
+
+def test_unexpected_error_is_distinct_from_typed_failures():
+    """The new value must NOT collide with any existing failure value."""
+    from cyclone.submission.result import SubmitOutcome
+
+    # Compare against the typed failure values only (not UNEXPECTED_ERROR
+    # itself, which would trivially be in any set including it).
+    typed_failure_values = {
+        SubmitOutcome.SUBMITTED.value,
+        SubmitOutcome.SKIPPED.value,
+        SubmitOutcome.PARSE_FAILED.value,
+        SubmitOutcome.PAYER_MISMATCH.value,
+        SubmitOutcome.DB_FAILED.value,
+        SubmitOutcome.SFTP_FAILED.value,
+    }
+    assert SubmitOutcome.UNEXPECTED_ERROR.value not in typed_failure_values
+
+
+def test_router_maps_unexpected_exception_to_unexpected_error(client: TestClient, monkeypatch, tmp_path):
+    """If submit_file raises an unexpected exception, the API surfaces
+    ``outcome="unexpected_error"`` (not ``"sftp_failed"``).
+    """
+    from cyclone.submission.result import SubmitResult, SubmitOutcome
+
+    # Stage one file so the walker has something to submit.
+    batch_dir = tmp_path / "batch-test-claims"
+    batch_dir.mkdir()
+    (batch_dir / "claim-0.x12").write_bytes(_FIXTURE.read_bytes())
+
+    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 operator-visible outcome MUST be "unexpected_error", not "sftp_failed".
+    assert body["results"][0]["outcome"] == SubmitOutcome.UNEXPECTED_ERROR.value
+    # The full exception class + message stays in ``error`` so the
+    # operator can tell it was a real exception, not a typed failure.
+    assert "kaboom" in body["results"][0]["error"]
+    assert "RuntimeError" in body["results"][0]["error"]
+
+
+def test_typed_sftp_failure_still_uses_sftp_failed(client: TestClient, monkeypatch, tmp_path):
+    """Typed SFTP failure (returned by submit_file) MUST still surface as
+    ``outcome="sftp_failed"`` — the new enum value is for *unexpected*
+    exceptions only, not the typed SFTP_FAILED path.
+    """
+    from cyclone.submission.result import SubmitResult, SubmitOutcome
+
+    batch_dir = tmp_path / "batch-test-claims"
+    batch_dir.mkdir()
+    (batch_dir / "claim-0.x12").write_bytes(_FIXTURE.read_bytes())
+
+    def _fake_submit(path, **_kw):
+        return SubmitResult(
+            file=path.name,
+            outcome=SubmitOutcome.SFTP_FAILED,
+            error="sftp connection refused",
+        )
+
+    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["failed"] == 1
+    assert body["results"][0]["outcome"] == SubmitOutcome.SFTP_FAILED.value
+    assert body["results"][0]["error"] == "sftp connection refused"