package tilt

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/tidbcloud/controlplane9/internal/artifactsync"
	"github.com/tidbcloud/controlplane9/internal/config"
	"github.com/tidbcloud/controlplane9/internal/host"
	"github.com/tidbcloud/controlplane9/internal/verbose"
)

const (
	ControllerLocal  = "local"
	ControllerRemote = "remote"

	remoteControllerDirName = "controller"
	remoteRepositoryName    = "controlplane9"
	remoteUnitPrefix        = "cp9-tilt-"
)

// ponytail: one cp9 process owns one remote controller; serialize its rare
// directory publications until a process can legitimately operate multiple
// sessions.
var remoteDirectorySyncMu sync.Mutex

// RemoteMaterialization is the path handoff from a local cp9 command to one
// remote-host-owned Tilt controller.
type RemoteMaterialization struct {
	Root             string
	TiltfilePath     string
	ContextIndexPath string
	KubeconfigPath   string
	ReportsDir       string
	EvidenceDir      string
	LogPath          string
	Unit             string
	Request          *Request
	changedServices  map[string]bool
}

// RemoteObservation is the authoritative systemd state of a remote Tilt
// controller.
type RemoteObservation struct {
	LoadState   string
	ActiveState string
	SubState    string
	Result      string
	PID         int
}

type sourceContractDocument struct {
	Contracts []sourceContract `json:"contracts"`
}

type sourceContract struct {
	Name   string               `json:"name"`
	Source sourceContractSource `json:"source"`
}

type sourceContractSource struct {
	Repo string `json:"repo"`
	Main string `json:"main"`
}

// RemoteControllerRepositoryRoot returns the synced CP9 repository used by a
// remote Tilt controller.
func RemoteControllerRepositoryRoot(status *Status) (string, error) {
	if status == nil || status.Controller != ControllerRemote {
		return "", fmt.Errorf("remote Tilt controller status is required")
	}
	if strings.TrimSpace(status.RemoteRoot) == "" {
		return "", fmt.Errorf("remote Tilt controller root is missing")
	}
	return path.Join(status.RemoteRoot, "work", remoteRepositoryName), nil
}

func (o *RemoteObservation) Running() bool {
	return o != nil && o.ActiveState == "active" && o.PID > 0
}

// UsesRemoteController reports whether the selected host owns its Tilt
// controller.
func UsesRemoteController(selected *host.Session) bool {
	return selected != nil && selected.Remote != nil &&
		(selected.Mode == config.HostModeRemoteEC2K3d || selected.Mode == config.HostModeRemoteGCEK3d)
}

// UsesRemoteControllerForward reports whether local Tilt UI access needs the
// existing raw-SSH tunnel. GCE day-2 commands query Tilt through IAP directly.
func UsesRemoteControllerForward(selected *host.Session) bool {
	return selected != nil && selected.Mode == config.HostModeRemoteEC2K3d && selected.Remote != nil
}

// MaterializeRemoteController copies the session-owned runtime inputs to the
// selected remote host and rewrites the context contract to remote paths.
func MaterializeRemoteController(
	ctx context.Context,
	selected *host.Session,
	localTiltfile string,
	localContextIndex string,
	req *Request,
) (*RemoteMaterialization, error) {
	return materializeRemoteController(ctx, selected, localTiltfile, localContextIndex, req, true)
}

// RefreshRemoteControllerInputs overlays the current checkout and selected
// source repositories onto a running remote controller without replacing the
// controller or its resource graph.
func RefreshRemoteControllerInputs(
	ctx context.Context,
	selected *host.Session,
	stateDir string,
	status *Status,
) (map[string]bool, error) {
	changedServices := map[string]bool{}
	if status == nil || status.Controller != ControllerRemote {
		return changedServices, nil
	}
	localTiltfile := SessionAbsolutePath(stateDir, status.TiltfilePath)
	localContext := SessionAbsolutePath(stateDir, status.ContextIndexPath)
	materialized, err := materializeRemoteController(
		ctx,
		selected,
		localTiltfile,
		localContext,
		&Request{Port: status.APIPort},
		false,
	)
	if err != nil {
		return nil, err
	}
	return materialized.changedServices, nil
}

// RefreshRemoteArtifactContext copies only the resolved artifact snapshot and
// selected artifact content. It deliberately does not publish CP9 or service
// source trees. The returned set contains selected artifact resources whose
// remote bytes changed; recovery reports a prior partial transaction that
// remains pending until the command observes runtime convergence.
func RefreshRemoteArtifactContext(
	ctx context.Context,
	selected *host.Session,
	artifactStateDir, runtimeStateDir string,
	status *Status,
	artifacts map[string]bool,
	runtimeArtifacts map[string]bool,
	targetSnapshot *artifactsync.Snapshot,
) (changed, recovered map[string]bool, err error) {
	return refreshRemoteArtifactContext(
		ctx, selected, artifactStateDir, runtimeStateDir, status,
		artifacts, runtimeArtifacts, targetSnapshot, true,
	)
}

// RestoreRemoteArtifactContext republishes the prior generation during an
// in-process rollback. A fresh command owns the marker it is retargeting;
// normal recovery remains generation-strict and forward-only.
func RestoreRemoteArtifactContext(
	ctx context.Context,
	selected *host.Session,
	artifactStateDir, runtimeStateDir string,
	status *Status,
	artifacts map[string]bool,
	runtimeArtifacts map[string]bool,
	targetSnapshot *artifactsync.Snapshot,
) (changed, recovered map[string]bool, err error) {
	return refreshRemoteArtifactContext(
		ctx, selected, artifactStateDir, runtimeStateDir, status,
		artifacts, runtimeArtifacts, targetSnapshot, false,
	)
}

func refreshRemoteArtifactContext(
	ctx context.Context,
	selected *host.Session,
	artifactStateDir, runtimeStateDir string,
	status *Status,
	artifacts map[string]bool,
	runtimeArtifacts map[string]bool,
	targetSnapshot *artifactsync.Snapshot,
	validatePendingGeneration bool,
) (changed, recovered map[string]bool, err error) {
	changed = map[string]bool{}
	recovered = map[string]bool{}
	if status == nil || status.Controller != ControllerRemote {
		return changed, recovered, nil
	}
	if strings.TrimSpace(status.RemoteRoot) == "" || strings.TrimSpace(status.RemoteContextIndexPath) == "" {
		return nil, recovered, fmt.Errorf("refresh remote artifact context: persisted remote paths are missing")
	}
	if targetSnapshot == nil {
		return nil, recovered, errors.New("refresh remote artifact context: target snapshot is required")
	}
	materialized := &RemoteMaterialization{
		Root:             status.RemoteRoot,
		ContextIndexPath: status.RemoteContextIndexPath,
		KubeconfigPath:   status.RemoteKubeconfigPath,
		ReportsDir:       path.Join(status.RemoteRoot, "session", "reports"),
		EvidenceDir:      path.Join(status.RemoteRoot, "session", "reports", "evidence"),
		LogPath:          path.Join(status.RemoteRoot, "session", "tilt-up.log"),
	}
	localContext := SessionAbsolutePath(runtimeStateDir, status.ContextIndexPath)
	artifactRoot := path.Join(materialized.Root, "session", "artifacts")
	plannedRuntime := cloneArtifactNames(runtimeArtifacts)
	pendingData, pending, err := readRemoteRefreshTransaction(ctx, selected, path.Join(artifactRoot, artifactsync.RefreshTransactionFile))
	if err != nil {
		return nil, recovered, err
	}
	if pending {
		pendingTransaction, parseErr := artifactsync.ParseRefreshTransaction(pendingData)
		if parseErr != nil {
			return nil, recovered, parseErr
		}
		pendingSelection := map[string]bool{}
		for _, name := range pendingTransaction.Artifacts {
			pendingSelection[name] = true
		}
		if err := artifactsync.ValidateRefreshSelection(pendingSelection, artifacts); err != nil {
			return nil, pendingSelection, fmt.Errorf("remote %w", err)
		}
		if validatePendingGeneration {
			if err := artifactsync.ValidateRefreshGeneration(pendingTransaction, targetSnapshot.Digest); err != nil {
				return nil, pendingSelection, fmt.Errorf("remote %w", err)
			}
			plannedRuntime = map[string]bool{}
		}
		for _, name := range pendingTransaction.RuntimeArtifacts {
			plannedRuntime[name] = true
		}
		if validatePendingGeneration {
			recovered = cloneArtifactNames(plannedRuntime)
		}
	}
	if !pending || !validatePendingGeneration {
		remoteSnapshotData, remoteSnapshotPresent, readErr := readRemoteArtifactFile(
			ctx, selected, path.Join(artifactRoot, artifactsync.SnapshotFile), "artifact snapshot",
		)
		if readErr != nil {
			return nil, recovered, readErr
		}
		remoteTargets, targetErr := remoteArtifactRuntimeTargets(remoteSnapshotData, remoteSnapshotPresent, artifacts, targetSnapshot)
		if targetErr != nil {
			return nil, recovered, targetErr
		}
		for name := range remoteTargets {
			plannedRuntime[name] = true
		}
	}
	transaction, err := artifactsync.RefreshTransactionData(
		artifactsync.OutputDir(artifactStateDir), artifacts, plannedRuntime, targetSnapshot,
	)
	if err != nil {
		return nil, recovered, err
	}
	transactionPath := path.Join(artifactRoot, artifactsync.RefreshTransactionFile)
	if err := WriteRemoteFile(ctx, selected, transactionPath, transaction, 0o600); err != nil {
		return nil, recovered, fmt.Errorf("write remote artifact refresh transaction: %w", err)
	}
	if err := materializeRemoteContext(ctx, selected, localContext, materialized, plannedRuntime, changed); err != nil {
		return changed, recovered, err
	}
	markerChanged := false
	for name := range changed {
		if !plannedRuntime[name] {
			plannedRuntime[name] = true
			markerChanged = true
		}
	}
	if markerChanged {
		transaction, err = artifactsync.RefreshTransactionData(
			artifactsync.OutputDir(artifactStateDir), artifacts, plannedRuntime, targetSnapshot,
		)
		if err != nil {
			return changed, recovered, err
		}
		if err := WriteRemoteFile(ctx, selected, transactionPath, transaction, 0o600); err != nil {
			return changed, recovered, fmt.Errorf("update remote artifact refresh transaction: %w", err)
		}
	}
	return changed, recovered, nil
}

func cloneArtifactNames(names map[string]bool) map[string]bool {
	cloned := make(map[string]bool, len(names))
	for name, enabled := range names {
		if enabled {
			cloned[name] = true
		}
	}
	return cloned
}

