Skip to main content

Executor System

Executors define how a task runs once it has been dispatched to a target. A task supplies the unit of work, the runtime prepares the payload, and the executor is responsible for invoking that payload in a compatible way.

Core Concept​

Every executor implements an internal execution hook:

async def _execute(self, task: BaseTask) -> None:
    ...

The task itself is responsible for orchestration concerns such as input validation, event emission, error handling, and incrementing the run counter. The target owns dispatch and waiting. The executor focuses on execution only.

Command-based executors run their work through the target's channel rather than spawning processes directly. A channel-driven executor renders the payload with task.runtime, then calls task.target.run_command(...) and drives the returned ChannelProcess. Because the channel abstracts where a command runs, the same executor works on a local target or a remote SSH target with nothing installed on the remote side. See Targets are agentless channels.

In-process executors (PythonExecExecutor, PythonFunctionExecutor) run Python inside the orchestrator instead of through a channel, so they are only valid on local, in-process targets.

Contract​

  • Declare compatible runtime types via runtimes
  • Implement _execute(), not execute()
  • execute() is the public final entry point: it creates task.side_artifacts_dir through the target's channel (await task.target.mkdir(...)), runs ExecutorMiddleware, calls _execute(), then collects side artifacts. Collection is channel-based: collect_side_artifacts() lists the directory with list_dir and pulls each file/folder back to a local temp dir with get_file, so it works on local and remote targets. It is best-effort (failures are logged, never masking the task) and skips files over HorusRuntimeSettings.MAX_SIDE_ARTIFACT_BYTES. See Side Artifacts
  • Use kind: str as the registry discriminator

BaseTask validates runtime compatibility during model validation. If a task pairs an executor with an unsupported runtime, Horus raises IncompatibleRuntimeError.

Base Executor​

All executors inherit from BaseExecutor:

class BaseExecutor(AutoRegistry, entry_point="executor"):
    registry_key: ClassVar[str] = "kind"
    kind: str
    kind_name: ClassVar[str] = "Executor"
    kind_description: ClassVar[str] = _("Base executor")
    runtimes: ClassVar[tuple[type[BaseRuntime], ...]] = (BaseRuntime,)

    @final
    async def execute(self, task: BaseTask) -> None:
        """
        Public entry point wrapped by executor middleware.
        """
        ...

    @abstractmethod
    async def _execute(self, task: BaseTask) -> None:
        """
        Subclass hook that performs execution.
        """

execute() wraps _execute() in ExecutorMiddleware.call_with_middleware(...). See Middleware Overview.

Kind metadata​

Executors may expose kind_name and kind_description ClassVars to provide human-friendly text for registries and UIs. For kind_description, prefer using your plugin's own translator created via make_translator (commonly aliased to _(...)) so descriptions are translatable without depending on horus_runtime's internal translator.

Built-in Executors​

  • ShellExecutor: runs a CommandRuntime through the target's channel (task.target.run_command(...)), so it works on local and remote targets. It exposes HORUS_SIDE_ARTIFACTS_DIR in the command environment, raises TaskExecutionError on a non-zero exit, and on cancellation group-kills the process via the ChannelProcess handle.
  • PythonFunctionExecutor: executes a PythonFunctionRuntime in-process by calling the wrapped Python function directly (local targets only)
  • PythonExecExecutor: executes a PythonCodeStringRuntime in-process using exec() (local targets only)

Example​

from horus_builtin.executor.shell import ShellExecutor
from horus_builtin.runtime.command import CommandRuntime
from horus_builtin.target.local import LocalTarget
from horus_builtin.task.horus_task import HorusTask

task = HorusTask(
    name="echo",
    target=LocalTarget(),
    executor=ShellExecutor(),
    runtime=CommandRuntime(command="echo 'hello world'"),
)

Python Code Execution Example​

from horus_builtin.executor.python_exec import PythonExecExecutor
from horus_builtin.runtime.python_string import PythonCodeStringRuntime
from horus_builtin.target.local import LocalTarget
from horus_builtin.task.horus_task import HorusTask

task = HorusTask(
    name="python_step",
    target=LocalTarget(),
    executor=PythonExecExecutor(),
    runtime=PythonCodeStringRuntime(
        code="with open('hello.txt', 'w', encoding='utf-8') as f:\n    f.write('hello\\n')"
    ),
)

PythonExecExecutor executes the runtime's code string in-process with exec(). The execution scope includes ctx, task, and HORUS_SIDE_ARTIFACTS_DIR (a string path to task.side_artifacts_dir). See Side Artifacts.

Registering Custom Executors​

To register executor plugins, expose them through:

[project.entry-points."horus.executor"]

For more details, refer to the Auto-Registry documentation.