From 5d82d90452112f6bde2ff2b99af0355f5181d4e4 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Sun, 5 Apr 2026 17:22:19 -0400 Subject: [PATCH] feat(merge-sweep): add failure history tracking and Track A-resolve [OMN-7573] Add FailureCategory enum, FailureHistoryEntry/Summary dataclasses, and cross-run failure escalation logic to NodeMergeSweep. CHRONIC (>=5 consecutive failures) and RECIDIVIST (>=3 polishes still failing) PRs skip polish automatically. Add Track A-resolve for MERGEABLE+BLOCKED+GREEN PRs (thread resolution). Fix is_merge_ready to exclude BLOCKED PRs so they fall through to A-resolve. Bump contract version to 1.1.0. Add 6 new tests covering failure history escalation. --- .../nodes/node_merge_sweep/contract.yaml | 12 +- .../handlers/handler_merge_sweep.py | 179 ++++++++++++++- tests/test_golden_chain_merge_sweep.py | 212 +++++++++++++++++- 3 files changed, 385 insertions(+), 18 deletions(-) diff --git a/src/omnimarket/nodes/node_merge_sweep/contract.yaml b/src/omnimarket/nodes/node_merge_sweep/contract.yaml index 0bb1199ede..f2e65b5cfc 100644 --- a/src/omnimarket/nodes/node_merge_sweep/contract.yaml +++ b/src/omnimarket/nodes/node_merge_sweep/contract.yaml @@ -1,13 +1,15 @@ --- name: merge_sweep -contract_version: {major: 1, minor: 0, patch: 0} +contract_version: {major: 1, minor: 1, patch: 0} node_type: orchestrator -node_version: {major: 1, minor: 0, patch: 0} +node_version: {major: 1, minor: 1, patch: 0} description: > - Org-wide PR sweep. Classifies open PRs into merge-ready (Track A) and - needs-polish (Track B) queues. Enables GitHub auto-merge on ready PRs - and dispatches polish work on fixable PRs. + Org-wide PR sweep. Classifies open PRs into merge-ready (Track A), + thread-resolution (Track A-resolve), and needs-polish (Track B) queues. + Enables GitHub auto-merge on ready PRs, resolves review threads, and + dispatches polish work on fixable PRs. Tracks cross-run failure history + with escalation thresholds (STUCK, CHRONIC, RECIDIVIST). handler_routing: default_handler: handler:NodeMergeSweep diff --git a/src/omnimarket/nodes/node_merge_sweep/handlers/handler_merge_sweep.py b/src/omnimarket/nodes/node_merge_sweep/handlers/handler_merge_sweep.py index 220a6c349a..f7f96973d8 100644 --- a/src/omnimarket/nodes/node_merge_sweep/handlers/handler_merge_sweep.py +++ b/src/omnimarket/nodes/node_merge_sweep/handlers/handler_merge_sweep.py @@ -3,6 +3,7 @@ Classifies open PRs into tracks: - Track A-update: Stale branches needing update before merge - Track A: Merge-ready PRs for auto-merge +- Track A-resolve: PRs blocked only by unresolved review threads - Track B: PRs with fixable blocking issues for polish ONEX node type: ORCHESTRATOR @@ -19,10 +20,27 @@ class PRTrack(Enum): A_UPDATE = "A-update" A_MERGE = "A" + A_RESOLVE = "A-resolve" B_POLISH = "B" SKIP = "skip" +class FailureCategory(Enum): + """Standardized failure category strings for cross-run failure history.""" + + CI_TEST = "ci_test" + CI_LINT = "ci_lint" + CI_GATE = "ci_gate" + PR_TITLE = "pr_title" + CONFLICT = "conflict" + CHANGES_REQUESTED = "changes_requested" + THREADS_BLOCKED = "threads_blocked" + BRANCH_STALE = "branch_stale" + SCAN_FAILED = "scan_failed" + POLISH_FAILED = "polish_failed" + NEEDS_HUMAN = "needs_human" + + @dataclass class PRInfo: """Minimal PR representation for classification.""" @@ -45,6 +63,33 @@ class ClassifiedPR: pr: PRInfo track: PRTrack reason: str + failure_categories: list[str] = field(default_factory=list) + + +@dataclass +class FailureHistoryEntry: + """Cross-run failure tracking for a single PR.""" + + first_seen: str + last_seen: str + consecutive_failures: int = 0 + total_failures: int = 0 + total_polishes: int = 0 + last_failure_categories: list[str] = field(default_factory=list) + last_result: str | None = None + last_run_id: str | None = None + runs_seen: list[str] = field(default_factory=list) + + +@dataclass +class FailureHistorySummary: + """Aggregate failure history stats for ModelSkillResult.""" + + total_tracked: int = 0 + stuck_prs: int = 0 + chronic_prs: int = 0 + recidivist_prs: int = 0 + cleaned_merged: int = 0 @dataclass @@ -56,6 +101,8 @@ class MergeSweepRequest: merge_method: str = "squash" max_total_merges: int = 0 skip_polish: bool = False + failure_history: dict[str, FailureHistoryEntry] = field(default_factory=dict) + run_id: str = "" @dataclass @@ -64,6 +111,9 @@ class MergeSweepResult: classified: list[ClassifiedPR] = field(default_factory=list) status: str = "nothing_to_merge" + failure_history_summary: FailureHistorySummary = field( + default_factory=FailureHistorySummary + ) @property def track_a_update(self) -> list[ClassifiedPR]: @@ -73,6 +123,10 @@ def track_a_update(self) -> list[ClassifiedPR]: def track_a_merge(self) -> list[ClassifiedPR]: return [c for c in self.classified if c.track == PRTrack.A_MERGE] + @property + def track_a_resolve(self) -> list[ClassifiedPR]: + return [c for c in self.classified if c.track == PRTrack.A_RESOLVE] + @property def track_b_polish(self) -> list[ClassifiedPR]: return [c for c in self.classified if c.track == PRTrack.B_POLISH] @@ -85,13 +139,22 @@ def skipped(self) -> list[ClassifiedPR]: class NodeMergeSweep: """Classify PRs into merge/polish/skip tracks.""" + # Escalation thresholds + STUCK_THRESHOLD = 3 + CHRONIC_THRESHOLD = 5 + RECIDIVIST_POLISH_THRESHOLD = 3 + def handle(self, request: MergeSweepRequest) -> MergeSweepResult: """Classify all PRs and return the result.""" classified: list[ClassifiedPR] = [] for pr in request.prs: - track, reason = self._classify_pr(pr, request.require_approval) - classified.append(ClassifiedPR(pr=pr, track=track, reason=reason)) + track, reason, categories = self._classify_pr(pr, request.require_approval) + classified.append( + ClassifiedPR( + pr=pr, track=track, reason=reason, failure_categories=categories + ) + ) # Apply max_total_merges cap to Track A if request.max_total_merges > 0: @@ -110,39 +173,73 @@ def handle(self, request: MergeSweepRequest) -> MergeSweepResult: c.track = PRTrack.SKIP c.reason = "Polish skipped (--skip-polish)" + # Apply failure history escalation to Track B + for c in classified: + if c.track == PRTrack.B_POLISH: + skip, skip_reason = self._should_skip_polish( + request.failure_history, f"{c.pr.repo}#{c.pr.number}" + ) + if skip: + c.track = PRTrack.SKIP + c.reason = skip_reason or "Skipped by failure history" + has_actionable = any( - c.track in (PRTrack.A_UPDATE, PRTrack.A_MERGE, PRTrack.B_POLISH) + c.track + in (PRTrack.A_UPDATE, PRTrack.A_MERGE, PRTrack.A_RESOLVE, PRTrack.B_POLISH) for c in classified ) status = "queued" if has_actionable else "nothing_to_merge" - return MergeSweepResult(classified=classified, status=status) + # Compute failure history summary + summary = self._compute_failure_summary(request.failure_history) + + return MergeSweepResult( + classified=classified, status=status, failure_history_summary=summary + ) - def _classify_pr(self, pr: PRInfo, require_approval: bool) -> tuple[PRTrack, str]: - """Classify a single PR. First match wins.""" + def _classify_pr( + self, pr: PRInfo, require_approval: bool + ) -> tuple[PRTrack, str, list[str]]: + """Classify a single PR. First match wins. Returns (track, reason, categories).""" if pr.is_draft: - return PRTrack.SKIP, "Draft PR" + return PRTrack.SKIP, "Draft PR", [] # Track A-update: stale branches or unknown mergeable state if self._needs_branch_update(pr): - return PRTrack.A_UPDATE, f"Branch stale ({pr.merge_state_status})" + return ( + PRTrack.A_UPDATE, + f"Branch stale ({pr.merge_state_status})", + [FailureCategory.BRANCH_STALE.value], + ) # Track A: merge-ready if self._is_merge_ready(pr, require_approval): - return PRTrack.A_MERGE, "Merge-ready" + return PRTrack.A_MERGE, "Merge-ready", [] + + # Track A-resolve: BLOCKED by unresolved threads only + if self._needs_thread_resolution(pr): + return ( + PRTrack.A_RESOLVE, + "Blocked by unresolved review threads", + [FailureCategory.THREADS_BLOCKED.value], + ) # Track B: fixable blocking issues if self._needs_polish(pr, require_approval): - reason_parts = [] + categories: list[str] = [] + reason_parts: list[str] = [] if pr.mergeable == "CONFLICTING": reason_parts.append("conflicts") + categories.append(FailureCategory.CONFLICT.value) if not pr.required_checks_pass: reason_parts.append("CI failing") + categories.append(FailureCategory.CI_TEST.value) if require_approval and pr.review_decision == "CHANGES_REQUESTED": reason_parts.append("changes requested") - return PRTrack.B_POLISH, "Needs polish: " + ", ".join(reason_parts) + categories.append(FailureCategory.CHANGES_REQUESTED.value) + return PRTrack.B_POLISH, "Needs polish: " + ", ".join(reason_parts), categories - return PRTrack.SKIP, "No actionable state" + return PRTrack.SKIP, "No actionable state", [] def _needs_branch_update(self, pr: PRInfo) -> bool: if pr.mergeable == "MERGEABLE": @@ -154,12 +251,24 @@ def _needs_branch_update(self, pr: PRInfo) -> bool: def _is_merge_ready(self, pr: PRInfo, require_approval: bool) -> bool: if pr.mergeable != "MERGEABLE": return False + if pr.merge_state_status.upper() == "BLOCKED": + return False # BLOCKED PRs may need thread resolution — fall through to A-resolve if not pr.required_checks_pass: return False if require_approval: return pr.review_decision in ("APPROVED", None) return True + def _needs_thread_resolution(self, pr: PRInfo) -> bool: + """MERGEABLE + BLOCKED + GREEN = blocked by required_conversation_resolution.""" + if pr.mergeable != "MERGEABLE": + return False + if pr.merge_state_status.upper() != "BLOCKED": + return False + if not pr.required_checks_pass: + return False + return True + def _needs_polish(self, pr: PRInfo, require_approval: bool) -> bool: if pr.mergeable == "UNKNOWN": return False @@ -172,3 +281,49 @@ def _needs_polish(self, pr: PRInfo, require_approval: bool) -> bool: if require_approval and pr.review_decision == "CHANGES_REQUESTED": return True return False + + def _should_skip_polish( + self, history: dict[str, FailureHistoryEntry], pr_key: str + ) -> tuple[bool, str | None]: + """Check if a PR should skip polish based on failure history.""" + entry = history.get(pr_key) + if not entry: + return False, None + if entry.consecutive_failures >= self.CHRONIC_THRESHOLD: + return ( + True, + f"CHRONIC: {entry.consecutive_failures} consecutive failures — skipping polish", + ) + if ( + entry.total_polishes >= self.RECIDIVIST_POLISH_THRESHOLD + and entry.consecutive_failures > 0 + ): + return ( + True, + f"RECIDIVIST: polished {entry.total_polishes}x, still failing — skipping polish", + ) + return False, None + + def _compute_failure_summary( + self, history: dict[str, FailureHistoryEntry] + ) -> FailureHistorySummary: + """Compute aggregate failure history stats.""" + stuck = 0 + chronic = 0 + recidivist = 0 + for entry in history.values(): + if entry.consecutive_failures >= self.CHRONIC_THRESHOLD: + chronic += 1 + elif entry.consecutive_failures >= self.STUCK_THRESHOLD: + stuck += 1 + if ( + entry.total_polishes >= self.RECIDIVIST_POLISH_THRESHOLD + and entry.consecutive_failures > 0 + ): + recidivist += 1 + return FailureHistorySummary( + total_tracked=len(history), + stuck_prs=stuck, + chronic_prs=chronic, + recidivist_prs=recidivist, + ) diff --git a/tests/test_golden_chain_merge_sweep.py b/tests/test_golden_chain_merge_sweep.py index 46bc0f3206..760836ae70 100644 --- a/tests/test_golden_chain_merge_sweep.py +++ b/tests/test_golden_chain_merge_sweep.py @@ -1,6 +1,6 @@ """Golden chain test for node_merge_sweep. -Verifies PR classification logic and event bus wiring. +Verifies PR classification logic, failure history tracking, and event bus wiring. """ from __future__ import annotations @@ -11,6 +11,8 @@ from omnibase_core.event_bus.event_bus_inmemory import EventBusInmemory from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( + FailureCategory, + FailureHistoryEntry, MergeSweepRequest, NodeMergeSweep, PRInfo, @@ -45,6 +47,7 @@ async def test_merge_ready_pr_classified_track_a( assert result.status == "queued" assert len(result.track_a_merge) == 1 assert result.track_a_merge[0].pr.number == 42 + assert result.track_a_merge[0].failure_categories == [] async def test_behind_pr_classified_track_a_update( self, event_bus: EventBusInmemory @@ -65,6 +68,28 @@ async def test_behind_pr_classified_track_a_update( assert len(result.track_a_update) == 1 assert result.track_a_update[0].track == PRTrack.A_UPDATE + assert FailureCategory.BRANCH_STALE.value in result.track_a_update[0].failure_categories + + async def test_blocked_green_pr_classified_track_a_resolve( + self, event_bus: EventBusInmemory + ) -> None: + """A MERGEABLE + BLOCKED + GREEN PR goes to Track A-resolve.""" + handler = NodeMergeSweep() + pr = PRInfo( + number=517, + title="feat: visibility toggles", + repo="OmniNode-ai/omnidash", + mergeable="MERGEABLE", + merge_state_status="BLOCKED", + review_decision=None, + required_checks_pass=True, + ) + request = MergeSweepRequest(prs=[pr]) + result = handler.handle(request) + + assert len(result.track_a_resolve) == 1 + assert result.track_a_resolve[0].track == PRTrack.A_RESOLVE + assert FailureCategory.THREADS_BLOCKED.value in result.track_a_resolve[0].failure_categories async def test_conflicting_pr_classified_track_b( self, event_bus: EventBusInmemory @@ -85,6 +110,26 @@ async def test_conflicting_pr_classified_track_b( assert len(result.track_b_polish) == 1 assert "conflicts" in result.track_b_polish[0].reason + assert FailureCategory.CONFLICT.value in result.track_b_polish[0].failure_categories + + async def test_ci_failing_pr_has_failure_categories( + self, event_bus: EventBusInmemory + ) -> None: + """A PR with CI failures should have ci_test in failure_categories.""" + handler = NodeMergeSweep() + pr = PRInfo( + number=200, + title="feat: broken tests", + repo="OmniNode-ai/omniclaude", + mergeable="MERGEABLE", + merge_state_status="BLOCKED", + required_checks_pass=False, + ) + request = MergeSweepRequest(prs=[pr]) + result = handler.handle(request) + + assert len(result.track_b_polish) == 1 + assert FailureCategory.CI_TEST.value in result.track_b_polish[0].failure_categories async def test_draft_pr_skipped(self, event_bus: EventBusInmemory) -> None: """Draft PRs should be skipped.""" @@ -140,6 +185,171 @@ async def test_skip_polish_flag(self, event_bus: EventBusInmemory) -> None: assert len(result.track_b_polish) == 0 assert len(result.skipped) == 1 + +@pytest.mark.unit +class TestFailureHistory: + """Cross-run failure history tracking and escalation.""" + + async def test_chronic_pr_skips_polish(self, event_bus: EventBusInmemory) -> None: + """A PR with >=5 consecutive failures should skip polish.""" + handler = NodeMergeSweep() + pr = PRInfo( + number=99, + title="fix: always broken", + repo="OmniNode-ai/test", + mergeable="CONFLICTING", + merge_state_status="DIRTY", + ) + history = { + "OmniNode-ai/test#99": FailureHistoryEntry( + first_seen="2026-04-01T00:00:00Z", + last_seen="2026-04-05T00:00:00Z", + consecutive_failures=5, + total_failures=5, + total_polishes=3, + last_failure_categories=["conflict"], + last_result="blocked", + last_run_id="ms-prev", + runs_seen=["ms-1", "ms-2", "ms-3", "ms-4", "ms-prev"], + ), + } + request = MergeSweepRequest(prs=[pr], failure_history=history) + result = handler.handle(request) + + assert len(result.track_b_polish) == 0 + assert len(result.skipped) == 1 + assert "CHRONIC" in result.skipped[0].reason + + async def test_recidivist_pr_skips_polish( + self, event_bus: EventBusInmemory + ) -> None: + """A PR polished >=3 times but still failing should skip polish.""" + handler = NodeMergeSweep() + pr = PRInfo( + number=42, + title="fix: keeps breaking", + repo="OmniNode-ai/test", + mergeable="MERGEABLE", + merge_state_status="BLOCKED", + required_checks_pass=False, + ) + history = { + "OmniNode-ai/test#42": FailureHistoryEntry( + first_seen="2026-04-01T00:00:00Z", + last_seen="2026-04-05T00:00:00Z", + consecutive_failures=1, + total_failures=4, + total_polishes=3, + last_failure_categories=["ci_test"], + last_result="polished_and_queued", + last_run_id="ms-prev", + runs_seen=["ms-1", "ms-2", "ms-3", "ms-prev"], + ), + } + request = MergeSweepRequest(prs=[pr], failure_history=history) + result = handler.handle(request) + + assert len(result.track_b_polish) == 0 + assert len(result.skipped) == 1 + assert "RECIDIVIST" in result.skipped[0].reason + + async def test_stuck_pr_still_gets_polished( + self, event_bus: EventBusInmemory + ) -> None: + """A PR with 3-4 consecutive failures (STUCK) should still get polished.""" + handler = NodeMergeSweep() + pr = PRInfo( + number=10, + title="fix: struggling", + repo="OmniNode-ai/test", + mergeable="CONFLICTING", + merge_state_status="DIRTY", + ) + history = { + "OmniNode-ai/test#10": FailureHistoryEntry( + first_seen="2026-04-03T00:00:00Z", + last_seen="2026-04-05T00:00:00Z", + consecutive_failures=3, + total_failures=3, + total_polishes=1, + last_failure_categories=["conflict"], + last_result="blocked", + last_run_id="ms-prev", + runs_seen=["ms-1", "ms-2", "ms-prev"], + ), + } + request = MergeSweepRequest(prs=[pr], failure_history=history) + result = handler.handle(request) + + # STUCK but still below CHRONIC threshold — should still polish + assert len(result.track_b_polish) == 1 + + async def test_failure_history_summary_computed( + self, event_bus: EventBusInmemory + ) -> None: + """Failure history summary should aggregate stats correctly.""" + handler = NodeMergeSweep() + pr = PRInfo( + number=1, + title="test", + repo="test/repo", + mergeable="MERGEABLE", + merge_state_status="CLEAN", + review_decision="APPROVED", + required_checks_pass=True, + ) + history = { + "test/repo#10": FailureHistoryEntry( + first_seen="2026-04-01T00:00:00Z", + last_seen="2026-04-05T00:00:00Z", + consecutive_failures=3, + total_failures=3, + ), + "test/repo#20": FailureHistoryEntry( + first_seen="2026-04-01T00:00:00Z", + last_seen="2026-04-05T00:00:00Z", + consecutive_failures=6, + total_failures=6, + ), + "test/repo#30": FailureHistoryEntry( + first_seen="2026-04-01T00:00:00Z", + last_seen="2026-04-05T00:00:00Z", + consecutive_failures=1, + total_failures=4, + total_polishes=3, + ), + } + request = MergeSweepRequest(prs=[pr], failure_history=history) + result = handler.handle(request) + + assert result.failure_history_summary.total_tracked == 3 + assert result.failure_history_summary.stuck_prs == 1 # #10 (3 consecutive) + assert result.failure_history_summary.chronic_prs == 1 # #20 (6 consecutive) + assert result.failure_history_summary.recidivist_prs == 1 # #30 (3 polishes, still failing) + + async def test_no_failure_history_still_works( + self, event_bus: EventBusInmemory + ) -> None: + """Handler works correctly with empty failure history.""" + handler = NodeMergeSweep() + pr = PRInfo( + number=1, + title="feat: new", + repo="test/repo", + mergeable="CONFLICTING", + merge_state_status="DIRTY", + ) + request = MergeSweepRequest(prs=[pr]) + result = handler.handle(request) + + assert len(result.track_b_polish) == 1 + assert result.failure_history_summary.total_tracked == 0 + + +@pytest.mark.unit +class TestMergeSweepEventBus: + """Event bus wiring tests.""" + async def test_event_bus_wiring(self, event_bus: EventBusInmemory) -> None: """Handler can be wired to event bus for command/completion flow.""" handler = NodeMergeSweep()