func remoteArtifactRuntimeTargets(
	data []byte,
	present bool,
	selected map[string]bool,
	target *artifactsync.Snapshot,
) (map[string]bool, error) {
	previous := map[string]string{}
	if present {
		var snapshot artifactsync.Snapshot
		if err := json.Unmarshal(data, &snapshot); err != nil {
			return nil, fmt.Errorf("parse remote artifact snapshot: %w", err)
		}
		if snapshot.APIVersion != artifactsync.SnapshotVersion || strings.TrimSpace(snapshot.Digest) == "" {
			return nil, errors.New("remote artifact snapshot is not supported")
		}
		for i := range snapshot.Artifacts {
			previous[snapshot.Artifacts[i].Name] = snapshot.Artifacts[i].Digest
		}
	}
	targets := map[string]bool{}
	for i := range target.Artifacts {
		artifact := &target.Artifacts[i]
		if selected[artifact.Name] && previous[artifact.Name] != artifact.Digest {
			targets[artifact.Name] = true
		}
	}
	return targets, nil
}

// ClearRemoteArtifactRefreshTransaction removes the remote marker after the
// caller has observed runtime convergence on the authoritative controller.
func ClearRemoteArtifactRefreshTransaction(ctx context.Context, selected *host.Session, status *Status) error {
	if status == nil || status.Controller != ControllerRemote {
		return nil
	}
	if strings.TrimSpace(status.RemoteRoot) == "" {
		return fmt.Errorf("clear remote artifact refresh transaction: persisted remote root is missing")
	}
	transactionPath := path.Join(status.RemoteRoot, "session", "artifacts", artifactsync.RefreshTransactionFile)
	if err := removeRemoteFile(
		ctx,
		selected,
		transactionPath,
	); err != nil {
		return err
	}
	_, present, err := readRemoteRefreshTransaction(ctx, selected, transactionPath)
	if err != nil {
		return err
	}
	if present {
		return errors.New("remote artifact refresh transaction marker remains after clear")
	}
	return nil
}

// SyncRemoteControllerArtifacts mirrors remote evidence into the existing
// local session artifact paths. Local controller metadata such as
// tilt-status.json and context-report.json remains local.
func SyncRemoteControllerArtifacts(
	ctx context.Context,
	selected *host.Session,
	status *Status,
	stateDir string,
) error {
	if !UsesRemoteController(selected) || status == nil || status.Controller != ControllerRemote {
		return nil
	}
	if strings.TrimSpace(status.RemoteRoot) == "" {
		return fmt.Errorf("sync remote Tilt artifacts: persisted remote root is missing")
	}
	remoteEvidence := path.Join(status.RemoteRoot, "session", "reports", "evidence")
	localEvidence := filepath.Join(stateDir, "tilt", "reports", "evidence")
	return SyncRemoteArtifacts(
		ctx,
		selected,
		remoteEvidence,
		localEvidence,
		[]string{"dbaas-ui-readiness.json"},
	)
}

func materializeRemoteController(
	ctx context.Context,
	selected *host.Session,
	localTiltfile string,
	localContextIndex string,
	req *Request,
	replace bool,
) (*RemoteMaterialization, error) {
	changedServices := map[string]bool{}
	if !UsesRemoteController(selected) {
		return nil, fmt.Errorf("remote Tilt controller requires a persisted remote k3d host")
	}
	if strings.TrimSpace(localTiltfile) == "" {
		return nil, fmt.Errorf("remote Tilt controller requires a Tiltfile")
	}
	if strings.TrimSpace(localContextIndex) == "" {
		return nil, fmt.Errorf("remote Tilt controller requires a session context")
	}
	localTiltfile, err := filepath.Abs(localTiltfile)
	if err != nil {
		return nil, fmt.Errorf("resolve local Tiltfile: %w", err)
	}
	localContextIndex, err = filepath.Abs(localContextIndex)
	if err != nil {
		return nil, fmt.Errorf("resolve local session context: %w", err)
	}
	if _, err := os.Stat(localTiltfile); err != nil {
		return nil, fmt.Errorf("stat local Tiltfile: %w", err)
	}
	if _, err := os.Stat(localContextIndex); err != nil {
		return nil, fmt.Errorf("stat local session context: %w", err)
	}

	remoteRoot := path.Join(selected.Remote.RemoteStateDir, remoteControllerDirName)
	remoteRepositoryRoot := path.Join(remoteRoot, "work", remoteRepositoryName)
	var remoteTiltfile string
	if repositoryRoot, ok := findRepositoryRoot(localTiltfile); ok {
		relativeTiltfile, relErr := filepath.Rel(repositoryRoot, localTiltfile)
		if relErr != nil || relativeTiltfile == ".." ||
			strings.HasPrefix(relativeTiltfile, ".."+string(filepath.Separator)) {
			return nil, fmt.Errorf("resolve Tiltfile relative to repository root %s", repositoryRoot)
		}
		if err := syncDirectoryToRemote(ctx, selected, repositoryRoot, remoteRepositoryRoot, replace); err != nil {
			return nil, fmt.Errorf("sync controller repository: %w", err)
		}
		remoteTiltfile = path.Join(remoteRepositoryRoot, filepath.ToSlash(relativeTiltfile))
		sourceReceipt, err := syncSelectedSourceRepositories(
			ctx,
			selected,
			localTiltfile,
			remoteTiltfile,
			localContextIndex,
		)
		if err != nil {
			return nil, err
		}
		changedServices = sourceReceipt
	} else {
		remoteGeneratedRoot := path.Join(remoteRoot, "generated-tilt")
		if err := syncDirectoryToRemote(ctx, selected, filepath.Dir(localTiltfile), remoteGeneratedRoot, replace); err != nil {
			return nil, fmt.Errorf("sync generated Tilt handoff: %w", err)
		}
		remoteTiltfile = path.Join(remoteGeneratedRoot, filepath.Base(localTiltfile))
	}

	materialized := &RemoteMaterialization{
		Root:             remoteRoot,
		TiltfilePath:     remoteTiltfile,
		ContextIndexPath: path.Join(remoteRoot, "session", "context.json"),
		KubeconfigPath:   selected.Remote.RemoteKubeconfigPath,
		ReportsDir:       path.Join(remoteRoot, "session", "reports"),
		EvidenceDir:      path.Join(remoteRoot, "session", "reports", "evidence"),
		LogPath:          path.Join(remoteRoot, "session", "tilt-up.log"),
		Unit:             remoteControllerUnit(selected.ClusterName),
		changedServices:  changedServices,
	}
	if err := materializeRemoteContext(
		ctx,
		selected,
		localContextIndex,
		materialized,
		nil,
		nil,
	); err != nil {
		return nil, err
	}
	if err := removeRemoteFile(
		ctx,
		selected,
		path.Join(materialized.Root, "session", "artifacts", artifactsync.RefreshTransactionFile),
	); err != nil {
		return nil, fmt.Errorf("clear recovered remote artifact refresh transaction: %w", err)
	}
	materialized.Request = &Request{
		Tiltfile:    materialized.TiltfilePath,
		KubeContext: selected.Context,
		Kubeconfig:  materialized.KubeconfigPath,
		Host:        "127.0.0.1",
		Port:        normalizedRemotePort(req),
		LogPath:     materialized.LogPath,
	}
	return materialized, nil
}

// StartRemoteController installs and starts the session-scoped systemd unit,
// then waits for the remote Tilt API. The unit survives local command
// interruption and laptop disconnects.
func StartRemoteController(
	ctx context.Context,
	selected *host.Session,
	materialized *RemoteMaterialization,
) (*StartedProcess, error) {
	if !UsesRemoteController(selected) || materialized == nil || materialized.Request == nil {
		return nil, fmt.Errorf("start remote Tilt controller: materialized remote runtime is required")
	}
	unit := renderRemoteControllerUnit(selected, materialized)
	unitPath := "/etc/systemd/system/" + materialized.Unit
	script := fmt.Sprintf(
		"set -eu; sudo tee %s >/dev/null; sudo systemctl daemon-reload; sudo systemctl enable %s >/dev/null; sudo systemctl restart %s",
		host.ShellEscape(unitPath),
		host.ShellEscape(materialized.Unit),
		host.ShellEscape(materialized.Unit),
	)
	if _, err := host.RunSSH(ctx, selected, unit, "sh", "-lc", script); err != nil {
		return nil, fmt.Errorf("start remote Tilt systemd unit: %w", err)
	}
	observation, err := waitForRemoteController(ctx, selected, materialized)
	if err != nil {
		logs, _ := RemoteControllerJournal(ctx, selected, materialized.Unit, 200)
		if len(logs) > 0 {
			return nil, fmt.Errorf("%w\n%s", err, strings.TrimSpace(string(logs)))
		}
		return nil, err
	}
	startedAt := time.Now().UTC()
	return &StartedProcess{
		PID: observation.PID,
		Command: []string{
			"systemctl", "start", materialized.Unit,
		},
		LogPath:   materialized.LogPath,
		StartedAt: startedAt,
	}, nil
}

// RunRemoteCI runs synchronous Tilt CI on the selected remote host.
func RunRemoteCI(
	ctx context.Context,
	selected *host.Session,
	materialized *RemoteMaterialization,
) error {
	if !UsesRemoteController(selected) || materialized == nil || materialized.Request == nil {
		return fmt.Errorf("remote Tilt CI: materialized remote runtime is required")
	}
	request := materialized.Request
	argv := []string{
		"env",
		"KUBECONFIG=" + request.Kubeconfig,
		"CP9_SESSION_CONTEXT=" + materialized.ContextIndexPath,
	}
	argv = append(argv, remoteControllerEnvironment(selected)...)
	argv = append(argv, "tilt")
	argv = append(argv, ciArgs(request)...)
	if _, err := host.RunSSH(ctx, selected, nil, argv...); err != nil {
		return fmt.Errorf("remote tilt ci: %w", err)
	}
	return nil
}

// ObserveRemoteController reads the remote controller state from systemd.
func ObserveRemoteController(
	ctx context.Context,
	selected *host.Session,
	status *Status,
) (*RemoteObservation, error) {
	if !UsesRemoteController(selected) {
		return nil, fmt.Errorf("observe remote Tilt controller: remote host is required")
	}
	unit := remoteControllerUnit(selected.ClusterName)
	if status != nil && strings.TrimSpace(status.RemoteUnit) != "" {
		unit = status.RemoteUnit
	}
	output, err := host.RunSSH(
		ctx,
		selected,
		nil,
		"systemctl",
		"show",
		unit,
		"--property=LoadState,ActiveState,SubState,Result,MainPID",
	)
	if err != nil {
		return nil, fmt.Errorf("observe remote Tilt controller %s: %w", unit, err)
	}
	values := map[string]string{}
	for line := range strings.SplitSeq(strings.TrimSpace(string(output)), "\n") {
		key, value, ok := strings.Cut(line, "=")
		if ok {
			values[strings.TrimSpace(key)] = strings.TrimSpace(value)
		}
	}
	if values["ActiveState"] == "" {
		return nil, fmt.Errorf(
			"observe remote Tilt controller %s: unexpected systemd response %q",
			unit,
			strings.TrimSpace(string(output)),
		)
	}
	pid, _ := strconv.Atoi(values["MainPID"])
	return &RemoteObservation{
		LoadState:   values["LoadState"],
		ActiveState: values["ActiveState"],
		SubState:    values["SubState"],
		Result:      values["Result"],
		PID:         pid,
	}, nil
}

