Phase 0 milestone 0.5: Registrant vertical slice

Registrant model + schemas + router + migration 0003, exercising the
vendored framework end-to-end (CRUDService, RBAC, audit, version history).

- SSN app-layer encrypted (AES-256-GCM, HKDF key, column-bound AAD,
  versioned format) in security/column_crypto.py; write-only on the model;
  serializer exposes only ssn_masked (XXX-XX-1234). C.5.7.50 / SC-28.
- pii_read audit on GET, list (one row per page with ids), and history.
  C.5.6.9 #5 / AU-14.
- Portal-only callers (YOUTH/PARENT/HOST) limited to own row via
  Registrant.user_id; out-of-scope is 404. PARENT_USER sees nothing until
  Guardian lands. History gated on VersionHistory, not View.
- minor_flag derived from dob at read time.

Framework changes (flag to OWC on re-vendor):
- crud._to_dict honors __audit_redact__, keeping ciphertext out of
  audit_log and version_history.
- crud._to_dict serializes date (not only datetime) for JSON columns.
- CRUDService.history annotation uses builtins.list (method shadowed it).

98 tests passing, 94.83% coverage; ruff + mypy clean.
Coverage trace at docs/coverage/registrant.md.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

docs/coverage/registrant.md

@@ -0,0 +1,55 @@
+# Registrant — requirement coverage
+
+**Milestone:** Phase 0 / 0.5 (vertical slice) · **As of:** 2026-09-30
+**Entity:** `Registrant` (ARCHITECTURE.md §6.1) · **RBAC entity string:** `Registrant`
+
+Traces each requirement the Registrant slice claims to the code and test that
+evidence it. Auditors: file paths are relative to `ypms/`.
+
+---
+
+## Surface
+
+| Method | Path | Permission | Portal-only callers | PII-read audit |
+|---|---|---|---|---|
+| `GET` | `/registrants` | `List` | 403 | one `pii_read` row per page, `after.ids` = returned ids |
+| `GET` | `/registrants/{id}` | `View` | own row only (YOUTH_USER); else 404 | `pii_read` row, `entity_id` = id |
+| `POST` | `/registrants` | `Create` | 403 | — (create audit row) |
+| `PATCH` | `/registrants/{id}` | `Edit` | own row only (YOUTH_USER); else 404 | — (update audit row) |
+| `DELETE` | `/registrants/{id}` | `Delete` | 403 | — (delete audit row) |
+| `POST` | `/registrants/{id}/restore` | `Edit` | 403 | — (restore audit row) |
+| `GET` | `/registrants/{id}/history` | `VersionHistory` | 403 | `pii_read` row, `entity_id` = id |
+
+"Portal-only" = every role the caller holds is in `{YOUTH_USER, PARENT_USER, HOST_USER}`. A caller with any staff role gets staff behaviour. PARENT_USER sees no rows until the Guardian link lands (Phase 1). No bulk endpoints for Registrant in Phase 0.
+
+---
+
+## Requirement trace
+
+| Requirement | What the code does | Evidence (code) | Evidence (test) |
+|---|---|---|---|
+| Source RFP `C.5.3.1`, `C.5.3.4` — youth application / account management | Registrant CRUD with soft delete, restore, cursor pagination | `src/ypms/youth/router.py` | `tests/youth/test_registrant_api.py` |
+| Source RFP `C.5.7.50`; ARCH §6.3 — SSN masked in display | Read schema exposes only `ssn_masked` (`XXX-XX-1234`); masking in the serializer, so API consumers see it too | `src/ypms/youth/schemas.py` (`RegistrantRead`), `src/ypms/youth/models.py` (`ssn_masked`) | `test_create_masks_ssn_and_never_returns_it` |
+| NIST SC-28, SC-28(1); ARCH §6.3 — app-layer encryption of SSN | AES-256-GCM, HKDF-derived key, per-value nonce, column-bound AAD, versioned format; SSN is write-only on the model | `src/ypms/security/column_crypto.py`, `src/ypms/youth/models.py` (`ssn` setter, `decrypt_ssn`) | `test_roundtrip_and_fresh_nonce`, `test_ciphertext_is_bound_to_column`, `test_ssn_is_write_only` |
+| NIST AU-2, AU-3, AU-12 — every state change audited | Create / update / delete / restore emit audit rows with actor, before, after | `src/ypms/framework/crud.py` | `test_create_emits_redacted_audit_and_version_snapshot`, `test_update_bumps_version_and_audits`, `test_delete_requires_delete_permission_then_restore` |
+| NIST AU-14; source RFP `C.5.6.9` #5 — record access to sensitive data | GET, list, and history emit `pii_read` rows with actor and client IP | `src/ypms/framework/pii_audit.py`, `src/ypms/youth/router.py` | `test_get_emits_pii_read`, `test_list_emits_one_pii_read_with_ids`, `test_history_requires_version_history_permission` |
+| NIST AU-9 (protection of audit info) — no secrets in the audit trail | `__audit_redact__` replaces `ssn_encrypted` with `[REDACTED]` in audit before/after and version-history snapshots | `src/ypms/framework/crud.py` (`_to_dict`), `src/ypms/youth/models.py` | `test_create_emits_redacted_audit_and_version_snapshot`, `test_update_bumps_version_and_audits` |
+| NIST AC-3, AC-6 — least privilege | Data-driven RBAC per route; history gated on `VersionHistory` (not `View`); delete requires `Delete` (OYP_STAFF lacks it) | `src/ypms/youth/router.py`, `src/ypms/seed/permissions_matrix.json` | `test_history_requires_version_history_permission`, `test_delete_requires_delete_permission_then_restore`, `test_host_user_denied` |
+| NIST AC-3 — row-level restriction for external users (Phase 0 minimum) | Portal-only callers limited to own row by `user_id`; out-of-scope → 404 (no id probing) | `src/ypms/youth/router.py` (`_get_scoped`, `_require_staff`) | `test_youth_can_read_and_edit_only_own_row`, `test_youth_blocked_from_staff_routes`, `test_parent_sees_nothing_until_guardian_link` |
+| NIST SI-10 — input validation | SSN format + SSA never-issued ranges; DOB not in future; ZIP, email, length limits; NOT NULL fields can't be PATCHed to null; `user_id` not writable | `src/ypms/youth/schemas.py` | `test_create_validation`, `test_update_rejects_null_on_required_field`, `test_update_cannot_set_user_id` |
+| ARCH §3 "minors first" — minor status | `minor_flag` derived from DOB at read time, never stored | `src/ypms/youth/models.py` (`minor_flag`, `age_on`) | `test_minor_flag_derived_from_dob`, `test_age_on_boundaries` |
+| NIST CM-3; source RFP `C.5.8.12` — schema change control | New immutable Alembic revision | `migrations/versions/0003_registrants.py` | rendered offline (`alembic upgrade 0002_framework_tables:0003_registrants --sql`); live Postgres run pending |
+
+---
+
+## Deferred (not in 0.5)
+
+| Item | Why deferred | Lands in |
+|---|---|---|
+| Banking columns (encrypted account, last-4) | Needs PAYROLL_ADMIN-only field scoping built alongside it; payroll vendor not selected (`YPMS-POAM-052`) | Payroll phase |
+| SSN blind index (HMAC) for full-SSN dedupe/search | `ssn_last4` covers Phase 0 display/search; dedupe is an intake concern | Phase 1 (application intake) |
+| `Search` endpoint | Matrix grants `Search`; no route yet | Phase 1 |
+| Guardian-scoped access for PARENT_USER | Needs `Guardian` + `ConsentRecord` | Phase 1 |
+| Account-link flow setting `user_id` | Identity model open (`YPMS-POAM-050`) | Phase 1 |
+| Key Vault–backed key + `v2` ciphertext | Phase 0 uses env-var key material | Phase 1 (`YPMS-POAM-032`) |
+| Demographics, immigration-doc refs, tax elections | Not in PHASE0_PLAN §4 sketch | Phase 1+ |

