Gitea-Tools/stale_binding_recovery.py

"""Sanctioned recovery for stale env-bound task workspace bindings (#702).

When the MCP daemon dies from an unhandled exception it never runs teardown,
and the auto-reconnected daemon inherits ``GITEA_ACTIVE_WORKTREE`` from its
parent environment. That inherited value can permanently bind the fresh
session to a prior task's worktree (the PR #701 / ``review-pr-654`` incident).

This module classifies the active env binding against live session evidence
and produces a fail-closed recovery plan. Only two situations permit the
daemon to clear the binding on its own:

- the bound path no longer exists on disk (``provably_stale_missing_path``)
- the binding was inherited at daemon boot and a *sanctioned* in-session
  reviewer/merger lease later bound a different worktree
  (``superseded_by_session_lease``)

Everything else is surfaced (``unverified_inherited``) and left for the
managed reconnect path — never cleared silently, never rebound to a guessed
worktree.
"""

from __future__ import annotations

import os
from typing import Any

ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"

CLASSIFICATION_UNBOUND = "unbound"
CLASSIFICATION_CORROBORATED = "corroborated"
CLASSIFICATION_UNVERIFIED_INHERITED = "unverified_inherited"
CLASSIFICATION_PROVABLY_STALE_MISSING_PATH = "provably_stale_missing_path"
CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE = "superseded_by_session_lease"

CLASSIFICATIONS = frozenset({
    CLASSIFICATION_UNBOUND,
    CLASSIFICATION_CORROBORATED,
    CLASSIFICATION_UNVERIFIED_INHERITED,
    CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
    CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
})

# Roles whose task workspace is owned by a lease lifecycle, so an inherited
# generic binding without a corroborating lease is suspect.
LEASE_BOUND_ROLES = frozenset({"reviewer", "merger"})

RECOVERY_ACTION_NONE = "none"
RECOVERY_ACTION_CLEAR_ENV = "clear_env"


def _norm(path: str | None) -> str:
    text = (path or "").strip()
    if not text:
        return ""
    return os.path.realpath(os.path.abspath(text))


def snapshot_boot_bindings(
    env: dict[str, str] | os._Environ | None = None,
) -> dict[str, Any]:
    """Capture worktree env bindings present when the daemon booted.

    A value present in this snapshot was inherited from the parent
    environment, not established by any sanctioned tool in this daemon's
    lifetime.
    """
    env_map = env if env is not None else os.environ
    return {
        "active_worktree": (env_map.get(ACTIVE_WORKTREE_ENV) or "").strip() or None,
    }


def classify_active_worktree_binding(
    *,
    active_value: str | None,
    role_env_value: str | None = None,
    session_lease_worktree: str | None = None,
    boot_inherited: bool,
    path_exists: bool | None,
    role_kind: str | None = None,
) -> dict[str, Any]:
    """Classify the GITEA_ACTIVE_WORKTREE binding against live evidence.

    Fail closed: unknown evidence never yields a clear-eligible class.
    """
    reasons: list[str] = []
    active = _norm(active_value)
    if not active:
        return {
            "classification": CLASSIFICATION_UNBOUND,
            "clear_eligible": False,
            "active_worktree": None,
            "reasons": ["no GITEA_ACTIVE_WORKTREE binding present"],
        }

    role = (role_kind or "").strip().lower()
    role_env = _norm(role_env_value)
    lease_wt = _norm(session_lease_worktree)

    result: dict[str, Any] = {
        "active_worktree": active,
        "boot_inherited": bool(boot_inherited),
        "role_kind": role or None,
        "session_lease_worktree": lease_wt or None,
        "role_env_worktree": role_env or None,
    }

    if path_exists is False:
        reasons.append(
            f"bound worktree '{active}' no longer exists on disk; the binding "
            "cannot name a valid task workspace"
        )
        result.update({
            "classification": CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
            "clear_eligible": True,
            "reasons": reasons,
        })
        return result

    if lease_wt and lease_wt == active:
        reasons.append("binding matches the sanctioned in-session lease worktree")
        result.update({
            "classification": CLASSIFICATION_CORROBORATED,
            "clear_eligible": False,
            "reasons": reasons,
        })
        return result

    if lease_wt and lease_wt != active and boot_inherited:
        reasons.append(
            "boot-inherited binding disagrees with the sanctioned in-session "
            f"lease worktree '{lease_wt}'; the lease is live authority"
        )
        result.update({
            "classification": CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
            "clear_eligible": True,
            "reasons": reasons,
        })
        return result

    if lease_wt and lease_wt != active and not boot_inherited:
        # Set after boot by tooling; disagreement is surfaced, never auto-cleared.
        reasons.append(
            "post-boot binding disagrees with the in-session lease worktree; "
            "surfacing only (fail closed, no automatic clear)"
        )
        result.update({
            "classification": CLASSIFICATION_UNVERIFIED_INHERITED,
            "clear_eligible": False,
            "reasons": reasons,
        })
        return result

    if role_env and role_env == active:
        reasons.append("binding matches the role-specific worktree env binding")
        result.update({
            "classification": CLASSIFICATION_CORROBORATED,
            "clear_eligible": False,
            "reasons": reasons,
        })
        return result

    if boot_inherited and role in LEASE_BOUND_ROLES and not lease_wt:
        reasons.append(
            "binding was inherited at daemon boot, no sanctioned in-session "
            f"lease corroborates it, and the {role} role binds workspaces "
            "through the lease lifecycle; treat as unverified until a fresh "
            "lease or managed reconnect re-establishes the workspace"
        )
        result.update({
            "classification": CLASSIFICATION_UNVERIFIED_INHERITED,
            "clear_eligible": False,
            "reasons": reasons,
        })
        return result

    reasons.append("no contradicting evidence; binding treated as corroborated")
    result.update({
        "classification": CLASSIFICATION_CORROBORATED,
        "clear_eligible": False,
        "reasons": reasons,
    })
    return result


