From d0726ca992fc8976f20e81ba7bdd86145f1bc0a6 Mon Sep 17 00:00:00 2001
From: Samuel Enocsson <samuel.enocsson@gmail.com>
Date: Tue, 13 Oct 2026 08:07:45 +0200
Subject: [PATCH] feat(master,agent): deploy digest-pinned refs and route them
 to protocol 3 agents (#9)

---
 internal/agent/updater/compose.go      |   4 +-
 internal/agent/updater/compose_test.go |  18 +++
 internal/agent/updater/resolve.go      |  21 +++-
 internal/agent/updater/resolve_test.go |  15 ++-
 internal/agent/updater/swarm.go        |   4 +-
 internal/agent/updater/swarm_test.go   |  16 +++
 internal/e2e/agent_master_test.go      |   4 +-
 internal/e2e/jobs_test.go              |   4 +-
 internal/e2e/webhook_test.go           |   3 +-
 internal/master/hub/jobs_test.go       |   4 +-
 internal/master/hub/route.go           |  13 +-
 internal/master/hub/route_test.go      |  39 +++---
 internal/master/jobs/jobs.go           |  41 ++++++-
 internal/master/jobs/jobs_test.go      | 157 ++++++++++++++++++++++++-
 internal/protocol/protocol.go          |  13 +-
 15 files changed, 303 insertions(+), 53 deletions(-)

diff --git a/internal/agent/updater/compose.go b/internal/agent/updater/compose.go
index 1e20aac..74eddff 100644
--- a/internal/agent/updater/compose.go
+++ b/internal/agent/updater/compose.go
@@ -91,7 +91,7 @@ func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run
 	return &ComposeExecutor{cli: cli, resolve: resolve, selfID: selfID, run: run, build: NewBuildChecker(cli, selfID, runOut)}
 }
 
-// Execute pins job.Ref to its current digest, writes it to the stack's
+// Execute pins job.Ref to its current digest (a ref that has one is used as is), writes it to the stack's
 // override file and recreates the service. Working directory and config
 // files come from the service's container labels on this machine, never from
 // the job. It returns the pinned ref.
@@ -150,7 +150,7 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, ou
 	if err != nil {
 		return "", err
 	}
-	fmt.Fprintf(out, "resolved %s\n", pinned)
+	fmt.Fprintln(out, resolveNote(job.Ref, pinned))
 
 	if _, err := WriteOverride(dir, service, pinned); err != nil {
 		return "", err
diff --git a/internal/agent/updater/compose_test.go b/internal/agent/updater/compose_test.go
index ab9b637..0b9f9cb 100644
--- a/internal/agent/updater/compose_test.go
+++ b/internal/agent/updater/compose_test.go
@@ -108,6 +108,24 @@ func TestComposeExecutor(t *testing.T) {
 		assert.Equal(t, pinned, serviceImage(doc, "web"))
 		assert.Contains(t, out.String(), "resolved "+pinned)
 	})
