package cmd

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"regexp"
	"slices"
	"strconv"
	"strings"
	"syscall"
	"time"

	"github.com/innerr/ticat/pkg/core/model"
	"gopkg.in/yaml.v3"

	"github.com/tidbcloud/controlplane9/internal/cluster"
	"github.com/tidbcloud/controlplane9/internal/config"
	cp9errors "github.com/tidbcloud/controlplane9/internal/errors"
	"github.com/tidbcloud/controlplane9/internal/host"
	"github.com/tidbcloud/controlplane9/internal/lifecycle"
	"github.com/tidbcloud/controlplane9/internal/persona"
	cpterraform "github.com/tidbcloud/controlplane9/internal/terraform"
	cptilt "github.com/tidbcloud/controlplane9/internal/tilt"
)

var (
	loadSelectedHost        = host.LoadOrResolve
	loadDay2Artifacts       = lifecycle.LoadArtifacts
	loadSelectedSessionHost = host.Load
	loadLifecycleState      = cluster.LoadState
	loadTiltStatus          = cptilt.LoadStatusForState
	readTiltReadiness       = cptilt.ResourceReadinessForHost
	runTiltTrigger          = func(ctx context.Context, status *cptilt.Status, resource string) error {
		return cptilt.NewCLI().Trigger(ctx, status.APIHost, status.APIPort, resource)
	}
	runTiltLogs = func(ctx context.Context, status *cptilt.Status, resources []string, query logsQuery) ([]byte, error) {
		return cptilt.NewCLI().Logs(ctx, &cptilt.LogsRequest{
			Host:      status.APIHost,
			Port:      status.APIPort,
			Resources: resources,
			Since:     query.Since,
			Level:     query.Level,
			Tail:      query.Limit,
		})
	}
	runDedicatedDiagnosticsKubectl = kubectlOutput
	day2ProcessRunning             = func(pid int) bool {
		if pid <= 0 {
			return false
		}
		process, err := os.FindProcess(pid)
		if err != nil {
			return false
		}
		return process.Signal(syscall.Signal(0)) == nil
	}
	runBuildRealStep = func(ctx context.Context, step *lifecycle.BuildRealStep) error {
		return step.Run(ctx)
	}
	runGenStubsStep = func(ctx context.Context, step *lifecycle.GenStubsStep) error {
		return step.Run(ctx)
	}
	runConfigGenStep = func(ctx context.Context, step *lifecycle.ConfigGenStep) error {
		return step.Run(ctx)
	}
	runDeployStep = func(ctx context.Context, step *lifecycle.DeployStep) error {
		return step.Run(ctx)
	}
	runReadinessStep = func(ctx context.Context, step *lifecycle.ReadinessStep) error {
		return step.Run(ctx)
	}
)

type deployResult struct {
	Service        string `json:"service"`
	Runtime        string `json:"runtime,omitempty"`
	Role           string `json:"role"`
	Resource       string `json:"resource,omitempty"`
	ImageRef       string `json:"image_ref,omitempty"`
	Namespace      string `json:"namespace,omitempty"`
	WorkloadKind   string `json:"workload_kind,omitempty"`
	WorkloadName   string `json:"workload_name,omitempty"`
	OverlayDir     string `json:"overlay_dir,omitempty"`
	DryRun         bool   `json:"dry_run,omitempty"`
	TriggerCommand string `json:"trigger_command,omitempty"`
}

func (r *deployResult) FormatText() string {
	runtime := ""
	if r.Runtime != "" {
		runtime = " on runtime " + r.Runtime
	}
	if r.DryRun {
		target := strings.TrimSpace(r.Resource)
		if target == "" {
			target = strings.TrimSpace(r.WorkloadKind + "/" + r.WorkloadName)
		}
		if target == "" {
			target = "selected service"
		}
		if strings.TrimSpace(r.TriggerCommand) == "" {
			return fmt.Sprintf("dry run: would deploy %s (%s) in %s%s", target, r.Service, r.Namespace, runtime)
		}
		return fmt.Sprintf("dry run: would trigger %s (%s) in %s%s with `%s`", target, r.Service, r.Namespace, runtime, r.TriggerCommand)
	}
	if strings.TrimSpace(r.Resource) != "" {
		return fmt.Sprintf("triggered %s (%s) via Tilt resource %s in %s%s", r.Service, r.Role, r.Resource, r.Namespace, runtime)
	}
	target := r.WorkloadKind + "/" + r.WorkloadName
	if r.WorkloadKind == "" || r.WorkloadName == "" {
		target = "overlay applied"
	}
	return fmt.Sprintf("deployed %s (%s) to %s%s as %s", r.Service, r.Role, r.Namespace, runtime, target)
}

type logEntry struct {
	Cluster   string `json:"cluster"`
	Substrate string `json:"substrate,omitempty"`
	Context   string `json:"context,omitempty"`
	Namespace string `json:"namespace,omitempty"`
	Service   string `json:"service"`
	Pod       string `json:"pod"`
	Container string `json:"container,omitempty"`
	Stream    string `json:"stream,omitempty"`
	Line      string `json:"line"`
}

type logSourceAttempt struct {
	Source    string `json:"source"`
	Substrate string `json:"substrate,omitempty"`
	Context   string `json:"context,omitempty"`
	Namespace string `json:"namespace,omitempty"`
	Workload  string `json:"workload,omitempty"`
	Status    string `json:"status"`
	Message   string `json:"message,omitempty"`
}

type logsResult struct {
	Status      string             `json:"status"`
	Sources     []string           `json:"sources"`
	Entries     []logEntry         `json:"entries"`
	Service     string             `json:"service,omitempty"`
	TraceID     string             `json:"trace_id,omitempty"`
	Since       string             `json:"since,omitempty"`
	Level       string             `json:"level,omitempty"`
	Limit       int                `json:"limit,omitempty"`
	EmptyReason string             `json:"empty_reason,omitempty"`
	Attempted   []logSourceAttempt `json:"attempted_sources,omitempty"`
	NextQuery   string             `json:"next_query,omitempty"`
}

func (r *logsResult) FormatText() string {
	var b strings.Builder
	if len(r.Sources) > 0 {
		sourceLine := "Sources: " + strings.Join(r.Sources, ", ")
		if r.Limit > 0 {
			sourceLine += fmt.Sprintf(" limit=%d", r.Limit)
		}
		if r.Since != "" {
			sourceLine += " since=" + r.Since
		}
		if r.Level != "" {
			sourceLine += " level=" + r.Level
		}
		if r.TraceID != "" {
			sourceLine += " trace=" + r.TraceID
		}
		sourceLine += "\n"
		b.WriteString(sourceLine)
	}
	if len(r.Entries) == 0 {
		if len(r.Sources) > 0 || r.Status == "partial" || r.Status == "incomplete" {
			fmt.Fprintf(&b, "Status: %s\n", firstNonEmptyString(r.Status, "empty"))
		}
		b.WriteString(r.emptyMessage())
	} else {
		b.WriteString(strings.Join(r.formatEntryLines(), "\n"))
	}
	if r.Status == "incomplete" || r.Status == "partial" || strings.TrimSpace(r.NextQuery) != "" {
		if b.Len() > 0 {
			b.WriteString("\n")
		}
		if len(r.Attempted) > 0 {
			b.WriteString("Attempted sources:\n")
			for _, attempt := range r.Attempted {
				identity := firstNonEmptyString(attempt.Substrate, attempt.Source)
				if attempt.Namespace != "" {
					identity += "/" + attempt.Namespace
				}
				if attempt.Workload != "" {
					identity += "/" + attempt.Workload
				}
				fmt.Fprintf(&b, "- %s: %s", identity, attempt.Status)
				if strings.TrimSpace(attempt.Message) != "" {
					fmt.Fprintf(&b, " (%s)", strings.TrimSpace(attempt.Message))
				}
				b.WriteString("\n")
			}
		}
		if strings.TrimSpace(r.NextQuery) != "" {
			fmt.Fprintf(&b, "Next query: %s", strings.TrimSpace(r.NextQuery))
		}
	}
	return strings.TrimRight(b.String(), "\n")
}

func (r *logsResult) formatEntryLines() []string {
	var lines []string
	for i := range r.Entries {
		entry := &r.Entries[i]
		source := firstNonEmptyString(entry.Substrate, entry.Cluster)
		identity := strings.Trim(strings.Join([]string{entry.Pod, entry.Container, entry.Stream}, "/"), "/")
		if entry.Namespace != "" {
			identity = entry.Namespace + "/" + identity
		}
		lines = append(lines, fmt.Sprintf("[%s] %s/%s %s", source, entry.Service, identity, entry.Line))
	}
	return lines
}

func (r *logsResult) emptyMessage() string {
	if strings.TrimSpace(r.EmptyReason) != "" {
		return r.EmptyReason
	}
	scope := "all services"
	if r.Service != "" {
		scope = r.Service
	}
	if r.TraceID != "" {
		return fmt.Sprintf("no log lines matched %s for trace %s", scope, r.TraceID)
	}
	return fmt.Sprintf("no log lines matched %s", scope)
}

func deployCmd() model.NormalCmd {
	return func(args model.ArgVals, cc *model.Cli, env *model.Env, _ []model.ParsedCmd) error {
		syncVerbose(env)

		service := strings.TrimSpace(args.GetRaw("service"))
		if service == "" {
			return cp9errors.New(cp9errors.CategoryConfig, "service argument is required")
		}
		dryRun := strings.EqualFold(strings.TrimSpace(args.GetRaw("dry-run")), "true")

		cfgPath := args.GetRaw("config")
		localPath, localProvided := explicitLocalPath(args)
		version, err := config.DetectAPIVersion(cfgPath)
		if err != nil {
			return err
		}
		if version == config.APIVersionV1Alpha2 {
			result, err := runV1Alpha2DeployForRuntime(
				context.Background(), cfgPath, localPath, localProvided, service, strings.TrimSpace(args.GetRaw("runtime")), dryRun,
			)
			if err != nil {
				return wrapDay2CommandError("deploy "+service, err)
			}
			return model.Output(cc, env, result)
		}

		cfg, local, err := config.Load(cfgPath, localPath)
		if err != nil {
			return err
		}
		result, err := runDeploy(context.Background(), cfg, local, service, dryRun)
		if err != nil {
			return wrapDay2CommandError("deploy "+service, err)
		}
		return model.Output(cc, env, result)
	}
}

func logsCmd() model.NormalCmd {
	return func(args model.ArgVals, cc *model.Cli, env *model.Env, _ []model.ParsedCmd) error {
		syncVerbose(env)

		cfgPath := args.GetRaw("config")
		localPath, localProvided := explicitLocalPath(args)
		version, err := config.DetectAPIVersion(cfgPath)
		if err != nil {
			return err
		}
		requestedRuntime := strings.TrimSpace(args.GetRaw("runtime"))
		query := logsQuery{
			Service: strings.TrimSpace(args.GetRaw("service")),
			TraceID: strings.TrimSpace(args.GetRaw("trace-id")),
			Since:   strings.TrimSpace(args.GetRaw("since")),
			Level:   strings.TrimSpace(args.GetRaw("level")),
		}
		query.Limit, err = parseLogsLimit(args.GetRaw("limit"))
		if err != nil {
			return err
		}
		if version == config.APIVersionV1Alpha2 {
			result, err := runV1Alpha2Logs(context.Background(), cfgPath, localPath, localProvided, query, requestedRuntime)
			if result != nil {
				if outputErr := model.Output(cc, env, result); outputErr != nil {
					return outputErr
				}
			}
			if err != nil {
				return wrapDay2CommandError("collect logs", err)
			}
			return nil
		}

		cfg, local, err := config.Load(cfgPath, localPath)
		if err != nil {
			return err
		}
		selected, err := loadSelectedHost(cfg, local)
		if err != nil {
			return wrapDay2CommandError("load selected host", err)
		}
		artifacts, err := loadDay2Artifacts(selected.StateDir)
		if err != nil {
			return wrapDay2CommandError("load artifacts", err)
		}

		result, err := collectLogs(context.Background(), selected, artifacts, query)
		if err != nil {
			return wrapDay2CommandError("collect logs", err)
		}
		return model.Output(cc, env, result)
	}
}

func wrapDay2CommandError(prefix string, err error) error {
	var cp9Err *cp9errors.Error
	if errors.As(err, &cp9Err) {
		return cp9Err
	}
	return cp9errors.Newf(cp9errors.CategoryResource, "%s: %v", prefix, err)
}

type v1alpha2TiltRuntime struct {
	Session         *config.SessionConfig
	StateDir        string
	RuntimeID       string
	RuntimeStateDir string
	MultiRuntime    bool
	State           *cluster.State
	Selected        *host.Session
	Status          *cptilt.Status
}

func loadV1Alpha2TiltRuntime(cfgPath, localPath string, localProvided bool) (*v1alpha2TiltRuntime, error) {
	session, stateDir, err := loadV1Alpha2SessionStateDir(cfgPath, localPath, localProvided)
	if err != nil {
		return nil, err
	}
	state, err := loadLifecycleState(stateDir)
	if err != nil {
		return nil, cp9errors.Newf(cp9errors.CategoryResource, "read state: %v", err)
	}
	selected, err := loadSelectedSessionHost(stateDir)
	if err != nil {
		return nil, cp9errors.Newf(cp9errors.CategoryResource, "load selected host: %v", err)
	}
	status, err := loadTiltStatus(stateDir, state)
	if err != nil {
		return nil, cp9errors.Newf(cp9errors.CategoryResource, "load tilt status: %v", err)
	}
	runtime := &v1alpha2TiltRuntime{
		Session:         session,
		StateDir:        stateDir,
		RuntimeID:       cptilt.RuntimePrimary,
		RuntimeStateDir: stateDir,
		State:           state,
		Selected:        selected,
		Status:          status,
	}
	if err := checkSessionContextStillMatches(cfgPath, session, stateDir, state); err != nil {
		return nil, err
	}
	return runtime, nil
}

func loadV1Alpha2TiltRuntimeForResource(
	cfgPath, localPath string,
	localProvided bool,
	resource, requestedRuntime string,
) (*v1alpha2TiltRuntime, error) {
	runtime, err := loadV1Alpha2TiltRuntime(cfgPath, localPath, localProvided)
	if err != nil {
		return nil, err
	}
	requestedRuntime = strings.TrimSpace(requestedRuntime)
	plan, found, err := loadValidatedRuntimePlan(runtime.StateDir, runtime.Session)
	if err != nil {
		return nil, err
	}
	if !found {
		if requestedRuntime != "" && requestedRuntime != cptilt.RuntimePrimary {
			return nil, cp9errors.Newf(cp9errors.CategoryConfig, "runtime %q is not configured", requestedRuntime)
		}
		return runtime, nil
	}
	membership, err := plan.Resolve(resource, requestedRuntime)
	if err != nil {
		return nil, cp9errors.Newf(cp9errors.CategoryConfig, "%v", err)
	}
	runtimeStateDir := filepath.Clean(filepath.Join(runtime.StateDir, filepath.FromSlash(membership.StateDir)))
	selected, err := loadSelectedSessionHost(runtimeStateDir)
	if err != nil {
		return nil, cp9errors.Newf(cp9errors.CategoryResource, "load %s host: %v", membership.ID, err)
	}
	status, err := cptilt.LoadStatus(cptilt.PathsForState(runtimeStateDir, nil).Status)
	if err != nil {
		return nil, cp9errors.Newf(cp9errors.CategoryResource, "load %s Tilt status: %v", membership.ID, err)
	}
	runtime.RuntimeID = membership.ID
	runtime.RuntimeStateDir = runtimeStateDir
	runtime.MultiRuntime = true
	runtime.Selected = selected
	runtime.Status = status
	return runtime, nil
}

func loadValidatedRuntimePlan(stateDir string, session *config.SessionConfig) (*cptilt.RuntimePlan, bool, error) {
	plan, err := cptilt.LoadRuntimePlan(stateDir)
	if errors.Is(err, os.ErrNotExist) {
		if config.DedicatedGCPEnabled(session) {
			return nil, false, cp9errors.New(cp9errors.CategoryResource, "runtime plan is missing; rerun cp9 up to finish the dual-runtime handoff")
		}
		return nil, false, nil
	}
	if err != nil {
		return nil, false, cp9errors.Newf(cp9errors.CategoryResource, "load runtime plan: %v", err)
	}
	if err := plan.ValidateTiltfile(cptilt.SessionAbsolutePath(stateDir, plan.TiltfilePath)); err != nil {
		return nil, false, cp9errors.Newf(cp9errors.CategoryConfig, "validate runtime plan: %v", err)
	}
	return plan, true, nil
}

// checkSessionContextStillMatches rejects config that the running session cannot
// adopt. The session context is written once, during cp9 up, and the Tiltfile
// reads its build parallelism from there -- so editing the value afterwards used
// to change nothing and say nothing, which reads as "my setting had no effect"
// rather than "this needs a new session".
func checkSessionContextStillMatches(
	cfgPath string,
	session *config.SessionConfig,
	stateDir string,
	state *cluster.State,
) error {
	// Only an explicit setting can be ignored, so only an explicit setting is
	// worth refusing over. Comparing the defaulted value would fail every
	// session created before the default changed, for operators who never set
	// the field and cannot act on the message.
	if !sessionSetsSourceBuildParallelism(cfgPath) {
		return nil
	}
	raw, err := os.ReadFile(cptilt.PathsForState(stateDir, state).ContextIndex)
	if err != nil {
		// No persisted context yet means nothing has been promised to Tilt.
		return nil
	}
	var persisted struct {
		SourceBuildParallelism int `json:"source_build_parallelism"`
	}
	if err := json.Unmarshal(raw, &persisted); err != nil {
		return nil
	}
	want := session.Services.SourceBuildParallelism
	if persisted.SourceBuildParallelism == 0 || want == 0 || persisted.SourceBuildParallelism == want {
		return nil
	}
	return cp9errors.Newf(
		cp9errors.CategoryConfig,
		"services.source_build_parallelism is %d in %s but the running session was created with %d; "+
			"the session context is written once at cp9 up, so change it back or recreate the session "+
			"with cp9 destroy && cp9 up",
		want,
		strings.TrimSpace(session.Name),
		persisted.SourceBuildParallelism,
	)
}

func loadV1Alpha2SessionStateDir(cfgPath, localPath string, localProvided bool) (*config.SessionConfig, string, error) {
	if err := rejectV1Alpha2LocalConfigOverride(localPath, localProvided); err != nil {
		return nil, "", err
	}
	session, err := config.LoadSession(cfgPath)
	if err != nil {
		return nil, "", err
	}
	return session, cluster.SessionStateDir(session.Name), nil
}

