Gitea-Tools/tests/test_issue_790_heartbeat_mcp_path.py

"""Task heartbeat through the native MCP author path (#790 Slice A, AC-N6).

Assessor-level coverage is not sufficient here, and this project has already
paid for learning that: in review #499 on PR #791 the #760 renewal waiver was
computed correctly and then *discarded* at two later gates, so every real
renewal still failed while the unit suite stayed green. AC-N6 exists because of
that, and requires driving the real tools against a real git repository and a
real durable lock file, composing the gates in production order.

These tests therefore call ``gitea_lock_issue`` and
``gitea_heartbeat_issue_lock`` themselves and assert on what lands on disk,
never on an assessor's return value alone.
"""

from __future__ import annotations

import os
import subprocess
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__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from mutation_profile_fixture import shared_mutation_env  # noqa: E402

import issue_lock_provenance  # noqa: E402
import issue_lock_store  # noqa: E402
import lease_policy  # noqa: E402
import mcp_server  # noqa: E402

ISSUE = 9791
BRANCH = f"fix/issue-{ISSUE}-heartbeat-mcp"
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"


def _ts(moment: datetime) -> str:
    return (
        moment.astimezone(timezone.utc)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )


class _HeartbeatMcpBase(unittest.TestCase):
    """Real git repo plus a real durable lock, driven through the real tools."""

    def setUp(self):
        self.lock_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.lock_dir.cleanup)
        self.repo = tempfile.mkdtemp(prefix="issue790-mcp-")
        self.addCleanup(lambda: subprocess.run(["rm", "-rf", self.repo], check=False))
        self._init_worktree()
        self.remotes = patch.dict(
            mcp_server.REMOTES,
            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
        )
        self.remotes.start()
        self.addCleanup(patch.stopall)
        mcp_server._IDENTITY_CACHE.clear()

    def _git(self, *args):
        return subprocess.run(
            ["git", "-C", self.repo, *args], capture_output=True, text=True, check=True
        )

    def _init_worktree(self):
        self._git("init", "-q", "-b", "master")
        self._git("config", "user.email", "[email protected]")
        self._git("config", "user.name", "Test")
        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
            fh.write("seed\n")
        self._git("add", "seed.txt")
        self._git("commit", "-q", "-m", "seed")
        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
        # A fresh claim starts base-equivalent, which is the ordinary first-lock
        # shape and exercises assess_issue_lock_worktree on its normal path.
        self._git("checkout", "-q", "-b", BRANCH)
        self.head_sha = self.base_sha
        self.worktree = os.path.realpath(self.repo)

    def _lock_path(self):
        return issue_lock_store.lock_file_path(
            remote="prgs",
            org=ORG,
            repo=REPO,
            issue_number=ISSUE,
            lock_dir=self.lock_dir.name,
        )

    def _tool_env(self):
        env = shared_mutation_env(
            PROFILE, include_example_repo=True, GITEA_ISSUE_LOCK_DIR=self.lock_dir.name
        )
        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
        return env

    def _git_state(self, *, porcelain="", base_equivalent=True):
        return {
            "current_branch": BRANCH,
            "porcelain_status": porcelain,
            "base_equivalent": base_equivalent,
            "head_sha": self.head_sha,
            "inspected_git_root": self.worktree,
            "base_branch": "master",
        }

    def run_lock_issue(
        self,
        *,
        branch_entries=None,
        open_prs=None,
        git_state=None,
        identity=IDENTITY,
        profile=PROFILE,
    ):
        branch_entries = branch_entries if branch_entries is not None else []
        open_prs = open_prs if open_prs is not None else []
        git_state = git_state or self._git_state()
        env = self._tool_env()
        with patch(
            "mcp_server.api_get_all", return_value=list(branch_entries)
        ), patch(
            "mcp_server._list_open_pulls", return_value=list(open_prs)
        ), patch(
            "mcp_server.get_auth_header", return_value="token x"
        ), patch(
            "mcp_server._work_lease_claimant",
            return_value={"username": identity, "profile": profile},
        ), patch(
            "mcp_server.issue_lock_worktree.read_worktree_git_state",
            return_value=git_state,
        ), patch(
            "mcp_server.issue_duplicate_context_fetcher",
            side_effect=lambda h, o, r, auth, issue_number: (
                list(open_prs),
                [b.get("name") for b in branch_entries if isinstance(b, dict)],
                {"status": "not_claimed"},
            ),
        ), patch.dict(os.environ, env, clear=True):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_lock_issue(
                issue_number=ISSUE,
                branch_name=BRANCH,
                remote="prgs",
                worktree_path=self.worktree,
            )

    def run_heartbeat(
        self, *, task_session_id, identity=IDENTITY, profile=PROFILE, **kwargs
    ):
        env = self._tool_env()
        with patch(
            "mcp_server._work_lease_claimant",
            return_value={"username": identity, "profile": profile},
        ), patch("mcp_server.get_auth_header", return_value="token x"), patch.dict(
            os.environ, env, clear=True
        ):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_heartbeat_issue_lock(
                issue_number=ISSUE,
                branch_name=kwargs.pop("branch_name", BRANCH),
                task_session_id=task_session_id,
                remote="prgs",
                worktree_path=kwargs.pop("worktree_path", self.worktree),
                **kwargs,
            )

    def write_legacy_lock(self, *, hours_old: float = 3.0, ttl_hours: float = 4.0):
        """A durable lock in the shape the store wrote before this slice."""
        now = datetime.now(timezone.utc)
        claimant = {"username": IDENTITY, "profile": PROFILE}
        created = now - timedelta(hours=hours_old)
        record = {
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "remote": "prgs",
            "org": ORG,
            "repo": REPO,
            "worktree_path": self.worktree,
            "session_pid": os.getpid(),
            "pid": os.getpid(),
            "lock_generation": 1,
            "work_lease": {
                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
                "issue_number": ISSUE,
                "pr_number": None,
                "branch": BRANCH,
                "worktree_path": self.worktree,
                "claimant": claimant,
                "created_at": _ts(created),
                # The legacy signature: never advanced past creation.
                "last_heartbeat_at": _ts(created),
                "expires_at": _ts(created + timedelta(hours=ttl_hours)),
            },
            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
                tool="gitea_lock_issue", claimant=claimant
            ),
        }
        path = self._lock_path()
        record["lock_file_path"] = path
        issue_lock_store.save_lock_file(path, record)
        return record


