Write the agent task

Now the code. A CXP agent is an ordinary Trigger.dev task — there is no base class to extend and no framework to adopt. What makes it an agent is that it understands one payload shape and reports its state back.

1. What the platform sends

Every platform-triggered run arrives with the same envelope, regardless of what started it:

type CxpaPayload<TInput> = {
  /** Platform run id. Its presence means the platform owns the run row. */
  runId: string
  /** Whatever was in the input surface, the schedule, or the trigger call. */
  input?: TInput
  /** integration_id → connection, for this agent's credential requirements. */
  connectionIds?: Record<string, { integrationId: string; connectionId: string }>
  /** Short-lived (60 min) JWT that authorises credential resolution. */
  credentialsToken?: string
}

The canonical definition lives in the SDK reference — CxpaPayload — and is repeated here because it is the first thing a task has to get right.

Note what is not there. No credential values — only connection ids and the credentialsToken, which resolve a secret together with the agent API key. No callback URL and no callback-specific token, because the callback is addressed by runId and authenticated by that same agent API key. And no agent API key itself: it is the one factor the payload never carries, and your task reads it from the environment.

Your domain input is nested under input, not spread at the top level. This is the mistake most first tasks make.

Resolve every credential you need before branching, not at the point of use. The credentials token is valid for 60 minutes from the start of the run, so a task that first resolves at minute 90 — or inside a branch it only sometimes takes — gets a 401 deep into a long job. See Credentials.

2. The task skeleton

The minimum viable agent reads its input, does the work, and returns a result:

import { task } from '@trigger.dev/sdk'

type Input = { question: string; sessionId: string }

type Payload = {
  runId?: string
  input?: Input
  connectionIds?: Record<string, { integrationId: string; connectionId: string }>
  credentialsToken?: string
}

export const revenueChatAgent = task({
  id: 'revenue-chat-agent',
  maxDuration: 300,
  run: async (payload: Payload) => {
    const question = payload.input?.question
    if (!question) throw new Error('revenue-chat-agent: missing input.question')

    const answer = await answerQuestion(question)

    return {
      answer,
      outputs: 1,
      generatedAt: new Date().toISOString(),
    }
  },
})

The task id must match the Task ID field on the agent record exactly.

That really is enough. There is no callback here, and the run still shows up on the platform with its status, output, and duration.

3. Reporting state back

The platform tracks every platform-managed run by polling the runtime. Each run gets its own poll job at the agent's status poll interval (60 seconds by default). On each tick the platform asks the runtime for the run's status, and on the transition to a terminal state it fetches the result and records:

output_data — the value your task returned;
duration_ms — wall-clock time;
outputs — read from a top-level integer outputs key in that returned value.

So a task that just returns gets the full lifecycle for free. Note the two consequences of that mechanism: the returned value must be an object with at least one key (an empty object is left as null rather than written), and outputs has to be top-level in it — which is why the skeleton above returns outputs: 1 alongside the answer.

When to call back instead

The callback API exists for what polling cannot give you: immediacy, and states the runtime cannot express.

import { createAgentClient } from '@cxpa/sdk/agent'

const cxpa = createAgentClient({
  baseUrl: process.env.CXPA_API_URL!,
  apiKey: process.env.CXPA_REVENUE_CHAT_AGENT_API_KEY!,
  runId: payload.runId!,
  credentialsToken: payload.credentialsToken,
})

Call	Effect on the run	Could polling do it?
`cxpa.started()`	status → `running`	Yes, within one interval
`cxpa.output({ output })`	stores intermediate output mid-run	No
`cxpa.complete({ output, outputs })`	stores output, status → `completed`	Yes, within one interval
`cxpa.fail({ message, name?, stack? })`	stores a structured error, status → `failed`	Partly — polling records the failure, not your message
`cxpa.waiting({ tokenId, description, url? })`	status → `waiting`	No

Reach for it when:

A person is watching. Polling can leave a finished run showing as running for up to a poll interval. For an interactive agent that is the difference between instant and sluggish.
You want progress during the run. output() is the only way to record intermediate state; polling sees nothing until the run ends. Batch it — the callback endpoint allows 20 requests per minute per agent, shared with waitpoint completions, so one call per processed item will start returning 429s on any real batch. Report every N items or every few seconds, and use runtime metadata, which is not rate-limited, for finer progress.
The agent pauses for a human. A waitpoint has to be announced — the platform cannot infer waiting from runtime status.
The error message matters. fail() records your message; polling records that the run failed.

