gitops.go

package main

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"path"
	"path/filepath"
	"regexp"
	"slices"
	"strconv"
	"strings"

	"gopkg.in/yaml.v3"
)

const moduleBundleSchema = "codefly.dev/module-bundle/v1"

var (
	dnsLabelPattern   = regexp.MustCompile(`^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$`)
	unresolvedPattern = regexp.MustCompile(`(?i)REPLACE_ME|saas-starter|\$\{[^}]+\}|<[^>]*replace[^>]*>`)
	// A Cloud SQL instance is addressed by project:region:instance, which is not
	// a DNS name, so it travels beside the ExternalName rather than inside it.
	// The optional leading segment is the legacy domain-scoped project form
	// (example.com:project:region:instance) that older organizations still carry.
	instanceConnectionPattern = regexp.MustCompile(`^(?:[a-z0-9](?:[-a-z0-9]*[a-z0-9])?(?:\.[a-z0-9](?:[-a-z0-9]*[a-z0-9])?)+:)?[a-z][-a-z0-9]{4,28}[a-z0-9]:[a-z0-9]+(?:-[a-z0-9]+)*:[a-z](?:[-a-z0-9]{0,96}[a-z0-9])?$`)
)

type moduleManifest struct {
	Name         string             `yaml:"name"`
	ServiceEntry string             `yaml:"service-entry,omitempty"`
	Services     []serviceReference `yaml:"services"`
}

type workspaceManifest struct {
	Name         string               `yaml:"name"`
	Environments []*environmentConfig `yaml:"environments,omitempty"`
	root         string
}

type environmentConfig struct {
	Name      string                    `yaml:"name"`
	Namespace string                    `yaml:"namespace"`
	Cluster   environmentCluster        `yaml:"cluster"`
	Ingress   []environmentIngressRoute `yaml:"ingress,omitempty"`
	// KeyService is which key service holds this environment's keys, in the
	// same vocabulary the `key-service` configuration group's
	// KEY_SERVICE_BACKEND uses. It decides whether this module composes a Vault
	// here at all. Unstated means `vault`; resolveKeyService says why that is a
	// default rather than a refusal, and what has to change for it to become
	// one.
	KeyService      string                          `yaml:"key-service,omitempty"`
	ManagedServices map[string]managedServiceConfig `yaml:"managed-services,omitempty"`
}

type environmentCluster struct {
	Kind string `yaml:"kind"`
}

type environmentIngressRoute struct {
	Name     string   `yaml:"name"`
	Service  string   `yaml:"service"`
	Endpoint string   `yaml:"endpoint"`
	Hosts    []string `yaml:"hosts"`
}

type serviceReference struct {
	Name string  `yaml:"name"`
	Path *string `yaml:"path,omitempty"`
}

type serviceDefinition struct {
	name      string
	directory string
}

// moduleBundle is the typed, transport-neutral output the module plugin emits.
// It carries the module identity and, per declared environment, the
// module-owned Kubernetes overlay path plus the topology and placement metadata
// a promotion driver needs to compose repository transport (Argo, Flux, or
// otherwise) on its own. The plugin never records repositories, revisions, or
// Argo resources.
type moduleBundle struct {
	SchemaVersion string              `json:"schemaVersion"`
	Module        string              `json:"module"`
	Namespace     string              `json:"namespace"`
	ServiceEntry  string              `json:"serviceEntry"`
	Environments  []bundleEnvironment `json:"environments"`
}

type bundleEnvironment struct {
	Name                   string                  `json:"name"`
	Namespace              string                  `json:"namespace"`
	Cluster                string                  `json:"cluster"`
	ResourcePath           string                  `json:"resourcePath"`
	Services               []string                `json:"services"`
	Ingress                []bundleIngressRoute    `json:"ingress"`
	ManagedServiceHandoffs []managedServiceHandoff `json:"managedServiceHandoffs,omitempty"`
	DeployJobs             []bundleDeployJob       `json:"deployJobs,omitempty"`
}

// bundleDeployJob is a store-writing one-shot the promotion driver runs against
// a dependency's store after that dependency's own bring-up. Unlike a service's
// self-serving bootstrap Job (bootstrap_job_endpoints, which reaches the
// service's own endpoints), it runs one service's image but writes to a
// dependency it declares — carrying the generated artifact and the target the
// driver mounts and connects. The driver schedules it after the target's
// migration and fails the promotion if it exits non-zero.
type bundleDeployJob struct {
	Name    string             `json:"name"`
	Service string             `json:"service"`
	Command string             `json:"command"`
	Catalog string             `json:"catalog"`
	Force   bool               `json:"force,omitempty"`
	Writes  bundleDeployTarget `json:"writes"`
	After   []string           `json:"after,omitempty"`
}

type bundleDeployTarget struct {
	Service  string `json:"service"`
	Endpoint string `json:"endpoint"`
	Port     uint32 `json:"port"`
}

type bundleIngressRoute struct {
	Name     string   `json:"name"`
	Service  string   `json:"service"`
	Endpoint string   `json:"endpoint"`
	Port     uint32   `json:"port"`
	Hosts    []string `json:"hosts"`
}

type managedServiceHandoff struct {
	Service      string `json:"service"`
	Kind         string `json:"kind"`
	ExternalName string `json:"externalName"`
	AuthMode     string `json:"authMode,omitempty"`
	// EgressNamespace is the in-cluster namespace that holds this dependency,
	// set only for a kind the cell runs inside the same cluster. It is what the
	// caller's egress policy selects on; such a dependency has no stable address
	// to write a CIDR for, and the pod CIDR would authorize the whole cluster.
	EgressNamespace        string   `json:"egressNamespace,omitempty"`
	InstanceConnectionName string   `json:"instanceConnectionName,omitempty"`
	SecretReferences       []string `json:"secretReferences,omitempty"`
}

// UnmarshalJSON reads the handoff kind from "kind", falling back to the legacy
// "awsKind" key so bundles written before the rename still parse. Drop the
// fallback once every producer emits "kind".
func (h *managedServiceHandoff) UnmarshalJSON(data []byte) error {
	type alias managedServiceHandoff
	aux := struct {
		*alias
		LegacyAWSKind string `json:"awsKind"`
	}{alias: (*alias)(h)}
	if err := json.Unmarshal(data, &aux); err != nil {
		return err
	}
	if h.Kind == "" {
		h.Kind = aux.LegacyAWSKind
	}
	return nil
}

type managedServiceConfig struct {
	Kind                   string                   `yaml:"kind"`
	ExternalName           string                   `yaml:"external-name"`
	AuthMode               string                   `yaml:"auth-mode,omitempty"`
	InstanceConnectionName string                   `yaml:"instance-connection-name,omitempty"`
	EgressCIDRs            []string                 `yaml:"egress-cidrs,omitempty"`
	EgressNamespace        string                   `yaml:"egress-namespace,omitempty"`
	SecretReferences       []managedSecretReference `yaml:"secret-references,omitempty"`
}

type managedSecretReference struct {
	Name        string         `yaml:"name"`
	RemoteKey   string         `yaml:"remote-key"`
	SecretStore secretStoreRef `yaml:"secret-store"`
}

type secretStoreRef struct {
	Name string `yaml:"name"`
	Kind string `yaml:"kind"`
}

type objectMeta struct {
	Name       string            `yaml:"name"`
	Namespace  string            `yaml:"namespace,omitempty"`
	Labels     map[string]string `yaml:"labels,omitempty"`
	Finalizers []string          `yaml:"finalizers,omitempty"`
}

type kubeObject struct {
	APIVersion string     `yaml:"apiVersion"`
	Kind       string     `yaml:"kind"`
	Metadata   objectMeta `yaml:"metadata"`
	Spec       any        `yaml:"spec,omitempty"`
}

type kustomization struct {
	APIVersion string   `yaml:"apiVersion"`
	Kind       string   `yaml:"kind"`
	Namespace  string   `yaml:"namespace,omitempty"`
	Resources  []string `yaml:"resources"`
}

type environmentPlan struct {
	environment *environmentConfig
	cluster     string
	services    []string
	ingress     []ingressRoutePlan
	handoffs    []managedServiceHandoff
	managed     map[string]managedServiceConfig
	// keyService is the resolved `key-service` selection for this environment.
	keyService string
	// withdrawn holds the services this environment's key-service selection
	// takes OUT of the composition: not an in-cluster workload, and not a
	// managed dependency either — nothing reaches them, so every edge to one is
	// dropped rather than needing a handoff. It is a third state beside
	// in-cluster and managed, and it exists because "the keys are in the cell's
	// cloud key service" names nothing for a handoff to point at.
	withdrawn map[string]struct{}
}

type ingressRoutePlan struct {
	name     string
	service  string
	endpoint string
	// port is the pod port the ingress gateway's traffic arrives on (what the
	// AuthorizationPolicy and NetworkPolicy match); servicePort is the
	// Kubernetes Service port the VirtualService routes to.
	port        uint32
	servicePort uint32
	hosts       []string
}

type deploymentTopology struct {
	Version    string              `yaml:"version"`
	Module     topologyModule      `yaml:"module"`
	Interface  []topologyInterface `yaml:"interface"`
	Services   []topologyService   `yaml:"services"`
	DeployJobs []topologyDeployJob `yaml:"deploy_jobs,omitempty"`

	// internalMethodPaths holds, per service, the gRPC method paths marked
	// EXPOSURE_INTERNAL in that service's generated authorization catalog. It is
	// populated after decoding and drives the mesh reach gate for the internal
	// authority surface.
	internalMethodPaths map[string][]string
}

type topologyModule struct {
	Name         string         `yaml:"name"`
	Namespace    string         `yaml:"namespace"`
	ServiceEntry string         `yaml:"service_entry"`
	Description  string         `yaml:"description"`
	Agent        map[string]any `yaml:"agent,omitempty"`
}

type topologyInterface struct {
	Service    string `yaml:"service"`
	Endpoint   string `yaml:"endpoint"`
	Visibility string `yaml:"visibility"`
}

type topologyService struct {
	Name                               string                               `yaml:"name"`
	Version                            string                               `yaml:"version,omitempty"`
	Description                        string                               `yaml:"description,omitempty"`
	Agent                              map[string]any                       `yaml:"agent,omitempty"`
	Kubernetes                         *topologyKubernetesIdentity          `yaml:"kubernetes,omitempty"`
	WorkspaceConfigurationDependencies []string                             `yaml:"workspace_configuration_dependencies,omitempty"`
	SecretServiceConfigurations        []topologySecretServiceConfiguration `yaml:"secret_service_configurations,omitempty"`
	Endpoints                          []topologyEndpoint                   `yaml:"endpoints"`
	BootstrapJobEndpoints              []string                             `yaml:"bootstrap_job_endpoints,omitempty"`
	Dependencies                       []topologyDependency                 `yaml:"dependencies,omitempty"`
	InternalHTTPRoutes                 []topologyInternalHTTPRoute          `yaml:"internal_http_routes,omitempty"`
	PublicEgressPorts                  []uint32                             `yaml:"public_egress_ports,omitempty"`
	Spec                               map[string]any                       `yaml:"spec,omitempty"`
}

// topologyInternalHTTPRoute is an HTTP route carrying cluster-internal
// authority on a port that also serves browser-facing traffic. Path and method
// are the only thing separating the two, so it is authored rather than derived.
type topologyInternalHTTPRoute struct {
	Path    string   `yaml:"path"`
	Methods []string `yaml:"methods"`
}

type topologyKubernetesIdentity struct {
	ServiceName string `yaml:"service-name"`
	AppLabel    string `yaml:"app-label"`
}

type topologySecretServiceConfiguration struct {
	Name    string                                    `yaml:"name"`
	Entries []topologySecretServiceConfigurationEntry `yaml:"entries"`
}

type topologySecretServiceConfigurationEntry struct {
	Key string `yaml:"key"`
}

// topologyEndpoint carries both of an endpoint's ports. Port is the pod port
// the process binds — what a NetworkPolicy and an Istio AuthorizationPolicy
// match. ServicePort is the Kubernetes Service port Codefly's render allocates
// (network.DeployedEndpointPorts) — what a VirtualService destination names.
type topologyEndpoint struct {
	Name        string `yaml:"name"`
	API         string `yaml:"api,omitempty"`
	Visibility  string `yaml:"visibility"`
	Port        uint32 `yaml:"port"`
	ServicePort uint32 `yaml:"service_port"`
}

type topologyDependency struct {
	Service   string   `yaml:"service"`
	Endpoints []string `yaml:"endpoints"`
}

// topologyDeployJob declares a store-writing one-shot the promotion driver runs
// as a deploy step. It runs Service's image but writes to a dependency's
// endpoint (Writes) rather than serving its own — the boundary a self-serving
// bootstrap_job_endpoints Job cannot express — consuming the generated Catalog
// artifact. Force selects the importer's authoritative-catalog semantics: the
// generated catalog is base-tracked and only changes through contribution →
// regeneration, so a removal in it is intentional and applied past the
// truncation guard.
type topologyDeployJob struct {
	Name    string                  `yaml:"name"`
	Service string                  `yaml:"service"`
	Command string                  `yaml:"command"`
	Catalog string                  `yaml:"catalog"`
	Force   bool                    `yaml:"force,omitempty"`
	Writes  topologyDeployJobTarget `yaml:"writes"`
	After   []string                `yaml:"after,omitempty"`
}

type topologyDeployJobTarget struct {
	Service  string `yaml:"service"`
	Endpoint string `yaml:"endpoint"`
}

func loadWorkspaceManifest(root string) (*workspaceManifest, error) {
	data, err := os.ReadFile(filepath.Join(root, workspaceYamlPath))
	if err != nil {
		return nil, err
	}
	var workspace workspaceManifest
	if err := yaml.Unmarshal(data, &workspace); err != nil {
		return nil, err
	}
	workspace.root = root
	return &workspace, nil
}

