// Copyright 2025 Emiliano Spinella (eminwux)
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

package runner

import (
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"

	containerd "github.com/containerd/containerd/v2/client"
	"github.com/eminwux/kukeon/internal/apischeme"
	"github.com/eminwux/kukeon/internal/cgroupcheck"
	"github.com/eminwux/kukeon/internal/cni"
	"github.com/eminwux/kukeon/internal/consts"
	"github.com/eminwux/kukeon/internal/ctr"
	"github.com/eminwux/kukeon/internal/errdefs"
	"github.com/eminwux/kukeon/internal/metadata"
	intmodel "github.com/eminwux/kukeon/internal/modelhub"
	"github.com/eminwux/kukeon/internal/util/fs"
	"github.com/eminwux/kukeon/internal/util/naming"
)

// ensureCgroupParams holds parameters for ensureCgroupInternal.
type ensureCgroupParams struct {
	spec           ctr.CgroupSpec
	docName        string
	cgroupPath     *string
	createErr      error
	updateErr      error
	logLabel       string
	updateMetadata func() error
}

func (r *Exec) provisionNewRealm(realm intmodel.Realm) (intmodel.Realm, error) {
	// Default empty Spec.Namespace to <realm>.kukeon.io so the runner-layer
	// API behaves symmetrically with the controller layer's defaulting
	// (internal/controller/create_realm.go:57-62). Without this, callers
	// that go straight to the runner — notably ReconcileSpace's "ensure
	// parent realm exists" stub at internal/controller/apply/reconcile.go
	// — would create a containerd namespace under the empty string,
	// silently colocating every "naked" realm. Mirror the controller's
	// behavior here so both layers converge on the same safe default.
	if strings.TrimSpace(realm.Spec.Namespace) == "" {
		realm.Spec.Namespace = consts.RealmNamespace(strings.TrimSpace(realm.Metadata.Name))
	}

	// Update realm metadata with Creating state
	if err := r.UpdateRealmMetadata(realm); err != nil {
		return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateRealmMetadata, err)
	}

	// Create realm namespace. A pre-existing containerd namespace is benign on the
	// idempotent kuke init path — kuke-system.kukeon.io is routinely seeded by
	// `ctr -n kuke-system.kukeon.io images import` before init runs. Treat it as
	// a no-op so we still proceed to cgroup creation and the Ready transition,
	// matching the symmetry of ensureRealmContainerdNamespace.
	if err := r.createRealmContainerdNamespace(realm); err != nil &&
		!errors.Is(err, errdefs.ErrNamespaceAlreadyExists) {
		// Set state to Failed and update metadata before returning error
		realm.Status.State = intmodel.RealmStateFailed
		_ = r.UpdateRealmMetadata(realm) // Best effort to save failed state
		return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrCreateRealmNamespace, err)
	}

	// Create realm cgroup
	cgroupPath, subtreeControllers, err := r.createRealmCgroup(realm)
	if err != nil {
		// Set state to Failed and update metadata before returning error
		realm.Status.State = intmodel.RealmStateFailed
		_ = r.UpdateRealmMetadata(realm) // Best effort to save failed state
		return intmodel.Realm{}, err
	}

	// Update CgroupPath, SubtreeControllers (issue #328), and state in internal model
	realm.Status.CgroupPath = cgroupPath
	realm.Status.SubtreeControllers = subtreeControllers
	realm.Status.State = intmodel.RealmStateReady

	// Always update metadata after cgroup creation to ensure CgroupPath is saved
	// (similar to ensureRealmCgroup pattern for consistency)
	if err = r.UpdateRealmMetadata(realm); err != nil {
		return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateRealmMetadata, err)
	}

	return realm, nil
}

func (r *Exec) createRealmContainerdNamespace(realm intmodel.Realm) error {
	// Create realm containerd namespace
	if err := r.ensureClientConnected(); err != nil {
		return fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	exists, err := r.ctrClient.ExistsNamespace(realm.Spec.Namespace)
	if err != nil {
		r.logger.InfoContext(r.ctx, "failed to check if kukeon namespace exists", "err", fmt.Sprintf("%v", err))
		return fmt.Errorf("%w: %w", errdefs.ErrCheckNamespaceExists, err)
	}

	if exists {
		r.logger.InfoContext(r.ctx, "kukeon namespace already exists", "namespace", realm.Spec.Namespace)
		return errdefs.ErrNamespaceAlreadyExists
	}

	fmt.Fprintf(os.Stdout, "Creating containerd namespace '%s'\n", realm.Spec.Namespace)
	err = r.ctrClient.CreateNamespace(realm.Spec.Namespace)
	if err != nil {
		r.logger.InfoContext(r.ctx, "failed to create kukeon namespace", "err", fmt.Sprintf("%v", err))
		return fmt.Errorf("%w: %w", errdefs.ErrCreateNamespace, err)
	}
	r.logger.InfoContext(r.ctx, "created kukeon namespace", "namespace", realm.Spec.Namespace)

	return nil
}

func (r *Exec) ensureRealmContainerdNamespace(realm intmodel.Realm) (intmodel.Realm, error) {
	// Ensure containerd namespace exists for the realm
	if err := r.ensureClientConnected(); err != nil {
		return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	exists, err := r.ctrClient.ExistsNamespace(realm.Spec.Namespace)
	if err != nil {
		return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrCheckNamespaceExists, err)
	}
	if !exists {
		if err = r.ctrClient.CreateNamespace(realm.Spec.Namespace); err != nil {
			return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrCreateNamespace, err)
		}
		r.logger.InfoContext(
			r.ctx,
			"recreated missing containerd namespace for realm",
			"namespace",
			realm.Spec.Namespace,
		)
	}
	return realm, nil
}

func (r *Exec) provisionNewSpace(space intmodel.Space) (intmodel.Space, error) {
	// Update space metadata
	if err := r.UpdateSpaceMetadata(space); err != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateSpaceMetadata, err)
	}

	// Create space network (strict create)
	cniConfigPath, err := r.createSpaceCNIConfig(space)
	if err != nil {
		return intmodel.Space{}, err
	}
	// Update internal model with CNIConfigPath
	space.Spec.CNIConfigPath = cniConfigPath

	// Create space cgroup
	cgroupPath, subtreeControllers, createErr := r.createSpaceCgroup(space)
	if createErr != nil {
		return intmodel.Space{}, createErr
	}

	// Update CgroupPath, SubtreeControllers (issue #328), and state in internal model
	space.Status.CgroupPath = cgroupPath
	space.Status.SubtreeControllers = subtreeControllers
	space.Status.State = intmodel.SpaceStateReady

	// Realize egress policy on the host firewall. iptables rules are safe
	// to install before the bridge exists — they will match nothing until
	// the CNI plugin brings the bridge up for the first container.
	if policyErr := r.applySpaceEgressPolicy(space); policyErr != nil {
		return intmodel.Space{}, policyErr
	}

	// Update space metadata
	if updateErr := r.UpdateSpaceMetadata(space); updateErr != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateSpaceMetadata, updateErr)
	}

	return space, nil
}

func (r *Exec) ensureSpaceCNIConfig(space intmodel.Space) (intmodel.Space, error) {
	// Initialize CNI manager
	mgr, err := cni.NewManager(
		r.cniConf.CniBinDir,
		r.cniConf.CniConfigDir,
		r.cniConf.CniCacheDir,
	)
	if err != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrInitCniManager, err)
	}

	// Convert to external model for other operations (CNIConfigPath handling)
	spaceDoc, err := apischeme.BuildSpaceExternalFromInternal(space, apischeme.VersionV1Beta1)
	if err != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrConversionFailed, err)
	}

	networkName, err := naming.BuildSpaceNetworkName(space.Spec.RealmName, space.Metadata.Name)
	if err != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrCheckNetworkExists, err)
	}
	confPath, err := fs.SpaceNetworkConfigPath(r.opts.RunPath, space.Spec.RealmName, space.Metadata.Name)
	if err != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrCheckNetworkExists, err)
	}

	// Ensure a network exists for this space. Self-heal stale conflists on
	// disk: if the file exists but its bridge name disagrees with
	// SafeBridgeName(networkName), the conflist was written before the
	// IFNAMSIZ fix (commit 6a46bc4) and will fail netlink with ERANGE when
	// the CNI plugin tries to create the bridge. Regenerate in place so
	// fixes to the name generator propagate to hosts with pre-existing
	// state on re-run.
	exists, _, err := mgr.ExistsNetworkConfig(networkName, confPath)
	if err != nil && !errors.Is(err, errdefs.ErrNetworkNotFound) {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrCheckNetworkExists, err)
	}
	regenerate, regenReason, regenErr := r.shouldRegenerateSpaceCNI(mgr, space, networkName, confPath, exists)
	if regenErr != nil {
		return intmodel.Space{}, regenErr
	}
	if regenerate {
		if regenReason != "" {
			r.logger.InfoContext(
				r.ctx,
				"regenerating space network conflist",
				"space", space.Metadata.Name,
				"network", networkName,
				"reason", regenReason,
			)
		}
		subnet, subnetErr := r.subnetForRegenerate(mgr, space, confPath, exists)
		if subnetErr != nil {
			return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrCreateNetwork, subnetErr)
		}
		if writeErr := fs.WriteSpaceNetworkConfig(confPath, networkName, subnet); writeErr != nil {
			return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrCreateNetwork, writeErr)
		}
	}
	specUpdated := spaceDoc.Spec.CNIConfigPath != confPath
	if specUpdated {
		spaceDoc.Spec.CNIConfigPath = confPath
		// Convert back to internal model for UpdateSpaceMetadata
		// Note: CNIConfigPath will be lost in internal model, but UpdateSpaceMetadata
		// converts to external which will have CNIConfigPath from spaceDoc
		// Actually, we need to preserve CNIConfigPath, so we'll convert spaceDoc which has it
		updatedSpace, err := apischeme.ConvertSpaceDocToInternal(spaceDoc)
		if err != nil {
			return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrConversionFailed, err)
		}
		// Preserve any internal-only fields that might have been set
		updatedSpace.Status.CgroupPath = space.Status.CgroupPath
		// This path bypasses UpdateSpaceMetadata (CNIConfigPath would be
		// dropped through the internal-model conversion), so apply the
		// same lifecycle-timestamp invariants here — otherwise UpdatedAt
		// would not advance on this persist. See #419, #418.
		stampSpaceLifecycle(&updatedSpace, r.nowUTC())
		// UpdateSpaceMetadata will convert to external, but CNIConfigPath is not in internal model
		// So we need to ensure it's preserved. Since UpdateSpaceMetadata converts internal->external,
		// and CNIConfigPath is not in internal, we need to handle this differently.
		// Solution: Call UpdateSpaceMetadata with spaceDoc directly converted, but that requires
		// changing UpdateSpaceMetadata signature... Actually, we can build external from the
		// updatedSpace and then manually set CNIConfigPath before saving.
		spaceDocForSave, err := apischeme.BuildSpaceExternalFromInternal(updatedSpace, apischeme.VersionV1Beta1)
		if err != nil {
			return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrConversionFailed, err)
		}
		// Preserve CNIConfigPath from the original spaceDoc
		spaceDocForSave.Spec.CNIConfigPath = spaceDoc.Spec.CNIConfigPath
		// Now save using the external model directly
		metadataFilePath := fs.SpaceMetadataPath(
			r.opts.RunPath,
			spaceDocForSave.Spec.RealmID,
			spaceDocForSave.Metadata.Name,
		)
		if updateErr := metadata.WriteMetadata(r.ctx, r.logger, spaceDocForSave, metadataFilePath); updateErr != nil {
			return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateSpaceMetadata, updateErr)
		}
	}
	r.logger.InfoContext(
		r.ctx,
		"space network exists",
		"space",
		space.Metadata.Name,
		"network",
		networkName,
		"conf",
		confPath,
	)
	// Re-apply egress policy on every Ensure so a daemon restart (or a
	// manual `kuke apply` with a changed policy) converges the host
	// firewall to the spec. iptables rules are idempotent via ensureRule.
	if policyErr := r.applySpaceEgressPolicy(space); policyErr != nil {
		return intmodel.Space{}, policyErr
	}
	return space, nil
}

