diff --git a/Makefile b/Makefile index 3f3b9c94e..5b4f85137 100644 --- a/Makefile +++ b/Makefile @@ -339,7 +339,7 @@ trigger-pipeline: @echo "Triggering pipeline for issue: $(JIRA_ISSUE) (force_cve_triage=$(FORCE_CVE_TRIAGE))" $(COMPOSE_AGENTS) exec valkey redis-cli LPUSH triage_queue '{"metadata": {"issue": "$(JIRA_ISSUE)", "force_cve_triage": $(FORCE_CVE_TRIAGE)}}' -# Manually enqueue a reproducer job (triage auto-enqueue is currently disabled). +# Manually enqueue a reproducer job (also when TRIAGE_ENQUEUE_REPRODUCER=false). # Required: JIRA_ISSUE, PACKAGE # Optional: CVE_ID, FIX_VERSION, TARGET_BRANCH, TRIAGE_SUMMARY, USER_TRIGGERED=true .PHONY: trigger-reproducer diff --git a/README-agents.md b/README-agents.md index dd8525b6e..fafe7cf5a 100644 --- a/README-agents.md +++ b/README-agents.md @@ -117,7 +117,7 @@ make trigger-pipeline JIRA_ISSUE=RHEL-12345 # Force triage of Y-stream CVEs (normally skipped) make trigger-pipeline JIRA_ISSUE=RHEL-12345 FORCE_CVE_TRIAGE=true -# Manually enqueue a reproducer job (triage auto-enqueue is currently disabled) +# Manually enqueue a reproducer job (also used when TRIAGE_ENQUEUE_REPRODUCER=false) make trigger-reproducer JIRA_ISSUE=RHEL-12345 PACKAGE=bind make trigger-reproducer JIRA_ISSUE=RHEL-12345 PACKAGE=bind CVE_ID=CVE-2025-12345 FIX_VERSION=rhel-10.1 ``` diff --git a/compose.yaml b/compose.yaml index 8dbc2d003..3e57ab15f 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:-true} REQUESTS_CA_BUNDLE: /etc/pki/tls/certs/ca-bundle.crt SENTRY_ENVIRONMENT: ${SENTRY_ENVIRONMENT:-development} diff --git a/docs/reproducer_architecture.md b/docs/reproducer_architecture.md index f934141e8..4371f10e8 100644 --- a/docs/reproducer_architecture.md +++ b/docs/reproducer_architecture.md @@ -19,7 +19,8 @@ stream creates waste and conflict: The Reproducer Agent automates this: after triage, it designs or reuses a tmt/BeakerLib test, verifies it on Testing Farm for the issue’s stream, opens or updates a single tests-repo MR labeled `ymir_reproducer`, and uses a Redis -lock so sibling-stream workers serialize create/adapt work. +lock so sibling-stream workers serialize the full reproducer run (analysis, +TF verification, and MR push). ## High-Level Architecture @@ -28,7 +29,7 @@ lock so sibling-stream workers serialize create/adapt work. │ │ ─────────────────────────▶│ rebase / backport │ │ Triage │ │ / rebuild queues │ │ Agent │ └────────────────────┘ -│ │ AUTO_CHAIN (parallel) +│ │ TRIAGE_ENQUEUE_REPRODUCER (parallel) │ │ ─────────────────────────▶┌────────────────────┐ │ │ rebase|backport|rebuild │ reproducer_queue │ │ │ |not-affected │ (+ _todo twin) │ @@ -38,13 +39,24 @@ lock so sibling-stream workers serialize create/adapt work. │ ▼ ┌────────────────────┐ - │ Reproducer Agent │ + │ Queue orchestration │ + │ (per task) │ + │ • package gate │ + │ • acquire lock │ + │ • in_progress label │ + └─────────┬──────────┘ + │ lock held + ▼ + ┌────────────────────┐ + │ Reproducer workflow │ │ │ + │ 0. Bootstrap tests│ + │ clone + MR tip │ │ 1. LLM analysis │ - │ (TF verify / │ + │ (TF verify / │ │ reuse / adapt)│ │ 2. create/update │ - │ MR (under lock)│ + │ MR │ │ 3. Jira labels + │ │ comment │ └────────────────────┘ @@ -53,43 +65,68 @@ lock so sibling-stream workers serialize create/adapt work. ▼ ▼ ▼ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ tests repo MR │ │ Testing Farm │ │ Redis lock hash │ - │ ymir_reproducer │ │ reserve / run / │ │ package:lock_id │ - │ │ │ cancel │ │ :active │ + │ ymir_reproducer │ │ reserve / run / │ │ + blocked lists │ + │ │ │ cancel │ │ per package:lock │ └──────────────────┘ └──────────────────┘ └──────────────────┘ ``` -Triage still sends fix work to rebase/backport/rebuild as before. The -reproducer job is an **additional** parallel enqueue for eligible resolutions. +Triage still sends fix work to rebase/backport/rebuild when `AUTO_CHAIN=true`. +The reproducer job is an **additional** parallel enqueue when triage +auto-enqueue is enabled (see [Triggering](#triggering)). + +## Package Enablement + +Before enqueue (triage) or before running (queue worker), orchestration reads +the `reproducer` section of `ymir.yaml` from +`gitlab.com/redhat/centos-stream/rules/` via `fetch_reproducer_config`. + +| Config | Behavior | +|--------|----------| +| File missing or no `reproducer` key | Default: **disabled** | +| `reproducer.enabled: true` | Proceed | +| `reproducer.enabled: false` | Skip silently (no workflow, no terminal reproducer label) | +| Malformed `reproducer` section | Skip; at queue time post a Jira error comment asking maintainers to fix `ymir.yaml` | + +Triage enqueue also skips when config is disabled or invalid. Manual +`make trigger-reproducer` bypasses triage enqueue but the queue worker still +checks config before acquiring the lock. ## Triggering -### Auto-chain from triage +### Auto-enqueue from triage -> **Temporarily disabled.** Triage no longer LPUSHes to `reproducer_queue`. -> Submit jobs manually with `make trigger-reproducer` (see -> [Manual queue submit](#manual-queue-submit) below). The enqueue helpers -> (`_build_reproducer_input`, `_enqueue_reproducer`) remain in place for when -> auto-chain is re-enabled. +> **Enabled by default** (`TRIAGE_ENQUEUE_REPRODUCER=true` in +> `openshift/configmap-agents-env.yml` and local `compose.yaml`). Set +> `TRIAGE_ENQUEUE_REPRODUCER=false` to disable triage LPUSH and submit jobs +> manually with `make trigger-reproducer` (see +> [Manual queue submit](#manual-queue-submit) below). -When `AUTO_CHAIN=true` (default) and triage resolves as one of: +`TRIAGE_ENQUEUE_REPRODUCER` is **independent** of `AUTO_CHAIN`. Fix-agent +enqueue uses `AUTO_CHAIN`; reproducer enqueue uses its own flag. -| Resolution | Fix agent queued? | Reproducer queued? | -|------------------|-------------------|--------------------| -| `rebase` | Yes | Yes | -| `backport` | Yes | Yes | -| `rebuild` | Yes | Yes | -| `not-affected` | No | Yes | -| `postponed` | No (postponed list) | No | +When `TRIAGE_ENQUEUE_REPRODUCER` is true and triage resolves as one of: + +| Resolution | Fix agent queued? (`AUTO_CHAIN`) | Reproducer eligible? | +|------------------|----------------------------------|----------------------| +| `rebase` | Yes (if `AUTO_CHAIN`) | Yes | +| `backport` | Yes (if `AUTO_CHAIN`) | Yes | +| `rebuild` | Yes (if `AUTO_CHAIN`) | Yes | +| `not-affected` | No | Yes | +| `postponed` | No (postponed list) | No | | `clarification-needed` / `open-ended-analysis` / `error` | Special | No | -Triage builds a flat `ReproducerInputSchema` (not full `TriageState`) via -`_build_reproducer_input` and `lpush`es a `Task` to `reproducer_queue` or -`reproducer_queue_todo` when `user_triggered`. +Additional gates before LPUSH: + +1. `_build_reproducer_input` must resolve `package` on the resolution data. +2. `fetch_reproducer_config` must return `enabled: true`. + +Triage builds a flat `ReproducerInputSchema` (not full `TriageState`) and +`lpush`es a `Task` to `reproducer_queue` or `reproducer_queue_todo` when +`user_triggered`. -Required for enqueue: `package` on the resolution data. For `not-affected`, -applicability overrides copy `package` / `cve_id` / `fix_version` / -`triage_summary` / `patch_urls` from the prior resolution, and the N/A -explanation is folded into `triage_summary` for the reproducer prompt. +For `not-affected`, applicability overrides copy `package` / `cve_id` / +`fix_version` / `triage_summary` / `patch_urls` from the prior resolution, and +the N/A explanation is folded into `triage_summary` for the reproducer prompt. ### Standalone / queue / test modes @@ -102,28 +139,62 @@ explanation is folded into `triage_summary` for the reproducer prompt. For ad-hoc queue testing, you can also manually Redis-`lpush` a `Task` whose `metadata` validates as `ReproducerInputSchema`. -## Redis Queues and Delayed Retries +Standalone mode runs `run_workflow` once with no Redis: no queue lock, no +duplicate guard, and no package gate unless `PACKAGE` is passed via queue +metadata (standalone with only `JIRA_ISSUE` leaves `package` unset unless the +agent resolves it from Jira). + +## Redis Queues and Retries | Key | Type | Role | |-----|------|------| | `reproducer_queue` | List | Normal reproducer tasks | | `reproducer_queue_todo` | List | Priority twin (`ymir_todo` / user-triggered) | | `reproducer_queue_delayed` | ZSET | Deferred retries (score = unix ready-time) | +| `reproducer_blocked:{package}:{lock_id}` | List | Tasks waiting for the create/adapt lock | | `completed_reproducer_list` | List | Finished non-retry outcomes | +| `reproducer_creation_lock` | Hash | Active lock entries (`{package}:{lock_id}:active`) | The worker BRPOPs `[reproducer_queue_todo, reproducer_queue]`. Each poll cycle also: -1. `sweep_stale_reproducer_locks()` — clear abandoned create/adapt locks +1. `sweep_stale_reproducer_locks()` — clear abandoned locks (default 6h) and + promote any tasks on matching blocked lists 2. `promote_due_tasks()` — move ready delayed payloads back onto the appropriate list (`_todo` if `user_triggered`) -Delayed retries are used for: +**Delayed ZSET retries** (`reproducer_queue_delayed`) are used only for +**retryable infra** (`retryable_error`, e.g. Testing Farm provisioning). Default +delay: `REPRODUCER_RETRY_DELAY_SECONDS` (1800s). + +**Lock contention** does **not** use the delayed ZSET. A worker that cannot +acquire the lock parks the task on `reproducer_blocked:{package}:{lock_id}` via +`enqueue_blocked_reproducer_task`. When the holder releases the lock (or stale +sweep removes it), `promote_blocked_reproducer_tasks` LPUSHes blocked payloads +back to `reproducer_queue` or `reproducer_queue_todo` (preserving +`user_triggered`). The blocked waiter still receives +`ymir_reproducer_in_progress` and a user ack when `user_triggered`. -- **Retryable infra** (`retryable_error`, e.g. Testing Farm provisioning) -- **Lock contention** (`lock_deferred` — another worker holds create/adapt) +After `MAX_RETRIES` failed queue attempts (uncaught exception or missing package), +the worker sets `ymir_reproducer_errored` and pushes to `error_list`. -Default delay: `REPRODUCER_RETRY_DELAY_SECONDS` (1800s). +The legacy output flag `lock_deferred` is no longer set by the agent; +orchestration blocks at workflow start instead. + +## Working Directory Layout + +Each run uses a per-issue directory under `GIT_REPO_BASEPATH` (default +`/git-repos`): + +``` +/git-repos/Reproducer//tests-/ +``` + +`run_workflow` removes and recreates `Reproducer//` at the start of +each run. Orchestration bootstraps the tests clone at +`tests-` before the LLM runs (see below). SCP allowlisting permits paths +under `/git-repos` and `/tmp`; remote tools scope copies to the job's Reproducer +tree via MCP metadata (`jira_issue`, `package`). ## Cross-Stream Reuse and Adapt @@ -133,38 +204,78 @@ CVE tests are often under `Security//` and bug tests under agent MUST return that path as `test_directory` in its output; orchestration never invents or guesses the location. +### Orchestration bootstrap + +Before the LLM runs (when `package` is known), orchestration: + +1. Clones `https://gitlab.com/redhat/rhel/tests/` into the working + directory. +2. Lists open MRs with label `ymir_reproducer`. +3. Matches an open MR for this CVE or Jira issue (see [MR matching](#mr-matching) below). +4. If matched: fetches and checks out the MR source branch; discovers + `existing_test_directory` on that branch. +5. Passes bootstrap context into the prompt (`tests_clone_ready`, + `tests_clone_path`, `existing_mr_url`, `mr_source_branch`, + `existing_test_directory`) so the agent adapts in place and does not + re-clone over the checked-out branch. + ### Agent behavior (prompt) -1. Clone `https://gitlab.com/redhat/rhel/tests/` (default branch). -2. Look for an existing test directory and grep for issue/CVE references. -3. List open MRs with label `ymir_reproducer` via - `list_project_merge_requests`. -4. **If found:** reserve Testing Farm for **this** stream’s compose and run - the existing test. +When bootstrap did not run, the agent clones via `clone_repository` and follows +the prompt workflow: + +1. Look for an existing test directory and grep for issue/CVE references. +2. List open MRs with label `ymir_reproducer` via `list_project_merge_requests`. +3. **If found:** reserve Testing Farm for **this** stream’s compose and run the + existing test. - Works → `success=true`, `test_already_exists=true`, `adapted_existing=false` (no new MR); still set `test_directory`. - - Fails on this stream → adapt the test to be portable across streams, - re-verify, set `adapted_existing=true`, `existing_mr_url`, and - `test_directory` to the adapted path. -5. **If not found:** create a new test and set `test_directory` to its - relative path under the clone. + - Fails on this stream → adapt the test in the **same directory path** on the + open MR branch, re-verify, set `adapted_existing=true`, `existing_mr_url`, + and `test_directory`. +4. **If not found:** create a new test and set `test_directory` to its relative + path under the clone. + +The agent does not set `test_mr_url` or `lock_deferred`; orchestration owns MR +URLs and lock scheduling. + +### MR matching + +Open reproducer MRs are matched by **MR title only** using canonical bracket +tags (descriptions are ignored): + +| Reproducer type | Title pattern | Cross-stream behavior | +|-----------------|---------------|------------------------| +| CVE | `package: [CVE-…] ymir reproducer test` | One MR per CVE set (stable across streams) | +| Regression (bug) | `package: [RHEL-…] ymir reproducer test` | Sibling streams extend the same MR; title accumulates keys e.g. `[RHEL-100, RHEL-200]` | + +Matching order: explicit `existing_mr_url` from agent output, then title CVE +tags, then title Jira tags (issue key or **Cloners-chain root** for regression +siblings). `_match_regression_sibling_mr` handles clone-chain grouping. ### Orchestration (`create_merge_request`) -Uses `result.test_directory` (relative to the tests clone) as the sole -source of truth for which files to `git add`. Rejects absolute paths and -`..` segments. +Uses `result.test_directory` (relative to the tests clone) as the sole source of +truth for which files to `git add`. Rejects absolute paths and `..` segments. | Result flags | Action | |--------------|--------| | `test_already_exists` and not `adapted_existing` | Skip MR | -| `adapted_existing` and success | Acquire lock; push to existing MR source branch (or `reproducer/` fallback) | -| New success | Acquire lock; commit `test_directory`, fork, open MR with label `ymir_reproducer` | -| `lock_deferred` / `retryable_error` | Skip MR; do not write terminal Jira labels | +| `adapted_existing` and success | Push to existing MR source branch (or `reproducer/` fallback) | +| New success | Commit `test_directory`, fork, open MR with label `ymir_reproducer` | +| `retryable_error` | Skip MR; do not write terminal Jira labels (delayed retry) | | Missing/invalid `test_directory` when MR needed | Fail; skip MR | -Branch resolution for adapt: list open `ymir_reproducer` MRs and match by -`existing_mr_url` or CVE/issue text in title/description. +When bootstrap found `existing_test_directory` and the agent reports +`adapted_existing`, `test_directory` must match that path; otherwise MR creation +is skipped. + +For **new** MRs, orchestration may call `request_mr_qe_reviews` when +`target_branch` or `fix_version` resolves a dist-git branch and +`ASSIGN_MR_REVIEWERS=true`. + +Orchestration GitLab tools (not in the agent MCP allowlist): `get_merge_request_details`, +`fetch_branch`, `fork_repository`, `commit_push_and_open_mr` (via `tasks.py`). ### not-affected semantics @@ -177,58 +288,73 @@ test that would detect the issue **if present**: ## Multi-Worker Clash Prevention (Redis Lock) -Sibling issues for the same CVE (different streams) contend on one lock so -only one worker creates or adapts the canonical test/MR at a time. +Sibling issues for the same CVE (different streams) or regression clones in the +same Cloners chain contend on one lock so only one worker runs the full +reproducer workflow (analysis, TF, MR) at a time. + +### Lock id + +Resolved by `resolve_reproducer_lock_id`: + +| Job type | `lock_id` | +|----------|-----------| +| CVE (`cve_id` set) | Normalized CVE id(s): sorted, comma-joined if multiple | +| Bug (no CVE) | Root issue of the Jira **Cloners** chain (Y-stream root), via issuelinks; falls back to issue key if resolution fails | -### Data structure +See `reproducer_lock_id()` and `resolve_clone_root()`. + +### Data structures Redis **Hash** `reproducer_creation_lock`: | Field pattern | Meaning | |---------------|---------| -| `{package}:{lock_id}:active` | Worker currently creating or adapting | - -**Lock id:** normalized CVE id(s) (sorted, comma-joined if multiple) or, for -non-CVE bugs, the Jira issue key. See `reproducer_lock_id()`. +| `{package}:{lock_id}:active` | Worker currently holding the workflow lock | -Unlike MR consolidation, there is **no pending slot** — waiters requeue on -`reproducer_queue_delayed` and retry later. +Per-lock **blocked list** `reproducer_blocked:{package}:{lock_id}` — tasks that +could not acquire the lock (not scanned by the Jira fetcher's queue dedup; see +[Fetcher stale recovery](#fetcher-stale-recovery)). -### Operations +Unlike MR consolidation, there is **no pending slot** in the hash — waiters park +on the blocked list until release or stale sweep. -**`try_acquire_reproducer_lock(package, lock_id)`** — Lua `HEXISTS` then -`HSET` if absent. Returns an ownership token (serialized lock entry JSON) on -success, or `None` if busy. +### Lifecycle (queue mode) -**`release_reproducer_lock(package, lock_id, token)`** — compare-and-delete the -`:active` field only when *token* still matches (same Lua as the stale -sweeper). Always called in `finally` after create/adapt push; a late release -from worker A cannot wipe worker B's re-acquired lock. +1. **`try_acquire_reproducer_lock(package, lock_id)`** at the start of + `process_task` (before `run_workflow`). Lua `HEXISTS` then `HSET` if absent. + Returns an ownership token (serialized `ReproducerLockEntry` JSON) on success, + or `None` if busy. +2. On success: set `ymir_reproducer_in_progress`, run the full workflow. +3. **`release_reproducer_lock(package, lock_id, token)`** in `finally` after the + workflow completes. Compare-and-delete the `:active` field only when *token* + still matches. Then `promote_blocked_reproducer_tasks` for that lock. +4. On busy: stage in-progress label (and user ack if triggered), RPUSH task to + blocked list, return without running the workflow. -**`sweep_stale_reproducer_locks(threshold=6h)`** — Removes `:active` entries -whose `activated_at` is older than the threshold, using compare-and-delete so -a lock released and re-acquired between snapshot and delete is not wiped. +**`sweep_stale_reproducer_locks(threshold=6h)`** — Removes stale `:active` +entries and promotes blocked tasks for each swept lock, using compare-and-delete +so a re-acquired lock is not wiped. ### Covered race -rhel-10 finishes and opens the MR; later rhel-9 and rhel-8 both find the test -fails on their compose and would adapt the same MR. The same `package:cve` -lock serializes adapters. The loser delayed-retries, re-clones / re-lists MRs, -re-verifies first, and often exits as reuse-only after the first portable -adaptation lands. - -### When the lock is required +rhel-10 finishes and opens the MR; later rhel-9 and rhel-8 both need to adapt +the same CVE MR. The same `package:cve` lock serializes full runs. Waiters sit +on the blocked list until rhel-10 releases; they then re-bootstrap the tests +clone (often already on the MR branch), re-verify, and often exit as reuse-only +after the first portable adaptation lands. -| Path | Lock? | -|------|-------| -| Create new MR | Yes | -| Adapt and push to existing MR | Yes | -| Pure verify of existing test (no mutate) | No | +### Standalone mode -Standalone / direct mode (`redis_conn is None`) skips the lock. +Queue locking lives in `process_task`, not inside `run_workflow`. Standalone / +direct mode (`JIRA_ISSUE` env, no Redis consumer) never acquires the lock. ## Workflow Steps +### Step 0: Bootstrap (orchestration, before LLM) + +See [Orchestration bootstrap](#orchestration-bootstrap) (`_bootstrap_tests_clone`). +Skipped when `package` is unset. + ### Step 1: `run_reproducer_analysis` BeeAI `ReasoningAgent` with tools for Jira, patches, maintainer rules, git @@ -248,13 +374,13 @@ deferred until after parallel tools finish — no extra inference round. ### Step 2: `create_merge_request` See orchestration table above. Commits under -`/git-repos/tests-`, forks the tests project, opens or updates the -MR via `commit_push_and_open_mr`. +`GIT_REPO_BASEPATH/Reproducer//tests-`, forks the tests +project, opens or updates the MR via `commit_push_and_open_mr`. ### Step 3: `handle_results` -Writes terminal Jira labels and a comment unless `retryable_error` or -`lock_deferred` (keeps `ymir_reproducer_in_progress` for retry). +Writes terminal Jira labels and a comment unless `retryable_error` (keeps +`ymir_reproducer_in_progress` for delayed retry). | Outcome | Label | |---------|-------| @@ -262,10 +388,11 @@ Writes terminal Jira labels and a comment unless `retryable_error` or | Existing verified, no adapt | `ymir_reproducer_already_exists` | | Not reproducible | `ymir_reproducer_not_reproducible` | | Other failure | `ymir_reproducer_failed` | -| Exhausted retries | `ymir_reproducer_errored` | +| Exhausted queue retries | `ymir_reproducer_errored` | -Dedup: before work, skip if a terminal reproducer label is present and the -issue is not in-progress (unless `user_triggered`). +**Queue dedup:** before work, skip if a terminal reproducer label is present and +the issue is not in-progress (unless `user_triggered`). Staging in-progress +removes terminal reproducer labels so a fresh run can replace them. ## Jira Labels and Fetcher @@ -281,22 +408,53 @@ runs afterward without clearing those. Fetcher stale recovery for labels and only treats another `ymir_reproducer_*` label as proof the stage already finished. +### Fetcher stale recovery + +When `ymir_reproducer_in_progress` looks abandoned (no Jira update within +`STALE_LABEL_THRESHOLD_HOURS`, default 24h) and the issue is not already queued +in `reproducer_queue` / `reproducer_queue_todo`, the fetcher flips the label to +`ymir_retry_needed` and re-enqueues to **triage** (full re-triage). The +original `ReproducerInputSchema` payload only existed in Redis and is not +recoverable from Jira alone. + +Blocked tasks on `reproducer_blocked:*` are **not** included in the fetcher's +`existing_keys` scan. A task blocked on lock for longer than the stale threshold +could theoretically be treated as abandoned while still parked (edge case). + See also [jira_label_workflow_routing.md](../jira_label_workflow_routing.md). +## Environment Variables + +| Variable | Default | Role | +|----------|---------|------| +| `TRIAGE_ENQUEUE_REPRODUCER` | `true` | Triage LPUSH to reproducer queues | +| `AUTO_CHAIN` | `true` | Triage → fix-agent queues (separate from reproducer) | +| `REPRODUCER_RETRY_DELAY_SECONDS` | `1800` | Delayed ZSET retry for `retryable_error` | +| `REPRODUCER_POLL_TIMEOUT` | `30` | BRPOP timeout (seconds) | +| `MAX_RETRIES` | `3` | Queue task retries before `ymir_reproducer_errored` | +| `MAX_CONCURRENT_TASKS` | `1` | Reproducer worker concurrency | +| `STALE_LABEL_THRESHOLD_HOURS` | `24` | Fetcher abandoned in-flight detection | +| `ASSIGN_MR_REVIEWERS` | `true` (OpenShift) | QE reviewer on new test MRs | +| `GIT_REPO_BASEPATH` | `/git-repos` | Per-issue Reproducer working dirs | +| `DRY_RUN` | `false` | Skip MR and Jira finalization | + ## Safety Invariants | Invariant | Mechanism | |-----------|-----------| | No duplicate triage→reproducer for ineligible resolutions | `_REPRODUCER_ELIGIBLE_RESOLUTIONS` gate | | No enqueue without package | `_build_reproducer_input` returns `None` | -| One create/adapt at a time per package+CVE/issue | Redis lock + Lua acquire; release is compare-and-delete by ownership token | -| Waiters do not spin | Delayed ZSET retry on lock busy | +| No run for disabled packages | `fetch_reproducer_config` + `enabled: true` | +| One full reproducer run at a time per package+lock_id | Redis lock at `process_task` start; release in `finally` | +| Waiters do not spin on lock busy | Blocked list; promoted on release or stale sweep | | Abandoned locks do not block forever | 6h stale sweep with compare-and-delete | -| No second concurrent run of same Jira issue | `ymir_reproducer_in_progress` + terminal labels | +| No second concurrent run of same Jira issue | `ymir_reproducer_in_progress` + terminal labels (+ queue dedup) | | No TF machine leaks | Agent cancel + `TFReservationCleanupMiddleware` | -| Remote SSH/SCP cannot target arbitrary hosts or other clones | `ssh_host` allowlisted from reservation details; SCP paths limited to `tests-` / `/tmp` | +| Remote SSH/SCP cannot target arbitrary hosts | `ssh_host` allowlisted from reservation details | +| SCP paths scoped | Under `/git-repos` or `/tmp`; job metadata scopes Reproducer tree | | No real writes in dry-run / tests | `DRY_RUN` skips MR and Jira finalization | -| Adapt targets the right MR | Match open `ymir_reproducer` MRs by URL / CVE / issue; fetch/checkout MR source branch before overlaying local adaptations | +| Adapt targets the right MR | Title-only bracket-tag match; bootstrap checks out MR branch before agent runs | +| Adapt uses correct test path | `test_directory` must match `existing_test_directory` when adapting open MR | ## Running and Testing the Agent @@ -310,7 +468,9 @@ make run-reproducer-agent-standalone JIRA_ISSUE=RHEL-12345 DRY_RUN=true ``` In this mode Redis is unused: the agent runs `run_workflow` once and exits. -Create/adapt locking is skipped when there is no Redis connection. +No queue lock or package gate (unless you pass full metadata via a custom +integration). For realistic runs, use queue mode with `make trigger-reproducer` +or LPUSH a full `ReproducerInputSchema` task. ### E2E tests @@ -325,15 +485,15 @@ make run-reproducer-agent-e2e-tests These are regression tests for the agent — they are not a way to produce reproducer MRs for real issues. -Unit tests for enqueue helpers, labels, and the create/adapt lock live under +Unit tests for enqueue helpers, labels, blocked lock, and MR helpers live under `ymir/agents/tests/unit/` and `ymir/common/tests/unit/` (see File Map). ### Queue mode (production) Start the compose `reproducer-agent` service (agents profile) with Redis and -**without** `JIRA_ISSUE`. Triage auto-enqueue is currently **disabled**; use -`make trigger-reproducer` (below) to feed `reproducer_queue` / -`reproducer_queue_todo`. +**without** `JIRA_ISSUE`. With default settings, triage auto-enqueues eligible +resolutions to the reproducer queues. Use `make trigger-reproducer` (below) for +ad-hoc jobs or when `TRIAGE_ENQUEUE_REPRODUCER=false`. On OpenShift, `openshift/deployment-reproducer-agent.yml` runs the same queue worker (`beeai-agent:c10s`, module `ymir.agents.reproducer_agent`). Testing Farm @@ -352,8 +512,9 @@ Same optional flags as the local target (`CVE_ID`, `FIX_VERSION`, ### Manual queue submit -With the agents stack running (`make start` / `make start DRY_RUN=true`) and -the `reproducer-agent` worker up, enqueue a job: +For ad-hoc enqueue or when `TRIAGE_ENQUEUE_REPRODUCER=false`. With the agents +stack running (`make start` / `make start DRY_RUN=true`) and the +`reproducer-agent` worker up, enqueue a job: ```bash # Minimum (required fields) @@ -376,12 +537,16 @@ make trigger-reproducer JIRA_ISSUE=RHEL-12345 PACKAGE=bind USER_TRIGGERED=true |----------|----------|---------|-------| | `JIRA_ISSUE` | yes | — | Issue key (e.g. `RHEL-12345`) | | `PACKAGE` | yes | — | Downstream component / tests-repo name | -| `CVE_ID` | no | unset | CVE id(s); used for lock id and Security/ path hints | +| `CVE_ID` | no | unset | CVE id(s); lock id and Security/ path hints | | `FIX_VERSION` | no | unset | Jira fix version (e.g. `rhel-10.1`) | | `TARGET_BRANCH` | no | unset | Dist-git / stream hint (e.g. `c10s`) | | `TRIAGE_SUMMARY` | no | unset | Free-text context for the reproducer prompt | | `USER_TRIGGERED` | no | `false` | `true` → `reproducer_queue_todo` + user ack comments | +`ReproducerInputSchema` also supports `patch_urls` (set by triage enqueue, not +exposed by `make trigger-reproducer`). For manual jobs, LPUSH JSON with full +metadata if patch context is needed. + The target LPUSHes a `Task` whose `metadata` validates as `ReproducerInputSchema` onto Valkey. Watch progress with: @@ -390,29 +555,32 @@ $(COMPOSE) -f compose.yaml --profile=agents logs -f reproducer-agent # or Redis Commander at http://localhost:8081/ ``` -Standalone mode (`make run-reproducer-agent-standalone`) bypasses Redis entirely -and is better for one-off dry runs without a running worker. +Standalone mode bypasses Redis entirely and is better for quick prompt/dry-run +experiments without a running worker. ## File Map | File | Purpose | |------|---------| -| `ymir/agents/triage_agent.py` | `_build_reproducer_input`, `_enqueue_reproducer`, parallel dispatch | -| `ymir/agents/reproducer_agent.py` | Workflow, queue consumer, MR create/adapt, lock integration | +| `ymir/agents/triage_agent.py` | `_build_reproducer_input`, `_enqueue_reproducer`, `TRIAGE_ENQUEUE_REPRODUCER` gate | +| `ymir/agents/reproducer_agent.py` | Workflow, queue consumer, bootstrap, MR create/adapt, lock integration | +| `ymir/agents/tasks.py` | `fetch_reproducer_config`, `commit_push_and_open_mr`, `request_mr_qe_reviews` | | `ymir/agents/reasoning_agent/context_management.py` | `manage_context` tool + deferred memory compaction | -| `ymir/common/reproducer_lock.py` | Acquire / release / stale sweep | +| `ymir/common/reproducer_lock.py` | Acquire / release / stale sweep / blocked queue promote | | `ymir/common/delayed_queue.py` | Delayed retry ZSET helpers | -| `ymir/common/models.py` | `ReproducerInputSchema`, `ReproducerOutputSchema`, enriched `NotAffectedData` | +| `ymir/common/models.py` | `ReproducerInputSchema`, `ReproducerOutputSchema`, `PackageReproducerConfig` | | `ymir/common/constants.py` | Queue names, `get_reproducer_queue()`, Jira labels | | `ymir/agents/prompts/reproducer/prompt.j2` | LLM workflow (reuse / verify / adapt) | | `ymir/agents/prompts/reproducer/output_format.j2` | Expected agent JSON | | `ymir/agents/tf_cleanup_middleware.py` | TF reservation leak cleanup | -| `ymir/tools/privileged/testing_farm.py` | TF MCP tools | -| `ymir/jira_issue_fetcher/jira_issue_fetcher.py` | `IN_FLIGHT` includes reproducer | +| `ymir/tools/privileged/testing_farm.py` | TF MCP tools, SSH/SCP allowlisting | +| `ymir/tools/privileged/gitlab.py` | Fork, MR, branch fetch (orchestration) | +| `ymir/jira_issue_fetcher/jira_issue_fetcher.py` | `IN_FLIGHT` includes reproducer; stale → triage re-queue | | `agents_as_skills/reproducer/SKILL.md` | Skill mirror of the workflow | -| `ymir/agents/tests/unit/test_reproducer_agent.py` | Unit tests: label / MR-need helpers | +| `ymir/agents/tests/unit/test_reproducer_agent.py` | Unit tests: labels, MR helpers, blocked lock | | `ymir/agents/tests/unit/test_context_management.py` | Unit tests: manage_context compaction | | `ymir/agents/tests/unit/test_triage_agent.py` | Unit tests: enqueue input builder | -| `ymir/common/tests/unit/test_reproducer_lock.py` | Unit tests: create/adapt lock | +| `ymir/common/tests/unit/test_reproducer_lock.py` | Unit tests: create/adapt lock + blocked queue | | `ymir/agents/tests/e2e/reproducer_agent/` | E2E test suite (mock repos / fixtures) | | `openshift/deployment-reproducer-agent.yml` | Production OpenShift Deployment | +| `openshift/configmap-agents-env.yml` | `TRIAGE_ENQUEUE_REPRODUCER`, `ASSIGN_MR_REVIEWERS`, etc. | diff --git a/jira_label_workflow_routing.md b/jira_label_workflow_routing.md index 6e18e51a6..48295cb18 100644 --- a/jira_label_workflow_routing.md +++ b/jira_label_workflow_routing.md @@ -136,9 +136,9 @@ These labels are applied to GitLab merge requests (not Jira issues): | `clarification_needed_queue` | Input | Resolution=CLARIFICATION | `ymir_needs_attention` | Active (AUTO_CHAIN only) | | `error_list` | Output | Any error | `ymir_*_errored` | Active | | `open_ended_analysis_list` | Output | Resolution=OPEN_ENDED_ANALYSIS | `ymir_triaged` | Active (AUTO_CHAIN only) | -| `reproducer_queue` | Input | Manual `make trigger-reproducer` (triage auto-enqueue temporarily disabled) | `ymir_reproducer_in_progress` → terminal `ymir_reproducer_*` | Manual enqueue only | -| `reproducer_queue_todo` | Input (priority) | Same when `USER_TRIGGERED=true` | Same as `reproducer_queue` | Manual enqueue only | -| `reproducer_queue_delayed` | Delayed (ZSET) | Retryable TF infra or create/adapt lock contention | Keeps `ymir_reproducer_in_progress` until retry completes | Active | +| `reproducer_queue` | Input | Triage auto-enqueue (`TRIAGE_ENQUEUE_REPRODUCER`) or `make trigger-reproducer` | `ymir_reproducer_in_progress` → terminal `ymir_reproducer_*` | Active when auto-enqueue enabled | +| `reproducer_queue_todo` | Input (priority) | Same when `user_triggered` / `USER_TRIGGERED=true` | Same as `reproducer_queue` | Active when auto-enqueue enabled | +| `reproducer_queue_delayed` | Delayed (ZSET) | Retryable TF infra (`retryable_error`) | Keeps `ymir_reproducer_in_progress` until retry completes | Active | | `completed_rebase_list` | Output | Rebase success | `ymir_rebased` | Active | | `completed_backport_list` | Output | Backport success | `ymir_backported` | Active | | `completed_reproducer_list` | Output | Reproducer success/failure (non-retry) | `ymir_reproducer_*` | Active | diff --git a/openshift/Makefile b/openshift/Makefile index d28590d31..45c16af15 100644 --- a/openshift/Makefile +++ b/openshift/Makefile @@ -132,7 +132,7 @@ process: '{"metadata":{"issue":"$(ISSUE)"},"attempts":0,"user_triggered":false}' @echo "✓ Task queued successfully. View queue with: make show-triage-queue" -# Manually enqueue a reproducer job (triage auto-enqueue is currently disabled). +# Manually enqueue a reproducer job (also when TRIAGE_ENQUEUE_REPRODUCER=false). # Same interface as the local `make trigger-reproducer` target. # Required: JIRA_ISSUE, PACKAGE # Optional: CVE_ID, FIX_VERSION, TARGET_BRANCH, TRIAGE_SUMMARY, USER_TRIGGERED=true diff --git a/openshift/README.md b/openshift/README.md index f4c26e724..0b8852413 100644 --- a/openshift/README.md +++ b/openshift/README.md @@ -185,8 +185,10 @@ Set `force_cve_triage` to `false` for a normal triage run. This mirrors the `mak ### Triggering a reproducer job -Triage auto-enqueue of reproducer jobs is currently disabled. Enqueue manually -(same flags as local `make trigger-reproducer`): +When triage auto-enqueue is enabled (`TRIAGE_ENQUEUE_REPRODUCER=true`, the +OpenShift default), eligible triage resolutions LPUSH reproducer jobs +automatically. You can still enqueue manually (same flags as local +`make trigger-reproducer`): ```bash make trigger-reproducer JIRA_ISSUE=RHEL-12345 PACKAGE=bind diff --git a/openshift/configmap-agents-env.yml b/openshift/configmap-agents-env.yml index 6d9722f4b..94be3a532 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. + # Set to "false" to disable auto-enqueue (manual: make trigger-reproducer). + TRIAGE_ENQUEUE_REPRODUCER: "true" immutable: false kind: ConfigMap metadata: 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 ddb4202ab..ff302d7c3 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 @@ -22,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, @@ -34,7 +34,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 @@ -54,11 +53,14 @@ ReproducerOutputSchema as OutputSchema, ) 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, ) +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 @@ -156,17 +158,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) @@ -248,6 +294,71 @@ 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(): + by_name = { + child.name.upper(): child + for child in security.iterdir() + if child.is_dir() and _is_reproducer_test_dir(child) + } + for cve in cves: + if cve in by_name: + return by_name[cve] + 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(): @@ -295,13 +406,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 "" @@ -310,17 +430,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. @@ -335,6 +473,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 "" @@ -344,6 +483,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 @@ -365,11 +506,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. @@ -379,23 +634,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") @@ -419,84 +682,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: @@ -530,7 +765,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: @@ -546,6 +788,53 @@ 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, + 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, @@ -570,21 +859,25 @@ 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] ) + 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(), ) @@ -617,33 +910,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 = ( @@ -671,29 +937,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, ) @@ -713,7 +997,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, @@ -727,6 +1011,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: @@ -739,17 +1038,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 +1113,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 +1281,102 @@ 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, + await retry( + task, + ErrorData( + details="Missing package in reproducer task metadata", + jira_issue=input_data.jira_issue, + ).model_dump_json(), + ) + 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, + 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(), ) - 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." + 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 +1402,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 +1428,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/tasks.py b/ymir/agents/tasks.py index 6591a16ca..9ffaf2cec 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 @@ -445,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, @@ -878,6 +910,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 +936,6 @@ async def fetch_consolidation_config( Returns: Parsed consolidation config. """ - import yaml - try: raw = await run_tool( "get_maintainer_rules", @@ -967,3 +1001,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 (disabled) 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 f52c3842c..553b82f8f 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 @@ -7,21 +8,27 @@ 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, _reproducer_mr_title_tags, _resolve_reproducer_mr_target, _resolve_test_dir, + _reviewer_lookup_branch, _should_finalize_jira, 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 @@ -75,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 @@ -116,6 +138,55 @@ 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_rejects_cve_prefix_collision(tmp_path: Path): + repo = tmp_path / "tests-pkg" + repo.mkdir() + for cve in ("CVE-2026-1", "CVE-2026-10"): + test_dir = repo / "Security" / cve + test_dir.mkdir(parents=True) + (test_dir / "main.fmf").write_text(f"summary: {cve}\n") + + discovered = _discover_existing_reproducer_test_dir( + repo, + cve_id="CVE-2026-10", + jira_issue="RHEL-1", + reproducer_type="cve", + ) + assert discovered == repo / "Security" / "CVE-2026-10" + + +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"] @@ -167,13 +238,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(): @@ -401,80 +572,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" # ============================================================================= @@ -483,11 +691,50 @@ 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_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 ( + 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 +813,8 @@ 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( result=MagicMock(success=True, retryable_error=False, lock_deferred=False, summary="ok") @@ -588,6 +837,8 @@ 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( result=MagicMock(success=True, retryable_error=False, lock_deferred=False, summary="ok") @@ -609,6 +860,8 @@ 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( result=MagicMock(success=True, retryable_error=False, lock_deferred=False, summary="ok") @@ -616,3 +869,105 @@ 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, + _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 False + + +@pytest.mark.asyncio +async def test_fetch_reproducer_config_parses_enabled(): + with patch("ymir.agents.tasks.run_tool", new_callable=AsyncMock) as mock_run: + mock_run.return_value = "reproducer:\n enabled: true\n" + 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/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/agents/triage_agent.py b/ymir/agents/triage_agent.py index c9a1baa17..c6d415a84 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())) @@ -1136,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", "true").lower() == "true" force_cve_triage = os.getenv("FORCE_CVE_TRIAGE", "false").lower() == "true" if jira_issue := os.getenv("JIRA_ISSUE", None): @@ -1235,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)) @@ -1650,23 +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: - # await _enqueue_reproducer(redis, state, user_triggered) - # 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) diff --git a/ymir/common/models.py b/ymir/common/models.py index 38ab2fbff..7e804d49d 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=False, + description="Whether to run the Ymir reproducer workflow for this package", + ) + + class MRConsolidationInputSchema(BaseModel): """Input schema for the MR consolidation agent.""" @@ -1346,8 +1359,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 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")