task

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.

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`
Call an external API, an LLM, or anything with a connector	`serviceCall`
Read from / write to the app’s own database	`importFromDatabase` / `persistToDatabase`

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:

// 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`
  },
});

// 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

{
  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
`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 and reference the row’s columns with {{item.X}}. This is the common file-processing shape:

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

Platform Docs

Synthetiq Desktop

Synthetiq CLI

App Framework

Services Framework

Workflows Framework

Platform API

Deployments

When to use a task step

Defining a task

Referencing a task in a workflow

Configuration

The task context

The canonical pipeline: import → task → persist

Execution semantics

Validation

​When to use a task step

​Defining a task

​Referencing a task in a workflow

​Configuration

​The task context

​The canonical pipeline: import → task → persist

​Execution semantics

​Validation

When to use a task step

Defining a task

Referencing a task in a workflow

Configuration

The task context

The canonical pipeline: import → task → persist

Execution semantics

Validation