package lifecycle

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/url"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"gopkg.in/yaml.v3"

	"github.com/tidbcloud/controlplane9/internal/cluster"
	"github.com/tidbcloud/controlplane9/internal/config"
	"github.com/tidbcloud/controlplane9/internal/dbaasui"
	"github.com/tidbcloud/controlplane9/internal/host"
	cptilt "github.com/tidbcloud/controlplane9/internal/tilt"
	"github.com/tidbcloud/controlplane9/internal/verbose"
)

type tiltRuntime interface {
	VerifyBinary() error
	RunCI(ctx context.Context, req *cptilt.Request) error
	StartUp(ctx context.Context, req *cptilt.Request) (*cptilt.StartedProcess, error)
	ResourceReadiness(ctx context.Context, apiHost string, port int) (*cptilt.ResourceReadiness, error)
}

// TiltReadyStep turns the generated Tilt handoff into a runnable or validated
// session and persists runtime metadata for later day-2 commands.
type TiltReadyStep struct {
	StateDir                    string
	GCPRuntime                  *GCPRuntimeTiltStep
	Runner                      tiltRuntime
	PortReservation             *cptilt.PortReservation
	ECRDockerLogin              func(context.Context, ECRDockerLoginRequest) error
	KubeAPITunnelStarter        func(context.Context, *host.Session, string) (*cptilt.StartedProcess, error)
	KubeAPITunnelSettleDuration time.Duration
}

func (s *TiltReadyStep) Name() cluster.StepName { return cluster.StepTiltReady }

func (s *TiltReadyStep) Run(_ context.Context) error {
	return fmt.Errorf("tilt_ready requires lifecycle state")
}

func (s *TiltReadyStep) RunWithState(ctx context.Context, state *cluster.State) error {
	primaryRunning, err := s.primaryRemoteControllerRunning(ctx, state)
	if err != nil {
		return err
	}
	if !primaryRunning {
		if err := s.runPrimaryWithState(ctx, state); err != nil {
			return err
		}
	}
	if s.GCPRuntime == nil {
		return nil
	}
	if err := s.GCPRuntime.RunWithState(ctx, state); err != nil {
		return err
	}
	return s.persistRuntimePlan(ctx, state)
}

func (s *TiltReadyStep) runPrimaryWithState(ctx context.Context, state *cluster.State) error {
	if strings.TrimSpace(s.StateDir) == "" {
		return fmt.Errorf("tilt_ready: state dir is required")
	}
	if state == nil {
		return fmt.Errorf("tilt_ready: lifecycle state is required")
	}
	paths := cptilt.PathsForState(s.StateDir, state)
	resolved, err := config.LoadSession(resolvedSessionPathForState(s.StateDir, state))
	if err != nil {
		return fmt.Errorf("tilt_ready: load resolved session: %w", err)
	}
	report, err := cptilt.LoadContextReport(paths.ContextReport)
	if err != nil {
		return fmt.Errorf("tilt_ready: load context report: %w", err)
	}
	index, err := cptilt.LoadContextIndex(paths.ContextIndex)
	if err != nil {
		return fmt.Errorf("tilt_ready: load context index: %w", err)
	}
	if err := validateGeneratedTiltArtifacts(s.StateDir, &paths, report); err != nil {
		return err
	}
	status, err := cptilt.BuildStatus(s.StateDir, &paths, resolved.Tilt.Mode, report, index)
	if err != nil {
		return fmt.Errorf("tilt_ready: build status: %w", err)
	}
	status.APIHost = cptilt.DefaultAPIHost
	status.APIPort = normalizedTiltAPIPort(resolved.Tilt.Port)
	kubeconfigPath := resolveTiltReportFilePath(s.StateDir, report.KubeconfigPath)
	selectedHost, err := loadSelectedHostForTilt(s.StateDir)
	if err != nil {
		return fmt.Errorf("tilt_ready: %w", err)
	}
	selected := selectedHost.Session
	remoteController := cptilt.UsesRemoteController(selected)
	tunnelSelected, tunnelKubeconfigPath := remoteKubeAPITunnelTargetForTilt(selected, kubeconfigPath)
	if remoteController {
		status.Controller = cptilt.ControllerRemote
	}
	kubectlEnv := selectedHostKubectlEnv(selected)
	starterSourceDockerEnv, err := s.ensureStarterSourceECRLogin(ctx, resolved)
	if err != nil {
		return fmt.Errorf("tilt_ready: %w", err)
	}
	tiltEnv := append([]string(nil), kubectlEnv...)
	tiltEnv = append(tiltEnv, starterSourceDockerEnv...)
	tiltEnv = append(tiltEnv, "CP9_SESSION_CONTEXT="+absTiltReadyPath(paths.ContextIndex))

	switch resolved.Tilt.Mode {
	case "generate-only":
		if err := s.releasePortReservation(); err != nil {
			return fmt.Errorf("tilt_ready: release Tilt port reservation: %w", err)
		}
		if err := cptilt.SaveStatus(paths.Status, status); err != nil {
			return fmt.Errorf("tilt_ready: persist tilt status: %w", err)
		}
		return nil
	case "ci":
		if remoteController {
			if err := s.releasePortReservation(); err != nil {
				return fmt.Errorf("tilt_ready: release Tilt port reservation: %w", err)
			}
			materialized, err := cptilt.MaterializeRemoteController(
				ctx,
				selected,
				paths.Tiltfile,
				paths.ContextIndex,
				&cptilt.Request{Port: status.APIPort},
			)
			if err != nil {
				return fmt.Errorf("tilt_ready: %w", err)
			}
			applyRemoteControllerStatus(status, materialized)
			if err := cptilt.RunRemoteCI(ctx, selected, materialized); err != nil {
				return fmt.Errorf("tilt_ready: %w", err)
			}
			if err := cptilt.SaveStatus(paths.Status, status); err != nil {
				return fmt.Errorf("tilt_ready: persist tilt status: %w", err)
			}
			return nil
		}
		runner := s.runtime()
		var tunnel *cptilt.StartedProcess
		if tunnelSelected != nil {
			tunnel, err = s.startKubeAPITunnel(ctx, tunnelSelected, tunnelKubeconfigPath)
			if err != nil {
				return fmt.Errorf("tilt_ready: %w", err)
			}
		}
		defer stopStartedProcess(tunnel)
		if err := s.releasePortReservation(); err != nil {
			return fmt.Errorf("tilt_ready: release Tilt port reservation: %w", err)
		}
		req := &cptilt.Request{
			Tiltfile:    paths.Tiltfile,
			KubeContext: report.KubeContext,
			Kubeconfig:  kubeconfigPath,
			Env:         tiltEnv,
			Host:        status.APIHost,
			Port:        status.APIPort,
		}
		if err := runner.RunCI(ctx, req); err != nil {
			return fmt.Errorf("tilt_ready: %w", err)
		}
		if err := cptilt.SaveStatus(paths.Status, status); err != nil {
			return fmt.Errorf("tilt_ready: persist tilt status: %w", err)
		}
		return nil
	case "up":
		if remoteController {
			if err := s.releasePortReservation(); err != nil {
				return fmt.Errorf("tilt_ready: release Tilt port reservation: %w", err)
			}
			materialized, err := cptilt.MaterializeRemoteController(
				ctx,
				selected,
				paths.Tiltfile,
				paths.ContextIndex,
				&cptilt.Request{Port: status.APIPort},
			)
			if err != nil {
				return fmt.Errorf("tilt_ready: %w", err)
			}
			applyRemoteControllerStatus(status, materialized)
			started, err := cptilt.StartRemoteController(ctx, selected, materialized)
			if err != nil {
				return fmt.Errorf("tilt_ready: %w", err)
			}
			status.Running = true
			status.PID = started.PID
			status.Command = append([]string(nil), started.Command...)
			startedAt := started.StartedAt.UTC()
			status.StartedAt = &startedAt
			if err := cptilt.SaveStatus(paths.Status, status); err != nil {
				return fmt.Errorf("tilt_ready: persist remote Tilt status: %w", err)
			}
			forward, err := cptilt.StartRemoteControllerForward(
				ctx,
				selected,
				s.StateDir,
				status.APIPort,
			)
			if err != nil {
				return fmt.Errorf(
					"tilt_ready: remote controller is running but local UI forward failed: %w",
					err,
				)
			}
			status.ControllerTunnelPID = forward.PID
			status.ControllerTunnelCommand = append([]string(nil), forward.Command...)
			status.ControllerTunnelLogPath = cptilt.SessionRelativePath(s.StateDir, forward.LogPath)
			status.ControllerTunnelChildPID = cptilt.SessionRelativePath(s.StateDir, forward.ChildPIDPath)
			forwardStartedAt := forward.StartedAt.UTC()
			status.ControllerTunnelStartedAt = &forwardStartedAt
			var tunnel *cptilt.StartedProcess
			if tunnelSelected != nil {
				tunnel, err = s.startKubeAPITunnel(ctx, tunnelSelected, tunnelKubeconfigPath)
				if err != nil {
					cptilt.StopStartedProcess(forward)
					return fmt.Errorf(
						"tilt_ready: remote controller is running but local kube API tunnel failed: %w",
						err,
					)
				}
				status.KubeAPITunnelPID = tunnel.PID
				status.KubeAPITunnelCommand = append([]string(nil), tunnel.Command...)
				status.KubeAPITunnelLogPath = cptilt.SessionRelativePath(s.StateDir, tunnel.LogPath)
				status.KubeAPITunnelChildPIDPath = cptilt.SessionRelativePath(s.StateDir, tunnel.ChildPIDPath)
				tunnelStartedAt := tunnel.StartedAt.UTC()
				status.KubeAPITunnelStartedAt = &tunnelStartedAt
			}
			if err := cptilt.SaveStatus(paths.Status, status); err != nil {
				cptilt.StopStartedProcess(forward)
				stopStartedProcess(tunnel)
				return fmt.Errorf("tilt_ready: persist remote controller forward: %w", err)
			}
			return nil
		}
		runner := s.runtime()
		var tunnel *cptilt.StartedProcess
		if tunnelSelected != nil {
			tunnel, err = s.startKubeAPITunnel(ctx, tunnelSelected, tunnelKubeconfigPath)
			if err != nil {
				return fmt.Errorf("tilt_ready: %w", err)
			}
			if err := ensureStartedProcessAlive(tunnel, "remote kube API tunnel"); err != nil {
				stopStartedProcess(tunnel)
				return fmt.Errorf("tilt_ready: %w", err)
			}
		}
		if err := s.releasePortReservation(); err != nil {
			stopStartedProcess(tunnel)
			return fmt.Errorf("tilt_ready: release Tilt port reservation: %w", err)
		}
		started, err := runner.StartUp(ctx, &cptilt.Request{
			Tiltfile:    paths.Tiltfile,
			KubeContext: report.KubeContext,
			Kubeconfig:  kubeconfigPath,
			Env:         tiltEnv,
			Host:        status.APIHost,
			Port:        status.APIPort,
			LogPath:     paths.UpLog,
		})
		if err != nil {
			stopStartedProcess(tunnel)
			return fmt.Errorf("tilt_ready: %w", err)
		}
		if err := s.waitForKubeAPITunnelStable(ctx, tunnel); err != nil {
			stopStartedProcess(started)
			stopStartedProcess(tunnel)
			return fmt.Errorf("tilt_ready: %w", err)
		}
		status.Running = true
		status.PID = started.PID
		status.Command = append([]string(nil), started.Command...)
		startedAt := started.StartedAt
		status.StartedAt = &startedAt
		if tunnel != nil {
			status.KubeAPITunnelPID = tunnel.PID
			status.KubeAPITunnelCommand = append([]string(nil), tunnel.Command...)
			status.KubeAPITunnelLogPath = cptilt.SessionRelativePath(s.StateDir, tunnel.LogPath)
			status.KubeAPITunnelChildPIDPath = cptilt.SessionRelativePath(s.StateDir, tunnel.ChildPIDPath)
			tunnelStartedAt := tunnel.StartedAt
			status.KubeAPITunnelStartedAt = &tunnelStartedAt
		}
		status.LogPath = cptilt.SessionRelativePath(s.StateDir, paths.UpLog)
		if err := cptilt.SaveStatus(paths.Status, status); err != nil {
			return fmt.Errorf("tilt_ready: persist tilt status: %w", err)
		}
		return nil
	default:
		return fmt.Errorf("tilt_ready: unsupported mode %q", resolved.Tilt.Mode)
	}
}

