from types import SimpleNamespace from unittest.mock import AsyncMock, Mock import pytest from pydantic import ValidationError from artemis.agents.flash.progress import TaskProgress from artemis.agents.flash.runner import FlashRunner, _TurnRecord def plan(phone="active", pc="pending"): return { "reason": "按用户要求先手机页脚，再电脑独立检查", "stages": [ { "id": "phone", "scope": "Android 主体、页脚及最后滚动", "resources": ["android"], "status": phone, "evidence": "步骤 12 页脚及 13 滚动截图" if phone in {"done", "blocked"} else "", }, { "id": "pc", "scope": "电脑检查", "resources": ["cua-driver"], "status": pc, "evidence": "步骤 20 电脑截图" if pc == "done" else "", }, ], } def test_ordered_stages_require_evidence_and_keep_outstanding_scope(tmp_path): progress = TaskProgress() progress.update(plan(), resources={"android", "cua-driver"}, directory=tmp_path) assert progress.guard("cua-driver") assert progress.guard("android") is None assert progress.guard(None) is None # read-only preflight assert progress.guard(None, completed=True) assert (tmp_path / "notes/task_progress.md").exists() with pytest.raises(ValidationError): progress.update(plan("pending", "active"), resources={"android", "cua-driver"}) omitted = plan() omitted["stages"] = omitted["stages"][1:] with pytest.raises(ValueError, match="不能删除"): progress.update(omitted, resources={"android", "cua-driver"}) progress.update(plan("done", "active"), resources={"android", "cua-driver"}) assert progress.guard("cua-driver") is None assert progress.guard("android") progress.update(plan("done", "done"), resources={"android", "cua-driver"}) assert progress.guard(None, completed=True) is None def test_pc_first_and_intentional_interleaving_are_supported(): progress = TaskProgress() value = { "reason": "路由器调整后在手机验证", "stages": [ { "id": "mixed", "scope": "电脑改参数，手机核对，循环验证", "resources": ["android", "cua-driver"], "status": "active", } ], } progress.update(value, resources={"android", "cua-driver"}) assert progress.guard("cua-driver") is None assert progress.guard("android") is None value["stages"][0]["resources"] = ["cua-driver"] progress.update(value, resources={"cua-driver"}) assert progress.guard("cua-driver") is None def test_stage_updates_are_atomic_and_preserve_unmentioned_scope_and_order(tmp_path): progress = TaskProgress() resources = {"android", "cua-driver"} progress.update(plan(), resources=resources, directory=tmp_path) before = progress.value.model_dump() saved = (tmp_path / "notes/task_progress.md").read_bytes() with pytest.raises(ValidationError, match="完成或阻塞阶段必须记录"): progress.update( {"reason": "无证据不能完成", "stage_updates": [{"id": "phone", "status": "done"}]}, resources=resources, directory=tmp_path, ) assert progress.value.model_dump() == before assert (tmp_path / "notes/task_progress.md").read_bytes() == saved progress.update( { "reason": "手机已检查完", "stage_updates": [ {"id": "phone", "status": "done", "evidence": "实际步骤记录 real-step"} ], }, resources=resources, directory=tmp_path, ) assert progress.value.stages[1].model_dump() == before["stages"][1] assert [s.id for s in progress.value.stages] == ["phone", "pc"] assert progress.value.stages[0].scope == before["stages"][0]["scope"] progress.update( {"reason": "开始电脑检查", "stage_updates": [{"id": "pc", "status": "active"}]}, resources=resources, ) assert progress.guard("cua-driver") is None with pytest.raises(ValueError, match="未知 id"): progress.update( {"reason": "错拼不能覆盖", "stage_updates": [{"id": "typo", "status": "done"}]}, resources=resources, ) with pytest.raises(ValidationError, match="只能提供一种"): progress.update( {**plan(), "stage_updates": [{"id": "pc", "status": "active"}]}, resources=resources ) @pytest.mark.asyncio async def test_flash_preflight_does_not_authorize_pc_mutation(monkeypatch): plugins = SimpleNamespace( packages={"cua-driver": None}, bindings={"plug_pc_look": ("cua-driver", None), "plug_pc_click": ("cua-driver", None)}, tool_names={"plug_pc_look", "plug_pc_click"}, is_read_only=lambda name: name == "plug_pc_look", ) monkeypatch.setattr("artemis.plugins.runtime.get_plugin_runtime", lambda _: plugins) runner = FlashRunner.__new__(FlashRunner) runner.ctx = SimpleNamespace(data_engine=None, execution_target="android") runner.executor = SimpleNamespace(action_tool_names={"click"}) runner._progress = TaskProgress() runner._progress.update(plan(), resources={"android", "cua-driver"}) runner._execute_and_record_action = AsyncMock(return_value=(None, "", 0)) runner._refresh_observation = AsyncMock(return_value=(None, "")) messages = [] await runner._process_tool_calls( [ {"name": "plug_pc_look", "id": "look", "args": {}}, {"name": "plug_pc_click", "id": "click", "args": {}}, ], SimpleNamespace(indexed_elements=[]), messages, "预检", {}, None, "", 0, _TurnRecord(), ) runner._execute_and_record_action.assert_awaited_once() assert runner._execute_and_record_action.call_args.args[0] == "plug_pc_look" assert messages[-1].tool_call_id == "click" and messages[-1].status == "error" runner._refresh_observation.assert_not_awaited() @pytest.mark.asyncio async def test_final_report_reuses_exact_saved_body_and_rejects_missing_note(tmp_path): runner = FlashRunner.__new__(FlashRunner) runner.ctx = SimpleNamespace( data_engine=SimpleNamespace(base_dir=tmp_path, current_step_id="s", record_step=Mock()) ) runner.summarizer = None runner.final_report_received = False from artemis.utils.notes import save_verified_note body = "# 报告\nPC 是深色；手机检查未全部覆盖。" save_verified_note(tmp_path, "report", body) messages = [] args = {"status": "completed", "explanation": "全部完成", "report_note_key": "missing"} result = await runner._finalize_task_report( "report_task_status", args, "r", "", {}, None, "", messages ) assert result is None and not runner.final_report_received args["report_note_key"] = "report" result = await runner._finalize_task_report( "report_task_status", args, "r2", "", {}, None, "", messages ) assert result["explanation"] == body assert ( runner.ctx.data_engine.record_step.call_args.kwargs["action_taken"]["args"]["explanation"] == body )