fix(master): seed the tag digest from running services and hide registry error text (#9)

Renames last_auto_digest to last_tag_digest. FinishJob sets it to the pinned
digest after any successful job whose to_ref had no digest, so a rollback
holds however the update was started.

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

@@ -158,7 +158,7 @@ func run() error {
 	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), broker, poller.Config{
+	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{})

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.
 

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 (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.
+// 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{

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 (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.
+// 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 {

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
@@ -91,6 +95,18 @@ func (m *memStore) SetPollError(_ context.Context, ids []int64, msg string) erro
 	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.
 func (m *memStore) seedService(svc model.Service) {
 	m.mu.Lock()
@@ -229,6 +245,9 @@ 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, digest, _ := strings.Cut(requestedRef, "@")
 		if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) &&

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

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

@@ -99,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")

internal/master/hub/route.go

@@ -16,7 +16,8 @@ type candidate struct {
 
 // 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 when there is none.
+// 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))
@@ -64,6 +65,9 @@ func route(svc model.Service, conns []candidate, minProtocol int) (model.Route,
 		}
 	}
 	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

@@ -30,7 +30,7 @@ func TestRoute(t *testing.T) {
 		{"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.ErrAgentTooOld},
+		{"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},
@@ -47,7 +47,7 @@ func TestRoute(t *testing.T) {
 			[]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.ErrAgentTooOld},
+		{"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) {

internal/master/jobs/jobs.go

@@ -156,8 +156,9 @@ func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64
 }
 
 // Start starts an update of svc. It returns the job, which may already be
-// failed ("agent offline", "agent too old for updates", the latter also when
-// the ref has a digest or no tag and the agent speaks protocol version 2); 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.

internal/master/jobs/jobs_test.go

@@ -83,6 +83,9 @@ type fakeHub struct {
 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
@@ -481,7 +484,7 @@ func TestTriggerVersion_RequiresPinnedProtocol(t *testing.T) {
 	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 for updates", pinned.Error)
+	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")

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

@@ -61,9 +61,16 @@ type Service struct {
 	AvailableSource string // SourceWebhook or SourcePoll; "" when AvailableRef is empty
 	AvailableAt     *time.Time
 	LastPollError   string // last registry error for the followed ref; "" when fine
-	FirstSeenAt     time.Time
-	UpdatedAt       time.Time
-	RemovedAt       *time.Time
+	// 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.

internal/master/poller/poller.go

@@ -25,7 +25,9 @@ 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
 }
 
@@ -74,7 +76,7 @@ func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Lo
 func SkipReason(svc model.Service) string {
 	switch {
 	case jobs.FollowedRef(svc.ActualRef) == "":
-		return "follows no tag"
+		return model.ReasonNoTag
 	case svc.Built:
 		return "built from source"
 	case svc.ActualDigest == "":
@@ -208,35 +210,83 @@ func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, sta
 	stats.ok++
 	p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" })
 	for _, svc := range svcs {
-		if svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) {
+		if !p.suggest(ctx, svc, l, stats) {
 			continue
 		}
-		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)
-			continue
+		if autoWanted(svc, l.ref, l.digest) {
+			p.startAuto(ctx, svc.ID, l.ref, l.digest, stats)
 		}
-		stats.wrote = true
-		stats.suggested++
-		if svc.UpdateMode != model.ModeAuto {
-			continue
-		}
-		p.startAuto(ctx, svc, l.ref, stats)
 	}
 }
 
-// startAuto starts the update job of an auto service to ref. A service that
-// already has a job or cannot be updated is not an error for the cycle.
-func (p *Poller) startAuto(ctx context.Context, svc model.Service, ref string, stats *cycleStats) {
-	_, err := p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref})
+// 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", svc.ID, "ref", ref, "reason", err.Error())
+		p.log.Debug("poll_job_not_started", "service_id", id, "ref", ref, "reason", err.Error())
 	default:
-		p.log.Warn("poll_job_failed", "service_id", svc.ID, "ref", ref, "error", err)
+		p.log.Warn("poll_job_failed", "service_id", id, "ref", ref, "error", err)
 	}
 }
 

internal/master/poller/poller_test.go

@@ -30,9 +30,16 @@ type pollErrorCall struct {
 	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
@@ -43,6 +50,34 @@ 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()
@@ -175,7 +210,7 @@ func TestSkipReason(t *testing.T) {
 	}{
 		{"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"}, "follows no tag"},
+		{"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"}, ""},
@@ -252,8 +287,10 @@ func TestNewDigest(t *testing.T) {
 				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)
 		})
@@ -268,6 +305,7 @@ func TestSameSuggestionIsIdempotent(t *testing.T) {
 	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}}
@@ -303,6 +341,89 @@ func TestStartErrorsDoNotAbortCycle(t *testing.T) {
 
 	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) {
@@ -384,18 +505,37 @@ func TestListErrorEndsTheCycle(t *testing.T) {
 }
 
 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{delay: 20 * time.Millisecond}
-	p, _ := newPoller(st, &fakeJobs{}, reg, Config{Workers: 3})
+	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) }()
 
-	p.Cycle(context.Background())
+	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)
-	assert.Equal(t, 3, reg.maxSeen)
 }
 
 func TestCycleStopsOnCancel(t *testing.T) {
@@ -485,3 +625,27 @@ func TestRunReturnsDuringFirstDelay(t *testing.T) {
 }
 
 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

@@ -4,11 +4,19 @@ 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"
 )
