Workflows and activities

A Long Tail workflow is an async function with two kinds of call. proxyActivities runs work your code controls and checkpoints the result. conditional waits on work it does not control and makes the wait a row. This page covers the first kind and how workflows are registered, started and composed.

Anatomy

import { Durable } from '@hotmeshio/hotmesh';
import { conditional, type LTEnvelope } from '@hotmeshio/long-tail';

import * as activities from '../activities';

const { analyzeContent } = Durable.workflow.proxyActivities<typeof activities>({ activities });

export async function reviewContent(envelope: LTEnvelope) {
  const analysis = await analyzeContent(envelope.data.content);       // checkpointed
  if (analysis.confidence >= 0.85) return { data: { approved: true } };

  const { workflowId } = Durable.workflow.workflowInfo();
  const decision = await conditional<{ approved: boolean }>(`review-${workflowId}`, {
    role: 'reviewer',
    metadata: { contentId: envelope.data.contentId },
    timeout: '72h',
  });                                                                  // a row, then a wait

  return { data: { approved: decision ? decision.approved : false } };
}

The workflow body is deterministic: given the same stored results it makes the same calls in the same order. Anything that touches the outside world (a model, an API, a database, the clock) goes in an activity.

The envelope

Every workflow receives one argument, an LTEnvelope.

Field	Holds
`data`	Business input, from the invoke call
`metadata`	Control data for the run
`lt`	Set by Long Tail: `userId`, `principal` (id, type, roles), `initiatedBy` when a run executes as another account
`resolver`	The resolver payload when a run is started by a resolution strategy (see Triage)

Activities

An activity is a plain async function with typed input and output.

export async function analyzeContent(content: string) {
  const result = await model.classify(content);
  return { confidence: result.confidence, flags: result.flags };
}

proxyActivities wraps each function so its result is written to Postgres when it completes:

A completed activity never runs again. After a crash the workflow replays and reads the stored result.
The workflow resumes from its last checkpoint, not from the start.
An activity that was mid-flight during a crash is redelivered once its stream reservation lapses (HMSH_RESERVATION_TIMEOUT_S, about 90 seconds worst case) and runs again. Delivery is at least once, so make side effects (payments, emails, bookings) idempotent.

Retries

Failed activities retry with exponential backoff. Configure the policy per proxy:

const { chargeCard } = Durable.workflow.proxyActivities<typeof activities>({
  activities,
  retry: { maximumAttempts: 3, initialInterval: '1s', backoffCoefficient: 2, maximumInterval: '30s' },
});

Field	Default
`maximumAttempts`	50
`initialInterval`	`'1s'`
`backoffCoefficient`	10
`maximumInterval`	`'120s'`
`throwOnError`	`true`: the workflow receives the error once attempts run out

Identity in activities

Activities read who started the run, and resolve that principal’s credentials, with getActivityIdentity():

import { getActivityIdentity } from '@hotmeshio/long-tail';

export async function draftReply(input: { threadId: string }) {
  const identity = getActivityIdentity();
  if (!identity) throw new Error('getActivityIdentity runs inside an activity');
  const token = await identity?.getCredential('google');   // executing principal, then the invoker, then env
  // identity.principal: { id, type, roles, displayName }
}

getCredential throws when no credential resolves, so a missing credential can become a row a person fixes. getActivityIdentity() returns null outside a run that carries a principal.

Register workers

Pass workers to start(). taskQueue groups workflows onto worker processes; config registers the workflow so people and services can start it.

await start({
  database: { connectionString: process.env.DATABASE_URL },
  workers: [
    {
      taskQueue: 'default',
      workflow: reviewContent,
      config: {
        description: 'Score content and send low scores to a reviewer',
        invocable: true,
        invocationRoles: ['editor'],
      },
    },
  ],
});

`config` field	Effect
`invocable`	Allows starting the workflow from the dashboard, HTTP, SDK, CLI and MCP. Default `false`
`invocationRoles`	Roles allowed to start it. Empty means any authenticated user. A superadmin grant, or the `admin` role held with `admin` type, starts any workflow
`description`	Shown in the registry and on the Invoke Tool page
`inputSchema`	An x-lt-* form for the Invoke Tool page. The invoke API validates `data` against it (`422`)
`inputLookups`	Versioned knowledge entries the invoke form reads as `lookup.*`
`envelopeSchema`	A JSON template that prefills the plain invoke form
`executeAs`	A bot account the workflow runs as
`readSafe`	Marks a side-effect-free workflow, invocable by read-scoped MCP callers
`toolTags`	MCP tool tags used for tool discovery
`icon`	A curated icon for the registry and Invoke Tool page
`reset`	Per-entry ownership override for code-owned config

Each escalation’s form lives on its target role, never on the workflow config. A workflow can park rows on many roles, and each role owns its own versioned form.

Start a workflow

curl -X POST http://localhost:3000/api/workflows/reviewContent/invoke \
  -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  -d '{"data":{"contentId":"post-1","content":"Hello"},"metadata":{"source":"api"}}'
# 202 { "workflowId": "reviewContent-...", "message": "Workflow started" }

const result = await lt.workflows.invoke({
  type: 'reviewContent',
  data: { contentId: 'post-1', content: 'Hello' },
});
// result.status === 202, result.data.workflowId

ltc wf invoke reviewContent --data '{"contentId":"post-1","content":"Hello"}'
ltc wf status <workflowId>
ltc wf result <workflowId>

Call invoke_workflow with workflow_type and data (plus optional metadata). invoke_workflow_read_safe starts only workflows registered with readSafe.

The invoke gate checks invocationRoles against the caller’s current role grants, then starts the run with the caller’s identity in envelope.lt.

Compose workflows

A workflow can start other workflows. executeChild starts a child and waits for its result; startChild starts one and returns its id.

export async function fulfillOrder(envelope: LTEnvelope) {
  const { orderId, serialNumber } = envelope.data;

  const inspection = await Durable.workflow.executeChild<{ passed: boolean }>({
    workflowName: 'inspectPart',
    args: [{ data: { serialNumber }, metadata: {} }],
    taskQueue: 'default',
    workflowId: `inspect-${orderId}`,
  });

  if (!inspection.passed) return { data: { orderId, status: 'held' } };
  return { data: { orderId, status: 'shipped' } };
}

A child that parks its own conditional row waits for its own resolver, and the parent waits on the child. To fan out to many children and collect them with one wake, start each with startChild and park one batch row that each child fills when it finishes. See Batches and accumulators.

Turn failures into rows

When an activity runs out of retries, the workflow receives the error. Catch it and park a row so a person decides what happens next.

const { chargeCard } = Durable.workflow.proxyActivities<typeof activities>({
  activities,
  retry: { maximumAttempts: 3 },
});

try {
  await chargeCard(order);
} catch (err) {
  if (Durable.workflow.didInterrupt(err as Error)) throw err;   // engine control flow, rethrow

  const { workflowId } = Durable.workflow.workflowInfo();
  const next = await conditional<{ action: 'retry' | 'refund' }>(`charge-failed-${workflowId}`, {
    role: 'billing',
    type: 'error',
    description: `Charge failed: ${(err as Error).message}`,
    metadata: { orderId: order.id },
  });
  // branch on next
}

conditional in depthEvery option, return value and guarantee of the wait.

Batches and accumulatorsCollect many contributions on one row and wake once.

Errors become escalationsA failure lands in a queue with its context, not in a log.

Tools as durable activitiesThe same function is an activity and an MCP tool.