fix: validate Cua window discovery filters and guide recovery

tests/cua_discovery_smoke.py

@@ -0,0 +1,88 @@
+"""Opt-in read-only app discovery against an already open, named application."""
+
+import argparse
+import asyncio
+import json
+import os
+from pathlib import Path
+
+import jsonschema
+from artemis.plugins.models import ServerConfig
+from artemis.plugins.session import PluginSession
+
+
+async def run(executable: Path, app: str, output: Path):
+    session = await PluginSession(
+        ServerConfig(command=str(executable)), executable.parent
+    ).start()
+    try:
+        schema = next(t.inputSchema for t in session.tools if t.name == "list_windows")
+        query = {"app": app, "on_screen_only": True}
+        bad_query = {**query, "pid": 0}
+        try:
+            jsonschema.validate(bad_query, schema)
+        except jsonschema.ValidationError:
+            pass
+        else:
+            raise AssertionError("Host schema still accepts PID 0")
+        bad = await session.call("list_windows", bad_query)
+        assert bad.isError
+        assert bad.structuredContent["artemis"]["execution"] == "not_started"
+        assert "omit pid" in "\n".join(b.text for b in bad.content if b.type == "text")
+
+        jsonschema.validate(query, schema)
+        discovered = await session.call("list_windows", query)
+        assert not discovered.isError
+        windows = discovered.structuredContent["windows"]
+        assert windows, "Open the intended app before running this opt-in probe"
+
+        nullable = {**query, "pid": None}
+        jsonschema.validate(nullable, schema)
+        found = await session.call("list_windows", nullable)
+        assert not found.isError and found.structuredContent["windows"]
+        assert "pid" not in found.structuredContent["lookup"]["filters"]
+
+        # This Python test process is not the requested application. A conflict
+        # must stay empty and suggest a new app-only query, without a guessed PID.
+        conflict = await session.call("list_windows", {**query, "pid": os.getpid()})
+        assert not conflict.isError and conflict.structuredContent["windows"] == []
+        next_query = conflict.structuredContent["lookup"]["suggested_arguments"]
+        assert next_query == query
+        assert "frontmost" not in "\n".join(
+            b.text for b in conflict.content if b.type == "text"
+        )
+        recovered = await session.call("list_windows", next_query)
+        assert not recovered.isError and recovered.structuredContent["windows"]
+
+        window = recovered.structuredContent["windows"][0]
+        target = {k: window[k] for k in ("pid", "window_id")}
+        state = await session.call("get_window_state", {**target, "max_elements": 100})
+        assert not state.isError, state
+        assert any(b.type == "image" for b in state.content)
+        assert state.structuredContent["snapshot_id"]
+        assert state.structuredContent["tree_markdown"]
+        assert state.structuredContent["artemis"]["source"]["target_id"] == str(
+            target["window_id"]
+        )
+        report = {
+            "app": app,
+            "zero_pid_rejected_by_host_and_adapter": True,
+            "app_only_window_count": len(windows),
+            "null_pid_omitted": True,
+            "conflicting_pid_recovered_without_substitution": True,
+            "screenshot_and_tree": True,
+        }
+        output.parent.mkdir(parents=True, exist_ok=True)
+        output.write_text(json.dumps(report, ensure_ascii=False, indent=2), "utf-8")
+        print(json.dumps(report, ensure_ascii=False))
+    finally:
+        await session.aclose()
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(__doc__)
+    parser.add_argument("executable", type=Path)
+    parser.add_argument("--app", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    asyncio.run(run(args.executable.resolve(), args.app, args.output))