diff --git a/docs/build-plans/03-game-center-token-layer.md b/docs/build-plans/03-game-center-token-layer.md new file mode 100644 index 0000000..226d8c6 --- /dev/null +++ b/docs/build-plans/03-game-center-token-layer.md @@ -0,0 +1,1351 @@ +# Game Center Token Layer — Implementation Plan (Signal v2 infrastructure) + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build STTIL's patient identity tokenization layer ("Game Center") as an additive, default-off Signal v2 seam: STTIL mints cryptographically random tokens, the clinic holds the only token-to-identity crosswalk, and Signal processes tokens exactly as it processes today's opaque `patient_id`. + +**Architecture:** A new isolated package `python-backend/core/tokens/` (format, mint, registry) plus a blind mint endpoint whose request schema physically cannot carry identity, plus a mode-gated identity path in the CSV normalizer. The scoring pipeline (dedup, coverage, readiness, rollup, persistence) is untouched — it already treats `patient_id` as an opaque string and already stores only `sha256(patient_id)`. Default behavior stays byte-identical; token mode is opt-in per deployment via env var. + +**Tech Stack:** Python 3.11 / FastAPI / Pydantic on Railway; Supabase Postgres (service-role access, RLS enabled); pytest; Clerk JWT auth (org derived from claims). + +**Plan authored:** 2026-07-07 by the Fable orchestrator session, from a cto-architect design fan-out + a primary-source compliance verification pass + direct repo grounding. Concept origin: `STTIL-Vault/Projects/sttil-privacy-token-platform-2026-06-03.md`. + +--- + +## 0. Read this first (context for a fresh session) + +**What Game Center is.** Every patient in any STTIL system gets a cryptographically random token. STTIL generates and stores tokens; the crosswalk between token and real identity (name, DOB, MRN) lives ONLY with the clinic/supplier. STTIL's tools process tokens and never hold the identity map. The design goal is architectural de-identification, not pseudonymization-by-contract. It upgrades Signal's privacy story and is a prerequisite for Vitality Med Tracker (see `signal/CLAUDE.md` "Gate: Token Layer"). + +**The one binding constraint (never trade this away):** No identity attribute may EVER cross to STTIL, including inside the token-generation request. The mint request accepts exactly one field (`count: int`) with `extra="forbid"` — identity cannot cross because there is nowhere to put it. Any future change that adds a string field to the mint path, or a per-patient "tokenize this MRN" endpoint, breaks the entire compliance thesis. Reject such changes. + +### Execution preconditions — check ALL before Task 1 + +- [ ] **The in-flight readiness lane has landed.** As of 2026-07-07 the signal repo has uncommitted WIP (`python-backend/api/main.py`, `core/worklist_readiness.py`, untracked `core/rollup.py`, `tests/test_rollup.py`) belonging to the readiness-model P2–P5 lane (`docs/build-plans/02-readiness-model-completion.md`). Run `git status --short` — if `python-backend/` is not clean, STOP and confirm with Kisa before proceeding (decision K5 below). Both lanes edit `normalizer.py`/`main.py`; sequence, don't merge blind. +- [ ] `cd /Users/sttil-solutions/projects/signal && git pull --rebase` succeeds and `context/current-state.md` has been read. +- [ ] Full test suite green before you start: `python -m pytest tests/ -v` (132+ tests as of 2026-07-07). Record the pass count; you must not lower it. +- [ ] Kisa's pending decisions (K1–K6 below) checked in `context/current-state.md` Decisions section. This plan implements the recommended defaults; if Kisa reversed any, adapt before building. + +### Kisa decision register (defaults this plan implements) + +| # | Decision | Default implemented here | Status | +|---|---|---|---| +| K1 | Stable per-clinic opaque tokens (overturns the current-state.md "rotating RAP-ID/CSP-ID" note; rotation re-homed to optional v2 display aliases) | Stable | PENDING Kisa | +| K2 | Opaque `TKN_` format, not memorable "BlueDragon_4821" (wordlists collapse entropy and invite mnemonic leakage; friendly names become clinic-local, STTIL never sees them) | Opaque | PENDING Kisa | +| K3 | v1 clinic artifact = crosswalk spreadsheet template + runbook; offline mapping helper is a fast-follow | Template + runbook | PENDING Kisa | +| K4 | Token mode is opt-in per deployment; the live Gaboro pilot stays on `patient_id` | Opt-in, default off | PENDING Kisa | +| K5 | Build starts only after the readiness lane lands | Sequenced after | PENDING Kisa | +| K6 | STTIL is sole minter; registry enforcement rejects non-STTIL tokens | Sole minter | PENDING Kisa | + +Design rationale (alternatives, GROUNDED/GRAFTED labels, threat model) lives in section 7. Executing model: do NOT `kg_add` any of these — the Decision Protocol runs only when Kisa confirms. + +--- + +## 1. Compliance guardrails (verified against primary sources 2026-07-07) + +These bind every task and every future edit to this layer. + +1. **Random, never derived.** Tokens come from `secrets` (CSPRNG) with zero relationship to any patient attribute. HHS guidance: a code "cannot be derived from" identity; even a keyed hash of an MRN fails. A random token passes. (45 CFR 164.514(c); HHS de-identification guidance.) +2. **The crosswalk never exists at STTIL.** No schema in STTIL's DB may hold, and no endpoint may accept, any identity attribute alongside a token. The `tokens` table has no identity columns and none may ever be added. +3. **Dates finding (attorney-bound, do not resolve in code):** full shipment dates in data flowing to STTIL break HIPAA safe-harbor de-identification regardless of token quality ("all elements of dates except year" is one of the 18 identifiers). Signal needs granular dates to function, so the viable path is expert determination (164.514(b)(1)) — a formal, documented expert engagement, not a design self-judgment. This plan does NOT truncate dates; it flags the question to the attorney brief. Nothing in this plan claims the resulting data is "de-identified." +4. **BA status is the attorney's call.** Datavant — the $1B+ tokenization precedent — executes BAAs and accepts Business Associate status rather than claiming exemption. STTIL claims architecture facts only. +5. **NEVER-CLAIM list (external copy, decks, LOIs):** "not a HIPAA Business Associate", "HIPAA compliant", "data is de-identified", "STTIL cannot access patient identity" (as a guarantee), "architecture eliminates PHI risk", "safe-harbor compliant", "no BAA needed", "we never hold PHI". SAFE-TO-SAY: design facts — "STTIL's system is designed so STTIL does not receive names, DOBs, SSNs, or contact info"; "the crosswalk is held only by the clinic in our design"; "tokens are cryptographically random with no relationship to any patient attribute." +6. **Attorney brief questions** (assemble before first production use; sources and full text in section 8): 164.514(c) assignment when the vendor generates the code but the clinic builds the crosswalk; the de-identification precondition for (c); dates via expert determination; whether the current supplier patient_id contract clause is sufficient (contract ≠ cryptographic randomness — a live Signal exposure, not just a Game Center question); FTC HBNR applicability if STTIL is not HIPAA-covered; crosswalk-destruction contract language for offboarding; AWS BAA activation scope; approved pitch-deck phrasing. + +--- + +## 2. File map + +``` +python-backend/core/tokens/__init__.py CREATE (empty package marker) +python-backend/core/tokens/format.py CREATE Task 1 +python-backend/core/tokens/mint.py CREATE Task 2 +python-backend/core/tokens/registry.py CREATE Task 3 +supabase/migrations/20260708000001_token_registry.sql CREATE Task 3 +python-backend/core/audit_logger.py MODIFY (2 enum lines) Task 4 +python-backend/api/main.py MODIFY (mint endpoint; upload guard; 2 audit spots) Tasks 4, 6, 8 +python-backend/api/normalizer.py MODIFY (token mode) Task 5 +docs/onboarding/crosswalk-template.csv CREATE Task 9 +docs/onboarding/token-crosswalk-runbook.md CREATE Task 9 +tests/test_token_format.py CREATE Task 1 +tests/test_token_mint.py CREATE Task 2 +tests/test_token_registry.py CREATE Task 3 +tests/test_token_mint_endpoint.py CREATE Task 4 +tests/test_normalizer_token_mode.py CREATE Task 5 +tests/test_upload_token_guard.py CREATE Task 6 +tests/test_token_pipeline_e2e.py CREATE Task 10 +``` + +UNCHANGED by design (verify at the end that your diff never touched them): `core/dedup.py`, `core/coverage_calculator.py`, `core/worklist_readiness.py`, `core/readiness.py`, `core/rollup.py`, `core/persistence.py`, `core/doc_state_machine.py`, `core/rules.py`. They all treat `patient_id` as an opaque string; a token IS an opaque string. + +All test files start with the repo's standard path shim (copy exactly): + +```python +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) +``` + +Run tests from the repo root: `cd /Users/sttil-solutions/projects/signal && python -m pytest tests/ -v`. + +--- + +## 3. Database schema (Task 3 migration, shown here for reference) + +```sql +-- Game Center token registry. STTIL stores ONLY opaque random tokens. +-- COMPLIANCE LOCK: no identity columns exist here and none may ever be added. +create table if not exists tokens ( + id uuid primary key default gen_random_uuid(), + org_id uuid not null references organizations(id) on delete cascade, + token text not null, + status text not null default 'active' check (status in ('active', 'revoked')), + minted_at timestamptz not null default now(), + revoked_at timestamptz, + unique (org_id, token) +); +create index if not exists tokens_org_status_idx on tokens (org_id, status); + +-- Mint-event audit: counts only, never token values. +create table if not exists token_mint_batches ( + id uuid primary key default gen_random_uuid(), + org_id uuid not null references organizations(id) on delete cascade, + count int not null, + minted_by text, + created_at timestamptz not null default now() +); + +alter table tokens enable row level security; +alter table token_mint_batches enable row level security; +-- No anon/public policies: the backend uses the service role, mirroring +-- existing Signal tables (see 20260607000001_audit_log_worm.sql for style). +``` + +Operational tables are untouched: `persistence.py` already stores `sha256(patient_id)`; in token mode the same code path hashes the token. + +--- + +## Task 1: Token format (`core/tokens/format.py`) + +**Files:** Create `python-backend/core/tokens/__init__.py` (empty), `python-backend/core/tokens/format.py`, `tests/test_token_format.py`. + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_token_format.py +""" +Game Center token format tests. + +Canonical token: "TKN_" + 26 Crockford-Base32 chars (130 bits, CSPRNG) ++ 1 checksum char. Tokens are NEVER derived from patient attributes +(45 CFR 164.514(c): the code must not be "derived from or related to +information about the individual"). +""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +from core.tokens.format import ( + ALPHABET, + BODY_LENGTH, + TOKEN_LENGTH, + TOKEN_PREFIX, + _checksum_char, + generate, + is_valid_token, +) + + +def test_alphabet_is_crockford_base32(): + assert len(ALPHABET) == 32 + for ambiguous in "ILOU": + assert ambiguous not in ALPHABET + + +def test_generate_shape(): + token = generate() + assert len(token) == TOKEN_LENGTH == len(TOKEN_PREFIX) + BODY_LENGTH + 1 + assert token.startswith(TOKEN_PREFIX) + assert all(c in ALPHABET for c in token[len(TOKEN_PREFIX):]) + + +def test_generate_is_unique_across_1000(): + assert len({generate() for _ in range(1000)}) == 1000 + + +def test_valid_roundtrip(): + for _ in range(100): + assert is_valid_token(generate()) + + +def test_checksum_catches_every_single_char_substitution(): + token = generate() + body_start = len(TOKEN_PREFIX) + for i in range(body_start, TOKEN_LENGTH - 1): + original = token[i] + substitute = ALPHABET[(ALPHABET.index(original) + 1) % 32] + corrupted = token[:i] + substitute + token[i + 1:] + assert not is_valid_token(corrupted), f"substitution at {i} not caught" + + +def test_rejects_mrn_like_values(): + for bad in ("MRN00123", "12345", "PT-2024-001", "", "TKN_", "TKN_SHORT", + "tkn_" + "A" * 27, "TKN_" + "a" * 27): + assert not is_valid_token(bad) + + +def test_rejects_non_string(): + assert not is_valid_token(None) + assert not is_valid_token(12345) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_token_format.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'core.tokens'` + +- [ ] **Step 3: Write the implementation** + +```python +# python-backend/core/tokens/format.py +""" +Game Center token format. + +Canonical token: "TKN_" + 26 Crockford-Base32 chars (130 bits from the OS +CSPRNG) + 1 checksum char. Total length 31. + +Tokens are opaque crosswalk keys, NEVER derived from any patient attribute +(compliance-critical: 45 CFR 164.514(c) requires a re-identification code +be "not derived from or related to information about the individual"). +The "TKN_" prefix makes accidental leaks greppable; a token without the +clinic-held crosswalk is just a random string. +""" + +import secrets + +TOKEN_PREFIX = "TKN_" +# Crockford Base32 — excludes I, L, O, U (ambiguous glyphs, accidental words). +ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" +BODY_LENGTH = 26 # 26 chars x 5 bits = 130 bits of entropy +TOKEN_LENGTH = len(TOKEN_PREFIX) + BODY_LENGTH + 1 # prefix + body + checksum + + +def _checksum_char(body: str) -> str: + """Unweighted mod-32 sum: provably catches ANY single-character + substitution (the delta of two distinct alphabet indexes is never 0 + mod 32). Transpositions are not caught — acceptable, tokens are + copy-pasted or VLOOKUP'd, rarely hand-keyed.""" + total = sum(ALPHABET.index(c) for c in body) + return ALPHABET[total % 32] + + +def generate() -> str: + """One canonical token from the OS CSPRNG. Never uses `random`.""" + body = "".join(secrets.choice(ALPHABET) for _ in range(BODY_LENGTH)) + return f"{TOKEN_PREFIX}{body}{_checksum_char(body)}" + + +def is_valid_token(value) -> bool: + """Format + checksum validation only. Registry membership (was this + token actually minted for this org?) is a separate check.""" + if not isinstance(value, str) or len(value) != TOKEN_LENGTH: + return False + if not value.startswith(TOKEN_PREFIX): + return False + body, check = value[len(TOKEN_PREFIX):-1], value[-1] + if any(c not in ALPHABET for c in body) or check not in ALPHABET: + return False + return _checksum_char(body) == check +``` + +Also create the empty package marker: `python-backend/core/tokens/__init__.py` (zero bytes is fine). + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/test_token_format.py -v` +Expected: 7 passed + +- [ ] **Step 5: Commit** + +```bash +git add python-backend/core/tokens/__init__.py python-backend/core/tokens/format.py tests/test_token_format.py +git commit -m "feat(tokens): Game Center token format - 130-bit CSPRNG, checksum, validator" +``` + +--- + +## Task 2: Blind mint (`core/tokens/mint.py`) + +**Files:** Create `python-backend/core/tokens/mint.py`, `tests/test_token_mint.py`. + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_token_mint.py +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +import pytest + +from core.tokens.format import is_valid_token +from core.tokens.mint import MINT_CAP, mint + + +def test_mint_returns_requested_count_of_valid_tokens(): + tokens = mint(50) + assert len(tokens) == 50 + assert all(is_valid_token(t) for t in tokens) + + +def test_mint_batch_is_unique(): + assert len(set(mint(500))) == 500 + + +def test_mint_rejects_nonpositive_and_noninteger(): + for bad in (0, -1, 2.5, "10", None): + with pytest.raises(ValueError): + mint(bad) + + +def test_mint_rejects_over_cap(): + with pytest.raises(ValueError): + mint(MINT_CAP + 1) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_token_mint.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'core.tokens.mint'` + +- [ ] **Step 3: Write the implementation** + +```python +# python-backend/core/tokens/mint.py +""" +Blind bulk mint. The ONLY input is a count — identity has nowhere to ride. +This is the structural guarantee of the Game Center layer: STTIL mints +tokens without ever learning who they will map to. Never add parameters +that could carry patient attributes. +""" + +from .format import generate + +MINT_CAP = 50_000 # per-request cap; resource-exhaustion guard + + +def mint(count) -> list[str]: + if not isinstance(count, int) or isinstance(count, bool) or count < 1: + raise ValueError("count must be a positive integer") + if count > MINT_CAP: + raise ValueError(f"count exceeds mint cap of {MINT_CAP}") + tokens: set[str] = set() + while len(tokens) < count: # collision retry; astronomically rare at 130 bits + tokens.add(generate()) + return sorted(tokens) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/test_token_mint.py -v` +Expected: 4 passed + +- [ ] **Step 5: Commit** + +```bash +git add python-backend/core/tokens/mint.py tests/test_token_mint.py +git commit -m "feat(tokens): blind bulk mint - count-only input, 50k cap" +``` + +--- + +## Task 3: Registry (`core/tokens/registry.py` + migration) + +**Files:** Create `python-backend/core/tokens/registry.py`, `supabase/migrations/20260708000001_token_registry.sql` (content in section 3 above), `tests/test_token_registry.py`. + +- [ ] **Step 1: Write the migration file** — copy the SQL from section 3 verbatim into `supabase/migrations/20260708000001_token_registry.sql`. + +- [ ] **Step 2: Write the failing tests** (unit-level with a fake client; no live Supabase needed) + +```python +# tests/test_token_registry.py +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +from core.tokens import registry + + +class FakeResponse: + def __init__(self, data): + self.data = data + + +class FakeTable: + """Minimal supabase-py table stub recording calls and returning canned data.""" + def __init__(self, store, name): + self.store, self.name = store, name + self._filters = {} + + def insert(self, rows): + self.store.setdefault(self.name, []).extend(rows if isinstance(rows, list) else [rows]) + self._result = FakeResponse(rows if isinstance(rows, list) else [rows]) + return self + + def select(self, *_): + return self + + def eq(self, col, val): + self._filters[col] = val + return self + + def in_(self, col, vals): + rows = [r for r in self.store.get(self.name, []) + if r.get(col) in set(vals) + and all(r.get(k) == v for k, v in self._filters.items() if k != "status")] + self._result = FakeResponse(rows) + return self + + def delete(self): + remaining, removed = [], [] + for r in self.store.get(self.name, []): + (removed if all(r.get(k) == v for k, v in self._filters.items()) else remaining).append(r) + self.store[self.name] = remaining + self._result = FakeResponse(removed) + return self + + def execute(self): + return getattr(self, "_result", FakeResponse([])) + + +class FakeClient: + def __init__(self): + self.store = {} + + def table(self, name): + return FakeTable(self.store, name) + + +def test_record_mint_persists_tokens_and_batch(monkeypatch): + fake = FakeClient() + monkeypatch.setattr(registry, "get_client", lambda: fake) + registry.record_mint("org-1", ["TKN_A", "TKN_B"], minted_by="user_1") + assert len(fake.store["tokens"]) == 2 + assert all(r["org_id"] == "org-1" for r in fake.store["tokens"]) + assert fake.store["token_mint_batches"][0]["count"] == 2 + # COMPLIANCE LOCK: no identity-shaped keys ever persisted + for row in fake.store["tokens"]: + assert set(row.keys()) == {"org_id", "token"} + + +def test_minted_subset_returns_only_known_org_tokens(monkeypatch): + fake = FakeClient() + monkeypatch.setattr(registry, "get_client", lambda: fake) + registry.record_mint("org-1", ["TKN_A", "TKN_B"]) + known = registry.minted_subset("org-1", ["TKN_A", "TKN_X"]) + assert known == {"TKN_A"} + + +def test_revoke_org_purges_and_reports_counts(monkeypatch): + fake = FakeClient() + monkeypatch.setattr(registry, "get_client", lambda: fake) + registry.record_mint("org-1", ["TKN_A", "TKN_B"]) + registry.record_mint("org-2", ["TKN_C"]) + counts = registry.revoke_org("org-1") + assert counts["tokens"] == 2 + assert len(fake.store["tokens"]) == 1 # org-2 untouched +``` + +- [ ] **Step 3: Run tests to verify they fail** + +Run: `python -m pytest tests/test_token_registry.py -v` +Expected: FAIL with `ModuleNotFoundError` or `AttributeError` on `registry` + +- [ ] **Step 4: Write the implementation** + +```python +# python-backend/core/tokens/registry.py +""" +Token registry — Supabase persistence for minted tokens. + +STTIL stores ONLY opaque token values, org-scoped. COMPLIANCE LOCK: no +identity columns exist in this schema and none may ever be added. The +insert row shape is exactly {org_id, token}; widening it is a design +violation, not a feature. +""" + +from core.supabase_client import get_client + +# Org-scoped operational tables purged on offboarding, verified against +# persistence.py/audit_logger.py table usage 2026-07-07. audit_log is +# intentionally EXCLUDED (WORM — the tombstone lives there). +ORG_SCOPED_PURGE_TABLES = [ + "tokens", + "token_mint_batches", + "normalized_records", + "mapping_decisions", + "report_runs", + "upload_batches", + "source_files", + "export_files", + "confirmed_visits", + "doc_status", +] + + +def record_mint(org_id: str, tokens: list[str], minted_by: str | None = None) -> None: + client = get_client() + rows = [{"org_id": org_id, "token": t} for t in tokens] + client.table("tokens").insert(rows).execute() + client.table("token_mint_batches").insert( + {"org_id": org_id, "count": len(tokens), "minted_by": minted_by} + ).execute() + + +def minted_subset(org_id: str, tokens: list[str]) -> set[str]: + """Return the subset of `tokens` that are active minted tokens for this + org. Chunked to keep IN-lists bounded.""" + client = get_client() + found: set[str] = set() + for i in range(0, len(tokens), 200): + chunk = tokens[i:i + 200] + resp = ( + client.table("tokens") + .select("token") + .eq("org_id", org_id) + .eq("status", "active") + .in_("token", chunk) + .execute() + ) + found.update(row["token"] for row in (resp.data or [])) + return found + + +def revoke_org(org_id: str) -> dict[str, int]: + """Offboarding purge: hard-delete all org-scoped rows. The clinic-side + crosswalk destruction (contractual) is the actual de-identification + event; this purge is defense in depth. Returns per-table row counts + for the audit tombstone (counts only — never token values).""" + client = get_client() + counts: dict[str, int] = {} + for table in ORG_SCOPED_PURGE_TABLES: + try: + resp = client.table(table).delete().eq("org_id", org_id).execute() + counts[table] = len(resp.data or []) + except Exception: + counts[table] = -1 # -1 = purge failed for this table; caller surfaces it + return counts +``` + +Note: some operational tables key rows by `batch_id`/`report_run_id` rather than `org_id` directly. After writing this, run `grep -n "org_id" python-backend/core/persistence.py` and check each table in `ORG_SCOPED_PURGE_TABLES` actually has an `org_id` column in `supabase/migrations/`; remove any that don't and instead delete them via their parent's cascade (the migration FKs use `on delete cascade` where that holds). Record what you found in the Task 3 commit message. + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `python -m pytest tests/test_token_registry.py -v` +Expected: 3 passed + +- [ ] **Step 6: Apply the migration** (CLI only, mirroring how `20260607000001_audit_log_worm.sql` was applied): `supabase db push` from the repo root, or `psql $SUPABASE_DB_URL -f supabase/migrations/20260708000001_token_registry.sql` if the CLI is not linked. Verify: query `select count(*) from tokens;` returns 0, not an error. + +- [ ] **Step 7: Commit** + +```bash +git add python-backend/core/tokens/registry.py supabase/migrations/20260708000001_token_registry.sql tests/test_token_registry.py +git commit -m "feat(tokens): org-scoped token registry + Supabase migration + offboarding purge" +``` + +--- + +## Task 4: Blind mint endpoint (`POST /api/token/mint`) + +**Files:** Modify `python-backend/api/main.py` (new endpoint + models), modify `python-backend/core/audit_logger.py` (2 enum lines), create `tests/test_token_mint_endpoint.py`. + +- [ ] **Step 1: Add audit actions** — in `python-backend/core/audit_logger.py`, extend the `AuditAction` enum (currently ends at `RECORD_VIEW = "record_view"`, line ~51): + +```python + TOKEN_MINT = "token_mint" + ORG_REVOKE = "org_revoke" +``` + +- [ ] **Step 2: Write the failing tests** + +```python +# tests/test_token_mint_endpoint.py +""" +The mint endpoint is BLIND: its schema accepts exactly one field (count). +The identity-cannot-cross test is the compliance-critical one — any extra +field must be rejected at the schema boundary (422), never silently ignored. +""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +import pytest +from fastapi.testclient import TestClient + +import api.main as main_mod +from api.main import app, require_auth + + +@pytest.fixture +def client(monkeypatch): + app.dependency_overrides[require_auth] = lambda: { + "sub": "test_user", "o": {"id": "org_clerk_test"}, + } + monkeypatch.setattr(main_mod, "get_or_create_org", lambda clerk_org_id=None: "org-uuid-1") + recorded = {} + + def fake_record_mint(org_id, tokens, minted_by=None): + recorded["org_id"], recorded["tokens"], recorded["minted_by"] = org_id, tokens, minted_by + + import core.tokens.registry as reg + monkeypatch.setattr(reg, "record_mint", fake_record_mint) + monkeypatch.setattr(main_mod, "log_event", lambda **kw: recorded.setdefault("audit", kw)) + yield TestClient(app), recorded + app.dependency_overrides.clear() + + +def test_mint_returns_valid_tokens(client): + c, recorded = client + resp = c.post("/api/token/mint", json={"count": 5}) + assert resp.status_code == 200 + body = resp.json() + assert body["count"] == 5 and len(body["tokens"]) == 5 + assert recorded["org_id"] == "org-uuid-1" + + from core.tokens.format import is_valid_token + assert all(is_valid_token(t) for t in body["tokens"]) + + +def test_identity_cannot_cross_extra_fields_rejected(client): + c, _ = client + for smuggle in ( + {"count": 5, "patient_name": "John Doe"}, + {"count": 5, "mrn": "MRN0042"}, + {"count": 5, "note": "for the Smith family"}, + ): + resp = c.post("/api/token/mint", json=smuggle) + assert resp.status_code == 422, f"extra field accepted: {smuggle}" + + +def test_mint_cap_enforced(client): + c, _ = client + assert c.post("/api/token/mint", json={"count": 50_001}).status_code == 422 + assert c.post("/api/token/mint", json={"count": 0}).status_code == 422 + + +def test_audit_logs_count_never_token_values(client): + c, recorded = client + resp = c.post("/api/token/mint", json={"count": 3}) + assert resp.status_code == 200 + audit_str = str(recorded.get("audit", {})) + for token in resp.json()["tokens"]: + assert token not in audit_str +``` + +If `fastapi.testclient` raises `ImportError: httpx`, run `pip install httpx` (test-only dependency; do not add it to the deploy requirements). + +- [ ] **Step 3: Run tests to verify they fail** + +Run: `python -m pytest tests/test_token_mint_endpoint.py -v` +Expected: FAIL with 404 on `/api/token/mint` (route doesn't exist) + +- [ ] **Step 4: Implement the endpoint** — in `python-backend/api/main.py`: + +(a) Extend the pydantic import at line ~18 from `from pydantic import BaseModel` to: + +```python +from pydantic import BaseModel, ConfigDict, Field +``` + +(If the installed pydantic is v1 — check with `python -c "import pydantic; print(pydantic.VERSION)"` — use `class Config: extra = "forbid"` inside the model instead of `model_config`.) + +(b) Add near the other request models (e.g., after `class ExportRequest`, line ~891): + +```python +class MintRequest(BaseModel): + """Blind mint request. COMPLIANCE LOCK: exactly one field, extra + forbidden. Identity cannot cross because there is nowhere to put it. + Never add a string field to this model.""" + model_config = ConfigDict(extra="forbid") + count: int = Field(..., ge=1, le=50_000) + + +class MintResponse(BaseModel): + tokens: list[str] + count: int +``` + +(c) Add the endpoint (place after the `/api/export` endpoint): + +```python +@app.post("/api/token/mint", response_model=MintResponse) +async def mint_tokens( + body: MintRequest, + claims: dict = Depends(require_auth), +): + """Game Center blind mint: returns freshly minted org-scoped tokens. + The clinic maps them to patients in ITS OWN crosswalk; STTIL never + sees, accepts, or stores the mapping.""" + from core.tokens.mint import mint + from core.tokens import registry as token_registry + + clerk_org_id = ( + claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None + ) + org_id = get_or_create_org(clerk_org_id=clerk_org_id) + if not org_id: + raise HTTPException( + status_code=400, detail="No organization resolved from auth claims." + ) + + tokens = mint(body.count) + token_registry.record_mint(org_id, tokens, minted_by=claims.get("sub", "demo_user")) + + from core.supabase_client import get_client + log_event( + action=AuditAction.TOKEN_MINT, + resource_id=f"token-mint:{body.count}", + user_id=claims.get("sub", "demo_user"), + outcome="success", + ip_address="0.0.0.0", + detail=f"minted {len(tokens)} tokens", # counts only, NEVER token values + db_conn=get_client(), + ) + return MintResponse(tokens=tokens, count=len(tokens)) +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `python -m pytest tests/test_token_mint_endpoint.py -v` +Expected: 4 passed + +- [ ] **Step 6: Run the FULL suite** — `python -m pytest tests/ -v` — expected: no regressions vs the precondition count. + +- [ ] **Step 7: Commit** + +```bash +git add python-backend/api/main.py python-backend/core/audit_logger.py tests/test_token_mint_endpoint.py +git commit -m "feat(tokens): blind mint endpoint - schema-locked count-only request" +``` + +--- + +## Task 5: Normalizer token mode + +**Files:** Modify `python-backend/api/normalizer.py`, create `tests/test_normalizer_token_mode.py`. + +Behavior: `IDENTITY_MODE=patient_id` (default) → byte-identical to today. `IDENTITY_MODE=token` → the identity value must be a valid `TKN_` token (arriving via a token-named column or the patient_id column); any non-token identifier is skipped with a reason that NEVER echoes the raw value (it could be an MRN — echoing it would put identity in responses and logs). + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_normalizer_token_mode.py +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +from api.normalizer import normalize_csv +from core.tokens.format import generate + +CSV_LEGACY = ( + "patient_id,device_type,shipment_date,quantity,payer\n" + "PT-001,Dexcom G7,2026-06-01,3,Medicare Part B\n" + "PT-002,Libre 3,2026-06-15,1,Aetna\n" +) + + +def test_default_mode_is_byte_identical_regression_lock(monkeypatch): + monkeypatch.delenv("IDENTITY_MODE", raising=False) + records, skipped, summary = normalize_csv(CSV_LEGACY) + assert [r.patient_id for r in records] == ["PT-001", "PT-002"] + assert skipped == [] + assert summary["required_missing"] == [] + + +def test_token_mode_accepts_token_column(monkeypatch): + monkeypatch.setenv("IDENTITY_MODE", "token") + t1, t2 = generate(), generate() + csv_text = ( + "token,device_type,shipment_date,quantity,payer\n" + f"{t1},Dexcom G7,2026-06-01,3,Medicare Part B\n" + f"{t2},Libre 3,2026-06-15,1,Aetna\n" + ) + records, skipped, _ = normalize_csv(csv_text) + assert [r.patient_id for r in records] == [t1, t2] + assert skipped == [] + + +def test_token_mode_accepts_token_in_patient_id_column(monkeypatch): + monkeypatch.setenv("IDENTITY_MODE", "token") + t1 = generate() + csv_text = ( + "patient_id,device_type,shipment_date\n" + f"{t1},Dexcom G7,2026-06-01\n" + ) + records, skipped, _ = normalize_csv(csv_text) + assert records[0].patient_id == t1 + + +def test_token_mode_rejects_raw_mrn_without_echoing_it(monkeypatch): + monkeypatch.setenv("IDENTITY_MODE", "token") + csv_text = ( + "patient_id,device_type,shipment_date\n" + "MRN-SECRET-0042,Dexcom G7,2026-06-01\n" + ) + records, skipped, _ = normalize_csv(csv_text) + assert records == [] + assert len(skipped) == 1 + assert "MRN-SECRET-0042" not in skipped[0] # NEVER echo a rejected identifier + assert "token" in skipped[0].lower() + + +def test_token_mode_flags_missing_identity_column(monkeypatch): + monkeypatch.setenv("IDENTITY_MODE", "token") + csv_text = "device_type,shipment_date\nDexcom G7,2026-06-01\n" + _, _, summary = normalize_csv(csv_text) + assert "sttil_token" in summary["required_missing"] +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_normalizer_token_mode.py -v` +Expected: the default-mode test passes; the four token-mode tests FAIL (no token handling exists) + +- [ ] **Step 3: Implement** — in `python-backend/api/normalizer.py`: + +(a) Add `import os` to the imports (line ~8 block) and, after the `from core.coverage_calculator import ShipmentRecord` import: + +```python +from core.tokens.format import is_valid_token + + +def _identity_mode() -> str: + """"patient_id" (default, legacy) or "token" (Game Center). Read per + call so tests and deployments flip it via env without reimport.""" + mode = os.getenv("IDENTITY_MODE", "patient_id").strip().lower() + return mode if mode in ("patient_id", "token") else "patient_id" +``` + +(b) Add a token entry to `HEADER_MAP` (after the `"patient_id"` entry): + +```python + "sttil_token": [ + "sttil_token", "sttil token", "token", "signal_token", "signal token", + "patient_token", "patient token", "game_center_token", "gc_token", + ], +``` + +(c) In `normalize_csv`, replace the `required_missing` line (currently `required_missing = [f for f in required_fields if f not in mapping_detail]`) with: + +```python + required_missing = [f for f in required_fields if f not in mapping_detail] + identity_mode = _identity_mode() + if identity_mode == "token": + required_missing = [ + f for f in ("device_type", "shipment_date") if f not in mapping_detail + ] + if "sttil_token" not in mapping_detail and "patient_id" not in mapping_detail: + required_missing.append("sttil_token") +``` + +(d) In the row loop, replace the identity block (currently the three lines starting `patient_id = mapped.get("patient_id", "").strip()` through `continue`) with: + +```python + if identity_mode == "token": + candidate = ( + mapped.get("sttil_token", "").strip() + or mapped.get("patient_id", "").strip() + ) + if not candidate: + skipped.append(f"Row {i}: missing token") + continue + if not is_valid_token(candidate): + # NEVER echo the rejected value — it could be a raw MRN. + skipped.append( + f"Row {i}: identifier is not a valid STTIL token " + "(token mode accepts only TKN_-format tokens; raw IDs are rejected)" + ) + continue + patient_id = candidate + else: + patient_id = mapped.get("patient_id", "").strip() + if not patient_id: + skipped.append(f"Row {i}: missing patient_id") + continue +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/test_normalizer_token_mode.py -v` +Expected: 5 passed + +- [ ] **Step 5: Regression check the full suite** — `python -m pytest tests/ -v` — the pre-existing normalizer/e2e tests must all still pass (they run in default mode). + +- [ ] **Step 6: Commit** + +```bash +git add python-backend/api/normalizer.py tests/test_normalizer_token_mode.py +git commit -m "feat(tokens): IDENTITY_MODE token path in normalizer - default byte-identical" +``` + +--- + +## Task 6: Upload registry-enforcement guard + +**Files:** Modify `python-backend/api/main.py` (`upload_csv`), create `tests/test_upload_token_guard.py`. + +Behavior: with `IDENTITY_MODE=token` AND `TOKEN_ENFORCE_REGISTRY=true`, an upload containing tokens not minted for the caller's org is rejected 422 (sole-minter rule, K6). Registry unreachable in enforce mode → 503 fail-CLOSED. Both flags off → zero change. + +- [ ] **Step 1: Write the failing tests** + +```python +# tests/test_upload_token_guard.py +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +import io +import pytest +from fastapi.testclient import TestClient + +import api.main as main_mod +from api.main import app, require_auth +from core.tokens.format import generate + + +@pytest.fixture +def token_mode_client(monkeypatch): + monkeypatch.setenv("IDENTITY_MODE", "token") + monkeypatch.setenv("TOKEN_ENFORCE_REGISTRY", "true") + app.dependency_overrides[require_auth] = lambda: { + "sub": "test_user", "o": {"id": "org_clerk_test"}, + } + monkeypatch.setattr(main_mod, "get_or_create_org", lambda clerk_org_id=None: "org-uuid-1") + yield TestClient(app), monkeypatch + app.dependency_overrides.clear() + + +def _csv_upload(client, tokens): + rows = "\n".join(f"{t},Dexcom G7,2026-06-01,3,Medicare Part B" for t in tokens) + text = f"token,device_type,shipment_date,quantity,payer\n{rows}\n" + return client.post( + "/api/upload", + files={"file": ("orders.csv", io.BytesIO(text.encode()), "text/csv")}, + ) + + +def test_unminted_tokens_rejected(token_mode_client): + client, monkeypatch = token_mode_client + import core.tokens.registry as reg + minted = generate() + stranger = generate() + monkeypatch.setattr(reg, "minted_subset", lambda org_id, toks: {minted}) + resp = _csv_upload(client, [minted, stranger]) + assert resp.status_code == 422 + assert "not minted" in str(resp.json()) + + +def test_registry_failure_fails_closed(token_mode_client): + client, monkeypatch = token_mode_client + import core.tokens.registry as reg + + def boom(org_id, toks): + raise RuntimeError("supabase down") + + monkeypatch.setattr(reg, "minted_subset", boom) + resp = _csv_upload(client, [generate()]) + assert resp.status_code == 503 +``` + +Note: the all-minted happy path continues past the guard into Supabase-backed persistence, so it is covered by the pipeline test in Task 10 rather than here. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `python -m pytest tests/test_upload_token_guard.py -v` +Expected: FAIL — uploads proceed (200) or fail later, instead of 422/503 at the guard + +- [ ] **Step 3: Implement** — in `python-backend/api/main.py`, inside `upload_csv`, immediately AFTER the `org_id = get_or_create_org(clerk_org_id=clerk_org_id)` line and BEFORE `confirmed_visits = ...`: + +```python + # Game Center guard (K6, sole minter): in token mode with enforcement on, + # every uploaded token must have been minted for THIS org. + from api.normalizer import _identity_mode + if ( + _identity_mode() == "token" + and os.getenv("TOKEN_ENFORCE_REGISTRY", "false").strip().lower() == "true" + ): + if not org_id: + raise HTTPException( + status_code=400, + detail="Token enforcement requires an organization context.", + ) + from core.tokens import registry as token_registry + upload_tokens = list({r.patient_id for r in records}) + try: + known = token_registry.minted_subset(org_id, upload_tokens) + except Exception as e: + # Fail CLOSED: in enforce mode an unreachable registry rejects + # the upload rather than letting unverified tokens through. + raise HTTPException( + status_code=503, + detail="Token registry unavailable; upload rejected in enforce mode.", + ) from e + unknown = sorted(set(upload_tokens) - known) + if unknown: + raise HTTPException( + status_code=422, + detail={ + "message": ( + f"{len(unknown)} token(s) in this file were not minted " + "for this organization." + ), + "unknown_tokens_sample": unknown[:5], + }, + ) +``` + +(`api.normalizer` is imported lazily inside the function to match the file's existing local-import style; `os` is already imported at module top.) + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `python -m pytest tests/test_upload_token_guard.py -v` +Expected: 2 passed + +- [ ] **Step 5: Commit** + +```bash +git add python-backend/api/main.py tests/test_upload_token_guard.py +git commit -m "feat(tokens): registry-enforcement upload guard - fail-closed, sole-minter" +``` + +--- + +## Task 7: Audit-log identifier hardening + +**Files:** Modify `python-backend/api/main.py` (2 spots). + +`signal/CLAUDE.md` rule: "All logs use hashed patient_id, never raw." Two audit calls currently embed a raw 8-char identifier prefix. Fix them unconditionally (strictly more compliant in both modes). + +- [ ] **Step 1: Find both spots** + +Run: `grep -n "patient_id\[:8\]" python-backend/api/main.py` +Expected: exactly 2 hits (`confirm-visit:` ~line 1101, `update-doc-status:` ~line 1193). If the readiness lane moved them, adjust line numbers, not the approach. + +- [ ] **Step 2: Replace both.** Change + +```python + resource_id=f"confirm-visit:{body.patient_id[:8]}", +``` + +to + +```python + resource_id=f"confirm-visit:{_sha256(body.patient_id)[:12]}", +``` + +and the same pattern for `update-doc-status:`. Add the import next to the other `core.persistence` imports at the top of `main.py`: + +```python +from core.persistence import _sha256 +``` + +(If `main.py` imports persistence functions lazily, follow that style instead: `from core.persistence import _sha256` inside each handler.) + +- [ ] **Step 3: Verify no raw-identifier patterns remain** + +Run: `grep -n "patient_id\[" python-backend/api/main.py` +Expected: no matches embedding raw slices into log/audit strings. + +- [ ] **Step 4: Full suite** — `python -m pytest tests/ -v` — expected: no regressions (if an existing test asserts the old `resource_id` format, update that assertion to the hashed form — the raw form violated the repo's own logging rule). + +- [ ] **Step 5: Commit** + +```bash +git add python-backend/api/main.py +git commit -m "fix(audit): hash identifier prefixes in audit resource_ids per PHI logging rule" +``` + +--- + +## Task 8: Offboarding endpoint (`POST /api/org/revoke`) + +**Files:** Modify `python-backend/api/main.py`. Registry logic already exists (Task 3); this exposes it with an explicit confirmation phrase and an audit tombstone. + +- [ ] **Step 1: Add the endpoint** (after the mint endpoint): + +```python +class RevokeOrgRequest(BaseModel): + """Offboarding is destructive. The caller must type the org's Clerk ID + as confirmation; no other body fields exist.""" + model_config = ConfigDict(extra="forbid") + confirm_clerk_org_id: str + + +@app.post("/api/org/revoke") +async def revoke_org_endpoint( + body: RevokeOrgRequest, + claims: dict = Depends(require_auth), +): + """Game Center offboarding: purge all org-scoped rows (tokens + operational + data) and write an audit tombstone with row counts only. The clinic-side + crosswalk destruction (contractual) is the actual de-identification event; + this purge is defense in depth.""" + clerk_org_id = ( + claims.get("o", {}).get("id") if isinstance(claims.get("o"), dict) else None + ) + if not clerk_org_id or body.confirm_clerk_org_id != clerk_org_id: + raise HTTPException( + status_code=400, + detail="Confirmation mismatch: confirm_clerk_org_id must equal the caller's org.", + ) + org_id = get_or_create_org(clerk_org_id=clerk_org_id) + if not org_id: + raise HTTPException(status_code=400, detail="No organization resolved.") + + from core.tokens import registry as token_registry + counts = token_registry.revoke_org(org_id) + + from core.supabase_client import get_client + log_event( + action=AuditAction.ORG_REVOKE, + resource_id=f"org-revoke:{org_id}", + user_id=claims.get("sub", "demo_user"), + outcome="failure" if any(v == -1 for v in counts.values()) else "success", + ip_address="0.0.0.0", + detail=f"purged row counts: {counts}", # counts only, never token values + db_conn=get_client(), + ) + if any(v == -1 for v in counts.values()): + raise HTTPException( + status_code=500, + detail={"message": "Purge incomplete; see counts.", "counts": counts}, + ) + return {"revoked_org": org_id, "purged": counts} +``` + +- [ ] **Step 2: Add tests** — append to `tests/test_token_mint_endpoint.py` (same fixture pattern; monkeypatch `reg.revoke_org` to return `{"tokens": 2}` for success and `{"tokens": -1}` for the failure case): + +```python +def test_revoke_requires_matching_confirmation(client): + c, _ = client + resp = c.post("/api/org/revoke", json={"confirm_clerk_org_id": "wrong_org"}) + assert resp.status_code == 400 + + +def test_revoke_purges_and_reports_counts(client, monkeypatch): + c, _ = client + import core.tokens.registry as reg + monkeypatch.setattr(reg, "revoke_org", lambda org_id: {"tokens": 2, "doc_status": 1}) + resp = c.post("/api/org/revoke", json={"confirm_clerk_org_id": "org_clerk_test"}) + assert resp.status_code == 200 + assert resp.json()["purged"]["tokens"] == 2 + + +def test_revoke_surfaces_partial_failure(client, monkeypatch): + c, _ = client + import core.tokens.registry as reg + monkeypatch.setattr(reg, "revoke_org", lambda org_id: {"tokens": -1}) + resp = c.post("/api/org/revoke", json={"confirm_clerk_org_id": "org_clerk_test"}) + assert resp.status_code == 500 +``` + +- [ ] **Step 3: Run** `python -m pytest tests/test_token_mint_endpoint.py -v` — expected: all pass (7 total in the file now). + +- [ ] **Step 4: Commit** + +```bash +git add python-backend/api/main.py tests/test_token_mint_endpoint.py +git commit -m "feat(tokens): org offboarding endpoint - confirmed purge + audit tombstone" +``` + +--- + +## Task 9: Clinic onboarding artifacts + +**Files:** Create `docs/onboarding/crosswalk-template.csv`, `docs/onboarding/token-crosswalk-runbook.md`. Documentation only — review gate, no tests. + +- [ ] **Step 1: Create `docs/onboarding/crosswalk-template.csv`** + +```csv +token,your_patient_id,your_patient_name,notes +TKN_EXAMPLE0000000000000000000,,,"Replace this row. Paste STTIL-minted tokens in column 1; fill columns 2-4 from YOUR system. THIS FILE NEVER LEAVES YOUR ORGANIZATION." +``` + +- [ ] **Step 2: Create `docs/onboarding/token-crosswalk-runbook.md`** + +```markdown +# Signal Token Onboarding Runbook (Game Center) + +For clinic/supplier staff. STTIL never sees your patient identities. This +runbook is how that stays true. + +## What you hold vs what STTIL holds +- YOU hold: the crosswalk file (token -> your patient). Store it wherever + you store PHI today. It never leaves your organization. +- STTIL holds: the tokens and documentation-status data only. No names, + no DOBs, no MRNs, no contact info. + +## Onboarding (one-time, ~30 minutes) +1. Your Signal admin signs in and requests tokens: your account team runs + the mint for your expected patient count plus headroom (e.g. 500 + patients -> mint 750). You receive a one-column CSV of tokens. +2. Open the crosswalk template (crosswalk-template.csv). Paste the minted + tokens into column 1. Fill your patient identifiers in columns 2-4 from + your own system, one row per patient. Save it in your PHI-grade storage. + Delete the example row. +3. THIS FILE IS THE CROSSWALK. Never email it to STTIL, never attach it to + a support ticket, never upload it to Signal. If STTIL ever receives it, + report it to your compliance officer and to STTIL immediately. + +## Every upload +1. Export your order data as usual (Brightree or equivalent). +2. Replace the patient ID column with the token column: VLOOKUP each + patient against your crosswalk file (token for patient_id). +3. Name the column "token". Upload the CSV to Signal as usual. +4. Signal's worklist shows tokens. To act on a flagged patient, look the + token up in YOUR crosswalk. STTIL cannot do this lookup for you - by + design, it does not have the file. + +## Rejected rows +If Signal reports "identifier is not a valid STTIL token", a raw patient +ID slipped through your VLOOKUP. Fix the mapping and re-upload. Signal +rejects these rows without storing or echoing them. + +## Adding patients +Request another mint batch any time. Add the new tokens to your crosswalk. + +## Offboarding +1. STTIL purges all your organization's data (tokens + documentation + status) and provides the purge receipt (row counts). +2. You destroy your crosswalk file. Once destroyed, nothing anywhere can + link a token to a patient - that destruction is the point. +``` + +- [ ] **Step 3: Commit** + +```bash +git add docs/onboarding/crosswalk-template.csv docs/onboarding/token-crosswalk-runbook.md +git commit -m "docs(tokens): clinic crosswalk template + onboarding runbook" +``` + +--- + +## Task 10: Pipeline end-to-end test (token mode) + +**Files:** Create `tests/test_token_pipeline_e2e.py`. Hermetic (no Supabase): drives mint → CSV → normalize → dedup → coverage exactly as `upload_csv` does. + +- [ ] **Step 1: Write the test** + +```python +# tests/test_token_pipeline_e2e.py +""" +End-to-end (in-process): minted tokens flow through the normalizer and the +scoring pipeline exactly like legacy patient_ids, because every engine +treats the identifier as an opaque string. This is the structural claim of +the Game Center seam - prove it. +""" +import sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +from datetime import date + +from api.normalizer import normalize_csv +from core.coverage_calculator import calculate_batch +from core.tokens.mint import mint + + +def test_tokens_flow_through_scoring_pipeline(monkeypatch): + monkeypatch.setenv("IDENTITY_MODE", "token") + tokens = mint(3) + rows = "\n".join( + f"{t},Dexcom G7,2026-06-0{i + 1},3,Medicare Part B" + for i, t in enumerate(tokens) + ) + csv_text = f"token,device_type,shipment_date,quantity,payer\n{rows}\n" + + records, skipped, summary = normalize_csv(csv_text) + assert skipped == [] + assert summary["required_missing"] == [] + assert sorted(r.patient_id for r in records) == sorted(tokens) + + results = calculate_batch(records, as_of=date(2026, 7, 1), confirmed_visits={}) + assert len(results) == len(tokens) + result_ids = {getattr(r, "patient_id", None) or r.get("patient_id") for r in results} \ + if results and isinstance(results[0], dict) else {r.patient_id for r in results} + assert result_ids == set(tokens) + + +def test_mixed_file_keeps_tokens_drops_raw_ids(monkeypatch): + monkeypatch.setenv("IDENTITY_MODE", "token") + good = mint(1)[0] + csv_text = ( + "token,device_type,shipment_date\n" + f"{good},Dexcom G7,2026-06-01\n" + "MRN-LEAKED-99,Dexcom G7,2026-06-02\n" + ) + records, skipped, _ = normalize_csv(csv_text) + assert [r.patient_id for r in records] == [good] + assert len(skipped) == 1 and "MRN-LEAKED-99" not in skipped[0] +``` + +Before running, check `calculate_batch`'s return shape (`grep -n "def calculate_batch" -A 10 python-backend/core/coverage_calculator.py`) and keep whichever branch of the `result_ids` extraction matches; delete the other. + +- [ ] **Step 2: Run** `python -m pytest tests/test_token_pipeline_e2e.py -v` — expected: 2 passed. + +- [ ] **Step 3: Full suite + count check** — `python -m pytest tests/ -v` — expected: precondition count + ~23 new tests, all green. + +- [ ] **Step 4: Commit** + +```bash +git add tests/test_token_pipeline_e2e.py +git commit -m "test(tokens): pipeline e2e - tokens traverse scoring unchanged, raw IDs dropped" +``` + +--- + +## Task 11: Deploy dark + live verification + +Token mode ships default-OFF (K4): deploying this code changes nothing for the live Gaboro pilot until `IDENTITY_MODE=token` is set on a deployment. + +- [ ] **Step 1: Deploy** — from the repo root: `railway up --detach` (code change → fresh build; `railway redeploy` only reuses the image). Then poll `railway status --json` until the active deployment shows `status: SUCCESS` and the instance is `RUNNING`. +- [ ] **Step 2: Verify legacy behavior unchanged** — `curl -s https://signal-api-production-91c2.up.railway.app/health` returns healthy; run the existing smoke flow (`tests/smoke_test.py` if wired, or a manual legacy-CSV upload via the UI) and confirm identical results to pre-deploy. +- [ ] **Step 3: Verify the mint endpoint exists and is auth-gated** — `curl -s -o /dev/null -w "%{http_code}" -X POST https://signal-api-production-91c2.up.railway.app/api/token/mint -H "Content-Type: application/json" -d '{"count": 1}'` — expected: 401/403 (auth required), NOT 404. +- [ ] **Step 4: Do NOT set `IDENTITY_MODE=token` on the production service.** Token mode gets its own staging deployment or a dedicated env when the first token-mode clinic onboards (Kisa's call, K4). +- [ ] **Step 5: Commit anything outstanding, then update the ship ledger** — add a line to `docs/ship-status-ledger.md` describing what shipped and that token mode is dark. + +--- + +## Task 12: State sync + handoff hygiene (ALWAYS run, even if stopping early) + +- [ ] **Step 1: Update `context/current-state.md`** via the state-sync skill: note Game Center progress in ACTIVE/NEXT, append ONE author-stamped decision line (e.g. `- 2026-07-XX []: Game Center token layer Tasks 1-N built per docs/build-plans/03-game-center-token-layer.md; token mode dark; K1-K6 pending Kisa.`), commit + push. +- [ ] **Step 2: If Kisa confirmed any K-decision in-session,** run the Decision Protocol (kg_query for conflicts, then kg_add). Otherwise add nothing to the KG. +- [ ] **Step 3: Reconcile the RAP-ID note (only if K1 confirmed):** edit `context/current-state.md` Stack line "tokenization via RAP-ID/CSP-ID (rotating anonymous ID, not static UUID)" to "tokenization via Game Center stable per-clinic opaque tokens (rotation re-homed to optional v2 display aliases)". +- [ ] **Step 4: Update `signal/CLAUDE.md` "Gate: Token Layer" table** rows to reflect what is now built (token generation service; crosswalk handoff protocol; Signal accepts tokens) — only mark an item DONE if its tests are green and committed. +- [ ] **Step 5: Attorney brief** — confirm the questions in section 8 are queued for the attorney email bundle (see reactivation plan `docs/build-plans/00-signal-reactivation-plan.md`, attorney thread). + +--- + +## 7. Design rationale (for reviewers; the tasks above already encode these) + +- **Blind mint over per-patient tokenize-API:** a "tokenize this MRN" endpoint requires identity to cross the wire even if discarded — STTIL would have *touched* the map, destroying the "provably never held it" thesis. The count-only schema makes crossing structurally impossible. GROUNDED in the concept doc; schema-lock mechanism GRAFTED (design choice). +- **One token per (clinic, patient), not global:** a global token requires STTIL to know the same person exists at two clinics — which requires identity. Per-clinic minting defeats cross-clinic correlation as a free structural property. GROUNDED (strengthens concept-doc Q2 from "safer" to "forced"). +- **Stable tokens, not rotating (K1):** Signal's value is longitudinal — `confirmed_visits` and `doc_status` persist across uploads keyed by `sha256(patient_id)`; rotation orphans all of it and breaks dedup. Rotation adds no protection here because the threats it addresses (linkable leaked identifier) are already neutralized by no-map-at-STTIL + per-clinic scoping. GROUNDED against `persistence.py`. +- **Opaque format, not memorable names (K2):** wordlists collapse entropy, invite informal mnemonics that become identifying, and feel safe enough to paste into tickets. Friendly names belong in the clinic's own crosswalk file, which STTIL never sees. GROUNDED on entropy math; overturns the concept doc's "BlueDragon_4821" illustration. +- **Supabase, not a token microservice (K6-adjacent):** the token store holds nothing sensitive (opaque strings, no map), Supabase Pro is already paid, and a clean `core/tokens/` boundary keeps later extraction reversible. Standalone service is v3+ (licensed-product horizon), YAGNI now. GROUNDED in signal/CLAUDE.md stack + parked-AWS ruling. +- **Fail-closed enforcement guard:** in enforce mode, a down registry rejects uploads. Honest failure over silent acceptance of unverifiable tokens. +- **Threat model residuals:** tokens in the export CSV are intended (the clinic needs them for lookup); small-cell/aggregate risk is deferred until any benchmarking product exists (flag attorney question 5 then); mint-endpoint abuse bounded by cap + auth. The clinic crosswalk becomes the crown jewels — protection moves to contract + runbook, which is exactly where the design wants it. + +## 8. Attorney brief questions (full text) + +1. Is STTIL a HIPAA Business Associate for Signal's current and token-mode operation ("acting on behalf of" a covered entity, even receiving only tokenized data)? Datavant precedent: accepts BA status. +2. Does 164.514(c) tolerate the vendor (STTIL) generating the code where the covered entity builds and solely holds the crosswalk — or must the covered entity itself "assign" the code? +3. 164.514(c) operates on data already de-identified under 164.514(b). What process, run by whom, de-identifies the operational payload (device, dates, quantity, payer) BEFORE the token attaches, and who certifies it? +4. Full shipment dates break safe harbor ("all elements of dates except year"). Signal needs granular dates. Is expert determination (164.514(b)(1)) viable, what expert engagement is required, and must it precede first production token-mode use? +5. If any cross-org aggregate/benchmark output is ever built: small-cell suppression standard (HHS ≥11 convention)? +6. Current Signal exposure: does the "supplier confirms patient_id is not PHI-derivable" contract clause suffice when the supplier's IDs may be sequential/MRN-derived? Must STTIL verify randomness? Can the clause be retired once token mode is live, or must both coexist? +7. If STTIL is determined NOT HIPAA-covered: does the FTC Health Breach Notification Rule (2024 amendments; breach includes unauthorized disclosure) reach Signal as a PHR-related entity? +8. Offboarding contract language: crosswalk-destruction obligation as the de-identification event + protection standard for the crosswalk while active. +9. AWS BAA: confirm executed-and-active vs signed-and-parked, and required activation scope if PHI status is ambiguous. +10. Approved external phrasing for the privacy architecture (see NEVER-CLAIM / SAFE-TO-SAY lists in section 1). + +Primary sources consulted (2026-07-07): 45 CFR 164.514 (Cornell LII), HHS de-identification guidance, Datavant BAA + de-identification pages, FTC HBNR 2024 final rule + Operation AI Comply releases, HHS/FTC coordination coverage. URLs in the session's compliance research output; re-verify before the attorney meeting. diff --git a/python-backend/api/main.py b/python-backend/api/main.py index b9ba1e2..0b074a2 100644 --- a/python-backend/api/main.py +++ b/python-backend/api/main.py @@ -35,6 +35,11 @@ from core.persistence import ( load_doc_statuses_for_org, ) from api.normalizer import normalize_csv +from core.overrides import ( + Override, + OverrideKey, + overrides_from_saved, +) from core.worklist_readiness import ( build_readiness_index, merged_to_record, @@ -349,22 +354,129 @@ VALID_DOC_STATUSES = { "pa": {"not_required", "requested", "approved", "denied"}, } -DOC_STATUS_LABELS = { - "pending": "Pending", - "requested": "Requested", - "on_file": "On File", - "not_required": "Not Required", - "approved": "Approved", - "denied": "Denied", -} +# Single source of truth for stored-status labels lives with the override +# engine (the same labels feed grading and display). +from core.overrides import STORED_STATUS_LABELS as DOC_STATUS_LABELS # noqa: E402 + + +def _recompute_patient( + patient_id: str, + echo_lines: list["EchoLine"], + saved_doc_statuses: dict, + confirmed_visit: Optional[date] = None, +): + """Stateless patient recompute (P4): rebuild the patient's lines from the + echoed CSV-truth inputs, apply ALL persisted staff overrides (the write + that triggered this call is already persisted) plus the confirmed visit, + re-grade through the same engine, and roll up through the same truth + table. Returns (lines_out, rollup_status).""" + from core.rollup import rollup_patient + from core.worklist_readiness import build_readiness_index as _bri + + records = [] + for el in echo_lines: + try: + ship = ( + date_type.fromisoformat(el.shipment_date) + if el.shipment_date + else date.today() + ) + except ValueError: + ship = date.today() + csv_visit = None + if el.csv_visit_date: + try: + csv_visit = date_type.fromisoformat(el.csv_visit_date) + except ValueError: + csv_visit = None + records.append( + ShipmentRecord( + patient_id=patient_id, + device_type=el.device_type, + shipment_date=ship, + quantity=1, + payer="", + csv_swo_status=el.csv_swo_status, + csv_visit_date=csv_visit, + csv_pecos_verified=el.csv_pecos_verified, + csv_pa_status=el.csv_pa_status, + csv_diagnosis_on_file=el.csv_diagnosis_on_file, + plan_type=(el.plan_type or "").strip().lower() or None, + ) + ) + + patient_hash = hashlib.sha256(patient_id.encode()).hexdigest() + confirmed = ( + {patient_hash: confirmed_visit} if confirmed_visit else None + ) + overrides = overrides_from_saved( + {patient_hash: saved_doc_statuses.get(patient_hash, {})} + ) + index = _bri(records, confirmed_visits=confirmed, overrides=overrides) + + lines_out = [] + for line_key, line in index.lines.items(): + verdict = index.by_line.get(line_key) + lines_out.append( + LineRecomputeOut( + device_type=line.device_type, + plan_type=None if line.plan_type == "unknown" else line.plan_type, + line_status=line.line_status, + doc_items=_readiness_items_out(verdict), + ) + ) + rollup = rollup_patient(list(index.lines.values())) + return lines_out, ( + rollup.rollup_status.value if rollup.rollup_status else None + ) + + +class EchoLine(BaseModel): + """One coverage line echoed back for a stateless patient recompute. + Carries CSV-truth grading inputs only; the server applies all persisted + staff overrides and confirmed visits on top, then re-grades.""" + + device_type: str + gradeable: bool = True + plan_type: Optional[str] = None + shipment_date: Optional[str] = None # ISO; defaults to today (doc axis + # grading does not depend on it) + csv_swo_status: Optional[str] = None + csv_visit_date: Optional[str] = None + csv_pecos_verified: Optional[str] = None + csv_pa_status: Optional[str] = None + csv_diagnosis_on_file: Optional[str] = None + + +class LineRecomputeOut(BaseModel): + device_type: str + plan_type: Optional[str] = None + line_status: Optional[str] = None + doc_items: Optional[list["ReadinessItemOut"]] = None class DocStatusRequest(BaseModel): patient_id: str doc_type: str status: str + # Device scope (P4). '' or omitted writes a legacy patient-scoped row + # that applies to all devices until a device-scoped row is saved. + device_type: str = "" status_date: Optional[str] = None expiry_date: Optional[str] = None + # Optional echo of the patient's lines for the {line, rollup} recompute. + lines: Optional[list[EchoLine]] = None + + +def _saved_for_device( + doc_statuses: dict, patient_hash: str, device_type: str +) -> dict: + """Resolve the device-scoped saved statuses for one record: legacy '' + (patient-scoped) rows apply to all devices; a device-scoped row wins.""" + by_pat = doc_statuses.get(patient_hash, {}) + merged = dict(by_pat.get("", {})) + merged.update(by_pat.get(device_type, {})) + return merged def _readiness_items_out(readiness) -> Optional[list[ReadinessItemOut]]: @@ -683,8 +795,11 @@ async def upload_csv( # Readiness index (Phase 2a: additive, augments the legacy scoring — # replaces nothing). Row-to-line membership comes from the dedup output - # itself, never re-derived from a row's own plan_type. - index = build_readiness_index(records, confirmed_visits) + # itself, never re-derived from a row's own plan_type. Staff overrides + # (P4) feed the grading inputs, device-scoped. + index = build_readiness_index( + records, confirmed_visits, overrides=overrides_from_saved(doc_statuses) + ) out = [] used_lines = {} @@ -718,8 +833,10 @@ async def upload_csv( confirmed_visit_date=confirmed_visits.get( hashlib.sha256(r.patient_id.encode()).hexdigest() ), - saved_doc_statuses=doc_statuses.get( - hashlib.sha256(r.patient_id.encode()).hexdigest(), {} + saved_doc_statuses=_saved_for_device( + doc_statuses, + hashlib.sha256(r.patient_id.encode()).hexdigest(), + r.device_type, ), readiness=verdict, ) @@ -883,6 +1000,10 @@ class ConfirmVisitRequest(BaseModel): # Client-mapped plan type echoed from the upload — never guessed. None # keeps the recompute honest: the verdict reads Plan Type Needed. plan_type: str | None = None + # Optional echo of ALL the patient's lines: the confirmed visit fans out + # to every line (the visit is the person), so the P4 recompute re-grades + # each and returns the re-rolled patient alongside the legacy record. + lines: Optional[list[EchoLine]] = None @app.post("/api/confirm-visit") @@ -985,9 +1106,23 @@ async def confirm_visit( db_conn=db_conn, ) - return _to_record_out( + record_out = _to_record_out( result, record=record, confirmed_visit_date=confirmed, readiness=verdict ) + if not body.lines: + return record_out + + # P4: re-grade the patient's echoed lines with the new confirmed date + # (visit fans out to all lines) and return the re-rolled patient. All + # legacy RecordOut fields stay present. + saved = load_doc_statuses_for_org(org_id) + lines_out, rollup_status = _recompute_patient( + body.patient_id, body.lines, saved, confirmed_visit=confirmed + ) + payload = record_out.model_dump() + payload["lines"] = [l.model_dump() for l in lines_out] + payload["rollup_status"] = rollup_status + return payload @app.put("/api/doc-status") @@ -1039,7 +1174,13 @@ async def update_doc_status( patient_hash = hashlib.sha256(body.patient_id.encode()).hexdigest() success = upsert_doc_status( - org_id, patient_hash, body.doc_type, body.status, status_date, expiry_date + org_id, + patient_hash, + body.doc_type, + body.status, + status_date, + expiry_date, + device_type=body.device_type or "", ) if not success: raise HTTPException(status_code=503, detail="Failed to save doc status.") @@ -1057,9 +1198,24 @@ async def update_doc_status( db_conn=db_conn, ) - return { + response = { "patient_id": body.patient_id, "doc_type": body.doc_type, "status": body.status, + "device_type": body.device_type or "", "display": DOC_STATUS_LABELS.get(body.status, body.status), } + + # P4 recompute contract: when the client echoes the patient's lines, + # re-grade them (the write above is already persisted, so a fresh load + # picks it up) and return {lines, rollup} from the same engine. + if body.lines: + saved = load_doc_statuses_for_org(org_id) + confirmed = load_confirmed_visits_for_org(org_id).get(patient_hash) + lines_out, rollup_status = _recompute_patient( + body.patient_id, body.lines, saved, confirmed_visit=confirmed + ) + response["lines"] = [l.model_dump() for l in lines_out] + response["rollup_status"] = rollup_status + + return response diff --git a/python-backend/core/overrides.py b/python-backend/core/overrides.py new file mode 100644 index 0000000..5d8b51f --- /dev/null +++ b/python-backend/core/overrides.py @@ -0,0 +1,130 @@ +""" +overrides.py +Signal — STTIL Solutions + +Device-keyed staff overrides (build plan 02, P4; phase-2 design Item 4). + +Pilot scope is LOCKED to two override kinds: +- visit: patient-scoped (the visit is the person). It rides the existing + confirmed_visits path, which already fans out to every line of the patient + by construction (evaluate_readiness receives the confirmed date per + patient). This module does not duplicate that mechanism. +- swo / pa: line-scoped. They apply only to lines whose device_type matches; + device_type '' marks a legacy patient-scoped row and matches all devices + until staff re-saves it device-scoped. + +Overrides feed the GRADING inputs (staff knowledge wins over the CSV cell, +matching the existing display precedence), then affected lines re-grade. + +PHI CONTRACT: overrides carry patient_id_hash only (the storage key). Lines +carry raw patient_id; matching hashes it. Nothing here logs patient data. +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from datetime import date, datetime +from typing import Optional + +# Stored status -> display/grading label. Single source of truth; the API's +# display map imports this. The labels feed the readiness vocabularies +# ("On File" grades GOOD, "Denied" grades BAD, "Requested" grades PENDING). +STORED_STATUS_LABELS = { + "pending": "Pending", + "requested": "Requested", + "on_file": "On File", + "not_required": "Not Required", + "approved": "Approved", + "denied": "Denied", +} + +# Legacy patient-scoped rows carry an empty device_type and match all devices. +LEGACY_ALL_DEVICES = "" + +OVERRIDABLE_DOC_TYPES = ("swo", "pa") + + +@dataclass(frozen=True) +class OverrideKey: + org_id: str + patient_hash: str + device_type: str # '' = legacy patient-scoped, applies to all devices + doc_type: str # "swo" | "pa" (visit rides confirmed_visits) + + +@dataclass +class Override: + key: OverrideKey + status: str # stored value, e.g. "on_file" | "approved" | "denied" + status_date: Optional[date] = None + expiry_date: Optional[date] = None + confirmed_by: str = "staff" + confirmed_at: Optional[datetime] = None + + +def _hash_pid(patient_id: str) -> str: + return hashlib.sha256(patient_id.encode()).hexdigest() + + +def grading_value(override: Override) -> str: + """The label fed to the readiness vocabularies for this override.""" + return STORED_STATUS_LABELS.get(override.status, override.status) + + +def overrides_from_saved(saved: dict) -> list[Override]: + """Build Override objects from the persistence load shape: + patient_hash -> device_type -> doc_type -> {status, status_date, ...}.""" + out: list[Override] = [] + for patient_hash, by_device in (saved or {}).items(): + for device_type, by_doc in by_device.items(): + for doc_type, row in by_doc.items(): + if doc_type not in OVERRIDABLE_DOC_TYPES: + continue + out.append( + Override( + key=OverrideKey( + org_id="", + patient_hash=patient_hash, + device_type=device_type, + doc_type=doc_type, + ), + status=row.get("status", ""), + ) + ) + return out + + +def apply_overrides(lines: dict, overrides: list[Override]) -> set[str]: + """Apply swo/pa overrides onto the grading inputs of matching lines. + + Mutates each matching line's current_shipment csv fields (staff knowledge + wins over the CSV cell). Device-scoped rows win over legacy '' rows. + Returns the raw patient_ids of touched lines. Grading (or re-grading) + happens in the caller — build_readiness_index grades after application. + """ + if not overrides: + return set() + + by_patient: dict[str, list[Override]] = {} + for o in overrides: + by_patient.setdefault(o.key.patient_hash, []).append(o) + + touched: set[str] = set() + for line in lines.values(): + patient_overrides = by_patient.get(_hash_pid(line.patient_id)) + if not patient_overrides: + continue + # Legacy '' first so a device-scoped row applied after it wins. + for o in sorted( + patient_overrides, key=lambda o: o.key.device_type != LEGACY_ALL_DEVICES + ): + if o.key.device_type not in (LEGACY_ALL_DEVICES, line.device_type): + continue + label = grading_value(o) + if o.key.doc_type == "swo": + line.current_shipment.csv_swo_status = label + elif o.key.doc_type == "pa": + line.current_shipment.csv_pa_status = label + touched.add(line.patient_id) + return touched diff --git a/python-backend/core/persistence.py b/python-backend/core/persistence.py index 3af094b..da34d02 100644 --- a/python-backend/core/persistence.py +++ b/python-backend/core/persistence.py @@ -227,9 +227,12 @@ def upsert_doc_status( status: str, status_date: "date | None" = None, expiry_date: "date | None" = None, + device_type: str = "", ) -> bool: """ - Insert or update a doc status (swo or pa) for a patient. + Insert or update a doc status (swo or pa), scoped to a device. + device_type '' writes a legacy patient-scoped row (applies to all of the + patient's devices until a device-scoped row is saved). Returns True on success, False if Supabase unavailable. """ client = get_client() @@ -241,6 +244,7 @@ def upsert_doc_status( "patient_id_hash": patient_id_hash, "doc_type": doc_type, "status": status, + "device_type": device_type or "", "updated_at": "now()", } if status_date is not None: @@ -248,7 +252,7 @@ def upsert_doc_status( if expiry_date is not None: payload["expiry_date"] = expiry_date.isoformat() client.table("doc_status").upsert( - payload, on_conflict="org_id,patient_id_hash,doc_type" + payload, on_conflict="org_id,patient_id_hash,doc_type,device_type" ).execute() return True except Exception as e: @@ -258,23 +262,24 @@ def upsert_doc_status( def load_doc_statuses_for_org(org_id: str) -> dict: """ - Load all SWO and PA statuses for an org. - Returns dict: patient_id_hash -> {doc_type: {status, status_date, expiry_date}} + Load all SWO and PA statuses for an org, device-scoped. + Returns dict: patient_id_hash -> device_type -> {doc_type: {status, + status_date, expiry_date}}. The '' device bucket holds legacy + patient-scoped rows (they apply to every device until re-saved). """ client = get_client() if not client: return {} try: result = client.table("doc_status") \ - .select("patient_id_hash,doc_type,status,status_date,expiry_date") \ + .select("patient_id_hash,device_type,doc_type,status,status_date,expiry_date") \ .eq("org_id", org_id) \ .execute() out: dict = {} for row in (result.data or []): h = row["patient_id_hash"] - if h not in out: - out[h] = {} - out[h][row["doc_type"]] = { + device = row.get("device_type") or "" + out.setdefault(h, {}).setdefault(device, {})[row["doc_type"]] = { "status": row["status"], "status_date": row.get("status_date"), "expiry_date": row.get("expiry_date"), diff --git a/python-backend/core/worklist_readiness.py b/python-backend/core/worklist_readiness.py index 860a2ae..f925550 100644 --- a/python-backend/core/worklist_readiness.py +++ b/python-backend/core/worklist_readiness.py @@ -82,18 +82,27 @@ def _confirmed_for( def build_readiness_index( records: list[ShipmentRecord], confirmed_visits: Optional[dict[str, date]] = None, + overrides: Optional[list] = None, ) -> ReadinessIndex: - """Dedup records into coverage lines, grade each gradeable line, and - record the group->line membership map so row-level resolution never - re-derives it. + """Dedup records into coverage lines, apply staff overrides to the + grading inputs (P4: swo/pa device-scoped; staff knowledge wins over the + CSV cell), grade each gradeable line, and record the group->line + membership map so row-level resolution never re-derives it. Grades the CURRENT shipment only (latest DOS per line — locked rule); prior shipments are collapsed history. Non-gradeable lines are skipped. + Visit overrides arrive via confirmed_visits (patient-scoped fan-out by + construction). """ confirmed_visits = confirmed_visits or {} merged = dedup(records) lines = assign_to_coverage_lines(merged) + if overrides: + from core.overrides import apply_overrides + + apply_overrides(lines, overrides) + index = ReadinessIndex() for key, line in lines.items(): line_key = (key.patient_id, key.device_type, key.plan_type_id) diff --git a/signal-ui/src/components/WorklistTable.jsx b/signal-ui/src/components/WorklistTable.jsx index c4499dc..ac912a4 100644 --- a/signal-ui/src/components/WorklistTable.jsx +++ b/signal-ui/src/components/WorklistTable.jsx @@ -347,22 +347,22 @@ export default function WorklistTable({ records, activeFilter, onFilterChange, o
setLocalDocStates(prev => ({ - ...prev, [r.patient_id]: { ...prev[r.patient_id], [type]: label } + ...prev, [`${r.patient_id}:${r.device_type}`]: { ...prev[`${r.patient_id}:${r.device_type}`], [type]: label } }))} /> setLocalDocStates(prev => ({ - ...prev, [r.patient_id]: { ...prev[r.patient_id], [type]: label } + ...prev, [`${r.patient_id}:${r.device_type}`]: { ...prev[`${r.patient_id}:${r.device_type}`], [type]: label } }))} /> @@ -492,7 +492,7 @@ function ActionCell({ flag, cascade, showConfirmVisit, onConfirmVisit, onExpand return null; } -function DocItem({ label, value, patientId, docType, cycle, getToken, onUpdated }) { +function DocItem({ label, value, patientId, deviceType = "", docType, cycle, getToken, onUpdated }) { const [saving, setSaving] = useState(false); const isGood = value && ( @@ -514,7 +514,7 @@ function DocItem({ label, value, patientId, docType, cycle, getToken, onUpdated const nextLabel = DOC_STATUS_LABELS[nextKey]; setSaving(true); const token = getToken ? await getToken().catch(() => null) : null; - const result = await updateDocStatus(patientId, docType, nextKey, null, null, token); + const result = await updateDocStatus(patientId, docType, nextKey, null, null, token, deviceType); setSaving(false); if (result && onUpdated) onUpdated(docType, nextLabel); }; diff --git a/signal-ui/src/lib/api.js b/signal-ui/src/lib/api.js index ad4985b..daea4c2 100644 --- a/signal-ui/src/lib/api.js +++ b/signal-ui/src/lib/api.js @@ -141,7 +141,7 @@ export async function confirmVisit(payload, token = null) { * @param {string|null} token - Clerk JWT * @returns {Promise<{patient_id, doc_type, status, display}|null>} */ -export async function updateDocStatus(patientId, docType, status, statusDate = null, expiryDate = null, token = null) { +export async function updateDocStatus(patientId, docType, status, statusDate = null, expiryDate = null, token = null, deviceType = "") { const authHeader = token ? { "Authorization": `Bearer ${token}` } : {}; @@ -153,6 +153,9 @@ export async function updateDocStatus(patientId, docType, status, statusDate = n patient_id: patientId, doc_type: docType, status, + // Device scope (P4): an SWO/PA update applies to this device's line + // only, never to every device the patient has. + ...(deviceType && { device_type: deviceType }), ...(statusDate && { status_date: statusDate }), ...(expiryDate && { expiry_date: expiryDate }), }), diff --git a/tests/test_overrides.py b/tests/test_overrides.py new file mode 100644 index 0000000..778e328 --- /dev/null +++ b/tests/test_overrides.py @@ -0,0 +1,311 @@ +""" +Tests for device-keyed staff overrides (build plan 02, P4). + +Locked semantics (phase-2 design Item 4 / brief section 6): +- visit overrides are patient-scoped and fan out to ALL the patient's lines + (they ride the confirmed_visits path by construction); +- swo/pa overrides apply only to lines whose device_type matches; +- device_type '' marks a legacy patient-scoped row and matches all devices; +- staff knowledge wins over the CSV cell; +- after a write, the recompute returns {lines, rollup} from the same engine. + +PHI CONTRACT: synthetic patient_ids only; overrides carry hashes. +""" + +import hashlib +import io +import sys +from datetime import date, timedelta +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "python-backend")) + +from core.coverage_calculator import ShipmentRecord # noqa: E402 +from core.overrides import ( # noqa: E402 + Override, + OverrideKey, + apply_overrides, + overrides_from_saved, +) +from core.worklist_readiness import build_readiness_index # noqa: E402 + +TODAY = date.today() + + +def _hash(pid: str) -> str: + return hashlib.sha256(pid.encode()).hexdigest() + + +def _record(**kwargs) -> ShipmentRecord: + defaults = dict( + patient_id="PT-OV-1", + device_type="dexcom_g7", + shipment_date=TODAY - timedelta(days=5), + quantity=3, + payer="Medicare Part B", + component="sensor", + plan_type="medicare", + csv_visit_date=TODAY - timedelta(days=30), + csv_pecos_verified="Yes", + csv_diagnosis_on_file="Yes", + ) + defaults.update(kwargs) + return ShipmentRecord(**defaults) + + +def _swo_override(pid="PT-OV-1", device="dexcom_g7", status="on_file"): + return Override( + key=OverrideKey( + org_id="org", patient_hash=_hash(pid), device_type=device, + doc_type="swo", + ), + status=status, + ) + + +def test_override_swo_applies_only_to_one_device(): + # Two devices, both missing SWO -> Action Needed. Override device A only. + g7 = _record(device_type="dexcom_g7") + libre = _record(device_type="freestyle_libre_2") + index = build_readiness_index( + [g7, libre], overrides=[_swo_override(device="dexcom_g7")] + ) + assert index.by_line[("PT-OV-1", "dexcom_g7", "medicare")].line_status.value == "Clear to Ship" + assert index.by_line[("PT-OV-1", "freestyle_libre_2", "medicare")].line_status.value == "Action Needed" + + +def test_legacy_blank_device_row_applies_patient_wide(): + g7 = _record(device_type="dexcom_g7") + libre = _record(device_type="freestyle_libre_2") + index = build_readiness_index( + [g7, libre], overrides=[_swo_override(device="")] + ) + for key, verdict in index.by_line.items(): + assert verdict.line_status.value == "Clear to Ship", key + + +def test_device_scoped_row_wins_over_legacy_row(): + g7 = _record(device_type="dexcom_g7") + index = build_readiness_index( + [g7], + overrides=[ + _swo_override(device="", status="on_file"), + _swo_override(device="dexcom_g7", status="pending"), + ], + ) + verdict = index.by_line[("PT-OV-1", "dexcom_g7", "medicare")] + swo = next(i for i in verdict.doc_items if i.doc_type == "swo") + assert swo.quality.value == "PENDING" # device-scoped Pending won + + +def test_override_triggers_recompute_to_better_status(): + # SWO missing is the last open required item; the override flips the line. + rec = _record() + base = build_readiness_index([rec]) + assert base.by_line[("PT-OV-1", "dexcom_g7", "medicare")].line_status.value == "Action Needed" + fixed = build_readiness_index([rec], overrides=[_swo_override()]) + assert fixed.by_line[("PT-OV-1", "dexcom_g7", "medicare")].line_status.value == "Clear to Ship" + + +def test_override_visit_applies_to_all_patient_lines(): + # Visit rides confirmed_visits: patient-scoped fan-out by construction. + g7 = _record(device_type="dexcom_g7", csv_visit_date=None, + csv_swo_status="On File") + libre = _record(device_type="freestyle_libre_2", csv_visit_date=None, + csv_swo_status="On File") + confirmed = {_hash("PT-OV-1"): TODAY - timedelta(days=10)} + index = build_readiness_index([g7, libre], confirmed_visits=confirmed) + for key, verdict in index.by_line.items(): + visit = next(i for i in verdict.doc_items if i.doc_type == "visit") + assert visit.satisfied, key + + +def test_overrides_from_saved_shape(): + saved = { + _hash("PT-OV-1"): { + "": {"swo": {"status": "on_file"}}, + "dexcom_g7": {"pa": {"status": "approved"}}, + } + } + overrides = overrides_from_saved(saved) + assert len(overrides) == 2 + kinds = {(o.key.device_type, o.key.doc_type, o.status) for o in overrides} + assert kinds == {("", "swo", "on_file"), ("dexcom_g7", "pa", "approved")} + + +def test_apply_overrides_returns_touched_patients(): + g7 = _record() + other = _record(patient_id="PT-OV-2") + index_lines = build_readiness_index([g7, other]).lines + touched = apply_overrides(index_lines, [_swo_override()]) + assert touched == {"PT-OV-1"} + + +# --- API level --------------------------------------------------------------- + +httpx = pytest.importorskip("httpx", reason="httpx required for TestClient") +from fastapi.testclient import TestClient # noqa: E402 + +import api.main as api_main # noqa: E402 + +client = TestClient(api_main.app) + + +def test_doc_status_put_accepts_device_and_returns_line_and_rollup(monkeypatch): + saved_writes = {} + + def fake_upsert(org_id, patient_hash, doc_type, status, + status_date=None, expiry_date=None, device_type=""): + saved_writes[(patient_hash, doc_type, device_type)] = status + return True + + def fake_load(org_id): + # Echo the persisted write back, as a fresh load would. + out = {} + for (ph, dt, dev), status in saved_writes.items(): + out.setdefault(ph, {}).setdefault(dev, {})[dt] = {"status": status} + return out + + monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test") + monkeypatch.setattr(api_main, "upsert_doc_status", fake_upsert) + monkeypatch.setattr(api_main, "load_doc_statuses_for_org", fake_load) + monkeypatch.setattr( + api_main, "load_confirmed_visits_for_org", lambda org: {} + ) + + resp = client.put( + "/api/doc-status", + json={ + "patient_id": "PT-API-1", + "doc_type": "swo", + "status": "on_file", + "device_type": "dexcom_g7", + "lines": [ + { + "device_type": "dexcom_g7", + "plan_type": "medicare", + "csv_visit_date": (TODAY - timedelta(days=20)).isoformat(), + "csv_pecos_verified": "Yes", + "csv_diagnosis_on_file": "Yes", + }, + { + "device_type": "freestyle_libre_2", + "plan_type": "medicare", + "csv_visit_date": (TODAY - timedelta(days=20)).isoformat(), + "csv_pecos_verified": "Yes", + "csv_diagnosis_on_file": "Yes", + }, + ], + }, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["device_type"] == "dexcom_g7" + by_dev = {l["device_type"]: l for l in body["lines"]} + # The override satisfied G7's SWO; the Libre line is untouched (no leak). + assert by_dev["dexcom_g7"]["line_status"] == "Clear to Ship" + assert by_dev["freestyle_libre_2"]["line_status"] == "Action Needed" + assert body["rollup_status"] == "Action Needed" # worst line + + +def test_doc_status_put_without_lines_keeps_legacy_shape(monkeypatch): + monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test") + monkeypatch.setattr( + api_main, "upsert_doc_status", lambda *a, **kw: True + ) + resp = client.put( + "/api/doc-status", + json={"patient_id": "PT-API-2", "doc_type": "pa", "status": "approved"}, + ) + assert resp.status_code == 200 + body = resp.json() + assert body["display"] == "Approved" + assert "lines" not in body + + +def test_confirm_visit_with_lines_returns_rerolled_patient(monkeypatch): + monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test") + monkeypatch.setattr( + api_main, "upsert_confirmed_visit", lambda *a, **kw: True + ) + monkeypatch.setattr( + api_main, "load_doc_statuses_for_org", lambda org: {} + ) + ship = (TODAY - timedelta(days=5)).isoformat() + resp = client.post( + "/api/confirm-visit", + json={ + "patient_id": "PT-API-3", + "confirmed_date": (TODAY - timedelta(days=15)).isoformat(), + "shipment_date": ship, + "payer": "Medicare Part B", + "device_type": "dexcom_g7", + "quantity": 3, + "csv_swo_status": "On File", + "csv_pecos_verified": "Yes", + "csv_diagnosis_on_file": "Yes", + "plan_type": "medicare", + "lines": [ + { + "device_type": "dexcom_g7", + "plan_type": "medicare", + "csv_swo_status": "On File", + "csv_pecos_verified": "Yes", + "csv_diagnosis_on_file": "Yes", + }, + { + "device_type": "freestyle_libre_2", + "plan_type": "medicare", + "csv_swo_status": "On File", + "csv_pecos_verified": "Yes", + "csv_diagnosis_on_file": "Yes", + }, + ], + }, + ) + assert resp.status_code == 200 + body = resp.json() + # Legacy record fields still present + assert body["readiness_status"] == "Clear to Ship" + # The confirmed visit fanned out to BOTH echoed lines + assert all(l["line_status"] == "Clear to Ship" for l in body["lines"]) + assert body["rollup_status"] == "Clear to Ship" + + +def test_upload_grades_with_saved_device_scoped_override(monkeypatch): + """An SWO override saved for device A feeds the verdict for A's line only.""" + pid_hash = hashlib.sha256("PT-UP-1".encode()).hexdigest() + monkeypatch.setattr( + api_main, + "load_doc_statuses_for_org", + lambda org: { + pid_hash: {"dexcom_g7": {"swo": {"status": "on_file"}}} + }, + ) + monkeypatch.setattr(api_main, "get_or_create_org", lambda **kw: "org-test") + monkeypatch.setattr( + api_main, "load_confirmed_visits_for_org", lambda org: {} + ) + ship = (TODAY - timedelta(days=5)).isoformat() + visit = (TODAY - timedelta(days=30)).isoformat() + header = ( + "patient_id,device_type,shipment_date,quantity,payer,plan_type," + "visit_date,pecos_verified,diagnosis_on_file" + ) + csv_text = ( + header + + f"\nPT-UP-1,Dexcom G7,{ship},3,Medicare Part B,medicare,{visit},Yes,Yes" + + f"\nPT-UP-1,Libre 2,{ship},1,Medicare Part B,medicare,{visit},Yes,Yes" + ) + resp = client.post( + "/api/upload", + files={"file": ("f.csv", io.BytesIO(csv_text.encode()), "text/csv")}, + ) + assert resp.status_code == 200 + recs = {r["device_type"]: r for r in resp.json()["records"]} + assert recs["dexcom_g7"]["readiness_status"] == "Clear to Ship" + assert recs["freestyle_libre_2"]["readiness_status"] == "Action Needed" + # Display follows the same device scope + assert recs["dexcom_g7"]["doc_state"]["swo"] == "On File"