Test Workflow: Daily Briefing Pipeline

Purpose

First integration test for Cambium. Simple enough to debug, complex enough to exercise all major subsystems.

Workflow Overview

[Source A] ──┐
[Source B] ──┼──→ [Merge] → [Summarize] → [Format] → [Review Gate] → [Deliver]
[Source C] ──┘        ↑
   │                  │
   └─ (fallback) ─────┘

Agents Required

Workflow Graph

Nodes

1. gather_a (task) — Fetch RSS feed A - capability: fetch_rss - retry: 2, fallback edge to merge

2. gather_b (task) — Fetch RSS feed B - capability: fetch_rss - retry: 2, fallback edge to merge

3. gather_c (task) — Fetch web source C - capability: fetch_web - retry: 1, fallback edge to merge

4. merge (task) — Merge gathered content - join: waits for all gatherers (succeeds with partial if some fail) - capability: text_analysis

5. summarize (task) — Summarize merged content - capability: summarization - checkpoint: before and after

6. format (task) — Format summary for delivery - capability: text_formatting

7. review_gate (approval) — Human reviews before delivery - policy: “briefing_review” (single approver) - timeout: 30 minutes (auto-approve if no response)

8. deliver (task) — Deliver formatted briefing - capability: delivery

9. success (terminal_success)

10. failure (terminal_failure)

Edges

• gather_a → merge (default)
• gather_a → merge (fallback, on failure after retries)
• gather_b → merge (default)
• gather_b → merge (fallback, on failure after retries)
• gather_c → merge (default)
• gather_c → merge (fallback, on failure after retries)
• merge → summarize (default)
• merge → failure (conditional: all sources failed)
• summarize → format (default)
• format → review_gate (default)
• review_gate → deliver (on approval)
• review_gate → failure (on rejection)
• deliver → success (default)

Expected Signal Flow (Happy Path)

1. lifecycle.run_started
2. lifecycle.agent_genesis × 6
3. coord.agent_ready × 6
4. coord.work_available × 3 (gather tasks)
5. [gather agents place quality markers on sources]
6. node_completed × 3
7. lifecycle.checkpoint_created (pre-summarize)
8. node_completed (merge)
9. [summarizer reads quality markers to weight sources]
10. node_completed (summarize)
11. lifecycle.checkpoint_created (post-summarize)
12. node_completed (format)
13. governance.approval_required
14. governance.approval_granted
15. node_completed (deliver)
16. lifecycle.run_completed
17. lifecycle.agent_composted × 6

Stress Test Variant

Inject these failures to test resilience:

Test A: Source failure

• Make Source C return 500
• Expected: gather_c fails, retries once, takes fallback edge
• Stigmergy: danger.error_prone marker on source C
• Merge succeeds with 2/3 sources
• Rest of workflow completes

Test B: LLM slowdown

• Add artificial delay to summarizer (5s → 15s)
• Expected: Health detects stress.latency_spike
• Stigmergy: path.slow marker on summarize path
• Run marked “degraded” in Loom
• Workflow still completes (no threshold crossed)

Test C: LLM failure with cascade

• Make summarizer fail completely (provider down)
• Expected: 1. Retries exhausted 2. stress.error_rate signal 3. Hypha checks fallback: none configured → workflow fails 4. Seed has checkpoint from pre-summarize 5. Operator can replay from checkpoint after provider recovers

Test D: Anomalous behavior

• Make gatherer_a return 10x normal content volume
• Expected: 1. Immune detects anomaly (abnormal output size) 2. immune.anomaly_detected (severity: 0.4) 3. Warning visible in Loom 4. Below quarantine threshold — workflow continues 5. Composting captures the pattern

Stigmergic Behavior

Markers placed during run:

• Each gatherer places quality.high or quality.low on their source region
• On failure: danger.error_prone on the source
• Summarizer reads quality markers to weight source reliability
• After successful delivery: path.hot on the entire workflow path

Cross-run learning:

• If Source C has danger.error_prone from previous runs (marker hasn’t decayed)
• Next run: Hypha may deprioritize Source C or allocate fewer resources to it
• This is stigmergic coordination — no rule says “skip C”, the environment says “C is unreliable”

Composting Value

After each run: - Which sources returned useful content (feed into source quality scoring) - Summarizer performance (latency, token usage) - Was the review gate useful or always auto-approved (policy tuning) - Template effectiveness for each agent type

Definition as JSON (for Cambium loader)

{
  "name": "Daily Briefing Pipeline",
  "description": "Gather news from multiple sources, summarize, format, review, deliver",
  "version": 1,
  "checkpoint_policy": {
    "at_nodes": ["summarize"],
    "position": "before_and_after"
  },
  "policy_bindings": {
    "review_gate": {
      "policy_id": "briefing_review",
      "type": "single_approver",
      "timeout_ms": 1800000,
      "timeout_action": "auto_approve"
    }
  },
  "graph": {
    "nodes": [
      { "id": "gather_a", "type": "task", "config": { "capability": "fetch_rss", "retry": 2, "source": "feed_a" } },
      { "id": "gather_b", "type": "task", "config": { "capability": "fetch_rss", "retry": 2, "source": "feed_b" } },
      { "id": "gather_c", "type": "task", "config": { "capability": "fetch_web", "retry": 1, "source": "web_c" } },
      { "id": "merge", "type": "task", "config": { "capability": "text_analysis", "join": "partial" } },
      { "id": "summarize", "type": "task", "config": { "capability": "summarization" } },
      { "id": "format", "type": "task", "config": { "capability": "text_formatting" } },
      { "id": "review_gate", "type": "approval", "config": { "policy": "briefing_review" } },
      { "id": "deliver", "type": "task", "config": { "capability": "delivery" } },
      { "id": "success", "type": "terminal_success" },
      { "id": "failure", "type": "terminal_failure" }
    ],
    "edges": [
      { "from": "gather_a", "to": "merge", "type": "default" },
      { "from": "gather_a", "to": "merge", "type": "fallback" },
      { "from": "gather_b", "to": "merge", "type": "default" },
      { "from": "gather_b", "to": "merge", "type": "fallback" },
      { "from": "gather_c", "to": "merge", "type": "default" },
      { "from": "gather_c", "to": "merge", "type": "fallback" },
      { "from": "merge", "to": "summarize", "type": "default" },
      { "from": "merge", "to": "failure", "type": "conditional", "condition": "all_sources_failed" },
      { "from": "summarize", "to": "format", "type": "default" },
      { "from": "format", "to": "review_gate", "type": "default" },
      { "from": "review_gate", "to": "deliver", "type": "default" },
      { "from": "review_gate", "to": "failure", "type": "conditional", "condition": "rejected" },
      { "from": "deliver", "to": "success", "type": "default" }
    ]
  }
}