Phase 0 milestone 0.7: retention scaffold

Every mapped table resolves a retention policy; a report-only job runs
over all of them and logs "N records to purge". ARCH §6.4; NIST AU-11,
SI-12; source RFP C.5.7.42.

- framework/retention.py: RetentionPolicy (data class, disposal, period,
  anchor column, cadence, approved, authority), run_retention(), and a
  `python -m ypms.framework.retention` entrypoint. Phase 0 never mutates
  data: interim-period rows are counted (interim_eligible), approved-period
  rows are counted (to_purge), nothing is purged. One retention_run audit
  row per run.
- retention_classes.py: PIA §9 data classes at their interim periods, all
  approved=False pending YPMS-POAM-054. Classes without a PIA basis
  (hosts, operational logs, accounts, consent-derived) are unset and logged
  by name with their governing POA&M item. Vendored framework tables are
  mapped by table name so vendored files stay unchanged.
- stub_models.py + migration 0004: id + mixin models for the 26 stub
  entities so each can carry a policy. Migration uses a literal table list.
- ypms/models.py: single import-all module used by Alembic env, the job,
  and the test fixtures.
- Drift guard: test upgrades a scratch SQLite DB to head and compares to
  Base.metadata. Found pre-existing drift in 0002 (named unique
  constraints duplicating the unique indexes on roles.code, users.username,
  users.email); allow-listed, follow-up noted in PHASE0_STATUS.

156 tests passing; ruff + mypy clean.

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

migrations/env.py

@@ -1,7 +1,7 @@
 """Alembic environment.
 
 Model modules are imported here so `Base.metadata` is populated before Alembic
-compares against it. Add every new model module to the import block below.
+compares against it. New model modules go in `src/ypms/models.py`, not here.
 """
 
 from __future__ import annotations
@@ -14,12 +14,8 @@ from sqlalchemy import engine_from_config, pool
 from ypms.db import Base
 from ypms.settings import get_settings
 
-# ---- Import model modules here so Base.metadata is populated ----
-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
+# ---- Populate Base.metadata ----
+import ypms.models  # noqa: F401 — every model module; see src/ypms/models.py
 
 config = context.config
 if config.config_file_name is not None:

migrations/versions/0004_stub_entity_tables.py