func ensureTiltSessionRunning(ctx context.Context, runtime *v1alpha2TiltRuntime) error {
	if runtime == nil {
		return cp9errors.New(cp9errors.CategoryResource, "persisted Tilt runtime is not available")
	}
	status := runtime.Status
	if status == nil {
		return cp9errors.New(cp9errors.CategoryResource, "persisted Tilt status is not available")
	}
	if status.Mode != "up" {
		return cp9errors.New(cp9errors.CategoryResource,
			"v1alpha2 Tilt session is not running; run `cp9 up` with tilt.mode=up first")
	}
	if cptilt.UsesRemoteController(runtime.Selected) {
		status.Controller = cptilt.ControllerRemote
		observation, err := cptilt.ObserveRemoteController(ctx, runtime.Selected, status)
		if err != nil {
			return cp9errors.Newf(cp9errors.CategoryResource, "remote Tilt controller is unreachable: %v", err)
		}
		if !observation.Running() {
			return cp9errors.Newf(
				cp9errors.CategoryResource,
				"remote Tilt controller is not running (active=%s sub=%s result=%s); rerun `cp9 up` to reconcile it",
				observation.ActiveState,
				observation.SubState,
				observation.Result,
			)
		}
		status.Running = true
		status.PID = observation.PID
		return nil
	}
	if !status.Running {
		return cp9errors.New(cp9errors.CategoryResource,
			"v1alpha2 Tilt session is not running; run `cp9 up` with tilt.mode=up first")
	}
	if status.PID <= 0 {
		return cp9errors.New(cp9errors.CategoryResource, "persisted Tilt session PID is missing")
	}
	if !day2ProcessRunning(status.PID) {
		return cp9errors.New(cp9errors.CategoryResource,
			"persisted Tilt session is not alive; rerun `cp9 up` to restart Tilt")
	}
	return nil
}

func runV1Alpha2DeployForRuntime(
	ctx context.Context,
	cfgPath, localPath string,
	localProvided bool,
	service, requestedRuntime string,
	dryRun bool,
) (*deployResult, error) {
	runtime, err := loadV1Alpha2TiltRuntimeForResource(
		cfgPath, localPath, localProvided, strings.TrimSpace(service), requestedRuntime,
	)
	if err != nil {
		return nil, err
	}
	if err := ensureTiltSessionRunning(ctx, runtime); err != nil {
		return nil, err
	}
	resource, err := v1alpha2Day2TiltResource(runtime, service)
	if err != nil {
		return nil, cp9errors.Newf(cp9errors.CategoryConfig, "%v", err)
	}
	if strings.TrimSpace(resource) == "" {
		return nil, cp9errors.Newf(cp9errors.CategoryResource, "service %s is missing a Tilt resource mapping", service)
	}
	result := deployResultFromTiltResource(service, resource, runtime.Status)
	if runtime.MultiRuntime {
		result.Runtime = runtime.RuntimeID
	}
	var beforeRefresh *cptilt.ResourceReadiness
	sourceChanged := false
	imageResource := resource + "-image"
	imageStampBefore := ""
	if !dryRun && cptilt.UsesRemoteController(runtime.Selected) && runtime.Status.Controller == cptilt.ControllerRemote {
		beforeRefresh, err = cptilt.ResourceReadinessForHost(ctx, runtime.Selected, runtime.Status)
		if err != nil {
			return nil, fmt.Errorf("read Tilt state before source refresh: %w", err)
		}
		if beforeRefresh == nil {
			return nil, fmt.Errorf("read Tilt state before source refresh: empty response")
		}
		if runtime.RuntimeID == cptilt.RuntimeGCPRegional {
			if _, sourceBacked := beforeRefresh.ResourceStates[resource+"-live-binary"]; sourceBacked {
				return nil, cp9errors.Newf(
					cp9errors.CategoryConfig,
					"GCP runtime source build is unsupported for %s; use its immutable owner image",
					service,
				)
			}
		}
		liveBinary := resource + "-live-binary"
		if state, ok := beforeRefresh.ResourceStates[liveBinary]; ok && state.TriggerMode != v1alpha2ManualTriggerMode {
			return nil, fmt.Errorf(
				"tilt resource %s is not in manual trigger mode; rerun `cp9 up` before deploying with this CP9 version",
				liveBinary,
			)
		}
		if _, hasLiveBinary := beforeRefresh.ResourceStates[liveBinary]; hasLiveBinary {
			if imageState, hasImage := beforeRefresh.ResourceStates[imageResource]; hasImage {
				imageStampBefore = imageState.ImageStamp
			}
		}
		changedServices, refreshErr := cptilt.RefreshRemoteControllerInputs(
			ctx,
			runtime.Selected,
			runtime.RuntimeStateDir,
			runtime.Status,
		)
		if refreshErr != nil {
			return nil, fmt.Errorf("refresh remote Tilt controller inputs: %w", refreshErr)
		}
		sourceChanged = changedServices[service]
	}
	readiness, err := cptilt.ResourceReadinessForHost(ctx, runtime.Selected, runtime.Status)
	if err != nil {
		return nil, fmt.Errorf("read Tilt state before deploy: %w", err)
	}
	if readiness == nil {
		return nil, fmt.Errorf("read Tilt state before deploy: empty response")
	}
	if !dryRun {
		if imageState, ok := readiness.ResourceStates[imageResource]; ok && imageState.TriggerMode == v1alpha2ManualWithAutoInitTriggerMode {
			// The image builder's fallback snapshot is the source of truth; the
			// atomically replaced Git root is not a reliable Tilt file-watch signal.
			if imageStampBefore == "" {
				imageStampBefore = imageState.ImageStamp
			}
			serviceBefore, ok := readiness.ResourceStates[resource]
			if !ok {
				return nil, fmt.Errorf("tilt resource %s was not found before triggering %s", resource, imageResource)
			}
			if err := triggerV1Alpha2TiltResource(ctx, runtime, imageResource); err != nil {
				return nil, err
			}
			if err := waitForV1Alpha2TiltResource(ctx, runtime, imageResource, &imageState); err != nil {
				return nil, err
			}
			imageReadiness, err := cptilt.ResourceReadinessForHost(ctx, runtime.Selected, runtime.Status)
			if err != nil {
				return nil, fmt.Errorf("read Tilt state after triggering %s: %w", imageResource, err)
			}
			if imageReadiness == nil {
				return nil, fmt.Errorf("read Tilt state after triggering %s: empty response", imageResource)
			}
			imageAfter, ok := imageReadiness.ResourceStates[imageResource]
			if !ok {
				return nil, fmt.Errorf("tilt resource %s disappeared after its update", imageResource)
			}
			imageStampAfter := imageAfter.ImageStamp
			if imageStampAfter == "" {
				return nil, fmt.Errorf("tilt resource %s did not publish its image stamp", imageResource)
			}
			// An empty pre-trigger stamp is only stamp initialization (first
			// deploy or a Tiltfile reload). Keep going so this command still
			// updates the selected service: source-only deploys must not ask
			// for a rerun, and unchanged source must not return success after
			// only publishing the stamp.
			if imageStampBefore != "" && imageStampAfter != imageStampBefore {
				if err := waitForV1Alpha2TiltResource(ctx, runtime, resource, &serviceBefore); err != nil {
					return nil, err
				}
				afterService, err := cptilt.ResourceReadinessForHost(ctx, runtime.Selected, runtime.Status)
				if err != nil {
					return nil, fmt.Errorf("read Tilt state after updating %s: %w", resource, err)
				}
				if afterService == nil {
					return nil, fmt.Errorf("read Tilt state after updating %s: empty response", resource)
				}
				liveBinary := resource + "-live-binary"
				live, ok := afterService.ResourceStates[liveBinary]
				if !ok || !tiltResourceUpdateFailed(&live) {
					return result, nil
				}
				// Image-path success leaves a previously failed manual
				// live-binary in Tilt error, so cp9 status stays failed.
				// Continue and compile current source to clear that error.
				readiness = afterService
			}
		}
	}
	resources, err := v1alpha2DeployResources(readiness, beforeRefresh, sourceChanged, resource)
	if err != nil {
		return nil, err
	}
	if dryRun {
		result.DryRun = true
		for i := range resources {
			if resources[i].trigger {
				result.TriggerCommand = tiltTriggerCommand(runtime.Status, resources[i].resource)
				break
			}
		}
		return result, nil
	}
	for i := range resources {
		if resources[i].trigger {
			if err := triggerV1Alpha2TiltResource(ctx, runtime, resources[i].resource); err != nil {
				return nil, err
			}
		}
		if err := waitForV1Alpha2TiltResource(ctx, runtime, resources[i].resource, &resources[i].before); err != nil {
			return nil, err
		}
	}
	return result, nil
}

func tiltResourceUpdateFailed(state *cptilt.ResourceState) bool {
	if state == nil {
		return false
	}
	switch strings.ToLower(strings.TrimSpace(state.UpdateStatus)) {
	case "error", "failed":
		return true
	default:
		return false
	}
}

func triggerV1Alpha2TiltResource(ctx context.Context, runtime *v1alpha2TiltRuntime, resource string) error {
	if cptilt.UsesRemoteController(runtime.Selected) && runtime.Status.Controller == cptilt.ControllerRemote {
		return cptilt.TriggerForHost(ctx, runtime.Selected, runtime.Status, resource)
	}
	return runTiltTrigger(ctx, runtime.Status, resource)
}

const (
	// A cold image rebuild plus rollout is minutes; live-update is seconds. The
	// timeout has to clear the slow path without stranding an operator on it.
	v1alpha2DeployConvergeTimeout         = 20 * time.Minute
	v1alpha2ManualWithAutoInitTriggerMode = 1
	v1alpha2ManualTriggerMode             = 2
)