// ReconcilePersistedRuntimeForHost makes cp9 up idempotently resume the
// authoritative controller for the selected host.
func ReconcilePersistedRuntimeForHost(
	ctx context.Context,
	selected *host.Session,
	statusPath string,
	stateDir string,
) (*RuntimeReconcileResult, error) {
	if !UsesRemoteController(selected) {
		return ReconcilePersistedRuntime(ctx, statusPath, stateDir)
	}
	status, err := LoadStatus(statusPath)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return &RuntimeReconcileResult{}, nil
		}
		return nil, err
	}
	if status.Mode != "up" {
		return &RuntimeReconcileResult{}, nil
	}
	status.Controller = ControllerRemote
	observation, err := ObserveRemoteController(ctx, selected, status)
	if err != nil {
		return nil, err
	}
	if observation.Running() {
		status.Running = true
		status.PID = observation.PID
		if err := SaveStatus(statusPath, status); err != nil {
			return nil, err
		}
		message := ""
		if UsesRemoteControllerForward(selected) {
			message, err = EnsureRemoteControllerForward(ctx, selected, statusPath, stateDir)
			if err != nil {
				return nil, err
			}
		}
		return &RuntimeReconcileResult{Message: message}, nil
	}
	status.Running = false
	status.PID = 0
	status.StartedAt = nil
	if err := SaveStatus(statusPath, status); err != nil {
		return nil, err
	}
	return &RuntimeReconcileResult{
		RestartRequired: true,
		Message: fmt.Sprintf(
			"remote Tilt controller requires restart (active=%s sub=%s result=%s)",
			observation.ActiveState,
			observation.SubState,
			observation.Result,
		),
	}, nil
}

// ResourceReadinessForHost reads the correct controller, without requiring a
// local Tilt process or tunnel for remote sessions.
func ResourceReadinessForHost(
	ctx context.Context,
	selected *host.Session,
	status *Status,
) (*ResourceReadiness, error) {
	if status == nil {
		return nil, fmt.Errorf("tilt status is required")
	}
	if !UsesRemoteController(selected) || status.Controller != ControllerRemote {
		return NewCLI().ResourceReadiness(ctx, status.APIHost, status.APIPort)
	}
	// UIResource snapshots must not contend with the shared SSH connection
	// that owns the long-lived controller forward and transfer channels.
	argv := append([]string{"tilt"}, resourceReadinessArgs("127.0.0.1", status.APIPort)...)
	output, err := host.RunSSHIsolated(ctx, selected, nil, argv...)
	if err != nil {
		return nil, fmt.Errorf("remote tilt get uiresources: %w", err)
	}
	return ParseUIResourceReadiness(output)
}

// TriggerForHost triggers a resource on the authoritative controller.
func TriggerForHost(ctx context.Context, selected *host.Session, status *Status, resource string) error {
	if status == nil {
		return fmt.Errorf("tilt status is required")
	}
	if !UsesRemoteController(selected) || status.Controller != ControllerRemote {
		return NewCLI().Trigger(ctx, status.APIHost, status.APIPort, resource)
	}
	_, err := host.RunSSH(
		ctx,
		selected,
		nil,
		append([]string{"tilt"}, triggerArgs("127.0.0.1", status.APIPort, resource)...)...,
	)
	if err != nil {
		return fmt.Errorf("remote tilt trigger %s: %w", resource, err)
	}
	return nil
}

// LogsForHost reads resource logs from the authoritative controller.
func LogsForHost(
	ctx context.Context,
	selected *host.Session,
	status *Status,
	req *LogsRequest,
) ([]byte, error) {
	if status == nil {
		return nil, fmt.Errorf("tilt status is required")
	}
	if !UsesRemoteController(selected) || status.Controller != ControllerRemote {
		return NewCLI().Logs(ctx, req)
	}
	remoteReq := *req
	remoteReq.Host = "127.0.0.1"
	remoteReq.Port = status.APIPort
	output, err := host.RunSSH(ctx, selected, nil, append([]string{"tilt"}, logsArgs(&remoteReq)...)...)
	if err != nil {
		return nil, fmt.Errorf("remote tilt logs: %w", err)
	}
	return output, nil
}

// DownForHost stops the remote controller before deleting its Kubernetes
// objects. Local sessions retain the existing direct Tilt behavior.
func DownForHost(
	ctx context.Context,
	selected *host.Session,
	status *Status,
	req *Request,
) error {
	if !UsesRemoteController(selected) || status == nil || status.Controller != ControllerRemote {
		return NewCLI().Down(ctx, req)
	}
	unit := remoteControllerUnit(selected.ClusterName)
	if strings.TrimSpace(status.RemoteUnit) != "" {
		unit = status.RemoteUnit
	}
	observation, err := ObserveRemoteController(ctx, selected, status)
	if err != nil {
		return fmt.Errorf("remote tilt stop observation: %w", err)
	}
	if observation.LoadState != "not-found" {
		if _, err := host.RunSSH(ctx, selected, nil, "sudo", "systemctl", "disable", "--now", unit); err != nil {
			return fmt.Errorf("remote tilt stop: %w", err)
		}
	}
	stopped := "if sudo systemctl is-active --quiet " + host.ShellEscape(unit) +
		"; then echo 'remote Tilt unit is still active' >&2; exit 1; fi"
	if _, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", stopped); err != nil {
		return fmt.Errorf("remote tilt stop verification: %w", err)
	}
	remoteReq := &Request{
		Tiltfile:    status.RemoteTiltfilePath,
		KubeContext: selected.Context,
		Kubeconfig:  status.RemoteKubeconfigPath,
	}
	if strings.TrimSpace(remoteReq.Tiltfile) == "" || strings.TrimSpace(remoteReq.Kubeconfig) == "" {
		return fmt.Errorf("remote tilt down: persisted remote Tiltfile and kubeconfig paths are required")
	}
	argv := []string{
		"env",
		"KUBECONFIG=" + remoteReq.Kubeconfig,
		"CP9_SESSION_CONTEXT=" + status.RemoteContextIndexPath,
		"tilt",
	}
	argv = append(argv, downArgs(remoteReq)...)
	if _, err := host.RunSSH(ctx, selected, nil, argv...); err != nil {
		return fmt.Errorf("remote tilt down: %w", err)
	}
	cleanup := "set -eu; sudo rm -f " + host.ShellEscape("/etc/systemd/system/"+unit) +
		"; sudo systemctl daemon-reload; " + stopped
	if _, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", cleanup); err != nil {
		return fmt.Errorf("remote tilt unit cleanup: %w", err)
	}
	return nil
}

// StartRemoteControllerForward exposes the EC2 loopback-only Tilt API on the
// same local port. The controller remains remote-authoritative if this
// disposable local forward exits.
func StartRemoteControllerForward(
	ctx context.Context,
	selected *host.Session,
	stateDir string,
	port int,
) (*StartedProcess, error) {
	if !UsesRemoteControllerForward(selected) {
		return nil, fmt.Errorf("remote Tilt controller forward requires a raw-SSH remote host")
	}
	if port <= 0 || port > 65535 {
		return nil, fmt.Errorf("remote Tilt controller forward has invalid port %d", port)
	}
	baseArgs, err := host.SSHBaseArgs(selected)
	if err != nil {
		return nil, fmt.Errorf("remote Tilt controller forward: %w", err)
	}
	if len(baseArgs) == 0 {
		return nil, fmt.Errorf("remote Tilt controller forward: empty ssh arguments")
	}
	destination := baseArgs[len(baseArgs)-1]
	args := append([]string{}, baseArgs[:len(baseArgs)-1]...)
	args = append(
		args,
		"-o", "ExitOnForwardFailure=yes",
		"-N",
		"-L", fmt.Sprintf("127.0.0.1:%d:127.0.0.1:%d", port, port),
		destination,
	)
	supervisorPath, childPIDPath, err := prepareTunnelSupervisor(stateDir, "controller-tunnel")
	if err != nil {
		return nil, fmt.Errorf("remote Tilt controller forward: %w", err)
	}
	command := append([]string{supervisorPath, childPIDPath, "ssh"}, args...)
	logPath := filepath.Join(stateDir, "tilt", "controller-tunnel.log")
	started, err := startKubeAPITunnelCommand(ctx, command, logPath)
	if err != nil {
		return nil, fmt.Errorf("remote Tilt controller forward: %w", err)
	}
	return started, nil
}

// EnsureRemoteControllerForward repairs only the disposable local UI/API
// forward. It never treats a local PID as the controller state.
func EnsureRemoteControllerForward(
	ctx context.Context,
	selected *host.Session,
	statusPath string,
	stateDir string,
) (string, error) {
	status, err := LoadStatus(statusPath)
	if err != nil {
		return "", err
	}
	if status.Controller != ControllerRemote || !status.Running {
		return "", nil
	}
	supervisorPath, err := tunnelSupervisorPath(stateDir, "controller-tunnel")
	if err != nil {
		return "", err
	}
	pids, err := tunnelSupervisorPIDs(ctx, supervisorPath)
	if err != nil {
		return "", err
	}
	if len(pids) == 1 && pids[0] == status.ControllerTunnelPID {
		command, commandErr := processCommand(ctx, pids[0])
		if commandErr == nil && controllerTunnelMatches(status, command) {
			return "", nil
		}
	}
	if _, stopErr := NewProcessStopper().stopRecordedControllerTunnelChild(ctx, status, stateDir); stopErr != nil {
		return "", stopErr
	}
	if _, stopErr := stopExactTunnelSupervisors(ctx, supervisorPath); stopErr != nil {
		return "", stopErr
	}
	started, err := StartRemoteControllerForward(ctx, selected, stateDir, status.APIPort)
	if err != nil {
		return "", err
	}
	recordControllerTunnel(stateDir, status, started)
	if err := SaveStatus(statusPath, status); err != nil {
		StopStartedProcess(started)
		return "", err
	}
	if len(pids) == 0 {
		return fmt.Sprintf("started remote Tilt controller forward %d", started.PID), nil
	}
	return fmt.Sprintf("replaced remote Tilt controller forwards %v -> %d", pids, started.PID), nil
}

// StopRemoteControllerForward stops the disposable local forward and clears
// its persisted metadata.
func StopRemoteControllerForward(
	ctx context.Context,
	statusPath string,
	stateDir string,
) (string, error) {
	status, err := LoadStatus(statusPath)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			pids, stopErr := stopExactTunnelSupervisorsForRole(ctx, stateDir, "controller-tunnel")
			if stopErr != nil {
				return "", stopErr
			}
			if len(pids) == 0 {
				return "", nil
			}
			return fmt.Sprintf("remote Tilt controller forwards stopped without status: %v", pids), nil
		}
		return "", err
	}
	childMessage, childErr := NewProcessStopper().stopRecordedControllerTunnelChild(ctx, status, stateDir)
	if childErr != nil {
		return "", childErr
	}
	pids, err := stopExactTunnelSupervisorsForRole(ctx, stateDir, "controller-tunnel")
	if err != nil {
		return "", err
	}
	clearControllerTunnel(status)
	if err := SaveStatus(statusPath, status); err != nil {
		return "", err
	}
	stopMessage := ""
	if len(pids) > 0 {
		stopMessage = fmt.Sprintf("remote Tilt controller forwards stopped: %v", pids)
	}
	return strings.Join(compactStopMessages(stopMessage, childMessage), "; "), nil
}

// RemoteControllerJournal returns bounded systemd logs for diagnosis.
func RemoteControllerJournal(
	ctx context.Context,
	selected *host.Session,
	unit string,
	tail int,
) ([]byte, error) {
	if tail <= 0 {
		tail = 200
	}
	return host.RunSSH(
		ctx,
		selected,
		nil,
		"journalctl",
		"--unit",
		unit,
		"--no-pager",
		"--output=cat",
		"--lines",
		strconv.Itoa(tail),
	)
}

func waitForRemoteController(
	ctx context.Context,
	selected *host.Session,
	materialized *RemoteMaterialization,
) (*RemoteObservation, error) {
	deadline := time.Now().Add(defaultReadyTimeout)
	var lastObservation *RemoteObservation
	var lastErr error
	for time.Now().Before(deadline) {
		observation, err := ObserveRemoteController(ctx, selected, &Status{RemoteUnit: materialized.Unit})
		if err == nil {
			lastObservation = observation
			if observation.Running() {
				probe := &Status{
					Controller: ControllerRemote,
					APIPort:    materialized.Request.Port,
				}
				_, probeErr := ResourceReadinessForHost(ctx, selected, probe)
				if probeErr == nil {
					return observation, nil
				}
				lastErr = probeErr
			} else if observation.ActiveState == "failed" || observation.Result == "failed" {
				return nil, fmt.Errorf(
					"remote Tilt controller %s failed (active=%s sub=%s result=%s)",
					materialized.Unit,
					observation.ActiveState,
					observation.SubState,
					observation.Result,
				)
			}
		} else {
			lastErr = err
		}
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(defaultReadyPollInterval):
		}
	}
	if lastObservation != nil {
		return nil, fmt.Errorf(
			"remote Tilt controller %s did not become ready (active=%s sub=%s result=%s): %w",
			materialized.Unit,
			lastObservation.ActiveState,
			lastObservation.SubState,
			lastObservation.Result,
			lastErr,
		)
	}
	return nil, fmt.Errorf("remote Tilt controller %s did not become observable: %w", materialized.Unit, lastErr)
}

func renderRemoteControllerUnit(selected *host.Session, materialized *RemoteMaterialization) []byte {
	request := materialized.Request
	user := strings.TrimSpace(selected.Remote.SSHUser)
	if user == "" {
		user = "ec2-user"
	}
	execArgs := append([]string{"/usr/local/bin/tilt"}, upArgs(request)...)
	var b strings.Builder
	b.WriteString("[Unit]\n")
	b.WriteString("Description=CP9 remote Tilt controller for " + systemdEscape(selected.ClusterName) + "\n")
	b.WriteString("After=network-online.target docker.service\n")
	b.WriteString("Wants=network-online.target\n\n")
	b.WriteString("[Service]\n")
	b.WriteString("Type=simple\n")
	b.WriteString("User=" + systemdEscape(user) + "\n")
	b.WriteString("WorkingDirectory=" + systemdEscape(filepath.ToSlash(filepath.Dir(request.Tiltfile))) + "\n")
	b.WriteString("Environment=HOME=" + systemdEscape(remoteHome(selected)) + "\n")
	b.WriteString("Environment=PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin\n")
	b.WriteString("Environment=KUBECONFIG=" + systemdEscape(request.Kubeconfig) + "\n")
	b.WriteString("Environment=CP9_SESSION_CONTEXT=" + systemdEscape(materialized.ContextIndexPath) + "\n")
	for _, value := range remoteControllerEnvironment(selected) {
		b.WriteString("Environment=" + systemdEscape(value) + "\n")
	}
	b.WriteString("ExecStart=" + systemdJoin(execArgs) + "\n")
	b.WriteString("Restart=on-failure\n")
	b.WriteString("RestartSec=3\n")
	b.WriteString("KillMode=mixed\n")
	b.WriteString("TimeoutStopSec=30\n")
	b.WriteString("StandardOutput=append:" + systemdEscape(materialized.LogPath) + "\n")
	b.WriteString("StandardError=append:" + systemdEscape(materialized.LogPath) + "\n\n")
	b.WriteString("[Install]\n")
	b.WriteString("WantedBy=multi-user.target\n")
	return []byte(b.String())
}

func materializeRemoteContext(
	ctx context.Context,
	selected *host.Session,
	localContextPath string,
	materialized *RemoteMaterialization,
	selectedArtifacts map[string]bool,
	changedArtifacts map[string]bool,
) error {
	data, err := os.ReadFile(localContextPath)
	if err != nil {
		return fmt.Errorf("read local session context: %w", err)
	}
	var raw map[string]any
	if err := json.Unmarshal(data, &raw); err != nil {
		return fmt.Errorf("parse local session context: %w", err)
	}
	raw["kubeconfig_path"] = materialized.KubeconfigPath
	raw["tiltfile_path"] = materialized.TiltfilePath
	raw["reports_dir"] = materialized.ReportsDir
	raw["evidence_dir"] = materialized.EvidenceDir

	sessionDir := path.Join(materialized.Root, "session")
	if localPath := stringValue(raw["resolved_session_path"]); localPath != "" {
		remotePath := path.Join(sessionDir, "resolved-session.yaml")
		if err := copyFileToRemote(ctx, selected, localPath, remotePath); err != nil {
			return fmt.Errorf("copy resolved session to remote controller: %w", err)
		}
		raw["resolved_session_path"] = remotePath
	}
	if terraform, ok := raw["terraform"].(map[string]any); ok {
		for _, key := range []string{"outputs_path", "contract_path"} {
			if localPath := stringValue(terraform[key]); localPath != "" {
				remotePath := path.Join(sessionDir, filepath.Base(localPath))
				if err := copyFileToRemote(ctx, selected, localPath, remotePath); err != nil {
					return fmt.Errorf("copy Terraform %s to remote controller: %w", key, err)
				}
				terraform[key] = remotePath
			}
		}
	}
	if ui, ok := raw["ui"].(map[string]any); ok {
		if localPath := stringValue(ui["runtime_config_path"]); localPath != "" {
			remotePath := path.Join(sessionDir, "dbaas-ui-config.js")
			if err := copyFileToRemote(ctx, selected, localPath, remotePath); err != nil {
				return fmt.Errorf("copy DBaaS UI runtime config to remote controller: %w", err)
			}
			ui["runtime_config_path"] = remotePath
		}
	}
	if artifacts, ok := raw["artifacts"].(map[string]any); ok {
		if localPath := stringValue(artifacts["snapshot_path"]); localPath != "" {
			snapshot, loadErr := artifactsync.LoadSnapshot(filepath.Dir(localPath))
			if loadErr != nil {
				return fmt.Errorf("load artifact snapshot for remote controller: %w", loadErr)
			}
			remoteRoot := path.Join(sessionDir, "artifacts")
			remotePath := path.Join(remoteRoot, artifactsync.SnapshotFile)
			artifacts["snapshot_path"] = remotePath
			triggerDir := filepath.Join(filepath.Dir(localPath), artifactsync.ArtifactTriggerDir)
			if selectedArtifacts != nil {
				for i := range snapshot.Artifacts {
					artifact := &snapshot.Artifacts[i]
					if artifact.Kind != "service" || !selectedArtifacts[artifact.Name] {
						continue
					}
					pending, marshalErr := artifactsync.ArtifactTriggerData(artifact.Name, "pending")
					if marshalErr != nil {
						return fmt.Errorf("marshal remote artifact trigger %s: %w", artifact.Name, marshalErr)
					}
					remoteTriggerPath := path.Join(remoteRoot, artifactsync.ArtifactTriggerDir, artifact.Name+".json")
					changed, writeErr := writeRemoteFileChanged(ctx, selected, remoteTriggerPath, pending, 0o600)
					if writeErr != nil {
						return fmt.Errorf("publish pending remote artifact trigger %s: %w", artifact.Name, writeErr)
					}
					if changed && changedArtifacts != nil {
						changedArtifacts[artifact.Name] = true
					}
				}
			}
			for i := range snapshot.Artifacts {
				artifact := &snapshot.Artifacts[i]
				if artifact.Kind != "crd" || (selectedArtifacts != nil && !selectedArtifacts[artifact.Name]) {
					continue
				}
				relative, relErr := filepath.Rel(filepath.Dir(localPath), artifact.LocalPath)
				if relErr != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
					return fmt.Errorf("artifact CRD %s path escapes its snapshot directory", artifact.Name)
				}
				if selectedArtifacts == nil {
					if err := syncDirectoryToRemote(
						ctx,
						selected,
						artifact.LocalPath,
						path.Join(remoteRoot, filepath.ToSlash(relative)),
						true,
					); err != nil {
						return fmt.Errorf("sync artifact CRD %s to remote controller: %w", artifact.Name, err)
					}
					continue
				}
				changed, syncErr := syncDirectoryToRemoteChanged(
					ctx,
					selected,
					artifact.LocalPath,
					path.Join(remoteRoot, filepath.ToSlash(relative)),
					true,
				)
				if syncErr != nil {
					return fmt.Errorf("sync artifact CRD %s to remote controller: %w", artifact.Name, syncErr)
				}
				if changed {
					changedArtifacts[artifact.Name] = true
				}
			}
			if changedArtifacts == nil {
				if err := copyFileToRemote(ctx, selected, localPath, remotePath); err != nil {
					return fmt.Errorf("copy artifact snapshot to remote controller: %w", err)
				}
			} else {
				var copyErr error
				_, copyErr = copyFileToRemoteChanged(ctx, selected, localPath, remotePath)
				if copyErr != nil {
					return fmt.Errorf("copy artifact snapshot to remote controller: %w", copyErr)
				}
			}
			if selectedArtifacts == nil {
				if stat, statErr := os.Stat(triggerDir); statErr == nil && stat.IsDir() {
					if err := syncDirectoryToRemote(
						ctx,
						selected,
						triggerDir,
						path.Join(remoteRoot, artifactsync.ArtifactTriggerDir),
						true,
					); err != nil {
						return fmt.Errorf("sync artifact triggers to remote controller: %w", err)
					}
				}
			} else {
				for i := range snapshot.Artifacts {
					artifact := &snapshot.Artifacts[i]
					if artifact.Kind != "service" || !selectedArtifacts[artifact.Name] {
						continue
					}
					localTriggerPath := filepath.Join(triggerDir, artifact.Name+".json")
					remoteTriggerPath := path.Join(remoteRoot, artifactsync.ArtifactTriggerDir, artifact.Name+".json")
					changed := false
					var copyErr error
					if changedArtifacts == nil {
						copyErr = copyFileToRemote(ctx, selected, localTriggerPath, remoteTriggerPath)
					} else {
						changed, copyErr = copyFileToRemoteChanged(ctx, selected, localTriggerPath, remoteTriggerPath)
					}
					if copyErr != nil {
						return fmt.Errorf("copy artifact trigger %s to remote controller: %w", artifact.Name, copyErr)
					}
					if changed {
						changedArtifacts[artifact.Name] = true
					}
				}
			}
		}
	}
	if hostContext, ok := raw["host"].(map[string]any); ok {
		hostContext["controller"] = ControllerRemote
		hostContext["state_dir"] = selected.Remote.RemoteStateDir
		hostContext["kubeconfig_path"] = materialized.KubeconfigPath
		if remote, ok := hostContext["remote"].(map[string]any); ok {
			remote["ssh_private_key_path"] = ""
		}
		if dedicated, ok := hostContext["dedicated_api"].(map[string]any); ok && selected.DedicatedAPI != nil {
			localDedicatedDir := filepath.Dir(selected.DedicatedAPI.AdminKubeconfigPath)
			remoteDedicatedDir := path.Dir(selected.DedicatedAPI.RemoteAdminKubeconfigPath)
			if err := syncDirectoryToRemote(
				ctx,
				selected,
				localDedicatedDir,
				remoteDedicatedDir,
				false,
			); err != nil {
				return fmt.Errorf("sync Dedicated API identities to remote controller: %w", err)
			}
			dedicated["admin_kubeconfig_path"] = selected.DedicatedAPI.RemoteAdminKubeconfigPath
			dedicated["internal_kubeconfig_path"] = selected.DedicatedAPI.RemoteAdminKubeconfigPath
			for _, item := range []struct {
				key       string
				localPath string
				name      string
			}{
				{"host_manifest_path", selected.DedicatedAPI.HostManifestPath, "dedicated-api-host.yaml"},
				{"bootstrap_manifest_path", selected.DedicatedAPI.BootstrapManifestPath, "dedicated-api-bootstrap.yaml"},
			} {
				if strings.TrimSpace(item.localPath) == "" {
					continue
				}
				remotePath := path.Join(sessionDir, item.name)
				if err := copyFileToRemote(ctx, selected, item.localPath, remotePath); err != nil {
					return fmt.Errorf("copy %s to remote controller: %w", item.key, err)
				}
				dedicated[item.key] = remotePath
			}
		}
	}
	remoteContext, err := json.MarshalIndent(raw, "", "  ")
	if err != nil {
		return fmt.Errorf("marshal remote session context: %w", err)
	}
	if err := WriteRemoteFile(ctx, selected, materialized.ContextIndexPath, remoteContext, 0o600); err != nil {
		return fmt.Errorf("write remote session context: %w", err)
	}
	if _, err := host.RunSSH(
		ctx,
		selected,
		nil,
		"mkdir",
		"-p",
		materialized.ReportsDir,
		materialized.EvidenceDir,
		filepath.ToSlash(filepath.Dir(materialized.LogPath)),
	); err != nil {
		return fmt.Errorf("create remote controller report directories: %w", err)
	}
	return nil
}