// generateDeploymentBundle renders the module-owned Kubernetes overlay for every
// declared environment and writes a typed, transport-neutral bundle manifest.
// It requires no Git binary, checkout, network, or cluster access.
func generateDeploymentBundle(moduleDir string, workspace *workspaceManifest) error {
	manifest, err := loadModuleManifest(moduleDir)
	if err != nil {
		return err
	}
	services, err := validateServiceInventory(moduleDir, manifest.Services)
	if err != nil {
		return err
	}
	topology, err := loadDeploymentTopology(moduleDir, manifest.Name, services)
	if err != nil {
		return err
	}
	environments, err := selectEnvironments(workspace)
	if err != nil {
		return err
	}

	root := filepath.Join(moduleDir, filepath.FromSlash(bundleRelativeDir))
	if err := os.MkdirAll(filepath.Dir(root), 0o755); err != nil {
		return fmt.Errorf("create deployment directory: %w", err)
	}
	stage, err := os.MkdirTemp(filepath.Dir(root), ".kustomize-stage-*")
	if err != nil {
		return fmt.Errorf("create bundle staging directory: %w", err)
	}
	defer func() { _ = os.RemoveAll(stage) }()

	bundle := moduleBundle{
		SchemaVersion: moduleBundleSchema,
		Module:        manifest.Name,
		Namespace:     topology.Module.Namespace,
		ServiceEntry:  topology.Module.ServiceEntry,
	}
	for _, environment := range environments {
		_, managed, kind, err := classifyEnvironment(environment)
		if err != nil {
			return err
		}
		plan, err := planEnvironment(environment, serviceNames(services), topology, kind, managed)
		if err != nil {
			return err
		}
		if err := renderEnvironment(stage, workspace.Name, manifest.Name, topology, plan); err != nil {
			return err
		}
		bundle.Environments = append(bundle.Environments, bundleEnvironment{
			Name:                   plan.environment.Name,
			Namespace:              plan.environment.Namespace,
			Cluster:                plan.cluster,
			ResourcePath:           path.Join("overlays", plan.environment.Name),
			Services:               append([]string(nil), plan.services...),
			Ingress:                bundleIngress(plan.ingress),
			ManagedServiceHandoffs: append([]managedServiceHandoff(nil), plan.handoffs...),
			DeployJobs:             planDeployJobs(topology, plan),
		})
	}
	if err := writeJSON(filepath.Join(stage, "bundle.json"), bundle); err != nil {
		return err
	}
	if err := validateGeneratedBundle(stage, bundle); err != nil {
		return err
	}
	return replaceGeneratedTree(stage, root)
}

func bundleIngress(routes []ingressRoutePlan) []bundleIngressRoute {
	result := make([]bundleIngressRoute, 0, len(routes))
	for _, route := range routes {
		result = append(result, bundleIngressRoute{
			Name:     route.name,
			Service:  route.service,
			Endpoint: route.endpoint,
			Port:     route.port,
			Hosts:    append([]string(nil), route.hosts...),
		})
	}
	return result
}

// planDeployJobs records the deploy Jobs the promotion driver must run for this
// environment: those whose running service is in-cluster and whose write target
// is reachable (in-cluster or a managed handoff). Each carries the generated
// catalog artifact and the resolved target endpoint so the driver mounts the
// document and connects the store without consulting the topology itself.
func planDeployJobs(topology deploymentTopology, plan environmentPlan) []bundleDeployJob {
	inCluster := make(map[string]struct{}, len(plan.services))
	for _, service := range plan.services {
		inCluster[service] = struct{}{}
	}
	jobs := make([]bundleDeployJob, 0, len(topology.DeployJobs))
	for _, job := range topology.DeployJobs {
		if _, exists := inCluster[job.Service]; !exists {
			continue
		}
		_, targetInCluster := inCluster[job.Writes.Service]
		_, targetManaged := plan.managed[job.Writes.Service]
		if !targetInCluster && !targetManaged {
			continue
		}
		target, _ := topologyServiceByName(topology, job.Writes.Service)
		endpoint, _ := topologyEndpointByName(target, job.Writes.Endpoint)
		jobs = append(jobs, bundleDeployJob{
			Name:    job.Name,
			Service: job.Service,
			Command: job.Command,
			Catalog: job.Catalog,
			Force:   job.Force,
			Writes: bundleDeployTarget{
				Service:  job.Writes.Service,
				Endpoint: job.Writes.Endpoint,
				Port:     endpoint.Port,
			},
			After: append([]string(nil), job.After...),
		})
	}
	if len(jobs) == 0 {
		return nil
	}
	return jobs
}

func replaceGeneratedTree(stage, root string) error {
	if _, err := os.Lstat(root); errors.Is(err, os.ErrNotExist) {
		if err := os.Rename(stage, root); err != nil {
			return fmt.Errorf("install generated bundle directory: %w", err)
		}
		return nil
	} else if err != nil {
		return fmt.Errorf("inspect generated bundle directory: %w", err)
	}

	backup, err := os.MkdirTemp(filepath.Dir(root), ".kustomize-backup-*")
	if err != nil {
		return fmt.Errorf("create bundle backup path: %w", err)
	}
	if err := os.Remove(backup); err != nil {
		return fmt.Errorf("prepare bundle backup path: %w", err)
	}
	if err := os.Rename(root, backup); err != nil {
		return fmt.Errorf("back up generated bundle directory: %w", err)
	}
	if err := os.Rename(stage, root); err != nil {
		if rollbackErr := os.Rename(backup, root); rollbackErr != nil {
			return fmt.Errorf("install generated bundle directory: %w (rollback failed: %v; previous tree remains at %s)", err, rollbackErr, backup)
		}
		return fmt.Errorf("install generated bundle directory: %w", err)
	}
	if err := os.RemoveAll(backup); err != nil {
		return fmt.Errorf("remove previous bundle directory: %w", err)
	}
	return nil
}

func loadModuleManifest(moduleDir string) (moduleManifest, error) {
	data, err := os.ReadFile(filepath.Join(moduleDir, moduleYamlPath))
	if err != nil {
		return moduleManifest{}, fmt.Errorf("read %s: %w", moduleYamlPath, err)
	}
	var manifest moduleManifest
	if err := yaml.Unmarshal(data, &manifest); err != nil {
		return moduleManifest{}, fmt.Errorf("parse %s: %w", moduleYamlPath, err)
	}
	if err := validateDNSLabel("module name", manifest.Name); err != nil {
		return moduleManifest{}, err
	}
	if len(manifest.Services) == 0 {
		return moduleManifest{}, fmt.Errorf("module %q declares no services", manifest.Name)
	}
	return manifest, nil
}

func validateServiceInventory(moduleDir string, references []serviceReference) ([]serviceDefinition, error) {
	declared := make(map[string]string, len(references))
	declaredPaths := make(map[string]string, len(references))
	services := make([]serviceDefinition, 0, len(references))
	serviceRoot := filepath.Join(moduleDir, "services")
	for _, reference := range references {
		if err := validateDNSLabel("service name", reference.Name); err != nil {
			return nil, err
		}
		if _, exists := declared[reference.Name]; exists {
			return nil, fmt.Errorf("module declares service %q more than once", reference.Name)
		}
		servicePath := filepath.Join(serviceRoot, reference.Name)
		displayPath := reference.Name
		if reference.Path != nil {
			override := *reference.Path
			if strings.ContainsRune(override, '\x00') {
				return nil, fmt.Errorf("service %q path contains NUL", reference.Name)
			}
			if filepath.IsAbs(override) {
				servicePath = filepath.Clean(override)
				displayPath = servicePath
			} else {
				relative, err := cleanRelativeFilesystemPath(override)
				if err != nil {
					return nil, fmt.Errorf("service %q path: %w", reference.Name, err)
				}
				servicePath = filepath.Join(serviceRoot, relative)
				displayPath = filepath.ToSlash(relative)
			}
		}
		canonicalPath, err := filepath.Abs(servicePath)
		if err != nil {
			return nil, fmt.Errorf("service %q path: %w", reference.Name, err)
		}
		if owner, exists := declaredPaths[canonicalPath]; exists {
			return nil, fmt.Errorf("services %q and %q declare the same path %q", owner, reference.Name, displayPath)
		}
		declared[reference.Name] = canonicalPath
		declaredPaths[canonicalPath] = reference.Name
		services = append(services, serviceDefinition{name: reference.Name, directory: canonicalPath})

		if !filepath.IsAbs(valueOrEmpty(reference.Path)) {
			current := serviceRoot
			relative, err := filepath.Rel(serviceRoot, canonicalPath)
			if err != nil {
				return nil, fmt.Errorf("service %q path: %w", reference.Name, err)
			}
			for _, element := range strings.Split(relative, string(filepath.Separator)) {
				current = filepath.Join(current, element)
				info, statErr := os.Lstat(current)
				if statErr != nil {
					return nil, fmt.Errorf("declared service %q path %q is missing: %w", reference.Name, displayPath, statErr)
				}
				if !info.IsDir() {
					return nil, fmt.Errorf("declared service %q path %q is not a directory", reference.Name, displayPath)
				}
			}
		} else {
			info, statErr := os.Stat(canonicalPath)
			if statErr != nil {
				return nil, fmt.Errorf("declared service %q path %q is missing: %w", reference.Name, displayPath, statErr)
			}
			if !info.IsDir() {
				return nil, fmt.Errorf("declared service %q path %q is not a directory", reference.Name, displayPath)
			}
		}
	}

	expectedPaths := make(map[string]struct{}, len(declared))
	for _, absolute := range declared {
		relative, err := filepath.Rel(serviceRoot, absolute)
		if err == nil && filepath.IsLocal(relative) {
			expectedPaths[relative] = struct{}{}
		}
	}
	if _, err := os.Stat(serviceRoot); errors.Is(err, os.ErrNotExist) && len(expectedPaths) == 0 {
		return services, nil
	}
	if err := filepath.WalkDir(serviceRoot, func(file string, entry os.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if !entry.IsDir() || file == serviceRoot {
			return nil
		}
		relative, err := filepath.Rel(serviceRoot, file)
		if err != nil {
			return err
		}
		if _, exists := expectedPaths[relative]; exists {
			return filepath.SkipDir
		}
		prefix := relative + string(filepath.Separator)
		for expected := range expectedPaths {
			if strings.HasPrefix(expected, prefix) {
				return nil
			}
		}
		return fmt.Errorf("service path %q has no module service declaration", filepath.ToSlash(relative))
	}); err != nil {
		return nil, fmt.Errorf("validate service paths: %w", err)
	}
	return services, nil
}

func valueOrEmpty(value *string) string {
	if value == nil {
		return ""
	}
	return *value
}

func serviceNames(services []serviceDefinition) []string {
	names := make([]string, 0, len(services))
	for _, service := range services {
		names = append(names, service.name)
	}
	return names
}

func selectEnvironments(workspace *workspaceManifest) ([]*environmentConfig, error) {
	if err := validateDNSLabel("workspace name", workspace.Name); err != nil {
		return nil, err
	}
	if len(workspace.Environments) == 0 {
		return nil, fmt.Errorf("workspace %q declares no deployment environments", workspace.Name)
	}
	seen := make(map[string]struct{}, len(workspace.Environments))
	for _, environment := range workspace.Environments {
		if environment == nil {
			return nil, fmt.Errorf("workspace contains an empty environment")
		}
		if err := validateDNSLabel("environment name", environment.Name); err != nil {
			return nil, err
		}
		if _, exists := seen[environment.Name]; exists {
			return nil, fmt.Errorf("workspace declares environment %q more than once", environment.Name)
		}
		seen[environment.Name] = struct{}{}
		if err := validateDNSLabel("environment namespace", environment.Namespace); err != nil {
			return nil, fmt.Errorf("environment %q: %w", environment.Name, err)
		}
	}
	return workspace.Environments, nil
}

func planEnvironment(
	environment *environmentConfig,
	services []string,
	topology deploymentTopology,
	cluster string,
	managedCapable bool,
) (environmentPlan, error) {
	plan := environmentPlan{
		environment: environment,
		cluster:     cluster,
		managed:     make(map[string]managedServiceConfig),
	}
	keyService, withdrawn, err := resolveKeyService(environment, services)
	if err != nil {
		return environmentPlan{}, err
	}
	plan.keyService, plan.withdrawn = keyService, withdrawn
	if err := validateManagedServices(environment, services, topology, managedCapable, &plan); err != nil {
		return environmentPlan{}, err
	}
	for _, service := range services {
		if _, managed := plan.managed[service]; managed {
			continue
		}
		if _, gone := plan.withdrawn[service]; gone {
			continue
		}
		plan.services = append(plan.services, service)
	}
	slices.Sort(plan.services)
	if err := validateIngressRoutes(environment, topology, &plan); err != nil {
		return environmentPlan{}, err
	}
	return plan, nil
}

