diff --git a/ariadne/services/hermes_autotriage.py b/ariadne/services/hermes_autotriage.py index 8f7f36b..3726b31 100644 --- a/ariadne/services/hermes_autotriage.py +++ b/ariadne/services/hermes_autotriage.py @@ -8,7 +8,8 @@ import httpx from ..settings import settings from ..utils.logging import get_logger -from . import hermes_agent_client, hermes_autotriage_repair, hermes_code_flow, hermes_infra_signals +from . import hermes_agent_client, hermes_autotriage_repair, hermes_code_flow, hermes_incident_issue +from . import hermes_infra_signals from . import hermes_autotriage_decision as hermes_decision from . import hermes_autotriage_events as hermes_events from . import hermes_autotriage_evidence as hermes_evidence @@ -68,8 +69,9 @@ def run_hermes_autotriage(storage: Any) -> dict[str, Any]: started = time.time() incidents = hermes_events.incident_state(storage) jobs: dict[str, Any] = {} + tick_state: dict[str, Any] = {} for job in settings.hermes_autotriage_job_allowlist: - jobs[job] = _process_job(storage, job, incidents) + jobs[job] = _process_job(storage, job, incidents, tick_state) HERMES_TRIAGE_DURATION_SECONDS.labels(phase="total").set(time.time() - started) logger.info( "hermes autotriage tick finished", @@ -78,7 +80,9 @@ def run_hermes_autotriage(storage: Any) -> dict[str, Any]: return {"status": "ok", "jobs": jobs} -def _process_job(storage: Any, job: str, incidents: dict[str, dict[str, Any]]) -> dict[str, Any]: +def _process_job( + storage: Any, job: str, incidents: dict[str, dict[str, Any]], tick_state: dict[str, Any] +) -> dict[str, Any]: """Inspect one allowlisted job's last build and advance its incidents.""" last_build = _fetch_last_build(job) @@ -89,7 +93,7 @@ def _process_job(storage: Any, job: str, incidents: dict[str, dict[str, Any]]) - if result == "SUCCESS": return _resolve_on_success(storage, job, last_build, incidents) if result == "FAILURE": - return _handle_failure(storage, job, last_build, incidents) + return _handle_failure(storage, job, last_build, incidents, tick_state) return {"status": "ignored", "result": result} @@ -148,8 +152,8 @@ def _resolve_on_success( return {"status": "healthy", "resolved": resolved} -def _handle_failure( - storage: Any, job: str, last_build: dict[str, Any], incidents: dict[str, dict[str, Any]] +def _handle_failure( # noqa: PLR0913 - the tick's issue budget travels with the incident context + storage: Any, job: str, last_build: dict[str, Any], incidents: dict[str, dict], tick_state: dict[str, Any] ) -> dict[str, Any]: """Route a terminal build failure to dedupe, rebuild-failure, or triage.""" @@ -162,7 +166,7 @@ def _handle_failure( if stale is not None: base = {"incident_id": incident_id, "job": job, "build_number": number} return _mark_rebuild_failure(storage, stale, base) - return _run_pipeline(storage, incident_id, job, last_build) + return _run_pipeline(storage, incident_id, job, last_build, tick_state) def _awaiting_rebuild_incident( @@ -200,8 +204,8 @@ def _mark_rebuild_failure( } -def _run_pipeline( - storage: Any, incident_id: str, job: str, last_build: dict[str, Any] +def _run_pipeline( # noqa: PLR0913 - the tick's issue budget travels with the incident context + storage: Any, incident_id: str, job: str, last_build: dict[str, Any], tick_state: dict[str, Any] ) -> dict[str, Any]: """Run detect, evidence, diagnosis, and authorization for a new incident.""" @@ -224,13 +228,14 @@ def _run_pipeline( reason = f"hermes_run_{run.status}" hermes_events.record_diagnosis(storage, base, run, None, hermes_events.Authorization(False, reason)) hermes_events.record_incident(storage, base, "human_required", {"reason": reason}) + _file_incident_issue(storage, base, _diagnosis(bundle, None, reason, run.run_id), tick_state) return {"status": "human_required", "incident_id": incident_id, "reason": reason} outcome = hermes_decision.parse_triage_response(run.output, incident_id) - return _authorize_and_execute(storage, base, run, outcome, bundle) + return _authorize_and_execute(storage, base, run, outcome, bundle, tick_state) -def _authorize_and_execute( - storage: Any, base: dict[str, Any], run: Any, outcome: Any, bundle: dict[str, Any] +def _authorize_and_execute( # noqa: PLR0913 - the tick's issue budget travels with the incident context + storage: Any, base: dict[str, Any], run: Any, outcome: Any, bundle: dict[str, Any], tick_state: dict[str, Any] ) -> dict[str, Any]: """Gate the parsed diagnosis and run its action when every gate passes.""" @@ -252,10 +257,35 @@ def _authorize_and_execute( if not allowed: HERMES_TRIAGE_ACTION_TOTAL.labels(action=_action_label(outcome), result="rejected").inc() hermes_events.record_incident(storage, base, "human_required", {"reason": reason}) + _file_incident_issue(storage, base, _diagnosis(bundle, outcome, reason, run.run_id), tick_state) return {"status": "human_required", "incident_id": incident_id, "reason": reason} return _execute_action(storage, base, outcome, marker) +def _file_incident_issue( + storage: Any, base: dict[str, Any], diagnosis: dict[str, Any], tick_state: dict[str, Any] +) -> None: + """Hand a human_required incident to its service repository as an issue. + + Filing is opt-in per job, additive, and never mutates anything, so any + failure is swallowed here and the tick continues unaffected. + """ + + try: + hermes_incident_issue.maybe_file_issue(storage, settings, base, diagnosis, tick_state) + except Exception as exc: + logger.info( + "hermes autotriage issue filing failed", + extra={"event": "hermes_autotriage", "status": "issue_error", "detail": str(exc)}, + ) + + +def _diagnosis(bundle: dict[str, Any], outcome: Any, reason: str, run_id: str | None) -> dict[str, Any]: + """Pack the diagnosis inputs the issue body is rendered from.""" + + return {"bundle": bundle, "outcome": outcome, "authorize_reason": reason, "run_id": run_id} + + def _evidence_signature(outcome: Any, bundle: dict[str, Any], incident_id: str) -> tuple[bool, str | None]: """Run the signature check that belongs to the requested action id. diff --git a/ariadne/services/hermes_code_candidates.py b/ariadne/services/hermes_code_candidates.py new file mode 100644 index 0000000..b5d04eb --- /dev/null +++ b/ariadne/services/hermes_code_candidates.py @@ -0,0 +1,240 @@ +"""Select which source files a build's failures implicate. + +The code-repair flow used to patch one hardcoded file. Real service +repositories hold many, and the failing console output is the only evidence +that says which of them are in play: pytest tracebacks, Rust diagnostics, and +JS/TS stack frames all name a `path:line`. This module turns those references +into a short, ranked, safety-filtered candidate list. + +`FILE_REFERENCE_PATTERNS` is deliberately reviewable: every entry is a named +regex an operator can read, grep for, and test in isolation. + +Ranking is tiered rather than a single score, because the two signals answer +different questions. The console failure region a path first appears in says +*which failure* implicates it, and the earliest enforced failure is the one +worth repairing; reference count and "is this a test file" only break ties +inside one region. A test file stays a candidate - sometimes the test is the +thing that is wrong - it just loses to a source file that the same failure +implicates equally. +""" + +from __future__ import annotations + +import re +from typing import Any + + +DEFAULT_MAX_CANDIDATES = 3 +DEFAULT_TEST_PATH_MARKERS: tuple[str, ...] = ("tests/", "test_", "_test.") + +# Path characters accepted inside a console file reference. Deliberately +# excludes the backslash, whitespace, quotes, and brackets so a match stops at +# the surrounding console punctuation instead of swallowing it. +_PATH_CHARS = r"[A-Za-z0-9_./+\-]" +_FILE_REF = _PATH_CHARS + r"+\.[A-Za-z0-9]{1,6}" +# Bundler and code-host references carry a query or anchor between the file +# and its line number (`src/panel.tsx?t=1699:10:5`); it is consumed here so +# the reference is still recognised, and never captured as part of the path. +_NOISE = r"(?:[?#][^\s:]*)?" + +FILE_REFERENCE_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( + # python: File "path/to/file.py", line 123 + ("python_traceback", re.compile(r'File "(' + _FILE_REF + r')", line \d+')), + # pytest: FAILED tests/test_x.py::test_y / ERROR tests/test_x.py + ("pytest_summary", re.compile(r"(?:FAILED|ERROR)\s+(" + _FILE_REF + r")(?=::|\s|$)")), + # rust: --> src/lib.rs:20:5 + ("rust_diagnostic", re.compile(r"-->\s+(" + _FILE_REF + r")" + _NOISE + r":\d+:\d+")), + # js/ts: at fn (path/to/file.ts:10:5) / at path/to/file.tsx:10:5 + ( + "js_stack_frame", + re.compile(r"\bat\s+(?:[^\s()]+\s+\()?(" + _FILE_REF + r")" + _NOISE + r":\d+:\d+"), + ), + # generic: path/to/file.[?query|#anchor]:[:] + ("path_with_line", re.compile(r"(" + _FILE_REF + r")" + _NOISE + r":\d+(?::\d+)?")), +) + +# Anything up to and including a `/workspace//` segment is the build +# agent's checkout directory, not part of the repository-relative path. +_WORKSPACE_PREFIX = re.compile(r"^.*?/workspace/[^/]+/") +_REPEATED_SLASHES = re.compile(r"/{2,}") +_NOISE_SEPARATORS = ("?", "#") + + +def extract_candidate_paths(bundle: dict, cfg: dict) -> list[str]: + """Rank the repository files that a build's console failures implicate. + + Inputs: an evidence bundle from `collect_evidence` (the + `jenkins.console_failures` regions in chronological order, then + `jenkins.console_tail`) and a cfg carrying `allowed_suffixes`, + `allowed_path_prefixes`, `max_candidates` (default 3), and optional + `test_path_markers`. + Outputs: at most `max_candidates` repository-relative paths, best first: + earliest failure region wins, then more references, then non-test paths. + Paths outside the allowed prefixes/suffixes and any traversal, absolute, + backslash, or NUL-bearing path are dropped. Never raises; returns [] when + nothing matches. + """ + + try: + references = _references(_search_texts(bundle), cfg) + ranked = sorted(references, key=lambda path: _rank_key(path, references[path], cfg)) + return ranked[: _max_candidates(cfg)] + except Exception: + return [] + + +def is_test_path(path: str, cfg: dict) -> bool: + """Report whether a repository path looks like a test file. + + Inputs: a repository-relative path and a cfg whose optional + `test_path_markers` overrides `DEFAULT_TEST_PATH_MARKERS`. A marker ending + in "/" matches a directory segment; any other marker matches the file name + at a non-alphanumeric boundary, so "test_" hits `test_x.py` but not + `latest_run.py`. + Outputs: True when any marker matches. Never raises. + """ + + try: + segments = str(path).split("/") + name = segments[-1] + directories = segments[:-1] + return any( + _marker_hit(marker, name, directories) for marker in _test_markers(cfg) + ) + except Exception: + return False + + +def _test_markers(cfg: dict) -> tuple[str, ...]: + """Return the configured test-path markers, or the module defaults.""" + + markers = cfg.get("test_path_markers") if isinstance(cfg, dict) else None + if not markers: + return DEFAULT_TEST_PATH_MARKERS + return tuple(str(marker) for marker in markers if str(marker)) + + +def _marker_hit(marker: str, name: str, directories: list[str]) -> bool: + """Match one test marker against a file name or its directories.""" + + if marker.endswith("/"): + return marker.rstrip("/") in directories + if not marker[0].isalnum(): + return marker in name + index = name.find(marker) + while index >= 0: + if index == 0 or not name[index - 1].isalnum(): + return True + index = name.find(marker, index + 1) + return False + + +def _search_texts(bundle: dict) -> list[str]: + """Return the console texts to scan, earliest failure region first.""" + + if not isinstance(bundle, dict): + return [] + jenkins = bundle.get("jenkins") + if not isinstance(jenkins, dict): + return [] + regions = jenkins.get("console_failures") + texts = [ + str(region.get("text") or "") + for region in (regions if isinstance(regions, list) else []) + if isinstance(region, dict) + ] + texts.append(str(jenkins.get("console_tail") or "")) + return texts + + +def _references(texts: list[str], cfg: dict) -> dict[str, dict[str, Any]]: + """Collect every allowed path reference with where it first appeared.""" + + found: dict[str, dict[str, Any]] = {} + for region_index, text in enumerate(texts): + for line_index, line in enumerate(text.splitlines()): + for path, column in _line_paths(line, cfg): + entry = found.setdefault( + path, {"first": (region_index, line_index, column), "lines": set()} + ) + entry["lines"].add((region_index, line_index)) + return found + + +def _line_paths(line: str, cfg: dict) -> list[tuple[str, int]]: + """Return the allowed paths one console line names, leftmost first. + + A single line can match several patterns (a generic `path:line` hit and a + language-specific one), so hits are collapsed per path here; that keeps + the reference count a count of *lines*, not of regexes that fired. + """ + + columns: dict[str, int] = {} + for _name, pattern in FILE_REFERENCE_PATTERNS: + for match in pattern.finditer(line): + path = _normalize(match.group(1)) + if not _is_allowed(path, cfg): + continue + column = match.start(1) + if column < columns.get(path, column + 1): + columns[path] = column + return sorted(columns.items(), key=lambda item: (item[1], item[0])) + + +def _normalize(raw: str) -> str: + """Reduce a raw console file reference to a repository-relative path.""" + + path = raw.strip() + for separator in _NOISE_SEPARATORS: + path = path.split(separator)[0] + path = _REPEATED_SLASHES.sub("/", path) + path = _WORKSPACE_PREFIX.sub("", path) + while path.startswith("./"): + path = path[2:] + return path + + +def _is_allowed(path: str, cfg: dict) -> bool: + """Gate one normalized path on safety and the configured allowlists.""" + + if not _is_safe(path): + return False + suffixes = [str(item) for item in (cfg.get("allowed_suffixes") or [])] + prefixes = [str(item) for item in (cfg.get("allowed_path_prefixes") or [])] + if not any(path.endswith(suffix) for suffix in suffixes): + return False + return any(path.startswith(prefix) for prefix in prefixes) + + +def _is_safe(path: str) -> bool: + """Reject traversal, absolute, backslash, and NUL-bearing paths.""" + + if not path or "\x00" in path or "\\" in path: + return False + if path.startswith("/"): + return False + return ".." not in path + + +def _rank_key(path: str, entry: dict[str, Any], cfg: dict) -> tuple: + """Build the deterministic sort key for one candidate path.""" + + region_index, line_index, column = entry["first"] + return ( + region_index, + -len(entry["lines"]), + is_test_path(path, cfg), + line_index, + column, + path, + ) + + +def _max_candidates(cfg: dict) -> int: + """Return the configured candidate cap, defaulting when it is unusable.""" + + try: + value = int(cfg.get("max_candidates")) + except (TypeError, ValueError): + return DEFAULT_MAX_CANDIDATES + return value if value > 0 else DEFAULT_MAX_CANDIDATES diff --git a/ariadne/services/hermes_code_flow.py b/ariadne/services/hermes_code_flow.py index a31b8dd..87818a5 100644 --- a/ariadne/services/hermes_code_flow.py +++ b/ariadne/services/hermes_code_flow.py @@ -1,3 +1,14 @@ +"""Run the bounded code-repair proposal flow for one failed build. + +The flow is per-job: `hermes_code_repos` maps a Jenkins job to the repository +its failures live in (see `hermes_code_repos` for the settings contract), the +console evidence says which files that build implicates (see +`hermes_code_candidates`), and this module fetches, prompts, validates, and +publishes. The original single-repo demo settings still work unchanged - a +job matching `hermes_code_job` keeps patching its one fixed +`hermes_code_candidate_path` in `hermes_code_owner/hermes_code_repo`. +""" + from __future__ import annotations from dataclasses import dataclass @@ -5,7 +16,13 @@ import json from typing import Any from ..utils.logging import get_logger -from . import hermes_agent_client, hermes_code_patch, hermes_code_repair +from . import ( + hermes_agent_client, + hermes_code_candidates, + hermes_code_patch, + hermes_code_repair, + hermes_code_repos, +) logger = get_logger(__name__) @@ -13,15 +30,18 @@ logger = get_logger(__name__) CODE_PROPOSAL_EVENT_TYPE = "hermes_autotriage_code_proposal" _RUN_COMPLETED = "completed" -_GITEA_TIMEOUT_SECONDS = 15.0 _EXISTING_PROPOSAL_REASON = "existing_proposal_open" +_NO_REPO_MAPPING_REASON = "no_repo_mapping" +_NO_CANDIDATE_FILES_REASON = "no_candidate_files" _PATCH_PROMPT_TEMPLATE = """Use $triage-titan-test-failures. You are proposing a MINIMAL source fix for incident __INCIDENT_ID__. The repository is __OWNER__/__REPO__ branch __BASE_BRANCH__. Return ONLY a single JSON object with exactly these keys and no others: {"incident_id": "", "analysis": "", "patch": {"path": "", "original": "", "replacement": "", "rationale": ""} or null, "human_required": , "reason": ""} -`original` must be an exact substring of the file shown below, appearing exactly once. +`patch.path` MUST be exactly one of these candidate paths, copied character for character: +__CANDIDATE_LIST__ +`original` must be an exact substring of the content shown below for THAT file, appearing exactly once. Change as few lines as possible; do not reformat; do not add dependencies. Ariadne validates and pushes the change — you do not execute anything. Set human_required to true if the fix is not a small localized source change. @@ -29,8 +49,16 @@ Set human_required to true if the fix is not a small localized source change. Failing test evidence bundle: __BUNDLE__ -Current content of the candidate file __PATH__: -__FILE_CONTENTS__""" +__FILE_SECTIONS__""" + +_FILE_SECTION_TEMPLATE = "Current content of the candidate file {path}:\n{contents}" + + +@dataclass(frozen=True) +class _Incident: + incident_id: str + job: str + build_number: int @dataclass(frozen=True) @@ -57,10 +85,13 @@ def propose_code_fix( # noqa: PLR0913 - flow contract carries the full incident path config from `code_config`. Outputs: {"status": "pr_opened", ...} when a branch and pull request were created, otherwise {"status": "human_required", "reason": ...}. Always records one - hermes_autotriage_code_proposal event without patch bodies or tokens. + hermes_autotriage_code_proposal event carrying the resolved repository, + the candidate paths offered, and the chosen path - never patch bodies or + tokens. """ - result, event = _propose(incident_id, build_number, bundle, hermes_cfg, code_cfg) + incident = _Incident(incident_id=incident_id, job=job, build_number=build_number) + result, event = _propose(incident, bundle, hermes_cfg, code_cfg) storage.record_event( CODE_PROPOSAL_EVENT_TYPE, {"incident_id": incident_id, "job": job, "build_number": build_number, **event}, @@ -76,52 +107,137 @@ def code_config(config: Any) -> dict[str, Any]: """Build the code-repair cfg dict from a settings-like object. Inputs: an object exposing the hermes_code_* and hermes_gitea_* settings. - Outputs: the cfg dict consumed by the patch validator and the Gitea - client (validation limits, repository identity, and credentials). + Outputs: the settings-level cfg dict built by `hermes_code_repos`, which + also documents the multi-repository fields and their defaults; + `resolve_repo_config` narrows it to one job before any Gitea call. """ - return { - "candidate_path": config.hermes_code_candidate_path, - "allowed_path_prefixes": list(config.hermes_code_allowed_prefixes), - "allowed_suffixes": list(config.hermes_code_allowed_suffixes), - "max_patch_bytes": config.hermes_code_max_patch_bytes, - "max_changed_lines": config.hermes_code_max_changed_lines, - "gitea_base_url": config.hermes_gitea_base_url, - "gitea_token": config.hermes_gitea_token, - "owner": config.hermes_code_owner, - "repo": config.hermes_code_repo, - "base_branch": config.hermes_code_base_branch, - "timeout_seconds": _GITEA_TIMEOUT_SECONDS, - } + return hermes_code_repos.build_config(config) + + +def resolve_repo_config(job: str, settings_cfg: dict) -> dict[str, Any] | None: + """Narrow the settings-level cfg to the repository one job publishes to. + + Inputs: the Jenkins job name and the cfg from `code_config`. Outputs: the + merged per-job cfg consumed by the patch validator and the Gitea client, + or None when the job maps to no repository - the caller must then make no + HTTP call. See `hermes_code_repos.resolve` for the merge rules. + """ + + return hermes_code_repos.resolve(job, settings_cfg) def _propose( - incident_id: str, build_number: int, bundle: dict, hermes_cfg: dict, code_cfg: dict + incident: _Incident, bundle: dict, hermes_cfg: dict, code_cfg: dict ) -> tuple[dict[str, Any], dict[str, Any]]: - """Run fetch, diagnosis, validation, and publication for one incident.""" + """Run resolution, fetch, diagnosis, validation, and publication. - duplicate = _duplicate_proposal(code_cfg, incident_id) + `context` is filled in as the flow learns the repository, the candidate + paths it offered the model, and the path the model chose, so the recorded + event describes the proposal however it ended. + """ + + context: dict[str, Any] = {"repo": None, "candidates": [], "chosen_path": None} + result, event = _run_proposal(incident, bundle, hermes_cfg, code_cfg, context) + return result, {**event, **context} + + +def _run_proposal( + incident: _Incident, bundle: dict, hermes_cfg: dict, code_cfg: dict, context: dict +) -> tuple[dict[str, Any], dict[str, Any]]: + """Resolve the repository and gather candidate files before diagnosis.""" + + cfg = resolve_repo_config(incident.job, code_cfg) + if cfg is None: + return _human_required(_NO_REPO_MAPPING_REASON, None, validated=False) + context["repo"] = f"{cfg.get('owner') or ''}/{cfg.get('repo') or ''}" + duplicate = _duplicate_proposal(cfg, incident.incident_id) if duplicate is not None: return duplicate - path = str(code_cfg.get("candidate_path") or "") - contents, fetch_error = hermes_code_repair.fetch_file(code_cfg, path) - if contents is None: - return _human_required(f"candidate_fetch_failed: {fetch_error}", None, validated=False) - run = hermes_agent_client.run_triage( - hermes_cfg, _build_patch_prompt(incident_id, bundle, code_cfg, contents) - ) + fetched, failure = _candidate_contents(cfg, bundle) + context["candidates"] = list(fetched) + if failure is not None: + return _human_required(failure, None, validated=False) + return _diagnose(incident, bundle, hermes_cfg, cfg, fetched, context) + + +def _diagnose( # noqa: PLR0913 - diagnosis needs both configs plus the fetched context + incident: _Incident, + bundle: dict, + hermes_cfg: dict, + cfg: dict, + fetched: dict[str, str], + context: dict, +) -> tuple[dict[str, Any], dict[str, Any]]: + """Ask Hermes for a patch and gate it against the file it names.""" + + prompt = _build_patch_prompt(incident.incident_id, bundle, cfg, fetched) + run = hermes_agent_client.run_triage(hermes_cfg, prompt) if run.status != _RUN_COMPLETED or not run.output: return _human_required(f"hermes_run_{run.status}", run.run_id, validated=False) - patch, reject_reason = _validated_patch(run.output, incident_id, path, code_cfg, contents) + patch, reject_reason = _validated_patch(run.output, incident.incident_id, cfg, fetched) if patch is None: return _human_required(reject_reason, run.run_id, validated=False) + context["chosen_path"] = patch.path proposal = _Proposal( patch=patch, analysis=hermes_code_patch.parse_analysis(run.output), - patched_contents=hermes_code_patch.apply_patch(contents, patch), + patched_contents=hermes_code_patch.apply_patch(fetched[patch.path], patch), run_id=run.run_id, ) - return _publish(code_cfg, incident_id, build_number, proposal) + return _publish(cfg, incident, proposal) + + +def _candidate_contents(cfg: dict, bundle: dict) -> tuple[dict[str, str], str | None]: + """Pick the candidate paths for this job and fetch their contents.""" + + candidates = _candidate_paths(cfg, bundle) + if not candidates: + return {}, _NO_CANDIDATE_FILES_REASON + fetched, error = _fetch_within_budget(cfg, candidates) + if not fetched: + return {}, f"candidate_fetch_failed: {error}" + return fetched, None + + +def _candidate_paths(cfg: dict, bundle: dict) -> list[str]: + """Return the ranked candidate paths, honouring the legacy fixed path.""" + + legacy_path = str(cfg.get("candidate_path") or "") + if legacy_path: + return [legacy_path] + return hermes_code_candidates.extract_candidate_paths(bundle, cfg) + + +def _fetch_within_budget(cfg: dict, candidates: list[str]) -> tuple[dict[str, str], str | None]: + """Fetch candidates in rank order while the context budget allows. + + A file that would exceed the remaining budget is skipped whole and never + truncated: a patch anchor has to match the file exactly, so a partial + file can only produce an unapplicable patch. Returns the fetched contents + keyed by path in rank order plus the first fetch error seen. + """ + + budget = hermes_code_repos.positive_int( + cfg.get("max_context_chars"), hermes_code_repos.DEFAULT_MAX_CONTEXT_CHARS + ) + fetched: dict[str, str] = {} + first_error: str | None = None + used = 0 + for path in candidates: + contents, error = hermes_code_repair.fetch_file(cfg, path) + if contents is None: + first_error = first_error or str(error or "unknown error") + continue + if used + len(contents) > budget: + logger.info( + "hermes code candidate skipped for context budget", + extra={"event": "hermes_code_flow", "status": "skipped", "detail": path}, + ) + continue + fetched[path] = contents + used += len(contents) + return fetched, first_error def _duplicate_proposal( @@ -129,7 +245,7 @@ def _duplicate_proposal( ) -> tuple[dict[str, Any], dict[str, Any]] | None: """Suppress this proposal when an earlier repair pull request is still open. - Runs before the candidate fetch and before Hermes, so a duplicate costs no + Runs before candidate selection and before Hermes, so a duplicate costs no model tokens. Returns the human_required result/event pair when an open proposal exists, otherwise None so the flow continues — including when the lookup itself failed, which fails open rather than dropping real work. @@ -166,34 +282,47 @@ def _duplicate_proposal( def _validated_patch( - raw_output: str, incident_id: str, path: str, code_cfg: dict, contents: str + raw_output: str, incident_id: str, code_cfg: dict, fetched: dict[str, str] ) -> tuple[hermes_code_patch.ProposedPatch | None, str]: - """Parse and gate the patch response against the fetched candidate file.""" + """Parse and gate the patch response against the file the model chose.""" outcome = hermes_code_patch.parse_patch_response(raw_output, incident_id) if not outcome.valid or outcome.patch is None: return None, f"patch_invalid: {outcome.reject_reason}" - if outcome.patch.path != path: - return None, f"patch_path_mismatch: got {outcome.patch.path!r} expected {path!r}" - ok, gate = hermes_code_patch.validate_patch(outcome.patch, code_cfg, contents) + path = outcome.patch.path + legacy_path = str(code_cfg.get("candidate_path") or "") + if legacy_path and path != legacy_path: + return None, f"patch_path_mismatch: got {path!r} expected {legacy_path!r}" + if path not in fetched: + return None, f"patch_path_not_offered: got {path!r} offered {sorted(fetched)!r}" + ok, gate = hermes_code_patch.validate_patch(outcome.patch, code_cfg, fetched[path]) if not ok: return None, f"patch_rejected: {gate}" return outcome.patch, "" def _publish( - code_cfg: dict, incident_id: str, build_number: int, proposal: _Proposal + code_cfg: dict, incident: _Incident, proposal: _Proposal ) -> tuple[dict[str, Any], dict[str, Any]]: """Push the repair branch and open the human-review pull request.""" push = hermes_code_repair.push_branch( - code_cfg, incident_id, build_number, proposal.patch, proposal.patched_contents + code_cfg, + incident.incident_id, + incident.build_number, + proposal.patch, + proposal.patched_contents, ) branch = str(push.get("branch") or "") if not push.get("committed"): return _human_required(f"branch_push_failed: {push.get('error')}", proposal.run_id, validated=True) pull = hermes_code_repair.open_pull_request( - code_cfg, incident_id, build_number, branch, proposal.patch, proposal.analysis + code_cfg, + incident.incident_id, + incident.build_number, + branch, + proposal.patch, + proposal.analysis, ) if pull.get("error"): return _human_required( @@ -235,16 +364,23 @@ def _human_required( return result, event -def _build_patch_prompt(incident_id: str, bundle: dict, code_cfg: dict, contents: str) -> str: +def _build_patch_prompt( + incident_id: str, bundle: dict, code_cfg: dict, fetched: dict[str, str] +) -> str: """Render the frozen patch prompt with incident, bundle, and file context.""" compact = json.dumps(bundle, separators=(",", ":"), ensure_ascii=True) + listing = "\n".join(f"- {path}" for path in fetched) + sections = "\n\n".join( + _FILE_SECTION_TEMPLATE.format(path=path, contents=contents) + for path, contents in fetched.items() + ) return ( _PATCH_PROMPT_TEMPLATE.replace("__INCIDENT_ID__", incident_id) .replace("__OWNER__", str(code_cfg.get("owner") or "")) .replace("__REPO__", str(code_cfg.get("repo") or "")) .replace("__BASE_BRANCH__", str(code_cfg.get("base_branch") or "")) - .replace("__PATH__", str(code_cfg.get("candidate_path") or "")) + .replace("__CANDIDATE_LIST__", listing) .replace("__BUNDLE__", compact) - .replace("__FILE_CONTENTS__", contents) + .replace("__FILE_SECTIONS__", sections) ) diff --git a/ariadne/services/hermes_code_repos.py b/ariadne/services/hermes_code_repos.py new file mode 100644 index 0000000..fb5dd65 --- /dev/null +++ b/ariadne/services/hermes_code_repos.py @@ -0,0 +1,166 @@ +"""Map Jenkins jobs onto the repositories their code fixes belong in. + +`hermes_code_repos` is the multi-repository index: `job=owner/repo` pairs, +with sibling settings narrowing the allowed path prefixes, file suffixes, and +base branch per job. The original single-repo demo settings remain the +fallback, so a job matching `hermes_code_job` resolves to +`hermes_code_owner/hermes_code_repo` and keeps its one fixed candidate path. + +Every multi-repository setting is read with `getattr(settings, name, default)` +so this module keeps working against a settings object that has not grown the +field yet; the defaults reproduce the single-repo behaviour exactly. +""" + +from __future__ import annotations + +from typing import Any + + +DEFAULT_MAX_CANDIDATES = 3 +DEFAULT_MAX_CONTEXT_CHARS = 60000 + +_GITEA_TIMEOUT_SECONDS = 15.0 +_PAIR_SEPARATOR = "," +_VALUE_SEPARATOR = "|" + + +def build_config(config: Any) -> dict[str, Any]: + """Build the settings-level code-repair cfg from a settings-like object. + + Inputs: an object exposing the hermes_code_* and hermes_gitea_* settings. + The multi-repository fields (`hermes_code_repos`, `hermes_code_prefixes`, + `hermes_code_suffixes`, `hermes_code_base_branches`, + `hermes_code_max_candidates`, `hermes_code_max_context_chars`) are read + defensively, so a settings object without them yields the original + single-repo behaviour. + Outputs: the cfg dict; `resolve` narrows it to one job before any Gitea + call. + """ + + return { + "candidate_path": config.hermes_code_candidate_path, + "allowed_path_prefixes": list(config.hermes_code_allowed_prefixes), + "allowed_suffixes": list(config.hermes_code_allowed_suffixes), + "max_patch_bytes": config.hermes_code_max_patch_bytes, + "max_changed_lines": config.hermes_code_max_changed_lines, + "gitea_base_url": config.hermes_gitea_base_url, + "gitea_token": config.hermes_gitea_token, + "owner": config.hermes_code_owner, + "repo": config.hermes_code_repo, + "base_branch": config.hermes_code_base_branch, + "timeout_seconds": _GITEA_TIMEOUT_SECONDS, + "legacy_job": str(getattr(config, "hermes_code_job", "") or ""), + "repos": _parse_repo_map(getattr(config, "hermes_code_repos", "")), + "job_prefixes": _parse_list_map(getattr(config, "hermes_code_prefixes", "")), + "job_suffixes": _parse_list_map(getattr(config, "hermes_code_suffixes", "")), + "job_base_branches": dict(_parse_pairs(getattr(config, "hermes_code_base_branches", ""))), + "max_candidates": positive_int( + getattr(config, "hermes_code_max_candidates", DEFAULT_MAX_CANDIDATES), + DEFAULT_MAX_CANDIDATES, + ), + "max_context_chars": positive_int( + getattr(config, "hermes_code_max_context_chars", DEFAULT_MAX_CONTEXT_CHARS), + DEFAULT_MAX_CONTEXT_CHARS, + ), + } + + +def resolve(job: str, settings_cfg: dict) -> dict[str, Any] | None: + """Narrow the settings-level cfg to the repository one job publishes to. + + Inputs: the Jenkins job name and the cfg from `build_config`. The job's + repository comes from `hermes_code_repos`, falling back to the legacy + single-repo owner/repo when the job is `hermes_code_job`. Per-job + prefixes, suffixes, and base branch override their single-repo + counterparts when configured. + Outputs: the merged cfg consumed by the patch validator and the Gitea + client, or None when the job maps to no repository - the caller must then + make no HTTP call. `candidate_path` survives only for the legacy demo + job; every other job selects candidates from the evidence bundle. + """ + + name = str(job or "") + identity = _repo_identity(name, settings_cfg) + if identity is None: + return None + resolved = dict(settings_cfg) + resolved.update(identity) + resolved.update(_job_overrides(name, settings_cfg)) + if name != str(settings_cfg.get("legacy_job") or ""): + resolved["candidate_path"] = "" + return resolved + + +def positive_int(value: Any, default: int) -> int: + """Coerce a setting to a positive int, falling back to the default. + + Inputs: any settings value and the default to use. Outputs: the value as + a positive int, or the default when it is missing, unparseable, or not + positive. Never raises. + """ + + try: + number = int(value) + except (TypeError, ValueError): + return default + return number if number > 0 else default + + +def _repo_identity(job: str, cfg: dict) -> dict[str, str] | None: + """Resolve one job's owner/repo, or None when it maps nowhere.""" + + entry = (cfg.get("repos") or {}).get(job) + if isinstance(entry, dict) and entry.get("owner") and entry.get("repo"): + return {"owner": str(entry["owner"]), "repo": str(entry["repo"])} + if job and job == str(cfg.get("legacy_job") or "") and cfg.get("owner") and cfg.get("repo"): + return {"owner": cfg["owner"], "repo": cfg["repo"]} + return None + + +def _job_overrides(job: str, cfg: dict) -> dict[str, Any]: + """Return only the per-job settings that override the single-repo ones.""" + + overrides: dict[str, Any] = {} + branch = (cfg.get("job_base_branches") or {}).get(job) + if branch: + overrides["base_branch"] = branch + prefixes = (cfg.get("job_prefixes") or {}).get(job) + if prefixes: + overrides["allowed_path_prefixes"] = list(prefixes) + suffixes = (cfg.get("job_suffixes") or {}).get(job) + if suffixes: + overrides["allowed_suffixes"] = list(suffixes) + return overrides + + +def _parse_pairs(raw: Any) -> list[tuple[str, str]]: + """Parse a `key=value,key=value` setting into ordered pairs.""" + + pairs: list[tuple[str, str]] = [] + for item in str(raw or "").split(_PAIR_SEPARATOR): + key, separator, value = item.strip().partition("=") + if separator and key.strip() and value.strip(): + pairs.append((key.strip(), value.strip())) + return pairs + + +def _parse_repo_map(raw: Any) -> dict[str, dict[str, str]]: + """Parse `job=owner/repo` pairs into per-job repository identities.""" + + repos: dict[str, dict[str, str]] = {} + for job, value in _parse_pairs(raw): + owner, separator, repo = value.partition("/") + if separator and owner.strip() and repo.strip(): + repos[job] = {"owner": owner.strip(), "repo": repo.strip()} + return repos + + +def _parse_list_map(raw: Any) -> dict[str, list[str]]: + """Parse `job=a|b` pairs into per-job value lists.""" + + parsed: dict[str, list[str]] = {} + for job, value in _parse_pairs(raw): + values = [part.strip() for part in value.split(_VALUE_SEPARATOR) if part.strip()] + if values: + parsed[job] = values + return parsed diff --git a/ariadne/services/hermes_incident_body.py b/ariadne/services/hermes_incident_body.py new file mode 100644 index 0000000..00baa10 --- /dev/null +++ b/ariadne/services/hermes_incident_body.py @@ -0,0 +1,213 @@ +"""Render the title, body, and dedupe marker of a Hermes triage issue. + +Pure formatting: nothing here talks to Gitea, reads settings, or raises. The +body is what an operator actually reads, so every list is capped and every +statement clipped — a diagnosis must fit in an issue without burying the one +line that says why a person is needed. + +The hidden marker comment is the whole dedupe mechanism. A repeatedly failing +job opens a fresh incident per build; the marker is what lets the next build +recognize the issue already filed for the same job and classification. +""" + +from __future__ import annotations + +import re +from typing import Any + + +DEFAULT_MAX_BODY_CHARS = 8000 +UNDIAGNOSED = "undiagnosed" + +_MAX_TITLE_CHARS = 120 +_MAX_FACTS = 10 +_MAX_FACT_CHARS = 300 +_MAX_INFERENCES = 6 +_ELLIPSIS = "..." +_FACT_KEYS = ("statement", "source", "reference") +_TRUNCATION_NOTICE = ( + "\n\n_Truncated by Ariadne: the diagnosis exceeded the configured issue body limit. " + "The complete evidence bundle is in the Ariadne audit trail._" +) +_AUDIT_NOTE = ( + "Full evidence bundle and audit trail live in Ariadne at `/api/admin/audit/events`, " + "event types `hermes_autotriage_incident` and `hermes_autotriage_diagnosis`." +) +_FOOTER_TEMPLATE = ( + "Filed automatically by Ariadne from a Hermes Agent diagnosis (run `{run_id}`). " + "Hermes has no write access to this repository; no files or infrastructure were changed." +) +_MARKER_PATTERN = re.compile( + r"" +) + + +def issue_marker(job: str, classification: str, incident_id: str) -> str: + """Render the hidden marker comment that identifies a filed issue. + + Inputs: the Jenkins job, the diagnosis classification, and the incident id. + Outputs: a single-line HTML comment safe to embed in an issue body; values + are flattened so a stray newline or comment terminator cannot break the + marker or the markdown around it. + """ + + return ( + f"" + ) + + +def parse_issue_marker(body: Any) -> dict[str, str] | None: + """Extract the hermes-triage marker fields from an issue body. + + Inputs: an issue body, which may be missing or not a string. Outputs: + {"job", "classification", "incident"} for the first marker found, or None + when the body carries no marker. + """ + + if not isinstance(body, str): + return None + match = _MARKER_PATTERN.search(body) + if match is None: + return None + return {name: match.group(name).strip() for name in ("job", "classification", "incident")} + + +def issue_title(context: dict) -> str: + """Render the issue title, bounded so issue lists stay readable. + + Inputs: the issue context (job, build_number, classification). Outputs: + `[hermes] {job} #{build}: {classification}` clipped to 120 characters. + """ + + prefix = f"[hermes] {context.get('job')} #{context.get('build_number')}: " + classification = str(context.get("classification") or UNDIAGNOSED) + room = _MAX_TITLE_CHARS - len(prefix) + if room <= len(_ELLIPSIS): + return f"{prefix}{classification}"[:_MAX_TITLE_CHARS] + return f"{prefix}{_clip(classification, room)}" + + +def issue_body(context: dict, max_chars: int = DEFAULT_MAX_BODY_CHARS) -> str: + """Render the markdown issue body with the hidden marker line last. + + Inputs: the issue context (incident identity, classification, confidence, + first_failed_gate, reason, authorize_reason, facts, inferences, build_url, + run_id) and the whole-body character cap. Outputs: the markdown body, + truncated with an explicit notice when it would exceed the cap, always + ending in the dedupe marker so truncation can never drop it. + """ + + marker = issue_marker( + str(context.get("job") or ""), + str(context.get("classification") or UNDIAGNOSED), + str(context.get("incident_id") or ""), + ) + sections = [ + _summary_line(context), + _human_section(context), + _facts_section(context), + _inferences_section(context), + _links_section(context), + _FOOTER_TEMPLATE.format(run_id=context.get("run_id") or "unknown"), + ] + body = "\n\n".join(section for section in sections if section) + return f"{_bounded_body(body, max_chars - len(marker) - 2)}\n\n{marker}" + + +def fact_fields(fact: Any) -> dict[str, str]: + """Normalize one cited fact from a triage dataclass or a plain dict. + + Inputs: a TriageFact or a {statement, source, reference} mapping. Outputs: + the three fields as strings, empty when absent. + """ + + if isinstance(fact, dict): + return {key: str(fact.get(key) or "") for key in _FACT_KEYS} + return {key: str(getattr(fact, key, "") or "") for key in _FACT_KEYS} + + +def _summary_line(context: dict) -> str: + """Render the one-line summary that opens the issue body.""" + + confidence = context.get("confidence") + return ( + f"Hermes auto-triage classified incident `{context.get('incident_id')}` as " + f"**{context.get('classification')}** (confidence {'n/a' if confidence is None else confidence}); " + f"first failed gate: `{context.get('first_failed_gate') or 'unknown'}`." + ) + + +def _human_section(context: dict) -> str: + """Render the section explaining why the incident needs a person.""" + + lines = ["## Why a human is needed", str(context.get("reason") or "no reason recorded")] + authorize_reason = str(context.get("authorize_reason") or "") + if authorize_reason: + lines.append(f"Ariadne did not authorize automated remediation: `{authorize_reason}`.") + return "\n\n".join(lines) + + +def _facts_section(context: dict) -> str: + """Render the bounded evidence list cited by the diagnosis.""" + + facts = context.get("facts") + if not isinstance(facts, list) or not facts: + return "" + lines = ["## Facts"] + for fact in facts[:_MAX_FACTS]: + fields = fact_fields(fact) + statement = _clip(fields["statement"], _MAX_FACT_CHARS) + lines.append(f"- **{fields['source'] or 'unknown'}** — {statement} (`{fields['reference']}`)") + return "\n".join(lines) + + +def _inferences_section(context: dict) -> str: + """Render the bounded inference list from the diagnosis.""" + + inferences = context.get("inferences") + if not isinstance(inferences, list) or not inferences: + return "" + lines = ["## Inferences"] + lines.extend(f"- {_clip(str(item), _MAX_FACT_CHARS)}" for item in inferences[:_MAX_INFERENCES]) + return "\n".join(lines) + + +def _links_section(context: dict) -> str: + """Render the links pointing back at Jenkins and the Ariadne audit log.""" + + build_url = str(context.get("build_url") or "") + return "\n".join( + [ + "## Links", + f"- Failed build: {build_url}" if build_url else "- Failed build: url unavailable", + f"- {_AUDIT_NOTE}", + ] + ) + + +def _bounded_body(body: str, budget: int) -> str: + """Clip the rendered body to its budget with an explicit notice.""" + + if budget <= 0: + return "" + if len(body) <= budget: + return body + room = budget - len(_TRUNCATION_NOTICE) + return body[:room] + _TRUNCATION_NOTICE if room > 0 else body[:budget] + + +def _clip(value: str, limit: int) -> str: + """Clip one string to a limit, marking that it was shortened.""" + + if len(value) <= limit: + return value + return value[: limit - len(_ELLIPSIS)] + _ELLIPSIS + + +def _marker_value(value: str) -> str: + """Flatten one marker field so it cannot break the comment or the body.""" + + return " ".join(str(value).replace("-->", "").split()) or "unknown" diff --git a/ariadne/services/hermes_incident_issue.py b/ariadne/services/hermes_incident_issue.py new file mode 100644 index 0000000..4fbad81 --- /dev/null +++ b/ariadne/services/hermes_incident_issue.py @@ -0,0 +1,406 @@ +"""File Gitea issues for triage incidents that concluded a human is needed. + +When auto-triage decides a real service failure cannot be remediated under its +own authorization policy, the diagnosis is worth more in the failing service's +own issue tracker than in Ariadne's audit log. This module posts that +diagnosis as an issue in that repository so the operator finds it where they +already work. + +The operation is additive and non-destructive: an issue is opened, nothing is +changed. Hermes never holds write access to these repositories — Ariadne posts +under its own automation token, and the issue body says so. + +Filing is opt-in twice over: a master switch, and a per-job map of the +repositories that may receive issues. An unmapped job never causes a single +HTTP call. +""" + +from __future__ import annotations + +from typing import Any + +import httpx + +from ..utils.logging import get_logger +from . import hermes_incident_body as body + + +logger = get_logger(__name__) + +ISSUE_EVENT_TYPE = "hermes_autotriage_issue" + +HTTP_OK = 200 +HTTP_CREATED = 201 + +DEDUPE_BY_CLASSIFICATION = "classification" +DEDUPE_BY_INCIDENT = "incident" + +_DEFAULT_TIMEOUT_SECONDS = 15.0 +_OPEN_ISSUES_LIMIT = 50 +_DEFAULT_MAX_PER_TICK = 2 +_FILED_KEY = "issues_filed" +_REPO_PAIR_LENGTH = 2 + + +def find_open_incident_issue(cfg: dict, job: str, classification: str, incident_id: str) -> dict[str, Any]: + """Find an already-open Hermes triage issue for this failure. + + Inputs: `cfg` with gitea_base_url, gitea_token, owner, repo, and optional + dedupe_scope and timeout_seconds; the job, classification, and incident id + about to be filed. Outputs: {"found", "issue_number", "url", "error"}, + returning the lowest-numbered match so repeated checks stay stable while + the issue sits open. + + Dedupe scope "classification" (the default) treats any open issue carrying + the marker for the same job and classification as this failure's issue, + which is what stops a job that fails every build — a new incident each + time — from filing a new issue each time. Scope "incident" matches only + the same incident id. + + Fails open by design: every HTTP, parse, or transport failure returns + found=False with `error` set. A false "already filed" would silently + swallow a real escalation and the operator would never learn the service + needs them, while a false "not filed yet" costs one duplicate issue. + Never raises and never logs the token. + """ + + base_url = _base_url(cfg) + if not base_url: + return _no_issue("gitea base url is empty") + try: + with httpx.Client(timeout=_timeout(cfg)) as client: + response = client.get( + f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/issues", + headers=_headers(cfg), + params={"state": "open", "limit": _OPEN_ISSUES_LIMIT}, + ) + except Exception as exc: + return _no_issue(f"open issue lookup failed: {exc}") + if response.status_code != HTTP_OK: + return _no_issue(f"open issue lookup http {response.status_code}") + return _lowest_matching_issue(response, _dedupe_scope(cfg), job, classification, incident_id) + + +def create_incident_issue(cfg: dict, context: dict) -> dict[str, Any]: + """Open the triage issue that hands one incident to a human. + + Inputs: `cfg` as for `find_open_incident_issue` plus optional + max_body_chars; `context` with incident_id, job, build_number, build_url, + classification, confidence, first_failed_gate, reason, facts, inferences, + authorize_reason, and run_id. Outputs: {"issue_number", "url", "error"}. + + Only HTTP 201 counts as success. Never raises, never puts credentials in + the title or body, and never logs the token. + """ + + base_url = _base_url(cfg) + if not base_url: + return {"issue_number": None, "url": None, "error": "gitea base url is empty"} + payload = { + "title": body.issue_title(context), + "body": body.issue_body(context, _max_body_chars(cfg)), + } + try: + with httpx.Client(timeout=_timeout(cfg)) as client: + response = client.post( + f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/issues", + headers=_headers(cfg), + json=payload, + ) + except Exception as exc: + return {"issue_number": None, "url": None, "error": f"issue create failed: {exc}"} + if response.status_code != HTTP_CREATED: + return {"issue_number": None, "url": None, "error": f"issue create http {response.status_code}"} + return _created_issue(response) + + +def maybe_file_issue( + storage: Any, + config: Any, + base: dict[str, Any], + diagnosis: dict[str, Any], + tick_state: dict[str, Any], +) -> dict[str, Any] | None: + """File the incident issue when every opt-in gate allows it. + + Inputs: the incident event storage; a settings-like object exposing the + hermes_issue_* and hermes_gitea_* values; the incident identity fields; + `diagnosis` with the evidence bundle, the parsed outcome (or None when the + run never completed), the authorization reason, and the Hermes run id; and + the tick-scoped state dict holding this tick's filed-issue counter. + Outputs: the recorded event detail, or None when nothing was attempted. + + Files nothing unless the master switch is on, the job is mapped to a + repository, and this tick is still under its issue budget. Records one + bounded hermes_autotriage_issue event whenever a lookup ran, including the + deduped case. Never raises: issue filing sits on top of triage as a + courtesy and must never be able to break a tick. + """ + + try: + return _file_issue(storage, config, base, diagnosis, tick_state) + except Exception as exc: + logger.info( + "hermes incident issue filing failed", + extra={ + "event": "hermes_incident_issue", + "status": "error", + "incident_id": str(base.get("incident_id") or ""), + "detail": str(exc), + }, + ) + return None + + +def issue_config(config: Any) -> dict[str, Any]: + """Build the issue-filing cfg dict from a settings-like object. + + Inputs: an object exposing the hermes_issue_* and hermes_gitea_* settings. + Outputs: the cfg dict consumed by the lookup and create calls, reusing the + Gitea base URL and automation token already configured for code repair. + """ + + return { + "enabled": bool(getattr(config, "hermes_issues_enabled", False)), + "repos": dict(getattr(config, "hermes_issue_repos", None) or {}), + "dedupe_scope": getattr(config, "hermes_issue_dedupe_scope", DEDUPE_BY_CLASSIFICATION), + "max_per_tick": getattr(config, "hermes_issue_max_per_tick", _DEFAULT_MAX_PER_TICK), + "gitea_base_url": getattr(config, "hermes_gitea_base_url", ""), + "gitea_token": getattr(config, "hermes_gitea_token", ""), + "timeout_seconds": _DEFAULT_TIMEOUT_SECONDS, + } + + +def issue_context(base: dict[str, Any], diagnosis: dict[str, Any]) -> dict[str, Any]: + """Flatten the incident, evidence bundle, and outcome into body inputs. + + Inputs: the incident identity fields and the diagnosis dict passed to + `maybe_file_issue`. Outputs: the context dict `create_incident_issue` + renders. An incident with no parsed decision still yields a context, with + classification "undiagnosed" so those failures dedupe together. + """ + + decision = getattr(diagnosis.get("outcome"), "decision", None) + bundle = diagnosis.get("bundle") + jenkins = bundle.get("jenkins") if isinstance(bundle, dict) else None + authorize_reason = str(diagnosis.get("authorize_reason") or "") + return { + "incident_id": str(base.get("incident_id") or ""), + "job": str(base.get("job") or ""), + "build_number": base.get("build_number"), + "build_url": str(jenkins.get("url") or "") if isinstance(jenkins, dict) else "", + "classification": str(getattr(decision, "classification", "") or body.UNDIAGNOSED), + "confidence": getattr(decision, "confidence", None), + "first_failed_gate": str(getattr(decision, "first_failed_gate", "") or ""), + "reason": str(getattr(decision, "reason", "") or authorize_reason), + "facts": [body.fact_fields(fact) for fact in getattr(decision, "facts", None) or []], + "inferences": list(getattr(decision, "inferences", None) or []), + "authorize_reason": authorize_reason, + "run_id": diagnosis.get("run_id"), + } + + +def _file_issue( + storage: Any, + config: Any, + base: dict[str, Any], + diagnosis: dict[str, Any], + tick_state: dict[str, Any], +) -> dict[str, Any] | None: + """Run the opt-in gates, the dedupe lookup, and the create.""" + + cfg = issue_config(config) + job = str(base.get("job") or "") + target = _repo_target(cfg["repos"].get(job)) + if not cfg["enabled"] or target is None or _filed_count(tick_state) >= _max_per_tick(cfg): + return None + repo_cfg = {**cfg, "owner": target[0], "repo": target[1]} + context = issue_context(base, diagnosis) + existing = find_open_incident_issue( + repo_cfg, job, str(context["classification"]), str(context["incident_id"]) + ) + if existing.get("found"): + return _record(storage, context, existing, skipped=True) + tick_state[_FILED_KEY] = _filed_count(tick_state) + 1 + return _record(storage, context, create_incident_issue(repo_cfg, context), skipped=False) + + +def _record( + storage: Any, context: dict[str, Any], result: dict[str, Any], *, skipped: bool +) -> dict[str, Any]: + """Append the bounded issue event and return its detail.""" + + detail = { + "incident_id": context["incident_id"], + "job": context["job"], + "build_number": context["build_number"], + "classification": context["classification"], + "issue_number": result.get("issue_number"), + "url": result.get("url"), + "skipped": skipped, + "error": result.get("error"), + } + storage.record_event(ISSUE_EVENT_TYPE, detail) + logger.info( + "hermes incident issue handled", + extra={ + "event": "hermes_incident_issue", + "status": "skipped" if skipped else "filed", + "incident_id": str(context["incident_id"]), + "detail": str(result.get("error") or ""), + }, + ) + return detail + + +def _lowest_matching_issue( + response: Any, scope: str, job: str, classification: str, incident_id: str +) -> dict[str, Any]: + """Pick the lowest-numbered open issue matching the dedupe scope.""" + + try: + payload = response.json() + except Exception as exc: + return _no_issue(f"open issue parse failed: {exc}") + if not isinstance(payload, list): + return _no_issue("open issue payload is not a list") + matches = [ + issue for issue in payload if _issue_matches(issue, scope, job, classification, incident_id) + ] + if not matches: + return _no_issue(None) + lowest = min(matches, key=lambda issue: int(issue["number"])) + return { + "found": True, + "issue_number": int(lowest["number"]), + "url": str(lowest.get("html_url") or "") or None, + "error": None, + } + + +def _issue_matches(issue: Any, scope: str, job: str, classification: str, incident_id: str) -> bool: + """Report whether one open issue is this incident's already-filed issue.""" + + if not isinstance(issue, dict): + return False + number = issue.get("number") + if isinstance(number, bool) or not isinstance(number, int): + return False + marker = body.parse_issue_marker(issue.get("body")) + if marker is None: + return False + if scope == DEDUPE_BY_INCIDENT: + return marker["incident"] == incident_id + return marker["job"] == job and marker["classification"] == classification + + +def _created_issue(response: Any) -> dict[str, Any]: + """Map a 201 issue payload to the create result shape.""" + + try: + payload = response.json() + except Exception: + payload = {} + if not isinstance(payload, dict): + payload = {} + number = payload.get("number") + url = str(payload.get("html_url") or "") or None + if isinstance(number, bool) or not isinstance(number, int): + return {"issue_number": None, "url": url, "error": "issue create response had no number"} + return {"issue_number": number, "url": url, "error": None} + + +def _repo_target(value: Any) -> tuple[str, str] | None: + """Resolve one issue-repo mapping entry to (owner, repo).""" + + if isinstance(value, dict): + owner, repo = str(value.get("owner") or ""), str(value.get("repo") or "") + elif isinstance(value, (tuple, list)) and len(value) == _REPO_PAIR_LENGTH: + owner, repo = str(value[0]), str(value[1]) + elif isinstance(value, str): + owner, _, repo = value.partition("/") + else: + return None + owner, repo = owner.strip(), repo.strip() + return (owner, repo) if owner and repo else None + + +def _no_issue(error: str | None) -> dict[str, Any]: + """Build the fail-open result meaning "no open issue found".""" + + return {"found": False, "issue_number": None, "url": None, "error": error} + + +def _dedupe_scope(cfg: dict) -> str: + """Return the configured dedupe scope, defaulting to classification.""" + + scope = str(cfg.get("dedupe_scope") or "").strip().lower() + return DEDUPE_BY_INCIDENT if scope == DEDUPE_BY_INCIDENT else DEDUPE_BY_CLASSIFICATION + + +def _filed_count(tick_state: Any) -> int: + """Return how many issues this tick has already filed.""" + + return _int_value(tick_state.get(_FILED_KEY)) if isinstance(tick_state, dict) else 0 + + +def _max_per_tick(cfg: dict) -> int: + """Return the per-tick issue budget, defaulting when unset or invalid.""" + + try: + return int(cfg.get("max_per_tick")) + except (TypeError, ValueError): + return _DEFAULT_MAX_PER_TICK + + +def _max_body_chars(cfg: dict) -> int: + """Return the issue body character cap, defaulting when unset or invalid.""" + + try: + value = int(cfg.get("max_body_chars")) + except (TypeError, ValueError): + return body.DEFAULT_MAX_BODY_CHARS + return value if value > 0 else body.DEFAULT_MAX_BODY_CHARS + + +def _int_value(value: Any) -> int: + """Coerce a value to int, defaulting to zero.""" + + try: + return int(value) + except (TypeError, ValueError): + return 0 + + +def _headers(cfg: dict) -> dict[str, str]: + """Build the Gitea token authorization header.""" + + return {"Authorization": f"token {cfg.get('gitea_token') or ''}"} + + +def _base_url(cfg: dict) -> str: + """Return the configured Gitea base URL without a trailing slash.""" + + return str(cfg.get("gitea_base_url") or "").strip().rstrip("/") + + +def _owner(cfg: dict) -> str: + """Return the configured repository owner.""" + + return str(cfg.get("owner") or "") + + +def _repo(cfg: dict) -> str: + """Return the configured repository name.""" + + return str(cfg.get("repo") or "") + + +def _timeout(cfg: dict) -> float: + """Return the bounded request timeout in seconds.""" + + try: + value = float(cfg.get("timeout_seconds")) + except (TypeError, ValueError): + return _DEFAULT_TIMEOUT_SECONDS + return value if value > 0 else _DEFAULT_TIMEOUT_SECONDS diff --git a/ariadne/settings.py b/ariadne/settings.py index 0280f26..862cf3e 100644 --- a/ariadne/settings.py +++ b/ariadne/settings.py @@ -11,6 +11,7 @@ from .settings_sections import ( _game_stream_config, _hermes_autotriage_config, _hermes_code_config, + _hermes_issue_config, _image_sweeper_config, _jenkins_build_weather_config, _jenkins_workspace_cleanup_config, @@ -200,6 +201,12 @@ class Settings: hermes_code_owner: str hermes_code_repo: str hermes_code_base_branch: str + hermes_code_repos: str + hermes_code_prefixes: str + hermes_code_suffixes: str + hermes_code_base_branches: str + hermes_code_max_candidates: int + hermes_code_max_context_chars: int hermes_code_candidate_path: str hermes_code_allowed_prefixes: list[str] hermes_code_allowed_suffixes: list[str] @@ -207,6 +214,10 @@ class Settings: hermes_code_max_changed_lines: int hermes_gitea_base_url: str hermes_gitea_token: str + hermes_issues_enabled: bool + hermes_issue_repos: dict[str, tuple[str, str]] + hermes_issue_dedupe_scope: str + hermes_issue_max_per_tick: int vaultwarden_namespace: str vaultwarden_pod_label: str @@ -326,6 +337,7 @@ class Settings: testing_triage_cfg = _testing_triage_config() hermes_autotriage_cfg = _hermes_autotriage_config() hermes_code_cfg = _hermes_code_config() + hermes_issue_cfg = _hermes_issue_config() vaultwarden_cfg = _vaultwarden_config() schedule_cfg = _schedule_config() cluster_cfg = _cluster_state_config() @@ -372,6 +384,7 @@ class Settings: **testing_triage_cfg, **hermes_autotriage_cfg, **hermes_code_cfg, + **hermes_issue_cfg, **vaultwarden_cfg, **schedule_cfg, **cluster_cfg, diff --git a/ariadne/settings_sections.py b/ariadne/settings_sections.py index 462ebf0..32827c9 100644 --- a/ariadne/settings_sections.py +++ b/ariadne/settings_sections.py @@ -309,6 +309,12 @@ def _hermes_code_config() -> dict[str, Any]: "hermes_code_owner": _env("ARIADNE_HERMES_CODE_OWNER", "bstein"), "hermes_code_repo": _env("ARIADNE_HERMES_CODE_REPO", "hermes-code-demo"), "hermes_code_base_branch": _env("ARIADNE_HERMES_CODE_BASE_BRANCH", "master"), + "hermes_code_repos": _env("ARIADNE_HERMES_CODE_REPOS", ""), + "hermes_code_prefixes": _env("ARIADNE_HERMES_CODE_PREFIXES", ""), + "hermes_code_suffixes": _env("ARIADNE_HERMES_CODE_SUFFIXES", ""), + "hermes_code_base_branches": _env("ARIADNE_HERMES_CODE_BASE_BRANCHES", ""), + "hermes_code_max_candidates": _env_int("ARIADNE_HERMES_CODE_MAX_CANDIDATES", 3), + "hermes_code_max_context_chars": _env_int("ARIADNE_HERMES_CODE_MAX_CONTEXT_CHARS", 60000), "hermes_code_candidate_path": _env("ARIADNE_HERMES_CODE_CANDIDATE_PATH", "src/discount.py"), "hermes_code_allowed_prefixes": [ item.strip() @@ -327,6 +333,25 @@ def _hermes_code_config() -> dict[str, Any]: } +def _repo_map(raw: str) -> dict[str, tuple[str, str]]: + mapping: dict[str, tuple[str, str]] = {} + for job, value in _pair_map(raw).items(): + owner, _, repo = value.partition("/") + if owner.strip() and repo.strip(): + mapping[job] = (owner.strip(), repo.strip()) + return mapping + + +def _hermes_issue_config() -> dict[str, Any]: + scope = _env("ARIADNE_HERMES_ISSUE_DEDUPE_SCOPE", "classification").strip().lower() + return { + "hermes_issues_enabled": _env_bool("ARIADNE_HERMES_ISSUES_ENABLED", "false"), + "hermes_issue_repos": _repo_map(_env("ARIADNE_HERMES_ISSUE_REPOS", "")), + "hermes_issue_dedupe_scope": scope if scope in {"classification", "incident"} else "classification", + "hermes_issue_max_per_tick": _env_int("ARIADNE_HERMES_ISSUE_MAX_PER_TICK", 2), + } + + def _vaultwarden_config() -> dict[str, Any]: return { "vaultwarden_namespace": _env("VAULTWARDEN_NAMESPACE", "vaultwarden"), diff --git a/tests/hermes_autotriage_harness.py b/tests/hermes_autotriage_harness.py index 4dfb7bd..13c82e4 100644 --- a/tests/hermes_autotriage_harness.py +++ b/tests/hermes_autotriage_harness.py @@ -77,6 +77,10 @@ def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] "hermes_code_max_changed_lines": 20, "hermes_gitea_base_url": "https://scm.example", "hermes_gitea_token": "gitea-token", + "hermes_issues_enabled": False, + "hermes_issue_repos": {}, + "hermes_issue_dedupe_scope": "classification", + "hermes_issue_max_per_tick": 2, "jenkins_base_url": "https://ci.example", "jenkins_api_user": "user", "jenkins_api_token": "token", diff --git a/tests/test_hermes_autotriage.py b/tests/test_hermes_autotriage.py index 7791a62..8281f05 100644 --- a/tests/test_hermes_autotriage.py +++ b/tests/test_hermes_autotriage.py @@ -293,3 +293,137 @@ def test_jenkins_request_uses_basic_auth_and_tree(monkeypatch) -> None: url, params = env.calls["gets"][0] assert url == f"https://ci.example/job/{JOB}/api/json" assert params == {"tree": "lastBuild[number,result,building,timestamp,duration,url]"} + + +def _issue_settings(**overrides): # type: ignore[no-untyped-def] + values = { + "hermes_issues_enabled": True, + "hermes_issue_repos": {JOB: ("bstein", JOB)}, + } + values.update(overrides) + return _settings(**values) + + +def _install_issue_tracker(monkeypatch, calls) -> None: # type: ignore[no-untyped-def] + filed: dict = {} + + def fake_find(cfg, job, classification, incident_id): # type: ignore[no-untyped-def] + calls["lookups"].append((cfg["owner"], cfg["repo"], job, classification, incident_id)) + existing = filed.get((job, classification)) + if existing is None: + return {"found": False, "issue_number": None, "url": None, "error": None} + return {"found": True, "issue_number": existing, "url": f"https://scm/issues/{existing}", "error": None} + + def fake_create(cfg, context): # type: ignore[no-untyped-def] + calls["creates"].append(context) + number = 40 + len(calls["creates"]) + filed[(context["job"], context["classification"])] = number + return {"issue_number": number, "url": f"https://scm/issues/{number}", "error": None} + + monkeypatch.setattr(module.hermes_incident_issue, "find_open_incident_issue", fake_find) + monkeypatch.setattr(module.hermes_incident_issue, "create_incident_issue", fake_create) + + +def _issue_env(monkeypatch, cfg=None, **kwargs): # type: ignore[no-untyped-def] + calls: dict = {"lookups": [], "creates": []} + env = _prepare(monkeypatch, cfg=cfg if cfg is not None else _issue_settings(), **kwargs) + _install_issue_tracker(monkeypatch, calls) + env.calls.update(calls) + return env + + +def _human_required_output(build_number: int = 12) -> str: + return _model_output( + incident_id=f"{JOB}/{build_number}", + classification="unknown_build_failure", + requested_action=None, + human_required=True, + reason="the failure matches no known signature", + ) + + +def _issue_events(storage): # type: ignore[no-untyped-def] + return _events(storage, module.hermes_incident_issue.ISSUE_EVENT_TYPE) + + +def test_human_required_incident_files_one_issue(monkeypatch) -> None: + env = _issue_env(monkeypatch, run=_run(output=_human_required_output())) + + summary = module.run_hermes_autotriage(env.storage) + + assert summary["jobs"][JOB]["status"] == "human_required" + assert env.calls["lookups"] == [("bstein", JOB, JOB, "unknown_build_failure", INCIDENT_ID)] + context = env.calls["creates"][0] + assert context["incident_id"] == INCIDENT_ID + assert context["classification"] == "unknown_build_failure" + assert context["reason"] == "the failure matches no known signature" + assert context["run_id"] == "run-1" + assert _issue_events(env.storage) == [ + { + "incident_id": INCIDENT_ID, + "job": JOB, + "build_number": 12, + "classification": "unknown_build_failure", + "issue_number": 41, + "url": "https://scm/issues/41", + "skipped": False, + "error": None, + } + ] + assert _statuses(env.storage)[-1] == "human_required" + + +def test_second_incident_with_the_same_classification_skips(monkeypatch) -> None: + env = _issue_env(monkeypatch, run=_run(output=_human_required_output())) + module.run_hermes_autotriage(env.storage) + + _prepare( + monkeypatch, + cfg=_issue_settings(), + last_build=_build(13, "FAILURE"), + run=_run(output=_human_required_output(13)), + storage=env.storage, + ) + module.run_hermes_autotriage(env.storage) + + assert len(env.calls["creates"]) == 1 + events = _issue_events(env.storage) + assert [event["skipped"] for event in events] == [False, True] + assert events[1]["incident_id"] == f"{JOB}/13" + assert events[1]["issue_number"] == 41 + + +def test_unmapped_job_and_disabled_switch_file_nothing(monkeypatch) -> None: + for cfg in (_issue_settings(hermes_issue_repos={}), _issue_settings(hermes_issues_enabled=False)): + env = _issue_env(monkeypatch, cfg=cfg, run=_run(output=_human_required_output())) + + module.run_hermes_autotriage(env.storage) + + assert env.calls["lookups"] == [] + assert env.calls["creates"] == [] + assert _issue_events(env.storage) == [] + + +def test_remediated_incident_files_no_issue(monkeypatch) -> None: + env = _issue_env(monkeypatch) + + summary = module.run_hermes_autotriage(env.storage) + + assert summary["jobs"][JOB]["status"] == "awaiting_rebuild" + assert env.calls["creates"] == [] + assert _issue_events(env.storage) == [] + + +def test_issue_filing_failure_never_breaks_the_tick(monkeypatch) -> None: + env = _issue_env(monkeypatch, run=_run(output=_human_required_output())) + + def explode(*args, **kwargs): # type: ignore[no-untyped-def] + raise RuntimeError("gitea is unreachable") + + monkeypatch.setattr(module.hermes_incident_issue, "maybe_file_issue", explode) + + summary = module.run_hermes_autotriage(env.storage) + + assert summary["status"] == "ok" + assert summary["jobs"][JOB]["status"] == "human_required" + assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"] diff --git a/tests/test_hermes_code_candidates.py b/tests/test_hermes_code_candidates.py new file mode 100644 index 0000000..cb5c5b2 --- /dev/null +++ b/tests/test_hermes_code_candidates.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import pytest + +from ariadne.services import hermes_code_candidates as module + + +def _cfg(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "allowed_path_prefixes": ["src/", "tests/"], + "allowed_suffixes": [".py", ".rs", ".ts", ".tsx"], + "max_candidates": 3, + } + base.update(overrides) + return base + + +def _bundle(*regions: str, tail: str = "") -> dict: + return { + "jenkins": { + "console_failures": [ + {"marker": "FAILED ", "line_number": index + 1, "text": text} + for index, text in enumerate(regions) + ], + "console_tail": tail, + } + } + + +def _paths(text: str, **cfg_overrides) -> list[str]: # type: ignore[no-untyped-def] + return module.extract_candidate_paths(_bundle(text), _cfg(**cfg_overrides)) + + +@pytest.mark.parametrize( + ("line", "expected"), + [ + (' File "src/ledger.py", line 12, in balance', "src/ledger.py"), + ("src/ledger.py:14: AssertionError", "src/ledger.py"), + ("FAILED tests/test_ledger.py::test_balance - assert 2 == 3", "tests/test_ledger.py"), + ("ERROR tests/test_ledger.py", "tests/test_ledger.py"), + ("thread 'main' panicked at src/lib.rs:12:34:", "src/lib.rs"), + (" --> src/lib.rs:20:5", "src/lib.rs"), + (" at src/client.ts:10:5", "src/client.ts"), + (" at Object. (src/panel.tsx:10:5)", "src/panel.tsx"), + ("compiling src/build.rs:1:1 failed", "src/build.rs"), + ], +) +def test_each_language_reference_pattern_is_matched(line: str, expected: str) -> None: + assert _paths(line) == [expected] + + +def test_named_patterns_are_reviewable() -> None: + names = [name for name, _pattern in module.FILE_REFERENCE_PATTERNS] + assert names == [ + "python_traceback", + "pytest_summary", + "rust_diagnostic", + "js_stack_frame", + "path_with_line", + ] + + +def test_workspace_prefix_is_stripped() -> None: + line = "/home/jenkins/agent/workspace/titan-api/src/ledger.py:14: AssertionError" + assert _paths(line) == ["src/ledger.py"] + + +def test_leading_dot_slash_and_duplicate_slashes_are_normalized() -> None: + assert _paths(".//src//ledger.py:14: boom") == ["src/ledger.py"] + + +def test_query_and_anchor_noise_is_dropped() -> None: + assert _paths("src/ledger.py?raw=1:14: boom") == ["src/ledger.py"] + assert _paths("src/ledger.py#L14:14: boom") == ["src/ledger.py"] + + +def test_suffix_outside_the_allowlist_is_rejected() -> None: + assert _paths("src/ledger.rb:14: boom") == [] + assert _paths("src/ledger.rs:14: boom", allowed_suffixes=[".py"]) == [] + + +def test_prefix_outside_the_allowlist_is_rejected() -> None: + assert _paths("vendor/ledger.py:14: boom") == [] + assert _paths("tests/test_ledger.py:2: boom", allowed_path_prefixes=["src/"]) == [] + + +def test_empty_allowlists_accept_nothing() -> None: + assert _paths("src/ledger.py:14: boom", allowed_suffixes=[]) == [] + assert _paths("src/ledger.py:14: boom", allowed_path_prefixes=[]) == [] + + +def test_traversal_paths_are_rejected() -> None: + assert _paths("src/../../etc/shadow.py:1: boom") == [] + + +def test_absolute_paths_are_rejected() -> None: + assert _paths("/etc/hosts.py:1: boom", allowed_path_prefixes=["/etc/", "src/"]) == [] + + +def test_backslash_paths_never_become_candidates() -> None: + found = _paths("src\\ledger.py:12: boom", allowed_path_prefixes=["src/", "ledger.py"]) + assert "src\\ledger.py" not in found + + +def test_nul_bearing_paths_are_rejected() -> None: + assert _paths("src/led\x00ger.py:1: boom") == [] + + +def test_earliest_region_outranks_a_busier_later_region() -> None: + bundle = _bundle( + "src/first.py:3: AssertionError", + "src/second.py:4: boom\nsrc/second.py:5: boom\nsrc/second.py:6: boom", + ) + assert module.extract_candidate_paths(bundle, _cfg()) == ["src/first.py", "src/second.py"] + + +def test_reference_count_breaks_ties_inside_one_region() -> None: + text = "src/one.py:1: boom\nsrc/many.py:2: boom\nsrc/many.py:3: boom" + assert _paths(text) == ["src/many.py", "src/one.py"] + + +def test_non_test_paths_rank_before_test_paths_at_equal_score() -> None: + text = "FAILED tests/test_ledger.py::test_balance\nsrc/ledger.py:2: AssertionError" + assert _paths(text) == ["src/ledger.py", "tests/test_ledger.py"] + + +def test_console_tail_is_scanned_after_the_regions() -> None: + bundle = _bundle("src/region.py:1: boom", tail="src/tail.py:9: boom") + assert module.extract_candidate_paths(bundle, _cfg()) == ["src/region.py", "src/tail.py"] + + +def test_tail_only_bundle_still_yields_candidates() -> None: + bundle = {"jenkins": {"console_failures": [], "console_tail": "src/tail.py:9: boom"}} + assert module.extract_candidate_paths(bundle, _cfg()) == ["src/tail.py"] + + +def test_repeated_references_are_deduped() -> None: + text = 'src/ledger.py:1: boom\nFile "src/ledger.py", line 1, in balance\nsrc/ledger.py:1: boom' + assert _paths(text) == ["src/ledger.py"] + + +def test_multiple_patterns_on_one_line_count_once() -> None: + text = " --> src/lib.rs:20:5\nsrc/other.rs:1: boom\nsrc/other.rs:2: boom" + assert _paths(text) == ["src/other.rs", "src/lib.rs"] + + +def test_max_candidates_caps_the_result() -> None: + text = "\n".join(f"src/file{index}.py:{index}: boom" for index in range(6)) + assert len(_paths(text)) == 3 + assert len(_paths(text, max_candidates=2)) == 2 + + +def test_max_candidates_defaults_when_missing_or_unusable() -> None: + text = "\n".join(f"src/file{index}.py:{index}: boom" for index in range(6)) + cfg = {"allowed_path_prefixes": ["src/"], "allowed_suffixes": [".py"]} + assert len(module.extract_candidate_paths(_bundle(text), cfg)) == module.DEFAULT_MAX_CANDIDATES + assert len(_paths(text, max_candidates=0)) == module.DEFAULT_MAX_CANDIDATES + assert len(_paths(text, max_candidates="nope")) == module.DEFAULT_MAX_CANDIDATES + + +@pytest.mark.parametrize( + "bundle", + [ + {}, + {"jenkins": None}, + {"jenkins": {}}, + {"jenkins": {"console_failures": None, "console_tail": None}}, + {"jenkins": {"console_failures": "not-a-list", "console_tail": 7}}, + {"jenkins": {"console_failures": ["not-a-dict", {"text": None}, {}]}}, + {"jenkins": {"console_failures": [{"text": "nothing to see"}]}}, + ], +) +def test_empty_or_malformed_bundles_yield_no_candidates(bundle) -> None: # type: ignore[no-untyped-def] + assert module.extract_candidate_paths(bundle, _cfg()) == [] + + +@pytest.mark.parametrize("bundle", [None, "text", 7, [], {"jenkins": 5}]) +def test_never_raises_on_hostile_input(bundle) -> None: # type: ignore[no-untyped-def] + assert module.extract_candidate_paths(bundle, _cfg()) == [] + + +def test_never_raises_on_hostile_config() -> None: + bundle = _bundle("src/ledger.py:1: boom") + assert module.extract_candidate_paths(bundle, None) == [] + assert module.extract_candidate_paths(bundle, {"allowed_suffixes": 5}) == [] + + +@pytest.mark.parametrize( + ("path", "expected"), + [ + ("tests/test_ledger.py", True), + ("src/test_helpers.py", True), + ("src/ledger_test.py", True), + ("tests/conftest.py", True), + ("src/ledger.py", False), + ("src/latest_run.py", False), + ("src/contest.py", False), + ], +) +def test_is_test_path_uses_default_markers(path: str, expected: bool) -> None: + assert module.is_test_path(path, {}) is expected + + +def test_is_test_path_honours_configured_markers() -> None: + cfg = {"test_path_markers": ("spec/",)} + assert module.is_test_path("spec/ledger_spec.py", cfg) is True + assert module.is_test_path("tests/test_ledger.py", cfg) is False + + +def test_is_test_path_never_raises() -> None: + assert module.is_test_path(None, {}) is False + assert module.is_test_path("src/a.py", {"test_path_markers": [None]}) is False diff --git a/tests/test_hermes_code_flow.py b/tests/test_hermes_code_flow.py index 750c5fb..5d020fe 100644 --- a/tests/test_hermes_code_flow.py +++ b/tests/test_hermes_code_flow.py @@ -47,6 +47,13 @@ def _code_cfg(**overrides) -> dict: # type: ignore[no-untyped-def] "repo": "hermes-code-demo", "base_branch": "master", "timeout_seconds": 15.0, + "legacy_job": JOB, + "repos": {}, + "job_prefixes": {}, + "job_suffixes": {}, + "job_base_branches": {}, + "max_candidates": 3, + "max_context_chars": 60000, } base.update(overrides) return base @@ -98,6 +105,8 @@ def _install(monkeypatch, *, fetch=None, run=None, push=None, pull=None, existin def fake_fetch(cfg, path): # type: ignore[no-untyped-def] calls["fetches"].append((cfg, path)) + if isinstance(fetch, dict): + return fetch.get(path, (None, "file fetch http 404")) return fetch if fetch is not None else (FILE_CONTENTS, None) def fake_run(cfg, prompt): # type: ignore[no-untyped-def] @@ -161,6 +170,9 @@ def test_happy_path_opens_pull_request(monkeypatch) -> None: "branch": "hermes-repair/7", "pr_number": 5, "url": "https://scm.example/pulls/5", + "repo": "bstein/hermes-code-demo", + "candidates": ["src/discount.py"], + "chosen_path": "src/discount.py", } serialized = json.dumps(detail) assert "return price" not in serialized @@ -198,6 +210,9 @@ def test_existing_open_proposal_suppresses_duplicate(monkeypatch) -> None: "branch": "hermes-repair/4", "pr_number": 1, "url": "https://scm.example/pulls/1", + "repo": "bstein/hermes-code-demo", + "candidates": [], + "chosen_path": None, } @@ -311,6 +326,7 @@ def test_pull_request_failure_requires_human(monkeypatch) -> None: def test_code_config_maps_settings() -> None: config = SimpleNamespace( + hermes_code_job=JOB, hermes_code_candidate_path="src/discount.py", hermes_code_allowed_prefixes=["src/"], hermes_code_allowed_suffixes=[".py"], @@ -324,7 +340,6 @@ def test_code_config_maps_settings() -> None: ) assert module.code_config(config) == _code_cfg() - def _orchestrator_settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] values = { "hermes_autotriage_enabled": True, diff --git a/tests/test_hermes_code_flow_repos.py b/tests/test_hermes_code_flow_repos.py new file mode 100644 index 0000000..2dc91a5 --- /dev/null +++ b/tests/test_hermes_code_flow_repos.py @@ -0,0 +1,309 @@ +"""Multi-repository code-proposal tests for the Hermes code flow. + +The legacy single-repo demo path lives in test_hermes_code_flow.py; this file +covers per-job repository resolution and evidence-driven candidate selection. +Shared fakes are imported from that module so both drive the same storage, +Gitea, and Hermes stand-ins. +""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +from ariadne.services import hermes_code_flow as module +from tests.test_hermes_code_flow import ( + JOB, + FakeStorage, + _code_cfg, + _event, + _hermes_cfg, + _install, + _run, +) + + +SERVICE_JOB = "titan-api" +SERVICE_INCIDENT_ID = f"{SERVICE_JOB}/9" +SERVICE_SOURCE = "def balance(rows):\n return sum(rows) - 1\n" +SERVICE_TEST = "def test_balance():\n assert balance([1, 2]) == 3\n" +SERVICE_BUNDLE = { + "incident_id": SERVICE_INCIDENT_ID, + "jenkins": { + "job": SERVICE_JOB, + "console_failures": [ + { + "marker": "FAILED ", + "line_number": 40, + "text": ( + "FAILED tests/test_ledger.py::test_balance\n" + "src/ledger.py:2: AssertionError\n" + "src/ledger.py:2: in balance" + ), + } + ], + "console_tail": "ERROR: script returned exit code 1", + }, + "log_evidence": {"records": []}, +} + + +def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] + values = { + "hermes_code_job": JOB, + "hermes_code_candidate_path": "src/discount.py", + "hermes_code_allowed_prefixes": ["src/"], + "hermes_code_allowed_suffixes": [".py"], + "hermes_code_max_patch_bytes": 4000, + "hermes_code_max_changed_lines": 20, + "hermes_gitea_base_url": "https://scm.example", + "hermes_gitea_token": "secret-token", + "hermes_code_owner": "bstein", + "hermes_code_repo": "hermes-code-demo", + "hermes_code_base_branch": "master", + } + values.update(overrides) + return SimpleNamespace(**values) + + +def _service_cfg(**overrides) -> dict: # type: ignore[no-untyped-def] + settings = _settings( + hermes_code_repos=f"{SERVICE_JOB}=titan/api, other-job = titan/other", + hermes_code_prefixes=f"{SERVICE_JOB}=src/|tests/", + hermes_code_suffixes=f"{SERVICE_JOB}=.py", + hermes_code_base_branches=f"{SERVICE_JOB}=main", + **overrides, + ) + return module.code_config(settings) + + +def _propose_service(monkeypatch, cfg=None, job=SERVICE_JOB, **kwargs): # type: ignore[no-untyped-def] + storage = FakeStorage() + calls = _install(monkeypatch, **kwargs) + result = module.propose_code_fix( + storage, + SERVICE_INCIDENT_ID, + job, + 9, + SERVICE_BUNDLE, + _hermes_cfg(), + _service_cfg() if cfg is None else cfg, + ) + return storage, calls, result + + +def _service_fetch(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "src/ledger.py": (SERVICE_SOURCE, None), + "tests/test_ledger.py": (SERVICE_TEST, None), + } + base.update(overrides) + return base + + +def _service_output(path: str, original: str, replacement: str) -> str: + return json.dumps( + { + "incident_id": SERVICE_INCIDENT_ID, + "analysis": "The balance helper is off by one.", + "patch": { + "path": path, + "original": original, + "replacement": replacement, + "rationale": "restore the intended balance", + }, + "human_required": False, + "reason": "small localized fix", + } + ) + + +def test_code_config_defaults_when_multi_repo_settings_absent() -> None: + cfg = module.code_config(_settings()) + assert cfg["repos"] == {} + assert cfg["job_prefixes"] == {} + assert cfg["job_suffixes"] == {} + assert cfg["job_base_branches"] == {} + assert cfg["max_candidates"] == 3 + assert cfg["max_context_chars"] == 60000 + + +def test_code_config_parses_per_job_maps() -> None: + cfg = _service_cfg(hermes_code_max_candidates=5, hermes_code_max_context_chars=1234) + assert cfg["repos"] == { + SERVICE_JOB: {"owner": "titan", "repo": "api"}, + "other-job": {"owner": "titan", "repo": "other"}, + } + assert cfg["job_prefixes"] == {SERVICE_JOB: ["src/", "tests/"]} + assert cfg["job_suffixes"] == {SERVICE_JOB: [".py"]} + assert cfg["job_base_branches"] == {SERVICE_JOB: "main"} + assert cfg["max_candidates"] == 5 + assert cfg["max_context_chars"] == 1234 + + +def test_code_config_ignores_malformed_pairs() -> None: + cfg = module.code_config( + _settings( + hermes_code_repos="broken, =owner/repo, job=, job2=owneronly, job3=titan/api", + hermes_code_prefixes="job3=,job4=src/", + hermes_code_max_candidates="not-a-number", + ) + ) + assert cfg["repos"] == {"job3": {"owner": "titan", "repo": "api"}} + assert cfg["job_prefixes"] == {"job4": ["src/"]} + assert cfg["max_candidates"] == 3 + + +def test_resolve_repo_config_uses_per_job_mapping() -> None: + resolved = module.resolve_repo_config(SERVICE_JOB, _service_cfg()) + assert resolved is not None + assert resolved["owner"] == "titan" + assert resolved["repo"] == "api" + assert resolved["base_branch"] == "main" + assert resolved["allowed_path_prefixes"] == ["src/", "tests/"] + assert resolved["allowed_suffixes"] == [".py"] + assert resolved["candidate_path"] == "" + assert resolved["gitea_token"] == "secret-token" + assert resolved["max_patch_bytes"] == 4000 + + +def test_resolve_repo_config_falls_back_to_single_repo_settings() -> None: + resolved = module.resolve_repo_config(JOB, _service_cfg()) + assert resolved is not None + assert (resolved["owner"], resolved["repo"]) == ("bstein", "hermes-code-demo") + assert resolved["base_branch"] == "master" + assert resolved["allowed_path_prefixes"] == ["src/"] + assert resolved["candidate_path"] == "src/discount.py" + + +def test_resolve_repo_config_is_unchanged_for_legacy_only_settings() -> None: + assert module.resolve_repo_config(JOB, _code_cfg()) == _code_cfg() + + +def test_resolve_repo_config_returns_none_for_unmapped_job() -> None: + assert module.resolve_repo_config("unmapped-job", _service_cfg()) is None + assert module.resolve_repo_config("", _service_cfg()) is None + + +def test_unmapped_job_requires_human_without_http(monkeypatch) -> None: + storage, calls, result = _propose_service(monkeypatch, job="unmapped-job") + assert result == {"status": "human_required", "reason": "no_repo_mapping"} + assert calls["lookups"] == [] + assert calls["fetches"] == [] + assert calls["runs"] == [] + detail = _event(storage) + assert detail["reject_reason"] == "no_repo_mapping" + assert detail["repo"] is None + assert detail["candidates"] == [] + assert detail["chosen_path"] is None + + +def test_no_candidate_files_makes_no_model_call(monkeypatch) -> None: + bundle = {"incident_id": SERVICE_INCIDENT_ID, "jenkins": {"console_tail": "no file here"}} + storage = FakeStorage() + calls = _install(monkeypatch) + result = module.propose_code_fix( + storage, SERVICE_INCIDENT_ID, SERVICE_JOB, 9, bundle, _hermes_cfg(), _service_cfg() + ) + assert result == {"status": "human_required", "reason": "no_candidate_files"} + assert len(calls["lookups"]) == 1 + assert calls["fetches"] == [] + assert calls["runs"] == [] + assert _event(storage)["reject_reason"] == "no_candidate_files" + + +def test_service_repo_offers_ranked_candidates(monkeypatch) -> None: + storage, calls, result = _propose_service( + monkeypatch, + fetch=_service_fetch(), + run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), + ) + assert result["status"] == "pr_opened" + assert [path for _cfg, path in calls["fetches"]] == ["src/ledger.py", "tests/test_ledger.py"] + push_cfg, _incident, build_number, patch, contents = calls["pushes"][0] + assert (push_cfg["owner"], push_cfg["repo"], push_cfg["base_branch"]) == ("titan", "api", "main") + assert (build_number, patch.path) == (9, "src/ledger.py") + assert contents == "def balance(rows):\n return sum(rows)\n" + detail = _event(storage) + assert detail["repo"] == "titan/api" + assert detail["candidates"] == ["src/ledger.py", "tests/test_ledger.py"] + assert detail["chosen_path"] == "src/ledger.py" + + +def test_prompt_lists_every_offered_candidate(monkeypatch) -> None: + _, calls, _ = _propose_service( + monkeypatch, + fetch=_service_fetch(), + run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), + ) + prompt = calls["runs"][0][1] + assert "The repository is titan/api branch main." in prompt + assert "- src/ledger.py\n- tests/test_ledger.py" in prompt + assert "`patch.path` MUST be exactly one of these candidate paths" in prompt + assert "Current content of the candidate file src/ledger.py:" in prompt + assert "Current content of the candidate file tests/test_ledger.py:" in prompt + assert SERVICE_SOURCE in prompt + assert prompt.rstrip().endswith(SERVICE_TEST.rstrip()) + + +def test_failed_candidate_fetch_is_skipped(monkeypatch) -> None: + storage, calls, result = _propose_service( + monkeypatch, + fetch=_service_fetch(**{"tests/test_ledger.py": (None, "file fetch http 404")}), + run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), + ) + assert result["status"] == "pr_opened" + assert len(calls["fetches"]) == 2 + prompt = calls["runs"][0][1] + assert "tests/test_ledger.py" not in prompt.split("Failing test evidence bundle:")[0] + assert _event(storage)["candidates"] == ["src/ledger.py"] + + +def test_all_candidate_fetches_failing_requires_human(monkeypatch) -> None: + storage, calls, result = _propose_service( + monkeypatch, fetch={"src/ledger.py": (None, "file fetch http 500")} + ) + assert result == {"status": "human_required", "reason": "candidate_fetch_failed: file fetch http 500"} + assert calls["runs"] == [] + assert _event(storage)["candidates"] == [] + + +def test_context_budget_skips_oversized_candidate(monkeypatch) -> None: + cfg = _service_cfg(hermes_code_max_context_chars=len(SERVICE_SOURCE) + 1) + storage, calls, result = _propose_service( + monkeypatch, + cfg=cfg, + fetch=_service_fetch(), + run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), + ) + assert result["status"] == "pr_opened" + assert len(calls["fetches"]) == 2 + assert SERVICE_TEST not in calls["runs"][0][1] + assert _event(storage)["candidates"] == ["src/ledger.py"] + + +def test_patch_path_not_offered_requires_human(monkeypatch) -> None: + storage, calls, result = _propose_service( + monkeypatch, + fetch=_service_fetch(), + run=_run(output=_service_output("src/other.py", "sum(rows) - 1", "sum(rows)")), + ) + assert result["status"] == "human_required" + assert result["reason"].startswith("patch_path_not_offered: got 'src/other.py'") + assert calls["pushes"] == [] + assert _event(storage)["chosen_path"] is None + + +def test_validation_resolves_against_the_chosen_file(monkeypatch) -> None: + storage, calls, result = _propose_service( + monkeypatch, + fetch=_service_fetch(), + run=_run(output=_service_output("tests/test_ledger.py", "== 3", "== 2")), + ) + assert result["status"] == "pr_opened" + assert result["path"] == "tests/test_ledger.py" + _cfg, _incident, _build, patch, contents = calls["pushes"][0] + assert patch.path == "tests/test_ledger.py" + assert contents == "def test_balance():\n assert balance([1, 2]) == 2\n" + assert _event(storage)["chosen_path"] == "tests/test_ledger.py" + diff --git a/tests/test_hermes_incident_issue.py b/tests/test_hermes_incident_issue.py new file mode 100644 index 0000000..e06eff2 --- /dev/null +++ b/tests/test_hermes_incident_issue.py @@ -0,0 +1,484 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from ariadne.services import hermes_incident_body as body +from ariadne.services import hermes_incident_issue as module + + +JOB = "metis" +INCIDENT_ID = f"{JOB}/12" +CLASSIFICATION = "dependency_resolution_failure" +TOKEN = "super-secret-token" + + +class FakeResponse: + def __init__(self, status_code: int, payload=None) -> None: # type: ignore[no-untyped-def] + self.status_code = status_code + self._payload = payload + + def json(self): # type: ignore[no-untyped-def] + if isinstance(self._payload, Exception): + raise self._payload + return self._payload + + +class FakeStorage: + def __init__(self, explode: bool = False) -> None: + self.events: list[tuple[str, dict]] = [] + self.explode = explode + + def record_event(self, event_type, detail) -> None: # type: ignore[no-untyped-def] + if self.explode: + raise RuntimeError("storage is down") + self.events.append((event_type, detail)) + + +def _install_http(monkeypatch, responses=None) -> dict: # type: ignore[no-untyped-def] + calls: dict = {"requests": [], "kwargs": None} + queue = list(responses or []) + + class FakeClient: + def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] + calls["kwargs"] = kwargs + + def __enter__(self): # type: ignore[no-untyped-def] + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def _next(self, method, url, kwargs): # type: ignore[no-untyped-def] + calls["requests"].append((method, url, kwargs)) + item = queue.pop(0) + if isinstance(item, Exception): + raise item + return item + + def get(self, url, **kwargs): # type: ignore[no-untyped-def] + return self._next("GET", url, kwargs) + + def post(self, url, **kwargs): # type: ignore[no-untyped-def] + return self._next("POST", url, kwargs) + + monkeypatch.setattr(module.httpx, "Client", FakeClient) + return calls + + +def _cfg(**overrides) -> dict: # type: ignore[no-untyped-def] + values = { + "gitea_base_url": "https://scm.example", + "gitea_token": TOKEN, + "owner": "bstein", + "repo": JOB, + "timeout_seconds": 7.5, + } + values.update(overrides) + return values + + +def _context(**overrides) -> dict: # type: ignore[no-untyped-def] + values = { + "incident_id": INCIDENT_ID, + "job": JOB, + "build_number": 12, + "build_url": "https://ci.example/job/metis/12/", + "classification": CLASSIFICATION, + "confidence": 0.91, + "first_failed_gate": "dependencies", + "reason": "no allowlisted action fits this failure", + "facts": [{"statement": "pip could not resolve urllib3", "source": "jenkins", "reference": "console"}], + "inferences": ["an upstream index published a broken constraint"], + "authorize_reason": "classification_not_actionable", + "run_id": "run-9", + } + values.update(overrides) + return values + + +def _issue(number: int, job: str = JOB, classification: str = CLASSIFICATION, incident: str = INCIDENT_ID) -> dict: + return { + "number": number, + "html_url": f"https://scm.example/bstein/{job}/issues/{number}", + "body": f"diagnosis\n\n{body.issue_marker(job, classification, incident)}", + } + + +def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] + values = { + "hermes_issues_enabled": True, + "hermes_issue_repos": {JOB: ("bstein", JOB)}, + "hermes_issue_dedupe_scope": "classification", + "hermes_issue_max_per_tick": 2, + "hermes_gitea_base_url": "https://scm.example", + "hermes_gitea_token": TOKEN, + } + values.update(overrides) + return SimpleNamespace(**values) + + +def _base(build_number: int = 12) -> dict: + return {"incident_id": f"{JOB}/{build_number}", "job": JOB, "build_number": build_number} + + +def _diagnosis(**overrides) -> dict: # type: ignore[no-untyped-def] + decision = SimpleNamespace( + classification=CLASSIFICATION, + confidence=0.91, + first_failed_gate="dependencies", + reason="the failure has no known signature", + facts=[SimpleNamespace(statement="pip failed", source="jenkins", reference="console")], + inferences=["upstream index broke"], + ) + values = { + "bundle": {"jenkins": {"url": "https://ci.example/job/metis/12/"}}, + "outcome": SimpleNamespace(decision=decision), + "authorize_reason": "classification_not_actionable", + "run_id": "run-9", + } + values.update(overrides) + return values + + +def test_marker_round_trips_through_the_body() -> None: + marker = body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID) + + assert body.parse_issue_marker(f"text\n{marker}\nmore") == { + "job": JOB, + "classification": CLASSIFICATION, + "incident": INCIDENT_ID, + } + + +def test_marker_flattens_values_that_would_break_the_comment() -> None: + marker = body.issue_marker("job\nname", "a --> b", INCIDENT_ID) + + assert "\n" not in marker + assert body.parse_issue_marker(marker) == { + "job": "job name", + "classification": "a b", + "incident": INCIDENT_ID, + } + + +def test_marker_parsing_tolerates_missing_and_non_string_bodies() -> None: + assert body.parse_issue_marker(None) is None + assert body.parse_issue_marker(12) is None + assert body.parse_issue_marker("a plain human issue") is None + + +def test_classification_scope_matches_a_different_incident_of_the_same_job(monkeypatch) -> None: + calls = _install_http(monkeypatch, [FakeResponse(200, [_issue(7, incident=f"{JOB}/9")])]) + + found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID) + + assert found == { + "found": True, + "issue_number": 7, + "url": "https://scm.example/bstein/metis/issues/7", + "error": None, + } + method, url, kwargs = calls["requests"][0] + assert (method, url) == ("GET", "https://scm.example/api/v1/repos/bstein/metis/issues") + assert kwargs["params"] == {"state": "open", "limit": 50} + assert kwargs["headers"] == {"Authorization": f"token {TOKEN}"} + + +def test_classification_scope_ignores_other_jobs_and_classifications(monkeypatch) -> None: + _install_http( + monkeypatch, + [FakeResponse(200, [_issue(7, job="lesavka"), _issue(8, classification="flaky_test")])], + ) + + found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID) + + assert found["found"] is False + assert found["error"] is None + + +def test_incident_scope_matches_only_the_same_incident(monkeypatch) -> None: + _install_http( + monkeypatch, + [FakeResponse(200, [_issue(7, incident=f"{JOB}/9"), _issue(9, incident=INCIDENT_ID)])], + ) + + found = module.find_open_incident_issue( + _cfg(dedupe_scope="incident"), JOB, CLASSIFICATION, INCIDENT_ID + ) + + assert found["found"] is True + assert found["issue_number"] == 9 + + +def test_lookup_returns_the_lowest_numbered_match(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(200, [_issue(31), _issue(12), _issue(20)])]) + + found = module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID) + + assert found["issue_number"] == 12 + + +def test_lookup_ignores_unmarked_and_malformed_issues(monkeypatch) -> None: + payload = [ + {"number": 3, "body": "a human filed this"}, + {"number": True, "body": _issue(4)["body"]}, + {"number": "5", "body": _issue(5)["body"]}, + "not an issue", + ] + _install_http(monkeypatch, [FakeResponse(200, payload)]) + + assert module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID)["found"] is False + + +def test_lookup_fails_open_on_http_parse_and_transport_errors(monkeypatch) -> None: + _install_http( + monkeypatch, + [ + FakeResponse(503, None), + FakeResponse(200, ValueError("no json")), + FakeResponse(200, {"issues": []}), + RuntimeError("connection reset"), + ], + ) + + statuses = [ + module.find_open_incident_issue(_cfg(), JOB, CLASSIFICATION, INCIDENT_ID) for _ in range(4) + ] + + assert [result["found"] for result in statuses] == [False, False, False, False] + assert statuses[0]["error"] == "open issue lookup http 503" + assert "parse failed" in statuses[1]["error"] + assert statuses[2]["error"] == "open issue payload is not a list" + assert "connection reset" in statuses[3]["error"] + + +def test_lookup_without_a_base_url_never_calls_gitea(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + + found = module.find_open_incident_issue(_cfg(gitea_base_url=" "), JOB, CLASSIFICATION, INCIDENT_ID) + + assert found == {"found": False, "issue_number": None, "url": None, "error": "gitea base url is empty"} + assert calls["requests"] == [] + + +def test_create_posts_the_issue_and_returns_its_identity(monkeypatch) -> None: + payload = {"number": 41, "html_url": "https://scm.example/bstein/metis/issues/41"} + calls = _install_http(monkeypatch, [FakeResponse(201, payload)]) + + created = module.create_incident_issue(_cfg(), _context()) + + assert created == {"issue_number": 41, "url": payload["html_url"], "error": None} + method, url, kwargs = calls["requests"][0] + assert (method, url) == ("POST", "https://scm.example/api/v1/repos/bstein/metis/issues") + assert kwargs["headers"] == {"Authorization": f"token {TOKEN}"} + assert kwargs["json"]["title"] == f"[hermes] {JOB} #12: {CLASSIFICATION}" + + +def test_create_reports_http_errors_exceptions_and_missing_numbers(monkeypatch) -> None: + _install_http( + monkeypatch, + [FakeResponse(422, None), RuntimeError("tls handshake failed"), FakeResponse(201, {"html_url": "u"})], + ) + + results = [module.create_incident_issue(_cfg(), _context()) for _ in range(3)] + + assert results[0] == {"issue_number": None, "url": None, "error": "issue create http 422"} + assert "tls handshake failed" in results[1]["error"] + assert results[2] == {"issue_number": None, "url": "u", "error": "issue create response had no number"} + + +def test_create_without_a_base_url_never_calls_gitea(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + + created = module.create_incident_issue(_cfg(gitea_base_url=""), _context()) + + assert created["error"] == "gitea base url is empty" + assert calls["requests"] == [] + + +def test_title_truncates_a_long_classification() -> None: + title = body.issue_title(_context(classification="x" * 400)) + + assert len(title) == 120 + assert title.startswith(f"[hermes] {JOB} #12: ") + assert title.endswith("...") + + +def test_body_carries_the_marker_the_build_url_and_the_no_write_footer() -> None: + rendered = body.issue_body(_context()) + + assert rendered.rstrip().endswith(body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID)) + assert "https://ci.example/job/metis/12/" in rendered + assert "Hermes has no write access to this repository" in rendered + assert "no files or infrastructure were changed" in rendered + assert "run `run-9`" in rendered + assert "/api/admin/audit/events" in rendered + assert "## Why a human is needed" in rendered + assert "classification_not_actionable" in rendered + assert "- **jenkins** — pip could not resolve urllib3 (`console`)" in rendered + + +def test_body_never_contains_the_gitea_token() -> None: + rendered = body.issue_body(_context(reason=f"failure while using {TOKEN[:4]}")) + + assert TOKEN not in rendered + + +def test_body_caps_facts_inferences_and_statement_length() -> None: + rendered = body.issue_body( + _context( + facts=[{"statement": "s" * 500, "source": "jenkins", "reference": f"r{i}"} for i in range(20)], + inferences=[f"inference {i}" for i in range(12)], + ) + ) + + assert rendered.count("- **jenkins**") == 10 + assert rendered.count("- inference ") == 6 + assert "s" * 500 not in rendered + assert "s" * 297 + "..." in rendered + + +def test_body_truncates_to_the_configured_cap_and_keeps_the_marker() -> None: + rendered = body.issue_body(_context(reason="r" * 20000), max_chars=1200) + + assert len(rendered) <= 1200 + assert "Truncated by Ariadne" in rendered + assert rendered.rstrip().endswith(body.issue_marker(JOB, CLASSIFICATION, INCIDENT_ID)) + + +def test_context_defaults_to_undiagnosed_when_no_decision_was_parsed() -> None: + context = module.issue_context(_base(), _diagnosis(outcome=None, run_id=None)) + + assert context["classification"] == "undiagnosed" + assert context["reason"] == "classification_not_actionable" + assert context["facts"] == [] + assert context["build_url"] == "https://ci.example/job/metis/12/" + + +def test_maybe_file_issue_files_and_records_one_event(monkeypatch) -> None: + payload = {"number": 5, "html_url": "https://scm.example/bstein/metis/issues/5"} + calls = _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, payload)]) + storage, tick = FakeStorage(), {} + + detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick) + + assert detail == { + "incident_id": INCIDENT_ID, + "job": JOB, + "build_number": 12, + "classification": CLASSIFICATION, + "issue_number": 5, + "url": payload["html_url"], + "skipped": False, + "error": None, + } + assert storage.events == [(module.ISSUE_EVENT_TYPE, detail)] + assert tick == {"issues_filed": 1} + assert [call[0] for call in calls["requests"]] == ["GET", "POST"] + + +def test_maybe_file_issue_skips_when_an_open_issue_already_exists(monkeypatch) -> None: + calls = _install_http(monkeypatch, [FakeResponse(200, [_issue(7, incident=f"{JOB}/9")])]) + storage, tick = FakeStorage(), {} + + detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick) + + assert detail["skipped"] is True + assert detail["issue_number"] == 7 + assert detail["error"] is None + assert [call[0] for call in calls["requests"]] == ["GET"] + assert tick == {} + + +def test_disabled_flag_files_nothing_and_calls_nothing(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + storage = FakeStorage() + + assert module.maybe_file_issue(storage, _settings(hermes_issues_enabled=False), _base(), _diagnosis(), {}) is None + assert calls["requests"] == [] + assert storage.events == [] + + +def test_unmapped_job_files_nothing_and_calls_nothing(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + storage = FakeStorage() + + result = module.maybe_file_issue( + storage, _settings(hermes_issue_repos={"other-job": ("bstein", "other")}), _base(), _diagnosis(), {} + ) + + assert result is None + assert calls["requests"] == [] + assert storage.events == [] + + +def test_max_per_tick_stops_a_burst_of_failures(monkeypatch) -> None: + payload = {"number": 5, "html_url": "u"} + calls = _install_http( + monkeypatch, + [FakeResponse(200, []), FakeResponse(201, payload), FakeResponse(200, []), FakeResponse(201, payload)], + ) + storage, tick = FakeStorage(), {} + config = _settings(hermes_issue_max_per_tick=2) + + results = [ + module.maybe_file_issue(storage, config, _base(build), _diagnosis(), tick) for build in (12, 13, 14) + ] + + assert [result is None for result in results] == [False, False, True] + assert tick == {"issues_filed": 2} + assert len(calls["requests"]) == 4 + + +def test_lookup_failure_fails_open_and_still_files(monkeypatch) -> None: + calls = _install_http(monkeypatch, [FakeResponse(500, None), FakeResponse(201, {"number": 6})]) + storage, tick = FakeStorage(), {} + + detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick) + + assert detail["skipped"] is False + assert detail["issue_number"] == 6 + assert [call[0] for call in calls["requests"]] == ["GET", "POST"] + + +def test_create_failure_is_recorded_on_the_event(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(403, None)]) + storage, tick = FakeStorage(), {} + + detail = module.maybe_file_issue(storage, _settings(), _base(), _diagnosis(), tick) + + assert detail["issue_number"] is None + assert detail["error"] == "issue create http 403" + assert detail["skipped"] is False + assert tick == {"issues_filed": 1} + + +def test_maybe_file_issue_never_raises(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, {"number": 5})]) + + assert module.maybe_file_issue(FakeStorage(explode=True), _settings(), _base(), _diagnosis(), {}) is None + + +def test_repo_map_entries_accept_pairs_dicts_and_strings(monkeypatch) -> None: + calls = _install_http(monkeypatch, [FakeResponse(200, []), FakeResponse(201, {"number": 5})] * 2) + for repos in ({JOB: {"owner": "bstein", "repo": "metis"}}, {JOB: "bstein/metis"}): + module.maybe_file_issue( + FakeStorage(), _settings(hermes_issue_repos=repos), _base(), _diagnosis(), {} + ) + + assert [call[1] for call in calls["requests"]] == [ + "https://scm.example/api/v1/repos/bstein/metis/issues" + ] * 4 + + +def test_broken_repo_map_entries_file_nothing(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + + for repos in ({JOB: "bstein"}, {JOB: ("bstein", "")}, {JOB: 12}, {JOB: None}): + assert ( + module.maybe_file_issue( + FakeStorage(), _settings(hermes_issue_repos=repos), _base(), _diagnosis(), {} + ) + is None + ) + assert calls["requests"] == [] diff --git a/tests/test_settings.py b/tests/test_settings.py index 570c533..8c79981 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -158,3 +158,46 @@ def test_hermes_action_defaults_stay_conservative(monkeypatch) -> None: assert cfg.hermes_allowed_actions == ["repair_demo_fixture"] assert cfg.hermes_parameterized_jobs == ["hermes-triage-demo"] + + +def test_from_env_includes_hermes_issue_settings(monkeypatch) -> None: + monkeypatch.setenv("ARIADNE_HERMES_ISSUES_ENABLED", "true") + monkeypatch.setenv( + "ARIADNE_HERMES_ISSUE_REPOS", + " metis=bstein/metis , lesavka=bstein/lesavka , broken , nope= , =bstein/x ", + ) + monkeypatch.setenv("ARIADNE_HERMES_ISSUE_DEDUPE_SCOPE", " Incident ") + monkeypatch.setenv("ARIADNE_HERMES_ISSUE_MAX_PER_TICK", "5") + + cfg = Settings.from_env() + + assert cfg.hermes_issues_enabled is True + assert cfg.hermes_issue_repos == { + "metis": ("bstein", "metis"), + "lesavka": ("bstein", "lesavka"), + } + assert cfg.hermes_issue_dedupe_scope == "incident" + assert cfg.hermes_issue_max_per_tick == 5 + + +def test_hermes_issue_defaults_stay_opt_in(monkeypatch) -> None: + for name in ( + "ARIADNE_HERMES_ISSUES_ENABLED", + "ARIADNE_HERMES_ISSUE_REPOS", + "ARIADNE_HERMES_ISSUE_DEDUPE_SCOPE", + "ARIADNE_HERMES_ISSUE_MAX_PER_TICK", + ): + monkeypatch.delenv(name, raising=False) + + cfg = Settings.from_env() + + assert cfg.hermes_issues_enabled is False + assert cfg.hermes_issue_repos == {} + assert cfg.hermes_issue_dedupe_scope == "classification" + assert cfg.hermes_issue_max_per_tick == 2 + + +def test_hermes_issue_dedupe_scope_rejects_unknown_values(monkeypatch) -> None: + monkeypatch.setenv("ARIADNE_HERMES_ISSUE_DEDUPE_SCOPE", "everything") + + assert Settings.from_env().hermes_issue_dedupe_scope == "classification"