var v1alpha2DeployPollInterval = 2 * time.Second

type v1alpha2DeployStep struct {
	resource string
	trigger  bool
	before   cptilt.ResourceState
}

// v1alpha2DeployResources builds the explicit resource plan for a deploy.
// A source-build service compiles in <resource>-live-binary and its output
// automatically drives the service live update, so the service step waits
// without a second trigger. Remote-controller live-binary resources are manual,
// so deploy triggers the companion exactly once after refreshing source inputs.
// A service without a live-binary companion uses one trigger.
func v1alpha2DeployResources(
	readiness *cptilt.ResourceReadiness,
	beforeRefresh *cptilt.ResourceReadiness,
	sourceChanged bool,
	resource string,
) ([]v1alpha2DeployStep, error) {
	if readiness == nil {
		return nil, fmt.Errorf("read Tilt resources before deploy: empty response")
	}
	liveBinary := resource + "-live-binary"
	_, currentHasLiveBinary := readiness.ResourceStates[liveBinary]
	if beforeRefresh != nil {
		_, beforeHasLiveBinary := beforeRefresh.ResourceStates[liveBinary]
		if beforeHasLiveBinary != currentHasLiveBinary {
			return nil, fmt.Errorf("tilt resource %s companion presence changed across source refresh", liveBinary)
		}
	}
	if sourceChanged {
		if beforeRefresh == nil {
			return nil, fmt.Errorf("cannot plan a fresh source deploy without a pre-refresh Tilt snapshot")
		}
	}
	steps := []v1alpha2DeployStep{{resource: resource, trigger: true}}
	if !currentHasLiveBinary {
		state, ok := readiness.ResourceStates[resource]
		if !ok {
			return nil, fmt.Errorf("tilt resource %s was not found before triggering it", resource)
		}
		steps[0].before = state
		return steps, nil
	}
	if beforeRefresh != nil && readiness.ResourceStates[liveBinary].TriggerMode != v1alpha2ManualTriggerMode {
		return nil, fmt.Errorf("tilt resource %s is not in manual trigger mode", liveBinary)
	}
	liveBefore := readiness.ResourceStates[liveBinary]
	steps = []v1alpha2DeployStep{
		{resource: liveBinary, trigger: true, before: liveBefore},
		{resource: resource, trigger: false},
	}
	for i := range steps {
		if i == 0 {
			continue
		}
		state, ok := readiness.ResourceStates[steps[i].resource]
		if !ok {
			return nil, fmt.Errorf("tilt resource %s was not found before triggering it", steps[i].resource)
		}
		if sourceChanged {
			state, ok = beforeRefresh.ResourceStates[steps[i].resource]
			if !ok {
				return nil, fmt.Errorf("tilt resource %s was not present before source refresh", steps[i].resource)
			}
		}
		steps[i].before = state
	}
	return steps, nil
}

// waitForV1Alpha2TiltResource blocks until Tilt finishes the update it was just
// asked for. `tilt trigger` only enqueues, so without this the command reports
// success while the cluster still runs the old build. A new build identity
// must appear after this trigger; a previous ready state is not evidence of
// this deploy.
func waitForV1Alpha2TiltResource(
	ctx context.Context,
	runtime *v1alpha2TiltRuntime,
	resource string,
	before *cptilt.ResourceState,
) error {
	waitCtx, cancel := context.WithTimeout(ctx, v1alpha2DeployConvergeTimeout)
	defer cancel()
	poll := func() error {
		timer := time.NewTimer(v1alpha2DeployPollInterval)
		defer timer.Stop()
		select {
		case <-waitCtx.Done():
			return waitCtx.Err()
		case <-timer.C:
			return nil
		}
	}
	var lastStatus string
	timeoutError := func() error {
		return fmt.Errorf(
			"tilt resource %s did not finish updating within %s (last update status %q)",
			resource,
			v1alpha2DeployConvergeTimeout,
			lastStatus,
		)
	}
	for {
		readiness, err := readTiltReadiness(waitCtx, runtime.Selected, runtime.Status)
		if err != nil {
			if errors.Is(err, context.DeadlineExceeded) && waitCtx.Err() != nil {
				return timeoutError()
			}
			return fmt.Errorf("read tilt state while waiting for %s: %w", resource, err)
		}
		if readiness == nil {
			return fmt.Errorf("read tilt state while waiting for %s: empty response", resource)
		}
		state, ok := readiness.ResourceStates[resource]
		if !ok {
			return fmt.Errorf("tilt resource %s disappeared while waiting for it to converge", resource)
		}
		lastStatus = state.UpdateStatus
		if v1alpha2DeployErrorChanged(before, &state) {
			return fmt.Errorf("tilt resource %s failed to update: %s", resource, strings.TrimSpace(state.Message))
		}
		fresh := v1alpha2DeployBuildChanged(before, &state)
		if fresh && strings.EqualFold(state.UpdateStatus, "ok") {
			return nil
		}
		if err := poll(); err != nil {
			if errors.Is(err, context.DeadlineExceeded) {
				return timeoutError()
			}
			return fmt.Errorf("wait for tilt resource %s: %w", resource, err)
		}
	}
}

func v1alpha2DeployBuildChanged(before, after *cptilt.ResourceState) bool {
	if before == nil || after == nil || after.BuildHistoryCount == 0 ||
		after.LatestBuildSpanID == "" || after.LatestBuildStart == "" {
		return false
	}
	if after.BuildHistoryCount < before.BuildHistoryCount {
		return false
	}
	if after.BuildHistoryCount > before.BuildHistoryCount {
		return true
	}
	return before.LatestBuildSpanID != after.LatestBuildSpanID ||
		before.LatestBuildStart != after.LatestBuildStart
}

func v1alpha2DeployErrorChanged(before, after *cptilt.ResourceState) bool {
	if after == nil || !strings.EqualFold(after.UpdateStatus, "error") {
		return false
	}
	if before == nil {
		return true
	}
	return !strings.EqualFold(before.UpdateStatus, "error") ||
		before.Reason != after.Reason ||
		before.Message != after.Message ||
		before.LastTransitionTime != after.LastTransitionTime ||
		v1alpha2DeployBuildChanged(before, after)
}

func deployResultFromTiltResource(service, resource string, status *cptilt.Status) *deployResult {
	return &deployResult{
		Service:   service,
		Resource:  resource,
		Namespace: status.Namespace,
	}
}

func tiltTriggerCommand(status *cptilt.Status, resource string) string {
	apiHost := ""
	port := 0
	if status != nil {
		apiHost = status.APIHost
		port = status.APIPort
	}
	apiHost = firstNonEmptyString(apiHost, "localhost")
	if port <= 0 {
		port = cptilt.DefaultAPIPort
	}
	return fmt.Sprintf("tilt trigger %s --host %s --port %d", resource, apiHost, port)
}

func runDeploy(
	ctx context.Context,
	cfg *config.Config,
	local *config.LocalConfig,
	service string,
	dryRun bool,
) (*deployResult, error) {
	selected, err := loadSelectedHost(cfg, local)
	if err != nil {
		return nil, fmt.Errorf("load selected host: %w", err)
	}

	svcPlan, err := loadServicePlan(selected.StateDir, service)
	if err != nil {
		return nil, err
	}
	if dryRun {
		artifacts, err := loadDay2Artifacts(selected.StateDir)
		if err != nil {
			return nil, fmt.Errorf("load artifacts: %w", err)
		}
		artifact := artifacts.Services[service]
		if artifact == nil {
			return nil, fmt.Errorf("artifact not found for %s", service)
		}
		return &deployResult{
			Service:      service,
			Role:         string(svcPlan.Role),
			ImageRef:     artifact.ImageRef,
			Namespace:    firstNonEmptyString(artifact.Namespace, selected.Namespace),
			WorkloadKind: artifact.WorkloadKind,
			WorkloadName: artifact.WorkloadName,
			OverlayDir:   artifact.OverlayDir,
			DryRun:       true,
		}, nil
	}

	switch svcPlan.Role {
	case persona.RoleSource:
		if err := runBuildRealStep(ctx, &lifecycle.BuildRealStep{
			StateDir:     selected.StateDir,
			Selected:     selected,
			Services:     []string{service},
			ForceRebuild: true,
		}); err != nil {
			return nil, err
		}
	case persona.RoleLive:
		// Live services do not have a local build/gen phase in the narrow wedge.
	case persona.RoleStub:
		if err := runGenStubsStep(ctx, &lifecycle.GenStubsStep{
			StateDir: selected.StateDir,
			Selected: selected,
			Services: []string{service},
		}); err != nil {
			return nil, err
		}
	default:
		return nil, fmt.Errorf("service %s has unsupported role %q", service, svcPlan.Role)
	}

	if err := runConfigGenStep(ctx, &lifecycle.ConfigGenStep{
		StateDir: selected.StateDir,
		Config:   cfg,
	}); err != nil {
		return nil, err
	}
	if err := runDeployStep(ctx, &lifecycle.DeployStep{
		StateDir:     selected.StateDir,
		Selected:     selected,
		Services:     []string{service},
		ForceRollout: true,
	}); err != nil {
		return nil, err
	}
	if err := runReadinessStep(ctx, &lifecycle.ReadinessStep{
		StateDir: selected.StateDir,
		Selected: selected,
		Config:   cfg,
		Services: []string{service},
	}); err != nil {
		return nil, err
	}

	artifacts, err := loadDay2Artifacts(selected.StateDir)
	if err != nil {
		return nil, fmt.Errorf("load artifacts: %w", err)
	}
	artifact := artifacts.Services[service]
	if artifact == nil {
		return nil, fmt.Errorf("artifact not found for %s", service)
	}
	return &deployResult{
		Service:      service,
		Role:         string(svcPlan.Role),
		ImageRef:     artifact.ImageRef,
		Namespace:    firstNonEmptyString(artifact.Namespace, selected.Namespace),
		WorkloadKind: artifact.WorkloadKind,
		WorkloadName: artifact.WorkloadName,
		OverlayDir:   artifact.OverlayDir,
	}, nil
}

