feat(sp41): wire claim-id dedup into submit_file pre-flight

Inserts a per-claim-id check_duplicate() walk into submit_file between
the payer-mismatch guard and the BatchRecord837 DB write (Task 9).

If any CLM01 in the parsed 837P file was submitted within the 30-day
window recorded in the submission_dedup table, submit_file raises
DuplicateClaimError before any DB write or SFTP upload. The DB-first,
upload-second invariant is preserved: a duplicate blocks both the
BatchRecord837 write AND the paramiko upload.

The api_routers/submission.submit_batch per-file loop special-cases
DuplicateClaimError BEFORE the generic Exception handler so the
200-with-results status-code contract documented at the top of the
module is preserved (per-file failures stay in the body). The
exception remains 409-class by design — a future caller that
propagates it past the per-file loop will surface 409 at the FastAPI
layer.

Approach A per the Task 9 spec (raise, don't return a SubmitResult).

Tests:
- test_submission_dedup.test_submit_file_raises_on_duplicate_claim_id
  pre-populates the dedup table via record_submission('CLM001', ...),
  calls submit_file against minimal_837p.txt, expects DuplicateClaimError
  with structured claim_id / original_submission_at attributes, and
  pins the DB-write invariant (no Batch row landed) plus an
  AssertionError sentinel on the SFTP factory.

backend/src/cyclone/api_routers/submission.py

@@ -33,6 +33,7 @@ from pydantic import BaseModel, ConfigDict, Field
 
 from cyclone.auth.deps import matrix_gate
 from cyclone.store import store as cycl_store
+from cyclone.store.exceptions import DuplicateClaimError
 from cyclone.submission.core import submit_file
 from cyclone.submission.result import SubmitOutcome, SubmitResult
 
@@ -154,6 +155,32 @@ def submit_batch(body: SubmitBatchRequest):
                 # monkey-patch submit_file itself instead of wiring a
                 # factory here.
             )
