Stage-Per-Item

A pipeline framework for processing thousands of entities through multi-stage research workflows — with versioning, staleness detection, and automatic backfill.

1. The Problem

You're building a system that monitors 18,000 investment theses scraped from a members-only club. For each thesis, you need to: fetch the HTML, extract structured fields (ticker, thesis type, quality score), run an LLM to validate the extraction, compute returns at 5 horizons, and generate embeddings for ML prediction. That's 6 stages × 18,000 items = 108,000 stage-executions.

Halfway through, you improve your extraction prompt. Now 12,000 items have stale extractions. You don't want to re-fetch the HTML (it hasn't changed and the source is rate-limited). You want to re-run just the extraction and its downstream stages on the cached data.

A week later, you add a 7th stage: embed the thesis text for ML training. You need it to run on all 18,000 items — but only on items that already have a valid extraction. And you want the system to figure that out automatically.

This is not a scheduling problem (Airflow). It's not an asset materialization problem (Dagster). It's not a durable function problem (Temporal). It's a stage-per-item tracking problem, and no mainstream framework provides it natively.

2. Core Idea

Think of it like a factory assembly line, but where each product moves at its own pace, any station can be upgraded independently, and upgrading a station automatically flags products that passed through the old version for re-inspection.

The key properties that distinguish this from existing approaches:

3. Architecture

System Overview

Data Model

Three tables capture all pipeline state:

Plus an events table for the audit trail (circuit breaker trips, backfill pauses, error classifications).

Handler Pattern

A handler is a Python module exporting one function: process_stage(). It dispatches on stage name via match/case:

async def process_stage(*, item_key, data, job, stage) -> dict | None:
    match stage:
        case "fetch":     return await _fetch(item_key, data, job)
        case "extract":   return _extract(item_key, data, job)
        case "enrich":    return await _enrich(item_key, data, job)

HANDLER_VERSION = {"fetch": "1", "extract": "2026-02-27.1", "enrich": "2"}

VERSION_DEPS = {
    "enrich": [_ENRICH_PROMPT, _ENRICH_SYSTEM, str(ENRICH_LLM)],
}

The runner calls process_stage() once per entity per stage, stores the return dict in the results table with the current handler_version, and advances item_stages.

Version Tracking

Two complementary mechanisms detect when results become stale:

The dashboard compares each result's stored handler_version against the current version. Mismatches surface as a stale count. One click resets stale items to pending for that stage — re-running on cached upstream data.

Stage Execution: Event-Driven, Not Polling

Each stage has a dedicated async worker. When stage N finishes an item, it fires stage N+1's asyncio.Event immediately (<1ms). Workers sleep on the event when idle — zero DB queries, zero CPU. A 30-second safety poll prevents stalls.

This matters for pipelines where the first stage is fast (fetch cached HTML) but the third is slow (LLM call). Items flow through without waiting for the entire batch.

Error Classification

Every exception passes through an LLM classifier (~$0.001 per call) that categorizes it and selects an action:

The critical insight: an item-specific parse error should never pause the entire pipeline. Only systemic issues (affecting all items) warrant a pause. Traditional circuit breakers can't distinguish these — they just count consecutive failures.

4. Motivating Examples

resources:
  groq_transcribe:
    concurrency: 3    # 3 concurrent across ALL pipelines

guards:
  daily_cost_limit: 5.00    # Pause if $5 spent today
  max_pending: 200          # Don't queue more than 200

5. Prior Work Comparison

Existing orchestration frameworks solve adjacent problems well. The gap is specific: per-entity-per-stage tracking with versioning and surgical reprocessing.

Where each framework excels

Airflow is the standard for scheduled ETL — daily batch jobs with complex inter-task dependencies. If your work is "run this DAG every morning at 6am," Airflow is battle-tested.

Dagster comes closest to the versioning story. Its code_version and "Unsynced" labels solve staleness at the asset level. For teams building analytics where the output is a table or model, Dagster's asset graph is excellent. The gap: it models assets, not entities within an asset. Tracking 18,000 items at different stages requires DynamicPartitionsDefinition with custom staleness logic on top.

Temporal is the strongest per-entity model — each entity can be a child workflow with durable state. For long-running processes (order fulfillment, subscription lifecycle), Temporal is unmatched. The tradeoff: you're running a distributed systems platform (Temporal server, persistence layer, worker fleet), and the stage pipeline isn't declarative — it's imperative code.

Prefect has the best developer ergonomics — pure Python, no DAG boilerplate. Its .map() fans out per-item with independent tracking. But there's no versioning, no staleness detection, and no persistent entity registry between runs.

The gap we fill

6. Measuring Usefulness

Direct metrics

Process metrics

Ablation: What happens without it

Before this framework existed, each pipeline was a standalone script with:

• Custom "already_processed" tracking (a set, a file, a separate DB table)
• No versioning — prompt changes silently produced mixed-version results
• No cross-pipeline rate limiting — concurrent scripts overwhelmed APIs
• No error classification — transient 429s killed entire runs
• No cost tracking — overnight runs occasionally spent $100+ before being noticed

Limitations

• SQLite ceiling: Single-writer constraint limits throughput to ~5,000 writes/sec. Not suitable for high-frequency trading or real-time event processing.
• Single-machine: No distributed execution. All workers run in one process. Fine for research workloads; insufficient for production services with SLA requirements.
• No UI for stage editing: Adding stages requires editing YAML and restarting. A web UI for pipeline design would improve accessibility.
• LLM error classification cost: At $0.001/error, a pipeline with 10% error rate processing 100K items costs $10 just for error classification. Batch classification or local models could reduce this.

7. Walkthrough: VIC Thesis Pipeline

The VIC (Value Investors Club) pipeline processes 18,000+ investment theses through 3 stages. Here's the end-to-end flow:

Configuration

vic_ideas:
  handler: vic_ideas
  kind: backfill
  stages:
    - name: fetch
      concurrency: 4
    - name: extract
      concurrency: 8
    - name: check_enrich
      concurrency: 4
      resource: groq_transcribe
  pacing:
    max_per_hour: 200
    retry_max_attempts: 3
  guards:
    daily_cost_limit: 10.00
  storage:
    base_dir: jobs/data/vic_ideas
    pattern: "{idea_id}"

Stage-by-stage flow for a single entity

What happens when you change the extraction prompt

Pipeline status view

8. Vision & Roadmap

Where it stands

The framework runs 12 pipelines in production, processing ~20,000 entities across investment research, earnings analysis, YouTube content monitoring, and entity intelligence. It's proven the core model — stage-per-item with versioning and cached reprocessing — across diverse workloads.

Independent library roadmap

Why it compounds

North star

A world where building a research pipeline is as simple as writing a function per stage and declaring the pipeline in YAML. The framework handles entity tracking, versioning, staleness, retries, cost control, and observability. You focus on what each stage does — the framework handles everything else.