// subnetForRegenerate picks the subnet to write into a regenerated conflist.
// Order of preference:
//
//  1. An existing allocator state file (post-#131 spaces): use the
//     persisted subnet so re-runs are stable.
//  2. The current conflist's IPAM subnet (legacy spaces created before the
//     allocator landed): preserve it so #131 deliberately leaves
//     pre-existing shared-subnet spaces alone — migration is #133.
//  3. As a last resort (no state file, no readable conflist subnet),
//     allocate a fresh per-space /24.
//
// The legacy-preservation branch never writes allocator state, so a legacy
// space stays legacy until it is recreated post-#131 or migrated by #133.
func (r *Exec) subnetForRegenerate(
	mgr *cni.Manager, space intmodel.Space, confPath string, conflistExists bool,
) (string, error) {
	alloc := r.subnetAllocator
	if persisted, err := alloc.LoadAssigned(space.Spec.RealmName, space.Metadata.Name); err != nil {
		return "", err
	} else if persisted != "" {
		return persisted, nil
	}
	if conflistExists {
		legacy, err := mgr.ReadSubnetCIDR(confPath)
		if err == nil && legacy != "" {
			return legacy, nil
		}
	}
	return alloc.Allocate(space.Spec.RealmName, space.Metadata.Name)
}

// shouldRegenerateSpaceCNI decides whether ensureSpaceCNIConfig needs to
// rewrite the on-disk conflist. Returns (regenerate, humanReason, fatalErr).
// The reason is empty when the decision is the trivial "file missing" case so
// callers can suppress logging on first-time provision.
func (r *Exec) shouldRegenerateSpaceCNI(
	mgr *cni.Manager,
	space intmodel.Space,
	networkName, confPath string,
	exists bool,
) (bool, string, error) {
	if !exists {
		return true, "", nil
	}
	if r.opts.ForceRegenerateCNI {
		return true, "force-regenerate-cni flag set", nil
	}
	onDiskBridge, readErr := mgr.ReadBridgeName(confPath)
	switch {
	case readErr == nil:
		expected := cni.SafeBridgeName(networkName)
		if onDiskBridge != expected {
			return true, fmt.Sprintf(
				"bridge name %q does not match expected %q (stale pre-SafeBridgeName conflist)",
				onDiskBridge, expected,
			), nil
		}
		return false, "", nil
	case errors.Is(readErr, errdefs.ErrBridgePluginMissing):
		return true, "bridge plugin missing from conflist", nil
	case errors.Is(readErr, errdefs.ErrNetworkNotFound):
		// Race: ExistsNetworkConfig saw the file; it's gone now. Regenerate.
		return true, "conflist disappeared after existence check", nil
	default:
		return false, "", fmt.Errorf("%w: %w", errdefs.ErrCheckNetworkExists, readErr)
	}
}

func (r *Exec) createSpaceCNIConfig(space intmodel.Space) (string, error) {
	// Initialize CNI manager
	mgr, err := cni.NewManager(
		r.cniConf.CniBinDir,
		r.cniConf.CniConfigDir,
		r.cniConf.CniCacheDir,
	)
	if err != nil {
		return "", fmt.Errorf("%w: %w", errdefs.ErrInitCniManager, err)
	}

	networkName, err := naming.BuildSpaceNetworkName(space.Spec.RealmName, space.Metadata.Name)
	if err != nil {
		return "", fmt.Errorf("%w: %w", errdefs.ErrCheckNetworkExists, err)
	}
	confPath, err := fs.SpaceNetworkConfigPath(r.opts.RunPath, space.Spec.RealmName, space.Metadata.Name)
	if err != nil {
		return "", fmt.Errorf("%w: %w", errdefs.ErrCheckNetworkExists, err)
	}

	// Check if network already exists
	exists, _, err := mgr.ExistsNetworkConfig(networkName, confPath)
	if err != nil && !errors.Is(err, errdefs.ErrNetworkNotFound) {
		return "", fmt.Errorf("%w: %w", errdefs.ErrCheckNetworkExists, err)
	}
	if exists {
		r.logger.InfoContext(r.ctx, "space network already exists", "network", networkName)
		return "", errdefs.ErrNetworkAlreadyExists
	}

	subnet, allocErr := r.subnetAllocator.Allocate(space.Spec.RealmName, space.Metadata.Name)
	if allocErr != nil {
		return "", fmt.Errorf("%w: %w", errdefs.ErrCreateNetwork, allocErr)
	}

	fmt.Fprintf(os.Stdout, "Creating space network '%s'\n", networkName)
	if writeErr := fs.WriteSpaceNetworkConfig(confPath, networkName, subnet); writeErr != nil {
		r.logger.InfoContext(r.ctx, "failed to create space network", "err", fmt.Sprintf("%v", writeErr))
		return "", fmt.Errorf("%w: %w", errdefs.ErrCreateNetwork, writeErr)
	}
	r.logger.InfoContext(
		r.ctx,
		"created space network",
		"space",
		space.Metadata.Name,
		"network",
		networkName,
		"conf",
		confPath,
		"subnet",
		subnet,
	)
	return confPath, nil
}

