tasks/CONTRACT.md

Hanzo Tasks — Canonical Contract

The single durable async substrate for every Hanzo Go service.

This document is law. If your service needs retries, durability, scheduling, cron, or any async work that must survive a process restart, you use github.com/hanzoai/tasks. There is no second async system.

1. When to use tasks vs sync

Sync HTTP handler (no tasks):

• Total work fits in <100ms.
• Caller blocks for the response and the result is the whole point of the call.
• Single round-trip with no external I/O that can flake (OTP code generation, JWT issue, field validation, in-memory CLOB match, JWKS lookup).
• Failure is the response — the caller decides what to do.

Tasks (durable workflow):

• Any side-effect that must survive a pod restart or panic.
• Anything with retry policy (webhook delivery, mint/burn, settlement, notification send, third-party API fan-out).
• Anything scheduled (cron, delay, send_at, daily/quarterly sweeps).
• Anything that fans out across multiple steps where a partial completion is worse than no completion (saga-style flows).
• Long jobs (>5s) where the HTTP caller should not block.

Rule of thumb: if you find yourself writing go func() { ... }() to fire work in the background, you are writing a bug. Use tasks.

2. Task definition shape

Every workflow is named, namespaced, and has a typed input + output.

// pkg/tasks/<feature>.go in your service.
package <feature>tasks

import (
    "time"

    "github.com/hanzoai/tasks/pkg/sdk/workflow"
)

// Input is the workflow + activity payload. JSON-serializable, no
// pointers, no interfaces.
type Input struct {
    ID         string         `json:"id"`
    TenantSlug string         `json:"tenant_slug"`
    // … domain fields …
}

// Output is the workflow return shape.
type Output struct {
    ID     string `json:"id"`
    Status string `json:"status"`
    Error  string `json:"error,omitempty"`
}

// FeatureWorkflow is the durable entry point. Must be deterministic on
// its inputs — all I/O happens inside activities.
func FeatureWorkflow(ctx workflow.Context, in Input) (Output, error) {
    actCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
        StartToCloseTimeout: 60 * time.Second,
        RetryPolicy: &workflow.RetryPolicy{
            InitialInterval:    5 * time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    2 * time.Minute,
            MaximumAttempts:    5,
        },
    })
    var out Output
    if err := workflow.ExecuteActivity(actCtx, "Deliver", in).Get(actCtx, &out); err != nil {
        return Output{ID: in.ID, Status: "failed", Error: err.Error()}, err
    }
    return out, nil
}

Defaults — apply unless your workflow explicitly contradicts them:

Field	Default	When to change
StartToCloseTimeout	60s	I/O heavy >60s; never above 10m
InitialInterval	5s	Fast-retry only if upstream <1s
BackoffCoefficient	2.0	Never lower (thundering herd)
MaximumInterval	2m	Long-tail jobs may bump to 30m
MaximumAttempts	5	Idempotent ops can go to 10

Idempotency is the activity's responsibility. The tasks server replays on retry; your activity must detect "already done" and return the prior result.

3. Worker registration pattern

One worker per service, started during boot, drained on shutdown.

// in your service's main.go boot path
import (
    tasksclient "github.com/hanzoai/tasks/pkg/sdk/client"
    tasksworker "github.com/hanzoai/tasks/pkg/sdk/worker"
)

cli, err := tasksclient.Dial(tasksclient.Options{
    Address:   os.Getenv("TASKS_ADDR"),     // "tasks.hanzo.svc:9999" or "/run/tasks.sock"
    Namespace: tenantNamespace(orgID),       // see §6
})
if err != nil {
    return fmt.Errorf("tasks dial: %w", err)
}

wk := tasksworker.New(cli, "<service>-default", tasksworker.Options{
    Logger: logger,
})
wk.RegisterWorkflow(FeatureWorkflow)
wk.RegisterActivity(activities.Deliver)
if err := wk.Start(); err != nil {
    return fmt.Errorf("tasks worker start: %w", err)
}
defer wk.Stop()

Task queue name: <service>-<purpose> (e.g. notify-send, bd-webhooks, ta-chain, ats-settlement). One queue per workflow family. Do not multiplex unrelated workflows on the same queue.

4. Client API

The sync caller's interaction with tasks is exactly two calls.

// Enqueue — returns immediately with task id.
run, err := cli.ExecuteWorkflow(ctx, tasksclient.StartWorkflowOptions{
    ID:        "<feature>-" + in.ID, // deterministic ID = dedup key
    TaskQueue: "<service>-<purpose>",
}, "FeatureWorkflow", in)
if err != nil {
    return "", fmt.Errorf("tasks enqueue: %w", err)
}
taskID := run.GetID()

// Optional wait for completion (sync-over-async).
var out Output
if err := run.Get(ctx, &out); err != nil {
    return out, err
}