func copyFileToRemote(
	ctx context.Context,
	selected *host.Session,
	localPath string,
	remotePath string,
) error {
	data, err := os.ReadFile(localPath)
	if err != nil {
		return err
	}
	return WriteRemoteFile(ctx, selected, remotePath, data, 0o600)
}

func readRemoteRefreshTransaction(
	ctx context.Context,
	selected *host.Session,
	remotePath string,
) ([]byte, bool, error) {
	return readRemoteArtifactFile(ctx, selected, remotePath, "artifact refresh transaction")
}

func readRemoteArtifactFile(
	ctx context.Context,
	selected *host.Session,
	remotePath string,
	label string,
) ([]byte, bool, error) {
	script := fmt.Sprintf(
		"set -eu; if [ -f %s ]; then printf 'present\\n'; cat -- %s; else printf 'missing\\n'; fi",
		host.ShellEscape(remotePath),
		host.ShellEscape(remotePath),
	)
	output, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", script)
	if err != nil {
		return nil, false, fmt.Errorf("read remote %s: %w", label, err)
	}
	state, data, ok := strings.Cut(string(output), "\n")
	switch strings.TrimSpace(state) {
	case "missing":
		if ok && strings.TrimSpace(data) != "" {
			return nil, false, fmt.Errorf("read remote %s: unexpected missing file output", label)
		}
		return nil, false, nil
	case "present":
		if !ok || strings.TrimSpace(data) == "" {
			return nil, true, fmt.Errorf("read remote %s: file is empty", label)
		}
		return []byte(data), true, nil
	default:
		return nil, false, fmt.Errorf("read remote %s: unexpected state %q", label, strings.TrimSpace(state))
	}
}

func removeRemoteFile(ctx context.Context, selected *host.Session, remotePath string) error {
	script := fmt.Sprintf(
		"set -eu; if [ -e %s ]; then rm -f -- %s; fi",
		host.ShellEscape(remotePath),
		host.ShellEscape(remotePath),
	)
	_, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", script)
	return err
}

func copyFileToRemoteChanged(
	ctx context.Context,
	selected *host.Session,
	localPath string,
	remotePath string,
) (bool, error) {
	data, err := os.ReadFile(localPath)
	if err != nil {
		return false, err
	}
	return writeRemoteFileChanged(ctx, selected, remotePath, data, 0o600)
}

// WriteRemoteFile atomically materializes one controller-owned file remotely.
func WriteRemoteFile(
	ctx context.Context,
	selected *host.Session,
	remotePath string,
	data []byte,
	mode os.FileMode,
) error {
	_, err := writeRemoteFileChanged(ctx, selected, remotePath, data, mode)
	return err
}

func writeRemoteFileChanged(
	ctx context.Context,
	selected *host.Session,
	remotePath string,
	data []byte,
	mode os.FileMode,
) (bool, error) {
	script := fmt.Sprintf(
		"set -eu; mkdir -p %s; umask 077; temporary=$(mktemp %s); trap 'rm -f -- \"$temporary\"' EXIT; cat > \"$temporary\"; chmod %04o \"$temporary\"; if [ -f %s ] && cmp -s \"$temporary\" %s; then rm -f -- \"$temporary\"; printf 'unchanged\\n'; else mv -f -- \"$temporary\" %s; printf 'changed\\n'; fi; trap - EXIT",
		host.ShellEscape(path.Dir(remotePath)),
		host.ShellEscape(path.Join(path.Dir(remotePath), "."+path.Base(remotePath)+".XXXXXX")),
		mode.Perm(),
		host.ShellEscape(remotePath),
		host.ShellEscape(remotePath),
		host.ShellEscape(remotePath),
	)
	output, err := host.RunSSH(ctx, selected, data, "sh", "-lc", script)
	if err != nil {
		return false, err
	}
	return parseRemoteChangeState(remotePath, output)
}

func parseRemoteChangeState(remotePath string, output []byte) (bool, error) {
	switch strings.TrimSpace(string(output)) {
	case "changed":
		return true, nil
	case "unchanged":
		return false, nil
	default:
		return false, fmt.Errorf("remote file %s returned unexpected change state %q", remotePath, strings.TrimSpace(string(output)))
	}
}

func syncDirectoryToRemote(
	ctx context.Context,
	selected *host.Session,
	localDir string,
	remoteDir string,
	replace bool,
) error {
	_, err := syncDirectoryToRemoteWithFilter(ctx, selected, localDir, remoteDir, replace, true, false)
	return err
}

func syncDirectoryToRemoteChanged(
	ctx context.Context,
	selected *host.Session,
	localDir string,
	remoteDir string,
	replace bool,
) (bool, error) {
	return syncDirectoryToRemoteWithFilter(ctx, selected, localDir, remoteDir, replace, true, true)
}

func syncGitDirectoryToRemote(
	ctx context.Context,
	selected *host.Session,
	localDir string,
	remoteDir string,
) (bool, error) {
	return syncDirectoryToRemoteWithFilter(ctx, selected, localDir, remoteDir, true, false, true)
}

