Improve ambiguity handling in argument processing

formatters/ai_scriptless.py

@@ -37,6 +37,13 @@ def command_selection_policy_info() -> List[str]:
         "Structural helpers (add_logical_step, add_loop, add_condition, comment, wait) are OK; "
         "keep observable steps AI-driven when possible.",
         "Call get_command_definitions only for the AI command_ids you will use.",
+        "cmd_arguments keys are the parameter names returned by get_command_definitions "
+        "(mandatory_parameters / optional_parameters); undeclared names are rejected.",
+        "Keep command arguments nested inside cmd_arguments: the 'action' parameter of ai_user-action "
+        "collides with the tool's own action key if flattened into args.",
+        "Values are constants by default; pass {\"data_source\": \"VARIABLE\", \"value\": \"<variable name>\"} "
+        "to bind an argument to a script variable.",
+        "modify_command merges: only the arguments sent are replaced, the others keep their current value.",
     ]
 
 def format_ai_scriptless_tests_filter_values(tests: dict[str, Any], params: Optional[dict] = None) -> dict[str, Any]:

main.py

@@ -5,6 +5,25 @@
 import sys
 from typing import Literal, cast
 
+# Patch MCP ArgModelBase so tools with an "arguments" param receive the full payload
+# when the client sends {"action": "x", "key": "value"} instead of {"arguments": {...}}
+from mcp.server.fastmcp.utilities import func_metadata
+from pydantic import model_validator
+
+_OriginalArgModelBase = func_metadata.ArgModelBase
+
+
+class _PatchedArgModelBase(_OriginalArgModelBase):
+    @model_validator(mode="before")
+    @classmethod
+    def _wrap_root_as_arguments(cls, data: object) -> object:
+        if isinstance(data, dict) and "arguments" not in data:
+            return {"arguments": data}
+        return data
+
+
+func_metadata.ArgModelBase = _PatchedArgModelBase
+
 from mcp.server.fastmcp import FastMCP, Icon
 
 from config.perfecto import SECURITY_TOKEN_FILE_ENV_NAME, SECURITY_TOKEN_ENV_NAME, PERFECTO_CLOUD_NAME_ENV_NAME, \

tests/conftest.py

@@ -17,8 +17,46 @@
 import pytest
 
 from config.token import PerfectoToken
+from models.result import BaseResult
+from tools.ai_scriptless import definitions
 
 
 @pytest.fixture
 def perfecto_token() -> PerfectoToken:
     return PerfectoToken("test-token", "demo")
+
+
+@pytest.fixture(autouse=True)
+def offline_command_definitions(monkeypatch):
+    """Keep cmd_arguments validation offline.
+
+    Validation resolves declared parameters over HTTP and memoizes them, so the
+    request is stubbed out (no declared parameters = validation fails open) and the
+    cache is reset on both ends. Use declare_command_parameters to opt into validation.
+    """
+    definitions.reset_declared_parameters_cache()
+
+    async def offline_api_request(*_args, **_kwargs):
+        return BaseResult(error="command definitions are not fetched in tests")
+
+    monkeypatch.setattr(definitions, "api_request", offline_api_request)
+    yield
+    definitions.reset_declared_parameters_cache()
+
+
+@pytest.fixture
+def declare_command_parameters(monkeypatch, offline_command_definitions):
+    """Declare parameters per command_id: {command_id: (mandatory, optional)}."""
+
+    def declare(declarations: dict[str, tuple[list[str], list[str]]]) -> None:
+        async def fake_fetch(_token, command_id):
+            declaration = declarations.get(command_id)
+            if declaration is None:
+                return None
+            mandatory, optional = declaration
+            return frozenset(mandatory), frozenset(optional)
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "_fetch_declared_parameters", fake_fetch)
+
+    return declare

tests/test_ai_scriptless_definitions.py

