merge: SP37 canonical submit-batch flow into main

12-commit feature branch delivering:
  * Migration 0020: add batches.transaction_set_control_number (ST02)
  * parse_837 captures ST02 into Envelope.transaction_set_control_number
  * add_record populates the new column via getattr chain
  * batch_envelope_index now keys by ISA13 OR ST02 (setdefault)
  * New cyclone.submission package with submit_file helper
    (DB-first / upload-second invariant; paramiko direct factory
    because SftpClient wrapper lacks stat())
  * New cyclone submit-batch CLI + POST /api/submit-batch endpoint
    (thin wrappers over submit_file, byte-for-byte walker parity)
  * 31 new tests across 5 files (8 + 11 + 4 + 3 + 5)
  * Live-verified end-to-end: 999 with AK2*837*991102977~ now
    resolves to claim_id=CLM001 via the new join key (was orphan)

End-to-end orphan reduction going forward: every batch ingested via
the canonical submit path captures ST02; 999 acks referencing that
ST02 resolve via the updated batch_envelope_index.

Spec: docs/superpowers/specs/2026-07-07-cyclone-submit-batch-canonical-flow-design.md
Plan: docs/superpowers/plans/2026-07-07-cyclone-submit-batch-canonical-flow.md
Tracker: /tmp/refactor-cyclone.md

Reviewed-by: pr-reviewer subagent (PASS, all 8 checklist items)
Live-tested-by: end-to-end 999 POST against real dev DB
               (/tmp/sp37-task7-live-evidence.txt)
Merge-shape: --no-ff, single atomic commit (per cyclone-spec skill)

.env.example

@@ -17,4 +17,24 @@ VITE_API_BASE_URL=
 
 # Optional. Set to 1 to disable auth entirely (DEV ONLY). When set,
 # the backend auto-grants admin access without checking credentials.
-# CYCLONE_AUTH_DISABLED=0
+# CYCLONE_AUTH_DISABLED=0
+
+# ─── Gainwell / HCPF SFTP credentials (operator convention) ─────────────
+# These mirror what /home/tyler/EDI/scripts/upload_claims.sh exports and
+# what the operator's SFTP client uses. Cyclone's secrets module looks up
+# CYCLONE_SFTP_PASSWORD (or _FILE), not GAINWELL_SFTP_PASS — to bridge:
+#
+#   export CYCLONE_SFTP_PASSWORD="$GAINWELL_SFTP_PASS"
+#
+# or for the daemon, set CYCLONE_SFTP_PASSWORD_FILE to a 0600 file.
+GAINWELL_SFTP_USER=colorado-fts\coxix_prod_11525703
+GAINWELL_SFTP_HOST=mft.gainwelltechnologies.com
+GAINWELL_SFTP_PASS=
+GAINWELL_REMOTE_DIR=/CO XIX/PROD/coxix_prod_11525703/ToHPE
+
+# CYCLONE_SFTP_PASSWORD=                 # mirror of GAINWELL_SFTP_PASS
+# CYCLONE_SFTP_PASSWORD_FILE=            # path to a 0600 file containing it
+
+# CYCLONE_SCHEDULER_AUTOSTART=           # 1/true/yes to start the MFT poll loop on API launch
+# CYCLONE_SCHEDULER_POLL_SECONDS=60      # poll interval when autostart is on
+# CYCLONE_TAIL_HEARTBEAT_S=15            # NDJSON stream heartbeat interval

.gitignore

@@ -20,6 +20,16 @@ build/
 *.swp
 *.swo
 
+# macOS extraction residue (AppleDouble / __MACOSX from unzip on macOS).
+# These are paired with every regular file in an extracted zip — never data.
+__MACOSX/
+._*
+
+# Operator drop zone (untracked working dir; contents are HCPF-delivered
+# inbound files, never source). Use `mkdir -p ingest && touch ingest/.gitkeep`
+# if you want the directory itself in the repo.
+ingest/
+
 # Local config
 .env
 .env.local

CLAUDE.md