class TestLockIssueMintsTheLifecycle(_HeartbeatMcpBase):
    """Durable lock creation and read-back through the real tool."""

    def test_native_lock_writes_the_marker_and_a_task_session_id(self):
        result = self.run_lock_issue()
        self.assertTrue(result["success"], result)

        written = issue_lock_store.read_lock_file(result["lock_file_path"])
        lease = written["work_lease"]
        self.assertEqual(
            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
        )
        self.assertTrue(lease["task_session_id"])
        self.assertFalse(issue_lock_store.is_legacy_lease(written))
        # AC-N1: the ownership key is not the daemon pid, which is recorded
        # separately as evidence.
        self.assertNotIn(str(written["session_pid"]), lease["task_session_id"])
        self.assertEqual(written["session_pid"], os.getpid())

    def test_native_lease_uses_the_policy_window_not_four_hours(self):
        result = self.run_lock_issue()
        lease = result["work_lease"]
        created = datetime.fromisoformat(lease["created_at"].replace("Z", "+00:00"))
        expires = datetime.fromisoformat(lease["expires_at"].replace("Z", "+00:00"))
        policy = lease_policy.policy_for(lease_policy.TASK_CLASS_AUTHOR_ISSUE_WORK)
        self.assertEqual(
            (expires - created).total_seconds() / 60.0, policy.initial_ttl_minutes
        )

    def test_freshness_of_a_new_native_lock_is_live(self):
        result = self.run_lock_issue()
        self.assertEqual(
            result["lock_freshness"]["status"], issue_lock_store.STATUS_LIVE
        )
        self.assertTrue(result["lock_freshness"]["live"])


