diff --git a/crates/ployz-build/src/lib.rs b/crates/ployz-build/src/lib.rs
index f1d30c58c6..bb560b4877 100644
--- a/crates/ployz-build/src/lib.rs
+++ b/crates/ployz-build/src/lib.rs
@@ -127,15 +127,11 @@ pub struct BuiltImage {
 }
 
 impl BuiltImage {
-    /// Name this manifest in the repository used by its Build.
+    /// Name this manifest in the caller's requested repository.
     /// # Errors
-    /// Rejects missing or malformed repository evidence.
-    pub fn repository_reference(&self) -> Result<String, BuildError> {
-        let tag = self
-            .tags
-            .first()
-            .ok_or_else(|| BuildError::Result("Build image has no tags".into()))?;
-        let reference = tag
+    /// Rejects malformed repository references.
+    pub fn repository_reference(&self, image: &str) -> Result<String, BuildError> {
+        let reference = image
             .parse::<oci_client::Reference>()
             .map_err(|error| BuildError::Result(error.to_string()))?;
         Ok(format!(
diff --git a/crates/ployz-build/src/tests.rs b/crates/ployz-build/src/tests.rs
index 2357dfbf02..d8b7e1a3d3 100644
--- a/crates/ployz-build/src/tests.rs
+++ b/crates/ployz-build/src/tests.rs
@@ -552,3 +552,17 @@ exit 0
     );
     std::fs::remove_dir_all(directory).unwrap();
 }
+
+#[test]
+fn repository_reference_rejects_a_malformed_service_image() {
+    let image = BuiltImage {
+        reference: format!("sha256:{}", "1".repeat(64)),
+        tags: vec!["example.test/api:latest".into()],
+        platforms: vec!["linux/amd64".into()],
+        location: "unix:///var/run/docker.sock".into(),
+    };
+    assert!(matches!(
+        image.repository_reference("not a repository"),
+        Err(BuildError::Result(_))
+    ));
+}
diff --git a/crates/ployz/src/compose/build.rs b/crates/ployz/src/compose/build.rs
index 5faf8586ac..dc4cf8d471 100644
--- a/crates/ployz/src/compose/build.rs
+++ b/crates/ployz/src/compose/build.rs
@@ -858,7 +858,7 @@ fn ssh_paths(key: &Value) -> Result<(&str, &str), ComposeError> {
         })
 }
 
-fn is_remote_context(source: &str) -> bool {
+pub(super) fn is_remote_context(source: &str) -> bool {
     source.contains("://") || source.starts_with("git@") || source.starts_with("service:")
 }
 
diff --git a/crates/ployz/src/compose/build_inputs.rs b/crates/ployz/src/compose/build_inputs.rs
index daa1f3a689..9b0c529e66 100644
--- a/crates/ployz/src/compose/build_inputs.rs
+++ b/crates/ployz/src/compose/build_inputs.rs
@@ -16,6 +16,8 @@ use sha2::{Digest as _, Sha256};
 
 use super::ComposeError;
 
+mod service;
+
 /// Private, attempt-scoped copies. Never persisted as deployment history.
 pub(super) struct BuildInputs {
     root: PathBuf,
@@ -124,51 +126,6 @@ impl BuildInputs {
         })
     }
 
-    /// Isolate one target's generated recipe while retaining captured bytes.
-    pub(super) fn for_service(&self, name: &str) -> Result<Self, ComposeError> {
-        let inputs = Self::new()?;
-        // ponytail: copy the already-filtered capture per remote target. Select
-        // per-target source trees if large multi-Service captures become costly.
-        for area in ["source", "private"] {
-            let source = self.root.join(area);
-            for entry in entries(&source, &source, None).map_err(input_error)? {
-                copy(
-                    &entry,
-                    &inputs
-                        .root
-                        .join(area)
-                        .join(entry.file_name().expect("entry")),
-                    &source,
-                    None,
-                )
-                .map_err(input_error)?;
-            }
-        }
-        let metadata = inputs.root.join("private/railpack.json");
-        if metadata.exists() {
-            let mut recipes: Vec<ployz_build::Railpack> =
-                serde_json::from_slice(&fs::read(&metadata).map_err(input_error)?)
-                    .map_err(|error| ComposeError::Invalid(error.to_string()))?;
-            recipes.retain(|recipe| recipe.name == name);
-            fs::remove_file(metadata).map_err(input_error)?;
-            inputs.railpack(&recipes)?;
-        }
-        let mut recipe: serde_norway::Value = serde_norway::from_slice(
-            &fs::read(self.root.join("compose.yaml")).map_err(input_error)?,
-        )
-        .map_err(|error| ComposeError::Invalid(error.to_string()))?;
-        recipe
-            .get_mut("services")
-            .and_then(serde_norway::Value::as_mapping_mut)
-            .ok_or_else(|| ComposeError::Invalid("captured Build has no services".into()))?
-            .retain(|key, _| key.as_str() == Some(name));
-        inputs.compose(
-            &serde_norway::to_string(&recipe)
-                .map_err(|error| ComposeError::Invalid(error.to_string()))?,
-        )?;
-        Ok(inputs)
-    }
-
     /// Copy a source once, rejecting edits observed during the copy.
     ///
     /// # Errors
@@ -675,42 +632,6 @@ fn validate_link(path: &Path, root: &Path, selection: Option<&Selection>) -> io:
 mod tests {
     use super::*;
 
-    #[test]
-    fn per_service_capture_keeps_only_its_railpack_recipe() {
-        let inputs = BuildInputs::new().unwrap();
-        inputs
-            .compose("services: {api: {build: .}, web: {build: .}, base: {build: .}}\n")
-            .unwrap();
-        inputs
-            .railpack(&["api", "web"].map(|name| ployz_build::Railpack {
-                name: name.into(),
-                context: "source".into(),
-                variables: BTreeMap::new(),
-                refresh_cache: false,
-            }))
-            .unwrap();
-        for name in ["api", "web", "base"] {
-            let selected = inputs.for_service(name).unwrap();
-            let metadata = selected.root().join("private/railpack.json");
-            if name == "base" {
-                assert!(!metadata.exists());
-            } else {
-                let recipes: Vec<ployz_build::Railpack> =
-                    serde_json::from_slice(&fs::read(metadata).unwrap()).unwrap();
-                assert_eq!(recipes.len(), 1);
-                assert_eq!(recipes.first().unwrap().name, name);
-            }
-        }
-        let original: Vec<ployz_build::Railpack> =
-            serde_json::from_slice(&fs::read(inputs.root().join("private/railpack.json")).unwrap())
-                .unwrap();
-        assert_eq!(
-            original.len(),
-            2,
-            "per-target selection changed the original capture"
-        );
-    }
-
     use std::os::unix::net::UnixListener;
 
     #[test]
diff --git a/crates/ployz/src/compose/build_inputs/service.rs b/crates/ployz/src/compose/build_inputs/service.rs
new file mode 100644
index 0000000000..e71dd4a986
--- /dev/null
+++ b/crates/ployz/src/compose/build_inputs/service.rs
@@ -0,0 +1,285 @@
+//! Materialize only the captured inputs referenced by one remote Build.
+
+use serde_norway::Value;
+
+use crate::compose::build::is_remote_context;
+
+use super::*;
+
+impl BuildInputs {
+    pub(in crate::compose) fn for_service(&self, name: &str) -> Result<Self, ComposeError> {
+        let invalid = || ComposeError::Invalid("invalid captured Build recipe".into());
+        let mut recipe: Value = serde_norway::from_slice(
+            &fs::read(self.root.join("compose.yaml")).map_err(input_error)?,
+        )
+        .map_err(|_| invalid())?;
+        let services = recipe
+            .get_mut("services")
+            .and_then(Value::as_mapping_mut)
+            .ok_or_else(invalid)?;
+        services.retain(|key, _| key.as_str() == Some(name));
+        let build = services
+            .get(name)
+            .and_then(|service| service.get("build"))
+            .and_then(Value::as_mapping)
+            .ok_or_else(invalid)?;
+        let mut paths = BTreeSet::new();
+        let context = build
+            .get("context")
+            .and_then(Value::as_str)
+            .ok_or_else(invalid)?;
+        if !is_remote_context(context) {
+            paths.insert(PathBuf::from(context));
+            if let Some(dockerfile) = build.get("dockerfile").and_then(Value::as_str) {
+                let dockerfile = Path::new(context).join(dockerfile);
+                let mut ignore = dockerfile.as_os_str().to_owned();
+                ignore.push(".dockerignore");
+                if self.root.join(&ignore).try_exists().map_err(input_error)? {
+                    paths.insert(ignore.into());
+                }
+                paths.insert(dockerfile);
+            }
+        }
+        if let Some(contexts) = build.get("additional_contexts") {
+            let local_path =
+                |context: &str| (!is_remote_context(context)).then(|| PathBuf::from(context));
+            match contexts {
+                Value::Mapping(values) => paths.extend(
+                    values
+                        .values()
+                        .filter_map(Value::as_str)
+                        .filter_map(local_path),
+                ),
+                Value::Sequence(values) => paths.extend(
+                    values
+                        .iter()
+                        .filter_map(|value| value.as_str()?.split_once('=').map(|(_, path)| path))
+                        .filter_map(local_path),
+                ),
+                Value::Null => {}
+                Value::Bool(_) | Value::Number(_) | Value::String(_) | Value::Tagged(_) => {
+                    return Err(invalid());
+                }
+            }
+        }
+        if let Some(ssh) = build.get("ssh").and_then(Value::as_sequence) {
+            for key in ssh {
+                let (_, keys) = key
+                    .as_str()
+                    .and_then(|key| key.split_once('='))
+                    .ok_or_else(invalid)?;
+                paths.extend(keys.split(',').map(PathBuf::from));
+            }
+        }
+        let names = build
+            .get("secrets")
+            .and_then(Value::as_sequence)
+            .into_iter()
+            .flatten()
+            .map(|secret| {
+                secret
+                    .as_str()
+                    .or_else(|| secret.get("source").and_then(Value::as_str))
+                    .map(str::to_owned)
+                    .ok_or_else(invalid)
+            })
+            .collect::<Result<BTreeSet<_>, _>>()?;
+        if let Some(secrets) = recipe.get_mut("secrets").and_then(Value::as_mapping_mut) {
+            secrets.retain(|key, _| key.as_str().is_some_and(|name| names.contains(name)));
+            for secret in secrets.values() {
+                paths.insert(PathBuf::from(
+                    secret
+                        .get("file")
+                        .and_then(Value::as_str)
+                        .ok_or_else(invalid)?,
+                ));
+            }
+        }
+        // The explicitly supplied, normalized Docker configuration is command-wide.
+        // Its original capture can contain discarded fields and is never uploaded.
+        if self.root.join("private/docker").exists() {
+            paths.insert("private/docker".into());
+        }
+        let inputs = Self::new()?;
+        let paths = paths
+            .into_iter()
+            .map(|path| {
+                let source = self.root.join(path).canonicalize().map_err(input_error)?;
+                let relative = source
+                    .strip_prefix(&self.root)
+                    .map_err(|_| invalid())?
+                    .to_owned();
+                if !relative.starts_with("source") && !relative.starts_with("private") {
+                    return Err(invalid());
+                }
+                Ok(relative)
+            })
+            .collect::<Result<BTreeSet<_>, ComposeError>>()?;
+        for path in &paths {
+            if paths
+                .iter()
+                .any(|parent| parent != path && path.starts_with(parent))
+            {
+                continue;
+            }
+            let source = self.root.join(path);
+            copy(&source, &inputs.root.join(path), &source, None).map_err(input_error)?;
+        }
+        let metadata = self.root.join("private/railpack.json");
+        if metadata.exists() {
+            let mut recipes: Vec<ployz_build::Railpack> =
+                serde_json::from_slice(&fs::read(metadata).map_err(input_error)?)
+                    .map_err(|_| invalid())?;
+            recipes.retain(|recipe| recipe.name == name);
+            inputs.railpack(&recipes)?;
+        }
+        inputs.compose(&serde_norway::to_string(&recipe).map_err(|_| invalid())?)?;
+        Ok(inputs)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn per_service_upload_contains_only_referenced_inputs() {
+        let inputs = BuildInputs::new().unwrap();
+        for path in [
+            "source/api",
+            "source/web",
+            "source/shared",
+            "private/docker",
+        ] {
+            fs::create_dir(inputs.root.join(path)).unwrap();
+        }
+        for path in [
+            "source/api/code",
+            "source/web/code",
+            "source/shared/artifact",
+            "source/api.Dockerfile",
+            "source/api.Dockerfile.dockerignore",
+            "source/web.Dockerfile",
+            "private/api-key",
+            "private/web-key",
+            "private/api-token",
+            "private/web-token",
+            "private/shared-token",
+            "private/docker/config.json",
+            "private/original-docker-config",
+        ] {
+            fs::write(inputs.root.join(path), path).unwrap();
+        }
+        inputs
+            .compose(
+                r#"
+services:
+  api:
+    build:
+      context: source/api
+      dockerfile: ../api.Dockerfile
+      additional_contexts: {shared: source/shared, base: 'service:web'}
+      ssh: ['default=private/api-key']
+      secrets: [api-token, {source: shared-token, target: shared}]
+  web:
+    build:
+      context: source/web
+      dockerfile: ../web.Dockerfile
+      additional_contexts: ['shared=source/shared']
+      ssh: ['default=private/web-key']
+      secrets: [web-token, shared-token]
+secrets:
+  api-token: {file: private/api-token}
+  web-token: {file: private/web-token}
+  shared-token: {file: private/shared-token}
+"#,
+            )
+            .unwrap();
+        for (name, other) in [("api", "web"), ("web", "api")] {
+            let upload = inputs.for_service(name).unwrap();
+            for required in [
+                format!("source/{name}/code"),
+                format!("source/{name}.Dockerfile"),
+                format!("private/{name}-key"),
+                format!("private/{name}-token"),
+                "source/shared/artifact".into(),
+                "private/shared-token".into(),
+                "private/docker/config.json".into(),
+            ] {
+                assert!(
+                    upload.root.join(&required).exists(),
+                    "{name} lost {required}"
+                );
+            }
+            for forbidden in [
+                format!("source/{other}"),
+                format!("source/{other}.Dockerfile"),
+                format!("private/{other}-key"),
+                format!("private/{other}-token"),
+                "private/original-docker-config".into(),
+            ] {
+                assert!(
+                    !upload.root.join(&forbidden).exists(),
+                    "{name} received {forbidden}"
+                );
+            }
+            assert_eq!(
+                upload
+                    .root
+                    .join("source/api.Dockerfile.dockerignore")
+                    .exists(),
+                name == "api"
+            );
+            let recipe: serde_norway::Value =
+                serde_norway::from_slice(&fs::read(upload.root.join("compose.yaml")).unwrap())
+                    .unwrap();
+            assert!(
+                recipe
+                    .get("secrets")
+                    .unwrap()
+                    .get(format!("{other}-token"))
+                    .is_none()
+            );
+            assert!(recipe.get("services").unwrap().get(other).is_none());
+        }
+    }
+
+    #[test]
+    fn per_service_capture_keeps_only_its_railpack_recipe() {
+        let inputs = BuildInputs::new().unwrap();
+        for name in ["api", "web", "base"] {
+            fs::create_dir(inputs.root.join("source").join(name)).unwrap();
+        }
+        inputs
+            .compose("services: {api: {build: {context: source/api}}, web: {build: {context: source/web}}, base: {build: {context: source/base}}}\n")
+            .unwrap();
+        inputs
+            .railpack(&["api", "web"].map(|name| ployz_build::Railpack {
+                name: name.into(),
+                context: format!("source/{name}").into(),
+                variables: BTreeMap::new(),
+                refresh_cache: false,
+            }))
+            .unwrap();
+        for name in ["api", "web", "base"] {
+            let selected = inputs.for_service(name).unwrap();
+            let metadata = selected.root().join("private/railpack.json");
+            if name == "base" {
+                assert!(!metadata.exists());
+            } else {
+                let recipes: Vec<ployz_build::Railpack> =
+                    serde_json::from_slice(&fs::read(metadata).unwrap()).unwrap();
+                assert_eq!(recipes.len(), 1);
+                assert_eq!(recipes.first().unwrap().name, name);
+            }
+        }
+        let original: Vec<ployz_build::Railpack> =
+            serde_json::from_slice(&fs::read(inputs.root().join("private/railpack.json")).unwrap())
+                .unwrap();
+        assert_eq!(
+            original.len(),
+            2,
+            "per-target selection changed the original capture"
+        );
+    }
+}
diff --git a/crates/ployz/src/compose/capture.rs b/crates/ployz/src/compose/capture.rs
index 4671f04416..6462c49536 100644
--- a/crates/ployz/src/compose/capture.rs
+++ b/crates/ployz/src/compose/capture.rs
@@ -41,12 +41,14 @@ impl CapturedCompose {
                 .iter()
                 .find(|build| build.name == service.name.as_str())
             {
-                service.container.image = build.built.repository_reference().map_err(|error| {
-                    super::ComposeError::Build {
-                        services: build.name.clone(),
-                        source: error,
-                    }
-                })?;
+                service.container.image =
+                    build
+                        .built
+                        .repository_reference(&build.image)
+                        .map_err(|error| super::ComposeError::Build {
+                            services: build.name.clone(),
+                            source: error,
+                        })?;
                 service.container.pull_policy = ployz_core::PullPolicy::Never;
             }
         }
@@ -140,7 +142,10 @@ fn deploy_binds_each_service_to_its_build_when_requested_tags_are_shared() {
         location: BuildLocation::Machine(ployz_core::MachineId::parse("a".repeat(32)).unwrap()),
         built: ployz_build::BuiltImage {
             reference: digest.into(),
-            tags: vec!["example.test/shared:latest".into()],
+            tags: vec![
+                "auxiliary.test:5000/other:extra".into(),
+                "example.test/shared:latest".into(),
+            ],
             platforms: vec!["linux/amd64".into()],
             location: "unix:///var/run/docker.sock".into(),
         },
diff --git a/crates/ployz/src/compose/remote_steps.rs b/crates/ployz/src/compose/remote_steps.rs
index 67eca93644..aeb1eeb0d0 100644
--- a/crates/ployz/src/compose/remote_steps.rs
+++ b/crates/ployz/src/compose/remote_steps.rs
@@ -165,9 +165,12 @@ impl CapturedBuild {
                 contexts.insert(
                     dependency.to_owned(),
                     ployz_build::ImageContext {
-                        reference: image.built.repository_reference().map_err(|error| {
-                            failed(Stage::Preparation, error.to_string()).with_work(work.clone())
-                        })?,
+                        reference: image.built.repository_reference(&image.image).map_err(
+                            |error| {
+                                failed(Stage::Preparation, error.to_string())
+                                    .with_work(work.clone())
+                            },
+                        )?,
                         platforms: image.built.platforms.clone(),
                         source,
                     },
diff --git a/crates/ployz/src/deploy/pipeline.rs b/crates/ployz/src/deploy/pipeline.rs
index ecb74b06cf..a34047c7ef 100644
--- a/crates/ployz/src/deploy/pipeline.rs
+++ b/crates/ployz/src/deploy/pipeline.rs
@@ -557,7 +557,7 @@ async fn push_image(
             // Retain each digest even when multiple Services requested one tag.
             let tag = service
                 .built
-                .repository_reference()
+                .repository_reference(&service.image)
                 .map_err(|error| crate::image::PushError::InvalidReference {
                     reference: service.built.reference.clone(),
                     message: error.to_string(),
@@ -577,6 +577,7 @@ async fn push_image(
             crate::image::push_from_machine_using_machines(
                 client,
                 &service.built,
+                &service.image,
                 source,
                 targets,
                 machines,
diff --git a/crates/ployz/src/handlers/deploy_remote_tests.rs b/crates/ployz/src/handlers/deploy_remote_tests.rs
index 59bab96a1d..9c1a9bd7a1 100644
--- a/crates/ployz/src/handlers/deploy_remote_tests.rs
+++ b/crates/ployz/src/handlers/deploy_remote_tests.rs
@@ -172,7 +172,10 @@ async fn remote_transfer_keeps_exact_source_successes_failures_and_omissions() {
     );
     let image = ployz_build::BuiltImage {
         reference: format!("sha256:{}", "1".repeat(64)),
-        tags: vec!["registry.invalid/shared:latest".into()],
+        tags: vec![
+            "auxiliary.test:5000/other:extra".into(),
+            "registry.invalid/shared:latest".into(),
+        ],
         platforms: vec!["linux/amd64".into()],
         location: "unix:///var/run/docker.sock".into(),
     };
@@ -209,6 +212,7 @@ async fn remote_transfer_keeps_exact_source_successes_failures_and_omissions() {
     let result = crate::image::push_from_machine(
         &mut client,
         &image,
+        "registry.invalid/shared:latest",
         source.machine.id,
         &[],
         &tokio_util::sync::CancellationToken::new(),
@@ -238,17 +242,13 @@ async fn remote_transfer_keeps_exact_source_successes_failures_and_omissions() {
             .contains("containerd image store")
     );
     assert_eq!(result.omissions, [missing.machine.id]);
-    assert!(
-        builds
-            .pulls
-            .lock()
-            .unwrap()
-            .iter()
-            .all(
-                |(_, pull)| pull.image == image.repository_reference().unwrap()
-                    && pull.source.management_address == source.machine.management_address()
-            )
-    );
+    assert!(builds.pulls.lock().unwrap().iter().all(|(_, pull)| {
+        pull.image
+            == image
+                .repository_reference("registry.invalid/shared:latest")
+                .unwrap()
+            && pull.source.management_address == source.machine.management_address()
+    }));
     builds
         .stores
         .lock()
@@ -260,6 +260,7 @@ async fn remote_transfer_keeps_exact_source_successes_failures_and_omissions() {
         crate::image::push_from_machine(
             &mut client,
             &image,
+            "registry.invalid/shared:latest",
             source.machine.id,
             &[success.machine.id.to_string()],
             &tokio_util::sync::CancellationToken::new(),
diff --git a/crates/ployz/src/image/built.rs b/crates/ployz/src/image/built.rs
index 9a9aa90a36..3d8e1179c3 100644
--- a/crates/ployz/src/image/built.rs
+++ b/crates/ployz/src/image/built.rs
@@ -13,6 +13,7 @@ use super::*;
 pub async fn push_from_machine(
     client: &mut Client,
     image: &BuiltImage,
+    repository: &str,
     source: MachineId,
     selectors: &[String],
     cancellation: &tokio_util::sync::CancellationToken,
@@ -20,13 +21,22 @@ pub async fn push_from_machine(
     let machines = Cancellation::new(cancellation)
         .race(client.machines())
         .await??;
-    push_from_machine_using_machines(client, image, source, selectors, &machines, cancellation)
-        .await
+    push_from_machine_using_machines(
+        client,
+        image,
+        repository,
+        source,
+        selectors,
+        &machines,
+        cancellation,
+    )
+    .await
 }
 
 pub(crate) async fn push_from_machine_using_machines(
     client: &mut Client,
     image: &BuiltImage,
+    repository: &str,
     source: MachineId,
     selectors: &[String],
     machines: &[MachineObservation],
@@ -42,12 +52,13 @@ pub(crate) async fn push_from_machine_using_machines(
         return Ok(result);
     }
     let source = serve_build_image(client, image, source, cancellation).await?;
-    let reference = image
-        .repository_reference()
-        .map_err(|error| PushError::InvalidReference {
-            reference: image.reference.clone(),
-            message: error.to_string(),
-        })?;
+    let reference =
+        image
+            .repository_reference(repository)
+            .map_err(|error| PushError::InvalidReference {
+                reference: image.reference.clone(),
+                message: error.to_string(),
+            })?;
     let mut cancellation = Cancellation::new(cancellation);
     let mut remaining = selection.targets.into_iter();
     while let Some(machine) = remaining.next() {
diff --git a/crates/ployz/tests/build_layer3/remote.rs b/crates/ployz/tests/build_layer3/remote.rs
index 25a05a89d5..4c9ba995c4 100644
--- a/crates/ployz/tests/build_layer3/remote.rs
+++ b/crates/ployz/tests/build_layer3/remote.rs
@@ -506,6 +506,7 @@ async fn remote_build_delivers_dependency_content_and_deploys_without_a_registry
     let result = ployz::image::push_from_machine(
         &mut client,
         &app.built,
+        &app.image,
         destination,
         &[selected.to_string()],
         &CancellationToken::new(),
diff --git a/crates/ployzd/src/daemon.rs b/crates/ployzd/src/daemon.rs
index bc905194f5..4c985614e5 100644
--- a/crates/ployzd/src/daemon.rs
+++ b/crates/ployzd/src/daemon.rs
@@ -792,6 +792,7 @@ mod tests {
         use std::os::unix::fs::PermissionsExt as _;
         let root = TestDir::new("ployzd-builder-restart");
         let (config, socket) = test_config(&root.0, ContainerMode::Absent);
+        fs::set_permissions(&root.0, fs::Permissions::from_mode(0o700)).unwrap();
         let policy = ployz_build::HostPolicy {
             state_directory: root.0.clone(),
             docker: root.0.join("docker"),
diff --git a/crates/ployzd/src/docker/peer_pull.rs b/crates/ployzd/src/docker/peer_pull.rs
index 12a703d4ff..d95ecfdb84 100644
--- a/crates/ployzd/src/docker/peer_pull.rs
+++ b/crates/ployzd/src/docker/peer_pull.rs
@@ -38,28 +38,50 @@ pub(crate) async fn pull_from_ingest(
     };
     let proxy = ImageProxy::open(source).await?;
     let pulled = localhost_registry_reference(proxy.port, image);
-    docker_cli(["pull", &pulled]).await?;
-    // Docker cannot tag repository@digest. A content-specific tag retains
-    // that repository's digest without another client's tag overwriting it.
-    docker_cli(["tag", &pulled, retained.as_deref().unwrap_or(image)]).await?;
-    if let Some((_, digest)) = image.split_once('@') {
-        let descriptor = docker_cli([
-            "image",
-            "inspect",
-            image,
-            "--format",
-            "{{json .Descriptor}}",
-        ])
-        .await?;
-        let descriptor: serde_json::Value = serde_json::from_str(&descriptor)?;
-        if descriptor.get("digest").and_then(serde_json::Value::as_str) != Some(digest) {
-            return Err(Error::PeerPull(format!(
-                "the image store did not retain {image}"
-            )));
+    pull_and_tag(
+        image,
+        &pulled,
+        retained.as_deref(),
+        std::path::Path::new("docker"),
+    )
+    .await
+}
+
+async fn pull_and_tag(
+    image: &str,
+    pulled: &str,
+    retained: Option<&str>,
+    docker: &std::path::Path,
+) -> Result<(), Error> {
+    let result = async {
+        docker_cli(docker, ["pull", pulled]).await?;
+        if let Some((_, digest)) = image.split_once('@') {
+            let descriptor = docker_cli(
+                docker,
+                [
+                    "image",
+                    "inspect",
+                    pulled,
+                    "--format",
+                    "{{json .Descriptor}}",
+                ],
+            )
+            .await?;
+            let descriptor: serde_json::Value = serde_json::from_str(&descriptor)?;
+            if descriptor.get("digest").and_then(serde_json::Value::as_str) != Some(digest) {
+                return Err(Error::PeerPull(format!(
+                    "the image store did not retain {image}"
+                )));
+            }
         }
+        // Verify before publishing the deterministic retention tag: a failed
+        // attempt must not leave a tag or remove one from an earlier delivery.
+        docker_cli(docker, ["tag", pulled, retained.unwrap_or(image)]).await?;
+        Ok(())
     }
-    let _ = docker_cli(["image", "rm", &pulled]).await;
-    Ok(())
+    .await;
+    let _ = docker_cli(docker, ["image", "rm", pulled]).await;
+    result
 }
 
 fn localhost_registry_reference(port: u16, image: &str) -> String {
@@ -108,8 +130,11 @@ impl Drop for ImageProxy {
     }
 }
 
-async fn docker_cli<const N: usize>(args: [&str; N]) -> Result<String, Error> {
-    let output = Command::new("docker")
+async fn docker_cli<const N: usize>(
+    docker: &std::path::Path,
+    args: [&str; N],
+) -> Result<String, Error> {
+    let output = Command::new(docker)
         .args(args)
         .output()
         .await
@@ -128,6 +153,79 @@ mod tests {
     use super::*;
     use ployz_core::UNREGISTRY_PORT;
 
+    #[tokio::test]
+    async fn failed_digest_delivery_cleans_up_without_publishing_unverified_content() {
+        use std::{fs, os::unix::fs::PermissionsExt as _};
+
+        let root = std::env::temp_dir().join(format!(
+            "ployz-peer-pull-{}",
+            ployz_core::MachineId::random()
+        ));
+        fs::create_dir(&root).unwrap();
+        let docker = root.join("docker");
+        fs::write(
+            &docker,
+            r#"#!/bin/sh
+cd "$(dirname "$0")"
+printf '%s\n' "$*" >> calls
+mode=$(cat mode)
+case "$1 $2" in
+  'image inspect')
+    case "$mode" in
+      inspect) echo inspect-failed >&2; exit 1 ;;
+      json) echo not-json ;;
+      digest) echo '{"digest":"sha256:wrong"}' ;;
+      *) cat descriptor ;;
+    esac ;;
+  'image rm') echo cleanup-failed >&2; exit 1 ;;
+  *) if [ "$1" = "$mode" ]; then echo "$mode-failed" >&2; exit 1; fi ;;
+esac
+"#,
+        )
+        .unwrap();
+        fs::set_permissions(&docker, fs::Permissions::from_mode(0o700)).unwrap();
+        let digest = format!("sha256:{}", "1".repeat(64));
+        let image = format!("example.test/api@{digest}");
+        let retained = image.replace("@sha256:", ":ployz-sha256-");
+        let pulled = format!("127.0.0.1:1234/{image}");
+        fs::write(
+            root.join("descriptor"),
+            serde_json::json!({"digest": digest}).to_string(),
+        )
+        .unwrap();
+        for mode in ["inspect", "json", "digest", "tag", "pull", "success"] {
+            fs::write(root.join("mode"), mode).unwrap();
+            fs::write(root.join("calls"), "").unwrap();
+            let result = pull_and_tag(&image, &pulled, Some(&retained), &docker).await;
+            assert_eq!(result.is_ok(), mode == "success", "{mode}: {result:?}");
+            if let Err(error) = result {
+                assert!(!error.to_string().contains("cleanup-failed"), "{error}");
+                if ["inspect", "tag", "pull"].contains(&mode) {
+                    assert!(
+                        error.to_string().contains(&format!("{mode}-failed")),
+                        "{error}"
+                    );
+                }
+            }
+            let calls = fs::read_to_string(root.join("calls")).unwrap();
+            assert_eq!(
+                calls.lines().last(),
+                Some(format!("image rm {pulled}").as_str()),
+                "{mode}: {calls}"
+            );
+            assert_eq!(
+                calls.lines().any(|line| line.starts_with("tag ")),
+                ["tag", "success"].contains(&mode),
+                "{mode}: {calls}"
+            );
+            assert!(
+                !calls.contains(&format!("image rm {retained}")),
+                "must preserve previously retained content"
+            );
+        }
+        fs::remove_dir_all(root).unwrap();
+    }
+
     #[test]
     fn localhost_pull_reference_keeps_the_image_path() {
         assert_eq!(