feat(governance): constraint inventory + CASCADE-delete risk

backend/app/api/snapshots.py

@@ -19,6 +19,7 @@
 )
 from app.permissions import require_project_member
 from app.schemas import (
+    ConstraintInventoryOut,
     FkCyclesOut,
     InferredRelationshipOut,
     MigrationSafetyOut,
@@ -35,6 +36,7 @@
 from app.ddl.migration import snapshot_diff_to_migration_sql
 from app.ddl.migration_safety import analyze_migration_safety
 from app.diff.schema_diff import diff_snapshots
+from app.spec.constraint_inventory import build_constraint_inventory
 from app.spec.fk_cycles import detect_fk_cycles
 from app.spec.data_dictionary import snapshot_to_data_dictionary_md
 from app.spec.naming_lint import lint_naming
@@ -438,6 +440,32 @@ async def sensitive_columns(
     )
 
 
+@router.get(
+    "/{schema_snapshot_uuid}/constraint-inventory",
+    response_model=ConstraintInventoryOut,
+)
+async def constraint_inventory(
+    schema_snapshot_uuid: uuid.UUID,
+    user: CurrentUser = Depends(get_current_user),
+    session: AsyncSession = Depends(get_read_session),
+) -> ConstraintInventoryOut:
+    """Inventory CHECK-constraint business rules and FK delete-action risks
+    (ON DELETE CASCADE = warning, SET NULL = info).
+
+    IDOR-safe (uniform not-found for missing/unauthorized snapshots).
+    """
+    snap = await _get_authorized_snapshot(session, schema_snapshot_uuid, user)
+    if snap is None:
+        return ConstraintInventoryOut(
+            schema_snapshot_uuid=schema_snapshot_uuid, status="not_found", report=None
+        )
+    data = await session.get(SchemaSnapshotData, schema_snapshot_uuid)
+    report = build_constraint_inventory(data.snapshot_json if data else None)
+    return ConstraintInventoryOut(
+        schema_snapshot_uuid=schema_snapshot_uuid, status="ok", report=report
+    )
+
+
 @router.get(
     "/{schema_snapshot_uuid}/data-dictionary.md", response_class=PlainTextResponse
 )

backend/app/schemas.py

@@ -163,6 +163,14 @@ class SensitiveColumnsOut(BaseModel):
     report: dict | None
 
 
+class ConstraintInventoryOut(BaseModel):
+    """CHECK-rule inventory and FK delete-action risks for a snapshot."""
+
+    schema_snapshot_uuid: uuid.UUID
+    status: str
+    report: dict | None
+
+
 class DiagramViewCreateIn(BaseModel):
     """Request body for saving an ERD canvas view."""
 

backend/app/spec/constraint_inventory.py

@@ -0,0 +1,94 @@
+"""Inventory business rules living in constraints, and flag CASCADE risk.
+
+Reverse engineering isn't just tables and columns -- the *business rules* often
+hide in CHECK constraints, and the operational hazards hide in FK referential
+actions. This inventories both:
+
+* **check_rules** -- every CHECK constraint with its expression: the closest
+  thing a legacy database has to documented invariants.
+* **cascade_deletes** (warning) -- ``ON DELETE CASCADE`` foreign keys: deleting
+  a parent silently deletes children; every operator should know these paths.
+* **set_null_deletes** (info) -- ``ON DELETE SET NULL``: orphaned-but-kept rows.
+
+Pure and dialect-agnostic (PostgreSQL action codes 'c'/'n' plus spelled-out
+variants are both handled).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+WARNING = "warning"
+INFO = "info"
+
+_CASCADE = {"c", "cascade"}
+_SET_NULL = {"n", "set null", "set_null"}
+
+
+def build_constraint_inventory(snapshot: dict[str, Any] | None) -> dict[str, Any]:
+    """Return CHECK-rule inventory and FK delete-action risk findings."""
+    snapshot = snapshot or {}
+    constraints = snapshot.get("constraints") or []
+
+    check_rules: list[dict[str, Any]] = []
+    cascade_items: list[dict[str, Any]] = []
+
+    for con in constraints:
+        ctype = str(con.get("constraint_type") or "").lower()
+        table = f"{con.get('schema_name')}.{con.get('relation_name')}"
+        name = str(con.get("constraint_name") or "")
+
+        if ctype == "c":
+            expr = con.get("check_expr") or con.get("constraint_def")
+            check_rules.append(
+                {
+                    "table": table,
+                    "constraint": name,
+                    "expression": str(expr or ""),
+                }
+            )
+        elif ctype == "f":
+            on_delete = str(con.get("fk_on_delete") or "").lower()
+            target = (
+                f"{con.get('foreign_schema_name')}.{con.get('foreign_relation_name')}"
+            )
+            if on_delete in _CASCADE:
+                cascade_items.append(
+                    {
+                        "category": "cascade_delete",
+                        "severity": WARNING,
+                        "table": table,
+                        "constraint": name,
+                        "references": target,
+                        "detail": (
+                            f"Deleting a row in {target} silently deletes rows in "
+                            f"{table} (ON DELETE CASCADE via '{name}')."
+                        ),
+                    }
+                )
+            elif on_delete in _SET_NULL:
+                cascade_items.append(
+                    {
+                        "category": "set_null_delete",
+                        "severity": INFO,
+                        "table": table,
+                        "constraint": name,
+                        "references": target,
+                        "detail": (
+                            f"Deleting a row in {target} nulls the reference in "
+                            f"{table} (ON DELETE SET NULL via '{name}') — rows are kept but orphaned."
+                        ),
+                    }
+                )
+
+    check_rules.sort(key=lambda r: (r["table"], r["constraint"]))
+    cascade_items.sort(
+        key=lambda i: (0 if i["severity"] == WARNING else 1, i["table"], i["constraint"])
+    )
+
+    summary = {
+        "check_rules": len(check_rules),
+        "cascade_deletes": sum(1 for i in cascade_items if i["category"] == "cascade_delete"),
+        "set_null_deletes": sum(1 for i in cascade_items if i["category"] == "set_null_delete"),
+    }
+    return {"check_rules": check_rules, "delete_actions": cascade_items, "summary": summary}

backend/tests/test_constraint_inventory.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from app.spec.constraint_inventory import build_constraint_inventory
+
+
+def _con(ctype, name, table="orders", ftable=None, on_delete=None, check_expr=None):
+    return {
+        "constraint_type": ctype,
+        "constraint_name": name,
+        "schema_name": "public",
+        "relation_name": table,
+        "foreign_schema_name": "public" if ftable else None,
+        "foreign_relation_name": ftable,
+        "fk_on_delete": on_delete,
+        "check_expr": check_expr,
+        "constraint_def": f"DEF {name}",
+    }
+
+
+def test_inventories_check_constraints_with_expressions():
+    snap = {"constraints": [
+        _con("c", "chk_qty_positive", check_expr="(quantity > 0)"),
+        _con("c", "chk_status", table="member", check_expr="(status IN ('active','banned'))"),
+        _con("p", "pk_orders"),
+    ]}
+    inv = build_constraint_inventory(snap)
+    assert inv["summary"]["check_rules"] == 2
+    assert inv["check_rules"][0]["table"] == "public.member"  # sorted by table
+    assert inv["check_rules"][1]["expression"] == "(quantity > 0)"
+
+
+def test_flags_cascade_delete_as_warning():
+    snap = {"constraints": [
+        _con("f", "fk_item_order", table="order_item", ftable="orders", on_delete="c"),
+    ]}
+    inv = build_constraint_inventory(snap)
+    assert inv["summary"]["cascade_deletes"] == 1
+    item = inv["delete_actions"][0]
+    assert item["severity"] == "warning"
+    assert "public.orders" in item["detail"] and "public.order_item" in item["detail"]
+
+
+def test_set_null_is_info_and_spelled_out_codes_work():
+    snap = {"constraints": [
+        _con("f", "fk_a", table="a", ftable="b", on_delete="set null"),
+        _con("f", "fk_c", table="c", ftable="d", on_delete="CASCADE"),
+        _con("f", "fk_plain", table="e", ftable="f", on_delete="a"),  # no action
+    ]}
+    inv = build_constraint_inventory(snap)
+    assert inv["summary"]["set_null_deletes"] == 1
+    assert inv["summary"]["cascade_deletes"] == 1
+    # warnings sort before infos
+    assert [i["severity"] for i in inv["delete_actions"]] == ["warning", "info"]
+
+
+def test_check_without_expr_falls_back_to_def_and_empty():
+    snap = {"constraints": [_con("c", "chk_x", check_expr=None)]}
+    inv = build_constraint_inventory(snap)
+    assert inv["check_rules"][0]["expression"] == "DEF chk_x"
+    assert build_constraint_inventory({})["summary"]["check_rules"] == 0