Guard against schema drift and fields dropped on the wire

Two mistakes have recurred in this codebase: the hand-maintained schema in
db/connection.py drifting from the Alembic chain, and a field quietly going
missing from the hand-written mapping onto the plan response. Tests existed
for individual instances of the second, written after each one was found.

Both are now checked structurally. One test builds a database from the
migrations and another from the hand-written schema and compares their real
shape, ignoring cosmetic column ordering but failing on a missing column,
wrong nullability, absent foreign key or index. The other walks the proposal
dataclass and asserts every field reaches the response, so a field added
later is covered without anyone remembering to extend the test. Anything
deliberately kept off the wire goes in an explicit allow-list.

Both were confirmed by breaking what they guard and watching them name it.

backend/tests/integration/test_schema_parity.py

@@ -0,0 +1,251 @@
+"""Mechanism test: Alembic migrations vs. `_FULL_SCHEMA` must describe one schema.
+
+Production boots via `run_migrations()` (the real Alembic chain in
+`backend/migrations/versions/`). All 518+ existing tests instead get their schema
+from `ensure_schema()`, which applies `_FULL_SCHEMA` — a hand-maintained DDL string
+in `app/db/connection.py` — directly, bypassing Alembic entirely (see
+`tests/conftest.py::connection`).
+
+Nothing has ever compared the two. They have drifted before (see
+`PLANNING_ENGINE_SPEC.md` §10.5: "Do not forget `_FULL_SCHEMA`" is called out
+precisely because someone forgot it once already). When they drift, every test
+in the suite keeps passing against a schema production does not have, and the
+gap is only discovered when production breaks.
+
+This test builds one SQLite database via each path and asserts the resulting
+schemas are structurally equivalent: same tables, same columns (name, type,
+nullability, default, primary-key membership), same CHECK constraints, same
+foreign keys, same indexes/uniqueness. Column *order* and cosmetic DDL
+formatting are deliberately ignored — `_FULL_SCHEMA` writes each table's final
+column layout by hand, while the migration chain appends columns via `ALTER
+TABLE ... ADD COLUMN`, so the two legitimately end up in different column
+orders for the same set of columns. That is not a bug and must not fail this
+test.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import sqlite3
+from pathlib import Path
+
+import aiosqlite
+import pytest
+
+from app.db.connection import ensure_schema, run_migrations
+
+# Bookkeeping tables that exist only because of *how* a schema was applied,
+# never part of the application's own data model.
+_IGNORED_TABLES = {"sqlite_sequence", "alembic_version"}
+
+
+def _is_app_table(name: str) -> bool:
+    return name not in _IGNORED_TABLES and not name.startswith("sqlite_")
+
+
+def _table_names(cur: sqlite3.Cursor) -> set[str]:
+    rows = cur.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
+    return {row[0] for row in rows if _is_app_table(row[0])}
+
+
+def _normalize_type(sql_type: str | None) -> str:
+    return (sql_type or "").strip().upper()
+
+
+def _normalize_default(default: object) -> str | None:
+    if default is None:
+        return None
+    return str(default).strip()
+
+
+def _columns(cur: sqlite3.Cursor, table: str) -> dict[str, tuple[str, bool, str | None, int]]:
+    """name -> (normalized type, notnull, normalized default, pk ordinal)."""
+    rows = cur.execute(f'PRAGMA table_info("{table}")').fetchall()
+    # row: cid, name, type, notnull, dflt_value, pk
+    return {
+        row[1]: (_normalize_type(row[2]), bool(row[3]), _normalize_default(row[4]), row[5])
+        for row in rows
+    }
+
+
+def _foreign_keys(cur: sqlite3.Cursor, table: str) -> set[tuple[str, str, str, str, str]]:
+    rows = cur.execute(f'PRAGMA foreign_key_list("{table}")').fetchall()
+    # row: id, seq, table, from, to, on_update, on_delete, match
+    return {(row[2], row[3], row[4] or "", row[5], row[6]) for row in rows}
+
+
+def _indexes(cur: sqlite3.Cursor, table: str) -> set[tuple[bool, tuple[str, ...]]]:
+    """(is_unique, ordered column names) per index — index *names* are not compared,
+    they are SQLite-generated (`sqlite_autoindex_...`) and carry no meaning of their own.
+    """
+    index_rows = cur.execute(f'PRAGMA index_list("{table}")').fetchall()
+    # row: seq, name, unique, origin, partial
+    result: set[tuple[bool, tuple[str, ...]]] = set()
+    for row in index_rows:
+        index_name, is_unique = row[1], bool(row[2])
+        info_rows = cur.execute(f'PRAGMA index_info("{index_name}")').fetchall()
+        # row: seqno, cid, name — order by seqno for a stable column tuple.
+        cols = tuple(c[2] for c in sorted(info_rows, key=lambda c: c[0]))
+        result.add((is_unique, cols))
+    return result
+
+
+def _extract_checks(create_table_sql: str) -> list[str]:
+    """Pull out each balanced `CHECK ( ... )` clause's inner text, raw."""
+    checks: list[str] = []
+    upper = create_table_sql.upper()
+    pos = 0
+    while True:
+        found = upper.find("CHECK", pos)
+        if found == -1:
+            break
+        paren_start = create_table_sql.find("(", found)
+        if paren_start == -1:
+            break
+        depth = 0
+        end = paren_start
+        for end in range(paren_start, len(create_table_sql)):
+            char = create_table_sql[end]
+            if char == "(":
+                depth += 1
+            elif char == ")":
+                depth -= 1
+                if depth == 0:
+                    break
+        checks.append(create_table_sql[paren_start + 1 : end])
+        pos = end + 1
+    return checks
+
+
+def _normalize_check(check_text: str) -> str:
+    return re.sub(r"\s+", " ", check_text).strip()
+
+
+def _checks(cur: sqlite3.Cursor, table: str) -> frozenset[str]:
+    row = cur.execute(
+        "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?", (table,)
+    ).fetchone()
+    sql = row[0] if row and row[0] else ""
+    return frozenset(_normalize_check(c) for c in _extract_checks(sql))
+
+
+def _diff_schemas(migrated: sqlite3.Connection, full_schema: sqlite3.Connection) -> list[str]:
+    """Return a list of human-readable mismatch descriptions; empty means equivalent."""
+    diffs: list[str] = []
+    migrated_cur = migrated.cursor()
+    full_schema_cur = full_schema.cursor()
+
+    migrated_tables = _table_names(migrated_cur)
+    full_schema_tables = _table_names(full_schema_cur)
+
+    only_in_migrations = migrated_tables - full_schema_tables
+    only_in_full_schema = full_schema_tables - migrated_tables
+    if only_in_migrations:
+        diffs.append(
+            f"tables only produced by run_migrations() (missing from _FULL_SCHEMA): "
+            f"{sorted(only_in_migrations)}"
+        )
+    if only_in_full_schema:
+        diffs.append(
+            f"tables only produced by _FULL_SCHEMA (missing from the migration chain): "
+            f"{sorted(only_in_full_schema)}"
+        )
+
+    for table in sorted(migrated_tables & full_schema_tables):
+        migrated_cols = _columns(migrated_cur, table)
+        full_schema_cols = _columns(full_schema_cur, table)
+
+        missing_in_full_schema = migrated_cols.keys() - full_schema_cols.keys()
+        missing_in_migrations = full_schema_cols.keys() - migrated_cols.keys()
+        if missing_in_full_schema:
+            diffs.append(
+                f"table '{table}': column(s) {sorted(missing_in_full_schema)} exist after "
+                f"run_migrations() but are missing from _FULL_SCHEMA"
+            )
+        if missing_in_migrations:
+            diffs.append(
+                f"table '{table}': column(s) {sorted(missing_in_migrations)} exist in "
+                f"_FULL_SCHEMA but the migration chain never creates them"
+            )
+
+        for col in sorted(migrated_cols.keys() & full_schema_cols.keys()):
+            m_type, m_notnull, m_default, m_pk = migrated_cols[col]
+            f_type, f_notnull, f_default, f_pk = full_schema_cols[col]
+            if (m_type, m_notnull, m_default, m_pk) != (f_type, f_notnull, f_default, f_pk):
+                diffs.append(
+                    f"table '{table}' column '{col}': migrations gives "
+                    f"(type={m_type!r}, notnull={m_notnull}, default={m_default!r}, pk={m_pk}) "
+                    f"but _FULL_SCHEMA gives "
+                    f"(type={f_type!r}, notnull={f_notnull}, default={f_default!r}, pk={f_pk})"
+                )
+
+        m_checks = _checks(migrated_cur, table)
+        f_checks = _checks(full_schema_cur, table)
+        if m_checks != f_checks:
+            diffs.append(
+                f"table '{table}': CHECK constraints differ — "
+                f"migrations only: {sorted(m_checks - f_checks)}; "
+                f"_FULL_SCHEMA only: {sorted(f_checks - m_checks)}"
+            )
+
+        m_fks = _foreign_keys(migrated_cur, table)
+        f_fks = _foreign_keys(full_schema_cur, table)
+        if m_fks != f_fks:
+            diffs.append(
+                f"table '{table}': foreign keys differ (ref_table, from_col, to_col, "
+                f"on_update, on_delete) — "
+                f"migrations only: {sorted(m_fks - f_fks)}; "
+                f"_FULL_SCHEMA only: {sorted(f_fks - m_fks)}"
+            )
+
+        m_idx = _indexes(migrated_cur, table)
+        f_idx = _indexes(full_schema_cur, table)
+        if m_idx != f_idx:
+            diffs.append(
+                f"table '{table}': indexes differ (is_unique, columns) — "
+                f"migrations only: {sorted(m_idx - f_idx)}; "
+                f"_FULL_SCHEMA only: {sorted(f_idx - m_idx)}"
+            )
+
+    return diffs
+
+
+async def _apply_full_schema(db_path: Path) -> None:
+    conn = await aiosqlite.connect(db_path)
+    try:
+        await ensure_schema(conn)
+    finally:
+        await conn.close()
+
+
+def test_alembic_migrations_and_full_schema_agree(tmp_path: Path) -> None:
+    """`run_migrations()` (production) and `ensure_schema()` (tests) must build the
+    same schema. A failure here means the 518+ tests using `_FULL_SCHEMA` are
+    exercising a schema production does not actually have — see module docstring.
+    """
+    migrated_db = tmp_path / "via_migrations.db"
+    full_schema_db = tmp_path / "via_full_schema.db"
+
+    # Alembic chain, exactly as production's app lifespan runs it.
+    run_migrations(f"sqlite:///{migrated_db}")
+
+    # `ensure_schema()` is async (an aiosqlite connection); `run_migrations()` above
+    # must run to completion first since it drives its own `asyncio.run()` internally
+    # and cannot be called from within a running event loop.
+    asyncio.run(_apply_full_schema(full_schema_db))
+
+    migrated_conn = sqlite3.connect(migrated_db)
+    full_schema_conn = sqlite3.connect(full_schema_db)
+    try:
+        diffs = _diff_schemas(migrated_conn, full_schema_conn)
+    finally:
+        migrated_conn.close()
+        full_schema_conn.close()
+
+    if diffs:
+        pytest.fail(
+            "Schema drift between Alembic migrations (production) and _FULL_SCHEMA "
+            "(tests) in backend/app/db/connection.py:\n- " + "\n- ".join(diffs),
+            pytrace=False,
+        )

