add dockerfile, docker config, pipeline and container entrypoint

2026-09-09 22:14:22 +02:00 · 2026-09-09 22:14:22 +02:00 · 80ca38af0a
commit 80ca38af0a
parent 2ec749d94d
15 changed files with 1136 additions and 19 deletions
--- a/.dockerignore
+++ b/.dockerignore
@ -0,0 +1,12 @@
+# Allowlist the build context: credentials, data, Git history and tests stay out.
+**
+!Dockerfile
+!pyproject.toml
+!uv.lock
+!README.md
+!src/
+!src/car_traker/
+!src/car_traker/**
+src/**/__pycache__/
+src/**/*.pyc
+!searches.toml
--- a/30
+++ b/30
@ -0,0 +1,30 @@
+FROM ghcr.io/astral-sh/uv:0.9.3 AS uv
+FROM python:3.13-slim-bookworm
+
+COPY --from=uv /uv /usr/local/bin/uv
+ENV UV_PROJECT_ENVIRONMENT=/opt/venv \
+    UV_PYTHON_DOWNLOADS=never \
+    UV_LINK_MODE=copy \
+    PLAYWRIGHT_BROWSERS_PATH=/opt/playwright \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    HOME=/tmp \
+    PATH=/opt/venv/bin:$PATH
+WORKDIR /app
+
+COPY pyproject.toml uv.lock README.md ./
+RUN uv sync --locked --no-dev --no-install-project \
+    && python -m playwright install --with-deps chromium \
+    && apt-get update \
+    && apt-get install -y --no-install-recommends xvfb \
+    && rm -rf /var/lib/apt/lists/* /tmp/uv-cache /tmp/.cache \
+    && useradd --uid 1000 --no-create-home app \
+    && mkdir /data \
+    && chown app:app /data
+COPY src/ ./src/
+RUN uv sync --locked --no-dev --no-editable && rm -rf /tmp/.cache
+COPY searches.toml ./searches.toml
+
+USER 1000:1000
+ENTRYPOINT ["python", "-m", "car_traker.container_entrypoint"]
+CMD ["car-traker-pipeline", "--headed", "--config", "/app/searches.toml", "--output", "/data", "--family", "saab-900", "--max-pages", "1", "--max-details", "1"]
--- a/PROJECT_PLAN.md
+++ b/PROJECT_PLAN.md
@ -24,6 +24,13 @@ PostgreSQL containers are disposable tests only. Homelab provisioning and remote
 connection setup remain user actions documented in `docs/POSTGRESQL.md`; no homelab
 database has been accessed or modified during implementation.

+The manual homelab job is now packaged in `Dockerfile` and `ops/app/compose.yaml`.
+`car-traker-pipeline` performs a database preflight, bounded collection, and an
+idempotent import of the exact returned manifest, retaining partial coverage and
+separate operational outcomes. There is no scheduler or automatic deployment.
+See `docs/DEPLOYMENT.md`; container tests use isolated PostgreSQL 17 and synthetic
+browser responses. Actual live access from the homelab still needs verification.
+
 ## Purpose

 Build a French used-car asking-price observatory focused initially on youngtimers.
--- a/README.md
+++ b/README.md
@ -3,7 +3,8 @@
 First implementation: collect unnormalized listing observations from Leboncoin
 search pages and individual ads, including descriptions and source-provided locations.
 Finalized raw runs can now be cleaned and imported idempotently into PostgreSQL.
-Scheduling and deployment are not implemented yet.
+Manual Docker Compose execution is available; scheduling and automated deployment
+remain deferred.
 Asking prices are not sale prices; disappearing listings do not prove a sale.

 ## Run
@ -51,12 +52,16 @@ src/car_traker/       # Installed Python package
  __main__.py        # python -m car_traker entry point
  collect.py         # Collector and CLI
  database.py        # PostgreSQL migrations and import CLI
+  pipeline.py        # Manual collect -> raw -> import job
+  container_entrypoint.py # Virtual display and graceful shutdown
  transform.py       # Pure, conservative silver field cleaning
  migrations/        # Versioned SQL schema and analytical views
 tests/               # Offline tests
 docs/POSTGRESQL.md    # Homelab provisioning, import, queries, database tests
 docs/CI.md            # Gitea 1.23 CI and repository-scoped runner registration
 ops/ci-runner/        # Optional CI runner infrastructure, not app deployment
+ops/app/             # Manual homelab job and runtime configuration example
+docs/DEPLOYMENT.md    # Storage, credentials, network access, and recovery
 searches.toml        # Search configuration
 pyproject.toml       # Dependencies, build configuration, and CLI entry point
 data/                # Local snapshots (ignored by Git)
@ -85,6 +90,35 @@ separate timestamps; source coverage and data-quality issues remain explicit.
 SQL views expose typed listing history, latest observations, price history, coverage,
 and validation issues. Generation/trim inference, redaction, and scheduling are deferred.

+## Manual Homelab Job
+
+`car-traker-pipeline` checks database connectivity/schema, collects raw observations,
+and imports that exact finalized run. Partial/failed collections remain marked as
+such even when their import succeeds. A lock prevents overlapping pipeline jobs on
+the same output directory, and `pipeline.json` records the import outcome separately
+from the immutable raw manifest.
+
+For a laptop smoke run with the existing database connection configured:
+
+```sh
+uv run car-traker-pipeline --headed --family saab-900 --max-pages 1 --max-details 1
+```
+
+For Docker on the homelab, follow [manual deployment](docs/DEPLOYMENT.md) first to
+configure private raw storage, the password file, and container network access.
+Then, from the homelab checkout:
+
+```sh
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml build app
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app car-traker-db check
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app
+```
+
+The default image command is bounded to Saab 900, one search page and one detail
+request. It uses headed Chromium through a virtual display, not a desktop/VNC
+service. No PostgreSQL server, scheduler, or automatic migration is added.
+Live Leboncoin access from the homelab still needs a manual smoke test.
+
 ## Raw Format

 **Gzip-compressed JSON Lines (`.jsonl.gz`)**, with one observation per line and one
--- a/docs/DEPLOYMENT.md
+++ b/docs/DEPLOYMENT.md
@ -0,0 +1,234 @@
+# Manual Homelab Deployment
+
+## Scope
+The next milestone is one manual **collection -> raw files -> PostgreSQL import**
+job on the homelab. Development stays on the laptop; build and run from a repository
+checkout on the existing Debian VM, alongside Gitea, its runner, and PostgreSQL 17.
+Use the existing database and role, currently both `car_tracker`.
+Do not provision another production PostgreSQL server or recreate those objects.
+
+`ops/app/compose.yaml` is the one-shot application stack, separate from
+`ops/ci-runner`. There is no scheduling, restart loop, or automatic CI deployment.
+Run commands from the homelab checkout's root with Docker Engine and Compose
+available. Review the revision and `searches.toml`; do not run these in CI.
+After these changes are published, update the homelab checkout before building.
+
+## Private Runtime Configuration
+Use an external, nonsecret Compose env file, for example
+`/srv/car-tracker/app.env`. These absolute paths are examples, not secrets.
+Inspect existing files first: never overwrite an existing env file or `.pgpass`.
+Use `ops/app/.env.example` as the reference when creating a new file locally:
+
+```dotenv
+APP_UID=1000
+APP_GID=1000
+RAW_DATA_DIR=/srv/car-tracker/raw
+DB_CONFIG_DIR=/srv/car-tracker/db
+PGHOST=postgres.your-lan.example
+PGPORT=5432
+PGDATABASE=car_tracker
+PGUSER=car_tracker
+PGSSLMODE=require
+APP_SUBNET=172.30.17.0/28
+```
+
+Replace the hostname with the existing VM's reachable LAN/VPN hostname and choose
+the numeric `APP_UID`/`APP_GID` that will own the mounted files. Keep the application
+user non-root. Both bind directories must already exist; Compose refuses to create
+missing source directories. After inspecting `/srv` and any existing application
+directories, create only missing directories and apply permissions deliberately:
+
+```sh
+sudo install -d -m 0700 -o 1000 -g 1000 /srv/car-tracker/raw /srv/car-tracker/db
+```
+
+Use your chosen UID/GID instead of `1000` in all ownership commands. Both
+`RAW_DATA_DIR` and `DB_CONFIG_DIR` must have mode `0700` and the correct ownership.
+The raw directory must be writable by the container user; the DB directory must be
+traversable by that user even though its container mount is read-only.
+
+Create or carefully edit `/srv/car-tracker/db/pgpass` locally, using a private
+editor/session. Its matching entry has this form (do not use the placeholder):
+
+```text
+postgres.your-lan.example:5432:car_tracker:car_tracker:<password entered locally>
+```
+
+Escape `:` and `\` in the password according to libpq password-file rules.
+Do not put passwords in commands, the env file, Git, logs, or chat. Do not send
+tokens or credentials to anyone to complete this setup. Preserve existing password
+files and entries rather than replacing them blindly. Protect the dedicated file:
+
+```sh
+sudo chown 1000:1000 /srv/car-tracker/db/pgpass
+sudo chmod 0600 /srv/car-tracker/db/pgpass
+```
+
+The password is supplied only through `/run/db/pgpass`, mounted from
+`DB_CONFIG_DIR`. Incorrect ownership prevents reads; libpq ignores a password file
+with overly permissive permissions. Keep this directory dedicated to app DB files,
+not unrelated credentials.
+
+Compose interpolates the external env file but passes only the runtime whitelist:
+`PGHOST`, `PGPORT`, `PGDATABASE`, `PGUSER`, and `PGSSLMODE`, plus fixed
+`PGPASSFILE=/run/db/pgpass` and `PGSSLROOTCERT=/run/db/root.crt`.
+It does not pass through `DATABASE_URL`, `PGPASSWORD`, or arbitrary host variables.
+Shell variables can override Compose env-file interpolation; unset stale matching
+variables before using this configuration. Always specify `--env-file` explicitly.
+
+## TLS And Network Access
+`PGSSLMODE` defaults to `require`: require encryption, but do not treat this as
+server identity verification. Prefer `verify-full` with the trusted CA certificate
+at `/srv/car-tracker/db/root.crt`, readable by the application user. The existing
+directory mount exposes it as `/run/db/root.crt`; `PGHOST` must match the server
+certificate. Configure PostgreSQL server TLS before running. Never fix production
+connectivity by disabling TLS or adding `trust` authentication.
+
+`localhost` inside the container is not the VM's PostgreSQL server. Prefer the VM's
+LAN/VPN hostname. Compose also provides `host.docker.internal` through
+`extra_hosts: host-gateway`, but PostgreSQL must listen on that gateway address if
+you use it. A listener only on loopback or the LAN address does not necessarily
+accept gateway connections. Certificate hostname matching still applies.
+
+The app has a dedicated bridge subnet, default `172.30.17.0/28`. Check existing
+Docker, LAN, and VPN routes for conflicts before starting; configure `APP_SUBNET`
+to a non-overlapping range if needed. Coordinate any network changes with the
+existing runner and services. Do not expose PostgreSQL publicly.
+
+PostgreSQL may see a container client address different from the laptop's address.
+The laptop's working HBA rule is therefore not proof this job can connect. Inspect
+the actual client address in PostgreSQL logs and existing firewall/HBA rules. A new
+narrow rule may be needed for the default app subnet:
+
+```text
+hostssl  car_tracker  car_tracker  172.30.17.0/28  scram-sha-256
+```
+
+Use the configured subnet and actual observed addressing, not a blindly copied
+rule. HBA rules are first-match; preserve unrelated rules and permit only the
+necessary source/port through the firewall. Check `pg_hba_file_rules` for errors
+and reload authentication changes with `SELECT pg_reload_conf();` as an admin.
+Changing `listen_addresses` requires a PostgreSQL restart: coordinate downtime for
+the shared VM's clients rather than restarting casually. See
+[PostgreSQL troubleshooting](POSTGRESQL.md#connection-troubleshooting) for details;
+that document's `car_traker` names are examples, not this VM's `car_tracker` names.
+
+## Build And Smoke Run
+Validate without printing interpolated configuration, then build on the homelab:
+
+```sh
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml config --quiet
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml build app
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app car-traker-db check
+```
+
+`check` is read-only connectivity testing, not a schema or migration-privilege
+check. The pipeline additionally checks access to `silver.collection_runs` before
+opening the browser. No new migrations are required for this milestone. If the
+existing packaged migrations have not yet been applied, apply them explicitly:
+
+```sh
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app car-traker-db migrate
+```
+
+Migrations are not automatic. Once DB access/schema are ready, run the image's
+default bounded headed pipeline: Saab 900, one page per selected search, at most
+one unique detail request for the run, with the default five-second request delay.
+
+```sh
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app
+```
+
+Inspect the reported manifest and import outcome before expanding. After a usable
+smoke run, this is the explicit larger manual command:
+
+```sh
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app \
+  car-traker-pipeline --headed --config /app/searches.toml --output /data \
+  --family saab-900 --max-pages 3 --max-details 5
+```
+
+Limits can intentionally produce partial coverage. Do not loop or retry a blocked
+source automatically; stop and investigate access failures before another scrape.
+
+## Outcomes And Recovery
+The pipeline prints `Manifest: /data/<date>/<run-id>/manifest.json`, reports a JSON
+summary after successful import, and imports
+that exact finalized run, including partial or failed collection runs. Host files
+are under `/srv/car-tracker/raw/<date>/<run-id>/`. Import success does not imply
+complete source coverage. Keep finalized raw files and manifests immutable.
+
+| Pipeline exit | Meaning |
+| --- | --- |
+| 0 | Complete collection imported, without rejected records |
+| 1 | Incomplete collection imported, without rejected records |
+| 2 | Import or replay has rejected records; inspect validation issues |
+| 3 | Operational, configuration, locking, or database failure |
+| 130 | Cancellation/interruption |
+
+`pipeline.json` is a separate operational sidecar, not a rewrite of raw coverage.
+It records `collection_status`, `import_status`, and an import summary on success.
+An interrupted import or lost DB response can leave `import_status: unknown`:
+the transaction may already have committed. Recover by idempotently replaying the
+exact manifest with the DB CLI, not by rerunning the scrape. Substitute the exact
+date/run ID printed by the pipeline:
+
+```sh
+docker compose --env-file /srv/car-tracker/app.env -f ops/app/compose.yaml run --rm app \
+  car-traker-db import '/data/<date>/<run-id>/manifest.json'
+```
+
+An identical replay reports `already_imported` without duplicating observations.
+The standalone DB CLI does not update `pipeline.json`; retain its replay result
+separately. Its exit codes differ; see [import documentation](POSTGRESQL.md#migrate-and-import).
+Do not alter raw files to force a replay past integrity checks.
+
+`browser_cleanup.json` may record a nonfatal browser teardown error after raw
+finalization; the pipeline can still import that run. Allow graceful cancellation
+to finish. A hard kill, OOM, or expired stop grace can leave a `running` manifest,
+no sidecar, or stale `running` import state. A `running` manifest is not imported;
+investigate rather than marking it complete or assuming a DB rollback occurred.
+
+The output-root `.pipeline.lock` covers the whole manual pipeline job, including
+collection and import. Do not delete an active lock file; the file can remain after
+exit and its presence alone does not mean a job is active. Standalone collection
+and DB CLIs do not honor this lock. Never overlap them with a pipeline or other
+manual job using the same output root, including during recovery.
+
+## Runtime Boundaries
+
+Compose runs non-root with a read-only root filesystem, all capabilities dropped,
+and `no-new-privileges`. There is no Docker socket mount or published port.
+Runtime limits are 1.5 CPUs, 2 GiB memory, 256 PIDs, 512 MiB shared memory,
+512 MiB `/tmp` tmpfs, a 1 MiB X11 socket tmpfs, and a 30-second stop grace period. These do not constrain
+image builds. Keep load low on the VM shared with Gitea, its runner, and PostgreSQL.
+
+The image installs Chromium matching the locked Playwright version. Its custom
+Python entrypoint starts Xvfb for headed operation, forwards SIGTERM to the child,
+and waits for child cleanup before removing remaining processes and the display.
+No external GUI or noVNC service is needed.
+
+Chromium's sandbox remains disabled by Playwright's default launch behavior; an
+explicit sandbox trial failed under host security restrictions. These container
+settings are not a claim of secure browser isolation. Do not add privileged mode,
+`SYS_ADMIN`, or `seccomp=unconfined` to work around this. The browser container can
+read mounted DB credentials and shares the VM kernel. Prefer a separate VM for
+future browser isolation rather than weakening the shared host's protections.
+
+## Verification And Retention
+
+Live source access from this homelab has not been verified here. All 109 tests
+passed inside the built non-root image with its read-only filesystem and resource
+limits, using isolated PostgreSQL 17 and synthetic browser responses. An actual
+Docker stop also allowed the child to finish cleanup and exit 130 before the
+display was removed. These tests do not guarantee live source access.
+For optional testing, see [PostgreSQL tests](POSTGRESQL.md#tests) and
+`tests/test_browser_pipeline.py`. Browser tests require `RUN_BROWSER_TESTS=1` and
+`TEST_DATABASE_URL`; use only a disposable database explicitly ending in `_test`,
+never homelab production. Tests truncate data. Default CI skips browser tests.
+
+Retention is still manual: start with 30 days for raw files and private database
+copies, including descriptions and raw payloads, and account for relevant backups.
+Deleting raw files does not delete DB copies. No retention or backup automation is
+provided here; arrange and verify private backups manually before relying on the
+data. Keep raw files, DB copies, and backups private and outside Git/CI artifacts.
--- a/ops/app/.env.example
+++ b/ops/app/.env.example
@ -0,0 +1,17 @@
+# Nonsecret Compose settings. Use a separate runtime env file on the homelab.
+# Do not put passwords or DATABASE_URL in this file.
+APP_UID=1000
+APP_GID=1000
+RAW_DATA_DIR=/srv/car-tracker/raw
+DB_CONFIG_DIR=/srv/car-tracker/db
+
+# Use the existing VM's reachable PostgreSQL hostname/IP, not container localhost.
+PGHOST=postgres.your-lan.example
+PGPORT=5432
+PGDATABASE=car_tracker
+PGUSER=car_tracker
+
+# Encryption required. Prefer verify-full once root.crt and hostname are configured.
+PGSSLMODE=require
+# Check this subnet does not overlap existing Docker/LAN/VPN networks.
+APP_SUBNET=172.30.17.0/28
--- a/ops/app/compose.yaml
+++ b/ops/app/compose.yaml
@ -0,0 +1,58 @@
+name: car-traker-app
+
+services:
+  app:
+    build:
+      context: ../..
+      dockerfile: Dockerfile
+    image: car-traker:local
+    # One-shot manual job only. Never restart a blocked scraper automatically.
+    restart: "no"
+    init: true
+    stop_grace_period: 30s
+    user: "${APP_UID:-1000}:${APP_GID:-1000}"
+    read_only: true
+    cap_drop: [ALL]
+    security_opt:
+      - no-new-privileges:true
+    cpus: 1.5
+    mem_limit: 2g
+    pids_limit: 256
+    shm_size: 512m
+    tmpfs:
+      - /tmp:rw,nosuid,nodev,size=512m,mode=1777
+      - /tmp/.X11-unix:rw,nosuid,nodev,size=1m,mode=1777
+    environment:
+      PGHOST: "${PGHOST:?Set the existing PostgreSQL hostname or IP}"
+      PGPORT: "${PGPORT:-5432}"
+      PGDATABASE: "${PGDATABASE:?Set the application database name}"
+      PGUSER: "${PGUSER:?Set the application database role}"
+      PGSSLMODE: "${PGSSLMODE:-require}"
+      PGPASSFILE: /run/db/pgpass
+      PGSSLROOTCERT: /run/db/root.crt
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    volumes:
+      - type: bind
+        source: "${RAW_DATA_DIR:?Set an existing private raw storage directory}"
+        target: /data
+        bind:
+          create_host_path: false
+      - type: bind
+        source: "${DB_CONFIG_DIR:?Set an existing private directory containing pgpass}"
+        target: /run/db
+        read_only: true
+        bind:
+          create_host_path: false
+      - type: bind
+        source: ../../searches.toml
+        target: /app/searches.toml
+        read_only: true
+    networks:
+      app:
+
+networks:
+  app:
+    ipam:
+      config:
+        - subnet: "${APP_SUBNET:-172.30.17.0/28}"
--- a/pyproject.toml
+++ b/pyproject.toml
@ -18,6 +18,7 @@ select = ["E4", "E7", "E9", "F"]
 [project.scripts]
 car-traker = "car_traker.collect:main"
 car-traker-db = "car_traker.database:main"
+car-traker-pipeline = "car_traker.pipeline:main"

 [build-system]
 requires = ["setuptools>=77"]
--- a/src/car_traker/collect.py
+++ b/src/car_traker/collect.py
@ -4,6 +4,7 @@ import argparse
 import gzip
 import json
 import math
+import sys
 import time
 import tomllib
 import uuid
@ -257,8 +258,8 @@ def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, m
    return path, manifest["status"]


-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
+def parse_collection_args(argv=None, description=None):
+    parser = argparse.ArgumentParser(description=description or __doc__)
    parser.add_argument("--config", type=Path, default=Path("searches.toml"))
    parser.add_argument("--output", type=Path, default=Path("data/raw/leboncoin"))
    parser.add_argument("--family", help="Collect only this configured family")
@ -267,7 +268,7 @@ def main():
    parser.add_argument("--headed", action="store_true")
    parser.add_argument("--search-only", action="store_true", help="Skip individual listing pages")
    parser.add_argument("--max-details", type=int, help="Maximum unique detail requests per run")
-    args = parser.parse_args()
+    args = parser.parse_args(argv)
    if args.max_pages < 1 or not math.isfinite(args.delay) or args.delay < 1:
        parser.error("max-pages must be positive and delay must be finite and >= 1 second")
    if args.max_details is not None and args.max_details < 1:
@ -284,26 +285,50 @@ def main():
        searches = [s for s in searches if s["family"] == args.family]
    if not searches:
        parser.error("no matching searches")
+    args.searches = searches
+    return args

+
+def collect_browser(args):
    from playwright.sync_api import sync_playwright

-    with sync_playwright() as playwright:
-        browser = playwright.chromium.launch(headless=not args.headed)
+    result = None
+    try:
+        with sync_playwright() as playwright:
+            browser = playwright.chromium.launch(headless=not args.headed)
+            try:
+                context = browser.new_context(locale="fr-FR")
+                page = context.new_page()
+
+                def fetch(url):
+                    response = page.goto(url, wait_until="domcontentloaded", timeout=60_000)
+                    if response is None or response.status != 200:
+                        raise SourceAccessError(response.status if response else None)
+                    return page.locator("script#__NEXT_DATA__").text_content(timeout=15_000)
+
+                result = collect(
+                    fetch, args.searches, args.output, args.max_pages, args.delay,
+                    args.search_only, args.max_details,
+                )
+            finally:
+                browser.close()
+    except KeyboardInterrupt:
+        if result is not None:
+            print(f"Finalized manifest before interruption: {result[0]}", file=sys.stderr)
+        raise
+    except Exception as exc:
+        if result is None:
+            raise
+        # Browser teardown must not discard an already-finalized raw run.
+        print(f"Browser cleanup failed ({type(exc).__name__}); raw run is finalized.", file=sys.stderr)
        try:
-            context = browser.new_context(locale="fr-FR")
-            page = context.new_page()
+            write_json(result[0].parent / "browser_cleanup.json", {"error_type": type(exc).__name__})
+        except OSError:
+            print("Could not record browser cleanup failure.", file=sys.stderr)
+    return result

-            def fetch(url):
-                response = page.goto(url, wait_until="domcontentloaded", timeout=60_000)
-                if response is None or response.status != 200:
-                    raise SourceAccessError(response.status if response else None)
-                return page.locator("script#__NEXT_DATA__").text_content(timeout=15_000)

-            path, status = collect(
-                fetch, searches, args.output, args.max_pages, args.delay,
-                args.search_only, args.max_details,
-            )
-        finally:
-            browser.close()
+def main():
+    path, status = collect_browser(parse_collection_args())
    print(f"{status}: {path}")
    return 0 if status == "complete" else 1
--- a/src/car_traker/container_entrypoint.py
+++ b/src/car_traker/container_entrypoint.py
@ -0,0 +1,69 @@
+"""Keep the virtual display alive until the command finishes handling shutdown."""
+
+import os
+import select
+import signal
+import subprocess
+import sys
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("A container command is required.", file=sys.stderr)
+        return 3
+    display = child = None
+    stopped = None
+
+    def forward(signum, frame):
+        nonlocal stopped
+        stopped = signum
+        if child is not None and child.poll() is None:
+            child.send_signal(signum)
+
+    previous = {sig: signal.signal(sig, forward) for sig in (signal.SIGINT, signal.SIGTERM)}
+    reader, writer = os.pipe()
+    try:
+        display = subprocess.Popen([
+            "Xvfb", "-displayfd", str(writer), "-screen", "0", "1280x1024x24", "-nolisten", "tcp",
+        ], pass_fds=(writer,), start_new_session=True)
+        os.close(writer)
+        writer = None
+        ready, _, _ = select.select([reader], [], [], 10)
+        if stopped:
+            return 128 + stopped
+        number = os.read(reader, 32).decode("ascii").strip() if ready else ""
+        if not number.isdigit():
+            print("Virtual display did not start.", file=sys.stderr)
+            return 3
+        child = subprocess.Popen(sys.argv[1:], env={**os.environ, "DISPLAY": f":{number}"},
+                                 start_new_session=True)
+        if stopped:
+            child.send_signal(stopped)
+        code = child.wait()
+        return code if code >= 0 else 128 - code
+    except (OSError, ValueError):
+        print("Could not start the virtual display or container command.", file=sys.stderr)
+        return 3
+    finally:
+        # Only clean up descendants after the command has finalized its files/transaction.
+        if child is not None:
+            try:
+                os.killpg(child.pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+        if display is not None:
+            display.terminate()
+            try:
+                display.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                display.kill()
+                display.wait()
+        os.close(reader)
+        if writer is not None:
+            os.close(writer)
+        for sig, handler in previous.items():
+            signal.signal(sig, handler)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/src/car_traker/pipeline.py
+++ b/src/car_traker/pipeline.py
@ -0,0 +1,106 @@
+"""Collect raw observations and import that exact run into PostgreSQL."""
+
+from contextlib import redirect_stderr
+import fcntl
+import io
+import json
+import os
+import signal
+import sys
+
+import psycopg
+
+from .collect import collect_browser, parse_collection_args, write_json
+from .database import InputError, database_error_hint, import_run
+
+
+def run_pipeline(args):
+    """Run one locked collection/import job and return its CLI exit code."""
+    try:
+        if not any(os.environ.get(key) for key in (
+            "DATABASE_URL", "PGSERVICE", "PGHOST", "PGHOSTADDR",
+        )):
+            raise InputError("database_configuration_required")
+        conninfo = os.environ.get("DATABASE_URL", "")
+        args.output.mkdir(mode=0o700, parents=True, exist_ok=True)
+        # Keep the inode: unlinking would let another process lock a different file.
+        with (args.output / ".pipeline.lock").open("a") as lock:
+            try:
+                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                print("Another pipeline job holds the output lock.", file=sys.stderr)
+                return 3
+            sidecar = None
+            state = None
+            try:
+                with psycopg.connect(conninfo, connect_timeout=10, autocommit=True) as conn:
+                    conn.execute("SELECT 1 FROM silver.collection_runs LIMIT 0")
+                path, status = collect_browser(args)
+                print(f"Manifest: {path}", flush=True)
+                sidecar = path.parent / "pipeline.json"
+                state = {"schema_version": 1, "collection_status": status, "import_status": "not_attempted"}
+                manifest = json.loads(path.read_text(encoding="utf-8"))
+                if manifest.get("error_type") == "KeyboardInterrupt":
+                    raise KeyboardInterrupt
+                write_json(sidecar, state)
+                state["import_status"] = "running"
+                write_json(sidecar, state)
+                with psycopg.connect(conninfo, connect_timeout=10, autocommit=True) as conn:
+                    summary = import_run(conn, path)
+                state.update(import_status="succeeded", import_summary=summary)
+                write_json(sidecar, state)
+                print(json.dumps(state), flush=True)
+                if summary.get("rejected", 0):
+                    return 2
+                return 0 if status == "complete" else 1
+            except (Exception, KeyboardInterrupt) as exc:
+                if state is not None:
+                    if state["import_status"] == "running":
+                        # A lost response or interruption can occur after PostgreSQL commits.
+                        state["import_status"] = "unknown"
+                    state["error_type"] = type(exc).__name__
+                    try:
+                        write_json(sidecar, state)
+                    except Exception:
+                        print("Could not write pipeline outcome; raw files are retained.", file=sys.stderr)
+                raise
+    except KeyboardInterrupt:
+        print("Pipeline interrupted; raw files are retained.", file=sys.stderr)
+        return 130
+    except InputError:
+        print("Pipeline input/configuration error; check configuration and raw files.", file=sys.stderr)
+    except psycopg.Error as exc:
+        print(f"Database error: {database_error_hint(exc)}", file=sys.stderr)
+    except Exception:
+        print("Pipeline failed; check browser, configuration and storage. Raw files are retained.", file=sys.stderr)
+    return 3
+
+
+def main(argv=None):
+    def interrupted(signum, frame):
+        raise KeyboardInterrupt
+
+    # Docker stops one-shot jobs with SIGTERM; let the collector finalize coverage.
+    previous = signal.signal(signal.SIGTERM, interrupted)
+    try:
+        try:
+            # argparse can echo arbitrary argument values; keep configuration errors private.
+            with redirect_stderr(io.StringIO()):
+                args = parse_collection_args(argv, description=__doc__)
+        except KeyboardInterrupt:
+            return 130
+        except SystemExit as exc:
+            if exc.code == 0:
+                return 0
+            print("Invalid pipeline arguments or collection configuration.", file=sys.stderr)
+            return 3
+        except Exception:
+            print("Could not load collection configuration.", file=sys.stderr)
+            return 3
+        return run_pipeline(args)
+    finally:
+        signal.signal(signal.SIGTERM, previous)
+
+
+if __name__ == "__main__":
+    sys.exit(main())
--- a/tests/test_browser_pipeline.py
+++ b/tests/test_browser_pipeline.py
@ -0,0 +1,202 @@
+"""Opt-in headed Chromium tests against a disposable *_test PostgreSQL database."""
+
+from contextlib import redirect_stderr, redirect_stdout
+from decimal import Decimal
+import gzip
+import io
+import json
+import os
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import unittest
+from unittest.mock import patch
+
+from psycopg import sql
+
+from car_traker import database, pipeline
+from car_traker.collect import search_url
+import test_database_integration as database_tests
+
+
+@unittest.skipUnless(
+    os.environ.get("RUN_BROWSER_TESTS") == "1" and os.environ.get("TEST_DATABASE_URL"),
+    "RUN_BROWSER_TESTS=1 and TEST_DATABASE_URL are required",
+)
+class BrowserPipelineTests(unittest.TestCase):
+    tables = database_tests.DatabaseIntegrationTests.tables
+
+    @classmethod
+    def setUpClass(cls):
+        cls.conn = database_tests.DatabaseIntegrationTests.connect()
+        cls.addClassCleanup(cls.conn.close)
+        database.migrate(cls.conn)
+
+    def setUp(self):
+        self.conn.execute(sql.SQL("TRUNCATE {} RESTART IDENTITY").format(
+            sql.SQL(", ").join(sql.Identifier("silver", table) for table in self.tables)
+        ))
+        temporary = TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.directory = Path(temporary.name)
+        self.config = self.directory / "searches.toml"
+        self.config.write_text(
+            '[[searches]]\nfamily = "synthetic"\ntext = "synthetic car"\n',
+            encoding="utf-8",
+        )
+        self.output = self.directory / "raw"
+        self.search = {"list_id": 123, "price": ["12000"]}
+        self.detail = {
+            "list_id": 123, "price": ["12500.50"],
+            "body": "Synthetic private description.\nRetain this raw text exactly.",
+            "location": {"lat": 48.125, "lng": 2.375},
+        }
+
+    def run_browser_pipeline(self, *, detail_status=200, import_status="succeeded"):
+        # Import only when opted in; never replace the browser, collector or importer.
+        from playwright.sync_api import BrowserContext
+
+        original_new_page = BrowserContext.new_page
+        search = search_url("synthetic car", 1)
+        detail = "https://www.leboncoin.fr/ad/voitures/123"
+        requests, responses, unexpected = [], [], []
+
+        def handle(route):
+            url = route.request.url
+            requests.append(url)
+            if url == search:
+                status = 200
+                props = {"searchData": {"ads": [self.search], "total": 1, "max_pages": 1}}
+            elif url == detail:
+                status = detail_status
+                # Even the 403 has valid data: ignoring HTTP status must fail this test.
+                props = {"ad": self.detail}
+            else:
+                unexpected.append(url)
+                route.abort()
+                return
+            data = json.dumps({"props": {"pageProps": props}})
+            route.fulfill(
+                status=status, content_type="text/html",
+                body=f'<!doctype html><script id="__NEXT_DATA__" type="application/json">{data}</script>',
+            )
+
+        def new_page(context, *args, **kwargs):
+            page = original_new_page(context, *args, **kwargs)
+            page.route("**/*", handle)
+            page.on("response", lambda response: responses.append((response.url, response.status)))
+            return page
+
+        stdout, stderr = io.StringIO(), io.StringIO()
+        # Preserve DISPLAY/browser paths, but discard inherited libpq/homelab settings.
+        environment = {key: value for key, value in os.environ.items() if not key.startswith("PG")}
+        environment["DATABASE_URL"] = os.environ["TEST_DATABASE_URL"]
+        with patch.dict(os.environ, environment, clear=True), patch.object(
+            BrowserContext, "new_page", new_page,
+        ), patch("car_traker.collect.time.sleep"), redirect_stdout(stdout), redirect_stderr(stderr):
+            code = pipeline.main([
+                "--headed", "--config", str(self.config), "--output", str(self.output),
+                "--delay", "1", "--max-pages", "1", "--max-details", "1",
+            ])
+
+        self.assertEqual(unexpected, [])
+        self.assertEqual(requests, [search, detail])
+        self.assertEqual(responses, [(search, 200), (detail, detail_status)])
+        self.assertNotIn(self.detail["body"], stdout.getvalue() + stderr.getvalue())
+        self.assertNotIn("48.125", stdout.getvalue() + stderr.getvalue())
+        self.assertNotIn("2.375", stdout.getvalue() + stderr.getvalue())
+        paths = list(self.output.rglob("manifest.json"))
+        self.assertEqual(len(paths), 1)
+        path = paths[0]
+        manifest = json.loads(path.read_text(encoding="utf-8"))
+        state = json.loads((path.parent / "pipeline.json").read_text(encoding="utf-8"))
+        self.assertEqual(state["collection_status"], manifest["status"])
+        self.assertEqual(state["import_status"], import_status)
+        if import_status == "succeeded":
+            self.assertEqual(state["import_summary"]["status"], "imported")
+            self.assertEqual(state["import_summary"]["source_status"], manifest["status"])
+            self.assertEqual(state["import_summary"]["rejected"], 0)
+        self.assertEqual(self.conn.execute(
+            "SELECT run_id::text, source_status, manifest FROM silver.collection_runs"
+        ).fetchall(), [(manifest["run_id"], manifest["status"], manifest)])
+        self.assertEqual(self.conn.execute(
+            "SELECT family, status, pages_observed, records_reported FROM silver.search_coverage"
+        ).fetchall(), [("synthetic", "complete", 1, 1)])
+        with gzip.open(path.parent / manifest["searches"][0]["pages"][0]["file"], "rt") as stream:
+            records = [json.loads(line) for line in stream]
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["record_type"], "search")
+        self.assertEqual(records[0]["payload"], self.search)
+        return code, path, manifest
+
+    def test_complete_run_preserves_raw_and_typed_detail_and_replays(self):
+        code, path, manifest = self.run_browser_pipeline()
+        self.assertEqual(code, 0)
+        self.assertEqual(manifest["status"], "complete")
+        with gzip.open(path.parent / manifest["details"]["listings"]["123"]["file"], "rt") as stream:
+            records = [json.loads(line) for line in stream]
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0]["record_type"], "detail")
+        self.assertEqual(records[0]["payload"], self.detail)
+        self.assertEqual(self.conn.execute(
+            "SELECT source_listing_id, record_type, price_eur, currency, description, "
+            "latitude, longitude, raw_payload FROM silver.observations ORDER BY record_type"
+        ).fetchall(), [
+            ("123", "detail", Decimal("12500.50"), "EUR", self.detail["body"],
+             48.125, 2.375, self.detail),
+            ("123", "search", Decimal("12000"), "EUR", None, None, None, self.search),
+        ])
+        self.assertEqual(self.conn.execute(
+            "SELECT source_listing_id, status FROM silver.detail_coverage"
+        ).fetchall(), [("123", "complete")])
+        before = self.state()
+        self.assertEqual({table: len(rows) for table, rows in before.items()}, {
+            "collection_runs": 1, "raw_files": 2, "search_coverage": 1,
+            "detail_coverage": 1, "listings": 1, "observations": 2, "rejected_records": 0,
+        })
+        with redirect_stdout(io.StringIO()):
+            replay = database.import_run(self.conn, path)
+        self.assertEqual(replay, {
+            "run_id": manifest["run_id"], "status": "already_imported", "rejected": 0,
+        })
+        self.assertEqual(self.state(), before)
+
+    def test_blocked_detail_imports_partial_run_and_retains_search(self):
+        code, path, manifest = self.run_browser_pipeline(detail_status=403)
+        self.assertEqual(code, 1)
+        self.assertEqual(manifest["status"], "partial")
+        self.assertEqual(manifest["error_type"], "SourceAccessError")
+        self.assertEqual(manifest["details"]["status"], "partial")
+        self.assertEqual(manifest["details"]["listings"]["123"]["http_status"], 403)
+        self.assertFalse((path.parent / "detail-123.jsonl.gz").exists())
+        self.assertEqual(self.conn.execute(
+            "SELECT source_listing_id, status, reason, http_status FROM silver.detail_coverage"
+        ).fetchall(), [("123", "failed", "SourceAccessError", 403)])
+        self.assertEqual(self.conn.execute(
+            "SELECT source_listing_id, record_type, price_eur, raw_payload FROM silver.observations"
+        ).fetchall(), [("123", "search", Decimal("12000"), self.search)])
+        self.assertEqual(self.conn.execute(
+            "SELECT record_type, expected_records, accepted_records, rejected_records FROM silver.raw_files"
+        ).fetchall(), [("search", 1, 1, 0)])
+        self.assertEqual(self.conn.execute("SELECT count(*) FROM silver.rejected_records").fetchone(), (0,))
+
+    def state(self):
+        return {table: self.conn.execute(sql.SQL(
+            "SELECT to_jsonb(t) FROM silver.{} t ORDER BY to_jsonb(t)::text"
+        ).format(sql.Identifier(table))).fetchall() for table in self.tables}
+
+    def test_interrupt_after_commit_is_unknown_and_replay_is_safe(self):
+        def committed_then_interrupted(conn, path):
+            database.import_run(conn, path)
+            raise KeyboardInterrupt
+
+        with patch.object(pipeline, "import_run", side_effect=committed_then_interrupted):
+            code, path, _ = self.run_browser_pipeline(import_status="unknown")
+        self.assertEqual(code, 130)
+        before = self.state()
+        self.assertEqual(len(before["observations"]), 2)
+        self.assertEqual(database.import_run(self.conn, path)["status"], "already_imported")
+        self.assertEqual(self.state(), before)
+
+
+if __name__ == "__main__":
+    unittest.main()
--- a/tests/test_collect_browser.py
+++ b/tests/test_collect_browser.py
@ -0,0 +1,43 @@
+from argparse import Namespace
+import io
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import unittest
+from unittest.mock import patch
+
+from car_traker.collect import collect_browser
+
+
+class BrowserCleanupTests(unittest.TestCase):
+    def setUp(self):
+        temporary = TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.path = Path(temporary.name) / "manifest.json"
+        self.path.write_text('{"status":"complete"}')
+        self.args = Namespace(headed=False, searches=[], output=self.path.parent,
+                              max_pages=1, delay=1, search_only=True, max_details=None)
+        self.playwright = self.enterContext(patch("playwright.sync_api.sync_playwright"))
+        self.browser = self.playwright.return_value.__enter__.return_value.chromium.launch.return_value
+        self.collect = self.enterContext(patch("car_traker.collect.collect", return_value=(self.path, "complete")))
+        self.stderr = self.enterContext(patch("sys.stderr", new_callable=io.StringIO))
+
+    def test_browser_close_failure_retains_finalized_result(self):
+        self.browser.close.side_effect = RuntimeError("SYNTHETIC_SECRET")
+        self.assertEqual(collect_browser(self.args), (self.path, "complete"))
+        self.assertTrue((self.path.parent / "browser_cleanup.json").exists())
+        self.assertNotIn("SYNTHETIC_SECRET", self.stderr.getvalue())
+
+    def test_driver_cleanup_failure_retains_finalized_result(self):
+        self.playwright.return_value.__exit__.side_effect = RuntimeError("cleanup failed")
+        self.assertEqual(collect_browser(self.args), (self.path, "complete"))
+
+    def test_collection_failure_is_not_suppressed(self):
+        self.collect.side_effect = OSError("write failed")
+        with self.assertRaises(OSError):
+            collect_browser(self.args)
+
+    def test_cleanup_interruption_remains_cancellation_with_manifest_path(self):
+        self.browser.close.side_effect = KeyboardInterrupt
+        with self.assertRaises(KeyboardInterrupt):
+            collect_browser(self.args)
+        self.assertIn(str(self.path), self.stderr.getvalue())
--- a/tests/test_container_entrypoint.py
+++ b/tests/test_container_entrypoint.py
@ -0,0 +1,57 @@
+import io
+import os
+import select
+import signal
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+
+from car_traker import container_entrypoint
+
+
+class ContainerEntrypointTests(unittest.TestCase):
+    def test_requires_a_command(self):
+        with patch("sys.argv", ["entrypoint"]), patch("sys.stderr", new_callable=io.StringIO):
+            self.assertEqual(container_entrypoint.main(), 3)
+
+
+@unittest.skipUnless(os.environ.get("RUN_BROWSER_TESTS") == "1", "Xvfb container tests are opt-in")
+class VirtualDisplayTests(unittest.TestCase):
+    def test_child_exit_code_is_preserved(self):
+        result = subprocess.run([
+            sys.executable, "-m", "car_traker.container_entrypoint",
+            sys.executable, "-c", "raise SystemExit(7)",
+        ], capture_output=True, text=True, timeout=20)
+        self.assertEqual(result.returncode, 7, result.stderr)
+
+    def test_sigterm_waits_for_child_finalization(self):
+        command = """
+import signal, sys, time
+def stop(signum, frame):
+    print('cleanup-start', flush=True)
+    time.sleep(1)
+    print('cleanup-done', flush=True)
+    sys.exit(130)
+signal.signal(signal.SIGTERM, stop)
+print('ready', flush=True)
+while True:
+    signal.pause()
+"""
+        process = subprocess.Popen([
+            sys.executable, "-m", "car_traker.container_entrypoint",
+            sys.executable, "-c", command,
+        ], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        try:
+            ready, _, _ = select.select([process.stdout], [], [], 15)
+            self.assertTrue(ready, "Child did not become ready")
+            self.assertEqual(process.stdout.readline().strip(), "ready")
+            process.send_signal(signal.SIGTERM)
+            stdout, stderr = process.communicate(timeout=15)
+            self.assertEqual(process.returncode, 130, stderr)
+            self.assertIn("cleanup-start", stdout)
+            self.assertIn("cleanup-done", stdout)
+        finally:
+            if process.poll() is None:
+                process.terminate()
+                process.communicate(timeout=15)
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@ -0,0 +1,222 @@
+import argparse
+from contextlib import redirect_stderr, redirect_stdout
+import fcntl
+import io
+import json
+import os
+import signal
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from car_traker import pipeline
+
+
+class PipelineTests(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.output = Path(self.directory.name) / "raw"
+        self.args = argparse.Namespace(output=self.output, searches=[])
+        self.stdout, self.stderr = io.StringIO(), io.StringIO()
+        self.enterContext(redirect_stdout(self.stdout))
+        self.enterContext(redirect_stderr(self.stderr))
+        self.enterContext(patch.dict(os.environ, {"PGHOST": "homelab"}, clear=True))
+        self.preflight, self.import_connection = MagicMock(), MagicMock()
+        self.preflight.__enter__.return_value = self.preflight
+        self.import_connection.__enter__.return_value = self.import_connection
+        self.connect = self.enterContext(patch.object(
+            pipeline.psycopg, "connect", side_effect=[self.preflight, self.import_connection],
+        ))
+        self.collect = self.enterContext(patch.object(pipeline, "collect_browser", side_effect=self.fixture))
+        self.summary = {"status": "imported", "accepted": 1, "rejected": 0}
+        self.importer = self.enterContext(patch.object(pipeline, "import_run", return_value=self.summary))
+        self.path = self.output / "returned-run" / "manifest.json"
+        self.raw = self.path.parent / "search-001.jsonl.gz"
+
+    def fixture(self, args, status="complete", error_type=None):
+        self.preflight.__exit__.assert_called_once()
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        manifest = {"status": status}
+        if error_type:
+            manifest["error_type"] = error_type
+        self.path.write_text(json.dumps(manifest), encoding="utf-8")
+        self.raw.write_bytes(b"raw snapshot")
+        self.manifest_bytes = self.path.read_bytes()
+        return self.path, status
+
+    def state(self):
+        return json.loads((self.path.parent / "pipeline.json").read_text())
+
+    def test_collection_statuses_import_exact_manifest(self):
+        for status, code in (("complete", 0), ("partial", 1), ("failed", 1)):
+            with self.subTest(status=status):
+                self.preflight.reset_mock()
+                self.connect.side_effect = [self.preflight, self.import_connection]
+                self.collect.side_effect = lambda args: self.fixture(args, status)
+                stale = self.output / "zz-newer-run" / "manifest.json"
+                stale.parent.mkdir(parents=True, exist_ok=True)
+                stale.write_text('{"status": "complete"}')
+                output_mode = self.output.stat().st_mode
+                self.assertEqual(pipeline.run_pipeline(self.args), code)
+                self.assertEqual(self.output.stat().st_mode, output_mode)
+                self.importer.assert_called_with(self.import_connection, self.path)
+                self.assertEqual(self.path.read_bytes(), self.manifest_bytes)
+                self.assertEqual(self.state()["collection_status"], status)
+                self.assertEqual(self.state()["import_status"], "succeeded")
+        self.preflight.execute.assert_called_once_with("SELECT 1 FROM silver.collection_runs LIMIT 0")
+        for call in self.connect.call_args_list:
+            self.assertEqual(call.args, ("",))
+            self.assertEqual(call.kwargs, {"connect_timeout": 10, "autocommit": True})
+
+    def test_new_output_directory_is_private(self):
+        self.assertEqual(pipeline.run_pipeline(self.args), 0)
+        self.assertEqual(self.output.stat().st_mode & 0o777, 0o700)
+
+    def test_rejections_take_priority_over_partial_collection(self):
+        for status in ("complete", "partial", "failed"):
+            with self.subTest(status=status):
+                self.preflight.reset_mock()
+                self.connect.side_effect = [self.preflight, self.import_connection]
+                self.collect.side_effect = lambda args: self.fixture(args, status)
+                self.summary["rejected"] = 1
+                self.assertEqual(pipeline.run_pipeline(self.args), 2)
+
+    def test_import_failure_preserves_raw_and_reports_promptly_without_secrets(self):
+        def fail(conn, path):
+            self.assertIn(str(path), self.stdout.getvalue())
+            raise pipeline.psycopg.OperationalError("password authentication failed TOP_SECRET")
+        self.importer.side_effect = fail
+        self.assertEqual(pipeline.run_pipeline(self.args), 3)
+        self.assertEqual(self.raw.read_bytes(), b"raw snapshot")
+        self.assertEqual(self.path.read_bytes(), self.manifest_bytes)
+        self.assertEqual(self.state()["import_status"], "unknown")
+        self.assertEqual(self.state()["error_type"], "OperationalError")
+        self.assertNotIn("TOP_SECRET", self.stderr.getvalue() + self.stdout.getvalue() + json.dumps(self.state()))
+
+    def test_preflight_failure_never_collects(self):
+        self.preflight.execute.side_effect = pipeline.psycopg.OperationalError("TOP_SECRET")
+        self.assertEqual(pipeline.run_pipeline(self.args), 3)
+        self.collect.assert_not_called()
+        self.importer.assert_not_called()
+        self.preflight.__exit__.assert_called_once()
+        self.assertNotIn("TOP_SECRET", self.stderr.getvalue())
+
+    def test_finalized_interruption_is_not_imported(self):
+        self.collect.side_effect = lambda args: self.fixture(args, "partial", "KeyboardInterrupt")
+        self.assertEqual(pipeline.run_pipeline(self.args), 130)
+        self.importer.assert_not_called()
+        self.assertEqual(self.state()["import_status"], "not_attempted")
+        self.assertEqual(self.state()["error_type"], "KeyboardInterrupt")
+        self.assertEqual(self.path.read_bytes(), self.manifest_bytes)
+
+    def test_interruption_before_collector_returns(self):
+        self.collect.side_effect = KeyboardInterrupt
+        self.assertEqual(pipeline.run_pipeline(self.args), 130)
+        self.importer.assert_not_called()
+
+    def test_interruption_during_import_exits_connection_with_interrupt(self):
+        self.importer.side_effect = KeyboardInterrupt
+        self.assertEqual(pipeline.run_pipeline(self.args), 130)
+        self.assertIs(self.import_connection.__exit__.call_args.args[0], KeyboardInterrupt)
+        self.assertEqual(self.state()["import_status"], "unknown")
+
+    def test_lock_refusal_and_release_keeps_inode(self):
+        self.output.mkdir()
+        lock_path = self.output / ".pipeline.lock"
+        with lock_path.open("a") as lock:
+            inode = lock_path.stat().st_ino
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            self.assertEqual(pipeline.run_pipeline(self.args), 3)
+            self.collect.assert_not_called()
+            self.connect.assert_not_called()
+        self.collect.side_effect = RuntimeError("TOP_SECRET browser crash")
+        self.assertEqual(pipeline.run_pipeline(self.args), 3)
+        self.assertEqual(lock_path.stat().st_ino, inode)
+        with lock_path.open("a") as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        self.importer.assert_not_called()
+        self.assertNotIn("TOP_SECRET", self.stderr.getvalue())
+
+    def test_no_implicit_database(self):
+        with patch.dict(os.environ, {"PGUSER": "user", "PGDATABASE": "db"}, clear=True):
+            self.assertEqual(pipeline.run_pipeline(self.args), 3)
+        self.connect.assert_not_called()
+        self.collect.assert_not_called()
+
+    def test_lock_is_held_through_preflight_collection_and_import(self):
+        def assert_locked():
+            with (self.output / ".pipeline.lock").open("a") as lock:
+                with self.assertRaises(BlockingIOError):
+                    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+        def collect(args):
+            assert_locked()
+            return self.fixture(args)
+
+        def import_run(conn, path):
+            assert_locked()
+            return self.summary
+
+        self.preflight.execute.side_effect = lambda query: assert_locked()
+        self.collect.side_effect = collect
+        self.importer.side_effect = import_run
+        self.assertEqual(pipeline.run_pipeline(self.args), 0)
+        with (self.output / ".pipeline.lock").open("a") as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+    def test_explicit_database_configuration(self):
+        for env, conninfo in (({"DATABASE_URL": "postgresql://configured"}, "postgresql://configured"),
+                              ({"PGSERVICE": "homelab"}, ""), ({"PGHOSTADDR": "192.0.2.1"}, "")):
+            with self.subTest(env=env), patch.dict(os.environ, env, clear=True):
+                self.preflight.reset_mock()
+                self.connect.side_effect = [self.preflight, self.import_connection]
+                self.assertEqual(pipeline.run_pipeline(self.args), 0)
+                self.connect.assert_called_with(conninfo, connect_timeout=10, autocommit=True)
+
+    def test_input_error_message_is_not_logged(self):
+        self.importer.side_effect = pipeline.InputError("TOP_SECRET")
+        self.assertEqual(pipeline.run_pipeline(self.args), 3)
+        self.assertNotIn("TOP_SECRET", self.stderr.getvalue() + json.dumps(self.state()))
+
+    def test_sidecar_storage_failure_prevents_import(self):
+        with patch.object(pipeline, "write_json", side_effect=OSError("TOP_SECRET")):
+            self.assertEqual(pipeline.run_pipeline(self.args), 3)
+        self.importer.assert_not_called()
+        self.assertTrue(self.raw.exists())
+        self.assertNotIn("TOP_SECRET", self.stderr.getvalue())
+
+    def test_main_delegates_parser_and_runner(self):
+        with patch.object(pipeline, "parse_collection_args", return_value=self.args) as parser:
+            with patch.object(pipeline, "run_pipeline", return_value=2) as run:
+                self.assertEqual(pipeline.main(["--search-only"]), 2)
+        parser.assert_called_once_with(["--search-only"], description=pipeline.__doc__)
+        run.assert_called_once_with(self.args)
+
+    def test_main_sanitizes_config_failures_and_handles_interrupt(self):
+        for error, code in ((ValueError("TOP_SECRET"), 3), (SystemExit(2), 3),
+                            (KeyboardInterrupt(), 130), (SystemExit(0), 0)):
+            with self.subTest(error=type(error).__name__):
+                def fail(*args, **kwargs):
+                    print("TOP_SECRET", file=pipeline.sys.stderr)
+                    raise error
+                with patch.object(pipeline, "parse_collection_args", side_effect=fail):
+                    self.assertEqual(pipeline.main([]), code)
+        self.assertNotIn("TOP_SECRET", self.stderr.getvalue())
+
+    def test_sigterm_is_handled_and_previous_handler_restored(self):
+        previous = signal.getsignal(signal.SIGTERM)
+
+        def terminated(args):
+            os.kill(os.getpid(), signal.SIGTERM)
+
+        self.collect.side_effect = terminated
+        with patch.object(pipeline, "parse_collection_args", return_value=self.args):
+            self.assertEqual(pipeline.main([]), 130)
+        self.importer.assert_not_called()
+        self.assertIs(signal.getsignal(signal.SIGTERM), previous)
+
+
+if __name__ == "__main__":
+    unittest.main()