diff --git a/docs/jobcoordinator.md b/docs/jobcoordinator.md
new file mode 100644
index 00000000..cb24b8c2
--- /dev/null
+++ b/docs/jobcoordinator.md
@@ -0,0 +1,143 @@
+# Job Coordinator (database-backed)
+
+The job coordinator turns **any executable** into a distributed worker. It is a
+database-backed sibling of the file-based grid wrapper: instead of
+`.LOCKED/.PROCESSED/.FAILED` marker files on a shared filesystem, the queue and
+its coordination state live in a database (SQLite or PostgreSQL).
+
+A **job** is a whole batch/run; a **task** is one unit of work within it
+(typically a file path). Two commands make up the workflow:
+
+| Command | Purpose |
+| --- | --- |
+| `claimed propagate_jobs` | Populate a job with tasks from a glob/directory (status `pending`). |
+| `claimed work_jobs`      | Claim the job's tasks one-by-one and run a worker script per task. |
+
+Because the database is the single source of truth and claiming is atomic, many
+workers — on one machine or across many nodes — cooperate safely and never
+process the same task twice.
+
+## Concepts
+
+Each task is one row in the `claimed_jobs` table, identified by its **job** (the
+batch namespace) plus its **task_name** (the file path emitted by
+`propagate_jobs`). The job is a mandatory namespace: it lets many independent
+batches share one database, and workers only ever claim tasks from the job they
+were pointed at. Uniqueness is per `(job, task_name)`, so the same file path may
+appear in more than one job. A task moves through:
+
+```
+pending ── claim ──▶ processing ── worker exit 0 ──▶ succeeded
+                               └── worker exit ≠0 ──▶ failed
+```
+
+The table also records `worker_id`, `attempts`, timestamps, and (on failure) an
+`error` message.
+
+## `propagate_jobs`
+
+```bash
+claimed propagate_jobs --db <url> --job <name> <pattern>
+```
+
+- `--db <url>` — a SQLite file path (e.g. `jobs.db`, `/tmp/jobs.db`) or a full
+  database URL. Bare paths and `*.db` / `*.sqlite` become SQLite; `sqlite:///…`
+  and `postgresql://…` (also the legacy `postgres://…`) are used as given.
+- `--job <name>` — **mandatory** namespace for this batch. Use a distinct job
+  name per independent run so several runs can share one database.
+- `<pattern>` — a glob (with `**` recursion) or a directory. **Quote it** so the
+  shell does not expand the glob before CLAIMED sees it.
+
+The command is **idempotent**: re-running only inserts `(job, task_name)` pairs
+that do not already exist (`INSERT … ON CONFLICT DO NOTHING`), so you can grow
+the queue incrementally.
+
+```bash
+claimed propagate_jobs --db /tmp/jobs.db --job run-2026 'data/**/*.tif'
+# Inserted 42 tasks into job 'run-2026' (0 already existed). Total pending: 42
+```
+
+## `work_jobs`
+
+```bash
+claimed work_jobs --db <url> --job <name> --worker <script> [options]
+```
+
+The worker loop claims the oldest pending task **in `--job`**, marks it
+`processing`, runs the worker, then marks it `succeeded` (exit 0) or `failed`
+(any other exit code or a launch error). It exits when the job's queue is
+empty. The job must match the one used in `propagate_jobs`.
+
+The task name is passed to the worker **both** ways:
+
+- as the first positional argument (`$1`)
+- as the `CLAIMED_TASK` environment variable
+
+`CLAIMED_JOB` and `CLAIMED_WORKER_ID` are also exported.
+
+Options:
+
+| Option | Default | Meaning |
+| --- | --- | --- |
+| `--worker-id NAME`   | `<hostname>-<pid>` | Identifier recorded on claimed tasks. |
+| `--max-tasks N`      | drain queue        | Stop after N tasks this run. |
+| `--poll-interval S`  | `0`                | When empty, wait S seconds and retry instead of exiting. |
+
+### Running workers in parallel
+
+Just launch `work_jobs` more than once against the same `--db` and `--job`:
+
+```bash
+for i in 1 2 3 4; do
+  claimed work_jobs --db postgresql://user:pass@host/claimed --job run-2026 --worker ./worker.sh &
+done
+wait
+```
+
+Two runs that use **different** `--job` values against the same database run
+fully independently — neither sees or claims the other's tasks.
+
+## Atomic claiming
+
+Claiming is always scoped to the worker's `--job`.
+
+- **PostgreSQL** uses `SELECT … FOR UPDATE SKIP LOCKED` so concurrent workers
+  skip rows another worker is already claiming — no contention, no duplicates.
+- **SQLite** has no `SKIP LOCKED`; writers are serialized with `BEGIN IMMEDIATE`
+  plus a guarded `UPDATE … WHERE status='pending'` and a short retry loop, with
+  WAL mode and a busy timeout enabled for cross-process concurrency.
+
+## PostgreSQL support
+
+PostgreSQL needs the `psycopg2` driver, available via the project extra:
+
+```bash
+pip install 'claimed[postgresql]'
+```
+
+Remote hosts should use SSL — pass it in the URL, e.g.
+`postgresql://user:pass@host:5432/claimed?sslmode=require`.
+
+## Writing a worker
+
+Any executable works. Read the task name, do the work, and use the exit code to
+report success or failure:
+
+```bash
+#!/usr/bin/env bash
+set -euo pipefail
+TASK="${1:-$CLAIMED_TASK}"
+python process.py --input "$TASK" --output "out/$(basename "$TASK")"
+```
+
+A complete, runnable example lives in
+[`examples/jobcoordinator_example/`](https://github.com/claimed-framework/claimed/tree/main/examples/jobcoordinator_example).
+
+## Relationship to the grid wrapper
+
+The [grid wrapper](c3/create-gridwrapper.md) coordinates work through marker
+files on a shared filesystem (or object store) and wraps a CLAIMED *component*.
+The job coordinator coordinates through a database and wraps an *arbitrary
+script*. Use the grid wrapper when you already have a containerised component and
+a shared filesystem; use the job coordinator when you want a database as the
+source of truth, robust multi-node claiming, or a plain shell/Python worker.
diff --git a/examples/jobcoordinator_example/README.md b/examples/jobcoordinator_example/README.md
new file mode 100644
index 00000000..92ca1385
--- /dev/null
+++ b/examples/jobcoordinator_example/README.md
@@ -0,0 +1,78 @@
+# Job Coordinator example
+
+A database-backed coordinator that turns any shell script into a distributed
+worker. One command fills a **job** (a batch of tasks) from a set of files;
+another pulls the job's **tasks** one-by-one and runs your worker for each. The
+database (SQLite or PostgreSQL) holds the queue and provides atomic claiming, so
+many worker sessions — on the same machine or across many nodes — can cooperate
+without a shared filesystem and without ever processing the same task twice.
+
+See [`../../docs/jobcoordinator.md`](../../docs/jobcoordinator.md) for the full
+reference.
+
+## 1. Create some input files
+
+```bash
+mkdir -p /tmp/jc/data
+touch /tmp/jc/data/a.tif /tmp/jc/data/b.tif /tmp/jc/data/c.tif
+```
+
+## 2. Propagate tasks (status: pending)
+
+Every batch is namespaced by a mandatory `--job`, so many independent jobs can
+share one database without colliding.
+
+Local SQLite (no server needed — a bare path becomes a SQLite database):
+
+```bash
+claimed propagate_jobs --db /tmp/jc/jobs.db --job scenes-2026 '/tmp/jc/data/**/*.tif'
+# Inserted 3 tasks into job 'scenes-2026' (0 already existed). Total pending: 3
+```
+
+PostgreSQL (real multi-node coordination):
+
+```bash
+claimed propagate_jobs \
+  --db postgresql://user:pass@localhost:5432/claimed \
+  --job scenes-2026 \
+  '/data/scenes/**/*.tif'
+```
+
+Quote the glob so your shell does not expand it before CLAIMED sees it.
+`propagate_jobs` is idempotent: re-running skips `(job, task_name)` pairs that
+already exist.
+
+## 3. Work the queue
+
+```bash
+claimed work_jobs --db /tmp/jc/jobs.db --job scenes-2026 --worker ./worker.sh
+```
+
+Each task is claimed atomically and marked `processing`, then `succeeded` /
+`failed` based on the worker's exit code. The task name (the file path) is passed
+to the worker as `$1` **and** as `$CLAIMED_TASK` (with `$CLAIMED_JOB` and
+`$CLAIMED_WORKER_ID` also exported). Workers only claim tasks from their own
+`--job`.
+
+Run several workers at once for parallelism — just launch the command multiple
+times (or on multiple nodes) against the same `--db` and `--job`:
+
+```bash
+claimed work_jobs --db /tmp/jc/jobs.db --job scenes-2026 --worker ./worker.sh &
+claimed work_jobs --db /tmp/jc/jobs.db --job scenes-2026 --worker ./worker.sh &
+wait
+```
+
+### Useful options
+
+- `--worker-id NAME`   label recorded on claimed tasks (default `<host>-<pid>`)
+- `--max-tasks N`      stop after N tasks (default: drain the queue)
+- `--poll-interval S`  when the queue is empty, wait S seconds and retry instead
+                       of exiting — handy when producers keep adding tasks
+
+## Writing your own worker
+
+`worker.sh` here is a template. Read the task name from `$1` or
+`$CLAIMED_TASK`, do your work, and exit `0` on success or non-zero on
+failure. Anything executable works — a Python script, a `bsub`/`sbatch`
+submission wrapper, a container invocation, etc.
diff --git a/examples/jobcoordinator_example/worker.sh b/examples/jobcoordinator_example/worker.sh
new file mode 100755
index 00000000..f9de9ec7
--- /dev/null
+++ b/examples/jobcoordinator_example/worker.sh
@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+#
+# Example CLAIMED job-coordinator worker.
+#
+# The coordinator (`claimed work_jobs`) invokes this script once per task and
+# passes the task name (the file path recorded by `claimed propagate_jobs`) in
+# two ways, so use whichever is convenient:
+#
+#   $1                  -> the task name (first positional argument)
+#   $CLAIMED_TASK       -> the task name (environment variable)
+#   $CLAIMED_JOB        -> the job (batch) this task belongs to
+#   $CLAIMED_WORKER_ID  -> id of the worker session running this task
+#
+# Contract:
+#   exit 0        -> the coordinator marks the task "succeeded"
+#   exit non-zero -> the coordinator marks the task "failed" (error recorded)
+#
+# Replace the body with your real work: load the file at "$TASK", process it,
+# and write results wherever you need.
+set -euo pipefail
+
+TASK="${1:-${CLAIMED_TASK:-}}"
+
+if [[ -z "$TASK" ]]; then
+  echo "worker.sh: no task name provided" >&2
+  exit 2
+fi
+
+echo "worker ${CLAIMED_WORKER_ID:-?} handling task: $TASK"
+
+# --- do the real work here -------------------------------------------------
+# e.g. gdalinfo "$TASK"; python process.py --input "$TASK" --output out/
+sleep 1
+# ---------------------------------------------------------------------------
+
+# Demo of the failure path: any task whose name contains FAIL exits non-zero.
+if [[ "$TASK" == *FAIL* ]]; then
+  echo "worker.sh: simulated failure for $TASK" >&2
+  exit 1
+fi
+
+echo "worker done: $TASK"
diff --git a/mkdocs.yml b/mkdocs.yml
index 9f7c1ba5..29716ebf 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -81,6 +81,7 @@ nav:
       - create_operator: c3/create-operator.md
       - create_gridwrapper: c3/create-gridwrapper.md
       - operator_utils: c3/operator-utils.md
+  - Job Coordinator: jobcoordinator.md
   - MLX Backend:
       - Overview: mlx/index.md
       - cos_backend: mlx/cos-backend.md
diff --git a/pyproject.toml b/pyproject.toml
index a4008e5e..2fb90a2d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -96,6 +96,8 @@ test = [
 ]
 utility    = ["geobench"]
 nvidia     = ["pynvml"]
+# Used by iterate2's PostgreSQL coordinator plugin and by the claimed job
+# coordinator (`claimed propagate_jobs` / `claimed work_jobs`) for Postgres URLs.
 postgresql = ["psycopg2-binary>=2.9"]
 amd        = ["pyrsmi"]
 
diff --git a/src/claimed/claimed.py b/src/claimed/claimed.py
index cb75fda3..3b717406 100644
--- a/src/claimed/claimed.py
+++ b/src/claimed/claimed.py
@@ -116,6 +116,11 @@ def main():
         _run_module(sys.argv[2:])
         return
 
+    if len(sys.argv) > 1 and sys.argv[1] in ('propagate_jobs', 'work_jobs'):
+        from claimed.jobcoordinator.cli import main as jobcoordinator_main
+        jobcoordinator_main(sys.argv[1], sys.argv[2:])
+        return
+
     dir_path = os.path.dirname(os.path.realpath(__file__))
     return subprocess.call(
         f'{dir_path}/scripts/claimed ' + ' '.join(sys.argv[1:]), shell=True
diff --git a/src/claimed/jobcoordinator/__init__.py b/src/claimed/jobcoordinator/__init__.py
new file mode 100644
index 00000000..6c6b5136
--- /dev/null
+++ b/src/claimed/jobcoordinator/__init__.py
@@ -0,0 +1,13 @@
+"""Database-backed job coordinator for CLAIMED.
+
+A **job** is a whole batch/run; a **task** is one unit of work within it
+(typically a file path).  Populate a job with tasks from a glob/directory of
+files (``propagate_jobs``) and let many worker sessions pull the job's tasks
+one-by-one (``work_jobs``), invoking any shell script as the worker.  The
+database is the single source of truth and provides atomic claiming via row
+locks, so workers can run across many nodes without a shared filesystem.
+
+See :mod:`claimed.jobcoordinator.db` for the storage/claim logic and
+:mod:`claimed.jobcoordinator.cli` for the command-line entry points wired into
+``claimed propagate_jobs`` / ``claimed work_jobs``.
+"""
diff --git a/src/claimed/jobcoordinator/cli.py b/src/claimed/jobcoordinator/cli.py
new file mode 100644
index 00000000..e0179e58
--- /dev/null
+++ b/src/claimed/jobcoordinator/cli.py
@@ -0,0 +1,196 @@
+"""Command-line entry points for the CLAIMED database-backed job coordinator.
+
+A **job** (``--job``) is a whole batch/run; a **task** is one unit of work
+within it (typically a file path).  Two verbs, dispatched from
+``claimed.claimed.main``:
+
+``claimed propagate_jobs --db <url> --job <name> <pattern>``
+    Expand *pattern* (a glob or directory) and insert each matching path as a
+    pending task of the job.  Idempotent -- re-running skips tasks that already
+    exist.
+
+``claimed work_jobs --db <url> --job <name> --worker <script> [options]``
+    Pull the job's pending tasks one-by-one, running *script* per task.  The
+    task name (the file path) is passed to the worker both as the first
+    positional argument (``$1``) and as the ``CLAIMED_TASK`` environment
+    variable.  Exit code 0 marks the task ``succeeded``; anything else marks it
+    ``failed``.
+
+Any executable can act as a worker -- see ``examples/jobcoordinator_example/``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import glob
+import logging
+import os
+import socket
+import subprocess
+import sys
+import time
+
+from claimed.jobcoordinator import db
+
+logger = logging.getLogger("claimed.jobcoordinator.cli")
+
+
+# ---------------------------------------------------------------------------
+# propagate_jobs
+# ---------------------------------------------------------------------------
+
+def _expand_pattern(pattern: str) -> list:
+    """Return the list of task names for *pattern* (glob or directory)."""
+    if os.path.isdir(pattern):
+        return sorted(
+            os.path.join(pattern, name) for name in os.listdir(pattern)
+        )
+    # recursive=True enables the ** wildcard, e.g. dir/**/*.tif
+    return sorted(glob.glob(pattern, recursive=True))
+
+
+def propagate_jobs(argv) -> int:
+    parser = argparse.ArgumentParser(
+        prog="claimed propagate_jobs",
+        description="Create pending tasks for a job in the coordinator database "
+                    "from a glob or directory.",
+    )
+    parser.add_argument("--db", required=True, help="Database URL or SQLite file path.")
+    parser.add_argument(
+        "--job", required=True,
+        help="Job name. Namespaces this batch of tasks so multiple independent "
+             "jobs can share one database. Workers only process their own job.",
+    )
+    parser.add_argument(
+        "pattern",
+        help="Glob pattern (e.g. 'data/**/*.tif') or a directory. Quote globs so the "
+             "shell does not expand them.",
+    )
+    args = parser.parse_args(argv)
+
+    names = _expand_pattern(args.pattern)
+    if not names:
+        print(f"No files matched pattern: {args.pattern!r}", file=sys.stderr)
+        return 1
+
+    engine = db.get_engine(args.db)
+    inserted = db.insert_tasks(engine, args.job, names)
+    already = len(names) - inserted
+    pending = db.count_pending(engine, args.job)
+    print(
+        f"Inserted {inserted} tasks into job '{args.job}' "
+        f"({already} already existed). Total pending: {pending}"
+    )
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# work_jobs
+# ---------------------------------------------------------------------------
+
+def _run_worker(worker_script: str, task_name: str, job: str,
+                worker_id: str) -> subprocess.CompletedProcess:
+    """Invoke the worker with the task name as $1 and via CLAIMED_TASK."""
+    env = dict(os.environ)
+    env["CLAIMED_TASK"] = task_name
+    env["CLAIMED_JOB"] = job
+    env["CLAIMED_WORKER_ID"] = worker_id
+    return subprocess.run([worker_script, task_name], env=env)
+
+
+def work_jobs(argv) -> int:
+    parser = argparse.ArgumentParser(
+        prog="claimed work_jobs",
+        description="Pull a job's pending tasks one-by-one and run a worker script "
+                    "for each.",
+    )
+    parser.add_argument("--db", required=True, help="Database URL or SQLite file path.")
+    parser.add_argument(
+        "--job", required=True,
+        help="Job to process. Must match the job used in propagate_jobs. "
+             "Workers only claim tasks from this job.",
+    )
+    parser.add_argument(
+        "--worker", required=True,
+        help="Path to the worker script/executable. Receives the task name as $1 "
+             "and as $CLAIMED_TASK.",
+    )
+    parser.add_argument(
+        "--worker-id", default=None,
+        help="Identifier recorded on claimed tasks. Default: <hostname>-<pid>.",
+    )
+    parser.add_argument(
+        "--max-tasks", type=int, default=None,
+        help="Stop after processing this many tasks (default: drain the queue).",
+    )
+    parser.add_argument(
+        "--poll-interval", type=float, default=0.0,
+        help="When the queue is empty, wait this many seconds and retry instead of "
+             "exiting. 0 (default) exits as soon as the queue is empty.",
+    )
+    args = parser.parse_args(argv)
+
+    worker_id = args.worker_id or f"{socket.gethostname()}-{os.getpid()}"
+    engine = db.get_engine(args.db)
+
+    processed = 0
+    while args.max_tasks is None or processed < args.max_tasks:
+        task_name = db.claim_next(engine, args.job, worker_id)
+        if task_name is None:
+            if args.poll_interval > 0:
+                logger.info("Queue empty; sleeping %.1fs before retry.", args.poll_interval)
+                time.sleep(args.poll_interval)
+                continue
+            break
+
+        print(f"[{worker_id}] processing: {task_name}")
+        try:
+            result = _run_worker(args.worker, task_name, args.job, worker_id)
+        except Exception as exc:  # worker could not be launched at all
+            db.mark(engine, args.job, task_name, db.FAILED, error=str(exc))
+            print(f"[{worker_id}] FAILED (launch error): {task_name}: {exc}", file=sys.stderr)
+        else:
+            if result.returncode == 0:
+                db.mark(engine, args.job, task_name, db.SUCCEEDED)
+                print(f"[{worker_id}] succeeded: {task_name}")
+            else:
+                db.mark(
+                    engine, args.job, task_name, db.FAILED,
+                    error=f"worker exited with code {result.returncode}",
+                )
+                print(
+                    f"[{worker_id}] FAILED (exit {result.returncode}): {task_name}",
+                    file=sys.stderr,
+                )
+        processed += 1
+
+    summary = db.stats(engine, args.job)
+    summary_str = ", ".join(f"{k}={v}" for k, v in sorted(summary.items())) or "(empty)"
+    print(
+        f"[{worker_id}] done. Processed {processed} task(s) this run. "
+        f"Job '{args.job}': {summary_str}"
+    )
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+_VERBS = {
+    "propagate_jobs": propagate_jobs,
+    "work_jobs": work_jobs,
+}
+
+
+def main(verb: str, argv) -> None:
+    """Entry point called from ``claimed.claimed.main`` for the two job verbs."""
+    logging.basicConfig(
+        level=os.environ.get("CLAIMED_LOG_LEVEL", "INFO"),
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    handler = _VERBS.get(verb)
+    if handler is None:  # pragma: no cover - guarded by caller
+        print(f"Unknown job coordinator verb: {verb}", file=sys.stderr)
+        sys.exit(2)
+    sys.exit(handler(argv))
diff --git a/src/claimed/jobcoordinator/db.py b/src/claimed/jobcoordinator/db.py
new file mode 100644
index 00000000..43632cb3
--- /dev/null
+++ b/src/claimed/jobcoordinator/db.py
@@ -0,0 +1,295 @@
+"""Storage engine, schema and atomic claim logic for the CLAIMED job coordinator.
+
+A single ``jobs`` table holds the work queue.  A **job** is a whole batch/run
+(the namespace); a **task** is one unit of work within it (typically a file
+path).  Each row is one task and moves through the lifecycle::
+
+    pending -> processing -> succeeded
+                          \\-> failed
+
+Two backends are supported through one URL-based abstraction (SQLAlchemy Core):
+
+* **PostgreSQL** -- real multi-node coordination.  Atomic claiming uses
+  ``SELECT ... FOR UPDATE SKIP LOCKED`` so N workers never grab the same job.
+* **SQLite** -- local / testing.  ``SELECT ... FOR UPDATE`` does not exist, so
+  claiming serializes writers with ``BEGIN IMMEDIATE`` plus a guarded
+  ``UPDATE ... WHERE status='pending'`` and a small retry loop.
+
+The URL-normalisation helpers mirror the iterate2 coordinator plugins
+(``terratorch_iterate/iterate2/plugin/coordinator/{sqlite,postgresql}.py``) so the
+accepted ``--db`` syntax is consistent across CLAIMED.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from datetime import datetime, timezone
+
+from sqlalchemy import (
+    Column,
+    Integer,
+    MetaData,
+    String,
+    Table,
+    Text,
+    UniqueConstraint,
+    create_engine,
+    event,
+    func,
+    select,
+    text,
+    update,
+)
+from sqlalchemy.engine import Engine
+
+logger = logging.getLogger("claimed.jobcoordinator")
+
+# Job lifecycle states.
+PENDING = "pending"
+PROCESSING = "processing"
+SUCCEEDED = "succeeded"
+FAILED = "failed"
+
+_SQLITE_BUSY_TIMEOUT = 30  # seconds
+
+_metadata = MetaData()
+
+JOBS = Table(
+    "claimed_jobs",
+    _metadata,
+    Column("id", Integer, primary_key=True, autoincrement=True),
+    # job namespaces independent runs so many parallel jobs can share one
+    # database. Uniqueness is per (job, task_name), not task_name alone.
+    Column("job", Text, nullable=False),
+    Column("task_name", Text, nullable=False),
+    Column("status", String(16), nullable=False, default=PENDING),
+    Column("worker_id", Text, nullable=True),
+    Column("attempts", Integer, nullable=False, default=0),
+    Column("created_at", Text, nullable=True),
+    Column("updated_at", Text, nullable=True),
+    Column("error", Text, nullable=True),
+    UniqueConstraint("job", "task_name", name="uq_claimed_jobs_job_task"),
+)
+
+
+# ---------------------------------------------------------------------------
+# URL handling
+# ---------------------------------------------------------------------------
+
+def _redact(url: str) -> str:
+    """Replace the password in a DB URL with '***' for safe logging."""
+    return re.sub(r"(://[^:/]+:)[^@]+(@)", r"\1***\2", url)
+
+
+def normalize_url(db_url: str) -> str:
+    """Turn a user-supplied ``--db`` value into a SQLAlchemy URL.
+
+    * ``postgres://`` is rewritten to ``postgresql://`` (SQLAlchemy 1.4+ dropped
+      the short form).
+    * An explicit scheme (``postgresql://``, ``sqlite:///``) is passed through.
+    * Anything else is treated as a filesystem path to a SQLite database.
+    """
+    if db_url.startswith("postgres://") and not db_url.startswith("postgresql://"):
+        db_url = "postgresql://" + db_url[len("postgres://"):]
+    if db_url.startswith("postgresql://") or db_url.startswith("sqlite:///"):
+        return db_url
+    # Bare path (foo.db, /abs/path, ./rel/path.sqlite, ...) -> SQLite.
+    return f"sqlite:///{db_url}"
+
+
+def is_sqlite(engine: Engine) -> bool:
+    return engine.dialect.name == "sqlite"
+
+
+def get_engine(db_url: str) -> Engine:
+    """Create a SQLAlchemy engine and ensure the ``claimed_jobs`` table exists."""
+    url = normalize_url(db_url)
+    logger.info("Job coordinator database: %s", _redact(url))
+
+    if url.startswith("sqlite:///"):
+        # future=True keeps 2.0 semantics; timeout lets SQLite wait on the busy
+        # write lock instead of raising immediately under contention.
+        engine = create_engine(
+            url,
+            future=True,
+            connect_args={"timeout": _SQLITE_BUSY_TIMEOUT},
+        )
+
+        @event.listens_for(engine, "connect")
+        def _set_sqlite_pragma(dbapi_conn, _record):  # pragma: no cover - trivial
+            cur = dbapi_conn.cursor()
+            # WAL improves read/write concurrency between worker processes.
+            cur.execute("PRAGMA journal_mode=WAL")
+            cur.execute(f"PRAGMA busy_timeout={_SQLITE_BUSY_TIMEOUT * 1000}")
+            cur.close()
+    else:
+        if url.startswith("postgresql://"):
+            _require_psycopg2()
+        engine = create_engine(url, future=True, pool_pre_ping=True)
+
+    _metadata.create_all(engine, checkfirst=True)
+    return engine
+
+
+def _require_psycopg2() -> None:
+    try:
+        import psycopg2  # noqa: F401
+    except ImportError:
+        raise ImportError(
+            "psycopg2 is not installed but is required for PostgreSQL support.\n\n"
+            "Install options:\n"
+            "  pip install psycopg2-binary        # pre-built wheel, no compiler\n"
+            "  pip install 'claimed[postgresql]'  # via the project extra\n"
+        ) from None
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# Queue operations
+# ---------------------------------------------------------------------------
+
+def insert_tasks(engine: Engine, job: str, names) -> int:
+    """Insert *names* as pending tasks of *job*, ignoring duplicates.
+
+    Uniqueness is per ``(job, task_name)``, so the same file path may exist
+    independently in different jobs.  Returns the number of rows inserted.
+    """
+    names = list(dict.fromkeys(n for n in names if n))  # de-dupe, keep order
+    if not names:
+        return 0
+
+    now = _now()
+    rows = [{"job": job, "task_name": n, "status": PENDING,
+             "attempts": 0, "created_at": now, "updated_at": now} for n in names]
+
+    before = count_all(engine, job)
+    with engine.begin() as conn:
+        # ON CONFLICT DO NOTHING is supported by both PostgreSQL and SQLite
+        # (>= 3.24).  Use the dialect-specific insert to express it.
+        if is_sqlite(engine):
+            from sqlalchemy.dialects.sqlite import insert as _insert
+        else:
+            from sqlalchemy.dialects.postgresql import insert as _insert
+        stmt = _insert(JOBS).values(rows)
+        stmt = stmt.on_conflict_do_nothing(
+            index_elements=[JOBS.c.job, JOBS.c.task_name]
+        )
+        conn.execute(stmt)
+    after = count_all(engine, job)
+    return after - before
+
+
+def claim_next(engine: Engine, job: str, worker_id: str):
+    """Atomically claim the oldest pending task of *job* for *worker_id*.
+
+    Returns the claimed ``task_name`` or ``None`` when the job's queue is empty.
+    Claiming is scoped to *job*, so parallel jobs in the same database never
+    contend for or steal each other's tasks.
+    """
+    if is_sqlite(engine):
+        return _claim_next_sqlite(engine, job, worker_id)
+    return _claim_next_postgres(engine, job, worker_id)
+
+
+def _claim_next_postgres(engine: Engine, job: str, worker_id: str):
+    stmt = text(
+        """
+        UPDATE claimed_jobs
+           SET status='processing',
+               worker_id=:w,
+               attempts=attempts + 1,
+               updated_at=:ts
+         WHERE id = (
+               SELECT id FROM claimed_jobs
+                WHERE status='pending' AND job=:j
+                ORDER BY id
+                   FOR UPDATE SKIP LOCKED
+                LIMIT 1
+         )
+        RETURNING task_name
+        """
+    )
+    with engine.begin() as conn:
+        row = conn.execute(
+            stmt, {"w": worker_id, "ts": _now(), "j": job}
+        ).fetchone()
+    return row[0] if row else None
+
+
+def _claim_next_sqlite(engine: Engine, job: str, worker_id: str):
+    # SKIP LOCKED is unavailable; serialize writers with BEGIN IMMEDIATE and a
+    # guarded UPDATE.  Retry only on a lost race (0 rows updated) where another
+    # writer grabbed the row we just selected.
+    for _ in range(50):
+        with engine.begin() as conn:
+            conn.execute(text("BEGIN IMMEDIATE"))
+            row = conn.execute(
+                select(JOBS.c.id, JOBS.c.task_name, JOBS.c.attempts)
+                .where(JOBS.c.status == PENDING, JOBS.c.job == job)
+                .order_by(JOBS.c.id)
+                .limit(1)
+            ).fetchone()
+            if row is None:
+                return None
+            task_id, task_name, attempts = row
+            result = conn.execute(
+                update(JOBS)
+                .where(JOBS.c.id == task_id, JOBS.c.status == PENDING)
+                .values(
+                    status=PROCESSING,
+                    worker_id=worker_id,
+                    attempts=attempts + 1,
+                    updated_at=_now(),
+                )
+            )
+            if result.rowcount == 1:
+                return task_name
+        # Lost the race for this row; try again.
+    logger.warning("claim_next gave up after repeated lost races (SQLite)")
+    return None
+
+
+def mark(engine: Engine, job: str, task_name: str, status: str,
+         error: str | None = None) -> None:
+    """Record a terminal *status* for *task_name* within *job*."""
+    with engine.begin() as conn:
+        conn.execute(
+            update(JOBS)
+            .where(JOBS.c.job == job, JOBS.c.task_name == task_name)
+            .values(status=status, error=error, updated_at=_now())
+        )
+
+
+def count_all(engine: Engine, job: str) -> int:
+    with engine.connect() as conn:
+        return int(
+            conn.execute(
+                select(func.count()).select_from(JOBS)
+                .where(JOBS.c.job == job)
+            ).scalar_one()
+        )
+
+
+def count_pending(engine: Engine, job: str) -> int:
+    with engine.connect() as conn:
+        return int(
+            conn.execute(
+                select(func.count()).select_from(JOBS)
+                .where(JOBS.c.status == PENDING, JOBS.c.job == job)
+            ).scalar_one()
+        )
+
+
+def stats(engine: Engine, job: str) -> dict:
+    """Return a ``{status: count}`` mapping for *job*."""
+    with engine.connect() as conn:
+        rows = conn.execute(
+            select(JOBS.c.status, func.count())
+            .where(JOBS.c.job == job)
+            .group_by(JOBS.c.status)
+        ).fetchall()
+    return {status: int(n) for status, n in rows}
diff --git a/tests/integration/test_jobcoordinator.py b/tests/integration/test_jobcoordinator.py
new file mode 100644
index 00000000..54b64c69
--- /dev/null
+++ b/tests/integration/test_jobcoordinator.py
@@ -0,0 +1,168 @@
+"""
+Integration tests for the database-backed CLAIMED job coordinator.
+
+SQLite      – always runs (uses a temp file).
+PostgreSQL  – skipped unless POSTGRES_URL is set, e.g.:
+
+    export POSTGRES_URL="postgresql://user:password@localhost:5432/claimed_test"
+    pytest tests/integration/test_jobcoordinator.py -v
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import threading
+
+import pytest
+
+from claimed.jobcoordinator import db
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def sqlite_engine():
+    with tempfile.TemporaryDirectory() as tmp:
+        engine = db.get_engine(os.path.join(tmp, "jobs.db"))
+        yield engine
+        engine.dispose()
+
+
+# ---------------------------------------------------------------------------
+# URL normalisation
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize(
+    "given,expected",
+    [
+        ("jobs.db", "sqlite:///jobs.db"),
+        ("/tmp/x.sqlite", "sqlite:////tmp/x.sqlite"),
+        ("sqlite:///already.db", "sqlite:///already.db"),
+        ("postgres://u:p@h/d", "postgresql://u:p@h/d"),
+        ("postgresql://u:p@h/d", "postgresql://u:p@h/d"),
+    ],
+)
+def test_normalize_url(given, expected):
+    assert db.normalize_url(given) == expected
+
+
+# ---------------------------------------------------------------------------
+# Lifecycle
+# ---------------------------------------------------------------------------
+
+JOB = "job1"
+
+
+def test_propagate_and_lifecycle(sqlite_engine):
+    engine = sqlite_engine
+    inserted = db.insert_tasks(engine, JOB, ["a.tif", "b.tif", "c.tif"])
+    assert inserted == 3
+    assert db.count_pending(engine, JOB) == 3
+
+    # Claim, then mark terminal states.
+    n1 = db.claim_next(engine, JOB, "worker-1")
+    n2 = db.claim_next(engine, JOB, "worker-1")
+    assert {n1, n2} == {"a.tif", "b.tif"}  # oldest first
+    assert db.count_pending(engine, JOB) == 1
+
+    db.mark(engine, JOB, n1, db.SUCCEEDED)
+    db.mark(engine, JOB, n2, db.FAILED, error="boom")
+
+    n3 = db.claim_next(engine, JOB, "worker-1")
+    assert n3 == "c.tif"
+    db.mark(engine, JOB, n3, db.SUCCEEDED)
+
+    assert db.claim_next(engine, JOB, "worker-1") is None  # queue drained
+    assert db.stats(engine, JOB) == {db.SUCCEEDED: 2, db.FAILED: 1}
+
+
+def test_propagate_is_idempotent(sqlite_engine):
+    engine = sqlite_engine
+    assert db.insert_tasks(engine, JOB, ["a", "b", "c"]) == 3
+    # Re-propagating the same names inserts nothing.
+    assert db.insert_tasks(engine, JOB, ["a", "b", "c"]) == 0
+    # A partial overlap inserts only the new ones.
+    assert db.insert_tasks(engine, JOB, ["c", "d", "e"]) == 2
+    assert db.count_pending(engine, JOB) == 5
+
+
+def test_jobs_are_isolated(sqlite_engine):
+    """The same task name lives independently in different jobs, and claiming
+    in one job never consumes another job's tasks."""
+    engine = sqlite_engine
+    assert db.insert_tasks(engine, "alpha", ["shared", "a-only"]) == 2
+    assert db.insert_tasks(engine, "beta", ["shared", "b-only"]) == 2
+
+    # Drain alpha entirely.
+    alpha_claimed = set()
+    while (t := db.claim_next(engine, "alpha", "w")) is not None:
+        alpha_claimed.add(t)
+    assert alpha_claimed == {"shared", "a-only"}
+
+    # beta is untouched.
+    assert db.count_pending(engine, "beta") == 2
+    beta_claimed = set()
+    while (t := db.claim_next(engine, "beta", "w")) is not None:
+        beta_claimed.add(t)
+    assert beta_claimed == {"shared", "b-only"}
+
+
+def test_concurrent_claims_are_disjoint(sqlite_engine):
+    """Two threads draining one queue must claim disjoint task sets."""
+    engine = sqlite_engine
+    names = [f"task-{i:03d}" for i in range(60)]
+    db.insert_tasks(engine, JOB, names)
+
+    claimed: dict[str, list] = {"w1": [], "w2": []}
+
+    def drain(worker_id):
+        while True:
+            task = db.claim_next(engine, JOB, worker_id)
+            if task is None:
+                break
+            claimed[worker_id].append(task)
+
+    t1 = threading.Thread(target=drain, args=("w1",))
+    t2 = threading.Thread(target=drain, args=("w2",))
+    t1.start()
+    t2.start()
+    t1.join()
+    t2.join()
+
+    all_claimed = claimed["w1"] + claimed["w2"]
+    assert sorted(all_claimed) == sorted(names)   # every job claimed once
+    assert len(all_claimed) == len(set(all_claimed))  # no duplicates
+
+
+# ---------------------------------------------------------------------------
+# PostgreSQL (optional)
+# ---------------------------------------------------------------------------
+
+@pytest.mark.skipif(
+    not os.environ.get("POSTGRES_URL"),
+    reason="POSTGRES_URL not set; skipping PostgreSQL job coordinator test",
+)
+def test_postgres_lifecycle():
+    url = os.environ["POSTGRES_URL"]
+    engine = db.get_engine(url)
+    try:
+        # Isolate from any prior run.
+        with engine.begin() as conn:
+            from sqlalchemy import text
+            conn.execute(text("DELETE FROM claimed_jobs"))
+
+        assert db.insert_tasks(engine, "pg", ["p1", "p2", "p3"]) == 3
+        got = {db.claim_next(engine, "pg", "pw") for _ in range(3)}
+        assert got == {"p1", "p2", "p3"}
+        assert db.claim_next(engine, "pg", "pw") is None
+        for name in ["p1", "p2", "p3"]:
+            db.mark(engine, "pg", name, db.SUCCEEDED)
+        assert db.stats(engine, "pg") == {db.SUCCEEDED: 3}
+    finally:
+        with engine.begin() as conn:
+            from sqlalchemy import text
+            conn.execute(text("DELETE FROM claimed_jobs"))
+        engine.dispose()