migrations/env.py

@@ -19,6 +19,7 @@ import ypms.users.models  # noqa: F401
 import ypms.security.models  # noqa: F401
 import ypms.audit.models  # noqa: F401
 import ypms.framework.version_history  # noqa: F401
+import ypms.youth.models  # noqa: F401
 
 config = context.config
 if config.config_file_name is not None:

migrations/versions/0003_registrants.py

@@ -0,0 +1,71 @@
+"""Registrants table — Phase 0 vertical slice.
+
+Revision ID: 0003_registrants
+Revises: 0002_framework_tables
+Create Date: 2026-09-30
+
+Milestone 0.5. ARCHITECTURE.md §6.1 / PHASE0_PLAN.md §4.
+
+- `ssn_encrypted` holds AES-256-GCM ciphertext from security/column_crypto.py
+  (NIST SC-28; source RFP `C.5.7.50`). Never plaintext.
+- `ssn_last4` is the only SSN-derived value stored in the clear — for search
+  and masked display (``XXX-XX-1234``).
+- No `minor_flag` column: derived from `dob` at read time.
+- `user_id` links the youth's own portal account (Phase 1 account-link flow).
+
+Per repo rule (CONTRIBUTING.md): this file is immutable once merged.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0003_registrants"
+down_revision: str | Sequence[str] | None = "0002_framework_tables"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "registrants",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("first_name", sa.String(length=100), nullable=False),
+        sa.Column("middle_name", sa.String(length=100), nullable=True),
+        sa.Column("last_name", sa.String(length=100), nullable=False),
+        sa.Column("dob", sa.Date(), nullable=False),
+        sa.Column("ssn_encrypted", sa.String(length=255), nullable=True),
+        sa.Column("ssn_last4", sa.String(length=4), nullable=True),
+        sa.Column("email", sa.String(length=255), nullable=True),
+        sa.Column("phone", sa.String(length=32), nullable=True),
+        sa.Column("address_line1", sa.String(length=255), nullable=True),
+        sa.Column("address_line2", sa.String(length=255), nullable=True),
+        sa.Column("city", sa.String(length=100), nullable=True),
+        sa.Column("state", sa.String(length=2), nullable=True),
+        sa.Column("zip", sa.String(length=10), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("version", sa.Integer(), nullable=False, server_default=sa.text("1")),
+    )
+    op.create_index("ix_registrants_user_id", "registrants", ["user_id"])
+    op.create_index("ix_registrants_last_name", "registrants", ["last_name"])
+    op.create_index("ix_registrants_dob", "registrants", ["dob"])
+    op.create_index("ix_registrants_ssn_last4", "registrants", ["ssn_last4"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_registrants_ssn_last4", table_name="registrants")
+    op.drop_index("ix_registrants_dob", table_name="registrants")
+    op.drop_index("ix_registrants_last_name", table_name="registrants")
+    op.drop_index("ix_registrants_user_id", table_name="registrants")
+    op.drop_table("registrants")

src/ypms/framework/crud.py

@@ -12,11 +12,18 @@ upstream:
   which combined with `where id > cursor` on the next call silently skipped
   that row. Caught by `tests/framework/test_crud.py::test_list_pagination_forwards_cursor`.
   Flag to OWC when re-vendoring.
+- **Snapshot redaction** (milestone 0.5): models may declare
+  `__audit_redact__ = ("col", ...)`; `_to_dict` replaces those columns' values
+  with `"[REDACTED]"` (nulls stay null, so "was it set" is still visible).
+  Keeps app-layer ciphertext (e.g. `registrants.ssn_encrypted`) out of
+  `audit_log` and `version_history`. Flag to OWC — Claimant SSN has the same
+  exposure there.
 """
 
 from __future__ import annotations
 
-from datetime import UTC, datetime
+import builtins
+from datetime import UTC, date, datetime
 from typing import Any, Generic, TypeVar
 
 from pydantic import BaseModel
@@ -33,14 +40,19 @@ ModelT = TypeVar("ModelT")
 CreateT = TypeVar("CreateT", bound=BaseModel)
 UpdateT = TypeVar("UpdateT", bound=BaseModel)
 
+REDACTED = "[REDACTED]"
+
 
 def _to_dict(obj: Any) -> dict[str, Any]:
     if obj is None:
         return {}
+    redact = getattr(obj, "__audit_redact__", ())
     out: dict[str, Any] = {}
     for c in obj.__table__.columns:
         v = getattr(obj, c.name, None)
-        if isinstance(v, datetime):
+        if c.name in redact and v is not None:
+            v = REDACTED
+        elif isinstance(v, (datetime, date)):
             v = v.isoformat()
         out[c.name] = v
     return out
@@ -157,7 +169,7 @@ class CRUDService(Generic[ModelT, CreateT, UpdateT]):  # noqa: UP046 — TypeVar
         )
         db.flush()
 
