"""Machine-readable security and governance review for EwoooC."""

from __future__ import annotations

import ast
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from services.http_access_policy_service import (
    INTERNAL_ONLY_ENDPOINTS,
    INTENTIONALLY_PUBLIC_ENDPOINTS,
    SESSION_REQUIRED_BLUEPRINTS,
    SESSION_REQUIRED_ENDPOINTS,
)


POLICY = "global_product_governance_v2_security_review_v1"
ROOT = Path(__file__).resolve().parents[1]
ASSET_INVENTORY_PATH = Path("governance/ewoooc_asset_inventory.json")
QUALITY_BASELINE_PATH = Path("governance/security_governance_review_baseline.json")
SOURCE_RECEIPT_PATH = Path("governance/security_governance_source_receipt.json")
RUNTIME_RECEIPT_PATH = Path("governance/security_governance_runtime_receipt.json")
AUTH_IDENTITY_RUNTIME_RECEIPT_PATH = Path("governance/auth_identity_runtime_receipt.json")

MUTATING_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
AUTH_DECORATOR_MARKERS = (
    "login_required",
    "admin_required",
    "role_required",
    "permission_required",
    "check_alert_auth",
    "internal_key_required",
    "require_api_token",
    "verify_webhook",
)
INLINE_AUTH_CALLS = {
    "_require_internal_key",
    "verify_internal_token",
    "verify_webhook_signature",
    "compare_digest",
}
FORBIDDEN_SOURCE_MARKERS = (
    "github.com",
    "api.github.com",
    "raw.githubusercontent.com",
    "codeload.github.com",
    "ghcr.io",
)
ASSET_REQUIRED_FIELDS = {
    "canonical_id",
    "asset_type",
    "owner_lane",
    "source_truth",
    "runtime_identity",
    "signal_freshness",
    "policy",
    "executor",
    "verifier",
    "backup_restore",
    "learning_targets",
}
DOCKERIGNORE_REQUIRED_PATTERNS = {
    ".git",
    ".gitea",
    ".env",
    ".env.*",
    "data",
    "logs",
    "backups",
    "config/google_credentials.json",
    "config/google_token.json",
    "config/google_token.pickle",
    "config/*.json",
}


def _read(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="ignore")
    except OSError:
        return ""


def _blueprint_names(tree: ast.Module) -> dict[str, str]:
    names: dict[str, str] = {}
    for node in tree.body:
        if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call):
            continue
        func = node.value.func
        if not isinstance(func, ast.Name) or func.id != "Blueprint" or not node.value.args:
            continue
        first = node.value.args[0]
        if not isinstance(first, ast.Constant) or not isinstance(first.value, str):
            continue
        for target in node.targets:
            if isinstance(target, ast.Name):
                names[target.id] = first.value
    return names


def _route_metadata(decorator: ast.expr) -> tuple[str, set[str]] | None:
    if not isinstance(decorator, ast.Call) or not isinstance(decorator.func, ast.Attribute):
        return None
    if decorator.func.attr != "route" or not decorator.args:
        return None
    first = decorator.args[0]
    route = first.value if isinstance(first, ast.Constant) and isinstance(first.value, str) else ""
    methods = {"GET"}
    for keyword in decorator.keywords:
        if keyword.arg != "methods" or not isinstance(keyword.value, (ast.List, ast.Tuple)):
            continue
        parsed = {
            item.value.upper()
            for item in keyword.value.elts
            if isinstance(item, ast.Constant) and isinstance(item.value, str)
        }
        if parsed:
            methods = parsed
    return route, methods


def _called_names(node: ast.AST) -> set[str]:
    names: set[str] = set()
    for child in ast.walk(node):
        if not isinstance(child, ast.Call):
            continue
        if isinstance(child.func, ast.Name):
            names.add(child.func.id)
        elif isinstance(child.func, ast.Attribute):
            names.add(child.func.attr)
    return names


def _scan_routes(root: Path) -> dict[str, Any]:
    rows: list[dict[str, Any]] = []
    parse_errors: list[str] = []
    for path in sorted((root / "routes").glob("*.py")):
        try:
            tree = ast.parse(_read(path), filename=str(path))
        except SyntaxError:
            parse_errors.append(path.relative_to(root).as_posix())
            continue
        blueprints = _blueprint_names(tree)
        for node in tree.body:
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            decorator_text = [ast.unparse(item) for item in node.decorator_list]
            explicit_guard = any(
                marker in text
                for marker in AUTH_DECORATOR_MARKERS
                for text in decorator_text
            )
            inline_guard = bool(_called_names(node) & INLINE_AUTH_CALLS)
            for decorator in node.decorator_list:
                metadata = _route_metadata(decorator)
                if metadata is None:
                    continue
                route, methods = metadata
                blueprint_var = ""
                if isinstance(decorator, ast.Call) and isinstance(decorator.func, ast.Attribute):
                    if isinstance(decorator.func.value, ast.Name):
                        blueprint_var = decorator.func.value.id
                blueprint = blueprints.get(blueprint_var, blueprint_var.removesuffix("_bp"))
                endpoint = f"{blueprint}.{node.name}" if blueprint else node.name
                central_guard = (
                    blueprint in SESSION_REQUIRED_BLUEPRINTS
                    or endpoint in SESSION_REQUIRED_ENDPOINTS
                    or endpoint in INTERNAL_ONLY_ENDPOINTS
                )
                rows.append({
                    "file": path.relative_to(root).as_posix(),
                    "line": node.lineno,
                    "endpoint": endpoint,
                    "route": route,
                    "methods": sorted(methods),
                    "mutating": bool(methods & MUTATING_METHODS),
                    "guarded": explicit_guard or inline_guard or central_guard,
                    "guard_source": (
                        "decorator" if explicit_guard else
                        "inline" if inline_guard else
                        "central_policy" if central_guard else
                        "none"
                    ),
                })

    unguarded = [item for item in rows if not item["guarded"]]
    mutating_unguarded = [item for item in unguarded if item["mutating"]]
    unclassified_unguarded = [
        item for item in unguarded
        if item["endpoint"] not in INTENTIONALLY_PUBLIC_ENDPOINTS
    ]
    return {
        "route_count": len(rows),
        "guarded_count": sum(1 for item in rows if item["guarded"]),
        "unguarded_count": len(unguarded),
        "unguarded": unguarded,
        "unclassified_unguarded_count": len(unclassified_unguarded),
        "unclassified_unguarded": unclassified_unguarded,
        "mutating_unguarded_count": len(mutating_unguarded),
        "mutating_unguarded": mutating_unguarded[:30],
        "parse_errors": parse_errors,
    }


