From dd7da182797f7affe672372782c276cd25af5407 Mon Sep 17 00:00:00 2001 From: Nora Date: Wed, 24 Jun 2026 22:05:55 -0600 Subject: [PATCH 1/6] =?UTF-8?q?fix(sftp):=20swap=20inbound/outbound=20path?= =?UTF-8?q?s=20=E2=80=94=20FromHPE=20is=20inbound,=20ToHPE=20is=20outbound?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dzinesco SP9 seed had `paths.inbound` and `paths.outbound` mapped to the wrong Gainwell MFT directories: - paths.outbound was FromHPE/ (HPE sends files FROM here TO us — inbound) - paths.inbound was ToHPE/ (we send files TO here — outbound) So `/api/clearhouse/submit` was writing 837P claims to FromHPE (where HPE puts acks/835s) and the SP16 scheduler was polling ToHPE (where our claims go). 999 / TA1 / 835 files in the real FromHPE inbox were unreachable. Semantics per operator (2026-06-24): - FromHPE = HPE/Gainwell → us = 999, TA1, 835 (inbound) - ToHPE = us → HPE/Gainwell = 837P claims (outbound) **Runtime code (the actual fix):** - backend/src/cyclone/store.py — SP9 seed paths flipped - backend/src/cyclone/edi/filenames.py — docstring corrected **Test fixtures + assertions (would otherwise fail on the new seed):** - backend/tests/test_clearhouse_api.py - backend/tests/test_providers_seed.py - backend/tests/test_sftp_stub.py — incl. inbound dir paths - backend/tests/test_sftp_paramiko.py - backend/tests/test_store_update_clearhouse.py - backend/tests/test_api_clearhouse_patch.py - backend/tests/test_scheduler.py - backend/tests/test_api_scheduler.py **Docs (text-only — keeps the codebase self-consistent):** - README.md - docs/reference/co-medicaid.md - docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md **Operator action required after merge:** the existing clearhouse row in `~/.local/share/cyclone/cyclone.db` was seeded with the old wrong paths. Easiest recovery: 1. `rm ~/.local/share/cyclone/cyclone.db` and let `ensure_clearhouse_seeded` re-run on next boot, OR 2. PATCH /api/clearhouse with the new `paths` block (the SP25 reconfigure hook picks it up live, including by the running scheduler). **Verification:** - 82 tests in the affected files pass (test_clearhouse_api, test_providers_seed, test_sftp_stub, test_sftp_paramiko, test_store_update_clearhouse, test_api_clearhouse_patch, test_scheduler, test_api_scheduler, test_filenames). - Full backend suite: 1029 pass + 36 pre-existing order-dependent flakes unrelated to this change (verified by running the same tests in isolation). --- README.md | 6 +++--- backend/src/cyclone/edi/filenames.py | 2 +- backend/src/cyclone/store.py | 4 ++-- backend/tests/test_api_clearhouse_patch.py | 4 ++-- backend/tests/test_api_scheduler.py | 6 +++--- backend/tests/test_clearhouse_api.py | 4 ++-- backend/tests/test_providers_seed.py | 4 ++-- backend/tests/test_scheduler.py | 8 ++++---- backend/tests/test_sftp_paramiko.py | 4 ++-- backend/tests/test_sftp_stub.py | 12 ++++++------ backend/tests/test_store_update_clearhouse.py | 4 ++-- docs/reference/co-medicaid.md | 12 ++++++------ ...26-06-20-cyclone-multi-payer-npi-sftp-design.md | 14 +++++++------- 13 files changed, 42 insertions(+), 42 deletions(-) diff --git a/README.md b/README.md index 5a2497c..79ec516 100644 --- a/README.md +++ b/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 diff --git a/backend/src/cyclone/edi/filenames.py b/backend/src/cyclone/edi/filenames.py index 92157d1..52c213e 100644 --- a/backend/src/cyclone/edi/filenames.py +++ b/backend/src/cyclone/edi/filenames.py @@ -7,7 +7,7 @@ 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 Example: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12 diff --git a/backend/src/cyclone/store.py b/backend/src/cyclone/store.py index 5079658..7449357 100644 --- a/backend/src/cyclone/store.py +++ b/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", - "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE", + "outbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE", + "inbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE", }, "stub": True, "staging_dir": "./var/sftp/staging", diff --git a/backend/tests/test_api_clearhouse_patch.py b/backend/tests/test_api_clearhouse_patch.py index 3717a0e..e4259d0 100644 --- a/backend/tests/test_api_clearhouse_patch.py +++ b/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", - "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE", + "outbound": "/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", diff --git a/backend/tests/test_api_scheduler.py b/backend/tests/test_api_scheduler.py index cb05a31..79b7dd6 100644 --- a/backend/tests/test_api_scheduler.py +++ b/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 diff --git a/backend/tests/test_clearhouse_api.py b/backend/tests/test_clearhouse_api.py index 7057bbd..bf41827 100644 --- a/backend/tests/test_clearhouse_api.py +++ b/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"]["inbound"] + assert "ToHPE" in body["sftp_block"]["paths"]["outbound"] + assert "FromHPE" in body["sftp_block"]["paths"]["inbound"] def test_list_providers(client): diff --git a/backend/tests/test_providers_seed.py b/backend/tests/test_providers_seed.py index 42d82b9..27bb18e 100644 --- a/backend/tests/test_providers_seed.py +++ b/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["inbound"] + assert "ToHPE" in ch.sftp_block.paths["outbound"] + assert "FromHPE" in ch.sftp_block.paths["inbound"] def test_seed_creates_co_txix_payer_with_both_configs(): diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py index 3a19fc4..db13929 100644 --- a/backend/tests/test_scheduler.py +++ b/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", - "inbound": "/ToHPE", + "outbound": "/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) diff --git a/backend/tests/test_sftp_paramiko.py b/backend/tests/test_sftp_paramiko.py index fe0e4b0..affea05 100644 --- a/backend/tests/test_sftp_paramiko.py +++ b/backend/tests/test_sftp_paramiko.py @@ -40,8 +40,8 @@ def _block( port=22, username="testuser", paths={ - "outbound": "/CO XIX/PROD/test/FromHPE", - "inbound": "/CO XIX/PROD/test/ToHPE", + "outbound": "/CO XIX/PROD/test/ToHPE", + "inbound": "/CO XIX/PROD/test/FromHPE", }, stub=stub, staging_dir=staging_dir, diff --git a/backend/tests/test_sftp_stub.py b/backend/tests/test_sftp_stub.py index 8ce3548..bf390ec 100644 --- a/backend/tests/test_sftp_stub.py +++ b/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", - "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE", + "outbound": "/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") diff --git a/backend/tests/test_store_update_clearhouse.py b/backend/tests/test_store_update_clearhouse.py index 51fe765..c06f26d 100644 --- a/backend/tests/test_store_update_clearhouse.py +++ b/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", - "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE", + "outbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE", + "inbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE", }, stub=False, staging_dir="./var/sftp/staging", diff --git a/docs/reference/co-medicaid.md b/docs/reference/co-medicaid.md index 009008a..fc8dd26 100644 --- a/docs/reference/co-medicaid.md +++ b/docs/reference/co-medicaid.md @@ -25,8 +25,8 @@ the 837P file. | Submitter contact | Tyler Martinez | | SFTP host | `mft.gainwelltechnologies.com` | | SFTP username | `colorado-fts\coxix_prod_11525703` | -| SFTP outbound dir | `/CO XIX/PROD/coxix_prod_11525703/FromHPE` | -| SFTP inbound dir | `/CO XIX/PROD/coxix_prod_11525703/ToHPE` | +| SFTP outbound 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` -- **Inbound** (HPE/Gainwell sends to us): `/CO XIX/PROD/coxix_prod_11525703/ToHPE` +- **Outbound** (we send): `/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/ -ls -la "./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/ToHPE/" ``` ## Payer IDs diff --git a/docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md b/docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md index 55cb54a..0d0e69a 100644 --- a/docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md +++ b/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",' - '"inbound":"/CO XIX/PROD/coxix_prod_11525703/ToHPE"},' + '"paths":{"outbound":"/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) | From a436538c155446b1a3d381bf042c03a0f54e8281 Mon Sep 17 00:00:00 2001 From: Nora Date: Wed, 24 Jun 2026 22:15:52 -0600 Subject: [PATCH 2/6] fix(scheduler): read inbound bytes from the cached local_path, not read_file(f.name) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In real (paramiko) mode, `_download_and_parse` called `client.read_file(f.name)` with a bare filename. paramiko's `sftp.open(f.name)` opens at the SFTP root, not at `paths.inbound` (FromHPE) — so the scheduler would fail to download any file in real mode even with the path swap from the previous commit. But this round-trip is also unnecessary: `_list_inbound_paramiko` already downloads each entry into the local cache (cache_path) and returns it as `InboundFile.local_path` as part of the listing pass. Reading from disk is faster than re-fetching and avoids the path bug. Stub mode was already reading from `f.local_path`. Now both modes do, which is the simpler invariant. Verified: 36 tests pass (test_scheduler, test_api_scheduler, test_sftp_stub, test_sftp_paramiko). --- backend/src/cyclone/scheduler.py | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 2734f0f..3aac5cc 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -650,20 +650,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 - (set by ``SftpClient._list_inbound_stub``). Real mode: the - remote name is ``f.name`` and we round-trip through paramiko. + Both stub and real modes read from ``f.local_path`` — the + inbound file is already on disk: + * 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) + content = f.local_path.read_bytes() text = content.decode("utf-8") handler = HANDLERS[file_type] parser_used, claim_count = handler(text, f.name) From c3a6c5309661646e961ce930b7fc8022b799a66b Mon Sep 17 00:00:00 2001 From: tyler Date: Wed, 24 Jun 2026 23:23:46 -0600 Subject: [PATCH 3/6] fix(sftp): case-insensitive inbound regex, skip _warn.txt, add targeted pull MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three changes that unblock the daily inbound pull from Gainwell's FromHPE MFT path: 1. INBOUND_RE now accepts both 'TP' and 'tp' prefixes via inline (?i:TP) scoping — case-folding the whole pattern would let lowercase '837p' / tracking IDs through, which is invalid HCPF. The rest of the pattern is still case-sensitive. 2. _list_inbound_paramiko skips *_warn.txt entries. Gainwell's MFT drops ~583 advisory text-format notes in the same inbound dir; they come first alphabetically and were padding every poll with ~80 min of pointless downloads. 3. New SftpClient.list_inbound_names() + download_inbound() pair gives a metadata-only listing and on-demand fetch. The scheduler's existing full-listing path still works (now without the warn padding); the new path is what the new /api/admin/scheduler/pull-inbound endpoint and the 'cyclone pull-inbound' CLI use to fast-target a date range without paying the cost of a full ~6000-file download. Scheduler.process_inbound_files() runs the same per-file pipeline as a regular tick on the pre-fetched list, so dedup via processed_inbound_files still applies. Tests added in test_filenames.py (lowercase + mixed-case cases), test_sftp_paramiko.py (warn skip + no-download listing), and test_scheduler.py (process_inbound_files idempotency). With this, the daily 385-file pull for 20260624 completes in seconds via 'docker exec cyclone-backend-1 python -m cyclone pull-inbound --date 20260624 --block dzinesco' (or the equivalent POST to /api/admin/scheduler/pull-inbound?date=20260624). --- backend/src/cyclone/api.py | 124 ++++++++++++++++ backend/src/cyclone/clearhouse/__init__.py | 115 +++++++++++++++ backend/src/cyclone/cli.py | 164 +++++++++++++++++++++ backend/src/cyclone/edi/filenames.py | 30 ++-- backend/src/cyclone/scheduler.py | 38 +++++ backend/tests/test_filenames.py | 48 +++++- backend/tests/test_scheduler.py | 60 +++++++- backend/tests/test_sftp_paramiko.py | 60 ++++++++ 8 files changed, 627 insertions(+), 12 deletions(-) diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 7a67a2a..dba36d3 100644 --- a/backend/src/cyclone/api.py +++ b/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 @@ -3419,6 +3421,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).""" diff --git a/backend/src/cyclone/clearhouse/__init__.py b/backend/src/cyclone/clearhouse/__init__.py index 9fcba7c..2144375 100644 --- a/backend/src/cyclone/clearhouse/__init__.py +++ b/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() diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py index 931d1bb..2c3ddf6 100644 --- a/backend/src/cyclone/cli.py +++ b/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) diff --git a/backend/src/cyclone/edi/filenames.py b/backend/src/cyclone/edi/filenames.py index 52c213e..ad11727 100644 --- a/backend/src/cyclone/edi/filenames.py +++ b/backend/src/cyclone/edi/filenames.py @@ -8,8 +8,11 @@ Outbound (we send): Example: tp11525703-837P-20260620132243505-1of1.x12 Inbound (HPE sends to our FromHPE): - TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12 - Example: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12 + [Tt][Pp]{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.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\d+)-(?P[A-Z0-9]+)-(?P\d{17})-1of1\.(?P[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 -# - track: M + 1+ alnum (e.g. M019048402) — the M is part of the -# tracking value, not a separator. +# - orig_tx: 1+ uppercase alnum +# - track: M + 1+ uppercase alnum (e.g. M019048402) — the M is +# 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\d+)-(?P[A-Z0-9]+)_(?PM[A-Z0-9]+)" + r"^(?i:TP)(?P\d+)-(?P[A-Z0-9]+)_(?PM[A-Z0-9]+)" r"-(?P\d{17})-1of1_(?P[A-Z0-9]+)\.(?Px12)$" ) @@ -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. diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 3aac5cc..3569cb8 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -494,6 +494,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: diff --git a/backend/tests/test_filenames.py b/backend/tests/test_filenames.py index b802539..a566273 100644 --- a/backend/tests/test_filenames.py +++ b/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 - assert not is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12") + # Lowercase tp prefix is now accepted too — Gainwell's filer has + # 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") diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py index db13929..72e0a3f 100644 --- a/backend/tests/test_scheduler.py +++ b/backend/tests/test_scheduler.py @@ -284,4 +284,62 @@ class TestRoutedFileTypes: type. These tests are the regression net.""" def test_handlers_cover_all_routed_types(self): - assert set(HANDLERS.keys()) == ROUTED_FILE_TYPES \ No newline at end of file + 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) \ No newline at end of file diff --git a/backend/tests/test_sftp_paramiko.py b/backend/tests/test_sftp_paramiko.py index affea05..6ec55db 100644 --- a/backend/tests/test_sftp_paramiko.py +++ b/backend/tests/test_sftp_paramiko.py @@ -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): From 1381a7652d30eef393c0b2bb1d82d06d50d6bf0c Mon Sep 17 00:00:00 2001 From: tyler Date: Wed, 24 Jun 2026 23:55:58 -0600 Subject: [PATCH 4/6] fix(acks): make 999 source_batch_id unique per file + surface PCN MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gainwell's MFT ships every 999 with the same ISA interchange control number (`000000001`) and one 999 ack covers a whole batch, not a single claim — so the AK2 set_control_number (patient_control_number) is the same for the ~96 999s in a batch. With the old synthetic-id formula (`999-{icn}`), all 385 daily acks collapsed onto a single row the operator couldn't distinguish. The new formula is `999-{pcn}-{filename_hash8}` (or `999-{icn}-{filename_hash8}` for envelope-only 999s without an AK2). The PCN gives the operator a human-readable handle to the claim batch; the 8-char hash of the inbound filename guarantees uniqueness within a batch. Fits in the VARCHAR(32) source_batch_id column (max 22 chars). Also surface `patient_control_number` in /api/acks list response (extracted from raw_json's set_responses[0].set_control_number) and in the Acks UI as the primary label, with the synthetic id shown dimmed after a middle dot. The detail endpoint already exposed raw_json for the full 999 parse tree. --- backend/src/cyclone/api_routers/acks.py | 17 ++++++++- backend/src/cyclone/scheduler.py | 51 +++++++++++++++++++++++-- src/lib/api.ts | 2 + src/pages/Acks.tsx | 13 ++++++- src/types/index.ts | 8 ++++ 5 files changed, 85 insertions(+), 6 deletions(-) diff --git a/backend/src/cyclone/api_routers/acks.py b/backend/src/cyclone/api_routers/acks.py index 3ea8f09..d95b8a8 100644 --- a/backend/src/cyclone/api_routers/acks.py +++ b/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") diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 3569cb8..e32153d 100644 --- a/backend/src/cyclone/scheduler.py +++ b/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): @@ -357,9 +366,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: - """Synthetic batches.id for a received 999 with no source batch.""" - return f"999-{(interchange_control_number or '').strip() or '000000001'}" +def _ack_synthetic_source_batch_id( + interchange_control_number: str, + *, + 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: diff --git a/src/lib/api.ts b/src/lib/api.ts index 0e5d302..ead3361 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -738,6 +738,7 @@ interface RawAckRow { received_count: number; ack_code: "A" | "E" | "R" | "P"; parsed_at: string; + patient_control_number?: string | null; } function mapAck(row: RawAckRow): Ack { @@ -749,6 +750,7 @@ function mapAck(row: RawAckRow): Ack { receivedCount: row.received_count, ackCode: row.ack_code, parsedAt: row.parsed_at, + patientControlNumber: row.patient_control_number ?? null, }; } diff --git a/src/pages/Acks.tsx b/src/pages/Acks.tsx index f707e11..1ee2652 100644 --- a/src/pages/Acks.tsx +++ b/src/pages/Acks.tsx @@ -365,7 +365,18 @@ export function Acks() { {a.id} - {a.sourceBatchId} + {a.patientControlNumber ? ( + <> + + {a.patientControlNumber} + + + {" "}· {a.sourceBatchId} + + + ) : ( + a.sourceBatchId + )} {a.acceptedCount} diff --git a/src/types/index.ts b/src/types/index.ts index d307918..6dab372 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -529,6 +529,14 @@ 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; } // --------------------------------------------------------------------------- From 6507a8c874939b3d96a1fdb20e260f99148fd958 Mon Sep 17 00:00:00 2001 From: Nora Date: Thu, 25 Jun 2026 00:26:13 -0600 Subject: [PATCH 5/6] fix(acks): accept IK5 from Gainwell, trust set-level codes over bogus AK9, surface TA1 in UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related fixes land together because the UI was reporting "1 accepted 1 rejected" for every 999 even though every inbound file Gainwell ships has IK5=A. 1. Gainwell's MFT uses IK5 where the X12 005010X231A1 spec calls for AK5 (the per-set accept/reject segment). The parser only recognized AK5, so set_responses[0].set_accept_reject.code defaulted to 'R' and the count summary showed all rejections. _consume_ak2 now accepts either AK5 or IK5; the orchestrator's segment-skip set picks up IK5 too. A new fixture (minimal_999_ik5_gainwell.txt) is a verbatim copy of one of the files in the FromHPE inbound staging dir. 2. _ack_count_summary (api + scheduler) now trusts the per-set IK5 codes over the functional-group AK9. Gainwell's AK9 is internally inconsistent — the per-claim IK5=A but the AK9 reports accepted=1, rejected=1, received=1 (sum exceeds received). Trusting the per-set codes restores the right answer: accepted=1, rejected=0, code='A'. 3. The Acks page now has a TA1 envelope register alongside the 999 register. TA1s are the lower-level sibling of the 999 (one row per inbound ISA/IEA). The backend surface (parser, store, API at /api/ta1-acks) was already in place; this adds the UI: Ta1Ack type, listTa1Acks API method, useTa1Acks hook, and a Ta1AcksSection card with KPIs + table. After reprocessing 1056 cached 999s through the new code: every row shows code='A' with accepted=1, rejected=0 — matches the Gainwell portal's per-claim accepted state. The user's earlier observation ("the claims look to be accepted in the portal") was correct: the underlying claim state was always fine, only the displayed count was wrong. - backend/src/cyclone/parsers/parse_999.py | 21 ++- - backend/src/cyclone/api.py | 11 +- - backend/src/cyclone/scheduler.py | 13 +- - backend/tests/test_parse_999.py | 32 ++++ - backend/tests/fixtures/minimal_999_ik5_gainwell.txt - src/types/index.ts | 32 ++++ - src/lib/api.ts | 62 +++++- - src/hooks/useTa1Acks.ts | 26 +++ (new) - src/pages/Acks.tsx | 209 +++++++++++++++++++- --- backend/src/cyclone/api.py | 11 +- backend/src/cyclone/parsers/parse_999.py | 21 +- backend/src/cyclone/scheduler.py | 13 +- .../fixtures/minimal_999_ik5_gainwell.txt | 10 + backend/tests/test_parse_999.py | 32 +++ src/hooks/useTa1Acks.ts | 26 +++ src/lib/api.ts | 62 +++++- src/pages/Acks.tsx | 209 +++++++++++++++++- src/types/index.ts | 32 +++ 9 files changed, 399 insertions(+), 17 deletions(-) create mode 100644 backend/tests/fixtures/minimal_999_ik5_gainwell.txt create mode 100644 src/hooks/useTa1Acks.ts diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index dba36d3..e30563e 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -669,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 - to summing per-set codes if no AK9 was found. + 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") diff --git a/backend/src/cyclone/parsers/parse_999.py b/backend/src/cyclone/parsers/parse_999.py index 00aa1f5..1173bcd 100644 --- a/backend/src/cyclone/parsers/parse_999.py +++ b/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 diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index e32153d..1f5724b 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -346,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") diff --git a/backend/tests/fixtures/minimal_999_ik5_gainwell.txt b/backend/tests/fixtures/minimal_999_ik5_gainwell.txt new file mode 100644 index 0000000..e6deecc --- /dev/null +++ b/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~ diff --git a/backend/tests/test_parse_999.py b/backend/tests/test_parse_999.py index d3c5208..8cfbe85 100644 --- a/backend/tests/test_parse_999.py +++ b/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 diff --git a/src/hooks/useTa1Acks.ts b/src/hooks/useTa1Acks.ts new file mode 100644 index 0000000..0385eac --- /dev/null +++ b/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>({ + queryKey: ["ta1-acks", params], + queryFn: () => api.listTa1Acks(params), + enabled: api.isConfigured, + }); +} diff --git a/src/lib/api.ts b/src/lib/api.ts index ead3361..9984b17 100644 --- a/src/lib/api.ts +++ b/src/lib/api.ts @@ -39,6 +39,7 @@ import type { Payee835, Provider, ReassociationTrace, + Ta1Ack, UnmatchedClaim, UnmatchedResponse, BatchSummary as ParserBatchSummary, @@ -738,7 +739,6 @@ interface RawAckRow { received_count: number; ack_code: "A" | "E" | "R" | "P"; parsed_at: string; - patient_control_number?: string | null; } function mapAck(row: RawAckRow): Ack { @@ -750,7 +750,6 @@ function mapAck(row: RawAckRow): Ack { receivedCount: row.received_count, ackCode: row.ack_code, parsedAt: row.parsed_at, - patientControlNumber: row.patient_control_number ?? null, }; } @@ -780,6 +779,64 @@ async function getAck(id: number): Promise { 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> { + if (!isConfigured) throw notConfiguredError(); + const query: Record = {}; + 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. * @@ -873,4 +930,5 @@ export const api = { unmatchClaim, listAcks, getAck, + listTa1Acks, }; diff --git a/src/pages/Acks.tsx b/src/pages/Acks.tsx index 1ee2652..f909e3b 100644 --- a/src/pages/Acks.tsx +++ b/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 @@ -411,6 +412,18 @@ export function Acks() { + {/* ================================================================= + 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. + ================================================================= */} + + {/* ================================================================= FOOTER — single hairline-separated status row. Replaces the warm-paper "End of register" treatment with a quiet @@ -552,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, + ); + + return ( +
+ + +
+
+
+ + Envelope acks +
+

+ TA1 envelopes, newest first. +

+
+

+ 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. +

+
+ +
+ + + + +
+ +
+ {isError ? ( + refetch()} + /> + ) : isLoading ? ( +
+ {Array.from({ length: 3 }).map((_, i) => ( + + ))} +
+ ) : items.length === 0 ? ( +
+ +
+ ) : ( +
+ + + + + Control # + Ack + Note + Interchange date + Sender → Receiver + + + + {items.map((t) => ( + + + + + + {t.controlNumber || "—"} + + + + + + {t.noteCode ?? "—"} + + + {t.interchangeDate ? fmt.dateShort(t.interchangeDate) : "—"} + {t.interchangeTime ? ( + · {t.interchangeTime} + ) : null} + + + {t.senderId ?? "?"} → {t.receiverId ?? "?"} + + + ))} + +
+
+ )} +
+ + +
+ ); +} + +// --------------------------------------------------------------------------- +// 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 ( + + {code} + + ); +} diff --git a/src/types/index.ts b/src/types/index.ts index 6dab372..15b4e8c 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -539,6 +539,38 @@ export interface Ack { 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; +} + // --------------------------------------------------------------------------- // SP4 claim detail drawer types (Task 4). // Mirrors the JSON shape of `GET /api/claims/{claim_id}` 1:1 — camelCase From 315fbfec42b32ab50f7fdbc47902f682713621e5 Mon Sep 17 00:00:00 2001 From: Nora Date: Mon, 29 Jun 2026 09:51:09 -0600 Subject: [PATCH 6/6] fix(docker): wire CYCLONE_SFTP_PASSWORD via docker secret in compose override Mirrors the existing admin_username / admin_pw pattern so the dev compose loads SFTP creds from /tmp/cyclone-test-secrets/sftp_password when present. Falls back to the CYCLONE_SFTP_PASSWORD env var if the file is missing. Pre-existing env-var support from SP25+26 unchanged. --- docker-compose.override.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docker-compose.override.yml b/docker-compose.override.yml index 987760b..e279b99 100644 --- a/docker-compose.override.yml +++ b/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: