ariadne/ariadne/services/hermes_code_flow.py

"""Run the bounded code-repair proposal flow for one failed build.

The flow is per-job: `hermes_code_repos` maps a Jenkins job to the repository
its failures live in (see `hermes_code_repos` for the settings contract), the
console evidence says which files that build implicates (see
`hermes_code_candidates`), and this module fetches, prompts, validates, and
publishes. The original single-repo demo settings still work unchanged - a
job matching `hermes_code_job` keeps patching its one fixed
`hermes_code_candidate_path` in `hermes_code_owner/hermes_code_repo`.

Two entry points share that flow. `propose_code_fix` is the demo job's
dedicated path, which replaces diagnosis outright. `propose_for_incident` is
the additive step every other mapped job takes: it runs after triage has
already diagnosed the failure and decided a human is needed, so a real service
failure can produce both an issue and a pull request.
"""

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_candidates,
    hermes_code_patch,
    hermes_code_repair,
    hermes_code_repos,
)
from .hermes_autotriage_metrics import HERMES_TRIAGE_ACTION_TOTAL, PROPOSE_CODE_FIX_ACTION


logger = get_logger(__name__)

CODE_PROPOSAL_EVENT_TYPE = "hermes_autotriage_code_proposal"

_RUN_COMPLETED = "completed"
_EXISTING_PROPOSAL_REASON = "existing_proposal_open"
_NO_REPO_MAPPING_REASON = "no_repo_mapping"
_NO_CANDIDATE_FILES_REASON = "no_candidate_files"

_PATCH_PROMPT_TEMPLATE = """Use $triage-titan-test-failures.
You are proposing a MINIMAL source fix for incident __INCIDENT_ID__.
The repository is __OWNER__/__REPO__ branch __BASE_BRANCH__.
Return ONLY a single JSON object with exactly these keys and no others:
{"incident_id": "<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>"}
`patch.path` MUST be exactly one of these candidate paths, copied character for character:
__CANDIDATE_LIST__
`original` must be an exact substring of the content shown below for THAT file, appearing exactly once.
Change as few lines as possible; do not reformat; do not add dependencies.
Ariadne validates and pushes the change — you do not execute anything.
Set human_required to true if the fix is not a small localized source change.

Failing test evidence bundle:
__BUNDLE__

__FILE_SECTIONS__"""

_FILE_SECTION_TEMPLATE = "Current content of the candidate file {path}:\n{contents}"


@dataclass(frozen=True)
class _Incident:
    incident_id: str
    job: str
    build_number: int


@dataclass(frozen=True)
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 carrying the resolved repository,
    the candidate paths offered, and the chosen path - never patch bodies or
    tokens.
    """

    incident = _Incident(incident_id=incident_id, job=job, build_number=build_number)
    result, event = _propose(incident, bundle, hermes_cfg, code_cfg)
    storage.record_event(
        CODE_PROPOSAL_EVENT_TYPE,
        {"incident_id": incident_id, "job": job, "build_number": build_number, **event},
    )
    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 settings-level cfg dict built by `hermes_code_repos`, which
    also documents the multi-repository fields and their defaults;
    `resolve_repo_config` narrows it to one job before any Gitea call.
    """

    return hermes_code_repos.build_config(config)


def resolve_repo_config(job: str, settings_cfg: dict) -> dict[str, Any] | None:
    """Narrow the settings-level cfg to the repository one job publishes to.

    Inputs: the Jenkins job name and the cfg from `code_config`. Outputs: the
    merged per-job cfg consumed by the patch validator and the Gitea client,
    or None when the job maps to no repository - the caller must then make no
    HTTP call. See `hermes_code_repos.resolve` for the merge rules.
    """

    return hermes_code_repos.resolve(job, settings_cfg)


def propose_for_incident(
    storage: Any, base: dict, bundle: dict, hermes_cfg: dict, config: Any
) -> dict[str, Any]:
    """Attempt an extra code-fix proposal for an escalating incident.

    Inputs: the incident event storage; the incident identity dict
    (incident_id, job, build_number); the evidence bundle already collected
    for that incident; the Hermes agent run config; and a settings-like object
    exposing the hermes_code_* fields. Outputs: the phase detail to merge into
    the incident's human_required phase - {"code_proposal": {branch, pr_number,
    url}} when a pull request was opened, {"code_proposal": {"reason": ...}}
    when the proposal declined, and {} when nothing was attempted.

    Attempts nothing unless the code path is enabled, the job is a real
    service job rather than the legacy demo job (which keeps its own dedicated
    short-circuit), and the job maps to a repository - an unmapped job costs
    no HTTP call and no model tokens. Additive on top of triage and never
    raises: the incident's outcome is decided before this runs and must
    survive any proposal failure.
    """

    try:
        settings_cfg = _eligible_config(str(base.get("job") or ""), config)
        if settings_cfg is None:
            return {}
        return _incident_proposal(storage, base, bundle, hermes_cfg, settings_cfg)
    except Exception as exc:
        logger.info(
            "hermes incident code proposal failed",
            extra={
                "event": "hermes_code_flow",
                "status": "error",
                "incident_id": str(base.get("incident_id") or ""),
                "detail": str(exc),
            },
        )
        return {}


