Signal/docs/build-plans/03-game-center-token-layer.md
Kisa 03afd62caf docs(tokens): Game Center token layer implementation plan (03) - handoff-ready
Design fan-out (cto-architect) + primary-source compliance verification
(lead-sttil) + independent adversarial audit (code-auditor, 2 HIGH findings
fixed: registry test fixture eager-delete, offboarding purge table list vs
real schema.sql). Executable by a fresh Sonnet/Opus session. Token mode
ships dark; K1-K6 decisions pending Kisa.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 05:49:17 -04:00

61 KiB
Raw Blame History

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 P2P5 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 (K1K6 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):

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/<file> -v.


3. Database schema (Task 3 migration, shown here for reference)

-- 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
# 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-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
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
# 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-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
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)

# 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-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

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):
    TOKEN_MINT = "token_mint"
    ORG_REVOKE = "org_revoke"
  • Step 2: Write the failing tests
# 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:

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):

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):

@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 suitepython -m pytest tests/ -v — expected: no regressions vs the precondition count.

  • Step 7: Commit

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
# 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:

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):

    "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:

    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:

        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 suitepython -m pytest tests/ -v — the pre-existing normalizer/e2e tests must all still pass (they run in default mode).

  • Step 6: Commit

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
# 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 {}:
    # 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
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
        resource_id=f"confirm-visit:{body.patient_id[:8]}",

to

        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:

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 suitepython -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

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):
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):
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

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
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
# 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
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
# 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 checkpython -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

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 unchangedcurl -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-gatedcurl -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 [<you>]: 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.