End-to-End Pipeline

Note: the Mermaid diagrams below include click links with _blank targets. They work in Mermaid renderers that allow interactivity, usually with securityLevel: "loose". Some Markdown hosts may render the diagram but disable clicks or new-window behavior.

Full Request Path

flowchart TD
  A["WAHA Core"] --> D["POST /analyze"]
  B["Webhook / ping"] --> D
  C["Playground upload / testing tools"] --> D
  K["Kafka content-intelligence.flow-runs.v1"] --> L["broker worker"]
  L --> D

  D --> E["FlowRunRequest validation"]
  E --> F["request_normalizer"]
  F --> G{"payload.source == waha and raw events?"}
  G -->|yes| H["adapters.waha.adapt_waha_payload"]
  G -->|no| I["FlowPayload validation"]
  H --> I
  I --> V["chat_name_unification"]
  V --> J["resolve_flow_recipe"]
  J --> Z["Temporal payload preflight"]
  Z -->|inline| M["ProcessConversationChunk"]
  Z -->|oversized eligible Daily| X["split manifest in private object store"]
  X --> Y["bounded partition child workflows"]
  Y --> AA["exact recomposition + prompt budget gate"]
  AA --> P
  M --> N{"recipe media policy"}
  N -->|audio/image/sticker/GIF/video/document/link enabled| O["media fan-out activities"]
  O --> OB{"source representation"}
  OB -->|audio/document/video| OF["bounded SSRF-safe stream to /tmp/wardian"]
  OB -->|image/sticker/GIF/link| OM["bounded in-memory bytes"]
  OF --> P
  OM --> P
  N -->|text only| P["run_flow_pipeline activity"]
  P --> Q["conversation_builder"]
  Q --> R["AI JSON step(s) primary + shadow"]
  R --> S["FlowRunResult"]
  S --> T["GET /runs/{run_id}"]
  S --> U["Kafka content-intelligence.flow-results.v1"]
  S --> W["Kafka content-intelligence.flow-diagnostics.v1 when degraded/failed"]

  click A "/dev/docs/waha-integration" "WAHA integration" _blank
  click B "/dev/docs/payload-contracts" "HTTP payload contracts" _blank
  click C "/dev/docs/payload-contracts" "Payload contracts for dev tools" _blank
  click D "/dev/docs/payload-contracts" "POST /analyze contract" _blank
  click E "/dev/docs/payload-contracts" "FlowRunRequest validation" _blank
  click F "/dev/docs/payload-contracts" "Request normalization" _blank
  click G "/dev/docs/waha-integration" "WAHA source detection" _blank
  click H "/dev/docs/waha-integration" "WAHA adapter" _blank
  click I "/dev/docs/payload-contracts" "FlowPayload contract" _blank
  click J "/dev/docs/flows-recipes" "Flow and recipe resolution" _blank
  click K "/dev/docs/broker-integration" "Broker input topic" _blank
  click L "/dev/docs/broker-integration" "Broker worker" _blank
  click M "/dev/docs/temporal" "Temporal workflow" _blank
  click Z "/dev/docs/temporal" "Temporal payload preflight" _blank
  click X "/dev/docs/temporal" "Internal split artifacts" _blank
  click Y "/dev/docs/temporal" "Partition child workflows" _blank
  click N "/dev/docs/flows-recipes" "Recipe media policy" _blank
  click O "/dev/docs/temporal" "Media fan-out activities" _blank
  click P "/dev/docs/prompts-stages" "Flow pipeline activity" _blank
  click Q "/dev/docs/conversation-builder" "Conversation builder" _blank
  click R "/dev/docs/prompts-stages" "AI JSON steps and prompts" _blank
  click S "/dev/docs/output-contracts" "FlowRunResult contract" _blank
  click T "/dev/docs/output-contracts" "Run status response" _blank
  click U "/dev/docs/broker-integration" "Broker result topic" _blank
  click V "/dev/docs/waha-integration" "Chat name unification" _blank
  click W "/dev/docs/broker-integration" "Broker diagnostics topic" _blank

Flow Families

