fix(master): review fixes for drift restore and accept (#10)

Adds services.awaiting_inventory, Service.Drifted/RunsRef/RunningRef, AcceptRunning, MoveCandidates, MergeSources and MergeService, and maps a job for a deleted service to ErrNotFound.

CLAUDE.md

@@ -64,6 +64,11 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
   list in `known_jobs`, and only then accepts jobs for that connection.
 - **The webhook never creates services; new services always start `manual`.** The upsert
   does not touch `update_mode`.
+- **Drift is never corrected automatically; poll/webhook jobs are refused for a drifted
+  service** (`jobs.Start`, `ReasonDrifted`). Only a manual job, `RestoreDesired` or an
+  accept gets it out.
+- **Merging is never automatic; merge is the only path that deletes a `services` row**
+  (`store.MergeService`). The old row keeps its id and takes the duplicate's location.
 - **Auth**: agents by token over pinned TLS; people by OIDC; CI by the `UPDATER_API_KEY`
   bearer token on `POST /update` only. Every state-changing UI request is a POST with a
   CSRF token.
@@ -138,6 +143,25 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
   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.
+- **Drift compares by digest, not by string** (`model.RunsRef`): `repo:tag@D` is run by
+  `repo:tag` reporting digest D. Same digest under another tag is not drift. Different
+  repo strings (`nginx` vs `docker.io/library/nginx`) are.
+- **`services.awaiting_inventory`** is true from a successful `FinishJob` until the next
+  inventory write for the row; `Service.Drifted()` is false meanwhile, since `desired_ref`
+  is ahead of `actual_ref` for a debounce interval after every job.
+- **Accept never touches `last_tag_digest`, `available_*` or `update_mode`**, and the
+  `actual` form field (`Service.RunningRef()`, ref plus digest) is only a guard against a
+  changed service; the value written comes from the row. `RestoreDesired` sets
+  `last_tag_digest` to the drifted digest when it is another digest of the desired tag and
+  `last_tag_digest` is empty or equals the desired digest (otherwise it holds back a newer
+  digest and stays).
+- **`awaiting_inventory` can hide real drift** until the next inventory change when a job
+  result arrives after the snapshot that shows the drift (an agent reconnect resends its
+  outbox after the snapshot), or when Swarm coalesces an update and a rollback into one
+  debounce interval. This is accepted.
+- **Merge lock order is agent, then service rows by id** (as in inventory writes), and the
+  duplicate row is deleted before the old row is updated, because `location_key` is unique
+  and not deferrable. A removal racing a merge can be lost; the next snapshot fixes it.
 
 ## References
 
@@ -146,6 +170,7 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/
 - 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`
+- Phase 5 plan: `docs/superpowers/plans/2026-10-13-master-agent-phase-5-drift-merge.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

@@ -10,6 +10,7 @@ Design: [master/agent spec](docs/superpowers/specs/2026-10-12-master-agent-desig
 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).
+Phase 5 plan: [drift and merge](docs/superpowers/plans/2026-10-13-master-agent-phase-5-drift-merge.md).
 
 ## What an update does
 
@@ -105,6 +106,9 @@ Response (`200` or `500`):
 | `500` | At least one result is `failed` or `timeout`. |
 | `503` | `UPDATER_API_KEY` is not set on the master. |
 
+A service that has **drifted** (see [Drift and moved services](#drift-and-moved-services))
+is refused with its reason, in auto mode too, until someone restores or accepts it.
+
 The webhook never creates services: the first deployment of a stack is done by hand. A
 new service always starts in **manual** mode; switch it with the *Make auto* button on the
 overview or the service page. Auto is refused for the updater's own services, `build:`
@@ -191,6 +195,42 @@ that differs from the newest one the master deployed. The newer version it was r
 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.
 
+## Drift and moved services
+
+A service has **drifted** when it runs something other than the version the last update
+job deployed. A service the updater never touched has nothing to drift from. Drift is
+decided by what runs, not by how the reference is written: the digest decides when both
+sides know one, so `repo:tag@sha256:…` is run by a service reporting `repo:tag` with that
+digest, and a different tag with the same digest is not drift. (The design spec compares
+the two reference strings; this is deliberately looser.) Between a successful job and the
+agent's next inventory report the service is not counted as drifted, because the report
+has not arrived yet.
+
+The master never corrects drift on its own, because a manual `docker stack deploy` of a
+newer tag may be intentional. Polling and the webhook start no job for a drifted
+service, in auto mode too; a person can still start one by hand. There are two ways out:
+
+- **Restore desired** deploys the recorded version again (a manual job to the exact
+  `repo:tag@digest`; an agent speaking protocol 3 is needed). If the service had drifted
+  to another digest of the same tag, that digest is remembered, so polling does not
+  deploy it again until the registry shows a newer one (unless `last_tag_digest` already
+  holds back a newer digest, which then stays).
+- **Accept running** records what runs now as the desired version. It starts no job and
+  leaves the available suggestion and the update mode alone.
+
+When a Compose service disappears from one agent and exactly one service with the same
+project and service name appears on another, and the removed one has history (a desired
+version, auto mode or jobs) while the new one has none, the service page suggests that it
+moved. **Merge** moves the old row, with its history, mode and desired version, to the new
+location and deletes the duplicate. Merging is never automatic. It can also be done by
+hand for any removed service of the same kind (Compose or Swarm, names may differ), but
+not for plain containers, the updater itself, or a service with a queued or running job.
+A merge keeps the old row's mode and desired version, so a manually merged target with its
+own history may show drift right after the merge and can be accepted.
+
+The restore, accept and merge buttons arrive with the UI follow-up (issue #28); the endpoints
+(`POST /services/{id}/restore`, `/accept`, `/merge`) and the data for them are in place.
+
 ## Adding agents
 
 Open the master's **Agents** page and add one. It shows a ready-to-run compose file with

docs/superpowers/plans/2026-10-13-master-agent-phase-5-drift-merge.md

@@ -0,0 +1,104 @@
+# Master/Agent Phase 5 — Drift and Merge Implementation Plan
+
+> This plan documents what was built for issue #10 (backend only) and where it deviates from the spec. Code in "Frozen contracts" is final; tasks give the contract and the test cases.
+
+**Goal:** The master knows when a service runs something other than the version the last job deployed (drift), never corrects it by itself, and offers two ways out: restore the desired version or accept the running one. When a Compose service disappears from one agent and appears on another, the master suggests that it moved, and a merge moves the old row (history, mode, desired ref) to the new location. Merging can also be done by hand.
+
+**Architecture:**
+- Model: `Service.Drifted()` plus `RunsRef` and `RunningRef` decide drift and what accept writes. `Move` carries a suggestion.
+- Store: migration 0005 adds `services.awaiting_inventory`; `FinishJob` sets it, every inventory upsert clears it. New `AcceptRunning`, `MoveCandidates`, `MergeSources`, `MergeService` (the only code that deletes a `services` row).
+- Jobs: `Start` refuses poll and webhook jobs for a drifted service. `Restorable` and `RestoreDesired` deploy the recorded ref again as a manual job.
+- Web: three POST routes (`/services/{id}/restore`, `/accept`, `/merge`) and new `pageData` fields. No template or CSS changes: issue #18 (the Ledger rewrite of the Services and Agents pages) is a separate UI change; the restore, accept and merge buttons arrive with the UI follow-up, issue #28.
+
+**Tech Stack:** no new dependencies.
+
+**Spec:** `docs/superpowers/specs/2026-10-12-master-agent-design.md`, sections "Drift" and "Service identity and moves". Phase 4 plan conventions still apply.
+
+## Global Constraints
+
+- `go test ./...` passes with no Docker and no Postgres. Postgres tests carry `//go:build integration`.
+- No changes to `internal/master/web/templates/*`, `static/*`, `layout.html` or `groupServices`.
+- Interfaces declared at the consumer (`jobs.Store`, `web.Store`, `web.Jobs`). Hand-written fakes and testify, table-driven where there is more than one case.
+- GoDoc on every exported identifier. Conventional Commits, each ending with ` (#10)`.
+- Every state-changing request is a POST with a CSRF token. Refusals render the service page with the reason (409), not a bare error.
+- The invariants of earlier phases hold: one active job per service (`update_jobs_one_active`), the webhook and poller never create services, the updater's own services are never updated or merged.
+
+## Deliberate deviations from the spec in this phase
+
+1. **Drift compares what runs, not reference strings.** The spec says a service drifts when `actual_ref` differs from `desired_ref`. `desired_ref` is always `repo:tag@digest` while `actual_ref` is the raw image string, so a string comparison would flag every healthy service. `RunsRef` compares the repo exactly (no `docker.io` normalisation), then the digest when both sides know one, else the followed tag. The same digest under a different tag is therefore not drift.
+2. **`awaiting_inventory`.** The agent's upsert arrives a debounce interval after `job.result`; until then `desired_ref` is newer than `actual_ref`. The flag, set by `FinishJob` and cleared by the next upsert of the row, keeps that window from counting as drift.
+3. **Poll and webhook jobs are refused for a drifted service** (`ReasonDrifted`). Manual jobs are allowed, so a person can still update or roll back.
+4. **Move suggestion is stricter than "exactly one appears".** Besides one removed and one live row on each side, the live row must be newer, have no desired ref and no jobs, and the removed row must have something worth moving (desired ref, auto mode or jobs). Otherwise every restarted stack would produce a suggestion.
+5. **Manual merge is allowed for Compose and Swarm of the same kind**, names may differ, same agent allowed. Plain containers and the updater itself are refused.
+6. **Accept does not clear `available_*`**, and the form's `actual` field (`Service.RunningRef()`) is only a guard; the value written comes from the row.
+7. **The overview does not show moves or drift yet.** `MoveCandidates` is only read on the service page. `Service.Drifted()` is a model method, so the UI follow-up needs no further wiring.
+8. **`CreateJob` maps the foreign key violation on `update_jobs_service_id_fkey` to `ErrNotFound`**, so a job for a service that a merge just deleted fails cleanly.
+
+## Review Focus
+
+| Failure case | Test that covers it |
+|---|---|
+| Drift by digest, not string: tag+digest vs tag and reported digest, bare image, tagless, port colon, other repo, case. | model `TestRunsRef`, `TestDrifted` |
+| Accept writes `repo:tag@digest` from the running version. | model `TestRunningRef` |
+| A fresh job result is not drift until the inventory arrives. | store `TestFinishJob_SetsAwaitingInventoryUntilTheNextUpsert`; e2e `TestNoDriftAfterASuccessfulJob` |
+| Accept is guarded by the running version, refuses removed/never-updated, leaves `last_tag_digest`, `available_*`, mode alone. | store `TestAcceptRunning_*` |
+| Poll and webhook start no job for a drifted service; manual does. | jobs `TestStart_RefusesPollAndWebhookWhileDrifted`; e2e `TestDriftedAutoServiceIsNotUpdatedByPollOrWebhook` |
+| Restore deploys the recorded ref and remembers the drifted digest of the same tag. | jobs `TestRestoreDesired_*`; e2e `TestRestoreDeploysThePinnedRefOnTheAgent` |
+| Move suggestion only for a unique, new, history-less target and a removed source with history. | store `TestMoveCandidates` |
+| Merge keeps id, mode, desired ref, first seen; takes location and inventory; moves jobs; deletes the duplicate. | store `TestMergeService_MovesHistoryModeAndDesiredToTheNewLocation` |
+| Merge refusals change nothing (active job, kinds, container, self, live old, removed new, revoked agent, unknown). | store `TestMergeService_Refusals` |
+| Inventory after a merge updates the survivor; the old agent returning creates a new row. | store `TestMergeService_InventoryAfterTheMerge` |
+| A job for a deleted service is `ErrNotFound`. | store `TestCreateJob_UnknownServiceIsNotFound` |
+| Handlers: redirects, 400/404/409, signals published, survivor id in the redirect. | web `TestRestoreDesired`, `TestAcceptRunning_*`, `TestMergeService_*`, `TestServiceDetail_DriftAndMoveData` |
+
+## Frozen contracts
+
+### C1. Migration `0005_drift_merge.sql`
+`services.awaiting_inventory boolean NOT NULL DEFAULT false`; index `services_by_compose_name` on `(compose_project, compose_service) WHERE kind = 'compose'`.
+
+### C2. `internal/master/model`
+`Service.AwaitingInventory`; `Service.Drifted()`; `RunsRef(desired, actualRef, actualDigest)`; `RunningRef(actualRef, actualDigest)`; `Move{FromID, ToID, FromAgent, ToAgent}` (FromID is the removed row). `ReasonDrifted`, `ReasonNotDrifted`; `MergeRefusedError{Reason}` with the `ReasonMerge*` constants (`ReasonSelf` is reused).
+
+### C3. Store
+- `upsertService` sets `awaiting_inventory = false`; `FinishJob` success sets it to true.
+- `AcceptRunning(ctx, id, actualRef, actualDigest, desired)`: updates only a live, previously updated row that still reports that actual ref and digest. `ErrNotFound` for unknown or removed, `ErrConflict` otherwise.
+- `MoveCandidates(ctx) ([]model.Move, error)`: Compose only, not self, other agent, unique on both sides before the extra filters (live row newer, no desired ref, no jobs; removed row has a desired ref, auto mode or jobs). Agent names by `LEFT JOIN`, so a revoked agent keeps its name.
+- `MergeSources(ctx, toID)`: removed rows of the same kind (not container, not self), same name first, then `removed_at DESC`, at most 50.
+- `MergeService(ctx, fromID, toID)`: one transaction. Lock order: the live row's agent (`FOR SHARE`, `ErrRevoked`), then both rows `ORDER BY id FOR UPDATE`. Validates (`MergeRefusedError`), refuses an active job on either row (`ErrJobActive`), moves the duplicate's jobs, reads its location and inventory, **deletes it**, then updates the old row (keeps id, kind, `first_seen_at`, `update_mode`, `desired_ref`; takes agent, location, inventory columns, available suggestion and poll error; `last_tag_digest` only if the old row has none; `removed_at = NULL`, `awaiting_inventory = false`). The delete comes first because `location_key` is unique and not deferrable.
+- `CreateJob`: foreign key violation `update_jobs_service_id_fkey` becomes `ErrNotFound`.
+
+### C4. Jobs
+`Start` returns `NotUpdatableError{ReasonDrifted}` for a non-manual trigger on a drifted service. `Restorable(svc)`: drifted, then `updatable(svc, true)`, and the desired ref's repo is not the agent image repo. `RestoreDesired(ctx, id, by)`: `Start` with `Ref: svc.DesiredRef`, manual. After a created job, `SetLastTagDigest` to the drifted digest when the actual tag equals the desired tag, the digest is well formed, differs from the desired digest and from `last_tag_digest`, and `last_tag_digest` is empty or equals the desired digest (otherwise it holds back a newer digest and stays).
+
+### C5. Web
+Routes `POST /services/{id}/restore`, `/accept` (form `actual` required, the `Service.RunningRef()` the user saw, compared with the row's; optional `return=overview`), `/merge` (form `from` = removed row, `{id}` = live row; redirects to the surviving id). `pageData` gains `RestoreBlocked`, `MoveFrom`, `MoveTo`, `MergeSources`. Log events `drift_accepted` and `service_merged`.
+
+### C6. Concurrency
+An upsert for the live row's location waits on its lock, then updates the survivor. A snapshot after the merge lists the survivor's key, so it is not marked removed. A removal racing a merge can be lost; the next snapshot fixes it. `CreateJob` on the deleted row gets `ErrNotFound`; the poller and webhook re-read and log it.
+
+## Tasks
+
+### Task 1: Store and model (C1, C2, C3)
+Migration, `imageref.IsDigest` (replaces the local pattern in the store), model types and functions, store methods. Tests: model tables; integration tests listed in Review Focus.
+
+### Task 2: Jobs (C4)
+`Store.SetLastTagDigest` on the consumer interface, `Start` refusal, `Restorable`, `RestoreDesired`. Tests: refusal table, manual job to the desired ref with protocol 3, active job, old agent, last-tag-digest table.
+
+### Task 3: Web (C5)
+Handlers, interfaces, `pageData`. Tests with fakes.
+
+### Task 4: End-to-end
+`memStore.FinishJob` sets `AwaitingInventory`. `internal/e2e/drift_test.go`.
+
+### Task 5: Documentation
+README section "Drift and moved services", webhook note, CLAUDE.md design intent and gotchas, this plan.
+
+## 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/
+```

internal/e2e/agent_master_test.go

@@ -11,7 +11,6 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
-	"regexp"
 	"sort"
 	"strings"
 	"sync"
@@ -25,6 +24,7 @@ import (
 	"github.com/shcizo/package-updater/internal/agent/conn"
 	"github.com/shcizo/package-updater/internal/agent/updater"
 	"github.com/shcizo/package-updater/internal/agenttoken"
+	"github.com/shcizo/package-updater/internal/imageref"
 	"github.com/shcizo/package-updater/internal/master/events"
 	"github.com/shcizo/package-updater/internal/master/hub"
 	"github.com/shcizo/package-updater/internal/master/model"
@@ -32,9 +32,6 @@ 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
@@ -245,7 +242,8 @@ 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.AwaitingInventory = true
+		if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && imageref.IsDigest(d) {
 			svc.LastTagDigest = d
 		}
 		// Same rule as the SQL: the job was created for the suggested ref after it was reported.

internal/e2e/drift_test.go

@@ -0,0 +1,113 @@
+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/model"
+	"github.com/shcizo/package-updater/internal/master/poller"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+const (
+	driftDesired = "sha256:" + "d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1"
+	driftOther   = "sha256:" + "0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a"
+)
+
+// seedDrifted replaces the environment's service with an auto service that was
+// updated to slusk:1.4.2 at driftDesired but runs driftOther now.
+func (e *jobEnv) seedDrifted() {
+	e.store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
+		AgentID: ptr(agentID), ImageRepo: slusk, UpdateMode: model.ModeAuto,
+		DesiredRef: slusk + ":1.4.2@" + driftDesired,
+		ActualRef:  slusk + ":1.4.2", ActualDigest: driftOther})
+}
+
+// inventoryWrite is what the store does for an agent's upsert of the service.
+func (e *jobEnv) inventoryWrite(id int64, actualRef, actualDigest string) {
+	e.store.mu.Lock()
+	defer e.store.mu.Unlock()
+	svc := e.store.svcRows[id]
+	svc.ActualRef, svc.ActualDigest, svc.AwaitingInventory = actualRef, actualDigest, false
+	e.store.svcRows[id] = svc
+}
+
+func TestDriftedAutoServiceIsNotUpdatedByPollOrWebhook(t *testing.T) {
+	e := newJobEnv(t)
+	e.seedDrifted()
+	reg := &fixedRegistry{digests: map[string]string{slusk + ":1.4.2": "sha256:new"}}
+	p := poller.New(e.store, e.jobs, reg, e.m.broker, poller.Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil)))
+
+	p.Cycle(context.Background())
+
+	assert.Equal(t, 1, reg.calls)
+	assert.Zero(t, e.store.jobCount(), "polling starts no job for a drifted service")
+	svc, err := e.store.GetService(context.Background(), 1)
+	require.NoError(t, err)
+	assert.Equal(t, slusk+":1.4.2", svc.AvailableRef, "the suggestion is still recorded")
+
+	code, body := callWebhook(t, e.webhookServer(t), `{"image":"`+slusk+`:ignored","tag":"2.0.0"}`)
+
+	assert.Equal(t, 200, code)
+	require.Len(t, body.Results, 1)
+	assert.Equal(t, "refused", body.Results[0].Status)
+	assert.Contains(t, body.Results[0].Error, "drifted")
+	assert.Zero(t, e.store.jobCount(), "the webhook starts no job either")
+}
+
+func TestRestoreDeploysThePinnedRefOnTheAgent(t *testing.T) {
+	e := newJobEnv(t)
+	e.seedDrifted()
+	close(e.exec.release)
+
+	job, err := e.jobs.RestoreDesired(context.Background(), 1, "alice")
+
+	require.NoError(t, err)
+	assert.Equal(t, model.TriggerManual, job.Trigger)
+	select {
+	case ref := <-e.exec.started:
+		assert.Equal(t, slusk+":1.4.2@"+driftDesired, ref, "the agent receives exactly the desired ref")
+	case <-time.After(waitFor):
+		t.Fatal("the agent never received the job")
+	}
+	done := e.waitStatus(t, job.ID, model.JobSucceeded)
+	assert.Equal(t, slusk+":1.4.2@"+driftDesired, done.ToRef)
+	svc, err := e.store.GetService(context.Background(), 1)
+	require.NoError(t, err)
+	assert.Equal(t, driftOther, svc.LastTagDigest, "the drifted digest is remembered so polling does not redeploy it")
+	assert.False(t, svc.Drifted(), "the inventory has not caught up yet")
+
+	e.inventoryWrite(1, slusk+":1.4.2", driftDesired)
+	svc, err = e.store.GetService(context.Background(), 1)
+	require.NoError(t, err)
+	assert.False(t, svc.Drifted())
+}
+
+func TestNoDriftAfterASuccessfulJob(t *testing.T) {
+	e := newJobEnv(t)
+	close(e.exec.release)
+
+	job, err := e.jobs.Trigger(context.Background(), 1, "alice")
+
+	require.NoError(t, err)
+	e.waitStatus(t, job.ID, model.JobSucceeded)
+	svc, err := e.store.GetService(context.Background(), 1)
+	require.NoError(t, err)
+	assert.Equal(t, slusk+":1.4.2@"+e2eDigest, svc.DesiredRef)
+	assert.Equal(t, slusk+":1.4.2@sha256:old", svc.ActualRef, "the agent's inventory write has not arrived")
+	assert.False(t, svc.Drifted(), "between job.result and the inventory write the row is not drifted")
+
+	e.inventoryWrite(1, slusk+":1.4.2", e2eDigest)
+	svc, _ = e.store.GetService(context.Background(), 1)
+	assert.False(t, svc.Drifted())
+
+	e.inventoryWrite(1, slusk+":1.4.2", driftOther)
+	svc, _ = e.store.GetService(context.Background(), 1)
+	assert.True(t, svc.Drifted(), "a later inventory write of another digest is drift")
+}

internal/imageref/imageref.go

@@ -2,7 +2,10 @@
 // and digest.
 package imageref
 
-import "strings"
+import (
+	"regexp"
+	"strings"
+)
 
 // Ref is an image reference split into its parts. Tag and Digest are empty
 // when the reference has none.
@@ -41,3 +44,8 @@ func (r Ref) String() string {
 	}
 	return s
 }
+
+var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+
+// IsDigest reports whether d is a well-formed sha256 digest.
+func IsDigest(d string) bool { return digestPattern.MatchString(d) }

internal/imageref/imageref_test.go

@@ -1,6 +1,7 @@
 package imageref
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -27,3 +28,19 @@ func TestParse(t *testing.T) {
 		})
 	}
 }
+
+func TestIsDigest(t *testing.T) {
+	cases := []struct {
+		in   string
+		want bool
+	}{
+		{"sha256:" + strings.Repeat("a", 64), true},
+		{"sha256:" + strings.Repeat("a", 63), false},
+		{"sha256:" + strings.Repeat("A", 64), false},
+		{"sha512:" + strings.Repeat("a", 64), false},
+		{"", false},
+	}
+	for _, c := range cases {
+		assert.Equal(t, c.want, IsDigest(c.in), c.in)
+	}
+}

internal/master/jobs/drift_test.go

@@ -0,0 +1,237 @@
+package jobs
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+const (
+	driftRepo = "gitea.shcizo.se/shcizo/slusk"
+	driftD1   = "sha256:1111111111111111111111111111111111111111111111111111111111111111"
+	driftD2   = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
+	driftD3   = "sha256:3333333333333333333333333333333333333333333333333333333333333333"
+)
+
+// driftedSvc was updated to 1.4.2 at D1 but now runs 1.4.2 at D2.
+func driftedSvc() model.Service {
+	svc := composeSvc
+	svc.ImageRepo = driftRepo
+	svc.DesiredRef = driftRepo + ":1.4.2@" + driftD1
+	svc.ActualRef = driftRepo + ":1.4.2"
+	svc.ActualDigest = driftD2
+	return svc
+}
+
+func notUpdatableReason(t *testing.T, err error) string {
+	t.Helper()
+	var nu *model.NotUpdatableError
+	require.True(t, errors.As(err, &nu), "got %v", err)
+	return nu.Reason
+}
+
+func TestRestorable(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*model.Service)
+		reason string // "" means restorable
+	}{
+		{"drifted", func(*model.Service) {}, ""},
+		{"drifted service that follows no tag", func(s *model.Service) {
+			s.ActualRef = driftRepo + "@" + driftD2
+		}, ""},
+		{"not drifted", func(s *model.Service) { s.ActualDigest = driftD1 }, model.ReasonNotDrifted},
+		{"never updated", func(s *model.Service) { s.DesiredRef = "" }, model.ReasonNotDrifted},
+		{"awaiting inventory", func(s *model.Service) { s.AwaitingInventory = true }, model.ReasonNotDrifted},
+		{"removed", func(s *model.Service) { s.RemovedAt = ptr(time.Now()) }, model.ReasonRemoved},
+		{"flagged self", func(s *model.Service) { s.Self = true }, model.ReasonSelf},
+		{"built", func(s *model.Service) { s.Built = true }, model.ReasonBuilt},
+		{"container", func(s *model.Service) { s.Kind = protocol.KindContainer }, model.ReasonContainer},
+		{"desired ref is the agent image", func(s *model.Service) {
+			s.DesiredRef = selfImage + "@" + driftD1
+		}, model.ReasonSelf},
+	}
+	s, _ := newService(&fakeStore{}, &fakeHub{})
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc := driftedSvc()
+			tt.mutate(&svc)
+			err := s.Restorable(svc)
+			if tt.reason == "" {
+				assert.NoError(t, err)
+				return
+			}
+			assert.Equal(t, tt.reason, notUpdatableReason(t, err))
+		})
+	}
+}
+
+func TestRestoreDesired_StartsAManualJobToTheDesiredRef(t *testing.T) {
+	svc := driftedSvc()
+	st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
+	h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+	s, _ := newService(st, h)
+
+	job, err := s.RestoreDesired(context.Background(), 1, "sub-1")
+
+	require.NoError(t, err)
+	assert.Equal(t, model.JobQueued, job.Status)
+	require.Len(t, st.created, 1)
+	assert.Equal(t, svc.DesiredRef, st.created[0].ToRef)
+	assert.Equal(t, svc.ActualRef, st.created[0].FromRef)
+	assert.Equal(t, "sub-1", st.created[0].RequestedBy)
+	assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
+	assert.Equal(t, []int{protocol.MinPinnedRefVersion}, h.mins)
+	require.Len(t, h.starts, 1)
+	assert.Equal(t, svc.DesiredRef, h.starts[0].start.Ref)
+}
+
+func TestRestoreDesired_Refusals(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*model.Service)
+		reason string
+	}{
+		{"not drifted", func(s *model.Service) { s.ActualDigest = driftD1 }, model.ReasonNotDrifted},
+		{"self", func(s *model.Service) { s.Self = true }, model.ReasonSelf},
+		{"built", func(s *model.Service) { s.Built = true }, model.ReasonBuilt},
+		{"container", func(s *model.Service) { s.Kind = protocol.KindContainer }, model.ReasonContainer},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc := driftedSvc()
+			tt.mutate(&svc)
+			st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, pub := newService(st, h)
+
+			_, err := s.RestoreDesired(context.Background(), 1, "sub-1")
+
+			assert.Equal(t, tt.reason, notUpdatableReason(t, err))
+			assert.Empty(t, st.created)
+			assert.Empty(t, h.starts)
+			assert.Empty(t, pub.topics)
+			assert.Empty(t, st.lastTag, "last_tag_digest untouched")
+		})
+	}
+}
+
+func TestRestoreDesired_UnknownService(t *testing.T) {
+	s, _ := newService(&fakeStore{services: map[int64]model.Service{}}, &fakeHub{})
+
+	_, err := s.RestoreDesired(context.Background(), 9, "sub-1")
+
+	assert.ErrorIs(t, err, model.ErrNotFound)
+}
+
+func TestRestoreDesired_ActiveJobReturnsItAndMarksNothing(t *testing.T) {
+	activeJob := model.Job{ID: 99, Status: model.JobRunning}
+	st := &fakeStore{services: map[int64]model.Service{1: driftedSvc()}, jobs: map[int64]model.Job{99: activeJob}, active: &activeJob}
+	s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}})
+
+	job, err := s.RestoreDesired(context.Background(), 1, "sub-1")
+
+	assert.ErrorIs(t, err, model.ErrJobActive)
+	assert.Equal(t, int64(99), job.ID)
+	assert.Empty(t, st.lastTag)
+}
+
+func TestRestoreDesired_OldAgentGivesAFailedJob(t *testing.T) {
+	st := &fakeStore{services: map[int64]model.Service{1: driftedSvc()}, jobs: map[int64]model.Job{}}
+	h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2}
+	s, _ := newService(st, h)
+
+	job, err := s.RestoreDesired(context.Background(), 1, "sub-1")
+
+	require.NoError(t, err)
+	assert.Equal(t, model.JobFailed, job.Status)
+	assert.Equal(t, model.ErrAgentTooOldForPinned.Error(), job.Error)
+	assert.Empty(t, h.starts)
+}
+
+func TestRestoreDesired_LastTagDigest(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*model.Service)
+		want   string // "" means SetLastTagDigest is not called
+	}{
+		{"drifted to another digest of the same tag", func(*model.Service) {}, driftD2},
+		{"digest in the actual ref", func(s *model.Service) {
+			s.ActualRef, s.ActualDigest = driftRepo+":1.4.2@"+driftD3, ""
+		}, driftD3},
+		{"last tag digest holds back a newer one", func(s *model.Service) {
+			s.DesiredRef = driftRepo + ":1.4.2@" + driftD1
+			s.LastTagDigest = driftD3
+		}, ""},
+		{"last tag digest equals the desired digest", func(s *model.Service) { s.LastTagDigest = driftD1 }, driftD2},
+		{"already remembered", func(s *model.Service) { s.LastTagDigest = driftD2 }, ""},
+		{"different tag", func(s *model.Service) { s.ActualRef = driftRepo + ":1.5.0" }, ""},
+		{"actual digest malformed", func(s *model.Service) { s.ActualDigest = "sha256:abc" }, ""},
+		{"desired follows no tag", func(s *model.Service) { s.DesiredRef = driftRepo + "@" + driftD1 }, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc := driftedSvc()
+			tt.mutate(&svc)
+			st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
+			s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}})
+
+			_, err := s.RestoreDesired(context.Background(), 1, "sub-1")
+
+			require.NoError(t, err)
+			if tt.want == "" {
+				assert.Empty(t, st.lastTag)
+				return
+			}
+			assert.Equal(t, map[int64]string{1: tt.want}, st.lastTag)
+		})
+	}
+}
+
+func TestStart_RefusesPollAndWebhookWhileDrifted(t *testing.T) {
+	tests := []struct {
+		trigger model.JobTrigger
+		refused bool
+	}{
+		{model.TriggerPoll, true},
+		{model.TriggerWebhook, true},
+		{model.TriggerManual, false},
+	}
+	for _, tt := range tests {
+		t.Run(string(tt.trigger), func(t *testing.T) {
+			svc := driftedSvc()
+			st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, _ := newService(st, h)
+
+			_, err := s.Start(context.Background(), svc, StartRequest{Trigger: tt.trigger, Ref: driftRepo + ":2.0.0"})
+
+			if tt.refused {
+				assert.Equal(t, model.ReasonDrifted, notUpdatableReason(t, err))
+				assert.Empty(t, st.created)
+				return
+			}
+			require.NoError(t, err)
+			assert.Len(t, st.created, 1)
+		})
+	}
+}
+
+func TestStart_PollAllowedWhenNotDrifted(t *testing.T) {
+	svc := driftedSvc()
+	svc.AwaitingInventory = true
+	st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
+	s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}})
+
+	_, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerPoll})
+
+	require.NoError(t, err)
+	assert.Len(t, st.created, 1)
+}

internal/master/jobs/jobs.go

@@ -21,6 +21,7 @@ type Store interface {
 	FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error)
 	GetJob(ctx context.Context, id int64) (model.Job, error)
 	GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error)
+	SetLastTagDigest(ctx context.Context, id int64, digest string) error
 }
 
 // Hub routes jobs to agent connections. *hub.Hub satisfies it.
@@ -114,6 +115,81 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error {
 	return nil
 }
 
+// Restorable returns nil or a *model.NotUpdatableError saying why the service
+// cannot be restored to its desired version: it must have drifted, and the
+// desired version is deployed like an explicit ref, so a service that follows
+// no tag is fine. The service page calls it for every render.
+func (s *Service) Restorable(svc model.Service) error {
+	if svc.RemovedAt != nil {
+		return &model.NotUpdatableError{Reason: model.ReasonRemoved}
+	}
+	if !svc.Drifted() {
+		return &model.NotUpdatableError{Reason: model.ReasonNotDrifted}
+	}
+	if err := s.updatable(svc, true); err != nil {
+		return err
+	}
+	if imageref.Parse(svc.DesiredRef).Repo == s.selfRepo {
+		return &model.NotUpdatableError{Reason: model.ReasonSelf}
+	}
+	return nil
+}
+
+// RestoreDesired starts a manual job that deploys the drifted service's
+// desired ref again. The ref is read from the row, never from the caller. It
+// returns model.ErrNotFound for an unknown service and otherwise what Start
+// returns, with a *model.NotUpdatableError from Restorable if it cannot be
+// restored.
+//
+// When the service had drifted to another digest of the same tag, that digest
+// becomes last_tag_digest, so the poller does not deploy it again before the
+// registry shows a newer one (the same rule as for any rollback). It is not
+// written when last_tag_digest holds back a newer digest, that is when it is
+// set and differs from the digest the desired ref names.
+func (s *Service) RestoreDesired(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) {
+	svc, err := s.store.GetService(ctx, serviceID)
+	if err != nil {
+		return model.Job{}, err
+	}
+	if err := s.Restorable(svc); err != nil {
+		return model.Job{}, err
+	}
+	job, err := s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: svc.DesiredRef})
+	if err != nil {
+		return job, err
+	}
+	if d := driftedTagDigest(svc); d != "" && d != svc.LastTagDigest && holdsNothingNewer(svc) {
+		if err := s.store.SetLastTagDigest(ctx, svc.ID, d); err != nil {
+			s.log.Warn("last_tag_digest_not_set", "service_id", svc.ID, "error", err)
+		}
+	}
+	return job, nil
+}
+
+// holdsNothingNewer reports whether last_tag_digest may be overwritten by a
+// restore: it is empty or equals the digest the desired ref names. Anything
+// else is a digest the poller must keep holding back.
+func holdsNothingNewer(svc model.Service) bool {
+	return svc.LastTagDigest == "" || svc.LastTagDigest == imageref.Parse(svc.DesiredRef).Digest
+}
+
+// driftedTagDigest returns the digest a drifted service runs when it runs the
+// same tag as desired at another digest, else "".
+func driftedTagDigest(svc model.Service) string {
+	tag := FollowedRef(svc.DesiredRef)
+	if tag == "" || tag != FollowedRef(svc.ActualRef) {
+		return ""
+	}
+	d := imageref.Parse(svc.ActualRef).Digest
+	if d == "" {
+		d = svc.ActualDigest
+	}
+	if !imageref.IsDigest(d) || d == imageref.Parse(svc.DesiredRef).Digest {
+		return ""
+	}
+	return d
+}
+
 // StartRequest says who asks for an update and what to deploy.
 type StartRequest struct {
 	Trigger     model.JobTrigger
@@ -161,8 +237,12 @@ func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64
 // 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.
+// running. Poll and webhook jobs are refused for a drifted service; only a
+// person starts one then.
 func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest) (model.Job, error) {
+	if req.Trigger != model.TriggerManual && svc.Drifted() {
+		return model.Job{}, &model.NotUpdatableError{Reason: model.ReasonDrifted}
+	}
 	if err := s.updatable(svc, req.Ref != ""); err != nil {
 		return model.Job{}, err
 	}

internal/master/jobs/jobs_test.go

@@ -23,7 +23,8 @@ type fakeStore struct {
 	active   *model.Job // returned with ErrJobActive when set
 	versions map[int64]model.ImageVersion
 
-	revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these
+	revokedAgents map[int64]bool   // CreateJob returns ErrAgentOffline for these
+	lastTag       map[int64]string // SetLastTagDigest calls: service ID to digest
 }
 
 func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) {
@@ -63,6 +64,14 @@ func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVer
 	return v, nil
 }
 
+func (f *fakeStore) SetLastTagDigest(_ context.Context, id int64, digest string) error {
+	if f.lastTag == nil {
+		f.lastTag = map[int64]string{}
+	}
+	f.lastTag[id] = digest
+	return nil
+}
+
 func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil }
 
 type startCall struct {

internal/master/model/job.go

@@ -44,6 +44,27 @@ const (
 	ReasonNoTag     = "follows no tag"
 	ReasonRemoved   = "service no longer exists"
 	ReasonBuilt     = "built from source (build:)"
+
+	ReasonDrifted    = "drifted: restore or accept the running version first"
+	ReasonNotDrifted = "runs its desired version"
+)
+
+// MergeRefusedError says why two services cannot be merged. Reason is shown
+// in the UI as is.
+type MergeRefusedError struct {
+	Reason string
+}
+
+func (e *MergeRefusedError) Error() string { return "cannot merge: " + e.Reason }
+
+// Reasons two services cannot be merged. ReasonSelf is reused for the
+// updater itself.
+const (
+	ReasonMergeSame      = "a service cannot be merged into itself"
+	ReasonMergeLiveOld   = "the old service still exists"
+	ReasonMergeGoneNew   = "the new service no longer exists"
+	ReasonMergeKind      = "only services of the same kind can be merged"
+	ReasonMergeContainer = "plain containers cannot be merged"
 )
 
 // JobStatus is the state of an update job.

internal/master/model/model.go

@@ -15,6 +15,8 @@ import (
 // Sentinel errors returned by storage.
 var (
 	ErrNotFound = errors.New("not found")
+	// ErrConflict means the row already exists, or that it changed meanwhile
+	// (AcceptRunning, MergeService).
 	ErrConflict = errors.New("already exists")
 	ErrRevoked  = errors.New("agent revoked")
 )
@@ -68,9 +70,13 @@ type Service struct {
 	// not start another job for the same digest, so a rollback sticks until the
 	// registry shows a different one.
 	LastTagDigest string
-	FirstSeenAt   time.Time
+	// AwaitingInventory is true from a successful job until the next inventory
-	UpdatedAt     time.Time
+	// write for the row. Until then DesiredRef is newer than ActualRef without
-	RemovedAt     *time.Time
+	// the service having drifted.
+	AwaitingInventory bool
+	FirstSeenAt       time.Time
+	UpdatedAt         time.Time
+	RemovedAt         *time.Time
 }
 
 // Update modes of a service.
@@ -113,6 +119,68 @@ func (v ImageVersion) Ref() string {
 	return imageref.Ref{Repo: v.Repo, Tag: v.Tag, Digest: v.Digest}.String()
 }
 
+// Drifted reports whether the service runs something other than the version
+// the last job deployed. A service the updater never touched, a removed one
+// and one whose inventory has not caught up with the last job do not drift.
+func (s Service) Drifted() bool {
+	return s.DesiredRef != "" && s.RemovedAt == nil && !s.AwaitingInventory &&
+		!RunsRef(s.DesiredRef, s.ActualRef, s.ActualDigest)
+}
+
+// RunsRef reports whether a service reporting actualRef and actualDigest runs
+// the desired ref. The digest decides when both sides know one, so
+// "repo:tag@D" is run by "repo:tag" reporting digest D, and by "repo@D".
+// Otherwise the repo (compared exactly, without registry normalisation) and
+// the followed tag must be equal.
+func RunsRef(desired, actualRef, actualDigest string) bool {
+	d, a := imageref.Parse(desired), imageref.Parse(actualRef)
+	if d.Repo != a.Repo {
+		return false
+	}
+	ad := a.Digest
+	if ad == "" {
+		ad = actualDigest
+	}
+	if d.Digest != "" && ad != "" {
+		return d.Digest == ad
+	}
+	return followedTag(d) == followedTag(a)
+}
+
+// followedTag is the tag a ref follows: a bare image follows "latest" and a
+// ref with a digest but no tag follows nothing.
+func followedTag(r imageref.Ref) string {
+	if r.Tag == "" && r.Digest == "" {
+		return "latest"
+	}
+	return r.Tag
+}
+
+// RunningRef is the ref to record as the desired one when the running version
+// is accepted: actualRef itself if it carries a digest, "repo:tag@digest" if
+// only actualDigest is known (a bare image is "latest"), else actualRef.
+func RunningRef(actualRef, actualDigest string) string {
+	r := imageref.Parse(actualRef)
+	if r.Digest != "" || !imageref.IsDigest(actualDigest) {
+		return actualRef
+	}
+	r.Tag, r.Digest = followedTag(r), actualDigest
+	return r.String()
+}
+
+// RunningRef returns the ref accepting the running version records as the
+// desired one; see the function RunningRef. Templates post it as the guard of
+// the accept form.
+func (s Service) RunningRef() string { return RunningRef(s.ActualRef, s.ActualDigest) }
+
+// Move says that the removed service FromID has probably moved to the live
+// service ToID on another agent. The names are the agents' own; a revoked
+// agent keeps its name.
+type Move struct {
+	FromID, ToID       int64
+	FromAgent, ToAgent string
+}
+
 // DisplayName is the human-facing name of the service.
 func (s Service) DisplayName() string {
 	switch s.Kind {

internal/master/model/model_test.go

@@ -2,6 +2,7 @@ package model
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -58,6 +59,84 @@ func TestImageVersionRef(t *testing.T) {
 	}
 }
 
+const (
+	dA = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	dB = "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+)
+
+func TestRunsRef(t *testing.T) {
+	cases := []struct {
+		name                          string
+		desired, actualRef, actualDig string
+		want                          bool
+	}{
+		{"same string", "repo:v1@" + dA, "repo:v1@" + dA, "", true},
+		{"tag and digest, actual reports it separately", "repo:tag@" + dA, "repo:tag", dA, true},
+		{"bare image reports latest digest", "postgres:latest@" + dA, "postgres", dA, true},
+		{"tagless actual with the same digest", "repo:tag@" + dA, "repo@" + dA, "", true},
+		{"same digest under another tag", "repo:v1@" + dA, "repo:v2@" + dA, "", true},
+		{"same digest, actual tag differs, digest reported", "repo:v1@" + dA, "repo:v2", dA, true},
+		{"same tag, actual digest unknown", "repo:tag@" + dA, "repo:tag", "", true},
+		{"desired without digest, same tag", "repo:tag", "repo:tag@" + dA, "", true},
+		{"bare desired, latest actual", "repo", "repo:latest", "", true},
+		{"registry with port", "localhost:5000/foo:v1@" + dA, "localhost:5000/foo:v1", dA, true},
+		{"registry with port, other digest", "localhost:5000/foo:v1@" + dA, "localhost:5000/foo:v1", dB, false},
+		{"same tag, other digest", "repo:tag@" + dA, "repo:tag", dB, false},
+		{"other digest in the ref", "repo:tag@" + dA, "repo:tag@" + dB, "", false},
+		{"other tag, digest unknown", "repo:v1@" + dA, "repo:v2", "", false},
+		{"tagless desired, tagged actual, digest unknown", "repo@" + dA, "repo:v1", "", false},
+		{"other repo", "repo:tag@" + dA, "other:tag", dA, false},
+		{"no docker.io normalisation", "nginx:1@" + dA, "docker.io/library/nginx:1", dA, false},
+		{"case-sensitive repo", "Repo:1@" + dA, "repo:1", dA, false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, RunsRef(c.desired, c.actualRef, c.actualDig))
+		})
+	}
+}
+
+func TestDrifted(t *testing.T) {
+	now := time.Now()
+	drifted := Service{DesiredRef: "repo:tag@" + dA, ActualRef: "repo:tag", ActualDigest: dB}
+	cases := []struct {
+		name   string
+		mutate func(*Service)
+		want   bool
+	}{
+		{"runs another digest", func(*Service) {}, true},
+		{"never updated", func(s *Service) { s.DesiredRef = "" }, false},
+		{"removed", func(s *Service) { s.RemovedAt = &now }, false},
+		{"awaiting inventory", func(s *Service) { s.AwaitingInventory = true }, false},
+		{"runs the desired digest", func(s *Service) { s.ActualDigest = dA }, false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			svc := drifted
+			c.mutate(&svc)
+			assert.Equal(t, c.want, svc.Drifted())
+		})
+	}
+}
+
+func TestRunningRef(t *testing.T) {
+	cases := []struct {
+		name, actualRef, actualDigest, want string
+	}{
+		{"ref with digest is kept", "repo:v1@" + dB, dA, "repo:v1@" + dB},
+		{"tag plus reported digest", "repo:v1", dA, "repo:v1@" + dA},
+		{"bare image becomes latest", "postgres", dA, "postgres:latest@" + dA},
+		{"registry with port", "localhost:5000/foo:v1", dA, "localhost:5000/foo:v1@" + dA},
+		{"no digest known", "repo:v1", "", "repo:v1"},
+		{"malformed digest", "repo:v1", "sha256:abc", "repo:v1"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, RunningRef(c.actualRef, c.actualDigest))
+		})
+	}
+}
+
 func TestJob_Active(t *testing.T) {
 	cases := []struct {
 		status JobStatus

internal/master/store/drift_merge_integration_test.go

@@ -0,0 +1,550 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+var (
+	digNew     = "sha256:" + strings.Repeat("e", 64)
+	digOld     = "sha256:" + strings.Repeat("f", 64)
+	sluskDrift = "gitea.shcizo.se/shcizo/slusk"
+)
+
+// serviceID returns the ID of the service the agent reports under svc's key.
+func serviceID(t *testing.T, s *Store, agentID int64, svc protocol.Service) int64 {
+	t.Helper()
+	var id int64
+	require.NoError(t, s.pool.QueryRow(context.Background(),
+		`SELECT id FROM services WHERE location_key = $1`, model.LocationKey(agentID, "", svc.Key())).Scan(&id))
+	return id
+}
+
+// finishJob runs a job for the service to success with the deployed ref.
+func finishJob(t *testing.T, s *Store, agentID, svcID int64, toRef, deployed string) {
+	t.Helper()
+	ctx := context.Background()
+	nj := newJob(svcID, agentID)
+	nj.ToRef = toRef
+	j, err := s.CreateJob(ctx, nj)
+	require.NoError(t, err)
+	ok, err := s.FinishJob(ctx, agentID, j.ID, model.JobFinish{Status: model.JobSucceeded, Ref: deployed, StartedAt: t0, FinishedAt: t0})
+	require.NoError(t, err)
+	require.True(t, ok)
+}
+
+func mustExec(t *testing.T, s *Store, sql string, args ...any) {
+	t.Helper()
+	_, err := s.pool.Exec(context.Background(), sql, args...)
+	require.NoError(t, err)
+}
+
+func TestFinishJob_SetsAwaitingInventoryUntilTheNextUpsert(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.False(t, got.AwaitingInventory)
+
+	finishJob(t, s, agent, svc, "r:1", "r:1@"+digNew)
+	got, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.True(t, got.AwaitingInventory, "set by the successful job")
+	assert.False(t, got.Drifted(), "desired is ahead of actual only until the inventory arrives")
+
+	require.NoError(t, s.UpsertService(ctx, agent, "", slusk, t0.Add(time.Minute)))
+	got, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.False(t, got.AwaitingInventory, "cleared by the agent's upsert")
+	assert.True(t, got.Drifted(), "the agent still reports the old version")
+}
+
+func TestFinishJob_FailedJobDoesNotSetAwaitingInventory(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	agent, svc := jobFixture(t, s, "a")
+	j, err := s.CreateJob(ctx, newJob(svc, agent))
+	require.NoError(t, err)
+
+	ok, err := s.FinishJob(ctx, agent, j.ID, model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0})
+	require.NoError(t, err)
+	require.True(t, ok)
+
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.False(t, got.AwaitingInventory)
+}
+
+func TestCreateJob_UnknownServiceIsNotFound(t *testing.T) {
+	s := openTestStore(t)
+	agent := newAgent(t, s, "a")
+
+	_, err := s.CreateJob(context.Background(), newJob(9999, agent))
+
+	assert.ErrorIs(t, err, model.ErrNotFound)
+}
+
+// driftedFixture is a service updated to digNew whose agent now reports digD1.
+func driftedFixture(t *testing.T, s *Store) (agent, svc int64) {
+	t.Helper()
+	agent, svc = jobFixture(t, s, "a")
+	finishJob(t, s, agent, svc, "r:1", sluskDrift+":1.4.2@"+digNew)
+	require.NoError(t, s.UpsertService(context.Background(), agent, "", slusk, t0.Add(time.Minute)))
+	return agent, svc
+}
+
+func TestAcceptRunning_WritesTheDesiredRef(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	_, svc := driftedFixture(t, s)
+	require.NoError(t, s.SetLastTagDigest(ctx, svc, "sha256:ltd"))
+	require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskDrift + ":1.5.0", Source: model.SourcePoll, At: t0}))
+	require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto))
+	before, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	require.True(t, before.Drifted())
+
+	running := model.RunningRef(before.ActualRef, before.ActualDigest)
+	require.NoError(t, s.AcceptRunning(ctx, svc, before.ActualRef, before.ActualDigest, running))
+
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, sluskDrift+":1.4.2@"+digD1, got.DesiredRef)
+	assert.False(t, got.Drifted())
+	assert.False(t, got.AwaitingInventory)
+	assert.Equal(t, "sha256:ltd", got.LastTagDigest, "last_tag_digest untouched")
+	assert.Equal(t, sluskDrift+":1.5.0", got.AvailableRef, "available suggestion untouched")
+	assert.Equal(t, model.ModeAuto, got.UpdateMode)
+}
+
+func TestAcceptRunning_Refusals(t *testing.T) {
+	tests := []struct {
+		name      string
+		prepare   func(t *testing.T, s *Store, agent, svc int64) (id int64, actualRef, actualDigest string)
+		wantError error
+	}{
+		{"running version changed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			return svc, sluskDrift + ":1.4.2", digOld
+		}, model.ErrConflict},
+		{"actual ref changed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			return svc, sluskDrift + ":1.3.0", digD1
+		}, model.ErrConflict},
+		{"never updated", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			mustExec(t, s, `UPDATE services SET desired_ref = '' WHERE id = $1`, svc)
+			return svc, sluskDrift + ":1.4.2", digD1
+		}, model.ErrConflict},
+		{"removed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			require.NoError(t, s.RemoveService(context.Background(), agent, "", slusk.Key(), t0.Add(time.Hour)))
+			return svc, sluskDrift + ":1.4.2", digD1
+		}, model.ErrNotFound},
+		{"unknown", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			return 9999, sluskDrift + ":1.4.2", digD1
+		}, model.ErrNotFound},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := openTestStore(t)
+			agent, svc := driftedFixture(t, s)
+			id, ref, digest := tt.prepare(t, s, agent, svc)
+
+			err := s.AcceptRunning(context.Background(), id, ref, digest, "x:1@"+digOld)
+
+			assert.ErrorIs(t, err, tt.wantError)
+			got, gerr := s.GetService(context.Background(), svc)
+			require.NoError(t, gerr)
+			assert.NotEqual(t, "x:1@"+digOld, got.DesiredRef, "nothing written")
+		})
+	}
+}
+
+// moveFixture has agent "a" with slusk removed after it was updated (history,
+// auto mode) and agent "b" reporting a new slusk row.
+type moveFixture struct {
+	a, b     int64
+	from, to int64
+}
+
+func newMoveFixture(t *testing.T, s *Store) moveFixture {
+	t.Helper()
+	ctx := context.Background()
+	f := moveFixture{a: newAgent(t, s, "a"), b: newAgent(t, s, "b")}
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{slusk}, t0))
+	f.from = serviceID(t, s, f.a, slusk)
+	finishJob(t, s, f.a, f.from, "r:1", sluskDrift+":1.4.2@"+digNew)
+	require.NoError(t, s.SetUpdateMode(ctx, f.from, model.ModeAuto))
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, nil, t0.Add(time.Hour)))
+	require.NoError(t, s.ApplySnapshot(ctx, f.b, worker, []protocol.Service{slusk}, t0.Add(2*time.Hour)))
+	f.to = serviceID(t, s, f.b, slusk)
+	return f
+}
+
+func TestMoveCandidates(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(t *testing.T, s *Store, f moveFixture)
+		want   bool
+		from   string // agent name of the removed row
+	}{
+		{"one removed and one live", func(*testing.T, *Store, moveFixture) {}, true, "a"},
+		{"removed row's agent was revoked", func(t *testing.T, s *Store, f moveFixture) {
+			_, err := s.RevokeAgent(context.Background(), f.a, t0.Add(3*time.Hour))
+			require.NoError(t, err)
+		}, true, "a"},
+		{"two live targets", func(t *testing.T, s *Store, f moveFixture) {
+			c := newAgent(t, s, "c")
+			require.NoError(t, s.ApplySnapshot(context.Background(), c, worker, []protocol.Service{slusk}, t0.Add(3*time.Hour)))
+		}, false, ""},
+		{"two removed sources", func(t *testing.T, s *Store, f moveFixture) {
+			ctx := context.Background()
+			d := newAgent(t, s, "d")
+			require.NoError(t, s.ApplySnapshot(ctx, d, worker, []protocol.Service{slusk}, t0))
+			require.NoError(t, s.ApplySnapshot(ctx, d, worker, nil, t0.Add(time.Hour)))
+		}, false, ""},
+		{"target has history", func(t *testing.T, s *Store, f moveFixture) {
+			finishJob(t, s, f.b, f.to, "r:1", sluskDrift+":1.4.2@"+digNew)
+		}, false, ""},
+		{"target has a desired ref", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET desired_ref = 'r:1@x' WHERE id = $1`, f.to)
+		}, false, ""},
+		{"source has no history", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `DELETE FROM update_jobs WHERE service_id = $1`, f.from)
+			mustExec(t, s, `UPDATE services SET desired_ref = '', update_mode = 'manual' WHERE id = $1`, f.from)
+		}, false, ""},
+		{"source only has auto mode", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `DELETE FROM update_jobs WHERE service_id = $1`, f.from)
+			mustExec(t, s, `UPDATE services SET desired_ref = '' WHERE id = $1`, f.from)
+		}, true, "a"},
+		{"source only has jobs", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET desired_ref = '', update_mode = 'manual' WHERE id = $1`, f.from)
+		}, true, "a"},
+		{"target is older than the source", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET first_seen_at = $2 WHERE id = $1`, f.to, t0.Add(-time.Hour))
+		}, false, ""},
+		{"plain containers", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET kind = 'container' WHERE id IN ($1, $2)`, f.from, f.to)
+		}, false, ""},
+		{"swarm services", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET kind = 'swarm' WHERE id IN ($1, $2)`, f.from, f.to)
+		}, false, ""},
+		{"the updater itself (target)", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to)
+		}, false, ""},
+		{"the updater itself (source)", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from)
+		}, false, ""},
+		{"live row also removed", func(t *testing.T, s *Store, f moveFixture) {
+			require.NoError(t, s.RemoveService(context.Background(), f.b, "", slusk.Key(), t0.Add(3*time.Hour)))
+		}, false, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := openTestStore(t)
+			f := newMoveFixture(t, s)
+			tt.mutate(t, s, f)
+
+			got, err := s.MoveCandidates(context.Background())
+
+			require.NoError(t, err)
+			if !tt.want {
+				assert.Empty(t, got)
+				return
+			}
+			assert.Equal(t, []model.Move{{FromID: f.from, ToID: f.to, FromAgent: tt.from, ToAgent: "b"}}, got)
+		})
+	}
+}
+
+func TestMergeService_MovesHistoryModeAndDesiredToTheNewLocation(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	finishJob(t, s, f.b, f.to, "r:2", sluskDrift+":2.0.0@"+digOld)
+	require.NoError(t, s.SetLastTagDigest(ctx, f.to, "sha256:fromto"))
+	require.NoError(t, s.SetAvailable(ctx, f.to, model.Available{Ref: sluskDrift + ":3", Digest: digNew, Source: model.SourcePoll, At: t0}))
+	require.NoError(t, s.SetPollError(ctx, []int64{f.to}, "registry down"))
+	// A job on the old row is finished, and awaiting_inventory is set on both.
+	before, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+
+	require.NoError(t, s.MergeService(ctx, f.from, f.to))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, f.from, got.ID, "the old id survives")
+	assert.Nil(t, got.RemovedAt)
+	assert.False(t, got.AwaitingInventory)
+	assert.Equal(t, "b", got.AgentName)
+	assert.Equal(t, model.LocationKey(f.b, "", slusk.Key()), got.LocationKey)
+	assert.Equal(t, before.FirstSeenAt, got.FirstSeenAt, "first_seen_at kept")
+	assert.Equal(t, model.ModeAuto, got.UpdateMode, "mode kept")
+	assert.Equal(t, before.DesiredRef, got.DesiredRef, "desired ref kept")
+	assert.Equal(t, slusk.Image, got.ActualRef, "inventory from the duplicate")
+	assert.Equal(t, digD1, got.ActualDigest)
+	assert.Equal(t, "running", got.Status)
+	assert.Equal(t, sluskDrift+":3", got.AvailableRef)
+	assert.Equal(t, digNew, got.AvailableDigest)
+	assert.Equal(t, "registry down", got.LastPollError)
+	assert.Equal(t, before.LastTagDigest, got.LastTagDigest, "an existing last_tag_digest wins")
+
+	_, err = s.GetService(ctx, f.to)
+	assert.ErrorIs(t, err, model.ErrNotFound, "the duplicate is deleted")
+	jobs, err := s.ListJobs(ctx, model.JobFilter{ServiceID: f.from})
+	require.NoError(t, err)
+	assert.Len(t, jobs, 2, "the duplicate's job moved to the survivor")
+	live, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"slusk/web"}, names(live))
+}
+
+func TestMergeService_TakesLastTagDigestOfTheDuplicateWhenTheOldRowHasNone(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	mustExec(t, s, `UPDATE services SET last_tag_digest = '' WHERE id = $1`, f.from)
+	require.NoError(t, s.SetLastTagDigest(ctx, f.to, "sha256:fromto"))
+
+	require.NoError(t, s.MergeService(ctx, f.from, f.to))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, "sha256:fromto", got.LastTagDigest)
+}
+
+func TestMergeService_ManualMergeOfAnyRemovedServiceOfTheSameKind(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"}
+	require.NoError(t, s.UpsertService(ctx, f.b, "", other, t0.Add(3*time.Hour)))
+	target := serviceID(t, s, f.b, other)
+
+	require.NoError(t, s.MergeService(ctx, f.from, target))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, "renamed/app", got.DisplayName(), "names may differ")
+}
+
+func TestMergeService_SameAgentIsAllowed(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"}
+	require.NoError(t, s.UpsertService(ctx, f.a, "", other, t0.Add(3*time.Hour)))
+
+	require.NoError(t, s.MergeService(ctx, f.from, serviceID(t, s, f.a, other)))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, "a", got.AgentName)
+}
+
+func TestMergeService_Swarm(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	m1 := newAgent(t, s, "m1")
+	require.NoError(t, s.ApplySnapshot(ctx, m1, manager, []protocol.Service{gitea}, t0))
+	oldID := serviceIDByName(t, s, "gitea_gitea")
+	require.NoError(t, s.ApplySnapshot(ctx, m1, manager, nil, t0.Add(time.Hour)))
+	renamed := gitea
+	renamed.SwarmService = "gitea2_gitea"
+	require.NoError(t, s.ApplySnapshot(ctx, m1, manager, []protocol.Service{renamed}, t0.Add(2*time.Hour)))
+	newID := serviceIDByName(t, s, "gitea2_gitea")
+
+	require.NoError(t, s.MergeService(ctx, oldID, newID))
+
+	got, err := s.GetService(ctx, oldID)
+	require.NoError(t, err)
+	assert.Equal(t, "gitea2_gitea", got.SwarmService)
+	assert.Nil(t, got.RemovedAt)
+}
+
+func serviceIDByName(t *testing.T, s *Store, swarmService string) int64 {
+	t.Helper()
+	var id int64
+	require.NoError(t, s.pool.QueryRow(context.Background(),
+		`SELECT id FROM services WHERE swarm_service = $1`, swarmService).Scan(&id))
+	return id
+}
+
+func TestMergeService_Refusals(t *testing.T) {
+	refused := func(reason string) func(*testing.T, error) {
+		return func(t *testing.T, err error) {
+			var mr *model.MergeRefusedError
+			require.True(t, errors.As(err, &mr), "got %v", err)
+			assert.Equal(t, reason, mr.Reason)
+		}
+	}
+	is := func(want error) func(*testing.T, error) {
+		return func(t *testing.T, err error) { assert.ErrorIs(t, err, want) }
+	}
+	tests := []struct {
+		name   string
+		run    func(t *testing.T, s *Store, f moveFixture) error
+		assert func(*testing.T, error)
+	}{
+		{"into itself", func(t *testing.T, s *Store, f moveFixture) error {
+			return s.MergeService(context.Background(), f.to, f.to)
+		}, refused(model.ReasonMergeSame)},
+		{"the old service still exists", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET removed_at = NULL WHERE id = $1`, f.from)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeLiveOld)},
+		{"the new service was removed", func(t *testing.T, s *Store, f moveFixture) error {
+			require.NoError(t, s.RemoveService(context.Background(), f.b, "", slusk.Key(), t0.Add(3*time.Hour)))
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeGoneNew)},
+		{"different kinds", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET kind = 'swarm' WHERE id = $1`, f.to)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeKind)},
+		{"plain containers", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET kind = 'container' WHERE id IN ($1, $2)`, f.from, f.to)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeContainer)},
+		{"the updater itself (new)", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonSelf)},
+		{"the updater itself (old)", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonSelf)},
+		{"active job on the old service", func(t *testing.T, s *Store, f moveFixture) error {
+			_, err := s.CreateJob(context.Background(), newJob(f.from, f.a))
+			require.NoError(t, err)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, is(model.ErrJobActive)},
+		{"active job on the new service", func(t *testing.T, s *Store, f moveFixture) error {
+			_, err := s.CreateJob(context.Background(), newJob(f.to, f.b))
+			require.NoError(t, err)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, is(model.ErrJobActive)},
+		{"unknown old service", func(t *testing.T, s *Store, f moveFixture) error {
+			return s.MergeService(context.Background(), 9999, f.to)
+		}, is(model.ErrNotFound)},
+		{"unknown new service", func(t *testing.T, s *Store, f moveFixture) error {
+			return s.MergeService(context.Background(), f.from, 9999)
+		}, is(model.ErrNotFound)},
+		{"the new service's agent is revoked", func(t *testing.T, s *Store, f moveFixture) error {
+			_, err := s.RevokeAgent(context.Background(), f.b, t0.Add(3*time.Hour))
+			require.NoError(t, err)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, is(model.ErrRevoked)},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := openTestStore(t)
+			f := newMoveFixture(t, s)
+			var rowsBefore int
+			require.NoError(t, s.pool.QueryRow(context.Background(), `SELECT count(*) FROM services`).Scan(&rowsBefore))
+
+			err := tt.run(t, s, f)
+
+			tt.assert(t, err)
+			var rowsAfter int
+			require.NoError(t, s.pool.QueryRow(context.Background(), `SELECT count(*) FROM services`).Scan(&rowsAfter))
+			assert.Equal(t, rowsBefore, rowsAfter, "nothing is deleted when the merge is refused")
+		})
+	}
+}
+
+func TestMergeService_InventoryAfterTheMerge(t *testing.T) {
+	setup := func(t *testing.T) (*Store, moveFixture) {
+		s := openTestStore(t)
+		f := newMoveFixture(t, s)
+		require.NoError(t, s.MergeService(context.Background(), f.from, f.to))
+		return s, f
+	}
+
+	t.Run("an upsert for the new location updates the survivor", func(t *testing.T) {
+		s, f := setup(t)
+		ctx := context.Background()
+		moved := slusk
+		moved.Digest = digOld
+		require.NoError(t, s.UpsertService(ctx, f.b, "", moved, t0.Add(4*time.Hour)))
+
+		live, err := s.ListServices(ctx)
+		require.NoError(t, err)
+		require.Len(t, live, 1)
+		assert.Equal(t, f.from, live[0].ID)
+		assert.Equal(t, digOld, live[0].ActualDigest)
+	})
+
+	t.Run("a snapshot keeps the survivor", func(t *testing.T) {
+		s, f := setup(t)
+		ctx := context.Background()
+		require.NoError(t, s.ApplySnapshot(ctx, f.b, worker, []protocol.Service{slusk}, t0.Add(4*time.Hour)))
+
+		got, err := s.GetService(ctx, f.from)
+		require.NoError(t, err)
+		assert.Nil(t, got.RemovedAt)
+		live, err := s.ListServices(ctx)
+		require.NoError(t, err)
+		assert.Len(t, live, 1)
+	})
+
+	t.Run("the old agent returning with the stack creates a new row", func(t *testing.T) {
+		s, f := setup(t)
+		ctx := context.Background()
+		require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{slusk}, t0.Add(4*time.Hour)))
+
+		returned := serviceID(t, s, f.a, slusk)
+		assert.NotEqual(t, f.from, returned)
+		got, err := s.GetService(ctx, returned)
+		require.NoError(t, err)
+		assert.Equal(t, model.ModeManual, got.UpdateMode)
+		assert.Empty(t, got.DesiredRef)
+		live, err := s.ListServices(ctx)
+		require.NoError(t, err)
+		assert.Len(t, live, 2)
+	})
+}
+
+func TestMergeSources(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	// A removed service with another name, removed later than the matching one.
+	other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"}
+	require.NoError(t, s.UpsertService(ctx, f.a, "", other, t0))
+	require.NoError(t, s.RemoveService(ctx, f.a, "", other.Key(), t0.Add(5*time.Hour)))
+	otherID := serviceID(t, s, f.a, other)
+	// Other kinds, the updater itself and a live service never show up.
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{adhoc}, t0.Add(6*time.Hour)))
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, nil, t0.Add(7*time.Hour)))
+	selfSvc := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "pu", ComposeService: "agent", Image: "pu:1", Status: "running", Self: true}
+	require.NoError(t, s.UpsertService(ctx, f.a, "", selfSvc, t0))
+	require.NoError(t, s.RemoveService(ctx, f.a, "", selfSvc.Key(), t0.Add(8*time.Hour)))
+
+	got, err := s.MergeSources(ctx, f.to)
+
+	require.NoError(t, err)
+	var ids []int64
+	for _, m := range got {
+		ids = append(ids, m.ID)
+	}
+	assert.Equal(t, []int64{f.from, otherID}, ids, "same name first, then the most recently removed")
+	assert.Equal(t, "a", got[0].AgentName)
+
+	_, err = s.MergeSources(ctx, 9999)
+	assert.ErrorIs(t, err, model.ErrNotFound)
+
+	containerID := serviceID(t, s, f.a, adhoc)
+	got, err = s.MergeSources(ctx, containerID)
+	require.NoError(t, err)
+	assert.Empty(t, got, "plain containers cannot be merged")
+}

internal/master/store/jobs.go

@@ -10,6 +10,7 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 
+	"github.com/shcizo/package-updater/internal/imageref"
 	"github.com/shcizo/package-updater/internal/master/model"
 	"github.com/shcizo/package-updater/internal/protocol"
 )
@@ -90,6 +91,9 @@ func (s *Store) CreateJob(ctx context.Context, j model.NewJob) (model.Job, error
 			}
 			return active, model.ErrJobActive
 		}
+		if errors.As(err, &pgErr) && pgErr.Code == "23503" && pgErr.ConstraintName == "update_jobs_service_id_fkey" {
+			return model.Job{}, model.ErrNotFound
+		}
 		if err != nil {
 			return model.Job{}, fmt.Errorf("create job: %w", err)
 		}
@@ -232,11 +236,11 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job
 		// 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) {
+		if _, d, _ := strings.Cut(f.Ref, "@"); requestedDigest == "" && imageref.IsDigest(d) {
 			tagDigest = d
 		}
 		if _, err := tx.Exec(ctx, `
-			UPDATE services SET desired_ref = $2,
+			UPDATE services SET desired_ref = $2, awaiting_inventory = true,
 				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,

internal/master/store/merge.go

@@ -0,0 +1,249 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// moveCandidatesSQL pairs a removed Compose service with a live one of the
+// same project and service name on another agent. A pair counts only when it
+// is the single one for both rows, and the live row is new (newer than the
+// removed one, never updated, no jobs) while the removed one has something
+// worth moving (a desired ref, auto mode or jobs).
+const moveCandidatesSQL = `
+WITH pairs AS (
+	SELECT f.id AS from_id, t.id AS to_id
+	FROM services f
+	JOIN services t ON t.kind = 'compose' AND t.removed_at IS NULL AND NOT t.self
+	 AND t.compose_project = f.compose_project AND t.compose_service = f.compose_service
+	 AND t.agent_id IS DISTINCT FROM f.agent_id
+	WHERE f.kind = 'compose' AND f.removed_at IS NOT NULL AND NOT f.self
+), uniq AS (
+	SELECT from_id, to_id FROM pairs p
+	WHERE (SELECT count(*) FROM pairs q WHERE q.from_id = p.from_id) = 1
+	  AND (SELECT count(*) FROM pairs q WHERE q.to_id   = p.to_id)   = 1
+)
+SELECT u.from_id, u.to_id, COALESCE(fa.name, ''), COALESCE(ta.name, '')
+FROM uniq u
+JOIN services f ON f.id = u.from_id
+JOIN services t ON t.id = u.to_id
+LEFT JOIN agents fa ON fa.id = f.agent_id
+LEFT JOIN agents ta ON ta.id = t.agent_id
+WHERE t.first_seen_at > f.first_seen_at
+  AND t.desired_ref = ''
+  AND NOT EXISTS (SELECT 1 FROM update_jobs j WHERE j.service_id = t.id)
+  AND (f.desired_ref <> '' OR f.update_mode = 'auto'
+       OR EXISTS (SELECT 1 FROM update_jobs j WHERE j.service_id = f.id))
+ORDER BY u.to_id`
+
+// MoveCandidates returns the removed Compose services that look like they
+// moved to another agent; see model.Move. It only suggests: nothing is merged.
+func (s *Store) MoveCandidates(ctx context.Context) ([]model.Move, error) {
+	rows, err := s.pool.Query(ctx, moveCandidatesSQL)
+	if err != nil {
+		return nil, fmt.Errorf("move candidates: %w", err)
+	}
+	defer rows.Close()
+	var out []model.Move
+	for rows.Next() {
+		var m model.Move
+		if err := rows.Scan(&m.FromID, &m.ToID, &m.FromAgent, &m.ToAgent); err != nil {
+			return nil, fmt.Errorf("scan move: %w", err)
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
+
+// MergeSources returns up to 50 removed services that could be merged into
+// the service toID: the same kind (Compose or Swarm), not the updater itself,
+// those with the same name first, then the most recently removed. Returns
+// model.ErrNotFound if toID does not exist.
+func (s *Store) MergeSources(ctx context.Context, toID int64) ([]model.Service, error) {
+	to, err := s.GetService(ctx, toID)
+	if err != nil {
+		return nil, err
+	}
+	if to.Kind == protocol.KindContainer {
+		return nil, nil
+	}
+	rows, err := s.pool.Query(ctx, `
+		SELECT `+serviceColumns+`
+		FROM services s LEFT JOIN agents a ON a.id = s.agent_id
+		WHERE s.removed_at IS NOT NULL AND s.kind = $1 AND NOT s.self AND s.id <> $2
+		ORDER BY (s.compose_project, s.compose_service, s.swarm_service) = ($3, $4, $5) DESC,
+		         s.removed_at DESC, s.id DESC
+		LIMIT 50`,
+		string(to.Kind), toID, to.ComposeProject, to.ComposeService, to.SwarmService)
+	if err != nil {
+		return nil, fmt.Errorf("merge sources: %w", err)
+	}
+	defer rows.Close()
+	var out []model.Service
+	for rows.Next() {
+		m, err := scanService(rows)
+		if err != nil {
+			return nil, fmt.Errorf("scan service: %w", err)
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
+
+// mergeRow is what MergeService needs to know about one of the two rows.
+type mergeRow struct {
+	kind    string
+	agentID *int64
+	removed bool
+	self    bool
+}
+
+// MergeService moves the live service toID onto the removed service fromID:
+// the old row keeps its id, first_seen_at, update mode and desired ref, takes
+// the location and inventory of the live row (and its available suggestion),
+// becomes live, and the live row is deleted. The jobs of the live row move to
+// the old one. This is the only place a services row is deleted.
+//
+// It returns *model.MergeRefusedError when the pair cannot be merged,
+// model.ErrJobActive when either row has a queued or running job,
+// model.ErrRevoked when the live row's agent is revoked, model.ErrNotFound
+// when a row does not exist, and model.ErrConflict when the live row moved to
+// another agent while the merge started.
+//
+// Locks are taken in the order inventory writes use: the agent row first,
+// then the service rows by id. The live row is deleted before the old one is
+// updated because location_key is unique.
+func (s *Store) MergeService(ctx context.Context, fromID, toID int64) error {
+	if fromID == toID {
+		return &model.MergeRefusedError{Reason: model.ReasonMergeSame}
+	}
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin merge: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	var toAgent *int64
+	err = tx.QueryRow(ctx, `SELECT agent_id FROM services WHERE id = $1`, toID).Scan(&toAgent)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return model.ErrNotFound
+	}
+	if err != nil {
+		return fmt.Errorf("merge: find new service: %w", err)
+	}
+	if toAgent != nil {
+		if err := lockActiveAgent(ctx, tx, *toAgent); err != nil {
+			return err
+		}
+	}
+
+	rows, err := tx.Query(ctx, `
+		SELECT id, kind, agent_id, removed_at IS NOT NULL, self FROM services
+		WHERE id IN ($1, $2) ORDER BY id FOR UPDATE`, fromID, toID)
+	if err != nil {
+		return fmt.Errorf("merge: lock services: %w", err)
+	}
+	locked := map[int64]mergeRow{}
+	for rows.Next() {
+		var id int64
+		var r mergeRow
+		if err := rows.Scan(&id, &r.kind, &r.agentID, &r.removed, &r.self); err != nil {
+			rows.Close()
+			return fmt.Errorf("merge: scan service: %w", err)
+		}
+		locked[id] = r
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("merge: lock services: %w", err)
+	}
+	from, okFrom := locked[fromID]
+	to, okTo := locked[toID]
+	if !okFrom || !okTo {
+		return model.ErrNotFound
+	}
+	if (to.agentID == nil) != (toAgent == nil) || (toAgent != nil && *to.agentID != *toAgent) {
+		return model.ErrConflict
+	}
+	if reason := mergeRefusal(from, to); reason != "" {
+		return &model.MergeRefusedError{Reason: reason}
+	}
+
+	var active bool
+	if err := tx.QueryRow(ctx, `
+		SELECT EXISTS (SELECT 1 FROM update_jobs
+		               WHERE service_id IN ($1, $2) AND status IN ('queued', 'running'))`,
+		fromID, toID).Scan(&active); err != nil {
+		return fmt.Errorf("merge: check active jobs: %w", err)
+	}
+	if active {
+		return model.ErrJobActive
+	}
+
+	if _, err := tx.Exec(ctx, `UPDATE update_jobs SET service_id = $1 WHERE service_id = $2`, fromID, toID); err != nil {
+		return fmt.Errorf("merge: move jobs: %w", err)
+	}
+	var t model.Service
+	var lastTagDigest string
+	err = tx.QueryRow(ctx, `
+		SELECT agent_id, location_key, compose_project, compose_service, working_dir, config_files, built,
+		       swarm_cluster_id, swarm_service, container_name, image_repo, actual_ref, actual_digest, status, self,
+		       updated_at, available_ref, available_digest, available_source, available_at, last_poll_error,
+		       last_tag_digest
+		FROM services WHERE id = $1`, toID).Scan(
+		&t.AgentID, &t.LocationKey, &t.ComposeProject, &t.ComposeService, &t.WorkingDir, &t.ConfigFiles, &t.Built,
+		&t.SwarmClusterID, &t.SwarmService, &t.ContainerName, &t.ImageRepo, &t.ActualRef, &t.ActualDigest, &t.Status, &t.Self,
+		&t.UpdatedAt, &t.AvailableRef, &t.AvailableDigest, &t.AvailableSource, &t.AvailableAt, &t.LastPollError,
+		&lastTagDigest)
+	if err != nil {
+		return fmt.Errorf("merge: read new service: %w", err)
+	}
+	if _, err := tx.Exec(ctx, `DELETE FROM services WHERE id = $1`, toID); err != nil {
+		return fmt.Errorf("merge: delete new service: %w", err)
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE services SET
+			agent_id = $2, location_key = $3, compose_project = $4, compose_service = $5, working_dir = $6,
+			config_files = $7, built = $8, swarm_cluster_id = $9, swarm_service = $10, container_name = $11,
+			image_repo = $12, actual_ref = $13, actual_digest = $14, status = $15, self = $16, updated_at = $17,
+			available_ref = $18, available_digest = $19, available_source = $20, available_at = $21,
+			last_poll_error = $22,
+			last_tag_digest = CASE WHEN last_tag_digest <> '' THEN last_tag_digest ELSE $23 END,
+			removed_at = NULL, awaiting_inventory = false
+		WHERE id = $1`,
+		fromID, t.AgentID, t.LocationKey, t.ComposeProject, t.ComposeService, t.WorkingDir,
+		t.ConfigFiles, t.Built, t.SwarmClusterID, t.SwarmService, t.ContainerName,
+		t.ImageRepo, t.ActualRef, t.ActualDigest, t.Status, t.Self, t.UpdatedAt,
+		t.AvailableRef, t.AvailableDigest, t.AvailableSource, t.AvailableAt,
+		t.LastPollError, lastTagDigest); err != nil {
+		return fmt.Errorf("merge: update old service: %w", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit merge: %w", err)
+	}
+	return nil
+}
+
+// mergeRefusal returns why from (the old row) cannot be merged with to (the
+// live one), or "".
+func mergeRefusal(from, to mergeRow) string {
+	switch {
+	case !from.removed:
+		return model.ReasonMergeLiveOld
+	case to.removed:
+		return model.ReasonMergeGoneNew
+	case from.kind != to.kind:
+		return model.ReasonMergeKind
+	case from.kind == string(protocol.KindContainer):
+		return model.ReasonMergeContainer
+	case from.self || to.self:
+		return model.ReasonSelf
+	}
+	return ""
+}

internal/master/store/migrations/0005_drift_merge.sql

@@ -0,0 +1,7 @@
+-- True from a successful job until the next inventory write for the row. The
+-- agent's upsert arrives a debounce interval after job.result, and until then
+-- desired_ref is newer than actual_ref without the service having drifted.
+ALTER TABLE services ADD COLUMN awaiting_inventory boolean NOT NULL DEFAULT false;
+
+-- Move suggestions look up Compose rows by project and service across agents.
+CREATE INDEX services_by_compose_name ON services (compose_project, compose_service) WHERE kind = 'compose';

internal/master/store/services.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"regexp"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -55,10 +54,8 @@ ON CONFLICT (location_key) DO UPDATE SET
 	status        = EXCLUDED.status,
 	self          = EXCLUDED.self,
 	updated_at    = EXCLUDED.updated_at,
-	removed_at    = NULL`
+	removed_at    = NULL,
-
+	awaiting_inventory = false`
-// 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)
@@ -75,7 +72,7 @@ func upsertService(ctx context.Context, db execer, agentID int64, clusterID stri
 	if err != nil {
 		return fmt.Errorf("upsert service %s: %w", svc.Key(), err)
 	}
-	if digestPattern.MatchString(svc.Digest) {
+	if imageref.IsDigest(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 == "" {
@@ -178,7 +175,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.last_tag_digest, s.first_seen_at, s.updated_at, s.removed_at`
+	s.last_poll_error, s.last_tag_digest, s.awaiting_inventory, s.first_seen_at, s.updated_at, s.removed_at`
 
 func scanService(row pgx.Row) (model.Service, error) {
 	var m model.Service
@@ -187,7 +184,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.LastTagDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
+		&m.LastPollError, &m.LastTagDigest, &m.AwaitingInventory, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
 	m.Kind = protocol.Kind(kind)
 	return m, err
 }
@@ -262,8 +259,8 @@ func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error
 }
 
 // SetLastTagDigest sets last_tag_digest; see model.Service.LastTagDigest. The
-// poller calls it, and FinishJob sets the same column after a tag job.
+// poller and jobs.RestoreDesired call it, and FinishJob sets the same column
-// Inventory writes never change it.
+// 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 {
@@ -331,6 +328,34 @@ func (s *Store) SetUpdateMode(ctx context.Context, id int64, mode string) error
 	return nil
 }
 
+// AcceptRunning makes the running version the desired one: desired_ref =
+// desired, provided the service is live, has been updated before and still
+// reports actualRef and actualDigest (what the caller saw when it decided).
+// It starts no job and leaves last_tag_digest, the available suggestion and
+// the update mode alone. Returns model.ErrNotFound for an unknown or removed
+// service and model.ErrConflict when the service was never updated or the
+// running version changed meanwhile.
+func (s *Store) AcceptRunning(ctx context.Context, id int64, actualRef, actualDigest, desired string) error {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE services SET desired_ref = $4, awaiting_inventory = false
+		WHERE id = $1 AND removed_at IS NULL AND desired_ref <> ''
+		  AND actual_ref = $2 AND actual_digest = $3`, id, actualRef, actualDigest, desired)
+	if err != nil {
+		return fmt.Errorf("accept running: %w", err)
+	}
+	if tag.RowsAffected() == 1 {
+		return nil
+	}
+	svc, err := s.GetService(ctx, id)
+	if err != nil {
+		return err
+	}
+	if svc.RemovedAt != nil {
+		return model.ErrNotFound
+	}
+	return model.ErrConflict
+}
+
 // GetService returns one service, including a removed one (RemovedAt set).
 // Returns model.ErrNotFound if the ID does not exist.
 func (s *Store) GetService(ctx context.Context, id int64) (model.Service, error) {

internal/master/web/drift_test.go

@@ -0,0 +1,297 @@
+package web
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/master/events"
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+const (
+	driftDesired = "sha256:1111111111111111111111111111111111111111111111111111111111111111"
+	driftActual  = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
+)
+
+type acceptCall struct {
+	id                                  int64
+	actualRef, actualDigest, desiredRef string
+}
+
+func (f *fakeStore) AcceptRunning(_ context.Context, id int64, actualRef, actualDigest, desired string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.acceptCalls = append(f.acceptCalls, acceptCall{id, actualRef, actualDigest, desired})
+	return f.acceptErr
+}
+
+func (f *fakeStore) MoveCandidates(context.Context) ([]model.Move, error) { return f.moves, nil }
+
+func (f *fakeStore) MergeSources(context.Context, int64) ([]model.Service, error) {
+	return f.sources, nil
+}
+
+func (f *fakeStore) MergeService(_ context.Context, from, to int64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.merged = append(f.merged, [2]int64{from, to})
+	return f.mergeErr
+}
+
+func (f *fakeJobs) Restorable(model.Service) error { return f.restorable }
+
+func (f *fakeJobs) RestoreDesired(_ context.Context, serviceID int64, by string) (model.Job, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.requestedBy = by
+	f.restored = append(f.restored, serviceID)
+	return f.job, f.err
+}
+
+// driftService makes the first fixture service a drifted one: updated to
+// 1.4.2 at driftDesired, now running 1.4.2 at driftActual.
+func driftService(store *fakeStore) {
+	store.services[0].DesiredRef = "gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftDesired
+	store.services[0].ActualDigest = driftActual
+}
+
+func postTo(t *testing.T, srv *httptest.Server, path string, form url.Values) (*http.Response, string) {
+	t.Helper()
+	form.Set("csrf_token", "csrf")
+	resp, err := noRedirectClient().PostForm(srv.URL+path, form)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	return resp, string(body)
+}
+
+func TestRestoreDesired(t *testing.T) {
+	tests := []struct {
+		name         string
+		job          model.Job
+		err          error
+		path         string
+		wantCode     int
+		wantLocation string
+		wantBody     string
+	}{
+		{name: "started", job: model.Job{ID: 42}, path: "/services/1/restore",
+			wantCode: http.StatusSeeOther, wantLocation: "/jobs/42"},
+		{name: "already running", job: model.Job{ID: 41}, err: model.ErrJobActive, path: "/services/1/restore",
+			wantCode: http.StatusSeeOther, wantLocation: "/jobs/41"},
+		{name: "refused", err: &model.NotUpdatableError{Reason: model.ReasonNotDrifted}, path: "/services/1/restore",
+			wantCode: http.StatusConflict, wantBody: "Cannot restore: runs its desired version."},
+		{name: "unknown service", err: model.ErrNotFound, path: "/services/1/restore", wantCode: http.StatusNotFound},
+		{name: "invalid id", path: "/services/x/restore", wantCode: http.StatusBadRequest},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srv, _, _, _, jobs := newTestEnv(t)
+			jobs.job, jobs.err = tt.job, tt.err
+
+			resp, body := postTo(t, srv, tt.path, url.Values{})
+
+			assert.Equal(t, tt.wantCode, resp.StatusCode)
+			assert.Equal(t, tt.wantLocation, resp.Header.Get("Location"))
+			assert.Contains(t, body, tt.wantBody)
+			if tt.wantCode != http.StatusBadRequest {
+				assert.Equal(t, "alice", jobs.triggeredBy())
+			}
+		})
+	}
+}
+
+func TestAcceptRunning_WritesRunningRefAndPublishes(t *testing.T) {
+	srv, store, _, broker, _ := newTestEnv(t)
+	driftService(store)
+	ch, cancel := broker.Subscribe()
+	defer cancel()
+
+	resp, _ := postTo(t, srv, "/services/1/accept", url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}})
+
+	assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
+	assert.Equal(t, "/services/1", resp.Header.Get("Location"))
+	assert.Equal(t, []acceptCall{{
+		id: 1, actualRef: "gitea.shcizo.se/shcizo/slusk:1.4.2", actualDigest: driftActual,
+		desiredRef: "gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual,
+	}}, store.acceptCalls, "the value written comes from the row, not the form")
+	select {
+	case topic := <-ch:
+		assert.Equal(t, events.Services, topic)
+	case <-time.After(time.Second):
+		t.Fatal("no services signal")
+	}
+}
+
+func TestAcceptRunning_ReturnToOverview(t *testing.T) {
+	srv, store, _, _, _ := newTestEnv(t)
+	driftService(store)
+
+	resp, _ := postTo(t, srv, "/services/1/accept",
+		url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}, "return": {"overview"}})
+
+	assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
+	assert.Equal(t, "/", resp.Header.Get("Location"))
+}
+
+func TestAcceptRunning_Refusals(t *testing.T) {
+	tests := []struct {
+		name       string
+		path       string
+		form       url.Values
+		notDrifted bool
+		removed    bool
+		acceptErr  error
+		wantCode   int
+		wantBody   string
+	}{
+		{name: "missing actual", path: "/services/1/accept", form: url.Values{}, wantCode: http.StatusBadRequest},
+		{name: "stale actual", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.3.0"}},
+			wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
+		{name: "same ref, other digest", path: "/services/1/accept",
+			form:     url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftDesired}},
+			wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
+		{name: "not drifted", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
+			notDrifted: true, wantCode: http.StatusConflict, wantBody: "Cannot accept: the service runs its desired version."},
+		{name: "changed under our feet", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
+			acceptErr: model.ErrConflict, wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
+		{name: "unknown service", path: "/services/9/accept", form: url.Values{"actual": {"x"}}, wantCode: http.StatusNotFound},
+		{name: "removed", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
+			removed: true, wantCode: http.StatusNotFound},
+		{name: "removed meanwhile", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
+			acceptErr: model.ErrNotFound, wantCode: http.StatusNotFound},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srv, store, _, _, _ := newTestEnv(t)
+			if !tt.notDrifted {
+				driftService(store)
+			}
+			if tt.removed {
+				removedAt := time.Now()
+				store.services[0].RemovedAt = &removedAt
+			}
+			store.acceptErr = tt.acceptErr
+
+			resp, body := postTo(t, srv, tt.path, tt.form)
+
+			assert.Equal(t, tt.wantCode, resp.StatusCode)
+			assert.Contains(t, body, tt.wantBody)
+			if tt.acceptErr == nil {
+				assert.Empty(t, store.acceptCalls)
+			}
+		})
+	}
+}
+
+func TestMergeService_RedirectsToTheSurvivorAndPublishes(t *testing.T) {
+	srv, store, _, broker, _ := newTestEnv(t)
+	ch, cancel := broker.Subscribe()
+	defer cancel()
+
+	resp, _ := postTo(t, srv, "/services/1/merge", url.Values{"from": {"17"}})
+
+	assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
+	assert.Equal(t, "/services/17", resp.Header.Get("Location"))
+	assert.Equal(t, [][2]int64{{17, 1}}, store.merged)
+	var got []events.Topic
+	for len(got) < 2 {
+		select {
+		case topic := <-ch:
+			got = append(got, topic)
+		case <-time.After(time.Second):
+			t.Fatalf("signals so far: %v", got)
+		}
+	}
+	assert.ElementsMatch(t, []events.Topic{events.Services, events.Jobs}, got)
+}
+
+func TestMergeService_Refusals(t *testing.T) {
+	tests := []struct {
+		name     string
+		id, from string
+		err      error
+		wantCode int
+		wantBody string
+	}{
+		{name: "invalid from", id: "1", from: "x", wantCode: http.StatusBadRequest},
+		{name: "empty from", id: "1", from: "", wantCode: http.StatusBadRequest},
+		{name: "refused", id: "1", from: "17", err: &model.MergeRefusedError{Reason: model.ReasonMergeLiveOld},
+			wantCode: http.StatusConflict, wantBody: "Cannot merge: the old service still exists."},
+		{name: "job active", id: "1", from: "17", err: model.ErrJobActive,
+			wantCode: http.StatusConflict, wantBody: "Cannot merge: an update is already in progress for this service."},
+		{name: "agent revoked", id: "1", from: "17", err: model.ErrRevoked, wantCode: http.StatusConflict, wantBody: "Cannot merge:"},
+		{name: "not found", id: "1", from: "17", err: model.ErrNotFound, wantCode: http.StatusNotFound},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			srv, store, _, _, _ := newTestEnv(t)
+			store.mergeErr = tt.err
+
+			resp, body := postTo(t, srv, "/services/"+tt.id+"/merge", url.Values{"from": {tt.from}})
+
+			assert.Equal(t, tt.wantCode, resp.StatusCode)
+			assert.Contains(t, body, tt.wantBody)
+			assert.Empty(t, resp.Header.Get("Location"))
+		})
+	}
+}
+
+// serviceDetail's data is what the service template reads.
+func TestServiceDetail_DriftAndMoveData(t *testing.T) {
+	removed := time.Now()
+	live := model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", ActualRef: "slusk:1"}
+	gone := model.Service{ID: 2, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", ActualRef: "slusk:1", RemovedAt: &removed}
+	swarm := model.Service{ID: 3, Kind: protocol.KindSwarm, SwarmService: "gitea", ActualRef: "gitea:1"}
+	container := model.Service{ID: 4, Kind: protocol.KindContainer, ContainerName: "adhoc", ActualRef: "redis:7"}
+	self := model.Service{ID: 5, Kind: protocol.KindCompose, ComposeProject: "pu", ComposeService: "agent", ActualRef: "pu:1", Self: true}
+	move := model.Move{FromID: 2, ToID: 1, FromAgent: "old", ToAgent: "new"}
+	sources := []model.Service{gone}
+
+	tests := []struct {
+		name        string
+		id          int64
+		restorable  error
+		wantBlocked string
+		wantFrom    *model.Move
+		wantTo      *model.Move
+		wantSources []model.Service
+	}{
+		{name: "live row gets the suggestion and merge choices", id: 1, wantFrom: &move, wantSources: sources},
+		{name: "removed row gets where it went", id: 2, wantTo: &move},
+		{name: "swarm service can merge, no suggestion", id: 3, wantSources: sources},
+		{name: "container cannot merge", id: 4},
+		{name: "self cannot merge", id: 5},
+		{name: "restore refusal is passed on", id: 1, restorable: &model.NotUpdatableError{Reason: model.ReasonSelf},
+			wantBlocked: model.ReasonSelf, wantFrom: &move, wantSources: sources},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			store := &fakeStore{services: []model.Service{live, gone, swarm, container, self},
+				moves: []model.Move{move}, sources: sources}
+			jobs := &fakeJobs{restorable: tt.restorable}
+			s, err := New(store, &fakeHub{}, jobs, events.NewBroker(), Config{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
+			require.NoError(t, err)
+			req := httptest.NewRequest(http.MethodGet, "/services/"+strconv.FormatInt(tt.id, 10), nil)
+
+			data, err := s.serviceDetail(req, tt.id)
+
+			require.NoError(t, err)
+			assert.Equal(t, tt.wantBlocked, data.RestoreBlocked)
+			assert.Equal(t, tt.wantFrom, data.MoveFrom)
+			assert.Equal(t, tt.wantTo, data.MoveTo)
+			assert.Equal(t, tt.wantSources, data.MergeSources)
+		})
+	}
+}

internal/master/web/server.go

@@ -18,6 +18,7 @@ import (
 	"github.com/shcizo/package-updater/internal/master/events"
 	"github.com/shcizo/package-updater/internal/master/model"
 	"github.com/shcizo/package-updater/internal/master/webauth"
+	"github.com/shcizo/package-updater/internal/protocol"
 )
 
 //go:embed templates/*.html static/*
@@ -40,6 +41,16 @@ type Store interface {
 	RecentVersions(ctx context.Context, repo string, limit int) ([]model.ImageVersion, error)
 	// SetUpdateMode returns model.ErrNotFound for an unknown or removed service.
 	SetUpdateMode(ctx context.Context, id int64, mode string) error
+	// AcceptRunning sets the desired ref of a drifted service; model.ErrNotFound
+	// for an unknown or removed service, model.ErrConflict when the running
+	// version is no longer actualRef@actualDigest.
+	AcceptRunning(ctx context.Context, id int64, actualRef, actualDigest, desired string) error
+	// MoveCandidates returns the suggested moves of removed services.
+	MoveCandidates(ctx context.Context) ([]model.Move, error)
+	// MergeSources returns the removed services that could be merged into toID.
+	MergeSources(ctx context.Context, toID int64) ([]model.Service, error)
+	// MergeService moves the live service toID onto the removed fromID.
+	MergeService(ctx context.Context, fromID, toID int64) error
 	GetJob(ctx context.Context, id int64) (model.Job, error)
 	ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error)
 	ListAgents(ctx context.Context) ([]model.Agent, error)
@@ -63,6 +74,10 @@ type Jobs interface {
 	// TriggerVersion updates to exactly the image version versionID of the
 	// service's repo; model.ErrNotFound for an unknown service or version.
 	TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error)
+	// Restorable explains why a drifted service cannot be restored.
+	Restorable(svc model.Service) error
+	// RestoreDesired deploys the service's desired ref again.
+	RestoreDesired(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error)
 }
 
 // Broker delivers change signals to open streams and accepts signals from
