fix(sp37): submit_file default factory uses paramiko directly

SftpClient wrapper lacks a stat() method, so the previous default
factory produced a client whose stat() raised AttributeError — making
the SKIPPED outcome unreachable in production and silently
misclassifying all uploads as SFTP_FAILED. The helper now opens a
paramiko SFTP session directly (same pattern as
resubmit-rejected-claims in cli.py:620-650), so stat() and write_file()
both work end-to-end.

Also: tighten typing (SftpBlock instead of Any), add PAYER_MISMATCH
test, drop duplicate _db fixture, rename misleading test, add class
docstrings to SubmitResult/SubmitOutcome.

backend/src/cyclone/submission/__init__.py

@@ -14,4 +14,4 @@ upload via stat-then-put).
 from .core import submit_file
 from .result import SubmitOutcome, SubmitResult
 
-__all__ = ["submit_file", "SubmitOutcome", "SubmitResult"]
+__all__ = ["submit_file", "SubmitOutcome", "SubmitResult"]

backend/src/cyclone/submission/core.py

@@ -9,6 +9,11 @@ no SFTP call is made. If the SFTP call fails after a successful DB
 write, the row exists but the file never landed — re-running is safe
 (idempotent DB write via add_record's s.get check; idempotent SFTP
 upload via stat-then-put).
+
+The default SFTP factory uses paramiko directly (not the
+``SftpClient`` wrapper) because the wrapper exposes ``write_file``,
+``list_inbound``, and ``read_file`` but no ``stat()`` — which makes
+the SKIPPED outcome unreachable in production.
 """
 from __future__ import annotations
 
@@ -20,9 +25,9 @@ from typing import Any, Callable
 
 from cyclone import db as db_mod
 from cyclone.audit_log import AuditEvent, append_event
-from cyclone.clearhouse import SftpClient
 from cyclone.parsers.parse_837 import parse as parse_837_text
 from cyclone.parsers.payer import PayerConfig
+from cyclone.providers import SftpBlock
 from cyclone.store import store as cycl_store
 from cyclone.store.records import BatchRecord837
 
@@ -35,13 +40,55 @@ log = logging.getLogger(__name__)
 EXPECTED_PAYER_ID = "CO_TXIX"
 
 
+def _default_sftp_factory(sftp_block: SftpBlock) -> Any:
+    """Open a paramiko SFTP session to the real MFT. Mirrors
+    resubmit_rejected_claims._open_session (cli.py:620-640).
+    Caller is responsible for closing the session.
+
+    The returned ``sftp`` is a ``paramiko.SFTPClient`` — it exposes
+    ``stat(remote_path)`` (used for the idempotency check) and
+    ``put(local_path, remote_path)`` (used for the upload). The
+    underlying SSH handle is stashed on ``sftp._cyclone_ssh`` so the
+    caller can close it cleanly via ``getattr(sftp, "_cyclone_ssh",
+    None).close()`` after the upload finishes.
+
+    Raises:
+        RuntimeError: if the SFTP block is in stub mode (the CLI/HTTP
+            layer should already guard against this, but we re-check
+            here because paramiko will try to connect even when
+            ``stub=True``).
+    """
+    if sftp_block.stub:
+        # Same posture as resubmit_rejected_claims in cli.py:592-595:
+        # the operator refuses to upload in stub mode, and the helper
+        # surfaces that as SFTP_FAILED with an explicit error.
+        raise RuntimeError("SFTP block is in stub mode")
+
+    from paramiko import AutoAddPolicy, SSHClient
+
+    from cyclone.secrets import get_secret
+
+    pw = get_secret(sftp_block.auth.get("password_keychain_account", ""))
+    ssh = SSHClient()
+    ssh.set_missing_host_key_policy(AutoAddPolicy())
+    ssh.connect(
+        sftp_block.host, port=sftp_block.port,
+        username=sftp_block.username, password=pw,
+        timeout=15, banner_timeout=15, auth_timeout=15,
+    )
+    sftp = ssh.open_sftp()
+    # Attach ssh to sftp so the caller can close it cleanly.
+    sftp._cyclone_ssh = ssh
+    return sftp
+
+
 def submit_file(
     path: Path,
     *,
-    sftp_block: Any,
+    sftp_block: SftpBlock,
     actor: str,
     validate: bool = True,
-    sftp_client_factory: Callable[[Any], Any] | None = None,
+    sftp_client_factory: Callable[[SftpBlock], Any] | None = None,
 ) -> SubmitResult:
     """Submit one 837P file: parse → DB write → SFTP upload.
 