flowchart LR
  A["smoke-test"] --> A1["smoke-test-default"]
  B["initial-ingestion-report"] --> B1["initial-ingestion-report-text-only"]
  B --> B2["initial-ingestion-report-text-audio"]
  B --> B3["initial-ingestion-report-all-media"]
  C["daily-summary-and-notifications-report"] --> C1["daily-summary-and-notifications-report-text-only"]
  C --> C2["daily-summary-and-notifications-report-text-audio"]
  C --> C3["daily-summary-and-notifications-report-all-media"]
  D["interweek-report"] --> D1["interweek-report-default"]

  click A "/dev/docs/flows-recipes" "Smoke test flow" _blank
  click A1 "/dev/docs/flows-recipes" "Smoke test recipe" _blank
  click B "/dev/docs/flows-recipes" "Initial ingestion flow" _blank
  click B1 "/dev/docs/flows-recipes" "Text-only recipe" _blank
  click B2 "/dev/docs/flows-recipes" "Text and audio recipe" _blank
  click B3 "/dev/docs/flows-recipes" "All-media recipe" _blank
  click C "/dev/docs/flows-recipes" "Daily summary flow" _blank
  click C1 "/dev/docs/flows-recipes" "Text-only recipe" _blank
  click C2 "/dev/docs/flows-recipes" "Text and audio recipe" _blank
  click C3 "/dev/docs/flows-recipes" "All-media recipe" _blank
  click D "/dev/docs/flows-recipes" "Interweek flow" _blank
  click D1 "/dev/docs/flows-recipes" "Interweek recipe" _blank

Adapter To AI Steps

sequenceDiagram
  participant API as FastAPI
  participant Adapter as WAHA Adapter
  participant Names as Chat Name Unification
  participant Workflow as Temporal Workflow
  participant Store as Private Artifact Store
  participant Media as Media Activities
  participant Builder as Conversation Builder
  participant AI as OpenRouter JSON Gateway

  API->>API: validate flow, settings, payload
  API->>Adapter: raw WAHA events when source=waha
  Adapter-->>API: FlowPayload with conversations
  API->>Names: collapse @lid/@c.us aliases and choose names
  Names-->>API: canonical conversations
  API->>Workflow: run_id + normalized request preflight
  alt inline budget
    Workflow->>Media: process only media enabled by recipe
  else oversized eligible Daily
    Workflow->>Store: content-addressed manifest + deterministic partitions
    Workflow->>Media: bounded partition child workflows
    Workflow->>Store: partition media results
    Workflow->>Store: reload and verify exact recomposition
  end
  Media->>Media: stream audio/document/video to owned temporary files
  Media->>Media: keep image/animated/link bytes under small in-memory limits
  Media-->>Workflow: transcriptions/descriptions/excerpts/link summaries
  Workflow->>Builder: FlowPayload + media_processed
  Builder-->>Workflow: rendered conversation documents
  Workflow->>AI: system prompt + user prompt, primary model
  Workflow->>AI: same prompt, shadow model
  AI-->>Workflow: JSON outputs
  Workflow->>Store: internal final result ref for split mode

Broker Path

flowchart LR
  A["Kafka topic content-intelligence.flow-runs.v1"] --> B["BrokerWorkerService"]
  B --> C["FlowRunEnvelope validation"]
  C --> D["workflow_launcher.start_analysis"]
  D --> E["Temporal workflow"]
  E --> F["RunStatusResponse completed/failed"]
  F --> G["FlowResultEnvelope"]
  G --> H["Kafka topic content-intelligence.flow-results.v1"]
  G --> J["content-intelligence.flow-diagnostics.v1 when degraded/failed"]
  C -->|invalid| I["content-intelligence.flow-runs.dlq.v1"]

  click A "/dev/docs/broker-integration" "Broker input topic" _blank
  click B "/dev/docs/broker-integration" "Broker worker service" _blank
  click C "/dev/docs/payload-contracts" "FlowRunEnvelope validation" _blank
  click D "/dev/docs/temporal" "Workflow launcher" _blank
  click E "/dev/docs/temporal" "Temporal workflow" _blank
  click F "/dev/docs/output-contracts" "Run status response" _blank
  click G "/dev/docs/output-contracts" "FlowResultEnvelope contract" _blank
  click H "/dev/docs/broker-integration" "Broker result topic" _blank
  click I "/dev/docs/broker-integration" "Broker DLQ" _blank
  click J "/dev/docs/broker-integration" "Broker diagnostics" _blank

Error Behavior

• Missing flow, unsupported flow, unknown recipe, recipe/flow mismatch, malformed payload, and WAHA payloads with no analyzable messages return verbose validation details.
• AI JSON parse/schema failures fail the step and surface on GET /runs/{run_id}.
• Optional media activity failures degrade the media item, render [MEDIA_ERROR],

and publish diagnostics while the run can still complete.

• Shadow model failures are non-fatal and recorded as failed shadow output.
• WAHA/CP resolves temporal evidence before media fan-out. In enforce/strict,

an untrusted clock is retained upstream for reconciliation and any non-zero quarantine count is rejected before CIS statistics, prompts, memory or reports; direct raw CIS input fails explicitly. In shadow/disabled, the side-contract is observational: CIS may render the supplied conversations, including messages whose timestamps remain untrusted.