func (s *TiltReadyStep) primaryRemoteControllerRunning(ctx context.Context, state *cluster.State) (bool, error) {
	if s == nil || s.GCPRuntime == nil {
		return false, nil
	}
	paths := cptilt.PathsForState(s.StateDir, state)
	status, err := cptilt.LoadStatus(paths.Status)
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("tilt_ready: load primary Tilt status: %w", err)
	}
	if status.Mode != "up" || status.Controller != cptilt.ControllerRemote {
		return false, nil
	}
	selected, err := host.Load(s.StateDir)
	if err != nil {
		return false, fmt.Errorf("tilt_ready: load primary host: %w", err)
	}
	observation, err := cptilt.ObserveRemoteController(ctx, selected, status)
	if err != nil {
		return false, fmt.Errorf("tilt_ready: observe primary Tilt controller: %w", err)
	}
	if !observation.Running() {
		return false, nil
	}
	status.Running = true
	status.PID = observation.PID
	if err := cptilt.SaveStatus(paths.Status, status); err != nil {
		return false, fmt.Errorf("tilt_ready: persist primary Tilt observation: %w", err)
	}
	return true, nil
}

func (s *TiltReadyStep) persistRuntimePlan(ctx context.Context, state *cluster.State) error {
	resolved, err := config.LoadSession(resolvedSessionPathForState(s.StateDir, state))
	if err != nil {
		return fmt.Errorf("tilt_ready: load resolved session for runtime plan: %w", err)
	}
	if resolved.Tilt.Mode != "up" {
		return nil
	}
	primaryPaths := cptilt.PathsForState(s.StateDir, state)
	primaryStatus, err := cptilt.LoadStatus(primaryPaths.Status)
	if err != nil {
		return fmt.Errorf("tilt_ready: load primary status for runtime plan: %w", err)
	}
	primaryHost, err := host.Load(s.StateDir)
	if err != nil {
		return fmt.Errorf("tilt_ready: load primary host for runtime plan: %w", err)
	}
	primaryReadiness, err := cptilt.ResourceReadinessForHost(ctx, primaryHost, primaryStatus)
	if err != nil {
		return fmt.Errorf("tilt_ready: read primary graph for runtime plan: %w", err)
	}
	if primaryReadiness == nil {
		return fmt.Errorf("tilt_ready: read primary graph for runtime plan: empty response")
	}
	gcpStateDir := host.GCPRegionalStateDir(s.StateDir)
	gcpPaths := cptilt.PathsForState(gcpStateDir, nil)
	gcpStatus, err := cptilt.LoadStatus(gcpPaths.Status)
	if err != nil {
		return fmt.Errorf("tilt_ready: load GCP status for runtime plan: %w", err)
	}
	gcpHost, err := host.Load(gcpStateDir)
	if err != nil {
		return fmt.Errorf("tilt_ready: load GCP host for runtime plan: %w", err)
	}
	gcpReadiness, err := cptilt.ResourceReadinessForHost(ctx, gcpHost, gcpStatus)
	if err != nil {
		return fmt.Errorf("tilt_ready: read GCP graph for runtime plan: %w", err)
	}
	if gcpReadiness == nil {
		return fmt.Errorf("tilt_ready: read GCP graph for runtime plan: empty response")
	}
	primaryMembership, err := cptilt.Membership(s.StateDir, cptilt.RuntimePrimary, s.StateDir, primaryReadiness)
	if err != nil {
		return fmt.Errorf("tilt_ready: build primary runtime membership: %w", err)
	}
	gcpMembership, err := cptilt.Membership(s.StateDir, cptilt.RuntimeGCPRegional, gcpStateDir, gcpReadiness)
	if err != nil {
		return fmt.Errorf("tilt_ready: build GCP runtime membership: %w", err)
	}
	plan, err := cptilt.BuildRuntimePlan(
		s.StateDir,
		resolved.Name,
		primaryPaths.Tiltfile,
		primaryMembership,
		gcpMembership,
	)
	if err != nil {
		return fmt.Errorf("tilt_ready: build runtime plan: %w", err)
	}
	if err := cptilt.SaveRuntimePlan(s.StateDir, plan); err != nil {
		return fmt.Errorf("tilt_ready: persist runtime plan: %w", err)
	}
	return nil
}

