"""Session-immutable MCP mutation context (#714).

Pins profile, remote, host, repository, identity, and role for the life of an
MCP process (or until an explicit ``gitea_activate_profile`` re-bind).

Capability resolution and mutation gates must evaluate only the active
profile for the requested remote. Silent cross-host / cross-profile
substitution is forbidden.
"""

from __future__ import annotations

import os
import threading
import urllib.parse
from dataclasses import dataclass
from typing import Any, Mapping


@dataclass(frozen=True, slots=True)
class _SessionContext:
    """One atomic, immutable process-session binding."""

    profile_name: str | None
    remote: str | None
    host: str | None
    identity: str | None
    repository: str | None
    org: str | None
    role_kind: str | None
    expected_username: str | None
    source: str
    pid: int

    def as_dict(self) -> dict[str, Any]:
        return {
            "profile_name": self.profile_name,
            "remote": self.remote,
            "host": self.host,
            "identity": self.identity,
            "repository": self.repository,
            "org": self.org,
            "role_kind": self.role_kind,
            "expected_username": self.expected_username,
            "source": self.source,
            "pid": self.pid,
        }


# Process-local only — never a shared file (same rationale as mutation authority).
# The frozen value prevents partial mutation, while the lock makes first-bind and
# sanctioned rebind atomic across concurrent MCP calls.
_SESSION_CONTEXT: _SessionContext | None = None
_SESSION_CONTEXT_LOCK = threading.RLock()


def _reset_session_context_for_testing() -> None:
    """Reset at a pytest test boundary; unavailable to production callers.

    Production sessions transition only through process startup/PID change or
    the explicit profile-activation rebind.  Keeping this helper private and
    requiring pytest's per-test marker prevents it from becoming an MCP/runtime
    bypass.
    """
    if "PYTEST_CURRENT_TEST" not in os.environ:
        raise RuntimeError("session context reset is restricted to pytest boundaries")
    global _SESSION_CONTEXT
    with _SESSION_CONTEXT_LOCK:
        _SESSION_CONTEXT = None


def get_session_context() -> dict[str, Any] | None:
    """Return a detached snapshot of the bound context, or None if unbound."""
    with _SESSION_CONTEXT_LOCK:
        if _SESSION_CONTEXT is None:
            return None
        return _SESSION_CONTEXT.as_dict()


def profile_host(profile: dict | None) -> str | None:
    """Hostname from profile base_url, lowercased, or None."""
    if not profile:
        return None
    base = (profile.get("base_url") or "").strip()
    if not base:
        return None
    try:
        parsed = urllib.parse.urlparse(base)
        host = (parsed.netloc or parsed.path or "").strip().lower()
        return host or None
    except Exception:
        return None


def remote_host(remote: str | None, remotes: dict | None) -> str | None:
    """Hostname for a known remote key."""
    if not remote or not remotes:
        return None
    entry = remotes.get(remote) or {}
    return (entry.get("host") or "").strip().lower() or None


def profile_matches_remote(
    profile: dict | None,
    remote: str | None,
    remotes: dict | None,
    *,
    contexts: dict | None = None,
) -> bool:
    """True when *profile* is bound to the same host/context as *remote*."""
    if not profile or not remote:
        return False
    r_host = remote_host(remote, remotes)
    p_host = profile_host(profile)
    if r_host and p_host:
        return r_host == p_host
    # Fall back to context name heuristics when base_url missing.
    ctx = (profile.get("context") or "").strip().lower()
    if not ctx or not contexts:
        return False
    ctx_data = contexts.get(ctx) or {}
    gitea = ctx_data.get("gitea") or {}
    base = (gitea.get("base_url") or "").strip()
    if not base or not r_host:
        return False
    try:
        parsed = urllib.parse.urlparse(base)
        c_host = (parsed.netloc or parsed.path or "").strip().lower()
    except Exception:
        return False
    return bool(c_host) and c_host == r_host


def bind_session_context(
    *,
    profile_name: str,
    remote: str | None,
    host: str | None,
    identity: str | None,
    repository: str | None = None,
    org: str | None = None,
    role_kind: str | None = None,
    expected_username: str | None = None,
    source: str = "bind",
) -> dict[str, Any]:
    """Atomically bind/re-bind context (the explicit activation path)."""
    with _SESSION_CONTEXT_LOCK:
        return _bind_session_context_unlocked(
            profile_name=profile_name,
            remote=remote,
            host=host,
            identity=identity,
            repository=repository,
            org=org,
            role_kind=role_kind,
            expected_username=expected_username,
            source=source,
        )


