"""Four-level ontology and data-governance application service.

The storage model is intentionally graph-shaped but uses the existing SQLite
abstraction so a desktop deployment works without an external graph database.
Elements are graph nodes; important relations are reified assertions carrying
provenance, evidence, validity and review state.  This keeps the API portable
to RDF/Neo4j later without making either one a prerequisite for the MVP.
"""
from __future__ import annotations

import hashlib
import json
from collections import defaultdict, deque
from typing import Any, Dict, Iterable, List, Optional

from agentpaas.db.models import gen_id, now_utc


LAYERS = ("M3", "M2", "M1", "M0")
PROFILES = ("meta", "upper", "business", "software", "governance", "cross")
ELEMENT_STATUSES = ("candidate", "published", "deprecated", "retired")
ASSERTION_TYPES = ("declared", "extracted", "observed", "inferred")
REVIEW_STATUSES = ("pending", "accepted", "rejected", "expired")
SEVERITIES = ("low", "medium", "high", "critical")

_LAYER_RANK = {"M0": 0, "M1": 1, "M2": 2, "M3": 3}
_JSON_FIELDS = {
    "ontology_elements": ("attributes",),
    "governance_evidence": ("metadata",),
    "governance_assertions": ("evidence_ids",),
    "governance_rules": ("expression",),
    "governance_issues": ("evidence_ids",),
    "governance_agent_runs": ("result_json", "trace_json"),
}


class GovernanceError(ValueError):
    """A domain validation or tenant-scope failure."""

    def __init__(self, message: str, *, code: str = "INVALID_GOVERNANCE_DATA", status: int = 400):
        super().__init__(message)
        self.code = code
        self.status = status


def _json_load(value: Any, default: Any) -> Any:
    if value in (None, ""):
        return default
    if isinstance(value, (dict, list)):
        return value
    try:
        return json.loads(value)
    except (TypeError, ValueError):
        return default


def _row(table: str, value: Optional[dict]) -> Optional[dict]:
    if value is None:
        return None
    out = dict(value)
    for field in _JSON_FIELDS.get(table, ()):
        default = [] if field.endswith("_ids") or field == "trace_json" else {}
        out[field] = _json_load(out.get(field), default)
    if table == "governance_rules":
        out["enabled"] = bool(out.get("enabled"))
    return out


def _rows(table: str, values: Iterable[dict]) -> List[dict]:
    return [_row(table, value) for value in values]  # type: ignore[list-item]


def _dump(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True)


def require_project(db, tenant_id: str, project_id: str) -> dict:
    project = db.fetchone(
        "SELECT * FROM governance_projects WHERE id=? AND tenant_id=?",
        (project_id, tenant_id),
    )
    if not project:
        raise GovernanceError("本体项目不存在", code="PROJECT_NOT_FOUND", status=404)
    return project


def create_project(db, tenant_id: str, data: Dict[str, Any]) -> dict:
    name = str(data.get("name", "")).strip()
    namespace = str(data.get("namespace", "")).strip().rstrip("#/")
    if not name or not namespace:
        raise GovernanceError("name 和 namespace 不能为空")
    existing = db.fetchone(
        "SELECT * FROM governance_projects WHERE tenant_id=? AND namespace=?",
        (tenant_id, namespace),
    )
    if existing:
        return existing
    project_id = gen_id("gp_")
    now = now_utc()
    db.execute(
        "INSERT INTO governance_projects "
        "(id,tenant_id,name,description,namespace,version,status,created_at,updated_at) "
        "VALUES (?,?,?,?,?,?,?,?,?)",
        (
            project_id, tenant_id, name, str(data.get("description", "")), namespace,
            str(data.get("version", "1.0.0")), str(data.get("status", "active")), now, now,
        ),
    )
    db.commit()
    return require_project(db, tenant_id, project_id)


def list_projects(db, tenant_id: str) -> List[dict]:
    return db.fetchall(
        "SELECT * FROM governance_projects WHERE tenant_id=? ORDER BY updated_at DESC",
        (tenant_id,),
    )


