merge: fix 999 IK5 and TA1 UI into Version-1.0.0

README.md

@@ -690,7 +690,7 @@ backup create` manually retains full control.
 The `clearhouse.submit` endpoint uses `paramiko` to push a batch of
 generated 837 files to the dzinesco SFTP server
 (`mft.gainwelltechnologies.com:22`, path
-`/CO XIX/PROD/coxix_prod_11525703/FromHPE`). The SFTP credential is
+`/CO XIX/PROD/coxix_prod_11525703/ToHPE`). The SFTP credential is
 fetched from the macOS Keychain at call time — never read from YAML,
 never logged, never written to disk. The wire-up honors the file-naming
 template stored in the `clearhouse` config:
@@ -898,7 +898,7 @@ Shipped sub-projects (most recent first):
 - **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed
   `SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint
   actually pushes to
-  `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
+  `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/ToHPE`.
   SFTP credentials are read from the macOS Keychain at call time.
 - **Sub-project 12 (shipped) — Encryption at rest.** Optional
   SQLCipher AES-256 encryption of the SQLite file, with the key
@@ -1160,7 +1160,7 @@ the one-time setup recipe.
 
 - `POST /api/clearhouse/submit` — same endpoint as SP9; the
   implementation is now a real `paramiko` `SftpClient.write` to
-  `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
+  `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/ToHPE`.
   SFTP credentials are fetched from the macOS Keychain at call time.
 
 ## License

backend/src/cyclone/api.py

@@ -18,6 +18,7 @@ Additional origins (LAN IPs, staging hosts) can be appended via the
 
 from __future__ import annotations
 
+import asyncio
 import csv
 import io
 import json
@@ -36,6 +37,7 @@ from pydantic import ValidationError
 
 from cyclone import __version__, db
 from cyclone.auth.deps import matrix_gate
+from cyclone.clearhouse import InboundFile
 from cyclone.db import Batch, Claim, ClaimState, Remittance
 from sqlalchemy import desc, or_
 from sqlalchemy.exc import IntegrityError
@@ -667,12 +669,13 @@ async def parse_835_endpoint(
 def _ack_count_summary(result) -> tuple[int, int, int, str]:
     """Aggregate (received, accepted, rejected, ack_code) from a ParseResult999.
 
-    The first functional group carries the canonical counts; falls back
+    Counts are derived from the set-level ``IK5`` responses (one per
-    to summing per-set codes if no AK9 was found.
+    AK2 in the 999), not the functional-group ``AK9``. Gainwell's
+    MFT ships AK9 segments that contradict the per-set IK5
+    (e.g. ``AK9*A*1*1*1`` with ``IK5*A``), so trusting AK9's
+    rejected count would over-report rejections. The set-level
+    IK5 is the authoritative per-claim accept/reject signal.
     """
-    if result.functional_group_acks:
-        fg = result.functional_group_acks[0]
-        return (fg.received_count, fg.accepted_count, fg.rejected_count, fg.ack_code)
     sets = result.set_responses
     received = len(sets)
     accepted = sum(1 for s in sets if s.set_accept_reject.code == "A")
@@ -3419,6 +3422,128 @@ async def scheduler_tick() -> Any:
     return {"ok": True, "tick": result.as_dict()}
 
 
+@app.post("/api/admin/scheduler/pull-inbound", dependencies=[Depends(matrix_gate)])
+async def scheduler_pull_inbound(
+    date: str = Query(
+        ..., pattern=r"^\d{8}$",
+        description="Date filter as YYYYMMDD; only filenames whose 8-digit "
+                    "timestamp (the 9th positional group in the inbound "
+                    "filename) matches are downloaded and processed.",
+    ),
+    file_types: str | None = Query(
+        default=None,
+        description="Optional comma-separated whitelist of file_types "
+                    "(999, TA1, 277, 277CA, 835). Defaults to 999+TA1.",
+    ),
+    limit: int = Query(default=2000, ge=1, le=10000),
+) -> Any:
+    """Targeted pull: list, filter to a date, download, and process.
+
+    Bypasses the alphabetical full-listing pass. Workflow:
+      1. ``SftpClient.list_inbound_names()`` — sub-second metadata-only
+         listing of the inbound MFT dir (skips ``*_warn.txt``).
+      2. Client-side filter: keep files whose 8-digit timestamp
+         substring equals ``date`` and whose ``file_type`` is in the
+         allowlist.
+      3. ``SftpClient.download_inbound(f)`` for each — fetches bytes
+         into the local cache.
+      4. ``Scheduler.process_inbound_files(files)`` — runs the same
+         per-file pipeline as a regular tick (already-processed files
+         are deduped via ``processed_inbound_files``).
+
+    Use this for the daily "process today's 999s" workflow without
+    paying the cost of downloading the full inbound set.
+
+    Returns ``{"ok": True, "summary": {...}}`` with
+    ``listed / matched / downloaded / processed / skipped / errored``
+    counters and the date / file_type filters applied.
+    """
+    import time
+
+    from cyclone.clearhouse import SftpClient
+    from cyclone.edi.filenames import (
+        ALLOWED_FILE_TYPES,
+        parse_inbound_filename,
+    )
+    from cyclone.providers import SftpBlock
+
+    sched = _scheduler_or_503()
+    block: SftpBlock = sched._sftp_block  # noqa: SLF001 — internal but stable
+    client = SftpClient(block)
+
+    if file_types:
+        wanted = {t.strip().upper() for t in file_types.split(",") if t.strip()}
+        unknown = wanted - ALLOWED_FILE_TYPES
+        if unknown:
+            raise HTTPException(
+                status_code=400,
+                detail=f"file_types {sorted(unknown)!r} not in "
+                       f"{sorted(ALLOWED_FILE_TYPES)}",
+            )
+    else:
+        wanted = {"999", "TA1"}  # daily default — what the operator needs
+
+    started = time.monotonic()
+    try:
+        # Single SFTP listdir — fast, no download.
+        all_files = await asyncio.to_thread(client.list_inbound_names)
+    except Exception as exc:
+        log.exception("SFTP list_inbound_names failed")
+        raise HTTPException(
+            status_code=502,
+            detail=f"SFTP list failed: {type(exc).__name__}: {exc}",
+        ) from exc
+
+    listed = len(all_files)
+    matched: list[InboundFile] = []
+    for f in all_files:
+        if f.name.find(date) == -1:
+            continue
+        try:
+            parsed = parse_inbound_filename(f.name)
+        except ValueError:
+            continue
+        if parsed.file_type not in wanted:
+            continue
+        matched.append(f)
+        if len(matched) >= limit:
+            break
+
+    # Download in parallel-ish via to_thread (SftpClient serializes per
+    # connection; the overhead is dominated by the SFTP round trip).
+    downloaded = 0
+    download_errors: list[str] = []
+    for f in matched:
+        try:
+            await asyncio.to_thread(client.download_inbound, f)
+            downloaded += 1
+        except Exception as exc:  # noqa: BLE001
+            log.warning("Failed to download %s: %s", f.name, exc)
+            download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}")
+
+    # Hand off to the scheduler pipeline (idempotent; dedupes via
+    # processed_inbound_files).
+    tick = await sched.process_inbound_files(matched)
+    duration = round(time.monotonic() - started, 3)
+    return {
+        "ok": True,
+        "summary": {
+            "date": date,
+            "file_types": sorted(wanted),
+            "limit": limit,
+            "listed": listed,
+            "matched": len(matched),
+            "downloaded": downloaded,
+            "download_errors": download_errors,
+            "processed": tick.files_processed,
+            "skipped": tick.files_skipped,
+            "errored": tick.files_errored,
+            "duration_s": duration,
+        },
+        "tick": tick.as_dict(),
+    }
+
+
 @app.get("/api/admin/scheduler/status", dependencies=[Depends(matrix_gate)])
 def scheduler_status() -> Any:
     """Return the scheduler's runtime snapshot (running, counters, last tick)."""

backend/src/cyclone/api_routers/acks.py

@@ -32,8 +32,14 @@ def _ack_to_ui(row) -> dict:
     Field names match the rest of the Cyclone API (snake_case). The
     frontend ``useAcks`` hook re-shapes this to the camelCase ``Ack``
     interface in ``src/types/index.ts``.
+
+    Adds ``patient_control_number`` pulled from ``raw_json`` so the
+    operator can correlate each 999 back to the original claim
+    batch. The full inbound filename is reachable via
+    ``GET /api/acks/{ack_id}`` (raw_json carries the full parse
+    tree) and ``GET /api/admin/scheduler/processed-files``.
     """
-    return {
+    body = {
         "id": row.id,
         "source_batch_id": row.source_batch_id,
         "accepted_count": row.accepted_count,
@@ -45,7 +51,16 @@ def _ack_to_ui(row) -> dict:
             if row.parsed_at is not None
             else ""
         ),
+        "patient_control_number": None,
     }
+    raw = row.raw_json or {}
+    try:
+        set_responses = raw.get("set_responses") or []
+        if set_responses:
+            body["patient_control_number"] = set_responses[0].get("set_control_number")
+    except (AttributeError, TypeError):
+        pass
+    return body
 
 
 @router.get("/api/acks")

backend/src/cyclone/clearhouse/__init__.py

@@ -87,11 +87,70 @@ class SftpClient:
         dir and returns :class:`InboundFile` records pointing at the
         cache copy. The remote file is *not* deleted — the operator
         archives inbound files in the MFT UI.
+
+        Gainwell's MFT puts advisory ``*_warn.txt`` files in the same
+        inbound path. They're text-format side-channel notes (not X12
+        envelopes) and are skipped at list time. Use
+        :meth:`list_inbound_names` if you need the raw list without
+        the download.
         """
         if self._stub:
             return self._list_inbound_stub()
         return self._list_inbound_paramiko()
 