-    def history(self, db: Session, id: int) -> list[VersionHistory]:
+    def history(self, db: Session, id: int) -> builtins.list[VersionHistory]:
         stmt = (
             select(VersionHistory)
             .where(VersionHistory.entity == self.entity, VersionHistory.entity_id == id)

src/ypms/framework/pii_audit.py

@@ -50,3 +50,28 @@ def emit_pii_read(
         entity_id=entity_id,
         ip=ip,
     )
+
+
+def emit_pii_list_read(
+    db: Session,
+    *,
+    actor: User | None,
+    entity: str,
+    entity_ids: list[int],
+    ip: str | None = None,
+) -> AuditLog:
+    """Record one PII-read event for a list/search page.
+
+    One row per page (not per record) keeps the log proportional to requests
+    while still answering "who saw which records" — the returned ids are in
+    `after.ids`.
+    """
+    return audit.record(
+        db,
+        actor=actor,
+        action=PII_READ_ACTION,
+        entity=entity,
+        entity_id=None,
+        after={"ids": entity_ids, "count": len(entity_ids)},
+        ip=ip,
+    )

src/ypms/main.py

@@ -17,6 +17,7 @@ from ypms.db import engine
 from ypms.logging import configure_logging, get_logger
 from ypms.middleware import RequestIdMiddleware
 from ypms.settings import get_settings
+from ypms.youth.router import router as registrant_router
 
 
 def create_app() -> FastAPI:
@@ -42,6 +43,7 @@ def create_app() -> FastAPI:
     )
 
     app.include_router(auth_router)
+    app.include_router(registrant_router)
 
     @app.get("/healthz", tags=["system"])
     def healthz() -> dict[str, str]:

src/ypms/security/column_crypto.py

@@ -0,0 +1,72 @@
+"""App-layer column encryption for the highest-sensitivity PII fields.
+
+ARCHITECTURE.md §6.3: SSN, banking, tax ID, and immigration-doc references are
+encrypted at the application layer on top of Postgres TDE. Satisfies NIST
+SC-28 / SC-28(1) and source RFP `C.5.7.50` (defense-in-depth for SSN at rest).
+
+Scheme: AES-256-GCM (FIPS-approved AEAD). The 32-byte data key is derived with
+HKDF-SHA256 from `COLUMN_ENCRYPTION_KEY`. Each value gets a fresh 96-bit nonce.
+The column name is bound in as associated data, so ciphertext copied from one
+column into another fails to decrypt.
+
+Stored format: ``v1.<urlsafe-b64(nonce || ciphertext+tag)>``. The version
+prefix leaves room for key rotation — Phase 1 swaps the env-var key for Azure
+Key Vault and introduces ``v2`` without re-encrypting in place.
+
+Plaintext never leaves this module except as the return value of `decrypt`.
+Never log inputs or outputs.
+"""
+
+from __future__ import annotations
+
+import base64
+import os
+from functools import lru_cache
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+
+from ypms.settings import get_settings
+
+_VERSION = "v1"
+_NONCE_BYTES = 12
+_MIN_KEY_MATERIAL = 32
+_HKDF_INFO = b"ypms-column-encryption-v1"
+
+
+class ColumnCryptoError(Exception):
+    """Raised on malformed ciphertext, wrong key, or wrong column binding."""
+
+
+@lru_cache
+def _aead() -> AESGCM:
+    material = get_settings().column_encryption_key.encode()
+    if len(material) < _MIN_KEY_MATERIAL:
+        raise ValueError(
+            f"COLUMN_ENCRYPTION_KEY must be at least {_MIN_KEY_MATERIAL} bytes"
+        )
+    key = HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=_HKDF_INFO).derive(
+        material
+    )
+    return AESGCM(key)
+
+
+def encrypt(plaintext: str, *, column: str) -> str:
+    """Encrypt `plaintext` for storage in `column` (e.g. ``"registrants.ssn"``)."""
+    nonce = os.urandom(_NONCE_BYTES)
+    ct = _aead().encrypt(nonce, plaintext.encode(), column.encode())
+    return f"{_VERSION}.{base64.urlsafe_b64encode(nonce + ct).decode()}"
+
+
+def decrypt(token: str, *, column: str) -> str:
+    """Decrypt a value produced by `encrypt` for the same `column`."""
+    version, _, body = token.partition(".")
+    if version != _VERSION or not body:
+        raise ColumnCryptoError("unrecognized ciphertext format")
+    try:
+        raw = base64.urlsafe_b64decode(body)
+        pt = _aead().decrypt(raw[:_NONCE_BYTES], raw[_NONCE_BYTES:], column.encode())
+    except Exception as e:  # InvalidTag, binascii.Error, ValueError
+        raise ColumnCryptoError("decryption failed") from e
+    return pt.decode()

src/ypms/youth/models.py