// EnsureKukeonRootCgroup ensures the kukeon root cgroup (/kukeon) exists at the
// cgroup hierarchy root. This is the base under which all realms are created.
// It bypasses buildCgroupPath (which would otherwise nest /kukeon under itself)
// and creates the cgroup directly at the discovered mountpoint.
// Returns (existsPre, created, err).
func (r *Exec) EnsureKukeonRootCgroup() (bool, bool, error) {
	if err := r.ensureClientConnected(); err != nil {
		return false, false, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	mountpoint := r.ctrClient.GetCgroupMountpoint()
	group := consts.KukeonCgroupRoot

	if _, err := r.ctrClient.LoadCgroup(group, mountpoint); err == nil {
		r.logger.DebugContext(r.ctx, "kukeon root cgroup already exists", "group", group, "mountpoint", mountpoint)
		return true, false, nil
	}

	spec := ctr.CgroupSpec{Group: group, Mountpoint: mountpoint}
	if _, err := r.ctrClient.NewCgroup(spec); err != nil {
		return false, false, fmt.Errorf("%w: %w", errdefs.ErrCreateRealmCgroup, err)
	}
	r.logger.InfoContext(r.ctx, "created kukeon root cgroup", "group", group, "mountpoint", mountpoint)
	return false, true, nil
}

// buildCgroupPath discovers the cgroup mountpoint and current process cgroup path,
// then combines the spec's Group path relative to the current process's cgroup.
// Returns the updated spec with Group and Mountpoint set, and the full filesystem path.
func (r *Exec) buildCgroupPath(spec ctr.CgroupSpec) (ctr.CgroupSpec, string, error) {
	mountpoint := r.ctrClient.GetCgroupMountpoint()
	currentCgroupPath, err := r.ctrClient.GetCurrentCgroupPath()
	if err != nil {
		return spec, "", fmt.Errorf("failed to get current cgroup path: %w", err)
	}

	// Create the cgroup path relative to the current process's cgroup
	// e.g., if current is /user.slice/... and spec.Group is /kukeon/kuke-system,
	// the full path becomes /user.slice/.../kukeon/kuke-system
	relativeGroup := strings.TrimPrefix(spec.Group, "/")
	// Ensure the cgroup path starts with / for cgroup2
	combinedPath := filepath.Join(currentCgroupPath, relativeGroup)
	if !strings.HasPrefix(combinedPath, "/") {
		combinedPath = "/" + combinedPath
	}
	fullCgroupPath := filepath.Join(mountpoint, strings.TrimPrefix(combinedPath, "/"))
	spec.Group = combinedPath
	spec.Mountpoint = mountpoint

	return spec, fullCgroupPath, nil
}

// createCgroupInternal handles client initialization, connection, and cgroup creation.
// It assumes the spec already has Group and Mountpoint set correctly.
func (r *Exec) createCgroupInternal(spec ctr.CgroupSpec) (string, error) {
	if err := r.ensureClientConnected(); err != nil {
		return "", fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	if _, err := r.ctrClient.NewCgroup(spec); err != nil {
		return "", err
	}

	return spec.Group, nil
}

// ensureCgroupInternal handles the common ensure cgroup logic: checking existence,
// creating if missing, verifying creation, and backfilling metadata if needed.
func (r *Exec) ensureCgroupInternal(params ensureCgroupParams) error {
	// Build the cgroup path
	spec, fullCgroupPath, err := r.buildCgroupPath(params.spec)
	if err != nil {
		return err
	}

	r.logger.DebugContext(
		r.ctx,
		fmt.Sprintf("checking if %s cgroup exists", params.logLabel),
		params.logLabel,
		params.docName,
		"cgroup_path",
		spec.Group,
		"mountpoint",
		spec.Mountpoint,
		"full_path",
		fullCgroupPath,
		"metadata_cgroup_path",
		*params.cgroupPath,
	)

	// Check if cgroup exists
	_, err = r.ctrClient.LoadCgroup(spec.Group, spec.Mountpoint)
	if err != nil {
		r.logger.InfoContext(
			r.ctx,
			fmt.Sprintf("%s cgroup does not exist, attempting to create", params.logLabel),
			params.logLabel,
			params.docName,
			"cgroup_path",
			spec.Group,
			"mountpoint",
			spec.Mountpoint,
			"full_path",
			fullCgroupPath,
			"load_error",
			fmt.Sprintf("%v", err),
		)

		// Attempt to create if missing
		manager, createErr := r.ctrClient.NewCgroup(spec)
		if createErr != nil {
			r.logger.ErrorContext(
				r.ctx,
				fmt.Sprintf("failed to create %s cgroup", params.logLabel),
				params.logLabel,
				params.docName,
				"cgroup_path",
				spec.Group,
				"mountpoint",
				spec.Mountpoint,
				"full_path",
				fullCgroupPath,
				"error",
				fmt.Sprintf("%v", createErr),
			)
			return fmt.Errorf("%w: %w", params.createErr, createErr)
		}
		if manager == nil {
			r.logger.ErrorContext(
				r.ctx,
				"NewCgroup returned nil manager without error",
				params.logLabel,
				params.docName,
				"cgroup_path",
				spec.Group,
			)
			return fmt.Errorf("%w: NewCgroup returned nil manager", params.createErr)
		}

		// Verify the cgroup was actually created
		actualPath := fullCgroupPath
		path, pathErr := r.ctrClient.CgroupPath(spec.Group, spec.Mountpoint)
		if pathErr != nil {
			r.logger.WarnContext(
				r.ctx,
				"failed to get actual cgroup path after creation",
				params.logLabel,
				params.docName,
				"error",
				fmt.Sprintf("%v", pathErr),
			)
		} else {
			actualPath = path
			// Check if the path actually exists
			if _, statErr := os.Stat(actualPath); statErr != nil {
				r.logger.WarnContext(
					r.ctx,
					"cgroup path does not exist on filesystem after creation",
					params.logLabel,
					params.docName,
					"expected_path",
					actualPath,
					"error",
					fmt.Sprintf("%v", statErr),
				)
			}
		}

		r.logger.InfoContext(
			r.ctx,
			fmt.Sprintf("successfully created %s cgroup", params.logLabel),
			params.logLabel,
			params.docName,
			"cgroup_path",
			spec.Group,
			"actual_path",
			actualPath,
			"full_path",
			fullCgroupPath,
		)

		*params.cgroupPath = spec.Group
		if err = params.updateMetadata(); err != nil {
			r.logger.ErrorContext(
				r.ctx,
				fmt.Sprintf("failed to update %s metadata after cgroup creation", params.logLabel),
				params.logLabel,
				params.docName,
				"error",
				fmt.Sprintf("%v", err),
			)
			return fmt.Errorf("%w: %w", params.updateErr, err)
		}
		r.logger.InfoContext(
			r.ctx,
			fmt.Sprintf("recreated missing %s cgroup", params.logLabel),
			params.logLabel,
			params.docName,
			"path",
			spec.Group,
		)
		return nil
	}

	// Cgroup exists
	r.logger.DebugContext(
		r.ctx,
		fmt.Sprintf("%s cgroup exists", params.logLabel),
		params.logLabel,
		params.docName,
		"cgroup_path",
		spec.Group,
	)

	// Backfill metadata if path is empty
	if *params.cgroupPath == "" {
		r.logger.InfoContext(
			r.ctx,
			"cgroup exists but metadata path is empty, backfilling",
			params.logLabel,
			params.docName,
			"cgroup_path",
			spec.Group,
		)
		*params.cgroupPath = spec.Group
		if err = params.updateMetadata(); err != nil {
			return fmt.Errorf("%w: %w", params.updateErr, err)
		}
		r.logger.InfoContext(
			r.ctx,
			fmt.Sprintf("backfilled %s cgroup path", params.logLabel),
			params.logLabel,
			params.docName,
			"path",
			*params.cgroupPath,
		)
	}

	return nil
}

func (r *Exec) ensureSpaceCgroup(space intmodel.Space) (intmodel.Space, error) {
	// Extract realm name from space and validate
	if space.Spec.RealmName == "" {
		return intmodel.Space{}, errdefs.ErrRealmNameRequired
	}

	if err := r.ensureClientConnected(); err != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	spec := ctr.DefaultSpaceSpec(space)

	// Convert to external model for ensureCgroupInternal (requires external model for CgroupPath updates)
	spaceDoc, err := apischeme.BuildSpaceExternalFromInternal(space, apischeme.VersionV1Beta1)
	if err != nil {
		return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrConversionFailed, err)
	}

	// Capture space for closure
	spaceForUpdate := space
	ensureErr := r.ensureCgroupInternal(ensureCgroupParams{
		spec:       spec,
		docName:    spaceDoc.Metadata.Name,
		cgroupPath: &spaceDoc.Status.CgroupPath,
		createErr:  errdefs.ErrCreateSpaceCgroup,
		updateErr:  errdefs.ErrUpdateSpaceMetadata,
		logLabel:   "space",
		updateMetadata: func() error {
			// Update internal model's CgroupPath from external model
			spaceForUpdate.Status.CgroupPath = spaceDoc.Status.CgroupPath
			return r.UpdateSpaceMetadata(spaceForUpdate)
		},
	})
	if ensureErr != nil {
		return intmodel.Space{}, ensureErr
	}

	// Update internal model's CgroupPath from external model after ensureCgroupInternal
	space.Status.CgroupPath = spaceDoc.Status.CgroupPath

	// Always update metadata after ensureCgroupInternal to ensure CgroupPath is saved
	// (closure may not have been called if cgroup already existed with path)
	if spaceDoc.Status.CgroupPath != "" {
		if err := r.UpdateSpaceMetadata(space); err != nil {
			return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateSpaceMetadata, err)
		}
		// Idempotent re-toggle of the space's subtree controllers — see the
		// matching realm ensure-path note for the rationale (issue #327).
		// Persist the effective set so SpaceStatus.SubtreeControllers
		// reflects the live host-filtered set after every ensure-pass
		// (issue #328). On error skip the persist — the warn-and-continue
		// pattern matches the cell ensure path and avoids overwriting a
		// previously-correct value with stale data.
		if subtreeControllers, subtreeErr := r.enableAncestorSubtreeControllers(
			spaceDoc.Status.CgroupPath, r.ctrClient.GetCgroupMountpoint()); subtreeErr != nil {
			r.logger.WarnContext(r.ctx,
				"failed to enable space subtree controllers on ensure-path",
				"space", space.Metadata.Name, "error", subtreeErr)
		} else if !equalStringSlices(space.Status.SubtreeControllers, subtreeControllers) {
			space.Status.SubtreeControllers = subtreeControllers
			if metaErr := r.UpdateSpaceMetadata(space); metaErr != nil {
				return intmodel.Space{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateSpaceMetadata, metaErr)
			}
		}
	}

	return space, nil
}

func (r *Exec) createSpaceCgroup(space intmodel.Space) (string, []string, error) {
	// Extract realm name from space and validate
	if space.Spec.RealmName == "" {
		return "", nil, errdefs.ErrRealmNameRequired
	}

	// Fetch the realm internally for DefaultSpaceSpec
	lookupRealm := intmodel.Realm{
		Metadata: intmodel.RealmMetadata{
			Name: space.Spec.RealmName,
		},
	}
	internalRealm, err := r.GetRealm(lookupRealm)
	if err != nil {
		return "", nil, err
	}

	spec := ctr.DefaultSpaceSpec(space)

	// Ensure client is initialized and connected
	if err = r.ensureClientConnected(); err != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	// Build the cgroup path
	spec, _, err = r.buildCgroupPath(spec)
	if err != nil {
		return "", nil, err
	}

	// Create the cgroup
	cgroupPath, createErr := r.createCgroupInternal(spec)
	if createErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateSpaceCgroup, createErr)
	}

	// Delegate the kukeon resource subset on the space's own subtree_control
	// so an empty space (no descendant stack/cell yet) still carries the
	// controllers — issue #327. The returned effective set bubbles back up
	// so the caller can persist it on SpaceStatus.SubtreeControllers
	// (issue #328).
	subtreeControllers, subtreeErr := r.enableAncestorSubtreeControllers(spec.Group, spec.Mountpoint)
	if subtreeErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateSpaceCgroup, subtreeErr)
	}

	r.logger.InfoContext(
		r.ctx,
		"created space cgroup",
		"space",
		space.Metadata.Name,
		"realm",
		internalRealm.Metadata.Name,
		"path",
		cgroupPath,
	)
	return cgroupPath, subtreeControllers, nil
}

func (r *Exec) createRealmCgroup(realm intmodel.Realm) (string, []string, error) {
	spec := ctr.DefaultRealmSpec(realm)

	// Ensure client is initialized and connected
	if err := r.ensureClientConnected(); err != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	// Build the cgroup path
	var err error
	spec, _, err = r.buildCgroupPath(spec)
	if err != nil {
		return "", nil, err
	}

	// Create the cgroup
	cgroupPath, createErr := r.createCgroupInternal(spec)
	if createErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateRealmCgroup, createErr)
	}

	// Delegate the kukeon resource subset on the realm's own subtree_control
	// so an empty realm (no descendant space/stack/cell yet) still carries
	// the controllers — issue #327. Mirrors the cell-level call site below
	// so each level explicitly populates its own subtree instead of
	// relying on the first cell's ancestor walk to widen everything. The
	// returned effective set bubbles back up so the caller can persist it
	// on RealmStatus.SubtreeControllers (issue #328).
	subtreeControllers, subtreeErr := r.enableAncestorSubtreeControllers(spec.Group, spec.Mountpoint)
	if subtreeErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateRealmCgroup, subtreeErr)
	}

	r.logger.InfoContext(
		r.ctx,
		"created realm cgroup",
		"realm",
		realm.Metadata.Name,
		"path",
		cgroupPath,
	)
	return cgroupPath, subtreeControllers, nil
}