For an unattended job like nightly-etl, none of those apply — returning a value is the right amount of machinery. For revenue-chat, where someone is waiting on the answer, the callback is worth it.

Doing both is safe. Whichever path writes a terminal status first wins, and the other becomes a no-op — the platform guards against overwriting a terminal state. So a task can call complete() and return the same object, which is also what the synchronous endpoints (trigger-and-wait) and anything subscribed to the run on the runtime will read.

Lowering the agent's poll interval is the other lever: the minimum is 10 seconds, at the cost of more requests to the runtime.

4. The `outputs` count

outputs is an integer, and it is not the same thing as output. output is your domain payload; outputs is how many units of work the agent produced — rows written, images generated, reports sent, questions answered. The platform aggregates it into the value metrics a customer sees.

Pick a unit that means something for this agent and stay consistent: outputs: 1 per answered question for revenue-chat, outputs: <rows loaded> for nightly-etl. Do not leave it out — a metric of zero is indistinguishable from an agent that does nothing.

Choose the unit as the thing the customer would count. The platform multiplies it by two rates configured per agent — hours saved per output, and cost per output — to produce the hours-saved and ROI figures on the customer's dashboard. So the right unit is whatever a person would have had to do by hand: emails triaged, invoices reconciled, reports written. "API calls made" or "rows scanned" are activity, not value, and they make the ROI number meaningless.

Two consequences of how it aggregates:

Only successful runs contribute. A failed run still counts as a run, but its outputs are ignored.
The rates are applied at read time, not frozen onto the run. Changing an agent's hours-per-output later rewrites every historical figure on the dashboard. Pick the unit once and leave it alone; changing what one output means silently rewrites history.

5. A scheduled task

nightly-etl differs from the on-demand shape in two ways: its input comes from the schedule, and it has to know where it left off.

import { task } from '@trigger.dev/sdk'

type Input = { mode?: 'incremental' | 'full' }

export const nightlyEtl = task({
  id: 'nightly-etl',
  maxDuration: 7200,
  machine: 'large-1x',
  run: async (payload: { runId?: string; input?: Input }) => {
    const full = payload.input?.mode === 'full'

    // Where we left off. Stored in YOUR database, not the platform's.
    const since = full ? EPOCH : await readWatermark('nightly-etl')

    const batch = await fetchChangedSince(since)
    if (batch.length === 0) {
      return { loaded: 0, outputs: 0, since }
    }

    // Dedupe on a natural key so a re-run or an overlapping run is harmless.
    const loaded = await upsertByNaturalKey(batch)

    // Advance only after the write succeeds, and only to what we actually read.
    await writeWatermark('nightly-etl', maxTimestamp(batch))

    return { loaded, outputs: loaded, since }
  },
})

The watermark is yours. The platform stores JSONB per run and has no notion of "where this agent got to" — there is no cursor, no last-run pointer, nothing to read at the start of a run. Keep it in your own database, keyed by agent.

Three rules make it safe, and all three come from the scheduling behaviour described in Triggering runs:

Advance the watermark after the write, never before. A crash between the two must re-process, not skip.
Advance it to what you actually read, not to "now". Records that arrive late but timestamped earlier are lost otherwise.
Dedupe on a natural key regardless. Overlapping runs are possible, missed fires are not backfilled, and a re-run of a failed run will re-read the same window. Idempotent writes make all three harmless.

A full-reload mode is worth having from the start — it is the recovery path when the watermark is wrong, and it costs one branch.

6. Accepting a second trigger source

If a zone app triggers the runtime directly rather than going through the platform, the payload arrives in whatever shape the zone chose and there is no runId — the platform does not know the run exists yet. A task can support both sources by reading its input from either position and using the presence of runId to tell them apart:

import { registerRun } from '@cxpa/sdk/agent'

const question = payload.question ?? payload.input?.question
const triggeredByPlatform = Boolean(payload.runId)

let runId = payload.runId
let credentialsToken = payload.credentialsToken
let connectionIds = payload.connectionIds

