fix(context): support nested explicit input projection

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<agent>[a-zA-Z_][a-zA-Z0-9_]*)\.output(?:\.(?P<field>[a-zA-Z_][a-zA-Z0-9_]*))?|"
+    r"(?P<agent>[a-zA-Z_][a-zA-Z0-9_]*)\.output(?:\.(?P<field>[a-zA-Z_][a-zA-Z0-9_]*)(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*)?|"
     r"(?P<parallel>[a-zA-Z_][a-zA-Z0-9_]*)\.(?P<pg_kind>outputs|errors)(?:\.(?P<pg_agent>[a-zA-Z_][a-zA-Z0-9_]*)(?:\.(?P<pg_field>[a-zA-Z_][a-zA-Z0-9_]*))?)?|"
-    r"workflow\.input\.(?P<input>[a-zA-Z_][a-zA-Z0-9_]*)"
+    r"workflow\.input\.(?P<input>[a-zA-Z_][a-zA-Z0-9_]*)|"
+    r"(?!workflow\b)(?![a-zA-Z_][a-zA-Z0-9_]*\.(?:outputs|errors)(?:\.|$))"
+    r"(?P<shorthand>[a-zA-Z_][a-zA-Z0-9_]*)\.(?P<sh_field>[a-zA-Z_][a-zA-Z0-9_]*)(?:\.[a-zA-Z_][a-zA-Z0-9_]*)*"
     r")(?P<optional>\?)?$"
 )
 
@@ -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", "<field>", ...]
+            # 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:

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

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: