Multi-Agent AI Processing Pipelines on Cloudflare Workers

Org Status: 🟡 Dormant Cloudflare: N/A Last Audited: 2026-04-28

Most multi-agent AI frameworks assume you have a Python server, a Redis cluster, and a deployment pipeline. They assume your agents will run for minutes, consume gigabytes of memory, and need a Kubernetes cluster to orchestrate. What if you could build the same thing on a serverless platform where each agent runs in isolation, costs nothing at rest, and scales to zero between invocations?

This article shows how to build multi-agent AI processing pipelines entirely on Cloudflare Workers. Not toy chatbots — production systems where specialized AI agents evaluate data, debate each other, form consensus, remember what they learned, and make recommendations. All running on edge infrastructure with sub-millisecond cold starts, D1-backed persistent memory, and per-request cost tracking measured in fractions of a cent.

What you will learn:

• How to architect a pipeline of specialized AI agent processors on Cloudflare Workers
• The Durable Object coordinator pattern (OpportunityHarness) that governs agent execution
• How to implement a multi-perspective debate system where agents argue, challenge, and reach consensus
• Convergence detection: knowing when agents have collectively figured something out
• Agent memory with core beliefs, episodic recall, and reflective updates
• LLM integration through Vercel AI SDK with Zod-validated structured output
• Budget-aware execution that adapts model selection based on remaining spend
• Durable Workflows for scatter-gather batch processing with automatic retries
• Why this Workers-native approach beats LangChain, CrewAI, and AutoGen for this class of problem

• The Problem: Why General-Purpose Agent Frameworks Fall Short
• Architecture Overview
• Core Concepts
  • The Agent Processor
  • The Harness Config
  • The Processing Pipeline
  • The Agent Ledger
• Pattern 1: The Durable Object Coordinator
• Pattern 2: Phased Pipeline Execution
• Pattern 3: The Debate System
• Pattern 4: Convergence Detection
• Pattern 5: Agent Memory and Core Beliefs
• Pattern 6: Budget-Aware Model Selection
• Pattern 7: Durable Workflow Orchestration
• Small Examples
• Comparisons
• Anti-Patterns
• References

You want to build a system that takes raw data — say, market opportunities, content ideas, or investment signals — and runs it through multiple AI agents that each analyze a different dimension. One agent evaluates demand. Another challenges assumptions. A third looks for hidden risks. A fourth synthesizes everything into a recommendation.

The standard approach is to reach for LangChain, CrewAI, or AutoGen. Here is what happens next:

You deploy a Python server. Now you need a VM, a container, or a serverless function with Python runtime support. Cold starts are measured in seconds. You pay for idle time. You need to manage dependencies, virtual environments, and deployment pipelines.

You add Redis for state. Agents need to share context, remember past runs, and coordinate. Redis adds another service to manage, another failure mode, and another cost line.

You hit CPU time limits. A pipeline that runs 6 agents sequentially can take 30-60 seconds of LLM inference. Most serverless platforms have execution limits that force you into background job infrastructure.

You cannot track costs. When Agent A calls GPT-4 and Agent B calls Claude, the bill arrives as a single line item. You have no idea which agent consumed what, or whether the expensive model actually improved output quality.

You scale to infinity, not zero. When there is no work, your Python server still runs. When there is a burst, your single-threaded agent framework processes items sequentially.

The Workers-native approach solves all of these:

Key insight: The multi-agent pipeline is not a monolithic process. It is a series of small, focused computations with state checkpoints between them. This maps perfectly to Workers + Durable Objects + Workflows.

The system has four layers:

                    +----------------------------+
                    |     API / Cron Trigger      |
                    |  (Worker fetch handler)     |
                    +-----------+----------------+
                                |
                                v
                    +----------------------------+
                    |   OpportunityHarness DO     |
                    |  (Coordinator / Governor)   |
                    |  - Budget tracking          |
                    |  - Phase management          |
                    |  - Alarm-driven tick loop    |
                    +-----------+----------------+
                                |
              +-----------------+------------------+
              |                 |                   |
              v                 v                   v
    +--------------+  +--------------+  +--------------+
    | Gather Phase |  | Analyze Phase|  | Evaluate     |
    | data-coverage|  | genre-scanner|  | evaluator    |
    | (deterministic) | niche-analyzer| | critic       |
    +--------------+  | (LLM)       |  | lens-applicator
                      +--------------+  | (LLM)       |
                                        +--------------+
                                                |
                                                v
                                     +-------------------+
                                     | Debate System      |
                                     | 6 agent roles      |
                                     | 3-phase protocol   |
                                     +-------------------+
                                                |
                                                v
                                     +-------------------+
                                     | Agent Memory       |
                                     | Core beliefs       |
                                     | Episodic memory    |
                                     | Reflective updates |
                                     +-------------------+

Each processor is a pure function: it receives a context (database handle, AI binding, budget state, model selector) and returns an array of results. The harness calls processors one at a time, records telemetry, and advances through phases until the cycle completes.

The key primitives from Cloudflare:

• Durable Objects: The harness coordinator. One DO per pipeline instance. Persistent state, alarm-driven execution, single-threaded consistency.
• D1: Agent memory, processing ledger, evaluation results, debate messages. SQLite at the edge.
• Workers AI: Free tier for lightweight models. 10,000 Neurons/day at no cost.
• Workflows: Durable execution for batch operations. Automatic retries, step-level checkpointing.
• AI Gateway: Optional caching and logging for external LLM calls.

The Agent Processor

Every agent in the pipeline is a processor: a pure function registered by name, called by the harness when its phase is active.

// The processor signature — every agent implements this
export type ProcessorFn = (ctx: ProcessorContext) => Promise<ProcessorResult[]>;

export interface ProcessorContext {
  db: D1Database;
  ai: Ai;                          // Workers AI binding
  apiKey: string;                   // External LLM API key (e.g., Gemini)
  config: HarnessConfig;           // Full pipeline configuration
  processor: ProcessorConfig;       // This processor's specific config
  cycle: CycleState;               // Current cycle state (budget, counters, work queues)
  selectModel: () => string;       // Budget-aware model selector
  remainingBudget: () => number;   // USD remaining for this processor
}

export interface ProcessorResult {
  processor: string;               // 'evaluator', 'critic', etc.
  entity_type: string;             // What was processed
  entity_id: string;               // Unique identifier
  model_used: string | null;       // null for deterministic processors
  tokens_input: number;
  tokens_output: number;
  cost_usd: number;
  duration_ms: number;
  success: boolean;
  error_message?: string;
  result_summary?: string;
  spawned_entities?: Array<{ type: string; id: string }>;
}

