From 70c94df4c772c5e6133a700f42a9eea871fed248 Mon Sep 17 00:00:00 2001 From: Jiri Podivin Date: Wed, 12 Aug 2026 12:41:30 +0200 Subject: [PATCH 1/2] Add type hints for state Signed-off-by: Jiri Podivin --- ymir/agents/mr_consolidation_agent.py | 34 +++++++++++++-------------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/ymir/agents/mr_consolidation_agent.py b/ymir/agents/mr_consolidation_agent.py index ac43d7690..e9c33e11e 100644 --- a/ymir/agents/mr_consolidation_agent.py +++ b/ymir/agents/mr_consolidation_agent.py @@ -299,7 +299,7 @@ def _mr_type_from_labels(mr: dict) -> str: async def _resolve_source_issues( - state, + state: ConsolidationState, project_path: str, issue_keys: list[str], gateway_tools: list, @@ -444,7 +444,7 @@ async def run_workflow( workflow = Workflow(ConsolidationState, name="MRConsolidationWorkflow") - async def list_open_mrs(state): + async def list_open_mrs(state: ConsolidationState): """List open backport and rebuild MRs for the package/branch.""" if state.mr_branches: logger.info( @@ -503,7 +503,7 @@ async def list_open_mrs(state): state.all_open_mrs = all_mrs return "fork_and_prepare_dist_git" - async def fork_and_prepare_dist_git(state): + async def fork_and_prepare_dist_git(state: ConsolidationState): working_id = f"consolidation-{package}-{dist_git_branch}-{int(time.time())}" ( state.local_clone, @@ -723,7 +723,7 @@ async def fork_and_prepare_dist_git(state): return "per_commit_flow" if release_strategy == "per_commit" else "run_consolidation_agent" - async def run_consolidation_agent(state): + async def run_consolidation_agent(state: ConsolidationState): prompt = render_template( "mr_consolidation/prompt.j2", MRConsolidationInputSchema( @@ -766,7 +766,7 @@ async def run_consolidation_agent(state): return "handle_failure" return "run_build_agent" - async def run_build_agent(state): + async def run_build_agent(state: ConsolidationState): if not state.consolidation_result or not state.consolidation_result.srpm_path: logger.warning("No SRPM generated, skipping build verification") return "stage_changes" @@ -781,7 +781,7 @@ async def run_build_agent(state): ), ) - def _retry_step_for_build(state): + def _retry_step_for_build(state: ConsolidationState): """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: @@ -860,7 +860,7 @@ async def _run_prep(clone) -> tuple[bool, str]: return False, output return True, output - async def per_commit_flow(state): + async def per_commit_flow(state: ConsolidationState): """Cherry-pick base branch, then incrementally adapt commits from the other branch. 1. Choose the branch with larger patches as the "base" — cherry-pick @@ -1134,7 +1134,7 @@ async def per_commit_flow(state): return "run_build_agent" - async def rebuild_append_flow(state): + async def rebuild_append_flow(state: ConsolidationState): """Append rebuild ticket(s) to the backport MR without cherry-picking. The backport branch is cherry-picked as base (it has patches + Release @@ -1320,7 +1320,7 @@ async def rebuild_append_flow(state): return "run_build_agent" - async def update_release(state): + async def update_release(state: ConsolidationState): try: await tasks.update_release( local_clone=state.local_clone, @@ -1338,7 +1338,7 @@ async def update_release(state): return "handle_failure" return "stage_changes" - async def stage_changes(state): + async def stage_changes(state: ConsolidationState): try: files_to_stage = _files_to_stage_for_patches(state.local_clone, package) logger.info("Staging files: %s", files_to_stage) @@ -1358,7 +1358,7 @@ async def stage_changes(state): return "commit_push_and_open_mr" return "run_log_agent" - async def run_log_agent(state): + async def run_log_agent(state: ConsolidationState): summary_parts = [ f"Consolidated {len(state.mr_branches)} backport branches " f"for {package} on {dist_git_branch}.", @@ -1393,7 +1393,7 @@ async def run_log_agent(state): ) return "stage_changes" - async def commit_push_and_open_mr(state): + async def commit_push_and_open_mr(state: ConsolidationState): """Squash all changes into a single commit (merged strategy), then push.""" if state.log_result: commit_message = f"{state.log_result.title}\n\n{state.log_result.description}" @@ -1455,7 +1455,7 @@ async def commit_push_and_open_mr(state): return "push_and_open_mr" - async def push_and_open_mr(state): + async def push_and_open_mr(state: ConsolidationState): """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( @@ -1539,7 +1539,7 @@ async def push_and_open_mr(state): return "mark_original_mrs" - async def mark_original_mrs(state): + async def mark_original_mrs(state: ConsolidationState): """Label original MRs as consolidated so they are excluded from future runs.""" if dry_run: logger.info( @@ -1563,7 +1563,7 @@ async def mark_original_mrs(state): return "update_jira_issues" - async def update_jira_issues(state): + async def update_jira_issues(state: ConsolidationState): if not state.jira_issues_collected or not state.merge_request_url: return "requeue_if_needed" @@ -1603,7 +1603,7 @@ async def update_jira_issues(state): return "requeue_if_needed" - async def requeue_if_needed(state): + async def requeue_if_needed(state: ConsolidationState): current_count = state.current_mrs_count remaining = current_count - 2 if remaining < 1: @@ -1625,7 +1625,7 @@ async def requeue_if_needed(state): ) return Workflow.END - async def handle_failure(state): + async def handle_failure(state: ConsolidationState): logger.error( "MR consolidation failed for %s/%s: %s", package, From 1d27cefa419534e393175f8767b8de0eacc32bcb Mon Sep 17 00:00:00 2001 From: Jiri Podivin Date: Wed, 12 Aug 2026 17:21:28 +0200 Subject: [PATCH 2/2] Report all significant actions taken while processing consolidation workflow The `MRConsolidationOutputSchema` now includes fields `status` and `status_detail`. Valid values of the `status` are limited to `ConsolidationStatus` literal. The `handle_failure` step now routes to `update_jira_issues` step, only logging error and optionally filling `jira_issues_collected` if it is not set. The `update_jira_issues` returns `Workflow.END` if the status field of `MRConsolidationOutputSchema` is set to "failed", "nothing_to_consolidate" or "error". Signed-off-by: Jiri Podivin Assisted-by: Claude Sonnet 4.5 via Claude Code --- ymir/agents/backport_agent.py | 14 +- ymir/agents/mr_consolidation_agent.py | 220 +++++++++++++++++++++----- ymir/agents/rebuild_agent.py | 1 + ymir/agents/tasks.py | 31 +++- ymir/agents/tests/unit/test_tasks.py | 170 +++++++++++++++++++- ymir/common/models.py | 14 +- ymir/common/tests/unit/test_models.py | 71 +++++++++ 7 files changed, 476 insertions(+), 45 deletions(-) diff --git a/ymir/agents/backport_agent.py b/ymir/agents/backport_agent.py index 9f3500250..335118c0f 100644 --- a/ymir/agents/backport_agent.py +++ b/ymir/agents/backport_agent.py @@ -744,6 +744,7 @@ async def submit_consolidation_job(state): state.dist_git_branch, gateway_tools, redis_conn, + jira_issue=state.jira_issue, ) except InvalidConsolidationConfigError as e: logger.warning("Invalid consolidation config for %s: %s", state.package, e) @@ -768,9 +769,16 @@ async def comment_in_jira(state): if dry_run: return Workflow.END if state.backport_result.success: - comment_text = ( - state.merge_request_url if state.merge_request_url else state.backport_result.status - ) + if state.merge_request_url: + if not state.merge_request_newly_created: + comment_text = ( + f"MR: {state.merge_request_url}\n\n" + "The existing MR was reused; MR consolidation will not run for this issue." + ) + else: + comment_text = f"New merge request was created: {state.merge_request_url}" + else: + comment_text = state.backport_result.status is_error = False else: comment_text = f"Agent failed to perform a backport: {state.backport_result.error}" diff --git a/ymir/agents/mr_consolidation_agent.py b/ymir/agents/mr_consolidation_agent.py index e9c33e11e..7ecf159d5 100644 --- a/ymir/agents/mr_consolidation_agent.py +++ b/ymir/agents/mr_consolidation_agent.py @@ -188,6 +188,19 @@ def _extract_cves_from_cve_footer_lines(text: str) -> list[str]: return sorted(set(cves)) +def _extract_jira_from_mr_descriptions(mrs: list[dict]) -> list[str]: + """Extract Jira issue keys from MR description Resolves:/Related: footers. + + Used in auto mode to seed jira_issues_collected before the git clone is + available, so early-exit paths can still post Jira comments. + """ + issues: list[str] = [] + for mr in mrs: + desc = mr.get("description") or "" + issues.extend(_extract_jira_issues_from_resolves_footer_lines(desc)) + return sorted(set(issues)) + + def _extract_jira_issues_from_resolves_footer_lines(text: str) -> list[str]: """Extract RHEL keys only from ``Resolves:`` / ``Related:`` lines. @@ -349,10 +362,14 @@ async def _resolve_source_issues( ) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status=f"Could not find an open MR for {issue_key}", + status="mr_not_found", + status_detail=f"Could not find an open MR for {issue_key}", error=f"No open MR matching {issue_key} in {project_path}", ) - return Workflow.END + # When have no collected issues post updates under those from issue_keys + if not state.jira_issues_collected: + state.jira_issues_collected = issue_keys + return "update_jira_issues" matched_mrs.append(mr) logger.info( @@ -375,9 +392,13 @@ async def _resolve_source_issues( 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.", + status="nothing_to_consolidate", + status_detail="Fewer than 2 unique MRs resolved; nothing to do.", ) - return Workflow.END + # When have no collected issues post updates under those from issue_keys + if not state.jira_issues_collected: + state.jira_issues_collected = issue_keys + return "update_jira_issues" state.all_open_mrs = matched_mrs state.mr_urls = [mr["url"] for mr in matched_mrs] @@ -496,9 +517,12 @@ async def list_open_mrs(state: ConsolidationState): ) state.consolidation_result = MRConsolidationOutputSchema( success=True, - status="Fewer than 2 MRs to consolidate; nothing to do.", + status="nothing_to_consolidate", + status_detail="Fewer than 2 MRs to consolidate; nothing to do.", ) - return Workflow.END + if not state.jira_issues_collected and all_mrs: + state.jira_issues_collected = _extract_jira_from_mr_descriptions(all_mrs) + return "update_jira_issues" state.all_open_mrs = all_mrs return "fork_and_prepare_dist_git" @@ -622,9 +646,12 @@ async def fork_and_prepare_dist_git(state: ConsolidationState): ) state.consolidation_result = MRConsolidationOutputSchema( success=True, - status="Fewer than 2 MRs based on current HEAD; nothing to do.", + status="nothing_to_consolidate", + status_detail="Fewer than 2 MRs based on current HEAD; nothing to do.", ) - return Workflow.END + if not state.jira_issues_collected and state.all_open_mrs: + state.jira_issues_collected = _extract_jira_from_mr_descriptions(state.all_open_mrs) + return "update_jira_issues" # Sort by type priority (backport first, then rebuild) # and select the two highest-priority MRs. @@ -641,10 +668,13 @@ async def fork_and_prepare_dist_git(state: ConsolidationState): ) state.consolidation_result = MRConsolidationOutputSchema( success=True, - status="No backport MR on current HEAD; " + status="nothing_to_consolidate", + status_detail="No backport MR on current HEAD; " "consolidation without a backport is not supported.", ) - return Workflow.END + if not state.jira_issues_collected and state.all_open_mrs: + state.jira_issues_collected = _extract_jira_from_mr_descriptions(state.all_open_mrs) + return "update_jira_issues" selected = sorted_mrs[:2] @@ -679,7 +709,8 @@ async def fork_and_prepare_dist_git(state: ConsolidationState): ) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Failed to diff branch", + status="failed", + status_detail="Failed to diff branch", error=f"git diff {dist_git_branch}...{branch_name} " f"failed (exit {exit_code}): {err_msg}", ) @@ -702,7 +733,8 @@ async def fork_and_prepare_dist_git(state: ConsolidationState): logger.error("Failed to collect commit footers: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Failed to collect commit footers", + status="failed", + status_detail="Failed to collect commit footers", error=str(e), ) return "handle_failure" @@ -751,14 +783,16 @@ async def run_consolidation_agent(state: ConsolidationState): logger.error("Consolidation agent error: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Agent error", + status="failed", + status_detail="Agent error", error=str(e), ) except Exception as e: logger.error("Unexpected consolidation error: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Unexpected error", + status="failed", + status_detail="Unexpected error", error=str(e), ) @@ -1113,10 +1147,14 @@ async def per_commit_flow(state: ConsolidationState): ), ) output = srpm_result.result - srpm_path = output.strip() if "FAILED" not in output else None + output_stripped = output.strip() + srpm_path = output_stripped if output_stripped and "FAILED" not in output else None state.consolidation_result = MRConsolidationOutputSchema( success=srpm_path is not None, - status="per_commit consolidation complete", + status="consolidation_complete" if srpm_path else "failed", + status_detail="per_commit consolidation complete" + if srpm_path + else "per_commit flow failed", srpm_path=srpm_path, ) if not srpm_path: @@ -1127,7 +1165,8 @@ async def per_commit_flow(state: ConsolidationState): logger.error("per_commit flow error: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="per_commit flow failed", + status="failed", + status_detail="per_commit flow failed", error=str(e), ) return "handle_failure" @@ -1299,10 +1338,14 @@ async def rebuild_append_flow(state: ConsolidationState): ), ) output = srpm_result.result - srpm_path = output.strip() if "FAILED" not in output else None + output_stripped = output.strip() + srpm_path = output_stripped if output_stripped and "FAILED" not in output else None state.consolidation_result = MRConsolidationOutputSchema( success=srpm_path is not None, - status="rebuild_append consolidation complete", + status="consolidation_complete" if srpm_path else "failed", + status_detail="rebuild_append consolidation complete" + if srpm_path + else "rebuild_append flow failed", srpm_path=srpm_path, ) if not srpm_path: @@ -1313,7 +1356,8 @@ async def rebuild_append_flow(state: ConsolidationState): logger.error("rebuild_append flow error: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="rebuild_append flow failed", + status="failed", + status_detail="rebuild_append flow failed", error=str(e), ) return "handle_failure" @@ -1332,7 +1376,8 @@ async def update_release(state: ConsolidationState): logger.error("Error updating release: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Failed to update release", + status="failed", + status_detail="Failed to update release", error=str(e), ) return "handle_failure" @@ -1350,7 +1395,8 @@ async def stage_changes(state: ConsolidationState): logger.error("Error staging changes: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Failed to stage changes", + status="failed", + status_detail="Failed to stage changes", error=str(e), ) return "handle_failure" @@ -1365,8 +1411,8 @@ async def run_log_agent(state: ConsolidationState): ] if state.mr_titles: summary_parts.extend(f" - {title}" for title in state.mr_titles) - if state.consolidation_result and state.consolidation_result.status: - summary_parts.append(f"Result: {state.consolidation_result.status}") + if state.consolidation_result and state.consolidation_result.status_detail: + summary_parts.append(f"Result: {state.consolidation_result.status_detail}") changes_summary = "\n".join(summary_parts) log_prompt = render_template( @@ -1448,10 +1494,11 @@ async def commit_push_and_open_mr(state: ConsolidationState): logger.error("Failed to finalize commit: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Failed to create consolidated commit", + status="failed", + status_detail="Failed to create consolidated commit", error=str(e), ) - return Workflow.END + return "handle_failure" return "push_and_open_mr" @@ -1532,10 +1579,11 @@ async def push_and_open_mr(state: ConsolidationState): logger.error("Failed to create consolidated MR: %s", e) state.consolidation_result = MRConsolidationOutputSchema( success=False, - status="Failed to create consolidated MR", + status="failed", + status_detail="Failed to create consolidated MR", error=str(e), ) - return Workflow.END + return "handle_failure" return "mark_original_mrs" @@ -1564,25 +1612,79 @@ async def mark_original_mrs(state: ConsolidationState): return "update_jira_issues" async def update_jira_issues(state: ConsolidationState): - if not state.jira_issues_collected or not state.merge_request_url: + """Post status updates to Jira issues based on consolidation outcome.""" + if not state.jira_issues_collected: + return "requeue_if_needed" + + # Determine if we should post a comment based on the status + if not state.consolidation_result: + return "requeue_if_needed" + + status = state.consolidation_result.status + + # Skip comment posting only when there's genuinely nothing to report + # (i.e., no failure, no success, and no informational status) + if not state.merge_request_url and status == "consolidation_complete": + logger.warning( + "Consolidation marked complete for %s/%s but no MR URL available - possible workflow bug", + package, + dist_git_branch, + ) return "requeue_if_needed" has_rebuild = any(t == "rebuild" for t in state.mr_types.values()) for issue_key in state.jira_issues_collected: - if has_rebuild: + comment = None + + if status == "failed": + # Report failure details from consolidation_result + error_detail = state.consolidation_result.error or "unknown" + comment = f"MR consolidation failed for {package}/{dist_git_branch}: {error_detail}" + elif status == "consolidation_complete": + # Successfully consolidated and created an MR + 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}" + ) + elif status == "nothing_to_consolidate": + # Informational: not enough MRs to consolidate + detail = ( + state.consolidation_result.status_detail or "Nothing to consolidate at this time." + ) comment = ( - f"Your MR has been consolidated (backport + rebuild) " - f"into a single MR: {state.merge_request_url}" + f"MR consolidation check completed for {package}/{dist_git_branch}.\n\n" + f"{detail}\n\n" + f"Your MR will be evaluated again when more MRs are available for consolidation." ) - else: + elif status == "mr_not_found": + # Could not find MR for the specified issue + detail = state.consolidation_result.status_detail or "Could not find MR for this issue." comment = ( - f"Your backport MR has been consolidated with other fixes " - f"into a single MR: {state.merge_request_url}" + f"MR consolidation could not proceed for {package}/{dist_git_branch}.\n\n{detail}" ) + + if not comment: + logger.warning( + ( + "No comment generated for issue %s with status '%s' " + "- unhandled ConsolidationStatus value" + ), + issue_key, + status, + ) + continue + if dry_run: logger.info( - "Dry run: would post consolidation comment on %s", + "Dry run: would post consolidation comment on %s: %s", issue_key, + comment, ) continue try: @@ -1601,6 +1703,10 @@ async def update_jira_issues(state: ConsolidationState): e, ) + if status in ("failed", "nothing_to_consolidate", "mr_not_found"): + # End the workflow if we have experienced failure, there is nothing to do or MR was not found + return Workflow.END + return "requeue_if_needed" async def requeue_if_needed(state: ConsolidationState): @@ -1626,13 +1732,18 @@ async def requeue_if_needed(state: ConsolidationState): return Workflow.END async def handle_failure(state: ConsolidationState): + """Log failure and request update of all linked Jira items.""" logger.error( "MR consolidation failed for %s/%s: %s", package, dist_git_branch, state.consolidation_result.error if state.consolidation_result else "unknown", ) - return Workflow.END + # In auto mode, jira_issues_collected may not be populated yet + # (failure before footer collection). Fall back to MR descriptions. + if not state.jira_issues_collected and state.all_open_mrs: + state.jira_issues_collected = _extract_jira_from_mr_descriptions(state.all_open_mrs) + return "update_jira_issues" workflow.add_step("list_open_mrs", list_open_mrs) workflow.add_step("fork_and_prepare_dist_git", fork_and_prepare_dist_git) @@ -1664,8 +1775,39 @@ async def handle_failure(state: ConsolidationState): # CVE / Jira lists are collected from commit footers after branches # are fetched — do not seed from branch metadata. - response = await workflow.run(initial_state) - return response.state + try: + response = await workflow.run(initial_state) + return response.state + except Exception: + logger.exception( + "Unhandled error in consolidation workflow for %s/%s", + package, + dist_git_branch, + ) + # Notify all known Jira issues — prefer keys collected from commit + # footers/MR descriptions; fall back to the source_issues seed. + issue_keys_to_notify = initial_state.jira_issues_collected or list(source_issues or []) + if issue_keys_to_notify and not dry_run: + msg = ( + f"MR consolidation failed for {package}/{dist_git_branch} " + f"with an unexpected error. Please check the agent logs." + ) + for issue_key in issue_keys_to_notify: + try: + await run_tool( + "add_jira_comment", + issue_key=issue_key, + comment=msg, + private=True, + available_tools=gateway_tools, + ) + except Exception as e: + logger.warning( + "Failed to post unhandled-failure comment on %s: %s", + issue_key, + e, + ) + raise _CONSOLIDATED_MARKER = "## Consolidated Backport MR" diff --git a/ymir/agents/rebuild_agent.py b/ymir/agents/rebuild_agent.py index c0a72f64d..1e0337114 100644 --- a/ymir/agents/rebuild_agent.py +++ b/ymir/agents/rebuild_agent.py @@ -294,6 +294,7 @@ async def submit_consolidation_job(state): state.dist_git_branch, gateway_tools, redis_conn, + jira_issue=state.jira_issue, ) except InvalidConsolidationConfigError as e: logger.warning("Invalid consolidation config for %s: %s", state.package, e) diff --git a/ymir/agents/tasks.py b/ymir/agents/tasks.py index 6591a16ca..16f81d336 100644 --- a/ymir/agents/tasks.py +++ b/ymir/agents/tasks.py @@ -937,11 +937,19 @@ async def try_submit_consolidation_job( dist_git_branch: str, gateway_tools: list, redis_conn, + jira_issue: str | None = None, ) -> None: """Fetch consolidation config and submit a job if enabled. Shared logic used by both the backport and rebuild agents after - creating an MR. + creating an MR. Posts a Jira comment when consolidation is triggered. + + Args: + package: The package name + dist_git_branch: The dist-git branch + gateway_tools: List of available MCP tools + redis_conn: Redis connection for job submission + jira_issue: Optional Jira issue key to post consolidation notification Raises: InvalidConsolidationConfigError: When ymir.yaml exists but the @@ -965,5 +973,26 @@ async def try_submit_consolidation_job( ) if submitted: logger.info("Submitted consolidation job for %s/%s", package, dist_git_branch) + + # Post a Jira comment notifying that consolidation has been triggered + if jira_issue: + try: + await run_tool( + "add_jira_comment", + issue_key=jira_issue, + comment=JIRA_COMMENT_TEMPLATE.substitute( + AGENT_TYPE="MR Consolidation", + JIRA_COMMENT=( + f"Your MR has been queued for consolidation with other open MRs " + f"for {package} on {dist_git_branch}. " + f"A consolidated MR will be created automatically once processing completes." + ), + ), + private=True, + available_tools=gateway_tools, + ) + logger.info("Posted consolidation notification to %s", jira_issue) + except Exception as e: + logger.warning("Failed to post consolidation notification to %s: %s", jira_issue, e) else: logger.info("Consolidation job already queued for %s/%s", package, dist_git_branch) diff --git a/ymir/agents/tests/unit/test_tasks.py b/ymir/agents/tests/unit/test_tasks.py index a825f0561..9c9e58d77 100644 --- a/ymir/agents/tests/unit/test_tasks.py +++ b/ymir/agents/tests/unit/test_tasks.py @@ -13,9 +13,10 @@ handle_zstream_branch_stale_error, needs_zstream_target_label, post_user_ack_once, + try_submit_consolidation_job, ) from ymir.common.constants import JiraLabels, RedisQueues -from ymir.common.models import Task +from ymir.common.models import PackageConsolidationConfig, Task @asynccontextmanager @@ -681,3 +682,170 @@ async def test_handle_zstream_branch_stale_error_skips_comment_on_dry_run(): mock_labels.assert_awaited_once() mock_comment.assert_not_awaited() redis.lpush.assert_awaited_once() + + +# -- try_submit_consolidation_job --------------------------------------------- + + +def _consolidation_config(merge_mrs: bool = True) -> PackageConsolidationConfig: + return PackageConsolidationConfig(merge_mrs=merge_mrs) + + +@pytest.mark.asyncio +async def test_try_submit_posts_jira_comment_when_submitted_with_issue(): + """When the job is newly submitted and jira_issue is given, a Jira comment is posted.""" + with ( + patch( + "ymir.agents.tasks.fetch_consolidation_config", + new_callable=AsyncMock, + return_value=_consolidation_config(), + ), + patch( + "ymir.agents.tasks.submit_merge_job", + new_callable=AsyncMock, + return_value=True, + ), + patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run_tool, + ): + await try_submit_consolidation_job( + package="bash", + dist_git_branch="c10s", + gateway_tools=[], + redis_conn=AsyncMock(), + jira_issue="RHEL-12345", + ) + + mock_run_tool.assert_awaited_once() + assert mock_run_tool.call_args.args[0] == "add_jira_comment" + assert mock_run_tool.call_args.kwargs["issue_key"] == "RHEL-12345" + + +@pytest.mark.asyncio +async def test_try_submit_no_jira_comment_when_issue_is_none(): + """When jira_issue=None no Jira comment is posted, even if the job was submitted.""" + with ( + patch( + "ymir.agents.tasks.fetch_consolidation_config", + new_callable=AsyncMock, + return_value=_consolidation_config(), + ), + patch( + "ymir.agents.tasks.submit_merge_job", + new_callable=AsyncMock, + return_value=True, + ), + patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run_tool, + ): + await try_submit_consolidation_job( + package="bash", + dist_git_branch="c10s", + gateway_tools=[], + redis_conn=AsyncMock(), + jira_issue=None, + ) + + mock_run_tool.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_try_submit_no_jira_comment_when_already_queued(): + """When the job is already queued (submit_merge_job returns False) no comment is posted.""" + with ( + patch( + "ymir.agents.tasks.fetch_consolidation_config", + new_callable=AsyncMock, + return_value=_consolidation_config(), + ), + patch( + "ymir.agents.tasks.submit_merge_job", + new_callable=AsyncMock, + return_value=False, + ), + patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run_tool, + ): + await try_submit_consolidation_job( + package="bash", + dist_git_branch="c10s", + gateway_tools=[], + redis_conn=AsyncMock(), + jira_issue="RHEL-12345", + ) + + mock_run_tool.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_try_submit_jira_comment_failure_is_swallowed(): + """A failure posting the Jira comment must not propagate — it is logged and ignored.""" + with ( + patch( + "ymir.agents.tasks.fetch_consolidation_config", + new_callable=AsyncMock, + return_value=_consolidation_config(), + ), + patch( + "ymir.agents.tasks.submit_merge_job", + new_callable=AsyncMock, + return_value=True, + ), + patch( + "ymir.agents.tasks.run_tool", + new_callable=AsyncMock, + side_effect=RuntimeError("Jira down"), + ), + ): + await try_submit_consolidation_job( + package="bash", + dist_git_branch="c10s", + gateway_tools=[], + redis_conn=AsyncMock(), + jira_issue="RHEL-12345", + ) + + +@pytest.mark.asyncio +async def test_try_submit_skips_when_consolidation_disabled(): + """When merge_mrs=False no job is submitted and no Jira comment is posted.""" + with ( + patch( + "ymir.agents.tasks.fetch_consolidation_config", + new_callable=AsyncMock, + return_value=_consolidation_config(merge_mrs=False), + ), + patch("ymir.agents.tasks.submit_merge_job", new_callable=AsyncMock) as mock_submit, + patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run_tool, + ): + await try_submit_consolidation_job( + package="bash", + dist_git_branch="c10s", + gateway_tools=[], + redis_conn=AsyncMock(), + jira_issue="RHEL-12345", + ) + + mock_submit.assert_not_awaited() + mock_run_tool.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_try_submit_skips_when_redis_is_none(): + """Without a Redis connection (direct mode) no job is submitted and no comment is posted.""" + with ( + patch( + "ymir.agents.tasks.fetch_consolidation_config", + new_callable=AsyncMock, + return_value=_consolidation_config(), + ), + patch("ymir.agents.tasks.submit_merge_job", new_callable=AsyncMock) as mock_submit, + patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run_tool, + ): + await try_submit_consolidation_job( + package="bash", + dist_git_branch="c10s", + gateway_tools=[], + redis_conn=None, + jira_issue="RHEL-12345", + ) + + mock_submit.assert_not_awaited() + mock_run_tool.assert_not_awaited() diff --git a/ymir/common/models.py b/ymir/common/models.py index 7e6569723..4990ad319 100644 --- a/ymir/common/models.py +++ b/ymir/common/models.py @@ -817,11 +817,23 @@ class MRConsolidationInputSchema(BaseModel): build_error: str | None = Field(default=None, description="Error encountered during package build") +ConsolidationStatus = Literal[ + "nothing_to_consolidate", # Fewer than 2 MRs, or no backport MR found + "mr_not_found", # Could not find MR for specified issue + "consolidation_complete", # Successfully consolidated and created MR + "failed", # Consolidation failed (see error field for details) +] + + class MRConsolidationOutputSchema(BaseModel): """Output schema for the MR consolidation agent.""" success: bool = Field(description="Whether the consolidation was successfully completed") - status: str = Field(description="Consolidation status with details of how the merge was performed") + status: ConsolidationStatus = Field(description="Consolidation status indicating the outcome type") + status_detail: str | None = Field( + default=None, + description="Human-readable details about the status", + ) srpm_path: Path | None = Field(default=None, description="Absolute path to generated SRPM") error: str | None = Field(default=None, description="Specific details about an error") files_to_git_add: list[str] | None = Field( diff --git a/ymir/common/tests/unit/test_models.py b/ymir/common/tests/unit/test_models.py index e65cf7883..2cf698973 100644 --- a/ymir/common/tests/unit/test_models.py +++ b/ymir/common/tests/unit/test_models.py @@ -1,3 +1,6 @@ +import pytest +from pydantic import ValidationError + from ymir.common.models import ( AUTOMATED_RESOLUTION_NOT_SUPPORTED, TRIAGE_DISCLAIMER, @@ -6,6 +9,7 @@ ClarificationNeededData, ConsolidatedIssue, ErrorData, + MRConsolidationOutputSchema, NotAffectedData, OpenEndedAnalysisData, PostponedData, @@ -639,3 +643,70 @@ def test_reproducer_output_retryable_error(): assert data.test_already_exists is False restored = ReproducerOutputSchema.model_validate_json(data.model_dump_json()) assert restored.retryable_error is True + + +# --- MRConsolidationOutputSchema tests --- + + +class TestMRConsolidationOutputSchema: + @pytest.mark.parametrize( + "status", + [ + "nothing_to_consolidate", + "mr_not_found", + "consolidation_complete", + "failed", + ], + ) + def test_valid_status_values_accepted(self, status): + schema = MRConsolidationOutputSchema(success=True, status=status) + assert schema.status == status + + def test_invalid_status_raises_validation_error(self): + with pytest.raises(ValidationError): + MRConsolidationOutputSchema(success=True, status="in_progress") + + def test_status_detail_defaults_to_none(self): + schema = MRConsolidationOutputSchema(success=True, status="consolidation_complete") + assert schema.status_detail is None + + def test_status_detail_is_set(self): + schema = MRConsolidationOutputSchema( + success=False, + status="failed", + status_detail="Agent error during consolidation", + ) + assert schema.status_detail == "Agent error during consolidation" + + def test_failed_status_with_error_field(self): + schema = MRConsolidationOutputSchema( + success=False, + status="failed", + status_detail="Unexpected error", + error="Traceback (most recent call last): ...", + ) + assert schema.success is False + assert schema.status == "failed" + assert schema.status_detail == "Unexpected error" + assert schema.error is not None + + def test_nothing_to_consolidate_is_successful(self): + """A nothing_to_consolidate result is a success (not a failure).""" + schema = MRConsolidationOutputSchema( + success=True, + status="nothing_to_consolidate", + status_detail="Fewer than 2 unique MRs resolved; nothing to do.", + ) + assert schema.success is True + assert schema.error is None + + def test_mr_not_found_with_error(self): + schema = MRConsolidationOutputSchema( + success=False, + status="mr_not_found", + status_detail="Could not find an open MR for RHEL-99999", + error="No open MR matching RHEL-99999 in rpms/bash", + ) + assert schema.status == "mr_not_found" + assert "RHEL-99999" in schema.status_detail + assert "RHEL-99999" in schema.error