# Author a production task

> Source: https://docs.clonepartner.com/guides/task-authoring/

An Envoy task is one streaming pipeline. It declares the connector types it needs, accepts typed runtime arguments, and passes records through named steps.

Keep task YAML portable: declare connector slots by name and type, but keep URLs, credentials, paths, and account-specific configuration in stored connectors or `connectors.yaml`.

## Design before writing YAML

Define:

1. the source and its pagination behavior;
2. the destination operation;
3. a stable source identity;
4. the canonical payload used for change detection;
5. failure and retry behavior;
6. the expected result and reconciliation checks;
7. the smallest useful test dataset.

A task has one source stream. Split independent source pulls into separate tasks and compose them with a job. Keep dependent child pulls in the same task when each child query uses its parent `input`.

## Start from a minimal task

```yaml
name: contacts/stage
description: Stage contacts from CSV into SQLite
output: none

connectors:
  source:
    type: csv
  staging:
    type: sqlite

arguments:
  dry_run:
    type: boolean
    default: true

steps:
  - name: read_contacts
    type: call_method
    config:
      connector: source
      method: read
      args:
        resource: contacts.csv

  - name: normalize
    type: transform
    config: |
      {
        "id": $string(input.id),
        "name": $trim(input.name),
        "email": $lowercase(input.email)
      }

  - name: write_contact
    type: call_method
    run_if: "arguments.dry_run = false"
    config: |
      {
        "connector": "staging",
        "method": "upsert",
        "args": {
          "table": "contacts",
          "data": input,
          "conflict_columns": ["id"]
        }
      }
```

`output: none` is appropriate because the final step writes to a connector. Use `ndjson` when stdout is the intended data output. Avoid `json` for large pipelines because it collects all results.

## Use static and dynamic configuration correctly

A YAML object is static. Use it when the step does not need the current record. Use a JSONata string when referencing `input`, `arguments`, `variables`, `job_config`, or earlier step output.

`call_method` replaces the current `input` with the connector result. If a later step needs the original identity after a write, carry it in a preceding transform and read `steps.<name>.output`.

## Filter explicitly

`run_if` and `skip_if` decide whether a step executes; skipped records pass through unchanged. Use `filter` to remove records:

```yaml
- name: usable_email
  type: filter
  config: "$exists(input.email) and input.email != ''"
```

A transform that evaluates to a missing value also passes its original input through before execution. Do not use that behavior as a filter.

## Preserve streaming

Use a `call_method` step for list, find, read, and other collection methods. It can consume async results with backpressure.

Use `$callMethod()` inside JSONata only for bounded lookups such as `find_one`, `count`, or a grouped aggregate. It buffers the complete method result.

Add `concurrency` to a slow per-record step only after measuring correctness and provider limits:

```yaml
- name: write_contact
  type: call_method
  concurrency: 5
  error_handling: ignore
  config: |
    { ... }
```

Use `batch` to emit fixed-size arrays. Use `spool` only when the next step truly requires the whole dataset; it is intentionally unbounded.

## Choose error handling

- `fail` stops the task and is the default.
- `ignore` records the error, drops that step result, and continues.
- `ignore_rest` skips remaining steps for the affected item.

If a write uses `ignore`, guard the success-mark step against a missing result and record a separate error state. Otherwise a failed write can be marked successful.

## Validate and test

In the UI, validate before saving. From the CLI:

```bash
envoy validate tasks/contacts/stage.yaml
```

Static validation catches structure, known step types, connector references, argument schemas, and expression compilation. It does not validate live credentials, upstream fields, or evaluation against real data.

Use this progression:

1. inspect one representative source record;
2. run with writes disabled;
3. inspect transformed output and logs;
4. enable writes for a small bounded set;
5. verify destination records and mappings;
6. rerun the same set and confirm no duplicates;
7. test one invalid record;
8. scale page size or concurrency gradually.

See [Task structure](/reference/tasks/task-structure), [Step types](/reference/tasks/step-types), and [Task patterns](/reference/tasks/patterns).