func (r *Exec) ensureRealmCgroup(realm intmodel.Realm) (intmodel.Realm, error) {
	if err := r.ensureClientConnected(); err != nil {
		return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	spec := ctr.DefaultRealmSpec(realm)

	// Convert to external model for ensureCgroupInternal (requires external model for CgroupPath updates)
	realmDoc, err := apischeme.BuildRealmExternalFromInternal(realm, apischeme.VersionV1Beta1)
	if err != nil {
		return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrConversionFailed, err)
	}

	// Capture realm for closure
	realmForUpdate := realm
	ensureErr := r.ensureCgroupInternal(ensureCgroupParams{
		spec:       spec,
		docName:    realmDoc.Metadata.Name,
		cgroupPath: &realmDoc.Status.CgroupPath,
		createErr:  errdefs.ErrCreateRealmCgroup,
		updateErr:  errdefs.ErrUpdateRealmMetadata,
		logLabel:   "realm",
		updateMetadata: func() error {
			// Update internal model's CgroupPath from external model
			realmForUpdate.Status.CgroupPath = realmDoc.Status.CgroupPath
			return r.UpdateRealmMetadata(realmForUpdate)
		},
	})
	if ensureErr != nil {
		return intmodel.Realm{}, ensureErr
	}

	// Update internal model's CgroupPath from external model after ensureCgroupInternal
	// This ensures realm.Status.CgroupPath matches what was set/backfilled in ensureCgroupInternal
	realm.Status.CgroupPath = realmDoc.Status.CgroupPath

	// Always update metadata after ensureCgroupInternal to ensure CgroupPath is saved
	// This is critical because:
	// 1. If cgroup was created, closure already saved it, but we ensure it's saved with current realm
	// 2. If cgroup existed and path was backfilled, closure saved it, but we ensure consistency
	// 3. If cgroup existed with path already set, we ensure metadata matches
	if realmDoc.Status.CgroupPath != "" {
		if err := r.UpdateRealmMetadata(realm); err != nil {
			return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateRealmMetadata, err)
		}
		// Idempotent re-toggle of the realm's subtree controllers on every
		// ensure-pass — covers daemon restart against pre-#327 realms whose
		// subtree_control was only populated as a side effect of the first
		// cell. No-op on already-delegated subtrees. Log + continue on
		// failure to mirror the cell ensure-path pattern. Persist the
		// effective set so RealmStatus.SubtreeControllers reflects the
		// live host-filtered set after every ensure-pass (issue #328).
		if subtreeControllers, subtreeErr := r.enableAncestorSubtreeControllers(
			realmDoc.Status.CgroupPath, r.ctrClient.GetCgroupMountpoint()); subtreeErr != nil {
			r.logger.WarnContext(r.ctx,
				"failed to enable realm subtree controllers on ensure-path",
				"realm", realm.Metadata.Name, "error", subtreeErr)
		} else if !equalStringSlices(realm.Status.SubtreeControllers, subtreeControllers) {
			realm.Status.SubtreeControllers = subtreeControllers
			if metaErr := r.UpdateRealmMetadata(realm); metaErr != nil {
				return intmodel.Realm{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateRealmMetadata, metaErr)
			}
		}
	} else {
		// If cgroup path is still empty after ensureCgroupInternal, log warning
		// This should not happen - ensureCgroupInternal should have created cgroup or backfilled path
		r.logger.WarnContext(
			r.ctx,
			"realm cgroup path is empty after ensureCgroupInternal",
			"realm",
			realm.Metadata.Name,
			"spec_group",
			spec.Group,
		)
	}

	return realm, nil
}

func (r *Exec) provisionNewStack(stack intmodel.Stack) (intmodel.Stack, error) {
	// Update stack metadata
	if err := r.UpdateStackMetadata(stack); err != nil {
		return intmodel.Stack{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateStackMetadata, err)
	}

	// Create stack cgroup
	cgroupPath, subtreeControllers, err := r.createStackCgroup(stack)
	if err != nil {
		return intmodel.Stack{}, err
	}

	// Update CgroupPath, SubtreeControllers (issue #328), and state in internal model
	stack.Status.CgroupPath = cgroupPath
	stack.Status.SubtreeControllers = subtreeControllers
	stack.Status.State = intmodel.StackStateReady

	// Update stack metadata
	if err = r.UpdateStackMetadata(stack); err != nil {
		return intmodel.Stack{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateStackMetadata, err)
	}

	return stack, nil
}

func (r *Exec) createStackCgroup(stack intmodel.Stack) (string, []string, error) {
	// Extract realm and space names from stack and validate
	if stack.Spec.RealmName == "" {
		return "", nil, errdefs.ErrRealmNameRequired
	}
	if stack.Spec.SpaceName == "" {
		return "", nil, errdefs.ErrSpaceNameRequired
	}

	spec := ctr.DefaultStackSpec(stack)

	// Ensure client is initialized and connected
	if err := r.ensureClientConnected(); err != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	// Build the cgroup path
	var err error
	spec, _, err = r.buildCgroupPath(spec)
	if err != nil {
		return "", nil, err
	}

	// Create the cgroup
	cgroupPath, createErr := r.createCgroupInternal(spec)
	if createErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateStackCgroup, createErr)
	}

	// Delegate the kukeon resource subset on the stack's own subtree_control
	// so an empty stack (no descendant cell yet) still carries the
	// controllers — issue #327. The returned effective set bubbles back up
	// so the caller can persist it on StackStatus.SubtreeControllers
	// (issue #328).
	subtreeControllers, subtreeErr := r.enableAncestorSubtreeControllers(spec.Group, spec.Mountpoint)
	if subtreeErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateStackCgroup, subtreeErr)
	}

	r.logger.InfoContext(
		r.ctx,
		"created stack cgroup",
		"stack",
		stack.Metadata.Name,
		"path",
		cgroupPath,
	)
	return cgroupPath, subtreeControllers, nil
}

func (r *Exec) ensureStackCgroup(stack intmodel.Stack) (intmodel.Stack, error) {
	realmName := strings.TrimSpace(stack.Spec.RealmName)
	if realmName == "" {
		return intmodel.Stack{}, errdefs.ErrRealmNameRequired
	}
	spaceName := strings.TrimSpace(stack.Spec.SpaceName)
	if spaceName == "" {
		return intmodel.Stack{}, errdefs.ErrSpaceNameRequired
	}

	stackDoc, err := apischeme.BuildStackExternalFromInternal(stack, apischeme.VersionV1Beta1)
	if err != nil {
		return intmodel.Stack{}, fmt.Errorf("%w: %w", errdefs.ErrConversionFailed, err)
	}

	if err = r.ensureClientConnected(); err != nil {
		return intmodel.Stack{}, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	spec := ctr.DefaultStackSpec(stack)

	// Capture stack for closure
	stackForUpdate := stack
	ensureErr := r.ensureCgroupInternal(ensureCgroupParams{
		spec:       spec,
		docName:    stackDoc.Metadata.Name,
		cgroupPath: &stackDoc.Status.CgroupPath,
		createErr:  errdefs.ErrCreateStackCgroup,
		updateErr:  errdefs.ErrUpdateStackMetadata,
		logLabel:   "stack",
		updateMetadata: func() error {
			// Update internal model's CgroupPath from external model
			stackForUpdate.Status.CgroupPath = stackDoc.Status.CgroupPath
			return r.UpdateStackMetadata(stackForUpdate)
		},
	})
	if ensureErr != nil {
		return intmodel.Stack{}, ensureErr
	}

	// Update internal model's CgroupPath from external model after ensureCgroupInternal
	stack.Status.CgroupPath = stackDoc.Status.CgroupPath

	// Always update metadata after ensureCgroupInternal to ensure CgroupPath is saved
	// (closure may not have been called if cgroup already existed with path)
	if stackDoc.Status.CgroupPath != "" {
		if err := r.UpdateStackMetadata(stack); err != nil {
			return intmodel.Stack{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateStackMetadata, err)
		}
		// Idempotent re-toggle of the stack's subtree controllers — see the
		// matching realm ensure-path note for the rationale (issue #327).
		// Persist the effective set so StackStatus.SubtreeControllers
		// reflects the live host-filtered set after every ensure-pass
		// (issue #328).
		if subtreeControllers, subtreeErr := r.enableAncestorSubtreeControllers(
			stackDoc.Status.CgroupPath, r.ctrClient.GetCgroupMountpoint()); subtreeErr != nil {
			r.logger.WarnContext(r.ctx,
				"failed to enable stack subtree controllers on ensure-path",
				"stack", stack.Metadata.Name, "error", subtreeErr)
		} else if !equalStringSlices(stack.Status.SubtreeControllers, subtreeControllers) {
			stack.Status.SubtreeControllers = subtreeControllers
			if metaErr := r.UpdateStackMetadata(stack); metaErr != nil {
				return intmodel.Stack{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateStackMetadata, metaErr)
			}
		}
	}

	return stack, nil
}

// cellSpaceBridgeName returns the host-side bridge interface name for the
// cell's space. Returns an error if the realm or space is missing on the
// cell spec — callers treat that as best-effort and log/continue.
func cellSpaceBridgeName(cell intmodel.Cell) (string, error) {
	networkName, err := naming.BuildSpaceNetworkName(cell.Spec.RealmName, cell.Spec.SpaceName)
	if err != nil {
		return "", err
	}
	return cni.SafeBridgeName(networkName), nil
}

func (r *Exec) provisionNewCell(cell intmodel.Cell) (intmodel.Cell, error) {
	// The inner closure owns the actual provisioning work. The outer
	// function captures provisionStarted via closure and, on any error
	// past the stub write, flips the on-disk record to Failed with a
	// Reason and Message so `kuke get cell -o yaml` shows the operator
	// the failing stage instead of a derived Stopped (issue #504).
	// A pre-stub UpdateCellMetadata failure leaves no metadata file to
	// stamp Failed onto, so provisionStarted stays false and the helper
	// is skipped. Closure pattern (instead of a named retErr + defer
	// like StartCell / StartContainer) keeps the new code lint-clean
	// against nonamedreturns.
	var provisionStarted bool
	out, err := func() (intmodel.Cell, error) {
		// Update cell metadata
		if err := r.UpdateCellMetadata(cell); err != nil {
			return intmodel.Cell{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateCellMetadata, err)
		}

		// Past the stub write: any subsequent error must flip the
		// on-disk record to Failed.
		provisionStarted = true

		// Create cell cgroup
		cgroupPath, subtreeControllers, err := r.createCellCgroup(cell)
		if err != nil {
			return intmodel.Cell{}, err
		}

		// Update internal model with cgroup path and effective subtree
		// controllers (issue #328 — surfaces the result of
		// enableCellControllers so operators can confirm the cell-level
		// delegation from `kuke get cells -o yaml`).
		cell.Status.CgroupPath = cgroupPath
		cell.Status.SubtreeControllers = subtreeControllers

		// Persist the host-side bridge this cell's space lands on.
		// Computing it here (instead of the read path) keeps the cell
		// self-describing for debug tooling and survives daemon
		// restarts because the cell metadata file is rewritten in
		// UpdateCellMetadata below.
		if bridge, brErr := cellSpaceBridgeName(cell); brErr == nil {
			cell.Status.Network.BridgeName = bridge
		} else {
			// Bridge name is best-effort metadata; failing to derive
			// it must not block provisioning (the cell can still come
			// up — the iface is created by CNI from the conflist
			// regardless of what we record here).
			r.logger.WarnContext(r.ctx, "could not derive cell bridge name",
				"cell", cell.Metadata.Name, "error", brErr)
		}

		// Create pause container and all containers for the cell
		if _, err = r.createCellContainers(&cell); err != nil {
			return intmodel.Cell{}, err
		}

		markCellReady(&cell)

		// Update cell metadata
		if err = r.UpdateCellMetadata(cell); err != nil {
			return intmodel.Cell{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateCellMetadata, err)
		}

		return cell, nil
	}()
	if err != nil && provisionStarted {
		r.markCellFailed(cell, "CreateCellFailed", err)
	}
	return out, err
}

func (r *Exec) createCellCgroup(cell intmodel.Cell) (string, []string, error) {
	// Extract realm, space, and stack names from cell and validate
	if cell.Spec.RealmName == "" {
		return "", nil, errdefs.ErrRealmNameRequired
	}
	if cell.Spec.SpaceName == "" {
		return "", nil, errdefs.ErrSpaceNameRequired
	}
	if cell.Spec.StackName == "" {
		return "", nil, errdefs.ErrStackNameRequired
	}

	spec := ctr.DefaultCellSpec(cell)

	// Ensure client is initialized and connected
	if err := r.ensureClientConnected(); err != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	// Build the cgroup path
	var err error
	spec, _, err = r.buildCgroupPath(spec)
	if err != nil {
		return "", nil, err
	}

	// Create the cgroup
	cgroupPath, createErr := r.createCgroupInternal(spec)
	if createErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateCellCgroup, createErr)
	}

	// Enable the cell-level resource controllers in the cgroup tree before
	// any container task lands inside the cell. Without this, container
	// task cgroups nested under <cell>/<containerd-id> have no controllers
	// available and cell-level UpdateCgroup limits are decorative — exactly
	// what issue #312 reports. Cells with NestedCgroupRuntime opt in to
	// delegating the full host-available controller set instead of the
	// resource subset (issue #314). The returned effective set bubbles
	// back up so the caller can persist it on
	// CellStatus.SubtreeControllers (issue #328).
	subtreeControllers, subtreeErr := r.enableCellControllers(cell, spec.Group, spec.Mountpoint)
	if subtreeErr != nil {
		return "", nil, fmt.Errorf("%w: %w", errdefs.ErrCreateCellCgroup, subtreeErr)
	}

	r.logger.InfoContext(
		r.ctx,
		"created cell cgroup",
		"cell",
		cell.Metadata.Name,
		"path",
		cgroupPath,
	)
	return cgroupPath, subtreeControllers, nil
}