@@ -0,0 +1,73 @@
+"""Stub entity tables — id + framework mixin columns only.
+
+Revision ID: 0004_stub_entity_tables
+Revises: 0003_registrants
+Create Date: 2026-10-01
+
+Milestone 0.7. One table per Phase 0 stub entity (`src/ypms/stub_models.py`)
+so every entity can carry a retention policy. Domain phases add columns in new
+revisions; table names are stable.
+
+The table list is a literal on purpose — migrations must not import app code,
+or a later rename would silently rewrite history.
+
+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 = "0004_stub_entity_tables"
+down_revision: str | Sequence[str] | None = "0003_registrants"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+TABLES: tuple[str, ...] = (
+    "guardians",
+    "consent_records",
+    "hosts",
+    "host_contacts",
+    "worksites",
+    "positions",
+    "job_postings",
+    "programs",
+    "cohorts",
+    "eligibility_rules",
+    "applications",
+    "eligibility_decisions",
+    "certifications",
+    "orientations",
+    "enrollments",
+    "time_entries",
+    "timesheets",
+    "payroll_batches",
+    "tax_documents",
+    "transportation_vouchers",
+    "attachments",
+    "notes",
+    "integration_events",
+    "notifications",
+    "education_records",
+    "foster_indicators",
+)
+
+
+def upgrade() -> None:
+    for table in TABLES:
+        op.create_table(
+            table,
+            sa.Column("id", sa.Integer(), primary_key=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")),
+        )
+
+
+def downgrade() -> None:
+    for table in reversed(TABLES):
+        op.drop_table(table)

src/ypms/framework/retention.py

@@ -0,0 +1,196 @@
+"""Retention scaffold — policy type + nightly job (milestone 0.7; YPMS addition).
+
+ARCHITECTURE.md §6.4: every entity declares a retention policy; a nightly job
+scans for records past their envelope and applies the policy. NIST AU-11,
+SI-12; source RFP `C.5.7.42`, `C.5.21.34`.
+
+**Phase 0 is report-only.** The job never modifies data. For each mapped
+table it resolves the policy and logs one of:
+
+- `retain`       — kept by design (configuration tables);
+- `unset`        — no period yet; named so the gap stays visible;
+- `interim`      — PIA §9 interim period, not yet signed off. Rows past the
+                   interim envelope are *counted* (`interim_eligible`) but
+                   never purged;
+- `approved`     — signed-off period. Rows past it are counted as
+                   `to_purge`. Disposal execution lands in Phase 5 with the
+                   ingest pipelines; until then the job still only reports.
+
+Every run writes one `retention_run` audit row with the summary.
+
+Policy catalog: `ypms/retention_classes.py`. Entity models declare
+`retention_policy: ClassVar[RetentionPolicy]`; vendored framework tables are
+mapped by table name in `retention_classes.FRAMEWORK_TABLE_POLICIES` so the
+vendored files stay unchanged.
+
+Run: ``python -m ypms.framework.retention`` (nightly schedule is wired at
+deployment — Azure Container Apps job / cron; see ARCH §9).
+"""
+
+from __future__ import annotations
+
+import enum
+import json
+from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from ypms.audit import service as audit
+from ypms.logging import get_logger
+
+RETENTION_RUN_ACTION = "retention_run"
+RETENTION_ENTITY = "Retention"
+
+
+class Disposal(enum.StrEnum):
+    TOMBSTONE = "tombstone"            # soft-delete + scrub, keep the row shell
+    HARD_DELETE = "hard_delete"        # delete with audit
+    ARCHIVE = "archive"                # move to cold storage, then delete online copy
+    ARCHIVE_SCRUB = "archive_scrub"    # archive, scrub PII from the online copy
+    RETAIN = "retain"                  # never purged (configuration)
+
+
+class Cadence(enum.StrEnum):
+    NIGHTLY = "nightly"
+    ANNUAL = "annual"
+
+
+@dataclass(frozen=True)
+class RetentionPolicy:
+    data_class: str
+    disposal: Disposal
+    period_days: int | None
+    anchor: str = "created_at"
+    cadence: Cadence = Cadence.NIGHTLY
+    approved: bool = False
+    authority: str = ""
+
+    @property
+    def status(self) -> str:
+        if self.disposal is Disposal.RETAIN:
+            return "retain"
+        if self.period_days is None:
+            return "unset"
+        return "approved" if self.approved else "interim"
+
+
+@dataclass
+class TableResult:
+    table: str
+    data_class: str
+    status: str
+    period_days: int | None
+    authority: str
+    interim_eligible: int = 0
+    to_purge: int = 0
+
+
+@dataclass
+class RetentionReport:
+    ran_at: str
+    dry_run: bool = True
+    tables: list[TableResult] = field(default_factory=list)
+
+    @property
+    def to_purge(self) -> int:
+        return sum(t.to_purge for t in self.tables)
+
+    @property
+    def interim_eligible(self) -> int:
+        return sum(t.interim_eligible for t in self.tables)
+
+    def tables_with(self, status: str) -> list[str]:
+        return sorted(t.table for t in self.tables if t.status == status)
+
+    def summary(self) -> dict[str, Any]:
+        return {
+            "ran_at": self.ran_at,
+            "dry_run": self.dry_run,
+            "to_purge": self.to_purge,
+            "interim_eligible": self.interim_eligible,
+            "unset": self.tables_with("unset"),
+            "tables": len(self.tables),
+        }
+
+
+def resolve_policy(model: type[Any]) -> RetentionPolicy | None:
+    from ypms.retention_classes import FRAMEWORK_TABLE_POLICIES
+
+    policy: RetentionPolicy | None = getattr(model, "retention_policy", None)
+    if policy is not None:
+        return policy
+    return FRAMEWORK_TABLE_POLICIES.get(model.__tablename__)
+
+
+def mapped_models() -> list[type[Any]]:
+    import ypms.models  # noqa: F401 — registers every model on Base
+    from ypms.db import Base
+
+    return sorted((m.class_ for m in Base.registry.mappers), key=lambda c: c.__tablename__)
+
+
+def _count_past(db: Session, model: type[Any], policy: RetentionPolicy, now: datetime) -> int:
+    assert policy.period_days is not None
+    anchor = getattr(model, policy.anchor)
+    cutoff = now - timedelta(days=policy.period_days)
+    stmt = select(func.count()).select_from(model).where(anchor < cutoff)
+    return int(db.execute(stmt).scalar_one())
+
+
+def run_retention(db: Session, *, now: datetime | None = None) -> RetentionReport:
+    """Report-only retention pass over every mapped table. Never mutates data
+    other than the single `retention_run` audit row; caller commits."""
+    log = get_logger(__name__)
+    now = now or datetime.now(UTC)
+    report = RetentionReport(ran_at=now.isoformat())
+
+    for model in mapped_models():
+        policy = resolve_policy(model)
+        if policy is None:
+            raise LookupError(f"no retention policy for table {model.__tablename__}")
+        result = TableResult(
+            table=model.__tablename__, data_class=policy.data_class,
+            status=policy.status, period_days=policy.period_days,
+            authority=policy.authority,
+        )
+        if result.status in {"interim", "approved"}:
+            n = _count_past(db, model, policy, now)
+            if result.status == "approved":
+                result.to_purge = n
+            else:
+                result.interim_eligible = n
+        report.tables.append(result)
+        log.info("retention.table", **asdict(result))
+
+    for result in report.tables:
+        if result.status == "unset":
+            log.warning("retention.policy_unset", table=result.table, authority=result.authority)
+    log.info(
+        "retention.run",
+        message=f"{report.to_purge} records to purge",
+        **report.summary(),
+    )
+    audit.record(
+        db, actor=None, action=RETENTION_RUN_ACTION, entity=RETENTION_ENTITY,
+        after=report.summary(),
+    )
+    return report
+
+
+def main() -> int:
+    from ypms.db import session_factory
+    from ypms.logging import configure_logging
+
+    configure_logging()
+    with session_factory()() as db:
+        report = run_retention(db)
+        db.commit()
+    print(json.dumps(report.summary()))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/ypms/models.py

@@ -0,0 +1,12 @@
+"""Import every model module so `Base.metadata` / `Base.registry` is complete.
+
+Used by Alembic (`migrations/env.py`), the retention job, and the test
+fixtures. Add new model modules here — one place, not three.
+"""
+
+import ypms.audit.models  # noqa: F401
+import ypms.framework.version_history  # noqa: F401
+import ypms.security.models  # noqa: F401
+import ypms.stub_models  # noqa: F401
+import ypms.users.models  # noqa: F401
+import ypms.youth.models  # noqa: F401

src/ypms/retention_classes.py

@@ -0,0 +1,132 @@
+"""Retention policy catalog — PIA §9 data classes.
+
+Every period here is the PIA §9 **interim** posture and is `approved=False`
+until the DOES Records Officer signs the retention policy (`YPMS-POAM-054`).
+Flipping `approved=True` is the sign-off act; cite the signed schedule in
+`authority` when you do.
+
+Classes with no PIA §9 basis have `period_days=None` and report as `unset`
+on every run — deliberately loud rather than a guessed default.
+
+Anchors: "after last activity" uses `updated_at`; ingest-scoped classes use
+`created_at` (ingest date); framework tables use their own timestamp column.
+"""
+
+from __future__ import annotations
+
+from ypms.framework.retention import Cadence, Disposal, RetentionPolicy
+
+_PIA = "PIA §9 interim (YPMS-POAM-054)"
+_YEAR = 365
+
+# Program participation records: application, eligibility decision,
+# enrollment, hours worked, payment record — 7 years after last activity.
+PARTICIPATION = RetentionPolicy(
+    data_class="participation",
+    disposal=Disposal.TOMBSTONE,
+    period_days=7 * _YEAR,
+    anchor="updated_at",
+    authority=_PIA,
+)
+
+# Uploaded documents — 7 years after last activity or 1 year post-program,
+# whichever is greater. Phase 0 encodes the 7-year floor; the post-program
+# branch needs Enrollment end dates (Phase 6).
+DOCUMENTS = RetentionPolicy(
+    data_class="documents",
+    disposal=Disposal.HARD_DELETE,
+    period_days=7 * _YEAR,
+    anchor="updated_at",
+    authority=_PIA + "; post-program branch deferred to Phase 6",
+)
+
+# Consent records — match the retention of the record they relate to.
+# Needs the ConsentRecord → subject link (Phase 1).
+CONSENT = RetentionPolicy(
+    data_class="consent",
+    disposal=Disposal.TOMBSTONE,
+    period_days=None,
+    authority=_PIA + "; derived from related record — Phase 1",
+)
+
+# Audit logs — 1 year online + 3 years archived. The job archives at 1 year;
+# the 3-year archive purge is a storage lifecycle rule, not this job.
+AUDIT = RetentionPolicy(
+    data_class="audit",
+    disposal=Disposal.ARCHIVE,
+    period_days=1 * _YEAR,
+    anchor="ts",
+    authority=_PIA + "; source RFP C.5.7.42; NIST AU-11",
+)
+
+# CFSA foster/ward indicators — current year and one prior; annual purge.
+FOSTER = RetentionPolicy(
+    data_class="cfsa_foster",
+    disposal=Disposal.HARD_DELETE,
+    period_days=2 * _YEAR,
+    cadence=Cadence.ANNUAL,
+    authority=_PIA + "; ARCH §6.4; DC Code Title 4",
+)
+
+# OSSE enrollment records — current year and one prior; annual purge.
+EDUCATION = RetentionPolicy(
+    data_class="osse_ferpa",
+    disposal=Disposal.HARD_DELETE,
+    period_days=2 * _YEAR,
+    cadence=Cadence.ANNUAL,
+    authority=_PIA + "; ARCH §6.4; FERPA",
+)
+
+# Host-side business records. Not covered by PIA §9 — left unset rather
+# than assumed equal to participation.
+HOST_RECORDS = RetentionPolicy(
+    data_class="host_records",
+    disposal=Disposal.TOMBSTONE,
+    period_days=None,
+    authority="not covered by PIA §9 — YPMS-POAM-054",
+)
+
+# Operational logs (integration calls, notifications). Not covered by PIA §9.
+OPERATIONAL = RetentionPolicy(
+    data_class="operational",
+    disposal=Disposal.HARD_DELETE,
+    period_days=None,
+    authority="not covered by PIA §9 — YPMS-POAM-054",
+)
+
+# Version-history snapshots carry the full record over time — they follow
+# the participation envelope of the record they snapshot.
+VERSION_SNAPSHOTS = RetentionPolicy(
+    data_class="version_history",
+    disposal=Disposal.HARD_DELETE,
+    period_days=7 * _YEAR,
+    anchor="ts",
+    authority=_PIA + "; follows participation records",
+)
+
+# User accounts — inactivity disable/purge is YPMS-POAM-071, gated on -054.
+ACCOUNTS = RetentionPolicy(
+    data_class="accounts",
+    disposal=Disposal.TOMBSTONE,
+    period_days=None,
+    authority="YPMS-POAM-071 (depends on YPMS-POAM-054)",
+)
+
+# Program configuration and RBAC — retained by design.
+CONFIGURATION = RetentionPolicy(
+    data_class="configuration",
+    disposal=Disposal.RETAIN,
+    period_days=None,
+    authority="configuration — not personal data",
+)
+
+# Vendored framework tables, mapped by name so the vendored models stay
+# byte-identical to OWC apart from documented divergences.
+FRAMEWORK_TABLE_POLICIES: dict[str, RetentionPolicy] = {
+    "audit_log": AUDIT,
+    "version_history": VERSION_SNAPSHOTS,
+    "users": ACCOUNTS,
+    "roles": CONFIGURATION,
+    "user_roles": CONFIGURATION,
+    "permissions": CONFIGURATION,
+}

src/ypms/stub_models.py

@@ -0,0 +1,158 @@
+"""Phase 0 stub models — id + framework mixins + retention policy only.
+
+PHASE0_PLAN.md §4 / milestone 0.7: every entity in ARCH §6.1 has a model so
+it can declare `retention_policy`, even while its router is a 501 stub
+(`ypms/stubs.py`). No domain fields — those arrive with each domain phase,
+which moves the class into its own module (`ypms/<domain>/models.py`, ARCH
+§5.1) and adds columns in a new migration. The table name stays put.
+
+Tables created by `migrations/versions/0004_stub_entity_tables.py`.
+"""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from sqlalchemy.orm import Mapped, mapped_column
+
+from ypms.db import Base
+from ypms.framework.base import SoftDeleteMixin, TimestampMixin, VersionedMixin
+from ypms.framework.retention import RetentionPolicy
+from ypms.retention_classes import (
+    CONFIGURATION,
+    CONSENT,
+    DOCUMENTS,
+    EDUCATION,
+    FOSTER,
+    HOST_RECORDS,
+    OPERATIONAL,
+    PARTICIPATION,
+)
+
+
+class _Stub(SoftDeleteMixin, TimestampMixin, VersionedMixin):
+    id: Mapped[int] = mapped_column(primary_key=True)
+    retention_policy: ClassVar[RetentionPolicy]
+
+
+# ---- parent/ ----
+
+class Guardian(Base, _Stub):
+    __tablename__ = "guardians"
+    retention_policy = PARTICIPATION
+
+class ConsentRecord(Base, _Stub):
+    __tablename__ = "consent_records"
+    retention_policy = CONSENT
+
+# ---- host/ ----
+
+class Host(Base, _Stub):
+    __tablename__ = "hosts"
+    retention_policy = HOST_RECORDS
+
+class HostContact(Base, _Stub):
+    __tablename__ = "host_contacts"
+    retention_policy = HOST_RECORDS
+
+class Worksite(Base, _Stub):
+    __tablename__ = "worksites"
+    retention_policy = HOST_RECORDS
+
+class Position(Base, _Stub):
+    __tablename__ = "positions"
+    retention_policy = HOST_RECORDS
+
+class JobPosting(Base, _Stub):
+    __tablename__ = "job_postings"
+    retention_policy = HOST_RECORDS
+
+# ---- program/ ----
+
+class Program(Base, _Stub):
+    __tablename__ = "programs"
+    retention_policy = CONFIGURATION
+
+class Cohort(Base, _Stub):
+    __tablename__ = "cohorts"
+    retention_policy = CONFIGURATION
+
+class EligibilityRule(Base, _Stub):
+    __tablename__ = "eligibility_rules"
+    retention_policy = CONFIGURATION
+
+# ---- application/ ----
+
+class Application(Base, _Stub):
+    __tablename__ = "applications"
+    retention_policy = PARTICIPATION
+
+class EligibilityDecision(Base, _Stub):
+    __tablename__ = "eligibility_decisions"
+    retention_policy = PARTICIPATION
+
+class Certification(Base, _Stub):
+    __tablename__ = "certifications"
+    retention_policy = PARTICIPATION
+
+# ---- enrollment/, orientation/ ----
+
+class Orientation(Base, _Stub):
+    __tablename__ = "orientations"
+    retention_policy = PARTICIPATION
+
+class Enrollment(Base, _Stub):
+    __tablename__ = "enrollments"
+    retention_policy = PARTICIPATION
+
+# ---- timekeeping/ ----
+
+class TimeEntry(Base, _Stub):
+    __tablename__ = "time_entries"
+    retention_policy = PARTICIPATION
+
+class Timesheet(Base, _Stub):
+    __tablename__ = "timesheets"
+    retention_policy = PARTICIPATION
+
+# ---- payroll/ ----
+
+class PayrollBatch(Base, _Stub):
+    __tablename__ = "payroll_batches"
+    retention_policy = PARTICIPATION
+
+class TaxDocument(Base, _Stub):
+    __tablename__ = "tax_documents"
+    retention_policy = PARTICIPATION
+
+class TransportationVoucher(Base, _Stub):
+    __tablename__ = "transportation_vouchers"
+    retention_policy = PARTICIPATION
+
+# ---- cross-cutting ----
+
+class Attachment(Base, _Stub):
+    __tablename__ = "attachments"
+    retention_policy = DOCUMENTS
+
+class Note(Base, _Stub):
+    __tablename__ = "notes"
+    retention_policy = PARTICIPATION
+
+class IntegrationEvent(Base, _Stub):
+    __tablename__ = "integration_events"
+    retention_policy = OPERATIONAL
+
+class Notification(Base, _Stub):
+    __tablename__ = "notifications"
+    retention_policy = OPERATIONAL
+
+# ---- integrations/ — FERPA (OSSE) and DC Code Title 4 (CFSA) ----
+
+class EducationRecord(Base, _Stub):
+    __tablename__ = "education_records"
+    retention_policy = EDUCATION
+
+class FosterIndicator(Base, _Stub):
+    __tablename__ = "foster_indicators"
+    retention_policy = FOSTER

src/ypms/youth/models.py

@@ -27,6 +27,8 @@ from sqlalchemy.orm import Mapped, mapped_column
 
 from ypms.db import Base
 from ypms.framework.base import SoftDeleteMixin, TimestampMixin, VersionedMixin
+from ypms.framework.retention import RetentionPolicy
+from ypms.retention_classes import PARTICIPATION
 from ypms.security import column_crypto
 
 SSN_COLUMN = "registrants.ssn"
@@ -40,6 +42,7 @@ def age_on(dob: date, today: date) -> int:
 class Registrant(Base, TimestampMixin, SoftDeleteMixin, VersionedMixin):
     __tablename__ = "registrants"
     __audit_redact__: ClassVar[tuple[str, ...]] = ("ssn_encrypted",)
+    retention_policy: ClassVar[RetentionPolicy] = PARTICIPATION
 
     id: Mapped[int] = mapped_column(primary_key=True)
 

tests/conftest.py

@@ -36,11 +36,7 @@ def db() -> Iterator[Session]:
     Import happens inside the fixture so the model modules register themselves
     on Base.metadata before create_all runs.
     """
