From f1ad66524261ecf75284c579a6380065e4148876 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 03:57:38 -0400 Subject: [PATCH] feat: isolate MCP workspace binding across role namespaces (Closes #510) 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. --- .env.example | 9 + author_mutation_worktree.py | 2 + docs/llm-workflow-runbooks.md | 39 +++ gitea_mcp_server.py | 212 ++++++++++++--- namespace_workspace_binding.py | 307 ++++++++++++++++++++++ tests/test_mcp_server.py | 11 +- tests/test_namespace_workspace_binding.py | 240 +++++++++++++++++ 7 files changed, 772 insertions(+), 48 deletions(-) create mode 100644 namespace_workspace_binding.py create mode 100644 tests/test_namespace_workspace_binding.py diff --git a/.env.example b/.env.example index 5463787..4777fac 100644 --- a/.env.example +++ b/.env.example @@ -46,3 +46,12 @@ GITEA_TOKEN_SOURCE=GITEA_TOKEN # profile's values. Leave unset for pure env-based configuration. GITEA_MCP_CONFIG=/Users/jasonwalker/.config/gitea-tools/profiles.json GITEA_MCP_PROFILE=prgs + +# Namespace-scoped active task workspaces (#510). Each MCP namespace uses only +# its own role env var; foreign bindings (e.g. GITEA_AUTHOR_WORKTREE in a +# merger process) are ignored. +# GITEA_AUTHOR_WORKTREE=/path/to/repo/branches/issue-123-work +# GITEA_REVIEWER_WORKTREE=/path/to/repo/branches/review-pr456 +# GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456 +# GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456 +# GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override diff --git a/author_mutation_worktree.py b/author_mutation_worktree.py index 29dc05d..dccb94a 100644 --- a/author_mutation_worktree.py +++ b/author_mutation_worktree.py @@ -12,6 +12,8 @@ import subprocess BASE_BRANCHES = frozenset({"master", "main", "dev"}) ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE" AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" +# Author-only: reviewer/merger/reconciler namespaces use role-specific env vars +# via namespace_workspace_binding (#510). def _normalize_path(path: str) -> str: diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index d67769d..cb904c4 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -823,6 +823,45 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push ``` +## Namespace workspace binding (#510) + +Each MCP namespace resolves its **own** active task workspace. Foreign role +worktree environment variables must not poison another namespace's purity +checks. + +| Namespace | Workspace env vars (in priority under `GITEA_ACTIVE_WORKTREE`) | Allowed roots | +|-----------|------------------------------------------------------------------|---------------| +| author | `GITEA_AUTHOR_WORKTREE` | `branches/` worktree only (#274) | +| reviewer | `GITEA_REVIEWER_WORKTREE` | clean `branches/` worktree | +| merger | `GITEA_MERGER_WORKTREE` | clean `branches/` worktree **or** clean control checkout | +| reconciler | `GITEA_RECONCILER_WORKTREE` | clean `branches/` worktree **or** clean control checkout | + +`GITEA_AUTHOR_WORKTREE` is **author-only**. Reviewer, merger, and reconciler +MCP processes ignore it even when it points at a dirty author WIP tree. + +### Safe reconnect / rebind procedure + +When a mutation blocks on workspace binding: + +1. Read the error — it names the **resolved workspace path**, **role + namespace**, and **binding source** (tool arg, env var, or process root). +2. Reconnect or relaunch the correct namespace MCP server from the intended + workspace (or set the role-specific env var before launch). +3. Pass `worktree_path` on reviewer/merger mutation tools when the active + branches/ worktree differs from the MCP process root. +4. **Do not** clean, reset, or discard foreign role worktrees to unblock your + own namespace — that destroys another agent's WIP. + +### CTH guidance for workspace binding blockers + +When posting a Canonical Thread Handoff after a binding blocker: + +- State which namespace was active (author / reviewer / merger / reconciler). +- Quote the resolved workspace path and binding source from the error. +- Name the safe reconnect action (relaunch MCP from `branches/...`, set + `GITEA_*_WORKTREE`, or pass `worktree_path`). +- Explicitly note that foreign worktrees must not be cleaned to unblock. + ## Safety notes - Never place raw tokens or passwords in any LLM MCP config; reference secrets diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index cc1cbcf..7b62c18 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -174,12 +174,76 @@ _preflight_reviewer_violation_files: list[str] = [] ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE" AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" +REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE" +MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE" +RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE" + +import namespace_workspace_binding as nwb # noqa: E402 def _preflight_in_test_mode() -> bool: return "pytest" in sys.modules or "unittest" in sys.modules +def _reviewer_session_worktree() -> str | None: + import reviewer_pr_lease as _reviewer_pr_lease + + session = _reviewer_pr_lease.get_session_lease() + if not session: + return None + worktree = (session.get("worktree") or "").strip() + return worktree or None + + +def _effective_workspace_role() -> str: + """Resolve the namespace key used for workspace binding (#510).""" + profile = get_profile() + role = _preflight_resolved_role + if not role: + role = _role_kind( + profile.get("allowed_operations") or [], + profile.get("forbidden_operations") or [], + ) + return nwb.normalize_role_kind( + role, + profile_name=profile.get("profile_name"), + ) + + +def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: + """Resolve the namespace-scoped workspace root inspected by pre-flight guards.""" + role = _effective_workspace_role() + workspace, _source = nwb.resolve_namespace_workspace( + role_kind=role, + worktree_path=worktree_path, + process_project_root=PROJECT_ROOT, + session_lease_worktree=( + _reviewer_session_worktree() if role in {"reviewer", "merger"} else None + ), + profile_name=get_profile().get("profile_name"), + ) + return workspace + + +def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict: + """Canonical namespace workspace + repository root for guards (#460/#510).""" + role = _effective_workspace_role() + return nwb.resolve_namespace_mutation_context( + role_kind=role, + worktree_path=worktree_path, + process_project_root=PROJECT_ROOT, + session_lease_worktree=( + _reviewer_session_worktree() if role in {"reviewer", "merger"} else None + ), + profile_name=get_profile().get("profile_name"), + ) + + +def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict: + """Backward-compatible alias for namespace workspace context.""" + return _resolve_namespace_mutation_context(worktree_path) + + def _ensure_process_start_porcelain() -> str: """Capture the shared-worktree baseline once per MCP process (#252).""" global _process_start_porcelain @@ -188,26 +252,6 @@ def _ensure_process_start_porcelain() -> str: return _process_start_porcelain -def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: - """Resolve the workspace root inspected by pre-flight guards.""" - return author_mutation_worktree.resolve_mutation_workspace( - worktree_path, - PROJECT_ROOT, - active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), - author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), - ) - - -def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict: - """Canonical workspace + repository root for runtime_context and guards (#460).""" - return author_mutation_worktree.resolve_author_mutation_context( - worktree_path, - PROJECT_ROOT, - active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV), - author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV), - ) - - def _get_git_root(path: str) -> str | None: try: res = subprocess.run( @@ -275,7 +319,7 @@ def _format_preflight_files(files: list[str]) -> str: def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict: - ctx = _resolve_author_mutation_context(worktree_path) + ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] inspected_root = _get_git_root(workspace) process_root = ctx["process_project_root"] @@ -293,6 +337,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st "dirty_files": list(dirty_files), "dirty_scope": dirty_scope, "workspace_roots_aligned": ctx["roots_aligned"], + "workspace_role_kind": ctx.get("workspace_role_kind"), + "workspace_binding_source": ctx.get("workspace_binding_source"), + "ignored_bindings": list(ctx.get("ignored_bindings") or []), } if not ctx["roots_aligned"]: details["workspace_root_mismatch"] = ( @@ -303,13 +350,19 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st def _format_preflight_workspace_details(details: dict) -> str: - return ( - f"MCP server process root: {details.get('mcp_server_process_root')}; " - f"active task workspace root: {details.get('active_task_workspace_root')}; " - f"inspected git root: {details.get('inspected_git_root')}; " - f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; " - f"dirty scope: {details.get('dirty_scope')}" - ) + parts = [ + f"MCP server process root: {details.get('mcp_server_process_root')}", + f"active task workspace root: {details.get('active_task_workspace_root')}", + f"workspace role: {details.get('workspace_role_kind')}", + f"binding source: {details.get('workspace_binding_source')}", + f"inspected git root: {details.get('inspected_git_root')}", + f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}", + f"dirty scope: {details.get('dirty_scope')}", + ] + ignored = details.get("ignored_bindings") or [] + if ignored: + parts.append(f"ignored foreign bindings: {'; '.join(ignored)}") + return "; ".join(parts) def assess_preflight_status(worktree_path: str | None = None) -> dict: @@ -332,6 +385,25 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict: "Active task workspace has tracked file edits before mutation " f"({_format_preflight_workspace_details(workspace_details)})" ) + role = _effective_workspace_role() + if role in nwb.NON_AUTHOR_ROLES: + binding = nwb.assess_metadata_only_worktree_binding( + role_kind=role, + declared_worktree_path=worktree_path, + mutation_workspace=_resolve_preflight_workspace_path(worktree_path), + process_project_root=PROJECT_ROOT, + profile_name=get_profile().get("profile_name"), + ) + if binding.get("block"): + reasons.append(binding["reasons"][0]) + reasons.append( + nwb.format_namespace_workspace_binding_error( + role_kind=role, + workspace_path=binding["mutation_workspace"], + binding_source="MCP server process root (default)", + reasons=binding.get("reasons"), + ) + ) return { "preflight_ready": not reasons, "preflight_block_reasons": reasons, @@ -419,13 +491,14 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None: """#274: author file/branch mutations must run from a branches/ worktree. - Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a - Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE`` - (#468). + Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr`` + is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE`` + (#468). Non-author namespaces use dedicated workspace env vars (#510). """ - if _preflight_resolved_role in ("reviewer", "reconciler"): + role = _effective_workspace_role() + if role in nwb.NON_AUTHOR_ROLES: return - ctx = _resolve_author_mutation_context(worktree_path) + ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] git_state = issue_lock_worktree.read_worktree_git_state(workspace) assessment = author_mutation_worktree.assess_author_mutation_worktree( @@ -459,11 +532,12 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" ) - ctx = _resolve_author_mutation_context(worktree_path) + ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] canonical_root = ctx["canonical_repo_root"] process_root = ctx["process_project_root"] real_workspace = os.path.realpath(workspace) + role = ctx.get("workspace_role_kind") or _effective_workspace_role() if real_workspace != process_root: if not _preflight_in_test_mode(): @@ -480,11 +554,15 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace))) if dirty_files: - details = _preflight_workspace_details(workspace, dirty_files) raise RuntimeError( - "Pre-flight order violation: Active task workspace has tracked " - "file edits before mutation (fail closed). " - f"{_format_preflight_workspace_details(details)}" + nwb.format_namespace_workspace_binding_error( + role_kind=role, + workspace_path=workspace, + binding_source=ctx.get("workspace_binding_source") + or "unknown binding source", + dirty_files=dirty_files, + ignored_bindings=ctx.get("ignored_bindings"), + ) ) else: if _preflight_whoami_violation: @@ -500,20 +578,58 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None f"{_format_preflight_files(_preflight_capability_violation_files)}" ) - if _preflight_resolved_role == "reviewer": + if role in {"reviewer", "merger"}: current = _get_workspace_porcelain() baseline = _preflight_capability_baseline_porcelain or "" reviewer_delta = _new_tracked_changes_since(baseline, current) _preflight_reviewer_violation_files = reviewer_delta if reviewer_delta: raise RuntimeError( - "Reviewer role violation: Reviewer profile is forbidden from modifying " + f"{role.title()} role violation: profile is forbidden from modifying " "tracked workspace files (fail closed). Offending files: " f"{_format_preflight_files(reviewer_delta)}" ) _enforce_branches_only_author_mutation(worktree_path) + +def _verify_role_mutation_workspace( + remote: str | None = None, + *, + worktree_path: str | None = None, + worktree: str | None = None, +) -> str: + """Bind reviewer/merger mutations to the active namespace workspace (#510).""" + role = _effective_workspace_role() + git_state = issue_lock_worktree.read_worktree_git_state( + _resolve_preflight_workspace_path(worktree_path) + ) + assessment = nwb.assess_namespace_mutation_workspace( + role_kind=role, + worktree_path=worktree_path, + worktree=worktree, + process_project_root=PROJECT_ROOT, + session_lease_worktree=( + _reviewer_session_worktree() if role in {"reviewer", "merger"} else None + ), + profile_name=get_profile().get("profile_name"), + current_branch=git_state.get("current_branch"), + ) + if assessment["block"]: + raise RuntimeError( + nwb.format_namespace_workspace_binding_error( + role_kind=role, + workspace_path=assessment["mutation_workspace"], + binding_source=assessment.get("workspace_binding_source") + or "unknown binding source", + reasons=assessment.get("reasons"), + ignored_bindings=assessment.get("ignored_bindings"), + ) + ) + resolved = assessment["mutation_workspace"] + verify_preflight_purity(remote, worktree_path=resolved) + return resolved + from mcp.server.fastmcp import FastMCP # noqa: E402 from gitea_auth import ( # noqa: E402 @@ -2649,9 +2765,10 @@ def _evaluate_pr_review_submission( *, live: bool, final_review_decision_ready: bool = False, + worktree_path: str | None = None, ) -> dict: """Shared gate chain for live submit and dry-run review tools.""" - verify_preflight_purity(remote) + _verify_role_mutation_workspace(remote, worktree_path=worktree_path) action = (action or "").strip().lower() result = { "requested_action": action, @@ -3051,6 +3168,7 @@ def gitea_dry_run_pr_review( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Validate review submission mechanics without a live PR mutation.""" return _evaluate_pr_review_submission( @@ -3063,6 +3181,7 @@ def gitea_dry_run_pr_review( org=org, repo=repo, live=False, + worktree_path=worktree_path, ) @@ -3078,6 +3197,7 @@ def gitea_submit_pr_review( org: str | None = None, repo: str | None = None, final_review_decision_ready: bool = False, + worktree_path: str | None = None, ) -> dict: """Gated PR review mutation: comment findings, request changes, or approve. @@ -3096,6 +3216,7 @@ def gitea_submit_pr_review( repo=repo, live=True, final_review_decision_ready=final_review_decision_ready, + worktree_path=worktree_path, ) @@ -3456,6 +3577,7 @@ def gitea_merge_pr( host: str | None = None, org: str | None = None, repo: str | None = None, + worktree_path: str | None = None, ) -> dict: """Gated merge of a Gitea pull request (#16). @@ -3502,6 +3624,10 @@ def gitea_merge_pr( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + worktree_path: Merger workspace path under ``branches/`` or clean + control checkout; defaults to ``GITEA_MERGER_WORKTREE``, + ``GITEA_ACTIVE_WORKTREE``, or the MCP server process root. Ignores + foreign ``GITEA_AUTHOR_WORKTREE`` bindings (#510). Returns: dict describing the attempt: performed, authenticated user, profile @@ -3509,7 +3635,7 @@ def gitea_merge_pr( reasons/gates passed or blocked, and merge result / merge commit if available. Never secrets. """ - verify_preflight_purity(remote) + _verify_role_mutation_workspace(remote, worktree_path=worktree_path) do = (do or "").strip().lower() result = { "performed": False, @@ -4992,7 +5118,7 @@ def gitea_acquire_reviewer_pr_lease( "permission_report": _permission_block_report("gitea.pr.comment"), } - verify_preflight_purity(remote) + _verify_role_mutation_workspace(remote, worktree=worktree) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) profile = get_profile() diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py new file mode 100644 index 0000000..4831907 --- /dev/null +++ b/namespace_workspace_binding.py @@ -0,0 +1,307 @@ +"""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 [], + } \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 651b253..179159d 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3879,7 +3879,7 @@ class TestPreflightVerification(unittest.TestCase): os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n" with self.assertRaises(RuntimeError) as ctx: mcp_server.verify_preflight_purity() - self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception)) + self.assertIn("forbidden from modifying tracked workspace files", str(ctx.exception)) self.assertIn("reviewer_edit.py", str(ctx.exception)) # Foreign pre-existing dirty state does not block when unchanged. @@ -3945,7 +3945,8 @@ class TestPreflightVerification(unittest.TestCase): with self.assertRaises(RuntimeError) as ctx: mcp_server.verify_preflight_purity(worktree_path=worktree) msg = str(ctx.exception) - self.assertIn("active task workspace root", msg) - self.assertIn("inspected git root", msg) - self.assertIn("dirty files: task_file.py", msg) - self.assertIn("dirty scope:", msg) + self.assertIn("resolved workspace", msg) + self.assertIn(worktree, msg) + self.assertIn("worktree_path argument", msg) + self.assertIn("task_file.py", msg) + self.assertIn("author namespace", msg) diff --git a/tests/test_namespace_workspace_binding.py b/tests/test_namespace_workspace_binding.py new file mode 100644 index 0000000..b9c270c --- /dev/null +++ b/tests/test_namespace_workspace_binding.py @@ -0,0 +1,240 @@ +"""Tests for namespace-scoped MCP workspace binding (#510).""" + +from __future__ import annotations + +import os +import sys +import unittest +from pathlib import Path +from unittest import mock +from unittest.mock import MagicMock + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server as srv # noqa: E402 +import namespace_workspace_binding as nwb # noqa: E402 + +REPO_ROOT = Path(__file__).resolve().parent.parent +if REPO_ROOT.parent.name == "branches": + CONTROL_ROOT = str(REPO_ROOT.parent.parent) +else: + CONTROL_ROOT = str(REPO_ROOT) + +AUTHOR_DIRTY = f"{CONTROL_ROOT}/branches/mcp-author-worktree" +MERGER_CLEAN = f"{CONTROL_ROOT}/branches/merge-pr487-submit" +REVIEWER_CLEAN = f"{CONTROL_ROOT}/branches/review-pr487-submit" +RECONCILER_CLEAN = f"{CONTROL_ROOT}/branches/reconcile-pr487" +MCP_PROCESS_ROOT = CONTROL_ROOT + + +class TestNamespaceWorkspaceModule(unittest.TestCase): + def test_author_env_ignored_for_merger_namespace(self): + workspace, source = nwb.resolve_namespace_workspace( + role_kind="merger", + worktree_path=None, + process_project_root=MCP_PROCESS_ROOT, + env={ + nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, + nwb.MERGER_WORKTREE_ENV: MERGER_CLEAN, + }, + profile_name="gitea-merger", + ) + self.assertEqual(workspace, os.path.realpath(MERGER_CLEAN)) + self.assertEqual(source, f"{nwb.MERGER_WORKTREE_ENV} environment variable") + + def test_author_env_ignored_for_reviewer_namespace(self): + workspace, source = nwb.resolve_namespace_workspace( + role_kind="reviewer", + worktree_path=None, + process_project_root=MCP_PROCESS_ROOT, + env={ + nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, + nwb.REVIEWER_WORKTREE_ENV: REVIEWER_CLEAN, + }, + ) + self.assertEqual(workspace, os.path.realpath(REVIEWER_CLEAN)) + self.assertEqual(source, f"{nwb.REVIEWER_WORKTREE_ENV} environment variable") + + def test_author_env_ignored_for_reconciler_namespace(self): + workspace, source = nwb.resolve_namespace_workspace( + role_kind="reconciler", + worktree_path=None, + process_project_root=MCP_PROCESS_ROOT, + env={ + nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY, + nwb.RECONCILER_WORKTREE_ENV: RECONCILER_CLEAN, + }, + ) + self.assertEqual(workspace, os.path.realpath(RECONCILER_CLEAN)) + self.assertEqual(source, f"{nwb.RECONCILER_WORKTREE_ENV} environment variable") + + def test_merger_profile_maps_to_merger_namespace(self): + role = nwb.normalize_role_kind("reviewer", profile_name="gitea-merger") + self.assertEqual(role, "merger") + + def test_error_message_includes_path_and_binding_source(self): + msg = nwb.format_namespace_workspace_binding_error( + role_kind="merger", + workspace_path=AUTHOR_DIRTY, + binding_source=f"{nwb.AUTHOR_WORKTREE_ENV} environment variable", + dirty_files=["gitea_mcp_server.py"], + ignored_bindings=[f"{nwb.AUTHOR_WORKTREE_ENV}={AUTHOR_DIRTY} (ignored for merger namespace)"], + ) + self.assertIn(AUTHOR_DIRTY, msg) + self.assertIn("via", msg.lower()) + self.assertIn(nwb.AUTHOR_WORKTREE_ENV, msg) + self.assertIn("Do not clean or reset foreign role worktrees", msg) + + +class TestNamespaceWorkspaceIntegration(unittest.TestCase): + def setUp(self): + self._saved = { + "whoami_called": srv._preflight_whoami_called, + "capability_called": srv._preflight_capability_called, + "resolved_role": srv._preflight_resolved_role, + "whoami_violation": srv._preflight_whoami_violation, + "capability_violation": srv._preflight_capability_violation, + "in_test": srv._preflight_in_test_mode, + } + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + srv._preflight_in_test_mode = lambda: False + self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False) + self._env_patch.start() + + def tearDown(self): + srv._preflight_whoami_called = self._saved["whoami_called"] + srv._preflight_capability_called = self._saved["capability_called"] + srv._preflight_resolved_role = self._saved["resolved_role"] + srv._preflight_whoami_violation = self._saved["whoami_violation"] + srv._preflight_capability_violation = self._saved["capability_violation"] + srv._preflight_in_test_mode = self._saved["in_test"] + self._env_patch.stop() + for key in ( + nwb.AUTHOR_WORKTREE_ENV, + nwb.MERGER_WORKTREE_ENV, + nwb.REVIEWER_WORKTREE_ENV, + nwb.RECONCILER_WORKTREE_ENV, + nwb.ACTIVE_WORKTREE_ENV, + ): + os.environ.pop(key, None) + + def _merger_profile(self): + return { + "profile_name": "gitea-merger", + "allowed_operations": ["gitea.pr.merge", "gitea.read"], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + } + + def _reviewer_profile(self): + return { + "profile_name": "prgs-reviewer", + "allowed_operations": ["gitea.pr.approve", "gitea.pr.review", "gitea.read"], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + } + + def _reconciler_profile(self): + return { + "profile_name": "prgs-reconciler", + "allowed_operations": ["gitea.pr.close", "gitea.read"], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.create", + ], + } + + def _author_profile(self): + return { + "profile_name": "prgs-author", + "allowed_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.read"], + "forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"], + } + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_author_worktree_does_not_block_merger_with_clean_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN + srv._preflight_resolved_role = "reviewer" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): + srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_author_worktree_does_not_block_reviewer_with_clean_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + os.environ[nwb.REVIEWER_WORKTREE_ENV] = REVIEWER_CLEAN + srv._preflight_resolved_role = "reviewer" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._reviewer_profile()): + srv.verify_preflight_purity("prgs", worktree_path=REVIEWER_CLEAN) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_author_worktree_does_not_block_reconciler_with_clean_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + os.environ[nwb.RECONCILER_WORKTREE_ENV] = RECONCILER_CLEAN + srv._preflight_resolved_role = "reconciler" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._reconciler_profile()): + srv.verify_preflight_purity("prgs", worktree_path=RECONCILER_CLEAN) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_dirty_active_task_workspace_still_blocks_mutations( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN + srv._preflight_resolved_role = "reviewer" + dirty = " M namespace_workspace_binding.py\n" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): + with mock.patch("gitea_mcp_server._get_workspace_porcelain", return_value=dirty): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN) + self.assertIn(MERGER_CLEAN, str(ctx.exception)) + self.assertIn("binding", str(ctx.exception).lower()) + + def test_root_workspace_mutation_still_blocked_for_author(self): + srv._preflight_resolved_role = "author" + with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._author_profile()): + with mock.patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master"}, + ): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity("prgs") + self.assertIn("stable control checkout", str(ctx.exception)) + + @mock.patch("subprocess.run") + @mock.patch("os.path.isdir", return_value=True) + @mock.patch("os.path.exists", return_value=True) + def test_pr487_style_merge_binds_clean_merger_workspace( + self, _exists, _isdir, mock_run + ): + mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n") + os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY + srv._preflight_resolved_role = "reviewer" + with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT): + with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()): + resolved = srv._verify_role_mutation_workspace("prgs") + self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT)) \ No newline at end of file