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

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.

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. Auto-creates enrich jobs for each configured LLM provider
6. Creates follow-up harvest jobs for referenced laws (respects depth limit of 1000)

Enrich Worker

The enrich worker:

1. Polls the queue for pending enrich jobs
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. Creates per-provider branches (e.g., enrich/opencode)

Chunked enrichment of large laws

One LLM session cannot enrich a large law (hundreds of articles) within the session/RSS limits. With ENRICH_MAX_ARTICLES_PER_RUN = N (default 15, 0 disables chunking) each enrich run processes at most N articles, in document order, from a worker-owned cursor:

• The cursor (enrich_cursor + enrich_cursor_path) persists in the .enrichment.yaml on the enrich/{provider} branch. It only applies when recorded for the same YAML path and within bounds; otherwise it resets to 0 (covers new law versions and legacy metadata).
• Each successful chunk commits and pushes its own result, so a failing later chunk never loses earlier chunks.
• While the law is not finished (law_complete = false), its status stays enriching and a continuation job is created in the same database transaction as the job completion (respecting the unique active-enrich-job index), so there is never a law in enriching without an active/pending job.
• MvT research runs only in the first chunk (cursor 0); reverse validation is limited to the articles of the chunk. A chunk may legitimately add zero machine_readable sections when the agent records a chunk_report in .enrichment-result.yaml that references at least one article of the chunk’s window; a chunk without any output (or with an empty/unrelated report) fails retryable (never terminal).
• Termination is guaranteed in ceil(articles_total / N) successful runs, independent of LLM behavior; the last chunk marks the law enriched.
• Task-flow enrichments (deliver=task) always run whole-law (chunking off).

One session per window

A window is one law and one article range: the translation pass plus the feedback rounds of the three gates. Every one of those calls used to be a cold CLI process that read the law, the context brief, the skills and the schema again, up to seven starts per window. ENRICH_SESSION_REUSE decides whether they share one agent session instead. It applies to the claude provider only; the worker picks the session id itself and passes --session-id on the first call and --resume after that.

• window (default): every call in the window continues the same session. Each resumed feedback prompt opens with an instruction to read the file from disk before answering. A gate is meant to be a fresh look at what stands there, and an agent that remembers writing it can otherwise defend its own choice instead of reading the finding.
• repair: the translation pass and the schema gate share a session, the checks and marking gates run cold. A schema error is a fact about the file; those two gates ask for judgement.
• off: every call its own cold process, the behaviour before this existed.

The session never crosses a window: a continuation chunk opens its own. An agent that kept everything it wrote would carry half a large law into the last chunk, which costs more than starting over.

Whether reuse is cheaper has a number behind it. A resumed round pays every turn over the context the translation pass ended at, so it wins only if knowing the law already shortens the round. Every call is therefore accounted (agent_calls in the job result: step, whether it was resumed, input/output/ cache-read tokens and cost), with the window total beside it (usage) and the mode that produced them (session_reuse). enrich-once --session-reuse runs the same loop locally and prints the table.

LLM Providers

The LLM provider is configurable via LLM_PROVIDER (default: opencode). Provider-specific paths and models are set via environment variables (e.g., OPENCODE_PATH, OPENCODE_MODEL).

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

Configuration

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

Two tables with PostgreSQL enums:

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).

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