@@ -0,0 +1,95 @@
+"""Registrant — the youth applying to and participating in OYP programs.
+
+ARCHITECTURE.md §6.1: Registrant is the spine; everything user-facing hangs off
+it. Phase 0 vertical slice (PHASE0_PLAN.md §4) — real fields, not a stub.
+
+PII handling:
+
+- SSN is write-only on the model. Assigning `registrant.ssn = "123456789"`
+  encrypts into `ssn_encrypted` (security/column_crypto.py) and denormalizes
+  `ssn_last4` for search and masked display. There is no plaintext getter;
+  the one sanctioned read path is `decrypt_ssn()` (SSA verification, payroll
+  export — Phase 1+). Source RFP `C.5.7.50`; NIST SC-28.
+- `ssn_encrypted` is listed in `__audit_redact__`, so audit before/after and
+  version-history snapshots record that the field exists and changed, never
+  its ciphertext.
+- `minor_flag` is derived from `dob` at read time, never stored — a stored
+  flag goes stale on the registrant's 18th birthday.
+"""
+
+from __future__ import annotations
+
+from datetime import date
+from typing import ClassVar
+
+from sqlalchemy import Date, ForeignKey, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from ypms.db import Base
+from ypms.framework.base import SoftDeleteMixin, TimestampMixin, VersionedMixin
+from ypms.security import column_crypto
+
+SSN_COLUMN = "registrants.ssn"
+MINOR_AGE = 18
+
+
+def age_on(dob: date, today: date) -> int:
+    return today.year - dob.year - ((today.month, today.day) < (dob.month, dob.day))
+
+
+class Registrant(Base, TimestampMixin, SoftDeleteMixin, VersionedMixin):
+    __tablename__ = "registrants"
+    __audit_redact__: ClassVar[tuple[str, ...]] = ("ssn_encrypted",)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+
+    # Portal account of the youth themself (YOUTH_USER). Set by the Phase 1
+    # account-link flow; never writable through the Phase 0 API. Drives the
+    # own-row scoping in youth/router.py.
+    user_id: Mapped[int | None] = mapped_column(
+        ForeignKey("users.id", ondelete="SET NULL"), index=True
+    )
+
+    first_name: Mapped[str] = mapped_column(String(100), nullable=False)
+    middle_name: Mapped[str | None] = mapped_column(String(100))
+    last_name: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
+    dob: Mapped[date] = mapped_column(Date, nullable=False, index=True)
+
+    ssn_encrypted: Mapped[str | None] = mapped_column(String(255))
+    ssn_last4: Mapped[str | None] = mapped_column(String(4), index=True)
+
+    email: Mapped[str | None] = mapped_column(String(255))
+    phone: Mapped[str | None] = mapped_column(String(32))
+
+    address_line1: Mapped[str | None] = mapped_column(String(255))
+    address_line2: Mapped[str | None] = mapped_column(String(255))
+    city: Mapped[str | None] = mapped_column(String(100))
+    state: Mapped[str | None] = mapped_column(String(2))
+    zip: Mapped[str | None] = mapped_column(String(10))
+
+    # ---- SSN: write-only ----
+
+    def _set_ssn(self, value: str | None) -> None:
+        if value is None:
+            self.ssn_encrypted = None
+            self.ssn_last4 = None
+            return
+        self.ssn_encrypted = column_crypto.encrypt(value, column=SSN_COLUMN)
+        self.ssn_last4 = value[-4:]
+
+    ssn = property(fset=_set_ssn, doc="Write-only. Use decrypt_ssn() for the sanctioned read path.")
+
+    def decrypt_ssn(self) -> str | None:
+        if self.ssn_encrypted is None:
+            return None
+        return column_crypto.decrypt(self.ssn_encrypted, column=SSN_COLUMN)
+
+    # ---- derived display fields ----
+
+    @property
+    def ssn_masked(self) -> str | None:
+        return f"XXX-XX-{self.ssn_last4}" if self.ssn_last4 else None
+
+    @property
+    def minor_flag(self) -> bool:
+        return age_on(self.dob, date.today()) < MINOR_AGE

src/ypms/youth/router.py