+	t.Run("a pinned ref is written verbatim and not resolved", func(t *testing.T) {
+		t.Cleanup(cleanOverride)
+		for _, ref := range []string{"gitea.shcizo.se/shcizo/slusk:1.3@" + d, "gitea.shcizo.se/shcizo/slusk@" + d} {
+			rec := &recorder{}
+			dist := &fakeDist{digest: "sha256:other"}
+			var out strings.Builder
+			pinnedJob := job
+			pinnedJob.Ref = ref
+			got, err := setup(fakeCompose{rw, []container.Summary{ct}}, dist, rec).Execute(context.Background(), pinnedJob, &out)
+			require.NoError(t, err)
+			assert.Equal(t, ref, got)
+			assert.Empty(t, dist.refs, "no registry lookup")
+			_, doc := readOverride(t, stack)
+			assert.Equal(t, ref, serviceImage(doc, "web"))
+			assert.Contains(t, out.String(), "pinned "+ref)
+			assert.NotContains(t, out.String(), "resolved")
+		}
+	})
 	t.Run("built from source is refused before anything is written", func(t *testing.T) {
 		t.Cleanup(cleanOverride)
 		rec := &recorder{}
diff --git a/internal/agent/updater/resolve.go b/internal/agent/updater/resolve.go
index a3ddab2..677b04e 100644
--- a/internal/agent/updater/resolve.go
+++ b/internal/agent/updater/resolve.go
@@ -16,7 +16,8 @@ type DistributionClient interface {
 }
 
 // Resolver turns "repo:tag" into "repo:tag@digest" by asking the local
-// daemon, which contacts the registry.
+// daemon, which contacts the registry. A ref that already has a digest is
+// passed through.
 type Resolver struct {
 	cli  DistributionClient
 	auth *RegistryAuth
@@ -27,13 +28,15 @@ func NewResolver(cli DistributionClient, auth *RegistryAuth) *Resolver {
 	return &Resolver{cli: cli, auth: auth}
 }
 
-// Resolve returns ref pinned to the digest its tag currently points at. It
-// fails for a ref without a tag or with a digest already.
+// Resolve returns ref pinned to the digest its tag currently points at. A ref
+// that already has a digest ("repo:tag@digest" or "repo@digest") is returned
+// unchanged without asking the daemon. It fails for a ref with neither tag nor
+// digest.
 func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) {
 	p := imageref.Parse(ref)
 	switch {
 	case p.Digest != "":
-		return "", fmt.Errorf("image %q already has a digest", ref)
+		return ref, nil
 	case p.Tag == "":
 		return "", fmt.Errorf("image %q has no tag to resolve", ref)
 	}
@@ -46,3 +49,13 @@ func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) {
 	}
 	return ref + "@" + info.Descriptor.Digest.String(), nil
 }
+
+// resolveNote is the job-output line that says which ref is deployed:
+// "resolved" when pinned came from resolving a tag, "pinned" when ref already
+// carried its digest.
+func resolveNote(ref, pinned string) string {
+	if pinned == ref {
+		return "pinned " + pinned
+	}
+	return "resolved " + pinned
+}
diff --git a/internal/agent/updater/resolve_test.go b/internal/agent/updater/resolve_test.go
index a6eab28..e84bb59 100644
--- a/internal/agent/updater/resolve_test.go
+++ b/internal/agent/updater/resolve_test.go
@@ -21,14 +21,21 @@ func TestResolver(t *testing.T) {
 		assert.Equal(t, []string{"gitea.shcizo.se/shcizo/slusk:1.4"}, dist.refs)
 		assert.Equal(t, []string{"enc-gitea"}, dist.auths)
 	})