func applyRemoteControllerStatus(status *cptilt.Status, materialized *cptilt.RemoteMaterialization) {
	if status == nil || materialized == nil {
		return
	}
	status.Controller = cptilt.ControllerRemote
	status.APIHost = "127.0.0.1"
	status.RemoteUnit = materialized.Unit
	status.RemoteRoot = materialized.Root
	status.RemoteTiltfilePath = materialized.TiltfilePath
	status.RemoteContextIndexPath = materialized.ContextIndexPath
	status.RemoteKubeconfigPath = materialized.KubeconfigPath
	status.LogPath = "remote:" + materialized.LogPath
}

func (s *TiltReadyStep) releasePortReservation() error {
	if s == nil || s.PortReservation == nil {
		return nil
	}
	err := s.PortReservation.Release()
	s.PortReservation = nil
	return err
}

func absTiltReadyPath(path string) string {
	if strings.TrimSpace(path) == "" {
		return ""
	}
	abs, err := filepath.Abs(path)
	if err != nil {
		return path
	}
	return abs
}

type selectedHostForTilt struct {
	Session *host.Session
}

func loadSelectedHostForTilt(stateDir string) (selectedHostForTilt, error) {
	selected, err := host.Load(stateDir)
	if err != nil {
		if os.IsNotExist(err) {
			return selectedHostForTilt{}, nil
		}
		return selectedHostForTilt{}, fmt.Errorf("load selected host: %w", err)
	}
	return selectedHostForTilt{Session: selected}, nil
}

func selectedHostKubectlEnv(selected *host.Session) []string {
	if selected == nil {
		return nil
	}
	return selected.KubectlEnv()
}

func (s *TiltReadyStep) runtime() tiltRuntime {
	if s.Runner != nil {
		return s.Runner
	}
	return cptilt.NewCLI()
}

func (s *TiltReadyStep) startKubeAPITunnel(
	ctx context.Context,
	selected *host.Session,
	kubeconfigPath string,
) (*cptilt.StartedProcess, error) {
	starter := s.KubeAPITunnelStarter
	if starter == nil {
		starter = startRemoteKubeAPITunnel
	}
	return starter(ctx, selected, kubeconfigPath)
}

func needsRemoteKubeAPITunnel(selected *host.Session) bool {
	return selected != nil && selected.Mode == config.HostModeRemoteEC2K3d
}

func remoteKubeAPITunnelTargetForTilt(selected *host.Session, kubeconfigPath string) (*host.Session, string) {
	if needsRemoteKubeAPITunnel(selected) {
		return selected, kubeconfigPath
	}
	return nil, ""
}

func startRemoteKubeAPITunnel(
	ctx context.Context,
	selected *host.Session,
	kubeconfigPath string,
) (*cptilt.StartedProcess, error) {
	port, err := prepareKubeconfigForLocalTunnel(kubeconfigPath)
	if err != nil {
		return nil, err
	}
	if err := ensureLocalTCPPortAvailable(port); err != nil {
		return nil, fmt.Errorf("remote kube API tunnel: %w", err)
	}
	forwards, err := remoteKubeAPIForwards(selected, port)
	if err != nil {
		return nil, fmt.Errorf("remote kube API tunnel: %w", err)
	}
	for _, forward := range forwards[1:] {
		localPort, err := localForwardPort(forward)
		if err != nil {
			return nil, fmt.Errorf("remote kube API tunnel: %w", err)
		}
		if err := ensureLocalTCPPortAvailable(localPort); err != nil {
			return nil, fmt.Errorf("remote kube API tunnel: %w", err)
		}
	}
	baseArgs, err := host.SSHBaseArgs(selected)
	if err != nil {
		return nil, fmt.Errorf("remote kube API tunnel: build ssh args: %w", err)
	}
	if len(baseArgs) == 0 {
		return nil, fmt.Errorf("remote kube API tunnel: empty ssh args")
	}
	destination := baseArgs[len(baseArgs)-1]
	args := append([]string{}, baseArgs[:len(baseArgs)-1]...)
	args = append(args, "-o", "ExitOnForwardFailure=yes", "-N")
	for _, forward := range forwards {
		args = append(args, "-L", forward)
	}
	args = append(args, destination)
	supervisorPath, childPIDPath, err := cptilt.PrepareKubeAPITunnelSupervisor(selected.StateDir)
	if err != nil {
		return nil, fmt.Errorf("remote kube API tunnel: %w", err)
	}
	command := append([]string{supervisorPath, childPIDPath, "ssh"}, args...)
	logPath := filepath.Join(selected.StateDir, "tilt", "reports", "kube-api-tunnel.log")
	startedProcess, err := cptilt.StartSupervisedTunnel(ctx, command, logPath)
	if err != nil {
		return nil, fmt.Errorf("start remote kube API tunnel: %w", err)
	}
	return startedProcess, nil
}

func remoteKubeAPIForwards(selected *host.Session, hostAPIPort int) ([]string, error) {
	if hostAPIPort < 1 || hostAPIPort > 65535 {
		return nil, fmt.Errorf("invalid host Kubernetes API port %d", hostAPIPort)
	}
	forwards := []string{fmt.Sprintf("127.0.0.1:%d:127.0.0.1:%d", hostAPIPort, hostAPIPort)}
	if selected == nil || selected.DedicatedAPI == nil {
		return forwards, nil
	}
	dedicatedPort := selected.DedicatedAPI.HostPort
	if dedicatedPort < 1 || dedicatedPort > 65535 {
		return nil, fmt.Errorf("invalid Dedicated API port %d", dedicatedPort)
	}
	if dedicatedPort == hostAPIPort {
		return nil, fmt.Errorf("dedicated API port %d collides with host Kubernetes API", dedicatedPort)
	}
	return append(forwards, fmt.Sprintf("127.0.0.1:%d:127.0.0.1:%d", dedicatedPort, dedicatedPort)), nil
}

func localForwardPort(forward string) (int, error) {
	parts := strings.Split(forward, ":")
	if len(parts) != 4 {
		return 0, fmt.Errorf("invalid SSH local forward %q", forward)
	}
	port, err := strconv.Atoi(parts[1])
	if err != nil || port < 1 || port > 65535 {
		return 0, fmt.Errorf("invalid SSH local forward port in %q", forward)
	}
	return port, nil
}

func ensureLocalTCPPortAvailable(port int) error {
	listener, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)))
	if err != nil {
		return fmt.Errorf("local port %d is already in use before starting tunnel: %w", port, err)
	}
	return listener.Close()
}

func prepareKubeconfigForLocalTunnel(kubeconfigPath string) (int, error) {
	port, err := kubeconfigServerPort(kubeconfigPath)
	if err != nil {
		return 0, err
	}
	if err := normalizeKubeconfigServerHostForTunnel(kubeconfigPath); err != nil {
		return 0, err
	}
	return port, nil
}

