feat(health): circular foreign-key dependency detection

backend/app/api/snapshots.py

@@ -18,12 +18,14 @@
 )
 from app.permissions import require_project_member
 from app.schemas import (
+    FkCyclesOut,
     SnapshotCreateIn,
     SnapshotDetailOut,
     SnapshotOut,
     WideTablesOut,
 )
 from app.ddl.export import snapshot_json_to_sql
+from app.spec.fk_cycles import detect_fk_cycles
 from app.spec.wide_tables import detect_wide_tables
 from app.jobs.valkey_queue import enqueue_job_signal
 from app.spec.llm import (
@@ -279,3 +281,26 @@ async def list_snapshots(
         )
         for s in snaps
     ]
+
+
+@router.get("/{schema_snapshot_uuid}/fk-cycles", response_model=FkCyclesOut)
+async def fk_cycles(
+    schema_snapshot_uuid: uuid.UUID,
+    user: CurrentUser = Depends(get_current_user),
+    session: AsyncSession = Depends(get_read_session),
+) -> FkCyclesOut:
+    """Report circular foreign-key dependencies (migration-ordering hazards).
+
+    Multi-table cycles are warnings; self-references are informational.
+    IDOR-safe (uniform not-found for missing/unauthorized snapshots).
+    """
+    snap = await _get_authorized_snapshot(session, schema_snapshot_uuid, user)
+    if snap is None:
+        return FkCyclesOut(
+            schema_snapshot_uuid=schema_snapshot_uuid, status="not_found", report=None
+        )
+    data = await session.get(SchemaSnapshotData, schema_snapshot_uuid)
+    report = detect_fk_cycles(data.snapshot_json if data else None)
+    return FkCyclesOut(
+        schema_snapshot_uuid=schema_snapshot_uuid, status="ok", report=report
+    )

backend/app/schemas.py

@@ -105,3 +105,11 @@ class MeOut(BaseModel):
     user_account_uuid: uuid.UUID
     subject: str
     display_name: str | None
+
+
+class FkCyclesOut(BaseModel):
+    """Circular foreign-key dependency findings for a snapshot."""
+
+    schema_snapshot_uuid: uuid.UUID
+    status: str
+    report: dict | None

backend/app/spec/fk_cycles.py

@@ -0,0 +1,129 @@
+"""Detect circular foreign-key dependencies in a snapshot.
+
+A cycle in the FK graph (A → B → … → A) means there's no order in which the
+tables can be created, dropped, or bulk-loaded without deferring constraints --
+a real migration hazard. Self-references (``employee.manager_id → employee``)
+are a normal, benign special case and are reported separately.
+
+Pure and dialect-agnostic. Uses an iterative Tarjan SCC so a deep FK chain can't
+blow the recursion limit.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+WARNING = "warning"
+INFO = "info"
+
+
+def _sccs(nodes: list[str], adj: dict[str, set[str]]) -> list[list[str]]:
+    """Tarjan's strongly-connected components (iterative)."""
+    index: dict[str, int] = {}
+    low: dict[str, int] = {}
+    on_stack: set[str] = set()
+    stack: list[str] = []
+    result: list[list[str]] = []
+    counter = 0
+
+    for root in nodes:
+        if root in index:
+            continue
+        index[root] = low[root] = counter
+        counter += 1
+        stack.append(root)
+        on_stack.add(root)
+        work = [(root, iter(sorted(adj.get(root, set()))))]
+        while work:
+            v, it = work[-1]
+            pushed = False
+            for w in it:
+                if w not in index:
+                    index[w] = low[w] = counter
+                    counter += 1
+                    stack.append(w)
+                    on_stack.add(w)
+                    work.append((w, iter(sorted(adj.get(w, set())))))
+                    pushed = True
+                    break
+                if w in on_stack:
+                    low[v] = min(low[v], index[w])
+            if pushed:
+                continue
+            if low[v] == index[v]:
+                comp = []
+                while True:
+                    w = stack.pop()
+                    on_stack.discard(w)
+                    comp.append(w)
+                    if w == v:
+                        break
+                result.append(comp)
+            work.pop()
+            if work:
+                low[work[-1][0]] = min(low[work[-1][0]], low[v])
+    return result
+
+
+def detect_fk_cycles(snapshot: dict[str, Any] | None) -> dict[str, Any]:
+    """Return circular-FK findings (multi-table cycles + self-references)."""
+    snapshot = snapshot or {}
+    relations = snapshot.get("relations") or []
+    fk_edges = snapshot.get("fk_edges") or []
+
+    def _qname(oid: Any) -> str | None:
+        rel = rel_by_oid.get(oid)
+        if rel is None:
+            return None
+        return f"{rel.get('schema_name')}.{rel.get('relation_name')}"
+
+    rel_by_oid = {r.get("relation_oid"): r for r in relations}
+
+    adj: dict[str, set[str]] = {}
+    nodes: set[str] = set()
+    self_refs: set[str] = set()
+    for edge in fk_edges:
+        child = _qname(edge.get("child_relation_oid"))
+        parent = _qname(edge.get("parent_relation_oid"))
+        if child is None or parent is None:
+            continue
+        nodes.add(child)
+        nodes.add(parent)
+        if child == parent:
+            self_refs.add(child)
+        else:
+            adj.setdefault(child, set()).add(parent)
+
+    items: list[dict[str, Any]] = []
+    for comp in _sccs(sorted(nodes), adj):
+        if len(comp) >= 2:
+            tables = sorted(comp)
+            items.append(
+                {
+                    "category": "circular_dependency",
+                    "severity": WARNING,
+                    "tables": tables,
+                    "detail": (
+                        "Circular foreign-key dependency: "
+                        + " → ".join(tables)
+                        + " — no create/drop/load order works without deferring constraints."
+                    ),
+                }
+            )
+    for table in sorted(self_refs):
+        items.append(
+            {
+                "category": "self_reference",
+                "severity": INFO,
+                "tables": [table],
+                "detail": f"{table} references itself (hierarchical/tree) — usually fine; load roots first.",
+            }
+        )
+
+    items.sort(key=lambda i: (0 if i["severity"] == WARNING else 1, i["tables"]))
+    summary = {
+        "circular_dependencies": sum(1 for i in items if i["category"] == "circular_dependency"),
+        "self_references": len(self_refs),
+        "total": len(items),
+    }
+    return {"items": items, "summary": summary}

