fix: exclude epic and child-only containers from allocator selection (Closes #844)

Epics and parent issues whose body delegates implementation to children are
skipped before ranking with structured reason epic_or_child_only_container.
Title-only "epic" mentions without body/label evidence remain eligible.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

allocator_service.py

@@ -53,6 +53,8 @@ OUTCOME_CANDIDATE_SET_DRIFT = "candidate_set_drift"
 SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session"
 # #776: controller-supplied pre-rank exclusion.
 SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller"
+# #844: epic / child-only implementation container (pre-rank).
+SKIP_EPIC_OR_CHILD_ONLY_CONTAINER = "epic_or_child_only_container"
 
 # Ownership verdicts for a live claim on a candidate (#765).
 OWNERSHIP_OWN = "own"
@@ -130,6 +132,39 @@ ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
 }
 
 
+# Body phrases that prove an issue is an implementation container, not a
+# unit of direct author work (#844). Matched case-insensitively against the
+# issue body. Title alone is never sufficient (ordinary issues may mention
+# "epic" incidentally).
+_CHILD_ONLY_BODY_MARKERS: tuple[str, ...] = (
+    "implementation is delivered via child issues only",
+    "implementation is delivered through child issues only",
+    "implementation is delivered via child issues",
+    "implementation is delivered through child issues",
+    "do not implement product features in this epic",
+    "do not implement product features in this epic issue itself",
+    "no product feature implementation is claimed complete solely on this epic",
+    "implementable child issues remain independently eligible",
+    "owns the product roadmap and linkage",
+    "this epic owns the product roadmap",
+    "coordination container",
+    "child-only container",
+    "implementation is delegated to child",
+)
+
+# Explicit epic / umbrella labels (structured evidence preferred over title).
+_EPIC_LABELS: frozenset[str] = frozenset(
+    {
+        "type:epic",
+        "epic",
+        "kind:epic",
+        "scope:epic",
+        "type:umbrella",
+        "umbrella",
+    }
+)
+
+
 @dataclass
 class WorkCandidate:
     """One assignable Gitea issue or PR presented to the allocator."""
@@ -139,6 +174,7 @@ class WorkCandidate:
     state: str = "open"
     labels: tuple[str, ...] = ()
     title: str = ""
+    body: str = ""
     priority: int = 0
     head_sha: str | None = None
     # Routing signals (callers derive from Gitea / review feedback).
@@ -158,6 +194,7 @@ class WorkCandidate:
         self.labels = tuple(
             str(x).strip().lower() for x in (self.labels or ()) if str(x).strip()
         )
+        self.body = str(self.body or "")
         if self.kind not in WORK_KINDS:
             raise InvalidWorkKindError(
                 f"candidate kind '{self.kind}' is not assignable; only "
@@ -171,6 +208,7 @@ class WorkCandidate:
             "state": self.state,
             "labels": list(self.labels),
             "title": self.title,
+            "body": self.body,
             "priority": self.priority,
             "head_sha": self.head_sha,
             "request_changes_current_head": self.request_changes_current_head,
@@ -184,6 +222,51 @@ class WorkCandidate:
         }
 
 
+def classify_epic_or_child_only_container(
+    c: WorkCandidate,
+) -> tuple[bool, str | None]:
+    """Return whether *c* is an epic / child-only implementation container (#844).
+
+    Exclusion uses structured evidence first (labels, body scope language).
+    A bare title containing the word "epic" is **not** enough — ordinary
+    implementable issues may mention epics incidentally. A title that is
+    explicitly prefixed ``Epic:`` only counts when the body also proves
+    child-only / no-direct-implementation scope (or an epic label is present).
+
+    PRs are never classified as containers here (they already have a head).
+    """
+    if c.kind != "issue":
+        return False, None
+
+    labels = set(c.labels)
+    epic_label = sorted(labels & _EPIC_LABELS)
+    body_l = (c.body or "").lower()
+    title = (c.title or "").strip()
+    title_l = title.lower()
+
+    body_hits = [m for m in _CHILD_ONLY_BODY_MARKERS if m in body_l]
+    title_epic_prefix = title_l.startswith("epic:") or title_l.startswith("epic ")
+
+    if epic_label:
+        detail = f"label={epic_label[0]}"
+        if body_hits:
+            detail = f"{detail}; body_marker={body_hits[0]!r}"
+        return True, detail
+
+    if body_hits:
+        # Body proves child-only / umbrella scope. Title "Epic:" is corroborating
+        # but not required — containers without the word still exclude.
+        detail = f"body_marker={body_hits[0]!r}"
+        if title_epic_prefix:
+            detail = f"title_epic_prefix; {detail}"
+        return True, detail
+
+    # Title-only "Epic:" without body scope evidence is insufficient (#844 AC:
+    # eligibility does not rely solely on the word "Epic" in a title).
+    # Similarly, incidental "epic" mid-title without markers stays eligible.
+    return False, None
+
+
 @dataclass
 class SkipRecord:
     kind: str
