Gitea-Tools/namespace_workspace_binding.py

"""Namespace-scoped MCP workspace binding (#510).

Each role namespace (author, reviewer, merger, reconciler) resolves its own
active task workspace. Foreign role worktree environment variables must not
poison workspace purity checks in another namespace.
"""

from __future__ import annotations

import os

import author_mutation_worktree as amw

ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV
REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"

ROLE_WORKTREE_ENVS: dict[str, str] = {
    "author": AUTHOR_WORKTREE_ENV,
    "reviewer": REVIEWER_WORKTREE_ENV,
    "merger": MERGER_WORKTREE_ENV,
    "reconciler": RECONCILER_WORKTREE_ENV,
}

NON_AUTHOR_ROLES = frozenset({"reviewer", "merger", "reconciler"})


def normalize_role_kind(
    role_kind: str | None,
    *,
    profile_name: str | None = None,
) -> str:
    """Map profile/task role to a workspace namespace key."""
    role = (role_kind or "author").strip().lower()
    profile = (profile_name or "").strip().lower()
    if role == "reviewer" and "merger" in profile:
        return "merger"
    if role in ROLE_WORKTREE_ENVS:
        return role
    return "author"


def _env_value(env: dict[str, str] | os._Environ, key: str) -> str | None:
    text = (env.get(key) or "").strip()
    return text or None


def resolve_namespace_workspace(
    *,
    role_kind: str,
    worktree_path: str | None = None,
    worktree: str | None = None,
    process_project_root: str,
    env: dict[str, str] | os._Environ | None = None,
    session_lease_worktree: str | None = None,
    profile_name: str | None = None,
    demotions: list[str] | None = None,
    verify_paths: bool = False,
) -> tuple[str, str]:
    """Return ``(resolved_path, binding_source)`` for *role_kind*.

    With *verify_paths*, env-sourced candidates whose path no longer exists
    are demoted (#702): a binding to a deleted worktree can never name a
    valid task workspace, so resolution falls through to the next candidate.
    Explicit arguments are never demoted — a caller-declared path must fail
    loudly downstream rather than silently rebind. Demotion notes are
    appended to *demotions* when provided. Runtime-context and mutation
    guards resolve through :func:`resolve_namespace_mutation_context`, which
    always verifies.
    """
    env_map = env if env is not None else os.environ
    role = normalize_role_kind(role_kind, profile_name=profile_name)
    role_env_key = ROLE_WORKTREE_ENVS[role]

    for candidate, source, env_sourced in (
        (worktree_path, "worktree_path argument", False),
        (worktree, "worktree argument", False),
        (_env_value(env_map, ACTIVE_WORKTREE_ENV),
         f"{ACTIVE_WORKTREE_ENV} environment variable", True),
        (_env_value(env_map, role_env_key),
         f"{role_env_key} environment variable", True),
        (session_lease_worktree if role in {"reviewer", "merger"} else None,
         "reviewer PR lease worktree", False),
    ):
        text = (candidate or "").strip()
        if not text:
            continue
        real = os.path.realpath(os.path.abspath(text))
        if verify_paths and env_sourced and not os.path.isdir(real):
            if demotions is not None:
                demotions.append(
                    f"{source} '{real}' demoted: path no longer exists "
                    "(stale binding, #702)"
                )
            continue
        return real, source

    return os.path.realpath(process_project_root), "MCP server process root (default)"


def resolve_namespace_mutation_context(
    *,
    role_kind: str,
    worktree_path: str | None,
    process_project_root: str,
    env: dict[str, str] | os._Environ | None = None,
    session_lease_worktree: str | None = None,
    worktree: str | None = None,
    profile_name: str | None = None,
    configured_canonical_root: str | None = None,
) -> dict:
    """Shared workspace resolution for runtime_context and mutation guards.

    When *configured_canonical_root* is supplied (a cross-repository namespace
    bound to an external target repository, #706), the canonical repository root
    is that configured target rather than the MCP install checkout. This keeps
    the branches-only / worktree-membership guards (#274) evaluating against the
    repository the namespace actually mutates. Without it the single-repo
    default is preserved: the canonical root follows the process checkout.
    """
    demotions: list[str] = []
    workspace, binding_source = resolve_namespace_workspace(
        role_kind=role_kind,
        worktree_path=worktree_path,
        worktree=worktree,
        process_project_root=process_project_root,
        env=env,
        session_lease_worktree=session_lease_worktree,
        profile_name=profile_name,
        demotions=demotions,
        verify_paths=True,
    )
    process_root = os.path.realpath(process_project_root)
    role = normalize_role_kind(role_kind, profile_name=profile_name)
    pollution = assess_foreign_role_worktree_pollution(
        role_kind=role,
        resolved_workspace=workspace,
        binding_source=binding_source,
        env=env,
        profile_name=profile_name,
    )
    configured = (configured_canonical_root or "").strip()
    if configured:
        canonical_root = os.path.realpath(configured)
    else:
        canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
    return {
        "workspace_path": workspace,
        "workspace_binding_source": binding_source,
        "workspace_role_kind": role,
        "ignored_bindings": demotions + (pollution.get("ignored_bindings") or []),
        "process_project_root": process_root,
        "canonical_repo_root": canonical_root,
        "roots_aligned": canonical_root == process_root,
    }


def assess_foreign_role_worktree_pollution(
    *,
    role_kind: str,
    resolved_workspace: str,
    binding_source: str,
    env: dict[str, str] | os._Environ | None = None,
    profile_name: str | None = None,
) -> dict:
    """Detect when a foreign role env would have hijacked workspace binding."""
    env_map = env if env is not None else os.environ
    role = normalize_role_kind(role_kind, profile_name=profile_name)
    if role == "author":
        return {"would_pollute": False, "ignored_bindings": []}

    ignored: list[str] = []
    author_path = _env_value(env_map, AUTHOR_WORKTREE_ENV)
    if author_path:
        author_real = os.path.realpath(os.path.abspath(author_path))
        resolved_real = os.path.realpath(resolved_workspace)
        if author_real != resolved_real and binding_source != f"{AUTHOR_WORKTREE_ENV} environment variable":
            ignored.append(
                f"{AUTHOR_WORKTREE_ENV}={author_real} (ignored for {role} namespace)"
            )
    return {
        "would_pollute": bool(ignored),
        "ignored_bindings": ignored,
    }


def assess_metadata_only_worktree_binding(
    *,
    role_kind: str,
    declared_worktree_path: str | None,
    mutation_workspace: str,
    process_project_root: str,
    profile_name: str | None = None,
) -> dict:
    """Fail closed when declared worktree_path would not redirect mutations."""
    declared = (declared_worktree_path or "").strip()
    process_root = os.path.realpath(process_project_root)
    mutation_root = os.path.realpath(mutation_workspace)
    role = normalize_role_kind(role_kind, profile_name=profile_name)
    if not declared:
        return {"block": False, "reasons": [], "metadata_only": False}

    declared_root = os.path.realpath(os.path.abspath(declared))
    if declared_root == mutation_root:
        return {"block": False, "reasons": [], "metadata_only": False}

    if declared_root != process_root and mutation_root == process_root:
        return {
            "block": True,
            "metadata_only": True,
            "reasons": [
                f"worktree_path is metadata-only for {role} mutations: preflight "
                f"inspected '{declared_root}' but mutation tools would still "
                f"validate MCP server process root '{process_root}'"
            ],
            "declared_worktree_path": declared_root,
            "mutation_workspace": mutation_root,
            "process_project_root": process_root,
        }

    return {"block": False, "reasons": [], "metadata_only": False}


