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

A worker that holds a job records a heartbeat on it every 30 seconds (heartbeat_at), from the moment it claims the job until it is done. A job in Processing whose last heartbeat is older than the orphan timeout (default 5 minutes) is reset to Pending or marked Failed. That is how a crashed or restarted worker is handled.

How long a job has been running does not count: a worker that keeps beating keeps its job, and the job timeout (WORKER_JOB_TIMEOUT_SECS) decides when it gives up. The heartbeat itself stops 10 minutes after the job timeout, so a worker that is alive but stuck (a hung git push) loses its job to the reaper instead of holding it until the pod restarts. A database outage longer than the orphan timeout also costs the jobs in flight, because no heartbeat lands while the database is down. Before the heartbeat, the reaper measured from started_at, so a run still working past the orphan timeout was put back to Pending and its result refused when it finished.

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. Writes progress to the job every 10 seconds (see Progress and stalls)
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 applies to the claude provider, which runs on a personal subscription token, whatever the worker’s own LLM_PROVIDER is: a user picks the provider per enrichment in the editor. It counts claude runs per clock hour (Europe/Amsterdam), and because it is counted in the jobs table it survives a restart. Once the claude runs of the hour are used up, claude jobs wait for the next hour and the worker keeps running jobs for other providers. The opencode provider (VLAM) has no cap for now. The budget fails closed: without ENRICH_HOURLY_LIMIT, or with 0, no claude job runs, so a forgotten variable cannot spend the subscription on the whole corpus.

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) walk the same windows but push nothing. Each window is its own job with its own review tasks. The next job takes the current job’s result as its input blobs: the proposal, so the next window builds on what this one enriched, and .enrichment.yaml, which carries the cursor. A window’s review tasks cover the articles that window changed. The last window is wider: its closing reconcile pass and binding check run over the whole law, so it can also carry tasks for articles from earlier windows.

A new law (new_law, from an upload or a harvest into a traject) is created as a whole, so its windows produce no review task along the way. Only the last window does, with the complete law as one proposal. If a window fails for good, the requester gets a failure task and the earlier windows’ work is lost; the law has to be uploaded or harvested again.

When an existing law takes more than one window, a window whose proposal cannot be compared article by article fails and is retried, the last window included. It does not fall back to one task for the whole law, because that proposal also contains the earlier windows’ articles, and approving it would silently restore changes a reviewer had rejected.

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.

Progress and stalls

The claude provider runs with --output-format stream-json, so the worker reads each event as it happens: assistant turns, tool calls, a partial event per token, and the closing result with the token use. Every 10 seconds the worker writes jobs.progress with two parts:

• at the top level, the note the agent writes before each step (phase, step, total_steps), as before;
• under worker, what the worker saw: the current call (translate or a gate), how long it has run, how long ago the agent last wrote anything, the number of turns and tool calls, the last few tool calls (Edit laws/…/law.yaml), and how many of the law’s articles carry machine_readable so far. The tool summary names the tool and what it works on, never the text it writes.

The job detail sheet on the Harvester page shows this while a job runs, and keeps the last state on a job that stopped. That is the place to look when a run times out: the error itself also ends with a line on what the agent did last. When the claude CLI exits with an error, such as an API call that kept failing, the error also carries the reason the CLI gave in its closing result event. The CLI writes nothing to stderr in that case, so without it the job would only say exit status: 1.

A claude call that writes nothing for ENRICH_AGENT_IDLE_SECS (default 600) is stopped as stalled, and fails like a timeout, with the same retries. With partial events on, a long answer arrives as a stream of events, so silence means no token at all. The limit is generous because an API retry after an overload is silent for a while too. 0 turns the check off. The check leaves opencode alone, because it writes an event per finished part and can be quiet for longer, and it leaves the document and law conversions alone, because a converter run through Bash is silent while it works.

With ENRICH_WINDOW_CONCURRENCY above 1 the windows of a job share one view of the agent: the counts add up, and the stall check fires only once every window has gone quiet.

Definitions and units

The law-generate skill writes every constant in definitions in the structured form, with a type and, for a quantity, a type_spec.unit (RFC-023). Money is type: amount with unit: eurocent.

The deterministic check definition-type reports a definition that holds a number without a type, because nothing then says whether 0.137 is a share or a factor, or whether 3971900 is cents. Its findings go back to the agent as a feedback round.

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. LLM_PROVIDER is the worker’s default: a job that names a provider in its payload runs on that one, which is how the editor lets a user pick a model per enrichment.