func validateIngressRoutes(
	environment *environmentConfig,
	topology deploymentTopology,
	plan *environmentPlan,
) error {
	// Public ingress is optional: an environment may front the mesh with a
	// gateway this module does not own, or expose nothing publicly yet. Such an
	// environment renders module-owned baseline resources without a Gateway.
	if len(environment.Ingress) == 0 {
		return nil
	}

	routeNames := make(map[string]struct{}, len(environment.Ingress))
	hosts := make(map[string]string)
	for _, route := range environment.Ingress {
		if err := validateDNSLabel("ingress route name", route.Name); err != nil {
			return fmt.Errorf("environment %q: %w", environment.Name, err)
		}
		if _, duplicate := routeNames[route.Name]; duplicate {
			return fmt.Errorf("environment %q repeats ingress route %q", environment.Name, route.Name)
		}
		routeNames[route.Name] = struct{}{}
		if !slices.Contains(plan.services, route.Service) {
			return fmt.Errorf(
				"environment %q ingress route %q targets service %q outside its in-cluster inventory",
				environment.Name,
				route.Name,
				route.Service,
			)
		}
		service, exists := topologyServiceByName(topology, route.Service)
		if !exists {
			return fmt.Errorf(
				"environment %q ingress route %q targets unknown service %q",
				environment.Name,
				route.Name,
				route.Service,
			)
		}
		endpoint, exists := topologyEndpointByName(service, route.Endpoint)
		if !exists || !topologyExposesPublicEndpoint(topology, route.Service, route.Endpoint) {
			return fmt.Errorf(
				"environment %q ingress route %q target %s/%s is not a public module interface",
				environment.Name,
				route.Name,
				route.Service,
				route.Endpoint,
			)
		}
		if len(route.Hosts) == 0 {
			return fmt.Errorf("environment %q ingress route %q declares no exact hosts", environment.Name, route.Name)
		}
		planned := ingressRoutePlan{
			name:        route.Name,
			service:     route.Service,
			endpoint:    route.Endpoint,
			port:        endpoint.Port,
			servicePort: endpoint.ServicePort,
		}
		for _, host := range route.Hosts {
			host = strings.TrimSpace(host)
			if !validExternalName(host) {
				return fmt.Errorf(
					"environment %q ingress route %q host %q is not an exact DNS name",
					environment.Name,
					route.Name,
					host,
				)
			}
			if owner, duplicate := hosts[host]; duplicate {
				return fmt.Errorf(
					"environment %q ingress host %q is declared by routes %q and %q",
					environment.Name,
					host,
					owner,
					route.Name,
				)
			}
			hosts[host] = route.Name
			planned.hosts = append(planned.hosts, host)
		}
		plan.ingress = append(plan.ingress, planned)
	}
	return nil
}

func topologyExposesPublicEndpoint(topology deploymentTopology, service, endpoint string) bool {
	return slices.ContainsFunc(topology.Interface, func(exposed topologyInterface) bool {
		return exposed.Service == service &&
			exposed.Endpoint == endpoint &&
			exposed.Visibility == "public"
	})
}

func validateManagedServices(
	environment *environmentConfig,
	services []string,
	topology deploymentTopology,
	managedCapable bool,
	plan *environmentPlan,
) error {
	if !managedCapable && len(environment.ManagedServices) > 0 {
		return fmt.Errorf("environment %q declares managed services for an in-cluster environment", environment.Name)
	}
	declared := make(map[string]struct{}, len(services))
	for _, service := range services {
		declared[service] = struct{}{}
	}
	for service, config := range environment.ManagedServices {
		_, exists := declared[service]
		if !exists {
			return fmt.Errorf("environment %q declares unexpected managed service %q", environment.Name, service)
		}
		// A withdrawn service is not an in-cluster workload AND not something a
		// handoff can point at: the selection that withdrew it says the keys are
		// in the cell's cloud key service, so there is no Vault to name an
		// ExternalName for. Accepting both would render reach to an address
		// nothing dials and leave the reader unable to say which declaration won.
		if _, gone := plan.withdrawn[service]; gone {
			return fmt.Errorf(
				"environment %q declares both key-service: %s and a managed service %q: the selection means the keys are in the cell's cloud key-management service, so there is no Vault for a handoff to point at — declare key-service: %s with the %q handoff, or key-service: %s with neither",
				environment.Name, plan.keyService, service, keyServiceVault, managedKindCellVault, keyServiceKMS)
		}
		switch config.Kind {
		case "elasticache", "rds-postgresql", "s3", "secrets-manager", "azure-postgres-flexible", "cloud-sql-postgres", managedKindCellVault:
		default:
			return fmt.Errorf("environment %q managed service %q kind %q is not supported", environment.Name, service, config.Kind)
		}
		if !validExternalName(config.ExternalName) {
			return fmt.Errorf("environment %q managed service %q external-name %q is not an exact DNS name", environment.Name, service, config.ExternalName)
		}
		authMode, instanceConnection, err := validateManagedAuth(environment.Name, service, config)
		if err != nil {
			return err
		}
		egressNamespace, err := validateManagedEgress(environment.Name, service, config)
		if err != nil {
			return err
		}
		// Store what was validated rather than what was written: the renderer and
		// the bundle both read these back, so a padded or defaulted spelling must
		// not survive past the one place that checked it.
		config.AuthMode = authMode
		config.InstanceConnectionName = instanceConnection
		config.EgressNamespace = egressNamespace
		for _, cidr := range config.EgressCIDRs {
			if _, _, err := net.ParseCIDR(cidr); err != nil {
				return fmt.Errorf("environment %q managed service %q egress CIDR %q is invalid", environment.Name, service, cidr)
			}
		}
		referenceNames := make(map[string]struct{}, len(config.SecretReferences))
		handoff := managedServiceHandoff{
			Service:                service,
			Kind:                   config.Kind,
			ExternalName:           config.ExternalName,
			EgressNamespace:        egressNamespace,
			InstanceConnectionName: instanceConnection,
		}
		// Only a non-default mode earns a key: every consumer already reads an
		// absent authMode as the password-backed shape, so emitting it for the
		// default would rewrite every existing handoff to say what it already said.
		if authMode != "password" {
			handoff.AuthMode = authMode
		}
		for _, reference := range config.SecretReferences {
			if err := validateDNSLabel("managed secret reference name", reference.Name); err != nil {
				return fmt.Errorf("environment %q managed service %q: %w", environment.Name, service, err)
			}
			if _, duplicate := referenceNames[reference.Name]; duplicate {
				return fmt.Errorf("environment %q managed service %q repeats secret reference %q", environment.Name, service, reference.Name)
			}
			referenceNames[reference.Name] = struct{}{}
			if strings.TrimSpace(reference.RemoteKey) == "" ||
				strings.ContainsAny(reference.RemoteKey, "\r\n") ||
				strings.TrimSpace(reference.SecretStore.Name) == "" ||
				(reference.SecretStore.Kind != "SecretStore" && reference.SecretStore.Kind != "ClusterSecretStore") {
				return fmt.Errorf("environment %q managed service %q secret reference %q is incomplete", environment.Name, service, reference.Name)
			}
			handoff.SecretReferences = append(handoff.SecretReferences, reference.Name)
		}
		plan.managed[service] = config
		plan.handoffs = append(plan.handoffs, handoff)
	}
	slices.SortFunc(plan.handoffs, func(left, right managedServiceHandoff) int {
		return strings.Compare(left.Service, right.Service)
	})
	return validateManagedDependencyCIDRs(environment.Name, topology, plan.managed)
}

// The key services an environment may select, spelled exactly as
// KEY_SERVICE_BACKEND spells them in the `key-service` configuration group
// (module/configurations/local/key-service.env), so one word means one thing on
// both sides of the deployment.
const (
	keyServiceVault = "vault"
	keyServiceKMS   = "kms"
)

// keyServiceVaultService is the service this module composes to BE the `vault`
// key service. It is the only service a key-service selection withdraws, and it
// is named here rather than discovered because the relationship runs the other
// way: the service exists to serve that selection.
const keyServiceVaultService = "vault"

// keyServiceGroup is the workspace-configuration group the `vault` key service
// is REACHED through: the address, the auth method, the AppRole mount and, in
// its secret half, the credential that uses them.
//
// It is named beside the service rather than derived from it because the two
// are different facts about the same selection, and only one of them governs
// the group. A group is required by the BACKEND an environment selected — it is
// how a process reaches that backend — not by whether a workload of the same
// name happens to be in the inventory. The two coincide for this module, which
// is exactly why keying the group off the service read as correct while being
// the wrong owner: a cell that selects the cloud key service and still composes
// a Vault for something else would have kept a binding nothing uses, and a cell
// that selects the vault backend from a Vault the CELL runs — composed here as
// a managed cell-vault — must keep the group though no Vault of this module's
// is ever rendered.
const keyServiceGroup = "vault"

// resolveKeyService answers, for one environment, which key service holds its
// keys — and therefore whether this module composes a Vault here.
//
// WHY THIS IS A MANIFEST FACT AND NOT A CONFIGURATION READ. accounts selects its
// backend at runtime from the `key-service` group, and on
// KEY_SERVICE_BACKEND=kms with KEY_SERVICE_SIGNING_BACKEND=kms-wrapped it opens
// no Vault connection at all (services/accounts/code/pkg/keyservice/load.go:
// the Vault binding is built only by the `vault` sealer, and the wrapped signing
// key is unwrapped from configuration). The INVENTORY, though, is decided here,
// before any process exists: what is rendered, scheduled and given a volume.
// Those two planes had no declaration in common, so a cell whose keys live in
// Cloud KMS still got a Vault StatefulSet — rendered, scheduled, refusing to
// start without the auto-unseal seal it is forbidden to hold, and crash-looping
// as a permanently degraded workload that nothing reads.
//
// The renderer could have read the group's .env instead. It must not: a group's
// values are the configuration plane's to deliver per environment, the renderer
// has never read one, and a file that may legitimately be absent would make the
// render fall back to composing a Vault — which is exactly the failure. So the
// selection is DECLARED on the environment, where the inventory is decided.
//
// AN UNSTATED SELECTION IS ANSWERED BY THE INVENTORY, which is the other
// declaration a composition makes about this module's Vault and the only one it
// can write today: composing the `vault` service selects the vault backend,
// composing none selects the cell's cloud key service. That keeps the inventory
// a SOURCE of the selection while the selection alone decides what follows from
// it — which workloads are rendered, which edges exist, and which configuration
// group is declared. On a deployed environment an unstated selection should be
// refused outright, exactly as `keyservice.selectedBackend` refuses the same
// unstated fact at boot. It is not refused yet, and the reason is a tooling gap
// rather than a judgement:
//
// THE ENVIRONMENT BLOCK'S SCHEMA IS NOT THIS MODULE'S. `workspace.codefly.yaml`
// is parsed by the CLI's own `pkg/environments.Environment`, strictly and
// recursively, and that type has no extension seam — core's
// `resources.Environment` has an inline `Extensions map[string]YAMLValue` for
// host-owned keys, but the CLI does not use it here. So until the CLI accepts
// `key-service`, a composition cannot write the line, and refusing an unstated
// selection would refuse every composition instead of the one mistake it is
// aimed at. Verified against codefly 0.1.181: the field is rejected with
// "field key-service not found in type environments.Environment", and so are
// the `auth-mode` and `egress-namespace` keys this module's own cell-vault
// handoff already documents.
//
// Making it required is therefore one line here plus the CLI's field; issue
// #1128 carries that. Everything else below is live now: a composition that
// composes no `vault` service at all selects the cloud key service by saying
// so in the inventory, which the CLI does parse, and the selection withdraws
// the workload and the `vault` configuration group with it.
//
// A local environment composes this module's own Vault, which is what `codefly
// run service --fixture dev-admin` reaches over loopback with a token, so the
// default is right there in any case.
func resolveKeyService(
	environment *environmentConfig,
	services []string,
) (string, map[string]struct{}, error) {
	// Whether this composition has a Vault AT ALL. A cell that runs its own
	// declares `vault` as a managed service of kind cell-vault, which
	// validateManagedServices requires to be in the inventory too, so one
	// question answers both shapes.
	composed := slices.Contains(services, keyServiceVaultService)
	withdrawn := make(map[string]struct{})
	switch stated := strings.TrimSpace(environment.KeyService); stated {
	case "":
		// UNSTATED: the inventory answers. A composition that composes a
		// `vault` service has a Vault for this environment to use and no other
		// key service named, which is the selection this module had before the
		// field existed; one that composes none has said the keys are elsewhere
		// — it is the only half of this selection a composition can write today
		// — and there is nothing here to withdraw.
		if composed {
			return keyServiceVault, withdrawn, nil
		}
		return keyServiceKMS, withdrawn, nil
	case keyServiceVault:
		// STATED vault with no Vault to select is the contradiction an unstated
		// selection cannot be read as: the environment names a key service this
		// composition does not have, so accounts would be delivered the `vault`
		// group's keys with nothing behind them.
		if !composed {
			return "", nil, fmt.Errorf(
				"environment %q selects key-service: %s, but this composition composes no %q service: the %q configuration group is the vault backend's binding (VAULT_ADDR, VAULT_AUTH_METHOD, VAULT_APPROLE_MOUNT and the AppRole credential) and there would be nothing behind it — compose the %q service in module.codefly.yaml `services:`, declare it there as a managed service of kind %q when the cell runs its own, or select key-service: %s",
				environment.Name, keyServiceVault, keyServiceVaultService, keyServiceGroup,
				keyServiceVaultService, managedKindCellVault, keyServiceKMS)
		}
		return keyServiceVault, withdrawn, nil
	case keyServiceKMS:
		// Only a service this composition actually composes can be withdrawn: a
		// consumer that already left `vault` out of its module.codefly.yaml has
		// nothing here to take away.
		if composed {
			withdrawn[keyServiceVaultService] = struct{}{}
		}
		return keyServiceKMS, withdrawn, nil
	default:
		return "", nil, fmt.Errorf(
			"environment %q key-service %q is not supported: it is %q or %q, the same two KEY_SERVICE_BACKEND takes in the `key-service` configuration group",
			environment.Name, environment.KeyService, keyServiceVault, keyServiceKMS)
	}
}

// compositionSelectsVault answers whether the vault key service is selected
// ANYWHERE in this composition, which is the grain the `vault` configuration
// group is declared at.
//
// WHY AN AGGREGATE. The selection is per environment; the group is declared
// once, in a service manifest, for the whole composition. The two grains do not
// match, so the only sound reduction is agreement: the group is withdrawn when
// no environment selects the backend that needs it, and kept the moment one
// does. A composition that runs this module's own Vault locally and the cell's
// cloud key service in its cell therefore keeps the group — accounts needs
// VAULT_ADDR and the token for the local run — and only a composition on the
// cloud key service everywhere drops it. Making the drop per environment means
// delivering a different set of groups per environment, which is the
// configuration plane's to decide and not expressible in a manifest.
func compositionSelectsVault(workspace *workspaceManifest, services []string) (bool, error) {
	asked := false
	for _, environment := range workspace.Environments {
		if environment == nil {
			// selectEnvironments refuses this when the bundle is rendered; here
			// it is simply not an answer.
			continue
		}
		asked = true
		selection, _, err := resolveKeyService(environment, services)
		if err != nil {
			return false, err
		}
		if selection == keyServiceVault {
			return true, nil
		}
	}
	if !asked {
		// Nothing to ask, so the unstated selection answers, exactly as it does
		// for an environment that states none.
		return slices.Contains(services, keyServiceVaultService), nil
	}
	return false, nil
}

