feat(evidence): enforce the traceability_events mutation policy and derive waiver approvers from the principal

alembic/versions/20261028_08c7cf36ddbe_v141_traceability_events_append_only.py

@@ -0,0 +1,198 @@
+"""v141 — fsma.traceability_events mutation policy (column allowlist trigger).
+
+``fsma.traceability_events`` is the canonical FSMA 204 evidence table and
+the chain hash is computed over its payload columns, yet until now nothing
+in the database stopped an UPDATE or DELETE on it. ``fsma.cte_events`` has
+had an append-only trigger since v107; the canonical table had none, so a
+``UPDATE fsma.traceability_events SET quantity = ...`` from any role with
+UPDATE privilege silently broke the evidence the FDA package is built on.
+
+This revision installs a BEFORE UPDATE OR DELETE row trigger plus a BEFORE
+TRUNCATE statement trigger that together enforce the sanctioned mutation
+surface, which is exactly what the three production writers do today:
+
+* ``services/shared/canonical_persistence/writer.py`` marks the row an
+  amendment supersedes: ``status = 'superseded', amended_at = NOW()``.
+* ``services/shared/rules/engine.py`` records the event-level verdict:
+  ``evaluation_verdict``, ``no_verdict_reason``, ``evaluated_at``, and the
+  evaluation error marker: ``evaluation_error``, ``evaluated_at``.
+
+Allowed on UPDATE: changes confined to the allowlisted columns
+(``status``, ``amended_at``, ``evaluation_verdict``, ``no_verdict_reason``,
+``evaluation_error``, ``evaluated_at``), and ``status`` may only move from
+``active`` to ``superseded``. Any other column change, any other status
+transition, every DELETE and every TRUNCATE raise ``check_violation``.
+
+Break-glass: ``SET LOCAL fsma.allow_mutation = 'true'`` lets the statement
+through with a WARNING, the same GUC v095 and v107 use. The runtime resets
+it per transaction (``services/shared/tenant_context.py``) and per pooled
+connection (``services/admin/app/database.py``); the retention purge is
+its only production user.
+
+The comparison is ``to_jsonb(NEW) - allowlist = to_jsonb(OLD) - allowlist``
+rather than a list of frozen column names, so the trigger is correct on
+every catalog variant of the table (the Supabase twin types the tenant
+column as text and has no ``ingested_at``) and keeps protecting columns
+added later.
+
+Idempotent: ``CREATE OR REPLACE`` functions, ``DROP TRIGGER IF EXISTS``
+before ``CREATE TRIGGER``, ``to_regclass`` guard. No grant changes (v111
+still grants UPDATE to the runtime role; the trigger is the enforcement).
+
+Revision ID: 08c7cf36ddbe
+Revises: b8c1e4f2a901
+Create Date: 2026-10-28
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "08c7cf36ddbe"  # pragma: allowlist secret
+down_revision: Union[str, Sequence[str], None] = "b8c1e4f2a901"  # pragma: allowlist secret
+branch_labels = None
+depends_on = None
+
+TABLE = "fsma.traceability_events"
+ROW_TRIGGER = "traceability_events_mutation_policy"
+TRUNCATE_TRIGGER = "traceability_events_no_truncate"
+ROW_FUNCTION = "fsma.prevent_traceability_event_mutation"
+TRUNCATE_FUNCTION = "fsma.prevent_traceability_event_truncate"
+
+# Columns the sanctioned writers may change. Keep in step with the module
+# docstring and docs/engineering/SCHEMA_CHANGE_POLICY.md §5.
+ALLOWLIST: tuple[str, ...] = (
+    "status",
+    "amended_at",
+    "evaluation_verdict",
+    "no_verdict_reason",
+    "evaluation_error",
+    "evaluated_at",
+)
+
+_ALLOWLIST_SQL = ", ".join(f"'{c}'" for c in ALLOWLIST)
+
+_ROW_FUNCTION_SQL = f"""
+CREATE OR REPLACE FUNCTION {ROW_FUNCTION}()
+RETURNS TRIGGER LANGUAGE plpgsql AS $$
+DECLARE
+    allowlist text[] := ARRAY[{_ALLOWLIST_SQL}];
+    changed   text[];
+BEGIN
+    IF current_setting('fsma.allow_mutation', true) = 'true' THEN
+        RAISE WARNING
+            'traceability_events modification via break-glass. op=% user=% event_id=%',
+            TG_OP, current_user, COALESCE(NEW.event_id::text, OLD.event_id::text);
+        IF TG_OP = 'UPDATE' THEN
+            RETURN NEW;
+        END IF;
+        RETURN OLD;
+    END IF;
+
+    IF TG_OP = 'DELETE' THEN
+        RAISE EXCEPTION
+            'traceability_events is append-only (FDA 21 CFR 1.1455). DELETE rejected for event_id=%. Set fsma.allow_mutation=true for authorized break-glass.',
+            OLD.event_id
+            USING ERRCODE = 'check_violation';
+    END IF;
+
+    -- UPDATE: every changed column must be on the allowlist.
+    SELECT COALESCE(array_agg(key ORDER BY key), ARRAY[]::text[])
+      INTO changed
+      FROM (
+          SELECT key FROM jsonb_each(to_jsonb(NEW) - allowlist)
+          EXCEPT
+          SELECT key FROM jsonb_each(to_jsonb(OLD) - allowlist)
+          UNION
+          SELECT n.key
+            FROM jsonb_each(to_jsonb(NEW) - allowlist) n
+            JOIN jsonb_each(to_jsonb(OLD) - allowlist) o ON o.key = n.key
+           WHERE n.value IS DISTINCT FROM o.value
+      ) diff;
+    IF array_length(changed, 1) IS NOT NULL THEN
+        RAISE EXCEPTION
+            'traceability_events is append-only (FDA 21 CFR 1.1455). UPDATE rejected for event_id=%: column(s) % are not amendable. Corrections are new events that supersede the old one. Set fsma.allow_mutation=true for authorized break-glass.',
+            OLD.event_id, array_to_string(changed, ', ')
+            USING ERRCODE = 'check_violation';
+    END IF;
+
+    -- status may only move active -> superseded (an amendment landed).
+    -- IS NOT TRUE, not NOT (...): the Supabase twin leaves status nullable,
+    -- and a NULL on either side would turn NOT (...) into NULL and let the
+    -- change through.
+    IF NEW.status IS DISTINCT FROM OLD.status
+       AND (OLD.status = 'active' AND NEW.status = 'superseded') IS NOT TRUE THEN
+        RAISE EXCEPTION
+            'traceability_events status may only move from active to superseded; % -> % rejected for event_id=%. Set fsma.allow_mutation=true for authorized break-glass.',
+            OLD.status, NEW.status, OLD.event_id
+            USING ERRCODE = 'check_violation';
+    END IF;
+
+    RETURN NEW;
+END;
+$$;
+"""
+
+_TRUNCATE_FUNCTION_SQL = f"""
+CREATE OR REPLACE FUNCTION {TRUNCATE_FUNCTION}()
+RETURNS TRIGGER LANGUAGE plpgsql AS $$
+BEGIN
+    IF current_setting('fsma.allow_mutation', true) = 'true' THEN
+        RAISE WARNING 'traceability_events TRUNCATE via break-glass. user=%', current_user;
+        RETURN NULL;
+    END IF;
+    RAISE EXCEPTION
+        'traceability_events is append-only (FDA 21 CFR 1.1455). TRUNCATE rejected. Set fsma.allow_mutation=true for authorized break-glass.'
+        USING ERRCODE = 'check_violation';
+END;
+$$;
+"""
+
+_TRIGGERS_SQL = f"""
+DO $$
+BEGIN
+    IF to_regclass('{TABLE}') IS NULL THEN
+        RETURN;
+    END IF;
+    DROP TRIGGER IF EXISTS {ROW_TRIGGER} ON {TABLE};
+    CREATE TRIGGER {ROW_TRIGGER}
+        BEFORE UPDATE OR DELETE ON {TABLE}
+        FOR EACH ROW EXECUTE FUNCTION {ROW_FUNCTION}();
+    DROP TRIGGER IF EXISTS {TRUNCATE_TRIGGER} ON {TABLE};
+    CREATE TRIGGER {TRUNCATE_TRIGGER}
+        BEFORE TRUNCATE ON {TABLE}
+        FOR EACH STATEMENT EXECUTE FUNCTION {TRUNCATE_FUNCTION}();
+END $$;
+"""
+
+_DOWNGRADE_SQL = f"""
+DO $$
+BEGIN
+    IF to_regclass('{TABLE}') IS NOT NULL THEN
+        DROP TRIGGER IF EXISTS {ROW_TRIGGER} ON {TABLE};
+        DROP TRIGGER IF EXISTS {TRUNCATE_TRIGGER} ON {TABLE};
+    END IF;
+END $$;
+DROP FUNCTION IF EXISTS {ROW_FUNCTION}();
+DROP FUNCTION IF EXISTS {TRUNCATE_FUNCTION}();
+"""
+
+
+# CREATE/DROP TRIGGER takes SHARE ROW EXCLUSIVE on the table and would queue
+# behind a long ingest or export transaction, holding every later writer
+# behind it. Bound the wait (same 10s as v134) so the migration fails fast
+# and can be retried instead of stalling production. Alembic runs each
+# path in one transaction, so SET LOCAL covers every statement below.
+_LOCK_TIMEOUT_SQL = "SET LOCAL lock_timeout = '10s'"
+
+
+def upgrade() -> None:
+    op.execute(_LOCK_TIMEOUT_SQL)
+    op.execute(_ROW_FUNCTION_SQL)
+    op.execute(_TRUNCATE_FUNCTION_SQL)
+    op.execute(_TRIGGERS_SQL)
+
+
+def downgrade() -> None:
+    op.execute(_LOCK_TIMEOUT_SQL)
+    op.execute(_DOWNGRADE_SQL)

docs/engineering/SCHEMA_CHANGE_POLICY.md

@@ -89,6 +89,37 @@ Engineering must implement. Reviewers must enforce.
 | Every verdict records `rule_version_id`, `fact_version_ids`, `authority_pointer_ids` | NOT NULL + FK constraints |
 | Missing data → `INDETERMINATE` | Application-level (never NULL verdicts) |
 | Corrections are new versions | No UPDATE on verdict tables (trigger-enforced) |
+| `fsma.traceability_events` is append-only | v141 trigger (`traceability_events_mutation_policy`, `traceability_events_no_truncate`); see 5.1 |
+| Legacy CTE tables are read-only retention | No new writers, columns or indexes on `fsma.cte_events` and `fsma.critical_tracking_events`; see 5.2 |
+
+### 5.1 Sanctioned mutations of `fsma.traceability_events`
+
+The canonical evidence table rejects every UPDATE, DELETE and TRUNCATE at the database (migration
+v141, `08c7cf36ddbe`) except the two writers the application already relies on:
+
+| Writer | Columns it may change | Constraint |
+|--------|-----------------------|------------|
+| Amendment (`services/shared/canonical_persistence/writer.py`) | `status`, `amended_at` | `status` may only move `active -> superseded` |
+| Rules engine (`services/shared/rules/engine.py`) | `evaluation_verdict`, `no_verdict_reason`, `evaluation_error`, `evaluated_at` | none |
+
+Any statement that changes another column, moves `status` anywhere else, deletes or truncates
+raises `check_violation`. A correction is a new event that supersedes the old one, never an edit.
+The break-glass is `SET LOCAL fsma.allow_mutation = 'true'` inside one transaction (the same GUC
+as `fsma.cte_events` and `fsma.correction_patches`); the runtime resets it per transaction and per
+pooled connection, and the retention purge is its only production user. Adding a column to the
+allowlist is a schema change under this policy: update the trigger in a new migration, this table,
+and `tests/migrations/test_v141_traceability_events_append_only.py` together.
+
+### 5.2 Read freeze on the legacy CTE tables (SEL-589)
+
+`fsma.traceability_events` is the committed canonical contract for all new FSMA evidence (product
+decision 2026-10-27, `docs/product/PRODUCT_PRD_2026-10.md`). `fsma.cte_events` and
+`fsma.critical_tracking_events` are read-only retention: no new writers, columns or indexes. Their
+reads are not frozen yet: the v1 FDA export and recall readers still default to `fsma.cte_events`
+behind `FDA_EXPORT_V1_READ_CANONICAL` (default off, `services/ingestion/app/fda_export/queries.py`).
+Each export, recall, trace and verification reader is cut over one at a time, only after its
+historical-row and canonical coverage is documented. A migration that adds to either legacy table
+is rejected under section 7.
 
 ---
 
@@ -115,6 +146,6 @@ Engineering must implement. Reviewers must enforce.
 
 ## Appendix: Current Schema Version
 
-- **Version:** 1.1.2
-- **Last Updated:** 2026-01-22
+- **Version:** 1.2.0
+- **Last Updated:** 2026-10-28 (v141 mutation policy on `fsma.traceability_events`; SEL-589 read freeze)
 - **Git SHA:** (auto-populated by migration)

services/ingestion/app/exception_router.py

@@ -88,8 +88,13 @@ class ResolveRequest(BaseModel):
 
 
 class WaiveRequest(BaseModel):
+    """Waiver body. The approver is the authenticated principal, never a body field.
+
+    A client-supplied ``waiver_approved_by`` is ignored (pydantic drops
+    unknown fields) so a caller cannot sign a waiver in someone else's name.
+    """
+
     waiver_reason: str
-    waiver_approved_by: str
 
 
 class AddCommentRequest(BaseModel):
@@ -237,9 +242,23 @@ async def resolve_exception(
     return {"case_id": case_id, "status": "resolved"}
 
 
+def _actor_from_principal(principal: IngestionPrincipal) -> str:
+    """Identity string recorded as the actor of a signoff.
+
+    Mirrors ``_audit_identity`` in the FDA export router: identity fields
+    resolve from the authenticated principal, never from the request body.
+    ``key_id`` is ``session:<user id>`` for browser sessions and the API key
+    id for scoped keys. An empty principal cannot approve anything.
+    """
+    actor = (getattr(principal, "key_id", None) or "").strip()
+    if not actor:
+        raise HTTPException(status_code=401, detail="Authenticated principal required to approve a waiver")
+    return actor
+
+
 @router.patch(
     "/{case_id}/waive",
-    summary="Waive exception (requires reason + approver)",
+    summary="Waive exception (requires reason; approver is the authenticated principal)",
 )
 async def waive_exception(
     case_id: str,
@@ -250,7 +269,7 @@ async def waive_exception(
 ):
     tid = resolve_tenant(tenant_id, principal)
     svc = _get_service(db_session)
-    svc.waive_exception(tid, case_id, body.waiver_reason, body.waiver_approved_by)
+    svc.waive_exception(tid, case_id, body.waiver_reason, _actor_from_principal(principal))
     return {"case_id": case_id, "status": "waived"}
 
 

services/ingestion/tests/test_auditor_pagination_ties_pg.py

@@ -139,6 +139,7 @@ def seeded_tenant():
         yield engine, tenant_id, set(lot_codes)
     finally:
         with engine.begin() as conn:
+            conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'"))  # v141 append-only
             conn.execute(
                 text("DELETE FROM fsma.traceability_events WHERE tenant_id = :tid"),
                 {"tid": tenant_id},

services/ingestion/tests/test_disaster_recovery_data_volume_pg.py

@@ -98,6 +98,7 @@ def seeded_tenants():
         yield engine, tenant_id
     finally:
         with engine.begin() as conn:
+            conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'"))  # v141 append-only
             conn.execute(
                 text("DELETE FROM fsma.traceability_events WHERE tenant_id IN (:tid, :other)"),
                 {"tid": tenant_id, "other": other_tenant_id},

services/ingestion/tests/test_exception_router.py

@@ -236,22 +236,52 @@ def test_resolve_exception(client, mock_svc):
 # ---------------------------------------------------------------------------
 
 
-def test_waive_exception(client, mock_svc):
+def test_waive_exception_records_principal_as_approver(client, mock_svc):
+    """S5: the approver is the authenticated principal's key_id, not a body field."""
     resp = client.patch(
         "/api/v1/exceptions/exc-1/waive",
-        json={
-            "waiver_reason": "Low-risk supplier",
-            "waiver_approved_by": "admin-1",
-        },
+        json={"waiver_reason": "Low-risk supplier"},
     )
     assert resp.status_code == 200
     assert resp.json()["status"] == "waived"
     mock_svc.waive_exception.assert_called_once_with(
         TENANT,
         "exc-1",
         "Low-risk supplier",
-        "admin-1",
+        "test-key",
+    )
+
+
+def test_waive_exception_ignores_body_supplied_approver(client, mock_svc):
+    """A caller cannot sign a waiver in someone else's name."""
+    resp = client.patch(
+        "/api/v1/exceptions/exc-1/waive",
+        json={
+            "waiver_reason": "Low-risk supplier",
+            "waiver_approved_by": "someone-else",
+        },
+    )
+    assert resp.status_code == 200
+    assert mock_svc.waive_exception.call_args.args[3] == "test-key"
+
+
+def test_waive_exception_requires_reason(client, mock_svc):
+    resp = client.patch("/api/v1/exceptions/exc-1/waive", json={})
+    assert resp.status_code == 422
+    mock_svc.waive_exception.assert_not_called()
+
+
+def test_waive_exception_rejects_empty_principal_identity():
+    principal = _make_principal()
+    principal = principal.model_copy(update={"key_id": ""})
+    svc = MagicMock()
+    client = _build_client(principal, svc)
+    resp = client.patch(
+        "/api/v1/exceptions/exc-1/waive",
+        json={"waiver_reason": "Low-risk supplier"},
     )