@@ -52,8 +99,9 @@ def submit_file(
         validate: parse the file before upload (default True). Catches
             bad byte-fixes early.
         sftp_client_factory: optional callable that returns an
-            SftpClient-compatible object. Defaults to the real
-            ``SftpClient(block=sftp_block)``. Tests inject a fake.
+            SFTP-client-compatible object (must expose ``stat()`` and
+            ``write_file()``). Defaults to :func:`_default_sftp_factory`
+            which opens a real paramiko session. Tests inject a fake.
 
     Returns:
         SubmitResult with file, outcome, batch_id (when written),
@@ -115,9 +163,12 @@ def submit_file(
         log.warning("submit_file %s: DB write failed: %s", file_label, exc)
         return SubmitResult(file_label, SubmitOutcome.DB_FAILED, error=str(exc))
 
-    # 3. SFTP upload.
+    # 3. SFTP upload. ``_default_sftp_factory`` opens a paramiko
+    # session directly because the SftpClient wrapper has no ``stat()``
+    # method — that omission made the SKIPPED outcome unreachable in
+    # production. Mirror cli.py:620-650.
     remote_path = f"{sftp_block.paths['outbound']}/{file_label}"
-    factory = sftp_client_factory or (lambda b: SftpClient(block=b))
+    factory = sftp_client_factory or _default_sftp_factory
     sftp = factory(sftp_block)
     local_size = len(content)
     try:
@@ -136,6 +187,16 @@ def submit_file(
             file_label, SubmitOutcome.SFTP_FAILED,
             batch_id=batch_id, error=str(exc),
         )
+    finally:
+        # Close the paramiko SSH handle the helper stashed on the
+        # SFTP client (paramiko does not auto-close on GC and a leak
+        # here costs a slot in MOVEit's per-IP session table).
+        ssh_handle = getattr(sftp, "_cyclone_ssh", None)
+        if ssh_handle is not None:
+            try:
+                ssh_handle.close()
+            except Exception:  # noqa: BLE001
+                pass
 
     # 4. Audit event — same shape the legacy resubmit_rejected_claims CLI
     # uses (event_type="clearhouse.submitted", entity_type="claim_file",
@@ -161,4 +222,4 @@ def submit_file(
             "submit_file %s: audit event failed: %s", file_label, exc,
         )
 
-    return SubmitResult(file_label, SubmitOutcome.SUBMITTED, batch_id=batch_id)
+    return SubmitResult(file_label, SubmitOutcome.SUBMITTED, batch_id=batch_id)

backend/src/cyclone/submission/result.py

@@ -10,6 +10,15 @@ from enum import Enum
 
 
 class SubmitOutcome(str, Enum):
+    """Outcome of a single submit_file invocation.
+
+    Audit-event invariant: ``SUBMITTED`` emits a ``clearhouse.submitted``
+    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.
+    """
     SUBMITTED = "submitted"
     SKIPPED = "skipped"
     PARSE_FAILED = "parse_failed"
@@ -20,7 +29,12 @@ class SubmitOutcome(str, Enum):
 
 @dataclass(frozen=True)
 class SubmitResult:
+    """Return value of ``submit_file``: filename, outcome, and optional
+    ``batch_id`` (populated when the DB write succeeded) and ``error``
+    (populated on any failure outcome; ``None`` on ``SUBMITTED`` /
+    ``SKIPPED``).
+    """
     file: str
     outcome: SubmitOutcome
     batch_id: str | None = None
-    error: str | None = None
+    error: str | None = None

backend/tests/test_submission.py

@@ -1,8 +1,8 @@
 """SP37 Task 4: submit_file owns parse → DB write → SFTP upload.
 
 Tests use a fake SFTP client to avoid hitting the real clearhouse.
-The DB path is real (autouse fixture) so the join-key wiring is
-exercised end-to-end.
+The DB path is real (autouse ``_auto_init_db`` fixture in conftest.py)
+so the join-key wiring is exercised end-to-end.
 """
 from __future__ import annotations
 
@@ -12,7 +12,6 @@ from unittest.mock import MagicMock
 import pytest
 
 from cyclone import db as db_mod
-from cyclone.clearhouse import SftpClient
 from cyclone.submission.core import submit_file
 from cyclone.submission.result import SubmitOutcome
 
@@ -48,14 +47,6 @@ class _FakeSftp:
         self.existing_files[remote_path] = len(content)
 
 
-@pytest.fixture(autouse=True)
-def _db(tmp_path, monkeypatch):
-    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
-    db_mod._reset_for_tests()
-    db_mod.init_db()
-    yield
-
-
 def test_submit_file_happy_path():
     """parse → DB write → SFTP upload, all steps succeed."""
     fake = _FakeSftp()
@@ -79,8 +70,14 @@ def test_submit_file_happy_path():
         assert s.get(db_mod.Batch, result.batch_id) is not None
 
 
-def test_submit_file_idempotent_db():
-    """Re-running submit_file on the same file is a no-op on the DB side."""
+def test_submit_file_skipped_when_remote_size_matches():
+    """Re-running submit_file on an already-uploaded file is SKIPPED.
+
+    The ``SKIPPED`` outcome is reached via the SFTP ``stat()`` check
+    (not via the DB) — same-size remote file ⇒ idempotency short-circuit.
+    Re-runs therefore don't emit a duplicate ``clearhouse.submitted``
+    audit event.
+    """
     fake = _FakeSftp()
     sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
     factory = lambda _b: fake
@@ -137,4 +134,62 @@ def test_submit_file_sftp_fail():
     # DB row was written before the SFTP call.
     assert result.batch_id is not None
     with db_mod.SessionLocal()() as s:
-        assert s.get(db_mod.Batch, result.batch_id) is not None
+        assert s.get(db_mod.Batch, result.batch_id) is not None
+
+
+def test_submit_file_payer_mismatch():
+    """PAYER_MISMATCH → no DB write, no SFTP call.
+
+    We monkeypatch ``parse_837_text`` to return a single claim whose
+    payer.id is ``OTHER_PAYER`` (the helper gates on the first
+    claim's payer — same posture as the legacy ``resubmit_rejected_claims``
+    CLI at cli.py:657-668).
+    """
+    from unittest.mock import patch
+
+    bad_claim = MagicMock()
+    bad_claim.payer.id = "OTHER_PAYER"
+    parsed_with_bad_payer = MagicMock()
+    parsed_with_bad_payer.claims = [bad_claim]
+    parsed_with_bad_payer.parsed_at = None
+    parsed_with_bad_payer.envelope.transaction_set_control_number = "ST123"
+
+    fake = _FakeSftp()
+    sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
+
+    with patch("cyclone.submission.core.parse_837_text", return_value=parsed_with_bad_payer):
+        result = submit_file(
+            _FIXTURE,
+            sftp_block=sftp_block,
+            actor="t",
+            validate=True,
+            sftp_client_factory=lambda _b: fake,
+        )
+    assert result.outcome == SubmitOutcome.PAYER_MISMATCH
+    assert "OTHER_PAYER" in (result.error or "")
+    assert len(fake.put_calls) == 0
+
+
+def test_submit_file_uses_default_paramiko_factory(monkeypatch):
+    """When no factory is supplied, helper should use paramiko directly
+    (not SftpClient wrapper, which lacks stat()).
+    """
+    fake_paramiko_sftp = _FakeSftp()
+    monkeypatch.setattr(
+        "cyclone.submission.core._default_sftp_factory",
+        lambda _block: fake_paramiko_sftp,
+    )
+
+    sftp_block = MagicMock()
+    sftp_block.stub = False
+    sftp_block.paths = {"outbound": "/ToHPE"}
+
+    result = submit_file(
+        _FIXTURE,
+        sftp_block=sftp_block,
+        actor="t",
+        validate=True,
+        # NO sftp_client_factory — should use the paramiko default
+    )
+    assert result.outcome == SubmitOutcome.SUBMITTED
+    assert len(fake_paramiko_sftp.put_calls) == 1