Fixed multiple pipeline being fixed simultaneously.

Author: srtab

CHANGELOG.md

@@ -21,10 +21,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - GitLab client now `retry_transient_errors=True` to be more resilient to transient errors.
 - Improved assessment of request for changes on `ReviewAddressorAgent` to allow the agent to reason before responding.
 - Improved response prompt for `ReviewAddressorAgent` to avoid answering like "I'll update the code to replace", as the agent is not able to update the code but only answer questions.
+- Configuration of webhooks now takes a default `EXTERNAL_URL` for `--base-url` to avoid having to pass it on every call, and now configures `pipeline_events` instead of `job_events`. **You must run command `setup_webhooks` to update the webhooks on your GitLab projects.**
+- Turn `PullRequestDescriberAgent` more resilient to errors by defining fallbacks to use a model from other provider.
 
 ### Fixed
 
 - Fixed index updates to branches that don't exist anymore, like when a branch is marked to delete after a merge request is merged: #153.
+- `SnippetReplacerAgent` was replacing multiple snippets when only one was expected. Now it will return an error if multiple snippets are found to instruct the llm to provide a more specific original snippet.
+- `PipelineFixerAgent` was trying to fix multiple jobs from the same stage at the same time, causing multiple fixes being applied simultaneously to the same files which could lead to conflicts or a job being fixed with outdated code. Now it will fix one job at a time. #164
+
+### Removed
+
+- `get_repository_tree` was removed from the `RepoClient` as it's no longer used.
 
 ## [0.1.0-alpha.15] - 2024-12-30
 

daiv/automation/agents/pr_describer/agent.py

@@ -4,7 +4,7 @@
 from langchain_core.runnables import Runnable
 
 from automation.agents import BaseAgent
-from automation.agents.base import GENERIC_COST_EFFICIENT_MODEL_NAME
+from automation.agents.base import CODING_COST_EFFICIENT_MODEL_NAME, GENERIC_COST_EFFICIENT_MODEL_NAME
 from codebase.base import FileChange
 
 from .prompts import human, system
@@ -28,4 +28,8 @@ def compile(self) -> Runnable:
         prompt = ChatPromptTemplate.from_messages([system, human]).partial(
             branch_name_convention=None, extra_details={}
         )
-        return prompt | self.model.with_structured_output(PullRequestDescriberOutput, method="json_schema")
+        return prompt | self.model.with_structured_output(
+            PullRequestDescriberOutput, method="json_schema"
+        ).with_fallbacks([
+            self.get_model(model=CODING_COST_EFFICIENT_MODEL_NAME).with_structured_output(PullRequestDescriberOutput)
+        ])

daiv/automation/agents/snippet_replacer/agent.py

@@ -54,7 +54,10 @@ def _replace_content_snippet(self, input_data: SnippetReplacerInput) -> SnippetR
         if not original_snippet_found:
             return "error: Original snippet not found."
 
-        replaced_content = input_data["content"].replace(original_snippet_found, input_data["replacement_snippet"])
+        if len(original_snippet_found) > 1:
+            return "error: Multiple original snippets found. Please provide a more specific original snippet."
+
+        replaced_content = input_data["content"].replace(original_snippet_found[0], input_data["replacement_snippet"])
         if not replaced_content:
             return "error: Snippet replacement failed."
 

daiv/automation/agents/snippet_replacer/utils.py

@@ -2,7 +2,7 @@
 import re
 
 
-def find_original_snippet(snippet: str, file_contents: str, threshold=0.8, initial_line_threshold=0.9) -> str | None:
+def find_original_snippet(snippet: str, file_contents: str, threshold=0.8, initial_line_threshold=0.9) -> list[str]:
     """
     This function finds the original snippet of code in a file given a snippet and the file contents.
 
@@ -19,18 +19,19 @@ def find_original_snippet(snippet: str, file_contents: str, threshold=0.8, initi
                                         with a line in the file.
 
     Returns:
-        tuple[str, int, int] | None: A tuple containing the original snippet from the file, start index, and end index,
-                                     or None if the snippet could not be found.
+        list[str]: A list of original snippets from the file.
     """
     if snippet.strip() == "":
-        return None
+        return []
 
     snippet_lines = [line for line in snippet.split("\n") if line.strip()]
     file_lines = file_contents.split("\n")
 
     # Find the first non-empty line in the snippet
     first_snippet_line = next((line for line in snippet_lines if line.strip()), "")
 
+    all_matches = []
+
     # Search for a matching initial line in the file
     for start_index, file_line in enumerate(file_lines):
         if compute_similarity(first_snippet_line, file_line) >= initial_line_threshold:
@@ -55,9 +56,9 @@ def find_original_snippet(snippet: str, file_contents: str, threshold=0.8, initi
 
             if snippet_index == len(snippet_lines):
                 # All lines in the snippet have been matched
-                return "\n".join(file_lines[start_index:file_index])
+                all_matches.append("\n".join(file_lines[start_index:file_index]))
 
-    return None
+    return all_matches
 
 
 def compute_similarity(text1: str, text2: str, ignore_whitespace=True) -> float:
@@ -67,13 +68,13 @@ def compute_similarity(text1: str, text2: str, ignore_whitespace=True) -> float:
     difflib.SequenceMatcher uses the Ratcliff/Obershelp algorithm: it computes the doubled number of matching
     characters divided by the total number of characters in the two strings.
 
-    Parameters:
-    text1 (str): The first piece of text.
-    text2 (str): The second piece of text.
-    ignore_whitespace (bool): If True, ignores whitespace when comparing the two pieces of text.
+    Args:
+        text1 (str): The first piece of text.
+        text2 (str): The second piece of text.
+        ignore_whitespace (bool): If True, ignores whitespace when comparing the two pieces of text.
 
     Returns:
-    float: The similarity ratio between the two pieces of text.
+        float: The similarity ratio between the two pieces of text.
     """
     if ignore_whitespace:
         text1 = re.sub(r"\s+", "", text1)

daiv/automation/tools/repository.py

@@ -176,6 +176,10 @@ class ReplaceSnippetInFileTool(BaseRepositoryTool):
         For multiple replacements, call this tool multiple times.
         Do not alter indentation levels unless intentionally modifying code block structures.
         Inspect the code beforehand to understand what exaclty needs to change.
+
+        IMPORTANT:
+        - Provide at least 3 lines before and 3 lines after the snippet you want to replace.
+        - Include unique identifiers such as variable names or function calls that appear only once in the entire file.
         """  # noqa: E501
     )
 

daiv/codebase/api/callbacks_gitlab.py

@@ -1,17 +1,26 @@
 import logging
-import re
 from functools import cached_property
 from typing import Any, Literal
 
 from asgiref.sync import sync_to_async
 
 from codebase.api.callbacks import BaseCallback
-from codebase.api.models import Issue, IssueAction, MergeRequest, Note, NoteableType, NoteAction, Project, User
+from codebase.api.models import (
+    Issue,
+    IssueAction,
+    MergeRequest,
+    Note,
+    NoteableType,
+    NoteAction,
+    Pipeline,
+    PipelineBuild,
+    Project,
+    User,
+)
 from codebase.base import MergeRequest as BaseMergeRequest
 from codebase.clients import RepoClient
 from codebase.tasks import address_issue_task, address_review_task, fix_pipeline_job_task, update_index_repository
 from core.config import RepositoryConfig
-from core.utils import generate_uuid
 
 PIPELINE_JOB_REF_SUFFIX = "refs/merge-requests/"
 
@@ -176,68 +185,73 @@ def related_merge_requests(self) -> list[BaseMergeRequest]:
         return client.get_commit_related_merge_requests(self.project.path_with_namespace, commit_sha=self.checkout_sha)
 
 
-class PipelineJobCallback(BaseCallback):
+class PipelineStatusCallback(BaseCallback):
     """
-    Gitlab Pipeline Job Webhook
+    Gitlab Pipeline Status Webhook
     """
 
-    object_kind: Literal["build"]
+    object_kind: Literal["pipeline"]
     project: Project
-    sha: str
-    ref: str
-    build_id: int
-    build_name: str
-    build_allow_failure: bool
-    build_status: Literal[
-        "created", "pending", "running", "failed", "success", "canceled", "skipped", "manual", "scheduled"
-    ]
-    build_failure_reason: str
+    merge_request: MergeRequest | None = None
+    object_attributes: Pipeline
+    builds: list[PipelineBuild]
 
     def model_post_init(self, __context: Any):
         self._repo_config = RepositoryConfig.get_config(self.project.path_with_namespace)
 
     def accept_callback(self) -> bool:
         """
-        Accept the webhook if the pipeline job failed due to a script failure and there are related merge requests.
+        Accept callback if the pipeline failed and has a failed build to fix.
         """
         return (
-            not self.build_allow_failure
-            and self._repo_config.features.autofix_pipeline_enabled
-            and self.build_status == "failed"
-            # Only fix pipeline jobs that failed due to a script failure.
-            and self.build_failure_reason == "script_failure"
-            # Only fix pipeline jobs of the latest commit of the merge request.
-            and self.merge_request is not None
-            and self.merge_request.is_daiv()
-            and self.merge_request.sha == self.sha
+            self._repo_config.features.autofix_pipeline_enabled
+            and self.object_attributes.status == "failed"
+            and self._first_failed_build is not None
+            and self._merge_request is not None
+            and self._merge_request.is_daiv()
         )
 
     async def process_callback(self):
         """
-        Trigger the task to fix the pipeline job.
+        Trigger the task to fix the pipeline failed build.
+
+        Only one build is fixed at a time to avoid two or more fixes being applied simultaneously to the same files,
+        which could lead to conflicts or a job being fixed with outdated code.
         """
