sysadmin
3d11e1f12b
Namespace-scoped workspace resolution prevents GITEA_AUTHOR_WORKTREE from poisoning reviewer, merger, and reconciler purity checks. Each role uses its own worktree env var with actionable binding-source errors and safe reconnect guidance. Preserves #274/#475 author and control-checkout guards.
307 lines
11 KiB
Python
307 lines
11 KiB
Python
"""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,
|
|
) -> tuple[str, str]:
|
|
"""Return ``(resolved_path, binding_source)`` for *role_kind*."""
|
|
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 in (
|
|
(worktree_path, "worktree_path argument"),
|
|
(worktree, "worktree argument"),
|
|
(_env_value(env_map, ACTIVE_WORKTREE_ENV), f"{ACTIVE_WORKTREE_ENV} environment variable"),
|
|
(_env_value(env_map, role_env_key), f"{role_env_key} environment variable"),
|
|
(session_lease_worktree if role in {"reviewer", "merger"} else None,
|
|
"reviewer PR lease worktree"),
|
|
):
|
|
text = (candidate or "").strip()
|
|
if text:
|
|
return os.path.realpath(os.path.abspath(text)), 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,
|
|
) -> dict:
|
|
"""Shared workspace resolution for runtime_context and mutation guards."""
|
|
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,
|
|
)
|
|
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,
|
|
)
|
|
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": 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,
|
|
) -> 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,
|
|
)
|
|
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 [],
|
|
} |