diff --git a/ymir/tools/base.py b/ymir/tools/base.py index 082673d55..88cda79f6 100644 --- a/ymir/tools/base.py +++ b/ymir/tools/base.py @@ -15,19 +15,49 @@ @contextmanager -def tool_error_context(error_message: str, **additional_context): +def tool_error_context( + error_message: str, + include_exception_message_for: tuple[type[Exception], ...] = (), + **additional_context, +): + """Context manager for unified tool error handling with observability. + + Catches exceptions and wraps them as ToolErrorWithContext with a clean + LLM-facing message while preserving more specific details (exception type, + exception message and additional context) for logs and traces. + + Args: + error_message: Clean error message shown to the LLM. + include_exception_message_for: Exception types whose exception messages + should be appended to the LLM-facing error_message. + **additional_context: Key-value pairs for observability, automatically + redacted for credentials. + + Raises: + ToolErrorWithContext: Wraps any caught exception (except ToolErrorWithContext + which passes through unchanged) and adds provided additional context + for observability. + """ try: yield except ToolErrorWithContext: raise except Exception as e: + if isinstance(e, include_exception_message_for): + error_message = f"{error_message}: {redact_credentials(str(e))}" + additional_context["exception"] = f"{type(e).__name__}: {e}" - redacted_additional_context = {k: redact_credentials(str(v)) for k, v in additional_context.items()} raise ToolErrorWithContext( - error_message, cause=e, additional_context=redacted_additional_context + error_message, + cause=e, + additional_context=make_additional_context(**additional_context), ) from e +def make_additional_context(**additional_context) -> dict[str, str]: + return {k: redact_credentials(str(v)) for k, v in additional_context.items()} + + class CloneableTool(Tool[TInput, TRunOptions, TOutput]): """Tool with clone method and built-in timeout handling""" diff --git a/ymir/tools/privileged/copr.py b/ymir/tools/privileged/copr.py index 919b1d84b..8152ef44b 100644 --- a/ymir/tools/privileged/copr.py +++ b/ymir/tools/privileged/copr.py @@ -14,7 +14,6 @@ from beeai_framework.emitter import Emitter from beeai_framework.tools import ( JSONToolOutput, - ToolError, ToolRunOptions, ) from copr.v3 import BuildProxy, ProjectChrootProxy, ProjectProxy @@ -22,11 +21,13 @@ from pydantic import BaseModel, Field from ymir.common import load_rhel_config -from ymir.common.base_utils import KerberosError, init_kerberos_ticket +from ymir.common.base_utils import init_kerberos_ticket from ymir.common.validators import AbsolutePath from ymir.common.version_utils import parse_branch_name from ymir.tools.base import CloneableTool as Tool +from ymir.tools.base import make_additional_context, tool_error_context from ymir.tools.constants import AIOHTTP_TIMEOUT, YMIR_USER_AGENT +from ymir.tools.errors import ToolErrorWithContext from ymir.tools.http import aiohttp_get_with_retries COPR_CONFIG = { @@ -124,15 +125,11 @@ async def _run( srpm_path = tool_input.srpm_path dist_git_branch = tool_input.dist_git_branch jira_issue = tool_input.jira_issue - try: + with tool_error_context("Failed to initialize Kerberos ticket"): principal = await init_kerberos_ticket() - except KerberosError as e: - raise ToolError(f"Failed to initialize Kerberos ticket: {e}") from e copr_user = principal.split("@", maxsplit=1)[0] - try: + with tool_error_context("Failed to read SRPM header", srpm_path=str(srpm_path)): exclusive_arches = await self.get_exclusive_arches(srpm_path) - except Exception as e: - raise ToolError(f"Failed to read SRPM header: {e}") from e # build for the fastest supported arch (see COPR_ARCH_PREFERENCE); # default to x86_64 when the package has no ExclusiveArch build_arch = next( @@ -141,11 +138,13 @@ async def _run( ) rhel_config = await load_rhel_config() upcoming_z_streams = rhel_config.get("upcoming_z_streams", {}) - try: + with tool_error_context( + "Failed to deduce Copr chroot", + dist_git_branch=dist_git_branch, + build_arch=build_arch, + ): chroot_base, majorver = await self.branch_to_chroot(dist_git_branch, upcoming_z_streams) chroot = f"{chroot_base}-{build_arch}" - except ValueError as e: - raise ToolError(f"Failed to deduce Copr chroot: {e}") from e logger.info(f"Connecting to Copr API at {COPR_CONFIG['copr_url']} for project creation/update") project_proxy = ProjectProxy({"username": copr_user, **COPR_CONFIG}) kwargs = { @@ -164,14 +163,30 @@ async def _run( kwargs["chroots"] = sorted(set(project.chroot_repos.keys()) | {chroot}) await _copr_api_call(project_proxy.edit, **kwargs) except Exception as e: - raise ToolError(f"Failed to create or update Copr project: {_copr_error_detail(e)}") from e + raise ToolErrorWithContext( + "Failed to create or update Copr project", + cause=e, + additional_context=make_additional_context( + copr_user=copr_user, + project=jira_issue, + chroot=chroot, + exception=_copr_error_detail(e), + ), + ) from e if chroot.startswith("custom-"): # make sure the chroot has access to corresponding buildroot repository logger.info(f"Connecting to Copr API to update chroot configuration for {chroot}") chroot_proxy = ProjectChrootProxy({"username": copr_user, **COPR_CONFIG}) bootstrap_image = f"registry.access.redhat.com/ubi{majorver}/ubi" if not (internal_repos_host := rhel_config.get("internal_repos_host")): - raise ToolError("Internal repos host not configured") + raise ToolErrorWithContext( + "Internal repos host not configured", + additional_context=make_additional_context( + copr_user=copr_user, + project=jira_issue, + chroot=chroot, + ), + ) buildroot_repo_url = urljoin( internal_repos_host, f"brewroot/repos/{dist_git_branch}-z-build/latest/{build_arch}", @@ -207,7 +222,16 @@ async def _run( **kwargs, ) except Exception as e: - raise ToolError(f"Failed to update Copr chroot: {_copr_error_detail(e)}") from e + raise ToolErrorWithContext( + "Failed to update Copr chroot", + cause=e, + additional_context=make_additional_context( + copr_user=copr_user, + project=jira_issue, + chroot=chroot, + exception=_copr_error_detail(e), + ), + ) from e logger.info(f"Connecting to Copr API to submit build for {srpm_path}") build_proxy = BuildProxy({"username": copr_user, **COPR_CONFIG}) try: @@ -218,10 +242,19 @@ async def _run( path=str(srpm_path), buildopts={"chroots": [chroot], "timeout": COPR_BUILD_TIMEOUT}, ) - except Exception as e: - raise ToolError(f"Failed to submit Copr build: {_copr_error_detail(e)}") from e - else: logger.info(f"{jira_issue}: build of {srpm_path} in {chroot} started: {build.id:08d}") + except Exception as e: + raise ToolErrorWithContext( + "Failed to submit Copr build", + cause=e, + additional_context=make_additional_context( + copr_user=copr_user, + project=jira_issue, + chroot=chroot, + srpm_path=str(srpm_path), + exception=_copr_error_detail(e), + ), + ) from e async def get_artifacts_urls(build): if build.source_package and (package := build.source_package.get("name")): @@ -370,21 +403,26 @@ async def _run( for url in artifacts_urls: logger.info(f"Downloading build artifact from: {url}") try: - async with aiohttp_get_with_retries(session, url) as response: - if response.status < 400: - target = Path(Path(urlparse(url).path).name) - content = await response.read() - if ".log" in target.suffixes: - if content.startswith(b"\x1f\x8b"): - # decompress logs on-the-fly - content = gzip.decompress(content) - if target.suffix == ".gz": - target = target.with_suffix("") - (target_path / target).write_bytes(content) - else: - raise ValueError(f"{response.status} {response.reason}") - except Exception as e: + with tool_error_context( + "Failed to download build artifact", + include_exception_message_for=(ValueError,), + artifacts_url=url, + ): + async with aiohttp_get_with_retries(session, url) as response: + if response.status < 400: + target = Path(Path(urlparse(url).path).name) + content = await response.read() + if ".log" in target.suffixes: + if content.startswith(b"\x1f\x8b"): + # decompress logs on-the-fly + content = gzip.decompress(content) + if target.suffix == ".gz": + target = target.with_suffix("") + (target_path / target).write_bytes(content) + else: + raise ValueError(f"{response.status} {response.reason}") + except Exception: # Cleanup temporary dir rmtree(target_path) - raise ToolError(f"Failed to download {url}: {e}") from e + raise return DownloadArtifactsToolOutput(result=DownloadArtifactsResult(target_path=target_path)) diff --git a/ymir/tools/privileged/distgit.py b/ymir/tools/privileged/distgit.py index ddf824931..c48dbf8fa 100644 --- a/ymir/tools/privileged/distgit.py +++ b/ymir/tools/privileged/distgit.py @@ -15,11 +15,12 @@ from pydantic import BaseModel, Field from specfile import Specfile -from ymir.common.base_utils import KerberosError, init_kerberos_ticket +from ymir.common.base_utils import init_kerberos_ticket from ymir.common.utils import get_latest_candidate_build, get_latest_z_pending_build from ymir.common.version_utils import is_older_zstream, parse_zstream_branch_name from ymir.tools.base import CloneableTool as Tool -from ymir.tools.base import tool_error_context +from ymir.tools.base import make_additional_context, tool_error_context +from ymir.tools.errors import ToolErrorWithContext from ymir.tools.privileged.utils import sanitize_url logger = logging.getLogger(__name__) @@ -183,11 +184,8 @@ async def _run( ) -> StringToolOutput: package = tool_input.package branch = tool_input.branch - try: + with tool_error_context("Failed to initialize Kerberos ticket"): principal = await init_kerberos_ticket() - except KerberosError as e: - logger.error("Kerberos initialization failed: %s", e) - raise ToolError("Failed to initialize Kerberos ticket") from e username = principal.split("@", maxsplit=1)[0] token = os.environ["GITLAB_TOKEN"] gitlab_repo_url = f"https://oauth2:{token}@gitlab.com/redhat/rhel/rpms/{package}" @@ -200,7 +198,12 @@ async def _run( result=f"Z-Stream branch {branch} already exists, no need to create it" ) with ( - tool_error_context("Failed to create Z-Stream branch", package=package, branch=branch), + tool_error_context( + "Failed to create Z-Stream branch", + include_exception_message_for=(ToolError,), + package=package, + branch=branch, + ), tempfile.TemporaryDirectory() as path, ): # Username is taken from the Kerberos principal and embedded in @@ -257,12 +260,6 @@ async def _clone(): return StringToolOutput(result=f"Successfully created Z-Stream branch {branch}") except git.exc.GitCommandError as e: if not _is_transient_git_error(e): - logger.error( - "Failed to poll GitLab mirror for %s/%s: %s", - package, - branch, - sanitize_url(str(e)), - ) raise ToolError("Failed to poll GitLab mirror") from e logger.warning(f"Transient error polling GitLab mirror sync: {sanitize_url(str(e))}") elapsed = int(time.monotonic() - start_time) @@ -270,10 +267,11 @@ async def _clone(): f"Waiting for GitLab mirror sync of {package} branch {branch} ({elapsed}s elapsed)" ) await asyncio.sleep(30) - logger.error( - "GitLab mirror sync timed out for %s branch %s after %ds", - package, - branch, - SYNC_TIMEOUT, + raise ToolErrorWithContext( + "GitLab mirror sync timed out", + additional_context=make_additional_context( + package=package, + branch=branch, + timeout=SYNC_TIMEOUT, + ), ) - raise ToolError("GitLab mirror sync timed out") diff --git a/ymir/tools/privileged/errata.py b/ymir/tools/privileged/errata.py index 29a350509..e1448808c 100644 --- a/ymir/tools/privileged/errata.py +++ b/ymir/tools/privileged/errata.py @@ -163,7 +163,11 @@ async def _run( erratum_id = erratum_id.rstrip("/").split("/")[-1] logger.info("Getting erratum %s (full=%s)", erratum_id, tool_input.full) - with tool_error_context(f"Failed to get erratum {erratum_id}", full=tool_input.full): + with tool_error_context( + f"Failed to get erratum {erratum_id}", + erratum_id=erratum_id, + full=tool_input.full, + ): erratum = await asyncio.to_thread(_get_erratum, erratum_id, full=tool_input.full) return JSONToolOutput(result=erratum.model_dump(mode="json")) @@ -198,7 +202,11 @@ async def _run( package_name = tool_input.package_name logger.info("Getting build NVR for %s in erratum %s", package_name, erratum_id) - with tool_error_context(f"Failed to get build NVR for {package_name} in erratum {erratum_id}"): + with tool_error_context( + f"Failed to get build NVR for {package_name} in erratum {erratum_id}", + erratum_id=erratum_id, + package_name=package_name, + ): builds_by_release = await asyncio.to_thread(_et_api_get, f"erratum/{erratum_id}/builds_list") for release_info in builds_by_release.values(): for builds_map in release_info["builds"]: @@ -516,7 +524,10 @@ async def _run( ) -> JSONToolOutput[dict[str, Any]]: erratum_id = tool_input.erratum_id logger.info("Getting transition rules for erratum %s", erratum_id) - with tool_error_context(f"Failed to get transition rules for erratum {erratum_id}"): + with tool_error_context( + f"Failed to get transition rules for erratum {erratum_id}", + erratum_id=erratum_id, + ): rule_set = await asyncio.to_thread(_get_erratum_transition_rules, erratum_id) return JSONToolOutput(result=rule_set.model_dump(mode="json")) @@ -546,7 +557,10 @@ async def _run( ) -> JSONToolOutput[dict[str, Any]]: erratum_id = tool_input.erratum_id logger.info("Getting build map for erratum %s", erratum_id) - with tool_error_context(f"Failed to get build map for erratum {erratum_id}"): + with tool_error_context( + f"Failed to get build map for erratum {erratum_id}", + erratum_id=erratum_id, + ): build_map = await asyncio.to_thread(_get_erratum_build_map, erratum_id) return JSONToolOutput(result=build_map.model_dump(mode="json")) @@ -580,7 +594,11 @@ async def _run( erratum_id = tool_input.erratum_id package_name = tool_input.package_name logger.info("Getting previous erratum for %s in erratum %s", package_name, erratum_id) - with tool_error_context(f"Failed to get previous erratum for {package_name} in erratum {erratum_id}"): + with tool_error_context( + f"Failed to get previous erratum for {package_name} in erratum {erratum_id}", + erratum_id=erratum_id, + package_name=package_name, + ): prev_id, prev_nvr = await asyncio.to_thread(_get_previous_erratum, erratum_id, package_name) return JSONToolOutput(result={"id": prev_id, "nvr": prev_nvr}) @@ -610,7 +628,10 @@ async def _run( ) -> JSONToolOutput[dict[str, Any]]: erratum_id = tool_input.erratum_id logger.info("Getting stage push details for erratum %s", erratum_id) - with tool_error_context(f"Failed to get stage push details for erratum {erratum_id}"): + with tool_error_context( + f"Failed to get stage push details for erratum {erratum_id}", + erratum_id=erratum_id, + ): details = await asyncio.to_thread(_get_erratum_stage_push_details, erratum_id) return JSONToolOutput(result=details.model_dump(mode="json")) @@ -642,7 +663,10 @@ async def _run( result=f"Dry run, not pushing erratum {erratum_id} to stage (this is expected, not an error)" ) logger.info("Pushing erratum %s to stage", erratum_id) - with tool_error_context(f"Failed to push erratum {erratum_id} to stage"): + with tool_error_context( + f"Failed to push erratum {erratum_id} to stage", + erratum_id=erratum_id, + ): await asyncio.to_thread(_et_api_post, f"erratum/{erratum_id}/push", {"defaults": "stage"}) return StringToolOutput(result=f"Successfully pushed erratum {erratum_id} to stage") @@ -677,7 +701,11 @@ async def _run( f"(this is expected, not an error)" ) logger.info("Changing state of erratum %s to %s", erratum_id, new_state) - with tool_error_context(f"Failed to change state of erratum {erratum_id} to {new_state}"): + with tool_error_context( + f"Failed to change state of erratum {erratum_id} to {new_state}", + erratum_id=erratum_id, + new_state=new_state, + ): await asyncio.to_thread( _et_api_post, f"erratum/{erratum_id}/change_state", @@ -715,7 +743,10 @@ async def _run( result=f"Dry run, not adding comment to erratum {erratum_id} (this is expected, not an error)" ) logger.info("Adding comment to erratum %s", erratum_id) - with tool_error_context(f"Failed to add comment to erratum {erratum_id}"): + with tool_error_context( + f"Failed to add comment to erratum {erratum_id}", + erratum_id=erratum_id, + ): await asyncio.to_thread( _et_api_post, f"erratum/{erratum_id}/add_comment", @@ -754,7 +785,10 @@ async def _run( f"(this is expected, not an error)" ) logger.info("Refreshing security alerts for erratum %s", erratum_id) - with tool_error_context(f"Failed to refresh security alerts for erratum {erratum_id}"): + with tool_error_context( + f"Failed to refresh security alerts for erratum {erratum_id}", + erratum_id=erratum_id, + ): await asyncio.to_thread( _et_api_post, f"erratum/{erratum_id}/security_alerts/refresh", diff --git a/ymir/tools/privileged/gitlab.py b/ymir/tools/privileged/gitlab.py index f6636cb1d..84bd19162 100644 --- a/ymir/tools/privileged/gitlab.py +++ b/ymir/tools/privileged/gitlab.py @@ -21,7 +21,7 @@ ToolRunOptions, ) from mcp.server.lowlevel.server import request_ctx -from ogr.exceptions import GitlabAPIException, OgrException +from ogr.exceptions import GitlabAPIException from ogr.factory import get_project from ogr.services.gitlab.project import GitlabProject from ogr.services.gitlab.pull_request import GitlabPullRequest @@ -37,7 +37,9 @@ ) from ymir.common.validators import AbsolutePath from ymir.tools.base import CloneableTool as Tool +from ymir.tools.base import make_additional_context, tool_error_context from ymir.tools.constants import AIOHTTP_TIMEOUT, YMIR_USER_AGENT +from ymir.tools.errors import ToolErrorWithContext from ymir.tools.http import aiohttp_get_with_retries from ymir.tools.privileged.utils import clean_stale_repositories, sanitize_url @@ -390,61 +392,66 @@ async def _run( ) -> StringToolOutput: repository = tool_input.repository logger.info(f"Connecting to GitLab API to fork repository: {repository}") - project = await asyncio.to_thread(get_project, url=repository, token=os.getenv("GITLAB_TOKEN")) - if not project: - raise ToolError("Failed to get the specified repository") - - if urlparse(project.service.instance_url).hostname != "gitlab.com": - raise ToolError("Unexpected git forge, expected gitlab.com/redhat") - - namespace = project.gitlab_repo.namespace["full_path"].split("/") - if not namespace or namespace[0] != "redhat": - raise ToolError("Unexpected GitLab project, expected gitlab.com/redhat") - - fork_namespace = os.getenv("FORK_NAMESPACE") - - def get_fork(): - target = fork_namespace or project.service.user.get_username() - for fork in project.get_forks(): - if fork.gitlab_repo.namespace["full_path"] == target: + with tool_error_context( + "Failed to fork repository", + include_exception_message_for=(ToolError,), + repository=repository, + ): + project = await asyncio.to_thread(get_project, url=repository, token=os.getenv("GITLAB_TOKEN")) + if not project: + raise ToolError("Failed to get the specified repository") + + if urlparse(project.service.instance_url).hostname != "gitlab.com": + raise ToolError("Unexpected git forge, expected gitlab.com/redhat") + + namespace = project.gitlab_repo.namespace["full_path"].split("/") + if not namespace or namespace[0] != "redhat": + raise ToolError("Unexpected GitLab project, expected gitlab.com/redhat") + + fork_namespace = os.getenv("FORK_NAMESPACE") + + def get_fork(): + target = fork_namespace or project.service.user.get_username() + for fork in project.get_forks(): + if fork.gitlab_repo.namespace["full_path"] == target: + return fork + return None + + if fork := await asyncio.to_thread(get_fork): + return StringToolOutput(result=fork.get_git_urls()["git"]) + + if os.getenv("DRY_RUN", "False").lower() == "true": + logger.info("DRY_RUN is set, skipping fork creation — returning original repo URL") + return StringToolOutput(result=project.get_git_urls()["git"]) + + def create_fork(): + prefix = "_".join(ns.replace("centos-stream", "centos") for ns in namespace[1:]) + fork_name = (f"{prefix}_" if prefix else "") + project.gitlab_repo.name + data = {"name": fork_name, "path": fork_name} + if fork_namespace: + data["namespace"] = fork_namespace + try: + fork = project.gitlab_repo.forks.create(data=data) + except GitlabAPIException: + if not fork_namespace: + raise + logger.info("Fork creation failed, checking if it was created by another deployment") + fork = get_fork() + if not fork: + raise return fork - return None + return GitlabProject( + namespace=fork.namespace["full_path"], + service=project.service, + repo=fork.path, + ) - if fork := await asyncio.to_thread(get_fork): + fork = await asyncio.to_thread(create_fork) + if not fork: + raise ToolError("Failed to fork the specified repository") + await asyncio.to_thread(_wait_for_fork_ready, fork) return StringToolOutput(result=fork.get_git_urls()["git"]) - if os.getenv("DRY_RUN", "False").lower() == "true": - logger.info("DRY_RUN is set, skipping fork creation — returning original repo URL") - return StringToolOutput(result=project.get_git_urls()["git"]) - - def create_fork(): - prefix = "_".join(ns.replace("centos-stream", "centos") for ns in namespace[1:]) - fork_name = (f"{prefix}_" if prefix else "") + project.gitlab_repo.name - data = {"name": fork_name, "path": fork_name} - if fork_namespace: - data["namespace"] = fork_namespace - try: - fork = project.gitlab_repo.forks.create(data=data) - except GitlabAPIException: - if not fork_namespace: - raise - logger.info("Fork creation failed, checking if it was created by another deployment") - fork = get_fork() - if not fork: - raise - return fork - return GitlabProject( - namespace=fork.namespace["full_path"], - service=project.service, - repo=fork.path, - ) - - fork = await asyncio.to_thread(create_fork) - if not fork: - raise ToolError("Failed to fork the specified repository") - await asyncio.to_thread(_wait_for_fork_ready, fork) - return StringToolOutput(result=fork.get_git_urls()["git"]) - class OpenMergeRequestToolInput(BaseModel): fork_url: str = Field(description="URL of the fork to open the MR from") @@ -512,29 +519,38 @@ async def _run( source = tool_input.source labels = tool_input.labels logger.info(f"Connecting to GitLab API to open merge request from fork: {fork_url}") - project = await asyncio.to_thread(get_project, url=fork_url, token=os.getenv("GITLAB_TOKEN")) - if not project: - raise ToolError("Failed to get the specified fork") - is_new_mr = True - try: - pr = await asyncio.to_thread(self._create_mr, project, title, description, target, source, labels) - except GitlabAPIException as ex: - logger.info("Gitlab API exception: %s", ex) - if ex.response_code == 409: - prs = await asyncio.to_thread(project.parent.get_pr_list) - for pr in prs: - if pr.source_branch == source and pr.target_branch == target: - logger.info("Reusing existing MR %s", pr) - pr.description = description - pr.title = title - is_new_mr = False - break + with tool_error_context( + "Failed to open merge request", + include_exception_message_for=(ToolError,), + fork_url=fork_url, + source=source, + target=target, + ): + project = await asyncio.to_thread(get_project, url=fork_url, token=os.getenv("GITLAB_TOKEN")) + if not project: + raise ToolError("Failed to get the specified fork") + is_new_mr = True + try: + pr = await asyncio.to_thread( + self._create_mr, project, title, description, target, source, labels + ) + except GitlabAPIException as ex: + logger.info("Gitlab API exception: %s", ex) + if ex.response_code == 409: + prs = await asyncio.to_thread(project.parent.get_pr_list) + for pr in prs: + if pr.source_branch == source and pr.target_branch == target: + logger.info("Reusing existing MR %s", pr) + pr.description = description + pr.title = title + is_new_mr = False + break + else: + raise else: raise - else: - raise - if not pr: - raise ToolError("Failed to open the merge request") + if not pr: + raise ToolError("No merge request was created or found") return JSONToolOutput(result=OpenMergeRequestResult(url=pr.url, is_new_mr=is_new_mr)) @@ -570,7 +586,12 @@ async def _run( repository_url = f"https://gitlab.com/redhat/rhel/rpms/{package}" logger.info(f"Connecting to GitLab API to get branches for package: {repository_url}") - try: + with tool_error_context( + f"Failed to get branches for package {package}", + include_exception_message_for=(ToolError,), + package=package, + repository_url=repository_url, + ): project = await asyncio.to_thread( get_project, url=repository_url, token=os.getenv("GITLAB_TOKEN") ) @@ -578,12 +599,9 @@ async def _run( raise ToolError(f"Failed to get repository for package: {package}") branches = await asyncio.to_thread(project.get_branches) - logger.info(f"Found {len(branches)} branches for package {package}: {branches}") - return JSONToolOutput(result=branches) - except OgrException as ex: - logger.warning(f"Failed to get branches for package {package}: {ex}") - raise ToolError(f"Failed to get branches for package {package}: {ex}") from ex + logger.info(f"Found {len(branches)} branches for package {package}: {branches}") + return JSONToolOutput(result=branches) class CloneRepositoryToolInput(BaseModel): @@ -621,52 +639,59 @@ async def _run( branch = tool_input.branch clone_path = tool_input.clone_path - basepath = Path(os.getenv("GIT_REPO_BASEPATH", "/git-repos")).resolve() - resolved = clone_path.resolve() - if resolved == basepath or not resolved.is_relative_to(basepath): - raise ToolError(f"clone_path must be under {basepath} (the shared volume). Got: {clone_path}") - clone_path = resolved - - await clean_stale_repositories() - - auth_args = _get_git_auth_args(repository) - git_env = _get_mock_git_env() - - safe_url = sanitize_url(repository) - - await asyncio.to_thread(_remove_existing_clone_path, clone_path) - - if branch: - clone_path.mkdir(parents=True, exist_ok=True) - await _run_git_cmd( - ["git", "init"], - label=f"git init {clone_path}", - cwd=clone_path, - env=git_env, - timeout=None, - ) + with tool_error_context( + "Failed to clone repository", + include_exception_message_for=(ToolError,), + repository=repository, + branch=str(branch), + clone_path=str(clone_path), + ): + basepath = Path(os.getenv("GIT_REPO_BASEPATH", "/git-repos")).resolve() + resolved = clone_path.resolve() + if resolved == basepath or not resolved.is_relative_to(basepath): + raise ToolError(f"clone_path must be under {basepath} (the shared volume). Got: {clone_path}") + clone_path = resolved + + await clean_stale_repositories() + + auth_args = _get_git_auth_args(repository) + git_env = _get_mock_git_env() + + safe_url = sanitize_url(repository) + + await asyncio.to_thread(_remove_existing_clone_path, clone_path) + + if branch: + clone_path.mkdir(parents=True, exist_ok=True) + await _run_git_cmd( + ["git", "init"], + label=f"git init {clone_path}", + cwd=clone_path, + env=git_env, + timeout=None, + ) - await _run_git_cmd( - ["git", *auth_args, "fetch", repository, f"{branch}:refs/heads/{branch}"], - label=f"git fetch {safe_url} branch={branch}", - cwd=clone_path, - env=git_env, - ) + await _run_git_cmd( + ["git", *auth_args, "fetch", repository, f"{branch}:refs/heads/{branch}"], + label=f"git fetch {safe_url} branch={branch}", + cwd=clone_path, + env=git_env, + ) - await _run_git_cmd( - ["git", "checkout", branch], - label=f"git checkout branch={branch}", - cwd=clone_path, - env=git_env, - timeout=None, - ) - else: - clone_path.parent.mkdir(parents=True, exist_ok=True) - await _run_git_cmd( - ["git", *auth_args, "clone", repository, str(clone_path)], - label=f"git clone {safe_url}", - env=git_env, - ) + await _run_git_cmd( + ["git", "checkout", branch], + label=f"git checkout branch={branch}", + cwd=clone_path, + env=git_env, + timeout=None, + ) + else: + clone_path.parent.mkdir(parents=True, exist_ok=True) + await _run_git_cmd( + ["git", *auth_args, "clone", repository, str(clone_path)], + label=f"git clone {safe_url}", + env=git_env, + ) return StringToolOutput(result=f"Successfully cloned the specified repository to {clone_path}") @@ -703,20 +728,29 @@ async def _run( clone_path = tool_input.clone_path force = tool_input.force safe_url = sanitize_url(repository) - auth_args = _get_git_auth_args(repository) - git_env = _get_mock_git_env() - - command = ["git", *auth_args, "push", repository, branch] - if force: - command.append("--force") - - await _run_git_cmd( - command, - label=f"git push {safe_url} branch={branch} force={force}", - cwd=clone_path, - env=git_env, - timeout=None, - ) + + with tool_error_context( + "Failed to push to remote repository", + include_exception_message_for=(ToolError,), + repository=safe_url, + branch=branch, + clone_path=str(clone_path), + force=str(force), + ): + auth_args = _get_git_auth_args(repository) + git_env = _get_mock_git_env() + + command = ["git", *auth_args, "push", repository, branch] + if force: + command.append("--force") + + await _run_git_cmd( + command, + label=f"git push {safe_url} branch={branch} force={force}", + cwd=clone_path, + env=git_env, + timeout=None, + ) return StringToolOutput(result=f"Successfully pushed the specified branch to {safe_url}") @@ -752,16 +786,24 @@ async def _run( branch = tool_input.branch clone_path = tool_input.clone_path safe_url = sanitize_url(repository) - auth_args = _get_git_auth_args(repository) - git_env = _get_mock_git_env() - - await _run_git_cmd( - ["git", *auth_args, "fetch", repository, f"{branch}:refs/heads/{branch}"], - label=f"git fetch {safe_url} branch={branch}", - cwd=clone_path, - env=git_env, - timeout=None, - ) + + with tool_error_context( + "Failed to fetch branch", + include_exception_message_for=(ToolError,), + repository=safe_url, + branch=branch, + clone_path=str(clone_path), + ): + auth_args = _get_git_auth_args(repository) + git_env = _get_mock_git_env() + + await _run_git_cmd( + ["git", *auth_args, "fetch", repository, f"{branch}:refs/heads/{branch}"], + label=f"git fetch {safe_url} branch={branch}", + cwd=clone_path, + env=git_env, + timeout=None, + ) return StringToolOutput(result=f"Successfully fetched branch {branch} from {safe_url}") @@ -793,17 +835,18 @@ async def _run( ) -> StringToolOutput: merge_request_url = tool_input.merge_request_url labels = tool_input.labels - try: + with tool_error_context( + "Failed to add labels to merge request", + include_exception_message_for=(ValueError,), + merge_request_url=merge_request_url, + labels=str(labels), + ): mr = await _get_merge_request_from_url(merge_request_url) for label in labels: await asyncio.to_thread(mr.add_label, label) - return StringToolOutput( - result=f"Successfully added labels {labels} to merge request {merge_request_url}" - ) - except ToolError: - raise - except Exception as e: - raise ToolError(f"Failed to add labels to merge request: {e}") from e + return StringToolOutput( + result=f"Successfully added labels {labels} to merge request {merge_request_url}" + ) class SetMergeRequestReviewersToolInput(BaseModel): @@ -833,7 +876,12 @@ async def _run( ) -> StringToolOutput: merge_request_url = tool_input.merge_request_url reviewer_ids = tool_input.reviewer_ids - try: + with tool_error_context( + "Failed to set reviewers on merge request", + include_exception_message_for=(ValueError,), + merge_request_url=merge_request_url, + reviewer_ids=str(reviewer_ids), + ): mr = await _get_merge_request_from_url(merge_request_url) def set_reviewers(): @@ -841,11 +889,9 @@ def set_reviewers(): mr._raw_pr.save() await asyncio.to_thread(set_reviewers) - return StringToolOutput( - result=f"Successfully set reviewers {reviewer_ids} on merge request {merge_request_url}" - ) - except Exception as e: - raise ToolError(f"Failed to set reviewers on merge request: {e}") from e + return StringToolOutput( + result=f"Successfully set reviewers {reviewer_ids} on merge request {merge_request_url}" + ) class ResolveReviewersToolInput(BaseModel): @@ -875,7 +921,12 @@ async def _run( ) -> JSONToolOutput[list[int]]: from ymir.tools.privileged.reviewer_resolver import resolve_reviewers - reviewer_ids = await resolve_reviewers(tool_input.package, tool_input.dist_git_branch) + with tool_error_context( + "Failed to resolve reviewers", + package=tool_input.package, + dist_git_branch=tool_input.dist_git_branch, + ): + reviewer_ids = await resolve_reviewers(tool_input.package, tool_input.dist_git_branch) return JSONToolOutput(result=reviewer_ids) @@ -932,14 +983,14 @@ async def _run( ) -> StringToolOutput: merge_request_url = tool_input.merge_request_url comment = tool_input.comment - try: + with tool_error_context( + "Failed to add comment to merge request", + include_exception_message_for=(ValueError,), + merge_request_url=merge_request_url, + ): mr = await _get_merge_request_from_url(merge_request_url) await asyncio.to_thread(mr._raw_pr.notes.create, {"body": comment}) - return StringToolOutput(result=f"Successfully added comment to merge request {merge_request_url}") - except ToolError: - raise - except Exception as e: - raise ToolError(f"Failed to add comment to merge request: {e}") from e + return StringToolOutput(result=f"Successfully added comment to merge request {merge_request_url}") class AddBlockingMergeRequestCommentToolInput(BaseModel): @@ -973,7 +1024,12 @@ async def _run( ) -> StringToolOutput: merge_request_url = tool_input.merge_request_url comment = tool_input.comment - try: + with tool_error_context( + "Failed to add blocking comment to merge request", + include_exception_message_for=(ValueError,), + merge_request_url=merge_request_url, + blocking_comment=tool_input.comment, + ): mr = await _get_merge_request_from_url(merge_request_url) def check_existing_comment(): @@ -1000,11 +1056,9 @@ def check_existing_comment(): {"body": comment}, ) - return StringToolOutput( - result=f"Successfully added blocking comment to merge request {merge_request_url}" - ) - except Exception as e: - raise ToolError(f"Failed to add blocking comment to merge request: {e}") from e + return StringToolOutput( + result=f"Successfully added blocking comment to merge request {merge_request_url}" + ) class RetryPipelineJobToolInput(BaseModel): @@ -1035,7 +1089,11 @@ async def _run( project_url = tool_input.project_url job_id = tool_input.job_id logger.info(f"Connecting to GitLab API to retry job {job_id} for project: {project_url}") - try: + with tool_error_context( + f"Failed to retry job {job_id}", + project_url=project_url, + job_id=job_id, + ): project = await asyncio.to_thread(get_project, url=project_url, token=os.getenv("GITLAB_TOKEN")) def retry_gitlab_job(): @@ -1045,12 +1103,8 @@ def retry_gitlab_job(): job = await asyncio.to_thread(retry_gitlab_job) - logger.info(f"Successfully retried job {job_id} for project {project_url}") - return StringToolOutput(result=f"Successfully retried job {job_id}. Status: {job.status}") - - except Exception as e: - logger.error(f"Failed to retry job {job_id} for project {project_url}: {e}") - raise ToolError(f"Failed to retry job: {e}") from e + logger.info(f"Successfully retried job {job_id} for project {project_url}") + return StringToolOutput(result=f"Successfully retried job {job_id}. Status: {job.status}") class GetFailedPipelineJobsFromMergeRequestToolInput(BaseModel): @@ -1085,7 +1139,11 @@ async def _run( context: RunContext, ) -> JSONToolOutput[list[FailedPipelineJob]]: merge_request_url = tool_input.merge_request_url - try: + with tool_error_context( + "Failed to get failed jobs from merge request", + include_exception_message_for=(ValueError,), + merge_request_url=merge_request_url, + ): mr = await _get_merge_request_from_url(merge_request_url) def get_latest_pipeline_jobs(): @@ -1118,12 +1176,8 @@ def get_latest_pipeline_jobs(): failed_jobs = await asyncio.to_thread(get_latest_pipeline_jobs) - logger.info(f"Found {len(failed_jobs)} failed jobs in latest pipeline for MR {merge_request_url}") - return JSONToolOutput(result=failed_jobs) - - except Exception as e: - logger.error(f"Failed to get failed jobs from MR {merge_request_url}: {e}") - raise ToolError(f"Failed to get failed jobs from merge request: {e}") from e + logger.info(f"Found {len(failed_jobs)} failed jobs in latest pipeline for MR {merge_request_url}") + return JSONToolOutput(result=failed_jobs) def _get_authorized_member_ids(project: GitlabProject) -> set[int]: @@ -1211,13 +1265,13 @@ async def _run( context: RunContext, ) -> JSONToolOutput[list[MergeRequestComment]]: merge_request_url = tool_input.merge_request_url - try: + with tool_error_context( + "Failed to get authorized comments from merge request", + include_exception_message_for=(ValueError,), + merge_request_url=merge_request_url, + ): comments = await _fetch_authorized_comments_from_merge_request_url(merge_request_url) - return JSONToolOutput(result=comments) - except ToolError: - raise - except Exception as e: - raise ToolError(f"Failed to get authorized comments from merge request: {e}") from e + return JSONToolOutput(result=comments) class GetMergeRequestDetailsToolInput(BaseModel): @@ -1251,26 +1305,26 @@ async def _run( context: RunContext, ) -> JSONToolOutput[MergeRequestDetails]: merge_request_url = tool_input.merge_request_url - try: + with tool_error_context( + "Failed to get merge request details", + include_exception_message_for=(ValueError,), + merge_request_url=merge_request_url, + ): mr = await _get_merge_request_from_url(merge_request_url) comments = await _fetch_authorized_comments_from_merge_request_url(merge_request_url) username = mr.source_project.service.user.get_username() - return JSONToolOutput( - result=MergeRequestDetails( - source_repo=mr.source_project.get_git_urls()["git"], - source_branch=mr.source_branch, - target_repo_name=mr.target_project.gitlab_repo.name, - target_branch=mr.target_branch, - title=mr.title, - description=mr.description, - last_updated_at=mr._raw_pr.updated_at, - comments=[c for c in comments if f"@{username}" in c.message], - ) + return JSONToolOutput( + result=MergeRequestDetails( + source_repo=mr.source_project.get_git_urls()["git"], + source_branch=mr.source_branch, + target_repo_name=mr.target_project.gitlab_repo.name, + target_branch=mr.target_branch, + title=mr.title, + description=mr.description, + last_updated_at=mr._raw_pr.updated_at, + comments=[c for c in comments if f"@{username}" in c.message], ) - except ToolError: - raise - except Exception as e: - raise ToolError(f"Failed to get merge request details: {e}") from e + ) MAX_PATCH_CONTENT_LENGTH = 2000 @@ -1325,7 +1379,11 @@ async def _run( request_url = _get_api_diff_url(patch_url) headers = _get_auth_headers(request_url) - try: + with tool_error_context( + f"Failed to fetch patch from {patch_url}", + patch_url=patch_url, + request_url=request_url, + ): async with ( aiohttp.ClientSession(timeout=AIOHTTP_TIMEOUT) as session, aiohttp_get_with_retries(session, request_url, headers=headers) as response, @@ -1338,8 +1396,6 @@ async def _run( result=f"Error: Failed to fetch patch from {patch_url}: HTTP {response.status}" ) text = await response.text() - except (aiohttp.ClientError, TimeoutError) as e: - raise ToolError(f"Failed to fetch patch from {patch_url}: {e}") from e try: hunks = json.loads(text) except json.decoder.JSONDecodeError: @@ -1431,7 +1487,15 @@ async def _run( except (aiohttp.ClientError, TimeoutError) as e: # Here we handle ClientError as ToolError, because client error # signals networking issues which should be flagged (DNS resolution failure, timeouts etc) - raise ToolError(f"Failed to fetch MR notes for !{input.mr_iid} in {input.project}: {e}") from e + raise ToolErrorWithContext( + f"Failed to fetch MR notes for !{input.mr_iid} in {input.project}", + cause=e, + additional_context=make_additional_context( + project=input.project, + mr_iid=input.mr_iid, + exception=f"{type(e).__name__}: {e}", + ), + ) from e except Exception as e: logger.error("Error fetching GitLab MR notes: %s", e) return StringToolOutput(result=f"Error fetching GitLab MR notes: {e}") @@ -1488,7 +1552,12 @@ async def _run( headers = _get_auth_headers(f"https://gitlab.com/{project}") logger.info("Searching MRs for %s in %s (state=%s)", search, project, state) - try: + with tool_error_context( + f"Failed to search MRs in {project}", + project=project, + search=search, + state=state, + ): async with ( aiohttp.ClientSession(timeout=AIOHTTP_TIMEOUT) as session, aiohttp_get_with_retries(session, url, headers=headers, params=params) as response, @@ -1509,13 +1578,8 @@ async def _run( for mr in data ] - logger.info("Found %d MR(s) for %s in %s", len(results), search, project) - return JSONToolOutput(result=results) - - except ToolError: - raise - except Exception as e: - raise ToolError(f"Failed to search MRs in {project}: {e}") from e + logger.info("Found %d MR(s) for %s in %s", len(results), search, project) + return JSONToolOutput(result=results) class ListProjectMergeRequestsToolInput(BaseModel): @@ -1599,7 +1663,14 @@ async def _run( tool_input.author_username, ) - try: + with tool_error_context( + f"Failed to list MRs for {tool_input.project}", + project=tool_input.project, + state=tool_input.state, + target_branch=tool_input.target_branch, + labels=str(tool_input.labels), + author_username=tool_input.author_username, + ): async with ( aiohttp.ClientSession(timeout=AIOHTTP_TIMEOUT) as session, aiohttp_get_with_retries(session, url, headers=headers, params=params) as response, @@ -1623,8 +1694,5 @@ async def _run( for mr in data ] - logger.info("Found %d MR(s) for project %s", len(results), tool_input.project) - return JSONToolOutput(result=results) - - except Exception as e: - raise ToolError(f"Failed to list MRs for {tool_input.project}: {e}") from e + logger.info("Found %d MR(s) for project %s", len(results), tool_input.project) + return JSONToolOutput(result=results) diff --git a/ymir/tools/privileged/jira.py b/ymir/tools/privileged/jira.py index ebe01a29d..d27d9c7b7 100644 --- a/ymir/tools/privileged/jira.py +++ b/ymir/tools/privileged/jira.py @@ -781,7 +781,10 @@ async def _run( logger.info(f"Connecting to JIRA API to check CVE eligibility: {jira_url}") async with aiohttpClientSession(timeout=AIOHTTP_TIMEOUT) as session: - with tool_error_context(f"Failed to get Jira data for {issue_key}"): + with tool_error_context( + f"Failed to get Jira data for {issue_key}", + issue_key=issue_key, + ): async with aiohttp_get_with_retries( session, jira_url, @@ -1238,7 +1241,11 @@ async def _run( available = ", ".join(t.get("to", {}).get("name", "?") for t in transitions) raise ToolError(f"Status '{status}' is not available for {issue_key}. Available: {available}") - with tool_error_context(f"Failed to change status of {issue_key} to {status}"): + with tool_error_context( + f"Failed to change status of {issue_key} to {status}", + issue_key=issue_key, + status=status, + ): async with session.post( jira_url, json={"transition": {"id": transition["id"]}}, @@ -1352,7 +1359,10 @@ async def _run( logger.info(f"Connecting to JIRA API to verify issue author: {jira_url}") async with aiohttpClientSession(timeout=AIOHTTP_TIMEOUT) as session: - with tool_error_context(f"Failed to get Jira data for {issue_key}"): + with tool_error_context( + f"Failed to get Jira data for {issue_key}", + issue_key=issue_key, + ): async with aiohttp_get_with_retries( session, jira_url, @@ -1375,7 +1385,10 @@ async def _run( elif author_key: params["key"] = author_key - with tool_error_context(f"Failed to get user groups for issue {issue_key}"): + with tool_error_context( + f"Failed to get user groups for issue {issue_key}", + issue_key=issue_key, + ): async with aiohttp_get_with_retries( session, urljoin(os.getenv("JIRA_URL"), "rest/api/3/user"), @@ -1481,7 +1494,10 @@ async def _fetch_dev_status_details( aggregated detail records for every application type found under *summary_category* (e.g. ``"repository"`` or ``"pullrequest"``).""" issue_url = urljoin(jira_base, f"rest/api/3/issue/{issue_key}") - with tool_error_context(f"Failed to resolve issue ID for {issue_key}"): + with tool_error_context( + f"Failed to resolve issue ID for {issue_key}", + issue_key=issue_key, + ): async with aiohttp_get_with_retries( session, issue_url, params={"fields": ""}, headers=headers ) as response: @@ -1758,7 +1774,11 @@ async def _run( logger.info("Updating comment %s on %s", comment_id, issue_key) async with aiohttpClientSession(timeout=AIOHTTP_TIMEOUT) as session: - with tool_error_context(f"Failed to update comment {comment_id} on {issue_key}"): + with tool_error_context( + f"Failed to update comment {comment_id} on {issue_key}", + issue_key=issue_key, + comment_id=comment_id, + ): async with session.put( jira_url, json={"body": comment}, @@ -1920,7 +1940,10 @@ async def _get_user_identifier(session: Any, headers: dict, email: str) -> tuple """ jira_base = os.getenv("JIRA_URL") url = urljoin(jira_base, "rest/api/3/user/search") - with tool_error_context(f"Failed to search for user {email}"): + with tool_error_context( + f"Failed to search for user {email}", + email=email, + ): async with session.get(url, params={"query": email}, headers=headers) as response: response.raise_for_status() users = await response.json() diff --git a/ymir/tools/privileged/lookaside.py b/ymir/tools/privileged/lookaside.py index 78cef0caf..e2668baa3 100644 --- a/ymir/tools/privileged/lookaside.py +++ b/ymir/tools/privileged/lookaside.py @@ -18,6 +18,7 @@ from ymir.common.base_utils import KerberosError, init_kerberos_ticket, is_cs_branch from ymir.common.validators import AbsolutePath from ymir.tools.base import CloneableTool as Tool +from ymir.tools.base import tool_error_context logger = logging.getLogger(__name__) @@ -122,31 +123,32 @@ async def _run( if not sources_path.exists(): return StringToolOutput(result="No sources file found, nothing to download") - try: + with tool_error_context("Failed to parse sources file", sources_path=str(sources_path)): sources = pyrpkg.sources.SourcesFile(str(sources_path), "bsd") - except (pyrpkg.errors.MalformedLineError, ValueError, OSError) as e: - raise ToolError(f"Failed to parse sources file: {e}") from e loop = asyncio.get_running_loop() resolved_dist_git = dist_git_path.resolve() async def download_entry(entry): outfile = (dist_git_path / entry.file).resolve() - try: + with tool_error_context( + f"Invalid source filename in sources file: {entry.file}", + outfile=str(outfile), + ): relative_path = outfile.relative_to(resolved_dist_git) if ".git" in relative_path.parts: raise ValueError("Access to .git directory is forbidden") - except ValueError as e: - raise ToolError(f"Invalid source filename in sources file: {entry.file}") from e - try: + with tool_error_context( + f"Failed to download {entry.file}", + package=tool_input.package, + dist_git_branch=tool_input.dist_git_branch, + ): await loop.run_in_executor( None, partial( cache.download, qualified_name, entry.file, entry.hash, str(outfile), entry.hashtype ), ) - except Exception as e: - raise ToolError(f"Failed to download {entry.file}: {e}") from e unique_entries = [] seen_files: set[str] = set() @@ -192,10 +194,8 @@ async def _run( if os.getenv("DRY_RUN", "False").lower() == "true": return StringToolOutput(result="Dry run, not uploading sources (this is expected, not an error)") - try: + with tool_error_context("Failed to initialize Kerberos ticket"): await init_kerberos_ticket() - except KerberosError as e: - raise ToolError(f"Failed to initialize Kerberos ticket: {e}") from e config = _get_config(tool_input.dist_git_branch) cache = _get_cache(config) @@ -206,10 +206,8 @@ async def _run( if not sources_path.exists(): sources_path.touch() - try: + with tool_error_context("Failed to parse sources file", sources_path=str(sources_path)): sources = pyrpkg.sources.SourcesFile(str(sources_path), "bsd") - except (pyrpkg.errors.MalformedLineError, ValueError, OSError) as e: - raise ToolError(f"Failed to parse sources file: {e}") from e sources.entries.clear() loop = asyncio.get_running_loop() @@ -219,27 +217,34 @@ async def _run( if filename in new_filenames: continue filepath = (dist_git_path / filename).resolve() - try: + with tool_error_context( + f"Invalid source file path: {filename}", + filepath=str(filepath), + ): relative_path = filepath.relative_to(resolved_dist_git) if ".git" in relative_path.parts: raise ValueError("Access to .git directory is forbidden") - except ValueError as e: - raise ToolError(f"Invalid source file path: {filename}") from e if not filepath.is_file(): raise ToolError(f"Source file not found: {filepath}") - try: + with tool_error_context( + f"Failed to hash {filename}", + package=tool_input.package, + filepath=str(filepath), + ): hash_value = await loop.run_in_executor(None, partial(cache.hash_file, str(filepath))) - except Exception as e: - raise ToolError(f"Failed to hash {filename}: {e}") from e - try: - await loop.run_in_executor( - None, partial(cache.upload, qualified_name, str(filepath), hash_value) - ) - except pyrpkg.errors.AlreadyUploadedError: - logger.info("%s is already present in lookaside cache", filename) - except Exception as e: - raise ToolError(f"Failed to upload {filename}: {e}") from e + + with tool_error_context( + f"Failed to upload {filename}", + package=tool_input.package, + filepath=str(filepath), + ): + try: + await loop.run_in_executor( + None, partial(cache.upload, qualified_name, str(filepath), hash_value) + ) + except pyrpkg.errors.AlreadyUploadedError: + logger.info("%s is already present in lookaside cache", filename) sources.add_entry(cache.hashtype, filename, hash_value) new_filenames.add(filename) diff --git a/ymir/tools/privileged/maintainer_rules.py b/ymir/tools/privileged/maintainer_rules.py index 0d434e932..36afa17e6 100644 --- a/ymir/tools/privileged/maintainer_rules.py +++ b/ymir/tools/privileged/maintainer_rules.py @@ -62,7 +62,10 @@ async def _run( headers["PRIVATE-TOKEN"] = token with tool_error_context( - f"Failed to fetch maintainer rules for {tool_input.package}", file_path=tool_input.file_path + f"Failed to fetch maintainer rules for {tool_input.package}", + include_exception_message_for=(ToolError,), + package=tool_input.package, + file_path=tool_input.file_path, ): try: async with ( @@ -81,4 +84,4 @@ async def _run( result=f"Failed to fetch maintainer rules (HTTP {response.status}): {text}" ) except TimeoutError as e: - raise ToolError(f"Timeout while fetching maintainer rules for {tool_input.package}") from e + raise ToolError("Timeout while fetching maintainer rules") from e diff --git a/ymir/tools/privileged/testing_farm.py b/ymir/tools/privileged/testing_farm.py index 6d4ef65a3..330a95331 100644 --- a/ymir/tools/privileged/testing_farm.py +++ b/ymir/tools/privileged/testing_farm.py @@ -23,7 +23,8 @@ TestingFarmRequestResult, ) from ymir.tools.base import CloneableTool as Tool -from ymir.tools.base import tool_error_context +from ymir.tools.base import make_additional_context, tool_error_context +from ymir.tools.errors import ToolErrorWithContext logger = logging.getLogger(__name__) @@ -180,7 +181,10 @@ async def _run( ) -> JSONToolOutput[dict[str, Any]]: logger.info("Getting Testing Farm request %s", tool_input.request_id) - with tool_error_context(f"Failed to get Testing Farm request {tool_input.request_id}"): + with tool_error_context( + f"Failed to get Testing Farm request {tool_input.request_id}", + request_id=tool_input.request_id, + ): response = await asyncio.to_thread(_testing_farm_api_get, f"requests/{tool_input.request_id}") tf_request = _parse_tf_request(response) @@ -257,9 +261,14 @@ def create_new_environment(env: dict) -> dict: new_env["artifacts"] = [{"id": build_nvr, "type": "redhat-brew-build", "order": 40}] return new_env - raise ToolError( + raise ToolErrorWithContext( "Cannot reproduce Testing Farm request: " - "cannot determine how to replace build in environment." + "cannot determine how to replace build in environment.", + additional_context=make_additional_context( + build_nvr=build_nvr, + has_builds_var=builds_var is not None, + artifacts_count=len(artifacts) if artifacts else 0, + ), ) body = { @@ -447,7 +456,11 @@ async def _run( } ) - try: + with tool_error_context( + "Failed to list Testing Farm composes", + ranch=tool_input.ranch, + arch=tool_input.arch, + ): base_url = _testing_farm_url().rsplit("/v0", 1)[0] url = f"{base_url}/v0.2/composes/{tool_input.ranch}" response = await asyncio.to_thread(requests.get, url, headers=_testing_farm_headers(), timeout=30) @@ -472,9 +485,6 @@ async def _run( return JSONToolOutput(result={"composes": composes}) - except Exception as e: - raise ToolError(f"Failed to list Testing Farm composes: {e}") from e - class ReserveTestingFarmMachineToolInput(BaseModel): compose: str = Field(description="Compose to reserve, e.g. RHEL-9.8.0-Nightly") @@ -524,7 +534,11 @@ async def _run( } ) - try: + with tool_error_context( + "Failed to reserve Testing Farm machine", + compose=tool_input.compose, + arch=tool_input.arch, + ): # Always use the gateway's own SSH key so run_remote_command can authenticate ssh_public_key = await asyncio.to_thread(_ensure_gateway_ssh_key) if tool_input.ssh_public_key and tool_input.ssh_public_key != ssh_public_key: @@ -574,8 +588,6 @@ async def _run( } response = await asyncio.to_thread(_testing_farm_api_post, "requests", json=body) - except Exception as e: - raise ToolError(f"Failed to reserve Testing Farm machine: {e}") from e return JSONToolOutput(result={"id": response["id"]}) @@ -621,34 +633,34 @@ async def _run( _TRANSIENT_HTTP_CODES = (502, 503, 504) for attempt in range(1, max_attempts + 1): - try: - response = await asyncio.to_thread(_testing_farm_api_get, f"requests/{tool_input.request_id}") - except requests.RequestException as e: - is_transient = False - if isinstance(e, requests.HTTPError) and e.response is not None: - if e.response.status_code in _TRANSIENT_HTTP_CODES: + with tool_error_context( + f"Failed to get Testing Farm reservation details {tool_input.request_id}", + request_id=tool_input.request_id, + ): + try: + response = await asyncio.to_thread( + _testing_farm_api_get, f"requests/{tool_input.request_id}" + ) + except requests.RequestException as e: + is_transient = False + if isinstance(e, requests.HTTPError) and e.response is not None: + if e.response.status_code in _TRANSIENT_HTTP_CODES: + is_transient = True + elif isinstance(e, (requests.ConnectionError, requests.Timeout)): is_transient = True - elif isinstance(e, (requests.ConnectionError, requests.Timeout)): - is_transient = True - if is_transient: - logger.warning( - "Transient error %s polling TF %s (attempt %d/%d)", - e, - tool_input.request_id, - attempt, - max_attempts, - ) - if attempt < max_attempts: - await asyncio.sleep(poll_interval) - continue - raise ToolError( - f"Failed to get Testing Farm reservation details {tool_input.request_id}: {e}" - ) from e - except Exception as e: - raise ToolError( - f"Failed to get Testing Farm reservation details {tool_input.request_id}: {e}" - ) from e + if is_transient: + logger.warning( + "Transient error %s polling TF %s (attempt %d/%d)", + e, + tool_input.request_id, + attempt, + max_attempts, + ) + if attempt < max_attempts: + await asyncio.sleep(poll_interval) + continue + raise state = response.get("state", "unknown") @@ -761,10 +773,11 @@ async def _run( } ) - try: + with tool_error_context( + f"Failed to cancel Testing Farm request {request_id}", + request_id=request_id, + ): await asyncio.to_thread(_testing_farm_api_delete, f"requests/{request_id}") - except Exception as e: - raise ToolError(f"Failed to cancel Testing Farm request {request_id}: {e}") from e clear_allowed_ssh_hosts() return JSONToolOutput(result={"cancelled": True, "request_id": request_id}) @@ -813,28 +826,33 @@ async def _run( } ) - try: - await asyncio.to_thread(_ensure_gateway_ssh_key) - proc = await asyncio.create_subprocess_exec( - "ssh", - "-i", - str(_SSH_KEY_PATH), - "-o", - "StrictHostKeyChecking=no", - "-o", - "UserKnownHostsFile=/dev/null", - ssh_host, - command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) - except TimeoutError as e: - proc.kill() - await proc.wait() - raise ToolError(f"Command timed out after {timeout}s on {ssh_host}: {command}") from e - except Exception as e: - raise ToolError(f"Failed to run command on {ssh_host}: {e}") from e + with tool_error_context( + f"Failed to run remote command on {ssh_host}", + include_exception_message_for=(ToolError,), + ssh_host=ssh_host, + command=command, + timeout=timeout, + ): + try: + await asyncio.to_thread(_ensure_gateway_ssh_key) + proc = await asyncio.create_subprocess_exec( + "ssh", + "-i", + str(_SSH_KEY_PATH), + "-o", + "StrictHostKeyChecking=no", + "-o", + "UserKnownHostsFile=/dev/null", + ssh_host, + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except TimeoutError as e: + proc.kill() + await proc.wait() + raise ToolError(f"Command timed out after {timeout}s") from e return JSONToolOutput( result={ @@ -923,43 +941,49 @@ async def _run( ] active_proc = None - try: - # Create the remote directory - active_proc = await asyncio.create_subprocess_exec( - "ssh", - *ssh_opts, - ssh_host, - "mkdir", - "-p", - remote_dir, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - _, stderr = await asyncio.wait_for(active_proc.communicate(), timeout=timeout) - if active_proc.returncode != 0: - raise RuntimeError( - f"Failed to create remote directory {remote_dir}: {stderr.decode().strip()}" + with tool_error_context( + f"Failed to copy files to {ssh_host}:{remote_dir}", + include_exception_message_for=(ToolError,), + ssh_host=ssh_host, + remote_dir=remote_dir, + local_paths=str(local_paths), + timeout=timeout, + ): + try: + # Create the remote directory + active_proc = await asyncio.create_subprocess_exec( + "ssh", + *ssh_opts, + ssh_host, + "mkdir", + "-p", + remote_dir, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, ) + _, stderr = await asyncio.wait_for(active_proc.communicate(), timeout=timeout) + if active_proc.returncode != 0: + raise RuntimeError( + f"Failed to create remote directory {remote_dir}: {stderr.decode().strip()}" + ) - # Copy files via scp - active_proc = await asyncio.create_subprocess_exec( - "scp", - *ssh_opts, - "-r", - *local_paths, - f"{ssh_host}:{remote_dir}", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - _, stderr = await asyncio.wait_for(active_proc.communicate(), timeout=timeout) - if active_proc.returncode != 0: - raise RuntimeError(f"SCP failed: {stderr.decode().strip()}") - except TimeoutError as e: - if active_proc: - active_proc.kill() - await active_proc.wait() - raise ToolError(f"Copy operation timed out after {timeout}s to {ssh_host}:{remote_dir}") from e - except Exception as e: - raise ToolError(f"Failed to copy files to {ssh_host}:{remote_dir}: {e}") from e + # Copy files via scp + active_proc = await asyncio.create_subprocess_exec( + "scp", + *ssh_opts, + "-r", + *local_paths, + f"{ssh_host}:{remote_dir}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + _, stderr = await asyncio.wait_for(active_proc.communicate(), timeout=timeout) + if active_proc.returncode != 0: + raise RuntimeError(f"SCP failed: {stderr.decode().strip()}") + except TimeoutError as e: + if active_proc: + active_proc.kill() + await active_proc.wait() + raise ToolError(f"Copy operation timed out after {timeout}s") from e return JSONToolOutput(result={"copied": True, "remote_dir": remote_dir, "files": local_paths}) diff --git a/ymir/tools/privileged/zstream_search.py b/ymir/tools/privileged/zstream_search.py index b9e57c5ef..ae686ee06 100644 --- a/ymir/tools/privileged/zstream_search.py +++ b/ymir/tools/privileged/zstream_search.py @@ -237,6 +237,8 @@ def _not_found(): with tool_error_context( f"Failed to check z-stream status for {tool_input.component}/{tool_input.fix_version}", + component=tool_input.component, + fix_version=tool_input.fix_version, ): older = await is_older_zstream(tool_input.fix_version) @@ -263,6 +265,8 @@ def _not_found(): with tool_error_context( f"Failed to search Jira for related z-stream issues" f" for {tool_input.component}/{tool_input.fix_version}", + component=tool_input.component, + fix_version=tool_input.fix_version, jql=jql, ): search_result = await run_tool(