Give validate and run one shared notion of an action (#241)

src/orchestrator/control_systems/hybrid_control_system.py

@@ -7,6 +7,7 @@
 from datetime import datetime
 
 from .model_based_control_system import ModelBasedControlSystem
+from ..core.actions import BUILTIN_ACTION_HANDLERS
 from ..core.task import Task
 from ..core.action_loop_task import ActionLoopTask
 from ..core.expressions import (
@@ -161,54 +162,25 @@ async def _execute_task_impl(self, task: Task, context: Dict[str, Any]) -> Any:
             logger.debug("Routing to tool handler: %s", tool_name)
             return await self._handle_tool_execution(task, tool_name, context)
 
-        # Check if this is a control flow operation
-        if action_str == "control_flow":
-            return await self._handle_control_flow(task, context)
+        # Actions the runtime runs itself, dispatched off the shared registry
+        # in core/actions.py. Driving dispatch from that mapping -- rather than
+        # keeping a parallel chain of `if action_str == ...` branches -- is
+        # what stops it from drifting out of step with the validator, which
+        # accepts exactly the same names (#241).
+        handler_name = BUILTIN_ACTION_HANDLERS.get(action_str.strip())
+        if handler_name:
+            logger.debug("Routing to built-in action handler: %s", handler_name)
+            return await getattr(self, handler_name)(task, context)
 
-        # Check if this is a simple echo/print operation
+        # The remaining two families are matched by pattern, not by name: an
+        # action like "echo hello" or "write the following content to report.md"
+        # is prose, so there is no name to register.
         if self._is_echo_operation(action_str):
             return await self._handle_echo_operation(task, context)
 
-        # Check if this is a file operation
-        if self._is_file_operation(action_str) or action_str == "filesystem":
+        if self._is_file_operation(action_str):
             return await self._handle_file_operation(task, context)
 
-        # Check if this is a data processing operation
-        if action_str == "process":
-            return await self._handle_data_processing(task, context)
-
-        # Check if this is a validation operation
-        if action_str == "validate":
-            return await self._handle_validation(task, context)
-
-        # Check if this is a loop completion marker
-        if action_str == "loop_complete":
-            return await self._handle_loop_complete(task, context)
-
-        # Check if this is a capture result marker
-        if action_str == "capture_result":
-            return await self._handle_capture_result(task, context)
-
-        # Check if this is a condition evaluation
-        if action_str == "evaluate_condition":
-            return await self._handle_evaluate_condition(task, context)
-
-        # Check if this is a parallel queue execution
-        if action_str == "create_parallel_queue":
-            return await self._handle_create_parallel_queue(task, context)
-
-        # Check if this is an action loop
-        if action_str == "action_loop":
-            return await self._handle_action_loop(task, context)
-
-        # Check if this is text analysis
-        if action_str == "analyze_text" or action_str == "analyze":
-            return await self._handle_analyze_text(task, context)
-
-        # Check if this is text generation
-        if action_str == "generate_text" or action_str == "generate":
-            return await self._handle_generate_text(task, context)
-
         # Otherwise use model-based execution
         return await super()._execute_task_impl(task, context)
 

src/orchestrator/control_systems/model_based_control_system.py

@@ -6,6 +6,7 @@
 import re
 from typing import Any, Dict, List, Optional
 
+from ..core.actions import STRUCTURED_ACTIONS
 from ..core.control_system import ControlSystem
 from ..core.pipeline import Pipeline
 from ..core.task import Task
@@ -15,14 +16,6 @@
 
 logger = logging.getLogger(__name__)
 
-# Every other action in this project is spelled with underscores
-# (`generate_text`, `analyze_text`, `evaluate_condition`, `loop_complete`).
-# `generate-structured` was the lone hyphenated one, so `generate_structured`
-# fell through to the natural-language branch below, which turns an unknown
-# action into a *prompt*: the step returned a sentence instead of an object and
-# still reported success. Both spellings dispatch here.
-STRUCTURED_ACTIONS = frozenset({"generate-structured", "generate_structured"})
-
 
 class ModelBasedControlSystem(ControlSystem):
     """Control system that executes tasks using real AI models."""

src/orchestrator/core/actions.py

@@ -0,0 +1,80 @@
+"""The actions the runtime executes itself, without a `tool:`.
+
+A step names either a tool and an operation on it::
+
+    tool: filesystem
+    action: read
+
+or an action the runtime handles directly::
+
+    action: generate
+
+Both the executor and the validator need to know which names fall in the
+second group, and they used to know separately. `ToolValidator` treated a
+step's `action:` as a tool name whenever the step had no `tool:` key, so a
+model step was reported as ``Tool 'generate' not found in registry`` while the
+executor ran the very same step correctly -- #241, `validate` and `run`
+disagreeing about the same document.
+
+This module is the single source of truth. `HybridControlSystem` *dispatches*
+off `BUILTIN_ACTION_HANDLERS`, so the mapping cannot claim an action the
+executor does not actually run, and the validator accepts exactly the names in
+it. Adding a runtime action means adding one entry here; both sides follow.
+
+Not covered here are the natural-language action families -- ``"echo ..."``,
+``"write the following content to report.md"`` -- which are matched by pattern
+rather than by name. Those are `HybridControlSystem._is_echo_operation` and
+`_is_file_operation`.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, FrozenSet
+
+# Action name -> the `HybridControlSystem` method that runs it.
+#
+# Values are method *names* rather than functions because the handlers are
+# bound methods on the control system; the dispatcher resolves them per call.
+BUILTIN_ACTION_HANDLERS: Dict[str, str] = {
+    "control_flow": "_handle_control_flow",
+    # `file` and `filesystem` are the two spellings the executor has always
+    # accepted for the filesystem handler.
+    "file": "_handle_file_operation",
+    "filesystem": "_handle_file_operation",
+    "process": "_handle_data_processing",
+    "validate": "_handle_validation",
+    "loop_complete": "_handle_loop_complete",
+    "capture_result": "_handle_capture_result",
+    "evaluate_condition": "_handle_evaluate_condition",
+    "create_parallel_queue": "_handle_create_parallel_queue",
+    "action_loop": "_handle_action_loop",
+    "analyze": "_handle_analyze_text",
+    "analyze_text": "_handle_analyze_text",
+    "generate": "_handle_generate_text",
+    "generate_text": "_handle_generate_text",
+}
+
+# Structured generation is handled one level down, by
+# `ModelBasedControlSystem`, so it has no entry above -- but it is still an
+# action the runtime executes without a tool, and the validator must accept it.
+#
+# Every other action here uses underscores. `generate-structured` was the lone
+# hyphenated one, and for a long time the only spelling that dispatched: the
+# underscore fell through to the natural-language branch, which turns an
+# unrecognised action into a *prompt*, so the step returned a sentence instead
+# of an object and still reported success. Both spellings are supported.
+STRUCTURED_ACTIONS: FrozenSet[str] = frozenset(
+    {"generate-structured", "generate_structured"}
+)
+
+#: Every action name the runtime executes without a `tool:`.
+BUILTIN_ACTIONS: FrozenSet[str] = frozenset(BUILTIN_ACTION_HANDLERS) | STRUCTURED_ACTIONS
+
+
+def is_builtin_action(action: str) -> bool:
+    """Whether `action` names something the runtime executes without a tool.
+
+    Comparison is case-insensitive and ignores surrounding whitespace, matching
+    the executor, which lowercases a task's action before dispatching on it.
+    """
+    return action.strip().lower() in BUILTIN_ACTIONS

src/orchestrator/validation/tool_validator.py

@@ -4,6 +4,7 @@
 from typing import Any, Dict, List, Optional, Set, Type, Union
 from dataclasses import dataclass
 
+from ..core.actions import is_builtin_action
 from ..tools.base import Tool, ToolRegistry, default_registry
 from .template_validator import TemplateValidationError
 
@@ -139,8 +140,22 @@ def validate_pipeline_tools(self, pipeline_def: Dict[str, Any]) -> ToolValidatio
 
             # Check tool availability
             tool_available = self._check_tool_availability(tool_name)
+
+            # A step with no `tool:` names an action, and some actions are run
+            # by the runtime itself rather than by a tool -- `generate`,
+            # `evaluate_condition`, `loop_complete` and the rest of
+            # core.actions.BUILTIN_ACTIONS. Looking those up as tools is what
+            # made `validate` reject a model pipeline that `run` executes
+            # perfectly well: "Tool 'generate' not found in registry" (#241).
+            #
+            # The tool lookup still comes first, so the legacy single-field
+            # form (`action: filesystem`, naming the tool itself) keeps its
+            # full parameter validation.
+            if not tool_available and "tool" not in step and is_builtin_action(tool_name):
+                continue
+
             tool_availability[tool_name] = tool_available
-            
+
             if not tool_available:
                 if self.allow_unknown_tools:
                     warnings.append(ToolValidationError(

tests/test_builtin_actions.py

@@ -0,0 +1,176 @@
+"""The built-in action registry, and the two consumers that must agree on it.
+
+`validate` and `run` used to hold separate notions of what an action is. The
+validator treated a step's `action:` as a tool name whenever the step had no
+`tool:` key, so `action: generate` was rejected as a missing *tool* while the
+executor ran the same step correctly (#241).
+
+`core.actions` is now the single source of truth. These tests pin both
+directions of that: the executor dispatches every name in the registry to a
+handler that exists, and the validator accepts every name in the registry.
+"""
+
+import asyncio
+
+import pytest
+
+from orchestrator.core.actions import (
+    BUILTIN_ACTION_HANDLERS,
+    BUILTIN_ACTIONS,
+    STRUCTURED_ACTIONS,
+    is_builtin_action,
+)
+from orchestrator.control_systems.hybrid_control_system import HybridControlSystem
+from orchestrator.models.model_registry import ModelRegistry
+from orchestrator.validation.tool_validator import ToolValidator
+from tests.test_infrastructure import MockTestModel
+
+pytestmark = [pytest.mark.contract]
+
+
+@pytest.fixture
+def control_system():
+    registry = ModelRegistry()
+    registry.register_model(MockTestModel())
+    return HybridControlSystem(model_registry=registry)
+
+
+@pytest.fixture
+def validator():
+    """A strict validator: unknown tools are errors, not warnings."""
+    return ToolValidator(allow_unknown_tools=False)
+
+
+# ---------------------------------------------------------------------------
+# the registry cannot promise what the executor cannot deliver
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("action", sorted(BUILTIN_ACTION_HANDLERS))
+def test_every_registered_action_has_a_real_handler(action, control_system):
+    """Dispatch is driven off this mapping, so a bad entry is an AttributeError.
+
+    Catching it here names the offending action instead of surfacing as a
+    mystery failure inside whichever pipeline happened to use it.
+    """
+    handler_name = BUILTIN_ACTION_HANDLERS[action]
+    handler = getattr(control_system, handler_name, None)
+
+    assert handler is not None, (
+        f"action {action!r} is registered to {handler_name!r}, "
+        f"which does not exist on HybridControlSystem"
+    )
+    assert callable(handler), f"{handler_name!r} is not callable"
+
+
+def test_structured_actions_are_builtin_but_handled_one_level_down():
+    """They have no handler entry: ModelBasedControlSystem runs them.
+
+    They must still be in BUILTIN_ACTIONS or the validator would reject them
+    as unknown tools -- the original #241 failure, in a different action.
+    """
+    assert STRUCTURED_ACTIONS <= BUILTIN_ACTIONS
+    assert not (STRUCTURED_ACTIONS & set(BUILTIN_ACTION_HANDLERS))
+
+
+def test_is_builtin_action_matches_the_executors_normalisation():
+    """The executor lowercases and strips before dispatching; so must this."""
+    assert is_builtin_action("generate")
+    assert is_builtin_action("  GENERATE  ")
+    assert is_builtin_action("Generate_Text")
+    assert not is_builtin_action("summarise the report in three sentences")
+    assert not is_builtin_action("")
+
+
+# ---------------------------------------------------------------------------
+# the validator accepts exactly what the executor runs
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("action", sorted(BUILTIN_ACTIONS))
+def test_validator_accepts_every_builtin_action(action, validator):
+    """No built-in action may be reported as a missing tool."""
+    result = validator.validate_pipeline_tools(
+        {"steps": [{"id": "step_one", "action": action, "parameters": {}}]}
+    )
+
+    unknown = [e for e in result.errors if e.error_type == "unknown_tool"]
+    assert not unknown, (
+        f"built-in action {action!r} was rejected as a tool: "
+        f"{[e.message for e in unknown]}"
+    )
+
+
+def test_validator_still_rejects_an_action_that_is_neither_tool_nor_builtin(validator):
+    """The fix must not turn the validator into a rubber stamp."""
+    result = validator.validate_pipeline_tools(
+        {"steps": [{"id": "step_one", "action": "definitely_not_a_real_thing"}]}
+    )
+
+    assert [e for e in result.errors if e.error_type == "unknown_tool"], (
+        "an unknown action still has to be reported"
+    )
+
+
+def test_validator_still_validates_the_legacy_single_field_form(validator):
+    """`action: filesystem` names the tool itself and keeps full validation.
+
+    `filesystem` is both a registered tool and a built-in action name. The tool
+    lookup has to win, or this form would silently lose its parameter checks.
+    """
+    result = validator.validate_pipeline_tools(
+        {"steps": [{"id": "step_one", "action": "filesystem", "parameters": {}}]}
+    )
+
+    assert "filesystem" in result.tool_availability, (
+        "the legacy form must still be resolved as a tool, not skipped as an action"
+    )
+    assert result.tool_availability["filesystem"] is True
+
+
+def test_a_tool_step_is_unaffected(validator):
+    """The two-field form still resolves `tool:`, never `action:`."""
+    result = validator.validate_pipeline_tools(
+        {
+            "steps": [
+                {
+                    "id": "step_one",
+                    "tool": "filesystem",
+                    "action": "read",
+                    "parameters": {"path": "x.txt"},
+                }
+            ]
+        }
+    )
+
+    assert "filesystem" in result.tool_availability
+    assert "read" not in result.tool_availability, (
+        "`read` is an operation on the tool, not a tool to look up"
+    )
+
+
+# ---------------------------------------------------------------------------
+# end to end: the two agree on the same document
+# ---------------------------------------------------------------------------
+
+def test_builtin_action_dispatches_away_from_the_model(control_system):
+    """A built-in action must reach its handler, not the prompt fallback.
+
+    The executor's last resort turns an unrecognised action into natural
+    language for the model. That is how `generate_structured` returned a
+    sentence instead of an object for as long as it did, so the boundary is
+    pinned: `evaluate_condition` is a marker action with no model involvement,
+    and reaching the model would produce prose instead of a verdict.
+    """
+    from orchestrator.core.task import Task
+
+    task = Task(
+        id="check",
+        name="check",
+        action="evaluate_condition",
+        parameters={"condition": "1 == 1"},
+    )
+    result = asyncio.run(control_system._execute_task_impl(task, {}))
+
+    assert isinstance(result, dict), f"expected the handler's verdict, got {type(result)}"
+    assert "result" in result or "condition" in result, (
+        f"this does not look like _handle_evaluate_condition output: {result}"
+    )