feat: master/agent phase 4 — registry polling and version history (#9) (#23)

## Summary
- `POST /update` on the master (outside OIDC, bearer `UPDATER_API_KEY`; unset → 503). Matches `image` against `image_repo` on all agents, marks `available_*` on every match, starts webhook-triggered jobs for `auto` services targeting `<image_repo>:<tag>` (followed tag when no tag is given), and waits for them up to `UPDATE_TIMEOUT` (default 5m). 404 on no match, 500 if any auto job failed or timed out, otherwise 200.
- Manual/auto toggle per service on the overview and service page (`POST /services/{id}/mode`, CSRF); auto is blocked for self, `build:` and container services with the reason shown. "Update available" badge and an "Available" row.
- `FinishJob` clears `available_*` only when the successful job was created for that exact ref, after it was reported. Migration `0003_webhook.sql`, phase-3 plan doc, README / gitea-action README / CLAUDE.md / deploy example updated. `action.yml` unchanged.

Closes #8

## Behaviour changes for CI
- Mixed results now return 500 (the old 207 is gone); no match returns 404 instead of 200.
- With a `tag`, Compose services get pinned to `<image_repo>:<tag>` via `compose.override.yaml`; the tag must be a pushed tag.
- `busy` (job already active) and `refused` (not updatable) are reported but return 200.
- The reverse proxy read timeout must exceed `UPDATE_TIMEOUT`.

## Test plan
- [x] `go test ./...` and `go test -race ./internal/agent/... ./internal/master/... ./internal/e2e/`
- [x] `go test -tags integration ./internal/master/store/` against throwaway Postgres
- [x] UI verified in browser (mode toggle, disabled reason, badge via SSE, Available row)
- [ ] Manual: webhook with manual service → `marked` + badge
- [ ] Manual: webhook with auto service → waits, `succeeded`, badge cleared, override pinned
- [ ] Manual: wrong token 401, unknown image 404, offline agent 500
- [ ] Gitea action pointed at the master's `/update` in a real CI run

Reviewed-on: #22

CLAUDE.md

@@ -36,6 +36,8 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
 - `internal/master/config`   — env-var loading
 - `internal/master/store`    — Postgres access and migrations
 - `internal/master/hub`      — agent connections: auth, per-connection writer, routing, job reconcile on hello
+- `internal/master/registry` — HEAD-only digest lookup against a registry (go-containerregistry, `DOCKER_CONFIG` credentials)
+- `internal/master/poller`   — every `POLL_INTERVAL`: one lookup per followed `repo:tag`, sets the available version, starts auto jobs
 - `internal/master/jobs`     — decides whether a service may be updated and starts a job on the right agent
 - `internal/master/webhook`  — `POST /update`: records the available version, starts jobs for auto services, waits for them
 - `internal/master/web`, `webauth` — HTMX UI behind OIDC login, SSE signals
@@ -68,6 +70,26 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
 
 ## Gotchas
 
+- **Registry access is HEAD only** (`internal/master/registry`). Docker Hub counts a manifest
+  GET as a pull; the client never falls back to GET, and a test asserts this.
+- **`image_versions` uses `''` instead of NULL** for a missing tag or digest, so
+  `UNIQUE (repo, tag, digest)` deduplicates. The webhook records `(repo, tag, '')`; that
+  row is hidden from `RecentVersions` when a row with the same tag and a digest was first
+  seen at or after it.
+- **The UI picks a version by `image_versions.id`, never by a ref from the form**, and the row
+  must belong to the service's `image_repo`. A digest-pinned or tagless pick needs agent
+  protocol 3 (`protocol.MinPinnedRefVersion`); `Hub.Route` takes the minimum version.
+- **Picking a tagless version (`repo@digest`) leaves the service following no tag:** it is no
+  longer polled and the plain Update button is refused.
+- **`services.last_tag_digest`** is set by the poller when it starts an auto job and when it
+  sees the service already running the tag's digest, and by `FinishJob` after any successful
+  plain tag job (not pinned/tagless picks; the digest is the one the agent pinned). The poller
+  starts no job for that digest, nor for one `desired_ref` (`ref@digest`) already names, and
+  re-reads the service before `Start`; so a rollback of an auto service holds until the
+  registry shows a different digest. `ErrJobActive` and `NotUpdatableError` leave it unset so
+  the next cycle retries.
+- **A bare image's running version is recorded with tag `latest`** (like `jobs.FollowedRef`);
+  `repo@digest` is recorded with tag `''`. Only `sha256:<64 hex>` digests are recorded.
 - **Image matching is case-sensitive.** The tricky case is `localhost:5000/foo:v1` where
   the first colon is a port, not a tag. See `internal/imageref`.
 - **Path safety uses `filepath.Rel` + ".." prefix check**, NOT `strings.HasPrefix`.
@@ -102,9 +124,10 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
   repo, never from the request's image string. With an explicit target the "follows no tag"
   refusal does not apply; removed, self, built and container still do.
 - **`available_*` is cleared by `FinishJob`** on success only when the job's `to_ref`
-  (as created, without `@digest`) equals `available_ref` and `available_at <= job.created_at`
-  (that version is now running). A manual Update of the followed tag does not clear a
-  badge for a different tag.
+  (as created, without `@digest`) equals `available_ref`, `available_at <= job.created_at`
+  (that version is now running) and, when the job's `to_ref` carries a digest, that digest
+  equals `available_digest`. A rollback to an older digest of the tag therefore keeps the
+  newer suggestion, and a digest-pinned pick does not clear a digest-less (webhook) one.
 - **Job output is capped at `protocol.MaxJobOutput` (128 KiB, tail).** NUL bytes are
   stripped by the hub because Postgres text rejects them.
 
@@ -114,6 +137,7 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
 - Phase 1 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-1-inventory.md`
 - Phase 2 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-2-jobs.md`
 - Phase 3 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-3-webhook.md`
+- Phase 4 plan: `docs/superpowers/plans/2026-10-13-master-agent-phase-4-polling.md`
 - Original single-instance design (historical):
   `docs/superpowers/specs/2026-05-22-package-updater-design.md`
 - Consumer-side CI integration: `gitea-action/` (calls the master's `POST /update`)

README.md

@@ -9,6 +9,7 @@ the same image: `updater-master` and `updater-agent` (the default entrypoint).
 Design: [master/agent spec](docs/superpowers/specs/2026-10-12-master-agent-design.md).
 Phase 2 plan: [jobs and manual update](docs/superpowers/plans/2026-10-12-master-agent-phase-2-jobs.md).
 Phase 3 plan: [webhook and auto mode](docs/superpowers/plans/2026-10-12-master-agent-phase-3-webhook.md).
+Phase 4 plan: [registry polling and version history](docs/superpowers/plans/2026-10-13-master-agent-phase-4-polling.md).
 
 ## What an update does
 
@@ -33,6 +34,8 @@ pinned TLS, reached directly on the internal network).
 |---|---|---|---|
 | `UPDATER_API_KEY` | no | — | Bearer token for `POST /update`. Without it the webhook answers `503` and the master logs `webhook_disabled`. |
 | `UPDATE_TIMEOUT` | no | `5m` | How long one `POST /update` waits for its auto jobs (Go duration). |
+| `POLL_INTERVAL` | no | `6h` | Time between registry polling cycles (Go duration, must be positive). The first cycle runs one minute after start. |
+| `DOCKER_CONFIG` | no | `~/.docker` | Directory with a `config.json` holding registry credentials for polling. Only inline `auths` entries work; see [Registry polling](#registry-polling). |
 
 The master's other settings (`DATABASE_URL`, `PUBLIC_URL`, `AGENT_URL`, `OIDC_*`,
 `ALLOWED_GROUP`, `AGENT_IMAGE`, `PORT`, `AGENT_PORT`, `LOG_LEVEL`) are shown in
@@ -112,6 +115,77 @@ The request is open for as long as the jobs run, so the reverse proxy in front o
 master must allow a read timeout longer than `UPDATE_TIMEOUT`. If the client disconnects,
 marking and starting jobs still complete; only the waiting stops.
 
+## Registry polling
+
+Besides the webhook, the master asks each registry whether the tag a service follows has
+a new digest. Every `POLL_INTERVAL` it sends one `HEAD /v2/<repo>/manifests/<tag>` per
+distinct followed `repo:tag` (a bare image counts as `:latest`; `repo:tag@digest` is polled
+as `repo:tag`). It never downloads a manifest: Docker Hub counts a `GET` as a pull against
+the rate limit and a `HEAD` as nothing
+([source](https://docs.docker.com/docker-hub/usage/pulls/)), and Docker Hub also has a
+separate abuse limit for many pulls from one IP address, so the poller stays HEAD-only. For a
+multi-arch image the digest is the index digest, the same one `docker inspect` shows in
+`RepoDigests`.
+
+- **What counts as new.** A digest different from the one the service runs becomes the
+  service's *available* version (the overview badge, source `poll`). A digest the master
+  already suggested is not suggested again. A service in **auto** mode also gets an update
+  job to `repo:tag` (the agent resolves the digest itself, so agents on protocol version 2
+  keep working). The master remembers the newest digest it started or deployed for the tag
+  in `services.last_tag_digest` (set when polling starts a job, when polling sees the
+  service already running that digest, and when any plain tag job succeeds, whoever started
+  it) and starts no job for that digest again, whether the job
+  failed or the user rolled back afterwards. If the service already had a job when the
+  digest was found, the next cycle tries again.
+- **What is not polled.** Services that follow no tag (`repo@digest`), `build:` services, and
+  services without a registry digest (locally built images). The service page says which.
+  The updater's own services and plain containers are polled, so their badge works, but
+  jobs for them are still refused.
+- **Errors.** A failed lookup is stored for the services that follow that tag and shown as a
+  "poll failed" badge on the overview and in the "Polling" row of the service page; other
+  tags continue. A later success clears it.
+- **Credentials.** The master reads `$DOCKER_CONFIG/config.json` (a file written by
+  `docker login`), mounted read-only into the container; without it registries are accessed
+  anonymously. Only inline `auths` entries are supported. A `credsStore` or `credHelpers`
+  entry makes every lookup fail with `error getting credentials ... executable file not found
+  in $PATH`, because the container has no credential helper binaries. The master logs
+  `registry_credentials` at startup with the directory and whether `config.json` exists,
+  never its contents.
+- **Requests go wherever the image name points.** The master sends the lookup to the
+  registry host in the image name, including a `GET /v2/` ping that follows redirects.
+  Anyone who can start containers on a managed host can therefore make the master send
+  blind requests to internal hosts. The registry's answer is not shown: an error keeps only
+  the HTTP status and the registry's error codes; any other failure shows only "timeout",
+  "cannot reach registry" or "lookup failed". The full error goes to the master's log
+  (`registry_lookup_failed`).
+
+### Version history and rollback
+
+The master records every version it sees (webhook, polling, and what agents report as
+running) in `image_versions`. The service page lists the 10 most recently first seen
+versions of the service's image repo, marks the one that is running, and has a *Deploy*
+button on each row. A webhook announces a tag but not its digest, so such a row has no
+digest until polling or an agent reports one; it is hidden when a row with the same tag
+and a digest was first seen at or after it. A bare image (`postgres`) is recorded as tag
+`latest`.
+
+Picking a version starts a manual job to exactly that reference. Picking an older one is a
+rollback, including to a digest-pinned reference. A version without a tag deploys
+`repo@digest` and the service then follows no tag, so it is no longer polled or updatable
+with the plain *Update* button until it is moved to a tagged version again. Picks of
+digest-pinned or tagless versions need an agent speaking protocol version 3 (this
+release); on an older agent the job fails with "agent too old to deploy a pinned version
+(needs protocol 3)".
+
+A rollback made before the first poll cycle after the upgrade has seen the service is not
+protected, because the master has not yet recorded the digest it was running.
+
+A service in **auto** mode stays auto after a rollback. Polling does not undo it: the
+service stays on the older version until the registry shows a digest of the followed tag
+that differs from the newest one the master deployed. The newer version it was rolled back
+from stays marked as available. A webhook that names a tag still starts a job for an auto
+service every time, so a CI push moves it forward again.
+
 ## Adding agents
 
 Open the master's **Agents** page and add one. It shows a ready-to-run compose file with

cmd/master/main.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"strings"
 	"syscall"
 	"time"
@@ -20,6 +21,8 @@ import (
 	"github.com/shcizo/package-updater/internal/master/events"
 	"github.com/shcizo/package-updater/internal/master/hub"
 	"github.com/shcizo/package-updater/internal/master/jobs"
+	"github.com/shcizo/package-updater/internal/master/poller"
+	"github.com/shcizo/package-updater/internal/master/registry"
 	"github.com/shcizo/package-updater/internal/master/store"
 	"github.com/shcizo/package-updater/internal/master/web"
 	"github.com/shcizo/package-updater/internal/master/webauth"
@@ -37,6 +40,29 @@ func main() {
 	}
 }
 
+// pollFirstDelay is how long the master runs before its first registry
+// polling cycle, so agents have reported their inventory.
+const pollFirstDelay = time.Minute
+
+// pollWorkers is how many registry lookups run at once.
+const pollWorkers = 4
+
+// registryCredentialsDir returns the directory the registry client reads
+// credentials from (DOCKER_CONFIG, else ~/.docker) and whether a config.json
+// is in it. Only the path is logged, never the file.
+func registryCredentialsDir() (dir string, found bool) {
+	dir = os.Getenv("DOCKER_CONFIG")
+	if dir == "" {
+		home, err := os.UserHomeDir()
+		if err != nil {
+			return "", false
+		}
+		dir = filepath.Join(home, ".docker")
+	}
+	_, err := os.Stat(filepath.Join(dir, "config.json"))
+	return dir, err == nil
+}
+
 func run() error {
 	cfg, err := config.Load()
 	if err != nil {
@@ -81,6 +107,7 @@ func run() error {
 
 	ui, err := web.New(st, agentHub, jobService, broker, web.Config{
 		AgentURL: cfg.AgentURL, MasterKey: fingerprint, AgentImage: cfg.AgentImage,
+		PollInterval: cfg.PollInterval,
 	}, logger)
 	if err != nil {
 		return err
@@ -127,22 +154,39 @@ func run() error {
 		BaseContext: baseContext,
 	}
 
+	credDir, credFound := registryCredentialsDir()
+	logger.Info("registry_credentials", "docker_config", credDir, "config_json_found", credFound)
+	pollCtx, stopPolling := context.WithCancel(ctx)
+	defer stopPolling()
+	registryPoller := poller.New(st, jobService, registry.New("package-updater-master/"+version, logger), broker, poller.Config{
+		Interval: cfg.PollInterval, FirstDelay: pollFirstDelay, Workers: pollWorkers,
+	}, logger)
+	pollDone := make(chan struct{})
+	go func() {
+		defer close(pollDone)
+		registryPoller.Run(pollCtx)
+	}()
+
 	errCh := make(chan error, 2)
 	go func() { errCh <- webSrv.ListenAndServe() }()
 	go func() { errCh <- agentSrv.ListenAndServeTLS("", "") }()
 	logger.Info("listening", "web", webSrv.Addr, "agents", agentSrv.Addr)
 
+	var serveErr error
 	select {
 	case <-ctx.Done():
 		logger.Info("shutdown_signal")
 	case err := <-errCh:
 		if !errors.Is(err, http.ErrServerClosed) {
-			return err
+			serveErr = err
 		}
 	}
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	_ = webSrv.Shutdown(shutdownCtx)
 	_ = agentSrv.Shutdown(shutdownCtx)
-	return nil
+	// The poller writes to the store, which is closed when run returns.
+	stopPolling()
+	<-pollDone
+	return serveErr
 }

deploy/master.stack.yml

@@ -10,6 +10,11 @@
 #      To run without the webhook, remove the pu_updater_api_key lines (the UPDATER_API_KEY
 #      part of the entrypoint, the service secret and the secrets entry); the endpoint then answers 503.
 #      The reverse proxy must allow a read timeout longer than UPDATE_TIMEOUT (default 5m).
+#   5. Optional, for registry polling of private images: put a config.json written by
+#      `docker login` (inline "auths" only; no credsStore) in /path/to/docker-config and
+#      uncomment DOCKER_CONFIG and the volumes entry below. The directory must exist on every
+#      manager node (the master is placed on a manager), or Swarm rejects the task. Without
+#      it registries are accessed anonymously.
 services:
   master:
     image: gitea.shcizo.se/shcizo/package-updater:latest
@@ -21,10 +26,14 @@ services:
       OIDC_CLIENT_ID: package-updater
       ALLOWED_GROUP: updater-admins
       # UPDATE_TIMEOUT: 5m
+      # POLL_INTERVAL: 6h
+      # DOCKER_CONFIG: /docker-config   # step 5
     secrets:
       - pu_database_url
       - pu_oidc_client_secret
       - pu_updater_api_key
+    # volumes:                        # step 5
+    #   - /path/to/docker-config:/docker-config:ro
     ports:
       # Agents connect here directly over the internal network, not via the proxy.
       - target: 8090

docs/superpowers/plans/2026-10-13-master-agent-phase-4-polling.md

@@ -0,0 +1,311 @@
+# Master/Agent Phase 4 — Registry Polling and Version History Implementation Plan
+
+> This plan documents what to build for issue #9 and where it deviates from the spec. Code in "Frozen contracts" is final; tasks give the contract and the test cases.
+
+**Goal:** The master records every version it sees (webhook, registry polling, running inventory) in `image_versions`. Every `POLL_INTERVAL` (default 6h) it checks each distinct followed `repo:tag` against its registry with one HEAD request. A new digest becomes the service's available suggestion, and `auto` services get a job. The Service page lists the 10 most recent versions of the service's repo. Picking one starts a manual job to exactly that reference; picking an older one is a rollback, including to a digest-pinned or tagless reference.
+
+**Architecture:**
+- New package `internal/master/registry`: thin wrapper around `go-containerregistry`'s `remote.Head` with `authn.DefaultKeychain` (reads `DOCKER_CONFIG`/`config.json`).
+- New package `internal/master/poller`: owns the cycle (list services, group by followed ref, run HEADs with bounded concurrency, then set available / start auto jobs / store poll error per service). `cmd/master` runs it as a background goroutine on the signal context.
+- Store: `image_versions` written in `SetAvailable` (webhook and poll, same transaction) and in `upsertService` (running inventory, inside the existing snapshot/upsert transactions).
+- Jobs gains `TriggerVersion`, which takes a version ID, never a ref from the form. `Hub.Route` gains a minimum protocol version so a pinned or tagless ref only goes to agents that understand it.
+- Agent: resolver passes digest-pinned refs through unchanged. Protocol version goes to 3.
+
+**Tech Stack:** adds `github.com/google/go-containerregistry` (`pkg/name`, `pkg/v1/remote`, `pkg/authn`; `pkg/registry` and `pkg/v1/random` in tests only).
+
+**Spec:** `docs/superpowers/specs/2026-10-12-master-agent-design.md`, sections "Versions: what the system has seen", "Data model", "Update", "Failure handling", "Registry polling", "UI". Phase 3 plan conventions still apply.
+
+## Global Constraints
+
+- `go test ./...` passes with no Docker, no Postgres and no network. Registry tests use `httptest` plus `ggcr/pkg/registry`. Postgres tests carry `//go:build integration`.
+- Interfaces declared at the consumer (`poller.Store`, `poller.Jobs`, `poller.Registry`). Hand-written fakes and testify, table-driven where there is more than one case.
+- GoDoc on every exported identifier. Conventional Commits, each ending with ` (#9)`.
+- Registry credentials never appear in logs or in `last_poll_error`. Error text stored truncated to 500 bytes.
+- Registry traffic is HEAD only: never `GET /v2/<repo>/manifests/...` (Docker Hub counts GETs as pulls). A test asserts this.
+- A pick from the UI is a POST with the CSRF token and a version ID. The ref is built server-side from the `image_versions` row, and the row must belong to the service's `image_repo`.
+- The poller never overlaps itself. Next cycle is scheduled after the previous one finishes.
+- `go.mod` `go` directive must stay `1.26.3` after `go get`.
+
+## Deliberate deviations from the spec in this phase
+
+1. **Webhook records `(repo, tag, '')`, no digest.** The master does not resolve the digest at webhook time (no registry round-trip/credentials in the CI path). The digest arrives later from polling or running inventory. The picker hides a digest-less row when a row with the same `(repo, tag)` and a digest was first seen at or after it.
+2. **Empty string instead of NULL** for missing tag or digest in `image_versions`, so `UNIQUE (repo, tag, digest)` works. A CHECK forbids both being empty.
+3. **`first_seen_at` is never bumped.** Re-observation is `ON CONFLICT DO NOTHING`; `source` is the first source that saw it. "10 most recent" = "10 most recently first seen". A new digest of a tag is a new row at the top; old digest stays below (enables rollback of moving tags).
+4. **Not polled:** services that follow no tag, `build:` services, services with empty `actual_digest`. UI shows the reason. Self and container services ARE polled (badge informative; jobs still refused by `jobs`).
+5. **Poller does not re-suggest the same version.** If `available_ref == ref && available_digest == D`, the badge is not written again (no `available_at` bump). The auto job is decided separately: it starts only if the service is auto, `actual_digest != D`, `D != last_tag_digest` and `desired_ref != ref@D`, on a fresh read of the row. `last_tag_digest` is set when a job row was created (not on `ErrJobActive`/`NotUpdatableError`, which retry next cycle). A failed auto job is not retried every interval; a rollback of an auto service is not undone by polling until a new digest appears. The poller also sets `last_tag_digest` to D when `actual_digest == D` (seeding services that predate the column). `FinishJob` also sets `last_tag_digest` to the pinned digest after any successful job whose `to_ref` as created has no digest (manual, webhook or poll); pinned and tagless picks leave it, so a rollback holds whoever started the update.
+6. **FinishJob clearing rule gets a digest condition** (C3). Rollback to an older digest of the same tag no longer clears the newer-version badge.
+7. **Poll auto jobs use `repo:tag`, not `repo:tag@D`.** v2 agents keep working for poll and webhook jobs. Only UI picks of digest-pinned or tagless versions need a v3 agent.
+8. **First cycle runs 1 minute after start**, then every `POLL_INTERVAL` measured from the end of the previous cycle.
+9. **No `last_polled_at` column** (YAGNI).
+10. **Overview gets two new badges: "follows no tag" and "poll failed"** (error as `title`). Service page gets a "Polling" row. "Drifted" is not in this phase.
+11. **Credentials from `DOCKER_CONFIG`** (a directory), same name and meaning as on the agent. Only inline `auths` entries supported in the container.
+
+## Implementation notes (differences found while building)
+
+- The plan's test names `TestResolve` and `TestComposeWritesPinnedRef` became `TestResolver`
+  (extended), a subtest of `TestComposeExecutor` and `TestSwarmUsesPinnedRef`.
+- Tasks 2 and 3 are one commit: changing `Hub.Route`'s signature breaks `jobs` and the e2e hub
+  wrapper until both are updated.
+- `Poller.Cycle` runs the registry lookups concurrently and then records the results
+  sequentially (on `context.WithoutCancel`).
+- `web.Config` gained `PollInterval` for the "every <interval>" text.
+- `authn.DefaultKeychain` with a `credsStore` whose helper binary is missing makes the lookup fail
+  with `error getting credentials - err: exec: "docker-credential-...": executable file not found in $PATH`
+  (covered by `TestCredsStoreWithoutHelperFailsTheLookup`); it does not fall back to anonymous access.
+
+## Review Focus
+
+| Failure case | Test that covers it |
+|---|---|
+| Registry access is HEAD only; no manifest GET ever happens. | registry `TestDigestUsesHeadOnly` |
+| Multi-arch: digest returned is the index digest (same as `RepoDigests` / Swarm spec). | registry `TestDigestOfIndexIsIndexDigest` |
+| Credentials from `$DOCKER_CONFIG/config.json` are sent; anonymous works without the file. | registry `TestCredentialsFromDockerConfig`, `TestAnonymousWithoutConfig` |
+| One HEAD per distinct followed ref (`repo:tag`, `repo:tag@d`, bare → `:latest`). | poller `TestOneHeadPerDistinctRef` |
+| No tag / built / no digest are skipped, no HEAD. | poller `TestSkipped` |
+| Error on one ref goes into its services' `last_poll_error`; others continue; later success clears it. | poller `TestErrorIsPerRef`, `TestErrorClearedOnSuccess` |
+| Already-suggested digest neither re-suggested nor re-started. | poller `TestSameSuggestionIsIdempotent` |
+| Auto gets exactly one `TriggerPoll` job to `repo:tag`; manual only gets marked. | poller `TestNewDigest` table |
+| `ErrJobActive` / `NotUpdatableError` from `Start` do not abort the cycle. | poller `TestStartErrorsDoNotAbortCycle` |
+| Concurrency within limit; cancellation stops the cycle; marking/starting on `WithoutCancel`. | poller `TestConcurrencyLimit`, `TestRunStopsOnCancel` |
+| Form cannot deploy an arbitrary ref: unknown version ID → 404; version of another repo → 404, no job. | jobs `TestTriggerVersion_*`, web `TestUpdateService_Version*` |
+| Pinned/tagless refs routed only to protocol ≥ 3 agents; old agents → failed "agent too old to deploy a pinned version (needs protocol 3)" job (`ErrAgentTooOldForPinned`, which also satisfies `errors.Is(err, ErrAgentTooOld)`). | jobs `TestTriggerVersion_RequiresPinnedProtocol`, hub `TestRoute` |
+| Agent deploys a pinned ref verbatim, no registry resolve. | updater `TestResolve`, `TestComposeWritesPinnedRef`, `TestSwarmUsesPinnedRef` |
+| Rollback to older digest of same tag keeps newer suggestion; pick of suggested digest clears it. | store `TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed` (extended) |
+| Running versions recorded only with a digest; re-snapshot does not duplicate or reorder. | store `TestSnapshotRecordsRunningVersions` |
+| Digest-less webhook row hidden once the tag's digest is known; order, limit, repo filter. | store `TestRecentVersions` |
+| `available_source = 'poll'` passes the CHECK. | store `TestSetAvailable_RecordsVersion` |
+| E2E: pick of a pinned version reaches the agent as that exact ref. | e2e `TestPickPinnedVersion` |
+
+## Frozen contracts
+
+### C1. Migration `internal/master/store/migrations/0004_polling.sql`
+
+```sql
+ALTER TABLE services DROP CONSTRAINT services_available_source_check;
+ALTER TABLE services ADD CONSTRAINT services_available_source_check
+    CHECK (available_source IN ('', 'webhook', 'poll'));
+ALTER TABLE services ADD COLUMN available_digest text NOT NULL DEFAULT '';
+ALTER TABLE services ADD COLUMN last_poll_error  text NOT NULL DEFAULT '';
+ALTER TABLE services ADD COLUMN last_tag_digest text NOT NULL DEFAULT '';
+
+-- Every version the master has seen. '' means "no tag" / "digest unknown";
+-- never NULL, so the unique constraint deduplicates.
+CREATE TABLE image_versions (
+    id            bigserial   PRIMARY KEY,
+    repo          text        NOT NULL,
+    tag           text        NOT NULL DEFAULT '',
+    digest        text        NOT NULL DEFAULT '',
+    source        text        NOT NULL CHECK (source IN ('webhook', 'poll', 'running')),
+    first_seen_at timestamptz NOT NULL,
+    CONSTRAINT image_versions_tag_or_digest CHECK (tag <> '' OR digest <> ''),
+    CONSTRAINT image_versions_unique UNIQUE (repo, tag, digest)
+);
+CREATE INDEX image_versions_recent ON image_versions (repo, first_seen_at DESC, id DESC);
+```
+
+Confirm the existing constraint name in `0003_webhook.sql` before relying on it.
+
+### C2. `internal/master/model`
+
+```go
+// Service gains:
+AvailableDigest string // digest of AvailableRef when known (poll); "" from the webhook
+LastPollError   string // last registry error for this service's followed ref; "" when fine
+
+const SourcePoll = "poll"       // alongside SourceWebhook
+const SourceRunning = "running" // image_versions only
+
+type Available struct { Ref, Digest, Source string; At time.Time } // Digest added
+
+// ImageVersion is one row of image_versions.
+type ImageVersion struct {
+    ID          int64
+    Repo, Tag, Digest, Source string
+    FirstSeenAt time.Time
+}
+// Ref returns "repo:tag@digest", "repo:tag" or "repo@digest".
+func (v ImageVersion) Ref() string
+```
+
+### C3. Store
+
+```go
+// SetAvailable sets available_ref/_digest/_source/_at and, in the same
+// transaction, inserts image_versions(repo, tag parsed from a.Ref, a.Digest,
+// a.Source, a.At) ON CONFLICT DO NOTHING.
+SetAvailable(ctx, id int64, a model.Available) error
+
+// RecentVersions returns up to limit versions of repo, newest first_seen_at
+// first (ties: id DESC). It leaves out a digest-less row when a row with the
+// same repo and tag and a digest was first seen at or after it.
+RecentVersions(ctx, repo string, limit int) ([]model.ImageVersion, error)
+
+// GetImageVersion returns one version or model.ErrNotFound.
+GetImageVersion(ctx, id int64) (model.ImageVersion, error)
+
+// SetPollError sets last_poll_error on the given services ("" clears).
+SetPollError(ctx, ids []int64, msg string) error
+```
+
+`RecentVersions` SQL:
+
+```sql
+SELECT id, repo, tag, digest, source, first_seen_at FROM image_versions v
+WHERE v.repo = $1
+  AND NOT (v.digest = '' AND EXISTS (
+        SELECT 1 FROM image_versions d
+        WHERE d.repo = v.repo AND d.tag = v.tag AND d.digest <> ''
+          AND d.first_seen_at >= v.first_seen_at))
+ORDER BY v.first_seen_at DESC, v.id DESC
+LIMIT $2
+```
+
+`upsertService` (running inventory): after the service upsert, on the same `execer`, only when `svc.Digest` matches `^sha256:[0-9a-f]{64}$`:
+
+```sql
+INSERT INTO image_versions (repo, tag, digest, source, first_seen_at)
+VALUES ($1, $2, $3, 'running', $4) ON CONFLICT DO NOTHING
+```
+
+`repo`/`tag` from `imageref.Parse(svc.Image)`, `digest` = `svc.Digest`. A bare image (no tag, no digest in the name) is recorded with `tag = 'latest'`, like `jobs.FollowedRef`, so it dedupes with the poll row; `repo@digest` is recorded with `tag = ''`. Covers `ApplySnapshot` and `UpsertService` with no hub change.
+
+`serviceColumns` / `scanService` gain `s.available_digest, s.last_poll_error, s.last_tag_digest`. `SetLastTagDigest(ctx, id, digest)` sets the last column; inventory writes never touch it.
+
+`FinishJob` clearing rule (replaces phase 3's). With `requestedRef, requestedDigest, _ := strings.Cut(to_ref_as_created, "@")`:
+
+```sql
+available_ref = $4 AND available_at <= $3
+  AND ($5 = '' OR available_digest = $5)
+```
+
+`$5` = `requestedDigest`. A digest-pinned job does NOT clear a digest-less (webhook) suggestion: a rollback to `main@old` must not hide that a newer `main` was announced. The badge then stays until a plain tag job succeeds; that is accepted. Clearing sets `available_ref=''`, `available_digest=''`, `available_source=''`, `available_at=NULL`.
+
+### C4. `internal/master/registry`
+
+```go
+// Client looks up tag digests with HEAD requests only.
+type Client struct{ /* keychain authn.Keychain, userAgent string */ }
+// New uses authn.DefaultKeychain ($DOCKER_CONFIG/config.json, else ~/.docker/config.json).
+func New(userAgent string) *Client
+// Digest returns the digest ref ("repo:tag", any digest suffix ignored) points
+// at: the index digest for multi-arch images. HEAD /v2/<repo>/manifests/<tag> only.
+func (c *Client) Digest(ctx context.Context, ref string) (string, error)
+```
+
+Implementation: `name.ParseReference(repo+":"+tag)`; `remote.Head(r, remote.WithContext(ctx), remote.WithAuthFromKeychain(c.keychain), remote.WithUserAgent(c.userAgent))`; return `desc.Digest.String()`. Never fall back to GET if `Docker-Content-Digest` is missing.
+
+### C5. `internal/master/poller`
+
+```go
+type Store interface {
+    ListServices(ctx context.Context) ([]model.Service, error)
+    GetService(ctx context.Context, id int64) (model.Service, error)
+    SetAvailable(ctx context.Context, id int64, a model.Available) error
+    SetLastTagDigest(ctx context.Context, id int64, digest string) error
+    SetPollError(ctx context.Context, ids []int64, msg string) error
+}
+type Jobs interface {
+    Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error)
+}
+type Registry interface { Digest(ctx context.Context, ref string) (string, error) }
+type Publisher interface { Publish(t events.Topic) }
+
+type Config struct {
+    Interval   time.Duration // POLL_INTERVAL
+    FirstDelay time.Duration // 1m in production; small in tests
+    Workers    int           // 4
+}
+func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Logger) *Poller
+// Run polls until ctx is done. Blocking.
+func (p *Poller) Run(ctx context.Context)
+// Cycle runs one poll over all services (exported for tests and e2e).
+func (p *Poller) Cycle(ctx context.Context)
+// SkipReason says why svc is not polled: "follows no tag", "built from
+// source", "no registry digest", or "" when it is polled.
+func SkipReason(svc model.Service) string
+```
+
+Cycle semantics, per distinct `ref = jobs.FollowedRef(svc.ActualRef)` over non-removed, non-skipped services:
+1. `D, err := reg.Digest(ctx, ref)` with a 30s timeout per call.
+2. On error: `SetPollError(ids whose LastPollError != msg, msg)`, log `poll_failed` (ref, error).
+3. On success: `SetPollError(ids with LastPollError != "", "")`. Then per service:
+   - Suggestion: `svc.ActualDigest == D` → nothing; `svc.AvailableRef == ref && svc.AvailableDigest == D` → no write; otherwise `SetAvailable({ref, D, poll, now})`.
+   - Auto job, independent of the suggestion: if `UpdateMode == auto && ActualDigest != D && D != LastTagDigest && DesiredRef != ref+"@"+D`, re-read the service (`GetService`), re-check the same condition and removed/mode on the fresh row, then `Start(TriggerPoll, Ref: ref)`. If a job row was created, `SetLastTagDigest(D)`. `ErrJobActive` and `NotUpdatableError` are logged at debug and ignored, leaving `last_tag_digest` unset so the next cycle retries.
+
+Steps 2 and 3 use `context.WithoutCancel(ctx)`. At the end, publish `events.Services` if anything was written; log `poll_cycle` with refs, ok, failed, suggested, jobs, skipped, duration_ms.
+
+### C6. `internal/master/jobs` and hub
+
+```go
+// Hub.Route gains the minimum protocol version for this job:
+Route(svc model.Service, minProtocol int) (model.Route, error)
+
+// Store gains:
+GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error)
+
+// TriggerVersion starts a manual job to exactly version versionID of the
+// service's repo. model.ErrNotFound for an unknown service, an unknown
+// version, or a version of another repo. Otherwise what Start returns.
+func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error)
+```
+
+- Ref = `svc.ImageRepo` + `":"+tag` if tag set + `"@"+digest` if digest set.
+- Calls `Start` with `TriggerManual` and explicit ref ("follows no tag" refusal does not apply; removed, self, built, container still do).
+- `Start` passes `protocol.MinPinnedRefVersion` to `Route` when the ref has a digest or no tag, otherwise `protocol.MinJobVersion`.
+
+### C7. Protocol and agent
+
+```go
+const Version = 3
+const MinJobVersion = 2       // unchanged: repo:tag jobs
+const MinPinnedRefVersion = 3 // jobs whose Ref has a digest or no tag
+```
+
+`JobStart` doc: "Ref is `repo:tag`, `repo:tag@digest` or `repo@digest`. For `repo:tag` the agent resolves the tag's current digest; a ref with a digest is deployed as is."
+
+`Resolver.Resolve`: digest present → return ref unchanged, no `DistributionInspect`; tag present → resolve as today; neither → error "has no tag to resolve" (unchanged). Compose writes the ref verbatim into the override, Swarm sets it as the image. Executors print `pinned <ref>` instead of `resolved <ref>` when nothing was resolved.
+
+### C8. Master configuration
+
+- `POLL_INTERVAL`: default `6h`, positive Go duration, error `"POLL_INTERVAL must be a positive duration such as 6h, got %q"`.
+- `DOCKER_CONFIG`: not parsed by `config`; `go-containerregistry` reads it. `main` logs `registry_credentials` with the path and whether `config.json` exists, never its contents.
+
+### C9. Web
+
+- `POST /services/{id}/update` gains optional form field `version` (an `image_versions.id`): empty → `Trigger` as today; non-numeric → `400 invalid version`; otherwise `TriggerVersion`. Results: `303` to the job (also on `ErrJobActive`), `404` on `ErrNotFound`, `409` re-render "Cannot update: <reason>."
+- `web.Store` gains `RecentVersions`; `web.Jobs` gains `TriggerVersion`.
+- Service page: "Versions" table, up to 10 rows: ref (shortRef, full ref as title), source, first seen, "running" marker when digest equals `ActualDigest`, a POST form with `csrf_token` and `version`. Forms shown only when `AutoBlocked == ""`. Rows with empty tag note "the service will follow no tag afterwards". When the service is `auto`, header notes auto mode will move it forward at the next new version. A "Polling" row shows `SkipReason`, or `last_poll_error`, or "every <interval>".
+- Overview: badge "follows no tag" when `FollowedRef(ActualRef) == ""`; badge "poll failed" (title = error) when `LastPollError != ""`. Template func `followsNoTag`.
+
+## Tasks
+
+### Task 1: Store and model (C1, C2, C3)
+Migration, model fields, store methods, `upsertService` running versions, `FinishJob` rule. Integration tests: SetAvailable writes digest/source poll and records version, second call keeps first_seen_at; TestSnapshotRecordsRunningVersions (tag+digest, tagless digest, no digest skipped, re-snapshot idempotent); TestRecentVersions (order, limit 10, repo filter, digest-less row hidden only when a digest row for the same tag was first seen at or after it); GetImageVersion ErrNotFound; SetPollError sets/clears, upsert does not touch last_poll_error/available_*; FinishJob table extended (tag job clears poll suggestion; pick tag@old keeps suggestion tag@new; pick tag@new clears; webhook suggestion (digest '') cleared by matching tag; tagless pick keeps it). Update the e2e `memStore` in the same commit.
+
+### Task 2: Protocol, agent and hub routing (C7, C6 Route)
+Bump protocol, MinPinnedRefVersion, docs; Resolver pass-through; compose/swarm output line; `hub.Route(svc, minProtocol)`. Tests: TestResolve table (tag resolves; tag@d and @d pass through with zero inspect calls; bare errors); compose pinned ref verbatim in override; swarm spec image equals pinned ref; route rows v2 agent + min 3 → ErrAgentTooOld, v3 → ok, swarm fallback to v3 manager when reporting manager is v2.
+
+### Task 3: Jobs (C6)
+`Start` picks min version from the ref; `TriggerVersion`. Tests: version of other repo → ErrNotFound, no CreateJob; ref construction table; min version passed to fake hub (plain Trigger → 2, pinned → 3); explicit ref bypasses no-tag but not removed/self/built/container; existing tests green.
+
+### Task 4: Registry client and poller (C4, C5)
+Registry tests (local httptest + ggcr/pkg/registry, no network): random.Index digest equals index digest; recording handler shows only HEAD on /manifests/; basic-auth handler succeeds with temp DOCKER_CONFIG (t.Setenv) and fails without; 404 → error. Verify behaviour of DefaultKeychain when config.json has credsStore but no helper binary; document. Poller tests per Review Focus, plus Run test (first cycle after delay, no overlap, returns on cancel).
+
+### Task 5: Config, wiring, web (C8, C9)
+Config POLL_INTERVAL with tests; add to setAll list. `cmd/master/main.go`: registry.New, poller, `go p.Run(ctx)` with done channel, wait for it after HTTP shutdown and before st.Close; log registry_credentials. Web handlers/templates per C9 with tests (picker ≤10 rows with CSRF and version IDs; hidden for built/self/container/removed; POST with version → TriggerVersion, without → Trigger; 400/404/409; overview badges; Polling row).
+
+### Task 6: End-to-end
+`internal/e2e/versions_test.go`: TestPickPinnedVersion; TestPollStartsAutoJob (poller.Cycle with fake registry → one TriggerPoll job to repo:tag; second Cycle starts nothing). `currentHub.Route` new signature.
+
+### Task 7: Documentation
+README (POLL_INTERVAL, DOCKER_CONFIG, "Registry polling" section: polled/skipped, credentials only inline auths, Docker Hub note, version picker & rollback, auto caveat, agent protocol 3 for pinned/tagless picks). Spec: replace "Believed not to count against Docker Hub pull limits for HEAD requests — verify during implementation" with verified statement + source link (docs.docker.com/docker-hub/usage/pulls: "Using GET emulates a real pull and counts towards the limit. Using HEAD won't."; note the separate unquantified abuse rate limit). `deploy/master.stack.yml`: mount config dir read-only, set DOCKER_CONFIG. CLAUDE.md: architecture entries for registry and poller; gotchas (HEAD only; digest-less webhook row hidden once digest known; image_versions uses '' not NULL; picks by version ID and pinned/tagless need protocol 3; FinishJob rule includes digest (replace existing bullet); rollback to repo@digest makes the service follow no tag); phase 4 plan reference. Save this plan as `docs/superpowers/plans/2026-10-13-master-agent-phase-4-polling.md`.
+
+## Final verification
+
+```bash
+gofmt -l . ; go vet ./...
+go build ./cmd/master ./cmd/agent && rm -f master agent
+go test ./...
+go test -race ./internal/agent/... ./internal/master/... ./internal/e2e/
+DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
+```

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

@@ -247,8 +247,12 @@ them. An agent recreating itself would kill the job it is running.
 - 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.
+- Interval `POLL_INTERVAL`, default 6h. Docker Hub documents that HEAD requests do not
+  count against the pull limit ("Using GET emulates a real pull and counts towards the
+  limit. Using HEAD won't."; verified against
+  https://docs.docker.com/docker-hub/usage/pulls/). The same page mentions a separate,
+  unquantified abuse rate limit for many pulls from one IP address, so the poller stays
+  HEAD-only and never downloads a manifest.
 - Credentials: a Docker `config.json` (as written by `docker login`) mounted into the
   master. No credential UI, no encryption in Postgres.
 

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/coreos/go-oidc/v3 v3.21.0
 	github.com/distribution/reference v0.6.0
 	github.com/docker/docker v28.5.2+incompatible
+	github.com/google/go-containerregistry v0.22.1
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/opencontainers/go-digest v1.0.0
 	github.com/stretchr/testify v1.11.1
@@ -21,6 +22,8 @@ require (
 	github.com/containerd/errdefs/pkg v0.3.0 // indirect
 	github.com/containerd/log v0.1.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/docker/cli v29.7.2+incompatible // indirect
+	github.com/docker/docker-credential-helpers v0.9.3 // indirect
 	github.com/docker/go-connections v0.7.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
@@ -30,6 +33,7 @@ require (
 	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/klauspost/compress v1.19.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
@@ -37,14 +41,15 @@ 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/sirupsen/logrus v1.9.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect
 	go.opentelemetry.io/otel v1.43.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 // indirect
 	go.opentelemetry.io/otel/metric v1.43.0 // indirect
 	go.opentelemetry.io/otel/trace v1.43.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
-	golang.org/x/sys v0.42.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.35.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
 	gotest.tools/v3 v3.5.2 // indirect

go.sum

@@ -21,8 +21,12 @@ 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=
 github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
+github.com/docker/cli v29.7.2+incompatible h1:dlkwallR8XqfeVnA2ELEhdwvb4lsSwuB4IgsG8Q9cLY=
+github.com/docker/cli v29.7.2+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
 github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM=
 github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker-credential-helpers v0.9.3 h1:gAm/VtF9wgqJMoxzT3Gj5p4AqIjCBS4wrsOh9yRqcz8=
+github.com/docker/docker-credential-helpers v0.9.3/go.mod h1:x+4Gbw9aGmChi3qTLZj8Dfn0TD20M/fuWy0E5+WDeCo=
 github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c=
 github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -38,6 +42,8 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
 github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/go-containerregistry v0.22.1 h1:RZuuSYhTvlDvtsK+NkutoCZ//C0X2ebLK8X8l3ULs84=
+github.com/google/go-containerregistry v0.22.1/go.mod h1:bJR35SK8XgisYmhg/FMQ/5RK0S/XrOAqLBV5/LR2XE0=
 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=
@@ -50,6 +56,8 @@ 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.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8=
+github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -74,8 +82,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 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/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w=
+github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g=
 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=
@@ -101,18 +109,22 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09
 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
 go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g=
 go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk=
+golang.org/x/mod v0.39.0 h1:UF5zwQdCRRUpHfyPwr7d4UrGiVeldIsogtzWVnczL74=
+golang.org/x/mod v0.39.0/go.mod h1:bvIbwjQ0HUFFf5AKukeeYQG4ZBUG9yxQbR9aEweIwYY=
 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/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
 golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
-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/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
 golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
 golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
 golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
+golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI=
+golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo=
 google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA=
 google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:7QBABkRtR8z+TEnmXTqIqwJLlzrZKVfAUm7tY3yGv0M=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg=

internal/agent/updater/compose.go

@@ -91,10 +91,10 @@ func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run
 	return &ComposeExecutor{cli: cli, resolve: resolve, selfID: selfID, run: run, build: NewBuildChecker(cli, selfID, runOut)}
 }
 
-// Execute pins job.Ref to its current digest, writes it to the stack's
-// override file and recreates the service. Working directory and config
-// files come from the service's container labels on this machine, never from
-// the job. It returns the pinned ref.
+// Execute pins job.Ref to its current digest (a ref that has one is used as
+// is), writes it to the stack's override file and recreates the service.
+// Working directory and config files come from the service's container
+// labels on this machine, never from the job. It returns the pinned ref.
 func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
 	project, service := job.ComposeProject, job.ComposeService
 	cts, err := e.cli.ContainerList(ctx, container.ListOptions{
@@ -150,7 +150,7 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, ou
 	if err != nil {
 		return "", err
 	}
-	fmt.Fprintf(out, "resolved %s\n", pinned)
+	fmt.Fprintln(out, resolveNote(job.Ref, pinned))
 
 	if _, err := WriteOverride(dir, service, pinned); err != nil {
 		return "", err

internal/agent/updater/compose_test.go

@@ -108,6 +108,24 @@ func TestComposeExecutor(t *testing.T) {
 		assert.Equal(t, pinned, serviceImage(doc, "web"))
 		assert.Contains(t, out.String(), "resolved "+pinned)
 	})
+	t.Run("a pinned ref is written verbatim and not resolved", func(t *testing.T) {
+		t.Cleanup(cleanOverride)
+		for _, ref := range []string{"gitea.shcizo.se/shcizo/slusk:1.3@" + d, "gitea.shcizo.se/shcizo/slusk@" + d} {
+			rec := &recorder{}
+			dist := &fakeDist{digest: "sha256:other"}
+			var out strings.Builder
+			pinnedJob := job
+			pinnedJob.Ref = ref
+			got, err := setup(fakeCompose{rw, []container.Summary{ct}}, dist, rec).Execute(context.Background(), pinnedJob, &out)
+			require.NoError(t, err)
+			assert.Equal(t, ref, got)
+			assert.Empty(t, dist.refs, "no registry lookup")
+			_, doc := readOverride(t, stack)
+			assert.Equal(t, ref, serviceImage(doc, "web"))
+			assert.Contains(t, out.String(), "pinned "+ref)
+			assert.NotContains(t, out.String(), "resolved")
+		}
+	})
 	t.Run("built from source is refused before anything is written", func(t *testing.T) {
 		t.Cleanup(cleanOverride)
 		rec := &recorder{}

internal/agent/updater/resolve.go

@@ -16,7 +16,8 @@ type DistributionClient interface {
 }
 
 // Resolver turns "repo:tag" into "repo:tag@digest" by asking the local
-// daemon, which contacts the registry.
+// daemon, which contacts the registry. A ref that already has a digest is
+// passed through.
 type Resolver struct {
 	cli  DistributionClient
 	auth *RegistryAuth
@@ -27,13 +28,15 @@ func NewResolver(cli DistributionClient, auth *RegistryAuth) *Resolver {
 	return &Resolver{cli: cli, auth: auth}
 }
 
-// Resolve returns ref pinned to the digest its tag currently points at. It
-// fails for a ref without a tag or with a digest already.
+// Resolve returns ref pinned to the digest its tag currently points at. A ref
+// that already has a digest ("repo:tag@digest" or "repo@digest") is returned
+// unchanged without asking the daemon. It fails for a ref with neither tag nor
+// digest.
 func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) {
 	p := imageref.Parse(ref)
 	switch {
 	case p.Digest != "":
-		return "", fmt.Errorf("image %q already has a digest", ref)
+		return ref, nil
 	case p.Tag == "":
 		return "", fmt.Errorf("image %q has no tag to resolve", ref)
 	}
@@ -46,3 +49,13 @@ func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) {
 	}
 	return ref + "@" + info.Descriptor.Digest.String(), nil
 }
+
+// resolveNote is the job-output line that says which ref is deployed:
+// "resolved" when pinned came from resolving a tag, "pinned" when ref already
+// carried its digest.
+func resolveNote(ref, pinned string) string {
+	if pinned == ref {
+		return "pinned " + pinned
+	}
+	return "resolved " + pinned
+}

internal/agent/updater/resolve_test.go

@@ -21,14 +21,21 @@ func TestResolver(t *testing.T) {
 		assert.Equal(t, []string{"gitea.shcizo.se/shcizo/slusk:1.4"}, dist.refs)
 		assert.Equal(t, []string{"enc-gitea"}, dist.auths)
 	})
-	t.Run("digest or missing tag is refused without asking the daemon", func(t *testing.T) {
-		for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "repo"} {
+	t.Run("a ref with a digest passes through without asking the daemon", func(t *testing.T) {
+		for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "localhost:5000/repo:1@sha256:x"} {
 			dist := &fakeDist{digest: d}
-			_, err := NewResolver(dist, nil).Resolve(context.Background(), ref)
-			assert.Error(t, err, ref)
+			got, err := NewResolver(dist, nil).Resolve(context.Background(), ref)
+			require.NoError(t, err, ref)
+			assert.Equal(t, ref, got)
 			assert.Empty(t, dist.refs)
 		}
 	})
+	t.Run("a bare ref is refused without asking the daemon", func(t *testing.T) {
+		dist := &fakeDist{digest: d}
+		_, err := NewResolver(dist, nil).Resolve(context.Background(), "repo")
+		assert.ErrorContains(t, err, "has no tag to resolve")
+		assert.Empty(t, dist.refs)
+	})
 	t.Run("daemon error is wrapped", func(t *testing.T) {
 		cause := errors.New("unauthorized")
 		_, err := NewResolver(&fakeDist{err: cause}, nil).Resolve(context.Background(), "repo:1")

internal/agent/updater/swarm.go

@@ -35,11 +35,11 @@ func NewSwarmExecutor(cli SwarmClient, resolve *Resolver, auth *RegistryAuth) *S
 }
 
 // Execute checks that the local daemon is a manager of the job's cluster,
-// pins job.Ref to its current digest and updates the service to it. The
-// update is addressed by the ID and version of the service that was
-// inspected, so it changes exactly that object. Success means the daemon
-// accepted the new spec, not that the tasks have converged. It returns the
-// pinned ref.
+// pins job.Ref to its current digest (a ref that has one is used as is) and
+// updates the service to it. The update is addressed by the ID and version of
+// the service that was inspected, so it changes exactly that object. Success
+// means the daemon accepted the new spec, not that the tasks have converged.
+// It returns the pinned ref.
 func (e *SwarmExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
 	info, err := e.cli.Info(ctx)
 	if err != nil {
@@ -71,7 +71,7 @@ func (e *SwarmExecutor) Execute(ctx context.Context, job protocol.JobStart, out
 	if err != nil {
 		return "", err
 	}
-	fmt.Fprintf(out, "resolved %s\n", pinned)
+	fmt.Fprintln(out, resolveNote(job.Ref, pinned))
 
 	spec := svc.Spec
 	spec.TaskTemplate.ContainerSpec.Image = pinned

internal/agent/updater/swarm_test.go

@@ -77,6 +77,22 @@ func swarmSvc(name string) swarm.Service {
 	}
 }
 
+func TestSwarmUsesPinnedRef(t *testing.T) {
+	const ref = "gitea/gitea:1.26@sha256:abababababababababababababababababababababababababababababababab"
+	job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: ref}
+	fake := &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea")}
+	dist := &fakeDist{digest: "sha256:other"}
+	var out bytes.Buffer
+
+	got, err := NewSwarmExecutor(fake, NewResolver(dist, nil), nil).Execute(context.Background(), job, &out)
+
+	require.NoError(t, err)
+	assert.Equal(t, ref, got)
+	assert.Equal(t, ref, fake.updSpec.TaskTemplate.ContainerSpec.Image)
+	assert.Empty(t, dist.refs, "no registry lookup")
+	assert.Contains(t, out.String(), "pinned "+ref)
+}
+
 func TestSwarmExecutor(t *testing.T) {
 	job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"}
 	workerInfo := managerOf("c1")

internal/e2e/agent_master_test.go

@@ -11,6 +11,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"regexp"
 	"sort"
 	"strings"
 	"sync"
@@ -31,6 +32,9 @@ import (
 	"github.com/shcizo/package-updater/internal/protocol"
 )
 
+// digestPattern is the digest format the real store records.
+var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+
 type memStore struct {
 	mu        sync.Mutex
 	agents    map[string]model.Agent
@@ -44,6 +48,63 @@ type memStore struct {
 	jobs          map[int64]*model.Job
 	nextJob       int64
 	finishApplied int // FinishJob calls that returned true
+	versions      map[int64]model.ImageVersion
+}
+
+// seedVersion adds an image_versions row that GetImageVersion returns.
+func (m *memStore) seedVersion(v model.ImageVersion) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.versions == nil {
+		m.versions = map[int64]model.ImageVersion{}
+	}
+	m.versions[v.ID] = v
+}
+
+func (m *memStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	v, ok := m.versions[id]
+	if !ok {
+		return model.ImageVersion{}, model.ErrNotFound
+	}
+	return v, nil
+}
+
+func (m *memStore) ListServices(_ context.Context) ([]model.Service, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var out []model.Service
+	for _, svc := range m.svcRows {
+		if svc.RemovedAt == nil {
+			out = append(out, svc)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
+	return out, nil
+}
+
+func (m *memStore) SetPollError(_ context.Context, ids []int64, msg string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, id := range ids {
+		svc := m.svcRows[id]
+		svc.LastPollError = msg
+		m.svcRows[id] = svc
+	}
+	return nil
+}
+
+func (m *memStore) SetLastTagDigest(_ context.Context, id int64, digest string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	svc, ok := m.svcRows[id]
+	if !ok {
+		return model.ErrNotFound
+	}
+	svc.LastTagDigest = digest
+	m.svcRows[id] = svc
+	return nil
 }
 
 // seedService adds a service row that jobs can be triggered for.
@@ -88,7 +149,7 @@ func (m *memStore) SetAvailable(_ context.Context, id int64, a model.Available)
 	if !ok {
 		return model.ErrNotFound
 	}
-	svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Source, &a.At
+	svc.AvailableRef, svc.AvailableDigest, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Digest, a.Source, &a.At
 	m.svcRows[id] = svc
 	return nil
 }
@@ -184,10 +245,14 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo
 	if f.Status == model.JobSucceeded {
 		svc := m.svcRows[j.ServiceID]
 		svc.DesiredRef = f.Ref
+		if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && digestPattern.MatchString(d) {
+			svc.LastTagDigest = d
+		}
 		// Same rule as the SQL: the job was created for the suggested ref after it was reported.
-		requested, _, _ := strings.Cut(requestedRef, "@")
-		if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) {
-			svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = "", "", nil
+		requested, digest, _ := strings.Cut(requestedRef, "@")
+		if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) &&
+			(digest == "" || svc.AvailableDigest == digest) {
+			svc.AvailableRef, svc.AvailableDigest, svc.AvailableSource, svc.AvailableAt = "", "", "", nil
 		}
 		m.svcRows[j.ServiceID] = svc
 	}
@@ -282,7 +347,9 @@ type currentHub struct{ p atomic.Pointer[hub.Hub] }
 
 func (c *currentHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { c.p.Load().ServeHTTP(w, r) }
 func (c *currentHub) Disconnect(id int64)                              { c.p.Load().Disconnect(id) }
-func (c *currentHub) Route(svc model.Service) (model.Route, error)     { return c.p.Load().Route(svc) }
+func (c *currentHub) Route(svc model.Service, minProtocol int) (model.Route, error) {
+	return c.p.Load().Route(svc, minProtocol)
+}
 func (c *currentHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
 	return c.p.Load().StartJob(r, id, s)
 }

internal/e2e/jobs_test.go

@@ -32,6 +32,9 @@ const (
 	lostMessage = "lost: the agent did not know this job when it reconnected"
 )
 
+// e2eDigest is the digest the fake executor pins a tag to.
+const e2eDigest = "sha256:" + "e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0e2e0"
+
 // gateExecutor writes a line of output, then blocks until release is closed
 // or the job is cancelled.
 type gateExecutor struct {
@@ -52,7 +55,10 @@ func (g *gateExecutor) Execute(ctx context.Context, job protocol.JobStart, out i
 		return "", ctx.Err()
 	}
 	_, _ = io.WriteString(out, "recreated\n")
-	return job.Ref + "@sha256:e2e", nil
+	if strings.Contains(job.Ref, "@") {
+		return job.Ref, nil
+	}
+	return job.Ref + "@" + e2eDigest, nil
 }
 
 // jobAgent is a connected agent with a real runner and its two lifetimes: the
@@ -119,7 +125,7 @@ func (e *jobEnv) waitRoutable(t *testing.T) {
 	t.Helper()
 	svc, err := e.store.GetService(context.Background(), 1)
 	require.NoError(t, err)
-	require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc); return err == nil }, waitFor, pollEvery)
+	require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc, protocol.MinJobVersion); return err == nil }, waitFor, pollEvery)
 }
 
 func (e *jobEnv) job(t *testing.T, id int64) model.Job {
@@ -159,12 +165,12 @@ func TestManualJobFlowsToAgentAndBack(t *testing.T) {
 	j := e.trigger(t)
 	done := e.waitStatus(t, j.ID, model.JobSucceeded)
 
-	assert.Equal(t, slusk+":1.4.2@sha256:e2e", done.ToRef)
+	assert.Equal(t, slusk+":1.4.2@"+e2eDigest, done.ToRef)
 	assert.Contains(t, done.Output, "pulling "+slusk+":1.4.2")
 	assert.Contains(t, done.Output, "recreated")
 	svc, err := e.store.GetService(context.Background(), 1)
 	require.NoError(t, err)
-	assert.Equal(t, slusk+":1.4.2@sha256:e2e", svc.DesiredRef)
+	assert.Equal(t, slusk+":1.4.2@"+e2eDigest, svc.DesiredRef)
 	require.Eventually(t, func() bool { return len(e.agent.runner.Known()) == 0 }, waitFor, pollEvery, "the result must be acked")
 }
 
@@ -240,7 +246,7 @@ func TestPhaseOneAgentGetsNoJobs(t *testing.T) {
 	require.NoError(t, wsjson.Write(ctx, c, snap))
 
 	require.Eventually(t, func() bool {
-		_, err := m.hub.Route(svcRow)
+		_, err := m.hub.Route(svcRow, protocol.MinJobVersion)
 		return errors.Is(err, model.ErrAgentTooOld)
 	}, waitFor, pollEvery)
 

internal/e2e/versions_test.go

@@ -0,0 +1,133 @@
+package e2e
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/jobs"
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/master/poller"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// jobCount is the number of jobs created so far.
+func (m *memStore) jobCount() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return len(m.jobs)
+}
+
+// fixedRegistry answers every lookup from a fixed table.
+type fixedRegistry struct {
+	digests map[string]string
+	calls   int
+}
+
+func (f *fixedRegistry) Digest(_ context.Context, ref string) (string, error) {
+	f.calls++
+	return f.digests[ref], nil
+}
+
+func TestPickPinnedVersion(t *testing.T) {
+	env := newJobEnv(t)
+	env.store.seedVersion(model.ImageVersion{ID: 5, Repo: slusk, Tag: "1.3.0", Digest: "sha256:old130", Source: model.SourceRunning})
+
+	job, err := env.jobs.TriggerVersion(context.Background(), 1, 5, "alice")
+
+	require.NoError(t, err)
+	assert.Equal(t, slusk+":1.3.0@sha256:old130", job.ToRef)
+	select {
+	case ref := <-env.exec.started:
+		assert.Equal(t, slusk+":1.3.0@sha256:old130", ref, "the agent receives exactly the picked version")
+	case <-time.After(waitFor):
+		t.Fatal("the agent never received the job")
+	}
+	close(env.exec.release)
+	done := env.waitStatus(t, job.ID, model.JobSucceeded)
+	assert.Equal(t, model.TriggerManual, done.Trigger)
+	assert.Equal(t, "alice", done.RequestedBy)
+}
+
+func TestPickVersionOfAnotherRepoIsRefused(t *testing.T) {
+	env := newJobEnv(t)
+	env.store.seedVersion(model.ImageVersion{ID: 6, Repo: "evil/other", Tag: "1", Digest: "sha256:x", Source: model.SourcePoll})
+
+	_, err := env.jobs.TriggerVersion(context.Background(), 1, 6, "alice")
+
+	assert.ErrorIs(t, err, model.ErrNotFound)
+	assert.Zero(t, env.store.jobCount(), "no job was created")
+}
+
+func TestPollStartsAutoJob(t *testing.T) {
+	env := newJobEnv(t)
+	env.store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
+		AgentID: ptr(agentID), ImageRepo: slusk, ActualRef: slusk + ":1.4.2@sha256:old", ActualDigest: "sha256:old",
+		UpdateMode: model.ModeAuto})
+	reg := &fixedRegistry{digests: map[string]string{slusk + ":1.4.2": "sha256:new"}}
+	p := poller.New(env.store, env.jobs, reg, env.m.broker, poller.Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil)))
+
+	p.Cycle(context.Background())
+
+	select {
+	case ref := <-env.exec.started:
+		assert.Equal(t, slusk+":1.4.2", ref, "the agent gets repo:tag and resolves the digest itself")
+	case <-time.After(waitFor):
+		t.Fatal("the agent never received the job")
+	}
+	require.Equal(t, 1, env.store.jobCount())
+	job := env.job(t, 1)
+	assert.Equal(t, model.TriggerPoll, job.Trigger)
+	assert.Equal(t, slusk+":1.4.2", job.ToRef)
+	svc, err := env.store.GetService(context.Background(), 1)
+	require.NoError(t, err)
+	assert.Equal(t, slusk+":1.4.2", svc.AvailableRef)
+	assert.Equal(t, "sha256:new", svc.AvailableDigest)
+	assert.Equal(t, model.SourcePoll, svc.AvailableSource)
+	assert.Equal(t, "sha256:new", svc.LastTagDigest)
+
+	p.Cycle(context.Background())
+
+	assert.Equal(t, 2, reg.calls, "the registry is asked again")
+	assert.Equal(t, 1, env.store.jobCount(), "the same digest does not start another job")
+	close(env.exec.release)
+	env.waitStatus(t, job.ID, model.JobSucceeded)
+}
+
+// An update started by the webhook, then a rollback by hand: polling must not
+// redo the update while the registry still shows the digest that was deployed.
+func TestPollKeepsARollbackAfterAWebhookUpdate(t *testing.T) {
+	env := newJobEnv(t)
+	ctx := context.Background()
+	close(env.exec.release)
+	env.store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
+		AgentID: ptr(agentID), ImageRepo: slusk, ActualRef: slusk + ":main@sha256:old", ActualDigest: "sha256:old",
+		UpdateMode: model.ModeAuto})
+	env.store.seedVersion(model.ImageVersion{ID: 5, Repo: slusk, Tag: "main", Digest: "sha256:old", Source: model.SourceRunning})
+	svc, err := env.store.GetService(ctx, 1)
+	require.NoError(t, err)
+
+	hook, err := env.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerWebhook, Ref: slusk + ":main"})
+	require.NoError(t, err)
+	env.waitStatus(t, hook.ID, model.JobSucceeded)
+	svc, err = env.store.GetService(ctx, 1)
+	require.NoError(t, err)
+	assert.Equal(t, e2eDigest, svc.LastTagDigest, "the deployed digest of the tag is remembered")
+
+	pick, err := env.jobs.TriggerVersion(ctx, 1, 5, "alice")
+	require.NoError(t, err)
+	env.waitStatus(t, pick.ID, model.JobSucceeded)
+	jobsBefore := env.store.jobCount()
+
+	reg := &fixedRegistry{digests: map[string]string{slusk + ":main": e2eDigest}}
+	p := poller.New(env.store, env.jobs, reg, env.m.broker, poller.Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	p.Cycle(ctx)
+
+	assert.Equal(t, 1, reg.calls)
+	assert.Equal(t, jobsBefore, env.store.jobCount(), "the rollback stays")
+}