// managedKindCellVault names a Vault the cell runs beside this module rather
// than one this module composes. It exists because the module's own `vault`
// service is a development convenience whose state is the product's secrets: a
// cell that holds the signing key and the Transit key wants its own durable
// Vault, and the product must reach that one by name rather than inherit
// whatever the composition happened to render.
//
// Everything about the kind follows from being in-cluster. The destination is
// selected by namespace, because no egress CIDR can describe a pod whose
// address moves. And this handoff projects no connection secret of its own: the
// caller authenticates with a credential the configuration plane delivers into
// its own secret group, which is why the mode must be stated rather than
// inherited from the password default — a default would have the promotion
// driver project a secret nothing issued.
const managedKindCellVault = "cell-vault"

// validateManagedEgress resolves how a caller is granted reach to a managed
// dependency. An out-of-cluster dependency has a stable address, so it is named
// by exact CIDR; an in-cluster one does not, and the pod CIDR that would cover
// it covers every other workload too — so those are selected by namespace, and
// mixing the two spellings is an error rather than a union.
func validateManagedEgress(environment, service string, config managedServiceConfig) (namespace string, err error) {
	namespace = strings.TrimSpace(config.EgressNamespace)
	if config.Kind != managedKindCellVault {
		if namespace != "" {
			return "", fmt.Errorf("environment %q managed service %q kind %q does not take an egress-namespace: it is reached by exact egress-cidrs", environment, service, config.Kind)
		}
		return "", nil
	}
	if namespace == "" {
		return "", fmt.Errorf("environment %q managed service %q kind %q requires an egress-namespace: the namespace holding the cell's Vault, which is what the caller's egress policy selects on", environment, service, config.Kind)
	}
	if err := validateDNSLabel("managed service egress namespace", namespace); err != nil {
		return "", fmt.Errorf("environment %q managed service %q: %w", environment, service, err)
	}
	if len(config.EgressCIDRs) > 0 {
		return "", fmt.Errorf("environment %q managed service %q kind %q is in-cluster and declares egress-cidrs: a pod address is not stable and the pod CIDR would authorize every workload in the cluster", environment, service, config.Kind)
	}
	return namespace, nil
}

// validateManagedAuth resolves how callers authenticate to a managed service.
// An unset auth-mode keeps the password-backed shape the password-bearing kinds
// have always had; external-identity means the workload authenticates as itself,
// so there is no connection secret to project and none may be declared.
func validateManagedAuth(environment, service string, config managedServiceConfig) (mode, instance string, err error) {
	instance = strings.TrimSpace(config.InstanceConnectionName)
	mode = strings.TrimSpace(config.AuthMode)
	if config.Kind == managedKindCellVault {
		// A long-lived Vault token rendered beside the Service is the shape this
		// kind exists to replace: it is the same credential for every replica and
		// it outlives the pod that read it. The caller logs in for a short-lived
		// token of its own instead, with a credential delivered through its own
		// secret group, so this handoff renders none and the mode is stated
		// rather than inherited from the password default.
		if mode != "external-identity" {
			return "", "", fmt.Errorf("environment %q managed service %q kind %q requires auth-mode: external-identity — this handoff projects no connection secret, because the caller logs in with a credential the configuration plane delivers into its own secret group", environment, service, config.Kind)
		}
	}
	if config.Kind == "cloud-sql-postgres" {
		// A silent password default here would have the driver project a
		// connection secret an IAM-only instance never issued, so the choice is
		// stated rather than inherited.
		if mode == "" {
			return "", "", fmt.Errorf("environment %q managed service %q kind %q requires an explicit auth-mode", environment, service, config.Kind)
		}
		if !instanceConnectionPattern.MatchString(instance) {
			return "", "", fmt.Errorf("environment %q managed service %q instance-connection-name %q is not a project:region:instance name", environment, service, config.InstanceConnectionName)
		}
	} else if instance != "" {
		return "", "", fmt.Errorf("environment %q managed service %q kind %q does not take an instance-connection-name", environment, service, config.Kind)
	}
	switch mode {
	case "", "password":
		return "password", instance, nil
	case "external-identity":
		if len(config.SecretReferences) > 0 {
			return "", "", fmt.Errorf("environment %q managed service %q authenticates as an external identity and declares secret references", environment, service)
		}
		return "external-identity", instance, nil
	}
	return "", "", fmt.Errorf("environment %q managed service %q auth-mode %q is not supported", environment, service, config.AuthMode)
}

func validateManagedDependencyCIDRs(environment string, topology deploymentTopology, managed map[string]managedServiceConfig) error {
	for _, service := range topology.Services {
		for _, dependency := range service.Dependencies {
			config, exists := managed[dependency.Service]
			if !exists {
				continue
			}
			// An in-cluster kind is reached by namespace, which
			// validateManagedEgress already required; demanding a CIDR as well
			// would make the kind undeclarable.
			if config.EgressNamespace != "" {
				continue
			}
			if len(config.EgressCIDRs) == 0 {
				return fmt.Errorf(
					"environment %q managed service %q requires at least one exact egress CIDR for caller %q",
					environment,
					dependency.Service,
					service.Name,
				)
			}
		}
	}
	return nil
}

func validExternalName(value string) bool {
	name := strings.TrimSuffix(strings.TrimSpace(value), ".")
	if name == "" || net.ParseIP(name) != nil || len(name) > 253 {
		return false
	}
	for _, label := range strings.Split(name, ".") {
		if len(label) > 63 || !dnsLabelPattern.MatchString(label) {
			return false
		}
	}
	return true
}

func classifyEnvironment(environment *environmentConfig) (local bool, managedCapable bool, cluster string, err error) {
	kind := strings.TrimSpace(environment.Cluster.Kind)
	switch kind {
	case "k3d":
		return true, false, kind, nil
	case "eks", "aks", "gke":
		return false, true, kind, nil
	case "":
		if strings.HasPrefix(environment.Name, "local") {
			return true, false, "k3d", nil
		}
	}
	return false, false, "", fmt.Errorf("environment %q cluster kind %q is not supported by the SaaS deployment generator", environment.Name, kind)
}

func renderEnvironment(
	root,
	workspaceName,
	moduleName string,
	topology deploymentTopology,
	plan environmentPlan,
) error {
	environmentRoot := filepath.Join(root, "overlays", plan.environment.Name)
	baseRoot := filepath.Join(environmentRoot, "base")
	if err := os.MkdirAll(baseRoot, 0o755); err != nil {
		return err
	}

	identityLabels := map[string]string{
		"app.kubernetes.io/managed-by": "codefly",
		"app.kubernetes.io/part-of":    kubernetesName(workspaceName, moduleName),
		"codefly.dev/workspace":        workspaceName,
		"codefly.dev/module":           moduleName,
		"codefly.dev/environment":      plan.environment.Name,
	}
	// The Namespace object carries the environment identity, so — like the
	// host-bearing Gateway and VirtualService — it lives in the overlay, not the
	// base. The base stays identity-neutral (no Namespace object, no host); the
	// overlay names the namespace both on this object and, through its namespace
	// transformer, on every resource the base contributes.
	namespaceLabels := map[string]string{
		"istio.io/dataplane-mode":                    "ambient",
		"pod-security.kubernetes.io/enforce":         "baseline",
		"pod-security.kubernetes.io/enforce-version": "latest",
		"pod-security.kubernetes.io/audit":           "restricted",
		"pod-security.kubernetes.io/audit-version":   "latest",
		"pod-security.kubernetes.io/warn":            "restricted",
		"pod-security.kubernetes.io/warn-version":    "latest",
		"kubernetes.io/metadata.name":                plan.environment.Namespace,
	}
	if topologyRequiresWaypoint(topology, plan) {
		// Route service traffic through the namespace waypoint so the L7
		// internal-authority AuthorizationPolicy is enforced in the ambient mesh.
		namespaceLabels["istio.io/use-waypoint"] = meshWaypointName
	}
	namespace := kubeObject{
		APIVersion: "v1",
		Kind:       "Namespace",
		Metadata: objectMeta{
			Name:   plan.environment.Namespace,
			Labels: mergeLabels(identityLabels, namespaceLabels),
		},
	}
	if err := writeYAML(filepath.Join(environmentRoot, "namespace.yaml"), namespace); err != nil {
		return err
	}
	basePaths, ingress, err := renderSharedResources(baseRoot, plan.environment.Namespace, identityLabels, topology, plan)
	if err != nil {
		return err
	}

	baseResources := []string{
		"resource-quota.yaml",
		"limit-range.yaml",
	}
	baseResources = append(baseResources, basePaths...)
	if err := writeYAML(filepath.Join(baseRoot, "kustomization.yaml"), kustomization{
		APIVersion: "kustomize.config.k8s.io/v1beta1",
		Kind:       "Kustomization",
		Resources:  baseResources,
	}); err != nil {
		return err
	}

	overlayResources := []string{"base", "namespace.yaml"}
	if len(ingress) > 0 {
		if err := writeYAMLDocuments(filepath.Join(environmentRoot, "ingress.yaml"), ingress); err != nil {
			return err
		}
		overlayResources = append(overlayResources, "ingress.yaml")
	}
	if err := writeYAML(filepath.Join(environmentRoot, "kustomization.yaml"), kustomization{
		APIVersion: "kustomize.config.k8s.io/v1beta1",
		Kind:       "Kustomization",
		Namespace:  plan.environment.Namespace,
		Resources:  overlayResources,
	}); err != nil {
		return err
	}
	return nil
}

func renderSharedResources(
	root,
	namespace string,
	labels map[string]string,
	topology deploymentTopology,
	plan environmentPlan,
) ([]string, []kubeObject, error) {
	name := labels["app.kubernetes.io/part-of"]
	quota := kubeObject{
		APIVersion: "v1",
		Kind:       "ResourceQuota",
		Metadata:   objectMeta{Name: name, Namespace: namespace, Labels: labels},
		Spec: map[string]any{"hard": map[string]string{
			"configmaps":             "60",
			"limits.cpu":             "16",
			"limits.memory":          "32Gi",
			"persistentvolumeclaims": "20",
			"pods":                   "60",
			"requests.cpu":           "8",
			"requests.memory":        "16Gi",
			"requests.storage":       "100Gi",
			"secrets":                "60",
			"services":               "30",
		}},
	}
	if err := writeYAML(filepath.Join(root, "resource-quota.yaml"), quota); err != nil {
		return nil, nil, err
	}
	limit := kubeObject{
		APIVersion: "v1",
		Kind:       "LimitRange",
		Metadata:   objectMeta{Name: name, Namespace: namespace, Labels: labels},
		Spec: map[string]any{"limits": []any{
			map[string]any{
				"type":           "Container",
				"default":        map[string]string{"cpu": "500m", "memory": "512Mi"},
				"defaultRequest": map[string]string{"cpu": "100m", "memory": "128Mi"},
				"max":            map[string]string{"cpu": "4", "memory": "8Gi"},
				"min":            map[string]string{"cpu": "10m", "memory": "16Mi"},
			},
			map[string]any{
				"type": "PersistentVolumeClaim",
				"max":  map[string]string{"storage": "50Gi"},
				"min":  map[string]string{"storage": "1Gi"},
			},
		}},
	}
	if err := writeYAML(filepath.Join(root, "limit-range.yaml"), limit); err != nil {
		return nil, nil, err
	}
	networkPolicies, err := topologyNetworkPolicies(topology, plan, namespace, labels)
	if err != nil {
		return nil, nil, err
	}
	if err := writeYAMLDocuments(filepath.Join(root, "network-policy.yaml"), networkPolicies); err != nil {
		return nil, nil, err
	}
	istio, gateway, err := topologyIstioResources(topology, plan, namespace, labels)
	if err != nil {
		return nil, nil, err
	}
	if err := writeYAMLDocuments(filepath.Join(root, "istio-mtls.yaml"), istio); err != nil {
		return nil, nil, err
	}
	paths := []string{"network-policy.yaml", "istio-mtls.yaml"}
	// The Gateway and VirtualService carry the environment's public host, so they
	// belong in the overlay rather than the base and are handed back to the
	// caller. Their DestinationRules carry no public host and stay in the base
	// with the rest of the mesh baseline.
	var ingress []kubeObject
	var destinationRules []kubeObject
	for _, object := range gateway {
		switch object.Kind {
		case "Gateway", "VirtualService":
			ingress = append(ingress, object)
		default:
			destinationRules = append(destinationRules, object)
		}
	}
	if len(destinationRules) > 0 {
		if err := writeYAMLDocuments(filepath.Join(root, "destination-rules.yaml"), destinationRules); err != nil {
			return nil, nil, err
		}
		paths = append(paths, "destination-rules.yaml")
	}
	handoffPaths, err := renderManagedServiceHandoffs(root, namespace, labels, topology, plan)
	if err != nil {
		return nil, nil, err
	}
	paths = append(paths, handoffPaths...)
	return paths, ingress, nil
}