@@ -0,0 +1,187 @@
+"""Registrant REST surface — Phase 0 vertical slice (milestone 0.5).
+
+Hand-built on `CRUDService` rather than `framework.router.make_router` because
+Registrant needs three things the generic factory doesn't provide:
+
+1. **PII-read audit** on every read path (GET, list, history) via
+   `framework.pii_audit` — source RFP `C.5.6.9` #5, NIST AU-14.
+2. **Own-row scoping for portal roles.** The matrix grants YOUTH_USER and
+   PARENT_USER View/Edit on Registrant; RBAC alone would let them read anyone.
+   Portal-only callers may GET/PATCH only the row linked to their own account
+   (`Registrant.user_id == actor.id`); PARENT_USER sees nothing until the
+   Guardian link lands in Phase 1. Every other route is staff-only. Not a
+   generic scope layer — that's Phase 2+ (PHASE0_STATUS.md gotcha 7).
+3. **History gated on `VersionHistory`**, not `View` — snapshots carry the
+   full PII record over time.
+
+No bulk endpoints for Registrant in Phase 0.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import Session
+
+from ypms.db import get_db
+from ypms.framework.crud import CRUDService
+from ypms.framework.pii_audit import emit_pii_list_read, emit_pii_read
+from ypms.security.rbac import require_permission, user_role_codes
+from ypms.users.models import User
+from ypms.youth.models import Registrant
+from ypms.youth.schemas import RegistrantCreate, RegistrantRead, RegistrantUpdate
+
+ENTITY = "Registrant"
+PORTAL_ROLES = frozenset({"YOUTH_USER", "PARENT_USER", "HOST_USER"})
+
+crud: CRUDService[Registrant, RegistrantCreate, RegistrantUpdate] = CRUDService(
+    Registrant, ENTITY
+)
+router = APIRouter(prefix="/registrants", tags=["registrants"])
+
+
+def _serialize(obj: Registrant) -> dict[str, Any]:
+    return RegistrantRead.model_validate(obj).model_dump(mode="json")
+
+
+def _ip(request: Request) -> str | None:
+    return request.client.host if request.client else None
+
+
+def _is_portal_only(actor: User) -> bool:
+    codes = user_role_codes(actor)
+    return bool(codes) and codes <= PORTAL_ROLES
+
+
+def _require_staff(actor: User) -> None:
+    if _is_portal_only(actor):
+        raise HTTPException(
+            status.HTTP_403_FORBIDDEN,
+            detail={"error": "forbidden", "reason": "staff-only route", "entity": ENTITY},
+        )
+
+
+def _get_scoped(db: Session, id: int, actor: User) -> Registrant:
+    """Fetch a live registrant, 404 if missing or outside the caller's scope.
+
+    Out-of-scope returns 404, not 403, so portal users can't probe for ids.
+    """
+    try:
+        obj = crud.get(db, id)
+    except NoResultFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+    if _is_portal_only(actor):
+        owns = "YOUTH_USER" in user_role_codes(actor) and obj.user_id == actor.id
+        if not owns:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, f"{ENTITY} {id} not found")
+    return obj
+
+
+@router.get("")
+def list_registrants(
+    request: Request,
+    cursor: int | None = None,
+    limit: int = 50,
+    include_deleted: bool = False,
+    newest_first: bool = False,
+    db: Session = Depends(get_db),
+    actor: User = Depends(require_permission("List", ENTITY)),
+) -> dict[str, Any]:
+    _require_staff(actor)
+    limit = max(1, min(limit, 200))
+    page = crud.list(
+        db, cursor=cursor, limit=limit,
+        include_deleted=include_deleted, newest_first=newest_first,
+    )
+    emit_pii_list_read(
+        db, actor=actor, entity=ENTITY,
+        entity_ids=[r.id for r in page.items], ip=_ip(request),
+    )
+    db.commit()
+    return page.to_dict(_serialize)
+
+
+@router.get("/{id}")
+def get_registrant(
+    id: int,
+    request: Request,
+    db: Session = Depends(get_db),
+    actor: User = Depends(require_permission("View", ENTITY)),
+) -> dict[str, Any]:
+    obj = _get_scoped(db, id, actor)
+    emit_pii_read(db, actor=actor, entity=ENTITY, entity_id=obj.id, ip=_ip(request))
+    db.commit()
+    return _serialize(obj)
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+def create_registrant(
+    payload: RegistrantCreate,
+    db: Session = Depends(get_db),
+    actor: User = Depends(require_permission("Create", ENTITY)),
+) -> dict[str, Any]:
+    _require_staff(actor)
+    obj = crud.create(db, payload, actor=actor)
+    db.commit()
+    return _serialize(obj)
+
+
+@router.patch("/{id}")
+def update_registrant(
+    id: int,
+    payload: RegistrantUpdate,
+    db: Session = Depends(get_db),
+    actor: User = Depends(require_permission("Edit", ENTITY)),
+) -> dict[str, Any]:
+    _get_scoped(db, id, actor)
+    obj = crud.update(db, id, payload, actor=actor)
+    db.commit()
+    return _serialize(obj)
+
+
+@router.delete("/{id}")
+def delete_registrant(
+    id: int,
+    db: Session = Depends(get_db),
+    actor: User = Depends(require_permission("Delete", ENTITY)),
+) -> dict[str, Any]:
+    _require_staff(actor)
+    try:
+        obj = crud.soft_delete(db, id, actor=actor)
+    except NoResultFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+    db.commit()
+    return _serialize(obj)
+
+
+@router.post("/{id}/restore")
+def restore_registrant(
+    id: int,
+    db: Session = Depends(get_db),
+    actor: User = Depends(require_permission("Edit", ENTITY)),
+) -> dict[str, Any]:
+    _require_staff(actor)
+    try:
+        obj = crud.restore(db, id, actor=actor)
+    except NoResultFound as e:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, str(e)) from e
+    db.commit()
+    return _serialize(obj)
+
+
+@router.get("/{id}/history")
+def registrant_history(
+    id: int,
+    request: Request,
+    db: Session = Depends(get_db),
+    actor: User = Depends(require_permission("VersionHistory", ENTITY)),
+) -> list[dict[str, Any]]:
+    _require_staff(actor)
+    rows = crud.history(db, id)
+    if not rows:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"{ENTITY} {id} not found")
+    emit_pii_read(db, actor=actor, entity=ENTITY, entity_id=id, ip=_ip(request))
+    db.commit()
+    return [{"version": r.version, "ts": r.ts.isoformat(), "snapshot": r.snapshot} for r in rows]

src/ypms/youth/schemas.py

@@ -0,0 +1,108 @@
+"""Registrant request/response schemas.
+
+The read schema has no `ssn` field — only `ssn_masked` (``XXX-XX-1234``).
+Masking happens here in the serializer, not the UI, so every API consumer sees
+the masked form (ARCHITECTURE.md §6.3; source RFP `C.5.7.50`).
+
+`user_id` is deliberately absent from the write schemas: linking a registrant
+to a portal account is a Phase 1 identity flow, not a PATCH.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import date, datetime
+
+from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
+
+_ZIP = r"^\d{5}(-\d{4})?$"
+
+
+def normalize_ssn(value: str | None) -> str | None:
+    """Accept ``123-45-6789`` or ``123456789``; return 9 digits.
+
+    Rejects numbers the SSA never issues: area 000, 666, or 900–999; group 00;
+    serial 0000. Error messages never echo the input.
+    """
+    if value is None:
+        return None
+    digits = re.sub(r"[\s-]", "", value)
+    if not re.fullmatch(r"\d{9}", digits):
+        raise ValueError("SSN must be 9 digits")
+    area, group, serial = digits[:3], digits[3:5], digits[5:]
+    if area in {"000", "666"} or area >= "900" or group == "00" or serial == "0000":
+        raise ValueError("SSN is not a valid issued number")
+    return digits
+
+
+def _not_future(value: date | None) -> date | None:
+    if value is not None and value > date.today():
+        raise ValueError("dob cannot be in the future")
+    return value
+
+
+class RegistrantCreate(BaseModel):
+    first_name: str = Field(min_length=1, max_length=100)
+    middle_name: str | None = Field(default=None, max_length=100)
+    last_name: str = Field(min_length=1, max_length=100)
+    dob: date
+    ssn: str | None = None
+    email: EmailStr | None = None
+    phone: str | None = Field(default=None, max_length=32)
+    address_line1: str | None = Field(default=None, max_length=255)
+    address_line2: str | None = Field(default=None, max_length=255)
+    city: str | None = Field(default=None, max_length=100)
+    state: str | None = Field(default=None, min_length=2, max_length=2)
+    zip: str | None = Field(default=None, pattern=_ZIP)
+
+    _ssn = field_validator("ssn")(normalize_ssn)
+    _dob = field_validator("dob")(_not_future)
+
+
+class RegistrantUpdate(BaseModel):
+    first_name: str | None = Field(default=None, min_length=1, max_length=100)
+    middle_name: str | None = Field(default=None, max_length=100)
+    last_name: str | None = Field(default=None, min_length=1, max_length=100)
+    dob: date | None = None
+    ssn: str | None = None
+    email: EmailStr | None = None
+    phone: str | None = Field(default=None, max_length=32)
+    address_line1: str | None = Field(default=None, max_length=255)
+    address_line2: str | None = Field(default=None, max_length=255)
+    city: str | None = Field(default=None, max_length=100)
+    state: str | None = Field(default=None, min_length=2, max_length=2)
+    zip: str | None = Field(default=None, pattern=_ZIP)
+
+    _ssn = field_validator("ssn")(normalize_ssn)
+    _dob = field_validator("dob")(_not_future)
+
+    @field_validator("first_name", "last_name", "dob")
+    @classmethod
+    def _required_not_null(cls, v: object) -> object:
+        # Omit to leave unchanged; explicit null would violate NOT NULL.
+        if v is None:
+            raise ValueError("cannot be null")
+        return v
+
+
+class RegistrantRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    first_name: str
+    middle_name: str | None
+    last_name: str
+    dob: date
+    ssn_masked: str | None
+    minor_flag: bool
+    email: str | None
+    phone: str | None
+    address_line1: str | None
+    address_line2: str | None
+    city: str | None
+    state: str | None
+    zip: str | None
+    version: int
+    created_at: datetime
+    updated_at: datetime
+    deleted_at: datetime | None