internal/e2e/webhook_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/shcizo/package-updater/internal/master/model"
 	"github.com/shcizo/package-updater/internal/master/webhook"
+	"github.com/shcizo/package-updater/internal/protocol"
 )
 
 // webhookServer mounts the webhook handler on a test server, wired to the
@@ -98,7 +99,7 @@ func TestWebhookWaitsForAutoJob(t *testing.T) {
 	job := e.job(t, r.body.Results[0].JobID)
 	assert.Equal(t, model.TriggerWebhook, job.Trigger)
 	assert.Equal(t, model.JobSucceeded, job.Status)
-	assert.Equal(t, slusk+":2.0.0@sha256:e2e", job.ToRef)
+	assert.Equal(t, slusk+":2.0.0@"+e2eDigest, job.ToRef)
 	svc, err := e.store.GetService(context.Background(), 1)
 	require.NoError(t, err)
 	assert.Empty(t, svc.AvailableRef, "the job deployed the suggestion, so it is cleared")
@@ -111,7 +112,7 @@ func TestWebhookFailedAutoJobIs500(t *testing.T) {
 	e.agent.stopConn() // no agent connection: the job fails at once with "agent offline"
 	require.Eventually(t, func() bool {
 		svc, _ := e.store.GetService(context.Background(), 1)
-		_, err := e.m.hub.Route(svc)
+		_, err := e.m.hub.Route(svc, protocol.MinJobVersion)
 		return err != nil
 	}, waitFor, pollEvery)
 

internal/master/config/config.go

@@ -28,11 +28,17 @@ type Config struct {
 	// UpdateTimeout bounds how long one webhook request waits for its auto
 	// jobs (UPDATE_TIMEOUT, default 5m).
 	UpdateTimeout time.Duration
+	// PollInterval is the time between registry polling cycles
+	// (POLL_INTERVAL, default 6h).
+	PollInterval time.Duration
 }
 
 // defaultUpdateTimeout is used when UPDATE_TIMEOUT is unset.
 const defaultUpdateTimeout = 5 * time.Minute
 
+// defaultPollInterval is used when POLL_INTERVAL is unset.
+const defaultPollInterval = 6 * time.Hour
+
 // Load reads and validates the master configuration, failing fast on
 // anything missing so a half-configured master never starts.
 func Load() (*Config, error) {
@@ -50,6 +56,7 @@ func Load() (*Config, error) {
 		LogLevel:         getenvDefault("LOG_LEVEL", "info"),
 		UpdaterAPIKey:    os.Getenv("UPDATER_API_KEY"),
 		UpdateTimeout:    defaultUpdateTimeout,
+		PollInterval:     defaultPollInterval,
 	}
 	for _, r := range []struct{ name, value string }{
 		{"DATABASE_URL", cfg.DatabaseURL},
@@ -77,6 +84,13 @@ func Load() (*Config, error) {
 		}
 		cfg.UpdateTimeout = d
 	}
+	if v := os.Getenv("POLL_INTERVAL"); v != "" {
+		d, err := time.ParseDuration(v)
+		if err != nil || d <= 0 {
+			return nil, fmt.Errorf("POLL_INTERVAL must be a positive duration such as 6h, got %q", v)
+		}
+		cfg.PollInterval = d
+	}
 	return cfg, nil
 }
 

internal/master/config/config_test.go

@@ -22,7 +22,7 @@ func setAll(t *testing.T) {
 	for k, v := range required {
 		t.Setenv(k, v)
 	}
-	for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL", "UPDATER_API_KEY", "UPDATE_TIMEOUT"} {
+	for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL", "UPDATER_API_KEY", "UPDATE_TIMEOUT", "POLL_INTERVAL"} {
 		t.Setenv(k, "")
 	}
 }
@@ -38,6 +38,26 @@ func TestLoad_Defaults(t *testing.T) {
 	assert.Equal(t, "updater-admins", cfg.AllowedGroup)
 	assert.Empty(t, cfg.UpdaterAPIKey)
 	assert.Equal(t, 5*time.Minute, cfg.UpdateTimeout)
+	assert.Equal(t, 6*time.Hour, cfg.PollInterval)
+}
+
+func TestLoad_PollInterval(t *testing.T) {
+	setAll(t)
+	t.Setenv("POLL_INTERVAL", "30m")
+	cfg, err := Load()
+	require.NoError(t, err)
+	assert.Equal(t, 30*time.Minute, cfg.PollInterval)
+}
+
+func TestLoad_InvalidPollInterval(t *testing.T) {
+	for _, v := range []string{"abc", "0s", "-1h", "6"} {
+		t.Run(v, func(t *testing.T) {
+			setAll(t)
+			t.Setenv("POLL_INTERVAL", v)
+			_, err := Load()
+			assert.ErrorContains(t, err, "POLL_INTERVAL must be a positive duration such as 6h, got")
+		})
+	}
 }
 
 func TestLoad_WebhookSettings(t *testing.T) {

internal/master/hub/jobs_test.go

@@ -43,7 +43,7 @@ func waitRoute(t *testing.T, h *Hub) model.Route {
 	var r model.Route
 	require.Eventually(t, func() bool {
 		var err error
-		r, err = h.Route(svc)
+		r, err = h.Route(svc, protocol.MinJobVersion)
 		return err == nil
 	}, 2*time.Second, 5*time.Millisecond)
 	return r
@@ -61,7 +61,7 @@ func TestHub_ReconcilesBeforeRoutable(t *testing.T) {
 	send(t, c, protocol.TypeHello, protocol.Hello{Hostname: "node-2", ProtocolVersion: 2, KnownJobs: []string{"7", "9", "x"}})
 	svc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(node2.ID)}
 	time.Sleep(50 * time.Millisecond) // hello is now blocked inside FailLostJobs
-	_, err = h.Route(svc)
+	_, err = h.Route(svc, protocol.MinJobVersion)
 	assert.ErrorIs(t, err, model.ErrAgentOffline)
 
 	close(store.lostGate)

internal/master/hub/route.go

@@ -14,9 +14,11 @@ type candidate struct {
 	hello   protocol.Hello
 }
 
-// Route picks the connection a job for svc goes to. It returns
-// model.ErrAgentOffline or model.ErrAgentTooOld when there is none.
-func (h *Hub) Route(svc model.Service) (model.Route, error) {
+// Route picks the connection a job for svc goes to, among agents that speak
+// at least protocol version minProtocol. It returns model.ErrAgentOffline or
+// model.ErrAgentTooOld (model.ErrAgentTooOldForPinned when minProtocol asks
+// for a pinned ref) when there is none.
+func (h *Hub) Route(svc model.Service, minProtocol int) (model.Route, error) {
 	h.mu.Lock()
 	conns := make([]candidate, 0, len(h.conns))
 	for id, c := range h.conns {
@@ -25,7 +27,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) {
 		}
 	}
 	h.mu.Unlock()
-	return route(svc, conns)
+	return route(svc, conns, minProtocol)
 }
 
 // route picks the connection a job for svc goes to. Compose services and
@@ -33,7 +35,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) {
 // run on the reporting agent if it qualifies, else on any other connected
 // manager of the same cluster (lowest agent ID first). Too-old wins over
 // offline only when a connection would otherwise have qualified.
-func route(svc model.Service, conns []candidate) (model.Route, error) {
+func route(svc model.Service, conns []candidate, minProtocol int) (model.Route, error) {
 	sort.Slice(conns, func(a, b int) bool { return conns[a].agentID < conns[b].agentID })
 	tooOld := false
 	try := func(c candidate) bool {
@@ -42,7 +44,7 @@ func route(svc model.Service, conns []candidate) (model.Route, error) {
 				return false
 			}
 		}
-		if c.hello.ProtocolVersion < protocol.MinJobVersion {
+		if c.hello.ProtocolVersion < minProtocol {
 			tooOld = true
 			return false
 		}
@@ -63,6 +65,9 @@ func route(svc model.Service, conns []candidate) (model.Route, error) {
 		}
 	}
 	if tooOld {
+		if minProtocol >= protocol.MinPinnedRefVersion {
+			return model.Route{}, model.ErrAgentTooOldForPinned
+		}
 		return model.Route{}, model.ErrAgentTooOld
 	}
 	return model.Route{}, model.ErrAgentOffline

internal/master/hub/route_test.go

@@ -12,37 +12,46 @@ import (
 func TestRoute(t *testing.T) {
 	composeSvc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(int64(2))}
 	swarmSvc := model.Service{Kind: protocol.KindSwarm, AgentID: ptr(int64(1)), SwarmClusterID: "c1"}
-	v2 := protocol.Hello{ProtocolVersion: protocol.Version}
-	mgrV2 := protocol.Hello{ProtocolVersion: protocol.Version, SwarmManager: true, SwarmClusterID: "c1"}
+	v2 := protocol.Hello{ProtocolVersion: 2}
+	v3 := protocol.Hello{ProtocolVersion: 3}
+	mgrV2 := protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c1"}
+	mgrV3 := protocol.Hello{ProtocolVersion: 3, SwarmManager: true, SwarmClusterID: "c1"}
+	jobs, pinned := protocol.MinJobVersion, protocol.MinPinnedRefVersion
 
 	tests := []struct {
 		name  string
 		svc   model.Service
 		conns []candidate
+		min   int
 		want  model.Route
 		err   error
 	}{
-		{"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, model.Route{AgentID: 2, ConnID: 20}, nil},
-		{"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, model.Route{}, model.ErrAgentTooOld},
-		{"compose, own agent missing", composeSvc, nil, model.Route{}, model.ErrAgentOffline},
-		{"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, model.Route{}, model.ErrAgentOffline},
-		{"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 1, ConnID: 10}, nil},
-		{"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
-		{"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
+		{"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, jobs, model.Route{AgentID: 2, ConnID: 20}, nil},
+		{"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, jobs, model.Route{}, model.ErrAgentTooOld},
+		{"compose, own agent missing", composeSvc, nil, jobs, model.Route{}, model.ErrAgentOffline},
+		{"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, jobs, model.Route{}, model.ErrAgentOffline},
+		{"compose, v2 agent and a pinned ref", composeSvc, []candidate{{2, 20, v2}}, pinned, model.Route{}, model.ErrAgentTooOldForPinned},
+		{"compose, v3 agent and a pinned ref", composeSvc, []candidate{{2, 20, v3}}, pinned, model.Route{AgentID: 2, ConnID: 20}, nil},
+		{"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 1, ConnID: 10}, nil},
+		{"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
+		{"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
 		{"swarm, other manager in another cluster", swarmSvc,
-			[]candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, model.Route{}, model.ErrAgentOffline},
+			[]candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, jobs, model.Route{}, model.ErrAgentOffline},
 		{"swarm, only a v0 manager of the cluster", swarmSvc,
-			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentTooOld},
+			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentTooOld},
 		{"swarm, v0 manager but a v2 manager exists", swarmSvc,
-			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
+			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
 		{"swarm, worker of the cluster", swarmSvc,
-			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentOffline},
+			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentOffline},
 		{"swarm, manager with empty cluster id", model.Service{Kind: protocol.KindSwarm},
-			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, model.Route{}, model.ErrAgentOffline},
+			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, jobs, model.Route{}, model.ErrAgentOffline},
+		{"swarm, pinned ref falls back to a v3 manager when the reporting one is v2", swarmSvc,
+			[]candidate{{1, 10, mgrV2}, {3, 30, mgrV3}}, pinned, model.Route{AgentID: 3, ConnID: 30}, nil},
+		{"swarm, pinned ref and only v2 managers", swarmSvc, []candidate{{1, 10, mgrV2}}, pinned, model.Route{}, model.ErrAgentTooOldForPinned},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := route(tt.svc, tt.conns)
+			got, err := route(tt.svc, tt.conns, tt.min)
 			assert.ErrorIs(t, err, tt.err)
 			assert.Equal(t, tt.want, got)
 		})

internal/master/jobs/jobs.go

@@ -20,13 +20,15 @@ type Store interface {
 	CreateJob(ctx context.Context, j model.NewJob) (model.Job, error)
 	FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error)
 	GetJob(ctx context.Context, id int64) (model.Job, error)
+	GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error)
 }
 
 // Hub routes jobs to agent connections. *hub.Hub satisfies it.
 type Hub interface {
-	// Route picks the connection for svc: model.ErrAgentOffline or
+	// Route picks the connection for svc among agents speaking at least
+	// protocol version minProtocol: model.ErrAgentOffline or
 	// model.ErrAgentTooOld when there is none.
-	Route(svc model.Service) (model.Route, error)
+	Route(svc model.Service, minProtocol int) (model.Route, error)
 	// StartJob sends job.start over exactly the routed connection.
 	// model.ErrAgentOffline if that connection is gone or was replaced.
 	StartJob(r model.Route, jobID int64, start protocol.JobStart) error
@@ -116,8 +118,9 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error {
 type StartRequest struct {
 	Trigger     model.JobTrigger
 	RequestedBy string // OIDC subject for manual jobs; "" otherwise
-	// Ref is the ref to deploy ("repo:tag"). Empty means the service's
-	// followed tag, derived from its actual ref.
+	// Ref is the ref to deploy ("repo:tag", "repo:tag@digest" or
+	// "repo@digest"). Empty means the service's followed tag, derived from
+	// its actual ref.
 	Ref string
 }
 
@@ -131,8 +134,31 @@ func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy stri
 	return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy})
 }
 
+// TriggerVersion starts a manual update of the service to exactly the image
+// version versionID, which must be a version of the service's own repo; the
+// ref is built from the stored version, never taken from the caller. It
+// returns model.ErrNotFound for an unknown service, an unknown version or a
+// version of another repo, and otherwise what Start returns. A version
+// without a tag leaves the service following no tag.
+func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error) {
+	svc, err := s.store.GetService(ctx, serviceID)
+	if err != nil {
+		return model.Job{}, err
+	}
+	v, err := s.store.GetImageVersion(ctx, versionID)
+	if err != nil {
+		return model.Job{}, err
+	}
+	if v.Repo != svc.ImageRepo {
+		return model.Job{}, model.ErrNotFound
+	}
+	return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: v.Ref()})
+}
+
 // Start starts an update of svc. It returns the job, which may already be
-// failed ("agent offline", "agent too old for updates"); a
+// failed ("agent offline", "agent too old for updates", or "agent too old to
+// deploy a pinned version" when the ref has a digest or no tag and the agent
+// speaks protocol version 2); a
 // *model.NotUpdatableError when the service cannot be updated (no job
 // created); the active job and model.ErrJobActive when one is queued or
 // running.
@@ -145,7 +171,11 @@ func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest
 		ref = FollowedRef(svc.ActualRef) // updatable checked the tag
 	}
 
-	r, routeErr := s.hub.Route(svc)
+	minProtocol := protocol.MinJobVersion
+	if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" {
+		minProtocol = protocol.MinPinnedRefVersion
+	}
+	r, routeErr := s.hub.Route(svc, minProtocol)
 	nj := model.NewJob{
 		ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref,
 		Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(),

internal/master/jobs/jobs_test.go

@@ -21,6 +21,7 @@ type fakeStore struct {
 	jobs     map[int64]model.Job
 	created  []model.NewJob
 	active   *model.Job // returned with ErrJobActive when set
+	versions map[int64]model.ImageVersion
 
 	revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these
 }
@@ -54,6 +55,14 @@ func (f *fakeStore) FailJob(_ context.Context, id int64, msg string, at time.Tim
 	return true, nil
 }
 
+func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) {
+	v, ok := f.versions[id]
+	if !ok {
+		return model.ImageVersion{}, model.ErrNotFound
+	}
+	return v, nil
+}
+
 func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil }
 
 type startCall struct {
@@ -63,13 +72,24 @@ type startCall struct {
 }
 
 type fakeHub struct {
-	route    model.Route
-	routeErr error
-	startErr error
-	starts   []startCall
+	route         model.Route
+	routeErr      error
+	startErr      error
+	starts        []startCall
+	agentProtocol int   // when set, Route fails with ErrAgentTooOld for a higher minProtocol
+	mins          []int // minProtocol of every Route call
 }
 
-func (h *fakeHub) Route(model.Service) (model.Route, error) { return h.route, h.routeErr }
+func (h *fakeHub) Route(_ model.Service, minProtocol int) (model.Route, error) {
+	h.mins = append(h.mins, minProtocol)
+	if h.agentProtocol != 0 && minProtocol > h.agentProtocol {
+		if minProtocol >= protocol.MinPinnedRefVersion {
+			return model.Route{}, model.ErrAgentTooOldForPinned
+		}
+		return model.Route{}, model.ErrAgentTooOld
+	}
+	return h.route, h.routeErr
+}
 
 func (h *fakeHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
 	h.starts = append(h.starts, startCall{r, id, s})
@@ -341,3 +361,133 @@ func TestFollowedRef(t *testing.T) {
 	assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo"))
 	assert.Empty(t, FollowedRef("redis@sha256:abc"))
 }
+
+func TestStart_RoutesWithTheMinimumProtocolOfTheRef(t *testing.T) {
+	tests := []struct {
+		name string
+		ref  string // "" means the followed tag
+		want int
+	}{
+		{"followed tag", "", protocol.MinJobVersion},
+		{"repo:tag", "x/y:2", protocol.MinJobVersion},
+		{"repo:tag@digest", "x/y:2@sha256:a", protocol.MinPinnedRefVersion},
+		{"repo@digest", "x/y@sha256:a", protocol.MinPinnedRefVersion},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, _ := newService(st, h)
+
+			_, err := s.Start(context.Background(), composeSvc, StartRequest{Trigger: model.TriggerManual, Ref: tt.ref})
+
+			require.NoError(t, err)
+			assert.Equal(t, []int{tt.want}, h.mins)
+		})
+	}
+}
+
+func TestTriggerVersion_Ref(t *testing.T) {
+	const repo = "gitea.shcizo.se/shcizo/slusk"
+	noTag := model.Service{ID: 4, Kind: protocol.KindCompose, ComposeProject: "n", ComposeService: "n",
+		AgentID: ptr(int64(7)), ImageRepo: repo, ActualRef: repo + "@sha256:d1"}
+	svc := composeSvc
+	svc.ImageRepo = repo
+	tests := []struct {
+		name    string
+		svc     model.Service
+		version model.ImageVersion
+		wantRef string
+		wantMin int
+	}{
+		{"tag and digest", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
+		{"tag only", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "main"}, repo + ":main", protocol.MinJobVersion},
+		{"digest only", svc, model.ImageVersion{ID: 10, Repo: repo, Digest: "sha256:old"}, repo + "@sha256:old", protocol.MinPinnedRefVersion},
+		{"service that follows no tag", noTag, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			st := &fakeStore{
+				services: map[int64]model.Service{tt.svc.ID: tt.svc},
+				jobs:     map[int64]model.Job{},
+				versions: map[int64]model.ImageVersion{10: tt.version},
+			}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, _ := newService(st, h)
+
+			job, err := s.TriggerVersion(context.Background(), tt.svc.ID, 10, "sub-1")
+
+			require.NoError(t, err)
+			assert.Equal(t, model.JobQueued, job.Status)
+			require.Len(t, st.created, 1)
+			assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
+			assert.Equal(t, "sub-1", st.created[0].RequestedBy)
+			assert.Equal(t, tt.wantRef, st.created[0].ToRef)
+			require.Len(t, h.starts, 1)
+			assert.Equal(t, tt.wantRef, h.starts[0].start.Ref)
+			assert.Equal(t, []int{tt.wantMin}, h.mins)
+		})
+	}
+}
+
+func TestTriggerVersion_NotFound(t *testing.T) {
+	const repo = "gitea.shcizo.se/shcizo/slusk"
+	svc := composeSvc
+	svc.ImageRepo = repo
+	tests := []struct {
+		name      string
+		serviceID int64
+		versionID int64
+	}{
+		{"unknown service", 9, 10},
+		{"unknown version", 1, 99},
+		{"version of another repo", 1, 11},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			st := &fakeStore{
+				services: map[int64]model.Service{1: svc},
+				jobs:     map[int64]model.Job{},
+				versions: map[int64]model.ImageVersion{
+					10: {ID: 10, Repo: repo, Tag: "1"},
+					11: {ID: 11, Repo: "evil/other", Tag: "1"},
+				},
+			}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, pub := newService(st, h)
+
+			_, err := s.TriggerVersion(context.Background(), tt.serviceID, tt.versionID, "sub-1")
+
+			assert.ErrorIs(t, err, model.ErrNotFound)
+			assert.Empty(t, st.created, "no job")
+			assert.Empty(t, h.starts)
+			assert.Empty(t, pub.topics)
+		})
+	}
+}
+
+func TestTriggerVersion_RequiresPinnedProtocol(t *testing.T) {
+	const repo = "gitea.shcizo.se/shcizo/slusk"
+	svc := composeSvc
+	svc.ImageRepo = repo
+	st := &fakeStore{
+		services: map[int64]model.Service{1: svc},
+		jobs:     map[int64]model.Job{},
+		versions: map[int64]model.ImageVersion{
+			10: {ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"},
+			11: {ID: 11, Repo: repo, Tag: "main"},
+		},
+	}
+	h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2}
+	s, _ := newService(st, h)
+
+	pinned, err := s.TriggerVersion(context.Background(), 1, 10, "sub-1")
+	require.NoError(t, err)
+	assert.Equal(t, model.JobFailed, pinned.Status)
+	assert.Equal(t, "agent too old to deploy a pinned version (needs protocol 3)", pinned.Error)
+	assert.Empty(t, h.starts, "nothing sent to the old agent")
+
+	plain, err := s.TriggerVersion(context.Background(), 1, 11, "sub-1")
+	require.NoError(t, err)
+	assert.Equal(t, model.JobQueued, plain.Status)
+}

internal/master/model/job.go

@@ -15,8 +15,20 @@ var (
 	// ErrAgentTooOld means the only agent that could run the job speaks a
 	// protocol version without jobs.
 	ErrAgentTooOld = errors.New("agent too old for updates")
+	// ErrAgentTooOldForPinned means the only agent that could run the job
+	// speaks protocol version 2, which cannot deploy a digest-pinned or
+	// tagless ref. errors.Is(err, ErrAgentTooOld) is true for it as well.
+	ErrAgentTooOldForPinned error = agentTooOldForPinned{}
 )
 
+type agentTooOldForPinned struct{}
+
+func (agentTooOldForPinned) Error() string {
+	return "agent too old to deploy a pinned version (needs protocol 3)"
+}
+
+func (agentTooOldForPinned) Is(target error) bool { return target == ErrAgentTooOld }
+
 // NotUpdatableError says why a service cannot be updated. Reason is shown
 // in the UI as is.
 type NotUpdatableError struct {

internal/master/model/model.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/shcizo/package-updater/internal/imageref"
 	"github.com/shcizo/package-updater/internal/protocol"
 )
 
@@ -56,11 +57,20 @@ type Service struct {
 	// AvailableRef is the newest ref a webhook or poll reported for this
 	// service; "" if none, or if a later job already deployed it.
 	AvailableRef    string
-	AvailableSource string // SourceWebhook; "" when AvailableRef is empty
+	AvailableDigest string // digest of AvailableRef when known (poll); "" from the webhook
+	AvailableSource string // SourceWebhook or SourcePoll; "" when AvailableRef is empty
 	AvailableAt     *time.Time
-	FirstSeenAt     time.Time
-	UpdatedAt       time.Time
-	RemovedAt       *time.Time
+	LastPollError   string // last registry error for the followed ref; "" when fine
+	// LastTagDigest is the newest digest of the followed tag that was deployed
+	// or started: set when the poller starts an auto job, when it sees the service
+	// already running that digest, and when any plain tag job succeeds (the
+	// digest the agent pinned); "" if never. The poller does
+	// not start another job for the same digest, so a rollback sticks until the
+	// registry shows a different one.
+	LastTagDigest string
+	FirstSeenAt   time.Time
+	UpdatedAt     time.Time
+	RemovedAt     *time.Time
 }
 
 // Update modes of a service.
@@ -69,18 +79,40 @@ const (
 	ModeAuto   = "auto"
 )
 
-// Sources of an available ref.
+// Sources of an available ref and of an ImageVersion.
 const (
 	SourceWebhook = "webhook"
+	SourcePoll    = "poll"
+	// SourceRunning is used by image_versions only: an agent reported the
+	// version as running.
+	SourceRunning = "running"
 )
 
 // Available is a newer ref reported for a service.
 type Available struct {
 	Ref    string
-	Source string // SourceWebhook
+	Digest string // "" when unknown (webhook)
+	Source string // SourceWebhook or SourcePoll
 	At     time.Time
 }
 
+// ImageVersion is one version of an image repo the master has seen. Tag and
+// Digest are "" when the version has no tag or its digest is unknown; they
+// are never both empty.
+type ImageVersion struct {
+	ID          int64
+	Repo        string
+	Tag         string
+	Digest      string
+	Source      string // SourceWebhook, SourcePoll or SourceRunning: the first source that saw it
+	FirstSeenAt time.Time
+}
+
+// Ref returns "repo:tag@digest", "repo:tag" or "repo@digest".
+func (v ImageVersion) Ref() string {
+	return imageref.Ref{Repo: v.Repo, Tag: v.Tag, Digest: v.Digest}.String()
+}
+
 // DisplayName is the human-facing name of the service.
 func (s Service) DisplayName() string {
 	switch s.Kind {

internal/master/model/model_test.go

@@ -40,6 +40,24 @@ func TestDisplayName(t *testing.T) {
 	}
 }
 
+func TestImageVersionRef(t *testing.T) {
+	cases := []struct {
+		name string
+		v    ImageVersion
+		want string
+	}{
+		{"tag and digest", ImageVersion{Repo: "r/x", Tag: "1", Digest: "sha256:a"}, "r/x:1@sha256:a"},
+		{"tag only", ImageVersion{Repo: "r/x", Tag: "1"}, "r/x:1"},
+		{"digest only", ImageVersion{Repo: "r/x", Digest: "sha256:a"}, "r/x@sha256:a"},
+		{"registry port", ImageVersion{Repo: "localhost:5000/x", Tag: "1"}, "localhost:5000/x:1"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, c.v.Ref())
+		})
+	}
+}
+
 func TestJob_Active(t *testing.T) {
 	cases := []struct {
 		status JobStatus

internal/master/poller/poller.go

@@ -0,0 +1,320 @@
+// Package poller checks the registry for new digests of the tags the
+// services follow. A new digest becomes the service's available version, and
+// a service in auto mode gets an update job.
+package poller
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/shcizo/package-updater/internal/master/events"
+	"github.com/shcizo/package-updater/internal/master/jobs"
+	"github.com/shcizo/package-updater/internal/master/model"
+)
+
+// lookupTimeout bounds one registry lookup.
+const lookupTimeout = 30 * time.Second
+
+// maxErrorBytes caps the registry error stored in services.last_poll_error.
+const maxErrorBytes = 500
+
+// Store is the storage the poller needs. *store.Store satisfies it.
+type Store interface {
+	ListServices(ctx context.Context) ([]model.Service, error)
+	GetService(ctx context.Context, id int64) (model.Service, error)
+	SetAvailable(ctx context.Context, id int64, a model.Available) error
+	SetLastTagDigest(ctx context.Context, id int64, digest string) error
+	SetPollError(ctx context.Context, ids []int64, msg string) error
+}
+
+// Jobs starts updates. *jobs.Service satisfies it.
+type Jobs interface {
+	Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error)
+}
+
+// Registry looks up the digest a "repo:tag" ref points at. *registry.Client
+// satisfies it.
+type Registry interface {
+	Digest(ctx context.Context, ref string) (string, error)
+}
+
+// Publisher receives change signals. *events.Broker satisfies it.
+type Publisher interface{ Publish(t events.Topic) }
+
+// Config holds the poller settings.
+type Config struct {
+	Interval   time.Duration // time between the end of a cycle and the start of the next (POLL_INTERVAL)
+	FirstDelay time.Duration // wait before the first cycle; 1 minute in production
+	Workers    int           // registry lookups in flight at once; 4 in production
+}
+
+// Poller runs polling cycles.
+type Poller struct {
+	store Store
+	jobs  Jobs
+	reg   Registry
+	pub   Publisher
+	cfg   Config
+	log   *slog.Logger
+	now   func() time.Time
+}
+
+// New returns a Poller.
+func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Logger) *Poller {
+	if cfg.Workers < 1 {
+		cfg.Workers = 1
+	}
+	return &Poller{store: st, jobs: j, reg: reg, pub: pub, cfg: cfg, log: log, now: time.Now}
+}
+
+// SkipReason says why svc is not polled: "follows no tag", "built from
+// source", "no registry digest", or "" when it is polled.
+func SkipReason(svc model.Service) string {
+	switch {
+	case jobs.FollowedRef(svc.ActualRef) == "":
+		return model.ReasonNoTag
+	case svc.Built:
+		return "built from source"
+	case svc.ActualDigest == "":
+		return "no registry digest"
+	}
+	return ""
+}
+
+// Run polls until ctx is done. The first cycle starts Config.FirstDelay after
+// the call and each following one Config.Interval after the previous cycle
+// ended, so cycles never overlap. It blocks.
+func (p *Poller) Run(ctx context.Context) {
+	timer := time.NewTimer(p.cfg.FirstDelay)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+		}
+		p.Cycle(ctx)
+		timer.Reset(p.cfg.Interval)
+	}
+}
+
+// lookup is the registry's answer for one followed ref.
+type lookup struct {
+	ref    string
+	digest string
+	err    error
+}
+
+// cycleStats is what one cycle did, for the poll_cycle log line.
+type cycleStats struct {
+	refs, ok, failed, suggested, jobs, skipped int
+	wrote                                      bool
+}
+
+// Cycle runs one poll over all services. It looks up every distinct followed
+// ref once, with at most Config.Workers lookups at a time, then records the
+// results. Writing results and starting jobs is not cut short by ctx ending
+// during the lookups, so a job that was marked is always started.
+func (p *Poller) Cycle(ctx context.Context) {
+	began := p.now()
+	svcs, err := p.store.ListServices(ctx)
+	if err != nil {
+		p.log.Error("poll_list_failed", "error", err)
+		return
+	}
+	var stats cycleStats
+	byRef := map[string][]model.Service{}
+	var refs []string
+	for _, svc := range svcs {
+		if svc.RemovedAt != nil {
+			continue
+		}
+		if SkipReason(svc) != "" {
+			stats.skipped++
+			continue
+		}
+		ref := jobs.FollowedRef(svc.ActualRef)
+		if _, seen := byRef[ref]; !seen {
+			refs = append(refs, ref)
+		}
+		byRef[ref] = append(byRef[ref], svc)
+	}
+	stats.refs = len(refs)
+
+	write := context.WithoutCancel(ctx)
+	for _, l := range p.lookupAll(ctx, refs) {
+		if l.err != nil && ctx.Err() != nil {
+			continue // cancelled mid-lookup: not the registry's fault
+		}
+		p.record(write, l, byRef[l.ref], &stats)
+	}
+	if stats.wrote {
+		p.pub.Publish(events.Services)
+	}
+	p.log.Info("poll_cycle", "refs", stats.refs, "ok", stats.ok, "failed", stats.failed,
+		"suggested", stats.suggested, "jobs", stats.jobs, "skipped", stats.skipped,
+		"duration_ms", p.now().Sub(began).Milliseconds())
+}
+
+// lookupAll asks the registry for every ref, Config.Workers at a time. It
+// returns the answers in ref order and stops starting new lookups once ctx
+// is done; refs never looked up are missing from the result.
+func (p *Poller) lookupAll(ctx context.Context, refs []string) []lookup {
+	results := make([]*lookup, len(refs))
+	next := make(chan int)
+	var wg sync.WaitGroup
+	for w := 0; w < p.cfg.Workers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := range next {
+				lctx, cancel := context.WithTimeout(ctx, lookupTimeout)
+				d, err := p.reg.Digest(lctx, refs[i])
+				cancel()
+				results[i] = &lookup{ref: refs[i], digest: d, err: err}
+			}
+		}()
+	}
+feed:
+	for i := range refs {
+		select {
+		case next <- i:
+		case <-ctx.Done():
+			break feed
+		}
+	}
+	close(next)
+	wg.Wait()
+	var out []lookup
+	for _, r := range results {
+		if r != nil {
+			out = append(out, *r)
+		}
+	}
+	return out
+}
+
+// record stores one lookup result for the services that follow l.ref.
+func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, stats *cycleStats) {
+	if l.err != nil {
+		stats.failed++
+		msg := truncate(l.err.Error())
+		p.log.Warn("poll_failed", "ref", l.ref, "error", msg)
+		p.setPollError(ctx, svcs, msg, stats, func(s model.Service) bool { return s.LastPollError != msg })
+		return
+	}
+	stats.ok++
+	p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" })
+	for _, svc := range svcs {
+		if !p.suggest(ctx, svc, l, stats) {
+			continue
+		}
+		if autoWanted(svc, l.ref, l.digest) {
+			p.startAuto(ctx, svc.ID, l.ref, l.digest, stats)
+		}
+	}
+}
+
+// suggest marks l.digest as the available version of svc unless the service
+// already runs it or already has exactly that suggestion. A service that
+// runs l.digest also gets it as last_tag_digest. It returns false only when
+// writing the suggestion failed.
+func (p *Poller) suggest(ctx context.Context, svc model.Service, l lookup, stats *cycleStats) bool {
+	if svc.ActualDigest == l.digest && svc.LastTagDigest != l.digest {
+		// Seed last_tag_digest for a service that was not updated by a tag job
+		// (it predates the column or was updated by hand), so that a later
+		// rollback is not undone.
+		if err := p.store.SetLastTagDigest(ctx, svc.ID, l.digest); err != nil {
+			p.log.Error("poll_set_last_tag_digest_failed", "service_id", svc.ID, "error", err)
+		}
+	}
+	if svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) {
+		return true
+	}
+	err := p.store.SetAvailable(ctx, svc.ID, model.Available{Ref: l.ref, Digest: l.digest, Source: model.SourcePoll, At: p.now()})
+	if err != nil {
+		p.log.Error("poll_set_available_failed", "service_id", svc.ID, "ref", l.ref, "error", err)
+		return false
+	}
+	stats.wrote = true
+	stats.suggested++
+	return true
+}
+
+// autoWanted reports whether an update of svc to ref at digest should be
+// started automatically: the service is in auto mode, not removed, does not
+// already run that digest, was not already updated automatically to it
+// (otherwise a rollback to an older version would be undone at the next
+// cycle), and its last job did not already deploy it (desired_ref holds the
+// digest until the agent reports the new inventory).
+func autoWanted(svc model.Service, ref, digest string) bool {
+	return svc.RemovedAt == nil &&
+		svc.UpdateMode == model.ModeAuto &&
+		svc.ActualDigest != digest &&
+		svc.LastTagDigest != digest &&
+		svc.DesiredRef != ref+"@"+digest
+}
+
+// startAuto starts the update job of an auto service to ref and remembers
+// digest as services.last_tag_digest, so the same digest is not started
+// again, for instance after the user rolled back. It re-reads the service
+// first, because a cycle can run for a long time and the user may have
+// switched the service to manual meanwhile. A service that already has a
+// job or cannot be updated is not an error for the cycle, and last_tag_digest
+// is left alone so the next cycle tries again.
+func (p *Poller) startAuto(ctx context.Context, id int64, ref, digest string, stats *cycleStats) {
+	svc, err := p.store.GetService(ctx, id)
+	if err != nil {
+		p.log.Warn("poll_get_service_failed", "service_id", id, "error", err)
+		return
+	}
+	if !autoWanted(svc, ref, digest) {
+		return
+	}
+	_, err = p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref})
+	var notUpdatable *model.NotUpdatableError
+	switch {
+	case err == nil:
+		stats.jobs++
+		if err := p.store.SetLastTagDigest(ctx, id, digest); err != nil {
+			p.log.Error("poll_set_last_tag_digest_failed", "service_id", id, "error", err)
+		}
+	case errors.Is(err, model.ErrJobActive), errors.As(err, &notUpdatable):
+		p.log.Debug("poll_job_not_started", "service_id", id, "ref", ref, "reason", err.Error())
+	default:
+		p.log.Warn("poll_job_failed", "service_id", id, "ref", ref, "error", err)
+	}
+}
+
+// setPollError sets msg as last_poll_error on the services for which
+// changed returns true; with none, it writes nothing.
+func (p *Poller) setPollError(ctx context.Context, svcs []model.Service, msg string, stats *cycleStats, changed func(model.Service) bool) {
+	var ids []int64
+	for _, svc := range svcs {
+		if changed(svc) {
+			ids = append(ids, svc.ID)
+		}
+	}
+	if len(ids) == 0 {
+		return
+	}
+	if err := p.store.SetPollError(ctx, ids, msg); err != nil {
+		p.log.Error("poll_set_error_failed", "error", err)
+		return
+	}
+	stats.wrote = true
+}
+
+// truncate cuts msg to maxErrorBytes without leaving a broken UTF-8
+// sequence, and drops NUL bytes, which Postgres text cannot hold.
+func truncate(msg string) string {
+	msg = strings.ReplaceAll(msg, "\x00", "")
+	if len(msg) > maxErrorBytes {
+		msg = strings.ToValidUTF8(msg[:maxErrorBytes], "")
+	}
+	return msg
+}