func topologyIstioResources(
	topology deploymentTopology,
	plan environmentPlan,
	namespace string,
	labels map[string]string,
) ([]kubeObject, []kubeObject, error) {
	name := labels["app.kubernetes.io/part-of"]
	istio := []kubeObject{
		{
			APIVersion: "security.istio.io/v1",
			Kind:       "PeerAuthentication",
			Metadata:   objectMeta{Name: name + "-mtls", Namespace: namespace, Labels: labels},
			Spec:       map[string]any{"mtls": map[string]string{"mode": "STRICT"}},
		},
		{
			APIVersion: "security.istio.io/v1",
			Kind:       "AuthorizationPolicy",
			Metadata:   objectMeta{Name: "default-deny", Namespace: namespace, Labels: labels},
			Spec:       map[string]any{},
		},
	}
	istio = append(istio, topologyInternalAuthorizationPolicies(topology, plan, namespace, labels)...)
	istio = append(istio, topologyInternalAuthorityAllowPolicies(topology, plan, namespace, labels)...)
	istio = append(istio, topologyInternalHTTPDenyPolicies(topology, plan, namespace, labels)...)
	if topologyRequiresWaypoint(topology, plan) {
		// The internal-authority allow and the internal-HTTP deny both match on
		// request path. In the ambient data plane ztunnel enforces L4 only, so a
		// waypoint proxy must front the namespace for that match to be evaluated.
		// The namespace opts in via istio.io/use-waypoint (see renderEnvironment).
		istio = append(istio, kubeObject{
			APIVersion: "gateway.networking.k8s.io/v1",
			Kind:       "Gateway",
			Metadata: objectMeta{
				Name:      meshWaypointName,
				Namespace: namespace,
				Labels:    mergeLabels(labels, map[string]string{"istio.io/waypoint-for": "service"}),
			},
			Spec: map[string]any{
				"gatewayClassName": "istio-waypoint",
				"listeners": []any{map[string]any{
					"name":     "mesh",
					"port":     15008,
					"protocol": "HBONE",
				}},
			},
		})
	}
	if len(plan.ingress) == 0 {
		// No public ingress: emit the mTLS baseline only, no Gateway.
		return istio, nil, nil
	}
	ingressPorts := make(map[string]map[uint32]struct{})
	for _, route := range plan.ingress {
		if ingressPorts[route.service] == nil {
			ingressPorts[route.service] = make(map[uint32]struct{})
		}
		ingressPorts[route.service][route.port] = struct{}{}
	}
	ingressServices := make([]string, 0, len(ingressPorts))
	for service := range ingressPorts {
		ingressServices = append(ingressServices, service)
	}
	slices.Sort(ingressServices)
	for _, service := range ingressServices {
		topologyService, _ := topologyServiceByName(topology, service)
		ports := make([]uint32, 0, len(ingressPorts[service]))
		for port := range ingressPorts[service] {
			ports = append(ports, port)
		}
		slices.Sort(ports)
		renderedPorts := make([]string, 0, len(ports))
		for _, port := range ports {
			renderedPorts = append(renderedPorts, strconv.FormatUint(uint64(port), 10))
		}
		istio = append(istio, kubeObject{
			APIVersion: "security.istio.io/v1",
			Kind:       "AuthorizationPolicy",
			Metadata: objectMeta{
				Name:      "allow-istio-ingress-to-" + service,
				Namespace: namespace,
				Labels:    labels,
			},
			Spec: map[string]any{
				"selector": map[string]any{"matchLabels": map[string]string{"app": topologyKubernetesAppLabel(topologyService)}},
				"rules": []any{map[string]any{
					"from": []any{map[string]any{"source": map[string]any{
						"principals": []string{"cluster.local/ns/istio-system/sa/istio-ingressgateway-service-account"},
					}}},
					"to": []any{map[string]any{"operation": map[string]any{
						"ports": renderedPorts,
					}}},
				}},
			},
		})
	}

	gatewayHosts := []string{"*"}
	var httpRoutes []any
	if len(plan.ingress[0].hosts) == 0 {
		route := plan.ingress[0]
		service, _ := topologyServiceByName(topology, route.service)
		httpRoutes = []any{map[string]any{
			"name":  route.name,
			"match": []any{map[string]any{"uri": map[string]string{"prefix": "/"}}},
			"route": []any{map[string]any{"destination": map[string]any{
				"host": topologyKubernetesServiceName(service) + "." + namespace + ".svc.cluster.local",
				"port": map[string]any{"number": route.servicePort},
			}}},
			"timeout": "30s",
		}}
	} else {
		gatewayHosts = nil
		hostSet := make(map[string]struct{})
		for _, route := range plan.ingress {
			service, _ := topologyServiceByName(topology, route.service)
			matches := make([]any, 0, len(route.hosts))
			for _, host := range route.hosts {
				if _, exists := hostSet[host]; !exists {
					hostSet[host] = struct{}{}
					gatewayHosts = append(gatewayHosts, host)
				}
				matches = append(matches, map[string]any{
					"authority": map[string]string{
						"regex": "^" + regexp.QuoteMeta(host) + "(:[0-9]+)?$",
					},
				})
			}
			httpRoutes = append(httpRoutes, map[string]any{
				"name":  route.name,
				"match": matches,
				"route": []any{map[string]any{"destination": map[string]any{
					"host": topologyKubernetesServiceName(service) + "." + namespace + ".svc.cluster.local",
					"port": map[string]any{"number": route.servicePort},
				}}},
				"timeout": "30s",
			})
		}
		slices.Sort(gatewayHosts)
	}
	gatewayName := topology.Module.Name
	gateway := []kubeObject{
		{
			APIVersion: "networking.istio.io/v1",
			Kind:       "Gateway",
			Metadata:   objectMeta{Name: gatewayName, Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"selector": map[string]string{"istio": "ingressgateway"},
				"servers": []any{map[string]any{
					"port":  map[string]any{"number": 80, "name": "http", "protocol": "HTTP"},
					"hosts": gatewayHosts,
				}},
			},
		},
		{
			APIVersion: "networking.istio.io/v1",
			Kind:       "VirtualService",
			Metadata:   objectMeta{Name: name, Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"hosts":    gatewayHosts,
				"gateways": []string{gatewayName},
				"http":     httpRoutes,
			},
		},
	}
	for _, service := range plan.services {
		topologyService, _ := topologyServiceByName(topology, service)
		gateway = append(gateway, kubeObject{
			APIVersion: "networking.istio.io/v1",
			Kind:       "DestinationRule",
			Metadata: objectMeta{
				Name:      kubernetesName(name, service, "defaults"),
				Namespace: namespace,
				Labels:    labels,
			},
			Spec: map[string]any{
				"host":     topologyKubernetesServiceName(topologyService) + "." + namespace + ".svc.cluster.local",
				"exportTo": []string{"."},
				"trafficPolicy": map[string]any{
					"connectionPool": map[string]any{
						"tcp":  map[string]any{"maxConnections": 100},
						"http": map[string]any{"http2MaxRequests": 1000, "maxRequestsPerConnection": 10, "maxRetries": 3},
					},
					"outlierDetection": map[string]any{
						"consecutive5xxErrors": 5,
						"interval":             "10s",
						"baseEjectionTime":     "30s",
						"maxEjectionPercent":   50,
					},
				},
			},
		})
	}
	return istio, gateway, nil
}

// topologyServiceAccount returns the Kubernetes ServiceAccount a service's pods
// run under: spec.service-account.name when the service declares one (the
// go-grpc agent renders the SA object and binds serviceAccountName off it),
// otherwise the namespace default. A caller without a distinct SA (e.g. the
// nextjs frontend, whose agent does not render one yet) stays on "default".
func topologyServiceAccount(service topologyService) string {
	spec, ok := service.Spec["service-account"].(map[string]any)
	if !ok {
		return "default"
	}
	if name, ok := spec["name"].(string); ok && name != "" {
		return name
	}
	return "default"
}

func meshPrincipal(namespace, serviceAccount string) string {
	return "cluster.local/ns/" + namespace + "/sa/" + serviceAccount
}

// topologyTargetCallerPrincipals maps each in-cluster dependency target to the
// sorted, de-duplicated SPIFFE principals of its declared callers — each
// caller's workload SA (or sa/default when it has none). This is the positive
// allowlist the internal-authority AuthorizationPolicies gate on: only a
// target's declared callers may reach it, deny-by-default for everyone else
// (the ingress gateway included), instead of the fail-open notPrincipals
// exemption of the shared sa/default.
func topologyTargetCallerPrincipals(topology deploymentTopology, plan environmentPlan, namespace string) map[string][]string {
	inCluster := make(map[string]struct{}, len(plan.services))
	for _, service := range plan.services {
		inCluster[service] = struct{}{}
	}
	callers := make(map[string]map[string]struct{})
	addCaller := func(target, principal string) {
		if callers[target] == nil {
			callers[target] = make(map[string]struct{})
		}
		callers[target][principal] = struct{}{}
	}
	for _, caller := range topology.Services {
		if _, ok := inCluster[caller.Name]; !ok {
			continue
		}
		principal := meshPrincipal(namespace, topologyServiceAccount(caller))
		for _, dependency := range caller.Dependencies {
			if _, ok := inCluster[dependency.Service]; !ok {
				continue
			}
			addCaller(dependency.Service, principal)
		}
	}
	// A service's bootstrap job runs under the service's own SA and reaches its
	// own endpoints, so it is a declared caller of itself.
	for _, service := range topology.Services {
		if _, ok := inCluster[service.Name]; !ok {
			continue
		}
		if len(service.BootstrapJobEndpoints) > 0 {
			addCaller(service.Name, meshPrincipal(namespace, topologyServiceAccount(service)))
		}
	}
	result := make(map[string][]string, len(callers))
	for target, set := range callers {
		principals := make([]string, 0, len(set))
		for principal := range set {
			principals = append(principals, principal)
		}
		slices.Sort(principals)
		result[target] = principals
	}
	return result
}

func topologyInternalAuthorizationPolicies(
	topology deploymentTopology,
	plan environmentPlan,
	namespace string,
	labels map[string]string,
) []kubeObject {
	inCluster := make(map[string]struct{}, len(plan.services))
	for _, service := range plan.services {
		inCluster[service] = struct{}{}
	}
	targetPorts := make(map[string]map[uint32]struct{})
	addPort := func(service string, port uint32) {
		if targetPorts[service] == nil {
			targetPorts[service] = make(map[uint32]struct{})
		}
		targetPorts[service][port] = struct{}{}
	}
	for _, caller := range topology.Services {
		if _, exists := inCluster[caller.Name]; !exists {
			continue
		}
		for _, dependency := range caller.Dependencies {
			if _, exists := inCluster[dependency.Service]; !exists {
				continue
			}
			target, _ := topologyServiceByName(topology, dependency.Service)
			for _, endpointName := range dependency.Endpoints {
				endpoint, _ := topologyEndpointByName(target, endpointName)
				addPort(target.Name, endpoint.Port)
			}
		}
	}
	for _, service := range topology.Services {
		if _, exists := inCluster[service.Name]; !exists {
			continue
		}
		for _, endpointName := range service.BootstrapJobEndpoints {
			endpoint, _ := topologyEndpointByName(service, endpointName)
			addPort(service.Name, endpoint.Port)
		}
	}

	targets := make([]string, 0, len(targetPorts))
	for target := range targetPorts {
		targets = append(targets, target)
	}
	slices.Sort(targets)
	policies := make([]kubeObject, 0, len(targets))
	callerPrincipals := topologyTargetCallerPrincipals(topology, plan, namespace)
	// Once a waypoint fronts the namespace, east-west traffic reaches the
	// destination's ztunnel from the waypoint, not from the caller, so the
	// caller-scoped allowlist below would drop every in-mesh request unless the
	// waypoint's own identity is admitted too. It runs under the ServiceAccount
	// named after its Gateway. Caller identity is still enforced above it, by the
	// L7 policies the waypoint evaluates.
	waypointPrincipal := ""
	if topologyRequiresWaypoint(topology, plan) {
		waypointPrincipal = meshPrincipal(namespace, meshWaypointName)
	}
	for _, target := range targets {
		service, _ := topologyServiceByName(topology, target)
		ports := make([]uint32, 0, len(targetPorts[target]))
		for port := range targetPorts[target] {
			ports = append(ports, port)
		}
		slices.Sort(ports)
		renderedPorts := make([]string, 0, len(ports))
		for _, port := range ports {
			renderedPorts = append(renderedPorts, strconv.FormatUint(uint64(port), 10))
		}
		policies = append(policies, kubeObject{
			APIVersion: "security.istio.io/v1",
			Kind:       "AuthorizationPolicy",
			Metadata: objectMeta{
				Name:      "allow-" + target + "-from-internal-topology",
				Namespace: namespace,
				Labels:    labels,
			},
			Spec: map[string]any{
				"selector": map[string]any{"matchLabels": map[string]string{"app": topologyKubernetesAppLabel(service)}},
				"rules": []any{map[string]any{
					"from": []any{map[string]any{"source": map[string]any{
						"principals": portAllowPrincipals(callerPrincipals[target], waypointPrincipal),
					}}},
					"to": []any{map[string]any{"operation": map[string]any{
						"ports": renderedPorts,
					}}},
				}},
			},
		})
	}
	return policies
}

// portAllowPrincipals returns the L4 allowlist for a target: its declared
// callers plus, when one fronts the namespace, the waypoint that relays their
// traffic.
func portAllowPrincipals(callers []string, waypointPrincipal string) []string {
	if waypointPrincipal == "" {
		return callers
	}
	principals := append(slices.Clone(callers), waypointPrincipal)
	slices.Sort(principals)
	return slices.Compact(principals)
}