ExecuteWorkflow is idempotent on the workflow ID. Re-submitting the same ID with the workflow still running returns the existing handle. Use the business key (message_id, withdrawal_id, intent_id) as the workflow ID — that is your idempotency token.

5. ZAP transport

Production: ZAP only. tasksd listens on TCP/9999 by default -- or on a unix socket when --zap is given a path, which is the right shape when caller and engine share a host; the SDK dials directly. No HTTP, no JSON envelope on the hot path, no gRPC.

Local dev: TASKS_ADDR="" switches the SDK to in-process mode (no durability, used for tests and dev loops). Production must set TASKS_ADDR=tasks.hanzo.svc:9999 (or the per-cluster service name).

Auth flows via the same JWT the caller already holds. The SDK attaches Authorization: Bearer <jwt> from tasksclient.Options.AuthToken or from the context's outbound header bag. tasksd validates against IAM JWKS (see hanzoai/tasks LLM.md §"Auth: IAM only").

6. Per-tenant namespace isolation

Namespaces map 1:1 to IAM orgs. This is non-negotiable.

Layer	Value
IAM org	mlc
Tasks namespace	mlc
SQLite store	data/mlc.db

Workers dial with Namespace: <orgSlug>. All schedules, workflows, and visibility queries are namespace-scoped at the server. An org-a worker cannot see org-b workflows even if both share a binary. The strip + write identity middleware in tasksd enforces this on every RPC.

Multi-tenant services hold one client + worker per active org. Onboarding a new org allocates the namespace lazily on first ExecuteWorkflow.

Single-tenant services (notify when serving Hanzo only) may run a single default namespace; the moment they serve a second tenant, switch to per-org namespacing — no retrofitting allowed in production.

7. Failure modes

• Activity error returned: tasks server retries per the workflow's RetryPolicy. After MaximumAttempts, the workflow run is marked failed and the result is queryable via cli.DescribeWorkflowExecution.
• Activity panic: caught by the SDK; counted as an attempt; retried.
• Worker process killed mid-activity: the heartbeat times out; the task is rescheduled on another worker. Activities MUST be idempotent.
• Workflow exceeds WorkflowExecutionTimeout: terminated; downstream saga must reconcile or fire a compensating workflow.
• Server unreachable on enqueue: ExecuteWorkflow returns an error. The caller decides — usually: persist the intent to local storage and resubmit on a recovery cron, never silently drop.

Dead-letter behavior: there is no separate DLQ collection. Failed runs stay in the visibility store with status=failed and are queryable. The operator surface (CRD LiquidTasks / HanzoTasks .status) surfaces counts of failed_24h per namespace per queue.

Alerting: tasks_workflow_runs_failed_total{namespace,queue} Prometheus metric → alerts at >0.5% failure ratio over 5m or any 10 consecutive failures of the same workflow ID. The alert lands in the owning team's PagerDuty service via Hanzo Insights.

8. Observability

UI: tasks.hanzo.ai (hanzoai/tasks Web UI). Filter by namespace, task queue, status, time range. Inspect input/output/history for any run.

CLI:

# describe a stuck workflow
tasksctl workflow describe --namespace <org> <workflow-id>

# replay events for debugging
tasksctl workflow history --namespace <org> <workflow-id> --json

# queue depth (per-queue backlog)
tasksctl queue stats --namespace <org> <queue-name>

# force-fail a runaway workflow
tasksctl workflow terminate --namespace <org> <workflow-id> --reason <text>

Metrics (Prometheus, scraped from tasksd /metrics):

Metric	Labels
tasks_workflow_runs_started_total	namespace, workflow_type
tasks_workflow_runs_completed_total	namespace, workflow_type, status
tasks_workflow_runs_failed_total	namespace, workflow_type
tasks_activity_attempts_total	namespace, activity_type, outcome
tasks_queue_backlog	namespace, queue
tasks_workflow_run_duration_seconds	namespace, workflow_type (histogram)

Traces: every workflow + activity emits an OTEL span. Span links connect the originating HTTP handler → workflow start → each activity attempt.

Logs: workflow + activity logs are structured slog records with workflow_id, run_id, namespace, task_queue, attempt. They land in the cluster's logging backend (Loki, in our case).

9. What this replaces

This contract supersedes:

• Per-service pkg/taskqueue/ and pkg/tasks/ wrappers that wrap a fork of the SDK with HTTP fallbacks. The wrappers will be removed once direct callers migrate to github.com/hanzoai/tasks/pkg/sdk.
• go.temporal.io/sdk/* imports anywhere outside this repo. Any service that still imports the legacy SDK gets migrated in the dispatched follow-up work; new code must not import it.
• Ad-hoc time.NewTicker + go func() for retries, cron, or scheduled send. If you wrote one of those, you owe a tasks workflow.

10. Versioning

The Hanzo Tasks module pins to v1.x.x forever per the global Go package policy (no v2+ semver-import-versioning). New features land as minor bumps; fixes as patches. See LLM.md §"Versioning policy".