-    import ypms.audit.models  # noqa: F401
-    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
+    import ypms.models  # noqa: F401
     from ypms.db import Base
 
     # StaticPool + check_same_thread=False share the same in-memory DB across

tests/framework/_fixtures.py

@@ -5,16 +5,23 @@ slice; until then, this Thing model is enough to prove CRUD + audit + versioning
 
 from __future__ import annotations
 
+from typing import ClassVar
+
 from pydantic import BaseModel
 from sqlalchemy import String
 from sqlalchemy.orm import Mapped, mapped_column
 
 from ypms.db import Base
 from ypms.framework.base import SoftDeleteMixin, TimestampMixin, VersionedMixin
+from ypms.framework.retention import Disposal, RetentionPolicy
 
 
 class Thing(Base, TimestampMixin, SoftDeleteMixin, VersionedMixin):
     __tablename__ = "things"
+    # Every mapped table needs a policy, test-only ones included (milestone 0.7).
+    retention_policy: ClassVar[RetentionPolicy] = RetentionPolicy(
+        "test_fixture", Disposal.RETAIN, None, authority="test-only model"
+    )
 
     id: Mapped[int] = mapped_column(primary_key=True)
     name: Mapped[str] = mapped_column(String(64), nullable=False)

tests/framework/test_retention.py

