Workflow state

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

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

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}

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

client = PolosClient()

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

const client = PolosClient.fromEnv();

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

class ConfigState(WorkflowState):
    retry_limit: int = 3
    timeout_seconds: int = 30
    debug_mode: bool = False

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>;