def format_namespace_workspace_binding_error(
    *,
    role_kind: str,
    workspace_path: str,
    binding_source: str,
    reasons: list[str] | None = None,
    ignored_bindings: list[str] | None = None,
    dirty_files: list[str] | None = None,
) -> str:
    """Canonical error when namespace workspace binding blocks mutations."""
    role = normalize_role_kind(role_kind)
    workspace = os.path.realpath(workspace_path)
    parts = [
        f"Namespace workspace binding blocked ({role} namespace, #510): "
        f"resolved workspace '{workspace}' via {binding_source}."
    ]
    if ignored_bindings:
        parts.append(
            "Foreign role bindings ignored: " + "; ".join(ignored_bindings) + "."
        )
    if dirty_files:
        parts.append(
            "Dirty tracked files in active task workspace: "
            + ", ".join(dirty_files)
            + "."
        )
    if reasons:
        parts.append("Details: " + "; ".join(reasons) + ".")
    parts.append(
        "Remediation: reconnect or relaunch the MCP server from a clean dedicated "
        f"branches/ {role} worktree, set "
        f"{ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} or {ACTIVE_WORKTREE_ENV} "
        "to that path, or pass worktree_path on mutation tools. Do not clean or "
        "reset foreign role worktrees to unblock this namespace."
    )
    return " ".join(parts)


def assess_namespace_mutation_workspace(
    *,
    role_kind: str,
    worktree_path: str | None,
    worktree: str | None,
    process_project_root: str,
    env: dict[str, str] | os._Environ | None = None,
    session_lease_worktree: str | None = None,
    profile_name: str | None = None,
    current_branch: str | None = None,
    configured_canonical_root: str | None = None,
) -> dict:
    """Evaluate namespace workspace binding before preflight/mutation."""
    ctx = resolve_namespace_mutation_context(
        role_kind=role_kind,
        worktree_path=worktree_path,
        worktree=worktree,
        process_project_root=process_project_root,
        env=env,
        session_lease_worktree=session_lease_worktree,
        profile_name=profile_name,
        configured_canonical_root=configured_canonical_root,
    )
    mutation_workspace = ctx["workspace_path"]
    binding_source = ctx["workspace_binding_source"]
    role = ctx["workspace_role_kind"]
    process_root = ctx["process_project_root"]

    metadata = assess_metadata_only_worktree_binding(
        role_kind=role,
        declared_worktree_path=worktree_path,
        mutation_workspace=mutation_workspace,
        process_project_root=process_root,
        profile_name=profile_name,
    )
    pollution = assess_foreign_role_worktree_pollution(
        role_kind=role,
        resolved_workspace=mutation_workspace,
        binding_source=binding_source,
        env=env,
        profile_name=profile_name,
    )

    reasons = list(metadata.get("reasons") or [])
    if role == "author":
        branches = amw.assess_author_mutation_worktree(
            workspace_path=mutation_workspace,
            project_root=ctx["canonical_repo_root"],
            current_branch=current_branch,
        )
        if branches["block"]:
            reasons.extend(branches["reasons"])
    elif (
        role == "reviewer"
        and mutation_workspace == process_root
        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
    ):
        reasons.append(
            f"{role} mutation blocked: workspace is the stable control checkout; "
            f"create or reconnect to a session-owned worktree under branches/ "
            f"or set {ROLE_WORKTREE_ENVS[role]} / {ACTIVE_WORKTREE_ENV}"
        )
    elif (
        role in {"reviewer", "merger"}
        and mutation_workspace != process_root
        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
    ):
        reasons.append(
            f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
            f"'{ctx['canonical_repo_root']}/branches/'"
        )

    block = bool(reasons)
    return {
        "block": block,
        "reasons": reasons,
        "mutation_workspace": mutation_workspace,
        "workspace_binding_source": binding_source,
        "workspace_role_kind": role,
        "process_project_root": process_root,
        "canonical_repo_root": ctx["canonical_repo_root"],
        "metadata_only": metadata.get("metadata_only", False),
        "declared_worktree_path": metadata.get("declared_worktree_path"),
        "ignored_bindings": pollution.get("ignored_bindings") or [],
    }