diff --git a/.github/workflows/vibecad-workflow-harness.yml b/.github/workflows/vibecad-workflow-harness.yml new file mode 100644 index 00000000..275ef574 --- /dev/null +++ b/.github/workflows/vibecad-workflow-harness.yml @@ -0,0 +1,40 @@ +name: VibeCAD workflow harness + +on: + pull_request: + push: + branches: + - main + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: vibecad-workflow-harness-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + harness: + name: Fake-channel workflow assertions + runs-on: ubuntu-24.04 + timeout-minutes: 10 + env: + PYTHONPATH: src/Mod/VibeCAD + steps: + - name: Check out source + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 + with: + python-version: "3.11" + + - name: Install pytest + run: python -m pip install --disable-pip-version-check pytest + + - name: Run click-channel workflow tests without a TypeSafe key + run: python -m pytest -q src/Mod/VibeCAD/vibecad_tests/test_workflow_harness.py + + - name: Run the fake-channel harness entry point + run: python src/Mod/VibeCAD/VibeCADWorkflowHarness.py --fake-channel diff --git a/docs/vibecad-agent-control.md b/docs/vibecad-agent-control.md index 816c6c22..7c16480b 100644 --- a/docs/vibecad-agent-control.md +++ b/docs/vibecad-agent-control.md @@ -238,6 +238,11 @@ curl -s -H "Authorization: Bearer %VIBECAD_TOKEN%" ^ -d "{\"kind\":\"ribbon\",\"text\":\"Aero\"}" ^ http://127.0.0.1:8766/v1/ui/click +curl -s -H "Authorization: Bearer %VIBECAD_TOKEN%" ^ + -H "Content-Type: application/json" ^ + -d "{\"kind\":\"command\",\"text\":\"VibeCADRibbonNew\"}" ^ + http://127.0.0.1:8766/v1/ui/click + curl -s -H "Authorization: Bearer %VIBECAD_TOKEN%" ^ http://127.0.0.1:8766/v1/screenshot @@ -267,6 +272,7 @@ python "%VIBECAD_CLI%" close python "%VIBECAD_CLI%" ui-menus python "%VIBECAD_CLI%" ui-ribbon python "%VIBECAD_CLI%" ui-click --kind ribbon --text Aero +python "%VIBECAD_CLI%" ui-click --kind command --text VibeCADRibbonNew python "%VIBECAD_CLI%" screenshot --path C:\Evidence\vibecad.png python "%VIBECAD_CLI%" run --path C:\Models\part.FCStd --script C:\Work\edit.py python "%VIBECAD_CLI%" run --python "result = [obj.Name for obj in App.ActiveDocument.Objects]" @@ -321,7 +327,7 @@ owner's network controls. | POST | `/v1/close` | optional `document`, explicit `discard_unsaved` | Close without silently discarding a modified document | | GET | `/v1/ui/menus` | | Live top-level menu names, indices, visibility, and screen geometry | | GET | `/v1/ui/ribbon` | | Live ribbon names, workbenches, indices, selection, and screen geometry | -| POST | `/v1/ui/click` | `{"kind":"menu|ribbon","text":"..."}`, optional exact PID/index | Activate one semantic Qt target without moving or clicking the OS cursor | +| POST | `/v1/ui/click` | `{"kind":"menu|ribbon|command","text":"..."}`, optional exact PID/index | Activate one semantic Qt target without moving or clicking the OS cursor | | POST | `/v1/run` | `{"python":"..."}` or `{"script":"..."}` plus optional `path`, `recompute` | Exec against the active doc | | GET | `/v1/operations/{operation_id}` | | Read the in-memory state/result of a client-identified operation without entering the document thread | | GET/POST | `/v1/aero` | operation payload for POST | Bounded Aero context and operations | @@ -409,6 +415,14 @@ window's visible, enabled top-level menus and enabled ribbon tabs. This keeps th tester reusable across checkouts without assuming that an optional product feature is installed. +`kind=command` is additive: it activates one named ribbon button or menu +action (`VibeCADRibbonNew`, `Sketcher_NewSketch`, `PartDesign_Pad`, +`Std_Export`, or the matching command id / tooltip) through the same +in-process click site. The workflow harness in +[vibecad-workflow-harness.md](vibecad-workflow-harness.md) uses that channel +for New document, Sketch then pad, and Export. The visible tour stays the +demo; it does not become the CI suite. + ### Visible-window screenshots `GET /v1/screenshot` captures the visible VibeCAD main window under the private diff --git a/docs/vibecad-workflow-harness.md b/docs/vibecad-workflow-harness.md new file mode 100644 index 00000000..128d54af --- /dev/null +++ b/docs/vibecad-workflow-harness.md @@ -0,0 +1,48 @@ +# VibeCAD click-channel workflow harness + +`src/Mod/VibeCAD/VibeCADWorkflowHarness.py` drives three operator workflows +through the existing authenticated loopback click channel +(`POST /v1/ui/click` / `VibeCADAgentCli ui-click`). It does not move the OS +cursor and it does not replace +[`Invoke-VibeCAD-VisibleTour.ps1`](../Invoke-VibeCAD-VisibleTour.ps1). The +tour remains the human-watchable demo. + +## Workflows + +| Workflow | Clicks | Post-click checks | +| --- | --- | --- | +| `new_document` | `command` `VibeCADRibbonNew` (`Std_New`) | Semantic click verified; document count +1; active document present | +| `sketch_then_pad` | New document, `ribbon` `Model`, `Sketcher_NewSketch`, `PartDesign_Pad` | Model tab selected; sketch and pad objects present | +| `export` | New document, `command` `Std_Export` | Semantic click verified; active document still present | + +Each step records the click payload plus document/object state. Code owns +timeout, click success, and pass/fail. + +## Run + +Live GUI (same agent home / token as the visible tour): + +```sh +python src/Mod/VibeCAD/VibeCADWorkflowHarness.py --gui-only +``` + +Headless / CI (fake click channel, no display, no TypeSafe key): + +```sh +python src/Mod/VibeCAD/VibeCADWorkflowHarness.py --fake-channel +python -m pytest -q src/Mod/VibeCAD/vibecad_tests/test_workflow_harness.py +``` + +Set `PYTHONPATH=src/Mod/VibeCAD` when running from the repository root. + +## Optional Jev classification + +After a step, one System One request may classify: + +- `landed` (Noul): did the step actually complete +- `failure_class` (Choice): `geometry_bug`, `wrong_button`, `model_lied`, `flake` +- `progress` (Score): how far the workflow got + +This path is default-off. A live call happens only when both +`VIBECAD_JEV_JUDGE=1` and `TYPESAFE_API_KEY` are set. Low confidence never +changes a green result. CI mocks or skips the judge. Do not commit a key. diff --git a/src/Mod/VibeCAD/CMakeLists.txt b/src/Mod/VibeCAD/CMakeLists.txt index a3149910..f06ac1d5 100644 --- a/src/Mod/VibeCAD/CMakeLists.txt +++ b/src/Mod/VibeCAD/CMakeLists.txt @@ -45,6 +45,8 @@ set(VibeCAD_Scripts VibeCADGeometryInspection.py VibeCADPlacementInspection.py VibeCADGui.py + VibeCADJevJudge.py + VibeCADWorkflowHarness.py VibeCADTokenUsage.py VibeCADLegacyArchitecture.py VibeCADLiveAcceptanceOracle.py @@ -1626,6 +1628,7 @@ if(BUILD_TEST AND BUILD_GUI) vibecad_tests/test_agent_control_grok_bot.py vibecad_tests/test_dev_launcher_contract.py vibecad_tests/test_visible_operator_contract.py + vibecad_tests/test_workflow_harness.py vibecad_tests/test_reference_image_downscale.py vibecad_tests/test_ollama_inspect.py vibecad_tests/test_live_acceptance_oracle.py diff --git a/src/Mod/VibeCAD/VibeCADAgentCli.py b/src/Mod/VibeCAD/VibeCADAgentCli.py index 926e97cd..4786d77a 100644 --- a/src/Mod/VibeCAD/VibeCADAgentCli.py +++ b/src/Mod/VibeCAD/VibeCADAgentCli.py @@ -119,7 +119,7 @@ def build_parser() -> argparse.ArgumentParser: ui_click_parser.add_argument( "--kind", required=True, - choices=("ribbon", "menu"), + choices=("ribbon", "menu", "command"), help="Target family to activate.", ) ui_click_parser.add_argument("--text", required=True, help="Exact visible target text.") diff --git a/src/Mod/VibeCAD/VibeCADAgentControl.py b/src/Mod/VibeCAD/VibeCADAgentControl.py index e3ad58cd..8c96fd57 100644 --- a/src/Mod/VibeCAD/VibeCADAgentControl.py +++ b/src/Mod/VibeCAD/VibeCADAgentControl.py @@ -1475,7 +1475,7 @@ def brief_path() -> Path: | POST | `/v1/close` | optional `{{"document":"Name","discard_unsaved":false}}` | Close without silently discarding changes | | GET | `/v1/ui/ribbon` | | Live semantic tab names and screen geometry | | GET | `/v1/ui/menus` | | Live top-level menu names and screen geometry | -| POST | `/v1/ui/click` | `{{"kind":"ribbon","text":"Model"}}` | Activate a semantic Qt target without moving the physical cursor | +| POST | `/v1/ui/click` | `{{"kind":"ribbon","text":"Model"}}` or `{{"kind":"command","text":"Std_New"}}` | Activate a semantic Qt target without moving the physical cursor | | GET/POST | `/v1/screenshot` | optional `{{"path":"...png","overwrite":false}}` | Capture the visible VibeCAD window | | POST | `/v1/run` | `{{"python":"..."}}` or `{{"script":"..."}}` (+ optional `path`, `recompute`) | Run against the active document | | GET | `/v1/operations/` | | Prove completion after a client timeout | @@ -2434,6 +2434,83 @@ def _cursor_coordinates(QtGui: Any) -> dict[str, int]: return {"x": int(point.x()), "y": int(point.y())} +def _normalized_ui_text(value: Any) -> str: + return str(value or "").replace("&", "").strip() + + +def _command_identity(target: Any) -> set[str]: + """Collect visible names and command ids for one Qt button or action.""" + + names: set[str] = set() + if target is None: + return names + for attribute in ("objectName", "toolTip", "accessibleName", "text"): + reader = getattr(target, attribute, None) + if callable(reader): + text = _normalized_ui_text(reader()) + if text: + names.add(text) + property_reader = getattr(target, "property", None) + if callable(property_reader): + command_id = _normalized_ui_text(property_reader("VibeCADCommandId")) + if command_id: + names.add(command_id) + default_action = getattr(target, "defaultAction", None) + if callable(default_action): + names.update(_command_identity(default_action())) + return names + + +def _iter_menu_actions(root: Any) -> list[Any]: + found: list[Any] = [] + pending = [root] + seen: set[int] = set() + while pending: + current = pending.pop() + marker = id(current) + if marker in seen or current is None: + continue + seen.add(marker) + actions_reader = getattr(current, "actions", None) + if not callable(actions_reader): + continue + try: + actions = list(actions_reader() or []) + except Exception: + continue + for action in actions: + if action is None: + continue + found.append(action) + nested = getattr(action, "menu", None) + if callable(nested): + pending.append(nested()) + return found + + +def _collect_command_candidates(main_window: Any, QtWidgets: Any) -> list[tuple[str, Any, Any]]: + """Return unique (kind, target, identity) tuples for live command widgets.""" + + candidates: list[tuple[str, Any, Any]] = [] + finder = getattr(main_window, "findChildren", None) + button_type = getattr(QtWidgets, "QToolButton", None) + if callable(finder) and button_type is not None: + try: + buttons = list(finder(button_type) or []) + except Exception: + buttons = [] + for button in buttons: + names = _command_identity(button) + if names: + candidates.append(("button", button, names)) + menu_bar = main_window.menuBar() if main_window is not None else None + for action in _iter_menu_actions(menu_bar): + names = _command_identity(action) + if names: + candidates.append(("action", action, names)) + return candidates + + def ui_click_target( kind: str, text: str, @@ -2446,10 +2523,12 @@ def ui_click_target( target_kind = str(kind or "").strip().lower().replace("-", "_") if target_kind in {"tab", "ribbon_tab"}: target_kind = "ribbon" - if target_kind not in {"ribbon", "menu"}: + if target_kind in {"action", "button"}: + target_kind = "command" + if target_kind not in {"ribbon", "menu", "command"}: return failure( "UI_TARGET_KIND_INVALID", - "kind must be 'ribbon' or 'menu'.", + "kind must be 'ribbon', 'menu', or 'command'.", stage="schema", ) target_text = str(text or "").strip() @@ -2585,6 +2664,10 @@ def interaction_state() -> dict[str, bool]: left_button = QtCore.Qt.LeftButton no_modifier = QtCore.Qt.NoModifier + def inject_mouse_click(widget: Any, click_point: Any) -> None: + # One in-process Qt click site for ribbon tabs and command buttons. + QtTest.QTest.mouseClick(widget, left_button, no_modifier, click_point) + if target_kind == "ribbon": menu_bar = main_window.menuBar() active_action_before = None @@ -2637,7 +2720,7 @@ def interaction_state() -> dict[str, bool]: widget.tabText(int(widget.currentIndex())) or "" ).replace("&", "").strip() click_point = widget.tabRect(target_index).center() - QtTest.QTest.mouseClick(widget, left_button, no_modifier, click_point) + inject_mouse_click(widget, click_point) process_events() if active_action_observed: current_active_action = menu_bar.activeAction() @@ -2669,6 +2752,94 @@ def interaction_state() -> dict[str, bool]: "click_queued": False, **state, } + elif target_kind == "command": + matches = [ + (source_kind, target, names) + for source_kind, target, names in _collect_command_candidates( + main_window, QtWidgets + ) + if target_text in names + ] + unique: list[tuple[str, Any, Any]] = [] + seen_targets: set[int] = set() + for source_kind, target, names in matches: + marker = id(target) + if marker in seen_targets: + continue + seen_targets.add(marker) + unique.append((source_kind, target, names)) + if len(unique) != 1: + return failure( + "UI_TARGET_NOT_UNIQUE", + ( + f"Expected exactly one command target named {target_text!r}; " + f"found {len(unique)}." + ), + stage="precondition", + ) + source_kind, target, names = unique[0] + if required_index is not None and required_index != 0: + return failure( + "UI_TARGET_INDEX_MISMATCH", + f"Command {target_text!r} is index 0, not {required_index}.", + stage="precondition", + ) + enabled_reader = getattr(target, "isEnabled", None) + visible_reader = getattr(target, "isVisible", None) + if (callable(enabled_reader) and not bool(enabled_reader())) or ( + callable(visible_reader) and not bool(visible_reader()) + ): + return failure( + "UI_TARGET_DISABLED", + f"Command {target_text!r} is disabled or hidden.", + stage="precondition", + ) + command_id = next( + ( + name + for name in sorted(names) + if name.startswith(("Std_", "Sketcher_", "PartDesign_")) + ), + target_text, + ) + used_mouse_click = False + if source_kind == "button": + rect_reader = getattr(target, "rect", None) + if not callable(rect_reader): + return failure( + "UI_TARGET_GEOMETRY_UNAVAILABLE", + f"Command button {target_text!r} has no clickable rectangle.", + stage="precondition", + ) + click_point = rect_reader().center() + inject_mouse_click(target, click_point) + used_mouse_click = True + else: + trigger = getattr(target, "trigger", None) + if not callable(trigger): + return failure( + "UI_TARGET_HAS_NO_ACTION", + f"Command {target_text!r} cannot be triggered in-process.", + stage="precondition", + ) + trigger() + process_events() + state = interaction_state() + verified = bool(state["interaction_restored"]) + details = { + "target_kind": target_kind, + "target_text": target_text, + "target_index": 0, + "command_id": command_id, + "command_source": source_kind, + "click_queued": False, + **state, + } + details["input_method"] = ( + "qt_in_process_mouse_click" + if used_mouse_click + else "qt_in_process_action_trigger" + ) else: widget = main_window.menuBar() if widget is None or not bool(widget.isVisible()): @@ -2780,13 +2951,15 @@ def interaction_state() -> dict[str, bool]: } cursor_after = _cursor_coordinates(QtGui) + if target_kind == "ribbon": + input_method = "qt_in_process_mouse_click" + elif target_kind == "command": + input_method = str(details.get("input_method") or "qt_in_process_mouse_click") + else: + input_method = "qt_in_process_menu_popup" details.update( { - "input_method": ( - "qt_in_process_mouse_click" - if target_kind == "ribbon" - else "qt_in_process_menu_popup" - ), + "input_method": input_method, "physical_cursor_control": "none", "physical_cursor_before": cursor_before, "physical_cursor_after": cursor_after, diff --git a/src/Mod/VibeCAD/VibeCADJevJudge.py b/src/Mod/VibeCAD/VibeCADJevJudge.py new file mode 100644 index 00000000..8436e03d --- /dev/null +++ b/src/Mod/VibeCAD/VibeCADJevJudge.py @@ -0,0 +1,176 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later +"""Optional post-step Jev classification for the workflow harness. + +Jev is a judge, not a clicker. Code owns click, timeout, and pass/fail. +This helper is default-off. A live System One call happens only when both +``VIBECAD_JEV_JUDGE`` is enabled and ``TYPESAFE_API_KEY`` is set. Low +confidence never changes a green result. CI should mock or skip this path. +""" + +from __future__ import annotations + +import json +import os +from typing import Any +from urllib import error, request + + +TYPESAFE_ENDPOINT = "https://api.typesafe.ai/v1/systemone" +TYPESAFE_MODEL = "jev-latest" +API_KEY_ENV = "TYPESAFE_API_KEY" +JUDGE_FLAG_ENV = "VIBECAD_JEV_JUDGE" +LOW_CONFIDENCE = 0.5 +FAILURE_CLASSES = ( + "geometry_bug", + "wrong_button", + "model_lied", + "flake", +) +ENABLED_FLAGS = frozenset({"1", "true", "yes", "on"}) + + +def judge_enabled(environ: dict[str, str] | None = None) -> bool: + """Return True only when the operator opted in and a key is present.""" + + env = os.environ if environ is None else environ + flag = str(env.get(JUDGE_FLAG_ENV) or "").strip().lower() + if flag not in ENABLED_FLAGS: + return False + return bool(str(env.get(API_KEY_ENV) or "").strip()) + + +def _system_one_payload(state: dict[str, Any]) -> dict[str, Any]: + return { + "state": state, + "model": TYPESAFE_MODEL, + "questions": { + "landed": { + "type": "noul", + "instructions": ( + "Did this VibeCAD workflow step actually complete against " + "the post-click UI and document evidence?" + ), + }, + "failure_class": { + "type": "choice", + "instructions": ( + "If the step failed, which class best describes the failure? " + "Use flake when the evidence is inconsistent or timing-related." + ), + "criteria": { + "geometry_bug": "The command ran but the resulting geometry is wrong.", + "wrong_button": "The click hit the wrong control or command.", + "model_lied": "The UI or document reported success that the evidence contradicts.", + "flake": "The failure looks intermittent or environment-related.", + }, + }, + "progress": { + "type": "score", + "instructions": "How far did this workflow get after the step?", + "criteria": [ + "The step did not start.", + "The click happened but the document or UI did not change.", + "The step completed.", + ], + }, + }, + } + + +def _ask_typesafe(state: dict[str, Any], api_key: str) -> dict[str, Any]: + raw = json.dumps(_system_one_payload(state)).encode("utf-8") + http_request = request.Request( + TYPESAFE_ENDPOINT, + data=raw, + headers={ + "Authorization": f"Bearer {api_key}", + "Accept": "application/json", + "Content-Type": "application/json", + }, + method="POST", + ) + try: + response = request.urlopen(http_request, timeout=20) + try: + payload = json.loads(response.read().decode("utf-8")) + finally: + response.close() + except error.HTTPError as exc: + try: + detail = exc.read().decode("utf-8", errors="replace") + finally: + exc.close() + raise RuntimeError(f"TypeSafe HTTP {exc.code}: {detail}") from exc + if not isinstance(payload, dict): + raise RuntimeError("TypeSafe returned a non-object payload.") + return payload + + +def _answers(payload: dict[str, Any]) -> dict[str, Any]: + answers = payload.get("answers") + return answers if isinstance(answers, dict) else {} + + +def classify_step( + state: dict[str, Any], + *, + asker: Any = None, + environ: dict[str, str] | None = None, +) -> dict[str, Any] | None: + """Return one optional classification, or None when the judge is skipped.""" + + env = os.environ if environ is None else environ + if asker is None and not judge_enabled(env): + return { + "skipped": True, + "reason": "jev_disabled_or_unkeyed", + "used_for_pass_fail": False, + } + + api_key = str(env.get(API_KEY_ENV) or "").strip() + try: + payload = asker(state) if asker is not None else _ask_typesafe(state, api_key) + except Exception as exc: + return { + "skipped": True, + "reason": "jev_request_failed", + "error": str(exc), + "used_for_pass_fail": False, + } + + answers = _answers(payload) + landed = answers.get("landed") if isinstance(answers.get("landed"), dict) else {} + failure = ( + answers.get("failure_class") + if isinstance(answers.get("failure_class"), dict) + else {} + ) + progress = answers.get("progress") if isinstance(answers.get("progress"), dict) else {} + failure_confidence = failure.get("confidence") + progress_confidence = progress.get("confidence") + try: + lowest_confidence = min( + float(value) + for value in (failure_confidence, progress_confidence) + if value is not None + ) + except ValueError: + lowest_confidence = None + low_confidence = lowest_confidence is None or lowest_confidence < LOW_CONFIDENCE + return { + "skipped": False, + "landed": landed.get("noul"), + "failure_class": failure.get("choice"), + "progress": progress.get("score"), + "confidence": lowest_confidence, + "low_confidence": low_confidence, + "used_for_pass_fail": False, + "answers": answers, + } + + +def cannot_flip_green(code_passed: bool, judge: dict[str, Any] | None) -> bool: + """Code owns pass/fail. Jev never turns a passing step red.""" + + del judge + return bool(code_passed) diff --git a/src/Mod/VibeCAD/VibeCADWorkflowHarness.py b/src/Mod/VibeCAD/VibeCADWorkflowHarness.py new file mode 100644 index 00000000..be6f27c6 --- /dev/null +++ b/src/Mod/VibeCAD/VibeCADWorkflowHarness.py @@ -0,0 +1,482 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later +"""Click-channel workflows for New document, Sketch then pad, and Export. + +This is a CI/operator harness, not the Windows visible tour. Clicks go through +the same authenticated ``/v1/ui/click`` channel used by +``Invoke-VibeCAD-VisibleTour.ps1``. The harness never moves the OS cursor. +``Invoke-VibeCAD-VisibleTour.ps1`` remains the human-watchable demo. + +Optional Jev classification is default-off and never overrides code pass/fail. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from dataclasses import dataclass, field +from typing import Any, Protocol + +import VibeCADAgentCli as cli +import VibeCADJevJudge as jev + + +WORKFLOW_NAMES = ("new_document", "sketch_then_pad", "export") +INSPECT_OBJECTS_PYTHON = """ +doc = App.ActiveDocument +result = [] +if doc is not None: + result = [ + { + "name": str(getattr(obj, "Name", "") or ""), + "label": str(getattr(obj, "Label", "") or ""), + "type_id": str(getattr(obj, "TypeId", "") or ""), + } + for obj in list(getattr(doc, "Objects", []) or []) + ] +""" + + +class ClickChannel(Protocol): + def click(self, kind: str, text: str) -> dict[str, Any]: + """Activate one semantic target through the existing click channel.""" + + def documents(self) -> dict[str, Any]: + """Return the current document listing.""" + + def inspect_objects(self) -> list[dict[str, Any]]: + """Return active-document objects for post-click assertions.""" + + +@dataclass(frozen=True) +class WorkflowStep: + name: str + kind: str + text: str + document_count_delta: int | None = None + selected_after: str | None = None + require_type_ids: tuple[str, ...] = () + require_active_document: bool = False + require_command_id: str | None = None + + +NEW_DOCUMENT_STEP = WorkflowStep( + name="new_document", + kind="command", + text="VibeCADRibbonNew", + document_count_delta=1, + require_active_document=True, + require_command_id="Std_New", +) + +WORKFLOWS: dict[str, tuple[WorkflowStep, ...]] = { + "new_document": (NEW_DOCUMENT_STEP,), + "sketch_then_pad": ( + NEW_DOCUMENT_STEP, + WorkflowStep( + name="select_model_ribbon", + kind="ribbon", + text="Model", + selected_after="Model", + ), + WorkflowStep( + name="create_sketch", + kind="command", + text="Sketcher_NewSketch", + require_type_ids=("Sketcher::SketchObject",), + require_command_id="Sketcher_NewSketch", + ), + WorkflowStep( + name="create_pad", + kind="command", + text="PartDesign_Pad", + require_type_ids=("Sketcher::SketchObject", "PartDesign::Pad"), + require_command_id="PartDesign_Pad", + ), + ), + "export": ( + NEW_DOCUMENT_STEP, + WorkflowStep( + name="export", + kind="command", + text="Std_Export", + require_active_document=True, + require_command_id="Std_Export", + ), + ), +} + + +class AgentClickChannel: + """Live or in-process adapter over ``VibeCADAgentCli`` / ``/v1/ui/click``.""" + + def __init__(self, *, timeout_seconds: float = 30.0, gui_only: bool = False) -> None: + self.timeout_seconds = timeout_seconds + self.gui_only = gui_only + + def _call(self, command: str, arguments: dict[str, Any] | None = None) -> dict[str, Any]: + args = argparse.Namespace( + command=command, + local=False, + gui_only=self.gui_only, + timeout=self.timeout_seconds, + kind=None, + text=None, + expected_process_id=None, + expected_index=None, + path=None, + document=None, + overwrite=False, + discard_unsaved=False, + script=None, + python=None, + no_recompute=False, + ) + for key, value in dict(arguments or {}).items(): + setattr(args, key, value) + payload = cli.execute(args) + if not isinstance(payload, dict): + return { + "ok": False, + "failure_code": "CHANNEL_INVALID_RESPONSE", + "error": "The click channel did not return a JSON object.", + } + return payload + + def click(self, kind: str, text: str) -> dict[str, Any]: + return self._call("ui-click", {"kind": kind, "text": text}) + + def documents(self) -> dict[str, Any]: + return self._call("documents") + + def inspect_objects(self) -> list[dict[str, Any]]: + payload = self._call("run", {"python": INSPECT_OBJECTS_PYTHON, "no_recompute": True}) + result = payload.get("result") + if isinstance(result, list): + return [item for item in result if isinstance(item, dict)] + return [] + + +class FakeClickChannel: + """Headless stand-in that uses the same click/document/object shapes.""" + + def __init__(self) -> None: + self.documents_list: list[dict[str, Any]] = [] + self.objects: list[dict[str, Any]] = [] + self.selected_ribbon = "Start" + self.export_activated = False + self.clicks: list[tuple[str, str]] = [] + + def click(self, kind: str, text: str) -> dict[str, Any]: + self.clicks.append((kind, text)) + if kind == "ribbon": + self.selected_ribbon = text + return { + "ok": True, + "target_kind": "ribbon", + "target_text": text, + "selected_after": text, + "semantic_verified": True, + "input_method": "qt_in_process_mouse_click", + "physical_cursor_control": "none", + } + if text in {"VibeCADRibbonNew", "Std_New", "New document"}: + name = f"Unnamed{len(self.documents_list)}" + self.documents_list.append( + { + "document": name, + "label": name, + "path": "", + "active": True, + "object_count": len(self.objects), + "modified": True, + } + ) + for item in self.documents_list[:-1]: + item["active"] = False + return { + "ok": True, + "target_kind": "command", + "target_text": text, + "command_id": "Std_New", + "semantic_verified": True, + "input_method": "qt_in_process_mouse_click", + "physical_cursor_control": "none", + } + if text == "Sketcher_NewSketch": + self.objects.append( + { + "name": "Sketch", + "label": "Sketch", + "type_id": "Sketcher::SketchObject", + } + ) + return { + "ok": True, + "target_kind": "command", + "target_text": text, + "command_id": "Sketcher_NewSketch", + "semantic_verified": True, + "input_method": "qt_in_process_mouse_click", + "physical_cursor_control": "none", + } + if text == "PartDesign_Pad": + self.objects.append( + { + "name": "Pad", + "label": "Pad", + "type_id": "PartDesign::Pad", + } + ) + return { + "ok": True, + "target_kind": "command", + "target_text": text, + "command_id": "PartDesign_Pad", + "semantic_verified": True, + "input_method": "qt_in_process_mouse_click", + "physical_cursor_control": "none", + } + if text == "Std_Export": + self.export_activated = True + return { + "ok": True, + "target_kind": "command", + "target_text": text, + "command_id": "Std_Export", + "semantic_verified": True, + "input_method": "qt_in_process_action_trigger", + "physical_cursor_control": "none", + } + return { + "ok": False, + "failure_code": "UI_TARGET_NOT_UNIQUE", + "error": f"Fake channel has no target {text!r}.", + "semantic_verified": False, + } + + def documents(self) -> dict[str, Any]: + return { + "ok": True, + "document_count": len(self.documents_list), + "documents": list(self.documents_list), + } + + def inspect_objects(self) -> list[dict[str, Any]]: + return list(self.objects) + + +@dataclass +class StepResult: + name: str + passed: bool + click: dict[str, Any] + documents_before: dict[str, Any] + documents_after: dict[str, Any] + objects: list[dict[str, Any]] + errors: list[str] = field(default_factory=list) + judge: dict[str, Any] | None = None + + +def _type_ids(objects: list[dict[str, Any]]) -> set[str]: + return { + str(item.get("type_id") or "") + for item in objects + if str(item.get("type_id") or "") + } + + +def _active_documents(payload: dict[str, Any]) -> list[dict[str, Any]]: + documents = payload.get("documents") + if not isinstance(documents, list): + return [] + return [item for item in documents if isinstance(item, dict) and item.get("active")] + + +def assert_step_state( + step: WorkflowStep, + *, + click: dict[str, Any], + documents_before: dict[str, Any], + documents_after: dict[str, Any], + objects: list[dict[str, Any]], +) -> list[str]: + """Reuse the tour's semantic-target + post-click checks for one step.""" + + errors: list[str] = [] + if click.get("ok") is not True: + errors.append( + f"{step.name}: click channel failed: " + f"{click.get('failure_code') or click.get('error') or click}" + ) + return errors + if click.get("semantic_verified") is not True: + errors.append(f"{step.name}: click was not semantically verified.") + if click.get("physical_cursor_control") not in {None, "none"}: + errors.append(f"{step.name}: click channel reported physical cursor control.") + if step.selected_after and click.get("selected_after") != step.selected_after: + errors.append( + f"{step.name}: selected_after was {click.get('selected_after')!r}, " + f"expected {step.selected_after!r}." + ) + if step.require_command_id: + command_id = str(click.get("command_id") or click.get("target_text") or "") + if command_id != step.require_command_id and click.get("target_text") != step.text: + errors.append( + f"{step.name}: command_id was {command_id!r}, " + f"expected {step.require_command_id!r}." + ) + if step.document_count_delta is not None: + before = int(documents_before.get("document_count") or 0) + after = int(documents_after.get("document_count") or 0) + if after - before != step.document_count_delta: + errors.append( + f"{step.name}: document_count delta was {after - before}, " + f"expected {step.document_count_delta}." + ) + if step.require_active_document and not _active_documents(documents_after): + errors.append(f"{step.name}: no active document after the click.") + missing = [type_id for type_id in step.require_type_ids if type_id not in _type_ids(objects)] + if missing: + errors.append(f"{step.name}: missing document objects {missing!r}.") + return errors + + +def run_workflow( + name: str, + channel: ClickChannel, + *, + judge_asker: Any = None, + judge_environ: dict[str, str] | None = None, +) -> dict[str, Any]: + if name not in WORKFLOWS: + raise ValueError(f"Unknown workflow {name!r}; expected one of {list(WORKFLOWS)}.") + + steps: list[StepResult] = [] + for step in WORKFLOWS[name]: + documents_before = channel.documents() + click = channel.click(step.kind, step.text) + documents_after = channel.documents() + objects = channel.inspect_objects() + errors = assert_step_state( + step, + click=click, + documents_before=documents_before, + documents_after=documents_after, + objects=objects, + ) + code_passed = not errors + judge = jev.classify_step( + { + "workflow": name, + "step": step.name, + "kind": step.kind, + "text": step.text, + "click": click, + "documents_before": documents_before, + "documents_after": documents_after, + "objects": objects, + "code_passed": code_passed, + "errors": errors, + }, + asker=judge_asker, + environ=judge_environ, + ) + passed = jev.cannot_flip_green(code_passed, judge) + steps.append( + StepResult( + name=step.name, + passed=passed, + click=click, + documents_before=documents_before, + documents_after=documents_after, + objects=objects, + errors=errors, + judge=judge, + ) + ) + if not passed: + break + + return { + "ok": all(step.passed for step in steps) and len(steps) == len(WORKFLOWS[name]), + "workflow": name, + "channel": "vibecad-agent-control", + "demo": "Invoke-VibeCAD-VisibleTour.ps1 remains the visible demo tour.", + "completed_steps": sum(1 for step in steps if step.passed), + "step_count": len(WORKFLOWS[name]), + "steps": [ + { + "name": step.name, + "ok": step.passed, + "errors": step.errors, + "click": step.click, + "document_count_before": step.documents_before.get("document_count"), + "document_count_after": step.documents_after.get("document_count"), + "objects": step.objects, + "judge": step.judge, + } + for step in steps + ], + } + + +def run_all_workflows( + channel: ClickChannel, + *, + judge_asker: Any = None, + judge_environ: dict[str, str] | None = None, +) -> dict[str, Any]: + reports = [ + run_workflow( + name, + channel, + judge_asker=judge_asker, + judge_environ=judge_environ, + ) + for name in WORKFLOW_NAMES + ] + return { + "ok": all(report["ok"] for report in reports), + "workflows": reports, + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--workflow", + choices=WORKFLOW_NAMES, + help="Run one workflow. Default: all three.", + ) + parser.add_argument( + "--fake-channel", + action="store_true", + help="Use the in-process fake click channel (CI / no display).", + ) + parser.add_argument( + "--gui-only", + action="store_true", + help="Fail if the live GUI loopback click channel is not listening.", + ) + parser.add_argument("--timeout", type=float, default=30.0) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + if args.fake_channel: + channel: ClickChannel = FakeClickChannel() + else: + channel = AgentClickChannel(timeout_seconds=args.timeout, gui_only=args.gui_only) + if args.workflow: + report = run_workflow(args.workflow, channel) + else: + report = run_all_workflows(channel) + json.dump(report, sys.stdout, indent=2) + sys.stdout.write("\n") + return 0 if report.get("ok") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/Mod/VibeCAD/vibecad_tests/test_visible_operator_contract.py b/src/Mod/VibeCAD/vibecad_tests/test_visible_operator_contract.py index 2eae50ea..b35d509d 100644 --- a/src/Mod/VibeCAD/vibecad_tests/test_visible_operator_contract.py +++ b/src/Mod/VibeCAD/vibecad_tests/test_visible_operator_contract.py @@ -20,7 +20,10 @@ AGENT_CONTROL = ( REPOSITORY_ROOT / "src" / "Mod" / "VibeCAD" / "VibeCADAgentControl.py" ) -PHYSICAL_INPUT_IMPLEMENTATIONS = (OPERATOR_SCRIPT, AGENT_CONTROL) +WORKFLOW_HARNESS = ( + REPOSITORY_ROOT / "src" / "Mod" / "VibeCAD" / "VibeCADWorkflowHarness.py" +) +PHYSICAL_INPUT_IMPLEMENTATIONS = (OPERATOR_SCRIPT, AGENT_CONTROL, WORKFLOW_HARNESS) PROHIBITED_PHYSICAL_INPUT_APIS = ( "SetCursorPos", "SendInput", diff --git a/src/Mod/VibeCAD/vibecad_tests/test_workflow_harness.py b/src/Mod/VibeCAD/vibecad_tests/test_workflow_harness.py new file mode 100644 index 00000000..a3551bce --- /dev/null +++ b/src/Mod/VibeCAD/vibecad_tests/test_workflow_harness.py @@ -0,0 +1,232 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later +"""Headless contracts for the click-channel workflow harness.""" + +from __future__ import annotations + +import sys +from types import SimpleNamespace +from typing import Any + +import VibeCADAgentCli as cli +import VibeCADAgentControl as control +import VibeCADJevJudge as jev +import VibeCADWorkflowHarness as harness + + +def test_cli_accepts_command_kind_on_the_existing_click_route() -> None: + parsed = cli.build_parser().parse_args( + ["ui-click", "--kind", "command", "--text", "VibeCADRibbonNew"] + ) + assert cli._http_route(parsed.command) == ("POST", "/v1/ui/click") + assert cli._command_arguments(parsed) == { + "kind": "command", + "text": "VibeCADRibbonNew", + "expected_process_id": None, + "expected_index": None, + } + + +def test_command_click_uses_the_same_in_process_mouse_injection(monkeypatch) -> None: + class Point: + def __init__(self, x: int, y: int) -> None: + self._x = x + self._y = y + + def x(self) -> int: + return self._x + + def y(self) -> int: + return self._y + + class Rect: + def center(self) -> Point: + return Point(18, 10) + + class Button: + def __init__(self) -> None: + self.clicked = False + + def objectName(self) -> str: # noqa: N802 + return "VibeCADRibbonNew" + + def toolTip(self) -> str: # noqa: N802 + return "New document" + + def accessibleName(self) -> str: # noqa: N802 + return "New document" + + def property(self, name: str): + return "Std_New" if name == "VibeCADCommandId" else None + + def isEnabled(self) -> bool: # noqa: N802 + return True + + def isVisible(self) -> bool: # noqa: N802 + return True + + def rect(self) -> Rect: + return Rect() + + def defaultAction(self): # noqa: N802 + return None + + button = Button() + window = SimpleNamespace( + findChild=lambda *_args: None, + findChildren=lambda _kind: [button], + menuBar=lambda: SimpleNamespace(actions=lambda: []), + ) + application = SimpleNamespace(focus=object(), active_window=window, popup=None) + qt_core = SimpleNamespace( + Qt=SimpleNamespace( + LeftButton="left", + NoModifier="none", + OtherFocusReason="other", + ) + ) + qt_gui = SimpleNamespace(QCursor=SimpleNamespace(pos=lambda: Point(4, 8))) + qt_widgets = SimpleNamespace( + QTabBar=object, + QToolButton=Button, + QApplication=SimpleNamespace( + processEvents=lambda: None, + focusWidget=lambda: application.focus, + activeWindow=lambda: application.active_window, + activePopupWidget=lambda: application.popup, + ), + ) + clicks: list[tuple[object, object, object, object]] = [] + + class QTest: + @staticmethod + def mouseClick(widget, button_name, modifiers, point) -> None: # noqa: N802 + clicks.append((widget, button_name, modifiers, point)) + widget.clicked = True + + monkeypatch.setitem( + sys.modules, + "PySide", + SimpleNamespace(QtCore=qt_core, QtGui=qt_gui, QtWidgets=qt_widgets), + ) + monkeypatch.setitem( + sys.modules, + "PySide6", + SimpleNamespace(QtTest=SimpleNamespace(QTest=QTest)), + ) + monkeypatch.setattr( + control, + "_gui", + lambda: SimpleNamespace(GuiUp=True, getMainWindow=lambda: window), + ) + monkeypatch.setattr(control, "_document_thread_dispatch", lambda operation: operation()) + freecad = sys.modules["FreeCAD"] + monkeypatch.setattr(freecad, "isRestoring", lambda: False, raising=False) + + payload = control.dispatch("ui_click", {"kind": "command", "text": "VibeCADRibbonNew"}) + assert payload["ok"] is True + assert payload["semantic_verified"] is True + assert payload["command_id"] == "Std_New" + assert payload["input_method"] == "qt_in_process_mouse_click" + assert payload["physical_cursor_control"] == "none" + assert len(clicks) == 1 + assert button.clicked is True + + +def test_three_workflows_pass_on_the_fake_click_channel_without_a_key( + monkeypatch, +) -> None: + monkeypatch.delenv(jev.API_KEY_ENV, raising=False) + monkeypatch.delenv(jev.JUDGE_FLAG_ENV, raising=False) + report = harness.run_all_workflows(harness.FakeClickChannel()) + assert report["ok"] is True + names = [item["workflow"] for item in report["workflows"]] + assert names == ["new_document", "sketch_then_pad", "export"] + sketch = next(item for item in report["workflows"] if item["workflow"] == "sketch_then_pad") + type_ids = { + obj["type_id"] + for step in sketch["steps"] + for obj in step["objects"] + } + assert "Sketcher::SketchObject" in type_ids + assert "PartDesign::Pad" in type_ids + for workflow in report["workflows"]: + for step in workflow["steps"]: + assert step["judge"]["skipped"] is True + assert step["judge"]["used_for_pass_fail"] is False + + +def test_new_document_fails_when_the_channel_does_not_add_a_document() -> None: + channel = harness.FakeClickChannel() + + def silent_new(kind: str, text: str) -> dict[str, Any]: + if text in {"VibeCADRibbonNew", "Std_New", "New document"}: + return { + "ok": True, + "target_kind": "command", + "target_text": text, + "command_id": "Std_New", + "semantic_verified": True, + "physical_cursor_control": "none", + } + return harness.FakeClickChannel.click(channel, kind, text) + + channel.click = silent_new # type: ignore[method-assign] + report = harness.run_workflow("new_document", channel) + assert report["ok"] is False + assert any("document_count" in error for error in report["steps"][0]["errors"]) + + +def test_jev_is_optional_and_cannot_flip_a_green(monkeypatch) -> None: + monkeypatch.setenv(jev.JUDGE_FLAG_ENV, "1") + monkeypatch.setenv(jev.API_KEY_ENV, "test-key") + + def asker(_state): + return { + "answers": { + "landed": {"type": "noul", "noul": 0.1}, + "failure_class": { + "type": "choice", + "choice": "model_lied", + "confidence": 0.2, + }, + "progress": {"type": "score", "score": 0.0, "confidence": 0.1}, + } + } + + report = harness.run_workflow( + "new_document", + harness.FakeClickChannel(), + judge_asker=asker, + judge_environ={ + jev.JUDGE_FLAG_ENV: "1", + jev.API_KEY_ENV: "test-key", + }, + ) + assert report["ok"] is True + judge = report["steps"][0]["judge"] + assert judge["skipped"] is False + assert judge["low_confidence"] is True + assert judge["used_for_pass_fail"] is False + assert jev.cannot_flip_green(True, judge) is True + + +def test_jev_stays_skipped_in_ci_without_a_key(monkeypatch) -> None: + monkeypatch.setenv(jev.JUDGE_FLAG_ENV, "1") + monkeypatch.delenv(jev.API_KEY_ENV, raising=False) + assert jev.judge_enabled() is False + classification = jev.classify_step({"step": "new_document"}) + assert classification is not None + assert classification["skipped"] is True + assert classification["used_for_pass_fail"] is False + + +def test_harness_cli_fake_channel_exits_zero_without_typesafe(monkeypatch) -> None: + monkeypatch.delenv(jev.API_KEY_ENV, raising=False) + monkeypatch.delenv(jev.JUDGE_FLAG_ENV, raising=False) + assert harness.main(["--fake-channel"]) == 0 + + +def test_visible_tour_stays_the_demo() -> None: + source = harness.__doc__ or "" + assert "Invoke-VibeCAD-VisibleTour.ps1" in source + assert "never moves the OS cursor" in source