The processor registry is a simple Map:

const PROCESSOR_REGISTRY = new Map<string, ProcessorFn>();

export function registerProcessor(name: string, fn: ProcessorFn) {
  PROCESSOR_REGISTRY.set(name, fn);
}

export function getProcessor(name: string): ProcessorFn | undefined {
  return PROCESSOR_REGISTRY.get(name);
}

Processors self-register at module load time. When a file like evaluator.ts is imported, its registerProcessor('evaluator', async (ctx) => { ... }) call adds it to the registry. The harness looks up processors by name from the config.

Key insight: Processors are pure functions, not classes. They do not hold state. All state lives in D1 or the cycle object. This means any processor can be tested in isolation by constructing a ProcessorContext with a test database.

The Harness Config

The pipeline is configured entirely through a database record. This means a dashboard can change processor weights, model tiers, budget limits, and batch sizes while the pipeline is running — the harness reloads config from DB at the start of every tick.

export interface HarnessConfig {
  id: number;
  slug: string;
  name: string;
  version: number;
  status: string;                    // 'active' | 'paused' | 'error'

  // Budget
  budget_mode: 'soft' | 'hard';     // soft = skip expensive, hard = stop entirely
  budget_max_per_cycle: number;      // USD cap per processing cycle
  budget_lambda: number;             // decay factor for adaptive budgeting

  // Models (tiered by cost)
  model_default: string;             // e.g., 'gemini-2.0-flash'
  model_expensive: string;           // e.g., 'gemini-2.5-flash'
  model_premium: string;             // e.g., 'gemini-2.5-pro'
  model_upgrade_threshold: number;   // score above which to upgrade model

  // Cycle control
  cycle_interval_ms: number;         // time between cycles
  cycle_max_iterations: number;
  cycle_genres_limit: number;
  cycle_niches_per_genre: number;
  cycle_critique_threshold: number;  // minimum score to trigger critique

  // Processors (ordered)
  processors: ProcessorConfig[];

  // Focus (optional filtering)
  focus_genres: string[] | null;
  focus_niches: string[] | null;

  // Memory
  memory_enabled: boolean;
  memory_reflect_every_n: number;    // reflect after every N debates
}

export interface ProcessorConfig {
  name: string;                      // 'data-coverage', 'evaluator', etc.
  type: 'deterministic' | 'llm';
  enabled: boolean;
  phase: string;                     // 'gather' | 'analyze' | 'evaluate' | 'critique'

  // LLM-specific
  model_tier?: 'default' | 'expensive' | 'premium';
  max_tokens?: number;
  temperature?: number;
  prompt_version?: number;

  // Budget allocation
  weight: number;                    // share of budget (0.0-1.0)
  max_cost_per_call: number;         // hard cap per invocation

  // Execution
  batch_size: number;                // entities per tick
}

A typical config stores its processors in JSON:

{
  "slug": "market-scanner",
  "budget_mode": "soft",
  "budget_max_per_cycle": 0.50,
  "model_default": "gemini-2.0-flash",
  "model_expensive": "gemini-2.5-flash",
  "model_premium": "gemini-2.5-pro",
  "processors": [
    {
      "name": "data-coverage",
      "type": "deterministic",
      "phase": "gather",
      "enabled": true,
      "weight": 0,
      "max_cost_per_call": 0,
      "batch_size": 10
    },
    {
      "name": "genre-scanner",
      "type": "deterministic",
      "phase": "gather",
      "enabled": true,
      "weight": 0,
      "max_cost_per_call": 0,
      "batch_size": 1
    },
    {
      "name": "niche-analyzer",
      "type": "llm",
      "phase": "analyze",
      "enabled": true,
      "model_tier": "default",
      "weight": 0.3,
      "max_cost_per_call": 0.02,
      "batch_size": 3,
      "temperature": 0.3
    },
    {
      "name": "evaluator",
      "type": "llm",
      "phase": "evaluate",
      "enabled": true,
      "model_tier": "default",
      "weight": 0.3,
      "max_cost_per_call": 0.05,
      "batch_size": 2,
      "temperature": 0.2
    },
    {
      "name": "critic",
      "type": "llm",
      "phase": "critique",
      "enabled": true,
      "model_tier": "expensive",
      "weight": 0.25,
      "max_cost_per_call": 0.05,
      "batch_size": 2,
      "temperature": 0.4
    },
    {
      "name": "lens-applicator",
      "type": "llm",
      "phase": "evaluate",
      "enabled": true,
      "model_tier": "default",
      "weight": 0.15,
      "max_cost_per_call": 0.03,
      "batch_size": 2,
      "temperature": 0.3
    }
  ]
}

Key insight: The entire pipeline topology — which agents run, in what order, with what models, at what cost — is a JSON document in D1. No code changes required to reconfigure. A dashboard writes to the DB, and the next tick picks it up.

The Processing Pipeline

The pipeline moves through four phases in strict order:

gather → analyze → evaluate → critique

Each phase has zero or more processors. The harness advances to the next phase only when all processors in the current phase have no more work to do.

const PHASE_ORDER = ['gather', 'analyze', 'evaluate', 'critique'];

Within a phase, processors are tried sequentially. When a processor produces results, the tick ends (one processor per tick to stay under CPU limits). When no processor in the current phase produces work, the harness advances.

This creates a pull-based data flow:

1. Gather: data-coverage checks which data sources are ready. genre-scanner identifies candidate niches from ready data. Both are deterministic — zero LLM cost.
2. Analyze: niche-analyzer takes the niche queue populated by gather and produces deep analysis using LLM. Spawns demographic sub-niches for further analysis.
3. Evaluate: evaluator takes analyzed niches and produces full evaluations with deterministic tests (demand, supply, monetization, fragmentation) plus LLM tests (buildable, marketable, confidence). lens-applicator enriches evaluations with viral scoring and cross-pollination ideas.
4. Critique: critic stress-tests high-scoring evaluations, challenges assumptions, and adjusts scores.

// CycleState tracks the work queues that flow between phases
export interface CycleState {
  cycle_id: number | null;
  cycle_number: number;
  phase: string;
  phase_index: number;
  cost_so_far: number;
  tokens_input: number;
  tokens_output: number;

  // Work queues populated by earlier phases
  genres_to_scan: string[];
  niches_to_analyze: Array<{
    genre: string;
    niche_id: string;
    keywords: string[];
  }>;
  evaluations_to_critique: string[];

