From ff65c1e158fd40e6fe62dfa13a498aeb7e1a91f6 Mon Sep 17 00:00:00 2001 From: Tomas Korbar Date: Fri, 21 Aug 2026 09:22:01 +0200 Subject: [PATCH 1/5] Reproducer: acquire create/adapt lock at workflow start Serialize sibling-stream jobs before analysis and MR push, park contended tasks on per-lock blocked queues, and promote them when the lock releases instead of using a fixed 30-minute delayed retry. Co-authored-by: Cursor --- ymir/agents/reproducer_agent.py | 206 +++++++++++------- .../tests/unit/test_reproducer_agent.py | 61 +++++- ymir/common/models.py | 4 +- ymir/common/reproducer_lock.py | 72 ++++++ .../common/tests/unit/test_reproducer_lock.py | 28 +++ 5 files changed, 289 insertions(+), 82 deletions(-) diff --git a/ymir/agents/reproducer_agent.py b/ymir/agents/reproducer_agent.py index ddb4202ab..ab43158a3 100644 --- a/ymir/agents/reproducer_agent.py +++ b/ymir/agents/reproducer_agent.py @@ -54,6 +54,7 @@ ReproducerOutputSchema as OutputSchema, ) from ymir.common.reproducer_lock import ( + enqueue_blocked_reproducer_task, release_reproducer_lock, resolve_reproducer_lock_id, sweep_stale_reproducer_locks, @@ -617,33 +618,6 @@ async def create_merge_request(state): package = result.package agent_input = InputSchema(jira_issue=state.jira_issue) if input_data is None else input_data - lock_id = await resolve_reproducer_lock_id( - agent_input.cve_id, - state.jira_issue, - fetch_issuelinks=fetch_jira_issue_issuelinks, - ) - lock_token: str | None = None - - if redis_conn is not None: - lock_token = await try_acquire_reproducer_lock( - redis_conn, - package, - lock_id, - jira_issue=state.jira_issue, - ) - if lock_token is None: - result.lock_deferred = True - result.summary = ( - (result.summary or "") - + " (Deferred: another worker holds the reproducer create/adapt lock)" - ).strip() - logger.info( - "Reproducer lock busy for %s/%s — deferring %s", - package, - lock_id, - state.jira_issue, - ) - return "handle_results" try: tests_clone = ( @@ -739,17 +713,6 @@ async def create_merge_request(state): result.test_mr_url = None result.success = False result.summary += f" (MR creation failed: {e})" - finally: - if lock_token is not None and redis_conn is not None: - try: - await release_reproducer_lock(redis_conn, package, lock_id, lock_token) - except Exception as e: - logger.warning( - "Failed to release reproducer lock for %s/%s: %s", - package, - lock_id, - e, - ) return "handle_results" @@ -825,6 +788,35 @@ async def handle_results(state): await tf_cleanup.cleanup(gateway_tools) +async def _stage_reproducer_in_progress( + *, + jira_issue: str, + dry_run: bool, + user_triggered: bool, + task: Task, +) -> None: + """Stamp ``ymir_reproducer_in_progress`` before queue work or while blocked on lock.""" + await tasks.set_jira_labels( + jira_issue=jira_issue, + labels_to_add=[JiraLabels.REPRODUCER_IN_PROGRESS.value], + labels_to_remove=list(_REPRODUCER_TERMINAL_LABELS), + dry_run=dry_run, + user_triggered=user_triggered, + critical=True, + ) + await tasks.post_user_ack_once( + task=task, + jira_issue=jira_issue, + agent_type="Reproducer", + comment_text=( + "Ymir picked up your request and started processing. " + "Results will be posted here when reproducer analysis completes." + ), + user_triggered=user_triggered, + dry_run=dry_run, + ) + + async def main() -> None: init_sentry() @@ -964,49 +956,91 @@ async def retry( ) await fix_await(redis.lpush(RedisQueues.ERROR_LIST.value, error)) - # ymir_reproducer_in_progress is the dedup anchor for the next - # fetcher sweep. If we cannot write it, we must not proceed — - # otherwise the fetcher will re-enqueue this issue and a second - # reproducer will run in parallel. - try: - await tasks.set_jira_labels( - jira_issue=input_data.jira_issue, - labels_to_add=[JiraLabels.REPRODUCER_IN_PROGRESS.value], - labels_to_remove=list(_REPRODUCER_TERMINAL_LABELS), - dry_run=dry_run, - user_triggered=user_triggered, - critical=True, + if not input_data.package: + logger.error( + "Reproducer task for %s is missing package metadata; cannot acquire lock", + input_data.jira_issue, ) - logger.info(f"Cleaned up existing labels for {input_data.jira_issue}") - # Post acknowledgement comment for user-triggered runs now that - # the in-progress label write succeeded. This prevents duplicate - # comments if the critical label write were to fail. - await tasks.post_user_ack_once( - task=task, - jira_issue=input_data.jira_issue, - agent_type="Reproducer", - comment_text=( - "Ymir picked up your request and started processing. " - "Results will be posted here when reproducer analysis completes." - ), - user_triggered=user_triggered, - dry_run=dry_run, + await retry( + task, + ErrorData( + details="Missing package in reproducer task metadata", + jira_issue=input_data.jira_issue, + ).model_dump_json(), ) - except Exception as e: - logger.error( - f"Could not set {JiraLabels.REPRODUCER_IN_PROGRESS.value} on " - f"{input_data.jira_issue} after retries: {e}; re-queuing to avoid duplicate reproducer." + return + + lock_id = await resolve_reproducer_lock_id( + input_data.cve_id, + input_data.jira_issue, + fetch_issuelinks=fetch_jira_issue_issuelinks, + ) + lock_token = await try_acquire_reproducer_lock( + redis, + input_data.package, + lock_id, + jira_issue=input_data.jira_issue, + ) + if lock_token is None: + try: + await _stage_reproducer_in_progress( + jira_issue=input_data.jira_issue, + dry_run=dry_run, + user_triggered=user_triggered, + task=task, + ) + except Exception as e: + logger.error( + "Could not set %s on blocked reproducer %s: %s", + JiraLabels.REPRODUCER_IN_PROGRESS.value, + input_data.jira_issue, + e, + ) + await retry( + task, + ErrorData( + details=f"Failed to set in-progress label while blocked: {e}", + jira_issue=input_data.jira_issue, + ).model_dump_json(), + ) + await asyncio.sleep(60) + return + + await enqueue_blocked_reproducer_task( + redis, + input_data.package, + lock_id, + task.model_dump_json(), + ) + logger.info( + "Reproducer lock busy for %s/%s — blocked %s until lock is released", + input_data.package, + lock_id, + input_data.jira_issue, ) - error_msg = f"Failed to set in-progress label: {e}" - error_data = ErrorData(details=error_msg, jira_issue=input_data.jira_issue) - await retry(task, error_data.model_dump_json()) - # Long sleep on purpose: critical-write retries already burned - # ~7s, so we're past transient blips. Typical Jira outages last - # minutes; cycling faster just spams the API. - await asyncio.sleep(60) return try: + try: + await _stage_reproducer_in_progress( + jira_issue=input_data.jira_issue, + dry_run=dry_run, + user_triggered=user_triggered, + task=task, + ) + logger.info(f"Cleaned up existing labels for {input_data.jira_issue}") + except Exception as e: + logger.error( + f"Could not set {JiraLabels.REPRODUCER_IN_PROGRESS.value} on " + f"{input_data.jira_issue} after retries: {e}; " + "re-queuing to avoid duplicate reproducer." + ) + error_msg = f"Failed to set in-progress label: {e}" + error_data = ErrorData(details=error_msg, jira_issue=input_data.jira_issue) + await retry(task, error_data.model_dump_json()) + await asyncio.sleep(60) + return + logger.info(f"Starting reproducer processing for {input_data.jira_issue}") with span_processor.start_transaction(input_data.jira_issue, workflow="reproducer"): state = await run_workflow( @@ -1032,16 +1066,15 @@ async def retry( ErrorData(details=error, jira_issue=input_data.jira_issue).model_dump_json(), ) else: - if output.retryable_error or output.lock_deferred: - reason = "lock contention" if output.lock_deferred else "retryable infra error" + if output.retryable_error: logger.info( - f"Reproducer {reason} for {input_data.jira_issue}; " + f"Reproducer retryable infra error for {input_data.jira_issue}; " f"scheduling retry in {retry_delay_seconds:.0f}s" ) await retry( task, ErrorData( - details=output.summary or f"Reproducer deferred: {reason}", + details=output.summary or "Reproducer deferred: retryable infra error", jira_issue=input_data.jira_issue, ).model_dump_json(), delay_seconds=retry_delay_seconds, @@ -1059,6 +1092,21 @@ async def retry( logger.info( f"Pushed {input_data.jira_issue} to {RedisQueues.COMPLETED_REPRODUCER_LIST.value}" ) + finally: + try: + await release_reproducer_lock( + redis, + input_data.package, + lock_id, + lock_token, + ) + except Exception as e: + logger.warning( + "Failed to release reproducer lock for %s/%s: %s", + input_data.package, + lock_id, + e, + ) await run_task_loop( redis, diff --git a/ymir/agents/tests/unit/test_reproducer_agent.py b/ymir/agents/tests/unit/test_reproducer_agent.py index f52c3842c..32662bc97 100644 --- a/ymir/agents/tests/unit/test_reproducer_agent.py +++ b/ymir/agents/tests/unit/test_reproducer_agent.py @@ -1,5 +1,6 @@ """Unit tests for reproducer agent label and comment helpers.""" +import contextlib from datetime import UTC, datetime from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -483,11 +484,33 @@ async def fake_run_tool(name, available_tools=None, **kwargs): def _make_reproducer_payload(issue: str = "RHEL-99999", user_triggered: bool = False) -> bytes: - input_data = ReproducerInputSchema(jira_issue=issue) + input_data = ReproducerInputSchema(jira_issue=issue, package="bind") task = Task(metadata=input_data.model_dump(), user_triggered=user_triggered) return task.model_dump_json().encode() +@contextlib.contextmanager +def _mock_workflow_lock(): + with ( + patch( + "ymir.agents.reproducer_agent.resolve_reproducer_lock_id", + new_callable=AsyncMock, + return_value="RHEL-99999", + ), + patch( + "ymir.agents.reproducer_agent.try_acquire_reproducer_lock", + new_callable=AsyncMock, + return_value='{"package":"bind","lock_id":"RHEL-99999","jira_issue":"RHEL-99999"}', + ), + patch( + "ymir.agents.reproducer_agent.release_reproducer_lock", + new_callable=AsyncMock, + return_value=True, + ), + ): + yield + + async def _run_process_task(payload: bytes) -> None: """Run reproducer main() in queue mode, invoking process_task with payload once. @@ -566,6 +589,7 @@ async def test_process_task_proceeds_despite_terminal_label_when_user_triggered( patch("ymir.agents.tasks.set_jira_labels", new_callable=AsyncMock), patch("ymir.agents.tasks.post_user_ack_once", new_callable=AsyncMock), patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + _mock_workflow_lock(), ): mock_workflow.return_value = MagicMock( result=MagicMock(success=True, retryable_error=False, lock_deferred=False, summary="ok") @@ -588,6 +612,7 @@ async def test_process_task_proceeds_when_terminal_label_and_in_progress(): patch("ymir.agents.tasks.set_jira_labels", new_callable=AsyncMock), patch("ymir.agents.tasks.post_user_ack_once", new_callable=AsyncMock), patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + _mock_workflow_lock(), ): mock_workflow.return_value = MagicMock( result=MagicMock(success=True, retryable_error=False, lock_deferred=False, summary="ok") @@ -609,6 +634,7 @@ async def test_process_task_proceeds_when_no_terminal_labels(): patch("ymir.agents.tasks.set_jira_labels", new_callable=AsyncMock), patch("ymir.agents.tasks.post_user_ack_once", new_callable=AsyncMock), patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + _mock_workflow_lock(), ): mock_workflow.return_value = MagicMock( result=MagicMock(success=True, retryable_error=False, lock_deferred=False, summary="ok") @@ -616,3 +642,36 @@ async def test_process_task_proceeds_when_no_terminal_labels(): await _run_process_task(_make_reproducer_payload()) mock_workflow.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_process_task_blocks_when_workflow_lock_busy(): + """Busy create/adapt locks park the task until the holder releases.""" + with ( + patch( + "ymir.agents.tasks.get_jira_issue_metadata", + new_callable=AsyncMock, + return_value=([], "New"), + ), + patch("ymir.agents.tasks.set_jira_labels", new_callable=AsyncMock), + patch("ymir.agents.tasks.post_user_ack_once", new_callable=AsyncMock), + patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + patch( + "ymir.agents.reproducer_agent.resolve_reproducer_lock_id", + new_callable=AsyncMock, + return_value="CVE-2026-56132", + ), + patch( + "ymir.agents.reproducer_agent.try_acquire_reproducer_lock", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "ymir.agents.reproducer_agent.enqueue_blocked_reproducer_task", + new_callable=AsyncMock, + ) as mock_enqueue_blocked, + ): + await _run_process_task(_make_reproducer_payload()) + + mock_workflow.assert_not_awaited() + mock_enqueue_blocked.assert_awaited_once() diff --git a/ymir/common/models.py b/ymir/common/models.py index 38ab2fbff..08106e313 100644 --- a/ymir/common/models.py +++ b/ymir/common/models.py @@ -1346,8 +1346,8 @@ class ReproducerOutputSchema(BaseModel): lock_deferred: bool = Field( default=False, description=( - "True when create/adapt could not proceed because another worker holds " - "the reproducer lock; the task should be scheduled for delayed retry" + "Legacy output flag; queue orchestration now blocks at workflow start " + "instead of deferring MR creation with a long delayed retry" ), ) retryable_error: bool = Field( diff --git a/ymir/common/reproducer_lock.py b/ymir/common/reproducer_lock.py index e139d02ce..e6ed09e1b 100644 --- a/ymir/common/reproducer_lock.py +++ b/ymir/common/reproducer_lock.py @@ -4,6 +4,11 @@ ``package:lock_id`` so only one worker creates or adapts the canonical ``Security//`` or ``Regression//`` test at a time. +The lock is acquired at **workflow start** (queue mode) and held until the +worker finishes analysis, Testing Farm verification, and MR push (or skip). +Tasks that cannot acquire the lock are parked on a per-lock blocked list and +promoted back to the main reproducer queue when the lock is released. + For CVE jobs *lock_id* is the normalized CVE id. For non-CVE bugs it is the root issue of the Jira Cloners chain (Y-stream root), resolved via issuelinks. """ @@ -21,6 +26,7 @@ logger = logging.getLogger(__name__) REPRODUCER_LOCK_HASH = "reproducer_creation_lock" +REPRODUCER_BLOCKED_QUEUE_PREFIX = "reproducer_blocked" _DEFAULT_STALE_THRESHOLD = timedelta(hours=6) _ACQUIRE_LUA = """ @@ -186,6 +192,66 @@ def _active_field(package: str, lock_id: str) -> str: return f"{package}:{lock_id}:active" +def blocked_reproducer_queue_key(package: str, lock_id: str) -> str: + """Per-lock Redis list for tasks waiting on ``package:lock_id``.""" + return f"{REPRODUCER_BLOCKED_QUEUE_PREFIX}:{package}:{lock_id}" + + +async def enqueue_blocked_reproducer_task( + redis_conn, + package: str, + lock_id: str, + payload: str, +) -> None: + """Park a task until the create/adapt lock for ``package:lock_id`` is free.""" + key = blocked_reproducer_queue_key(package, lock_id) + await fix_await(redis_conn.rpush(key, payload)) + logger.info( + "Blocked reproducer task for %s/%s — waiting for lock (queue %s)", + package, + lock_id, + key, + ) + + +async def promote_blocked_reproducer_tasks( + redis_conn, + package: str, + lock_id: str, +) -> int: + """Move tasks blocked on ``package:lock_id`` back to their target list queues.""" + from ymir.common.constants import RedisQueues + from ymir.common.models import Task + + key = blocked_reproducer_queue_key(package, lock_id) + promoted = 0 + while True: + raw = await fix_await(redis_conn.lpop(key)) + if raw is None: + break + payload = raw.decode() if isinstance(raw, bytes) else str(raw) + try: + task = Task.model_validate_json(payload) + except Exception: + logger.warning("Skipping invalid blocked reproducer payload on %s", key) + continue + target = ( + RedisQueues.REPRODUCER_QUEUE_TODO.value + if task.user_triggered + else RedisQueues.REPRODUCER_QUEUE.value + ) + await fix_await(redis_conn.lpush(target, payload)) + promoted += 1 + logger.info( + "Promoted blocked reproducer task for %s to %s (lock %s/%s)", + task.metadata.get("jira_issue", "?"), + target, + package, + lock_id, + ) + return promoted + + async def try_acquire_reproducer_lock( redis_conn, package: str, @@ -238,6 +304,7 @@ async def release_reproducer_lock( deleted = await fix_await(redis_conn.eval(_CONDITIONAL_HDEL_LUA, 1, REPRODUCER_LOCK_HASH, field, token)) if deleted: logger.info("Released reproducer lock for %s/%s", package, lock_id) + await promote_blocked_reproducer_tasks(redis_conn, package, lock_id) return True logger.warning( "Reproducer lock for %s/%s was not released — token no longer matches " @@ -287,5 +354,10 @@ async def sweep_stale_reproducer_locks( age, threshold, ) + await promote_blocked_reproducer_tasks( + redis_conn, + entry.package, + entry.lock_id, + ) return removed diff --git a/ymir/common/tests/unit/test_reproducer_lock.py b/ymir/common/tests/unit/test_reproducer_lock.py index b59c73301..89c379e1c 100644 --- a/ymir/common/tests/unit/test_reproducer_lock.py +++ b/ymir/common/tests/unit/test_reproducer_lock.py @@ -9,6 +9,9 @@ REPRODUCER_LOCK_HASH, ReproducerLockEntry, _immediate_clone_parent, + blocked_reproducer_queue_key, + enqueue_blocked_reproducer_task, + promote_blocked_reproducer_tasks, release_reproducer_lock, reproducer_lock_id, resolve_clone_root, @@ -148,6 +151,7 @@ async def test_try_acquire_reproducer_lock_busy(): async def test_release_reproducer_lock_compare_and_delete(): redis = MagicMock() redis.eval = AsyncMock(return_value=1) + redis.lpop = AsyncMock(return_value=None) token = ReproducerLockEntry(package="bind", lock_id="CVE-1", jira_issue="RHEL-1").model_dump_json() assert await release_reproducer_lock(redis, "bind", "CVE-1", token) is True @@ -158,6 +162,29 @@ async def test_release_reproducer_lock_compare_and_delete(): assert args[3] == "bind:CVE-1:active" assert args[4] == token redis.hdel.assert_not_called() + redis.lpop.assert_awaited_once_with(blocked_reproducer_queue_key("bind", "CVE-1")) + + +@pytest.mark.asyncio +async def test_promote_blocked_reproducer_tasks(): + payload = '{"metadata":{"jira_issue":"RHEL-2","package":"bind"},"attempts":0,"user_triggered":false}' + redis = MagicMock() + redis.lpop = AsyncMock(side_effect=[payload.encode(), None]) + redis.lpush = AsyncMock() + + promoted = await promote_blocked_reproducer_tasks(redis, "bind", "CVE-1") + assert promoted == 1 + redis.lpush.assert_awaited_once_with("reproducer_queue", payload) + + +@pytest.mark.asyncio +async def test_enqueue_blocked_reproducer_task(): + redis = MagicMock() + redis.rpush = AsyncMock() + payload = '{"metadata":{"jira_issue":"RHEL-2","package":"bind"}}' + + await enqueue_blocked_reproducer_task(redis, "bind", "CVE-1", payload) + redis.rpush.assert_awaited_once_with(blocked_reproducer_queue_key("bind", "CVE-1"), payload) @pytest.mark.asyncio @@ -197,6 +224,7 @@ async def test_sweep_stale_reproducer_locks_removes_old(): } ) redis.eval = AsyncMock(return_value=1) + redis.lpop = AsyncMock(return_value=None) removed = await sweep_stale_reproducer_locks(redis, threshold=timedelta(hours=6)) assert removed == 1 From 5be985a5cc3de226568d142cd3d30ace82df78ee Mon Sep 17 00:00:00 2001 From: Tomas Korbar Date: Fri, 21 Aug 2026 13:20:33 +0200 Subject: [PATCH 2/5] Reproducer: bootstrap MR branch before agent and unify clone-chain MRs Prepare the tests clone on the open MR fork branch before the agent runs so adapt jobs edit the canonical directory in place. Match regression clone chains to one MR via clone-root title tags, aligned with the create/adapt lock. Co-authored-by: Cursor --- ymir/agents/prompts/reproducer/prompt.j2 | 25 +- ymir/agents/reproducer_agent.py | 471 ++++++++++++++---- .../tests/unit/test_reproducer_agent.py | 253 ++++++++-- 3 files changed, 601 insertions(+), 148 deletions(-) diff --git a/ymir/agents/prompts/reproducer/prompt.j2 b/ymir/agents/prompts/reproducer/prompt.j2 index 081ebebde..5bd861520 100644 --- a/ymir/agents/prompts/reproducer/prompt.j2 +++ b/ymir/agents/prompts/reproducer/prompt.j2 @@ -94,11 +94,22 @@ Execute the following steps in order. same MR. Search open MRs primarily by **`{{ cve_id }}`** in title/description, not only by `{{ jira_issue }}`. + {% if tests_clone_ready %} + **Pre-provisioned tests clone (orchestration):** The tests repository is already + cloned at `{{ tests_clone_path }}`{% if mr_source_branch %} on branch + `{{ mr_source_branch }}`{% endif %}{% if existing_mr_url %} for open MR + {{ existing_mr_url }}{% endif %}{% if existing_test_directory %}. The existing + reproducer test is at `{{ existing_test_directory }}/` — **adapt it in place**; + do NOT create a parallel directory or call `clone_repository` for this path{% else %}. + Do NOT call `clone_repository` for this path — it would destroy the checked-out + branch{% endif %}. + {% else %} * Clone the RHEL tests repository using `clone_repository`: - URL: `https://gitlab.com/redhat/rhel/tests/` - Do NOT specify a `branch` parameter — omit it so the tool clones the default branch. - Use clone path `{{ reproducer_working_dir }}/tests-`. - If the clone path already exists, `clone_repository` removes and re-clones it — do NOT delete it yourself with `rm -rf`. + {% endif %} * Determine the expected test directory by inspecting the tests repo layout for this package (names are package-specific; do not assume a fixed scheme): - For CVEs: typically under `Security/` (often `Security//`, but @@ -126,11 +137,11 @@ Execute the following steps in order. (CVE, stable across streams) or ``: [RHEL-…] ymir reproducer test`` (regression). When a regression test is adapted for another stream, the MR title gains an additional ``[RHEL-…]`` key (e.g. ``[RHEL-100, RHEL-200]``). - If an open MR matches, note its URL as `existing_mr_url`. **Check out the MR - source branch** in the clone (via `get_merge_request_details` + `fetch_branch` + - `git checkout`) **before** copying/running the test — the test usually lives - only on that branch, not on the default branch yet. The MR does not need to be - merged first. + If an open MR matches, note its URL as `existing_mr_url`.{% if not tests_clone_ready %} + **Check out the MR source branch** in the clone (via `get_merge_request_details` + + `fetch_branch` + `git checkout`) **before** copying/running the test — the test + usually lives only on that branch, not on the default branch yet.{% endif %} + The MR does not need to be merged first. * **If an existing test directory / matching test / open MR is found:** 1. Continue to steps 3–6 to reserve a Testing Farm machine for **this** @@ -239,7 +250,11 @@ Execute the following steps in order. 4.1. Use the Already-Cloned Tests Repository + {% if tests_clone_ready %} + The tests repository was prepared by orchestration at `{{ tests_clone_path }}`{% if mr_source_branch %} on branch `{{ mr_source_branch }}`{% endif %}{% if existing_test_directory %} with the canonical test at `{{ existing_test_directory }}/`{% endif %}. Do NOT call `clone_repository` for this path. + {% else %} The tests repository was already cloned in step 1.7 at `{{ reproducer_working_dir }}/tests-`. If for any reason the clone is missing, re-clone it using `clone_repository` with URL `https://gitlab.com/redhat/rhel/tests/` (no `branch` parameter) to `{{ reproducer_working_dir }}/tests-`. + {% endif %} * Create the test directory under the tests clone. Prefer package convention: - CVEs: often `/Security//` diff --git a/ymir/agents/reproducer_agent.py b/ymir/agents/reproducer_agent.py index ab43158a3..f4d7bce55 100644 --- a/ymir/agents/reproducer_agent.py +++ b/ymir/agents/reproducer_agent.py @@ -3,10 +3,9 @@ import logging import os import re -import shutil import sys -import tempfile import traceback +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -34,7 +33,6 @@ mcp_tools, render_template, resolve_chat_model_override, - run_subprocess, run_tool, ) from ymir.common.base_utils import fix_await, redis_client, run_task_loop @@ -56,6 +54,7 @@ from ymir.common.reproducer_lock import ( enqueue_blocked_reproducer_task, release_reproducer_lock, + resolve_clone_root, resolve_reproducer_lock_id, sweep_stale_reproducer_locks, try_acquire_reproducer_lock, @@ -157,17 +156,61 @@ class _PromptContext(InputSchema): dry_run: bool = Field(default=False) reproducer_working_dir: str = Field(description="Per-issue working directory on the shared git volume") + tests_clone_ready: bool = Field( + default=False, + description="True when orchestration already cloned the tests repo before the agent runs", + ) + tests_clone_path: str | None = Field( + default=None, + description="Absolute path to the pre-provisioned tests clone", + ) + existing_mr_url: str | None = Field( + default=None, + description="Open reproducer MR URL when the tests clone was bootstrapped for adapt", + ) + mr_source_branch: str | None = Field( + default=None, + description="MR source branch checked out in the pre-provisioned tests clone", + ) + existing_test_directory: str | None = Field( + default=None, + description="Relative test directory path already on the MR branch", + ) + + +@dataclass +class PreparedTestsClone: + """Tests-repo layout prepared before the reproducer agent runs.""" + tests_clone: Path + existing_mr_url: str | None = None + mr_source_branch: str | None = None + existing_test_directory: str | None = None + matched_mr: dict | None = None -def _render_prompt(input_data: InputSchema, dry_run: bool = False) -> str: + +TestsCloneBootstrap = PreparedTestsClone + + +def _render_prompt( + input_data: InputSchema, + dry_run: bool = False, + bootstrap: TestsCloneBootstrap | None = None, +) -> str: """Render the reproducer prompt template with the input schema fields.""" working_dir = ( Path(os.environ.get("GIT_REPO_BASEPATH", "/git-repos")) / "Reproducer" / input_data.jira_issue ) + default_clone = working_dir / f"tests-{input_data.package}" if input_data.package else working_dir context = _PromptContext( **input_data.model_dump(), dry_run=dry_run, reproducer_working_dir=str(working_dir), + tests_clone_ready=bootstrap is not None, + tests_clone_path=str(bootstrap.tests_clone if bootstrap else default_clone), + existing_mr_url=bootstrap.existing_mr_url if bootstrap else None, + mr_source_branch=bootstrap.mr_source_branch if bootstrap else None, + existing_test_directory=bootstrap.existing_test_directory if bootstrap else None, ) return render_template(_PROMPT_TEMPLATE, context) @@ -249,6 +292,76 @@ def _resolve_test_dir(tests_clone: Path, test_directory: str | None) -> Path | N return None +def _is_reproducer_test_dir(path: Path) -> bool: + """Return whether *path* looks like a reproducer test directory.""" + return path.is_dir() and ( + (path / "main.fmf").is_file() + or (path / "runtest.sh").is_file() + or (path / "ai-test-description").is_file() + ) + + +def _discover_existing_reproducer_test_dir( + tests_clone: Path, + *, + cve_id: str | None, + jira_issue: str, + reproducer_type: str, + clone_root: str | None = None, +) -> Path | None: + """Find the reproducer test directory already on an open MR branch. + + When a sibling stream adapts an existing MR, the agent may report a fresh + ``test_directory`` under the default branch layout. After checking out the + MR tip, prefer the directory that is already part of that MR. + """ + cves = _cve_only_needles(cve_id) + if cves: + for cve in cves: + candidate = tests_clone / "Security" / cve + if _is_reproducer_test_dir(candidate): + return candidate + + security = tests_clone / "Security" + if security.is_dir(): + matches = [ + child + for child in security.iterdir() + if child.is_dir() + and _is_reproducer_test_dir(child) + and any(cve in child.name.upper() for cve in cves) + ] + if len(matches) == 1: + return matches[0] + for cve in cves: + for match in matches: + if match.name.upper() == cve: + return match + return None + + if reproducer_type == "bug": + search_keys: list[str] = [] + root = (clone_root or "").upper() + issue = jira_issue.upper() + if root and root not in search_keys: + search_keys.append(root) + if issue not in search_keys: + search_keys.append(issue) + for key in search_keys: + candidate = tests_clone / "Regression" / key + if _is_reproducer_test_dir(candidate): + return candidate + + regression = tests_clone / "Regression" + if regression.is_dir(): + matches = [ + child for child in regression.iterdir() if child.is_dir() and _is_reproducer_test_dir(child) + ] + if len(matches) == 1: + return matches[0] + return None + + def _cve_only_needles(cve_id: str | None) -> list[str]: """CVE id strings used to match sibling-stream reproducer MRs.""" if not cve_id or not cve_id.strip(): @@ -296,13 +409,22 @@ def _is_reproducer_mr_title(title: str) -> bool: return "ymir reproducer test" in title.lower() -def _match_regression_sibling_mr(mrs: list[dict], jira_issue: str) -> dict | None: +def _match_regression_sibling_mr( + mrs: list[dict], + jira_issue: str, + *, + clone_root: str | None = None, +) -> dict | None: """Find the canonical regression reproducer MR to extend for another stream. - When the current issue is not yet listed in the title, match a sole open - regression reproducer MR (no ``[CVE-…]`` tag in the title). + Clone-chain siblings share one MR keyed by the root issue's ``[RHEL-…]`` tag + (same id as the create/adapt lock). When the current issue is not yet listed + in the title, match an MR tagged with the clone root before falling back to a + sole open regression reproducer MR. """ wanted = jira_issue.upper() + root = (clone_root or wanted).upper() + root_match: dict | None = None candidates: list[dict] = [] for mr in mrs: title = mr.get("title") or "" @@ -311,17 +433,35 @@ def _match_regression_sibling_mr(mrs: list[dict], jira_issue: str) -> dict | Non continue if wanted in title_jiras: return mr + if root in title_jiras: + root_match = mr candidates.append(mr) + if root_match is not None: + return root_match if len(candidates) == 1: return candidates[0] return None +async def _resolve_reproducer_clone_root(jira_issue: str) -> str: + """Return the Cloners-chain root issue key (uppercase) for MR/lock grouping.""" + try: + return (await resolve_clone_root(jira_issue, fetch_jira_issue_issuelinks)).upper() + except Exception: + logger.warning( + "Failed to resolve clone root for %s; using issue key for reproducer MR match", + jira_issue, + exc_info=True, + ) + return jira_issue.upper() + + def _match_open_reproducer_mr( mrs: list[dict], *, cve_ids: list[str] | None = None, jira_issue: str | None = None, + clone_root: str | None = None, existing_mr_url: str | None = None, ) -> dict | None: """Return the open reproducer MR for this CVE or Jira issue. @@ -336,6 +476,7 @@ def _match_open_reproducer_mr( wanted_cves = {cve.upper() for cve in cve_ids or [] if cve} wanted_jira = jira_issue.upper() if jira_issue else None + root_jira = clone_root.upper() if clone_root else None for mr in mrs: title = mr.get("title") or "" @@ -345,6 +486,8 @@ def _match_open_reproducer_mr( return mr if wanted_jira and wanted_jira in title_jiras: return mr + if root_jira and root_jira in title_jiras: + return mr return None @@ -366,11 +509,125 @@ async def _list_open_reproducer_mrs(package: str, available_tools: list[Any]) -> return mrs if isinstance(mrs, list) else [] +async def _match_open_reproducer_mr_for_input( + input_data: InputSchema, + mrs: list[dict], +) -> dict | None: + """Match an open reproducer MR from queue input (before the agent runs).""" + cve_needles = _cve_only_needles(input_data.cve_id) + if cve_needles: + return _match_open_reproducer_mr(mrs, cve_ids=cve_needles) + clone_root = await _resolve_reproducer_clone_root(input_data.jira_issue) + matched = _match_open_reproducer_mr( + mrs, + jira_issue=input_data.jira_issue, + clone_root=clone_root, + ) + if matched is None: + matched = _match_regression_sibling_mr( + mrs, + input_data.jira_issue, + clone_root=clone_root, + ) + return matched + + +async def _bootstrap_tests_clone( + working_dir: Path, + input_data: InputSchema, + available_tools: list[Any], +) -> TestsCloneBootstrap: + """Clone the tests repo and check out an open reproducer MR branch when present.""" + package = input_data.package + if not package: + raise ValueError("package is required to bootstrap tests clone") + + tests_clone = working_dir / f"tests-{package}" + repository = f"https://gitlab.com/redhat/rhel/tests/{package}" + + await run_tool( + "clone_repository", + repository=repository, + clone_path=str(tests_clone), + available_tools=available_tools, + ) + + mrs = await _list_open_reproducer_mrs(package, available_tools) + matched = await _match_open_reproducer_mr_for_input(input_data, mrs) + if not matched: + logger.info( + "No open reproducer MR for %s — tests clone left on default branch", + input_data.jira_issue, + ) + return TestsCloneBootstrap(tests_clone=tests_clone) + + mr_url = matched.get("url") + if not mr_url: + logger.warning("Matched reproducer MR for %s has no URL", input_data.jira_issue) + return TestsCloneBootstrap(tests_clone=tests_clone, matched_mr=matched) + + details_raw = await run_tool( + "get_merge_request_details", + merge_request_url=mr_url, + available_tools=available_tools, + ) + details = MergeRequestDetails.model_validate(details_raw) + branch = details.source_branch or matched.get("source_branch") + if not branch: + raise RuntimeError(f"Open reproducer MR {mr_url} has no source branch") + + await run_tool( + "fetch_branch", + repository=details.source_repo, + branch=branch, + clone_path=str(tests_clone), + available_tools=available_tools, + ) + await check_subprocess(["git", "checkout", "-f", branch], cwd=tests_clone) + + reproducer_type = "cve" if _cve_only_needles(input_data.cve_id) else "bug" + clone_root = None + if reproducer_type == "bug": + clone_root = await _resolve_reproducer_clone_root(input_data.jira_issue) + discovered = _discover_existing_reproducer_test_dir( + tests_clone, + cve_id=input_data.cve_id, + jira_issue=input_data.jira_issue, + reproducer_type=reproducer_type, + clone_root=clone_root, + ) + existing_test_directory = None + if discovered: + existing_test_directory = str(discovered.relative_to(tests_clone)) + else: + logger.warning( + "Checked out reproducer MR %s on %s but found no test directory on branch", + mr_url, + branch, + ) + + logger.info( + "Bootstrapped tests clone for %s on MR branch %s (test dir: %s)", + input_data.jira_issue, + branch, + existing_test_directory or "unknown", + ) + return TestsCloneBootstrap( + tests_clone=tests_clone, + existing_mr_url=mr_url, + mr_source_branch=branch, + existing_test_directory=existing_test_directory, + matched_mr=matched, + ) + + async def _resolve_reproducer_mr_target( result: OutputSchema, agent_input: InputSchema, package: str, available_tools: list[Any], + *, + bootstrap: TestsCloneBootstrap | None = None, ) -> tuple[str | None, str, dict | None]: """Resolve MR URL, git branch, and matched MR metadata for create/adapt push. @@ -380,23 +637,31 @@ async def _resolve_reproducer_mr_target( MR title when another stream extends the same open MR. """ fallback_branch = f"reproducer/{result.jira_issue}" - mrs = await _list_open_reproducer_mrs(package, available_tools) - - cve_needles = _cve_only_needles(agent_input.cve_id) - if cve_needles: - matched = _match_open_reproducer_mr( - mrs, - cve_ids=cve_needles, - existing_mr_url=result.existing_mr_url, - ) + if bootstrap and bootstrap.matched_mr: + matched = bootstrap.matched_mr else: - matched = _match_open_reproducer_mr( - mrs, - jira_issue=result.jira_issue, - existing_mr_url=result.existing_mr_url, - ) - if matched is None and result.reproducer_type == "bug": - matched = _match_regression_sibling_mr(mrs, result.jira_issue) + mrs = await _list_open_reproducer_mrs(package, available_tools) + cve_needles = _cve_only_needles(agent_input.cve_id) + if cve_needles: + matched = _match_open_reproducer_mr( + mrs, + cve_ids=cve_needles, + existing_mr_url=result.existing_mr_url, + ) + else: + clone_root = await _resolve_reproducer_clone_root(agent_input.jira_issue) + matched = _match_open_reproducer_mr( + mrs, + jira_issue=result.jira_issue, + clone_root=clone_root, + existing_mr_url=result.existing_mr_url, + ) + if matched is None and result.reproducer_type == "bug": + matched = _match_regression_sibling_mr( + mrs, + result.jira_issue, + clone_root=clone_root, + ) if matched: mr_url = matched.get("url") @@ -420,84 +685,56 @@ async def _prepare_reproducer_branch( test_dir: Path, update_branch: str, *, - adapted_existing: bool, existing_mr_url: str | None, available_tools: list[Any], -) -> str: - """Checkout the commit/push branch while preserving local ``test_dir`` edits. + bootstrap: TestsCloneBootstrap | None = None, +) -> tuple[str, Path]: + """Ensure the local clone is on the branch that will be pushed. - For adaptations of an open MR, fetch and check out that MR's source-branch - tip first (same idea as ``prepare_dist_git_from_merge_request``) so a later - force-push cannot drop sibling commits already on the MR. New MRs create - ``update_branch`` from the current local HEAD. + When orchestration bootstrapped an open MR before the agent ran, the agent + already worked on the fork branch in place — do not re-checkout or overlay. """ - with tempfile.TemporaryDirectory() as tmp: - snapshot = Path(tmp) / "adapted_test" - shutil.copytree(test_dir, snapshot) + if bootstrap and bootstrap.mr_source_branch and bootstrap.existing_mr_url: + branch = bootstrap.mr_source_branch + head, _ = await check_subprocess(["git", "branch", "--show-current"], cwd=tests_clone) + if head.strip() != branch: + await check_subprocess(["git", "checkout", "-f", branch], cwd=tests_clone) + return branch, test_dir - branch = update_branch + if existing_mr_url: try: - if existing_mr_url: - try: - details_raw = await run_tool( - "get_merge_request_details", - merge_request_url=existing_mr_url, - available_tools=available_tools, - ) - details = MergeRequestDetails.model_validate(details_raw) - branch = details.source_branch or update_branch - - # Leave the target branch so fetch can update refs/heads/. - _, head_branch, _ = await run_subprocess( - ["git", "branch", "--show-current"], cwd=tests_clone - ) - if head_branch.strip() == branch: - await check_subprocess( - ["git", "checkout", "--detach"], - cwd=tests_clone, - ) - - await run_tool( - "fetch_branch", - repository=details.source_repo, - branch=branch, - clone_path=str(tests_clone), - available_tools=available_tools, - ) - await check_subprocess( - ["git", "checkout", "-f", branch], - cwd=tests_clone, - ) - logger.info( - "Checked out existing MR source branch %s for adapt (%s)", - branch, - existing_mr_url, - ) - except Exception as e: - logger.warning( - "Failed to fetch/checkout existing MR branch for %s " - "(wanted %s); falling back to checkout -B from local HEAD: %s", - existing_mr_url, - update_branch, - e, - ) - branch = update_branch - await check_subprocess( - ["git", "checkout", "-B", branch], - cwd=tests_clone, - ) - else: - await check_subprocess( - ["git", "checkout", "-B", branch], - cwd=tests_clone, - ) - finally: - # Overlay agent adaptations onto whatever tip we checked out. - if test_dir.exists(): - shutil.rmtree(test_dir) - shutil.copytree(snapshot, test_dir) + details_raw = await run_tool( + "get_merge_request_details", + merge_request_url=existing_mr_url, + available_tools=available_tools, + ) + details = MergeRequestDetails.model_validate(details_raw) + branch = details.source_branch or update_branch + await run_tool( + "fetch_branch", + repository=details.source_repo, + branch=branch, + clone_path=str(tests_clone), + available_tools=available_tools, + ) + await check_subprocess(["git", "checkout", "-f", branch], cwd=tests_clone) + logger.info( + "Checked out existing MR source branch %s for adapt (%s)", + branch, + existing_mr_url, + ) + return branch, test_dir + except Exception as e: + logger.warning( + "Failed to fetch/checkout existing MR branch for %s " + "(wanted %s); falling back to checkout -B from local HEAD: %s", + existing_mr_url, + update_branch, + e, + ) - return branch + await check_subprocess(["git", "checkout", "-B", update_branch], cwd=tests_clone) + return update_branch, test_dir def _build_mr_description(result: OutputSchema, input_data: InputSchema) -> str: @@ -531,7 +768,14 @@ def _build_mr_description(result: OutputSchema, input_data: InputSchema) -> str: def _build_commit_message(result: OutputSchema, input_data: InputSchema) -> str: """Build the commit message for the reproducer test.""" - if result.reproducer_type == "cve": + if result.adapted_existing: + if result.reproducer_type == "cve": + title = f"{result.package}: adapt security reproducer for {result.jira_issue}" + body = f"Adapt security test for {input_data.cve_id} in {result.package} for this stream." + else: + title = f"{result.package}: adapt regression reproducer for {result.jira_issue}" + body = f"Adapt regression test for {result.jira_issue} in {result.package} for this stream." + elif result.reproducer_type == "cve": title = f"{result.package}: add security reproducer for {result.jira_issue}" body = f"Add security test for {input_data.cve_id} in {result.package}." else: @@ -576,16 +820,19 @@ async def run_workflow( gateway_tools, local_tool_options, extra_middlewares=[tf_cleanup] ) + agent_input = InputSchema(jira_issue=jira_issue) if input_data is None else input_data + bootstrap: TestsCloneBootstrap | None = None + if agent_input.package: + bootstrap = await _bootstrap_tests_clone(working_dir, agent_input, gateway_tools) + workflow = Workflow(ReproducerState, name="ReproducerWorkflow") async def run_reproducer_analysis(state): """Run the reproducer agent.""" logger.info(f"Running reproducer analysis for {state.jira_issue}") - agent_input = InputSchema(jira_issue=state.jira_issue) if input_data is None else input_data - response = await reproducer_agent.run( - _render_prompt(agent_input, dry_run=dry_run), + _render_prompt(agent_input, dry_run=dry_run, bootstrap=bootstrap), expected_output=render_template("reproducer/output_format.j2"), **get_agent_execution_config(), ) @@ -645,29 +892,47 @@ async def create_merge_request(state): return "handle_results" logger.info("Using test directory %s for MR creation", test_dir) + if bootstrap and bootstrap.existing_test_directory and result.adapted_existing: + expected = bootstrap.existing_test_directory + actual = (result.test_directory or "").strip().lstrip("/") + if actual != expected: + logger.error( + "Adapt for %s used test_directory=%r but open MR test is at %r", + state.jira_issue, + actual, + expected, + ) + result.success = False + result.summary += ( + f" (MR creation skipped: when adapting open MR, " + f"test_directory must be {expected})" + ) + return "handle_results" + existing_mr_url, update_branch, matched_mr = await _resolve_reproducer_mr_target( result, agent_input, package, gateway_tools, + bootstrap=bootstrap, ) - update_branch = await _prepare_reproducer_branch( + update_branch, commit_dir = await _prepare_reproducer_branch( tests_clone, test_dir, update_branch, - adapted_existing=bool(result.adapted_existing), existing_mr_url=existing_mr_url, available_tools=gateway_tools, + bootstrap=bootstrap, ) # Make shell scripts executable before staging - for script in test_dir.glob("*.sh"): + for script in commit_dir.glob("*.sh"): script.chmod(0o755) - for script in test_dir.glob("*.ksh"): + for script in commit_dir.glob("*.ksh"): script.chmod(0o755) await check_subprocess( - ["git", "add", str(test_dir.relative_to(tests_clone))], + ["git", "add", str(commit_dir.relative_to(tests_clone))], cwd=tests_clone, ) diff --git a/ymir/agents/tests/unit/test_reproducer_agent.py b/ymir/agents/tests/unit/test_reproducer_agent.py index 32662bc97..2cb48b457 100644 --- a/ymir/agents/tests/unit/test_reproducer_agent.py +++ b/ymir/agents/tests/unit/test_reproducer_agent.py @@ -8,11 +8,15 @@ import pytest from ymir.agents.reproducer_agent import ( + PreparedTestsClone, + _bootstrap_tests_clone, _build_mr_title, _cve_only_needles, _determine_comment_resolution, _determine_result_label, + _discover_existing_reproducer_test_dir, _match_open_reproducer_mr, + _match_open_reproducer_mr_for_input, _match_regression_sibling_mr, _needs_merge_request, _prepare_reproducer_branch, @@ -117,6 +121,38 @@ def test_resolve_test_dir_rejects_traversal_and_missing(tmp_path: Path): assert _resolve_test_dir(tmp_path, "Security/CVE-missing") is None +def test_discover_existing_reproducer_test_dir_prefers_security_cve(tmp_path: Path): + repo = tmp_path / "tests-pkg" + repo.mkdir() + canonical = repo / "Security" / "CVE-2026-50219" + canonical.mkdir(parents=True) + (canonical / "main.fmf").write_text("summary: test\n") + + discovered = _discover_existing_reproducer_test_dir( + repo, + cve_id="CVE-2026-50219", + jira_issue="RHEL-220981", + reproducer_type="cve", + ) + assert discovered == canonical + + +def test_discover_existing_reproducer_test_dir_finds_sole_regression_dir(tmp_path: Path): + repo = tmp_path / "tests-pkg" + repo.mkdir() + regression = repo / "Regression" / "RHEL-100" + regression.mkdir(parents=True) + (regression / "runtest.sh").write_text("#!/bin/bash\n") + + discovered = _discover_existing_reproducer_test_dir( + repo, + cve_id=None, + jira_issue="RHEL-200", + reproducer_type="bug", + ) + assert discovered == regression + + def test_cve_only_needles_splits_and_normalizes(): assert _cve_only_needles("CVE-2026-56132") == ["CVE-2026-56132"] assert _cve_only_needles("cve-1; CVE-2") == ["CVE-1", "CVE-2"] @@ -168,13 +204,113 @@ def test_match_regression_sibling_mr_when_issue_not_yet_in_title(): }, { "url": "https://gitlab.com/a/2", - "title": "bind: [RHEL-200] ymir reproducer test", + "title": "bind: [RHEL-500] ymir reproducer test", }, ] assert _match_regression_sibling_mr(mrs, "RHEL-300") is None + assert _match_regression_sibling_mr(mrs, "RHEL-300", clone_root="RHEL-100") == mrs[0] single = [mrs[0]] assert _match_regression_sibling_mr(single, "RHEL-200") == single[0] + assert _match_regression_sibling_mr(single, "RHEL-200", clone_root="RHEL-100") == single[0] + + +def test_match_open_reproducer_mr_uses_clone_root_tag(): + mrs = [ + { + "url": "https://gitlab.com/a/1", + "title": "bind: [RHEL-100] ymir reproducer test", + }, + { + "url": "https://gitlab.com/a/2", + "title": "bind: [RHEL-500] ymir reproducer test", + }, + ] + assert _match_open_reproducer_mr(mrs, jira_issue="RHEL-200", clone_root="RHEL-100") == mrs[0] + assert _match_open_reproducer_mr(mrs, jira_issue="RHEL-600", clone_root="RHEL-500") == mrs[1] + + +def test_discover_existing_reproducer_test_dir_prefers_clone_root_regression_path(tmp_path: Path): + repo = tmp_path / "tests-pkg" + repo.mkdir() + root_dir = repo / "Regression" / "RHEL-100" + root_dir.mkdir(parents=True) + (root_dir / "runtest.sh").write_text("#!/bin/bash\n") + + discovered = _discover_existing_reproducer_test_dir( + repo, + cve_id=None, + jira_issue="RHEL-300", + reproducer_type="bug", + clone_root="RHEL-100", + ) + assert discovered == root_dir + + +@pytest.mark.asyncio +async def test_match_open_reproducer_mr_for_input_uses_clone_root(monkeypatch): + mrs = [ + { + "url": "https://gitlab.com/a/1", + "title": "bind: [RHEL-100] ymir reproducer test", + }, + { + "url": "https://gitlab.com/a/2", + "title": "bind: [RHEL-500] ymir reproducer test", + }, + ] + monkeypatch.setattr( + "ymir.agents.reproducer_agent._resolve_reproducer_clone_root", + AsyncMock(return_value="RHEL-100"), + ) + input_data = ReproducerInputSchema(jira_issue="RHEL-300", package="bind") + matched = await _match_open_reproducer_mr_for_input(input_data, mrs) + assert matched == mrs[0] + + +@pytest.mark.asyncio +async def test_resolve_reproducer_mr_target_extends_clone_chain_mr_with_multiple_open(): + result = _output( + jira_issue="RHEL-300", + success=True, + test_directory="Regression/RHEL-100", + package="bind", + reproducer_type="bug", + ) + agent_input = ReproducerInputSchema(jira_issue="RHEL-300", package="bind") + open_mrs = [ + { + "url": "https://gitlab.com/redhat/rhel/tests/bind/-/merge_requests/5", + "title": "bind: [RHEL-100] ymir reproducer test", + "source_branch": "reproducer/RHEL-100", + }, + { + "url": "https://gitlab.com/redhat/rhel/tests/bind/-/merge_requests/6", + "title": "bind: [RHEL-500] ymir reproducer test", + "source_branch": "reproducer/RHEL-500", + }, + ] + + async def fake_run_tool(name, available_tools=None, **kwargs): + if name == "list_project_merge_requests": + return open_mrs + raise AssertionError(name) + + with ( + patch("ymir.agents.reproducer_agent.run_tool", new=AsyncMock(side_effect=fake_run_tool)), + patch( + "ymir.agents.reproducer_agent._resolve_reproducer_clone_root", + new=AsyncMock(return_value="RHEL-100"), + ), + ): + mr_url, branch, matched_mr = await _resolve_reproducer_mr_target(result, agent_input, "bind", []) + + assert mr_url == open_mrs[0]["url"] + assert branch == "reproducer/RHEL-100" + assert ( + _build_mr_title(result, agent_input, matched_mr=matched_mr) + == "bind: [RHEL-100, RHEL-300] ymir reproducer test" + ) def test_build_mr_title_appends_jira_on_regression_adapt(): @@ -402,80 +538,117 @@ async def test_prepare_reproducer_branch_new_mr_preserves_test_dir(tmp_path: Pat test_dir.mkdir(parents=True) (test_dir / "runtest.sh").write_text("adapted-on-default\n") - branch = await _prepare_reproducer_branch( + branch, commit_dir = await _prepare_reproducer_branch( repo, test_dir, "reproducer/RHEL-1", - adapted_existing=False, existing_mr_url=None, available_tools=[], ) assert branch == "reproducer/RHEL-1" + assert commit_dir == test_dir head, _ = await check_subprocess(["git", "branch", "--show-current"], cwd=repo) assert head.strip() == "reproducer/RHEL-1" assert (test_dir / "runtest.sh").read_text() == "adapted-on-default\n" @pytest.mark.asyncio -async def test_prepare_reproducer_branch_adapt_keeps_sibling_commits(tmp_path: Path): - """Adapt must land on the MR tip (sibling commit), not wipe it via checkout -B HEAD.""" +async def test_prepare_reproducer_branch_bootstrapped_adapt_preserves_worktree(tmp_path: Path): + """Bootstrapped adapt must not re-overlay — agent edits are already on the MR branch.""" repo = tmp_path / "tests-pkg" repo.mkdir() await _git_init_with_main(repo) - # Simulate an existing MR branch that already has a sibling commit. await check_subprocess(["git", "checkout", "-b", "reproducer/RHEL-1"], cwd=repo) - mr_dir = repo / "Security" / "CVE-1" - mr_dir.mkdir(parents=True) - (mr_dir / "runtest.sh").write_text("sibling-stream\n") + test_dir = repo / "Security" / "CVE-1" + test_dir.mkdir(parents=True) + (test_dir / "main.fmf").write_text("summary: sibling\n") + (test_dir / "runtest.sh").write_text("local-adapt\n") await check_subprocess(["git", "add", "Security"], cwd=repo) await check_subprocess(["git", "commit", "-m", "sibling adapt"], cwd=repo) sibling_sha, _ = await check_subprocess(["git", "rev-parse", "HEAD"], cwd=repo) - # Agent continued on main with a local adaptation of the same test path. - await check_subprocess(["git", "checkout", "main"], cwd=repo) - test_dir = repo / "Security" / "CVE-1" - test_dir.mkdir(parents=True) - (test_dir / "runtest.sh").write_text("local-adapt\n") + bootstrap = PreparedTestsClone( + tests_clone=repo, + existing_mr_url="https://gitlab.com/redhat/rhel/tests/pkg/-/merge_requests/1", + mr_source_branch="reproducer/RHEL-1", + existing_test_directory="Security/CVE-1", + matched_mr={"url": "https://gitlab.com/redhat/rhel/tests/pkg/-/merge_requests/1"}, + ) - details = MergeRequestDetails( - source_repo="https://gitlab.com/fork/tests-pkg.git", - source_branch="reproducer/RHEL-1", - target_repo_name="pkg", - target_branch="main", - title="adapt", - description="", - last_updated_at=datetime.now(UTC), - comments=[], + branch, commit_dir = await _prepare_reproducer_branch( + repo, + test_dir, + "reproducer/RHEL-1", + existing_mr_url=bootstrap.existing_mr_url, + available_tools=[], + bootstrap=bootstrap, ) + assert branch == "reproducer/RHEL-1" + assert commit_dir == test_dir + sha, _ = await check_subprocess(["git", "rev-parse", "HEAD"], cwd=repo) + assert sha.strip() == sibling_sha.strip() + assert (test_dir / "runtest.sh").read_text() == "local-adapt\n" + + +@pytest.mark.asyncio +async def test_bootstrap_tests_clone_checks_out_existing_mr_branch(tmp_path: Path, monkeypatch): + working_dir = tmp_path / "Reproducer" / "RHEL-2" + working_dir.mkdir(parents=True) + repo = working_dir / "tests-bind" + repo.mkdir() + cve_id = "CVE-2026-0001" + async def fake_run_tool(name, available_tools=None, **kwargs): + if name == "clone_repository": + (repo / "README").write_text("cloned\n") + return "ok" + if name == "list_project_merge_requests": + return [ + { + "url": "https://gitlab.com/redhat/rhel/tests/bind/-/merge_requests/9", + "title": f"bind: [{cve_id}] ymir reproducer test", + "source_branch": "reproducer/RHEL-1", + } + ] if name == "get_merge_request_details": - return details.model_dump(mode="json") + return MergeRequestDetails( + source_repo="https://gitlab.com/fork/tests-bind.git", + source_branch="reproducer/RHEL-1", + target_repo_name="bind", + target_branch="main", + title=f"bind: [{cve_id}] ymir reproducer test", + description="", + last_updated_at=datetime.now(UTC), + comments=[], + ).model_dump(mode="json") if name == "fetch_branch": - # Local stand-in: branch already exists; nothing to fetch. + await _git_init_with_main(repo) + await check_subprocess(["git", "checkout", "-b", "reproducer/RHEL-1"], cwd=repo) + mr_dir = repo / "Security" / cve_id + mr_dir.mkdir(parents=True) + (mr_dir / "main.fmf").write_text("summary: on mr\n") + await check_subprocess(["git", "add", "Security"], cwd=repo) + await check_subprocess(["git", "commit", "-m", "mr test"], cwd=repo) return "ok" raise AssertionError(f"unexpected tool {name}") - with patch("ymir.agents.reproducer_agent.run_tool", new=AsyncMock(side_effect=fake_run_tool)): - branch = await _prepare_reproducer_branch( - repo, - test_dir, - "reproducer/RHEL-1", - adapted_existing=True, - existing_mr_url="https://gitlab.com/redhat/rhel/tests/pkg/-/merge_requests/1", - available_tools=[], - ) + monkeypatch.setattr("ymir.agents.reproducer_agent.run_tool", fake_run_tool) - assert branch == "reproducer/RHEL-1" + input_data = ReproducerInputSchema( + jira_issue="RHEL-2", + package="bind", + cve_id=cve_id, + ) + bootstrap = await _bootstrap_tests_clone(working_dir, input_data, []) + + assert bootstrap.existing_mr_url.endswith("/merge_requests/9") + assert bootstrap.mr_source_branch == "reproducer/RHEL-1" + assert bootstrap.existing_test_directory == f"Security/{cve_id}" head, _ = await check_subprocess(["git", "branch", "--show-current"], cwd=repo) assert head.strip() == "reproducer/RHEL-1" - # HEAD is still the sibling commit (not a reset of main). - sha, _ = await check_subprocess(["git", "rev-parse", "HEAD"], cwd=repo) - assert sha.strip() == sibling_sha.strip() - # Local adaptations were restored on top of that tip. - assert (test_dir / "runtest.sh").read_text() == "local-adapt\n" # ============================================================================= From 5cdc82d7013014ed316aa51fe3ae9f2417cae09f Mon Sep 17 00:00:00 2001 From: Tomas Korbar Date: Fri, 21 Aug 2026 13:30:17 +0200 Subject: [PATCH 3/5] Add per-package reproducer opt-out via rules ymir.yaml Allow maintainers to disable the reproducer workflow with a reproducer.enabled flag in the rules repo, mirroring MR consolidation's merge_mrs config. Co-authored-by: Cursor --- ymir/agents/reproducer_agent.py | 50 +++++++++++- ymir/agents/tasks.py | 60 +++++++++++++- .../tests/unit/test_reproducer_agent.py | 81 +++++++++++++++++++ ymir/agents/triage_agent.py | 23 +++++- ymir/common/models.py | 13 +++ 5 files changed, 222 insertions(+), 5 deletions(-) diff --git a/ymir/agents/reproducer_agent.py b/ymir/agents/reproducer_agent.py index f4d7bce55..d1b3ff9ed 100644 --- a/ymir/agents/reproducer_agent.py +++ b/ymir/agents/reproducer_agent.py @@ -21,6 +21,7 @@ from ymir.agents.constants import I_AM_YMIR, mr_description_footer from ymir.agents.observability import setup_observability from ymir.agents.reasoning_agent import ReasoningAgent +from ymir.agents.tasks import InvalidReproducerConfigError from ymir.agents.tf_cleanup_middleware import TFReservationCleanupMiddleware from ymir.agents.utils import ( build_agent_factory_with_mock_repos, @@ -791,6 +792,41 @@ def _build_commit_message(result: OutputSchema, input_data: InputSchema) -> str: ) +async def _reproducer_enabled_for_package( + package: str, + jira_issue: str, + gateway_tools: list, + *, + dry_run: bool, + user_triggered: bool, +) -> bool: + """Return False when reproducer is disabled or rules config is invalid.""" + try: + config = await tasks.fetch_reproducer_config(package, gateway_tools) + except InvalidReproducerConfigError as e: + logger.warning("Invalid reproducer config for %s: %s", package, e) + if not dry_run: + await tasks.comment_in_jira( + jira_issue=jira_issue, + agent_type="Reproducer", + comment_text=( + f"ymir.yaml for {package} has a malformed reproducer " + f"section: {e}\n\nReproducer analysis was skipped. Please fix " + f"the config file in the rules repository." + ), + is_error=True, + available_tools=gateway_tools, + user_triggered=user_triggered, + ) + return False + + if not config.enabled: + logger.info("Reproducer not enabled for %s, skipping", package) + return False + + return True + + async def run_workflow( jira_issue: str, dry_run: bool, @@ -815,12 +851,13 @@ async def run_workflow( working_dir.mkdir(parents=True, exist_ok=True) async with mcp_tools(os.getenv("MCP_GATEWAY_URL"), call_meta=call_meta) as gateway_tools: + agent_input = InputSchema(jira_issue=jira_issue) if input_data is None else input_data + tf_cleanup = TFReservationCleanupMiddleware() reproducer_agent = reproducer_agent_factory( gateway_tools, local_tool_options, extra_middlewares=[tf_cleanup] ) - agent_input = InputSchema(jira_issue=jira_issue) if input_data is None else input_data bootstrap: TestsCloneBootstrap | None = None if agent_input.package: bootstrap = await _bootstrap_tests_clone(working_dir, agent_input, gateway_tools) @@ -1235,6 +1272,17 @@ async def retry( ) return + call_meta = {"jira_issue": input_data.jira_issue, "package": input_data.package} + async with mcp_tools(os.getenv("MCP_GATEWAY_URL"), call_meta=call_meta) as gateway_tools: + if not await _reproducer_enabled_for_package( + input_data.package, + input_data.jira_issue, + gateway_tools, + dry_run=dry_run, + user_triggered=user_triggered, + ): + return + lock_id = await resolve_reproducer_lock_id( input_data.cve_id, input_data.jira_issue, diff --git a/ymir/agents/tasks.py b/ymir/agents/tasks.py index 6591a16ca..3c2ee9744 100644 --- a/ymir/agents/tasks.py +++ b/ymir/agents/tasks.py @@ -7,6 +7,7 @@ from pathlib import Path from urllib.parse import urlparse +import yaml from beeai_framework.tools import Tool from specfile import Specfile @@ -30,6 +31,7 @@ MergeRequestDetails, OpenMergeRequestResult, PackageConsolidationConfig, + PackageReproducerConfig, Task, ) from ymir.common.utils import get_all_sources, get_latest_candidate_build, get_latest_z_pending_build @@ -878,6 +880,10 @@ class InvalidConsolidationConfigError(Exception): """Raised when ymir.yaml exists but the consolidation section cannot be parsed.""" +class InvalidReproducerConfigError(Exception): + """Raised when ymir.yaml exists but the reproducer section cannot be parsed.""" + + async def fetch_consolidation_config( package: str, available_tools: list, @@ -900,8 +906,6 @@ async def fetch_consolidation_config( Returns: Parsed consolidation config. """ - import yaml - try: raw = await run_tool( "get_maintainer_rules", @@ -967,3 +971,55 @@ async def try_submit_consolidation_job( 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) + + +async def fetch_reproducer_config( + package: str, + available_tools: list, +) -> PackageReproducerConfig: + """Fetch the reproducer config from the per-package rules repo. + + Reads the ``reproducer`` section from ``ymir.yaml`` at + ``gitlab.com/redhat/centos-stream/rules/``. + Returns the default config (enabled) when the file is absent + or has no ``reproducer`` key. + + Raises: + InvalidReproducerConfigError: When the file exists but the + ``reproducer`` section does not conform to the expected schema. + + Args: + package: RPM package name. + available_tools: MCP gateway tools (must include ``get_maintainer_rules``). + + Returns: + Parsed reproducer config. + """ + try: + raw = await run_tool( + "get_maintainer_rules", + package=package, + file_path="ymir.yaml", + available_tools=available_tools, + ) + except Exception as e: + logger.warning("Failed to fetch ymir.yaml for %s: %s", package, e) + return PackageReproducerConfig() + + if "not found" in raw.lower(): + return PackageReproducerConfig() + + try: + data = yaml.safe_load(raw) + except yaml.YAMLError as e: + raise InvalidReproducerConfigError(f"ymir.yaml for {package} is not valid YAML: {e}") from e + + if not isinstance(data, dict) or "reproducer" not in data: + return PackageReproducerConfig() + + try: + return PackageReproducerConfig.model_validate(data["reproducer"]) + except Exception as e: + raise InvalidReproducerConfigError( + f"ymir.yaml reproducer section for {package} is malformed: {e}" + ) from e diff --git a/ymir/agents/tests/unit/test_reproducer_agent.py b/ymir/agents/tests/unit/test_reproducer_agent.py index 2cb48b457..f4484ffef 100644 --- a/ymir/agents/tests/unit/test_reproducer_agent.py +++ b/ymir/agents/tests/unit/test_reproducer_agent.py @@ -27,6 +27,7 @@ create_reproducer_agent, main, ) +from ymir.agents.tasks import InvalidReproducerConfigError, fetch_reproducer_config from ymir.common.base_utils import check_subprocess from ymir.common.constants import JiraLabels from ymir.common.models import MergeRequestDetails, ReproducerInputSchema, ReproducerOutputSchema, Task @@ -662,6 +663,23 @@ def _make_reproducer_payload(issue: str = "RHEL-99999", user_triggered: bool = F return task.model_dump_json().encode() +@contextlib.contextmanager +def _mock_reproducer_config_enabled(): + enabled_config = MagicMock(enabled=True) + + @contextlib.asynccontextmanager + async def fake_mcp_tools(*_args, **_kwargs): + yield [] + + with ( + patch( + "ymir.agents.tasks.fetch_reproducer_config", new_callable=AsyncMock, return_value=enabled_config + ), + patch("ymir.agents.reproducer_agent.mcp_tools", side_effect=fake_mcp_tools), + ): + yield + + @contextlib.contextmanager def _mock_workflow_lock(): with ( @@ -762,6 +780,7 @@ async def test_process_task_proceeds_despite_terminal_label_when_user_triggered( patch("ymir.agents.tasks.set_jira_labels", new_callable=AsyncMock), patch("ymir.agents.tasks.post_user_ack_once", new_callable=AsyncMock), patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + _mock_reproducer_config_enabled(), _mock_workflow_lock(), ): mock_workflow.return_value = MagicMock( @@ -785,6 +804,7 @@ async def test_process_task_proceeds_when_terminal_label_and_in_progress(): patch("ymir.agents.tasks.set_jira_labels", new_callable=AsyncMock), patch("ymir.agents.tasks.post_user_ack_once", new_callable=AsyncMock), patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + _mock_reproducer_config_enabled(), _mock_workflow_lock(), ): mock_workflow.return_value = MagicMock( @@ -807,6 +827,7 @@ async def test_process_task_proceeds_when_no_terminal_labels(): patch("ymir.agents.tasks.set_jira_labels", new_callable=AsyncMock), patch("ymir.agents.tasks.post_user_ack_once", new_callable=AsyncMock), patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + _mock_reproducer_config_enabled(), _mock_workflow_lock(), ): mock_workflow.return_value = MagicMock( @@ -843,8 +864,68 @@ async def test_process_task_blocks_when_workflow_lock_busy(): "ymir.agents.reproducer_agent.enqueue_blocked_reproducer_task", new_callable=AsyncMock, ) as mock_enqueue_blocked, + _mock_reproducer_config_enabled(), ): await _run_process_task(_make_reproducer_payload()) mock_workflow.assert_not_awaited() mock_enqueue_blocked.assert_awaited_once() + + +# -- fetch_reproducer_config --------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fetch_reproducer_config_returns_default_when_not_found(): + with patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run: + mock_run.return_value = "No maintainer rules found for package 'bind' (file 'ymir.yaml' not found)" + config = await fetch_reproducer_config("bind", []) + + assert config.enabled is True + + +@pytest.mark.asyncio +async def test_fetch_reproducer_config_parses_disabled(): + with patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run: + mock_run.return_value = "reproducer:\n enabled: false\n" + config = await fetch_reproducer_config("bind", []) + + assert config.enabled is False + + +@pytest.mark.asyncio +async def test_fetch_reproducer_config_raises_on_malformed_section(): + with patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run: + mock_run.return_value = "reproducer:\n enabled: not_a_bool\n" + with pytest.raises(InvalidReproducerConfigError, match="malformed"): + await fetch_reproducer_config("bind", []) + + +@pytest.mark.asyncio +async def test_process_task_skips_when_reproducer_disabled(): + disabled_config = MagicMock(enabled=False) + + @contextlib.asynccontextmanager + async def fake_mcp_tools(*_args, **_kwargs): + yield [] + + with ( + patch( + "ymir.agents.tasks.get_jira_issue_metadata", + new_callable=AsyncMock, + return_value=([], "New"), + ), + patch( + "ymir.agents.tasks.fetch_reproducer_config", new_callable=AsyncMock, return_value=disabled_config + ), + patch("ymir.agents.reproducer_agent.mcp_tools", side_effect=fake_mcp_tools), + patch("ymir.agents.reproducer_agent.run_workflow", new_callable=AsyncMock) as mock_workflow, + patch( + "ymir.agents.reproducer_agent.try_acquire_reproducer_lock", + new_callable=AsyncMock, + ) as mock_acquire_lock, + ): + await _run_process_task(_make_reproducer_payload()) + + mock_workflow.assert_not_awaited() + mock_acquire_lock.assert_not_awaited() diff --git a/ymir/agents/triage_agent.py b/ymir/agents/triage_agent.py index c9a1baa17..086acbb4a 100644 --- a/ymir/agents/triage_agent.py +++ b/ymir/agents/triage_agent.py @@ -28,6 +28,7 @@ queue_siblings_for_triage, ) from ymir.agents.rebuild_consolidation import find_rebuild_siblings +from ymir.agents.tasks import InvalidReproducerConfigError from ymir.agents.utils import ( build_agent_factory_with_mock_repos, get_agent_execution_config, @@ -173,7 +174,7 @@ def _build_reproducer_input(state) -> ReproducerInputSchema | None: ) -async def _enqueue_reproducer(redis, state, user_triggered: bool) -> None: +async def _enqueue_reproducer(redis, state, user_triggered: bool, gateway_tools) -> None: """Push a reproducer job when triage resolution is eligible.""" if state.triage_result is None: return @@ -186,6 +187,23 @@ async def _enqueue_reproducer(redis, state, user_triggered: bool) -> None: state.jira_issue, ) return + if reproducer_input.package: + try: + config = await tasks.fetch_reproducer_config(reproducer_input.package, gateway_tools) + except InvalidReproducerConfigError as e: + logger.warning( + "Invalid reproducer config for %s: %s; skipping enqueue", + reproducer_input.package, + e, + ) + return + if not config.enabled: + logger.info( + "Reproducer not enabled for %s, skipping enqueue for %s", + reproducer_input.package, + state.jira_issue, + ) + return queue = RedisQueues.get_reproducer_queue(user_triggered) task = Task(metadata=reproducer_input.model_dump(), user_triggered=user_triggered) await fix_await(redis.lpush(queue, task.model_dump_json())) @@ -1654,7 +1672,8 @@ async def retry(task, error, input=input, user_triggered=user_triggered): # Submit manually instead: # make trigger-reproducer JIRA_ISSUE=… PACKAGE=… # if auto_chain and output.resolution in _REPRODUCER_ELIGIBLE_RESOLUTIONS: - # await _enqueue_reproducer(redis, state, user_triggered) + # async with mcp_tools(os.environ["MCP_GATEWAY_URL"]) as gateway_tools: + # await _enqueue_reproducer(redis, state, user_triggered, gateway_tools) # elif not auto_chain and output.resolution in _REPRODUCER_ELIGIBLE_RESOLUTIONS: # logger.info( # "AUTO_CHAIN disabled, skipping reproducer queue for %s", diff --git a/ymir/common/models.py b/ymir/common/models.py index 08106e313..0b55cd1c1 100644 --- a/ymir/common/models.py +++ b/ymir/common/models.py @@ -816,6 +816,19 @@ class PackageConsolidationConfig(BaseModel): ) +class PackageReproducerConfig(BaseModel): + """Machine-readable reproducer config from the per-package rules repo. + + Parsed from the ``reproducer`` section of + ``gitlab.com/redhat/centos-stream/rules//ymir.yaml``. + """ + + enabled: bool = Field( + default=True, + description="Whether to run the Ymir reproducer workflow for this package", + ) + + class MRConsolidationInputSchema(BaseModel): """Input schema for the MR consolidation agent.""" From aa0fd214da3e718c367cb273fadc5c64666cac18 Mon Sep 17 00:00:00 2001 From: Tomas Korbar Date: Fri, 21 Aug 2026 13:46:56 +0200 Subject: [PATCH 4/5] Assign package QE as reviewer on new reproducer test MRs Resolve QA Contact from bugzilla component data and set GitLab reviewers on new tests-repo MRs, mirroring backport MR reviewer assignment. Co-authored-by: Cursor --- ymir/agents/reproducer_agent.py | 30 +++++++++++++- ymir/agents/tasks.py | 30 ++++++++++++++ .../tests/unit/test_reproducer_agent.py | 16 ++++++++ ymir/agents/tests/unit/test_tasks.py | 27 +++++++++++++ ymir/tools/privileged/gateway.py | 2 + ymir/tools/privileged/gitlab.py | 26 ++++++++++++ ymir/tools/privileged/reviewer_resolver.py | 39 ++++++++++++++++-- .../privileged/tests/unit/test_gitlab.py | 12 ++++++ .../tests/unit/test_reviewer_resolver.py | 40 +++++++++++++++++++ 9 files changed, 218 insertions(+), 4 deletions(-) diff --git a/ymir/agents/reproducer_agent.py b/ymir/agents/reproducer_agent.py index d1b3ff9ed..106cd1242 100644 --- a/ymir/agents/reproducer_agent.py +++ b/ymir/agents/reproducer_agent.py @@ -60,6 +60,7 @@ sweep_stale_reproducer_locks, try_acquire_reproducer_lock, ) +from ymir.common.version_utils import construct_internal_branch_name, parse_rhel_version from ymir.tools.privileged.jira import fetch_jira_issue_issuelinks from ymir.tools.unprivileged.commands import RunShellCommandTool from ymir.tools.unprivileged.text import CreateTool, SearchTextTool, ViewTool @@ -792,6 +793,18 @@ def _build_commit_message(result: OutputSchema, input_data: InputSchema) -> str: ) +def _reviewer_lookup_branch(input_data: InputSchema) -> str | None: + """Map reproducer task metadata to a dist-git branch for reviewer lookup.""" + if input_data.target_branch: + return input_data.target_branch + if input_data.fix_version: + parsed = parse_rhel_version(input_data.fix_version) + if parsed: + major, minor, _ = parsed + return construct_internal_branch_name(major, minor) + return None + + async def _reproducer_enabled_for_package( package: str, jira_issue: str, @@ -989,7 +1002,7 @@ async def create_merge_request(state): mr_description = _build_mr_description(result, agent_input) commit_message = _build_commit_message(result, agent_input) - mr_url, _ = await tasks.commit_push_and_open_mr( + mr_url, is_new_mr = await tasks.commit_push_and_open_mr( local_clone=tests_clone, commit_message=commit_message, fork_url=fork_url, @@ -1003,6 +1016,21 @@ async def create_merge_request(state): result.test_mr_url = mr_url if mr_url: logger.info(f"Created/updated reproducer MR: {mr_url}") + if is_new_mr: + reviewer_branch = _reviewer_lookup_branch(agent_input) + if reviewer_branch: + await tasks.request_mr_qe_reviews( + package, + reviewer_branch, + mr_url, + gateway_tools, + ) + else: + logger.info( + "Skipping QE reviewer assignment for %s — " + "no target_branch or fix_version in reproducer input", + state.jira_issue, + ) if result.adapted_existing: result.existing_mr_url = result.existing_mr_url or mr_url else: diff --git a/ymir/agents/tasks.py b/ymir/agents/tasks.py index 3c2ee9744..2ed1825f5 100644 --- a/ymir/agents/tasks.py +++ b/ymir/agents/tasks.py @@ -447,6 +447,36 @@ async def request_mr_reviews( logger.warning("Failed to assign reviewers to MR %s: %s", mr_url, e) +async def request_mr_qe_reviews( + package: str, + dist_git_branch: str, + mr_url: str, + available_tools: list[Tool], +) -> None: + """Best-effort QE reviewer assignment — logs warnings but never raises.""" + if os.getenv("ASSIGN_MR_REVIEWERS", "false").lower() != "true": + return + try: + reviewer_ids = await run_tool( + "resolve_qe_reviewers", + package=package, + dist_git_branch=dist_git_branch, + available_tools=available_tools, + ) + if not reviewer_ids: + logger.info("No QE reviewers resolved for %s (%s)", package, dist_git_branch) + return + await run_tool( + "set_merge_request_reviewers", + merge_request_url=mr_url, + reviewer_ids=reviewer_ids, + available_tools=available_tools, + ) + logger.info("Assigned QE reviewers %s to MR %s", reviewer_ids, mr_url) + except Exception as e: + logger.warning("Failed to assign QE reviewers to MR %s: %s", mr_url, e) + + async def commit_push_and_open_mr( local_clone: Path, commit_message: str, diff --git a/ymir/agents/tests/unit/test_reproducer_agent.py b/ymir/agents/tests/unit/test_reproducer_agent.py index f4484ffef..8c74bcff8 100644 --- a/ymir/agents/tests/unit/test_reproducer_agent.py +++ b/ymir/agents/tests/unit/test_reproducer_agent.py @@ -23,6 +23,7 @@ _reproducer_mr_title_tags, _resolve_reproducer_mr_target, _resolve_test_dir, + _reviewer_lookup_branch, _should_finalize_jira, create_reproducer_agent, main, @@ -81,6 +82,21 @@ def test_adapted_existing_uses_created_label(): assert _determine_comment_resolution(result) == "adapted-existing" +@pytest.mark.parametrize( + ("input_data", "expected"), + [ + (ReproducerInputSchema(jira_issue="RHEL-1", package="bind", target_branch="c10s"), "c10s"), + ( + ReproducerInputSchema(jira_issue="RHEL-1", package="bind", fix_version="rhel-9.8"), + "rhel-9.8.0", + ), + (ReproducerInputSchema(jira_issue="RHEL-1", package="bind"), None), + ], +) +def test_reviewer_lookup_branch(input_data, expected): + assert _reviewer_lookup_branch(input_data) == expected + + def test_should_finalize_jira_false_for_retryable_error(): assert _should_finalize_jira(_output(success=False, retryable_error=True)) is False assert _should_finalize_jira(_output(success=False, lock_deferred=True)) is False diff --git a/ymir/agents/tests/unit/test_tasks.py b/ymir/agents/tests/unit/test_tasks.py index a825f0561..a5973b058 100644 --- a/ymir/agents/tests/unit/test_tasks.py +++ b/ymir/agents/tests/unit/test_tasks.py @@ -13,6 +13,7 @@ handle_zstream_branch_stale_error, needs_zstream_target_label, post_user_ack_once, + request_mr_qe_reviews, ) from ymir.common.constants import JiraLabels, RedisQueues from ymir.common.models import Task @@ -371,6 +372,32 @@ async def mock_run_tool(name, *, available_tools=None, **kwargs): assert reviewer_calls[0][1]["reviewer_ids"] == [42, 99] +@pytest.mark.asyncio +async def test_request_mr_qe_reviews_assigns_qe_only(tmp_path, monkeypatch): + monkeypatch.setenv("ASSIGN_MR_REVIEWERS", "true") + tool_calls = [] + + async def mock_run_tool(name, *, available_tools=None, **kwargs): + tool_calls.append((name, kwargs)) + if name == "resolve_qe_reviewers": + return [99] + return None + + with patch("ymir.agents.tasks.run_tool", side_effect=mock_run_tool): + await request_mr_qe_reviews( + "bind", + "c10s", + "https://gitlab.com/redhat/rhel/tests/bind/-/merge_requests/1", + [], + ) + + assert tool_calls[0][0] == "resolve_qe_reviewers" + assert tool_calls[0][1] == {"package": "bind", "dist_git_branch": "c10s"} + reviewer_calls = [(n, kw) for n, kw in tool_calls if n == "set_merge_request_reviewers"] + assert len(reviewer_calls) == 1 + assert reviewer_calls[0][1]["reviewer_ids"] == [99] + + @pytest.mark.asyncio async def test_commit_push_and_open_mr_reviewer_failure_does_not_fail(tmp_path, monkeypatch): monkeypatch.setenv("ASSIGN_MR_REVIEWERS", "true") diff --git a/ymir/tools/privileged/gateway.py b/ymir/tools/privileged/gateway.py index 75c36846d..59e36ffba 100644 --- a/ymir/tools/privileged/gateway.py +++ b/ymir/tools/privileged/gateway.py @@ -40,6 +40,7 @@ ListProjectMergeRequestsTool, OpenMergeRequestTool, PushToRemoteRepositoryTool, + ResolveQeReviewersTool, ResolveReviewersTool, RetryPipelineJobTool, SearchGitlabProjectMrsTool, @@ -129,6 +130,7 @@ async def _async_main(): FetchGitlabMrNotesTool(options=tool_options), SearchGitlabProjectMrsTool(options=tool_options), ResolveReviewersTool(options=tool_options), + ResolveQeReviewersTool(options=tool_options), SetMergeRequestReviewersTool(options=tool_options), GetErratumTool(options=tool_options), GetErratumBuildNvrTool(options=tool_options), diff --git a/ymir/tools/privileged/gitlab.py b/ymir/tools/privileged/gitlab.py index 4b74b97cb..f6636cb1d 100644 --- a/ymir/tools/privileged/gitlab.py +++ b/ymir/tools/privileged/gitlab.py @@ -879,6 +879,32 @@ async def _run( return JSONToolOutput(result=reviewer_ids) +class ResolveQeReviewersTool(Tool[ResolveReviewersToolInput, ToolRunOptions, JSONToolOutput[list[int]]]): + name = "resolve_qe_reviewers" + timeout = 120 + description = """ + Resolve QE reviewer GitLab user IDs for a package from the bugzilla QA Contact. + """ + input_schema = ResolveReviewersToolInput + + def _create_emitter(self) -> Emitter: + return Emitter.root().child( + namespace=["tool", "gitlab", self.name], + creator=self, + ) + + async def _run( + self, + tool_input: ResolveReviewersToolInput, + options: ToolRunOptions | None, + context: RunContext, + ) -> JSONToolOutput[list[int]]: + from ymir.tools.privileged.reviewer_resolver import resolve_qe_reviewers + + reviewer_ids = await resolve_qe_reviewers(tool_input.package, tool_input.dist_git_branch) + return JSONToolOutput(result=reviewer_ids) + + class AddMergeRequestCommentToolInput(BaseModel): merge_request_url: str = Field(description="URL of the merge request") comment: str = Field(description="Comment text") diff --git a/ymir/tools/privileged/reviewer_resolver.py b/ymir/tools/privileged/reviewer_resolver.py index 50b387d1e..29104916f 100644 --- a/ymir/tools/privileged/reviewer_resolver.py +++ b/ymir/tools/privileged/reviewer_resolver.py @@ -257,8 +257,41 @@ async def _lookup_gitlab_user_by_username( async def resolve_reviewers(package: str, dist_git_branch: str) -> list[int]: """Resolve reviewer GitLab user IDs for a package on a given branch. + Includes both the default assignee (maintainer) and QA contact. + + Returns a (possibly empty) list of user IDs. Never raises. + """ + return await _resolve_component_reviewers( + package, + dist_git_branch, + include_assignee=True, + include_qa=True, + ) + + +async def resolve_qe_reviewers(package: str, dist_git_branch: str) -> list[int]: + """Resolve QE reviewer GitLab user IDs for a package on a given branch. + + Uses only the bugzilla component ``QA Contact`` field. + Returns a (possibly empty) list of user IDs. Never raises. """ + return await _resolve_component_reviewers( + package, + dist_git_branch, + include_assignee=False, + include_qa=True, + ) + + +async def _resolve_component_reviewers( + package: str, + dist_git_branch: str, + *, + include_assignee: bool, + include_qa: bool, +) -> list[int]: + """Resolve GitLab reviewer IDs from bugzilla component contacts.""" try: parsed = parse_branch_name(dist_git_branch) if not parsed: @@ -271,13 +304,13 @@ async def resolve_reviewers(package: str, dist_git_branch: str) -> list[int]: return [] emails: list[str] = [] - if assignee := component_data.get("Default Assignee"): + if include_assignee and (assignee := component_data.get("Default Assignee")): emails.append(assignee) - if (qa_contact := component_data.get("QA Contact")) and qa_contact not in emails: + if include_qa and (qa_contact := component_data.get("QA Contact")) and qa_contact not in emails: emails.append(qa_contact) if not emails: - logger.info("No assignee or QA contact for %s (RHEL%s)", package, rhel_major) + logger.info("No matching component contacts for %s (RHEL%s)", package, rhel_major) return [] reviewer_ids: list[int] = [] diff --git a/ymir/tools/privileged/tests/unit/test_gitlab.py b/ymir/tools/privileged/tests/unit/test_gitlab.py index 31e673dc7..7b9c46a35 100644 --- a/ymir/tools/privileged/tests/unit/test_gitlab.py +++ b/ymir/tools/privileged/tests/unit/test_gitlab.py @@ -23,6 +23,7 @@ GetFailedPipelineJobsFromMergeRequestTool, OpenMergeRequestTool, PushToRemoteRepositoryTool, + ResolveQeReviewersTool, ResolveReviewersTool, RetryPipelineJobTool, SetMergeRequestReviewersTool, @@ -1315,3 +1316,14 @@ async def test_resolve_reviewers_tool(): ): result = await ResolveReviewersTool().run(input={"package": "bash", "dist_git_branch": "c10s"}) assert result.result == [42, 99] + + +@pytest.mark.asyncio +async def test_resolve_qe_reviewers_tool(): + with patch( + "ymir.tools.privileged.reviewer_resolver.resolve_qe_reviewers", + new_callable=AsyncMock, + return_value=[99], + ): + result = await ResolveQeReviewersTool().run(input={"package": "bash", "dist_git_branch": "c10s"}) + assert result.result == [99] diff --git a/ymir/tools/privileged/tests/unit/test_reviewer_resolver.py b/ymir/tools/privileged/tests/unit/test_reviewer_resolver.py index 0c7ce845e..df376b5f6 100644 --- a/ymir/tools/privileged/tests/unit/test_reviewer_resolver.py +++ b/ymir/tools/privileged/tests/unit/test_reviewer_resolver.py @@ -11,6 +11,7 @@ fetch_bugzilla_component_data, parse_component_file, resolve_gitlab_user_id, + resolve_qe_reviewers, resolve_reviewers, ) @@ -208,6 +209,45 @@ async def json(): assert sorted(result) == [42, 99] +@pytest.mark.asyncio +async def test_resolve_qe_reviewers_only_uses_qa_contact(monkeypatch): + monkeypatch.setenv("GITLAB_TOKEN", "test-token") + + @asynccontextmanager + async def get(url, headers=None, params=None): + if "gitlab.cee.redhat.com" in url: + + async def text(): + return SAMPLE_COMPONENT_FILE + + yield flexmock(status=200, text=text) + elif params: + search = params["search"] + if search == "qaengineer@redhat.com": + + async def json(): + return [{"id": 99, "username": "qaengineer"}] + + yield flexmock(status=200, json=json) + else: + + async def json(): + return [] + + yield flexmock(status=200, json=json) + else: + + async def json(): + return {"id": 99, "public_email": "qaengineer@redhat.com"} + + yield flexmock(status=200, json=json) + + flexmock(aiohttp.ClientSession).should_receive("get").replace_with(get) + + result = await resolve_qe_reviewers("bash", "c10s") + assert result == [99] + + @pytest.mark.asyncio async def test_resolve_reviewers_partial_failure(monkeypatch): monkeypatch.setenv("GITLAB_TOKEN", "test-token") From 39cd67ef49749eba37e692472008dd786aefb929 Mon Sep 17 00:00:00 2001 From: Tomas Korbar Date: Fri, 21 Aug 2026 15:50:34 +0200 Subject: [PATCH 5/5] Re-enable triage reproducer auto-enqueue behind TRIAGE_ENQUEUE_REPRODUCER. Restore post-triage reproducer job submission as an opt-in feature so deployments can enable it explicitly while local and prod defaults stay off. Co-authored-by: Cursor --- compose.yaml | 1 + openshift/configmap-agents-env.yml | 3 +++ ymir/agents/triage_agent.py | 32 +++++++++++++----------------- 3 files changed, 18 insertions(+), 18 deletions(-) diff --git a/compose.yaml b/compose.yaml index 8dbc2d003..3682ed608 100644 --- a/compose.yaml +++ b/compose.yaml @@ -14,6 +14,7 @@ x-beeai-env: &beeai-env JIRA_ALLOW_STATUS_CHANGES: ${JIRA_ALLOW_STATUS_CHANGES:-false} ERRATA_ALLOW_STATUS_CHANGES: ${ERRATA_ALLOW_STATUS_CHANGES:-false} AUTO_CHAIN: ${AUTO_CHAIN:-true} + TRIAGE_ENQUEUE_REPRODUCER: ${TRIAGE_ENQUEUE_REPRODUCER:-false} REQUESTS_CA_BUNDLE: /etc/pki/tls/certs/ca-bundle.crt SENTRY_ENVIRONMENT: ${SENTRY_ENVIRONMENT:-development} diff --git a/openshift/configmap-agents-env.yml b/openshift/configmap-agents-env.yml index 6d9722f4b..7a446a975 100644 --- a/openshift/configmap-agents-env.yml +++ b/openshift/configmap-agents-env.yml @@ -21,6 +21,9 @@ data: # When "true", agents assign reviewers to newly created MRs based on # the package's bugzilla component contacts (Default Assignee + QA Contact). ASSIGN_MR_REVIEWERS: "true" + # When "true", triage enqueues reproducer jobs for eligible resolutions. + # Unset or "false" disables auto-enqueue (manual: make trigger-reproducer). + TRIAGE_ENQUEUE_REPRODUCER: "false" immutable: false kind: ConfigMap metadata: diff --git a/ymir/agents/triage_agent.py b/ymir/agents/triage_agent.py index 086acbb4a..6ad5fe2d3 100644 --- a/ymir/agents/triage_agent.py +++ b/ymir/agents/triage_agent.py @@ -1154,6 +1154,7 @@ async def main() -> None: dry_run = os.getenv("DRY_RUN", "False").lower() == "true" auto_chain = os.getenv("AUTO_CHAIN", "true").lower() == "true" + enqueue_reproducer = os.getenv("TRIAGE_ENQUEUE_REPRODUCER", "false").lower() == "true" force_cve_triage = os.getenv("FORCE_CVE_TRIAGE", "false").lower() == "true" if jira_issue := os.getenv("JIRA_ISSUE", None): @@ -1253,7 +1254,11 @@ async def main() -> None: return - logger.info(f"Starting triage agent in queue mode (AUTO_CHAIN={'enabled' if auto_chain else 'disabled'})") + logger.info( + "Starting triage agent in queue mode (AUTO_CHAIN=%s, TRIAGE_ENQUEUE_REPRODUCER=%s)", + "enabled" if auto_chain else "disabled", + "enabled" if enqueue_reproducer else "disabled", + ) max_concurrent_tasks = int(os.getenv("MAX_CONCURRENT_TASKS", 1)) async with redis_client(os.environ["REDIS_URL"]) as redis: max_retries = int(os.getenv("MAX_RETRIES", 3)) @@ -1668,24 +1673,15 @@ async def retry(task, error, input=input, user_triggered=user_triggered): else: logger.info(f"AUTO_CHAIN disabled, skipping downstream queue for {input.issue}") - # Auto-enqueue of reproducer jobs is temporarily disabled. - # Submit manually instead: - # make trigger-reproducer JIRA_ISSUE=… PACKAGE=… - # if auto_chain and output.resolution in _REPRODUCER_ELIGIBLE_RESOLUTIONS: - # async with mcp_tools(os.environ["MCP_GATEWAY_URL"]) as gateway_tools: - # await _enqueue_reproducer(redis, state, user_triggered, gateway_tools) - # elif not auto_chain and output.resolution in _REPRODUCER_ELIGIBLE_RESOLUTIONS: - # logger.info( - # "AUTO_CHAIN disabled, skipping reproducer queue for %s", - # input.issue, - # ) if output.resolution in _REPRODUCER_ELIGIBLE_RESOLUTIONS: - logger.info( - "Skipping auto-enqueue of reproducer for %s " - "(manual trigger: make trigger-reproducer JIRA_ISSUE=%s PACKAGE=…)", - input.issue, - input.issue, - ) + if enqueue_reproducer: + async with mcp_tools(os.environ["MCP_GATEWAY_URL"]) as gateway_tools: + await _enqueue_reproducer(redis, state, user_triggered, gateway_tools) + else: + logger.info( + "TRIAGE_ENQUEUE_REPRODUCER disabled, skipping reproducer queue for %s", + input.issue, + ) shutdown_event = asyncio.Event() install_shutdown_handler(asyncio.get_running_loop(), shutdown_event)