Collect clones and postpone triaging of y streams for CVEs

Makefile

@@ -27,7 +27,12 @@ run-triage-agent-standalone:
 		-e MOCK_JIRA=$(MOCK_JIRA) \
 		triage-agent
 
-
+.PHONY: run-clones-analyzer-agent-standalone
+run-clones-analyzer-agent-standalone:
+	$(COMPOSE_AGENTS) run --rm \
+		-e JIRA_ISSUE=$(JIRA_ISSUE) \
+		-e DRY_RUN=$(DRY_RUN) \
+		clones-analyzer-agent
 
 
 .PHONY: run-rebase-agent-c9s-standalone

agents/clones_analyzer_agent.py

@@ -0,0 +1,142 @@
+import os
+import copy
+import logging
+from typing import Any
+from textwrap import dedent
+
+from beeai_framework.agents.requirement import RequirementAgent
+from beeai_framework.agents.requirement.prompts import RequirementAgentSystemPrompt
+from beeai_framework.agents.requirement.requirements.conditional import (
+    ConditionalRequirement,
+)
+from beeai_framework.memory import UnconstrainedMemory
+from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
+from beeai_framework.tools import Tool
+from beeai_framework.tools.think import ThinkTool
+from beeai_framework.workflows import Workflow
+
+from pydantic import BaseModel, Field
+from observability import setup_observability
+
+from tools.commands import RunShellCommandTool
+from tools.version_mapper import VersionMapperTool
+from common.models import ClonesInputSchema, ClonesOutputSchema
+from utils import get_chat_model, get_tool_call_checker_config
+from utils import mcp_tools, get_agent_execution_config
+
+logger = logging.getLogger(__name__)
+
+
+def get_instructions() -> str:
+    return """
+      You are an expert on finding other Jira issues related to a given Jira issue
+      in RHEL Jira project by analyzing the Jira fields and comments.
+
+      To find other Jira issues which are clones of <JIRA_ISSUE> Jira issue, do the following:
+
+      1. Search for other Jira issues which have the same affected component as <JIRA_ISSUE> Jira issue in RHEL Jira project and extract their titles.
+
+      2. Compare the titles of the found Jira issues with the title of <JIRA_ISSUE> Jira issue and identify the ones which are clones.
+      For example, if the title of <JIRA_ISSUE> Jira issue is "CVE-YYYY-XXXXX libsoup3: Out-of-Bounds Read in Cookie Date Handling of libsoup HTTP Library [rhel-10.1]"
+      and you have found another Jira issue with the title "CVE-YYYY-XXXXX libsoup3: Out-of-Bounds Read in Cookie Date Handling of libsoup HTTP Library [rhel-10.0z]",
+      then it is a clone of <JIRA_ISSUE> Jira issue or <JIRA_ISSUE> Jira issue is a clone of the found Jira issue.
+
+      3.Usually clones are already linked to each other in Jira through the "Issue Links" field.
+      If not, link the found Jira issues to <JIRA_ISSUE> Jira issue and the <JIRA_ISSUE> Jira issue to the found Jira issues
+      through the "is related" relationship.
+
+     General instructions:
+      - If in DRY RUN mode, do not link the Jira issues to each other but tell the user that you would have linked them.
+    """
+
+
+def get_prompt(input: ClonesInputSchema) -> str:
+    return f"""
+      Find other Jira issues which are clones of {input.jira_issue} Jira issue and link them to each other.
+      Also check if {input.jira_issue} Jira issue is a clone of any of the found Jira issues and link them to each other.
+    """
+
+def get_agent_definition(gateway_tools: list[Tool]) -> dict[str, Any]:
+    return {
+    "name": "ExistingClonesAnalyzerAgent",
+    "llm": get_chat_model(),
+    "tool_call_checker": get_tool_call_checker_config(),
+    "tools": [ThinkTool(), RunShellCommandTool(), VersionMapperTool()]
+        + [t for t in gateway_tools if t.name in ["get_jira_details", "set_jira_fields"]],
+    "memory": UnconstrainedMemory(),
+    "requirements": [
+        ConditionalRequirement(
+            ThinkTool,
+            force_at_step=1,
+            force_after=Tool,
+            consecutive_allowed=False,
+            only_success_invocations=False,
+        ),
+        ],
+    "middlewares": [GlobalTrajectoryMiddleware(pretty=True)],
+    "role": "Red Hat Enterprise Linux developer",
+    "instructions": get_instructions(),
+    "templates": {"system": copy.deepcopy(RequirementAgentSystemPrompt)}
+    }
+
+def create_clones_analyzer_agent(mcp_tools: list[Tool], local_tool_options: dict[str, Any]) -> RequirementAgent:
+    return RequirementAgent(**get_agent_definition(mcp_tools))
+
+WORKFLOW_STEP_INSTRUCTIONS = dedent("""
+                            The final answer must be a JSON object with the following fields:
+                            - `clones`: a list of Jira issue keys and branches that are clones of the given Jira issue or the given Jira issue is a clone of the found Jira issues
+                            - `links`: a list of links you have added between the given Jira issue and the found Jira issues or the found Jira issues and the given Jira issue
+                            ```json
+                            {
+                            "clones": [{"jira_issue": "RHEL-12345", "branch": "rhel-9.6z"},
+                                       {"jira_issue": "RHEL-12346", "branch": "rhel-9.7"}],
+                            "links": [{"source": "RHEL-12345", "target": "RHEL-12346"},
+                                      {"source": "RHEL-12346", "target": "RHEL-12345"}]
+                            }
+                            ```
+""")
+
+async def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+
+    setup_observability(os.environ["COLLECTOR_ENDPOINT"])
+
+    dry_run = os.getenv("DRY_RUN", "False").lower() == "true"
+
+    class State(BaseModel):
+        jira_issue: str
+        clones_result: ClonesOutputSchema | None = Field(default=None)
+
+    async def run_workflow(jira_issue):
+        async with mcp_tools(os.getenv("MCP_GATEWAY_URL")) as gateway_tools:
+            clones_analyzer_agent = RequirementAgent(
+                **get_agent_definition(gateway_tools),
+            )
+
+            async def identify_existing_clones(state):
+                """Identify and link clones of the given Jira issue"""
+                logger.info(f"Identifying and linking clones of {state.jira_issue}")
+                response = await clones_analyzer_agent.run(
+                    get_prompt(ClonesInputSchema(jira_issue=state.jira_issue)),
+                    expected_output=WORKFLOW_STEP_INSTRUCTIONS,
+                    **get_agent_execution_config(),
+                    )
+
+                state.clones_result = ClonesOutputSchema.model_validate_json(response.last_message.text)
+                return Workflow.END
+
+            workflow = Workflow(State, name="ClonesAnalyzerWorkflow")
+            workflow.add_step("identify_existing_clones", identify_existing_clones)
+            await workflow.run(State(jira_issue=jira_issue))
+
+    jira_issue = os.getenv("JIRA_ISSUE")
+    if not jira_issue:
+        logger.error("JIRA_ISSUE environment variable is required")
+        return
+
+    await run_workflow(jira_issue)
+
+
+if __name__ == "__main__":
+    import asyncio
+    asyncio.run(main())

