From 3c8f6a1215aef6fda17e304df7762742f169fea1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ond=C5=99ej=20Poho=C5=99elsk=C3=BD?= Date: Thu, 16 Jul 2026 15:07:25 +0200 Subject: [PATCH 1/3] Enable rebuild+backport MR consolidation The consolidation agent now discovers both ymir_backport and ymir_rebuild MRs. When a package has one backport and one rebuild MR open for the same branch, the rebuild is folded into the backport via an append-only flow: the backport branch is cherry-picked as base and the rebuild ticket(s) plus CVEs are appended to the commit message without a second Release bump. Selection priority: backport+backport pairs are still consolidated first; backport+rebuild only when fewer than 2 backport MRs exist. The rebuild agent now submits a consolidation job after creating its MR (same mechanism as the backport agent), so the consolidation agent can pick up both MR types on the next queue poll. CVE handling: CVE IDs are extracted from both MR descriptions and merged into the consolidated commit only if the base backport already included a CVE: line (respecting per-package maintainer rules). Assisted-by: Claude Opus 4.6 (Cursor) --- docs/mr_consolidation_architecture.md | 41 +- jira_label_workflow_routing.md | 1 + ymir/agents/mr_consolidation_agent.py | 434 +++++++++++++++--- ymir/agents/rebuild_agent.py | 54 +++ .../tests/unit/test_mr_consolidation.py | 107 ++++- 5 files changed, 571 insertions(+), 66 deletions(-) diff --git a/docs/mr_consolidation_architecture.md b/docs/mr_consolidation_architecture.md index 3df73756b..ed7742e7f 100644 --- a/docs/mr_consolidation_architecture.md +++ b/docs/mr_consolidation_architecture.md @@ -18,11 +18,17 @@ originals as consolidated. ┌──────────────┐ submit_merge_job() ┌────────────────────┐ │ │ ─────────────────────────────▶ │ │ │ Backport │ (after filing an MR, if │ Redis Queue │ -│ Agent │ ymir.yaml enables it │ (Hash-based, │ -│ │ it for the package) │ per pkg/branch) │ +│ Agent │ ymir.yaml enables it) │ (Hash-based, │ +│ │ │ per pkg/branch) │ └──────────────┘ └────────┬───────────┘ │ ┌──────────────┐ submit_merge_job() │ +│ Rebuild │ ─────────────────────────────▶ │ +│ Agent │ (after filing an MR, if │ +│ │ ymir.yaml enables it) │ +└──────────────┘ │ + │ +┌──────────────┐ submit_merge_job() │ │ Jira Issue │ ─────────────────────────────▶ │ │ Fetcher │ (when ymir_consolidate_base │ │ │ + _next labels match) │ @@ -38,7 +44,7 @@ originals as consolidated. │ 1. list_open_mrs │ │ 2. fork & clone │ │ 3. filter by HEAD │ - │ 4. LLM merges │ + │ 4. merge/append │ │ 5. build verify │ │ 6. log & commit │ │ 7. mark originals │ @@ -295,6 +301,34 @@ exactly those two MRs — it does **not** fall back to the "oldest two" heuristi `ymir_consolidate_next` for the same package/branch (or vice versa), the fetcher logs a warning and leaves the labels untouched for the next sweep. +## Rebuild + Backport Consolidation + +When a package has both an open backport MR (`ymir_backport`) and a rebuild MR +(`ymir_rebuild`) for the same branch, the consolidation agent can fold the rebuild +into the backport. This avoids a double Release bump since the backport already +bumps Release. + +### How it works + +1. **Discovery**: `list_open_mrs` searches for both `ymir_backport` and `ymir_rebuild` + labeled MRs. At least one backport MR must be present. +2. **Selection priority**: backport+backport pairs are consolidated first. Only when + fewer than 2 backport MRs exist does the agent pick a backport+rebuild pair. +3. **Append-only flow**: The backport branch is cherry-picked as the base. The rebuild + branch is NOT cherry-picked (its Release bump + changelog are redundant). Instead, + only the rebuild's Jira ticket(s) and CVE(s) are appended to the commit message. +4. **CVE propagation**: CVE IDs from both MRs are collected. The `CVE:` line in the + commit message is updated only if the base backport already had one (respecting + per-package maintainer rules that may omit it). +5. **Output**: The consolidated MR is labeled `ymir_backport`. The original rebuild MR + is labeled `ymir_consolidated`. + +### Triggering + +Both the backport agent and the rebuild agent submit consolidation jobs after +creating their MRs (when `ymir.yaml` enables consolidation). The consolidation +agent discovers eligible pairs regardless of which agent triggered the job. + ## Safety Invariants | Invariant | Mechanism | @@ -321,6 +355,7 @@ a warning and leaves the labels untouched for the next sweep. | `ymir/common/constants.py` | `MERGE_CONSOLIDATION_QUEUE` Redis key name | | `ymir/tools/privileged/gitlab.py` | `ListProjectMergeRequestsTool`, `AddMergeRequestLabelsTool` | | `ymir/agents/backport_agent.py` | `submit_consolidation_job` step (triggers consolidation) | +| `ymir/agents/rebuild_agent.py` | `submit_consolidation_job` step (triggers consolidation for rebuild+backport) | | `ymir/jira_issue_fetcher/jira_issue_fetcher.py` | Label-triggered consolidation scanning (`_process_consolidation_labels`) | | `ymir/agents/tests/unit/test_mr_consolidation.py` | Unit tests for Redis queue logic | | `ymir/agents/tests/e2e/mr_consolidation/` | E2E test suite with LLM judge evaluation | diff --git a/jira_label_workflow_routing.md b/jira_label_workflow_routing.md index 39f686aed..154f0760b 100644 --- a/jira_label_workflow_routing.md +++ b/jira_label_workflow_routing.md @@ -103,6 +103,7 @@ These labels are applied to GitLab merge requests (not Jira issues): | Label | Purpose | Effect | |-------|---------|--------| | `ymir_backport` | Marks an MR as a backport | Used by the consolidation agent to discover candidate MRs | +| `ymir_rebuild` | Marks an MR as a rebuild | Used by the consolidation agent to discover rebuild MRs for backport+rebuild consolidation | | `ymir_consolidated` | Marks an MR that has been folded into a consolidated MR | The MR stays open but is excluded from future consolidation searches | ## Queue Types Summary diff --git a/ymir/agents/mr_consolidation_agent.py b/ymir/agents/mr_consolidation_agent.py index dfc69f9db..688779df8 100644 --- a/ymir/agents/mr_consolidation_agent.py +++ b/ymir/agents/mr_consolidation_agent.py @@ -139,6 +139,8 @@ class ConsolidationState(PackageUpdateState): patches_per_mr: dict[str, list[str]] = Field(default_factory=dict) all_open_mrs: list[dict] = Field(default_factory=list) current_mrs_count: int = Field(default=0) + mr_types: dict[str, str] = Field(default_factory=dict) + cves_collected: list[str] = Field(default_factory=list) def _extract_jira_issues_from_description(description: str) -> list[str]: @@ -173,6 +175,22 @@ def _extract_jira_issues_from_description(description: str) -> list[str]: return list(set(issues)) +def _extract_cves_from_text(text: str) -> list[str]: + """Extract CVE IDs (e.g. CVE-2024-12345) from arbitrary text.""" + return sorted(set(re.findall(r"CVE-\d{4}-\d+", text))) + + +_CONSOLIDATION_MR_LABELS = ["ymir_backport", "ymir_rebuild"] + + +def _mr_type_from_labels(mr: dict) -> str: + """Determine if an MR is a backport or rebuild from its GitLab labels.""" + labels = mr.get("labels") or [] + if "ymir_rebuild" in labels: + return "rebuild" + return "backport" + + async def _resolve_source_issues( state, project_path: str, @@ -188,27 +206,31 @@ async def _resolve_source_issues( """ matched_mrs = [] for issue_key in issue_keys: - kwargs: dict[str, Any] = { - "project": project_path, - "state": "opened", - "labels": ["ymir_backport"], - "order_by": "updated_at", - "sort": "desc", - "available_tools": gateway_tools, - } - if target_branch: - kwargs["target_branch"] = target_branch - result = await run_tool("list_project_merge_requests", **kwargs) - mrs = json.loads(result) if isinstance(result, str) else result - mrs = [m for m in mrs if JiraLabels.MR_CONSOLIDATED.value not in m.get("labels", [])] + all_mrs_for_issue: list[dict] = [] + for label in _CONSOLIDATION_MR_LABELS: + kwargs: dict[str, Any] = { + "project": project_path, + "state": "opened", + "labels": [label], + "order_by": "updated_at", + "sort": "desc", + "available_tools": gateway_tools, + } + if target_branch: + kwargs["target_branch"] = target_branch + result = await run_tool("list_project_merge_requests", **kwargs) + mrs = json.loads(result) if isinstance(result, str) else result + all_mrs_for_issue.extend( + m for m in mrs if JiraLabels.MR_CONSOLIDATED.value not in m.get("labels", []) + ) mr = next( - (m for m in mrs if issue_key in m.get("title", "")), + (m for m in all_mrs_for_issue if issue_key in m.get("title", "")), None, ) if not mr: mr = next( - (m for m in mrs if issue_key in m.get("description", "")), + (m for m in all_mrs_for_issue if issue_key in m.get("description", "")), None, ) @@ -227,21 +249,42 @@ async def _resolve_source_issues( matched_mrs.append(mr) logger.info( - "Resolved %s to MR '%s' (branch: %s)", + "Resolved %s to MR '%s' (branch: %s, type: %s)", issue_key, mr["title"], mr["source_branch"], + _mr_type_from_labels(mr), ) + seen_urls: set[str] = set() + unique_mrs: list[dict] = [] + for mr in matched_mrs: + if mr["url"] not in seen_urls: + seen_urls.add(mr["url"]) + unique_mrs.append(mr) + matched_mrs = unique_mrs + + if len(matched_mrs) < 2: + logger.info("Fewer than 2 unique MRs resolved, nothing to consolidate") + state.consolidation_result = MRConsolidationOutputSchema( + success=True, + status="Fewer than 2 unique MRs resolved; nothing to do.", + ) + return Workflow.END + state.mr_urls = [mr["url"] for mr in matched_mrs] state.mr_branches = [mr["source_branch"] for mr in matched_mrs] state.mr_titles = [mr["title"] for mr in matched_mrs] state.mr_descriptions = [mr.get("description", "") for mr in matched_mrs] + state.mr_types = {mr["source_branch"]: _mr_type_from_labels(mr) for mr in matched_mrs} all_jira: list[str] = list(issue_keys) + all_cves: list[str] = [] for mr in matched_mrs: all_jira.extend(_extract_jira_issues_from_description(mr.get("description", ""))) + all_cves.extend(_extract_cves_from_text(mr.get("description", ""))) state.jira_issues_collected = sorted(set(all_jira)) + state.cves_collected = sorted(set(all_cves)) state.jira_issue = state.jira_issues_collected[0] if state.jira_issues_collected else None logger.info( @@ -298,7 +341,7 @@ async def run_workflow( workflow = Workflow(ConsolidationState, name="MRConsolidationWorkflow") async def list_open_mrs(state): - """List open backport MRs for the package/branch and pick the two oldest.""" + """List open backport and rebuild MRs for the package/branch.""" if state.mr_branches: logger.info( "Branch info pre-populated (%s), skipping GitLab MR lookup", @@ -318,24 +361,32 @@ async def list_open_mrs(state): target_branch=dist_git_branch, ) - result = await run_tool( - "list_project_merge_requests", - project=project_path, - state="opened", - target_branch=dist_git_branch, - labels=["ymir_backport"], - order_by="created_at", - sort="asc", - available_tools=gateway_tools, - ) - - mrs = json.loads(result) if isinstance(result, str) else result - mrs = [mr for mr in mrs if JiraLabels.MR_CONSOLIDATED.value not in mr.get("labels", [])] - state.all_open_mrs = mrs - - if len(mrs) < 2: + all_mrs: list[dict] = [] + seen_urls: set[str] = set() + for label in _CONSOLIDATION_MR_LABELS: + result = await run_tool( + "list_project_merge_requests", + project=project_path, + state="opened", + target_branch=dist_git_branch, + labels=[label], + order_by="created_at", + sort="asc", + available_tools=gateway_tools, + ) + mrs = json.loads(result) if isinstance(result, str) else result + for mr in mrs: + if JiraLabels.MR_CONSOLIDATED.value in mr.get("labels", []): + continue + if mr.get("url") in seen_urls: + continue + seen_urls.add(mr.get("url")) + all_mrs.append(mr) + + has_backport = any(_mr_type_from_labels(mr) == "backport" for mr in all_mrs) + if not has_backport or len(all_mrs) < 2: logger.info( - "Fewer than 2 open MRs for %s/%s, nothing to consolidate", + "Fewer than 2 open MRs (or no backport MR) for %s/%s, nothing to consolidate", package, dist_git_branch, ) @@ -345,6 +396,7 @@ async def list_open_mrs(state): ) return Workflow.END + state.all_open_mrs = all_mrs return "fork_and_prepare_dist_git" async def fork_and_prepare_dist_git(state): @@ -469,16 +521,42 @@ async def fork_and_prepare_dist_git(state): ) return Workflow.END - oldest_two = current_mrs[:2] - state.mr_urls = [mr["url"] for mr in oldest_two] - state.mr_branches = [mr["source_branch"] for mr in oldest_two] - state.mr_titles = [mr["title"] for mr in oldest_two] - state.mr_descriptions = [mr.get("description", "") for mr in oldest_two] + # Selection priority: backport+backport first; backport+rebuild + # only when fewer than 2 backport MRs exist. + backport_mrs = [mr for mr in current_mrs if _mr_type_from_labels(mr) == "backport"] + rebuild_mrs = [mr for mr in current_mrs if _mr_type_from_labels(mr) == "rebuild"] - all_jira = [] - for mr in oldest_two: + if len(backport_mrs) >= 2: + selected = backport_mrs[:2] + elif backport_mrs and rebuild_mrs: + selected = [backport_mrs[0], rebuild_mrs[0]] + elif not backport_mrs: + logger.info( + "No backport MRs on current HEAD for %s/%s, cannot consolidate rebuild-only", + package, + dist_git_branch, + ) + state.consolidation_result = MRConsolidationOutputSchema( + success=True, + status="No backport MR on current HEAD; rebuild-only consolidation not supported.", + ) + return Workflow.END + else: + selected = current_mrs[:2] + + state.mr_urls = [mr["url"] for mr in selected] + state.mr_branches = [mr["source_branch"] for mr in selected] + state.mr_titles = [mr["title"] for mr in selected] + state.mr_descriptions = [mr.get("description", "") for mr in selected] + state.mr_types = {mr["source_branch"]: _mr_type_from_labels(mr) for mr in selected} + + all_jira: list[str] = [] + all_cves: list[str] = [] + for mr in selected: all_jira.extend(_extract_jira_issues_from_description(mr.get("description", ""))) + all_cves.extend(_extract_cves_from_text(mr.get("description", ""))) state.jira_issues_collected = sorted(set(all_jira)) + state.cves_collected = sorted(set(all_cves)) state.jira_issue = state.jira_issues_collected[0] if state.jira_issues_collected else None if state.jira_issues_collected: current_jira_issue.set(",".join(state.jira_issues_collected)) @@ -486,8 +564,9 @@ async def fork_and_prepare_dist_git(state): state.current_mrs_count = len(current_mrs) logger.info( - "Selected MRs for consolidation (same HEAD): %s", - [mr["url"] for mr in oldest_two], + "Selected MRs for consolidation (same HEAD): %s (types: %s)", + [mr["url"] for mr in selected], + state.mr_types, ) for branch_name in state.mr_branches: @@ -517,6 +596,12 @@ async def fork_and_prepare_dist_git(state): state.patches_per_mr[branch_name] = new_patches logger.info("Per-MR patch mapping: %s", state.patches_per_mr) + # If all non-base branches are rebuild MRs, use the simplified + # append-only flow (no cherry-pick, no Release bump for rebuilds). + non_backport = [b for b in state.mr_branches if state.mr_types.get(b) == "rebuild"] + if non_backport and len(non_backport) == len(state.mr_branches) - 1: + return "rebuild_append_flow" + return "per_commit_flow" if release_strategy == "per_commit" else "run_consolidation_agent" async def run_consolidation_agent(state): @@ -576,6 +661,16 @@ async def run_build_agent(state): jira_issue=state.jira_issue, ), ) + + def _retry_step_for_build(state): + """Determine which flow to retry on build failure.""" + has_rebuild_other = any(t == "rebuild" for t in state.mr_types.values()) + if has_rebuild_other and len(state.mr_types) > 1: + return "rebuild_append_flow" + if release_strategy == "per_commit": + return "per_commit_flow" + return "run_consolidation_agent" + try: build_result = await build_agent.run( build_prompt, @@ -586,21 +681,15 @@ async def run_build_agent(state): if not build_output.success: state.build_error = build_output.error state.attempts_remaining -= 1 - retry_step = ( - "per_commit_flow" if release_strategy == "per_commit" else "run_consolidation_agent" - ) if state.attempts_remaining > 0: - return retry_step + return _retry_step_for_build(state) return "handle_failure" except Exception as e: logger.error("Build verification error: %s", e) state.build_error = str(e) state.attempts_remaining -= 1 - retry_step = ( - "per_commit_flow" if release_strategy == "per_commit" else "run_consolidation_agent" - ) if state.attempts_remaining > 0: - return retry_step + return _retry_step_for_build(state) return "handle_failure" if release_strategy == "per_commit": @@ -922,6 +1011,196 @@ async def per_commit_flow(state): return "run_build_agent" + async def rebuild_append_flow(state): + """Append rebuild ticket(s) to the backport MR without cherry-picking. + + The backport branch is cherry-picked as base (it has patches + Release + bump). Rebuild branches are skipped entirely — only their Jira tickets + and CVEs are recorded for inclusion in the final commit message and + changelog. + """ + git_env = local_tool_options.get("env") + spec_name = f"{package}.spec" + + try: + # The backport branch is always the base + backport_branches = [b for b in state.mr_branches if state.mr_types.get(b) != "rebuild"] + rebuild_branches = [b for b in state.mr_branches if state.mr_types.get(b) == "rebuild"] + + if not backport_branches: + raise RuntimeError("No backport branch found for rebuild_append_flow") + + base_branch = backport_branches[0] + + # Cherry-pick the backport branch as-is + _, base_commits_raw, _ = await run_subprocess( + ["git", "rev-list", "--reverse", f"{dist_git_branch}..{base_branch}"], + cwd=state.local_clone, + env=git_env, + ) + base_commits = [c for c in (base_commits_raw or "").strip().splitlines() if c] + + if not base_commits: + raise RuntimeError( + f"No commits found on backport branch {base_branch} " + f"relative to {dist_git_branch}. Cannot consolidate." + ) + + logger.info( + "Cherry-picking %d commit(s) from backport branch %s", + len(base_commits), + base_branch, + ) + await check_subprocess( + ["git", "cherry-pick", *base_commits], + cwd=state.local_clone, + env=git_env, + ) + + # The base branch may have added new source tarballs to the + # lookaside cache (updated `sources` file). Re-download so + # subsequent SRPM builds find them. + await run_tool( + "download_sources", + dist_git_path=str(state.local_clone), + package=package, + dist_git_branch=dist_git_branch, + available_tools=gateway_tools, + ) + logger.info( + "Waiting %ds for NFS attribute cache after source download", + _NFS_CACHE_WAIT, + ) + await asyncio.sleep(_NFS_CACHE_WAIT) + + # Collect Jira issues and CVEs from rebuild branches + rebuild_jira: list[str] = [] + rebuild_cves: list[str] = [] + for rb_branch in rebuild_branches: + idx = state.mr_branches.index(rb_branch) + desc = state.mr_descriptions[idx] if idx < len(state.mr_descriptions) else "" + rebuild_jira.extend(_extract_jira_issues_from_description(desc)) + rebuild_cves.extend(_extract_cves_from_text(desc)) + + # Also extract CVEs from the base backport commit messages + for commit_sha in base_commits: + _, msg, _ = await run_subprocess( + ["git", "log", "-1", "--format=%B", commit_sha], + cwd=state.local_clone, + env=git_env, + ) + state.cves_collected = sorted( + set(state.cves_collected + _extract_cves_from_text(msg or "")) + ) + + state.cves_collected = sorted(set(state.cves_collected + rebuild_cves)) + + # Update changelog with the rebuild ticket reference + rebuild_issues_str = ", ".join(sorted(set(rebuild_jira))) or "rebuild" + log_prompt = render_template( + get_log_prompt(), + LogInputSchema( + jira_issue=rebuild_issues_str, + changes_summary=( + f"Rebuild for {rebuild_issues_str} consolidated into backport. " + f"No additional source changes — the backport's Release bump covers " + f"both the backport and the rebuild." + ), + ), + ) + try: + await log_agent.run( + log_prompt, + expected_output=LogOutputSchema, + **get_agent_execution_config(), + ) + except Exception as e: + logger.warning("Log agent error (non-fatal) for rebuild append: %s", e) + + await tasks.stage_changes( + local_clone=state.local_clone, + files_to_commit=[spec_name], + ) + + # Amend the last backport commit to include rebuild tickets + _, last_msg, _ = await run_subprocess( + ["git", "log", "-1", "--format=%B"], + cwd=state.local_clone, + env=git_env, + ) + last_msg = (last_msg or "").strip() + + # Append rebuild Resolves: compare against what the commit + # message already contains, not state.jira_issues_collected + # (which includes rebuild tickets from discovery). + already_in_commit = set(re.findall(r"RHEL-\d+", last_msg)) + extra_resolves = sorted(set(rebuild_jira) - already_in_commit) + if extra_resolves: + state.jira_issues_collected = sorted(set(state.jira_issues_collected + extra_resolves)) + + # Add rebuild CVEs to commit message only if original already had CVE line + if rebuild_cves and re.search(r"^CVE:", last_msg, re.MULTILINE): + existing_cves = _extract_cves_from_text(last_msg) + all_cves = sorted(set(existing_cves + rebuild_cves)) + last_msg = re.sub( + r"^CVE:.*$", + f"CVE: {', '.join(all_cves)}", + last_msg, + count=1, + flags=re.MULTILINE, + ) + + # Append rebuild tickets to Resolves line + if extra_resolves: + resolves_addition = ", ".join(extra_resolves) + if re.search(r"^Resolves:", last_msg, re.MULTILINE): + last_msg = re.sub( + r"^(Resolves:.*)$", + rf"\1, {resolves_addition}", + last_msg, + count=1, + flags=re.MULTILINE, + ) + else: + last_msg += f"\nResolves: {resolves_addition}" + + await check_subprocess( + ["git", "commit", "--amend", "-m", last_msg], + cwd=state.local_clone, + env=git_env, + ) + + # Build SRPM for verification + srpm_tool = BuildSrpmTool(options=local_tool_options) + srpm_result = await srpm_tool.run( + input=BuildSrpmInput( + dist_git_path=state.local_clone, + package=package, + dist_git_branch=dist_git_branch, + ), + ) + output = srpm_result.result + srpm_path = output.strip() if "FAILED" not in output else None + state.consolidation_result = MRConsolidationOutputSchema( + success=srpm_path is not None, + status="rebuild_append consolidation complete", + srpm_path=srpm_path, + ) + if not srpm_path: + state.consolidation_result.error = output + return "handle_failure" + + except Exception as e: + logger.error("rebuild_append flow error: %s", e) + state.consolidation_result = MRConsolidationOutputSchema( + success=False, + status="rebuild_append flow failed", + error=str(e), + ) + return "handle_failure" + + return "run_build_agent" + async def update_release(state): try: await tasks.update_release( @@ -1057,14 +1336,20 @@ async def commit_push_and_open_mr(state): async def push_and_open_mr(state): """Push commits and open the consolidated MR on GitLab.""" + has_rebuild = any(t == "rebuild" for t in state.mr_types.values()) combined_description = _build_consolidated_description( state.mr_titles, state.mr_descriptions, state.mr_urls, state.jira_issues_collected, package, + cves=state.cves_collected, + has_rebuild=has_rebuild, ) - title = f"[Consolidated] Backport fixes for {package} ({dist_git_branch})" + if has_rebuild: + title = f"[Consolidated] Backport + rebuild fixes for {package} ({dist_git_branch})" + else: + title = f"[Consolidated] Backport fixes for {package} ({dist_git_branch})" try: if dry_run: @@ -1154,11 +1439,18 @@ async def update_jira_issues(state): if not state.jira_issues_collected or not state.merge_request_url: return "requeue_if_needed" + has_rebuild = any(t == "rebuild" for t in state.mr_types.values()) for issue_key in state.jira_issues_collected: - comment = ( - f"Your backport MR has been consolidated with other fixes " - f"into a single MR: {state.merge_request_url}" - ) + if has_rebuild: + comment = ( + f"Your MR has been consolidated (backport + rebuild) " + f"into a single MR: {state.merge_request_url}" + ) + else: + comment = ( + f"Your backport MR has been consolidated with other fixes " + f"into a single MR: {state.merge_request_url}" + ) if dry_run: logger.info( "Dry run: would post consolidation comment on %s", @@ -1219,6 +1511,7 @@ async def handle_failure(state): workflow.add_step("run_consolidation_agent", run_consolidation_agent) workflow.add_step("run_build_agent", run_build_agent) workflow.add_step("per_commit_flow", per_commit_flow) + workflow.add_step("rebuild_append_flow", rebuild_append_flow) workflow.add_step("update_release", update_release) workflow.add_step("stage_changes", stage_changes) workflow.add_step("run_log_agent", run_log_agent) @@ -1302,6 +1595,8 @@ def _build_consolidated_description( mr_urls: list[str], jira_issues: list[str], package: str, + cves: list[str] | None = None, + has_rebuild: bool = False, ) -> str: """Build the description for the consolidated MR. @@ -1315,12 +1610,27 @@ def _build_consolidated_description( desc = mr_descriptions[i] if i < len(mr_descriptions) else None all_sub_mrs.extend(_extract_sub_mrs(title, desc or "", url)) - parts = [ - "## Consolidated Backport MR\n", - f"This MR consolidates multiple backport merge requests {I_AM_YMIR} " - "Learn more about [MR consolidation and configuration of its behavior]" - "(https://ymir.pages.redhat.com/docs/agents/mr-consolidation/).\n", - ] + if has_rebuild: + header = "## Consolidated Backport + Rebuild MR\n" + intro = ( + f"This MR consolidates backport and rebuild merge requests {I_AM_YMIR} " + "Learn more about [MR consolidation and configuration of its behavior]" + "(https://ymir.pages.redhat.com/docs/agents/mr-consolidation/).\n" + ) + else: + header = "## Consolidated Backport MR\n" + intro = ( + f"This MR consolidates multiple backport merge requests {I_AM_YMIR} " + "Learn more about [MR consolidation and configuration of its behavior]" + "(https://ymir.pages.redhat.com/docs/agents/mr-consolidation/).\n" + ) + + parts = [header, intro] + + if cves: + parts.append("### CVEs Fixed\n") + parts.extend(f"- {cve}" for cve in cves) + parts.append("") if jira_issues: parts.append("### Resolved Jira Issues\n") diff --git a/ymir/agents/rebuild_agent.py b/ymir/agents/rebuild_agent.py index b97c1c41c..642e00c41 100644 --- a/ymir/agents/rebuild_agent.py +++ b/ymir/agents/rebuild_agent.py @@ -15,6 +15,7 @@ from ymir.agents.log_agent import get_prompt as get_log_prompt from ymir.agents.observability import setup_observability from ymir.agents.package_update_steps import PackageUpdateState +from ymir.agents.tasks import InvalidConsolidationConfigError, fetch_consolidation_config, submit_merge_job from ymir.agents.utils import ( format_mr_triage_details, get_agent_execution_config, @@ -77,6 +78,7 @@ async def run_workflow( consolidation_summary=None, side_tag=None, user_triggered=False, + redis_conn=None, ): local_tool_options = {"working_directory": None} if mock_env := get_mock_local_tool_env(jira_issue): @@ -265,6 +267,56 @@ async def commit_push_and_open_mr(state): state.merge_request_url = None state.rebuild_success = False state.rebuild_error = f"Could not commit and open MR: {e}" + return "submit_consolidation_job" + + async def submit_consolidation_job(state): + if ( + not state.merge_request_url + or not state.rebuild_success + or not state.merge_request_newly_created + ): + return "comment_in_jira" + + try: + config = await fetch_consolidation_config(state.package, gateway_tools) + except InvalidConsolidationConfigError as e: + logger.warning("Invalid consolidation config for %s: %s", state.package, e) + return "comment_in_jira" + + try: + if not config.merge_mrs: + logger.info( + "MR consolidation not enabled for %s, skipping", + state.package, + ) + return "comment_in_jira" + + if redis_conn is not None: + submitted = await submit_merge_job( + redis_conn, + state.package, + state.dist_git_branch, + release_strategy=config.release_strategy.value, + ) + if submitted: + logger.info( + "Submitted consolidation job for %s/%s", + state.package, + state.dist_git_branch, + ) + else: + logger.info( + "Consolidation job already queued for %s/%s", + state.package, + state.dist_git_branch, + ) + else: + logger.info( + "No Redis connection (direct mode), skipping consolidation job submission" + ) + except Exception as e: + logger.warning("Failed to submit consolidation job: %s", e) + return "comment_in_jira" async def comment_in_jira(state): @@ -302,6 +354,7 @@ async def comment_in_jira(state): workflow.add_step("stage_changes", stage_changes) workflow.add_step("run_log_agent", run_log_agent) workflow.add_step("commit_push_and_open_mr", commit_push_and_open_mr) + workflow.add_step("submit_consolidation_job", submit_consolidation_job) workflow.add_step("comment_in_jira", comment_in_jira) response = await workflow.run( @@ -468,6 +521,7 @@ async def retry( consolidation_summary=rebuild_data.consolidation_summary, side_tag=rebuild_data.side_tag, user_triggered=user_triggered, + redis_conn=redis, ) logger.info( f"Rebuild processing completed for {rebuild_data.jira_issue}, " diff --git a/ymir/agents/tests/unit/test_mr_consolidation.py b/ymir/agents/tests/unit/test_mr_consolidation.py index c6b206eb8..4b77e97fb 100644 --- a/ymir/agents/tests/unit/test_mr_consolidation.py +++ b/ymir/agents/tests/unit/test_mr_consolidation.py @@ -4,7 +4,11 @@ import pytest -from ymir.agents.mr_consolidation_agent import _extract_jira_issues_from_description +from ymir.agents.mr_consolidation_agent import ( + _extract_cves_from_text, + _extract_jira_issues_from_description, + _mr_type_from_labels, +) from ymir.agents.tasks import ( _CONSOLIDATION_HASH_KEY as HASH_KEY, ) @@ -605,3 +609,104 @@ async def test_sweep_uses_activated_at_not_submitted_at(fake_redis): assert removed == 0 assert await fake_redis.hget(HASH_KEY, active_key) is not None + + +# -- _extract_cves_from_text --------------------------------------------------- + + +class TestExtractCves: + def test_single_cve(self): + assert _extract_cves_from_text("Fixed CVE-2024-12345") == ["CVE-2024-12345"] + + def test_multiple_cves(self): + text = "CVE: CVE-2024-1111, CVE-2024-2222\nAlso fixes CVE-2023-99999" + result = _extract_cves_from_text(text) + assert result == ["CVE-2023-99999", "CVE-2024-1111", "CVE-2024-2222"] + + def test_no_cves(self): + assert _extract_cves_from_text("No vulnerabilities here") == [] + + def test_deduplication(self): + text = "CVE-2024-1111 and CVE-2024-1111 again" + assert _extract_cves_from_text(text) == ["CVE-2024-1111"] + + def test_cve_in_commit_message_format(self): + text = "Rebuild gnutls\n\nCVE: CVE-2024-55555\nResolves: RHEL-12345" + assert _extract_cves_from_text(text) == ["CVE-2024-55555"] + + +# -- _mr_type_from_labels ------------------------------------------------------ + + +class TestMrTypeFromLabels: + def test_backport_label(self): + mr = {"labels": ["ymir_backport"]} + assert _mr_type_from_labels(mr) == "backport" + + def test_rebuild_label(self): + mr = {"labels": ["ymir_rebuild"]} + assert _mr_type_from_labels(mr) == "rebuild" + + def test_both_labels_rebuild_wins(self): + mr = {"labels": ["ymir_backport", "ymir_rebuild"]} + assert _mr_type_from_labels(mr) == "rebuild" + + def test_no_labels_defaults_to_backport(self): + mr = {"labels": []} + assert _mr_type_from_labels(mr) == "backport" + + def test_none_labels_defaults_to_backport(self): + mr = {} + assert _mr_type_from_labels(mr) == "backport" + + +# -- MR selection priority (backport+backport > backport+rebuild) -------------- + + +class TestMrSelectionPriority: + """Verify the selection logic prefers backport+backport over backport+rebuild.""" + + def _make_mr(self, branch, label): + return { + "source_branch": branch, + "labels": [label], + "url": f"https://gitlab.com/mr/{branch}", + "title": f"MR for {branch}", + "description": f"Resolves: RHEL-{branch[-3:]}", + } + + def test_two_backports_selected_over_rebuild(self): + mrs = [ + self._make_mr("bp1", "ymir_backport"), + self._make_mr("bp2", "ymir_backport"), + self._make_mr("rb1", "ymir_rebuild"), + ] + backport_mrs = [mr for mr in mrs if _mr_type_from_labels(mr) == "backport"] + rebuild_mrs = [mr for mr in mrs if _mr_type_from_labels(mr) == "rebuild"] + + if len(backport_mrs) >= 2: + selected = backport_mrs[:2] + elif backport_mrs and rebuild_mrs: + selected = [backport_mrs[0], rebuild_mrs[0]] + else: + selected = mrs[:2] + + assert all(_mr_type_from_labels(mr) == "backport" for mr in selected) + + def test_one_backport_one_rebuild_selected(self): + mrs = [ + self._make_mr("bp1", "ymir_backport"), + self._make_mr("rb1", "ymir_rebuild"), + ] + backport_mrs = [mr for mr in mrs if _mr_type_from_labels(mr) == "backport"] + rebuild_mrs = [mr for mr in mrs if _mr_type_from_labels(mr) == "rebuild"] + + if len(backport_mrs) >= 2: + selected = backport_mrs[:2] + elif backport_mrs and rebuild_mrs: + selected = [backport_mrs[0], rebuild_mrs[0]] + else: + selected = mrs[:2] + + assert _mr_type_from_labels(selected[0]) == "backport" + assert _mr_type_from_labels(selected[1]) == "rebuild" From 8e5f27f21b1338d040d7426d37af00d9b97c43e7 Mon Sep 17 00:00:00 2001 From: Tomas Korbar Date: Tue, 21 Jul 2026 15:30:19 +0200 Subject: [PATCH 2/3] Use sorting approach for prioritizing of MRs to consolidate --- ymir/agents/mr_consolidation_agent.py | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/ymir/agents/mr_consolidation_agent.py b/ymir/agents/mr_consolidation_agent.py index 688779df8..b908341d9 100644 --- a/ymir/agents/mr_consolidation_agent.py +++ b/ymir/agents/mr_consolidation_agent.py @@ -182,6 +182,8 @@ def _extract_cves_from_text(text: str) -> list[str]: _CONSOLIDATION_MR_LABELS = ["ymir_backport", "ymir_rebuild"] +_MR_TYPE_PRIORITY = {"backport": 0, "rebuild": 1} + def _mr_type_from_labels(mr: dict) -> str: """Determine if an MR is a backport or rebuild from its GitLab labels.""" @@ -521,28 +523,27 @@ async def fork_and_prepare_dist_git(state): ) return Workflow.END - # Selection priority: backport+backport first; backport+rebuild - # only when fewer than 2 backport MRs exist. - backport_mrs = [mr for mr in current_mrs if _mr_type_from_labels(mr) == "backport"] - rebuild_mrs = [mr for mr in current_mrs if _mr_type_from_labels(mr) == "rebuild"] + # Sort by type priority (backport first, then rebuild) + # and select the two highest-priority MRs. + sorted_mrs = sorted( + current_mrs, + key=lambda mr: _MR_TYPE_PRIORITY.get(_mr_type_from_labels(mr), 99), + ) - if len(backport_mrs) >= 2: - selected = backport_mrs[:2] - elif backport_mrs and rebuild_mrs: - selected = [backport_mrs[0], rebuild_mrs[0]] - elif not backport_mrs: + if _mr_type_from_labels(sorted_mrs[0]) != "backport": logger.info( - "No backport MRs on current HEAD for %s/%s, cannot consolidate rebuild-only", + "No backport MRs on current HEAD for %s/%s, cannot consolidate without a backport", package, dist_git_branch, ) state.consolidation_result = MRConsolidationOutputSchema( success=True, - status="No backport MR on current HEAD; rebuild-only consolidation not supported.", + status="No backport MR on current HEAD; " + "consolidation without a backport is not supported.", ) return Workflow.END - else: - selected = current_mrs[:2] + + selected = sorted_mrs[:2] state.mr_urls = [mr["url"] for mr in selected] state.mr_branches = [mr["source_branch"] for mr in selected] From 7579fcce450dc4f4bf3ad2dff741f0715b258406 Mon Sep 17 00:00:00 2001 From: Tomas Korbar Date: Tue, 21 Jul 2026 15:40:26 +0200 Subject: [PATCH 3/3] Refactor consolidation job submission into shared helper Extract the duplicated fetch-config/check-enabled/submit-to-Redis logic from both backport_agent and rebuild_agent into a single try_submit_consolidation_job() function in tasks.py. Each agent keeps its own thin wrapper for precondition checks and error handling (backport posts a Jira comment on malformed config, rebuild just logs). Co-authored-by: Cursor --- ymir/agents/backport_agent.py | 40 ++++++--------------------------- ymir/agents/rebuild_agent.py | 42 ++++++----------------------------- ymir/agents/tasks.py | 37 ++++++++++++++++++++++++++++++ 3 files changed, 51 insertions(+), 68 deletions(-) diff --git a/ymir/agents/backport_agent.py b/ymir/agents/backport_agent.py index c9b415d55..2314fa56c 100644 --- a/ymir/agents/backport_agent.py +++ b/ymir/agents/backport_agent.py @@ -30,7 +30,7 @@ from ymir.agents.observability import setup_observability from ymir.agents.package_update_steps import PackageUpdateState from ymir.agents.reasoning_agent import ReasoningAgent -from ymir.agents.tasks import InvalidConsolidationConfigError, fetch_consolidation_config, submit_merge_job +from ymir.agents.tasks import InvalidConsolidationConfigError from ymir.agents.utils import ( check_subprocess, format_mr_triage_details, @@ -729,7 +729,12 @@ async def submit_consolidation_job(state): return "comment_in_jira" try: - config = await fetch_consolidation_config(state.package, gateway_tools) + await tasks.try_submit_consolidation_job( + state.package, + state.dist_git_branch, + gateway_tools, + redis_conn, + ) except InvalidConsolidationConfigError as e: logger.warning("Invalid consolidation config for %s: %s", state.package, e) await tasks.comment_in_jira( @@ -744,37 +749,6 @@ async def submit_consolidation_job(state): available_tools=gateway_tools, user_triggered=user_triggered, ) - return "comment_in_jira" - - try: - if not config.merge_mrs: - logger.info( - "MR consolidation not enabled for %s, skipping", - state.package, - ) - return "comment_in_jira" - - if redis_conn is not None: - submitted = await submit_merge_job( - redis_conn, - state.package, - state.dist_git_branch, - release_strategy=config.release_strategy.value, - ) - if submitted: - logger.info( - "Submitted consolidation job for %s/%s", - state.package, - state.dist_git_branch, - ) - else: - logger.info( - "Consolidation job already queued for %s/%s", - state.package, - state.dist_git_branch, - ) - else: - logger.info("No Redis connection (direct mode), skipping consolidation job submission") except Exception as e: logger.warning("Failed to submit consolidation job: %s", e) diff --git a/ymir/agents/rebuild_agent.py b/ymir/agents/rebuild_agent.py index 642e00c41..cc200e406 100644 --- a/ymir/agents/rebuild_agent.py +++ b/ymir/agents/rebuild_agent.py @@ -15,7 +15,7 @@ from ymir.agents.log_agent import get_prompt as get_log_prompt from ymir.agents.observability import setup_observability from ymir.agents.package_update_steps import PackageUpdateState -from ymir.agents.tasks import InvalidConsolidationConfigError, fetch_consolidation_config, submit_merge_job +from ymir.agents.tasks import InvalidConsolidationConfigError from ymir.agents.utils import ( format_mr_triage_details, get_agent_execution_config, @@ -278,42 +278,14 @@ async def submit_consolidation_job(state): return "comment_in_jira" try: - config = await fetch_consolidation_config(state.package, gateway_tools) + await tasks.try_submit_consolidation_job( + state.package, + state.dist_git_branch, + gateway_tools, + redis_conn, + ) except InvalidConsolidationConfigError as e: logger.warning("Invalid consolidation config for %s: %s", state.package, e) - return "comment_in_jira" - - try: - if not config.merge_mrs: - logger.info( - "MR consolidation not enabled for %s, skipping", - state.package, - ) - return "comment_in_jira" - - if redis_conn is not None: - submitted = await submit_merge_job( - redis_conn, - state.package, - state.dist_git_branch, - release_strategy=config.release_strategy.value, - ) - if submitted: - logger.info( - "Submitted consolidation job for %s/%s", - state.package, - state.dist_git_branch, - ) - else: - logger.info( - "Consolidation job already queued for %s/%s", - state.package, - state.dist_git_branch, - ) - else: - logger.info( - "No Redis connection (direct mode), skipping consolidation job submission" - ) except Exception as e: logger.warning("Failed to submit consolidation job: %s", e) diff --git a/ymir/agents/tasks.py b/ymir/agents/tasks.py index 241d23877..54a92af93 100644 --- a/ymir/agents/tasks.py +++ b/ymir/agents/tasks.py @@ -766,3 +766,40 @@ async def fetch_consolidation_config( raise InvalidConsolidationConfigError( f"ymir.yaml consolidation section for {package} is malformed: {e}" ) from e + + +async def try_submit_consolidation_job( + package: str, + dist_git_branch: str, + gateway_tools: list, + redis_conn, +) -> None: + """Fetch consolidation config and submit a job if enabled. + + Shared logic used by both the backport and rebuild agents after + creating an MR. + + Raises: + InvalidConsolidationConfigError: When ymir.yaml exists but the + consolidation section is malformed. + """ + config = await fetch_consolidation_config(package, gateway_tools) + + if not config.merge_mrs: + logger.info("MR consolidation not enabled for %s, skipping", package) + return + + if redis_conn is None: + logger.info("No Redis connection (direct mode), skipping consolidation job submission") + return + + submitted = await submit_merge_job( + redis_conn, + package, + dist_git_branch, + release_strategy=config.release_strategy.value, + ) + if submitted: + logger.info("Submitted consolidation job for %s/%s", package, dist_git_branch) + else: + logger.info("Consolidation job already queued for %s/%s", package, dist_git_branch)