def plan_recovery(classification_result: dict[str, Any]) -> dict[str, Any]:
    """Turn a classification into an explicit, fail-closed recovery plan."""
    classification = classification_result.get("classification")
    clear_eligible = bool(classification_result.get("clear_eligible"))
    reasons = list(classification_result.get("reasons") or [])

    if classification not in CLASSIFICATIONS:
        return {
            "action": RECOVERY_ACTION_NONE,
            "clear_allowed": False,
            "classification": classification,
            "reasons": reasons + ["unknown classification (fail closed)"],
        }

    if clear_eligible and classification in {
        CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
        CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
    }:
        return {
            "action": RECOVERY_ACTION_CLEAR_ENV,
            "clear_allowed": True,
            "classification": classification,
            "active_worktree": classification_result.get("active_worktree"),
            "reasons": reasons,
        }

    exact_next_action = None
    if classification == CLASSIFICATION_UNVERIFIED_INHERITED:
        exact_next_action = (
            "managed recovery required: reconnect the MCP namespace through "
            "the managed client configuration (or start a fresh session) so "
            "the daemon boots without the inherited GITEA_ACTIVE_WORKTREE, or "
            "corroborate the binding by acquiring a lease for that worktree; "
            "do not clear or rewrite the environment manually"
        )
    return {
        "action": RECOVERY_ACTION_NONE,
        "clear_allowed": False,
        "classification": classification,
        "active_worktree": classification_result.get("active_worktree"),
        "reasons": reasons,
        **({"exact_next_action": exact_next_action} if exact_next_action else {}),
    }


def apply_recovery(
    plan: dict[str, Any],
    env: dict[str, str] | os._Environ | None = None,
) -> dict[str, Any]:
    """Apply a clear-eligible plan by removing the stale env binding.

    This is the sanctioned in-daemon mechanism (#702 AC2); callers must
    persist the returned record for audit. Refuses anything but an explicit
    ``clear_env`` plan.
    """
    env_map = env if env is not None else os.environ
    if not plan.get("clear_allowed") or plan.get("action") != RECOVERY_ACTION_CLEAR_ENV:
        return {
            "performed": False,
            "action": RECOVERY_ACTION_NONE,
            "reasons": ["recovery plan does not authorize clearing (fail closed)"],
        }
    cleared_value = (env_map.get(ACTIVE_WORKTREE_ENV) or "").strip() or None
    env_map.pop(ACTIVE_WORKTREE_ENV, None)
    return {
        "performed": True,
        "action": RECOVERY_ACTION_CLEAR_ENV,
        "cleared_env": ACTIVE_WORKTREE_ENV,
        "cleared_value": cleared_value,
        "classification": plan.get("classification"),
        "reasons": list(plan.get("reasons") or []),
    }