"""Conflict gate honors heartbeat-lifecycle non-live reclaim bands (#790 review #502/#516). ``assess_expired_lock_reclaim`` already permits reclaim for the heartbeat-lifecycle stale bands (``stale_missed_heartbeat``, ``stale_absolute_cap``) without a dead PID. But ``assess_same_issue_lease_conflict`` used to enter that reclaim branch only under ``is_lease_expired`` (``expires_at <= now``). For a heartbeat-lifecycle lease that is non-live yet whose ``expires_at`` is still in the future, the acquisition gate fell through to the foreign "already has an active lease" block and never consulted the reclaim assessor — so the load-bearing heartbeat was not load-bearing for foreign reclaim, the exact abandonment scenario #790 exists to fix. Two future-``expires_at`` non-live shapes are reachable: * ``stale_absolute_cap`` — a session that keeps heartbeating past the 8h absolute cap has ``expires_at = last_heartbeat + TTL`` in the future (default policy). * ``stale_missed_heartbeat`` — under an independent TTL>grace policy the heartbeat grace lapses while ``expires_at`` is still ahead. These tests pin: both reclaim from a foreign acquirer; a live lease still blocks a foreign acquirer; the same owner may reclaim its own abandoned heartbeat lease; and a legacy (pre-lifecycle) lock keeps its absolute-``expires_at`` clock — a non-expired legacy lock with a dead PID is *not* reclaimable through this path. """ from __future__ import annotations import os import sys import unittest from datetime import datetime, timedelta, timezone from unittest import mock sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import issue_lock_store as ils # noqa: E402 import lease_policy # noqa: E402 ISSUE = 790 OWNER_BRANCH = f"fix/issue-{ISSUE}-slice-a-heartbeat-policy" OWNER_WORKTREE = "/tmp/wt-790-owner" FOREIGN_BRANCH = f"fix/issue-{ISSUE}-foreign-attempt" FOREIGN_WORKTREE = "/tmp/wt-790-foreign" def _ts(moment: datetime) -> str: return ( moment.astimezone(timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) class _ConflictGateBase(unittest.TestCase): def setUp(self): self.now = datetime(2026, 7, 23, 12, 0, 0, tzinfo=timezone.utc) def _lock( self, *, lifecycle: str | None, created_ago: timedelta, heartbeat_ago: timedelta, expires_in: timedelta, pid: int, ) -> dict: lease: dict = { "operation_type": ils.AUTHOR_ISSUE_WORK_LEASE, "issue_number": ISSUE, "branch": OWNER_BRANCH, "worktree_path": OWNER_WORKTREE, "created_at": _ts(self.now - created_ago), "last_heartbeat_at": _ts(self.now - heartbeat_ago), "expires_at": _ts(self.now + expires_in), } if lifecycle is not None: lease["lifecycle_version"] = lifecycle lease["task_session_id"] = "author_issue_work-deadbeefdeadbeef" return { "issue_number": ISSUE, "branch_name": OWNER_BRANCH, "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "worktree_path": OWNER_WORKTREE, "session_pid": pid, "pid": pid, "work_lease": lease, } def _foreign_conflict(self, existing: dict) -> str | None: return ils.assess_same_issue_lease_conflict( existing, issue_number=ISSUE, branch_name=FOREIGN_BRANCH, worktree_path=FOREIGN_WORKTREE, now=self.now, ) def _same_owner_conflict(self, existing: dict) -> str | None: return ils.assess_same_issue_lease_conflict( existing, issue_number=ISSUE, branch_name=OWNER_BRANCH, worktree_path=OWNER_WORKTREE, now=self.now, ) class TestHeartbeatNonLiveFutureExpiresReclaim(_ConflictGateBase): def test_stale_absolute_cap_future_expires_allows_foreign_reclaim(self): # Created >8h ago, heartbeated one minute ago, expires 9 min in the FUTURE, # owner PID alive: freshness = stale_absolute_cap, live=False, not expired. existing = self._lock( lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1, created_ago=timedelta(hours=9), heartbeat_ago=timedelta(minutes=1), expires_in=timedelta(minutes=9), pid=os.getpid(), ) freshness = ils.assess_lock_freshness(existing, now=self.now) self.assertEqual(freshness["status"], ils.STATUS_STALE_ABSOLUTE_CAP) self.assertFalse(freshness["live"]) self.assertFalse(ils.is_lease_expired(existing, now=self.now)) with mock.patch.object(ils, "is_process_alive", return_value=True): self.assertIsNone(self._foreign_conflict(existing)) def test_missed_heartbeat_ttl_gt_grace_future_expires_allows_foreign_reclaim(self): # Heartbeat grace (default 10 min) lapsed 5 min ago, but a TTL>grace policy # leaves expires_at 20 min in the FUTURE: stale_missed_heartbeat, not expired. existing = self._lock( lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1, created_ago=timedelta(minutes=30), heartbeat_ago=timedelta(minutes=15), expires_in=timedelta(minutes=20), pid=os.getpid(), ) freshness = ils.assess_lock_freshness(existing, now=self.now) self.assertEqual(freshness["status"], ils.STATUS_STALE_MISSED_HEARTBEAT) self.assertFalse(freshness["live"]) self.assertFalse(ils.is_lease_expired(existing, now=self.now)) with mock.patch.object(ils, "is_process_alive", return_value=True): self.assertIsNone(self._foreign_conflict(existing)) def test_same_owner_may_reclaim_its_own_abandoned_heartbeat_lease(self): existing = self._lock( lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1, created_ago=timedelta(hours=9), heartbeat_ago=timedelta(minutes=1), expires_in=timedelta(minutes=9), pid=os.getpid(), ) with mock.patch.object(ils, "is_process_alive", return_value=True): self.assertIsNone(self._same_owner_conflict(existing)) class TestLiveAndLegacyStillBlockForeign(_ConflictGateBase): def test_live_heartbeat_lease_still_blocks_foreign(self): existing = self._lock( lifecycle=lease_policy.LIFECYCLE_HEARTBEAT_V1, created_ago=timedelta(minutes=5), heartbeat_ago=timedelta(minutes=1), expires_in=timedelta(minutes=9), pid=os.getpid(), ) freshness = ils.assess_lock_freshness(existing, now=self.now) self.assertTrue(freshness["live"]) with mock.patch.object(ils, "is_process_alive", return_value=True): block = self._foreign_conflict(existing) self.assertIn("already has an active", block or "") def test_legacy_lease_future_expires_dead_pid_is_not_reclaimed_here(self): # AC-N8: a legacy lock keeps its absolute expires_at clock. Non-expired + # dead PID is non-live, but the widened band excludes legacy, so the # foreign acquirer is still blocked rather than silently reclaiming. existing = self._lock( lifecycle=None, created_ago=timedelta(hours=9), heartbeat_ago=timedelta(hours=9), expires_in=timedelta(hours=2), pid=4_194_304, # far above any live pid on a test host ) with mock.patch.object(ils, "is_process_alive", return_value=False): freshness = ils.assess_lock_freshness(existing, now=self.now) self.assertTrue(freshness["legacy_lease"]) self.assertFalse(freshness["live"]) self.assertFalse(ils.is_lease_expired(existing, now=self.now)) block = self._foreign_conflict(existing) self.assertIn("already has an active", block or "") if __name__ == "__main__": unittest.main()