refactor(config): split step definitions into a discriminated union

CHANGELOG.md

@@ -41,6 +41,33 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   without replacing tool-based output, weakening schema validation, or
   changing authored system prompts.
 
+### Changed
+
+- **Step definitions split into concrete models with a discriminated union**
+  (#517) — the single monolithic `AgentDef` that carried every step type's
+  fields is now a set of focused Pydantic models: `AgentDef` (provider-backed
+  LLM agents only), `HumanGateStepDef`, `QuestionsStepDef`, `ScriptStepDef`,
+  `MCPStepDef`, `WaitStepDef`, `SetStepDef`, `TerminateStepDef`, and
+  `WorkflowStepDef`, united by the static `StepDef` union discriminated on
+  `type`. Every model owns exactly the fields meaningful for its kind with
+  `extra="forbid"`, so a misplaced field is rejected next to its step instead
+  of being silently ignored, and the published JSON Schema now exposes a
+  `oneOf` + `discriminator` mapping with per-variant
+  `additionalProperties: false` for editors and tooling. Compatibility:
+  workflow YAML is unchanged — an omitted `type` or an explicit `type: null`
+  still loads as an LLM agent (now canonicalized to `type: "agent"` at parse
+  time) — and `AgentDef(...)` keeps working for LLM agents, including
+  `AgentDef(type=None, ...)`. What changed: constructing a non-LLM step
+  programmatically via `AgentDef(type="script", ...)` no longer works — use
+  the named step class (e.g. `ScriptStepDef(...)`) — and a field that belongs
+  to a different step type now fails with Pydantic's standard
+  `extra_forbidden` error rather than the previous custom
+  `"<type> agents cannot have '<field>'"` messages. Human gates additionally
+  lost the inert `model` field and can no longer appear as a `for_each`
+  inline agent (concurrent iterations would compete for one interactive
+  channel; route to a gate from the group's `routes:` instead). All classes
+  are exported from `conductor.config` and `conductor.config.schema`.
+
 ## [0.1.37](https://github.com/microsoft/conductor/compare/v0.1.36...v0.1.37) - 2026-09-09
 
 ### Added

src/conductor/cli/app.py

@@ -968,6 +968,7 @@ def show(
 
     try:
         from conductor.config.loader import load_config as load_workflow_config
+        from conductor.config.schema import RoutableStepBase
 
         config = load_workflow_config(workflow_path)
     except Exception as e:
@@ -1015,9 +1016,12 @@ def show(
     agent_table.add_column("Routes")
 
     for agent in config.agents:
-        agent_type = agent.type or "agent"
-        routes = ", ".join(r.to + (f" (when {r.when})" if r.when else "") for r in agent.routes)
-        agent_table.add_row(agent.name, agent_type, agent.description or "-", routes or "-")
+        agent_routes = (
+            ", ".join(r.to + (f" (when {r.when})" if r.when else "") for r in agent.routes)
+            if isinstance(agent, RoutableStepBase)
+            else ""
+        )
+        agent_table.add_row(agent.name, agent.type, agent.description or "-", agent_routes or "-")
 
     # Include parallel groups
     for pg in config.parallel:

src/conductor/cli/plugin.py

@@ -228,6 +228,7 @@ def _list_enabled_plugins(config, workflow_path: Path, sources) -> None:  # noqa
     different builds, so they are reported — and their component counts
     computed — separately rather than one silently standing in for both.
     """
+    from conductor.config.schema import AgentDef, StepDef
     from conductor.plugins.errors import PluginError
     from conductor.plugins.manifest import PluginFlavor
     from conductor.plugins.registry import resolve_plugins
@@ -246,8 +247,8 @@ def _list_enabled_plugins(config, workflow_path: Path, sources) -> None:  # noqa
     groups: dict[tuple[tuple[tuple[str, bool, bool, bool], ...], PluginFlavor | None], list[str]]
     groups = {}
 
-    def _record(name: str, agent) -> None:  # noqa: ANN001
-        if agent.type not in (None, "agent"):
+    def _record(name: str, agent: StepDef) -> None:
+        if not isinstance(agent, AgentDef):
             return
         entries = agent.plugins if agent.plugins is not None else config.workflow.runtime.plugins
         if not entries:

src/conductor/cli/run.py

@@ -25,6 +25,7 @@
 from rich.text import Text
 
 from conductor.config.loader import load_config
+from conductor.config.schema import AgentDef
 from conductor.console import MarkupFreeConsole, join, make_console, styled
 from conductor.engine.workflow import ExecutionPlan, WorkflowEngine
 from conductor.exceptions import WorkflowTerminated
@@ -2409,7 +2410,9 @@ async def run_workflow_async(
         plugin_marketplaces = await _prefetch_plugin_sources(config, workflow_path)
 
         # Check if workflow uses multiple providers (has per-agent provider overrides)
-        uses_multi_provider = any(agent.provider is not None for agent in config.agents)
+        uses_multi_provider = any(
+            isinstance(agent, AgentDef) and agent.provider is not None for agent in config.agents
+        )
 
         if uses_multi_provider:
             verbose_log("Multi-provider mode: agents use different providers", style="cyan")
@@ -2801,7 +2804,6 @@ def _find_agent(self, name: str) -> Any:
             return next((a for a in self.config.agents if a.name == name), None)
 
     # Use a real WorkflowEngine but with a mock provider
-    from conductor.config.schema import AgentDef
     from conductor.providers.base import AgentOutput, AgentProvider
 
     class _MockProvider(AgentProvider, abstract=True):

src/conductor/cli/validate.py

@@ -408,7 +408,9 @@ def display_validation_success(
     for_each_group_count = len(config.for_each)
 
     # Count conditional routes
-    conditional_route_count = sum(1 for a in config.agents for r in a.routes if r.when)
+    conditional_route_count = sum(
+        1 for a in config.agents for r in getattr(a, "routes", []) if r.when
+    )
 
     # Determine workflow patterns
     patterns = []
@@ -419,7 +421,7 @@ def display_validation_success(
     agent_names = [a.name for a in config.agents]
     has_loop = False
     for i, agent in enumerate(config.agents):
-        for route in agent.routes:
+        for route in getattr(agent, "routes", []):
             if route.to in agent_names:
                 target_idx = agent_names.index(route.to)
                 if target_idx <= i:
@@ -483,13 +485,14 @@ def display_validation_success(
         for agent in config.agents:
             agent_type = agent.type or "agent"
             model = (
-                agent.model
+                getattr(agent, "model", None)
                 or config.workflow.runtime.default_model
                 or Text.from_markup("[dim]default[/dim]")
             )
 
-            if agent.routes:
-                route_targets = [r.to for r in agent.routes]
+            routes = getattr(agent, "routes", [])
+            if routes:
+                route_targets = [r.to for r in routes]
                 routes_str = ", ".join(route_targets[:3])
                 if len(route_targets) > 3:
                     routes_str += f" (+{len(route_targets) - 3} more)"

src/conductor/config/__init__.py

@@ -16,14 +16,25 @@
     ContextConfig,
     DialogConfig,
     GateOption,
+    HumanGateStepDef,
     InputDef,
     LimitsConfig,
+    MCPStepDef,
     OutputField,
+    QuestionsStepDef,
+    RoutableStepBase,
     RouteDef,
     RuntimeConfig,
+    ScriptStepDef,
+    SetStepDef,
+    StepBase,
+    StepDef,
+    TerminateStepDef,
     ValidatorConfig,
+    WaitStepDef,
     WorkflowConfig,
     WorkflowDef,
+    WorkflowStepDef,
 )
 from conductor.config.validator import validate_workflow_config
 
@@ -40,13 +51,24 @@
     "DialogConfig",
     "GateOption",
     "InputDef",
+    "HumanGateStepDef",
     "LimitsConfig",
     "OutputField",
+    "MCPStepDef",
+    "QuestionsStepDef",
     "RouteDef",
     "RuntimeConfig",
+    "RoutableStepBase",
+    "ScriptStepDef",
+    "SetStepDef",
+    "StepBase",
+    "StepDef",
+    "TerminateStepDef",
     "ValidatorConfig",
     "WorkflowConfig",
     "WorkflowDef",
+    "WorkflowStepDef",
+    "WaitStepDef",
     # Validator
     "validate_workflow_config",
 ]