-	t.Run("digest or missing tag is refused without asking the daemon", func(t *testing.T) {
-		for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "repo"} {
+	t.Run("a ref with a digest passes through without asking the daemon", func(t *testing.T) {
+		for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "localhost:5000/repo:1@sha256:x"} {
 			dist := &fakeDist{digest: d}
-			_, err := NewResolver(dist, nil).Resolve(context.Background(), ref)
-			assert.Error(t, err, ref)
+			got, err := NewResolver(dist, nil).Resolve(context.Background(), ref)
+			require.NoError(t, err, ref)
+			assert.Equal(t, ref, got)
 			assert.Empty(t, dist.refs)
 		}
 	})
+	t.Run("a bare ref is refused without asking the daemon", func(t *testing.T) {
+		dist := &fakeDist{digest: d}
+		_, err := NewResolver(dist, nil).Resolve(context.Background(), "repo")
+		assert.ErrorContains(t, err, "has no tag to resolve")
+		assert.Empty(t, dist.refs)
+	})
 	t.Run("daemon error is wrapped", func(t *testing.T) {
 		cause := errors.New("unauthorized")
 		_, err := NewResolver(&fakeDist{err: cause}, nil).Resolve(context.Background(), "repo:1")
diff --git a/internal/agent/updater/swarm.go b/internal/agent/updater/swarm.go
index 63e2ffd..d644c2d 100644
--- a/internal/agent/updater/swarm.go
+++ b/internal/agent/updater/swarm.go
@@ -35,7 +35,7 @@ func NewSwarmExecutor(cli SwarmClient, resolve *Resolver, auth *RegistryAuth) *S
 }
 
 // Execute checks that the local daemon is a manager of the job's cluster,
-// pins job.Ref to its current digest and updates the service to it. The
+// pins job.Ref to its current digest (a ref that has one is used as is) and updates the service to it. The
 // update is addressed by the ID and version of the service that was
 // inspected, so it changes exactly that object. Success means the daemon
 // accepted the new spec, not that the tasks have converged. It returns the
@@ -71,7 +71,7 @@ func (e *SwarmExecutor) Execute(ctx context.Context, job protocol.JobStart, out
 	if err != nil {
 		return "", err
 	}
-	fmt.Fprintf(out, "resolved %s\n", pinned)
+	fmt.Fprintln(out, resolveNote(job.Ref, pinned))
 
 	spec := svc.Spec
 	spec.TaskTemplate.ContainerSpec.Image = pinned
diff --git a/internal/agent/updater/swarm_test.go b/internal/agent/updater/swarm_test.go
index d44ddfa..429248f 100644
--- a/internal/agent/updater/swarm_test.go
+++ b/internal/agent/updater/swarm_test.go
@@ -77,6 +77,22 @@ func swarmSvc(name string) swarm.Service {
 	}
 }
 
+func TestSwarmUsesPinnedRef(t *testing.T) {
+	const ref = "gitea/gitea:1.26@sha256:abababababababababababababababababababababababababababababababab"
+	job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: ref}
+	fake := &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea")}
+	dist := &fakeDist{digest: "sha256:other"}
+	var out bytes.Buffer
+
+	got, err := NewSwarmExecutor(fake, NewResolver(dist, nil), nil).Execute(context.Background(), job, &out)
+
+	require.NoError(t, err)
+	assert.Equal(t, ref, got)
+	assert.Equal(t, ref, fake.updSpec.TaskTemplate.ContainerSpec.Image)
+	assert.Empty(t, dist.refs, "no registry lookup")
+	assert.Contains(t, out.String(), "pinned "+ref)
+}
+
 func TestSwarmExecutor(t *testing.T) {
 	job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"}
 	workerInfo := managerOf("c1")
diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go
index 2974cf8..bb737ef 100644
--- a/internal/e2e/agent_master_test.go
+++ b/internal/e2e/agent_master_test.go
@@ -328,7 +328,9 @@ type currentHub struct{ p atomic.Pointer[hub.Hub] }
 
 func (c *currentHub) ServeHTTP(w http.ResponseWriter, r *http.Request) { c.p.Load().ServeHTTP(w, r) }
 func (c *currentHub) Disconnect(id int64)                              { c.p.Load().Disconnect(id) }
-func (c *currentHub) Route(svc model.Service) (model.Route, error)     { return c.p.Load().Route(svc) }
+func (c *currentHub) Route(svc model.Service, minProtocol int) (model.Route, error) {
+	return c.p.Load().Route(svc, minProtocol)
+}
 func (c *currentHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
 	return c.p.Load().StartJob(r, id, s)
 }
diff --git a/internal/e2e/jobs_test.go b/internal/e2e/jobs_test.go
index 912ea93..a4d1669 100644
--- a/internal/e2e/jobs_test.go
+++ b/internal/e2e/jobs_test.go
@@ -119,7 +119,7 @@ func (e *jobEnv) waitRoutable(t *testing.T) {
 	t.Helper()
 	svc, err := e.store.GetService(context.Background(), 1)
 	require.NoError(t, err)
-	require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc); return err == nil }, waitFor, pollEvery)
+	require.Eventually(t, func() bool { _, err := e.m.hub.Route(svc, protocol.MinJobVersion); return err == nil }, waitFor, pollEvery)
 }
 
 func (e *jobEnv) job(t *testing.T, id int64) model.Job {
@@ -240,7 +240,7 @@ func TestPhaseOneAgentGetsNoJobs(t *testing.T) {
 	require.NoError(t, wsjson.Write(ctx, c, snap))
 
 	require.Eventually(t, func() bool {
-		_, err := m.hub.Route(svcRow)
+		_, err := m.hub.Route(svcRow, protocol.MinJobVersion)
 		return errors.Is(err, model.ErrAgentTooOld)
 	}, waitFor, pollEvery)
 
diff --git a/internal/e2e/webhook_test.go b/internal/e2e/webhook_test.go
index 5e38c82..ed9ef10 100644
--- a/internal/e2e/webhook_test.go
+++ b/internal/e2e/webhook_test.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/shcizo/package-updater/internal/master/model"
 	"github.com/shcizo/package-updater/internal/master/webhook"
+	"github.com/shcizo/package-updater/internal/protocol"
 )
 
 // webhookServer mounts the webhook handler on a test server, wired to the
@@ -111,7 +112,7 @@ func TestWebhookFailedAutoJobIs500(t *testing.T) {
 	e.agent.stopConn() // no agent connection: the job fails at once with "agent offline"
 	require.Eventually(t, func() bool {
 		svc, _ := e.store.GetService(context.Background(), 1)
-		_, err := e.m.hub.Route(svc)
+		_, err := e.m.hub.Route(svc, protocol.MinJobVersion)
 		return err != nil
 	}, waitFor, pollEvery)
 
diff --git a/internal/master/hub/jobs_test.go b/internal/master/hub/jobs_test.go
index ce7fe00..61fad3f 100644
--- a/internal/master/hub/jobs_test.go
+++ b/internal/master/hub/jobs_test.go
@@ -43,7 +43,7 @@ func waitRoute(t *testing.T, h *Hub) model.Route {
 	var r model.Route
 	require.Eventually(t, func() bool {
 		var err error
-		r, err = h.Route(svc)
+		r, err = h.Route(svc, protocol.MinJobVersion)
 		return err == nil
 	}, 2*time.Second, 5*time.Millisecond)
 	return r
@@ -61,7 +61,7 @@ func TestHub_ReconcilesBeforeRoutable(t *testing.T) {
 	send(t, c, protocol.TypeHello, protocol.Hello{Hostname: "node-2", ProtocolVersion: 2, KnownJobs: []string{"7", "9", "x"}})
 	svc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(node2.ID)}
 	time.Sleep(50 * time.Millisecond) // hello is now blocked inside FailLostJobs
-	_, err = h.Route(svc)
+	_, err = h.Route(svc, protocol.MinJobVersion)
 	assert.ErrorIs(t, err, model.ErrAgentOffline)
 
 	close(store.lostGate)
diff --git a/internal/master/hub/route.go b/internal/master/hub/route.go
index 6013ee7..19fd603 100644
--- a/internal/master/hub/route.go
+++ b/internal/master/hub/route.go
@@ -14,9 +14,10 @@ type candidate struct {
 	hello   protocol.Hello
 }
 
-// Route picks the connection a job for svc goes to. It returns
-// model.ErrAgentOffline or model.ErrAgentTooOld when there is none.
-func (h *Hub) Route(svc model.Service) (model.Route, error) {
+// Route picks the connection a job for svc goes to, among agents that speak
+// at least protocol version minProtocol. It returns model.ErrAgentOffline or
+// model.ErrAgentTooOld when there is none.
+func (h *Hub) Route(svc model.Service, minProtocol int) (model.Route, error) {
 	h.mu.Lock()
 	conns := make([]candidate, 0, len(h.conns))
 	for id, c := range h.conns {
@@ -25,7 +26,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) {
 		}
 	}
 	h.mu.Unlock()
-	return route(svc, conns)
+	return route(svc, conns, minProtocol)
 }
 
 // route picks the connection a job for svc goes to. Compose services and
@@ -33,7 +34,7 @@ func (h *Hub) Route(svc model.Service) (model.Route, error) {
 // run on the reporting agent if it qualifies, else on any other connected
 // manager of the same cluster (lowest agent ID first). Too-old wins over
 // offline only when a connection would otherwise have qualified.
-func route(svc model.Service, conns []candidate) (model.Route, error) {
+func route(svc model.Service, conns []candidate, minProtocol int) (model.Route, error) {
 	sort.Slice(conns, func(a, b int) bool { return conns[a].agentID < conns[b].agentID })
 	tooOld := false
 	try := func(c candidate) bool {
@@ -42,7 +43,7 @@ func route(svc model.Service, conns []candidate) (model.Route, error) {
 				return false
 			}
 		}
-		if c.hello.ProtocolVersion < protocol.MinJobVersion {
+		if c.hello.ProtocolVersion < minProtocol {
 			tooOld = true
 			return false
 		}
diff --git a/internal/master/hub/route_test.go b/internal/master/hub/route_test.go
index 1ad86d8..9393cc8 100644
--- a/internal/master/hub/route_test.go
+++ b/internal/master/hub/route_test.go
@@ -12,37 +12,46 @@ import (
 func TestRoute(t *testing.T) {
 	composeSvc := model.Service{Kind: protocol.KindCompose, AgentID: ptr(int64(2))}
 	swarmSvc := model.Service{Kind: protocol.KindSwarm, AgentID: ptr(int64(1)), SwarmClusterID: "c1"}
-	v2 := protocol.Hello{ProtocolVersion: protocol.Version}
-	mgrV2 := protocol.Hello{ProtocolVersion: protocol.Version, SwarmManager: true, SwarmClusterID: "c1"}
+	v2 := protocol.Hello{ProtocolVersion: 2}
+	v3 := protocol.Hello{ProtocolVersion: 3}
+	mgrV2 := protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c1"}
+	mgrV3 := protocol.Hello{ProtocolVersion: 3, SwarmManager: true, SwarmClusterID: "c1"}
+	jobs, pinned := protocol.MinJobVersion, protocol.MinPinnedRefVersion
 
 	tests := []struct {
 		name  string
 		svc   model.Service
 		conns []candidate
+		min   int
 		want  model.Route
 		err   error
 	}{
-		{"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, model.Route{AgentID: 2, ConnID: 20}, nil},
-		{"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, model.Route{}, model.ErrAgentTooOld},
-		{"compose, own agent missing", composeSvc, nil, model.Route{}, model.ErrAgentOffline},
-		{"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, model.Route{}, model.ErrAgentOffline},
-		{"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 1, ConnID: 10}, nil},
-		{"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
-		{"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
+		{"compose on own agent", composeSvc, []candidate{{2, 20, v2}}, jobs, model.Route{AgentID: 2, ConnID: 20}, nil},
+		{"compose, own agent too old", composeSvc, []candidate{{2, 20, protocol.Hello{}}}, jobs, model.Route{}, model.ErrAgentTooOld},
+		{"compose, own agent missing", composeSvc, nil, jobs, model.Route{}, model.ErrAgentOffline},
+		{"compose, only another agent online", composeSvc, []candidate{{5, 50, v2}}, jobs, model.Route{}, model.ErrAgentOffline},
+		{"compose, v2 agent and a pinned ref", composeSvc, []candidate{{2, 20, v2}}, pinned, model.Route{}, model.ErrAgentTooOld},
+		{"compose, v3 agent and a pinned ref", composeSvc, []candidate{{2, 20, v3}}, pinned, model.Route{AgentID: 2, ConnID: 20}, nil},
+		{"swarm on reporting agent", swarmSvc, []candidate{{1, 10, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 1, ConnID: 10}, nil},
+		{"swarm, reporting agent offline, other manager", swarmSvc, []candidate{{3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
+		{"swarm, lowest agent id first", swarmSvc, []candidate{{4, 40, mgrV2}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
 		{"swarm, other manager in another cluster", swarmSvc,
-			[]candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, model.Route{}, model.ErrAgentOffline},
+			[]candidate{{3, 30, protocol.Hello{ProtocolVersion: 2, SwarmManager: true, SwarmClusterID: "c2"}}}, jobs, model.Route{}, model.ErrAgentOffline},
 		{"swarm, only a v0 manager of the cluster", swarmSvc,
-			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentTooOld},
+			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentTooOld},
 		{"swarm, v0 manager but a v2 manager exists", swarmSvc,
-			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, model.Route{AgentID: 3, ConnID: 30}, nil},
+			[]candidate{{1, 10, protocol.Hello{SwarmManager: true, SwarmClusterID: "c1"}}, {3, 30, mgrV2}}, jobs, model.Route{AgentID: 3, ConnID: 30}, nil},
 		{"swarm, worker of the cluster", swarmSvc,
-			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, model.Route{}, model.ErrAgentOffline},
+			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmClusterID: "c1"}}}, jobs, model.Route{}, model.ErrAgentOffline},
 		{"swarm, manager with empty cluster id", model.Service{Kind: protocol.KindSwarm},
-			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, model.Route{}, model.ErrAgentOffline},
+			[]candidate{{1, 10, protocol.Hello{ProtocolVersion: 2, SwarmManager: true}}}, jobs, model.Route{}, model.ErrAgentOffline},
+		{"swarm, pinned ref falls back to a v3 manager when the reporting one is v2", swarmSvc,
+			[]candidate{{1, 10, mgrV2}, {3, 30, mgrV3}}, pinned, model.Route{AgentID: 3, ConnID: 30}, nil},
+		{"swarm, pinned ref and only v2 managers", swarmSvc, []candidate{{1, 10, mgrV2}}, pinned, model.Route{}, model.ErrAgentTooOld},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := route(tt.svc, tt.conns)
+			got, err := route(tt.svc, tt.conns, tt.min)
 			assert.ErrorIs(t, err, tt.err)
 			assert.Equal(t, tt.want, got)
 		})
diff --git a/internal/master/jobs/jobs.go b/internal/master/jobs/jobs.go
index 6b2fe1b..f479dd3 100644
--- a/internal/master/jobs/jobs.go
+++ b/internal/master/jobs/jobs.go
@@ -20,13 +20,15 @@ type Store interface {
 	CreateJob(ctx context.Context, j model.NewJob) (model.Job, error)
 	FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error)
 	GetJob(ctx context.Context, id int64) (model.Job, error)
+	GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error)
 }
 
 // Hub routes jobs to agent connections. *hub.Hub satisfies it.
 type Hub interface {
-	// Route picks the connection for svc: model.ErrAgentOffline or
+	// Route picks the connection for svc among agents speaking at least
+	// protocol version minProtocol: model.ErrAgentOffline or
 	// model.ErrAgentTooOld when there is none.
-	Route(svc model.Service) (model.Route, error)
+	Route(svc model.Service, minProtocol int) (model.Route, error)
 	// StartJob sends job.start over exactly the routed connection.
 	// model.ErrAgentOffline if that connection is gone or was replaced.
 	StartJob(r model.Route, jobID int64, start protocol.JobStart) error
@@ -116,8 +118,9 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error {
 type StartRequest struct {
 	Trigger     model.JobTrigger
 	RequestedBy string // OIDC subject for manual jobs; "" otherwise
-	// Ref is the ref to deploy ("repo:tag"). Empty means the service's
-	// followed tag, derived from its actual ref.
+	// Ref is the ref to deploy ("repo:tag", "repo:tag@digest" or
+	// "repo@digest"). Empty means the service's followed tag, derived from
+	// its actual ref.
 	Ref string
 }
 
@@ -131,8 +134,30 @@ func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy stri
 	return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy})
 }
 
+// TriggerVersion starts a manual update of the service to exactly the image
+// version versionID, which must be a version of the service's own repo; the
+// ref is built from the stored version, never taken from the caller. It
+// returns model.ErrNotFound for an unknown service, an unknown version or a
+// version of another repo, and otherwise what Start returns. A version
+// without a tag leaves the service following no tag.
+func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error) {
+	svc, err := s.store.GetService(ctx, serviceID)
+	if err != nil {
+		return model.Job{}, err
+	}
+	v, err := s.store.GetImageVersion(ctx, versionID)
+	if err != nil {
+		return model.Job{}, err
+	}
+	if v.Repo != svc.ImageRepo {
+		return model.Job{}, model.ErrNotFound
+	}
+	return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: v.Ref()})
+}
+
 // Start starts an update of svc. It returns the job, which may already be
-// failed ("agent offline", "agent too old for updates"); a
+// failed ("agent offline", "agent too old for updates", the latter also when
+// the ref has a digest or no tag and the agent speaks protocol version 2); a
 // *model.NotUpdatableError when the service cannot be updated (no job
 // created); the active job and model.ErrJobActive when one is queued or
 // running.
@@ -145,7 +170,11 @@ func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest
 		ref = FollowedRef(svc.ActualRef) // updatable checked the tag
 	}
 
-	r, routeErr := s.hub.Route(svc)
+	minProtocol := protocol.MinJobVersion
+	if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" {
+		minProtocol = protocol.MinPinnedRefVersion
+	}
+	r, routeErr := s.hub.Route(svc, minProtocol)
 	nj := model.NewJob{
 		ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref,
 		Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(),
diff --git a/internal/master/jobs/jobs_test.go b/internal/master/jobs/jobs_test.go
index 3918fb2..7ad6044 100644
--- a/internal/master/jobs/jobs_test.go
+++ b/internal/master/jobs/jobs_test.go
@@ -21,6 +21,7 @@ type fakeStore struct {
 	jobs     map[int64]model.Job
 	created  []model.NewJob
 	active   *model.Job // returned with ErrJobActive when set
+	versions map[int64]model.ImageVersion
 
 	revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these
 }
@@ -54,6 +55,14 @@ func (f *fakeStore) FailJob(_ context.Context, id int64, msg string, at time.Tim
 	return true, nil
 }
 
+func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) {
+	v, ok := f.versions[id]
+	if !ok {
+		return model.ImageVersion{}, model.ErrNotFound
+	}
+	return v, nil
+}
+
 func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil }
 
 type startCall struct {
@@ -63,13 +72,21 @@ type startCall struct {
 }
 
 type fakeHub struct {
-	route    model.Route
-	routeErr error
-	startErr error
-	starts   []startCall
+	route         model.Route
+	routeErr      error
+	startErr      error
+	starts        []startCall
+	agentProtocol int   // when set, Route fails with ErrAgentTooOld for a higher minProtocol
+	mins          []int // minProtocol of every Route call
 }
 
-func (h *fakeHub) Route(model.Service) (model.Route, error) { return h.route, h.routeErr }
+func (h *fakeHub) Route(_ model.Service, minProtocol int) (model.Route, error) {
+	h.mins = append(h.mins, minProtocol)
+	if h.agentProtocol != 0 && minProtocol > h.agentProtocol {
+		return model.Route{}, model.ErrAgentTooOld
+	}
+	return h.route, h.routeErr
+}
 
 func (h *fakeHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
 	h.starts = append(h.starts, startCall{r, id, s})
@@ -341,3 +358,133 @@ func TestFollowedRef(t *testing.T) {
 	assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo"))
 	assert.Empty(t, FollowedRef("redis@sha256:abc"))
 }
+
+func TestStart_RoutesWithTheMinimumProtocolOfTheRef(t *testing.T) {
+	tests := []struct {
+		name string
+		ref  string // "" means the followed tag
+		want int
+	}{
+		{"followed tag", "", protocol.MinJobVersion},
+		{"repo:tag", "x/y:2", protocol.MinJobVersion},
+		{"repo:tag@digest", "x/y:2@sha256:a", protocol.MinPinnedRefVersion},
+		{"repo@digest", "x/y@sha256:a", protocol.MinPinnedRefVersion},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, _ := newService(st, h)
+
+			_, err := s.Start(context.Background(), composeSvc, StartRequest{Trigger: model.TriggerManual, Ref: tt.ref})
+
+			require.NoError(t, err)
+			assert.Equal(t, []int{tt.want}, h.mins)
+		})
+	}
+}
+
+func TestTriggerVersion_Ref(t *testing.T) {
+	const repo = "gitea.shcizo.se/shcizo/slusk"
+	noTag := model.Service{ID: 4, Kind: protocol.KindCompose, ComposeProject: "n", ComposeService: "n",
+		AgentID: ptr(int64(7)), ImageRepo: repo, ActualRef: repo + "@sha256:d1"}
+	svc := composeSvc
+	svc.ImageRepo = repo
+	tests := []struct {
+		name    string
+		svc     model.Service
+		version model.ImageVersion
+		wantRef string
+		wantMin int
+	}{
+		{"tag and digest", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
+		{"tag only", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "main"}, repo + ":main", protocol.MinJobVersion},
+		{"digest only", svc, model.ImageVersion{ID: 10, Repo: repo, Digest: "sha256:old"}, repo + "@sha256:old", protocol.MinPinnedRefVersion},
+		{"service that follows no tag", noTag, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			st := &fakeStore{
+				services: map[int64]model.Service{tt.svc.ID: tt.svc},
+				jobs:     map[int64]model.Job{},
+				versions: map[int64]model.ImageVersion{10: tt.version},
+			}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, _ := newService(st, h)
+
+			job, err := s.TriggerVersion(context.Background(), tt.svc.ID, 10, "sub-1")
+
+			require.NoError(t, err)
+			assert.Equal(t, model.JobQueued, job.Status)
+			require.Len(t, st.created, 1)
+			assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
+			assert.Equal(t, "sub-1", st.created[0].RequestedBy)
+			assert.Equal(t, tt.wantRef, st.created[0].ToRef)
+			require.Len(t, h.starts, 1)
+			assert.Equal(t, tt.wantRef, h.starts[0].start.Ref)
+			assert.Equal(t, []int{tt.wantMin}, h.mins)
+		})
+	}
+}
+
+func TestTriggerVersion_NotFound(t *testing.T) {
+	const repo = "gitea.shcizo.se/shcizo/slusk"
+	svc := composeSvc
+	svc.ImageRepo = repo
+	tests := []struct {
+		name      string
+		serviceID int64
+		versionID int64
+	}{
+		{"unknown service", 9, 10},
+		{"unknown version", 1, 99},
+		{"version of another repo", 1, 11},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			st := &fakeStore{
+				services: map[int64]model.Service{1: svc},
+				jobs:     map[int64]model.Job{},
+				versions: map[int64]model.ImageVersion{
+					10: {ID: 10, Repo: repo, Tag: "1"},
+					11: {ID: 11, Repo: "evil/other", Tag: "1"},
+				},
+			}
+			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
+			s, pub := newService(st, h)
+
+			_, err := s.TriggerVersion(context.Background(), tt.serviceID, tt.versionID, "sub-1")
+
+			assert.ErrorIs(t, err, model.ErrNotFound)
+			assert.Empty(t, st.created, "no job")
+			assert.Empty(t, h.starts)
+			assert.Empty(t, pub.topics)
+		})
+	}
+}
+
+func TestTriggerVersion_RequiresPinnedProtocol(t *testing.T) {
+	const repo = "gitea.shcizo.se/shcizo/slusk"
+	svc := composeSvc
+	svc.ImageRepo = repo
+	st := &fakeStore{
+		services: map[int64]model.Service{1: svc},
+		jobs:     map[int64]model.Job{},
+		versions: map[int64]model.ImageVersion{
+			10: {ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"},
+			11: {ID: 11, Repo: repo, Tag: "main"},
+		},
+	}
+	h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2}
+	s, _ := newService(st, h)
+
+	pinned, err := s.TriggerVersion(context.Background(), 1, 10, "sub-1")
+	require.NoError(t, err)
+	assert.Equal(t, model.JobFailed, pinned.Status)
+	assert.Equal(t, "agent too old for updates", pinned.Error)
+	assert.Empty(t, h.starts, "nothing sent to the old agent")
+
+	plain, err := s.TriggerVersion(context.Background(), 1, 11, "sub-1")
+	require.NoError(t, err)
+	assert.Equal(t, model.JobQueued, plain.Status)
+}
diff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
index af25a2c..2d67989 100644
--- a/internal/protocol/protocol.go
+++ b/internal/protocol/protocol.go
@@ -16,11 +16,16 @@ import (
 
 // Version is the protocol version this build speaks. Agents send it in
 // Hello. Phase-1 agents do not send the field, which reads as 0.
-const Version = 2
+const Version = 3
 
 // MinJobVersion is the lowest agent protocol version the master sends jobs to.
 const MinJobVersion = 2
 
+// MinPinnedRefVersion is the lowest agent protocol version the master sends
+// jobs whose Ref has a digest or no tag to. Version-2 agents refuse a Ref
+// with a digest.
+const MinPinnedRefVersion = 3
+
 // MaxJobOutput caps the job output an agent keeps and the master stores, in
 // bytes. Only the tail is kept. With JSON escaping (worst case 6 bytes per
 // byte) a JobResult carrying this much output still fits the 1 MiB read limit.
@@ -123,8 +128,10 @@ type Remove struct {
 	Key string `json:"key"`
 }
 
-// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag";
-// the agent resolves the tag's current digest and deploys "repo:tag@digest".
+// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag",
+// "repo:tag@digest" or "repo@digest". For "repo:tag" the agent resolves the
+// tag's current digest and deploys "repo:tag@digest"; a Ref with a digest is
+// deployed as is.
 // The agent reads the Compose working directory and config files from its
 // own Docker daemon, not from this message.
 type JobStart struct {