"""Database-driven runtime context for the leoclean Hermes profile."""

from __future__ import annotations

import hashlib
import json
import os
import re
import subprocess
import sys
import threading
from collections import OrderedDict
from collections.abc import Callable
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

DEFAULT_TIMEOUT_SECONDS = 10
MAX_QUERY_CHARS = 16_000
MAX_PENDING_SNAPSHOTS = 128
COMPILED_CONTRACT_IDS = frozenset(
    {
        "mixed_packet_composition",
        "source_intake",
        "proposal_state_readback",
        "named_packet_readback",
        "decision_matrix_readback",
        "source_link_audit",
        "demo_capability_readback",
        "identity_canonicality_readback",
        "telegram_participant_identity",
        "runtime_persistence",
        "shared_claims_agent_positions",
        "forecast_resolution_schema",
        "telegram_delivery_proof",
        "claim_supersession_schema",
    }
)
DECISION_MATRIX_TABLES = tuple(
    f"{schema}.{table}"
    for schema in ("public", "kb_stage")
    for table in ("matrix_voters", "proposal_votes", "proposal_decisions")
)
DATABASE_UNAVAILABLE_RESPONSE = (
    "Live database readback is unavailable, so I cannot safely answer this database-state question from memory. "
    "No canonical, proposal, source, identity, or apply claim should be inferred.\n\n"
    "Next proof-changing follow-up: restore the read-only teleo-kb context lookup and rerun the exact question before "
    "any review, apply, or demo decision."
)

_SNAPSHOT_LOCK = threading.Lock()
_PENDING_SNAPSHOTS: OrderedDict[tuple[str, str], dict[str, Any]] = OrderedDict()


