diff --git a/coordinator/.dockerignore b/coordinator/.dockerignore
new file mode 100644
index 0000000..a4d0d5c
--- /dev/null
+++ b/coordinator/.dockerignore
@@ -0,0 +1,14 @@
+# Keep the build context small and never bake secrets or local state into an image.
+.env
+.git
+.gitignore
+*.md
+Makefile
+docker-compose.yml
+Dockerfile
+.dockerignore
+
+# Local build artifacts
+/coordinator
+/bin/
+*.out
diff --git a/coordinator/.env.example b/coordinator/.env.example
new file mode 100644
index 0000000..2887295
--- /dev/null
+++ b/coordinator/.env.example
@@ -0,0 +1,14 @@
+# Copy to .env and adjust. All settings are read from the environment.
+
+COORDINATOR_ADDR=:8080
+DATABASE_URL=postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable
+
+# Shared bearer token every worker must present. Leave empty to disable auth (dev only).
+WORKER_AUTH_TOKEN=change-me
+
+# Optional tuning (defaults shown).
+DB_MAX_CONNS=10
+REQUEST_TIMEOUT=15s
+LEASE_DURATION=2m
+DEFAULT_MAX_ATTEMPTS=3
+REAPER_INTERVAL=30s
diff --git a/coordinator/.gitignore b/coordinator/.gitignore
new file mode 100644
index 0000000..c44a5e4
--- /dev/null
+++ b/coordinator/.gitignore
@@ -0,0 +1,4 @@
+/coordinator
+/bin/
+.env
+*.out
diff --git a/coordinator/.golangci.yml b/coordinator/.golangci.yml
new file mode 100644
index 0000000..428bfed
--- /dev/null
+++ b/coordinator/.golangci.yml
@@ -0,0 +1,54 @@
+version: "2"
+
+run:
+  timeout: 3m
+
+linters:
+  # "standard" = errcheck, govet, ineffassign, staticcheck, unused.
+  default: standard
+  enable:
+    # Catches `err == ErrFoo` where errors.Is is required. Directly relevant
+    # here: domain exposes sentinel errors that use cases may wrap with %w.
+    - errorlint
+    # Returning nil after checking a non-nil error — a silent bug factory.
+    - nilerr
+    # http.Get/Do without a context: every outbound call must be cancellable.
+    - noctx
+    # Unclosed response bodies leak connections.
+    - bodyclose
+    # Common security mistakes (weak crypto, unhandled file perms).
+    - gosec
+    # Style and naming consistency.
+    - revive
+    - misspell
+    - unconvert
+
+  settings:
+    errcheck:
+      # Deferred Close/Rollback are intentionally ignored in a few places
+      # (rollback after commit is a documented no-op).
+      check-type-assertions: true
+    revive:
+      rules:
+        - name: exported
+          disabled: true # internal packages need no exported-symbol comments
+    gosec:
+      excludes:
+        - G404 # math/rand is fine for jitter; nothing here is security-sensitive
+
+  exclusions:
+    rules:
+      # Tests may skip error checks and use long literals freely.
+      - path: _test\.go
+        linters:
+          - errcheck
+          - gosec
+
+formatters:
+  enable:
+    - gofmt
+    - goimports
+  settings:
+    goimports:
+      local-prefixes:
+        - github.com/emil28092005/SciMesh/coordinator
diff --git a/coordinator/ARCHITECTURE.md b/coordinator/ARCHITECTURE.md
new file mode 100644
index 0000000..77b6976
--- /dev/null
+++ b/coordinator/ARCHITECTURE.md
@@ -0,0 +1,144 @@
+# Архитектура координатора
+
+Карта кода. Читать сверху вниз: сначала «где что лежит», потом «как проходит
+запрос», в конце — «куда добавлять новое».
+
+---
+
+## 1. Четыре слоя
+
+```
+   infra       конфиг, пул БД, часы, HTTP-сервер, reaper    ← драйверы
+   transport   HTTP-хендлеры          ← входящее: кто зовёт нас
+   storage     репозитории на SQL     ← исходящее: кого зовём мы
+   usecase     операции + ПОРТЫ (интерфейсы)                ← прикладные правила
+   domain      Task, Job и их инварианты                    ← бизнес-правила
+
+                       ┌── transport ──┐
+   domain ◄── usecase ◄┤               ├◄── infra
+                       └── storage ────┘
+```
+
+`transport` и `storage` — один и тот же слой (в книгах он зовётся «адаптеры»),
+просто разделённый по направлению: транспорт принимает запросы снаружи, storage
+обращается наружу сам. Так путь к файлу говорит о его роли, а не о категории.
+
+**Единственное правило:** зависимости идут только внутрь. `domain` не импортирует
+ничего из проекта. `usecase` видит только `domain`. `transport` и `storage` не
+знают друг о друге.
+
+Проверить в любой момент:
+
+```sh
+go list -f '{{range .Imports}}{{.}}{{"\n"}}{{end}}' ./internal/domain | grep internal
+# пусто = правило соблюдено
+```
+
+---
+
+## 2. Где что лежит
+
+| Файл | Что внутри | Строк |
+| --- | --- | --- |
+| `domain/task.go` | `Task` и **все** переходы состояний: аренда, завершение, провал, истечение | ~245 |
+| `domain/job.go` | `Job`, разбиение на чанки, вывод статуса из счётчиков задач | ~107 |
+| `domain/errors.go` | Нарушения бизнес-правил (`ErrLeaseConflict`, `ErrStaleAttempt`, …) | ~18 |
+| `usecase/ports.go` | **Порты**: `TaskRepository`, `JobRepository`, `TxManager`, `Clock` | ~79 |
+| `usecase/task.go` | Операции над задачей: claim, renew, complete, fail, expire | ~200 |
+| `usecase/job.go` | Операции над job: create, status, results, stitch | ~180 |
+| `usecase/dto.go` | Входные структуры юзкейсов | ~51 |
+| `transport/http/server.go` | Роутер и сборка middleware | ~60 |
+| `transport/http/handlers.go` | По хендлеру на эндпоинт | ~180 |
+| `transport/http/dto.go` | JSON-форматы запросов и ответов | ~118 |
+| `transport/http/middleware.go` | request-ID, access-лог, bearer-авторизация | ~103 |
+| `transport/http/errors.go` | Маппинг доменных ошибок в HTTP-коды | ~55 |
+| `storage/postgres/task_repo.go` | SQL по задачам, включая атомарный claim | ~109 |
+| `storage/postgres/job_repo.go` | SQL по job'ам | ~39 |
+| `storage/postgres/tx.go` | `TxManager`: транзакция через контекст | ~65 |
+| `infra/*.go` | Конфиг, пул, часы, сервер, reaper | ~240 |
+| `cmd/coordinator/main.go` | **Composition root** — единственное место со всеми конкретными типами | ~73 |
+
+---
+
+## 3. Трасса запроса: `POST /tasks/claim`
+
+Как воркер получает задачу. Четыре остановки, по одной на слой:
+
+```
+  ①  transport/http/handlers.go → handleClaim
+        разбирает JSON, отдаёт usecase.ClaimTaskInput
+        │
+        ▼
+  ②  usecase/task.go → ClaimTask.Execute
+        сначала подчищает протухшие аренды, потом просит одну задачу
+        через ПОРТ TaskRepository (реализацию не знает)
+        │
+        ▼
+  ③  usecase/ports.go → TaskRepository.ClaimNext
+        контракт: «атомарно выдай одну задачу»
+        │
+        ▼
+  ④  storage/postgres/task_repo.go → claimNextSQL
+        SELECT ... FOR UPDATE SKIP LOCKED + UPDATE одним запросом
+```
+
+Обратно поднимается `*domain.Task`, юзкейс сужает его до `domain.ClaimedTask`
+(воркеру не отдаём `version`, `max_attempts` и чужие ошибки), хендлер
+превращает в JSON. Пустая очередь — это `nil, nil` на шаге ② и `204` на ①.
+
+**Трасса `POST /tasks/{id}/result`** такая же, но с одним отличием: решение
+принимает **сущность**, а не юзкейс.
+
+```
+  handlers.go → CompleteTask.Execute → tx.WithinTx(
+        GetForUpdate  →  task.CompleteWith(...)  ←── ЗДЕСЬ правила
+                              │                      (чужая аренда? устаревший
+        Update  ←─────────────┘                       attempt? повтор того же
+        syncJobStatus                                 манифеста?)
+  )
+```
+
+---
+
+## 4. Куда добавлять новое
+
+| Хочу… | Правлю |
+| --- | --- |
+| новое бизнес-правило (когда задачу можно повторить) | `domain/task.go` + тест рядом |
+| новую операцию (отменить job) | `usecase/job.go` + порт в `ports.go`, если нужен новый запрос к БД |
+| новый HTTP-эндпоинт | `transport/http/handlers.go` + маршрут в `server.go` + DTO в `dto.go` |
+| новый SQL-запрос | `storage/postgres/*_repo.go` |
+| новую настройку | `infra/config.go` + `.env.example` |
+| поменять код ответа на ошибку | `transport/http/errors.go` |
+
+**Правило при сомнении:** если код можно описать фразой «когда X, то Y» без
+упоминания HTTP, SQL и конфигов — это `domain`. Если он оркеструет несколько
+шагов и транзакцию — `usecase`. Если знает про JSON — `transport`, про SQL — `storage`.
+
+---
+
+## 5. Три вещи, которые надо понять один раз
+
+**Порты объявляет потребитель.** `TaskRepository` описан в `usecase/ports.go`, а
+реализован в `storage/postgres`. Поэтому `usecase` не импортирует `storage` —
+стрелка зависимости смотрит внутрь, хотя вызов на рантайме идёт наружу.
+
+**Транзакция едет в контексте.** `TxManager.WithinTx` кладёт `pgx.Tx` в контекст
+по неэкспортируемому ключу; репозитории достают её через `conn(ctx, pool)`.
+Благодаря этому юзкейс говорит «сделай это атомарно», ни разу не упомянув pgx.
+
+**Атомарный claim нельзя разложить на шаги.** `ClaimNext` — один SQL-запрос,
+потому что `SELECT` + отдельный `UPDATE` вернул бы гонку, при которой одну
+задачу выдают двум воркерам. Поэтому `ClaimTask.Execute` выглядит тонким: там
+нечего оркестровать, вся гарантия — внутри запроса.
+
+---
+
+## 6. Что уже работает, а что заглушка
+
+Работает: слои и проводка, роутинг, авторизация, access-лог, маппинг ошибок,
+транзакции, graceful shutdown, миграции, **весь domain с 12 юнит-тестами без БД**.
+
+Заглушки (`ErrNotImplemented` → HTTP 501): методы репозиториев. SQL для двух
+главных операций уже написан в `task_repo.go` — `claimNextSQL` и
+`expireLeasesSQL`, осталось их подключить.
diff --git a/coordinator/Dockerfile b/coordinator/Dockerfile
new file mode 100644
index 0000000..bf0b1dc
--- /dev/null
+++ b/coordinator/Dockerfile
@@ -0,0 +1,47 @@
+# syntax=docker/dockerfile:1
+#
+# Requires BuildKit (the RUN --mount cache lines below). Docker 23+ enables it
+# by default when the buildx plugin is present; install `docker-buildx` if a
+# build fails with "the --mount option requires BuildKit".
+
+# --- build stage ----------------------------------------------------------
+FROM golang:1.24-alpine AS build
+
+WORKDIR /src
+
+# Copy manifests first: this layer stays cached until dependencies actually
+# change, so editing Go sources does not re-download the module graph.
+COPY go.mod go.sum ./
+RUN --mount=type=cache,target=/go/pkg/mod go mod download
+
+COPY . .
+
+# The cache mounts persist the module cache and the compiler's build cache
+# *across* builds, so a rebuild after a code edit recompiles only what changed
+# instead of the whole dependency tree.
+#
+# CGO_ENABLED=0 produces a fully static binary, so the runtime image needs no
+# libc. -trimpath strips local paths; -s -w drop the symbol table and DWARF.
+RUN --mount=type=cache,target=/go/pkg/mod \
+    --mount=type=cache,target=/root/.cache/go-build \
+    CGO_ENABLED=0 GOOS=linux go build \
+        -trimpath -ldflags="-s -w" \
+        -o /out/coordinator ./cmd/coordinator
+
+# --- runtime stage --------------------------------------------------------
+FROM alpine:3.20
+
+# ca-certificates for outbound TLS; wget backs the container healthcheck.
+RUN apk add --no-cache ca-certificates wget \
+ && adduser -D -H -u 10001 coordinator
+
+COPY --from=build /out/coordinator /usr/local/bin/coordinator
+
+# Never run as root: a compromised process should not own the container.
+USER coordinator
+
+EXPOSE 8080
+
+# Exec form, not shell: the binary becomes PID 1 and receives SIGTERM directly,
+# which is what its graceful shutdown depends on.
+ENTRYPOINT ["/usr/local/bin/coordinator"]
diff --git a/coordinator/Makefile b/coordinator/Makefile
new file mode 100644
index 0000000..d85c19e
--- /dev/null
+++ b/coordinator/Makefile
@@ -0,0 +1,60 @@
+.PHONY: build run test vet tidy migrate-up migrate-down up down logs ps rebuild psql
+
+# --- build / run ---------------------------------------------------------
+build:
+	go build ./...
+
+run:
+	go run ./cmd/coordinator
+
+test:
+	go test ./...
+
+vet:
+	go vet ./...
+
+# Runs golangci-lint without installing it system-wide. Install it for speed:
+#   pacman -S golangci-lint   (Arch)
+LINT_VERSION := v2.12.2
+lint:
+	@command -v golangci-lint >/dev/null 2>&1 \
+		&& golangci-lint run ./... \
+		|| go run github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(LINT_VERSION) run ./...
+
+tidy:
+	go mod tidy
+
+# --- migrations ----------------------------------------------------------
+# Requires the golang-migrate CLI:
+#   go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@latest
+# DATABASE_URL must be set, e.g.:
+#   export DATABASE_URL='postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable'
+migrate-up:
+	migrate -path migrations -database "$(DATABASE_URL)" up
+
+migrate-down:
+	migrate -path migrations -database "$(DATABASE_URL)" down 1
+
+# --- docker --------------------------------------------------------------
+# `up` starts Postgres, applies migrations, then launches the coordinator.
+up:
+	docker compose up -d --build
+
+down:
+	docker compose down
+
+# Also drops the database volume — use when the schema is beyond repair.
+down-clean:
+	docker compose down -v
+
+logs:
+	docker compose logs -f coordinator
+
+ps:
+	docker compose ps
+
+rebuild:
+	docker compose up -d --build --force-recreate coordinator
+
+psql:
+	docker compose exec postgres psql -U scimesh -d scimesh
diff --git a/coordinator/README.md b/coordinator/README.md
new file mode 100644
index 0000000..f83a9be
--- /dev/null
+++ b/coordinator/README.md
@@ -0,0 +1,143 @@
+# SciMesh Coordinator
+
+Durable task-queue server for SciMesh, in Go on PostgreSQL. It owns all database
+access; workers talk to it only over HTTP and never receive DB credentials.
+
+Built as a **modular monolith following Clean Architecture** — one binary, four
+layers, dependencies pointing strictly inward. See
+`docs/database-integration-task.md` and `docs/worker-daemon-task.md` in the repo
+root for the full contract.
+
+## Layers
+
+```
+        infra       config, pgxpool, http.Server, clock    ← frameworks & drivers
+        transport   http handlers      ← inbound: who calls us
+        storage     sql repositories   ← outbound: who we call
+        usecase     business operations + PORTS            ← application rules
+        domain      Task, Job + their invariants           ← enterprise rules
+
+                            ┌── transport ──┐
+        domain ◄── usecase ◄┤               ├◄── infra
+                            └── storage ────┘
+```
+
+`transport` and `storage` are one layer — the "interface adapters" ring — split
+by direction rather than by category, so a file's path tells you its role.
+
+The rule that matters: **source dependencies point only inward**. `domain`
+imports nothing from this module; `usecase` sees only `domain`; `transport` and
+`storage` know nothing of each other. Verify it at any time with:
+
+```sh
+go list -f '{{range .Imports}}{{.}}{{"\n"}}{{end}}' ./internal/domain | grep internal   # must be empty
+```
+
+## Layout
+
+```
+coordinator/
+  cmd/coordinator/main.go      # composition root: the only place with concrete types
+  internal/
+    domain/                    # entities + rules, no I/O
+      task.go                    Task, lease/complete/fail/expire transitions
+      job.go                     Job, chunk fan-out, status derivation
+      errors.go                  business-rule violations
+    usecase/                   # one type per operation, dependencies injected
+      ports.go                   TaskRepository, JobRepository, TxManager, Clock
+      dto.go                     use-case boundary inputs
+      task.go                    claim, renew, complete, fail, expire
+      job.go                     create, status, results, stitch
+    transport/http/            # routing, DTOs, middleware, error mapping
+    storage/postgres/          # SQL behind the ports; TxManager via context
+    infra/                     # config.go db.go clock.go server.go
+  migrations/                  # golang-migrate SQL, run as an explicit command
+```
+
+A full map — file-by-file table, a request traced through every layer, and a
+"where do I add X" guide — lives in [ARCHITECTURE.md](ARCHITECTURE.md).
+
+## Quickstart
+
+### With Docker (nothing to install but Docker)
+
+```sh
+make up                       # Postgres → migrations → coordinator
+curl localhost:8080/health
+make logs                     # follow the coordinator
+make down                     # stop (add down-clean to drop the DB volume)
+```
+
+`up` starts three services in order: Postgres waits until `pg_isready` passes, a
+one-shot `migrate` container applies the schema and exits, and only then does the
+coordinator start — so it never queries a database that has no tables.
+
+> **Needs BuildKit.** The Dockerfile uses `RUN --mount=type=cache` to reuse the
+> Go module and compiler caches between builds. If the build fails with
+> *"the --mount option requires BuildKit"*, install the buildx plugin —
+> `pacman -S docker-buildx` on Arch, `apt install docker-buildx-plugin` on Debian.
+
+### Locally, against your own Postgres
+
+```sh
+cp .env.example .env          # then edit DATABASE_URL / WORKER_AUTH_TOKEN
+                              # it is loaded automatically — no export needed
+
+make tidy                     # fetch deps (needs network once)
+make migrate-up               # apply schema (needs the migrate CLI)
+make run                      # start the server
+```
+
+## Configuration
+
+Settings come from the environment. A `.env` file is loaded at startup via
+`godotenv` as a local-dev convenience (override its path with `ENV_FILE`):
+
+- a missing `.env` is not an error — production injects real env vars;
+- **real environment variables always win** over the file, so an orchestrator's
+  values are never shadowed by a stale `.env` baked into an image.
+
+See `.env.example`; only `DATABASE_URL` is required.
+
+## Endpoints
+
+| Method | Path                          | Purpose                                    |
+| ------ | ----------------------------- | ------------------------------------------ |
+| POST   | `/jobs`                       | Create job + pending tasks transactionally |
+| POST   | `/tasks/claim`                | Atomically lease one task (`204` if none)  |
+| POST   | `/tasks/{task_id}/heartbeat`  | Renew the caller's lease                   |
+| POST   | `/tasks/{task_id}/result`     | Record a completed result (idempotent)     |
+| POST   | `/tasks/{task_id}/failure`    | Record failure / retryable state           |
+| GET    | `/jobs/{job_id}`              | Aggregate job progress                     |
+| GET    | `/health`                     | Liveness (unauthenticated)                 |
+
+## Status
+
+Scaffold with a **complete, tested domain**. Layers, wiring, routing, auth,
+access logging, error mapping, transactions, migrations, and graceful shutdown
+are in place. Repository methods are stubs returning `ErrNotImplemented`
+(→ HTTP 501); the SQL for claiming and lease expiry is written and ready to wire.
+
+Roadmap:
+
+1. schema + migrations ✅
+2. `ClaimNext`, `InsertBatch` — atomic claim via `FOR UPDATE SKIP LOCKED`
+3. `GetForUpdate`, `Update`, `CountByStatus` — completes the result/failure paths
+4. file upload / chunk download
+5. `ExpireLeases` — SQL is written, needs wiring
+6. stitcher: merge per-chunk top-k into the final CSV
+7. integration tests against real Postgres via `TEST_DATABASE_URL`
+
+## Tests
+
+`internal/domain` is covered by unit tests that need **no database** — lease
+ownership, stale attempts, idempotent replays, retry budgets, and expiry are all
+pure functions of entity state:
+
+```sh
+go test ./...
+go vet ./...
+```
+
+Integration tests (concurrent claiming, migrations) come in phase 7 and require
+a real PostgreSQL instance supplied through `TEST_DATABASE_URL`.
diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go
new file mode 100644
index 0000000..d2fb7ec
--- /dev/null
+++ b/coordinator/cmd/coordinator/main.go
@@ -0,0 +1,98 @@
+// Command coordinator is the SciMesh task-queue server. It owns all database
+// access; workers reach it only over HTTP and never receive DB credentials.
+// Migrations are a separate explicit command (see Makefile) — this binary never
+// mutates schema at startup.
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/infra"
+	"github.com/emil28092005/SciMesh/coordinator/internal/storage/postgres"
+	httptransport "github.com/emil28092005/SciMesh/coordinator/internal/transport/http"
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+func main() {
+	log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
+
+	// All work happens in run() so its defers (pool.Close, signal stop) still
+	// execute: os.Exit skips deferred calls entirely.
+	if err := run(log); err != nil {
+		log.Error("fatal", "err", err)
+		os.Exit(1)
+	}
+}
+
+func run(log *slog.Logger) error {
+	cfg, err := infra.Load()
+	if err != nil {
+		return err
+	}
+
+	// One cancellation source for the whole process: HTTP server and reaper
+	// both observe it and wind down together.
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	pool, err := infra.NewPool(ctx, cfg)
+	if err != nil {
+		return err
+	}
+	defer pool.Close()
+
+	// --- composition root: the only place that knows concrete types ---
+	//
+	// Wiring reads outward-in: adapters are constructed, then injected into
+	// use cases through their ports. Nothing below this function can see a
+	// pgxpool, and nothing above the repositories can see SQL.
+	var (
+		clk      = infra.NewClock()
+		tx       = postgres.NewTxManager(pool)
+		taskRepo = postgres.NewTaskRepo(pool)
+		jobRepo  = postgres.NewJobRepo(pool)
+	)
+
+	useCases := httptransport.UseCases{
+		CreateJob:    usecase.NewCreateJob(jobRepo, taskRepo, tx, clk),
+		ClaimTask:    usecase.NewClaimTask(taskRepo, clk, cfg.LeaseDuration),
+		RenewLease:   usecase.NewRenewLease(taskRepo, tx, clk, cfg.LeaseDuration),
+		CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, tx, clk),
+		FailTask:     usecase.NewFailTask(taskRepo, jobRepo, tx, clk),
+		GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo),
+	}
+
+	// Background workers are tracked so shutdown can wait for them. Without
+	// this the process would exit while the reaper sat mid-UPDATE, and the
+	// deferred pool.Close() would pull connections out from under it.
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		infra.RunReaper(ctx, log, usecase.NewExpireLeases(taskRepo, clk), cfg.ReaperInterval)
+	}()
+
+	api := httptransport.NewServer(useCases, log, cfg.RequestTimeout)
+	err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.WorkerAuthToken))
+
+	// Shutdown order matters, and defers alone cannot express it (they run
+	// LIFO, so the deferred stop() would fire *after* the wait below).
+	//
+	//   1. stop()    cancel the context, telling the reaper to finish
+	//   2. wg.Wait() let it return from its current tick
+	//   3. deferred pool.Close() closes an idle pool, not a busy one
+	//
+	// Calling stop() here also covers the path where RunServer failed on its
+	// own: the context would never be cancelled otherwise and wg.Wait()
+	// would block forever.
+	stop()
+	wg.Wait()
+	log.Info("shutdown complete")
+
+	return err
+}
diff --git a/coordinator/docker-compose.yml b/coordinator/docker-compose.yml
new file mode 100644
index 0000000..1cbe5b9
--- /dev/null
+++ b/coordinator/docker-compose.yml
@@ -0,0 +1,67 @@
+name: scimesh
+
+services:
+  postgres:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_USER: ${POSTGRES_USER:-scimesh}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-scimesh}
+      POSTGRES_DB: ${POSTGRES_DB:-scimesh}
+    ports:
+      - "${POSTGRES_PORT:-5432}:5432"
+    volumes:
+      - pgdata:/var/lib/postgresql/data
+    healthcheck:
+      # Everything else waits on this, so the check must prove the server
+      # accepts queries — not merely that the port is open.
+      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-scimesh} -d ${POSTGRES_DB:-scimesh}"]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+      start_period: 5s
+
+  # One-shot: applies migrations, then exits. Schema changes stay an explicit
+  # deployment step — the coordinator binary never migrates on startup.
+  migrate:
+    image: migrate/migrate:v4.17.1
+    depends_on:
+      postgres:
+        condition: service_healthy
+    volumes:
+      - ./migrations:/migrations:ro
+    command:
+      - -path=/migrations
+      - -database=postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres:5432/${POSTGRES_DB:-scimesh}?sslmode=disable
+      - up
+    restart: on-failure
+
+  coordinator:
+    build:
+      context: .
+    depends_on:
+      postgres:
+        condition: service_healthy
+      # Start only once the schema exists, otherwise the first query fails.
+      migrate:
+        condition: service_completed_successfully
+    environment:
+      COORDINATOR_ADDR: ":8080"
+      # Host is the service name: compose resolves it on the project network.
+      DATABASE_URL: postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres:5432/${POSTGRES_DB:-scimesh}?sslmode=disable
+      WORKER_AUTH_TOKEN: ${WORKER_AUTH_TOKEN:-dev-token}
+      DB_MAX_CONNS: "10"
+      REQUEST_TIMEOUT: "15s"
+      LEASE_DURATION: "2m"
+      REAPER_INTERVAL: "30s"
+    ports:
+      - "${COORDINATOR_PORT:-8080}:8080"
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8080/health"]
+      interval: 10s
+      timeout: 3s
+      retries: 3
+      start_period: 5s
+    restart: unless-stopped
+
+volumes:
+  pgdata:
diff --git a/coordinator/go.mod b/coordinator/go.mod
new file mode 100644
index 0000000..dd02541
--- /dev/null
+++ b/coordinator/go.mod
@@ -0,0 +1,19 @@
+module github.com/emil28092005/SciMesh/coordinator
+
+go 1.22
+
+require (
+	github.com/cenkalti/backoff/v4 v4.3.0
+	github.com/google/uuid v1.6.0
+	github.com/jackc/pgx/v5 v5.6.0
+	github.com/joho/godotenv v1.5.1
+)
+
+require (
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/puddle/v2 v2.2.1 // indirect
+	golang.org/x/crypto v0.17.0 // indirect
+	golang.org/x/sync v0.1.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+)
diff --git a/coordinator/go.sum b/coordinator/go.sum
new file mode 100644
index 0000000..c078c1d
--- /dev/null
+++ b/coordinator/go.sum
@@ -0,0 +1,34 @@
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
+github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
+github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
+github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
+github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
+golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/coordinator/internal/domain/errors.go b/coordinator/internal/domain/errors.go
new file mode 100644
index 0000000..f1e69e8
--- /dev/null
+++ b/coordinator/internal/domain/errors.go
@@ -0,0 +1,18 @@
+package domain
+
+import "errors"
+
+// Business-rule violations. They live in the innermost layer because they
+// describe what the rules are, not how a transport reports them: the HTTP
+// adapter maps these to status codes, and nothing here knows 409 exists.
+//
+// Always compare with errors.Is — outer layers may wrap these with %w.
+var (
+	ErrJobNotFound    = errors.New("job not found")
+	ErrTaskNotFound   = errors.New("task not found")
+	ErrLeaseConflict  = errors.New("task leased to another worker")
+	ErrStaleAttempt   = errors.New("attempt does not match lease")
+	ErrResultConflict = errors.New("different result already recorded")
+	ErrInvalidInput   = errors.New("invalid input")
+	ErrTaskNotLeased  = errors.New("task is not currently leased")
+)
diff --git a/coordinator/internal/domain/job.go b/coordinator/internal/domain/job.go
new file mode 100644
index 0000000..b6e9f4a
--- /dev/null
+++ b/coordinator/internal/domain/job.go
@@ -0,0 +1,107 @@
+package domain
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+)
+
+type JobStatus string
+
+const (
+	JobPending   JobStatus = "pending"
+	JobRunning   JobStatus = "running"
+	JobCompleted JobStatus = "completed"
+	JobFailed    JobStatus = "failed"
+	JobCancelled JobStatus = "cancelled"
+)
+
+// Job is one user submission that fans out into one or more tasks.
+type Job struct {
+	ID          uuid.UUID
+	Workload    string
+	InputURI    string
+	Parameters  map[string]any
+	Status      JobStatus
+	CreatedAt   time.Time
+	CompletedAt *time.Time
+}
+
+// ChunkSpec describes one piece a job is split into. Callers build these from
+// whatever chunking strategy the workload uses; the domain only validates them.
+type ChunkSpec struct {
+	ChunkIndex  int
+	Workload    string // empty inherits the job's workload
+	InputURI    string
+	InputSHA256 string
+	Parameters  map[string]any
+	MaxAttempts int
+}
+
+// NewJobWithTasks builds a job together with all of its tasks, validating the
+// set as a whole. Returning both from one constructor keeps the invariant
+// visible: a job without tasks, or with duplicate chunk indexes, cannot exist.
+func NewJobWithTasks(workload, inputURI string, params map[string]any,
+	chunks []ChunkSpec, now time.Time) (*Job, []*Task, error) {
+
+	if workload == "" || inputURI == "" || len(chunks) == 0 {
+		return nil, nil, ErrInvalidInput
+	}
+
+	job := &Job{
+		ID:         uuid.New(),
+		Workload:   workload,
+		InputURI:   inputURI,
+		Parameters: params,
+		Status:     JobPending,
+		CreatedAt:  now,
+	}
+
+	seen := make(map[int]struct{}, len(chunks))
+	tasks := make([]*Task, 0, len(chunks))
+	for _, c := range chunks {
+		if _, dup := seen[c.ChunkIndex]; dup {
+			return nil, nil, ErrInvalidInput // unique (job_id, chunk_index)
+		}
+		seen[c.ChunkIndex] = struct{}{}
+
+		w := c.Workload
+		if w == "" {
+			w = workload
+		}
+		task, err := NewTask(job.ID, c.ChunkIndex, w, c.InputURI, c.InputSHA256,
+			c.Parameters, c.MaxAttempts, now)
+		if err != nil {
+			return nil, nil, err
+		}
+		tasks = append(tasks, task)
+	}
+	return job, tasks, nil
+}
+
+// JobProgress is the aggregate view of a job and the state of its tasks.
+type JobProgress struct {
+	Job     Job
+	Total   int
+	Pending int
+	Leased  int
+	Done    int
+	Failed  int
+}
+
+// DeriveStatus computes what the job's status should be from its task counts,
+// so the rule lives here rather than in a SQL trigger or a handler.
+func (p JobProgress) DeriveStatus() JobStatus {
+	switch {
+	case p.Total == 0:
+		return JobPending
+	case p.Done == p.Total:
+		return JobCompleted
+	case p.Failed > 0 && p.Done+p.Failed == p.Total:
+		return JobFailed
+	case p.Leased > 0 || p.Done > 0 || p.Failed > 0:
+		return JobRunning
+	default:
+		return JobPending
+	}
+}
diff --git a/coordinator/internal/domain/task.go b/coordinator/internal/domain/task.go
new file mode 100644
index 0000000..6704552
--- /dev/null
+++ b/coordinator/internal/domain/task.go
@@ -0,0 +1,244 @@
+// Package domain holds SciMesh's entities and the rules that govern them. It
+// is the innermost layer: it imports nothing from this module and knows nothing
+// about HTTP, SQL, or configuration. Every state transition a task can undergo
+// is a method here, so the rules are unit-testable without a database.
+package domain
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+)
+
+type TaskStatus string
+
+const (
+	TaskPending   TaskStatus = "pending"
+	TaskLeased    TaskStatus = "leased"
+	TaskCompleted TaskStatus = "completed"
+	TaskFailed    TaskStatus = "failed"
+	TaskCancelled TaskStatus = "cancelled"
+)
+
+// ErrCodeLeaseExpired marks tasks failed by the reaper rather than by a worker.
+const ErrCodeLeaseExpired = "lease_expired"
+
+// Task is one independently executable chunk of a job.
+//
+// Nullable columns are pointers so "no lease" stays distinguishable from
+// "lease owned by the empty string" — a plain string cannot express both.
+type Task struct {
+	ID             uuid.UUID
+	JobID          uuid.UUID
+	ChunkIndex     int
+	Workload       string
+	InputURI       string
+	InputSHA256    string
+	Parameters     map[string]any
+	Status         TaskStatus
+	Attempt        int
+	MaxAttempts    int
+	LeaseOwner     *string
+	LeaseExpiresAt *time.Time
+	ResultURI      *string
+	ResultSHA256   *string
+	Metrics        map[string]any
+	ErrorCode      *string
+	ErrorMessage   *string
+	CreatedAt      time.Time
+	StartedAt      *time.Time
+	CompletedAt    *time.Time
+	Version        int
+}
+
+// NewTask builds a pending task. maxAttempts <= 0 falls back to the default.
+func NewTask(jobID uuid.UUID, chunkIndex int, workload, inputURI, inputSHA256 string,
+	params map[string]any, maxAttempts int, now time.Time) (*Task, error) {
+
+	if inputURI == "" {
+		return nil, ErrInvalidInput
+	}
+	if inputSHA256 == "" {
+		return nil, ErrInvalidInput // checksum is mandatory: workers verify inputs
+	}
+	if chunkIndex < 0 {
+		return nil, ErrInvalidInput
+	}
+	if maxAttempts <= 0 {
+		maxAttempts = DefaultMaxAttempts
+	}
+	return &Task{
+		ID:          uuid.New(),
+		JobID:       jobID,
+		ChunkIndex:  chunkIndex,
+		Workload:    workload,
+		InputURI:    inputURI,
+		InputSHA256: inputSHA256,
+		Parameters:  params,
+		Status:      TaskPending,
+		Attempt:     0,
+		MaxAttempts: maxAttempts,
+		CreatedAt:   now,
+	}, nil
+}
+
+// DefaultMaxAttempts applies when a task does not specify its own ceiling.
+const DefaultMaxAttempts = 3
+
+// CanRetry reports whether any attempts remain.
+func (t *Task) CanRetry() bool { return t.Attempt < t.MaxAttempts }
+
+// IsLeaseHeldBy reports whether worker currently holds this task at attempt.
+func (t *Task) IsLeaseHeldBy(worker string, attempt int) bool {
+	return t.LeaseOwner != nil && *t.LeaseOwner == worker && t.Attempt == attempt
+}
+
+// AsClaimed projects the task into the trimmed view handed to a worker:
+// everything needed to execute, nothing it has no business seeing.
+func (t *Task) AsClaimed() ClaimedTask {
+	ct := ClaimedTask{
+		TaskID:      t.ID,
+		JobID:       t.JobID,
+		ChunkIndex:  t.ChunkIndex,
+		Workload:    t.Workload,
+		InputURI:    t.InputURI,
+		InputSHA256: t.InputSHA256,
+		Parameters:  t.Parameters,
+		Attempt:     t.Attempt,
+	}
+	if t.LeaseOwner != nil {
+		ct.LeaseOwner = *t.LeaseOwner
+	}
+	if t.LeaseExpiresAt != nil {
+		ct.LeaseExpiresAt = *t.LeaseExpiresAt
+	}
+	return ct
+}
+
+// verifyLease is the guard every worker-driven transition shares: the caller
+// must own the lease and reference the attempt it was granted.
+func (t *Task) verifyLease(worker string, attempt int) error {
+	if t.Status != TaskLeased {
+		return ErrTaskNotLeased
+	}
+	if t.LeaseOwner == nil || *t.LeaseOwner != worker {
+		return ErrLeaseConflict
+	}
+	if t.Attempt != attempt {
+		return ErrStaleAttempt
+	}
+	return nil
+}
+
+// RenewLease extends the lease of the worker that holds it.
+func (t *Task) RenewLease(worker string, attempt int, until time.Time) error {
+	if err := t.verifyLease(worker, attempt); err != nil {
+		return err
+	}
+	t.LeaseExpiresAt = &until
+	t.Version++
+	return nil
+}
+
+// CompleteWith records a successful result.
+//
+// Idempotency comes first deliberately: a worker whose network dropped will
+// retry the same manifest, and that must succeed rather than trip the lease
+// check on a task the coordinator already finished. A *different* manifest for
+// an already-completed task is a genuine conflict.
+func (t *Task) CompleteWith(resultURI, resultSHA256 string, metrics map[string]any,
+	worker string, attempt int, now time.Time) error {
+
+	if resultURI == "" || resultSHA256 == "" {
+		return ErrInvalidInput
+	}
+
+	if t.Status == TaskCompleted {
+		if t.Attempt == attempt && t.ResultURI != nil && *t.ResultURI == resultURI &&
+			t.ResultSHA256 != nil && *t.ResultSHA256 == resultSHA256 {
+			return nil // same attempt, same manifest — replay of a successful call
+		}
+		return ErrResultConflict
+	}
+
+	if err := t.verifyLease(worker, attempt); err != nil {
+		return err
+	}
+
+	t.Status = TaskCompleted
+	t.ResultURI = &resultURI
+	t.ResultSHA256 = &resultSHA256
+	t.Metrics = metrics
+	t.CompletedAt = &now
+	t.LeaseOwner = nil
+	t.LeaseExpiresAt = nil
+	t.ErrorCode = nil
+	t.ErrorMessage = nil
+	t.Version++
+	return nil
+}
+
+// Fail records a worker-reported failure. A retryable failure with attempts
+// left returns the task to the queue; otherwise it terminates as failed.
+func (t *Task) Fail(worker string, attempt int, code, message string, retryable bool, now time.Time) error {
+	if err := t.verifyLease(worker, attempt); err != nil {
+		return err
+	}
+	t.ErrorCode = &code
+	t.ErrorMessage = &message
+	t.LeaseOwner = nil
+	t.LeaseExpiresAt = nil
+	t.Version++
+
+	if retryable && t.CanRetry() {
+		t.Status = TaskPending
+		return nil
+	}
+	t.Status = TaskFailed
+	t.CompletedAt = &now
+	return nil
+}
+
+// ExpireLease is applied by the reaper when a lease elapses without a
+// heartbeat: requeue while attempts remain, otherwise fail terminally.
+func (t *Task) ExpireLease(now time.Time) {
+	if t.Status != TaskLeased {
+		return
+	}
+	t.LeaseOwner = nil
+	t.LeaseExpiresAt = nil
+	t.Version++
+
+	if t.CanRetry() {
+		t.Status = TaskPending
+		return
+	}
+	code, msg := ErrCodeLeaseExpired, "lease expired after the final attempt"
+	t.ErrorCode = &code
+	t.ErrorMessage = &msg
+	t.Status = TaskFailed
+	t.CompletedAt = &now
+}
+
+// ClaimedTask is the worker-facing projection of a leased task.
+type ClaimedTask struct {
+	TaskID         uuid.UUID
+	JobID          uuid.UUID
+	ChunkIndex     int
+	Workload       string
+	InputURI       string
+	InputSHA256    string
+	Parameters     map[string]any
+	Attempt        int
+	LeaseOwner     string
+	LeaseExpiresAt time.Time
+}
+
+// ResultManifest is a completed task's output, ordered for the stitcher.
+type ResultManifest struct {
+	TaskID       uuid.UUID
+	ChunkIndex   int
+	ResultURI    string
+	ResultSHA256 string
+	Metrics      map[string]any
+}
diff --git a/coordinator/internal/domain/task_test.go b/coordinator/internal/domain/task_test.go
new file mode 100644
index 0000000..3650478
--- /dev/null
+++ b/coordinator/internal/domain/task_test.go
@@ -0,0 +1,183 @@
+package domain
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+var (
+	testNow    = time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)
+	testLater  = testNow.Add(time.Hour)
+	testWorker = "worker-1"
+)
+
+// leasedTask builds a task already leased to testWorker at the given attempt.
+func leasedTask(attempt, maxAttempts int) *Task {
+	owner := testWorker
+	expires := testLater
+	return &Task{
+		ID:             uuid.New(),
+		JobID:          uuid.New(),
+		Status:         TaskLeased,
+		Attempt:        attempt,
+		MaxAttempts:    maxAttempts,
+		LeaseOwner:     &owner,
+		LeaseExpiresAt: &expires,
+	}
+}
+
+func TestCompleteWithRecordsResult(t *testing.T) {
+	task := leasedTask(1, 3)
+
+	if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if task.Status != TaskCompleted {
+		t.Errorf("status = %q, want completed", task.Status)
+	}
+	if task.LeaseOwner != nil || task.LeaseExpiresAt != nil {
+		t.Error("lease must be released on completion")
+	}
+	if task.CompletedAt == nil || !task.CompletedAt.Equal(testNow) {
+		t.Error("completed_at must be stamped")
+	}
+}
+
+// A worker whose network dropped retries the same manifest; that must succeed
+// rather than fail on the lease it has already given up.
+func TestCompleteWithIsIdempotentForSameManifest(t *testing.T) {
+	task := leasedTask(1, 3)
+	if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil {
+		t.Fatalf("first call: %v", err)
+	}
+	versionAfterFirst := task.Version
+
+	if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testLater); err != nil {
+		t.Fatalf("replay must be idempotent, got %v", err)
+	}
+	if task.Version != versionAfterFirst {
+		t.Error("replay must not mutate the task")
+	}
+}
+
+func TestCompleteWithRejectsDifferentManifest(t *testing.T) {
+	task := leasedTask(1, 3)
+	if err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow); err != nil {
+		t.Fatalf("first call: %v", err)
+	}
+
+	err := task.CompleteWith("s3://other.csv", "def", nil, testWorker, 1, testLater)
+	if !errors.Is(err, ErrResultConflict) {
+		t.Errorf("err = %v, want ErrResultConflict", err)
+	}
+}
+
+func TestCompleteWithRejectsForeignWorker(t *testing.T) {
+	task := leasedTask(1, 3)
+
+	err := task.CompleteWith("s3://r.csv", "abc", nil, "worker-2", 1, testNow)
+	if !errors.Is(err, ErrLeaseConflict) {
+		t.Errorf("err = %v, want ErrLeaseConflict", err)
+	}
+}
+
+func TestCompleteWithRejectsStaleAttempt(t *testing.T) {
+	task := leasedTask(2, 3) // task is on attempt 2
+
+	err := task.CompleteWith("s3://r.csv", "abc", nil, testWorker, 1, testNow) // worker thinks it is 1
+	if !errors.Is(err, ErrStaleAttempt) {
+		t.Errorf("err = %v, want ErrStaleAttempt", err)
+	}
+}
+
+func TestFailRequeuesWhileAttemptsRemain(t *testing.T) {
+	task := leasedTask(1, 3)
+
+	if err := task.Fail(testWorker, 1, "boom", "exploded", true, testNow); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if task.Status != TaskPending {
+		t.Errorf("status = %q, want pending", task.Status)
+	}
+	if task.LeaseOwner != nil {
+		t.Error("lease must be released so another worker can claim it")
+	}
+}
+
+func TestFailTerminatesOnFinalAttempt(t *testing.T) {
+	task := leasedTask(3, 3) // no attempts left
+
+	if err := task.Fail(testWorker, 3, "boom", "exploded", true, testNow); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if task.Status != TaskFailed {
+		t.Errorf("status = %q, want failed", task.Status)
+	}
+}
+
+func TestFailIsTerminalWhenNotRetryable(t *testing.T) {
+	task := leasedTask(1, 3) // attempts remain, but the error is fatal
+
+	if err := task.Fail(testWorker, 1, "bad_input", "checksum mismatch", false, testNow); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if task.Status != TaskFailed {
+		t.Errorf("status = %q, want failed", task.Status)
+	}
+}
+
+// This is the MVP acceptance criterion: a dead worker must not strand its task.
+func TestExpireLeaseRequeuesWhileAttemptsRemain(t *testing.T) {
+	task := leasedTask(1, 3)
+
+	task.ExpireLease(testNow)
+
+	if task.Status != TaskPending {
+		t.Errorf("status = %q, want pending", task.Status)
+	}
+	if task.LeaseOwner != nil || task.LeaseExpiresAt != nil {
+		t.Error("expired lease must be cleared")
+	}
+}
+
+func TestExpireLeaseFailsAfterFinalAttempt(t *testing.T) {
+	task := leasedTask(3, 3)
+
+	task.ExpireLease(testNow)
+
+	if task.Status != TaskFailed {
+		t.Errorf("status = %q, want failed", task.Status)
+	}
+	if task.ErrorCode == nil || *task.ErrorCode != ErrCodeLeaseExpired {
+		t.Error("expected a lease_expired error code")
+	}
+}
+
+func TestExpireLeaseIgnoresUnleasedTasks(t *testing.T) {
+	task := &Task{Status: TaskCompleted, Attempt: 1, MaxAttempts: 3}
+
+	task.ExpireLease(testNow)
+
+	if task.Status != TaskCompleted {
+		t.Errorf("status = %q, completed tasks must be untouched", task.Status)
+	}
+}
+
+func TestRenewLeaseExtendsOnlyForHolder(t *testing.T) {
+	task := leasedTask(1, 3)
+	until := testLater.Add(time.Hour)
+
+	if err := task.RenewLease(testWorker, 1, until); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !task.LeaseExpiresAt.Equal(until) {
+		t.Error("lease must be extended")
+	}
+
+	if err := task.RenewLease("worker-2", 1, until); !errors.Is(err, ErrLeaseConflict) {
+		t.Errorf("err = %v, want ErrLeaseConflict", err)
+	}
+}
diff --git a/coordinator/internal/infra/clock.go b/coordinator/internal/infra/clock.go
new file mode 100644
index 0000000..eedbcde
--- /dev/null
+++ b/coordinator/internal/infra/clock.go
@@ -0,0 +1,13 @@
+// Clock: the real implementation of the usecase.Clock port. It lives out here
+// because reading the system clock is infrastructure; tests substitute a fixed one.
+package infra
+
+import "time"
+
+type System struct{}
+
+func NewClock() System { return System{} }
+
+// Now returns UTC so every timestamp the coordinator writes is comparable
+// regardless of the host's timezone.
+func (System) Now() time.Time { return time.Now().UTC() }
diff --git a/coordinator/internal/infra/config.go b/coordinator/internal/infra/config.go
new file mode 100644
index 0000000..db07876
--- /dev/null
+++ b/coordinator/internal/infra/config.go
@@ -0,0 +1,135 @@
+// Config: coordinator settings, read only from the environment, so the same
+// binary behaves identically in CI, local, and prod.
+package infra
+
+import (
+	"errors"
+	"fmt"
+	"io/fs"
+	"math"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/joho/godotenv"
+)
+
+// defaultEnvFile is loaded by Load unless ENV_FILE points elsewhere.
+const defaultEnvFile = ".env"
+
+type Config struct {
+	// HTTP listen address, e.g. ":8080".
+	Addr string
+	// PostgreSQL connection string (pgx format / libpq URL).
+	DatabaseURL string
+	// Shared bearer token workers must present. Empty disables auth (dev only).
+	WorkerAuthToken string
+
+	// Connection pool upper bound.
+	DBMaxConns int32
+	// Per-request context timeout applied to handlers and DB calls.
+	RequestTimeout time.Duration
+
+	// Default lease length handed out on claim.
+	LeaseDuration time.Duration
+	// Default attempt ceiling for newly created tasks.
+	DefaultMaxAttempts int
+	// How often the background lease-reaper runs.
+	ReaperInterval time.Duration
+}
+
+// Load reads the environment and fails fast on anything required-but-missing
+// or malformed, so a misconfigured process never limps along half-wired.
+//
+// A .env file (path overridable via ENV_FILE) is loaded first as a local-dev
+// convenience. It only fills variables the environment does not already define.
+func Load() (Config, error) {
+	envFile := os.Getenv("ENV_FILE")
+	if envFile == "" {
+		envFile = defaultEnvFile
+	}
+	// godotenv.Load never overwrites variables already present in the
+	// environment, so an orchestrator's values always beat the file. A missing
+	// file is expected in production, where env vars are injected directly.
+	if err := godotenv.Load(envFile); err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return Config{}, fmt.Errorf("load env file %q: %w", envFile, err)
+	}
+
+	cfg := Config{
+		Addr:               getEnv("COORDINATOR_ADDR", ":8080"),
+		DatabaseURL:        os.Getenv("DATABASE_URL"),
+		WorkerAuthToken:    os.Getenv("WORKER_AUTH_TOKEN"),
+		DBMaxConns:         10,
+		RequestTimeout:     15 * time.Second,
+		LeaseDuration:      2 * time.Minute,
+		DefaultMaxAttempts: 3,
+		ReaperInterval:     30 * time.Second,
+	}
+
+	if cfg.DatabaseURL == "" {
+		return Config{}, fmt.Errorf("DATABASE_URL is required")
+	}
+
+	var err error
+	if cfg.DBMaxConns, err = getEnvInt32("DB_MAX_CONNS", cfg.DBMaxConns); err != nil {
+		return Config{}, err
+	}
+	if cfg.RequestTimeout, err = getEnvDuration("REQUEST_TIMEOUT", cfg.RequestTimeout); err != nil {
+		return Config{}, err
+	}
+	if cfg.LeaseDuration, err = getEnvDuration("LEASE_DURATION", cfg.LeaseDuration); err != nil {
+		return Config{}, err
+	}
+	if cfg.ReaperInterval, err = getEnvDuration("REAPER_INTERVAL", cfg.ReaperInterval); err != nil {
+		return Config{}, err
+	}
+	if cfg.DefaultMaxAttempts, err = getEnvInt("DEFAULT_MAX_ATTEMPTS", cfg.DefaultMaxAttempts); err != nil {
+		return Config{}, err
+	}
+
+	return cfg, nil
+}
+
+func getEnv(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
+
+func getEnvInt(key string, def int) (int, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return def, nil
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return 0, fmt.Errorf("%s: %w", key, err)
+	}
+	return n, nil
+}
+
+func getEnvInt32(key string, def int32) (int32, error) {
+	n, err := getEnvInt(key, int(def))
+	if err != nil {
+		return 0, err
+	}
+	// On 64-bit builds int is wider than int32, so an oversized value would
+	// wrap silently — DB_MAX_CONNS=2147483648 becoming a negative pool size.
+	if n < math.MinInt32 || n > math.MaxInt32 {
+		return 0, fmt.Errorf("%s: %d is out of range for int32", key, n)
+	}
+	return int32(n), nil
+}
+
+func getEnvDuration(key string, def time.Duration) (time.Duration, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return def, nil
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil {
+		return 0, fmt.Errorf("%s: %w", key, err)
+	}
+	return d, nil
+}
diff --git a/coordinator/internal/infra/db.go b/coordinator/internal/infra/db.go
new file mode 100644
index 0000000..29899b3
--- /dev/null
+++ b/coordinator/internal/infra/db.go
@@ -0,0 +1,30 @@
+// DB: the PostgreSQL connection pool.
+package infra
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// NewPool builds the single shared pool. The caller owns its lifetime and must
+// Close() it on shutdown.
+func NewPool(ctx context.Context, cfg Config) (*pgxpool.Pool, error) {
+	poolCfg, err := pgxpool.ParseConfig(cfg.DatabaseURL)
+	if err != nil {
+		return nil, err
+	}
+	poolCfg.MaxConns = cfg.DBMaxConns
+
+	pool, err := pgxpool.NewWithConfig(ctx, poolCfg)
+	if err != nil {
+		return nil, err
+	}
+	// pgxpool.New is lazy, so without this ping a bad DATABASE_URL would only
+	// surface on the first request instead of at startup.
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, err
+	}
+	return pool, nil
+}
diff --git a/coordinator/internal/infra/doc.go b/coordinator/internal/infra/doc.go
new file mode 100644
index 0000000..4e64f47
--- /dev/null
+++ b/coordinator/internal/infra/doc.go
@@ -0,0 +1,7 @@
+// Package infra holds the outermost layer: frameworks, drivers, and process
+// wiring. It reads configuration, opens the database pool, supplies the real
+// clock, and runs the HTTP server and background reaper.
+//
+// Nothing inward depends on this package — it is the last thing constructed and
+// the first thing that would be swapped when the runtime environment changes.
+package infra
diff --git a/coordinator/internal/infra/server.go b/coordinator/internal/infra/server.go
new file mode 100644
index 0000000..c017726
--- /dev/null
+++ b/coordinator/internal/infra/server.go
@@ -0,0 +1,72 @@
+// Server: the HTTP listener and the background lease reaper, both shut down
+// cleanly on a signal.
+package infra
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+const shutdownGrace = 15 * time.Second
+
+// Run serves handler until ctx is cancelled, then drains in-flight requests.
+func RunServer(ctx context.Context, log *slog.Logger, addr string, handler http.Handler) error {
+	srv := &http.Server{
+		Addr:              addr,
+		Handler:           handler,
+		ReadHeaderTimeout: 5 * time.Second,
+	}
+
+	// Buffered so this goroutine can exit even when nobody reads the channel
+	// (the ctx.Done branch below) — an unbuffered send would leak it forever.
+	errCh := make(chan error, 1)
+	go func() {
+		log.Info("coordinator listening", "addr", addr)
+		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			errCh <- err
+		}
+	}()
+
+	select {
+	case err := <-errCh:
+		return err
+	case <-ctx.Done():
+		log.Info("shutdown signal received")
+	}
+
+	// A fresh context: ctx is already cancelled, and reusing it would abort the
+	// very requests we are trying to let finish.
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGrace)
+	defer cancel()
+	return srv.Shutdown(shutdownCtx)
+}
+
+// RunReaper periodically reclaims tasks whose lease elapsed, so a worker that
+// died without a heartbeat cannot strand its task in 'leased' forever.
+func RunReaper(ctx context.Context, log *slog.Logger, uc *usecase.ExpireLeases, interval time.Duration) {
+	t := time.NewTicker(interval)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			n, err := uc.Execute(ctx)
+			if err != nil {
+				// Demoted to debug while the repository is still a stub;
+				// raise to Warn once phase 5 lands.
+				log.Debug("reaper skipped", "err", err)
+				continue
+			}
+			if n > 0 {
+				log.Info("reaper requeued expired leases", "count", n)
+			}
+		}
+	}
+}
diff --git a/coordinator/internal/storage/postgres/job_repo.go b/coordinator/internal/storage/postgres/job_repo.go
new file mode 100644
index 0000000..3751801
--- /dev/null
+++ b/coordinator/internal/storage/postgres/job_repo.go
@@ -0,0 +1,40 @@
+package postgres
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+// JobRepo implements usecase.JobRepository.
+type JobRepo struct {
+	pool *pgxpool.Pool
+}
+
+func NewJobRepo(pool *pgxpool.Pool) *JobRepo {
+	return &JobRepo{pool: pool}
+}
+
+var _ usecase.JobRepository = (*JobRepo)(nil)
+
+// TODO(phase 2-3): replace stubs with real pgx queries.
+
+func (r *JobRepo) Insert(ctx context.Context, j *domain.Job) error {
+	// Phase 2: INSERT INTO jobs ...; runs inside the caller's transaction.
+	return usecase.ErrNotImplemented
+}
+
+func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) {
+	// Phase 3: SELECT ... WHERE id = $1; no rows -> domain.ErrJobNotFound.
+	return nil, usecase.ErrNotImplemented
+}
+
+func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error {
+	// Phase 3: UPDATE jobs SET status = $2, completed_at = $3 WHERE id = $1.
+	return usecase.ErrNotImplemented
+}
diff --git a/coordinator/internal/storage/postgres/retry.go b/coordinator/internal/storage/postgres/retry.go
new file mode 100644
index 0000000..8f0ae86
--- /dev/null
+++ b/coordinator/internal/storage/postgres/retry.go
@@ -0,0 +1,82 @@
+package postgres
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+// Transient PostgreSQL failures. Under concurrent claiming these are expected
+// rather than exceptional: two coordinators touching neighbouring rows can
+// deadlock or fail to serialize, and the correct response is to try again.
+const (
+	codeSerializationFailure = "40001"
+	codeDeadlockDetected     = "40P01"
+	codeTooManyConnections   = "53300"
+	codeCannotConnectNow     = "57P03"
+)
+
+// Retry budget: short and bounded. A worker polling for tasks would rather get
+// a fast error and poll again than have its request hang for half a minute.
+const (
+	retryInitialInterval = 50 * time.Millisecond
+	retryMaxInterval     = 1 * time.Second
+	retryMaxElapsedTime  = 5 * time.Second
+)
+
+// isTransient reports whether err is worth retrying.
+//
+// The default is *not* to retry: a constraint violation or a syntax error will
+// fail identically every time, and retrying it only multiplies the damage.
+func isTransient(err error) bool {
+	if err == nil {
+		return false
+	}
+	// A cancelled caller does not want another attempt.
+	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+		return false
+	}
+
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		switch pgErr.Code {
+		case codeSerializationFailure, codeDeadlockDetected,
+			codeTooManyConnections, codeCannotConnectNow:
+			return true
+		default:
+			return false
+		}
+	}
+
+	// Connection-level trouble (dropped socket, closed pool). pgconn knows
+	// whether the query could have been executed before the failure — retrying
+	// a maybe-executed write would risk duplicating it.
+	return pgconn.SafeToRetry(err)
+}
+
+// withRetry runs op, retrying only transient database failures with
+// exponential backoff and jitter, and giving up as soon as ctx is done.
+//
+// Jitter matters here: without it, several coordinators that collide once will
+// retry in lockstep and collide again at exactly the same moment.
+func withRetry(ctx context.Context, op func(context.Context) error) error {
+	b := backoff.NewExponentialBackOff()
+	b.InitialInterval = retryInitialInterval
+	b.MaxInterval = retryMaxInterval
+	b.MaxElapsedTime = retryMaxElapsedTime
+	// RandomizationFactor defaults to 0.5, which is the jitter.
+
+	return backoff.Retry(func() error {
+		err := op(ctx)
+		if err == nil {
+			return nil
+		}
+		if !isTransient(err) {
+			return backoff.Permanent(err) // stop now, do not burn the budget
+		}
+		return err
+	}, backoff.WithContext(b, ctx))
+}
diff --git a/coordinator/internal/storage/postgres/retry_test.go b/coordinator/internal/storage/postgres/retry_test.go
new file mode 100644
index 0000000..bdf77ec
--- /dev/null
+++ b/coordinator/internal/storage/postgres/retry_test.go
@@ -0,0 +1,98 @@
+package postgres
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+func TestIsTransient(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"serialization failure", &pgconn.PgError{Code: codeSerializationFailure}, true},
+		{"deadlock", &pgconn.PgError{Code: codeDeadlockDetected}, true},
+		{"too many connections", &pgconn.PgError{Code: codeTooManyConnections}, true},
+		// A unique-violation repeats identically forever — retrying is pointless.
+		{"unique violation", &pgconn.PgError{Code: "23505"}, false},
+		{"syntax error", &pgconn.PgError{Code: "42601"}, false},
+		{"context cancelled", context.Canceled, false},
+		{"deadline exceeded", context.DeadlineExceeded, false},
+		{"unknown error", errors.New("boom"), false},
+		// Wrapping must not hide the cause: errors.As walks the chain.
+		{"wrapped deadlock", errors2Wrap(&pgconn.PgError{Code: codeDeadlockDetected}), true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isTransient(tt.err); got != tt.want {
+				t.Errorf("isTransient(%v) = %v, want %v", tt.err, got, tt.want)
+			}
+		})
+	}
+}
+
+func errors2Wrap(err error) error {
+	return errors.Join(errors.New("query failed"), err)
+}
+
+func TestWithRetrySucceedsAfterTransientFailures(t *testing.T) {
+	calls := 0
+	err := withRetry(context.Background(), func(context.Context) error {
+		calls++
+		if calls < 3 {
+			return &pgconn.PgError{Code: codeSerializationFailure}
+		}
+		return nil
+	})
+
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if calls != 3 {
+		t.Errorf("calls = %d, want 3", calls)
+	}
+}
+
+func TestWithRetryStopsOnPermanentError(t *testing.T) {
+	permanent := &pgconn.PgError{Code: "23505"} // unique violation
+	calls := 0
+
+	err := withRetry(context.Background(), func(context.Context) error {
+		calls++
+		return permanent
+	})
+
+	if !errors.Is(err, permanent) {
+		t.Errorf("err = %v, want the original error", err)
+	}
+	if calls != 1 {
+		t.Errorf("calls = %d, want 1 — a permanent error must not be retried", calls)
+	}
+}
+
+func TestWithRetryHonoursContextCancellation(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	calls := 0
+	start := time.Now()
+	err := withRetry(ctx, func(context.Context) error {
+		calls++
+		return &pgconn.PgError{Code: codeDeadlockDetected}
+	})
+
+	if err == nil {
+		t.Fatal("expected an error once the context expired")
+	}
+	// Must abort at the deadline, not run the full 5s retry budget.
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Errorf("took %v, expected to stop at the context deadline", elapsed)
+	}
+}
diff --git a/coordinator/internal/storage/postgres/task_repo.go b/coordinator/internal/storage/postgres/task_repo.go
new file mode 100644
index 0000000..631a88b
--- /dev/null
+++ b/coordinator/internal/storage/postgres/task_repo.go
@@ -0,0 +1,114 @@
+package postgres
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+// TaskRepo implements usecase.TaskRepository.
+type TaskRepo struct {
+	pool *pgxpool.Pool
+}
+
+func NewTaskRepo(pool *pgxpool.Pool) *TaskRepo {
+	return &TaskRepo{pool: pool}
+}
+
+var _ usecase.TaskRepository = (*TaskRepo)(nil)
+
+// claimNextSQL leases one task in a single statement.
+//
+// FOR UPDATE SKIP LOCKED is what makes concurrent coordinators safe: each
+// process locks a different candidate row instead of queueing on the same one,
+// so no task is ever handed to two workers and no claim blocks behind another.
+// Splitting this into SELECT + UPDATE would reintroduce exactly that race.
+//
+//nolint:unused // wired up in phase 2; kept beside the repository it belongs to
+const claimNextSQL = `
+WITH candidate AS (
+    SELECT id
+    FROM tasks
+    WHERE status = 'pending'
+      AND attempt < max_attempts
+      AND (cardinality($1::text[]) = 0 OR workload = ANY($1))
+    ORDER BY created_at, chunk_index
+    FOR UPDATE SKIP LOCKED
+    LIMIT 1
+)
+UPDATE tasks
+SET status           = 'leased',
+    attempt          = attempt + 1,
+    lease_owner      = $2,
+    lease_expires_at = $3,
+    started_at       = COALESCE(started_at, $4),
+    version          = version + 1
+FROM candidate
+WHERE tasks.id = candidate.id
+RETURNING tasks.id, tasks.job_id, tasks.chunk_index, tasks.workload,
+          tasks.input_uri, tasks.input_sha256, tasks.parameters,
+          tasks.status, tasks.attempt, tasks.max_attempts,
+          tasks.lease_owner, tasks.lease_expires_at, tasks.version;
+`
+
+// expireLeasesSQL applies the lease-expiry rule set-based, mirroring
+// domain.Task.ExpireLease: requeue while attempts remain, otherwise fail.
+//
+//nolint:unused // wired up in phase 5; mirrors domain.Task.ExpireLease
+const expireLeasesSQL = `
+UPDATE tasks
+SET status           = CASE WHEN attempt < max_attempts THEN 'pending'::task_status
+                            ELSE 'failed'::task_status END,
+    lease_owner      = NULL,
+    lease_expires_at = NULL,
+    error_code       = CASE WHEN attempt >= max_attempts THEN $2 ELSE error_code END,
+    error_message    = CASE WHEN attempt >= max_attempts
+                            THEN 'lease expired after the final attempt'
+                            ELSE error_message END,
+    completed_at     = CASE WHEN attempt >= max_attempts THEN $1 ELSE completed_at END,
+    version          = version + 1
+WHERE status = 'leased' AND lease_expires_at < $1;
+`
+
+// TODO(phase 2-6): replace stubs with real pgx queries; the SQL above is ready
+// to wire up. Each method maps 1:1 to a roadmap phase.
+
+func (r *TaskRepo) ClaimNext(ctx context.Context, f usecase.ClaimFilter) (*domain.Task, error) {
+	// Phase 2: run claimNextSQL; pgx.ErrNoRows -> (nil, nil).
+	return nil, usecase.ErrNotImplemented
+}
+
+func (r *TaskRepo) GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task, error) {
+	// Phase 3: SELECT ... WHERE id = $1 FOR UPDATE; no rows -> domain.ErrTaskNotFound.
+	return nil, usecase.ErrNotImplemented
+}
+
+func (r *TaskRepo) Update(ctx context.Context, t *domain.Task) error {
+	// Phase 3: UPDATE ... WHERE id = $1 AND version = $2 (optimistic concurrency).
+	return usecase.ErrNotImplemented
+}
+
+func (r *TaskRepo) InsertBatch(ctx context.Context, tasks []*domain.Task) error {
+	// Phase 2: pgx.Batch or COPY; runs inside the caller's transaction.
+	return usecase.ErrNotImplemented
+}
+
+func (r *TaskRepo) ListCompleted(ctx context.Context, jobID uuid.UUID) ([]*domain.Task, error) {
+	// Phase 6: WHERE job_id = $1 AND status = 'completed' ORDER BY chunk_index.
+	return nil, usecase.ErrNotImplemented
+}
+
+func (r *TaskRepo) CountByStatus(ctx context.Context, jobID uuid.UUID) (map[domain.TaskStatus]int, error) {
+	// Phase 3: SELECT status, count(*) ... GROUP BY status.
+	return nil, usecase.ErrNotImplemented
+}
+
+func (r *TaskRepo) ExpireLeases(ctx context.Context, now time.Time) (int64, error) {
+	// Phase 5: run expireLeasesSQL, return the affected row count.
+	return 0, usecase.ErrNotImplemented
+}
diff --git a/coordinator/internal/storage/postgres/tx.go b/coordinator/internal/storage/postgres/tx.go
new file mode 100644
index 0000000..72dfe9b
--- /dev/null
+++ b/coordinator/internal/storage/postgres/tx.go
@@ -0,0 +1,69 @@
+// Package postgres implements the usecase repository ports on PostgreSQL.
+// SQL and pgx types never escape this package.
+package postgres
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// querier is satisfied by both *pgxpool.Pool and pgx.Tx, letting every
+// repository method run identically inside or outside a transaction.
+//
+//nolint:unused // used by repository methods once phase 2 replaces the stubs
+type querier interface {
+	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
+	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
+	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
+}
+
+// txKey is an unexported struct type, so no other package can collide with it
+// or reach the transaction we stash in the context.
+type txKey struct{}
+
+// TxManager implements usecase.TxManager.
+type TxManager struct {
+	pool *pgxpool.Pool
+}
+
+func NewTxManager(pool *pgxpool.Pool) *TxManager {
+	return &TxManager{pool: pool}
+}
+
+// WithinTx runs fn inside one transaction, committing on success and rolling
+// back on any error or panic.
+//
+// The transaction travels in the context rather than in fn's signature, which
+// is what lets the usecase layer express "do these repository calls atomically"
+// without its port ever mentioning pgx.
+func (m *TxManager) WithinTx(ctx context.Context, fn func(ctx context.Context) error) error {
+	if _, ok := ctx.Value(txKey{}).(pgx.Tx); ok {
+		return fn(ctx) // already inside a transaction — join it, don't nest
+	}
+
+	tx, err := m.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	// Rollback after a successful Commit is a no-op, so this defer is safe and
+	// also covers the panic path.
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	if err := fn(context.WithValue(ctx, txKey{}, tx)); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
+}
+
+// conn returns the transaction bound to ctx, or the pool when there is none.
+//
+//nolint:unused // every repository method will route through this in phase 2
+func conn(ctx context.Context, pool *pgxpool.Pool) querier {
+	if tx, ok := ctx.Value(txKey{}).(pgx.Tx); ok {
+		return tx
+	}
+	return pool
+}
diff --git a/coordinator/internal/transport/http/dto.go b/coordinator/internal/transport/http/dto.go
new file mode 100644
index 0000000..b308843
--- /dev/null
+++ b/coordinator/internal/transport/http/dto.go
@@ -0,0 +1,119 @@
+package http
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+// Wire formats. Keeping them separate from domain entities means the API
+// contract can evolve without reshaping the database, and nothing internal
+// (version counters, other workers' errors) leaks by accident.
+
+type createJobRequest struct {
+	Workload   string         `json:"workload"`
+	InputURI   string         `json:"input_uri"`
+	Parameters map[string]any `json:"parameters"`
+	Chunks     []chunkDTO     `json:"chunks"`
+}
+
+type chunkDTO struct {
+	ChunkIndex  int            `json:"chunk_index"`
+	Workload    string         `json:"workload"`
+	InputURI    string         `json:"input_uri"`
+	InputSHA256 string         `json:"input_sha256"`
+	Parameters  map[string]any `json:"parameters"`
+	MaxAttempts int            `json:"max_attempts"`
+}
+
+type claimRequest struct {
+	WorkerID  string   `json:"worker_id"`
+	Workloads []string `json:"workloads"`
+}
+
+type heartbeatRequest struct {
+	WorkerID string `json:"worker_id"`
+	Attempt  int    `json:"attempt"`
+}
+
+type resultRequest struct {
+	WorkerID     string         `json:"worker_id"`
+	Attempt      int            `json:"attempt"`
+	ResultURI    string         `json:"result_uri"`
+	ResultSHA256 string         `json:"result_sha256"`
+	Metrics      map[string]any `json:"metrics"`
+}
+
+type failureRequest struct {
+	WorkerID     string `json:"worker_id"`
+	Attempt      int    `json:"attempt"`
+	ErrorCode    string `json:"error_code"`
+	ErrorMessage string `json:"error_message"`
+	Retryable    bool   `json:"retryable"`
+}
+
+type jobResponse struct {
+	ID     uuid.UUID `json:"id"`
+	Status string    `json:"status"`
+}
+
+type taskResponse struct {
+	ID     uuid.UUID `json:"id"`
+	JobID  uuid.UUID `json:"job_id"`
+	Status string    `json:"status"`
+}
+
+type claimedTaskResponse struct {
+	TaskID         uuid.UUID      `json:"task_id"`
+	JobID          uuid.UUID      `json:"job_id"`
+	ChunkIndex     int            `json:"chunk_index"`
+	Workload       string         `json:"workload"`
+	InputURI       string         `json:"input_uri"`
+	InputSHA256    string         `json:"input_sha256"`
+	Parameters     map[string]any `json:"parameters"`
+	Attempt        int            `json:"attempt"`
+	LeaseExpiresAt time.Time      `json:"lease_expires_at"`
+}
+
+type jobProgressResponse struct {
+	ID      uuid.UUID `json:"id"`
+	Status  string    `json:"status"`
+	Total   int       `json:"total"`
+	Pending int       `json:"pending"`
+	Leased  int       `json:"leased"`
+	Done    int       `json:"completed"`
+	Failed  int       `json:"failed"`
+}
+
+type errorResponse struct {
+	Error     string `json:"error"`
+	RequestID string `json:"request_id,omitempty"`
+}
+
+func toClaimedTaskResponse(c domain.ClaimedTask) claimedTaskResponse {
+	return claimedTaskResponse{
+		TaskID:         c.TaskID,
+		JobID:          c.JobID,
+		ChunkIndex:     c.ChunkIndex,
+		Workload:       c.Workload,
+		InputURI:       c.InputURI,
+		InputSHA256:    c.InputSHA256,
+		Parameters:     c.Parameters,
+		Attempt:        c.Attempt,
+		LeaseExpiresAt: c.LeaseExpiresAt,
+	}
+}
+
+func toJobProgressResponse(p domain.JobProgress) jobProgressResponse {
+	return jobProgressResponse{
+		ID:      p.Job.ID,
+		Status:  string(p.DeriveStatus()),
+		Total:   p.Total,
+		Pending: p.Pending,
+		Leased:  p.Leased,
+		Done:    p.Done,
+		Failed:  p.Failed,
+	}
+}
diff --git a/coordinator/internal/transport/http/errors.go b/coordinator/internal/transport/http/errors.go
new file mode 100644
index 0000000..d052a90
--- /dev/null
+++ b/coordinator/internal/transport/http/errors.go
@@ -0,0 +1,55 @@
+package http
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+func decodeJSON(r *http.Request, dst any) error {
+	dec := json.NewDecoder(r.Body)
+	// Reject unknown fields: silently ignoring a misspelled "worker_ID" would
+	// surface later as a baffling validation failure.
+	dec.DisallowUnknownFields()
+	return dec.Decode(dst)
+}
+
+// writeError translates domain errors into status codes. This mapping is the
+// only place in the codebase that knows HTTP status codes exist — the inner
+// layers speak only in business terms.
+func (s *Server) writeError(w http.ResponseWriter, r *http.Request, err error) {
+	reqID := requestIDFrom(r.Context())
+
+	status := http.StatusInternalServerError
+	switch {
+	case errors.Is(err, domain.ErrInvalidInput):
+		status = http.StatusBadRequest
+	case errors.Is(err, domain.ErrJobNotFound), errors.Is(err, domain.ErrTaskNotFound):
+		status = http.StatusNotFound
+	case errors.Is(err, domain.ErrLeaseConflict),
+		errors.Is(err, domain.ErrStaleAttempt),
+		errors.Is(err, domain.ErrResultConflict),
+		errors.Is(err, domain.ErrTaskNotLeased):
+		status = http.StatusConflict
+	case errors.Is(err, usecase.ErrNotImplemented):
+		status = http.StatusNotImplemented
+	}
+
+	if status >= 500 {
+		// Never echo an internal error: it can carry table names, query
+		// fragments, and values. The request ID is the bridge to the logs.
+		s.log.Error("request failed", "request_id", reqID, "path", r.URL.Path, "err", err)
+		writeJSON(w, status, errorResponse{Error: "internal error", RequestID: reqID})
+		return
+	}
+	writeJSON(w, status, errorResponse{Error: err.Error(), RequestID: reqID})
+}
diff --git a/coordinator/internal/transport/http/handlers.go b/coordinator/internal/transport/http/handlers.go
new file mode 100644
index 0000000..00e4f0e
--- /dev/null
+++ b/coordinator/internal/transport/http/handlers.go
@@ -0,0 +1,181 @@
+package http
+
+import (
+	"context"
+	"net/http"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+// Every handler follows the same shape: decode, map to a use-case input,
+// execute, translate. Anything resembling a rule belongs one layer inward.
+
+func (s *Server) handleCreateJob(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+
+	var req createJobRequest
+	if err := decodeJSON(r, &req); err != nil {
+		s.writeError(w, r, domain.ErrInvalidInput)
+		return
+	}
+
+	in := usecase.CreateJobInput{
+		Workload:   req.Workload,
+		InputURI:   req.InputURI,
+		Parameters: req.Parameters,
+	}
+	for _, c := range req.Chunks {
+		in.Chunks = append(in.Chunks, usecase.ChunkInput(c))
+	}
+
+	job, err := s.uc.CreateJob.Execute(ctx, in)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusCreated, jobResponse{ID: job.ID, Status: string(job.Status)})
+}
+
+func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+
+	var req claimRequest
+	if err := decodeJSON(r, &req); err != nil {
+		s.writeError(w, r, domain.ErrInvalidInput)
+		return
+	}
+
+	claimed, err := s.uc.ClaimTask.Execute(ctx, usecase.ClaimTaskInput{
+		WorkerID:  req.WorkerID,
+		Workloads: req.Workloads,
+	})
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	if claimed == nil {
+		w.WriteHeader(http.StatusNoContent) // empty queue, not an error
+		return
+	}
+	writeJSON(w, http.StatusOK, toClaimedTaskResponse(*claimed))
+}
+
+func (s *Server) handleHeartbeat(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+
+	taskID, ok := s.pathUUID(w, r, "task_id")
+	if !ok {
+		return
+	}
+	var req heartbeatRequest
+	if err := decodeJSON(r, &req); err != nil {
+		s.writeError(w, r, domain.ErrInvalidInput)
+		return
+	}
+
+	claimed, err := s.uc.RenewLease.Execute(ctx, usecase.RenewLeaseInput{
+		TaskID:   taskID,
+		WorkerID: req.WorkerID,
+		Attempt:  req.Attempt,
+	})
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, toClaimedTaskResponse(*claimed))
+}
+
+func (s *Server) handleResult(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+
+	taskID, ok := s.pathUUID(w, r, "task_id")
+	if !ok {
+		return
+	}
+	var req resultRequest
+	if err := decodeJSON(r, &req); err != nil {
+		s.writeError(w, r, domain.ErrInvalidInput)
+		return
+	}
+
+	task, err := s.uc.CompleteTask.Execute(ctx, usecase.CompleteTaskInput{
+		TaskID:       taskID,
+		WorkerID:     req.WorkerID,
+		Attempt:      req.Attempt,
+		ResultURI:    req.ResultURI,
+		ResultSHA256: req.ResultSHA256,
+		Metrics:      req.Metrics,
+	})
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, taskResponse{ID: task.ID, JobID: task.JobID, Status: string(task.Status)})
+}
+
+func (s *Server) handleFailure(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+
+	taskID, ok := s.pathUUID(w, r, "task_id")
+	if !ok {
+		return
+	}
+	var req failureRequest
+	if err := decodeJSON(r, &req); err != nil {
+		s.writeError(w, r, domain.ErrInvalidInput)
+		return
+	}
+
+	task, err := s.uc.FailTask.Execute(ctx, usecase.FailTaskInput{
+		TaskID:       taskID,
+		WorkerID:     req.WorkerID,
+		Attempt:      req.Attempt,
+		ErrorCode:    req.ErrorCode,
+		ErrorMessage: req.ErrorMessage,
+		Retryable:    req.Retryable,
+	})
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, taskResponse{ID: task.ID, JobID: task.JobID, Status: string(task.Status)})
+}
+
+func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+
+	jobID, ok := s.pathUUID(w, r, "job_id")
+	if !ok {
+		return
+	}
+	progress, err := s.uc.GetJobStatus.Execute(ctx, jobID)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, toJobProgressResponse(progress))
+}
+
+// --- helpers ---
+
+func (s *Server) reqCtx(r *http.Request) (context.Context, context.CancelFunc) {
+	return context.WithTimeout(r.Context(), s.requestTimeout)
+}
+
+func (s *Server) pathUUID(w http.ResponseWriter, r *http.Request, name string) (uuid.UUID, bool) {
+	id, err := uuid.Parse(r.PathValue(name))
+	if err != nil {
+		s.writeError(w, r, domain.ErrInvalidInput)
+		return uuid.Nil, false
+	}
+	return id, true
+}
diff --git a/coordinator/internal/transport/http/middleware.go b/coordinator/internal/transport/http/middleware.go
new file mode 100644
index 0000000..feb3f4f
--- /dev/null
+++ b/coordinator/internal/transport/http/middleware.go
@@ -0,0 +1,103 @@
+package http
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/subtle"
+	"encoding/hex"
+	"log/slog"
+	"net/http"
+	"strings"
+	"time"
+)
+
+type ctxKey string
+
+const requestIDKey ctxKey = "request_id"
+
+// withRequestID stamps every request with an ID for correlated logs and error
+// bodies. It wraps the auth middleware rather than the other way round, so even
+// a rejected request carries an ID the caller can quote in a bug report.
+func withRequestID(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		id := newRequestID()
+		w.Header().Set("X-Request-ID", id)
+		next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), requestIDKey, id)))
+	})
+}
+
+func requestIDFrom(ctx context.Context) string {
+	if v, ok := ctx.Value(requestIDKey).(string); ok {
+		return v
+	}
+	return ""
+}
+
+func newRequestID() string {
+	var b [8]byte
+	_, _ = rand.Read(b[:])
+	return hex.EncodeToString(b[:])
+}
+
+// withAuth enforces the shared bearer token every worker presents.
+// An empty token disables the check (local development only).
+func withAuth(token string) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if token == "" {
+				next.ServeHTTP(w, r)
+				return
+			}
+			presented := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
+			// Constant-time compare: a byte-by-byte early exit would let an
+			// attacker recover the token by timing responses.
+			if subtle.ConstantTimeCompare([]byte(presented), []byte(token)) != 1 {
+				w.Header().Set("WWW-Authenticate", "Bearer")
+				writeJSON(w, http.StatusUnauthorized, errorResponse{
+					Error:     "unauthorized",
+					RequestID: requestIDFrom(r.Context()),
+				})
+				return
+			}
+			next.ServeHTTP(w, r)
+		})
+	}
+}
+
+// statusRecorder captures the status code for the access log.
+type statusRecorder struct {
+	http.ResponseWriter
+	status int
+}
+
+func (s *statusRecorder) WriteHeader(code int) {
+	s.status = code
+	s.ResponseWriter.WriteHeader(code)
+}
+
+// withAccessLog records one structured line per request — the minimum needed to
+// debug a distributed system after the fact.
+func withAccessLog(log *slog.Logger) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			start := time.Now()
+			rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK}
+			next.ServeHTTP(rec, r)
+			log.Info("request",
+				"request_id", requestIDFrom(r.Context()),
+				"method", r.Method,
+				"path", r.URL.Path,
+				"status", rec.status,
+				"duration_ms", time.Since(start).Milliseconds(),
+			)
+		})
+	}
+}
+
+// chain applies middleware so that the first argument is the outermost layer.
+func chain(h http.Handler, mw ...func(http.Handler) http.Handler) http.Handler {
+	for i := len(mw) - 1; i >= 0; i-- {
+		h = mw[i](h)
+	}
+	return h
+}
diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go
new file mode 100644
index 0000000..d2731a3
--- /dev/null
+++ b/coordinator/internal/transport/http/server.go
@@ -0,0 +1,60 @@
+// Package http adapts the use-case layer to HTTP. Handlers decode requests,
+// map them onto use-case inputs, and translate results and errors back — no
+// business rules live here.
+package http
+
+import (
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+// UseCases collects everything the transport needs. Depending on concrete
+// use-case types (not one fat interface) keeps each handler's dependency
+// explicit and the wiring visible in the composition root.
+type UseCases struct {
+	CreateJob    *usecase.CreateJob
+	ClaimTask    *usecase.ClaimTask
+	RenewLease   *usecase.RenewLease
+	CompleteTask *usecase.CompleteTask
+	FailTask     *usecase.FailTask
+	GetJobStatus *usecase.GetJobStatus
+}
+
+type Server struct {
+	uc             UseCases
+	log            *slog.Logger
+	requestTimeout time.Duration
+}
+
+func NewServer(uc UseCases, log *slog.Logger, requestTimeout time.Duration) *Server {
+	return &Server{uc: uc, log: log, requestTimeout: requestTimeout}
+}
+
+// Handler builds the router. Go 1.22's ServeMux matches on method and path
+// wildcards, so no third-party router is needed.
+func (s *Server) Handler(token string) http.Handler {
+	protected := http.NewServeMux()
+	protected.HandleFunc("POST /jobs", s.handleCreateJob)
+	protected.HandleFunc("GET /jobs/{job_id}", s.handleGetJob)
+	protected.HandleFunc("POST /tasks/claim", s.handleClaim)
+	protected.HandleFunc("POST /tasks/{task_id}/heartbeat", s.handleHeartbeat)
+	protected.HandleFunc("POST /tasks/{task_id}/result", s.handleResult)
+	protected.HandleFunc("POST /tasks/{task_id}/failure", s.handleFailure)
+
+	mux := http.NewServeMux()
+	// A more specific pattern wins, so /health stays outside the auth wall.
+	mux.HandleFunc("GET /health", s.handleHealth)
+	mux.Handle("/", chain(protected,
+		withRequestID,        // outermost: every response gets an ID,
+		withAccessLog(s.log), // including the 401s below
+		withAuth(token),
+	))
+	return mux
+}
+
+func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
+}
diff --git a/coordinator/internal/usecase/dto.go b/coordinator/internal/usecase/dto.go
new file mode 100644
index 0000000..d7cb76c
--- /dev/null
+++ b/coordinator/internal/usecase/dto.go
@@ -0,0 +1,51 @@
+package usecase
+
+import "github.com/google/uuid"
+
+// Use-case boundary types. Adapters map their wire formats onto these, so the
+// HTTP shape can change without touching business code.
+
+type CreateJobInput struct {
+	Workload   string
+	InputURI   string
+	Parameters map[string]any
+	Chunks     []ChunkInput
+}
+
+type ChunkInput struct {
+	ChunkIndex  int
+	Workload    string
+	InputURI    string
+	InputSHA256 string
+	Parameters  map[string]any
+	MaxAttempts int
+}
+
+type ClaimTaskInput struct {
+	WorkerID  string
+	Workloads []string
+}
+
+type RenewLeaseInput struct {
+	TaskID   uuid.UUID
+	WorkerID string
+	Attempt  int
+}
+
+type CompleteTaskInput struct {
+	TaskID       uuid.UUID
+	WorkerID     string
+	Attempt      int
+	ResultURI    string
+	ResultSHA256 string
+	Metrics      map[string]any
+}
+
+type FailTaskInput struct {
+	TaskID       uuid.UUID
+	WorkerID     string
+	Attempt      int
+	ErrorCode    string
+	ErrorMessage string
+	Retryable    bool
+}
diff --git a/coordinator/internal/usecase/job.go b/coordinator/internal/usecase/job.go
new file mode 100644
index 0000000..1fc455f
--- /dev/null
+++ b/coordinator/internal/usecase/job.go
@@ -0,0 +1,174 @@
+package usecase
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+// Job operations: the submitter-facing lifecycle of a whole submission.
+//
+//	CreateJob    register a job and fan it out into tasks
+//	GetJobStatus aggregate progress
+//	ListResults  completed manifests, ordered for the stitcher
+//	StitchJob    merge partial results into the final artifact
+
+// --- CreateJob -----------------------------------------------------------
+
+type CreateJob struct {
+	jobs  JobRepository
+	tasks TaskRepository
+	tx    TxManager
+	clock Clock
+}
+
+func NewCreateJob(jobs JobRepository, tasks TaskRepository, tx TxManager, clock Clock) *CreateJob {
+	return &CreateJob{jobs: jobs, tasks: tasks, tx: tx, clock: clock}
+}
+
+// Execute builds the job and its tasks, then writes them in one transaction.
+// The all-or-none guarantee comes from TxManager: a half-created job would
+// leave chunks no worker could ever complete.
+func (uc *CreateJob) Execute(ctx context.Context, in CreateJobInput) (*domain.Job, error) {
+	chunks := make([]domain.ChunkSpec, 0, len(in.Chunks))
+	for _, c := range in.Chunks {
+		chunks = append(chunks, domain.ChunkSpec(c))
+	}
+
+	job, tasks, err := domain.NewJobWithTasks(in.Workload, in.InputURI, in.Parameters, chunks, uc.clock.Now())
+	if err != nil {
+		return nil, err
+	}
+
+	err = uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		if err := uc.jobs.Insert(ctx, job); err != nil {
+			return err
+		}
+		return uc.tasks.InsertBatch(ctx, tasks)
+	})
+	if err != nil {
+		return nil, err
+	}
+	return job, nil
+}
+
+// --- GetJobStatus --------------------------------------------------------
+
+type GetJobStatus struct {
+	jobs  JobRepository
+	tasks TaskRepository
+}
+
+func NewGetJobStatus(jobs JobRepository, tasks TaskRepository) *GetJobStatus {
+	return &GetJobStatus{jobs: jobs, tasks: tasks}
+}
+
+func (uc *GetJobStatus) Execute(ctx context.Context, jobID uuid.UUID) (domain.JobProgress, error) {
+	job, err := uc.jobs.Get(ctx, jobID)
+	if err != nil {
+		return domain.JobProgress{}, err
+	}
+	counts, err := uc.tasks.CountByStatus(ctx, jobID)
+	if err != nil {
+		return domain.JobProgress{}, err
+	}
+	return progressFrom(*job, counts), nil
+}
+
+// --- ListResults ---------------------------------------------------------
+
+type ListResults struct {
+	tasks TaskRepository
+}
+
+func NewListResults(tasks TaskRepository) *ListResults {
+	return &ListResults{tasks: tasks}
+}
+
+// Execute preserves chunk_index order: the stitcher merges these into one
+// artifact, and a non-deterministic order would make the final result depend on
+// which worker happened to finish first.
+func (uc *ListResults) Execute(ctx context.Context, jobID uuid.UUID) ([]domain.ResultManifest, error) {
+	tasks, err := uc.tasks.ListCompleted(ctx, jobID)
+	if err != nil {
+		return nil, err
+	}
+
+	manifests := make([]domain.ResultManifest, 0, len(tasks))
+	for _, t := range tasks {
+		if t.ResultURI == nil || t.ResultSHA256 == nil {
+			continue // a completed task always carries both; skip defensively
+		}
+		manifests = append(manifests, domain.ResultManifest{
+			TaskID:       t.ID,
+			ChunkIndex:   t.ChunkIndex,
+			ResultURI:    *t.ResultURI,
+			ResultSHA256: *t.ResultSHA256,
+			Metrics:      t.Metrics,
+		})
+	}
+	return manifests, nil
+}
+
+// --- StitchJob -----------------------------------------------------------
+
+// StitchJob merges every chunk's partial result into the job's final artifact.
+// For similarity search that means concatenating each worker's local top-k,
+// sorting by similarity, and keeping the global top-k — the distributed result
+// must match what a single local run would produce.
+type StitchJob struct {
+	results *ListResults
+}
+
+func NewStitchJob(results *ListResults) *StitchJob {
+	return &StitchJob{results: results}
+}
+
+// Execute returns the URI of the assembled artifact.
+//
+// TODO(phase 6): fetch each manifest's CSV, merge, and persist the result.
+func (uc *StitchJob) Execute(ctx context.Context, jobID uuid.UUID) (string, error) {
+	if _, err := uc.results.Execute(ctx, jobID); err != nil {
+		return "", err
+	}
+	return "", ErrNotImplemented
+}
+
+// --- shared helpers ------------------------------------------------------
+
+// progressFrom turns a status histogram into the domain's progress view.
+func progressFrom(job domain.Job, counts map[domain.TaskStatus]int) domain.JobProgress {
+	p := domain.JobProgress{
+		Job:     job,
+		Pending: counts[domain.TaskPending],
+		Leased:  counts[domain.TaskLeased],
+		Done:    counts[domain.TaskCompleted],
+		Failed:  counts[domain.TaskFailed],
+	}
+	for _, n := range counts {
+		p.Total += n
+	}
+	return p
+}
+
+// syncJobStatus recomputes a job's status from its task counts and persists it.
+// Shared by CompleteTask and FailTask so both close a job by the same rule —
+// the rule itself lives in domain.JobProgress.DeriveStatus.
+func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository,
+	jobID uuid.UUID, now time.Time) error {
+
+	counts, err := tasks.CountByStatus(ctx, jobID)
+	if err != nil {
+		return err
+	}
+	status := progressFrom(domain.Job{}, counts).DeriveStatus()
+
+	var completedAt *time.Time
+	if status == domain.JobCompleted || status == domain.JobFailed {
+		completedAt = &now
+	}
+	return jobs.UpdateStatus(ctx, jobID, status, completedAt)
+}
diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go
new file mode 100644
index 0000000..af0e06d
--- /dev/null
+++ b/coordinator/internal/usecase/ports.go
@@ -0,0 +1,80 @@
+// Package usecase holds the application's business operations. Each use case is
+// a small type with its dependencies injected and a single Execute method.
+//
+// The interfaces below are *ports*: they are declared here, by the consumer,
+// and implemented further out in storage/postgres. That is what keeps the
+// dependency rule intact — usecase never imports storage or transport.
+package usecase
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+// ClaimFilter narrows which task a worker may be handed.
+type ClaimFilter struct {
+	Workloads  []string // workloads this worker can execute
+	Owner      string   // worker ID taking the lease
+	Now        time.Time
+	LeaseUntil time.Time
+}
+
+// TaskRepository persists tasks.
+//
+// ClaimNext is deliberately coarse: leasing must be a single atomic statement
+// (SELECT ... FOR UPDATE SKIP LOCKED + UPDATE), so it cannot be decomposed into
+// Get+Update without losing the guarantee that one task goes to one worker.
+type TaskRepository interface {
+	// ClaimNext atomically leases one matching pending task.
+	// Returns (nil, nil) when nothing is available.
+	ClaimNext(ctx context.Context, f ClaimFilter) (*domain.Task, error)
+
+	// GetForUpdate reads a task and locks its row for the enclosing
+	// transaction, so read-modify-write use cases stay serialized.
+	GetForUpdate(ctx context.Context, id uuid.UUID) (*domain.Task, error)
+
+	// Update persists a mutated task, honouring its Version for optimistic
+	// concurrency.
+	Update(ctx context.Context, t *domain.Task) error
+
+	InsertBatch(ctx context.Context, tasks []*domain.Task) error
+
+	// ListCompleted returns completed tasks ordered by chunk_index.
+	ListCompleted(ctx context.Context, jobID uuid.UUID) ([]*domain.Task, error)
+
+	// CountByStatus aggregates a job's tasks for progress reporting.
+	CountByStatus(ctx context.Context, jobID uuid.UUID) (map[domain.TaskStatus]int, error)
+
+	// ExpireLeases applies the lease-expiry rule to every elapsed task and
+	// reports how many were affected.
+	ExpireLeases(ctx context.Context, now time.Time) (int64, error)
+}
+
+// JobRepository persists jobs.
+type JobRepository interface {
+	Insert(ctx context.Context, j *domain.Job) error
+	Get(ctx context.Context, id uuid.UUID) (*domain.Job, error)
+	UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error
+}
+
+// TxManager runs a function inside one database transaction. The transaction
+// travels in the context, so repositories pick it up without this port ever
+// mentioning pgx.
+type TxManager interface {
+	WithinTx(ctx context.Context, fn func(ctx context.Context) error) error
+}
+
+// Clock supplies the current time. Injecting it keeps lease and expiry rules
+// testable without sleeping or freezing the system clock.
+type Clock interface {
+	Now() time.Time
+}
+
+// ErrNotImplemented marks scaffold code with no body yet. Unlike the errors in
+// domain, it describes the state of this codebase, not a business rule.
+var ErrNotImplemented = errors.New("not implemented")
diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go
new file mode 100644
index 0000000..c9dc801
--- /dev/null
+++ b/coordinator/internal/usecase/task.go
@@ -0,0 +1,206 @@
+package usecase
+
+import (
+	"context"
+	"time"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+// Task operations: the worker-facing lifecycle of a single chunk.
+//
+//	ClaimTask    lease the next available task
+//	RenewLease   extend a held lease (heartbeat)
+//	CompleteTask record a successful result
+//	FailTask     record a failure
+//	ExpireLeases reclaim leases that elapsed without a heartbeat
+
+// --- ClaimTask -----------------------------------------------------------
+
+type ClaimTask struct {
+	tasks         TaskRepository
+	clock         Clock
+	leaseDuration time.Duration
+}
+
+func NewClaimTask(tasks TaskRepository, clock Clock, leaseDuration time.Duration) *ClaimTask {
+	return &ClaimTask{tasks: tasks, clock: clock, leaseDuration: leaseDuration}
+}
+
+// Execute reclaims elapsed leases first, then hands out one task.
+//
+// Sweeping before claiming matters: otherwise a task abandoned by a dead worker
+// stays invisible until the reaper's next tick, and a waiting worker is told the
+// queue is empty while work sits idle.
+//
+// This use case is thin by design — the atomicity that makes claiming correct
+// lives in one SQL statement behind ClaimNext, and splitting it across the layer
+// boundary would break it.
+func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.ClaimedTask, error) {
+	if in.WorkerID == "" {
+		return nil, domain.ErrInvalidInput
+	}
+	now := uc.clock.Now()
+
+	if _, err := uc.tasks.ExpireLeases(ctx, now); err != nil {
+		return nil, err
+	}
+
+	task, err := uc.tasks.ClaimNext(ctx, ClaimFilter{
+		Workloads:  in.Workloads,
+		Owner:      in.WorkerID,
+		Now:        now,
+		LeaseUntil: now.Add(uc.leaseDuration),
+	})
+	if err != nil {
+		return nil, err
+	}
+	if task == nil {
+		return nil, nil // empty queue is a normal state, not an error
+	}
+
+	claimed := task.AsClaimed()
+	return &claimed, nil
+}
+
+// --- RenewLease ----------------------------------------------------------
+
+type RenewLease struct {
+	tasks         TaskRepository
+	tx            TxManager
+	clock         Clock
+	leaseDuration time.Duration
+}
+
+func NewRenewLease(tasks TaskRepository, tx TxManager, clock Clock, leaseDuration time.Duration) *RenewLease {
+	return &RenewLease{tasks: tasks, tx: tx, clock: clock, leaseDuration: leaseDuration}
+}
+
+// Execute is a read-modify-write, so it runs inside a transaction with the row
+// locked: two concurrent heartbeats must not interleave into a lost update.
+// Whether the caller may renew at all is decided by the entity, not here.
+func (uc *RenewLease) Execute(ctx context.Context, in RenewLeaseInput) (*domain.ClaimedTask, error) {
+	var claimed domain.ClaimedTask
+
+	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		task, err := uc.tasks.GetForUpdate(ctx, in.TaskID)
+		if err != nil {
+			return err
+		}
+		if err := task.RenewLease(in.WorkerID, in.Attempt, uc.clock.Now().Add(uc.leaseDuration)); err != nil {
+			return err
+		}
+		if err := uc.tasks.Update(ctx, task); err != nil {
+			return err
+		}
+		claimed = task.AsClaimed()
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &claimed, nil
+}
+
+// --- CompleteTask --------------------------------------------------------
+
+type CompleteTask struct {
+	tasks TaskRepository
+	jobs  JobRepository
+	tx    TxManager
+	clock Clock
+}
+
+func NewCompleteTask(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *CompleteTask {
+	return &CompleteTask{tasks: tasks, jobs: jobs, tx: tx, clock: clock}
+}
+
+// Execute applies the result and, when that was the job's last outstanding
+// task, closes the job in the same transaction — so a caller who sees a
+// completed task never observes its job still marked running.
+//
+// Lease ownership, staleness, and idempotent replays are all decided by
+// Task.CompleteWith; this use case only orchestrates.
+func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*domain.Task, error) {
+	var out *domain.Task
+
+	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		task, err := uc.tasks.GetForUpdate(ctx, in.TaskID)
+		if err != nil {
+			return err
+		}
+		now := uc.clock.Now()
+		if err := task.CompleteWith(in.ResultURI, in.ResultSHA256, in.Metrics,
+			in.WorkerID, in.Attempt, now); err != nil {
+			return err
+		}
+		if err := uc.tasks.Update(ctx, task); err != nil {
+			return err
+		}
+		out = task
+		return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now)
+	})
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// --- FailTask ------------------------------------------------------------
+
+type FailTask struct {
+	tasks TaskRepository
+	jobs  JobRepository
+	tx    TxManager
+	clock Clock
+}
+
+func NewFailTask(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *FailTask {
+	return &FailTask{tasks: tasks, jobs: jobs, tx: tx, clock: clock}
+}
+
+// Execute delegates the requeue-or-terminate decision to Task.Fail, then keeps
+// the parent job's status consistent in the same transaction.
+func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task, error) {
+	var out *domain.Task
+
+	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		task, err := uc.tasks.GetForUpdate(ctx, in.TaskID)
+		if err != nil {
+			return err
+		}
+		now := uc.clock.Now()
+		if err := task.Fail(in.WorkerID, in.Attempt, in.ErrorCode, in.ErrorMessage, in.Retryable, now); err != nil {
+			return err
+		}
+		if err := uc.tasks.Update(ctx, task); err != nil {
+			return err
+		}
+		out = task
+		return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now)
+	})
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// --- ExpireLeases --------------------------------------------------------
+
+type ExpireLeases struct {
+	tasks TaskRepository
+	clock Clock
+}
+
+func NewExpireLeases(tasks TaskRepository, clock Clock) *ExpireLeases {
+	return &ExpireLeases{tasks: tasks, clock: clock}
+}
+
+// Execute reports how many tasks were reclaimed.
+//
+// The sweep is one set-based statement rather than a load-decide-save loop:
+// several coordinators run it concurrently, and a single atomic UPDATE makes
+// the duplicate work harmless — the loser simply updates 0 rows.
+func (uc *ExpireLeases) Execute(ctx context.Context) (int64, error) {
+	return uc.tasks.ExpireLeases(ctx, uc.clock.Now())
+}
diff --git a/coordinator/migrations/0001_init.down.sql b/coordinator/migrations/0001_init.down.sql
new file mode 100644
index 0000000..78df969
--- /dev/null
+++ b/coordinator/migrations/0001_init.down.sql
@@ -0,0 +1,8 @@
+BEGIN;
+
+DROP TABLE IF EXISTS tasks;
+DROP TABLE IF EXISTS jobs;
+DROP TYPE  IF EXISTS task_status;
+DROP TYPE  IF EXISTS job_status;
+
+COMMIT;
diff --git a/coordinator/migrations/0001_init.up.sql b/coordinator/migrations/0001_init.up.sql
new file mode 100644
index 0000000..c84c469
--- /dev/null
+++ b/coordinator/migrations/0001_init.up.sql
@@ -0,0 +1,58 @@
+BEGIN;
+
+CREATE TYPE job_status  AS ENUM ('pending','running','completed','failed','cancelled');
+CREATE TYPE task_status AS ENUM ('pending','leased','completed','failed','cancelled');
+
+-- One user submission, possibly split into several tasks.
+CREATE TABLE jobs (
+    id           uuid        PRIMARY KEY,
+    workload     text        NOT NULL,
+    input_uri    text        NOT NULL,
+    parameters   jsonb       NOT NULL DEFAULT '{}'::jsonb,
+    status       job_status  NOT NULL DEFAULT 'pending',
+    created_at   timestamptz NOT NULL DEFAULT now(),
+    completed_at timestamptz
+);
+
+-- One independently executable chunk.
+CREATE TABLE tasks (
+    id               uuid        PRIMARY KEY,
+    job_id           uuid        NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+    chunk_index      integer     NOT NULL,
+    workload         text        NOT NULL,
+    input_uri        text        NOT NULL,
+    input_sha256     text        NOT NULL,
+    parameters       jsonb       NOT NULL DEFAULT '{}'::jsonb,
+    status           task_status NOT NULL DEFAULT 'pending',
+    attempt          integer     NOT NULL DEFAULT 0,
+    max_attempts     integer     NOT NULL DEFAULT 3,
+    lease_owner      text,
+    lease_expires_at timestamptz,
+    result_uri       text,
+    result_sha256    text,
+    metrics          jsonb,
+    error_code       text,
+    error_message    text,
+    created_at       timestamptz NOT NULL DEFAULT now(),
+    started_at       timestamptz,
+    completed_at     timestamptz,
+    version          integer     NOT NULL DEFAULT 0,
+
+    CONSTRAINT uq_tasks_job_chunk    UNIQUE (job_id, chunk_index),
+    CONSTRAINT ck_tasks_attempt      CHECK (attempt >= 0),
+    CONSTRAINT ck_tasks_max_attempts CHECK (max_attempts > 0),
+    -- A completed task must carry its result manifest.
+    CONSTRAINT ck_tasks_completed_result CHECK (
+        status <> 'completed' OR (result_uri IS NOT NULL AND result_sha256 IS NOT NULL)
+    ),
+    -- A leased task must carry its lease.
+    CONSTRAINT ck_tasks_leased_owner CHECK (
+        status <> 'leased' OR (lease_owner IS NOT NULL AND lease_expires_at IS NOT NULL)
+    )
+);
+
+-- Claim path: find the oldest pending task fast.
+CREATE INDEX ix_tasks_claim ON tasks (status, lease_expires_at, created_at);
+CREATE INDEX ix_tasks_job   ON tasks (job_id);
+
+COMMIT;