Streams

What it is. Swirls-managed persistent storage for workflow output: structured, typed records.

Use it when you want to keep what a workflow produced and reuse it, in other workflows or later runs of the same one.

Works with workflows (the source), type: stream nodes (the reader), and schemas (the record shape). For files use a disk; for your existing database use postgres.

Streams in .swirls files are declared at the top level: a stream block names a persisted store, points at a source workflow, picks an active writer version, and defines one or more versions: entries with per-version schema, optional condition, and required prepare @ts blocks. A separate workflow uses a type: stream node with an explicit version pin to read rows from that version's table.

Stream data is written and read only in deployed projects. Deploy to Swirls Cloud to exercise streams end to end.

Top-level stream block

Per-version fields (inside versions:):

In condition and prepare, use context.nodes for node outputs and context.output for leaf node outputs keyed by node name (see below).

stream store_topic_tokens {
  label: "Store topic tokens"
  workflow: count_topic_tokens
  enabled: true
  version: v1

  versions: {
    v1 {
      schema: @json {
        {
          "type": "object",
          "required": ["topic", "tokens"],
          "properties": {
            "topic": { "type": "string" },
            "tokens": { "type": "number" }
          }
        }
      }
      condition: @ts {
        return context.output.tokens.tokens > 0
      }
      prepare: @ts {
        return {
          topic: context.nodes.root.output.topic,
          tokens: context.output.tokens.tokens,
        }
      }
    }
  }
}

Deploy provisions one Postgres table per (stream, version) (for example stream_store_topic_tokens_v1 in the project schema). Workflow completion writes only to the deployment's current version. Re-deploying with a changed schema for an existing version id fails with a drift error. Evolve a stream by adding a new versions: entry and moving the version: pointer; readers stay pinned to their version until you migrate them.

Persist failures are soft. A throwing condition or prepare, or a prepare result that fails schema validation, skips the row and records an audit event. The workflow run itself does not fail.

context.output and switch workflows

• Leaves not downstream of a switch are treated as must-run; their keys on context.output are required in typings.
• Leaves downstream of a switch may not run in a given execution; each such leaf is optional on context.output.
• The LSP does not model mutual exclusion between branches, so multiple optional leaves may appear even though only one runs at runtime.

In prepare, narrow which branch ran (for example with if (context.output.short_tokens) { … }) before returning the row. See stream-switch.swirls in the language examples package for a full pattern.

type: stream node

Reads rows from a named stream block at a pinned version.

node pull {
  type: stream
  label: "Matching rows"
  stream: store_topic_tokens
  version: v1
  filter: @ts {
    return {
      topic: { eq: context.nodes.root.output.topic },
    }
  }
}

Filter operators: eq, ne, gt, gte, lt, lte, like, in. like uses SQL LIKE semantics with % wildcards; in with an empty array matches nothing. Multiple top-level keys and multiple operators on one field combine with AND. Field keys name either a payload field from the version's schema or a system field via the bare aliases id, created_at, deployment_id, workflow_execution_id (mapped to the underscore-prefixed system columns). System fields are filterable but stripped from returned rows.

Downstream nodes read the array as context.nodes.pull.output (or your node name). Rows come back newest first. There are no pagination, sorting, or limit controls; sort: and limit: fields are not parsed. Zero matches is not an error.

A stream node has no incoming edge, so it counts as a root candidate. When one workflow reads several streams, fan out from a single real root { } to each stream node, or validation fails with multiple roots.

Row keys come back snake_case

Payload keys are snake_cased into columns on write, and reads do not convert them back. A discoveredAt field in prepare comes back as discovered_at on read. Filters match camelCase keys case-insensitively against their snake_case columns; returned rows do not. Prefer snake_case keys in prepare so the written and read shapes match.

Where stream data lives

After deploy, each stream version has a dedicated table under the project stream schema (project_<uuid>.stream_<name>_<version>). The platform maps JSON Schema properties to Postgres columns; required paths become NOT NULL. Every table carries four system columns: _id, _created_at, _deployment_id, _workflow_execution_id. Writers insert on workflow completion with the active deployment tagged on each row. Stream nodes and the API read by pinned version across all deployments in the project (optional filters can still target deployment_id).

Stream storage is filter-only. There is no raw SQL surface over stream tables; for raw SQL against your own database use a postgres node.

Common mistakes

Using the removed SQL query form. query: and querySql: on stream nodes were removed. The validator errors with querySql and query are no longer supported on stream nodes; use filter (@ts returning a filter object). Return a filter object from filter: @ts { } instead.

Omitting version on a stream node. Every read pins a version explicitly. There is no implicit default.

Leaving stream nodes parentless. Each stream node without an incoming edge counts as a root. Fan out from one real root { } when reading multiple streams.

Treating the output as one record. A stream node returns an array of all matching rows, newest first. Iterate or take the first row.

Further reading

• Workflows: DAG structure
• Context: context.nodes and iteration
• Syntax: @ts / @json blocks
• Views: spreadsheets composed from streams