from __future__ import annotations import warnings from collections.abc import Iterator, Mapping from typing import Any, cast import pytest from tests.authoring.helpers import ( AutoBindInput, AutoBindOutput, AutoBindState, auto_bind_node, ) from wf_authoring import ( WorkflowBuilder, input_from, input_path, output_to, state, state_path, ) from wf_authoring.builder.mapping import normalize_input_mapping from wf_core import ( END, EndNode, NodeDef, RunStatus, ValidationIssueCode, Workflow, WorkflowExecutionError, ) from wf_core.models.steps import ( InputExpressionBinding, InputPathBinding, InputValueBinding, ) from wf_core.paths import GraphSourcePath, LocalPath, StatePath from wf_platform import CapabilityRef def _editable_three_step_builder() -> WorkflowBuilder: builder = WorkflowBuilder( name="editable_three_step", input_schema={"type": "object", "properties": {}}, state_schema={"type": "object", "properties": {}}, output_schema={"type": "object", "properties": {}}, start="first", ) builder.use_ref("demo.first", id="first") builder.use_ref("demo.second", id="second") builder.use_ref("demo.third", id="third") builder.connect("first", "ok", "second") builder.connect("second", "ok", "third") builder.connect("third", "ok", END) return builder def test_builder_round_trip_preserves_complete_workflow() -> None: original = Workflow.model_validate( { "name": "round_trip", "input_schema": { "type": "object", "properties": {"topic": {"type": "string"}}, }, "state_schema": { "type": "object", "properties": {"result": {"type": "string"}}, }, "output_schema": { "type": "object", "properties": {"result": {"type": "string"}}, }, "output": [{"path": "state.result", "target": "result"}], "node_defs": [ { "name": "app.default.search", "input_schema": {"type": "object", "properties": {}}, "output_schema": { "type": "object", "properties": {"result": {"type": "string"}}, }, "outcomes": ["ok"], } ], "outcomes": ["ok"], "start": "search", "nodes": [ { "id": "search", "type": "node", "node": "app.default.search", "input": [], "output": [{"source": "result", "target": "state.result"}], }, {"id": "end_ok", "type": "end", "outcome": "ok"}, ], "edges": [{"from": "search", "outcome": "ok", "to": "end_ok"}], } ) builder = WorkflowBuilder.from_workflow(original) rebuilt = builder.compile() assert rebuilt.model_dump(mode="json", by_alias=True) == original.model_dump( mode="json", by_alias=True ) builder.set_output([{"value": "changed", "target": "result"}]) original_output = original.output[0] assert isinstance(original_output, InputPathBinding) assert str(original_output.path) == "state.result" def test_set_route_replaces_unique_source_outcome_edge() -> None: builder = _editable_three_step_builder() builder.set_route("first", "ok", "third") matching = [ edge for edge in builder.edges if edge.from_ == "first" and edge.outcome == "ok" ] assert [(edge.from_, edge.outcome, edge.to) for edge in matching] == [ ("first", "ok", "third") ] def test_remove_route_removes_only_requested_source_outcome_pair() -> None: builder = _editable_three_step_builder() builder.connect("first", "error", "third") builder.remove_route("first", "ok") assert [(edge.from_, edge.outcome) for edge in builder.edges] == [ ("second", "ok"), ("third", "ok"), ("first", "error"), ] def test_remove_step_rejects_referenced_step() -> None: builder = _editable_three_step_builder() with pytest.raises(ValueError, match="still referenced by route"): builder.remove_step("second") def test_use_contract_registers_schema_only_external_node() -> None: builder = WorkflowBuilder( name="contract_builder", input_schema={"type": "object", "properties": {"topic": {"type": "string"}}}, state_schema={ "type": "object", "properties": {"result": {"type": "string"}}, }, output_schema={"type": "object", "properties": {"result": {"type": "string"}}}, ) contract = NodeDef.model_validate( { "name": "app.remote.search", "input_schema": { "type": "object", "properties": {"topic": {"type": "string"}}, }, "output_schema": { "type": "object", "properties": {"result": {"type": "string"}}, }, "outcomes": ["ok"], } ) step = builder.use_contract(contract, id="search") builder.set_entry_point(step) builder.connect(step, "ok", END) assert step.node == contract.name assert contract.name in {node_def.name for node_def in builder.compile().node_defs} assert builder.node_specs == {} assert builder.registry() == {} assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath.input("topic") assert step.output[0].target == StatePath.of("result") def test_validate_structure_reports_unknown_start_when_unset() -> None: builder = WorkflowBuilder( name="missing_start", input_schema={}, state_schema={"type": "object"}, output_schema={}, ) report = builder.validate_structure() assert report.errors[0].code == ValidationIssueCode.UNKNOWN_START def test_builder_auto_binds_matching_node_inputs_and_outputs_to_state() -> None: builder = WorkflowBuilder( name="auto_bind_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, start="update", ) step = builder.use(auto_bind_node, id="update") builder.connect(step, "ok", "__end__") run = builder.execute( {"text": "hello", "count": 1}, ) assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath.state("text") assert step.input[0].target == LocalPath.of("text") assert isinstance(step.input[1], InputPathBinding) assert step.input[1].path == GraphSourcePath.state("count") assert step.input[1].target == LocalPath.of("count") assert step.output[0].source == LocalPath.of("text") assert step.output[0].target == StatePath.of("text") assert step.output[1].source == LocalPath.of("count") assert step.output[1].target == StatePath.of("count") assert run.status == RunStatus.COMPLETED assert run.state["text"] == "HELLO" assert run.state["count"] == 2 def test_builder_preserves_explicit_nested_node_local_maps() -> None: builder = WorkflowBuilder( name="nested_local_maps", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) step = builder.use( auto_bind_node, input=[input_from(state_path("text"), "payload.text")], output=[output_to("payload.text", state_path("text"))], ) assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath.state("text") assert step.input[0].target == LocalPath.of("payload.text") assert step.output[0].source == LocalPath.of("payload.text") assert step.output[0].target == StatePath.of("text") def test_builder_use_accepts_typed_paths_and_literal_iterable_paths() -> None: builder = WorkflowBuilder( name="typed_path_maps", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) step = builder.use( auto_bind_node, input=[input_from(input_path('"text.with.dot"'), ("payload.text",))], output=[output_to(("payload.text",), state_path(("state field",)))], ) assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath("input", ("text.with.dot",)) assert step.input[0].target == LocalPath(("payload.text",)) assert step.output[0].source == LocalPath(("payload.text",)) assert step.output[0].target == StatePath(("state field",)) def test_builder_use_accepts_canonical_binding_dicts_with_structural_paths() -> None: builder = WorkflowBuilder( name="canonical_binding_dicts", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) step = builder.use( auto_bind_node, input=[ { "target": {"root": "local", "parts": ["payload.text"]}, "path": {"root": "input", "parts": ["text.with.dot"]}, }, { "target": {"root": "local", "parts": ["static.limit"]}, "value": 3, }, ], output=[ { "source": {"root": "local", "parts": ["payload.text"]}, "target": {"root": "state", "parts": ["text.with.dot"]}, } ], ) assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath("input", ("text.with.dot",)) assert step.input[0].target == LocalPath(("payload.text",)) assert isinstance(step.input[1], InputValueBinding) assert step.input[1].target == LocalPath(("static.limit",)) assert step.input[1].value == 3 assert step.output[0].source == LocalPath(("payload.text",)) assert step.output[0].target == StatePath(("text.with.dot",)) def test_builder_preserves_explicit_root_node_local_maps() -> None: builder = WorkflowBuilder( name="root_local_maps", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) step = builder.use( auto_bind_node, input=[input_from(state_path("text"), ".")], output=[output_to(".", state_path("text"))], ) assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath.state("text") assert step.input[0].target == LocalPath.root() assert step.output[0].source == LocalPath.root() assert step.output[0].target == StatePath.of("text") def test_builder_emits_canonical_node_bindings() -> None: builder = WorkflowBuilder( name="canonical_bindings", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, start="update", ) step = builder.use( auto_bind_node, id="update", input=[input_from(input_path("text"), "text")], output=[output_to("text", state_path("text"))], ) builder.connect(step, "ok", END) dumped_node = builder.compile().model_dump(mode="json")["nodes"][0] assert dumped_node["input"][0]["path"] == "input.text" assert dumped_node["input"][0]["target"] == "text" assert dumped_node["output"][0]["source"] == "text" assert dumped_node["output"][0]["target"] == "state.text" assert "in_map" not in dumped_node assert "input_values" not in dumped_node assert "out_map" not in dumped_node def test_builder_can_auto_id_node_uses_from_spec_name() -> None: builder = WorkflowBuilder( name="auto_id_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, start="test_auto_bind", ) first = builder.use(auto_bind_node) second = builder.use(auto_bind_node) assert first.id == "test_auto_bind" assert second.id == "test_auto_bind_2" def test_builder_can_compile_with_explicit_start_set_later() -> None: builder = WorkflowBuilder( name="optional_start_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) step = builder.use(auto_bind_node) builder.set_entry_point(step) workflow = builder.compile() assert workflow.start == "test_auto_bind" def test_builder_requires_explicit_start_before_compile() -> None: builder = WorkflowBuilder( name="missing_start_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) with pytest.raises(WorkflowExecutionError, match="start"): builder.compile() def test_builder_registry_exports_used_node_specs() -> None: builder = WorkflowBuilder( name="registry_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, start="test_auto_bind", ) builder.use(auto_bind_node) assert set(builder.registry()) == {"test.auto_bind"} def test_builder_execute_compiles_and_runs_with_used_registry() -> None: builder = WorkflowBuilder( name="execute_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) step = builder.use(auto_bind_node) builder.set_entry_point(step) builder.connect(step, "ok", "__end__") run = builder.execute({"text": "hello", "count": 1}) assert run.status == RunStatus.COMPLETED assert run.state["text"] == "HELLO" assert run.state["count"] == 2 def test_builder_can_auto_id_condition_foreach_and_interrupt() -> None: builder = WorkflowBuilder( name="auto_id_control_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) first_condition = builder.condition(check=state("count").gt(0)) second_condition = builder.condition(check=state("count").gt(1)) foreach = builder.foreach(over="state.tags", as_="tag") interrupt = builder.interrupt(kind="approval") assert first_condition.id == "state_count" assert second_condition.id == "state_count_2" assert foreach.id == "foreach_tag" assert interrupt.id == "interrupt_approval" def test_builder_interrupt_accepts_canonical_request_and_resume_bindings() -> None: builder = WorkflowBuilder( name="interrupt_bindings", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) interrupt = builder.interrupt( kind="approval", request=[input_from(input_path("text"), "message")], resume=[output_to("text", state_path("text"))], ) assert isinstance(interrupt.request[0], InputPathBinding) assert interrupt.request[0].path == GraphSourcePath.input("text") assert interrupt.request[0].target == LocalPath.of("message") assert interrupt.resume[0].source == LocalPath.of("text") assert interrupt.resume[0].target == StatePath.of("text") def test_builder_interrupt_accepts_composite_request_binding() -> None: builder = WorkflowBuilder( name="interrupt_composite_request", input_schema={"type": "object"}, state_schema={"type": "object"}, output_schema={"type": "object"}, ) interrupt = builder.interrupt( kind="approval", request=[ { "target": "message", "expression": { "kind": "object", "fields": {"text": {"kind": "literal", "value": "review"}}, }, } ], ) assert isinstance(interrupt.request[0], InputExpressionBinding) def test_builder_interrupt_accepts_request_and_resume_schemas() -> None: builder = WorkflowBuilder( name="interrupt_contract", input_schema={"type": "object", "properties": {}}, state_schema={"fields": {}}, output_schema={"type": "object", "properties": {}}, ) interrupt = builder.interrupt( kind="approval", request_schema={ "type": "object", "properties": {"message": {"type": "string"}}, "required": ["message"], }, resume_schema={ "type": "object", "properties": {"approved": {"type": "boolean"}}, "required": ["approved"], }, ) assert interrupt.request_schema["required"] == ["message"] assert interrupt.resume_schema["required"] == ["approved"] assert interrupt.has_explicit_contract is True def test_builder_connect_can_use_node_specs_and_returns_resolved_refs() -> None: builder = WorkflowBuilder( name="connect_specs_demo", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) source, target = builder.connect(auto_bind_node, "ok", auto_bind_node) assert not isinstance(source, str) assert not isinstance(target, str) assert source.id == "test_auto_bind" assert target.id == "test_auto_bind_2" assert builder.edges[0].from_ == "test_auto_bind" assert builder.edges[0].outcome == "ok" assert builder.edges[0].to == "test_auto_bind_2" def test_builder_use_ref_creates_external_node_use_without_node_def() -> None: builder = WorkflowBuilder( name="external_ref_demo", input_schema={}, state_schema={"fields": {}}, output_schema={}, ) step = builder.use_ref( "demo.echo", id="echo", input=[input_from(input_path("text"), "text")], output=[output_to("echoed", state_path("echoed"))], ) builder.set_entry_point(step) builder.connect(step, "ok", END) workflow = builder.compile() assert step.node == "demo.echo" assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath.input("text") assert step.input[0].target == LocalPath.of("text") assert step.output[0].source == LocalPath.of("echoed") assert step.output[0].target == StatePath.of("echoed") assert workflow.node_defs == [] def test_builder_use_ref_accepts_canonical_binding_dicts() -> None: builder = WorkflowBuilder( name="external_ref_canonical_bindings", input_schema={}, state_schema={"fields": {}}, output_schema={}, ) step = builder.use_ref( "demo.echo", id="echo", input=[ { "target": {"root": "local", "parts": ["text"]}, "path": {"root": "input", "parts": ["text"]}, } ], output=[ { "source": {"root": "local", "parts": ["echoed"]}, "target": {"root": "state", "parts": ["echoed"]}, } ], ) assert step.node == "demo.echo" assert isinstance(step.input[0], InputPathBinding) assert step.input[0].path == GraphSourcePath.input("text") assert step.output[0].target == StatePath.of("echoed") def test_builder_use_accepts_composite_node_input_binding() -> None: builder = WorkflowBuilder( name="composite_node_input", input_schema={"type": "object"}, state_schema={"type": "object"}, output_schema={"type": "object"}, ) step = builder.use( auto_bind_node, id="concat", input=[ { "target": "items", "expression": { "kind": "array", "items": [ {"kind": "path", "path": "state.value"}, {"kind": "literal", "value": "!"}, ], }, } ], ) assert isinstance(step.input[0], InputExpressionBinding) def test_builder_use_ref_accepts_composite_node_input_binding() -> None: builder = WorkflowBuilder( name="composite_node_input_ref", input_schema={"type": "object"}, state_schema={"type": "object"}, output_schema={"type": "object"}, ) step = builder.use_ref( "demo.concat", id="concat", input=[ { "target": "items", "expression": { "kind": "array", "items": [ {"kind": "path", "path": "state.value"}, {"kind": "literal", "value": "!"}, ], }, } ], ) assert isinstance(step.input[0], InputExpressionBinding) def test_builder_use_ref_accepts_structural_capability_ref() -> None: builder = WorkflowBuilder( name="external_ref_structural", input_schema={}, state_schema={"fields": {}}, output_schema={}, ) step = builder.use_ref( CapabilityRef.parse("demo.personal.echo"), input=[input_from(input_path("text"), "text")], ) assert step.node == "demo.personal.echo" assert step.id == "demo_personal_echo" def test_builder_warns_when_explicit_deprecated_maps_are_used() -> None: builder = WorkflowBuilder( name="deprecated_maps", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) with pytest.warns(DeprecationWarning, match="canonical input/output"): builder.use( auto_bind_node, in_map={"input.text": "text"}, out_map={"text": "state.text"}, ) def test_builder_auto_mapping_does_not_warn() -> None: builder = WorkflowBuilder( name="auto_map_no_warning", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) with warnings.catch_warnings(): warnings.simplefilter("error", DeprecationWarning) builder.use(auto_bind_node) def test_builder_rejects_mixed_canonical_and_deprecated_input_styles() -> None: builder = WorkflowBuilder( name="mixed_input_styles", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) with pytest.raises(TypeError, match="cannot mix canonical input"): cast(Any, builder.use)( auto_bind_node, input=[{"target": "text", "path": "input.text"}], in_map={"input.text": "text"}, ) def test_builder_rejects_mixed_canonical_and_deprecated_output_styles() -> None: builder = WorkflowBuilder( name="mixed_output_styles", input_schema=AutoBindInput, state_schema=AutoBindState, output_schema=AutoBindOutput, ) with pytest.raises(TypeError, match="cannot mix canonical output"): cast(Any, builder.use)( auto_bind_node, output=[{"source": "text", "target": "state.text"}], out_map={"text": "state.text"}, ) def test_builder_adds_explicit_end_node() -> None: builder = WorkflowBuilder( name="explicit_end", input_schema={}, state_schema={"fields": {}}, output_schema={}, outcomes=["ok", "error"], ) terminal = builder.end("error", id="end_error") assert isinstance(terminal, EndNode) assert terminal.id == "end_error" assert terminal.outcome == "error" assert builder.nodes[-1] is terminal builder.set_entry_point(terminal) assert builder.compile().outcomes == ["ok", "error"] class _StructuralKeyMap: def __getitem__(self, key: object) -> object: raise KeyError(key) def __iter__(self) -> Iterator[object]: return iter(()) def __len__(self) -> int: return 1 def items(self) -> list[tuple[dict[str, object], str]]: # Deliberately violates Mapping's item-view contract to exercise the # runtime guard for structural dict keys from malformed mappings. return [ ( {"root": "input", "parts": ["email.address"]}, "payload.email", ) ] def test_input_map_rejects_structural_dict_keys_with_clear_message() -> None: with pytest.raises(TypeError, match="structural path dicts cannot be map keys"): normalize_input_mapping(cast(Mapping[object, object], _StructuralKeyMap()))