Add GPT semantic command stack with guarded execution

.test/test_gpt_semantic_executor.py

@@ -0,0 +1,145 @@
+"""Executor behavior tests for GPT semantic commands.
+
+Purpose:
+- Verifies action mapping, browser fallback behavior, synchronization hooks, and stop-on-error policy.
+
+Called from:
+- Pytest suite in `user/talon-ai-tools/.test`.
+"""
+
+import sys
+
+sys.path.append(".")
+
+from GPT.semantic import gpt_semantic_executor_helpers as helpers
+from GPT.semantic import gpt_semantic_step_executor as step_executor
+from GPT.semantic.gpt_semantic_executor import GptSemanticExecutionError, execute_plan
+from GPT.semantic.gpt_semantic_types import GptSemanticPlan, GptSemanticStep
+
+
+class Proxy:
+    def __init__(self, runner: "FakeRunner", prefix: str):
+        self._runner = runner
+        self._prefix = prefix
+
+    def __getattr__(self, name: str):
+        def call(*args):
+            self._runner.record(f"{self._prefix}.{name}", *args)
+
+        return call
+
+
+class FakeRunner:
+    def __init__(self, fail_call: str | None = None, fail_message: str = "boom"):
+        self.calls: list[tuple[str, tuple[object, ...]]] = []
+        self.fail_call = fail_call
+        self.fail_message = fail_message
+        self.user = Proxy(self, "user")
+        self.browser = Proxy(self, "browser")
+        self.app = Proxy(self, "app")
+        self.edit = Proxy(self, "edit")
+
+    def record(self, name: str, *args) -> None:
+        self.calls.append((name, args))
+        if name == self.fail_call:
+            raise RuntimeError(self.fail_message)
+
+    def insert(self, text: str) -> None:
+        self.record("insert", text)
+
+    def key(self, combo: str) -> None:
+        self.record("key", combo)
+
+    def sleep(self, value: str) -> None:
+        self.record("sleep", value)
+
+
+def test_mapping_includes_launch_app(monkeypatch) -> None:
+    monkeypatch.setattr(helpers, "resolve_launch_command", lambda _name: "text-editor")
+    plan = GptSemanticPlan(
+        steps=[
+            GptSemanticStep("launch_app", {"app_name": "Text Editor"}),
+            GptSemanticStep("new_tab", {}),
+        ]
+    )
+    runner = FakeRunner()
+    execute_plan(plan, runner=runner)
+    assert runner.calls[0] == ("user.switcher_launch", ("text-editor",))
+    assert runner.calls[1] == ("app.tab_open", ())
+
+
+def test_launch_app_uses_resolved_command(monkeypatch) -> None:
+    monkeypatch.setattr(
+        helpers, "resolve_launch_command", lambda _name: "gnome-text-editor"
+    )
+    plan = GptSemanticPlan([GptSemanticStep("launch_app", {"app_name": "Gedit"})])
+    runner = FakeRunner()
+    execute_plan(plan, runner=runner)
+    assert runner.calls[0] == ("user.switcher_launch", ("gnome-text-editor",))
+
+
+def test_stop_on_first_error() -> None:
+    plan = GptSemanticPlan(
+        [GptSemanticStep("new_tab", {}), GptSemanticStep("copy", {})]
+    )
+    runner = FakeRunner(fail_call="app.tab_open")
+    try:
+        execute_plan(plan, runner=runner)
+        assert False
+    except GptSemanticExecutionError as exc:
+        assert "Step 1 (new_tab)" in str(exc)
+    assert len(runner.calls) == 1
+
+
+def test_browser_fallbacks() -> None:
+    msg = "Action 'browser.go' exists but the Module method is empty and no Context reimplements it"
+    runner = FakeRunner(fail_call="browser.go", fail_message=msg)
+    execute_plan(
+        GptSemanticPlan([GptSemanticStep("go_url", {"url": "https://x"})]), runner
+    )
+    assert ("key", ("ctrl-l",)) in runner.calls
+    assert ("insert", ("https://x",)) in runner.calls
+
+
+def test_switch_app_waits_for_focus(monkeypatch) -> None:
+    waits: list[tuple[str, str | None]] = []
+    monkeypatch.setattr(
+        step_executor, "wait_for_app_focus", lambda _r, n, c=None: waits.append((n, c))
+    )
+    plan = GptSemanticPlan(
+        [GptSemanticStep("switch_app", {"app_name": "Google Chrome"})]
+    )
+    runner = FakeRunner()
+    execute_plan(plan, runner=runner)
+    assert runner.calls[0] == ("user.switcher_focus", ("Google Chrome",))
+    assert waits == [("Google Chrome", None)]
+
+
+def test_execute_plan_applies_settle_after_each_step(monkeypatch) -> None:
+    settled: list[str] = []
+    monkeypatch.setattr(
+        step_executor, "settle_after_step", lambda action, _r: settled.append(action)
+    )
+    plan = GptSemanticPlan(
+        [GptSemanticStep("new_tab", {}), GptSemanticStep("copy", {})]
+    )
+    execute_plan(plan, runner=FakeRunner())
+    assert settled == ["new_tab", "copy"]
+
+
+def test_switch_app_falls_back_to_launch_if_not_running(monkeypatch) -> None:
+    waits: list[tuple[str, str | None]] = []
+    monkeypatch.setattr(step_executor, "is_running_app", lambda _name: False)
+    monkeypatch.setattr(
+        step_executor, "launch_candidates", lambda _name: ["google-chrome"]
+    )
+    monkeypatch.setattr(
+        step_executor, "wait_for_app_focus", lambda _r, n, c=None: waits.append((n, c))
+    )
+    plan = GptSemanticPlan(
+        [GptSemanticStep("switch_app", {"app_name": "Google Chrome"})]
+    )
+    runner = FakeRunner()
+    execute_plan(plan, runner=runner)
+    assert runner.calls[0] == ("user.switcher_launch", ("google-chrome",))
+    assert waits == [("Google Chrome", "google-chrome")]

.test/test_gpt_semantic_guardrails.py

@@ -0,0 +1,53 @@
+import sys
+
+sys.path.append(".")
+
+from GPT.semantic.gpt_semantic_guardrails import (
+    GptSemanticGuardrailError,
+    validate_guardrails,
+)
+from GPT.semantic.gpt_semantic_types import GptSemanticPlan, GptSemanticStep
+
+
+def test_step_limit() -> None:
+    plan = GptSemanticPlan([GptSemanticStep("new_tab", {}) for _ in range(2)])
+    try:
+        validate_guardrails(
+            plan, max_steps=1, max_total_sleep_ms=2000, max_insert_chars=50
+        )
+        assert False
+    except GptSemanticGuardrailError as exc:
+        assert "maximum is 1" in str(exc)
+
+
+def test_sleep_budget() -> None:
+    plan = GptSemanticPlan([GptSemanticStep("sleep", {"ms": 3000})])
+    try:
+        validate_guardrails(
+            plan, max_steps=5, max_total_sleep_ms=2000, max_insert_chars=50
+        )
+        assert False
+    except GptSemanticGuardrailError as exc:
+        assert "Total sleep is" in str(exc)
+
+
+def test_insert_length() -> None:
+    plan = GptSemanticPlan([GptSemanticStep("insert_text", {"text": "x" * 10})])
+    try:
+        validate_guardrails(
+            plan, max_steps=5, max_total_sleep_ms=2000, max_insert_chars=5
+        )
+        assert False
+    except GptSemanticGuardrailError as exc:
+        assert "insert_text exceeds 5" in str(exc)
+
+
+def test_blocked_combo() -> None:
+    plan = GptSemanticPlan([GptSemanticStep("key", {"combo": "ctrl-w"})])
+    try:
+        validate_guardrails(
+            plan, max_steps=5, max_total_sleep_ms=2000, max_insert_chars=50
+        )
+        assert False
+    except GptSemanticGuardrailError as exc:
+        assert "blocked" in str(exc)

.test/test_gpt_semantic_launch_catalog.py

@@ -0,0 +1,19 @@
+import sys
+
+sys.path.append(".")
+
+from GPT.semantic import gpt_semantic_launch_catalog as catalog
+
+
+def test_resolve_launch_command_prefers_name(monkeypatch) -> None:
+    rows = (("Text Editor", "gnome-text-editor"), ("Terminal", "gnome-terminal"))
+    monkeypatch.setattr(catalog, "launch_entries", lambda: rows)
+    assert catalog.resolve_launch_command("text editor") == "gnome-text-editor"
+
+
+def test_launch_context_text(monkeypatch) -> None:
+    rows = (("Text Editor", "gnome-text-editor"),)
+    monkeypatch.setattr(catalog, "launch_entries", lambda: rows)
+    text = catalog.launch_context_text()
+    assert "Launchable apps for launch_app" in text
+    assert "Text Editor => gnome-text-editor" in text

.test/test_gpt_semantic_launch_provider.py

