import asyncio import json from types import SimpleNamespace from unittest.mock import AsyncMock, Mock import pytest from apps.admin_console.services.automation_store import AutomationStore from artemis.config.exploration import ExplorationConfig from artemis.exploration.demonstrations import Demonstrations from artemis.exploration.store import ExplorationConflict, ExplorationStore from artemis.runtime import task_control @pytest.fixture def demo_session(tmp_path, monkeypatch): repository = ExplorationStore(AutomationStore(tmp_path / "history.db")) monkeypatch.setattr("artemis.exploration.runtime.explorations", repository) monkeypatch.setattr("artemis.exploration.demo_runtime.explorations", repository) monkeypatch.setattr(task_control, "TRACES_PATH", tmp_path / "traces") session = repository.create("验证操作结果", ExplorationConfig(), {}, "create") run = repository.reserve_run( session["id"], "start", kind="exploration", goal="验证", task_config={} ) repository.set_run_state(run["id"], "running") return repository, session, run @pytest.mark.parametrize( "window_class", ["Chrome_WidgetWin_1", "Progman", "WorkerW", "Shell_TrayWnd"] ) def test_window_demo_bounds_use_foreground_hwnd_not_virtual_desktop(monkeypatch, window_class): from artemis.exploration import demo_runtime user32 = Mock() user32.GetForegroundWindow.return_value = 2**40 + 123 user32.IsIconic.return_value = False user32.SetThreadDpiAwarenessContext.return_value = 17 def class_name(hwnd, buffer, size): buffer.value = window_class return len(window_class) def bounds(hwnd, pointer): rect = pointer._obj rect.left, rect.top, rect.right, rect.bottom = -1200, 50, -400, 650 return True def title(hwnd, buffer, size): buffer.value = "示范浏览器" return len(buffer.value) user32.GetClassNameW.side_effect = class_name user32.GetWindowRect.side_effect = bounds user32.GetWindowTextW.side_effect = title monkeypatch.setattr( demo_runtime.ctypes, "windll", SimpleNamespace(user32=user32), raising=False ) monkeypatch.setattr(demo_runtime, "os", SimpleNamespace(name="nt")) result = demo_runtime.desktop_source("window") user32.GetSystemMetrics.assert_not_called() assert user32.SetThreadDpiAwarenessContext.call_count == 2 if window_class != "Chrome_WidgetWin_1": assert result is None return assert result.target_id == str(2**40 + 123) assert result.label == "示范浏览器" assert result.window == {"left": -1200, "top": 50, "width": 800, "height": 600} assert result.screen is None assert (result.width, result.height) == (800, 600) def test_demo_command_retries_do_not_extend_twice_or_finish_other_sessions( demo_session, monkeypatch ): repository, session, run = demo_session now = [100.0] monkeypatch.setattr("artemis.exploration.demonstrations.time.time", lambda: now[0]) demos = Demonstrations(repository) demo = demos.request(session["id"], "both", "request") assert demos.request(session["id"], "both", "request")["id"] == demo["id"] with pytest.raises(ExplorationConflict): demos.request(session["id"], "android", "second") demos.transition(demo["id"], "recording") for _ in range(2): demos.command(session["id"], demo["id"], "extend", "extend", seconds=60) assert demos.get(demo["id"])["deadline"] == 1060 other = repository.create("另一个目标", ExplorationConfig(), {}, "other") with pytest.raises(KeyError): demos.command(other["id"], demo["id"], "finish", "finish") now[0] = 1061 with pytest.raises(ExplorationConflict): demos.command(session["id"], demo["id"], "finish", "finish") demos.transition(demo["id"], "timed_out", media=[{"saved": True}]) assert repository.get(session["id"])["active_demonstration_id"] is None assert repository.run(run["id"])["outcome"] is None assert demos.transition(demo["id"], "completed")["status"] == "timed_out" @pytest.mark.asyncio async def test_demo_safe_point_keeps_lease_and_restores_capture_after_success( demo_session, monkeypatch ): from artemis.exploration.runtime import ExplorationControl repository, session, run = demo_session demo = Demonstrations(repository).request(session["id"], "android", "demo") lock = SimpleNamespace(device_id="test-device", release=Mock()) control = ExplorationControl(run["id"], lock) control.context = SimpleNamespace() stop, resume = AsyncMock(), AsyncMock() async def perform(ctx, observed_control, record): assert record["id"] == demo["id"] stop.assert_awaited_once() lock.release.assert_not_called() observed_control._state("demonstrating") assert task_control.read_state(run["id"])["status"] == "demonstrating" return "" monkeypatch.setattr("artemis.exploration.demo_runtime.perform_demonstration", perform) task_control.request_pause(run["id"], reason="demonstration") assert await control.checkpoint(on_pause=stop, on_resume=resume) lock.release.assert_not_called() resume.assert_awaited_once() @pytest.mark.asyncio async def test_demo_timeout_saves_media_then_releases_lease_without_failing( demo_session, monkeypatch ): from artemis.exploration.runtime import ExplorationControl repository, session, run = demo_session demo = Demonstrations(repository).request(session["id"], "android", "demo") lock = SimpleNamespace(device_id="test-device", release=Mock()) control = ExplorationControl(run["id"], lock) control.context = SimpleNamespace() async def perform(*_): Demonstrations(repository).transition( demo["id"], "timed_out", media=[{"file": "saved.mp4"}] ) return "demonstration_timed_out" monkeypatch.setattr("artemis.exploration.demo_runtime.perform_demonstration", perform) task_control.request_pause(run["id"], reason="demonstration") waiting = asyncio.create_task(control.checkpoint()) for _ in range(100): await asyncio.sleep(0.01) if waiting.done(): waiting.result() if repository.get(session["id"])["status"] == "paused": break try: assert lock.release.called assert repository.get(session["id"])["status"] == "paused" assert repository.run(run["id"])["outcome"] is None assert Demonstrations(repository).get(demo["id"])["media"] finally: waiting.cancel() await asyncio.gather(waiting, return_exceptions=True) def test_system_demo_summary_cannot_answer_pending_user_question(demo_session): from artemis.exploration.runtime import ExplorationControl repository, session, run = demo_session control = ExplorationControl(run["id"], Mock()) question = repository.ask(run["id"], "预期是哪个状态？", "开启", "q") task_control.add_instruction(run["id"], "视频看到了点击", source="demonstration") assert not control.question_answered() repository.answer_question(session["id"], question["id"]) assert control.question_answered() def test_legacy_demo_source_survives_recent_message_window_without_copying_media(demo_session): from artemis.exploration.runtime import conversation_message repository, session, run = demo_session demo = { "id": "old-demo", "run_id": run["id"], "exploration_id": session["id"], "source": "computer", "status": "completed", "media": [{"kind": "computer", "segments": [{"file": "private/path/recording.mp4"}]}], } repository.database.put("exploration_demo", demo["id"], demo) repository.database.put( "exploration_demo", "failed-demo", {**demo, "id": "failed-demo", "status": "failed"} ) repository.add_message( session["id"], "打开浏览器看首页", "legacy-summary", role="assistant", metadata={"demonstration_id": demo["id"], "kind": "demonstration"}, ) for i in range(4): repository.add_message(session["id"], "最新指导" * 2000, f"guidance-{i}") ctx = SimpleNamespace(exploration_id=session["id"], exploration_run_id=run["id"]) payload = json.loads(conversation_message(ctx).content.split("\n", 1)[1]) assert not any(m.get("demonstration_id") for m in payload["messages"]) assert payload["demonstrations"] == [ { "id": "old-demo", "run_id": run["id"], "source": "computer", "recorded_sources": ["computer"], } ] assert "private/path" not in str(payload) @pytest.mark.asyncio @pytest.mark.parametrize("needs_user", [True, False]) @pytest.mark.parametrize( ("source", "recorded_sources"), [("computer", ["computer"]), ("android", ["android"]), ("both", ["android", "computer"])], ) async def test_demo_source_survives_clarification_and_execution_restart( demo_session, monkeypatch, needs_user, source, recorded_sources ): from apps.admin_console.services.exploration_service import ExplorationService from artemis.exploration import demo_runtime from artemis.exploration.runtime import conversation_message repository, session, run = demo_session demos = Demonstrations(repository) demo = demos.request(session["id"], source, "demo") demos.transition(demo["id"], "recording") demos.command(session["id"], demo["id"], "finish", "finish") capture = SimpleNamespace( start=AsyncMock(), stop=AsyncMock(), errors=[], media=[{"id": f"track-{kind}", "kind": kind} for kind in recorded_sources], ) monkeypatch.setattr(demo_runtime, "DemonstrationCapture", lambda *_: capture) monkeypatch.setattr( demo_runtime, "analyze_demonstration", AsyncMock( return_value=demo_runtime.DemonstrationAnalysis( summary="打开浏览器，浏览首页后返回顶部。", needs_user=needs_user, question="目标网址是什么？" if needs_user else "", ) ), ) async def run_interruptibly(operation, **_): return False, await operation control = SimpleNamespace( session_id=run["id"], directory=task_control.control_dir(run["id"]), _state=Mock(), run_interruptibly=run_interruptibly, ) ctx = SimpleNamespace(exploration_id=session["id"], exploration_run_id=run["id"]) reason = await demo_runtime.perform_demonstration(ctx, control, demo) assert reason == ("question" if needs_user else "") pending = repository.get(session["id"])["pending_question"] assert bool(pending) == needs_user ExplorationService(repository).message(session["id"], "www.playones.com，首页看一遍就行", "url") assert repository.get(session["id"])["pending_question"] is None # Reopen the durable store: source must not depend on in-memory capture, # the model repeating a platform name, or whether analysis needed an answer. restored = ExplorationStore(repository.database) monkeypatch.setattr("artemis.exploration.runtime.explorations", restored) payload = json.loads(conversation_message(ctx).content.split("\n", 1)[1]) assert payload["demonstrations"] == [ { "id": demo["id"], "run_id": run["id"], "source": source, "recorded_sources": recorded_sources, } ] summary = next(m for m in payload["messages"] if m.get("demonstration_id") == demo["id"]) assert "打开浏览器" in summary["text"] assert ("电脑" if source == "computer" else "安卓") in summary["text"] instructions = task_control.consume_instructions(control.directory)["instruction"] assert "打开浏览器" in instructions assert "www.playones.com" in instructions assert ("电脑" if source == "computer" else "安卓") in instructions # Explicit user changes remain later guidance; provenance is evidence, # never a hard global execution-target switch that would forbid mixed tests. ExplorationService(restored).message(session["id"], "接下来切到手机核对页面", "switch") changed = json.loads(conversation_message(ctx).content.split("\n", 1)[1]) assert changed["messages"][-1]["text"] == "接下来切到手机核对页面" assert changed["demonstrations"] == payload["demonstrations"] assert restored.run(run["id"])["task_config"] == {} @pytest.mark.asyncio async def test_saved_video_extracts_copy_and_preserves_source_and_offsets(tmp_path, monkeypatch): from artemis.exploration.media import SavedVideoSource original = tmp_path / "demo.mp4" original.write_bytes(b"evidence") async def render(segments, start, end, target, **kwargs): assert (start, end) == (12, 14) target.write_bytes(b"temporary") return True monkeypatch.setattr("artemis.exploration.media.render_timeline_clip", render) source = SavedVideoSource( tmp_path, { "id": "demo-computer", "kind": "computer", "segments": [{"file": "demo.mp4", "offset_ms": 12000, "duration_ms": 2000}], }, ) result = await source.extract_segment_metadata(0, None) assert result.actual_start_relative_time == 12 assert result.source_revision == "demo-computer" result.video_path.unlink() result.video_path.parent.rmdir() assert original.read_bytes() == b"evidence" with pytest.raises(ValueError): SavedVideoSource( tmp_path, {"segments": [{"file": "../private.mp4", "offset_ms": 0, "duration_ms": 1}]} ) def test_saved_demo_sampling_reaches_end_of_long_interval(monkeypatch, tmp_path): from artemis.agents.video_analyzer.universal_engine import _gather_universal_frames sampling = Mock(return_value=[]) monkeypatch.setattr( "artemis.agents.video_analyzer.video_analyzer.extract_frames_at_timestamps", sampling ) analyzer = SimpleNamespace( media_source=object(), _dense_action_offsets=lambda *args: [], max_dense_action_frames=24 ) _gather_universal_frames(analyzer, tmp_path / "video.mp4", 32.249, 92.249, 32.249) offsets = sampling.call_args_list[0].args[1] assert len(offsets) <= 30 assert offsets[0] == 0 assert offsets[-1] == pytest.approx(59.9) @pytest.mark.asyncio async def test_saved_demo_chunk_keeps_millisecond_coverage(monkeypatch, tmp_path): from artemis.agents.video_analyzer.blackboard import VideoBlackboard from artemis.agents.video_analyzer import chunk_runner board = VideoBlackboard( "demo", db_path=tmp_path / "board.db", evidence_dir=tmp_path / "evidence" ) analyzer = SimpleNamespace( max_retries=0, ctx=SimpleNamespace(device=None), blackboard=board, model_name="test", video_processing="static", chunk_size_seconds=60, _exec_single_chunk_universal=AsyncMock(return_value="actual summary"), _parse_chunk_result=lambda _: ("已分析完整区间", "没有推断缺失操作"), ) monkeypatch.setattr(chunk_runner, "_prepare_chunk_media", AsyncMock(return_value=(None, 1))) monkeypatch.setattr(chunk_runner, "_build_chunk_prompt", lambda *args: "query") await chunk_runner.exec_single_chunk(analyzer, 92.249, 137.428, "query") assert board.missing_intervals(92.249, 137.428, "query") == [] @pytest.mark.asyncio async def test_failed_render_removes_partial_temporary_clip_but_not_source(tmp_path, monkeypatch): from artemis.exploration.media import SavedVideoSource original = tmp_path / "recording.mp4" original.write_bytes(b"saved") targets = [] async def render(*args, **kwargs): target = args[3] targets.append(target) target.write_bytes(b"partial") return False monkeypatch.setattr("artemis.exploration.media.render_timeline_clip", render) source = SavedVideoSource( tmp_path, { "id": "demo", "kind": "android", "segments": [{"file": original.name, "offset_ms": 0, "duration_ms": 1000}], }, ) assert not (await source.extract_segment_metadata(0, 1)).success assert not targets[0].parent.exists() assert original.read_bytes() == b"saved" def test_temporary_recording_cleanup_keeps_demonstration_playback(tmp_path): from artemis.context import ArtemisContext from artemis.data_engine.engine import DataEngine from unittest.mock import MagicMock from uuid import uuid4 import time ctx = MagicMock(spec=ArtemisContext) ctx.execution_setup = MagicMock(traces_path=str(tmp_path / "traces")) ctx.device = None engine = DataEngine(ctx) sid = engine.start_session(goal="test") keep = tmp_path / "demonstrations" keep.mkdir() demo = keep / "demo.mp4" temporary = tmp_path / "temporary.mp4" demo.write_bytes(b"saved") temporary.write_bytes(b"temporary") demo_id, temp_id = uuid4(), uuid4() for vid, path in ((demo_id, demo), (temp_id, temporary)): engine.record_video_start(vid, "device", path) engine.record_video_stop(vid, "device", path, time.time() - 10, time.time()) engine.storage.discard_session_recordings(str(sid), preserve_directory=keep) assert engine.storage.get_video_recording(demo_id).local_video_path == str(demo) assert engine.storage.get_video_recording(temp_id).local_video_path is None assert engine.storage.get_session(sid).video_filepath == str(demo)