jcwalker3andClaude Opus 4.8
c763161702
Addresses the five blocking findings of review 644 on PR #972.
B1 — live ownership and status revalidation. resolve_missing_worktree_binding
now re-reads the authoritative lease, session, checkpoint, and issue-lock rows
immediately before mutating and diffs them against the snapshot the audit
recorded (binding path and identity, lease id/status/session/owner pid,
checkpoint path/status, live-session evidence, trustworthy ownership evidence,
issue-lock state). Any drift fails closed without mutation, and the binding is
reclassified from the live values rather than the audit snapshot. A candidate
carrying no audited snapshot is refused rather than trusted.
B2 — server-enforced cleanup authorization. Apply mode no longer accepts a
client-supplied operator_authorized boolean; it is rejected outright at the MCP
tool and in the module (#709 F1 / review 434). Authorization is now the
project's own reconciliation cleanup gate, required at both the task-capability
boundary (new reconciler-only reconcile_missing_worktree_bindings capability,
gitea.branch.delete, role-exclusive) and the production mutation boundary
(an authorized audit_reconciliation_mode cleanup phase, re-checked at the point
of mutation so a forged authorization mapping cannot stand in for the gate).
Dry-run remains available to any gitea.read profile and stays non-mutating.
Existing role, repository, parity, and provenance gates are unchanged.
B3 — expected-path compare-and-swap. retire_session_checkpoint_worktree_path
now requires expected_path and performs a guarded update keyed on the stored
path, refusing without mutation when the stored path was moved, replaced, or
concurrently changed, when the row is unknown, or when a selector matches more
than one checkpoint. retire_lease_worktree_path gains the same treatment plus
optional status/session/owner-pid compare-and-swap, and its UPDATE is keyed on
the audited path. Both report an idempotent already_retired outcome instead of
falsely reporting a retirement.
B4 — live-session and issue-lock evidence. session_active is now derived from
the control-plane sessions table instead of never being set, along two axes:
genuine liveness (recorded active, PID not dead, heartbeat fresh — the rule
reused from restart_coordinator) and weaker but still trustworthy recorded
ownership. A non-terminal lease now protects its binding regardless of whether
the recorded PID is alive, so dead-PID evidence alone can no longer retire a
lease the control plane still holds. The previously unused issue_lock_store is
now read: a live durable issue lock binding the path or branch blocks cleanup,
and locks whose own paths are missing are reported for release through their
own lifecycle rather than retired here.
B5 — adversarial regression coverage. The suite now drives the registered MCP
tools through mcp_server, the real ControlPlaneDB, and the real cleanup gate,
covering lease status/ownership/session/path drift, expected-path mismatch,
concurrent recreation, an unauthorized caller submitting operator_authorized,
wrong profile and missing capability, live-session and trustworthy-owner
evidence, conflicting issue locks, non-mutating dry-run, exact-binding-only
retirement, preservation of unrelated worktrees and git metadata, idempotent
re-execution, and worktrees-dimension resolution.
All original #970 acceptance criteria are preserved, including the distinctions
between deleted paths, moved paths, unavailable hosts or mounts, transient
filesystem failures, live ownership, and concurrent recreation.
Tests: focused #970 suite 47 passed. Adjacent suites (capability role
invariants, audit reconciliation mode, control plane DB, lease lifecycle,
reconciler cleanup integration, delete-branch capability, restart coordinator,
bootstrap lock contract) 252 passed / 93 subtests. Full suite 28 failed /
6000 passed, an exact match of the pre-change baseline's 28 failing test ids
at 3f584352 (28 failed / 5960 passed).
Closes #970
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
1084 lines
46 KiB
Python
1084 lines
46 KiB
Python
"""Tests for issue #970: safely resolve worktree bindings whose paths are missing.
|
|
|
|
Covers all acceptance criteria for #970:
|
|
- missing path reporting with repo/branch/issue/session/lease/host correlation
|
|
- fail closed on unavailable host or mount failure
|
|
- fail closed on moved worktree paths
|
|
- fail closed on live lease or active session/process
|
|
- atomic pre-mutation re-validation preventing recreation races
|
|
- targeted retirement preserving unrelated worktrees & metadata
|
|
- safe and idempotent repeated execution
|
|
- worktrees dimension resolution post-cleanup
|
|
- resolution of the two historical post-PR #968 records (#795 / #635)
|
|
|
|
Plus the adversarial coverage required by review 644 (B5). Every test below
|
|
drives the *production* paths — the registered MCP tools loaded through
|
|
``mcp_server``, the real ``ControlPlaneDB`` compare-and-swap, and the real
|
|
``audit_reconciliation_mode`` cleanup gate — so each one fails if the
|
|
corresponding protection is removed:
|
|
|
|
- B1: lease status / ownership / session / path drift between audit and apply
|
|
- B2: server-enforced cleanup authorization at both the task-capability and
|
|
the production mutation boundary, including a rejected ``operator_authorized``
|
|
- B3: expected-path compare-and-swap on lease *and* checkpoint retirement
|
|
- B4: live-session, trustworthy-owner, and conflicting-issue-lock evidence
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import audit_reconciliation_mode as arm
|
|
import control_plane_db as cpd
|
|
import issue_lock_store
|
|
import mcp_server
|
|
import missing_worktree_reconcile as mwr
|
|
import task_capability_map
|
|
|
|
# Profiles mirroring the configured role split. The reconciler is the only role
|
|
# the project grants cleanup authority to (#729 / #970 review 644 B2).
|
|
RECONCILER_PROFILE = {
|
|
"profile_name": "prgs-reconciler",
|
|
"role": "reconciler",
|
|
"role_kind": "reconciler",
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.pr.close",
|
|
"gitea.issue.comment",
|
|
"gitea.branch.delete",
|
|
],
|
|
"forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"],
|
|
"audit_label": "prgs-reconciler",
|
|
}
|
|
|
|
REVIEWER_PROFILE = {
|
|
"profile_name": "prgs-reviewer",
|
|
"role": "reviewer",
|
|
"role_kind": "reviewer",
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.pr.review",
|
|
"gitea.pr.approve",
|
|
"gitea.issue.comment",
|
|
],
|
|
"forbidden_operations": ["gitea.branch.delete", "gitea.repo.commit"],
|
|
"audit_label": "prgs-reviewer",
|
|
}
|
|
|
|
AUTHOR_WITH_DELETE_PROFILE = {
|
|
# Holds the cleanup *permission* but not the cleanup *role*: permission
|
|
# alone must never authorize the mutation.
|
|
"profile_name": "prgs-author-delete",
|
|
"role": "author",
|
|
"role_kind": "author",
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.branch.delete",
|
|
"gitea.repo.commit",
|
|
],
|
|
"forbidden_operations": [],
|
|
"audit_label": "prgs-author-delete",
|
|
}
|
|
|
|
|
|
class _MissingWorktreeTestBase(unittest.TestCase):
|
|
"""Shared fixture: a real control-plane DB, git root, and lock dir."""
|
|
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.db_path = os.path.join(self.temp_dir, "control_plane.sqlite3")
|
|
self.db = cpd.ControlPlaneDB(self.db_path)
|
|
self.project_root = os.path.join(self.temp_dir, "repo")
|
|
os.makedirs(os.path.join(self.project_root, "branches"), exist_ok=True)
|
|
# Initialize git repo in project_root for git rev-parse checks.
|
|
# Argument list form: no shell, so a temp path can never be interpreted
|
|
# as shell syntax.
|
|
subprocess.run(
|
|
["git", "-C", self.project_root, "init", "-q"],
|
|
check=True,
|
|
capture_output=True,
|
|
)
|
|
|
|
# Point the durable issue-lock store at an isolated directory so the
|
|
# real lock reader runs against test data, not the operator's locks.
|
|
self.lock_dir = os.path.join(self.temp_dir, "issue-locks")
|
|
os.makedirs(self.lock_dir, exist_ok=True)
|
|
self._prior_lock_dir = os.environ.get(issue_lock_store.LOCK_DIR_ENV)
|
|
os.environ[issue_lock_store.LOCK_DIR_ENV] = self.lock_dir
|
|
|
|
arm.clear_phase()
|
|
|
|
def tearDown(self):
|
|
arm.clear_phase()
|
|
if self._prior_lock_dir is None:
|
|
os.environ.pop(issue_lock_store.LOCK_DIR_ENV, None)
|
|
else:
|
|
os.environ[issue_lock_store.LOCK_DIR_ENV] = self._prior_lock_dir
|
|
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────
|
|
|
|
def _age_session_heartbeat(self, session_id, *, seconds=None):
|
|
"""Age a session's heartbeat past the staleness grace.
|
|
|
|
``assign_and_lease`` opens a session row whose heartbeat is fresh by
|
|
construction, and nothing closes it on release. Real abandoned bindings
|
|
— the ones #970 exists to retire — have long-stale heartbeats, so the
|
|
fixture models that rather than testing against a session that is live
|
|
only because the test just created it.
|
|
"""
|
|
age = seconds or (mwr.SESSION_HEARTBEAT_STALE_SECONDS + 3600)
|
|
stale_ts = (
|
|
datetime.now(timezone.utc) - timedelta(seconds=age)
|
|
).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
|
with self.db._tx() as conn: # noqa: SLF001 - fixture ages durable state
|
|
conn.execute(
|
|
"UPDATE sessions SET last_heartbeat_at = ? WHERE session_id = ?",
|
|
(stale_ts, session_id),
|
|
)
|
|
|
|
def _stale_lease(
|
|
self,
|
|
*,
|
|
number=402,
|
|
path=None,
|
|
session_id="sess-402",
|
|
org="org",
|
|
repo="repo",
|
|
):
|
|
"""Create a released lease whose recorded worktree path is missing."""
|
|
stale_path = path or os.path.join(self.temp_dir, f"stale_missing_{number}")
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs", org=org, repo=repo, kind="issue", number=number,
|
|
role="author", session_id=session_id, worktree_path=stale_path,
|
|
)
|
|
self.db.release_lease_recorded(res.lease_id, session_id=session_id)
|
|
self._age_session_heartbeat(session_id)
|
|
return res.lease_id, stale_path
|
|
|
|
def _audit(self):
|
|
return mwr.audit_missing_worktree_bindings(
|
|
self.db, project_root=self.project_root, remote="prgs"
|
|
)
|
|
|
|
def _candidate(self):
|
|
audit = self._audit()
|
|
candidates = audit["confirmed_stale_candidates"]
|
|
self.assertTrue(candidates, "expected one confirmed stale candidate")
|
|
return candidates[0]
|
|
|
|
def _mint_cleanup_authorization(self, role="reconciler"):
|
|
"""Mint a genuine server-side cleanup authorization (#419 gate)."""
|
|
arm.clear_phase()
|
|
arm.enter_audit_phase("reconcile_merged_cleanups")
|
|
phase = arm.authorize_cleanup_phase(
|
|
operator_approved=True,
|
|
delete_capability_proven=True,
|
|
safety_proof={"safe_to_remove_worktree": True},
|
|
before_after_snapshot={
|
|
"before": "binding recorded with missing path",
|
|
"after": "binding retired",
|
|
},
|
|
)
|
|
self.assertTrue(phase["authorized"], phase)
|
|
return mwr.assess_cleanup_authorization(
|
|
role=role,
|
|
capability_blockers=[],
|
|
operator_authorized=False,
|
|
task_capability_resolved=True,
|
|
)
|
|
|
|
def _lease_row(self, lease_id):
|
|
state = self.db.get_lease_workflow_state(lease_id)
|
|
return dict(state["lease"])
|
|
|
|
def _set_lease_columns(self, lease_id, **columns):
|
|
"""Simulate a concurrent writer changing the lease between audit/apply."""
|
|
assignments = ", ".join(f"{k} = ?" for k in columns)
|
|
params = list(columns.values()) + [lease_id]
|
|
with self.db._tx() as conn: # noqa: SLF001 - deliberate race simulation
|
|
conn.execute(
|
|
f"UPDATE leases SET {assignments} WHERE lease_id = ?", params
|
|
)
|
|
|
|
def _write_issue_lock(self, *, issue_number, worktree_path, branch, pid=None):
|
|
"""Write a durable issue lock through the production lock store."""
|
|
path = issue_lock_store.lock_file_path(
|
|
remote="prgs",
|
|
org="org",
|
|
repo="repo",
|
|
issue_number=issue_number,
|
|
lock_dir=self.lock_dir,
|
|
)
|
|
issue_lock_store.save_lock_file(path, {
|
|
"issue_number": issue_number,
|
|
"branch_name": branch,
|
|
"worktree_path": worktree_path,
|
|
"session_pid": pid if pid is not None else os.getpid(),
|
|
"pid": pid if pid is not None else os.getpid(),
|
|
"claimant": {"username": "jcwalker3", "profile": "prgs-author"},
|
|
})
|
|
return path
|
|
|
|
|
|
class TestMissingWorktreeReconcile(_MissingWorktreeTestBase):
|
|
"""Original #970 acceptance criteria."""
|
|
|
|
def test_audit_missing_worktree_bindings_reports_associations(self):
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
kind="pr",
|
|
number=795,
|
|
expected_head_sha="head795",
|
|
role="reviewer",
|
|
session_id="reviewer-pr795-test-session",
|
|
worktree_path="/tmp/nonexistent_wt_path_795",
|
|
)
|
|
lease_id = res.lease_id
|
|
self.assertIsNotNone(lease_id)
|
|
# Release the lease (dead/released)
|
|
self.db.release_lease_recorded(lease_id, session_id="reviewer-pr795-test-session")
|
|
self._age_session_heartbeat("reviewer-pr795-test-session")
|
|
|
|
audit = mwr.audit_missing_worktree_bindings(
|
|
self.db,
|
|
project_root=self.project_root,
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
)
|
|
|
|
self.assertTrue(audit["host_mount_healthy"])
|
|
self.assertEqual(audit["missing_count"], 1)
|
|
self.assertEqual(audit["confirmed_stale_count"], 1)
|
|
|
|
candidate = audit["confirmed_stale_candidates"][0]
|
|
self.assertEqual(candidate["lease_id"], lease_id)
|
|
self.assertEqual(candidate["work_number"], 795)
|
|
self.assertEqual(candidate["session_id"], "reviewer-pr795-test-session")
|
|
self.assertEqual(candidate["classification"], mwr.CLASS_CONFIRMED_STALE_DELETED)
|
|
self.assertTrue(candidate["retire_eligible"])
|
|
|
|
def test_unavailable_host_mount_blocks_retirement(self):
|
|
# Point to a non-existent project_root
|
|
bad_root = os.path.join(self.temp_dir, "nonexistent_repo_dir")
|
|
audit = mwr.audit_missing_worktree_bindings(
|
|
self.db,
|
|
project_root=bad_root,
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(audit["host_mount_healthy"])
|
|
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=100,
|
|
role="author", session_id="sess-100", worktree_path="/tmp/missing_100",
|
|
)
|
|
self.db.release_lease_recorded(res.lease_id, session_id="sess-100")
|
|
self._age_session_heartbeat("sess-100")
|
|
|
|
audit2 = mwr.audit_missing_worktree_bindings(self.db, project_root=bad_root, remote="prgs")
|
|
self.assertEqual(audit2["unavailable_host_count"], 1)
|
|
self.assertEqual(audit2["confirmed_stale_count"], 0)
|
|
self.assertFalse(audit2["missing_bindings"][0]["retire_eligible"])
|
|
|
|
def test_live_lease_and_live_session_block_retirement(self):
|
|
self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=200,
|
|
role="author", session_id="sess-live-200", worktree_path="/tmp/missing_live_200",
|
|
owner_pid=os.getpid(),
|
|
)
|
|
|
|
audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
|
|
self.assertEqual(audit["live_protected_count"], 1)
|
|
self.assertEqual(audit["confirmed_stale_count"], 0)
|
|
|
|
# Attempting resolution on a live protected binding is refused.
|
|
c = audit["missing_bindings"][0]
|
|
res_mut = mwr.resolve_missing_worktree_binding(
|
|
self.db,
|
|
binding=c,
|
|
dry_run=False,
|
|
cleanup_authorization=self._mint_cleanup_authorization(),
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertFalse(res_mut["success"])
|
|
self.assertEqual(res_mut["reason"], "revalidation_failed")
|
|
|
|
def test_moved_worktree_path_blocks_retirement(self):
|
|
# Create a worktree under branches/ to simulate moved path
|
|
moved_dir = os.path.join(self.project_root, "branches", "issue-635-project-registry-api")
|
|
os.makedirs(moved_dir, exist_ok=True)
|
|
|
|
res = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=635,
|
|
role="author", session_id="sess-635", worktree_path="/tmp/old_missing_path_635",
|
|
)
|
|
self.db.release_lease_recorded(res.lease_id, session_id="sess-635")
|
|
self._age_session_heartbeat("sess-635")
|
|
|
|
cls_info = mwr.classify_worktree_binding(
|
|
recorded_path="/tmp/old_missing_path_635",
|
|
branch="issue-635-project-registry-api",
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertEqual(cls_info["classification"], mwr.CLASS_MOVED_WORKTREE)
|
|
self.assertFalse(cls_info["retire_eligible"])
|
|
self.assertEqual(os.path.realpath(cls_info["moved_to_path"]), os.path.realpath(moved_dir))
|
|
|
|
def test_recreation_race_revalidation_blocks_mutation(self):
|
|
path = os.path.join(self.temp_dir, "recreated_wt")
|
|
lease_id, _ = self._stale_lease(number=300, path=path, session_id="sess-300")
|
|
|
|
# Initial audit when path is missing
|
|
candidate = self._candidate()
|
|
|
|
# Simulate concurrent recreation of worktree path
|
|
os.makedirs(path, exist_ok=True)
|
|
|
|
# Pre-mutation revalidation detects recreation and blocks
|
|
res_mut = mwr.resolve_missing_worktree_binding(
|
|
self.db,
|
|
binding=candidate,
|
|
dry_run=False,
|
|
cleanup_authorization=self._mint_cleanup_authorization(),
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertFalse(res_mut["success"])
|
|
self.assertEqual(res_mut["reason"], "revalidation_failed")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], path)
|
|
|
|
def test_authorized_cleanup_retires_stale_binding_and_is_idempotent(self):
|
|
# Create 2 leases: 1 stale missing, 1 valid existing worktree
|
|
valid_path = os.path.join(self.project_root, "branches", "valid_wt")
|
|
os.makedirs(valid_path, exist_ok=True)
|
|
unrelated_marker = os.path.join(valid_path, "keep-me.txt")
|
|
with open(unrelated_marker, "w", encoding="utf-8") as handle:
|
|
handle.write("unrelated worktree content")
|
|
|
|
stale_path = os.path.join(self.temp_dir, "stale_missing_400")
|
|
|
|
res1 = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=401,
|
|
role="author", session_id="sess-401", worktree_path=valid_path,
|
|
)
|
|
|
|
res2 = self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=402,
|
|
role="author", session_id="sess-402", worktree_path=stale_path,
|
|
)
|
|
self.db.release_lease_recorded(res2.lease_id, session_id="sess-402")
|
|
self._age_session_heartbeat("sess-402")
|
|
|
|
# Dry-run reconciliation
|
|
dry = mwr.reconcile_missing_worktree_bindings(
|
|
self.db, project_root=self.project_root, remote="prgs", dry_run=True
|
|
)
|
|
self.assertTrue(dry["success"])
|
|
self.assertTrue(dry["dry_run"])
|
|
self.assertEqual(dry["stale_candidates_count"], 1)
|
|
|
|
# Dry-run mutated nothing.
|
|
self.assertEqual(
|
|
self._lease_row(res2.lease_id)["worktree_path"], stale_path
|
|
)
|
|
|
|
# Apply reconciliation with a genuine server-minted authorization
|
|
applied = mwr.reconcile_missing_worktree_bindings(
|
|
self.db,
|
|
project_root=self.project_root,
|
|
remote="prgs",
|
|
dry_run=False,
|
|
cleanup_authorization=self._mint_cleanup_authorization(),
|
|
)
|
|
self.assertTrue(applied["success"])
|
|
self.assertFalse(applied["dry_run"])
|
|
self.assertTrue(applied["worktrees_dimension_resolved"])
|
|
self.assertEqual(applied["after_audit"]["missing_count"], 0)
|
|
|
|
# Verify DB: lease 2 worktree_path is cleared, lease 1 worktree_path remains intact
|
|
l1 = self.db.get_lease_workflow_state(res1.lease_id)
|
|
self.assertEqual(l1["lease"]["worktree_path"], valid_path)
|
|
|
|
l2 = self.db.get_lease_workflow_state(res2.lease_id)
|
|
self.assertEqual(l2["lease"]["worktree_path"], "")
|
|
prov = json.loads(l2["lease"]["provenance_json"])
|
|
self.assertTrue(prov.get("worktree_path_retired"))
|
|
self.assertEqual(prov.get("retired_worktree_path"), stale_path)
|
|
|
|
# Unrelated worktree contents and git metadata are untouched.
|
|
self.assertTrue(os.path.isfile(unrelated_marker))
|
|
self.assertTrue(os.path.isdir(os.path.join(self.project_root, ".git")))
|
|
|
|
# Re-run (idempotency check)
|
|
rerun = mwr.reconcile_missing_worktree_bindings(
|
|
self.db,
|
|
project_root=self.project_root,
|
|
remote="prgs",
|
|
dry_run=False,
|
|
cleanup_authorization=self._mint_cleanup_authorization(),
|
|
)
|
|
self.assertTrue(rerun["success"])
|
|
self.assertTrue(rerun["worktrees_dimension_resolved"])
|
|
self.assertEqual(rerun["stale_candidates_count"], 0)
|
|
|
|
def test_historical_post_pr968_records_resolution(self):
|
|
# Create historical records matching #970 description
|
|
res_795 = self.db.assign_and_lease(
|
|
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", kind="pr", number=795,
|
|
expected_head_sha="head795hist", role="reviewer", session_id="reviewer-pr795-hist",
|
|
worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/review-feat-issue-628-autonomous-handoffs-orchestration",
|
|
)
|
|
self.db.release_lease_recorded(res_795.lease_id, session_id="reviewer-pr795-hist")
|
|
self._age_session_heartbeat("reviewer-pr795-hist")
|
|
|
|
res_635 = self.db.assign_and_lease(
|
|
remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", kind="issue", number=635,
|
|
role="author", session_id="author-issue635-hist",
|
|
worktree_path="/Users/jasonwalker/Development/Gitea-Tools/branches/issue-635-project-registry-api",
|
|
)
|
|
self.db.release_lease_recorded(res_635.lease_id, session_id="author-issue635-hist")
|
|
self._age_session_heartbeat("author-issue635-hist")
|
|
|
|
# Audit finds both historical records
|
|
audit = mwr.audit_missing_worktree_bindings(self.db, project_root=self.project_root, remote="prgs")
|
|
self.assertEqual(audit["confirmed_stale_count"], 2)
|
|
|
|
# Reconcile safely resolves both
|
|
rec = mwr.reconcile_missing_worktree_bindings(
|
|
self.db,
|
|
project_root=self.project_root,
|
|
remote="prgs",
|
|
dry_run=False,
|
|
cleanup_authorization=self._mint_cleanup_authorization(),
|
|
)
|
|
self.assertTrue(rec["worktrees_dimension_resolved"])
|
|
self.assertEqual(rec["after_audit"]["missing_count"], 0)
|
|
|
|
|
|
class TestB1StaleSnapshotRevalidation(_MissingWorktreeTestBase):
|
|
"""B1: safety-relevant drift between audit and apply must fail closed."""
|
|
|
|
def _apply(self, candidate):
|
|
return mwr.resolve_missing_worktree_binding(
|
|
self.db,
|
|
binding=candidate,
|
|
dry_run=False,
|
|
cleanup_authorization=self._mint_cleanup_authorization(),
|
|
project_root=self.project_root,
|
|
)
|
|
|
|
def test_lease_reactivated_between_audit_and_apply_fails_closed(self):
|
|
"""The reviewer's exact probe: released at audit, active at apply."""
|
|
lease_id, stale_path = self._stale_lease(number=501, session_id="sess-501")
|
|
candidate = self._candidate()
|
|
|
|
# Concurrent writer re-activates the same lease with a live process.
|
|
self._set_lease_columns(lease_id, status="active", owner_pid=os.getpid())
|
|
|
|
result = self._apply(candidate)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertEqual(result["reason"], "safety_state_changed")
|
|
self.assertIn("lease_status", result["changed_fields"])
|
|
# Nothing was mutated.
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_lease_owner_session_change_between_audit_and_apply_fails_closed(self):
|
|
lease_id, stale_path = self._stale_lease(number=502, session_id="sess-502")
|
|
candidate = self._candidate()
|
|
|
|
self.db.upsert_session(
|
|
session_id="someone-else-503", role="author", profile="prgs-author",
|
|
pid=os.getpid(),
|
|
)
|
|
self._set_lease_columns(lease_id, session_id="someone-else-503")
|
|
|
|
result = self._apply(candidate)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "safety_state_changed")
|
|
self.assertIn("lease_session_id", result["changed_fields"])
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_lease_owner_pid_change_between_audit_and_apply_fails_closed(self):
|
|
lease_id, stale_path = self._stale_lease(number=504, session_id="sess-504")
|
|
candidate = self._candidate()
|
|
|
|
self._set_lease_columns(lease_id, owner_pid=999_999_999)
|
|
|
|
result = self._apply(candidate)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "safety_state_changed")
|
|
self.assertIn("lease_owner_pid", result["changed_fields"])
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_binding_path_change_between_audit_and_apply_fails_closed(self):
|
|
lease_id, _ = self._stale_lease(number=505, session_id="sess-505")
|
|
candidate = self._candidate()
|
|
|
|
moved_target = os.path.join(self.temp_dir, "relocated_missing_505")
|
|
self._set_lease_columns(lease_id, worktree_path=moved_target)
|
|
|
|
result = self._apply(candidate)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "safety_state_changed")
|
|
self.assertIn("binding_path", result["changed_fields"])
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], moved_target)
|
|
|
|
def test_live_session_appearing_after_audit_fails_closed(self):
|
|
lease_id, stale_path = self._stale_lease(number=506, session_id="sess-506")
|
|
candidate = self._candidate()
|
|
|
|
# A session row for the owning session appears (or revives) after audit.
|
|
self.db.upsert_session(
|
|
session_id="sess-506", role="author", profile="prgs-author",
|
|
pid=os.getpid(),
|
|
)
|
|
|
|
result = self._apply(candidate)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "safety_state_changed")
|
|
self.assertTrue(
|
|
{"session_live", "session_active", "session_status", "session_pid_alive"}
|
|
& set(result["changed_fields"])
|
|
)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_binding_without_audited_snapshot_is_refused(self):
|
|
"""A caller cannot skip the drift check by omitting the snapshot."""
|
|
lease_id, stale_path = self._stale_lease(number=507, session_id="sess-507")
|
|
candidate = dict(self._candidate())
|
|
candidate.pop("audited_state", None)
|
|
|
|
result = self._apply(candidate)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "safety_state_changed")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_revalidation_reads_live_rows_not_audit_snapshot(self):
|
|
"""read_live_safety_state must come from the DB, not the candidate."""
|
|
lease_id, _ = self._stale_lease(number=508, session_id="sess-508")
|
|
candidate = dict(self._candidate())
|
|
# Lie in the snapshot: claim a status the DB never held.
|
|
candidate["lease_status"] = "active"
|
|
|
|
live = mwr.read_live_safety_state(self.db, candidate)
|
|
self.assertEqual(live["snapshot"]["lease_status"], "released")
|
|
self.assertEqual(live["lease_row"]["lease_id"], lease_id)
|
|
|
|
|
|
class TestB2ServerEnforcedCleanupAuthorization(_MissingWorktreeTestBase):
|
|
"""B2: authorization is a server artifact, never a client boolean."""
|
|
|
|
def test_operator_authorized_boolean_is_rejected_by_module(self):
|
|
self._stale_lease(number=601, session_id="sess-601")
|
|
candidate = self._candidate()
|
|
|
|
result = mwr.resolve_missing_worktree_binding(
|
|
self.db,
|
|
binding=candidate,
|
|
dry_run=False,
|
|
operator_authorized=True,
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "operator_authorized_rejected")
|
|
|
|
def test_apply_without_authorization_is_refused(self):
|
|
lease_id, stale_path = self._stale_lease(number=602, session_id="sess-602")
|
|
candidate = self._candidate()
|
|
|
|
result = mwr.resolve_missing_worktree_binding(
|
|
self.db,
|
|
binding=candidate,
|
|
dry_run=False,
|
|
cleanup_authorization=None,
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "authorization_required")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_forged_authorization_without_cleanup_phase_is_refused(self):
|
|
"""A hand-built authorization mapping cannot stand in for the gate."""
|
|
lease_id, stale_path = self._stale_lease(number=603, session_id="sess-603")
|
|
candidate = self._candidate()
|
|
|
|
forged = {
|
|
"authorized": True,
|
|
"source": mwr.CLEANUP_AUTHORIZATION_SOURCE,
|
|
"role": "reconciler",
|
|
"reasons": [],
|
|
}
|
|
arm.clear_phase() # the server never authorized a cleanup phase
|
|
result = mwr.resolve_missing_worktree_binding(
|
|
self.db,
|
|
binding=candidate,
|
|
dry_run=False,
|
|
cleanup_authorization=forged,
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "authorization_required")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_authorization_requires_reconciler_role(self):
|
|
arm.clear_phase()
|
|
arm.enter_audit_phase("reconcile_merged_cleanups")
|
|
arm.authorize_cleanup_phase(
|
|
operator_approved=True,
|
|
delete_capability_proven=True,
|
|
safety_proof={"safe_to_remove_worktree": True},
|
|
before_after_snapshot={"before": "a", "after": "b"},
|
|
)
|
|
auth = mwr.assess_cleanup_authorization(
|
|
role="author", capability_blockers=[], task_capability_resolved=True
|
|
)
|
|
self.assertFalse(auth["authorized"])
|
|
self.assertTrue(any("reconciler" in r for r in auth["reasons"]))
|
|
|
|
def test_authorization_requires_delete_capability(self):
|
|
arm.clear_phase()
|
|
arm.enter_audit_phase("reconcile_merged_cleanups")
|
|
arm.authorize_cleanup_phase(
|
|
operator_approved=True,
|
|
delete_capability_proven=True,
|
|
safety_proof={"safe_to_remove_worktree": True},
|
|
before_after_snapshot={"before": "a", "after": "b"},
|
|
)
|
|
auth = mwr.assess_cleanup_authorization(
|
|
role="reconciler",
|
|
capability_blockers=["profile forbids gitea.branch.delete"],
|
|
task_capability_resolved=True,
|
|
)
|
|
self.assertFalse(auth["authorized"])
|
|
|
|
def test_authorization_requires_resolved_task_capability(self):
|
|
arm.clear_phase()
|
|
arm.enter_audit_phase("reconcile_merged_cleanups")
|
|
arm.authorize_cleanup_phase(
|
|
operator_approved=True,
|
|
delete_capability_proven=True,
|
|
safety_proof={"safe_to_remove_worktree": True},
|
|
before_after_snapshot={"before": "a", "after": "b"},
|
|
)
|
|
auth = mwr.assess_cleanup_authorization(
|
|
role="reconciler", capability_blockers=[], task_capability_resolved=False
|
|
)
|
|
self.assertFalse(auth["authorized"])
|
|
self.assertTrue(any(mwr.CLEANUP_TASK in r for r in auth["reasons"]))
|
|
|
|
def test_operator_authorized_rejected_even_for_authorized_reconciler(self):
|
|
auth = mwr.assess_cleanup_authorization(
|
|
role="reconciler",
|
|
capability_blockers=[],
|
|
operator_authorized=True,
|
|
task_capability_resolved=True,
|
|
)
|
|
self.assertFalse(auth["authorized"])
|
|
self.assertTrue(auth["operator_authorized_rejected"])
|
|
|
|
def test_cleanup_task_capability_is_reconciler_only(self):
|
|
self.assertEqual(
|
|
task_capability_map.required_role(mwr.CLEANUP_TASK), "reconciler"
|
|
)
|
|
self.assertEqual(
|
|
task_capability_map.required_permission(mwr.CLEANUP_TASK),
|
|
mwr.CLEANUP_REQUIRED_PERMISSION,
|
|
)
|
|
self.assertIn(mwr.CLEANUP_TASK, task_capability_map.ROLE_EXCLUSIVE_TASKS)
|
|
|
|
|
|
class TestB2ProductionMcpBoundary(_MissingWorktreeTestBase):
|
|
"""B2 at the registered MCP tool — the real production entrypoint."""
|
|
|
|
CANONICAL_ORG = "Scaled-Tech-Consulting"
|
|
CANONICAL_REPO = "Gitea-Tools"
|
|
|
|
def _mcp_stale_lease(self, *, number, session_id):
|
|
"""A stale lease under the org/repo the tool resolves for 'prgs'."""
|
|
return self._stale_lease(
|
|
number=number,
|
|
session_id=session_id,
|
|
org=self.CANONICAL_ORG,
|
|
repo=self.CANONICAL_REPO,
|
|
)
|
|
|
|
def _call_reconcile(self, **kwargs):
|
|
# org/repo are passed explicitly: the 'prgs' remote defaults to the
|
|
# Timesheet repo, so relying on resolution would audit a different
|
|
# repository than the fixture wrote to.
|
|
kwargs.setdefault("org", self.CANONICAL_ORG)
|
|
kwargs.setdefault("repo", self.CANONICAL_REPO)
|
|
with patch.object(mcp_server, "_control_plane_db_or_error",
|
|
return_value=(self.db, [])), \
|
|
patch.object(mcp_server, "_canonical_local_git_root",
|
|
return_value=self.project_root):
|
|
return mcp_server.gitea_reconcile_missing_worktree_bindings(**kwargs)
|
|
|
|
def test_production_tools_are_registered(self):
|
|
for name in (
|
|
"gitea_audit_missing_worktree_bindings",
|
|
"gitea_reconcile_missing_worktree_bindings",
|
|
):
|
|
self.assertTrue(
|
|
callable(getattr(mcp_server, name, None)),
|
|
f"{name} is not registered in the production MCP module",
|
|
)
|
|
|
|
@patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)
|
|
def test_reviewer_with_operator_authorized_true_is_rejected(self, _profile):
|
|
lease_id, stale_path = self._mcp_stale_lease(number=701, session_id="sess-701")
|
|
|
|
result = self._call_reconcile(
|
|
dry_run=False, operator_authorized=True, remote="prgs"
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "operator_authorized_rejected")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
@patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)
|
|
def test_reviewer_apply_without_cleanup_capability_is_rejected(self, _profile):
|
|
lease_id, stale_path = self._mcp_stale_lease(number=702, session_id="sess-702")
|
|
|
|
result = self._call_reconcile(dry_run=False, remote="prgs")
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "cleanup_authorization_required")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
@patch.object(mcp_server, "get_profile", return_value=AUTHOR_WITH_DELETE_PROFILE)
|
|
def test_wrong_role_with_delete_permission_is_rejected(self, _profile):
|
|
"""Permission alone must not authorize: the role gate must hold."""
|
|
lease_id, stale_path = self._mcp_stale_lease(number=703, session_id="sess-703")
|
|
|
|
result = self._call_reconcile(dry_run=False, remote="prgs")
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "cleanup_authorization_required")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
@patch.object(mcp_server, "get_profile", return_value=RECONCILER_PROFILE)
|
|
def test_reconciler_apply_without_authorized_cleanup_phase_is_rejected(self, _profile):
|
|
"""Correct role and capability, but no minted cleanup phase."""
|
|
lease_id, stale_path = self._mcp_stale_lease(number=704, session_id="sess-704")
|
|
arm.clear_phase()
|
|
|
|
result = self._call_reconcile(dry_run=False, remote="prgs")
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["reason"], "cleanup_authorization_required")
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
@patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)
|
|
def test_dry_run_is_allowed_and_non_mutating_for_read_profile(self, _profile):
|
|
lease_id, stale_path = self._mcp_stale_lease(number=705, session_id="sess-705")
|
|
|
|
result = self._call_reconcile(dry_run=True, remote="prgs")
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(result["dry_run"])
|
|
self.assertEqual(result["stale_candidates_count"], 1)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
@patch.object(mcp_server, "get_profile", return_value=RECONCILER_PROFILE)
|
|
def test_authorized_reconciler_apply_retires_binding(self, _profile):
|
|
lease_id, _ = self._mcp_stale_lease(number=706, session_id="sess-706")
|
|
arm.clear_phase()
|
|
arm.enter_audit_phase("reconcile_merged_cleanups")
|
|
arm.authorize_cleanup_phase(
|
|
operator_approved=True,
|
|
delete_capability_proven=True,
|
|
safety_proof={"safe_to_remove_worktree": True},
|
|
before_after_snapshot={"before": "bound", "after": "retired"},
|
|
)
|
|
|
|
result = self._call_reconcile(dry_run=False, remote="prgs")
|
|
self.assertTrue(result["success"], result)
|
|
self.assertTrue(result["worktrees_dimension_resolved"])
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], "")
|
|
|
|
@patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)
|
|
def test_audit_tool_is_read_only(self, _profile):
|
|
lease_id, stale_path = self._mcp_stale_lease(number=707, session_id="sess-707")
|
|
with patch.object(mcp_server, "_control_plane_db_or_error",
|
|
return_value=(self.db, [])), \
|
|
patch.object(mcp_server, "_canonical_local_git_root",
|
|
return_value=self.project_root):
|
|
audit = mcp_server.gitea_audit_missing_worktree_bindings(
|
|
remote="prgs", org=self.CANONICAL_ORG, repo=self.CANONICAL_REPO
|
|
)
|
|
self.assertEqual(audit["confirmed_stale_count"], 1)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
|
|
class TestB3ExpectedPathCompareAndSwap(_MissingWorktreeTestBase):
|
|
"""B3: both retirement paths enforce the expected stored path."""
|
|
|
|
def _checkpoint(self, *, session_id="cp-sess", worktree_path, number=800):
|
|
self.db.write_session_checkpoint(
|
|
remote="prgs", org="org", repo="repo",
|
|
session_id=session_id, role="author",
|
|
work_kind="issue", work_number=number,
|
|
worktree_path=worktree_path, branch=f"fix/issue-{number}-x",
|
|
)
|
|
rows = self.db.list_session_checkpoints(session_id=session_id)
|
|
self.assertEqual(len(rows), 1)
|
|
return rows[0]
|
|
|
|
def test_checkpoint_retirement_refuses_wrong_expected_path(self):
|
|
"""The reviewer's probe: wrong expected_path returned retired=True."""
|
|
missing = os.path.join(self.temp_dir, "cp_missing_801")
|
|
row = self._checkpoint(worktree_path=missing, number=801)
|
|
|
|
with self.assertRaises(cpd.ControlPlaneError):
|
|
self.db.retire_session_checkpoint_worktree_path(
|
|
session_id="cp-sess",
|
|
checkpoint_id=row["checkpoint_id"],
|
|
expected_path="/some/other/path",
|
|
)
|
|
|
|
after = self.db.list_session_checkpoints(session_id="cp-sess")[0]
|
|
self.assertEqual(after["worktree_path"], missing)
|
|
|
|
def test_checkpoint_retirement_requires_expected_path(self):
|
|
missing = os.path.join(self.temp_dir, "cp_missing_802")
|
|
row = self._checkpoint(worktree_path=missing, number=802)
|
|
|
|
with self.assertRaises(cpd.ControlPlaneError):
|
|
self.db.retire_session_checkpoint_worktree_path(
|
|
session_id="cp-sess", checkpoint_id=row["checkpoint_id"]
|
|
)
|
|
after = self.db.list_session_checkpoints(session_id="cp-sess")[0]
|
|
self.assertEqual(after["worktree_path"], missing)
|
|
|
|
def test_checkpoint_retirement_refuses_unknown_row(self):
|
|
with self.assertRaises(cpd.ControlPlaneError):
|
|
self.db.retire_session_checkpoint_worktree_path(
|
|
session_id="no-such-session", expected_path="/tmp/whatever"
|
|
)
|
|
|
|
def test_checkpoint_retirement_succeeds_on_exact_path(self):
|
|
missing = os.path.join(self.temp_dir, "cp_missing_803")
|
|
row = self._checkpoint(worktree_path=missing, number=803)
|
|
|
|
proof = self.db.retire_session_checkpoint_worktree_path(
|
|
session_id="cp-sess",
|
|
checkpoint_id=row["checkpoint_id"],
|
|
expected_path=missing,
|
|
)
|
|
self.assertTrue(proof["retired"])
|
|
after = self.db.list_session_checkpoints(session_id="cp-sess")[0]
|
|
self.assertEqual(after["worktree_path"], "")
|
|
|
|
# Idempotent: a second retirement is a no-op, not a failure.
|
|
again = self.db.retire_session_checkpoint_worktree_path(
|
|
session_id="cp-sess",
|
|
checkpoint_id=row["checkpoint_id"],
|
|
expected_path=missing,
|
|
)
|
|
self.assertFalse(again["retired"])
|
|
self.assertTrue(again["already_retired"])
|
|
|
|
def test_lease_retirement_refuses_wrong_expected_path(self):
|
|
lease_id, stale_path = self._stale_lease(number=804, session_id="sess-804")
|
|
with self.assertRaises(cpd.ControlPlaneError):
|
|
self.db.retire_lease_worktree_path(
|
|
lease_id, expected_path="/not/the/stored/path"
|
|
)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_lease_retirement_refuses_changed_status(self):
|
|
lease_id, stale_path = self._stale_lease(number=805, session_id="sess-805")
|
|
with self.assertRaises(cpd.ControlPlaneError):
|
|
self.db.retire_lease_worktree_path(
|
|
lease_id, expected_path=stale_path, expected_status="active"
|
|
)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_lease_retirement_refuses_changed_session(self):
|
|
lease_id, stale_path = self._stale_lease(number=806, session_id="sess-806")
|
|
with self.assertRaises(cpd.ControlPlaneError):
|
|
self.db.retire_lease_worktree_path(
|
|
lease_id,
|
|
expected_path=stale_path,
|
|
expected_session_id="a-different-session",
|
|
)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_lease_retirement_requires_expected_path(self):
|
|
lease_id, stale_path = self._stale_lease(number=807, session_id="sess-807")
|
|
with self.assertRaises(cpd.ControlPlaneError):
|
|
self.db.retire_lease_worktree_path(lease_id)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
|
|
class TestB4OwnershipEvidence(_MissingWorktreeTestBase):
|
|
"""B4: live-session, trustworthy-owner, and issue-lock evidence."""
|
|
|
|
def test_active_lease_with_dead_pid_is_not_confirmed_stale(self):
|
|
"""Dead-PID evidence alone must not retire a lease still held."""
|
|
dead_pid = 999_999_999
|
|
self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=901,
|
|
role="author", session_id="sess-901",
|
|
worktree_path=os.path.join(self.temp_dir, "missing_901"),
|
|
owner_pid=dead_pid,
|
|
)
|
|
audit = self._audit()
|
|
self.assertEqual(audit["confirmed_stale_count"], 0)
|
|
self.assertEqual(
|
|
audit["missing_bindings"][0]["classification"],
|
|
mwr.CLASS_LIVE_LEASE_PROTECTED,
|
|
)
|
|
|
|
def test_session_active_is_populated_from_the_sessions_table(self):
|
|
self.db.assign_and_lease(
|
|
remote="prgs", org="org", repo="repo", kind="issue", number=902,
|
|
role="author", session_id="sess-902",
|
|
worktree_path=os.path.join(self.temp_dir, "missing_902"),
|
|
owner_pid=os.getpid(),
|
|
)
|
|
evidence = mwr.collect_session_evidence(self.db)
|
|
self.assertIn("sess-902", evidence)
|
|
self.assertTrue(evidence["sess-902"]["session_active"])
|
|
self.assertTrue(evidence["sess-902"]["session_pid_alive"])
|
|
|
|
binding = self._audit()["missing_bindings"][0]
|
|
self.assertTrue(binding["session_active"])
|
|
|
|
def test_live_session_blocks_cleanup_of_released_lease(self):
|
|
lease_id, stale_path = self._stale_lease(number=903, session_id="sess-903")
|
|
# The lease is released, but its session is live.
|
|
self.db.upsert_session(
|
|
session_id="sess-903", role="author", profile="prgs-author",
|
|
pid=os.getpid(),
|
|
)
|
|
audit = self._audit()
|
|
self.assertEqual(audit["confirmed_stale_count"], 0)
|
|
self.assertEqual(
|
|
audit["missing_bindings"][0]["classification"],
|
|
mwr.CLASS_LIVE_SESSION_PROTECTED,
|
|
)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_trusted_owner_evidence_without_live_process_blocks_cleanup(self):
|
|
"""Session recorded active, process gone: still a valid owner signal.
|
|
|
|
No terminal lease contradicts the session record here (a checkpoint-style
|
|
binding), so the recorded owner is the best evidence available and must
|
|
block cleanup even though nothing is running.
|
|
"""
|
|
result = mwr.classify_worktree_binding(
|
|
recorded_path=os.path.join(self.temp_dir, "missing_904"),
|
|
owner_pid=999_999_999,
|
|
session_id="sess-904",
|
|
session_active=True,
|
|
session_pid_alive=False,
|
|
session_status="active",
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertEqual(result["classification"], mwr.CLASS_TRUSTED_OWNER_PROTECTED)
|
|
self.assertFalse(result["retire_eligible"])
|
|
|
|
def test_released_lease_outranks_a_lingering_active_session_row(self):
|
|
"""An explicit release is later and more specific than a session row.
|
|
|
|
Session rows are opened on assignment and never closed on release, so
|
|
treating one as protective would make every released lease permanently
|
|
unretireable — the exact degradation #970 exists to clear.
|
|
"""
|
|
result = mwr.classify_worktree_binding(
|
|
recorded_path=os.path.join(self.temp_dir, "missing_908"),
|
|
lease_status="released",
|
|
owner_pid=999_999_999,
|
|
session_id="sess-908",
|
|
session_active=True,
|
|
session_pid_alive=False,
|
|
session_status="active",
|
|
project_root=self.project_root,
|
|
)
|
|
self.assertEqual(result["classification"], mwr.CLASS_CONFIRMED_STALE_DELETED)
|
|
self.assertTrue(result["retire_eligible"])
|
|
|
|
def test_checkpoint_binding_with_active_session_is_protected(self):
|
|
"""A checkpoint has no lease to contradict its owning session."""
|
|
missing = os.path.join(self.temp_dir, "cp_missing_909")
|
|
self.db.write_session_checkpoint(
|
|
remote="prgs", org="org", repo="repo",
|
|
session_id="sess-909", role="author",
|
|
work_kind="issue", work_number=909,
|
|
worktree_path=missing, branch="fix/issue-909-x",
|
|
)
|
|
self.db.upsert_session(
|
|
session_id="sess-909", role="author", profile="prgs-author",
|
|
pid=999_999_999,
|
|
)
|
|
audit = self._audit()
|
|
checkpoint_bindings = [
|
|
b for b in audit["missing_bindings"]
|
|
if b.get("source") == "session_checkpoint"
|
|
]
|
|
self.assertEqual(len(checkpoint_bindings), 1)
|
|
self.assertEqual(
|
|
checkpoint_bindings[0]["classification"],
|
|
mwr.CLASS_TRUSTED_OWNER_PROTECTED,
|
|
)
|
|
|
|
def test_conflicting_issue_lock_blocks_cleanup(self):
|
|
stale_path = os.path.join(self.temp_dir, "missing_905")
|
|
lease_id, _ = self._stale_lease(
|
|
number=905, path=stale_path, session_id="sess-905"
|
|
)
|
|
self._write_issue_lock(
|
|
issue_number=905,
|
|
worktree_path=stale_path,
|
|
branch="fix/issue-905-x",
|
|
)
|
|
|
|
audit = self._audit()
|
|
self.assertEqual(audit["confirmed_stale_count"], 0)
|
|
self.assertEqual(
|
|
audit["missing_bindings"][0]["classification"],
|
|
mwr.CLASS_ISSUE_LOCK_PROTECTED,
|
|
)
|
|
self.assertEqual(self._lease_row(lease_id)["worktree_path"], stale_path)
|
|
|
|
def test_issue_lock_missing_path_is_reported_but_never_retired(self):
|
|
lock_path = os.path.join(self.temp_dir, "missing_906")
|
|
self._write_issue_lock(
|
|
issue_number=906, worktree_path=lock_path, branch="fix/issue-906-x"
|
|
)
|
|
audit = self._audit()
|
|
self.assertEqual(audit["issue_lock_missing_count"], 1)
|
|
reported = audit["issue_lock_missing_bindings"][0]
|
|
self.assertFalse(reported["retire_eligible"])
|
|
self.assertEqual(reported["issue_number"], 906)
|
|
|
|
def test_issue_lock_with_dead_owner_does_not_block_unrelated_binding(self):
|
|
"""A lock is only protective while it is live: no over-blocking."""
|
|
stale_path = os.path.join(self.temp_dir, "missing_907")
|
|
self._stale_lease(number=907, path=stale_path, session_id="sess-907")
|
|
self._write_issue_lock(
|
|
issue_number=907,
|
|
worktree_path=os.path.join(self.temp_dir, "other_missing_907"),
|
|
branch="fix/issue-907-other",
|
|
pid=999_999_999,
|
|
)
|
|
audit = self._audit()
|
|
self.assertEqual(audit["confirmed_stale_count"], 1)
|
|
self.assertEqual(
|
|
audit["confirmed_stale_candidates"][0]["worktree_path"], stale_path
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|