  // Counters
  genres_scanned: number;
  niches_found: number;
  evaluations_created: number;
  evaluations_critiqued: number;
  research_tasks_created: number;

  // Per-processor telemetry
  processor_stats: Record<
    string,
    { cost: number; calls: number; total_ms: number; failures: number }
  >;
}

The Agent Ledger

The agent framework tracks every processing event in an agent_ledger table. This enables incremental processing: agents only process entities they have not seen, or entities whose upstream data has changed.

export interface AgentConfig {
  name: string;
  version: number;
  stages?: { name: string; version: number }[];
}

// An entity needs processing when:
//   1. It has never been processed by this agent
//   2. It was processed at an older agent version
//   3. It was processed at an older stage version
//   4. Its input hash has changed (upstream data changed)
export async function getUnprocessed(
  db: D1Database,
  agent: AgentConfig,
  entityType: string,
  candidateIds: string[],
  stage = 'main',
  inputHashes?: Map<string, string>,
): Promise<string[]> {
  if (candidateIds.length === 0) return [];

  const stageVersion =
    agent.stages?.find((s) => s.name === stage)?.version ?? agent.version;

  const existing = new Map<
    string,
    { agent_version: number; stage_version: number; input_hash: string | null }
  >();

  // Query in chunks of 80 (D1 variable limit)
  for (let i = 0; i < candidateIds.length; i += 80) {
    const chunk = candidateIds.slice(i, i + 80);
    const placeholders = chunk.map(() => '?').join(',');
    const rows = await db
      .prepare(
        `SELECT entity_id, agent_version, stage_version, input_hash
         FROM agent_ledger
         WHERE agent = ? AND entity_type = ? AND stage = ?
           AND entity_id IN (${placeholders})`,
      )
      .bind(agent.name, entityType, stage, ...chunk)
      .all();

    for (const row of rows.results) {
      existing.set(row.entity_id as string, {
        agent_version: row.agent_version as number,
        stage_version: row.stage_version as number,
        input_hash: row.input_hash as string | null,
      });
    }
  }

  return candidateIds.filter((id) => {
    const entry = existing.get(id);
    if (!entry) return true;                               // never processed
    if (entry.agent_version < agent.version) return true;  // agent upgraded
    if (entry.stage_version < stageVersion) return true;   // stage upgraded
    if (inputHashes) {
      const currentHash = inputHashes.get(id);
      if (currentHash && entry.input_hash !== currentHash) return true;
    }
    return false;
  });
}

Version bumping is the escape hatch. When you improve an agent’s prompt or logic, bump its version number. Every entity it previously processed gets flagged for reprocessing automatically. No migration scripts, no manual resets.

// Simple hash for change detection
export function hashInputs(
  ...values: (string | number | null | undefined)[]
): string {
  const str = values.map((v) => String(v ?? '')).join('|');
  let hash = 0;
  for (let i = 0; i < str.length; i++) {
    const char = str.charCodeAt(i);
    hash = ((hash << 5) - hash) + char;
    hash |= 0;
  }
  return hash.toString(36);
}

The D1 schema for the ledger:

CREATE TABLE agent_ledger (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  agent TEXT NOT NULL,
  agent_version INTEGER NOT NULL,
  entity_type TEXT NOT NULL,
  entity_id TEXT NOT NULL,
  stage TEXT NOT NULL DEFAULT 'main',
  stage_version INTEGER NOT NULL,
  input_hash TEXT,
  result_json TEXT,
  duration_ms INTEGER,
  computed_at TEXT NOT NULL,
  UNIQUE(agent, entity_type, entity_id, stage)
);

CREATE INDEX idx_ledger_agent ON agent_ledger(agent, entity_type, stage);

Key insight: The agent ledger is the foundation of incremental processing. It answers “what has changed since last time?” without scanning every record. Version bumps + input hashes give you full control over what gets reprocessed.

The OpportunityHarness is a Durable Object that serves as the control plane for the entire pipeline. It does not do analysis — it governs execution.

The harness is responsible for:

1. Budget gating — can we afford the next step?
2. Phase selection — what processor runs next?
3. Entity selection — what data to process?
4. Model routing — cheap model or expensive model?
5. Executing the processor — calling the registered function
6. Recording telemetry — cost, tokens, success/failure
7. Adapting — adjusting batch sizes based on success rate

The DO Shell

import { DurableObject } from 'cloudflare:workers';

type Env = {
  DB: D1Database;
  AI: Ai;
  GEMINI_API_KEY: string;
  OPPORTUNITY_HARNESS: DurableObjectNamespace;
};

export class OpportunityHarness extends DurableObject<Env> {
  private state: HarnessState = DEFAULT_STATE;
  private config: HarnessConfig | null = null;

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    ctx.blockConcurrencyWhile(async () => {
      const saved = await ctx.storage.get<HarnessState>('state');
      if (saved) this.state = { ...DEFAULT_STATE, ...saved };
      // Resume alarm if the DO was evicted while running
      if (this.state.running) {
        const existing = await ctx.storage.getAlarm();
        if (!existing) {
          await ctx.storage.setAlarm(Date.now() + 5000);
        }
      }
    });
  }
}

The blockConcurrencyWhile call in the constructor is critical. It prevents any fetch or alarm from executing until the state is fully loaded. Without it, you get race conditions on startup.

The Alarm-Driven Main Loop

The harness uses DO alarms to drive execution. Each alarm tick runs one unit of work, then schedules the next tick. This avoids the Workers CPU time limit by breaking long pipelines into many small steps.

override async alarm() {
  if (!this.state.running) return;

  try {
    // Reload config from DB every tick -- dashboard can change it live
    this.config = await this.loadConfig();
    if (!this.config) {
      this.state.error_message = 'Harness config not found in DB';
      this.state.running = false;
      await this.save();
      return;
    }

    if (this.config.status === 'paused') {
      // Config says paused -- keep alarm alive but do not work
      await this.ctx.storage.setAlarm(Date.now() + 30_000);
      return;
    }

    // Start new cycle if needed
    if (!this.state.current_cycle) {
      await this.startCycle();
    }

    // Run one tick of work
    const done = await this.tick();

    if (done) {
      await this.finalizeCycle();
      await this.ctx.storage.setAlarm(
        Date.now() + this.config.cycle_interval_ms,
      );
    } else {
      // More work to do -- tick again in 2 seconds
      await this.ctx.storage.setAlarm(Date.now() + 2000);
    }

    await this.save();
  } catch (err: any) {
    this.state.error_message = err.message || 'Unknown error';
    await this.save();
    // Retry in 60s on error
    await this.ctx.storage.setAlarm(Date.now() + 60_000);
  }
}

