Eduard van Valkenburg (eavanvalkenburg) · Oct 15, 2026 · Oct 7, 2026 · Oct 8, 2026 · Oct 8, 2026 · Oct 9, 2026
@@ -46,6 +46,7 @@
 from agent_framework._workflows._state import State
 
 from ._errors import DeclarativeWorkflowError
+from ._powerfx_functions import whole_call_args
 from ._powerfx_limits import _PowerFxStateBudget, _validate_powerfx_state  # pyright: ignore[reportPrivateUsage]
 from ._state_path import _is_safe_path_segment  # pyright: ignore[reportPrivateUsage]
 
@@ -721,14 +722,13 @@ def _eval_custom_function(self, formula: str) -> Any | None:
 
         Returns None if the formula is not a custom function call.
         """
-        import re
+        formula = formula.strip()
 
         # Concat/Concatenate - string concatenation
         # In standard PowerFx, Concatenate is for strings, Concat is for tables.
         # Copilot Studio uses Concat for strings, so we support both.
-        match = re.match(r"(?:Concat|Concatenate)\((.+)\)$", formula.strip())
-        if match:
-            args_str = match.group(1)
+        args_str = whole_call_args(formula, "Concat", "Concatenate")
+        if args_str is not None:
             # Parse comma-separated arguments (handling nested parentheses)
             args = self._parse_function_args(args_str)
             evaluated_args: list[str] = []
@@ -747,24 +747,24 @@ def _eval_custom_function(self, formula: str) -> Any | None:
             return "".join(evaluated_args)
 
         # UserMessage(expr) - creates a user message dict
-        match = re.match(r"UserMessage\((.+)\)$", formula.strip())
-        if match:
-            inner_expr = match.group(1).strip()
+        inner_args = whole_call_args(formula, "UserMessage")
+        if inner_args is not None:
+            inner_expr = inner_args.strip()
             # Evaluate the inner expression
             text = self.eval(f"={inner_expr}")
             return {"role": "user", "text": str(text) if text else ""}
 
         # AgentMessage(expr) - creates an assistant message dict
-        match = re.match(r"AgentMessage\((.+)\)$", formula.strip())
-        if match:
-            inner_expr = match.group(1).strip()
+        inner_args = whole_call_args(formula, "AgentMessage")
+        if inner_args is not None:
+            inner_expr = inner_args.strip()
             text = self.eval(f"={inner_expr}")
             return {"role": "assistant", "text": str(text) if text else ""}
 
         # MessageText(expr) - extracts text from the last message
-        match = re.match(r"MessageText\((.+)\)$", formula.strip())
-        if match:
-            inner_expr = match.group(1).strip()
+        inner_args = whole_call_args(formula, "MessageText")
+        if inner_args is not None:
+            inner_expr = inner_args.strip()
             # Reuse the helper method for consistent text extraction
             return self._eval_and_replace_message_text(inner_expr)
 

@@ -15,6 +15,54 @@
 from typing import Any, cast
 
 
+def whole_call_args(formula: str, *names: str) -> str | None:
+    """Return the argument text when ``formula`` is exactly one ``Name(...)`` call.
+
+    A custom-function handler must not fire for a formula that merely *starts*
+    with its name: the closing parenthesis has to be the call's own and the
+    formula's last character. Otherwise a trailing operator is swallowed into the
+    argument list and the formula is evaluated as something the author never
+    wrote, instead of being left to PowerFx.
+
+    Args:
+        formula: The stripped formula to inspect.
+        *names: Candidate function names, matched case-sensitively.
+
+    Returns:
+        The text between the call's parentheses, or None when ``formula`` is not
+        a single complete call to one of ``names``. Empty arguments yield None so
+        that an argument-less call falls through to PowerFx, which reports the
+        arity error.
+    """
+    for name in names:
+        prefix = f"{name}("
+        if not formula.startswith(prefix):
+            continue
+        depth = 0
+        in_string = False
+        string_char: str | None = None
+        for index in range(len(name), len(formula)):
+            char = formula[index]
+            if char in ('"', "'") and not in_string:
+                in_string = True
+                string_char = char
+            elif char == string_char and in_string:
+                in_string = False
+                string_char = None
+            elif char == "(" and not in_string:
+                depth += 1
+            elif char == ")" and not in_string:
+                depth -= 1
+                if depth == 0:
+                    if index != len(formula) - 1:
+                        # Something follows the call, so the formula is a larger
+                        # expression that the caller must not claim.
+                        return None
+                    return formula[len(prefix) : index] or None
+        return None
+    return None
+
+
 def message_text(messages: Any) -> str:
     """Extract text content from a message or list of messages.
 

