pulse.yaml en un coup d'œil

Vérifié contre les schémas d'opérateurs livrés et les exemples de référence — c'est la syntaxe que le validateur au deploy accepte.

name: my-app                # id de l'app (les topics deviennent my-app.in, my-app.<stage>.out)
persist: results            # optionnel — collection App-Data durable, génère l'API

source:                     # OBSERVER — un type de connecteur + réglages.
  kind: webhook             # webhook · file-tail · file-source · http-poll · jdbc-source · …
                            # Omettez `source:` pour l'ingress natif par défaut.

stages:                     # DÉCIDER — ordonné ; un moteur par 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" }   # optionnel — gestion des retardataires

  - name: gate
    engine: rule-based
    rules:                               # des objets, pas des chaînes nues
      - { condition: "n > 20", action: flag }
      - { condition: "total >= 10000", action: "emit:review-queue" }

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

  - name: act
    engine: mcp
    mcpTools: [ tool.name ]              # s'affiche `blocked` tant que son plugin n'est pas installé

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

expose: [shape, gate]        # optionnel — flux SSE en direct pour ces stages

Les quatre moteurs

Moteur	À utiliser pour	Coût
streaming	mettre en forme le flux avec les opérateurs déclaratifs (ci-dessous)	pas de modèle — µs
rule-based	trancher : rules: [{ condition, action }] — actions : flag · block · hold · review · emit:<topic>	pas de modèle — µs
llm	raisonner sur le véritablement ambigu — systemPrompt, temperature / maxTokens optionnels	votre provider
mcp	agir : mcpTools: [ … ] — outils, tickets, workflows, approbation humaine	latence de l'outil

Opérateurs streaming (types canoniques)

filter (condition) · map (fields / targetType) · flatMap (splitField) · keyBy (field) · window (spec, aggregations, trigger) · cep (sequence, within) · enrich (lookupTopic, keyField) · enrichAsync (url, …) · branch · dedup (key, window optionnel) · 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 pour les pipelines avancés.

Entrée multi-topics : un stage consomme des topics supplémentaires avec inputTopics: [a, b] dans sa config ; un join s'abonne automatiquement à son rightTopic sans déclaration supplémentaire.

Secrets

${secret:NOM} (magasin de la CLI) et ${env:VAR} (environnement). Résolus au deploy ; une référence non résolue interrompt le deploy avec une erreur nommée.

Contraintes honnêtes

• Un stage llm sans provider joignable se déclare unhealthy — visiblement.
• Un stage mcp dont le plugin n'est pas installé s'affiche blocked dans la vue pipeline.
• La spec de fenêtre et chaque chaîne d'agrégation sont validées au deploy — une fonction inconnue est une 400 nommée, pas un écho silencieux.
• Les événements véritablement en retard (derrière le watermark) partent en dead-letter queue plutôt que de corrompre silencieusement les fenêtres.