"""Durable MCP session validation state shared across daemon process pools (#559).

The IDE often routes sequential MCP tool calls to different daemon processes.
Session-scoped proofs (workflow load, review decision lock) must therefore
survive process boundaries while remaining fail-closed against spoofing.

Security notes (extends #211):
- Never store under host-global ``/tmp`` (world-writable, spoofable).
- Default root is ``~/.cache/gitea-tools/session-state`` (mode ``0o700``).
- Files are written atomically with mode ``0o600``.
- Records are keyed by remote + org + repo + profile identity, not by PID.
- TTL prevents indefinitely stale reuse across unrelated sessions.
"""

from __future__ import annotations

import fcntl
import json
import os
import re
import tempfile
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Any

STATE_DIR_ENV = "GITEA_MCP_SESSION_STATE_DIR"
DEFAULT_STATE_DIR = os.path.expanduser("~/.cache/gitea-tools/session-state")
TTL_HOURS_ENV = "GITEA_MCP_SESSION_STATE_TTL_HOURS"
DEFAULT_TTL_HOURS = 4.0

KIND_WORKFLOW_LOAD = "review_workflow_load"
KIND_DECISION_LOCK = "review_decision_lock"
KIND_REVIEW_DRAFT = "review_draft"
# Durable marker set when a worker session attempts a direct stable-branch push
# or a root-checkout local commit (#671). Keyed per profile identity like the
# other session proofs; a contaminated session fails closed on gated mutations
# until a reconciler audits and clears it.
KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
# Durable shadow of the in-memory reviewer session lease (#702). Written on
# every sanctioned record/heartbeat and removed on sanctioned clear, so a
# daemon that dies without teardown leaves provable orphan evidence (owner
# pid + session id) for the guarded lease-cleanup path. Never a substitute
# for the in-session lease: mutation gates ignore it.
KIND_REVIEWER_SESSION_LEASE = "reviewer_session_lease"
# Durable audit record written whenever the daemon's sanctioned stale-binding
# recovery clears an inherited GITEA_ACTIVE_WORKTREE binding (#702).
KIND_STALE_BINDING_RECOVERY = "stale_binding_recovery"
# #709: archive prior terminal decision ledgers instead of silent overwrite.
KIND_DECISION_LOCK_ARCHIVE = "review_decision_lock_archive"
# #709: post-merge cleanup/audit reconciliation-required durable record.
KIND_POST_MERGE_DECISION_RECOVERY = "post_merge_decision_recovery"
# #709: truthful record when historical terminal evidence is irrecoverably gone.
KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
# #709 F1: server-side non-forgeable authorization artifact for recovery.
KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"

# Kinds that must survive the default session-state TTL (forensic / recovery).
#
# KIND_DECISION_LOCK is recovery-critical (#720): terminal review provenance is
# not disposable cache. A generic four-hour TTL must not drop old-head evidence
# or make ``fresh_review_on_current_head_allowed`` unreachable. Same-head / same-
# run #332 protections still apply once the ledger is loadable. Other session
# kinds (workflow load, drafts, etc.) remain TTL-bound.
RECOVERY_CRITICAL_KINDS = frozenset(
    {
        KIND_DECISION_LOCK,
        KIND_DECISION_LOCK_ARCHIVE,
        KIND_POST_MERGE_DECISION_RECOVERY,
        KIND_IRRECOVERABLE_DECISION_PROVENANCE,
        KIND_IRRECOVERABLE_PROVENANCE_AUTH,
        # #702 crash-orphan evidence (must outlive TTL; F4)
        KIND_REVIEWER_SESSION_LEASE,
        KIND_STALE_BINDING_RECOVERY,
    }
)


_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"


def default_state_dir() -> str:
    """Resolve the durable session-state root.

    When production native MCP transport is bound (#695 AC2), the directory
    pinned at transport bind is authoritative: later overrides of
    ``GITEA_MCP_SESSION_STATE_DIR`` cannot manufacture a second authority
    domain (PR #701 recurrence: ``.mcp_session_701`` evasion of cross-PR
    decision locks).
    """
    try:
        import mcp_daemon_guard

        pinned = mcp_daemon_guard.pinned_session_state_dir()
        if pinned:
            return pinned
    except Exception:
        # Fail open to env/default only when guard is unavailable (e.g. partial
        # import during bootstrap). Mutation gates still fail closed separately.
        pass
    raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
    return raw or DEFAULT_STATE_DIR


