142 lines
5.3 KiB
Python
142 lines
5.3 KiB
Python
"""Opt-in actual HWND capture, action, artifact, recording and shutdown proof."""
|
|
|
|
import asyncio
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from artemis.config.agent import RecordingConfig
|
|
from artemis.plugins.evidence import (
|
|
decode_receipt,
|
|
persist_attachments,
|
|
source_identity,
|
|
validate_tools,
|
|
)
|
|
from artemis.plugins.models import PluginCallResult, PluginManifest, ServerConfig
|
|
from artemis.plugins.recording import PluginRecordings
|
|
from artemis.plugins.session import PluginSession
|
|
|
|
|
|
async def run(executable, output):
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix="artemis-cua-smoke-") as temporary:
|
|
ready = Path(temporary) / "ready"
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"powershell.exe",
|
|
"-NoProfile",
|
|
"-File",
|
|
str(Path(__file__).with_name("capture-fixture.ps1")),
|
|
"-Title",
|
|
"ARTEMIS Cua Smoke",
|
|
"-ReadyFile",
|
|
str(ready),
|
|
"-PerMonitorDpi",
|
|
creationflags=subprocess.CREATE_NO_WINDOW,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
session = recorder = None
|
|
try:
|
|
for _ in range(100):
|
|
if ready.exists():
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
assert ready.exists()
|
|
artifacts = output / "artifacts/plugins/cua-driver"
|
|
session = await PluginSession(
|
|
ServerConfig(
|
|
command=str(executable),
|
|
env={"ARTEMIS_ARTIFACTS_DIR": str(artifacts)},
|
|
),
|
|
executable.parent,
|
|
).start()
|
|
manifest = PluginManifest.model_validate_json(
|
|
(
|
|
Path(__file__).parents[1] / "plugins/cua-driver/plugin.json"
|
|
).read_text("utf-8")
|
|
)
|
|
validate_tools(manifest, session.tools)
|
|
windows = await session.call("list_windows", {"pid": proc.pid})
|
|
target = {
|
|
"pid": proc.pid,
|
|
"window_id": windows.structuredContent["windows"][0]["window_id"],
|
|
}
|
|
raw = await session.call("get_window_state", target)
|
|
assert not raw.isError, raw
|
|
outcome = PluginCallResult(
|
|
status="success", structured_content=raw.structuredContent
|
|
)
|
|
decode_receipt(outcome)
|
|
assert outcome.source and outcome.source.width > 0
|
|
snapshot = raw.structuredContent["snapshot_id"]
|
|
clicked = await session.call(
|
|
"click", {**target, "element_token": f"{snapshot}:0"}
|
|
)
|
|
assert not clicked.isError, clicked
|
|
assert Path(str(ready) + ".clicked").exists()
|
|
assert clicked.structuredContent["artemis"]["source"]
|
|
assert clicked.structuredContent["observation"]["snapshot_id"] != snapshot
|
|
saved = await session.call(
|
|
"save_screenshot", {**target, "filename": "cua-proof.png"}
|
|
)
|
|
assert not saved.isError, saved
|
|
receipt = saved.structuredContent["artemis"]
|
|
files = persist_attachments(output, "cua-driver", receipt["attachments"])
|
|
duplicate = await session.call(
|
|
"save_screenshot", {**target, "filename": "cua-proof.png"}
|
|
)
|
|
assert (
|
|
duplicate.isError
|
|
and duplicate.structuredContent["artemis"]["execution"] == "not_started"
|
|
)
|
|
context = SimpleNamespace(
|
|
data_engine=SimpleNamespace(
|
|
base_dir=output, session_start_time=time.time()
|
|
),
|
|
agent_config=SimpleNamespace(
|
|
recording=RecordingConfig(codec="h264", computer_scope="window")
|
|
),
|
|
execution_setup=SimpleNamespace(
|
|
record_video_enabled=True, video_recording_tools_enabled=False
|
|
),
|
|
)
|
|
recorder = PluginRecordings(context)
|
|
await recorder.observe(
|
|
source_identity("cua-driver", outcome.source),
|
|
"cua-driver",
|
|
outcome.source,
|
|
)
|
|
assert recorder.active
|
|
await asyncio.sleep(5)
|
|
await recorder.stop()
|
|
clips = [
|
|
s for track in recorder.sources.values() for s in track["segments"]
|
|
]
|
|
assert clips and clips[0]["duration"] >= 3
|
|
result = {
|
|
"actual_click": True,
|
|
"source": outcome.source.model_dump(),
|
|
"attachments": files,
|
|
"recordings": clips,
|
|
}
|
|
(output / "result.json").write_text(
|
|
json.dumps(result, ensure_ascii=False, indent=2), "utf-8"
|
|
)
|
|
print(json.dumps(result, ensure_ascii=False))
|
|
finally:
|
|
if recorder:
|
|
await recorder.stop()
|
|
if session:
|
|
await session.aclose()
|
|
proc.terminate()
|
|
await asyncio.wait_for(proc.wait(), 10)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
|
|
asyncio.run(run(Path(sys.argv[1]).resolve(), Path(sys.argv[2]).resolve()))
|