@@ -850,7 +933,8 @@ def allocate_next_work(
     ownership_defects: list[dict[str, Any]] = []
     controller_excluded: list[dict[str, Any]] = []
 
-    # #776 AC2: remove excluded numbers *before* ranking / selection / lease.
+    # #776 AC2 + #844: remove excluded numbers *and* epic/child-only containers
+    # *before* ranking / selection / lease so they never receive assignments.
     rankable: list[WorkCandidate] = []
     for c in candidates:
         if int(c.number) in exclude_set:
@@ -929,6 +1013,23 @@ def allocate_next_work(
                     },
                 }
             continue
+        # #844: epics / child-only containers are never direct implement targets.
+        is_container, container_detail = classify_epic_or_child_only_container(c)
+        if is_container:
+            detail = container_detail or "epic or child-only container"
+            reason = (
+                f"{c.kind}#{c.number} {SKIP_EPIC_OR_CHILD_ONLY_CONTAINER}: "
+                f"{detail}; implementation is delegated to child issues"
+            )
+            skipped.append(
+                SkipRecord(
+                    c.kind,
+                    c.number,
+                    reason,
+                    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
+                )
+            )
+            continue
         rankable.append(c)
 
     ordered = sort_candidates(rankable)
@@ -1341,6 +1442,7 @@ def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate:
         state=str(data.get("state") or "open"),
         labels=tuple(data.get("labels") or ()),
         title=str(data.get("title") or ""),
+        body=str(data.get("body") or ""),
         priority=priority,
         head_sha=data.get("head_sha"),
         request_changes_current_head=bool(data.get("request_changes_current_head")),

gitea_mcp_server.py