+    def list_inbound_names(self) -> list[InboundFile]:
+        """Lightweight listing: returns metadata only, no file download.
+
+        Use this when you want to filter the inbound set (e.g. by date)
+        before paying the download cost. Pair with
+        :meth:`download_inbound` to fetch a filtered subset on demand.
+
+        Real mode is implemented as a single SFTP ``listdir_attr`` call
+        — sub-second on Gainwell's MFT — versus the full
+        :meth:`list_inbound` which downloads every file. The
+        ``*_warn.txt`` advisory files are filtered out the same way.
+
+        Stub mode returns the same as :meth:`list_inbound` (the stub
+        only knows about local files; no download cost).
+        """
+        if self._stub:
+            return self._list_inbound_stub()
+        return self._list_inbound_names_paramiko()
+
+    def download_inbound(self, f: InboundFile) -> Path:
+        """Download a single inbound file to its ``local_path``.
+
+        Idempotent: if ``f.local_path`` already exists and is
+        non-empty, the download is skipped. Callers should use the
+        ``InboundFile`` returned by :meth:`list_inbound_names` and pass
+        it back here — ``local_path`` is the planned cache location
+        and matches the path the scheduler will read from.
+
+        Returns:
+            The on-disk path (same as ``f.local_path``).
+
+        Raises:
+            FileNotFoundError: if the file is missing locally in stub
+                mode, or if the remote file disappears between list
+                and download in real mode.
+        """
+        if f.local_path.exists() and f.local_path.stat().st_size > 0:
+            log.debug(
+                "SFTP: %s already cached at %s, skipping download",
+                f.name, f.local_path,
+            )
+            return f.local_path
+        if self._stub:
+            # Stub mode: no remote — the file is supposed to already be
+            # at f.local_path (operator-dropped). If it isn't there, the
+            # operator hasn't seeded the stub; raise loudly.
+            if not f.local_path.is_file():
+                raise FileNotFoundError(
+                    f"inbound stub file not found: {f.local_path}"
+                )
+            return f.local_path
+        return self._download_inbound_paramiko(f)
+
     def read_file(self, remote_path: str) -> bytes:
         """Read bytes from a remote path.
 
@@ -284,6 +343,12 @@ class SftpClient:
                 if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000:
                     # Directory entry — skip.
                     continue
+                if attr.filename.endswith("_warn.txt"):
+                    # Gainwell's MFT drops text-format advisory notes in
+                    # the same inbound path. They're side-channel noise,
+                    # not X12 envelopes — skip at list time so we don't
+                    # download ~600 advisory files per poll.
+                    continue
                 remote = f"{inbound_dir.rstrip('/')}/{attr.filename}"
                 cache_path = cache_dir / attr.filename
                 # Download into cache. We use ``prefetch`` to keep memory
@@ -299,6 +364,56 @@ class SftpClient:
                 ))
             return files
 
+    def _list_inbound_names_paramiko(self) -> list[InboundFile]:
+        """List inbound names via paramiko; do NOT download (lightweight).
+
+        Same ``listdir_attr`` iteration as
+        :meth:`_list_inbound_paramiko`, but the returned
+        :class:`InboundFile` records have ``local_path`` set to the
+        planned cache location without actually fetching the file.
+        Pair with :meth:`_download_inbound_paramiko` to fetch on
+        demand. Skips ``*_warn.txt`` advisory files the same way.
+        """
+        with self._connect() as (ssh, sftp):
+            inbound_dir = self._block.paths.get("inbound", "/")
+            staging = Path(self._block.staging_dir).resolve()
+            inbound_rel = inbound_dir.lstrip("/")
+            cache_dir = staging / inbound_rel
+            cache_dir.mkdir(parents=True, exist_ok=True)
+
+            files: list[InboundFile] = []
+            try:
+                attrs = sftp.listdir_attr(inbound_dir)
+            except IOError as exc:
+                log.warning("SFTP: cannot list %s: %s", inbound_dir, exc)
+                return []
+
+            for attr in sorted(attrs, key=lambda a: a.filename):
+                if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000:
+                    # Directory entry — skip.
+                    continue
+                if attr.filename.endswith("_warn.txt"):
+                    continue
+                cache_path = cache_dir / attr.filename
+                files.append(InboundFile(
+                    name=attr.filename,
+                    size=attr.st_size or 0,
+                    modified_at=datetime.fromtimestamp(attr.st_mtime or 0),
+                    local_path=cache_path,
+                ))
+            return files
+
+    def _download_inbound_paramiko(self, f: InboundFile) -> Path:
+        """Download a single ``f`` to ``f.local_path`` (idempotent on size>0)."""
+        with self._connect() as (ssh, sftp):
+            inbound_dir = self._block.paths.get("inbound", "/")
+            remote = f"{inbound_dir.rstrip('/')}/{f.name}"
+            f.local_path.parent.mkdir(parents=True, exist_ok=True)
+            with sftp.open(remote, "rb") as src, open(f.local_path, "wb") as dst:
+                shutil.copyfileobj(src, dst, length=64 * 1024)
+            log.info("SFTP: downloaded %d bytes for %s", f.local_path.stat().st_size, f.name)
+            return f.local_path
+
     def _read_file_paramiko(self, remote_path: str) -> bytes:
         with self._connect() as (ssh, sftp):
             buf = io.BytesIO()

backend/src/cyclone/cli.py

@@ -569,3 +569,167 @@ def backup_status() -> None:
     snap = svc.status()
     import json
     click.echo(json.dumps(snap, indent=2, default=str))
+
+
+# ---------------------------------------------------------------------------
+# SP-N fix: `cyclone pull-inbound` — targeted inbound pull + process
+#
+# Mirrors POST /api/admin/scheduler/pull-inbound. For operators who
+# want to drive the daily "process today's 999s" workflow from a cron
+# job or shell script without going through the HTTP API.
+#
+# Exit codes:
+#   0  — processed ≥1 file (or processed 0 with no matches: not an
+#        error, just nothing to do)
+#   1  — unexpected exception
+#   2  — SFTP / config error
+# ---------------------------------------------------------------------------
+
+
+@main.command("pull-inbound")
+@click.option(
+    "--date", "date_str",
+    required=True,
+    help="Date filter YYYYMMDD — only files whose 8-digit timestamp "
+         "substring matches are downloaded and processed.",
+)
+@click.option(
+    "--block", "sftp_block_name",
+    default="dzinesco",
+    show_default=True,
+    help="Name of the SftpBlock in config/payers.yaml to use.",
+)
+@click.option(
+    "--file-types", "file_types_csv",
+    default=None,
+    help="Comma-separated whitelist (default: 999,TA1).",
+)
+@click.option(
+    "--limit", default=2000, show_default=True, type=int,
+)
+def pull_inbound(
+    date_str: str,
+    sftp_block_name: str,
+    file_types_csv: str | None,
+    limit: int,
+) -> None:
+    """List, date-filter, download, and process inbound MFT files.
+
+    Bypasses the alphabetical full-listing pass so a daily pull of
+    ~400 files takes seconds, not hours. Files already in the local
+    cache are skipped (idempotent).
+    """
+    import asyncio as _asyncio
+    from cyclone import db as db_mod
+    from cyclone import scheduler as scheduler_mod
+    from cyclone.edi.filenames import ALLOWED_FILE_TYPES
+    from cyclone.clearhouse import SftpClient
+    from cyclone.providers import SftpBlock
+
+    # Validate the date filter.
+    if not (len(date_str) == 8 and date_str.isdigit()):
+        click.echo(f"--date must be YYYYMMDD, got {date_str!r}", err=True)
+        sys.exit(2)
+
+    db_mod.init_db()
+
+    # Find the SftpBlock. The dzinesco clearhouse singleton is seeded
+    # by store.ensure_clearhouse_seeded() (SP9) and carries the
+    # production SFTP block; that's the only one we have at the
+    # moment. For multi-provider SFTP, a config-loader is the right
+    # thing — out of scope for this CLI which is the daily-pull path.
+    from cyclone import store as store_mod
+    store_mod.store.ensure_clearhouse_seeded()
+    clearhouse = store_mod.store.get_clearhouse()
+    if clearhouse is None or clearhouse.name != sftp_block_name:
+        # Fall back: try the named block, but v1 only ships the
+        # dzinesco singleton.
+        if clearhouse is None:
+            click.echo(
+                f"No clearhouse seeded — cannot find SftpBlock "
+                f"{sftp_block_name!r}.",
+                err=True,
+            )
+            sys.exit(2)
+        click.echo(
+            f"SftpBlock {sftp_block_name!r} not seeded; only "
+            f"{clearhouse.name!r} is available.",
+            err=True,
+        )
+        sys.exit(2)
+    block: SftpBlock = clearhouse.sftp_block
+
+    if file_types_csv:
+        wanted = {t.strip().upper() for t in file_types_csv.split(",") if t.strip()}
+        unknown = wanted - ALLOWED_FILE_TYPES
+        if unknown:
+            click.echo(
+                f"file_types {sorted(unknown)!r} not in {sorted(ALLOWED_FILE_TYPES)}",
+                err=True,
+            )
+            sys.exit(2)
+    else:
+        wanted = {"999", "TA1"}
+
+    # Wire up the scheduler singleton (re-use the same pipeline the
+    # HTTP endpoint uses, including the dedup via processed_inbound_files).
+    scheduler_mod.configure_scheduler(
+        block, sftp_block_name=sftp_block_name, force=True,
+    )
+    sched = scheduler_mod.get_scheduler()
+    client = SftpClient(block)
+
+    async def _run() -> dict:
+        from cyclone.edi.filenames import parse_inbound_filename
+
+        all_files = await _asyncio.to_thread(client.list_inbound_names)
+        matched: list = []
+        for f in all_files:
+            if f.name.find(date_str) == -1:
+                continue
+            try:
+                parsed = parse_inbound_filename(f.name)
+            except ValueError:
+                continue
+            if parsed.file_type not in wanted:
+                continue
+            matched.append(f)
+            if len(matched) >= limit:
+                break
+
+        download_errors: list[str] = []
+        downloaded = 0
+        for f in matched:
+            try:
+                await _asyncio.to_thread(client.download_inbound, f)
+                downloaded += 1
+            except Exception as exc:  # noqa: BLE001
+                download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}")
+
+        tick = await sched.process_inbound_files(matched)
+        return {
+            "listed": len(all_files),
+            "matched": len(matched),
+            "downloaded": downloaded,
+            "download_errors": download_errors,
+            "processed": tick.files_processed,
+            "skipped": tick.files_skipped,
+            "errored": tick.files_errored,
+        }
+
+    try:
+        summary = _asyncio.run(_run())
+    except Exception as exc:  # noqa: BLE001
+        click.echo(f"pull-inbound failed: {type(exc).__name__}: {exc}", err=True)
+        sys.exit(1)
+
+    click.echo(
+        f"date={date_str} file_types={sorted(wanted)} block={sftp_block_name} "
+        f"listed={summary['listed']} matched={summary['matched']} "
+        f"downloaded={summary['downloaded']} processed={summary['processed']} "
+        f"skipped={summary['skipped']} errored={summary['errored']}"
+    )
+    if summary["download_errors"]:
+        click.echo("download errors:", err=True)
+        for e in summary["download_errors"]:
+            click.echo(f"  {e}", err=True)

backend/src/cyclone/edi/filenames.py

@@ -7,9 +7,12 @@ Outbound (we send):
     tp{tpid}-{transaction_type}-{yyyymmddhhmmssSSS_MT}-1of1.{ext}
     Example: tp11525703-837P-20260620132243505-1of1.x12
 
-Inbound (HPE sends to our ToHPE):
+Inbound (HPE sends to our FromHPE):
-    TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12
+    [Tt][Pp]{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12
-    Example: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12
+    Example: tp11525703-837P_M019048402-20260520231513488-1of1_999.x12
+    (legacy / prodfiles may use uppercase TP; the regex is case-insensitive
+    on the prefix to accept both — Gainwell's filer has used both
+    casings over time.)
 
 Both use Mountain Time (MT) timestamps with 17-digit millisecond precision
 (yyyymmddhhmmssSSS = 4+2+2+2+2+2+3 = 17 digits). Sequence is always "1of1"
@@ -39,16 +42,23 @@ OUTBOUND_RE = re.compile(
     r"^tp(?P<tpid>\d+)-(?P<tx>[A-Z0-9]+)-(?P<ts>\d{17})-1of1\.(?P<ext>[A-Za-z0-9]+)$"
 )
 
-# Inbound: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12
+# Inbound: [Tt][Pp]11525703-837P_M019048402-20260520231513488-1of1_999.x12
+#   - prefix:  literal "TP" or "tp" (case-insensitive — Gainwell's
+#     production filer has used both)
 #   - tpid:    1+ digits (inside TP<...>)
-#   - orig_tx: 1+ alnum
+#   - orig_tx: 1+ uppercase alnum
-#   - track:   M + 1+ alnum (e.g. M019048402) — the M is part of the
+#   - track:   M + 1+ uppercase alnum (e.g. M019048402) — the M is
-#     tracking value, not a separator.
+#     part of the tracking value, not a separator.
 #   - ts:      17 digits
 #   - seq:     literal "1of1"
-#   - ft:      1+ alnum (e.g. 999, TA1, 271, 277, 277CA, 820, 834, 835, ENCR)
+#   - ft:      1+ uppercase alnum (e.g. 999, TA1, 271, 277, 277CA,
+#              820, 834, 835, ENCR)
+#
+# Case insensitivity is scoped to the ``TP`` prefix via ``(?i:TP)``
+# — the rest of the pattern is case-sensitive so we still reject a
+# stray ``837p`` or ``m019048402`` (they'd be invalid HCPF).
 INBOUND_RE = re.compile(
-    r"^TP(?P<tpid>\d+)-(?P<orig_tx>[A-Z0-9]+)_(?P<tracking>M[A-Z0-9]+)"
+    r"^(?i:TP)(?P<tpid>\d+)-(?P<orig_tx>[A-Z0-9]+)_(?P<tracking>M[A-Z0-9]+)"
     r"-(?P<ts>\d{17})-1of1_(?P<file_type>[A-Z0-9]+)\.(?P<ext>x12)$"
 )
 
@@ -116,7 +126,9 @@ def parse_inbound_filename(name: str) -> InboundFilename:
     """Parse an inbound HCPF filename.
 
     Args:
-        name: Filename like "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
+        name: Filename like "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
+            (case-insensitive on the ``TP`` prefix; both ``TP`` and
+            ``tp`` are accepted.)
 
     Returns:
         InboundFilename with tpid, orig_tx, tracking, ts, file_type, ext.

backend/src/cyclone/parsers/parse_999.py

@@ -8,6 +8,12 @@ Single-pass walker over the tokenized segment list:
 - AK3 (Segment Context) + AK4 (Element Context) — optional per-segment errors
 - AK5 (Transaction Set Response Status) — per-set accept/reject
 - AK9 (Functional Group Response Status) — per-group counts + ack code
+- IK5 — a non-standard synonym for ``AK5`` that Gainwell's MFT ships
+  in place of the spec-defined ``AK5``. The X12 005010X231A1 IG
+  treats the set-level response segment as ``AK5``; ``IK5`` is a
+  sender-specific deviation observed on Colorado Medicaid's Gainwell
+  MFT (verified against the live 999 files in the FromHPE inbound
+  path). We accept either.
 - SE / GE / IEA
 
 Errors at the file level raise :class:`CycloneParseError`. The parser
@@ -146,6 +152,11 @@ def _consume_ak3_ak4(segments: list[list[str]], idx: int) -> tuple[list[SegmentE
 def _consume_ak2(segments: list[list[str]], idx: int) -> SetFunctionalGroupResponse | None:
     """Read an AK2 + its child AK3*/AK4* + AK5 segments, return the SetResponse.
 
+    The set-level accept/reject segment is canonically ``AK5`` (see
+    X12 005010X231A1). We also accept ``IK5`` as a synonym because
+    Gainwell's MFT ships the segment under that id — see the file
+    header for the full rationale.
+
     Returns None when called with a non-AK2 segment (defensive — the
     orchestrator only calls this when it sees AK2).
     """
@@ -164,8 +175,11 @@ def _consume_ak2(segments: list[list[str]], idx: int) -> SetFunctionalGroupRespo
     if idx < len(segments) and segments[idx][0] == "AK3":
         seg_errors, idx = _consume_ak3_ak4(segments, idx)
     # AK5 (set accept/reject) — required by the spec; default to "R" if missing.
+    # Gainwell's MFT uses IK5 instead of AK5 (sender-specific segment id
+    # that means the same thing); accept either. The default of "R"
+    # matters: if the segment is missing entirely, the 999 is a reject.
     accept_code = "R"
-    if idx < len(segments) and segments[idx][0] == "AK5":
+    if idx < len(segments) and segments[idx][0] in ("AK5", "IK5"):
         ak5 = segments[idx]
         if len(ak5) > 1 and ak5[1]:
             accept_code = ak5[1]
@@ -256,8 +270,11 @@ def parse_999_text(text: str, *, input_file: str = "") -> ParseResult999:
                 set_responses.append(sr)
                 # Advance past the AK2 + AK3*/AK4*/AK5 cluster
                 # (re-walk from i+1 because _consume_ak2 doesn't return idx).
+                # ``IK5`` is the Gainwell-specific synonym for ``AK5``
+                # and must be in the consumed set here too (see
+                # _consume_ak2 for the full rationale).
                 i += 1
-                while i < len(segments) and segments[i][0] in {"AK3", "AK4", "AK5"}:
+                while i < len(segments) and segments[i][0] in {"AK3", "AK4", "AK5", "IK5"}:
                     i += 1
             else:
                 i += 1

backend/src/cyclone/scheduler.py

@@ -155,7 +155,16 @@ def _handle_999(text: str, source_file: str) -> tuple[str, int]:
 
     received, accepted, rejected, ack_code = _ack_count_summary(result)
     icn = result.envelope.control_number
-    synthetic_id = _ack_synthetic_source_batch_id(icn)
+    # The natural unique key for a 999 is the AK2 set_control_number
+    # (= the original claim's patient_control_number). Each 999 ack
+    # covers exactly one claim, so the PCN is 1:1 with the 999 and
+    # far more useful for the operator than the ISA interchange
+    # control number (Gainwell's MFT ships every 999 with the same
+    # default ICN, which used to collapse all 385 daily acks onto
+    # ``999-000000001``). Fall back to ICN → ``unknown`` if the AK2 is
+    # missing.
+    pcn = result.set_responses[0].set_control_number if result.set_responses else None
+    synthetic_id = _ack_synthetic_source_batch_id(icn, pcn=pcn, source_filename=source_file)
 
     with db.SessionLocal()() as session:
         def _lookup(pcn: str):
@@ -337,13 +346,14 @@ def _ack_count_summary(result: Any) -> tuple[int, int, int, str]:
 
     Mirrors the logic in ``cyclone.api._ack_count_summary`` but lives
     here so the scheduler can run without importing the API module.
+
+    Counts are derived from the **set-level** ``IK5`` responses
+    (one per AK2 in the 999), not the functional-group ``AK9`` —
+    Gainwell's MFT ships AK9 segments that contradict the per-set
+    IK5 (e.g. ``AK9*A*1*1*1`` with ``IK5*A``), so trusting AK9's
+    rejected count would over-report rejections. The set-level
+    IK5 is the authoritative per-claim accept/reject signal.
     """
-    if result.functional_group_acks:
-        fg = result.functional_group_acks[0]
-        return (
-            fg.received_count, fg.accepted_count,
-            fg.rejected_count, fg.ack_code,
-        )
     sets = result.set_responses
     received = len(sets)
     accepted = sum(1 for s in sets if s.set_accept_reject.code == "A")
@@ -357,9 +367,43 @@ def _ack_count_summary(result: Any) -> tuple[int, int, int, str]:
     return (received, accepted, rejected, code)
 
 
-def _ack_synthetic_source_batch_id(interchange_control_number: str) -> str:
+def _ack_synthetic_source_batch_id(
-    """Synthetic batches.id for a received 999 with no source batch."""
+    interchange_control_number: str,
-    return f"999-{(interchange_control_number or '').strip() or '000000001'}"
+    *,
+    pcn: str | None = None,
+    source_filename: str | None = None,
+) -> str:
+    """Synthetic batches.id for a received 999 with no source batch.
+
+    Gainwell's MFT ships every 999 with the same default ISA
+    interchange control number (``000000001``), so the ICN alone
+    collapses all daily acks onto one row. The AK2
+    ``set_control_number`` (= the original claim's
+    patient_control_number) is per-batch — Gainwell's 999
+    acks are per-batch, not per-claim, so a daily pull of 385
+    999s typically has only ~4 distinct PCNs. To make every
+    acks row distinguishable in the UI, the source_batch_id
+    always includes an 8-char hash of the inbound filename.
+
+    Precedence for the human-readable part of the id
+    (column is VARCHAR(32)):
+
+      1. ``999-{pcn}-{hash8}`` if AK2 set_control_number is
+         present (the common case). 4+9+1+8 = 22 chars max.
+      2. ``999-{icn}-{hash8}`` if no AK2 (envelope-only 999).
+      3. ``999-{hash12}`` if no filename either (shouldn't
+         happen in production).
+    """
+    import hashlib
+    short_hash = ""
+    if source_filename:
+        short_hash = hashlib.sha1(source_filename.encode("utf-8")).hexdigest()[:8]
+    if pcn and pcn.strip():
+        return f"999-{pcn.strip()}-{short_hash}"
+    icn = (interchange_control_number or "").strip() or "000000001"
+    if short_hash:
+        return f"999-{icn}-{short_hash}"
+    return f"999-{short_hash or icn}"
 
 
 def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str:
@@ -494,6 +538,44 @@ class Scheduler:
         finally:
             self._tick_in_progress = False
 
+    async def process_inbound_files(
+        self, files: list[InboundFile],
+    ) -> TickResult:
+        """Run the per-file processing pipeline over a pre-fetched list.
+
+        Unlike :meth:`tick`, this does **not** call SFTP — the caller
+        is expected to have already downloaded (or arranged the local
+        copy of) each ``InboundFile.local_path``. Used by the
+        ``/api/admin/scheduler/pull-inbound`` admin endpoint and the
+        ``cyclone pull-inbound`` CLI command to process a date-filtered
+        subset of the inbound MFT path without paying the cost of a
+        full poll.
+
+        Honors ``_stop_event`` between files. Concurrent calls are
+        coalesced the same way :meth:`tick` does, so a slow handler
+        can't be stampeded by a parallel admin invocation.
+        """
+        while self._tick_in_progress:
+            await asyncio.sleep(0.05)
+        self._tick_in_progress = True
+        try:
+            started = datetime.now(timezone.utc)
+            result = TickResult(started_at=started, files_seen=len(files))
+            for f in files:
+                if self._stop_event.is_set():
+                    break
+                await self._handle_one(f, result)
+            result.finished_at = datetime.now(timezone.utc)
+            self._last_tick = result
+            self._last_poll_at = result.finished_at
+            self._poll_count += 1
+            self._total_processed += result.files_processed
+            self._total_skipped += result.files_skipped
+            self._total_errored += result.files_errored
+            return result
+        finally:
+            self._tick_in_progress = False
+
     # ---- Internals --------------------------------------------------------
 
     async def _run(self) -> None:
@@ -650,20 +732,21 @@ class Scheduler:
     def _download_and_parse(
         self, f: InboundFile, file_type: str,
     ) -> tuple[Path, str, int]:
-        """Download from MFT, run the right handler. Returns (path, parser, count).
+        """Run the right handler on one inbound file. Returns (path, parser, count).
 
-        Stub mode: ``f.local_path`` already points at the staged file
+        Both stub and real modes read from ``f.local_path`` — the
-        (set by ``SftpClient._list_inbound_stub``). Real mode: the
+        inbound file is already on disk:
-        remote name is ``f.name`` and we round-trip through paramiko.
+          * Stub mode: ``_list_inbound_stub`` points ``local_path`` at
+            the operator-dropped staging file.
+          * Real mode: ``_list_inbound_paramiko`` downloads each
+            ``listdir_attr`` entry into the local cache as part of the
+            listing pass. Re-reading from the MFT would require
+            ``SftpClient.read_file`` with a full remote path, which the
+            scheduler was passing just ``f.name`` for (i.e. a bare
+            filename at the SFTP root, not the inbound dir). Use the
+            cached bytes instead.
         """
-        if self._sftp_block.stub:
-            # In stub mode the InboundFile already has a local_path;
-            # reading the staged bytes directly avoids the stub's
-            # remote-path semantics (which expect a full inbound path).
         content = f.local_path.read_bytes()
-        else:
-            client = self._sftp_client_factory(self._sftp_block)
-            content = client.read_file(f.name)
         text = content.decode("utf-8")
         handler = HANDLERS[file_type]
         parser_used, claim_count = handler(text, f.name)

backend/src/cyclone/store.py

@@ -2370,8 +2370,8 @@ class CycloneStore:
                         "port": 22,
                         "username": "colorado-fts\\coxix_prod_11525703",
                         "paths": {
-                            "outbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
+                            "outbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
-                            "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
+                            "inbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
                         },
                         "stub": True,
                         "staging_dir": "./var/sftp/staging",

backend/tests/fixtures/minimal_999_ik5_gainwell.txt

@@ -0,0 +1,10 @@
+ISA*00*          *00*          *ZZ*COMEDASSISTPROG*ZZ*11525703       *260527*2303*^*00501*000000001*0*P*:~
+GS*FA*COMEDASSISTPROG*11525703*20260527*2303*1*X*005010X231A1~
+ST*999*0001*005010X231A1~
+AK1*HC*1*005010X222A1~
+AK2*837*991102989*005010X222A1~
+IK5*A~
+AK9*A*1*1*1~
+SE*6*0001~
+GE*1*1~
+IEA*1*000000001~

backend/tests/test_api_clearhouse_patch.py

@@ -121,8 +121,8 @@ def test_patch_without_session_returns_401(client):
             "port": 22,
             "username": "colorado-fts\\coxix_prod_11525703",
             "paths": {
-                "outbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
+                "outbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
-                "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
+                "inbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
             },
             "stub": overrides.get("stub", False),
             "staging_dir": "./var/sftp/staging",

backend/tests/test_api_scheduler.py

@@ -34,13 +34,13 @@ def _stub_scheduler_env(tmp_path, monkeypatch):
     db._reset_for_tests()
 
     staging = tmp_path / "staging"
-    inbound = staging / "ToHPE"
+    inbound = staging / "FromHPE"
     inbound.mkdir(parents=True)
     sftp_block = SftpBlock(
         host="mft.example.com",
         port=22,
         username="test",
-        paths={"outbound": "/FromHPE", "inbound": "/ToHPE"},
+        paths={"outbound": "/ToHPE", "inbound": "/FromHPE"},
         stub=True,
         staging_dir=str(staging),
         poll_seconds=60,
@@ -54,7 +54,7 @@ def _stub_scheduler_env(tmp_path, monkeypatch):
 
 
 def _drop_file(staging: Path, name: str, body: bytes) -> Path:
-    p = staging / "ToHPE" / name
+    p = staging / "FromHPE" / name
     p.write_bytes(body)
     return p
 

backend/tests/test_clearhouse_api.py

@@ -30,8 +30,8 @@ def test_get_clearhouse_seeded(client):
     assert body["name"] == "dzinesco"
     assert body["tpid"] == "11525703"
     assert body["sftp_block"]["stub"] is True
-    assert "FromHPE" in body["sftp_block"]["paths"]["outbound"]
+    assert "ToHPE" in body["sftp_block"]["paths"]["outbound"]
-    assert "ToHPE" in body["sftp_block"]["paths"]["inbound"]
+    assert "FromHPE" in body["sftp_block"]["paths"]["inbound"]
 
 
 def test_list_providers(client):

backend/tests/test_filenames.py

@@ -107,6 +107,48 @@ def test_parse_inbound_277():
     assert parsed.file_type == "277"
 
 
+def test_parse_inbound_lowercase_tp_prefix_999():
+    # Gainwell's production filer uses lowercase `tp` for inbound 999/TA1.
+    # The inbound regex must accept both casings on the TP prefix.
+    name = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
+    parsed = parse_inbound_filename(name)
+    assert parsed.tpid == "11525703"
+    assert parsed.orig_tx == "837P"
+    assert parsed.tracking == "M019048402"
+    assert parsed.file_type == "999"
+    assert parsed.ext == "x12"
+
+
+def test_parse_inbound_lowercase_tp_prefix_ta1():
+    name = "tp11525703-837P_M019044969-20260520180505477-1of1_TA1.x12"
+    parsed = parse_inbound_filename(name)
+    assert parsed.file_type == "TA1"
+
+
+def test_is_inbound_filename_accepts_both_cases():
+    # is_inbound_filename() is the fast path used by the scheduler to
+    # filter the listing. It must accept both casings.
+    upper = "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
+    lower = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12"
+    assert is_inbound_filename(upper)
+    assert is_inbound_filename(lower)
+
+
+def test_parse_inbound_rejects_mixed_case_tracking():
+    # Tracking value must stay uppercase alnum; the case-insensitive
+    # flag is intentionally scoped to the TP prefix by the file_type
+    # and timestamp constraints, so a mixed-case tracking should still
+    # be rejected (it'd be invalid HCPF).
+    # We exercise the obvious "totally lowercase" rejection to confirm
+    # the rest of the pattern is still strict.
+    with pytest.raises(ValueError, match="Not a valid HCPF inbound"):
+        # Lowercase orig_tx; the orig_tx class is [A-Z0-9]+ so it
+        # must be uppercase.
+        parse_inbound_filename(
+            "tp11525703-837p_M019048402-20260520231513488-1of1_999.x12"
+        )
+
+
 def test_parse_inbound_rejects_missing_tp_prefix():
     with pytest.raises(ValueError, match="Not a valid HCPF inbound"):
         parse_inbound_filename("11525703-837P_M019048402-20260520231513488-1of1_999.x12")
@@ -157,8 +199,10 @@ def test_is_outbound_filename():
 
 def test_is_inbound_filename():
     assert is_inbound_filename("TP11525703-837P_M019048402-20260520231513488-1of1_999.x12")
-    # Lowercase tp prefix is the outbound shape, not inbound
+    # Lowercase tp prefix is now accepted too — Gainwell's filer has
-    assert not is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12")
+    # used both casings on inbound 999/TA1 files.
+    assert is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12")
+    # Outbound shape is still rejected (no tracking/ts/file_type).
     assert not is_inbound_filename("11525703-837P-20260620132243505-1of1.x12")
 
 

backend/tests/test_parse_999.py

@@ -11,6 +11,12 @@ from cyclone.parsers.parse_999 import parse_999_text
 
 ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_999.txt"
 REJECTED = Path(__file__).parent / "fixtures" / "minimal_999_rejected.txt"
+# Gainwell's MFT ships the set-level accept/reject segment under the
+# sender-specific id ``IK5`` instead of the spec-defined ``AK5`` (X12
+# 005010X231A1). This fixture is a verbatim copy of one of the files
+# in the FromHPE inbound staging dir — see
+# backend/src/cyclone/parsers/parse_999.py for the rationale.
+GAINWELL_IK5 = Path(__file__).parent / "fixtures" / "minimal_999_ik5_gainwell.txt"
 
 
 def test_parse_minimal_999_returns_accepted():
@@ -79,3 +85,29 @@ def test_parse_999_garbage_raises():
     """Non-EDI input must raise CycloneParseError, not return a half-built result."""
     with pytest.raises(CycloneParseError):
         parse_999_text("not edi at all", input_file="bad.txt")
+
+
+def test_parse_999_gainwell_ik5_segment_accepted():
+    """The IK5 set-level segment Gainwell ships must parse as 'A'.
+
+    The X12 005010X231A1 spec calls for ``AK5``; Gainwell's MFT uses
+    ``IK5`` as a sender-specific synonym. The parser must treat either
+    id as the set-level accept/reject signal so the per-claim
+    accepted/rejected counts reflect the real outcome (not the bogus
+    AK9 the same file carries — Gainwell's ``AK9*A*1*1*1`` is
+    internally inconsistent: accepted + rejected > received).
+    """
+    text = GAINWELL_IK5.read_text()
+    result = parse_999_text(text, input_file=GAINWELL_IK5.name)
+    assert len(result.set_responses) == 1
+    s = result.set_responses[0]
+    assert s.set_accept_reject.code == "A"
+    assert s.transaction_set_identifier == "837"
+    assert s.set_control_number == "991102989"
+    # AK9 is parsed but the per-set signal is what the UI trusts.
+    assert result.functional_group_acks[0].ack_code == "A"
+    assert result.functional_group_acks[0].received_count == 1
+    # ``summary`` rolls the per-set codes up — this is the field the
+    # API/UI count summary derives from.
+    assert result.summary.passed == 1
+    assert result.summary.failed == 0

backend/tests/test_providers_seed.py

@@ -54,8 +54,8 @@ def test_seed_creates_clearhouse_singleton():
     assert ch.submitter_name == "Dzinesco"
     assert ch.sftp_block.host == "mft.gainwelltechnologies.com"
     assert ch.sftp_block.stub is True
-    assert "FromHPE" in ch.sftp_block.paths["outbound"]
+    assert "ToHPE" in ch.sftp_block.paths["outbound"]
-    assert "ToHPE" in ch.sftp_block.paths["inbound"]
+    assert "FromHPE" in ch.sftp_block.paths["inbound"]
 
 
 def test_seed_creates_co_txix_payer_with_both_configs():

backend/tests/test_scheduler.py

@@ -35,15 +35,15 @@ from cyclone.scheduler import (
 @pytest.fixture
 def sftp_block(tmp_path):
     staging = tmp_path / "staging"
-    inbound_dir = staging / "ToHPE"
+    inbound_dir = staging / "FromHPE"
     inbound_dir.mkdir(parents=True)
     return SftpBlock(
         host="mft.example.com",
         port=22,
         username="test",
         paths={
-            "outbound": "/FromHPE",
+            "outbound": "/ToHPE",
-            "inbound": "/ToHPE",
+            "inbound": "/FromHPE",
         },
         stub=True,
         staging_dir=str(staging),
@@ -55,7 +55,7 @@ def sftp_block(tmp_path):
 @pytest.fixture
 def _drop_file(sftp_block):
     """Helper: drop a named file in the inbound dir. Returns the path."""
-    inbound_dir = Path(sftp_block.staging_dir) / "ToHPE"
+    inbound_dir = Path(sftp_block.staging_dir) / "FromHPE"
 
     def _drop(name: str, body: bytes) -> Path:
         inbound_dir.mkdir(parents=True, exist_ok=True)
@@ -285,3 +285,61 @@ class TestRoutedFileTypes:
 
     def test_handlers_cover_all_routed_types(self):
         assert set(HANDLERS.keys()) == ROUTED_FILE_TYPES
+
+
+class TestProcessInboundFiles:
+    """Scheduler.process_inbound_files() — date-filtered pull path.
+
+    The /api/admin/scheduler/pull-inbound endpoint and the
+    ``cyclone pull-inbound`` CLI both call this. It must:
+      * process each file in the provided list (no SFTP listdir)
+      * dedupe via ``processed_inbound_files`` (idempotent on rerun)
+      * not touch SFTP at all — files are expected to be on local
+        disk at ``f.local_path``
+    """
+
+    @pytest.mark.asyncio
+    async def test_processes_provided_files_without_listdir(
+        self, sftp_block, _drop_file, tmp_path,
+    ):
+        from cyclone.clearhouse import InboundFile
+        from cyclone.scheduler import STATUS_OK
+
+        # Drop a known 999 file on disk (the stub scheduler doesn't
+        # need SFTP — it reads from staging dir).
+        _drop_file(
+            "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12",
+            b"not a real 999 -- handler will parse_error",
+        )
+        sched = _make_scheduler(sftp_block, tmp_path)
+
+        # Build the InboundFile records manually — caller is
+        # responsible for staging (mirrors the targeted-pull flow).
+        path = tmp_path / "staging" / "FromHPE" / \
+            "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12"
+        files = [InboundFile(
+            name=path.name,
+            size=path.stat().st_size,
+            modified_at=datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc),
+            local_path=path,
+        )]
+
+        result = await sched.process_inbound_files(files)
+        # The file's bytes are intentionally invalid — we only care
+        # that process_inbound_files invokes the handler and
+        # records the outcome (here: error from parse).
+        assert result.files_seen == 1
+        assert result.files_seen == result.files_processed + result.files_errored
+        # Idempotent: a second call is a no-op (already-processed dedup).
+        result2 = await sched.process_inbound_files(files)
+        assert result2.files_seen == 1
+        # Either skipped (because the prior call recorded it as error)
+        # or error — both prove the dedup branch fired. We just check
+        # it's not re-processed successfully.
+        with db.SessionLocal()() as s:
+            rows = s.query(ProcessedInboundFile).filter_by(
+                sftp_block_name="test-block",
+                name=path.name,
+            ).all()
+            assert len(rows) == 1
+            assert rows[0].status in (STATUS_OK, STATUS_ERROR, STATUS_SKIPPED)

backend/tests/test_sftp_paramiko.py

@@ -40,8 +40,8 @@ def _block(
         port=22,
         username="testuser",
         paths={
-            "outbound": "/CO XIX/PROD/test/FromHPE",
+            "outbound": "/CO XIX/PROD/test/ToHPE",
-            "inbound": "/CO XIX/PROD/test/ToHPE",
+            "inbound": "/CO XIX/PROD/test/FromHPE",
         },
         stub=stub,
         staging_dir=staging_dir,
@@ -259,6 +259,66 @@ class TestRealModeListInbound:
         assert len(files) == 1
         assert files[0].name == "real.x12"
 
+    def test_list_skips_warn_txt_files(self, monkeypatch, tmp_path: Path):
+        # Gainwell's MFT drops advisory *_warn.txt files in the same
+        # inbound dir. They're text-format side-channel notes (not X12
+        # envelopes) and must be skipped at list time.
+        real_attr = MagicMock()
+        real_attr.filename = "TP123-837P_M456-20260520231513488-1of1_999.x12"
+        real_attr.st_mode = 0o100644
+        real_attr.st_size = 1024
+        real_attr.st_mtime = 1718899200
+
+        warn_attr = MagicMock()
+        warn_attr.filename = "TP123-837P-202606181208100000-1of1_warn.txt"
+        warn_attr.st_mode = 0o100644
+        warn_attr.st_size = 200
+        warn_attr.st_mtime = 1718899200
+
+        mock_ssh, mock_sftp = _make_mock_paramiko(
+            monkeypatch, sftp_attrs=[real_attr, warn_attr],
+        )
+
+        def _open(path, mode="rb"):
+            m = MagicMock()
+            m.__enter__.return_value = io.BytesIO(b"x12 content")
+            return m
+
+        mock_sftp.open.side_effect = _open
+
+        block = _block(staging_dir=str(tmp_path / "staging"))
+        client = SftpClient(block)
+        files = client.list_inbound()
+
+        names = [f.name for f in files]
+        assert "TP123-837P_M456-20260520231513488-1of1_999.x12" in names
+        assert not any(n.endswith("_warn.txt") for n in names)
+        assert len(files) == 1
+
+    def test_list_inbound_names_does_not_download(self, monkeypatch, tmp_path: Path):
+        # list_inbound_names() must do a metadata-only SFTP listing —
+        # no sftp.open() / no file written to the cache.
+        attr = MagicMock()
+        attr.filename = "TP123-837P_M456-20260520231513488-1of1_999.x12"
+        attr.st_mode = 0o100644
+        attr.st_size = 1024
+        attr.st_mtime = 1718899200
+
+        mock_ssh, mock_sftp = _make_mock_paramiko(
+            monkeypatch, sftp_attrs=[attr],
+        )
+
+        block = _block(staging_dir=str(tmp_path / "staging"))
+        client = SftpClient(block)
+        files = client.list_inbound_names()
+
+        # sftp.open() must NOT have been called (no download).
+        mock_sftp.open.assert_not_called()
+        assert len(files) == 1
+        # The InboundFile.local_path is set to the planned cache path
+        # but the file itself doesn't exist yet.
+        assert not files[0].local_path.exists()
+
 
 class TestRealModeReadFile:
     def test_read_returns_bytes(self, monkeypatch):

backend/tests/test_sftp_stub.py

@@ -19,8 +19,8 @@ def sftp_block(tmp_path):
         port=22,
         username="colorado-fts\\coxix_prod_11525703",
         paths={
-            "outbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
+            "outbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
-            "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
+            "inbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
         },
         stub=True,
         staging_dir=str(staging),
@@ -55,7 +55,7 @@ def test_stub_list_inbound_empty_when_no_local_files(sftp_block):
 
 def test_stub_list_inbound_returns_local_files(sftp_block):
     # Simulate operator dropping a file in the inbound staging dir
-    inbound_dir = Path(sftp_block.staging_dir) / "CO XIX/PROD/coxix_prod_11525703/ToHPE"
+    inbound_dir = Path(sftp_block.staging_dir) / "CO XIX/PROD/coxix_prod_11525703/FromHPE"
     inbound_dir.mkdir(parents=True, exist_ok=True)
     (inbound_dir / "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12").write_bytes(b"X")
     client = SftpClient(sftp_block)
@@ -78,17 +78,17 @@ def test_stub_read_file_returns_bytes(sftp_block, tmp_path):
     Lets the inbound scheduler exercise the same code path on a
     workstation without a real MFT connection.
     """
-    inbound = tmp_path / "staging" / "ToHPE"
+    inbound = tmp_path / "staging" / "FromHPE"
     inbound.mkdir(parents=True)
     (inbound / "TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12").write_bytes(
         b"hello-world",
     )
     client = SftpClient(sftp_block)
-    body = client.read_file("/ToHPE/TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12")
+    body = client.read_file("/FromHPE/TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12")
     assert body == b"hello-world"
 
 
 def test_stub_read_file_missing_raises(sftp_block):
     client = SftpClient(sftp_block)
     with pytest.raises(FileNotFoundError):
-        client.read_file("/ToHPE/does-not-exist.x12")
+        client.read_file("/FromHPE/does-not-exist.x12")

backend/tests/test_store_update_clearhouse.py

@@ -27,8 +27,8 @@ def test_update_clearhouse_round_trip():
         port=22,
         username="colorado-fts\\coxix_prod_11525703",
         paths={
-            "outbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
+            "outbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
-            "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
+            "inbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
         },
         stub=False,
         staging_dir="./var/sftp/staging",

docker-compose.override.yml

@@ -15,6 +15,8 @@ secrets:
     file: /tmp/cyclone-test-secrets/admin_username
   cyclone_admin_password:
     file: /tmp/cyclone-test-secrets/admin_pw
+  cyclone_sftp_password:
+    file: /tmp/cyclone-test-secrets/sftp_password
 
 services:
   frontend:

docs/reference/co-medicaid.md

@@ -25,8 +25,8 @@ the 837P file.
 | Submitter contact | Tyler Martinez <tyler@dzinesco.com> |
 | SFTP host | `mft.gainwelltechnologies.com` |
 | SFTP username | `colorado-fts\coxix_prod_11525703` |
-| SFTP outbound dir | `/CO XIX/PROD/coxix_prod_11525703/FromHPE` |
+| SFTP outbound dir | `/CO XIX/PROD/coxix_prod_11525703/ToHPE` |
-| SFTP inbound dir | `/CO XIX/PROD/coxix_prod_11525703/ToHPE` |
+| SFTP inbound dir | `/CO XIX/PROD/coxix_prod_11525703/FromHPE` |
 
 ## dzinesco's 3 billing-provider NPIs
 
@@ -49,8 +49,8 @@ dzinesco submits 837P files to Gainwell's MFT (Managed File Transfer)
 at `mft.gainwelltechnologies.com`. The full SFTP path layout is
 specified by the user (2026-06-20):
 
-- **Outbound** (we send): `/CO XIX/PROD/coxix_prod_11525703/FromHPE`
+- **Outbound** (we send): `/CO XIX/PROD/coxix_prod_11525703/ToHPE`
-- **Inbound** (HPE/Gainwell sends to us): `/CO XIX/PROD/coxix_prod_11525703/ToHPE`
+- **Inbound** (HPE/Gainwell sends to us): `/CO XIX/PROD/coxix_prod_11525703/FromHPE`
 
 ### File naming
 
@@ -107,8 +107,8 @@ curl -X POST http://localhost:8000/api/clearhouse/submit \
   -H 'Content-Type: application/json' \
   -d '{"claim_ids": ["CLM-1", "CLM-2"], "payer_id": "CO_TXIX"}'
 
-# Files appear at ./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/
+# Files appear at ./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/ToHPE/
-ls -la "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/"
+ls -la "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/ToHPE/"
 ```
 
 ## Payer IDs

docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md

@@ -21,7 +21,7 @@ Replace the single hard-coded `PayerConfig` factory dict (currently in `api.py:9
 - **SP10** — 277CA parser + "Payer-Rejected" lane in the Inbox.
 - **SP11** — Tamper-evident hash-chained `audit_log` table.
 - **SP12** — SQLCipher encryption at rest + Keychain-stored DB key.
-- **SP13** — Replace the SFTP stub with real `paramiko` connection to `mft.gainwelltechnologies.com` and actually push to `/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
+- **SP13** — Replace the SFTP stub with real `paramiko` connection to `mft.gainwelltechnologies.com` and actually push to `/CO XIX/PROD/coxix_prod_11525703/ToHPE`.
 - **Real SFTP credentials in Keychain** — schema and call sites are in place; the actual secret is created manually by the operator.
 
 ## 2. Goals
@@ -109,8 +109,8 @@ VALUES (1, 'dzinesco', '11525703', 'Dzinesco', 'Tyler Martinez', 'tyler@dzinesco
   '"inbound_template":"TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12"}',
   '{"host":"mft.gainwelltechnologies.com","port":22,'
   '"username":"colorado-fts\\coxix_prod_11525703",'
-  '"paths":{"outbound":"/CO XIX/PROD/coxix_prod_11525703/FromHPE",'
+  '"paths":{"outbound":"/CO XIX/PROD/coxix_prod_11525703/ToHPE",'
-  '"inbound":"/CO XIX/PROD/coxix_prod_11525703/ToHPE"},'
+  '"inbound":"/CO XIX/PROD/coxix_prod_11525703/FromHPE"},'
   '"stub":true,"staging_dir":"./var/sftp/staging","poll_seconds":300}',
   '2026-06-20T00:00:00Z');
 ```
@@ -163,13 +163,13 @@ All 3 share the same address because all 3 are registered to the Montrose corpor
 
 Per the **HCPF X12 File Naming Standards Quick Guide** (https://hcpf.colorado.gov/tp-x12-filenaming):
 
-**Outbound** (we send to `/CO XIX/PROD/coxix_prod_11525703/FromHPE`):
+**Outbound** (we send to `/CO XIX/PROD/coxix_prod_11525703/ToHPE`):
 ```
 {tpid}-{transaction_type}-{yyyymmddhhmmssSSS_MT}-1of1.{ext}
 ```
 Example: `11525703-837P-20260620132243505-1of1.x12`
 
-**Inbound** (HPE sends to `/CO XIX/PROD/coxix_prod_11525703/ToHPE`):
+**Inbound** (HPE sends to `/CO XIX/PROD/coxix_prod_11525703/FromHPE`):
 ```
 TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12
 ```
@@ -209,7 +209,7 @@ Handler:
      "ok": true,
      "submitted": [
        {"claim_id": "CLM-001", "filename": "11525703-837P-20260620132243505-1of1.x12",
-        "staging_path": "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/11525703-..."}
+        "staging_path": "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/ToHPE/11525703-..."}
      ],
      "stub": true
    }
@@ -367,7 +367,7 @@ All R200-R210 run on parse AND on serialize, so a 837 file with an unknown NPI f
 
 | # | Question | Resolution |
 |---|----------|-----------|
-| 1 | SFTP outbound + inbound paths on `mft.gainwelltechnologies.com` | `/CO XIX/PROD/coxix_prod_11525703/FromHPE` (out), `/CO XIX/PROD/coxix_prod_11525703/ToHPE` (in) — user-provided 2026-06-20 |
+| 1 | SFTP outbound + inbound paths on `mft.gainwelltechnologies.com` | `/CO XIX/PROD/coxix_prod_11525703/ToHPE` (out), `/CO XIX/PROD/coxix_prod_11525703/FromHPE` (in) — user-provided 2026-06-20 |
 | 2 | 3 NPI street addresses + ZIPs | All 3 NPIs share Montrose corporate address (user confirmed 2026-06-20) — seed from prod files |
 | 3 | 3 NPI taxonomy codes | `251E00000X` for all 3, self-served from 136 prod files |
 | 4 | 277CA filename suffix | Use `277` per HCPF doc; distinguish 277CA by `ST*277CA` content (user confirmed 2026-06-20) |

src/hooks/useTa1Acks.ts

@@ -0,0 +1,26 @@
+import { useQuery } from "@tanstack/react-query";
+import { api, type PaginatedResponse } from "@/lib/api";
+import type { Ta1Ack } from "@/types";
+
+/**
+ * Lists persisted TA1 (Interchange Acknowledgment) rows, newest
+ * first. Mirrors `useAcks` but for the lower-level envelope ack.
+ *
+ * A TA1 is one row per inbound ISA/IEA interchange — distinct from
+ * a 999, which is per-batch. Colorado Medicaid's Gainwell MFT
+ * currently only ships 999s in the FromHPE path, but historically
+ * they've sent TA1s, so the hook stays in the surface for when
+ * they reappear.
+ *
+ * No in-memory fallback: there is no zustand sample-data path for
+ * TA1s in v1. The hook is `enabled: api.isConfigured` so the page
+ * treats an empty list as "no TA1s on file" rather than a
+ * configuration error.
+ */
+export function useTa1Acks(params: { limit?: number } = {}) {
+  return useQuery<PaginatedResponse<Ta1Ack>>({
+    queryKey: ["ta1-acks", params],
+    queryFn: () => api.listTa1Acks(params),
+    enabled: api.isConfigured,
+  });
+}

src/lib/api.ts

@@ -39,6 +39,7 @@ import type {
   Payee835,
   Provider,
   ReassociationTrace,
+  Ta1Ack,
   UnmatchedClaim,
   UnmatchedResponse,
   BatchSummary as ParserBatchSummary,
@@ -778,6 +779,64 @@ async function getAck(id: number): Promise<Ack & { rawJson: unknown }> {
   return { ...mapAck(row), rawJson: row.raw_json };
 }
 
+// ---------------------------------------------------------------------------
+// Public surface — TA1 ACKs
+// The TA1 is the lowest-level X12 envelope ack (one per inbound
+// ISA/IEA interchange), distinct from the per-batch 999. Colorado
+// Medicaid's Gainwell MFT only ships 999s today, but historically
+// they've sent TA1s, so the UI shows them whenever one is on file.
+// ---------------------------------------------------------------------------
+
+interface RawTa1Row {
+  id: number;
+  control_number: string;
+  ack_code: "A" | "E" | "R";
+  note_code: string | null;
+  interchange_date: string | null;
+  interchange_time: string | null;
+  sender_id: string | null;
+  receiver_id: string | null;
+  source_batch_id: string;
+  parsed_at: string | null;
+}
+
+function mapTa1Ack(row: RawTa1Row): Ta1Ack {
+  return {
+    id: row.id,
+    controlNumber: row.control_number,
+    ackCode: row.ack_code,
+    noteCode: row.note_code,
+    interchangeDate: row.interchange_date,
+    interchangeTime: row.interchange_time,
+    senderId: row.sender_id,
+    receiverId: row.receiver_id,
+    sourceBatchId: row.source_batch_id,
+    parsedAt: row.parsed_at,
+  };
+}
+
+async function listTa1Acks(
+  params: { limit?: number } = {},
+): Promise<PaginatedResponse<Ta1Ack>> {
+  if (!isConfigured) throw notConfiguredError();
+  const query: Record<string, unknown> = {};
+  if (params.limit !== undefined) query.limit = params.limit;
+  const body = await authedFetch<{
+    items: RawTa1Row[];
+    total: number;
+  }>(`/api/ta1-acks${qs(query)}`);
+  // The TA1 list endpoint returns `{ total, items }` (no `has_more` /
+  // `returned` — it's a simple cap-based list, not a paginated one).
+  // Synthesize the `PaginatedResponse` shape so the UI can share the
+  // same hook contract as `listAcks`.
+  return {
+    items: body.items.map(mapTa1Ack),
+    total: body.total,
+    returned: body.items.length,
+    has_more: body.items.length < body.total,
+  };
+}
+
 /**
  * Download a ZIP of regenerated X12 837 files for a parsed batch.
  *
@@ -871,4 +930,5 @@ export const api = {
   unmatchClaim,
   listAcks,
   getAck,
+  listTa1Acks,
 };

src/pages/Acks.tsx

@@ -1,5 +1,5 @@
 import { useCallback, useState } from "react";
-import { CheckCircle2, Download, ShieldCheck } from "lucide-react";
+import { CheckCircle2, Download, Mail, ShieldCheck } from "lucide-react";
 import {
   Table,
   TableBody,
@@ -18,11 +18,12 @@ import { KeyboardCheatsheet } from "@/components/KeyboardCheatsheet";
 import { AckDrawer } from "@/components/AckDrawer";
 import { useAckDrawerUrlState } from "@/hooks/useAckDrawerUrlState";
 import { useAcks } from "@/hooks/useAcks";
+import { useTa1Acks } from "@/hooks/useTa1Acks";
 import { useRowKeyboard } from "@/hooks/useRowKeyboard";
 import { api } from "@/lib/api";
 import { fmt } from "@/lib/format";
 import { cn } from "@/lib/utils";
-import type { Ack } from "@/types";
+import type { Ack, Ta1Ack } from "@/types";
 
 /**
  * 999 ACK register. The page reads the persisted 999 Implementation
@@ -365,7 +366,18 @@ export function Acks() {
                                 {a.id}
                               </TableCell>
                               <TableCell className="font-mono text-[12px] text-muted-foreground truncate max-w-[180px]">
-                                {a.sourceBatchId}
+                                {a.patientControlNumber ? (
+                                  <>
+                                    <span className="text-foreground">
+                                      {a.patientControlNumber}
+                                    </span>
+                                    <span className="text-muted-foreground/60">
+                                      {" "}· {a.sourceBatchId}
+                                    </span>
+                                  </>
+                                ) : (
+                                  a.sourceBatchId
+                                )}
                               </TableCell>
                               <TableCell className="text-right display mono text-[hsl(var(--success))]">
                                 {a.acceptedCount}
@@ -400,6 +412,18 @@ export function Acks() {
           </Card>
         </section>
 
+        {/* =================================================================
+            TA1 ENVELOPE ACKS — per-interchange acknowledgments.
+            A TA1 (X12 Interchange Acknowledgment) is one row per
+            inbound ISA/IEA envelope — the lower-level sibling of the
+            999 (per-batch). Colorado Medicaid's Gainwell MFT currently
+            only ships 999s, but the section stays in the surface so
+            the operator can see TA1s as soon as one shows up. Kept as
+            a quieter, narrower section than the 999 register so the
+            999s remain the page's primary instrument.
+            ================================================================= */}
+        <Ta1AcksSection />
+
         {/* =================================================================
             FOOTER — single hairline-separated status row. Replaces the
             warm-paper "End of register" treatment with a quiet
@@ -541,3 +565,195 @@ function downloadBlob(filename: string, content: string) {
   a.click();
   URL.revokeObjectURL(url);
 }
+
+// ---------------------------------------------------------------------------
+// Ta1AcksSection — per-interchange TA1 register.
+//
+// Deliberately quieter than the 999 register above: smaller KPI strip,
+// single-card table, and an empty state that explains the
+// Colorado-specific context (TA1s aren't shipped today, so the
+// section is empty unless Gainwell starts sending them).
+// ---------------------------------------------------------------------------
+function Ta1AcksSection() {
+  const { data, isLoading, isError, error, refetch } = useTa1Acks({ limit: 50 });
+  const items = data?.items ?? [];
+
+  const totals = items.reduce(
+    (acc, t) => {
+      acc[t.ackCode] = (acc[t.ackCode] ?? 0) + 1;
+      return acc;
+    },
+    {} as Record<string, number>,
+  );
+
+  return (
+    <section
+      aria-label="TA1 envelope acknowledgments"
+      className="animate-fade-in-up"
+    >
+      <Card>
+        <CardContent className="p-6 lg:p-7 space-y-5">
+          <div className="flex items-end justify-between gap-6 flex-wrap">
+            <div className="min-w-0">
+              <div className="eyebrow flex items-center gap-2 mb-2">
+                <span className="inline-block h-px w-6 bg-foreground/20" />
+                Envelope acks
+              </div>
+              <h2 className="display text-[22px] leading-[1.05] tracking-[-0.02em]">
+                TA1 <span className="italic">envelopes</span>, newest first.
+              </h2>
+            </div>
+            <p className="text-[12.5px] text-muted-foreground/80 max-w-sm">
+              One row per inbound ISA/IEA interchange — the
+              envelope-level sibling of the 999. Gainwell's Colorado
+              MFT does not ship TA1s today; this section surfaces them
+              when they appear.
+            </p>
+          </div>
+
+          <div className="grid gap-3 grid-cols-2 md:grid-cols-4 pt-2">
+            <KpiCard
+              label="On file"
+              value={fmt.num(items.length)}
+              accent="default"
+              hint="persisted TA1s"
+            />
+            <KpiCard
+              label="Accepted"
+              value={fmt.num(totals.A ?? 0)}
+              accent="success"
+              hint="ack_code = A"
+            />
+            <KpiCard
+              label="Envelope errors"
+              value={fmt.num(totals.E ?? 0)}
+              accent="warning"
+              hint="ack_code = E"
+            />
+            <KpiCard
+              label="Rejected"
+              value={fmt.num(totals.R ?? 0)}
+              accent="destructive"
+              hint="ack_code = R"
+            />
+          </div>
+
+          <div className="pt-5 border-t border-border/40">
+            {isError ? (
+              <ErrorState
+                message="Couldn't load TA1 acks from the backend."
+                detail={error instanceof Error ? error.message : String(error)}
+                onRetry={() => refetch()}
+              />
+            ) : isLoading ? (
+              <div className="rounded-md border border-border/60 bg-card/40 p-4 space-y-2">
+                {Array.from({ length: 3 }).map((_, i) => (
+                  <Skeleton key={i} variant="row" />
+                ))}
+              </div>
+            ) : items.length === 0 ? (
+              <div className="rounded-md border border-dashed border-border/60 bg-card/40">
+                <EmptyState
+                  eyebrow="TA1 envelopes · none on file"
+                  message="Gainwell's Colorado MFT does not ship TA1 files today. When one does arrive (or you upload one), it will land here."
+                />
+              </div>
+            ) : (
+              <div className="rounded-md border border-border/60 overflow-hidden bg-card/40">
+                <Table>
+                  <TableHeader>
+                    <TableRow>
+                      <TableHead className="w-10" aria-label="Status" />
+                      <TableHead>Control #</TableHead>
+                      <TableHead>Ack</TableHead>
+                      <TableHead>Note</TableHead>
+                      <TableHead>Interchange date</TableHead>
+                      <TableHead>Sender → Receiver</TableHead>
+                    </TableRow>
+                  </TableHeader>
+                  <TableBody>
+                    {items.map((t) => (
+                      <TableRow key={t.id}>
+                        <TableCell>
+                          <Mail
+                            className={cn(
+                              "h-3.5 w-3.5",
+                              t.ackCode === "A"
+                                ? "text-[hsl(var(--success))]"
+                                : t.ackCode === "R"
+                                  ? "text-destructive"
+                                  : "text-[hsl(var(--warning))]",
+                            )}
+                            strokeWidth={1.75}
+                            aria-hidden
+                          />
+                        </TableCell>
+                        <TableCell className="display mono text-[13px] text-foreground">
+                          {t.controlNumber || "—"}
+                        </TableCell>
+                        <TableCell>
+                          <Ta1CodeBadge code={t.ackCode} />
+                        </TableCell>
+                        <TableCell className="mono text-[12px] text-muted-foreground">
+                          {t.noteCode ?? "—"}
+                        </TableCell>
+                        <TableCell className="mono text-[12.5px] text-muted-foreground/70">
+                          {t.interchangeDate ? fmt.dateShort(t.interchangeDate) : "—"}
+                          {t.interchangeTime ? (
+                            <span className="text-muted-foreground/50"> · {t.interchangeTime}</span>
+                          ) : null}
+                        </TableCell>
+                        <TableCell className="mono text-[12px] text-muted-foreground truncate max-w-[260px]">
+                          {t.senderId ?? "?"} → {t.receiverId ?? "?"}
+                        </TableCell>
+                      </TableRow>
+                    ))}
+                  </TableBody>
+                </Table>
+              </div>
+            )}
+          </div>
+        </CardContent>
+      </Card>
+    </section>
+  );
+}
+
+// ---------------------------------------------------------------------------
+// Ta1CodeBadge — same A/E/R → success/warning/destructive mapping as the
+// 999 AckCodeBadge above. Kept as a separate component so the TA1
+// surface can evolve independently (e.g. add a tooltip explaining the
+// X12 005010X231A1 note code table) without churning the 999 surface.
+// ---------------------------------------------------------------------------
+function Ta1CodeBadge({ code }: { code: Ta1Ack["ackCode"] }) {
+  const color =
+    code === "A"
+      ? {
+          text: "hsl(var(--success))",
+          bg: "hsl(var(--success) / 0.10)",
+          border: "hsl(var(--success) / 0.30)",
+        }
+      : code === "R"
+        ? {
+            text: "hsl(var(--destructive))",
+            bg: "hsl(var(--destructive) / 0.10)",
+            border: "hsl(var(--destructive) / 0.30)",
+          }
+        : {
+            text: "hsl(var(--warning))",
+            bg: "hsl(var(--warning) / 0.10)",
+            border: "hsl(var(--warning) / 0.30)",
+          };
+  return (
+    <span
+      className="inline-flex items-center gap-1 rounded-sm border px-2 py-0.5 mono text-[10.5px] font-semibold uppercase tracking-[0.14em]"
+      style={{
+        color: color.text,
+        backgroundColor: color.bg,
+        borderColor: color.border,
+      }}
+    >
+      {code}
+    </span>
+  );
+}

src/types/index.ts

@@ -529,6 +529,46 @@ export interface Ack {
   receivedCount: number;
   ackCode: "A" | "E" | "R" | "P";
   parsedAt: string;
+  /**
+   * AK2 set_control_number from the inbound 999 — the
+   * patient_control_number of the original claim batch this 999
+   * acks. Surfaced in the list endpoint so the operator can
+   * correlate a 999 to a claim batch in the UI without a second
+   * round-trip to the detail endpoint.
+   */
+  patientControlNumber?: string | null;
+}
+
+// ---------------------------------------------------------------------------
+// TA1 (Interchange Acknowledgment) — mirrors the `ta1_acks` table
+// and the `/api/ta1-acks` response. A TA1 is the lowest-level X12
+// envelope ack: one per inbound ISA/IEA interchange, separate from
+// the per-batch 999. Colorado Medicaid currently doesn't ship TA1s
+// in the FromHPE path (the inbound MFT only carries 999s) but
+// Gainwell has shipped them historically, so the UI shows them
+// whenever one is on file.
+// ---------------------------------------------------------------------------
+
+/**
+ * One persisted TA1 ACK row, camelCased for the UI. Re-shaped in
+ * `src/lib/api.ts` (`mapTa1Ack`) from the snake_case backend payload.
+ *
+ * ``ackCode`` is the interchange ack code: ``A`` = accepted,
+ * ``E`` = accepted with envelope errors, ``R`` = rejected.
+ * ``noteCode`` is an optional 3-digit X12 note code (e.g. ``000`` = no
+ * error, ``001`` = unsupported interchange version, etc.).
+ */
+export interface Ta1Ack {
+  id: number;
+  controlNumber: string;
+  ackCode: "A" | "E" | "R";
+  noteCode: string | null;
+  interchangeDate: string | null;
+  interchangeTime: string | null;
+  senderId: string | null;
+  receiverId: string | null;
+  sourceBatchId: string;
+  parsedAt: string | null;
 }
 
 // ---------------------------------------------------------------------------