agents/triage_agent.py

@@ -10,6 +10,7 @@
 from typing import Union
 
 from pydantic import BaseModel, Field
+from typing import List
 
 from beeai_framework.agents.requirement import RequirementAgent
 from beeai_framework.agents.requirement.requirements.conditional import (
@@ -25,6 +26,7 @@
 from beeai_framework.utils.strings import to_json
 
 import tasks
+from agents import clones_analyzer_agent
 from common.config import load_rhel_config
 from common.models import (
     Task,
@@ -37,6 +39,9 @@
     NoActionData,
     ErrorData,
     CVEEligibilityResult,
+    WhenEligibility,
+    ClonesOutputSchema,
+    Clone,
 )
 from common.utils import redis_client, fix_await
 from common.constants import JiraLabels, RedisQueues
@@ -300,6 +305,7 @@ async def main() -> None:
 
     class State(BaseModel):
         jira_issue: str
+        clones: List[Clone] | None = Field(default=None)
         cve_eligibility_result: CVEEligibilityResult | None = Field(default=None)
         triage_result: OutputSchema | None = Field(default=None)
         target_branch: str | None = Field(default=None)
@@ -341,6 +347,21 @@ async def run_workflow(jira_issue):
 
             workflow = Workflow(State, name="TriageWorkflow")
 
+            async def run_clones_analyzer_agent(state):
+                """Run the clones analyzer agent"""
+                logger.info(f"Running clones analyzer agent for {state.jira_issue}")
+                clones_analyzer_agent_definition = RequirementAgent(
+                    **clones_analyzer_agent.get_agent_definition(gateway_tools),
+                )
+
+                response = await clones_analyzer_agent_definition.run(
+                    clones_analyzer_agent.get_prompt(clones_analyzer_agent.ClonesInputSchema(jira_issue=state.jira_issue)),
+                    expected_output=clones_analyzer_agent.WORKFLOW_STEP_INSTRUCTIONS,
+                    **clones_analyzer_agent.get_agent_execution_config(),
+                    )
+                state.clones = ClonesOutputSchema.model_validate_json(response.last_message.text).clones
+                return "check_cve_eligibility"
+
             async def check_cve_eligibility(state):
                 """Check CVE eligibility for the issue"""
                 logger.info(f"Checking CVE eligibility for {state.jira_issue}")
@@ -354,28 +375,59 @@ async def check_cve_eligibility(state):
                 logger.info(f"CVE eligibility result: {state.cve_eligibility_result}")
 
                 # If not eligible for triage, end workflow
-                if not state.cve_eligibility_result.is_eligible_for_triage:
-                    logger.info(f"Issue {state.jira_issue} not eligible for triage: {state.cve_eligibility_result.reason}")
-                    if state.cve_eligibility_result.error:
+                match state.cve_eligibility_result.when_eligible_for_triage:
+                    case WhenEligibility.NEVER:
+                        logger.info(f"Issue {state.jira_issue} not eligible for triage: {state.cve_eligibility_result.reason}")
+                        if state.cve_eligibility_result.error:
+                            state.triage_result = OutputSchema(
+                            resolution=Resolution.ERROR,
+                            data=ErrorData(
+                                details=f"CVE eligibility check error: {state.cve_eligibility_result.error}",
+                                jira_issue=state.jira_issue
+                            )
+                        )
+                        return "comment_in_jira"
+                    case WhenEligibility.LATER:
+                        logger.info(f"Issue {state.jira_issue} is eligible for triage, but could be postponed: {state.cve_eligibility_result.reason}")
+                        return "run_postponed_triage_analysis"
+                    case WhenEligibility.IMMEDIATELY:
+                        logger.info(f"Issue {state.jira_issue} is eligible for triage: {state.cve_eligibility_result.reason}")
+                        return "run_triage_analysis"
+                    case _:
+                        logger.error(f"Unknown eligibility result: {state.cve_eligibility_result}")
                         state.triage_result = OutputSchema(
-                        resolution=Resolution.ERROR,
-                        data=ErrorData(
-                            details=f"CVE eligibility check error: {state.cve_eligibility_result.error}",
-                            jira_issue=state.jira_issue
+                            resolution=Resolution.ERROR,
+                            data=ErrorData(
+                                details=f"Unknown eligibility result: {state.cve_eligibility_result}",
+                                jira_issue=state.jira_issue
+                            )
                         )
-                    )
-                    else:
+                        return "comment_in_jira"
+
+            async def run_postponed_triage_analysis(state):
+                """Run the postponed triage analysis,
+                check if the Z-stream errata has been shipped
+                before proceeding with the triage analysis in a Y-stream"""
+                logger.info(f"Running postponed triage analysis for {state.jira_issue}")
+                for clone in state.clones:
+                    z_streams_errata_shipped = await run_tool(
+                        "check_z_stream_errata_shipped",
+                        available_tools=gateway_tools,
+                        issue_key=clone.jira_issue,
+                        branch=clone.branch)
+                    if not z_streams_errata_shipped:
+                        msg = f"Z-stream errata not shipped yet for issue {clone.jira_issue} for branch {clone.branch}, postponing triage analysis"
+                        logger.info(msg)
                         state.triage_result = OutputSchema(
-                            resolution=Resolution.NO_ACTION,
+                            resolution=Resolution.POSTPONED,
                             data=NoActionData(
-                                reasoning=f"CVE eligibility check decided to skip triaging: {state.cve_eligibility_result.reason}",
+                                reasoning=msg,
                                 jira_issue=state.jira_issue
                             )
                         )
-                    return "comment_in_jira"
+                        return "comment_in_jira"
 
-                reason = state.cve_eligibility_result.reason
-                logger.info(f"Issue {state.jira_issue} is eligible for triage: {reason}")
+                logger.info(f"All z-stream erratas shipped, proceeding with triage analysis")
                 return "run_triage_analysis"
 
             async def run_triage_analysis(state):
@@ -499,9 +551,23 @@ async def comment_in_jira(state):
                     comment_text=comment_text,
                     available_tools=gateway_tools,
                 )