@@ -0,0 +1,39 @@
+import sys
+
+sys.path.append(".")
+
+from GPT.semantic.gpt_semantic_launch_provider import LaunchProvider
+
+
+def test_read_entries_prefers_community(monkeypatch) -> None:
+    provider = LaunchProvider([])
+    monkeypatch.setattr(
+        provider, "_community_entries", lambda: (("B", "2"), ("A", "1"))
+    )
+    monkeypatch.setattr(provider, "_desktop_entries", lambda: (("X", "x"),))
+    assert provider.read_entries() == (("B", "2"), ("A", "1"))
+
+
+def test_read_entries_falls_back_to_desktop(monkeypatch) -> None:
+    provider = LaunchProvider([])
+    monkeypatch.setattr(provider, "_community_entries", lambda: ())
+    monkeypatch.setattr(provider, "_desktop_entries", lambda: (("X", "x"),))
+    assert provider.read_entries() == (("X", "x"),)
+
+
+def test_load_community_apps_filters_empty_values(monkeypatch) -> None:
+    provider = LaunchProvider([])
+    monkeypatch.setattr(
+        provider,
+        "_app_getter",
+        lambda: lambda: {
+            "Editor": "gedit",
+            "": "bad",
+            "Term": "",
+            "Calc": "gnome-calculator",
+        },
+    )
+    assert provider._load_community_apps() == {
+        "Editor": "gedit",
+        "Calc": "gnome-calculator",
+    }

.test/test_gpt_semantic_parser.py

@@ -0,0 +1,50 @@
+import sys
+
+sys.path.append(".")
+
+from GPT.semantic.gpt_semantic_parser import GptSemanticParseError, parse_plan
+
+
+def test_parse_valid_plan() -> None:
+    raw = (
+        '{"steps":[{"action":"launch_app","args":{"app_name":"Text Editor"}},'
+        '{"action":"go_url","args":{"url":"https://example.com"}}]}'
+    )
+    plan = parse_plan(raw)
+    assert len(plan.steps) == 2
+    assert plan.steps[0].args["app_name"] == "Text Editor"
+
+
+def test_parse_malformed_json() -> None:
+    try:
+        parse_plan("{")
+        assert False
+    except GptSemanticParseError as exc:
+        assert "not valid JSON" in str(exc)
+
+
+def test_parse_unknown_action() -> None:
+    raw = '{"steps":[{"action":"launch_missiles","args":{}}]}'
+    try:
+        parse_plan(raw)
+        assert False
+    except GptSemanticParseError as exc:
+        assert "unsupported action" in str(exc)
+
+
+def test_parse_bad_arg_type() -> None:
+    raw = '{"steps":[{"action":"sleep","args":{"ms":"200"}}]}'
+    try:
+        parse_plan(raw)
+        assert False
+    except GptSemanticParseError as exc:
+        assert "must be int" in str(exc)
+
+
+def test_parse_rejects_extra_fields() -> None:
+    raw = '{"steps":[{"action":"new_tab","args":{},"extra":true}],"garbage":1}'
+    try:
+        parse_plan(raw)
+        assert False
+    except GptSemanticParseError as exc:
+        assert "unsupported" in str(exc)

.test/test_gpt_semantic_transport.py

@@ -0,0 +1,63 @@
+import sys
+
+sys.path.append(".")
+
+from GPT.semantic import gpt_semantic_transport as transport
+
+
+def test_routes_to_api(monkeypatch) -> None:
+    calls: list[str] = []
+
+    def fake_api(request, system_message, model):
+        calls.append(f"api:{model}:{system_message}:{request['content'][0]['text']}")
+        return {"type": "text", "text": "ok-api"}
+
+    def fake_llm(*_args, **_kwargs):
+        calls.append("llm")
+        return {"type": "text", "text": "ok-llm"}
+
+    monkeypatch.setattr(transport, "_model_endpoint", lambda: "https://example.com")
+    monkeypatch.setattr(
+        transport,
+        "_helpers",
+        lambda: {
+            "resolve_model_name": lambda m: m,
+            "format_message": lambda text: {"type": "text", "text": text},
+            "extract_message": lambda resp: resp["text"],
+            "send_request_to_api": fake_api,
+            "send_request_to_llm_cli": fake_llm,
+        },
+    )
+
+    result = transport.request_completion("sys", "user", "m", debug=False)
+    assert result == "ok-api"
+    assert calls[0].startswith("api:m:sys:user")
+
+
+def test_routes_to_llm(monkeypatch) -> None:
+    calls: list[str] = []
+
+    def fake_api(*_args, **_kwargs):
+        calls.append("api")
+        return {"type": "text", "text": "ok-api"}
+
+    def fake_llm(prompt, _content, system, model, _thread):
+        calls.append(f"llm:{model}:{system}:{prompt['text']}")
+        return {"type": "text", "text": "ok-llm"}
+
+    monkeypatch.setattr(transport, "_model_endpoint", lambda: "llm")
+    monkeypatch.setattr(
+        transport,
+        "_helpers",
+        lambda: {
+            "resolve_model_name": lambda m: f"resolved-{m}",
+            "format_message": lambda text: {"type": "text", "text": text},
+            "extract_message": lambda resp: resp["text"],
+            "send_request_to_api": fake_api,
+            "send_request_to_llm_cli": fake_llm,
+        },
+    )
+
+    result = transport.request_completion("sys", "user", "m", debug=False)
+    assert result == "ok-llm"
+    assert calls[0].startswith("llm:resolved-m:sys:user")