def _bind_session_context_unlocked(
    *,
    profile_name: str,
    remote: str | None,
    host: str | None,
    identity: str | None,
    repository: str | None,
    org: str | None,
    role_kind: str | None,
    expected_username: str | None,
    source: str,
) -> dict[str, Any]:
    """Store a complete immutable context while the caller holds the lock."""
    global _SESSION_CONTEXT
    _SESSION_CONTEXT = _SessionContext(
        profile_name=(profile_name or "").strip() or None,
        remote=(remote or "").strip() or None,
        host=(host or "").strip().lower() or None,
        identity=(identity or "").strip() or None,
        repository=(repository or "").strip() or None,
        org=(org or "").strip() or None,
        role_kind=(role_kind or "").strip() or None,
        expected_username=(expected_username or "").strip() or None,
        source=source,
        pid=os.getpid(),
    )
    return _SESSION_CONTEXT.as_dict()


def seed_session_context_if_unbound(
    *,
    profile_name: str,
    remote: str | None,
    host: str | None,
    identity: str | None,
    repository: str | None = None,
    org: str | None = None,
    role_kind: str | None = None,
    expected_username: str | None = None,
    source: str = "seed",
) -> dict[str, Any]:
    """Atomically bind only when this process has no current context.

    A changed environment or an interleaved call is not a session boundary and
    therefore cannot replace an established binding.  A newly started/forked
    process is recognized by PID; explicit ``gitea_activate_profile`` uses
    :func:`bind_session_context` as its sanctioned logical-session transition.
    """
    with _SESSION_CONTEXT_LOCK:
        if _SESSION_CONTEXT is None or _SESSION_CONTEXT.pid != os.getpid():
            return _bind_session_context_unlocked(
                profile_name=profile_name,
                remote=remote,
                host=host,
                identity=identity,
                repository=repository,
                org=org,
                role_kind=role_kind,
                expected_username=expected_username,
                source=source,
            )
        return _SESSION_CONTEXT.as_dict()


def assess_session_context(
    *,
    profile_name: str | None,
    remote: str | None,
    host: str | None = None,
    identity: str | None = None,
    repository: str | None = None,
    org: str | None = None,
    expected_username: str | None = None,
    require_bound: bool = False,
) -> dict[str, Any]:
    """Compare live values against the bound session context.

    Returns ``proven`` / ``block`` / ``reasons``. When unbound and
    ``require_bound`` is false, does not block (caller may seed). When
    unbound and ``require_bound`` is true, fails closed.
    """
    reasons: list[str] = []
    with _SESSION_CONTEXT_LOCK:
        bound = _SESSION_CONTEXT
        ctx = bound.as_dict() if bound is not None else None
    if ctx is None or ctx.get("pid") != os.getpid():
        if require_bound:
            reasons.append(
                "session mutation context is unbound; call gitea_whoami or "
                "gitea_activate_profile before mutating (fail closed)"
            )
            return _assessment(False, reasons, ctx)
        return _assessment(True, reasons, ctx)

    live_profile = (profile_name or "").strip() or None
    live_remote = (remote or "").strip() or None
    live_host = (host or "").strip().lower() or None
    live_identity = (identity or "").strip() or None
    live_repo = (repository or "").strip() or None
    live_org = (org or "").strip() or None

    if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"):
        reasons.append(
            f"profile drift: live '{live_profile}' != bound "
            f"'{ctx.get('profile_name')}' (fail closed)"
        )
    if ctx.get("remote") and live_remote and live_remote != ctx.get("remote"):
        reasons.append(
            f"remote drift: live '{live_remote}' != bound "
            f"'{ctx.get('remote')}' (fail closed)"
        )
    if ctx.get("host") and live_host and live_host != ctx.get("host"):
        reasons.append(
            f"host drift: live '{live_host}' != bound "
            f"'{ctx.get('host')}' (fail closed)"
        )
    if (
        ctx.get("identity")
        and live_identity
        and live_identity != ctx.get("identity")
    ):
        reasons.append(
            f"identity drift: live '{live_identity}' != bound "
            f"'{ctx.get('identity')}' (fail closed)"
        )
    if ctx.get("repository") and live_repo and live_repo != ctx.get("repository"):
        reasons.append(
            f"repository drift: live '{live_repo}' != bound "
            f"'{ctx.get('repository')}' (fail closed)"
        )
    if ctx.get("org") and live_org and live_org != ctx.get("org"):
        reasons.append(
            f"org drift: live '{live_org}' != bound "
            f"'{ctx.get('org')}' (fail closed)"
        )

    expected = expected_username or ctx.get("expected_username")
    if expected and live_identity and live_identity != expected:
        reasons.append(
            f"identity mismatch: authenticated '{live_identity}' != "
            f"profile expected '{expected}' (fail closed)"
        )

    return _assessment(not reasons, reasons, ctx)


