"""Multi-repository code-proposal tests for the Hermes code flow. The legacy single-repo demo path lives in test_hermes_code_flow.py; this file covers per-job repository resolution, evidence-driven candidate selection, and the additive proposal a real service job earns when triage escalates it. Shared fakes are imported from that module and from the auto-triage harness so every file drives the same storage, Gitea, and Hermes stand-ins. """ from __future__ import annotations import json from types import SimpleNamespace import pytest from ariadne.services import hermes_autotriage as autotriage from ariadne.services import hermes_code_flow as module from tests.hermes_autotriage_harness import ( INCIDENT_ID as TRIAGE_INCIDENT_ID, JOB as TRIAGE_JOB, _code_settings, _counter, _events, _model_output, _prepare_code, _statuses, ) from tests.test_hermes_code_flow import ( JOB, FakeStorage, _code_cfg, _event, _hermes_cfg, _install, _run, ) SERVICE_JOB = "titan-api" SERVICE_INCIDENT_ID = f"{SERVICE_JOB}/9" SERVICE_SOURCE = "def balance(rows):\n return sum(rows) - 1\n" SERVICE_TEST = "def test_balance():\n assert balance([1, 2]) == 3\n" SERVICE_BUNDLE = { "incident_id": SERVICE_INCIDENT_ID, "jenkins": { "job": SERVICE_JOB, "console_failures": [ { "marker": "FAILED ", "line_number": 40, "text": ( "FAILED tests/test_ledger.py::test_balance\n" "src/ledger.py:2: AssertionError\n" "src/ledger.py:2: in balance" ), } ], "console_tail": "ERROR: script returned exit code 1", }, "log_evidence": {"records": []}, } def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] values = { "hermes_code_job": JOB, "hermes_code_candidate_path": "src/discount.py", "hermes_code_allowed_prefixes": ["src/"], "hermes_code_allowed_suffixes": [".py"], "hermes_code_max_patch_bytes": 4000, "hermes_code_max_changed_lines": 20, "hermes_gitea_base_url": "https://scm.example", "hermes_gitea_token": "secret-token", "hermes_code_owner": "bstein", "hermes_code_repo": "hermes-code-demo", "hermes_code_base_branch": "master", } values.update(overrides) return SimpleNamespace(**values) def _service_cfg(**overrides) -> dict: # type: ignore[no-untyped-def] settings = _settings( hermes_code_repos=f"{SERVICE_JOB}=titan/api, other-job = titan/other", hermes_code_prefixes=f"{SERVICE_JOB}=src/|tests/", hermes_code_suffixes=f"{SERVICE_JOB}=.py", hermes_code_base_branches=f"{SERVICE_JOB}=main", **overrides, ) return module.code_config(settings) def _propose_service(monkeypatch, cfg=None, job=SERVICE_JOB, **kwargs): # type: ignore[no-untyped-def] storage = FakeStorage() calls = _install(monkeypatch, **kwargs) result = module.propose_code_fix( storage, SERVICE_INCIDENT_ID, job, 9, SERVICE_BUNDLE, _hermes_cfg(), _service_cfg() if cfg is None else cfg, ) return storage, calls, result def _service_fetch(**overrides) -> dict: # type: ignore[no-untyped-def] base = { "src/ledger.py": (SERVICE_SOURCE, None), "tests/test_ledger.py": (SERVICE_TEST, None), } base.update(overrides) return base def _service_output(path: str, original: str, replacement: str) -> str: return json.dumps( { "incident_id": SERVICE_INCIDENT_ID, "analysis": "The balance helper is off by one.", "patch": { "path": path, "original": original, "replacement": replacement, "rationale": "restore the intended balance", }, "human_required": False, "reason": "small localized fix", } ) def test_code_config_defaults_when_multi_repo_settings_absent() -> None: cfg = module.code_config(_settings()) assert cfg["repos"] == {} assert cfg["job_prefixes"] == {} assert cfg["job_suffixes"] == {} assert cfg["job_base_branches"] == {} assert cfg["max_candidates"] == 3 assert cfg["max_context_chars"] == 60000 def test_code_config_parses_per_job_maps() -> None: cfg = _service_cfg(hermes_code_max_candidates=5, hermes_code_max_context_chars=1234) assert cfg["repos"] == { SERVICE_JOB: {"owner": "titan", "repo": "api"}, "other-job": {"owner": "titan", "repo": "other"}, } assert cfg["job_prefixes"] == {SERVICE_JOB: ["src/", "tests/"]} assert cfg["job_suffixes"] == {SERVICE_JOB: [".py"]} assert cfg["job_base_branches"] == {SERVICE_JOB: "main"} assert cfg["max_candidates"] == 5 assert cfg["max_context_chars"] == 1234 def test_code_config_ignores_malformed_pairs() -> None: cfg = module.code_config( _settings( hermes_code_repos="broken, =owner/repo, job=, job2=owneronly, job3=titan/api", hermes_code_prefixes="job3=,job4=src/", hermes_code_max_candidates="not-a-number", ) ) assert cfg["repos"] == {"job3": {"owner": "titan", "repo": "api"}} assert cfg["job_prefixes"] == {"job4": ["src/"]} assert cfg["max_candidates"] == 3 def test_resolve_repo_config_uses_per_job_mapping() -> None: resolved = module.resolve_repo_config(SERVICE_JOB, _service_cfg()) assert resolved is not None assert resolved["owner"] == "titan" assert resolved["repo"] == "api" assert resolved["base_branch"] == "main" assert resolved["allowed_path_prefixes"] == ["src/", "tests/"] assert resolved["allowed_suffixes"] == [".py"] assert resolved["candidate_path"] == "" assert resolved["gitea_token"] == "secret-token" assert resolved["max_patch_bytes"] == 4000 def test_resolve_repo_config_falls_back_to_single_repo_settings() -> None: resolved = module.resolve_repo_config(JOB, _service_cfg()) assert resolved is not None assert (resolved["owner"], resolved["repo"]) == ("bstein", "hermes-code-demo") assert resolved["base_branch"] == "master" assert resolved["allowed_path_prefixes"] == ["src/"] assert resolved["candidate_path"] == "src/discount.py" def test_resolve_repo_config_is_unchanged_for_legacy_only_settings() -> None: assert module.resolve_repo_config(JOB, _code_cfg()) == _code_cfg() def test_resolve_repo_config_returns_none_for_unmapped_job() -> None: assert module.resolve_repo_config("unmapped-job", _service_cfg()) is None assert module.resolve_repo_config("", _service_cfg()) is None def test_unmapped_job_requires_human_without_http(monkeypatch) -> None: storage, calls, result = _propose_service(monkeypatch, job="unmapped-job") assert result == {"status": "human_required", "reason": "no_repo_mapping"} assert calls["lookups"] == [] assert calls["fetches"] == [] assert calls["runs"] == [] detail = _event(storage) assert detail["reject_reason"] == "no_repo_mapping" assert detail["repo"] is None assert detail["candidates"] == [] assert detail["chosen_path"] is None def test_no_candidate_files_makes_no_model_call(monkeypatch) -> None: bundle = {"incident_id": SERVICE_INCIDENT_ID, "jenkins": {"console_tail": "no file here"}} storage = FakeStorage() calls = _install(monkeypatch) result = module.propose_code_fix( storage, SERVICE_INCIDENT_ID, SERVICE_JOB, 9, bundle, _hermes_cfg(), _service_cfg() ) assert result == {"status": "human_required", "reason": "no_candidate_files"} assert len(calls["lookups"]) == 1 assert calls["fetches"] == [] assert calls["runs"] == [] assert _event(storage)["reject_reason"] == "no_candidate_files" def test_service_repo_offers_ranked_candidates(monkeypatch) -> None: storage, calls, result = _propose_service( monkeypatch, fetch=_service_fetch(), run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), ) assert result["status"] == "pr_opened" assert [path for _cfg, path in calls["fetches"]] == ["src/ledger.py", "tests/test_ledger.py"] push_cfg, _incident, build_number, patch, contents = calls["pushes"][0] assert (push_cfg["owner"], push_cfg["repo"], push_cfg["base_branch"]) == ("titan", "api", "main") assert (build_number, patch.path) == (9, "src/ledger.py") assert contents == "def balance(rows):\n return sum(rows)\n" detail = _event(storage) assert detail["repo"] == "titan/api" assert detail["candidates"] == ["src/ledger.py", "tests/test_ledger.py"] assert detail["chosen_path"] == "src/ledger.py" def test_prompt_lists_every_offered_candidate(monkeypatch) -> None: _, calls, _ = _propose_service( monkeypatch, fetch=_service_fetch(), run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), ) prompt = calls["runs"][0][1] assert "The repository is titan/api branch main." in prompt assert "- src/ledger.py\n- tests/test_ledger.py" in prompt assert "`patch.path` MUST be exactly one of these candidate paths" in prompt assert "Current content of the candidate file src/ledger.py:" in prompt assert "Current content of the candidate file tests/test_ledger.py:" in prompt assert SERVICE_SOURCE in prompt assert prompt.rstrip().endswith(SERVICE_TEST.rstrip()) def test_failed_candidate_fetch_is_skipped(monkeypatch) -> None: storage, calls, result = _propose_service( monkeypatch, fetch=_service_fetch(**{"tests/test_ledger.py": (None, "file fetch http 404")}), run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), ) assert result["status"] == "pr_opened" assert len(calls["fetches"]) == 2 prompt = calls["runs"][0][1] assert "tests/test_ledger.py" not in prompt.split("Failing test evidence bundle:")[0] assert _event(storage)["candidates"] == ["src/ledger.py"] def test_all_candidate_fetches_failing_requires_human(monkeypatch) -> None: storage, calls, result = _propose_service( monkeypatch, fetch={"src/ledger.py": (None, "file fetch http 500")} ) assert result == {"status": "human_required", "reason": "candidate_fetch_failed: file fetch http 500"} assert calls["runs"] == [] assert _event(storage)["candidates"] == [] def test_context_budget_skips_oversized_candidate(monkeypatch) -> None: cfg = _service_cfg(hermes_code_max_context_chars=len(SERVICE_SOURCE) + 1) storage, calls, result = _propose_service( monkeypatch, cfg=cfg, fetch=_service_fetch(), run=_run(output=_service_output("src/ledger.py", "sum(rows) - 1", "sum(rows)")), ) assert result["status"] == "pr_opened" assert len(calls["fetches"]) == 2 assert SERVICE_TEST not in calls["runs"][0][1] assert _event(storage)["candidates"] == ["src/ledger.py"] def test_patch_path_not_offered_requires_human(monkeypatch) -> None: storage, calls, result = _propose_service( monkeypatch, fetch=_service_fetch(), run=_run(output=_service_output("src/other.py", "sum(rows) - 1", "sum(rows)")), ) assert result["status"] == "human_required" assert result["reason"].startswith("patch_path_not_offered: got 'src/other.py'") assert calls["pushes"] == [] assert _event(storage)["chosen_path"] is None def test_validation_resolves_against_the_chosen_file(monkeypatch) -> None: storage, calls, result = _propose_service( monkeypatch, fetch=_service_fetch(), run=_run(output=_service_output("tests/test_ledger.py", "== 3", "== 2")), ) assert result["status"] == "pr_opened" assert result["path"] == "tests/test_ledger.py" _cfg, _incident, _build, patch, contents = calls["pushes"][0] assert patch.path == "tests/test_ledger.py" assert contents == "def test_balance():\n assert balance([1, 2]) == 2\n" assert _event(storage)["chosen_path"] == "tests/test_ledger.py" ESCALATION = _model_output( classification="unknown_build_failure", requested_action=None, human_required=True, reason="the failure matches no known signature", ) def _last_phase(storage) -> dict: # type: ignore[no-untyped-def] return _events(storage, autotriage.INCIDENT_EVENT_TYPE)[-1]["phase"] def test_escalated_service_job_gets_both_an_issue_and_a_pull_request(monkeypatch) -> None: requested = _counter("propose_code_fix", "requested") success = _counter("propose_code_fix", "success") env = _prepare_code(monkeypatch, run=_run(output=ESCALATION)) summary = autotriage.run_hermes_autotriage(env.storage) assert summary["jobs"][TRIAGE_JOB] == { "status": "human_required", "incident_id": TRIAGE_INCIDENT_ID, "reason": "human_required", } assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"] incident_id, job, build_number, bundle, code_cfg = env.calls["proposals"][0] assert (incident_id, job, build_number) == (TRIAGE_INCIDENT_ID, TRIAGE_JOB, 12) assert bundle["incident_id"] == TRIAGE_INCIDENT_ID assert code_cfg["repos"] == {TRIAGE_JOB: {"owner": "bstein", "repo": TRIAGE_JOB}} assert _last_phase(env.storage) == { "reason": "human_required", "code_proposal": { "branch": "hermes-repair/12", "pr_number": 8, "url": "https://scm.example/pulls/8", }, } assert env.calls["issues"][0]["code_proposal_url"] == "https://scm.example/pulls/8" assert _counter("propose_code_fix", "requested") == requested + 1.0 assert _counter("propose_code_fix", "success") == success + 1.0 def test_remediated_service_job_is_never_patched(monkeypatch) -> None: env = _prepare_code(monkeypatch) summary = autotriage.run_hermes_autotriage(env.storage) assert summary["jobs"][TRIAGE_JOB]["status"] == "awaiting_rebuild" assert env.calls["proposals"] == [] assert env.calls["gitea"] == [] @pytest.mark.parametrize("overrides", [{"hermes_code_repos": ""}, {"hermes_code_enabled": False}]) def test_out_of_scope_job_proposes_nothing_and_calls_no_gitea(monkeypatch, overrides) -> None: env = _prepare_code(monkeypatch, cfg=_code_settings(**overrides), run=_run(output=ESCALATION)) summary = autotriage.run_hermes_autotriage(env.storage) assert summary["jobs"][TRIAGE_JOB]["status"] == "human_required" assert _last_phase(env.storage) == {"reason": "human_required"} assert env.calls["proposals"] == [] assert env.calls["gitea"] == [] def test_unfinished_hermes_run_proposes_nothing(monkeypatch) -> None: env = _prepare_code(monkeypatch, run=_run(status="timeout")) summary = autotriage.run_hermes_autotriage(env.storage) assert summary["jobs"][TRIAGE_JOB]["reason"] == "hermes_run_timeout" assert env.calls["proposals"] == [] def test_declined_proposal_is_recorded_and_counted_rejected(monkeypatch) -> None: rejected = _counter("propose_code_fix", "rejected") env = _prepare_code( monkeypatch, result={"status": "human_required", "reason": "no_candidate_files"}, run=_run(output=ESCALATION), ) summary = autotriage.run_hermes_autotriage(env.storage) assert summary["jobs"][TRIAGE_JOB]["reason"] == "human_required" assert _last_phase(env.storage) == { "reason": "human_required", "code_proposal": {"reason": "no_candidate_files"}, } assert env.calls["issues"][0]["code_proposal_url"] == "" assert _counter("propose_code_fix", "rejected") == rejected + 1.0 def test_exploding_proposal_never_breaks_the_tick(monkeypatch) -> None: env = _prepare_code( monkeypatch, error=RuntimeError("gitea is unreachable"), run=_run(output=ESCALATION) ) summary = autotriage.run_hermes_autotriage(env.storage) assert summary["status"] == "ok" assert summary["jobs"][TRIAGE_JOB]["status"] == "human_required" assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"] assert _last_phase(env.storage) == {"reason": "human_required"} assert env.calls["issues"][0]["incident_id"] == TRIAGE_INCIDENT_ID