Plans

A plan in the harness is a typed Pydantic schema that the planner emits and the executor consumes; when the action list is polymorphic, TaggedUnion builds a Pydantic...

A plan in the harness is a typed Pydantic schema that the planner emits and the executor consumes; when the action list is polymorphic, TaggedUnion builds a Pydantic discriminated union over a kind field.

from pydantic import BaseModel

from flowai_harness import TaggedUnion, define_plan

class PriceChange(BaseModel):
    kind: str = "price_change"
    product_id: str
    new_price: float

class PromotionLaunch(BaseModel):
    kind: str = "promotion_launch"
    product_ids: list[str]
    discount_pct: float

ScenarioAction = TaggedUnion(PriceChange, PromotionLaunch)

class ScenarioPlan(BaseModel):
    scope_ref: str
    actions: list[ScenarioAction]
    rationale: str

scenario_plan = define_plan(name="ScenarioPlan", schema=ScenarioPlan)

What a plan is

define_plan(name=..., schema=..., display_aliases=...) produces a PlanSpec. The schema argument accepts a Pydantic model class or any value that normalize_schema(...) can resolve to JSON Schema. The spec carries the resolved JSON Schema on the wire; the Rust runtime (flowai-runtime) compiles it into its native plan state machine.

Both the planner and the executor attach the same PlanSpec so their I/O is typed:

from flowai_harness import define_executor, define_planner

planner = define_planner(
    name="scenario_planner",
    model="claude-sonnet-4-6",
    plan=scenario_plan,
    prompt="You produce typed scenario plans.",
)
executor = define_executor(
    name="scenario_executor",
    model="claude-sonnet-4-6",
    plan=scenario_plan,
    tools=[search_products],
    prompt="You execute approved scenario plans action by action.",
)

Tagged action unions

When the planner can emit several action variants, declare them as separate Pydantic models and combine them with TaggedUnion. The harness supports two forms of discriminator:

kind: str default

from pydantic import BaseModel

from flowai_harness import TaggedUnion

class PriceChange(BaseModel):
    kind: str = "price_change"
    product_id: str
    new_price: float

class PromotionLaunch(BaseModel):
    kind: str = "promotion_launch"
    product_ids: list[str]
    discount_pct: float

ScenarioAction = TaggedUnion(PriceChange, PromotionLaunch)

Literal[...] annotation

from typing import Literal

from pydantic import BaseModel

from flowai_harness import TaggedUnion

class PriceChange(BaseModel):
    kind: Literal["price_change"]
    product_id: str
    new_price: float

class PromotionLaunch(BaseModel):
    kind: Literal["promotion_launch"]
    product_ids: list[str]
    discount_pct: float

ScenarioAction = TaggedUnion(PriceChange, PromotionLaunch)

The discriminator field defaults to kind; pass discriminator="some_other_field" to change it. TaggedUnion validates at construction time that each variant either carries a single-value Literal or a non-empty string default, and rejects duplicate discriminator values.

Static typing

TaggedUnion(...) is a Pydantic alias factory; its return value is typed as Any because type checkers cannot infer a precise alias from a runtime call. If you need static narrowing in your own code, declare a typing.TypeAlias alongside the union. Runtime validation and JSON Schema generation still use Pydantic's discriminated union machinery.

Plan lifecycle

Every plan instance moves through a fixed status lifecycle owned by the runtime:

draft --> approved --> executing --> executed
                           |
                           +------> failed

Each transition has a clear owner:

The planner emits a plan in draft.
The approval gate moves draft → approved. The runtime pauses, emits an approval event, and waits for the host to respond — see Approvals.
The executor moves approved → executing, then finishes at executed on success or failed on error.

These five statuses are the PlanStatus values accepted by the Python API; anything outside the set is rejected at spec construction time.

Display aliases

You can override the lifecycle labels surfaced to end users with display_aliases:

from flowai_harness import define_plan

scenario_plan = define_plan(
    name="ScenarioPlan",
    schema=ScenarioPlan,
    display_aliases={
        "draft": "Draft scenario",
        "approved": "Approved scenario",
        "executing": "Running scenario",
        "executed": "Scenario complete",
        "failed": "Scenario failed",
    },
)

Statuses outside the allowed set raise ValueError immediately. Aliases are pure presentation metadata — the runtime still tracks the underlying status internally.