diff --git a/candy/plugin-deploy-pod/config_remove.go b/candy/plugin-deploy-pod/config_remove.go
index dbfb2a9..53d999e 100644
--- a/candy/plugin-deploy-pod/config_remove.go
+++ b/candy/plugin-deploy-pod/config_remove.go
@@ -8,6 +8,7 @@ import (
 	"os/exec"
 	"strings"
 
+	"github.com/opencharly/sdk"
 	"github.com/opencharly/sdk/kit"
 	pb "github.com/opencharly/spec/proto"
 	"github.com/opencharly/spec/spec"
@@ -17,31 +18,43 @@ import (
 // VERBATIM (no seams needed — purely resolveBoxName + IsDirectDeploy + podman/systemctl exec +
 // findPodSidecarQuadlets, all portable). Distinct from `charly remove`/OpPostTeardown, which tears
 // down the whole deploy; this disables the quadlet service (the quadlet file remains).
-func invokeConfigRemove(_ context.Context, req *pb.InvokeRequest) (*pb.InvokeReply, error) {
+func invokeConfigRemove(ctx context.Context, req *pb.InvokeRequest) (*pb.InvokeReply, error) {
+	// The executor is OPTIONAL here: remove is driven from the runtime configuration and
+	// needs no venue work, so an Invoke with no executor attached must keep working — it
+	// simply resolves the selection from the runtime level (loadDeploy degrades to nil).
+	ex, _ := sdk.ExecutorFromInvoke(req.GetExecutorBrokerId())
 	var r spec.PodConfigRemoveRequest
 	if err := json.Unmarshal(req.GetParamsJson(), &r); err != nil {
 		return nil, fmt.Errorf("plugin-deploy-pod config-remove: decode request: %w", err)
 	}
-	if err := runPodConfigRemove(&r); err != nil {
+	if err := runPodConfigRemove(ctx, ex, &r); err != nil {
 		return nil, err
 	}
 	return marshalReply(spec.PodConfigRemoveReply{})
 }
 
-func runPodConfigRemove(c *spec.PodConfigRemoveRequest) error {
+func runPodConfigRemove(ctx context.Context, ex *sdk.Executor, c *spec.PodConfigRemoveRequest) error {
 	rt, err := kit.ResolveRuntime()
 	if err != nil {
 		return err
 	}
 	boxName := resolveBoxName(c.Box)
+	dc, err := loadDeploy(ctx, ex, "charly config remove")
+	if err != nil {
+		return err
+	}
+	sel, err := resolvePodSelectionFromConfig(dc, boxName, c.Instance, rt)
+	if err != nil {
+		return err
+	}
 
 	if rt.RunMode == "direct" || IsDirectDeploy(boxName, c.Instance) {
 		name := kit.ContainerNameInstance(boxName, c.Instance)
-		stop := engineStopArgv(rt.RunEngine, name)
+		stop := engineStopArgv(sel.Target, name)
 		_ = exec.Command(stop[0], stop[1:]...).Run()
-		rm := engineRemoveArgv(rt.RunEngine, name)
+		rm := engineRemoveArgv(sel.Target, name)
 		if out, err := exec.Command(rm[0], rm[1:]...).CombinedOutput(); err != nil {
-			fmt.Fprintf(os.Stderr, "Warning: %s rm -f %s: %v\n%s", rt.RunEngine, name, err, strings.TrimSpace(string(out)))
+			fmt.Fprintf(os.Stderr, "Warning: %s rm -f %s: %v\n%s", sel.Engine, name, err, strings.TrimSpace(string(out)))
 		}
 		if err := removeDirectDeployMarker(boxName, c.Instance); err != nil {
 			fmt.Fprintf(os.Stderr, "Warning: removing direct-mode marker: %v\n", err)
diff --git a/candy/plugin-deploy-pod/config_remove_test.go b/candy/plugin-deploy-pod/config_remove_test.go
index 197c174..2b200b3 100644
--- a/candy/plugin-deploy-pod/config_remove_test.go
+++ b/candy/plugin-deploy-pod/config_remove_test.go
@@ -28,7 +28,9 @@ func TestRunPodConfigRemove_DirectModeAllowed(t *testing.T) {
 	_ = os.Setenv("CHARLY_RUN_MODE", "direct")
 	defer os.Unsetenv("CHARLY_RUN_MODE") //nolint:errcheck
 
-	err := runPodConfigRemove(&spec.PodConfigRemoveRequest{Box: "fedora-test"})
+	// A nil executor is the proof that no overlay read is required here: loadDeploy degrades
+	// to (nil, nil) with a warning, and the selection falls back to the runtime level.
+	err := runPodConfigRemove(t.Context(), nil, &spec.PodConfigRemoveRequest{Box: "fedora-test"})
 	// Direct-mode remove of a non-existent deploy is best-effort — podman rm prints a
 	// warning but returns nil. The pre-cutover "run_mode=quadlet required" hard error must NOT fire.
 	if err != nil && strings.Contains(err.Error(), "run_mode=quadlet") {
diff --git a/candy/plugin-deploy-pod/config_setup.go b/candy/plugin-deploy-pod/config_setup.go
index 5ab19d0..4993a68 100644
--- a/candy/plugin-deploy-pod/config_setup.go
+++ b/candy/plugin-deploy-pod/config_setup.go
@@ -91,12 +91,6 @@ func runPodConfigSetup(ctx context.Context, ex *sdk.Executor, c *spec.PodConfigS
 	default:
 		return fmt.Errorf("charly config requires run_mode=quadlet, systemd-unit, or direct (current: %s)", rt.RunMode)
 	}
-	// A rootful target is explicit, and this plugin cannot finish one safely: the image store
-	// staging runs rootless and system-scope unit emission is unimplemented. Refuse with the
-	// remedy rather than silently deploying the rootless arm the operator did not ask for.
-	if rootfulUnsupported(rt) {
-		return rootfulRefusal(rt.RunEngine, rt.RunMode)
-	}
 	if c.Box == "" {
 		return fmt.Errorf("image name is required")
 	}
@@ -107,7 +101,22 @@ func runPodConfigSetup(ctx context.Context, ex *sdk.Executor, c *spec.PodConfigS
 	if err := deploykit.RejectImageRefAsDeployName(c.Box); err != nil {
 		return err
 	}
-	return runConfig(ctx, ex, rt, c)
+	// Resolve the selection ONCE for the whole config run. A rootful target is explicit, and
+	// this plugin cannot finish one safely: the image store staging runs rootless and
+	// system-scope unit emission is unimplemented. Refuse with the remedy rather than silently
+	// deploying the rootless arm the operator did not ask for.
+	dc, err := loadDeploy(ctx, ex, "charly config")
+	if err != nil {
+		return err
+	}
+	sel, err := resolvePodSelectionFromConfig(dc, c.Box, c.Instance, rt)
+	if err != nil {
+		return err
+	}
+	if rootfulUnsupported(sel.Target, rt.RunMode) {
+		return rootfulRefusal(sel.Engine, rt.RunMode)
+	}
+	return runConfig(ctx, ex, rt, sel, c)
 }
 
 // resolveDeployRef mirrors the former BoxConfigSetupCmd.resolveDeployRef, resolved plugin-side
@@ -177,7 +186,7 @@ var mutateDeploy = func(ctx context.Context, ex *sdk.Executor, caller string, mu
 }
 
 //nolint:gocyclo // ported 1:1 from charly-core BoxConfigSetupCmd.runConfig — see the file header; splitting further would fragment the seam-call sequencing across files for no clarity gain
-func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c *spec.PodConfigSetupRequest) error {
+func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, sel podSelection, c *spec.PodConfigSetupRequest) error {
 	// Device detection runs plugin-side (detect_devices.go — the relocated
 	// "pod-config-detect-devices" seam, peer InvokeProvider verb:gpu). The former seam's call
 	// passed NO engine (only NoAutoDetect), so the ensure-cdi leg never fired at config time —
@@ -188,13 +197,13 @@ func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c
 	if err != nil {
 		return err
 	}
-	if err := ensureImagePresent(ctx, ex, imageRef, rt.RunEngine); err != nil {
+	if err := ensureImagePresent(ctx, ex, imageRef, sel.Target); err != nil {
 		return err
 	}
 	// ExtractMetadata PLUGIN-SIDE (#55 coneC-dsh — the pod-config-ensure-image host seam shrinks to
 	// the host-coupled ensureImagePresent leg; the deploykit label extract runs here, shedding the
 	// host leg's deploykit import).
-	metaPtr, err := deploykit.ExtractMetadata(rt.RunEngine, imageRef)
+	metaPtr, err := deploykit.ExtractMetadata(string(sel.Engine), imageRef)
 	if err != nil {
 		return err
 	}
@@ -255,7 +264,7 @@ func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c
 	// emission path below (the direct return and the quadlet write) and before any start, so the
 	// ownership is right however the container is created; `charly remove` keeps named volumes, so
 	// the re-run of an existing deploy is the normal case this must repair.
-	if err := reconcileNamedVolumeRoots(ctx, ex, rt.RunEngine, volumeRootKeepID(rt.RunMode, rt.RunEngine, len(bindMounts)), uid, gid, imageRef, volumes); err != nil {
+	if err := reconcileNamedVolumeRoots(ctx, ex, string(sel.Engine), volumeRootKeepID(rt.RunMode, string(sel.Engine), len(bindMounts)), uid, gid, imageRef, volumes); err != nil {
 		return err
 	}
 
@@ -337,17 +346,17 @@ func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c
 	}
 	envVars = appendAutoDetectedEnv(envVars, detected)
 
-	resolvedNetwork, netErr := kit.ResolveNetwork(network, rt.RunEngine)
+	resolvedNetwork, netErr := kit.ResolveNetwork(network, string(sel.Engine))
 	if netErr != nil {
 		return netErr
 	}
 
-	if conflicts := kit.CheckPortAvailability(ports, rt.BindAddress, rt.RunEngine); len(conflicts) > 0 {
+	if conflicts := kit.CheckPortAvailability(ports, rt.BindAddress, string(sel.Engine)); len(conflicts) > 0 {
 		fmt.Fprintf(os.Stderr, "Warning: port conflicts detected:%s", kit.FormatPortConflicts(conflicts, c.Box))
 	}
 
 	autoGen := c.Password == "auto"
-	provisioned, provFallback, provResolutions, isKeyring, perr := resolvePodProvisionSecrets(ctx, ex, &meta, c.Box, c.Instance, rt.RunEngine, autoGen, c.RefreshSecret)
+	provisioned, provFallback, provResolutions, isKeyring, perr := resolvePodProvisionSecrets(ctx, ex, &meta, c.Box, c.Instance, string(sel.Engine), autoGen, c.RefreshSecret)
 	if perr != nil {
 		return perr
 	}
@@ -379,7 +388,7 @@ func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c
 	var resolvedSidecars []deploykit.ResolvedSidecar
 	var deploySidecars map[string]json.RawMessage
 	if len(deploySidecarsRaw) > 0 {
-		scRes, err := resolvePodSidecars(ctx, ex, deploySidecarsRaw, deploykit.SidecarTemplatesOf(dc), c.Env, c.Box, c.Instance, rt.RunEngine, autoGen, c.RefreshSecret)
+		scRes, err := resolvePodSidecars(ctx, ex, deploySidecarsRaw, deploykit.SidecarTemplatesOf(dc), c.Env, c.Box, c.Instance, string(sel.Engine), autoGen, c.RefreshSecret)
 		if err != nil {
 			return scErr(err)
 		}
@@ -424,11 +433,11 @@ func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c
 	deploySaveState(ctx, ex, c.Box, c.Instance, saveInput)
 
 	if rt.RunMode == "direct" {
-		return runConfigDirect(rt, qcfg, bindMounts, resolvedSidecars, tunnelCfg)
+		return runConfigDirect(rt, sel, qcfg, bindMounts, resolvedSidecars, tunnelCfg)
 	}
 
 	if rt.RunMode == "systemd-unit" {
-		written, werr := writeSystemdUnitConfig(rt, qcfg, bindMounts, resolvedSidecars, tunnelCfg)
+		written, werr := writeSystemdUnitConfig(rt, sel, qcfg, bindMounts, resolvedSidecars, tunnelCfg)
 		if werr != nil {
 			return werr
 		}
@@ -550,7 +559,7 @@ func runConfig(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c
 		if err := start.Run(); err != nil {
 			fmt.Fprintf(os.Stderr, "Warning: failed to start %s for post_enable hook: %v\n", svc, err)
 		} else {
-			engine := kit.EngineBinary(rt.RunEngine)
+			engine := sel.binary()
 			hookSecretEnv := resolvePodHookSecretEnv(ctx, ex, &meta, c.Box, c.Instance)
 			hookEnv := append(append([]string{}, c.Env...), hookSecretEnv...)
 			if err := runHook(engine, ctrName, hooks.PostEnable, hookEnv); err != nil {
diff --git a/candy/plugin-deploy-pod/config_setup_helpers.go b/candy/plugin-deploy-pod/config_setup_helpers.go
index b0a82bf..d893588 100644
--- a/candy/plugin-deploy-pod/config_setup_helpers.go
+++ b/candy/plugin-deploy-pod/config_setup_helpers.go
@@ -487,17 +487,17 @@ func writeDirectDeployMarker(m directDeployMarker) error {
 // meant a direct-mode deploy ignored the resolved engine (containerd/docker still ran podman).
 // The engine is now a parameter, the argv comes from the engine class, and the caller executes
 // that argv.
-func directRunArgs(qcfg deploykit.QuadletConfig, bindMounts []deploykit.ResolvedBindMount, engine string, rootful bool) ([]string, error) {
+func directRunArgs(qcfg deploykit.QuadletConfig, bindMounts []deploykit.ResolvedBindMount, sel podSelection) ([]string, error) {
 	name := kit.ContainerNameInstance(qcfg.BoxName, qcfg.Instance)
 	network := qcfg.Network
 	if network == "" {
 		network = "charly"
 	}
-	argv, _, err := engineRunArgv(engine, rootful, podEngineRun{
+	argv, _, err := engineRunArgv(sel.Target, podEngineRun{
 		Image: qcfg.ImageRef, Name: name, Network: network,
 		Hostname: name, Restart: "always",
 		Detach: true,
-		KeepID: engineKeepID(engine, len(bindMounts)),
+		KeepID: engineKeepID(string(sel.Engine), len(bindMounts)),
 		UID:    qcfg.UID, GID: qcfg.GID,
 		Ports:      localizedPorts(qcfg.Ports, qcfg.BindAddress),
 		Volumes:    volumeArgs(qcfg.Volumes, bindMounts),
@@ -510,7 +510,7 @@ func directRunArgs(qcfg deploykit.QuadletConfig, bindMounts []deploykit.Resolved
 }
 
 // runConfigDirect mirrors the former BoxConfigSetupCmd.runConfigDirect, now engine-driven.
-func runConfigDirect(rt *kit.ResolvedRuntime, qcfg deploykit.QuadletConfig, bindMounts []deploykit.ResolvedBindMount, sidecars []deploykit.ResolvedSidecar, tunnelCfg *spec.TunnelConfig) error {
+func runConfigDirect(rt *kit.ResolvedRuntime, sel podSelection, qcfg deploykit.QuadletConfig, bindMounts []deploykit.ResolvedBindMount, sidecars []deploykit.ResolvedSidecar, tunnelCfg *spec.TunnelConfig) error {
 	if len(sidecars) > 0 {
 		fmt.Fprintf(os.Stderr, "Warning: sidecars are not supported in direct mode (skipping); use run_mode=quadlet for sidecar deploys.\n")
 	}
@@ -520,16 +520,16 @@ func runConfigDirect(rt *kit.ResolvedRuntime, qcfg deploykit.QuadletConfig, bind
 	if deploykit.HasEncryptedBindMounts(bindMounts) {
 		fmt.Fprintf(os.Stderr, "Warning: encrypted bind mounts require systemd-run; encrypted volumes will not be initialized in direct mode.\n")
 	}
-	if rootfulUnsupported(rt) {
-		return rootfulRefusal(rt.RunEngine, rt.RunMode)
+	if rootfulUnsupported(sel.Target, rt.RunMode) {
+		return rootfulRefusal(sel.Engine, rt.RunMode)
 	}
-	engine := rt.RunEngine
+	engine := string(sel.Engine)
 	name := kit.ContainerNameInstance(qcfg.BoxName, qcfg.Instance)
-	stop := engineStopArgv(engine, name)
+	stop := engineStopArgv(sel.Target, name)
 	_ = exec.Command(stop[0], stop[1:]...).Run()
-	rm := engineRemoveArgv(engine, name)
+	rm := engineRemoveArgv(sel.Target, name)
 	_ = exec.Command(rm[0], rm[1:]...).Run()
-	args, err := directRunArgs(qcfg, bindMounts, engine, podRootful(rt))
+	args, err := directRunArgs(qcfg, bindMounts, sel)
 	if err != nil {
 		return err
 	}
@@ -583,12 +583,11 @@ func podConfigWriteSelf(r spec.PodConfigWriteRequest) (spec.PodConfigWriteReply,
 func provisionData(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntime, c *spec.PodConfigSetupRequest, meta *spec.BoxMetadata, imageRef string, dc *deploykit.DeployConfig, bindMounts []deploykit.ResolvedBindMount, volumes []deploykit.VolumeMount, deployVolumes []spec.DeployVolume) error {
 	dataMeta := meta
 	dataRef := imageRef
-	dataEngine := rt.RunEngine
-	if dc != nil {
-		if entry, ok := dc.Lookup(c.Box, c.Instance); ok && entry.Engine != "" {
-			dataEngine = string(entry.Engine)
-		}
+	sel, err := resolvePodSelectionFromConfig(dc, c.Box, c.Instance, rt)
+	if err != nil {
+		return err
 	}
+	dataEngine := string(sel.Engine)
 
 	if c.DataFrom != "" {
 		dataRef = c.DataFrom
@@ -597,7 +596,7 @@ func provisionData(ctx context.Context, ex *sdk.Executor, rt *kit.ResolvedRuntim
 				dataRef = resolved
 			}
 		}
-		if err := ensureImagePresent(ctx, ex, dataRef, rt.RunEngine); err != nil {
+		if err := ensureImagePresent(ctx, ex, dataRef, sel.Target); err != nil {
 			return fmt.Errorf("extracting metadata from data image %s: %w", dataRef, err)
 		}
 		dm, err := deploykit.ExtractMetadata(dataEngine, dataRef)
@@ -701,11 +700,18 @@ func updateAllDeployedQuadlets(ctx context.Context, ex *sdk.Executor, rt *kit.Re
 				imageRef = ref
 			}
 		}
-		if err := ensureImagePresent(ctx, ex, imageRef, rt.RunEngine); err != nil {
+		// Every entry resolves its OWN selection: a per-deploy `engine:`/`namespace:` in
+		// the overlay must reach that entry's quadlet, not the runtime's global engine.
+		updateSel, selErr := resolvePodSelectionFromConfig(dc, boxName, instance, rt)
+		if selErr != nil {
+			fmt.Fprintf(os.Stderr, "Warning: could not resolve the engine for %s, skipping quadlet update: %v\n", key, selErr)
+			continue
+		}
+		if err := ensureImagePresent(ctx, ex, imageRef, updateSel.Target); err != nil {
 			fmt.Fprintf(os.Stderr, "Warning: could not read metadata for %s, skipping quadlet update\n", key)
 			continue
 		}
-		metaPtr, err := deploykit.ExtractMetadata(rt.RunEngine, imageRef)
+		metaPtr, err := deploykit.ExtractMetadata(string(updateSel.Engine), imageRef)
 		if err != nil || metaPtr == nil {
 			continue
 		}
@@ -721,7 +727,7 @@ func updateAllDeployedQuadlets(ctx context.Context, ex *sdk.Executor, rt *kit.Re
 			continue
 		}
 		envVars = kit.EnrichNoProxy(envVars, deploykit.DeployedContainerNames(dc))
-		resolvedNetwork, _ := kit.ResolveNetwork(meta.Network, rt.RunEngine)
+		resolvedNetwork, _ := kit.ResolveNetwork(meta.Network, string(updateSel.Engine))
 
 		// Device detection runs plugin-side (detect_devices.go). The former seam call passed an
 		// empty request (NoAutoDetect=false, Engine="") and SWALLOWED the error — the
@@ -759,7 +765,7 @@ func updateAllDeployedQuadlets(ctx context.Context, ex *sdk.Executor, rt *kit.Re
 		}
 		envVars = appendAutoDetectedEnv(envVars, detected)
 
-		provisioned, _, provResolutions, isKeyring, perr := resolvePodProvisionSecrets(ctx, ex, &meta, boxName, instance, rt.RunEngine, true, nil)
+		provisioned, _, provResolutions, isKeyring, perr := resolvePodProvisionSecrets(ctx, ex, &meta, boxName, instance, string(updateSel.Engine), true, nil)
 		if perr != nil {
 			fmt.Fprintf(os.Stderr, "Warning: provisioning secrets for %s: %v\n", key, perr)
 			continue
@@ -786,7 +792,7 @@ func updateAllDeployedQuadlets(ctx context.Context, ex *sdk.Executor, rt *kit.Re
 		var resolvedSidecars []deploykit.ResolvedSidecar
 		podName := ""
 		if len(deploySidecarsRaw) > 0 {
-			scRes, err := resolvePodSidecars(ctx, ex, deploySidecarsRaw, deploykit.SidecarTemplatesOf(dc), nil, boxName, instance, rt.RunEngine, true, nil)
+			scRes, err := resolvePodSidecars(ctx, ex, deploySidecarsRaw, deploykit.SidecarTemplatesOf(dc), nil, boxName, instance, string(updateSel.Engine), true, nil)
 			if err != nil {
 				fmt.Fprintf(os.Stderr, "Warning: resolving sidecars for %s: %v\n", key, err)
 				continue
@@ -1029,16 +1035,16 @@ func mergeVolumeConfigsLocal(base, overrides []spec.DeployVolume) []spec.DeployV
 // through the engine `start_plan` body (engine_ops.go): the engine word selects the binary, the
 // keep-id fact, the GPU flag style and the runtime handler, so a containerd/docker deploy no
 // longer runs a podman argv. rootful selects the engine's mode-specific workload-user fact.
-func buildStartArgs(engine string, rootful bool, imageRef string, uid, gid int, ports []string, name string, volumes []deploykit.VolumeMount, bindMounts []deploykit.ResolvedBindMount, gpu bool, bindAddr string, envVars []string, security spec.SecurityConfig, entrypoint []string, workingDir string, network ...string) ([]string, error) {
+func buildStartArgs(sel podSelection, imageRef string, uid, gid int, ports []string, name string, volumes []deploykit.VolumeMount, bindMounts []deploykit.ResolvedBindMount, gpu bool, bindAddr string, envVars []string, security spec.SecurityConfig, entrypoint []string, workingDir string, network ...string) ([]string, error) {
 	net := ""
 	if len(network) > 0 {
 		net = network[0]
 	}
-	argv, _, err := engineRunArgv(engine, rootful, podEngineRun{
+	argv, _, err := engineRunArgv(sel.Target, podEngineRun{
 		Image: imageRef, Name: name, Network: net,
 		WorkingDir: workingDir, Hostname: name,
 		Detach: true, Remove: true,
-		KeepID: engineKeepID(engine, len(bindMounts)),
+		KeepID: engineKeepID(string(sel.Engine), len(bindMounts)),
 		UID:    uid, GID: gid,
 		Ports:      localizedPorts(ports, bindAddr),
 		Volumes:    volumeArgs(volumes, bindMounts),
diff --git a/candy/plugin-deploy-pod/enc_tunnel_resolve.go b/candy/plugin-deploy-pod/enc_tunnel_resolve.go
index 49beb1e..40fa853 100644
--- a/candy/plugin-deploy-pod/enc_tunnel_resolve.go
+++ b/candy/plugin-deploy-pod/enc_tunnel_resolve.go
@@ -93,7 +93,11 @@ func resolvePodTunnelPlan(dc *deploykit.DeployConfig, box, instance string) *spe
 	if err != nil {
 		return nil
 	}
-	engine := rt.RunEngine
+	sel, err := resolvePodSelectionFromConfig(dc, box, instance, rt)
+	if err != nil {
+		return nil
+	}
+	engine := string(sel.Engine)
 	ctrName := kit.ContainerNameInstance(box, instance)
 	imageRef := kit.ContainerImage(engine, ctrName)
 	if imageRef == "" {
diff --git a/candy/plugin-deploy-pod/engine_ops.go b/candy/plugin-deploy-pod/engine_ops.go
index 2bd81c5..40a5001 100644
--- a/candy/plugin-deploy-pod/engine_ops.go
+++ b/candy/plugin-deploy-pod/engine_ops.go
@@ -45,18 +45,18 @@ type podEngineRun struct {
 }
 
 // engineRunArgv builds a launch argv through the engine class's `start_plan` body and returns
-// the engine's own run mode. It is always the ROOTLESS argv: a rootful target that would need
-// a privilege prefix is refused by rootfulUnsupported before any caller reaches here. Every
-// launch of an engine that declares runtime_handlers carries `--runtime <handler>` selected
-// from that set, so a containerd launch never relies on the daemon's default handler.
-func engineRunArgv(engine string, rootful bool, r podEngineRun) ([]string, string, error) {
+// the engine's own run mode. The engine word and every target fact come from the RESOLVED
+// selection (selection.go), so the argv, the store and the unit all address the same target.
+// A rootful target that would need a privilege prefix is refused by rootfulUnsupported before
+// any caller reaches here. Every launch of an engine that declares runtime_handlers carries
+// `--runtime <handler>` selected from that set, so a containerd launch never relies on the
+// daemon's default handler.
+func engineRunArgv(target deploykit.ContainerdTarget, r podEngineRun) ([]string, string, error) {
+	engine := string(target.Engine)
 	c, ok := container.EngineCapabilityFor(engine)
 	if !ok {
 		return nil, "", fmt.Errorf("engine.run: no capability row for %q", engine)
 	}
-	if rootful && !c.SupportsRootful {
-		return nil, "", fmt.Errorf("engine %q does not support rootful launches", engine)
-	}
 	extra := make([]string, 0, 16)
 	if r.Remove && !c.NoRemoveWithDetach {
 		extra = append(extra, "--rm")
@@ -94,7 +94,7 @@ func engineRunArgv(engine string, rootful bool, r podEngineRun) ([]string, strin
 		Image:     r.Image,
 		Engine:    engine,
 		Network:   r.Network,
-		Rootful:   rootful,
+		Rootful:   target.Rootful,
 		Runtime:   engineDefaultRuntime(c),
 		Detach:    r.Detach,
 		Ports:     r.Ports,
@@ -118,7 +118,7 @@ func engineRunArgv(engine string, rootful bool, r podEngineRun) ([]string, strin
 	if rep.Error != "" {
 		return nil, "", fmt.Errorf("engine start_plan (%s): %s", engine, rep.Error)
 	}
-	argv := prependEngineGlobalArgs(c, rep.Argv)
+	argv := prependEngineGlobalArgs(target, rep.Argv)
 	// No privilege_prefix is applied here: rootfulUnsupported refuses every rootful engine
 	// that declares one (containerd), and an engine without one has an empty prefix, so this
 	// argv is always the rootless form. Adding a prefix on an unreachable arm would be a
@@ -136,24 +136,25 @@ func engineDefaultRuntime(c spec.EngineCapability) string {
 	return c.RuntimeHandlers[0]
 }
 
-// engineGlobalArgs returns the engine CLI's connection flags, from the capability row. Only
-// an engine with a separate daemon endpoint declares them (containerd: address + namespace),
-// so podman and docker are unchanged.
-func engineGlobalArgs(c spec.EngineCapability) []string {
+// engineGlobalArgs returns the engine CLI's connection flags from the RESOLVED target. Only
+// an engine with a separate daemon endpoint declares an address (containerd), so podman and
+// docker are unchanged; the namespace is the resolved one (deploy > settings > capability
+// default), never the capability default alone.
+func engineGlobalArgs(target deploykit.ContainerdTarget) []string {
 	out := make([]string, 0, 2)
-	if c.Address != "" {
-		out = append(out, "--address", c.Address)
+	if target.Address != "" {
+		out = append(out, "--address", target.Address)
 	}
-	if c.Namespace != "" {
-		out = append(out, "--namespace", c.Namespace)
+	if target.Namespace != "" {
+		out = append(out, "--namespace", target.Namespace)
 	}
 	return out
 }
 
 // prependEngineGlobalArgs inserts the connection flags BEFORE the subcommand, which is where
 // the engine CLI accepts persistent flags (nerdctl: `nerdctl --address … --namespace … run …`).
-func prependEngineGlobalArgs(c spec.EngineCapability, argv []string) []string {
-	globals := engineGlobalArgs(c)
+func prependEngineGlobalArgs(target deploykit.ContainerdTarget, argv []string) []string {
+	globals := engineGlobalArgs(target)
 	if len(globals) == 0 || len(argv) == 0 {
 		return argv
 	}
@@ -165,15 +166,13 @@ func prependEngineGlobalArgs(c spec.EngineCapability, argv []string) []string {
 }
 
 // engineStopArgv builds `<binary> <globals> stop <name>`.
-func engineStopArgv(engine, name string) []string {
-	c, _ := container.EngineCapabilityFor(engine)
-	return prependEngineGlobalArgs(c, []string{kit.EngineBinary(engine), "stop", name})
+func engineStopArgv(target deploykit.ContainerdTarget, name string) []string {
+	return prependEngineGlobalArgs(target, []string{kit.EngineBinary(string(target.Engine)), "stop", name})
 }
 
 // engineRemoveArgv builds `<binary> <globals> rm -f <name>`.
-func engineRemoveArgv(engine, name string) []string {
-	c, _ := container.EngineCapabilityFor(engine)
-	return prependEngineGlobalArgs(c, []string{kit.EngineBinary(engine), "rm", "-f", name})
+func engineRemoveArgv(target deploykit.ContainerdTarget, name string) []string {
+	return prependEngineGlobalArgs(target, []string{kit.EngineBinary(string(target.Engine)), "rm", "-f", name})
 }
 
 // engineUnitFiles renders the supervision unit through the engine class's `unit_emit` body.
@@ -210,33 +209,23 @@ func engineUsesCDI(engine string) bool {
 	return ok && c.GPUArgStyle == "cdi"
 }
 
-// podRootful reports whether the host config asks for the rootful mode. `auto` is the
-// pre-resolution selector; every concrete mode ("sudo", "native", "machine") is an explicit
-// rootful request.
-func podRootful(rt *kit.ResolvedRuntime) bool {
-	return rt.Rootful != "" && rt.Rootful != "auto"
-}
-
 // rootfulRefusal is the honest answer where this plugin cannot finish a rootful deploy safely:
 // system-scope units and a rootful image store are not implemented here, so silently running
 // the rootless arm would deploy a different target than the operator asked for.
-func rootfulRefusal(engine, mode string) error {
+func rootfulRefusal(engine spec.EngineName, mode string) error {
 	return fmt.Errorf("engine %q: rootful %s deploys are not implemented by plugin-deploy-pod yet (system-scope unit emission and a rootful image store are missing); remedy: run rootless (engine.rootful=auto), or use engine: podman (quadlet) for a rootful pod", engine, mode)
 }
 
-// rootfulUnsupported reports the rootful targets this plugin refuses. It is capability-driven:
-// a systemd-unit mode needs system-scope emission, and an engine with a privilege_prefix
-// expects a root daemon the rootless staging path cannot reach. An engine with neither (podman
-// quadlet) keeps its existing behaviour untouched.
-func rootfulUnsupported(rt *kit.ResolvedRuntime) bool {
-	if !podRootful(rt) {
+// rootfulUnsupported reports the rootful targets this plugin refuses. It reads the RESOLVED
+// target: `Rootful` already folds the flag/deploy/env/settings levels, and the privilege prefix
+// is the capability fact. A systemd-unit mode needs system-scope emission, and an engine with a
+// privilege prefix expects a root daemon the rootless staging path cannot reach. An engine with
+// neither (podman quadlet) keeps its existing behaviour untouched.
+func rootfulUnsupported(target deploykit.ContainerdTarget, runMode string) bool {
+	if !target.Rootful {
 		return false
 	}
-	if rt.RunMode == "systemd-unit" {
-		return true
-	}
-	c, ok := container.EngineCapabilityFor(rt.RunEngine)
-	return ok && c.PrivilegePrefix != ""
+	return runMode == "systemd-unit" || target.PrivilegePrefix != ""
 }
 
 // envArgs renders the pod's env entries as engine CLI `-e` arguments, VERBATIM. An engine
diff --git a/candy/plugin-deploy-pod/engine_ops_test.go b/candy/plugin-deploy-pod/engine_ops_test.go
index 665bd50..b7e3e4d 100644
--- a/candy/plugin-deploy-pod/engine_ops_test.go
+++ b/candy/plugin-deploy-pod/engine_ops_test.go
@@ -37,7 +37,7 @@ func hasFlag(argv []string, flag string) bool {
 }
 
 func TestEngineRunArgv_ContainerdUnitPlan(t *testing.T) {
-	argv, mode, err := engineRunArgv("containerd", false, podEngineRun{
+	argv, mode, err := engineRunArgv(selectionForTest(t, deploykit.Selection{}, "containerd").Target, podEngineRun{
 		Image:      "ghcr.io/opencharly/demo:latest",
 		Name:       "charly-demo",
 		Network:    "charly",
@@ -80,7 +80,7 @@ func TestEngineRunArgv_ContainerdUnitPlan(t *testing.T) {
 }
 
 func TestEngineRunArgv_PodmanPlanUnchanged(t *testing.T) {
-	argv, mode, err := engineRunArgv("podman", false, podEngineRun{
+	argv, mode, err := engineRunArgv(selectionForTest(t, deploykit.Selection{}, "podman").Target, podEngineRun{
 		Image:      "ghcr.io/opencharly/demo:latest",
 		Name:       "charly-demo",
 		Network:    "charly",
@@ -132,7 +132,8 @@ func TestEngineRunArgv_PodmanPlanUnchanged(t *testing.T) {
 }
 
 func TestEngineUnitFiles_ContainerdUnitWrapsStartPlan(t *testing.T) {
-	argv, _, err := engineRunArgv("containerd", false, podEngineRun{
+	sel := selectionForTest(t, deploykit.Selection{}, "containerd")
+	argv, _, err := engineRunArgv(sel.Target, podEngineRun{
 		Image: "ghcr.io/opencharly/demo:latest", Name: "charly-demo",
 	})
 	if err != nil {
@@ -140,7 +141,7 @@ func TestEngineUnitFiles_ContainerdUnitWrapsStartPlan(t *testing.T) {
 	}
 	files, err := engineUnitFiles("containerd", spec.EngineUnitRequest{
 		Name: "demo", RunMode: "systemd-unit", StartArgv: argv,
-		StopArgv: engineStopArgv("containerd", "charly-demo"),
+		StopArgv: engineStopArgv(sel.Target, "charly-demo"),
 	})
 	if err != nil {
 		t.Fatalf("engineUnitFiles(containerd): %v", err)
@@ -158,8 +159,8 @@ func TestEngineUnitFiles_ContainerdUnitWrapsStartPlan(t *testing.T) {
 }
 
 func TestRootfulContainerdRefused(t *testing.T) {
-	rt := kitResolvedRuntimeForTest("containerd", "systemd-unit", "sudo")
-	if !rootfulUnsupported(rt) {
+	containerd := selectionForTest(t, deploykit.Selection{Rootful: "sudo"}, "containerd")
+	if !rootfulUnsupported(containerd.Target, "systemd-unit") {
 		t.Fatal("a rootful containerd systemd-unit target must be refused, not silently run rootless")
 	}
 	if err := rootfulRefusal("containerd", "systemd-unit"); err == nil ||
@@ -168,8 +169,8 @@ func TestRootfulContainerdRefused(t *testing.T) {
 	}
 	// The podman quadlet control: its rootful mode is not a refusal (the capability declares no
 	// privilege prefix and no system-scope unit need here), so the podman plan stays unchanged.
-	podman := kitResolvedRuntimeForTest("podman", "quadlet", "sudo")
-	if rootfulUnsupported(podman) {
+	podman := selectionForTest(t, deploykit.Selection{Rootful: "sudo"}, "podman")
+	if rootfulUnsupported(podman.Target, "quadlet") {
 		t.Error("podman quadlet rootful must keep its existing behaviour")
 	}
 }
@@ -189,7 +190,7 @@ func TestWriteSystemdUnitConfig_Containerd(t *testing.T) {
 	rt := kitResolvedRuntimeForTest("containerd", "systemd-unit", "auto")
 	qcfg := deploykit.QuadletConfig{BoxName: "demo", ImageRef: "img", Home: "/home/user"}
 
-	written, err := writeSystemdUnitConfig(rt, qcfg, nil, nil, nil)
+	written, err := writeSystemdUnitConfig(rt, selectionForTest(t, deploykit.Selection{}, "containerd"), qcfg, nil, nil, nil)
 	if err != nil {
 		t.Fatalf("writeSystemdUnitConfig: %v", err)
 	}
@@ -213,7 +214,7 @@ func TestWriteSystemdUnitConfig_Containerd(t *testing.T) {
 
 	// A rootful systemd-unit target is refused before any file is written.
 	rootful := kitResolvedRuntimeForTest("containerd", "systemd-unit", "sudo")
-	if _, err := writeSystemdUnitConfig(rootful, qcfg, nil, nil, nil); err == nil || !strings.Contains(err.Error(), "rootless") {
+	if _, err := writeSystemdUnitConfig(rootful, selectionForTest(t, deploykit.Selection{Rootful: "sudo"}, "containerd"), qcfg, nil, nil, nil); err == nil || !strings.Contains(err.Error(), "rootless") {
 		t.Errorf("rootful systemd-unit must be refused with the remedy, got %v", err)
 	}
 }
@@ -342,7 +343,7 @@ func TestEngineRunArgv_EnvEntriesVerbatim(t *testing.T) {
 	for _, engine := range []string{"podman", "containerd"} {
 		for _, tc := range cases {
 			t.Run(engine+"/"+tc.name, func(t *testing.T) {
-				argv, _, err := engineRunArgv(engine, false, podEngineRun{
+				argv, _, err := engineRunArgv(selectionForTest(t, deploykit.Selection{}, engine).Target, podEngineRun{
 					Image: "ghcr.io/opencharly/demo:latest",
 					Name:  "charly-demo",
 					Env:   tc.env,
@@ -375,7 +376,7 @@ func TestEngineRunArgv_EnvEntriesVerbatim(t *testing.T) {
 func TestWriteSystemdUnitConfig_RejectsBareEnv(t *testing.T) {
 	rt := kitResolvedRuntimeForTest("containerd", "systemd-unit", "auto")
 	qcfg := deploykit.QuadletConfig{BoxName: "demo", ImageRef: "img", Env: []string{"FOO"}}
-	if _, err := writeSystemdUnitConfig(rt, qcfg, nil, nil, nil); err == nil || !strings.Contains(err.Error(), "inherit host environment variable") {
+	if _, err := writeSystemdUnitConfig(rt, selectionForTest(t, deploykit.Selection{}, "containerd"), qcfg, nil, nil, nil); err == nil || !strings.Contains(err.Error(), "inherit host environment variable") {
 		t.Errorf("bare env entry must be refused, got %v", err)
 	}
 }
diff --git a/candy/plugin-deploy-pod/go.mod b/candy/plugin-deploy-pod/go.mod
index 0133dc5..4c18f52 100644
--- a/candy/plugin-deploy-pod/go.mod
+++ b/candy/plugin-deploy-pod/go.mod
@@ -2,7 +2,7 @@ module github.com/opencharly/plugin-deploy-pod/candy/plugin-deploy-pod
 
 go 1.26.8
 
-require github.com/opencharly/sdk v0.2026295.1359
+require github.com/opencharly/sdk v0.2026296.1207
 
 require (
 	github.com/containerd/containerd/v2 v2.4.1 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/mattn/go-isatty v0.0.17 // indirect
 	github.com/mitchellh/go-wordwrap v1.0.1 // indirect
 	github.com/oklog/run v1.1.0 // indirect
-	github.com/opencharly/spec v0.2026295.1243
+	github.com/opencharly/spec v0.2026296.1036
 	github.com/pelletier/go-toml/v2 v2.4.3 // indirect
 	github.com/protocolbuffers/txtpbfmt v0.0.0-20260217160748-a481f6a22f94 // indirect
 	go.yaml.in/yaml/v3 v3.0.5 // indirect
diff --git a/candy/plugin-deploy-pod/go.sum b/candy/plugin-deploy-pod/go.sum
index db24f46..f709514 100644
--- a/candy/plugin-deploy-pod/go.sum
+++ b/candy/plugin-deploy-pod/go.sum
@@ -88,10 +88,10 @@ github.com/opencharly/plugin-egress/candy/plugin-egress v0.2026237.1419 h1:n3aw1
 github.com/opencharly/plugin-egress/candy/plugin-egress v0.2026237.1419/go.mod h1:5Uz4H7GYjW9WS8yPJGlr+dRv/jrR985d+PbaxJeAJZ8=
 github.com/opencharly/plugin-init/candy/plugin-init v0.2026237.1424 h1:Bldh1DoESB1OhP9itppLcViak/z/INZRIFv+Nm28KNg=
 github.com/opencharly/plugin-init/candy/plugin-init v0.2026237.1424/go.mod h1:tzB4pMZllnxK6Jpyeq5jxg4EQe1p544rQkhNdmPn788=
-github.com/opencharly/sdk v0.2026295.1359 h1:YdhGilt2+7rG8lm1wNoyNbM6L03N7BcRM+PZp6wy0hY=
-github.com/opencharly/sdk v0.2026295.1359/go.mod h1:0muFDG0u8Nro8a4IrgXUYYv2ZH8f7HzqvtwwJUUQErs=
-github.com/opencharly/spec v0.2026295.1243 h1:70q66n3NLbwWgHImNKBGeAViJKKo/yQGjfPWHlHVMN8=
-github.com/opencharly/spec v0.2026295.1243/go.mod h1:VrjzjP2glMbDwP4JNm50OV0MraHFZnFKTl+bVlouoso=
+github.com/opencharly/sdk v0.2026296.1207 h1:FcLqgjV8q6TsFo8cQ4pcAWLYTKM37BFeTj6dpUGf9pY=
+github.com/opencharly/sdk v0.2026296.1207/go.mod h1:JATa64xjMSgqEdOhRRm4iiptgfaetvmHOwybt9kKZko=
+github.com/opencharly/spec v0.2026296.1036 h1:F3Up2zG5hLw2SFVOR772cMQ1JegU/+HjoK8dOPGR/AU=
+github.com/opencharly/spec v0.2026296.1036/go.mod h1:VrjzjP2glMbDwP4JNm50OV0MraHFZnFKTl+bVlouoso=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040=
diff --git a/candy/plugin-deploy-pod/hostname_parity_test.go b/candy/plugin-deploy-pod/hostname_parity_test.go
index cfe3fde..05c997c 100644
--- a/candy/plugin-deploy-pod/hostname_parity_test.go
+++ b/candy/plugin-deploy-pod/hostname_parity_test.go
@@ -17,7 +17,7 @@ import (
 
 func TestBuildStartArgs_HostnameParity(t *testing.T) {
 	argv, err := buildStartArgs(
-		"podman", false,
+		selectionForTest(t, deploykit.Selection{}, "podman"),
 		"ghcr.io/opencharly/demo:latest",
 		1000, 1000,
 		nil,                             // ports
diff --git a/candy/plugin-deploy-pod/image_ensure.go b/candy/plugin-deploy-pod/image_ensure.go
index 2be578a..6e6f7d1 100644
--- a/candy/plugin-deploy-pod/image_ensure.go
+++ b/candy/plugin-deploy-pod/image_ensure.go
@@ -6,8 +6,7 @@ import (
 	"fmt"
 
 	"github.com/opencharly/sdk"
-	"github.com/opencharly/sdk/kit"
-	"github.com/opencharly/spec/container"
+	"github.com/opencharly/sdk/deploykit"
 	"github.com/opencharly/spec/spec"
 )
 
@@ -16,51 +15,109 @@ import (
 // (K-wave W3a B6, the A2-proven peer-dispatch pattern: candy/plugin-preempt already reaches a
 // COMPILED-IN provider word via a plain resolve+invoke; here the plugin reaches build:ensure the
 // SAME way core's own dispatchBuildEnsure does, just via the plugin-to-plugin
-// Executor.InvokeProvider leg instead of core's private providerRegistry). container.LocalImageExists
-// / container.TransferImage are ALREADY portable (spec/container, used directly by candy/plugin-build
-// per that package's own doc comment) — no seam needed for those tiers either. The
-// "pod-config-ensure-image" HostBuild seam + its core handler are DELETED; dispatchBuildEnsure
-// itself STAYS core (it has other core-internal callers: plugin_executor_reverse.go — the former
-// check-run preflight caller is gone too, K-wave 2 cone R4: the check preflight now reaches
-// build:ensure over the SAME InvokeProvider peer dispatch this file uses).
+// Executor.InvokeProvider leg instead of core's private providerRegistry).
 //
-// THE ENGINE IS THE RESOLVED RUN ENGINE. It used to be a package constant naming podman, which
-// made every pod stage its image into podman's store no matter what `engine.run` said, and then
-// transfer it across stores. The pod now stages into the store of the engine it will actually
-// launch on; `build:ensure` receives that same engine word.
-
-// ensureImagePresent guarantees imageRef is available in the resolved run engine's local store,
-// driving the store itself (no host round-trip): (1) already-present short-circuit, (2)
-// build:ensure peer-dispatch (pulls from the registry, falling back to a local build for a
-// project charly.yml entry).
-func ensureImagePresent(ctx context.Context, ex *sdk.Executor, imageRef, runEngine string) error {
-	engine := resolveRunEngine(runEngine)
-	if container.LocalImageExists(engine, imageRef) {
-		return nil
-	}
-	if err := dispatchBuildEnsurePeer(ctx, ex, imageRef, engine); err == nil {
-		return nil
-	}
-	return fmt.Errorf("%w: %s", spec.ErrImageNotLocal, imageRef)
+// THE STORE IS THE W3.7 ONE. The presence probe is deploykit's engine-neutral ImageStore, built
+// from the RESOLVED target (selection.go) — not container.LocalImageExists over an engine word.
+// The store addresses the engine CLI through the capability row (binary, presence argv, daemon
+// address, namespace), so a containerd deploy probes the resolved namespace instead of assuming
+// the capability default, and a daemon failure is surfaced rather than folded into "absent".
+//
+// The staging step stays build:ensure (W3.6), which receives the SAME resolved engine and lands
+// the image in the store this file just probed. The sdk's ImageResolver.Build hook expresses
+// that step for a caller that owns a CharlyImageSource (the engine-neutral built-image store);
+// no production implementation of that source exists at this pin (opencharly/sdk#460), so the
+// documented W3.7 order — target store, then build:ensure — is driven here instead of through
+// the Build hook.
+
+// reverseDeployExecutor adapts the plugin-side executor handle to the spec.DeployExecutor
+// surface deploykit.NewImageStore addresses. The image store calls RunCapture (and nothing
+// else), so every other method fails loudly instead of inventing a local answer.
+type reverseDeployExecutor struct{ ex *sdk.Executor }
+
+func (e reverseDeployExecutor) Venue() string { return "plugin-reverse-channel" }
+func (e reverseDeployExecutor) Kind() string  { return "plugin-reverse-channel" }
+
+func (e reverseDeployExecutor) RunCapture(ctx context.Context, script string) (string, string, int, error) {
+	return e.ex.RunCapture(ctx, script)
+}
+
+func (e reverseDeployExecutor) RunSystem(context.Context, string, spec.EmitOpts) error {
+	return spec.ErrNotSupported
+}
+
+func (e reverseDeployExecutor) RunUser(context.Context, string, spec.EmitOpts) error {
+	return spec.ErrNotSupported
+}
+
+func (e reverseDeployExecutor) RunBuilder(context.Context, spec.BuilderRunOpts) ([]byte, error) {
+	return nil, spec.ErrNotSupported
+}
+
+func (e reverseDeployExecutor) PutFile(context.Context, string, string, uint32, bool, spec.EmitOpts) error {
+	return spec.ErrNotSupported
+}
+
+func (e reverseDeployExecutor) GetFile(context.Context, string, bool, spec.EmitOpts) ([]byte, error) {
+	return nil, spec.ErrNotSupported
+}
+
+func (e reverseDeployExecutor) RunInteractive(context.Context, string) (int, error) {
+	return 0, spec.ErrNotSupported
 }
 
-// resolveRunEngine is the ONE fallback for an unset engine word: the schema's default engine.
-// The runtime resolver always yields a concrete word; this guards direct callers and tests
-// without inventing a literal.
-func resolveRunEngine(runEngine string) string {
-	if runEngine == "" {
-		return spec.DefaultContainerEngine
+func (e reverseDeployExecutor) RunStream(context.Context, string) (int, error) {
+	return 0, spec.ErrNotSupported
+}
+
+func (e reverseDeployExecutor) ResolveHome(context.Context, string) (string, error) {
+	return "", spec.ErrNotSupported
+}
+
+// newImageStore builds the W3.7 store for an already-resolved target.
+func newImageStore(ex *sdk.Executor, target deploykit.ContainerdTarget) (deploykit.ImageStore, error) {
+	if ex == nil {
+		return nil, fmt.Errorf("image store: no executor")
 	}
-	return runEngine
+	return deploykit.NewImageStore(reverseDeployExecutor{ex: ex}, target)
 }
 
-// currentRunEngine resolves the configured run engine for call sites that do not already carry
-// a ResolvedRuntime. A resolution failure degrades to the schema default, never to a literal.
-func currentRunEngine() string {
-	if rt, err := kit.ResolveRuntime(); err == nil {
-		return resolveRunEngine(rt.RunEngine)
+// ensureImagePresent guarantees imageRef is available in the RESOLVED target store, driving the
+// store itself (no host round-trip): (1) the target store already addresses it, (2) otherwise
+// build:ensure (W3.6) stages it — pulling from the registry or building locally — and the store
+// is re-probed to prove the boundary.
+func ensureImagePresent(ctx context.Context, ex *sdk.Executor, imageRef string, target deploykit.ContainerdTarget) error {
+	store, err := newImageStore(ex, target)
+	if err != nil {
+		return err
+	}
+	// W3.7's resolution order starts at the target store. ResolveImage answers the
+	// "absent" case, but it folds a probe FAILURE into the same error, so re-probe on the
+	// error path to keep a daemon fault distinct from a confirmed miss.
+	resolver := &deploykit.ImageResolver{}
+	if _, err := resolver.ResolveImage(ctx, imageRef, store); err == nil {
+		return nil
 	}
-	return resolveRunEngine("")
+	present, err := store.Has(ctx, imageRef)
+	if err != nil {
+		return err
+	}
+	if present {
+		return nil
+	}
+	// Step 4 — build:ensure. It receives the SAME resolved engine word, so the image lands in
+	// the store just probed.
+	if err := dispatchBuildEnsurePeer(ctx, ex, imageRef, string(target.Engine)); err != nil {
+		return fmt.Errorf("%w: %s: %v", spec.ErrImageNotLocal, imageRef, err)
+	}
+	present, err = store.Has(ctx, imageRef)
+	if err != nil {
+		return err
+	}
+	if !present {
+		return fmt.Errorf("%w: %s (build:ensure did not stage it into the %s store)", spec.ErrImageNotLocal, imageRef, target.Engine)
+	}
+	return nil
 }
 
 // dispatchBuildEnsurePeer InvokeProviders the compiled-in build:ensure word — the SAME word
diff --git a/candy/plugin-deploy-pod/image_ensure_test.go b/candy/plugin-deploy-pod/image_ensure_test.go
deleted file mode 100644
index 8cf571f..0000000
--- a/candy/plugin-deploy-pod/image_ensure_test.go
+++ /dev/null
@@ -1,70 +0,0 @@
-package deploypod
-
-import (
-	"errors"
-	"testing"
-
-	"github.com/opencharly/spec/container"
-	"github.com/opencharly/spec/spec"
-)
-
-// image_ensure_test.go — the staging leg stages into the store of the RESOLVED run engine.
-// The retired subject hardcoded podman and transferred across stores; these arms pin the
-// replacement: the configured engine is the only store consulted, `build:ensure` receives that
-// same engine, and an unset word falls to the schema default.
-
-func TestEnsureImagePresent(t *testing.T) {
-	orig := container.LocalImageExists
-	t.Cleanup(func() { container.LocalImageExists = orig })
-
-	t.Run("configured engine has image", func(t *testing.T) {
-		var probed []string
-		container.LocalImageExists = func(engine, ref string) bool {
-			probed = append(probed, engine)
-			return engine == "containerd"
-		}
-		if err := ensureImagePresent(t.Context(), nil, "myimage:latest", "containerd"); err != nil {
-			t.Errorf("expected no error, got: %v", err)
-		}
-		if len(probed) != 1 || probed[0] != "containerd" {
-			t.Errorf("staging must probe exactly the configured engine, got %q", probed)
-		}
-	})
-
-	t.Run("unset engine uses the schema default", func(t *testing.T) {
-		var probed []string
-		container.LocalImageExists = func(engine, ref string) bool {
-			probed = append(probed, engine)
-			return true
-		}
-		if err := ensureImagePresent(t.Context(), nil, "myimage:latest", ""); err != nil {
-			t.Errorf("expected no error, got: %v", err)
-		}
-		if len(probed) != 1 || probed[0] != spec.DefaultContainerEngine {
-			t.Errorf("unset engine must resolve to the schema default %q, got %q", spec.DefaultContainerEngine, probed)
-		}
-	})
-
-	t.Run("image missing reaches build:ensure tier", func(t *testing.T) {
-		container.LocalImageExists = func(engine, ref string) bool { return false }
-		var err error
-		func() {
-			defer func() {
-				if r := recover(); r != nil {
-					// Reached dispatchBuildEnsurePeer's ex.InvokeProvider with a nil test
-					// executor — proves the local-store tier was exhausted first.
-					// build:ensure's own resolve/decode behavior is out of this package's
-					// test scope.
-					err = errors.New("reached build:ensure tier (nil-executor panic, expected)")
-				}
-			}()
-			err = ensureImagePresent(t.Context(), nil, "myimage:latest", "containerd")
-		}()
-		if err == nil {
-			t.Fatal("expected an error (either spec.ErrImageNotLocal or the tier-3 panic marker)")
-		}
-		if !errors.Is(err, spec.ErrImageNotLocal) && err.Error() != "reached build:ensure tier (nil-executor panic, expected)" {
-			t.Errorf("expected spec.ErrImageNotLocal or the tier-3 reach marker, got: %v", err)
-		}
-	})
-}
diff --git a/candy/plugin-deploy-pod/lifecycle.go b/candy/plugin-deploy-pod/lifecycle.go
index 7152062..bdc940f 100644
--- a/candy/plugin-deploy-pod/lifecycle.go
+++ b/candy/plugin-deploy-pod/lifecycle.go
@@ -331,11 +331,14 @@ func podPrepareVenue(ctx context.Context, exec *sdk.Executor, p lifecycleParams)
 	var baseRegistry string
 	if reply.BaseImage != "" {
 		if rt, rterr := kit.ResolveRuntime(); rterr == nil {
-			if baseMeta, merr := deploykit.ExtractMetadata(rt.RunEngine, reply.BaseImage); merr == nil && baseMeta != nil {
-				baseUser = baseMeta.User
-				sec := baseMeta.Security
-				baseSecurity = &sec
-				baseRegistry = baseMeta.Registry
+			sel, selErr := resolvePodSelectionFor(ctx, exec, baseName, "", rt)
+			if selErr == nil {
+				if baseMeta, merr := deploykit.ExtractMetadata(string(sel.Engine), reply.BaseImage); merr == nil && baseMeta != nil {
+					baseUser = baseMeta.User
+					sec := baseMeta.Security
+					baseSecurity = &sec
+					baseRegistry = baseMeta.Registry
+				}
 			}
 		}
 	}
diff --git a/candy/plugin-deploy-pod/resolve.go b/candy/plugin-deploy-pod/resolve.go
index 51226f4..6dc0495 100644
--- a/candy/plugin-deploy-pod/resolve.go
+++ b/candy/plugin-deploy-pod/resolve.go
@@ -43,8 +43,18 @@ func resolvePodStartPlan(ctx context.Context, ex *sdk.Executor, box, instance st
 // .service (unit_emit) is started with `systemctl --user start <svc>`, exactly like the quadlet
 // path starts podman's generated unit. The unit file's existence is the "configured" signal.
 func resolvePodStartSystemdUnit(ctx context.Context, ex *sdk.Executor, box, instance string, rt *kit.ResolvedRuntime) (*spec.PodLifecyclePlan, error) {
-	if rootfulUnsupported(rt) {
-		return nil, rootfulRefusal(rt.RunEngine, rt.RunMode)
+	dc, err := loadDeploy(ctx, ex, "charly start")
+	if err != nil {
+		return nil, err
+	}
+	sel, err := resolvePodSelectionFromConfig(dc, box, instance, rt)
+	if err != nil {
+		return nil, err
+	}
+	// The rootful refusal stays FIRST (as it was): the operator gets the reason they cannot
+	// run this deployment, not a secondary "not configured".
+	if rootfulUnsupported(sel.Target, rt.RunMode) {
+		return nil, rootfulRefusal(sel.Engine, rt.RunMode)
 	}
 	svc := kit.ServiceNameInstance(box, instance)
 	svcDir, err := kit.SystemdUserDir()
@@ -57,19 +67,11 @@ func resolvePodStartSystemdUnit(ctx context.Context, ex *sdk.Executor, box, inst
 		}
 		return nil, err
 	}
-	engine, err := boxEngineForDeploy(ctx, ex, box, instance, rt.RunEngine)
-	if err != nil {
-		return nil, err
-	}
-	dc, err := loadDeploy(ctx, ex, "charly start")
-	if err != nil {
-		return nil, err
-	}
 	plan := &spec.PodLifecyclePlan{
 		Mode:          "systemd-unit",
 		SvcName:       svc,
 		ContainerName: kit.ContainerNameInstance(box, instance),
-		EngineBin:     kit.EngineBinary(engine),
+		EngineBin:     sel.binary(),
 	}
 	encJSON, err := resolvePodEncEnsurePlan(ctx, ex, dc, box, instance, false)
 	if err != nil {
@@ -92,11 +94,11 @@ func resolvePodStartQuadlet(ctx context.Context, ex *sdk.Executor, box, instance
 	if !exists && !directDeploy {
 		return nil, fmt.Errorf("not configured; run 'charly config %s' first", box)
 	}
-	engine, err := boxEngineForDeploy(ctx, ex, box, instance, rt.RunEngine)
+	dc, err := loadDeploy(ctx, ex, "charly start")
 	if err != nil {
 		return nil, err
 	}
-	dc, err := loadDeploy(ctx, ex, "charly start")
+	sel, err := resolvePodSelectionFromConfig(dc, box, instance, rt)
 	if err != nil {
 		return nil, err
 	}
@@ -105,7 +107,7 @@ func resolvePodStartQuadlet(ctx context.Context, ex *sdk.Executor, box, instance
 		SvcName:       kit.ServiceNameInstance(box, instance),
 		ContainerName: kit.ContainerNameInstance(box, instance),
 		DirectDeploy:  directDeploy,
-		EngineBin:     kit.EngineBinary(engine),
+		EngineBin:     sel.binary(),
 	}
 	// Encrypted-volume mounts are skipped in direct-deploy mode (those require systemd-run --scope).
 	if !directDeploy {
@@ -119,30 +121,12 @@ func resolvePodStartQuadlet(ctx context.Context, ex *sdk.Executor, box, instance
 	return plan, nil
 }
 
-// boxEngineForDeploy reads the per-host deploy config's Engine override PLUGIN-SIDE via loadDeploy
-// (the cycle-free loaderkit read) — replicating deploykit.ResolveBoxEngineForDeploy's lookup WITHOUT
-// the DeployStateHost package-var dependency (a plugin calling the raw deploykit function would
-// silently no-op out-of-process — the placement-dependency hazard). #55 coneC-dsh — the
-// pod-config-box-engine host seam is DELETED.
-func boxEngineForDeploy(ctx context.Context, ex *sdk.Executor, box, instance, globalEngine string) (string, error) {
-	dc, err := loadDeploy(ctx, ex, "ResolveBoxEngineForDeploy")
-	if err != nil {
-		return globalEngine, err
-	}
-	if dc != nil {
-		if entry, ok := dc.Lookup(box, instance); ok && entry.Engine != "" {
-			return string(entry.Engine), nil
-		}
-	}
-	return globalEngine, nil
-}
-
 // podRuntimeImage is the resolved pod image context shared by the start-plan resolver and the F12
-// shell resolver (R3): the runtime-detected devices, the resolved engine + image ref + baked
-// metadata, the per-host deploy config, and the resolved volume backing.
+// shell resolver (R3): the runtime-detected devices, the RESOLVED selection (engine + target),
+// the image ref + baked metadata, the per-host deploy config, and the resolved volume backing.
 type podRuntimeImage struct {
 	detected   spec.DetectedDevices
-	engine     string
+	sel        podSelection
 	imageRef   string
 	meta       *spec.BoxMetadata
 	dc         *deploykit.DeployConfig
@@ -153,17 +137,29 @@ type podRuntimeImage struct {
 // resolvePodRuntimeImage resolves the pod's runtime image context — the identical HEAD both
 // resolvePodStartDirect (`charly start` direct-mode) and resolvePodShellPlan (`charly shell`)
 // compute: device detection + CDI, the deploy overlay, the image ref, and the volume backing.
+// It loads the overlay and resolves the selection once, then hands both to
+// resolvePodRuntimeImageWith.
 func resolvePodRuntimeImage(ctx context.Context, ex *sdk.Executor, box, instance, tag string, rt *kit.ResolvedRuntime, noAutoDetect bool, volumeFlag, bind []string) (*podRuntimeImage, error) {
-	// Device detection runs plugin-side (detect_devices.go — the relocated
-	// "pod-config-detect-devices" seam). The former seam passed Engine=rt.RunEngine, so the
-	// ensure-cdi leg fires for podman — preserved exactly.
-	detected := detectDevices(ctx, ex, noAutoDetect, rt.RunEngine)
-	engine := rt.RunEngine
-
 	dc, err := loadDeploy(ctx, ex, "charly pod runtime image")
 	if err != nil {
 		return nil, err
 	}
+	sel, err := resolvePodSelectionFromConfig(dc, box, instance, rt)
+	if err != nil {
+		return nil, err
+	}
+	return resolvePodRuntimeImageWith(ctx, ex, box, instance, tag, rt, dc, sel, noAutoDetect, volumeFlag, bind)
+}
+
+// resolvePodRuntimeImageWith resolves the image context against an already-loaded overlay and an
+// already-resolved selection, so a caller that must refuse a rootful target BEFORE any image
+// work does not pay a second overlay read.
+func resolvePodRuntimeImageWith(ctx context.Context, ex *sdk.Executor, box, instance, tag string, rt *kit.ResolvedRuntime, dc *deploykit.DeployConfig, sel podSelection, noAutoDetect bool, volumeFlag, bind []string) (*podRuntimeImage, error) {
+	// Device detection runs plugin-side (detect_devices.go — the relocated
+	// "pod-config-detect-devices" seam). It receives the RESOLVED engine, so the CDI ensure
+	// leg fires for the engine that will actually launch the container.
+	detected := detectDevices(ctx, ex, noAutoDetect, string(sel.Engine))
+
 	var deployVolumes []spec.DeployVolume
 	if dc != nil {
 		if overlay, ok := dc.Lookup(box, instance); ok {
@@ -175,10 +171,10 @@ func resolvePodRuntimeImage(ctx context.Context, ex *sdk.Executor, box, instance
 	if err != nil {
 		return nil, err
 	}
-	if err := ensureImagePresent(ctx, ex, imageRef, rt.RunEngine); err != nil {
+	if err := ensureImagePresent(ctx, ex, imageRef, sel.Target); err != nil {
 		return nil, err
 	}
-	metaPtr, err := deploykit.ExtractMetadata(rt.RunEngine, imageRef)
+	metaPtr, err := deploykit.ExtractMetadata(string(sel.Engine), imageRef)
 	if err != nil {
 		return nil, err
 	}
@@ -186,9 +182,6 @@ func resolvePodRuntimeImage(ctx context.Context, ex *sdk.Executor, box, instance
 		return nil, fmt.Errorf("image %s has no embedded metadata; rebuild with latest charly", imageRef)
 	}
 	meta := *metaPtr
-	if meta.Engine != "" {
-		engine = meta.Engine
-	}
 	if dc != nil {
 		deploykit.MergeDeployOntoMetadata(&meta, dc, box, instance)
 	}
@@ -200,7 +193,7 @@ func resolvePodRuntimeImage(ctx context.Context, ex *sdk.Executor, box, instance
 			imageRef = ref
 		}
 	}
-	return &podRuntimeImage{detected: detected, engine: engine, imageRef: imageRef, meta: &meta, dc: dc, volumes: volumes, bindMounts: bindMounts}, nil
+	return &podRuntimeImage{detected: detected, sel: sel, imageRef: imageRef, meta: &meta, dc: dc, volumes: volumes, bindMounts: bindMounts}, nil
 }
 
 // resolvePodStartDirect resolves the direct-mode (non-unit) start plan: the launch argv the
@@ -208,14 +201,22 @@ func resolvePodRuntimeImage(ctx context.Context, ex *sdk.Executor, box, instance
 // instead of running it, and threads the enc-ensure + tunnel legs. The sidecar-in-direct-mode
 // rejection is preserved (sidecars require a unit run mode).
 func resolvePodStartDirect(ctx context.Context, ex *sdk.Executor, box, instance string, rt *kit.ResolvedRuntime, opts spec.PodStartOpts) (*spec.PodLifecyclePlan, error) {
-	if rootfulUnsupported(rt) {
-		return nil, rootfulRefusal(rt.RunEngine, rt.RunMode)
+	dc, err := loadDeploy(ctx, ex, "charly start")
+	if err != nil {
+		return nil, err
+	}
+	sel, err := resolvePodSelectionFromConfig(dc, box, instance, rt)
+	if err != nil {
+		return nil, err
+	}
+	if rootfulUnsupported(sel.Target, rt.RunMode) {
+		return nil, rootfulRefusal(sel.Engine, rt.RunMode)
 	}
-	img, err := resolvePodRuntimeImage(ctx, ex, box, instance, "", rt, opts.NoAutoDetect, opts.VolumeFlag, opts.Bind)
+	img, err := resolvePodRuntimeImageWith(ctx, ex, box, instance, "", rt, dc, sel, opts.NoAutoDetect, opts.VolumeFlag, opts.Bind)
 	if err != nil {
 		return nil, err
 	}
-	detected, engine, imageRef, meta, dc := img.detected, img.engine, img.imageRef, img.meta, img.dc
+	detected, imageRef, meta := img.detected, img.imageRef, img.meta
 	volumes, bindMounts := img.volumes, img.bindMounts
 
 	if dc != nil {
@@ -258,7 +259,7 @@ func resolvePodStartDirect(ctx context.Context, ex *sdk.Executor, box, instance
 	}
 	envVars = appendAutoDetectedEnv(envVars, detected)
 
-	resolvedNetwork, netErr := kit.ResolveNetwork(network, engine)
+	resolvedNetwork, netErr := kit.ResolveNetwork(network, string(sel.Engine))
 	if netErr != nil {
 		return nil, netErr
 	}
@@ -272,7 +273,7 @@ func resolvePodStartDirect(ctx context.Context, ex *sdk.Executor, box, instance
 		// deleted "deploy-config-save-state" host seam.
 		deploySaveState(ctx, ex, box, instance, spec.SaveDeployStateInput{Ports: ports, SetPorts: true})
 	}
-	if conflicts := kit.CheckPortAvailability(ports, rt.BindAddress, engine); len(conflicts) > 0 {
+	if conflicts := kit.CheckPortAvailability(ports, rt.BindAddress, string(sel.Engine)); len(conflicts) > 0 {
 		return nil, fmt.Errorf("port conflicts detected:%s", kit.FormatPortConflicts(conflicts, box))
 	}
 
@@ -290,7 +291,7 @@ func resolvePodStartDirect(ctx context.Context, ex *sdk.Executor, box, instance
 
 	name := kit.ContainerNameInstance(box, instance)
 	workDir := deploykit.ResolveWorkingDir(volumes, bindMounts, home, box, instance)
-	argv, err := buildStartArgs(engine, podRootful(rt), imageRef, uid, gid, ports, name, volumes, bindMounts, detected.GPU, rt.BindAddress, envVars, security, entrypoint, workDir, resolvedNetwork)
+	argv, err := buildStartArgs(sel, imageRef, uid, gid, ports, name, volumes, bindMounts, detected.GPU, rt.BindAddress, envVars, security, entrypoint, workDir, resolvedNetwork)
 	if err != nil {
 		return nil, err
 	}
@@ -298,7 +299,7 @@ func resolvePodStartDirect(ctx context.Context, ex *sdk.Executor, box, instance
 	tunnel := resolvePodTunnelPlan(dc, box, instance)
 
 	return &spec.PodLifecyclePlan{
-		Mode: "direct", ContainerName: name, RunArgv: argv, EngineBin: kit.EngineBinary(engine),
+		Mode: "direct", ContainerName: name, RunArgv: argv, EngineBin: sel.binary(),
 		Enc: encJSON, Tunnel: tunnel,
 	}, nil
 }
@@ -310,18 +311,18 @@ func resolvePodStopPlan(ctx context.Context, ex *sdk.Executor, box, instance str
 		return nil, err
 	}
 	quadletActive, _ := kit.QuadletExistsInstance(box, instance)
-	engine, err := boxEngineForDeploy(ctx, ex, box, instance, rt.RunEngine)
+	dc, err := loadDeploy(ctx, ex, "charly stop")
 	if err != nil {
 		return nil, err
 	}
-	dc, err := loadDeploy(ctx, ex, "charly stop")
+	sel, err := resolvePodSelectionFromConfig(dc, box, instance, rt)
 	if err != nil {
 		return nil, err
 	}
 	plan := &spec.PodLifecyclePlan{
 		ContainerName: kit.ContainerNameInstance(box, instance),
 		SvcName:       kit.ServiceNameInstance(box, instance),
-		EngineBin:     kit.EngineBinary(engine),
+		EngineBin:     sel.binary(),
 		Unmount:       unmount,
 		Tunnel:        resolvePodTunnelPlan(dc, box, instance),
 	}
diff --git a/candy/plugin-deploy-pod/resolve_f12.go b/candy/plugin-deploy-pod/resolve_f12.go
index 4513169..d8f83f0 100644
--- a/candy/plugin-deploy-pod/resolve_f12.go
+++ b/candy/plugin-deploy-pod/resolve_f12.go
@@ -39,7 +39,7 @@ func resolvePodShellPlan(ctx context.Context, ex *sdk.Executor, box, instance st
 	if err != nil {
 		return nil, err
 	}
-	detected, engine, imageRef, meta, dc := img.detected, img.engine, img.imageRef, img.meta, img.dc
+	detected, sel, imageRef, meta, dc := img.detected, img.sel, img.imageRef, img.meta, img.dc
 	volumes, bindMounts := img.volumes, img.bindMounts
 
 	uid, gid, home := meta.UID, meta.GID, meta.Home
@@ -65,10 +65,10 @@ func resolvePodShellPlan(ctx context.Context, ex *sdk.Executor, box, instance st
 	agentFwd := kit.ResolveAgentForwarding(rt, deployBox, home)
 
 	name := kit.ContainerNameInstance(box, instance)
-	if containerRunningLocal(engine, name) {
+	if containerRunningLocal(sel.Target, name) {
 		execEnv := append(append([]string{}, envVars...), agentFwd.Env...)
 		workDir := deploykit.ResolveWorkingDir(volumes, bindMounts, home, box, instance)
-		argv := buildExecArgs(engine, name, uid, gid, command, execEnv, workDir, opts.Interactive)
+		argv := buildExecArgs(sel.Target, name, uid, gid, command, execEnv, workDir, opts.Interactive)
 		return &spec.PodLiveStdioPlan{Script: hostAttachScript(argv, opts.WrapPTY)}, nil
 	}
 
@@ -86,24 +86,25 @@ func resolvePodShellPlan(ctx context.Context, ex *sdk.Executor, box, instance st
 		security.Mounts = deploykit.AppendUnique(security.Mounts, v)
 	}
 	envVars = append(envVars, agentFwd.Env...)
-	resolvedNetwork, err := kit.ResolveNetwork(network, engine)
+	resolvedNetwork, err := kit.ResolveNetwork(network, string(sel.Engine))
 	if err != nil {
 		return nil, err
 	}
 	workDir := deploykit.ResolveWorkingDir(volumes, bindMounts, home, box, instance)
-	argv := buildShellArgs(engine, imageRef, uid, gid, ports, volumes, bindMounts, detected.GPU, command, rt.BindAddress, envVars, security, workDir, opts.Interactive, resolvedNetwork)
+	argv := buildShellArgs(sel.Target, imageRef, uid, gid, ports, volumes, bindMounts, detected.GPU, command, rt.BindAddress, envVars, security, workDir, opts.Interactive, resolvedNetwork)
 	return &spec.PodLiveStdioPlan{Script: hostAttachScript(argv, opts.WrapPTY)}, nil
 }
 
 // resolvePodCmdPlan resolves `charly cmd`: an `<engine> exec -i [-e agentEnv] <ctr> sh -c <command>`
 // into the RUNNING container (or a named sidecar), with agent-forwarding env resolved.
 func resolvePodCmdPlan(ctx context.Context, ex *sdk.Executor, box, instance string, cmd []string, opts spec.PodCmdOpts) (*spec.PodLiveStdioPlan, error) {
-	var engine, name string
+	var sel podSelection
+	var name string
 	var err error
 	if opts.Sidecar != "" {
-		engine, name, err = resolveSidecarContainerLocal(ctx, ex, box, instance, opts.Sidecar)
+		sel, name, err = resolveSidecarContainerLocal(ctx, ex, box, instance, opts.Sidecar)
 	} else {
-		engine, name, err = resolveContainerLocal(ctx, ex, box, instance)
+		sel, name, err = resolveContainerLocal(ctx, ex, box, instance)
 	}
 	if err != nil {
 		return nil, err
@@ -121,7 +122,7 @@ func resolvePodCmdPlan(ctx context.Context, ex *sdk.Executor, box, instance stri
 		agentEnv = agentFwd.Env
 	}
 
-	argv := []string{engine, "exec", "-i"}
+	argv := prependEngineGlobalArgs(sel.Target, []string{sel.binary(), "exec", "-i"})
 	argv = append(argv, envArgs(agentEnv)...)
 	argv = append(argv, name, "sh", "-c", strings.Join(cmd, " "))
 	return &spec.PodLiveStdioPlan{Script: shellQuoteArgs(argv)}, nil
@@ -151,16 +152,15 @@ func resolvePodLogsPlan(ctx context.Context, ex *sdk.Executor, box, instance str
 		return &spec.PodLiveStdioPlan{Script: shellQuoteArgs(argv)}, nil
 	}
 
-	engineName, err := boxEngineForDeploy(ctx, ex, boxName, instance, rt.RunEngine)
+	sel, err := resolvePodSelectionFor(ctx, ex, boxName, instance, rt)
 	if err != nil {
 		return nil, err
 	}
-	engine := kit.EngineBinary(engineName)
 	name := kit.ContainerNameInstance(boxName, instance)
 	if opts.Sidecar != "" {
 		name = kit.SidecarContainerNameInstance(boxName, instance, opts.Sidecar)
 	}
-	argv := []string{engine, "logs"}
+	argv := prependEngineGlobalArgs(sel.Target, []string{sel.binary(), "logs"})
 	if opts.Follow {
 		argv = append(argv, "-f")
 	}
@@ -184,7 +184,8 @@ func hostAttachScript(argv []string, wrapPTY bool) string {
 
 // buildShellArgs constructs the `<engine> run --rm -it|-i …` argument list. interactive is the
 // HOST-resolved tty boolean (replaces the former package-level forceTTY||isTerminal() read).
-func buildShellArgs(engine, imageRef string, uid, gid int, ports []string, volumes []deploykit.VolumeMount, bindMounts []deploykit.ResolvedBindMount, gpu bool, command string, bindAddr string, envVars []string, security spec.SecurityConfig, workingDir string, interactive bool, network ...string) []string {
+func buildShellArgs(target deploykit.ContainerdTarget, imageRef string, uid, gid int, ports []string, volumes []deploykit.VolumeMount, bindMounts []deploykit.ResolvedBindMount, gpu bool, command string, bindAddr string, envVars []string, security spec.SecurityConfig, workingDir string, interactive bool, network ...string) []string {
+	engine := string(target.Engine)
 	binary := kit.EngineBinary(engine)
 	interactiveFlag := "-i"
 	if interactive {
@@ -226,12 +227,12 @@ func buildShellArgs(engine, imageRef string, uid, gid int, ports []string, volum
 	if command != "" {
 		args = append(args, "-c", command)
 	}
-	return args
+	return prependEngineGlobalArgs(target, args)
 }
 
 // buildExecArgs constructs the `<engine> exec -it|-i …` list for attaching to a running container.
-func buildExecArgs(engine, name string, uid, gid int, command string, envVars []string, workingDir string, interactive bool) []string {
-	binary := kit.EngineBinary(engine)
+func buildExecArgs(target deploykit.ContainerdTarget, name string, uid, gid int, command string, envVars []string, workingDir string, interactive bool) []string {
+	binary := kit.EngineBinary(string(target.Engine))
 	interactiveFlag := "-i"
 	if interactive {
 		interactiveFlag = "-it"
@@ -246,7 +247,7 @@ func buildExecArgs(engine, name string, uid, gid int, command string, envVars []
 	if command != "" {
 		args = append(args, "-c", command)
 	}
-	return args
+	return prependEngineGlobalArgs(target, args)
 }
 
 // shellQuoteArgs joins args into a shell-safe command string.
@@ -264,8 +265,11 @@ func shellQuoteArgs(args []string) string {
 
 // containerRunningLocal mirrors charly-core shell.go's defaultContainerRunning (the `<engine>
 // container inspect --format '{{.State.Running}}'`-based check) VERBATIM — pure engine-exec probe.
-func containerRunningLocal(engine, name string) bool {
-	out, err := exec.Command(kit.EngineBinary(engine), "container", "inspect", "--format", "{{.State.Running}}", name).Output()
+func containerRunningLocal(target deploykit.ContainerdTarget, name string) bool {
+	argv := prependEngineGlobalArgs(target, []string{
+		kit.EngineBinary(string(target.Engine)), "container", "inspect", "--format", "{{.State.Running}}", name,
+	})
+	out, err := exec.Command(argv[0], argv[1:]...).Output()
 	if err != nil {
 		return false
 	}
@@ -273,42 +277,40 @@ func containerRunningLocal(engine, name string) bool {
 }
 
 // resolveContainerLocal mirrors charly-core container.go's resolveContainer.
-func resolveContainerLocal(ctx context.Context, ex *sdk.Executor, box, instance string) (engine, name string, err error) {
+func resolveContainerLocal(ctx context.Context, ex *sdk.Executor, box, instance string) (podSelection, string, error) {
 	if box == "." {
-		return "", "", nil
+		return podSelection{}, "", nil
 	}
 	rt, err := kit.ResolveRuntime()
 	if err != nil {
-		return "", "", err
+		return podSelection{}, "", err
 	}
 	boxName := resolveBoxName(box)
-	runEngine, err := boxEngineForDeploy(ctx, ex, boxName, instance, rt.RunEngine)
+	sel, err := resolvePodSelectionFor(ctx, ex, boxName, instance, rt)
 	if err != nil {
-		return "", "", err
+		return podSelection{}, "", err
 	}
-	engine = kit.EngineBinary(runEngine)
-	name = kit.ContainerNameInstance(boxName, instance)
-	if !containerRunningLocal(engine, name) {
-		return "", "", fmt.Errorf("container %s is not running", name)
+	name := kit.ContainerNameInstance(boxName, instance)
+	if !containerRunningLocal(sel.Target, name) {
+		return podSelection{}, "", fmt.Errorf("container %s is not running", name)
 	}
-	return engine, name, nil
+	return sel, name, nil
 }
 
 // resolveSidecarContainerLocal mirrors sdk/deploykit.ResolveSidecarContainer.
-func resolveSidecarContainerLocal(ctx context.Context, ex *sdk.Executor, box, instance, sidecar string) (engine, name string, err error) {
+func resolveSidecarContainerLocal(ctx context.Context, ex *sdk.Executor, box, instance, sidecar string) (podSelection, string, error) {
 	rt, err := kit.ResolveRuntime()
 	if err != nil {
-		return "", "", err
+		return podSelection{}, "", err
 	}
 	boxName := resolveBoxName(box)
-	runEngine, err := boxEngineForDeploy(ctx, ex, boxName, instance, rt.RunEngine)
+	sel, err := resolvePodSelectionFor(ctx, ex, boxName, instance, rt)
 	if err != nil {
-		return "", "", err
+		return podSelection{}, "", err
 	}
-	engine = kit.EngineBinary(runEngine)
-	name = kit.SidecarContainerNameInstance(boxName, instance, sidecar)
-	if !containerRunningLocal(engine, name) {
-		return "", "", fmt.Errorf("sidecar container %s is not running", name)
+	name := kit.SidecarContainerNameInstance(boxName, instance, sidecar)
+	if !containerRunningLocal(sel.Target, name) {
+		return podSelection{}, "", fmt.Errorf("sidecar container %s is not running", name)
 	}
-	return engine, name, nil
+	return sel, name, nil
 }
diff --git a/candy/plugin-deploy-pod/resolve_f12_test.go b/candy/plugin-deploy-pod/resolve_f12_test.go
index d697889..e8e6814 100644
--- a/candy/plugin-deploy-pod/resolve_f12_test.go
+++ b/candy/plugin-deploy-pod/resolve_f12_test.go
@@ -20,7 +20,7 @@ import (
 // functions did not move).
 
 func TestBuildShellArgs(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -34,7 +34,7 @@ func TestBuildShellArgs(t *testing.T) {
 }
 
 func TestBuildShellArgsCustomUIDGID(t *testing.T) {
-	args := buildShellArgs("docker", "fedora:latest", 1001, 1002, nil, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "fedora:latest", 1001, 1002, nil, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -48,7 +48,7 @@ func TestBuildShellArgsCustomUIDGID(t *testing.T) {
 }
 
 func TestBuildShellArgsWithPorts(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/fedora:latest", 1000, 1000, []string{"9090:9090", "8080:8080"}, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/fedora:latest", 1000, 1000, []string{"9090:9090", "8080:8080"}, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -64,7 +64,7 @@ func TestBuildShellArgsWithPorts(t *testing.T) {
 }
 
 func TestBuildShellArgsWithSinglePort(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/fedora:latest", 1000, 1000, []string{"8080"}, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/fedora:latest", 1000, 1000, []string{"8080"}, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -82,7 +82,7 @@ func TestBuildShellArgsWithVolumes(t *testing.T) {
 	volumes := []deploykit.VolumeMount{
 		{VolumeName: "charly-openclaw-data", ContainerPath: "/home/user/.openclaw"},
 	}
-	args := buildShellArgs("docker", "ghcr.io/opencharly/openclaw:latest", 1000, 1000, nil, volumes, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/openclaw:latest", 1000, 1000, nil, volumes, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -97,7 +97,7 @@ func TestBuildShellArgsWithVolumes(t *testing.T) {
 }
 
 func TestBuildShellArgsWithGPU(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, true, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, true, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -112,7 +112,7 @@ func TestBuildShellArgsWithGPU(t *testing.T) {
 }
 
 func TestBuildShellArgsWithGPUPodman(t *testing.T) {
-	args := buildShellArgs("podman", "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, true, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "podman"), "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, true, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"podman", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -127,7 +127,7 @@ func TestBuildShellArgsWithGPUPodman(t *testing.T) {
 }
 
 func TestBuildShellArgsWithoutGPU(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", false)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", false)
 	for _, arg := range args {
 		if arg == "--gpus" {
 			t.Error("buildShellArgs(gpu=false) should not contain --gpus")
@@ -136,7 +136,7 @@ func TestBuildShellArgsWithoutGPU(t *testing.T) {
 }
 
 func TestBuildShellArgsWithCommand(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "echo hello", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", false)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "echo hello", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", false)
 	want := []string{
 		"docker", "run", "--rm", "-i",
 		"-w", "/workspace",
@@ -151,7 +151,7 @@ func TestBuildShellArgsWithCommand(t *testing.T) {
 }
 
 func TestBuildShellArgsWithCommandAndGPU(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, true, "nvidia-smi", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", false)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/ollama:latest", 1000, 1000, nil, nil, nil, true, "nvidia-smi", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", false)
 	want := []string{
 		"docker", "run", "--rm", "-i",
 		"-w", "/workspace",
@@ -167,7 +167,7 @@ func TestBuildShellArgsWithCommandAndGPU(t *testing.T) {
 }
 
 func TestBuildExecArgs(t *testing.T) {
-	args := buildExecArgs("docker", "charly-fedora", 1000, 1000, "", nil, "/workspace", true)
+	args := buildExecArgs(targetForTest(t, "docker"), "charly-fedora", 1000, 1000, "", nil, "/workspace", true)
 	want := []string{
 		"docker", "exec", "-it",
 		"--user", "1000:1000",
@@ -181,7 +181,7 @@ func TestBuildExecArgs(t *testing.T) {
 }
 
 func TestBuildExecArgsWithCommand(t *testing.T) {
-	args := buildExecArgs("docker", "charly-openclaw", 1000, 1000, "echo hello", nil, "/workspace", false)
+	args := buildExecArgs(targetForTest(t, "docker"), "charly-openclaw", 1000, 1000, "echo hello", nil, "/workspace", false)
 	want := []string{
 		"docker", "exec", "-i",
 		"--user", "1000:1000",
@@ -196,7 +196,7 @@ func TestBuildExecArgsWithCommand(t *testing.T) {
 }
 
 func TestBuildShellArgsWithCommandTTY(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "openclaw tui", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "openclaw tui", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -211,7 +211,7 @@ func TestBuildShellArgsWithCommandTTY(t *testing.T) {
 }
 
 func TestBuildExecArgsWithCommandTTY(t *testing.T) {
-	args := buildExecArgs("docker", "charly-openclaw", 1000, 1000, "openclaw tui", nil, "/workspace", true)
+	args := buildExecArgs(targetForTest(t, "docker"), "charly-openclaw", 1000, 1000, "openclaw tui", nil, "/workspace", true)
 	want := []string{
 		"docker", "exec", "-it",
 		"--user", "1000:1000",
@@ -229,7 +229,7 @@ func TestBuildExecArgsWithCommandTTY(t *testing.T) {
 // real terminal, interactive = force||isTerminal = true||false = true (the exact scenario
 // hostAttachScript's wrapPTY additionally handles by wrapping in script(1) — tested separately).
 func TestBuildShellArgsForceTTY(t *testing.T) {
-	args := buildShellArgs("docker", "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "openclaw models auth login", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "ghcr.io/opencharly/fedora:latest", 1000, 1000, nil, nil, nil, false, "openclaw models auth login", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 	want := []string{
 		"docker", "run", "--rm", "-it",
 		"-w", "/workspace",
@@ -243,7 +243,7 @@ func TestBuildShellArgsForceTTY(t *testing.T) {
 }
 
 func TestBuildExecArgsForceTTY(t *testing.T) {
-	args := buildExecArgs("docker", "charly-openclaw", 1000, 1000, "openclaw models auth login", nil, "/workspace", true)
+	args := buildExecArgs(targetForTest(t, "docker"), "charly-openclaw", 1000, 1000, "openclaw models auth login", nil, "/workspace", true)
 	want := []string{
 		"docker", "exec", "-it",
 		"--user", "1000:1000",
@@ -258,7 +258,7 @@ func TestBuildExecArgsForceTTY(t *testing.T) {
 }
 
 func TestBuildExecArgsCustomUIDGID(t *testing.T) {
-	args := buildExecArgs("podman", "charly-ubuntu", 1001, 1002, "", nil, "/workspace", true)
+	args := buildExecArgs(targetForTest(t, "podman"), "charly-ubuntu", 1001, 1002, "", nil, "/workspace", true)
 	want := []string{
 		"podman", "exec", "-it",
 		"--user", "1001:1002",
@@ -273,7 +273,7 @@ func TestBuildExecArgsCustomUIDGID(t *testing.T) {
 
 func TestBuildShellArgsWithEnvVars(t *testing.T) {
 	envVars := []string{"DB_HOST=localhost", "SECRET=abc"}
-	args := buildShellArgs("docker", "myapp:latest", 1000, 1000, nil, nil, nil, false, "", "127.0.0.1", envVars, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "myapp:latest", 1000, 1000, nil, nil, nil, false, "", "127.0.0.1", envVars, spec.SecurityConfig{}, "/workspace", true)
 
 	entryIdx := -1
 	envIdx := -1
@@ -309,7 +309,7 @@ func TestBuildShellArgsWithBindMounts(t *testing.T) {
 	bindMounts := []deploykit.ResolvedBindMount{
 		{Name: "data", HostPath: "/home/user/data", ContPath: "/home/user/.myapp"},
 	}
-	args := buildShellArgs("docker", "myapp:latest", 1000, 1000, nil, nil, bindMounts, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "myapp:latest", 1000, 1000, nil, nil, bindMounts, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 
 	found := false
 	for i, arg := range args {
@@ -333,7 +333,7 @@ func TestBuildShellArgsWithBindMountsPodman(t *testing.T) {
 	bindMounts := []deploykit.ResolvedBindMount{
 		{Name: "data", HostPath: "/home/user/data", ContPath: "/home/user/.myapp"},
 	}
-	args := buildShellArgs("podman", "myapp:latest", 1000, 1000, nil, nil, bindMounts, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "podman"), "myapp:latest", 1000, 1000, nil, nil, bindMounts, false, "", "127.0.0.1", nil, spec.SecurityConfig{}, "/workspace", true)
 
 	found := slices.Contains(args, "--userns=keep-id:uid=1000,gid=1000")
 	if !found {
@@ -346,7 +346,7 @@ func TestBuildShellArgsWithCapAdd(t *testing.T) {
 		CapAdd:  []string{"SYS_ADMIN"},
 		Devices: []string{"/dev/fuse"},
 	}
-	args := buildShellArgs("docker", "myimage:latest", 0, 0, nil, nil, nil, false, "", "127.0.0.1", nil, sec, "/workspace", true)
+	args := buildShellArgs(targetForTest(t, "docker"), "myimage:latest", 0, 0, nil, nil, nil, false, "", "127.0.0.1", nil, sec, "/workspace", true)
 	foundCap := false
 	foundDev := false
 	for i, arg := range args {
@@ -367,7 +367,7 @@ func TestBuildShellArgsWithCapAdd(t *testing.T) {
 
 func TestBuildExecArgsWithEnvVars(t *testing.T) {
 	envVars := []string{"FOO=bar"}
-	args := buildExecArgs("docker", "charly-myapp", 1000, 1000, "", envVars, "/workspace", true)
+	args := buildExecArgs(targetForTest(t, "docker"), "charly-myapp", 1000, 1000, "", envVars, "/workspace", true)
 	want := []string{
 		"docker", "exec", "-it",
 		"--user", "1000:1000",
diff --git a/candy/plugin-deploy-pod/resolve_ref.go b/candy/plugin-deploy-pod/resolve_ref.go
index df2f5f1..fda3d68 100644
--- a/candy/plugin-deploy-pod/resolve_ref.go
+++ b/candy/plugin-deploy-pod/resolve_ref.go
@@ -41,8 +41,18 @@ func resolveDeployRefLocal(ctx context.Context, ex *sdk.Executor, box, instance,
 		return "", "", err
 	}
 	deployBoxName = resolveDeployBoxNameLocal(ctx, ex, dc, box, instance)
-	if ov := resolvedOverlayImage(dc, box, instance); ov != "" && kit.LocalImageExists(currentRunEngine(), ov) {
-		return deployBoxName, ov, nil
+	if ov := resolvedOverlayImage(dc, box, instance); ov != "" {
+		rt, rtErr := kit.ResolveRuntime()
+		if rtErr != nil {
+			return "", "", rtErr
+		}
+		sel, selErr := resolvePodSelectionFromConfig(dc, box, instance, rt)
+		if selErr != nil {
+			return "", "", selErr
+		}
+		if kit.LocalImageExists(string(sel.Engine), ov) {
+			return deployBoxName, ov, nil
+		}
 	}
 	return deployBoxName, kit.ResolveShellImageRef("", deployBoxName, tag), nil
 }
diff --git a/candy/plugin-deploy-pod/selection.go b/candy/plugin-deploy-pod/selection.go
new file mode 100644
index 0000000..5e77f01
--- /dev/null
+++ b/candy/plugin-deploy-pod/selection.go
@@ -0,0 +1,82 @@
+package deploypod
+
+import (
+	"context"
+
+	"github.com/opencharly/sdk"
+	"github.com/opencharly/sdk/deploykit"
+	"github.com/opencharly/sdk/kit"
+	"github.com/opencharly/spec/spec"
+)
+
+// selection.go — the pod substrate's ONE consumer of the sdk selection model. Every site in this
+// package that needs to know which engine a deploy runs on calls resolvePodSelection; none reads
+// rt.RunEngine, re-derives the deploy-entry-over-runtime precedence by hand, or carries an engine
+// word literal.
+//
+// The precedence is the sdk's (flag > deploy entity > env > settings file > built-in default).
+// This package has no engine FLAG, so the flag level is always empty; the deploy entity comes
+// from the per-host overlay entry, and the env/settings fold arrives already resolved on
+// rt.RunEngine / rt.Namespace (hostenv.ResolveRuntime owns it). `auto` is a selector the resolver
+// handles at its own level.
+//
+// The candy `engine:` constraint has no source on this side of the plugin boundary: the deploy
+// plugin receives a deploy node, not the resolved candy graph. It is therefore left empty here,
+// exactly as the sibling deploy plugin does; the constraint stays enforced where the graph exists
+// (`charly deploy add`).
+
+// podSelection is the resolved selection for one deploy: the engine word and the engine target
+// facts (address, namespace, rootful, privilege prefix) that a run plan, unit emission and image
+// store need.
+type podSelection struct {
+	Engine spec.EngineName
+	Target deploykit.ContainerdTarget
+}
+
+// binary is the engine's CLI binary. It is a capability fact (containerd is driven through
+// nerdctl), never an engine-word spelling.
+func (s podSelection) binary() string { return kit.EngineBinary(string(s.Engine)) }
+
+// deploySelection maps the deploy entity level of the precedence chain onto the resolver's
+// Selection. An absent entry contributes nothing, so every lower level is consulted.
+func deploySelection(dc *deploykit.DeployConfig, box, instance string) deploykit.Selection {
+	entry, ok := dc.Lookup(box, instance)
+	if !ok {
+		return deploykit.Selection{}
+	}
+	return deploykit.Selection{
+		Engine:    entry.Engine,
+		Rootful:   entry.Rootful,
+		Namespace: entry.Namespace,
+	}
+}
+
+// resolvePodSelection runs the ONE selection model: the engine word first, then the target facts
+// derived from that word. ResolveContainerdTarget receives the already-resolved engine, so the
+// address/namespace/rootful facts can never disagree with the engine that launches.
+func resolvePodSelection(deploy deploykit.Selection, rt *kit.ResolvedRuntime) (podSelection, error) {
+	engine, err := deploykit.ResolvePodEngine("", deploy, rt)
+	if err != nil {
+		return podSelection{}, err
+	}
+	target, err := deploykit.ResolveContainerdTarget(engine, deploy, rt)
+	if err != nil {
+		return podSelection{}, err
+	}
+	return podSelection{Engine: engine, Target: target}, nil
+}
+
+// resolvePodSelectionFromConfig resolves against an already-loaded overlay.
+func resolvePodSelectionFromConfig(dc *deploykit.DeployConfig, box, instance string, rt *kit.ResolvedRuntime) (podSelection, error) {
+	return resolvePodSelection(deploySelection(dc, box, instance), rt)
+}
+
+// resolvePodSelectionFor loads the per-host overlay once and resolves against it. Callers that
+// already hold the config use resolvePodSelectionFromConfig instead.
+func resolvePodSelectionFor(ctx context.Context, ex *sdk.Executor, box, instance string, rt *kit.ResolvedRuntime) (podSelection, error) {
+	dc, err := loadDeploy(ctx, ex, "resolve pod selection")
+	if err != nil {
+		return podSelection{}, err
+	}
+	return resolvePodSelectionFromConfig(dc, box, instance, rt)
+}
diff --git a/candy/plugin-deploy-pod/selection_test.go b/candy/plugin-deploy-pod/selection_test.go
new file mode 100644
index 0000000..1eac2c5
--- /dev/null
+++ b/candy/plugin-deploy-pod/selection_test.go
@@ -0,0 +1,199 @@
+package deploypod
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/opencharly/sdk"
+	"github.com/opencharly/sdk/deploykit"
+	"github.com/opencharly/sdk/kit"
+	pb "github.com/opencharly/spec/proto"
+	"github.com/opencharly/spec/spec"
+	"google.golang.org/grpc"
+)
+
+// selection_test.go — the W3.1/W3.7 consumer arms. Each migrated call site is pinned through the
+// resolver output it now consumes, and every arm is written so the PRE-CHANGE tree fails it:
+// the pre-change lookup had no vocabulary check, the argv builder read the capability's default
+// namespace, the rootful predicate ignored the deploy entry, and the presence probe ran
+// container.LocalImageExists without a daemon address or namespace.
+
+// selectionForTest resolves a selection the way production does, from a deploy level + runtime.
+func selectionForTest(t *testing.T, deploy deploykit.Selection, runtimeEngine string) podSelection {
+	t.Helper()
+	rt := &kit.ResolvedRuntime{RunEngine: runtimeEngine}
+	sel, err := resolvePodSelection(deploy, rt)
+	if err != nil {
+		t.Fatalf("resolvePodSelection(%+v, %q): %v", deploy, runtimeEngine, err)
+	}
+	return sel
+}
+
+// targetForTest is the resolved target for an engine with no deploy-level overrides.
+func targetForTest(t *testing.T, engine string) deploykit.ContainerdTarget {
+	t.Helper()
+	return selectionForTest(t, deploykit.Selection{}, engine).Target
+}
+
+// TestResolvePodSelection_DeployLevelWinsOverRuntime pins the precedence level the old
+// hand-rolled deploy-entry-else-runtime lookup implemented — now through the ONE resolver.
+func TestResolvePodSelection_DeployLevelWinsOverRuntime(t *testing.T) {
+	sel := selectionForTest(t, deploykit.Selection{Engine: "containerd"}, "podman")
+	if sel.Engine != "containerd" {
+		t.Fatalf("engine = %q, want containerd (the deploy entity outranks the runtime)", sel.Engine)
+	}
+	if sel.Target.Engine != "containerd" {
+		t.Fatalf("target engine = %q, want containerd", sel.Target.Engine)
+	}
+}
+
+// TestResolvePodSelection_RejectsRetiredEngineWord is the control for the silent-wrong-engine
+// class: the pre-change lookup returned the authored word verbatim, and
+// container.EngineCapabilityFor — the table kit.EngineBinary reads — resolves ANY unknown word to
+// the default engine. A retired `nerdctl` deploy therefore ran podman without an error. The
+// resolver rejects it.
+func TestResolvePodSelection_RejectsRetiredEngineWord(t *testing.T) {
+	rt := &kit.ResolvedRuntime{RunEngine: "podman"}
+	if _, err := resolvePodSelection(deploykit.Selection{Engine: "nerdctl"}, rt); err == nil {
+		t.Fatal("a retired engine word must be rejected, not silently resolved to the default engine")
+	}
+	// The control that proves the pre-change path really did fall back silently.
+	if got := kit.EngineBinary("nerdctl"); got != spec.DefaultContainerEngine {
+		t.Fatalf("control: EngineBinary(nerdctl) = %q, want the silent default %q", got, spec.DefaultContainerEngine)
+	}
+}
+
+// TestDeployNamespaceReachesEngineArgv pins the resolved namespace into the argv the engine
+// actually receives. The pre-change engineGlobalArgs read the capability's default ("default"
+// for containerd), so a deploy/settings namespace never reached the CLI.
+func TestDeployNamespaceReachesEngineArgv(t *testing.T) {
+	deploy := deploykit.Selection{Engine: "containerd", Namespace: "edge"}
+	sel := selectionForTest(t, deploy, "podman")
+	if sel.Target.Namespace != "edge" {
+		t.Fatalf("target namespace = %q, want edge", sel.Target.Namespace)
+	}
+	argv := strings.Join(engineStopArgv(sel.Target, "charly-demo"), " ")
+	if !strings.Contains(argv, "--namespace edge") {
+		t.Errorf("stop argv must carry the resolved namespace: %q", argv)
+	}
+	if strings.Contains(argv, "--namespace default") {
+		t.Errorf("stop argv must not carry the capability default when a deploy namespace is set: %q", argv)
+	}
+
+	// podman declares no daemon address/namespace: the argv is unchanged by the migration.
+	podman := selectionForTest(t, deploykit.Selection{Engine: "podman"}, "podman")
+	if got := engineStopArgv(podman.Target, "charly-demo"); strings.Join(got, " ") != "podman stop charly-demo" {
+		t.Errorf("podman stop argv changed: %q", got)
+	}
+}
+
+// TestDeployRootfulReachesTarget pins the deploy-entity rootful level: the pre-change predicate read rt only,
+// so `rootful: sudo` on the entry never reached the launch. The resolver folds it into the
+// target, which is what rootfulUnsupported and the launch argv read.
+func TestDeployRootfulReachesTarget(t *testing.T) {
+	podman := selectionForTest(t, deploykit.Selection{Engine: "podman", Rootful: "sudo"}, "podman")
+	if !podman.Target.Rootful {
+		t.Fatal("a deploy-level rootful mode must reach the resolved target")
+	}
+	if rootfulUnsupported(podman.Target, "quadlet") {
+		t.Error("a rootful podman quadlet target must keep its existing behaviour")
+	}
+
+	containerd := selectionForTest(t, deploykit.Selection{Engine: "containerd", Rootful: "sudo"}, "podman")
+	if !rootfulUnsupported(containerd.Target, "systemd-unit") {
+		t.Error("a rootful containerd systemd-unit target must be refused, not silently run rootless")
+	}
+}
+
+// TestEngineRunArgv_UsesResolvedTarget carries the target through the launch builder too: the
+// containerd plan addresses the resolved namespace and reads its rootful mode from the target.
+func TestEngineRunArgv_UsesResolvedTarget(t *testing.T) {
+	sel := selectionForTest(t, deploykit.Selection{Engine: "containerd", Namespace: "edge"}, "podman")
+	argv, _, err := engineRunArgv(sel.Target, podEngineRun{Image: "img", Name: "charly-demo"})
+	if err != nil {
+		t.Fatalf("engineRunArgv: %v", err)
+	}
+	joined := strings.Join(argv, " ")
+	if !strings.Contains(joined, "--namespace edge") {
+		t.Errorf("launch argv must carry the resolved namespace: %q", joined)
+	}
+	if argv[0] != spec.EngineBinaries["containerd"] {
+		t.Errorf("launch binary = %q, want the containerd capability binary", argv[0])
+	}
+}
+
+// recordExecutorClient is a capture-only pb.ExecutorServiceClient: RunCapture answers from a
+// handler and records every script, so the test sees exactly what the plugin sent to the venue.
+type recordExecutorClient struct {
+	pb.ExecutorServiceClient
+	scripts []string
+	handle  func(script string) *pb.CaptureReply
+	invoked []string
+}
+
+func (c *recordExecutorClient) RunCapture(_ context.Context, in *pb.RunRequest, _ ...grpc.CallOption) (*pb.CaptureReply, error) {
+	c.scripts = append(c.scripts, in.GetScript())
+	return c.handle(in.GetScript()), nil
+}
+
+func (c *recordExecutorClient) InvokeProvider(_ context.Context, in *pb.InvokeProviderRequest, _ ...grpc.CallOption) (*pb.InvokeReply, error) {
+	c.invoked = append(c.invoked, in.GetClass()+":"+in.GetReserved())
+	return &pb.InvokeReply{}, nil
+}
+
+// TestEnsureImagePresent_ProbesResolvedStore pins W3.7's store as the staging boundary: the
+// presence probe is the RESOLVED target's argv (address + namespace), not
+// container.LocalImageExists run without either. The absent arm then reaches build:ensure and
+// re-probes the same store, proving the image landed.
+func TestEnsureImagePresent_ProbesResolvedStore(t *testing.T) {
+	sel := selectionForTest(t, deploykit.Selection{Engine: "containerd", Namespace: "edge"}, "podman")
+
+	t.Run("present in the resolved store", func(t *testing.T) {
+		client := &recordExecutorClient{handle: func(string) *pb.CaptureReply {
+			return &pb.CaptureReply{ExitCode: 0}
+		}}
+		if err := ensureImagePresent(t.Context(), sdk.NewInProcExecutor(client), "myimage:latest", sel.Target); err != nil {
+			t.Fatalf("ensureImagePresent: %v", err)
+		}
+		if len(client.scripts) != 1 {
+			t.Fatalf("probe count = %d, want exactly the target-store probe: %q", len(client.scripts), client.scripts)
+		}
+		probe := client.scripts[0]
+		if !strings.Contains(probe, "--namespace 'edge'") || !strings.Contains(probe, "image inspect") {
+			t.Errorf("presence probe must address the resolved store: %q", probe)
+		}
+	})
+
+	t.Run("absent reaches build:ensure and re-probes", func(t *testing.T) {
+		probes := 0
+		client := &recordExecutorClient{handle: func(script string) *pb.CaptureReply {
+			if strings.Contains(script, "image inspect") {
+				probes++
+				// The pre-build probes (ResolveImage's step 1 and the explicit re-probe
+				// that keeps a daemon fault distinct from a miss) answer absent; the
+				// post-build re-probe answers present.
+				if probes <= 2 {
+					return &pb.CaptureReply{ExitCode: 1}
+				}
+				return &pb.CaptureReply{ExitCode: 0}
+			}
+			// The engine's own daemon probe (`info`) answers, so exit 1 means absent.
+			return &pb.CaptureReply{ExitCode: 0}
+		}}
+		if err := ensureImagePresent(t.Context(), sdk.NewInProcExecutor(client), "myimage:latest", sel.Target); err != nil {
+			t.Fatalf("ensureImagePresent: %v", err)
+		}
+		if len(client.invoked) != 1 || client.invoked[0] != "build:ensure" {
+			t.Fatalf("build:ensure dispatches = %q, want exactly one", client.invoked)
+		}
+		if probes < 2 {
+			t.Fatalf("the store must be re-probed after build:ensure, probes = %d", probes)
+		}
+		for _, script := range client.scripts {
+			if strings.Contains(script, "image inspect") && !strings.Contains(script, "--namespace 'edge'") {
+				t.Errorf("every presence probe must address the resolved namespace: %q", script)
+			}
+		}
+	})
+}
diff --git a/candy/plugin-deploy-pod/systemd_unit.go b/candy/plugin-deploy-pod/systemd_unit.go
index 46503fb..107aafd 100644
--- a/candy/plugin-deploy-pod/systemd_unit.go
+++ b/candy/plugin-deploy-pod/systemd_unit.go
@@ -21,10 +21,10 @@ import (
 // writeSystemdUnitConfig emits the engine's supervision unit and (when configured) the
 // cloudflare tunnel companion unit, then returns the written paths. The caller reloads the
 // systemd user daemon and enables the tunnel exactly as it already does for the quadlet path.
-func writeSystemdUnitConfig(rt *kit.ResolvedRuntime, qcfg deploykit.QuadletConfig, bindMounts []deploykit.ResolvedBindMount, sidecars []deploykit.ResolvedSidecar, tunnelCfg *spec.TunnelConfig) ([]string, error) {
-	engine := rt.RunEngine
-	if rootfulUnsupported(rt) {
-		return nil, rootfulRefusal(engine, rt.RunMode)
+func writeSystemdUnitConfig(rt *kit.ResolvedRuntime, sel podSelection, qcfg deploykit.QuadletConfig, bindMounts []deploykit.ResolvedBindMount, sidecars []deploykit.ResolvedSidecar, tunnelCfg *spec.TunnelConfig) ([]string, error) {
+	engine := string(sel.Engine)
+	if rootfulUnsupported(sel.Target, rt.RunMode) {
+		return nil, rootfulRefusal(sel.Engine, rt.RunMode)
 	}
 	// A bare `KEY` env entry means "inherit this variable from the host"; systemd's
 	// Environment= has no inherit form and EngineUnitRequest carries no PassEnvironment
@@ -43,7 +43,7 @@ func writeSystemdUnitConfig(rt *kit.ResolvedRuntime, qcfg deploykit.QuadletConfi
 
 	name := kit.ContainerNameInstance(qcfg.BoxName, qcfg.Instance)
 	workingDir := deploykit.ResolveWorkingDir(qcfg.Volumes, bindMounts, qcfg.Home, qcfg.BoxName, qcfg.Instance)
-	argv, _, err := engineRunArgv(engine, false, podEngineRun{
+	argv, _, err := engineRunArgv(sel.Target, podEngineRun{
 		Image:      qcfg.ImageRef,
 		Name:       name,
 		Network:    qcfg.Network,
@@ -73,7 +73,7 @@ func writeSystemdUnitConfig(rt *kit.ResolvedRuntime, qcfg deploykit.QuadletConfi
 		Name:        unitName,
 		RunMode:     "systemd-unit",
 		StartArgv:   argv,
-		StopArgv:    engineStopArgv(engine, name),
+		StopArgv:    engineStopArgv(sel.Target, name),
 		Description: "charly deployment " + name,
 		WorkingDir:  workingDir,
 	})
diff --git a/candy/plugin-deploy-pod/volume_root_test.go b/candy/plugin-deploy-pod/volume_root_test.go
index 2589e5a..85dc525 100644
--- a/candy/plugin-deploy-pod/volume_root_test.go
+++ b/candy/plugin-deploy-pod/volume_root_test.go
@@ -174,12 +174,13 @@ func TestVolumeRootKeepIDCoversEveryEmitter(t *testing.T) {
 
 			// Direct mode has TWO container creators, and either may be the one that runs.
 			// Both now build through the engine start_plan body, so both consult engineKeepID.
-			directArgv, err := directRunArgs(qcfg, binds, tc.engine, false)
+			sel := selectionForTest(t, deploykit.Selection{}, tc.engine)
+			directArgv, err := directRunArgs(qcfg, binds, sel)
 			if err != nil {
 				t.Fatalf("directRunArgs(%s): %v", tc.engine, err)
 			}
 			directEmits := hasKeepID(directArgv)
-			startArgv, err := buildStartArgs(tc.engine, false, "img", tc.uid, tc.uid, nil, "ctr", nil, binds,
+			startArgv, err := buildStartArgs(sel, "img", tc.uid, tc.uid, nil, "ctr", nil, binds,
 				false, "", nil, spec.SecurityConfig{}, nil, "/")
 			if err != nil {
 				t.Fatalf("buildStartArgs(%s): %v", tc.engine, err)
@@ -192,7 +193,7 @@ func TestVolumeRootKeepIDCoversEveryEmitter(t *testing.T) {
 
 			// buildShellArgs is the transient `charly run` twin of buildStartArgs and attaches the
 			// same named volumes; it carries the same condition, so it must agree with it.
-			shell := hasKeepID(buildShellArgs(tc.engine, "img", tc.uid, tc.uid, nil, nil, binds,
+			shell := hasKeepID(buildShellArgs(sel.Target, "img", tc.uid, tc.uid, nil, nil, binds,
 				false, "", "", nil, spec.SecurityConfig{}, "/", true))
 			if shell != startEmits {
 				t.Errorf("buildShellArgs emits keep-id = %v but buildStartArgs = %v", shell, startEmits)