if (!triggeredByPlatform) {
  // The platform has no run row for this execution — announce it, and
  // receive the same credential pair a trigger payload would have carried.
  try {
    const res = await registerRun({
      baseUrl: process.env.CXPA_API_URL!,
      ingestApiKey: process.env.CXPA_REVENUE_CHAT_INGEST_API_KEY!,
      agentId: payload.agentId!,
      externalRunId: ctx.run.id,
      input: { question },
    })
    runId = res.runId
    credentialsToken ??= res.credentialsToken
    connectionIds ??= res.connectionIds
  } catch (err) {
    logger.warn('registerRun failed — continuing without platform tracking', {
      error: err instanceof Error ? err.message : String(err),
    })
  }
}

Three details matter here.

Only register when the platform did not trigger you. Calling it on a platform-triggered run races the platform's own update of the same row.

Registering gets you credentials too. The response carries connectionIds alongside the token — the same pair, in the same shape, that a trigger payload would have delivered. So the resolution code from Credentials works unchanged on this path; feed it whichever source produced a value.

Decide deliberately what a registration failure means. Swallowing it, as above, keeps the agent working when only run tracking was unavailable — reasonable for an agent whose real job is answering a question. But note that a swallowed failure also costs you the credential pair, so let it propagate if the task cannot do its job without customer secrets.

registerRun requires the agent's Allow external triggers switch to be on, and it verifies that the run really exists in the runtime before issuing a token.

Most integrations still do not need this. Prefer having the zone trigger through the platform — see Build the zone app — which creates the run row for you and keeps the ingest key out of your runtime entirely.

7. Failing on purpose

Distinguish two kinds of failure.

An unexpected error — a network blip, a transient upstream 503 — should throw normally, so Trigger.dev's retry policy gets a chance:

throw new Error('warehouse unreachable')

A definite failure that retrying cannot fix should abort the run outright:

import { AbortTaskRunError } from '@trigger.dev/sdk'

if (breaches.length > 0) {
  throw new AbortTaskRunError(
    `nightly-etl completed but reconciliation found ${breaches.length} breach(es)`,
  )
}

AbortTaskRunError fails the run without consuming retry attempts. Use it for validation failures, reconciliation breaches, and bad input — anything where a second attempt would produce the same result at the same cost. Re-running an hour-long ETL twice to reach the same conclusion is expensive and it delays the alert.

8. Resource knobs

For anything heavier than an API call, set these explicitly:

export const nightlyEtl = task({
  id: 'nightly-etl',
  maxDuration: 7200,
  machine: 'large-1x',
  retry: {
    maxAttempts: 2,
    minTimeoutInMs: 120_000,
    maxTimeoutInMs: 300_000,
    factor: 2,
    randomize: true,
  },
  run: async (payload, { ctx }) => { /* … */ },
})

maxDuration in seconds. Keep it consistent with the agent's Max run duration on the platform, which is in minutes.
machine — bump it when the task holds a lot in memory. Out-of-memory failures look like unexplained crashes with no useful stack.

Handling overlap — not with a queue

Two runs of the same agent overlapping is a real risk: a schedule plus a manual Run Now makes it easy, and for anything writing to shared tables it corrupts data.

The obvious fix — a runtime queue with concurrencyLimit: 1 — does not work on this platform. A run waiting in the runtime's queue is reported as queued, and the platform fails any run that stays queued for more than 10 minutes, with "Run never started (queued timeout exceeded)". The ceiling is not configurable per agent. Worse, the runtime still holds the run and will execute it when the queue drains, so you get a run recorded as failed that actually ran.

Serialise inside the task instead, and return early rather than waiting:

run: async (payload) => {
  const lock = await acquireLock('nightly-etl')   // advisory lock, or a status row
  if (!lock) {
    return { skipped: true, reason: 'already running', outputs: 0 }
  }
  try {
    return await doTheWork(payload)
  } finally {
    await releaseLock(lock)
  }
}

A skipped run completes normally, records why, and leaves no ambiguity in the run history — which is better than a queued run that may or may not have been dropped.

Pair it with deduplication on a natural key — the source record's id, the provider's message id — so that even two runs that do overlap cannot double-process the same item. That protects you regardless of triggering, which matters because most trigger paths do not deduplicate at all; see Errors § Idempotency.

9. Deploy

bash

npm run trigger:dev

runs the task on your machine against the Trigger.dev development environment — the fast inner loop. It must stay running: a run triggered against a dev key with no trigger dev process attached just queues silently.