The 2-second interval between ticks is deliberate. It keeps the harness responsive (you can pause it by changing DB config within 2 seconds) while avoiding unnecessary polling. For idle periods between cycles, the interval stretches to cycle_interval_ms (typically 1-24 hours).

HTTP Control API

The harness exposes a simple HTTP API for control operations:

override async fetch(request: Request): Promise<Response> {
  const url = new URL(request.url);
  const path = url.pathname;

  if (request.method === 'GET' && path === '/status') {
    return Response.json({
      running: this.state.running,
      harness_slug: this.state.harness_slug,
      cycle_count: this.state.cycle_count,
      last_cycle_at: this.state.last_cycle_at,
      error: this.state.error_message,
      current_cycle: this.state.current_cycle
        ? {
            phase: this.state.current_cycle.phase,
            cost_so_far: this.state.current_cycle.cost_so_far,
            genres_scanned: this.state.current_cycle.genres_scanned,
            niches_found: this.state.current_cycle.niches_found,
            evaluations_created: this.state.current_cycle.evaluations_created,
            processor_stats: this.state.current_cycle.processor_stats,
          }
        : null,
    });
  }

  if (request.method === 'POST' && path === '/start') {
    const body = await request.json<{ slug: string }>();
    this.state.running = true;
    this.state.harness_slug = body.slug;
    this.state.error_message = null;
    await this.save();
    await this.ctx.storage.setAlarm(Date.now() + 1000);
    return Response.json({ started: true, slug: body.slug });
  }

  if (request.method === 'POST' && path === '/stop') {
    this.state.running = false;
    await this.save();
    return Response.json({ stopped: true });
  }

  if (request.method === 'POST' && path === '/tick') {
    // Manual tick for testing
    await this.alarm();
    return Response.json({ ok: true, state: this.state });
  }

  return new Response('Not found', { status: 404 });
}

The Worker routes requests to the DO using idFromName:

// In your Worker's fetch handler
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname.startsWith('/harness/')) {
      const slug = url.pathname.split('/')[2]; // /harness/{slug}/status
      const id = env.OPPORTUNITY_HARNESS.idFromName(slug);
      const stub = env.OPPORTUNITY_HARNESS.get(id);
      const harnessPath = '/' + url.pathname.split('/').slice(3).join('/');
      return stub.fetch(new Request(url.origin + harnessPath, request));
    }

    return new Response('Not found', { status: 404 });
  },
};

The Tick: One Unit of Work

Each tick executes exactly one processor’s batch, then returns. This keeps CPU usage predictable and allows the harness to check budget between every operation.

private async tick(): Promise<boolean> {
  if (!this.config || !this.state.current_cycle) return true;
  const cycle = this.state.current_cycle;
  const config = this.config;

  // Budget gate (hard mode stops everything)
  if (
    config.budget_mode === 'hard' &&
    cycle.cost_so_far >= config.budget_max_per_cycle
  ) {
    if (cycle.cycle_id) {
      await this.env.DB.prepare(
        `UPDATE harness_cycles SET status = 'budget_exceeded' WHERE id = ?`,
      )
        .bind(cycle.cycle_id)
        .run();
    }
    return true; // done -- out of budget
  }

  // Get enabled processors for current phase
  const phaseProcessors = config.processors.filter(
    (p) => p.enabled && p.phase === cycle.phase,
  );

  if (phaseProcessors.length === 0) {
    return this.advancePhase();
  }

  let didWork = false;
  for (const proc of phaseProcessors) {
    const fn = getProcessor(proc.name);
    if (!fn) continue;

    // Soft budget: skip processors that exhausted their allocation
    if (proc.type === 'llm' && config.budget_mode === 'soft') {
      const allocated = config.budget_max_per_cycle * proc.weight;
      const spent = cycle.processor_stats[proc.name]?.cost || 0;
      if (spent >= allocated) continue;
    }

    const ctx: ProcessorContext = {
      db: this.env.DB,
      ai: this.env.AI,
      apiKey: this.env.GEMINI_API_KEY || '',
      config,
      processor: proc,
      cycle,
      selectModel: () => this.selectModel(proc, cycle),
      remainingBudget: () => {
        const allocated = config.budget_max_per_cycle * proc.weight;
        const spent = cycle.processor_stats[proc.name]?.cost || 0;
        return Math.max(0, allocated - spent);
      },
    };

    try {
      const results = await fn(ctx);
      for (const r of results) {
        this.recordResult(cycle, r);
      }
      didWork = results.length > 0;
      if (didWork) break; // one processor per tick
    } catch (err: any) {
      this.recordResult(cycle, {
        processor: proc.name,
        entity_type: 'error',
        entity_id: 'tick',
        model_used: null,
        tokens_input: 0,
        tokens_output: 0,
        cost_usd: 0,
        duration_ms: 0,
        success: false,
        error_message: err.message,
      });
    }
  }

  if (!didWork) {
    return this.advancePhase();
  }

  return false; // more work to do
}

private advancePhase(): boolean {
  if (!this.state.current_cycle) return true;
  const currentIdx = PHASE_ORDER.indexOf(this.state.current_cycle.phase);
  if (currentIdx < 0 || currentIdx >= PHASE_ORDER.length - 1) {
    return true; // last phase done
  }
  this.state.current_cycle.phase = PHASE_ORDER[currentIdx + 1];
  this.state.current_cycle.phase_index = 0;
  return false;
}

Key insight: The “one processor per tick” rule is the secret to staying under Workers CPU limits. Each tick runs one batch of one processor (typically 2-3 LLM calls), which takes 1-5 seconds. The alarm schedules the next tick. From the outside, it looks like a continuous process. From the inside, it is a sequence of independent, retryable steps.

The pipeline’s power comes from how processors in different phases feed data to each other. This is not configured with explicit wiring — it flows through the shared CycleState.

Phase 1: Gather (Deterministic)

The gather phase runs zero-cost processors that assess data readiness and identify candidates for analysis.

DataCoverage checks whether genres have sufficient data for meaningful analysis:

registerProcessor(
  'data-coverage',
  async (ctx: ProcessorContext): Promise<ProcessorResult[]> => {
    const { db, cycle } = ctx;
    const results: ProcessorResult[] = [];

    // One-shot: only run once per cycle
    if (cycle.processor_stats['data-coverage']?.calls > 0) return [];

    for (const genre of cycle.genres_to_scan) {
      const stats = await db
        .prepare(
          `SELECT
           COUNT(*) as total,
           SUM(CASE WHEN enrichment_source = 'scraper' THEN 1 ELSE 0 END) as enriched,
           SUM(CASE WHEN competitiveness_score IS NOT NULL THEN 1 ELSE 0 END) as scored
         FROM app_registry
         WHERE primary_genre = ?`,
        )
        .bind(genre)
        .first<any>();

      const enriched = stats?.enriched || 0;
      const enrichedPct =
        stats?.total > 0 ? Math.round((enriched / stats.total) * 100) : 0;
      const scored = stats?.scored || 0;
      const scoredPct =
        enriched > 0 ? Math.round((scored / enriched) * 100) : 0;

      // Ready threshold: 200+ enriched apps OR 30%+ enrichment rate
      const ready =
        (enriched >= 200 || enrichedPct >= 30) &&
        (scored >= 50 || scoredPct >= 20);

      if (!ready) {
        // Remove unready genres from the scan queue
        cycle.genres_to_scan = cycle.genres_to_scan.filter((g) => g !== genre);

        // Create a research task for the gap
        await db
          .prepare(
            `INSERT OR IGNORE INTO research_tasks
             (type, target_id, target_name, reason, priority, status)
             VALUES ('app_enrich', ?, ?, ?, 'high', 'pending')`,
          )
          .bind(
            genre,
            `${genre} enrichment`,
            `Genre has only ${enrichedPct}% enrichment coverage`,
          )
          .run();
        cycle.research_tasks_created++;
      }

      results.push({
        processor: 'data-coverage',
        entity_type: 'genre',
        entity_id: genre,
        model_used: null,
        tokens_input: 0,
        tokens_output: 0,
        cost_usd: 0,
        duration_ms: 0,
        success: true,
        result_summary: `${genre}: ${enrichedPct}% enriched, ready=${ready}`,
      });
    }

    return results;
  },
);

GenreScanner takes ready genres and finds candidate niches by clustering keywords:

registerProcessor(
  'genre-scanner',
  async (ctx: ProcessorContext): Promise<ProcessorResult[]> => {
    const { db, cycle } = ctx;

    // Process one genre per tick (heavy queries)
    const alreadyScanned = cycle.processor_stats['genre-scanner']?.calls || 0;
    const genre = cycle.genres_to_scan[alreadyScanned];
    if (!genre) return [];

    // Find vulnerable apps (stale + proven demand)
    const vulnerable = await db
      .prepare(
        `SELECT track_id, track_name, rating_count, competitiveness_score
         FROM app_registry
         WHERE primary_genre = ?
           AND enrichment_source = 'scraper'
           AND rating_count >= 500
           AND days_since_update >= 180
         ORDER BY competitiveness_score DESC
         LIMIT 30`,
      )
      .bind(genre)
      .all();

    // Find keyword gaps (keywords where incumbents are beatable)
    const keywordGaps = await db
      .prepare(
        `SELECT cr.keyword,
                COUNT(DISTINCT cr.track_id) as app_count,
                ROUND(AVG(ar.competitiveness_score), 1) as avg_comp,
                SUM(ar.rating_count) as total_ratings
         FROM crawl_rankings cr
         JOIN app_registry ar ON cr.track_id = ar.track_id
         WHERE ar.primary_genre = ?
           AND ar.competitiveness_score IS NOT NULL
         GROUP BY cr.keyword
         HAVING COUNT(DISTINCT cr.track_id) >= 3
           AND AVG(ar.competitiveness_score) >= 30
         ORDER BY avg_comp DESC
         LIMIT 30`,
      )
      .bind(genre)
      .all();

    // Cluster keywords into niches using word overlap
    const niches = clusterKeywords(keywordGaps.results as KeywordGap[]);

    // Feed niches into the analysis queue
    for (const niche of niches) {
      cycle.niches_to_analyze.push({
        genre,
        niche_id: `${genre}::${niche.label}`,
        keywords: niche.keywords,
      });
    }

    return [
      {
        processor: 'genre-scanner',
        entity_type: 'genre',
        entity_id: genre,
        model_used: null,
        tokens_input: 0,
        tokens_output: 0,
        cost_usd: 0,
        duration_ms: 0,
        success: true,
        result_summary: `${genre}: ${niches.length} niches, ${vulnerable.results.length} vulnerable apps`,
      },
    ];
  },
);

The keyword clustering is deterministic — no LLM needed:

function clusterKeywords(gaps: KeywordGap[]): NicheCluster[] {
  const clusters: NicheCluster[] = [];
  const used = new Set<string>();
  const sorted = [...gaps].sort((a, b) => b.total_ratings - a.total_ratings);

  for (const kw of sorted) {
    if (used.has(kw.keyword)) continue;

    const words = new Set(
      kw.keyword
        .toLowerCase()
        .split(/\s+/)
        .filter((w) => w.length > 2),
    );
    const cluster: KeywordGap[] = [kw];
    used.add(kw.keyword);

    for (const other of sorted) {
      if (used.has(other.keyword)) continue;
      const otherWords = other.keyword
        .toLowerCase()
        .split(/\s+/)
        .filter((w) => w.length > 2);
      const overlap = otherWords.filter((w) => words.has(w));
      if (overlap.length >= 1) {
        cluster.push(other);
        used.add(other.keyword);
        otherWords.forEach((w) => words.add(w));
      }
    }

    clusters.push({
      label: cluster[0].keyword,
      keywords: cluster.map((k) => k.keyword),
      total_ratings: cluster.reduce((s, k) => s + k.total_ratings, 0),
      avg_competitiveness:
        cluster.reduce((s, k) => s + k.avg_comp, 0) / cluster.length,
    });
  }

  return clusters.sort((a, b) => b.total_ratings - a.total_ratings).slice(0, 15);
}

Phase 2: Analyze (LLM)

The niche-analyzer takes each niche from the queue, gathers fresh data, and uses an LLM to interpret it:

registerProcessor(
  'niche-analyzer',
  async (ctx: ProcessorContext): Promise<ProcessorResult[]> => {
    const { db, cycle, processor, selectModel, remainingBudget } = ctx;
    const results: ProcessorResult[] = [];

    const batchSize = Math.min(
      processor.batch_size || 3,
      cycle.niches_to_analyze.length,
    );

    for (let i = 0; i < batchSize; i++) {
      const niche = cycle.niches_to_analyze.shift();
      if (!niche) break;

      // Budget gate: stop if we cannot afford another call
      if (remainingBudget() < (processor.max_cost_per_call || 0.05)) {
        cycle.niches_to_analyze.unshift(niche);
        break;
      }

      // Cap recursion depth
      const depth = (niche.niche_id.match(/::/g) || []).length;
      if (depth >= 3) continue;

      const start = Date.now();

      // Gather apps and reviews for this niche
      const apps = await fetchNicheApps(db, niche.keywords);
      const reviews = await fetchLowRatedReviews(db, apps.slice(0, 5));

      // Build prompt with real data
      const dataBlock = buildNicheDataBlock(niche, apps, reviews);

      const model = selectModel();
      const result = await llmObject(
        {
          apiKey: ctx.apiKey,
          model,
          temperature: 0.3,
          maxTokens: 4096,
          thinkingBudget: 0,
          db: ctx.db,
          contextType: 'harness',
          contextId: ctx.config.slug,
        },
        nicheAnalysisSchema,
        dataBlock,
        NICHE_ANALYZER_SYSTEM,
        `harness-niche-${niche.niche_id}`,
      );

      // Spawn sub-niches for demographic opportunities
      const spawned = result.data.demographic_opportunities.map((d) => ({
        type: 'niche' as const,
        id: `${niche.niche_id}::${d.segment}`,
      }));

      results.push({
        processor: 'niche-analyzer',
        entity_type: 'niche',
        entity_id: niche.niche_id,
        model_used: model,
        tokens_input: result.usage.inputTokens,
        tokens_output: result.usage.outputTokens,
        cost_usd: result.costUsd,
        duration_ms: Date.now() - start,
        success: true,
        spawned_entities: spawned,
      });
    }

    return results;
  },
);

The Zod schema enforces structured output from the LLM:

const nicheAnalysisSchema = z.object({
  user_need: z.string().describe('The core user problem this niche serves'),
  weakness_pattern: z
    .string()
    .describe('The common failure pattern across incumbents'),
  entry_vector: z
    .string()
    .describe('The strongest way to enter this market'),
  monetization_model: z
    .string()
    .describe('How to make money (subscription, IAP, paid upfront)'),
  price_range: z
    .string()
    .describe('Recommended price range based on competitor data'),
  buildability: z.enum(['trivial', 'moderate', 'complex', 'very_complex']),
  build_weeks: z.number().describe('Estimated weeks to MVP'),
  viral_score: z.number().min(0).max(5),
  viral_signals: z.array(z.string()),
  demographic_opportunities: z.array(
    z.object({
      segment: z.string(),
      axis: z.string(),
      reasoning: z.string(),
    }),
  ),
  risks: z.array(z.string()),
  confidence: z.enum(['high', 'medium', 'low']),
});

Phase 3: Evaluate (Hybrid)

The evaluator combines deterministic tests (computed from data) with LLM judgment:

// Tests 1-4: deterministic, computed from database queries
const totalRatings = appData.reduce(
  (s, a) => s + (a.rating_count || 0),
  0,
);
const staleApps = appData.filter(
  (a) => a.days_since_update && a.days_since_update >= 180,
);

// Test 1: Demand (10K+ combined ratings)
const demandStatus =
  totalRatings >= 10000
    ? 'pass'
    : totalRatings >= 5000
      ? 'partial'
      : 'fail';

// Test 2: Supply weakness (3+ stale apps)
const supplyStatus =
  staleApps.length >= 3 || avgCompScore > 40
    ? 'pass'
    : staleApps.length >= 1 || avgCompScore > 30
      ? 'partial'
      : 'fail';

// Tests 5-7: LLM judgment
const llmResult = await llmObject(
  config,
  evaluationSchema,
  `${dataBlock}\n\nYOUR JOB: Evaluate tests 5, 6, 7`,
  EVALUATOR_SYSTEM,
);

// Combine into a composite score
const tests = [
  demandStatus,
  supplyStatus,
  monetizationStatus,
  fragmentationStatus,
  llmResult.data.buildable_status,
  llmResult.data.marketable_status,
  llmResult.data.confidence_status,
];
const testsPassed = tests.filter((t) => t === 'pass').length;

The evaluation schema demands specificity:

const evaluationSchema = z.object({
  buildable_status: z.enum(['pass', 'fail', 'partial', 'unknown']),
  buildable_reasoning: z.string(),
  marketable_status: z.enum(['pass', 'fail', 'partial', 'unknown']),
  marketable_reasoning: z.string(),
  confidence_status: z.enum(['pass', 'fail', 'partial', 'unknown']),
  confidence_reasoning: z.string(),

  demand_summary: z.string(),
  supply_summary: z.string(),
  monetization_summary: z.string(),
  marketing_summary: z.string(),
  differentiation_angle: z.string(),

  failure_reasons: z.array(
    z.object({
      test: z.string(),
      reason: z.string(),
      solvability: z.enum(['easy', 'moderate', 'hard', 'impossible']),
      how_to_solve: z.string(),
    }),
  ),

  overall_score: z.number().min(0).max(100),
  confidence_level: z.enum(['high', 'medium', 'low']),
  status: z.enum(['promising', 'solvable', 'blocked', 'monitor', 'dead']),
});

Phase 4: Critique (Adversarial LLM)

The critic receives high-scoring evaluations and stress-tests them:

registerProcessor(
  'critic',
  async (ctx: ProcessorContext): Promise<ProcessorResult[]> => {
    const { db, cycle, processor, selectModel, remainingBudget } = ctx;
    const results: ProcessorResult[] = [];

    const batchSize = Math.min(
      processor.batch_size || 2,
      cycle.evaluations_to_critique.length,
    );

    for (let i = 0; i < batchSize; i++) {
      const slug = cycle.evaluations_to_critique.shift();
      if (!slug) break;
      if (remainingBudget() < (processor.max_cost_per_call || 0.05)) {
        cycle.evaluations_to_critique.unshift(slug);
        break;
      }

      // Load the evaluation to critique
      const eval_ = await db
        .prepare(`SELECT * FROM opportunity_evaluations WHERE slug = ?`)
        .bind(slug)
        .first<any>();

      // Load competitor data and reviews
      const refApps = await loadReferenceApps(db, eval_.id);
      const reviews = await loadCompetitorReviews(db, refApps);

      const model = selectModel();
      const llmResult = await llmObject(
        {
          apiKey: ctx.apiKey,
          model,
          temperature: 0.4, // higher for adversarial thinking
          maxTokens: 4096,
          db: ctx.db,
          contextType: 'harness',
          contextId: ctx.config.slug,
        },
        critiqueSchema,
        buildCritiqueDataBlock(eval_, refApps, reviews),
        CRITIC_SYSTEM,
      );

      // Apply score adjustment
      const adjustedScore = Math.max(
        0,
        Math.min(100, eval_.overall_score + llmResult.data.score_adjustment),
      );

      await db
        .prepare(
          `UPDATE opportunity_evaluations SET
           overall_score = ?,
           confidence = ?,
           critique_json = ?,
           critique_conviction = ?,
           critique_verdict = ?,
           critique_score_delta = ?
         WHERE slug = ?`,
        )
        .bind(
          adjustedScore,
          llmResult.data.conviction,
          JSON.stringify(llmResult.data),
          llmResult.data.conviction,
          llmResult.data.one_line_verdict,
          llmResult.data.score_adjustment,
          slug,
        )
        .run();

      cycle.evaluations_critiqued++;
    }

    return results;
  },
);