type logsQuery struct {
	Service string
	TraceID string
	Since   string
	Level   string
	Limit   int
}

func parseLogsLimit(raw string) (int, error) {
	value := strings.TrimSpace(raw)
	if value == "" {
		return 200, nil
	}
	limit, err := strconv.Atoi(value)
	if err != nil || limit <= 0 {
		return 0, cp9errors.Newf(cp9errors.CategoryConfig, "logs limit must be a positive integer, got %q", raw)
	}
	return limit, nil
}

func logsLimit(query logsQuery) int {
	if query.Limit <= 0 {
		return 200
	}
	return query.Limit
}

var (
	errNoPodsForWorkload      = errors.New("no pods found for workload")
	errWorkloadMetadataAbsent = errors.New("workload metadata missing")
)

func collectLogs(ctx context.Context, selected *host.Session, artifacts *cluster.ArtifactState, query logsQuery) (*logsResult, error) {
	services, err := artifactTargets(artifacts, query.Service)
	if err != nil {
		return nil, err
	}

	result := &logsResult{
		Sources: []string{"host"},
		Service: query.Service,
		TraceID: query.TraceID,
		Since:   query.Since,
		Level:   query.Level,
		Limit:   query.Limit,
	}

	for _, name := range services {
		artifact := artifacts.Services[name]
		if artifact == nil {
			continue
		}
		if query.Service == "" && !artifactHasWorkloadMetadata(artifact) {
			continue
		}
		namespace := firstNonEmptyString(artifact.Namespace, selected.Namespace)
		pods, err := podNamesForArtifact(ctx, selected, namespace, artifact)
		if err != nil {
			if query.Service == "" && (errors.Is(err, errNoPodsForWorkload) || errors.Is(err, errWorkloadMetadataAbsent)) {
				continue
			}
			return nil, fmt.Errorf("%s pods: %w", name, err)
		}
		for _, pod := range pods {
			args := []string{"-n", namespace, "logs", pod, "--tail", strconv.Itoa(logsLimit(query))}
			if artifact.ContainerName != "" {
				args = append(args, "-c", artifact.ContainerName)
			} else {
				args = append(args, "--all-containers=true")
			}
			if query.Since != "" {
				args = append(args, "--since", query.Since)
			}
			output, err := kubectlOutput(ctx, selected, args...)
			if err != nil {
				return nil, fmt.Errorf("%s logs for %s: %w", name, pod, err)
			}
			for _, line := range boundedLogLines(filterLogLines(string(output), query.TraceID, query.Level), query.Limit) {
				result.Entries = append(result.Entries, logEntry{
					Cluster:   "host",
					Service:   name,
					Pod:       pod,
					Container: artifact.ContainerName,
					Line:      line,
				})
			}
		}
	}

	return finalizeLogsResult(result), nil
}

func runV1Alpha2Logs(
	ctx context.Context,
	cfgPath, localPath string,
	localProvided bool,
	query logsQuery,
	requestedRuntime string,
) (*logsResult, error) {
	if query.Service != "" {
		runtime, err := loadV1Alpha2TiltRuntimeForResource(
			cfgPath, localPath, localProvided, query.Service, requestedRuntime,
		)
		if err != nil {
			return nil, err
		}
		resource, err := v1alpha2Day2TiltResource(runtime, query.Service)
		if err != nil {
			return nil, cp9errors.Newf(cp9errors.CategoryConfig, "%v", err)
		}
		return collectV1Alpha2ServiceLogs(ctx, runtime, resource, query, cfgPath, requestedRuntime)
	}
	return collectV1Alpha2RuntimeGraphLogs(ctx, cfgPath, localPath, localProvided, query, requestedRuntime)
}

func collectV1Alpha2RuntimeGraphLogs(
	ctx context.Context,
	cfgPath, localPath string,
	localProvided bool,
	query logsQuery,
	requestedRuntime string,
) (*logsResult, error) {
	base, err := loadV1Alpha2TiltRuntime(cfgPath, localPath, localProvided)
	if err != nil {
		return nil, err
	}
	plan, found, err := loadValidatedRuntimePlan(base.StateDir, base.Session)
	if err != nil {
		return nil, err
	}
	if !found {
		if requestedRuntime != "" && requestedRuntime != cptilt.RuntimePrimary {
			return nil, cp9errors.Newf(cp9errors.CategoryConfig, "runtime %q is not configured", requestedRuntime)
		}
		return finalizeLogsResult(&logsResult{
			Sources: []string{"host"}, TraceID: query.TraceID, Since: query.Since, Level: query.Level, Limit: query.Limit,
		}), nil
	}
	result := &logsResult{TraceID: query.TraceID, Since: query.Since, Level: query.Level, Limit: logsLimit(query)}
	matchedRuntime := requestedRuntime == ""
	for i := range plan.Runtimes {
		membership := &plan.Runtimes[i]
		if requestedRuntime != "" && membership.ID != requestedRuntime {
			continue
		}
		matchedRuntime = true
		if len(membership.Resources) == 0 {
			result.Attempted = append(result.Attempted, logSourceAttempt{
				Source: "tilt", Substrate: membership.ID, Status: "empty", Message: "runtime graph has no resources",
			})
			continue
		}
		runtime, loadErr := loadV1Alpha2TiltRuntimeForResource(
			cfgPath, localPath, localProvided, membership.Resources[0], membership.ID,
		)
		if loadErr != nil {
			result.Attempted = append(result.Attempted, logSourceAttempt{
				Source: "tilt", Substrate: membership.ID, Status: "failed", Message: loadErr.Error(),
			})
			continue
		}
		if runErr := ensureTiltSessionRunning(ctx, runtime); runErr != nil {
			result.Attempted = append(result.Attempted, logSourceAttempt{
				Source: "tilt", Substrate: membership.ID, Status: "failed", Message: runErr.Error(),
			})
			continue
		}
		output, logsErr := cptilt.LogsForHost(ctx, runtime.Selected, runtime.Status, &cptilt.LogsRequest{
			Resources: membership.Resources,
			Since:     query.Since,
			Level:     query.Level,
			Tail:      query.Limit,
		})
		if logsErr != nil {
			result.Attempted = append(result.Attempted, logSourceAttempt{
				Source: "tilt", Substrate: membership.ID, Status: "failed", Message: logsErr.Error(),
			})
			continue
		}
		result.Sources = appendSource(result.Sources, membership.ID+"/tilt")
		result.Attempted = append(result.Attempted, logSourceAttempt{
			Source: "tilt", Substrate: membership.ID, Status: "queried",
		})
		for _, line := range boundedLogLines(filterLogLines(string(output), query.TraceID, query.Level), query.Limit) {
			result.Entries = append(result.Entries, logEntry{
				Cluster: membership.ID, Substrate: "tilt", Service: "*", Pod: "*", Stream: "tilt", Line: redactLogLine(line),
			})
		}
	}
	if !matchedRuntime {
		return nil, cp9errors.Newf(cp9errors.CategoryConfig, "runtime %q is not present in the runtime plan", requestedRuntime)
	}
	result.Entries = boundedLogEntries(result.Entries, result.Limit)
	failed := false
	completed := false
	for _, attempt := range result.Attempted {
		failed = failed || attempt.Status == "failed"
		completed = completed || attempt.Status == "queried" || attempt.Status == "empty"
	}
	if len(result.Entries) > 0 {
		result.Status = "matched"
		if failed {
			result.Status = "partial"
		}
		return result, nil
	}
	result.Status = "empty"
	result.EmptyReason = result.emptyMessage()
	if failed {
		if completed {
			result.Status = "partial"
			return result, nil
		}
		result.Status = "incomplete"
		return result, cp9errors.New(cp9errors.CategoryResource, "no runtime log source completed successfully")
	}
	return result, nil
}