def upsert_element(db, tenant_id: str, project_id: str, data: Dict[str, Any]) -> dict:
    project = require_project(db, tenant_id, project_id)
    layer = str(data.get("layer", "")).upper()
    profile = str(data.get("profile", "")).lower()
    status = str(data.get("status", "candidate")).lower()
    name = str(data.get("name", "")).strip()
    kind = str(data.get("kind", "")).strip()
    if layer not in LAYERS:
        raise GovernanceError(f"未知元建模层级: {layer}")
    if profile not in PROFILES:
        raise GovernanceError(f"未知 Profile: {profile}")
    if status not in ELEMENT_STATUSES:
        raise GovernanceError(f"未知元素状态: {status}")
    if not name or not kind:
        raise GovernanceError("元素 name 和 kind 不能为空")
    if layer == "M3" and profile != "meta":
        raise GovernanceError("M3 元素必须属于 meta Profile")
    if layer == "M2" and profile == "meta":
        raise GovernanceError("M2 元素应属于 upper 或领域 Profile")

    uri = str(data.get("uri", "")).strip()
    if not uri:
        uri = f"{project['namespace']}/{layer.lower()}/{profile}/{name}"
    existing = db.fetchone(
        "SELECT id FROM ontology_elements WHERE tenant_id=? AND project_id=? AND uri=?",
        (tenant_id, project_id, uri),
    )
    now = now_utc()
    attrs = _dump(data.get("attributes") or {})
    if existing:
        element_id = existing["id"]
        db.execute(
            "UPDATE ontology_elements SET layer=?,profile=?,kind=?,name=?,description=?,"
            "status=?,version=?,attributes=?,source=?,updated_at=? "
            "WHERE id=? AND tenant_id=?",
            (
                layer, profile, kind, name, str(data.get("description", "")), status,
                str(data.get("version", "1.0.0")), attrs, str(data.get("source", "")),
                now, element_id, tenant_id,
            ),
        )
    else:
        element_id = str(data.get("id") or gen_id("oe_"))
        db.execute(
            "INSERT INTO ontology_elements "
            "(id,tenant_id,project_id,uri,layer,profile,kind,name,description,status,"
            "version,attributes,source,created_at,updated_at) "
            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                element_id, tenant_id, project_id, uri, layer, profile, kind, name,
                str(data.get("description", "")), status,
                str(data.get("version", "1.0.0")), attrs,
                str(data.get("source", "")), now, now,
            ),
        )
    db.execute("UPDATE governance_projects SET updated_at=? WHERE id=?", (now, project_id))
    db.commit()
    return get_element(db, tenant_id, element_id)


def get_element(db, tenant_id: str, element_id: str) -> dict:
    value = db.fetchone(
        "SELECT * FROM ontology_elements WHERE id=? AND tenant_id=?", (element_id, tenant_id)
    )
    if not value:
        raise GovernanceError("本体元素不存在", code="ELEMENT_NOT_FOUND", status=404)
    return _row("ontology_elements", value)  # type: ignore[return-value]


def list_elements(
    db, tenant_id: str, project_id: str, *, layer: str = "", profile: str = "",
    status: str = "", query: str = "", limit: int = 200, offset: int = 0,
) -> Dict[str, Any]:
    require_project(db, tenant_id, project_id)
    where = ["tenant_id=?", "project_id=?"]
    params: List[Any] = [tenant_id, project_id]
    if layer:
        where.append("layer=?")
        params.append(layer.upper())
    if profile:
        where.append("profile=?")
        params.append(profile.lower())
    if status:
        where.append("status=?")
        params.append(status.lower())
    if query:
        where.append("(name LIKE ? OR uri LIKE ? OR description LIKE ?)")
        like = f"%{query}%"
        params.extend((like, like, like))
    clause = " AND ".join(where)
    total = db.fetchone(f"SELECT COUNT(*) AS n FROM ontology_elements WHERE {clause}", tuple(params))["n"]
    bounded = min(max(int(limit), 1), 500)
    values = db.fetchall(
        f"SELECT * FROM ontology_elements WHERE {clause} "
        "ORDER BY CASE layer WHEN 'M3' THEN 0 WHEN 'M2' THEN 1 WHEN 'M1' THEN 2 ELSE 3 END, "
        "profile,name LIMIT ? OFFSET ?",
        tuple(params + [bounded, max(int(offset), 0)]),
    )
    return {"items": _rows("ontology_elements", values), "total": total}


def create_evidence(db, tenant_id: str, project_id: str, data: Dict[str, Any]) -> dict:
    require_project(db, tenant_id, project_id)
    title = str(data.get("title", "")).strip()
    kind = str(data.get("kind", "SourceArtifact")).strip()
    if not title:
        raise GovernanceError("证据 title 不能为空")
    uri = str(data.get("uri", "")).strip()
    if uri:
        existing = db.fetchone(
            "SELECT * FROM governance_evidence WHERE tenant_id=? AND project_id=? AND uri=?",
            (tenant_id, project_id, uri),
        )
        if existing:
            return _row("governance_evidence", existing)  # type: ignore[return-value]
    content = str(data.get("content", ""))
    evidence_id = gen_id("ev_")
    now = now_utc()
    db.execute(
        "INSERT INTO governance_evidence "
        "(id,tenant_id,project_id,kind,uri,title,content,content_hash,source,observed_at,"
        "valid_from,valid_to,metadata,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
        (
            evidence_id, tenant_id, project_id, kind, uri, title, content,
            str(data.get("content_hash") or hashlib.sha256(content.encode("utf-8")).hexdigest()),
            str(data.get("source", "")), str(data.get("observed_at") or now),
            data.get("valid_from") or now, data.get("valid_to"),
            _dump(data.get("metadata") or {}), now,
        ),
    )
    db.commit()
    return _row(
        "governance_evidence",
        db.fetchone("SELECT * FROM governance_evidence WHERE id=?", (evidence_id,)),
    )  # type: ignore[return-value]


def list_evidence(db, tenant_id: str, project_id: str, limit: int = 200) -> List[dict]:
    require_project(db, tenant_id, project_id)
    return _rows(
        "governance_evidence",
        db.fetchall(
            "SELECT * FROM governance_evidence WHERE tenant_id=? AND project_id=? "
            "ORDER BY observed_at DESC LIMIT ?",
            (tenant_id, project_id, min(max(limit, 1), 500)),
        ),
    )