-        if self.merge_request:
+        if self.merge_request is not None and self._first_failed_build is not None:
             await sync_to_async(
                 fix_pipeline_job_task.si(
                     repo_id=self.project.path_with_namespace,
                     ref=self.merge_request.source_branch,
-                    merge_request_id=self.merge_request.merge_request_id,
-                    job_id=self.build_id,
-                    job_name=self.build_name,
-                    thread_id=generate_uuid(
-                        f"{self.project.path_with_namespace}{self.merge_request.merge_request_id}{self.build_name}"
-                    ),
+                    merge_request_id=self.merge_request.iid,
+                    job_id=self._first_failed_build.id,
+                    job_name=self._first_failed_build.name,
                 ).delay
             )()
 
     @cached_property
-    def merge_request(self) -> BaseMergeRequest | None:
+    def _merge_request(self) -> BaseMergeRequest | None:
         """
-        Get the merge request related to the job.
+        Get the merge request related to the pipeline to obtain associated labels and infer if is a DAIV MR.
         """
-        # The ref points to the source branch of a merge request.
-        match = re.search(rf"{PIPELINE_JOB_REF_SUFFIX}(\d+)(?:/\w+)?$", self.ref)
-        if match:
-            client = RepoClient.create_instance()
-            return client.get_merge_request(self.project.path_with_namespace, int(match.group(1)))
+        client = RepoClient.create_instance()
+        if self.merge_request is not None:
+            return client.get_merge_request(self.project.path_with_namespace, self.merge_request.iid)
         return None
+
+    @cached_property
+    def _first_failed_build(self) -> PipelineBuild | None:
+        """
+        Get the first failed build of the pipeline.
+        """
+        return next(
+            (
+                build
+                for build in self.builds
+                if build.status == "failed"
+                and not build.manual
+                and not build.allow_failure
+                and build.failure_reason == "script_failure"
+            ),
+            None,
+        )

daiv/codebase/api/models.py

@@ -1,6 +1,7 @@
 from enum import StrEnum
 from typing import Literal
 
+from ninja import Field
 from pydantic import BaseModel
 
 from core.constants import BOT_LABEL
@@ -53,13 +54,13 @@ class MergeRequest(BaseModel):
     id: int
     iid: int
     title: str
-    description: str
+    description: str | None = None
     state: str
-    work_in_progress: bool
+    work_in_progress: bool | None = None
     source_branch: str
     target_branch: str
-    assignee_id: int | None
-    labels: list[Label]
+    assignee_id: int | None = None
+    labels: list[Label] = Field(default_factory=list)
 
     def is_daiv(self) -> bool:
         """
@@ -164,3 +165,40 @@ class User(BaseModel):
     name: str
     username: str
     email: str
+
+
+class PipelineBuild(BaseModel):
+    """
+    Gitlab Pipeline Build
+    """
+
+    id: int
+    status: Literal["created", "pending", "running", "failed", "success", "canceled", "skipped", "manual", "scheduled"]
+    name: str
+    stage: str
+    allow_failure: bool
+    failure_reason: str | None = None
+    manual: bool
+
+
+class Pipeline(BaseModel):
+    """
+    Gitlab Pipeline
+    """
+
+    id: int
+    ref: str
+    sha: str
+    status: Literal[
+        "created",
+        "waiting_for_resource",
+        "preparing",
+        "pending",
+        "running",
+        "failed",
+        "success",
+        "canceled",
+        "skipped",
+        "manual",
+        "scheduled",
+    ]

daiv/codebase/api/views.py

@@ -3,15 +3,15 @@
 from ninja import Router
 
 from .callbacks import UnprocessableEntityResponse
-from .callbacks_gitlab import IssueCallback, NoteCallback, PipelineJobCallback, PushCallback
+from .callbacks_gitlab import IssueCallback, NoteCallback, PipelineStatusCallback, PushCallback
 
 logger = logging.getLogger("daiv.webhooks")
 
 router = Router()
 
 
 @router.post("/callbacks/gitlab/", response={204: None, 423: UnprocessableEntityResponse})
-async def gitlab_callback(request, payload: IssueCallback | NoteCallback | PushCallback | PipelineJobCallback):
+async def gitlab_callback(request, payload: IssueCallback | NoteCallback | PushCallback | PipelineStatusCallback):
     """
     GitLab callback endpoint for processing callbacks.
     """