@@ -441,7 +441,7 @@ def _eval_simple(self, formula: str) -> Any:
         Returns:
             The evaluated result
         """
-        from ._powerfx_functions import CUSTOM_FUNCTIONS
+        from ._powerfx_functions import CUSTOM_FUNCTIONS, whole_call_args
 
         formula = formula.strip()
 
@@ -452,15 +452,15 @@ def _eval_simple(self, formula: str) -> Any:
             return not bool(result)
 
         # Handle Not() function
-        if formula.startswith("Not(") and formula.endswith(")"):
-            inner = formula[4:-1].strip()
-            result = self._eval_simple(inner)
+        not_args = whole_call_args(formula, "Not")
+        if not_args is not None:
+            result = self._eval_simple(not_args.strip())
             return not bool(result)
 
         # Handle function calls
         for func_name, func in CUSTOM_FUNCTIONS.items():
-            if formula.startswith(f"{func_name}(") and formula.endswith(")"):
-                args_str = formula[len(func_name) + 1 : -1]
+            args_str = whole_call_args(formula, func_name)
+            if args_str is not None:
                 # Simple argument parsing (doesn't handle nested calls well)
                 args = self._parse_function_args(args_str)
                 evaluated_args = [self._eval_simple(arg) if isinstance(arg, str) else arg for arg in args]

@@ -4,6 +4,9 @@
 
 from typing import cast
 
+from agent_framework._workflows._state import State
+
+from agent_framework_declarative._workflows._declarative_base import DeclarativeWorkflowState
 from agent_framework_declarative._workflows._powerfx_functions import (
     CUSTOM_FUNCTIONS,
     assistant_message,
@@ -19,6 +22,7 @@
     system_message,
     upper,
     user_message,
+    whole_call_args,
 )
 
 
@@ -692,3 +696,87 @@ def test_search_table_false_is_a_value_not_a_blank(self):
         """Test search_table with a boolean False."""
         items = [{"ok": True, "label": "yes"}, {"ok": False, "label": "no"}]
         assert [item["label"] for item in search_table(items, False, "ok")] == ["no"]
+
+
+class TestWholeCallArgs:
+    """Tests for whole_call_args, which bounds a custom-function match to one call."""
+
+    def test_whole_call_returns_inner_text(self):
+        """A formula that is exactly one call yields its argument text."""
+        assert whole_call_args('Concat("a", "b")', "Concat") == '"a", "b"'
+
+    def test_trailing_operator_is_not_claimed(self):
+        """A call followed by more expression belongs to PowerFx, not the handler."""
+        assert whole_call_args('Concat("a") & Lower("B")', "Concat") is None
+        assert whole_call_args('MessageText("hi") & Upper("there")', "MessageText") is None
+
+    def test_trailing_text_after_closing_paren_is_not_claimed(self):
+        """Anything after the call's own closing paren disqualifies the match."""
+        assert whole_call_args('Concat("a") + 1', "Concat") is None
+
+    def test_name_must_start_the_formula(self):
+        """A call nested inside a larger formula is not a whole-formula match."""
+        assert whole_call_args('Lower("B") & Concat("a")', "Concat") is None
+
+    def test_nested_same_name_call_is_balanced(self):
+        """Nested parentheses are tracked, so the outer call still matches."""
+        assert whole_call_args('Concat(Concat("a", "b"), "c")', "Concat") == 'Concat("a", "b"), "c"'
+
+    def test_paren_inside_string_literal_does_not_close_the_call(self):
+        """A ``)`` inside a quoted literal is data, not the call's terminator."""
+        assert whole_call_args('Concat("a)", "b")', "Concat") == '"a)", "b"'
+        assert whole_call_args('Concat("MessageText(Local.S)")', "Concat") == '"MessageText(Local.S)"'
+
+    def test_empty_arguments_fall_through(self):
+        """An argument-less call is left to PowerFx so it reports the arity error."""
+        assert whole_call_args("Concat()", "Concat") is None
+
+    def test_unbalanced_parens_fall_through(self):
+        """An unterminated call is not matched."""
+        assert whole_call_args('Concat("a"', "Concat") is None
+
+    def test_first_matching_name_wins(self):
+        """Several candidate names may be offered in one call."""
+        assert whole_call_args('Concatenate("x")', "Concat", "Concatenate") == '"x"'
+        assert whole_call_args('Concat("x")', "Concat", "Concatenate") == '"x"'
+
+    def test_match_is_case_sensitive(self):
+        """PowerFx function names are matched exactly, as before."""
+        assert whole_call_args('concat("a")', "Concat") is None
+
+
+class TestEvalCustomFunctionExtent:
+    """A custom-function handler must only claim a formula that is one whole call.
+
+    Regression test for the end-anchored regexes that captured any formula
+    starting with a dialect function name and ending in ``)``, swallowing the
+    trailing operator into the argument list. These assertions need no PowerFx
+    engine: a handled formula returns its value, and an unhandled one returns
+    None so that eval() hands the formula to PowerFx.
+    """
+
+    def _state(self) -> DeclarativeWorkflowState:
+        state = DeclarativeWorkflowState(State())
+        state.initialize({})
+        return state
+
+    def test_whole_concat_call_is_still_handled(self):
+        """The Copilot Studio string alias keeps working for a complete call."""
+        assert self._state()._eval_custom_function('Concat("a", "b")') == "ab"
+        assert self._state()._eval_custom_function('Concatenate("a", "b")') == "ab"
+
+    def test_concat_with_trailing_operator_falls_through(self):
+        """A trailing operator must not be absorbed as a string literal."""
+        assert self._state()._eval_custom_function('Concat("a") & Lower("B")') is None
+        assert self._state()._eval_custom_function('Concatenate("x") & Upper("y")') is None
+
+    def test_message_helpers_with_trailing_operator_fall_through(self):
+        """UserMessage/AgentMessage/MessageText are bounded the same way."""
+        state = self._state()
+        assert state._eval_custom_function('UserMessage("hi") & "x"') is None
+        assert state._eval_custom_function('AgentMessage("hi") & "x"') is None
+        assert state._eval_custom_function('MessageText("hi") & Upper("x")') is None
+
+    def test_quoted_paren_inside_a_whole_call_is_preserved(self):
+        """A literal containing ``)`` does not truncate the call."""
+        assert self._state()._eval_custom_function('Concat("a)", "b")') == "a)b"
@@ -440,6 +440,45 @@ def test_workflow_inputs_reference(self):
         assert state._eval_simple("Workflow.Inputs.name") == "test"
 
 