// topologyInternalHTTPDenyPolicies gates a service's cluster-internal HTTP
// routes by caller workload identity. Those routes are multiplexed on a public
// endpoint's port with browser-facing traffic, so no NetworkPolicy and no L4
// mesh rule can separate them — only a path-and-method match can, and only a
// DENY can subtract them from the port-wide ALLOW that admits the ingress
// gateway. Every principal outside the target's declared callers is refused, so
// the routes are reachable from inside the mesh and from nowhere else; a
// service with no declared callers denies them outright rather than leaving
// them on the public front door.
func topologyInternalHTTPDenyPolicies(
	topology deploymentTopology,
	plan environmentPlan,
	namespace string,
	labels map[string]string,
) []kubeObject {
	inCluster := make(map[string]struct{}, len(plan.services))
	for _, service := range plan.services {
		inCluster[service] = struct{}{}
	}
	callerPrincipals := topologyTargetCallerPrincipals(topology, plan, namespace)
	services := make([]topologyService, 0, len(topology.Services))
	for _, service := range topology.Services {
		if _, ok := inCluster[service.Name]; ok && len(service.InternalHTTPRoutes) > 0 {
			services = append(services, service)
		}
	}
	slices.SortFunc(services, func(a, b topologyService) int { return strings.Compare(a.Name, b.Name) })
	policies := make([]kubeObject, 0, len(services))
	for _, service := range services {
		exempt := append(slices.Clone(callerPrincipals[service.Name]), meshIngressPrincipal)
		slices.Sort(exempt)
		rules := make([]any, 0, len(service.InternalHTTPRoutes))
		for _, route := range service.InternalHTTPRoutes {
			rules = append(rules, map[string]any{
				"from": []any{map[string]any{"source": map[string]any{
					"notPrincipals": exempt,
				}}},
				"to": []any{map[string]any{"operation": map[string]any{
					"paths":   internalHTTPPathPatterns(route.Path),
					"methods": route.Methods,
				}}},
			})
		}
		policies = append(policies, kubeObject{
			APIVersion: "security.istio.io/v1",
			Kind:       "AuthorizationPolicy",
			Metadata: objectMeta{
				Name:      "deny-" + service.Name + "-internal-http",
				Namespace: namespace,
				Labels:    labels,
			},
			Spec: map[string]any{
				"targetRefs": []any{map[string]any{
					"kind":  "Service",
					"group": "",
					"name":  topologyKubernetesServiceName(service),
				}},
				"action": "DENY",
				"rules":  rules,
			},
		})
	}
	return policies
}

// internalHTTPPathPatterns expands an authored route into the Istio path
// patterns that match it. The suffix form is what closes the gap between what
// Envoy matches and what the application routes: Istio's default path
// normalization does not merge duplicate slashes, so an exact match alone lets
// //api/... through while the application still serves it. Istio allows one
// wildcard, at the start or the end, so a suffix pattern is the widest single
// form available; percent-encoded separators are matchable by neither, which is
// why the route's own credential check remains the gate rather than a backstop.
func internalHTTPPathPatterns(path string) []string {
	return []string{"*" + path, path}
}

// meshIngressPrincipal is the identity Istio's ingress gateway presents. It is
// exempt from the internal-HTTP deny because a waypoint never sees
// ingress-originated traffic: Istio routes it straight to the workload unless
// the install sets istio.io/ingress-use-waypoint AND istiod's
// ENABLE_INGRESS_WAYPOINT_ROUTING, neither of which a module can render. Naming
// it in the deny would read as a boundary while enforcing nothing.
const meshIngressPrincipal = "cluster.local/ns/istio-system/sa/istio-ingressgateway-service-account"

// meshWaypointName is the namespace waypoint that fronts the ambient data plane
// so L7 (method-path) AuthorizationPolicies are enforced; ztunnel alone is L4.
const meshWaypointName = "waypoint"

// topologyRequiresWaypoint reports whether any in-cluster service exposes an
// internal surface — gRPC authority methods or authored HTTP routes — gated by
// an L7 AuthorizationPolicy that only a waypoint can evaluate.
func topologyRequiresWaypoint(topology deploymentTopology, plan environmentPlan) bool {
	inCluster := make(map[string]struct{}, len(plan.services))
	for _, service := range plan.services {
		inCluster[service] = struct{}{}
	}
	for target := range topology.internalMethodPaths {
		if _, ok := inCluster[target]; ok {
			return true
		}
	}
	for _, service := range topology.Services {
		if _, ok := inCluster[service.Name]; !ok {
			continue
		}
		if len(service.InternalHTTPRoutes) > 0 {
			return true
		}
	}
	return false
}

// topologyInternalAuthorityAllowPolicies gates each service's internal authority
// method paths by caller workload identity with a positive ALLOW: only a
// target's declared callers (named by their per-service ServiceAccount) may
// reach the internal RPC paths, deny-by-default for every other principal — the
// ingress gateway and any unlisted namespace workload included. This replaces
// the earlier notPrincipals:[sa/default] DENY, which fail-open: with every pod
// sharing sa/default, that exemption admitted the whole namespace. Istio makes a
// selected workload default-deny once any ALLOW rule targets it, so an unlisted
// SA is rejected at the mesh before the handler; the app-layer internal
// credential remains the identity gate. The L7 path match needs the namespace
// waypoint (see topologyRequiresWaypoint); the L4 port ALLOW
// (topologyInternalAuthorizationPolicies) is now caller-scoped too, so even
// without the waypoint the target is reachable only by its declared callers.
func topologyInternalAuthorityAllowPolicies(
	topology deploymentTopology,
	plan environmentPlan,
	namespace string,
	labels map[string]string,
) []kubeObject {
	inCluster := make(map[string]struct{}, len(plan.services))
	for _, service := range plan.services {
		inCluster[service] = struct{}{}
	}
	callerPrincipals := topologyTargetCallerPrincipals(topology, plan, namespace)
	targets := make([]string, 0, len(topology.internalMethodPaths))
	for target := range topology.internalMethodPaths {
		if _, ok := inCluster[target]; ok {
			targets = append(targets, target)
		}
	}
	slices.Sort(targets)
	policies := make([]kubeObject, 0, len(targets))
	for _, target := range targets {
		service, _ := topologyServiceByName(topology, target)
		policies = append(policies, kubeObject{
			APIVersion: "security.istio.io/v1",
			Kind:       "AuthorizationPolicy",
			Metadata: objectMeta{
				Name:      "allow-" + target + "-internal-authority",
				Namespace: namespace,
				Labels:    labels,
			},
			Spec: map[string]any{
				"targetRefs": []any{map[string]any{
					"kind":  "Service",
					"group": "",
					"name":  topologyKubernetesServiceName(service),
				}},
				"action": "ALLOW",
				"rules": []any{map[string]any{
					"from": []any{map[string]any{"source": map[string]any{
						"principals": callerPrincipals[target],
					}}},
					"to": []any{map[string]any{"operation": map[string]any{
						"paths": topology.internalMethodPaths[target],
					}}},
				}},
			},
		})
	}
	return policies
}

func loadDeploymentTopology(moduleDir, moduleName string, services []serviceDefinition) (deploymentTopology, error) {
	topology, err := assembleDeploymentTopology(moduleDir, moduleName, services)
	if err != nil {
		return deploymentTopology{}, err
	}
	declared := make(map[string]struct{}, len(services))
	for _, service := range services {
		declared[service.name] = struct{}{}
	}
	topologyServices := make(map[string]topologyService, len(topology.Services))
	kubernetesServices := make(map[string]string, len(topology.Services))
	kubernetesApps := make(map[string]string, len(topology.Services))
	for _, service := range topology.Services {
		if _, exists := topologyServices[service.Name]; exists {
			return deploymentTopology{}, fmt.Errorf("deployment topology repeats service %q", service.Name)
		}
		if _, exists := declared[service.Name]; !exists {
			return deploymentTopology{}, fmt.Errorf("deployment topology contains undeclared service %q", service.Name)
		}
		if len(service.Endpoints) == 0 {
			return deploymentTopology{}, fmt.Errorf("deployment topology service %q declares no endpoints", service.Name)
		}
		kubernetesService := topologyKubernetesServiceName(service)
		if err := validateDNSLabel("deployment topology Kubernetes service name", kubernetesService); err != nil {
			return deploymentTopology{}, fmt.Errorf("service %q: %w", service.Name, err)
		}
		if owner, exists := kubernetesServices[kubernetesService]; exists {
			return deploymentTopology{}, fmt.Errorf(
				"deployment topology services %q and %q share Kubernetes service name %q",
				owner,
				service.Name,
				kubernetesService,
			)
		}
		kubernetesServices[kubernetesService] = service.Name
		kubernetesApp := topologyKubernetesAppLabel(service)
		if err := validateDNSLabel("deployment topology Kubernetes app label", kubernetesApp); err != nil {
			return deploymentTopology{}, fmt.Errorf("service %q: %w", service.Name, err)
		}
		if owner, exists := kubernetesApps[kubernetesApp]; exists {
			return deploymentTopology{}, fmt.Errorf(
				"deployment topology services %q and %q share Kubernetes app label %q",
				owner,
				service.Name,
				kubernetesApp,
			)
		}
		kubernetesApps[kubernetesApp] = service.Name
		endpoints := make(map[string]struct{}, len(service.Endpoints))
		for _, endpoint := range service.Endpoints {
			if endpoint.Port == 0 || endpoint.Port > 65535 || endpoint.ServicePort == 0 || endpoint.ServicePort > 65535 {
				return deploymentTopology{}, fmt.Errorf("deployment topology endpoint %s/%s has invalid port", service.Name, endpoint.Name)
			}
			if _, exists := endpoints[endpoint.Name]; exists {
				return deploymentTopology{}, fmt.Errorf("deployment topology repeats endpoint %s/%s", service.Name, endpoint.Name)
			}
			endpoints[endpoint.Name] = struct{}{}
		}
		for _, endpoint := range service.BootstrapJobEndpoints {
			if _, exists := endpoints[endpoint]; !exists {
				return deploymentTopology{}, fmt.Errorf("deployment topology service %q bootstrap Job references missing endpoint %q", service.Name, endpoint)
			}
		}
		topologyServices[service.Name] = service
	}
	for service := range declared {
		if _, exists := topologyServices[service]; !exists {
			return deploymentTopology{}, fmt.Errorf("deployment topology is missing declared service %q", service)
		}
	}
	if _, exists := topologyServices[topology.Module.ServiceEntry]; !exists {
		return deploymentTopology{}, fmt.Errorf("deployment topology service entry %q is not declared", topology.Module.ServiceEntry)
	}
	for _, service := range topology.Services {
		for _, dependency := range service.Dependencies {
			target, exists := topologyServices[dependency.Service]
			if !exists {
				return deploymentTopology{}, fmt.Errorf("deployment topology service %q references undeclared dependency %q", service.Name, dependency.Service)
			}
			if len(dependency.Endpoints) == 0 {
				return deploymentTopology{}, fmt.Errorf(
					"deployment topology service %q dependency %q declares no endpoints",
					service.Name,
					dependency.Service,
				)
			}
			for _, endpoint := range dependency.Endpoints {
				if _, exists := topologyEndpointByName(target, endpoint); !exists {
					return deploymentTopology{}, fmt.Errorf("deployment topology service %q references missing endpoint %s/%s", service.Name, dependency.Service, endpoint)
				}
			}
		}
	}
	if err := validateDeployJobs(moduleDir, topology, topologyServices); err != nil {
		return deploymentTopology{}, err
	}
	internalMethodPaths, err := loadInternalMethodPaths(moduleDir, services)
	if err != nil {
		return deploymentTopology{}, err
	}
	topology.internalMethodPaths = internalMethodPaths
	return topology, nil
}

// validateDeployJobs checks that each deploy Job runs a declared service, ships
// a command, and writes to an endpoint the running service already declares as
// a dependency — so the mesh already permits the reach and only the Job pod's
// distinct identity needs its own policy. The catalog artifact must exist on
// disk: the generated document is the Job's only input, and a missing file
// would ship a Job the environment can never satisfy.
func validateDeployJobs(moduleDir string, topology deploymentTopology, services map[string]topologyService) error {
	names := make(map[string]struct{}, len(topology.DeployJobs))
	for _, job := range topology.DeployJobs {
		if err := validateDNSLabel("deployment topology deploy Job name", job.Name); err != nil {
			return err
		}
		if _, repeated := names[job.Name]; repeated {
			return fmt.Errorf("deployment topology repeats deploy Job %q", job.Name)
		}
		names[job.Name] = struct{}{}
		runner, exists := services[job.Service]
		if !exists {
			return fmt.Errorf("deployment topology deploy Job %q references undeclared service %q", job.Name, job.Service)
		}
		if job.Command == "" {
			return fmt.Errorf("deployment topology deploy Job %q declares no command", job.Name)
		}
		if job.Catalog == "" {
			return fmt.Errorf("deployment topology deploy Job %q declares no catalog artifact", job.Name)
		}
		catalog, err := cleanRelativeFilesystemPath(job.Catalog)
		if err != nil {
			return fmt.Errorf("deployment topology deploy Job %q catalog: %w", job.Name, err)
		}
		if _, err := os.Stat(filepath.Join(moduleDir, filepath.FromSlash(catalog))); err != nil {
			return fmt.Errorf("deployment topology deploy Job %q catalog artifact %q: %w", job.Name, job.Catalog, err)
		}
		target, exists := services[job.Writes.Service]
		if !exists {
			return fmt.Errorf("deployment topology deploy Job %q writes to undeclared service %q", job.Name, job.Writes.Service)
		}
		if _, exists := topologyEndpointByName(target, job.Writes.Endpoint); !exists {
			return fmt.Errorf("deployment topology deploy Job %q writes to missing endpoint %s/%s", job.Name, job.Writes.Service, job.Writes.Endpoint)
		}
		if !serviceDependsOn(runner, job.Writes.Service, job.Writes.Endpoint) {
			return fmt.Errorf(
				"deployment topology deploy Job %q writes to %s/%s but service %q does not declare that dependency",
				job.Name, job.Writes.Service, job.Writes.Endpoint, job.Service,
			)
		}
		// A write target that declares bootstrap_job_endpoints runs a migration Job
		// before it is ready to be written to; the deploy Job seeds rows that
		// migration creates, so it must be ordered after the target. after is
		// author-supplied, so require the migration-bearing target to appear in it
		// rather than trusting the ordering to be remembered — an import scheduled
		// before the store migration writes against a schema that does not yet exist.
		if len(target.BootstrapJobEndpoints) > 0 && !slices.Contains(job.After, job.Writes.Service) {
			return fmt.Errorf(
				"deployment topology deploy Job %q writes to %s, which runs a migration Job, so it must run after it: add %q to the Job's after list",
				job.Name, job.Writes.Service, job.Writes.Service,
			)
		}
		for _, after := range job.After {
			if _, exists := services[after]; !exists {
				return fmt.Errorf("deployment topology deploy Job %q runs after undeclared service %q", job.Name, after)
			}
		}
	}
	return nil
}

func serviceDependsOn(service topologyService, dependency, endpoint string) bool {
	for _, declared := range service.Dependencies {
		if declared.Service == dependency && slices.Contains(declared.Endpoints, endpoint) {
			return true
		}
	}
	return false
}