def create_assertion(db, tenant_id: str, project_id: str, data: Dict[str, Any]) -> dict:
    require_project(db, tenant_id, project_id)
    subject = get_element(db, tenant_id, str(data.get("subject_id", "")))
    if subject["project_id"] != project_id:
        raise GovernanceError("断言 subject 不属于当前项目")
    object_id = str(data.get("object_id") or "")
    if object_id:
        obj = get_element(db, tenant_id, object_id)
        if obj["project_id"] != project_id:
            raise GovernanceError("断言 object 不属于当前项目")
    predicate = str(data.get("predicate", "")).strip().upper()
    assertion_type = str(data.get("assertion_type", "inferred")).lower()
    review_status = str(data.get("review_status", "pending")).lower()
    confidence = float(data.get("confidence", 0.0))
    if not predicate:
        raise GovernanceError("predicate 不能为空")
    if assertion_type not in ASSERTION_TYPES:
        raise GovernanceError(f"未知断言类型: {assertion_type}")
    if review_status not in REVIEW_STATUSES:
        raise GovernanceError(f"未知审核状态: {review_status}")
    if not 0.0 <= confidence <= 1.0:
        raise GovernanceError("confidence 必须在 0 到 1 之间")
    evidence_ids = list(dict.fromkeys(data.get("evidence_ids") or []))
    for evidence_id in evidence_ids:
        if not db.fetchone(
            "SELECT id FROM governance_evidence WHERE id=? AND tenant_id=? AND project_id=?",
            (evidence_id, tenant_id, project_id),
        ):
            raise GovernanceError(f"证据不存在: {evidence_id}")

    existing = db.fetchone(
        "SELECT * FROM governance_assertions WHERE tenant_id=? AND project_id=? "
        "AND subject_id=? AND predicate=? AND COALESCE(object_id,'')=? "
        "AND COALESCE(object_value,'')=?",
        (
            tenant_id, project_id, subject["id"], predicate, object_id,
            str(data.get("object_value", "")),
        ),
    )
    if existing:
        return _row("governance_assertions", existing)  # type: ignore[return-value]
    assertion_id = gen_id("as_")
    now = now_utc()
    db.execute(
        "INSERT INTO governance_assertions "
        "(id,tenant_id,project_id,subject_id,predicate,object_id,object_value,"
        "assertion_type,confidence,evidence_ids,generated_by,valid_from,valid_to,"
        "review_status,reviewer,review_note,created_at,updated_at) "
        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
        (
            assertion_id, tenant_id, project_id, subject["id"], predicate,
            object_id or None, str(data.get("object_value", "")), assertion_type,
            confidence, _dump(evidence_ids), str(data.get("generated_by", "")),
            data.get("valid_from") or now, data.get("valid_to"), review_status,
            str(data.get("reviewer", "")), str(data.get("review_note", "")), now, now,
        ),
    )
    db.commit()
    return get_assertion(db, tenant_id, assertion_id)


def get_assertion(db, tenant_id: str, assertion_id: str) -> dict:
    value = db.fetchone(
        "SELECT * FROM governance_assertions WHERE id=? AND tenant_id=?",
        (assertion_id, tenant_id),
    )
    if not value:
        raise GovernanceError("断言不存在", code="ASSERTION_NOT_FOUND", status=404)
    return _row("governance_assertions", value)  # type: ignore[return-value]


def list_assertions(
    db, tenant_id: str, project_id: str, *, review_status: str = "",
    assertion_type: str = "", limit: int = 300,
) -> List[dict]:
    require_project(db, tenant_id, project_id)
    where = ["a.tenant_id=?", "a.project_id=?"]
    params: List[Any] = [tenant_id, project_id]
    if review_status:
        where.append("a.review_status=?")
        params.append(review_status)
    if assertion_type:
        where.append("a.assertion_type=?")
        params.append(assertion_type)
    params.append(min(max(limit, 1), 500))
    values = db.fetchall(
        "SELECT a.*, s.name AS subject_name, o.name AS object_name "
        "FROM governance_assertions a "
        "JOIN ontology_elements s ON s.id=a.subject_id "
        "LEFT JOIN ontology_elements o ON o.id=a.object_id "
        f"WHERE {' AND '.join(where)} ORDER BY a.updated_at DESC LIMIT ?",
        tuple(params),
    )
    return _rows("governance_assertions", values)


def review_assertion(
    db, tenant_id: str, assertion_id: str, status: str, reviewer: str, note: str = "",
) -> dict:
    assertion = get_assertion(db, tenant_id, assertion_id)
    status = status.lower()
    if status not in ("accepted", "rejected", "expired"):
        raise GovernanceError("审核结果必须是 accepted、rejected 或 expired")
    if not reviewer.strip():
        raise GovernanceError("reviewer 不能为空")
    now = now_utc()
    db.execute(
        "UPDATE governance_assertions SET review_status=?,reviewer=?,review_note=?,updated_at=? "
        "WHERE id=? AND tenant_id=?",
        (status, reviewer.strip(), note, now, assertion_id, tenant_id),
    )
    db.commit()
    return get_assertion(db, tenant_id, assertion_id)


