"""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()