backend/tests/unit/test_plan_router.py

@@ -15,6 +15,7 @@ Uses the same in-memory sqlite `connection` fixture as the rest of the suite
 graph without touching the real `foraldraledig.db` file or running Alembic.
 """
 
+import dataclasses
 from datetime import date, datetime
 from uuid import uuid4
 
@@ -35,6 +36,7 @@ from app.repositories_impl.sqlite.parental_leave_repository import SQLiteParenta
 from app.repositories_impl.sqlite.personal_data_repository import SQLitePersonalDataRepository
 from app.repositories_impl.sqlite.vacation_days_repository import SQLiteVacationDaysRepository
 from app.repositories_impl.sqlite.workspace_repository import SQLiteWorkspaceRepository
+from app.services.allocation_service import PlanProposal, WindowOutcome
 
 
 def _make_fetch(
@@ -809,3 +811,260 @@ class TestLeaveWishWindowModeOverTheWire:
         assert outcome["strategy"] == "unplaced"
         assert outcome["weeks_funded"] == 0.0
         assert outcome["shortfall_days"] == 7
+
+
+# ---------------------------------------------------------------------------
+# Systematic wire-contract guard for the plan proposal (mechanism test)
+# ---------------------------------------------------------------------------
+#
+# The classes above each guard exactly one field that was already silently
+# dropped from `app/models/plan.py`'s hand-written `from_domain()` mapping
+# once (surplus_advice, age4_projections, wish_name_message_key). That is
+# whack-a-mole: it only ever protects a field *after* losing it has already
+# been painful enough to write a regression test for.
+#
+# This class instead walks `dataclasses.fields()` of the domain `PlanProposal`
+# (and its nested `WindowOutcome`) and checks each one has a surviving
+# counterpart in the actual HTTP JSON. Critically, a domain field with no
+# entry in the rename map or allow-list below is checked under its OWN name
+# by default -- so a brand-new field added to `PlanProposal` (or
+# `WindowOutcome`) and forgotten in `from_domain()` fails HERE, without this
+# test file needing to be touched first. Only intentional renames/fan-outs
+# and intentional exclusions require a (documented) entry.
+
+# Domain field name -> wire field name(s) on `PlanProposalResponse`. Only
+# fields whose wire shape differs from a straight "same name" passthrough
+# need an entry. `balances_after` is the one case today: a single nested
+# `PlanBalances` dataclass that `from_domain()` fans out into two separate
+# top-level wire lists.
+_PLAN_PROPOSAL_RENAMES: dict[str, tuple[str, ...]] = {
+    "balances_after": ("pool_balances", "vacation_balances"),
+}
+
+# Domain fields on `PlanProposal` deliberately absent from the wire, with why.
+# Empty today: every `PlanProposal` field currently reaches the HTTP response.
+_PLAN_PROPOSAL_ALLOWLIST: dict[str, str] = {}
+
+# Same idea for `WindowOutcome` -> `WindowOutcomeResponse`: every field there
+# is a straight same-name passthrough today, so both maps are empty, but they
+# stay explicit (not just "no map at all") so the intent -- and the place to
+# add the next rename/exclusion -- is obvious.
+_WINDOW_OUTCOME_RENAMES: dict[str, tuple[str, ...]] = {}
+_WINDOW_OUTCOME_ALLOWLIST: dict[str, str] = {}
+
+
+def _wire_keys_for(domain_field: str, renames: dict[str, tuple[str, ...]]) -> tuple[str, ...]:
+    """The wire key(s) `domain_field` must appear as, absent an explicit rename."""
+    return renames.get(domain_field, (domain_field,))
+
+
+def _assert_domain_fields_reach_the_wire(
+    dataclass_type: type,
+    renames: dict[str, tuple[str, ...]],
+    allowlist: dict[str, str],
+    wire_obj: dict[str, object],
+    *,
+    context: str,
+    check_nonempty: bool = True,
+) -> None:
+    """Fail loudly, naming the exact field, if any `dataclass_type` field has
+    no live counterpart in `wire_obj` (a decoded JSON object from the HTTP
+    response) -- unless it is explicitly excused via `allowlist`.
+
+    `check_nonempty` additionally requires list-typed wire fields to be
+    non-empty. Leave it on for a single, richly-seeded top-level object (e.g.
+    the `PlanProposalResponse` body, where every list should have content);
+    turn it off when walking a list of *instances* of a nested dataclass
+    (e.g. each individual `WindowOutcomeResponse`) — a specific outcome's own
+    list field (like `remedies`) can be legitimately empty (a fully-funded
+    window has no remedies) without that being a dropped-field bug.
+    """
+    for f in dataclasses.fields(dataclass_type):
+        if f.name in allowlist:
+            continue
+        for wire_key in _wire_keys_for(f.name, renames):
+            assert wire_key in wire_obj, (
+                f"{context}: domain field '{f.name}' has no counterpart "
+                f"'{wire_key}' in the wire response. Either from_domain() in "
+                f"app/models/plan.py silently dropped it, or (if '{f.name}' is "
+                f"a brand-new PlanProposal/WindowOutcome field) it needs a "
+                f"wire mapping AND a rename/allow-list entry in this test."
+            )
+            if not check_nonempty:
+                continue
+            value = wire_obj[wire_key]
+            if isinstance(value, list):
+                assert value, (
+                    f"{context}: wire field '{wire_key}' (from domain field "
+                    f"'{f.name}') is present but empty. Either the seeded "
+                    f"workspace in this test needs to be richer, or "
+                    f"from_domain() is silently producing an empty list where "
+                    f"the domain data is not actually empty."
+                )
+
+
+class TestPlanProposalFieldParityWithDomain:
+    """Mechanism test: every `PlanProposal`/`WindowOutcome` field must survive
+    the hand-written `PlanProposalResponse.from_domain()` mapping, not just
+    the three fields the classes above already know to check for by name.
+    """
+
+    async def test_every_plan_proposal_and_window_outcome_field_reaches_the_wire(
+        self,
+        client: AsyncClient,
+        connection: aiosqlite.Connection,
+    ) -> None:
+        now = datetime.now()
+        workspace_repo = SQLiteWorkspaceRepository(connection)
+        personal_data_repo = SQLitePersonalDataRepository(connection)
+        child_repo = SQLiteChildRepository(connection)
+        parental_leave_repo = SQLiteParentalLeaveRepository(connection)
+        vacation_repo = SQLiteVacationDaysRepository(connection)
+
+        ws = await workspace_repo.create(Workspace(id=uuid4(), name="Familj", created_at=now))
+        parent = await personal_data_repo.create(
+            PersonalData(
+                uuid=uuid4(),
+                workspace_id=ws.id,
+                name="Alex",
+                salary_type="monthly",
+                created_at=now,
+            )
+        )
+
+        # Child A: pool fully exhausted + no vacation in the wish's funding
+        # year -> a guaranteed shortfall (populates outcomes[].remedies,
+        # outcome.data, outcome.cost_breakdown with a genuinely non-trivial
+        # story) — same recipe as TestProposeMessageKeysOverTheWire above.
+        child_shortfall = await child_repo.create(
+            Child(
+                id=uuid4(),
+                workspace_id=ws.id,
+                name="Barn Ett",
+                birth_date=date.today() - relativedelta(months=6),
+                created_at=now,
+            )
+        )
+        await parental_leave_repo.create(
+            _make_fetch(parent.uuid, ws.id, child_shortfall.id, sjukpenning_left=0, lagsta_left=0)
+        )
+
+        next_year = date.today().year + 1
+        await vacation_repo.create(
+            VacationDays(
+                id=uuid4(),
+                personal_data_uuid=parent.uuid,
+                workspace_id=ws.id,
+                year=next_year,
+                total_days=0,
+                saved_days_in=0,
+            )
+        )
+        # Spare vacation *this* year gives `carry_vacation_forward` something
+        # to point at: available (25) above the semesterlagen minimum (20).
+        await vacation_repo.create(
+            VacationDays(
+                id=uuid4(),
+                personal_data_uuid=parent.uuid,
+                workspace_id=ws.id,
+                year=date.today().year,
+                total_days=25,
+                saved_days_in=0,
+            )
+        )
+
+        preset_wish = await client.post(
+            "/api/leave-wishes",
+            json={
+                "workspace_id": str(ws.id),
+                "child_id": str(child_shortfall.id),
+                "name": "Vinter",
+                "name_message_key": "wishEditor.presetVinter",
+                "window_start_month": 12,
+                "window_start_day": 1,
+                "window_end_month": 12,
+                "window_end_day": 7,
+                "target_weeks": 1,
+                "recurrence": "once",
+                "first_year": next_year,
+            },
+        )
+        assert preset_wish.status_code == 201
+
+        # Child B: no `parental_leave_fetch` row at all -> the full default FP
+        # pool, untouched by the small wish below, still swamps the age-4
+        # ceiling -> guaranteed `surplus_advice` + `days_lost_at_age_4`
+        # warning + a real `age4_projections` story — same recipe as
+        # TestProposeSurplusAdviceOverTheWire above. The small wish itself
+        # (well inside the huge default pool) also guarantees `days` and a
+        # second, fully-funded `outcomes` entry are non-trivially populated.
+        child_surplus = await child_repo.create(
+            Child(
+                id=uuid4(),
+                workspace_id=ws.id,
+                name="Barn Två",
+                birth_date=date.today() - relativedelta(months=6),
+                created_at=now,
+            )
+        )
+        funded_wish = await client.post(
+            "/api/leave-wishes",
+            json={
+                "workspace_id": str(ws.id),
+                "child_id": str(child_surplus.id),
+                "name": "Sportlov",
+                "window_start_month": 2,
+                "window_start_day": 1,
+                "window_end_month": 2,
+                "window_end_day": 7,
+                "target_weeks": 1,
+                "recurrence": "once",
+                "first_year": next_year,
+            },
+        )
+        assert funded_wish.status_code == 201
+
+        response = await client.post("/api/plan/propose", json={"workspace_id": str(ws.id)})
+        assert response.status_code == 200
+        body = response.json()
+
+        # Sanity-check the seed actually produced a non-trivial proposal
+        # before trusting the generic field walk below to mean anything.
+        assert body["days"], "seed should have produced at least one funded day"
+        assert body["outcomes"], "seed should have produced at least one outcome"
+        assert body["warnings"], "seed should have produced at least one warning"
+        assert body["surplus_advice"], "seed should have produced surplus advice"
+        assert body["age4_projections"], "seed should have produced age4 projections"
+        assert body["pool_balances"], "seed should have produced pool balances"
+        assert body["vacation_balances"], "seed should have produced vacation balances"
+
+        _assert_domain_fields_reach_the_wire(
+            PlanProposal,
+            _PLAN_PROPOSAL_RENAMES,
+            _PLAN_PROPOSAL_ALLOWLIST,
+            body,
+            context="PlanProposal -> PlanProposalResponse",
+        )
+
+        for outcome in body["outcomes"]:
+            _assert_domain_fields_reach_the_wire(
+                WindowOutcome,
+                _WINDOW_OUTCOME_RENAMES,
+                _WINDOW_OUTCOME_ALLOWLIST,
+                outcome,
+                context="WindowOutcome -> WindowOutcomeResponse",
+                # A specific outcome's own list field (e.g. `remedies`) can be
+                # legitimately empty (a fully-funded window has none) --
+                # presence is still checked, only the blanket emptiness
+                # assertion is skipped here. See docstring above.
+                check_nonempty=False,
+            )
+
+        # `remedies` is a list-of-dataclass field inside WindowOutcome too —
+        # confirm the shortfall outcome specifically has a non-empty one
+        # (the generic walk above only guarantees each outcome's own
+        # `remedies` key exists and is non-empty where non-empty is expected;
+        # this pins down that the shortfall scenario, not the funded one, is
+        # the one carrying it).
+        shortfall_outcome = next(o for o in body["outcomes"] if o["shortfall_days"] > 0)
+        assert shortfall_outcome["remedies"], "expected the carry_vacation_forward remedy"