diff --git a/docs/examples/provider.go b/docs/examples/provider.go
index 89ef92c4ad..a948c24cdf 100644
--- a/docs/examples/provider.go
+++ b/docs/examples/provider.go
@@ -18,9 +18,12 @@ package main
 
 import (
 	"bufio"
+	"crypto/sha256"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net"
+	"net/http/httputil"
 	"os"
 	"os/exec"
 	"strings"
@@ -85,11 +88,88 @@ func composeCommand() *cobra.Command {
 		Args: cobra.ExactArgs(1),
 	}
 
-	c.AddCommand(upCmd, downCmd, stopCmd)
-	c.AddCommand(metadataCommand(upCmd, downCmd, stopCmd))
+	// pull is the image-distribution command: compose invokes it during the
+	// image phase (up) and on `docker compose pull`, passing the identity of
+	// the service image and the state of the local daemon cache. None of its
+	// flags is required: they are injected by compose, not declared by the
+	// user under provider.options.
+	pullCmd := &cobra.Command{
+		Use:  "pull",
+		Run:  pull,
+		Args: cobra.ExactArgs(1),
+	}
+	pullCmd.Flags().String("image", "", "Image reference as compose resolved it")
+	pullCmd.Flags().String("digest", "", "Image ID in the local daemon cache, when present")
+	pullCmd.Flags().String("created", "", "Creation time of the local cache entry, when present")
+	pullCmd.Flags().String("source", "", "Authority verdict: local (sync from the daemon) or registry (resolve upstream)")
+	pullCmd.Flags().String("policy", "", "missing (a usable version suffices) or always (ensure freshness)")
+
+	c.AddCommand(upCmd, downCmd, stopCmd, pullCmd)
+	c.AddCommand(metadataCommand(upCmd, downCmd, stopCmd, pullCmd))
 	return c
 }
 
+// pull demonstrates the image-distribution contract. A real provider would
+// compare the announced digest/created with its bookkeeping and either
+// resolve the reference upstream (source=registry) or request the local
+// bytes (source=local). This demo requests the stream whenever
+// PROVIDER_PULL_MARKER is set, hashes it, and records the outcome there.
+func pull(cmd *cobra.Command, args []string) {
+	image, _ := cmd.Flags().GetString("image")
+	digest, _ := cmd.Flags().GetString("digest")
+	created, _ := cmd.Flags().GetString("created")
+	source, _ := cmd.Flags().GetString("source")
+	policy, _ := cmd.Flags().GetString("policy")
+
+	emit := func(kind, message string) {
+		payload, _ := json.Marshal(map[string]string{"type": kind, "message": message})
+		fmt.Println(string(payload))
+	}
+	emit("info", fmt.Sprintf("ensuring image %s (source=%s, policy=%s)", image, source, policy))
+
+	marker := os.Getenv("PROVIDER_PULL_MARKER")
+	if marker == "" {
+		// nothing to synchronize in the demo: a real registry-sourced provider
+		// would pull the reference upstream here
+		return
+	}
+
+	// Request the image bytes from the local daemon. The answer is one JSON
+	// line, then — unless it carries an error — the tar as an HTTP/1.1
+	// chunked body (readable with any stock chunked reader), whose zero
+	// chunk marks a COMPLETE transfer.
+	request, _ := json.Marshal(map[string]string{"type": "get-image", "message": image})
+	fmt.Println(string(request))
+
+	stdin := bufio.NewReader(os.Stdin)
+	line, err := stdin.ReadString('\n')
+	if err != nil {
+		emit("error", "reading image-stream announce: "+err.Error())
+		os.Exit(1)
+	}
+	var announce struct {
+		Encoding string `json:"encoding"`
+		Error    string `json:"error"`
+	}
+	if err := json.Unmarshal([]byte(line), &announce); err != nil || announce.Error != "" || announce.Encoding != "chunked" {
+		emit("error", fmt.Sprintf("image stream unavailable: %q (err %v)", line, err))
+		os.Exit(1)
+	}
+	h := sha256.New()
+	n, err := io.Copy(h, httputil.NewChunkedReader(stdin))
+	if err != nil {
+		emit("error", "image stream truncated: "+err.Error())
+		os.Exit(1)
+	}
+	record := fmt.Sprintf("image=%s digest=%s created=%s source=%s policy=%s sha256=%x bytes=%d\n",
+		image, digest, created, source, policy, h.Sum(nil), n)
+	if err := os.WriteFile(marker, []byte(record), 0o600); err != nil {
+		emit("error", "writing marker: "+err.Error())
+		os.Exit(1)
+	}
+	emit("info", fmt.Sprintf("image received (%d bytes)", n))
+}
+
 // serveDemoCommand is the detached helper process behind the
 // publish-endpoint demonstration: a TCP server on the given address
 // answering every connection with a fixed HTTP response, exiting on its own
@@ -227,23 +307,27 @@ func stop(_ *cobra.Command, _ []string) {
 	}
 }
 