internal/master/poller/poller_test.go

@@ -0,0 +1,651 @@
+package poller
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/events"
+	"github.com/shcizo/package-updater/internal/master/jobs"
+	"github.com/shcizo/package-updater/internal/master/model"
+)
+
+type availableCall struct {
+	id int64
+	a  model.Available
+}
+
+type pollErrorCall struct {
+	ids []int64
+	msg string
+}
+
+type lastTagCall struct {
+	id     int64
+	digest string
+}
+
+type fakeStore struct {
+	mu         sync.Mutex
+	services   []model.Service
+	fresh      map[int64]model.Service // what GetService returns instead of the listed row
+	lastTag    []lastTagCall
+	listErr    error
+	available  []availableCall
+	pollErrors []pollErrorCall
+	ctxErrs    []error // ctx.Err() seen by every write
+}
+
+func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) {
+	return f.services, f.listErr
+}
+
+func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if s, ok := f.fresh[id]; ok {
+		return s, nil
+	}
+	for _, s := range f.services {
+		if s.ID == id {
+			return s, nil
+		}
+	}
+	return model.Service{}, model.ErrNotFound
+}
+
+// SetLastTagDigest records the call and, like the database, changes the
+// row the next cycle lists.
+func (f *fakeStore) SetLastTagDigest(_ context.Context, id int64, digest string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.lastTag = append(f.lastTag, lastTagCall{id, digest})
+	for i := range f.services {
+		if f.services[i].ID == id {
+			f.services[i].LastTagDigest = digest
+		}
+	}
+	return nil
+}
+
+func (f *fakeStore) SetAvailable(ctx context.Context, id int64, a model.Available) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.ctxErrs = append(f.ctxErrs, ctx.Err())
+	f.available = append(f.available, availableCall{id, a})
+	return nil
+}
+
+func (f *fakeStore) SetPollError(ctx context.Context, ids []int64, msg string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.ctxErrs = append(f.ctxErrs, ctx.Err())
+	f.pollErrors = append(f.pollErrors, pollErrorCall{ids, msg})
+	return nil
+}
+
+type startCall struct {
+	svcID int64
+	req   jobs.StartRequest
+}
+
+type fakeJobs struct {
+	mu      sync.Mutex
+	starts  []startCall
+	errFor  map[int64]error // by service ID
+	ctxErrs []error
+}
+
+func (f *fakeJobs) Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.ctxErrs = append(f.ctxErrs, ctx.Err())
+	f.starts = append(f.starts, startCall{svc.ID, req})
+	return model.Job{ID: int64(len(f.starts))}, f.errFor[svc.ID]
+}
+
+type fakeRegistry struct {
+	mu       sync.Mutex
+	digests  map[string]string // ref -> digest
+	errs     map[string]error  // ref -> error
+	calls    []string
+	inFlight int
+	maxSeen  int
+	delay    time.Duration
+	started  chan struct{} // receives a value whenever a lookup starts, if set
+	release  chan struct{} // lookups wait for it to be closed, if set
+	onResult func()        // called after a lookup computed its answer
+}
+
+func (f *fakeRegistry) Digest(ctx context.Context, ref string) (string, error) {
+	f.mu.Lock()
+	f.calls = append(f.calls, ref)
+	f.inFlight++
+	if f.inFlight > f.maxSeen {
+		f.maxSeen = f.inFlight
+	}
+	f.mu.Unlock()
+	defer func() {
+		f.mu.Lock()
+		f.inFlight--
+		f.mu.Unlock()
+	}()
+
+	if f.started != nil {
+		f.started <- struct{}{}
+	}
+	if f.release != nil {
+		select {
+		case <-f.release:
+		case <-ctx.Done():
+			return "", ctx.Err()
+		}
+	}
+	if f.delay > 0 {
+		time.Sleep(f.delay)
+	}
+	if f.onResult != nil {
+		f.onResult()
+	}
+	if err := f.errs[ref]; err != nil {
+		return "", err
+	}
+	return f.digests[ref], nil
+}
+
+func (f *fakeRegistry) sortedCalls() []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	out := append([]string(nil), f.calls...)
+	sort.Strings(out)
+	return out
+}
+
+type fakePub struct {
+	mu     sync.Mutex
+	topics []events.Topic
+}
+
+func (p *fakePub) Publish(t events.Topic) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.topics = append(p.topics, t)
+}
+
+const (
+	repoA = "gitea.shcizo.se/shcizo/slusk"
+	repoB = "gitea/gitea"
+)
+
+// svc builds a polled service whose running version has digest sha256:old.
+func svc(id int64, actualRef string) model.Service {
+	return model.Service{ID: id, Kind: "compose", ActualRef: actualRef, ActualDigest: "sha256:old", UpdateMode: model.ModeManual}
+}
+
+func newPoller(st *fakeStore, j *fakeJobs, reg *fakeRegistry, cfg Config) (*Poller, *fakePub) {
+	pub := &fakePub{}
+	if cfg.Workers == 0 {
+		cfg.Workers = 4
+	}
+	p := New(st, j, reg, pub, cfg, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	p.now = func() time.Time { return time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) }
+	return p, pub
+}
+
+func TestSkipReason(t *testing.T) {
+	tests := []struct {
+		name string
+		svc  model.Service
+		want string
+	}{
+		{"tag and digest", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a"}, ""},
+		{"bare image follows latest", model.Service{ActualRef: "r", ActualDigest: "sha256:a"}, ""},
+		{"digest only follows no tag", model.Service{ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, model.ReasonNoTag},
+		{"built", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Built: true}, "built from source"},
+		{"no digest", model.Service{ActualRef: "r:1"}, "no registry digest"},
+		{"self and container are polled", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Self: true, Kind: "container"}, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, SkipReason(tt.svc))
+		})
+	}
+}
+
+func TestOneHeadPerDistinctRef(t *testing.T) {
+	st := &fakeStore{services: []model.Service{
+		svc(1, repoA+":1.4"),
+		svc(2, repoA+":1.4@sha256:old"),
+		svc(3, repoB),
+		svc(4, repoB+":latest"),
+		svc(5, repoB+":1.27"),
+	}}
+	reg := &fakeRegistry{digests: map[string]string{}}
+	p, _ := newPoller(st, &fakeJobs{}, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Equal(t, []string{repoA + ":1.4", repoB + ":1.27", repoB + ":latest"}, reg.sortedCalls())
+}
+
+func TestSkipped(t *testing.T) {
+	removed := svc(4, "r:1")
+	removed.RemovedAt = ptr(time.Now())
+	st := &fakeStore{services: []model.Service{
+		{ID: 1, ActualRef: "r@sha256:a", ActualDigest: "sha256:a"},
+		{ID: 2, ActualRef: "r:1", ActualDigest: "sha256:a", Built: true},
+		{ID: 3, ActualRef: "r:1"},
+		removed,
+	}}
+	reg := &fakeRegistry{}
+	j := &fakeJobs{}
+	p, pub := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Empty(t, reg.calls, "no lookups")
+	assert.Empty(t, st.available)
+	assert.Empty(t, st.pollErrors)
+	assert.Empty(t, j.starts)
+	assert.Empty(t, pub.topics)
+}
+
+func TestNewDigest(t *testing.T) {
+	ref := repoA + ":1.4"
+	tests := []struct {
+		name      string
+		mode      string
+		wantStart bool
+	}{
+		{"auto gets one poll job to repo:tag", model.ModeAuto, true},
+		{"manual is only marked", model.ModeManual, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := svc(1, ref+"@sha256:old")
+			s.UpdateMode = tt.mode
+			st := &fakeStore{services: []model.Service{s}}
+			reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+			j := &fakeJobs{}
+			p, pub := newPoller(st, j, reg, Config{})
+
+			p.Cycle(context.Background())
+
+			require.Len(t, st.available, 1)
+			assert.Equal(t, int64(1), st.available[0].id)
+			assert.Equal(t, model.Available{Ref: ref, Digest: "sha256:new", Source: model.SourcePoll,
+				At: time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC)}, st.available[0].a)
+			if tt.wantStart {
+				assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts)
+				assert.Equal(t, []lastTagCall{{1, "sha256:new"}}, st.lastTag)
+			} else {
+				assert.Empty(t, j.starts)
+				assert.Empty(t, st.lastTag)
+			}
+			assert.Equal(t, []events.Topic{events.Services}, pub.topics)
+		})
+	}
+}
+
+func TestSameSuggestionIsIdempotent(t *testing.T) {
+	ref := repoA + ":1.4"
+	running := svc(1, ref+"@sha256:old")
+	running.UpdateMode = model.ModeAuto
+	running.ActualDigest = "sha256:new" // already on the registry's digest
+	suggested := svc(2, ref)
+	suggested.UpdateMode = model.ModeAuto
+	suggested.AvailableRef, suggested.AvailableDigest = ref, "sha256:new"
+	suggested.LastTagDigest = "sha256:new" // its job was started in an earlier cycle
+	staleSuggestion := svc(3, ref)
+	staleSuggestion.AvailableRef, staleSuggestion.AvailableDigest = ref, "sha256:older"
+	st := &fakeStore{services: []model.Service{running, suggested, staleSuggestion}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	require.Len(t, st.available, 1, "only the service with an older suggestion is updated")
+	assert.Equal(t, int64(3), st.available[0].id)
+	assert.Empty(t, j.starts, "no new job for an already suggested digest")
+}
+
+func TestStartErrorsDoNotAbortCycle(t *testing.T) {
+	ref := repoA + ":1.4"
+	var svcs []model.Service
+	for id := int64(1); id <= 3; id++ {
+		s := svc(id, ref)
+		s.UpdateMode = model.ModeAuto
+		svcs = append(svcs, s)
+	}
+	st := &fakeStore{services: svcs}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{errFor: map[int64]error{
+		1: model.ErrJobActive,
+		2: &model.NotUpdatableError{Reason: model.ReasonSelf},
+		3: errors.New("database down"),
+	}}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Len(t, st.available, 3)
+	assert.Len(t, j.starts, 3)
+	assert.Empty(t, st.lastTag, "a job that was not created is retried next cycle")
+}
+
+func TestRollbackOfAutoServiceIsNotUndone(t *testing.T) {
+	ref := repoA + ":main"
+	// Updated to sha256:new automatically, then rolled back to sha256:old by hand.
+	s := svc(1, ref+"@sha256:old")
+	s.UpdateMode = model.ModeAuto
+	s.LastTagDigest = "sha256:new"
+	s.DesiredRef = ref + "@sha256:old"
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+	p.Cycle(context.Background())
+
+	assert.Empty(t, j.starts, "same registry digest: the rollback stays")
+	assert.Equal(t, "sha256:new", st.available[0].a.Digest, "the newer version is still offered")
+
+	reg.digests[ref] = "sha256:newer"
+	p.Cycle(context.Background())
+
+	assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts, "a new digest moves it again")
+	assert.Equal(t, []lastTagCall{{1, "sha256:newer"}}, st.lastTag)
+}
+
+func TestActiveJobIsRetriedNextCycle(t *testing.T) {
+	ref := repoA + ":1.4"
+	s := svc(1, ref)
+	s.UpdateMode = model.ModeAuto
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{errFor: map[int64]error{1: model.ErrJobActive}}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+	require.Len(t, j.starts, 1)
+	assert.Empty(t, st.lastTag)
+
+	j.errFor = nil
+	p.Cycle(context.Background())
+
+	assert.Len(t, j.starts, 2)
+	assert.Equal(t, []lastTagCall{{1, "sha256:new"}}, st.lastTag)
+
+	p.Cycle(context.Background())
+	assert.Len(t, j.starts, 2, "started once the job exists")
+}
+
+func TestJobThatAlreadyDeployedTheDigestIsNotStartedAgain(t *testing.T) {
+	ref := repoA + ":1.4"
+	// FinishJob set desired_ref; the agent has not reported the new digest yet.
+	s := svc(1, ref+"@sha256:old")
+	s.UpdateMode = model.ModeAuto
+	s.DesiredRef = ref + "@sha256:new"
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Empty(t, j.starts)
+}
+
+func TestServiceSwitchedToManualDuringTheCycleIsNotStarted(t *testing.T) {
+	ref := repoA + ":1.4"
+	s := svc(1, ref)
+	s.UpdateMode = model.ModeAuto
+	manual := s
+	manual.UpdateMode = model.ModeManual
+	st := &fakeStore{services: []model.Service{s}, fresh: map[int64]model.Service{1: manual}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Len(t, st.available, 1, "still suggested")
+	assert.Empty(t, j.starts)
+	assert.Empty(t, st.lastTag)
+}
+
+func TestErrorIsPerRef(t *testing.T) {
+	bad, good := repoA+":1.4", repoB+":1.27"
+	badOld := svc(1, bad)
+	badSame := svc(2, bad)
+	badSame.LastPollError = "head: 401 unauthorized"
+	goodSvc := svc(3, good)
+	st := &fakeStore{services: []model.Service{badOld, badSame, goodSvc}}
+	reg := &fakeRegistry{
+		errs:    map[string]error{bad: errors.New("head: 401 unauthorized")},
+		digests: map[string]string{good: "sha256:new"},
+	}
+	p, pub := newPoller(st, &fakeJobs{}, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Equal(t, []pollErrorCall{{[]int64{1}, "head: 401 unauthorized"}}, st.pollErrors,
+		"only the service whose stored error differs is written")
+	require.Len(t, st.available, 1, "the other ref is still processed")
+	assert.Equal(t, int64(3), st.available[0].id)
+	assert.Equal(t, []events.Topic{events.Services}, pub.topics)
+}
+
+func TestErrorTextIsTruncated(t *testing.T) {
+	ref := repoA + ":1.4"
+	st := &fakeStore{services: []model.Service{svc(1, ref)}}
+	reg := &fakeRegistry{errs: map[string]error{ref: errors.New(strings.Repeat("é", 400) + "\x00")}}
+	p, _ := newPoller(st, &fakeJobs{}, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	require.Len(t, st.pollErrors, 1)
+	msg := st.pollErrors[0].msg
+	assert.LessOrEqual(t, len(msg), maxErrorBytes)
+	assert.Equal(t, strings.Repeat("é", 250), msg, "cut on a character boundary, NUL removed")
+}
+
+func TestErrorClearedOnSuccess(t *testing.T) {
+	ref := repoA + ":1.4"
+	failing := svc(1, ref)
+	failing.LastPollError = "head: boom"
+	fine := svc(2, ref)
+	failing.ActualDigest, fine.ActualDigest = "sha256:new", "sha256:new"
+	st := &fakeStore{services: []model.Service{failing, fine}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	p, pub := newPoller(st, &fakeJobs{}, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Equal(t, []pollErrorCall{{[]int64{1}, ""}}, st.pollErrors)
+	assert.Empty(t, st.available)
+	assert.Equal(t, []events.Topic{events.Services}, pub.topics)
+}
+
+func TestNothingWrittenNothingPublished(t *testing.T) {
+	ref := repoA + ":1.4"
+	s := svc(1, ref)
+	s.ActualDigest = "sha256:new"
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	p, pub := newPoller(st, &fakeJobs{}, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Empty(t, st.pollErrors)
+	assert.Empty(t, st.available)
+	assert.Empty(t, pub.topics)
+}
+
+func TestListErrorEndsTheCycle(t *testing.T) {
+	st := &fakeStore{listErr: errors.New("db down")}
+	reg := &fakeRegistry{}
+	p, _ := newPoller(st, &fakeJobs{}, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Empty(t, reg.calls)
+}
+
+func TestConcurrencyLimit(t *testing.T) {
+	const workers = 3
+	var svcs []model.Service
+	for i := 0; i < 8; i++ {
+		svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i)))
+	}
+	st := &fakeStore{services: svcs}
+	reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 8)}
+	p, _ := newPoller(st, &fakeJobs{}, reg, Config{Workers: workers})
+	done := make(chan struct{})
+	go func() { p.Cycle(context.Background()); close(done) }()
+
+	for i := 0; i < workers; i++ {
+		select {
+		case <-reg.started:
+		case <-time.After(2 * time.Second):
+			t.Fatalf("only %d lookups started, want %d", i, workers)
+		}
+	}
+	select {
+	case <-reg.started:
+		t.Fatal("more lookups in flight than Workers")
+	case <-time.After(100 * time.Millisecond):
+	}
+	close(reg.release)
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Cycle did not finish")
+	}
+	assert.Len(t, reg.calls, 8)
+}
+
+func TestCycleStopsOnCancel(t *testing.T) {
+	var svcs []model.Service
+	for i := 0; i < 6; i++ {
+		svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i)))
+	}
+	st := &fakeStore{services: svcs}
+	reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 6)}
+	p, pub := newPoller(st, &fakeJobs{}, reg, Config{Workers: 2})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { p.Cycle(ctx); close(done) }()
+	<-reg.started
+	<-reg.started
+
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Cycle did not return after cancel")
+	}
+	assert.Len(t, reg.calls, 2, "no new lookups after cancel")
+	assert.Empty(t, st.pollErrors, "cancelled lookups are not recorded as registry errors")
+	assert.Empty(t, st.available)
+	assert.Empty(t, pub.topics)
+}
+
+func TestMarkedJobsAreStartedEvenIfCancelledDuringTheCycle(t *testing.T) {
+	ref := repoA + ":1.4"
+	s := svc(1, ref)
+	s.UpdateMode = model.ModeAuto
+	st := &fakeStore{services: []model.Service{s}}
+	ctx, cancel := context.WithCancel(context.Background())
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}, onResult: cancel}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(ctx)
+
+	require.Len(t, st.available, 1)
+	require.Len(t, j.starts, 1)
+	assert.NoError(t, st.ctxErrs[0], "writes do not see the cancelled context")
+	assert.NoError(t, j.ctxErrs[0])
+}
+
+func TestRun(t *testing.T) {
+	ref := repoA + ":1.4"
+	st := &fakeStore{services: []model.Service{svc(1, ref)}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:old"}, delay: 30 * time.Millisecond}
+	p, _ := newPoller(st, &fakeJobs{}, reg, Config{Interval: 5 * time.Millisecond, FirstDelay: 60 * time.Millisecond, Workers: 1})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	began := time.Now()
+	go func() { p.Run(ctx); close(done) }()
+
+	require.Eventually(t, func() bool {
+		reg.mu.Lock()
+		defer reg.mu.Unlock()
+		return len(reg.calls) >= 3
+	}, 5*time.Second, 5*time.Millisecond)
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+	assert.GreaterOrEqual(t, time.Since(began), 60*time.Millisecond+2*30*time.Millisecond, "first cycle waits FirstDelay")
+	assert.Equal(t, 1, reg.maxSeen, "cycles never overlap")
+}
+
+func TestRunReturnsDuringFirstDelay(t *testing.T) {
+	p, _ := newPoller(&fakeStore{}, &fakeJobs{}, &fakeRegistry{}, Config{Interval: time.Hour, FirstDelay: time.Hour})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { p.Run(ctx); close(done) }()
+
+	cancel()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Run did not return after cancel")
+	}
+}
+
+func ptr[T any](v T) *T { return &v }
+
+func TestSeedsTheTagDigestOfARunningService(t *testing.T) {
+	ref := repoA + ":main"
+	s := svc(1, ref+"@sha256:new")
+	s.UpdateMode = model.ModeAuto
+	s.ActualDigest = "sha256:new" // updated outside the updater: last_tag_digest is empty
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Equal(t, []lastTagCall{{1, "sha256:new"}}, st.lastTag)
+	assert.Empty(t, j.starts)
+
+	// Rolled back by hand to the older digest.
+	st.services[0].ActualRef, st.services[0].ActualDigest = ref+"@sha256:old", "sha256:old"
+	st.services[0].DesiredRef = ref + "@sha256:old"
+	p.Cycle(context.Background())
+
+	assert.Empty(t, j.starts, "the rollback stays")
+	assert.Len(t, st.lastTag, 1, "nothing more to seed")
+}

