feat(master): store service inventory snapshots and web sessions

feat(master): postgres store with migrations, master identity and agents
feat(agent): add updater-agent binary and ship it in the image
2026-10-12 14:45:46 +02:00 · 2026-10-12 14:43:34 +02:00 · 2026-10-12 14:41:05 +02:00 · 2026-10-12 14:38:14 +02:00 · 2026-10-12 14:36:11 +02:00 · 2026-10-12 14:34:13 +02:00
54 changed files with 9922 additions and 18 deletions
@@ -24,3 +24,4 @@ coverage.html

 # Serena MCP workspace
 .serena/
+.superpowers/
@@ -0,0 +1,88 @@
+# package-updater — Claude context
+
+Webhook-driven Docker Compose updater. Go service, single binary, deployed as
+a container with `/var/run/docker.sock` mounted.
+
+See `README.md` for user-facing docs. This file is for working ON the code.
+
+## Commands
+
+```bash
+go test ./...                          # all tests, fast (no docker required)
+go test -run TestFoo ./internal/api    # single test
+go build ./cmd/server                  # produces ./server
+docker build -t package-updater:dev .  # multi-stage, builds golang:1.26-alpine
+
+# Local smoke test (builds from Dockerfile, runs against /var/run/docker.sock):
+cp env.sample .env && $EDITOR .env     # fill UPDATER_API_KEY
+docker compose -f docker-compose.local.yml up --build
+curl -sH "Authorization: Bearer $UPDATER_API_KEY" \
+  -d '{"image":"foo"}' http://localhost:8080/update | jq
+```
+
+## Architecture (one-line per package)
+
+- `cmd/server`        — wiring only: config → docker client → discovery → queue → http
+- `internal/api`      — HTTP handlers, bearer-token auth, request-id + access-log middleware
+- `internal/config`   — env-var loading; fails fast if `UPDATER_API_KEY` missing
+- `internal/discovery`— given an image, return Compose `Job`s to run (label parsing, path check, dedup)
+- `internal/discovery/swarm.go`  — SwarmDiscovery: same FindJobs signature, lists `docker service ls`, gates on service-level opt-in label instead of STACKS_ROOT path-check
+- `internal/updater`  — FIFO queue + single worker + `docker compose` subprocess executor
+- `internal/updater/swarm_executor.go` — SwarmExecutor: `docker service update --image` via the Docker API instead of a `docker compose` subprocess
+- `internal/selfupdate` — flush-then-exec wrapper for updating ourselves (NOT wired in live, see gotcha)
+- `internal/metrics`  — Prometheus collectors
+- `internal/logging`  — slog JSON, request-id context propagation
+
+## Design intent (do not break without discussion)
+
+- **Single FIFO worker by design** (`internal/updater/queue.go`). Spec §5.7. Never
+  parallelise the queue — two `docker compose` calls against the same stack race.
+- **Defense in depth: token AND opt-in label AND STACKS_ROOT prefix** must all
+  hold before a container is touched. Weakening any of these breaks the security
+  model — discuss before changing.
+- **Auth on `/update` only**. `/healthz`, `/metrics`, `/version` are intentionally
+  unauthenticated (internal network, scraper/healthcheck need them). See
+  `cmd/server/main.go` `routeAuth`.
+- **`metrics *Metrics` parameters may be nil**; constructors and call sites check.
+  Tests rely on this — don't drop the nil-check.
+- **Stateless**: no DB, no config file, no on-disk audit log. Docker daemon is
+  the source of truth.
+- **Compose mode and Swarm mode are selected once per deployment via `MODE`**, never
+  mixed at request time. Swarm mode's security gate is opt-in label only — there is
+  no STACKS_ROOT-equivalent path check, since Swarm services have no local compose
+  file. Don't add one; don't weaken Compose mode's three-factor gate to match.
+
+## Gotchas
+
+- **`internal/selfupdate.Wrapped` is implemented and unit-tested but NOT wired
+  into the live queue.** Spec §15 calls this out as a v1 gap. If you "fix" this,
+  read the package doc — flush-before-exec ordering is subtle.
+- **Image matching is case-sensitive and tag/digest-agnostic.** The tricky case
+  is `localhost:5000/foo:v1` where the first colon is a port, not a tag. See
+  `internal/discovery/matching.go` `NormaliseImage`.
+- **Path safety uses `filepath.Rel` + ".." prefix check**, NOT `strings.HasPrefix`.
+  Prevents the `/foo-evil` vs `/foo` confusion. See `internal/discovery/pathcheck.go`.
+- **`STACKS_ROOT` defaults to `/home/shcizo/self-hosted`** in prod but `env.sample`
+  suggests `/tmp` for smoke tests. The default in `config.go` is the prod value
+  — set explicitly in test/dev envs.
+- **Go 1.26.3** (`go.mod`). Dockerfile pins `golang:1.26-alpine`. Bumping one
+  without the other has caused a fix commit already.
+- **`<summary>` C# convention from global CLAUDE.md does not apply here** — this
+  is Go. Use idiomatic GoDoc (`// FuncName does X.`).
+- **Swarm mode requires manager-node API access.** `docker service update` fails
+  with a permission error against a worker-only node. This is an operator/deployment
+  concern (point `DOCKER_HOST` at a manager, or schedule the updater on a manager),
+  not something the code can detect or work around.
+
+## References
+
+- Design spec: `docs/superpowers/specs/2026-05-22-package-updater-design.md` (489 lines, authoritative)
+- Implementation plan: `docs/superpowers/plans/2026-05-22-package-updater-implementation.md`
+- Consumer-side CI integration: `gitea-action/`
+
+## Conventions
+
+- Conventional Commits (`feat:`, `fix:`, `docs:`, `ci:`, `build:`, `chore:`).
+- One package = one responsibility; interfaces defined at consumer site
+  (`api.Finder`, `api.Submitter`, `api.Pinger`) for testability.
+- Table-driven tests with `stretchr/testify`. No mocks beyond hand-written fakes.
@@ -8,15 +8,17 @@ COPY . .
 ARG VERSION=dev
 ARG COMMIT=unknown
 ARG BUILD_TIME=unknown