class TestHeartbeatThroughTheTool(_HeartbeatMcpBase):
    def _lock_and_session(self):
        result = self.run_lock_issue()
        self.assertTrue(result["success"], result)
        return result, result["work_lease"]["task_session_id"]

    def test_heartbeat_slides_the_lease_and_advances_the_generation(self):
        locked, session = self._lock_and_session()
        before = issue_lock_store.read_lock_file(locked["lock_file_path"])

        beat = self.run_heartbeat(task_session_id=session)

        self.assertTrue(beat["success"], beat)
        self.assertEqual(beat["operation"], "heartbeat")
        after = issue_lock_store.read_lock_file(locked["lock_file_path"])
        self.assertGreater(
            issue_lock_store.lock_generation(after),
            issue_lock_store.lock_generation(before),
        )
        self.assertGreaterEqual(
            after["work_lease"]["expires_at"], before["work_lease"]["expires_at"]
        )
        self.assertEqual(after["work_lease"]["heartbeat_count"], 2)

    def test_heartbeat_evidence_survives_the_downstream_mutation_gate(self):
        """The #499 F2 lesson, applied.

        A sanction that is computed and then discarded downstream is worthless.
        After a heartbeat the lock must still satisfy the gate every author
        mutation runs through.
        """
        locked, session = self._lock_and_session()
        self.run_heartbeat(task_session_id=session)

        written = issue_lock_store.read_lock_file(locked["lock_file_path"])
        verdict = issue_lock_store.verify_lock_for_mutation(
            written,
            issue_number=ISSUE,
            branch_name=BRANCH,
            worktree_path=self.worktree,
        )
        self.assertTrue(verdict["proven"], verdict)
        self.assertFalse(verdict["block"])

    def _duplicate_gate(self, *, open_prs, branches):
        env = self._tool_env()
        with patch("mcp_server.get_auth_header", return_value="token x"), patch(
            "mcp_server.issue_duplicate_context_fetcher",
            side_effect=lambda h, o, r, auth, issue_number: (
                list(open_prs),
                list(branches),
                {"status": "not_claimed"},
            ),
        ), patch.dict(os.environ, env, clear=True):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_assess_work_issue_duplicate(
                issue_number=ISSUE, branch_name=BRANCH, remote="prgs"
            )

    def test_heartbeat_does_not_change_the_duplicate_gate_verdict(self):
        """The gate must be invariant under heartbeating.

        The point is not that the gate passes — with a linked open PR at the
        lock phase it correctly blocks (#400), heartbeat or not. The property
        that matters is that sliding a lease neither loosens the gate nor
        corrupts the lock state it reads: the verdict before and after a
        heartbeat must be identical, for both the clear and the blocking shape.
        """
        _, session = self._lock_and_session()
        linked = [{"number": 4242, "head": {"ref": BRANCH, "sha": self.head_sha}}]

        clear_before = self._duplicate_gate(open_prs=[], branches=[])
        blocked_before = self._duplicate_gate(open_prs=linked, branches=[BRANCH])

        self.assertTrue(self.run_heartbeat(task_session_id=session)["success"])

        clear_after = self._duplicate_gate(open_prs=[], branches=[])
        blocked_after = self._duplicate_gate(open_prs=linked, branches=[BRANCH])

        self.assertEqual(clear_before["outcome"], clear_after["outcome"])
        self.assertFalse(clear_after["block"])
        self.assertEqual(blocked_before["outcome"], blocked_after["outcome"])
        self.assertTrue(blocked_after["block"])
        self.assertEqual(blocked_after["linked_open_pr"], 4242)

    def test_foreign_session_id_is_refused_through_the_tool(self):
        self._lock_and_session()
        beat = self.run_heartbeat(task_session_id="author_issue_work-ffffffffffffffff")
        self.assertFalse(beat["success"])
        self.assertIn("task_session_id does not match", " ".join(beat["reasons"]))

    def test_stale_generation_is_refused_through_the_tool(self):
        locked, session = self._lock_and_session()
        current = issue_lock_store.lock_generation(
            issue_lock_store.read_lock_file(locked["lock_file_path"])
        )
        beat = self.run_heartbeat(
            task_session_id=session, expected_generation=current + 5
        )
        self.assertFalse(beat["success"])
        self.assertIn("generation changed", beat["reasons"][0])

    def test_foreign_claimant_is_refused_through_the_tool(self):
        _, session = self._lock_and_session()
        beat = self.run_heartbeat(task_session_id=session, identity="someone-else")
        self.assertFalse(beat["success"])

    def test_heartbeat_cannot_acquire_a_missing_lock(self):
        beat = self.run_heartbeat(task_session_id="author_issue_work-000000000000")
        self.assertFalse(beat["success"])
        self.assertIn("no durable lock", beat["reasons"][0])

    def test_alive_pid_alone_does_not_keep_a_lease_live_through_the_tool(self):
        """PID-only refusal, end to end.

        The recorded pid is this live process. The lock is aged past its grace
        with no heartbeat, so the tool must refuse to slide it and the durable
        record must classify as a missed heartbeat rather than as live.
        """
        locked, session = self._lock_and_session()
        record = issue_lock_store.read_lock_file(locked["lock_file_path"])
        record["work_lease"]["last_heartbeat_at"] = _ts(
            datetime.now(timezone.utc) - timedelta(minutes=30)
        )
        record["work_lease"]["expires_at"] = _ts(
            datetime.now(timezone.utc) + timedelta(hours=2)
        )
        issue_lock_store.save_lock_file(locked["lock_file_path"], record)

        self.assertTrue(issue_lock_store.is_process_alive(record["session_pid"]))
        fresh = issue_lock_store.assess_lock_freshness(record)
        self.assertEqual(
            fresh["status"], issue_lock_store.STATUS_STALE_MISSED_HEARTBEAT
        )
        self.assertTrue(fresh["pid_alive"])

        beat = self.run_heartbeat(task_session_id=session)
        self.assertFalse(beat["success"])
        self.assertIn("reclaimed", " ".join(beat["reasons"]))


