Pipeline

The pipeline is a PostgreSQL-backed job queue and law status tracking system that orchestrates the law processing workflow.

Overview

• Language: Rust
• Location: packages/pipeline/
• Database: PostgreSQL
• Key feature: Reliable concurrent job processing with FOR UPDATE SKIP LOCKED

Architecture

The pipeline coordinates two processing stages: harvesting (downloading laws from wetten.nl) and enrichment (adding machine-readable logic via LLM).

Modules

Binaries

Run the four tools from packages/ with cargo run -p regelrecht-pipeline --bin <name> -- <args>. The usage of each is in the doc comment at the top of its source file.

Job Lifecycle

Workers claim jobs atomically using PostgreSQL’s FOR UPDATE SKIP LOCKED, so multiple workers can safely process jobs concurrently without blocking each other.

Automatic Retries

When a job fails and has attempts remaining (attempts < max_attempts), it returns to Pending for retry. Default max_attempts is 3.

Orphan Reaping

Jobs stuck in Processing beyond the orphan timeout (default: 30 minutes) are reset to Pending or marked Failed, which is how a crashed worker is handled.

Law Status Tracking

Each law in the corpus progresses through processing states:

NotHarvestable is the terminal one. A work can have no consolidated text to harvest because it was withdrawn, is not yet in force, or has only been announced. The skip reason is uniform, so the status is a single value and the precise reason and date go into the harvest job’s result. The job is completed rather than failed, so it is never retried; a future law can be re-harvested by hand once its text appears.

Pipeline API

Besides the workers, the crate builds one HTTP service, pipeline-api (src/bin/pipeline_api.rs, image regelrecht-pipeline-api). It has no public address and no authentication of its own. The editor API forwards /api/harvest/* to it (see PIPELINE_API_URL on Editor API) and does the auth checks in front of it.

Harvest Worker

The harvest worker:

1. Polls the queue for pending harvest jobs
2. Downloads law XML from BWB (wetten.nl)
3. Converts XML to YAML via the harvester library
4. Writes YAML to the corpus
5. Creates enrich jobs for each LLM provider, but only when ENRICH_AUTO_ENQUEUE is on (it is off by default) and the law is not enrich_exhausted
6. Creates follow-up harvest jobs for the laws the harvested text references

Two mechanisms pull in further laws, each with its own limit:

• References in the text. Step 6 follows the external references the harvester finds in the XML. A follow-up job carries its parent’s depth plus one, and the chain stops at MAX_HARVEST_DEPTH (1000, in harvest.rs). That constant guards against runaway recursion and is not a tuning knob; the chain normally ends because every referenced law is already harvested or queued.
• Related legislation from enrichment. After an enrich job, the related legislation the agent reports (delegated regelingen, cross-law sources, legal bases the harvester misses) is harvested only while the enrich job’s depth is below RELATED_HARVEST_MAX_DEPTH (default 2). This is the limit that bounds how far one enrichment can grow the corpus.

Enrich Worker

The enrich worker:

1. Polls the queue for pending enrich jobs, within the hourly budget (ENRICH_HOURLY_LIMIT)
2. Spawns an LLM CLI process to generate machine_readable sections
3. Tracks progress via .enrichment-progress.json (polled every 10s)
4. Computes coverage score (the stored law_entries.coverage_score is the cumulative fraction of articles with machine_readable; the per-run delta rides in the job result)
5. Pushes the result to a per-provider branch (enrich/ followed by the provider name, opencode or claude) for review

The hourly budget fails closed. Without ENRICH_HOURLY_LIMIT, or with 0, the worker enriches nothing, so a forgotten variable cannot spend a subscription on the whole corpus. The cap counts runs for the worker’s own LLM_PROVIDER per clock hour (Europe/Amsterdam), and because it is counted in the jobs table it survives a restart.

Large laws and agent sessions

A large law does not fit in one agent session. The worker splits it into windows of at most ENRICH_MAX_ARTICLES_PER_RUN articles (default 15) and owns the cursor itself: it lives in .enrichment.yaml on the enrich/{provider} branch, each chunk pushes its own result, and the next chunk is queued in the same transaction that completes the current one. A law of N articles is done in at most ceil(N / 15) successful runs, whatever the model does. Task-flow enrichments (deliver: task) always take the whole law.

Within one window, ENRICH_SESSION_REUSE decides how the translation pass and the feedback rounds of the gates share an agent session: all of them (window, the default), only the schema repair (repair), or none (off). It applies to the claude provider and never crosses a window. Token use and cost of every call land in the job result (agent_calls, usage), which is how the modes are compared; enrich-once --session-reuse prints the same table locally.

The reasoning behind both, including why a window follows document order rather than “the next articles without machine_readable”, is in the doc comments on plan_chunk and SessionReuse in enrich.rs.

LLM Providers

LLM_PROVIDER selects opencode (the default) or claude. The binary and model come from OPENCODE_PATH / OPENCODE_MODEL or CLAUDE_PATH / CLAUDE_MODEL, with LLM_PATH and LLM_MODEL as the fallback for either.

The LLM subprocess runs with a stripped environment (allowlisted vars only) for security.

Configuration

Database and workers

The corpus checkout is configured by the corpus library (packages/corpus/src/config.rs). Without CORPUS_REPO_URL the workers run without one.

Enrichment

LLM_TIMEOUT_SECS is a ceiling per agent call, and one run makes several: a translation pass, a feedback round per gate, the closing pass and the final schema gate. The worker lowers it when the job budget cannot hold that many, so raising LLM_TIMEOUT_SECS without raising WORKER_JOB_TIMEOUT_SECS buys nothing.

Database Schema

The pipeline crate owns the migrations (packages/pipeline/migrations/) for the whole platform database, so the editor API’s tables (accounts, trajects, notes, settings, feature flags) are created here too. The two the pipeline itself runs on:

jobs - Job queue with retry tracking, priority ordering, and JSONB payload/result/progress columns. Partial index WHERE status = 'pending' for efficient claiming.

law_entries - Per-law status tracking with foreign keys to harvest/enrich jobs and a coverage score (0.0–1.0).

Next to those, tasks and job_blobs carry review tasks and their payloads, and markings and untranslatables hold what enrichment reported.

Migrations run automatically at startup using an advisory lock for coordination.

Testing

Integration tests use testcontainers to spin up ephemeral PostgreSQL instances; no local database setup is required.

Further reading

• Harvester - the BWB law downloader used by harvest jobs
• Architecture - where the pipeline fits in the system