-RUN CGO_ENABLED=0 go build \
-    -ldflags="-s -w \
-      -X main.version=${VERSION} \
-      -X main.commit=${COMMIT} \
-      -X main.buildTime=${BUILD_TIME}" \
-    -o /out/package-updater ./cmd/server
+RUN for cmd in server:package-updater agent:updater-agent; do \
+      CGO_ENABLED=0 go build \
+        -ldflags="-s -w \
+          -X main.version=${VERSION} \
+          -X main.commit=${COMMIT} \
+          -X main.buildTime=${BUILD_TIME}" \
+        -o /out/${cmd#*:} ./cmd/${cmd%%:*} || exit 1; \
+    done

 FROM alpine:3.20
 RUN apk add --no-cache docker-cli docker-cli-compose ca-certificates wget
-COPY --from=build /out/package-updater /usr/local/bin/
+COPY --from=build /out/ /usr/local/bin/
 EXPOSE 8080
 ENTRYPOINT ["/usr/local/bin/package-updater"]
@@ -27,6 +27,25 @@ A container is eligible for update only if it has **both**:

 Defense in depth: a valid bearer token AND the opt-in label must both be present before any container is touched.

+## Swarm mode
+
+Set `MODE=swarm` to update Docker Swarm services instead of Compose stacks. The
+service runs `docker service update --image` via the Docker API instead of
+shelling out to `docker compose`.
+
+Two things to get right when running in Swarm mode:
+
+- **The opt-in label goes on the service, not the container/task.** Swarm mode
+  reads `Service.Spec.Labels`, so add it with
+  `docker service update --label-add se.shcizo.auto-update=true <service>` or
+  set it under `deploy.labels` in the stack file — a plain `labels:` entry on
+  the service (container-level) is not visible to Swarm mode's discovery.
+- **The updater must talk to a Swarm manager.** `docker service update`
+  requires manager API access, so either point `DOCKER_HOST` at a manager node
+  or schedule the updater container itself on a manager with the socket
+  mounted. This is a deployment concern the service cannot detect or work
+  around.
+
 ## Quick start

 1. Build and push the image (e.g. via your own CI).
@@ -49,6 +68,7 @@ All via environment variables.
 | `LOG_LEVEL` | no | `info` | `debug` / `info` / `warn` / `error`. |
 | `UPDATE_TIMEOUT` | no | `5m` | Per-job timeout (Go duration). |
 | `OPT_IN_LABEL` | no | `se.shcizo.auto-update` | Label name to check; value must equal `"true"`. |
+| `MODE` | no | `compose` | `compose` or `swarm`. Selects the update mechanism for the whole deployment; not mixed per-request. |

 ## Endpoints

@@ -80,5 +100,7 @@ These are tracked in the spec's section 2 and section 15 as deliberate out-of-sc

 - **Self-update wiring**: `internal/selfupdate.Wrapped` exists and is unit-tested but is not wired into the live queue. The HTTP response flush ordering for self-replacement is a future enhancement; for now, expect to manually rerun `docker compose up -d` on the host if pushing a new image of `package-updater` itself causes a mid-response interruption.
 - **No rollback**: Compose's "keep old container if new fails to start" is the only safety net.
- **Single host only**.
+- **Single host only in Compose mode**. Swarm mode (`MODE=swarm`) is the
+  multi-node path, but only from a manager node's point of view — the updater
+  itself still needs manager API access (see "Swarm mode" above).
 - **No per-repo API keys**: a single shared bearer token is used.
@@ -0,0 +1,89 @@
+// Command updater-agent reports this machine's Docker services to the
+// package-updater master.
+package main
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/docker/docker/client"
+
+	"github.com/shcizo/package-updater/internal/agent/config"
+	"github.com/shcizo/package-updater/internal/agent/conn"
+	"github.com/shcizo/package-updater/internal/agent/inventory"
+	"github.com/shcizo/package-updater/internal/logging"
+)
+
+// Build info is injected at link time via -ldflags="-X main.version=...".
+var version = "dev"
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintln(os.Stderr, "fatal:", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	cfg, err := config.Load()
+	if err != nil {
+		return err
+	}
+	logger := logging.New(cfg.LogLevel)
+
+	dockerCli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
+	if err != nil {
+		return fmt.Errorf("docker client: %w", err)
+	}
+	defer dockerCli.Close()
+
+	collector := inventory.New(dockerCli, cfg.SelfID)
+	agent := conn.New(conn.Config{
+		URL: cfg.MasterURL, Token: cfg.Token, Fingerprint: cfg.MasterKey, Version: version,
+	}, collector, logger)
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	go watchLoop(ctx, dockerCli, agent, logger)
+	logger.Info("agent_starting", "version", version, "master_url", cfg.MasterURL)
+	return agent.Run(ctx)
+}
+
+// watchLoop keeps a Docker event subscription alive. When the stream breaks
+// (daemon restart), it resubscribes and asks for a refresh, because events
+// may have been missed in between. It also asks for a full re-inventory every
+// 60 seconds, since a Swarm replica dying on another node emits no event here.
+func watchLoop(ctx context.Context, src inventory.EventSource, agent *conn.Agent, log *slog.Logger) {
+	go func() {
+		ticker := time.NewTicker(60 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				agent.Notify()
+			}
+		}
+	}()
+
+	for {
+		err := inventory.Watch(ctx, src, 2*time.Second, agent.Notify)
+		if ctx.Err() != nil {
+			return
+		}
+		log.Warn("docker_events_lost", "error", err)
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(5 * time.Second):
+		}
+		agent.Notify()
+	}
+}
@@ -42,7 +42,7 @@ func run() error {
 	logger := logging.New(cfg.LogLevel)
 	logger.Info("starting",
 		"version", version, "commit", commit, "port", cfg.Port,
-		"stacks_root", cfg.StacksRoot, "opt_in_label", cfg.OptInLabel,
+		"mode", cfg.Mode, "stacks_root", cfg.StacksRoot, "opt_in_label", cfg.OptInLabel,
 	)

 	dockerCli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
@@ -51,8 +51,21 @@ func run() error {
 	}
 	defer dockerCli.Close()

-	disc := discovery.New(dockerCli, cfg.StacksRoot, cfg.OptInLabel)
-	exec := updater.NewComposeExecutor()
+	var finder api.Finder
+	var exec updater.Executor
+	switch cfg.Mode {
+	case "swarm":
+		finder = discovery.NewSwarm(dockerCli, cfg.OptInLabel)
+		exec = updater.NewSwarmExecutor(dockerCli)
+	case "compose":
+		finder = discovery.New(dockerCli, cfg.StacksRoot, cfg.OptInLabel)
+		exec = updater.NewComposeExecutor()
+	default:
+		// Unreachable: config.Load already validates Mode, but guard
+		// here too rather than silently falling through to a nil
+		// finder/exec pair.
+		return fmt.Errorf("unknown MODE %q", cfg.Mode)
+	}

 	reg := prometheus.NewRegistry()
 	m := metrics.New(reg)
@@ -62,7 +75,7 @@ func run() error {
 	queue.Start(context.Background())
 	defer queue.Stop()

-	handlers := api.NewHandlers(disc, &submitterAdapter{queue: queue, timeout: cfg.UpdateTimeout}, dockerCli, version, commit, buildTime, m)
+	handlers := api.NewHandlers(finder, &submitterAdapter{queue: queue, timeout: cfg.UpdateTimeout}, dockerCli, version, commit, buildTime, m)

 	mux := http.NewServeMux()
 	mux.HandleFunc("POST /update", handlers.Update)
@@ -0,0 +1,859 @@
+# Docker Swarm Support Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add a `MODE=swarm` operating mode so package-updater can update Docker Swarm services (`docker service update --image`) as an alternative to the existing Compose-only flow (`docker compose pull && up -d`), selected once per deployment via config — not mixed within one running instance.
+
+**Architecture:** Introduce parallel `SwarmDiscovery` (internal/discovery) and `SwarmExecutor` (internal/updater) implementations alongside the existing `Discovery`/`ComposeExecutor`. Both new types satisfy the exact same `api.Finder` / `updater.Executor` interfaces already in use, so `internal/api` and the FIFO `Queue` need no structural changes — only `cmd/server/main.go` branches on `cfg.Mode` to decide which pair to construct. The Compose-specific security gate (`STACKS_ROOT` path-check) does not apply to Swarm services (no local compose file exists); it is replaced by the same opt-in label mechanism, checked against the Swarm service's own labels instead of container labels.
+
+**Tech Stack:** Go 1.26.3, `github.com/docker/docker` SDK v28.5.2+incompatible (`client`, `api/types`, `api/types/swarm`), `stretchr/testify`.
+
+## Global Constraints
+
+- Go 1.26.3 (go.mod) — do not bump without also bumping the Dockerfile `golang:1.26-alpine` base image.
+- Single FIFO worker in `internal/updater/queue.go` must remain single — never parallelize job execution, including for Swarm jobs.
+- Defense in depth for Compose mode (token + opt-in label + STACKS_ROOT prefix) must keep holding unchanged. Swarm mode's gate is opt-in label only (no filesystem path exists to check) — this is a deliberate, narrower model for Swarm and must not be backported to weaken Compose mode.
+- Auth remains on `/update` only; `/healthz`, `/metrics`, `/version` stay unauthenticated.
+- `metrics *Metrics` parameters may be nil; every new call site must follow the existing `if m.metrics != nil { ... }` pattern — never assume non-nil.
+- Stateless: no DB, no config file, no on-disk audit log required for Swarm mode either.
+- Conventional Commits (`feat:`, `fix:`, `docs:`, `test:`) for every commit in this plan.
+- Table-driven / one-assertion-focus tests with `testify/require`, matching existing style in `internal/discovery/discovery_test.go`.
+
+---
+
+### Task 1: Add `Mode` to config
+
+**Files:**
+- Modify: `internal/config/config.go`
+- Test: `internal/config/config_test.go`
+
+**Interfaces:**
+- Produces: `Config.Mode string` — always one of `"compose"` or `"swarm"` after `Load()` returns successfully. Later tasks (`cmd/server/main.go`) branch on this exact string.
+
+- [ ] **Step 1: Write the failing tests**
+
+Add to `internal/config/config_test.go` (create the file if it doesn't already cover this — check first; if it exists, add these functions alongside the existing ones):
+
+```go
+func TestLoad_DefaultsToComposeMode(t *testing.T) {
+	t.Setenv("UPDATER_API_KEY", "secret")
+	cfg, err := config.Load()
+	require.NoError(t, err)
+	require.Equal(t, "compose", cfg.Mode)
+}
+
+func TestLoad_AcceptsSwarmMode(t *testing.T) {
+	t.Setenv("UPDATER_API_KEY", "secret")
+	t.Setenv("MODE", "swarm")
+	cfg, err := config.Load()
+	require.NoError(t, err)
+	require.Equal(t, "swarm", cfg.Mode)
+}
+
+func TestLoad_RejectsInvalidMode(t *testing.T) {
+	t.Setenv("UPDATER_API_KEY", "secret")
+	t.Setenv("MODE", "kubernetes")
+	_, err := config.Load()
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "MODE")
+}
+```
+
+Make sure the file imports `"testing"`, `"github.com/shcizo/package-updater/internal/config"`, and `"github.com/stretchr/testify/require"` (match whatever the existing `config_test.go` already imports).
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `go test ./internal/config/... -run 'TestLoad_.*Mode' -v`
+Expected: FAIL — `cfg.Mode` does not compile (`Config` has no field `Mode`).
+
+- [ ] **Step 3: Implement**
+
+In `internal/config/config.go`, add the field and validation:
+
+```go
+// Config holds all runtime configuration for the service.
+type Config struct {
+	APIKey        string
+	StacksRoot    string
+	Port          string
+	LogLevel      string
+	UpdateTimeout time.Duration
+	OptInLabel    string
+	Mode          string
+}
+```
+
+In `Load()`, after the existing field assignments (right after `OptInLabel`), add:
+
+```go
+		Mode:       getenvDefault("MODE", "compose"),
+```
+
+so the literal becomes:
+
+```go
+	cfg := &Config{
+		APIKey:     os.Getenv("UPDATER_API_KEY"),
+		StacksRoot: getenvDefault("STACKS_ROOT", "/home/shcizo/self-hosted"),
+		Port:       getenvDefault("PORT", "8080"),
+		LogLevel:   getenvDefault("LOG_LEVEL", "info"),
+		OptInLabel: getenvDefault("OPT_IN_LABEL", "se.shcizo.auto-update"),
+		Mode:       getenvDefault("MODE", "compose"),
+	}
+```
+
+Then, after the `APIKey` check and before the `UPDATE_TIMEOUT` parsing, add mode validation:
+
+```go
+	if cfg.Mode != "compose" && cfg.Mode != "swarm" {
+		return nil, fmt.Errorf("MODE %q is invalid: must be %q or %q", cfg.Mode, "compose", "swarm")
+	}
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `go test ./internal/config/... -v`
+Expected: PASS — all config tests including the three new ones, and no regressions in `TestLoad_RequiresAPIKey`, `_AppliesDefaults`, `_OverridesViaEnv`, `_InvalidTimeoutErrors`.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/config/config.go internal/config/config_test.go
+git commit -m "feat(config): add MODE env var to select compose or swarm operation"
+```
+
+---
+
+### Task 2: Generalize `discovery.Job` for non-Compose jobs
+
+**Files:**
+- Modify: `internal/discovery/discovery.go`
+- Test: `internal/discovery/discovery_test.go`
+
+**Interfaces:**
+- Produces: `discovery.Job` gains two fields, `Image string` (the full image reference the update request asked for — needed later by `SwarmExecutor` to know what to update *to*, since Swarm has no local compose file to `pull` against) and `ServiceID string` (empty for Compose jobs; the Swarm service ID for Swarm jobs, needed later by `SwarmExecutor` to target the right service). Both default to `""` for existing Compose jobs, so no existing caller breaks.
+- Consumes: nothing new — this task only widens the struct and populates `Image` from the existing `FindJobs(ctx, image string)` parameter.
+
+- [ ] **Step 1: Write the failing test**
+
+Add to `internal/discovery/discovery_test.go`:
+
+```go
+func TestFindJobs_PopulatesImageField(t *testing.T) {
+	fake := &fakeDockerClient{containers: []types.Container{
+		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
+			"myapp-prod", "web",
+			"/home/shcizo/self-hosted/myapp-prod",
+			"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml",
+			true,
+		)),
+	}}
+	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Len(t, jobs, 1)
+	require.Equal(t, "registry.example.com/myapp", jobs[0].Image)
+	require.Empty(t, jobs[0].ServiceID)
+}
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `go test ./internal/discovery/... -run TestFindJobs_PopulatesImageField -v`
+Expected: FAIL — `jobs[0].Image` does not compile (`Job` has no field `Image`).
+
+- [ ] **Step 3: Implement**
+
+In `internal/discovery/discovery.go`, widen the struct:
+
+```go
+// Job describes a single (project, service, config_files) update to execute.
+type Job struct {
+	Project     string
+	Service     string
+	WorkingDir  string
+	ConfigFiles []string
+	// Image is the full image reference the update request asked for.
+	// Compose jobs ignore it (compose.yml already pins the reference to
+	// pull); Swarm jobs need it to know what to set on the service spec.
+	Image string
+	// ServiceID is the Swarm service ID. Empty for Compose jobs.
+	ServiceID string
+	// Refused is true when the WorkingDir falls outside STACKS_ROOT.
+	// The job is returned so the caller can surface a per-job "refused"
+	// result, but it MUST NOT be executed.
+	Refused       bool
+	RefusedReason string
+}
+```
+
+In `FindJobs`, populate `Image` on the constructed job (leave `ServiceID` as its zero value):
+
+```go
+		jobs = append(jobs, Job{
+			Project:       cl.Project,
+			Service:       cl.Service,
+			WorkingDir:    cl.WorkingDir,
+			ConfigFiles:   cl.ConfigFiles,
+			Image:         image,
+			Refused:       refused,
+			RefusedReason: reason,
+		})
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `go test ./internal/discovery/... -v`
+Expected: PASS — all discovery tests, including the new one and all pre-existing `TestFindJobs_*` cases (they don't assert on `Image`/`ServiceID` so they're unaffected).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/discovery/discovery.go internal/discovery/discovery_test.go
+git commit -m "feat(discovery): add Image and ServiceID fields to Job for swarm support"
+```
+
+---
+
+### Task 3: `SwarmDiscovery` — find matching Swarm services
+
+**Files:**
+- Create: `internal/discovery/swarm.go`
+- Test: `internal/discovery/swarm_test.go`
+
+**Interfaces:**
+- Consumes: `discovery.Job{Project, Service, Image, ServiceID, Refused, RefusedReason}` (Task 2), `discovery.NormaliseImage(string) string` and `discovery.ImagesMatch(a, b string) bool` (existing, `internal/discovery/matching.go`), `discovery.HasOptIn(labels map[string]string, key string) bool` (existing, `internal/discovery/labels.go`).
+- Produces: `discovery.SwarmDockerClient` interface, `discovery.NewSwarm(cli SwarmDockerClient, optInLabel string) *SwarmDiscovery`, method `(*SwarmDiscovery) FindJobs(ctx context.Context, image string) ([]Job, error)` — same signature as `*Discovery.FindJobs`, so both satisfy `api.Finder` (`internal/api/handlers.go`) without any change there.
+
+- [ ] **Step 1: Verify the Docker SDK method signature before writing code**
+
+Run: `go doc github.com/docker/docker/client.Client.ServiceList`
+
+Expected output shape (confirm it matches before proceeding — if the SDK signature differs, adjust the interface in Step 3 to match what `go doc` actually reports instead of the assumed signature below):
+
+```
+func (cli *Client) ServiceList(ctx context.Context, options types.ServiceListOptions) ([]swarm.Service, error)
+```
+
+Also run: `go doc github.com/docker/docker/api/types/swarm.Service` and `go doc github.com/docker/docker/api/types/swarm.ServiceSpec` to confirm `Service.ID`, `Service.Spec.Name` (via embedded `Annotations`), `Service.Spec.Labels` (via embedded `Annotations`), and `Service.Spec.TaskTemplate.ContainerSpec.Image` are reachable as described. Adjust field access in later steps if the embedding differs.
+
+- [ ] **Step 2: Write the failing tests**
+
+Create `internal/discovery/swarm_test.go`:
+
+```go
+package discovery_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeSwarmDockerClient struct {
+	services []swarm.Service
+	err      error
+}
+
+func (f *fakeSwarmDockerClient) ServiceList(_ context.Context, _ types.ServiceListOptions) ([]swarm.Service, error) {
+	return f.services, f.err
+}
+
+func mkService(id, name, image string, labels map[string]string) swarm.Service {
+	return swarm.Service{
+		ID: id,
+		Spec: swarm.ServiceSpec{
+			Annotations: swarm.Annotations{Name: name, Labels: labels},
+			TaskTemplate: swarm.TaskSpec{
+				ContainerSpec: &swarm.ContainerSpec{Image: image},
+			},
+		},
+	}
+}
+
+func TestSwarmFindJobs_MatchAndOptIn(t *testing.T) {
+	fake := &fakeSwarmDockerClient{services: []swarm.Service{
+		mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", map[string]string{
+			"se.shcizo.auto-update": "true",
+		}),
+		mkService("svc-other", "other_web", "registry.example.com/other:v1", map[string]string{
+			"se.shcizo.auto-update": "true",
+		}),
+	}}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Len(t, jobs, 1)
+	require.Equal(t, "myapp_web", jobs[0].Service)
+	require.Equal(t, "svc-myapp", jobs[0].ServiceID)
+	require.Equal(t, "registry.example.com/myapp", jobs[0].Image)
+	require.False(t, jobs[0].Refused)
+	require.Empty(t, jobs[0].WorkingDir)
+	require.Empty(t, jobs[0].ConfigFiles)
+}
+
+func TestSwarmFindJobs_SkipsWithoutOptIn(t *testing.T) {
+	fake := &fakeSwarmDockerClient{services: []swarm.Service{
+		mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", nil),
+	}}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Empty(t, jobs)
+}
+
+func TestSwarmFindJobs_DockerError(t *testing.T) {
+	fake := &fakeSwarmDockerClient{err: errors.New("connection refused")}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	_, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.Error(t, err)
+}
+
+func TestSwarmFindJobs_NoContainerSpecIsSkipped(t *testing.T) {
+	fake := &fakeSwarmDockerClient{services: []swarm.Service{
+		{
+			ID: "svc-weird",
+			Spec: swarm.ServiceSpec{
+				Annotations: swarm.Annotations{Name: "weird", Labels: map[string]string{"se.shcizo.auto-update": "true"}},
+				TaskTemplate: swarm.TaskSpec{ContainerSpec: nil},
+			},
+		},
+	}}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Empty(t, jobs)
+}
+```
+
+- [ ] **Step 3: Run tests to verify they fail**
+
+Run: `go test ./internal/discovery/... -run TestSwarmFindJobs -v`
+Expected: FAIL to compile — `discovery.NewSwarm` and `discovery.SwarmDockerClient` don't exist yet.
+
+- [ ] **Step 4: Implement**
+
+Create `internal/discovery/swarm.go`:
+
+```go
+package discovery
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/swarm"
+)
+
+// SwarmDockerClient is the subset of the Docker SDK SwarmDiscovery depends
+// on. Defined as an interface so tests can supply a fake.
+type SwarmDockerClient interface {
+	ServiceList(ctx context.Context, opts types.ServiceListOptions) ([]swarm.Service, error)
+}
+
+// SwarmDiscovery orchestrates "given an image, which Swarm services should
+// we update?". Unlike Discovery, it has no filesystem path to check — the
+// opt-in label on the service itself is the only gate, since Swarm services
+// have no local compose file to anchor a STACKS_ROOT check against.
+type SwarmDiscovery struct {
+	cli        SwarmDockerClient
+	optInLabel string
+}
+
+// NewSwarm returns a SwarmDiscovery bound to the given Docker client and
+// opt-in label.
+func NewSwarm(cli SwarmDockerClient, optInLabel string) *SwarmDiscovery {
+	return &SwarmDiscovery{cli: cli, optInLabel: optInLabel}
+}
+
+// FindJobs lists Swarm services, filters by image match + opt-in label, and
+// returns one Job per matching service. Signature matches Discovery.FindJobs
+// so both satisfy api.Finder.
+func (d *SwarmDiscovery) FindJobs(ctx context.Context, image string) ([]Job, error) {
+	services, err := d.cli.ServiceList(ctx, types.ServiceListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("docker service list: %w", err)
+	}
+
+	var jobs []Job
+	for _, svc := range services {
+		if svc.Spec.TaskTemplate.ContainerSpec == nil {
+			continue
+		}
+		if !ImagesMatch(image, svc.Spec.TaskTemplate.ContainerSpec.Image) {
+			continue
+		}
+		if !HasOptIn(svc.Spec.Labels, d.optInLabel) {
+			continue
+		}
+
+		jobs = append(jobs, Job{
+			Service:   svc.Spec.Name,
+			ServiceID: svc.ID,
+			Image:     image,
+		})
+	}
+
+	return jobs, nil
+}
+```
+
+- [ ] **Step 5: Run tests to verify they pass**
+
+Run: `go test ./internal/discovery/... -v`
+Expected: PASS — all discovery tests including the four new `TestSwarmFindJobs_*` cases, no regressions.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add internal/discovery/swarm.go internal/discovery/swarm_test.go
+git commit -m "feat(discovery): add SwarmDiscovery to find opted-in Swarm services by image"
+```
+
+---
+
+### Task 4: `SwarmExecutor` — run `docker service update --image`
+
+**Files:**
+- Create: `internal/updater/swarm_executor.go`
+- Test: `internal/updater/swarm_executor_test.go`
+
+**Interfaces:**
+- Consumes: `discovery.Job{Service, ServiceID, Image, Refused, RefusedReason}` (Task 2/3).
+- Produces: `updater.SwarmDockerClient` interface, `updater.NewSwarmExecutor(cli SwarmDockerClient) *SwarmExecutor`, method `(*SwarmExecutor) Execute(ctx context.Context, job discovery.Job) error` — satisfies `updater.Executor` (`internal/updater/executor.go`), so `Queue` (`internal/updater/queue.go`) needs no changes.
+
+- [ ] **Step 1: Verify the Docker SDK method signatures before writing code**
+
+Run: `go doc github.com/docker/docker/client.Client.ServiceInspectWithRaw` and `go doc github.com/docker/docker/client.Client.ServiceUpdate`
+
+Expected output shape (confirm before proceeding — adjust the interface in Step 3 if the SDK reports something different):
+
+```
+func (cli *Client) ServiceInspectWithRaw(ctx context.Context, serviceID string, options types.ServiceInspectOptions) (swarm.Service, []byte, error)
+func (cli *Client) ServiceUpdate(ctx context.Context, serviceID string, version swarm.Version, service swarm.ServiceSpec, options types.ServiceUpdateOptions) (types.ServiceUpdateResponse, error)
+```
+
+- [ ] **Step 2: Write the failing tests**
+
+Create `internal/updater/swarm_executor_test.go`:
+
+```go
+package updater_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/shcizo/package-updater/internal/updater"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeSwarmDockerClient struct {
+	inspectService swarm.Service
+	inspectErr     error
+	updateErr      error
+
+	updateCalledServiceID string
+	updateCalledVersion   swarm.Version
+	updateCalledSpec      swarm.ServiceSpec
+	updateCalledOpts      types.ServiceUpdateOptions
+}
+
+func (f *fakeSwarmDockerClient) ServiceInspectWithRaw(_ context.Context, _ string, _ types.ServiceInspectOptions) (swarm.Service, []byte, error) {
+	return f.inspectService, nil, f.inspectErr
+}
+
+func (f *fakeSwarmDockerClient) ServiceUpdate(_ context.Context, serviceID string, version swarm.Version, spec swarm.ServiceSpec, opts types.ServiceUpdateOptions) (types.ServiceUpdateResponse, error) {
+	f.updateCalledServiceID = serviceID
+	f.updateCalledVersion = version
+	f.updateCalledSpec = spec
+	f.updateCalledOpts = opts
+	return types.ServiceUpdateResponse{}, f.updateErr
+}
+
+func mkInspectService(version uint64, image string) swarm.Service {
+	return swarm.Service{
+		ID:      "svc-myapp",
+		Version: swarm.Version{Index: version},
+		Spec: swarm.ServiceSpec{
+			Annotations:  swarm.Annotations{Name: "myapp_web"},
+			TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: image}},
+		},
+	}
+}
+
+func TestSwarmExecutor_UpdatesImageAndVersion(t *testing.T) {
+	fake := &fakeSwarmDockerClient{inspectService: mkInspectService(7, "registry.example.com/myapp:v1")}
+	e := updater.NewSwarmExecutor(fake)
+
+	job := discovery.Job{Service: "myapp_web", ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"}
+	err := e.Execute(context.Background(), job)
+
+	require.NoError(t, err)
+	require.Equal(t, "svc-myapp", fake.updateCalledServiceID)
+	require.Equal(t, uint64(7), fake.updateCalledVersion.Index)
+	require.Equal(t, "registry.example.com/myapp:v2", fake.updateCalledSpec.TaskTemplate.ContainerSpec.Image)
+	require.True(t, fake.updateCalledOpts.QueryRegistry)
+}
+
+func TestSwarmExecutor_InspectError(t *testing.T) {
+	fake := &fakeSwarmDockerClient{inspectErr: errors.New("service not found")}
+	e := updater.NewSwarmExecutor(fake)
+	err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-missing", Image: "x:v2"})
+	require.Error(t, err)
+}
+
+func TestSwarmExecutor_UpdateError(t *testing.T) {
+	fake := &fakeSwarmDockerClient{
+		inspectService: mkInspectService(1, "registry.example.com/myapp:v1"),
+		updateErr:      errors.New("update rejected"),
+	}
+	e := updater.NewSwarmExecutor(fake)
+	err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"})
+	require.Error(t, err)
+}
+
+func TestSwarmExecutor_RefusedJobIsNotExecuted(t *testing.T) {
+	fake := &fakeSwarmDockerClient{}
+	e := updater.NewSwarmExecutor(fake)
+	job := discovery.Job{ServiceID: "svc-myapp", Image: "x:v2", Refused: true, RefusedReason: "not opted in"}
+	err := e.Execute(context.Background(), job)
+	require.Error(t, err)
+	require.Empty(t, fake.updateCalledServiceID)
+}
+```
+
+- [ ] **Step 3: Run tests to verify they fail**
+
+Run: `go test ./internal/updater/... -run TestSwarmExecutor -v`
+Expected: FAIL to compile — `updater.NewSwarmExecutor` doesn't exist yet.
+
+- [ ] **Step 4: Implement**
+
+Create `internal/updater/swarm_executor.go`:
+
+```go
+package updater
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/shcizo/package-updater/internal/discovery"
+)
+
+// SwarmDockerClient is the subset of the Docker SDK SwarmExecutor depends
+// on. Defined as an interface so tests can supply a fake.
+type SwarmDockerClient interface {
+	ServiceInspectWithRaw(ctx context.Context, serviceID string, opts types.ServiceInspectOptions) (swarm.Service, []byte, error)
+	ServiceUpdate(ctx context.Context, serviceID string, version swarm.Version, service swarm.ServiceSpec, opts types.ServiceUpdateOptions) (types.ServiceUpdateResponse, error)
+}
+
+// SwarmExecutor updates a Swarm service's image via the Docker API,
+// equivalent to `docker service update --image`.
+type SwarmExecutor struct {
+	cli SwarmDockerClient
+}
+
+// NewSwarmExecutor returns an executor that drives Swarm service updates
+// through the Docker API.
+func NewSwarmExecutor(cli SwarmDockerClient) *SwarmExecutor {
+	return &SwarmExecutor{cli: cli}
+}
+
+// Execute inspects the service to get its current spec + version (required
+// by the Docker API as an optimistic-concurrency token), sets the new image
+// on the container spec, and calls ServiceUpdate. QueryRegistry is set so a
+// floating tag (e.g. ":latest") resolves to a fresh digest and actually
+// triggers a rolling update instead of being treated as unchanged.
+func (e *SwarmExecutor) Execute(ctx context.Context, job discovery.Job) error {
+	if job.Refused {
+		return fmt.Errorf("refused: %s", job.RefusedReason)
+	}
+
+	svc, _, err := e.cli.ServiceInspectWithRaw(ctx, job.ServiceID, types.ServiceInspectOptions{})
+	if err != nil {
+		return fmt.Errorf("inspect service %s: %w", job.ServiceID, err)
+	}
+
+	spec := svc.Spec
+	spec.TaskTemplate.ContainerSpec.Image = job.Image
+
+	if _, err := e.cli.ServiceUpdate(ctx, job.ServiceID, svc.Version, spec, types.ServiceUpdateOptions{QueryRegistry: true}); err != nil {
+		return fmt.Errorf("update service %s: %w", job.ServiceID, err)
+	}
+	return nil
+}
+```
+
+- [ ] **Step 5: Run tests to verify they pass**
+
+Run: `go test ./internal/updater/... -v`
+Expected: PASS — all updater tests including the four new `TestSwarmExecutor_*` cases, no regressions in `TestQueue_*` or `TestWorker_*`.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add internal/updater/swarm_executor.go internal/updater/swarm_executor_test.go
+git commit -m "feat(updater): add SwarmExecutor to run docker service update via the Docker API"
+```
+
+---
+
+### Task 5: Surface `ServiceID` in the API response
+
+**Files:**
+- Modify: `internal/api/types.go`
+- Modify: `internal/api/handlers.go`
+- Test: `internal/api/swarm_result_test.go`
+
+**Interfaces:**
+- Consumes: `discovery.Job.ServiceID` (Task 2), `updater.Result{Job discovery.Job, Status Status, Error string, DurationMs int64}` (existing, `internal/updater/queue.go`).
+- Produces: `api.ResultDTO` gains `ServiceID string` (json `service_id`, omitempty) alongside the existing `ComposeFile` field, which stays empty for Swarm jobs (already guarded by the existing `len(res.Job.ConfigFiles) > 0` check).
+
+- [ ] **Step 1: Write the failing test**
+
+Create `internal/api/swarm_result_test.go`:
+
+```go
+package api_test
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/docker/docker/api/types"
+	"github.com/shcizo/package-updater/internal/api"
+	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/shcizo/package-updater/internal/updater"
+	"github.com/stretchr/testify/require"
+)
+
+type swarmResultFinder struct{ jobs []discovery.Job }
+
+func (f *swarmResultFinder) FindJobs(_ context.Context, _ string) ([]discovery.Job, error) {
+	return f.jobs, nil
+}
+
+type swarmResultSubmitter struct{ results []updater.Result }
+
+func (s *swarmResultSubmitter) Submit(_ context.Context, _ []discovery.Job) []updater.Result {
+	return s.results
+}
+
+type swarmResultPinger struct{}
+
+func (swarmResultPinger) Ping(_ context.Context) (types.Ping, error) { return types.Ping{}, nil }
+
+func TestUpdate_SwarmJobPopulatesServiceID(t *testing.T) {
+	job := discovery.Job{Service: "myapp_web", ServiceID: "svc123", Image: "registry.example.com/myapp:v2"}
+	finder := &swarmResultFinder{jobs: []discovery.Job{job}}
+	submitter := &swarmResultSubmitter{results: []updater.Result{
+		{Job: job, Status: updater.StatusUpdated, DurationMs: 42},
+	}}
+	h := api.NewHandlers(finder, submitter, swarmResultPinger{}, "v", "c", "b", nil)
+
+	req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`{"image":"registry.example.com/myapp"}`))
+	rec := httptest.NewRecorder()
+	h.Update(rec, req)
+
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Contains(t, rec.Body.String(), `"service_id":"svc123"`)
+	require.Contains(t, rec.Body.String(), `"service":"myapp_web"`)
+}
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `go test ./internal/api/... -run TestUpdate_SwarmJobPopulatesServiceID -v`
+Expected: FAIL — response body has no `service_id` key (`ResultDTO` has no such field yet).
+
+- [ ] **Step 3: Implement**
+
+In `internal/api/types.go`, add the field to `ResultDTO`:
+
+```go
+// ResultDTO is one row in UpdateResponse.Results.
+type ResultDTO struct {
+	Project     string `json:"project"`
+	Service     string `json:"service"`
+	ComposeFile string `json:"compose_file"`
+	ServiceID   string `json:"service_id,omitempty"`
+	Status      string `json:"status"`
+	Error       string `json:"error,omitempty"`
+	DurationMs  int64  `json:"duration_ms"`
+}
+```
+
+In `internal/api/handlers.go`, inside the `Update` handler's results loop, add `ServiceID` to the constructed `ResultDTO`:
+
+```go
+		resp.Results = append(resp.Results, ResultDTO{
+			Project:     res.Job.Project,
+			Service:     res.Job.Service,
+			ComposeFile: composeFile,
+			ServiceID:   res.Job.ServiceID,
+			Status:      string(res.Status),
+			Error:       res.Error,
+			DurationMs:  res.DurationMs,
+		})
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `go test ./internal/api/... -v`
+Expected: PASS — the new test plus all existing `TestUpdate_*`, `TestHealthz_*`, `TestVersion` cases (they don't assert on absence of `service_id`, and `omitempty` keeps it out of Compose-mode responses since `ServiceID` is `""` for those).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add internal/api/types.go internal/api/handlers.go internal/api/swarm_result_test.go
+git commit -m "feat(api): surface ServiceID in update results for swarm jobs"
+```
+
+---
+
+### Task 6: Wire `MODE` in `cmd/server/main.go`
+
+**Files:**
+- Modify: `cmd/server/main.go`
+
+**Interfaces:**
+- Consumes: `config.Config.Mode` (Task 1), `discovery.New` / `discovery.NewSwarm` (Task 3), `updater.NewComposeExecutor` / `updater.NewSwarmExecutor` (Task 4), `api.Finder` / `updater.Executor` (existing).
+- Produces: nothing new for other tasks to consume — this is the final wiring step.
+
+There is no unit test for `main.go` in this codebase (it's pure wiring); verification is a successful build plus the full test suite passing, run in Step 3.
+
+- [ ] **Step 1: Modify the wiring**
+
+In `cmd/server/main.go`, replace:
+
+```go
+	disc := discovery.New(dockerCli, cfg.StacksRoot, cfg.OptInLabel)
+	exec := updater.NewComposeExecutor()
+```
+
+with:
+
+```go
+	var finder api.Finder
+	var exec updater.Executor
+	switch cfg.Mode {
+	case "swarm":
+		finder = discovery.NewSwarm(dockerCli, cfg.OptInLabel)
+		exec = updater.NewSwarmExecutor(dockerCli)
+	case "compose":
+		finder = discovery.New(dockerCli, cfg.StacksRoot, cfg.OptInLabel)
+		exec = updater.NewComposeExecutor()
+	default:
+		// Unreachable: config.Load already validates Mode, but guard
+		// here too rather than silently falling through to a nil
+		// finder/exec pair.
+		return fmt.Errorf("unknown MODE %q", cfg.Mode)
+	}
+```
+
+Then update the `NewHandlers` call, which currently reads:
+
+```go
+	handlers := api.NewHandlers(disc, &submitterAdapter{queue: queue, timeout: cfg.UpdateTimeout}, dockerCli, version, commit, buildTime, m)
+```
+
+to use `finder` instead of `disc`:
+
+```go
+	handlers := api.NewHandlers(finder, &submitterAdapter{queue: queue, timeout: cfg.UpdateTimeout}, dockerCli, version, commit, buildTime, m)
+```
+
+Note `queue := updater.NewQueue(exec, m)` (a few lines above) already references `exec` — no change needed there since `exec` is now declared by the switch instead of a direct `updater.NewComposeExecutor()` call, but it's the same variable name so the line is unchanged. Also update the startup log line to include the mode:
+
+```go
+	logger.Info("starting",
+		"version", version, "commit", commit, "port", cfg.Port,
+		"mode", cfg.Mode, "stacks_root", cfg.StacksRoot, "opt_in_label", cfg.OptInLabel,
+	)
+```
+
+- [ ] **Step 2: Build**
+
+Run: `go build ./...`
+Expected: builds cleanly with no errors.
+
+- [ ] **Step 3: Run the full test suite**
+
+Run: `go test ./...`
+Expected: PASS across all packages — this exercises every task in this plan together for the first time.
+
+- [ ] **Step 4: Manual smoke check (compose mode, default)**
+
+Run: `go run ./cmd/server &` (with `UPDATER_API_KEY` set) and check the startup log line shows `"mode":"compose"` (or the equivalent field in whatever log format `internal/logging` renders). Stop the process afterward.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add cmd/server/main.go
+git commit -m "feat: wire MODE config to select compose or swarm discovery/executor"
+```
+
+---
+
+### Task 7: Documentation
+
+**Files:**
+- Modify: `README.md`
+- Modify: `CLAUDE.md`
+
+**Interfaces:** None — documentation only, no code interfaces produced or consumed.
+
+- [ ] **Step 1: Update `README.md`**
+
+Read the current README first to match its existing structure (env var table, "known limitations" section, etc. — check what's there before editing rather than guessing the exact heading names). Add:
+- `MODE` to the environment variable table: `MODE` — `compose` (default) or `swarm`. Selects the update mechanism for the whole deployment; not mixed per-request.
+- A short "Swarm mode" subsection explaining: services must carry the opt-in label (`OPT_IN_LABEL`, default `se.shcizo.auto-update`) directly on the *service* (`docker service update --label-add se.shcizo.auto-update=true <service>` or set in the stack file's `deploy.labels`), not on the container/task, since Swarm mode reads `Service.Spec.Labels`. Also note the updater must run on/against a Swarm manager node (`docker service update` requires manager API access) — pointing `DOCKER_HOST` at a manager, or scheduling the updater container itself on a manager node with the socket mounted, is the operator's responsibility.
+- Update/remove the existing "single host only" limitation note (if present) to say Compose mode remains single-host; Swarm mode is the multi-node path but only from a manager node's point of view.
+
+- [ ] **Step 2: Update `CLAUDE.md`**
+
+In the "Architecture" section, add a one-line entry for the new files:
+```
+- `internal/discovery/swarm.go`  — SwarmDiscovery: same FindJobs signature, lists `docker service ls`, gates on service-level opt-in label instead of STACKS_ROOT path-check
+- `internal/updater/swarm_executor.go` — SwarmExecutor: `docker service update --image` via the Docker API instead of a `docker compose` subprocess
+```
+
+In "Design intent (do not break without discussion)", add:
+```
+- **Compose mode and Swarm mode are selected once per deployment via `MODE`**, never
+  mixed at request time. Swarm mode's security gate is opt-in label only — there is
+  no STACKS_ROOT-equivalent path check, since Swarm services have no local compose
+  file. Don't add one; don't weaken Compose mode's three-factor gate to match.
+```
+
+In "Gotchas", add:
+```
+- **Swarm mode requires manager-node API access.** `docker service update` fails
+  with a permission error against a worker-only node. This is an operator/deployment
+  concern (point `DOCKER_HOST` at a manager, or schedule the updater on a manager),
+  not something the code can detect or work around.
+```
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add README.md CLAUDE.md
+git commit -m "docs: document MODE=swarm support and its opt-in label / manager-node requirements"
+```
@@ -0,0 +1,373 @@
+# Master/Agent Architecture — Design
+
+**Date:** 2026-10-12
+**Status:** Approved in brainstorming, pending written-spec review
+
+## Problem
+
+package-updater today is one instance per Docker daemon: a webhook (`POST /update`)
+finds opted-in containers by label and runs `docker compose` or `docker service update`.
+It is stateless and has no UI.
+
+The goal now is a tool that replaces Watchtower and Shepherd across the whole fleet:
+
+- One overview of every service on every machine — Swarm services and Compose stacks,
+  which can coexist on the same machine.
+- Per-service choice, made in the UI, between **manual** (show that an update exists,
+  a human clicks) and **auto** (apply as soon as a new version is detected).
+- Detect new versions both from CI (webhook) and by polling registries, so images we do
+  not build ourselves are covered too.
+- See available and completed updates, pick any of the recently seen versions
+  (which also gives rollback).
+- Live updates in the UI.
+
+Building this ourselves instead of adopting Komodo or Portainer is a deliberate choice:
+it is a hobby project, and owning it means it works exactly the way we want.
+
+## Environment and assumptions
+
+- Several machines on one internal network, reachable from each other by IP.
+- One machine (`swarm-mgr`) is a Swarm manager; most services are Swarm stacks
+  (`docker-stack.yml`). At least one Compose stack exists (`slusk`).
+- A central Postgres already runs and will be used.
+- Authentik runs and is used per application via OIDC (not via the reverse proxy).
+- A reverse proxy fronts the web UI.
+- Stack files are version-controlled. Image references are ad hoc: a mix of `repo:tag`,
+  `repo:tag@digest` and bare `repo@digest`. Many digests were written by a Claude instance
+  on the servers, not by an automated process.
+- Shepherd and Watchtower currently run and will be removed (see Migration).
+
+## Decisions
+
+### Topology: one master, one agent per machine
+
+- **Master** (one instance): Postgres, web UI, OIDC login, webhook, registry polling,
+  job decisions. Runs as a Swarm service on `swarm-mgr`.
+- **Agent** (one per machine): owns that machine's Docker daemon via `docker.sock`.
+  Reports inventory and runs update jobs. Runs via plain `docker compose` on each host
+  (not a global Swarm service), because each agent has its own token and a global
+  service would hand every node the same environment.
+  The agent mounts `docker.sock` and `STACKS_ROOT` **read-write** (today it is `:ro`),
+  because it writes `compose.override.yaml` files.
+
+### Connection: agent dials master over WebSocket, live
+
+The agent opens a long-lived WebSocket to the master and keeps it alive with reconnect
+and backoff. Inventory changes are pushed live from Docker events; the master forwards
+them to browsers. Chosen over the master polling agents over HTTP because live updates
+are an explicit goal.
+
+Messages are JSON in a common envelope `{type, id, payload}` with one Go type per
+message kind, shared in `internal/protocol`. JSON over WebSocket rather than gRPC:
+same tooling as the browser side, debuggable with `websocat`, no protobuf toolchain.
+
+### Trust in both directions
+
+Commands flow from master to agent and can restart anything on that machine, so the
+agent must authenticate the master, not only the other way round.
+
+- On first start the master generates a key pair and stores it in Postgres
+  (`master_identity`), so it survives container recreation and is covered by DB backups.
+- The master serves the agent endpoint on its own port with TLS, using a self-signed
+  certificate for that key. No reverse proxy, no CA.
+- The agent is configured with `MASTER_URL`, `MASTER_KEY` and `AGENT_TOKEN`.
+  `MASTER_KEY` is the SHA-256 fingerprint of the master's public key. The agent skips
+  normal chain verification and instead accepts the connection only if the
+  certificate's public key hashes to `MASTER_KEY` (same idea as SSH `known_hosts`).
+  The TLS handshake proves the master holds the private key.
+- `AGENT_TOKEN` is per agent, created in the UI ("Add agent", which shows a ready
+  compose snippet including token and `MASTER_KEY`). Only its hash is stored.
+  Agents can be revoked individually. A leaked token lets someone impersonate an agent
+  (feed false inventory), not command machines.
+- Losing the master key means every agent needs a new `MASTER_KEY`.
+
+Plain `ws://` with a signed handshake was rejected: anyone able to hijack the TCP
+stream after the handshake could inject commands.
+
+### State: the master remembers, agents report reality
+
+Agents are the truth for what is running. Postgres holds what the master must remember:
+desired version, update mode, version history and job history. This deliberately
+retires the "stateless" property for the master; agents stay stateless.
+
+### Security gate moves from labels to the UI
+
+The opt-in label no longer decides what may be updated; the per-service mode in the UI
+does. Consequences:
+
+- Every newly discovered service starts as `manual`. Nothing is ever `auto` by default.
+- Access to the UI is equivalent to being able to restart everything, so login is the
+  most important control (see Authentication).
+- The `STACKS_ROOT` path check stays on the agent as an independent guard for Compose,
+  and matters more now that the agent needs write access there (override files).
+- The master cannot reliably tell which service is its own Postgres or Authentik, so
+  they are not locked automatically. Putting them on `auto` is the operator's risk and
+  is documented.
+
+### Changing the version of a service
+
+| Kind | How the version is changed | What can revert it |
+|---|---|---|
+| Compose | Agent writes the service's image into `compose.override.yaml` next to the stack's compose file (the agent owns this file, preserving other services' entries in it), then `docker compose pull <svc>` + `up -d <svc>`. `docker compose` loads the override automatically, so manual `docker compose up -d` keeps the new version. | Deleting the override file. |
+| Swarm | `service update` via the Docker API with `repo:tag@digest` (the agent resolves the digest; without it Swarm does not re-pull when the tag is unchanged). | Re-running `docker stack deploy` with a file holding an older reference. |
+
+The updater always writes `repo:tag@digest`. The stack's own files in git are never
+edited. Writing version bumps back to git (GitOps style) was considered and deferred:
+it needs repo write access, YAML rewriting and conflict handling — a different kind of
+tool. It can be added later as a step after a successful update.
+
+### Drift
+
+A service drifts when `actual_ref` differs from `desired_ref`. A service the updater has
+never touched has no `desired_ref` and cannot drift. Drift is **never corrected
+automatically**, not even in `auto` mode, because a manual `stack deploy` with a new tag
+may be intentional. The UI shows the drift and offers "restore desired" or
+"accept running".
+
+### Service identity and moves
+
+A service row has its own ID. Location is an attribute with a unique index:
+Compose = agent + project + service; Swarm = cluster ID + service name (independent of
+which manager agent reports it); plain container = agent + container name.
+
+Identity without the agent was rejected: the same stack legitimately runs on several
+machines (the agent itself, node exporters), and they would be merged into one history.
+Automatic merging was rejected for the same reason.
+
+When a Compose service disappears on one agent and exactly one service with the same
+project and service name appears on another, the UI suggests "Moved from A?" with a
+**Merge** action: the old row is moved to the new location with its history, mode and
+`desired_ref`, and the duplicate is removed. Merging can also be done by hand.
+
+### Versions: what the system has seen
+
+Every version the system sees is recorded in `image_versions`, from three sources:
+webhook (every build CI announces), polling (every new digest under a followed tag) and
+inventory (whatever is running). The service page lists the 10 most recent versions for
+the service's repo; picking one creates a manual job to exactly that reference. Picking
+an older one is a rollback.
+
+`available_*` on the service remains as "the newest suggestion", which drives the
+"update available" badge and `auto` mode.
+
+| Reference format | Polling | Webhook | Pick a version |
+|---|---|---|---|
+| `repo:tag` | yes | yes | yes |
+| `repo:tag@digest` | yes (follows the tag) | yes | yes |
+| `repo@digest` | no — nothing to follow | yes (own images) | yes (seen versions) |
+
+Services without a tag are flagged "follows no tag" in the UI. Recommended convention
+for stack files: `repo:tag@digest`, tag = what to follow.
+
+## Data model (Postgres)
+
+**`master_identity`** — single row: private key, created_at.
+
+**`agents`** — `id`, `name` (set when adding), `token_hash`, `hostname`,
+`agent_version`, `is_swarm_manager` (reported), `last_seen_at`, `revoked_at`, `created_at`.
+Online/offline is held in memory only.
+
+**`services`**
+- `id`
+- `kind`: `compose` | `swarm` | `container` (plain containers are shown but cannot be
+  updated in v1)
+- location: `agent_id`, `compose_project`, `compose_service`, `swarm_cluster_id`,
+  `swarm_service`, `container_name` (populated per kind; unique index per kind)
+- `image_repo` — normalised, used to match webhooks and polling
+- `update_mode`: `manual` (default) | `auto`
+- `desired_ref`
+- `actual_ref`, `actual_digest`, `status` — last known, kept so the UI has something to
+  show while an agent is offline
+- `available_ref`, `available_digest`, `available_source` (`webhook` | `poll`),
+  `available_at`
+- `last_poll_error`
+- `first_seen_at`, `removed_at` — disappeared services are marked, not deleted
+
+**`image_versions`** — `repo`, `tag`, `digest`, `source` (`webhook` | `poll` | `running`),
+`first_seen_at`; unique on (repo, tag, digest).
+
+**`update_jobs`** — `id`, `service_id`, `from_ref`, `to_ref`,
+`trigger` (`manual` | `webhook` | `poll`), `requested_by` (OIDC subject for manual),
+`status` (`queued` | `running` | `succeeded` | `failed`), `output`, `error`,
+`created_at`, `started_at`, `finished_at`.
+
+**`sessions`** — web sessions for OIDC logins.
+
+## Flows
+
+### Inventory
+
+1. On connect the agent sends `hello` (hostname, version, Swarm manager or not, cluster
+   ID) and a full `snapshot`.
+2. The master diffs the snapshot against that agent's rows: new services are inserted as
+   `manual`, missing ones get `removed_at`, changed ones get new `actual_*`. Running
+   references are recorded in `image_versions` with source `running`.
+3. The agent subscribes to Docker events (container start/stop/die/destroy, service
+   update), debounces bursts (a Compose restart emits many events), re-inspects the
+   affected item and sends `upsert` or `remove`.
+4. Swarm services are reported only by agents on manager nodes. Agents skip containers
+   that belong to Swarm tasks, so a service is not listed once per node.
+5. The master pushes changes to browsers over SSE.
+
+### Update
+
+1. Triggers: a click in the UI, the webhook, or polling. Webhook and polling always set
+   `available_*` and record the version; they create a job only if the service is `auto`.
+2. The master inserts an `update_jobs` row and sends `job.start` to the owning agent
+   (for Swarm, a connected manager agent).
+3. The agent puts the job in its FIFO queue (single worker, as today) and executes it as
+   described under "Changing the version of a service".
+4. The agent sends `job.progress` and `job.result` with output. On success the master
+   sets `desired_ref` and clears `available_*` if that version is now running.
+
+### Failure handling
+
+| Situation | Behaviour |
+|---|---|
+| Agent offline when a job is created | Job fails immediately with "agent offline". It is not stored for later — a job that runs hours later surprises more than it helps. `available_*` remains, so the user can retry. |
+| Connection drops mid-job | The agent finishes the job (its queue is local) and keeps results in memory until the master acknowledges them. They are resent after reconnect along with the snapshot. |
+| Master restarts mid-job | Jobs in `running` are reconciled when the agent reconnects; jobs the agent does not know are marked `failed` ("lost"). |
+| New trigger while a job for the same service is queued or running | No new job; only `available_*` is updated. Two jobs never race on one service. |
+| Job exceeds `UPDATE_TIMEOUT` | The agent cancels it, as today. |
+| Polling fails for one image | Error stored in `last_poll_error` and shown; other images continue. |
+
+### Self-protection
+
+The agent reports its own container and the master knows its own service. Both are shown
+as "updated manually outside the system", and the master refuses to create jobs for
+them. An agent recreating itself would kill the job it is running.
+
+## Registry polling
+
+- The master polls, not the agents: credentials live in one place and identical
+  references are polled once.
+- Every distinct `repo:tag` in use is polled, including `repo:tag@digest`. Bare
+  `repo@digest` and locally built images without a registry digest are skipped and
+  shown as such.
+- Mechanism: `HEAD /v2/<repo>/manifests/<tag>`; the returned digest is compared with
+  `actual_digest`. Use `go-containerregistry` — it handles Docker Hub's token auth and
+  multi-arch indexes.
+- Interval `POLL_INTERVAL`, default 6h. Believed not to count against Docker Hub pull
+  limits for HEAD requests — verify during implementation.
+- Credentials: a Docker `config.json` (as written by `docker login`) mounted into the
+  master. No credential UI, no encryption in Postgres.
+
+## Webhook
+
+- `POST /update` with `{image, tag}` and a bearer token — the same contract as today, so
+  the existing Gitea action works unchanged; only its `endpoint` moves to the master.
+- `image` is matched against `image_repo` across all agents using today's normalisation
+  (`NormaliseImage`). Each match gets `available_ref = repo:tag` (source `webhook`) and
+  the version is recorded; `auto` services get a job.
+- The response waits for the `auto` jobs to finish (bounded by `UPDATE_TIMEOUT`), and
+  returns 5xx if any failed or timed out, so CI still surfaces a broken deploy. `manual` matches are reported as "marked", not as
+  errors.
+- No match returns **404**, which catches typos in CI configuration.
+- The webhook never creates services. The first deployment of a stack is done by hand.
+
+## Authentication and UI
+
+### Authentication
+
+Three separate mechanisms that never mix:
+
+| Who | How | Where |
+|---|---|---|
+| People | OIDC authorization code flow with PKCE against Authentik (`coreos/go-oidc`, `x/oauth2`) | Web UI, via reverse proxy |
+| CI | Shared bearer token | `POST /update`, via reverse proxy |
+| Agents | Per-agent token + pinned master key over TLS | Dedicated agent port, internal network |
+
+- Sessions are stored in Postgres; the cookie holds a session ID (`HttpOnly`, `Secure`,
+  `SameSite=Lax`).
+- The master also requires membership of `ALLOWED_GROUP` from the token's `groups`
+  claim. Restricting the application in Authentik alone is not enough: one misconfigured
+  binding would hand every Authentik user the ability to restart the fleet.
+- All state-changing UI requests carry a CSRF token.
+- `/healthz`, `/metrics` and `/version` stay unauthenticated, as today.
+
+### UI
+
+Server-rendered Go templates with htmx; live updates via htmx's SSE support. Embedded in
+the master binary with `go:embed`. No Node build step. Visual design is drafted
+separately; this spec defines what each page contains.
+
+1. **Overview** — all services grouped per machine (Swarm services per cluster). Each
+   shows status, running version, the manual/auto toggle, and badges for
+   *update available*, *drifted* and *follows no tag*. Live.
+2. **Service** — the 10 most recent versions with an *Update* action, job history with
+   logs, drift actions (restore / accept), and *Merge* for moves.
+3. **Agents** — online/offline, version. *Add agent* produces a compose snippet with
+   token and `MASTER_KEY`. *Revoke*.
+4. **Jobs** — full history; running jobs stream their log live.
+
+## Code layout
+
+- `cmd/master`, `cmd/agent` — replace `cmd/server`.
+- `internal/protocol` — message types shared by master and agent.
+- `internal/agent/...` — today's `discovery` and `updater` (queue, Compose and Swarm
+  executors) move here, plus inventory, Docker events and the connection loop.
+- `internal/master/...` — agent hub, jobs, polling, webhook, OIDC, storage.
+- `web/` — templates and static assets.
+
+Existing conventions continue: interfaces defined at the consumer, hand-written fakes,
+nil-able `*Metrics`, Conventional Commits.
+
+## Testing
+
+- `go test ./...` must keep passing without Docker or Postgres.
+- Master logic (mode decisions, drift, merge, webhook matching, job reconciliation) is
+  tested against a storage interface with a hand-written fake.
+- Postgres storage is tested by integration tests behind the `integration` build tag,
+  requiring `DATABASE_URL`.
+- Master and agent are tested together in-process: an `httptest` TLS server, a real
+  WebSocket client and a fake Docker. Covers the `MASTER_KEY` handshake (including a
+  wrong key being rejected), reconnect with snapshot, and jobs completed while
+  disconnected.
+- Table-driven tests with testify.
+
+## Migration from today's deployment
+
+1. Create the database and user in Postgres; register the application and group in
+   Authentik.
+2. Deploy the master.
+3. **Remove Shepherd and Watchtower.** Running alongside, they would update services
+   behind the master's back and every such update would show up as drift.
+4. Add one agent per machine through the UI. Inventory fills in; everything is `manual`.
+5. Point the Gitea action's `endpoint` at the master. Contract unchanged.
+6. Remove today's package-updater instance.
+7. Convert bare `repo@digest` references to `repo:tag@digest`, and set `auto` where
+   wanted. Recommended: record the convention in the self-hosted repo's `CLAUDE.md`
+   so Claude instances on the servers stop writing bare digests.
+
+`CLAUDE.md` in this repo is updated as part of the implementation: the "stateless" and
+"token AND opt-in label AND STACKS_ROOT" design-intent entries no longer describe the
+system, and the `MODE` entry is replaced by the agent model.
+
+## Implementation phases
+
+Each phase ends with something that works on its own.
+
+1. **Protocol, agent and master with inventory.** Pinned-key TLS, per-agent tokens,
+   snapshot and live events, Postgres storage, OIDC login, Overview and Agents pages.
+   The overview works; nothing can be updated yet.
+2. **Jobs and manual update.** Job protocol, agent queue and executors (Compose override
+   file, Swarm with digest), job history, Service and Jobs pages, self-protection.
+3. **Webhook and auto mode.** `POST /update` on the master, matching, waiting for
+   auto jobs, mode toggle.
+4. **Polling and version history.** `image_versions`, registry polling, version picker
+   and rollback.
+5. **Drift and merge.** Drift detection and actions, move suggestion and merge.
+
+## Out of scope (later specs)
+
+- Detecting newer version *tags* for images we do not build (e.g. `gitea:1.27.3` →
+  `1.27.4`, `postgres:17.10`, `loki:3.4.2`). Requires listing tags and ordering them;
+  major upgrades such as Postgres 17 → 18 must never be automatic. Revisit after
+  checking the other two servers' stack files.
+- Notifications.
+- Export/import of a fleet configuration to restore a setup. The `services` table
+  already holds what an export needs.
+- Writing version bumps back to git.
+- Self-update of master and agents (`internal/selfupdate` is the starting point).
+- Updating plain (non-Compose, non-Swarm) containers.
+- Credential management UI for registries.
@@ -3,8 +3,11 @@ module github.com/shcizo/package-updater
 go 1.26.3

 require (
+	github.com/coder/websocket v1.8.15
 	github.com/docker/docker v28.5.2+incompatible
+	github.com/jackc/pgx/v5 v5.10.0
 	github.com/prometheus/client_golang v1.23.2
+	github.com/prometheus/client_model v0.6.2
 	github.com/stretchr/testify v1.11.1
 )

@@ -22,6 +25,9 @@ require (
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/sys/atomicwriter v0.1.0 // indirect
 	github.com/moby/term v0.5.2 // indirect
@@ -31,7 +37,6 @@ require (
 	github.com/opencontainers/image-spec v1.1.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
@@ -41,7 +46,9 @@ require (
 	go.opentelemetry.io/otel/metric v1.43.0 // indirect
 	go.opentelemetry.io/otel/trace v1.43.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
+	golang.org/x/text v0.35.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
@@ -8,12 +8,15 @@ github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1x
 github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
+github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
 github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M=
 github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE=
 github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk=
 github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
 github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
+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/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
@@ -37,6 +40,14 @@ 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/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c=
+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-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -77,6 +88,9 @@ github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0t
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+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.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
@@ -105,6 +119,8 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
 go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
 golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
 golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
 golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
@@ -122,6 +138,7 @@ google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+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=
 gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q=
@@ -0,0 +1,51 @@
+// Package config loads the agent's configuration from environment variables.
+package config
+
+import (
+	"errors"
+	"os"
+	"regexp"
+	"strings"
+)
+
+var masterKeyPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+
+// Config holds the agent's runtime configuration.
+type Config struct {
+	MasterURL string // MASTER_URL, wss://host:port/agent
+	MasterKey string // MASTER_KEY, fingerprint shown in the master UI
+	Token     string // AGENT_TOKEN, shown once when the agent was added
+	SelfID    string // SELF_ID, defaults to the hostname (= container ID prefix)
+	LogLevel  string // LOG_LEVEL, default info
+}
+
+// Load reads and validates the agent configuration. It fails fast so a
+// misconfigured agent never starts dialling.
+func Load() (*Config, error) {
+	cfg := &Config{
+		MasterURL: os.Getenv("MASTER_URL"),
+		MasterKey: os.Getenv("MASTER_KEY"),
+		Token:     os.Getenv("AGENT_TOKEN"),
+		SelfID:    os.Getenv("SELF_ID"),
+		LogLevel:  os.Getenv("LOG_LEVEL"),
+	}
+	switch {
+	case cfg.MasterURL == "":
+		return nil, errors.New("MASTER_URL is required")
+	case !strings.HasPrefix(cfg.MasterURL, "wss://"):
+		return nil, errors.New("MASTER_URL must start with wss://")
+	case cfg.MasterKey == "":
+		return nil, errors.New("MASTER_KEY is required")
+	case !masterKeyPattern.MatchString(cfg.MasterKey):
+		return nil, errors.New("MASTER_KEY must look like sha256:<64 hex>")
+	case cfg.Token == "":
+		return nil, errors.New("AGENT_TOKEN is required")
+	}
+	if cfg.SelfID == "" {
+		cfg.SelfID, _ = os.Hostname()
+	}
+	if cfg.LogLevel == "" {
+		cfg.LogLevel = "info"
+	}
+	return cfg, nil
+}
@@ -0,0 +1,50 @@
+package config
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const validKey = "sha256:0000000000000000000000000000000000000000000000000000000000000000"
+
+func setValid(t *testing.T) {
+	t.Setenv("MASTER_URL", "wss://10.0.0.5:8090/agent")
+	t.Setenv("MASTER_KEY", validKey)
+	t.Setenv("AGENT_TOKEN", "pua_abc")
+	t.Setenv("SELF_ID", "")
+	t.Setenv("LOG_LEVEL", "")
+}
+
+func TestLoad_Valid(t *testing.T) {
+	setValid(t)
+	cfg, err := Load()
+	require.NoError(t, err)
+	host, _ := os.Hostname()
+	assert.Equal(t, &Config{
+		MasterURL: "wss://10.0.0.5:8090/agent", MasterKey: validKey, Token: "pua_abc",
+		SelfID: host, LogLevel: "info",
+	}, cfg)
+}
+
+func TestLoad_Invalid(t *testing.T) {
+	cases := []struct {
+		name, key, value, wantErr string
+	}{
+		{"missing url", "MASTER_URL", "", "MASTER_URL is required"},
+		{"plain ws", "MASTER_URL", "ws://10.0.0.5:8090/agent", "MASTER_URL must start with wss://"},
+		{"missing key", "MASTER_KEY", "", "MASTER_KEY is required"},
+		{"malformed key", "MASTER_KEY", "abc", "MASTER_KEY must look like sha256:<64 hex>"},
+		{"missing token", "AGENT_TOKEN", "", "AGENT_TOKEN is required"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			setValid(t)
+			t.Setenv(c.key, c.value)
+			_, err := Load()
+			assert.ErrorContains(t, err, c.wantErr)
+		})
+	}
+}
@@ -0,0 +1,171 @@
+// Package conn keeps an agent connected to the master: it dials over pinned
+// TLS, announces itself, sends its inventory, and streams changes until the
+// connection drops, then reconnects with backoff.
+package conn
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/wsjson"
+
+	"github.com/shcizo/package-updater/internal/agent/inventory"
+	"github.com/shcizo/package-updater/internal/pin"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// readLimit caps a single incoming message. The library default (32 KiB)
+// is too small for the master's future job messages; matches the master.
+const readLimit = 1 << 20
+
+// Source supplies what the agent reports. *inventory.Collector satisfies it.
+type Source interface {
+	Hello(ctx context.Context) (protocol.Hello, error)
+	Snapshot(ctx context.Context) ([]protocol.Service, error)
+}
+
+// Config holds connection settings. Token is sent as a bearer header and must
+// never be logged.
+type Config struct {
+	URL         string // wss://host:port/agent
+	Token       string // AGENT_TOKEN
+	Fingerprint string // MASTER_KEY
+	Version     string
+	MinBackoff  time.Duration // default 1s
+	MaxBackoff  time.Duration // default 30s
+}
+
+// Agent is the agent's side of the master connection.
+type Agent struct {
+	cfg     Config
+	src     Source
+	log     *slog.Logger
+	changes chan struct{}
+}
+
+// New returns an Agent. Call Run to connect.
+func New(cfg Config, src Source, log *slog.Logger) *Agent {
+	if cfg.MinBackoff == 0 {
+		cfg.MinBackoff = time.Second
+	}
+	if cfg.MaxBackoff == 0 {
+		cfg.MaxBackoff = 30 * time.Second
+	}
+	return &Agent{cfg: cfg, src: src, log: log, changes: make(chan struct{}, 1)}
+}
+
+// Notify tells the agent its inventory may have changed. It never blocks;
+// several calls before the agent gets to them collapse into one refresh.
+func (a *Agent) Notify() {
+	select {
+	case a.changes <- struct{}{}:
+	default:
+	}
+}
+
+// Run keeps the agent connected until ctx ends, then returns nil.
+func (a *Agent) Run(ctx context.Context) error {
+	backoff := a.cfg.MinBackoff
+	for {
+		connected, err := a.session(ctx)
+		if ctx.Err() != nil {
+			return nil
+		}
+		if connected {
+			backoff = a.cfg.MinBackoff
+		}
+		a.log.Warn("master_connection_lost", "error", err, "retry_in", backoff.String())
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(backoff):
+		}
+		backoff = min(backoff*2, a.cfg.MaxBackoff)
+	}
+}
+
+// session runs one connection until it fails. connected reports whether the
+// hello was delivered, so Run resets its backoff only after real progress.
+func (a *Agent) session(parent context.Context) (connected bool, err error) {
+	httpClient := &http.Client{Transport: &http.Transport{TLSClientConfig: pin.ClientConfig(a.cfg.Fingerprint)}}
+	c, _, err := websocket.Dial(parent, a.cfg.URL, &websocket.DialOptions{
+		HTTPClient: httpClient,
+		HTTPHeader: http.Header{"Authorization": []string{"Bearer " + a.cfg.Token}},
+	})
+	if err != nil {
+		return false, fmt.Errorf("dial master: %w", err)
+	}
+	defer c.CloseNow()
+	c.SetReadLimit(readLimit)
+
+	ctx, cancel := context.WithCancel(parent)
+	defer cancel()
+	// The master sends nothing yet in this phase, but reading is how a
+	// dropped connection is noticed, so a reader runs for the whole session.
+	go func() {
+		defer cancel()
+		for {
+			if _, _, err := c.Read(ctx); err != nil {
+				return
+			}
+		}
+	}()
+
+	hello, err := a.src.Hello(ctx)
+	if err != nil {
+		return false, err
+	}
+	hello.AgentVersion = a.cfg.Version
+	if err := send(ctx, c, protocol.TypeHello, hello); err != nil {
+		return false, err
+	}
+
+	last, err := a.src.Snapshot(ctx)
+	if err != nil {
+		return true, err
+	}
+	if err := send(ctx, c, protocol.TypeSnapshot, protocol.Snapshot{Services: last}); err != nil {
+		return true, err
+	}
+	a.log.Info("master_connected", "services", len(last))
+
+	for {
+		select {
+		case <-ctx.Done():
+			_ = c.Close(websocket.StatusNormalClosure, "")
+			return true, ctx.Err()
+		case <-a.changes:
+			next, err := a.src.Snapshot(ctx)
+			if err != nil {
+				return true, err
+			}
+			upserts, removes := inventory.Diff(last, next)
+			for _, s := range upserts {
+				if err := send(ctx, c, protocol.TypeUpsert, protocol.Upsert{Service: s}); err != nil {
+					return true, err
+				}
+			}
+			for _, k := range removes {
+				if err := send(ctx, c, protocol.TypeRemove, protocol.Remove{Key: k}); err != nil {
+					return true, err
+				}
+			}
+			last = next
+		}
+	}
+}
+
+func send(ctx context.Context, c *websocket.Conn, typ string, payload any) error {
+	env, err := protocol.Encode(typ, payload)
+	if err != nil {
+		return err
+	}
+	if err := wsjson.Write(ctx, c, env); err != nil {
+		return fmt.Errorf("send %s: %w", typ, err)
+	}
+	return nil
+}
@@ -0,0 +1,200 @@
+package conn
+
+import (
+	"context"
+	"crypto/tls"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/wsjson"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/pin"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+type fakeMaster struct {
+	srv         *httptest.Server
+	fingerprint string
+	envs        chan protocol.Envelope
+	auth        chan string
+	conns       chan *websocket.Conn
+	attempts    atomic.Int64
+}
+
+func newFakeMaster(t *testing.T) *fakeMaster {
+	t.Helper()
+	key, err := pin.GenerateKey()
+	require.NoError(t, err)
+	cert, err := pin.Certificate(key)
+	require.NoError(t, err)
+	fp, err := pin.Fingerprint(key.Public())
+	require.NoError(t, err)
+
+	fm := &fakeMaster{
+		fingerprint: fp,
+		envs:        make(chan protocol.Envelope, 100),
+		auth:        make(chan string, 10),
+		conns:       make(chan *websocket.Conn, 10),
+	}
+	fm.srv = httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fm.auth <- r.Header.Get("Authorization")
+		c, err := websocket.Accept(w, r, nil)
+		if err != nil {
+			return
+		}
+		c.SetReadLimit(1 << 20)
+		fm.conns <- c
+		for {
+			var env protocol.Envelope
+			if err := wsjson.Read(r.Context(), c, &env); err != nil {
+				return
+			}
+			fm.envs <- env
+		}
+	}))
+	fm.srv.Config.ConnState = func(_ net.Conn, state http.ConnState) {
+		if state == http.StateNew {
+			fm.attempts.Add(1)
+		}
+	}
+	fm.srv.TLS = &tls.Config{Certificates: []tls.Certificate{cert}}
+	fm.srv.StartTLS()
+	t.Cleanup(fm.srv.Close)
+	return fm
+}
+
+func (fm *fakeMaster) url() string { return "wss" + strings.TrimPrefix(fm.srv.URL, "https") }
+
+func (fm *fakeMaster) next(t *testing.T) protocol.Envelope {
+	t.Helper()
+	select {
+	case e := <-fm.envs:
+		return e
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for a message from the agent")
+		return protocol.Envelope{}
+	}
+}
+
+type fakeSource struct {
+	mu       sync.Mutex
+	services []protocol.Service
+}
+
+func (f *fakeSource) set(s ...protocol.Service) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.services = s
+}
+
+func (f *fakeSource) Hello(context.Context) (protocol.Hello, error) {
+	return protocol.Hello{Hostname: "node-2"}, nil
+}
+
+func (f *fakeSource) Snapshot(context.Context) ([]protocol.Service, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return append([]protocol.Service(nil), f.services...), nil
+}
+
+var (
+	svcA = protocol.Service{Kind: protocol.KindContainer, ContainerName: "a", Image: "a:1", Status: "running"}
+	svcB = protocol.Service{Kind: protocol.KindContainer, ContainerName: "b", Image: "b:1", Status: "running"}
+)
+
+func startAgent(t *testing.T, fm *fakeMaster, fingerprint string, src Source) *Agent {
+	t.Helper()
+	a := New(Config{
+		URL: fm.url(), Token: "pua_test", Fingerprint: fingerprint, Version: "v-test",
+		MinBackoff: 10 * time.Millisecond, MaxBackoff: 50 * time.Millisecond,
+	}, src, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { _ = a.Run(ctx); close(done) }()
+	t.Cleanup(func() { cancel(); <-done })
+	return a
+}
+
+func TestAgent_SendsHelloThenSnapshotWithToken(t *testing.T) {
+	fm := newFakeMaster(t)
+	src := &fakeSource{}
+	src.set(svcA)
+	startAgent(t, fm, fm.fingerprint, src)
+
+	assert.Equal(t, "Bearer pua_test", <-fm.auth)
+	hello := fm.next(t)
+	require.Equal(t, protocol.TypeHello, hello.Type)
+	h, err := protocol.Decode[protocol.Hello](hello)
+	require.NoError(t, err)
+	assert.Equal(t, protocol.Hello{Hostname: "node-2", AgentVersion: "v-test"}, h)
+
+	snap := fm.next(t)
+	require.Equal(t, protocol.TypeSnapshot, snap.Type)
+	s, err := protocol.Decode[protocol.Snapshot](snap)
+	require.NoError(t, err)
+	assert.Equal(t, []protocol.Service{svcA}, s.Services)
+}
+
+func TestAgent_NotifySendsDiff(t *testing.T) {
+	fm := newFakeMaster(t)
+	src := &fakeSource{}
+	src.set(svcA)
+	a := startAgent(t, fm, fm.fingerprint, src)
+	fm.next(t) // hello
+	fm.next(t) // snapshot
+
+	src.set(svcB)
+	a.Notify()
+
+	up := fm.next(t)
+	require.Equal(t, protocol.TypeUpsert, up.Type)
+	u, err := protocol.Decode[protocol.Upsert](up)
+	require.NoError(t, err)
+	assert.Equal(t, svcB, u.Service)
+
+	rm := fm.next(t)
+	require.Equal(t, protocol.TypeRemove, rm.Type)
+	r, err := protocol.Decode[protocol.Remove](rm)
+	require.NoError(t, err)
+	assert.Equal(t, "container|a", r.Key)
+}
+
+func TestAgent_ReconnectsAndResendsSnapshot(t *testing.T) {
+	fm := newFakeMaster(t)
+	src := &fakeSource{}
+	src.set(svcA)
+	startAgent(t, fm, fm.fingerprint, src)
+	fm.next(t)
+	fm.next(t)
+
+	first := <-fm.conns
+	first.CloseNow()
+
+	assert.Equal(t, protocol.TypeHello, fm.next(t).Type)
+	assert.Equal(t, protocol.TypeSnapshot, fm.next(t).Type)
+}
+
+func TestAgent_RefusesMasterWithWrongKey(t *testing.T) {
+	fm := newFakeMaster(t)
+	other, err := pin.GenerateKey()
+	require.NoError(t, err)
+	wrong, err := pin.Fingerprint(other.Public())
+	require.NoError(t, err)
+
+	startAgent(t, fm, wrong, &fakeSource{})
+	require.Eventually(t, func() bool { return fm.attempts.Load() >= 1 },
+		2*time.Second, 10*time.Millisecond, "the agent must actually try to connect")
+	time.Sleep(200 * time.Millisecond)
+	assert.Empty(t, fm.auth, "the token must never be sent to a server with the wrong key")
+}
@@ -0,0 +1,193 @@
+// Package inventory turns what the local Docker daemon runs into the
+// services an agent reports to the master.
+package inventory
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/image"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/system"
+
+	"github.com/shcizo/package-updater/internal/imageref"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+const (
+	labelProject    = "com.docker.compose.project"
+	labelService    = "com.docker.compose.service"
+	labelWorkingDir = "com.docker.compose.project.working_dir"
+	labelSwarmTask  = "com.docker.swarm.service.id"
+)
+
+// DockerClient is the subset of the Docker SDK the collector needs.
+// *client.Client satisfies it.
+type DockerClient interface {
+	ContainerList(ctx context.Context, opts container.ListOptions) ([]container.Summary, error)
+	ImageList(ctx context.Context, opts image.ListOptions) ([]image.Summary, error)
+	ServiceList(ctx context.Context, opts swarm.ServiceListOptions) ([]swarm.Service, error)
+	Info(ctx context.Context) (system.Info, error)
+}
+
+// Collector reads the local daemon's inventory.
+type Collector struct {
+	cli    DockerClient
+	selfID string
+}
+
+// New returns a Collector. selfID is a prefix of the agent's own container
+// ID (inside a container, the hostname); that container is marked Self.
+func New(cli DockerClient, selfID string) *Collector {
+	return &Collector{cli: cli, selfID: selfID}
+}
+
+// Hello returns the machine facts sent on connect. AgentVersion is left for
+// the caller to fill in.
+func (c *Collector) Hello(ctx context.Context) (protocol.Hello, error) {
+	info, err := c.cli.Info(ctx)
+	if err != nil {
+		return protocol.Hello{}, fmt.Errorf("docker info: %w", err)
+	}
+	h := protocol.Hello{Hostname: info.Name, SwarmManager: info.Swarm.ControlAvailable}
+	if info.Swarm.Cluster != nil {
+		h.SwarmClusterID = info.Swarm.Cluster.ID
+	}
+	return h, nil
+}
+
+// Snapshot returns every service on this machine, sorted by Key.
+func (c *Collector) Snapshot(ctx context.Context) ([]protocol.Service, error) {
+	info, err := c.cli.Info(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("docker info: %w", err)
+	}
+	containers, err := c.cli.ContainerList(ctx, container.ListOptions{All: true})
+	if err != nil {
+		return nil, fmt.Errorf("docker container list: %w", err)
+	}
+	images, err := c.cli.ImageList(ctx, image.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("docker image list: %w", err)
+	}
+	repoDigests := make(map[string][]string, len(images))
+	for _, img := range images {
+		repoDigests[img.ID] = img.RepoDigests
+	}
+
+	byKey := make(map[string]protocol.Service)
+	for _, ct := range containers {
+		if _, isTask := ct.Labels[labelSwarmTask]; isTask {
+			continue
+		}
+		svc := c.fromContainer(ct, repoDigests[ct.ImageID])
+		if prev, ok := byKey[svc.Key()]; ok {
+			svc = mergeReplicas(prev, svc)
+		}
+		byKey[svc.Key()] = svc
+	}
+
+	if info.Swarm.ControlAvailable {
+		services, err := c.cli.ServiceList(ctx, swarm.ServiceListOptions{Status: true})
+		if err != nil {
+			return nil, fmt.Errorf("docker service list: %w", err)
+		}
+		for _, s := range services {
+			if s.Spec.TaskTemplate.ContainerSpec == nil {
+				continue
+			}
+			svc := fromSwarmService(s)
+			byKey[svc.Key()] = svc
+		}
+	}
+
+	out := make([]protocol.Service, 0, len(byKey))
+	for _, s := range byKey {
+		out = append(out, s)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Key() < out[j].Key() })
+	return out, nil
+}
+
+func (c *Collector) fromContainer(ct container.Summary, repoDigests []string) protocol.Service {
+	svc := protocol.Service{
+		Image:  ct.Image,
+		Digest: digestFor(ct.Image, repoDigests),
+		Status: containerStatus(ct.State),
+		Self:   c.selfID != "" && strings.HasPrefix(ct.ID, c.selfID),
+	}
+	project, service := ct.Labels[labelProject], ct.Labels[labelService]
+	if project != "" && service != "" {
+		svc.Kind = protocol.KindCompose
+		svc.ComposeProject = project
+		svc.ComposeService = service
+		svc.WorkingDir = ct.Labels[labelWorkingDir]
+		return svc
+	}
+	svc.Kind = protocol.KindContainer
+	svc.ContainerName = ct.ID
+	if len(ct.Names) > 0 {
+		svc.ContainerName = strings.TrimPrefix(ct.Names[0], "/")
+	}
+	return svc
+}
+
+func fromSwarmService(s swarm.Service) protocol.Service {
+	ref := s.Spec.TaskTemplate.ContainerSpec.Image
+	return protocol.Service{
+		Kind:         protocol.KindSwarm,
+		SwarmService: s.Spec.Name,
+		Image:        ref,
+		Digest:       imageref.Parse(ref).Digest,
+		Status:       swarmStatus(s.ServiceStatus),
+	}
+}
+
+// mergeReplicas folds a second container of the same Compose service into
+// the first: the service counts as running if any replica runs.
+func mergeReplicas(a, b protocol.Service) protocol.Service {
+	if b.Status == "running" {
+		a.Status = "running"
+	}
+	a.Self = a.Self || b.Self
+	return a
+}
+
+// digestFor returns the registry digest the container's image was pulled
+// as. A reference that already pins a digest wins; otherwise the image's
+// RepoDigests entry for the same repository is used.
+func digestFor(ref string, repoDigests []string) string {
+	r := imageref.Parse(ref)
+	if r.Digest != "" {
+		return r.Digest
+	}
+	for _, rd := range repoDigests {
+		if p := imageref.Parse(rd); p.Repo == r.Repo {
+			return p.Digest
+		}
+	}
+	return ""
+}
+
+func containerStatus(state string) string {
+	if state == "running" {
+		return "running"
+	}
+	return "stopped"
+}
+
+func swarmStatus(st *swarm.ServiceStatus) string {
+	switch {
+	case st == nil:
+		return "unknown"
+	case st.DesiredTasks == 0:
+		return "stopped"
+	case st.RunningTasks >= st.DesiredTasks:
+		return "running"
+	default:
+		return "degraded"
+	}
+}
@@ -0,0 +1,170 @@
+package inventory
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/image"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/system"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+type fakeDocker struct {
+	containers       []container.Summary
+	images           []image.Summary
+	services         []swarm.Service
+	info             system.Info
+	containerErr     error
+	serviceListCalls int
+}
+
+func (f *fakeDocker) ContainerList(context.Context, container.ListOptions) ([]container.Summary, error) {
+	return f.containers, f.containerErr
+}
+
+func (f *fakeDocker) ImageList(context.Context, image.ListOptions) ([]image.Summary, error) {
+	return f.images, nil
+}
+
+func (f *fakeDocker) ServiceList(context.Context, swarm.ServiceListOptions) ([]swarm.Service, error) {
+	f.serviceListCalls++
+	return f.services, nil
+}
+
+func (f *fakeDocker) Info(context.Context) (system.Info, error) { return f.info, nil }
+
+func composeContainer(id, project, service, img, imageID, state string) container.Summary {
+	return container.Summary{
+		ID: id, Image: img, ImageID: imageID, State: state, Names: []string{"/" + project + "-" + service + "-1"},
+		Labels: map[string]string{
+			labelProject:    project,
+			labelService:    service,
+			labelWorkingDir: "/home/shcizo/self-hosted/" + project,
+		},
+	}
+}
+
+func managerInfo() system.Info {
+	var info system.Info
+	info.Name = "swarm-mgr"
+	info.Swarm.ControlAvailable = true
+	info.Swarm.Cluster = &swarm.ClusterInfo{ID: "cluster-1"}
+	return info
+}
+
+func TestSnapshot_ComposeContainerWithRepoDigest(t *testing.T) {
+	cli := &fakeDocker{
+		containers: []container.Summary{composeContainer("abc123", "slusk", "web", "gitea.shcizo.se/shcizo/slusk:1.4.2", "sha256:img1", "running")},
+		images:     []image.Summary{{ID: "sha256:img1", RepoDigests: []string{"gitea.shcizo.se/shcizo/slusk@sha256:d1"}}},
+	}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, []protocol.Service{{
+		Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
+		WorkingDir: "/home/shcizo/self-hosted/slusk",
+		Image:      "gitea.shcizo.se/shcizo/slusk:1.4.2", Digest: "sha256:d1", Status: "running",
+	}}, got)
+}
+
+func TestSnapshot_ReplicasMergeAndRunningWins(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		composeContainer("a1", "slusk", "worker", "slusk:1", "sha256:i", "exited"),
+		composeContainer("a2", "slusk", "worker", "slusk:1", "sha256:i", "running"),
+	}}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "running", got[0].Status)
+}
+
+func TestSnapshot_PlainContainerAndSwarmTaskSkipped(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		{ID: "p1", Names: []string{"/adhoc"}, Image: "redis:alpine", State: "exited"},
+		{ID: "t1", Names: []string{"/gitea_gitea.1.x"}, Image: "gitea/gitea:1.27.3", State: "running",
+			Labels: map[string]string{labelSwarmTask: "svc1"}},
+	}}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, []protocol.Service{{
+		Kind: protocol.KindContainer, ContainerName: "adhoc", Image: "redis:alpine", Status: "stopped",
+	}}, got)
+}
+
+func TestSnapshot_SelfFlag(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		composeContainer("0123456789abcdef", "updater-agent", "agent", "pu:1", "sha256:i", "running"),
+	}}
+	got, err := New(cli, "0123456789ab").Snapshot(context.Background())
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.True(t, got[0].Self)
+}
+
+func TestSnapshot_SwarmServicesOnlyOnManager(t *testing.T) {
+	svc := swarm.Service{
+		Spec: swarm.ServiceSpec{
+			Annotations:  swarm.Annotations{Name: "gitea_gitea"},
+			TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: "gitea/gitea:1.27.3@sha256:87a6"}},
+		},
+		ServiceStatus: &swarm.ServiceStatus{RunningTasks: 1, DesiredTasks: 1},
+	}
+
+	worker := &fakeDocker{services: []swarm.Service{svc}}
+	got, err := New(worker, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Empty(t, got)
+	assert.Equal(t, 0, worker.serviceListCalls, "workers must not call the manager-only API")
+
+	manager := &fakeDocker{services: []swarm.Service{svc}, info: managerInfo()}
+	got, err = New(manager, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, []protocol.Service{{
+		Kind: protocol.KindSwarm, SwarmService: "gitea_gitea",
+		Image: "gitea/gitea:1.27.3@sha256:87a6", Digest: "sha256:87a6", Status: "running",
+	}}, got)
+}
+
+func TestSwarmStatus(t *testing.T) {
+	cases := []struct {
+		name string
+		in   *swarm.ServiceStatus
+		want string
+	}{
+		{"no status", nil, "unknown"},
+		{"scaled to zero", &swarm.ServiceStatus{DesiredTasks: 0}, "stopped"},
+		{"all running", &swarm.ServiceStatus{RunningTasks: 2, DesiredTasks: 2}, "running"},
+		{"some running", &swarm.ServiceStatus{RunningTasks: 1, DesiredTasks: 2}, "degraded"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) { assert.Equal(t, c.want, swarmStatus(c.in)) })
+	}
+}
+
+func TestSnapshot_SortedByKey(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		{ID: "2", Names: []string{"/zeta"}, Image: "z", State: "running"},
+		{ID: "1", Names: []string{"/alpha"}, Image: "a", State: "running"},
+	}}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	require.Len(t, got, 2)
+	assert.Equal(t, "alpha", got[0].ContainerName)
+}
+
+func TestSnapshot_PropagatesDockerErrors(t *testing.T) {
+	cli := &fakeDocker{containerErr: errors.New("daemon down")}
+	_, err := New(cli, "").Snapshot(context.Background())
+	assert.ErrorContains(t, err, "daemon down")
+}
+
+func TestHello(t *testing.T) {
+	got, err := New(&fakeDocker{info: managerInfo()}, "").Hello(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "cluster-1"}, got)
+}
@@ -0,0 +1,25 @@
+package inventory
+
+import "github.com/shcizo/package-updater/internal/protocol"
+
+// Diff compares two snapshots. upserts holds services that are new or
+// changed in next; removes holds the keys of services missing from next.
+func Diff(prev, next []protocol.Service) (upserts []protocol.Service, removes []string) {
+	old := make(map[string]protocol.Service, len(prev))
+	for _, s := range prev {
+		old[s.Key()] = s
+	}
+	seen := make(map[string]struct{}, len(next))
+	for _, s := range next {
+		seen[s.Key()] = struct{}{}
+		if o, ok := old[s.Key()]; !ok || o != s {
+			upserts = append(upserts, s)
+		}
+	}
+	for _, s := range prev {
+		if _, ok := seen[s.Key()]; !ok {
+			removes = append(removes, s.Key())
+		}
+	}
+	return upserts, removes
+}
@@ -0,0 +1,37 @@
+package inventory
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+func TestDiff(t *testing.T) {
+	a := protocol.Service{Kind: protocol.KindContainer, ContainerName: "a", Image: "a:1", Status: "running"}
+	b := protocol.Service{Kind: protocol.KindContainer, ContainerName: "b", Image: "b:1", Status: "running"}
+	bStopped := b
+	bStopped.Status = "stopped"
+	c := protocol.Service{Kind: protocol.KindContainer, ContainerName: "c", Image: "c:1", Status: "running"}
+
+	cases := []struct {
+		name        string
+		prev, next  []protocol.Service
+		wantUpserts []protocol.Service
+		wantRemoves []string
+	}{
+		{"unchanged", []protocol.Service{a, b}, []protocol.Service{a, b}, nil, nil},
+		{"added", []protocol.Service{a}, []protocol.Service{a, c}, []protocol.Service{c}, nil},
+		{"changed", []protocol.Service{a, b}, []protocol.Service{a, bStopped}, []protocol.Service{bStopped}, nil},
+		{"removed", []protocol.Service{a, b}, []protocol.Service{a}, nil, []string{"container|b"}},
+		{"from empty", nil, []protocol.Service{a}, []protocol.Service{a}, nil},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ups, rms := Diff(tc.prev, tc.next)
+			assert.Equal(t, tc.wantUpserts, ups)
+			assert.Equal(t, tc.wantRemoves, rms)
+		})
+	}
+}
@@ -0,0 +1,52 @@
+package inventory
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/docker/docker/api/types/events"
+	"github.com/docker/docker/api/types/filters"
+)
+
+// EventSource is the subset of the Docker SDK Watch needs.
+// *client.Client satisfies it.
+type EventSource interface {
+	Events(ctx context.Context, opts events.ListOptions) (<-chan events.Message, <-chan error)
+}
+
+// Watch subscribes to container and service lifecycle events and calls
+// onChange once after each burst has been quiet for the given duration.
+// A Compose restart emits a dozen events; this turns them into one
+// inventory refresh. Returns nil when ctx ends, or an error if the event
+// stream breaks (for example when the daemon restarts) — the caller should
+// resubscribe and refresh, since events may have been missed.
+func Watch(ctx context.Context, src EventSource, quiet time.Duration, onChange func()) error {
+	args := filters.NewArgs(
+		filters.Arg("type", string(events.ContainerEventType)),
+		filters.Arg("type", string(events.ServiceEventType)),
+	)
+	for _, action := range []string{"create", "start", "stop", "die", "destroy", "rename", "update", "remove"} {
+		args.Add("event", action)
+	}
+	msgs, errs := src.Events(ctx, events.ListOptions{Filters: args})
+
+	timer := time.NewTimer(quiet)
+	timer.Stop()
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case err := <-errs:
+			if ctx.Err() != nil {
+				return nil
+			}
+			return fmt.Errorf("docker events: %w", err)
+		case <-msgs:
+			timer.Reset(quiet)
+		case <-timer.C:
+			onChange()
+		}
+	}
+}
@@ -0,0 +1,52 @@
+package inventory
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/docker/docker/api/types/events"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeEvents struct {
+	msgs chan events.Message
+	errs chan error
+}
+
+func newFakeEvents() *fakeEvents {
+	return &fakeEvents{msgs: make(chan events.Message, 10), errs: make(chan error, 1)}
+}
+
+func (f *fakeEvents) Events(context.Context, events.ListOptions) (<-chan events.Message, <-chan error) {
+	return f.msgs, f.errs
+}
+
+func TestWatch_DebouncesBursts(t *testing.T) {
+	src := newFakeEvents()
+	var calls atomic.Int32
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- Watch(ctx, src, 30*time.Millisecond, func() { calls.Add(1) }) }()
+
+	for range 3 {
+		src.msgs <- events.Message{Type: events.ContainerEventType, Action: "start"}
+	}
+	require.Eventually(t, func() bool { return calls.Load() == 1 }, time.Second, 5*time.Millisecond)
+
+	src.msgs <- events.Message{Type: events.ServiceEventType, Action: "update"}
+	require.Eventually(t, func() bool { return calls.Load() == 2 }, time.Second, 5*time.Millisecond)
+
+	cancel()
+	assert.NoError(t, <-done)
+}
+
+func TestWatch_ReturnsStreamError(t *testing.T) {
+	src := newFakeEvents()
+	src.errs <- errors.New("daemon restarted")
+	err := Watch(context.Background(), src, time.Millisecond, func() {})
+	assert.ErrorContains(t, err, "daemon restarted")
+}
@@ -0,0 +1,31 @@
+// Package agenttoken creates and hashes the per-agent tokens agents present
+// to the master.
+package agenttoken
+
+import (
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"fmt"
+)
+
+const prefix = "pua_"
+
+// Generate returns a new random agent token. Show it to the operator once;
+// store only Hash(token).
+func Generate() (string, error) {
+	var b [32]byte
+	if _, err := rand.Read(b[:]); err != nil {
+		return "", fmt.Errorf("generate agent token: %w", err)
+	}
+	return prefix + base64.RawURLEncoding.EncodeToString(b[:]), nil
+}
+
+// Hash returns the value stored for token. A plain SHA-256 is sufficient
+// because tokens carry 256 bits of randomness; salting protects low-entropy
+// passwords, which these are not.
+func Hash(token string) string {
+	sum := sha256.Sum256([]byte(token))
+	return hex.EncodeToString(sum[:])
+}
@@ -0,0 +1,25 @@
+package agenttoken
+
+import (
+	"regexp"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGenerate(t *testing.T) {
+	a, err := Generate()
+	require.NoError(t, err)
+	b, err := Generate()
+	require.NoError(t, err)
+
+	assert.Regexp(t, regexp.MustCompile(`^pua_[A-Za-z0-9_-]{43}$`), a)
+	assert.NotEqual(t, a, b)
+}
+
+func TestHash(t *testing.T) {
+	assert.Equal(t, Hash("pua_x"), Hash("pua_x"))
+	assert.NotEqual(t, Hash("pua_x"), Hash("pua_y"))
+	assert.Regexp(t, regexp.MustCompile(`^[0-9a-f]{64}$`), Hash("pua_x"))
+}
@@ -56,7 +56,12 @@ func (h *Handlers) Update(w http.ResponseWriter, r *http.Request) {
 		return
 	}

-	jobs, err := h.finder.FindJobs(r.Context(), req.Image)
+	requestedImage := req.Image
+	if req.Tag != "" {
+		requestedImage = req.Image + ":" + req.Tag
+	}
+
+	jobs, err := h.finder.FindJobs(r.Context(), requestedImage)
 	if err != nil {
 		writeJSONError(w, http.StatusInternalServerError, "discovery failed: "+err.Error())
 		return
@@ -86,6 +91,7 @@ func (h *Handlers) Update(w http.ResponseWriter, r *http.Request) {
 			Project:     res.Job.Project,
 			Service:     res.Job.Service,
 			ComposeFile: composeFile,
+			ServiceID:   res.Job.ServiceID,
 			Status:      string(res.Status),
 			Error:       res.Error,
 			DurationMs:  res.DurationMs,
@@ -21,9 +21,21 @@ import (
 type fakeFinder struct {
 	jobs []discovery.Job
 	err  error
+
+	// gotImage records the image argument passed to FindJobs, for
+	// assertions on the exact reference the handler built.
+	gotImage string
+	// swarmShaped, when true, makes FindJobs return a Job whose Image
+	// field carries the received image argument, simulating
+	// discovery.SwarmDiscovery.FindJobs.
+	swarmShaped bool
 }

-func (f *fakeFinder) FindJobs(_ context.Context, _ string) ([]discovery.Job, error) {
+func (f *fakeFinder) FindJobs(_ context.Context, image string) ([]discovery.Job, error) {
+	f.gotImage = image
+	if f.swarmShaped {
+		return []discovery.Job{{Service: "myapp_web", ServiceID: "svc1", Image: image}}, f.err
+	}
 	return f.jobs, f.err
 }

@@ -157,6 +169,42 @@ func TestHealthz_503WhenDockerDown(t *testing.T) {
 	require.Equal(t, http.StatusServiceUnavailable, w.Code)
 }

+func TestUpdate_FindJobsCalledWithTaggedImage(t *testing.T) {
+	finder := &fakeFinder{}
+	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
+	body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp", Tag: "v2"})
+	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
+	w := httptest.NewRecorder()
+	h.Update(w, req)
+	require.Equal(t, "registry.example.com/myapp:v2", finder.gotImage)
+}
+
+func TestUpdate_FindJobsCalledWithBareImageWhenNoTag(t *testing.T) {
+	finder := &fakeFinder{}
+	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
+	body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp"})
+	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
+	w := httptest.NewRecorder()
+	h.Update(w, req)
+	require.Equal(t, "registry.example.com/myapp", finder.gotImage)
+}
+
+func TestUpdate_SwarmModeThreadsTagThroughDiscovery(t *testing.T) {
+	finder := &fakeFinder{swarmShaped: true}
+	h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil)
+	body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp", Tag: "v2"})
+	req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body))
+	w := httptest.NewRecorder()
+	h.Update(w, req)
+	require.Equal(t, http.StatusOK, w.Code)
+	// The fake Finder's returned Job.Image (as SwarmDiscovery would build
+	// it) reflects the image argument it received from the handler.
+	// Asserting on the captured argument proves the full request ->
+	// discovery flow carries the tag through to what would reach the
+	// SwarmExecutor.
+	require.Equal(t, "registry.example.com/myapp:v2", finder.gotImage)
+}
+
 func TestVersion(t *testing.T) {
 	h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v1.2.3", "abcdef", "2026-05-22T00:00:00Z", nil)
 	req := httptest.NewRequest(http.MethodGet, "/version", nil)
@@ -0,0 +1,48 @@
+package api_test
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/docker/docker/api/types"
+	"github.com/shcizo/package-updater/internal/api"
+	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/shcizo/package-updater/internal/updater"
+	"github.com/stretchr/testify/require"
+)
+
+type swarmResultFinder struct{ jobs []discovery.Job }
+
+func (f *swarmResultFinder) FindJobs(_ context.Context, _ string) ([]discovery.Job, error) {
+	return f.jobs, nil
+}
+
+type swarmResultSubmitter struct{ results []updater.Result }
+
+func (s *swarmResultSubmitter) Submit(_ context.Context, _ []discovery.Job) []updater.Result {
+	return s.results
+}
+
+type swarmResultPinger struct{}
+
+func (swarmResultPinger) Ping(_ context.Context) (types.Ping, error) { return types.Ping{}, nil }
+
+func TestUpdate_SwarmJobPopulatesServiceID(t *testing.T) {
+	job := discovery.Job{Service: "myapp_web", ServiceID: "svc123", Image: "registry.example.com/myapp:v2"}
+	finder := &swarmResultFinder{jobs: []discovery.Job{job}}
+	submitter := &swarmResultSubmitter{results: []updater.Result{
+		{Job: job, Status: updater.StatusUpdated, DurationMs: 42},
+	}}
+	h := api.NewHandlers(finder, submitter, swarmResultPinger{}, "v", "c", "b", nil)
+
+	req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`{"image":"registry.example.com/myapp"}`))
+	rec := httptest.NewRecorder()
+	h.Update(rec, req)
+
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Contains(t, rec.Body.String(), `"service_id":"svc123"`)
+	require.Contains(t, rec.Body.String(), `"service":"myapp_web"`)
+}
@@ -20,6 +20,7 @@ type ResultDTO struct {
 	Project     string `json:"project"`
 	Service     string `json:"service"`
 	ComposeFile string `json:"compose_file"`
+	ServiceID   string `json:"service_id,omitempty"`
 	Status      string `json:"status"`
 	Error       string `json:"error,omitempty"`
 	DurationMs  int64  `json:"duration_ms"`
@@ -16,6 +16,7 @@ type Config struct {
 	LogLevel      string
 	UpdateTimeout time.Duration
 	OptInLabel    string
+	Mode          string
 }

 // Load reads configuration from environment variables, applies defaults,
@@ -28,12 +29,17 @@ func Load() (*Config, error) {
 		Port:       getenvDefault("PORT", "8080"),
 		LogLevel:   getenvDefault("LOG_LEVEL", "info"),
 		OptInLabel: getenvDefault("OPT_IN_LABEL", "se.shcizo.auto-update"),
+		Mode:       getenvDefault("MODE", "compose"),
 	}

 	if cfg.APIKey == "" {
 		return nil, errors.New("UPDATER_API_KEY is required")
 	}

+	if cfg.Mode != "compose" && cfg.Mode != "swarm" {
+		return nil, fmt.Errorf("MODE %q is invalid: must be %q or %q", cfg.Mode, "compose", "swarm")
+	}
+
 	timeoutStr := getenvDefault("UPDATE_TIMEOUT", "5m")
 	d, err := time.ParseDuration(timeoutStr)
 	if err != nil {
@@ -57,3 +57,26 @@ func TestLoad_InvalidTimeoutErrors(t *testing.T) {
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "UPDATE_TIMEOUT")
 }
+
+func TestLoad_DefaultsToComposeMode(t *testing.T) {
+	t.Setenv("UPDATER_API_KEY", "secret")
+	cfg, err := config.Load()
+	require.NoError(t, err)
+	require.Equal(t, "compose", cfg.Mode)
+}
+
+func TestLoad_AcceptsSwarmMode(t *testing.T) {
+	t.Setenv("UPDATER_API_KEY", "secret")
+	t.Setenv("MODE", "swarm")
+	cfg, err := config.Load()
+	require.NoError(t, err)
+	require.Equal(t, "swarm", cfg.Mode)
+}
+
+func TestLoad_RejectsInvalidMode(t *testing.T) {
+	t.Setenv("UPDATER_API_KEY", "secret")
+	t.Setenv("MODE", "kubernetes")
+	_, err := config.Load()
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "MODE")
+}
@@ -11,10 +11,16 @@ import (

 // Job describes a single (project, service, config_files) update to execute.
 type Job struct {
-	Project    string
-	Service    string
-	WorkingDir string
+	Project     string
+	Service     string
+	WorkingDir  string
 	ConfigFiles []string
+	// Image is the full image reference the update request asked for.
+	// Compose jobs ignore it (compose.yml already pins the reference to
+	// pull); Swarm jobs need it to know what to set on the service spec.
+	Image string
+	// ServiceID is the Swarm service ID. Empty for Compose jobs.
+	ServiceID string
 	// Refused is true when the WorkingDir falls outside STACKS_ROOT.
 	// The job is returned so the caller can surface a per-job "refused"
 	// result, but it MUST NOT be executed.
@@ -74,6 +80,7 @@ func (d *Discovery) FindJobs(ctx context.Context, image string) ([]Job, error) {
 			Service:       cl.Service,
 			WorkingDir:    cl.WorkingDir,
 			ConfigFiles:   cl.ConfigFiles,
+			Image:         image,
 			Refused:       refused,
 			RefusedReason: reason,
 		})
@@ -153,3 +153,20 @@ func TestFindJobs_NonComposeContainerIsSkipped(t *testing.T) {
 	require.NoError(t, err)
 	require.Empty(t, jobs)
 }
+
+func TestFindJobs_PopulatesImageField(t *testing.T) {
+	fake := &fakeDockerClient{containers: []types.Container{
+		mkContainer("registry.example.com/myapp:v1", mkComposeLabels(
+			"myapp-prod", "web",
+			"/home/shcizo/self-hosted/myapp-prod",
+			"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml",
+			true,
+		)),
+	}}
+	d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Len(t, jobs, 1)
+	require.Equal(t, "registry.example.com/myapp", jobs[0].Image)
+	require.Empty(t, jobs[0].ServiceID)
+}
@@ -0,0 +1,60 @@
+package discovery
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/docker/api/types/swarm"
+)
+
+// SwarmDockerClient is the subset of the Docker SDK SwarmDiscovery depends
+// on. Defined as an interface so tests can supply a fake.
+type SwarmDockerClient interface {
+	ServiceList(ctx context.Context, opts swarm.ServiceListOptions) ([]swarm.Service, error)
+}
+
+// SwarmDiscovery orchestrates "given an image, which Swarm services should
+// we update?". Unlike Discovery, it has no filesystem path to check — the
+// opt-in label on the service itself is the only gate, since Swarm services
+// have no local compose file to anchor a STACKS_ROOT check against.
+type SwarmDiscovery struct {
+	cli        SwarmDockerClient
+	optInLabel string
+}
+
+// NewSwarm returns a SwarmDiscovery bound to the given Docker client and
+// opt-in label.
+func NewSwarm(cli SwarmDockerClient, optInLabel string) *SwarmDiscovery {
+	return &SwarmDiscovery{cli: cli, optInLabel: optInLabel}
+}
+
+// FindJobs lists Swarm services, filters by image match + opt-in label, and
+// returns one Job per matching service. Signature matches Discovery.FindJobs
+// so both satisfy api.Finder.
+func (d *SwarmDiscovery) FindJobs(ctx context.Context, image string) ([]Job, error) {
+	services, err := d.cli.ServiceList(ctx, swarm.ServiceListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("docker service list: %w", err)
+	}
+
+	var jobs []Job
+	for _, svc := range services {
+		if svc.Spec.TaskTemplate.ContainerSpec == nil {
+			continue
+		}
+		if !ImagesMatch(image, svc.Spec.TaskTemplate.ContainerSpec.Image) {
+			continue
+		}
+		if !HasOptIn(svc.Spec.Labels, d.optInLabel) {
+			continue
+		}
+
+		jobs = append(jobs, Job{
+			Service:   svc.Spec.Name,
+			ServiceID: svc.ID,
+			Image:     image,
+		})
+	}
+
+	return jobs, nil
+}
@@ -0,0 +1,86 @@
+package discovery_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeSwarmDockerClient struct {
+	services []swarm.Service
+	err      error
+}
+
+func (f *fakeSwarmDockerClient) ServiceList(_ context.Context, _ swarm.ServiceListOptions) ([]swarm.Service, error) {
+	return f.services, f.err
+}
+
+func mkService(id, name, image string, labels map[string]string) swarm.Service {
+	return swarm.Service{
+		ID: id,
+		Spec: swarm.ServiceSpec{
+			Annotations: swarm.Annotations{Name: name, Labels: labels},
+			TaskTemplate: swarm.TaskSpec{
+				ContainerSpec: &swarm.ContainerSpec{Image: image},
+			},
+		},
+	}
+}
+
+func TestSwarmFindJobs_MatchAndOptIn(t *testing.T) {
+	fake := &fakeSwarmDockerClient{services: []swarm.Service{
+		mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", map[string]string{
+			"se.shcizo.auto-update": "true",
+		}),
+		mkService("svc-other", "other_web", "registry.example.com/other:v1", map[string]string{
+			"se.shcizo.auto-update": "true",
+		}),
+	}}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Len(t, jobs, 1)
+	require.Equal(t, "myapp_web", jobs[0].Service)
+	require.Equal(t, "svc-myapp", jobs[0].ServiceID)
+	require.Equal(t, "registry.example.com/myapp", jobs[0].Image)
+	require.False(t, jobs[0].Refused)
+	require.Empty(t, jobs[0].WorkingDir)
+	require.Empty(t, jobs[0].ConfigFiles)
+}
+
+func TestSwarmFindJobs_SkipsWithoutOptIn(t *testing.T) {
+	fake := &fakeSwarmDockerClient{services: []swarm.Service{
+		mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", nil),
+	}}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Empty(t, jobs)
+}
+
+func TestSwarmFindJobs_DockerError(t *testing.T) {
+	fake := &fakeSwarmDockerClient{err: errors.New("connection refused")}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	_, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.Error(t, err)
+}
+
+func TestSwarmFindJobs_NoContainerSpecIsSkipped(t *testing.T) {
+	fake := &fakeSwarmDockerClient{services: []swarm.Service{
+		{
+			ID: "svc-weird",
+			Spec: swarm.ServiceSpec{
+				Annotations:  swarm.Annotations{Name: "weird", Labels: map[string]string{"se.shcizo.auto-update": "true"}},
+				TaskTemplate: swarm.TaskSpec{ContainerSpec: nil},
+			},
+		},
+	}}
+	d := discovery.NewSwarm(fake, "se.shcizo.auto-update")
+	jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp")
+	require.NoError(t, err)
+	require.Empty(t, jobs)
+}
@@ -0,0 +1,43 @@
+// Package imageref splits container image references into repository, tag
+// and digest.
+package imageref
+
+import "strings"
+
+// Ref is an image reference split into its parts. Tag and Digest are empty
+// when the reference has none.
+type Ref struct {
+	Repo   string
+	Tag    string
+	Digest string
+}
+
+// Parse splits ref into repository, tag and digest. A colon before the last
+// "/" belongs to a registry host and port ("localhost:5000/foo"), not a tag.
+func Parse(ref string) Ref {
+	var r Ref
+	if at := strings.Index(ref, "@"); at >= 0 {
+		r.Digest = ref[at+1:]
+		ref = ref[:at]
+	}
+	slash := strings.LastIndex(ref, "/")
+	last := ref[slash+1:]
+	if colon := strings.Index(last, ":"); colon >= 0 {
+		r.Tag = last[colon+1:]
+		ref = ref[:slash+1+colon]
+	}
+	r.Repo = ref
+	return r
+}
+
+// String reassembles the reference as repo[:tag][@digest].
+func (r Ref) String() string {
+	s := r.Repo
+	if r.Tag != "" {
+		s += ":" + r.Tag
+	}
+	if r.Digest != "" {
+		s += "@" + r.Digest
+	}
+	return s
+}
@@ -0,0 +1,29 @@
+package imageref
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParse(t *testing.T) {
+	cases := []struct {
+		in   string
+		want Ref
+	}{
+		{"postgres", Ref{Repo: "postgres"}},
+		{"postgres:17.10", Ref{Repo: "postgres", Tag: "17.10"}},
+		{"gitea/gitea:1.27.3@sha256:87a6", Ref{Repo: "gitea/gitea", Tag: "1.27.3", Digest: "sha256:87a6"}},
+		{"lscr.io/linuxserver/jellyfin@sha256:0f42", Ref{Repo: "lscr.io/linuxserver/jellyfin", Digest: "sha256:0f42"}},
+		{"localhost:5000/foo", Ref{Repo: "localhost:5000/foo"}},
+		{"localhost:5000/foo:v1", Ref{Repo: "localhost:5000/foo", Tag: "v1"}},
+		{"localhost:5000/foo:v1@sha256:ab", Ref{Repo: "localhost:5000/foo", Tag: "v1", Digest: "sha256:ab"}},
+	}
+	for _, c := range cases {
+		t.Run(c.in, func(t *testing.T) {
+			got := Parse(c.in)
+			assert.Equal(t, c.want, got)
+			assert.Equal(t, c.in, got.String(), "String must round-trip")
+		})
+	}
+}
@@ -0,0 +1,88 @@
+// Package model holds the master's domain types shared by storage, the agent
+// hub and the web UI.
+package model
+
+import (
+	"errors"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// Sentinel errors returned by storage.
+var (
+	ErrNotFound = errors.New("not found")
+	ErrConflict = errors.New("already exists")
+)
+
+// Agent is a registered agent. Whether it is connected right now is held in
+// memory by the hub, not here.
+type Agent struct {
+	ID           int64
+	Name         string
+	Hostname     string
+	AgentVersion string
+	SwarmManager bool
+	LastSeenAt   *time.Time
+	RevokedAt    *time.Time
+	CreatedAt    time.Time
+}
+
+// Service is one service as last reported by an agent.
+type Service struct {
+	ID             int64
+	Kind           protocol.Kind
+	LocationKey    string
+	AgentID        *int64
+	AgentName      string
+	ComposeProject string
+	ComposeService string
+	WorkingDir     string
+	SwarmClusterID string
+	SwarmService   string
+	ContainerName  string
+	ImageRepo      string
+	ActualRef      string
+	ActualDigest   string
+	Status         string
+	Self           bool
+	UpdateMode     string
+	FirstSeenAt    time.Time
+	UpdatedAt      time.Time
+}
+
+// DisplayName is the human-facing name of the service.
+func (s Service) DisplayName() string {
+	switch s.Kind {
+	case protocol.KindCompose:
+		return s.ComposeProject + "/" + s.ComposeService
+	case protocol.KindSwarm:
+		return s.SwarmService
+	default:
+		return s.ContainerName
+	}
+}
+
+// Session is a logged-in browser session.
+type Session struct {
+	ID        string
+	Subject   string
+	Name      string
+	Email     string
+	CSRFToken string
+	ExpiresAt time.Time
+}
+
+// LocationKey turns an agent-relative protocol key into the fleet-wide key
+// stored in services.location_key. Compose services and containers belong to
+// the agent's machine; Swarm services belong to the cluster, so two manager
+// agents in one cluster report the same row.
+func LocationKey(agentID int64, clusterID, agentKey string) string {
+	kind, rest, _ := strings.Cut(agentKey, "|")
+	if kind == string(protocol.KindSwarm) {
+		return kind + "|" + clusterID + "|" + rest
+	}
+	return kind + "|" + strconv.FormatInt(agentID, 10) + "|" + rest
+}
@@ -0,0 +1,41 @@
+package model
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+func TestLocationKey(t *testing.T) {
+	cases := []struct {
+		name, agentKey, want string
+	}{
+		{"compose is per agent", "compose|slusk|web", "compose|7|slusk|web"},
+		{"container is per agent", "container|adhoc", "container|7|adhoc"},
+		{"swarm is per cluster", "swarm|gitea_gitea", "swarm|cluster-1|gitea_gitea"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, LocationKey(7, "cluster-1", c.agentKey))
+		})
+	}
+}
+
+func TestDisplayName(t *testing.T) {
+	cases := []struct {
+		name string
+		svc  Service
+		want string
+	}{
+		{"compose", Service{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web"}, "slusk/web"},
+		{"swarm", Service{Kind: protocol.KindSwarm, SwarmService: "gitea_gitea"}, "gitea_gitea"},
+		{"container", Service{Kind: protocol.KindContainer, ContainerName: "adhoc"}, "adhoc"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, c.svc.DisplayName())
+		})
+	}
+}
@@ -0,0 +1,91 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+const agentColumns = `id, name, hostname, agent_version, swarm_manager, last_seen_at, revoked_at, created_at`
+
+func scanAgent(row pgx.Row) (model.Agent, error) {
+	var a model.Agent
+	err := row.Scan(&a.ID, &a.Name, &a.Hostname, &a.AgentVersion, &a.SwarmManager, &a.LastSeenAt, &a.RevokedAt, &a.CreatedAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return model.Agent{}, model.ErrNotFound
+	}
+	return a, err
+}
+
+// CreateAgent registers an agent. Returns model.ErrConflict if the name is taken.
+func (s *Store) CreateAgent(ctx context.Context, name, tokenHash string) (model.Agent, error) {
+	a, err := scanAgent(s.pool.QueryRow(ctx,
+		`INSERT INTO agents (name, token_hash) VALUES ($1, $2) RETURNING `+agentColumns, name, tokenHash))
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
+		return model.Agent{}, model.ErrConflict
+	}
+	if err != nil {
+		return model.Agent{}, fmt.Errorf("create agent: %w", err)
+	}
+	return a, nil
+}
+
+// ListAgents returns all agents, including revoked ones, ordered by name.
+func (s *Store) ListAgents(ctx context.Context) ([]model.Agent, error) {
+	rows, err := s.pool.Query(ctx, `SELECT `+agentColumns+` FROM agents ORDER BY name`)
+	if err != nil {
+		return nil, fmt.Errorf("list agents: %w", err)
+	}
+	defer rows.Close()
+	var out []model.Agent
+	for rows.Next() {
+		a, err := scanAgent(rows)
+		if err != nil {
+			return nil, fmt.Errorf("scan agent: %w", err)
+		}
+		out = append(out, a)
+	}
+	return out, rows.Err()
+}
+
+// AgentByTokenHash finds the agent a token belongs to. Revoked agents are
+// returned too; the caller decides. Returns model.ErrNotFound if none.
+func (s *Store) AgentByTokenHash(ctx context.Context, hash string) (model.Agent, error) {
+	return scanAgent(s.pool.QueryRow(ctx, `SELECT `+agentColumns+` FROM agents WHERE token_hash = $1`, hash))
+}
+
+// RevokeAgent marks the agent revoked. Its token stops working immediately;
+// closing a live connection is the hub's job.
+func (s *Store) RevokeAgent(ctx context.Context, id int64, at time.Time) error {
+	if _, err := s.pool.Exec(ctx,
+		`UPDATE agents SET revoked_at = $2 WHERE id = $1 AND revoked_at IS NULL`, id, at); err != nil {
+		return fmt.Errorf("revoke agent: %w", err)
+	}
+	return nil
+}
+
+// RecordHello stores what the agent reported about itself on connect.
+func (s *Store) RecordHello(ctx context.Context, id int64, h protocol.Hello, at time.Time) error {
+	if _, err := s.pool.Exec(ctx,
+		`UPDATE agents SET hostname = $2, agent_version = $3, swarm_manager = $4, last_seen_at = $5 WHERE id = $1`,
+		id, h.Hostname, h.AgentVersion, h.SwarmManager, at); err != nil {
+		return fmt.Errorf("record hello: %w", err)
+	}
+	return nil
+}
+
+// MarkSeen updates last_seen_at, called when an agent disconnects.
+func (s *Store) MarkSeen(ctx context.Context, id int64, at time.Time) error {
+	if _, err := s.pool.Exec(ctx, `UPDATE agents SET last_seen_at = $2 WHERE id = $1`, id, at); err != nil {
+		return fmt.Errorf("mark agent seen: %w", err)
+	}
+	return nil
+}
@@ -0,0 +1,84 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/pin"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+func TestIdentity_CreatedOnceThenReused(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	k1, err := s.LoadOrCreateIdentity(ctx, pin.GenerateKey)
+	require.NoError(t, err)
+	k2, err := s.LoadOrCreateIdentity(ctx, pin.GenerateKey)
+	require.NoError(t, err)
+	assert.True(t, k1.Equal(k2), "a restarted master must keep its key or every agent breaks")
+}
+
+func TestMigrationsAreIdempotent(t *testing.T) {
+	s := openTestStore(t)
+	require.NoError(t, migrate(context.Background(), s.pool))
+}
+
+func TestAgents_CreateLookupRevoke(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+
+	created, err := s.CreateAgent(ctx, "node-2", "hash-2")
+	require.NoError(t, err)
+	assert.Equal(t, "node-2", created.Name)
+	assert.Nil(t, created.RevokedAt)
+
+	found, err := s.AgentByTokenHash(ctx, "hash-2")
+	require.NoError(t, err)
+	assert.Equal(t, created.ID, found.ID)
+
+	_, err = s.AgentByTokenHash(ctx, "nope")
+	assert.ErrorIs(t, err, model.ErrNotFound)
+
+	require.NoError(t, s.RevokeAgent(ctx, created.ID, time.Now()))
+	found, err = s.AgentByTokenHash(ctx, "hash-2")
+	require.NoError(t, err)
+	assert.NotNil(t, found.RevokedAt)
+}
+
+func TestAgents_DuplicateNameIsConflict(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	_, err := s.CreateAgent(ctx, "node-2", "h1")
+	require.NoError(t, err)
+	_, err = s.CreateAgent(ctx, "node-2", "h2")
+	assert.ErrorIs(t, err, model.ErrConflict)
+}
+
+func TestAgents_RecordHelloAndMarkSeen(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a, err := s.CreateAgent(ctx, "swarm-mgr", "h")
+	require.NoError(t, err)
+
+	at := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
+	require.NoError(t, s.RecordHello(ctx, a.ID, protocol.Hello{
+		Hostname: "swarm-mgr", AgentVersion: "v1", SwarmManager: true, SwarmClusterID: "c1",
+	}, at))
+	require.NoError(t, s.MarkSeen(ctx, a.ID, at.Add(time.Minute)))
+
+	agents, err := s.ListAgents(ctx)
+	require.NoError(t, err)
+	require.Len(t, agents, 1)
+	assert.Equal(t, "swarm-mgr", agents[0].Hostname)
+	assert.Equal(t, "v1", agents[0].AgentVersion)
+	assert.True(t, agents[0].SwarmManager)
+	require.NotNil(t, agents[0].LastSeenAt)
+	assert.True(t, agents[0].LastSeenAt.Equal(at.Add(time.Minute)))
+}
@@ -0,0 +1,43 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stretchr/testify/require"
+)
+
+// openTestStore returns a Store migrated into a fresh schema that is dropped
+// when the test ends, so tests never see each other's rows.
+func openTestStore(t *testing.T) *Store {
+	t.Helper()
+	url := os.Getenv("DATABASE_URL")
+	if url == "" {
+		t.Skip("DATABASE_URL not set")
+	}
+	ctx := context.Background()
+	admin, err := pgx.Connect(ctx, url)
+	require.NoError(t, err)
+	schema := fmt.Sprintf("test_%d", time.Now().UnixNano())
+	_, err = admin.Exec(ctx, "CREATE SCHEMA "+schema)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_, _ = admin.Exec(ctx, "DROP SCHEMA "+schema+" CASCADE")
+		_ = admin.Close(ctx)
+	})
+
+	cfg, err := pgxpool.ParseConfig(url)
+	require.NoError(t, err)
+	cfg.ConnConfig.RuntimeParams["search_path"] = schema
+	s, err := openWithConfig(ctx, cfg)
+	require.NoError(t, err)
+	t.Cleanup(s.Close)
+	return s
+}
@@ -0,0 +1,42 @@
+package store
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/x509"
+	"errors"
+	"fmt"
+)
+
+// LoadOrCreateIdentity returns the master's private key, creating and
+// storing one with generate on first start. Keeping it in Postgres means it
+// survives container recreation and is covered by database backups; losing
+// it would force every agent to be reconfigured with a new MASTER_KEY.
+func (s *Store) LoadOrCreateIdentity(ctx context.Context, generate func() (ed25519.PrivateKey, error)) (ed25519.PrivateKey, error) {
+	fresh, err := generate()
+	if err != nil {
+		return nil, err
+	}
+	der, err := x509.MarshalPKCS8PrivateKey(fresh)
+	if err != nil {
+		return nil, fmt.Errorf("marshal master key: %w", err)
+	}
+	if _, err := s.pool.Exec(ctx,
+		`INSERT INTO master_identity (id, private_key) VALUES (1, $1) ON CONFLICT (id) DO NOTHING`, der,
+	); err != nil {
+		return nil, fmt.Errorf("store master key: %w", err)
+	}
+	var stored []byte
+	if err := s.pool.QueryRow(ctx, `SELECT private_key FROM master_identity WHERE id = 1`).Scan(&stored); err != nil {
+		return nil, fmt.Errorf("load master key: %w", err)
+	}
+	parsed, err := x509.ParsePKCS8PrivateKey(stored)
+	if err != nil {
+		return nil, fmt.Errorf("parse master key: %w", err)
+	}
+	key, ok := parsed.(ed25519.PrivateKey)
+	if !ok {
+		return nil, errors.New("stored master key is not ed25519")
+	}
+	return key, nil
+}
@@ -0,0 +1,68 @@
+package store
+
+import (
+	"context"
+	"embed"
+	"fmt"
+	"io/fs"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+//go:embed migrations/*.sql
+var migrationsFS embed.FS
+
+// migrate applies every embedded migration not yet recorded in
+// schema_migrations, in file-name order, each in its own transaction.
+func migrate(ctx context.Context, pool *pgxpool.Pool) error {
+	if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
+		version    text PRIMARY KEY,
+		applied_at timestamptz NOT NULL DEFAULT now()
+	)`); err != nil {
+		return fmt.Errorf("create schema_migrations: %w", err)
+	}
+	entries, err := fs.ReadDir(migrationsFS, "migrations")
+	if err != nil {
+		return fmt.Errorf("read migrations: %w", err)
+	}
+	for _, e := range entries {
+		name := e.Name()
+		var applied bool
+		if err := pool.QueryRow(ctx,
+			`SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version = $1)`, name,
+		).Scan(&applied); err != nil {
+			return fmt.Errorf("check migration %s: %w", name, err)
+		}
+		if applied {
+			continue
+		}
+		sql, err := migrationsFS.ReadFile("migrations/" + name)
+		if err != nil {
+			return fmt.Errorf("read migration %s: %w", name, err)
+		}
+		if err := applyMigration(ctx, pool, name, string(sql)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func applyMigration(ctx context.Context, pool *pgxpool.Pool, name, sql string) error {
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin migration %s: %w", name, err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	// Exec without arguments uses the simple protocol, which allows the
+	// several statements a migration file contains.
+	if _, err := tx.Exec(ctx, sql); err != nil {
+		return fmt.Errorf("migration %s: %w", name, err)
+	}
+	if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, name); err != nil {
+		return fmt.Errorf("record migration %s: %w", name, err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit migration %s: %w", name, err)
+	}
+	return nil
+}
@@ -0,0 +1,48 @@
+CREATE TABLE master_identity (
+    id          int PRIMARY KEY CHECK (id = 1),
+    private_key bytea NOT NULL,
+    created_at  timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE agents (
+    id            bigserial PRIMARY KEY,
+    name          text NOT NULL UNIQUE,
+    token_hash    text NOT NULL UNIQUE,
+    hostname      text NOT NULL DEFAULT '',
+    agent_version text NOT NULL DEFAULT '',
+    swarm_manager boolean NOT NULL DEFAULT false,
+    last_seen_at  timestamptz,
+    revoked_at    timestamptz,
+    created_at    timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE services (
+    id               bigserial PRIMARY KEY,
+    kind             text NOT NULL CHECK (kind IN ('compose', 'swarm', 'container')),
+    location_key     text NOT NULL UNIQUE,
+    agent_id         bigint REFERENCES agents (id),
+    compose_project  text NOT NULL DEFAULT '',
+    compose_service  text NOT NULL DEFAULT '',
+    working_dir      text NOT NULL DEFAULT '',
+    swarm_cluster_id text NOT NULL DEFAULT '',
+    swarm_service    text NOT NULL DEFAULT '',
+    container_name   text NOT NULL DEFAULT '',
+    image_repo       text NOT NULL,
+    actual_ref       text NOT NULL,
+    actual_digest    text NOT NULL DEFAULT '',
+    status           text NOT NULL,
+    self             boolean NOT NULL DEFAULT false,
+    update_mode      text NOT NULL DEFAULT 'manual' CHECK (update_mode IN ('manual', 'auto')),
+    first_seen_at    timestamptz NOT NULL,
+    updated_at       timestamptz NOT NULL,
+    removed_at       timestamptz
+);
+
+CREATE TABLE sessions (
+    id         text PRIMARY KEY,
+    subject    text NOT NULL,
+    name       text NOT NULL DEFAULT '',
+    email      text NOT NULL DEFAULT '',
+    csrf_token text NOT NULL,
+    expires_at timestamptz NOT NULL
+);
@@ -0,0 +1,136 @@
+package store
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+
+	"github.com/shcizo/package-updater/internal/imageref"
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+type execer interface {
+	Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error)
+}
+
+const upsertServiceSQL = `
+INSERT INTO services (
+	kind, location_key, agent_id, compose_project, compose_service, working_dir,
+	swarm_cluster_id, swarm_service, container_name, image_repo, actual_ref, actual_digest,
+	status, self, first_seen_at, updated_at)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $15)
+ON CONFLICT (location_key) DO UPDATE SET
+	agent_id      = EXCLUDED.agent_id,
+	working_dir   = EXCLUDED.working_dir,
+	image_repo    = EXCLUDED.image_repo,
+	actual_ref    = EXCLUDED.actual_ref,
+	actual_digest = EXCLUDED.actual_digest,
+	status        = EXCLUDED.status,
+	self          = EXCLUDED.self,
+	updated_at    = EXCLUDED.updated_at,
+	removed_at    = NULL`
+
+func upsertService(ctx context.Context, db execer, agentID int64, clusterID string, svc protocol.Service, at time.Time) error {
+	clusterCol := ""
+	if svc.Kind == protocol.KindSwarm {
+		clusterCol = clusterID
+	}
+	_, err := db.Exec(ctx, upsertServiceSQL,
+		string(svc.Kind), model.LocationKey(agentID, clusterID, svc.Key()), agentID,
+		svc.ComposeProject, svc.ComposeService, svc.WorkingDir,
+		clusterCol, svc.SwarmService, svc.ContainerName,
+		imageref.Parse(svc.Image).Repo, svc.Image, svc.Digest,
+		svc.Status, svc.Self, at)
+	if err != nil {
+		return fmt.Errorf("upsert service %s: %w", svc.Key(), err)
+	}
+	return nil
+}
+
+// ApplySnapshot makes the stored inventory for this agent's scope equal to
+// services: everything listed is upserted, everything in scope but not
+// listed is marked removed. Scope is the agent's own Compose services and
+// containers, plus the cluster's Swarm services if the agent is a manager.
+// A manager without a cluster ID is treated as a non-manager, since its
+// Swarm rows could not be told apart from other agents' rows.
+func (s *Store) ApplySnapshot(ctx context.Context, agentID int64, h protocol.Hello, services []protocol.Service, at time.Time) error {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin snapshot: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	isManager := h.SwarmManager && h.SwarmClusterID != ""
+	keys := make([]string, 0, len(services))
+	for _, svc := range services {
+		if svc.Kind == protocol.KindSwarm && !isManager {
+			continue
+		}
+		if err := upsertService(ctx, tx, agentID, h.SwarmClusterID, svc, at); err != nil {
+			return err
+		}
+		keys = append(keys, model.LocationKey(agentID, h.SwarmClusterID, svc.Key()))
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE services SET removed_at = $1
+		WHERE removed_at IS NULL
+		  AND NOT (location_key = ANY($2))
+		  AND ((kind IN ('compose', 'container') AND agent_id = $3)
+		    OR (kind = 'swarm' AND $4 AND swarm_cluster_id = $5))`,
+		at, keys, agentID, isManager, h.SwarmClusterID); err != nil {
+		return fmt.Errorf("mark removed services: %w", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit snapshot: %w", err)
+	}
+	return nil
+}
+
+// UpsertService stores one changed or new service.
+func (s *Store) UpsertService(ctx context.Context, agentID int64, clusterID string, svc protocol.Service, at time.Time) error {
+	return upsertService(ctx, s.pool, agentID, clusterID, svc, at)
+}
+
+// RemoveService marks the service with the agent-relative key as removed.
+func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, agentKey string, at time.Time) error {
+	if _, err := s.pool.Exec(ctx,
+		`UPDATE services SET removed_at = $2 WHERE location_key = $1 AND removed_at IS NULL`,
+		model.LocationKey(agentID, clusterID, agentKey), at); err != nil {
+		return fmt.Errorf("remove service: %w", err)
+	}
+	return nil
+}
+
+// ListServices returns all services not marked removed: machine-bound
+// services grouped by agent name first, Swarm services last.
+func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) {
+	rows, err := s.pool.Query(ctx, `
+		SELECT s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''),
+		       s.compose_project, s.compose_service, s.working_dir, s.swarm_cluster_id,
+		       s.swarm_service, s.container_name, s.image_repo, s.actual_ref, s.actual_digest,
+		       s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at
+		FROM services s LEFT JOIN agents a ON a.id = s.agent_id
+		WHERE s.removed_at IS NULL
+		ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`)
+	if err != nil {
+		return nil, fmt.Errorf("list services: %w", err)
+	}
+	defer rows.Close()
+	var out []model.Service
+	for rows.Next() {
+		var m model.Service
+		var kind string
+		if err := rows.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName,
+			&m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.SwarmClusterID,
+			&m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.ActualRef, &m.ActualDigest,
+			&m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt); err != nil {
+			return nil, fmt.Errorf("scan service: %w", err)
+		}
+		m.Kind = protocol.Kind(kind)
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
@@ -0,0 +1,162 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+var (
+	t0      = time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
+	slusk   = protocol.Service{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", WorkingDir: "/srv/slusk", Image: "gitea.shcizo.se/shcizo/slusk:1.4.2", Digest: "sha256:d1", Status: "running"}
+	adhoc   = protocol.Service{Kind: protocol.KindContainer, ContainerName: "adhoc", Image: "redis:alpine", Status: "stopped"}
+	gitea   = protocol.Service{Kind: protocol.KindSwarm, SwarmService: "gitea_gitea", Image: "gitea/gitea:1.27.3@sha256:87a6", Digest: "sha256:87a6", Status: "running"}
+	worker  = protocol.Hello{Hostname: "node-2"}
+	manager = protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "c1"}
+)
+
+func names(svcs []model.Service) []string {
+	var out []string
+	for _, s := range svcs {
+		out = append(out, s.DisplayName())
+	}
+	return out
+}
+
+func newAgent(t *testing.T, s *Store, name string) int64 {
+	t.Helper()
+	a, err := s.CreateAgent(context.Background(), name, "hash-"+name)
+	require.NoError(t, err)
+	return a.ID
+}
+
+func TestApplySnapshot_InsertsAsManualWithAgentName(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "node-2")
+
+	require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk}, t0))
+
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "manual", got[0].UpdateMode)
+	assert.Equal(t, "node-2", got[0].AgentName)
+	assert.Equal(t, "gitea.shcizo.se/shcizo/slusk", got[0].ImageRepo)
+	assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:1.4.2", got[0].ActualRef)
+	assert.Equal(t, "sha256:d1", got[0].ActualDigest)
+	assert.Equal(t, "compose|"+itoa(id)+"|slusk|web", got[0].LocationKey)
+}
+
+func TestApplySnapshot_RemovedWhileOfflineThenBack(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "node-2")
+
+	require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk, adhoc}, t0))
+	before, err := s.ListServices(ctx)
+	require.NoError(t, err)
+
+	require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk}, t0.Add(time.Hour)))
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"slusk/web"}, names(got))
+
+	require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk, adhoc}, t0.Add(2*time.Hour)))
+	after, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	require.Len(t, after, 2)
+	for i := range before {
+		assert.Equal(t, before[i].ID, after[i].ID, "a returning service keeps its row")
+		assert.True(t, before[i].FirstSeenAt.Equal(after[i].FirstSeenAt))
+	}
+}
+
+func TestApplySnapshot_TwoManagersShareSwarmRowsAndKeepOwnCompose(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a := newAgent(t, s, "swarm-mgr")
+	b := newAgent(t, s, "swarm-mgr-2")
+
+	require.NoError(t, s.ApplySnapshot(ctx, a, manager, []protocol.Service{gitea, slusk}, t0))
+	require.NoError(t, s.ApplySnapshot(ctx, b, manager, []protocol.Service{gitea}, t0))
+
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"slusk/web", "gitea_gitea"}, names(got),
+		"one swarm row, and b's snapshot must not remove a's compose service")
+}
+
+func TestApplySnapshot_IgnoresSwarmFromWorker(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "node-2")
+	require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{gitea}, t0))
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.Empty(t, got)
+}
+
+func TestApplySnapshot_IgnoresSwarmFromManagerWithoutClusterID(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	a := newAgent(t, s, "mgr-a")
+	b := newAgent(t, s, "mgr-b")
+	noCluster := protocol.Hello{Hostname: "mgr", SwarmManager: true}
+
+	require.NoError(t, s.ApplySnapshot(ctx, a, manager, []protocol.Service{gitea}, t0))
+	require.NoError(t, s.ApplySnapshot(ctx, b, noCluster, []protocol.Service{gitea}, t0))
+
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	require.Len(t, got, 1, "the cluster-less manager adds no swarm row")
+	assert.Equal(t, "swarm|c1|gitea_gitea", got[0].LocationKey)
+
+	require.NoError(t, s.ApplySnapshot(ctx, b, noCluster, nil, t0.Add(time.Hour)))
+	got, err = s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.Len(t, got, 1, "the cluster-less manager removes no swarm rows")
+}
+
+func TestUpsertAndRemoveService(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "swarm-mgr")
+
+	require.NoError(t, s.UpsertService(ctx, id, "c1", gitea, t0))
+	changed := gitea
+	changed.Status = "degraded"
+	require.NoError(t, s.UpsertService(ctx, id, "c1", changed, t0.Add(time.Minute)))
+
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "degraded", got[0].Status)
+	assert.Equal(t, "swarm|c1|gitea_gitea", got[0].LocationKey)
+
+	require.NoError(t, s.RemoveService(ctx, id, "c1", gitea.Key(), t0.Add(2*time.Minute)))
+	got, err = s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.Empty(t, got)
+}
+
+func TestListServices_SwarmLast(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "swarm-mgr")
+	require.NoError(t, s.ApplySnapshot(ctx, id, manager, []protocol.Service{gitea, slusk}, t0))
+	got, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"slusk/web", "gitea_gitea"}, names(got))
+}
+
+func itoa(n int64) string { return strconv.FormatInt(n, 10) }
@@ -0,0 +1,50 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+)
+
+// CreateSession stores a new session and deletes expired ones, so the table
+// never grows without bound.
+func (s *Store) CreateSession(ctx context.Context, sess model.Session) error {
+	if _, err := s.pool.Exec(ctx, `DELETE FROM sessions WHERE expires_at < now()`); err != nil {
+		return fmt.Errorf("delete expired sessions: %w", err)
+	}
+	if _, err := s.pool.Exec(ctx,
+		`INSERT INTO sessions (id, subject, name, email, csrf_token, expires_at) VALUES ($1, $2, $3, $4, $5, $6)`,
+		sess.ID, sess.Subject, sess.Name, sess.Email, sess.CSRFToken, sess.ExpiresAt); err != nil {
+		return fmt.Errorf("create session: %w", err)
+	}
+	return nil
+}
+
+// GetSession returns the session unless it is missing or expired at now,
+// in which case it returns model.ErrNotFound.
+func (s *Store) GetSession(ctx context.Context, id string, now time.Time) (model.Session, error) {
+	var m model.Session
+	err := s.pool.QueryRow(ctx,
+		`SELECT id, subject, name, email, csrf_token, expires_at FROM sessions WHERE id = $1 AND expires_at > $2`,
+		id, now).Scan(&m.ID, &m.Subject, &m.Name, &m.Email, &m.CSRFToken, &m.ExpiresAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return model.Session{}, model.ErrNotFound
+	}
+	if err != nil {
+		return model.Session{}, fmt.Errorf("get session: %w", err)
+	}
+	return m, nil
+}
+
+// DeleteSession removes a session (logout).
+func (s *Store) DeleteSession(ctx context.Context, id string) error {
+	if _, err := s.pool.Exec(ctx, `DELETE FROM sessions WHERE id = $1`, id); err != nil {
+		return fmt.Errorf("delete session: %w", err)
+	}
+	return nil
+}
@@ -0,0 +1,34 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+)
+
+func TestSessions(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
+	sess := model.Session{ID: "s1", Subject: "sub", Name: "Samuel", Email: "s@example.com", CSRFToken: "c1", ExpiresAt: now.Add(time.Hour)}
+
+	require.NoError(t, s.CreateSession(ctx, sess))
+	got, err := s.GetSession(ctx, "s1", now)
+	require.NoError(t, err)
+	assert.Equal(t, "Samuel", got.Name)
+	assert.Equal(t, "c1", got.CSRFToken)
+
+	_, err = s.GetSession(ctx, "s1", now.Add(2*time.Hour))
+	assert.ErrorIs(t, err, model.ErrNotFound, "expired sessions are not returned")
+
+	require.NoError(t, s.DeleteSession(ctx, "s1"))
+	_, err = s.GetSession(ctx, "s1", now)
+	assert.ErrorIs(t, err, model.ErrNotFound)
+}
@@ -0,0 +1,41 @@
+// Package store persists the master's state in Postgres.
+package store
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// Store is the Postgres-backed storage for the master.
+type Store struct {
+	pool *pgxpool.Pool
+}
+
+// Open connects to Postgres and applies pending migrations.
+func Open(ctx context.Context, url string) (*Store, error) {
+	cfg, err := pgxpool.ParseConfig(url)
+	if err != nil {
+		return nil, fmt.Errorf("parse DATABASE_URL: %w", err)
+	}
+	return openWithConfig(ctx, cfg)
+}
+
+func openWithConfig(ctx context.Context, cfg *pgxpool.Config) (*Store, error) {
+	pool, err := pgxpool.NewWithConfig(ctx, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("connect postgres: %w", err)
+	}
+	if err := migrate(ctx, pool); err != nil {
+		pool.Close()
+		return nil, err
+	}
+	return &Store{pool: pool}, nil
+}
+
+// Close releases the connection pool.
+func (s *Store) Close() { s.pool.Close() }
+
+// Ping checks that Postgres answers. Used by /healthz.
+func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }
@@ -0,0 +1,91 @@
+// Package pin lets agents trust the master without a certificate authority.
+// The master serves a self-signed certificate for its own key pair; agents
+// are configured with the fingerprint of that public key and accept the
+// connection only if it matches. The TLS handshake proves the master holds
+// the private key, the same way SSH host keys work.
+package pin
+
+import (
+	"crypto"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/sha256"
+	"crypto/subtle"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"math/big"
+	"time"
+)
+
+// GenerateKey creates a new master key pair.
+func GenerateKey() (ed25519.PrivateKey, error) {
+	_, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		return nil, fmt.Errorf("generate key: %w", err)
+	}
+	return priv, nil
+}
+
+// Certificate returns a self-signed TLS certificate for key. Its validity is
+// effectively unlimited because trust comes from the pinned fingerprint, not
+// from the certificate's dates or issuer.
+func Certificate(key ed25519.PrivateKey) (tls.Certificate, error) {
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "package-updater-master"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().AddDate(100, 0, 0),
+		KeyUsage:     x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:  []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+	}
+	der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, key.Public(), key)
+	if err != nil {
+		return tls.Certificate{}, fmt.Errorf("create certificate: %w", err)
+	}
+	return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}, nil
+}
+
+// Fingerprint returns "sha256:" + hex SHA-256 of pub's PKIX DER encoding.
+// This is the MASTER_KEY value agents are configured with.
+func Fingerprint(pub crypto.PublicKey) (string, error) {
+	der, err := x509.MarshalPKIXPublicKey(pub)
+	if err != nil {
+		return "", fmt.Errorf("marshal public key: %w", err)
+	}
+	sum := sha256.Sum256(der)
+	return "sha256:" + hex.EncodeToString(sum[:]), nil
+}
+
+// ClientConfig returns a TLS config that accepts only a server whose
+// certificate carries the public key with the given fingerprint.
+func ClientConfig(fingerprint string) *tls.Config {
+	return &tls.Config{
+		MinVersion: tls.VersionTLS13,
+		// Chain and hostname verification are replaced by the fingerprint
+		// check below: the master's certificate is self-signed by design.
+		// The handshake still verifies the server's signature with the
+		// certificate's key, so presenting a copied certificate fails.
+		InsecureSkipVerify: true, //nolint:gosec // see comment above
+		VerifyPeerCertificate: func(raw [][]byte, _ [][]*x509.Certificate) error {
+			if len(raw) == 0 {
+				return errors.New("pin: server sent no certificate")
+			}
+			cert, err := x509.ParseCertificate(raw[0])
+			if err != nil {
+				return fmt.Errorf("pin: parse certificate: %w", err)
+			}
+			got, err := Fingerprint(cert.PublicKey)
+			if err != nil {
+				return err
+			}
+			if subtle.ConstantTimeCompare([]byte(got), []byte(fingerprint)) != 1 {
+				return fmt.Errorf("pin: fingerprint mismatch: server presented %s", got)
+			}
+			return nil
+		},
+	}
+}
@@ -0,0 +1,61 @@
+package pin
+
+import (
+	"crypto/tls"
+	"net/http"
+	"net/http/httptest"
+	"regexp"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newPinnedServer(t *testing.T) (*httptest.Server, string) {
+	t.Helper()
+	key, err := GenerateKey()
+	require.NoError(t, err)
+	cert, err := Certificate(key)
+	require.NoError(t, err)
+	fp, err := Fingerprint(key.Public())
+	require.NoError(t, err)
+
+	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte("ok"))
+	}))
+	srv.TLS = &tls.Config{Certificates: []tls.Certificate{cert}}
+	srv.StartTLS()
+	t.Cleanup(srv.Close)
+	return srv, fp
+}
+
+func clientFor(fp string) *http.Client {
+	return &http.Client{Transport: &http.Transport{TLSClientConfig: ClientConfig(fp)}}
+}
+
+func TestFingerprintFormat(t *testing.T) {
+	key, err := GenerateKey()
+	require.NoError(t, err)
+	fp, err := Fingerprint(key.Public())
+	require.NoError(t, err)
+	assert.Regexp(t, regexp.MustCompile(`^sha256:[0-9a-f]{64}$`), fp)
+}
+
+func TestClientConfigAcceptsPinnedKey(t *testing.T) {
+	srv, fp := newPinnedServer(t)
+	resp, err := clientFor(fp).Get(srv.URL)
+	require.NoError(t, err)
+	_ = resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+}
+
+func TestClientConfigRejectsOtherKey(t *testing.T) {
+	srv, _ := newPinnedServer(t)
+	other, err := GenerateKey()
+	require.NoError(t, err)
+	otherFP, err := Fingerprint(other.Public())
+	require.NoError(t, err)
+
+	_, err = clientFor(otherFP).Get(srv.URL)
+	assert.ErrorContains(t, err, "fingerprint mismatch")
+}
@@ -0,0 +1,102 @@
+// Package protocol defines the messages agents and the master exchange over
+// their WebSocket connection. Every message is an Envelope whose Payload is
+// one of the payload types below, selected by Type.
+package protocol
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+// Message types sent from agent to master.
+const (
+	TypeHello    = "hello"
+	TypeSnapshot = "snapshot"
+	TypeUpsert   = "upsert"
+	TypeRemove   = "remove"
+)
+
+// Envelope wraps every message on the wire.
+type Envelope struct {
+	Type    string          `json:"type"`
+	Payload json.RawMessage `json:"payload,omitempty"`
+}
+
+// Hello is the first message an agent sends on every connection.
+type Hello struct {
+	AgentVersion   string `json:"agent_version"`
+	Hostname       string `json:"hostname"`
+	SwarmManager   bool   `json:"swarm_manager"`
+	SwarmClusterID string `json:"swarm_cluster_id,omitempty"`
+}
+
+// Kind says how a service is deployed.
+type Kind string
+
+// Service kinds.
+const (
+	KindCompose   Kind = "compose"
+	KindSwarm     Kind = "swarm"
+	KindContainer Kind = "container"
+)
+
+// Service is one deployable unit as an agent sees it. Which location fields
+// are set depends on Kind.
+type Service struct {
+	Kind           Kind   `json:"kind"`
+	ComposeProject string `json:"compose_project,omitempty"`
+	ComposeService string `json:"compose_service,omitempty"`
+	WorkingDir     string `json:"working_dir,omitempty"`
+	SwarmService   string `json:"swarm_service,omitempty"`
+	ContainerName  string `json:"container_name,omitempty"`
+	Image          string `json:"image"`
+	Digest         string `json:"digest,omitempty"`
+	Status         string `json:"status"`
+	Self           bool   `json:"self,omitempty"`
+}
+
+// Key identifies the service within one agent's report. It is stable across
+// restarts and is what Remove refers to.
+func (s Service) Key() string {
+	switch s.Kind {
+	case KindCompose:
+		return "compose|" + s.ComposeProject + "|" + s.ComposeService
+	case KindSwarm:
+		return "swarm|" + s.SwarmService
+	default:
+		return "container|" + s.ContainerName
+	}
+}
+
+// Snapshot is the agent's complete inventory. Sent after every Hello.
+type Snapshot struct {
+	Services []Service `json:"services"`
+}
+
+// Upsert reports a service that appeared or changed.
+type Upsert struct {
+	Service Service `json:"service"`
+}
+
+// Remove reports that the service with Key is gone.
+type Remove struct {
+	Key string `json:"key"`
+}
+
+// Encode wraps payload in an Envelope of the given type.
+func Encode(typ string, payload any) (Envelope, error) {
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		return Envelope{}, fmt.Errorf("encode %s: %w", typ, err)
+	}
+	return Envelope{Type: typ, Payload: raw}, nil
+}
+
+// Decode unpacks env's payload into T.
+func Decode[T any](env Envelope) (T, error) {
+	var v T
+	if err := json.Unmarshal(env.Payload, &v); err != nil {
+		return v, fmt.Errorf("decode %s: %w", env.Type, err)
+	}
+	return v, nil
+}
@@ -0,0 +1,49 @@
+package protocol
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestServiceKey(t *testing.T) {
+	cases := []struct {
+		name string
+		svc  Service
+		want string
+	}{
+		{"compose", Service{Kind: KindCompose, ComposeProject: "slusk", ComposeService: "web"}, "compose|slusk|web"},
+		{"swarm", Service{Kind: KindSwarm, SwarmService: "gitea_gitea"}, "swarm|gitea_gitea"},
+		{"container", Service{Kind: KindContainer, ContainerName: "adhoc"}, "container|adhoc"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, c.svc.Key())
+		})
+	}
+}
+
+func TestEncodeDecodeRoundTrip(t *testing.T) {
+	in := Snapshot{Services: []Service{
+		{Kind: KindCompose, ComposeProject: "slusk", ComposeService: "web", Image: "slusk:1.4.2", Status: "running"},
+	}}
+	env, err := Encode(TypeSnapshot, in)
+	require.NoError(t, err)
+
+	raw, err := json.Marshal(env)
+	require.NoError(t, err)
+	var back Envelope
+	require.NoError(t, json.Unmarshal(raw, &back))
+
+	assert.Equal(t, TypeSnapshot, back.Type)
+	out, err := Decode[Snapshot](back)
+	require.NoError(t, err)
+	assert.Equal(t, in, out)
+}
+
+func TestDecodeRejectsMalformedPayload(t *testing.T) {
+	_, err := Decode[Hello](Envelope{Type: TypeHello, Payload: json.RawMessage(`{"hostname": 5}`)})
+	assert.ErrorContains(t, err, "decode hello")
+}
@@ -0,0 +1,52 @@
+package updater
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/shcizo/package-updater/internal/discovery"
+)
+
+// SwarmDockerClient is the subset of the Docker SDK SwarmExecutor depends
+// on. Defined as an interface so tests can supply a fake.
+type SwarmDockerClient interface {
+	ServiceInspectWithRaw(ctx context.Context, serviceID string, opts swarm.ServiceInspectOptions) (swarm.Service, []byte, error)
+	ServiceUpdate(ctx context.Context, serviceID string, version swarm.Version, service swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error)
+}
+
+// SwarmExecutor updates a Swarm service's image via the Docker API,
+// equivalent to `docker service update --image`.
+type SwarmExecutor struct {
+	cli SwarmDockerClient
+}
+
+// NewSwarmExecutor returns an executor that drives Swarm service updates
+// through the Docker API.
+func NewSwarmExecutor(cli SwarmDockerClient) *SwarmExecutor {
+	return &SwarmExecutor{cli: cli}
+}
+
+// Execute inspects the service to get its current spec and version (required
+// by the Docker API as an optimistic-concurrency token), sets the new image
+// on the container spec, and calls ServiceUpdate. QueryRegistry is set so a
+// floating tag (e.g. ":latest") resolves to a fresh digest and actually
+// triggers a rolling update instead of being treated as unchanged.
+func (e *SwarmExecutor) Execute(ctx context.Context, job discovery.Job) error {
+	if job.Refused {
+		return fmt.Errorf("refused: %s", job.RefusedReason)
+	}
+
+	svc, _, err := e.cli.ServiceInspectWithRaw(ctx, job.ServiceID, swarm.ServiceInspectOptions{})
+	if err != nil {
+		return fmt.Errorf("inspect service %s: %w", job.ServiceID, err)
+	}
+
+	spec := svc.Spec
+	spec.TaskTemplate.ContainerSpec.Image = job.Image
+
+	if _, err := e.cli.ServiceUpdate(ctx, job.ServiceID, svc.Version, spec, swarm.ServiceUpdateOptions{QueryRegistry: true}); err != nil {
+		return fmt.Errorf("update service %s: %w", job.ServiceID, err)
+	}
+	return nil
+}
@@ -0,0 +1,99 @@
+package updater_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/shcizo/package-updater/internal/updater"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeSwarmDockerClient struct {
+	inspectService swarm.Service
+	inspectErr     error
+	updateErr      error
+
+	inspectCalled         bool
+	updateCalled          bool
+	updateCalledServiceID string
+	updateCalledVersion   swarm.Version
+	updateCalledSpec      swarm.ServiceSpec
+	updateCalledOpts      swarm.ServiceUpdateOptions
+}
+
+func (f *fakeSwarmDockerClient) ServiceInspectWithRaw(_ context.Context, _ string, _ swarm.ServiceInspectOptions) (swarm.Service, []byte, error) {
+	f.inspectCalled = true
+	return f.inspectService, nil, f.inspectErr
+}
+
+func (f *fakeSwarmDockerClient) ServiceUpdate(_ context.Context, serviceID string, version swarm.Version, spec swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error) {
+	f.updateCalled = true
+	f.updateCalledServiceID = serviceID
+	f.updateCalledVersion = version
+	f.updateCalledSpec = spec
+	f.updateCalledOpts = opts
+	return swarm.ServiceUpdateResponse{}, f.updateErr
+}
+
+func mkInspectService(version uint64, image string) swarm.Service {
+	return swarm.Service{
+		ID:   "svc-myapp",
+		Meta: swarm.Meta{Version: swarm.Version{Index: version}},
+		Spec: swarm.ServiceSpec{
+			Annotations:  swarm.Annotations{Name: "myapp_web"},
+			TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: image}},
+		},
+	}
+}
+
+func TestSwarmExecutor_UpdatesImageAndVersion(t *testing.T) {
+	fake := &fakeSwarmDockerClient{inspectService: mkInspectService(7, "registry.example.com/myapp:v1")}
+	e := updater.NewSwarmExecutor(fake)
+
+	job := discovery.Job{Service: "myapp_web", ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"}
+	err := e.Execute(context.Background(), job)
+
+	require.NoError(t, err)
+	require.True(t, fake.updateCalled)
+	require.Equal(t, "svc-myapp", fake.updateCalledServiceID)
+	require.Equal(t, uint64(7), fake.updateCalledVersion.Index)
+	require.Equal(t, "registry.example.com/myapp:v2", fake.updateCalledSpec.TaskTemplate.ContainerSpec.Image)
+	require.True(t, fake.updateCalledOpts.QueryRegistry)
+}
+
+func TestSwarmExecutor_InspectError(t *testing.T) {
+	fake := &fakeSwarmDockerClient{inspectErr: errors.New("service not found")}
+	e := updater.NewSwarmExecutor(fake)
+
+	err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-missing", Image: "x:v2"})
+
+	require.Error(t, err)
+	require.False(t, fake.updateCalled)
+}
+
+func TestSwarmExecutor_UpdateError(t *testing.T) {
+	fake := &fakeSwarmDockerClient{
+		inspectService: mkInspectService(1, "registry.example.com/myapp:v1"),
+		updateErr:      errors.New("update rejected"),
+	}
+	e := updater.NewSwarmExecutor(fake)
+
+	err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"})
+
+	require.Error(t, err)
+}
+
+func TestSwarmExecutor_RefusedJobIsNotExecuted(t *testing.T) {
+	fake := &fakeSwarmDockerClient{}
+	e := updater.NewSwarmExecutor(fake)
+	job := discovery.Job{ServiceID: "svc-myapp", Image: "x:v2", Refused: true, RefusedReason: "not opted in"}
+
+	err := e.Execute(context.Background(), job)
+
+	require.Error(t, err)
+	require.False(t, fake.inspectCalled)
+	require.False(t, fake.updateCalled)
+}
Author	SHA1	Message	Date
shcizo	8a1fb45971	feat(master): store service inventory snapshots and web sessions	2026-10-12 14:45:46 +02:00
shcizo	99c553b7c6	feat(master): postgres store with migrations, master identity and agents	2026-10-12 14:43:34 +02:00
shcizo	9b74286c9b	feat(agent): add updater-agent binary and ship it in the image	2026-10-12 14:41:05 +02:00
shcizo	ad6ff1e066	feat(agent): connect to master over pinned TLS and stream inventory	2026-10-12 14:38:14 +02:00
shcizo	3edaf4e602	feat(agent): diff inventories and debounce docker events	2026-10-12 14:36:11 +02:00
shcizo	abee9abb1b	feat(agent): collect compose, swarm and container inventory from docker	2026-10-12 14:34:13 +02:00
shcizo	61e738e39d	feat: pinned master key TLS and per-agent tokens	2026-10-12 14:32:40 +02:00
shcizo	b482c2e29c	feat(protocol): define agent-to-master wire messages	2026-10-12 14:31:18 +02:00
shcizo	adeddf0254	feat(imageref): parse image references into repo, tag and digest	2026-10-12 14:29:52 +02:00
shcizo	a57df51532	Merge pull request 'docs: master/agent design spec and phase 1 plan' (#16 ) from feat/master-agent into main	2026-10-12 12:15:16 +00:00
shcizo	d2a8b548c9	docs: implementation plan for master/agent phase 1 (inventory)	2026-10-12 13:42:27 +02:00
shcizo	1c833f9fbe	docs: design for master/agent architecture with live inventory, update modes and polling	2026-10-12 13:20:50 +02:00
shcizo	9ea082de82	Merge branch 'worktree-docker-swarm-support' Adds MODE=compose\|swarm. Each instance owns one Docker daemon and runs one paradigm; see docs/superpowers/specs/2026-08-04-multi-server-fanout-design.md for why this exclusivity is the intended design.	2026-08-04 19:25:08 +02:00
shcizo	50ef1a9a07	Merge pull request 'feat: add Docker Swarm support (MODE=swarm)' (#4 ) from worktree-docker-swarm-support into main Docker Build / docker (push) Successful in 1m7s Details Reviewed-on: #4	2026-07-04 18:54:14 +00:00
shcizo	db84a2f6ba	fix(api): plumb request tag into discovery so swarm mode deploys the requested version req.Tag was only echoed in the HTTP response, never used to match jobs. Compose mode didn't care (ComposeExecutor re-pulls the compose file's own pinned tag), but SwarmExecutor sets the service image directly from Job.Image, which was built from the untagged req.Image alone -- so a Swarm deploy silently rewrote the service to :latest instead of the requested tag. Build the full image:tag reference once in the handler and pass it into FindJobs; NormaliseImage/ImagesMatch already strip tags before matching, so this doesn't change which jobs match in either mode.	2026-07-04 20:41:59 +02:00
shcizo	9881812eca	docs: document MODE=swarm support and its opt-in label / manager-node requirements	2026-07-04 20:32:22 +02:00
shcizo	ac3912367f	merge: bring in CLAUDE.md and CI workflow from main	2026-07-04 20:31:25 +02:00
shcizo	9a3a200873	feat: wire MODE config to select compose or swarm discovery/executor	2026-07-04 20:28:28 +02:00
shcizo	f92f8ac6b1	feat(api): surface ServiceID in update results for swarm jobs	2026-07-04 20:22:30 +02:00
shcizo	34865c9fd0	feat(updater): add SwarmExecutor to run docker service update via the Docker API Adds the Swarm-mode counterpart to ComposeExecutor: updates a Swarm service's image directly through the Docker SDK (ServiceInspectWithRaw + ServiceUpdate with QueryRegistry=true so floating tags resolve to a fresh digest), gated by the same Refused guard used in Compose mode. Satisfies the existing Executor interface unchanged, so Queue needs no changes.	2026-07-04 20:19:00 +02:00
shcizo	a8524e5ff4	feat(discovery): add SwarmDiscovery to find opted-in Swarm services by image Verified the Docker SDK v28.5.2+incompatible ServiceList signature via go doc before implementing: the options type is swarm.ServiceListOptions, not types.ServiceListOptions as the brief assumed. Everything else (Service.ID, Service.Spec via embedded Annotations for Name/Labels, TaskTemplate.ContainerSpec.Image) matched the brief exactly. Reuses discovery.ImagesMatch and discovery.HasOptIn rather than duplicating matching/opt-in logic. Opt-in label is read from the service spec's own labels since Swarm services have no local compose file to anchor a STACKS_ROOT path check against (unlike Compose mode).	2026-07-04 20:12:44 +02:00
shcizo	9cf484e164	feat(discovery): add Image and ServiceID fields to Job for swarm support	2026-07-04 20:09:39 +02:00
shcizo	1002bb7b96	feat(config): add MODE env var to select compose or swarm operation	2026-07-04 20:06:50 +02:00
shcizo	8f999c69bd	docs: add docker swarm support implementation plan	2026-07-04 20:05:02 +02:00
shcizo	df54bbaacf	docs: add CLAUDE.md with project context for Claude Code Captures design intent, gotchas, and references that aren't obvious from code alone — single-worker queue rationale, defense-in-depth security model, the not-yet-wired selfupdate package, and pointers to the design spec and implementation plan in docs/superpowers/.	2026-05-22 14:27:28 +02:00
shcizo	476945722b	Merge pull request 'ci: docker build + push workflow on v* tags' (#2 ) from feat/ci-docker-build into main Docker Build / docker (push) Successful in 51s Details	2026-05-22 14:09:02 +02:00