diff --git a/src/conductor/config/validator.py b/src/conductor/config/validator.py index e807068a..7e775229 100644 --- a/src/conductor/config/validator.py +++ b/src/conductor/config/validator.py @@ -82,15 +82,18 @@ def get(self, key: str, default: object = None) -> object: _MAX_ENUMERATED_PATHS = 100 # Pattern for input references: -# - agent.output(.field)? -# - parallel_group.outputs.agent(.field)? +# - agent.output[.field[.subfield...]] +# - parallel_group.outputs|errors[.agent[.field]] (parallel depth unchanged) # - workflow.input.param +# - agent.field[.subfield...] (shorthand; excludes workflow.*) # All with optional ? suffix INPUT_REF_PATTERN = re.compile( r"^(?:" - r"(?P[a-zA-Z_][a-zA-Z0-9_]*)\.output(?:\.(?P[a-zA-Z_][a-zA-Z0-9_]*))?|" + r"(?P[a-zA-Z_][a-zA-Z0-9_]*)\.output(?:\.(?P[a-zA-Z_][a-zA-Z0-9_]*)(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)?|" r"(?P[a-zA-Z_][a-zA-Z0-9_]*)\.(?Poutputs|errors)(?:\.(?P[a-zA-Z_][a-zA-Z0-9_]*)(?:\.(?P[a-zA-Z_][a-zA-Z0-9_]*))?)?|" - r"workflow\.input\.(?P[a-zA-Z_][a-zA-Z0-9_]*)" + r"workflow\.input\.(?P[a-zA-Z_][a-zA-Z0-9_]*)|" + r"(?!workflow\b)(?![a-zA-Z_][a-zA-Z0-9_]*\.(?:outputs|errors)(?:\.|$))" + r"(?P[a-zA-Z_][a-zA-Z0-9_]*)\.(?P[a-zA-Z_][a-zA-Z0-9_]*)(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*" r")(?P\?)?$" ) @@ -321,14 +324,17 @@ def _validate_input_references( if not match: errors.append( f"Agent '{agent_name}' has invalid input reference '{input_ref}'. " - "Expected format: 'agent_name.output', 'agent_name.output.field', " - "'parallel_group.outputs.agent_name', 'parallel_group.outputs.agent_name.field', " + "Expected formats: 'agent_name.output', " + "'agent_name.output.field[.subfield...]', " + "'agent_name.field[.subfield...]' (shorthand), " + "'parallel_group.outputs[.agent_name[.field]]', " + "'parallel_group.errors[.agent_name]', " "or 'workflow.input.param_name' (append '?' for optional)" ) continue # Check if referencing another agent's output - ref_agent = match.group("agent") + ref_agent = match.group("agent") or match.group("shorthand") if ref_agent and ref_agent not in agent_names: is_optional = match.group("optional") == "?" if is_optional: @@ -858,6 +864,9 @@ def _extract_template_refs(template: str) -> TemplateRefs: agent_refs.add(root) if kind == "output": # attrs is ["output"] or ["output", "", ...] + # Keep first-level precision only: deeper chains like + # a.output.foo.bar intentionally record field="foo" so advisory + # checks compare declared first-level fields. field: str | None = attrs[1] if len(attrs) >= 2 else None agent_output_fields.setdefault(root, set()).add(field) else: # kind == "outputs" @@ -1312,10 +1321,14 @@ def _validate_template_references( match = INPUT_REF_PATTERN.match(ref.rstrip("?")) if not match: continue - ref_agent = match.group("agent") + ref_agent = match.group("agent") or match.group("shorthand") if ref_agent: - field = match.group("field") - # field is None for bare ``a.output`` (whole output declared). + # For explicit refs, ``field`` is the first component after + # ``.output`` (or None for bare ``a.output``). For shorthand + # refs, ``sh_field`` is the first component after the agent + # name. Nested paths intentionally degrade to first-level + # advisory precision. + field = match.group("field") if match.group("agent") else match.group("sh_field") declared_agent_output_fields.setdefault(ref_agent, set()).add(field) ref_parallel = match.group("parallel") if ref_parallel: diff --git a/src/conductor/engine/context.py b/src/conductor/engine/context.py index 148a2523..62cb9305 100644 --- a/src/conductor/engine/context.py +++ b/src/conductor/engine/context.py @@ -276,8 +276,8 @@ def _add_explicit_input(self, ctx: dict[str, Any], input_ref: str) -> None: Input reference formats: - workflow.input.param_name - References a workflow input - agent_name.output - References an agent's entire output - - agent_name.output.field - References a specific output field - - agent_name.field - Shorthand for agent_name.output.field (deprecated but supported) + - agent_name.output.field[.subfield...] - Nested output projection + - agent_name.field[.subfield...] - Shorthand nested projection - parallel_group.outputs - References all parallel group outputs - parallel_group.outputs.agent_name - References a specific parallel agent's output - parallel_group.outputs.agent_name.field - Specific field from parallel agent @@ -347,6 +347,13 @@ def _add_agent_input( ) -> None: """Add a regular agent output reference to context. + Supported behaviors: + - ``agent_name`` or ``agent_name.output`` copies the full output. + - ``agent_name.output.field[.subfield...]`` projects nested fields. + - ``agent_name.field[.subfield...]`` projects the same nested fields via shorthand. + - Optional refs (``?``) skip missing leaves/intermediate paths. + - Projected leaves are deep-copied to avoid mutation aliasing. + Args: ctx: The context dictionary to update. agent_name: The name of the agent. @@ -359,6 +366,7 @@ def _add_agent_input( agent_output = self.agent_outputs[agent_name] is_dict_output = isinstance(agent_output, dict) + # Seed ctx[agent_name] so optional-skip returns still leave a stub entry. # Initialise ``output`` to an empty dict for dict-shaped outputs (so # subsequent field-writes have somewhere to land) or to ``None`` for # scalar/list outputs (which are assigned whole below). @@ -367,48 +375,53 @@ def _add_agent_input( elif "output" not in ctx[agent_name]: ctx[agent_name]["output"] = {} if is_dict_output else None - if not remaining_parts: - # Just agent_name - copy entire output - ctx[agent_name]["output"] = ( - copy.deepcopy(agent_output) if is_dict_output else (copy.copy(agent_output)) - ) - elif len(remaining_parts) == 1 and remaining_parts[0] == "output": - # agent_name.output - copy entire output + if not remaining_parts or remaining_parts == ["output"]: + # agent_name or agent_name.output — copy entire output ctx[agent_name]["output"] = ( copy.deepcopy(agent_output) if is_dict_output else (copy.copy(agent_output)) ) - elif len(remaining_parts) >= 2 and remaining_parts[0] == "output": - # agent_name.output.field - copy specific field. Only meaningful - # when the stored output is a dict; otherwise the field access - # is undefined and (when required) must raise. - field_name = remaining_parts[1] - if is_dict_output and field_name in agent_output: - # Ensure we have a dict to write the field into (the initial - # output slot above seeded ``None`` for non-dict outputs). - if not isinstance(ctx[agent_name]["output"], dict): - ctx[agent_name]["output"] = {} - ctx[agent_name]["output"][field_name] = agent_output[field_name] - elif not is_optional: - if not is_dict_output: + else: + # Resolve field path: + # agent_name.output.field[.subfield...] → path = [field, subfield, ...] + # agent_name.field[.subfield...] → path = [field, subfield, ...] (shorthand) + path = remaining_parts[1:] if remaining_parts[0] == "output" else remaining_parts + + if not is_dict_output: + if not is_optional: raise KeyError( - f"Cannot access field '{field_name}' on agent '{agent_name}': " + f"Cannot access field '{path[0]}' on agent '{agent_name}': " f"its output is a {type(agent_output).__name__}, not a dict" ) - raise KeyError(f"Missing output field '{field_name}' from agent '{agent_name}'") - elif len(remaining_parts) == 1 and remaining_parts[0] != "output": - # Shorthand format: agent_name.field -> agent_name.output.field - field_name = remaining_parts[0] - if is_dict_output and field_name in agent_output: - if not isinstance(ctx[agent_name]["output"], dict): - ctx[agent_name]["output"] = {} - ctx[agent_name]["output"][field_name] = agent_output[field_name] - elif not is_optional: - if not is_dict_output: + return + + # Traverse agent_output along path + value: Any = agent_output + for i, key in enumerate(path): + if isinstance(value, dict) and key in value: + value = value[key] + continue + if is_optional: + return + intermediate_path = ".".join(path[:i]) if i > 0 else agent_name + if not isinstance(value, dict): raise KeyError( - f"Cannot access field '{field_name}' on agent '{agent_name}': " - f"its output is a {type(agent_output).__name__}, not a dict" + f"Cannot access field '{key}' on agent '{agent_name}': " + f"intermediate value at '{intermediate_path}' is a " + f"{type(value).__name__}, not a dict" ) - raise KeyError(f"Missing output field '{field_name}' from agent '{agent_name}'") + raise KeyError(f"Missing output field '{'.'.join(path)}' from agent '{agent_name}'") + + # Write the resolved leaf value into ctx at the nested output path, + # creating intermediate dicts as needed. deepcopy matches the + # whole-output copy semantics above and prevents mutation aliasing. + if not isinstance(ctx[agent_name]["output"], dict): + ctx[agent_name]["output"] = {} + target = ctx[agent_name]["output"] + for key in path[:-1]: + if key not in target or not isinstance(target[key], dict): + target[key] = {} + target = target[key] + target[path[-1]] = copy.deepcopy(value) def _add_parallel_group_input( self, ctx: dict[str, Any], group_name: str, remaining_parts: list[str], is_optional: bool diff --git a/tests/test_config/test_validator.py b/tests/test_config/test_validator.py index 255eeae6..234d1bde 100644 --- a/tests/test_config/test_validator.py +++ b/tests/test_config/test_validator.py @@ -309,6 +309,79 @@ def test_optional_reference_to_unknown_agent_warns(self) -> None: warnings = validate_workflow_config(config) assert any("unknown" in w for w in warnings) + def test_valid_nested_explicit_output_reference(self) -> None: + """Nested explicit refs (agent.output.foo.bar) pass validation.""" + config = WorkflowConfig( + workflow=WorkflowDef(name="test", entry_point="producer"), + agents=[ + AgentDef( + name="producer", + model="gpt-4", + prompt="Produce", + routes=[RouteDef(to="consumer")], + ), + AgentDef( + name="consumer", + model="gpt-4", + prompt="Consume", + input=["producer.output.foo.bar"], + routes=[RouteDef(to="$end")], + ), + ], + ) + + validate_workflow_config(config) + + def test_valid_nested_shorthand_output_reference(self) -> None: + """Nested shorthand refs (agent.foo.bar) pass validation.""" + config = WorkflowConfig( + workflow=WorkflowDef(name="test", entry_point="producer"), + agents=[ + AgentDef( + name="producer", + model="gpt-4", + prompt="Produce", + routes=[RouteDef(to="consumer")], + ), + AgentDef( + name="consumer", + model="gpt-4", + prompt="Consume", + input=["producer.foo.bar"], + routes=[RouteDef(to="$end")], + ), + ], + ) + + validate_workflow_config(config) + + def test_invalid_nested_ref_message_mentions_explicit_and_shorthand_forms(self) -> None: + """Malformed nested refs should mention both supported nested forms.""" + config = WorkflowConfig( + workflow=WorkflowDef(name="test", entry_point="producer"), + agents=[ + AgentDef( + name="producer", + model="gpt-4", + prompt="Produce", + routes=[RouteDef(to="consumer")], + ), + AgentDef( + name="consumer", + model="gpt-4", + prompt="Consume", + input=["producer.output.foo..bar"], + routes=[RouteDef(to="$end")], + ), + ], + ) + + with pytest.raises(ConfigurationError) as exc_info: + validate_workflow_config(config) + msg = str(exc_info.value) + assert "agent_name.output.field[.subfield...]" in msg + assert "agent_name.field[.subfield...]" in msg + class TestToolValidation: """Tests for tool reference validation.""" @@ -810,6 +883,32 @@ def test_pg_kind_capture(self) -> None: assert m_err is not None assert m_err.group("pg_kind") == "errors" + @pytest.mark.parametrize( + "ref", + [ + # Legacy shapes (must still pass) + "agent.output", + "agent.output.field", + "agent.output?", + "agent.output.field?", + "workflow.input.param", + "workflow.input.param?", + "group.outputs.agent.field", + # New: nested explicit output + "agent.output.field.subfield", + "agent.output.field.subfield?", + "agent.output.a.b.c", + # New: shorthand + "agent.field", + "agent.field?", + "agent.field.subfield", + "agent.field.subfield?", + "agent.foo.bar.baz", + ], + ) + def test_pattern_accepts_nested_and_shorthand_shapes(self, ref: str) -> None: + assert INPUT_REF_PATTERN.match(ref) is not None + @pytest.mark.parametrize( "ref,expected_parallel", [ @@ -823,32 +922,17 @@ def test_pg_kind_capture(self) -> None: ("group.outputs?", "group"), ], ) - def test_pattern_accepts_new_shapes(self, ref: str, expected_parallel: str) -> None: + def test_pattern_accepts_parallel_shapes(self, ref: str, expected_parallel: str) -> None: match = INPUT_REF_PATTERN.match(ref) assert match is not None, f"{ref!r} should match INPUT_REF_PATTERN" assert match.group("parallel") == expected_parallel - @pytest.mark.parametrize( - "ref", - [ - "agent.output", - "agent.output.field", - "agent.output?", - "agent.output.field?", - "workflow.input.param", - "workflow.input.param?", - ], - ) - def test_pattern_still_accepts_legacy_shapes(self, ref: str) -> None: - assert INPUT_REF_PATTERN.match(ref) is not None - @pytest.mark.parametrize( "ref", [ "workflow.input", # bare workflow.input no longer accepted "agent", - "agent.foo", - "group.bogus", + "group.outputs.agent.field.subfield", # deeper parallel projection not supported ], ) def test_pattern_rejects_invalid_shapes(self, ref: str) -> None: diff --git a/tests/test_engine/test_context.py b/tests/test_engine/test_context.py index 6c8c97f8..1daa98cd 100644 --- a/tests/test_engine/test_context.py +++ b/tests/test_engine/test_context.py @@ -352,6 +352,161 @@ def test_explicit_mode_human_gate_unchanged(self) -> None: assert agent_ctx["workflow"]["input"] == {} + def test_explicit_mode_nested_field_shorthand(self) -> None: + """Shorthand agent.foo.bar exposes output.foo.bar; sibling keys are excluded.""" + ctx = WorkflowContext() + ctx.store( + "gate", + {"selected": "go", "additional_input": {"answer": "README.md", "other": "x"}}, + ) + + agent_ctx = ctx.build_for_agent( + "next", + ["gate.additional_input.answer"], + mode="explicit", + ) + + assert agent_ctx["gate"]["output"]["additional_input"]["answer"] == "README.md" + # Only the declared leaf is present — sibling 'other' is excluded + assert "other" not in agent_ctx["gate"]["output"]["additional_input"] + # Top-level 'selected' is also excluded (not declared) + assert "selected" not in agent_ctx["gate"]["output"] + + def test_explicit_mode_nested_field_output_prefix(self) -> None: + """agent.output.foo.bar exposes the same path as the shorthand form.""" + ctx = WorkflowContext() + ctx.store("gate", {"selected": "go", "additional_input": {"answer": "README.md"}}) + + agent_ctx = ctx.build_for_agent( + "next", + ["gate.output.additional_input.answer"], + mode="explicit", + ) + + assert agent_ctx["gate"]["output"]["additional_input"]["answer"] == "README.md" + assert "selected" not in agent_ctx["gate"]["output"] + + def test_explicit_mode_nested_field_multiple_declarations(self) -> None: + """Two declarations into the same parent dict both land without overwriting each other.""" + ctx = WorkflowContext() + ctx.store("agent1", {"data": {"x": 1, "y": 2, "z": 3}}) + + agent_ctx = ctx.build_for_agent( + "agent2", + ["agent1.data.x", "agent1.data.y"], + mode="explicit", + ) + + assert agent_ctx["agent1"]["output"]["data"]["x"] == 1 + assert agent_ctx["agent1"]["output"]["data"]["y"] == 2 + assert "z" not in agent_ctx["agent1"]["output"]["data"] + + def test_explicit_mode_nested_leaf_projection_excludes_sibling(self) -> None: + """Declaring a leaf path copies only that leaf, excluding sibling keys.""" + ctx = WorkflowContext() + ctx.store("a", {"foo": {"bar": 1, "baz": 2}}) + + agent_ctx = ctx.build_for_agent( + "next", + ["a.output.foo.bar"], + mode="explicit", + ) + + assert agent_ctx["a"]["output"]["foo"]["bar"] == 1 + assert "baz" not in agent_ctx["a"]["output"]["foo"] + + def test_explicit_mode_parent_decl_then_child_keeps_all_siblings(self) -> None: + """Declaring parent then child keeps full parent dict in projected output.""" + ctx = WorkflowContext() + ctx.store("a", {"foo": {"bar": 1, "baz": 2}}) + + agent_ctx = ctx.build_for_agent( + "next", + ["a.foo", "a.foo.bar"], + mode="explicit", + ) + + assert agent_ctx["a"]["output"]["foo"] == {"bar": 1, "baz": 2} + + def test_explicit_mode_child_decl_then_parent_overwrites(self) -> None: + """Declaring child then parent ends with the full parent dict.""" + ctx = WorkflowContext() + ctx.store("a", {"foo": {"bar": 1, "baz": 2}}) + + agent_ctx = ctx.build_for_agent( + "next", + ["a.foo.bar", "a.foo"], + mode="explicit", + ) + + assert agent_ctx["a"]["output"]["foo"] == {"bar": 1, "baz": 2} + + def test_explicit_mode_deep_three_level_projection(self) -> None: + """Nested projections beyond two levels keep only the declared deep leaf.""" + ctx = WorkflowContext() + ctx.store("a", {"x": {"y": {"z": 42, "other": 99}}}) + + agent_ctx = ctx.build_for_agent( + "next", + ["a.x.y.z"], + mode="explicit", + ) + + assert agent_ctx["a"]["output"]["x"]["y"]["z"] == 42 + assert "other" not in agent_ctx["a"]["output"]["x"]["y"] + + def test_explicit_mode_nested_field_missing_required_raises(self) -> None: + """Missing required nested path raises KeyError.""" + ctx = WorkflowContext() + ctx.store("gate", {"additional_input": {}}) + + with pytest.raises(KeyError, match="Missing output field"): + ctx.build_for_agent( + "next", + ["gate.additional_input.answer"], + mode="explicit", + ) + + def test_explicit_mode_nested_field_optional_missing_skipped(self) -> None: + """Optional missing nested path is silently skipped.""" + ctx = WorkflowContext() + ctx.store("gate", {"additional_input": {}}) + + agent_ctx = ctx.build_for_agent( + "next", + ["gate.additional_input.answer?"], + mode="explicit", + ) + + # Stub is always created; optional path not written so output dict is empty. + assert agent_ctx["gate"] == {"output": {}} + + def test_explicit_mode_nested_intermediate_not_dict_raises(self) -> None: + """Traversing through a non-dict intermediate value raises KeyError.""" + ctx = WorkflowContext() + ctx.store("agent1", {"foo": "scalar_not_a_dict"}) + + with pytest.raises(KeyError, match="intermediate value"): + ctx.build_for_agent( + "agent2", + ["agent1.foo.bar"], + mode="explicit", + ) + + def test_explicit_mode_deep_missing_mid_path_error_includes_path(self) -> None: + """Mid-path non-dict errors should include a non-empty intermediate path label.""" + ctx = WorkflowContext() + ctx.store("a", {"foo": "not_a_dict"}) + + with pytest.raises(KeyError, match="intermediate value") as exc_info: + ctx.build_for_agent( + "next", + ["a.foo.bar"], + mode="explicit", + ) + + assert "'a'" in str(exc_info.value) + class TestWorkflowContextOptionalDeps: """Tests for optional dependencies with ? suffix."""