def upsert_rule(db, tenant_id: str, project_id: str, data: Dict[str, Any]) -> dict:
    require_project(db, tenant_id, project_id)
    name = str(data.get("name", "")).strip()
    rule_type = str(data.get("rule_type", "integrity")).strip()
    severity = str(data.get("severity", "medium")).lower()
    if not name:
        raise GovernanceError("规则 name 不能为空")
    if severity not in SEVERITIES:
        raise GovernanceError(f"未知严重级别: {severity}")
    existing = db.fetchone(
        "SELECT id FROM governance_rules WHERE tenant_id=? AND project_id=? AND name=?",
        (tenant_id, project_id, name),
    )
    now = now_utc()
    if existing:
        rule_id = existing["id"]
        db.execute(
            "UPDATE governance_rules SET description=?,rule_type=?,target_layer=?,"
            "target_profile=?,target_kind=?,expression=?,severity=?,enabled=?,version=?,updated_at=? "
            "WHERE id=? AND tenant_id=?",
            (
                str(data.get("description", "")), rule_type,
                str(data.get("target_layer", "")), str(data.get("target_profile", "")),
                str(data.get("target_kind", "")), _dump(data.get("expression") or {}),
                severity, 1 if data.get("enabled", True) else 0,
                str(data.get("version", "1.0.0")), now, rule_id, tenant_id,
            ),
        )
    else:
        rule_id = gen_id("gr_")
        db.execute(
            "INSERT INTO governance_rules "
            "(id,tenant_id,project_id,name,description,rule_type,target_layer,target_profile,"
            "target_kind,expression,severity,enabled,version,created_at,updated_at) "
            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                rule_id, tenant_id, project_id, name, str(data.get("description", "")),
                rule_type, str(data.get("target_layer", "")),
                str(data.get("target_profile", "")), str(data.get("target_kind", "")),
                _dump(data.get("expression") or {}), severity,
                1 if data.get("enabled", True) else 0,
                str(data.get("version", "1.0.0")), now, now,
            ),
        )
    db.commit()
    return _row(
        "governance_rules", db.fetchone("SELECT * FROM governance_rules WHERE id=?", (rule_id,))
    )  # type: ignore[return-value]


def list_rules(db, tenant_id: str, project_id: str) -> List[dict]:
    require_project(db, tenant_id, project_id)
    return _rows(
        "governance_rules",
        db.fetchall(
            "SELECT * FROM governance_rules WHERE tenant_id=? AND project_id=? ORDER BY severity DESC,name",
            (tenant_id, project_id),
        ),
    )


def _finding(code: str, severity: str, title: str, description: str, target_id: str = "", assertion_id: str = "") -> dict:
    return {
        "code": code,
        "severity": severity,
        "title": title,
        "description": description,
        "target_id": target_id,
        "assertion_id": assertion_id,
    }