func kubeconfigServerPort(kubeconfigPath string) (int, error) {
	data, err := os.ReadFile(kubeconfigPath)
	if err != nil {
		return 0, fmt.Errorf("remote kube API tunnel: read kubeconfig: %w", err)
	}
	var doc kubeconfigDocument
	if err := yaml.Unmarshal(data, &doc); err != nil {
		return 0, fmt.Errorf("remote kube API tunnel: parse kubeconfig: %w", err)
	}
	for _, item := range doc.Clusters {
		server := strings.TrimSpace(item.Cluster.Server)
		if server == "" {
			continue
		}
		u, err := url.Parse(server)
		if err != nil {
			return 0, fmt.Errorf("remote kube API tunnel: parse server %q: %w", server, err)
		}
		portText := strings.TrimSpace(u.Port())
		if portText == "" {
			return 0, fmt.Errorf("remote kube API tunnel: kubeconfig server %q has no explicit port", server)
		}
		port, err := strconv.Atoi(portText)
		if err != nil || port <= 0 || port > 65535 {
			return 0, fmt.Errorf("remote kube API tunnel: kubeconfig server %q has invalid port", server)
		}
		return port, nil
	}
	return 0, fmt.Errorf("remote kube API tunnel: kubeconfig has no cluster server")
}

func normalizeKubeconfigServerHostForTunnel(kubeconfigPath string) error {
	data, err := os.ReadFile(kubeconfigPath)
	if err != nil {
		return fmt.Errorf("remote kube API tunnel: read kubeconfig: %w", err)
	}
	var doc yaml.Node
	if err := yaml.Unmarshal(data, &doc); err != nil {
		return fmt.Errorf("remote kube API tunnel: parse kubeconfig: %w", err)
	}
	changed := normalizeKubeconfigServerNodesForTunnel(&doc)
	if !changed {
		return nil
	}
	out, err := yaml.Marshal(&doc)
	if err != nil {
		return fmt.Errorf("remote kube API tunnel: render normalized kubeconfig: %w", err)
	}
	if err := os.WriteFile(kubeconfigPath, out, 0o600); err != nil {
		return fmt.Errorf("remote kube API tunnel: write normalized kubeconfig: %w", err)
	}
	return nil
}

func normalizeKubeconfigServerNodesForTunnel(node *yaml.Node) bool {
	if node == nil {
		return false
	}
	changed := false
	if node.Kind == yaml.DocumentNode && len(node.Content) > 0 {
		return normalizeKubeconfigServerNodesForTunnel(node.Content[0])
	}
	if node.Kind != yaml.MappingNode {
		return false
	}
	clusters := mappingValue(node, "clusters")
	if clusters == nil || clusters.Kind != yaml.SequenceNode {
		return false
	}
	for _, clusterItem := range clusters.Content {
		clusterNode := mappingValue(clusterItem, "cluster")
		if clusterNode == nil || clusterNode.Kind != yaml.MappingNode {
			continue
		}
		server := mappingValue(clusterNode, "server")
		if server == nil || server.Kind != yaml.ScalarNode {
			continue
		}
		normalized, ok := normalizedTunnelServer(server.Value)
		if !ok {
			continue
		}
		server.Value = normalized
		changed = true
	}
	return changed
}

func mappingValue(node *yaml.Node, key string) *yaml.Node {
	if node == nil || node.Kind != yaml.MappingNode {
		return nil
	}
	for i := 0; i+1 < len(node.Content); i += 2 {
		if node.Content[i].Value == key {
			return node.Content[i+1]
		}
	}
	return nil
}

func normalizedTunnelServer(server string) (string, bool) {
	u, err := url.Parse(strings.TrimSpace(server))
	if err != nil {
		return "", false
	}
	switch u.Hostname() {
	case "0.0.0.0", "::", "":
	default:
		return "", false
	}
	port := strings.TrimSpace(u.Port())
	if port == "" {
		return "", false
	}
	u.Host = net.JoinHostPort("127.0.0.1", port)
	return u.String(), true
}

type kubeconfigDocument struct {
	Clusters []kubeconfigNamedCluster `yaml:"clusters"`
}

type kubeconfigNamedCluster struct {
	Cluster kubeconfigCluster `yaml:"cluster"`
}

type kubeconfigCluster struct {
	Server string `yaml:"server"`
}

func (s *TiltReadyStep) waitForKubeAPITunnelStable(ctx context.Context, tunnel *cptilt.StartedProcess) error {
	if tunnel == nil || tunnel.Exit == nil {
		return nil
	}
	if err := ensureStartedProcessAlive(tunnel, "remote kube API tunnel"); err != nil {
		return err
	}
	settle := s.KubeAPITunnelSettleDuration
	if settle == 0 {
		settle = 2 * time.Second
	}
	if settle < 0 {
		return nil
	}
	timer := time.NewTimer(settle)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case err := <-tunnel.Exit:
		if err != nil {
			return withProcessLog(fmt.Errorf("remote kube API tunnel exited after readiness: %w", err), tunnel.LogPath)
		}
		return withProcessLog(fmt.Errorf("remote kube API tunnel exited after readiness"), tunnel.LogPath)
	case <-timer.C:
		return nil
	}
}

func ensureStartedProcessAlive(started *cptilt.StartedProcess, name string) error {
	if started == nil || started.Exit == nil {
		return nil
	}
	select {
	case err := <-started.Exit:
		if err != nil {
			return withProcessLog(fmt.Errorf("%s exited after readiness: %w", name, err), started.LogPath)
		}
		return withProcessLog(fmt.Errorf("%s exited after readiness", name), started.LogPath)
	default:
		return nil
	}
}

func withProcessLog(err error, logPath string) error {
	if err == nil || strings.TrimSpace(logPath) == "" {
		return err
	}
	data, readErr := os.ReadFile(logPath)
	if readErr != nil {
		return err
	}
	text := strings.TrimSpace(string(data))
	if text == "" {
		return err
	}
	return fmt.Errorf("%w\n%s", err, text)
}

func stopStartedProcess(started *cptilt.StartedProcess) {
	cptilt.StopStartedProcess(started)
}

type ECRDockerLoginRequest struct {
	Profile         string
	Region          string
	Registry        string
	DockerConfigDir string
}

func (s *TiltReadyStep) ensureStarterSourceECRLogin(ctx context.Context, resolved *config.SessionConfig) ([]string, error) {
	return nil, nil
}

// SessionReadyStep verifies that persisted Tilt runtime metadata exists and is
// sufficient to unblock v1alpha2 day-2 commands.
type SessionReadyStep struct {
	StateDir              string
	ConfigPath            string
	Session               *config.SessionConfig
	O11YEndpointReady     func(context.Context) error
	KubectlRunner         func(context.Context, *host.Session, []byte, ...string) ([]byte, error)
	ReadinessTimeout      time.Duration
	ReadinessPollInterval time.Duration
	TiltResourceReadiness func(context.Context, string, int) (*cptilt.ResourceReadiness, error)
	KubeAPITunnelRepairer *cptilt.KubeAPITunnelRepairer
}

func (s *SessionReadyStep) Name() cluster.StepName { return cluster.StepSessionReady }

func (s *SessionReadyStep) Run(_ context.Context) error {
	return fmt.Errorf("session_ready requires lifecycle state")
}

