Execution Plan API

plan

Execution plan primitive for multi-step agent workflows.

PlanStatus module-attribute

PlanStatus = Literal[
    "pending",
    "in-progress",
    "completed",
    "failed",
    "skipped",
]

Task status values understood by the frontend plan UI.

ExecutionPlan

Ordered list of tasks + helpers to emit status updates.

Tasks are mutable: set_task_status rewrites a task (or subtask) in place and returns the payload the consumer should forward to the frontend. When an EventBus is attached, the same payload is also emitted as an :class:AgentEvent so any subscriber (e.g. the SSE sidecar) can observe the change.

enable_ui_events

enable_ui_events(
    *, component_id: str = "plan"
) -> ExecutionPlan

Opt into dual-emission of typed ui.plan.{init,delta} events.

When enabled, :meth:set_task_status and :meth:emit_init emit a parallel typed event alongside the legacy plan.init / plan.task.updated events. The legacy events stay identical so existing consumers continue to work; new consumers subscribe to the typed channel via :class:UIEmitter conventions.

Returns self for chaining.

Source code in orqest/plan/execution_plan.py

def enable_ui_events(self, *, component_id: str = "plan") -> ExecutionPlan:
    """Opt into dual-emission of typed ``ui.plan.{init,delta}`` events.

    When enabled, :meth:`set_task_status` and :meth:`emit_init`
    emit a parallel typed event alongside the legacy ``plan.init``
    / ``plan.task.updated`` events. The legacy events stay
    identical so existing consumers continue to work; new
    consumers subscribe to the typed channel via
    :class:`UIEmitter` conventions.

    Returns ``self`` for chaining.
    """
    self._emit_ui_events = True
    self._ui_component_id = component_id
    return self

to_sse_init

to_sse_init() -> dict[str, Any]

Return the plan.init payload the frontend expects.

Source code in orqest/plan/execution_plan.py

def to_sse_init(self) -> dict[str, Any]:
    """Return the ``plan.init`` payload the frontend expects."""
    return {"tasks": [t.model_dump() for t in self.tasks]}

as_component

as_component(
    *, component_id: str | None = None
) -> PlanComponent

Wrap this plan as a :class:PlanComponent for the generative-UI pipeline.

Source code in orqest/plan/execution_plan.py

def as_component(
    self, *, component_id: str | None = None
) -> PlanComponent:
    """Wrap this plan as a :class:`PlanComponent` for the
    generative-UI pipeline."""
    from orqest.ui.components.plan import PlanComponent, PlanComponentData

    return PlanComponent(
        component_id=component_id or self._ui_component_id,
        data=PlanComponentData(tasks=list(self.tasks)),
    )

set_task_status async

set_task_status(
    task_id: str,
    status: PlanStatus,
    *,
    subtask_id: str | None = None,
    bus: EventBus | None = None,
    agent_name: str = "unknown",
) -> dict[str, Any]

Flip a task (or subtask) status and return the update payload.

Parameters:

Name	Type	Description	Default
task_id	str	ID of the :class:PlanTask to update.	required
status	PlanStatus	New status value.	required
subtask_id	str | None	If provided, update the matching subtask rather than the parent task.	None
bus	EventBus | None	Optional :class:EventBus to publish a plan.task.updated event to. Omit for tests or in-memory-only usage.	None
agent_name	str	Tagged on emitted events for source attribution.	'unknown'

Returns:

Type	Description
dict[str, Any]	A dict carrying task_id + status (and
dict[str, Any]	subtask_id if applicable). This is the exact shape the
dict[str, Any]	consumer forwards to the frontend, so it must remain stable.

Source code in orqest/plan/execution_plan.py