internal/master/registry/registry.go

@@ -0,0 +1,114 @@
+// Package registry looks up the digest a tag points at in a container
+// registry, using HEAD requests only.
+package registry
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/url"
+	"regexp"
+	"strings"
+
+	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/google/go-containerregistry/pkg/name"
+	"github.com/google/go-containerregistry/pkg/v1/remote"
+	"github.com/google/go-containerregistry/pkg/v1/remote/transport"
+
+	"github.com/shcizo/package-updater/internal/imageref"
+)
+
+// Client looks up tag digests with HEAD requests only. Docker Hub counts a
+// GET of a manifest as a pull against the rate limit and a HEAD as nothing,
+// so Client never issues a manifest GET.
+type Client struct {
+	keychain  authn.Keychain
+	userAgent string
+	log       *slog.Logger
+}
+
+// New returns a Client that finds credentials the way the Docker CLI does:
+// in $DOCKER_CONFIG/config.json, else ~/.docker/config.json. Registries
+// without an entry are accessed anonymously. userAgent is added to the
+// User-Agent header of every request. log receives the full error of a
+// failed lookup, which Digest itself returns only in a sanitised form.
+func New(userAgent string, log *slog.Logger) *Client {
+	return &Client{keychain: authn.DefaultKeychain, userAgent: userAgent, log: log}
+}
+
+// Digest returns the digest that "repo:tag" points at, as the registry
+// reports it: for a multi-arch image that is the digest of the index, the
+// same one `docker inspect` shows in RepoDigests. A digest suffix in ref is
+// ignored. It sends HEAD /v2/<repo>/manifests/<tag> and nothing else to the
+// manifest endpoint; it does not fall back to GET when the registry omits
+// the digest header.
+func (c *Client) Digest(ctx context.Context, ref string) (string, error) {
+	p := imageref.Parse(ref)
+	if p.Tag == "" {
+		return "", fmt.Errorf("image %q has no tag", ref)
+	}
+	r, err := name.ParseReference(p.Repo + ":" + p.Tag)
+	if err != nil {
+		return "", fmt.Errorf("parse %s: %w", ref, err)
+	}
+	// Credentials are resolved here, not by ggcr, so that a broken credential
+	// helper keeps its own message: that text comes from the local config.
+	auth, err := c.keychain.Resolve(r.Context())
+	if err != nil {
+		return "", fmt.Errorf("lookup %s: %w", r, err)
+	}
+	desc, err := remote.Head(r,
+		remote.WithContext(ctx),
+		remote.WithAuth(auth),
+		remote.WithUserAgent(c.userAgent))
+	if err != nil {
+		var terr *transport.Error
+		if errors.As(err, &terr) {
+			return "", fmt.Errorf("lookup %s: %s", r, describe(terr))
+		}
+		c.log.Warn("registry_lookup_failed", "ref", ref, "error", err)
+		return "", fmt.Errorf("lookup %s: %s", r, classify(err))
+	}
+	return desc.Digest.String(), nil
+}
+
+// classify returns a fixed text for an error that is not a registry answer.
+// ggcr copies registry-controlled text (token responses, authentication
+// challenges, redirect URLs, certificate names) into such errors, so none of
+// it is passed on.
+func classify(err error) string {
+	var netErr net.Error
+	var urlErr *url.Error
+	switch {
+	case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled):
+		return "timeout"
+	case errors.As(err, &netErr), errors.As(err, &urlErr):
+		return "cannot reach registry"
+	}
+	return "lookup failed"
+}
+
+// errorCodePattern matches the registry error codes of the distribution
+// spec ("MANIFEST_UNKNOWN"). Anything else is dropped from error text.
+var errorCodePattern = regexp.MustCompile(`^[A-Z_]{1,40}$`)
+
+// describe returns the HTTP status and the structured error codes of a
+// registry error, without the messages, details or raw body that ggcr copies
+// into its error text. The registry host comes from an image name, so
+// whatever it answers must not end up in the UI.
+func describe(e *transport.Error) string {
+	msg := fmt.Sprintf("registry answered %d %s", e.StatusCode, http.StatusText(e.StatusCode))
+	var codes []string
+	for _, d := range e.Errors {
+		if errorCodePattern.MatchString(string(d.Code)) {
+			codes = append(codes, string(d.Code))
+		}
+	}
+	if len(codes) > 0 {
+		msg += " (" + strings.Join(codes, ", ") + ")"
+	}
+	return msg
+}

