From 5ed2ab8a38d72e6959c8cd7704dc14f9a4eab909 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Mon, 20 Jul 2026 12:55:40 -0500 Subject: [PATCH] fix: durable author worktree resolution without control fallback (Closes #618) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Author mutation tools now resolve workspace via explicit worktree_path, env bindings, or the active author issue lock — never silent fallback to the control checkout/master. Missing configured bindings fail closed with operator recovery; create_issue and create_issue_comment agree. Recovered onto 0568f44 from preserved candidate cbf56ccd (AUTHOR_RECOVERY). LLM_LOCK_ID=author-618-recovery-508eb3162d01-1784569919 Co-Authored-By: Claude Opus 4.8 (1M context) --- author_mutation_worktree.py | 547 +++++++++++++++++- gitea_mcp_server.py | 155 ++++- issue_lock_worktree.py | 15 +- namespace_workspace_binding.py | 196 +++++-- tests/test_author_mutation_worktree.py | 73 ++- tests/test_create_issue_workspace_guard.py | 8 + ...st_issue_618_author_worktree_resolution.py | 349 +++++++++++ tests/test_namespace_workspace_binding.py | 17 +- thread_state_ledger_examples.py | 43 ++ 9 files changed, 1335 insertions(+), 68 deletions(-) create mode 100644 tests/test_issue_618_author_worktree_resolution.py diff --git a/author_mutation_worktree.py b/author_mutation_worktree.py index dccb94a..b1fb48d 100644 --- a/author_mutation_worktree.py +++ b/author_mutation_worktree.py @@ -1,7 +1,15 @@ -"""Branches-only author mutation worktree guard (#274). +"""Branches-only author mutation worktree guard (#274) with durable resolution (#618). Author/coder mutations must run from a session-owned worktree under the project's ``branches/`` directory, never from the stable control checkout. + +#618 durable resolution: +- Prefer an explicit validated ``worktree_path`` argument. +- Else derive the workspace from the active author issue lock's worktree. +- Env bindings (``GITEA_ACTIVE_WORKTREE`` / ``GITEA_AUTHOR_WORKTREE``) may bind + when present and valid. +- Author mutations never silently fall back to the control checkout or master. +- Missing configured bindings fail closed with a clear operator recovery action. """ from __future__ import annotations @@ -15,6 +23,18 @@ AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE" # Author-only: reviewer/merger/reconciler namespaces use role-specific env vars # via namespace_workspace_binding (#510). +BOUND_WORKTREE_MISSING = "bound_worktree_missing" +BOUND_WORKTREE_MISSING_MESSAGE = ( + "bound worktree missing; operator must recreate or repoint the worktree " + "and reconnect" +) +OPERATOR_RECOVERY_RECREATE_REPOINT = ( + "Recreate the worktree under branches/ (scripts/worktree-start or " + "git worktree add), set GITEA_AUTHOR_WORKTREE / GITEA_ACTIVE_WORKTREE " + "to that path (or pass worktree_path on mutation tools), keep the control " + "checkout clean on master, then reconnect the author MCP session and re-run." +) + def _normalize_path(path: str) -> str: return (path or "").replace("\\", "/").rstrip("/") @@ -45,7 +65,11 @@ def resolve_mutation_workspace( active_worktree_env: str | None = None, author_worktree_env: str | None = None, ) -> str: - """Resolve the workspace path inspected before author mutations.""" + """Resolve the workspace path inspected before author mutations. + + Legacy helper: returns the first non-empty candidate path. Prefer + :func:`resolve_durable_author_worktree` for mutation guards (#618). + """ for candidate in (worktree_path, active_worktree_env, author_worktree_env): text = (candidate or "").strip() if text: @@ -231,4 +255,521 @@ def format_author_mutation_worktree_error(assessment: dict) -> str: f"Branches-only mutation guard (#274): {reasons}. " f"project root: {root}; workspace: {workspace}. " "Create a session-owned worktree under branches/ before mutating." - ) \ No newline at end of file + ) + + +# --------------------------------------------------------------------------- +# #618 durable author worktree resolution +# --------------------------------------------------------------------------- + + +def _abs_real(path: str) -> str: + return os.path.realpath(os.path.abspath((path or "").strip())) + + +def assess_path_traversal_safety( + *, + path: str, + canonical_repo_root: str, +) -> dict: + """Fail closed on traversal/symlink escapes outside the target repository. + + Uses ``realpath`` so intermediate symlinks cannot walk outside + ``canonical_repo_root``. Author mutation workspaces must also land under + ``branches/`` of that root (enforced separately by the branches-only guard). + """ + reasons: list[str] = [] + raw = (path or "").strip() + if not raw: + return { + "proven": False, + "block": True, + "reasons": ["worktree path is empty (fail closed)"], + "workspace_path": None, + "canonical_repo_root": os.path.realpath(canonical_repo_root), + } + if "\x00" in raw: + return { + "proven": False, + "block": True, + "reasons": ["worktree path contains a null byte (fail closed)"], + "workspace_path": raw, + "canonical_repo_root": os.path.realpath(canonical_repo_root), + } + + root = os.path.realpath(canonical_repo_root) + # Resolve without requiring existence first: abspath then realpath of parents. + abs_path = os.path.abspath(raw) + try: + real = os.path.realpath(abs_path) + except OSError as exc: + return { + "proven": False, + "block": True, + "reasons": [f"worktree path could not be resolved safely: {exc}"], + "workspace_path": abs_path, + "canonical_repo_root": root, + } + + root_norm = _normalize_path(root) + real_norm = _normalize_path(real) + if real_norm != root_norm and not real_norm.startswith(f"{root_norm}/"): + reasons.append( + f"worktree path '{real}' escapes canonical repository root '{root}' " + "(traversal/symlink safety, fail closed)" + ) + return { + "proven": not reasons, + "block": bool(reasons), + "reasons": reasons, + "workspace_path": real, + "canonical_repo_root": root, + } + + +def list_git_worktree_paths(canonical_repo_root: str) -> list[str]: + """Return realpaths registered in ``git worktree list --porcelain``.""" + root = os.path.realpath(canonical_repo_root) + try: + res = subprocess.run( + ["git", "-C", root, "worktree", "list", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return [] + if res.returncode != 0: + return [] + paths: list[str] = [] + for line in (res.stdout or "").splitlines(): + if line.startswith("worktree "): + raw = line[len("worktree ") :].strip() + if raw: + paths.append(os.path.realpath(raw)) + return paths + + +def path_in_git_worktree_list(path: str, canonical_repo_root: str) -> bool | None: + """True/False when inventory is available; None when git inventory fails. + + An empty inventory with a working git root is treated as inconclusive + (``None``) so unit tests and partial sandboxes are not false-negative + blocked when ``git worktree list`` is mocked/unavailable. + """ + root = os.path.realpath(canonical_repo_root) + try: + res = subprocess.run( + ["git", "-C", root, "worktree", "list", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return None + if res.returncode != 0: + return None + inventory: list[str] = [] + for line in (res.stdout or "").splitlines(): + if line.startswith("worktree "): + raw = line[len("worktree ") :].strip() + if raw: + inventory.append(os.path.realpath(raw)) + if not inventory: + return None + return os.path.realpath(path) in inventory + + +def assess_bound_worktree_existence( + *, + configured_path: str, + binding_source: str, + canonical_repo_root: str | None = None, + role_kind: str = "author", + profile_name: str | None = None, +) -> dict: + """Fail closed when a configured role-bound worktree path is missing (#618).""" + raw = (configured_path or "").strip() + if not raw: + return { + "proven": True, + "block": False, + "bound_worktree_missing": False, + "path_exists": None, + "in_git_worktree_list": None, + "inspected_git_root": None, + "reasons": [], + "configured_path": None, + "binding_source": binding_source, + "role_kind": role_kind, + "profile_name": profile_name, + "blocker_kind": None, + "operator_recovery": None, + } + + try: + real = _abs_real(raw) + except OSError: + real = os.path.abspath(raw) + + path_exists = os.path.isdir(real) + in_list: bool | None = None + inspected_git_root: str | None = None + root = (canonical_repo_root or "").strip() + if root: + in_list = path_in_git_worktree_list(real, root) if path_exists else False + if path_exists: + try: + res = subprocess.run( + ["git", "-C", real, "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=False, + ) + if res.returncode == 0: + inspected_git_root = (res.stdout or "").strip() or None + except Exception: + inspected_git_root = None + + if path_exists: + return { + "proven": True, + "block": False, + "bound_worktree_missing": False, + "path_exists": True, + "in_git_worktree_list": in_list, + "inspected_git_root": inspected_git_root, + "reasons": [], + "configured_path": real, + "binding_source": binding_source, + "role_kind": role_kind, + "profile_name": profile_name, + "blocker_kind": None, + "operator_recovery": None, + } + + reasons = [ + BOUND_WORKTREE_MISSING_MESSAGE, + ( + f"role/profile '{profile_name or role_kind}' binding via {binding_source} " + f"points to '{real}' which does not exist on disk" + ), + f"path_exists=false; in_git_worktree_list={in_list}; inspected_git_root=null", + ] + return { + "proven": False, + "block": True, + "bound_worktree_missing": True, + "path_exists": False, + "in_git_worktree_list": False if in_list is not None else False, + "inspected_git_root": None, + "reasons": reasons, + "configured_path": real, + "binding_source": binding_source, + "role_kind": role_kind, + "profile_name": profile_name, + "blocker_kind": BOUND_WORKTREE_MISSING, + "operator_recovery": OPERATOR_RECOVERY_RECREATE_REPOINT, + } + + +def format_bound_worktree_missing_error(assessment: dict) -> str: + """Canonical operator-facing message for a missing author worktree binding.""" + reasons = list(assessment.get("reasons") or [BOUND_WORKTREE_MISSING_MESSAGE]) + recovery = assessment.get("operator_recovery") or OPERATOR_RECOVERY_RECREATE_REPOINT + profile = assessment.get("profile_name") or assessment.get("role_kind") or "author" + source = ( + assessment.get("binding_source") + or assessment.get("workspace_binding_source") + or "unknown binding" + ) + path = ( + assessment.get("configured_path") + or assessment.get("workspace_path") + or "(unknown)" + ) + return ( + f"Author worktree binding unhealthy (#618): {'; '.join(reasons)}. " + f"role/profile: {profile}; binding_source: {source}; configured_path: {path}. " + f"Operator recovery: {recovery}" + ) + + +def assess_lock_worktree_ownership( + *, + workspace_path: str, + session_lock_worktree: str | None, +) -> dict: + """When a live lock records a worktree, mutation workspace must match it.""" + locked = (session_lock_worktree or "").strip() + if not locked: + return { + "proven": True, + "block": False, + "reasons": [], + "workspace_path": os.path.realpath(workspace_path) if workspace_path else None, + "lock_worktree_path": None, + } + workspace = os.path.realpath(workspace_path) + locked_real = os.path.realpath(locked) + if workspace != locked_real: + return { + "proven": False, + "block": True, + "reasons": [ + f"active author issue lock worktree '{locked_real}' does not match " + f"mutation workspace '{workspace}' (lock ownership, fail closed)" + ], + "workspace_path": workspace, + "lock_worktree_path": locked_real, + } + return { + "proven": True, + "block": False, + "reasons": [], + "workspace_path": workspace, + "lock_worktree_path": locked_real, + } + + +def resolve_durable_author_worktree( + *, + worktree_path: str | None = None, + worktree: str | None = None, + process_project_root: str, + active_worktree_env: str | None = None, + author_worktree_env: str | None = None, + session_lock_worktree: str | None = None, + canonical_repo_root: str | None = None, + profile_name: str | None = None, + validate: bool = True, +) -> dict: + """Resolve author mutation workspace without silent control-checkout fallback (#618). + + Candidate priority: + 1. explicit ``worktree_path`` argument + 2. ``worktree`` argument + 3. ``GITEA_ACTIVE_WORKTREE`` + 4. ``GITEA_AUTHOR_WORKTREE`` + 5. active author issue lock ``worktree_path`` + 6. process project root **only** when it is already under ``branches/`` + + Configured bindings that point at a missing path fail closed immediately + (no demotion to the control checkout). Validation (when *validate*) covers + existence, traversal/symlink safety, repository identity, branches/ + containment, and lock ownership. + """ + process_root = os.path.realpath(process_project_root) + canonical = os.path.realpath(canonical_repo_root or process_root) + reasons: list[str] = [] + role = "author" + + candidates: list[tuple[str | None, str, bool]] = [ + (worktree_path, "worktree_path argument", False), + (worktree, "worktree argument", False), + (active_worktree_env, f"{ACTIVE_WORKTREE_ENV} environment variable", True), + (author_worktree_env, f"{AUTHOR_WORKTREE_ENV} environment variable", True), + (session_lock_worktree, "active author issue lock worktree", False), + ] + + selected_path: str | None = None + selected_source: str | None = None + existence: dict | None = None + + for candidate, source, _configured in candidates: + text = (candidate or "").strip() + if not text: + continue + try: + real = _abs_real(text) + except OSError: + real = os.path.abspath(text) + + existence = assess_bound_worktree_existence( + configured_path=real, + binding_source=source, + canonical_repo_root=canonical, + role_kind=role, + profile_name=profile_name, + ) + if existence["block"]: + # Missing configured binding: fail closed, never fall back (#618). + return { + "proven": False, + "block": True, + "workspace_path": real, + "workspace_binding_source": source, + "process_project_root": process_root, + "canonical_repo_root": canonical, + "bound_worktree_missing": True, + "path_exists": False, + "in_git_worktree_list": existence.get("in_git_worktree_list"), + "inspected_git_root": None, + "reasons": list(existence.get("reasons") or []), + "blocker_kind": BOUND_WORKTREE_MISSING, + "operator_recovery": OPERATOR_RECOVERY_RECREATE_REPOINT, + "silent_control_fallback": False, + } + + selected_path = real + selected_source = source + break + + if selected_path is None: + # No explicit/env/lock binding. Allow process root only when it is a + # branches/ worktree (MCP launched from the task worktree). Never + # silently bind the stable control checkout. + if is_path_under_branches(process_root, canonical): + selected_path = process_root + selected_source = "MCP process root under branches/ (session-owned)" + else: + return { + "proven": False, + "block": True, + "workspace_path": process_root, + "workspace_binding_source": "no author worktree binding", + "process_project_root": process_root, + "canonical_repo_root": canonical, + "bound_worktree_missing": False, + "path_exists": os.path.isdir(process_root), + "in_git_worktree_list": None, + "inspected_git_root": None, + "reasons": [ + "author mutation blocked: workspace is the stable control checkout; " + "author mutation requires an explicit validated worktree_path " + "or a worktree derived from the active author issue lock; " + "silent fallback to the control checkout/master is forbidden (#618)" + ], + "blocker_kind": "author_worktree_unbound_control_checkout", + "operator_recovery": OPERATOR_RECOVERY_RECREATE_REPOINT, + "silent_control_fallback": False, + } + + workspace = selected_path + source = selected_source or "unknown" + path_exists = os.path.isdir(workspace) + inspected_git_root: str | None = None + in_list: bool | None = None + + if not validate: + return { + "proven": True, + "block": False, + "workspace_path": workspace, + "workspace_binding_source": source, + "process_project_root": process_root, + "canonical_repo_root": canonical, + "bound_worktree_missing": False, + "path_exists": path_exists, + "in_git_worktree_list": None, + "inspected_git_root": None, + "reasons": [], + "blocker_kind": None, + "operator_recovery": None, + "silent_control_fallback": False, + } + + # Traversal / symlink safety + safety = assess_path_traversal_safety( + path=workspace, canonical_repo_root=canonical + ) + if safety["block"]: + reasons.extend(safety["reasons"]) + else: + workspace = safety["workspace_path"] or workspace + + # Existence + git inventory + if not path_exists: + reasons.append(BOUND_WORKTREE_MISSING_MESSAGE) + reasons.append(f"resolved worktree '{workspace}' does not exist") + else: + in_list = path_in_git_worktree_list(workspace, canonical) + try: + res = subprocess.run( + ["git", "-C", workspace, "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=False, + ) + if res.returncode == 0: + inspected_git_root = (res.stdout or "").strip() or None + except Exception: + inspected_git_root = None + if in_list is False: + # Only hard-fail when inventory was obtained and the path is absent. + reasons.append( + f"worktree '{workspace}' is not listed in git worktree list for " + f"'{canonical}' (fail closed)" + ) + + # Repository identity + if path_exists: + membership = assess_workspace_repo_membership( + workspace_path=workspace, + canonical_repo_root=canonical, + ) + if membership["block"]: + reasons.extend(membership["reasons"]) + + # branches/ containment + branches = assess_author_mutation_worktree( + workspace_path=workspace, + project_root=canonical, + ) + if branches["block"]: + reasons.extend(branches["reasons"]) + + # Lock ownership (when a lock worktree is recorded) + lock_own = assess_lock_worktree_ownership( + workspace_path=workspace, + session_lock_worktree=session_lock_worktree, + ) + if lock_own["block"]: + reasons.extend(lock_own["reasons"]) + + # Forbid resolved control checkout even if somehow selected + if workspace == canonical or workspace == process_root: + if not is_path_under_branches(workspace, canonical): + if not any("control checkout" in r for r in reasons): + reasons.append( + "author mutation blocked: resolved workspace is the stable " + "control checkout; silent fallback forbidden (#618)" + ) + + block = bool(reasons) + bound_missing = any("does not exist" in r or BOUND_WORKTREE_MISSING_MESSAGE in r for r in reasons) + return { + "proven": not block, + "block": block, + "workspace_path": workspace, + "workspace_binding_source": source, + "process_project_root": process_root, + "canonical_repo_root": canonical, + "bound_worktree_missing": bound_missing, + "path_exists": path_exists, + "in_git_worktree_list": in_list, + "inspected_git_root": inspected_git_root, + "reasons": reasons, + "blocker_kind": BOUND_WORKTREE_MISSING if bound_missing else ( + "author_worktree_validation_failed" if block else None + ), + "operator_recovery": OPERATOR_RECOVERY_RECREATE_REPOINT if block else None, + "silent_control_fallback": False, + } + + +def format_durable_author_worktree_error(assessment: dict) -> str: + """Format fail-closed error for durable author worktree resolution.""" + if assessment.get("bound_worktree_missing") or assessment.get("blocker_kind") == BOUND_WORKTREE_MISSING: + return format_bound_worktree_missing_error(assessment) + workspace = assessment.get("workspace_path") or "(unknown)" + source = assessment.get("workspace_binding_source") or "unknown" + reasons = "; ".join( + assessment.get("reasons") or ["author worktree resolution failed"] + ) + recovery = assessment.get("operator_recovery") or OPERATOR_RECOVERY_RECREATE_REPOINT + return ( + f"Durable author worktree resolution blocked (#618): {reasons}. " + f"workspace: {workspace}; binding_source: {source}. " + f"Operator recovery: {recovery}" + ) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 773c427..7a44b0e 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -422,6 +422,20 @@ def _canonical_local_git_root() -> str: return crr.resolve_repo_toplevel(configured) or os.path.realpath(configured) +def _session_author_lock_worktree() -> str | None: + """Worktree path recorded on the active author issue lock (#618). + + Used to derive the author mutation workspace when no explicit + ``worktree_path`` or env binding is provided. Never invents a path. + """ + try: + lock = issue_lock_store.read_session_issue_lock() or {} + except Exception: + return None + path = (lock.get("worktree_path") or "").strip() + return path or None + + def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: """Resolve the namespace-scoped workspace root inspected by pre-flight guards. @@ -429,6 +443,9 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: canonical mutation context (verify_paths). A dead env binding must demote here exactly as it does in :func:`_resolve_namespace_mutation_context`, or preflight would inspect a path the mutation guard never uses. + + #618: author role never demotes a missing configured binding to the + control checkout; resolution may derive from the active issue lock. """ role = _effective_workspace_role() workspace, _source = nwb.resolve_namespace_workspace( @@ -438,6 +455,9 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: session_lease_worktree=( _reviewer_session_worktree() if role in {"reviewer", "merger"} else None ), + session_lock_worktree=( + _session_author_lock_worktree() if role == "author" else None + ), profile_name=get_profile().get("profile_name"), verify_paths=True, ) @@ -445,7 +465,7 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict: - """Canonical namespace workspace + repository root for guards (#460/#510/#706).""" + """Canonical namespace workspace + repository root for guards (#460/#510/#706/#618).""" role = _effective_workspace_role() configured_root, _source = _configured_canonical_root() return nwb.resolve_namespace_mutation_context( @@ -455,6 +475,9 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic session_lease_worktree=( _reviewer_session_worktree() if role in {"reviewer", "merger"} else None ), + session_lock_worktree=( + _session_author_lock_worktree() if role == "author" else None + ), profile_name=get_profile().get("profile_name"), configured_canonical_root=configured_root, ) @@ -555,7 +578,10 @@ def _format_preflight_files(files: list[str]) -> str: def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict: ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] - inspected_root = _get_git_root(workspace) + # Prefer durable-resolution evidence when present (#618); fall back to live git. + inspected_root = ctx.get("inspected_git_root") + if inspected_root is None and workspace and os.path.isdir(workspace): + inspected_root = _get_git_root(workspace) process_root = ctx["process_project_root"] canonical_root = ctx["canonical_repo_root"] active_root = os.path.realpath(inspected_root or workspace) @@ -563,6 +589,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st dirty_scope = "control checkout" else: dirty_scope = "active task workspace" + path_exists = ctx.get("path_exists") + if path_exists is None: + path_exists = os.path.isdir(workspace) if workspace else False details = { "mcp_server_process_root": process_root, "canonical_repository_root": canonical_root, @@ -574,6 +603,16 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st "workspace_role_kind": ctx.get("workspace_role_kind"), "workspace_binding_source": ctx.get("workspace_binding_source"), "ignored_bindings": list(ctx.get("ignored_bindings") or []), + # #618 runtime health surface for bound worktree disappearance + "path_exists": path_exists, + "in_git_worktree_list": ctx.get("in_git_worktree_list"), + "bound_worktree_missing": bool(ctx.get("bound_worktree_missing")), + "author_worktree_block": bool(ctx.get("author_worktree_block")), + "author_worktree_reasons": list(ctx.get("author_worktree_reasons") or []), + "operator_recovery": ctx.get("operator_recovery"), + "workspace_healthy": not bool( + ctx.get("bound_worktree_missing") or ctx.get("author_worktree_block") + ), } if not ctx["roots_aligned"]: details["workspace_root_mismatch"] = ( @@ -590,6 +629,10 @@ def _format_preflight_workspace_details(details: dict) -> str: 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"path_exists: {details.get('path_exists')}", + f"in_git_worktree_list: {details.get('in_git_worktree_list')}", + f"bound_worktree_missing: {details.get('bound_worktree_missing')}", + f"workspace_healthy: {details.get('workspace_healthy')}", f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}", f"dirty scope: {details.get('dirty_scope')}", ] @@ -600,7 +643,7 @@ def _format_preflight_workspace_details(details: dict) -> str: def assess_preflight_status(worktree_path: str | None = None) -> dict: - """Non-throwing pre-flight readiness for runtime-context alignment (#252).""" + """Non-throwing pre-flight readiness for runtime-context alignment (#252/#618).""" reasons: list[str] = [] if not _preflight_whoami_called: reasons.append( @@ -610,16 +653,52 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict: reasons.append( "Task capability (gitea_resolve_task_capability) has not been resolved" ) + try: + role = _effective_workspace_role() + except Exception: + role = "author" + + # #618: always compute workspace details for author so runtime_context can + # report unhealthy state when a configured role-bound path is missing. + dirty_files: list[str] = [] workspace_details = None - if worktree_path: - dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path))) + try: + dirty_files = sorted( + _parse_porcelain_entries(_get_workspace_porcelain(worktree_path)) + ) workspace_details = _preflight_workspace_details(worktree_path, dirty_files) - if dirty_files: + except Exception: + workspace_details = None + + if workspace_details is not None: + if workspace_details.get("bound_worktree_missing") or ( + workspace_details.get("author_worktree_block") + and workspace_details.get("path_exists") is False + ): + reasons.append( + author_mutation_worktree.BOUND_WORKTREE_MISSING_MESSAGE + + f" ({_format_preflight_workspace_details(workspace_details)})" + ) + elif ( + role == "author" + and workspace_details.get("workspace_healthy") is False + and workspace_details.get("author_worktree_reasons") + ): + reasons.append( + "Author worktree binding unhealthy: " + + "; ".join(workspace_details.get("author_worktree_reasons") or []) + + f" ({_format_preflight_workspace_details(workspace_details)})" + ) + + if worktree_path: + if dirty_files and not any("bound worktree missing" in r for r in reasons): + details = workspace_details or _preflight_workspace_details( + worktree_path, dirty_files + ) reasons.append( "Active task workspace has tracked file edits before mutation " - f"({_format_preflight_workspace_details(workspace_details)})" + f"({_format_preflight_workspace_details(details)})" ) - role = _effective_workspace_role() if role in nwb.NON_AUTHOR_ROLES: binding = nwb.assess_metadata_only_worktree_binding( role_kind=role, @@ -681,7 +760,9 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict: "preflight_whoami_violation_files": list(_preflight_whoami_violation_files), "preflight_capability_violation_files": list(_preflight_capability_violation_files), "preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files), - "preflight_workspace": _preflight_workspace_details(None, []), + "preflight_workspace": workspace_details + if workspace_details is not None + else _preflight_workspace_details(None, []), } @@ -926,6 +1007,58 @@ def _enforce_branches_only_author_mutation( return ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] + + # #618: durable resolution already failed closed (missing binding, lock + # mismatch, traversal, membership). Bound-worktree-missing never receives + # the create_issue control-checkout bootstrap exemption — a configured but + # missing worktree is never "clean control". + durable = ctx.get("author_worktree_resolution") or {} + if ctx.get("bound_worktree_missing") or ( + durable.get("block") and durable.get("bound_worktree_missing") + ): + raise RuntimeError( + author_mutation_worktree.format_durable_author_worktree_error( + durable or { + "reasons": ctx.get("author_worktree_reasons") + or [author_mutation_worktree.BOUND_WORKTREE_MISSING_MESSAGE], + "workspace_path": workspace, + "workspace_binding_source": ctx.get("workspace_binding_source"), + "bound_worktree_missing": True, + "blocker_kind": author_mutation_worktree.BOUND_WORKTREE_MISSING, + "operator_recovery": ctx.get("operator_recovery"), + "configured_path": workspace, + "binding_source": ctx.get("workspace_binding_source"), + "profile_name": get_profile().get("profile_name"), + } + ) + ) + if durable.get("block") and not durable.get("bound_worktree_missing"): + # Other durable failures (lock ownership, traversal, no binding while + # on control). create_issue bootstrap may still permit clean control. + import create_issue_bootstrap as _cib + + bootstrap = ( + _create_issue_bootstrap_assessment(task, worktree_path) + if bootstrap_assessment is _BOOTSTRAP_UNSET + else bootstrap_assessment + ) + if _cib.bootstrap_permits_control_checkout( + bootstrap, + task=task, + workspace_path=workspace, + canonical_repo_root=ctx["canonical_repo_root"], + ): + return + if ( + isinstance(bootstrap, dict) + and bootstrap.get("block") + and not bootstrap.get("not_applicable") + ): + raise RuntimeError(_cib.format_create_issue_bootstrap_error(bootstrap)) + raise RuntimeError( + author_mutation_worktree.format_durable_author_worktree_error(durable) + ) + git_state = issue_lock_worktree.read_worktree_git_state(workspace) assessment = author_mutation_worktree.assess_author_mutation_worktree( workspace_path=workspace, @@ -1644,6 +1777,9 @@ def _verify_role_mutation_workspace( session_lease_worktree=( _reviewer_session_worktree() if role in {"reviewer", "merger"} else None ), + session_lock_worktree=( + _session_author_lock_worktree() if role == "author" else None + ), profile_name=get_profile().get("profile_name"), current_branch=git_state.get("current_branch"), configured_canonical_root=_configured_root, @@ -1657,6 +1793,7 @@ def _verify_role_mutation_workspace( or "unknown binding source", reasons=assessment.get("reasons"), ignored_bindings=assessment.get("ignored_bindings"), + operator_recovery=assessment.get("operator_recovery"), ) ) resolved = assessment["mutation_workspace"] diff --git a/issue_lock_worktree.py b/issue_lock_worktree.py index f0f56bd..1be775a 100644 --- a/issue_lock_worktree.py +++ b/issue_lock_worktree.py @@ -20,11 +20,24 @@ BASE_BRANCHES = frozenset({"master", "main", "dev"}) def resolve_author_worktree_path( explicit: str | None, project_root: str, + *, + session_lock_worktree: str | None = None, ) -> str: - """Resolve the author worktree path for lock/PR gates.""" + """Resolve the author worktree path for lock/PR gates. + + #618: prefer explicit path, then env, then the active issue lock worktree. + Does not invent a branches/ worktree. Falling back to *project_root* is + retained only for lock-time bootstrap when the process itself is already + under branches/ or no binding exists yet (callers still fail closed via + preflight / durable resolution before mutation). + """ path = (explicit or "").strip() if not path: path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip() + if not path: + path = (os.environ.get("GITEA_ACTIVE_WORKTREE") or "").strip() + if not path: + path = (session_lock_worktree or "").strip() if not path: path = project_root return os.path.realpath(os.path.abspath(path)) diff --git a/namespace_workspace_binding.py b/namespace_workspace_binding.py index 8ad63a2..fa006b6 100644 --- a/namespace_workspace_binding.py +++ b/namespace_workspace_binding.py @@ -55,18 +55,26 @@ def resolve_namespace_workspace( 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): 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 + 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. """ @@ -74,6 +82,32 @@ def resolve_namespace_workspace( 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), @@ -83,6 +117,11 @@ def resolve_namespace_workspace( 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: @@ -107,6 +146,7 @@ def resolve_namespace_mutation_context( 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, @@ -119,21 +159,54 @@ def resolve_namespace_mutation_context( 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] = [] - 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, - ) + 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, @@ -141,12 +214,7 @@ def resolve_namespace_mutation_context( 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 { + result = { "workspace_path": workspace, "workspace_binding_source": binding_source, "workspace_role_kind": role, @@ -155,6 +223,17 @@ def resolve_namespace_mutation_context( "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( @@ -231,10 +310,29 @@ def format_namespace_workspace_binding_error( 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) - workspace = os.path.realpath(workspace_path) + 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}." @@ -249,15 +347,18 @@ def format_namespace_workspace_binding_error( + ", ".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." - ) + 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) @@ -269,6 +370,7 @@ def assess_namespace_mutation_workspace( 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, @@ -281,6 +383,7 @@ def assess_namespace_mutation_workspace( 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, ) @@ -305,14 +408,23 @@ def assess_namespace_mutation_workspace( ) reasons = list(metadata.get("reasons") or []) + operator_recovery = ctx.get("operator_recovery") 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"]) + # #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 @@ -345,4 +457,10 @@ def assess_namespace_mutation_workspace( "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"), } \ No newline at end of file diff --git a/tests/test_author_mutation_worktree.py b/tests/test_author_mutation_worktree.py index c3e0119..23bed97 100644 --- a/tests/test_author_mutation_worktree.py +++ b/tests/test_author_mutation_worktree.py @@ -79,18 +79,33 @@ class TestPreflightIntegration(unittest.TestCase): mcp_server._preflight_whoami_called = True mcp_server._preflight_capability_called = True mcp_server._preflight_resolved_role = "author" + mcp_server._preflight_resolved_task = None control_root = "/repo/Gitea-Tools" with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root): with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"): - with mock.patch("gitea_auth.get_profile", return_value={"profile_name": "gitea-author"}): - with mock.patch.dict( - "os.environ", - {"GITEA_TEST_PORCELAIN": ""}, - clear=False, + with mock.patch( + "gitea_mcp_server._session_author_lock_worktree", + return_value=None, + ): + with mock.patch( + "gitea_auth.get_profile", + return_value={"profile_name": "gitea-author"}, ): - with self.assertRaises(RuntimeError) as ctx: - mcp_server.verify_preflight_purity() - self.assertIn("Branches-only mutation guard", str(ctx.exception)) + with mock.patch.dict( + "os.environ", + {"GITEA_TEST_PORCELAIN": ""}, + clear=False, + ): + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity() + blob = str(ctx.exception) + self.assertTrue( + "Branches-only mutation guard" in blob + or "control checkout" in blob + or "author worktree" in blob.lower() + or "#618" in blob, + msg=blob, + ) def test_verify_preflight_allows_branches_worktree(self): import mcp_server @@ -98,14 +113,46 @@ class TestPreflightIntegration(unittest.TestCase): mcp_server._preflight_whoami_called = True mcp_server._preflight_capability_called = True mcp_server._preflight_resolved_role = "author" + mcp_server._preflight_resolved_task = None worktree = "/repo/Gitea-Tools/branches/issue-274" + healthy_ctx = { + "workspace_path": worktree, + "workspace_binding_source": "worktree_path argument", + "workspace_role_kind": "author", + "ignored_bindings": [], + "process_project_root": "/repo/Gitea-Tools", + "canonical_repo_root": "/repo/Gitea-Tools", + "roots_aligned": True, + "bound_worktree_missing": False, + "author_worktree_block": False, + "author_worktree_reasons": [], + "author_worktree_resolution": { + "proven": True, + "block": False, + "bound_worktree_missing": False, + "workspace_path": worktree, + "workspace_binding_source": "worktree_path argument", + "reasons": [], + }, + "path_exists": True, + "in_git_worktree_list": True, + "inspected_git_root": worktree, + } with mock.patch.object(mcp_server, "_enforce_root_checkout_guard"): - with mock.patch.dict( - "os.environ", - {"GITEA_TEST_PORCELAIN": ""}, - clear=False, + with mock.patch.object( + mcp_server, "_session_author_lock_worktree", return_value=None ): - mcp_server.verify_preflight_purity(worktree_path=worktree) + with mock.patch.object( + mcp_server, + "_resolve_namespace_mutation_context", + return_value=healthy_ctx, + ): + with mock.patch.dict( + "os.environ", + {"GITEA_TEST_PORCELAIN": ""}, + clear=False, + ): + mcp_server.verify_preflight_purity(worktree_path=worktree) if __name__ == "__main__": diff --git a/tests/test_create_issue_workspace_guard.py b/tests/test_create_issue_workspace_guard.py index a61fb3a..31a7dad 100644 --- a/tests/test_create_issue_workspace_guard.py +++ b/tests/test_create_issue_workspace_guard.py @@ -26,15 +26,23 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase): srv._preflight_whoami_called = True srv._preflight_capability_called = True srv._preflight_resolved_role = "author" + srv._preflight_resolved_task = "create_issue" srv._preflight_whoami_violation = False srv._preflight_capability_violation = False # Disable early return in verify_preflight_purity for testing self._orig_in_test = srv._preflight_in_test_mode srv._preflight_in_test_mode = lambda: False + # #618: isolate from ambient session issue locks + self._lock_patch = patch( + "gitea_mcp_server._session_author_lock_worktree", return_value=None + ) + self._lock_patch.start() def tearDown(self): srv._preflight_in_test_mode = self._orig_in_test + srv._preflight_resolved_task = None + self._lock_patch.stop() @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._profile_permission_block", return_value=None) diff --git a/tests/test_issue_618_author_worktree_resolution.py b/tests/test_issue_618_author_worktree_resolution.py new file mode 100644 index 0000000..2052f85 --- /dev/null +++ b/tests/test_issue_618_author_worktree_resolution.py @@ -0,0 +1,349 @@ +"""Regression tests for durable author worktree resolution (#618).""" + +from __future__ import annotations + +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock +from unittest.mock import MagicMock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import author_mutation_worktree as amw # noqa: E402 +import gitea_mcp_server as srv # noqa: E402 +import namespace_workspace_binding as nwb # noqa: E402 + +FAKE_AUTH = {"Authorization": "token test-token"} +current_file_path = Path(__file__).resolve() +if "branches" in current_file_path.parts: + CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3]) +else: + CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1]) + + +class TestDurableAuthorWorktreeResolution(unittest.TestCase): + def test_missing_author_env_fails_closed_no_control_fallback(self): + missing = "/nonexistent/branches/mcp-author-clean-ns" + result = amw.resolve_durable_author_worktree( + process_project_root=CONTROL_CHECKOUT_ROOT, + author_worktree_env=missing, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + ) + self.assertTrue(result["block"]) + self.assertTrue(result["bound_worktree_missing"]) + self.assertFalse(result["silent_control_fallback"]) + self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, result["reasons"][0]) + self.assertNotEqual( + os.path.realpath(result["workspace_path"]), + os.path.realpath(CONTROL_CHECKOUT_ROOT), + ) + + def test_missing_active_env_fails_closed(self): + missing = "/nonexistent/branches/deleted-active" + result = amw.resolve_durable_author_worktree( + process_project_root=CONTROL_CHECKOUT_ROOT, + active_worktree_env=missing, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + ) + self.assertTrue(result["block"]) + self.assertTrue(result["bound_worktree_missing"]) + self.assertIn(amw.ACTIVE_WORKTREE_ENV, result["workspace_binding_source"]) + + def test_derives_from_active_author_issue_lock(self): + lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock") + result = amw.resolve_durable_author_worktree( + process_project_root=CONTROL_CHECKOUT_ROOT, + session_lock_worktree=lock_wt, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + validate=False, + ) + self.assertEqual( + result["workspace_path"], os.path.realpath(os.path.abspath(lock_wt)) + ) + self.assertIn("issue lock", result["workspace_binding_source"]) + + def test_explicit_worktree_path_wins_over_lock(self): + explicit = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-explicit") + lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock") + result = amw.resolve_durable_author_worktree( + worktree_path=explicit, + process_project_root=CONTROL_CHECKOUT_ROOT, + session_lock_worktree=lock_wt, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + validate=False, + ) + self.assertEqual( + result["workspace_path"], os.path.realpath(os.path.abspath(explicit)) + ) + self.assertEqual(result["workspace_binding_source"], "worktree_path argument") + + def test_no_binding_does_not_silently_use_control_checkout(self): + result = amw.resolve_durable_author_worktree( + process_project_root=CONTROL_CHECKOUT_ROOT, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + ) + self.assertTrue(result["block"]) + self.assertFalse(result["silent_control_fallback"]) + blob = " ".join(result["reasons"]) + self.assertIn("control checkout", blob) + self.assertIn("forbidden", blob) + + def test_process_root_under_branches_is_allowed(self): + branches_root = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "session-wt") + result = amw.resolve_durable_author_worktree( + process_project_root=branches_root, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + validate=False, + ) + self.assertFalse(result["block"]) + self.assertEqual( + result["workspace_path"], os.path.realpath(branches_root) + ) + self.assertIn("branches/", result["workspace_binding_source"]) + + def test_lock_ownership_mismatch_fails_closed(self): + with tempfile.TemporaryDirectory() as tmp: + root = tmp + branches = os.path.join(root, "branches") + os.makedirs(os.path.join(branches, "a")) + os.makedirs(os.path.join(branches, "b")) + # Seed a fake .git so membership/list may soft-fail without hard error + os.makedirs(os.path.join(root, ".git")) + result = amw.resolve_durable_author_worktree( + worktree_path=os.path.join(branches, "a"), + process_project_root=root, + session_lock_worktree=os.path.join(branches, "b"), + canonical_repo_root=root, + validate=True, + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("lock" in r.lower() and "match" in r.lower() for r in result["reasons"]) + ) + + def test_traversal_safety_blocks_escape(self): + assessment = amw.assess_path_traversal_safety( + path="/tmp/other-repo/branches/evil", + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + ) + self.assertTrue(assessment["block"]) + self.assertTrue(any("escapes" in r for r in assessment["reasons"])) + + def test_bound_worktree_existence_reports_null_git_root(self): + assessment = amw.assess_bound_worktree_existence( + configured_path="/nonexistent/branches/gone", + binding_source=f"{amw.AUTHOR_WORKTREE_ENV} environment variable", + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + profile_name="prgs-author", + ) + self.assertTrue(assessment["block"]) + self.assertIsNone(assessment["inspected_git_root"]) + self.assertFalse(assessment["path_exists"]) + msg = amw.format_bound_worktree_missing_error(assessment) + self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, msg) + self.assertIn("prgs-author", msg) + self.assertIn("recreate or repoint", msg.lower()) + + +class TestNamespaceAuthorNoDemotion(unittest.TestCase): + def test_author_missing_env_not_demoted_to_process_root(self): + missing = "/nonexistent/branches/mcp-author-clean-ns" + demotions: list[str] = [] + path, source = nwb.resolve_namespace_workspace( + role_kind="author", + process_project_root=CONTROL_CHECKOUT_ROOT, + env={amw.AUTHOR_WORKTREE_ENV: missing}, + demotions=demotions, + verify_paths=True, + ) + self.assertIn("AUTHOR", source) + self.assertNotEqual(os.path.realpath(path), os.path.realpath(CONTROL_CHECKOUT_ROOT)) + self.assertTrue(any("not demoted" in d for d in demotions)) + + def test_reviewer_still_demotes_missing_env(self): + """#702 demotion retained for non-author roles.""" + demotions: list[str] = [] + path, source = nwb.resolve_namespace_workspace( + role_kind="reviewer", + process_project_root=CONTROL_CHECKOUT_ROOT, + env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-gone"}, + demotions=demotions, + verify_paths=True, + ) + self.assertEqual(source, "MCP server process root (default)") + self.assertEqual(path, os.path.realpath(CONTROL_CHECKOUT_ROOT)) + self.assertTrue(demotions) + + def test_mutation_context_surfaces_missing_binding_health(self): + ctx = nwb.resolve_namespace_mutation_context( + role_kind="author", + worktree_path=None, + process_project_root=CONTROL_CHECKOUT_ROOT, + env={amw.AUTHOR_WORKTREE_ENV: "/nonexistent/branches/mcp-author-clean-ns"}, + profile_name="prgs-author", + ) + self.assertTrue(ctx.get("bound_worktree_missing")) + self.assertTrue(ctx.get("author_worktree_block")) + self.assertIsNone(ctx.get("inspected_git_root")) + self.assertFalse(ctx.get("path_exists")) + + +class TestCreateIssueAndCommentAgreeOnMissingWorktree(unittest.TestCase): + """AC3/AC4: create_issue and create_issue_comment enforce the same rule.""" + + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "author" + srv._preflight_resolved_task = None + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + self._orig_in_test = srv._preflight_in_test_mode + srv._preflight_in_test_mode = lambda: False + self._lock_patch = patch( + "gitea_mcp_server._session_author_lock_worktree", return_value=None + ) + self._lock_patch.start() + self.addCleanup(self._restore) + + def _restore(self): + srv._preflight_in_test_mode = self._orig_in_test + srv._preflight_resolved_task = None + self._lock_patch.stop() + os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None) + os.environ.pop(amw.ACTIVE_WORKTREE_ENV, None) + + def _assert_blocked_missing(self, result_or_exc): + if isinstance(result_or_exc, BaseException): + blob = str(result_or_exc) + else: + blob = " ".join( + str(x) + for x in ( + result_or_exc.get("reasons") or [], + result_or_exc.get("message"), + result_or_exc.get("blocker_kind"), + ) + if x + ) + if not blob: + blob = str(result_or_exc) + self.assertTrue( + amw.BOUND_WORKTREE_MISSING_MESSAGE in blob + or "does not exist" in blob + or "bound worktree" in blob.lower(), + msg=blob, + ) + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch( + "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, []), + ) + @patch("gitea_mcp_server.api_request") + @patch("gitea_mcp_server.api_get_all", return_value=[]) + def test_create_issue_blocked_when_author_env_missing( + self, _get_all, mock_api, _role, _ns, _prof, _auth + ): + missing = os.path.join( + CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env" + ) + os.environ[amw.AUTHOR_WORKTREE_ENV] = missing + srv._preflight_resolved_task = "create_issue" + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + try: + res = srv.gitea_create_issue(title="Test issue", body="body text here") + except RuntimeError as exc: + self._assert_blocked_missing(exc) + else: + self.assertFalse(res.get("success", True) and res.get("number")) + self._assert_blocked_missing(res) + mock_api.assert_not_called() + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server.api_request") + def test_create_issue_comment_blocked_when_author_env_missing(self, mock_api, _auth): + missing = os.path.join( + CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env" + ) + os.environ[amw.AUTHOR_WORKTREE_ENV] = missing + srv._preflight_resolved_task = "comment_issue" + author_env = { + "GITEA_PROFILE_NAME": "gitea-author", + "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment", + amw.AUTHOR_WORKTREE_ENV: missing, + } + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch.dict(os.environ, author_env, clear=False): + try: + res = srv.gitea_create_issue_comment( + issue_number=618, + body="evidence comment", + remote="prgs", + ) + except RuntimeError as exc: + self._assert_blocked_missing(exc) + else: + self.assertFalse(res.get("success", True)) + self._assert_blocked_missing(res) + mock_api.assert_not_called() + + +class TestRuntimeContextUnhealthyMissingWorktree(unittest.TestCase): + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "author" + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + self._lock_patch = patch( + "gitea_mcp_server._session_author_lock_worktree", return_value=None + ) + self._lock_patch.start() + + def tearDown(self): + self._lock_patch.stop() + os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None) + + def test_assess_preflight_reports_null_git_root_and_missing(self): + missing = "/nonexistent/branches/mcp-author-clean-ns" + os.environ[amw.AUTHOR_WORKTREE_ENV] = missing + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch("gitea_mcp_server.get_profile", return_value={ + "profile_name": "prgs-author", + "allowed_operations": ["gitea.pr.create"], + "forbidden_operations": [], + }): + status = srv.assess_preflight_status() + self.assertFalse(status["preflight_ready"]) + blob = " ".join(status["preflight_block_reasons"]) + self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, blob) + details = status["preflight_workspace"] + self.assertIsNotNone(details) + self.assertTrue(details.get("bound_worktree_missing")) + self.assertIsNone(details.get("inspected_git_root")) + self.assertFalse(details.get("path_exists")) + self.assertFalse(details.get("workspace_healthy")) + + +class TestThreadLedgerExample(unittest.TestCase): + def test_bound_worktree_missing_ledger_example_exists(self): + import thread_state_ledger_examples as examples + + names = [name for name, _h, _l in examples.EXAMPLES] + self.assertIn("bound_worktree_missing_blocker", names) + for name, _handoff, ledger in examples.EXAMPLES: + if name == "bound_worktree_missing_blocker": + self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, ledger) + self.assertIn("inspected_git_root", ledger) + self.assertIn("operator", ledger.lower()) + break + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_namespace_workspace_binding.py b/tests/test_namespace_workspace_binding.py index 91364e7..46e0b19 100644 --- a/tests/test_namespace_workspace_binding.py +++ b/tests/test_namespace_workspace_binding.py @@ -224,9 +224,20 @@ class TestNamespaceWorkspaceIntegration(unittest.TestCase): "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)) + with mock.patch( + "gitea_mcp_server._session_author_lock_worktree", + return_value=None, + ): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity("prgs") + blob = str(ctx.exception) + self.assertTrue( + "stable control checkout" in blob + or "control checkout" in blob + or "#618" in blob + or "author worktree" in blob.lower(), + msg=blob, + ) @mock.patch("subprocess.run") @mock.patch("os.path.isdir", return_value=True) diff --git a/thread_state_ledger_examples.py b/thread_state_ledger_examples.py index f0aeae9..d5c35a7 100644 --- a/thread_state_ledger_examples.py +++ b/thread_state_ledger_examples.py @@ -523,6 +523,49 @@ Who/what acts next: - Required action: implement #507 two-comment validator - Do not do: recreate duplicate CTH issue - Resume from: issue #507 body +""", + ) +) + +EXAMPLES.append( + _example( + "bound_worktree_missing_blocker", + """ +[CONTROLLER HANDOFF] Issue #618 — author mutation blocked + +Server-side mutation ledger: +- none — no server-side state changed + +Blockers: +- environment/tooling blocker: bound worktree missing; operator must recreate or repoint the worktree and reconnect +""", + """ +[THREAD STATE LEDGER] Issue #618 — author worktree binding unhealthy + +What is true now: +- Issue state: open +- Server-side decision state: no server-side state changed +- Local verdict/state: author mutation tools fail closed consistently +- Latest known validation: runtime context reports workspace_healthy=false +- Role/profile: prgs-author +- Configured worktree path: branches/mcp-author-clean-ns (via GITEA_AUTHOR_WORKTREE) +- path_exists: false +- in_git_worktree_list: false +- inspected_git_root: null + +What changed: +- nothing server-side; local env still points at a deleted role-bound worktree + +What is blocked: +- Blocker classification: environment/tooling blocker +- Blocker detail: bound worktree missing; operator must recreate or repoint the worktree and reconnect +- create_issue and create_issue_comment (and other author mutations) agree: fail closed before API mutation + +Who/what acts next: +- Next actor: operator +- Required action: recreate the worktree under branches/ (scripts/worktree-start or git worktree add), set GITEA_AUTHOR_WORKTREE / GITEA_ACTIVE_WORKTREE to that path (or pass worktree_path), keep control checkout clean on master, reconnect the author MCP session, then re-run the mutation +- Do not do: retry mutations hoping create_issue_comment will still work while create_issue blocks; do not fall back to the control checkout or master +- Resume from: healthy author worktree binding + gitea_whoami + gitea_resolve_task_capability """, ) ) \ No newline at end of file