async def set_task_status(
    self,
    task_id: str,
    status: PlanStatus,
    *,
    subtask_id: str | None = None,
    bus: EventBus | None = None,
    agent_name: str = "unknown",
) -> dict[str, Any]:
    """Flip a task (or subtask) status and return the update payload.

    Args:
        task_id: ID of the :class:`PlanTask` to update.
        status: New status value.
        subtask_id: If provided, update the matching subtask rather
            than the parent task.
        bus: Optional :class:`EventBus` to publish a
            ``plan.task.updated`` event to. Omit for tests or
            in-memory-only usage.
        agent_name: Tagged on emitted events for source attribution.

    Returns:
        A dict carrying ``task_id`` + ``status`` (and
        ``subtask_id`` if applicable). This is the exact shape the
        consumer forwards to the frontend, so it must remain stable.
    """
    task_idx: int | None = None
    sub_idx: int | None = None
    for idx, task in enumerate(self.tasks):
        if task.id != task_id:
            continue
        task_idx = idx
        if subtask_id is not None:
            for sidx, subtask in enumerate(task.subtasks):
                if subtask.id == subtask_id:
                    subtask.status = status
                    sub_idx = sidx
                    break
        else:
            task.status = status
        break

    payload: dict[str, Any] = {"task_id": task_id, "status": status}
    if subtask_id is not None:
        payload["subtask_id"] = subtask_id

    if bus is not None:
        await bus.emit(
            AgentEvent(
                event_type="plan.task.updated",
                agent_name=agent_name,
                data=payload,
            )
        )
        # Opt-in: also emit the typed UI delta event so generative-
        # UI consumers can subscribe to ui.plan.delta uniformly with
        # other component types.
        if self._emit_ui_events and task_idx is not None:
            from orqest.ui.spec import UIDeltaEvent

            if subtask_id is not None and sub_idx is not None:
                delta_path = f"tasks.{task_idx}.subtasks.{sub_idx}.status"
            else:
                delta_path = f"tasks.{task_idx}.status"
            delta = UIDeltaEvent(
                component_id=self._ui_component_id,
                component_type="plan",
                op="replace",
                path=delta_path,
                value=status,
            )
            await bus.emit(
                AgentEvent(
                    event_type="ui.plan.delta",
                    agent_name=agent_name,
                    data=delta.model_dump(mode="json"),
                )
            )
    return payload

emit_init async

emit_init(
    bus: EventBus, *, agent_name: str = "unknown"
) -> dict[str, Any]

Publish the initial plan to bus as a plan.init event.

Convenience for consumers that always want the plan mirrored to the bus right after construction. Returns the same payload as :meth:to_sse_init so the caller can also stream it inline.

Source code in orqest/plan/execution_plan.py

async def emit_init(
    self,
    bus: EventBus,
    *,
    agent_name: str = "unknown",
) -> dict[str, Any]:
    """Publish the initial plan to *bus* as a ``plan.init`` event.

    Convenience for consumers that always want the plan mirrored to
    the bus right after construction. Returns the same payload as
    :meth:`to_sse_init` so the caller can also stream it inline.
    """
    payload = self.to_sse_init()
    await bus.emit(
        AgentEvent(
            event_type="plan.init",
            agent_name=agent_name,
            data=payload,
        )
    )
    if self._emit_ui_events:
        component = self.as_component()
        await bus.emit(
            AgentEvent(
                event_type="ui.plan.init",
                agent_name=agent_name,
                data=component.to_event_data(),
            )
        )
    return payload

from_tasks_json classmethod

from_tasks_json(
    tasks_json: str | list[dict[str, Any]] | dict[str, Any],
) -> ExecutionPlan

Construct from either a JSON string, a bare task list, or a {"tasks": [...]} dict. Useful when an LLM produces the plan as an opaque JSON string.

Source code in orqest/plan/execution_plan.py

@classmethod
def from_tasks_json(
    cls, tasks_json: str | list[dict[str, Any]] | dict[str, Any]
) -> ExecutionPlan:
    """Construct from either a JSON string, a bare task list, or a
    ``{"tasks": [...]}`` dict. Useful when an LLM produces the plan
    as an opaque JSON string.
    """
    import json as _json

    if isinstance(tasks_json, str):
        data = _json.loads(tasks_json)
    else:
        data = tasks_json

    tasks = data if isinstance(data, list) else data.get("tasks", [])
    return cls.model_validate({"tasks": tasks})

PlanSubtask

A subtask nested under a :class:PlanTask.

PlanTask

A top-level task in an :class:ExecutionPlan.