# 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 readiness-model P2–P5 lane (`docs/build-plans/02-readiness-model-completion.md`) is actively committing to this repo; the exact dirty-file set changes hour to hour, so trust `git status --short`, not any file list written here. 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 — a confirming or reversing line will appear there once she rules; absent means still pending. This plan implements the recommended defaults, and everything ships DARK (default-off), so building ahead of the rulings is safe: the code is inert until `IDENTITY_MODE=token` is set, and no K-decision is irreversible in code. If Kisa reversed any, adapt before building. K1 note: `context/current-state.md` Stack still records the older "rotating RAP-ID/CSP-ID" direction — that conflict was surfaced to Kisa 2026-07-07 with this plan; do not treat the Stack line as a K1 ruling. ### 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; revoke endpoint) Tasks 4, 6, 7, 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): # Defer to execute(): the real client chains .delete().eq(...).execute(), # so filters are not yet set when delete() is called. self._pending_delete = True return self def execute(self): if getattr(self, "_pending_delete", False): removed = [r for r in self.store.get(self.name, []) if all(r.get(k) == v for k, v in self._filters.items())] self.store[self.name] = [ r for r in self.store.get(self.name, []) if r not in removed ] return FakeResponse(removed) 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 # Tables purged on offboarding. Verified against python-backend/db/schema.sql # 2026-07-07: ONLY these carry an org_id column. Child tables purge via # cascade — upload_batches cascades to source_files/normalized_records/ # mapping_decisions/raw_rows; report_runs cascades to export_files/ # report_items. Do NOT add a table here unless schema.sql shows it has # org_id (a delete().eq("org_id", ...) on a table without that column # raises, and offboarding would report failure on every call). # audit_log is intentionally EXCLUDED (WORM — the tombstone lives there). ORG_SCOPED_PURGE_TABLES = [ "tokens", "token_mint_batches", "upload_batches", "report_runs", "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: the operational DDL lives in `python-backend/db/schema.sql` (NOT in `supabase/migrations/`, which holds only two later migrations). Before running the registry tests, re-verify each table in `ORG_SCOPED_PURGE_TABLES` still has an `org_id` column there (`grep -n "org_id" python-backend/db/schema.sql`), and record the verification in the Task 3 commit message. IMPORTANT: the unit tests use a fake client and CANNOT catch a wrong table/column name — Task 11 includes the live check that a real `revoke_org` returns a clean receipt (no `-1` values). - [ ] **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 ``` (Installed pydantic verified 2.12.5 on 2026-07-07, so `ConfigDict` is correct.) (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`. Anchor precisely: `org_id = get_or_create_org(...)` appears in THREE handlers; the correct spot is the one inside `upload_csv`, identified by the unique line that follows it — insert this block immediately BEFORE `confirmed_visits = load_confirmed_visits_for_org(org_id) if org_id else {}`: ```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 embed a raw 8-char identifier prefix in `resource_id`. Honest framing: `build_audit_entry` already SHA-256s `resource_id` into `resource_hash` before any sink, so this is NOT an active raw-identifier leak — the fix is defense in depth (a low-entropy 8-char prefix inside the hashed string weakens it against brute-forcing, and pre-hashing keeps every code path consistent with the house rule). Apply unconditionally. - [ ] **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 + ~30 new tests (7 format, 4 mint, 3 registry, 7 endpoint, 5 normalizer, 2 guard, 2 e2e), 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** — first confirm the build image includes the new package: check the Dockerfile / Railway build config copies `python-backend/` wholesale (it must include `core/tokens/` — `normalizer.py` now imports from it at module load, so a missing package breaks the DEFAULT pipeline too). Then from the repo root: `railway up --detach` (code change → fresh build; `railway redeploy` only reuses the image). 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 4b: Live offboarding receipt check (required — the unit fake cannot catch schema drift):** against a staging/dev deployment with a throwaway org, mint a small batch, upload one token-mode CSV, then call `POST /api/org/revoke` and verify the receipt contains NO `-1` values and the counts look sane. A `-1` means a purge-table name/column no longer matches `python-backend/db/schema.sql`. - [ ] **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.