func collectV1Alpha2ServiceLogs(
	ctx context.Context,
	runtime *v1alpha2TiltRuntime,
	resource string,
	query logsQuery,
	configPath string,
	requestedRuntime string,
) (*logsResult, error) {
	runtimeID := ""
	if runtime.MultiRuntime {
		runtimeID = firstNonEmptyString(runtime.RuntimeID, cptilt.RuntimePrimary)
	}
	tiltSubstrate := firstNonEmptyString(runtimeID, "tilt")
	tiltSource := "tilt"
	kubernetesSubstrate := firstNonEmptyString(runtimeID, "k3d")
	kubernetesSource := "k3d"
	if runtimeID != "" {
		tiltSource = runtimeID + "/tilt"
		kubernetesSource = runtimeID + "/k3d"
	}
	result := &logsResult{
		Service: query.Service,
		TraceID: query.TraceID,
		Since:   query.Since,
		Level:   query.Level,
		Limit:   logsLimit(query),
	}
	// The Session Tiltfile owns the resource graph, so cp9 persists no per-service
	// status: the workload always has to be discovered from the session.
	discoverSessionWorkload := runtime.Session != nil && config.UsesSessionTiltfile(runtime.Session)
	tiltFailed := false
	if err := ensureTiltSessionRunning(ctx, runtime); err != nil {
		tiltFailed = true
		result.Attempted = append(
			result.Attempted,
			logSourceAttempt{Source: "tilt", Substrate: tiltSubstrate, Status: "failed", Message: err.Error()},
		)
	} else if strings.TrimSpace(resource) == "" {
		tiltFailed = true
		result.Attempted = append(
			result.Attempted,
			logSourceAttempt{Source: "tilt", Substrate: tiltSubstrate, Status: "failed", Message: "service is missing a Tilt resource mapping"},
		)
	} else {
		var output []byte
		var err error
		if cptilt.UsesRemoteController(runtime.Selected) && runtime.Status.Controller == cptilt.ControllerRemote {
			output, err = cptilt.LogsForHost(ctx, runtime.Selected, runtime.Status, &cptilt.LogsRequest{
				Resources: []string{resource},
				Since:     query.Since,
				Level:     query.Level,
				Tail:      query.Limit,
			})
		} else {
			output, err = runTiltLogs(ctx, runtime.Status, []string{resource}, query)
		}
		if err != nil {
			tiltFailed = true
			result.Attempted = append(
				result.Attempted,
				logSourceAttempt{Source: "tilt", Substrate: tiltSubstrate, Workload: resource, Status: "failed", Message: err.Error()},
			)
		} else {
			result.Sources = appendSource(result.Sources, tiltSource)
			result.Attempted = append(
				result.Attempted,
				logSourceAttempt{Source: "tilt", Substrate: tiltSubstrate, Workload: resource, Status: "queried"},
			)
			for _, line := range boundedLogLines(filterLogLines(string(output), query.TraceID, query.Level), query.Limit) {
				result.Entries = append(result.Entries, logEntry{
					Cluster: tiltSubstrate, Substrate: "tilt", Service: query.Service, Pod: resource, Stream: "tilt", Line: redactLogLine(line),
				})
			}
		}
	}

	remoteTarget, hasRemoteTarget := starterRemoteLogsTarget(query.Service)
	// cp9 persists no workload metadata, so the workload is always discovered
	// from the session.
	if discoverSessionWorkload {
		artifact, namespace, discoverErr := discoverExactNameWorkload(ctx, runtime.Selected, query.Service)
		if discoverErr != nil {
			status := "failed"
			if errors.Is(discoverErr, errExactNameWorkloadNotFound) {
				status = "not_found"
			}
			result.Attempted = append(result.Attempted, logSourceAttempt{
				Source: "kubernetes", Substrate: kubernetesSubstrate, Status: status, Message: discoverErr.Error(),
			})
		} else {
			entries, attempt := collectKubernetesWorkloadLogs(
				ctx, runtime.Selected, kubernetesSubstrate, query.Service, namespace, artifact, query,
			)
			result.Attempted = append(result.Attempted, attempt)
			if attempt.Status == "queried" {
				result.Sources = appendSource(result.Sources, kubernetesSource)
			}
			result.Entries = append(result.Entries, entries...)
		}
	}

	if hasRemoteTarget {
		remote, remoteErr := loadStarterRemoteEKSArtifact(runtime.StateDir)
		if remoteErr != nil {
			result.Attempted = append(
				result.Attempted,
				logSourceAttempt{
					Source:    "kubernetes",
					Substrate: "starter-eks",
					Namespace: remoteTarget.Namespace,
					Workload:  remoteTarget.Kind + "/" + remoteTarget.Name,
					Status:    "failed",
					Message:   remoteErr.Error(),
				},
			)
		} else {
			selected := starterRemoteLogsSession(runtime, remote, remoteTarget.Namespace)
			artifact := &cluster.ServiceArtifact{
				WorkloadKind: remoteTarget.Kind,
				WorkloadName: remoteTarget.Name,
				Namespace:    remoteTarget.Namespace,
			}
			entries, attempt := collectKubernetesWorkloadLogs(
				ctx, selected, "starter-eks", query.Service, remoteTarget.Namespace, artifact, query,
			)
			result.Attempted = append(result.Attempted, attempt)
			if attempt.Status == "queried" {
				result.Sources = appendSource(result.Sources, "starter-eks")
			}
			result.Entries = append(result.Entries, entries...)
		}
	} else if discoverSessionWorkload {
		remote, remoteErr := loadStarterRemoteEKSArtifact(runtime.StateDir)
		if remoteErr == nil {
			selected := starterRemoteLogsSession(runtime, remote, "")
			artifact, namespace, discoverErr := discoverExactNameWorkload(ctx, selected, query.Service)
			if discoverErr != nil {
				status := "failed"
				if errors.Is(discoverErr, errExactNameWorkloadNotFound) {
					status = "not_found"
				}
				result.Attempted = append(result.Attempted, logSourceAttempt{
					Source: "kubernetes", Substrate: "starter-eks", Status: status, Message: discoverErr.Error(),
				})
			} else {
				entries, attempt := collectKubernetesWorkloadLogs(
					ctx, selected, "starter-eks", query.Service, namespace, artifact, query,
				)
				result.Attempted = append(result.Attempted, attempt)
				if attempt.Status == "queried" {
					result.Sources = appendSource(result.Sources, "starter-eks")
				}
				result.Entries = append(result.Entries, entries...)
			}
		} else if remoteErr != nil && !errors.Is(remoteErr, os.ErrNotExist) {
			result.Attempted = append(result.Attempted, logSourceAttempt{
				Source: "kubernetes", Substrate: "starter-eks", Status: "failed", Message: remoteErr.Error(),
			})
		}
	}

	if resources, ok := dedicatedServiceDiagnosticResources(query.Service); ok {
		entries, attempt := collectDedicatedAPIDiagnostics(ctx, runtime.Selected, query.Service, resources, query)
		result.Attempted = append(result.Attempted, attempt)
		if attempt.Status == "queried" {
			result.Sources = appendSource(result.Sources, "dedicated-api")
		}
		result.Entries = append(result.Entries, entries...)
	}

	result.Entries = boundedLogEntries(result.Entries, result.Limit)
	if len(result.Entries) == 0 {
		result.Status = "incomplete"
		result.EmptyReason = "no workload log lines matched; evidence is incomplete"
		result.NextQuery = logsNextQuery(query, configPath, requestedRuntime)
		var failures []string
		for _, attempt := range result.Attempted {
			if attempt.Message != "" {
				failures = append(failures, attempt.Source+": "+attempt.Message)
			}
		}
		detail := "attempted sources are listed in output"
		if len(failures) > 0 {
			detail = strings.Join(failures, "; ")
		}
		return result, cp9errors.Newf(cp9errors.CategoryResource, "incomplete log evidence for %s: %s", query.Service, detail)
	}
	result.Status = "matched"
	for _, attempt := range result.Attempted {
		if attempt.Status == "failed" {
			result.Status = "partial"
			result.NextQuery = logsNextQuery(query, configPath, requestedRuntime)
			break
		}
	}
	if tiltFailed && slices.Contains(result.Sources, kubernetesSource) {
		fallback := "tilt-fallback"
		if runtimeID != "" {
			fallback = runtimeID + "/tilt-fallback"
		}
		result.Sources = appendSource(result.Sources, fallback)
	}
	return result, nil
}

func dedicatedServiceDiagnosticResources(service string) ([]string, bool) {
	resources, ok := map[string][]string{
		"regional-server":                   {"vpcs.dedicated.regionalserver.tidbcloud.com", "privatelinkservices.dedicated.regionalserver.tidbcloud.com", "upgradeoperations.dedicated.regionalserver.tidbcloud.com"},
		"cluster-control-server":            {"clusters.cloud.pingcap.com", "workflows.cloud.pingcap.com", "configchanges.cloud.pingcap.com"},
		"cluster-controller":                {"clusters.cloud.pingcap.com", "workflows.cloud.pingcap.com", "configchanges.cloud.pingcap.com"},
		"dedicated-aws-provider":            {"providerconfigs.aws.crossplane.io", "providerconfigusages.aws.crossplane.io"},
		"dedicated-ticdc-service":           {"ticdcs.cdc.tidbcloud.com", "sinkendpoints.cdc.tidbcloud.com"},
		"dedicated-infra-api-crds":          {"providerconfigs.aws.crossplane.io", "clusters.cloud.pingcap.com"},
		"dedicated-cluster-control-support": {"clusters.cloud.pingcap.com", "workflows.cloud.pingcap.com"},
	}[strings.TrimSpace(service)]
	return resources, ok
}