backend/tests/test_fk_cycles.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from app.spec.fk_cycles import detect_fk_cycles
+
+
+def _snap(edges, names=None):
+    names = names or sorted({e[0] for e in edges} | {e[1] for e in edges})
+    oid = {n: i + 1 for i, n in enumerate(names)}
+    return {
+        "relations": [{"relation_oid": oid[n], "schema_name": "public", "relation_name": n} for n in names],
+        "fk_edges": [
+            {"child_relation_oid": oid[c], "parent_relation_oid": oid[p], "child_column_name": "x", "parent_column_name": "y"}
+            for c, p in edges
+        ],
+    }
+
+
+def test_detects_multi_table_cycle():
+    # a -> b -> c -> a
+    report = detect_fk_cycles(_snap([("a", "b"), ("b", "c"), ("c", "a")]))
+    assert report["summary"]["circular_dependencies"] == 1
+    cyc = next(i for i in report["items"] if i["category"] == "circular_dependency")
+    assert set(cyc["tables"]) == {"public.a", "public.b", "public.c"}
+    assert cyc["severity"] == "warning"
+
+
+def test_two_node_cycle():
+    report = detect_fk_cycles(_snap([("a", "b"), ("b", "a")]))
+    assert report["summary"]["circular_dependencies"] == 1
+
+
+def test_self_reference_is_info_not_a_cycle():
+    report = detect_fk_cycles(_snap([("employee", "employee")], names=["employee"]))
+    assert report["summary"]["circular_dependencies"] == 0
+    assert report["summary"]["self_references"] == 1
+    assert report["items"][0]["severity"] == "info"
+
+
+def test_acyclic_graph_has_no_findings():
+    report = detect_fk_cycles(_snap([("orders", "member"), ("order_item", "orders")]))
+    assert report["items"] == []
+
+
+def test_two_independent_cycles_and_deep_chain_no_recursion_error():
+    edges = [("a", "b"), ("b", "a"), ("c", "d"), ("d", "c")]
+    # long acyclic chain to exercise the iterative SCC
+    chain = [f"n{i}" for i in range(1500)]
+    edges += [(chain[i], chain[i + 1]) for i in range(len(chain) - 1)]
+    report = detect_fk_cycles(_snap(edges))
+    assert report["summary"]["circular_dependencies"] == 2