Artemis-Eapil/tests/unit/agents/test_checker_structured_report.py

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""The structured verdict request must never end on a model turn (Gemini
rejects such requests with 400 "Requests ending with a model turn are not
supported"), which is the natural state after the loop's tool-free reply."""

import json
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock

import pytest
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from pydantic import ValidationError

from artemis.agents.checker.checker import CheckReport, _structured_report


def _llm(report: CheckReport):
    structured = MagicMock()
    structured.ainvoke = AsyncMock(return_value=report)
    llm = MagicMock()
    llm.with_structured_output.return_value = structured
    return llm, structured


@pytest.mark.asyncio
async def test_structured_report_appends_user_turn_after_model_reply():
    llm, structured = _llm(CheckReport(verdicts=[]))
    messages = [SystemMessage(content="s"), HumanMessage(content="h"), AIMessage(content="done")]

    await _structured_report(llm, messages)

    sent = structured.ainvoke.call_args.args[0]
    assert isinstance(sent[-1], HumanMessage)
    assert sent[-2] is messages[-1]
    assert len(messages) == 3  # caller's list untouched


@pytest.mark.asyncio
async def test_structured_report_keeps_user_ending_conversation_as_is():
    llm, structured = _llm(CheckReport(verdicts=[]))
    messages = [SystemMessage(content="s"), HumanMessage(content="final iteration")]

    await _structured_report(llm, messages)

    assert structured.ainvoke.call_args.args[0] is messages


def _result(status="passed", evidence=None):
    return {
        "status": status,
        "reason": "双端检查与报告已完成，已知风险不代表所有布局无缺陷。",
        "evidence": ["步骤 18、27：追加下滚后位置不变，双端已到页底。"]
        if evidence is None
        else evidence,
    }


@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["passed", "failed", "inconclusive", "blocked"])
async def test_complete_checker_json_does_not_request_another_verdict(status):
    payload = {"test_result": _result(status)}
    llm, structured = _llm(CheckReport())
    answer = AIMessage(
        content="核查步骤 [18]、[27]，控件 [28]。最终裁定\n```json\n"
        + json.dumps(payload)
        + "\n```\n范围仅限本次检查。"
    )

    result = await _structured_report(llm, [HumanMessage(content="review"), answer])

    assert result.test_result.status == status
    assert result.test_result.evidence == payload["test_result"]["evidence"]
    structured.ainvoke.assert_not_awaited()
    llm.with_structured_output.assert_not_called()


@pytest.mark.asyncio
@pytest.mark.parametrize("status", ["passed", "failed", "inconclusive"])
async def test_complete_verdict_alias_and_evidence_list_are_not_sent_for_reformatting(status):
    payload = {
        "verdicts": [
            {
                "kind": "assert",
                "item_text": "检查本轮页面",
                "verdict": status,
                "evidence": ["步骤 2 的实际页面", "步骤 3 的对照截图"],
            }
        ],
        "test_result": _result(status),
    }
    llm, structured = _llm(CheckReport())
    report = await _structured_report(
        llm,
        [AIMessage(content=json.dumps(payload))],
        [SimpleNamespace(kind="assert", text="检查本轮页面")],
    )
    assert report.verdicts[0].status == status
    assert report.verdicts[0].evidence == "步骤 2 的实际页面\n步骤 3 的对照截图"
    assert report.test_result.model_dump() == payload["test_result"]
    structured.ainvoke.assert_not_awaited()
    llm.with_structured_output.assert_not_called()


@pytest.mark.parametrize(
    "fields",
    [
        {"status": "failed", "verdict": "passed", "evidence": ["图 1"]},
        {"verdict": "unknown", "evidence": ["图 1"]},
        {"verdict": "passed", "evidence": [{"text": "图 1"}]},
        {"verdict": "passed", "evidence": [1]},
    ],
)
def test_conflicting_or_invalid_verdict_shapes_are_not_silently_accepted(fields):
    with pytest.raises(ValidationError):
        CheckReport.model_validate(
            {"verdicts": [{"kind": "assert", "item_text": "页面", **fields}]}
        )


def test_empty_evidence_list_does_not_turn_an_unsupported_failure_into_evidence():
    from artemis.agents.checker.checker import _normalize_report

    report = CheckReport.model_validate(
        {"verdicts": [{"kind": "assert", "item_text": "页面", "verdict": "failed", "evidence": []}]}
    )
    assert _normalize_report(report, []).verdicts[0].status == "inconclusive"


def test_nested_json_string_result_is_decoded_and_still_validated():
    result = _result()
    assert (
        CheckReport.model_validate({"test_result": json.dumps(result)}).test_result.model_dump()
        == result
    )
    for invalid in ("passed", "null", "[]", '{"status":"passed"}', json.dumps(_result("unknown"))):
        with pytest.raises(ValidationError):
            CheckReport.model_validate({"test_result": invalid})


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "kind", ["empty", "malformed", "truncated", "tool_call", "missing_check", "ambiguous", "prose"]
)
async def test_incomplete_or_ambiguous_answer_requires_structured_verdict(kind):
    text = json.dumps({"test_result": _result()})
    items = []
    kwargs = {}
    if kind == "empty":
        text = "{}"
    elif kind == "malformed":
        text = '{"test_result":{"status":"passed"}}'
    elif kind == "truncated":
        kwargs["response_metadata"] = {"finish_reason": "length"}
    elif kind == "tool_call":
        kwargs["tool_calls"] = [{"id": "read-1", "name": "search_history", "args": {}}]
    elif kind == "missing_check":
        items = [SimpleNamespace(kind="assert", text="must pass")]
    elif kind == "ambiguous":
        text += "\n" + json.dumps({"test_result": _result("failed")})
    elif kind == "prose":
        text = "测试通过。"
    llm, structured = _llm(CheckReport())

    await _structured_report(llm, [AIMessage(content=text, **kwargs)], items)

    structured.ainvoke.assert_awaited_once()


@pytest.mark.asyncio
async def test_local_verdict_preserves_evidence_normalization(monkeypatch):
    from artemis.agents.checker import checker

    llm, structured = _llm(CheckReport())
    monkeypatch.setattr(checker, "get_llm", lambda **_: llm)
    monkeypatch.setattr(
        checker,
        "acomplete",
        AsyncMock(
            return_value=AIMessage(
                content=json.dumps({"test_result": _result(evidence=[])}),
            )
        ),
    )
    ctx = SimpleNamespace(execution_setup=SimpleNamespace(checker_max_iterations=3))
    report = await checker._run_check_loop(ctx, [HumanMessage(content="review")], [], [])
    assert report.test_result.status == "inconclusive"
    structured.ainvoke.assert_not_awaited()