@@ -19,14 +27,16 @@ import (
 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.
-func New(userAgent string) *Client {
-	return &Client{keychain: authn.DefaultKeychain, userAgent: userAgent}
+// 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
@@ -44,12 +54,61 @@ func (c *Client) Digest(ctx context.Context, ref string) (string, error) {
 	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.WithAuthFromKeychain(c.keychain),
+		remote.WithAuth(auth),
 		remote.WithUserAgent(c.userAgent))
 	if err != nil {
-		return "", fmt.Errorf("head %s: %w", r, err)
+		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

@@ -3,10 +3,15 @@ package registry
 import (
 	"context"
 	"encoding/base64"
+	"errors"
+	"fmt"
 	"io"
 	"log"
+	"log/slog"
+	"net"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"os"
 	"path/filepath"
 	"strings"
@@ -22,6 +27,10 @@ import (
 	"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 }
 
@@ -108,7 +117,7 @@ func TestDigestOfIndexIsIndexDigest(t *testing.T) {
 	reg := newTestRegistry(t, "", "")
 	want := reg.pushIndex(t, "team/app", "1.2")
 
-	got, err := New("test").Digest(context.Background(), reg.host+"/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)
@@ -119,7 +128,7 @@ func TestDigestIgnoresDigestSuffix(t *testing.T) {
 	reg := newTestRegistry(t, "", "")
 	want := reg.pushIndex(t, "team/app", "1.2")
 
-	got, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2@sha256:"+strings.Repeat("0", 64))
+	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)
@@ -131,7 +140,7 @@ func TestDigestUsesHeadOnly(t *testing.T) {
 	reg.pushIndex(t, "team/app", "1.2")
 	reg.reset()
 
-	_, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2")
+	_, 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())
@@ -151,7 +160,7 @@ func TestDigestSendsUserAgent(t *testing.T) {
 	}))
 	t.Cleanup(srv.Close)
 
-	_, _ = New("package-updater-test/1").Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1")
+	_, _ = 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")
 }
@@ -162,7 +171,7 @@ func TestCredentialsFromDockerConfig(t *testing.T) {
 	auth := base64.StdEncoding.EncodeToString([]byte("bot:s3cret"))
 	dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`)
 
-	got, err := New("test").Digest(context.Background(), reg.host+"/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)
@@ -174,7 +183,7 @@ func TestWrongCredentialsFail(t *testing.T) {
 	auth := base64.StdEncoding.EncodeToString([]byte("bot:wrong"))
 	dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`)
 
-	_, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2")
+	_, 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")
@@ -187,11 +196,11 @@ func TestAnonymousWithoutConfig(t *testing.T) {
 	want := open.pushIndex(t, "team/app", "1.2")
 	protected := newTestRegistry(t, "bot", "s3cret")
 
-	got, err := New("test").Digest(context.Background(), open.host+"/team/app:1.2")
+	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 = New("test").Digest(context.Background(), protected.host+"/team/app:1.2")
+	_, err = newClient().Digest(context.Background(), protected.host+"/team/app:1.2")
 	assert.Error(t, err, "a registry that needs credentials fails without them")
 }
 
@@ -200,13 +209,13 @@ func TestUnknownTagIsAnError(t *testing.T) {
 	reg := newTestRegistry(t, "", "")
 	reg.pushIndex(t, "team/app", "1.2")
 
-	_, err := New("test").Digest(context.Background(), reg.host+"/team/app:nope")
+	_, err := newClient().Digest(context.Background(), reg.host+"/team/app:nope")
 
 	assert.Error(t, err)
 }
 
 func TestRefWithoutTagIsAnError(t *testing.T) {
-	_, err := New("test").Digest(context.Background(), "team/app@sha256:"+strings.Repeat("a", 64))
+	_, err := newClient().Digest(context.Background(), "team/app@sha256:"+strings.Repeat("a", 64))
 
 	assert.ErrorContains(t, err, "has no tag")
 }
@@ -220,7 +229,7 @@ func TestCredsStoreWithoutHelperFailsTheLookup(t *testing.T) {
 	reg.pushIndex(t, "team/app", "1.2")
 	dockerConfig(t, `{"credsStore":"no-such-helper-for-package-updater-tests"}`)
 
-	_, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2")
+	_, 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")
@@ -244,8 +253,84 @@ func TestMissingDigestHeaderDoesNotFallBackToDownload(t *testing.T) {
 	}))
 	t.Cleanup(srv.Close)
 
-	_, err := New("test").Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1")
+	_, 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

@@ -229,13 +229,20 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job
 	}
 	if f.Status == model.JobSucceeded {
 		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,
+				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); err != nil {
+			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

@@ -3,6 +3,7 @@ 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.

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,14 +70,18 @@ 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 svc.Digest != "" {
-		ref := imageref.Parse(svc.Image)
-		if err := insertVersion(ctx, db, ref.Repo, ref.Tag, svc.Digest, model.SourceRunning, at); err != nil {
+	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
 		}
 	}
@@ -87,7 +96,7 @@ func insertVersion(ctx context.Context, db execer, repo, tag, digest, source str
 		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:%s: %w", repo, tag, err)
+		return fmt.Errorf("record version %s: %w", imageref.Ref{Repo: repo, Tag: tag, Digest: digest}, err)
 	}
 	return nil
 }
@@ -169,7 +178,7 @@ const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.nam
 	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_digest, s.available_source, s.available_at,
-	s.last_poll_error, s.first_seen_at, s.updated_at, s.removed_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
@@ -178,7 +187,7 @@ func scanService(row pgx.Row) (model.Service, error) {
 		&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.AvailableDigest, &m.AvailableSource, &m.AvailableAt,
-		&m.LastPollError, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
+		&m.LastPollError, &m.LastTagDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
 	m.Kind = protocol.Kind(kind)
 	return m, err
 }
@@ -252,16 +261,28 @@ func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error
 	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 exists.
+// 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 <> ''))
+		        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 {

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

@@ -4,6 +4,7 @@ package store
 
 import (
 	"context"
+	"strings"
 	"testing"
 	"time"
 
@@ -16,6 +17,13 @@ import (
 
 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)
@@ -59,8 +67,8 @@ 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@sha256:87a6", Digest: "sha256:87a6", Status: "running"}
-	bare := protocol.Service{Kind: protocol.KindContainer, ContainerName: "pg", Image: "postgres", Digest: "sha256:pg1", Status: "running"}
+	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}
@@ -71,19 +79,31 @@ func TestSnapshotRecordsRunningVersions(t *testing.T) {
 	got := versionsOf(t, s, sluskRepo)
 	require.Len(t, got, 1)
 	assert.Equal(t, "1.4.2", got[0].Tag)
-	assert.Equal(t, "sha256:d1", got[0].Digest)
+	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, "sha256:87a6", tl[0].Digest)
+	assert.Equal(t, digGitea, tl[0].Digest)
 	b := versionsOf(t, s, "postgres")
 	require.Len(t, b, 1)
-	assert.Equal(t, "", b[0].Tag, "a bare image is recorded without a tag, not as latest")
+	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) {
@@ -123,7 +143,23 @@ func TestRecentVersions(t *testing.T) {
 		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@sha256:d1"}, refs)
+		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)
 	})
 }
 