@@ -0,0 +1,240 @@
+"""Milestone 0.7 — retention scaffold and schema-drift guard."""
+
+from __future__ import annotations
+
+import dataclasses
+import importlib.util
+import json
+from collections.abc import Iterator
+from datetime import UTC, date, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+from alembic import command
+from alembic.autogenerate import compare_metadata
+from alembic.config import Config
+from alembic.runtime.migration import MigrationContext
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+
+import ypms.retention_classes as catalog
+from ypms.audit.models import AuditLog
+from ypms.db import Base
+from ypms.framework import retention
+from ypms.framework.retention import (
+    RETENTION_RUN_ACTION,
+    Disposal,
+    RetentionPolicy,
+    mapped_models,
+    resolve_policy,
+    run_retention,
+)
+from ypms.settings import get_settings
+from ypms.stub_models import FosterIndicator
+from ypms.youth.models import Registrant
+
+ROOT = Path(__file__).resolve().parents[2]
+NOW = datetime(2026, 10, 1, tzinfo=UTC)
+
+# ---------------- policy coverage ----------------
+
+
+def test_every_mapped_table_has_a_policy() -> None:
+    missing = [m.__tablename__ for m in mapped_models() if resolve_policy(m) is None]
+    assert missing == []
+
+
+def test_entity_models_declare_policy_on_the_class() -> None:
+    """Framework tables may use the name map; YPMS entities must not."""
+    framework = set(catalog.FRAMEWORK_TABLE_POLICIES)
+    for m in mapped_models():
+        if m.__tablename__ not in framework:
+            assert "retention_policy" in vars(m), m.__tablename__
+
+
+def test_unset_policies_cite_poam() -> None:
+    for m in mapped_models():
+        p = resolve_policy(m)
+        assert p is not None
+        if p.status == "unset":
+            assert "POAM" in p.authority, m.__tablename__
+
+
+def test_nothing_is_approved_before_records_officer_signoff() -> None:
+    # Flip deliberately when YPMS-POAM-054 closes; update this test with it.
+    assert not any(p.approved for p in _all_policies())
+
+
+def test_anchor_columns_exist() -> None:
+    for m in mapped_models():
+        p = resolve_policy(m)
+        assert p is not None
+        if p.period_days is not None:
+            assert p.anchor in m.__table__.columns, (m.__tablename__, p.anchor)
+
+
+@pytest.mark.parametrize(
+    ("policy", "status"),
+    [
+        (RetentionPolicy("x", Disposal.RETAIN, None), "retain"),
+        (RetentionPolicy("x", Disposal.RETAIN, 10), "retain"),
+        (RetentionPolicy("x", Disposal.TOMBSTONE, None), "unset"),
+        (RetentionPolicy("x", Disposal.TOMBSTONE, 10), "interim"),
+        (RetentionPolicy("x", Disposal.TOMBSTONE, 10, approved=True), "approved"),
+    ],
+)
+def test_policy_status(policy: RetentionPolicy, status: str) -> None:
+    assert policy.status == status
+
+
+def _all_policies() -> list[RetentionPolicy]:
+    return [v for v in vars(catalog).values() if isinstance(v, RetentionPolicy)]
+
+
+# ---------------- job ----------------
+
+
+def _registrant(db: Session, *, updated_at: datetime) -> Registrant:
+    r = Registrant(first_name="A", last_name="B", dob=date(2009, 1, 1))
+    db.add(r)
+    db.flush()
+    r.updated_at = updated_at
+    db.flush()
+    return r
+
+
+def test_empty_run_reports_zero_and_audits(db: Session) -> None:
+    report = run_retention(db, now=NOW)
+    assert report.to_purge == 0
+    assert report.dry_run is True
+    assert {t.table for t in report.tables} == {m.__tablename__ for m in mapped_models()}
+    assert "consent_records" in report.tables_with("unset")
+    assert "permissions" in report.tables_with("retain")
+
+    (row,) = db.execute(select(AuditLog).where(AuditLog.action == RETENTION_RUN_ACTION)).scalars()
+    assert row.actor_id is None
+    assert row.after == report.summary()
+
+
+def test_interim_policy_counts_but_never_purges(db: Session) -> None:
+    old = _registrant(db, updated_at=NOW - timedelta(days=8 * 365))
+    _registrant(db, updated_at=NOW - timedelta(days=30))
+
+    report = run_retention(db, now=NOW)
+
+    reg = next(t for t in report.tables if t.table == "registrants")
+    assert (reg.status, reg.interim_eligible, reg.to_purge) == ("interim", 1, 0)
+    assert report.to_purge == 0
+    assert db.get(Registrant, old.id) is not None
+    assert old.deleted_at is None
+
+
+def test_approved_policy_counts_to_purge_but_phase0_still_reports_only(
+    db: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(
+        Registrant, "retention_policy",
+        dataclasses.replace(catalog.PARTICIPATION, approved=True),
+    )
+    old = _registrant(db, updated_at=NOW - timedelta(days=8 * 365))
+
+    report = run_retention(db, now=NOW)
+
+    reg = next(t for t in report.tables if t.table == "registrants")
+    assert (reg.status, reg.to_purge) == ("approved", 1)
+    assert report.to_purge == 1
+    assert db.get(Registrant, old.id) is not None and old.deleted_at is None
+
+
+def test_annual_ingest_classes_use_two_year_window(db: Session) -> None:
+    for days in (800, 100):
+        f = FosterIndicator()
+        db.add(f)
+        db.flush()
+        f.created_at = NOW - timedelta(days=days)
+    db.flush()
+    report = run_retention(db, now=NOW)
+    foster = next(t for t in report.tables if t.table == "foster_indicators")
+    assert foster.interim_eligible == 1
+
+
+def test_missing_policy_is_an_error(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
+    trimmed = {k: v for k, v in catalog.FRAMEWORK_TABLE_POLICIES.items() if k != "roles"}
+    monkeypatch.setattr(catalog, "FRAMEWORK_TABLE_POLICIES", trimmed)
+    with pytest.raises(LookupError, match="roles"):
+        run_retention(db, now=NOW)
+
+
+def test_main_runs_commits_and_prints_summary(
+    db: Session, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    import ypms.db
+
+    bind = db.get_bind()
+    monkeypatch.setattr(ypms.db, "session_factory", lambda: sessionmaker(bind=bind))
+    assert retention.main() == 0
+    out = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
+    assert out["to_purge"] == 0 and out["dry_run"] is True
+    assert db.execute(select(AuditLog).where(AuditLog.action == RETENTION_RUN_ACTION)).first()
+
+
+# ---------------- migrations ----------------
+
+
+def _load_migration(name: str) -> Any:
+    spec = importlib.util.spec_from_file_location(name, ROOT / "migrations" / "versions" / f"{name}.py")
+    assert spec and spec.loader
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def test_stub_migration_matches_stub_models() -> None:
+    import ypms.stub_models as stubs
+
+    tables = {
+        v.__tablename__ for v in vars(stubs).values()
+        if isinstance(v, type) and issubclass(v, Base) and v is not Base
+    }
+    assert set(_load_migration("0004_stub_entity_tables").TABLES) == tables
+
+
+# Pre-existing in 0002 (immutable): named unique constraints duplicated by the
+# unique indexes the models declare. Follow-up tracked in PHASE0_STATUS.md.
+KNOWN_0002_DRIFT = {("roles", "code"), ("users", "email"), ("users", "username")}
+
+# Mapped only when tests/framework/_fixtures.py is imported; never migrated.
+TEST_ONLY_TABLES = {"things"}
+
+
+@pytest.fixture
+def migrated_url(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[str]:
+    url = f"sqlite:///{tmp_path / 'drift.db'}"
+    monkeypatch.setenv("DATABASE_URL", url)
+    get_settings.cache_clear()
+    try:
+        command.upgrade(Config(str(ROOT / "alembic.ini")), "head")
+        yield url
+    finally:
+        monkeypatch.undo()
+        get_settings.cache_clear()
+
+
+def test_migrations_match_models(migrated_url: str) -> None:
+    engine = create_engine(migrated_url)
+    with engine.connect() as conn:
+        diff = compare_metadata(MigrationContext.configure(conn), Base.metadata)
+    engine.dispose()
+
+    unexpected = []
+    for op in diff:
+        if op[0] == "add_table" and op[1].name in TEST_ONLY_TABLES:
+            continue
+        if op[0] == "remove_constraint":
+            uc = op[1]
+            cols = [c.name for c in uc.columns]
+            if len(cols) == 1 and (uc.table.name, cols[0]) in KNOWN_0002_DRIFT:
+                continue
+        unexpected.append(op)
+    assert unexpected == []