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/<build> 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 <noreply@anthropic.com>

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."""
 

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": "<must equal __INCIDENT_ID__>", "analysis": "<string>", "patch": {"path": "<repository-relative file path>", "original": "<exact snippet from the file shown>", "replacement": "<replacement snippet>", "rationale": "<string>"} or null, "human_required": <bool>, "reason": "<string>"}
+`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)
+    )

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

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

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,

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"),

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",

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": "<must equal ' + 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

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())

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"] == []

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"