internal/master/registry/registry_test.go

@@ -0,0 +1,336 @@
+package registry
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/google/go-containerregistry/pkg/name"
+	ggcrregistry "github.com/google/go-containerregistry/pkg/registry"
+	"github.com/google/go-containerregistry/pkg/v1/random"
+	"github.com/google/go-containerregistry/pkg/v1/remote"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var discardLog = slog.New(slog.NewTextHandler(io.Discard, nil))
+
+func newClient() *Client { return New("test", discardLog) }
+
+// request is one request the test registry received.
+type request struct{ method, path string }
+
+// testRegistry is an in-memory registry that records every request.
+type testRegistry struct {
+	srv  *httptest.Server
+	host string // "127.0.0.1:port", the registry part of a ref
+
+	mu       sync.Mutex
+	requests []request
+}
+
+// newTestRegistry starts a registry. When user is set every request needs
+// those basic-auth credentials.
+func newTestRegistry(t *testing.T, user, pass string) *testRegistry {
+	t.Helper()
+	backend := ggcrregistry.New(ggcrregistry.Logger(log.New(io.Discard, "", 0)))
+	tr := &testRegistry{}
+	tr.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		tr.mu.Lock()
+		tr.requests = append(tr.requests, request{r.Method, r.URL.Path})
+		tr.mu.Unlock()
+		if user != "" {
+			if u, p, ok := r.BasicAuth(); !ok || u != user || p != pass {
+				w.Header().Set("WWW-Authenticate", `Basic realm="test"`)
+				w.WriteHeader(http.StatusUnauthorized)
+				return
+			}
+		}
+		backend.ServeHTTP(w, r)
+	}))
+	t.Cleanup(tr.srv.Close)
+	tr.host = strings.TrimPrefix(tr.srv.URL, "http://")
+	return tr
+}
+
+// pushIndex stores a random multi-arch index as repo:tag and returns its
+// digest. user and pass are used when the registry requires them.
+func (tr *testRegistry) pushIndex(t *testing.T, repo, tag string, opts ...remote.Option) string {
+	t.Helper()
+	idx, err := random.Index(64, 1, 2)
+	require.NoError(t, err)
+	ref, err := name.ParseReference(tr.host + "/" + repo + ":" + tag)
+	require.NoError(t, err)
+	require.NoError(t, remote.WriteIndex(ref, idx, opts...))
+	d, err := idx.Digest()
+	require.NoError(t, err)
+	return d.String()
+}
+
+// manifestRequests returns the requests that touched a manifest.
+func (tr *testRegistry) manifestRequests() []request {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	var out []request
+	for _, r := range tr.requests {
+		if strings.Contains(r.path, "/manifests/") {
+			out = append(out, r)
+		}
+	}
+	return out
+}
+
+// reset forgets the requests received so far.
+func (tr *testRegistry) reset() {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	tr.requests = nil
+}
+
+// dockerConfig points DOCKER_CONFIG at a fresh directory holding config.json
+// with the given content; "" leaves the directory empty.
+func dockerConfig(t *testing.T, configJSON string) {
+	t.Helper()
+	dir := t.TempDir()
+	if configJSON != "" {
+		require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte(configJSON), 0o600))
+	}
+	t.Setenv("DOCKER_CONFIG", dir)
+}
+
+func TestDigestOfIndexIsIndexDigest(t *testing.T) {
+	dockerConfig(t, "")
+	reg := newTestRegistry(t, "", "")
+	want := reg.pushIndex(t, "team/app", "1.2")
+
+	got, err := newClient().Digest(context.Background(), reg.host+"/team/app:1.2")
+
+	require.NoError(t, err)
+	assert.Equal(t, want, got)
+}
+
+func TestDigestIgnoresDigestSuffix(t *testing.T) {
+	dockerConfig(t, "")
+	reg := newTestRegistry(t, "", "")
+	want := reg.pushIndex(t, "team/app", "1.2")
+
+	got, err := newClient().Digest(context.Background(), reg.host+"/team/app:1.2@sha256:"+strings.Repeat("0", 64))
+
+	require.NoError(t, err)
+	assert.Equal(t, want, got)
+}
+
+func TestDigestUsesHeadOnly(t *testing.T) {
+	dockerConfig(t, "")
+	reg := newTestRegistry(t, "", "")
+	reg.pushIndex(t, "team/app", "1.2")
+	reg.reset()
+
+	_, err := newClient().Digest(context.Background(), reg.host+"/team/app:1.2")
+
+	require.NoError(t, err)
+	assert.Equal(t, []request{{http.MethodHead, "/v2/team/app/manifests/1.2"}}, reg.manifestRequests())
+	for _, r := range reg.requests {
+		if !strings.Contains(r.path, "/manifests/") {
+			assert.Equal(t, request{http.MethodGet, "/v2/"}, r, "the only other request is the API ping")
+		}
+	}
+}
+
+func TestDigestSendsUserAgent(t *testing.T) {
+	dockerConfig(t, "")
+	var ua string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ua = r.Header.Get("User-Agent")
+		w.WriteHeader(http.StatusNotFound)
+	}))
+	t.Cleanup(srv.Close)
+
+	_, _ = New("package-updater-test/1", discardLog).Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1")
+
+	assert.Contains(t, ua, "package-updater-test/1")
+}
+
+func TestCredentialsFromDockerConfig(t *testing.T) {
+	reg := newTestRegistry(t, "bot", "s3cret")
+	want := reg.pushIndex(t, "team/app", "1.2", remote.WithAuth(&authn.Basic{Username: "bot", Password: "s3cret"}))
+	auth := base64.StdEncoding.EncodeToString([]byte("bot:s3cret"))
+	dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`)
+
+	got, err := newClient().Digest(context.Background(), reg.host+"/team/app:1.2")
+
+	require.NoError(t, err)
+	assert.Equal(t, want, got)
+}
+
+func TestWrongCredentialsFail(t *testing.T) {
+	reg := newTestRegistry(t, "bot", "s3cret")
+	reg.pushIndex(t, "team/app", "1.2", remote.WithAuth(&authn.Basic{Username: "bot", Password: "s3cret"}))
+	auth := base64.StdEncoding.EncodeToString([]byte("bot:wrong"))
+	dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`)
+
+	_, err := newClient().Digest(context.Background(), reg.host+"/team/app:1.2")
+
+	require.Error(t, err)
+	assert.NotContains(t, err.Error(), "wrong", "credentials do not leak into errors")
+	assert.NotContains(t, err.Error(), auth)
+}
+
+func TestAnonymousWithoutConfig(t *testing.T) {
+	dockerConfig(t, "")
+	open := newTestRegistry(t, "", "")
+	want := open.pushIndex(t, "team/app", "1.2")
+	protected := newTestRegistry(t, "bot", "s3cret")
+
+	got, err := newClient().Digest(context.Background(), open.host+"/team/app:1.2")
+	require.NoError(t, err)
+	assert.Equal(t, want, got, "anonymous access works without a config file")
+
+	_, err = newClient().Digest(context.Background(), protected.host+"/team/app:1.2")
+	assert.Error(t, err, "a registry that needs credentials fails without them")
+}
+
+func TestUnknownTagIsAnError(t *testing.T) {
+	dockerConfig(t, "")
+	reg := newTestRegistry(t, "", "")
+	reg.pushIndex(t, "team/app", "1.2")
+
+	_, err := newClient().Digest(context.Background(), reg.host+"/team/app:nope")
+
+	assert.Error(t, err)
+}
+
+func TestRefWithoutTagIsAnError(t *testing.T) {
+	_, err := newClient().Digest(context.Background(), "team/app@sha256:"+strings.Repeat("a", 64))
+
+	assert.ErrorContains(t, err, "has no tag")
+}
+
+// A config.json whose credsStore names a helper that is not installed makes
+// the lookup fail for every registry, including ones that would work
+// anonymously. The master container has no credential helpers, so only
+// inline "auths" entries are supported there.
+func TestCredsStoreWithoutHelperFailsTheLookup(t *testing.T) {
+	reg := newTestRegistry(t, "", "")
+	reg.pushIndex(t, "team/app", "1.2")
+	dockerConfig(t, `{"credsStore":"no-such-helper-for-package-updater-tests"}`)
+
+	_, err := newClient().Digest(context.Background(), reg.host+"/team/app:1.2")
+
+	assert.ErrorContains(t, err, "error getting credentials")
+	assert.ErrorContains(t, err, "docker-credential-no-such-helper-for-package-updater-tests")
+}
+
+// A registry that leaves out Docker-Content-Digest must not make the client
+// retry with a manifest download, which Docker Hub counts as a pull.
+func TestMissingDigestHeaderDoesNotFallBackToDownload(t *testing.T) {
+	dockerConfig(t, "")
+	var mu sync.Mutex
+	var methods []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !strings.Contains(r.URL.Path, "/manifests/") {
+			return
+		}
+		mu.Lock()
+		methods = append(methods, r.Method)
+		mu.Unlock()
+		w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json")
+		w.Header().Set("Content-Length", "10")
+	}))
+	t.Cleanup(srv.Close)
+
+	_, err := newClient().Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1")
+
+	assert.Error(t, err)
+	assert.Equal(t, []string{http.MethodHead}, methods)
+}
+
+// ggcr copies up to 64 KiB of the registry's response into its errors. The
+// registry host comes from an image name, so that text must not reach the
+// stored poll error.
+func TestRegistryErrorBodyDoesNotLeak(t *testing.T) {
+	dockerConfig(t, "")
+	tests := []struct {
+		name string
+		body string
+		want string
+	}{
+		{"raw body", "internal-secret-body", "registry answered 400 Bad Request"},
+		{"structured errors", `{"errors":[{"code":"MANIFEST_UNKNOWN","message":"internal-secret-body","detail":"internal-secret-body"},{"code":"internal-secret-body"}]}`,
+			"registry answered 400 Bad Request (MANIFEST_UNKNOWN)"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				// HEAD answers carry no body, so the leak is in the GET /v2/ ping.
+				w.WriteHeader(http.StatusBadRequest)
+				_, _ = w.Write([]byte(tt.body))
+			}))
+			t.Cleanup(srv.Close)
+			host := strings.TrimPrefix(srv.URL, "http://")
+
+			_, err := newClient().Digest(context.Background(), host+"/a/b:1")
+
+			require.Error(t, err)
+			assert.NotContains(t, err.Error(), "internal-secret-body")
+			assert.Equal(t, "lookup "+host+"/a/b:1: "+tt.want, strings.Replace(err.Error(), "index.docker.io/", "", 1))
+		})
+	}
+}
+
+// ggcr puts the token server's answer into its error when the answer has no
+// token, and the token server is named by the registry's challenge.
+func TestNonRegistryErrorsAreFixedText(t *testing.T) {
+	dockerConfig(t, "")
+	tokenSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(`{"detail":"internal-secret-body"}`))
+	}))
+	t.Cleanup(tokenSrv.Close)
+	reg := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("WWW-Authenticate", `Bearer realm="`+tokenSrv.URL+`/internal-secret-realm",service="internal-secret-service"`)
+		w.WriteHeader(http.StatusUnauthorized)
+	}))
+	t.Cleanup(reg.Close)
+	host := strings.TrimPrefix(reg.URL, "http://")
+
+	_, err := newClient().Digest(context.Background(), host+"/a/b:1")
+
+	require.Error(t, err)
+	assert.NotContains(t, err.Error(), "internal-secret")
+	assert.Equal(t, "lookup "+host+"/a/b:1: lookup failed", err.Error())
+}
+
+func TestClassify(t *testing.T) {
+	refused := &url.Error{Op: "Get", URL: "http://internal-secret-host/v2/", Err: errors.New("connection refused")}
+	tests := []struct {
+		name string
+		err  error
+		want string
+	}{
+		{"deadline", fmt.Errorf("x: %w", context.DeadlineExceeded), "timeout"},
+		{"canceled", context.Canceled, "timeout"},
+		{"url error", refused, "cannot reach registry"},
+		{"net error", &net.DNSError{Name: "internal-secret-host"}, "cannot reach registry"},
+		{"deadline inside url error", &url.Error{Op: "Get", URL: "http://x", Err: context.DeadlineExceeded}, "timeout"},
+		{"anything else", errors.New("internal-secret"), "lookup failed"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, classify(tt.err))
+		})
+	}
+}

