sysadminandClaude Opus 4.8
243f52dc79
Slice A of Issue #790, per the controller reassessment in comment 13958. Does
not close the issue: terminal retirement (Slice B) and the read-side generation
check plus the #760 renewal re-scope (Slice C) are deliberately not implemented.
The defect. `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, and the recorded PID is the long-lived MCP
daemon rather than the authoring task, so an abandoned claim stayed live for the
full four hours. A tree-wide search found the field written in exactly one place
and advanced by nothing. Issue #787 / PR #789 hit this; Issue #760 / PR #791 hit
it again, blocking reconciliation for over five hours after its work had landed.
A1 — central policy. New `lease_policy` declares every duration for every task
class in one place: author initial/sliding TTL 10 minutes, heartbeat cadence 2,
stale warning 5, missed-heartbeat grace 10, absolute cap 8 hours, recovery grace
10, terminal race-drain 2. It ships first so the first heartbeat and TTL
behavior to run reads from it (AC-N7). The duplicated four-hour literal is gone
from both `issue_lock_store` and `gitea_mcp_server`. Reviewer, merger, and
conflict-fix classes are declared but not rewired — Slice C moves those call
sites — and a test asserts the declaration still equals the constants #747 and
`pr_work_lease` own, so the two cannot drift apart unnoticed.
A2 — load-bearing freshness, with two deliberate asymmetries. An alive PID never
establishes freshness anywhere (AC-N2); it is recorded as evidence and no branch
returns live because of it. A dead PID still marks a lease stale, and that band
still precedes every heartbeat evaluation, so #753 dead-session recovery keys on
exactly the classification it always did. New bands `stale_missed_heartbeat` and
`stale_absolute_cap` are classified in `branch_cleanup_guard` rather than
falling through to unknown-status, and still block unless the ownership record
proves `reclaim_allowed is True`. A heartbeat lease carrying no heartbeat is
contradictory and fails closed. `assess_expired_lock_reclaim` accepts a lapsed
heartbeat as reclaim grounds for heartbeat-lifecycle leases only: under this
lifecycle the heartbeat is the liveness proof, and also requiring a dead PID
would reinstate the original defect.
A3/A4 — task-session identity and the writer. `mint_task_session_id` produces an
ownership key containing no process identifier, since the daemon PID is reused
by every task it serves and identifies none of them. `heartbeat_session_lock`
writes inside the existing per-issue flock under the #772 generation
compare-and-swap, verifying exact issue, branch, realpath-normalized worktree,
claimant username, claimant profile, and recorded session identifier. It cannot
acquire, take over, or revive: a lease past its grace is refused and must use
the reclaim path, so a session that stopped proving liveness cannot restore
ownership retroactively. New `gitea_heartbeat_issue_lock` gates on the same
authority as `lock_issue`, being strictly narrower.
A5 — legacy compatibility (AC-N8). The explicit `lifecycle_version` marker, never
a timestamp comparison, discriminates legacy from heartbeat leases: a legacy lock
has `last_heartbeat_at == created_at` forever precisely because nothing advanced
it, and a freshly minted heartbeat lease has them equal too, so the equality
carries no information in either direction. Legacy locks keep their recorded
absolute expiry and are never evaluated against the short grace, so deployment
cannot make an existing claim instantly reclaimable. They leave that state only
by terminal retirement (Slice B) or by `rebind_legacy_lock`, which re-verifies
the exact owner and mints a genuine identifier and first heartbeat while
preserving the original claim under `legacy_origin`. Rebinding a lapsed legacy
lease is refused; that belongs to #760 renewal or #601 reclaim.
A6 — native coverage. Review #499 proved assessor-level tests miss discard
points, so `tests/test_issue_790_heartbeat_mcp_path.py` drives the real tools
against a real git repository and a real durable lock: lock creation and
read-back, policy window, freshness, survival of `verify_lock_for_mutation`,
invariance of the duplicate-work and linked-open-PR gates, CAS rejection,
foreign-session and foreign-claimant refusal, alive-PID-only refusal, missed
heartbeat, legacy protection on deployment, and legacy rebinding.
Tests. New suites 55 passed. Lock and lease regression set (issue_lock_store,
lease_lifecycle, #753, #755, #760 x2, #768, #772, lock registration, worktree,
adoption, duplicate gate, branch cleanup guard, capability invariants, claim
heartbeat, worktrees) 383 passed with 98 subtests. Full suite 4295 passed, 11
failed, 6 skipped, 499 subtests passed, against a clean master baseline worktree
at 620ed6e9 that reports 11 failed and 4240 passed — the same eleven node IDs.
The 55-test delta is exactly the new suites; no new failures.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_011u6GKSJwwrrYjguPjs1aK5
595 lines
24 KiB
Python
595 lines
24 KiB
Python
"""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()
|