State & the Entity Model

Typed, immutable, auditable Pydantic state and the activity return convention

State & the entity model

Audience: Customer developers. How workflow state is modelled — typed, immutable, and auditable — so decisions can be reasoned about and reviewed.

A workflow's state is not a loose bag of variables. It's a typed Pydantic model that the SDK serialises and deserialises around every step, and it grows by immutable addition: each step appends its result rather than overwriting what came before. That history is what lets a human — or an agent — reason about how a result was reached, and what lets an auditor trace it later.

The typed input

Every workflow receives a typed input. The canonical model is WorkflowInput (tenant, case id, a payload, runtime config, trace context); the runtime config is frozen (immutable). Validate the incoming payload at the top of run:

from workflow_sdk import WorkflowInput

@workflow.run
async def run(self, payload: dict) -> dict:
    wf_input = WorkflowInput.model_validate(payload)
    ...

The input a run starts with comes from the Studio start form, which is generated from the input_model you declare in WorkflowRequirements. See Running a workflow.

Managed state: your workflow entity

Pass entity_type=YourModel to reviewable_workflow and the SDK manages the entity for you. Each step's activity returns a typed result; the SDK serialises it, appends it to state under that step's key, and exposes the merged current entity as loop.entity:

async for loop in reviewable_workflow(state, loop_id="main", entity_type=ReviewRecord):
    await loop.start("ingest", ingest_activity, payload)   # state["ingest"] = typed result
    await loop.target(agent_review_activity, state_key="review")  # state["review"] = typed result
    ...
entity = loop.entity   # the merged ReviewRecord, revalidated each step

• Append, don't mutate. Each step writes under its own state_key; earlier entries are never changed. State accumulates one entry per step — an immutable trail.
• Always typed. On every step the SDK revalidates stored entries against your model, so state can never silently drift out of shape.

The entity contract

Your entity model is an ordinary Pydantic BaseModel that implements three methods the SDK calls:

from pydantic import BaseModel

class ReviewRecord(BaseModel):
    entries: list[FieldEntry] = []

    def add_entry(self, key, value, source):
        # return a NEW record with the entry appended — don't mutate self
        ...
    def hitl_flatten(self, display_fields):
        ...
    def resolve_field(self, key, value, source):
        ...

Because each field carries its source (which step or reviewer produced it), the entity is self-describing: you can see what was extracted, what an agent generated, and what a human corrected — all in one immutable record.

The activity contract: what each step returns

State only works because activities follow a return convention. An activity can't return a bare dict or a naked entity — the SDK wouldn't know how to re-type it. Instead, every activity returns one of a few small wrapper models (generic over your entity type). The SDK serialises the wrapper into state, and on the next step deserialises it and revalidates the inner data against your entity_type. That's the whole reason state stays typed across steps.

What an activity receives

Each activity takes a single activity_input: dict:

What an activity returns

from workflow_sdk import activity, InputData, Basic

@activity.defn
async def ingest_activity(activity_input: dict) -> InputData:
    payload = activity_input["arg"]
    record = ReviewRecord(request=payload.get("request", ""))
    return InputData(data=record, documents=[])          # entity + documents

@activity.defn
async def review_activity(activity_input: dict) -> Basic:
    state = activity_input["state"]
    prompt = activity_input.get("prompt")                # feedback on a repeat
    record = _entity_from(state, "ingest")
    try:
        record.add_entry("decision", "reviewed", source="review_activity")
        return Basic(decision=True, data=record)         # updated entity
    except Exception as exc:
        return Basic(decision=False, data=None, error=str(exc))   # surfaced failure

Key rules:

• Return a wrapper, always. InputData, Basic, or Value — never a raw dict/entity.
• data is your entity (or None when a step only decides and changes nothing).
• error is how a step reports failure into state, so later steps and reviewers can see it — distinct from a Temporal retry (which handles infrastructure failures).
• documents on InputData are the files a reviewer will see at a HITL gate.

Unmanaged state

If you don't need the managed entity, omit entity_type. State can be a plain dict, or a Pydantic model you pass and drive yourself (optionally with state_type= so the SDK revalidates it after each mutation). Use this for simple linear workflows without HITL correction.

Why immutable, typed state matters

• Reasoning — a reviewer (or an agent doing an agentic merge) can see the full derivation, not just a final value, and decide from evidence.
• Auditability — the append-only trail is a natural record of every step and decision.
• Reliability — typed (de)serialisation is what lets Temporal durably persist and replay your workflow across restarts without state drifting.

Next: Calling agents & tools.