-func metadataCommand(upCmd, downCmd, stopCmd *cobra.Command) *cobra.Command {
+func metadataCommand(upCmd, downCmd, stopCmd, pullCmd *cobra.Command) *cobra.Command {
 	return &cobra.Command{
 		Use: "metadata",
 		Run: func(cmd *cobra.Command, _ []string) {
-			metadata(upCmd, downCmd, stopCmd)
+			metadata(upCmd, downCmd, stopCmd, pullCmd)
 		},
 		Args: cobra.NoArgs,
 	}
 }
 
-func metadata(upCmd, downCmd, stopCmd *cobra.Command) {
+func metadata(upCmd, downCmd, stopCmd, pullCmd *cobra.Command) {
 	metadata := ProviderMetadata{}
 	metadata.Description = "Manage services on AwesomeCloud"
 	metadata.Up = commandParameters(upCmd)
 	metadata.Down = commandParameters(downCmd)
 	stopParams := commandParameters(stopCmd)
 	metadata.Stop = &stopParams
+	// like stop, declaring the pull block is what opts the provider into
+	// image distribution
+	pullParams := commandParameters(pullCmd)
+	metadata.Pull = &pullParams
 	jsonMetadata, err := json.Marshal(metadata)
 	if err != nil {
 		panic(err)
@@ -271,6 +355,7 @@ type ProviderMetadata struct {
 	Up          CommandMetadata  `json:"up"`
 	Down        CommandMetadata  `json:"down"`
 	Stop        *CommandMetadata `json:"stop,omitempty"`
+	Pull        *CommandMetadata `json:"pull,omitempty"`
 }
 
 type CommandMetadata struct {
diff --git a/docs/extension.md b/docs/extension.md
index 0a173e4436..afdf56779e 100644
--- a/docs/extension.md
+++ b/docs/extension.md
@@ -30,7 +30,10 @@ the resource(s) needed to run a service.
 If `provider.type` doesn't resolve into any of those, Compose will report an error and interrupt the `up` command.
 
 To be a valid Compose extension, provider command *MUST* accept a `compose` command (which can be hidden)
-with subcommands `up` and `down`. It *MAY* additionally implement a `stop` subcommand to support `docker compose stop`.
+with subcommands `up` and `down`. It *MAY* additionally implement a `stop` subcommand to support `docker compose stop`,
+and a `pull` subcommand to take part in image distribution (see [Image distribution](#image-distribution)).
+Optional subcommands are declared through the provider metadata: the presence of the command block is what
+opts the provider in.
 
 ## Up lifecycle
 
@@ -111,6 +114,68 @@ sequenceDiagram
     Compose-)Shell: service started
 ```
 
+## Image distribution
+
+A provider-backed service can declare `build` or `image` like any other service; the image then has to reach
+the provider's runtime, which may be nowhere near the local daemon. Providers opt into image distribution by
+declaring a `pull` block in their `metadata` output — like `stop`, the presence of the block is the
+declaration of support. Providers without it keep managing images on their own during `up`.
+
+When the provider declares `pull`, Compose invokes it during the image phase of `up` (after any build) and on
+`docker compose pull`:
+
+```console
+awesomecloud compose --project-name <NAME> pull --image=<ref> --source=<verdict> --policy=<policy> [--digest=<id> --created=<time>] "database"
+```
+
+- `--image`: the image reference as Compose resolved it (the `image` attribute, or `<project>-<service>` for a
+  build-only service).
+- `--digest` / `--created`: the state of the **local daemon cache**, present only when the image exists there.
+  They describe the cache, they are not instructions: persist them as the bookkeeping keys of what you ingest —
+  the digest as identity test, `created` as the ordering fallback for a backend that cannot preserve digests.
+  Beware that reproducible builds can freeze `created`, so a comparable digest always wins over it.
+- `--source` is the authority verdict, computed by Compose from the model and the invocation (`pull_policy`,
+  `--build`, what the current run just built), so providers never re-implement that arbitration:
+  - `local`: the desired state is the local daemon's image. Compare your bookkeeping with the announced
+    digest/created; when they differ, request the bytes with `get-image`.
+  - `registry`: resolve the reference upstream — this includes the common workflow where `build` is only the
+    recipe CI uses to publish the image that consumers pull. The local facts are an optimization, never an
+    obligation.
+- `--policy`:
+  - `missing` (the `up` path): a usable version present in your runtime suffices;
+  - `always` (`docker compose pull`): ensure your runtime holds the latest version of the authority.
+
+### Requesting the image bytes
+
+During `pull`, the provider can ask Compose for the image content with a regular JSON line on `stdout`
+(`platform` is optional and narrows a multi-platform image):
+
+```json
+{ "type": "get-image", "message": "<image ref>", "platform": "linux/arm64" }
+```
+
+Compose answers on the provider's `stdin` with one JSON line:
+
+```json
+{ "type": "image-stream", "encoding": "chunked", "media-type": "application/x-tar" }
+```
+
+followed — unless the line carries an `error` field instead — by the image tar encoded as HTTP/1.1 chunked
+data (RFC 9112 §7.1): every block of data prefixed by its length, terminated by the zero-length chunk. Unlike
+an HTTP message there is no trailer section nor final CRLF — the next byte after the zero chunk belongs to the
+next stdin answer. Length-prefixed framing needs no in-band delimiter (any byte value can appear inside a tar), and
+any language's stock chunked-body reader consumes it — Go providers can use `httputil.NewChunkedReader`. A
+stream that ends without the terminating zero chunk was aborted and must be discarded — Compose closes the
+answer channel after an aborted transfer, so the truncation is always observable as EOF. The tar is what
+`docker image save` produces: feed it to `docker load` or whatever your runtime ingests.
+
+As during `stop`, any `setenv`, `rawsetenv` or `publish-endpoint` message emitted during `pull` is accepted
+but ignored: dependent services are not being configured in this phase.
+
+The stream is exclusive on `stdin` for its whole duration — the chunked body must be contiguous, so answers to
+any other request emitted meanwhile are delivered after it. Drain the announced stream completely before
+expecting another answer.
+
 ## Connection to a service managed by a provider
 
 A service in the Compose application can declare dependency on a service managed by an external provider: 
@@ -269,6 +334,9 @@ The expected JSON output format is:
         "type": "string"
       }
     ]
+  },
+  "pull": {
+    "parameters": []
   }
 }
 ```
@@ -277,6 +345,9 @@ The top elements are:
 - `up`: Object describing the parameters accepted by the `up` command
 - `down`: Object describing the parameters accepted by the `down` command
 - `stop`: Object describing the parameters accepted by the `stop` command (optional)
+- `pull`: Object describing the parameters accepted by the `pull` command (optional — declaring the block is
+  what opts the provider into [image distribution](#image-distribution); the flags Compose injects need not be
+  listed)
 
 And for each command parameter, you should include the following properties:
 - `name`: The parameter name (without `--` prefix)
diff --git a/pkg/compose/build.go b/pkg/compose/build.go
index b422b50790..e70b16d5e0 100644
--- a/pkg/compose/build.go
+++ b/pkg/compose/build.go
@@ -106,7 +106,7 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti
 	return s.doBuildClassic(ctx, project, serviceToBuild, options)
 }
 
-func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool) error {
+func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool, skipProviders bool) error {
 	for name, service := range project.Services {
 		if service.Provider == nil && service.Image == "" && service.Build == nil {
 			return fmt.Errorf("invalid service %q. Must specify either image or build", name)
@@ -127,6 +127,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types.
 		return err
 	}
 
+	built := map[string]string{}
 	if buildOpts != nil {
 		err = tracing.SpanWrapFunc("project/build", tracing.ProjectOptions(ctx, project),
 			func(ctx context.Context) error {
@@ -136,6 +137,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types.
 				}
 
 				for name, digest := range builtImages {
+					built[name] = digest
 					images[name] = api.ImageSummary{
 						Repository:  name,
 						ID:          digest,
@@ -150,6 +152,20 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types.
 		}
 	}
 
+	// provider-backed services get their turn in the image phase: the
+	// provider makes the image available to ITS runtime (weak contract on
+	// the up path: a usable version present suffices)
+	if !skipProviders {
+		err = tracing.SpanWrapFunc("project/provider-pull", tracing.ProjectOptions(ctx, project),
+			func(ctx context.Context) error {
+				return s.ensureProviderImages(ctx, project, built, providerPullPolicyMissing)
+			},
+		)(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
 	// set digest as com.docker.compose.image label so we can detect outdated
 	// containers — the single writer of that label, so the platform-pinned
 	// resolution below can't be overwritten by another code path
diff --git a/pkg/compose/create.go b/pkg/compose/create.go
index b5496a68e9..9c7f4a9a6a 100644
--- a/pkg/compose/create.go
+++ b/pkg/compose/create.go
@@ -80,7 +80,7 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt
 		return err
 	}
 
-	err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull)
+	err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull, options.SkipProviders)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/compose/images_test.go b/pkg/compose/images_test.go
index a1b53ad477..53921e97b2 100644
--- a/pkg/compose/images_test.go
+++ b/pkg/compose/images_test.go
@@ -423,7 +423,7 @@ func TestPlatformPinnedDigest(t *testing.T) {
 			Return(client.ImageInspectResult{InspectResponse: multiPlatform}, nil)
 
 		project := newProject()
-		assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true))
+		assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true, false))
 		assert.Equal(t, project.Services["app"].CustomLabels[compose.ImageDigestLabel], "sha256:s390x")
 	})
 
diff --git a/pkg/compose/plugins.go b/pkg/compose/plugins.go
index ffd885b7bf..158e69c3fe 100644
--- a/pkg/compose/plugins.go
+++ b/pkg/compose/plugins.go
@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"net/http/httputil"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -33,8 +34,10 @@ import (
 
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/containerd/errdefs"
+	"github.com/containerd/platforms"
 	"github.com/docker/cli/cli-plugins/manager"
 	"github.com/docker/cli/cli/config"
+	"github.com/moby/moby/client"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 
@@ -44,6 +47,9 @@ import (
 type JsonMessage struct {
 	Type    string `json:"type"`
 	Message string `json:"message,omitempty"`
+	// Platform optionally narrows a get-image request to one platform of a
+	// multi-platform image (e.g. "linux/arm64").
+	Platform string `json:"platform,omitempty"`
 }
 
 const (
@@ -59,8 +65,29 @@ const (
 	// its stdin, one JSON line holding the resolved canonical configuration
 	// of the service it manages — answered from the in-memory model.
 	GetServiceConfigType = "get-service-config"
+
+	// GetImageType is a message the provider sends during its pull command
+	// to receive the service image from the local daemon: compose answers on
+	// the provider's stdin with one ImageStreamType JSON line, then — unless
+	// that line carries an error — the image tar encoded as an HTTP/1.1
+	// chunked body (see streamImageTo).
+	GetImageType = "get-image"
+
+	// ImageStreamType is the type of the JSON line answering a get-image
+	// request.
+	ImageStreamType = "image-stream"
 )
 
+// imageStreamAnswer is the stdin answer to a get-image request. On success
+// Encoding and MediaType describe the byte stream that follows the JSON line;
+// on failure Error carries the reason and no stream follows.
+type imageStreamAnswer struct {
+	Type      string `json:"type"`
+	Encoding  string `json:"encoding,omitempty"`
+	MediaType string `json:"media-type,omitempty"`
+	Error     string `json:"error,omitempty"`
+}
+
 type pluginVariables struct {
 	prefixed types.Mapping
 	raw      types.Mapping
@@ -73,7 +100,7 @@ type pluginVariables struct {
 
 var mux sync.Mutex
 
-func (s *composeService) runPlugin(ctx context.Context, project *types.Project, service types.ServiceConfig, command string) error {
+func (s *composeService) runPlugin(ctx context.Context, project *types.Project, service types.ServiceConfig, command string, extraArgs ...string) error {
 	provider := *service.Provider
 
 	plugin, err := s.getPluginBinaryPath(provider.Type)
@@ -81,7 +108,7 @@ func (s *composeService) runPlugin(ctx context.Context, project *types.Project,
 		return err
 	}
 
-	cmd, err := s.setupPluginCommand(ctx, project, service, plugin, command)
+	cmd, err := s.setupPluginCommand(ctx, project, service, plugin, command, extraArgs...)
 	if err != nil {
 		return err
 	}
@@ -89,12 +116,12 @@ func (s *composeService) runPlugin(ctx context.Context, project *types.Project,
 		return nil
 	}
 
-	variables, err := s.executePlugin(cmd, command, service)
+	variables, err := s.executePlugin(ctx, cmd, command, service)
 	if err != nil {
 		return err
 	}
 
-	if command == "stop" {
+	if command == "stop" || command == "pull" {
 		return nil
 	}
 
@@ -159,7 +186,7 @@ func parseEndpointMessage(message string) (int, string, error) {
 	return port, upstream, nil
 }
 
-func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service types.ServiceConfig) (pluginVariables, error) {
+func (s *composeService) executePlugin(ctx context.Context, cmd *exec.Cmd, command string, service types.ServiceConfig) (pluginVariables, error) {
 	var action string
 	switch command {
 	case "up":
@@ -171,6 +198,9 @@ func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service ty
 	case "stop":
 		s.events.On(stoppingEvent(service.Name))
 		action = "stop"
+	case "pull":
+		s.events.On(newEvent(service.Name, api.Working, "Pulling"))
+		action = "pull"
 	default:
 		return pluginVariables{}, fmt.Errorf("unsupported plugin command: %s", command)
 	}
@@ -243,47 +273,8 @@ func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service ty
 		if err != nil {
 			return pluginVariables{}, err
 		}
-		switch msg.Type {
-		case ErrorType:
-			s.events.On(newEvent(service.Name, api.Error, firstLine(msg.Message)))
-			return pluginVariables{}, errors.New(msg.Message)
-		case InfoType:
-			s.events.On(newEvent(service.Name, api.Working, firstLine(msg.Message)))
-		case SetEnvType:
-			key, val, found := strings.Cut(msg.Message, "=")
-			if !found {
-				return pluginVariables{}, fmt.Errorf("invalid message from plugin: %s", msg.Message)
-			}
-			variables.prefixed[key] = val
-		case RawSetEnvType:
-			key, val, found := strings.Cut(msg.Message, "=")
-			if !found {
-				return pluginVariables{}, fmt.Errorf("invalid message from plugin: %s", msg.Message)
-			}
-			variables.raw[key] = val
-		case GetServiceConfigType:
-			payload, err := json.Marshal(service)
-			if err != nil {
-				return pluginVariables{}, fmt.Errorf("failed to answer get-service-config: %w", err)
-			}
-			payload = append(payload, '\n')
-			answers.Add(1)
-			go func() {
-				defer answers.Done()
-				stdinMu.Lock()
-				defer stdinMu.Unlock()
-				_, _ = stdin.Write(payload)
-			}()
-		case PublishEndpointType:
-			port, upstream, err := parseEndpointMessage(msg.Message)
-			if err != nil {
-				return pluginVariables{}, fmt.Errorf("invalid message from plugin: %w", err)
-			}
-			variables.endpoints[port] = upstream
-		case DebugType:
-			logrus.Debugf("%s: %s", service.Name, msg.Message)
-		default:
-			return pluginVariables{}, fmt.Errorf("invalid message from plugin: %s", msg.Type)
+		if err := s.handlePluginMessage(ctx, msg, command, service, &variables, stdin, &stdinMu, &answers); err != nil {
+			return pluginVariables{}, err
 		}
 	}
 
@@ -300,10 +291,150 @@ func (s *composeService) executePlugin(cmd *exec.Cmd, command string, service ty
 		s.events.On(removedEvent(service.Name))
 	case "stop":
 		s.events.On(stoppedEvent(service.Name))
+	case "pull":
+		s.events.On(newEvent(service.Name, api.Done, "Pulled"))
 	}
 	return variables, nil
 }
 
+// imageStreamChunkSize is the fixed buffer compose fills from the image
+// export before flushing it as one chunk of the stream.
+const imageStreamChunkSize = 1024 * 1024
+
+// streamImageTo answers a get-image request on the provider's stdin: one
+// ImageStreamType JSON line announcing the transfer, then the image tar
+// encoded as HTTP/1.1 chunked data (RFC 9112 §7.1) — every block of data
+// prefixed by its length, the zero-length chunk marking a COMPLETE
+// transfer. Length-prefixed framing needs no in-band delimiter (any
+// byte value can appear inside a tar) and is consumable with any language's
+// chunked-body reader. A stream that ends without the terminating zero chunk
+// was aborted: the provider must discard it.
+// When the image cannot be exported, the announce line carries an error
+// instead and no stream follows.
+func (s *composeService) streamImageTo(ctx context.Context, w io.Writer, ref, platform string) error {
+	announce := func(a imageStreamAnswer) error {
+		payload, err := json.Marshal(a)
+		if err != nil {
+			return err
+		}
+		_, err = w.Write(append(payload, '\n'))
+		return err
+	}
+
+	var opts []client.ImageSaveOption
+	if platform != "" {
+		p, err := platforms.Parse(platform)
+		if err != nil {
+			_ = announce(imageStreamAnswer{Type: ImageStreamType, Error: fmt.Sprintf("invalid platform %q: %s", platform, err)})
+			return err
+		}
+		opts = append(opts, client.ImageSaveWithPlatforms(p))
+	}
+	tar, err := s.apiClient().ImageSave(ctx, []string{ref}, opts...)
+	if err != nil {
+		_ = announce(imageStreamAnswer{Type: ImageStreamType, Error: err.Error()})
+		return err
+	}
+	defer func() { _ = tar.Close() }()
+
+	if err := announce(imageStreamAnswer{Type: ImageStreamType, Encoding: "chunked", MediaType: "application/x-tar"}); err != nil {
+		return err
+	}
+	cw := httputil.NewChunkedWriter(w)
+	if _, err := io.CopyBuffer(cw, struct{ io.Reader }{tar}, make([]byte, imageStreamChunkSize)); err != nil {
+		return err
+	}
+	// ChunkedWriter.Close writes the zero-length chunk, which ends the
+	// stream. Unlike an HTTP message there is deliberately no trailer
+	// section nor final CRLF: a stock chunked reader stops at the zero
+	// chunk without consuming either, and leftover bytes would corrupt the
+	// next JSON answer a provider requesting several images reads.
+	return cw.Close()
+}
+
+// handlePluginMessage processes one provider message. Answer-bearing requests
+// (get-service-config, get-image) are served from their own goroutine — see
+// the answers/stdinMu contract in executePlugin. A returned error aborts the
+// provider run.
+func (s *composeService) handlePluginMessage(
+	ctx context.Context, msg JsonMessage, command string, service types.ServiceConfig,
+	variables *pluginVariables, stdin io.WriteCloser, stdinMu *sync.Mutex, answers *sync.WaitGroup,
+) error {
+	switch msg.Type {
+	case ErrorType:
+		s.events.On(newEvent(service.Name, api.Error, firstLine(msg.Message)))
+		return errors.New(msg.Message)
+	case InfoType:
+		s.events.On(newEvent(service.Name, api.Working, firstLine(msg.Message)))
+	case SetEnvType:
+		key, val, found := strings.Cut(msg.Message, "=")
+		if !found {
+			return fmt.Errorf("invalid message from plugin: %s", msg.Message)
+		}
+		variables.prefixed[key] = val
+	case RawSetEnvType:
+		key, val, found := strings.Cut(msg.Message, "=")
+		if !found {
+			return fmt.Errorf("invalid message from plugin: %s", msg.Message)
+		}
+		variables.raw[key] = val
+	case GetServiceConfigType:
+		payload, err := json.Marshal(service)
+		if err != nil {
+			return fmt.Errorf("failed to answer get-service-config: %w", err)
+		}
+		payload = append(payload, '\n')
+		answers.Add(1)
+		go func() {
+			defer answers.Done()
+			stdinMu.Lock()
+			defer stdinMu.Unlock()
+			_, _ = stdin.Write(payload)
+		}()
+	case GetImageType:
+		// image distribution belongs to the pull command: answering it
+		// elsewhere would let an image export stall a down or a stop
+		if command != "pull" {
+			return fmt.Errorf("invalid message from plugin: %s is only supported during the pull command", GetImageType)
+		}
+		ref, platform := msg.Message, msg.Platform
+		answers.Add(1)
+		go func() {
+			defer answers.Done()
+			// stdinMu is deliberately held for the whole transfer: the
+			// chunked body must be contiguous on stdin, any concurrent
+			// answer interleaved into it would corrupt the framing. A
+			// provider must therefore drain the announced stream before
+			// expecting any other answer (documented contract); compose
+			// cannot hang forever on a provider that stops reading — the
+			// error path kills the process, which EPIPEs the write.
+			stdinMu.Lock()
+			defer stdinMu.Unlock()
+			if err := s.streamImageTo(ctx, stdin, ref, platform); err != nil {
+				logrus.Warnf("provider %q: get-image %q: %v", service.Name, ref, err)
+				// A failure after the success announce leaves the stream
+				// without its terminating chunk, and the channel cannot be
+				// resynchronized (any byte would read as chunk data).
+				// Closing stdin makes the truncation observable — the
+				// provider gets EOF mid-chunk and discards, instead of
+				// blocking forever on a stream nobody will finish.
+				_ = stdin.Close()
+			}
+		}()
+	case PublishEndpointType:
+		port, upstream, err := parseEndpointMessage(msg.Message)
+		if err != nil {
+			return fmt.Errorf("invalid message from plugin: %w", err)
+		}
+		variables.endpoints[port] = upstream
+	case DebugType:
+		logrus.Debugf("%s: %s", service.Name, msg.Message)
+	default:
+		return fmt.Errorf("invalid message from plugin: %s", msg.Type)
+	}
+	return nil
+}
+
 func (s *composeService) getPluginBinaryPath(provider string) (path string, err error) {
 	if provider == "compose" {
 		return "", errors.New("'compose' is not a valid provider type")
@@ -322,7 +453,7 @@ func (s *composeService) getPluginBinaryPath(provider string) (path string, err
 	return path, err
 }
 
-func (s *composeService) setupPluginCommand(ctx context.Context, project *types.Project, service types.ServiceConfig, path, command string) (*exec.Cmd, error) {
+func (s *composeService) setupPluginCommand(ctx context.Context, project *types.Project, service types.ServiceConfig, path, command string, extraArgs ...string) (*exec.Cmd, error) {
 	cmdOptionsMetadata := s.getPluginMetadata(path, service.Provider.Type, project)
 	var currentCommandMetadata CommandMetadata
 	switch command {
@@ -335,6 +466,13 @@ func (s *composeService) setupPluginCommand(ctx context.Context, project *types.
 			return nil, nil
 		}
 		currentCommandMetadata = *cmdOptionsMetadata.Stop
+	case "pull":
+		// image distribution is opt-in, declared like stop by the presence
+		// of the command block in the provider metadata
+		if cmdOptionsMetadata.Pull == nil {
+			return nil, nil
+		}
+		currentCommandMetadata = *cmdOptionsMetadata.Pull
 	}
 
 	provider := *service.Provider
@@ -351,6 +489,7 @@ func (s *composeService) setupPluginCommand(ctx context.Context, project *types.
 			}
 		}
 	}
+	args = append(args, extraArgs...)
 	args = append(args, service.Name)
 
 	cmd := exec.CommandContext(ctx, path, args...)
@@ -402,6 +541,9 @@ type ProviderMetadata struct {
 	Up          CommandMetadata  `json:"up"`
 	Down        CommandMetadata  `json:"down"`
 	Stop        *CommandMetadata `json:"stop,omitempty"`
+	// Pull declares support for the image-distribution command; like Stop,
+	// the presence of the block is what opts the provider in.
+	Pull *CommandMetadata `json:"pull,omitempty"`
 }
 
 func (p ProviderMetadata) IsEmpty() bool {
diff --git a/pkg/compose/plugins_control_test.go b/pkg/compose/plugins_control_test.go
index 5d1179c852..3b789df84d 100644
--- a/pkg/compose/plugins_control_test.go
+++ b/pkg/compose/plugins_control_test.go
@@ -51,7 +51,7 @@ func TestExecutePlugin_GetServiceConfig(t *testing.T) {
 			Options: types.MultiOptions{"template": {"agent:latest"}},
 		},
 	}
-	variables, err := svc.(*composeService).executePlugin(cmd, "up", service)
+	variables, err := svc.(*composeService).executePlugin(t.Context(), cmd, "up", service)
 	assert.NilError(t, err)
 	assert.Equal(t, variables.prefixed["TEMPLATE"], "agent:latest")
 	// the channel stays usable for more than one request
diff --git a/pkg/compose/plugins_test.go b/pkg/compose/plugins_test.go
index 9ff2f653c4..93d2b47bc5 100644
--- a/pkg/compose/plugins_test.go
+++ b/pkg/compose/plugins_test.go
@@ -97,6 +97,25 @@ func TestProviderMetadata_StopAbsent(t *testing.T) {
 	assert.Assert(t, metadata.Stop == nil, "Stop should be nil when absent from JSON")
 }
 
+func TestProviderMetadata_PullAbsent(t *testing.T) {
+	raw := `{"description":"x","up":{"parameters":[]},"down":{"parameters":[]}}`
+
+	var metadata ProviderMetadata
+	err := json.Unmarshal([]byte(raw), &metadata)
+	assert.NilError(t, err)
+	assert.Assert(t, metadata.Pull == nil, "Pull should be nil when absent from JSON")
+}
+
+func TestProviderMetadata_PullPresent(t *testing.T) {
+	raw := `{"pull":{"parameters":[{"name":"registry"}]}}`
+
+	var metadata ProviderMetadata
+	err := json.Unmarshal([]byte(raw), &metadata)
+	assert.NilError(t, err)
+	assert.Assert(t, metadata.Pull != nil, "Pull should be non-nil when key present")
+	assert.Equal(t, metadata.Pull.Parameters[0].Name, "registry")
+}
+
 func TestProviderMetadata_StopAdvertisedWithoutParameters(t *testing.T) {
 	raw := `{"stop":{"parameters":null}}`
 
diff --git a/pkg/compose/provider_images.go b/pkg/compose/provider_images.go
new file mode 100644
index 0000000000..0150474b3d
--- /dev/null
+++ b/pkg/compose/provider_images.go
@@ -0,0 +1,132 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package compose
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/compose-spec/compose-go/v2/types"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/docker/compose/v5/pkg/api"
+)
+
+const (
+	// providerImageSourceLocal: the desired state is the local daemon's
+	// image; the provider synchronizes from it (get-image) when its
+	// bookkeeping says it holds something else.
+	providerImageSourceLocal = "local"
+	// providerImageSourceRegistry: the reference is resolvable upstream and
+	// upstream is the authority; local facts are an optimization, never an
+	// obligation.
+	providerImageSourceRegistry = "registry"
+
+	// providerPullPolicyMissing (up path): a usable version present in the
+	// provider's runtime suffices.
+	providerPullPolicyMissing = "missing"
+	// providerPullPolicyAlways (compose pull): the provider must ensure it
+	// holds the latest version of the authority.
+	providerPullPolicyAlways = "always"
+)
+
+// ensureProviderImages gives provider-backed services their turn in the image
+// phase: for every service whose provider declares the pull command (metadata
+// opt-in, like stop), compose invokes
+//
+//	<provider> compose pull --image=<ref> [--digest=… --created=…] --source=… --policy=… <service>
+//
+// The provider owns making the image available to ITS runtime: resolving the
+// reference upstream, or requesting the local bytes with a get-image message.
+// built maps the image names (re)built by the current run, which forces the
+// local-source verdict for them.
+func (s *composeService) ensureProviderImages(ctx context.Context, project *types.Project, built map[string]string, policy string) error {
+	// deliberately NOT errgroup.WithContext: providers are independent, and
+	// fail-fast cancellation would kill the siblings of the first failure
+	// mid-run, mangling their reports into context-canceled noise. Every
+	// provider runs to completion and every failure is reported.
+	var eg errgroup.Group
+	eg.SetLimit(s.maxConcurrency)
+	var mu sync.Mutex
+	var errs []error
+	for _, service := range project.Services {
+		if service.Provider == nil {
+			continue
+		}
+		image := api.GetImageNameOrDefault(service, project.Name)
+		_, justBuilt := built[image]
+		source := providerImageSource(service, justBuilt)
+		eg.Go(func() error {
+			if err := s.runProviderPull(ctx, project, service, image, source, policy); err != nil {
+				mu.Lock()
+				errs = append(errs, fmt.Errorf("provider service %q: %w", service.Name, err))
+				mu.Unlock()
+			}
+			return nil
+		})
+	}
+	_ = eg.Wait()
+	return errors.Join(errs...)
+}
+
+// providerImageSource computes the authority VERDICT for one invocation —
+// compose owns this arbitration (pull_policy, flags, what this run built) so
+// providers never re-implement it:
+//   - "local": the desired state is the local daemon's image — a declared
+//     build with no pullable reference, pull_policy build, or an image the
+//     current run just (re)built;
+//   - "registry": the provider resolves the reference upstream — including
+//     the build-in-CI workflow where a build section is only the recipe used
+//     to publish the image consumers pull.
+func providerImageSource(service types.ServiceConfig, justBuilt bool) string {
+	if service.Build == nil {
+		return providerImageSourceRegistry
+	}
+	if service.Image == "" || justBuilt {
+		return providerImageSourceLocal
+	}
+	if policy, _, err := service.GetPullPolicy(); err == nil && policy == types.PullPolicyBuild {
+		return providerImageSourceLocal
+	}
+	return providerImageSourceRegistry
+}
+
+func (s *composeService) runProviderPull(ctx context.Context, project *types.Project, service types.ServiceConfig, image, source, policy string) error {
+	args := []string{"--image=" + image, "--source=" + source, "--policy=" + policy}
+	if digest, created, ok := s.localImageFacts(ctx, image); ok {
+		args = append(args, "--digest="+digest)
+		if created != "" {
+			args = append(args, "--created="+created)
+		}
+	}
+	return s.runPlugin(ctx, project, service, "pull", args...)
+}
+
+// localImageFacts reports the state of the local daemon cache for ref: the
+// image ID and its creation time. These describe the cache, they are not
+// instructions — the provider persists them as the bookkeeping keys of what
+// it ingested (digest as identity test, created as the ordering fallback for
+// backends that cannot preserve digests).
+func (s *composeService) localImageFacts(ctx context.Context, ref string) (digest, created string, ok bool) {
+	inspected, err := s.apiClient().ImageInspect(ctx, ref)
+	if err != nil {
+		return "", "", false
+	}
+	return inspected.ID, inspected.Created, true
+}
diff --git a/pkg/compose/provider_images_test.go b/pkg/compose/provider_images_test.go
new file mode 100644
index 0000000000..80370d4c95
--- /dev/null
+++ b/pkg/compose/provider_images_test.go
@@ -0,0 +1,266 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package compose
+
+import (
+	"bufio"
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http/httputil"
+	"os"
+	"os/exec"
+	"strconv"
+	"testing"
+
+	"github.com/compose-spec/compose-go/v2/types"
+	"go.uber.org/mock/gomock"
+	"gotest.tools/v3/assert"
+
+	"github.com/docker/compose/v5/pkg/mocks"
+)
+
+// providerSourceService builds a ServiceConfig by assignment: Image and
+// PullPolicy are fields promoted from the embedded ContainerSpec, which
+// struct literals cannot set before go1.27.
+func providerSourceService(build *types.BuildConfig, image, pullPolicy string) types.ServiceConfig {
+	var s types.ServiceConfig
+	s.Build = build
+	s.Image = image
+	s.PullPolicy = pullPolicy
+	return s
+}
+
+func TestProviderImageSource(t *testing.T) {
+	build := &types.BuildConfig{Context: "."}
+	tests := []struct {
+		name      string
+		service   types.ServiceConfig
+		justBuilt bool
+		want      string
+	}{
+		{"no build: registry authority", providerSourceService(nil, "nginx", ""), false, providerImageSourceRegistry},
+		{"build without image: local-only name", providerSourceService(build, "", ""), false, providerImageSourceLocal},
+		{"build just run: local wins for this invocation", providerSourceService(build, "repo/app", ""), true, providerImageSourceLocal},
+		{"build as CI recipe, default policy: registry authority", providerSourceService(build, "repo/app", ""), false, providerImageSourceRegistry},
+		{"pull_policy build: local authority", providerSourceService(build, "repo/app", types.PullPolicyBuild), false, providerImageSourceLocal},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, providerImageSource(tc.service, tc.justBuilt), tc.want)
+		})
+	}
+}
+
+// TestStreamImageTo locks the wire format of the get-image answer: one JSON
+// announce line, then the tar bytes as an HTTP/1.1 chunked body terminated by
+// the zero chunk and a final CRLF — decodable with a stock chunked reader.
+func TestStreamImageTo(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	cli := mocks.NewMockCli(mockCtrl)
+	apiClient := mocks.NewMockAPIClient(mockCtrl)
+	cli.EXPECT().Client().Return(apiClient).AnyTimes()
+	svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{}))
+	assert.NilError(t, err)
+	s := svc.(*composeService)
+
+	payload := bytes.Repeat([]byte("compose-image-bytes\x00\x17\xff"), 100_000) // > one chunk, binary-hostile values included
+	apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-app"}).
+		Return(fakeImageSaveResult{io.NopCloser(bytes.NewReader(payload))}, nil)
+
+	var out bytes.Buffer
+	assert.NilError(t, s.streamImageTo(t.Context(), &out, "proj-app", ""))
+
+	rd := bufio.NewReader(&out)
+	line, err := rd.ReadString('\n')
+	assert.NilError(t, err)
+	var announce imageStreamAnswer
+	assert.NilError(t, json.Unmarshal([]byte(line), &announce))
+	assert.Equal(t, announce.Type, ImageStreamType)
+	assert.Equal(t, announce.Encoding, "chunked")
+	assert.Equal(t, announce.Error, "")
+
+	body, err := io.ReadAll(httputil.NewChunkedReader(rd))
+	assert.NilError(t, err)
+	assert.Assert(t, bytes.Equal(body, payload), "chunked round-trip must be byte-identical (%d vs %d bytes)", len(body), len(payload))
+
+	// nothing may follow the zero chunk: a stock chunked reader stops there
+	// without consuming further bytes, and any leftover would corrupt the
+	// next answer of a provider requesting several images
+	rest, err := io.ReadAll(rd)
+	assert.NilError(t, err)
+	assert.Equal(t, string(rest), "", "no byte may follow the terminating zero chunk")
+}
+
+// TestStreamImageTo_SaveError: when the daemon cannot export the image, the
+// announce line carries the error and no stream follows — the provider is
+// never left waiting.
+func TestStreamImageTo_SaveError(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	cli := mocks.NewMockCli(mockCtrl)
+	apiClient := mocks.NewMockAPIClient(mockCtrl)
+	cli.EXPECT().Client().Return(apiClient).AnyTimes()
+	svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{}))
+	assert.NilError(t, err)
+	s := svc.(*composeService)
+
+	apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-app"}).
+		Return(nil, errors.New("no such image"))
+
+	var out bytes.Buffer
+	assert.ErrorContains(t, s.streamImageTo(t.Context(), &out, "proj-app", ""), "no such image")
+
+	var announce imageStreamAnswer
+	assert.NilError(t, json.Unmarshal(out.Bytes(), &announce))
+	assert.Equal(t, announce.Type, ImageStreamType)
+	assert.ErrorContains(t, errors.New(announce.Error), "no such image")
+	assert.Equal(t, announce.Encoding, "")
+}
+
+type fakeImageSaveResult struct{ io.ReadCloser }
+
+// TestExecutePlugin_GetImage runs the pull command against a fake provider
+// (this test binary re-executed, see TestHelperProviderPull): the get-image
+// request must be answered with the announced chunked stream, which the
+// provider consumes and acknowledges by hashing.
+func TestExecutePlugin_GetImage(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	cli := mocks.NewMockCli(mockCtrl)
+	apiClient := mocks.NewMockAPIClient(mockCtrl)
+	cli.EXPECT().Client().Return(apiClient).AnyTimes()
+	svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{}))
+	assert.NilError(t, err)
+
+	payload := bytes.Repeat([]byte{0x17, 0x00, 0xAB}, 500_000) // ETB and NUL laden
+	sum := sha256.Sum256(payload)
+	apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-db"}).
+		Return(fakeImageSaveResult{io.NopCloser(bytes.NewReader(payload))}, nil)
+
+	cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull")
+	cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1")
+
+	service := types.ServiceConfig{
+		Name:     "db",
+		Provider: &types.ServiceProviderConfig{Type: "fake"},
+	}
+	variables, err := svc.(*composeService).executePlugin(t.Context(), cmd, "pull", service)
+	assert.NilError(t, err)
+	assert.Equal(t, variables.prefixed["SHA"], hex.EncodeToString(sum[:]))
+	assert.Equal(t, variables.prefixed["LEN"], strconv.Itoa(len(payload)))
+}
+
+// TestHelperProviderPull is not a test: it is the fake provider process
+// spawned by TestExecutePlugin_GetImage. It requests the image, decodes the
+// chunked stream with the stock reader, and reports what it received.
+func TestHelperProviderPull(t *testing.T) {
+	if os.Getenv("GO_WANT_HELPER_PROCESS") != "1" {
+		t.Skip("helper process for TestExecutePlugin_GetImage")
+	}
+	emit := func(msg JsonMessage) {
+		if err := json.NewEncoder(os.Stdout).Encode(msg); err != nil {
+			os.Exit(1)
+		}
+	}
+	emit(JsonMessage{Type: GetImageType, Message: "proj-db"})
+
+	stdin := bufio.NewReader(os.Stdin)
+	line, err := stdin.ReadString('\n')
+	if err != nil {
+		emit(JsonMessage{Type: ErrorType, Message: "reading announce: " + err.Error()})
+		os.Exit(1)
+	}
+	var announce imageStreamAnswer
+	if err := json.Unmarshal([]byte(line), &announce); err != nil || announce.Error != "" || announce.Encoding != "chunked" {
+		emit(JsonMessage{Type: ErrorType, Message: fmt.Sprintf("bad announce %q (err %v)", line, err)})
+		os.Exit(1)
+	}
+	h := sha256.New()
+	n, err := io.Copy(h, httputil.NewChunkedReader(stdin))
+	if err != nil {
+		emit(JsonMessage{Type: ErrorType, Message: "reading stream: " + err.Error()})
+		os.Exit(1)
+	}
+	emit(JsonMessage{Type: SetEnvType, Message: "SHA=" + hex.EncodeToString(h.Sum(nil))})
+	emit(JsonMessage{Type: SetEnvType, Message: fmt.Sprintf("LEN=%d", n)})
+	os.Exit(0)
+}
+
+// TestExecutePlugin_GetImageRefusedOutsidePull: image distribution belongs to
+// the pull command — a get-image emitted during up is a protocol error, so an
+// image export can never stall another lifecycle command.
+func TestExecutePlugin_GetImageRefusedOutsidePull(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	cli := mocks.NewMockCli(mockCtrl)
+	apiClient := mocks.NewMockAPIClient(mockCtrl)
+	cli.EXPECT().Client().Return(apiClient).AnyTimes()
+	svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{}))
+	assert.NilError(t, err)
+
+	cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull")
+	cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1")
+
+	service := types.ServiceConfig{
+		Name:     "db",
+		Provider: &types.ServiceProviderConfig{Type: "fake"},
+	}
+	_, err = svc.(*composeService).executePlugin(t.Context(), cmd, "up", service)
+	assert.ErrorContains(t, err, "get-image is only supported during the pull command")
+}
+
+// failingReader errors after serving its prefix, simulating a daemon export
+// dying mid-stream.
+type failingReader struct{ data []byte }
+
+func (f *failingReader) Read(p []byte) (int, error) {
+	if len(f.data) == 0 {
+		return 0, errors.New("export died mid-stream")
+	}
+	n := copy(p, f.data)
+	f.data = f.data[n:]
+	return n, nil
+}
+func (f *failingReader) Close() error { return nil }
+
+// TestExecutePlugin_GetImageAbortedMidStream: a transfer failing after the
+// success announce must end observably — compose closes the answer channel,
+// the provider reads EOF mid-chunk and reports, instead of blocking forever
+// on a stream nobody will finish.
+func TestExecutePlugin_GetImageAbortedMidStream(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	cli := mocks.NewMockCli(mockCtrl)
+	apiClient := mocks.NewMockAPIClient(mockCtrl)
+	cli.EXPECT().Client().Return(apiClient).AnyTimes()
+	svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{}))
+	assert.NilError(t, err)
+
+	apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-db"}).
+		Return(&failingReader{data: bytes.Repeat([]byte{0xAB}, 4096)}, nil)
+
+	cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull")
+	cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1")
+
+	service := types.ServiceConfig{
+		Name:     "db",
+		Provider: &types.ServiceProviderConfig{Type: "fake"},
+	}
+	_, err = svc.(*composeService).executePlugin(t.Context(), cmd, "pull", service)
+	assert.ErrorContains(t, err, "reading stream")
+}
diff --git a/pkg/compose/pull.go b/pkg/compose/pull.go
index b8a935fbf3..f35baa2f38 100644
--- a/pkg/compose/pull.go
+++ b/pkg/compose/pull.go
@@ -64,6 +64,7 @@ type imagePuller struct {
 }
 
 func (s *composeService) pull(ctx context.Context, project *types.Project, opts api.PullOptions) error {
+	rootCtx := ctx
 	images, _, err := s.getLocalImagesDigests(ctx, project)
 	if err != nil {
 		return err
@@ -98,8 +99,20 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts
 		logrus.Warnf("WARNING: Some service image(s) must be built from source by running:\n    docker compose build %s", strings.Join(p.mustBuild, " "))
 	}
 
+	// provider-backed services: pull is the strong contract — the provider
+	// must ensure its runtime holds the latest version of the authority.
+	// It runs even when a regular pull failed: skipping the provider phase
+	// because an unrelated service failed would silently leave provider
+	// runtimes stale. Its error is a per-service pull error and follows
+	// their exact regime — reported below, suppressed by IgnoreFailures.
+	// rootCtx, not ctx: the errgroup context above is canceled once Wait
+	// returns, and would kill the provider on the spot.
+	p.pullErrors = append(p.pullErrors, s.ensureProviderImages(rootCtx, project, nil, providerPullPolicyAlways))
+
 	if err != nil {
-		return err
+		// fatal errgroup error: report it with whatever per-service context
+		// accumulated (provider phase included) instead of dropping it
+		return errors.Join(append(p.pullErrors, err)...)
 	}
 	if opts.IgnoreFailures {
 		return nil
diff --git a/pkg/compose/run.go b/pkg/compose/run.go
index 1f233fe32f..081a908290 100644
--- a/pkg/compose/run.go
+++ b/pkg/compose/run.go
@@ -160,7 +160,7 @@ func (s *composeService) prepareRun(ctx context.Context, project *types.Project,
 
 	// Only ensure image exists for the target service, dependencies were already handled by startDependencies
 	buildOpts := prepareBuildOptions(options)
-	if err := s.ensureImagesExists(ctx, project, buildOpts, options.QuietPull); err != nil { // all dependencies already checked, but might miss service img
+	if err := s.ensureImagesExists(ctx, project, buildOpts, options.QuietPull, false); err != nil { // all dependencies already checked, but might miss service img
 		return prepareRunResult{}, err
 	}
 
diff --git a/pkg/compose/service_containers.go b/pkg/compose/service_containers.go
index af7f35dd3e..790d5c4f20 100644
--- a/pkg/compose/service_containers.go
+++ b/pkg/compose/service_containers.go
@@ -578,7 +578,6 @@ func (s *composeService) startService(ctx context.Context,
 	if service.Deploy != nil && service.Deploy.Replicas != nil && *service.Deploy.Replicas == 0 {
 		return nil
 	}
-
 	err := s.waitDependencies(ctx, project, service.Name, service.DependsOn, containers, timeout)
 	if err != nil {
 		return err
@@ -588,6 +587,13 @@ func (s *composeService) startService(ctx context.Context,
 		if service.GetScale() == 0 {
 			return nil
 		}
+		if service.Provider != nil {
+			// a provider-backed service usually has no container of its own
+			// (it gets one — the relay — only when the provider published
+			// endpoints), so a project made only of provider services
+			// legitimately reaches the start phase with no container at all
+			return nil
+		}
 		return fmt.Errorf("service %q has no container to start", service.Name)
 	}
 
diff --git a/pkg/e2e/providers_test.go b/pkg/e2e/providers_test.go
index 999c9df281..e905f60816 100644
--- a/pkg/e2e/providers_test.go
+++ b/pkg/e2e/providers_test.go
@@ -56,6 +56,24 @@ func TestProviderStopHook(t *testing.T) {
 			FileExists(marker))
 }
 
+func TestProviderImagePull(t *testing.T) {
+	// The example provider's pull subcommand requests the image bytes with a
+	// get-image message and records what it received (digest facts, sha256,
+	// size) at PROVIDER_PULL_MARKER.
+	marker := filepath.Join(t.TempDir(), "example-provider-pull-marker")
+	s := providerScenario(t, "a provider-backed service with a build must get the built image distributed to the provider")
+	s.Env("PROVIDER_PULL_MARKER=" + marker)
+	s.Step("up builds the image and streams it to the provider with the local-authority verdict",
+		ComposeCmd("up", "-d"),
+		FileContains(marker, "source=local"),
+		FileContains(marker, "policy=missing"),
+		FileContains(marker, "sha256="),
+		FileContains(marker, "digest=sha256:")).
+		Step("pull re-invokes the provider under the freshness contract",
+			ComposeCmd("pull"),
+			FileContains(marker, "policy=always"))
+}
+
 func TestDependsOnMultipleProviders(t *testing.T) {
 	providerScenario(t, "a service depending on several providers must receive each provider's variables").
 		Step("the service sees both providers' URLs",
diff --git a/pkg/e2e/testdata/TestProviderImagePull/Dockerfile b/pkg/e2e/testdata/TestProviderImagePull/Dockerfile
new file mode 100644
index 0000000000..7a34c97125
--- /dev/null
+++ b/pkg/e2e/testdata/TestProviderImagePull/Dockerfile
@@ -0,0 +1,2 @@
+FROM alpine
+RUN echo provider-image-pull-fixture > /fixture.txt
diff --git a/pkg/e2e/testdata/TestProviderImagePull/compose.yaml b/pkg/e2e/testdata/TestProviderImagePull/compose.yaml
new file mode 100644
index 0000000000..a210d17c3b
--- /dev/null
+++ b/pkg/e2e/testdata/TestProviderImagePull/compose.yaml
@@ -0,0 +1,9 @@
+services:
+  db:
+    build:
+      context: .
+    provider:
+      type: example-provider
+      options:
+        type: mysql
+        name: myDB