diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_builder.py b/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_builder.py index 9a6814af60d..9e7a18711e6 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_builder.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_declarative_builder.py @@ -173,7 +173,8 @@ def __init__( self._checkpoint_storage = checkpoint_storage self._pending_gotos: list[tuple[Any, str]] = [] # (goto_executor, target_id) self._validate = validate - self._seen_explicit_ids: set[str] = set() # Track explicit IDs for duplicate detection + self._seen_explicit_ids: set[str] = set() + self._generated_ids: set[str] = set() self._http_request_handler = http_request_handler self._mcp_tool_handler = mcp_tool_handler self._env_config: DeclarativeEnvConfig = env_config if env_config is not None else DeclarativeEnvConfig() @@ -201,6 +202,9 @@ def build(self) -> Workflow: if self._validate: self._validate_workflow(actions) + # Reserve nested and forward-declared executor IDs before naming internal nodes. + self._collect_explicit_ids(actions) + # Create a stable entry node as the start executor, then wire it to the first action. # This avoids needing a placeholder since the entry executor isn't known until after # _create_executors_for_actions runs (which itself needs the builder to add edges). @@ -238,6 +242,39 @@ def build(self) -> Workflow: return builder.build() + def _collect_explicit_ids(self, actions: list[dict[str, Any]] | None) -> None: + """Reserve authored executor IDs, excluding virtual structures and skipped actions.""" + if not actions: + return + for action_def in actions: + kind = action_def.get("kind", "") + creates_executor = ( + kind in ALL_ACTION_EXECUTORS + or kind in ("BreakLoop", "ContinueLoop") + or (kind == "GotoAction" and bool(action_def.get("target") or action_def.get("actionId"))) + ) + if creates_executor and (explicit_id := action_def.get("id")): + self._seen_explicit_ids.add(explicit_id) + if kind == "If": + self._collect_explicit_ids(action_def.get("then", action_def.get("actions", []))) + self._collect_explicit_ids(action_def.get("else", [])) + elif kind == "ConditionGroup": + for branch in action_def.get("conditions", []): + self._collect_explicit_ids(branch.get("actions", [])) + self._collect_explicit_ids(action_def.get("elseActions", [])) + elif kind == "Foreach": + self._collect_explicit_ids(action_def.get("actions", [])) + + def _allocate_generated_id(self, preferred_id: str) -> str: + """Retain the existing ID unless an authored or generated ID already uses it.""" + candidate = preferred_id + suffix = 2 + while candidate in self._seen_explicit_ids or candidate in self._generated_ids or candidate in self._executors: + candidate = f"{preferred_id}_{suffix}" + suffix += 1 + self._generated_ids.add(candidate) + return candidate + def _validate_workflow(self, actions: list[dict[str, Any]]) -> None: """Validate the workflow definition before building. @@ -489,7 +526,8 @@ def _create_executor_for_action( action_id = explicit_id else: parent_id = (parent_context or {}).get("parent_id") - action_id = f"{parent_id}_{kind}_{self._action_index}" if parent_id else f"{kind}_{self._action_index}" + preferred_id = f"{parent_id}_{kind}_{self._action_index}" if parent_id else f"{kind}_{self._action_index}" + action_id = self._allocate_generated_id(preferred_id) self._action_index += 1 # Pass agents/tools to specialized executors @@ -574,7 +612,7 @@ def _create_if_structure( evaluator = IfConditionEvaluatorExecutor( action_def, condition_expr, - id=f"{action_id}_eval", + id=self._allocate_generated_id(f"{action_id}_eval"), ) self._executors[evaluator.id] = evaluator @@ -588,7 +626,9 @@ def _create_if_structure( else_passthrough = None if not else_entry: # No else branch - create a passthrough for continuation when condition is false - else_passthrough = JoinExecutor({"kind": "ElsePassthrough"}, id=f"{action_id}_else_pass") + else_passthrough = JoinExecutor( + {"kind": "ElsePassthrough"}, id=self._allocate_generated_id(f"{action_id}_else_pass") + ) self._executors[else_passthrough.id] = else_passthrough # Wire evaluator to branches with conditions that check ConditionResult.branch_index @@ -673,7 +713,7 @@ def _create_condition_group_structure( evaluator: DeclarativeActionExecutor = ConditionGroupEvaluatorExecutor( action_def, conditions, - id=f"{action_id}_eval", + id=self._allocate_generated_id(f"{action_id}_eval"), ) self._executors[evaluator.id] = evaluator @@ -708,7 +748,9 @@ def _create_condition_group_structure( else: # No else actions - create a passthrough for the "no match" case # This allows the workflow to continue to the next action when no condition matches - default_passthrough = JoinExecutor({"kind": "DefaultPassthrough"}, id=f"{action_id}_default") + default_passthrough = JoinExecutor( + {"kind": "DefaultPassthrough"}, id=self._allocate_generated_id(f"{action_id}_default") + ) self._executors[default_passthrough.id] = default_passthrough branch_exits.append(default_passthrough) @@ -781,15 +823,17 @@ def _create_foreach_structure( self._action_index += 1 # Create foreach init executor - init_executor = ForeachInitExecutor(action_def, id=f"{action_id}_init") + init_executor = ForeachInitExecutor(action_def, id=self._allocate_generated_id(f"{action_id}_init")) self._executors[init_executor.id] = init_executor # Create foreach next executor (for advancing to next item) - next_executor = ForeachNextExecutor(action_def, init_executor.id, id=f"{action_id}_next") + next_executor = ForeachNextExecutor( + action_def, init_executor.id, id=self._allocate_generated_id(f"{action_id}_next") + ) self._executors[next_executor.id] = next_executor # Create join node for loop exit - join_executor = JoinExecutor({"kind": "Join"}, id=f"{action_id}_exit") + join_executor = JoinExecutor({"kind": "Join"}, id=self._allocate_generated_id(f"{action_id}_exit")) self._executors[join_executor.id] = join_executor # Create loop body @@ -865,7 +909,7 @@ def _create_goto_reference( return None # Create a pass-through executor for the goto - action_id = action_def.get("id") or f"goto_{target_id}_{self._action_index}" + action_id = action_def.get("id") or self._allocate_generated_id(f"goto_{target_id}_{self._action_index}") self._action_index += 1 # Use JoinExecutor as a simple pass-through node @@ -892,7 +936,7 @@ def _create_break_executor( if parent_context and "loop_next_executor" in parent_context: loop_next = parent_context["loop_next_executor"] - action_id = action_def.get("id") or f"Break_{self._action_index}" + action_id = action_def.get("id") or self._allocate_generated_id(f"Break_{self._action_index}") self._action_index += 1 executor = BreakLoopExecutor(action_def, loop_next.id, id=action_id) @@ -920,7 +964,7 @@ def _create_continue_executor( if parent_context and "loop_next_executor" in parent_context: loop_next = parent_context["loop_next_executor"] - action_id = action_def.get("id") or f"Continue_{self._action_index}" + action_id = action_def.get("id") or self._allocate_generated_id(f"Continue_{self._action_index}") self._action_index += 1 executor = ContinueLoopExecutor(action_def, loop_next.id, id=action_id) diff --git a/python/packages/declarative/tests/test_workflow_generated_ids.py b/python/packages/declarative/tests/test_workflow_generated_ids.py new file mode 100644 index 00000000000..3951c2de301 --- /dev/null +++ b/python/packages/declarative/tests/test_workflow_generated_ids.py @@ -0,0 +1,277 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Generated graph IDs must not shadow explicit declarative action IDs.""" + +from copy import deepcopy +from typing import Any + +import pytest + +from agent_framework_declarative import WorkflowFactory +from agent_framework_declarative._workflows import DeclarativeWorkflowBuilder +from agent_framework_declarative._workflows._errors import DeclarativeWorkflowError + +try: + import powerfx # noqa: F401 + + _powerfx_available = True +except (ImportError, RuntimeError): + _powerfx_available = False + +_requires_powerfx = pytest.mark.skipif(not _powerfx_available, reason="PowerFx engine not available") + + +@pytest.mark.parametrize("explicit_first", [False, True]) +async def test_unnamed_action_preserves_explicit_id(explicit_first: bool) -> None: + explicit_id = "SendActivity_1" if explicit_first else "SendActivity_0" + unnamed = {"kind": "SendActivity", "activity": "unnamed"} + explicit = {"kind": "SendActivity", "id": explicit_id, "activity": "explicit"} + actions = [explicit, unnamed] if explicit_first else [unnamed, explicit] + workflow = WorkflowFactory().create_workflow_from_definition({"actions": actions}) + + events = await workflow.run({}) + + assert events.get_outputs() == (["explicit", "unnamed"] if explicit_first else ["unnamed", "explicit"]) + assert workflow.executors[explicit_id].id == explicit_id + + +@pytest.mark.parametrize("explicit_first", [False, True]) +@pytest.mark.parametrize( + ("kind", "suffix"), + [ + pytest.param("If", "eval", marks=_requires_powerfx), + pytest.param("If", "else_pass", marks=_requires_powerfx), + pytest.param("ConditionGroup", "eval", marks=_requires_powerfx), + pytest.param("ConditionGroup", "default", marks=_requires_powerfx), + ("Foreach", "init"), + ("Foreach", "next"), + ("Foreach", "exit"), + ], +) +async def test_control_flow_internal_ids_preserve_explicit_actions( + kind: str, suffix: str, explicit_first: bool +) -> None: + child = {"kind": "SendActivity", "activity": "control"} + control: dict[str, Any] = {"kind": kind, "id": "branch"} + if kind == "If": + control.update(condition=True, then=[child]) + elif kind == "ConditionGroup": + control["conditions"] = [{"condition": "=true", "actions": [child]}] + else: + control.update(source=[1, 2], actions=[child]) + explicit_id = f"branch_{suffix}" + explicit = {"kind": "SendActivity", "id": explicit_id, "activity": "explicit"} + actions = [explicit, control] if explicit_first else [control, explicit] + workflow = WorkflowFactory().create_workflow_from_definition({"actions": actions}) + + events = await workflow.run({}) + + control_outputs = ["control", "control"] if kind == "Foreach" else ["control"] + assert events.get_outputs() == ( + ["explicit", *control_outputs] if explicit_first else [*control_outputs, "explicit"] + ) + assert workflow.executors[explicit_id].id == explicit_id + + +@_requires_powerfx +async def test_nested_explicit_id_is_reserved_before_parent_nodes_are_created() -> None: + workflow = WorkflowFactory().create_workflow_from_definition({ + "actions": [ + { + "kind": "If", + "id": "branch", + "condition": True, + "then": [{"kind": "SendActivity", "id": "branch_eval", "activity": "nested"}], + }, + {"kind": "SendActivity", "activity": "after"}, + ] + }) + + assert (await workflow.run({})).get_outputs() == ["nested", "after"] + assert "branch_eval" in workflow.executors + + +async def test_generated_suffix_avoids_other_explicit_ids_and_goto_targets() -> None: + workflow = WorkflowFactory().create_workflow_from_definition({ + "actions": [ + {"kind": "SendActivity", "activity": "first"}, + {"kind": "SendActivity", "id": "SendActivity_0_2", "activity": "suffix"}, + {"kind": "GotoAction", "actionId": "SendActivity_0"}, + {"kind": "SendActivity", "id": "SendActivity_0", "activity": "target"}, + ] + }) + + assert (await workflow.run({})).get_outputs() == ["first", "suffix", "target"] + assert {"SendActivity_0", "SendActivity_0_2"} <= workflow.executors.keys() + + +async def test_id_allocation_does_not_require_schema_validation() -> None: + workflow = DeclarativeWorkflowBuilder( + { + "actions": [ + {"kind": "SendActivity", "activity": "first"}, + {"kind": "SendActivity", "id": "SendActivity_0", "activity": "second"}, + ] + }, + validate=False, + ).build() + + assert (await workflow.run({})).get_outputs() == ["first", "second"] + + +async def test_noncolliding_generated_ids_are_unchanged() -> None: + workflow = WorkflowFactory().create_workflow_from_definition({ + "actions": [ + {"kind": "SendActivity", "activity": "first"}, + {"kind": "SendActivity", "activity": "second"}, + ] + }) + + assert list(workflow.executors) == ["_workflow_entry", "SendActivity_0", "SendActivity_1"] + assert (await workflow.run({})).get_outputs() == ["first", "second"] + + +def test_duplicate_explicit_action_ids_still_raise() -> None: + with pytest.raises(DeclarativeWorkflowError, match="Duplicate action ID 'same'"): + WorkflowFactory().create_workflow_from_definition({ + "actions": [ + {"kind": "SendActivity", "id": "same", "activity": "first"}, + {"kind": "SendActivity", "id": "same", "activity": "second"}, + ] + }) + + +@pytest.mark.parametrize("nested", [False, True]) +def test_reserved_entry_action_id_still_raises(nested: bool) -> None: + action: dict[str, Any] = {"kind": "SendActivity", "id": "_workflow_entry", "activity": "reserved"} + if nested: + action = {"kind": "If", "condition": True, "then": [action]} + + with pytest.raises(DeclarativeWorkflowError, match="reserved for internal use"): + WorkflowFactory().create_workflow_from_definition({"actions": [action]}) + + +@pytest.mark.parametrize("kind", ["BreakLoop", "ContinueLoop"]) +@pytest.mark.parametrize("collide_with_next", [False, True]) +async def test_loop_terminators_use_the_allocated_loop_node(kind: str, collide_with_next: bool) -> None: + terminator_prefix = "Break" if kind == "BreakLoop" else "Continue" + explicit_id = "loop_next" if collide_with_next else f"{terminator_prefix}_2" + definition: dict[str, Any] = { + "actions": [ + { + "kind": "Foreach", + "id": "loop", + "source": [1, 2], + "actions": [{"kind": "SendActivity", "activity": "inside"}, {"kind": kind}], + }, + {"kind": "SendActivity", "id": explicit_id, "activity": "after"}, + ] + } + normal_definition = deepcopy(definition) + normal_definition["actions"][1]["id"] = "after" + normal = WorkflowFactory().create_workflow_from_definition(normal_definition) + workflow = WorkflowFactory().create_workflow_from_definition(definition) + + assert (await workflow.run({})).get_outputs() == (await normal.run({})).get_outputs() + + +@pytest.mark.parametrize( + "kind", + [pytest.param("If", marks=_requires_powerfx), pytest.param("ConditionGroup", marks=_requires_powerfx), "Foreach"], +) +async def test_noncolliding_control_structure_name_is_unchanged(kind: str) -> None: + child = {"kind": "SendActivity", "activity": "inside"} + control: dict[str, Any] = {"kind": kind} + if kind == "If": + control.update(condition=True, then=[child]) + expected_ids = ["If_0_eval", "If_0_SendActivity_1", "If_0_else_pass"] + elif kind == "ConditionGroup": + control["conditions"] = [{"condition": "=true", "actions": [child]}] + expected_ids = ["ConditionGroup_0_eval", "ConditionGroup_0_case0_SendActivity_1", "ConditionGroup_0_default"] + else: + control.update(source=[1], actions=[child]) + expected_ids = ["Foreach_0_init", "SendActivity_1", "Foreach_0_next", "Foreach_0_exit"] + explicit_id = f"{kind}_0" + workflow = WorkflowFactory().create_workflow_from_definition({ + "actions": [control, {"kind": "SendActivity", "id": explicit_id, "activity": "after"}] + }) + + assert list(workflow.executors) == ["_workflow_entry", *expected_ids, explicit_id] + assert (await workflow.run({})).get_outputs() == ["inside", "after"] + + +@pytest.mark.parametrize("control_first", [False, True]) +@pytest.mark.parametrize( + "kind", + [pytest.param("If", marks=_requires_powerfx), pytest.param("ConditionGroup", marks=_requires_powerfx), "Foreach"], +) +async def test_authored_virtual_id_does_not_rename_an_ordinary_executor(kind: str, control_first: bool) -> None: + ordinary_id = "SendActivity_2" if control_first else "SendActivity_0" + child = {"kind": "SendActivity", "activity": "inside"} + control: dict[str, Any] = {"kind": kind, "id": ordinary_id} + if kind == "If": + control.update(condition=True, then=[child]) + internal_ids = [f"{ordinary_id}_eval", f"{ordinary_id}_SendActivity_1", f"{ordinary_id}_else_pass"] + elif kind == "ConditionGroup": + control["conditions"] = [{"condition": "=true", "actions": [child]}] + internal_ids = [f"{ordinary_id}_eval", f"{ordinary_id}_case0_SendActivity_1", f"{ordinary_id}_default"] + else: + control.update(source=[1], actions=[child]) + internal_ids = [f"{ordinary_id}_init", "SendActivity_1", f"{ordinary_id}_next", f"{ordinary_id}_exit"] + if not control_first: + internal_ids = [node.replace("SendActivity_1", "SendActivity_2") for node in internal_ids] + ordinary = {"kind": "SendActivity", "activity": "ordinary"} + actions = [control, ordinary] if control_first else [ordinary, control] + workflow = WorkflowFactory().create_workflow_from_definition({"actions": actions}) + + expected_ids = [*internal_ids, ordinary_id] if control_first else [ordinary_id, *internal_ids] + assert set(workflow.executors) == {"_workflow_entry", *expected_ids} + assert (await workflow.run({})).get_outputs() == ( + ["inside", "ordinary"] if control_first else ["ordinary", "inside"] + ) + + +@pytest.mark.parametrize("skipped_first", [False, True]) +@pytest.mark.parametrize("kind", ["UnknownAction", "GotoAction"]) +async def test_skipped_action_id_does_not_rename_an_ordinary_executor(kind: str, skipped_first: bool) -> None: + skipped = {"kind": kind, "id": "SendActivity_0", "actionId": ""} + ordinary = {"kind": "SendActivity", "activity": "ordinary"} + actions = [skipped, ordinary] if skipped_first else [ordinary, skipped] + workflow = WorkflowFactory().create_workflow_from_definition({"actions": actions}) + + assert list(workflow.executors) == ["_workflow_entry", "SendActivity_0"] + assert (await workflow.run({})).get_outputs() == ["ordinary"] + + +@pytest.mark.parametrize("kind", ["BreakLoop", "ContinueLoop"]) +def test_loop_terminators_outside_a_loop_still_raise(kind: str) -> None: + with pytest.raises(DeclarativeWorkflowError, match="can only be used inside a Foreach loop"): + WorkflowFactory().create_workflow_from_definition({ + "actions": [ + {"kind": "SendActivity", "activity": "ordinary"}, + {"kind": kind, "id": "SendActivity_0"}, + ] + }) + + +@_requires_powerfx +@pytest.mark.parametrize("null_then", [False, True]) +async def test_nullable_inactive_if_branch_remains_supported(null_then: bool) -> None: + child = {"kind": "SendActivity", "id": "child", "activity": "inside"} + workflow = WorkflowFactory().create_workflow_from_definition({ + "actions": [ + { + "kind": "If", + "id": "branch", + "condition": not null_then, + "then": None if null_then else [child], + "else": [child] if null_then else None, + } + ] + }) + + expected_ids = {"_workflow_entry", "branch_eval", "child"} + if not null_then: + expected_ids.add("branch_else_pass") + assert set(workflow.executors) == expected_ids + assert (await workflow.run({})).get_outputs() == ["inside"]