Pipeline YAML

A pipeline config is a YAML file. Schema below is validated at create-time (factflow config create), pre-flight (factflow config validate), and server startup (strict by default).

Top-level

version: "1.0"

routes:
  <route_id>:
    ...

init_message:
  route: <route_id>
  payload: {...}

error_handling:   # optional, inherits to all routes
  ...

backpressure:     # optional, inherits to all routes
  ...

One version: — today "1.0". Future major changes go behind version gates.

Route schema

<route_id>:
  name: "Human-readable name"
  description: "What this route does"

  inbound:
    queue: "/queue/route.inbound"       # broker-specific destination
    subscription: "processors-name"      # shared subscription across processors
    concurrency: 5                        # max in-flight messages per processor
    prefetch: 10                          # broker-side buffer

  adapters:
    - type: "adapter_name"
      config:
        ...adapter-specific fields...
      condition:                          # optional; gate on message fields
        field: "some_field"
        operator: "matches"
        value: "^https://"

  error_handling:                         # optional override
    max_retries: 3
    retry_delay_seconds: 10
    dead_letter_queue: "/queue/dlq"

  backpressure:                           # optional override
    target_latency_ms: 500
    max_concurrency: 20

inbound.queue naming

Provider-specific:

• Artemis (STOMP) — /queue/... or /topic/...
• RabbitMQ (AMQP) — simple names like route.inbound
• Pulsar — persistent://tenant/namespace/topic

The server’s NamingStrategyRegistry picks up the configured provider and validates names against its rules. Invalid names fail at validate-time, not at first message.

Adapters

Every adapter in the adapters: list runs on every message the route consumes, in order. Each adapter’s type: must be registered in the adapter registry; unknown types fail with E103.

config: is validated against the adapter’s Pydantic config class. Unknown fields fail with E301; missing required fields fail with E302.

See the adapter catalog for every shipped adapter’s config shape.

Conditions

Skip an adapter if the incoming message doesn’t match:

- type: "web_scraper"
  condition:
    field: "url"
    operator: "matches"
    value: "^https://.*\\.dnb\\.no/.*"

Supported operators

From factflow_engine.condition_operators.OperatorRegistry:

Operator	Purpose
eq / equals	Equality
neq / not_equals	Inequality
in	Value is in a list
not_in	Value is not in a list
contains	String / list contains substring / item
starts_with	String starts with prefix
ends_with	String ends with suffix
gt / greater_than	Numeric greater than
lt / less_than	Numeric less than
matches	Regex match
exists	Field is present (non-null)

Custom operators register via OperatorRegistry.register_function(name, fn).

Init message

init_message:
  route: "sitemap_scraper"
  payload:
    sitemap_url: "https://example.com/sitemap.xml"
    # any JSON-serialisable fields your first adapter expects

One init message per execution. Published to the named route’s inbound queue when the execution starts. Required — without it no work ever happens.

Error codes

Stable taxonomy produced by the validator. Format: <severity><numeric>; E = error (fails validation), W = warning (validation succeeds with log output).

Global (E001–E099)

• E001 — YAML parsing error
• E002 — Unknown top-level key
• E003 — Required top-level key missing
• E004 — Type mismatch
• E501 — Version unsupported

Routes (E101–E209)

• E101 — Unknown route field
• E102 — Route id invalid (naming rules)
• E103 — Unknown adapter type
• E104 — Adapter list empty
• E105 — Duplicate route id
• E106 — Duplicate queue across routes
• E107–E109 — Inbound queue / subscription / concurrency errors
• E110–E111 — Adapter position errors
• E201–E209 — Condition errors (unknown operator, bad value, missing field)

Adapters (E301–E399)

• E301 — Unknown config field for this adapter type
• E302 — Required config field missing
• E303–E399 — Adapter-specific validation

Init message (E401–E499)

• E401 — Missing init_message
• E402 — Route referenced by init message doesn’t exist
• E403 — Payload fails the first adapter’s input validation

Warnings (W501–W503)

• W501 — Deprecated field
• W502 — Suboptimal concurrency setting
• W503 — Queue-name prefix convention not followed

Config snapshot invariant

When an execution starts, the full YAML config is frozen into config_snapshot in the execution row. Subsequent edits to the stored config don’t affect in-flight executions. Replay resolves route → queue from the parent execution’s snapshot.

Related

• Configuration guide — day-to-day create / edit / validate
• Adapter catalog — every adapter’s config
• Concept: Pipelines