func syncDirectoryToRemoteWithFilter(
	ctx context.Context,
	selected *host.Session,
	localDir string,
	remoteDir string,
	replace bool,
	applyExcludes bool,
	trackChanges bool,
) (bool, error) {
	remoteDirectorySyncMu.Lock()
	defer remoteDirectorySyncMu.Unlock()
	localDir, err := filepath.Abs(localDir)
	if err != nil {
		return false, err
	}
	if stat, err := os.Stat(localDir); err != nil {
		return false, err
	} else if !stat.IsDir() {
		return false, fmt.Errorf("%s is not a directory", localDir)
	}
	stagingDir := remoteDir + ".cp9-incoming"
	if selected != nil && selected.Remote != nil && selected.Remote.SSHTransport == host.SSHTransportGCloudIAP {
		if err := syncDirectoryToGCloudIAP(ctx, selected, localDir, stagingDir, applyExcludes); err != nil {
			return false, err
		}
	} else {
		baseArgs, err := host.SSHBaseArgs(selected)
		if err != nil {
			return false, err
		}
		if len(baseArgs) == 0 {
			return false, fmt.Errorf("sync remote directory: empty ssh arguments")
		}
		destination := baseArgs[len(baseArgs)-1]
		remoteShell := "ssh " + host.ShellEscapeArgs(baseArgs[:len(baseArgs)-1]...)
		remoteRsync := "mkdir -p " + host.ShellEscape(path.Dir(stagingDir)) + " && rsync"
		localSource := localDir
		if !strings.HasSuffix(localSource, string(filepath.Separator)) {
			localSource += string(filepath.Separator)
		}
		rsyncArgs := []string{
			"--archive",
			"--compress",
			"--delete",
			"--partial-dir=.cp9-partial",
			"--timeout=30",
			"--rsh", remoteShell,
			"--rsync-path", remoteRsync,
		}
		if applyExcludes {
			rsyncArgs = append(rsyncArgs, remoteSyncExcludes()...)
		}
		rsyncArgs = append(rsyncArgs, localSource, destination+":"+stagingDir+"/")
		var (
			rsyncOutput []byte
			rsyncErr    error
		)
		for attempt := 1; attempt <= 3; attempt++ {
			rsyncOutput, rsyncErr = verbose.RunCommand(exec.CommandContext(ctx, "rsync", rsyncArgs...))
			if rsyncErr == nil {
				break
			}
			if err := ctx.Err(); err != nil {
				return false, fmt.Errorf("sync remote directory: %w", err)
			}
			if attempt < 3 {
				verbose.Logf("remote directory sync failed; retrying resumable transfer attempt=%d/3: %v", attempt+1, rsyncErr)
				timer := time.NewTimer(time.Duration(attempt) * time.Second)
				select {
				case <-ctx.Done():
					timer.Stop()
					return false, fmt.Errorf("sync remote directory: %w", ctx.Err())
				case <-timer.C:
				}
			}
		}
		if rsyncErr != nil {
			return false, fmt.Errorf(
				"sync remote directory after 3 attempts: %w\n%s",
				rsyncErr,
				strings.TrimSpace(string(rsyncOutput)),
			)
		}
	}
	changed := false
	if trackChanges {
		changed, err = remoteDirectoryChanged(ctx, selected, stagingDir, remoteDir)
		if err != nil {
			return false, fmt.Errorf("compare staged remote source: %w", err)
		}
		if !changed {
			cleanupScript := fmt.Sprintf("set -eu; rm -rf %s", host.ShellEscape(stagingDir))
			if _, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", cleanupScript); err != nil {
				return false, fmt.Errorf("discard unchanged staged remote source: %w", err)
			}
			return false, nil
		}
	}
	var publishScript string
	if replace {
		previousDir := remoteDir + ".cp9-previous"
		publishScript = fmt.Sprintf(
			"set -eu; test -d %s; rm -rf %s/.cp9-partial; find %s -type f -name '._*' -delete; rm -rf %s; if [ -e %s ]; then mv %s %s; fi; if mv %s %s; then rm -rf %s; else status=$?; if [ -e %s ] && [ ! -e %s ]; then mv %s %s; fi; exit \"$status\"; fi",
			host.ShellEscape(stagingDir),
			host.ShellEscape(stagingDir),
			host.ShellEscape(stagingDir),
			host.ShellEscape(previousDir),
			host.ShellEscape(remoteDir),
			host.ShellEscape(remoteDir),
			host.ShellEscape(previousDir),
			host.ShellEscape(stagingDir),
			host.ShellEscape(remoteDir),
			host.ShellEscape(previousDir),
			host.ShellEscape(previousDir),
			host.ShellEscape(remoteDir),
			host.ShellEscape(previousDir),
			host.ShellEscape(remoteDir),
		)
	} else {
		publishScript = fmt.Sprintf(
			"set -eu; test -d %s; rm -rf %s/.cp9-partial; find %s -type f -name '._*' -delete; mkdir -p %s; rsync -a --checksum %s/ %s/; find %s -type f -name '._*' -delete; rm -rf %s",
			host.ShellEscape(stagingDir),
			host.ShellEscape(stagingDir),
			host.ShellEscape(stagingDir),
			host.ShellEscape(remoteDir),
			host.ShellEscape(stagingDir),
			host.ShellEscape(remoteDir),
			host.ShellEscape(remoteDir),
			host.ShellEscape(stagingDir),
		)
	}
	if _, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", publishScript); err != nil {
		return false, fmt.Errorf("publish remote directory: %w", err)
	}
	return changed, nil
}

func syncDirectoryToGCloudIAP(
	ctx context.Context,
	selected *host.Session,
	localDir string,
	stagingDir string,
	applyExcludes bool,
) error {
	archive, err := os.CreateTemp("", "cp9-gce-runtime-*.tgz")
	if err != nil {
		return fmt.Errorf("create GCE runtime archive: %w", err)
	}
	archivePath := archive.Name()
	if err := archive.Close(); err != nil {
		_ = os.Remove(archivePath)
		return fmt.Errorf("close GCE runtime archive: %w", err)
	}
	defer os.Remove(archivePath)

	args := []string{"-C", localDir, "-czf", archivePath}
	if applyExcludes {
		for _, value := range remoteSyncExcludeNames() {
			args = append(args, "--exclude="+value)
		}
	}
	args = append(args, ".")
	if output, err := verbose.RunCommand(exec.CommandContext(ctx, "tar", args...)); err != nil {
		return fmt.Errorf("archive GCE runtime bundle: %w\n%s", err, strings.TrimSpace(string(output)))
	}

	remoteArchive := stagingDir + ".tgz"
	if _, err := host.RunSSH(ctx, selected, nil, "mkdir", "-p", path.Dir(stagingDir)); err != nil {
		return fmt.Errorf("prepare GCE runtime upload: %w", err)
	}
	if err := host.CopyFileToGCloudIAP(ctx, selected, archivePath, remoteArchive); err != nil {
		return fmt.Errorf("upload GCE runtime bundle: %w", err)
	}
	script := fmt.Sprintf(
		"set -eu; trap 'rm -f -- %s' EXIT; rm -rf %s; mkdir -p %s; tar -xzf %s -C %s; rm -f -- %s; trap - EXIT",
		host.ShellEscape(remoteArchive),
		host.ShellEscape(stagingDir),
		host.ShellEscape(stagingDir),
		host.ShellEscape(remoteArchive),
		host.ShellEscape(stagingDir),
		host.ShellEscape(remoteArchive),
	)
	if _, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", script); err != nil {
		return fmt.Errorf("extract GCE runtime bundle: %w", err)
	}
	return nil
}

func remoteSyncExcludes() []string {
	values := remoteSyncExcludeNames()
	args := make([]string, 0, len(values))
	for _, value := range values {
		args = append(args, "--exclude="+value)
	}
	return args
}

func remoteSyncExcludeNames() []string {
	return []string{
		".git",
		".cp9",
		".terraform",
		"node_modules",
		".cp9-build",
		"cp9.yaml",
		"./bin/cp9",
		"bin/cp9",
	}
}

func remoteDirectoryChanged(
	ctx context.Context,
	selected *host.Session,
	stagingDir string,
	remoteDir string,
) (bool, error) {
	script := fmt.Sprintf(
		"set -eu; if [ ! -d %s ]; then printf '__cp9_missing__\\n'; exit 0; fi; rsync --archive --delete --checksum --dry-run --itemize-changes --omit-dir-times --exclude=.cp9-partial --exclude='._*' --out-format='%%i %%n%%L' %s/ %s/",
		host.ShellEscape(remoteDir),
		host.ShellEscape(stagingDir),
		host.ShellEscape(remoteDir),
	)
	output, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", script)
	if err != nil {
		return false, err
	}
	return remoteDirectoryHasChanges(string(output)), nil
}

func remoteDirectoryHasChanges(output string) bool {
	for line := range strings.SplitSeq(output, "\n") {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		if line == "__cp9_missing__" {
			return true
		}
		code, name, ok := strings.Cut(line, " ")
		if !ok {
			return true
		}
		name = strings.TrimSpace(name)
		if code == "*deleting" {
			if strings.HasSuffix(name, "/") {
				continue
			}
			return true
		}
		if name != "" && strings.HasSuffix(name, "/") && len(code) >= 2 &&
			strings.ContainsRune(".<>ch", rune(code[0])) && code[1] == 'd' {
			continue
		}
		return true
	}
	return false
}

// SyncRemoteArtifacts mirrors one bounded remote artifact directory locally.
func SyncRemoteArtifacts(
	ctx context.Context,
	selected *host.Session,
	remoteDir string,
	localDir string,
	preserve []string,
) error {
	parent := filepath.Dir(localDir)
	if err := os.MkdirAll(parent, 0o755); err != nil {
		return fmt.Errorf("create local report parent: %w", err)
	}
	temporary, err := os.MkdirTemp(parent, "."+filepath.Base(localDir)+".sync-*")
	if err != nil {
		return fmt.Errorf("create temporary local report directory: %w", err)
	}
	defer os.RemoveAll(temporary)
	preserved := map[string][]byte{}
	for _, name := range preserve {
		if name == "" || filepath.Base(name) != name {
			return fmt.Errorf("invalid preserved local report name %q", name)
		}
		data, readErr := os.ReadFile(filepath.Join(localDir, name))
		if readErr == nil {
			preserved[name] = data
		} else if !errors.Is(readErr, os.ErrNotExist) {
			return fmt.Errorf("read preserved local report %s: %w", name, readErr)
		}
	}
	remoteScript := fmt.Sprintf(
		"set -eu; if [ -d %s ]; then tar -C %s -cf - .; else tar -cf - --files-from /dev/null; fi",
		host.ShellEscape(remoteDir),
		host.ShellEscape(remoteDir),
	)
	if selected != nil && selected.Remote != nil && selected.Remote.SSHTransport == host.SSHTransportGCloudIAP {
		// IAP exposes no raw SSH argv, so capture the same one-command archive
		// through the existing gcloud transport.
		archive, err := host.RunSSH(ctx, selected, nil, "sh", "-lc", remoteScript)
		if err != nil {
			return fmt.Errorf("archive remote reports: %w", err)
		}
		tarCmd := exec.CommandContext(ctx, "tar", "-C", temporary, "-xf", "-")
		tarCmd.Stdin = bytes.NewReader(archive)
		if output, err := verbose.RunCommand(tarCmd); err != nil {
			return fmt.Errorf("extract remote reports: %w\n%s", err, strings.TrimSpace(string(output)))
		}
	} else {
		baseArgs, err := host.SSHTransferBaseArgs(selected)
		if err != nil {
			return err
		}
		sshArgs := append(append([]string(nil), baseArgs...), host.SSHRemoteShellCommand(remoteScript))
		sshCmd := exec.CommandContext(ctx, "ssh", sshArgs...)
		tarCmd := exec.CommandContext(ctx, "tar", "-C", temporary, "-xf", "-")
		reader, writer := io.Pipe()
		sshCmd.Stdout = writer
		tarCmd.Stdin = reader
		var sshOutput bytes.Buffer
		var tarStderr bytes.Buffer
		sshCmd.Stderr = &sshOutput
		tarCmd.Stderr = &tarStderr
		if err := verbose.StartCommand(tarCmd); err != nil {
			_ = reader.Close()
			_ = writer.Close()
			return fmt.Errorf("start local report extractor: %w", err)
		}
		if err := verbose.StartCommand(sshCmd); err != nil {
			_ = writer.CloseWithError(err)
			_ = reader.Close()
			_ = tarCmd.Process.Kill()
			_ = tarCmd.Wait()
			return fmt.Errorf("start remote report archive: %w", err)
		}
		sshErr := sshCmd.Wait()
		_ = writer.CloseWithError(sshErr)
		tarErr := tarCmd.Wait()
		_ = reader.Close()
		if sshErr != nil {
			return fmt.Errorf("archive remote reports: %w\n%s", sshErr, strings.TrimSpace(sshOutput.String()))
		}
		if tarErr != nil {
			return fmt.Errorf("extract remote reports: %w\n%s", tarErr, strings.TrimSpace(tarStderr.String()))
		}
	}
	for name, data := range preserved {
		if err := os.WriteFile(filepath.Join(temporary, name), data, 0o600); err != nil {
			return fmt.Errorf("restore preserved local report %s: %w", name, err)
		}
	}
	if err := os.RemoveAll(localDir); err != nil {
		return fmt.Errorf("replace cached local reports: %w", err)
	}
	if err := os.Rename(temporary, localDir); err != nil {
		return fmt.Errorf("publish cached local reports: %w", err)
	}
	return nil
}