def env_session_state_dir_unpinned() -> str:
    """Raw env/default session-state dir ignoring production transport pin.

    Intended for diagnostics and tests that assert pin behavior — not for
    mutation-sensitive durable proofs under a bound native daemon.
    """
    raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
    return raw or DEFAULT_STATE_DIR


def ttl_hours() -> float:
    raw = (os.environ.get(TTL_HOURS_ENV) or "").strip()
    if not raw:
        return DEFAULT_TTL_HOURS
    try:
        value = float(raw)
    except ValueError:
        return DEFAULT_TTL_HOURS
    return value if value > 0 else DEFAULT_TTL_HOURS


def _sanitize_segment(value: str) -> str:
    text = (value or "").strip()
    if not text:
        return "_"
    return _SAFE_SEGMENT_RE.sub("_", text)


def current_profile_identity(
    profile_name: str | None = None,
    session_profile_lock: str | None = None,
    profile_identity: str | None = None,
) -> str:
    """Resolve the session profile identity used as the durable key."""
    env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
    explicit = (profile_identity or session_profile_lock or "").strip()
    lock = (explicit or env_lock or "").strip()
    name = (profile_name or "").strip()
    return lock or name or "unknown-profile"


def state_key(
    *,
    kind: str,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    profile_identity: str | None = None,
    instance_id: str | None = None,
) -> str:
    """Build durable filename key.

    Session proofs are one-active-per-profile (workflow load / decision lock),
    so the primary key is kind + profile identity. Remote/org/repo are stored
    inside the payload and validated on load (#559), which lets a later daemon
    process recover state without already knowing the remote argument.

    *instance_id* extends the key for kinds where one-per-profile collides —
    concurrent same-profile sessions each own their reviewer-lease shadow
    (#702 F5), keyed by their lease session id so a later heartbeat can never
    overwrite or misattribute another session's crash evidence.
    """
    # Keep remote/org/repo parameters for API stability / future kinds; they are
    # intentionally not part of the filename for session-scoped proofs.
    _ = (remote, org, repo)
    parts = [kind, profile_identity or "unknown-profile"]
    instance = (instance_id or "").strip()
    if instance:
        parts.append(instance)
    return "-".join(_sanitize_segment(part) for part in parts)


def state_file_path(
    *,
    kind: str,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    profile_identity: str | None = None,
    state_dir: str | None = None,
    instance_id: str | None = None,
) -> str:
    root = (state_dir or default_state_dir()).strip()
    name = state_key(
        kind=kind,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile_identity,
        instance_id=instance_id,
    )
    return os.path.join(root, f"{name}.json")


def _ensure_state_dir(state_dir: str | None = None) -> str:
    root = (state_dir or default_state_dir()).strip()
    os.makedirs(root, mode=0o700, exist_ok=True)
    return root


def _now_utc() -> datetime:
    return datetime.now(timezone.utc)


def _parse_iso(value: str | None) -> datetime | None:
    text = (value or "").strip()
    if not text:
        return None
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


@contextmanager
def _exclusive_file_lock(lock_path: str):
    os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX)
        yield fd
    finally:
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)


def _read_json(path: str) -> dict[str, Any] | None:
    if not path or not os.path.exists(path):
        return None
    try:
        with open(path, encoding="utf-8") as handle:
            data = json.load(handle)
    except (OSError, json.JSONDecodeError):
        return None
    return data if isinstance(data, dict) else None