func collectDedicatedAPIDiagnostics(
	ctx context.Context,
	selected *host.Session,
	service string,
	resources []string,
	query logsQuery,
) ([]logEntry, logSourceAttempt) {
	attempt := logSourceAttempt{Source: "kubernetes", Substrate: "dedicated-api", Workload: service, Status: "failed"}
	if selected == nil || selected.DedicatedAPI == nil || strings.TrimSpace(selected.DedicatedAPI.AdminKubeconfigPath) == "" {
		attempt.Message = "Dedicated API admin kubeconfig is unavailable"
		return nil, attempt
	}
	dedicated := &host.Session{
		Mode:           config.HostModeLocalK3d,
		StateDir:       selected.StateDir,
		ClusterName:    "dedicated-api",
		KubeconfigPath: selected.DedicatedAPI.AdminKubeconfigPath,
		Context:        "cp9-dedicated",
	}
	attempt.Context = dedicated.Context
	var entries []logEntry
	var failures []string
	for _, resource := range resources {
		output, err := runDedicatedDiagnosticsKubectl(
			ctx,
			dedicated,
			"get",
			resource,
			"-A",
			"--ignore-not-found",
			"-o",
			"custom-columns=NAMESPACE:.metadata.namespace,NAME:.metadata.name,GENERATION:.metadata.generation,OBSERVED:.status.observedGeneration,READY:.status.conditions[-1].status,REASON:.status.conditions[-1].reason",
		)
		if err != nil {
			failures = append(failures, resource+": "+err.Error())
			continue
		}
		entries = appendDedicatedDiagnosticLines(entries, service, "status/"+resource, string(output), query)
	}
	events, err := runDedicatedDiagnosticsKubectl(
		ctx,
		dedicated,
		"get",
		"events",
		"-A",
		"--sort-by=.lastTimestamp",
		"-o",
		"custom-columns=NAMESPACE:.metadata.namespace,TYPE:.type,KIND:.involvedObject.kind,OBJECT:.involvedObject.name,REASON:.reason,MESSAGE:.message",
	)
	if err != nil {
		failures = append(failures, "events: "+err.Error())
	} else {
		entries = appendDedicatedDiagnosticLines(entries, service, "events", string(events), query)
	}
	attempt.Status = "queried"
	if len(failures) > 0 {
		attempt.Message = strings.Join(failures, "; ")
		if len(entries) == 0 {
			attempt.Status = "failed"
		}
	}
	return entries, attempt
}

func appendDedicatedDiagnosticLines(entries []logEntry, service, stream, output string, query logsQuery) []logEntry {
	for _, line := range boundedLogLines(filterLogLines(output, query.TraceID, query.Level), logsLimit(query)) {
		entries = append(entries, logEntry{
			Cluster: "dedicated-api", Substrate: "dedicated-api", Context: "cp9-dedicated",
			Service: service, Pod: "metadata", Stream: stream, Line: redactLogLine(line),
		})
	}
	return entries
}

func starterRemoteLogsSession(runtime *v1alpha2TiltRuntime, remote *cpterraform.RemoteEKSOutput, namespace string) *host.Session {
	profile := ""
	region := strings.TrimSpace(remote.Region)
	if runtime.Session != nil {
		profile = strings.TrimSpace(runtime.Session.Target.AWS.IAM.Profile)
		region = firstNonEmptyString(region, runtime.Session.Target.Region)
	}
	return &host.Session{
		Mode: config.HostModeRemoteEKS, StateDir: runtime.StateDir,
		ClusterName: remote.ClusterName, Namespace: namespace,
		KubeconfigPath: filepath.Join(runtime.StateDir, "remote-eks", "kubeconfig.yaml"),
		Context:        firstNonEmptyString(remote.KubeconfigContext, remote.ClusterName),
		Profile:        profile,
		Region:         region,
	}
}

func logsNextQuery(query logsQuery, configPath, requestedRuntime string) string {
	configPath = strings.TrimSpace(configPath)
	if configPath == "" {
		configPath = config.DefaultConfigFile
	}
	command := "cp9 logs config=" + shellQuote(configPath) + " service=" + shellQuote(query.Service)
	if requestedRuntime != "" {
		command += " runtime=" + shellQuote(requestedRuntime)
	}
	return command + " since=30m limit=500"
}

var (
	errExactNameWorkloadNotFound  = errors.New("exact-name workload not found")
	errExactNameWorkloadAmbiguous = errors.New("exact-name workload is ambiguous")
)

type exactNameWorkloadMetadata struct {
	Name      string `json:"name"`
	Namespace string `json:"namespace"`
}

type exactNameWorkloadItem struct {
	Kind     string                    `json:"kind"`
	Metadata exactNameWorkloadMetadata `json:"metadata"`
}

type exactNameWorkloadList struct {
	Items []exactNameWorkloadItem `json:"items"`
}

func discoverExactNameWorkload(
	ctx context.Context,
	selected *host.Session,
	service string,
) (*cluster.ServiceArtifact, string, error) {
	service = strings.TrimSpace(service)
	if service == "" {
		return nil, "", fmt.Errorf("service name is required for k3d workload discovery")
	}
	out, err := kubectlOutput(ctx, selected, "get", "deployment,statefulset,daemonset", "-A", "-o", "json")
	if err != nil {
		return nil, "", fmt.Errorf("list session workloads across namespaces: %w", err)
	}
	var listed exactNameWorkloadList
	if err := json.Unmarshal(out, &listed); err != nil {
		return nil, "", fmt.Errorf("parse session workload inventory: %w", err)
	}
	var matches []cluster.ServiceArtifact
	for _, item := range listed.Items {
		if strings.TrimSpace(item.Metadata.Name) != service {
			continue
		}
		matches = append(matches, cluster.ServiceArtifact{
			WorkloadKind: strings.ToLower(strings.TrimSpace(item.Kind)),
			WorkloadName: service,
			Namespace:    strings.TrimSpace(item.Metadata.Namespace),
		})
	}
	if len(matches) == 0 {
		return nil, "", fmt.Errorf("%w: service %s has no exact-name deployment, statefulset, or daemonset", errExactNameWorkloadNotFound, service)
	}
	if len(matches) > 1 {
		locations := make([]string, 0, len(matches))
		for i := range matches {
			match := &matches[i]
			locations = append(locations, match.Namespace+"/"+match.WorkloadKind+"/"+match.WorkloadName)
		}
		slices.Sort(locations)
		return nil, "", fmt.Errorf("%w: service %s matched %s", errExactNameWorkloadAmbiguous, service, strings.Join(locations, ", "))
	}
	match := matches[0]
	return &match, match.Namespace, nil
}

type starterLogsTarget struct {
	Namespace string
	Kind      string
	Name      string
}

func starterRemoteLogsTarget(service string) (starterLogsTarget, bool) {
	service = strings.TrimSpace(service)
	rollouts := append(starterRemoteWorkloadRollouts(nil), starterRemoteAIFTSWorkloadRollouts()...)
	for _, rollout := range rollouts {
		kind, name, ok := strings.Cut(rollout.resource, "/")
		if !ok {
			continue
		}
		serviceName := name
		if name == "scaler" {
			serviceName = "tidb-worker-scaler"
		}
		if service != serviceName {
			continue
		}
		return starterLogsTarget{
			Namespace: rollout.namespace,
			Kind:      kind,
			Name:      name,
		}, true
	}
	return starterLogsTarget{}, false
}

type podLogContainer struct {
	Name string `json:"name"`
}

type podLogContainerStatus struct {
	Name         string `json:"name"`
	RestartCount int    `json:"restartCount"`
}

type podLogSpec struct {
	Containers []podLogContainer `json:"containers"`
}

type podLogStatus struct {
	ContainerStatuses []podLogContainerStatus `json:"containerStatuses"`
}

type podLogMetadata struct {
	Spec   podLogSpec   `json:"spec"`
	Status podLogStatus `json:"status"`
}

func collectKubernetesWorkloadLogs(
	ctx context.Context,
	selected *host.Session,
	substrate string,
	service string,
	namespace string,
	artifact *cluster.ServiceArtifact,
	query logsQuery,
) ([]logEntry, logSourceAttempt) {
	attempt := logSourceAttempt{
		Source: "kubernetes", Substrate: substrate, Context: selected.Context,
		Namespace: namespace, Workload: artifact.WorkloadKind + "/" + artifact.WorkloadName, Status: "failed",
	}
	pods, err := podNamesForArtifact(ctx, selected, namespace, artifact)
	if err != nil {
		attempt.Message = err.Error()
		return nil, attempt
	}
	var entries []logEntry
	var failures []string
	for _, pod := range pods {
		metadata := podLogMetadata{}
		output, metaErr := kubectlOutput(ctx, selected, "-n", namespace, "get", "pod", pod, "-o", "json")
		if metaErr == nil {
			metaErr = json.Unmarshal(output, &metadata)
		}
		containers := []string{}
		if metaErr == nil {
			for _, container := range metadata.Spec.Containers {
				containers = append(containers, container.Name)
			}
		} else if artifact.ContainerName != "" {
			containers = append(containers, artifact.ContainerName)
		}
		if len(containers) == 0 {
			failures = append(failures, pod+": container metadata unavailable")
			continue
		}
		restarts := map[string]int{}
		for _, status := range metadata.Status.ContainerStatuses {
			restarts[status.Name] = status.RestartCount
		}
		for _, container := range containers {
			streams := []string{"current"}
			if restarts[container] > 0 {
				streams = append(streams, "previous")
			}
			for _, stream := range streams {
				args := []string{"-n", namespace, "logs", pod, "--tail", strconv.Itoa(logsLimit(query)), "-c", container}
				if query.Since != "" {
					args = append(args, "--since", query.Since)
				}
				if stream == "previous" {
					args = append(args, "--previous")
				}
				output, logErr := kubectlOutput(ctx, selected, args...)
				if logErr != nil {
					failures = append(failures, pod+"/"+container+"/"+stream+": "+logErr.Error())
					continue
				}
				clusterName := selected.ClusterName
				if substrate == "k3d" {
					clusterName = "host"
				}
				for _, line := range filterLogLines(string(output), query.TraceID, query.Level) {
					entries = append(entries, logEntry{
						Cluster: clusterName, Substrate: substrate, Context: selected.Context,
						Namespace: namespace, Service: service, Pod: pod, Container: container, Stream: stream,
						Line: redactLogLine(line),
					})
				}
			}
		}
	}
	attempt.Status = "queried"
	if len(failures) > 0 {
		attempt.Message = strings.Join(failures, "; ")
		if len(entries) == 0 {
			attempt.Status = "failed"
		}
	}
	return entries, attempt
}