func (s *SessionReadyStep) RunWithState(ctx context.Context, state *cluster.State) error {
	if strings.TrimSpace(s.StateDir) == "" {
		return fmt.Errorf("session_ready: state dir is required")
	}
	if state == nil {
		return fmt.Errorf("session_ready: lifecycle state is required")
	}
	paths := cptilt.PathsForState(s.StateDir, state)
	status, err := cptilt.LoadStatus(paths.Status)
	if err != nil {
		return fmt.Errorf("session_ready: load tilt status: %w", err)
	}
	if status.Mode == "up" {
		selected, loadErr := host.Load(s.StateDir)
		remoteController := loadErr == nil && cptilt.UsesRemoteController(selected)
		if remoteController {
			status.Controller = cptilt.ControllerRemote
			observation, observeErr := cptilt.ObserveRemoteController(ctx, selected, status)
			if observeErr != nil {
				return fmt.Errorf("session_ready: %w", observeErr)
			}
			if !observation.Running() {
				return fmt.Errorf(
					"session_ready: remote Tilt controller is not running (active=%s sub=%s result=%s)",
					observation.ActiveState,
					observation.SubState,
					observation.Result,
				)
			}
			status.Running = true
			status.PID = observation.PID
			if saveErr := cptilt.SaveStatus(paths.Status, status); saveErr != nil {
				return fmt.Errorf("session_ready: persist observed remote Tilt PID: %w", saveErr)
			}
			if _, forwardErr := cptilt.EnsureRemoteControllerForward(
				ctx,
				selected,
				paths.Status,
				s.StateDir,
			); forwardErr != nil {
				return fmt.Errorf("session_ready: %w", forwardErr)
			}
			status, err = cptilt.LoadStatus(paths.Status)
			if err != nil {
				return fmt.Errorf("session_ready: reload remote Tilt status: %w", err)
			}
		} else {
			if status.Controller == cptilt.ControllerRemote && loadErr != nil {
				return fmt.Errorf("session_ready: load selected host for remote Tilt controller: %w", loadErr)
			}
			if !status.Running {
				return fmt.Errorf("session_ready: tilt session is not marked running")
			}
			if status.PID <= 0 {
				return fmt.Errorf("session_ready: tilt session pid is missing")
			}
		}
		status, err = s.ensureRemoteKubeAPITunnelAlive(ctx, paths.Status, status)
		if err != nil {
			return fmt.Errorf("session_ready: %w", err)
		}
	}
	if config.DedicatedGCPEnabled(s.Session) {
		if err := s.ensureGCPRuntimeReady(ctx); err != nil {
			return fmt.Errorf("session_ready: %w", err)
		}
	}
	readinessCtx, cancel := context.WithTimeout(ctx, s.starterReadinessTimeout())
	defer cancel()
	if s.Session != nil && config.StarterActivationRequested(s.Session) {
		checks, handoffErr := s.waitStarterPreProvisionHandoff(readinessCtx, state, status)
		state.ReadinessChecks = make([]cluster.ReadinessCheck, 0, len(checks))
		for i := range checks {
			check := &checks[i]
			state.ReadinessChecks = append(state.ReadinessChecks, cluster.ReadinessCheck{
				Name: check.Name, Status: check.Status, Message: check.Message,
			})
		}
		if handoffErr != nil {
			return finishSessionReadiness(
				ctx, readinessCtx, state,
				fmt.Errorf("session_ready: Starter activation handoff is not ready: %w", handoffErr),
			)
		}
	}
	if readinessErr := s.runRemoteUIReadiness(readinessCtx, status); readinessErr != nil {
		return finishSessionReadiness(ctx, readinessCtx, state, fmt.Errorf("session_ready: %w", readinessErr))
	}
	if status.Controller == cptilt.ControllerRemote {
		selected, loadErr := host.Load(s.StateDir)
		if loadErr != nil {
			return fmt.Errorf("session_ready: load selected host for remote artifact sync: %w", loadErr)
		}
		if syncErr := cptilt.SyncRemoteControllerArtifacts(readinessCtx, selected, status, s.StateDir); syncErr != nil {
			return finishSessionReadiness(ctx, readinessCtx, state, fmt.Errorf("session_ready: %w", syncErr))
		}
	}
	if s.O11YEndpointReady != nil {
		if err := s.O11YEndpointReady(readinessCtx); err != nil {
			return fmt.Errorf("session_ready: O11Y endpoint service is not ready: %w", err)
		}
	}
	return nil
}

func (s *SessionReadyStep) ensureGCPRuntimeReady(ctx context.Context) error {
	stateDir := host.GCPRegionalStateDir(s.StateDir)
	paths := cptilt.PathsForState(stateDir, nil)
	status, err := cptilt.LoadStatus(paths.Status)
	if err != nil {
		return fmt.Errorf("load GCP Tilt status: %w", err)
	}
	if status.Mode != "up" {
		return nil
	}
	selected, err := host.Load(stateDir)
	if err != nil {
		return fmt.Errorf("load GCP host for Tilt readiness: %w", err)
	}
	observation, err := cptilt.ObserveRemoteController(ctx, selected, status)
	if err != nil {
		return fmt.Errorf("observe GCP Tilt controller: %w", err)
	}
	if !observation.Running() {
		return fmt.Errorf(
			"GCP Tilt controller is not running (active=%s sub=%s result=%s)",
			observation.ActiveState,
			observation.SubState,
			observation.Result,
		)
	}
	status.Running = true
	status.PID = observation.PID
	if err := cptilt.SaveStatus(paths.Status, status); err != nil {
		return fmt.Errorf("persist GCP Tilt observation: %w", err)
	}
	readinessCtx, cancel := context.WithTimeout(ctx, s.starterReadinessTimeout())
	defer cancel()
	waiter := &SessionReadyStep{StateDir: stateDir, ReadinessPollInterval: s.ReadinessPollInterval}
	for {
		readiness, err := cptilt.ResourceReadinessForHost(readinessCtx, selected, status)
		if err != nil {
			return fmt.Errorf("read GCP Tilt resources: %w", err)
		}
		if readiness == nil {
			return fmt.Errorf("read GCP Tilt resources: empty response")
		}
		if readiness.TotalResources == 0 {
			if _, err := waiter.waitTiltResourceReady(readinessCtx, status, "(Tiltfile)"); err != nil {
				return fmt.Errorf("wait for GCP Tilt graph: %w", err)
			}
			continue
		}
		if readiness.ReadyResources == readiness.TotalResources {
			break
		}
		for _, resource := range readiness.NotReadyResources {
			if _, err := waiter.waitTiltResourceReady(readinessCtx, status, resource); err != nil {
				return fmt.Errorf("wait for GCP Tilt resource %s: %w", resource, err)
			}
		}
	}
	if err := cptilt.SyncRemoteControllerArtifacts(readinessCtx, selected, status, stateDir); err != nil {
		return fmt.Errorf("sync GCP Tilt artifacts: %w", err)
	}
	return nil
}

func finishSessionReadiness(
	parentCtx context.Context,
	readinessCtx context.Context,
	state *cluster.State,
	err error,
) error {
	if parentCtx.Err() != nil || !errors.Is(readinessCtx.Err(), context.DeadlineExceeded) {
		return err
	}
	state.ReadinessChecks = append(state.ReadinessChecks, cluster.ReadinessCheck{
		Name:    "session-readiness-timeout",
		Status:  "warning",
		Message: err.Error(),
	})
	return nil
}