def assess_identity_match(
    *,
    authenticated: str | None,
    expected_username: str | None,
) -> dict[str, Any]:
    """Fail closed when profile declares a username that does not match live auth."""
    reasons: list[str] = []
    auth = (authenticated or "").strip() or None
    expected = (expected_username or "").strip() or None
    if expected and auth and auth != expected:
        reasons.append(
            f"identity mismatch: authenticated '{auth}' != "
            f"profile expected '{expected}' (fail closed)"
        )
    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "authenticated": auth,
        "expected_username": expected,
    }


def filter_profiles_for_remote(
    config: dict | None,
    remote: str | None,
    remotes: dict | None,
) -> list[str]:
    """Profile names whose host/context matches *remote* (enabled only)."""
    if not config or not remote:
        return []
    profiles = config.get("profiles") or {}
    contexts = config.get("contexts") or {}
    names: list[str] = []
    for name, data in profiles.items():
        if not isinstance(data, dict):
            continue
        if not data.get("enabled", True):
            continue
        if profile_matches_remote(data, remote, remotes, contexts=contexts):
            names.append(name)
    return sorted(names)


def profile_allowed_for_remote(
    profile: dict | None,
    remote: str | None,
    remotes: dict | None,
    *,
    contexts: dict | None = None,
) -> dict[str, Any]:
    """Assess whether the active profile may serve *remote*."""
    reasons: list[str] = []
    if not profile:
        reasons.append("active profile unresolved (fail closed)")
        return {"proven": False, "block": True, "reasons": reasons}
    if not remote:
        return {"proven": True, "block": False, "reasons": reasons}
    # Legacy env-only profiles have no configured base URL or v2 context. Their
    # first call may establish the process remote/host pin; after that,
    # assess_session_context rejects any drift. Configured v2 profiles still
    # require positive host/context alignment here.
    if not profile_host(profile) and not (profile.get("context") or "").strip():
        return {"proven": True, "block": False, "reasons": reasons}
    if not profile_matches_remote(profile, remote, remotes, contexts=contexts):
        p_name = profile.get("profile_name") or profile.get("name") or "(unknown)"
        p_host = profile_host(profile) or "(none)"
        r_host = remote_host(remote, remotes) or "(none)"
        reasons.append(
            f"cross-host profile denial: profile '{p_name}' (host '{p_host}') "
            f"cannot serve remote '{remote}' (host '{r_host}') (fail closed)"
        )
    return {"proven": not reasons, "block": bool(reasons), "reasons": reasons}


def mutation_context_audit_fields(
    ctx: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    """Fields to include in pre-mutation audit records."""
    data = ctx if ctx is not None else get_session_context()
    if not data:
        return {
            "session_context_bound": False,
            "session_profile": None,
            "session_remote": None,
            "session_host": None,
            "session_identity": None,
            "session_repository": None,
            "session_org": None,
        }
    return {
        "session_context_bound": True,
        "session_profile": data.get("profile_name"),
        "session_remote": data.get("remote"),
        "session_host": data.get("host"),
        "session_identity": data.get("identity"),
        "session_repository": data.get("repository"),
        "session_org": data.get("org"),
        "session_role_kind": data.get("role_kind"),
        "session_context_source": data.get("source"),
    }


def _assessment(
    proven: bool, reasons: list[str], ctx: Mapping[str, Any] | None
) -> dict[str, Any]:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": list(reasons),
        "bound_context": dict(ctx) if ctx else None,
        "audit": mutation_context_audit_fields(ctx),
    }