func appendSource(sources []string, source string) []string {
	if source == "" || slices.Contains(sources, source) {
		return sources
	}
	return append(sources, source)
}

func boundedLogEntries(entries []logEntry, limit int) []logEntry {
	limit = logsLimit(logsQuery{Limit: limit})
	if len(entries) <= limit {
		return entries
	}
	return append([]logEntry(nil), entries[len(entries)-limit:]...)
}

var (
	awsAccessKeyLogPattern        = regexp.MustCompile(`\b(?:AKIA|ASIA)[A-Z0-9]{16}\b`)
	sensitiveLogAssignmentPattern = regexp.MustCompile(
		`(?i)((?:["']?[A-Z0-9_.-]*(?:PASSWORD|PASSWD|TOKEN|SECRET|CREDENTIAL|PRIVATE[_-]?KEY|API[_-]?KEY)[A-Z0-9_.-]*["']?)\s*[=:]\s*)(?:"[^"]*"|'[^']*'|[^\s,;}]+)`,
	)
	sensitiveLogHeaderPattern = regexp.MustCompile(`(?i)(\b(?:authorization|proxy-authorization|cookie|set-cookie|x-api-key)\s*[:=]\s*).*$`)
	jwtLogPattern             = regexp.MustCompile(`\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b`)
	knownTokenLogPattern      = regexp.MustCompile(`\b(?:gh[opusr]_[A-Za-z0-9]{20,}|sk_(?:live|test)_[A-Za-z0-9]{16,})\b`)
)

func redactLogLine(line string) string {
	line = awsAccessKeyLogPattern.ReplaceAllString(line, "[REDACTED]")
	line = sensitiveLogAssignmentPattern.ReplaceAllString(line, `${1}[REDACTED]`)
	line = sensitiveLogHeaderPattern.ReplaceAllString(line, `${1}[REDACTED]`)
	line = jwtLogPattern.ReplaceAllString(line, "[REDACTED]")
	return knownTokenLogPattern.ReplaceAllString(line, "[REDACTED]")
}

// v1alpha2Day2TiltResource resolves a day-2 service to its Tilt resource name.
// cp9 persists no per-service status, so the Session Tiltfile's own naming is
// the only source and the resource is the service name.
func v1alpha2Day2TiltResource(runtime *v1alpha2TiltRuntime, name string) (string, error) {
	if runtime.Session != nil && config.UsesSessionTiltfile(runtime.Session) {
		name = strings.TrimSpace(name)
		if name == "" {
			return "", fmt.Errorf("service name is required")
		}
		return name, nil
	}
	return "", fmt.Errorf("service %s is not resolvable without a Session Tiltfile", name)
}

func finalizeLogsResult(result *logsResult) *logsResult {
	if result == nil {
		return nil
	}
	if result.Limit <= 0 {
		result.Limit = 200
	}
	if len(result.Entries) == 0 {
		result.Status = "empty"
		result.EmptyReason = result.emptyMessage()
		return result
	}
	result.Status = "matched"
	return result
}

func boundedLogLines(lines []string, limit int) []string {
	limit = logsLimit(logsQuery{Limit: limit})
	if len(lines) <= limit {
		return lines
	}
	return append([]string(nil), lines[len(lines)-limit:]...)
}

func loadServicePlan(stateDir, service string) (*persona.ServicePlan, error) {
	plan, err := persona.LoadPlan(cluster.PlanPath(stateDir))
	if err != nil {
		return nil, fmt.Errorf("load plan: %w", err)
	}
	for i := range plan.Services {
		svc := &plan.Services[i]
		if svc.Name == service {
			if svc.Role == persona.RoleSkip {
				return nil, fmt.Errorf("service %s is skip in the active service plan", service)
			}
			return svc, nil
		}
	}
	return nil, fmt.Errorf("service %s not found in active service plan", service)
}

func artifactTargets(artifacts *cluster.ArtifactState, requested string) ([]string, error) {
	if requested != "" {
		if artifacts == nil || artifacts.Services[requested] == nil {
			return nil, fmt.Errorf("service %s not found in artifacts", requested)
		}
		return []string{requested}, nil
	}

	if artifacts == nil {
		return nil, fmt.Errorf("artifact state is empty")
	}
	var names []string
	for name := range artifacts.Services {
		names = append(names, name)
	}
	slices.Sort(names)
	return names, nil
}

type podMetadata struct {
	Name string `json:"name"`
}

type podItem struct {
	Metadata podMetadata `json:"metadata"`
}

type podList struct {
	Items []podItem `json:"items"`
}

type workloadSelector struct {
	Spec workloadSelectorSpec `json:"spec"`
}

type workloadSelectorSpec struct {
	Selector workloadLabelSelector `json:"selector"`
}

type workloadLabelSelector struct {
	MatchLabels map[string]string `json:"matchLabels"`
}

func artifactHasWorkloadMetadata(artifact *cluster.ServiceArtifact) bool {
	return strings.TrimSpace(artifact.WorkloadKind) != "" && strings.TrimSpace(artifact.WorkloadName) != ""
}

func podNamesForArtifact(ctx context.Context, selected *host.Session, namespace string, artifact *cluster.ServiceArtifact) ([]string, error) {
	if !artifactHasWorkloadMetadata(artifact) {
		return nil, errWorkloadMetadataAbsent
	}

	selector, err := workloadLabelSelectorForArtifact(ctx, selected, namespace, artifact)
	if err == nil && selector != "" {
		output, err := kubectlOutput(ctx, selected, "-n", namespace, "get", "pods", "-l", selector, "-o", "json")
		if err != nil {
			return nil, err
		}
		return podNamesFromList(output, "")
	}

	output, err := kubectlOutput(ctx, selected, "-n", namespace, "get", "pods", "-o", "json")
	if err != nil {
		return nil, err
	}
	return podNamesFromList(output, artifact.WorkloadName)
}

func workloadLabelSelectorForArtifact(
	ctx context.Context,
	selected *host.Session,
	namespace string,
	artifact *cluster.ServiceArtifact,
) (string, error) {
	kind := strings.ToLower(strings.TrimSpace(artifact.WorkloadKind))
	name := strings.TrimSpace(artifact.WorkloadName)
	if kind == "" || name == "" {
		return "", errWorkloadMetadataAbsent
	}

	output, err := kubectlOutput(ctx, selected, "-n", namespace, "get", kind, name, "-o", "json")
	if err != nil {
		return "", err
	}

	var selector workloadSelector
	if err := json.Unmarshal(output, &selector); err != nil {
		return "", fmt.Errorf("parse %s %s selector: %w", kind, name, err)
	}
	if len(selector.Spec.Selector.MatchLabels) == 0 {
		return "", fmt.Errorf("%s %s selector is empty", kind, name)
	}

	keys := make([]string, 0, len(selector.Spec.Selector.MatchLabels))
	for key := range selector.Spec.Selector.MatchLabels {
		keys = append(keys, key)
	}
	slices.Sort(keys)

	parts := make([]string, 0, len(keys))
	for _, key := range keys {
		parts = append(parts, fmt.Sprintf("%s=%s", key, selector.Spec.Selector.MatchLabels[key]))
	}
	return strings.Join(parts, ","), nil
}

func podNamesFromList(output []byte, workloadName string) ([]string, error) {
	var list podList
	if err := json.Unmarshal(output, &list); err != nil {
		return nil, fmt.Errorf("parse pod list: %w", err)
	}

	var pods []string
	for _, item := range list.Items {
		name := item.Metadata.Name
		if workloadName == "" || name == workloadName || strings.HasPrefix(name, workloadName+"-") {
			pods = append(pods, name)
		}
	}
	slices.Sort(pods)
	if len(pods) == 0 {
		return nil, fmt.Errorf("%w %s", errNoPodsForWorkload, workloadName)
	}
	return pods, nil
}

func filterLogLines(raw, traceID, level string) []string {
	var lines []string
	traceID = strings.TrimSpace(traceID)
	level = strings.ToLower(strings.TrimSpace(level))
	for line := range strings.SplitSeq(raw, "\n") {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		if traceID != "" && !strings.Contains(line, traceID) {
			continue
		}
		if level != "" && !strings.Contains(strings.ToLower(line), level) {
			continue
		}
		lines = append(lines, line)
	}
	return lines
}

func firstNonEmptyString(values ...string) string {
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value != "" {
			return value
		}
	}
	return ""
}

func kubectlOutput(ctx context.Context, selected *host.Session, args ...string) ([]byte, error) {
	return host.RunKubectl(ctx, selected, nil, args...)
}

// sessionSetsSourceBuildParallelism reports whether the operator wrote
// services.source_build_parallelism themselves, as opposed to inheriting the
// default that config loading fills in.
func sessionSetsSourceBuildParallelism(cfgPath string) bool {
	raw, err := os.ReadFile(cfgPath)
	if err != nil {
		return false
	}
	var declared declaredSessionServices
	if err := yaml.Unmarshal(raw, &declared); err != nil {
		return false
	}
	return declared.Services.SourceBuildParallelism != nil
}

// declaredSessionServices reads only what has to be told apart from its own
// default: a pointer is nil when the key is absent and set when it is present.
type declaredSessionServices struct {
	Services declaredSourceBuildParallelism `yaml:"services"`
}

type declaredSourceBuildParallelism struct {
	SourceBuildParallelism *int `yaml:"source_build_parallelism"`
}