func (s *SessionReadyStep) runRemoteUIReadiness(
	ctx context.Context,
	status *cptilt.Status,
) error {
	if s.Session == nil ||
		s.Session.Host.Mode != config.HostModeRemoteEC2K3d ||
		config.SessionPersona(s.Session) != "dev-full" ||
		status == nil || status.Mode == "generate-only" {
		return nil
	}
	selected, err := host.Load(s.StateDir)
	if err != nil {
		return s.saveUIReadinessFailure(
			&dbaasui.ReadinessEvidence{StaticMessage: "selected Remote host is unavailable"},
			fmt.Errorf("load selected host for dbaas-ui readiness: %w", err),
		)
	}
	if selected.Mode != config.HostModeRemoteEC2K3d || selected.Remote == nil {
		return s.saveUIReadinessFailure(
			&dbaasui.ReadinessEvidence{StaticMessage: "selected host does not satisfy the Remote UI placement contract"},
			fmt.Errorf("dbaas-ui requires a selected remote-ec2-k3d host"),
		)
	}
	uiURL := strings.TrimSpace(selected.Remote.PublicHTTPSURL)
	runUI := func(ctx context.Context, input []byte, args ...string) ([]byte, error) {
		return s.runUIKubectl(ctx, selected, input, args...)
	}
	run := func(ctx context.Context, selected *host.Session, args ...string) ([]byte, error) {
		return s.runUIKubectl(ctx, selected, nil, args...)
	}
	timeout := s.starterReadinessTimeout()
	var metadbErr error
	if status.Mode == "up" {
		_, metadbErr = s.waitTiltResourceReady(ctx, status, "premium-metadb")
		if metadbErr == nil {
			output, err := s.runKubectlRolloutWithObjectWait(
				ctx,
				selected,
				run,
				timeout,
				"-n", "cp9-system",
				"rollout", "status", "deployment/premium-metadb",
				"--timeout", timeout.String(),
			)
			if err != nil {
				metadbErr = fmt.Errorf("premium-metadb Deployment rollout: %w\n%s", err, string(output))
			}
		}
	}
	var identity dbaasui.Dimension
	if metadbErr != nil {
		identity = dbaasui.Dimension{
			Status:  "unavailable",
			Message: fmt.Sprintf("Account seed dependency premium-metadb is not ready: %v", metadbErr),
		}
	} else {
		identity = dbaasui.EnsureSeedIdentity(ctx, s.Session.UI.SeedIdentity.Email, runUI)
	}
	withIdentity := func(evidence *dbaasui.ReadinessEvidence) *dbaasui.ReadinessEvidence {
		evidence.UIIdentityReady = identity.Ready
		evidence.IdentityMessage = identity.Message
		return evidence
	}
	if uiURL == "" {
		return s.saveUIReadinessFailure(
			withIdentity(&dbaasui.ReadinessEvidence{StaticMessage: "Remote public HTTPS session URL is missing"}),
			fmt.Errorf("dbaas-ui requires the Remote public HTTPS session URL"),
		)
	}

	const (
		uiResource  = "dbaas-ui"
		uiNamespace = "cp9-global"
	)
	if status.Mode == "up" {
		if _, err := s.waitTiltResourceReady(
			ctx,
			status,
			uiResource,
		); err != nil {
			return s.saveUIReadinessFailure(
				withIdentity(&dbaasui.ReadinessEvidence{URL: uiURL, StaticMessage: "dbaas-ui Tilt resource is not ready"}),
				fmt.Errorf("wait for dbaas-ui Tilt resource: %w", err),
			)
		}
	}
	output, err := s.runKubectlRolloutWithObjectWait(
		ctx,
		selected,
		run,
		timeout,
		"-n", uiNamespace,
		"rollout", "status", "deployment/dbaas-ui",
		"--timeout", timeout.String(),
	)
	if err != nil {
		return s.saveUIReadinessFailure(
			withIdentity(&dbaasui.ReadinessEvidence{URL: uiURL, StaticMessage: "dbaas-ui Deployment is unavailable"}),
			fmt.Errorf("dbaas-ui Deployment rollout: %w\n%s", err, string(output)),
		)
	}
	expectedRuntimeConfig, err := dbaasui.RenderRuntimeConfig(SessionUIO11YRuntimeConfig(s.Session.UI.O11Y))
	if err != nil {
		return s.saveUIReadinessFailure(
			withIdentity(&dbaasui.ReadinessEvidence{URL: uiURL, StaticMessage: "dbaas-ui expected runtime config is invalid"}),
			fmt.Errorf("render expected dbaas-ui runtime config: %w", err),
		)
	}
	evidence, inspectErr := dbaasui.InspectReadiness(ctx, &dbaasui.ReadinessInspection{
		URL:                   uiURL,
		ExpectedRuntimeConfig: expectedRuntimeConfig,
		SeedIdentity:          s.Session.UI.SeedIdentity.Email,
		Run:                   runUI,
		Auth0HTTPClient:       verbose.NewHTTPClient(30 * time.Second),
		Auth0ReadToken:        strings.TrimSpace(os.Getenv("CP9_AUTH0_READ_TOKEN")),
		InspectDeployment: func(ctx context.Context) (string, error) {
			imageOutput, imageErr := runUI(
				ctx, nil,
				"-n", uiNamespace,
				"get", "deployment", "dbaas-ui",
				"-o", `jsonpath={.spec.template.spec.containers[?(@.name=="dbaas-ui")].image}`,
			)
			if imageErr != nil {
				return strings.TrimSpace(string(imageOutput)), fmt.Errorf("read dbaas-ui Deployment image: %w\n%s", imageErr, string(imageOutput))
			}
			return strings.TrimSpace(string(imageOutput)), nil
		},
		InspectIdentity: func(context.Context, string, dbaasui.CommandRunner) dbaasui.Dimension {
			return identity
		},
	})
	if err := dbaasui.SaveReadiness(s.StateDir, evidence); err != nil {
		return fmt.Errorf("persist dbaas-ui readiness evidence: %w", err)
	}
	return inspectErr
}

func (s *SessionReadyStep) saveUIReadinessFailure(evidence *dbaasui.ReadinessEvidence, err error) error {
	if writeErr := dbaasui.SaveReadiness(s.StateDir, evidence); writeErr != nil {
		return fmt.Errorf("%w; persist dbaas-ui readiness evidence: %w", err, writeErr)
	}
	return err
}

func (s *SessionReadyStep) runUIKubectl(
	ctx context.Context,
	selected *host.Session,
	input []byte,
	args ...string,
) ([]byte, error) {
	if s.KubectlRunner != nil {
		return s.KubectlRunner(ctx, selected, input, args...)
	}
	return host.RunKubectl(ctx, selected, input, args...)
}

func (s *SessionReadyStep) waitStarterPreProvisionHandoff(
	ctx context.Context,
	state *cluster.State,
	status *cptilt.Status,
) ([]StarterReadinessCheck, error) {
	if status == nil || status.Mode != "up" {
		return nil, nil
	}
	timeout := s.starterReadinessTimeout()
	targets := []struct {
		checkName string
		resource  string
	}{
		{checkName: "starter-control-plane-serverless-service", resource: "serverless-service"},
		{checkName: "starter-control-plane-serverless-global-service", resource: "serverless-global-service"},
	}
	type result struct {
		checkName string
		readiness *cptilt.ResourceReadiness
		err       error
	}
	results := make(chan result, len(targets))
	var wg sync.WaitGroup
	for _, target := range targets {
		wg.Go(func() {
			readiness, err := s.waitTiltResourceReady(ctx, status, target.resource)
			results <- result{checkName: target.checkName, readiness: readiness, err: err}
		})
	}
	wg.Wait()
	close(results)
	var pending []string
	for result := range results {
		if result.err != nil {
			pending = append(pending, result.checkName+": "+result.err.Error())
			continue
		}
	}
	selected, err := host.Load(s.StateDir)
	if err != nil {
		return nil, fmt.Errorf("load selected host for Starter activation prerequisites: %w", err)
	}
	runner := StarterPrerequisiteKubectl(nil)
	if s.KubectlRunner != nil {
		runner = func(ctx context.Context, selected *host.Session, args ...string) ([]byte, error) {
			return s.KubectlRunner(ctx, selected, nil, args...)
		}
	}
	if len(pending) > 0 {
		checks := CheckStarterActivationPrerequisites(ctx, selected, runner, s.ConfigPath)
		for i := range checks {
			for _, failure := range pending {
				if strings.HasPrefix(failure, checks[i].Name+":") {
					checks[i].Status = "not_ready"
					checks[i].Message = strings.TrimSpace(strings.TrimPrefix(failure, checks[i].Name+":"))
					break
				}
			}
		}
		return checks, fmt.Errorf(
			"named checks did not become ready: %s; run %s",
			strings.Join(pending, "; "),
			starterStatusRetryCommand(s.ConfigPath),
		)
	}
	if err := cptilt.SaveStatus(cptilt.PathsForState(s.StateDir, state).Status, status); err != nil {
		return nil, fmt.Errorf("persist Starter handoff Tilt readiness: %w", err)
	}
	prerequisiteCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	checks, err := WaitStarterActivationPrerequisites(
		prerequisiteCtx,
		selected,
		timeout,
		s.ReadinessPollInterval,
		runner,
		s.ConfigPath,
	)
	return checks, err
}

