From 2cb5d50fa86ee474e4819f0f520e7fc241d9c757 Mon Sep 17 00:00:00 2001 From: codex Date: Wed, 5 Aug 2026 19:18:26 -0300 Subject: [PATCH] feat(hermes-code): bounded source-patch proposal path Hermes proposes a minimal anchored patch; Ariadne validates it structurally and opens a pull request for human review. Nothing merges automatically and Hermes never holds Git credentials or executes anything. - hermes_code_patch: response parsing + patch validation (path prefix/suffix allowlist, size and changed-line caps, exact-single-occurrence anchor, replacement-differs) and exact application - hermes_code_repair: Gitea contents-API client (fetch, branch push with hermes-repair/ prefix and base-branch refusal, PR creation) - hermes_code_flow: proposal orchestration + audit event, no patch bodies or tokens recorded - hermes_autotriage: code-path branch for the configured repo job; a PR records human_required/code_fix_proposed, never auto-resolution, and is kept out of the fixture-action accounting - settings: ARIADNE_HERMES_CODE_* configuration, disabled by default 74 new tests; 214 pass in the hermes suite. Co-Authored-By: Claude Fable 5 --- ariadne/services/hermes_autotriage.py | 37 ++- ariadne/services/hermes_code_flow.py | 205 ++++++++++++++ ariadne/services/hermes_code_patch.py | 254 +++++++++++++++++ ariadne/services/hermes_code_repair.py | 247 ++++++++++++++++ ariadne/settings.py | 15 + ariadne/settings_sections.py | 25 ++ tests/test_hermes_autotriage.py | 12 + tests/test_hermes_code_flow.py | 375 +++++++++++++++++++++++++ tests/test_hermes_code_patch.py | 253 +++++++++++++++++ tests/test_hermes_code_repair.py | 258 +++++++++++++++++ tests/test_settings.py | 30 ++ 11 files changed, 1710 insertions(+), 1 deletion(-) create mode 100644 ariadne/services/hermes_code_flow.py create mode 100644 ariadne/services/hermes_code_patch.py create mode 100644 ariadne/services/hermes_code_repair.py create mode 100644 tests/test_hermes_code_flow.py create mode 100644 tests/test_hermes_code_patch.py create mode 100644 tests/test_hermes_code_repair.py diff --git a/ariadne/services/hermes_autotriage.py b/ariadne/services/hermes_autotriage.py index 1c8f927..6d5881a 100644 --- a/ariadne/services/hermes_autotriage.py +++ b/ariadne/services/hermes_autotriage.py @@ -8,7 +8,7 @@ import httpx from ..settings import settings from ..utils.logging import get_logger -from . import hermes_agent_client, hermes_autotriage_repair +from . import hermes_agent_client, hermes_autotriage_repair, hermes_code_flow from . import hermes_autotriage_decision as hermes_decision from . import hermes_autotriage_evidence as hermes_evidence from .hermes_autotriage_metrics import ( @@ -29,6 +29,7 @@ DIAGNOSIS_EVENT_TYPE = "hermes_autotriage_diagnosis" ACTION_EVENT_TYPE = "hermes_autotriage_action" EXPECTED_CLASSIFICATION = "known_demo_fixture_failure" REBUILD_FAILED_REASON = "repair rebuild failed" +CODE_FIX_PROPOSED_REASON = "code_fix_proposed" _LAST_BUILD_TREE = "lastBuild[number,result,building,timestamp,duration,url]" _EVENT_SCAN_LIMIT = 500 @@ -239,6 +240,8 @@ def _run_pipeline( phase_started = time.time() bundle = hermes_evidence.collect_evidence(incident_id, job, last_build) HERMES_TRIAGE_DURATION_SECONDS.labels(phase="evidence").set(time.time() - phase_started) + if settings.hermes_code_enabled and job == settings.hermes_code_job: + return _propose_code_fix(storage, base, bundle) phase_started = time.time() run = hermes_agent_client.run_triage(_hermes_run_config(), _build_prompt(incident_id, bundle)) HERMES_TRIAGE_DURATION_SECONDS.labels(phase="diagnosis").set(time.time() - phase_started) @@ -266,6 +269,38 @@ def _run_pipeline( return _execute_action(storage, base, outcome) +def _propose_code_fix(storage: Any, base: dict[str, Any], bundle: dict[str, Any]) -> dict[str, Any]: + """Run the code-repair proposal flow and park the incident with humans. + + A pull request always awaits human review, so the incident lands in + human_required either way; this path never touches the fixture-repair + action accounting. + """ + + result = hermes_code_flow.propose_code_fix( + storage, + str(base["incident_id"]), + str(base["job"]), + _int_value(base["build_number"]), + bundle, + _hermes_run_config(), + hermes_code_flow.code_config(settings), + ) + if result.get("status") == "pr_opened": + reason = CODE_FIX_PROPOSED_REASON + phase = { + "reason": reason, + "branch": result.get("branch"), + "pr_number": result.get("pr_number"), + "url": result.get("url"), + } + else: + reason = str(result.get("reason") or "code_fix_not_proposed") + phase = {"reason": reason} + _record_incident(storage, base, "human_required", phase) + return {"status": "human_required", "incident_id": str(base["incident_id"]), "reason": reason} + + def _execute_action(storage: Any, base: dict[str, Any], outcome: Any) -> dict[str, Any]: """Run the authorized repair action and request the verification rebuild.""" diff --git a/ariadne/services/hermes_code_flow.py b/ariadne/services/hermes_code_flow.py new file mode 100644 index 0000000..4dd60b4 --- /dev/null +++ b/ariadne/services/hermes_code_flow.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +from dataclasses import dataclass +import json +from typing import Any + +from ..utils.logging import get_logger +from . import hermes_agent_client, hermes_code_patch, hermes_code_repair + + +logger = get_logger(__name__) + +CODE_PROPOSAL_EVENT_TYPE = "hermes_autotriage_code_proposal" + +_RUN_COMPLETED = "completed" +_GITEA_TIMEOUT_SECONDS = 15.0 + +_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. +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. + +Failing test evidence bundle: +__BUNDLE__ + +Current content of the candidate file __PATH__: +__FILE_CONTENTS__""" + + +@dataclass(frozen=True) +class _Proposal: + patch: hermes_code_patch.ProposedPatch + analysis: str + patched_contents: str + run_id: str | None + + +def propose_code_fix( # noqa: PLR0913 - flow contract carries the full incident context + storage: Any, + incident_id: str, + job: str, + build_number: int, + bundle: dict, + hermes_cfg: dict, + code_cfg: dict, +) -> dict[str, Any]: + """Run the bounded code-repair proposal flow for one failed build. + + Inputs: the incident event storage, incident identity (id/job/build), the + already-built evidence bundle, the Hermes agent run config, and the code + 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. + """ + + result, event = _propose(incident_id, build_number, bundle, hermes_cfg, code_cfg) + storage.record_event( + CODE_PROPOSAL_EVENT_TYPE, + {"incident_id": incident_id, "job": job, "build_number": build_number, **event}, + ) + logger.info( + "hermes code fix proposal finished", + extra={"event": "hermes_code_flow", "status": result["status"], "incident_id": incident_id}, + ) + return result + + +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). + """ + + 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, + } + + +def _propose( + incident_id: str, build_number: int, bundle: dict, hermes_cfg: dict, code_cfg: dict +) -> tuple[dict[str, Any], dict[str, Any]]: + """Run fetch, diagnosis, validation, and publication for one incident.""" + + 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) + ) + 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) + if patch is None: + return _human_required(reject_reason, run.run_id, validated=False) + proposal = _Proposal( + patch=patch, + analysis=hermes_code_patch.parse_analysis(run.output), + patched_contents=hermes_code_patch.apply_patch(contents, patch), + run_id=run.run_id, + ) + return _publish(code_cfg, incident_id, build_number, proposal) + + +def _validated_patch( + raw_output: str, incident_id: str, path: str, code_cfg: dict, contents: str +) -> tuple[hermes_code_patch.ProposedPatch | None, str]: + """Parse and gate the patch response against the fetched candidate file.""" + + 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) + 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 +) -> 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 + ) + 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 + ) + if pull.get("error"): + return _human_required( + f"pull_request_failed: {pull.get('error')}", proposal.run_id, validated=True, branch=branch + ) + result = { + "status": "pr_opened", + "branch": branch, + "pr_number": pull.get("pr_number"), + "url": pull.get("url"), + "path": proposal.patch.path, + "run_id": proposal.run_id, + } + event = { + "run_id": proposal.run_id, + "validated": True, + "reject_reason": None, + "branch": branch, + "pr_number": pull.get("pr_number"), + "url": pull.get("url"), + } + return result, event + + +def _human_required( + reason: str, run_id: str | None, validated: bool, branch: str | None = None +) -> tuple[dict[str, Any], dict[str, Any]]: + """Build the human_required result and its proposal event detail.""" + + result = {"status": "human_required", "reason": reason} + event = { + "run_id": run_id, + "validated": validated, + "reject_reason": reason, + "branch": branch, + "pr_number": None, + "url": None, + } + return result, event + + +def _build_patch_prompt(incident_id: str, bundle: dict, code_cfg: dict, contents: str) -> str: + """Render the frozen patch prompt with incident, bundle, and file context.""" + + compact = json.dumps(bundle, separators=(",", ":"), ensure_ascii=True) + 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("__BUNDLE__", compact) + .replace("__FILE_CONTENTS__", contents) + ) diff --git a/ariadne/services/hermes_code_patch.py b/ariadne/services/hermes_code_patch.py new file mode 100644 index 0000000..dbfac3b --- /dev/null +++ b/ariadne/services/hermes_code_patch.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +from dataclasses import dataclass +import json +from typing import Any + + +VALID_REASON = "valid" + +_TOP_LEVEL_KEYS = {"incident_id", "analysis", "patch", "human_required", "reason"} +_PATCH_KEYS = {"path", "original", "replacement", "rationale"} +_STRING_FIELDS = ("incident_id", "analysis", "reason") +_NONEMPTY_PATCH_FIELDS = ("path", "original", "replacement") + + +@dataclass(frozen=True) +class ProposedPatch: + """Represent one minimal source patch proposed by Hermes. + + Inputs: a validated `patch` object from the frozen response schema. + Outputs: the repository-relative path, the exact snippet expected in the + current file, its replacement, and the model's rationale. + """ + + path: str + original: str + replacement: str + rationale: str + + +@dataclass(frozen=True) +class PatchOutcome: + """Represent the result of parsing a Hermes patch response. + + Inputs: raw Hermes output run through `parse_patch_response`. + Outputs: validity, the proposed patch when valid, and a specific reject + reason when invalid. + """ + + valid: bool + patch: ProposedPatch | None + reject_reason: str | None + + +def parse_patch_response(raw_output: str, expected_incident_id: str) -> PatchOutcome: + """Extract, parse, and validate a Hermes patch response payload. + + Inputs: raw model output (JSON, optionally fenced or wrapped in prose) + and the incident id the response must reference. + Outputs: a PatchOutcome; any schema violation, incident mismatch, missing + patch, or human_required escalation yields valid=False with a specific + reject_reason. Never raises. + """ + + candidate = _extract_json_object(str(raw_output or "")) + if candidate is None: + return _rejected("no_json_object_found") + try: + payload = json.loads(candidate) + except json.JSONDecodeError as exc: + return _rejected(f"invalid_json: {exc}") + error = _validate_payload(payload, expected_incident_id) + if error: + return _rejected(error) + patch = payload["patch"] + return PatchOutcome( + valid=True, + patch=ProposedPatch( + path=patch["path"], + original=patch["original"], + replacement=patch["replacement"], + rationale=patch["rationale"], + ), + reject_reason=None, + ) + + +def parse_analysis(raw_output: str) -> str: + """Return the analysis string from a patch response, or "" when absent. + + Inputs: the same raw Hermes output given to `parse_patch_response`. + Outputs: the top-level analysis string when one can be extracted; used + only for human-facing pull-request context. Never raises. + """ + + candidate = _extract_json_object(str(raw_output or "")) + if candidate is None: + return "" + try: + payload = json.loads(candidate) + except json.JSONDecodeError: + return "" + analysis = payload.get("analysis") if isinstance(payload, dict) else None + return analysis if isinstance(analysis, str) else "" + + +def validate_patch(patch: ProposedPatch, cfg: dict, file_contents: str) -> tuple[bool, str]: + """Apply every patch gate and name the first failing gate. + + Inputs: the proposed patch, cfg with allowed_path_prefixes, + allowed_suffixes, max_patch_bytes, and max_changed_lines, plus the + current contents of the candidate file. + Outputs: (ok, reason); reason is "valid" only when every gate passes, + otherwise it names the first failing gate. + """ + + prefixes = [str(item) for item in (cfg.get("allowed_path_prefixes") or [])] + suffixes = [str(item) for item in (cfg.get("allowed_suffixes") or [])] + max_bytes = _int_value(cfg.get("max_patch_bytes")) + max_lines = _int_value(cfg.get("max_changed_lines")) + occurrences = file_contents.count(patch.original) + changed_lines = max(patch.original.count("\n"), patch.replacement.count("\n")) + 1 + gates: list[tuple[bool, str]] = [ + (_path_is_safe(patch.path), "path_unsafe"), + (any(patch.path.startswith(prefix) for prefix in prefixes), "path_prefix_not_allowed"), + (any(patch.path.endswith(suffix) for suffix in suffixes), "path_suffix_not_allowed"), + (len(patch.original) + len(patch.replacement) <= max_bytes, "patch_too_large"), + (changed_lines <= max_lines, "too_many_changed_lines"), + (occurrences != 0, "original_missing"), + (occurrences <= 1, "original_ambiguous"), + (patch.replacement != patch.original, "replacement_identical"), + ] + for passed, reason in gates: + if not passed: + return False, reason + return True, VALID_REASON + + +def apply_patch(file_contents: str, patch: ProposedPatch) -> str: + """Replace the patch's original snippet with its replacement exactly once. + + Inputs: the current file contents and a validated ProposedPatch. + Outputs: the patched contents. Raises ValueError unless the original + snippet appears exactly once. + """ + + occurrences = file_contents.count(patch.original) + if occurrences != 1: + raise ValueError(f"original snippet must appear exactly once, found {occurrences}") + return file_contents.replace(patch.original, patch.replacement) + + +def _rejected(reason: str) -> PatchOutcome: + """Build the invalid PatchOutcome for one reject reason.""" + + return PatchOutcome(valid=False, patch=None, reject_reason=reason) + + +def _extract_json_object(raw: str) -> str | None: + """Return the first balanced top-level JSON object in raw text.""" + + depth = 0 + start = -1 + in_string = False + escaped = False + for index, char in enumerate(raw): + if in_string: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + in_string = False + continue + if char == '"' and depth > 0: + in_string = True + elif char == "{": + if depth == 0: + start = index + depth += 1 + elif char == "}" and depth > 0: + depth -= 1 + if depth == 0: + return raw[start : index + 1] + return None + + +def _validate_payload(payload: dict[str, Any], expected_incident_id: str) -> str | None: + """Validate the frozen patch-response schema, returning the first error.""" + + shape_error = ( + _validate_keys(payload) + or _validate_scalar_fields(payload) + or _validate_patch_field(payload["patch"]) + ) + if shape_error: + return shape_error + if payload["incident_id"] != expected_incident_id: + return f"incident_id_mismatch: got {payload['incident_id']!r} expected {expected_incident_id!r}" + if payload["human_required"]: + return "human_required" + if payload["patch"] is None: + return "patch_missing" + return None + + +def _validate_keys(payload: dict[str, Any]) -> str | None: + """Require exactly the frozen top-level keys.""" + + keys = set(payload) + missing = sorted(_TOP_LEVEL_KEYS - keys) + if missing: + return "missing_keys: " + ", ".join(missing) + extra = sorted(keys - _TOP_LEVEL_KEYS) + if extra: + return "unexpected_keys: " + ", ".join(extra) + return None + + +def _validate_scalar_fields(payload: dict[str, Any]) -> str | None: + """Type-check the scalar top-level fields of the response.""" + + for key in _STRING_FIELDS: + if not isinstance(payload[key], str): + return f"field_type_invalid: {key} must be a string" + if not isinstance(payload["human_required"], bool): + return "field_type_invalid: human_required must be a boolean" + return None + + +def _validate_patch_field(patch: Any) -> str | None: + """Validate the patch object shape, tolerating an explicit null.""" + + if patch is None: + return None + if not isinstance(patch, dict): + return "patch_invalid: must be an object or null" + if set(patch) != _PATCH_KEYS: + return "patch_invalid: must have exactly path, original, replacement, rationale" + if not all(isinstance(patch[key], str) for key in _PATCH_KEYS): + return "patch_invalid: fields must be strings" + for key in _NONEMPTY_PATCH_FIELDS: + if not patch[key]: + return f"patch_field_empty: {key}" + return None + + +def _path_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 _int_value(value: Any) -> int: + """Coerce a value to int, defaulting to zero.""" + + try: + return int(value) + except (TypeError, ValueError): + return 0 diff --git a/ariadne/services/hermes_code_repair.py b/ariadne/services/hermes_code_repair.py new file mode 100644 index 0000000..5f53a07 --- /dev/null +++ b/ariadne/services/hermes_code_repair.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +import base64 +from typing import Any + +import httpx + +from ..utils.logging import get_logger + + +logger = get_logger(__name__) + +HTTP_OK = 200 +HTTP_CREATED = 201 +HTTP_NOT_FOUND = 404 +HTTP_CONFLICT = 409 +HTTP_UNPROCESSABLE = 422 + +_DEFAULT_TIMEOUT_SECONDS = 15.0 +_PROTECTED_BRANCHES = {"master", "main"} +_COMMIT_IDENTITY = {"name": "Hermes Agent", "email": "hermes@bstein.dev"} +_COMMIT_OK_STATUSES = {HTTP_OK, HTTP_CREATED} +_BRANCH_RETRY_STATUSES = {HTTP_NOT_FOUND, HTTP_UNPROCESSABLE} + + +def fetch_file(cfg: dict, path: str) -> tuple[str | None, str | None]: + """Fetch one file's raw contents from Gitea at the base branch. + + Inputs: `cfg` with gitea_base_url, gitea_token, owner, repo, base_branch, + and timeout_seconds; the repository-relative file path. + Outputs: (contents, error) with exactly one side set. Never raises and + never logs the token. + """ + + base_url = _base_url(cfg) + if not base_url: + return None, "gitea base url is empty" + url = f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/raw/{path}" + try: + with httpx.Client(timeout=_timeout(cfg)) as client: + response = client.get(url, headers=_headers(cfg), params={"ref": _base_branch(cfg)}) + except Exception as exc: + return None, f"file fetch failed: {exc}" + if response.status_code != HTTP_OK: + return None, f"file fetch http {response.status_code}" + return response.text, None + + +def push_branch( + cfg: dict, incident_id: str, build_number: int, patch: Any, patched_contents: str +) -> dict[str, Any]: + """Commit the patched file to a fresh repair branch via the Gitea API. + + Inputs: `cfg` as for `fetch_file`; the incident id, the failed build + number that names the branch, the validated ProposedPatch, and the fully + patched file contents. Outputs: {"branch", "committed", "error"}. Never + force-pushes, refuses master/main or empty branch names, never raises, + and never logs the token. + """ + + branch = _branch_name(build_number) + if not branch.strip() or branch in _PROTECTED_BRANCHES: + return {"branch": branch, "committed": False, "error": f"refusing branch {branch!r}"} + base_url = _base_url(cfg) + if not base_url: + return {"branch": branch, "committed": False, "error": "gitea base url is empty"} + try: + with httpx.Client(timeout=_timeout(cfg)) as client: + error = _commit_patch(client, cfg, branch, incident_id, patch, patched_contents) + except Exception as exc: + error = f"branch push failed: {exc}" + if error is not None: + logger.info( + "hermes code repair branch push failed", + extra={"event": "hermes_code_repair", "status": "error", "branch": branch, "detail": error}, + ) + return {"branch": branch, "committed": error is None, "error": error} + + +def open_pull_request( # noqa: PLR0913 - flow contract mirrors push_branch identity fields + cfg: dict, incident_id: str, build_number: int, branch: str, patch: Any, analysis: str +) -> dict[str, Any]: + """Open the human-review pull request for a pushed repair branch. + + Inputs: `cfg` as for `fetch_file`; the incident id, failed build number, + pushed branch name, the validated ProposedPatch, and the model analysis. + Outputs: {"pr_number", "url", "error"}; an existing PR reported by a 409 + counts as success when the payload identifies it. Never raises and never + logs the token. + """ + + base_url = _base_url(cfg) + if not base_url: + return {"pr_number": None, "url": None, "error": "gitea base url is empty"} + payload = { + "head": branch, + "base": _base_branch(cfg), + "title": f"fix(hermes): repair {incident_id}", + "body": _pr_body(incident_id, patch, analysis), + } + try: + with httpx.Client(timeout=_timeout(cfg)) as client: + response = client.post( + f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/pulls", + headers=_headers(cfg), + json=payload, + ) + except Exception as exc: + return {"pr_number": None, "url": None, "error": f"pull request failed: {exc}"} + if response.status_code == HTTP_CREATED: + return _pr_result(response) + if response.status_code == HTTP_CONFLICT: + existing = _pr_result(response) + if existing["pr_number"] is not None: + return existing + return {"pr_number": None, "url": None, "error": "pull request already exists"} + return {"pr_number": None, "url": None, "error": f"pull request http {response.status_code}"} + + +def _commit_patch( # noqa: PLR0913 - commit needs the full branch/identity context + client: httpx.Client, + cfg: dict, + branch: str, + incident_id: str, + patch: Any, + patched_contents: str, +) -> str | None: + """PUT the patched file onto the new branch, trying both branch shapes.""" + + url = f"{_base_url(cfg)}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/contents/{patch.path}" + sha, sha_error = _file_sha(client, cfg, patch.path) + if sha_error is not None: + return sha_error + body = { + "message": f"fix(hermes): {patch.rationale} (incident {incident_id})", + "content": base64.b64encode(patched_contents.encode("utf-8")).decode("ascii"), + "sha": sha, + "author": dict(_COMMIT_IDENTITY), + "committer": dict(_COMMIT_IDENTITY), + } + first = client.put(url, headers=_headers(cfg), json={**body, "branch": _base_branch(cfg), "new_branch": branch}) + if first.status_code in _COMMIT_OK_STATUSES: + return None + if first.status_code not in _BRANCH_RETRY_STATUSES: + return f"commit http {first.status_code}" + fallback = client.put(url, headers=_headers(cfg), json={**body, "branch": branch}) + if fallback.status_code in _COMMIT_OK_STATUSES: + return None + return f"commit http {first.status_code} then fallback http {fallback.status_code}" + + +def _file_sha(client: httpx.Client, cfg: dict, path: str) -> tuple[str | None, str | None]: + """Read the file's current blob sha at the base branch.""" + + url = f"{_base_url(cfg)}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/contents/{path}" + response = client.get(url, headers=_headers(cfg), params={"ref": _base_branch(cfg)}) + if response.status_code != HTTP_OK: + return None, f"file sha http {response.status_code}" + sha = _json_payload(response).get("sha") + if not sha: + return None, "file sha missing from contents response" + return str(sha), None + + +def _pr_result(response: Any) -> dict[str, Any]: + """Map a pull-request response payload to the result shape.""" + + payload = _json_payload(response) + number = payload.get("number") + url = str(payload.get("html_url") or "") or None + if isinstance(number, bool) or not isinstance(number, int): + return {"pr_number": None, "url": url, "error": None} + return {"pr_number": number, "url": url, "error": None} + + +def _pr_body(incident_id: str, patch: Any, analysis: str) -> str: + """Render the markdown pull-request body for human review.""" + + return "\n".join( + [ + f"## Hermes repair proposal for incident {incident_id}", + "", + f"**Incident:** {incident_id}", + f"**File:** `{patch.path}`", + f"**Analysis:** {analysis}", + f"**Rationale:** {patch.rationale}", + "", + "Proposed by Hermes; validated and pushed by Ariadne; " + "requires human review — no automatic merge.", + ] + ) + + +def _branch_name(build_number: int) -> str: + """Derive the repair branch name for one failed build number.""" + + return f"hermes-repair/{build_number}" + + +def _json_payload(response: Any) -> dict[str, Any]: + """Return a response's JSON body as a dict, tolerating garbage.""" + + try: + payload = response.json() + except Exception: + return {} + return payload if isinstance(payload, dict) else {} + + +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 _base_branch(cfg: dict) -> str: + """Return the configured base branch name.""" + + return str(cfg.get("base_branch") or "") + + +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 76fee9f..0be6403 100644 --- a/ariadne/settings.py +++ b/ariadne/settings.py @@ -10,6 +10,7 @@ from .settings_sections import ( _firefly_config, _game_stream_config, _hermes_autotriage_config, + _hermes_code_config, _image_sweeper_config, _jenkins_build_weather_config, _jenkins_workspace_cleanup_config, @@ -192,6 +193,18 @@ class Settings: hermes_demo_namespace: str hermes_demo_fixture_configmap: str hermes_repair_image: str + hermes_code_enabled: bool + hermes_code_job: str + hermes_code_owner: str + hermes_code_repo: str + hermes_code_base_branch: str + hermes_code_candidate_path: str + hermes_code_allowed_prefixes: list[str] + hermes_code_allowed_suffixes: list[str] + hermes_code_max_patch_bytes: int + hermes_code_max_changed_lines: int + hermes_gitea_base_url: str + hermes_gitea_token: str vaultwarden_namespace: str vaultwarden_pod_label: str @@ -310,6 +323,7 @@ class Settings: jenkins_workspace_cleanup_cfg = _jenkins_workspace_cleanup_config() testing_triage_cfg = _testing_triage_config() hermes_autotriage_cfg = _hermes_autotriage_config() + hermes_code_cfg = _hermes_code_config() vaultwarden_cfg = _vaultwarden_config() schedule_cfg = _schedule_config() cluster_cfg = _cluster_state_config() @@ -355,6 +369,7 @@ class Settings: **jenkins_workspace_cleanup_cfg, **testing_triage_cfg, **hermes_autotriage_cfg, + **hermes_code_cfg, **vaultwarden_cfg, **schedule_cfg, **cluster_cfg, diff --git a/ariadne/settings_sections.py b/ariadne/settings_sections.py index f7ac66c..55a66d3 100644 --- a/ariadne/settings_sections.py +++ b/ariadne/settings_sections.py @@ -282,6 +282,31 @@ def _hermes_autotriage_config() -> dict[str, Any]: } +def _hermes_code_config() -> dict[str, Any]: + return { + "hermes_code_enabled": _env_bool("ARIADNE_HERMES_CODE_ENABLED", "false"), + "hermes_code_job": _env("ARIADNE_HERMES_CODE_JOB", "hermes-code-demo"), + "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_candidate_path": _env("ARIADNE_HERMES_CODE_CANDIDATE_PATH", "src/discount.py"), + "hermes_code_allowed_prefixes": [ + item.strip() + for item in _env("ARIADNE_HERMES_CODE_ALLOWED_PREFIXES", "src/").split(",") + if item.strip() + ], + "hermes_code_allowed_suffixes": [ + item.strip() + for item in _env("ARIADNE_HERMES_CODE_ALLOWED_SUFFIXES", ".py").split(",") + if item.strip() + ], + "hermes_code_max_patch_bytes": _env_int("ARIADNE_HERMES_CODE_MAX_PATCH_BYTES", 4000), + "hermes_code_max_changed_lines": _env_int("ARIADNE_HERMES_CODE_MAX_CHANGED_LINES", 20), + "hermes_gitea_base_url": _env("ARIADNE_HERMES_GITEA_BASE_URL", "https://scm.bstein.dev").rstrip("/"), + "hermes_gitea_token": _env("ARIADNE_HERMES_GITEA_TOKEN", ""), + } + + def _vaultwarden_config() -> dict[str, Any]: return { "vaultwarden_namespace": _env("VAULTWARDEN_NAMESPACE", "vaultwarden"), diff --git a/tests/test_hermes_autotriage.py b/tests/test_hermes_autotriage.py index 7a1998b..1095325 100644 --- a/tests/test_hermes_autotriage.py +++ b/tests/test_hermes_autotriage.py @@ -54,6 +54,18 @@ def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] "hermes_demo_namespace": "hermes-triage-demo", "hermes_demo_fixture_configmap": "hermes-triage-demo-fixture", "hermes_repair_image": "busybox:1.37", + "hermes_code_enabled": False, + "hermes_code_job": "hermes-code-demo", + "hermes_code_owner": "bstein", + "hermes_code_repo": "hermes-code-demo", + "hermes_code_base_branch": "master", + "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": "gitea-token", "jenkins_base_url": "https://ci.example", "jenkins_api_user": "user", "jenkins_api_token": "token", diff --git a/tests/test_hermes_code_flow.py b/tests/test_hermes_code_flow.py new file mode 100644 index 0000000..80a8d6f --- /dev/null +++ b/tests/test_hermes_code_flow.py @@ -0,0 +1,375 @@ +from __future__ import annotations + +import json +from types import SimpleNamespace + +from ariadne.services import hermes_autotriage as autotriage +from ariadne.services import hermes_code_flow as module +from ariadne.services.hermes_agent_client import HermesRunResult + + +JOB = "hermes-code-demo" +INCIDENT_ID = f"{JOB}/7" +FILE_CONTENTS = "def discount(price):\n return price * 0.5\n" +BUNDLE = {"incident_id": INCIDENT_ID, "jenkins": {"job": JOB}, "log_evidence": {"records": []}} + + +class FakeStorage: + def __init__(self) -> None: + self.events: list[dict] = [] + + def record_event(self, event_type, detail) -> None: # type: ignore[no-untyped-def] + self.events.append({"event_type": event_type, "detail": detail}) + + def list_events(self, limit=200, event_type=None): # type: ignore[no-untyped-def] + rows = [ + dict(row) + for row in reversed(self.events) + if event_type is None or row["event_type"] == event_type + ] + return rows[:limit] + + +def _hermes_cfg() -> dict: + return {"base_url": "http://hermes:8642", "api_key": "key", "total_timeout_seconds": 420.0} + + +def _code_cfg(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "candidate_path": "src/discount.py", + "allowed_path_prefixes": ["src/"], + "allowed_suffixes": [".py"], + "max_patch_bytes": 4000, + "max_changed_lines": 20, + "gitea_base_url": "https://scm.example", + "gitea_token": "secret-token", + "owner": "bstein", + "repo": "hermes-code-demo", + "base_branch": "master", + "timeout_seconds": 15.0, + } + base.update(overrides) + return base + + +def _model_output(**overrides) -> str: # type: ignore[no-untyped-def] + payload = { + "incident_id": INCIDENT_ID, + "analysis": "The multiplier regressed to 0.5.", + "patch": { + "path": "src/discount.py", + "original": "return price * 0.5", + "replacement": "return price * 0.9", + "rationale": "restore the intended discount", + }, + "human_required": False, + "reason": "small localized fix", + } + payload.update(overrides) + return json.dumps(payload) + + +def _run(status: str = "completed", output=None) -> HermesRunResult: # type: ignore[no-untyped-def] + return HermesRunResult( + status=status, + output=output, + run_id="run-1", + session_id="sess-1", + error=None, + duration_seconds=1.5, + denied_approvals=0, + ) + + +def _install(monkeypatch, *, fetch=None, run=None, push=None, pull=None) -> dict: # type: ignore[no-untyped-def] + calls: dict = {"fetches": [], "runs": [], "pushes": [], "pulls": []} + + def fake_fetch(cfg, path): # type: ignore[no-untyped-def] + calls["fetches"].append((cfg, path)) + return fetch if fetch is not None else (FILE_CONTENTS, None) + + def fake_run(cfg, prompt): # type: ignore[no-untyped-def] + calls["runs"].append((cfg, prompt)) + return run if run is not None else _run(output=_model_output()) + + def fake_push(cfg, incident_id, build_number, patch, contents): # type: ignore[no-untyped-def] + calls["pushes"].append((cfg, incident_id, build_number, patch, contents)) + return push if push is not None else {"branch": "hermes-repair/7", "committed": True, "error": None} + + def fake_pull(cfg, incident_id, build_number, branch, patch, analysis): # type: ignore[no-untyped-def] + calls["pulls"].append((cfg, incident_id, build_number, branch, patch, analysis)) + return pull if pull is not None else {"pr_number": 5, "url": "https://scm.example/pulls/5", "error": None} + + monkeypatch.setattr(module.hermes_code_repair, "fetch_file", fake_fetch) + monkeypatch.setattr(module.hermes_agent_client, "run_triage", fake_run) + monkeypatch.setattr(module.hermes_code_repair, "push_branch", fake_push) + monkeypatch.setattr(module.hermes_code_repair, "open_pull_request", fake_pull) + return calls + + +def _propose(monkeypatch, **kwargs): # type: ignore[no-untyped-def] + storage = FakeStorage() + calls = _install(monkeypatch, **kwargs) + result = module.propose_code_fix(storage, INCIDENT_ID, JOB, 7, BUNDLE, _hermes_cfg(), _code_cfg()) + return storage, calls, result + + +def _event(storage: FakeStorage) -> dict: + rows = [row for row in storage.events if row["event_type"] == module.CODE_PROPOSAL_EVENT_TYPE] + assert len(rows) == 1 + return rows[0]["detail"] + + +def test_happy_path_opens_pull_request(monkeypatch) -> None: + storage, calls, result = _propose(monkeypatch) + assert result == { + "status": "pr_opened", + "branch": "hermes-repair/7", + "pr_number": 5, + "url": "https://scm.example/pulls/5", + "path": "src/discount.py", + "run_id": "run-1", + } + assert calls["fetches"][0] == (_code_cfg(), "src/discount.py") + cfg, incident_id, build_number, patch, contents = calls["pushes"][0] + assert (incident_id, build_number) == (INCIDENT_ID, 7) + assert patch.original == "return price * 0.5" + assert contents == "def discount(price):\n return price * 0.9\n" + _, _, _, branch, _, analysis = calls["pulls"][0] + assert branch == "hermes-repair/7" + assert analysis == "The multiplier regressed to 0.5." + detail = _event(storage) + assert detail == { + "incident_id": INCIDENT_ID, + "job": JOB, + "build_number": 7, + "run_id": "run-1", + "validated": True, + "reject_reason": None, + "branch": "hermes-repair/7", + "pr_number": 5, + "url": "https://scm.example/pulls/5", + } + serialized = json.dumps(detail) + assert "return price" not in serialized + assert "secret-token" not in serialized + + +def test_prompt_is_frozen_shape(monkeypatch) -> None: + _, calls, _ = _propose(monkeypatch) + cfg, prompt = calls["runs"][0] + assert cfg == _hermes_cfg() + assert prompt.startswith("Use $triage-titan-test-failures.\n") + assert f"MINIMAL source fix for incident {INCIDENT_ID}" in prompt + assert "The repository is bstein/hermes-code-demo branch master." in prompt + assert '"incident_id": ""' in prompt + assert "appearing exactly once" in prompt + assert "Change as few lines as possible; do not reformat; do not add dependencies." in prompt + assert "Ariadne validates and pushes the change — you do not execute anything." in prompt + assert "Set human_required to true if the fix is not a small localized source change." in prompt + assert json.dumps(BUNDLE, separators=(",", ":")) in prompt + assert "Current content of the candidate file src/discount.py:" in prompt + assert prompt.rstrip().endswith(FILE_CONTENTS.rstrip()) + + +def test_fetch_failure_requires_human(monkeypatch) -> None: + storage, calls, result = _propose(monkeypatch, fetch=(None, "file fetch http 404")) + assert result == {"status": "human_required", "reason": "candidate_fetch_failed: file fetch http 404"} + assert calls["runs"] == [] + assert calls["pushes"] == [] + detail = _event(storage) + assert detail["run_id"] is None + assert detail["validated"] is False + assert detail["reject_reason"] == "candidate_fetch_failed: file fetch http 404" + + +def test_unfinished_run_requires_human(monkeypatch) -> None: + storage, calls, result = _propose(monkeypatch, run=_run(status="timeout")) + assert result == {"status": "human_required", "reason": "hermes_run_timeout"} + assert calls["pushes"] == [] + assert _event(storage)["run_id"] == "run-1" + + +def test_completed_run_without_output_requires_human(monkeypatch) -> None: + _, calls, result = _propose(monkeypatch, run=_run(status="completed", output=None)) + assert result == {"status": "human_required", "reason": "hermes_run_completed"} + assert calls["pushes"] == [] + + +def test_invalid_patch_response_requires_human(monkeypatch) -> None: + storage, calls, result = _propose(monkeypatch, run=_run(output="no json here")) + assert result == {"status": "human_required", "reason": "patch_invalid: no_json_object_found"} + assert calls["pushes"] == [] + assert _event(storage)["validated"] is False + + +def test_patch_path_mismatch_requires_human(monkeypatch) -> None: + output = _model_output( + patch={ + "path": "src/other.py", + "original": "return price * 0.5", + "replacement": "return price * 0.9", + "rationale": "r", + } + ) + _, calls, result = _propose(monkeypatch, run=_run(output=output)) + assert result["reason"] == "patch_path_mismatch: got 'src/other.py' expected 'src/discount.py'" + assert calls["pushes"] == [] + + +def test_rejected_gate_requires_human(monkeypatch) -> None: + output = _model_output( + patch={ + "path": "src/discount.py", + "original": "not in the file", + "replacement": "still not", + "rationale": "r", + } + ) + _, calls, result = _propose(monkeypatch, run=_run(output=output)) + assert result == {"status": "human_required", "reason": "patch_rejected: original_missing"} + assert calls["pushes"] == [] + + +def test_push_failure_requires_human(monkeypatch) -> None: + storage, calls, result = _propose( + monkeypatch, push={"branch": "hermes-repair/7", "committed": False, "error": "commit http 500"} + ) + assert result == {"status": "human_required", "reason": "branch_push_failed: commit http 500"} + assert calls["pulls"] == [] + detail = _event(storage) + assert detail["validated"] is True + assert detail["branch"] is None + + +def test_pull_request_failure_requires_human(monkeypatch) -> None: + storage, _, result = _propose( + monkeypatch, pull={"pr_number": None, "url": None, "error": "pull request http 500"} + ) + assert result == {"status": "human_required", "reason": "pull_request_failed: pull request http 500"} + detail = _event(storage) + assert detail["validated"] is True + assert detail["branch"] == "hermes-repair/7" + assert detail["pr_number"] is None + + +def test_code_config_maps_settings() -> None: + config = SimpleNamespace( + 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", + ) + assert module.code_config(config) == _code_cfg() + + +def _orchestrator_settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] + values = { + "hermes_autotriage_enabled": True, + "hermes_autotriage_job_allowlist": [JOB], + "hermes_api_url": "http://hermes:8642", + "hermes_api_key": "key", + "hermes_run_timeout_seconds": 420.0, + "hermes_code_enabled": True, + "hermes_code_job": JOB, + "hermes_code_owner": "bstein", + "hermes_code_repo": "hermes-code-demo", + "hermes_code_base_branch": "master", + "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", + } + values.update(overrides) + return SimpleNamespace(**values) + + +def _prepare_orchestrator(monkeypatch, flow_result, **setting_overrides): # type: ignore[no-untyped-def] + storage = FakeStorage() + calls: list = [] + monkeypatch.setattr(autotriage, "settings", _orchestrator_settings(**setting_overrides)) + monkeypatch.setattr( + autotriage, + "_fetch_last_build", + lambda job: {"number": 7, "result": "FAILURE", "building": False, "url": "https://ci.example/7/"}, + ) + monkeypatch.setattr(autotriage.hermes_evidence, "collect_evidence", lambda i, j, b: dict(BUNDLE)) + + def fake_propose(storage_arg, incident_id, job, build_number, bundle, hermes_cfg, code_cfg): # type: ignore[no-untyped-def] + calls.append((incident_id, job, build_number, bundle, hermes_cfg, code_cfg)) + return flow_result + + monkeypatch.setattr(autotriage.hermes_code_flow, "propose_code_fix", fake_propose) + triage_calls: list = [] + + def fake_run_triage(cfg, prompt): # type: ignore[no-untyped-def] + triage_calls.append(prompt) + return _run(status="error") + + monkeypatch.setattr(autotriage.hermes_agent_client, "run_triage", fake_run_triage) + return storage, calls, triage_calls + + +def test_orchestrator_records_code_fix_proposed_on_pr(monkeypatch) -> None: + storage, calls, triage_calls = _prepare_orchestrator( + monkeypatch, + {"status": "pr_opened", "branch": "hermes-repair/7", "pr_number": 5, "url": "https://scm.example/pulls/5"}, + ) + summary = autotriage.run_hermes_autotriage(storage) + assert summary["jobs"][JOB] == { + "status": "human_required", + "incident_id": INCIDENT_ID, + "reason": "code_fix_proposed", + } + incidents = [row["detail"] for row in storage.events if row["event_type"] == autotriage.INCIDENT_EVENT_TYPE] + assert [detail["status"] for detail in incidents] == ["detected", "human_required"] + assert incidents[-1]["phase"] == { + "reason": "code_fix_proposed", + "branch": "hermes-repair/7", + "pr_number": 5, + "url": "https://scm.example/pulls/5", + } + assert [row for row in storage.events if row["event_type"] == autotriage.ACTION_EVENT_TYPE] == [] + assert triage_calls == [] + incident_id, job, build_number, bundle, hermes_cfg, code_cfg = calls[0] + assert (incident_id, job, build_number) == (INCIDENT_ID, JOB, 7) + assert bundle == BUNDLE + assert hermes_cfg == _hermes_cfg() + assert code_cfg == _code_cfg() + + +def test_orchestrator_records_flow_failure_reason(monkeypatch) -> None: + storage, _, _ = _prepare_orchestrator( + monkeypatch, {"status": "human_required", "reason": "patch_rejected: path_unsafe"} + ) + summary = autotriage.run_hermes_autotriage(storage) + assert summary["jobs"][JOB]["reason"] == "patch_rejected: path_unsafe" + incidents = [row["detail"] for row in storage.events if row["event_type"] == autotriage.INCIDENT_EVENT_TYPE] + assert incidents[-1]["status"] == "human_required" + assert incidents[-1]["phase"] == {"reason": "patch_rejected: path_unsafe"} + + +def test_orchestrator_defaults_missing_flow_reason(monkeypatch) -> None: + storage, _, _ = _prepare_orchestrator(monkeypatch, {"status": "human_required"}) + summary = autotriage.run_hermes_autotriage(storage) + assert summary["jobs"][JOB]["reason"] == "code_fix_not_proposed" + + +def test_orchestrator_skips_code_path_for_other_jobs(monkeypatch) -> None: + storage, calls, triage_calls = _prepare_orchestrator( + monkeypatch, {"status": "pr_opened"}, hermes_code_job="another-job" + ) + summary = autotriage.run_hermes_autotriage(storage) + assert summary["jobs"][JOB]["status"] == "human_required" + assert summary["jobs"][JOB]["reason"] == "hermes_run_error" + assert calls == [] + assert len(triage_calls) == 1 diff --git a/tests/test_hermes_code_patch.py b/tests/test_hermes_code_patch.py new file mode 100644 index 0000000..cbad030 --- /dev/null +++ b/tests/test_hermes_code_patch.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +import json + +import pytest + +from ariadne.services import hermes_code_patch as module + + +INCIDENT_ID = "hermes-code-demo/7" + +FILE_CONTENTS = "def discount(price):\n return price * 0.5\n" + + +def _patch_dict(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "path": "src/discount.py", + "original": "return price * 0.5", + "replacement": "return price * 0.9", + "rationale": "restore the intended 10% discount", + } + base.update(overrides) + return base + + +def _payload(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "incident_id": INCIDENT_ID, + "analysis": "The discount multiplier regressed from 0.9 to 0.5.", + "patch": _patch_dict(), + "human_required": False, + "reason": "small localized fix", + } + base.update(overrides) + return base + + +def _parse(payload: dict | None = None, raw: str | None = None, incident: str = INCIDENT_ID): # type: ignore[no-untyped-def] + text = raw if raw is not None else json.dumps(payload if payload is not None else _payload()) + return module.parse_patch_response(text, incident) + + +def _proposed(**overrides) -> module.ProposedPatch: # type: ignore[no-untyped-def] + return module.ProposedPatch(**_patch_dict(**overrides)) + + +def _cfg(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "allowed_path_prefixes": ["src/"], + "allowed_suffixes": [".py"], + "max_patch_bytes": 4000, + "max_changed_lines": 20, + } + base.update(overrides) + return base + + +def test_parse_valid_response() -> None: + outcome = _parse() + + assert outcome.valid is True + assert outcome.reject_reason is None + assert outcome.patch == module.ProposedPatch( + path="src/discount.py", + original="return price * 0.5", + replacement="return price * 0.9", + rationale="restore the intended 10% discount", + ) + + +def test_parse_accepts_fenced_and_prose_wrapped_json() -> None: + raw = "Here you go.\n```json\n" + json.dumps(_payload()) + "\n```\nDone." + outcome = _parse(raw=raw) + assert outcome.valid is True + assert outcome.patch is not None + + +def test_parse_accepts_braces_and_escapes_inside_strings() -> None: + payload = _payload(analysis='Uses a {brace}, a "quoted" word, and a \\ escape.') + raw = "Prose with a stray } brace... " + json.dumps(payload) + ' Trailing {"not": "parsed"}' + outcome = _parse(raw=raw) + assert outcome.valid is True + assert outcome.patch is not None + + +def test_parse_rejects_missing_json() -> None: + assert _parse(raw="no json at all").reject_reason == "no_json_object_found" + assert _parse(raw="").reject_reason == "no_json_object_found" + + +def test_parse_rejects_invalid_json() -> None: + outcome = _parse(raw="{'incident_id': 'x'}") + assert outcome.valid is False + assert outcome.reject_reason.startswith("invalid_json:") + + +def test_parse_rejects_missing_keys() -> None: + payload = _payload() + payload.pop("analysis") + payload.pop("reason") + assert _parse(payload).reject_reason == "missing_keys: analysis, reason" + + +def test_parse_rejects_unexpected_keys() -> None: + assert _parse(_payload(extra=1)).reject_reason == "unexpected_keys: extra" + + +@pytest.mark.parametrize("field", ["incident_id", "analysis", "reason"]) +def test_parse_rejects_non_string_scalars(field) -> None: # type: ignore[no-untyped-def] + outcome = _parse(_payload(**{field: 5})) + assert outcome.reject_reason == f"field_type_invalid: {field} must be a string" + + +def test_parse_rejects_non_boolean_human_required() -> None: + outcome = _parse(_payload(human_required="no")) + assert outcome.reject_reason == "field_type_invalid: human_required must be a boolean" + + +def test_parse_rejects_non_object_patch() -> None: + assert _parse(_payload(patch=[1])).reject_reason == "patch_invalid: must be an object or null" + assert _parse(_payload(patch="diff")).reject_reason == "patch_invalid: must be an object or null" + + +def test_parse_rejects_patch_with_wrong_keys() -> None: + missing = _patch_dict() + missing.pop("rationale") + expected = "patch_invalid: must have exactly path, original, replacement, rationale" + assert _parse(_payload(patch=missing)).reject_reason == expected + assert _parse(_payload(patch=_patch_dict(extra="x"))).reject_reason == expected + + +def test_parse_rejects_non_string_patch_fields() -> None: + outcome = _parse(_payload(patch=_patch_dict(original=7))) + assert outcome.reject_reason == "patch_invalid: fields must be strings" + + +@pytest.mark.parametrize("field", ["path", "original", "replacement"]) +def test_parse_rejects_empty_patch_fields(field) -> None: # type: ignore[no-untyped-def] + outcome = _parse(_payload(patch=_patch_dict(**{field: ""}))) + assert outcome.reject_reason == f"patch_field_empty: {field}" + + +def test_parse_rejects_incident_mismatch() -> None: + outcome = _parse(_payload(incident_id="other/1")) + assert outcome.reject_reason == f"incident_id_mismatch: got 'other/1' expected {INCIDENT_ID!r}" + + +def test_parse_rejects_human_required_true() -> None: + assert _parse(_payload(human_required=True)).reject_reason == "human_required" + + +def test_parse_rejects_null_patch() -> None: + assert _parse(_payload(patch=None)).reject_reason == "patch_missing" + + +def test_parse_analysis_extracts_string() -> None: + assert module.parse_analysis(json.dumps(_payload())).startswith("The discount multiplier") + assert module.parse_analysis("no json") == "" + assert module.parse_analysis("{'bad': json}") == "" + assert module.parse_analysis('{"analysis": 5}') == "" + assert module.parse_analysis("") == "" + + +def test_validate_patch_accepts_valid_patch() -> None: + assert module.validate_patch(_proposed(), _cfg(), FILE_CONTENTS) == (True, "valid") + + +@pytest.mark.parametrize( + "path", + [ + "src/../secrets.py", + "/src/discount.py", + "src\\discount.py", + "src/disc\x00ount.py", + "", + ], +) +def test_validate_patch_rejects_unsafe_paths(path) -> None: # type: ignore[no-untyped-def] + ok, reason = module.validate_patch(_proposed(path=path), _cfg(), FILE_CONTENTS) + assert (ok, reason) == (False, "path_unsafe") + + +def test_validate_patch_rejects_disallowed_prefix() -> None: + ok, reason = module.validate_patch(_proposed(path="lib/discount.py"), _cfg(), FILE_CONTENTS) + assert (ok, reason) == (False, "path_prefix_not_allowed") + + +def test_validate_patch_rejects_disallowed_suffix() -> None: + ok, reason = module.validate_patch(_proposed(path="src/discount.sh"), _cfg(), FILE_CONTENTS) + assert (ok, reason) == (False, "path_suffix_not_allowed") + + +def test_validate_patch_rejects_empty_allowlists() -> None: + cfg = _cfg(allowed_path_prefixes=[], allowed_suffixes=[]) + ok, reason = module.validate_patch(_proposed(), cfg, FILE_CONTENTS) + assert (ok, reason) == (False, "path_prefix_not_allowed") + + +def test_validate_patch_treats_non_numeric_limits_as_zero() -> None: + cfg = _cfg(max_patch_bytes="lots", max_changed_lines=None) + ok, reason = module.validate_patch(_proposed(), cfg, FILE_CONTENTS) + assert (ok, reason) == (False, "patch_too_large") + + +def test_validate_patch_rejects_oversized_patch() -> None: + ok, reason = module.validate_patch(_proposed(), _cfg(max_patch_bytes=10), FILE_CONTENTS) + assert (ok, reason) == (False, "patch_too_large") + + +def test_validate_patch_rejects_too_many_changed_lines() -> None: + patch = _proposed(replacement="return (\n price\n * 0.9\n)") + ok, reason = module.validate_patch(patch, _cfg(max_changed_lines=2), FILE_CONTENTS) + assert (ok, reason) == (False, "too_many_changed_lines") + + +def test_validate_patch_counts_lines_from_larger_side() -> None: + patch = _proposed(original="return price * 0.5\n", replacement="return price * 0.9") + contents = FILE_CONTENTS + assert module.validate_patch(patch, _cfg(max_changed_lines=2), contents) == (True, "valid") + ok, reason = module.validate_patch(patch, _cfg(max_changed_lines=1), contents) + assert (ok, reason) == (False, "too_many_changed_lines") + + +def test_validate_patch_rejects_missing_original() -> None: + ok, reason = module.validate_patch(_proposed(original="return price * 0.75"), _cfg(), FILE_CONTENTS) + assert (ok, reason) == (False, "original_missing") + + +def test_validate_patch_rejects_ambiguous_original() -> None: + contents = FILE_CONTENTS + "\ndef other(price):\n return price * 0.5\n" + ok, reason = module.validate_patch(_proposed(), _cfg(), contents) + assert (ok, reason) == (False, "original_ambiguous") + + +def test_validate_patch_rejects_identical_replacement() -> None: + patch = _proposed(replacement="return price * 0.5") + ok, reason = module.validate_patch(patch, _cfg(), FILE_CONTENTS) + assert (ok, reason) == (False, "replacement_identical") + + +def test_apply_patch_replaces_single_occurrence() -> None: + patched = module.apply_patch(FILE_CONTENTS, _proposed()) + assert patched == "def discount(price):\n return price * 0.9\n" + + +def test_apply_patch_raises_on_zero_occurrences() -> None: + with pytest.raises(ValueError, match="exactly once, found 0"): + module.apply_patch("nothing here", _proposed()) + + +def test_apply_patch_raises_on_multiple_occurrences() -> None: + with pytest.raises(ValueError, match="exactly once, found 2"): + module.apply_patch(FILE_CONTENTS * 2, _proposed()) diff --git a/tests/test_hermes_code_repair.py b/tests/test_hermes_code_repair.py new file mode 100644 index 0000000..8c74bc4 --- /dev/null +++ b/tests/test_hermes_code_repair.py @@ -0,0 +1,258 @@ +from __future__ import annotations + +import base64 +import json + +from ariadne.services import hermes_code_repair as module +from ariadne.services.hermes_code_patch import ProposedPatch + + +INCIDENT_ID = "hermes-code-demo/7" +BRANCH = "hermes-repair/7" + + +class FakeResponse: + def __init__(self, status_code: int, payload=None, text: str = "") -> None: # type: ignore[no-untyped-def] + self.status_code = status_code + self._payload = payload + self.text = text + + def json(self): # type: ignore[no-untyped-def] + if self._payload is None: + raise ValueError("no json body") + return self._payload + + +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 put(self, url, **kwargs): # type: ignore[no-untyped-def] + return self._next("PUT", 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] + base = { + "gitea_base_url": "https://scm.example", + "gitea_token": "secret-token", + "owner": "bstein", + "repo": "hermes-code-demo", + "base_branch": "master", + "timeout_seconds": 7.5, + } + base.update(overrides) + return base + + +def _patch(**overrides) -> ProposedPatch: # type: ignore[no-untyped-def] + values = { + "path": "src/discount.py", + "original": "return price * 0.5", + "replacement": "return price * 0.9", + "rationale": "restore the intended discount", + } + values.update(overrides) + return ProposedPatch(**values) + + +def _push(monkeypatch, responses): # type: ignore[no-untyped-def] + calls = _install_http(monkeypatch, responses) + result = module.push_branch(_cfg(), INCIDENT_ID, 7, _patch(), "patched contents\n") + return calls, result + + +def test_fetch_file_success(monkeypatch) -> None: + calls = _install_http(monkeypatch, [FakeResponse(200, text="file body")]) + contents, error = module.fetch_file(_cfg(), "src/discount.py") + assert (contents, error) == ("file body", None) + method, url, kwargs = calls["requests"][0] + assert method == "GET" + assert url == "https://scm.example/api/v1/repos/bstein/hermes-code-demo/raw/src/discount.py" + assert kwargs["params"] == {"ref": "master"} + assert kwargs["headers"] == {"Authorization": "token secret-token"} + assert calls["kwargs"] == {"timeout": 7.5} + + +def test_fetch_file_http_error(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(404)]) + assert module.fetch_file(_cfg(), "src/discount.py") == (None, "file fetch http 404") + + +def test_fetch_file_request_exception(monkeypatch) -> None: + _install_http(monkeypatch, [RuntimeError("connect refused")]) + contents, error = module.fetch_file(_cfg(), "src/discount.py") + assert contents is None + assert error == "file fetch failed: connect refused" + + +def test_fetch_file_without_base_url(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + assert module.fetch_file(_cfg(gitea_base_url=""), "x") == (None, "gitea base url is empty") + assert calls["requests"] == [] + + +def test_fetch_file_uses_default_timeout(monkeypatch) -> None: + calls = _install_http(monkeypatch, [FakeResponse(200, text="ok")]) + module.fetch_file(_cfg(timeout_seconds="bad"), "src/discount.py") + assert calls["kwargs"] == {"timeout": 15.0} + + +def test_push_branch_success_with_new_branch_payload(monkeypatch) -> None: + calls, result = _push(monkeypatch, [FakeResponse(200, {"sha": "abc123"}), FakeResponse(201, {})]) + assert result == {"branch": BRANCH, "committed": True, "error": None} + get_method, get_url, get_kwargs = calls["requests"][0] + assert (get_method, get_kwargs["params"]) == ("GET", {"ref": "master"}) + assert get_url == "https://scm.example/api/v1/repos/bstein/hermes-code-demo/contents/src/discount.py" + put_method, put_url, put_kwargs = calls["requests"][1] + assert (put_method, put_url) == ("PUT", get_url) + body = put_kwargs["json"] + assert body["branch"] == "master" + assert body["new_branch"] == BRANCH + assert body["sha"] == "abc123" + assert body["message"] == f"fix(hermes): restore the intended discount (incident {INCIDENT_ID})" + assert base64.b64decode(body["content"]).decode() == "patched contents\n" + assert body["author"] == {"name": "Hermes Agent", "email": "hermes@bstein.dev"} + assert body["committer"] == {"name": "Hermes Agent", "email": "hermes@bstein.dev"} + + +def test_push_branch_falls_back_to_branch_only_payload(monkeypatch) -> None: + calls, result = _push( + monkeypatch, + [FakeResponse(200, {"sha": "abc123"}), FakeResponse(422, {}), FakeResponse(201, {})], + ) + assert result == {"branch": BRANCH, "committed": True, "error": None} + fallback_body = calls["requests"][2][2]["json"] + assert fallback_body["branch"] == BRANCH + assert "new_branch" not in fallback_body + + +def test_push_branch_reports_both_failed_payload_shapes(monkeypatch) -> None: + _, result = _push( + monkeypatch, + [FakeResponse(200, {"sha": "abc123"}), FakeResponse(404, {}), FakeResponse(422, {})], + ) + assert result["committed"] is False + assert result["error"] == "commit http 404 then fallback http 422" + + +def test_push_branch_non_retryable_commit_error(monkeypatch) -> None: + calls, result = _push(monkeypatch, [FakeResponse(200, {"sha": "abc123"}), FakeResponse(500, {})]) + assert result["error"] == "commit http 500" + assert len(calls["requests"]) == 2 + + +def test_push_branch_sha_read_error(monkeypatch) -> None: + _, result = _push(monkeypatch, [FakeResponse(404)]) + assert result == {"branch": BRANCH, "committed": False, "error": "file sha http 404"} + + +def test_push_branch_sha_missing(monkeypatch) -> None: + _, result = _push(monkeypatch, [FakeResponse(200, {})]) + assert result["error"] == "file sha missing from contents response" + + +def test_push_branch_request_exception(monkeypatch) -> None: + _, result = _push(monkeypatch, [RuntimeError("boom")]) + assert result == {"branch": BRANCH, "committed": False, "error": "branch push failed: boom"} + + +def test_push_branch_refuses_protected_branch_names(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + for name in ("master", "main", " "): + monkeypatch.setattr(module, "_branch_name", lambda build_number, name=name: name) + result = module.push_branch(_cfg(), INCIDENT_ID, 7, _patch(), "contents") + assert result["committed"] is False + assert result["error"] == f"refusing branch {name!r}" + assert calls["requests"] == [] + + +def test_push_branch_without_base_url(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + result = module.push_branch(_cfg(gitea_base_url=""), INCIDENT_ID, 7, _patch(), "contents") + assert result == {"branch": BRANCH, "committed": False, "error": "gitea base url is empty"} + assert calls["requests"] == [] + + +def test_open_pull_request_created(monkeypatch) -> None: + calls = _install_http( + monkeypatch, + [FakeResponse(201, {"number": 5, "html_url": "https://scm.example/pulls/5"})], + ) + result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "root cause analysis") + assert result == {"pr_number": 5, "url": "https://scm.example/pulls/5", "error": None} + method, url, kwargs = calls["requests"][0] + assert (method, url) == ("POST", "https://scm.example/api/v1/repos/bstein/hermes-code-demo/pulls") + payload = kwargs["json"] + assert payload["head"] == BRANCH + assert payload["base"] == "master" + assert payload["title"] == f"fix(hermes): repair {INCIDENT_ID}" + body = payload["body"] + assert INCIDENT_ID in body + assert "root cause analysis" in body + assert "restore the intended discount" in body + assert "`src/discount.py`" in body + assert "Proposed by Hermes; validated and pushed by Ariadne; requires human review — no automatic merge." in body + assert "secret-token" not in json.dumps(payload) + + +def test_open_pull_request_conflict_returns_existing(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(409, {"number": 9, "html_url": "https://scm.example/pulls/9"})]) + result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis") + assert result == {"pr_number": 9, "url": "https://scm.example/pulls/9", "error": None} + + +def test_open_pull_request_conflict_without_payload(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(409)]) + result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis") + assert result == {"pr_number": None, "url": None, "error": "pull request already exists"} + + +def test_open_pull_request_http_error(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(500)]) + result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis") + assert result == {"pr_number": None, "url": None, "error": "pull request http 500"} + + +def test_open_pull_request_created_with_bad_payload(monkeypatch) -> None: + _install_http(monkeypatch, [FakeResponse(201, {"number": True})]) + result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis") + assert result == {"pr_number": None, "url": None, "error": None} + + +def test_open_pull_request_request_exception(monkeypatch) -> None: + _install_http(monkeypatch, [RuntimeError("down")]) + result = module.open_pull_request(_cfg(), INCIDENT_ID, 7, BRANCH, _patch(), "analysis") + assert result == {"pr_number": None, "url": None, "error": "pull request failed: down"} + + +def test_open_pull_request_without_base_url(monkeypatch) -> None: + calls = _install_http(monkeypatch, []) + result = module.open_pull_request(_cfg(gitea_base_url=" "), INCIDENT_ID, 7, BRANCH, _patch(), "a") + assert result == {"pr_number": None, "url": None, "error": "gitea base url is empty"} + assert calls["requests"] == [] diff --git a/tests/test_settings.py b/tests/test_settings.py index d95a87b..7731d10 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -93,3 +93,33 @@ def test_from_env_includes_game_stream_settings(monkeypatch) -> None: assert cfg.wolf_gatekeeper_url == "http://wolf-gatekeeper.game-stream.svc.cluster.local:8087" assert cfg.game_stream_firewall_unlock_ttl_sec == 28800 assert cfg.game_stream_moonlight_host == "moonlight.bstein.dev" + + +def test_from_env_includes_hermes_code_settings(monkeypatch) -> None: + monkeypatch.setenv("ARIADNE_HERMES_CODE_ENABLED", "true") + monkeypatch.setenv("ARIADNE_HERMES_CODE_JOB", "hermes-code-demo") + monkeypatch.setenv("ARIADNE_HERMES_CODE_OWNER", "bstein") + monkeypatch.setenv("ARIADNE_HERMES_CODE_REPO", "hermes-code-demo") + monkeypatch.setenv("ARIADNE_HERMES_CODE_BASE_BRANCH", "master") + monkeypatch.setenv("ARIADNE_HERMES_CODE_CANDIDATE_PATH", "src/discount.py") + monkeypatch.setenv("ARIADNE_HERMES_CODE_ALLOWED_PREFIXES", "src/, app/") + monkeypatch.setenv("ARIADNE_HERMES_CODE_ALLOWED_SUFFIXES", ".py") + monkeypatch.setenv("ARIADNE_HERMES_CODE_MAX_PATCH_BYTES", "2000") + monkeypatch.setenv("ARIADNE_HERMES_CODE_MAX_CHANGED_LINES", "10") + monkeypatch.setenv("ARIADNE_HERMES_GITEA_BASE_URL", "https://scm.bstein.dev/") + monkeypatch.setenv("ARIADNE_HERMES_GITEA_TOKEN", "token") + + cfg = Settings.from_env() + + assert cfg.hermes_code_enabled is True + assert cfg.hermes_code_job == "hermes-code-demo" + assert cfg.hermes_code_owner == "bstein" + assert cfg.hermes_code_repo == "hermes-code-demo" + assert cfg.hermes_code_base_branch == "master" + assert cfg.hermes_code_candidate_path == "src/discount.py" + assert cfg.hermes_code_allowed_prefixes == ["src/", "app/"] + assert cfg.hermes_code_allowed_suffixes == [".py"] + assert cfg.hermes_code_max_patch_bytes == 2000 + assert cfg.hermes_code_max_changed_lines == 10 + assert cfg.hermes_gitea_base_url == "https://scm.bstein.dev" + assert cfg.hermes_gitea_token == "token"