def validate_project(db, tenant_id: str, project_id: str, *, persist: bool = True) -> dict:
    """Evaluate the first deterministic SHACL-equivalent governance gates."""
    require_project(db, tenant_id, project_id)
    elements = _rows(
        "ontology_elements",
        db.fetchall(
            "SELECT * FROM ontology_elements WHERE tenant_id=? AND project_id=?",
            (tenant_id, project_id),
        ),
    )
    assertions = _rows(
        "governance_assertions",
        db.fetchall(
            "SELECT * FROM governance_assertions WHERE tenant_id=? AND project_id=?",
            (tenant_id, project_id),
        ),
    )
    by_id = {item["id"]: item for item in elements}
    outgoing: Dict[str, List[dict]] = defaultdict(list)
    incoming: Dict[str, List[dict]] = defaultdict(list)
    findings: List[dict] = []
    for assertion in assertions:
        outgoing[assertion["subject_id"]].append(assertion)
        if assertion.get("object_id"):
            incoming[assertion["object_id"]].append(assertion)

        subject = by_id.get(assertion["subject_id"])
        obj = by_id.get(assertion.get("object_id"))
        predicate = assertion["predicate"]
        if predicate == "INSTANCE_OF" and subject and obj:
            if _LAYER_RANK[obj["layer"]] != _LAYER_RANK[subject["layer"]] + 1:
                findings.append(_finding(
                    "LAYER_INSTANCE_MISMATCH", "high", "instanceOf 层级不连续",
                    f"{subject['name']}({subject['layer']}) 必须实例化相邻上层，当前指向 {obj['name']}({obj['layer']})",
                    subject["id"], assertion["id"],
                ))
        if predicate in ("SUB_TYPE_OF", "SUBCLASS_OF") and subject and obj and subject["layer"] != obj["layer"]:
            findings.append(_finding(
                "SUBTYPE_CROSSES_LAYER", "high", "subTypeOf 不能跨元层",
                f"{subject['name']} 与 {obj['name']} 不在同一层", subject["id"], assertion["id"],
            ))
        if assertion["assertion_type"] == "inferred" and predicate in {
            "DELETE", "GRANT", "AUTHORIZE", "DOWNGRADE", "PUBLISH", "RETIRE",
        }:
            findings.append(_finding(
                "INFERRED_HIGH_RISK_ACTION", "critical", "推断断言不能触发高风险动作",
                f"{predicate} 必须经过确定性证据与人工审批", assertion["subject_id"], assertion["id"],
            ))
        if assertion["review_status"] == "accepted" and assertion["assertion_type"] != "declared":
            if not assertion.get("evidence_ids") or not assertion.get("generated_by"):
                findings.append(_finding(
                    "ACCEPTED_ASSERTION_WITHOUT_PROVENANCE", "high", "正式断言缺少证据或生成者",
                    "accepted 的 extracted/observed/inferred 断言必须关联证据并记录 generated_by",
                    assertion["subject_id"], assertion["id"],
                ))

    for element in elements:
        if element["layer"] == "M3" and element["profile"] != "meta":
            findings.append(_finding(
                "M3_DOMAIN_LEAK", "critical", "M3 混入领域概念",
                "M3 只能包含稳定的 meta 建模内核", element["id"],
            ))
        accepted_out = [a for a in outgoing[element["id"]] if a["review_status"] == "accepted"]
        accepted_in = [a for a in incoming[element["id"]] if a["review_status"] == "accepted"]
        if element["layer"] == "M0" and element["status"] == "published":
            if not any(a["predicate"] == "INSTANCE_OF" for a in accepted_out):
                findings.append(_finding(
                    "M0_UNKNOWN_TYPE", "high", "M0 事实未归类",
                    "已发布的 M0 事实必须实例化已发布的 M1 类型", element["id"],
                ))
        if element["layer"] == "M1" and element["profile"] == "business":
            if element["kind"] == "BusinessCapability" and not any(a["predicate"] == "REALIZED_BY" for a in accepted_out):
                findings.append(_finding(
                    "CAPABILITY_NOT_REALIZED", "high", "业务能力缺少软件实现",
                    f"{element['name']} 至少需要一个 REALIZED_BY 关系", element["id"],
                ))
            if element["kind"] == "BusinessObject" and not any(a["predicate"] == "REPRESENTS" for a in accepted_in):
                findings.append(_finding(
                    "BUSINESS_OBJECT_NOT_REPRESENTED", "high", "业务对象缺少技术表达",
                    f"{element['name']} 至少应由类、表、API Schema 或事件契约表达", element["id"],
                ))
        attrs = element.get("attributes") or {}
        if element["profile"] == "software" and attrs.get("environment") == "production":
            if not any(a["predicate"] == "OWNED_BY" for a in accepted_out):
                findings.append(_finding(
                    "PRODUCTION_ASSET_WITHOUT_OWNER", "critical", "生产资产缺少责任人",
                    f"{element['name']} 必须 OWNED_BY 某个 Owner/Steward", element["id"],
                ))
        if element["kind"] in ("Column", "PhysicalColumn") and attrs.get("sensitive"):
            policy_links = [a for a in accepted_out if a["predicate"] == "GOVERNED_BY"]
            if not any(by_id.get(a.get("object_id"), {}).get("kind") in ("AccessPolicy", "SecurityPolicy") for a in policy_links):
                findings.append(_finding(
                    "SENSITIVE_FIELD_WITHOUT_POLICY", "critical", "敏感字段缺少访问策略",
                    f"{element['name']} 必须关联分类分级和访问策略", element["id"],
                ))

    if persist:
        db.execute(
            "DELETE FROM governance_issues WHERE tenant_id=? AND project_id=? AND rule_id LIKE 'validation:%'",
            (tenant_id, project_id),
        )
        now = now_utc()
        for finding in findings:
            db.execute(
                "INSERT INTO governance_issues "
                "(id,tenant_id,project_id,rule_id,target_id,assertion_id,severity,title,description,"
                "status,evidence_ids,assigned_to,created_at,updated_at) "
                "VALUES (?,?,?,?,?,?,?,?,?,'open','[]','',?,?)",
                (
                    gen_id("gi_"), tenant_id, project_id, f"validation:{finding['code']}",
                    finding["target_id"], finding["assertion_id"], finding["severity"],
                    finding["title"], finding["description"], now, now,
                ),
            )
        db.commit()
    severity_counts = {severity: 0 for severity in SEVERITIES}
    for finding in findings:
        severity_counts[finding["severity"]] += 1
    return {
        "valid": not any(item["severity"] in ("critical", "high") for item in findings),
        "findings": findings,
        "counts": severity_counts,
        "checked": {"elements": len(elements), "assertions": len(assertions)},
    }


def list_issues(db, tenant_id: str, project_id: str, status: str = "open") -> List[dict]:
    require_project(db, tenant_id, project_id)
    where = "tenant_id=? AND project_id=?"
    params: List[Any] = [tenant_id, project_id]
    if status:
        where += " AND status=?"
        params.append(status)
    return _rows(
        "governance_issues",
        db.fetchall(
            f"SELECT * FROM governance_issues WHERE {where} "
            "ORDER BY CASE severity WHEN 'critical' THEN 0 WHEN 'high' THEN 1 "
            "WHEN 'medium' THEN 2 ELSE 3 END, updated_at DESC",
            tuple(params),
        ),
    )


