> ## Documentation Index
> Fetch the complete documentation index at: https://polos.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Workflow state

Workflow state persists data across the entire workflow execution. Use state for counters, accumulators, configuration, or any data that multiple steps need to access and modify.

## Defining workflow state

Create a schema for state using `WorkflowState`:

<CodeGroup>
  ```python Python theme={null}
  from polos import workflow, WorkflowContext, WorkflowState

  class ProcessingState(WorkflowState):
      processed_count: int = 0
      failed_items: list[str] = []
      started_at: str = ""

  @workflow(state_schema=ProcessingState)
  async def batch_processor(ctx: WorkflowContext, input: BatchProcessorInput):
      # Access state via ctx.state
      ctx.state.started_at = input.timestamp

      for item in input.items:
          try:
              await ctx.step.run(f"process_{item}", process_item, item)
              ctx.state.processed_count += 1
          except Exception as e:
              ctx.state.failed_items.append(item)

      return BatchProcessorOutput(
          processed=ctx.state.processed_count,
          failed=len(ctx.state.failed_items)
      )
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';
  import { z } from 'zod';

  const processingStateSchema = z.object({
    processedCount: z.number().default(0),
    failedItems: z.array(z.string()).default([]),
    startedAt: z.string().default(''),
  });

  type ProcessingState = z.infer<typeof processingStateSchema>;

  const batchProcessor = defineWorkflow<BatchProcessorInput, ProcessingState, BatchProcessorOutput>(
    { id: 'batch_processor', stateSchema: processingStateSchema },
    async (ctx, input) => {
      // Access state via ctx.state
      ctx.state.startedAt = input.timestamp;

      for (const item of input.items) {
        try {
          await ctx.step.run(`process_${item}`, () => processItem(item));
          ctx.state.processedCount += 1;
        } catch (e) {
          ctx.state.failedItems.push(item);
        }
      }

      return {
        processed: ctx.state.processedCount,
        failed: ctx.state.failedItems.length,
      };
    },
  );
  ```
</CodeGroup>

## When to use state

**Use state for:**

* ✅ Counters and accumulators
* ✅ Progress tracking
* ✅ Configuration set during execution
* ✅ Data shared across multiple steps

**Don't use state for:**

* ❌ Large data (state has 1MB limit)
* ❌ Temporary variables (use local variables)

State is saved when the workflow completes or fails.

## Setting initial state during workflow invocation

You can set the initial workflow state when invoking it:

<CodeGroup>
  ```python Python theme={null}
  class ParentState(WorkflowState):
      tasks_started: int = 0

  class ChildState(WorkflowState):
      parent_id: str = ""
      config: dict = {}

  @workflow(state_schema=ParentState)
  async def parent_workflow(ctx: WorkflowContext, input: ParentInput):
      ctx.state.tasks_started += 1

      # Invoke child workflow with some initial values for its state
      result = await ctx.step.invoke_and_wait(
          "call_child",
          child_workflow,
          payload={"data": input.data},
          initial_state=ChildState(
              parent_id=ctx.execution_id,
              config={"mode": "production"}
          )
      )

      return result

  @workflow(state_schema=ChildState)
  async def child_workflow(ctx: WorkflowContext, input: ChildInput):
      # Access initial state
      print(f"Parent ID: {ctx.state.parent_id}")
      print(f"Config: {ctx.state.config}")

      await ctx.step.run("process", process_data, input.data)
      return {"processed": True}
  ```

  ```typescript TypeScript theme={null}
  import { defineWorkflow } from '@polos/sdk';
  import { z } from 'zod';

  const parentStateSchema = z.object({
    tasksStarted: z.number().default(0),
  });

  type ParentState = z.infer<typeof parentStateSchema>;

  const childStateSchema = z.object({
    parentId: z.string().default(''),
    config: z.record(z.unknown()).default({}),
  });

  type ChildState = z.infer<typeof childStateSchema>;

  const parentWorkflow = defineWorkflow<ParentInput, ParentState, Result>(
    { id: 'parent_workflow', stateSchema: parentStateSchema },
    async (ctx, input) => {
      ctx.state.tasksStarted += 1;

      // Invoke child workflow with some initial values for its state
      const result = await ctx.step.invokeAndWait(
        'call_child',
        childWorkflow,
        {
          payload: { data: input.data },
          initialState: {
            parentId: ctx.executionId,
            config: { mode: 'production' },
          },
        },
      );

      return result;
    },
  );

  const childWorkflow = defineWorkflow<ChildInput, ChildState, { processed: boolean }>(
    { id: 'child_workflow', stateSchema: childStateSchema },
    async (ctx, input) => {
      // Access initial state
      console.log(`Parent ID: ${ctx.state.parentId}`);
      console.log(`Config: ${JSON.stringify(ctx.state.config)}`);

      await ctx.step.run('process', () => processData(input.data));
      return { processed: true };
    },
  );
  ```
</CodeGroup>

Invoke parent workflow with some initial values for its state:

<CodeGroup>
  ```python Python theme={null}
  client = PolosClient()

  result = await parent_workflow(
      client,
      ParentInput(data={"resume_from": 10}),
      initial_state=ParentState(tasks_started=10)
  )
  ```

  ```typescript TypeScript theme={null}
  const client = PolosClient.fromEnv();

  const result = await parentWorkflow.invoke(
    client,
    { data: { resumeFrom: 10 } },
    { initialState: { tasksStarted: 10 } },
  );
  ```
</CodeGroup>

## Best practices

* Do not use state to store large objects. Polos enforces a limit of 1MB (serialized JSON) on state size.

* Use default values

<CodeGroup>
  ```python Python theme={null}
  class ConfigState(WorkflowState):
      retry_limit: int = 3
      timeout_seconds: int = 30
      debug_mode: bool = False
  ```

  ```typescript TypeScript theme={null}
  import { z } from 'zod';

  const configStateSchema = z.object({
    retryLimit: z.number().default(3),
    timeoutSeconds: z.number().default(30),
    debugMode: z.boolean().default(false),
  });

  type ConfigState = z.infer<typeof configStateSchema>;
  ```
</CodeGroup>