def _eligible_config(job: str, config: Any) -> dict[str, Any] | None:
    """Return the code cfg when this job may receive an extra proposal."""

    if not job or not getattr(config, "hermes_code_enabled", False):
        return None
    if job == str(getattr(config, "hermes_code_job", "") or ""):
        return None
    settings_cfg = code_config(config)
    return settings_cfg if resolve_repo_config(job, settings_cfg) is not None else None


def _incident_proposal(
    storage: Any, base: dict, bundle: dict, hermes_cfg: dict, settings_cfg: dict
) -> dict[str, Any]:
    """Run the proposal for an eligible incident and count its outcome."""

    _count("requested")
    result = propose_code_fix(
        storage,
        str(base.get("incident_id") or ""),
        str(base.get("job") or ""),
        _int_value(base.get("build_number")),
        bundle,
        hermes_cfg,
        settings_cfg,
    )
    if result.get("status") != "pr_opened":
        _count("rejected")
        return {"code_proposal": {"reason": str(result.get("reason") or "code_fix_not_proposed")}}
    _count("success")
    detail = {
        "branch": result.get("branch"),
        "pr_number": result.get("pr_number"),
        "url": result.get("url"),
    }
    return {"code_proposal": detail}


def _count(result: str) -> None:
    """Count one additive code-proposal attempt under its bounded label."""

    HERMES_TRIAGE_ACTION_TOTAL.labels(action=PROPOSE_CODE_FIX_ACTION, result=result).inc()