// SelectedSourceBuildServices returns the source-build selection owned by a Session Tiltfile.
func SelectedSourceBuildServices(localTiltfile string) ([]string, error) {
	data, err := os.ReadFile(localTiltfile)
	if err != nil {
		return nil, fmt.Errorf("read Tiltfile source-build selection: %w", err)
	}
	return parseTiltStringList(string(data), "source_build_services")
}

// ResolveSourceDirectory returns the canonical local directory for a
// source-backed repo path selected by a Session Tiltfile.
func ResolveSourceDirectory(localTiltfile, relative string) (string, error) {
	tiltDir := filepath.Dir(localTiltfile)
	if source, ok := resolveTiltSourceDirectory(tiltDir, relative); ok {
		return source, nil
	}
	repositoryRoot, ok := findRepositoryRoot(localTiltfile)
	if ok && filepath.Base(filepath.Dir(repositoryRoot)) == "worktrees" {
		if source, ok := resolveTiltSourceDirectory(filepath.Dir(tiltDir), relative); ok {
			return source, nil
		}
	}
	return "", fmt.Errorf("source directory %s is not a readable canonical path", relative)
}

func syncSelectedSourceRepositories(
	ctx context.Context,
	selected *host.Session,
	localTiltfile string,
	remoteTiltfile string,
	localContextIndex string,
) (map[string]bool, error) {
	changedServices := map[string]bool{}
	contextData, err := os.ReadFile(localContextIndex)
	if err != nil {
		return nil, fmt.Errorf("read runtime identity from session context: %w", err)
	}
	var sessionContext struct {
		SourceBuildSupported *bool  `json:"source_build_supported"`
		GitHubAppClientID    string `json:"github_app_client_id"`
	}
	if err := json.Unmarshal(contextData, &sessionContext); err != nil {
		return nil, fmt.Errorf("parse runtime identity from session context: %w", err)
	}
	sourceBuildSupported := true
	if sessionContext.SourceBuildSupported != nil {
		sourceBuildSupported = *sessionContext.SourceBuildSupported
	}
	if !sourceBuildSupported {
		return changedServices, nil
	}
	services, err := SelectedSourceBuildServices(localTiltfile)
	if err != nil {
		return nil, err
	}
	if len(services) == 0 {
		return changedServices, nil
	}
	clientID := strings.TrimSpace(sessionContext.GitHubAppClientID)
	if clientID == "" {
		return nil, fmt.Errorf("session context github_app_client_id is required for selected source builds")
	}
	repositoryRoot, ok := findRepositoryRoot(localTiltfile)
	if !ok {
		return nil, fmt.Errorf("selected source-build services require a CP9 repository checkout")
	}
	registryPath := filepath.Join(repositoryRoot, "internal", "sourcebacked", "contracts.json")
	registry, err := os.ReadFile(registryPath)
	if err != nil {
		return nil, fmt.Errorf("read source-backed contract registry: %w", err)
	}
	var document sourceContractDocument
	if err := json.Unmarshal(registry, &document); err != nil {
		return nil, fmt.Errorf("parse source-backed contract registry: %w", err)
	}
	contracts := make(map[string]sourceContract, len(document.Contracts))
	for _, contract := range document.Contracts {
		contracts[contract.Name] = contract
	}

	type repositorySync struct {
		localRoot  string
		remoteRoot string
		services   []string
	}
	remoteTiltDir := path.Dir(remoteTiltfile)
	repositories := map[string]repositorySync{}
	remoteModuleRoots := map[string]bool{}
	for _, service := range services {
		contract, ok := contracts[service]
		if !ok || strings.TrimSpace(contract.Source.Repo) == "" ||
			strings.TrimSpace(contract.Source.Main) == "" {
			return nil, fmt.Errorf("selected source-build service %s has no complete source contract", service)
		}
		localSource, err := ResolveSourceDirectory(localTiltfile, contract.Source.Repo)
		if err != nil {
			return nil, fmt.Errorf(
				"selected source-build service %s has no readable local source directory %s: %w",
				service,
				contract.Source.Repo,
				err,
			)
		}
		localRoot, err := gitRepositoryRoot(ctx, localSource)
		if err != nil {
			return nil, fmt.Errorf("resolve selected source service %s git root: %w", service, err)
		}
		relativeService, err := filepath.Rel(localRoot, localSource)
		if err != nil || relativeService == ".." ||
			strings.HasPrefix(relativeService, ".."+string(filepath.Separator)) {
			return nil, fmt.Errorf("selected source service %s escapes git root %s", service, localRoot)
		}
		remoteRoot := remoteRepositoryRoot(
			path.Clean(path.Join(remoteTiltDir, filepath.ToSlash(contract.Source.Repo))),
			relativeService,
		)
		if existing, ok := repositories[localRoot]; ok {
			if existing.remoteRoot != remoteRoot {
				return nil, fmt.Errorf(
					"source services %s and %s map git root %s to different remote roots %s and %s",
					existing.services[0],
					service,
					localRoot,
					existing.remoteRoot,
					remoteRoot,
				)
			}
			existing.services = append(existing.services, service)
			repositories[localRoot] = existing
		} else {
			repositories[localRoot] = repositorySync{
				localRoot:  localRoot,
				remoteRoot: remoteRoot,
				services:   []string{service},
			}
		}
		moduleRoot, ok := findGoModuleRoot(localSource, contract.Source.Main, localRoot)
		if !ok {
			return nil, fmt.Errorf("selected source-build service %s has no go.mod within %s", service, localRoot)
		}
		relativeModuleRoot, err := filepath.Rel(localRoot, moduleRoot)
		if err != nil || relativeModuleRoot == ".." ||
			strings.HasPrefix(relativeModuleRoot, ".."+string(filepath.Separator)) {
			return nil, fmt.Errorf("selected source service %s module root escapes git root %s", service, localRoot)
		}
		remoteModuleRoots[path.Join(remoteRoot, filepath.ToSlash(relativeModuleRoot))] = true
	}

	manifestTool := filepath.Join(
		repositoryRoot,
		"personas",
		"dev-full",
		"build",
		"remote",
		"source-manifest.py",
	)
	remoteWorkRoot := path.Join(selected.Remote.RemoteStateDir, remoteControllerDirName, "work")
	for _, repository := range repositories {
		stagedRoots, cleanup, err := stageGitRepository(ctx, manifestTool, repository.localRoot)
		if err != nil {
			return nil, fmt.Errorf("stage selected source service %s: %w", repository.services[0], err)
		}
		repositoryChanged := false
		for _, staged := range stagedRoots {
			relativeRoot, relErr := filepath.Rel(repository.localRoot, staged.localRoot)
			if relErr != nil {
				err = relErr
				break
			}
			remoteRoot := path.Clean(path.Join(repository.remoteRoot, filepath.ToSlash(relativeRoot)))
			if remoteRoot != remoteWorkRoot && !strings.HasPrefix(remoteRoot, remoteWorkRoot+"/") {
				err = fmt.Errorf("local replacement root %s maps outside remote source workspace", staged.localRoot)
				break
			}
			var changed bool
			changed, err = syncGitDirectoryToRemote(ctx, selected, staged.stagedRoot, remoteRoot)
			repositoryChanged = repositoryChanged || changed
			if err != nil {
				break
			}
		}
		cleanup()
		if err != nil {
			return nil, fmt.Errorf("sync selected source service %s source roots: %w", repository.services[0], err)
		}
		for _, service := range repository.services {
			changedServices[service] = repositoryChanged
		}
	}
	helperPath := filepath.Join(
		repositoryRoot,
		"personas",
		"dev-full",
		"build",
		"remote",
		"github-user-module-download.sh",
	)
	userTokenHelperPath := filepath.Join(
		repositoryRoot,
		"personas",
		"dev-full",
		"build",
		"remote",
		"github-app-user-token.sh",
	)
	relativeTiltfile, err := filepath.Rel(repositoryRoot, localTiltfile)
	if err != nil {
		return nil, fmt.Errorf("resolve remote CP9 repository root: %w", err)
	}
	remoteCP9Root := remoteRepositoryRoot(remoteTiltfile, relativeTiltfile)
	relativeHelper, err := filepath.Rel(repositoryRoot, helperPath)
	if err != nil {
		return nil, fmt.Errorf("resolve GitHub App module downloader path: %w", err)
	}
	remoteHelperPath := path.Join(remoteCP9Root, filepath.ToSlash(relativeHelper))
	if _, err := host.RunSSH(ctx, selected, nil, "chmod", "700", remoteHelperPath); err != nil {
		return nil, fmt.Errorf("prepare GitHub user module downloader: %w", err)
	}
	// #nosec G702 -- the executable is a repository-owned helper; values are passed as argv without a shell.
	tokenCommand := exec.CommandContext(
		ctx,
		userTokenHelperPath,
		clientID,
	)
	var tokenStderr bytes.Buffer
	tokenCommand.Stderr = io.MultiWriter(os.Stderr, &tokenStderr)
	userToken, err := tokenCommand.Output()
	if err != nil {
		return nil, fmt.Errorf("authorize GitHub App user: %w: %s", err, strings.TrimSpace(tokenStderr.String()))
	}
	userToken = bytes.TrimSpace(userToken)
	if !bytes.HasPrefix(userToken, []byte("ghu_")) {
		return nil, fmt.Errorf("authorize GitHub App user: helper returned an invalid user token")
	}
	defer func() {
		for index := range userToken {
			userToken[index] = 0
		}
	}()
	for moduleRoot := range remoteModuleRoots {
		if _, err := host.RunSSH(
			ctx,
			selected,
			userToken,
			remoteHelperPath,
			moduleRoot,
		); err != nil {
			return nil, fmt.Errorf("download remote Go modules for %s with GitHub App user: %w", moduleRoot, err)
		}
	}
	return changedServices, nil
}

func gitRepositoryRoot(ctx context.Context, source string) (string, error) {
	output, err := verbose.RunCommand(
		exec.CommandContext(ctx, "git", "-C", source, "rev-parse", "--show-toplevel"),
	)
	if err != nil {
		return "", fmt.Errorf("%w\n%s", err, strings.TrimSpace(string(output)))
	}
	root := strings.TrimSpace(string(output))
	if root == "" {
		return "", fmt.Errorf("git returned an empty repository root")
	}
	return filepath.Abs(root)
}