// loadInternalMethodPaths reads each service's generated authorization catalog
// and returns the sorted gRPC method paths marked EXPOSURE_INTERNAL. Services
// without a catalog (e.g. non-gRPC agents) contribute nothing.
func loadInternalMethodPaths(moduleDir string, services []serviceDefinition) (map[string][]string, error) {
	paths := make(map[string][]string)
	for _, service := range services {
		file := filepath.Join(moduleDir, "services", service.name, "generated", "authz-methods.json")
		data, err := os.ReadFile(file)
		if os.IsNotExist(err) {
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("read authorization catalog for %q: %w", service.name, err)
		}
		var catalog struct {
			Methods []struct {
				Procedure string `json:"procedure"`
				Policy    struct {
					Exposure string `json:"exposure"`
				} `json:"policy"`
			} `json:"methods"`
		}
		if err := json.Unmarshal(data, &catalog); err != nil {
			return nil, fmt.Errorf("parse authorization catalog for %q: %w", service.name, err)
		}
		var procedures []string
		for _, method := range catalog.Methods {
			if method.Policy.Exposure == "EXPOSURE_INTERNAL" {
				procedures = append(procedures, method.Procedure)
			}
		}
		if len(procedures) > 0 {
			slices.Sort(procedures)
			paths[service.name] = procedures
		}
	}
	return paths, nil
}

func topologyServiceByName(topology deploymentTopology, name string) (topologyService, bool) {
	for _, service := range topology.Services {
		if service.Name == name {
			return service, true
		}
	}
	return topologyService{}, false
}

func topologyKubernetesServiceName(service topologyService) string {
	if service.Kubernetes == nil {
		return service.Name
	}
	return service.Kubernetes.ServiceName
}

func topologyKubernetesAppLabel(service topologyService) string {
	if service.Kubernetes == nil {
		return service.Name
	}
	return service.Kubernetes.AppLabel
}

func topologyEndpointByName(service topologyService, name string) (topologyEndpoint, bool) {
	for _, endpoint := range service.Endpoints {
		if endpoint.Name == name {
			return endpoint, true
		}
	}
	return topologyEndpoint{}, false
}

func topologyNetworkPolicies(
	topology deploymentTopology,
	plan environmentPlan,
	namespace string,
	labels map[string]string,
) ([]kubeObject, error) {
	policies := []kubeObject{
		{
			APIVersion: "networking.k8s.io/v1",
			Kind:       "NetworkPolicy",
			Metadata:   objectMeta{Name: "default-deny-all", Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"podSelector": map[string]any{},
				"policyTypes": []string{"Ingress", "Egress"},
			},
		},
		{
			APIVersion: "networking.k8s.io/v1",
			Kind:       "NetworkPolicy",
			Metadata:   objectMeta{Name: "allow-dns-egress", Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"podSelector": map[string]any{},
				"policyTypes": []string{"Egress"},
				"egress": []any{map[string]any{
					"to": []any{map[string]any{
						"namespaceSelector": map[string]any{"matchLabels": map[string]string{"kubernetes.io/metadata.name": "kube-system"}},
						"podSelector":       map[string]any{"matchLabels": map[string]string{"k8s-app": "kube-dns"}},
					}},
					"ports": []any{
						map[string]any{"protocol": "UDP", "port": 53},
						map[string]any{"protocol": "TCP", "port": 53},
					},
				}},
			},
		},
		{
			APIVersion: "networking.k8s.io/v1",
			Kind:       "NetworkPolicy",
			Metadata:   objectMeta{Name: "allow-istio-control-plane-egress", Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"podSelector": map[string]any{},
				"policyTypes": []string{"Egress"},
				"egress": []any{map[string]any{
					"to": []any{map[string]any{
						"namespaceSelector": map[string]any{"matchLabels": map[string]string{"kubernetes.io/metadata.name": "istio-system"}},
						"podSelector":       map[string]any{"matchLabels": map[string]string{"app": "istiod"}},
					}},
					"ports": []any{
						map[string]any{"protocol": "TCP", "port": 15012},
						map[string]any{"protocol": "TCP", "port": 15014},
					},
				}},
			},
		},
		{
			APIVersion: "networking.k8s.io/v1",
			Kind:       "NetworkPolicy",
			Metadata:   objectMeta{Name: "allow-ambient-hbone-transport", Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"podSelector": map[string]any{},
				"policyTypes": []string{"Ingress", "Egress"},
				"ingress": []any{map[string]any{
					"ports": []any{
						map[string]any{"protocol": "TCP", "port": 15008},
					},
				}},
				"egress": []any{map[string]any{
					"ports": []any{
						map[string]any{"protocol": "TCP", "port": 15008},
					},
				}},
			},
		},
	}
	inCluster := make(map[string]struct{}, len(plan.services))
	for _, service := range plan.services {
		inCluster[service] = struct{}{}
	}
	type edge struct {
		caller    string
		callerApp string
		target    string
		targetApp string
		ports     []uint32
	}
	var edges []edge
	for _, caller := range topology.Services {
		if _, exists := inCluster[caller.Name]; !exists {
			continue
		}
		for _, dependency := range caller.Dependencies {
			target, _ := topologyServiceByName(topology, dependency.Service)
			portSet := make(map[uint32]struct{})
			for _, endpointName := range dependency.Endpoints {
				endpoint, _ := topologyEndpointByName(target, endpointName)
				portSet[endpoint.Port] = struct{}{}
			}
			ports := make([]uint32, 0, len(portSet))
			for port := range portSet {
				ports = append(ports, port)
			}
			slices.Sort(ports)
			edges = append(edges, edge{
				caller:    caller.Name,
				callerApp: topologyKubernetesAppLabel(caller),
				target:    target.Name,
				targetApp: topologyKubernetesAppLabel(target),
				ports:     ports,
			})
		}
	}
	tokenEgress := make(map[string]struct{})
	for _, current := range edges {
		if _, exists := inCluster[current.target]; exists {
			policies = append(policies,
				dependencyIngressPolicy(namespace, labels, current.target, current.targetApp, current.caller, current.callerApp, current.ports),
				dependencyEgressPolicy(namespace, labels, current.caller, current.callerApp, current.target, current.targetApp, current.ports),
			)
			continue
		}
		if _, gone := plan.withdrawn[current.target]; gone {
			// The edge is withdrawn with its target, not left needing a handoff.
			// Under key-service: kms the caller opens no connection at all — the
			// `vault` sealer that builds one is never reached — so granting reach
			// to anything would be authorizing a dial that does not happen.
			continue
		}
		config, managed := plan.managed[current.target]
		if !managed {
			return nil, fmt.Errorf("dependency %s -> %s has no in-cluster service or managed handoff", current.caller, current.target)
		}
		if config.EgressNamespace != "" {
			// An in-cluster dependency the cell owns: the caller reaches the
			// namespace that holds it. Its credential is delivered as a secret,
			// so there is no workload-identity token to fetch and no
			// metadata-endpoint reach is granted.
			policies = append(policies, managedNamespaceEgressPolicy(namespace, labels, current.caller, current.callerApp, current.target, current.ports, config.EgressNamespace))
			continue
		}
		policies = append(policies, managedEgressPolicy(namespace, labels, current.caller, current.callerApp, current.target, current.ports, config.EgressCIDRs))
		if config.AuthMode == "external-identity" {
			if _, rendered := tokenEgress[current.caller]; !rendered {
				tokenEgress[current.caller] = struct{}{}
				policies = append(policies, workloadIdentityTokenEgressPolicy(namespace, labels, current.caller, current.callerApp))
			}
		}
	}
	for _, service := range topology.Services {
		if len(service.BootstrapJobEndpoints) == 0 {
			continue
		}
		if _, exists := inCluster[service.Name]; !exists {
			continue
		}
		portSet := make(map[uint32]struct{}, len(service.BootstrapJobEndpoints))
		for _, endpointName := range service.BootstrapJobEndpoints {
			endpoint, _ := topologyEndpointByName(service, endpointName)
			portSet[endpoint.Port] = struct{}{}
		}
		ports := make([]uint32, 0, len(portSet))
		for port := range portSet {
			ports = append(ports, port)
		}
		slices.Sort(ports)
		policies = append(policies,
			bootstrapJobIngressPolicy(namespace, labels, service.Name, topologyKubernetesAppLabel(service), ports),
			bootstrapJobEgressPolicy(namespace, labels, service.Name, topologyKubernetesAppLabel(service), ports),
		)
	}
	// A deploy Job reaches a dependency, not its own service, so its pod needs a
	// reach policy of its own — the running service's app-selector policies never
	// match the Job pod's distinct identity. Managed targets reach out of cluster
	// like a managed dependency, so the driver owns that egress, not this policy.
	for _, job := range topology.DeployJobs {
		if _, exists := inCluster[job.Service]; !exists {
			continue
		}
		if _, exists := inCluster[job.Writes.Service]; !exists {
			continue
		}
		target, _ := topologyServiceByName(topology, job.Writes.Service)
		endpoint, _ := topologyEndpointByName(target, job.Writes.Endpoint)
		policies = append(policies,
			deployJobIngressPolicy(namespace, labels, job.Name, target.Name, topologyKubernetesAppLabel(target), endpoint.Port),
			deployJobEgressPolicy(namespace, labels, job.Name, target.Name, topologyKubernetesAppLabel(target), endpoint.Port),
		)
	}
	ingressPorts := make(map[string]map[uint32]struct{})
	for _, route := range plan.ingress {
		if ingressPorts[route.service] == nil {
			ingressPorts[route.service] = make(map[uint32]struct{})
		}
		ingressPorts[route.service][route.port] = struct{}{}
	}
	ingressServices := make([]string, 0, len(ingressPorts))
	for service := range ingressPorts {
		ingressServices = append(ingressServices, service)
	}
	slices.Sort(ingressServices)
	for _, service := range ingressServices {
		topologyService, _ := topologyServiceByName(topology, service)
		ports := make([]uint32, 0, len(ingressPorts[service]))
		for port := range ingressPorts[service] {
			ports = append(ports, port)
		}
		slices.Sort(ports)
		policies = append(policies, kubeObject{
			APIVersion: "networking.k8s.io/v1",
			Kind:       "NetworkPolicy",
			Metadata:   objectMeta{Name: "allow-istio-ingress-to-" + service, Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"podSelector": map[string]any{"matchLabels": map[string]string{"app": topologyKubernetesAppLabel(topologyService)}},
				"policyTypes": []string{"Ingress"},
				"ingress": []any{map[string]any{
					"from": []any{map[string]any{
						"namespaceSelector": map[string]any{"matchLabels": map[string]string{"kubernetes.io/metadata.name": "istio-system"}},
						"podSelector":       map[string]any{"matchLabels": map[string]string{"istio": "ingressgateway"}},
					}},
					"ports": networkPorts(ports),
				}},
			},
		})
	}
	for _, service := range topology.Services {
		if len(service.PublicEgressPorts) == 0 || !slices.Contains(plan.services, service.Name) {
			continue
		}
		policies = append(policies, publicEgressPolicy(namespace, labels, service.Name, topologyKubernetesAppLabel(service), service.PublicEgressPorts))
	}
	return policies, nil
}

func dependencyIngressPolicy(namespace string, labels map[string]string, target, targetApp, caller, callerApp string, ports []uint32) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", target, "from", caller), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": targetApp}},
			"policyTypes": []string{"Ingress"},
			"ingress": []any{map[string]any{
				"from":  []any{map[string]any{"podSelector": map[string]any{"matchLabels": map[string]string{"app": callerApp}}}},
				"ports": networkPorts(ports),
			}},
		},
	}
}

func dependencyEgressPolicy(namespace string, labels map[string]string, caller, callerApp, target, targetApp string, ports []uint32) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", caller, "to", target), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": callerApp}},
			"policyTypes": []string{"Egress"},
			"egress": []any{map[string]any{
				"to":    []any{map[string]any{"podSelector": map[string]any{"matchLabels": map[string]string{"app": targetApp}}}},
				"ports": networkPorts(ports),
			}},
		},
	}
}

func bootstrapJobIngressPolicy(namespace string, labels map[string]string, service, serviceApp string, ports []uint32) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", service, "from", "bootstrap"), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": serviceApp}},
			"policyTypes": []string{"Ingress"},
			"ingress": []any{map[string]any{
				"from": []any{map[string]any{"podSelector": map[string]any{"matchLabels": map[string]string{
					"codefly.dev/bootstrap-service": service,
				}}}},
				"ports": networkPorts(ports),
			}},
		},
	}
}

func bootstrapJobEgressPolicy(namespace string, labels map[string]string, service, serviceApp string, ports []uint32) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", service, "bootstrap", "to", service), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{
				"codefly.dev/bootstrap-service": service,
			}},
			"policyTypes": []string{"Egress"},
			"egress": []any{map[string]any{
				"to":    []any{map[string]any{"podSelector": map[string]any{"matchLabels": map[string]string{"app": serviceApp}}}},
				"ports": networkPorts(ports),
			}},
		},
	}
}

// deployJobIngressPolicy admits the deploy Job pod to its write target. The Job
// pod carries codefly.dev/deploy-job plus the Kubernetes-set job-name, not the
// running service's app label, so its reach is scoped to this Job alone.
func deployJobIngressPolicy(namespace string, labels map[string]string, job, target, targetApp string, port uint32) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", target, "from", job), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": targetApp}},
			"policyTypes": []string{"Ingress"},
			"ingress": []any{map[string]any{
				"from": []any{map[string]any{"podSelector": map[string]any{"matchLabels": map[string]string{
					"codefly.dev/deploy-job": job,
					"job-name":               job,
				}}}},
				"ports": networkPorts([]uint32{port}),
			}},
		},
	}
}