internal/master/store/jobs.go

@@ -189,9 +189,11 @@ func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk str
 // started_at (the agent's, or the existing one if the agent sent none),
 // finished_at; to_ref = f.Ref when set. On success it also sets the
 // service's desired_ref = f.Ref and clears its available ref when that is
-// the ref the job was created for (ignoring any @digest) and was reported no
-// later than the job was created (that version is now running), in the same
-// transaction. Returns false if no
+// the ref the job was created for and was reported no later than the job was
+// created (that version is now running), in the same transaction. A job
+// created for "repo:tag@digest" clears only a suggestion with that same
+// digest, so rolling back to an older digest keeps a newer suggestion (and a
+// suggestion without a digest, such as a webhook's). Returns false if no
 // such active job belongs to agentID.
 func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error) {
 	tx, err := s.pool.Begin(ctx)
@@ -226,13 +228,21 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job
 		return false, fmt.Errorf("finish job: %w", err)
 	}
 	if f.Status == model.JobSucceeded {
-		requestedRef, _, _ = strings.Cut(requestedRef, "@")
+		requestedRef, requestedDigest, _ := strings.Cut(requestedRef, "@")
+		// A plain tag job, whoever started it, deployed the tag at the digest
+		// the agent pinned; remember it. Pinned and tagless picks do not.
+		tagDigest := ""
+		if _, d, _ := strings.Cut(f.Ref, "@"); requestedDigest == "" && digestPattern.MatchString(d) {
+			tagDigest = d
+		}
 		if _, err := tx.Exec(ctx, `
 			UPDATE services SET desired_ref = $2,
-				available_ref    = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_ref END,
-				available_source = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_source END,
-				available_at     = CASE WHEN available_ref = $4 AND available_at <= $3 THEN NULL ELSE available_at END
-			WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef); err != nil {
+				last_tag_digest  = CASE WHEN $6 <> '' THEN $6 ELSE last_tag_digest END,
+				available_ref    = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_ref END,
+				available_digest = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_digest END,
+				available_source = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_source END,
+				available_at     = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN NULL ELSE available_at END
+			WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef, requestedDigest, tagDigest); err != nil {
 			return false, fmt.Errorf("finish job: desired ref: %w", err)
 		}
 	}

internal/master/store/migrations/0004_polling.sql

@@ -0,0 +1,20 @@
+ALTER TABLE services DROP CONSTRAINT services_available_source_check;
+ALTER TABLE services ADD CONSTRAINT services_available_source_check
+    CHECK (available_source IN ('', 'webhook', 'poll'));
+ALTER TABLE services ADD COLUMN available_digest text NOT NULL DEFAULT '';
+ALTER TABLE services ADD COLUMN last_poll_error  text NOT NULL DEFAULT '';
+ALTER TABLE services ADD COLUMN last_tag_digest text NOT NULL DEFAULT '';
+
+-- Every version the master has seen. '' means "no tag" / "digest unknown";
+-- never NULL, so the unique constraint deduplicates.
+CREATE TABLE image_versions (
+    id            bigserial   PRIMARY KEY,
+    repo          text        NOT NULL,
+    tag           text        NOT NULL DEFAULT '',
+    digest        text        NOT NULL DEFAULT '',
+    source        text        NOT NULL CHECK (source IN ('webhook', 'poll', 'running')),
+    first_seen_at timestamptz NOT NULL,
+    CONSTRAINT image_versions_tag_or_digest CHECK (tag <> '' OR digest <> ''),
+    CONSTRAINT image_versions_unique UNIQUE (repo, tag, digest)
+);
+CREATE INDEX image_versions_recent ON image_versions (repo, first_seen_at DESC, id DESC);

internal/master/store/services.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"regexp"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -56,7 +57,11 @@ ON CONFLICT (location_key) DO UPDATE SET
 	updated_at    = EXCLUDED.updated_at,
 	removed_at    = NULL`
 
+// digestPattern is the form of a digest worth recording as a version.
+var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+
 func upsertService(ctx context.Context, db execer, agentID int64, clusterID string, svc protocol.Service, at time.Time) error {
+	image := imageref.Parse(svc.Image)
 	clusterCol := ""
 	if svc.Kind == protocol.KindSwarm {
 		clusterCol = clusterID
@@ -65,11 +70,34 @@ func upsertService(ctx context.Context, db execer, agentID int64, clusterID stri
 		string(svc.Kind), model.LocationKey(agentID, clusterID, svc.Key()), agentID,
 		svc.ComposeProject, svc.ComposeService, svc.WorkingDir, svc.ConfigFiles, svc.Built,
 		clusterCol, svc.SwarmService, svc.ContainerName,
-		imageref.Parse(svc.Image).Repo, svc.Image, svc.Digest,
+		image.Repo, svc.Image, svc.Digest,
 		svc.Status, svc.Self, at)
 	if err != nil {
 		return fmt.Errorf("upsert service %s: %w", svc.Key(), err)
 	}
+	if digestPattern.MatchString(svc.Digest) {
+		// A bare image runs "latest", as jobs.FollowedRef sees it, so it
+		// shares its row with the polled one. repo@digest has no tag.
+		if image.Tag == "" && image.Digest == "" {
+			image.Tag = "latest"
+		}
+		if err := insertVersion(ctx, db, image.Repo, image.Tag, svc.Digest, model.SourceRunning, at); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// insertVersion records a version in image_versions unless the same
+// (repo, tag, digest) is already there; the first sighting keeps its source
+// and first_seen_at.
+func insertVersion(ctx context.Context, db execer, repo, tag, digest, source string, at time.Time) error {
+	if _, err := db.Exec(ctx, `
+		INSERT INTO image_versions (repo, tag, digest, source, first_seen_at)
+		VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING`,
+		repo, tag, digest, source, at); err != nil {
+		return fmt.Errorf("record version %s: %w", imageref.Ref{Repo: repo, Tag: tag, Digest: digest}, err)
+	}
 	return nil
 }
 
@@ -149,8 +177,8 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age
 const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''),
 	s.compose_project, s.compose_service, s.working_dir, s.config_files, s.built, s.swarm_cluster_id,
 	s.swarm_service, s.container_name, s.image_repo, s.desired_ref, s.actual_ref, s.actual_digest,
-	s.status, s.self, s.update_mode, s.available_ref, s.available_source, s.available_at,
-	s.first_seen_at, s.updated_at, s.removed_at`
+	s.status, s.self, s.update_mode, s.available_ref, s.available_digest, s.available_source, s.available_at,
+	s.last_poll_error, s.last_tag_digest, s.first_seen_at, s.updated_at, s.removed_at`
 
 func scanService(row pgx.Row) (model.Service, error) {
 	var m model.Service
@@ -158,8 +186,8 @@ func scanService(row pgx.Row) (model.Service, error) {
 	err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName,
 		&m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.Built, &m.SwarmClusterID,
 		&m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.DesiredRef, &m.ActualRef, &m.ActualDigest,
-		&m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableSource, &m.AvailableAt,
-		&m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
+		&m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableDigest, &m.AvailableSource, &m.AvailableAt,
+		&m.LastPollError, &m.LastTagDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
 	m.Kind = protocol.Kind(kind)
 	return m, err
 }
@@ -198,16 +226,96 @@ func (s *Store) queryServices(ctx context.Context, what, extraWhere string, args
 }
 
 // SetAvailable records a newer ref for the service, replacing any earlier
-// suggestion.
+// suggestion, and records the version in image_versions in the same
+// transaction (the tag is parsed from a.Ref).
 func (s *Store) SetAvailable(ctx context.Context, id int64, a model.Available) error {
-	if _, err := s.pool.Exec(ctx, `
-		UPDATE services SET available_ref = $2, available_source = $3, available_at = $4
-		WHERE id = $1`, id, a.Ref, a.Source, a.At); err != nil {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("set available: begin: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	if _, err := tx.Exec(ctx, `
+		UPDATE services SET available_ref = $2, available_digest = $3, available_source = $4, available_at = $5
+		WHERE id = $1`, id, a.Ref, a.Digest, a.Source, a.At); err != nil {
 		return fmt.Errorf("set available: %w", err)
 	}
+	ref := imageref.Parse(a.Ref)
+	if err := insertVersion(ctx, tx, ref.Repo, ref.Tag, a.Digest, a.Source, a.At); err != nil {
+		return err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("set available: commit: %w", err)
+	}
 	return nil
 }
 
+// SetPollError sets last_poll_error on the given services; msg "" clears it.
+func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error {
+	if len(ids) == 0 {
+		return nil
+	}
+	if _, err := s.pool.Exec(ctx,
+		`UPDATE services SET last_poll_error = $2 WHERE id = ANY($1)`, ids, msg); err != nil {
+		return fmt.Errorf("set poll error: %w", err)
+	}
+	return nil
+}
+
+// SetLastTagDigest sets last_tag_digest; see model.Service.LastTagDigest. The
+// poller calls it, and FinishJob sets the same column after a tag job.
+// Inventory writes never change it.
+func (s *Store) SetLastTagDigest(ctx context.Context, id int64, digest string) error {
+	if _, err := s.pool.Exec(ctx,
+		`UPDATE services SET last_tag_digest = $2 WHERE id = $1`, id, digest); err != nil {
+		return fmt.Errorf("set last tag digest: %w", err)
+	}
+	return nil
+}
+
+// RecentVersions returns up to limit versions of repo, most recently first
+// seen first. A version without a digest is left out when a version with the
+// same tag and a digest was first seen at or after it.
+func (s *Store) RecentVersions(ctx context.Context, repo string, limit int) ([]model.ImageVersion, error) {
+	rows, err := s.pool.Query(ctx, `
+		SELECT id, repo, tag, digest, source, first_seen_at FROM image_versions v
+		WHERE v.repo = $1
+		  AND NOT (v.digest = '' AND EXISTS (
+		        SELECT 1 FROM image_versions d
+		        WHERE d.repo = v.repo AND d.tag = v.tag AND d.digest <> ''
+		          AND d.first_seen_at >= v.first_seen_at))
+		ORDER BY v.first_seen_at DESC, v.id DESC
+		LIMIT $2`, repo, limit)
+	if err != nil {
+		return nil, fmt.Errorf("recent versions: %w", err)
+	}
+	defer rows.Close()
+	var out []model.ImageVersion
+	for rows.Next() {
+		var v model.ImageVersion
+		if err := rows.Scan(&v.ID, &v.Repo, &v.Tag, &v.Digest, &v.Source, &v.FirstSeenAt); err != nil {
+			return nil, fmt.Errorf("scan version: %w", err)
+		}
+		out = append(out, v)
+	}
+	return out, rows.Err()
+}
+
+// GetImageVersion returns one version. Returns model.ErrNotFound if the ID
+// does not exist.
+func (s *Store) GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error) {
+	var v model.ImageVersion
+	err := s.pool.QueryRow(ctx, `
+		SELECT id, repo, tag, digest, source, first_seen_at FROM image_versions WHERE id = $1`, id,
+	).Scan(&v.ID, &v.Repo, &v.Tag, &v.Digest, &v.Source, &v.FirstSeenAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return model.ImageVersion{}, model.ErrNotFound
+	}
+	if err != nil {
+		return model.ImageVersion{}, fmt.Errorf("get image version: %w", err)
+	}
+	return v, nil
+}
+
 // SetUpdateMode sets the service's update mode (model.ModeManual or
 // model.ModeAuto). Returns model.ErrNotFound if the service does not exist
 // or is marked removed.

internal/master/store/services_integration_test.go

@@ -17,9 +17,9 @@ import (
 
 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"}
+	slusk   = protocol.Service{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", WorkingDir: "/srv/slusk", Image: "gitea.shcizo.se/shcizo/slusk:1.4.2", Digest: digD1, 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"}
+	gitea   = protocol.Service{Kind: protocol.KindSwarm, SwarmService: "gitea_gitea", Image: "gitea/gitea:1.27.3@sha256:87a6", Digest: digGitea, Status: "running"}
 	worker  = protocol.Hello{Hostname: "node-2"}
 	manager = protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "c1"}
 )
@@ -53,7 +53,7 @@ func TestApplySnapshot_InsertsAsManualWithAgentName(t *testing.T) {
 	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, digD1, got[0].ActualDigest)
 	assert.Equal(t, "compose|"+itoa(id)+"|slusk|web", got[0].LocationKey)
 }
 

internal/master/store/versions_integration_test.go

@@ -0,0 +1,216 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"strings"
+	"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"
+)
+
+const sluskRepo = "gitea.shcizo.se/shcizo/slusk"
+
+// Digests of the fixture services; only well-formed ones are recorded as versions.
+var (
+	digD1    = "sha256:" + strings.Repeat("d", 63) + "1"
+	digGitea = "sha256:" + strings.Repeat("a", 63) + "1"
+	digPg    = "sha256:" + strings.Repeat("b", 63) + "1"
+)
+
+func versionsOf(t *testing.T, s *Store, repo string) []model.ImageVersion {
+	t.Helper()
+	got, err := s.RecentVersions(context.Background(), repo, 100)
+	require.NoError(t, err)
+	return got
+}
+
+func TestSetAvailable_RecordsVersion(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	_, svc := jobFixture(t, s, "a")
+
+	require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourcePoll, At: t0}))
+
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, sluskRepo+":1.5.0", got.AvailableRef)
+	assert.Equal(t, "sha256:n1", got.AvailableDigest)
+	assert.Equal(t, model.SourcePoll, got.AvailableSource)
+	vs := versionsOf(t, s, sluskRepo)
+	var found model.ImageVersion
+	for _, v := range vs {
+		if v.Digest == "sha256:n1" {
+			found = v
+		}
+	}
+	require.NotZero(t, found.ID)
+	assert.Equal(t, "1.5.0", found.Tag)
+	assert.Equal(t, model.SourcePoll, found.Source)
+
+	// Seeing the same version again keeps the first sighting.
+	require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourceWebhook, At: t0.Add(time.Hour)}))
+	again, err := s.GetImageVersion(ctx, found.ID)
+	require.NoError(t, err)
+	assert.Equal(t, model.SourcePoll, again.Source)
+	assert.True(t, t0.Equal(again.FirstSeenAt))
+	assert.Len(t, versionsOf(t, s, sluskRepo), len(vs))
+}
+
+func TestSnapshotRecordsRunningVersions(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	id := newAgent(t, s, "node-2")
+	tagless := protocol.Service{Kind: protocol.KindSwarm, SwarmService: "g_g", Image: "gitea/gitea@" + digGitea, Digest: digGitea, Status: "running"}
+	bare := protocol.Service{Kind: protocol.KindContainer, ContainerName: "pg", Image: "postgres", Digest: digPg, Status: "running"}
+	nodigest := protocol.Service{Kind: protocol.KindContainer, ContainerName: "nd", Image: "nodigest:1", Status: "running"}
+	hello := protocol.Hello{Hostname: "node-2", SwarmManager: true, SwarmClusterID: "c1"}
+	svcs := []protocol.Service{slusk, tagless, bare, nodigest}
+
+	require.NoError(t, s.ApplySnapshot(ctx, id, hello, svcs, t0))
+	require.NoError(t, s.ApplySnapshot(ctx, id, hello, svcs, t0.Add(time.Hour)))
+
+	got := versionsOf(t, s, sluskRepo)
+	require.Len(t, got, 1)
+	assert.Equal(t, "1.4.2", got[0].Tag)
+	assert.Equal(t, digD1, got[0].Digest)
+	assert.Equal(t, model.SourceRunning, got[0].Source)
+	assert.True(t, t0.Equal(got[0].FirstSeenAt), "re-snapshot does not bump first_seen_at")
+	tl := versionsOf(t, s, "gitea/gitea")
+	require.Len(t, tl, 1)
+	assert.Equal(t, "", tl[0].Tag)
+	assert.Equal(t, digGitea, tl[0].Digest)
+	b := versionsOf(t, s, "postgres")
+	require.Len(t, b, 1)
+	assert.Equal(t, "latest", b[0].Tag, "a bare image runs latest, like the followed ref")
+	assert.Empty(t, versionsOf(t, s, "nodigest"))
+}
+
+func TestSnapshotRecordsOnlyWellFormedDigests(t *testing.T) {
+	s := openTestStore(t)
+	id := newAgent(t, s, "node-2")
+	bad := protocol.Service{Kind: protocol.KindContainer, ContainerName: "bad", Image: "bad:1", Digest: "sha256:short", Status: "running"}
+	upper := protocol.Service{Kind: protocol.KindContainer, ContainerName: "up", Image: "up:1", Digest: "sha256:" + strings.Repeat("A", 64), Status: "running"}
+
+	require.NoError(t, s.ApplySnapshot(context.Background(), id, protocol.Hello{Hostname: "node-2"}, []protocol.Service{bad, upper}, t0))
+
+	assert.Empty(t, versionsOf(t, s, "bad"))
+	assert.Empty(t, versionsOf(t, s, "up"))
+}
+
+func TestRecentVersions(t *testing.T) {
+	ctx := context.Background()
+	t.Run("order, limit and repo filter", func(t *testing.T) {
+		s := openTestStore(t)
+		_, svc := jobFixture(t, s, "a")
+		require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "other/repo:x", Digest: "sha256:ox", Source: model.SourcePoll, At: t0}))
+		// 12 versions of the service's repo, one per minute.
+		for i := 0; i < 12; i++ {
+			tag := "t" + string(rune('a'+i))
+			require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":" + tag, Digest: "sha256:" + tag, Source: model.SourcePoll, At: t0.Add(time.Duration(i+1) * time.Minute)}))
+		}
+
+		got, err := s.RecentVersions(ctx, sluskRepo, 10)
+
+		require.NoError(t, err)
+		require.Len(t, got, 10)
+		assert.Equal(t, "tl", got[0].Tag, "newest first")
+		assert.Equal(t, "tc", got[9].Tag)
+		for _, v := range got {
+			assert.Equal(t, sluskRepo, v.Repo)
+		}
+	})
+
+	t.Run("digest-less row is hidden once its tag has a digest", func(t *testing.T) {
+		s := openTestStore(t)
+		_, svc := jobFixture(t, s, "a")
+		add := func(tag, digest string, at time.Time) {
+			require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":" + tag, Digest: digest, Source: model.SourceWebhook, At: at}))
+		}
+		add("main", "", t0.Add(time.Minute))
+		add("solo", "", t0.Add(2*time.Minute))
+		require.Len(t, versionsOf(t, s, sluskRepo), 3, "running 1.4.2 plus two digest-less rows")
+
+		add("main", "sha256:m1", t0.Add(3*time.Minute))
+
+		var refs []string
+		for _, v := range versionsOf(t, s, sluskRepo) {
+			refs = append(refs, v.Ref())
+		}
+		assert.Equal(t, []string{sluskRepo + ":main@sha256:m1", sluskRepo + ":solo", sluskRepo + ":1.4.2@" + digD1}, refs)
+	})
+
+	t.Run("digest-less row newer than the digest row of its tag is shown", func(t *testing.T) {
+		s := openTestStore(t)
+		_, svc := jobFixture(t, s, "a")
+		add := func(tag, digest string, at time.Time) {
+			require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":" + tag, Digest: digest, Source: model.SourceWebhook, At: at}))
+		}
+		add("main", "sha256:m1", t0.Add(time.Minute))
+		add("main", "", t0.Add(2*time.Minute))
+
+		var refs []string
+		for _, v := range versionsOf(t, s, sluskRepo) {
+			refs = append(refs, v.Ref())
+		}
+		assert.Equal(t, []string{sluskRepo + ":main", sluskRepo + ":main@sha256:m1", sluskRepo + ":1.4.2@" + digD1}, refs)
+	})
+}
+
+func TestGetImageVersion_NotFound(t *testing.T) {
+	s := openTestStore(t)
+
+	_, err := s.GetImageVersion(context.Background(), 99999)
+
+	assert.ErrorIs(t, err, model.ErrNotFound)
+}
+
+func TestSetPollError(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	agent, svc := jobFixture(t, s, "a")
+	require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourcePoll, At: t0}))
+
+	require.NoError(t, s.SetPollError(ctx, []int64{svc}, "boom"))
+	require.NoError(t, s.SetPollError(ctx, nil, "ignored"))
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, "boom", got.LastPollError)
+
+	require.NoError(t, s.UpsertService(ctx, agent, "", slusk, t0.Add(time.Hour)))
+	got, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, "boom", got.LastPollError, "inventory writes do not touch the poll error")
+	assert.Equal(t, "sha256:n1", got.AvailableDigest, "nor the suggestion")
+
+	require.NoError(t, s.SetPollError(ctx, []int64{svc}, ""))
+	got, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Empty(t, got.LastPollError)
+}
+
+func TestSetLastTagDigest(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	agent, svc := jobFixture(t, s, "a")
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Empty(t, got.LastTagDigest)
+
+	require.NoError(t, s.SetLastTagDigest(ctx, svc, "sha256:n1"))
+	require.NoError(t, s.UpsertService(ctx, agent, "", slusk, t0.Add(time.Hour)))
+
+	got, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, "sha256:n1", got.LastTagDigest, "inventory writes do not touch it")
+	list, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	require.Len(t, list, 1)
+	assert.Equal(t, "sha256:n1", list[0].LastTagDigest)
+}

internal/master/store/webhook_integration_test.go

@@ -4,6 +4,7 @@ package store
 
 import (
 	"context"
+	"strings"
 	"testing"
 	"time"
 
@@ -91,27 +92,34 @@ func TestUpsertKeepsUpdateMode(t *testing.T) {
 func TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed(t *testing.T) {
 	succeed := model.JobFinish{Status: model.JobSucceeded, Ref: "r:2@sha256:d", StartedAt: t0, FinishedAt: t0}
 	tests := []struct {
-		name         string
-		availableRef string
-		availableAt  time.Time // job is created at t0 for r:2
-		finish       model.JobFinish
-		wantCleared  bool
+		name            string
+		availableRef    string
+		availableDigest string
+		availableAt     time.Time // the job is created at t0
+		toRef           string    // the job's to_ref as created
+		finish          model.JobFinish
+		wantCleared     bool
 	}{
-		{"same ref, suggestion before the job", "r:2", t0.Add(-time.Minute), succeed, true},
-		{"same ref, suggestion at the same instant", "r:2", t0, succeed, true},
-		{"same ref, suggestion after the job was created", "r:2", t0.Add(time.Minute), succeed, false},
-		{"different ref keeps the suggestion", "r:3", t0.Add(-time.Minute), succeed, false},
-		{"failed job keeps the suggestion", "r:2", t0.Add(-time.Minute),
+		{"same ref, suggestion before the job", "r:2", "", t0.Add(-time.Minute), "r:2", succeed, true},
+		{"same ref, suggestion at the same instant", "r:2", "", t0, "r:2", succeed, true},
+		{"same ref, suggestion after the job was created", "r:2", "", t0.Add(time.Minute), "r:2", succeed, false},
+		{"different ref keeps the suggestion", "r:3", "", t0.Add(-time.Minute), "r:2", succeed, false},
+		{"failed job keeps the suggestion", "r:2", "", t0.Add(-time.Minute), "r:2",
 			model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0}, false},
+		{"tag job clears a poll suggestion with a digest", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2", succeed, true},
+		{"pick of the suggested digest clears it", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2@sha256:new", succeed, true},
+		{"rollback to an older digest keeps the newer suggestion", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2@sha256:old", succeed, false},
+		{"digest pick keeps a suggestion without digest", "r:2", "", t0.Add(-time.Minute), "r:2@sha256:old", succeed, false},
+		{"tagless pick keeps the suggestion", "r:2", "sha256:new", t0.Add(-time.Minute), "r@sha256:old", succeed, false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			s := openTestStore(t)
 			ctx := context.Background()
 			agent, svc := jobFixture(t, s, "a")
-			require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: tt.availableRef, Source: model.SourceWebhook, At: tt.availableAt}))
+			require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: tt.availableRef, Digest: tt.availableDigest, Source: model.SourcePoll, At: tt.availableAt}))
 			nj := newJob(svc, agent)
-			nj.ToRef = "r:2"
+			nj.ToRef = tt.toRef
 			j, err := s.CreateJob(ctx, nj)
 			require.NoError(t, err)
 
@@ -123,11 +131,13 @@ func TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed(t *testing.T) {
 			require.NoError(t, err)
 			if tt.wantCleared {
 				assert.Empty(t, got.AvailableRef)
+				assert.Empty(t, got.AvailableDigest)
 				assert.Empty(t, got.AvailableSource)
 				assert.Nil(t, got.AvailableAt)
 			} else {
 				assert.Equal(t, tt.availableRef, got.AvailableRef)
-				assert.Equal(t, model.SourceWebhook, got.AvailableSource)
+				assert.Equal(t, tt.availableDigest, got.AvailableDigest)
+				assert.Equal(t, model.SourcePoll, got.AvailableSource)
 				assert.NotNil(t, got.AvailableAt)
 			}
 		})
@@ -151,3 +161,48 @@ func TestFinishJob_WithoutAvailableLeavesItEmpty(t *testing.T) {
 	assert.Nil(t, got.AvailableAt)
 	assert.Equal(t, "r:1@sha256:d", got.DesiredRef)
 }
+
+func TestFinishJob_RemembersTheDeployedDigestOfATagJob(t *testing.T) {
+	good := "sha256:" + strings.Repeat("a", 64)
+	old := "sha256:" + strings.Repeat("b", 64)
+	ok := func(ref string) model.JobFinish {
+		return model.JobFinish{Status: model.JobSucceeded, Ref: ref, StartedAt: t0, FinishedAt: t0}
+	}
+	tests := []struct {
+		name    string
+		trigger model.JobTrigger
+		toRef   string
+		finish  model.JobFinish
+		want    string
+	}{
+		{"webhook tag job", model.TriggerWebhook, "r:2", ok("r:2@" + good), good},
+		{"manual tag job", model.TriggerManual, "r:2", ok("r:2@" + good), good},
+		{"poll tag job", model.TriggerPoll, "r:2", ok("r:2@" + good), good},
+		{"pinned pick leaves it", model.TriggerManual, "r:2@" + old, ok("r:2@" + old), "sha256:before"},
+		{"tagless pick leaves it", model.TriggerManual, "r@" + old, ok("r@" + old), "sha256:before"},
+		{"failed job leaves it", model.TriggerWebhook, "r:2",
+			model.JobFinish{Status: model.JobFailed, Error: "boom", Ref: "r:2@" + good, StartedAt: t0, FinishedAt: t0}, "sha256:before"},
+		{"malformed digest leaves it", model.TriggerWebhook, "r:2", ok("r:2@sha256:d"), "sha256:before"},
+		{"no pinned ref leaves it", model.TriggerWebhook, "r:2", ok(""), "sha256:before"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := openTestStore(t)
+			ctx := context.Background()
+			agent, svc := jobFixture(t, s, "a")
+			require.NoError(t, s.SetLastTagDigest(ctx, svc, "sha256:before"))
+			nj := newJob(svc, agent)
+			nj.ToRef, nj.Trigger = tt.toRef, tt.trigger
+			j, err := s.CreateJob(ctx, nj)
+			require.NoError(t, err)
+
+			done, err := s.FinishJob(ctx, agent, j.ID, tt.finish)
+			require.NoError(t, err)
+			require.True(t, done)
+
+			got, err := s.GetService(ctx, svc)
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got.LastTagDigest)
+		})
+	}
+}

internal/master/web/server.go

@@ -26,12 +26,18 @@ var assets embed.FS
 // serviceJobsLimit is how many jobs the service page lists.
 const serviceJobsLimit = 20
 
+// versionsLimit is how many image versions the service page lists.
+const versionsLimit = 10
+
 var agentNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`)
 
 // Store is the storage the UI reads and writes. *store.Store satisfies it.
 type Store interface {
 	ListServices(ctx context.Context) ([]model.Service, error)
 	GetService(ctx context.Context, id int64) (model.Service, error)
+	// RecentVersions returns up to limit versions of repo the master has
+	// seen, most recently first seen first.
+	RecentVersions(ctx context.Context, repo string, limit int) ([]model.ImageVersion, error)
 	// SetUpdateMode returns model.ErrNotFound for an unknown or removed service.
 	SetUpdateMode(ctx context.Context, id int64, mode string) error
 	GetJob(ctx context.Context, id int64) (model.Job, error)
@@ -54,6 +60,9 @@ type Jobs interface {
 	Updatable(svc model.Service) error
 	AutoAllowed(svc model.Service) error
 	Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error)
+	// TriggerVersion updates to exactly the image version versionID of the
+	// service's repo; model.ErrNotFound for an unknown service or version.
+	TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error)
 }
 
 // Broker delivers change signals to open streams and accepts signals from
@@ -63,11 +72,12 @@ type Broker interface {
 	Publish(t events.Topic)
 }
 
-// Config holds values shown in the "add agent" snippet.
+// Config holds values shown in the "add agent" snippet and on the service page.
 type Config struct {
-	AgentURL   string // what agents dial, e.g. wss://10.0.0.5:8090/agent
-	MasterKey  string // the master's fingerprint
-	AgentImage string
+	AgentURL     string // what agents dial, e.g. wss://10.0.0.5:8090/agent
+	MasterKey    string // the master's fingerprint
+	AgentImage   string
+	PollInterval time.Duration // shown as "every <interval>" on the service page
 }
 
 // Server renders the UI.
@@ -85,7 +95,7 @@ type Server struct {
 // New parses the templates. Each page is its own template set so every page
 // can define its own "content" block.
 func New(store Store, hub Hub, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) (*Server, error) {
-	funcs := template.FuncMap{"shortRef": shortRef, "duration": jobDuration}
+	funcs := template.FuncMap{"shortRef": shortRef, "duration": jobDuration, "followsNoTag": followsNoTag}
 	pages := map[string]*template.Template{}
 	for _, page := range []string{"overview.html", "agents.html", "agent_created.html", "service.html", "jobs.html", "job.html"} {
 		t, err := template.New(page).Funcs(funcs).ParseFS(assets, "templates/layout.html", "templates/"+page)
@@ -132,6 +142,9 @@ type pageData struct {
 	ConfigFiles []string
 	Reason      string // why Service cannot be updated; "" when it can
 	AutoBlocked string // why Service cannot be set to auto; "" when it can
+	Versions    []versionRow
+	Polling     string // what the service page says about registry polling
+	PollFailed  bool   // Polling is the last registry error
 	Jobs        []model.Job
 	Job         model.Job
 }
@@ -222,6 +235,12 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) {
 		return data, err
 	}
 	data.AutoBlocked = s.autoBlocked(svc)
+	versions, err := s.store.RecentVersions(r.Context(), svc.ImageRepo, versionsLimit)
+	if err != nil {
+		return data, err
+	}
+	data.Versions = versionRows(versions, svc.ActualDigest)
+	data.Polling, data.PollFailed = pollingStatus(svc, s.cfg.PollInterval)
 	data.Jobs, err = s.store.ListJobs(r.Context(), model.JobFilter{ServiceID: id, Limit: serviceJobsLimit})
 	return data, err
 }
@@ -260,7 +279,18 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	sess, _ := webauth.SessionFrom(r.Context())
-	job, err := s.jobs.Trigger(r.Context(), id, sess.Subject)
+	var job model.Job
+	var err error
+	if v := r.PostFormValue("version"); v == "" {
+		job, err = s.jobs.Trigger(r.Context(), id, sess.Subject)
+	} else {
+		versionID, perr := strconv.ParseInt(v, 10, 64)
+		if perr != nil {
+			http.Error(w, "invalid version", http.StatusBadRequest)
+			return
+		}
+		job, err = s.jobs.TriggerVersion(r.Context(), id, versionID, sess.Subject)
+	}
 	var nu *model.NotUpdatableError
 	switch {
 	case err == nil || errors.Is(err, model.ErrJobActive):

internal/master/web/server_test.go

@@ -34,6 +34,17 @@ type fakeStore struct {
 	created      map[string]string // name → token hash
 	revokedID    int64
 	revokeFailed []int64 // job IDs RevokeAgent reports as failed
+	versions     []model.ImageVersion
+}
+
+func (f *fakeStore) RecentVersions(_ context.Context, repo string, limit int) ([]model.ImageVersion, error) {
+	var out []model.ImageVersion
+	for _, v := range f.versions {
+		if v.Repo == repo && len(out) < limit {
+			out = append(out, v)
+		}
+	}
+	return out, nil
 }
 
 func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) { return f.services, nil }
@@ -128,6 +139,9 @@ type fakeJobs struct {
 
 	mu          sync.Mutex
 	requestedBy string
+	triggered   []int64 // version IDs passed to TriggerVersion
+	triggerSvc  []int64 // service IDs passed to TriggerVersion
+	plainCalls  int     // Trigger calls
 }
 
 func (f *fakeJobs) Updatable(model.Service) error { return f.updatable }
@@ -138,6 +152,16 @@ func (f *fakeJobs) Trigger(_ context.Context, _ int64, by string) (model.Job, er
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.requestedBy = by
+	f.plainCalls++
+	return f.job, f.err
+}
+
+func (f *fakeJobs) TriggerVersion(_ context.Context, serviceID, versionID int64, by string) (model.Job, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.requestedBy = by
+	f.triggerSvc = append(f.triggerSvc, serviceID)
+	f.triggered = append(f.triggered, versionID)
 	return f.job, f.err
 }
 
@@ -177,7 +201,15 @@ func newTestServer(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *event
 	return srv, store, hub, broker
 }
 
+// newTestEnv starts the UI on fake storage, with polling off.
 func newTestEnv(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *events.Broker, *fakeJobs) {
+	t.Helper()
+	return newTestEnvPolling(t, 0)
+}
+
+// newTestEnvPolling is newTestEnv with pollInterval as the polling interval
+// the service page shows.
+func newTestEnvPolling(t *testing.T, pollInterval time.Duration) (*httptest.Server, *fakeStore, *fakeHub, *events.Broker, *fakeJobs) {
 	t.Helper()
 	agentID := int64(2)
 	store := &fakeStore{
@@ -201,6 +233,7 @@ func newTestEnv(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *events.B
 	jobs := &fakeJobs{}
 	s, err := New(store, hub, jobs, broker, Config{
 		AgentURL: "wss://10.0.0.5:8090/agent", MasterKey: "sha256:abc", AgentImage: "gitea.shcizo.se/shcizo/package-updater:latest",
+		PollInterval: pollInterval,
 	}, slog.New(slog.NewTextHandler(io.Discard, nil)))
 	require.NoError(t, err)
 	mux := http.NewServeMux()

internal/master/web/static/app.css

@@ -29,5 +29,6 @@ table.details { width: auto; }
 table.details th { padding-right: 24px; }
 pre.log { white-space: pre-wrap; }
 .badge-update { color: var(--warn); border-color: var(--warn); }
+.badge-error { color: var(--bad); border-color: var(--bad); }
 form.mode { display: inline; margin-left: 8px; }
 form.mode button { font-size: 12px; padding: 0 6px; }

internal/master/web/templates/overview.html

@@ -14,7 +14,7 @@
     <tbody>
     {{range .Services}}
       <tr>
-        <td><a href="/services/{{.ID}}">{{.DisplayName}}</a>{{if .Self}} <span class="badge">managed outside</span>{{end}}{{if .AvailableRef}} <span class="badge badge-update" title="{{.AvailableRef}}">update available</span>{{end}}</td>
+        <td><a href="/services/{{.ID}}">{{.DisplayName}}</a>{{if .Self}} <span class="badge">managed outside</span>{{end}}{{if .AvailableRef}} <span class="badge badge-update" title="{{.AvailableRef}}">update available</span>{{end}}{{if followsNoTag .Service}} <span class="badge">follows no tag</span>{{end}}{{if .LastPollError}} <span class="badge badge-error" title="{{.LastPollError}}">poll failed</span>{{end}}</td>
         <td><code>{{shortRef .ActualRef}}</code></td>
         <td class="status status-{{.Status}}">{{.Status}}</td>
         <td>{{.UpdateMode}}

internal/master/web/templates/service.html

@@ -10,6 +10,7 @@
   {{if .Service.DesiredRef}}<tr><th>Desired image</th><td><code>{{shortRef .Service.DesiredRef}}</code></td></tr>{{end}}
   {{if .Service.AvailableRef}}<tr><th>Available</th><td><code>{{.Service.AvailableRef}}</code> <span class="muted">via {{.Service.AvailableSource}}{{with .Service.AvailableAt}}, {{.Format "2006-01-02 15:04"}}{{end}}</span></td></tr>{{end}}
   <tr><th>Mode</th><td>{{.Service.UpdateMode}}</td></tr>
+  <tr><th>Polling</th><td{{if .PollFailed}} class="error"{{end}}>{{.Polling}}</td></tr>
   {{if .ConfigFiles}}<tr><th>Config files</th><td>{{range .ConfigFiles}}<code>{{.}}</code><br>{{end}}</td></tr>{{end}}
 </table>
 {{if .Reason}}
@@ -31,6 +32,29 @@
   <p class="muted">{{if .AutoBlocked}}Auto unavailable: {{.AutoBlocked}}.{{else}}The webhook will update this service as soon as CI pushes a new image.{{end}}</p>
   {{end}}
 </form>
+<h2>Versions</h2>
+{{if eq .Service.UpdateMode "auto"}}<p class="warning">This service is in auto mode: a version picked here is replaced again when a newer version is announced or found by polling.</p>{{end}}
+<table>
+  <thead><tr><th>Version</th><th>Seen via</th><th>First seen</th><th></th></tr></thead>
+  <tbody>
+  {{range .Versions}}
+    <tr>
+      <td><code title="{{.Ref}}">{{shortRef .Ref}}</code>{{if .Running}} <span class="badge">running</span>{{end}}{{if not .Tag}} <span class="muted">the service will follow no tag afterwards</span>{{end}}</td>
+      <td>{{.Source}}</td>
+      <td>{{.FirstSeenAt.Format "2006-01-02 15:04"}}</td>
+      {{/* AutoBlocked is the same check that decides whether the service can be updated with an explicit ref, which is what a pick is. */}}
+      <td>{{if not $.AutoBlocked}}
+        <form method="post" action="/services/{{$.Service.ID}}/update">
+          <input type="hidden" name="csrf_token" value="{{$.CSRF}}">
+          <input type="hidden" name="version" value="{{.ID}}">
+          <button type="submit">Deploy</button>
+        </form>{{end}}</td>
+    </tr>
+  {{else}}
+    <tr><td colspan="4">No versions seen yet.</td></tr>
+  {{end}}
+  </tbody>
+</table>
 <h2>Update history</h2>
 <div id="history" hx-get="/fragments/services/{{.Service.ID}}/jobs" hx-trigger="sse:jobs" hx-swap="innerHTML">
   {{template "history" .}}

internal/master/web/versions_test.go

@@ -0,0 +1,221 @@
+package web
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+)
+
+const sluskRepo = "gitea.shcizo.se/shcizo/slusk"
+
+func TestServicePage_VersionPicker(t *testing.T) {
+	srv, store, _, _, _ := newTestEnv(t)
+	store.services[0].ImageRepo = sluskRepo
+	store.services[0].ActualDigest = "sha256:cur"
+	first := time.Date(2026, 10, 12, 9, 5, 0, 0, time.UTC)
+	store.versions = []model.ImageVersion{
+		{ID: 31, Repo: sluskRepo, Tag: "1.5.0", Digest: "sha256:" + strings.Repeat("a", 40), Source: "poll", FirstSeenAt: first},
+		{ID: 32, Repo: sluskRepo, Tag: "1.4.2", Digest: "sha256:cur", Source: "running", FirstSeenAt: first},
+		{ID: 33, Repo: sluskRepo, Digest: "sha256:" + strings.Repeat("b", 40), Source: "running", FirstSeenAt: first},
+		{ID: 34, Repo: sluskRepo, Tag: "main", Source: "webhook", FirstSeenAt: first},
+		{ID: 40, Repo: "other/repo", Tag: "x", Source: "poll", FirstSeenAt: first},
+	}
+
+	_, body := get(t, srv, "/services/1")
+
+	for _, id := range []string{"31", "32", "33", "34"} {
+		assert.Contains(t, body, `<input type="hidden" name="version" value="`+id+`">`, id)
+	}
+	assert.NotContains(t, body, `name="version" value="40"`, "versions of other repos are not listed")
+	assert.Equal(t, 4, strings.Count(body, `name="version"`))
+	assert.Contains(t, body, `name="csrf_token" value="csrf"`)
+	assert.Contains(t, body, `title="`+sluskRepo+`:1.5.0@sha256:`+strings.Repeat("a", 40)+`"`, "full ref as title")
+	assert.Contains(t, body, sluskRepo+":1.5.0@sha256:"+strings.Repeat("a", 12), "short ref as text")
+	assert.Equal(t, 1, strings.Count(body, `<span class="badge">running</span>`), "only the running digest is marked")
+	assert.Equal(t, 1, strings.Count(body, "the service will follow no tag afterwards"), "only the tagless row")
+	assert.Contains(t, body, "2026-10-12 09:05")
+	assert.NotContains(t, body, "auto mode", "no auto warning for a manual service")
+}
+
+func TestServicePage_VersionPickerShowsAtMostTenRows(t *testing.T) {
+	srv, store, _, _, _ := newTestEnv(t)
+	store.services[0].ImageRepo = sluskRepo
+	for i := int64(1); i <= 12; i++ {
+		store.versions = append(store.versions, model.ImageVersion{ID: i, Repo: sluskRepo, Tag: "t", Digest: "sha256:d", Source: "poll"})
+	}
+
+	_, body := get(t, srv, "/services/1")
+
+	assert.Equal(t, 10, strings.Count(body, `name="version"`))
+}
+
+func TestServicePage_VersionPickerWarnsForAutoServices(t *testing.T) {
+	srv, store, _, _, _ := newTestEnv(t)
+	store.services[0].UpdateMode = "auto"
+
+	_, body := get(t, srv, "/services/1")
+
+	assert.Contains(t, body, "This service is in auto mode")
+}
+
+func TestServicePage_VersionPickerHiddenWhenServiceCannotBeUpdated(t *testing.T) {
+	for _, reason := range []string{model.ReasonSelf, model.ReasonContainer, model.ReasonRemoved, model.ReasonBuilt} {
+		t.Run(reason, func(t *testing.T) {
+			srv, store, _, _, jobs := newTestEnv(t)
+			store.services[0].ImageRepo = sluskRepo
+			store.versions = []model.ImageVersion{{ID: 31, Repo: sluskRepo, Tag: "1.5.0", Source: "poll"}}
+			jobs.autoErr = &model.NotUpdatableError{Reason: reason}
+
+			_, body := get(t, srv, "/services/1")
+
+			assert.Contains(t, body, "<h2>Versions</h2>")
+			assert.Contains(t, body, "1.5.0", "the history is still shown")
+			assert.NotContains(t, body, `name="version"`)
+		})
+	}
+}
+
+func TestServicePage_VersionPickerOffersVersionsToAServiceThatFollowsNoTag(t *testing.T) {
+	srv, store, _, _, jobs := newTestEnv(t)
+	store.services[0].ImageRepo = sluskRepo
+	store.versions = []model.ImageVersion{{ID: 31, Repo: sluskRepo, Tag: "1.5.0", Digest: "sha256:a", Source: "poll"}}
+	jobs.updatable = &model.NotUpdatableError{Reason: model.ReasonNoTag} // autoErr stays nil: an explicit ref is fine
+
+	_, body := get(t, srv, "/services/1")
+
+	assert.Contains(t, body, `name="version" value="31"`)
+	assert.NotContains(t, body, `<button type="submit">Update</button>`, "the plain Update button stays hidden")
+}
+
+func TestServicePage_PollingRow(t *testing.T) {
+	tests := []struct {
+		name   string
+		edit   func(*model.Service)
+		want   string
+		failed bool
+	}{
+		{"polled", func(s *model.Service) { s.ActualDigest = "sha256:cur" }, "every 6h", false},
+		{"poll error", func(s *model.Service) { s.ActualDigest = "sha256:cur"; s.LastPollError = "head x: 401 <b>" }, "last check failed: head x: 401 &lt;b&gt;", true},
+		{"built", func(s *model.Service) { s.ActualDigest = "sha256:cur"; s.Built = true }, "not polled: built from source", false},
+		{"no digest", func(s *model.Service) {}, "not polled: no registry digest", false},
+		{"no tag", func(s *model.Service) { s.ActualRef = "r@sha256:a"; s.ActualDigest = "sha256:a" }, "not polled: follows no tag", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srv, store, _, _, _ := newTestEnvPolling(t, 6*time.Hour)
+			tt.edit(&store.services[0])
+
+			_, body := get(t, srv, "/services/1")
+
+			assert.Contains(t, body, "<th>Polling</th><td")
+			assert.Contains(t, body, tt.want)
+			assert.Equal(t, tt.failed, strings.Contains(body, `<td class="error">`+tt.want))
+		})
+	}
+}
+
+func TestShortDuration(t *testing.T) {
+	tests := []struct {
+		d    time.Duration
+		want string
+	}{
+		{6 * time.Hour, "6h"},
+		{90 * time.Minute, "1h30m"},
+		{30 * time.Minute, "30m"},
+		{90 * time.Second, "1m30s"},
+		{45 * time.Second, "45s"},
+		{6*time.Hour + 5*time.Second, "6h0m5s"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.want, func(t *testing.T) {
+			assert.Equal(t, tt.want, shortDuration(tt.d))
+		})
+	}
+}
+
+func postVersion(t *testing.T, srv *httptest.Server, version string) (*http.Response, string) {
+	t.Helper()
+	resp, err := noRedirectClient().PostForm(srv.URL+"/services/1/update", url.Values{"csrf_token": {"csrf"}, "version": {version}})
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	return resp, string(body)
+}
+
+func TestUpdateService_VersionStartsThatVersion(t *testing.T) {
+	srv, _, _, _, jobs := newTestEnv(t)
+	jobs.job = model.Job{ID: 42}
+
+	resp, _ := postVersion(t, srv, "31")
+
+	assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
+	assert.Equal(t, "/jobs/42", resp.Header.Get("Location"))
+	assert.Equal(t, []int64{31}, jobs.triggered)
+	assert.Equal(t, []int64{1}, jobs.triggerSvc)
+	assert.Equal(t, "alice", jobs.triggeredBy())
+	assert.Zero(t, jobs.plainCalls)
+}
+
+func TestUpdateService_WithoutVersionTriggersTheFollowedTag(t *testing.T) {
+	srv, _, _, _, jobs := newTestEnv(t)
+	jobs.job = model.Job{ID: 42}
+
+	resp, _ := postVersion(t, srv, "")
+
+	assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
+	assert.Equal(t, 1, jobs.plainCalls)
+	assert.Empty(t, jobs.triggered)
+}
+
+func TestUpdateService_VersionErrors(t *testing.T) {
+	tests := []struct {
+		name     string
+		version  string
+		err      error
+		wantCode int
+		wantBody string
+		wantJob  string // Location header
+	}{
+		{"not a number", "abc", nil, http.StatusBadRequest, "invalid version", ""},
+		{"a ref instead of an id", "repo:evil@sha256:x", nil, http.StatusBadRequest, "invalid version", ""},
+		{"unknown version or other repo", "77", model.ErrNotFound, http.StatusNotFound, "not found", ""},
+		{"not updatable", "31", &model.NotUpdatableError{Reason: model.ReasonBuilt}, http.StatusConflict, "Cannot update: " + model.ReasonBuilt + ".", ""},
+		{"job already active", "31", model.ErrJobActive, http.StatusSeeOther, "", "/jobs/41"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srv, _, _, _, jobs := newTestEnv(t)
+			jobs.err, jobs.job = tt.err, model.Job{ID: 41}
+
+			resp, body := postVersion(t, srv, tt.version)
+
+			assert.Equal(t, tt.wantCode, resp.StatusCode)
+			assert.Contains(t, body, tt.wantBody)
+			assert.Equal(t, tt.wantJob, resp.Header.Get("Location"))
+			if tt.wantCode == http.StatusBadRequest {
+				assert.Empty(t, jobs.triggered, "no job for a malformed version")
+			}
+		})
+	}
+}
+
+func TestOverview_ShowsFollowsNoTagAndPollFailedBadges(t *testing.T) {
+	srv, store, _, _, _ := newTestEnv(t)
+	store.services[0].LastPollError = `head x: "boom" <script>`
+
+	for _, path := range []string{"/", "/fragments/services"} {
+		_, body := get(t, srv, path)
+		assert.Equal(t, 1, strings.Count(body, "follows no tag"), "only the digest-only Swarm service: "+path)
+		assert.Contains(t, body, `class="badge badge-error" title="head x: &#34;boom&#34; &lt;script&gt;">poll failed</span>`, path)
+		assert.Equal(t, 1, strings.Count(body, "poll failed"), path)
+	}
+}

internal/master/web/views.go

@@ -5,7 +5,9 @@ import (
 	"time"
 
 	"github.com/shcizo/package-updater/internal/imageref"
+	"github.com/shcizo/package-updater/internal/master/jobs"
 	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/master/poller"
 	"github.com/shcizo/package-updater/internal/protocol"
 )
 
@@ -54,6 +56,51 @@ func shortRef(ref string) string {
 	return r.String()
 }
 
+// followsNoTag says the service's image reference has a digest but no tag,
+// so there is no tag to poll or update.
+func followsNoTag(svc model.Service) bool {
+	return jobs.FollowedRef(svc.ActualRef) == ""
+}
+
+// versionRow is an image version as shown on the service page.
+type versionRow struct {
+	model.ImageVersion
+	Running bool // the service runs exactly this digest
+}
+
+// versionRows marks the version the service runs now, by digest.
+func versionRows(versions []model.ImageVersion, actualDigest string) []versionRow {
+	rows := make([]versionRow, len(versions))
+	for i, v := range versions {
+		rows[i] = versionRow{ImageVersion: v, Running: v.Digest != "" && v.Digest == actualDigest}
+	}
+	return rows
+}
+
+// pollingStatus is the text of the "Polling" row on the service page and
+// whether it is a registry error.
+func pollingStatus(svc model.Service, interval time.Duration) (text string, failed bool) {
+	switch {
+	case poller.SkipReason(svc) != "":
+		return "not polled: " + poller.SkipReason(svc), false
+	case svc.LastPollError != "":
+		return "last check failed: " + svc.LastPollError, true
+	}
+	return "every " + shortDuration(interval), false
+}
+
+// shortDuration formats d without trailing zero units: "6h", "1h30m", "90s".
+func shortDuration(d time.Duration) string {
+	s := d.String()
+	if strings.HasSuffix(s, "m0s") {
+		s = s[:len(s)-2]
+	}
+	if strings.HasSuffix(s, "h0m") {
+		s = s[:len(s)-2]
+	}
+	return s
+}
+
 type agentRow struct {
 	model.Agent
 	Online bool

internal/protocol/protocol.go

@@ -16,11 +16,16 @@ import (
 
 // Version is the protocol version this build speaks. Agents send it in
 // Hello. Phase-1 agents do not send the field, which reads as 0.
-const Version = 2
+const Version = 3
 
 // MinJobVersion is the lowest agent protocol version the master sends jobs to.
 const MinJobVersion = 2
 
+// MinPinnedRefVersion is the lowest agent protocol version the master sends
+// jobs whose Ref has a digest or no tag to. Version-2 agents refuse a Ref
+// with a digest.
+const MinPinnedRefVersion = 3
+
 // MaxJobOutput caps the job output an agent keeps and the master stores, in
 // bytes. Only the tail is kept. With JSON escaping (worst case 6 bytes per
 // byte) a JobResult carrying this much output still fits the 1 MiB read limit.
@@ -123,8 +128,10 @@ type Remove struct {
 	Key string `json:"key"`
 }
 
-// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag";
-// the agent resolves the tag's current digest and deploys "repo:tag@digest".
+// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag",
+// "repo:tag@digest" or "repo@digest". For "repo:tag" the agent resolves the
+// tag's current digest and deploys "repo:tag@digest"; a Ref with a digest is
+// deployed as is.
 // The agent reads the Compose working directory and config files from its
 // own Docker daemon, not from this message.
 type JobStart struct {