func remoteRepositoryRoot(remoteService, relativeService string) string {
	remoteRoot := remoteService
	if relativeService != "." {
		for range strings.SplitSeq(filepath.ToSlash(relativeService), "/") {
			remoteRoot = path.Dir(remoteRoot)
		}
	}
	return remoteRoot
}

type stagedGitRoot struct {
	localRoot  string
	stagedRoot string
}

func stageGitRepository(
	ctx context.Context,
	manifestTool string,
	repositoryRoot string,
) ([]stagedGitRoot, func(), error) {
	temporary, err := os.MkdirTemp("", "cp9-source-manifest-*")
	if err != nil {
		return nil, nil, err
	}
	cleanup := func() { _ = os.RemoveAll(temporary) }
	archive := filepath.Join(temporary, "source.tar")
	metadata := filepath.Join(temporary, "source.json")
	output, err := verbose.RunCommand(exec.CommandContext(
		ctx,
		"python3",
		manifestTool,
		"--repo",
		repositoryRoot,
		"--metadata",
		metadata,
		"--archive",
		archive,
	))
	if err != nil {
		cleanup()
		return nil, nil, fmt.Errorf("build git-aware source archive: %w\n%s", err, strings.TrimSpace(string(output)))
	}
	staged := filepath.Join(temporary, "source")
	if err := os.Mkdir(staged, 0o700); err != nil {
		cleanup()
		return nil, nil, err
	}
	output, err = verbose.RunCommand(exec.CommandContext(ctx, "tar", "-C", staged, "-xf", archive))
	if err != nil {
		cleanup()
		return nil, nil, fmt.Errorf("extract git-aware source archive: %w\n%s", err, strings.TrimSpace(string(output)))
	}
	var manifest struct {
		SyncRoot string   `json:"sync_root"`
		Roots    []string `json:"roots"`
	}
	data, err := os.ReadFile(metadata)
	if err != nil {
		cleanup()
		return nil, nil, err
	}
	if err := json.Unmarshal(data, &manifest); err != nil {
		cleanup()
		return nil, nil, fmt.Errorf("parse git-aware source metadata: %w", err)
	}
	stagedRoots := make([]stagedGitRoot, 0, len(manifest.Roots))
	for _, root := range manifest.Roots {
		relativeRoot, err := filepath.Rel(manifest.SyncRoot, root)
		if err != nil || relativeRoot == ".." || strings.HasPrefix(relativeRoot, ".."+string(filepath.Separator)) {
			cleanup()
			return nil, nil, fmt.Errorf("source root %s escapes manifest sync root %s", root, manifest.SyncRoot)
		}
		stagedRoots = append(stagedRoots, stagedGitRoot{
			localRoot:  root,
			stagedRoot: filepath.Join(staged, relativeRoot),
		})
	}
	return stagedRoots, cleanup, nil
}

func findGoModuleRoot(sourceDir, mainPath, repositoryRoot string) (string, bool) {
	current := filepath.Dir(filepath.Join(sourceDir, filepath.FromSlash(mainPath)))
	for {
		if _, err := os.Stat(filepath.Join(current, "go.mod")); err == nil {
			return current, true
		}
		if current == repositoryRoot {
			return "", false
		}
		parent := filepath.Dir(current)
		if parent == current {
			return "", false
		}
		current = parent
	}
}

func parseTiltStringList(document, assignment string) ([]string, error) {
	lines := strings.Split(document, "\n")
	inList := false
	var values []string
	for _, raw := range lines {
		line := strings.TrimSpace(raw)
		if !inList {
			name, after, ok := strings.Cut(line, "=")
			if !ok || strings.TrimSpace(name) != assignment {
				continue
			}
			line = strings.TrimSpace(after)
			if !strings.HasPrefix(line, "[") {
				return nil, fmt.Errorf("%s must be a string list literal", assignment)
			}
			inList = true
			line = strings.TrimSpace(strings.TrimPrefix(line, "["))
		}
		line = stripTiltComment(line)
		closed := false
		if before, after, ok := strings.Cut(line, "]"); ok {
			if strings.TrimSpace(after) != "" {
				return nil, fmt.Errorf("%s has content after closing bracket", assignment)
			}
			line = before
			closed = true
		}
		for value := range strings.SplitSeq(line, ",") {
			value = strings.TrimSpace(value)
			if value == "" {
				continue
			}
			parsed, ok := parseQuotedTiltValue(value)
			if !ok {
				return nil, fmt.Errorf("%s contains non-string list value %q", assignment, value)
			}
			values = appendUniqueString(values, parsed)
		}
		if closed {
			return values, nil
		}
	}
	if inList {
		return nil, fmt.Errorf("%s has an unterminated list", assignment)
	}
	return nil, nil
}

func stripTiltComment(line string) string {
	var quote byte
	for i := 0; i < len(line); i++ {
		switch line[i] {
		case '\'', '"':
			switch quote {
			case 0:
				quote = line[i]
			case line[i]:
				quote = 0
			}
		case '#':
			if quote == 0 {
				return strings.TrimSpace(line[:i])
			}
		}
	}
	return strings.TrimSpace(line)
}

func parseQuotedTiltValue(value string) (string, bool) {
	value = strings.TrimSpace(value)
	if len(value) < 2 {
		return "", false
	}
	quote := value[0]
	if (quote != '"' && quote != '\'') || value[len(value)-1] != quote {
		return "", false
	}
	return value[1 : len(value)-1], true
}

func resolveTiltSourceDirectory(tiltDir, relative string) (string, bool) {
	relative = strings.TrimSpace(relative)
	if relative == "" || filepath.IsAbs(relative) {
		return "", false
	}
	candidate := filepath.Clean(filepath.Join(tiltDir, filepath.FromSlash(relative)))
	stat, err := os.Stat(candidate)
	if err != nil || !stat.IsDir() {
		return "", false
	}
	canonical, err := filepath.EvalSymlinks(candidate)
	if err != nil {
		return "", false
	}
	absolute, err := filepath.Abs(canonical)
	return absolute, err == nil
}

func appendUniqueString(values []string, value string) []string {
	value = strings.TrimSpace(value)
	if value == "" {
		return values
	}
	if slices.Contains(values, value) {
		return values
	}
	return append(values, value)
}

func findRepositoryRoot(file string) (string, bool) {
	dir := filepath.Dir(file)
	for {
		if _, err := os.Stat(filepath.Join(dir, ".git")); err == nil {
			return dir, true
		}
		parent := filepath.Dir(dir)
		if parent == dir {
			return "", false
		}
		dir = parent
	}
}

func normalizedRemotePort(req *Request) int {
	if req != nil && req.Port > 0 {
		return req.Port
	}
	return DefaultAPIPort
}

func remoteHome(selected *host.Session) string {
	if selected != nil && selected.Remote != nil {
		stateDir := strings.TrimSpace(selected.Remote.RemoteStateDir)
		if before, _, ok := strings.Cut(stateDir, "/.cp9/"); ok && before != "" {
			return before
		}
		if stateDir != "" {
			return path.Dir(path.Dir(stateDir))
		}
		if user := strings.TrimSpace(selected.Remote.SSHUser); user != "" {
			return "/home/" + user
		}
	}
	return "/home/ec2-user"
}

func remoteGoEnvironment() []string {
	return []string{
		"GOPRIVATE=github.com/tidbcloud/*",
		"GONOSUMDB=github.com/tidbcloud/*",
		"GONOPROXY=github.com/tidbcloud/*",
		"GOPROXY=https://proxy.golang.org,direct",
	}
}

func remoteControllerEnvironment(selected *host.Session) []string {
	if selected != nil && selected.Mode == config.HostModeRemoteEC2K3d {
		return append([]string{"CP9_AWS_RUNTIME_IDENTITY=ec2-instance-profile"}, remoteGoEnvironment()...)
	}
	return nil
}

func remoteControllerUnit(session string) string {
	return remoteUnitPrefix + sanitizeRemoteName(session) + ".service"
}

func sanitizeRemoteName(value string) string {
	value = strings.TrimSpace(value)
	sanitized := make([]rune, 0, len(value))
	for _, r := range value {
		switch {
		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_':
			sanitized = append(sanitized, r)
		default:
			sanitized = append(sanitized, '-')
		}
	}
	value = strings.Trim(string(sanitized), "-")
	if value == "" {
		return "session"
	}
	return value
}

func systemdJoin(argv []string) string {
	quoted := make([]string, 0, len(argv))
	for _, value := range argv {
		quoted = append(quoted, systemdEscape(value))
	}
	return strings.Join(quoted, " ")
}

func systemdEscape(value string) string {
	value = strings.ReplaceAll(value, "%", "%%")
	if value == "" {
		return `""`
	}
	if !strings.ContainsAny(value, " \t\n\r\"\\") {
		return value
	}
	value = strings.ReplaceAll(value, `\`, `\\`)
	value = strings.ReplaceAll(value, `"`, `\"`)
	return `"` + value + `"`
}

func stringValue(value any) string {
	text, ok := value.(string)
	if !ok {
		return ""
	}
	return strings.TrimSpace(text)
}

func recordControllerTunnel(stateDir string, status *Status, started *StartedProcess) {
	if status == nil || started == nil {
		return
	}
	status.ControllerTunnelPID = started.PID
	status.ControllerTunnelCommand = append([]string(nil), started.Command...)
	status.ControllerTunnelLogPath = SessionRelativePath(stateDir, started.LogPath)
	status.ControllerTunnelChildPID = SessionRelativePath(stateDir, started.ChildPIDPath)
	startedAt := started.StartedAt.UTC()
	status.ControllerTunnelStartedAt = &startedAt
}

func clearControllerTunnel(status *Status) {
	if status == nil {
		return
	}
	status.ControllerTunnelPID = 0
	status.ControllerTunnelCommand = nil
	status.ControllerTunnelLogPath = ""
	status.ControllerTunnelChildPID = ""
	status.ControllerTunnelStartedAt = nil
}

func (s *ProcessStopper) stopRecordedControllerTunnelChild(
	ctx context.Context,
	status *Status,
	stateDir string,
) (string, error) {
	if status == nil {
		return "", nil
	}
	return s.stopRecordedTunnelChild(
		ctx,
		stateDir,
		status.ControllerTunnelChildPID,
		"remote Tilt controller forward",
		func(command string) bool {
			return controllerTunnelChildMatches(status, command)
		},
	)
}

func controllerTunnelMatches(status *Status, command string) bool {
	if status == nil {
		return false
	}
	return tunnelMatchesCommand(status.ControllerTunnelCommand, command)
}

func controllerTunnelChildMatches(status *Status, command string) bool {
	if status == nil {
		return false
	}
	return tunnelChildMatchesCommand(status.ControllerTunnelCommand, command)
}