def _write_json(path: str, data: dict[str, Any]) -> None:
    parent = os.path.dirname(path) or "."
    os.makedirs(parent, mode=0o700, exist_ok=True)
    payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
    fd, temp_path = tempfile.mkstemp(prefix=".session-", suffix=".json", dir=parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(temp_path, 0o600)
        os.replace(temp_path, path)
        try:
            os.chmod(path, 0o600)
        except OSError:
            pass
    finally:
        if os.path.exists(temp_path):
            try:
                os.remove(temp_path)
            except OSError:
                pass


def is_recovery_critical_record(record: dict[str, Any] | None, kind: str | None = None) -> bool:
    """True when a durable record must outlive the generic session-state TTL."""
    if not record and not kind:
        return False
    record_kind = ((record or {}).get("kind") or kind or "").strip()
    if record_kind in RECOVERY_CRITICAL_KINDS:
        return True
    return bool((record or {}).get("recovery_critical"))


def identity_match_reasons(
    record: dict[str, Any] | None,
    *,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    profile_identity: str | None = None,
) -> list[str]:
    """Return fail-closed reasons when durable record identity does not match."""
    if record is None:
        return []
    reasons: list[str] = []
    expected_profile = current_profile_identity(profile_identity=profile_identity)
    stored_profile = (
        (record.get("session_profile_lock") or record.get("profile_identity") or "")
        .strip()
    )
    if stored_profile and expected_profile and stored_profile != expected_profile:
        if expected_profile != "unknown-profile":
            reasons.append(
                "session state profile identity mismatch "
                f"(stored={stored_profile!r}, active={expected_profile!r}; fail closed)"
            )

    for field, expected in (
        ("remote", remote),
        ("org", org),
        ("repo", repo),
    ):
        want = (expected or "").strip()
        have = (str(record.get(field) or "")).strip()
        if want and have and want != have:
            reasons.append(
                f"session state {field} mismatch "
                f"(stored={have!r}, expected={want!r}; fail closed)"
            )

    recorded_at = _parse_iso(record.get("recorded_at") or record.get("updated_at"))
    if recorded_at is None:
        reasons.append("session state missing recorded_at timestamp (fail closed)")
    else:
        age = _now_utc() - recorded_at
        kind = (record.get("kind") or "").strip()
        ttl_exempt = is_recovery_critical_record(record, kind=kind)
        if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
            reasons.append(
                f"session state expired after {ttl_hours():g}h (fail closed)"
            )
        if age < timedelta(0):
            reasons.append("session state recorded_at is in the future (fail closed)")
    return reasons


def inspect_state_envelope(
    *,
    kind: str,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    profile_identity: str | None = None,
    state_dir: str | None = None,
) -> dict[str, Any]:
    """Read-only disk inspection for assessment when TTL would otherwise hide state (#720).

    Does **not** apply identity/TTL rejection to the returned presence flags.
    Callers use this to distinguish:
      * no file on disk
      * file present but TTL would reject a non-critical kind
      * recovery-critical ledger (e.g. KIND_DECISION_LOCK) still loadable
    Never mutates files. Never returns secrets.
    """
    profile = current_profile_identity(profile_identity=profile_identity)
    path = state_file_path(
        kind=kind,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
        state_dir=state_dir,
    )
    result: dict[str, Any] = {
        "kind": kind,
        "profile_identity": profile,
        "path_basename": os.path.basename(path),
        "on_disk": False,
        "has_payload": False,
        "recorded_at": None,
        "updated_at": None,
        "age_hours": None,
        "ttl_hours": ttl_hours(),
        "age_exceeds_default_ttl": False,
        "recovery_critical": kind in RECOVERY_CRITICAL_KINDS,
        "ttl_exempt": kind in RECOVERY_CRITICAL_KINDS,
        "would_ttl_reject": False,
        "identity_reasons": [],
        "summary": "no session-state file on disk",
    }
    if not path or not os.path.exists(path):
        return result
    result["on_disk"] = True
    envelope = _read_json(path)
    if not envelope:
        result["summary"] = "session-state file present but unreadable or empty"
        return result
    payload = envelope.get("payload")
    merged: dict[str, Any] = dict(payload) if isinstance(payload, dict) else {}
    result["has_payload"] = isinstance(payload, dict)
    for key in (
        "kind",
        "remote",
        "org",
        "repo",
        "profile_identity",
        "session_profile_lock",
        "recorded_at",
        "updated_at",
        "writer_pid",
        "recovery_critical",
    ):
        if key in envelope and key not in merged:
            merged[key] = envelope[key]
    if not merged.get("kind"):
        merged["kind"] = kind
    recorded_at = _parse_iso(merged.get("recorded_at") or merged.get("updated_at"))
    result["recorded_at"] = merged.get("recorded_at") or merged.get("updated_at")
    result["updated_at"] = merged.get("updated_at") or merged.get("recorded_at")
    if recorded_at is not None:
        age = _now_utc() - recorded_at
        age_hours = age.total_seconds() / 3600.0
        result["age_hours"] = age_hours
        result["age_exceeds_default_ttl"] = age > timedelta(hours=ttl_hours())
    ttl_exempt = is_recovery_critical_record(merged, kind=kind)
    result["recovery_critical"] = ttl_exempt
    result["ttl_exempt"] = ttl_exempt
    identity_reasons = identity_match_reasons(
        merged,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
    )
    result["identity_reasons"] = list(identity_reasons)
    result["would_ttl_reject"] = any("expired" in r for r in identity_reasons)
    if result["has_payload"] and ttl_exempt:
        result["summary"] = (
            "recovery-critical session-state present on disk and TTL-exempt; "
            "load via load_state for full payload"
        )
    elif result["has_payload"] and result["would_ttl_reject"]:
        result["summary"] = (
            "session-state file present on disk but generic TTL would reject load "
            f"(age_hours={result.get('age_hours')!r}, ttl={ttl_hours():g}h)"
        )
    elif result["has_payload"]:
        result["summary"] = "session-state file present and within TTL / identity gates"
    else:
        result["summary"] = "session-state file present without a dict payload"
    return result


def load_state(
    *,
    kind: str,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    profile_identity: str | None = None,
    state_dir: str | None = None,
    instance_id: str | None = None,
) -> dict[str, Any] | None:
    """Load durable state payload when identity checks pass."""
    profile = current_profile_identity(profile_identity=profile_identity)
    path = state_file_path(
        kind=kind,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
        state_dir=state_dir,
        instance_id=instance_id,
    )
    lock_path = f"{path}.lock"
    with _exclusive_file_lock(lock_path):
        envelope = _read_json(path)
    if not envelope:
        return None
    payload = envelope.get("payload")
    if not isinstance(payload, dict):
        return None
    # Identity fields live on both envelope and payload for convenience.
    merged = dict(payload)
    for key in (
        "kind",
        "remote",
        "org",
        "repo",
        "profile_identity",
        "session_profile_lock",
        "recorded_at",
        "updated_at",
        "writer_pid",
    ):
        if key in envelope and key not in merged:
            merged[key] = envelope[key]
    reasons = identity_match_reasons(
        merged,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
    )
    if reasons:
        return None
    return merged


def save_state(
    *,
    kind: str,
    payload: dict[str, Any] | None,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    profile_identity: str | None = None,
    state_dir: str | None = None,
    instance_id: str | None = None,
) -> dict[str, Any] | None:
    """Persist or clear durable state for the given session identity key."""
    profile = current_profile_identity(
        profile_name=payload.get("session_profile") if payload else None,
        session_profile_lock=(
            (payload or {}).get("session_profile_lock") or profile_identity
        ),
    )
    # Prefer explicit args over payload fields for key location.
    key_remote = remote if remote is not None else (payload or {}).get("remote")
    key_org = org if org is not None else (payload or {}).get("org")
    key_repo = repo if repo is not None else (payload or {}).get("repo")
    key_instance = (
        (instance_id or "").strip()
        or str((payload or {}).get("instance_id") or "").strip()
        or None
    )

    root = _ensure_state_dir(state_dir)
    path = state_file_path(
        kind=kind,
        remote=key_remote,
        org=key_org,
        repo=key_repo,
        profile_identity=profile,
        state_dir=root,
        instance_id=key_instance,
    )
    lock_path = f"{path}.lock"
    with _exclusive_file_lock(lock_path):
        if payload is None:
            for candidate in (path, lock_path):
                if os.path.exists(candidate):
                    try:
                        os.remove(candidate)
                    except OSError:
                        pass
            return None

        now = _now_utc().isoformat().replace("+00:00", "Z")
        body = dict(payload)
        body.setdefault("session_pid", os.getpid())
        body["writer_pid"] = os.getpid()
        body["profile_identity"] = profile
        if not (body.get("session_profile_lock") or "").strip():
            body["session_profile_lock"] = profile
        body["recorded_at"] = body.get("recorded_at") or now
        body["updated_at"] = now
        if key_remote is not None:
            body["remote"] = key_remote
        if key_org is not None:
            body["org"] = key_org
        if key_repo is not None:
            body["repo"] = key_repo
        # Stamp session-state authority used for this write (#695 AC2 / AC6).
        body.setdefault("session_state_dir", root)
        if key_instance:
            body["instance_id"] = key_instance
        try:
            import mcp_daemon_guard

            prov = mcp_daemon_guard.mutation_provenance_fields()
            body.setdefault(
                "native_token_fingerprint", prov.get("native_token_fingerprint")
            )
            body.setdefault(
                "native_mcp_transport", bool(prov.get("native_mcp_transport"))
            )
            body.setdefault(
                "production_native_mcp_transport",
                bool(prov.get("production_native_mcp_transport")),
            )
            body.setdefault("transport", prov.get("transport"))
        except Exception:
            body.setdefault("native_mcp_transport", False)
            body.setdefault("transport", "untrusted")

        envelope = {
            "kind": kind,
            "remote": key_remote,
            "org": key_org,
            "repo": key_repo,
            "profile_identity": profile,
            "session_profile_lock": body.get("session_profile_lock"),
            "recorded_at": body["recorded_at"],
            "updated_at": body["updated_at"],
            "writer_pid": body["writer_pid"],
            "session_state_dir": body.get("session_state_dir"),
            "transport": body.get("transport"),
            "payload": body,
        }
        if key_instance:
            envelope["instance_id"] = key_instance
        _write_json(path, envelope)
        return dict(body)


def clear_state(
    *,
    kind: str,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    profile_identity: str | None = None,
    state_dir: str | None = None,
    instance_id: str | None = None,
) -> None:
    save_state(
        kind=kind,
        payload=None,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile_identity,
        state_dir=state_dir,
        instance_id=instance_id,
    )

def list_states(
    *,
    kind: str,
    profile_identity: str | None = None,
    state_dir: str | None = None,
) -> list[dict[str, Any]]:
    """Enumerate durable records of *kind* across all instance keys.

    Crash recovery cannot know which session id left a shadow behind, so it
    must be able to enumerate every record of a kind (#702 F5). Identity
    checks (profile / TTL policy) apply per record exactly as in
    :func:`load_state`; records that fail closed are omitted.
    """
    root = (state_dir or default_state_dir()).strip()
    if not root or not os.path.isdir(root):
        return []
    profile = current_profile_identity(profile_identity=profile_identity)
    results: list[dict[str, Any]] = []
    try:
        names = sorted(os.listdir(root))
    except OSError:
        return []
    for name in names:
        if not name.endswith(".json") or name.startswith("."):
            continue
        envelope = _read_json(os.path.join(root, name))
        if not envelope or envelope.get("kind") != kind:
            continue
        payload = envelope.get("payload")
        if not isinstance(payload, dict):
            continue
        merged = dict(payload)
        for key in (
            "kind",
            "remote",
            "org",
            "repo",
            "profile_identity",
            "session_profile_lock",
            "recorded_at",
            "updated_at",
            "writer_pid",
        ):
            if key in envelope and key not in merged:
                merged[key] = envelope[key]
        if identity_match_reasons(merged, profile_identity=profile):
            continue
        results.append(merged)
    return results

def list_decision_lock_profile_identities(
    state_dir: str | None = None,
) -> list[str]:
    """Return profile identities that have a durable review_decision_lock file (#709).

    Filename form: ``review_decision_lock-<profile>.json`` (see ``state_key``).
    Does not validate TTL or identity — callers must load via ``load_state``.
    """
    root = (state_dir or default_state_dir()).strip()
    if not root or not os.path.isdir(root):
        return []
    prefix = f"{_sanitize_segment(KIND_DECISION_LOCK)}-"
    suffix = ".json"
    found: list[str] = []
    try:
        names = os.listdir(root)
    except OSError:
        return []
    for name in names:
        if not name.startswith(prefix) or not name.endswith(suffix):
            continue
        if name.endswith(".lock"):
            continue
        mid = name[len(prefix) : -len(suffix)]
        if mid:
            found.append(mid)
    return sorted(set(found))


def load_state_for_profile(
    *,
    kind: str,
    profile_identity: str,
    remote: str | None = None,
    org: str | None = None,
    repo: str | None = None,
    state_dir: str | None = None,
    skip_identity_match: bool = False,
    enforce_repo_scope: bool = True,
) -> dict[str, Any] | None:
    """Load durable state for an explicit profile identity (#709 cross-profile).

    When *skip_identity_match* is True, still requires the file's recorded
    profile_identity to equal the requested profile (anti-stomp), but does not
    require the *active* session identity to match — needed so a merger can
    inspect a reviewer lock after merge.

    #709 F3: remote/org/repo filter reasons are **enforced** (not merely
    computed). When *enforce_repo_scope* is True (default) and the caller
    supplies remote/org/repo, mismatches fail closed with ``None``.
    """
    # #709 F3: refuse traversal / malformed profile identities.
    try:
        from irrecoverable_provenance import assess_profile_path_identity

        path_gate = assess_profile_path_identity(profile_identity)
        if not path_gate.get("valid"):
            return None
    except Exception:
        # Module may be mid-import in edge bootstraps; fall through to
        # conservative checks below.
        raw = str(profile_identity or "")
        if ".." in raw or "/" in raw or "\\" in raw or "\x00" in raw:
            return None

    profile = current_profile_identity(profile_identity=profile_identity)
    root = _ensure_state_dir(state_dir)
    # Refuse symlink escape of the state root (#709 F3).
    try:
        real_root = os.path.realpath(root)
        path = state_file_path(
            kind=kind,
            remote=remote,
            org=org,
            repo=repo,
            profile_identity=profile,
            state_dir=root,
        )
        real_path = os.path.realpath(path) if os.path.exists(path) else path
        if os.path.exists(path) and not str(real_path).startswith(
            str(real_root) + os.sep
        ) and str(real_path) != str(real_root):
            return None
    except OSError:
        return None

    path = state_file_path(
        kind=kind,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
        state_dir=root,
    )
    lock_path = f"{path}.lock"
    with _exclusive_file_lock(lock_path):
        envelope = _read_json(path)
    if not envelope:
        return None
    payload = envelope.get("payload")
    if not isinstance(payload, dict):
        return None
    merged = dict(payload)
    for key in (
        "kind",
        "remote",
        "org",
        "repo",
        "profile_identity",
        "session_profile_lock",
        "recorded_at",
        "updated_at",
        "writer_pid",
    ):
        if key in envelope and key not in merged:
            merged[key] = envelope[key]
    stored = (merged.get("profile_identity") or "").strip()
    if stored and stored != profile:
        return None
    if skip_identity_match:
        # Enforce remote/org/repo when caller provides them (#709 F3).
        # Drop only *active session* profile-identity mismatches; keep
        # expiry, spoof, and repository-scope reasons.
        scope_remote = remote if enforce_repo_scope else None
        scope_org = org if enforce_repo_scope else None
        scope_repo = repo if enforce_repo_scope else None
        reasons = identity_match_reasons(
            merged,
            remote=scope_remote,
            org=scope_org,
            repo=scope_repo,
            profile_identity=stored or profile,
        )
        filtered = [
            r
            for r in reasons
            if "profile identity mismatch" not in r
            or (stored and stored != profile)
        ]
        # When caller requested a specific remote/org/repo, also fail if the
        # stored record lacks those identity fields entirely (legacy incomplete).
        if enforce_repo_scope:
            for field, want in (
                ("remote", remote),
                ("org", org),
                ("repo", repo),
            ):
                want_s = (want or "").strip()
                if not want_s:
                    continue
                have = (str(merged.get(field) or "")).strip()
                if not have:
                    filtered.append(
                        f"session state {field} missing on durable record "
                        f"(expected={want_s!r}; fail closed, #709 F3)"
                    )
                elif have != want_s:
                    # identity_match_reasons already adds mismatch; ensure kept
                    pass
        if filtered:
            return None
        return merged
    reasons = identity_match_reasons(
        merged,
        remote=remote,
        org=org,
        repo=repo,
        profile_identity=profile,
    )
    if reasons:
        return None
    return merged

