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,
    session_lock_worktree: str | None = None,
    profile_name: str | None = None,
    demotions: list[str] | None = None,
    verify_paths: bool = False,
    durable_author_result: dict | None = None,
) -> 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) for non-author roles: 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.

    Author role (#618): never demotes a missing configured binding to the
    control checkout. When *verify_paths* is true, resolution goes through
    :func:`author_mutation_worktree.resolve_durable_author_worktree` so
    mutations either use an explicit validated worktree, derive from the
    active author issue lock, or fail closed. 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]

    # #618: durable author resolution — no silent control/master fallback.
    if role == "author" and verify_paths:
        durable = durable_author_result
        if durable is None:
            durable = amw.resolve_durable_author_worktree(
                worktree_path=worktree_path,
                worktree=worktree,
                process_project_root=process_project_root,
                active_worktree_env=_env_value(env_map, ACTIVE_WORKTREE_ENV),
                author_worktree_env=_env_value(env_map, AUTHOR_WORKTREE_ENV),
                session_lock_worktree=session_lock_worktree,
                profile_name=profile_name,
                # Path selection only here; full validation is re-run in
                # resolve_namespace_mutation_context with the canonical root.
                validate=False,
            )
        workspace = durable.get("workspace_path") or os.path.realpath(
            process_project_root
        )
        source = durable.get("workspace_binding_source") or "no author worktree binding"
        if demotions is not None and durable.get("bound_worktree_missing"):
            demotions.append(
                f"{source} '{workspace}' not demoted: {amw.BOUND_WORKTREE_MISSING_MESSAGE}"
            )
        return workspace, source

    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),
        # Author lock derivation is handled by the durable path above when
        # verify_paths is true; when verify_paths is false, surface the lock
        # path as a non-demoted candidate so tooling can inspect it.
        (session_lock_worktree if role == "author" else None,
         "active author issue lock 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,
    session_lock_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.

    Author role (#618): uses durable worktree resolution (explicit path, env,
    or active issue lock) and never silently falls back to the control checkout.
    """
    demotions: list[str] = []
    env_map = env if env is not None else os.environ
    process_root = os.path.realpath(process_project_root)
    role = normalize_role_kind(role_kind, 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)

    durable: dict | None = None
    if role == "author":
        durable = amw.resolve_durable_author_worktree(
            worktree_path=worktree_path,
            worktree=worktree,
            process_project_root=process_root,
            active_worktree_env=_env_value(env_map, ACTIVE_WORKTREE_ENV),
            author_worktree_env=_env_value(env_map, AUTHOR_WORKTREE_ENV),
            session_lock_worktree=session_lock_worktree,
            canonical_repo_root=canonical_root,
            profile_name=profile_name,
            validate=True,
        )
        workspace = durable["workspace_path"]
        binding_source = durable["workspace_binding_source"]
        if durable.get("bound_worktree_missing"):
            demotions.append(
                f"{binding_source} '{workspace}' not demoted: "
                f"{amw.BOUND_WORKTREE_MISSING_MESSAGE}"
            )
    else:
        workspace, binding_source = resolve_namespace_workspace(
            role_kind=role,
            worktree_path=worktree_path,
            worktree=worktree,
            process_project_root=process_project_root,
            env=env,
            session_lease_worktree=session_lease_worktree,
            session_lock_worktree=session_lock_worktree,
            profile_name=profile_name,
            demotions=demotions,
            verify_paths=True,
        )

    pollution = assess_foreign_role_worktree_pollution(
        role_kind=role,
        resolved_workspace=workspace,
        binding_source=binding_source,
        env=env,
        profile_name=profile_name,
    )
    result = {
        "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,
    }
    if durable is not None:
        result["author_worktree_resolution"] = durable
        result["bound_worktree_missing"] = bool(durable.get("bound_worktree_missing"))
        result["path_exists"] = durable.get("path_exists")
        result["in_git_worktree_list"] = durable.get("in_git_worktree_list")
        result["inspected_git_root"] = durable.get("inspected_git_root")
        result["author_worktree_block"] = bool(durable.get("block"))
        result["author_worktree_reasons"] = list(durable.get("reasons") or [])
        result["author_worktree_blocker_kind"] = durable.get("blocker_kind")
        result["operator_recovery"] = durable.get("operator_recovery")
    return result


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,
    operator_recovery: str | None = None,
) -> str:
    """Canonical error when namespace workspace binding blocks mutations."""
    role = normalize_role_kind(role_kind)
    reason_list = list(reasons or [])
    # #618: prefer the durable author missing-worktree message when present.
    if role == "author" and any(
        amw.BOUND_WORKTREE_MISSING_MESSAGE in r for r in reason_list
    ):
        return amw.format_bound_worktree_missing_error(
            {
                "reasons": reason_list,
                "binding_source": binding_source,
                "configured_path": workspace_path,
                "role_kind": role,
                "operator_recovery": operator_recovery
                or amw.OPERATOR_RECOVERY_RECREATE_REPOINT,
            }
        )
    try:
        workspace = os.path.realpath(workspace_path)
    except OSError:
        workspace = 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 reason_list:
        parts.append("Details: " + "; ".join(reason_list) + ".")
    if operator_recovery:
        parts.append(f"Operator recovery: {operator_recovery}")
    else:
        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,
    session_lock_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,
        session_lock_worktree=session_lock_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 [])
    operator_recovery = ctx.get("operator_recovery")
    if role == "author":
        # #618 durable resolution already validated existence, membership,
        # branches/, lock ownership, and traversal safety when present.
        durable_reasons = list(ctx.get("author_worktree_reasons") or [])
        if durable_reasons:
            reasons.extend(durable_reasons)
        elif ctx.get("author_worktree_block"):
            reasons.append(amw.BOUND_WORKTREE_MISSING_MESSAGE)
        else:
            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 [],
        "bound_worktree_missing": bool(ctx.get("bound_worktree_missing")),
        "path_exists": ctx.get("path_exists"),
        "in_git_worktree_list": ctx.get("in_git_worktree_list"),
        "inspected_git_root": ctx.get("inspected_git_root"),
        "operator_recovery": operator_recovery,
        "blocker_kind": ctx.get("author_worktree_blocker_kind"),
    }