┌────────────────────────────────────────────────────┐
│                  Hanzo Console                      │
│          (Workflow browser, execution viewer)       │
└──────────────────────┬─────────────────────────────┘
                       │ gRPC
┌──────────────────────▼─────────────────────────────┐
│              Hanzo Tasks Server                     │
│   ┌──────────┬──────────┬──────────┬────────────┐  │
│   │ Frontend │ History  │ Matching │  Worker    │  │
│   │ Service  │ Service  │ Service  │  Service   │  │
│   └──────────┴──────────┴──────────┴────────────┘  │
│         Multi-tenant namespace routing              │
│         IAM OIDC authorizer                         │
└──────────────────────┬─────────────────────────────┘
                       │
            ┌──────────▼──────────┐
            │   Hanzo SQL         │
            │   (PostgreSQL)      │
            └─────────────────────┘

from hanzo.tasks import Client, workflow, activity
import asyncio

@activity.defn
async def send_email(to: str, subject: str) -> str:
    # Your activity logic here
    return f"Sent to {to}"

@workflow.defn
class OnboardingWorkflow:
    @workflow.run
    async def run(self, user_email: str) -> str:
        # Step 1: Send welcome email (retried automatically on failure)
        result = await workflow.execute_activity(
            send_email,
            args=[user_email, "Welcome to Hanzo!"],
            start_to_close_timeout=timedelta(seconds=30),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        # Step 2: Wait for user to complete profile (up to 7 days)
        await workflow.wait_condition(
            lambda: self.profile_completed,
            timeout=timedelta(days=7),
        )

        return "Onboarding complete"

    @workflow.signal
    async def complete_profile(self):
        self.profile_completed = True

async def main():
    client = await Client.connect(
        "tasks-api.hanzo.ai:443",
        # IAM token for multi-tenant auth
        api_key="your-hanzo-api-key",
        namespace="your-org",  # Hanzo org = Tasks namespace
    )

    handle = await client.start_workflow(
        OnboardingWorkflow.run,
        "new-user@example.com",
        id="onboard-user-123",
        task_queue="onboarding",
    )

    result = await handle.result()

package main

import (
    "context"
    "github.com/hanzoai/tasks/client"
    "github.com/hanzoai/tasks/worker"
    "github.com/hanzoai/tasks/workflow"
    "github.com/hanzoai/tasks/activity"
)

func OnboardingWorkflow(ctx workflow.Context, email string) error {
    opts := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy: &tasks.RetryPolicy{MaximumAttempts: 3},
    }
    ctx = workflow.WithActivityOptions(ctx, opts)

    err := workflow.ExecuteActivity(ctx, SendEmail, email, "Welcome!").Get(ctx, nil)
    if err != nil {
        return err
    }

    // Wait for signal (human-in-the-loop)
    var completed bool
    signalChan := workflow.GetSignalChannel(ctx, "complete-profile")
    signalChan.Receive(ctx, &completed)

    return nil
}

import { Client, Connection } from '@hanzo/tasks';

const connection = await Connection.connect({
  address: 'tasks-api.hanzo.ai:443',
});

const client = new Client({
  connection,
  namespace: 'your-org',
});

const handle = await client.workflow.start('OnboardingWorkflow', {
  args: ['new-user@example.com'],
  taskQueue: 'onboarding',
  workflowId: 'onboard-user-123',
});

const result = await handle.result();

@workflow.defn
class AgentPipeline:
    @workflow.run
    async def run(self, prompt: str):
        plan = await workflow.execute_activity(plan_task, prompt)
        for step in plan.steps:
            result = await workflow.execute_activity(
                execute_step, step,
                retry_policy=RetryPolicy(maximum_attempts=5),
            )
            if result.needs_human_review:
                approved = await workflow.wait_condition(
                    lambda: self.approved, timeout=timedelta(hours=24)
                )

@workflow.defn
class OrderSaga:
    @workflow.run
    async def run(self, order):
        # Each step has a compensating action
        charge = await workflow.execute_activity(charge_card, order)
        try:
            await workflow.execute_activity(reserve_inventory, order)
            await workflow.execute_activity(ship_order, order)
        except Exception:
            # Compensate: refund the charge
            await workflow.execute_activity(refund_charge, charge)
            raise

# Start a cron workflow
handle = await client.start_workflow(
    DailyReportWorkflow.run,
    id="daily-report",
    task_queue="reports",
    cron_schedule="0 9 * * MON-FRI",  # 9 AM weekdays
)

# compose.yml
services:
  tasks:
    image: ghcr.io/hanzoai/tasks:latest
    ports:
      - "7233:7233"
    environment:
      DB: postgres12
      POSTGRES_SEEDS: your-postgres-host
      POSTGRES_USER: tasks
      POSTGRES_PWD: ${TASKS_DB_PASSWORD}