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

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.

Dockerfile

@@ -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"]

cmd/agent/main.go

@@ -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()
+	}
+}

docs/superpowers/specs/2026-10-12-master-agent-design.md

@@ -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.

go.mod

@@ -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

go.sum

@@ -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=

internal/agent/config/config.go

@@ -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
+}

internal/agent/config/config_test.go

@@ -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)
+		})
+	}
+}

internal/agent/conn/conn.go

@@ -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
+}

internal/agent/conn/conn_test.go

@@ -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")
+}

internal/agent/inventory/collector.go

@@ -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"
+	}
+}

internal/agent/inventory/collector_test.go

@@ -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)
+}

internal/agent/inventory/diff.go

@@ -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
+}

internal/agent/inventory/diff_test.go

@@ -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)
+		})
+	}
+}

internal/agent/inventory/watch.go

@@ -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()
+		}
+	}
+}

internal/agent/inventory/watch_test.go

@@ -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")
+}

internal/agenttoken/agenttoken.go

@@ -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[:])
+}

internal/agenttoken/agenttoken_test.go

@@ -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"))
+}

internal/imageref/imageref.go

@@ -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
+}

internal/imageref/imageref_test.go

@@ -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")
+		})
+	}
+}

internal/master/model/model.go

@@ -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
+}

internal/master/model/model_test.go

@@ -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())
+		})
+	}
+}

internal/master/store/agents.go

@@ -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
+}

internal/master/store/agents_integration_test.go

@@ -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)))
+}

internal/master/store/helpers_integration_test.go

@@ -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
+}

internal/master/store/identity.go

@@ -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
+}

internal/master/store/migrate.go

@@ -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
+}

internal/master/store/migrations/0001_init.sql

@@ -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
+);

internal/master/store/services.go

@@ -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()
+}

internal/master/store/services_integration_test.go

@@ -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) }

internal/master/store/sessions.go

@@ -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
+}

internal/master/store/sessions_integration_test.go

@@ -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)
+}

internal/master/store/store.go

@@ -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) }

internal/pin/pin.go

@@ -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
+		},
+	}
+}

internal/pin/pin_test.go

@@ -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")
+}

internal/protocol/protocol.go

@@ -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
+}

internal/protocol/protocol_test.go

@@ -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")
+}