+        except DuplicateClaimError as exc:
+            # SP41 Task 9: 409-class domain exception (see the exception
+            # docstring on ``DuplicateClaimError``). The per-file loop
+            # preserves the 200-with-results status-code contract
+            # documented at the top of this module, so we surface the
+            # duplicate as a per-file failure with UNEXPECTED_ERROR
+            # outcome — the structured ``claim_id`` / ``original_submission_at``
+            # attributes ride along in the error string so the operator
+            # can see which CLM01 in the batch tripped the guard. Caught
+            # BEFORE the generic ``Exception`` handler so a future
+            # caller propagating the exception still gets 409-class
+            # treatment at the FastAPI layer.
+            log.warning(
+                "submit-batch duplicate claim on %s: claim_id=%r original=%s",
+                src.name, exc.claim_id, exc.original_submission_at,
+            )
+            r = SubmitResult(
+                file=src.name,
+                outcome=SubmitOutcome.UNEXPECTED_ERROR,
+                error=(
+                    f"DuplicateClaimError: claim_id {exc.claim_id!r} was "
+                    f"already submitted at "
+                    f"{exc.original_submission_at.isoformat()}; "
+                    f"within 30-day window"
+                ),
+            )
         except Exception as exc:  # noqa: BLE001
             log.exception(
                 "submit-batch unexpected error on %s", src.name,

backend/src/cyclone/submission/core.py

@@ -30,6 +30,7 @@ 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
+from cyclone.store.submission_dedup import check_duplicate
 
 from .result import SubmitOutcome, SubmitResult
 
@@ -137,6 +138,28 @@ def submit_file(
                 error=f"payer.id={mismatch.payer.id!r} (expected {EXPECTED_PAYER_ID!r})",
             )
 
+        # 1b. SP41 Task 9 — claim-id dedup pre-flight. Walks every
+        # CLM01 in the parsed file and asks ``check_duplicate`` whether
+        # any of them were submitted within the 30-day window. If so,
+        # raise ``DuplicateClaimError`` — this is a 409-class domain
+        # exception (mirrors the posture of ``AlreadyMatchedError`` /
+        # ``NotMatchedError`` / ``InvalidStateError`` per their docstrings)
+        # and is caught by ``api_routers/submission.submit_batch`` so
+        # the per-file 200-with-results contract is preserved.
+        #
+        # ``check_duplicate`` reads the configured ``cycl_db.SessionLocal()``
+        # — the same engine the BatchRecord837 DB write below uses — so
+        # we don't need to thread a session through here. The helper's
+        # optional ``db_url`` kwarg is ignored (signature-preserving
+        # shim for back-compat with the original Task 8 callers).
+        #
+        # We deliberately RAISE here instead of returning a SubmitResult:
+        # the exception carries ``claim_id`` + ``original_submission_at``
+        # so the caller can surface structured detail to the operator
+        # without re-querying. Approach A per the Task 9 spec.
+        for claim in parsed.claims:
+            check_duplicate(claim.claim_id)
+
     # 2. DB write — DB-first, upload-second invariant.
     # ``parsed`` is required to construct a BatchRecord837 (it embeds the
     # full ParseResult). validate=False therefore isn't a real path —

backend/tests/test_submission_dedup.py

@@ -1,5 +1,11 @@
 """Claim-id dedup at the SFTP pre-flight stage."""
-from datetime import datetime
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from cyclone.submission.core import submit_file
 from cyclone.store.exceptions import DuplicateClaimError
 from cyclone.store.submission_dedup import (
     record_submission,
@@ -7,6 +13,12 @@ from cyclone.store.submission_dedup import (
     DEFAULT_WINDOW_DAYS,
 )
 
+# Fixture file with a single CLM01 = "CLM001" claim (the minimal_837p
+# fixture is the source of truth; duplicated into submit-batch/single-claim.x12
+# but both byte-equal so we pick the canonical minimal path here).
+_FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt"
+_FIXTURE_CLAIM_ID = "CLM001"
+
 
 def test_check_duplicate_returns_none_for_first_submission(tmp_path):
     db = tmp_path / "test.db"
@@ -36,3 +48,73 @@ def test_check_duplicate_passes_after_window(tmp_path):
 
 def test_default_window_is_30_days():
     assert DEFAULT_WINDOW_DAYS == 30
+
+
+# --------------------------------------------------------------------------- #
+# SP41 Task 9 integration: dedup guard wired into submit_file pre-flight.
+# --------------------------------------------------------------------------- #
+
+
+def test_submit_file_raises_on_duplicate_claim_id(tmp_path):
+    """``submit_file`` raises ``DuplicateClaimError`` if any CLM01 in the
+    file was submitted within the 30-day dedup window, and the DB write
+    + SFTP factory are both skipped.
+
+    Mirrors the DB-first, upload-second invariant from ``submit_file``:
+    the dedup check sits between the payer-mismatch guard and the
+    BatchRecord837 DB write, so a duplicate blocks both the DB write
+    AND the SFTP upload. This test pins both invariants.
+
+    The conftest autouse fixture wires a fresh per-test SQLite DB at
+    ``tmp_path/test.db`` via the ``CYCLONE_DB_URL`` env var (so
+    ``cycl_db.SessionLocal()`` points there) — same engine the Batch
+    write would use, same engine the dedup helper reads.
+    """
+    # 1. Pre-populate the dedup table with the claim_id from the
+    #    fixture at a timestamp within the 30-day window. ``db_url``
+    #    is ignored inside the helper (signature-preserving shim) but
+    #    the argument is required by the current Task 8 signature, so
+    #    we pass the per-test DB URL for clarity.
+    record_submission(
+        _FIXTURE_CLAIM_ID,
+        submitted_at=datetime.now(timezone.utc),
+        db_url=f"sqlite:///{tmp_path}/test.db",
+    )
+
+    # 2. Sentinel: if submit_file reaches the SFTP factory call,
+    #    fail loudly. ``submit_file`` raises DuplicateClaimError
+    #    BEFORE the factory is invoked, so the AssertionError here
+    #    only fires if the dedup guard is bypassed (regression).
+    def _sftp_factory(_block):  # pragma: no cover - regression only
+        raise AssertionError(
+            "SFTP factory must not be called when dedup blocks submit_file"
+        )
+
+    sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
+
+    # 3. Expect the 409-class domain exception with structured
+    #    ``claim_id`` / ``original_submission_at`` attributes so the
+    #    operator can see which CLM01 tripped the guard.
+    with pytest.raises(DuplicateClaimError) as exc_info:
+        submit_file(
+            _FIXTURE,
+            sftp_block=sftp_block,
+            actor="test",
+            validate=True,
+            sftp_client_factory=_sftp_factory,
+        )
+    assert exc_info.value.claim_id == _FIXTURE_CLAIM_ID
+    assert exc_info.value.original_submission_at is not None
+
+    # 4. DB invariant: no Batch row was written (dedup blocks the
+    #    DB write, not just the SFTP upload). The autouse fixture
+    #    points cycl_db at tmp_path/test.db, so a fresh
+    #    ``db_mod.SessionLocal()()`` re-opens it.
+    from cyclone import db as db_mod
+
+    with db_mod.SessionLocal()() as session:
+        batch_count = session.query(db_mod.Batch).count()
+    assert batch_count == 0, (
+        "dedup guard must block the BatchRecord837 DB write; "
+        f"found {batch_count} Batch row(s)"
+    )