tasks/CONTRACT.md

Hanzo Tasks — Canonical Contract

The single durable async substrate for every Hanzo + Liquidity 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{
    HostPort:  os.Getenv("TASKS_ADDR"),     // "tasks.hanzo.svc:9999"
    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; 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. A liquidity worker cannot see mlc workflows even if both share a binary. The strip

• mint identity middleware in tasksd enforces this on every RPC.

Multi-tenant services (BD, TA, ATS) 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. Every Liquidity 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".