func (s *SessionReadyStep) ensureRemoteKubeAPITunnelAlive(
	ctx context.Context,
	statusPath string,
	status *cptilt.Status,
) (*cptilt.Status, error) {
	if status == nil || status.KubeAPITunnelPID <= 0 {
		return status, nil
	}
	repairer := s.KubeAPITunnelRepairer
	if repairer == nil {
		repairer = &cptilt.KubeAPITunnelRepairer{}
	}
	message, err := repairer.EnsureKubeAPITunnel(ctx, statusPath)
	if err != nil {
		return nil, withProcessLog(err, resolveTiltReportFilePath(s.StateDir, status.KubeAPITunnelLogPath))
	}
	if strings.TrimSpace(message) == "" {
		return status, nil
	}
	reloaded, err := cptilt.LoadStatus(statusPath)
	if err != nil {
		return nil, fmt.Errorf("reload repaired kube API tunnel status: %w", err)
	}
	return reloaded, nil
}

const starterRegionalSmokeEvidenceFileName = "starter-regional-smoke.json"

// StarterRegionalSmokeEvidence captures the remote regional deploy/readiness
// proof produced by session_ready for imported Starter routes.
type StarterRegionalSmokeEvidence struct {
	SessionName               string                                `json:"session_name"`
	Owner                     string                                `json:"owner,omitempty"`
	Namespace                 string                                `json:"namespace,omitempty"`
	RemoteEKSCluster          string                                `json:"remote_eks_cluster"`
	KubeContext               string                                `json:"kube_context"`
	KubeconfigPath            string                                `json:"kubeconfig_path"`
	TiltMode                  string                                `json:"tilt_mode"`
	ReadinessStatus           string                                `json:"readiness_status"`
	RemoteRegionalEndpointURL string                                `json:"remote_regional_endpoint_url,omitempty"`
	ClusterProxyEndpointURL   string                                `json:"cluster_proxy_endpoint_url,omitempty"`
	SQLGatewayDNSName         string                                `json:"sql_gateway_dns_name,omitempty"`
	Services                  []StarterRegionalSmokeEvidenceService `json:"services"`
}

// StarterRegionalSmokeEvidenceService records one regional service readiness
// check against the selected remote EKS target.
type StarterRegionalSmokeEvidenceService struct {
	Name            string `json:"name"`
	Role            string `json:"role,omitempty"`
	RuntimeHost     string `json:"runtime_host,omitempty"`
	Resource        string `json:"resource,omitempty"`
	Namespace       string `json:"namespace,omitempty"`
	WorkloadKind    string `json:"workload_kind,omitempty"`
	WorkloadName    string `json:"workload_name,omitempty"`
	ReadinessStatus string `json:"readiness_status"`
	Message         string `json:"message,omitempty"`
}

func StarterRegionalSmokeEvidencePath(stateDir string) string {
	return filepath.Join(stateDir, starterRegionalSmokeEvidenceFileName)
}

const starterGlobalWiringProbeEvidenceFileName = "starter-global-wiring-probe.json"

// StarterGlobalWiringProbeEvidence captures the no-mutation imported Starter probe
// that proves session_ready used the generated local-global/remote-regional
// wiring contract instead of shared staging fallbacks.
type StarterGlobalWiringProbeEvidence struct {
	SessionName               string                            `json:"session_name"`
	Owner                     string                            `json:"owner,omitempty"`
	Namespace                 string                            `json:"namespace,omitempty"`
	ProbeStatus               string                            `json:"probe_status"`
	LocalGlobalStatus         string                            `json:"local_global_status"`
	RemoteRegionalStatus      string                            `json:"remote_regional_status"`
	RemoteRegionalRouteStatus string                            `json:"remote_regional_route_status"`
	StarterRegionSeedStatus   string                            `json:"starter_region_seed_status,omitempty"`
	O11yStatus                string                            `json:"o11y_status"`
	RemoteEKSCluster          string                            `json:"remote_eks_cluster"`
	RemoteEKSEndpoint         string                            `json:"remote_eks_endpoint"`
	RemoteRegionalEndpointURL string                            `json:"remote_regional_endpoint_url,omitempty"`
	RemoteRegionalRequestID   string                            `json:"remote_regional_request_id,omitempty"`
	RemoteEKSContext          string                            `json:"remote_eks_context,omitempty"`
	O11yEndpointURL           string                            `json:"o11y_endpoint_url"`
	O11yTokenSource           string                            `json:"o11y_token_source"`
	EvidenceTable             string                            `json:"evidence_table,omitempty"`
	O11yScriptHash            string                            `json:"o11y_script_hash,omitempty"`
	O11yRequestID             string                            `json:"o11y_request_id"`
	Message                   string                            `json:"message,omitempty"`
	LocalGlobalServices       []StarterGlobalWiringProbeService `json:"local_global_services,omitempty"`
}

// StarterGlobalWiringProbeService records local-global readiness checked on
// the preserved local k3d host before the cross-environment probes run.
type StarterGlobalWiringProbeService struct {
	Name            string `json:"name"`
	RuntimeHost     string `json:"runtime_host,omitempty"`
	KubeContext     string `json:"kube_context,omitempty"`
	Namespace       string `json:"namespace,omitempty"`
	WorkloadKind    string `json:"workload_kind,omitempty"`
	WorkloadName    string `json:"workload_name,omitempty"`
	ReadinessStatus string `json:"readiness_status"`
	Message         string `json:"message,omitempty"`
}

func StarterGlobalWiringProbeEvidencePath(stateDir string) string {
	return filepath.Join(stateDir, starterGlobalWiringProbeEvidenceFileName)
}

const starterSessionReadyDefaultTimeout = 15 * time.Minute
const starterRemoteK3dSessionReadyDefaultTimeout = 60 * time.Minute

// WaitForTiltResources blocks an activation branch until its direct baseline
// resources are ready. It observes the running Session Tiltfile instead of
// introducing product-to-product dependencies between activation branches.
func WaitForTiltResources(
	ctx context.Context,
	stateDir string,
	timeout time.Duration,
	resources ...string,
) error {
	state, err := cluster.LoadState(stateDir)
	if err != nil {
		return fmt.Errorf("load lifecycle state: %w", err)
	}
	status, err := cptilt.LoadStatusForState(stateDir, state)
	if err != nil {
		return fmt.Errorf("load tilt status: %w", err)
	}
	if status.Mode == "ci" {
		return nil
	}
	if !status.Running {
		return fmt.Errorf("tilt session is not marked running")
	}
	if timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, timeout)
		defer cancel()
	}
	waiter := &SessionReadyStep{StateDir: stateDir}
	for _, resource := range resources {
		if _, err := waiter.waitTiltResourceReady(ctx, status, resource); err != nil {
			return err
		}
	}
	return nil
}

func (s *SessionReadyStep) starterReadinessTimeout() time.Duration {
	return s.starterReadinessTimeoutForSession(s.Session)
}

func (s *SessionReadyStep) starterReadinessTimeoutForSession(session *config.SessionConfig) time.Duration {
	if s.ReadinessTimeout > 0 {
		return s.ReadinessTimeout
	}
	if session != nil &&
		config.UsesSessionTiltfile(session) &&
		session.Host.Mode == config.HostModeRemoteEC2K3d &&
		config.SessionPersona(session) == "dev-full" &&
		(config.DedicatedGCPEnabled(session) ||
			config.StarterActivationRequested(session) ||
			config.EssentialV2Enabled(session)) {
		return starterRemoteK3dSessionReadyDefaultTimeout
	}
	return starterSessionReadyDefaultTimeout
}