def _asset_inventory(root: Path) -> dict[str, Any]:
    path = root / ASSET_INVENTORY_PATH
    try:
        payload = json.loads(_read(path))
    except (TypeError, json.JSONDecodeError):
        payload = {}
    assets = payload.get("assets") if isinstance(payload, dict) else []
    if not isinstance(assets, list):
        assets = []
    missing: list[dict[str, Any]] = []
    populated_fields = 0
    expected_fields = len(assets) * len(ASSET_REQUIRED_FIELDS)
    for index, asset in enumerate(assets):
        if not isinstance(asset, dict):
            missing.append({"index": index, "missing_fields": sorted(ASSET_REQUIRED_FIELDS)})
            continue
        missing_fields = sorted(field for field in ASSET_REQUIRED_FIELDS if not asset.get(field))
        populated_fields += len(ASSET_REQUIRED_FIELDS) - len(missing_fields)
        if missing_fields:
            missing.append({
                "index": index,
                "canonical_id": asset.get("canonical_id"),
                "missing_fields": missing_fields,
            })
    field_coverage = round((populated_fields / expected_fields * 100), 1) if expected_fields else 0.0
    reconciliation = payload.get("reconciliation") if isinstance(payload, dict) else {}
    return {
        "path": ASSET_INVENTORY_PATH.as_posix(),
        "asset_count": len(assets),
        "field_coverage_percent": field_coverage,
        "missing_field_assets": missing,
        "schema_valid": bool(assets) and not missing,
        "reconciliation_status": (reconciliation or {}).get("status", "missing"),
        "runtime_receipts_complete": bool((reconciliation or {}).get("runtime_receipts_complete")),
    }


def _quality_baseline(root: Path) -> dict[str, Any]:
    path = root / QUALITY_BASELINE_PATH
    try:
        payload = json.loads(_read(path))
    except (TypeError, json.JSONDecodeError):
        payload = {}
    collection = payload.get("collection") if isinstance(payload, dict) else {}
    full_suite = payload.get("full_suite_without_missing_contract") if isinstance(payload, dict) else {}
    focused = payload.get("focused_suites") if isinstance(payload, dict) else {}
    collection = collection if isinstance(collection, dict) else {}
    full_suite = full_suite if isinstance(full_suite, dict) else {}
    focused = focused if isinstance(focused, dict) else {}
    return {
        "path": QUALITY_BASELINE_PATH.as_posix(),
        "captured_at": payload.get("captured_at") if isinstance(payload, dict) else None,
        "collection_status": collection.get("status", "missing"),
        "collection_error_count": int(collection.get("error_count") or 0),
        "missing_contract": collection.get("missing_contract"),
        "full_suite_passed": int(full_suite.get("passed") or 0),
        "full_suite_failed": int(full_suite.get("failed") or 0),
        "full_suite_skipped": int(full_suite.get("skipped") or 0),
        "focused_suites": focused,
        "next_machine_action": payload.get("next_machine_action") if isinstance(payload, dict) else None,
    }


def _source_receipt(root: Path) -> dict[str, Any]:
    path = root / SOURCE_RECEIPT_PATH
    try:
        payload = json.loads(_read(path))
    except (TypeError, json.JSONDecodeError):
        payload = {}
    checks = payload.get("checks") if isinstance(payload, dict) else []
    checks_by_id = {
        item.get("id"): item
        for item in checks
        if isinstance(item, dict) and item.get("id")
    } if isinstance(checks, list) else {}
    source_commit = str(payload.get("source_commit") or "") if isinstance(payload, dict) else ""
    commit_shape_valid = len(source_commit) == 40 and all(
        char in "0123456789abcdefABCDEF" for char in source_commit
    )
    return {
        "path": SOURCE_RECEIPT_PATH.as_posix(),
        "available": bool(payload) and payload.get("receipt_kind") == "security_governance_source_receipt",
        "source_commit": source_commit,
        "source_commit_shape_valid": commit_shape_valid,
        "generated_at": payload.get("generated_at") if isinstance(payload, dict) else None,
        "release_gate_status": (
            (payload.get("release_gate") or {}).get("status")
            if isinstance(payload, dict) else None
        ),
        "_checks": checks_by_id,
    }