@@ -158,3 +194,23 @@ func TestSetPollError(t *testing.T) {
 	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"
 
@@ -160,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_test.go

@@ -201,9 +201,15 @@ func newTestServer(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *event
 	return srv, store, hub, broker
 }
 
-// newTestEnv starts the UI on fake storage. pollInterval is what the service
-// page shows as the polling interval; it defaults to zero.
-func newTestEnv(t *testing.T, pollInterval ...time.Duration) (*httptest.Server, *fakeStore, *fakeHub, *events.Broker, *fakeJobs) {
+// 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{
@@ -227,7 +233,7 @@ func newTestEnv(t *testing.T, pollInterval ...time.Duration) (*httptest.Server,
 	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: append(pollInterval, 0)[0],
+		PollInterval: pollInterval,
 	}, slog.New(slog.NewTextHandler(io.Discard, nil)))
 	require.NoError(t, err)
 	mux := http.NewServeMux()

internal/master/web/templates/service.html

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

internal/master/web/versions_test.go

@@ -111,7 +111,7 @@ func TestServicePage_PollingRow(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			srv, store, _, _, _ := newTestEnv(t, 6*time.Hour)
+			srv, store, _, _, _ := newTestEnvPolling(t, 6*time.Hour)
 			tt.edit(&store.services[0])
 
 			_, body := get(t, srv, "/services/1")