func (s *SessionReadyStep) waitTiltResourceReady(
	ctx context.Context,
	status *cptilt.Status,
	resource string,
) (*cptilt.ResourceReadiness, error) {
	resource = strings.TrimSpace(resource)
	if status == nil || resource == "" {
		return &cptilt.ResourceReadiness{}, nil
	}
	poll := s.ReadinessPollInterval
	if poll <= 0 {
		poll = 2 * time.Second
	}
	var lastReadiness *cptilt.ResourceReadiness
	var lastErr error
	for {
		var readiness *cptilt.ResourceReadiness
		var err error
		if status.Controller == cptilt.ControllerRemote {
			selected, loadErr := host.Load(s.StateDir)
			if loadErr != nil {
				err = loadErr
			} else {
				readiness, err = cptilt.ResourceReadinessForHost(ctx, selected, status)
			}
		} else {
			readiness, err = s.tiltResourceReadiness(ctx, status.APIHost, status.APIPort)
		}
		if err != nil {
			if lastErr == nil || (!errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded)) {
				lastErr = err
			}
		} else {
			lastReadiness = readiness
			if resourceIsReady(readiness, resource) {
				return readiness, nil
			}
			if failure, failed := tiltResourceTerminalFailure(readiness, resource); failed {
				return nil, formatTiltResourceWaitError(
					resource,
					fmt.Errorf("terminal Tilt failure: %s", failure),
					nil,
					readiness,
				)
			}
			lastErr = fmt.Errorf("tilt resource %s is not ready", resource)
			if !resourceWasObserved(readiness, resource) {
				lastErr = fmt.Errorf("tilt resource %s was not found", resource)
			}
		}
		timer := time.NewTimer(poll)
		select {
		case <-ctx.Done():
			timer.Stop()
			return nil, formatTiltResourceWaitError(resource, lastErr, ctx.Err(), lastReadiness)
		case <-timer.C:
		}
	}
}

func tiltResourceTerminalFailure(readiness *cptilt.ResourceReadiness, resource string) (string, bool) {
	if readiness == nil {
		return "", false
	}
	state, ok := readiness.ResourceStates[resource]
	if !ok {
		return "", false
	}
	terminalStatus := func(status string) bool {
		switch strings.ToLower(strings.TrimSpace(status)) {
		case "error", "failed":
			return true
		default:
			return false
		}
	}
	if !terminalStatus(state.UpdateStatus) && !terminalStatus(state.RuntimeStatus) {
		return "", false
	}
	return strings.TrimSpace(firstNonEmpty(state.Message, state.Reason, state.UpdateStatus, state.RuntimeStatus)), true
}

func formatTiltResourceWaitError(
	resource string,
	lastErr error,
	waitErr error,
	lastReadiness *cptilt.ResourceReadiness,
) error {
	state := formatTiltResourceState(lastReadiness, resource) + formatTiltReadinessSnapshot(lastReadiness)
	if lastErr == nil {
		return fmt.Errorf("wait for %s: %w%s", resource, waitErr, state)
	}
	if waitErr != nil && !errors.Is(lastErr, waitErr) {
		return fmt.Errorf("wait for %s: %w: %w%s", resource, lastErr, waitErr, state)
	}
	return fmt.Errorf("wait for %s: %w%s", resource, lastErr, state)
}

func formatTiltResourceState(readiness *cptilt.ResourceReadiness, resource string) string {
	if readiness == nil {
		return ""
	}
	state, ok := readiness.ResourceStates[resource]
	if !ok {
		return ""
	}
	return fmt.Sprintf(
		" (runtime=%s update=%s ready=%s last transition=%s reason=%s message=%s)",
		firstNonEmpty(state.RuntimeStatus, "unknown"),
		firstNonEmpty(state.UpdateStatus, "unknown"),
		firstNonEmpty(state.ReadyStatus, "unknown"),
		firstNonEmpty(state.LastTransitionTime, "unknown"),
		firstNonEmpty(state.Reason, "unknown"),
		firstNonEmpty(state.Message, "none"),
	)
}

func (s *SessionReadyStep) tiltResourceReadiness(
	ctx context.Context,
	apiHost string,
	port int,
) (*cptilt.ResourceReadiness, error) {
	if s.TiltResourceReadiness != nil {
		return s.TiltResourceReadiness(ctx, apiHost, port)
	}
	return cptilt.NewCLI().ResourceReadiness(ctx, apiHost, port)
}

func resourceIsReady(readiness *cptilt.ResourceReadiness, resource string) bool {
	if readiness == nil {
		return false
	}
	ready, ok := readiness.ResourceReady[resource]
	return ok && ready
}

func resourceWasObserved(readiness *cptilt.ResourceReadiness, resource string) bool {
	if readiness == nil {
		return false
	}
	if _, ok := readiness.ResourceReady[resource]; ok {
		return true
	}
	return slices.Contains(readiness.NotReadyResources, resource)
}

func formatTiltReadinessSnapshot(readiness *cptilt.ResourceReadiness) string {
	if readiness == nil {
		return ""
	}
	notReady := strings.Join(readiness.NotReadyResources, ", ")
	if notReady == "" {
		notReady = "none"
	}
	return fmt.Sprintf(" (ready %d/%d, not ready: %s)",
		readiness.ReadyResources, readiness.TotalResources, notReady)
}

func (s *SessionReadyStep) runKubectlRolloutWithObjectWait(
	ctx context.Context,
	selected *host.Session,
	run func(context.Context, *host.Session, ...string) ([]byte, error),
	timeout time.Duration,
	args ...string,
) ([]byte, error) {
	if timeout <= 0 {
		timeout = 3 * time.Minute
	}
	pollInterval := s.ReadinessPollInterval
	if pollInterval <= 0 {
		pollInterval = 2 * time.Second
	}
	deadline := time.Now().Add(timeout)
	for {
		output, err := run(ctx, selected, args...)
		if err == nil || !kubectlRolloutObjectMissing(output, err) || time.Now().After(deadline) {
			return output, err
		}
		timer := time.NewTimer(pollInterval)
		select {
		case <-ctx.Done():
			if !timer.Stop() {
				<-timer.C
			}
			return output, ctx.Err()
		case <-timer.C:
		}
	}
}

func kubectlRolloutObjectMissing(output []byte, err error) bool {
	if err == nil {
		return false
	}
	text := strings.ToLower(err.Error() + "\n" + string(output))
	return strings.Contains(text, "notfound") || strings.Contains(text, "not found")
}

func resolvedSessionPathForState(stateDir string, state *cluster.State) string {
	path := config.ResolvedSessionPath(stateDir)
	if state != nil && state.Paths != nil && strings.TrimSpace(state.Paths.ResolvedSession) != "" {
		path = state.Paths.ResolvedSession
	}
	return path
}

func normalizedTiltAPIPort(port int) int {
	if port <= 0 {
		return cptilt.DefaultAPIPort
	}
	return port
}

func validateGeneratedTiltArtifacts(stateDir string, paths *cptilt.Paths, report *cptilt.ContextReport) error {
	if report == nil {
		return fmt.Errorf("tilt_ready: context report is required")
	}
	if paths == nil {
		return fmt.Errorf("tilt_ready: tilt paths are required")
	}
	requiredFiles := []string{
		paths.Tiltfile,
		paths.ContextIndex,
		paths.ContextReport,
	}
	for _, path := range requiredFiles {
		if err := requireExistingFile(path); err != nil {
			return fmt.Errorf("tilt_ready: %w", err)
		}
	}
	if strings.TrimSpace(report.KubeconfigPath) != "" {
		if err := requireExistingFile(resolveTiltReportFilePath(stateDir, report.KubeconfigPath)); err != nil {
			return fmt.Errorf("tilt_ready: %w", err)
		}
	}
	return nil
}

func requireExistingFile(path string) error {
	info, err := os.Stat(path)
	if err != nil {
		return fmt.Errorf("required file %q: %w", path, err)
	}
	if info.IsDir() {
		return fmt.Errorf("required file %q is a directory", path)
	}
	return nil
}

func resolveTiltReportFilePath(stateDir, path string) string {
	path = strings.TrimSpace(path)
	if path == "" {
		return ""
	}
	if filepath.IsAbs(path) {
		return filepath.Clean(path)
	}
	sessionPath := cptilt.SessionAbsolutePath(stateDir, path)
	if _, err := os.Stat(sessionPath); err == nil {
		return sessionPath
	}
	// Older generated reports sometimes stored cwd-relative session paths such
	// as .cp9/<session>/kubeconfig.yaml instead of session-relative paths.
	legacyPath := filepath.Clean(filepath.FromSlash(path))
	if _, err := os.Stat(legacyPath); err == nil {
		return legacyPath
	}
	return sessionPath
}