pulse.yaml at a glance

Verified against the shipped operator schemas and the reference examples — this is the syntax the deploy-time validator accepts.

name: my-app                # app id (topics become my-app.in, my-app.<stage>.out)
persist: results            # optional — durable App-Data collection, drives the generated API

source:                     # OBSERVE — one connector kind + settings.
  kind: webhook             # webhook · file-tail · file-source · http-poll · jdbc-source · …
                            # Omit `source:` entirely for the default native ingress.

stages:                     # DECIDE — ordered; one engine per stage
  - name: shape
    engine: streaming
    operators:
      - { type: keyBy, field: user_id }
      - type: window
        spec: "sliding(5m,1m)"          # tumbling(60s) · sliding(60s,10s) · session(30s)
        aggregations:                    #   · count(100) · count_sliding(100,10) · global()
          n:    "count()"
          total: "sum(amount)"
          hi:   "countWhere(amount > 100)"
        trigger: { type: late, grace: "5m" }   # optional — late-event handling

  - name: gate
    engine: rule-based
    rules:                               # objects, not bare strings
      - { condition: "n > 20", action: flag }
      - { condition: "total >= 10000", action: "emit:review-queue" }

  - name: reason
    engine: llm
    systemPrompt: |                      # provider: ANTHROPIC_API_KEY or local Ollama
      …
    temperature: 0.2                     # optional
    maxTokens: 256                       # optional

  - name: act
    engine: mcp
    mcpTools: [ tool.name ]              # shows `blocked` until its plugin is installed

sink:                        # ACT — optional
  kind: webhook              # webhook · slack · kafka · jdbc · mqtt · s3 · file · …
  url: ${secret:HOOK}

expose: [shape, gate]        # optional — live SSE feeds for these stages

The four engines

Engine	Use for	Cost
streaming	shape the stream with declarative operators (below)	no model — µs
rule-based	gate: rules: [{ condition, action }] — actions: flag · block · hold · review · emit:<topic>	no model — µs
llm	reason over the genuinely ambiguous — systemPrompt, optional temperature / maxTokens	your provider
mcp	act: mcpTools: [ … ] — tools, tickets, workflows, human approval	tool latency

Streaming operators (canonical kinds)

filter (condition) · map (fields / targetType) · flatMap (splitField) · keyBy (field) · window (spec, aggregations, trigger) · cep (sequence, within) · enrich (lookupTopic, keyField) · enrichAsync (url, …) · branch · dedup (key, optional window) · join (leftTopic, rightTopic, window, on, joinType: inner|left|right|full) · process (timeout, key, timer: processing|event, onTimeout: emit|drop, stateFields) · reduce (function: sum|min|max|first|last|count|concat, field) · aggregate (aggregations, key) · explode (arrayField, elementField) · plus wasm, mapLlm, mcpCall, mlPredict, bandit for advanced pipelines.

Multi-topic input: a stage consumes extra topics with inputTopics: [a, b] in its config; a join auto-subscribes its rightTopic without any extra declaration.

Secrets

${secret:NAME} (CLI secret store) and ${env:VAR} (environment). Resolved at deploy; an unresolved reference aborts the deploy with a named error.

Honest constraints

• An llm stage without a reachable provider reports itself unhealthy — visibly.
• An mcp stage whose plugin isn't installed shows blocked in the pipeline view.
• The window spec and every aggregation string are validated at deploy — an unknown function is a named 400, not a silent echo.
• Genuinely-late events (behind the watermark) route to a dead-letter queue rather than silently corrupting windows.