From c2fc2683b97e3ba5dc7eaf4ac9798af76bd08dfc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 13 Jul 2026 16:40:58 -0400 Subject: [PATCH] fix(recovery): address PR #703 review findings F1-F6 (#702) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes the six defects from the formal REQUEST_CHANGES review at 889931d5532b32a46a181e03c1345946c7516aac: - F1: run sanctioned stale-binding recovery in gitea_resolve_task_capability BEFORE the terminal launcher probe, so a worktree removed mid-session can no longer wedge mutation resolution on 'missing cwd' before recovery runs. The fail-closed probe block now also carries the binding classification evidence. - F2: _resolve_preflight_workspace_path resolves with the same verify_paths existence checks as the canonical mutation context, and _get_workspace_porcelain returns a synthetic tracked-dirty sentinel when the workspace is missing or git fails — an uninspectable workspace can never read as clean/empty porcelain. - F3: the orphaned_expired_superseded_head cleanup matrix now requires affirmative safe worktree evidence (worktree_clean=true or worktree_exists=false), aligned with the sibling orphaned_owner_missing gate and the diagnosis path; unknown evidence fails closed. - F4: reviewer-session-lease shadows and stale-binding audit records are recovery-critical kinds exempt from the 4h session-state TTL; they persist until a sanctioned clear terminally reconciles them. - F5: session-lease shadows are keyed by lease session id (collision-safe identity) with a list_states enumeration API; concurrent same-profile sessions can no longer overwrite or misattribute each other's crash evidence. Legacy single-slot records remain readable. - F6: the durable audit record is persisted (status=pending) BEFORE the environment binding is cleared; if audit persistence fails the clear does not happen and the failure is reported explicitly. 30 new regression tests cover deleted-worktree recovery ordering, missing/ deleted/symlinked/mid-evaluation path changes, unknown-evidence cleanup, TTL boundary (before/at/after), interleaved and reconnected concurrent sessions, and audit-write failure/retry/idempotence/ordering. Issue #704 dotenv load-path prevention is intentionally NOT included. Full suite: 2695 passed, 6 skipped, 161 subtests passed. Co-Authored-By: Claude Fable 5 --- gitea_mcp_server.py | 138 +++- mcp_session_state.py | 97 ++- reviewer_pr_lease.py | 63 +- tests/test_issue_702_review_findings_f1_f6.py | 693 ++++++++++++++++++ 4 files changed, 947 insertions(+), 44 deletions(-) create mode 100644 tests/test_issue_702_review_findings_f1_f6.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 2b6a6d7..236900e 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -298,23 +298,61 @@ def _assess_stale_active_binding(auto_recover: bool = False) -> dict: not _preflight_in_test_mode() or os.environ.get("GITEA_FORCE_STALE_BINDING_RECOVERY") == "1" ) + audit_persisted: bool | None = None if auto_recover and plan.get("clear_allowed") and recovery_enabled: - applied = stale_binding_recovery.apply_recovery(plan) - if applied.get("performed"): - try: - import mcp_session_state as mss + # #702 F6: the durable audit record must exist BEFORE the environment + # is cleared. If audit persistence fails the recovery does not run — + # an env clear without durable proof is an unauditable mutation. + pending_value = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip() or None + audit_error: str | None = None + try: + import mcp_session_state as mss - mss.save_state( - kind=mss.KIND_STALE_BINDING_RECOVERY, - payload={ - "cleared_env": applied.get("cleared_env"), - "cleared_value": applied.get("cleared_value"), - "classification": applied.get("classification"), - "reasons": applied.get("reasons"), - }, - ) - except Exception: - pass + mss.save_state( + kind=mss.KIND_STALE_BINDING_RECOVERY, + payload={ + "cleared_env": ACTIVE_WORKTREE_ENV, + "cleared_value": pending_value, + "classification": plan.get("classification"), + "reasons": list(plan.get("reasons") or []), + "recovery_status": "pending", + }, + ) + audit_persisted = True + except Exception as exc: + audit_persisted = False + audit_error = f"{type(exc).__name__}: {exc}" + if audit_persisted: + applied = stale_binding_recovery.apply_recovery(plan) + if applied.get("performed"): + try: + import mcp_session_state as mss + + mss.save_state( + kind=mss.KIND_STALE_BINDING_RECOVERY, + payload={ + "cleared_env": applied.get("cleared_env"), + "cleared_value": applied.get("cleared_value"), + "classification": applied.get("classification"), + "reasons": applied.get("reasons"), + "recovery_status": "performed", + }, + ) + except Exception: + # The pre-clear "pending" record is already durable; the + # status refresh is best-effort and never rolls back. + pass + else: + applied = { + "performed": False, + "action": stale_binding_recovery.RECOVERY_ACTION_NONE, + "audit_write_failed": True, + "reasons": [ + "audit persistence failed; stale binding NOT cleared " + "(fail closed, #702 F6)" + ] + + ([audit_error] if audit_error else []), + } report = { "classification": classification.get("classification"), "active_worktree": classification.get("active_worktree"), @@ -324,6 +362,13 @@ def _assess_stale_active_binding(auto_recover: bool = False) -> dict: "recovery_performed": bool(applied.get("performed")), "reasons": classification.get("reasons") or [], } + if audit_persisted is not None: + report["audit_persisted"] = audit_persisted + if applied.get("audit_write_failed"): + report["audit_write_failed"] = True + report["reasons"] = list(report["reasons"]) + list( + applied.get("reasons") or [] + ) if applied.get("performed"): report["cleared_value"] = applied.get("cleared_value") if plan.get("exact_next_action"): @@ -332,7 +377,13 @@ def _assess_stale_active_binding(auto_recover: bool = False) -> dict: def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: - """Resolve the namespace-scoped workspace root inspected by pre-flight guards.""" + """Resolve the namespace-scoped workspace root inspected by pre-flight guards. + + #702 F2: preflight resolves with the same existence verification as the + 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. + """ role = _effective_workspace_role() workspace, _source = nwb.resolve_namespace_workspace( role_kind=role, @@ -342,6 +393,7 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str: _reviewer_session_worktree() if role in {"reviewer", "merger"} else None ), profile_name=get_profile().get("profile_name"), + verify_paths=True, ) return workspace @@ -388,6 +440,15 @@ def _get_git_root(path: str) -> str | None: return (res.stdout or "").strip() or None +# #702 F2: synthetic tracked-dirty porcelain returned when the preflight +# workspace cannot be inspected (missing path, git failure). Shaped as a +# modified tracked entry so every consumer treats "workspace unavailable" +# as dirty — an uninspectable workspace must never read as clean. +PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN = ( + " M __gitea_preflight_workspace_unavailable__\n" +) + + def _get_workspace_porcelain(worktree_path: str | None = None) -> str: """Return tracked-workspace porcelain for pre-flight attribution.""" if os.environ.get("GITEA_TEST_FORCE_DIRTY"): @@ -396,6 +457,8 @@ def _get_workspace_porcelain(worktree_path: str | None = None) -> str: if override is not None: return override workspace = _resolve_preflight_workspace_path(worktree_path) + if not os.path.isdir(workspace): + return PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN try: res = subprocess.run( ["git", "status", "--porcelain"], @@ -403,9 +466,11 @@ def _get_workspace_porcelain(worktree_path: str | None = None) -> str: text=True, cwd=workspace, ) - return res.stdout or "" except Exception: - return "" + return PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN + if res.returncode != 0: + return PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN + return res.stdout or "" def _parse_porcelain_entries(porcelain: str) -> dict[str, str]: @@ -8127,7 +8192,10 @@ def gitea_diagnose_reviewer_pr_lease_handoff( # shadow left behind by a crashed daemon. Only a provably dead recorded # owner pid yields False; PID liveness alone never proves ownership. orphan_evidence = reviewer_pr_lease.assess_crashed_session_lease_orphan( - reviewer_pr_lease.load_session_lease_shadow(), lease=active + reviewer_pr_lease.load_session_lease_shadow( + session_id=(active or {}).get("session_id") + ), + lease=active, ) diagnosis = reviewer_pr_lease.diagnose_reviewer_pr_lease_handoff( @@ -8379,7 +8447,10 @@ def gitea_cleanup_obsolete_reviewer_comment_lease( # session-lease shadow; an explicitly passed value always wins. if owner_process_alive is None: owner_process_alive = reviewer_pr_lease.assess_crashed_session_lease_orphan( - reviewer_pr_lease.load_session_lease_shadow(), lease=lease + reviewer_pr_lease.load_session_lease_shadow( + session_id=(lease or {}).get("session_id") + ), + lease=lease, ).get("owner_process_alive") assessment = reviewer_pr_lease.assess_obsolete_reviewer_comment_lease_cleanup( comments, @@ -11674,6 +11745,18 @@ def gitea_resolve_task_capability( "exact_safe_next_action": next_safe_action, } + # #702 F1: validate (and, when provably stale, recover) the inherited + # GITEA_ACTIVE_WORKTREE binding BEFORE the terminal launcher probe. The + # probe uses that binding as its cwd; a worktree removed mid-session + # would otherwise wedge every mutation-task resolution on 'missing cwd' + # before the sanctioned recovery could ever run. Recovery stays + # fail-closed: only the provably-stale classes clear, with the durable + # audit record persisted first (F6). + try: + stale_binding = _assess_stale_active_binding(auto_recover=True) + except Exception: + stale_binding = None + # ── Terminal Launcher Preflight Check (#556) ── if task in native_mcp_preference.GITEA_MUTATION_TASKS: in_test = _preflight_in_test_mode() @@ -11697,7 +11780,7 @@ def gitea_resolve_task_capability( "Do not attempt git/pytest finalization or unsafe fallbacks. " "Run gitea_diagnose_terminal, emit blocked-diagnose-report, and stop." ) - return { + blocked_result = { "requested_task": task, "required_operation_permission": required_permission, "required_role_kind": required_role, @@ -11717,6 +11800,11 @@ def gitea_resolve_task_capability( "different_mcp_namespace_required": False, "exact_safe_next_action": next_safe_action, } + if stale_binding is not None and stale_binding.get( + "classification" + ) != stale_binding_recovery.CLASSIFICATION_UNBOUND: + blocked_result["stale_binding_recovery"] = stale_binding + return blocked_result record_preflight_check("capability", required_role, resolved_task=task) @@ -11886,14 +11974,6 @@ def gitea_resolve_task_capability( record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task) - # #702: validate the inherited GITEA_ACTIVE_WORKTREE binding during - # capability resolution; provably-stale bindings are cleared by the - # sanctioned in-daemon mechanism with a durable audit record. - try: - stale_binding = _assess_stale_active_binding(auto_recover=True) - except Exception: - stale_binding = None - result = { "requested_task": task, "required_operation_permission": required_permission, diff --git a/mcp_session_state.py b/mcp_session_state.py index 1698189..330cf42 100644 --- a/mcp_session_state.py +++ b/mcp_session_state.py @@ -46,6 +46,16 @@ KIND_REVIEWER_SESSION_LEASE = "reviewer_session_lease" # recovery clears an inherited GITEA_ACTIVE_WORKTREE binding (#702). KIND_STALE_BINDING_RECOVERY = "stale_binding_recovery" +# Kinds that carry recovery-critical crash-orphan evidence (#702). They are +# exempt from TTL expiry: a crashed daemon can never refresh its own record, +# so a wall-clock TTL would silently destroy the only owner-exit proof the +# guarded cleanup path accepts. These records live until a sanctioned clear +# terminally reconciles them (durable lifecycle), never until a timer fires. +RECOVERY_CRITICAL_KINDS = frozenset({ + KIND_REVIEWER_SESSION_LEASE, + KIND_STALE_BINDING_RECOVERY, +}) + _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+") @@ -95,6 +105,7 @@ def state_key( org: str | None = None, repo: str | None = None, profile_identity: str | None = None, + instance_id: str | None = None, ) -> str: """Build durable filename key. @@ -102,17 +113,20 @@ def state_key( so the primary key is kind + profile identity. Remote/org/repo are stored inside the payload and validated on load (#559), which lets a later daemon process recover state without already knowing the remote argument. + + *instance_id* extends the key for kinds where one-per-profile collides — + concurrent same-profile sessions each own their reviewer-lease shadow + (#702 F5), keyed by their lease session id so a later heartbeat can never + overwrite or misattribute another session's crash evidence. """ # Keep remote/org/repo parameters for API stability / future kinds; they are # intentionally not part of the filename for session-scoped proofs. _ = (remote, org, repo) - return "-".join( - _sanitize_segment(part) - for part in ( - kind, - profile_identity or "unknown-profile", - ) - ) + parts = [kind, profile_identity or "unknown-profile"] + instance = (instance_id or "").strip() + if instance: + parts.append(instance) + return "-".join(_sanitize_segment(part) for part in parts) def state_file_path( @@ -123,6 +137,7 @@ def state_file_path( repo: str | None = None, profile_identity: str | None = None, state_dir: str | None = None, + instance_id: str | None = None, ) -> str: root = (state_dir or default_state_dir()).strip() name = state_key( @@ -131,6 +146,7 @@ def state_file_path( org=org, repo=repo, profile_identity=profile_identity, + instance_id=instance_id, ) return os.path.join(root, f"{name}.json") @@ -251,10 +267,12 @@ def identity_match_reasons( reasons.append("session state missing recorded_at timestamp (fail closed)") else: age = _now_utc() - recorded_at + record_kind = (str(record.get("kind") or "")).strip() if age > timedelta(hours=ttl_hours()): - reasons.append( - f"session state expired after {ttl_hours():g}h (fail closed)" - ) + if record_kind not in RECOVERY_CRITICAL_KINDS: + reasons.append( + f"session state expired after {ttl_hours():g}h (fail closed)" + ) if age < timedelta(0): reasons.append("session state recorded_at is in the future (fail closed)") return reasons @@ -268,6 +286,7 @@ def load_state( repo: str | None = None, profile_identity: str | None = None, state_dir: str | None = None, + instance_id: str | None = None, ) -> dict[str, Any] | None: """Load durable state payload when identity checks pass.""" profile = current_profile_identity(profile_identity=profile_identity) @@ -278,6 +297,7 @@ def load_state( repo=repo, profile_identity=profile, state_dir=state_dir, + instance_id=instance_id, ) lock_path = f"{path}.lock" with _exclusive_file_lock(lock_path): @@ -323,6 +343,7 @@ def save_state( repo: str | None = None, profile_identity: str | None = None, state_dir: str | None = None, + instance_id: str | None = None, ) -> dict[str, Any] | None: """Persist or clear durable state for the given session identity key.""" profile = current_profile_identity( @@ -344,6 +365,7 @@ def save_state( repo=key_repo, profile_identity=profile, state_dir=root, + instance_id=instance_id, ) lock_path = f"{path}.lock" with _exclusive_file_lock(lock_path): @@ -365,6 +387,8 @@ def save_state( body["session_profile_lock"] = profile body["recorded_at"] = body.get("recorded_at") or now body["updated_at"] = now + if (instance_id or "").strip(): + body["instance_id"] = instance_id.strip() if key_remote is not None: body["remote"] = key_remote if key_org is not None: @@ -396,6 +420,7 @@ def clear_state( repo: str | None = None, profile_identity: str | None = None, state_dir: str | None = None, + instance_id: str | None = None, ) -> None: save_state( kind=kind, @@ -405,4 +430,56 @@ def clear_state( repo=repo, profile_identity=profile_identity, state_dir=state_dir, + instance_id=instance_id, ) + + +def list_states( + *, + kind: str, + profile_identity: str | None = None, + state_dir: str | None = None, +) -> list[dict[str, Any]]: + """Enumerate durable records of *kind* across all instance keys. + + Crash recovery cannot know which session id left a shadow behind, so it + must be able to enumerate every record of a kind (#702 F5). Identity + checks (profile / TTL policy) apply per record exactly as in + :func:`load_state`; records that fail closed are omitted. + """ + root = (state_dir or default_state_dir()).strip() + if not root or not os.path.isdir(root): + return [] + profile = current_profile_identity(profile_identity=profile_identity) + results: list[dict[str, Any]] = [] + try: + names = sorted(os.listdir(root)) + except OSError: + return [] + for name in names: + if not name.endswith(".json") or name.startswith("."): + continue + envelope = _read_json(os.path.join(root, name)) + if not envelope or envelope.get("kind") != kind: + continue + payload = envelope.get("payload") + if not isinstance(payload, dict): + continue + merged = dict(payload) + for key in ( + "kind", + "remote", + "org", + "repo", + "profile_identity", + "session_profile_lock", + "recorded_at", + "updated_at", + "writer_pid", + ): + if key in envelope and key not in merged: + merged[key] = envelope[key] + if identity_match_reasons(merged, profile_identity=profile): + continue + results.append(merged) + return results diff --git a/reviewer_pr_lease.py b/reviewer_pr_lease.py index 5cdf004..796a4cb 100644 --- a/reviewer_pr_lease.py +++ b/reviewer_pr_lease.py @@ -413,30 +413,48 @@ def record_session_lease( def clear_session_lease() -> None: global _SESSION_LEASE + prior = _SESSION_LEASE _SESSION_LEASE = None - _persist_session_lease_shadow(None) + _persist_session_lease_shadow(None, prior=prior) def get_session_lease() -> dict[str, Any] | None: return dict(_SESSION_LEASE) if _SESSION_LEASE else None -def _persist_session_lease_shadow(lease: dict[str, Any] | None) -> None: +def _persist_session_lease_shadow( + lease: dict[str, Any] | None, + *, + prior: dict[str, Any] | None = None, +) -> None: """Best-effort durable shadow of the in-session lease (#702). A daemon that dies without teardown leaves this record behind, giving a later process provable orphan evidence (owner pid + session id) for the guarded cleanup path. Observability only — mutation gates never read it, and failures here must never block lease operations. + + Shadows are keyed by lease session id (#702 F5): concurrent same-profile + sessions each own a distinct record, so one session's heartbeat can never + overwrite or misattribute another's crash evidence. A sanctioned clear is + the terminal reconciliation of that record's lifecycle. """ try: import mcp_session_state as mss if lease is None: + sid = ((prior or {}).get("session_id") or "").strip() or None + if sid: + mss.clear_state( + kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=sid + ) + # Legacy single-slot record from pre-F5 code: clearing it is safe + # because collision-safe writes never target that key again. mss.clear_state(kind=mss.KIND_REVIEWER_SESSION_LEASE) return mss.save_state( kind=mss.KIND_REVIEWER_SESSION_LEASE, + instance_id=((lease.get("session_id") or "").strip() or None), payload={ "pr_number": lease.get("pr_number"), "session_id": lease.get("session_id"), @@ -454,12 +472,38 @@ def _persist_session_lease_shadow(lease: dict[str, Any] | None) -> None: pass -def load_session_lease_shadow() -> dict[str, Any] | None: - """Load the durable session-lease shadow left by this profile identity.""" +def load_session_lease_shadow( + session_id: str | None = None, +) -> dict[str, Any] | None: + """Load the durable session-lease shadow left by this profile identity. + + With *session_id*, load that session's collision-safe record (#702 F5), + falling back to the legacy single-slot record only when its payload names + the same session. Without *session_id*, return the legacy record or the + sole surviving instance record — ambiguity (multiple candidates) returns + ``None`` because a shadow that cannot be attributed proves nothing. + """ try: import mcp_session_state as mss - return mss.load_state(kind=mss.KIND_REVIEWER_SESSION_LEASE) + sid = (session_id or "").strip() + if sid: + record = mss.load_state( + kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=sid + ) + if record is not None: + return record + legacy = mss.load_state(kind=mss.KIND_REVIEWER_SESSION_LEASE) + if legacy and (legacy.get("session_id") or "").strip() == sid: + return legacy + return None + legacy = mss.load_state(kind=mss.KIND_REVIEWER_SESSION_LEASE) + if legacy is not None: + return legacy + records = mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE) + if len(records) == 1: + return records[0] + return None except Exception: return None @@ -1006,6 +1050,15 @@ def assess_obsolete_reviewer_comment_lease_cleanup( fail_closed_reasons.append( "owner process absent but worktree dirty; cleanup denied" ) + elif not (worktree_clean is True or worktree_exists is False): + # #702 F3: unknown worktree evidence must fail closed, exactly + # like the sibling orphaned_owner_missing gate and the + # diagnosis path — cleanup needs affirmative safe evidence. + fail_closed_reasons.append( + "owner process absent but worktree evidence is unknown; " + "cleanup requires affirmative safe evidence " + "(worktree_clean=true or worktree_exists=false)" + ) elif head_superseded and not has_terminal: classification = "ambiguous_conflicting_evidence" fail_closed_reasons.append( diff --git a/tests/test_issue_702_review_findings_f1_f6.py b/tests/test_issue_702_review_findings_f1_f6.py new file mode 100644 index 0000000..3ab3565 --- /dev/null +++ b/tests/test_issue_702_review_findings_f1_f6.py @@ -0,0 +1,693 @@ +"""Regression tests for the PR #703 REQUEST_CHANGES findings F1–F6 (#702). + +Formal review at head 889931d independently reproduced six defects: + +F1 — stale-binding recovery ran only AFTER the terminal launcher probe in + ``gitea_resolve_task_capability``; a worktree removed mid-session wedged + every mutation-task resolution on 'missing cwd' before recovery. +F2 — ``_resolve_preflight_workspace_path`` skipped the existence checks the + canonical mutation context applies; a dead binding could produce empty + porcelain and read as a clean workspace. +F3 — ``orphaned_expired_superseded_head`` cleanup accepted unknown + ``worktree_exists``/``worktree_clean`` evidence. +F4 — the 4-hour session-state TTL silently discarded recovery-critical + crash-orphan shadow evidence. +F5 — the single same-profile shadow slot let concurrent sessions overwrite + each other's crash evidence. +F6 — the environment was cleared BEFORE the audit record was persisted, and + audit-write failure was swallowed. +""" + +from __future__ import annotations + +import json +import os +import shutil +import sys +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import mcp_session_state as mss # noqa: E402 +import namespace_workspace_binding as nwb # noqa: E402 +import reviewer_pr_lease as leases # noqa: E402 +import stale_binding_recovery as sbr # noqa: E402 +import mcp_server # noqa: E402 + +REPO = "Scaled-Tech-Consulting/Gitea-Tools" +OLD_HEAD = "b" * 40 +NEW_HEAD = "6" * 40 + +CONFIG = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, + } + }, + "profiles": { + "prgs-author": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "jcwalker3", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": [ + "gitea.read", "gitea.issue.create", "gitea.issue.comment", + "gitea.pr.create", "gitea.branch.push", + ], + "forbidden_operations": [ + "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", + ], + "execution_profile": "prgs-author", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class _CapabilityHarness(unittest.TestCase): + """Shared config/env plumbing for resolve-capability integration tests.""" + + def setUp(self): + self._remotes = patch.dict(mcp_server.REMOTES, { + "prgs": { + "host": "gitea.example.com", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + }, + }) + self._remotes.start() + mcp_server._IDENTITY_CACHE.clear() + self._dir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self._dir.name, "profiles.json") + with open(self.config_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(CONFIG)) + + def tearDown(self): + self._remotes.stop() + mcp_server._IDENTITY_CACHE.clear() + self._dir.cleanup() + + def _env(self, **extra): + env = { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": "prgs-author", + "GITEA_TOKEN_AUTHOR": "author-pass", + } + env.update(extra) + return env + + def _deleted_worktree(self) -> str: + path = tempfile.mkdtemp(prefix="gitea-removed-worktree-") + shutil.rmtree(path) + return path + + +class TestF1RecoveryBeforeTerminalProbe(_CapabilityHarness): + """F1: recovery must run before the terminal cwd probe can wedge.""" + + @patch("mcp_server.api_request", return_value={"login": "jcwalker3"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_removed_worktree_recovers_before_probe(self, _auth, _api): + missing = self._deleted_worktree() + env = self._env( + GITEA_ACTIVE_WORKTREE=missing, + GITEA_FORCE_TERMINAL_PROBE="1", + GITEA_FORCE_STALE_BINDING_RECOVERY="1", + ) + with patch.dict(os.environ, env): + res = mcp_server.gitea_resolve_task_capability( + task="create_pr", remote="prgs" + ) + # The provably-stale binding was cleared before the probe ran. + self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ) + self.assertFalse(res.get("terminal_launcher_unhealthy")) + report = res.get("stale_binding_recovery") or {} + self.assertEqual( + report.get("classification"), + sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH, + ) + self.assertTrue(report.get("recovery_performed")) + self.assertTrue(res.get("allowed_in_current_session")) + + @patch("mcp_server.api_request", return_value={"login": "jcwalker3"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_probe_block_still_carries_recovery_report(self, _auth, _api): + # Recovery disabled (test mode without the force flag) preserves the + # fail-closed probe block, but the block must now carry the + # classification evidence instead of hiding it. + missing = self._deleted_worktree() + env = self._env( + GITEA_ACTIVE_WORKTREE=missing, + GITEA_FORCE_TERMINAL_PROBE="1", + ) + with patch.dict(os.environ, env): + res = mcp_server.gitea_resolve_task_capability( + task="create_pr", remote="prgs" + ) + self.assertIn("GITEA_ACTIVE_WORKTREE", os.environ) + self.assertTrue(res.get("terminal_launcher_unhealthy")) + report = res.get("stale_binding_recovery") or {} + self.assertEqual( + report.get("classification"), + sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH, + ) + self.assertFalse(report.get("recovery_performed")) + + @patch("mcp_server.api_request", return_value={"login": "jcwalker3"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_unverified_binding_is_never_cleared_by_reordering(self, _auth, _api): + # F1 must not weaken authorization: an existing, uncorroborated + # binding survives resolution untouched even with recovery forced. + existing = tempfile.mkdtemp(prefix="gitea-existing-worktree-") + self.addCleanup(shutil.rmtree, existing, ignore_errors=True) + env = self._env( + GITEA_ACTIVE_WORKTREE=existing, + GITEA_FORCE_TERMINAL_PROBE="1", + GITEA_FORCE_STALE_BINDING_RECOVERY="1", + ) + with patch.dict(os.environ, env): + res = mcp_server.gitea_resolve_task_capability( + task="create_pr", remote="prgs" + ) + self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), existing) + report = res.get("stale_binding_recovery") or {} + self.assertFalse(report.get("recovery_performed")) + + +class TestF2PreflightPathVerification(_CapabilityHarness): + """F2: preflight and mutation contexts verify resolved paths alike.""" + + def test_preflight_and_mutation_context_agree_on_dead_binding(self): + missing = self._deleted_worktree() + env = self._env(GITEA_ACTIVE_WORKTREE=missing) + with patch.dict(os.environ, env): + preflight = mcp_server._resolve_preflight_workspace_path(None) + mutation = mcp_server._resolve_namespace_mutation_context(None) + self.assertEqual(preflight, mutation["workspace_path"]) + self.assertNotEqual(preflight, os.path.realpath(missing)) + + def test_missing_explicit_worktree_never_reads_clean(self): + missing = self._deleted_worktree() + with patch.dict(os.environ, self._env()): + porcelain = mcp_server._get_workspace_porcelain(worktree_path=missing) + self.assertEqual( + porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN + ) + # The sentinel parses as a tracked dirty entry — never clean. + self.assertTrue(mcp_server._parse_porcelain_entries(porcelain)) + + def test_workspace_deleted_during_evaluation_never_reads_clean(self): + # TOCTOU: resolution succeeded, then the path vanished before git ran. + missing = self._deleted_worktree() + with patch.dict(os.environ, self._env()): + with patch.object( + mcp_server, + "_resolve_preflight_workspace_path", + return_value=missing, + ): + porcelain = mcp_server._get_workspace_porcelain() + self.assertEqual( + porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN + ) + + def test_git_failure_never_reads_clean(self): + class _Failed: + returncode = 128 + stdout = "" + stderr = "fatal: not a git repository" + + with patch.dict(os.environ, self._env()): + with patch.object( + mcp_server.subprocess, "run", return_value=_Failed() + ): + porcelain = mcp_server._get_workspace_porcelain() + self.assertEqual( + porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN + ) + + def test_symlinked_binding_to_deleted_target_is_demoted(self): + target = tempfile.mkdtemp(prefix="gitea-symlink-target-") + holder = tempfile.mkdtemp(prefix="gitea-symlink-holder-") + self.addCleanup(shutil.rmtree, holder, ignore_errors=True) + link = os.path.join(holder, "worktree-link") + os.symlink(target, link) + shutil.rmtree(target) + demotions: list[str] = [] + _path, source = nwb.resolve_namespace_workspace( + role_kind="reviewer", + process_project_root=os.getcwd(), + env={"GITEA_ACTIVE_WORKTREE": link}, + demotions=demotions, + verify_paths=True, + ) + self.assertEqual(source, "MCP server process root (default)") + self.assertEqual(len(demotions), 1) + + def test_dead_binding_falls_through_to_live_role_binding(self): + # Path changes during evaluation: the dead candidate demotes at + # verification time and never resurrects over the live fallback. + alive = tempfile.mkdtemp(prefix="gitea-alive-worktree-") + self.addCleanup(shutil.rmtree, alive, ignore_errors=True) + missing = self._deleted_worktree() + path, source = nwb.resolve_namespace_workspace( + role_kind="reviewer", + process_project_root=os.getcwd(), + env={ + "GITEA_ACTIVE_WORKTREE": missing, + "GITEA_REVIEWER_WORKTREE": alive, + }, + verify_paths=True, + ) + self.assertEqual(path, os.path.realpath(alive)) + self.assertEqual( + source, "GITEA_REVIEWER_WORKTREE environment variable" + ) + + +class TestF3AffirmativeWorktreeEvidence(unittest.TestCase): + """F3: unknown worktree evidence must fail closed for crash orphans.""" + + def _comments(self) -> list[dict]: + stale = _now() - timedelta(hours=3) + body = leases.format_lease_body( + repo=REPO, + pr_number=701, + issue_number=699, + reviewer_identity="sysadmin", + profile="prgs-reviewer", + session_id="65664-4f248d213d60", + worktree="branches/review-pr-654", + phase="validation-start", + candidate_head=OLD_HEAD, + target_branch="master", + target_branch_sha="2" * 40, + last_activity=stale, + expires_at=stale + timedelta(minutes=120), + ) + return [{ + "id": 11131, + "body": body, + "user": {"login": "sysadmin"}, + "author": "sysadmin", + "created_at": stale.isoformat(), + "updated_at": stale.isoformat(), + }] + + def _assess(self, **kwargs): + defaults = dict( + pr_number=701, + current_head_sha=NEW_HEAD, + formal_reviews=[], + repo=REPO, + expected_repo=REPO, + requesting_session_id="fresh-controller", + controller_recovery_authorized=True, + owner_process_alive=False, + ) + defaults.update(kwargs) + return leases.assess_obsolete_reviewer_comment_lease_cleanup( + self._comments(), **defaults + ) + + def test_unknown_worktree_evidence_fails_closed(self): + result = self._assess(worktree_exists=None, worktree_clean=None) + self.assertEqual( + result["classification"], "orphaned_expired_superseded_head" + ) + self.assertFalse(result["cleanup_allowed"]) + self.assertIn( + "affirmative safe evidence", + " ".join(result["fail_closed_reasons"]), + ) + + def test_unknown_cleanliness_with_existing_worktree_fails_closed(self): + result = self._assess(worktree_exists=True, worktree_clean=None) + self.assertFalse(result["cleanup_allowed"]) + + def test_affirmative_clean_worktree_is_eligible(self): + result = self._assess(worktree_exists=True, worktree_clean=True) + self.assertEqual( + result["classification"], "orphaned_expired_superseded_head" + ) + self.assertTrue(result["cleanup_allowed"]) + + def test_affirmative_absent_worktree_is_eligible(self): + result = self._assess(worktree_exists=False, worktree_clean=None) + self.assertTrue(result["cleanup_allowed"]) + + def test_matrix_matches_diagnosis_gate(self): + # The cleanup matrix and the diagnosis path must agree: unknown + # evidence yields acquire-only, affirmative evidence yields cleanup. + diagnosis_unknown = leases.diagnose_reviewer_pr_lease_handoff( + self._comments(), + pr_number=701, + current_session_id=None, + current_reviewer_identity="fresh-reviewer", + current_head_sha=NEW_HEAD, + formal_reviews=[], + owner_process_alive=False, + worktree_exists=None, + worktree_clean=None, + ) + assess_unknown = self._assess(worktree_exists=None, worktree_clean=None) + self.assertNotEqual( + diagnosis_unknown["next_action"], + "cleanup_obsolete_reviewer_comment_lease", + ) + self.assertFalse(assess_unknown["cleanup_allowed"]) + + +class TestF4CrashShadowDurability(unittest.TestCase): + """F4: crash-orphan evidence survives the former 4h TTL boundary.""" + + SID = "82984-abcabcabcabc" + + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self._env = patch.dict( + os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name} + ) + self._env.start() + leases._SESSION_LEASE = None + + def tearDown(self): + self._env.stop() + self._dir.cleanup() + leases._SESSION_LEASE = None + + def _age_record(self, kind: str, hours: float, instance_id: str | None = None): + path = mss.state_file_path(kind=kind, instance_id=instance_id) + with open(path, encoding="utf-8") as fh: + envelope = json.load(fh) + stamp = ( + (_now() - timedelta(hours=hours)).isoformat().replace("+00:00", "Z") + ) + envelope["recorded_at"] = stamp + envelope["updated_at"] = stamp + envelope["payload"]["recorded_at"] = stamp + envelope["payload"]["updated_at"] = stamp + with open(path, "w", encoding="utf-8") as fh: + json.dump(envelope, fh) + + def _save_shadow(self) -> None: + mss.save_state( + kind=mss.KIND_REVIEWER_SESSION_LEASE, + instance_id=self.SID, + payload={"session_id": self.SID, "owner_pid": os.getpid()}, + ) + + def _load_shadow(self): + return mss.load_state( + kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID + ) + + def test_shadow_loads_before_former_ttl_boundary(self): + self._save_shadow() + self._age_record( + mss.KIND_REVIEWER_SESSION_LEASE, hours=3.9, instance_id=self.SID + ) + self.assertIsNotNone(self._load_shadow()) + + def test_shadow_loads_at_former_ttl_boundary(self): + self._save_shadow() + self._age_record( + mss.KIND_REVIEWER_SESSION_LEASE, hours=4.0, instance_id=self.SID + ) + self.assertIsNotNone(self._load_shadow()) + + def test_shadow_loads_long_after_former_ttl_boundary(self): + self._save_shadow() + self._age_record( + mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID + ) + record = self._load_shadow() + self.assertIsNotNone(record) + self.assertEqual(record.get("session_id"), self.SID) + + def test_stale_binding_audit_record_is_also_durable(self): + mss.save_state( + kind=mss.KIND_STALE_BINDING_RECOVERY, + payload={"cleared_env": "GITEA_ACTIVE_WORKTREE"}, + ) + self._age_record(mss.KIND_STALE_BINDING_RECOVERY, hours=27.0) + self.assertIsNotNone( + mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY) + ) + + def test_non_recovery_kinds_keep_ttl(self): + mss.save_state( + kind=mss.KIND_WORKFLOW_LOAD, payload={"workflow_hash": "abc"} + ) + self._age_record(mss.KIND_WORKFLOW_LOAD, hours=5.0) + self.assertIsNone(mss.load_state(kind=mss.KIND_WORKFLOW_LOAD)) + + def test_sanctioned_clear_is_the_terminal_reconciliation(self): + self._save_shadow() + self._age_record( + mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID + ) + mss.clear_state( + kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID + ) + self.assertIsNone(self._load_shadow()) + self.assertEqual( + mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE), [] + ) + + def test_crash_orphan_recovery_works_after_former_ttl(self): + # End to end: a crashed session's shadow still proves owner exit a + # day later. + leases.record_session_lease( + { + "pr_number": 701, + "session_id": self.SID, + "worktree": "branches/review-pr-654", + "candidate_head": OLD_HEAD, + "repo": REPO, + "comment_id": 11131, + "profile": "prgs-reviewer", + "reviewer_identity": "sysadmin", + "phase": "claimed", + }, + lease_provenance={"source": "acquire", "comment_id": 11131}, + ) + leases._SESSION_LEASE = None # simulated crash: no sanctioned clear + self._age_record( + mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID + ) + shadow = leases.load_session_lease_shadow(session_id=self.SID) + self.assertIsNotNone(shadow) + verdict = leases.assess_crashed_session_lease_orphan( + shadow, + lease={"session_id": self.SID}, + current_pid=os.getpid() + 1, + pid_alive=False, + ) + self.assertTrue(verdict["orphan_evidence"]) + self.assertIs(verdict["owner_process_alive"], False) + + +class TestF5ConcurrentSessionShadows(unittest.TestCase): + """F5: concurrent same-profile sessions keep distinct shadow records.""" + + SID_A = "11111-aaaaaaaaaaaa" + SID_B = "22222-bbbbbbbbbbbb" + + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self._env = patch.dict( + os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name} + ) + self._env.start() + leases._SESSION_LEASE = None + + def tearDown(self): + self._env.stop() + self._dir.cleanup() + leases._SESSION_LEASE = None + + def _lease(self, session_id: str, pr_number: int, head: str) -> dict: + return { + "pr_number": pr_number, + "session_id": session_id, + "worktree": f"branches/review-pr-{pr_number}-{session_id[:5]}", + "candidate_head": head, + "repo": REPO, + "comment_id": 11221, + "profile": "prgs-reviewer", + "reviewer_identity": "sysadmin", + "phase": "claimed", + } + + def _record(self, lease: dict) -> None: + leases.record_session_lease( + lease, + lease_provenance={"source": "acquire", "comment_id": 11221}, + ) + + def test_interleaved_sessions_do_not_overwrite_each_other(self): + self._record(self._lease(self.SID_A, 703, OLD_HEAD)) + self._record(self._lease(self.SID_B, 701, NEW_HEAD)) + # Session A heartbeats again after B wrote. + updated_a = self._lease(self.SID_A, 703, OLD_HEAD) + updated_a["phase"] = "validation-start" + self._record(updated_a) + + shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A) + shadow_b = leases.load_session_lease_shadow(session_id=self.SID_B) + self.assertEqual(shadow_a.get("session_id"), self.SID_A) + self.assertEqual(shadow_a.get("pr_number"), 703) + self.assertEqual(shadow_a.get("phase"), "validation-start") + self.assertEqual(shadow_b.get("session_id"), self.SID_B) + self.assertEqual(shadow_b.get("pr_number"), 701) + self.assertEqual(shadow_b.get("candidate_head"), NEW_HEAD) + + def test_shadow_lookup_never_misattributes(self): + self._record(self._lease(self.SID_A, 703, OLD_HEAD)) + self.assertIsNone( + leases.load_session_lease_shadow(session_id=self.SID_B) + ) + verdict = leases.assess_crashed_session_lease_orphan( + leases.load_session_lease_shadow(session_id=self.SID_A), + lease={"session_id": self.SID_B}, + pid_alive=False, + ) + self.assertFalse(verdict["orphan_evidence"]) + + def test_clearing_one_session_preserves_the_other(self): + self._record(self._lease(self.SID_A, 703, OLD_HEAD)) + self._record(self._lease(self.SID_B, 701, NEW_HEAD)) + # Sanctioned clear of B (the currently recorded in-memory lease). + leases.clear_session_lease() + self.assertIsNone( + leases.load_session_lease_shadow(session_id=self.SID_B) + ) + shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A) + self.assertIsNotNone(shadow_a) + self.assertEqual(shadow_a.get("session_id"), self.SID_A) + + def test_reconnected_process_recovers_by_session_id(self): + self._record(self._lease(self.SID_A, 703, OLD_HEAD)) + self._record(self._lease(self.SID_B, 701, NEW_HEAD)) + # Simulated crash + reconnect: in-memory state is gone, durable + # records remain enumerable and individually attributable. + leases._SESSION_LEASE = None + records = mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE) + self.assertEqual( + sorted(r.get("session_id") for r in records), + sorted([self.SID_A, self.SID_B]), + ) + shadow = leases.load_session_lease_shadow(session_id=self.SID_B) + self.assertEqual(shadow.get("pr_number"), 701) + + def test_legacy_single_slot_record_still_resolves_by_session_id(self): + # Upgrade path: a record written by pre-F5 code (no instance key) + # remains usable when its payload names the requested session. + mss.save_state( + kind=mss.KIND_REVIEWER_SESSION_LEASE, + payload={"session_id": self.SID_A, "owner_pid": os.getpid()}, + ) + shadow = leases.load_session_lease_shadow(session_id=self.SID_A) + self.assertIsNotNone(shadow) + self.assertIsNone( + leases.load_session_lease_shadow(session_id=self.SID_B) + ) + + +class TestF6AuditBeforeClear(_CapabilityHarness): + """F6: the durable audit record is persisted before the env clears.""" + + def _recovery_env(self, missing: str) -> dict: + return self._env( + GITEA_ACTIVE_WORKTREE=missing, + GITEA_FORCE_STALE_BINDING_RECOVERY="1", + ) + + def test_audit_write_failure_blocks_the_clear(self): + missing = self._deleted_worktree() + with patch.dict(os.environ, self._recovery_env(missing)): + with patch.object( + mss, "save_state", side_effect=OSError("disk full") + ): + report = mcp_server._assess_stale_active_binding( + auto_recover=True + ) + self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), missing) + self.assertFalse(report["recovery_performed"]) + self.assertIs(report.get("audit_persisted"), False) + self.assertTrue(report.get("audit_write_failed")) + self.assertIn("NOT cleared", " ".join(report.get("reasons") or [])) + + def test_retry_after_audit_failure_recovers(self): + missing = self._deleted_worktree() + with patch.dict(os.environ, self._recovery_env(missing)): + with patch.object( + mss, "save_state", side_effect=OSError("disk full") + ): + first = mcp_server._assess_stale_active_binding( + auto_recover=True + ) + self.assertFalse(first["recovery_performed"]) + # Persistence recovered; the retry clears with a durable audit. + second = mcp_server._assess_stale_active_binding(auto_recover=True) + self.assertTrue(second["recovery_performed"]) + self.assertIs(second.get("audit_persisted"), True) + self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ) + audit = mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY) + self.assertIsNotNone(audit) + self.assertEqual(audit.get("recovery_status"), "performed") + self.assertEqual(audit.get("cleared_value"), missing) + + def test_audit_record_exists_before_env_clear(self): + missing = self._deleted_worktree() + observed: list[tuple[str | None, str | None]] = [] + real_save = mss.save_state + + def _spy(**kwargs): + observed.append( + ( + (kwargs.get("payload") or {}).get("recovery_status"), + os.environ.get("GITEA_ACTIVE_WORKTREE"), + ) + ) + return real_save(**kwargs) + + with patch.dict(os.environ, self._recovery_env(missing)): + with patch.object(mss, "save_state", side_effect=_spy): + report = mcp_server._assess_stale_active_binding( + auto_recover=True + ) + self.assertTrue(report["recovery_performed"]) + self.assertGreaterEqual(len(observed), 2) + pending_status, env_at_pending = observed[0] + performed_status, env_at_performed = observed[-1] + # Ordering proof: the binding was still present while the pending + # audit persisted, and gone by the time the performed status wrote. + self.assertEqual(pending_status, "pending") + self.assertEqual(env_at_pending, missing) + self.assertEqual(performed_status, "performed") + self.assertIsNone(env_at_performed) + + def test_recovery_is_idempotent_after_success(self): + missing = self._deleted_worktree() + with patch.dict(os.environ, self._recovery_env(missing)): + first = mcp_server._assess_stale_active_binding(auto_recover=True) + second = mcp_server._assess_stale_active_binding(auto_recover=True) + self.assertTrue(first["recovery_performed"]) + self.assertEqual(second["classification"], sbr.CLASSIFICATION_UNBOUND) + self.assertFalse(second["recovery_performed"]) + + +if __name__ == "__main__": + unittest.main()