Error Handling

from polos import workflow, WorkflowContext

@workflow(id="retry_example")
async def retry_example(ctx: WorkflowContext, payload):
    result = await ctx.step.run(
        "unreliable_op",
        unreliable_operation,
        max_retries=3,
        base_delay=1.0,
        max_delay=10.0,
    )
    return {"result": result}

import { defineWorkflow } from '@polos/sdk';

const retryExample = defineWorkflow<RetryPayload, unknown, RetryResult>(
  { id: 'retry_example' },
  async (ctx, payload) => {
    const result = await ctx.step.run(
      'unreliable_op',
      () => unreliableOperation(payload.failureRate, payload.operation),
      { maxRetries: 3, baseDelay: 1000, maxDelay: 10000 },
    );
    return { status: 'success', result };
  },
);

from polos import StepExecutionError

@workflow(id="error_recovery")
async def error_recovery(ctx: WorkflowContext, payload):
    results, errors = [], []

    for item in payload["items"]:
        try:
            result = await ctx.step.run(f"process_{item}", process_item, item)
            results.append(result)
        except StepExecutionError as e:
            errors.append({"item": item, "error": str(e)})

    return {"processed": len(results), "failed": len(errors)}

const errorRecovery = defineWorkflow<ErrorRecoveryPayload, unknown, ErrorRecoveryResult>(
  { id: 'error_recovery' },
  async (ctx, payload) => {
    const results: { item: string; status: string }[] = [];
    const errors: { item: string; error: string }[] = [];

    for (const item of payload.items) {
      try {
        const result = await ctx.step.run(
          `process_${item}`,
          () => processItem(item),
        );
        results.push(result);
      } catch (e) {
        errors.push({
          item,
          error: e instanceof Error ? e.message : String(e),
        });
      }
    }

    return { processed: results.length, failed: errors.length, results, errors };
  },
);

@workflow(id="fallback_pattern")
async def fallback_pattern(ctx: WorkflowContext, payload):
    try:
        return await ctx.step.run("primary", primary_process, payload)
    except StepExecutionError:
        return await ctx.step.run("fallback", fallback_process, payload)

const fallbackPattern = defineWorkflow<FallbackPayload, unknown, FallbackResult>(
  { id: 'fallback_pattern' },
  async (ctx, payload) => {
    try {
      const result = await ctx.step.run(
        'primary_method',
        () => primaryProcess(payload.data),
        { maxRetries: 2 },
      );
      return { method: 'primary', result };
    } catch {
      const result = await ctx.step.run(
        'fallback_method',
        () => fallbackProcess(payload.data),
      );
      return { method: 'fallback', result };
    }
  },
);

@workflow(id="compensation")
async def compensation(ctx: WorkflowContext, payload):
    completed = []
    try:
        await ctx.step.run("reserve", reserve_inventory, payload)
        completed.append("reserve")
        await ctx.step.run("charge", charge_payment, payload)
        completed.append("charge")
        return {"status": "success"}
    except StepExecutionError:
        # Rollback in reverse order
        for step in reversed(completed):
            await ctx.step.run(f"undo_{step}", compensate[step], payload)
        return {"status": "rolled_back"}

const compensationPattern = defineWorkflow<CompensationPayload, unknown, CompensationResult>(
  { id: 'compensation_pattern' },
  async (ctx, payload) => {
    const completedSteps: string[] = [];

    try {
      await ctx.step.run('reserve_inventory', () => reserveInventory(payload));
      completedSteps.push('reserve_inventory');

      await ctx.step.run('charge_payment', () => chargePayment(payload));
      completedSteps.push('charge_payment');

      return { status: 'success', completed: completedSteps };
    } catch (e) {
      // Rollback in reverse order
      for (const step of [...completedSteps].reverse()) {
        const compensationFn = getCompensation(step);
        if (compensationFn) {
          await ctx.step.run(`compensate_${step}`, () => compensationFn(payload));
        }
      }
      return {
        status: 'rolled_back',
        error: e instanceof Error ? e.message : String(e),
        compensated: [...completedSteps].reverse(),
      };
    }
  },
);

git clone https://github.com/polos-dev/polos.git
cd polos/python-examples/11-error-handling
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
uv sync
python main.py

git clone https://github.com/polos-dev/polos.git
cd polos/typescript-examples/11-error-handling
cp .env.example .env  # Add your POLOS_PROJECT_ID and API key
npm install
npx tsx main.ts