tests/conftest.py

@@ -39,6 +39,7 @@ def db() -> Iterator[Session]:
     import ypms.framework.version_history  # noqa: F401
     import ypms.security.models  # noqa: F401
     import ypms.users.models  # noqa: F401
+    import ypms.youth.models  # noqa: F401
     from ypms.db import Base
 
     # StaticPool + check_same_thread=False share the same in-memory DB across

tests/youth/test_registrant_api.py

@@ -0,0 +1,367 @@
+"""Registrant vertical slice, end-to-end through the FastAPI TestClient.
+
+Runs against the real seeded roles + permissions matrix (not hand-built
+grants), so these tests also pin the matrix's Registrant rows.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterator
+from datetime import date
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from ypms.audit.models import AuditLog
+from ypms.auth.security import create_access_token, hash_password
+from ypms.db import get_db
+from ypms.framework.crud import REDACTED
+from ypms.framework.pii_audit import PII_READ_ACTION
+from ypms.framework.version_history import VersionHistory
+from ypms.main import create_app
+from ypms.seed.permissions import seed_permissions
+from ypms.seed.run import seed_roles
+from ypms.users.models import ASSURANCE_STANDARD, Role, User, UserRole
+from ypms.youth.models import Registrant
+
+SSN = "123-45-6789"
+PAYLOAD: dict[str, Any] = {
+    "first_name": "Jordan",
+    "last_name": "Rivera",
+    "dob": "2010-06-15",
+    "ssn": SSN,
+    "email": "jordan@example.org",
+    "city": "Washington",
+    "state": "DC",
+    "zip": "20001",
+}
+
+Headers = dict[str, str]
+
+
+@pytest.fixture
+def roles(db: Session) -> dict[str, Role]:
+    r = seed_roles(db)
+    seed_permissions(db, r)
+    db.flush()
+    return r
+
+
+@pytest.fixture
+def make_user(db: Session, roles: dict[str, Role]) -> Callable[..., tuple[User, Headers]]:
+    def _make(username: str, *role_codes: str) -> tuple[User, Headers]:
+        user = User(
+            username=username,
+            password_hash=hash_password("unused-in-these-tests"),
+            is_active=True,
+            assurance_level=ASSURANCE_STANDARD,
+        )
+        db.add(user)
+        db.flush()
+        for code in role_codes:
+            db.add(UserRole(user_id=user.id, role_id=roles[code].id))
+        db.flush()
+        db.refresh(user)
+        token = create_access_token(
+            sub=username, uid=user.id, roles=list(role_codes),
+            assurance_level=ASSURANCE_STANDARD, mfa_verified=False,
+        )
+        return user, {"Authorization": f"Bearer {token}"}
+
+    return _make
+
+
+@pytest.fixture
+def staff(make_user: Callable[..., tuple[User, Headers]]) -> tuple[User, Headers]:
+    return make_user("oyp.staff", "OYP_STAFF")
+
+
+@pytest.fixture
+def api(db: Session) -> Iterator[TestClient]:
+    app = create_app()
+
+    def _override() -> Iterator[Session]:
+        yield db
+
+    app.dependency_overrides[get_db] = _override
+    with TestClient(app) as c:
+        yield c
+    app.dependency_overrides.clear()
+
+
+def _create(api: TestClient, headers: Headers, **overrides: Any) -> dict[str, Any]:
+    r = api.post("/registrants", json={**PAYLOAD, **overrides}, headers=headers)
+    assert r.status_code == 201, r.text
+    body: dict[str, Any] = r.json()
+    return body
+
+
+def _audit(db: Session, action: str) -> list[AuditLog]:
+    stmt = select(AuditLog).where(AuditLog.entity == "Registrant", AuditLog.action == action)
+    return list(db.execute(stmt).scalars())
+
+
+# ---------------- create + masking ----------------
+
+
+def test_create_masks_ssn_and_never_returns_it(
+    api: TestClient, db: Session, staff: tuple[User, Headers]
+) -> None:
+    body = _create(api, staff[1])
+    assert body["ssn_masked"] == "XXX-XX-6789"
+    assert "ssn" not in body
+    assert "ssn_encrypted" not in body
+    assert "ssn_last4" not in body
+    assert "6789" not in str({k: v for k, v in body.items() if k != "ssn_masked"})
+
+    row = db.get(Registrant, body["id"])
+    assert row is not None
+    assert row.ssn_encrypted and "123456789" not in row.ssn_encrypted
+    assert row.decrypt_ssn() == "123456789"
+
+
+def test_create_emits_redacted_audit_and_version_snapshot(
+    api: TestClient, db: Session, staff: tuple[User, Headers]
+) -> None:
+    body = _create(api, staff[1])
+    (row,) = _audit(db, "create")
+    assert row.actor_username == "oyp.staff"
+    assert row.entity_id == body["id"]
+    assert row.after is not None
+    assert row.after["ssn_encrypted"] == REDACTED
+    assert row.after["ssn_last4"] == "6789"
+    assert row.after["dob"] == "2010-06-15"
+
+    (snap,) = db.execute(select(VersionHistory)).scalars()
+    assert snap.snapshot["ssn_encrypted"] == REDACTED
+
+
+def test_minor_flag_derived_from_dob(api: TestClient, staff: tuple[User, Headers]) -> None:
+    minor = _create(api, staff[1], dob=date.today().replace(year=date.today().year - 16).isoformat())
+    adult = _create(api, staff[1], dob="1990-01-01", ssn=None)
+    assert minor["minor_flag"] is True
+    assert adult["minor_flag"] is False
+    assert adult["ssn_masked"] is None
+
+
+@pytest.mark.parametrize(
+    "bad",
+    [
+        {"ssn": "12345678"},
+        {"ssn": "000-12-3456"},
+        {"ssn": "666-12-3456"},
+        {"ssn": "900-12-3456"},
+        {"ssn": "123-00-4567"},
+        {"ssn": "123-45-0000"},
+        {"dob": "2999-01-01"},
+        {"zip": "2000"},
+        {"email": "not-an-email"},
+    ],
+)
+def test_create_validation(api: TestClient, staff: tuple[User, Headers], bad: dict[str, str]) -> None:
+    r = api.post("/registrants", json={**PAYLOAD, **bad}, headers=staff[1])
+    assert r.status_code == 422
+
+
+# ---------------- reads emit PII audit ----------------
+
+
+def test_get_emits_pii_read(api: TestClient, db: Session, staff: tuple[User, Headers]) -> None:
+    rid = _create(api, staff[1])["id"]
+    r = api.get(f"/registrants/{rid}", headers=staff[1])
+    assert r.status_code == 200
+    assert r.json()["ssn_masked"] == "XXX-XX-6789"
+    (row,) = _audit(db, PII_READ_ACTION)
+    assert row.entity_id == rid
+    assert row.actor_username == "oyp.staff"
+    assert row.ip is not None
+
+
+def test_list_emits_one_pii_read_with_ids(
+    api: TestClient, db: Session, staff: tuple[User, Headers]
+) -> None:
+    ids = [_create(api, staff[1])["id"] for _ in range(3)]
+    r = api.get("/registrants?limit=2", headers=staff[1])
+    assert r.status_code == 200
+    page = r.json()
+    assert [i["id"] for i in page["items"]] == ids[:2]
+    assert page["next_cursor"] == ids[1]
+    (row,) = _audit(db, PII_READ_ACTION)
+    assert row.entity_id is None
+    assert row.after == {"ids": ids[:2], "count": 2}
+
+
+def test_get_missing_is_404_and_not_audited(
+    api: TestClient, db: Session, staff: tuple[User, Headers]
+) -> None:
+    assert api.get("/registrants/999", headers=staff[1]).status_code == 404
+    assert _audit(db, PII_READ_ACTION) == []
+
+
+def test_unauthenticated_is_401(api: TestClient) -> None:
+    assert api.get("/registrants").status_code == 401
+
+
+# ---------------- update / delete / restore / history ----------------
+
+
+def test_update_bumps_version_and_audits(
+    api: TestClient, db: Session, staff: tuple[User, Headers]
+) -> None:
+    rid = _create(api, staff[1])["id"]
+    r = api.patch(f"/registrants/{rid}", json={"ssn": "234-56-4321", "city": "DC"}, headers=staff[1])
+    assert r.status_code == 200
+    body = r.json()
+    assert body["version"] == 2
+    assert body["ssn_masked"] == "XXX-XX-4321"
+    (row,) = _audit(db, "update")
+    assert row.before is not None and row.after is not None
+    assert row.before["ssn_last4"] == "6789"
+    assert row.after["ssn_last4"] == "4321"
+    assert row.before["ssn_encrypted"] == row.after["ssn_encrypted"] == REDACTED
+
+
+def test_update_rejects_null_on_required_field(
+    api: TestClient, staff: tuple[User, Headers]
+) -> None:
+    rid = _create(api, staff[1])["id"]
+    r = api.patch(f"/registrants/{rid}", json={"last_name": None}, headers=staff[1])
+    assert r.status_code == 422
+
+
+def test_update_cannot_set_user_id(
+    api: TestClient, db: Session, staff: tuple[User, Headers]
+) -> None:
+    rid = _create(api, staff[1])["id"]
+    api.patch(f"/registrants/{rid}", json={"user_id": staff[0].id}, headers=staff[1])
+    row = db.get(Registrant, rid)
+    assert row is not None and row.user_id is None
+
+
+def test_delete_requires_delete_permission_then_restore(
+    api: TestClient,
+    db: Session,
+    staff: tuple[User, Headers],
+    make_user: Callable[..., tuple[User, Headers]],
+) -> None:
+    rid = _create(api, staff[1])["id"]
+    # OYP_STAFF has no Delete on Registrant in the matrix.
+    assert api.delete(f"/registrants/{rid}", headers=staff[1]).status_code == 403
+
+    _, admin = make_user("admin", "DOES_SYSTEM_ADMIN")
+    r = api.delete(f"/registrants/{rid}", headers=admin)
+    assert r.status_code == 200 and r.json()["deleted_at"] is not None
+    assert api.get(f"/registrants/{rid}", headers=staff[1]).status_code == 404
+
+    r = api.post(f"/registrants/{rid}/restore", headers=staff[1])
+    assert r.status_code == 200 and r.json()["deleted_at"] is None
+    assert [a.action for a in db.execute(select(AuditLog).order_by(AuditLog.id)).scalars()][-2:] == [
+        "delete",
+        "restore",
+    ]
+
+
+def test_delete_missing_is_404(
+    api: TestClient, make_user: Callable[..., tuple[User, Headers]]
+) -> None:
+    _, admin = make_user("admin", "DOES_SYSTEM_ADMIN")
+    assert api.delete("/registrants/999", headers=admin).status_code == 404
+    assert api.post("/registrants/999/restore", headers=admin).status_code == 404
+
+
+def test_history_requires_version_history_permission(
+    api: TestClient,
+    db: Session,
+    staff: tuple[User, Headers],
+    make_user: Callable[..., tuple[User, Headers]],
+) -> None:
+    rid = _create(api, staff[1])["id"]
+    api.patch(f"/registrants/{rid}", json={"city": "DC"}, headers=staff[1])
+
+    # PAYROLL_ADMIN has View but not VersionHistory.
+    _, payroll = make_user("payroll", "PAYROLL_ADMIN")
+    assert api.get(f"/registrants/{rid}", headers=payroll).status_code == 200
+    assert api.get(f"/registrants/{rid}/history", headers=payroll).status_code == 403
+
+    r = api.get(f"/registrants/{rid}/history", headers=staff[1])
+    assert r.status_code == 200
+    assert [h["version"] for h in r.json()] == [1, 2]
+    assert all(h["snapshot"]["ssn_encrypted"] == REDACTED for h in r.json())
+    assert api.get("/registrants/999/history", headers=staff[1]).status_code == 404
+    reads = [a for a in _audit(db, PII_READ_ACTION) if a.actor_username == "oyp.staff"]
+    assert len(reads) == 1
+
+
+# ---------------- portal-role row scoping ----------------
+
+
+def test_youth_can_read_and_edit_only_own_row(
+    api: TestClient,
+    db: Session,
+    staff: tuple[User, Headers],
+    make_user: Callable[..., tuple[User, Headers]],
+) -> None:
+    youth, yh = make_user("youth", "YOUTH_USER")
+    mine = _create(api, staff[1])["id"]
+    theirs = _create(api, staff[1])["id"]
+    row = db.get(Registrant, mine)
+    assert row is not None
+    row.user_id = youth.id
+    db.flush()
+
+    assert api.get(f"/registrants/{mine}", headers=yh).status_code == 200
+    assert api.get(f"/registrants/{theirs}", headers=yh).status_code == 404
+    assert api.patch(f"/registrants/{mine}", json={"phone": "202-555-0100"}, headers=yh).status_code == 200
+    assert api.patch(f"/registrants/{theirs}", json={"phone": "x"}, headers=yh).status_code == 404
+
+
+def test_youth_blocked_from_staff_routes(
+    api: TestClient,
+    staff: tuple[User, Headers],
+    make_user: Callable[..., tuple[User, Headers]],
+) -> None:
+    _, yh = make_user("youth", "YOUTH_USER")
+    rid = _create(api, staff[1])["id"]
+    assert api.get("/registrants", headers=yh).status_code == 403
+    assert api.post("/registrants", json=PAYLOAD, headers=yh).status_code == 403
+    assert api.post(f"/registrants/{rid}/restore", headers=yh).status_code == 403
+    assert api.get(f"/registrants/{rid}/history", headers=yh).status_code == 403
+
+
+def test_parent_sees_nothing_until_guardian_link(
+    api: TestClient,
+    db: Session,
+    staff: tuple[User, Headers],
+    make_user: Callable[..., tuple[User, Headers]],
+) -> None:
+    parent, ph = make_user("parent", "PARENT_USER")
+    rid = _create(api, staff[1])["id"]
+    row = db.get(Registrant, rid)
+    assert row is not None
+    row.user_id = parent.id  # even a (bogus) direct link grants nothing
+    db.flush()
+    assert api.get(f"/registrants/{rid}", headers=ph).status_code == 404
+
+
+def test_staff_role_plus_portal_role_is_staff(
+    api: TestClient,
+    staff: tuple[User, Headers],
+    make_user: Callable[..., tuple[User, Headers]],
+) -> None:
+    _, h = make_user("dual", "OYP_STAFF", "YOUTH_USER")
+    rid = _create(api, staff[1])["id"]
+    assert api.get(f"/registrants/{rid}", headers=h).status_code == 200
+    assert api.get("/registrants", headers=h).status_code == 200
+
+
+def test_host_user_denied(
+    api: TestClient,
+    staff: tuple[User, Headers],
+    make_user: Callable[..., tuple[User, Headers]],
+) -> None:
+    _, hh = make_user("host", "HOST_USER")
+    rid = _create(api, staff[1])["id"]
+    assert api.get(f"/registrants/{rid}", headers=hh).status_code == 403

tests/youth/test_registrant_unit.py

@@ -0,0 +1,99 @@
+"""Unit tests for column crypto, SSN normalization, and Registrant model rules."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from datetime import date
+
+import pytest
+
+from ypms.security import column_crypto
+from ypms.security.column_crypto import ColumnCryptoError, decrypt, encrypt
+from ypms.settings import get_settings
+from ypms.youth.models import Registrant, age_on
+from ypms.youth.schemas import normalize_ssn
+
+# ---------------- column crypto ----------------
+
+
+def test_roundtrip_and_fresh_nonce() -> None:
+    a = encrypt("123456789", column="registrants.ssn")
+    b = encrypt("123456789", column="registrants.ssn")
+    assert a != b
+    assert a.startswith("v1.")
+    assert decrypt(a, column="registrants.ssn") == "123456789"
+
+
+def test_ciphertext_is_bound_to_column() -> None:
+    token = encrypt("123456789", column="registrants.ssn")
+    with pytest.raises(ColumnCryptoError):
+        decrypt(token, column="registrants.bank_account")
+
+
+@pytest.mark.parametrize("token", ["", "v1.", "v2.abcd", "v1.not-base64!!", "v1.AAAA"])
+def test_malformed_ciphertext_rejected(token: str) -> None:
+    with pytest.raises(ColumnCryptoError):
+        decrypt(token, column="registrants.ssn")
+
+
+@pytest.fixture
+def short_key(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    monkeypatch.setenv("COLUMN_ENCRYPTION_KEY", "too-short")
+    get_settings.cache_clear()
+    column_crypto._aead.cache_clear()
+    yield
+    monkeypatch.undo()
+    get_settings.cache_clear()
+    column_crypto._aead.cache_clear()
+
+
+@pytest.mark.usefixtures("short_key")
+def test_short_key_material_refused() -> None:
+    with pytest.raises(ValueError, match="at least 32 bytes"):
+        encrypt("x", column="c")
+
+
+# ---------------- SSN normalization ----------------
+
+
+@pytest.mark.parametrize("raw", ["123-45-6789", "123456789", " 123 45 6789 "])
+def test_normalize_ssn_accepts_common_forms(raw: str) -> None:
+    assert normalize_ssn(raw) == "123456789"
+
+
+def test_normalize_ssn_error_does_not_echo_input() -> None:
+    with pytest.raises(ValueError) as e:
+        normalize_ssn("123-45-678")
+    assert "678" not in str(e.value)
+
+
+# ---------------- model ----------------
+
+
+def test_ssn_is_write_only() -> None:
+    r = Registrant(first_name="A", last_name="B", dob=date(2010, 1, 1), ssn="123456789")
+    with pytest.raises(AttributeError):
+        _ = r.ssn  # type: ignore[misc]
+    assert r.ssn_last4 == "6789"
+    assert r.decrypt_ssn() == "123456789"
+
+
+def test_clearing_ssn_clears_both_columns() -> None:
+    r = Registrant(first_name="A", last_name="B", dob=date(2010, 1, 1), ssn="123456789")
+    r.ssn = None
+    assert r.ssn_encrypted is None and r.ssn_last4 is None
+    assert r.ssn_masked is None
+    assert r.decrypt_ssn() is None
+
+
+@pytest.mark.parametrize(
+    ("dob", "today", "age"),
+    [
+        (date(2008, 10, 1), date(2026, 9, 30), 17),  # day before 18th birthday
+        (date(2008, 9, 30), date(2026, 9, 30), 18),  # on 18th birthday
+        (date(2008, 2, 29), date(2026, 2, 28), 17),  # leap-day birthday
+        (date(2008, 2, 29), date(2026, 3, 1), 18),
+    ],
+)
+def test_age_on_boundaries(dob: date, today: date, age: int) -> None:
+    assert age_on(dob, today) == age