// enableCellControllers picks between the resource-subset and full-set
// subtree-control delegation based on cell.Spec.NestedCgroupRuntime, and
// invokes the matching ctr.Client entry point. The two paths share the
// same underlying writeSubtreeEnable + ToggleControllers logic; the only
// difference is which controller set is enabled. Issue #314.
//
// The resource subset comes from cgroupcheck.CellResourceControllers so
// the doctor pre-flight (`kuke doctor cgroups`) and the cell-creation
// path stay in lockstep — issue #324.
//
// Returns the effective controller set actually written (filtered against
// the host root) so the caller can persist it on
// CellStatus.SubtreeControllers (issue #328).
func (r *Exec) enableCellControllers(cell intmodel.Cell, group, mountpoint string) ([]string, error) {
	if cell.Spec.NestedCgroupRuntime {
		return r.ctrClient.EnableCellAllSubtreeControllers(group, mountpoint)
	}
	return r.ctrClient.EnableCellSubtreeControllers(group, mountpoint, cgroupcheck.CellResourceControllers())
}

// enableAncestorSubtreeControllers delegates the kukeon resource subset on
// a non-cell level (realm, space, stack) so the level's subtree_control is
// populated independently of the first cell that lands under it. Without
// this, an empty realm/space/stack carries no controllers — the cell-level
// ToggleControllers walk only widens ancestors when a cell is created
// (issue #327). Returns the effective controller set actually written so
// the caller can persist it on the level's Status.SubtreeControllers
// (issue #328); returning the underlying error lets callers in create-
// paths fail hard while ensure-pass call sites can choose to log + continue
// (matching the cell ensure-path pattern).
func (r *Exec) enableAncestorSubtreeControllers(group, mountpoint string) ([]string, error) {
	return r.ctrClient.EnsureSubtreeControllers(group, mountpoint, cgroupcheck.CellResourceControllers())
}