@@ -0,0 +1,151 @@
+"""
+Copyright 2025 Perforce Software, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import asyncio
+
+import httpx
+
+from models.result import BaseResult
+from tools.ai_scriptless import definitions
+from tools.ai_scriptless.definitions import (
+    declared_parameters,
+    empty_mandatory_note,
+    validate_argument_names,
+)
+
+USER_ACTION = (frozenset({"action"}), frozenset({"handsetId"}))
+
+
+class TestValidateArgumentNames:
+    def test_accepts_declared_names(self):
+        assert validate_argument_names(
+            "ai_user-action", {"action": "Tap", "handsetId": "DUT"}, USER_ACTION
+        ) is None
+
+    def test_rejects_undeclared_name_with_suggestion(self):
+        error = validate_argument_names("ai_user-action", {"actions": "Tap"}, USER_ACTION)
+        assert "'actions' (did you mean 'action'?)" in error
+        assert "Declared parameter names: action, handsetId" in error
+        assert "mandatory: action" in error
+
+    def test_reports_undeclared_name_without_close_match(self):
+        error = validate_argument_names("ai_user-action", {"xyz": "Tap"}, USER_ACTION)
+        assert "'xyz'" in error
+        assert "did you mean" not in error
+
+    def test_accepts_alias_in_either_direction(self):
+        # The spec canonicalizes waitDuration to duration; either name may be declared.
+        assert validate_argument_names("wait", {"duration": "3"}, (frozenset({"waitDuration"}), frozenset())) is None
+        assert validate_argument_names("wait", {"waitDuration": "3"}, (frozenset({"duration"}), frozenset())) is None
+
+    def test_fails_open_without_declared_parameters(self):
+        assert validate_argument_names("ai_user-action", {"anything": "value"}, None) is None
+
+    def test_accepts_variable_data_source_form(self):
+        assert validate_argument_names(
+            "ai_user-action",
+            {"action": {"data_source": "VARIABLE", "value": "loginStep"}},
+            USER_ACTION,
+        ) is None
+
+
+class TestEmptyMandatoryNote:
+    def test_notes_mandatory_left_empty_by_spec_default(self):
+        note = empty_mandatory_note("ai_user-action", None, USER_ACTION)
+        assert "Mandatory parameter(s) left empty on 'ai_user-action': action" in note
+
+    def test_no_note_when_mandatory_is_provided(self):
+        assert empty_mandatory_note("ai_user-action", {"action": "Tap"}, USER_ACTION) is None
+
+    def test_blank_string_counts_as_empty(self):
+        note = empty_mandatory_note("ai_user-action", {"action": "   "}, USER_ACTION)
+        assert "action" in note
+
+    def test_variable_binding_counts_as_provided(self):
+        assert empty_mandatory_note(
+            "ai_user-action",
+            {"action": {"data_source": "VARIABLE", "value": "loginStep"}},
+            USER_ACTION,
+        ) is None
+
+    def test_no_note_without_declared_parameters(self):
+        assert empty_mandatory_note("ai_user-action", None, None) is None
+
+
+class TestDeclaredParameters:
+    def test_parses_and_memoizes_definitions(self, perfecto_token, monkeypatch):
+        calls: list = []
+
+        async def fake_api_request(_token, _method, endpoint=None, result_formatter=None, **kwargs):
+            calls.append(kwargs.get("json"))
+            return BaseResult(result=result_formatter({
+                "definitions": [{
+                    "commandId": "ai_validation",
+                    "data": {
+                        "display": {"name": "AI Validation"},
+                        "mandatoryParameters": [{"name": "validation"}],
+                        "optionalParameters": [{"name": "handsetId"}],
+                    },
+                }],
+            }, None))
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        first = asyncio.run(declared_parameters(perfecto_token, "ai_validation"))
+        second = asyncio.run(declared_parameters(perfecto_token, "ai_validation"))
+
+        assert first == (frozenset({"validation"}), frozenset({"handsetId"}))
+        assert second == first
+        assert calls == [{"commandIds": ["ai_validation"]}]
+
+    def test_definition_without_parameters_is_treated_as_unknown(self, perfecto_token, monkeypatch):
+        async def fake_api_request(_token, _method, endpoint=None, result_formatter=None, **kwargs):
+            return BaseResult(result=result_formatter({
+                "definitions": [{
+                    "commandId": "wait",
+                    "data": {"display": {"name": "Wait"}, "mandatoryParameters": [], "optionalParameters": []},
+                }],
+            }, None))
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        assert asyncio.run(declared_parameters(perfecto_token, "wait")) is None
+
+    def test_fails_open_on_api_error(self, perfecto_token, monkeypatch):
+        async def fake_api_request(*_args, **_kwargs):
+            return BaseResult(error="Invalid credentials")
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        assert asyncio.run(declared_parameters(perfecto_token, "ai_validation")) is None
+
+    def test_fails_open_on_http_exception(self, perfecto_token, monkeypatch):
+        async def fake_api_request(*_args, **_kwargs):
+            request = httpx.Request("POST", "https://demo.perfectomobile.com/definitions")
+            raise httpx.HTTPStatusError(
+                "not found", request=request, response=httpx.Response(404, request=request)
+            )
+
+        definitions.reset_declared_parameters_cache()
+        monkeypatch.setattr(definitions, "api_request", fake_api_request)
+
+        assert asyncio.run(declared_parameters(perfecto_token, "ai_validation")) is None
+
+    def test_no_token_returns_none(self):
+        assert asyncio.run(declared_parameters(None, "ai_validation")) is None