def _runtime_receipt(root: Path) -> dict[str, Any]:
    path = root / RUNTIME_RECEIPT_PATH
    try:
        payload = json.loads(_read(path))
    except (TypeError, json.JSONDecodeError):
        payload = {}
    access = payload.get("access_control") if isinstance(payload, dict) else {}
    database = payload.get("database") if isinstance(payload, dict) else {}
    post_verifier = payload.get("post_verifier") if isinstance(payload, dict) else {}
    cd = payload.get("cd") if isinstance(payload, dict) else {}
    access = access if isinstance(access, dict) else {}
    database = database if isinstance(database, dict) else {}
    post_verifier = post_verifier if isinstance(post_verifier, dict) else {}
    cd = cd if isinstance(cd, dict) else {}
    source_commit = str(payload.get("source_commit") or "") if isinstance(payload, dict) else ""
    commit_shape_valid = len(source_commit) == 40 and all(
        char in "0123456789abcdefABCDEF" for char in source_commit
    )
    security_readback_passed = bool(
        payload.get("receipt_kind") == "security_governance_runtime_receipt"
        and commit_shape_valid
        and access.get("status") == "pass"
        and access.get("sensitive_routes_denied") is True
        and access.get("security_headers_passed") is True
        and access.get("secure_cookie") is True
        and access.get("metrics_public") is False
        and access.get("metrics_internal_canary") is True
        and database.get("unchanged") is True
        and post_verifier.get("health_status") == "healthy"
        and post_verifier.get("governance_gate") == "pass"
    ) if isinstance(payload, dict) else False
    return {
        "path": RUNTIME_RECEIPT_PATH.as_posix(),
        "available": bool(payload),
        "trace_id": payload.get("trace_id") if isinstance(payload, dict) else None,
        "run_id": payload.get("run_id") if isinstance(payload, dict) else None,
        "work_item_id": payload.get("work_item_id") if isinstance(payload, dict) else None,
        "source_commit": source_commit,
        "source_commit_shape_valid": commit_shape_valid,
        "deployed_version": payload.get("deployed_version") if isinstance(payload, dict) else None,
        "deployment_mode": payload.get("deployment_mode") if isinstance(payload, dict) else None,
        "security_readback_passed": security_readback_passed,
        "sensitive_route_count": int(access.get("sensitive_route_count") or 0),
        "denied_route_count": int(access.get("denied_route_count") or 0),
        "metrics_public": bool(access.get("metrics_public")),
        "metrics_internal_canary": bool(access.get("metrics_internal_canary")),
        "metrics_target_health": access.get("metrics_target_health"),
        "database_unchanged": bool(database.get("unchanged")),
        "health_status": post_verifier.get("health_status"),
        "governance_gate": post_verifier.get("governance_gate"),
        "cd_run_id": cd.get("run_id"),
        "cd_status": cd.get("status"),
        "cd_blocker": cd.get("blocker"),
        "learning_recorded": bool(payload.get("learning_recorded")) if isinstance(payload, dict) else False,
        "next_machine_action": payload.get("next_machine_action") if isinstance(payload, dict) else None,
    }


def _auth_identity_source_posture(root: Path) -> dict[str, Any]:
    files = {
        "auth": _read(root / "auth.py"),
        "policy": _read(root / "services" / "auth_identity_service.py"),
        "user_service": _read(root / "services" / "user_service.py"),
        "permission_service": _read(root / "services" / "permission_service.py"),
        "user_models": _read(root / "database" / "user_models.py"),
        "http_policy": _read(root / "services" / "http_access_policy_service.py"),
        "compose": _read(root / "docker-compose.yml"),
    }
    markers = {
        "database_authority": "database_auth_can_authorize" in files["auth"],
        "durable_lockout": "get_lockout_status" in files["user_service"]
        and "login_history" in files["policy"],
        "session_revocation": "identity_version" in files["auth"]
        and "session_revoked" in files["auth"],
        "role_matrix": "required_roles_for_request" in files["http_policy"]
        and "ROLE_MANAGER" in files["policy"],
        "trusted_proxy_policy": "MOMO_TRUSTED_PROXY_CIDRS" in files["compose"]
        and "resolve_client_ip" in files["auth"],
        "identity_mutation_audit": "_add_identity_audit" in files["user_service"]
        and "STATUS_IDENTITY_CHANGED" in files["user_models"],
        "permission_mutation_audit": "_add_permission_audit" in files["permission_service"]
        and "STATUS_PERMISSION_CHANGED" in files["user_models"],
        "runtime_readback": (
            root / "scripts" / "ops" / "report_auth_identity_governance.py"
        ).is_file(),
    }
    return {
        "markers": markers,
        "source_ready": all(markers.values()),
    }


def _auth_identity_runtime_receipt(root: Path) -> dict[str, Any]:
    path = root / AUTH_IDENTITY_RUNTIME_RECEIPT_PATH
    try:
        payload = json.loads(_read(path))
    except (TypeError, json.JSONDecodeError):
        payload = {}
    valid = bool(
        isinstance(payload, dict)
        and payload.get("receipt_kind") == "auth_identity_runtime_readback"
        and payload.get("work_item_id") == "SEC-P0-002"
    )
    checks = payload.get("checks") if valid else {}
    inventory = payload.get("inventory") if valid else {}
    checks = checks if isinstance(checks, dict) else {}
    inventory = inventory if isinstance(inventory, dict) else {}
    runtime_closed = bool(
        valid
        and payload.get("runtime_closed") is True
        and payload.get("auth_identity_mode") == "database"
        and checks.get("legacy_shared_password_retired") is True
    )
    return {
        "path": AUTH_IDENTITY_RUNTIME_RECEIPT_PATH.as_posix(),
        "available": valid,
        "generated_at": payload.get("generated_at") if valid else None,
        "status": payload.get("status") if valid else None,
        "configured_auth_identity_mode": (
            payload.get("configured_auth_identity_mode") if valid else None
        ),
        "auth_identity_mode": payload.get("auth_identity_mode") if valid else None,
        "canary_ready": bool(payload.get("canary_ready")) if valid else False,
        "runtime_closed": runtime_closed,
        "active_admin_count": int(inventory.get("active_admin_count") or 0),
        "users_total": int(inventory.get("users_total") or 0),
        "login_history_events": int(inventory.get("login_history_events") or 0),
        "identity_mutation_events": int(
            inventory.get("identity_mutation_events") or 0
        ),
        "permission_assignments": int(inventory.get("permission_assignments") or 0),
        "legacy_shared_password_retired": bool(
            checks.get("legacy_shared_password_retired")
        ),
        "remaining_blocker": payload.get("remaining_blocker") if valid else None,
        "next_machine_action": payload.get("next_machine_action") if valid else None,
        "secret_values_read": False,
    }


def _docker_build_context_posture(root: Path) -> dict[str, Any]:
    path = root / ".dockerignore"
    patterns = {
        line.strip().rstrip("/")
        for line in _read(path).splitlines()
        if line.strip() and not line.lstrip().startswith("#")
    }
    missing = sorted(DOCKERIGNORE_REQUIRED_PATTERNS - patterns)
    return {
        "path": ".dockerignore",
        "exists": path.is_file(),
        "required_pattern_count": len(DOCKERIGNORE_REQUIRED_PATTERNS),
        "missing_patterns": missing,
        "safe": path.is_file() and not missing,
    }