// equalStringSlices reports whether two []string values carry the same
// elements in the same order. Used by the ensure-paths to skip a metadata
// rewrite when SubtreeControllers is unchanged (issue #328 — the kernel
// re-toggle is already idempotent; the metadata save costs an fsync, so
// short-circuit on equality keeps the steady-state ensure pass cheap).
func equalStringSlices(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func (r *Exec) ensureCellCgroup(cell intmodel.Cell) (intmodel.Cell, error) {
	realmName := strings.TrimSpace(cell.Spec.RealmName)
	if realmName == "" {
		return intmodel.Cell{}, errdefs.ErrRealmNameRequired
	}
	spaceName := strings.TrimSpace(cell.Spec.SpaceName)
	if spaceName == "" {
		return intmodel.Cell{}, errdefs.ErrSpaceNameRequired
	}
	stackName := strings.TrimSpace(cell.Spec.StackName)
	if stackName == "" {
		return intmodel.Cell{}, errdefs.ErrStackNameRequired
	}

	cellDoc, err := apischeme.BuildCellExternalFromInternal(cell, apischeme.VersionV1Beta1)
	if err != nil {
		return intmodel.Cell{}, fmt.Errorf("%w: %w", errdefs.ErrConversionFailed, err)
	}

	if err = r.ensureClientConnected(); err != nil {
		return intmodel.Cell{}, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	spec := ctr.DefaultCellSpec(cell)

	// Capture cell for closure
	cellForUpdate := cell
	ensureErr := r.ensureCgroupInternal(ensureCgroupParams{
		spec:       spec,
		docName:    cellDoc.Metadata.Name,
		cgroupPath: &cellDoc.Status.CgroupPath,
		createErr:  errdefs.ErrCreateStackCgroup,
		updateErr:  errdefs.ErrUpdateStackMetadata,
		logLabel:   "stack",
		updateMetadata: func() error {
			// Update internal model's CgroupPath from external model
			cellForUpdate.Status.CgroupPath = cellDoc.Status.CgroupPath
			return r.UpdateCellMetadata(cellForUpdate)
		},
	})
	if ensureErr != nil {
		return intmodel.Cell{}, ensureErr
	}

	// Update internal model's CgroupPath from external model after ensureCgroupInternal
	cell.Status.CgroupPath = cellDoc.Status.CgroupPath

	// Always update metadata after ensureCgroupInternal to ensure CgroupPath is saved
	// (closure may not have been called if cgroup already existed with path)
	if cellDoc.Status.CgroupPath != "" {
		if err := r.UpdateCellMetadata(cell); err != nil {
			return intmodel.Cell{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateCellMetadata, err)
		}
		// Idempotent re-toggle of cell subtree controllers after every
		// ensure-pass — covers daemon restart against pre-existing cells
		// provisioned before issue #312 was fixed (and is a no-op on
		// already-enabled subtrees). Cells whose persisted spec carries
		// NestedCgroupRuntime=true take the full-set delegation path so a
		// daemon restart re-asserts the same controller set the cell was
		// provisioned with (issue #314). Persist the effective set so
		// CellStatus.SubtreeControllers reflects the live host-filtered set
		// after every ensure-pass (issue #328).
		if subtreeControllers, subtreeErr := r.enableCellControllers(
			cell, cellDoc.Status.CgroupPath, r.ctrClient.GetCgroupMountpoint()); subtreeErr != nil {
			r.logger.WarnContext(r.ctx,
				"failed to enable cell subtree controllers on ensure-path",
				"cell", cell.Metadata.Name, "error", subtreeErr)
		} else if !equalStringSlices(cell.Status.SubtreeControllers, subtreeControllers) {
			cell.Status.SubtreeControllers = subtreeControllers
			if metaErr := r.UpdateCellMetadata(cell); metaErr != nil {
				return intmodel.Cell{}, fmt.Errorf("%w: %w", errdefs.ErrUpdateCellMetadata, metaErr)
			}
		}
	}

	return cell, nil
}

// createCellContainers creates all containers defined in the Cell, including the root container.
// The root container is treated uniformly with other containers in the loop, but uses a different
// creation method (BuildRootContainerSpec + CreateContainer vs CreateContainerFromSpec).
// Modifies the cell in place: ensures root container is in Containers array and sets RootContainerID.
func (r *Exec) createCellContainers(cell *intmodel.Cell) (containerd.Container, error) {
	cellName := strings.TrimSpace(cell.Metadata.Name)
	if cellName == "" {
		return nil, errdefs.ErrCellNameRequired
	}

	cellID := cell.Spec.ID
	if cellID == "" {
		return nil, errdefs.ErrCellIDRequired
	}

	realmName := strings.TrimSpace(cell.Spec.RealmName)
	if realmName == "" {
		return nil, errdefs.ErrRealmNameRequired
	}

	spaceName := strings.TrimSpace(cell.Spec.SpaceName)
	if spaceName == "" {
		return nil, errdefs.ErrSpaceNameRequired
	}

	stackName := strings.TrimSpace(cell.Spec.StackName)
	if stackName == "" {
		return nil, errdefs.ErrStackNameRequired
	}

	if err := r.applyCellContainerDefaults(cell); err != nil {
		return nil, err
	}

	cniConfigPath, cniErr := r.ResolveSpaceCNIConfigPath(realmName, spaceName)
	if cniErr != nil {
		return nil, fmt.Errorf("failed to resolve space CNI config: %w", cniErr)
	}

	if err := r.ensureClientConnected(); err != nil {
		return nil, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	// Get realm to access namespace
	lookupRealm := intmodel.Realm{
		Metadata: intmodel.RealmMetadata{
			Name: realmName,
		},
	}
	internalRealm, err := r.GetRealm(lookupRealm)
	if err != nil {
		return nil, fmt.Errorf("failed to get realm: %w", err)
	}

	namespace := internalRealm.Spec.Namespace
	if namespace == "" {
		return nil, fmt.Errorf("realm %q has no namespace", realmName)
	}

	creds := ctr.ConvertRealmCredentials(internalRealm.Spec.RegistryCredentials)

	// Prepare root container: ensure it's in Containers array and RootContainerID is set
	rootContainerdID, err := naming.BuildRootContainerdID(spaceName, stackName, cellID)
	if err != nil {
		return nil, fmt.Errorf("failed to build root container containerd ID: %w", err)
	}

	rootContainerSpec, err := r.ensureCellRootContainerSpec(*cell)
	if err != nil {
		return nil, err
	}

	// Ensure root container base name is set (use "root" if empty)
	if rootContainerSpec.ID == "" {
		rootContainerSpec.ID = "root"
	}

	// Set containerd ID (hierarchical format)
	rootContainerSpec.ContainerdID = rootContainerdID

	// Determine root container ID for RootContainerID field (use base name from ID field)
	rootContainerBaseID := rootContainerSpec.ID

	// Ensure root container is in Containers array
	rootContainerInArray := false
	for i, container := range cell.Spec.Containers {
		if container.ID == rootContainerBaseID {
			// Update existing entry to ensure it has Root flag and correct fields
			cell.Spec.Containers[i] = rootContainerSpec
			rootContainerInArray = true
			break
		}
	}

	// Add root container to Containers array if not present
	if !rootContainerInArray {
		cell.Spec.Containers = append(cell.Spec.Containers, rootContainerSpec)
	}

	// Set RootContainerID if not already set (use base name, not hierarchical ID)
	if cell.Spec.RootContainerID == "" {
		cell.Spec.RootContainerID = rootContainerBaseID
	}

	// Per-cell /etc/hostname and initial /etc/hosts (no IP yet — CNI ADD runs
	// during StartCell and rewrites /etc/hosts with the assigned cell IP).
	// Files must exist before runc consumes the OCI bind-mount entries below,
	// so render them up front. Issue #345.
	if err := r.renderCellEtcFilesPreCNI(cell); err != nil {
		return nil, fmt.Errorf("render cell etc files: %w", err)
	}
	r.stampCellEtcFilePathsOnContainers(cell)

	// Track root container for return value
	var rootContainer containerd.Container

	// Resolve the attachable container so the non-root OCI build below can
	// merge `kuke run --env` runtime entries into the right container's env
	// (issue #834). Empty string means "no attachable container declared" —
	// mergeRuntimeEnvForContainer treats that as a no-op so unrelated
	// containers keep their authored env verbatim.
	attachableID := resolveAttachableContainerID(*cell)

	// Create all containers defined in the cell (including root container)
	for i := range cell.Spec.Containers {
		containerSpec := cell.Spec.Containers[i]

		// Determine if this is the root container
		isRoot := cell.Spec.RootContainerID != "" && containerSpec.ID == cell.Spec.RootContainerID

		// Ensure container spec has required names
		if containerSpec.CellName == "" {
			containerSpec.CellName = cellID
		}
		if containerSpec.SpaceName == "" {
			containerSpec.SpaceName = spaceName
		}
		if containerSpec.RealmName == "" {
			containerSpec.RealmName = realmName
		}
		if containerSpec.StackName == "" {
			containerSpec.StackName = stackName
		}
		if containerSpec.CNIConfigPath == "" {
			containerSpec.CNIConfigPath = cniConfigPath
		}
		// Cell-rooted cgroup placement for every container task. cell.Status.CgroupPath
		// is populated by createCellCgroup before this runs (and preserved through
		// recreate paths via desired.Status.CgroupPath = existing.Status.CgroupPath),
		// so it's safe to read here. See issue #312.
		setCellCgroupPath(&containerSpec, cell)
		cell.Spec.Containers[i] = containerSpec

		var createdContainer containerd.Container
		var createErr error

		if isRoot {
			// Root container: use BuildRootContainerSpec + CreateContainer
			// Use ContainerdID if available, otherwise build it
			containerdID := containerSpec.ContainerdID
			if containerdID == "" {
				var buildErr error
				containerdID, buildErr = naming.BuildRootContainerdID(spaceName, stackName, cellID)
				if buildErr != nil {
					return nil, fmt.Errorf("failed to build root container containerd ID: %w", buildErr)
				}
				containerSpec.ContainerdID = containerdID
				cell.Spec.Containers[i] = containerSpec
			}

			rootLabels := stampSpecHashOnLabels(buildRootContainerLabels(*cell), containerSpec)
			ctrContainerSpec := ctr.BuildRootContainerSpec(containerSpec, rootLabels, r.daemonDefaultBuildOpts()...)

			createdContainer, createErr = r.ctrClient.CreateContainer(namespace, ctrContainerSpec, creds)
			if createErr != nil {
				logFields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
				logFields = append(
					logFields,
					"space",
					spaceName,
					"realm",
					realmName,
					"cniConfig",
					cniConfigPath,
					"err",
					fmt.Sprintf("%v", createErr),
				)
				r.logger.ErrorContext(
					r.ctx,
					"failed to create root container",
					logFields...,
				)
				return nil, fmt.Errorf("%w: %w", errdefs.ErrCreateRootContainer, createErr)
			}

			rootContainer = createdContainer

			infoFields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
			infoFields = append(infoFields, "space", spaceName, "realm", realmName, "cniConfig", cniConfigPath)
			r.logger.InfoContext(
				r.ctx,
				"created root container",
				infoFields...,
			)
		} else {
			// Regular container: use CreateContainerFromSpec
			// Build containerd ID using hierarchical format for containerd operations
			// Store it in ContainerdID field, keep base name in ID field
			containerdID := containerSpec.ContainerdID
			if containerdID == "" {
				var buildErr error
				containerdID, buildErr = naming.BuildContainerdID(
					containerSpec.SpaceName,
					containerSpec.StackName,
					containerSpec.CellName,
					containerSpec.ID,
				)
				if buildErr != nil {
					return nil, fmt.Errorf("failed to build container containerd ID: %w", buildErr)
				}
				containerSpec.ContainerdID = containerdID
				cell.Spec.Containers[i] = containerSpec
			}

			// priorStages threads the controller-side ContainerStatus.Stages
			// snapshot into the renderer so the phase-C2 (#737) gate can omit
			// already-done runOn: create stages from the rendered ContainerDoc
			// on this boot.
			priorStages := priorStagesForContainer(*cell, containerSpec.ID)
			attachOpts, attachErr := r.attachableBuildOpts(namespace, containerSpec, creds, priorStages)
			if attachErr != nil {
				return nil, fmt.Errorf("failed to prepare attachable container %s: %w", containerdID, attachErr)
			}
			buildOpts := append(r.daemonDefaultBuildOpts(), attachOpts...)
			buildOpts = append(buildOpts, ctr.WithExtraLabels(specHashLabels(containerSpec)))
			// Merge `kuke run --env` runtime env into the attachable
			// container's spec env (issue #834). Returns containerSpec
			// unchanged for non-attachable containers or when
			// cell.Spec.RuntimeEnv is empty; the merged copy never writes
			// back to cell.Spec.Containers[i] (the OCI build sees the
			// runtime entries, on-disk metadata stays clean).
			ociSpec := mergeRuntimeEnvForContainer(containerSpec, attachableID, cell.Spec.RuntimeEnv)
			_, createErr = r.ctrClient.CreateContainerFromSpec(
				namespace,
				ociSpec,
				creds,
				buildOpts...,
			)
			if createErr != nil {
				fields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
				fields = append(
					fields,
					"space",
					spaceName,
					"realm",
					realmName,
					"cniConfig",
					containerSpec.CNIConfigPath,
					"err",
					fmt.Sprintf("%v", createErr),
				)
				r.logger.ErrorContext(
					r.ctx,
					"failed to create container from cell",
					fields...,
				)
				return nil, fmt.Errorf("failed to create container %s: %w", containerdID, createErr)
			}
			if chownErr := r.attachablePostCreateChown(namespace, containerSpec); chownErr != nil {
				return nil, fmt.Errorf(
					"failed to chown attachable tty dir for %s: %w", containerdID, chownErr,
				)
			}
		}
	}

	return rootContainer, nil
}

// applyCellContainerDefaults loads the parent Space for the cell and merges
// its spec.defaults.container into every container in the cell. The merge
// runs in place and is idempotent — containers that already carry the
// defaults are unchanged. This is the single entry point that every
// container-creating flow (CreateCell, UpdateCell, CreateContainer,
// UpdateContainer) funnels through.
func (r *Exec) applyCellContainerDefaults(cell *intmodel.Cell) error {
	spaceLookup := intmodel.Space{
		Metadata: intmodel.SpaceMetadata{Name: cell.Spec.SpaceName},
		Spec:     intmodel.SpaceSpec{RealmName: cell.Spec.RealmName},
	}
	parentSpace, err := r.GetSpace(spaceLookup)
	if err != nil {
		return fmt.Errorf("%w: %w", errdefs.ErrGetSpace, err)
	}
	for i := range cell.Spec.Containers {
		intmodel.ApplySpaceDefaultsToContainer(parentSpace, &cell.Spec.Containers[i])
		// Propagate the cell-level NestedCgroupRuntime opt-in (#314) to
		// every container so BuildContainerSpec can pair the private
		// cgroup-ns with a /sys/fs/cgroup mount over the delegated
		// subtree (#322). The flag is a per-cell decision because the
		// host-side subtree-controller delegation in
		// EnableCellAllSubtreeControllers is itself per-cell — a
		// container cannot opt out individually without breaking the
		// inner runtime that #318 enabled.
		cell.Spec.Containers[i].NestedCgroupRuntime = cell.Spec.NestedCgroupRuntime
	}
	return nil
}

// ensureCellContainers ensures the root container and all containers defined in the CellDoc exist.
// The root container is ensured first, then all containers in doc.Spec.Containers are ensured.
// If any container doesn't exist, it is created. Returns the root container or an error.
func (r *Exec) ensureCellContainers(cell *intmodel.Cell) (containerd.Container, error) {
	cellName := strings.TrimSpace(cell.Metadata.Name)
	if cellName == "" {
		return nil, errdefs.ErrCellNotFound
	}

	cellID := cell.Spec.ID
	if cellID == "" {
		return nil, errdefs.ErrCellIDRequired
	}

	realmName := cell.Spec.RealmName
	if realmName == "" {
		return nil, errdefs.ErrRealmNameRequired
	}

	spaceName := cell.Spec.SpaceName
	if spaceName == "" {
		return nil, errdefs.ErrSpaceNameRequired
	}

	stackName := cell.Spec.StackName
	if stackName == "" {
		return nil, errdefs.ErrStackNameRequired
	}

	if err := r.applyCellContainerDefaults(cell); err != nil {
		return nil, err
	}

	cniConfigPath, cniErr := r.ResolveSpaceCNIConfigPath(realmName, spaceName)
	if cniErr != nil {
		return nil, fmt.Errorf("failed to resolve space CNI config: %w", cniErr)
	}

	if err := r.ensureClientConnected(); err != nil {
		return nil, fmt.Errorf("%w: %w", errdefs.ErrConnectContainerd, err)
	}

	// Get realm to access namespace
	lookupRealm := intmodel.Realm{
		Metadata: intmodel.RealmMetadata{
			Name: realmName,
		},
	}
	internalRealm, err := r.GetRealm(lookupRealm)
	if err != nil {
		return nil, fmt.Errorf("failed to get realm: %w", err)
	}

	creds := ctr.ConvertRealmCredentials(internalRealm.Spec.RegistryCredentials)

	// Generate containerd ID with cell identifier for uniqueness
	containerID, err := naming.BuildRootContainerdID(spaceName, stackName, cellID)
	if err != nil {
		return nil, fmt.Errorf("failed to build root container containerd ID: %w", err)
	}

	// Declare container variable to be used in both branches
	var container containerd.Container

	// Check if container exists
	exists, err := r.ExistsContainer(internalRealm.Spec.Namespace, containerID)
	if err != nil {
		fields := appendCellLogFields([]any{"id", containerID}, cellID, cellName)
		fields = append(fields, "space", spaceName, "realm", realmName, "err", fmt.Sprintf("%v", err))
		r.logger.ErrorContext(
			r.ctx,
			"failed to check if root container exists",
			fields...,
		)
		return nil, fmt.Errorf("failed to check if root container exists: %w", err)
	}

	if exists {
		// Container exists, load it but continue to process other containers
		var loadErr error
		container, loadErr = r.ctrClient.GetContainer(internalRealm.Spec.Namespace, containerID)
		if loadErr != nil {
			fields := appendCellLogFields([]any{"id", containerID}, cellID, cellName)
			fields = append(fields, "space", spaceName, "realm", realmName, "err", fmt.Sprintf("%v", loadErr))
			r.logger.WarnContext(
				r.ctx,
				"root container exists but failed to load",
				fields...,
			)
			return nil, fmt.Errorf("failed to load existing root container: %w", loadErr)
		}
		fields := appendCellLogFields([]any{"id", containerID}, cellID, cellName)
		fields = append(fields, "space", spaceName, "realm", realmName)
		r.logger.DebugContext(
			r.ctx,
			"root container exists",
			fields...,
		)
		// Don't return early - continue to process other containers in the cell
	} else {
		// Container doesn't exist, create it
		createFields := appendCellLogFields([]any{"id", containerID}, cellID, cellName)
		createFields = append(createFields, "space", spaceName, "realm", realmName, "cniConfig", cniConfigPath)
		r.logger.InfoContext(
			r.ctx,
			"root container does not exist, creating",
			createFields...,
		)

		var rootContainerSpec intmodel.ContainerSpec
		rootContainerSpec, err = r.ensureCellRootContainerSpec(*cell)
		if err != nil {
			return nil, err
		}

		// Cell-rooted cgroup placement for the root container task (issue
		// #312). Mirrors the wiring in createCellContainers and the regular-
		// container loop below.
		setCellCgroupPath(&rootContainerSpec, cell)

		// Determine root container ID for RootContainerID field (use base name from ID field)
		rootContainerBaseID := rootContainerSpec.ID

		// Check if root container is already in Containers array
		rootContainerInArray := false
		for i, container := range cell.Spec.Containers {
			if container.ID == rootContainerBaseID {
				// Update existing entry to ensure it has Root flag and correct fields
				cell.Spec.Containers[i] = rootContainerSpec
				rootContainerInArray = true
				break
			}
		}

		// Add root container to Containers array if not present
		if !rootContainerInArray {
			cell.Spec.Containers = append(cell.Spec.Containers, rootContainerSpec)
		}

		// Set RootContainerID if not already set (use base name, not hierarchical ID)
		if cell.Spec.RootContainerID == "" {
			cell.Spec.RootContainerID = rootContainerBaseID
		}

		// Per-cell /etc/hostname and initial /etc/hosts (issue #345). Render
		// before BuildRootContainerSpec so the bind-mount sources exist when
		// runc creates the root container, and stamp the source paths onto
		// rootContainerSpec so the OCI spec emits the bind-mount entries.
		// The cell-wide stamp (after the if/else below) reaches non-root
		// containers; the root needs the per-spec stamp because the local
		// rootContainerSpec value is what feeds BuildRootContainerSpec here.
		if etcErr := r.renderCellEtcFilesPreCNI(cell); etcErr != nil {
			return nil, fmt.Errorf("render cell etc files: %w", etcErr)
		}
		r.stampContainerRecreateRuntimeFields(&rootContainerSpec, cell)

		rootLabels := stampSpecHashOnLabels(buildRootContainerLabels(*cell), rootContainerSpec)
		containerSpec := ctr.BuildRootContainerSpec(rootContainerSpec, rootLabels, r.daemonDefaultBuildOpts()...)

		var createErr error
		container, createErr = r.ctrClient.CreateContainer(internalRealm.Spec.Namespace, containerSpec, creds)
		if createErr != nil {
			fields := appendCellLogFields([]any{"id", containerID}, cellID, cellName)
			fields = append(
				fields,
				"space",
				spaceName,
				"realm",
				realmName,
				"cniConfig",
				cniConfigPath,
				"err",
				fmt.Sprintf("%v", createErr),
			)
			r.logger.ErrorContext(
				r.ctx,
				"failed to create root container",
				fields...,
			)
			return nil, fmt.Errorf("%w: %w", errdefs.ErrCreateRootContainer, createErr)
		}

		ensuredFields := appendCellLogFields([]any{"id", containerID}, cellID, cellName)
		ensuredFields = append(ensuredFields, "space", spaceName, "realm", realmName, "cniConfig", cniConfigPath)
		r.logger.InfoContext(
			r.ctx,
			"ensured root container exists",
			ensuredFields...,
		)
	}

	// Per-cell /etc/hostname / /etc/hosts: ensure the source files exist and
	// every non-root container in cell.Spec.Containers carries the bind-mount
	// path before the loop below builds OCI specs. On the existing-root
	// branch the cell IP may already be rendered into /etc/hosts by a prior
	// StartCell; use the existence-only helper so update_cell /
	// update_container ticks don't truncate the IP line back to localhost-
	// only. The no-root branch's pre-CNI render above remains the source of
	// content for fresh cells. Issue #345.
	if etcErr := r.ensureCellEtcFilesExistPreCNI(cell); etcErr != nil {
		return nil, fmt.Errorf("ensure cell etc files: %w", etcErr)
	}
	r.stampCellEtcFilePathsOnContainers(cell)

	// Log how many containers we're about to process
	containerCountFields := appendCellLogFields([]any{"cell", cellID}, cellID, cellName)
	containerCountFields = append(
		containerCountFields,
		"space",
		spaceName,
		"realm",
		realmName,
		"stack",
		stackName,
		"containerCount",
		len(cell.Spec.Containers),
	)
	r.logger.DebugContext(
		r.ctx,
		"processing containers from cell",
		containerCountFields...,
	)

	// Ensure all containers defined in the cell exist
	for i := range cell.Spec.Containers {
		containerSpec := cell.Spec.Containers[i]

		// Log which container we're processing
		processFields := appendCellLogFields([]any{"cell", cellID}, cellID, cellName)
		processFields = append(
			processFields,
			"space",
			spaceName,
			"realm",
			realmName,
			"stack",
			stackName,
			"containerName",
			containerSpec.ID,
			"containerIndex",
			i,
		)
		r.logger.DebugContext(
			r.ctx,
			"processing container from cell",
			processFields...,
		)

		if containerSpec.CNIConfigPath == "" {
			containerSpec.CNIConfigPath = cniConfigPath
			cell.Spec.Containers[i] = containerSpec
		}

		// Ensure container spec has required names
		if containerSpec.CellName == "" {
			containerSpec.CellName = cellID
		}
		if containerSpec.SpaceName == "" {
			containerSpec.SpaceName = spaceName
		}
		if containerSpec.RealmName == "" {
			containerSpec.RealmName = realmName
		}
		if containerSpec.StackName == "" {
			containerSpec.StackName = stackName
		}
		// Cell-rooted cgroup placement (issue #312). Mirrors the equivalent
		// fill-in for createCellContainers; only takes effect when the cell
		// has been provisioned and its Status.CgroupPath is populated.
		setCellCgroupPath(&containerSpec, cell)
		cell.Spec.Containers[i] = containerSpec

		// Build containerd ID using hierarchical format for containerd operations
		// Store it in ContainerdID field, keep base name in ID field
		containerdID := containerSpec.ContainerdID
		if containerdID == "" {
			var buildErr error
			containerdID, buildErr = naming.BuildContainerdID(
				containerSpec.SpaceName,
				containerSpec.StackName,
				containerSpec.CellName,
				containerSpec.ID,
			)
			if buildErr != nil {
				return nil, fmt.Errorf("failed to build container containerd ID: %w", buildErr)
			}
			containerSpec.ContainerdID = containerdID
			cell.Spec.Containers[i] = containerSpec
		}

		// Log the hierarchical container ID we built
		idFields := appendCellLogFields([]any{"cell", cellID}, cellID, cellName)
		idFields = append(
			idFields,
			"space",
			spaceName,
			"realm",
			realmName,
			"stack",
			stackName,
			"containerName",
			containerSpec.ID,
			"containerdID",
			containerdID,
		)
		r.logger.DebugContext(
			r.ctx,
			"built hierarchical container containerd ID",
			idFields...,
		)

		// Use containerd ID for containerd operations
		exists, err = r.ExistsContainer(internalRealm.Spec.Namespace, containerdID)
		if err != nil {
			// Some other error occurred (connection failure, permission error, etc.)
			fields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
			fields = append(
				fields,
				"space",
				spaceName,
				"realm",
				realmName,
				"containerName",
				containerSpec.ID,
				"err",
				fmt.Sprintf("%v", err),
			)
			r.logger.ErrorContext(
				r.ctx,
				"failed to check if hierarchical container exists",
				fields...,
			)
			return nil, fmt.Errorf("failed to check if hierarchical container exists: %w", err)
		}

		// Log the existence check result
		existsFields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
		existsFields = append(
			existsFields,
			"space",
			spaceName,
			"realm",
			realmName,
			"containerName",
			containerSpec.ID,
			"exists",
			exists,
		)
		r.logger.DebugContext(
			r.ctx,
			"checked container existence",
			existsFields...,
		)

		if !exists {
			fields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
			fields = append(fields, "space", spaceName, "realm", realmName, "cniConfig", containerSpec.CNIConfigPath)
			r.logger.InfoContext(
				r.ctx,
				"container does not exist, creating",
				fields...,
			)
			if containerSpec.CNIConfigPath == "" {
				containerSpec.CNIConfigPath = cniConfigPath
				cell.Spec.Containers[i] = containerSpec
			}

			// Container doesn't exist, create it
			// Log container spec details before creation
			createSpecFields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
			createSpecFields = append(
				createSpecFields,
				"space",
				spaceName,
				"realm",
				realmName,
				"stack",
				stackName,
				"containerName",
				containerSpec.ID,
				"image",
				containerSpec.Image,
				"command",
				containerSpec.Command,
			)
			r.logger.DebugContext(
				r.ctx,
				"creating container with spec",
				createSpecFields...,
			)

			// priorStages threads the controller-side ContainerStatus.Stages
			// snapshot into the renderer so the phase-C2 (#737) gate can omit
			// already-done runOn: create stages from the rendered ContainerDoc
			// on this boot.
			priorStages := priorStagesForContainer(*cell, containerSpec.ID)
			attachOpts, attachErr := r.attachableBuildOpts(internalRealm.Spec.Namespace, containerSpec, creds, priorStages)
			if attachErr != nil {
				return nil, fmt.Errorf("failed to prepare attachable container %s: %w", containerdID, attachErr)
			}
			buildOpts := append(r.daemonDefaultBuildOpts(), attachOpts...)
			buildOpts = append(buildOpts, ctr.WithExtraLabels(specHashLabels(containerSpec)))
			createdContainer, containerCreateErr := r.ctrClient.CreateContainerFromSpec(
				internalRealm.Spec.Namespace,
				containerSpec,
				creds,
				buildOpts...,
			)
			if containerCreateErr != nil {
				// Check if the error indicates the container already exists
				// This can happen due to race conditions where the container
				// was created between the existence check and creation attempt
				errMsg := containerCreateErr.Error()
				if strings.Contains(errMsg, "container already exists") {
					debugFields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
					debugFields = append(debugFields, "space", spaceName, "realm", realmName)
					r.logger.DebugContext(
						r.ctx,
						"container already exists (race condition), skipping",
						debugFields...,
					)
					continue
				}
				errorFields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
				errorFields = append(
					errorFields,
					"space",
					spaceName,
					"realm",
					realmName,
					"cniConfig",
					containerSpec.CNIConfigPath,
					"err",
					fmt.Sprintf("%v", containerCreateErr),
				)
				r.logger.ErrorContext(
					r.ctx,
					"failed to create container from cell",
					errorFields...,
				)
				return nil, fmt.Errorf("failed to create container %s: %w", containerdID, containerCreateErr)
			}
			if createdContainer != nil {
				successFields := appendCellLogFields([]any{"id", containerdID}, cellID, cellName)
				successFields = append(
					successFields,
					"space",
					spaceName,
					"realm",
					realmName,
					"containerName",
					containerSpec.ID,
				)
				r.logger.InfoContext(
					r.ctx,
					"created container from cell",
					successFields...,
				)
				if chownErr := r.attachablePostCreateChown(internalRealm.Spec.Namespace, containerSpec); chownErr != nil {
					return nil, fmt.Errorf(
						"failed to chown attachable tty dir for %s: %w", containerdID, chownErr,
					)
				}
			}
		} else {
			fields := appendCellLogFields([]any{"id", containerID}, cellID, cellName)
			fields = append(fields, "space", spaceName, "realm", realmName)
			r.logger.DebugContext(
				r.ctx,
				"container exists",
				fields...,
			)
		}
	}

	return container, nil
}

func (r *Exec) ensureCellRootContainerSpec(cell intmodel.Cell) (intmodel.ContainerSpec, error) {
	// Extract cell ID and validate
	cellID := strings.TrimSpace(cell.Spec.ID)
	if cellID == "" {
		return intmodel.ContainerSpec{}, errdefs.ErrCellIDRequired
	}

	// Extract and validate realm name
	realmName := strings.TrimSpace(cell.Spec.RealmName)
	if realmName == "" {
		return intmodel.ContainerSpec{}, errdefs.ErrRealmNameRequired
	}

	// Extract and validate space name
	spaceName := strings.TrimSpace(cell.Spec.SpaceName)
	if spaceName == "" {
		return intmodel.ContainerSpec{}, errdefs.ErrSpaceNameRequired
	}

	// Extract and validate stack name
	stackName := strings.TrimSpace(cell.Spec.StackName)
	if stackName == "" {
		return intmodel.ContainerSpec{}, errdefs.ErrStackNameRequired
	}

	// Resolve CNI config path
	cniConfigPath, err := r.ResolveSpaceCNIConfigPath(realmName, spaceName)
	if err != nil {
		return intmodel.ContainerSpec{}, fmt.Errorf("failed to resolve space CNI config: %w", err)
	}

	// Generate containerd ID
	containerdID, err := naming.BuildRootContainerdID(spaceName, stackName, cellID)
	if err != nil {
		return intmodel.ContainerSpec{}, fmt.Errorf("failed to build root container containerd ID: %w", err)
	}

	// Find root container in Containers array if RootContainerID is set
	var rootSpec intmodel.ContainerSpec
	if cell.Spec.RootContainerID != "" {
		// Look for root container in Containers array
		found := false
		for _, container := range cell.Spec.Containers {
			if container.ID == cell.Spec.RootContainerID {
				rootSpec = container
				found = true
				break
			}
		}
		if !found {
			return intmodel.ContainerSpec{}, fmt.Errorf(
				"rootContainerId %q not found in containers array",
				cell.Spec.RootContainerID,
			)
		}
		// Reject the misalignment instead of silently overriding the
		// user-provided root spec — see validateExplicitRootHostNetwork.
		if err := validateExplicitRootHostNetwork(cell, rootSpec); err != nil {
			return intmodel.ContainerSpec{}, err
		}
	} else {
		// Create default root container spec
		// Pass containerdID to set ContainerdID field, ID will be set to "root".
		// The kukepause host path (<RunPath>/bin/kukepause) is bind-mounted at
		// /.kukeon/bin/kukepause as the root container's PID 1 (issue #931). Stage it now so
		// daemon-only bootstrap paths (e2e `startKukeondDaemon`, and any future
		// caller that runs `kukeond serve --run-path X` without a prior
		// `kuke init`) self-heal — `kuke init` also stages it, but its copy is
		// not the *only* writer the daemon can rely on. Idempotent: returns the
		// existing path without re-copying when one is already there.
		kukepauseHostPath, stageErr := stageKukepauseBinary(r.opts.RunPath)
		if stageErr != nil {
			return intmodel.ContainerSpec{}, fmt.Errorf("stage kukepause: %w", stageErr)
		}
		rootSpec = ctr.DefaultRootContainerSpec(
			containerdID,
			cellID,
			realmName,
			spaceName,
			stackName,
			cniConfigPath,
			kukepauseHostPath,
		)
		// Propagate HostNetwork from any cell container to the default root
		// container. Non-root containers join the root container's netns
		// (see JoinContainerNamespaces), so the netns-owning root is the
		// only place that decision can be made — set it host-network if any
		// workload in the cell wants host-network. Used by the kukeond cell
		// (issue #96) so the daemon's CNI/iptables work lands in host scope.
		rootSpec.HostNetwork = cellWantsHostNetworkRoot(cell)
	}

	// Propagate NestedCgroupRuntime from the cell to the root spec so a
	// user-supplied or default-generated root container in a
	// NestedCgroupRuntime cell gets the same /sys/fs/cgroup wiring as
	// the rest of the cell's containers (#322). Mirrors the
	// applyCellContainerDefaults loop for non-root containers; needed
	// here because DefaultRootContainerSpec returns a freshly built
	// spec that never passes through that loop.
	rootSpec.NestedCgroupRuntime = cell.Spec.NestedCgroupRuntime

	// Ensure required fields are set
	rootSpec.Root = true
	if rootSpec.ID == "" {
		rootSpec.ID = "root"
	}
	// Set containerd ID (hierarchical format) if not already set
	if rootSpec.ContainerdID == "" {
		rootSpec.ContainerdID = containerdID
	}
	if rootSpec.CellName == "" {
		rootSpec.CellName = cellID
	}
	if rootSpec.RealmName == "" {
		rootSpec.RealmName = realmName
	}
	if rootSpec.SpaceName == "" {
		rootSpec.SpaceName = spaceName
	}
	if rootSpec.StackName == "" {
		rootSpec.StackName = stackName
	}
	if rootSpec.CNIConfigPath == "" {
		rootSpec.CNIConfigPath = cniConfigPath
	}

	return rootSpec, nil
}