def update_issue(db, tenant_id: str, issue_id: str, status: str, assigned_to: str = "") -> dict:
    if status not in ("open", "acknowledged", "resolved", "waived"):
        raise GovernanceError("未知问题状态")
    existing = db.fetchone(
        "SELECT * FROM governance_issues WHERE id=? AND tenant_id=?", (issue_id, tenant_id)
    )
    if not existing:
        raise GovernanceError("治理问题不存在", code="ISSUE_NOT_FOUND", status=404)
    now = now_utc()
    db.execute(
        "UPDATE governance_issues SET status=?,assigned_to=?,updated_at=?,resolved_at=? "
        "WHERE id=? AND tenant_id=?",
        (status, assigned_to, now, now if status in ("resolved", "waived") else None, issue_id, tenant_id),
    )
    db.commit()
    return _row(
        "governance_issues",
        db.fetchone("SELECT * FROM governance_issues WHERE id=?", (issue_id,)),
    )  # type: ignore[return-value]


def graph(db, tenant_id: str, project_id: str, *, include_pending: bool = True, limit: int = 500) -> dict:
    elements = list_elements(db, tenant_id, project_id, limit=limit)["items"]
    ids = {element["id"] for element in elements}
    statuses = ("accepted", "pending") if include_pending else ("accepted",)
    placeholders = ",".join("?" for _ in statuses)
    values = db.fetchall(
        "SELECT a.*, s.name AS subject_name, o.name AS object_name "
        "FROM governance_assertions a "
        "JOIN ontology_elements s ON s.id=a.subject_id AND s.tenant_id=a.tenant_id "
        "LEFT JOIN ontology_elements o ON o.id=a.object_id AND o.tenant_id=a.tenant_id "
        "WHERE a.tenant_id=? AND a.project_id=? "
        f"AND a.review_status IN ({placeholders}) ORDER BY a.created_at LIMIT ?",
        (tenant_id, project_id, *statuses, min(max(limit * 2, 1), 1000)),
    )
    edges = [
        item for item in _rows("governance_assertions", values)
        if item["subject_id"] in ids and (not item.get("object_id") or item["object_id"] in ids)
    ]
    return {"nodes": elements, "edges": edges}


def impact(db, tenant_id: str, element_id: str, max_depth: int = 4) -> dict:
    center = get_element(db, tenant_id, element_id)
    project_id = center["project_id"]
    elements = list_elements(db, tenant_id, project_id, limit=500)["items"]
    by_id = {element["id"]: element for element in elements}
    assertions = list_assertions(db, tenant_id, project_id, review_status="accepted", limit=500)
    adjacency: Dict[str, List[tuple[str, dict, str]]] = defaultdict(list)
    for edge in assertions:
        if not edge.get("object_id"):
            continue
        adjacency[edge["subject_id"]].append((edge["object_id"], edge, "out"))
        adjacency[edge["object_id"]].append((edge["subject_id"], edge, "in"))
    depth_limit = min(max(int(max_depth), 1), 8)
    queue = deque([(element_id, 0)])
    visited = {element_id: 0}
    used_edges: Dict[str, dict] = {}
    while queue:
        current, depth = queue.popleft()
        if depth >= depth_limit:
            continue
        for neighbour, edge, direction in adjacency[current]:
            used_edges[edge["id"]] = {**edge, "traversed_from": current, "direction": direction}
            if neighbour not in visited:
                visited[neighbour] = depth + 1
                queue.append((neighbour, depth + 1))
    nodes = [{**by_id[node_id], "distance": distance} for node_id, distance in visited.items() if node_id in by_id]
    nodes.sort(key=lambda item: (item["distance"], item["layer"], item["name"]))
    return {"center": center, "nodes": nodes, "edges": list(used_edges.values()), "max_depth": depth_limit}


def overview(db, tenant_id: str, project_id: str) -> dict:
    project = require_project(db, tenant_id, project_id)
    layer_counts = {layer: 0 for layer in LAYERS}
    for row in db.fetchall(
        "SELECT layer,COUNT(*) AS n FROM ontology_elements WHERE tenant_id=? AND project_id=? GROUP BY layer",
        (tenant_id, project_id),
    ):
        layer_counts[row["layer"]] = row["n"]
    profile_counts = {profile: 0 for profile in PROFILES}
    for row in db.fetchall(
        "SELECT profile,COUNT(*) AS n FROM ontology_elements WHERE tenant_id=? AND project_id=? GROUP BY profile",
        (tenant_id, project_id),
    ):
        profile_counts[row["profile"]] = row["n"]
    def count(table: str, extra: str = "", params: tuple = ()) -> int:
        row = db.fetchone(
            f"SELECT COUNT(*) AS n FROM {table} WHERE tenant_id=? AND project_id=? {extra}",
            (tenant_id, project_id, *params),
        )
        return int(row["n"])
    return {
        "project": project,
        "layers": layer_counts,
        "profiles": profile_counts,
        "totals": {
            "elements": sum(layer_counts.values()),
            "assertions": count("governance_assertions"),
            "pending_reviews": count("governance_assertions", "AND review_status='pending'"),
            "evidence": count("governance_evidence"),
            "rules": count("governance_rules", "AND enabled=1"),
            "open_issues": count("governance_issues", "AND status='open'"),
            "agent_runs": count("governance_agent_runs"),
        },
    }