+    assert resp.status_code == 401
+    svc.waive_exception.assert_not_called()
 
 
 # ---------------------------------------------------------------------------

services/ingestion/tests/test_product_catalog_derive_events_pg.py

@@ -98,6 +98,7 @@ def seeded_tenant():
         yield engine, tenant_id
     finally:
         with engine.begin() as conn:
+            conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'"))  # v141 append-only
             conn.execute(
                 text("DELETE FROM fsma.traceability_events WHERE tenant_id = :tid"),
                 {"tid": tenant_id},

tests/invariants/test_backfill_canonical.py

@@ -252,6 +252,8 @@ def _persist_dual(session, event):
 
 def _orphan_the_canonical_row(session, tenant_id, event_id):
     """Simulate the #1277 failure: legacy row survives, canonical row is lost."""
+    # v141 append-only trigger: the simulated loss needs the break-glass GUC.
+    session.execute(text("SET LOCAL fsma.allow_mutation = 'true'"))
     session.execute(
         text(
             "DELETE FROM fsma.traceability_events "

tests/invariants/test_verify_canonical_parity_live.py

@@ -758,6 +758,8 @@ def test_canonical_drift_on_recall_predicate_columns_is_detected(self, seeded_te
         database_url, engine, tenant = seeded_tenant
         with engine.begin() as conn:
             conn.execute(text("SELECT set_config('app.tenant_id', :tid, true)"), {"tid": tenant})
+            # v141 append-only trigger: simulated drift needs the break-glass GUC.
+            conn.execute(text("SET LOCAL fsma.allow_mutation = 'true'"))
             conn.execute(
                 text(
                     "UPDATE fsma.traceability_events SET quantity = 999.0, "