Task System

Tasks are the unit of work in Horus. A task binds together its input and output artifacts, execution variables, executor, and runtime.

Core Concept

Every task implements:

async def run(self) -> None:
    """
    Execute the task.
    """

Tasks also define completion and reset semantics:

def is_complete(self) -> bool: ...
def reset(self) -> None: ...

Contract

• run() is asynchronous
• is_complete() determines whether the task can be skipped
• reset() clears task outputs so the task can run again
• kind: str is the registry discriminator
• executor and runtime must be compatible
• interaction can carry a task-level runtime prompt transport

Runtime compatibility is validated automatically after model construction. An invalid executor/runtime pair raises IncompatibleRuntimeError.

Base Task

All tasks inherit from BaseTask:

class BaseTask(AutoRegistry, entry_point="task"):
    registry_key: ClassVar[str] = "kind"
    kind: str
    task_id: str | None = None
    name: str
    inputs: dict[str, BaseArtifact] = Field(default_factory=dict)
    outputs: dict[str, BaseArtifact] = Field(default_factory=dict)
    variables: dict[str, Any] = Field(default_factory=dict)
    executor: BaseExecutor
    runtime: BaseRuntime
    runs: int = 0
    skip_if_complete: bool = True
    interaction: BaseInteractionTransport | None = None

    @abstractmethod
    async def run(self) -> None:
        pass

    @abstractmethod
    def is_complete(self) -> bool:
        pass

    @abstractmethod
    def reset(self) -> None:
        pass

Built-in Tasks

• HorusTask: the standard task implementation for command-style execution
• FunctionTask: a code-first task that wraps a Python function and pairs it with PythonFunctionRuntime and PythonFunctionExecutor

HorusTask

HorusTask provides the default task behavior:

• Emits task start and completion events
• Validates that declared input artifacts exist before execution
• Delegates execution to the configured executor
• Raises TaskExecutionError on non-zero return codes
• Treats a task as complete when all declared output artifacts exist

If a task declares no outputs, HorusTask.is_complete() returns False, so the task always runs unless your workflow adds different logic.

FunctionTask

FunctionTask is the simplest way to build an in-memory workflow in Python:

@FunctionTask.task(wf)
def prepare_data() -> None:
    ...

The decorator creates a FunctionTask, wraps the function in a PythonFunctionRuntime, and registers the task in the workflow automatically.

It also defaults interaction to the built-in CLI transport, which makes interactive code-first tasks easy to author.

See FunctionTask for the full guide and examples.

Registering Custom Tasks

To register task plugins, expose them through:

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

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