def _int_value(value: Any) -> int:
    """Coerce a value to int, defaulting to zero."""

    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def _propose(
    incident: _Incident, bundle: dict, hermes_cfg: dict, code_cfg: dict
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Run resolution, fetch, diagnosis, validation, and publication.

    `context` is filled in as the flow learns the repository, the candidate
    paths it offered the model, and the path the model chose, so the recorded
    event describes the proposal however it ended.
    """

    context: dict[str, Any] = {"repo": None, "candidates": [], "chosen_path": None}
    result, event = _run_proposal(incident, bundle, hermes_cfg, code_cfg, context)
    return result, {**event, **context}


def _run_proposal(
    incident: _Incident, bundle: dict, hermes_cfg: dict, code_cfg: dict, context: dict
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Resolve the repository and gather candidate files before diagnosis."""

    cfg = resolve_repo_config(incident.job, code_cfg)
    if cfg is None:
        return _human_required(_NO_REPO_MAPPING_REASON, None, validated=False)
    context["repo"] = f"{cfg.get('owner') or ''}/{cfg.get('repo') or ''}"
    duplicate = _duplicate_proposal(cfg, incident.incident_id)
    if duplicate is not None:
        return duplicate
    fetched, failure = _candidate_contents(cfg, bundle)
    context["candidates"] = list(fetched)
    if failure is not None:
        return _human_required(failure, None, validated=False)
    return _diagnose(incident, bundle, hermes_cfg, cfg, fetched, context)


def _diagnose(  # noqa: PLR0913 - diagnosis needs both configs plus the fetched context
    incident: _Incident,
    bundle: dict,
    hermes_cfg: dict,
    cfg: dict,
    fetched: dict[str, str],
    context: dict,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Ask Hermes for a patch and gate it against the file it names."""

    prompt = _build_patch_prompt(incident.incident_id, bundle, cfg, fetched)
    run = hermes_agent_client.run_triage(hermes_cfg, prompt)
    if run.status != _RUN_COMPLETED or not run.output:
        return _human_required(f"hermes_run_{run.status}", run.run_id, validated=False)
    patch, reject_reason = _validated_patch(run.output, incident.incident_id, cfg, fetched)
    if patch is None:
        return _human_required(reject_reason, run.run_id, validated=False)
    context["chosen_path"] = patch.path
    proposal = _Proposal(
        patch=patch,
        analysis=hermes_code_patch.parse_analysis(run.output),
        patched_contents=hermes_code_patch.apply_patch(fetched[patch.path], patch),
        run_id=run.run_id,
    )
    return _publish(cfg, incident, proposal)


def _candidate_contents(cfg: dict, bundle: dict) -> tuple[dict[str, str], str | None]:
    """Pick the candidate paths for this job and fetch their contents."""

    candidates = _candidate_paths(cfg, bundle)
    if not candidates:
        return {}, _NO_CANDIDATE_FILES_REASON
    fetched, error = _fetch_within_budget(cfg, candidates)
    if not fetched:
        return {}, f"candidate_fetch_failed: {error}"
    return fetched, None


def _candidate_paths(cfg: dict, bundle: dict) -> list[str]:
    """Return the ranked candidate paths, honouring the legacy fixed path."""

    legacy_path = str(cfg.get("candidate_path") or "")
    if legacy_path:
        return [legacy_path]
    return hermes_code_candidates.extract_candidate_paths(bundle, cfg)


def _fetch_within_budget(cfg: dict, candidates: list[str]) -> tuple[dict[str, str], str | None]:
    """Fetch candidates in rank order while the context budget allows.

    A file that would exceed the remaining budget is skipped whole and never
    truncated: a patch anchor has to match the file exactly, so a partial
    file can only produce an unapplicable patch. Returns the fetched contents
    keyed by path in rank order plus the first fetch error seen.
    """

    budget = hermes_code_repos.positive_int(
        cfg.get("max_context_chars"), hermes_code_repos.DEFAULT_MAX_CONTEXT_CHARS
    )
    fetched: dict[str, str] = {}
    first_error: str | None = None
    used = 0
    queue = list(candidates)
    queued = set(queue)
    for path in queue:
        contents, error = hermes_code_repair.fetch_file(cfg, path)
        if contents is None:
            first_error = first_error or str(error or "unknown error")
            continue
        if used + len(contents) > budget:
            logger.info(
                "hermes code candidate skipped for context budget",
                extra={"event": "hermes_code_flow", "status": "skipped", "detail": path},
            )
            continue
        fetched[path] = contents
        used += len(contents)
        hermes_code_candidates.queue_imported_sources(cfg, path, contents, queue, queued)
    return fetched, first_error


def _duplicate_proposal(
    code_cfg: dict, incident_id: str
) -> tuple[dict[str, Any], dict[str, Any]] | None:
    """Suppress this proposal when an earlier repair pull request is still open.

    Runs before candidate selection and before Hermes, so a duplicate costs no
    model tokens. Returns the human_required result/event pair when an open
    proposal exists, otherwise None so the flow continues — including when the
    lookup itself failed, which fails open rather than dropping real work.
    """

    existing = hermes_code_repair.find_open_proposal(code_cfg)
    error = existing.get("error")
    if error:
        logger.info(
            "hermes open proposal lookup failed",
            extra={
                "event": "hermes_code_flow",
                "status": "error",
                "incident_id": incident_id,
                "detail": error,
            },
        )
        return None
    if not existing.get("found"):
        return None
    identity = {
        "pr_number": existing.get("pr_number"),
        "url": existing.get("url"),
        "branch": existing.get("branch"),
    }
    result = {"status": "human_required", "reason": _EXISTING_PROPOSAL_REASON, **identity}
    event = {
        "run_id": None,
        "validated": False,
        "reject_reason": _EXISTING_PROPOSAL_REASON,
        **identity,
    }
    return result, event


def _validated_patch(
    raw_output: str, incident_id: str, code_cfg: dict, fetched: dict[str, str]
) -> tuple[hermes_code_patch.ProposedPatch | None, str]:
    """Parse and gate the patch response against the file the model chose."""

    outcome = hermes_code_patch.parse_patch_response(raw_output, incident_id)
    if not outcome.valid or outcome.patch is None:
        return None, f"patch_invalid: {outcome.reject_reason}"
    path = outcome.patch.path
    legacy_path = str(code_cfg.get("candidate_path") or "")
    if legacy_path and path != legacy_path:
        return None, f"patch_path_mismatch: got {path!r} expected {legacy_path!r}"
    if path not in fetched:
        return None, f"patch_path_not_offered: got {path!r} offered {sorted(fetched)!r}"
    ok, gate = hermes_code_patch.validate_patch(outcome.patch, code_cfg, fetched[path])
    if not ok:
        return None, f"patch_rejected: {gate}"
    return outcome.patch, ""


def _publish(
    code_cfg: dict, incident: _Incident, proposal: _Proposal
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Push the repair branch and open the human-review pull request."""

    push = hermes_code_repair.push_branch(
        code_cfg,
        incident.incident_id,
        incident.build_number,
        proposal.patch,
        proposal.patched_contents,
    )
    branch = str(push.get("branch") or "")
    if not push.get("committed"):
        return _human_required(f"branch_push_failed: {push.get('error')}", proposal.run_id, validated=True)
    pull = hermes_code_repair.open_pull_request(
        code_cfg,
        incident.incident_id,
        incident.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, fetched: dict[str, str]
) -> str:
    """Render the frozen patch prompt with incident, bundle, and file context."""

    compact = json.dumps(bundle, separators=(",", ":"), ensure_ascii=True)
    listing = "\n".join(f"- {path}" for path in fetched)
    sections = "\n\n".join(
        _FILE_SECTION_TEMPLATE.format(path=path, contents=contents)
        for path, contents in fetched.items()
    )
    return (
        _PATCH_PROMPT_TEMPLATE.replace("__INCIDENT_ID__", incident_id)
        .replace("__OWNER__", str(code_cfg.get("owner") or ""))
        .replace("__REPO__", str(code_cfg.get("repo") or ""))
        .replace("__BASE_BRANCH__", str(code_cfg.get("base_branch") or ""))
        .replace("__CANDIDATE_LIST__", listing)
        .replace("__BUNDLE__", compact)
        .replace("__FILE_SECTIONS__", sections)
    )