Merge branch 'master' into feat/issue-609-prepared-review-verdict-resume
This commit is contained in:
@@ -561,12 +561,29 @@ class TestIssue691SupersededHeadCleanup(unittest.TestCase):
|
||||
self.assertTrue(any("owns the lease" in r for r in result["reasons"]))
|
||||
|
||||
def test_superseded_without_terminal_review_denied(self):
|
||||
# Pre-expiry the missing terminal review is ambiguous (fail closed);
|
||||
# post-expiry with unknown owner evidence it is a crash-orphan (#702)
|
||||
# that still fails closed until owner-process exit is proven.
|
||||
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
|
||||
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
|
||||
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
|
||||
result = _assess([claim], formal_reviews=[], now=now)
|
||||
result = _assess(
|
||||
[claim], formal_reviews=[], now=now, owner_process_alive=None
|
||||
)
|
||||
self.assertFalse(result["cleanup_allowed"])
|
||||
self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
|
||||
self.assertEqual(
|
||||
result["classification"], "orphaned_expired_superseded_head"
|
||||
)
|
||||
|
||||
fresh_expires = now + timedelta(minutes=60)
|
||||
fresh = _lease_comment(
|
||||
expires_at=fresh_expires, last_activity=now - timedelta(minutes=5)
|
||||
)
|
||||
pre_expiry = _assess([fresh], formal_reviews=[], now=now)
|
||||
self.assertFalse(pre_expiry["cleanup_allowed"])
|
||||
self.assertEqual(
|
||||
pre_expiry["classification"], "ambiguous_conflicting_evidence"
|
||||
)
|
||||
|
||||
|
||||
class TestIssue691DiagnoseClassifications(unittest.TestCase):
|
||||
|
||||
@@ -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()
|
||||
@@ -0,0 +1,517 @@
|
||||
"""Regression tests for #702: stale runtime binding + orphaned durable lease.
|
||||
|
||||
Reproduces the PR #701 incident (lease 65664-4f248d213d60, comments
|
||||
11129/11131): the MCP daemon crashed without teardown, destroying the
|
||||
in-memory session lease while the durable comment lease survived, and the
|
||||
auto-reconnected daemon inherited a stale ``GITEA_ACTIVE_WORKTREE`` binding
|
||||
(``branches/review-pr-654``) from its parent environment.
|
||||
|
||||
Covers the five mandated scenarios:
|
||||
|
||||
1. lost in-session leases — durable shadow survives a crash and yields
|
||||
provable owner-process evidence
|
||||
2. old-head leases — expired superseded-head lease with no terminal
|
||||
review becomes recoverable only with orphan evidence + controller authority
|
||||
3. runtime/worktree mismatch — env bindings to deleted worktrees are demoted,
|
||||
never silently bound
|
||||
4. managed reconnect — boot-inherited uncorroborated bindings are
|
||||
surfaced for managed recovery; provably stale ones are clear-eligible
|
||||
5. safe expiry — pre-expiry stays fail-closed wait (no steal);
|
||||
canonical expiry unlocks acquisition and guarded cleanup
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import namespace_workspace_binding as nwb # noqa: E402
|
||||
import reviewer_pr_lease as leases # noqa: E402
|
||||
import stale_binding_recovery as sbr # noqa: E402
|
||||
|
||||
# PR #701 incident fixtures.
|
||||
OLD_HEAD = "b4d0cb22e452a07c8e816745c704ddb0f43edf0b"
|
||||
NEW_HEAD = "6b675f5c834b41f9d74e8a54294ff44dddf28ae4"
|
||||
CRASHED_SESSION = "65664-4f248d213d60"
|
||||
LEASE_COMMENT = 11131
|
||||
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
|
||||
PR = 701
|
||||
ISSUE = 699
|
||||
LEASE_WORKTREE = "branches/review-fix-issue-699-structured-auth-mcp-errors"
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _lease_comment(
|
||||
*,
|
||||
phase: str = "validation-start",
|
||||
candidate_head: str = OLD_HEAD,
|
||||
session_id: str = CRASHED_SESSION,
|
||||
comment_id: int = LEASE_COMMENT,
|
||||
last_activity: datetime | None = None,
|
||||
expires_at: datetime | None = None,
|
||||
worktree: str = LEASE_WORKTREE,
|
||||
) -> dict:
|
||||
stamp = last_activity or _now()
|
||||
body = leases.format_lease_body(
|
||||
repo=REPO,
|
||||
pr_number=PR,
|
||||
issue_number=ISSUE,
|
||||
reviewer_identity="sysadmin",
|
||||
profile="prgs-reviewer",
|
||||
session_id=session_id,
|
||||
worktree=worktree,
|
||||
phase=phase,
|
||||
candidate_head=candidate_head,
|
||||
target_branch="master",
|
||||
target_branch_sha="2" * 40,
|
||||
last_activity=stamp,
|
||||
expires_at=expires_at,
|
||||
)
|
||||
return {
|
||||
"id": comment_id,
|
||||
"body": body,
|
||||
"user": {"login": "sysadmin"},
|
||||
"author": "sysadmin",
|
||||
"created_at": stamp.isoformat(),
|
||||
"updated_at": stamp.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def _expired_superseded_comments() -> list[dict]:
|
||||
stale = _now() - timedelta(hours=3)
|
||||
return [_lease_comment(
|
||||
last_activity=stale,
|
||||
expires_at=stale + timedelta(minutes=120),
|
||||
)]
|
||||
|
||||
|
||||
def _fresh_superseded_comments() -> list[dict]:
|
||||
recent = _now() - timedelta(minutes=10)
|
||||
return [_lease_comment(
|
||||
last_activity=recent,
|
||||
expires_at=recent + timedelta(minutes=120),
|
||||
)]
|
||||
|
||||
|
||||
class TestLostInSessionLeaseShadow(unittest.TestCase):
|
||||
"""Scenario 1: the durable shadow survives a daemon crash (#702 AC3)."""
|
||||
|
||||
def setUp(self):
|
||||
leases._SESSION_LEASE = None
|
||||
|
||||
def tearDown(self):
|
||||
leases._SESSION_LEASE = None
|
||||
|
||||
def _record(self) -> dict:
|
||||
return leases.record_session_lease(
|
||||
{
|
||||
"pr_number": PR,
|
||||
"issue_number": ISSUE,
|
||||
"session_id": CRASHED_SESSION,
|
||||
"reviewer_identity": "sysadmin",
|
||||
"profile": "prgs-reviewer",
|
||||
"worktree": LEASE_WORKTREE,
|
||||
"phase": "claimed",
|
||||
"candidate_head": OLD_HEAD,
|
||||
"repo": REPO,
|
||||
"comment_id": LEASE_COMMENT,
|
||||
},
|
||||
lease_provenance={"source": "acquire", "comment_id": LEASE_COMMENT},
|
||||
)
|
||||
|
||||
def test_shadow_written_on_record_and_survives_simulated_crash(self):
|
||||
self._record()
|
||||
# Simulated unhandled crash: the in-memory dict dies with the process
|
||||
# but the sanctioned clear path never runs.
|
||||
leases._SESSION_LEASE = None
|
||||
shadow = leases.load_session_lease_shadow()
|
||||
self.assertIsNotNone(shadow)
|
||||
self.assertEqual(shadow.get("session_id"), CRASHED_SESSION)
|
||||
self.assertEqual(shadow.get("pr_number"), PR)
|
||||
self.assertEqual(int(shadow.get("owner_pid")), os.getpid())
|
||||
|
||||
def test_sanctioned_clear_removes_shadow(self):
|
||||
self._record()
|
||||
leases.clear_session_lease()
|
||||
self.assertIsNone(leases.load_session_lease_shadow())
|
||||
|
||||
def test_orphan_assessment_dead_owner_pid_proves_exit(self):
|
||||
self._record()
|
||||
leases._SESSION_LEASE = None
|
||||
shadow = leases.load_session_lease_shadow()
|
||||
lease = {"session_id": CRASHED_SESSION}
|
||||
verdict = leases.assess_crashed_session_lease_orphan(
|
||||
shadow, lease=lease, current_pid=os.getpid() + 1, pid_alive=False
|
||||
)
|
||||
self.assertTrue(verdict["orphan_evidence"])
|
||||
self.assertIs(verdict["owner_process_alive"], False)
|
||||
|
||||
def test_orphan_assessment_alive_pid_is_not_ownership_proof(self):
|
||||
self._record()
|
||||
shadow = leases.load_session_lease_shadow()
|
||||
verdict = leases.assess_crashed_session_lease_orphan(
|
||||
shadow,
|
||||
lease={"session_id": CRASHED_SESSION},
|
||||
current_pid=os.getpid() + 1,
|
||||
pid_alive=True,
|
||||
)
|
||||
self.assertFalse(verdict["orphan_evidence"])
|
||||
self.assertIsNone(verdict["owner_process_alive"])
|
||||
|
||||
def test_orphan_assessment_session_mismatch_proves_nothing(self):
|
||||
self._record()
|
||||
shadow = leases.load_session_lease_shadow()
|
||||
verdict = leases.assess_crashed_session_lease_orphan(
|
||||
shadow, lease={"session_id": "other-999"}, pid_alive=False
|
||||
)
|
||||
self.assertFalse(verdict["orphan_evidence"])
|
||||
self.assertIsNone(verdict["owner_process_alive"])
|
||||
|
||||
def test_orphan_assessment_without_shadow_is_unknown(self):
|
||||
verdict = leases.assess_crashed_session_lease_orphan(
|
||||
None, lease={"session_id": CRASHED_SESSION}
|
||||
)
|
||||
self.assertFalse(verdict["orphan_evidence"])
|
||||
self.assertIsNone(verdict["owner_process_alive"])
|
||||
|
||||
|
||||
class TestOldHeadLeaseCleanup(unittest.TestCase):
|
||||
"""Scenario 2: expired superseded-head lease with no terminal review."""
|
||||
|
||||
def _assess(self, comments, **kwargs):
|
||||
defaults = dict(
|
||||
pr_number=PR,
|
||||
current_head_sha=NEW_HEAD,
|
||||
formal_reviews=[],
|
||||
repo=REPO,
|
||||
expected_repo=REPO,
|
||||
requesting_session_id="fresh-controller",
|
||||
controller_recovery_authorized=True,
|
||||
worktree_exists=True,
|
||||
worktree_clean=True,
|
||||
owner_process_alive=False,
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return leases.assess_obsolete_reviewer_comment_lease_cleanup(
|
||||
comments, **defaults
|
||||
)
|
||||
|
||||
def test_expired_orphan_with_full_evidence_is_cleanup_eligible(self):
|
||||
result = self._assess(_expired_superseded_comments())
|
||||
self.assertEqual(result["classification"], "orphaned_expired_superseded_head")
|
||||
self.assertTrue(result["cleanup_allowed"])
|
||||
self.assertIn("phase: released", result["release_body"] or "")
|
||||
self.assertEqual(
|
||||
result["exact_next_action"], "cleanup_obsolete_reviewer_comment_lease"
|
||||
)
|
||||
|
||||
def test_expired_orphan_without_owner_evidence_fails_closed(self):
|
||||
result = self._assess(
|
||||
_expired_superseded_comments(), owner_process_alive=None
|
||||
)
|
||||
self.assertFalse(result["cleanup_allowed"])
|
||||
self.assertIn(
|
||||
"provable owner-process-exit evidence",
|
||||
" ".join(result["fail_closed_reasons"]),
|
||||
)
|
||||
|
||||
def test_expired_orphan_without_controller_authority_fails_closed(self):
|
||||
result = self._assess(
|
||||
_expired_superseded_comments(), controller_recovery_authorized=False
|
||||
)
|
||||
self.assertFalse(result["cleanup_allowed"])
|
||||
|
||||
def test_expired_orphan_with_dirty_worktree_fails_closed(self):
|
||||
result = self._assess(
|
||||
_expired_superseded_comments(),
|
||||
worktree_clean=False,
|
||||
worktree_has_unpreserved_work=True,
|
||||
)
|
||||
self.assertFalse(result["cleanup_allowed"])
|
||||
|
||||
def test_unexpired_superseded_no_terminal_stays_ambiguous_wait(self):
|
||||
# Regression guard: pre-expiry there is still no steal path.
|
||||
result = self._assess(_fresh_superseded_comments())
|
||||
self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
|
||||
self.assertFalse(result["cleanup_allowed"])
|
||||
self.assertEqual(result["exact_next_action"], "wait")
|
||||
|
||||
|
||||
class TestRuntimeWorktreeMismatch(unittest.TestCase):
|
||||
"""Scenario 3: env bindings to deleted worktrees are demoted (#702 AC2)."""
|
||||
|
||||
def test_missing_active_env_binding_is_demoted(self):
|
||||
demotions: list[str] = []
|
||||
missing = "/nonexistent/branches/review-pr-654"
|
||||
path, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reviewer",
|
||||
process_project_root=os.getcwd(),
|
||||
env={"GITEA_ACTIVE_WORKTREE": missing},
|
||||
demotions=demotions,
|
||||
verify_paths=True,
|
||||
)
|
||||
self.assertEqual(source, "MCP server process root (default)")
|
||||
self.assertEqual(len(demotions), 1)
|
||||
self.assertIn("no longer exists", demotions[0])
|
||||
|
||||
def test_missing_active_falls_through_to_existing_role_env(self):
|
||||
existing = os.getcwd()
|
||||
path, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reviewer",
|
||||
process_project_root="/tmp",
|
||||
env={
|
||||
"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654",
|
||||
"GITEA_REVIEWER_WORKTREE": existing,
|
||||
},
|
||||
verify_paths=True,
|
||||
)
|
||||
self.assertEqual(path, os.path.realpath(existing))
|
||||
self.assertEqual(source, "GITEA_REVIEWER_WORKTREE environment variable")
|
||||
|
||||
def test_existing_active_env_binding_still_wins(self):
|
||||
existing = os.getcwd()
|
||||
path, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reviewer",
|
||||
process_project_root="/tmp",
|
||||
env={"GITEA_ACTIVE_WORKTREE": existing},
|
||||
verify_paths=True,
|
||||
)
|
||||
self.assertEqual(path, os.path.realpath(existing))
|
||||
self.assertEqual(source, "GITEA_ACTIVE_WORKTREE environment variable")
|
||||
|
||||
def test_explicit_worktree_path_argument_is_never_demoted(self):
|
||||
missing = "/nonexistent/branches/explicit"
|
||||
path, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reviewer",
|
||||
worktree_path=missing,
|
||||
process_project_root="/tmp",
|
||||
env={},
|
||||
verify_paths=True,
|
||||
)
|
||||
self.assertEqual(source, "worktree_path argument")
|
||||
|
||||
def test_mutation_context_reports_demotion_in_ignored_bindings(self):
|
||||
ctx = nwb.resolve_namespace_mutation_context(
|
||||
role_kind="reviewer",
|
||||
worktree_path=None,
|
||||
process_project_root=os.getcwd(),
|
||||
env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"},
|
||||
)
|
||||
joined = " ".join(ctx["ignored_bindings"])
|
||||
self.assertIn("stale binding, #702", joined)
|
||||
|
||||
|
||||
class TestManagedReconnectRecovery(unittest.TestCase):
|
||||
"""Scenario 4: boot-inherited bindings and the sanctioned recovery plan."""
|
||||
|
||||
def test_uncorroborated_inherited_reviewer_binding_is_unverified(self):
|
||||
classification = sbr.classify_active_worktree_binding(
|
||||
active_value=os.getcwd(),
|
||||
role_env_value=None,
|
||||
session_lease_worktree=None,
|
||||
boot_inherited=True,
|
||||
path_exists=True,
|
||||
role_kind="reviewer",
|
||||
)
|
||||
self.assertEqual(
|
||||
classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
|
||||
)
|
||||
self.assertFalse(classification["clear_eligible"])
|
||||
plan = sbr.plan_recovery(classification)
|
||||
self.assertFalse(plan["clear_allowed"])
|
||||
self.assertIn("managed", plan["exact_next_action"])
|
||||
self.assertIn("do not clear or rewrite", plan["exact_next_action"])
|
||||
|
||||
def test_missing_path_binding_is_provably_stale_and_clear_eligible(self):
|
||||
classification = sbr.classify_active_worktree_binding(
|
||||
active_value="/nonexistent/branches/review-pr-654",
|
||||
boot_inherited=True,
|
||||
path_exists=False,
|
||||
role_kind="reviewer",
|
||||
)
|
||||
self.assertEqual(
|
||||
classification["classification"],
|
||||
sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
|
||||
)
|
||||
plan = sbr.plan_recovery(classification)
|
||||
self.assertTrue(plan["clear_allowed"])
|
||||
|
||||
def test_inherited_binding_superseded_by_session_lease_is_clear_eligible(self):
|
||||
classification = sbr.classify_active_worktree_binding(
|
||||
active_value="/tmp",
|
||||
session_lease_worktree=os.getcwd(),
|
||||
boot_inherited=True,
|
||||
path_exists=True,
|
||||
role_kind="reviewer",
|
||||
)
|
||||
self.assertEqual(
|
||||
classification["classification"],
|
||||
sbr.CLASSIFICATION_SUPERSEDED_BY_SESSION_LEASE,
|
||||
)
|
||||
self.assertTrue(sbr.plan_recovery(classification)["clear_allowed"])
|
||||
|
||||
def test_post_boot_binding_conflict_is_surfaced_not_cleared(self):
|
||||
classification = sbr.classify_active_worktree_binding(
|
||||
active_value="/tmp",
|
||||
session_lease_worktree=os.getcwd(),
|
||||
boot_inherited=False,
|
||||
path_exists=True,
|
||||
role_kind="reviewer",
|
||||
)
|
||||
self.assertEqual(
|
||||
classification["classification"], sbr.CLASSIFICATION_UNVERIFIED_INHERITED
|
||||
)
|
||||
self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])
|
||||
|
||||
def test_corroborated_bindings_are_untouched(self):
|
||||
for kwargs in (
|
||||
dict(role_env_value=os.getcwd()),
|
||||
dict(session_lease_worktree=os.getcwd()),
|
||||
):
|
||||
classification = sbr.classify_active_worktree_binding(
|
||||
active_value=os.getcwd(),
|
||||
boot_inherited=True,
|
||||
path_exists=True,
|
||||
role_kind="reviewer",
|
||||
**kwargs,
|
||||
)
|
||||
self.assertEqual(
|
||||
classification["classification"], sbr.CLASSIFICATION_CORROBORATED
|
||||
)
|
||||
self.assertFalse(sbr.plan_recovery(classification)["clear_allowed"])
|
||||
|
||||
def test_author_inherited_binding_stays_corroborated(self):
|
||||
classification = sbr.classify_active_worktree_binding(
|
||||
active_value=os.getcwd(),
|
||||
boot_inherited=True,
|
||||
path_exists=True,
|
||||
role_kind="author",
|
||||
)
|
||||
self.assertEqual(
|
||||
classification["classification"], sbr.CLASSIFICATION_CORROBORATED
|
||||
)
|
||||
|
||||
def test_apply_recovery_clears_env_only_for_authorized_plan(self):
|
||||
env = {"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-pr-654"}
|
||||
plan = sbr.plan_recovery(
|
||||
sbr.classify_active_worktree_binding(
|
||||
active_value=env["GITEA_ACTIVE_WORKTREE"],
|
||||
boot_inherited=True,
|
||||
path_exists=False,
|
||||
role_kind="reviewer",
|
||||
)
|
||||
)
|
||||
applied = sbr.apply_recovery(plan, env=env)
|
||||
self.assertTrue(applied["performed"])
|
||||
self.assertNotIn("GITEA_ACTIVE_WORKTREE", env)
|
||||
self.assertIn("review-pr-654", applied["cleared_value"])
|
||||
|
||||
def test_apply_recovery_refuses_unauthorized_plan(self):
|
||||
env = {"GITEA_ACTIVE_WORKTREE": os.getcwd()}
|
||||
plan = sbr.plan_recovery(
|
||||
sbr.classify_active_worktree_binding(
|
||||
active_value=env["GITEA_ACTIVE_WORKTREE"],
|
||||
boot_inherited=True,
|
||||
path_exists=True,
|
||||
role_kind="reviewer",
|
||||
)
|
||||
)
|
||||
applied = sbr.apply_recovery(plan, env=env)
|
||||
self.assertFalse(applied["performed"])
|
||||
self.assertIn("GITEA_ACTIVE_WORKTREE", env)
|
||||
|
||||
|
||||
class TestSafeExpiryDiagnosis(unittest.TestCase):
|
||||
"""Scenario 5: canonical expiry flips wait into acquire/guarded cleanup."""
|
||||
|
||||
def _diagnose(self, comments, **kwargs):
|
||||
defaults = dict(
|
||||
pr_number=PR,
|
||||
current_session_id=None,
|
||||
current_reviewer_identity="fresh-reviewer",
|
||||
current_head_sha=NEW_HEAD,
|
||||
formal_reviews=[],
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return leases.diagnose_reviewer_pr_lease_handoff(comments, **defaults)
|
||||
|
||||
def setUp(self):
|
||||
leases._SESSION_LEASE = None
|
||||
|
||||
def test_pre_expiry_superseded_no_terminal_stays_wait(self):
|
||||
diagnosis = self._diagnose(_fresh_superseded_comments())
|
||||
self.assertEqual(
|
||||
diagnosis["classification"], "ambiguous_conflicting_evidence"
|
||||
)
|
||||
self.assertEqual(diagnosis["next_action"], "wait")
|
||||
|
||||
def test_post_expiry_without_orphan_evidence_unlocks_acquire(self):
|
||||
diagnosis = self._diagnose(_expired_superseded_comments())
|
||||
self.assertEqual(
|
||||
diagnosis["classification"], "orphaned_expired_superseded_head"
|
||||
)
|
||||
self.assertEqual(diagnosis["next_action"], "acquire")
|
||||
|
||||
def test_post_expiry_with_orphan_evidence_offers_guarded_cleanup(self):
|
||||
diagnosis = self._diagnose(
|
||||
_expired_superseded_comments(),
|
||||
owner_process_alive=False,
|
||||
worktree_clean=True,
|
||||
worktree_exists=True,
|
||||
)
|
||||
self.assertEqual(
|
||||
diagnosis["classification"], "orphaned_expired_superseded_head"
|
||||
)
|
||||
self.assertEqual(
|
||||
diagnosis["next_action"], "cleanup_obsolete_reviewer_comment_lease"
|
||||
)
|
||||
|
||||
def test_expired_lease_no_longer_blocks_fresh_acquisition(self):
|
||||
assessment = leases.assess_acquire_lease(
|
||||
_expired_superseded_comments(),
|
||||
pr_number=PR,
|
||||
reviewer_identity="fresh-reviewer",
|
||||
profile="prgs-reviewer",
|
||||
session_id="99999-fresh",
|
||||
repo=REPO,
|
||||
issue_number=ISSUE,
|
||||
worktree="branches/review-pr-701-fresh",
|
||||
candidate_head=NEW_HEAD,
|
||||
target_branch="master",
|
||||
target_branch_sha="2" * 40,
|
||||
)
|
||||
self.assertTrue(assessment.get("acquire_allowed"))
|
||||
|
||||
def test_unparseable_expiry_fails_closed_in_cleanup(self):
|
||||
comment = _lease_comment(
|
||||
last_activity=_now() - timedelta(hours=3), expires_at=None
|
||||
)
|
||||
comment["body"] = comment["body"].replace(
|
||||
"expires_at: ", "expires_at: not-a-timestamp"
|
||||
)
|
||||
result = leases.assess_obsolete_reviewer_comment_lease_cleanup(
|
||||
[comment],
|
||||
pr_number=PR,
|
||||
current_head_sha=NEW_HEAD,
|
||||
formal_reviews=[],
|
||||
repo=REPO,
|
||||
expected_repo=REPO,
|
||||
requesting_session_id="fresh-controller",
|
||||
controller_recovery_authorized=True,
|
||||
worktree_exists=True,
|
||||
worktree_clean=True,
|
||||
owner_process_alive=False,
|
||||
)
|
||||
self.assertFalse(result["cleanup_allowed"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user