The critique schema forces structured adversarial thinking:

const critiqueSchema = z.object({
  challenges: z.array(
    z.object({
      claim: z.string().describe('The assumption being challenged'),
      counter: z.string().describe('Why this might be wrong'),
      severity: z.enum(['minor', 'moderate', 'critical']),
      verdict: z.enum(['holds', 'weakened', 'broken']),
    }),
  ),

  hidden_risks: z.array(
    z.object({
      risk: z.string(),
      likelihood: z.enum(['low', 'medium', 'high']),
      impact: z.enum(['low', 'medium', 'high']),
      mitigation: z.string(),
    }),
  ),

  market_timing: z.enum(['too_early', 'good', 'late', 'uncertain']),
  platform_risk: z.enum(['low', 'medium', 'high']),
  platform_risk_detail: z.string(),

  score_adjustment: z
    .number()
    .min(-30)
    .max(10)
    .describe('Negative = overrated, positive = underrated'),
  adjustment_reasoning: z.string(),

  conviction: z.enum(['high', 'medium', 'low']),
  one_line_verdict: z.string(),
});

Key insight: The evaluator and critic are adversarial by design. The evaluator tries to find opportunity. The critic tries to kill it. The score adjustment (clamped to -30/+10) ensures the critic can downgrade significantly but cannot inflate. This asymmetry prevents the system from producing false positives.

Beyond sequential evaluation, the system supports multi-agent debates where 6 specialized agents argue about opportunities from different perspectives.

Agent Roles

const AGENT_ROLES = {
  blue_ocean: 'Blue Ocean Scout',     // Finds uncontested market spaces
  vulnerability: 'Vulnerability Analyst', // Identifies competitor weaknesses
  revenue: 'Revenue Strategist',       // Evaluates monetization potential
  timing: 'Timing Analyst',           // Assesses market timing windows
  contrarian: 'Contrarian',           // Argues against the consensus
  synthesizer: 'Synthesizer',         // Integrates all perspectives
} as const;

type AgentRole = keyof typeof AGENT_ROLES;

Three-Phase Debate Protocol

The debate runs in three phases, each producing structured output that feeds the next:

Phase 1: Independent Analysis. Each agent evaluates the opportunity independently, without seeing other agents’ opinions. This prevents anchoring bias.

interface Phase1Output {
  role: string;
  score: number;            // 0-10
  confidence: number;       // 0-100%
  top_opportunities: Array<{
    cluster_name: string;
    thesis: string;
    score: number;
    evidence: string[];
    risk: string;
  }>;
  market_observations: string[];
  warnings: string[];
  anticipated_disagreements: Array<{
    from_agent: string;
    their_likely_argument: string;
    my_preemptive_response: string;
  }>;
}

The anticipated_disagreements field is powerful. It forces each agent to consider what others will say before they say it. The Blue Ocean Scout might anticipate: “Revenue Strategist will argue the market is too small. My preemptive response: the adjacent health market adds $2B TAM.”

Phase 2: Debate and Critique. Each agent sees all Phase 1 outputs and responds with challenges, agreements, revised positions, and new insights.

interface Phase2Output {
  role: string;
  revised_score: number;
  revised_confidence: number;

  challenges: Array<{
    target_agent: string;
    their_claim: string;
    steel_man: string;          // best version of their argument
    my_challenge: string;       // why I still disagree
    what_would_change_my_mind: string;
  }>;

  agreements: Array<{
    with_agent: string;
    on_what: string;
    combined_insight: string;   // what we learned together
  }>;

  revised_opportunities: Array<{
    cluster_name: string;
    revised_thesis: string;
    revised_score: number;
    changed_because: string;
  }>;

  new_insights: string[];
  position_changes: string[];
}

The steel_man field is important. Before challenging another agent, the challenger must first present the strongest version of the argument they disagree with. This prevents straw-man attacks and forces genuine engagement with opposing views.

Phase 3: Synthesis. The Synthesizer agent reads all Phase 1 and Phase 2 outputs and produces the final report.

interface DebateReport {
  ranked_opportunities: Array<{
    rank: number;
    cluster_name: string;
    composite_score: number;
    app_concept_name: string;
    app_concept_tagline: string;
    go_no_go: string;
    first_move: string;
    key_evidence: string[];
    key_risks: string[];
    resolved_disagreements: string[];
  }>;

  market_narrative: string;
  biggest_surprise: string;
  meta_risk: string;

  debate_quality: {
    genuine_disagreements: number;
    position_changes: number;
    novel_insights: number;
    consensus_areas: string[];
    unresolved_tensions: string[];
  };
}

Debate Chat: Making Debates Readable

Raw debate JSON is impenetrable. The debate-chat module transforms structured debate data into a Slack-like timeline of messages:

export async function extractDebateMessages(
  db: D1Database,
  debateId: number,
  report: DebateReport,
): Promise<void> {
  const stmt = db.prepare(
    `INSERT INTO debate_messages
     (debate_id, phase, agent_role, message_type, content,
      target_agent, metadata_json, sort_order)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
  );

  const batch: D1PreparedStatement[] = [];
  let order = 0;

  // Phase 1: independent analyses become opening statements
  for (const [role, p1] of Object.entries(report.agent_evaluations.phase1)) {
    const name = AGENT_NAMES[role] || role;
    const parts: string[] = [];
    parts.push(`**${name}** -- Score: ${p1.score}/10`);

    if (p1.top_opportunities?.length) {
      parts.push('\n**My top opportunities:**');
      for (const opp of p1.top_opportunities) {
        parts.push(`- **${opp.cluster_name}** (${opp.score}/10): ${opp.thesis}`);
      }
    }

    batch.push(
      stmt.bind(
        debateId, 1, role, 'analysis',
        parts.join('\n'), null, null, order++,
      ),
    );
  }

  // Phase 2: challenges and agreements become threaded replies
  for (const [role, p2] of Object.entries(report.agent_evaluations.phase2)) {
    if (p2.challenges?.length) {
      for (const ch of p2.challenges) {
        const msg = [
          `**Challenging ${AGENT_NAMES[ch.target_agent]}:**`,
          `> Their claim: "${ch.their_claim}"`,
          `\n**Steel-man:** ${ch.steel_man}`,
          `\n**But I disagree:** ${ch.my_challenge}`,
          `\n*I'd change my mind if:* ${ch.what_would_change_my_mind}`,
        ].join('\n');
        batch.push(
          stmt.bind(
            debateId, 2, role, 'challenge',
            msg, ch.target_agent, null, order++,
          ),
        );
      }
    }

    if (p2.position_changes?.length) {
      for (const pc of p2.position_changes) {
        batch.push(
          stmt.bind(
            debateId, 2, role, 'position_change',
            `**I changed my mind:** ${pc}`, null, null, order++,
          ),
        );
      }
    }
  }

  // Phase 3: synthesis becomes the verdict
  for (const opp of report.ranked_opportunities) {
    batch.push(
      stmt.bind(
        debateId, 3, 'synthesizer', 'verdict',
        `**#${opp.rank}: ${opp.cluster_name}** -- Score: ${opp.composite_score}/10\n**Verdict:** ${opp.go_no_go}`,
        null,
        JSON.stringify({ rank: opp.rank, score: opp.composite_score }),
        order++,
      ),
    );
  }

  // Write in batches of 100 (D1 batch limit)
  for (let i = 0; i < batch.length; i += 100) {
    await db.batch(batch.slice(i, i + 100));
  }
}

The debate messages table:

CREATE TABLE debate_messages (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  debate_id INTEGER NOT NULL,
  phase INTEGER NOT NULL,        -- 1, 2, or 3
  agent_role TEXT NOT NULL,
  message_type TEXT NOT NULL,    -- analysis, challenge, agreement, position_change, verdict
  content TEXT NOT NULL,
  target_agent TEXT,             -- who they're responding to
  metadata_json TEXT,
  sort_order INTEGER NOT NULL,
  created_at TEXT DEFAULT (datetime('now'))
);

Key insight: The debate system is not a free-form conversation. It is a structured protocol with fixed phases and typed outputs. This makes it reliable (no infinite loops), auditable (every position change is recorded), and cost-predictable (exactly 6 + 6 + 1 = 13 LLM calls per debate).

Convergence detection answers the question: “Are the signals in our data getting stronger or weaker over time?”

In a market analysis context, convergence means two previously unrelated clusters are becoming similar — their shared keywords are growing, their app overlap is increasing. This is an opportunity signal: users are searching across categories, and no app serves the intersection.

export interface ConvergenceSignal {
  lineage_a: number;
  lineage_b: number;
  name_a: string;
  name_b: string;
  genre_a: string;
  genre_b: string;
  current_similarity: number;
  previous_similarity: number;
  delta: number;                  // positive = growing closer
  shared_keywords: string[];
  cross_genre: boolean;           // different genres = unexpected = higher signal
  first_detected_at: string | null;
  growth_rate: number;
  signal_strength: number;        // composite score
  insight: string;                // human-readable explanation
}

The detector compares current state against historical state:

export async function detectConvergence(db: D1Database): Promise<{
  signals: ConvergenceSignal[];
  stats: {
    pairs_analyzed: number;
    converging: number;
    diverging: number;
    stable: number;
  };
}> {
  // Get all current cluster relations
  const currentRels = await db
    .prepare(
      `SELECT cr.lineage_id_a, cr.lineage_id_b, cr.similarity, cr.app_overlap,
              cla.name as name_a, clb.name as name_b
       FROM cluster_relations cr
       JOIN cluster_lineages cla ON cr.lineage_id_a = cla.id
       JOIN cluster_lineages clb ON cr.lineage_id_b = clb.id
       WHERE cla.is_active = 1 AND clb.is_active = 1`,
    )
    .all();

  const signals: ConvergenceSignal[] = [];
  let converging = 0, diverging = 0, stable = 0;

  for (const rel of currentRels.results) {
    const currentSim = rel.similarity as number;

    // Get previous similarity for this pair
    const prevRel = await db
      .prepare(
        `SELECT similarity FROM cluster_relations
         WHERE lineage_id_a = ? AND lineage_id_b = ?
         ORDER BY detected_at DESC
         LIMIT 1 OFFSET 1`,
      )
      .bind(rel.lineage_id_a, rel.lineage_id_b)
      .first();

    const previousSim = (prevRel?.similarity as number) || 0;
    const delta = currentSim - previousSim;

    if (delta > 0.01) converging++;
    else if (delta < -0.01) diverging++;
    else stable++;

    // Signal strength = speed x unexpectedness x market size
    const crossGenre = genreA !== genreB;
    const unexpectedness = crossGenre ? 2.0 : 1.0;
    const marketSize = Math.log10(ratingsA + ratingsB);
    const signalStrength = Math.round(
      (delta > 0 ? delta * 100 : 0) * unexpectedness * marketSize,
    );

    // Generate human-readable insight
    let insight = '';
    if (delta > 0.05 && crossGenre) {
      insight = `Rapid cross-genre convergence: ${genreA} x ${genreB} merging around "${shared.slice(0, 3).join(', ')}". New app bridging both could capture untapped demand.`;
    } else if (delta > 0.05) {
      insight = `Same-genre clusters converging. Market subcategories increasingly share keywords. Consolidation opportunity.`;
    } else if (delta > 0 && crossGenre) {
      insight = `Slow cross-genre drift. Early signal -- monitor for acceleration.`;
    }

    signals.push({
      lineage_a: la,
      lineage_b: lb,
      name_a: rel.name_a as string,
      name_b: rel.name_b as string,
      genre_a: genreA,
      genre_b: genreB,
      current_similarity: currentSim,
      previous_similarity: previousSim,
      delta,
      shared_keywords: shared.slice(0, 10),
      cross_genre: crossGenre,
      first_detected_at: firstSeen,
      growth_rate: delta,
      signal_strength: signalStrength,
      insight,
    });
  }

  signals.sort((a, b) => b.signal_strength - a.signal_strength);

  return {
    signals,
    stats: {
      pairs_analyzed: currentRels.results.length,
      converging,
      diverging,
      stable,
    },
  };
}