class TestLegacyLocksThroughTheTool(_HeartbeatMcpBase):
    """AC-N8 end to end: protected on deployment, and rebindable."""

    def test_legacy_lock_stays_protected_after_deployment(self):
        record = self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
        fresh = issue_lock_store.assess_lock_freshness(record)
        self.assertEqual(fresh["status"], issue_lock_store.STATUS_LIVE)
        self.assertTrue(fresh["legacy_lease"])
        self.assertTrue(fresh["legacy_expiry_preserved"])
        # It had never heartbeated, so under the new grace alone it would be
        # long gone; the preserved absolute expiry is what protects it.
        self.assertEqual(
            record["work_lease"]["created_at"],
            record["work_lease"]["last_heartbeat_at"],
        )

    def test_tool_rebinds_a_legacy_lock_and_mints_a_first_heartbeat(self):
        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)

        result = self.run_heartbeat(task_session_id=None)

        self.assertTrue(result["success"], result)
        self.assertEqual(result["operation"], "legacy_rebind")
        self.assertTrue(result["task_session_id"])

        written = issue_lock_store.read_lock_file(self._lock_path())
        lease = written["work_lease"]
        self.assertEqual(
            lease["lifecycle_version"], lease_policy.LIFECYCLE_HEARTBEAT_V1
        )
        self.assertEqual(lease["heartbeat_count"], 1)
        self.assertNotEqual(
            lease["created_at"],
            written["legacy_rebind"]["legacy_origin"]["created_at"],
        )
        self.assertFalse(issue_lock_store.is_legacy_lease(written))

    def test_rebound_lock_then_heartbeats_through_the_tool(self):
        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
        rebound = self.run_heartbeat(task_session_id=None)
        beat = self.run_heartbeat(task_session_id=rebound["task_session_id"])
        self.assertTrue(beat["success"], beat)
        self.assertEqual(beat["operation"], "heartbeat")
        self.assertEqual(beat["heartbeat_count"], 2)

    def test_rebind_refuses_a_foreign_owner_through_the_tool(self):
        self.write_legacy_lock(hours_old=3.0, ttl_hours=4.0)
        result = self.run_heartbeat(task_session_id=None, identity="someone-else")
        self.assertFalse(result["success"])
        self.assertEqual(result["operation"], "legacy_rebind")


if __name__ == "__main__":
    unittest.main()