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 42aec8c49bb..f10a532eebf 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py @@ -983,8 +983,9 @@ def interpolate_string(self, text: str) -> str: Captures brace-delimited tokens whose root segment is an identifier (``[A-Za-z][A-Za-z0-9_]*``) followed by zero or more ``.`` separated dict-key segments. Resolution is delegated to :meth:`get`; unresolved - tokens are replaced with the empty string. Tokens that do not look - like state paths (e.g. ``{foo-bar}``, ``{Ctrl+C}``) are left literal. + tokens are replaced with the empty string. Substituted values remain + literal and are not interpolated again. Tokens that do not look like + state paths (e.g. ``{foo-bar}``, ``{Ctrl+C}``) are left literal. Args: text: Text that may contain {Variable.Path} references @@ -1004,12 +1005,7 @@ def replace_var(match: re.Match[str]) -> str: # per-segment safety on attribute traversal. pattern = r"\{([A-Za-z][A-Za-z0-9_]*(?:\.[^{}\s.]+)*)\}" - result = text - for match in re.finditer(pattern, text): - replacement = replace_var(match) - result = result.replace(match.group(0), replacement, 1) - - return result + return re.sub(pattern, replace_var, text) # Message types for inter-executor communication diff --git a/python/packages/declarative/tests/test_graph_executors.py b/python/packages/declarative/tests/test_graph_executors.py index c6b451e6f5a..af826ba626e 100644 --- a/python/packages/declarative/tests/test_graph_executors.py +++ b/python/packages/declarative/tests/test_graph_executors.py @@ -81,6 +81,71 @@ def track(executor: DeclarativeActionExecutor, store: State) -> DeclarativeWorkf forbidden.assert_not_called() +class TestFactoryTemplateInterpolation: + """Authored template tokens resolve once without rewriting substituted data.""" + + @pytest.mark.parametrize("mapping_activity", [False, True]) + @pytest.mark.parametrize( + "local, template, expected", + [ + ({"a": "{Local.b}", "b": "B"}, "{Local.a}|{Local.b}", "{Local.b}|B"), + ({"a": "prefix {Local.a}"}, "{Local.a}|{Local.a}", "prefix {Local.a}|prefix {Local.a}"), + ( + {"a": "{Local.b}/{Local.b}", "b": "B"}, + "{Local.a}|{Local.b}|{Local.b}", + "{Local.b}/{Local.b}|B|B", + ), + ({"a": "{Local.missing}"}, "{Local.a}|{Local.missing}", "{Local.missing}|"), + ({"a": ["{Local.b}"], "b": "B"}, "{Local.a}|{Local.b}", "['{Local.b}']|B"), + ({"a": {"text": "{Local.b}"}, "b": "B"}, "{Local.a}|{Local.b}", "{'text': '{Local.b}'}|B"), + ({"a": "safe"}, "{Local.a}|{Local.a}", "safe|safe"), + ({"a": "{Local.b}", "b": "B"}, "{Local.b}|{Local.a}", "B|{Local.b}"), + ({"a": "{Local.b}", "b": "B"}, "{Local.a}", "{Local.b}"), + ({}, "{Local.missing}|done", "|done"), + ({"a": None, "b": "B"}, "{Local.a}|{Local.b}", "|B"), + ({"a": False}, "{Local.a}|{Local.a}", "False|False"), + ({"a": 0}, "{Local.a}|{Local.a}", "0|0"), + ({"a": "文本"}, "{Local.a}|{Ctrl+C}|{foo-bar}", "文本|{Ctrl+C}|{foo-bar}"), + ({"a": r"\1\g<1>"}, "{Local.a}|{Local.a}", r"\1\g<1>|\1\g<1>"), + ({"a": "文本"}, "{{Local.a}}|{}|{Local. a}", "{文本}|{}|{Local. a}"), + ], + ids=[ + "later-token", + "self-token", + "multiple-tokens", + "missing-later-token", + "list-value", + "dict-value", + "ordinary-repeat", + "reverse-order", + "single-token", + "missing-token", + "none-value", + "false-value", + "zero-value", + "unicode-and-literals", + "backslashes", + "brace-boundaries", + ], + ) + async def test_factory_interpolates_only_authored_tokens( + self, mapping_activity: bool, local: dict[str, Any], template: str, expected: str + ) -> None: + workflow = WorkflowFactory().create_workflow_from_definition({ + "kind": "Workflow", + "actions": [ + *[{"kind": "SetVariable", "variable": f"Local.{key}", "value": value} for key, value in local.items()], + {"kind": "SendActivity", "activity": {"text": template} if mapping_activity else template}, + ], + }) + + result = await workflow.run({}) + assert result.get_outputs() == [expected] + + streamed_result = await workflow.run({}, stream=True).get_final_response() + assert streamed_result.get_outputs() == [expected] + + class TestDeclarativeWorkflowState: """Tests for DeclarativeWorkflowState."""