func deployJobEgressPolicy(namespace string, labels map[string]string, job, target, targetApp string, port uint32) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", job, "to", target), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{
				"codefly.dev/deploy-job": job,
				"job-name":               job,
			}},
			"policyTypes": []string{"Egress"},
			"egress": []any{map[string]any{
				"to":    []any{map[string]any{"podSelector": map[string]any{"matchLabels": map[string]string{"app": targetApp}}}},
				"ports": networkPorts([]uint32{port}),
			}},
		},
	}
}

func managedEgressPolicy(
	namespace string,
	labels map[string]string,
	caller,
	callerApp,
	target string,
	ports []uint32,
	cidrs []string,
) kubeObject {
	destinations := make([]any, 0, len(cidrs))
	for _, cidr := range cidrs {
		destinations = append(destinations, map[string]any{"ipBlock": map[string]string{"cidr": cidr}})
	}
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", caller, "to", target), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": callerApp}},
			"policyTypes": []string{"Egress"},
			"egress":      []any{map[string]any{"to": destinations, "ports": networkPorts(ports)}},
		},
	}
}

// managedNamespaceEgressPolicy grants a caller reach to a dependency the cell
// runs in another namespace of the same cluster, on the declared ports only. The
// destination is selected by namespace rather than by address because a pod's
// address is not stable and the CIDR that would cover it covers the cluster.
func managedNamespaceEgressPolicy(
	namespace string,
	labels map[string]string,
	caller,
	callerApp,
	target string,
	ports []uint32,
	targetNamespace string,
) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", caller, "to", target), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": callerApp}},
			"policyTypes": []string{"Egress"},
			"egress": []any{map[string]any{
				"to": []any{map[string]any{
					"namespaceSelector": map[string]any{"matchLabels": map[string]string{"kubernetes.io/metadata.name": targetNamespace}},
				}},
				"ports": networkPorts(ports),
			}},
		},
	}
}

func networkPorts(ports []uint32) []any {
	result := make([]any, 0, len(ports))
	for _, port := range ports {
		result = append(result, map[string]any{"protocol": "TCP", "port": port})
	}
	return result
}

var publicIPv4Exceptions = []string{
	"0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8", "169.254.0.0/16", "172.16.0.0/12",
	"192.0.0.0/24", "192.0.2.0/24", "192.31.196.0/24", "192.52.193.0/24", "192.88.99.0/24", "192.168.0.0/16",
	"192.175.48.0/24", "198.18.0.0/15", "198.51.100.0/24", "203.0.113.0/24", "224.0.0.0/4", "240.0.0.0/4",
}

var publicIPv6Exceptions = []string{
	"::/128", "::1/128", "64:ff9b::/96", "64:ff9b:1::/48", "100::/64", "2001::/23", "2001:db8::/32",
	"2002::/16", "fc00::/7", "fe80::/10", "ff00::/8",
}

// instanceMetadataCIDR is the node-local instance metadata endpoint every major
// cloud serves workload-identity tokens from (GCP, Azure IMDS and AWS IMDS all
// answer on this address).
const instanceMetadataCIDR = "169.254.169.254/32"

// workloadIdentityTokenEgressPolicy lets a caller mint the token that IS its
// credential under external-identity. The default-deny baseline denies all
// egress and publicEgressPolicy excepts link-local, so without this the
// passwordless mode has no path to the credential it is defined by — and the
// failure would land at connection time, long after generation reported success.
func workloadIdentityTokenEgressPolicy(namespace string, labels map[string]string, caller, callerApp string) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: kubernetesName("allow", caller, "workload-identity-token"), Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": callerApp}},
			"policyTypes": []string{"Egress"},
			"egress": []any{map[string]any{
				"to":    []any{map[string]any{"ipBlock": map[string]string{"cidr": instanceMetadataCIDR}}},
				"ports": []any{map[string]any{"protocol": "TCP", "port": 80}},
			}},
		},
	}
}

func publicEgressPolicy(namespace string, labels map[string]string, service, serviceApp string, ports []uint32) kubeObject {
	return kubeObject{
		APIVersion: "networking.k8s.io/v1",
		Kind:       "NetworkPolicy",
		Metadata:   objectMeta{Name: "allow-" + service + "-public-egress", Namespace: namespace, Labels: labels},
		Spec: map[string]any{
			"podSelector": map[string]any{"matchLabels": map[string]string{"app": serviceApp}},
			"policyTypes": []string{"Egress"},
			"egress": []any{map[string]any{
				"to": []any{
					map[string]any{"ipBlock": map[string]any{"cidr": "0.0.0.0/0", "except": publicIPv4Exceptions}},
					map[string]any{"ipBlock": map[string]any{"cidr": "::/0", "except": publicIPv6Exceptions}},
				},
				"ports": networkPorts(ports),
			}},
		},
	}
}

func renderManagedServiceHandoffs(
	root,
	namespace string,
	labels map[string]string,
	topology deploymentTopology,
	plan environmentPlan,
) ([]string, error) {
	var paths []string
	for _, handoff := range plan.handoffs {
		config := plan.managed[handoff.Service]
		service, _ := topologyServiceByName(topology, handoff.Service)
		ports := make([]any, 0, len(service.Endpoints))
		for _, endpoint := range service.Endpoints {
			ports = append(ports, map[string]any{
				"name":       endpoint.Name,
				"port":       endpoint.Port,
				"targetPort": endpoint.Port,
			})
		}
		objects := []kubeObject{{
			APIVersion: "v1",
			Kind:       "Service",
			Metadata:   objectMeta{Name: handoff.Service, Namespace: namespace, Labels: labels},
			Spec: map[string]any{
				"type":         "ExternalName",
				"externalName": config.ExternalName,
				"ports":        ports,
			},
		}}
		for _, reference := range config.SecretReferences {
			objects = append(objects, kubeObject{
				APIVersion: "external-secrets.io/v1",
				Kind:       "ExternalSecret",
				Metadata:   objectMeta{Name: reference.Name, Namespace: namespace, Labels: labels},
				Spec: map[string]any{
					"refreshInterval": "1h",
					"secretStoreRef": map[string]string{
						"name": reference.SecretStore.Name,
						"kind": reference.SecretStore.Kind,
					},
					"target":   map[string]string{"name": reference.Name, "creationPolicy": "Owner"},
					"dataFrom": []any{map[string]any{"extract": map[string]string{"key": reference.RemoteKey}}},
				},
			})
		}
		relative := path.Join("handoffs", handoff.Service+".yaml")
		if err := writeYAMLDocuments(filepath.Join(root, filepath.FromSlash(relative)), objects); err != nil {
			return nil, err
		}
		paths = append(paths, relative)
	}
	return paths, nil
}

func validateGeneratedBundle(root string, bundle moduleBundle) error {
	for _, environment := range bundle.Environments {
		rendered, err := renderKustomization(filepath.Join(root, "overlays", environment.Name))
		if err != nil {
			return fmt.Errorf("render environment %q: %w", environment.Name, err)
		}
		if err := validateRenderedObjects(rendered, environment); err != nil {
			return fmt.Errorf("validate environment %q: %w", environment.Name, err)
		}
	}
	return filepath.WalkDir(root, func(file string, entry os.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if entry.IsDir() {
			return nil
		}
		data, err := os.ReadFile(file)
		if err != nil {
			return err
		}
		if match := unresolvedPattern.Find(data); match != nil {
			return fmt.Errorf("%s contains an unresolved placeholder or starter identity %q", file, match)
		}
		return nil
	})
}

func renderKustomization(root string) ([]map[string]any, error) {
	data, err := os.ReadFile(filepath.Join(root, "kustomization.yaml"))
	if err != nil {
		return nil, err
	}
	var configuration kustomization
	if err := yaml.Unmarshal(data, &configuration); err != nil {
		return nil, err
	}
	if configuration.APIVersion != "kustomize.config.k8s.io/v1beta1" || configuration.Kind != "Kustomization" {
		return nil, fmt.Errorf("invalid kustomization type")
	}
	var objects []map[string]any
	for _, relative := range configuration.Resources {
		clean, err := cleanRelativePath(relative)
		if err != nil {
			return nil, err
		}
		target := filepath.Join(root, filepath.FromSlash(clean))
		info, err := os.Stat(target)
		if err != nil {
			return nil, err
		}
		if info.IsDir() {
			nested, err := renderKustomization(target)
			if err != nil {
				return nil, err
			}
			objects = append(objects, nested...)
			continue
		}
		documents, err := decodeYAMLDocuments(target)
		if err != nil {
			return nil, err
		}
		objects = append(objects, documents...)
	}
	applyNamespaceTransform(objects, configuration.Namespace)
	return objects, nil
}

// applyNamespaceTransform mirrors the kustomize namespace transformer over the
// object set a base emits: a Namespace object is renamed, and every other
// (namespaced) object is placed in that namespace. It lets the generator
// validate the effective render without taking on a dependency on kustomize.
func applyNamespaceTransform(objects []map[string]any, namespace string) {
	if namespace == "" {
		return
	}
	for _, object := range objects {
		metadata, ok := object["metadata"].(map[string]any)
		if !ok {
			continue
		}
		if object["kind"] == "Namespace" {
			metadata["name"] = namespace
			continue
		}
		metadata["namespace"] = namespace
	}
}

// moduleOwnedAPIVersions is the closed set of apiVersions the module plugin is
// allowed to emit. Any object outside it — a repository-transport resource such
// as an Argo or Flux kind, for example — fails generation. Expressing the
// boundary as a positive allowlist keeps repository-transport identifiers out
// of runtime code entirely (see TestRuntimePluginOwnsNoGitOrArgoTransport).
var moduleOwnedAPIVersions = map[string]struct{}{
	"v1":                           {},
	"networking.k8s.io/v1":         {},
	"security.istio.io/v1":         {},
	"networking.istio.io/v1":       {},
	"gateway.networking.k8s.io/v1": {},
	"external-secrets.io/v1":       {},
}

// validateRenderedObjects proves the generated overlay is module-owned only: no
// Secret material and no object outside the module-owned apiVersion allowlist.
func validateRenderedObjects(objects []map[string]any, environment bundleEnvironment) error {
	namespaceCount := 0
	for _, object := range objects {
		kind, _ := object["kind"].(string)
		apiVersion, _ := object["apiVersion"].(string)
		metadata, _ := object["metadata"].(map[string]any)
		if kind == "Secret" {
			return fmt.Errorf("bundle output contains a Kubernetes Secret")
		}
		if _, owned := moduleOwnedAPIVersions[apiVersion]; !owned {
			return fmt.Errorf("bundle output contains non-module-owned resource %s/%s", apiVersion, kind)
		}
		if kind == "Namespace" {
			namespaceCount++
			if metadata["name"] != environment.Namespace {
				return fmt.Errorf("namespace resource targets %v", metadata["name"])
			}
		}
	}
	if namespaceCount != 1 {
		return fmt.Errorf("render must contain exactly one Namespace")
	}
	return nil
}

func decodeYAMLDocuments(file string) ([]map[string]any, error) {
	data, err := os.ReadFile(file)
	if err != nil {
		return nil, err
	}
	return decodeYAMLBytes(data, file)
}

func decodeYAMLBytes(data []byte, source string) ([]map[string]any, error) {
	decoder := yaml.NewDecoder(bytes.NewReader(data))
	var documents []map[string]any
	for {
		var document map[string]any
		err := decoder.Decode(&document)
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("parse %s: %w", source, err)
		}
		if len(document) == 0 {
			continue
		}
		if document["apiVersion"] == nil || document["kind"] == nil || document["metadata"] == nil {
			return nil, fmt.Errorf("%s contains an incomplete Kubernetes object", source)
		}
		documents = append(documents, document)
	}
	return documents, nil
}

func cleanRelativeFilesystemPath(value string) (string, error) {
	if value == "" || strings.ContainsAny(value, "\x00\\") || !filepath.IsLocal(value) {
		return "", fmt.Errorf("%q is not a local relative path", value)
	}
	clean := filepath.Clean(value)
	if clean != value {
		return "", fmt.Errorf("%q is not a canonical local relative path", value)
	}
	return clean, nil
}

func writeYAML(file string, value any) error {
	data, err := yaml.Marshal(value)
	if err != nil {
		return err
	}
	return writeFile(file, data)
}

func writeYAMLDocuments(file string, values []kubeObject) error {
	var output bytes.Buffer
	encoder := yaml.NewEncoder(&output)
	encoder.SetIndent(2)
	for _, value := range values {
		if err := encoder.Encode(value); err != nil {
			return err
		}
	}
	if err := encoder.Close(); err != nil {
		return err
	}
	return writeFile(file, output.Bytes())
}

func writeJSON(file string, value any) error {
	data, err := json.MarshalIndent(value, "", "  ")
	if err != nil {
		return err
	}
	data = append(data, '\n')
	return writeFile(file, data)
}

func writeFile(file string, data []byte) error {
	if err := os.MkdirAll(filepath.Dir(file), 0o755); err != nil {
		return err
	}
	return os.WriteFile(file, data, 0o644)
}

func cleanRelativePath(value string) (string, error) {
	if value == "" || strings.Contains(value, "\\") || path.IsAbs(value) {
		return "", fmt.Errorf("%q is not a relative repository path", value)
	}
	clean := path.Clean(value)
	if clean == "." || clean == ".." || strings.HasPrefix(clean, "../") || clean != value {
		return "", fmt.Errorf("%q is not a canonical relative repository path", value)
	}
	return clean, nil
}

func validateDNSLabel(field, value string) error {
	if len(value) == 0 || len(value) > 63 || !dnsLabelPattern.MatchString(value) {
		return fmt.Errorf("%s %q must be a Kubernetes DNS label", field, value)
	}
	return nil
}

func kubernetesName(parts ...string) string {
	joined := strings.Join(parts, "-")
	if len(joined) <= 63 {
		return joined
	}
	digest := sha256.Sum256([]byte(joined))
	suffix := hex.EncodeToString(digest[:])[:10]
	return strings.TrimRight(joined[:52], "-") + "-" + suffix
}

func mergeLabels(sets ...map[string]string) map[string]string {
	result := make(map[string]string)
	for _, set := range sets {
		for key, value := range set {
			result[key] = value
		}
	}
	return result
}