def _workflow_supply_chain(root: Path) -> dict[str, Any]:
    files = sorted((root / ".gitea" / "workflows").glob("*.y*ml"))
    forbidden: list[dict[str, Any]] = []
    external_actions: list[dict[str, Any]] = []
    for path in files:
        for line_number, line in enumerate(_read(path).splitlines(), start=1):
            lower = line.lower()
            for marker in FORBIDDEN_SOURCE_MARKERS:
                if marker in lower:
                    forbidden.append({
                        "file": path.relative_to(root).as_posix(),
                        "line": line_number,
                        "marker": marker,
                    })
            stripped = line.strip()
            if stripped.startswith("uses:") or " uses:" in stripped:
                external_actions.append({
                    "file": path.relative_to(root).as_posix(),
                    "line": line_number,
                    "reference": stripped[:180],
                })
    return {
        "workflow_count": len(files),
        "forbidden_source_references": forbidden,
        "external_action_references": external_actions,
        "gitea_only": bool(files) and not forbidden and not external_actions,
    }


def _dependency_posture(root: Path) -> dict[str, Any]:
    requirements = []
    for raw in _read(root / "requirements.txt").splitlines():
        line = raw.split("#", 1)[0].strip()
        if line and not line.startswith(("-", "--")):
            requirements.append(line)
    exact = [line for line in requirements if "==" in line or " @ " in line]
    sbom_names = {"sbom.json", "bom.json", "cyclonedx.json", "spdx.json"}
    sbom_candidates: list[str] = []
    for name in sorted(sbom_names):
        candidate = root / name
        if candidate.is_file():
            sbom_candidates.append(candidate.relative_to(root).as_posix())
    for folder in ("artifacts", "build", "dist", "governance"):
        base = root / folder
        if not base.is_dir():
            continue
        sbom_candidates.extend(
            path.relative_to(root).as_posix()
            for path in base.rglob("*")
            if path.is_file() and path.name.lower() in sbom_names
        )
    return {
        "direct_requirement_count": len(requirements),
        "exact_requirement_count": len(exact),
        "all_direct_requirements_exact": bool(requirements) and len(exact) == len(requirements),
        "lock_exists": (root / "requirements.lock").exists(),
        "sbom_files": sbom_candidates,
        "automated_sca_configured": any(
            marker in _read(path).lower()
            for path in (root / ".gitea" / "workflows").glob("*.y*ml")
            for marker in ("pip-audit", "trivy", "grype", "osv-scanner")
        ),
    }


