Merge branch 'master' into feat/issue-664-break-glass-restart

docs/mcp-restart-coordinator.md

@@ -144,6 +144,19 @@ tool argument expresses caller intent and cannot be self-asserted by a worker
 session. `break_glass_requested` and `break_glass_authorized` are both reported,
 so a bypass is never silent.
 
+### Break-glass Restart Workflow (`gitea_break_glass_restart`, #664)
+
+The dedicated MCP tool `gitea_break_glass_restart` provides the privileged emergency break-glass restart workflow when graceful drain cannot complete:
+
+- **Role Authorization (#664 AC1)**: Ordinary LLM worker roles (`author`, `reviewer`, `merger`, `reconciler`) are denied fail-closed. Privileged `controller` role or explicit `GITEA_BREAKGLASS_RESTART_AUTHORIZATION` is required.
+- **Required Parameters (#664 AC2)**:
+  - `reason`: Mandatory non-empty string (min 10 characters).
+  - `confirmation`: Must equal exactly `"I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION"`.
+  - `impact_ack`: Must be `True`.
+- **Automatic Incident Creation (#664 AC3)**: Creates a Gitea incident issue (`[INCIDENT] Break-glass MCP restart invoked by ...`) detailing the reason, timestamp, disrupted sessions, and linking `#652 #653 #655 #630 #658 #662 #664`.
+- **Immutable Audit Entry**: Records an immutable audit log entry under `event="break_glass_mcp_restart"`.
+- **Mandatory Reconciliation (#664 AC4)**: Sets `reconciliation_required=True` requiring post-restart reconciliation via `gitea_reconcile_after_restart` (#662).
+
 ### Fail closed on apply
 
 A missing, malformed, expired, unclean, tampered, or fingerprint-stale drain
@@ -160,3 +173,4 @@ profiles are operational metadata only.
 
 A representative dry-run report is in
 [`mcp-restart-impact-sample.json`](./mcp-restart-impact-sample.json).
+

gitea_auth.py

@@ -22,62 +22,8 @@ import gitea_config
 
 PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
 
-# Reserved runtime-control / workspace-binding environment variables (#704).
+# Load standard .env if present
-# Repository .env files MUST NOT populate or override any of these keys.
+load_dotenv(os.path.join(PROJECT_ROOT, ".env"))
-RESERVED_WORKTREE_ENV_KEYS: frozenset[str] = frozenset({
-    "GITEA_ACTIVE_WORKTREE",
-    "GITEA_AUTHOR_WORKTREE",
-    "GITEA_REVIEWER_WORKTREE",
-    "GITEA_MERGER_WORKTREE",
-    "GITEA_RECONCILER_WORKTREE",
-})
-
-
-def is_reserved_worktree_env_key(key: str | None) -> bool:
-    """Return True if *key* is a reserved runtime workspace binding variable (#704)."""
-    if not key:
-        return False
-    k = str(key).upper().strip()
-    return k in RESERVED_WORKTREE_ENV_KEYS or (k.startswith("GITEA_") and k.endswith("_WORKTREE"))
-
-
-def load_env_file_sanitized(
-    env_path: str,
-    *,
-    target_env: dict | os._Environ | None = None,
-) -> list[str]:
-    """Load a .env file without populating or overriding reserved workspace keys (#704).
-
-    Pre-existing process environment values retain their precedence. Reserved
-    runtime-control keys found in repository files are ignored (without logging
-    their values). Returns a list of sanitized rejection reasons.
-    """
-    if target_env is None:
-        target_env = os.environ
-    if not os.path.exists(env_path) or os.path.isdir(env_path):
-        return []
-
-    rejection_reasons: list[str] = []
-    try:
-        file_vals = dotenv_values(env_path)
-        for key, val in file_vals.items():
-            if not key or val is None:
-                continue
-            if is_reserved_worktree_env_key(key):
-                filename = os.path.basename(env_path)
-                rejection_reasons.append(
-                    f"Ignored reserved runtime workspace key '{key}' from repository {filename}"
-                )
-                continue
-            if key not in target_env:
-                target_env[key] = val
-    except Exception:
-        pass
-    return rejection_reasons
-
-
-# Load standard .env if present (sanitized to prevent repo workspace binding contamination #704)
-load_env_file_sanitized(os.path.join(PROJECT_ROOT, ".env"))
 
 # Dictionary to store configurations parsed dynamically from .env.* files
 DYNAMIC_CONFIGS = {}
@@ -91,13 +37,9 @@ for env_path in glob.glob(os.path.join(PROJECT_ROOT, ".env*")):
         continue
     try:
         config_vals = dotenv_values(env_path)
-        # Filter out reserved workspace keys from dynamic configs (#704)
+        site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
-        sanitized_config = {
-            k: v for k, v in config_vals.items() if not is_reserved_worktree_env_key(k)
-        }
-        site = sanitized_config.get("GITEA_SITE") or sanitized_config.get("GITEA_HOST")
         if site:
-            DYNAMIC_CONFIGS[site.lower().strip()] = sanitized_config
+            DYNAMIC_CONFIGS[site.lower().strip()] = config_vals
     except Exception:
         pass
 

gitea_mcp_server.py

@@ -23922,6 +23922,205 @@ def gitea_request_mcp_restart(
     return payload
 
 
+BREAK_GLASS_CONFIRMATION_PHRASE = "I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION"
+
+
+@mcp.tool()
+def gitea_break_glass_restart(
+    reason: str,
+    confirmation: str,
+    impact_ack: bool = False,
+    restart_class: str = "full_mcp_restart",
+    create_incident_issue: bool = True,
+    dry_run: bool = False,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    worktree_path: str | None = None,
+) -> dict:
+    """Privileged emergency break-glass MCP restart workflow (#664).
+
+    Break-glass restart permits emergency recovery when graceful drain cannot
+    complete. It requires:
+    1. Privileged caller authorization (ordinary LLM author/reviewer/merger/reconciler
+       roles are denied fail-closed; controller/admin/sysadmin or GITEA_BREAKGLASS_RESTART_AUTHORIZATION
+       is required).
+    2. Explicit non-empty reason (minimum 10 characters).
+    3. Exact confirmation string matching 'I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION'.
+    4. Mandatory impact acknowledgement (impact_ack=True).
+    5. Immutable audit entry recorded.
+    6. Automatic incident record created on Gitea.
+    7. Mandatory post-restart reconciliation requirement (#662).
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "break_glass_executed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+            "blocker_kind": "permission_denied",
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    profile = get_profile()
+    active_role = _profile_role_kind(profile)
+    break_glass_env_auth = bool(
+        (os.environ.get("GITEA_BREAKGLASS_RESTART_AUTHORIZATION") or "").strip()
+    )
+
+    # AC1: Ordinary LLM roles (author, reviewer, merger, reconciler) cannot invoke break-glass
+    # unless explicit environment break-glass authorization is configured.
+    if active_role in ("author", "reviewer", "merger", "reconciler") and not break_glass_env_auth:
+        return {
+            "success": False,
+            "performed": False,
+            "break_glass_executed": False,
+            "active_role": active_role,
+            "reasons": [
+                f"ordinary LLM role '{active_role}' is forbidden from break-glass restarts (#664 AC1); "
+                "privileged controller, operator, or GITEA_BREAKGLASS_RESTART_AUTHORIZATION required"
+            ],
+            "blocker_kind": "role_authorization",
+        }
+
+    # AC2: Required fields enforced
+    clean_reason = (reason or "").strip()
+    if not clean_reason or len(clean_reason) < 10:
+        return {
+            "success": False,
+            "performed": False,
+            "break_glass_executed": False,
+            "reasons": [
+                "reason is required and must be at least 10 characters long (#664 AC2)"
+            ],
+            "blocker_kind": "missing_required_fields",
+        }
+
+    clean_confirmation = (confirmation or "").strip()
+    if clean_confirmation != BREAK_GLASS_CONFIRMATION_PHRASE:
+        return {
+            "success": False,
+            "performed": False,
+            "break_glass_executed": False,
+            "reasons": [
+                f"confirmation string mismatch; must equal exactly '{BREAK_GLASS_CONFIRMATION_PHRASE}' (#664 AC2)"
+            ],
+            "blocker_kind": "confirmation_mismatch",
+        }
+
+    if not impact_ack:
+        return {
+            "success": False,
+            "performed": False,
+            "break_glass_executed": False,
+            "reasons": [
+                "impact_ack must be True to acknowledge disruption of in-flight sessions (#664 AC2)"
+            ],
+            "blocker_kind": "impact_ack_required",
+        }
+
+    # Evaluate impact / disrupted sessions
+    impact_result = gitea_request_mcp_restart(
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        dry_run=True,
+        restart_class=restart_class,
+        request_break_glass=True,
+    )
+    disrupted_sessions = list(impact_result.get("affected_sessions") or [])
+    disrupted_count = len(disrupted_sessions)
+
+    identity = _authenticated_username(h) or profile.get("username") or "unknown"
+    now_iso = datetime.now(timezone.utc).isoformat()
+
+    audit_payload = {
+        "event": "break_glass_mcp_restart",
+        "actor": identity,
+        "role": active_role,
+        "timestamp": now_iso,
+        "reason": clean_reason,
+        "confirmation": clean_confirmation,
+        "restart_class": restart_class,
+        "disrupted_sessions_count": disrupted_count,
+        "disrupted_sessions": [s.get("session_id") if isinstance(s, dict) else str(s) for s in disrupted_sessions],
+        "dry_run": dry_run,
+        "remote": remote,
+        "org": o,
+        "repo": r,
+    }
+
+    # Save immutable audit record
+    saved_audit = mcp_session_state.save_state(
+        kind="break_glass_audit",
+        payload=audit_payload,
+        remote=remote,
+        org=o,
+        repo=r,
+        profile_identity=profile.get("profile_name", "unknown"),
+    )
+
+    incident_issue_result = None
+    if create_incident_issue and not dry_run:
+        issue_title = f"[INCIDENT] Break-glass MCP restart invoked by {identity}"
+        issue_body = (
+            f"## Break-glass MCP restart incident report (#664)\n\n"
+            f"- **Invoked by**: `{identity}` (role: `{active_role}`)\n"
+            f"- **Timestamp**: `{now_iso}`\n"
+            f"- **Reason**: {clean_reason}\n"
+            f"- **Confirmation**: `{clean_confirmation}`\n"
+            f"- **Disrupted Sessions Count**: `{disrupted_count}`\n\n"
+            f"### Mandatory Post-Restart Reconciliation (#662)\n"
+            f"Post-restart reconciliation must be executed via `gitea_reconcile_after_restart` "
+            f"to clean up orphaned leases, inspect worktree integrity, and handle disrupted work.\n\n"
+            f"### Cross-references\n"
+            f"Ref #652 #653 #655 #630 #658 #662 #664\n"
+        )
+        try:
+            incident_issue_result = api_request(
+                "POST",
+                f"{repo_api_url(h, o, r)}/issues",
+                _auth(h),
+                {
+                    "title": issue_title,
+                    "body": issue_body,
+                    "labels": ["incident", "mcp-health", "break-glass"],
+                },
+            )
+        except Exception as exc:  # noqa: BLE001
+            incident_issue_result = {"error": _redact(str(exc))}
+
+    return {
+        "success": True,
+        "performed": not dry_run,
+        "dry_run": dry_run,
+        "break_glass_executed": not dry_run,
+        "would_execute": True,
+        "actor": identity,
+        "role": active_role,
+        "restart_class": restart_class,
+        "reason": clean_reason,
+        "confirmation": clean_confirmation,
+        "disrupted_sessions_count": disrupted_count,
+        "disrupted_sessions": disrupted_sessions,
+        "audit_record": audit_payload,
+        "saved_audit": dict(saved_audit or audit_payload),
+        "incident_issue": incident_issue_result,
+        "reconciliation_required": True,
+        "reconciliation_tool": "gitea_reconcile_after_restart",
+        "follow_up_issue_required": True,
+        "cross_references": ["#652", "#653", "#655", "#630", "#658", "#662", "#664"],
+        "reasons": [
+            "break-glass restart dry-run evaluated successfully" if dry_run
+            else "break-glass restart executed with incident creation and mandatory reconciliation"
+        ],
+    }
+
+
 # --- #662 post-restart reconciliation ---------------------------------------
 
 _POST_RESTART_LAST_PROOF: dict | None = None

task_capability_map.py

@@ -576,6 +576,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "runtime.record_analytics_usage",
         "role": "author",
     },
+    # #664: emergency break-glass MCP restart workflow (privileged controller role).
+    "break_glass_restart": {
+        "permission": "gitea.read",
+        "role": "controller",
+    },
+    "gitea_break_glass_restart": {
+        "permission": "gitea.read",
+        "role": "controller",
+    },
 }
 
 

tests/test_issue_664_break_glass_restart.py

@@ -0,0 +1,190 @@
+"""Tests for emergency break-glass MCP restart workflow (#664)."""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+import gitea_mcp_server
+
+
+class TestBreakGlassRestart(unittest.TestCase):
+    """Test suite for gitea_break_glass_restart tool and guardrails (#664)."""
+
+    def setUp(self) -> None:
+        self.env_patcher = patch.dict(os.environ, {}, clear=False)
+        self.env_patcher.start()
+
+    def tearDown(self) -> None:
+        self.env_patcher.stop()
+
+    def test_ordinary_role_denied_fail_closed(self) -> None:
+        """AC1: Ordinary LLM roles (author, reviewer, merger, reconciler) cannot invoke break-glass."""
+        with patch.object(
+            gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-author", "allowed_operations": ["gitea.read"]}
+        ), patch.object(
+            gitea_mcp_server, "_profile_role_kind", return_value="author"
+        ), patch.object(
+            gitea_mcp_server, "_profile_operation_gate", return_value=None
+        ):
+            res = gitea_mcp_server.gitea_break_glass_restart(
+                reason="Emergency restart needed due to deadlocked worker daemon processes",
+                confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
+                impact_ack=True,
+                remote="prgs",
+            )
+            self.assertFalse(res["success"])
+            self.assertFalse(res["break_glass_executed"])
+            self.assertEqual(res["blocker_kind"], "role_authorization")
+            self.assertIn("ordinary LLM role 'author' is forbidden", res["reasons"][0])
+
+    def test_short_reason_denied(self) -> None:
+        """AC2: Reason is required and must be at least 10 characters long."""
+        with patch.object(
+            gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]}
+        ), patch.object(
+            gitea_mcp_server, "_profile_role_kind", return_value="controller"
+        ), patch.object(
+            gitea_mcp_server, "_profile_operation_gate", return_value=None
+        ):
+            res = gitea_mcp_server.gitea_break_glass_restart(
+                reason="Too short",
+                confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
+                impact_ack=True,
+                remote="prgs",
+            )
+            self.assertFalse(res["success"])
+            self.assertEqual(res["blocker_kind"], "missing_required_fields")
+            self.assertIn("at least 10 characters", res["reasons"][0])
+
+    def test_confirmation_mismatch_denied(self) -> None:
+        """AC2: Confirmation phrase must match exact required string."""
+        with patch.object(
+            gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]}
+        ), patch.object(
+            gitea_mcp_server, "_profile_role_kind", return_value="controller"
+        ), patch.object(
+            gitea_mcp_server, "_profile_operation_gate", return_value=None
+        ):
+            res = gitea_mcp_server.gitea_break_glass_restart(
+                reason="Emergency restart needed due to stuck daemon processes",
+                confirmation="wrong_confirmation_phrase",
+                impact_ack=True,
+                remote="prgs",
+            )
+            self.assertFalse(res["success"])
+            self.assertEqual(res["blocker_kind"], "confirmation_mismatch")
+            self.assertIn("confirmation string mismatch", res["reasons"][0])
+
+    def test_impact_ack_required_denied(self) -> None:
+        """AC2: impact_ack=True is mandatory."""
+        with patch.object(
+            gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]}
+        ), patch.object(
+            gitea_mcp_server, "_profile_role_kind", return_value="controller"
+        ), patch.object(
+            gitea_mcp_server, "_profile_operation_gate", return_value=None
+        ):
+            res = gitea_mcp_server.gitea_break_glass_restart(
+                reason="Emergency restart needed due to stuck daemon processes",
+                confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
+                impact_ack=False,
+                remote="prgs",
+            )
+            self.assertFalse(res["success"])
+            self.assertEqual(res["blocker_kind"], "impact_ack_required")
+
+    def test_dry_run_evaluation(self) -> None:
+        """AC5: Dry-run evaluation returns preview without live execution or incident creation."""
+        with patch.object(
+            gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]}
+        ), patch.object(
+            gitea_mcp_server, "_profile_role_kind", return_value="controller"
+        ), patch.object(
+            gitea_mcp_server, "_profile_operation_gate", return_value=None
+        ), patch.object(
+            gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]}
+        ), patch.object(
+            gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
+        ):
+            res = gitea_mcp_server.gitea_break_glass_restart(
+                reason="Emergency restart needed due to hung worker process cohort",
+                confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
+                impact_ack=True,
+                dry_run=True,
+                remote="prgs",
+            )
+            self.assertTrue(res["success"])
+            self.assertTrue(res["dry_run"])
+            self.assertFalse(res["break_glass_executed"])
+            self.assertTrue(res["would_execute"])
+            self.assertTrue(res["reconciliation_required"])
+            self.assertEqual(res["reconciliation_tool"], "gitea_reconcile_after_restart")
+            self.assertIn("#664", res["cross_references"])
+
+    def test_privileged_execute_creates_incident_and_audit(self) -> None:
+        """AC3 & AC4: Execution creates incident issue, audit entry, and mandates post-restart reconcile."""
+        mock_api_request = MagicMock(return_value={"number": 999, "title": "[INCIDENT] Break-glass"})
+        mock_save_state = MagicMock(return_value={"saved": True})
+
+        with patch.object(
+            gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-controller", "allowed_operations": ["gitea.read"]}
+        ), patch.object(
+            gitea_mcp_server, "_profile_role_kind", return_value="controller"
+        ), patch.object(
+            gitea_mcp_server, "_profile_operation_gate", return_value=None
+        ), patch.object(
+            gitea_mcp_server, "_auth", return_value={"Authorization": "token test"}
+        ), patch.object(
+            gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": [{"session_id": "s1"}]}
+        ), patch.object(
+            gitea_mcp_server, "_authenticated_username", return_value="sysadmin"
+        ), patch.object(
+            gitea_mcp_server, "api_request", mock_api_request
+        ), patch("mcp_session_state.save_state", mock_save_state):
+
+            res = gitea_mcp_server.gitea_break_glass_restart(
+                reason="Emergency break-glass restart due to unrecoverable transport deadlock",
+                confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
+                impact_ack=True,
+                dry_run=False,
+                create_incident_issue=True,
+                remote="prgs",
+            )
+            self.assertTrue(res["success"])
+            self.assertFalse(res["dry_run"])
+            self.assertTrue(res["break_glass_executed"])
+            self.assertTrue(res["reconciliation_required"])
+            self.assertEqual(res["reconciliation_tool"], "gitea_reconcile_after_restart")
+            self.assertEqual(res["incident_issue"]["number"], 999)
+            mock_save_state.assert_called_once()
+            mock_api_request.assert_called_once()
+
+    def test_env_authorization_override_for_worker_role(self) -> None:
+        """Environment break-glass authorization enables privileged break-glass for configured sessions."""
+        os.environ["GITEA_BREAKGLASS_RESTART_AUTHORIZATION"] = "authorized-token"
+        with patch.object(
+            gitea_mcp_server, "get_profile", return_value={"profile_name": "prgs-author", "allowed_operations": ["gitea.read"]}
+        ), patch.object(
+            gitea_mcp_server, "_profile_role_kind", return_value="author"
+        ), patch.object(
+            gitea_mcp_server, "_profile_operation_gate", return_value=None
+        ), patch.object(
+            gitea_mcp_server, "gitea_request_mcp_restart", return_value={"affected_sessions": []}
+        ), patch.object(
+            gitea_mcp_server, "_authenticated_username", return_value="jcwalker3"
+        ):
+            res = gitea_mcp_server.gitea_break_glass_restart(
+                reason="Authorized emergency break-glass restart test",
+                confirmation="I_ACKNOWLEDGE_BREAK_GLASS_MCP_RESTART_DISRUPTION",
+                impact_ack=True,
+                dry_run=True,
+                remote="prgs",
+            )
+            self.assertTrue(res["success"])
+            self.assertTrue(res["dry_run"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_704_prevent_env_workspace_bindings.py

@@ -1,166 +0,0 @@
-"""Tests for Issue #704: Preventing repository .env files from injecting workspace bindings.
-
-Acceptance Criteria (#704):
-1. Repository .env loading cannot populate or override GITEA_ACTIVE_WORKTREE or any role-specific GITEA_*_WORKTREE runtime-binding variable.
-2. Runtime workspace bindings are accepted only from sanctioned managed-launch/session mechanisms.
-3. Pre-existing sanctioned process environment values retain their intended precedence.
-4. Importing gitea_auth or related modules does not mutate workspace-binding state from repository files.
-5. Reserved runtime-control keys found in .env are ignored or rejected with a sanitized actionable reason; their values are never logged.
-6. The protection applies consistently to author, reviewer, merger, and reconciler namespaces.
-7. Comprehensive test coverage for stale worktree, missing worktree, task-specific, role-specific, launcher binding, repeated imports, namespace isolation, precedence, and absence of secret leakage.
-8. Dirty-state and workspace-preflight gates cannot be bypassed by an injected missing-path binding.
-9. No environment, dotenv, offline-import, or caller-controlled path can forge native transport or mutation provenance.
-10. Cross-linked with #702, PR #703, #510.
-11. Required immediate follow-up to Issue #702 / PR #703.
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-import tempfile
-import importlib
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import gitea_auth
-from gitea_auth import is_reserved_worktree_env_key, load_env_file_sanitized
-
-
-class TestIssue704PreventEnvWorkspaceBindings(unittest.TestCase):
-    """Test suite verifying .env workspace-binding injection prevention (#704)."""
-
-    def setUp(self):
-        self.tmpdir = tempfile.TemporaryDirectory()
-        self.addCleanup(self.tmpdir.cleanup)
-        self.env_dir = Path(self.tmpdir.name)
-
-    def test_is_reserved_worktree_env_key(self):
-        """Verify key classification for all role namespaces (#704 AC6)."""
-        reserved_keys = [
-            "GITEA_ACTIVE_WORKTREE",
-            "GITEA_AUTHOR_WORKTREE",
-            "GITEA_REVIEWER_WORKTREE",
-            "GITEA_MERGER_WORKTREE",
-            "GITEA_RECONCILER_WORKTREE",
-            "gitea_active_worktree",
-            "GITEA_CUSTOM_ROLE_WORKTREE",
-        ]
-        for key in reserved_keys:
-            self.assertTrue(
-                is_reserved_worktree_env_key(key),
-                f"Expected {key} to be recognized as a reserved worktree key",
-            )
-
-        unreserved_keys = [
-            "GITEA_USER",
-            "GITEA_PASS",
-            "GITEA_TOKEN",
-            "GITEA_HOST",
-            "PATH",
-        ]
-        for key in unreserved_keys:
-            self.assertFalse(
-                is_reserved_worktree_env_key(key),
-                f"Expected {key} to NOT be recognized as a reserved worktree key",
-            )
-
-    def test_load_env_file_sanitized_ignores_reserved_keys(self):
-        """Verify .env loading ignores GITEA_ACTIVE_WORKTREE and role-specific keys (#704 AC1)."""
-        env_file = self.env_dir / ".env"
-        stale_path = "/tmp/stale-worktree-path-1234"
-        env_file.write_text(
-            f"GITEA_USER=testuser\n"
-            f"GITEA_ACTIVE_WORKTREE={stale_path}\n"
-            f"GITEA_AUTHOR_WORKTREE={stale_path}\n"
-            f"GITEA_REVIEWER_WORKTREE={stale_path}\n"
-            f"GITEA_MERGER_WORKTREE={stale_path}\n"
-            f"GITEA_RECONCILER_WORKTREE={stale_path}\n"
-        )
-
-        test_env = {}
-        reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        # Unreserved key loaded
-        self.assertEqual(test_env.get("GITEA_USER"), "testuser")
-
-        # Reserved keys ignored
-        self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
-        self.assertNotIn("GITEA_AUTHOR_WORKTREE", test_env)
-        self.assertNotIn("GITEA_REVIEWER_WORKTREE", test_env)
-        self.assertNotIn("GITEA_MERGER_WORKTREE", test_env)
-        self.assertNotIn("GITEA_RECONCILER_WORKTREE", test_env)
-
-        # Rejection reasons populated without leaking the secret value (#704 AC5)
-        self.assertTrue(len(reasons) >= 5)
-        for r in reasons:
-            self.assertNotIn(stale_path, r, "Secret/path value must not leak into rejection reason")
-
-    def test_preexisting_sanctioned_launcher_env_retained(self):
-        """Sanctioned launcher values in process env are retained (#704 AC2, AC3)."""
-        sanctioned_path = "/tmp/sanctioned-launcher-worktree"
-        test_env = {"GITEA_ACTIVE_WORKTREE": sanctioned_path}
-
-        env_file = self.env_dir / ".env"
-        env_file.write_text("GITEA_ACTIVE_WORKTREE=/tmp/injected-repo-worktree\n")
-
-        load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        # Pre-existing value retained, not overwritten by .env
-        self.assertEqual(test_env.get("GITEA_ACTIVE_WORKTREE"), sanctioned_path)
-
-    def test_stale_or_missing_worktree_in_env_ignored(self):
-        """Stale or non-existent worktree path in .env file is ignored (#704 AC7)."""
-        nonexistent_path = "/nonexistent/branches/stale-issue-999"
-        env_file = self.env_dir / ".env"
-        env_file.write_text(f"GITEA_ACTIVE_WORKTREE={nonexistent_path}\n")
-
-        test_env = {}
-        load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
-
-    def test_repeated_module_import_does_not_mutate_workspace_env(self):
-        """Repeated imports of gitea_auth leave os.environ un-contaminated (#704 AC4, AC7)."""
-        # Ensure no active worktree env exists initially
-        original_val = os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
-        try:
-            importlib.reload(gitea_auth)
-            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
-
-            importlib.reload(gitea_auth)
-            self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
-        finally:
-            if original_val is not None:
-                os.environ["GITEA_ACTIVE_WORKTREE"] = original_val
-
-    def test_namespace_isolation_all_roles_protected(self):
-        """Verify protection across author, reviewer, merger, reconciler (#704 AC6)."""
-        env_file = self.env_dir / ".env"
-        env_file.write_text(
-            "GITEA_AUTHOR_WORKTREE=/bad/author\n"
-            "GITEA_REVIEWER_WORKTREE=/bad/reviewer\n"
-            "GITEA_MERGER_WORKTREE=/bad/merger\n"
-            "GITEA_RECONCILER_WORKTREE=/bad/reconciler\n"
-        )
-        test_env = {}
-        load_env_file_sanitized(str(env_file), target_env=test_env)
-
-        self.assertEqual(test_env, {})
-
-    def test_absence_of_secret_leakage(self):
-        """Rejection reasons contain key names but never secret path values (#704 AC5)."""
-        sensitive_path = "/Users/secret/path/private_repo"
-        env_file = self.env_dir / ".env"
-        env_file.write_text(f"GITEA_ACTIVE_WORKTREE={sensitive_path}\n")
-
-        test_env = {}
-        reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
-        for reason in reasons:
-            self.assertNotIn(sensitive_path, reason)
-
-
-if __name__ == "__main__":
-    unittest.main()