def _trace(record: dict[str, Any]) -> None:
    """Write test-only proof without retaining the operator's message."""

    raw_path = os.getenv("LEO_DB_CONTEXT_TRACE_PATH", "").strip()
    if not raw_path:
        return
    path = Path(raw_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\n")
    try:
        path.chmod(0o600)
    except OSError:
        pass


def _failure_context(reason: str) -> str:
    return (
        '<leo_current_runtime_contracts source="teleo-kb" status="unavailable">\n'
        "The read-only current database contract lookup failed. Do not infer current schema, row state, "
        "proposal applyability, or source-ingestion capability from memory. State that live readback is "
        f"unavailable and use the teleo-kb bridge before making a database claim. Failure: {reason}\n"
        "</leo_current_runtime_contracts>"
    )


def _requires_database_truth(query: str) -> bool:
    lowered = query.lower()
    named_packet_question = "helmer" in lowered and any(
        term in lowered for term in ("7 powers", "seven powers", "powers framework", "powers packet")
    )
    return bool(
        named_packet_question
        or ("telegram" in lowered and "gatewayrunner" in lowered)
        or re.search(
            r"\b(?:databases?|db|knowledge bases?|kb|proposals?|canonical|source_ref|source rows?|"
            r"decision[- ]matri(?:x|ces)|soul\.md|soul file|claim_evidence|claim_edges|reasoning_tools|"
            r"approve_claim|superseded_by|current[- ]schema)\b|"
            r"public\.|kb_stage",
            lowered,
        )
        or any(
            term in lowered
            for term in (
                "document",
                "pdf",
                "tweet",
                "attachment",
                "ingest",
                "intake",
                "absorb",
                "extract",
                "absorb this document",
                "ingest this document",
                "compose the database",
                "strategic framework",
                "governance rule",
            )
        )
    )


def _error_snapshot(query: str, query_sha256: str, reason: str) -> dict[str, Any]:
    requires_database_truth = _requires_database_truth(query)
    return {
        "status": "error",
        "query_sha256": query_sha256,
        "contracts": [],
        "compiled_response": DATABASE_UNAVAILABLE_RESPONSE if requires_database_truth else None,
        "requires_database_truth": requires_database_truth,
        "context": _failure_context(reason),
    }


def _format_context(contracts: list[dict[str, Any]]) -> str:
    payload = json.dumps(contracts, sort_keys=True, separators=(",", ":"))
    return (
        '<leo_current_runtime_contracts source="live teleo-kb context" status="ok">\n'
        "This trusted, read-only block was generated automatically for the current question. It is the "
        "binding contract for current schema and runtime capability. Base the answer on it, do not invent "
        "fields or capabilities, respect its reply budget, and distinguish canonical rows from staging. "
        "Do not claim that you called a tool; this context was injected before reasoning.\n"
        f"{payload}\n"
        "</leo_current_runtime_contracts>"
    )


def _snapshot_key(session_id: str, query_sha256: str) -> tuple[str, str]:
    return (str(session_id or ""), query_sha256)


def _store_snapshot(session_id: str, snapshot: dict[str, Any]) -> None:
    key = _snapshot_key(session_id, str(snapshot["query_sha256"]))
    with _SNAPSHOT_LOCK:
        _PENDING_SNAPSHOTS[key] = snapshot
        _PENDING_SNAPSHOTS.move_to_end(key)
        while len(_PENDING_SNAPSHOTS) > MAX_PENDING_SNAPSHOTS:
            _PENDING_SNAPSHOTS.popitem(last=False)


def _take_snapshot(session_id: str, query_sha256: str) -> dict[str, Any] | None:
    with _SNAPSHOT_LOCK:
        return _PENDING_SNAPSHOTS.pop(_snapshot_key(session_id, query_sha256), None)


def _load_database_snapshot(
    user_message: str,
    *,
    hermes_home: Path | None = None,
    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> dict[str, Any]:
    """Load one immutable DB-contract snapshot for generation and validation."""

    query = str(user_message or "")[:MAX_QUERY_CHARS]
    home = hermes_home or Path(os.getenv("HERMES_HOME", str(Path.home() / ".hermes")))
    kb_tool = home / "bin" / "kb_tool.py"
    query_sha256 = hashlib.sha256(query.encode("utf-8")).hexdigest()
    base_record: dict[str, Any] = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "event": "pre_llm_call",
        "query_sha256": query_sha256,
        "source": "kb_tool.py --local context",
    }

    if not kb_tool.is_file():
        record = base_record | {"status": "error", "error": "kb_tool_missing", "injected": True}
        _trace(record)
        return _error_snapshot(query, query_sha256, "kb_tool_missing")

    timeout = int(os.getenv("LEO_DB_CONTEXT_TIMEOUT_SECONDS", str(DEFAULT_TIMEOUT_SECONDS)))
    command = [
        sys.executable,
        str(kb_tool),
        "--local",
        "context",
        query,
        "--limit",
        "0",
        "--context-limit",
        "0",
        "--format",
        "json",
    ]
    try:
        completed = runner(command, capture_output=True, text=True, timeout=timeout, check=False)
    except subprocess.TimeoutExpired:
        record = base_record | {"status": "error", "error": "timeout", "injected": True}
        _trace(record)
        return _error_snapshot(query, query_sha256, "timeout")
    except OSError as exc:
        record = base_record | {"status": "error", "error": type(exc).__name__, "injected": True}
        _trace(record)
        return _error_snapshot(query, query_sha256, type(exc).__name__)

    if completed.returncode != 0:
        record = base_record | {
            "status": "error",
            "error": f"kb_tool_exit_{completed.returncode}",
            "injected": True,
        }
        _trace(record)
        return _error_snapshot(query, query_sha256, f"kb_tool_exit_{completed.returncode}")

    try:
        payload = json.loads(completed.stdout)
        contracts = payload.get("operational_contracts")
        if not isinstance(contracts, list) or not all(isinstance(item, dict) for item in contracts):
            raise ValueError("operational_contracts_missing")
        compiled_response = payload.get("compiled_response")
        if compiled_response is not None and not isinstance(compiled_response, str):
            raise ValueError("compiled_response_invalid")
    except (json.JSONDecodeError, TypeError, ValueError) as exc:
        record = base_record | {"status": "error", "error": str(exc), "injected": True}
        _trace(record)
        return _error_snapshot(query, query_sha256, str(exc))

    contract_json = json.dumps(contracts, sort_keys=True, separators=(",", ":"))
    record = base_record | {
        "status": "ok",
        "injected": True,
        "contract_ids": [str(item.get("id") or "") for item in contracts],
        "contract_sha256": hashlib.sha256(contract_json.encode("utf-8")).hexdigest(),
        "compiled_response_available": bool(compiled_response),
    }
    _trace(record)
    return {
        "status": "ok",
        "query_sha256": query_sha256,
        "contracts": contracts,
        "compiled_response": compiled_response,
        "requires_database_truth": bool({str(item.get("id") or "") for item in contracts} - {"reply_budget"}),
        "context": _format_context(contracts),
    }


def build_database_context(
    user_message: str,
    *,
    hermes_home: Path | None = None,
    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> str:
    """Return current operational contracts for one exact operator question."""

    return str(_load_database_snapshot(user_message, hermes_home=hermes_home, runner=runner)["context"])


def _word_count(value: str) -> int:
    return len(re.findall(r"\b\w+(?:[-']\w+)*\b", value))


def _reply_hard_max(contracts: list[dict[str, Any]]) -> int | None:
    for contract in contracts:
        if contract.get("id") == "reply_budget" and isinstance(contract.get("hard_max_words"), int):
            return int(contract["hard_max_words"])
    return None


def _mixed_packet_issues(response: str) -> list[str]:
    lowered = response.lower()
    issues: list[str] = []
    required_terms = {
        "claims": "claim",
        "sources": "source",
        "evidence": "evidence",
        "reasoning_tools": "reasoning_tools",
        "behavioral_rules": "behavioral_rules",
        "governance_gates": "governance_gates",
        "beliefs": "belief",
        "staging": "stage",
        "approve_claim": "approve_claim",
    }
    for label, term in required_terms.items():
        if term not in lowered:
            issues.append(f"missing_{label}")

    if not re.search(r"reviewed apply|review and authoriz|review.{0,50}apply", response, re.I | re.S):
        issues.append("missing_reviewed_apply_boundary")
    if not ("applied_at" in lowered and re.search(r"readback|postflight", response, re.I)):
        issues.append("missing_apply_receipt")
    if not re.search(
        r"approve_claim.{0,220}(?:supports neither|does not (?:support|cover|apply)|cannot apply|applies only)",
        response,
        re.I | re.S,
    ):
        issues.append("missing_approve_claim_boundary")

    if re.search(
        r"supported collections?\s*\([^)]*(?:belief|behavioral_rules|governance_gates|existing[- ]row)",
        response,
        re.I | re.S,
    ):
        issues.append("unsupported_collection_listed_as_supported")
    if re.search(
        r"(?:approve_claim\s+)?(?:applies?|apply) only[^.;\n]{0,220}"
        r"(?:belief updates?|behavioral_rules|governance_gates|existing[- ]row updates?)",
        response,
        re.I | re.S,
    ):
        issues.append("unsupported_collection_listed_in_apply")

    for segment in re.split(r"(?<=[.!?])\s+|\n+", response):
        negative = re.search(r"\b(?:no|not|never|without|must not|do not|does not|cannot)\b", segment, re.I)
        if re.search(r"reasoning_tools?.{0,100}\bscope\b", segment, re.I) and not negative:
            issues.append("reasoning_tools_scope_overclaim")
        if re.search(r"claims?.{0,100}\bfalsifier\b", segment, re.I) and not negative:
            issues.append("claims_falsifier_overclaim")
        if re.search(r"behavioral_rules?.{0,100}\bstatus\b", segment, re.I) and not negative:
            issues.append("behavioral_rules_status_overclaim")
        if (
            re.search(r"claim_edges?.{0,140}(?:reasoning_tools?|beliefs?)", segment, re.I | re.S)
            and re.search(r"point|connect|link|endpoint", segment, re.I)
            and not negative
        ):
            issues.append("non_claim_edge_endpoint")
    return sorted(set(issues))


def _source_intake_issues(response: str) -> list[str]:
    lowered = response.lower()
    issues: list[str] = []
    if not re.search(r"url|storage_path|storage path|content hash|file hash", response, re.I):
        issues.append("missing_real_source_identity")
    if not re.search(r"pending_review|pending review|proposal", response, re.I):
        issues.append("missing_staging")
    if not re.search(r"approval|authoriz", response, re.I) or "apply" not in lowered:
        issues.append("missing_apply_boundary")
    if not re.search(r"live.{0,80}(?:not shipped|not live|unavailable)", response, re.I | re.S):
        issues.append("missing_live_intake_ceiling")
    for segment in re.split(r"(?<=[.!?])\s+|\n+", response):
        negative = re.search(
            r"\b(?:no|not|never|without|must not|do not|does not|cannot|isn't|aren't)\b",
            segment,
            re.I,
        )
        if (
            re.search(
                r"public\.sources.{0,120}\b(?:author|title|publisher|publication date|published_at)\b",
                segment,
                re.I | re.S,
            )
            and not negative
        ):
            issues.append("unshipped_source_fields")
        if (
            re.search(
                r"\b(?:create|write|insert)\b.{0,80}public\.sources.{0,100}"
                r"(?:before|during).{0,80}(?:review|staging)",
                segment,
                re.I | re.S,
            )
            and not negative
        ):
            issues.append("canonical_source_before_review")
    return sorted(set(issues))


def _contract_map(contracts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    return {str(contract.get("id") or ""): contract for contract in contracts}


def _expected_count_readback(status_data: dict[str, Any] | None) -> str | None:
    counts = (status_data or {}).get("high_signal_rows") or {}
    labels = ("claims", "sources", "claim_edges", "claim_evidence", "kb_proposals")
    if not all(isinstance(counts.get(label), int) for label in labels):
        return None
    return "DB readback: " + "; ".join(f"{label}: {counts[label]}" for label in labels) + "."


def _has_expected_count_readback(response: str, contract: dict[str, Any]) -> bool:
    expected = _expected_count_readback(contract.get("database_status"))
    return bool(expected and expected.lower() in response.lower())


def _has_proposal_readback(response: str, proposal: dict[str, Any]) -> bool:
    readback_lines = re.findall(r"^\s*DB readback:\s*([^\n]+)$", response, re.I | re.M)
    expected_id = str(proposal.get("id") or "")
    expected_status = str(proposal.get("status") or "")
    expected_applied_at = str(proposal.get("applied_at") or "none")
    return any(
        expected_id in line
        and re.search(rf"\bstatus\s*:\s*{re.escape(expected_status)}\b", line, re.I)
        and re.search(rf"\bapplied_at\s*:\s*{re.escape(expected_applied_at)}\b", line, re.I)
        for line in readback_lines
    )


def _has_status_count(response: str, status: str, count: Any) -> bool:
    return isinstance(count, int) and bool(re.search(rf"\b{re.escape(status)}\s*:\s*{count}\b", response, re.I))


def _live_readback_issues(response: str, contracts: list[dict[str, Any]]) -> list[str]:
    """Require the delivered answer to carry the exact immutable DB snapshot."""

    by_id = _contract_map(contracts)
    active_ids = {
        "proposal_state_readback",
        "named_packet_readback",
        "decision_matrix_readback",
        "source_link_audit",
        "demo_capability_readback",
        "identity_canonicality_readback",
    } & set(by_id)
    if not active_ids:
        return []

    lowered = response.lower()
    issues: list[str] = []
    if "Next proof-changing follow-up:" not in response:
        issues.append("missing_proof_changing_followup")

    count_contract_ids = active_ids - {"named_packet_readback"}
    for contract_id in sorted(count_contract_ids):
        if not _has_expected_count_readback(response, by_id[contract_id]):
            issues.append(f"missing_live_count_readback:{contract_id}")

    proposal_state = by_id.get("proposal_state_readback")
    if proposal_state:
        states = (proposal_state.get("database_status") or {}).get("proposal_status_counts") or {}
        for state in ("applied", "approved", "pending_review", "canceled"):
            if not _has_status_count(response, state, states.get(state)):
                issues.append(f"missing_live_proposal_state:{state}")
        if "approved is not applied" not in lowered or "applied_at" not in lowered:
            issues.append("missing_proposal_apply_boundary")
        if re.search(r"all approved (?:rows|proposals).{0,40}(?:already )?(?:canonical|applied|live)", response, re.I):
            issues.append("contradictory_approved_state_claim")

    named_packet = by_id.get("named_packet_readback")
    if named_packet:
        matches = list(named_packet.get("matching_proposals") or [])
        primary = matches[0] if matches else None
        if primary and not _has_proposal_readback(response, primary):
            issues.append("missing_named_proposal_readback")
        if not primary and not _has_expected_count_readback(response, named_packet):
            issues.append("missing_named_count_readback")
        canonical_matches = named_packet.get("canonical_matches") or {}
        for table in ("reasoning_tools", "claims", "sources"):
            expected_count = len(canonical_matches.get(table) or [])
            if not _has_status_count(response, table, expected_count):
                issues.append(f"missing_named_canonical_count:{table}")
        for term in ("helmer", "canonical", "kb_stage", "applied_at", "authorization", "postflight"):
            if term not in lowered:
                issues.append(f"missing_named_packet_term:{term}")

    matrix = by_id.get("decision_matrix_readback")
    if matrix:
        table_states = (matrix.get("live_matrix_status") or {}).get("decision_matrix_tables") or {}
        unobserved_tables = [name for name in DECISION_MATRIX_TABLES if table_states.get(name) not in (True, False)]
        absent_tables = [name for name in DECISION_MATRIX_TABLES if table_states.get(name) is False]
        if unobserved_tables:
            if (
                any(name.lower() not in lowered for name in unobserved_tables)
                or "incomplete" not in lowered
                or "not observed" not in lowered
                or "cannot be inferred" not in lowered
            ):
                issues.append("missing_live_matrix_table_readback")
        elif absent_tables:
            if any(name.lower() not in lowered for name in absent_tables):
                issues.append("missing_live_matrix_table_readback")
        else:
            if not re.search(r"all required matrix tables are present|matrix schema is present", response, re.I):
                issues.append("missing_live_matrix_table_readback")
            if "proposal-specific" not in lowered or "not proven" not in lowered:
                issues.append("missing_matrix_decision_readback_ceiling")
        if "not a decision-matrix vote" not in lowered:
            issues.append("missing_matrix_approval_boundary")

    source_audit = by_id.get("source_link_audit")
    if source_audit:
        audit = source_audit.get("live_link_audit") or {}
        expected_fragments = (
            f"{audit.get('pending_or_approved')} pending/approved proposals",
            f"{audit.get('with_source_ref')} have source_ref",
            f"{audit.get('exact_public_source_matches')} exactly match",
        )
        if any(fragment.lower() not in lowered for fragment in expected_fragments):
            issues.append("missing_live_source_link_audit")
        for term in ("telegram_file_refs", "document_evaluations", "public.sources", "claim_evidence"):
            if term not in lowered:
                issues.append(f"missing_source_link_layer:{term}")
        if "does not" not in lowered and "not proven" not in lowered:
            issues.append("missing_source_link_causality_boundary")

    demo = by_id.get("demo_capability_readback")
    if demo:
        states = (demo.get("database_status") or {}).get("proposal_status_counts") or {}
        for state in ("applied", "approved", "pending_review"):
            if not _has_status_count(response, state, states.get(state)):
                issues.append(f"missing_demo_state:{state}")
        for term in ("demo", "pending_review", "authorization", "postflight", "artifact"):
            if term not in lowered:
                issues.append(f"missing_demo_term:{term}")
        if "does not prove" not in lowered:
            issues.append("missing_demo_claim_ceiling")

    identity = by_id.get("identity_canonicality_readback")
    if identity:
        for term in ("soul.md", "runtime/profile", "canonical postgres rows", "render/sync", "applied_at"):
            if term not in lowered:
                issues.append(f"missing_identity_term:{term}")
        if "does not change canonical postgres rows" not in lowered:
            issues.append("missing_identity_canonicality_boundary")
        if not re.search(r"not proven|no active.{0,100}proven", response, re.I | re.S):
            issues.append("missing_identity_renderer_ceiling")

    return sorted(set(issues))


def response_contract_issues(response: str, contracts: list[dict[str, Any]]) -> list[str]:
    contract_ids = {str(contract.get("id") or "") for contract in contracts}
    issues: list[str] = []
    hard_max = _reply_hard_max(contracts)
    if hard_max is not None and _word_count(response) > hard_max:
        issues.append("reply_budget_exceeded")
    if "mixed_packet_composition" in contract_ids:
        issues.extend(_mixed_packet_issues(response))
    elif "source_intake" in contract_ids:
        issues.extend(_source_intake_issues(response))
    issues.extend(_live_readback_issues(response, contracts))
    return sorted(set(issues))


def _pre_llm_call(**kwargs: Any) -> dict[str, str]:
    user_message = str(kwargs.get("user_message") or "")
    snapshot = _load_database_snapshot(user_message)
    _store_snapshot(str(kwargs.get("session_id") or ""), snapshot)
    return {"context": str(snapshot["context"])}


def _post_llm_call(**kwargs: Any) -> dict[str, str] | None:
    user_message = str(kwargs.get("user_message") or "")[:MAX_QUERY_CHARS]
    query_sha256 = hashlib.sha256(user_message.encode("utf-8")).hexdigest()
    snapshot = _take_snapshot(str(kwargs.get("session_id") or ""), query_sha256)
    response = str(kwargs.get("assistant_response") or "")
    base_record = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "event": "post_llm_call",
        "query_sha256": query_sha256,
        "response_sha256": hashlib.sha256(response.encode("utf-8")).hexdigest(),
        "validated": True,
    }
    if snapshot is None:
        fail_closed = _requires_database_truth(user_message)
        delivered = DATABASE_UNAVAILABLE_RESPONSE if fail_closed else response
        _trace(
            base_record
            | {
                "status": "error",
                "error": "contract_snapshot_missing",
                "transformed": fail_closed,
                "fail_closed": fail_closed,
                "delivered_response_sha256": hashlib.sha256(delivered.encode("utf-8")).hexdigest(),
            }
        )
        if fail_closed:
            return {"assistant_response": DATABASE_UNAVAILABLE_RESPONSE}
        return None
    if snapshot.get("status") != "ok":
        fail_closed = bool(snapshot.get("requires_database_truth"))
        delivered = DATABASE_UNAVAILABLE_RESPONSE if fail_closed else response
        _trace(
            base_record
            | {
                "status": "error" if fail_closed else "skipped",
                "reason": "contract_lookup_failed",
                "transformed": fail_closed,
                "fail_closed": fail_closed,
                "delivered_response_sha256": hashlib.sha256(delivered.encode("utf-8")).hexdigest(),
            }
        )
        if fail_closed:
            return {"assistant_response": DATABASE_UNAVAILABLE_RESPONSE}
        return None

    contracts = list(snapshot.get("contracts") or [])
    contract_ids = {str(item.get("id") or "") for item in contracts}
    issues = response_contract_issues(response, contracts)
    compiled_response = snapshot.get("compiled_response")
    compiled_issues = (
        response_contract_issues(str(compiled_response), contracts) if isinstance(compiled_response, str) else []
    )
    compiled_required = bool(contract_ids & COMPILED_CONTRACT_IDS)
    compiled_valid = bool(isinstance(compiled_response, str) and compiled_response.strip() and not compiled_issues)
    fail_closed = bool(compiled_required and not compiled_valid)
    enforce_compiled = bool(compiled_required and compiled_valid)
    transformed = bool(fail_closed or (enforce_compiled and str(compiled_response) != response))
    if fail_closed:
        delivered = DATABASE_UNAVAILABLE_RESPONSE
    elif enforce_compiled or (issues and compiled_valid):
        delivered = str(compiled_response)
    else:
        delivered = response
    _trace(
        base_record
        | {
            "status": "error" if fail_closed else "ok" if not issues or delivered != response else "error",
            "contract_ids": sorted(contract_ids),
            "issues": issues,
            "compiled_response_issues": compiled_issues,
            "transformed": transformed,
            "compiled_response_enforced": enforce_compiled,
            "fail_closed": fail_closed,
            "delivered_response_sha256": hashlib.sha256(delivered.encode("utf-8")).hexdigest(),
        }
    )
    if delivered != response:
        return {"assistant_response": delivered}
    return None


def register(ctx: Any) -> None:
    ctx.register_hook("pre_llm_call", _pre_llm_call)
    ctx.register_hook("post_llm_call", _post_llm_call)