def _large_python_files(root: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for folder in ("services", "routes"):
        for path in (root / folder).rglob("*.py"):
            try:
                line_count = sum(1 for _ in path.open(encoding="utf-8", errors="ignore"))
            except OSError:
                continue
            if line_count > 800:
                rows.append({
                    "file": path.relative_to(root).as_posix(),
                    "lines": line_count,
                })
    return sorted(rows, key=lambda item: item["lines"], reverse=True)


def _check(
    check_id: str,
    function: str,
    status: str,
    severity: str,
    summary: str,
    evidence: dict[str, Any],
    next_action: str,
    *,
    release_blocking: bool = False,
) -> dict[str, Any]:
    return {
        "id": check_id,
        "function": function,
        "status": status,
        "severity": severity,
        "summary": summary,
        "evidence": evidence,
        "next_machine_action": next_action,
        "release_blocking": release_blocking,
    }


def _work_items(
    checks: dict[str, dict[str, Any]],
    runtime_receipt: dict[str, Any],
    auth_identity_source: dict[str, Any],
    auth_identity_runtime: dict[str, Any],
) -> list[dict[str, Any]]:
    access_closed = runtime_receipt["security_readback_passed"]
    access_next_action = (
        "inventory_active_users_then_shadow_database_auth_before_cutover"
        if access_closed
        else "deny_public_metrics_then_verify_internal_prometheus_canary"
    )
    release_next_action = (
        "restore_dedicated_ewoooc_runner_then_replay_waiting_cd"
        if runtime_receipt["security_readback_passed"]
        else "deploy_current_source_then_read_back_security_matrix_and_container_identity"
    )
    identity_status = (
        "completed"
        if auth_identity_runtime["runtime_closed"]
        else "in_progress"
        if auth_identity_source["source_ready"] or auth_identity_runtime["canary_ready"]
        else "not_started"
    )
    if identity_status == "completed":
        identity_next_action = "advance_to_SEC-P0-003"
    elif auth_identity_runtime["canary_ready"]:
        identity_next_action = (
            auth_identity_runtime.get("next_machine_action")
            or "capture_database_admin_login_receipts_for_automatic_cutover"
        )
    elif auth_identity_source["source_ready"]:
        identity_next_action = "deploy_hybrid_database_identity_then_write_runtime_receipt"
    else:
        identity_next_action = (
            "implement_database_identity_durable_lockout_session_revocation_and_role_matrix"
        )
    return [
        {"order": 1, "id": "SEC-P0-001", "priority": "P0", "status": "completed" if access_closed else "in_progress", "lane": "deny_by_default_access", "objective": "Close all sensitive and mutating routes behind central auth or signed internal transport.", "next_machine_action": access_next_action},
        {"order": 2, "id": "SEC-P0-002", "priority": "P0", "status": identity_status, "lane": "identity_rbac", "objective": "Replace the shared admin password with database-backed identities, least-privilege RBAC, durable lockout and auditable sessions.", "next_machine_action": identity_next_action},
        {"order": 3, "id": "SEC-P0-003", "priority": "P0", "status": "in_progress", "lane": "webhook_trust", "objective": "Require Telegram and internal webhook transport secrets with replay protection.", "next_machine_action": "provision_telegram_webhook_secret_then_enable_required_mode_and_canary"},
        {"order": 4, "id": "SUPPLY-P0-001", "priority": "P0", "status": "in_progress", "lane": "software_supply_chain", "objective": "Use Gitea-only checkout, exact locks, SAST/SCA/secret scan, SBOM, image digest and provenance gates.", "next_machine_action": "add_internal_scanner_artifacts_and_exact_dependency_lock"},
        {"order": 5, "id": "GOV-P0-001", "priority": "P0", "status": "in_progress", "lane": "asset_graph", "objective": "Reconcile every host, service, data, AI, route, package and evidence asset against runtime truth.", "next_machine_action": "probe_all_assets_and_write_same_run_reconciliation_receipt"},
        {"order": 6, "id": "GOV-P0-002", "priority": "P0", "status": "not_started", "lane": "controlled_apply_contract", "objective": "Enforce one trace_id/run_id/work_item_id from sensor through verifier, rollback and learning acknowledgement.", "next_machine_action": "introduce_shared_governance_envelope_and_migrate_eventrouter_autoheal_first"},
        {"order": 7, "id": "RAG-P0-001", "priority": "P0", "status": "not_started", "lane": "internal_rag_canary", "objective": "Canary PixelRAG candidate knowledge into internal pgvector RAG without price or ai_insights promotion.", "next_machine_action": "run_internal_rag_candidate_canary"},
        {"order": 8, "id": "MCP-P0-001", "priority": "P0", "status": "in_progress", "lane": "mcp_rag_runtime", "objective": "Enable and verify MCP/RAG runtime health rather than registry-only readiness.", "next_machine_action": "close_mcp_ports_router_and_rag_runtime_readback"},
        {"order": 9, "id": "SEC-P0-004", "priority": "P0", "status": "not_started", "lane": "security_operations", "objective": "Publish security MTTA/MTTR, recurrence, false-positive, verifier and rollback metrics with incident closure receipts.", "next_machine_action": "add_security_event_metrics_and_incident_lifecycle_store"},
        {"order": 10, "id": "REL-P0-001", "priority": "P0", "status": "in_progress", "lane": "release_readback", "objective": "Separate source, test, CD, production runtime and visible proof for every release without touching momo-db.", "next_machine_action": release_next_action},
        {"order": 11, "id": "RES-P1-001", "priority": "P1", "status": "not_started", "lane": "resilience", "objective": "Automate checksum, offsite backup and isolated restore drills with measured RPO/RTO.", "next_machine_action": "build_non_destructive_restore_drill_receipt"},
        {"order": 12, "id": "PLAT-P1-001", "priority": "P1", "status": "not_started", "lane": "container_hardening", "objective": "Run application containers as non-root with minimal capabilities and read-only filesystems where compatible.", "next_machine_action": "shadow_nonroot_image_then_canary_three_app_containers"},
        {"order": 13, "id": "APPSEC-P1-001", "priority": "P1", "status": "in_progress", "lane": "browser_security", "objective": "Move CSP from report-only to enforced and remove unsafe DOM/inline sinks by measured rollout.", "next_machine_action": "collect_csp_violations_then_replace_high_risk_innerhtml_sinks"},
        {"order": 14, "id": "APPSEC-P1-002", "priority": "P1", "status": "not_started", "lane": "unsafe_cache_serialization", "objective": "Replace writable pickle caches with signed JSON or constrained serialization.", "next_machine_action": "migrate_dashboard_daily_sales_and_edm_shared_caches"},
        {"order": 15, "id": "ARCH-P1-001", "priority": "P1", "status": "in_progress", "lane": "modularization", "objective": "Split oversized services and routes so policy, executor and verifier can be independently tested.", "next_machine_action": "split_ai_automation_smoke_and_pchome_mapping_backlog_by_family"},
        {"order": 16, "id": "UX-P1-001", "priority": "P1", "status": "in_progress", "lane": "professional_product_ui", "objective": "Complete first-viewport cockpit, progressive disclosure, text-density reduction, accessibility and mobile visual smoke across primary pages.", "next_machine_action": "run_sitewide_visual_qa_then_fix_highest_traffic_text_walls"},
        {"order": 17, "id": "PIXELRAG-P1-001", "priority": "P1", "status": "not_started", "lane": "visual_embedding", "objective": "Benchmark Ollama-first multimodal embedding and pgvector-compatible visual metadata.", "next_machine_action": "benchmark_qwen3_vl_equivalent_on_gcp_a_gcp_b_111"},
        {"order": 18, "id": "MARKET-P1-001", "priority": "P1", "status": "in_progress", "lane": "marketplace_source_contracts", "objective": "Finish stable source contracts for Shopee, Coupang, Yahoo, ETMall, Friday and Rakuten with barrier classification.", "next_machine_action": "execute_provenance_rate_limit_and_public_boundary_replays"},
        {"order": 19, "id": "QA-P1-001", "priority": "P1", "status": "in_progress", "lane": "test_governance", "objective": "Restore clean collection, eliminate time/copy drift and split deterministic release tests from nightly integration and production canaries.", "next_machine_action": "restore_or_retire_missing_pchome_report_contract_then_baseline_legacy_failures"},
        {"order": 20, "id": "GOV-P2-001", "priority": "P2", "status": "not_started", "lane": "continuous_control_improvement", "objective": "Trend NIST CSF and OWASP ASVS control coverage and automatically create drift work items.", "next_machine_action": "persist_review_snapshots_and_diff_control_regressions"},
        {"order": 21, "id": "DATA-P2-001", "priority": "P2", "status": "not_started", "lane": "data_governance", "objective": "Classify business, personal, operational and model data and enforce retention with auditable non-destructive defaults.", "next_machine_action": "build_data_classification_and_retention_inventory"},
        {"order": 22, "id": "CHAOS-P2-001", "priority": "P2", "status": "not_started", "lane": "controlled_resilience", "objective": "Exercise model-host, MCP, queue, Telegram and app-container failures with bounded rollback and learning receipts.", "next_machine_action": "design_non_destructive_failure_drill_matrix"},
    ]


def build_security_governance_review(
    *,
    root: Path | str | None = None,
    detail_limit: int = 30,
) -> dict[str, Any]:
    """Review source and runtime flags without network, DB or secret access."""
    scan_root = Path(root).resolve() if root is not None else ROOT
    detail_limit = max(5, min(int(detail_limit or 30), 100))

    route_posture = _scan_routes(scan_root)
    route_posture["unguarded"] = route_posture["unguarded"][:detail_limit]
    route_posture["unclassified_unguarded"] = route_posture["unclassified_unguarded"][:detail_limit]
    route_posture["mutating_unguarded"] = route_posture["mutating_unguarded"][:detail_limit]
    assets = _asset_inventory(scan_root)
    source_receipt = _source_receipt(scan_root)
    supply_chain = _workflow_supply_chain(scan_root)
    docker_context = _docker_build_context_posture(scan_root)
    supply_chain["docker_build_context"] = docker_context
    supply_chain["source_chain_ready"] = bool(
        supply_chain["gitea_only"] and docker_context["safe"]
    )
    receipt_supply = source_receipt["_checks"].get("GV.SC-gitea-only-supply-chain", {})
    if (
        supply_chain["workflow_count"] == 0
        and source_receipt["available"]
        and source_receipt["source_commit_shape_valid"]
        and receipt_supply.get("status") == "pass"
    ):
        receipt_evidence = receipt_supply.get("evidence") or {}
        supply_chain = dict(receipt_evidence) if isinstance(receipt_evidence, dict) else {}
        supply_chain["evidence_source"] = "cd_source_receipt"
        supply_chain["source_commit"] = source_receipt["source_commit"]
        supply_chain["source_chain_ready"] = True
    dependencies = _dependency_posture(scan_root)
    quality = _quality_baseline(scan_root)
    runtime_receipt = _runtime_receipt(scan_root)
    auth_identity_source = _auth_identity_source_posture(scan_root)
    auth_identity_runtime = _auth_identity_runtime_receipt(scan_root)
    identity_runtime_next_action = (
        auth_identity_runtime.get("next_machine_action")
        or "capture_database_admin_login_receipts_for_automatic_cutover"
    )
    large_files = _large_python_files(scan_root)
    app_source = _read(scan_root / "app.py")
    alert_source = _read(scan_root / "routes" / "alert_routes.py")
    constitution = _read(scan_root / "CONSTITUTION.md")
    dockerfile = _read(scan_root / "Dockerfile")
    telegram_secret_configured = bool(os.getenv("TELEGRAM_WEBHOOK_SECRET_TOKEN"))
    mcp_enabled = os.getenv("MCP_ROUTER_ENABLED", "false").lower() in {"1", "true", "yes", "on"}
    rag_enabled = os.getenv("RAG_ENABLED", "false").lower() in {"1", "true", "yes", "on"}

    required_headers = {
        "X-Content-Type-Options",
        "X-Frame-Options",
        "Referrer-Policy",
        "Permissions-Policy",
        "Strict-Transport-Security",
        "Content-Security-Policy-Report-Only",
    }
    missing_headers = sorted(header for header in required_headers if header not in app_source)
    session_secure_default = "PUBLIC_URL', 'https://mo.wooo.work'" in app_source
    protected_resource_ready = all(
        marker in alert_source
        for marker in ("'momo-db'", "ALLOWED_AUTO_FIX_CONTAINERS", "PROTECTED_CONTAINERS")
    )
    central_policy_registered = "enforce_http_access_policy" in app_source

    core_contract_source = "\n".join(
        _read(scan_root / path)
        for path in (
            "services/event_router.py",
            "services/auto_heal_service.py",
            "services/agent_actions.py",
            "services/code_review_pipeline_service.py",
            "services/elephant_alpha_autonomous_engine.py",
        )
    )
    contract_fields = {
        field: field in core_contract_source
        for field in ("trace_id", "run_id", "work_item_id", "post_verifier", "rollback", "learning_recorded")
    }
    contract_coverage = sum(contract_fields.values())
    policy_drift_markers = [
        marker
        for marker in (
            "所有 API Token、密碼存放於 `config.py`",
            "nohup python3 app.py",
            "pkill -9 -f \"python3 app.py\"",
        )
        if marker in constitution
    ]
    policy_alignment_ready = bool(constitution) and not policy_drift_markers
    policy_evidence = {
        "legacy_markers": policy_drift_markers,
        "evidence_source": "source_file" if constitution else "missing",
    }
    receipt_policy = source_receipt["_checks"].get("GV.PO-policy-alignment", {})
    if (
        not constitution
        and source_receipt["available"]
        and source_receipt["source_commit_shape_valid"]
        and receipt_policy.get("status") == "pass"
    ):
        policy_alignment_ready = True
        policy_evidence = {
            "legacy_markers": [],
            "evidence_source": "cd_source_receipt",
            "source_commit": source_receipt["source_commit"],
        }

    access_contract_ready = bool(
        central_policy_registered
        and route_posture["mutating_unguarded_count"] == 0
        and route_posture["unclassified_unguarded_count"] == 0
    )

    checks = [
        _check(
            "GV.AM-asset-inventory",
            "GOVERN/IDENTIFY",
            "partial" if assets["schema_valid"] and not assets["runtime_receipts_complete"] else "pass" if assets["schema_valid"] else "fail",
            "high",
            f"Asset inventory has {assets['asset_count']} assets and {assets['field_coverage_percent']}% required-field coverage; runtime reconciliation remains {assets['reconciliation_status']}.",
            assets,
            "probe_all_assets_and_write_same_run_reconciliation_receipt",
            release_blocking=not assets["schema_valid"],
        ),
        _check(
            "PR.AA-route-access-control",
            "PROTECT",
            "pass" if access_contract_ready else "fail",
            "critical",
            f"Route map: {route_posture['route_count']} total, {route_posture['unguarded_count']} intentionally public, {route_posture['unclassified_unguarded_count']} unclassified and {route_posture['mutating_unguarded_count']} mutating unguarded after central policy.",
            route_posture,
            "verify_production_anonymous_matrix_and_move_metrics_behind_edge_auth",
            release_blocking=not access_contract_ready,
        ),
        _check(
            "PR.DS-session-and-browser-baseline",
            "PROTECT",
            "pass" if not missing_headers and session_secure_default else "fail",
            "high",
            "Security headers and secure-cookie-by-public-URL baseline are present." if not missing_headers and session_secure_default else "Browser/session baseline is incomplete.",
            {"missing_headers": missing_headers, "secure_cookie_default": session_secure_default},
            "enforce_csp_after_report_only_violation_cleanup",
            release_blocking=bool(missing_headers) or not session_secure_default,
        ),
        _check(
            "PR.AC-database-identity-rbac",
            "PROTECT/DETECT",
            "pass" if auth_identity_runtime["runtime_closed"] else "partial" if auth_identity_source["source_ready"] or auth_identity_runtime["canary_ready"] else "fail",
            "critical",
            "Database-only identity authority, durable lockout, revocable sessions and role enforcement are closed in production." if auth_identity_runtime["runtime_closed"] else "Database identity source is implemented; production hybrid/database canary and shared-authority retirement remain distinct gates." if auth_identity_source["source_ready"] else "Database identity, durable lockout, revocable sessions or role enforcement source is incomplete.",
            {
                "source": auth_identity_source,
                "runtime": auth_identity_runtime,
            },
            "advance_to_SEC-P0-003" if auth_identity_runtime["runtime_closed"] else identity_runtime_next_action if auth_identity_runtime["canary_ready"] else "deploy_hybrid_database_identity_then_write_runtime_receipt",
        ),
        _check(
            "PR.AA-production-security-readback",
            "PROTECT/DETECT",
            "pass" if runtime_receipt["security_readback_passed"] else "partial",
            "critical",
            (
                f"Production security readback passed for {runtime_receipt['denied_route_count']}/{runtime_receipt['sensitive_route_count']} sensitive routes; database identity remained unchanged."
                if runtime_receipt["security_readback_passed"]
                else "Production access, header, cookie, container and database readback receipt is not complete."
            ),
            runtime_receipt,
            (
                "inventory_active_users_then_shadow_database_auth_before_cutover"
                if runtime_receipt["security_readback_passed"]
                else "write_commit_bound_production_security_runtime_receipt"
            ),
        ),
        _check(
            "PR.IR-autoheal-resource-boundary",
            "PROTECT/RESPOND",
            "pass" if protected_resource_ready else "fail",
            "critical",
            "Auto-fix is deny-by-default for containers and explicitly protects momo-db." if protected_resource_ready else "Auto-fix container boundary is incomplete.",
            {"protected_resource_ready": protected_resource_ready},
            "keep_container_allowlist_and_add_independent_post_verifier",
            release_blocking=not protected_resource_ready,
        ),
        _check(
            "GV.SC-gitea-only-supply-chain",
            "GOVERN/PROTECT",
            "pass" if supply_chain["source_chain_ready"] else "fail",
            "critical",
            "Gitea workflow and Docker build context exclude external actions, forbidden sources, secrets and runtime data." if supply_chain["source_chain_ready"] else "Source workflow or Docker build context boundary is incomplete.",
            supply_chain,
            "keep_gitea_checkout_and_generate_commit_bound_source_receipt",
            release_blocking=not supply_chain["source_chain_ready"],
        ),
        _check(
            "ID.RA-dependency-vulnerability-management",
            "IDENTIFY/PROTECT",
            "pass" if dependencies["lock_exists"] and dependencies["sbom_files"] and dependencies["automated_sca_configured"] else "fail",
            "high",
            "Dependency lock, SBOM and automated SCA are not yet a complete release gate.",
            dependencies,
            "add_exact_lock_internal_sca_secret_scan_sbom_and_image_digest_receipts",
        ),
        _check(
            "PR.PS-container-least-privilege",
            "PROTECT",
            "pass" if "\nUSER " in f"\n{dockerfile}" else "fail",
            "high",
            "Application image declares a non-root runtime user." if "\nUSER " in f"\n{dockerfile}" else "Application image currently defaults to root.",
            {"dockerfile_user_declared": "\nUSER " in f"\n{dockerfile}"},
            "shadow_nonroot_image_then_canary_three_app_containers",
        ),
        _check(
            "RS.MA-controlled-apply-envelope",
            "RESPOND/RECOVER",
            "pass" if contract_coverage == len(contract_fields) else "fail",
            "critical",
            f"Core automation contract contains {contract_coverage}/{len(contract_fields)} required closure fields.",
            {"fields": contract_fields, "coverage_count": contract_coverage},
            "introduce_shared_governance_envelope_and_migrate_eventrouter_autoheal_first",
        ),
        _check(
            "RC.RP-restore-drill",
            "RECOVER",
            "partial" if (scan_root / "backup_system.py").exists() else "fail",
            "high",
            "Backup code exists, but no durable isolated restore-drill receipt is committed or read back.",
            {"backup_code_exists": (scan_root / "backup_system.py").exists(), "restore_receipt_exists": False},
            "build_non_destructive_restore_drill_receipt",
        ),
        _check(
            "GV.PO-policy-alignment",
            "GOVERN",
            "pass" if policy_alignment_ready else "fail",
            "high",
            "Constitution is aligned to current production/governance doctrine." if policy_alignment_ready else "Policy source is missing or legacy doctrine still conflicts with production truth.",
            policy_evidence,
            "sunset_legacy_rules_with_owner_expiry_replacement_and_verifier",
            release_blocking=not policy_alignment_ready,
        ),
        _check(
            "GV.OV-modularization",
            "GOVERN",
            "pass" if not large_files else "partial",
            "medium",
            f"{len(large_files)} service/route files exceed 800 lines; top file has {large_files[0]['lines'] if large_files else 0} lines.",
            {"oversized_file_count": len(large_files), "top_files": large_files[:15]},
            "split_ai_automation_smoke_and_pchome_mapping_backlog_by_family",
        ),
        _check(
            "GV.QA-test-governance",
            "GOVERN/IDENTIFY",
            "pass" if quality["collection_status"] == "passed" and quality["full_suite_failed"] == 0 else "partial",
            "high",
            f"Focused security suites are green, but full test governance has {quality['collection_error_count']} collection errors and {quality['full_suite_failed']} baseline failures.",
            quality,
            "restore_or_retire_missing_pchome_report_contract_then_baseline_legacy_failures",
        ),
        _check(
            "PR.AA-telegram-webhook-transport",
            "PROTECT",
            "pass" if telegram_secret_configured else "partial",
            "critical",
            "Telegram webhook secret is configured." if telegram_secret_configured else "Code supports Telegram secret verification, but production secret configuration is not proven.",
            {"secret_configured": telegram_secret_configured, "secret_value_read": False},
            "provision_telegram_webhook_secret_then_enable_required_mode_and_canary",
        ),
        _check(
            "DE.CM-mcp-rag-runtime",
            "DETECT",
            "pass" if mcp_enabled and rag_enabled else "partial",
            "high",
            f"Runtime flags: MCP_ROUTER_ENABLED={mcp_enabled}, RAG_ENABLED={rag_enabled}.",
            {"mcp_router_enabled": mcp_enabled, "rag_enabled": rag_enabled},
            "close_mcp_ports_router_and_rag_runtime_readback",
        ),
        _check(
            "AI.RAG-internal-candidate-canary",
            "DETECT/RESPOND",
            "pass" if (scan_root / "services/internal_rag_candidate_canary_service.py").exists() else "fail",
            "high",
            "Internal RAG candidate canary exists." if (scan_root / "services/internal_rag_candidate_canary_service.py").exists() else "PixelRAG candidate knowledge stops before an internal RAG canary.",
            {"service_exists": (scan_root / "services/internal_rag_candidate_canary_service.py").exists()},
            "run_internal_rag_candidate_canary",
        ),
    ]
    checks_by_id = {item["id"]: item for item in checks}
    blocking = [item for item in checks if item["release_blocking"] and item["status"] == "fail"]
    status_counts = {
        status: sum(1 for item in checks if item["status"] == status)
        for status in ("pass", "partial", "fail")
    }
    score_weights = {"pass": 1.0, "partial": 0.5, "fail": 0.0}
    program_percent = round(
        sum(score_weights[item["status"]] for item in checks) / len(checks) * 100,
        1,
    )
    runtime_checks = [
        checks_by_id["PR.AA-production-security-readback"],
        checks_by_id["PR.AC-database-identity-rbac"],
        checks_by_id["DE.CM-mcp-rag-runtime"],
        checks_by_id["AI.RAG-internal-candidate-canary"],
        checks_by_id["RC.RP-restore-drill"],
        checks_by_id["RS.MA-controlled-apply-envelope"],
    ]
    runtime_passed = sum(1 for item in runtime_checks if item["status"] == "pass")

    return {
        "success": True,
        "policy": POLICY,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "status": "complete" if status_counts["fail"] == 0 and status_counts["partial"] == 0 else "partial",
        "answer_to_owner": "The security/governance target is not complete. The project has many local receipts, but access control, identity, asset reconciliation, supply-chain evidence, same-run controlled apply, restore proof and MCP/RAG runtime closure are not yet one production loop.",
        "root_causes": [
            "Feature and receipt growth outran a single deny-by-default security policy.",
            "Source/test/UI evidence was repeatedly counted ahead of production closure evidence.",
            "AI lanes use many preview/review stages without one shared trace/run/work-item envelope.",
            "Asset, dependency, identity, backup and runtime controls are not reconciled by one machine-readable control plane.",
            "Large modules couple policy, execution and verification, making independent assurance expensive.",
        ],
        "benchmark_profile": {
            "nist_csf_2_0": ["GOVERN", "IDENTIFY", "PROTECT", "DETECT", "RESPOND", "RECOVER"],
            "owasp_asvs": "5.0.0",
            "nist_ssdf": "SP 800-218 v1.1",
            "application_level_target": "ASVS L2 with selected L3 controls for operator and automation surfaces",
        },
        "completion": {
            "program_percent": program_percent,
            "asset_field_coverage_percent": assets["field_coverage_percent"],
            "asset_runtime_reconciliation_complete": assets["runtime_receipts_complete"],
            "runtime_closure_passed": runtime_passed,
            "runtime_closure_total": len(runtime_checks),
            "status_counts": status_counts,
        },
        "checks": checks,
        "quality_baseline": quality,
        "runtime_receipt": runtime_receipt,
        "auth_identity_runtime_receipt": auth_identity_runtime,
        "source_receipt": {
            key: value for key, value in source_receipt.items() if key != "_checks"
        },
        "release_gate": {
            "status": "pass" if not blocking else "fail",
            "blocking_failure_count": len(blocking),
            "blocking_failures": [
                {"id": item["id"], "summary": item["summary"]}
                for item in blocking
            ],
            "debt_does_not_equal_release_pass": True,
        },
        "work_items": _work_items(
            checks_by_id,
            runtime_receipt,
            auth_identity_source,
            auth_identity_runtime,
        ),
        "safety": {
            "read_only": True,
            "network_calls": False,
            "database_reads": False,
            "database_writes": False,
            "secret_reads": False,
            "external_side_effects": False,
        },
        "next_machine_action": (
            "advance_to_SEC-P0-003"
            if auth_identity_runtime["runtime_closed"]
            else identity_runtime_next_action
            if auth_identity_runtime["canary_ready"]
            else "deploy_hybrid_database_identity_then_write_runtime_receipt"
            if runtime_receipt["security_readback_passed"] and auth_identity_source["source_ready"]
            else "inventory_active_users_then_shadow_database_auth_before_cutover"
            if runtime_receipt["security_readback_passed"]
            else "deny_public_metrics_then_verify_internal_prometheus_canary"
        ),
    }


__all__ = ["POLICY", "build_security_governance_review"]