+                if state.triage_result.resolution == Resolution.POSTPONED:
+                    await tasks.set_jira_labels(
+                        jira_issue=state.jira_issue,
+                        labels_to_add=[JiraLabels.POSTPONED.value],
+                        dry_run=dry_run
+                    )
+                elif JiraLabels.POSTPONED.value in JiraLabels.all_labels():
+                    await tasks.set_jira_labels(
+                        jira_issue=state.jira_issue,
+                        labels_to_remove=[JiraLabels.POSTPONED.value],
+                        dry_run=dry_run
+                    )
                 return Workflow.END
 
+            workflow.add_step("run_clones_analyzer_agent", run_clones_analyzer_agent)
             workflow.add_step("check_cve_eligibility", check_cve_eligibility)
+            workflow.add_step("run_postponed_triage_analysis", run_postponed_triage_analysis)
             workflow.add_step("run_triage_analysis", run_triage_analysis)
             workflow.add_step("verify_rebase_author", verify_rebase_author)
             workflow.add_step("determine_target_branch", determine_target_branch_step)

common/__init__.py

@@ -2,5 +2,6 @@
 
 from .config import load_rhel_config
 from .models import CVEEligibilityResult
+from .models import WhenEligibility
 
-__all__ = ["load_rhel_config", "CVEEligibilityResult"]
+__all__ = ["load_rhel_config", "CVEEligibilityResult", "WhenEligibility"]

