diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py b/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py index 25c1c3684fc..7338aa62260 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py @@ -46,6 +46,10 @@ from agent_framework._workflows._state import State from ._errors import DeclarativeWorkflowError +from ._powerfx_functions import ( + _iter_powerfx_expression_indices, # pyright: ignore[reportPrivateUsage] + 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] @@ -68,73 +72,6 @@ _ENV_REFERENCE_RE = re.compile(r"\bEnv\.([A-Za-z_][A-Za-z0-9_]*)") -def _skip_powerfx_opaque_token(formula: str, start: int) -> int: - """Return the end of an ordinary quoted token or comment, or start if neither.""" - quote = formula[start] - if quote in ('"', "'"): - pos = start + 1 - while pos < len(formula): - if formula[pos] == quote: - if pos + 1 < len(formula) and formula[pos + 1] == quote: - pos += 2 - continue - return pos + 1 - pos += 1 - return pos - if formula.startswith("//", start): - pos = start + 2 - while pos < len(formula) and formula[pos] not in "\r\n": - pos += 1 - return pos - if formula.startswith("/*", start): - end = formula.find("*/", start + 2) - return len(formula) if end == -1 else end + 2 - return start - - -def _iter_powerfx_expression_indices(formula: str) -> Iterator[int]: - """Yield code positions, excluding literals/comments but including interpolation expressions.""" - # None denotes interpolated text; integers track record braces in expression sections. - scopes: list[int | None] = [0] - cursor = 0 - while cursor < len(formula): - depth = scopes[-1] - char = formula[cursor] - if depth is None: - if formula.startswith(('""', "{{", "}}"), cursor): - cursor += 2 - continue - if char == "{": - scopes.append(0) - elif char == '"': - scopes.pop() - cursor += 1 - continue - - if formula.startswith('$"', cursor): - scopes.append(None) - cursor += 2 - continue - - token_end = _skip_powerfx_opaque_token(formula, cursor) - if token_end != cursor: - cursor = token_end - continue - - if char == "{": - scopes[-1] = depth + 1 - elif char == "}": - if depth > 0: - scopes[-1] = depth - 1 - elif len(scopes) > 1: - scopes.pop() - cursor += 1 - continue - - yield cursor - cursor += 1 - - @dataclass(frozen=True) class DeclarativeEnvConfig: """Configuration that populates the PowerFx ``Env`` symbol for a workflow. @@ -721,14 +658,14 @@ 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") + # Empty arguments fall through so PowerFx reports the arity error. + if args_str: # Parse comma-separated arguments (handling nested parentheses) args = self._parse_function_args(args_str) evaluated_args: list[str] = [] @@ -747,24 +684,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: + 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: + 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: + inner_expr = inner_args.strip() # Reuse the helper method for consistent text extraction return self._eval_and_replace_message_text(inner_expr) diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_powerfx_functions.py b/python/packages/declarative/agent_framework_declarative/_workflows/_powerfx_functions.py index e5b2f4156d0..ddaad1e299f 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_powerfx_functions.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_powerfx_functions.py @@ -12,9 +12,156 @@ from __future__ import annotations +from collections.abc import Iterator from typing import Any, cast +def _skip_powerfx_opaque_token(formula: str, start: int) -> int: + """Return the end of an ordinary quoted token or comment, or start if neither.""" + quote = formula[start] + if quote in ('"', "'"): + pos = start + 1 + while pos < len(formula): + if formula[pos] == quote: + if pos + 1 < len(formula) and formula[pos + 1] == quote: + pos += 2 + continue + return pos + 1 + pos += 1 + return pos + if formula.startswith("//", start): + pos = start + 2 + while pos < len(formula) and formula[pos] not in "\r\n": + pos += 1 + return pos + if formula.startswith("/*", start): + end = formula.find("*/", start + 2) + # An unterminated /* is not a complete comment: leave it as code so a + # trailing one is not mistaken for trivia following a complete call. + return start if end == -1 else end + 2 + return start + + +def _iter_powerfx_expression_indices(formula: str) -> Iterator[int]: + """Yield code positions, excluding literals/comments but including interpolation expressions.""" + # None denotes interpolated text; integers track record braces in expression sections. + scopes: list[int | None] = [0] + cursor = 0 + while cursor < len(formula): + depth = scopes[-1] + char = formula[cursor] + if depth is None: + if formula.startswith(('""', "{{", "}}"), cursor): + cursor += 2 + continue + if char == "{": + scopes.append(0) + elif char == '"': + scopes.pop() + cursor += 1 + continue + + if formula.startswith('$"', cursor): + scopes.append(None) + cursor += 2 + continue + + token_end = _skip_powerfx_opaque_token(formula, cursor) + if token_end != cursor: + cursor = token_end + continue + + if char == "{": + scopes[-1] = depth + 1 + elif char == "}": + if depth > 0: + scopes[-1] = depth - 1 + elif len(scopes) > 1: + scopes.pop() + cursor += 1 + continue + + yield cursor + cursor += 1 + + +def _after_trailing_trivia(formula: str, start: int) -> int: + """Return the index after trailing whitespace and complete comments. + + Unlike the expression walker, this never skips a string literal: a quoted + token after the call's own closing paren is trailing code, not trivia, so it + stops the scan and leaves the formula to PowerFx. + """ + index = start + while index < len(formula): + if formula[index].isspace(): + index += 1 + continue + if formula.startswith(("//", "/*"), index): + token_end = _skip_powerfx_opaque_token(formula, index) + if token_end == index: + # Unterminated /* is not a complete comment, so real code follows. + break + index = token_end + continue + break + return index + + +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 nothing + but trivia may follow it. 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. + + Parentheses inside string literals and comments are data, not delimiters, so + the scan walks the formula with the shared PowerFx expression walker: it + skips those tokens whole and counts only parentheses that are real code, + including those inside an interpolated string's ``{...}`` expression sections. + + Args: + formula: The stripped formula to inspect. + *names: Candidate function names, matched case-sensitively. + + Returns: + The text between the call's parentheses - an empty string for an + argument-less call - or None when ``formula`` is not a single complete + call to one of ``names``. Callers that require an argument check the + result for emptiness so the call still falls through to PowerFx, which + reports the arity error. + """ + for name in names: + prefix = f"{name}(" + if not formula.startswith(prefix): + continue + positions = _iter_powerfx_expression_indices(formula) + depth = 0 + close_index = -1 + for index in positions: + char = formula[index] + if char == "(": + depth += 1 + elif char == ")": + depth -= 1 + if depth == 0: + close_index = index + break + if close_index == -1: + # Unbalanced: the call never closes, so it is not a whole call. + return None + # Only whitespace and complete comments may follow the call's own closing + # paren. The suffix scan deliberately does not reuse the literal-skipping + # walker: a trailing string literal (or any other code) there means the + # formula is a larger expression the caller must not claim. + if _after_trailing_trivia(formula, close_index + 1) != len(formula): + return None + return formula[len(prefix) : close_index] + return None + + def message_text(messages: Any) -> str: """Extract text content from a message or list of messages. diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_state.py b/python/packages/declarative/agent_framework_declarative/_workflows/_state.py index a617f812e66..c2a18a0c7c4 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_state.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_state.py @@ -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] diff --git a/python/packages/declarative/tests/test_powerfx_functions.py b/python/packages/declarative/tests/test_powerfx_functions.py index 239303794cf..8927538605a 100644 --- a/python/packages/declarative/tests/test_powerfx_functions.py +++ b/python/packages/declarative/tests/test_powerfx_functions.py @@ -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,157 @@ 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_trailing_string_literal_after_the_call_is_not_claimed(self): + """A quoted string after the call's own closing paren is trailing code. + + The suffix scan must not reuse the literal-skipping walker, or the + trailing string is silently dropped and a malformed formula is claimed + instead of being left to PowerFx. + """ + assert whole_call_args('Concat("a") "tail"', "Concat") is None + assert whole_call_args('Upper("x") /* c */ "tail"', "Upper") 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_match_with_an_empty_string(self): + """An argument-less call is a match; emptiness is the caller's decision. + + Returning None here would make a zero-argument call indistinguishable + from no call at all, which silently disabled ``Or()``/``And()`` in the + fallback dispatcher. + """ + assert whole_call_args("Concat()", "Concat") == "" + assert whole_call_args("Or()", "Or") == "" + + 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 + + def test_paren_inside_a_block_comment_does_not_close_the_call(self): + """A ``)`` in a comment is text, so the call is still one whole call.""" + assert whole_call_args('UserMessage("hi" /* ) */)', "UserMessage") == '"hi" /* ) */' + + def test_paren_inside_a_line_comment_does_not_close_the_call(self): + """The same holds for a ``//`` comment inside the argument list.""" + assert whole_call_args('UserMessage("hi" // )\n)', "UserMessage") == '"hi" // )\n' + + def test_trailing_comment_after_the_call_is_trivia(self): + """A comment following the call does not make the formula a larger expression.""" + assert whole_call_args('Upper("x") // trailing (', "Upper") == '"x"' + assert whole_call_args('Upper("x") /* trailing */', "Upper") == '"x"' + + def test_unterminated_block_comment_falls_through(self): + """An unclosed comment swallows the rest of the formula, so nothing matches.""" + assert whole_call_args('Upper("x" /* unterminated', "Upper") is None + + def test_trailing_unterminated_block_comment_falls_through(self): + """A complete call followed by an unterminated /* is a larger expression.""" + assert whole_call_args('Concat("a") /* unterminated', "Concat") is None + assert whole_call_args('Upper("x") /* unterminated', "Upper") is None + + def test_escaped_quote_inside_a_literal_is_data(self): + """``""`` is an escaped quote, so the literal does not end there.""" + assert whole_call_args('Concat("a"")", "b")', "Concat") == '"a"")", "b"' + + def test_paren_inside_an_interpolation_string_does_not_close_the_call(self): + """A ``)`` in a quoted string inside a ``$"{...}"`` expression is data, not the terminator.""" + formula = 'UserMessage($"x {If(true, ")", "x")}")' + assert whole_call_args(formula, "UserMessage") == '$"x {If(true, ")", "x")}"' + + +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" + + def test_argument_less_call_falls_through(self): + """``Concat()`` is left to PowerFx, which reports the arity error.""" + assert self._state()._eval_custom_function("Concat()") is None + assert self._state()._eval_custom_function("MessageText()") is None + + def test_trailing_comment_does_not_disqualify_the_call(self): + """A comment after the call is trivia, not a larger expression. + + A comment *inside* the argument list is claimed as well, but how its + text is then split and unquoted belongs to the argument heuristic (see + issue #9072), not to the extent of the match. + """ + assert self._state()._eval_custom_function('Concat("a", "b") /* done */') == "ab" + assert self._state()._eval_custom_function('Concat("a", "b") // done') == "ab" + + def test_trailing_unterminated_comment_falls_through(self): + """An unterminated trailing /* is not trivia, so PowerFx handles the formula.""" + assert self._state()._eval_custom_function('Concat("a") /* unterminated') is None + + def test_trailing_string_literal_falls_through(self): + """A trailing string after a complete call is a larger expression for PowerFx.""" + assert self._state()._eval_custom_function('Concat("a") "tail"') is None diff --git a/python/packages/declarative/tests/test_workflow_state.py b/python/packages/declarative/tests/test_workflow_state.py index 531b6c75e5d..4cc8859eafc 100644 --- a/python/packages/declarative/tests/test_workflow_state.py +++ b/python/packages/declarative/tests/test_workflow_state.py @@ -440,6 +440,70 @@ 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" + + def test_argument_less_call_still_dispatches(self): + """A zero-argument call is a call, not an unmatched formula. + + ``Or()`` is false in PowerFx. Echoing the formula instead would hand a + truthy string back to a condition and invert it. + """ + state = WorkflowState() + assert state._eval_simple("Or()") is False + assert state._eval_simple("And()") is True + assert state._eval_simple("Not()") is True + + def test_commented_paren_does_not_truncate_the_call(self): + """A ``)`` inside a comment is text, so the call is still claimed. + + What the argument heuristic then makes of the comment text is issue + #9072's business; the point here is that the formula is dispatched + rather than echoed back unevaluated. + """ + formula = 'Upper("ab" /* ) */)' + assert WorkflowState()._eval_simple(formula) != formula + + def test_trailing_comment_does_not_disqualify_the_call(self): + """A comment after the call is trivia, not a larger expression.""" + assert WorkflowState()._eval_simple('Upper("ab") // done') == "AB" + + class TestWorkflowStateParseFunctionArgs: """Tests for _parse_function_args helper."""