54de192c8b
All checks were successful
Tests / Declarative: Post Actions passed: 1237
The three defect categories were implicit in code: an operator could neither see which were active nor switch one off. ARIADNE_HERMES_FIX_CATEGORIES now names them, an unrecognised entry is ignored rather than trusted, and an empty setting means all three so the default stays obvious. Kept separate from ARIADNE_HERMES_ALLOWED_ACTIONS on purpose. That allowlist gates what Ariadne executes on its own authority, with nothing between the decision and the change. Everything here becomes a pull request a person reads. Sharing one list would let a patch clear the same gate as an autonomous mutation, and that distinction is what makes the autonomous half defensible. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
493 lines
19 KiB
Python
493 lines
19 KiB
Python
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",
|
|
"max_open_proposals": 0,
|
|
"fix_categories": [],
|
|
"base_branch": "master",
|
|
"timeout_seconds": 15.0,
|
|
"legacy_job": JOB,
|
|
"repos": {},
|
|
"job_prefixes": {},
|
|
"job_suffixes": {},
|
|
"job_base_branches": {},
|
|
"max_candidates": 3,
|
|
"max_context_chars": 60000,
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
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 _no_existing(**overrides) -> dict: # type: ignore[no-untyped-def]
|
|
base = {"found": False, "pr_number": None, "url": None, "branch": None, "error": None}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
def _install(monkeypatch, *, fetch=None, run=None, push=None, pull=None, existing=None) -> dict: # type: ignore[no-untyped-def]
|
|
calls: dict = {"fetches": [], "runs": [], "pushes": [], "pulls": [], "lookups": []}
|
|
|
|
def fake_find(cfg): # type: ignore[no-untyped-def]
|
|
calls["lookups"].append(cfg)
|
|
return existing if existing is not None else _no_existing()
|
|
|
|
monkeypatch.setattr(module.hermes_code_repair, "find_open_proposal", fake_find)
|
|
|
|
def fake_fetch(cfg, path): # type: ignore[no-untyped-def]
|
|
calls["fetches"].append((cfg, path))
|
|
if isinstance(fetch, dict):
|
|
return fetch.get(path, (None, "file fetch http 404"))
|
|
return fetch if fetch is not None else (FILE_CONTENTS, None)
|
|
|
|
def fake_run(cfg, prompt): # type: ignore[no-untyped-def]
|
|
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, code_cfg=None, **kwargs): # type: ignore[no-untyped-def]
|
|
storage = FakeStorage()
|
|
calls = _install(monkeypatch, **kwargs)
|
|
cfg = code_cfg if code_cfg is not None else _code_cfg()
|
|
result = module.propose_code_fix(storage, INCIDENT_ID, JOB, 7, BUNDLE, _hermes_cfg(), 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",
|
|
"repo": "bstein/hermes-code-demo",
|
|
"candidates": ["src/discount.py"],
|
|
"chosen_path": "src/discount.py",
|
|
}
|
|
serialized = json.dumps(detail)
|
|
assert "return price" not in serialized
|
|
assert "secret-token" not in serialized
|
|
|
|
|
|
def test_existing_open_proposal_suppresses_duplicate(monkeypatch) -> None:
|
|
"""Only the ceiling suppresses new work, not a single open proposal."""
|
|
|
|
existing = {
|
|
"found": True,
|
|
"open_count": module.DEFAULT_MAX_OPEN_PROPOSALS,
|
|
"pr_number": 1,
|
|
"url": "https://scm.example/pulls/1",
|
|
"branch": "hermes-repair/4",
|
|
"error": None,
|
|
}
|
|
storage, calls, result = _propose(monkeypatch, existing=existing)
|
|
assert result == {
|
|
"status": "human_required",
|
|
"reason": "open_proposal_limit_reached",
|
|
"pr_number": 1,
|
|
"url": "https://scm.example/pulls/1",
|
|
"branch": "hermes-repair/4",
|
|
}
|
|
assert calls["lookups"] == [_code_cfg()]
|
|
assert calls["fetches"] == []
|
|
assert calls["runs"] == []
|
|
assert calls["pushes"] == []
|
|
assert calls["pulls"] == []
|
|
assert _event(storage) == {
|
|
"incident_id": INCIDENT_ID,
|
|
"job": JOB,
|
|
"build_number": 7,
|
|
"run_id": None,
|
|
"validated": False,
|
|
"reject_reason": "open_proposal_limit_reached",
|
|
"branch": "hermes-repair/4",
|
|
"pr_number": 1,
|
|
"url": "https://scm.example/pulls/1",
|
|
"repo": "bstein/hermes-code-demo",
|
|
"candidates": [],
|
|
"chosen_path": None,
|
|
}
|
|
|
|
|
|
def test_proposal_lookup_error_fails_open(monkeypatch) -> None:
|
|
storage, calls, result = _propose(
|
|
monkeypatch, existing=_no_existing(error="open proposal lookup http 503")
|
|
)
|
|
assert result["status"] == "pr_opened"
|
|
assert len(calls["fetches"]) == 1
|
|
assert len(calls["runs"]) == 1
|
|
assert _event(storage)["validated"] is True
|
|
|
|
|
|
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_job=JOB,
|
|
hermes_code_candidate_path="src/discount.py",
|
|
hermes_code_allowed_prefixes=["src/"],
|
|
hermes_code_allowed_suffixes=[".py"],
|
|
hermes_code_max_patch_bytes=4000,
|
|
hermes_code_max_changed_lines=20,
|
|
hermes_gitea_base_url="https://scm.example",
|
|
hermes_gitea_token="secret-token",
|
|
hermes_code_owner="bstein",
|
|
hermes_code_repo="hermes-code-demo",
|
|
hermes_code_base_branch="master",
|
|
)
|
|
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.hermes_jenkins_client,
|
|
"fetch_job_payload",
|
|
lambda job: {
|
|
"lastBuild": {
|
|
"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
|
|
|
|
|
|
def test_open_proposals_below_the_ceiling_do_not_suppress_new_work(monkeypatch) -> None:
|
|
"""A fix for one failure must not be withheld because another awaits review."""
|
|
|
|
existing = {
|
|
"found": True,
|
|
"open_count": 3,
|
|
"pr_number": 1,
|
|
"url": "https://scm.example/pulls/1",
|
|
"branch": "hermes-repair/4",
|
|
"error": None,
|
|
}
|
|
_storage, _calls, result = _propose(monkeypatch, existing=existing)
|
|
|
|
assert result["status"] == "pr_opened"
|
|
|
|
|
|
def test_the_ceiling_is_configurable(monkeypatch) -> None:
|
|
"""Review capacity is a property of the reviewer, not of the code."""
|
|
|
|
existing = {
|
|
"found": True,
|
|
"open_count": 2,
|
|
"pr_number": 1,
|
|
"url": "https://scm.example/pulls/1",
|
|
"branch": "hermes-repair/4",
|
|
"error": None,
|
|
}
|
|
cfg = {**_code_cfg(), "max_open_proposals": 2}
|
|
_storage, _calls, result = _propose(monkeypatch, existing=existing, code_cfg=cfg)
|
|
|
|
assert result["reason"] == "open_proposal_limit_reached"
|