@@ -19912,6 +19912,7 @@ def _allocator_candidates_from_gitea(
                         state="open",
                         labels=tuple(labels),
                         title=title,
+                        body=body,
                         priority=20 if "status:ready" in labels else 1,
                         blocked=blocked,
                         dependency_unmet=dep_unmet,

pr_work_lease.py

@@ -228,74 +228,25 @@ def find_active_reviewer_lease(
     return None
 
 
-def _conflict_fix_chain_key(lease: dict) -> tuple | None:
-    """Identity of the lease chain a conflict-fix marker belongs to (#842).
-
-    Keyed by PR number, profile, head_before, and branch. Returns None when any
-    required component (pr_number, profile, head_before) is missing or malformed.
-    """
-    raw = lease.get("raw_fields") or {}
-    pr_number = lease.get("pr_number")
-    profile = (lease.get("profile") or "").strip().lower()
-    head_before = lease.get("head_before")
-    branch = (lease.get("branch") or raw.get("branch") or "").strip()
-    if not (pr_number and profile and head_before):
-        return None
-    return (pr_number, profile, head_before, branch)
-
-
-def _conflict_fix_chain_matches(key1: tuple, key2: tuple) -> bool:
-    """True when two conflict-fix chain keys refer to the same lease chain."""
-    pr1, profile1, head1, branch1 = key1
-    pr2, profile2, head2, branch2 = key2
-    if pr1 != pr2 or profile1 != profile2 or head1 != head2:
-        return False
-    if branch1 and branch2 and branch1 != branch2:
-        return False
-    return True
-
-
-def _conflict_fix_chain_terminated_after(entries: list[dict], index: int) -> bool:
-    """True when a later marker terminates the conflict-fix chain of ``entries[index]``.
-
-    Append-only newest-wins: a terminal marker (phase=released/blocked/done)
-    ends only its matching claim chain (#842).
-    """
-    key = _conflict_fix_chain_key(entries[index])
-    if key is None:
-        return False
-    for later in entries[index + 1:]:
-        phase = (later.get("phase") or "").strip().lower()
-        if phase not in _TERMINAL_CONFLICT_FIX_PHASES:
-            continue
-        later_key = _conflict_fix_chain_key(later)
-        if later_key and _conflict_fix_chain_matches(key, later_key):
-            return True
-    return False
-
-
 def find_active_conflict_fix_lease(
     comments: list[dict],
     *,
     pr_number: int,
     now: datetime | None = None,
 ) -> dict[str, Any] | None:
-    """Return the newest unexpired, non-terminated conflict-fix lease for *pr_number*, if any."""
+    """Return the newest unexpired conflict-fix lease for *pr_number*, if any."""
     now = now or datetime.now(timezone.utc)
     candidates = [
         entry for entry in _comment_entries(comments, pr_number=pr_number)
         if entry.get("lease_kind") == "conflict_fix"
     ]
-    for index in range(len(candidates) - 1, -1, -1):
+    for lease in reversed(candidates):
-        lease = candidates[index]
         if _lease_expired(lease, now=now):
             continue
         phase = (lease.get("phase") or "").strip().lower()
         if phase in _TERMINAL_CONFLICT_FIX_PHASES:
             continue
         if phase in _ACTIVE_CONFLICT_FIX_PHASES or phase:
-            if _conflict_fix_chain_terminated_after(candidates, index):
-                continue
             return lease
     return None
 

tests/test_allocator_epic_container_exclusion.py

@@ -0,0 +1,243 @@
+"""Allocator epic / child-only container pre-rank exclusion (#844).
+
+Covers:
+* Issue #631-shaped child-only epic is excluded before ranking.
+* Implementable child issues remain eligible and can be selected.
+* Ordinary issues that merely mention "epic" in title/body are not excluded.
+* Excluded containers never receive assignments or workflow leases.
+* Structured skip reason ``epic_or_child_only_container`` is reported.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+
+from allocator_service import (
+    OUTCOME_ASSIGNED,
+    OUTCOME_PREVIEW,
+    SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
+    WorkCandidate,
+    allocate_next_work,
+    classify_epic_or_child_only_container,
+)
+from control_plane_db import ControlPlaneDB
+
+REMOTE = "prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+# Minimal body mirroring issue #631 authoritative scope language.
+_EPIC_631_BODY = """
+## Scope (umbrella)
+
+This epic owns the **product roadmap and linkage** for the Web Console.
+Implementation is delivered via child issues only.
+
+## Explicit non-goals
+
+* Do not implement product features in this epic issue itself.
+* No product feature implementation is claimed complete solely on this epic.
+"""
+
+_CHILD_BODY = """
+## Problem
+
+Operators need a workflow-event timeline model for Phase 1.
+
+## Acceptance criteria
+
+- [ ] Timeline model API exists
+"""
+
+
+def _issue(
+    number: int,
+    *,
+    title: str = "",
+    body: str = "",
+    labels: tuple[str, ...] = ("status:ready", "type:feature"),
+    priority: int = 20,
+) -> WorkCandidate:
+    return WorkCandidate(
+        kind="issue",
+        number=number,
+        state="open",
+        labels=labels,
+        title=title or f"issue {number}",
+        body=body,
+        priority=priority,
+    )
+
+
+class ClassifyEpicContainerTest(unittest.TestCase):
+    def test_631_shaped_body_and_title_is_container(self) -> None:
+        c = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIsNotNone(detail)
+        self.assertIn("body_marker", detail or "")
+
+    def test_body_markers_without_epic_title(self) -> None:
+        c = _issue(
+            900,
+            title="Control plane roadmap tracker",
+            body="Implementation is delivered via child issues only.",
+        )
+        is_c, _ = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+
+    def test_epic_label_alone_is_container(self) -> None:
+        c = _issue(
+            901,
+            title="Roadmap linkage",
+            body="Track children.",
+            labels=("status:ready", "type:epic"),
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertTrue(is_c)
+        self.assertIn("type:epic", detail or "")
+
+    def test_title_epic_prefix_alone_not_container(self) -> None:
+        """Title-only 'Epic:' without body scope evidence stays eligible (#844)."""
+        c = _issue(
+            902,
+            title="Epic: something mentioned only in title",
+            body="Implement a concrete fix for the allocator skip list.",
+        )
+        is_c, detail = classify_epic_or_child_only_container(c)
+        self.assertFalse(is_c)
+        self.assertIsNone(detail)
+
+    def test_incidental_epic_word_not_container(self) -> None:
+        c = _issue(
+            903,
+            title="Document epic handoff conventions",
+            body=(
+                "Update the docs so implementable issues that mention an epic "
+                "remain independently executable."
+            ),
+        )
+        is_c, _ = classify_epic_or_child_only_container(c)
+        self.assertFalse(is_c)
+
+    def test_prs_never_classified(self) -> None:
+        pr = WorkCandidate(
+            kind="pr",
+            number=10,
+            state="open",
+            title="Epic: fake",
+            body="Implementation is delivered via child issues only.",
+            head_sha="a" * 40,
+            priority=5,
+        )
+        is_c, _ = classify_epic_or_child_only_container(pr)
+        self.assertFalse(is_c)
+
+
+class AllocateEpicContainerExclusionTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def _alloc(self, candidates, **kwargs):
+        defaults = dict(
+            session_id="sess-844",
+            role="author",
+            remote=REMOTE,
+            org=ORG,
+            repo=REPO,
+            profile_name="prgs-author",
+            username="jcwalker3",
+            claims={},
+            apply=False,
+        )
+        defaults.update(kwargs)
+        return allocate_next_work(self.db, candidates=candidates, **defaults)
+
+    def test_631_shaped_epic_excluded_child_selected(self) -> None:
+        epic = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        child = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=_CHILD_BODY,
+        )
+        res = self._alloc([epic, child], apply=False)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(res["selected"]["number"], 637)
+        skipped = {s["number"]: s for s in res["skipped"]}
+        self.assertIn(631, skipped)
+        self.assertEqual(
+            skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
+        )
+        self.assertIn(SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[631]["reason"])
+
+    def test_container_cannot_receive_assignment_or_lease(self) -> None:
+        epic = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        res = self._alloc([epic], apply=True)
+        self.assertTrue(res["success"], res)
+        # Only container present → no safe work; never assigned_work.
+        self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertIsNone(res.get("assignment"))
+        self.assertIsNone(res.get("selected"))
+        skipped = {s["number"]: s for s in res["skipped"]}
+        self.assertEqual(
+            skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
+        )
+        # No lease row for the epic.
+        leases = self.db.list_active_leases(
+            remote=REMOTE, org=ORG, repo=REPO
+        ) if hasattr(self.db, "list_active_leases") else []
+        # Prefer generic inventory if available.
+        if not leases and hasattr(self.db, "list_leases"):
+            leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
+        for lease in leases or []:
+            work_number = lease.get("work_number") if isinstance(lease, dict) else None
+            self.assertNotEqual(work_number, 631)
+
+    def test_incidental_epic_title_remains_eligible(self) -> None:
+        ordinary = _issue(
+            700,
+            title="Document epic handoff conventions",
+            body="Write runbook text about epic vs child issues.",
+        )
+        res = self._alloc([ordinary], apply=False)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["selected"]["number"], 700)
+        self.assertEqual(res["skipped"], [])
+
+    def test_apply_selects_child_not_epic(self) -> None:
+        epic = _issue(
+            631,
+            title="Epic: MCP Control Plane Web Console",
+            body=_EPIC_631_BODY,
+        )
+        child = _issue(
+            637,
+            title="Web Console: Workflow-event timeline model (Phase 1)",
+            body=_CHILD_BODY,
+        )
+        res = self._alloc([epic, child], apply=True)
+        self.assertTrue(res["success"], res)
+        self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
+        self.assertEqual(res["selected"]["number"], 637)
+        self.assertEqual(res["assignment"]["work_number"], 637)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_work_lease.py

@@ -19,7 +19,6 @@ from pr_work_lease import (  # noqa: E402
     assess_reviewer_mutation_blocked,
     assess_reviewer_stale_head_final_report,
     format_conflict_fix_lease_body,
-    find_active_conflict_fix_lease,
     parse_conflict_fix_lease_comment,
     parse_reviewer_lease_comment,
 )
@@ -204,157 +203,5 @@ class TestFormatLease(unittest.TestCase):
         self.assertEqual(parsed["pr_number"], 376)
 
 
-class TestConflictFixLeaseLifecycle(unittest.TestCase):
-    def test_claim_followed_by_matching_release(self):
-        claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
-        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
-        release_body = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "branch: feat/fix-376",
-            "worktree: branches/fix-376",
-            "profile: prgs-author",
-            "phase: released",
-            f"head_before: {HEAD_A}",
-            f"head_after: {HEAD_B}",
-            f"expires_at: {expires}",
-        ])
-        comments = [{"body": claim_body}, {"body": release_body}]
-        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
-        self.assertIsNone(lease)
-
-    def test_expired_claim_without_release(self):
-        past_expires = (NOW - timedelta(minutes=10)).isoformat().replace("+00:00", "Z")
-        claim_body = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "phase: claimed",
-            f"head_before: {HEAD_A}",
-            f"expires_at: {past_expires}",
-            "profile: prgs-author",
-        ])
-        comments = [{"body": claim_body}]
-        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
-        self.assertIsNone(lease)
-
-    def test_mismatched_release_different_head(self):
-        claim_body = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
-        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
-        release_body = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "profile: prgs-author",
-            "phase: released",
-            f"head_before: {HEAD_B}",
-            f"expires_at: {expires}",
-        ])
-        comments = [{"body": claim_body}, {"body": release_body}]
-        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
-        self.assertIsNotNone(lease)
-        self.assertEqual(lease["phase"], "claimed")
-
-    def test_mismatched_release_different_branch(self):
-        claim_body = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "branch: feat/branch-A",
-            "phase: claimed",
-            f"head_before: {HEAD_A}",
-            f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
-            "profile: prgs-author",
-        ])
-        release_body = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "branch: feat/branch-B",
-            "phase: released",
-            f"head_before: {HEAD_A}",
-            f"expires_at: {(NOW + timedelta(minutes=60)).isoformat().replace('+00:00', 'Z')}",
-            "profile: prgs-author",
-        ])
-        comments = [{"body": claim_body}, {"body": release_body}]
-        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
-        self.assertIsNotNone(lease)
-        self.assertEqual(lease["phase"], "claimed")
-
-    def test_release_followed_by_newer_claim(self):
-        claim_1 = _conflict_fix_body(phase="claimed", worktree="branches/fix-376")
-        expires = (NOW + timedelta(minutes=60)).isoformat().replace("+00:00", "Z")
-        release_1 = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "profile: prgs-author",
-            "phase: released",
-            f"head_before: {HEAD_A}",
-            f"head_after: {HEAD_B}",
-            f"expires_at: {expires}",
-        ])
-        claim_2 = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "profile: prgs-author",
-            "phase: claimed",
-            f"head_before: {HEAD_B}",
-            f"expires_at: {expires}",
-        ])
-        comments = [{"body": claim_1}, {"body": release_1}, {"body": claim_2}]
-        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
-        self.assertIsNotNone(lease)
-        self.assertEqual(lease["head_before"], HEAD_B)
-
-    def test_malformed_or_ambiguous_markers(self):
-        malformed_release = "\n".join([
-            CONFLICT_FIX_LEASE_MARKER,
-            "pr: #376",
-            "phase: released",
-            # missing head_before and profile
-        ])
-        claim_body = _conflict_fix_body(phase="claimed")
-        comments = [{"body": claim_body}, {"body": malformed_release}]
-        lease = find_active_conflict_fix_lease(comments, pr_number=376, now=NOW)
-        self.assertIsNotNone(lease)
-
-    def test_pr818_historical_sequence(self):
-        comment_14696 = "\n".join([
-            "<!-- mcp-conflict-fix-lease:v1 -->",
-            "pr: #818",
-            "branch: feat/issue-638-webui-app-shell-phase1",
-            "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
-            "profile: prgs-author",
-            "session_id: unknown",
-            "phase: claimed",
-            "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
-            "expires_at: 2026-07-23T07:12:13Z",
-            "reviewer_active: no",
-        ])
-        comment_14730 = "\n".join([
-            "<!-- mcp-conflict-fix-lease:v1 -->",
-            "pr: #818",
-            "branch: feat/issue-638-webui-app-shell-phase1",
-            "worktree: /Users/jasonwalker/Development/Gitea-Tools/branches/issue-638-webui-app-shell-phase1",
-            "profile: prgs-author",
-            "session_id: prgs-author-61241-e5129c60",
-            "phase: released",
-            "head_before: 08061b7b8aebdd099a37d1abf5dafcf38e4fd3fb",
-            "head_after: 64b6eb5d5402663098de5ded3b0617cc3b3df98f",
-            "expires_at: 2026-07-23T06:05:00Z",
-            "reviewer_active: no",
-        ])
-        comments = [{"body": comment_14696}, {"body": comment_14730}]
-        check_now = datetime(2026, 7, 23, 6, 30, tzinfo=timezone.utc)
-        lease = find_active_conflict_fix_lease(comments, pr_number=818, now=check_now)
-        self.assertIsNone(lease)
-        
-        reviewer_gate = assess_reviewer_mutation_blocked(
-            pr_number=818,
-            comments=comments,
-            reviewed_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
-            live_head_sha="64b6eb5d5402663098de5ded3b0617cc3b3df98f",
-            mutation="approve",
-            now=check_now,
-        )
-        self.assertTrue(reviewer_gate["mutation_allowed"])
-
-
 if __name__ == "__main__":
     unittest.main()