def seed_tenant_isolation(db, tenant_id: str) -> dict:
    """Idempotently install the complete M3->M0 tenant-isolation slice."""
    project = create_project(db, tenant_id, {
        "name": "DCP 租户与组织域",
        "description": "OntoRefactor 四层元建模纵向样例：业务、软件、治理与运行事实统一建模",
        "namespace": "https://onto.qm.cn/ontorefactor/model/dcp/1.0",
        "version": "1.0.0",
    })
    pid = project["id"]
    nodes: Dict[str, dict] = {}

    def node(key: str, layer: str, profile: str, kind: str, name: str, description: str = "", attributes: Optional[dict] = None, uri: str = "") -> dict:
        value = upsert_element(db, tenant_id, pid, {
            "layer": layer, "profile": profile, "kind": kind, "name": name,
            "description": description, "attributes": attributes or {},
            "status": "published", "uri": uri,
            "source": "builtin:tenant-isolation",
        })
        nodes[key] = value
        return value

    # M3: minimal stable modeling language.
    for name in (
        "MetaElement", "MetaEntityType", "MetaRelationType", "MetaAttributeType",
        "MetaConstraintType", "MetaAssertionType", "MetaProfile", "MetaView",
    ):
        node(name, "M3", "meta", name, name, f"M3 本体建模内核：{name}", uri=f"https://onto.qm.cn/ontorefactor/meta/1.0#{name}")

    # M2 upper ontology.
    for name in ("Object", "Subject", "Activity", "Rule", "Result", "Evidence", "Assertion", "Term", "State", "TimeExtent"):
        node(name, "M2", "upper", "EntityType", name, f"M2 通用上层本体概念：{name}", uri=f"https://onto.qm.cn/ontorefactor/upper/1.0#{name}")

    profile_types = {
        "business": ("BusinessCapabilityType", "BusinessObjectType", "BusinessRuleType", "BusinessEventType"),
        "software": ("ModuleType", "ClassType", "ApiOperationType", "TableType", "ColumnType", "EventContractType"),
        "governance": ("OwnerType", "AccessPolicyType", "QualityRuleType", "QualityGateType"),
    }
    for profile, names in profile_types.items():
        for name in names:
            node(name, "M2", profile, "EntityType", name, f"M2 {profile} Profile 类型：{name}", uri=f"https://onto.qm.cn/ontorefactor/{profile}/1.0#{name}")

    # M1 unified DCP model.
    m1 = (
        ("TenantManagement", "business", "BusinessCapability", "租户管理业务能力", {}),
        ("Tenant", "business", "BusinessObject", "租户业务对象", {"core": True}),
        ("TenantIsolation", "business", "BusinessRule", "租户隔离业务不变量", {}),
        ("BaseIAM", "software", "Module", "身份与租户基础模块", {"environment": "production"}),
        ("TenantController", "software", "Class", "租户接口控制器", {}),
        ("CreateTenantAPI", "software", "ApiOperation", "创建租户 API", {}),
        ("DcpTenantTable", "software", "Table", "租户主表", {"environment": "production"}),
        ("TenantIdColumn", "software", "Column", "租户标识字段", {"environment": "production", "sensitive": True, "classification": "internal"}),
        ("TenantCreatedTopic", "software", "EventContract", "租户创建事件契约", {}),
        ("IAMDataOwner", "governance", "OwnerRole", "IAM 数据责任人", {}),
        ("TenantIsolationPolicy", "governance", "AccessPolicy", "租户隔离访问策略", {}),
        ("TenantIdCompleteness", "governance", "QualityRule", "租户标识完整性规则", {"threshold": 1.0}),
        ("ReleaseSecurityGate", "governance", "QualityGate", "发布安全门禁", {}),
    )
    for name, profile, kind, description, attrs in m1:
        node(name, "M1", profile, kind, name, description, attrs, f"https://onto.qm.cn/ontorefactor/model/dcp/1.0#{name}")

    # M0 concrete facts and observations.
    node("TenantControllerCommit", "M0", "software", "SourceArtifact", "TenantController@abc123", "Git 提交中的实际源文件", {"commit": "abc123"}, "git://dcp-base-tenant/commit/abc123/src/TenantController.java")
    node("TenantIdPhysical", "M0", "software", "PhysicalColumn", "prod.dcp_tenant.tenant_id", "生产库中的实际字段", {"environment": "production", "sensitive": True}, "db://prod/dcp_base_tenant/dcp_tenant/tenant_id")
    node("CompletenessRun", "M0", "governance", "Measurement", "tenant-id-completeness@20260817", "一次质量测量", {"value": 0.9987, "threshold": 1.0, "result": "failed"}, "measurement:tenant-id-completeness:20260817")

    evidence = create_evidence(db, tenant_id, pid, {
        "kind": "SourceArtifact", "uri": "evidence://dcp/tenant-isolation-design",
        "title": "租户隔离设计与 DDL", "content": "tenant_id NOT NULL; all tenant queries must filter tenant_id",
        "source": "builtin:tenant-isolation", "metadata": {"deterministic": True},
    })
    runtime_evidence = create_evidence(db, tenant_id, pid, {
        "kind": "RuntimeObservation", "uri": "evidence://trace/runtime-98271",
        "title": "生产调用 Trace 98271", "content": "observed tenant-scoped request and database access",
        "source": "apm://trace/98271", "metadata": {"environment": "production"},
    })

    type_map = {
        "TenantManagement": "BusinessCapabilityType", "Tenant": "BusinessObjectType",
        "TenantIsolation": "BusinessRuleType", "BaseIAM": "ModuleType",
        "TenantController": "ClassType", "CreateTenantAPI": "ApiOperationType",
        "DcpTenantTable": "TableType", "TenantIdColumn": "ColumnType",
        "TenantCreatedTopic": "EventContractType", "IAMDataOwner": "OwnerType",
        "TenantIsolationPolicy": "AccessPolicyType", "TenantIdCompleteness": "QualityRuleType",
        "ReleaseSecurityGate": "QualityGateType",
    }
    for child, parent in type_map.items():
        create_assertion(db, tenant_id, pid, {
            "subject_id": nodes[child]["id"], "predicate": "INSTANCE_OF", "object_id": nodes[parent]["id"],
            "assertion_type": "declared", "confidence": 1.0, "review_status": "accepted", "generated_by": "builtin:ontology-bootstrap",
        })
    for profile_names in profile_types.values():
        for child in profile_names:
            create_assertion(db, tenant_id, pid, {
                "subject_id": nodes[child]["id"], "predicate": "INSTANCE_OF", "object_id": nodes["MetaEntityType"]["id"],
                "assertion_type": "declared", "confidence": 1.0, "review_status": "accepted", "generated_by": "builtin:ontology-bootstrap",
            })
    for child in ("Object", "Subject", "Activity", "Rule", "Result", "Evidence", "Assertion", "Term", "State", "TimeExtent"):
        create_assertion(db, tenant_id, pid, {
            "subject_id": nodes[child]["id"], "predicate": "INSTANCE_OF", "object_id": nodes["MetaEntityType"]["id"],
            "assertion_type": "declared", "confidence": 1.0, "review_status": "accepted", "generated_by": "builtin:ontology-bootstrap",
        })

    relations = (
        ("TenantManagement", "REALIZED_BY", "BaseIAM", "declared"),
        ("BaseIAM", "OWNED_BY", "IAMDataOwner", "declared"),
        ("TenantController", "BELONGS_TO", "BaseIAM", "extracted"),
        ("BaseIAM", "EXPOSES", "CreateTenantAPI", "extracted"),
        ("DcpTenantTable", "REPRESENTS", "Tenant", "declared"),
        ("CreateTenantAPI", "OPERATES_ON", "Tenant", "declared"),
        ("TenantCreatedTopic", "REPRESENTS_EVENT", "Tenant", "declared"),
        ("Tenant", "GOVERNED_BY", "TenantIsolation", "declared"),
        ("DcpTenantTable", "OWNED_BY", "IAMDataOwner", "declared"),
        ("TenantIdColumn", "GOVERNED_BY", "TenantIsolationPolicy", "declared"),
        ("TenantIdColumn", "OWNED_BY", "IAMDataOwner", "declared"),
        ("TenantIdCompleteness", "VALIDATES", "TenantIdColumn", "declared"),
        ("ReleaseSecurityGate", "EVALUATES", "BaseIAM", "declared"),
        ("TenantControllerCommit", "INSTANCE_OF", "TenantController", "extracted"),
        ("TenantIdPhysical", "INSTANCE_OF", "TenantIdColumn", "observed"),
        ("TenantIdPhysical", "OWNED_BY", "IAMDataOwner", "observed"),
        ("TenantIdPhysical", "GOVERNED_BY", "TenantIsolationPolicy", "observed"),
        ("CompletenessRun", "INSTANCE_OF", "TenantIdCompleteness", "observed"),
        ("CompletenessRun", "VALIDATES", "TenantIdPhysical", "observed"),
    )
    for subject, predicate, obj, assertion_type in relations:
        create_assertion(db, tenant_id, pid, {
            "subject_id": nodes[subject]["id"], "predicate": predicate, "object_id": nodes[obj]["id"],
            "assertion_type": assertion_type, "confidence": 1.0 if assertion_type == "declared" else 0.98,
            "evidence_ids": [runtime_evidence["id"] if assertion_type == "observed" else evidence["id"]],
            "generated_by": "builtin:tenant-isolation", "review_status": "accepted",
        })

    builtin_rules = (
        ("相邻层实例化", "metamodel", "critical", {"predicate": "INSTANCE_OF", "layerDelta": 1}),
        ("关键业务能力必须有软件实现", "integrity", "high", {"kind": "BusinessCapability", "requires": "REALIZED_BY"}),
        ("生产数据资产必须有 Owner", "ownership", "critical", {"environment": "production", "requires": "OWNED_BY"}),
        ("敏感字段必须受访问策略约束", "security", "critical", {"sensitive": True, "requires": "GOVERNED_BY"}),
        ("正式非声明断言必须可追溯", "provenance", "high", {"reviewStatus": "accepted", "requires": ["evidence", "generatedBy"]}),
    )
    for name, rule_type, severity, expression in builtin_rules:
        upsert_rule(db, tenant_id, pid, {
            "name": name, "rule_type": rule_type, "severity": severity,
            "description": "OntoRefactor 首期确定性治理门禁", "expression": expression,
        })

    validation = validate_project(db, tenant_id, pid, persist=True)
    return {"project": project, "overview": overview(db, tenant_id, pid), "validation": validation}
