"""Central lease policy and load-bearing heartbeat freshness (#790 Slice A). Before this slice, ``issue_lock_store.assess_lock_freshness`` parsed ``last_heartbeat_at`` and then never consulted it: liveness was decided by an absolute four-hour ``expires_at`` and by PID liveness. Because the recorded PID is the long-lived MCP daemon rather than the authoring task, an abandoned claim stayed "live" for the full four hours, and a claim whose work had already landed blocked reconciliation for just as long (Issue #787 / PR #789, and again Issue #760 / PR #791). These tests pin the corrected semantics, including the two asymmetries that are easy to lose in a refactor: * an **alive** PID must never make anything live (AC-N2), while * a **dead** PID must still mark a lease stale, because #753 dead-session recovery keys on exactly that classification. Durable-state helpers here write real lock files through the real flock path; they are not mocks of the store. """ from __future__ import annotations import os import sys import tempfile import unittest from datetime import datetime, timedelta, timezone from unittest.mock import patch sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import issue_lock_store # noqa: E402 import lease_policy # noqa: E402 import pr_work_lease # noqa: E402 import reviewer_pr_lease # noqa: E402 ISSUE = 9790 BRANCH = f"fix/issue-{ISSUE}-heartbeat" IDENTITY = "example-user" PROFILE = "test-author-prgs" ORG = "Example-Org" REPO = "Example-Repo" REMOTE = "prgs" DEAD_PID = 2**22 # far above any live pid on a test host def _ts(moment: datetime) -> str: return ( moment.astimezone(timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) class _LockFixture(unittest.TestCase): def setUp(self): self.lock_dir = tempfile.TemporaryDirectory() self.addCleanup(self.lock_dir.cleanup) self.now = datetime.now(timezone.utc) self.worktree = os.path.realpath(tempfile.mkdtemp(prefix="issue790-")) self.addCleanup(patch.stopall) def _path(self): return issue_lock_store.lock_file_path( remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=self.lock_dir.name, ) def write_lock( self, *, lifecycle: str | None = lease_policy.LIFECYCLE_HEARTBEAT_V1, created_delta: timedelta = timedelta(minutes=1), heartbeat_delta: timedelta = timedelta(minutes=1), expires_delta: timedelta = timedelta(minutes=9), pid: int | None = None, task_session_id: str | None = "author_issue_work-aaaabbbbccccdddd", generation: int = 1, identity: str = IDENTITY, profile: str = PROFILE, branch: str = BRANCH, worktree: str | None = None, ) -> dict: """Write a real durable lock and return the record. Deltas are relative to ``self.now``; ``expires_delta`` is added, the others subtracted, so "in the past" reads naturally at each call site. """ lease: dict = { "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE, "issue_number": ISSUE, "pr_number": None, "branch": branch, "worktree_path": worktree or self.worktree, "claimant": {"username": identity, "profile": profile}, "created_at": _ts(self.now - created_delta), "last_heartbeat_at": _ts(self.now - heartbeat_delta), "expires_at": _ts(self.now + expires_delta), } if lifecycle is not None: lease["lifecycle_version"] = lifecycle if task_session_id is not None: lease["task_session_id"] = task_session_id pid_value = os.getpid() if pid is None else pid record = { "issue_number": ISSUE, "branch_name": branch, "remote": REMOTE, "org": ORG, "repo": REPO, "worktree_path": worktree or self.worktree, "session_pid": pid_value, "pid": pid_value, "lock_generation": generation, "work_lease": lease, } path = self._path() record["lock_file_path"] = path issue_lock_store.save_lock_file(path, record) return record class TestPolicyIsTheSingleSource(unittest.TestCase): """AC-N7: one authoritative configuration source for every duration.""" def test_author_policy_carries_the_agreed_values(self): policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK) self.assertEqual(policy.initial_ttl_minutes, 10.0) self.assertEqual(policy.heartbeat_cadence_minutes, 2.0) self.assertEqual(policy.stale_warning_minutes, 5.0) self.assertEqual(policy.missed_heartbeat_grace_minutes, 10.0) self.assertEqual(policy.absolute_cap_hours, 8.0) self.assertEqual(policy.recovery_grace_minutes, 10.0) self.assertEqual(policy.terminal_race_drain_minutes, 2.0) self.assertTrue(policy.terminal_retirement_eligible) self.assertTrue(policy.heartbeat_lifecycle_active) def test_the_four_hour_author_ttl_literal_is_gone(self): """The duplicated literal AC-N7 exists to remove.""" self.assertFalse(hasattr(issue_lock_store, "WORK_LEASE_TTL_HOURS")) import gitea_mcp_server self.assertFalse(hasattr(gitea_mcp_server, "WORK_LEASE_TTL_HOURS")) def test_declared_reviewer_values_match_the_module_still_using_them(self): """Slice A declares reviewer/merger numbers without rewiring them. Recording a value in two places is only safe if drift is detectable, so this asserts the declaration still equals the constants #747 owns. When Slice C migrates those call sites, this test becomes the proof the migration changed nothing. """ policy = lease_policy.policy_for(lease_policy.TASK_CLASS_REVIEWER_PR) self.assertEqual( policy.initial_ttl_minutes, float(reviewer_pr_lease.LEASE_TTL_MINUTES) ) self.assertEqual( policy.stale_warning_minutes, float(reviewer_pr_lease.STALE_WARNING_MINUTES), ) self.assertFalse(policy.heartbeat_lifecycle_active) def test_declared_conflict_fix_value_matches_its_module(self): policy = lease_policy.policy_for(lease_policy.TASK_CLASS_CONFLICT_FIX) self.assertEqual( policy.initial_ttl_minutes, float(pr_work_lease.DEFAULT_CONFLICT_FIX_TTL_MINUTES), ) self.assertFalse(policy.heartbeat_lifecycle_active) def test_environment_override_applies(self): var = lease_policy.env_var_name( lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes" ) with patch.dict(os.environ, {var: "7"}): self.assertEqual( lease_policy.policy_for( lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK ).initial_ttl_minutes, 7.0, ) def test_unusable_override_falls_back_instead_of_minting_a_zero_lease(self): """A typo must not make every claim instantly reclaimable.""" var = lease_policy.env_var_name( lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK, "initial_ttl_minutes" ) for bad in ("0", "-5", "not-a-number", " "): with self.subTest(value=bad), patch.dict(os.environ, {var: bad}): self.assertEqual( lease_policy.policy_for( lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK ).initial_ttl_minutes, 10.0, ) def test_unknown_task_class_does_not_raise(self): policy = lease_policy.policy_for("something-new") self.assertEqual(policy.task_class, lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK) class TestLifecycleDiscrimination(_LockFixture): """AC-N8: the marker, never a timestamp, decides legacy vs heartbeat.""" def test_missing_marker_reads_as_legacy(self): record = self.write_lock(lifecycle=None) self.assertTrue(issue_lock_store.is_legacy_lease(record)) self.assertEqual( issue_lock_store.lease_lifecycle_version(record), lease_policy.LIFECYCLE_LEGACY, ) def test_marker_present_reads_as_heartbeat_lifecycle(self): record = self.write_lock() self.assertFalse(issue_lock_store.is_legacy_lease(record)) def test_equal_created_and_heartbeat_never_implies_a_fresh_heartbeat(self): """The exact inversion AC-N8 forbids. A legacy lock has ``last_heartbeat_at == created_at`` forever because nothing ever advanced it. Reading that equality as "recently heartbeated" would classify every never-heartbeated lock as fresh. """ legacy = self.write_lock( lifecycle=None, created_delta=timedelta(hours=3), heartbeat_delta=timedelta(hours=3), ) lease = legacy["work_lease"] self.assertEqual(lease["created_at"], lease["last_heartbeat_at"]) self.assertTrue(issue_lock_store.is_legacy_lease(legacy)) # A brand-new heartbeat lease has them equal too, so the equality # carries no information in either direction. fresh = self.write_lock( created_delta=timedelta(seconds=0), heartbeat_delta=timedelta(seconds=0) ) self.assertEqual( fresh["work_lease"]["created_at"], fresh["work_lease"]["last_heartbeat_at"], ) self.assertFalse(issue_lock_store.is_legacy_lease(fresh)) def test_minted_session_id_contains_no_pid(self): """AC-N1: the ownership key must not be derived from the daemon pid.""" minted = issue_lock_store.mint_task_session_id() self.assertNotIn(str(os.getpid()), minted) self.assertNotEqual(minted, issue_lock_store.mint_task_session_id()) class TestFreshnessIsHeartbeatDriven(_LockFixture): """AC-N2 and the new bands.""" def test_fresh_heartbeat_is_live(self): record = self.write_lock(heartbeat_delta=timedelta(minutes=1)) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE) self.assertTrue(fresh["live"]) self.assertFalse(fresh["heartbeat_warning"]) def test_heartbeat_past_warning_is_still_live_but_flagged(self): record = self.write_lock(heartbeat_delta=timedelta(minutes=6)) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE) self.assertTrue(fresh["heartbeat_warning"]) def test_missed_heartbeat_past_grace_is_classified_explicitly(self): record = self.write_lock( heartbeat_delta=timedelta(minutes=11), expires_delta=timedelta(minutes=30), ) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual( fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT ) self.assertFalse(fresh["live"]) self.assertTrue(fresh["stale"]) def test_alive_pid_never_establishes_freshness(self): """The defect in one assertion. The recorded PID is this very process, so it is unambiguously alive — and the lease is still not live, because the task stopped heartbeating. """ record = self.write_lock( pid=os.getpid(), heartbeat_delta=timedelta(hours=4), expires_delta=timedelta(hours=4), ) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertTrue(fresh["pid_alive"]) self.assertFalse(fresh["live"]) self.assertEqual( fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT ) def test_dead_pid_still_marks_stale_for_issue_753(self): """The opposite asymmetry: dead-PID corroboration is preserved.""" record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1)) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE) self.assertFalse(fresh["live"]) self.assertIn("not alive", fresh["reason"]) def test_absolute_cap_requires_readoption(self): record = self.write_lock( created_delta=timedelta(hours=9), heartbeat_delta=timedelta(minutes=1) ) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual(fresh["status"], issue_lock_store.STATUS_STALE_ABSOLUTE_CAP) self.assertIn("re-adoption", fresh["reason"]) def test_heartbeat_lifecycle_without_a_heartbeat_fails_closed(self): record = self.write_lock() del record["work_lease"]["last_heartbeat_at"] issue_lock_store.save_lock_file(self._path(), record) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual( fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT ) self.assertIn("fail closed", fresh["reason"]) def test_absent_lock(self): fresh = issue_lock_store.assess_lock_freshness(None) self.assertEqual(fresh["status"], issue_lock_store.STATUS_ABSENT) self.assertFalse(fresh["stale"]) class TestLegacyLocksStayProtected(_LockFixture): """AC-N8: deployment must not retroactively shorten an existing claim.""" def test_legacy_lock_with_a_stale_heartbeat_remains_live(self): """The deployment-safety case. A four-hour legacy lease minted three hours ago has not heartbeated once. Under the new grace it would be long gone; under its preserved absolute expiry it is still live, and must stay that way. """ record = self.write_lock( lifecycle=None, created_delta=timedelta(hours=3), heartbeat_delta=timedelta(hours=3), expires_delta=timedelta(hours=1), ) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE) self.assertTrue(fresh["live"]) self.assertTrue(fresh["legacy_lease"]) self.assertTrue(fresh["legacy_expiry_preserved"]) def test_legacy_lock_past_its_absolute_expiry_is_expired_as_before(self): record = self.write_lock( lifecycle=None, created_delta=timedelta(hours=5), heartbeat_delta=timedelta(hours=5), expires_delta=timedelta(hours=-1), ) fresh = issue_lock_store.assess_lock_freshness(record, now=self.now) self.assertEqual(fresh["status"], issue_lock_store.STATUS_EXPIRED) def test_legacy_lock_is_never_reclaimed_by_the_heartbeat_band(self): record = self.write_lock( lifecycle=None, created_delta=timedelta(hours=3), heartbeat_delta=timedelta(hours=3), expires_delta=timedelta(hours=1), ) reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now) self.assertFalse(reclaim["reclaim_allowed"]) class TestReclaimAfterMissedHeartbeat(_LockFixture): def test_missed_heartbeat_makes_ownership_reclaimable(self): record = self.write_lock( pid=os.getpid(), heartbeat_delta=timedelta(minutes=15), expires_delta=timedelta(hours=3), ) reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now) self.assertTrue(reclaim["reclaim_allowed"]) self.assertIn("stale_missed_heartbeat", reclaim["reasons"][0]) def test_live_lease_is_never_reclaimable(self): record = self.write_lock(heartbeat_delta=timedelta(minutes=1)) reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now) self.assertFalse(reclaim["reclaim_allowed"]) def test_dead_pid_reclaim_path_is_unchanged(self): """#753 must keep working through its original conditions.""" record = self.write_lock(pid=DEAD_PID, heartbeat_delta=timedelta(minutes=1)) reclaim = issue_lock_store.assess_expired_lock_reclaim(record, now=self.now) self.assertTrue(reclaim["reclaim_allowed"]) self.assertTrue(reclaim["owner_pid_dead"]) class TestHeartbeatWriter(_LockFixture): """A4: flock + CAS + exact verification, and no revival path.""" def _heartbeat(self, **kwargs): params = { "remote": REMOTE, "org": ORG, "repo": REPO, "issue_number": ISSUE, "branch_name": BRANCH, "worktree_path": self.worktree, "identity": IDENTITY, "profile": PROFILE, "task_session_id": "author_issue_work-aaaabbbbccccdddd", "lock_dir": self.lock_dir.name, "now": self.now, } params.update(kwargs) return issue_lock_store.heartbeat_session_lock(**params) def test_heartbeat_slides_expiry_and_advances_generation(self): self.write_lock(heartbeat_delta=timedelta(minutes=4), generation=5) result = self._heartbeat() self.assertTrue(result["success"], result) self.assertEqual(result["prior_generation"], 5) self.assertEqual(result["lock_generation"], 6) self.assertEqual(result["heartbeat_count"], 1) self.assertEqual(result["last_heartbeat_at"], _ts(self.now)) self.assertEqual(result["expires_at"], _ts(self.now + timedelta(minutes=10))) self.assertTrue(result["freshness"]["live"]) def test_heartbeat_is_durable_and_repeatable(self): self.write_lock(heartbeat_delta=timedelta(minutes=4)) self._heartbeat() second = self._heartbeat(now=self.now + timedelta(minutes=1)) self.assertTrue(second["success"], second) self.assertEqual(second["heartbeat_count"], 2) written = issue_lock_store.read_lock_file(self._path()) self.assertEqual(written["work_lease"]["heartbeat_count"], 2) def test_stale_generation_is_refused(self): self.write_lock(generation=5) result = self._heartbeat(expected_generation=4) self.assertFalse(result["success"]) self.assertIn("generation changed", result["reasons"][0]) def test_foreign_session_is_refused(self): self.write_lock() result = self._heartbeat(task_session_id="author_issue_work-ffffffffffffffff") self.assertFalse(result["success"]) self.assertIn("task_session_id does not match", " ".join(result["reasons"])) def test_missing_session_id_is_refused(self): self.write_lock() result = self._heartbeat(task_session_id="") self.assertFalse(result["success"]) def test_foreign_claimant_is_refused(self): self.write_lock() for field, value in ( ("identity", "someone-else"), ("profile", "other-profile"), ): with self.subTest(field=field): result = self._heartbeat(**{field: value}) self.assertFalse(result["success"]) def test_branch_and_worktree_mismatch_are_refused(self): self.write_lock() wrong_branch = self._heartbeat(branch_name=f"fix/issue-{ISSUE}-other") self.assertFalse(wrong_branch["success"]) wrong_worktree = self._heartbeat(worktree_path="/tmp/not-the-worktree") self.assertFalse(wrong_worktree["success"]) def test_lapsed_lease_cannot_be_heartbeated_back_to_life(self): """No revival path (A4). A session that stopped proving liveness must reclaim under a fresh generation, not restore ownership retroactively. """ self.write_lock( heartbeat_delta=timedelta(minutes=30), expires_delta=timedelta(hours=1) ) result = self._heartbeat() self.assertFalse(result["success"]) self.assertIn("reclaimed", " ".join(result["reasons"])) def test_absent_lock_cannot_be_created_by_heartbeat(self): result = self._heartbeat() self.assertFalse(result["success"]) self.assertIn("no durable lock", result["reasons"][0]) def test_legacy_lock_is_refused_until_rebound(self): self.write_lock(lifecycle=None) result = self._heartbeat() self.assertFalse(result["success"]) self.assertTrue(result["legacy_lease"]) self.assertIn("rebound", " ".join(result["reasons"])) class TestLegacyRebind(_LockFixture): """AC-N8 exit route: canonical exact-owner rebinding.""" def _rebind(self, **kwargs): params = { "remote": REMOTE, "org": ORG, "repo": REPO, "issue_number": ISSUE, "branch_name": BRANCH, "worktree_path": self.worktree, "identity": IDENTITY, "profile": PROFILE, "lock_dir": self.lock_dir.name, "now": self.now, } params.update(kwargs) return issue_lock_store.rebind_legacy_lock(**params) def test_rebind_mints_a_session_and_a_genuine_first_heartbeat(self): self.write_lock( lifecycle=None, created_delta=timedelta(hours=3), heartbeat_delta=timedelta(hours=3), expires_delta=timedelta(hours=1), generation=2, ) result = self._rebind() self.assertTrue(result["success"], result) self.assertTrue(result["task_session_id"]) self.assertEqual(result["lock_generation"], 3) written = issue_lock_store.read_lock_file(self._path()) lease = written["work_lease"] self.assertEqual( lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1 ) self.assertEqual(lease["last_heartbeat_at"], _ts(self.now)) self.assertEqual(lease["expires_at"], _ts(self.now + timedelta(minutes=10))) self.assertFalse(issue_lock_store.is_legacy_lease(written)) # The original claim is preserved for audit rather than overwritten. origin = written["legacy_rebind"]["legacy_origin"] self.assertTrue(origin["created_at"]) self.assertEqual(origin["lifecycle"], lease_policy.LIFECYCLE_LEGACY) def test_rebound_lock_can_then_heartbeat(self): self.write_lock( lifecycle=None, created_delta=timedelta(hours=3), heartbeat_delta=timedelta(hours=3), expires_delta=timedelta(hours=1), ) rebound = self._rebind() beat = issue_lock_store.heartbeat_session_lock( remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree, identity=IDENTITY, profile=PROFILE, task_session_id=rebound["task_session_id"], lock_dir=self.lock_dir.name, now=self.now + timedelta(minutes=1), ) self.assertTrue(beat["success"], beat) def test_rebind_refuses_a_foreign_owner(self): self.write_lock(lifecycle=None, expires_delta=timedelta(hours=1)) result = self._rebind(identity="someone-else") self.assertFalse(result["success"]) def test_rebind_refuses_a_lock_already_on_the_lifecycle(self): self.write_lock() result = self._rebind() self.assertFalse(result["success"]) self.assertFalse(result["legacy_lease"]) def test_rebind_is_not_a_recovery_path_for_a_lapsed_legacy_lease(self): """An expired legacy lease belongs to #760 renewal or #601 reclaim.""" self.write_lock( lifecycle=None, created_delta=timedelta(hours=5), heartbeat_delta=timedelta(hours=5), expires_delta=timedelta(hours=-1), ) result = self._rebind() self.assertFalse(result["success"]) self.assertIn("not a recovery path", " ".join(result["reasons"])) if __name__ == "__main__": unittest.main()