+class TestWorkflowStateEvalSimpleCallExtent:
+    """The fallback dispatcher must only claim a formula that is one whole call.
+
+    ``_eval_simple`` previously matched ``formula.startswith(f"{name}(")`` with
+    ``formula.endswith(")")``, so a call followed by any operator had the
+    remainder of the formula absorbed into its arguments.
+    """
+
+    def test_trailing_operator_is_not_absorbed_into_arguments(self):
+        """The formula is left unclaimed instead of yielding a corrupted fragment.
+
+        ``_eval_simple`` echoes a formula it cannot evaluate, and ``eval``
+        reaches it only after PowerFx has already had its turn.
+        """
+        state = WorkflowState()
+        formula = 'Concat("a") & Lower("B")'
+        assert state._eval_simple(formula) == formula
+
+    def test_not_with_trailing_operator_is_not_absorbed(self):
+        """The Not() special case is bounded the same way.
+
+        Previously the handler claimed this formula and returned a bool built
+        from the mis-parsed fragment ``false) & Upper("x"``.
+        """
+        state = WorkflowState()
+        formula = 'Not(false) & Upper("x")'
+        assert state._eval_simple(formula) == formula
+
+    def test_whole_not_call_still_dispatches(self):
+        """A complete Not() call keeps negating its argument."""
+        assert WorkflowState()._eval_simple("Not(false)") is True
+
+    def test_whole_call_still_dispatches(self):
+        """A complete call keeps resolving through CUSTOM_FUNCTIONS."""
+        state = WorkflowState()
+        assert state._eval_simple('Concat("a", "b")') == "ab"
+        assert state._eval_simple('Upper("ab")') == "AB"
+
+
 class TestWorkflowStateParseFunctionArgs:
     """Tests for _parse_function_args helper."""