common/constants.py

@@ -69,6 +69,7 @@ class JiraLabels(Enum):
     REBASED = "jotnar_rebased"
     BACKPORTED = "jotnar_backported"
     MERGED = "jotnar_merged"
+    POSTPONED = "jotnar_postponed"
 
     REBASE_ERRORED = "jotnar_rebase_errored"
     BACKPORT_ERRORED = "jotnar_backport_errored"

common/models.py

@@ -10,6 +10,11 @@
 from pathlib import Path
 from enum import Enum
 
+class WhenEligibility(Enum):
+    """When eligibility for triage."""
+    IMMEDIATELY = "immediately"
+    LATER = "later"
+    NEVER = "never"
 
 class CVEEligibilityResult(BaseModel):
     """
@@ -21,8 +26,8 @@ class CVEEligibilityResult(BaseModel):
     is_cve: bool = Field(
         description="Whether this is a CVE (identified by SecurityTracking label)"
     )
-    is_eligible_for_triage: bool = Field(
-        description="Whether triage agent should process this CVE"
+    when_eligible_for_triage: WhenEligibility = Field(
+        description="Whether triage agent should process this issue immediately, later or never"
     )
     reason: str = Field(
         description="Explanation of the eligibility decision"
@@ -118,6 +123,7 @@ class Resolution(Enum):
     BACKPORT = "backport"
     CLARIFICATION_NEEDED = "clarification-needed"
     NO_ACTION = "no-action"
+    POSTPONED = "postponed"
     ERROR = "error"
 
 
@@ -292,3 +298,26 @@ class FailedPipelineJob(BaseModel):
     artifacts_url: str = Field(
         description="URL to browse job artifacts, empty string if no artifacts"
     )
+
+
+# ============================================================================
+# Clones Analyzer Agent Schemas
+# ============================================================================
+
+class ClonesInputSchema(BaseModel):
+    """Input schema for the clones analyzer agent."""
+    jira_issue: str = Field(description="Jira issue key to identify clones of")
+
+class Clone(BaseModel):
+    """A clone of a Jira issue."""
+    jira_issue: str = Field(description="Jira issue key")
+    branch: str = Field(description="Branch")
+
+class Link(BaseModel):
+    """A link between two Jira issues."""
+    source: str = Field(description="Source Jira issue key")
+    target: str = Field(description="Target Jira issue key")
+class ClonesOutputSchema(BaseModel):
+    """Output schema for the clones analyzer agent."""
+    clones: list[Clone] = Field(description="List of Jira issue keys and branches that are clones of the given Jira issue or the given Jira issue is a clone of the found Jira issues")
+    links: list[Link] = Field(description="List of links between the given Jira issue and the found Jira issues or the found Jira issues and the given Jira issue")

compose.yaml

@@ -135,6 +135,11 @@ services:
     command: ["python", "agents/triage_agent.py"]
     profiles: ["agents"]
 
+  clones-analyzer-agent:
+    <<: *beeai-agent-c10s
+    command: ["python", "agents/clones_analyzer_agent.py"]
+    profiles: ["agents"]
+
   backport-agent-c9s:
     <<: *beeai-agent-c9s
     command: ["python", "agents/backport_agent.py"]