> ## Documentation Index
> Fetch the complete documentation index at: https://synthetiq.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# task

> Run your own app code as a workflow step

A **task step** runs an app-defined function as a workflow step — for work SQL and service
calls can't do, like processing a file (transcode a video, render a thumbnail, parse a PDF) or
any bespoke computation. Its return value becomes the step's results table, exactly like a
service step, so later steps read it as an ordinary table.

> [!NOTE]
>
> **Developer-only.** Task steps run arbitrary app code, so — like `persistToDatabase` and
> `importFromDatabase` — they are for workflows you author in `src/server/workflows/`. The
> app's in-app AI agent cannot compose a task step; the framework rejects any job it submits
> that contains one.

## When to use a task step

A task is an **escape hatch** for work the declarative steps can't express. Reach for the
specific step first, and use a task only when none of them fit:

| You need to… | Use |
| - | - |
| Transform / aggregate / join data already in the workflow | [`sql`](/docs/platform-docs/workflows-framework/steps/sql) |
| Call an external API, an LLM, or anything with a connector | [`serviceCall`](/docs/platform-docs/workflows-framework/steps/service-call) |
| Read from / write to the app's own database | [`importFromDatabase`](/docs/platform-docs/workflows-framework/steps/import-from-db) / [`persistToDatabase`](/docs/platform-docs/workflows-framework/steps/persist-to-db) |

Use a **task** when you genuinely need to run code — typically over **bytes or files**, not rows:

* Processing an uploaded or generated file: transcode, resize, thumbnail, parse a PDF / CSV / XLSX, extract text, convert formats
* CPU-bound computation, or a library that operates on file contents
* Anything that must open a blob, do real work, and write a new blob

If the logic is really "call an external service," make it a **service** instead — services are
typed, reusable, and available to the in-app agent. Keep tasks small, deterministic, and
idempotent.

## Defining a task

Put each task in its own file under `src/server/tasks/`, and use `defineTask` with Zod schemas.
The function's params and return type are checked against those schemas **at build time**:

```typescript theme={"system"}
// src/server/tasks/processVideo.ts
import { defineTask } from "@synthetiq/app-framework/server";
import { z } from "zod";

export const processVideo = defineTask({
  input: z.object({ blobId: z.string(), fps: z.number().default(24) }),
  output: z.object({ newBlobId: z.string(), frameCount: z.number() }),
  async run(params, ctx) {        // params: { blobId: string; fps: number }
    const src = await Storage.open(params.blobId);           // stream in
    // ...do the work...
    const newBlobId = await Storage.put(outStream, { contentType: "video/mp4" });
    return [{ newBlobId, frameCount: 128 }];                 // must match `output`
  },
});
```

Register tasks in a thin manifest — one line per task:

```typescript theme={"system"}
// src/server/tasks/index.ts
import { processVideo } from "./processVideo";
import { generateThumbnail } from "./generateThumbnail";

export const tasks = { processVideo, generateThumbnail };
```

A plain object (`{ inputSchema, outputSchema, run }`) is also accepted, but `defineTask` is
preferred — it is what gives the compile-time param and return guarantee.

## Referencing a task in a workflow

```typescript theme={"system"}
{
  task: "processVideo",
  params: { blobId: "{{item.blob_id}}", fps: "{{var.fps}}" },
  outputTable: "processed",
  outputSchema: { newBlobId: "TEXT", frameCount: "INTEGER" },
}
```

### Configuration

| Field | Required | Description |
| - | - | - |
| `task` | Yes | Task name from the app's task registry (`src/server/tasks`) |
| `params` | No | Input to the task. Supports `{{var.X}}` and `{{item.X}}` |
| `outputTable` | Yes | Results table name |
| `outputSchema` | Yes | Columns written from the return value |
| `forEach` | No | Run once per source row — see [forEach](/docs/platform-docs/workflows-framework/foreach.md) |
| `maxRetries` | No | Per-step retry count |

The return value is written to `outputTable` / `outputSchema` like a service step: an array of
rows, a single row object, or a `{ data: [...] }`-shaped value.

## The task context

`run(params, ctx)` receives a context assembled by the worker:

| Field | Description |
| - | - |
| `userId` | The triggering user, or `system:*` for a scheduled job |
| `userScopes` | The caller's effective scopes, resolved at execution |
| `db` | A database client scoped to the caller (use this, not the unscoped client) |
| `idempotencyKey` | Stable across retries of this step — use it to make side effects idempotent |
| `jobId` / `stepId` / `taskName` / `attempt` | Run metadata |
| `log` | Structured logger |

Org is ordinary app data — pass it in via a `param` (for example `{{var.orgId}}`) like any other value.

## The canonical pipeline: import → task → persist

A task takes its input from `params`; to act on a prior step's rows, wrap it in
[`forEach`](/docs/platform-docs/workflows-framework/foreach) and reference the row's columns with
`{{item.X}}`. This is the common file-processing shape:

```typescript theme={"system"}
steps: [
  {
    id: "files",
    importFromDatabase: { appId: "{{var.appId}}", tableName: "Upload" },
    outputTable: "files",
    outputSchema: { id: "TEXT", blob_id: "TEXT" },
  },
  {
    id: "process",
    task: "processVideo",
    forEach: { source: "files", itemKey: "id" },
    params: { blobId: "{{item.blob_id}}", fps: "{{var.fps}}" },
    outputTable: "processed",
    outputSchema: { newBlobId: "TEXT", frameCount: "INTEGER" },
    dependsOn: ["files"],
  },
  {
    id: "save",
    persistToDatabase: {
      appId: "{{var.appId}}",
      tableName: "Upload",
      sourceTable: "processed",
      mode: "upsert",
      upsertKey: ["id"],
      fieldMapping: { processedBlobId: "newBlobId" },
    },
    dependsOn: ["process"],
  },
]
```

## Execution semantics

Execution is **at-least-once**. The results table is overwritten on retry (idempotent by
construction); make any *side effects* idempotent yourself using `ctx.idempotencyKey`. Retries,
`maxRetries`, and heartbeats match every other step type.

## Validation

| Check |
| - |
| `task` names a task that exists in the app's registry |
| `outputTable` and `outputSchema` are required |
| Params are present, known, and type-check against the task's `input` (literal, `{{item.X}}`, and `{{var.X}}` values) |
| `outputSchema` columns are ones the task actually returns |
| The task's output flows into cross-step checks (a downstream `sql` step referencing a missing column fails the build) |
| `defineTask` compiles — the function's params and return type match its schemas |


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.