@@ -118,6 +133,9 @@ func (s *Server) Register(mux *http.ServeMux, protect func(http.Handler) http.Ha
 	route("GET /fragments/services/{id}/jobs", s.serviceJobsFragment)
 	route("POST /services/{id}/update", s.updateService)
 	route("POST /services/{id}/mode", s.setMode)
+	route("POST /services/{id}/restore", s.restoreDesired)
+	route("POST /services/{id}/accept", s.acceptRunning)
+	route("POST /services/{id}/merge", s.mergeService)
 	route("GET /jobs", s.jobsPage)
 	route("GET /fragments/jobs", s.jobsFragment)
 	route("GET /jobs/{id}", s.jobPage)
@@ -145,8 +163,14 @@ type pageData struct {
 	Versions    []versionRow
 	Polling     string // what the service page says about registry polling
 	PollFailed  bool   // Polling is the last registry error
-	Jobs        []model.Job
+	// RestoreBlocked is why a drifted Service cannot be restored; "" when it
-	Job         model.Job
+	// can. Only meaningful when Service.Drifted().
+	RestoreBlocked string
+	MoveFrom       *model.Move     // a live Service: the removed service it probably replaces
+	MoveTo         *model.Move     // a removed Service: the live service it probably moved to
+	MergeSources   []model.Service // a live Compose or Swarm Service: removed ones to merge in
+	Jobs           []model.Job
+	Job            model.Job
 }
 
 func (s *Server) base(r *http.Request, title string) pageData {
@@ -235,6 +259,14 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) {
 		return data, err
 	}
 	data.AutoBlocked = s.autoBlocked(svc)
+	if err := s.jobs.Restorable(svc); errors.As(err, &nu) {
+		data.RestoreBlocked = nu.Reason
+	} else if err != nil {
+		return data, err
+	}
+	if err := s.moveData(r, &data); err != nil {
+		return data, err
+	}
 	versions, err := s.store.RecentVersions(r.Context(), svc.ImageRepo, versionsLimit)
 	if err != nil {
 		return data, err
@@ -245,6 +277,27 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) {
 	return data, err
 }
 
+// moveData fills the move suggestion and the merge choices of the service page.
+func (s *Server) moveData(r *http.Request, data *pageData) error {
+	svc := data.Service
+	moves, err := s.store.MoveCandidates(r.Context())
+	if err != nil {
+		return err
+	}
+	for i, m := range moves {
+		switch svc.ID {
+		case m.ToID:
+			data.MoveFrom = &moves[i]
+		case m.FromID:
+			data.MoveTo = &moves[i]
+		}
+	}
+	if svc.RemovedAt == nil && !svc.Self && svc.Kind != protocol.KindContainer {
+		data.MergeSources, err = s.store.MergeSources(r.Context(), svc.ID)
+	}
+	return err
+}
+
 func (s *Server) servicePage(w http.ResponseWriter, r *http.Request) {
 	id, ok := pathID(w, r)
 	if !ok {
@@ -296,18 +349,130 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) {
 	case err == nil || errors.Is(err, model.ErrJobActive):
 		http.Redirect(w, r, "/jobs/"+strconv.FormatInt(job.ID, 10), http.StatusSeeOther)
 	case errors.As(err, &nu):
-		data, derr := s.serviceDetail(r, id)
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot update: "+nu.Reason+".")
-		if derr != nil {
-			s.failLookup(w, derr)
-			return
-		}
-		data.Error = "Cannot update: " + nu.Reason + "."
-		s.render(w, http.StatusConflict, "service.html", "layout", data)
 	default:
 		s.failLookup(w, err)
 	}
 }
 
+// restoreDesired deploys the desired ref of a drifted service again.
+func (s *Server) restoreDesired(w http.ResponseWriter, r *http.Request) {
+	id, ok := pathID(w, r)
+	if !ok {
+		return
+	}
+	sess, _ := webauth.SessionFrom(r.Context())
+	job, err := s.jobs.RestoreDesired(r.Context(), id, sess.Subject)
+	var nu *model.NotUpdatableError
+	switch {
+	case err == nil || errors.Is(err, model.ErrJobActive):
+		http.Redirect(w, r, "/jobs/"+strconv.FormatInt(job.ID, 10), http.StatusSeeOther)
+	case errors.As(err, &nu):
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot restore: "+nu.Reason+".")
+	default:
+		s.failLookup(w, err)
+	}
+}
+
+// acceptRunning makes the running version of a drifted service the desired
+// one. The form's actual field is the running ref the user saw (Service.RunningRef);
+// it only guards against the service having changed since, the value written
+// comes from the row.
+func (s *Server) acceptRunning(w http.ResponseWriter, r *http.Request) {
+	id, ok := pathID(w, r)
+	if !ok {
+		return
+	}
+	actual := r.PostFormValue("actual")
+	if actual == "" {
+		http.Error(w, "missing actual", http.StatusBadRequest)
+		return
+	}
+	svc, err := s.store.GetService(r.Context(), id)
+	if err == nil && svc.RemovedAt != nil {
+		err = model.ErrNotFound
+	}
+	if err != nil {
+		s.failLookup(w, err)
+		return
+	}
+	if !svc.Drifted() {
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot accept: the service runs its desired version.")
+		return
+	}
+	if actual != svc.RunningRef() {
+		s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.")
+		return
+	}
+	desired := svc.RunningRef()
+	err = s.store.AcceptRunning(r.Context(), id, svc.ActualRef, svc.ActualDigest, desired)
+	switch {
+	case errors.Is(err, model.ErrConflict):
+		s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.")
+		return
+	case err != nil:
+		s.failLookup(w, err)
+		return
+	}
+	sess, _ := webauth.SessionFrom(r.Context())
+	s.log.Info("drift_accepted", "service_id", id, "subject", sess.Subject, "desired", desired)
+	s.broker.Publish(events.Services)
+	if r.PostFormValue("return") == "overview" {
+		http.Redirect(w, r, "/", http.StatusSeeOther)
+		return
+	}
+	http.Redirect(w, r, "/services/"+strconv.FormatInt(id, 10), http.StatusSeeOther)
+}
+
+// mergeService moves the live service {id} onto the removed service in the
+// form field "from", which keeps its history, mode and desired ref.
+func (s *Server) mergeService(w http.ResponseWriter, r *http.Request) {
+	id, ok := pathID(w, r)
+	if !ok {
+		return
+	}
+	from, err := strconv.ParseInt(r.PostFormValue("from"), 10, 64)
+	if err != nil {
+		http.Error(w, "invalid service", http.StatusBadRequest)
+		return
+	}
+	err = s.store.MergeService(r.Context(), from, id)
+	var refused *model.MergeRefusedError
+	switch {
+	case errors.As(err, &refused):
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: "+refused.Reason+".")
+		return
+	case errors.Is(err, model.ErrJobActive):
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: "+model.ErrJobActive.Error()+".")
+		return
+	case errors.Is(err, model.ErrRevoked):
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: the agent is revoked.")
+		return
+	case errors.Is(err, model.ErrConflict):
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: the service changed; check it again.")
+		return
+	case err != nil:
+		s.failLookup(w, err)
+		return
+	}
+	sess, _ := webauth.SessionFrom(r.Context())
+	s.log.Info("service_merged", "from_id", from, "to_id", id, "subject", sess.Subject)
+	s.broker.Publish(events.Services)
+	s.broker.Publish(events.Jobs)
+	http.Redirect(w, r, "/services/"+strconv.FormatInt(from, 10), http.StatusSeeOther)
+}
+
+// renderServiceError renders the service page of id with msg as its error.
+func (s *Server) renderServiceError(w http.ResponseWriter, r *http.Request, id int64, status int, msg string) {
+	data, err := s.serviceDetail(r, id)
+	if err != nil {
+		s.failLookup(w, err)
+		return
+	}
+	data.Error = msg
+	s.render(w, status, "service.html", "layout", data)
+}
+
 // setMode switches a service between manual and auto. Manual is always
 // allowed; auto only when jobs.AutoAllowed says so.
 func (s *Server) setMode(w http.ResponseWriter, r *http.Request) {
@@ -331,13 +496,7 @@ func (s *Server) setMode(w http.ResponseWriter, r *http.Request) {
 		}
 		var nu *model.NotUpdatableError
 		if err := s.jobs.AutoAllowed(svc); errors.As(err, &nu) {
-			data, derr := s.serviceDetail(r, id)
+			s.renderServiceError(w, r, id, http.StatusConflict, "Cannot enable auto: "+nu.Reason+".")
-			if derr != nil {
-				s.failLookup(w, derr)
-				return
-			}
-			data.Error = "Cannot enable auto: " + nu.Reason + "."
-			s.render(w, http.StatusConflict, "service.html", "layout", data)
 			return
 		} else if err != nil {
 			s.fail(w, err)

internal/master/web/server_test.go

@@ -35,6 +35,13 @@ type fakeStore struct {
 	revokedID    int64
 	revokeFailed []int64 // job IDs RevokeAgent reports as failed
 	versions     []model.ImageVersion
+
+	acceptCalls []acceptCall
+	acceptErr   error
+	moves       []model.Move
+	sources     []model.Service
+	merged      [][2]int64 // from, to
+	mergeErr    error
 }
 
 func (f *fakeStore) RecentVersions(_ context.Context, repo string, limit int) ([]model.ImageVersion, error) {
@@ -142,6 +149,9 @@ type fakeJobs struct {
 	triggered   []int64 // version IDs passed to TriggerVersion
 	triggerSvc  []int64 // service IDs passed to TriggerVersion
 	plainCalls  int     // Trigger calls
+
+	restorable error   // returned by Restorable
+	restored   []int64 // service IDs passed to RestoreDesired
 }
 
 func (f *fakeJobs) Updatable(model.Service) error { return f.updatable }