@@ -125,7 +125,7 @@ Frontend triplet for any live page: `use<X>(params)` (initial fetch) + `useTailS
 
 ## Backend at a glance
 
-`backend/src/cyclone/` is a single namespace. The two largest files are `api.py` (~3,548 LOC, the only large file) and `store.py` (~2,423 LOC, the `CycloneStore` facade — SP21 is in flight to split it into a `cyclone/store/` subpackage; the public API stays unchanged). Subpackages: `api_routers/` (acks, admin, health, ta1_acks), `clearhouse/` (Clearhouse + SftpClient), `edi/` (filenames), `parsers/` (X12 transaction parsers + models + validators + serializers), `workflow/` (placeholder for future sub-project 6).
+`backend/src/cyclone/` is a single namespace. The two largest files are `api.py` (~3,548 LOC, the only large file) and `store.py` (~2,423 LOC, the `CycloneStore` facade — SP21 is in flight to split it into a `cyclone/store/` subpackage; the public API stays unchanged). Subpackages: `api_routers/` (acks, admin, health, ta1_acks), `clearhouse/` (Clearhouse + SftpClient), `edi/` (filenames), `parsers/` (X12 transaction parsers + models + validators + serializers), `submission/` (SP37 — canonical `submit_file` helper shared by `cyclone submit-batch` CLI + `POST /api/submit-batch` HTTP endpoint; owns parse → DB-write → SFTP-upload → audit per file), `workflow/` (placeholder for future sub-project 6).
 
 The store is the only read/write surface for the database; every mutating endpoint goes through it. All persistence flows through SQLAlchemy sessions via `db.SessionLocal()()`. SQLAlchemy ORM models live in `db.py`; 12 SQL migrations under `migrations/` (0001_initial through 0012_backups) are walked in order by `db_migrate.py`.
 
@@ -137,6 +137,34 @@ Config: `config/payers.yaml` is the on-disk source for providers / payers / clea
 
 Secrets live in the macOS Keychain (via `keyring` + `cyclone.secrets`): SQLCipher key (service `cyclone`, account `cyclone.db.key`), SFTP password, backup passphrase. No secrets on disk in plaintext.
 
+## Production SFTP posture (this box: manual mode)
+
+This host runs in **manual SFTP mode** against Gainwell's MOVEit Transfer MFT at `mft.gainwelltechnologies.com`. The seeded `dzinesco` clearhouse keeps `sftp_block.stub: true`; the operator moves files to/from Gainwell with their SFTP client and drops inbound files into `ingest/` for cyclone to ingest. Do NOT flip `stub` to `false` from this host — see "Auth caveat" below.
+
+### Env var convention
+
+The operator's shell exports the Gainwell creds as `GAINWELL_SFTP_USER`, `GAINWELL_SFTP_PASS`, `GAINWELL_SFTP_HOST`, `GAINWELL_REMOTE_DIR`. Cyclone's `secrets.get_secret()` looks up `CYCLONE_SFTP_PASSWORD` (or `CYCLONE_SFTP_PASSWORD_FILE`) per `secrets._ENV_NAME_FOR`. The two are NOT the same name — bridge them per-call or in your shell rc:
+
+```bash
+export CYCLONE_SFTP_PASSWORD="$GAINWELL_SFTP_PASS"
+```
+
+### Inbound drop zone
+
+`/home/tyler/dev/cyclone/ingest/` is the operator-maintained staging dir for inbound MFT files (999 acks, TA1, 835 remittances, 277CA claim acks). Files here are NOT auto-watched; processing happens via the procedure in `docs/RUNBOOK.md` § "Manual SFTP mode". At time of writing this dir holds ~1500 unprocessed 999 acks + 1 TA1 + 1 835 from early July 2026 — the local ingest flow clears them.
+
+### Auth caveat (do not retry)
+
+Paramiko reaches `MOVEit Transfer SFTP` cleanly (SSH kex completes, MOVEit offers password auth) but `AuthenticationException: Authentication failed` is returned despite the operator confirming the credentials are known-good from another host. Most likely cause: MOVEit Transfer's IP-based access control — this host's public IP `103.14.26.95` is not whitelisted. Repeated auth attempts risk account lockout. Do NOT keep guessing. Resolve by either: (a) whitelisting this IP with Gainwell's MFT admin, or (b) staying in manual mode and moving files via the operator's SFTP client.
+
+### Inbound ingestion paths
+
+The canonical way to write 999/TA1/277CA/835 rows into the DB is `Scheduler.process_inbound_files`, exposed as the CLI `cyclone pull-inbound --date YYYYMMDD` and the HTTP `POST /api/admin/scheduler/pull-inbound`. **There is no `parse-999` / `parse-ta1` / `parse-277ca` CLI command** (CLAUDE.md's "CLI" section above is aspirational on those three). The `parse-837` and `parse-835` CLIs exist but only emit JSON files to `--output-dir`; they do NOT write to the DB. For DB writes, use `pull-inbound` (which dedupes via `processed_inbound_files`).
+
+### Daemon hot-reload
+
+`python -m cyclone serve` runs as root (started by `tini`) and keeps the SFTP block in memory. Flipping `stub` directly in the DB (`store.update_clearhouse(...)`) does NOT auto-reload the daemon's view — either restart the daemon or use `PATCH /api/clearhouse` (which calls `scheduler.reconfigure_scheduler` to hot-reload).
+
 ## Frontend at a glance
 
 `src/` is React 18 + TypeScript + Vite. Routes register in `src/App.tsx` (11 pages, all under a `<Layout>` route wrapper). Pages are pure renderers — every page pairs with a `use<X>` data hook in `src/hooks/` and renders a `<PageHeader>` + a table/list/KPI grid. Drawers (`ClaimDrawer/`, `RemitDrawer/`, plus the new `ProviderDrawer/` and `AckDrawer/`) are mounted by the page and their open/close state is mirrored to the URL via `useDrawerUrlState` so deep-links round-trip. Drill-stack navigation is provided by `<DrillStackProvider>` in `src/components/drill/`.

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,186 @@
+"""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.store import store as cycl_store
+from cyclone.submission.core import submit_file
+from cyclone.submission.result import SubmitOutcome, SubmitResult
+
+log = logging.getLogger(__name__)
+
+# No `prefix=` here — every other gated router in this package declares
+# the full path in the decorator (clearhouse.py uses "/api/clearhouse",
+# parse.py uses "/api/parse-837", etc.). The decorator sets the full URL.
+router = APIRouter(
+    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("/api/submit-batch")
+def 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.
+    """
+
+    # 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/src/cyclone/cli.py

@@ -768,6 +768,99 @@ def resubmit_rejected_claims(
     )
 
 
+@main.command("submit-batch")
+@click.option("--ingest-dir", default="ingest",
+              show_default=True,
+              help="Root dir holding batch-*-claims/ subfolders of .x12 files.")
+@click.option("--actor", default="cli-submit-batch",
+              show_default=True,
+              help="Audit-log actor tag for the clearhouse.submitted events.")
+@click.option("--validate/--no-validate", default=True,
+              help="Parse each file via parse_837 before upload.")
+@click.option("--limit", type=int, default=None,
+              help="Stop after N files (smoke-tests).")
+def submit_batch(
+    ingest_dir: str,
+    actor: str,
+    validate: bool,
+    limit: int | None,
+) -> None:
+    """Parse → DB-write → SFTP-upload each batch-*-claims/*.x12 (SP37).
+
+    Canonical outbound path. Mirrors ``resubmit-rejected-claims`` but
+    routes through ``cyclone.submission.submit_file``, which writes to
+    the DB before uploading. Use this as the preferred outbound path;
+    ``resubmit-rejected-claims`` remains for one-off cases where you
+    don't want a DB row (dzinesco-generated fixes not ready for
+    canonical tracking).
+
+    Exit codes: 0 = run completed (even with per-file failures), 1 =
+    unexpected exception, 2 = config-level failure (no clearhouse,
+    SFTP block in stub mode, ingest-dir missing).
+    """
+    from cyclone import db as db_mod
+    from cyclone.submission.core import submit_file
+    from cyclone.submission.result import SubmitOutcome
+    from cyclone.store import store as cycl_store
+
+    db_mod.init_db()
+    cycl_store.ensure_clearhouse_seeded()
+    clearhouse = cycl_store.get_clearhouse()
+    if clearhouse is None:
+        click.echo("No clearhouse seeded; cannot resolve SFTP block.", err=True)
+        sys.exit(2)
+    sftp_block = clearhouse.sftp_block
+    if sftp_block.stub:
+        click.echo("Clearhouse SFTP block is in stub mode; refusing to upload.", err=True)
+        sys.exit(2)
+
+    root = Path(ingest_dir).resolve()
+    if not root.exists():
+        click.echo(f"--ingest-dir does not exist: {root}", err=True)
+        sys.exit(2)
+
+    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("._")))
+    if not files:
+        click.echo(f"No .x12 files under {root}", err=True)
+        sys.exit(2)
+
+    click.echo(f"found {len(files)} files under {root}")
+
+    submitted = 0
+    skipped = 0
+    failed = 0
+    for i, src in enumerate(files, 1):
+        if limit is not None and i > limit:
+            break
+        try:
+            result = submit_file(
+                src, sftp_block=sftp_block, actor=actor, validate=validate,
+            )
+        except Exception as exc:  # noqa: BLE001
+            click.echo(f"UNEXPECTED  {src.name}: {exc.__class__.__name__}: {exc}", err=True)
+            failed += 1
+            continue
+        if result.outcome == SubmitOutcome.SUBMITTED:
+            submitted += 1
+            click.echo(f"submitted  {src.name}  batch={result.batch_id}")
+        elif result.outcome == SubmitOutcome.SKIPPED:
+            skipped += 1
+            click.echo(f"skipped    {src.name}  (already uploaded)")
+        else:
+            failed += 1
+            click.echo(
+                f"{result.outcome.value:<10}  {src.name}  {result.error or ''}",
+                err=True,
+            )
+
+    click.echo(f"\nsummary: submitted={submitted} skipped={skipped} failed={failed}")
+    # Per-file failures don't bump exit code; details in stdout.
+    # (Per cyclone-cli convention.)
+
+
 @main.group()
 def backup() -> None:
     """Encrypted DB backup management (SP17)."""

backend/src/cyclone/db.py

@@ -221,6 +221,13 @@ class Batch(Base):
     totals_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
     validation_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
     raw_result_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
+    # SP37 Task 2: source 837's ST02 (transaction set control number).
+    # Populated from ``Envelope.transaction_set_control_number`` by
+    # ``store.write.add_record`` for 837P batches; NULL for 835 batches
+    # (the column is an 837P-specific join key for 999 AK2 resolution).
+    # Migration 0020 adds the column additively; no backfill required for
+    # pre-existing rows that lack the value.
+    transaction_set_control_number: Mapped[Optional[str]] = mapped_column(String(32), nullable=True)
 
     claims: Mapped[list["Claim"]] = relationship(
         back_populates="batch", cascade="all, delete-orphan"

backend/src/cyclone/migrations/0020_add_batch_txn_set_control_number.sql

@@ -0,0 +1,27 @@
+-- version: 20
+-- SP37: Batch.transaction_set_control_number = parsed 837's ST02.
+--
+-- Today's 999 ack join (claim_acks.batch_envelope_index, Pass 1) matches
+-- on ``Batch.envelope.control_number == 999's set_control_number``. That
+-- never resolves in production because 999's set_control_number (AK201)
+-- echoes the source 837's ST02 (transaction set control number), not the
+-- ISA13 (interchange control number) that Envelope.control_number stores.
+-- Result: every AK2 set-response against a dzinesco-generated 837 turns
+-- into an orphan.
+--
+-- SP37 fixes this by adding a column populated from the parsed 837's
+-- ST02 on every ``add_record`` write, then updating Pass 1 to match on
+-- it (Task 2). This migration is the additive part: nullable, no
+-- default, backfills from ``raw_result_json.envelope.transaction_set_control_number``
+-- for any pre-existing batch rows that already carry the value.
+--
+-- No new index (column is a primary join key, not a range query; the
+-- existing batches table is small enough for a full scan during the
+-- 999 join — see SP37 §"Migration 0013").
+
+ALTER TABLE batches ADD COLUMN transaction_set_control_number TEXT;
+
+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;

backend/src/cyclone/parsers/models.py

@@ -120,6 +120,13 @@ class Envelope(_Base):
     implementation_guide: str | None = None
     # SP3 P1 T2: BHT06 transaction type code (was: transaction_set_purpose_code, which is BHT02).
     transaction_type_code: str | None = None
+    # SP37 Task 2: X12 ST02 (transaction set control number). Distinct
+    # from ``control_number`` above, which is the ISA13 interchange
+    # control number. 999 acks echo ST02 back as AK201, so this is the
+    # join key that lets ``add_record``'s batch row round-trip back to
+    # its source 837. Populated only by the 837P parser today; other
+    # parsers share this class but leave the field None.
+    transaction_set_control_number: str | None = None
 
 
 class BatchSummary(_Base):

backend/src/cyclone/parsers/parse_837.py

@@ -83,6 +83,12 @@ def _build_envelope(segments: list[list[str]], input_file: str = "") -> tuple[En
             except (IndexError, ValueError) as exc:
                 log.warning("Could not parse BHT date: %s", exc)
         elif seg[0] == "ST" and envelope is not None:
+            # SP37 Task 2: capture ST02 (transaction set control number).
+            # 999 acks echo this back as AK201, so this is what makes the
+            # batch row joinable once the 999 ingests. Distinct from ISA13
+            # (which is already on ``control_number``).
+            if len(seg) > 2:
+                envelope = envelope.model_copy(update={"transaction_set_control_number": seg[2].strip()})
             if len(seg) > 3:
                 envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
     return envelope, summary

backend/src/cyclone/store/__init__.py

@@ -325,7 +325,13 @@ class CycloneStore:
         return _remove_claim_ack(link_id, event_bus=event_bus)
 
     def batch_envelope_index(self):
-        """Return a {envelope.control_number: batch.id} map for D10 Pass 1."""
+        """Return a {key: batch.id} map populated from two columns (D10 Pass 1).
+
+        Each 837p batch contributes both ``Batch.raw_result_json.envelope.control_number``
+        (ISA13) and ``Batch.transaction_set_control_number`` (ST02); 835 batches
+        are excluded. Single ``.get(set_control_number)`` lookup resolves
+        either key — SP37 closes the 999 AK201 → source-batch gap.
+        """
         return _batch_envelope_index()
 
     # -- SP17: encrypted DB backups -------------------------------------

backend/src/cyclone/store/claim_acks.py

@@ -11,8 +11,10 @@ Five methods on top of the new ``ClaimAck`` ORM table:
   ack-orphans lane).
 * ``remove_claim_ack`` — unlink. Publishes ``claim_ack_dropped``.
 * ``batch_envelope_index`` — D10 in-memory map of
-  ``Batch.envelope.control_number → batch.id`` (cheap to rebuild;
-  re-built once per ingest).
+  ``{Batch.raw_result_json.envelope.control_number (ISA13) OR
+  Batch.transaction_set_control_number (ST02)} → batch.id``
+  (SP37: populated from two columns; cheap to rebuild, re-built
+  once per ingest).
 
 Each mutating method opens its own ``db.SessionLocal()()`` session
 so callers don't have to manage session lifecycles. Publishes are
@@ -67,26 +69,43 @@ def _safe_publish(event_bus: "EventBus | None", kind: str, payload: dict) -> Non
 
 
 def batch_envelope_index() -> dict[str, str]:
-    """Build a ``{envelope.control_number: batch.id}`` map.
+    """Build a ``{key: batch.id}`` map populated from two columns.
 
-    D10 primary join key (spec §D10). Built once per ingest from
-    every Batch row whose ``raw_result_json`` carries an envelope —
-    cost is O(N batches), currently ~16, so trivial. Kept as a
-    plain dict so callers can ``index.get(scn)`` to resolve.
+    D10 primary join key (spec §D10). Each 837p batch row contributes
+    up to two entries:
+
+    * ``Batch.raw_result_json.envelope.control_number`` (ISA13)
+    * ``Batch.transaction_set_control_number`` (ST02, new in SP37)
+
+    Either key resolves to the same ``batch.id``; callers do a single
+    ``idx.get(set_control_number)`` lookup. Pass 2 (PCN) is unchanged.
+
+    Built once per ingest from every Batch row whose kind is "837p";
+    cost is O(N batches), currently small, so trivial. 835 batches
+    are excluded — the column is an 837P-specific join key (see
+    ``Batch.transaction_set_control_number`` docstring).
     """
-    out: dict[str, str] = {}
+    idx: dict[str, str] = {}
     with db.SessionLocal()() as s:
         rows = (
-            s.query(Batch.id, Batch.raw_result_json)
+            s.query(
+                Batch.id,
+                Batch.raw_result_json,
+                Batch.transaction_set_control_number,
+            )
             .filter(Batch.kind == "837p")
             .all()
         )
-        for bid, raw in rows:
+        for bid, raw, stcn in rows:
             env = (raw or {}).get("envelope") or {}
-            ctrl = env.get("control_number")
-            if isinstance(ctrl, str) and ctrl:
-                out[ctrl] = bid
-    return out
+            isa_cn = env.get("control_number")
+            if isinstance(isa_cn, str) and isa_cn:
+                # First write wins (setdefault) — if ISA13 and ST02
+                # ever collide they map to the same batch anyway.
+                idx.setdefault(isa_cn, bid)
+            if isinstance(stcn, str) and stcn:
+                idx.setdefault(stcn, bid)
+    return idx
 
 
 # ---------------------------------------------------------------------------

backend/src/cyclone/store/write.py

@@ -77,6 +77,17 @@ def add_record(record: BatchRecord, *, event_bus=None) -> None:
             totals_json=None,
             validation_json=None,
             raw_result_json=json.loads(record.result.model_dump_json()),
+            # SP37 Task 2: mirror the parsed 837's ST02 onto the batch
+            # row so 999 AK201 set_control_numbers can resolve back via
+            # Pass 1. The ``getattr`` chain handles the 835 path: the
+            # shared ``Envelope`` class is used by both 837P and 835
+            # parsers, but only ``parse_837`` populates this field — for
+            # 835 records it stays ``None`` and the column is NULL.
+            transaction_set_control_number=getattr(
+                getattr(record.result, "envelope", None),
+                "transaction_set_control_number",
+                None,
+            ),
         )
         s.add(batch_row)
 

backend/src/cyclone/submission/__init__.py

@@ -0,0 +1,17 @@
+"""SP37 Task 4: canonical 837P submission flow.
+
+The single public helper is ``submit_file`` — it owns parse → DB
+write → SFTP upload per file. CLI (``cyclone submit-batch``) and HTTP
+(``POST /api/submit-batch``) are thin wrappers; both call this helper
+so the ordering, idempotency, and audit shape are identical.
+
+DB-first, upload-second (per spec decision §2): if the DB write fails,
+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).
+"""
+from .core import submit_file
+from .result import SubmitOutcome, SubmitResult
+
+__all__ = ["submit_file", "SubmitOutcome", "SubmitResult"]

backend/src/cyclone/submission/core.py

@@ -0,0 +1,225 @@
+"""SP37 Task 4: parse → DB write → SFTP upload, per file.
+
+Mirrors ``resubmit_rejected_claims`` but writes to the DB before
+uploading. Same idempotency check (``stat().st_size == local_size``)
+skips already-uploaded files without re-emitting audit events.
+
+DB-first, upload-second (per spec decision §2): if the DB write fails,
+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
+
+import logging
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+
+from cyclone import db as db_mod
+from cyclone.audit_log import AuditEvent, append_event
+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
+
+from .result import SubmitOutcome, SubmitResult
+
+log = logging.getLogger(__name__)
+
+# The companion-guide payer id we enforce for CO Medicaid submits.
+# Same value the legacy ``resubmit-rejected-claims`` CLI gates on.
+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: SftpBlock,
+    actor: str,
+    validate: bool = True,
+    sftp_client_factory: Callable[[SftpBlock], Any] | None = None,
+) -> SubmitResult:
+    """Submit one 837P file: parse → DB write → SFTP upload.
+
+    Args:
+        path: local 837P file.
+        sftp_block: the clearhouse SftpBlock (for paths + auth).
+        actor: audit-log actor tag (e.g. "api-submit-batch").
+        validate: parse the file before upload (default True). Catches
+            bad byte-fixes early.
+        sftp_client_factory: optional callable that returns an
+            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),
+        error (when failed).
+    """
+    file_label = path.name
+    try:
+        content = path.read_bytes()
+    except OSError as exc:
+        return SubmitResult(file_label, SubmitOutcome.PARSE_FAILED, error=str(exc))
+
+    # 1. Validate via parse (optional but recommended).
+    parsed: Any = None
+    if validate:
+        try:
+            parsed = parse_837_text(content.decode(), PayerConfig.co_medicaid())
+        except Exception as exc:  # noqa: BLE001
+            log.warning("submit_file %s: parse failed: %s", file_label, exc)
+            return SubmitResult(file_label, SubmitOutcome.PARSE_FAILED, error=str(exc))
+        mismatch = next(
+            (c for c in parsed.claims if c.payer.id != EXPECTED_PAYER_ID),
+            None,
+        )
+        if mismatch is not None:
+            return SubmitResult(
+                file_label,
+                SubmitOutcome.PAYER_MISMATCH,
+                error=f"payer.id={mismatch.payer.id!r} (expected {EXPECTED_PAYER_ID!r})",
+            )
+
+    # 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 —
+    # the CLI/HTTP always pass validate=True — so reject it loudly
+    # rather than silently building an empty row.
+    if parsed is None:
+        return SubmitResult(
+            file_label,
+            SubmitOutcome.PARSE_FAILED,
+            error="validate=False is not supported; must parse to construct a BatchRecord",
+        )
+
+    # Use the same uuid4().hex id the existing /api/parse-837 path uses.
+    batch_id = uuid.uuid4().hex
+    record = BatchRecord837(
+        id=batch_id,
+        input_filename=file_label,
+        parsed_at=datetime.now(timezone.utc),
+        result=parsed,
+    )
+
+    try:
+        # ``cycl_store.add`` is the public facade method (per the
+        # CycloneStore class in store/__init__.py:158). It delegates
+        # to ``store.write.add_record``, which is the underlying
+        # SQLAlchemy write path.
+        cycl_store.add(record)
+    except Exception as exc:  # noqa: BLE001
+        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. ``_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 _default_sftp_factory
+    sftp = factory(sftp_block)
+    local_size = len(content)
+    try:
+        try:
+            stat = sftp.stat(remote_path)
+            if stat.st_size == local_size:
+                # Already on remote at the right size — re-run is safe;
+                # do NOT emit a duplicate audit event.
+                return SubmitResult(file_label, SubmitOutcome.SKIPPED, batch_id=batch_id)
+        except (IOError, OSError):
+            pass  # not on remote yet
+        sftp.write_file(remote_path, content)
+    except Exception as exc:  # noqa: BLE001
+        log.warning("submit_file %s: SFTP failed: %s", file_label, exc)
+        return SubmitResult(
+            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",
+    # entity_id=filename, payload has remote_path + source + size).
+    # Best-effort: an audit failure must not roll back the upload.
+    try:
+        with db_mod.SessionLocal()() as session:
+            append_event(session, AuditEvent(
+                event_type="clearhouse.submitted",
+                entity_type="claim_file",
+                entity_id=file_label,
+                payload={
+                    "remote_path": remote_path,
+                    "source": "submit-batch",
+                    "size": local_size,
+                    "batch_id": batch_id,
+                },
+                actor=actor,
+            ))
+            session.commit()
+    except Exception as exc:  # noqa: BLE001
+        log.warning(
+            "submit_file %s: audit event failed: %s", file_label, exc,
+        )
+
+    return SubmitResult(file_label, SubmitOutcome.SUBMITTED, batch_id=batch_id)

backend/src/cyclone/submission/result.py

@@ -0,0 +1,40 @@
+"""SP37 Task 4: typed result for submit_file.
+
+The CLI and HTTP layer both consume SubmitResult; keeping it in one
+place means both surfaces agree on the response shape.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+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"
+    PAYER_MISMATCH = "payer_mismatch"
+    DB_FAILED = "db_failed"
+    SFTP_FAILED = "sftp_failed"
+
+
+@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

backend/tests/fixtures/submit-batch/single-claim.x12

@@ -0,0 +1,30 @@
+ISA*00*          *00*          *ZZ*11525703       *ZZ*COMEDASSISTPROG*260611*0814*^*00501*991102977*1*P*:~
+GS*HC*11525703*COMEDASSISTPROG*20260611*081417*991102977*X*005010X222A1~
+ST*837*991102977*005010X222A1~
+BHT*0019*00*ref-001*20260611*081417*CH~
+NM1*41*2*Test Submitter*****46*11525703~
+PER*IC*Test Contact*EM*test@example.com~
+NM1*40*2*COLORADO MEDICAL ASSISTANCE PROGRAM*****46*COMEDASSISTPROG~
+HL*1**20*1~
+PRV*BI*PXC*251E00000X~
+NM1*85*2*Test Provider Inc*****XX*1993999998~
+N3*123 Test St~
+N4*Denver*CO*80202~
+REF*EI*123456789~
+HL*2*1*22*0~
+SBR*P*18*******MC~
+NM1*IL*1*Doe*John****MI*ABC123~
+N3*456 Member St~
+N4*Denver*CO*80203~
+DMG*D8*19800101*M~
+NM1*PR*2*CO_TXIX*****PI*CO_TXIX~
+CLM*CLM001*100.00***12:B:1*Y*A*Y*Y~
+REF*G1*PA123~
+HI*ABK:Z00~
+LX*1~
+SV1*HC:99213*100.00*UN*1***1~
+DTP*472*D8*20260611~
+REF*6R*REF001~
+SE*26*991102977~
+GE*1*991102977~
+IEA*1*991102977~

backend/tests/test_acks.py

@@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table():
 
 def test_migration_latest_idempotent_on_fresh_db():
     """Re-running the migration on the same DB must be a no-op (PRAGMA
-    user_version already at the latest version — currently 19 after
+    user_version already at the latest version — currently 20 after
     0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
     providers/payers/clearhouse, SP10's 0008 payer_rejected,
     SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
@@ -61,15 +61,16 @@ def test_migration_latest_idempotent_on_fresh_db():
     claims.matched_remittance_id index, SP27-Task 17's 0017
     claim.patient_control_number backfill UPDATE, SP28's 0018
     claim_acks join table, SP32's 0019
-    rendering_provider_npi + service_provider_npi)."""
+    rendering_provider_npi + service_provider_npi, and SP37's 0020
+    transaction_set_control_number)."""
     with db.engine().begin() as c:
         v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v1 == 19
+    assert v1 == 20
     # A second run should not raise and should not bump the version.
     db_migrate.run(db.engine())
     with db.engine().begin() as c:
         v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v2 == 19
+    assert v2 == 20
 
 
 def test_add_ack_persists_row():

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

backend/tests/test_batch_envelope_index_sp37.py

@@ -0,0 +1,126 @@
+"""SP37 Task 3: batch_envelope_index populates from BOTH columns.
+
+The dict returned by ``batch_envelope_index`` should resolve lookups
+by EITHER ``Batch.raw_result_json.envelope.control_number`` (ISA13,
+preserved) OR ``Batch.transaction_set_control_number`` (ST02, new in
+SP37 Task 2). One dict, both keys — a single ``.get(set_control_number)``
+call hits whichever matches, so the D10 Pass 1 join in
+:func:`cyclone.claim_acks.lookup_claims_for_ack_set_response` resolves
+999 AK201 (which echoes the source 837's ST02) back to the right batch
+even when ST02 != ISA13.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+from cyclone import db as db_mod
+from cyclone.db import Batch
+from cyclone.store.claim_acks import batch_envelope_index
+
+
+@pytest.fixture(autouse=True)
+def _db(tmp_path, monkeypatch):
+    """Per-test DB; conftest's autouse already wires one too, but we
+    pin the URL again so this module is self-contained if anyone ever
+    lifts it out of the conftest tree."""
+    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
+    db_mod._reset_for_tests()
+    db_mod.init_db()
+    yield
+
+
+def _make_batch(
+    s,
+    *,
+    id: str,
+    icn: str,
+    stcn: str | None = None,
+    raw_json: dict | None = None,
+) -> Batch:
+    """Insert one Batch row whose envelope + ST02 mirror what
+    ``store.write.add_record`` writes for an 837P batch."""
+    row = Batch(
+        id=id,
+        kind="837p",
+        input_filename=id + ".x12",
+        parsed_at=datetime.now(timezone.utc),
+        raw_result_json=raw_json or {"envelope": {"control_number": icn}},
+        transaction_set_control_number=stcn,
+    )
+    s.add(row)
+    s.commit()
+    s.refresh(row)
+    return row
+
+
+def test_index_resolves_by_envelope_control_number():
+    """Pre-SP37 path: ISA13 (envelope.control_number) still resolves."""
+    with db_mod.SessionLocal()() as s:
+        _make_batch(s, id="b1", icn="ISA000001")
+    idx = batch_envelope_index()
+    assert idx.get("ISA000001") == "b1"
+
+
+def test_index_resolves_by_transaction_set_control_number():
+    """SP37 path: ST02 (transaction_set_control_number) resolves too.
+
+    The whole point of Task 3 — Gainwell batches have ST02 != ISA13,
+    so 999 AK201 echoes ST02 and must hit this branch.
+    """
+    with db_mod.SessionLocal()() as s:
+        _make_batch(
+            s, id="b2", icn="ISA000002", stcn="ST000002",
+            raw_json={"envelope": {"control_number": "ISA000002"}},
+        )
+    idx = batch_envelope_index()
+    assert idx.get("ST000002") == "b2"
+    # Backward compat: ISA still resolves.
+    assert idx.get("ISA000002") == "b2"
+
+
+def test_index_handles_row_with_no_transaction_set_control_number():
+    """Pre-migration rows (stcn NULL) should still resolve by ISA.
+
+    Spec says: 'Pre-migration rows (stcn NULL) should still resolve by ISA.'
+    Production row ``b1`` (the only one with claims) has ST02 NULL because
+    the migration's backfill only ran over rows whose raw_result_json
+    envelope had the ST02 key — and that row was written before the
+    parser populated it.
+    """
+    with db_mod.SessionLocal()() as s:
+        _make_batch(s, id="b3", icn="ISA000003")  # no stcn
+    idx = batch_envelope_index()
+    assert idx.get("ISA000003") == "b3"
+    assert idx.get("ST000003") is None
+
+
+def test_index_ignores_non_837_batches():
+    """835 batches don't contribute to the 999 join index.
+
+    The D10 two-pass join is specifically for 837 → 999 linkage; 835
+    remittances live on the response side of the loop and have no ST02
+    join key. Filtering by ``kind == "837p"`` keeps the index scoped
+    correctly and avoids an 835 row shadowing an 837 ST02 by accident.
+    """
+    with db_mod.SessionLocal()() as s:
+        # Add an 837 row first so we can prove the 835 doesn't leak in.
+        _make_batch(s, id="b-837", icn="ISA837", stcn="ST837")
+        # And an 835 row whose ISA13 / ST02 would otherwise pollute.
+        s.add(Batch(
+            id="b-835",
+            kind="835",
+            input_filename="b-835.x12",
+            parsed_at=datetime.now(timezone.utc),
+            raw_result_json={"envelope": {"control_number": "ISA999"}},
+            transaction_set_control_number="ST999",
+        ))
+        s.commit()
+    idx = batch_envelope_index()
+    # 837 row resolves both ways.
+    assert idx.get("ISA837") == "b-837"
+    assert idx.get("ST837") == "b-837"
+    # 835 row contributes nothing.
+    assert "ISA999" not in idx
+    assert "ST999" not in idx

backend/tests/test_batch_txn_set_cn.py

@@ -0,0 +1,103 @@
+"""SP37 Task 2: add_record populates Batch.transaction_set_control_number.
+
+The column should mirror the source 837 envelope's
+``transaction_set_control_number`` (ST02) so the join-key update in
+Task 3 can resolve 999 ``set_control_number`` (AK201) values back to
+the right batch. The 835 path leaves the column NULL because the
+field is an 837P-specific join key (835 remittances don't need to
+resolve back to a source 837 — they're the response side of the loop).
+"""
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+from pathlib import Path
+
+from cyclone import db as db_mod
+from cyclone.db import Batch
+from cyclone.parsers.parse_837 import parse as parse_837_text
+from cyclone.parsers.payer import PayerConfig
+from cyclone.store import store as cycl_store
+from cyclone.store.records import BatchRecord837, BatchRecord835
+
+FIXTURE_837 = Path(__file__).parent / "fixtures" / "minimal_837p.txt"
+FIXTURE_835 = Path(__file__).parent / "fixtures" / "minimal_835.txt"
+
+
+def test_add_record_populates_transaction_set_control_number_for_837():
+    """parse_837 → add_record writes the parsed ST02 onto the Batch row.
+
+    The fixture's ST02 (``991102977``) must round-trip into the new
+    ``batches.transaction_set_control_number`` column. This is the
+    join key Task 3's 999-ingest Pass 1 update relies on.
+    """
+    text = FIXTURE_837.read_text()
+    parsed = parse_837_text(text, PayerConfig.co_medicaid())
+
+    record = BatchRecord837(
+        id="b-txn-cn-1",
+        input_filename="minimal_837p.txt",
+        parsed_at=datetime.now(timezone.utc),
+        result=parsed,
+    )
+    cycl_store.add(record)
+
+    with db_mod.SessionLocal()() as s:
+        row = s.get(Batch, "b-txn-cn-1")
+        assert row is not None
+        assert row.transaction_set_control_number == parsed.envelope.transaction_set_control_number
+        # Sanity: the fixture's ST02 must be the parsed-and-stored value.
+        assert row.transaction_set_control_number == "991102977"
+
+
+def test_envelope_model_exposes_transaction_set_control_number():
+    """The Envelope Pydantic model carries the new field with a None default.
+
+    Guards against a regression where someone removes the field from
+    the model — the rest of the chain (parser + ORM + write path)
+    silently degrades to None if the model loses the attribute.
+    """
+    from cyclone.parsers.models import Envelope
+
+    env = Envelope(
+        sender_id="S",
+        receiver_id="R",
+        control_number="000000001",
+        transaction_date=date(2026, 1, 1),
+    )
+    assert env.transaction_set_control_number is None
+
+    env2 = env.model_copy(update={"transaction_set_control_number": "0001"})
+    assert env2.transaction_set_control_number == "0001"
+
+
+def test_add_record_leaves_835_column_null():
+    """835 batches don't carry an ST02 join key; column stays NULL.
+
+    The shared ``Envelope`` class is used by both 837P and 835 parsers,
+    but only ``parse_837`` populates ``transaction_set_control_number``.
+    For 835 records the field is None and ``add_record`` writes NULL
+    to the column — verified end-to-end via a real 835 ingest.
+    """
+    text = FIXTURE_835.read_text()
+    from cyclone.parsers.parse_835 import parse as parse_835_text
+    parsed835 = parse_835_text(text, payer_config=None)  # type: ignore[arg-type]
+
+    # Sanity: the 835 envelope also has the field (shared model class),
+    # but the parser doesn't populate it — so it must be None.
+    assert parsed835.envelope.transaction_set_control_number is None
+
+    record = BatchRecord835(
+        id="b-txn-cn-835-1",
+        input_filename="minimal_835.txt",
+        parsed_at=datetime.now(timezone.utc),
+        result=parsed835,
+    )
+    cycl_store.add(record)
+
+    with db_mod.SessionLocal()() as s:
+        row = s.get(Batch, "b-txn-cn-835-1")
+        assert row is not None
+        assert row.transaction_set_control_number is None
+        # And the kind/kind round-trip is correct (sanity check that
+        # we did exercise the 835 path, not the 837 path).
+        assert row.kind == "835"

backend/tests/test_db_migrate.py

@@ -126,14 +126,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
     SP28 bumped it to 18 with the claim_acks join table.
     SP32 bumped it to 19 with rendering_provider_npi +
     service_provider_npi on claims and remittances.
+    SP37 bumped it to 20 with batch transaction_set_control_number.
     """
     engine = _fresh_engine(tmp_path)
     db_migrate.run(engine)
     v_after_first = _user_version(engine)
-    assert v_after_first == 19, f"expected head=19, got {v_after_first}"
+    assert v_after_first == 20, f"expected head=20, got {v_after_first}"
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 19, "second run should not bump version"
+    assert _user_version(engine) == 20, "second run should not bump version"
 
 
 def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -159,7 +160,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
     engine = _fresh_engine(tmp_path)
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 19, f"expected head=19, got {_user_version(engine)}"
+    assert _user_version(engine) == 20, f"expected head=20, got {_user_version(engine)}"
 
     # Two claims in one batch with the same patient_control_number
     # must be insertable. If 0015's table recreation re-introduced a

backend/tests/test_migration_0020.py

@@ -0,0 +1,173 @@
+"""Tests for migration 0020_add_batch_txn_set_control_number.sql.
+
+SP37 adds a nullable ``batches.transaction_set_control_number`` column
+populated from the parsed 837's ST02 (transaction set control number)
+on every write. The 999 ack join (Pass 1) needs to resolve by ST02,
+not ISA13, so this column is the join key. This migration is purely
+additive: nullable, no default, backfills from
+``raw_result_json.envelope.transaction_set_control_number`` where the
+source JSON already carries it.
+
+For the backfill-shape tests we point ``db_migrate.MIGRATIONS_DIR``
+at the real migrations directory, apply all migrations once to bring
+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.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+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 _fresh_engine(path: Path) -> sa.Engine:
+    return sa.create_engine(f"sqlite:///{path}", future=True)
+
+
+def _table_info(engine: sa.Engine, table: str) -> list[tuple]:
+    """Return PRAGMA table_info rows for ``table`` as plain tuples."""
+    with engine.connect() as conn:
+        return list(conn.exec_driver_sql(f"PRAGMA table_info({table});").tuples())
+
+
+def _real_migrations_dir() -> Path:
+    return Path(__file__).parent.parent / "src" / "cyclone" / "migrations"
+
+
+@pytest.fixture
+def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    """Yield an engine at v20 against which every real migration has run.
+
+    Points ``db_migrate.MIGRATIONS_DIR`` at the real migrations
+    directory so the test exercises the actual 0020 file, then runs
+    the migration runner on a per-test fresh DB.
+    """
+    monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", _real_migrations_dir())
+    engine = _fresh_engine(tmp_path / "mig0020.db")
+    db_migrate.run(engine)
+
+    # Confirm head is 20 (every migration applied).
+    with engine.connect() as conn:
+        v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
+    assert v == 20, f"expected migration head=20, got {v}"
+
+    yield engine
+    engine.dispose()
+
+
+def test_migration_0020_creates_column(migrated_engine) -> None:
+    """Migration 0020 adds ``transaction_set_control_number`` to ``batches``."""
+    cols = _table_info(migrated_engine, "batches")
+    col_names = {row[1] for row in cols}
+    assert "transaction_set_control_number" in col_names, (
+        "batches.transaction_set_control_number missing — migration 0020 did not run. "
+        f"Existing columns: {sorted(col_names)}"
+    )
+
+
+def test_migration_0020_column_is_nullable(migrated_engine) -> None:
+    """The new column is nullable (no DEFAULT, no NOT NULL) so existing
+    batches that don't yet carry the ST02 stay valid."""
+    cols = {row[1]: row for row in _table_info(migrated_engine, "batches")}
+    row = cols["transaction_set_control_number"]
+    # PRAGMA table_info tuples: (cid, name, type, notnull, dflt_value, pk)
+    assert row[3] == 0, f"notnull flag must be 0 (nullable), got {row[3]}"
+    assert row[4] is None, f"dflt_value must be NULL, got {row[4]!r}"
+    assert "TEXT" in (row[2] or "").upper(), f"expected TEXT column, got {row[2]!r}"
+
+
+def test_migration_0020_backfills_when_key_present(
+    migrated_engine: sa.Engine,
+) -> None:
+    """Replay the migration's backfill UPDATE against a row whose
+    ``raw_result_json`` carries the key — must populate the new column.
+
+    Replay (rather than waiting for ``db_migrate.run()`` to do it) is
+    the only way to test the SQL against a row that didn't exist when
+    init_db() ran; the migration's UPDATE naturally runs only over
+    pre-existing rows.
+    """
+    st02_value = "ST0001"
+    raw_json = {"envelope": {"transaction_set_control_number": st02_value}}
+
+    with migrated_engine.begin() as conn:
+        conn.exec_driver_sql(
+            "INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
+            "VALUES ('B-ST02-1', '837p', 'mig0020-st02.txt', '2026-07-07 00:00:00', ?)",
+            (json.dumps(raw_json),),
+        )
+        conn.exec_driver_sql(BACKFILL_SQL)
+
+    with migrated_engine.connect() as conn:
+        row = conn.exec_driver_sql(
+            "SELECT transaction_set_control_number FROM batches WHERE id='B-ST02-1'"
+        ).first()
+    assert row is not None
+    assert row[0] == st02_value, (
+        f"backfill failed: expected {st02_value!r}, got {row[0]!r}"
+    )
+
+
+def test_migration_0020_backfill_conditional_on_key_present(
+    migrated_engine: sa.Engine,
+) -> None:
+    """Replay the UPDATE against a row whose ``raw_result_json`` does NOT
+    carry the key — must stay NULL. Proves the UPDATE is conditional
+    (the ``json_extract(...) IS NOT NULL`` guard) rather than
+    unconditionally overwriting with NULL."""
+    raw_json = {"envelope": {"control_number": "ISA0001"}}  # no txn-set key
+    with migrated_engine.begin() as conn:
+        conn.exec_driver_sql(
+            "INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
+            "VALUES ('B-NOST-1', '837p', 'mig0020-nost.txt', '2026-07-07 00:00:00', ?)",
+            (json.dumps(raw_json),),
+        )
+        conn.exec_driver_sql(BACKFILL_SQL)
+
+    with migrated_engine.connect() as conn:
+        row = conn.exec_driver_sql(
+            "SELECT transaction_set_control_number FROM batches WHERE id='B-NOST-1'"
+        ).first()
+    assert row is not None
+    assert row[0] is None, (
+        f"backfill must not overwrite when key is absent; got {row[0]!r}"
+    )
+
+
+def test_migration_0020_backfill_handles_null_raw_result_json(
+    migrated_engine: sa.Engine,
+) -> None:
+    """A Batch row with ``raw_result_json IS NULL`` (the prior SP's
+    unparsed state) must not crash the backfill and must stay NULL."""
+    with migrated_engine.begin() as conn:
+        conn.exec_driver_sql(
+            "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)
+
+    with migrated_engine.connect() as conn:
+        row = conn.exec_driver_sql(
+            "SELECT transaction_set_control_number FROM batches WHERE id='B-NULL-1'"
+        ).first()
+    assert row is not None
+    assert row[0] is None

backend/tests/test_submission.py

@@ -0,0 +1,208 @@
+"""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 ``_auto_init_db`` fixture in conftest.py)
+so the join-key wiring is exercised end-to-end.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from cyclone import db as db_mod
+from cyclone.submission.core import submit_file
+from cyclone.submission.result import SubmitOutcome
+
+
+_FIXTURE = Path(__file__).parent / "fixtures" / "submit-batch" / "single-claim.x12"
+
+
+class _FakeSftp:
+    """Implements just enough of SftpClient for submit_file's idempotency check.
+
+    After every successful ``write_file``, the file is added to
+    ``existing_files`` keyed by remote_path with the byte size — so a
+    subsequent ``stat`` returns the right size and the idempotency
+    check in ``submit_file`` short-circuits to ``SKIPPED``. Without
+    this bookkeeping, every re-run would look like the file isn't on
+    the remote and would re-upload.
+    """
+
+    def __init__(self, existing_files: dict[str, int] | None = None):
+        self.existing_files = existing_files or {}
+        self.put_calls: list[tuple[str, bytes]] = []
+
+    def stat(self, path: str):
+        if path not in self.existing_files:
+            raise IOError(f"no such file: {path}")
+        m = MagicMock()
+        m.st_size = self.existing_files[path]
+        return m
+
+    def write_file(self, remote_path: str, content: bytes):
+        self.put_calls.append((remote_path, content))
+        # Mirror the upload so a subsequent stat() sees it.
+        self.existing_files[remote_path] = len(content)
+
+
+def test_submit_file_happy_path():
+    """parse → DB write → SFTP upload, all steps succeed."""
+    fake = _FakeSftp()
+    sftp_block = MagicMock()
+    sftp_block.stub = False
+    sftp_block.paths = {"outbound": "/ToHPE"}
+
+    result = submit_file(
+        _FIXTURE,
+        sftp_block=sftp_block,
+        actor="test",
+        validate=True,
+        sftp_client_factory=lambda _block: fake,
+    )
+
+    assert result.outcome == SubmitOutcome.SUBMITTED
+    assert result.batch_id is not None
+    assert len(fake.put_calls) == 1
+    # DB row landed.
+    with db_mod.SessionLocal()() as s:
+        assert s.get(db_mod.Batch, result.batch_id) is not None
+
+
+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
+
+    r1 = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
+    r2 = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
+
+    # First call SUBMITTED, second call SKIPPED (size match on stat).
+    assert r1.outcome == SubmitOutcome.SUBMITTED
+    assert r2.outcome == SubmitOutcome.SKIPPED
+
+
+def test_submit_file_parse_fail(tmp_path):
+    """Bad 837 → PARSE_FAILED, no DB write, no SFTP call."""
+    bad = tmp_path / "bad.x12"
+    bad.write_bytes(b"not x12")
+    sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
+    fake = _FakeSftp()
+    factory = lambda _b: fake
+
+    result = submit_file(bad, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
+    assert result.outcome == SubmitOutcome.PARSE_FAILED
+    assert len(fake.put_calls) == 0
+
+
+def test_submit_file_db_fail(monkeypatch):
+    """If add raises, no SFTP call is made (DB-first invariant)."""
+    from cyclone.store import store as cycl_store
+    monkeypatch.setattr(cycl_store, "add", MagicMock(side_effect=RuntimeError("db down")))
+
+    sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
+    fake = _FakeSftp()
+    factory = lambda _b: fake
+
+    result = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
+    assert result.outcome == SubmitOutcome.DB_FAILED
+    assert len(fake.put_calls) == 0
+
+
+def test_submit_file_sftp_fail():
+    """SFTP failure after DB write → SFTP_FAILED, DB row still present."""
+
+    class _Boom:
+        def stat(self, _p):
+            raise IOError("nope")
+        def write_file(self, _p, _c):
+            raise RuntimeError("sftp down")
+
+    sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
+    factory = lambda _b: _Boom()
+
+    result = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
+    assert result.outcome == SubmitOutcome.SFTP_FAILED
+    # 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
+
+
+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
+
+
+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

docs/RUNBOOK.md

@@ -138,3 +138,156 @@ started without the secret mount), the scheduler surfaces a
 `RuntimeError` at the next tick that names the env var and the
 missing path — this is intentional, so a silent fall-through doesn't
 mask a real misconfiguration.
+
+## Manual SFTP mode (this box's current posture)
+
+Use this when the host's IP isn't whitelisted with Gainwell's MFT, when
+you don't want a daemon polling every minute, or when you prefer to
+drag-drop files with FileZilla / WinSCP. The seeded `dzinesco`
+clearhouse ships in this mode (`sftp_block.stub: true`); no daemon
+changes required.
+
+**Posture.** `SftpClient` reads/writes to local staging instead of
+`mft.gainwelltechnologies.com`:
+
+- Outbound (`write_file`): `./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/ToHPE/`
+- Inbound (`list_inbound`): `./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/`
+
+(Paths are relative to the backend cwd; create the directories if
+absent.)
+
+### Daily flow (manual mode)
+
+1. **Pull inbound from Gainwell** with your SFTP client:
+   `/CO XIX/PROD/coxix_prod_11525703/FromHPE/`
+2. **Stage locally** in `./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/`
+   (or, equivalently, drop them in `/home/tyler/dev/cyclone/ingest/`
+   and copy in).
+3. **Process** with `pull-inbound` (writes acks/835s to DB, dedupes
+   via `processed_inbound_files`):
+
+   ```bash
+   cd /home/tyler/dev/cyclone/backend
+   mkdir -p "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE"
+   cp /home/tyler/dev/cyclone/ingest/*.x12 \
+      "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/"
+   .venv/bin/python -m cyclone pull-inbound --date 20260701
+   .venv/bin/python -m cyclone pull-inbound --date 20260702
+   .venv/bin/python -m cyclone pull-inbound --date 20260703
+   ```
+
+4. **Submit outbound** with `POST /api/clearhouse/submit` (writes
+   serialized 837P files into the stub staging dir; you drag-drop
+   them to your SFTP client's `/CO XIX/PROD/coxix_prod_11525703/ToHPE/`).
+
+### Backfill a backlog
+
+`ingest/` often holds days of unprocessed acks. The copy + per-date
+`pull-inbound` flow above clears it. Files already in
+`processed_inbound_files` are skipped automatically — to re-process,
+delete the row first (`DELETE FROM processed_inbound_files WHERE name = ...`).
+
+### Submitting claims (canonical — SP37)
+
+For 837P files generated upstream (dzinesco) that you want cyclone to
+**track in the DB** before uploading, use the canonical submit path.
+This is the preferred outbound path going forward — it captures the
+batch's `transaction_set_control_number` (the ST02 control number that
+999 acks reference in AK201) so future 999 ack links resolve instead
+of becoming orphans.
+
+```bash
+cd /home/tyler/dev/cyclone/backend
+
+# 1. Lay out files in batch-*-claims subdirs under your ingest dir:
+#    ingest/batch-2026-07-08-claims/claim-001.x12
+#    ingest/batch-2026-07-08-claims/claim-002.x12
+#    ...
+
+# 2. CLI — walks ingest/, parses, writes to DB, then SFTP-uploads.
+.venv/bin/python -m cyclone submit-batch \
+    --ingest-dir /home/tyler/dev/cyclone/ingest \
+    --actor cli-submit-batch
+
+# Or via HTTP (auth-gated by matrix_gate):
+curl -s -X POST http://localhost:8000/api/submit-batch \
+    -H 'Content-Type: application/json' \
+    -d '{"ingest_dir": "/home/tyler/dev/cyclone/ingest", "actor": "api-submit-batch"}'
+```
+
+Both surfaces share `cyclone.submission.submit_file` for the
+parse → DB-write → SFTP-upload chain (DB-first, upload-second
+invariant). The walker pattern is identical: `batch-*-claims/*.x12`,
+sorted, with `._*` AppleDouble files skipped.
+
+**When to use `submit-batch` vs `resubmit-rejected-claims`:**
+- `submit-batch` — canonical path for fresh 837s from dzinesco (or any
+  source) that should be tracked in the DB before upload. Default choice.
+- `resubmit-rejected-claims` — one-off path for cases where you do NOT
+  want a DB row (e.g., dzinesco-generated fixes not ready for canonical
+  tracking). Legacy, retained for backward compat.
+
+**Status codes / exit codes:**
+- HTTP 200 on completed runs (per-file failures live in the JSON body);
+  401 unauthenticated; 404 no clearhouse; 409 stub mode; 422 validation.
+- CLI exit 0 on completed runs (per-file failures counted, not bumped);
+  2 on config-level failures (no clearhouse / stub mode / missing dir).
+
+### Note on per-file parse CLIs
+
+`parse-837` and `parse-835` exist as CLIs but only emit JSON files to
+`--output-dir`; they do NOT write to the DB. There is no
+`parse-999` / `parse-ta1` / `parse-277ca` CLI in this version of
+cyclone — the canonical 999/TA1/277CA/835 ingestion path is
+`pull-inbound` → `Scheduler.process_inbound_files`. For inspection of
+a single file without DB writes, use the Python API:
+
+```python
+from cyclone.parsers.parse_999 import parse as parse_999
+result = parse_999(open("path.999.x12").read())
+```
+
+### Switching from manual → real (and back)
+
+When this host's IP is whitelisted with Gainwell's MFT admin, the
+SP25 procedure above flips the block. The CLI one-liner:
+
+```bash
+export CYCLONE_SFTP_PASSWORD="$GAINWELL_SFTP_PASS"
+.venv/bin/python -c "
+from cyclone import db, store
+from cyclone.providers import Clearhouse, SftpBlock
+db.init_db()
+ch = store.store.get_clearhouse()
+sb = ch.sftp_block.model_dump()
+sb['stub'] = False
+sb['auth'] = {'password_keychain_account': 'sftp.gainwell.password'}
+new = Clearhouse(
+    id=1, name=ch.name, tpid=ch.tpid,
+    submitter_id_qual=ch.submitter_id_qual,
+    submitter_name=ch.submitter_name,
+    submitter_contact_name=ch.submitter_contact_name,
+    submitter_contact_email=ch.submitter_contact_email,
+    filename_block=ch.filename_block,
+    sftp_block=SftpBlock.model_validate(sb, strict=True),
+    updated_at=ch.updated_at,
+)
+print('updated:', store.store.update_clearhouse(new).sftp_block.stub)
+"
+```
+
+Hot-reload via `PATCH /api/clearhouse` (preferred over direct DB
+write — it also calls `scheduler.reconfigure_scheduler` so the running
+daemon picks up the new block without a restart):
+
+```bash
+curl -s -b cookies.txt http://127.0.0.1:8000/api/clearhouse > /tmp/ch.json
+jq '.sftp_block.stub = false
+    | .sftp_block.auth = {"password_keychain_account": "sftp.gainwell.password"}' \
+   /tmp/ch.json > /tmp/ch-patched.json
+curl -X PATCH http://127.0.0.1:8000/api/clearhouse \
+     -H 'Content-Type: application/json' -b cookies.txt \
+     --data @/tmp/ch-patched.json
+```
+
+To revert, set `stub: true` and `auth: {"method": "keychain", "secret_ref": "sftp.gainwell.password"}`.

docs/superpowers/specs/2026-07-07-cyclone-submit-batch-canonical-flow-design.md

@@ -0,0 +1,217 @@
+# Sub-project 37 — Canonical `submit-batch` flow: Design Spec
+
+**Date:** 2026-07-07
+**Status:** Draft, awaiting user sign-off
+**Branch:** `sp37-submit-batch-canonical-flow` (off main, post-SP36 merge)
+**Aesthetic direction:** No UI changes. Pure backend structural + data-model addition — the public HTTP surface gains one new endpoint; existing CLIs and endpoints are preserved unchanged.
+
+---
+
+## 1. Scope
+
+Today, cyclone's `claims` table is essentially empty in production (`clm-1` is the only row as of 2026-07-07). The reason is structural: dzinesco generates the 837P files (in `ingest/corrected/batch-*-claims/*.x12`), cyclone's `resubmit-rejected-claims` CLI only validates and re-uploads them to Gainwell's MFT — it never writes the claims to the DB. When 999 acknowledgments come back referencing those batches, every AK2 set-response becomes an orphan because there is no `Batch` row to join against.
+
+The 999→claim join (the D10 two-pass join in `claim_acks.lookup_claims_for_ack_set_response`) has two paths:
+
+- **Pass 1 (primary):** `Batch.envelope.control_number == 999's set_control_number`. This currently fails because the 999's `set_control_number` (AK201) echoes the source 837's **ST02** (transaction set control number), not the **ISA13** (interchange control number). The parsed `Envelope` model stores ISA13 in `control_number`, so the keys never collide even when the DB row exists.
+- **Pass 2 (fallback):** `Claim.patient_control_number == 999's set_control_number`. This also fails because the `_claim_837_row` ORM builder sets `patient_control_number = claim.claim_id` (CLM01), and CLM01 ≠ ST02 in production batches.
+
+The fix is twofold: (a) make cyclone write a `Batch` + `Claim` rows at submission time so there is something to join against, and (b) make the join key actually match — by storing the ST02 explicitly on `Batch` and using it in Pass 1.
+
+**In scope:**
+
+- A new top-level CLI `cyclone submit-batch` that does parse → DB write → SFTP upload per file, in that order.
+- A new HTTP endpoint `POST /api/submit-batch` with the same semantics.
+- A shared helper module `cyclone.submission.submit_file` so the CLI and HTTP path share one source of truth.
+- A new nullable column `Batch.transaction_set_control_number`, populated from the parsed 837's ST02 on every `add_record` write.
+- An Alembic migration `0013_add_batch_txn_set_control_number.py` (additive, nullable, backfills from `raw_result_json.envelope.transaction_set_control_number` where present).
+- An update to `claim_acks.batch_envelope_index` so Pass 1 also matches on `transaction_set_control_number` (Pass 2 unchanged).
+- Per-claim `claim_submitted` activity events and per-file `clearhouse.submitted` audit events (same shape `resubmit-rejected-claims` already emits).
+
+**Out of scope (explicit):**
+
+- Deprecating `parse-837` CLI or `/api/parse-837` endpoint. They remain for one-off testing and dry-run validation.
+- Deprecating `resubmit-rejected-claims`. It remains the "validate-then-upload without DB write" path for the dzinesco-generated files that aren't ready for canonical tracking.
+- UI changes. The Claims page already renders the fields a new tracked claim needs; the existing `claim_written` live-tail event fires automatically from `add_record`.
+- Bulk backfill of the existing `ingest/corrected/batch-*-claims/*.x12` backlog. A follow-up SP will walk that directory once `submit-batch` is proven.
+- Any change to the `claims` row schema beyond the new `Batch` column.
+- SFTP auth work. This box stays in manual mode (post-2026-07-07 incident); `submit-batch` works in stub mode for local testing and against a whitelisted IP for production.
+
+## 2. Decisions (locked during brainstorming)
+
+These were each a multiple-choice question with operator sign-off on 2026-07-07.
+
+1. **Where does the DB write happen?** New canonical `submit` flow. `submit-batch` parses → writes to DB → uploads, replacing neither `parse-837` nor `resubmit-rejected-claims` but superseding them as the preferred path.
+2. **DB vs upload ordering?** DB-first, upload-second. If DB write fails, the file never goes on the wire. If DB write succeeds but upload fails, re-running is safe (idempotent DB write, idempotent SFTP put).
+3. **Join key for 999 acks?** Add `Batch.transaction_set_control_number` column. Update Pass 1 to match on it. Don't repurpose `envelope.control_number` (preserves backward compat with existing joins) and don't force CLM01 == ST02 (Gainwell's existing 999s use different values).
+4. **Deprecation posture?** Add `submit-batch`; preserve `parse-837` and `resubmit-rejected-claims`. Document `submit-batch` as preferred in CLAUDE.md + RUNBOOK. Future SP can deprecate if usage shifts.
+
+## 3. Architecture
+
+The new flow is a single, small public surface — one CLI command, one HTTP endpoint, one shared helper. The helper owns the parse → write → upload sequence and is the only place that knows about the order.
+
+### Module layout
+
+A new package `cyclone.submission` (under `backend/src/cyclone/submission/`) holds the shared helper:
+
+- `__init__.py` — re-exports `submit_file`, `SubmitResult`, `SubmitOutcome`
+- `core.py` — `submit_file(path: Path, *, sftp_block, actor: str, event_bus=None, validate: bool = True) -> SubmitResult`. Owns its own DB session for the parse-write step, returns a result dataclass the caller inspects.
+- `cli.py` — Click handler for `cyclone submit-batch`. Walks `--ingest-dir`, calls `submit_file` per file, aggregates counts, exits 0/1/2 per cyclone-cli convention.
+- `api.py` — FastAPI handler for `POST /api/submit-batch`. Mounted in `api_routers/submission.py`. Calls `submit_file` per file. Auth-gated by the existing `matrix_gate`.
+
+### `submit_file` algorithm
+
+The walker globs `<ingest_dir>/batch-*-claims/*.x12` (mirrors `resubmit-rejected-claims` exactly — same pattern, same `_`-prefix skip for AppleDouble residue). For each matched file:
+
+1. Read bytes; if `validate=True`, run `parse_837_text(bytes, payer_cfg)`. On parse failure, record `{path, "parse_failed", exc}` and return without writing or uploading.
+2. Run a payer-id check (must be `CO_TXIX`); on mismatch, record and return.
+3. Call `store.add_record(record, event_bus=...)`. On DB failure, record and return (no SFTP call).
+4. Build the remote path `{outbound_root}/{filename}`. Open an `SftpClient` (the seeded singleton). On SFTP connect failure, record and return — the DB row is already written; the re-run will skip it via idempotency.
+5. `sftp_client.stat(remote_path)` — if `st_size == local_size`, mark as `skipped` (already uploaded; do not re-emit audit). Otherwise `sftp_client.write_file(remote_path, bytes)`.
+6. On successful upload, emit one `clearhouse.submitted` audit event with `source_file`, `batch_id`, `actor`.
+
+### Idempotency
+
+Two layers, both per-file:
+
+- **DB layer:** `store.add_record` already does a `s.get(Claim, claim.claim_id)` check before insert. A re-parse of the same file logs "claim already exists; skipping" and continues. Re-running `submit-batch` on an already-tracked file is a no-op on the DB.
+- **SFTP layer:** the `sftp.stat().st_size == local_size` check before `write_file` short-circuits already-uploaded files.
+
+### Failure modes summary
+
+- Parse fails → no DB write, no SFTP put, file counted as `failed`.
+- Payer-id mismatch → same as parse fail.
+- DB write fails → no SFTP put, file counted as `failed`. The DB session rolls back; no orphan rows.
+- SFTP connect fails → DB row already committed; file counted as `failed`. Re-run is safe.
+- SFTP stat succeeds + size match → counted as `skipped`; no audit event.
+- SFTP write fails → file counted as `failed`. DB row already committed. Re-run is safe.
+- Crash between DB write and SFTP put → DB row exists, file never landed. Re-run will see the size mismatch (remote doesn't have it) and upload.
+
+## 4. Data flow & error handling
+
+### `POST /api/submit-batch`
+
+Request body (JSON):
+- `ingest_dir: string` — root directory holding `batch-*-claims/*.x12` subfolders (mirrors the CLI flag).
+- `validate: bool = true` — pass-through to `submit_file`.
+- `actor: string = "api-submit-batch"` — audit actor tag.
+- `limit: int | null = null` — stop after N files (smoke test).
+
+Response body (200):
+- `submitted: int` — files uploaded this run.
+- `skipped: int` — files already on remote (size match).
+- `failed: int` — files that hit parse / payer / DB / SFTP errors.
+- `results: list<{file: string, outcome: "submitted"|"skipped"|"parse_failed"|"payer_mismatch"|"db_failed"|"sftp_failed", batch_id: string|null, error: string|null}>` — per-file detail.
+
+Status codes:
+- `200` — run completed (may have failures; details in the response body).
+- `401` — no auth cookie (matrix_gate).
+- `422` — body validation failure.
+- `500` — unexpected exception (caught by the existing `@app.exception_handler(Exception)`).
+
+### CLI exit codes (per `cyclone-cli` convention)
+
+- `0` — run completed, even with some per-file failures (the per-file detail is in stdout).
+- `1` — unexpected exception (uncaught error in the walker or the helper itself).
+- `2` — config-level failure (no clearhouse seeded, SFTP block in stub mode for non-stubbed invocation, ingest-dir doesn't exist).
+
+### Audit + live-tail events
+
+- `claim_submitted` (one per Claim row) — emitted by `add_record` already; no new code path.
+- `clearhouse.submitted` (one per successful upload) — emitted by `submit_file` after the SFTP put succeeds. Same actor tag and payload shape as the current `resubmit-rejected-claims` path uses.
+
+### Migration `0013_add_batch_txn_set_control_number.py`
+
+- `op.add_column('batches', sa.Column('transaction_set_control_number', sa.String(32), nullable=True))`.
+- Backfill: `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`.
+- Reversible: drop column.
+- The down migration is `op.drop_column('batches', 'transaction_set_control_number')`.
+
+### Join-key update in `claim_acks.batch_envelope_index`
+
+The existing function reads every `Batch` row whose `raw_result_json` has an envelope and returns `dict[str, str]` keyed by `envelope.control_number`. The updated version returns the **same shape** — a single `dict[str, str]` — but **populated from two columns**:
+
+- Each row contributes one entry keyed by `envelope.control_number` (preserved).
+- Each row also contributes one entry keyed by `transaction_set_control_number` (new), when that column is non-NULL.
+
+If both keys resolve to the same value (Gainwell batches where the ST02 happens to equal the ISA13 — unusual but legal), the second insert is a no-op. If they differ, both keys point to the same `batch.id` — a single `idx.get(set_control_number)` lookup will match either one. The dict may grow by up to `len(batches)` new entries; callers that previously assumed `len(index) == len(batches)` are unaffected because that invariant was already loose (multiple ST02s can share one ISA13).
+
+`lookup_claims_for_ack_set_response` Pass 1 is unchanged at the call site — one `idx.get(set_control_number)` lookup that succeeds whether the source 837's ST02 matches a `transaction_set_control_number` row or an `envelope.control_number` row. Pass 2 (PCN) fires as today if both miss.
+
+## 5. Testing approach
+
+Per `cyclone-tests` skill: pytest under `backend/tests/`, two flavors — `test_api_*` (FastAPI integration) and `test_*` (pure unit). Frontend tests out of scope (no UI changes).
+
+### Unit tests (`backend/tests/test_submission.py`)
+
+- `submit_file happy path` — mock SFTP, real DB; assert one Batch + N Claim rows + one audit event.
+- `submit_file idempotency on DB` — call `submit_file` twice with the same file; assert second call returns `skipped` (DB no-op).
+- `submit_file idempotency on SFTP` — mock SFTP `stat` to return matching size; assert `skipped`, no second `write_file` call, no second audit event.
+- `submit_file parse fail` — pass a bad 837; assert no DB write, no SFTP call, `parse_failed` outcome.
+- `submit_file payer mismatch` — pass a valid 837 with wrong payer_id; assert no DB write, no SFTP call, `payer_mismatch` outcome.
+- `submit_file DB fail` — monkey-patch `add_record` to raise; assert no SFTP call, `db_failed` outcome.
+- `submit_file SFTP fail` — mock SFTP `write_file` to raise; assert `sftp_failed` outcome, DB row still present.
+- `submit_file re-run after partial failure` — submit a file that fails SFTP, then submit again with SFTP working; assert second run uploads and emits audit.
+
+### Integration tests (`backend/tests/test_api_submit_batch.py`)
+
+- `POST /api/submit-batch` happy path — 3 valid 837 fixtures in `ingest_dir`; assert 200 + body shape.
+- `POST /api/submit-batch` auth gate — no cookie → 401 (matrix_gate).
+- `POST /api/submit-batch` body validation — missing `ingest_dir` → 422.
+
+### Migration tests (`backend/tests/test_migration_0013.py`)
+
+- Up + down on a fresh DB; column exists + is nullable; backfill populates the column from raw JSON.
+
+### Join-key update tests (`backend/tests/test_claim_acks_index.py`)
+
+- `batch_envelope_index` returns both maps; Pass 1 finds a claim by ST02 even when ISA13 differs; Pass 1 finds by ISA13 (backward compat); Pass 2 fallback still works.
+
+### Live test (after merge)
+
+Walk an actual `ingest/batch-*-claims/` directory via the CLI; verify:
+- `claims` table has new rows with the expected `patient_control_number` (CLM01).
+- `batches` table has new rows with `transaction_set_control_number` matching the parsed ST02.
+- Live-tail `claim_written` events fire on the API.
+
+Then drop a synthetic 999 (with the matching AK2 set_control_number) into `ingest/`, run `pull-inbound`, verify the `ClaimAck` link row is created (not an orphan).
+
+## 6. Threat model (post-SP24 alignment)
+
+No new attack surface. The new endpoint and CLI:
+
+- Are gated by the existing `matrix_gate` (HTTP) and `CYCLONE_AUTH_DISABLED` env var.
+- Use the same SFTP client and credential paths as `resubmit-rejected-claims` today.
+- Write to the same store facade; no new SQL surface.
+- Emit the same audit events with the same payload shape.
+
+The `transaction_set_control_number` column is operator-visible (it's in the raw `batches` row), but no auth boundary depends on it being absent. PII exposure is unchanged (the column carries the same value the 999 ack would have echoed back — already public-to-Gainwell).
+
+## 7. Risks & mitigations
+
+- **Risk:** The new `submission` package is the third "command runner" in the codebase (alongside `cli.py`'s standalone commands and `scheduler.py`). **Mitigation:** Keep `submit_file` as a pure helper with no FastAPI/Click dependencies; the API and CLI handlers are thin wrappers. Document the layering in `submission/__init__.py`'s module docstring.
+- **Risk:** `resubmit-rejected-claims` and `submit-batch` will diverge over time (one writes to DB, one doesn't). **Mitigation:** Both call the same `submit_file` helper if `submit-batch`'s `--write-to-db` flag is set to true (default). Document the divergence point in RUNBOOK.
+- **Risk:** Migration backfill could be slow on a large DB. **Mitigation:** `UPDATE ... WHERE json_extract(...) IS NOT NULL` is one pass; on a million-row DB this is sub-second. Index on `transaction_set_control_number` not added (low cardinality per batch; existing `ix_claims_patient_control_number` covers the hot path).
+- **Risk:** Live-tail event volume doubles when `submit-batch` runs (one `claim_submitted` per claim in addition to the historical `claim_written`). **Mitigation:** This is desired behavior — the operator wants the activity feed to show submissions. Document in the RUNBOOK that submitting a 1000-claim batch will publish ~1000 events.
+
+## 8. Rollout
+
+- **Branch:** `sp37-submit-batch-canonical-flow`, off main as of `710104f`.
+- **Implementation order:**
+  1. Migration `0013` + backfill (additive, no behavior change yet).
+  2. Update `claim_acks.batch_envelope_index` + Pass 1 lookup (no behavior change yet — the new map is built but unused until `submit-batch` writes rows).
+  3. `cyclone/submission/core.py` helper + unit tests.
+  4. CLI handler `cyclone submit-batch` + smoke test.
+  5. HTTP handler `POST /api/submit-batch` + integration tests.
+  6. Live test with one of the existing `ingest/batch-*-claims/*.x12` directories (will be in stub mode on this host).
+- **Live test cadence:** per the user's standing directive — live-test after each significant step, run autoreview, commit.
+- **Merge shape:** single atomic merge commit per SP-N convention. No squash. No rebase.
+- **Post-merge:** RUNBOOK update for the canonical flow; CLAUDE.md update to mark `submit-batch` as preferred.
+
+## 9. Open questions
+
+None blocking the spec. Tracked for follow-up SPs:
+
+- **Backfill command** — once `submit-batch` is proven, a `cyclone backfill-claims` CLI that walks the existing `ingest/corrected/batch-*-claims/*.x12` backlog (without uploading — they're already on Gainwell's side) and writes the DB rows. This is what would un-orphan the 1491+ orphan 999 acks currently in the DB.
+- **Deprecation timeline for `resubmit-rejected-claims`** — once `submit-batch` is the default, consider folding `resubmit-rejected-claims` into a `--no-write-to-db` flag on `submit-batch` to consolidate the surface.
+- **Cross-link to 277CA acks** — same `transaction_set_control_number` join key will resolve 277CA orphans if/when those arrive. The migration and index update cover both 999 and 277CA since both use Pass 1.