The opencode provider talks to VLAM, the language model API of SSC-ICT, and the editor shows it under that name. Its model lives in packages/pipeline/opencode.json; the model is kimi-k3-prepaid. The only route to VLAM is the platform service vlam on ZAD, which puts VLAM_API_URL (a proxy into VLAM) in the pod. At startup the enrich worker writes opencode’s user configuration with $VLAM_API_URL/v1 as the endpoint. Without that service the provider has no endpoint, and the worker logs a warning. The key comes from SSC-ICT, not from ZAD, and is set as VLAM_API_KEY; the worker writes it to opencode’s auth.json, readable by the worker alone. It does this only where the image’s opencode.json exists, so a worker started on a developer’s machine leaves their home alone. enrich-once never does it and runs on the opencode configuration and credentials the developer already has.

Before the worker starts, the image’s entrypoint logs VLAM check lines: the model as VLAM lists it, the result of one short chat call, and whether the model answers a call that offers it a tool. Together they show on every deploy whether the key works and whether the model can do the tool calls an enrichment depends on.

The chat call is asked three times: plainly, with reasoning_effort: low, and with chat_template_kwargs: {thinking: false}. Each line gives the completion tokens and whether the reasoning came back in a field of its own, by length only. A server that ignores a setting answers as if it were absent, so two lines with the same token count do not prove the setting was rejected.

opencode sends the model’s output limit (limit.output in opencode.json, 32000 for Kimi K3) as max_tokens on every call, and the chat call in the check asks for the same limit, so a deploy also shows whether VLAM accepts it. 32000 is also opencode’s own ceiling: a higher limit needs OPENCODE_EXPERIMENTAL_OUTPUT_TOKEN_MAX as well as limit.output.

Kimi K3 reasons before it acts, and the reasoning counts against that limit. VLAM reports it as zero reasoning tokens all the same: answering OK costs some sixty completion tokens. A step that runs into the limit ends without finishing what it was writing, and opencode then stops as if the work were done. The worker warns on agent step cut off by the output limit, and the error of a run that produced nothing says how many steps were cut off.

Every opencode step is logged with why it ended, the tokens it used, and how many characters it wrote as text, as reasoning and as tool arguments (text_chars, reasoning_chars, tool_input_chars). Counting needs opencode’s --thinking, which the worker passes; without it opencode leaves reasoning out of its output. Characters are not tokens (count three to four per token), but the split says where a cut-off step’s output went:

• Large reasoning_chars: the model was still thinking.
• Large text_chars: it was writing prose, or reasoning inline. If the startup check says VLAM returns no separate reasoning field, reasoning arrives as text.
• A tool call the limit interrupted shows up as invalid with a large tool_input_chars on the cut-off step (opencode repairs the half-written call into that tool, with the partial arguments in its error), or as <tool> (error) with no arguments just after it.

What differs between the two CLIs lives in one dialect per provider, under packages/pipeline/src/agent/: the command line, what a line of output means, and whether a call can continue a session, stalls when it goes quiet, or names its model. Starting the process and watching it is the same for both. The tests play recorded output of each provider through it, from packages/pipeline/tests/fixtures/agent-streams/; the opencode recordings there are still assembled by hand (#1895).

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

The image pins the claude CLI (packages/pipeline/Dockerfile, version 2.1.303). The CLI is a native binary: the image installs the platform package @anthropic-ai/claude-code-linux-x64-musl and links its binary as claude, because the wrapper package @anthropic-ai/claude-code only puts the binary in place from a postinstall script, which the install skips.

The claude provider runs Opus 5.5 (claude-opus-5-5) unless CLAUDE_MODEL or LLM_MODEL names another model. Left to itself the CLI picks a model from its version and the subscription behind the token, and that choice moved from Opus to Sonnet without a change on our side. The stripped environment keeps the CLI’s own variables (ANTHROPIC_MODEL, CLAUDE_CODE_EFFORT_LEVEL) out. Without LLM_EFFORT the CLI picks the effort too; that depends on a server-side setting, and the CLI reports it nowhere. The CLI does report the model, in the init event at the start of the stream-json output. The worker records it: per call in the job result’s agent_calls, as models on the result, as model in .enrichment.yaml (which said default when no model was set), and in the progress the Harvester page shows while a job runs.

opencode writes neither an init event nor a closing total. For the model the worker records the one opencode was told to use. When neither OPENCODE_MODEL nor LLM_MODEL names one, the enrich worker takes it from opencode.json. The usage of a call is the sum of its steps’ step_finish events, with reasoning tokens counted as output. An opencode call never continues a session, so an opencode job runs with session_reuse: off whatever ENRICH_SESSION_REUSE says: every call is recorded as cold, and no feedback prompt tells it that it wrote the file earlier in the same conversation.

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.

The shares are not equal. The translation pass writes every machine_readable in its window and is capped at three shares of the job budget, less a 30 s reserve; each feedback round the run may make is capped at one, counted from ENRICH_FEEDBACK_ROUNDS. A lower LLM_TIMEOUT_SECS stays as it is. With WORKER_JOB_TIMEOUT_SECS=3900 and one round per gate the caps are 1290 s for the translation and 430 s per round, where an even split gave each call about 550 s. More rounds per gate make every share smaller.

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