ExecutionPlan — typed multi-step workflow tracking

When an agent decomposes a user request into discrete steps (mesh → solve → analyze, research → synthesize → respond, …), the consumer usually wants to visualize progress: a task list in the UI with per- task status updates streaming in as work happens. Before ExecutionPlan, every consumer maintained its own Pydantic model, its own status-flip method, and its own hand-rolled EventBus glue to notify the frontend.

ExecutionPlan is the canonical primitive. It wraps a list of PlanTasks (optionally with PlanSubtasks), exposes a single set_task_status() that flips state and emits a typed plan.task.updated event, and keeps to_sse_init() byte-stable for frontend compatibility across Orqest releases.

Minimal example

import asyncio

from orqest import ExecutionPlan, PlanTask, PlanSubtask
from orqest.observability import EventBus


async def main() -> None:
    plan = ExecutionPlan(
        tasks=[
            PlanTask(
                id="mesh",
                title="Generate mesh",
                subtasks=[PlanSubtask(id="mesh.geo", title="Write .geo")],
            ),
            PlanTask(id="solve", title="Run solver", dependencies=["mesh"]),
        ]
    )

    bus = EventBus()
    bus.subscribe_all(lambda e: print(f"{e.event_type}: {e.data}"))

    # Initial payload the frontend renders once
    await plan.emit_init(bus, agent_name="orchestrator")

    # Later — flip a subtask as work starts and completes
    await plan.set_task_status(
        "mesh", "in-progress", subtask_id="mesh.geo", bus=bus,
        agent_name="orchestrator",
    )
    await plan.set_task_status(
        "mesh", "completed", subtask_id="mesh.geo", bus=bus,
        agent_name="orchestrator",
    )


asyncio.run(main())

Schema contract

Two payload shapes are stable — frontend code depends on them:

to_sse_init() → {"tasks": [...]} where each task has id, title, description, status, priority, level, dependencies, subtasks.

set_task_status() return value → {"task_id": ..., "status": ...} plus "subtask_id": ... when updating a subtask. The same shape lands on the EventBus as the data payload of plan.task.updated events.

Statuses

PlanStatus = "pending" | "in-progress" | "completed" | "failed" | "skipped". No additional values are introduced without a frontend coordination change.

When to reach for this

• You want a typed plan model instead of a handful of dict utilities.
• You want plan updates to flow onto your existing EventBus subscribers (metrics, SSE sidecar, logging) without extra wiring.
• You want a consistent wire format across multiple Orqest consumers so their frontends can share rendering code.

Reference

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})

PlanTask

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

PlanSubtask

A subtask nested under a :class:PlanTask.