//go:build functional

package spectests_test

import (
	"bufio"
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"
)

const (
	remoteSourceEnabled    = "CP9_REMOTE_SOURCE_DEVELOPMENT_ENABLED"
	remoteSourceResume     = "CP9_REMOTE_SOURCE_DEVELOPMENT_RESUME"
	remoteSourceRunDir     = "CP9_REMOTE_SOURCE_DEVELOPMENT_RUN_DIR"
	remoteSourceDestroy    = "CP9_REMOTE_SOURCE_DEVELOPMENT_DESTROY_SESSION"
	remoteSourceCase       = "remote-source-development"
	remoteSourceSSHHelper  = "CP9_REMOTE_SOURCE_SSH_HELPER"
	remoteSourceSSHReal    = "CP9_REMOTE_SOURCE_SSH_REAL"
	remoteSourceSSHMode    = "CP9_REMOTE_SOURCE_SSH_MODE"
	remoteSourceSSHTimings = "CP9_REMOTE_SOURCE_SSH_TIMINGS"
)

type remoteSourceRun struct {
	t                *testing.T
	originRoot       string
	root             string
	metadata         remoteSourceMetadata
	cp9Root          string
	config           string
	sourceRoot       string
	sourceFile       string
	session          string
	remoteSourceRoot string
	currentStage     string
	currentAttempt   string
	commandID        int
	commandsMu       sync.Mutex
	commandEnv       map[string]string
	stages           map[string]*remoteSourceStage
	failureKind      string
	cleanup          string
	cleanupMessage   string
	prepared         bool
	prepareAttempted bool
	allPassed        bool
	postflightFailed bool
	processes        []*remoteSourceProcess
}

type remoteSourceMetadata struct {
	Account   string `json:"account"`
	Region    string `json:"region"`
	Profile   string `json:"profile"`
	Session   string `json:"session"`
	Owner     string `json:"owner"`
	TTL       string `json:"ttl"`
	OriginCP9 string `json:"origin_cp9"`
	CP9Root   string `json:"cp9_root"`
	Config    string `json:"config"`
	WorkRoot  string `json:"work_root"`
	Head      string `json:"head"`
}

type remoteSourceStage struct {
	Name       string                       `json:"name"`
	Product    string                       `json:"product"`
	Harness    string                       `json:"harness"`
	Executed   bool                         `json:"executed"`
	Reason     string                       `json:"reason,omitempty"`
	Attempts   []remoteSourceStageAttempt   `json:"attempts,omitempty"`
	Checkpoint []remoteSourceCheckpointNote `json:"checkpoints,omitempty"`
}

type remoteSourceStageAttempt struct {
	Head       string `json:"head"`
	StartedAt  string `json:"started_at"`
	FinishedAt string `json:"finished_at"`
	Product    string `json:"product"`
	Harness    string `json:"harness"`
	Evidence   string `json:"evidence"`
}

type remoteSourceCheckpointNote struct {
	Name       string `json:"name"`
	StartedAt  string `json:"started_at"`
	FinishedAt string `json:"finished_at"`
	Status     string `json:"status"`
}

type remoteSourceCommandReceipt struct {
	Command    []string `json:"command"`
	StartedAt  string   `json:"started_at"`
	FinishedAt string   `json:"finished_at"`
	DurationMS int64    `json:"duration_ms"`
	ExitCode   int      `json:"exit_code"`
	Stdout     string   `json:"stdout"`
	Stderr     string   `json:"stderr"`
	Error      string   `json:"error,omitempty"`
}

type remoteSourceCommandResult struct {
	ExitCode int
	Stdout   string
	Stderr   string
	Duration time.Duration
}

type remoteSourceSavedAction struct {
	Status     string   `json:"status"`
	Head       string   `json:"head"`
	Directory  string   `json:"directory"`
	Command    []string `json:"command"`
	ExitCode   int      `json:"exit_code"`
	Stdout     string   `json:"stdout"`
	Stderr     string   `json:"stderr"`
	DurationNS int64    `json:"duration_ns"`
}

type remoteSourceProcess struct {
	cmd        *exec.Cmd
	ctx        context.Context
	cancel     context.CancelFunc
	stdout     *os.File
	stderr     *os.File
	stdoutPath string
	stderrPath string
	receipt    string
	argv       []string
	started    time.Time
	waited     bool
}

type remoteSourceStatus struct {
	SessionName string `json:"session_name"`
	Status      string `json:"status"`
	StateDir    string `json:"state_dir"`
	Layers      struct {
		Host struct {
			Execution struct {
				SourceSyncDestination string `json:"source_sync_destination"`
				Remote                struct {
					SSHTransport string `json:"ssh_transport"`
				} `json:"remote"`
			} `json:"execution"`
		} `json:"host"`
		Tilt struct {
			Controller        string               `json:"controller"`
			Running           bool                 `json:"running"`
			Status            string               `json:"status"`
			ReadyResources    int                  `json:"ready_resources"`
			TotalResources    int                  `json:"total_resources"`
			NotReadyResources []string             `json:"not_ready_resources"`
			FailedResources   []string             `json:"failed_resources"`
			Retryable         bool                 `json:"retryable"`
			ReadinessError    string               `json:"readiness_error"`
			TunnelProcesses   []remoteSourceTunnel `json:"tunnels"`
		} `json:"tilt"`
	} `json:"layers"`
}

type remoteSourceTunnel struct {
	Role           string `json:"role"`
	PID            int    `json:"pid"`
	State          string `json:"state"`
	ActionRequired bool   `json:"action_required"`
}

type remoteSourceTiltResource struct {
	Metadata struct {
		Name            string            `json:"name"`
		ResourceVersion string            `json:"resourceVersion"`
		Annotations     map[string]string `json:"annotations"`
	} `json:"metadata"`
	Status struct {
		TriggerMode  int    `json:"triggerMode"`
		UpdateStatus string `json:"updateStatus"`
		BuildHistory []struct {
			SpanID     string `json:"spanID"`
			StartTime  string `json:"startTime"`
			FinishTime string `json:"finishTime"`
		} `json:"buildHistory"`
	} `json:"status"`
}

type remoteSourcePodList struct {
	Items []struct {
		Metadata struct {
			Name string `json:"name"`
			UID  string `json:"uid"`
		} `json:"metadata"`
		Status struct {
			Phase      string `json:"phase"`
			Conditions []struct {
				Type   string `json:"type"`
				Status string `json:"status"`
			} `json:"conditions"`
			ContainerStatuses []struct {
				Name    string `json:"name"`
				ImageID string `json:"imageID"`
				Ready   bool   `json:"ready"`
			} `json:"containerStatuses"`
		} `json:"status"`
	} `json:"items"`
}

type remoteSourceWatchEvent struct {
	State struct {
		Status         string `json:"status"`
		TiltStatus     string `json:"tilt_status"`
		ProbeRetryable bool   `json:"probe_retryable"`
		ProbeError     string `json:"probe_error"`
	} `json:"state"`
	Action struct {
		Code string `json:"code"`
	} `json:"action"`
}

type remoteSourceProbeTiming struct {
	Actual  time.Duration `json:"actual"`
	Padding time.Duration `json:"padding"`
	Total   time.Duration `json:"total"`
}

func TestMain(m *testing.M) {
	if os.Getenv(remoteSourceSSHHelper) == "1" {
		os.Exit(runRemoteSourceSSHHelper())
	}
	os.Exit(m.Run())
}

// TestFunctional_RemoteSourceDevelopment proves refresh, targeted deploy, and
// transport-loss recovery against one real remote source-development session.
// It skips before build, AWS reads, or mutation unless explicitly enabled.
// # Remote source development survives changes and network loss
//
// This case proves one complete user journey in one Remote session: a developer
// starts CP9 from a source checkout, refreshes changed inputs, deploys the current
// source into the running Pod, loses the local SSH transport during a later source
// transfer, resumes the same remote controller without rebuilding the session,
// and keeps status usable when bounded remote probes are slow or time out.
//
// It covers issues [#977](https://github.com/tidbcloud/controlplane9/issues/977),
// [#995](https://github.com/tidbcloud/controlplane9/issues/995),
// [#996](https://github.com/tidbcloud/controlplane9/issues/996),
// [#855](https://github.com/tidbcloud/controlplane9/issues/855),
// [#910](https://github.com/tidbcloud/controlplane9/issues/910),
// [#1004](https://github.com/tidbcloud/controlplane9/issues/1004),
// [#1208](https://github.com/tidbcloud/controlplane9/issues/1208), and
// [#1091](https://github.com/tidbcloud/controlplane9/issues/1091).
//
// ## Agent execution
//
// **Mode: single agent.** Run every phase in order against the same checkout and
// session. Create the session once and clean it once. Spawn no sub-agents because
// source edits, fault injection, background processes, and cleanup share mutable
// state.
//
// Do not fast-fail after a phase result. Before phases 2 and 3, record a cheap
// status check proving the original session is still Ready and the source checkout
// is clean. Also require the remote copies of the tracked source probes to match
// and all phase-owned probe files to be absent. Run the phase when those checks pass.
// Otherwise record that phase and every later unsafe phase as `BLOCKED`, Product
// `NOT RUN`, with the failed health evidence, then perform the one cleanup.
//
// Native Go entrypoint (replace the key values with the existing caller-owned
// pair for the selected account):
//
// ```text
// CP9_REMOTE_SOURCE_DEVELOPMENT_ENABLED=1 \
// CP9_UI_API_AWS_KEY_PAIR=<existing-key-pair> \
// CP9_UI_API_SSH_PRIVATE_KEY=</absolute/path/to/private-key> \
// GOTOOLCHAIN=go1.26.1 GO_FUNCTIONAL_TEST_PACKAGES=./spec-tests \
// GO_FUNCTIONAL_TEST_RUN='^TestFunctional_RemoteSourceDevelopment$$' \
// make test-functional
// ```
//
// The test records its resumable run path in the test log and under
// `artifacts/spec-tests/`. To continue the same preserved scene, set
// `CP9_REMOTE_SOURCE_DEVELOPMENT_RESUME=1` and
// `CP9_REMOTE_SOURCE_DEVELOPMENT_RUN_DIR=<absolute-run-path>`, keep the exact
// same checkout and key inputs, and select the next safe phase with
// `GO_FUNCTIONAL_TEST_RUN`. Passed phases are reused. A failed phase keeps its
// original attempt evidence and may be retried in the same scene after diagnosis;
// completed action receipts are reused and each retry appends a new attempt
// instead of rewriting the first red. A partial network-loss phase fails closed
// because its owned background process cannot be reconstructed safely. An action
// left `started` also fails closed until its outcome is reconciled; it is never
// replayed automatically.
//
// ## Defaults and authority
//
// - Test the exact checked-out CP9 commit.
// - Use AWS account `804837308162`, region `us-west-2`, source profile
//   `dev-offline-afterworld`, role `cp9-developer`, and one fresh 10-hour
//   `remote-ec2-k3d` session.
// - Provide the existing caller-owned EC2 key pair and matching private key via
//   `CP9_UI_API_AWS_KEY_PAIR` and `CP9_UI_API_SSH_PRIVATE_KEY`; CP9 validates but
//   never creates or deletes them.
// - Use one clean shallow `tidbcloud/config-center` checkout. Its copied Dev Full
//   Tiltfile source-builds only `config-center`.
// - The instruction to execute this case authorizes the session, caller-owned
//   existing key pair/private key inputs, reversible source probes, scoped
//   transport fault, phase-owned local SSH delay/stall wrapper, targeted deploys,
//   and exact cleanup. GitHub Device Flow is the only possible human step. CP9
//   never creates or deletes the key pair.
//
// No TiDB cluster or product-line resource is created.
//
// ## Execute
//
// ### 1. Create the baseline once
//
// Freeze CP9 and `config-center`, build CP9, and verify the declared account,
// region, source, session, owner, and TTL with dry-run. Run setup, doctor, and the
// initial `cp9 up` once. Require the generated session to be `ready`, record its
// ID, and require that same ID through every later phase. Status must report one
// healthy local supervisor for each remote tunnel role (`controller` and
// `kube-api`), with distinct positive PIDs and no operator action required.
//
// If preparation fails before this baseline becomes usable, do not run any phase.
// Record all three phases as unexecuted with Product and Harness `NOT RUN` and a
// prerequisite reason, attempt cleanup for any run-owned state, emit the run
// summary with all three phases listed as unexecuted, and return a nonzero result.
//
// ### 2. Refresh changed inputs with the same `cp9 up`
//
// Make one reversible tracked change and one untracked probe in `config-center`.
// Run `cp9 up` again and require both changes in the source tree used by the
// remote controller. Restore both inputs, run `cp9 up` again, and require the
// tracked content restored and the untracked file absent. An optional manual
// deploy helper may remain idle; it must not block the session from becoming
// `ready`. After each repeated `cp9 up`, require exactly one healthy supervisor
// per tunnel role; a duplicate, stopped, or action-required supervisor is
// `FAIL`.
//
// ### 3. Deploy the current source into the running Pod
//
// Record the current `config-center` binary, Pod UID, and container image ID. Add
// a unique startup marker, clear only the current `config-center-image` UIResource
// stamp to reproduce first-deploy initialization, and run
// `cp9 deploy service=config-center`. Success requires the same command to publish
// a non-empty stamp, install a new running binary, and expose the marker in service
// logs before it returns, while the Pod UID and container image ID remain unchanged
// because this edit changes no image fallback input.
//
// Keep the source marker, add one unique comment to the service Dockerfile, and
// deploy again. Require a fresh image-resource build, a new Pod UID and container
// image ID, and the source marker from that exact new Pod's logs. A
// live-binary-only update, historical marker, or stale image is `FAIL`.
//
// Break that same source and deploy again. Require a nonzero result naming the
// failed resource while the last good Pod binary remains unchanged. Restore the
// source and Dockerfile, deploy again, and require the original binary, the exact
// remote Dockerfile content, and a clean checkout.
//
// Deploy `billing-svc` through its direct image-build path. Require a new build
// invocation and return success only after that invocation completes; a stale
// pre-trigger success is `FAIL`.
//
// ### 4. Resume after one scoped network loss
//
// Add a reversible transfer probe large enough to observe. Start `cp9 up`, wait
// until its source transfer is active, and interrupt only this case's SSH
// transport. Status must keep the same session ID and report the remote controller
// as `unreachable`, not stopped or failed.
//
// Restore transport. Let the original command continue or rerun the same command
// against the same session, then detach the local process without stopping the
// remote controller. Status watch and logs must remain usable. Run `cp9 up` again,
// require the original session to return `ready` with one remote controller, and
// require one healthy local supervisor per tunnel role and the complete transfer
// probe on EC2. A targeted `config-center` deploy must succeed from that complete
// tree.
//
// Remove the probe, run `cp9 up` once more, and require it absent remotely. If the
// persisted transport is SSM, run doctor and status inside their existing bounds;
// otherwise record that SSM-only assertion as `NOT RUN`.
//
// Use the phase-owned SSH wrapper to preserve each real controller-observation and
// resource-readiness result while padding each healthy command to at least 16
// seconds. Their aggregate duration must exceed one 30-second command budget while
// each remains below that budget. One `cp9 status` must still report the same
// session `ready` with the real Tilt resource counts.
//
// Next, stall only the resource-readiness command beyond its 30-second deadline
// and start `cp9 status --watch`. Its first bounded observation must report the
// session as `initializing`, Tilt as `unavailable`, the probe as retryable, the
// next action as `wait`, and name both `remote Tilt resource readiness` and the
// effective `30s` deadline. Remove the stall without restarting CP9 or the remote
// controller; the same watcher must later report `ready` rather than exit with a
// terminal failure. Finally, stall only remote artifact synchronization, run one
// bounded status observation, and require the same retryable `initializing`
// contract with `remote Tilt artifact sync` and `30s` in the error. These probes
// must not change remote state or rebuild the session.
//
// ### 5. Clean up once
//
// Require a clean source checkout and the same Ready session. Destroy only that
// session and generated worktree, leave the caller-owned key pair untouched, and
// record the exact cleanup receipt and durable accepted hosted-cleanup handoff.
// After destroy, require no process command to reference either exact-session
// tunnel supervisor. Do not wait for the asynchronous hosted cleanup run.
//
// The Go test performs this cleanup after all three phases pass. On first red it
// preserves the scene and reports Cleanup `FAIL` with the pending obligation. After reviewer disposition,
// set `CP9_REMOTE_SOURCE_DEVELOPMENT_DESTROY_SESSION=1` on the exact resume run to
// perform the same cleanup without treating preservation as success. The
// caller-owned key pair remains outside cleanup.
//
// ## Verdicts
//
// ### Phase 1: input refresh
//
// #### Product
//
// `PASS` requires the tracked and untracked changes to appear remotely after the
// second `cp9 up`, then return exactly to the original state after restoration.
// Each status snapshot must also report exactly one healthy `controller` and one
// healthy `kube-api` tunnel supervisor. Stale or partial remote inputs, duplicate
// supervisors, or an action-required tunnel are `FAIL`. If the initial session
// never becomes usable, this phase is `NOT RUN` unless direct product evidence
// already proves a failure.
//
// #### Harness
//
// `PASS` requires the source locators, hashes, SSH observation, and restoration
// evidence in `01-input-refresh/` to complete. A broken locator, comparison, or
// evidence write is Harness `FAIL`; unobserved Product assertions remain
// `NOT RUN`.
//
// ### Phase 2: deploy current source
//
// #### Product
//
// `PASS` requires the valid source change to reach the existing Pod before deploy
// returns, the Dockerfile change to build a new image and roll the Pod, the
// invalid source deploy to return nonzero without replacing the last good binary,
// restoration to return the original binary and Dockerfile, and direct
// `billing-svc` deploy to wait for a new build invocation. A stale success,
// unnecessary image rollout, skipped fallback rebuild, or wrong running artifact
// is `FAIL`. A failed pre-phase health check is `BLOCKED` and Product `NOT RUN`,
// not `FAIL`.
//
// #### Harness
//
// `PASS` requires the Pod and image identities, Tilt resources, log marker, build
// order, binary and source hashes, and restoration evidence in `02-deploy/` to
// complete. The source-only trace must show the manual image-resource check
// finishing with the same published image stamp before the live-binary build
// starts. A probe or parser defect is Harness `FAIL` and cannot manufacture a
// Product result.
//
// ### Phase 3: network resume
//
// #### Product
//
// `PASS` requires the original session to become unreachable during the scoped
// fault, recover without destroy/recreate, keep one remote controller, receive the
// complete source tree, restore exactly one healthy supervisor for each local
// tunnel role, accept a later deploy, keep two individually bounded healthy status
// probes correct when their aggregate duration exceeds 30 seconds, recover in the
// same watcher after a retryable readiness timeout, and classify an artifact-sync
// timeout as retryable. A changed session, false terminal failure, missing
// operation/deadline, stopped or duplicate controller or tunnel supervisor,
// partial source tree, action-required tunnel, or destroy/recreate recovery is
// `FAIL`. A failed pre-phase health check is `BLOCKED` and Product `NOT RUN`.
//
// #### Harness
//
// `PASS` requires bounded fault injection, process cleanup, transfer hashes,
// status/watch/log observations, measured per-probe timings, delay padding,
// retryable timeout events, and evidence in `03-network-resume/` to complete. A
// transport wrapper, locator, bounded-wait, timing capture, or evidence-capture
// defect is Harness `FAIL`.
//
// ### Run summary
//
// List executed and unexecuted phases. Product `PASS` requires all three phases to
// have executed and passed against one unchanged session ID. Any Product `FAIL`
// makes the summary `FAIL`; otherwise any `BLOCKED` or `NOT RUN` phase makes it
// `NOT RUN`/partial, never `PASS`.
//
// An expired login, missing tool, inaccessible source repository, or unavailable
// Device Flow human is product `NOT RUN` only when it prevents the affected phase
// from starting. Preserve direct results from completed phases.
//
// ### Cleanup
//
// Cleanup `PASS` requires every source and fault probe restored, background
// processes stopped, both exact-session tunnel supervisors absent, the exact
// session removed, and the required receipts recorded. The caller-owned key pair
// is not a run-owned resource and must remain untouched. Known residue or a
// missing receipt is cleanup `FAIL`, not product `FAIL`. Cleanup is `NOT RUN` only
// when the case created no run-owned state.

func TestFunctional_RemoteSourceDevelopment(t *testing.T) {
	if os.Getenv(remoteSourceEnabled) != "1" {
		t.Skipf("set %s=1 to run the remote source-development journey", remoteSourceEnabled)
	}

	run := newRemoteSourceRun(t)
	t.Cleanup(func() { run.finalize(t) })

	run.runStage(t, "input-refresh", run.inputRefresh)
	run.runStage(t, "deploy-current-source", run.deployCurrentSource)
	run.runStage(t, "network-resume", run.networkResume)
	run.allPassed = run.stages["input-refresh"].Product == "pass" && run.stages["input-refresh"].Harness == "pass" &&
		run.stages["deploy-current-source"].Product == "pass" && run.stages["deploy-current-source"].Harness == "pass" &&
		run.stages["network-resume"].Product == "pass" && run.stages["network-resume"].Harness == "pass"
}

func newRemoteSourceRun(t *testing.T) *remoteSourceRun {
	t.Helper()
	originRoot := repositoryRoot(t)
	resume := enabled(os.Getenv(remoteSourceResume))
	root := strings.TrimSpace(os.Getenv(remoteSourceRunDir))
	if resume && root == "" {
		t.Fatalf("%s=1 requires %s", remoteSourceResume, remoteSourceRunDir)
	}
	if root == "" {
		root = filepath.Join(originRoot, "artifacts", "spec-tests", time.Now().UTC().Format("20060102T150405Z")+"-remote-source-development")
	}
	root, err := filepath.Abs(root)
	if err != nil {
		t.Fatalf("resolve run directory: %v", err)
	}
	if err := os.MkdirAll(root, 0o700); err != nil {
		t.Fatalf("create run directory: %v", err)
	}
	r := &remoteSourceRun{
		t:          t,
		originRoot: originRoot,
		root:       root,
		commandEnv: map[string]string{},
		cleanup:    "not_run",
		stages: map[string]*remoteSourceStage{
			"input-refresh":         {Name: "input-refresh", Product: "not_run", Harness: "not_run"},
			"deploy-current-source": {Name: "deploy-current-source", Product: "not_run", Harness: "not_run"},
			"network-resume":        {Name: "network-resume", Product: "not_run", Harness: "not_run"},
		},
	}
	if resume {
		r.loadState(t)
	}
	return r
}

func (r *remoteSourceRun) runStage(t *testing.T, name string, body func(*testing.T)) {
	t.Helper()
	stage := r.stages[name]
	stage.Product, stage.Harness = remoteSourceStageVerdicts(stage)
	ran := false
	reused := false
	bodyCompleted := false
	bodyFailureKind := ""
	started := time.Now()
	attemptDir := filepath.Join(r.root, fmt.Sprintf("%s-%s", time.Now().UTC().Format("20060102T150405.000000000Z"), name))
	passed := t.Run(name, func(t *testing.T) {
		ran = true
		r.currentStage = name
		r.currentAttempt = attemptDir
		if err := os.MkdirAll(attemptDir, 0o700); err != nil {
			r.failureKind = "harness"
			t.Fatalf("create stage evidence directory: %v", err)
		}
		if enabled(os.Getenv(remoteSourceResume)) && stage.Product == "pass" && stage.Harness == "pass" {
			reused = true
			t.Skip("reuse preserved passing stage")
		}
		if enabled(os.Getenv(remoteSourceResume)) && name == "network-resume" && len(stage.Checkpoint) > 0 {
			r.failureKind = "blocked"
			t.Fatal("partial network-resume cannot safely recreate the owned background process; preserve the scene for disposition")
		}
		if name != "input-refresh" {
			if !r.preflight(t, attemptDir, "preflight") {
				t.Fatal("the preserved session is not safe for this phase")
			}
		}
		stage.Executed = true
		r.failureKind = "product"
		r.postflightFailed = false
		defer func() {
			bodyFailureKind = r.failureKind
			if r.prepared && !r.preflight(t, attemptDir, "postflight") {
				r.postflightFailed = true
				t.Errorf("phase did not restore a healthy session and clean source tree")
			}
		}()
		body(t)
		bodyCompleted = true
	})
	if !ran {
		if stage.Product != "pass" || stage.Harness != "pass" {
			stage.Reason = "not selected by GO_FUNCTIONAL_TEST_RUN"
		}
		r.writeState(t)
		return
	}
	if reused {
		return
	}
	stage.Executed = true
	attemptProduct := "pass"
	attemptHarness := "pass"
	if !passed {
		kind := bodyFailureKind
		if kind == "" {
			kind = r.failureKind
		}
		if kind == "prerequisite" {
			attemptProduct = "not_run"
			attemptHarness = "not_run"
		} else if kind == "blocked" {
			attemptProduct = "not_run"
			attemptHarness = "pass"
		} else if kind == "harness" {
			attemptProduct = "not_run"
			attemptHarness = "fail"
		} else {
			attemptProduct = "fail"
			attemptHarness = "pass"
		}
		if r.postflightFailed {
			attemptHarness = "fail"
			if bodyCompleted {
				attemptProduct = "pass"
			}
		}
	}
	stage.Attempts = append(stage.Attempts, remoteSourceStageAttempt{
		Head:       r.currentHead(t),
		StartedAt:  started.UTC().Format(time.RFC3339Nano),
		FinishedAt: time.Now().UTC().Format(time.RFC3339Nano),
		Product:    attemptProduct,
		Harness:    attemptHarness,
		Evidence:   filepath.Base(attemptDir),
	})
	stage.Product = aggregateRemoteSourceLane(stage.Product, attemptProduct)
	stage.Harness = aggregateRemoteSourceLane(stage.Harness, attemptHarness)
	if stage.Product == "fail" || stage.Harness == "fail" {
		stage.Reason = "first red preserved; inspect the attempt history and latest repair evidence"
	}
	r.writeState(t)
}

func (r *remoteSourceRun) stageAttemptDir(stage string) string {
	if r.currentAttempt == "" || filepath.Base(r.currentAttempt) == "" || !strings.HasSuffix(filepath.Base(r.currentAttempt), "-"+stage) {
		panic("remote source stage attempt is not active")
	}
	return r.currentAttempt
}

func (r *remoteSourceRun) checkpoint(t *testing.T, stage, name string, body func(*testing.T)) {
	t.Helper()
	ran := false
	reused := false
	started := time.Now()
	passed := t.Run(name, func(t *testing.T) {
		ran = true
		if enabled(os.Getenv(remoteSourceResume)) && remoteSourceCheckpointPassed(r.stages[stage], name) {
			reused = true
			t.Skip("reuse preserved passing checkpoint")
		}
		body(t)
	})
	if reused {
		return
	}
	status := "pass"
	if !ran {
		status = "not_run"
	} else if !passed {
		status = "fail"
	}
	r.stages[stage].Checkpoint = append(r.stages[stage].Checkpoint, remoteSourceCheckpointNote{
		Name:       name,
		StartedAt:  started.UTC().Format(time.RFC3339Nano),
		FinishedAt: time.Now().UTC().Format(time.RFC3339Nano),
		Status:     status,
	})
	r.writeState(t)
	if !ran {
		r.failureKind = "harness"
		t.Fatalf("checkpoint %s was filtered; select the complete ordered phase", name)
	}
	if !passed {
		t.FailNow()
	}
}

func remoteSourceCheckpointPassed(stage *remoteSourceStage, name string) bool {
	for index := len(stage.Checkpoint) - 1; index >= 0; index-- {
		if stage.Checkpoint[index].Name == name {
			return stage.Checkpoint[index].Status == "pass"
		}
	}
	return false
}

func (r *remoteSourceRun) inputRefresh(t *testing.T) {
	stage := "input-refresh"
	var original []byte
	baselinePath := filepath.Join(r.root, "state", "input-source-baseline.go")
	refreshPath := filepath.Join(r.root, "state", "input-source-refresh.go")
	untracked := ""
	if r.sourceRoot != "" {
		untracked = filepath.Join(r.sourceRoot, "cp9-spec-untracked-probe.txt")
	}
	if fileExists(baselinePath) {
		original = []byte(readFile(t, baselinePath))
	}
	defer func() {
		if len(original) > 0 {
			_ = os.WriteFile(r.sourceFile, original, 0o644)
		}
		if untracked != "" {
			_ = os.Remove(untracked)
		}
	}()

	r.checkpoint(t, stage, "create-baseline", func(t *testing.T) {
		r.ensurePrepared(t)
		attempt := r.stageAttemptDir(stage)
		r.mustCommand(t, "harness", attempt, "build-cp9", 20*time.Minute, r.cp9Root, nil, "make", "build")
		dry := r.mustCommand(t, "harness", attempt, "up-dry-run", 2*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "up", "dry-run=true", "config="+r.config)
		var scope struct {
			NoMutation bool   `json:"no_mutation"`
			Session    string `json:"session"`
			Owner      string `json:"owner"`
			TTL        string `json:"ttl"`
			AWS        struct {
				Account string `json:"account"`
				Region  string `json:"region"`
			} `json:"aws"`
		}
		r.decodeResult(t, "harness", dry, &scope)
		if !scope.NoMutation || scope.Session != r.session || scope.Owner == "" || scope.TTL == "" || scope.AWS.Account != "804837308162" || scope.AWS.Region != "us-west-2" {
			r.harnessFatalf(t, "dry-run scope does not match the declared session/account/region")
		}
		r.mustAction(t, "product", attempt, "setup", 30*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "setup", "checkout=false", "config="+r.config)
		r.mustCommand(t, "product", attempt, "doctor", 10*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "doctor", "config="+r.config)
		r.mustAction(t, "product", attempt, "up-initial", 3*time.Hour, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "up", "config="+r.config)
		status := r.status(t, attempt, "status-initial")
		r.requireHealthyStatus(t, status)
		r.remoteSourceRoot = filepath.Join(status.Layers.Host.Execution.SourceSyncDestination, "controller", "work", "config-center")
		r.writeState(t)
	})

	r.checkpoint(t, stage, "refresh-tracked-and-untracked-inputs", func(t *testing.T) {
		attempt := r.stageAttemptDir(stage)
		if untracked == "" {
			untracked = filepath.Join(r.sourceRoot, "cp9-spec-untracked-probe.txt")
		}
		if len(original) == 0 {
			var err error
			original, err = os.ReadFile(r.sourceFile)
			if err != nil {
				r.harnessFatalf(t, "read source baseline: %v", err)
			}
			if err := os.MkdirAll(filepath.Dir(baselinePath), 0o700); err != nil {
				r.harnessFatalf(t, "create input state directory: %v", err)
			}
			if err := os.WriteFile(refreshPath, append(append([]byte(nil), original...), fmt.Appendf(nil, "\n// cp9-up-refresh-%d\n", time.Now().UnixNano())...), 0o600); err != nil {
				r.harnessFatalf(t, "persist tracked source probe: %v", err)
			}
			if err := os.WriteFile(baselinePath, original, 0o600); err != nil {
				r.harnessFatalf(t, "persist source baseline: %v", err)
			}
		}
		if _, err := os.Lstat(untracked); !os.IsNotExist(err) {
			r.harnessFatalf(t, "untracked probe target already exists: %s", untracked)
		}
		if err := os.WriteFile(r.sourceFile, []byte(readFile(t, refreshPath)), 0o644); err != nil {
			r.harnessFatalf(t, "write tracked source probe: %v", err)
		}
		if err := os.WriteFile(untracked, []byte("cp9 up refresh probe\n"), 0o600); err != nil {
			r.harnessFatalf(t, "write untracked source probe: %v", err)
		}
		wantHash := fileSHA256(t, r.sourceFile)
		r.mustAction(t, "product", attempt, "up-refresh", 30*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "up", "config="+r.config)
		r.requireHealthyStatus(t, r.status(t, attempt, "status-after-refresh"))
		remoteHash := strings.Fields(r.readCommand(t, "product", r.mustCommand(t, "product", attempt, "remote-tracked-hash", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "sha256sum", filepath.Join(r.remoteSourceRoot, strings.TrimPrefix(r.sourceFile, r.sourceRoot+string(filepath.Separator))))))
		if len(remoteHash) == 0 || remoteHash[0] != wantHash {
			r.productFatalf(t, "remote tracked input hash did not refresh")
		}
		probe := r.readCommand(t, "product", r.mustCommand(t, "product", attempt, "remote-untracked", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "cat", filepath.Join(r.remoteSourceRoot, filepath.Base(untracked))))
		if strings.TrimSpace(probe) != "cp9 up refresh probe" {
			r.productFatalf(t, "remote untracked input did not refresh")
		}
	})

	r.checkpoint(t, stage, "restore-inputs", func(t *testing.T) {
		attempt := r.stageAttemptDir(stage)
		if untracked == "" {
			untracked = filepath.Join(r.sourceRoot, "cp9-spec-untracked-probe.txt")
		}
		if err := os.WriteFile(r.sourceFile, original, 0o644); err != nil {
			r.harnessFatalf(t, "restore tracked source: %v", err)
		}
		if err := os.Remove(untracked); err != nil && !os.IsNotExist(err) {
			r.harnessFatalf(t, "remove untracked probe: %v", err)
		}
		r.mustAction(t, "product", attempt, "up-restore", 30*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "up", "config="+r.config)
		r.requireHealthyStatus(t, r.status(t, attempt, "status-after-restore"))
		remoteHash := strings.Fields(r.readCommand(t, "product", r.mustCommand(t, "product", attempt, "remote-restored-hash", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "sha256sum", filepath.Join(r.remoteSourceRoot, strings.TrimPrefix(r.sourceFile, r.sourceRoot+string(filepath.Separator))))))
		if len(remoteHash) == 0 || remoteHash[0] != fileSHA256(t, r.sourceFile) {
			r.productFatalf(t, "remote tracked input did not restore")
		}
		absent := r.command(t, attempt, "remote-untracked-absent", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "test", "!", "-e", filepath.Join(r.remoteSourceRoot, "cp9-spec-untracked-probe.txt"))
		if absent.ExitCode != 0 {
			r.productFatalf(t, "remote untracked input remains; see %s", absent.Stderr)
		}
		r.requireCleanSource(t)
	})
}

func (r *remoteSourceRun) deployCurrentSource(t *testing.T) {
	stage := "deploy-current-source"
	attempt := r.stageAttemptDir(stage)
	service := "config-center"
	probeDir := filepath.Join(r.sourceRoot, "cmd", "server")
	probeFile := filepath.Join(probeDir, "cp9_spec_probe.go")
	fallbackFile := filepath.Join(r.sourceRoot, "Dockerfile")
	fallback, err := os.ReadFile(fallbackFile)
	if err != nil {
		r.harnessFatalf(t, "read Dockerfile baseline: %v", err)
	}
	defer func() {
		_ = os.Remove(probeFile)
		_ = os.WriteFile(fallbackFile, fallback, 0o644)
	}()
	var podBefore remoteSourcePod
	var podAfter remoteSourcePod
	var podAfterFallback remoteSourcePod
	var binaryBefore string
	var binaryAfterFallback string
	var imageBefore remoteSourceTiltResource
	var imageAfter remoteSourceTiltResource
	var liveBefore remoteSourceTiltResource
	var billingBefore remoteSourceTiltResource
	var billingBeforeCaptured bool
	var marker string
	var tiltLogBefore string
	statePath := filepath.Join(r.root, "state", "deploy.json")
	fallbackBaseline := fallback
	if fileExists(statePath) {
		var state remoteSourceDeployState
		decodeFile(t, statePath, &state)
		if len(state.FallbackBaseline) == 0 {
			r.harnessFatalf(t, "saved deploy state has no Dockerfile baseline")
		}
		podBefore, podAfter, podAfterFallback = state.PodBefore, state.PodAfter, state.PodAfterFallback
		binaryBefore, binaryAfterFallback = state.BinaryBefore, state.BinaryAfterFallback
		imageBefore, imageAfter, liveBefore = state.ImageBefore, state.ImageAfter, state.LiveBefore
		billingBefore = state.BillingBefore
		billingBeforeCaptured = state.BillingBeforeCaptured
		fallbackBaseline = state.FallbackBaseline
		marker, tiltLogBefore = state.Marker, state.TiltLogBefore
	}
	fallback = fallbackBaseline
	saveState := func() {
		writeJSON(t, statePath, remoteSourceDeployState{
			PodBefore: podBefore, PodAfter: podAfter, PodAfterFallback: podAfterFallback,
			BinaryBefore: binaryBefore, BinaryAfterFallback: binaryAfterFallback,
			ImageBefore: imageBefore, ImageAfter: imageAfter, LiveBefore: liveBefore, BillingBefore: billingBefore,
			BillingBeforeCaptured: billingBeforeCaptured, FallbackBaseline: fallbackBaseline,
			Marker: marker, TiltLogBefore: tiltLogBefore,
		})
	}
	if enabled(os.Getenv(remoteSourceResume)) && remoteSourceCheckpointPassed(r.stages[stage], "capture-deploy-baseline") && !remoteSourceCheckpointPassed(r.stages[stage], "restore-source-image-path-and-health") {
		if marker == "" {
			r.harnessFatalf(t, "saved deploy baseline has no source marker")
		}
		probe := fmt.Sprintf("package main\n\nimport \"log\"\n\nfunc init() { log.Print(%q) }\n", marker)
		if remoteSourceCheckpointPassed(r.stages[stage], "failed-source-preserves-last-good-binary") {
			probe += "\nthis is not valid Go\n"
		}
		if err := os.WriteFile(probeFile, []byte(probe), 0o600); err != nil {
			r.harnessFatalf(t, "restore source probe for resume: %v", err)
		}
		resumedFallback := fallback
		if remoteSourceCheckpointPassed(r.stages[stage], "deploy-dockerfile-through-image-fallback") {
			resumedFallback = append(append([]byte(nil), fallback...), []byte("\n# "+marker+" image fallback\n")...)
		}
		if err := os.WriteFile(fallbackFile, resumedFallback, 0o644); err != nil {
			r.harnessFatalf(t, "restore Dockerfile probe for resume: %v", err)
		}
	}

	r.checkpoint(t, stage, "capture-deploy-baseline", func(t *testing.T) {
		if _, err := os.Lstat(probeFile); !os.IsNotExist(err) {
			r.harnessFatalf(t, "source probe target already exists: %s", probeFile)
		}
		deployDry := r.mustCommand(t, "harness", attempt, "deploy-dry-run", 2*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "deploy", "service="+service, "dry-run=true", "config="+r.config)
		var selected struct {
			DryRun         bool   `json:"dry_run"`
			TriggerCommand string `json:"trigger_command"`
		}
		r.decodeResult(t, "harness", deployDry, &selected)
		if !selected.DryRun || !strings.Contains(selected.TriggerCommand, service+"-live-binary") {
			r.harnessFatalf(t, "config-center dry-run did not select its live-binary resource")
		}
		billingDry := r.mustCommand(t, "harness", attempt, "billing-deploy-dry-run", 2*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "deploy", "service=billing-svc", "dry-run=true", "config="+r.config)
		billingDryText := r.readCommand(t, "harness", billingDry)
		if !strings.Contains(billingDryText, `"resource":"billing-svc"`) && !strings.Contains(billingDryText, `"resource": "billing-svc"`) {
			r.harnessFatalf(t, "billing dry-run did not select billing-svc")
		}
		if strings.Contains(billingDryText, "billing-svc-live-binary") {
			r.harnessFatalf(t, "billing dry-run unexpectedly selected a live-binary companion")
		}
		imageBefore = r.waitForTiltResource(t, attempt, service+"-image", true)
		liveBefore = r.waitForTiltResource(t, attempt, service+"-live-binary", false)
		r.waitForTiltResource(t, attempt, service, true)
		podBefore = r.singleConfigCenterPod(t, attempt, "pods-before")
		binaryBefore = r.podBinaryHash(t, attempt, "pod-binary-before", podBefore.Name)
		tiltLogBefore = r.captureRemoteTiltLog(t, attempt, "remote-tilt-before-deploy")
		marker = fmt.Sprintf("cp9-spec-probe-%d", time.Now().UnixNano())
		probe := fmt.Sprintf("package main\n\nimport \"log\"\n\nfunc init() { log.Print(%q) }\n", marker)
		if err := os.WriteFile(probeFile, []byte(probe), 0o600); err != nil {
			r.harnessFatalf(t, "write source marker probe: %v", err)
		}
		saveState()
		r.clearImageStamp(t, attempt, service+"-image")
		imageBefore = r.tiltResource(t, attempt, service+"-image", "image-before-empty-stamp-deploy")
		if imageStamp(imageBefore) != "" {
			r.harnessFatalf(t, "image-stamp regression fixture did not expose an empty stamp")
		}
		saveState()
	})

	r.checkpoint(t, stage, "deploy-source-without-image-rollout", func(t *testing.T) {
		r.mustAction(t, "product", attempt, "deploy-source", 30*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "deploy", "service="+service, "config="+r.config)
		r.requireHealthyStatus(t, r.status(t, attempt, "status-after-source-deploy"))
		liveAfter := r.tiltResource(t, attempt, service+"-live-binary", "live-binary-after")
		imageAfter = r.tiltResource(t, attempt, service+"-image", "image-after")
		podAfter = r.singleConfigCenterPod(t, attempt, "pods-after")
		if podAfter.UID != podBefore.UID || podAfter.ImageID != podBefore.ImageID {
			r.productFatalf(t, "source-only deploy changed Pod UID or image ID")
		}
		if r.podBinaryHash(t, attempt, "pod-binary-after", podAfter.Name) == binaryBefore {
			r.productFatalf(t, "source-only deploy returned with the previous binary")
		}
		logs := r.readCommand(t, "product", r.mustCommand(t, "product", attempt, "service-marker", 2*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "logs", "service="+service, "limit=200", "config="+r.config))
		if !strings.Contains(logs, marker) {
			r.productFatalf(t, "source marker is absent from current service logs")
		}
		if imageStamp(imageAfter) == "" {
			r.productFatalf(t, "source-only deploy did not publish the previously empty image stamp")
		}
		if !newBuild(imageAfter, imageBefore) || !newBuild(liveAfter, liveBefore) {
			// The quiescent baseline is also preserved as a command receipt. A missing
			// decoded baseline here means the harness could not prove a fresh invocation.
			r.harnessFatalf(t, "source-only deploy did not expose fresh terminal image and live-binary invocations")
		}
		imageFinished, imageErr := time.Parse(time.RFC3339Nano, imageAfter.Status.BuildHistory[0].FinishTime)
		liveStarted, liveErr := time.Parse(time.RFC3339Nano, liveAfter.Status.BuildHistory[0].StartTime)
		if imageErr != nil || liveErr != nil || !imageFinished.Before(liveStarted) {
			r.harnessFatalf(t, "image build must finish before live-binary compilation starts: image=%q live=%q", imageAfter.Status.BuildHistory[0].FinishTime, liveAfter.Status.BuildHistory[0].StartTime)
		}
		afterLog := r.captureRemoteTiltLog(t, attempt, "remote-tilt-after-deploy")
		order := sourceDeployOrder(tiltLogBefore, afterLog)
		if order.LiveBinaryBuild != 1 || order.AutomaticLiveUpdate != 1 || order.ManualImageTrigger != 0 || !order.Ordered {
			r.harnessFatalf(t, "source deploy build order is incomplete: %+v", order)
		}
		if order.ManualLiveBinaryTrigger < 0 || order.ManualLiveBinaryTrigger > 1 {
			r.harnessFatalf(t, "source deploy emitted unexpected manual live-binary triggers: %+v", order)
		}
		writeJSON(t, filepath.Join(attempt, "source-deploy-order.json"), order)
		saveState()
	})

	r.checkpoint(t, stage, "deploy-dockerfile-through-image-fallback", func(t *testing.T) {
		if err := appendFile(fallbackFile, "\n# "+marker+" image fallback\n"); err != nil {
			r.harnessFatalf(t, "write Dockerfile marker: %v", err)
		}
		r.mustAction(t, "product", attempt, "deploy-image-fallback", 45*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "deploy", "service="+service, "config="+r.config)
		imageFallback := r.tiltResource(t, attempt, service+"-image", "image-after-fallback")
		podAfterFallback = r.singleConfigCenterPod(t, attempt, "pods-after-fallback")
		if podAfterFallback.UID == podAfter.UID || podAfterFallback.ImageID == podAfter.ImageID {
			r.productFatalf(t, "Dockerfile deploy did not roll both Pod UID and image ID")
		}
		if imageStamp(imageFallback) == "" || imageStamp(imageFallback) == imageStamp(imageAfter) || !newBuild(imageFallback, imageAfter) {
			r.harnessFatalf(t, "Dockerfile deploy did not expose a fresh image invocation and stamp")
		}
		binaryAfterFallback = r.podBinaryHash(t, attempt, "pod-binary-after-fallback", podAfterFallback.Name)
		logs := r.readCommand(t, "product", r.mustCommand(t, "product", attempt, "pod-marker-after-fallback", 2*time.Minute, r.cp9Root, nil,
			r.sessionKubectl(), r.session, "-n", "cp9-global", "logs", podAfterFallback.Name, "-c", "config-center", "--tail=200"))
		if !strings.Contains(logs, marker) {
			r.productFatalf(t, "new image Pod logs do not contain the source marker")
		}
		saveState()
	})

	r.checkpoint(t, stage, "failed-source-preserves-last-good-binary", func(t *testing.T) {
		if err := appendFile(probeFile, "\nthis is not valid Go\n"); err != nil {
			r.harnessFatalf(t, "break source probe: %v", err)
		}
		failed := r.action(t, attempt, "deploy-compile-failure", 30*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "deploy", "service="+service, "config="+r.config)
		if failed.ExitCode == 0 {
			r.productFatalf(t, "invalid source deploy returned success")
		}
		failureText := r.readCommand(t, "harness", failed) + readFile(t, failed.Stderr)
		if !strings.Contains(failureText, service+"-live-binary") {
			r.productFatalf(t, "invalid source deploy did not name the failed resource")
		}
		if got := r.podBinaryHash(t, attempt, "pod-binary-after-failure", podAfterFallback.Name); got != binaryAfterFallback {
			r.productFatalf(t, "invalid source replaced the last good Pod binary")
		}
	})

	r.checkpoint(t, stage, "restore-source-image-path-and-health", func(t *testing.T) {
		if err := os.Remove(probeFile); err != nil {
			r.harnessFatalf(t, "remove source probe: %v", err)
		}
		if err := os.WriteFile(fallbackFile, fallback, 0o644); err != nil {
			r.harnessFatalf(t, "restore Dockerfile: %v", err)
		}
		r.mustAction(t, "product", attempt, "deploy-restore", 45*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "deploy", "service="+service, "config="+r.config)
		podRestored := r.singleConfigCenterPod(t, attempt, "pods-restored")
		if got := r.podBinaryHash(t, attempt, "pod-binary-restored", podRestored.Name); got != binaryBefore {
			r.productFatalf(t, "restored source did not restore the original binary")
		}
		remoteDocker := strings.Fields(r.readCommand(t, "product", r.mustCommand(t, "product", attempt, "remote-dockerfile-restored", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "sha256sum", filepath.Join(r.remoteSourceRoot, "Dockerfile"))))
		if len(remoteDocker) == 0 || remoteDocker[0] != bytesSHA256(fallback) {
			r.productFatalf(t, "remote Dockerfile did not restore exactly")
		}
		r.requireCleanSource(t)
		r.requireHealthyStatus(t, r.status(t, attempt, "status-after-restore"))
	})

	r.checkpoint(t, stage, "direct-image-build-waits-for-fresh-invocation", func(t *testing.T) {
		missing := r.command(t, attempt, "billing-live-binary-absent", 2*time.Minute, r.cp9Root, nil,
			r.sessionTiltGet(), r.session, "billing-svc-live-binary")
		missingText := strings.ToLower(r.readCommand(t, "harness", missing) + readFile(t, missing.Stderr))
		if missing.ExitCode == 0 || !strings.Contains(missingText, "not found") || !strings.Contains(missingText, "uiresource") {
			r.harnessFatalf(t, "billing-svc live-binary absence was not proven by Tilt NotFound")
		}
		if !billingBeforeCaptured {
			billingBefore = r.tiltResource(t, attempt, "billing-svc", "billing-before")
			billingBeforeCaptured = true
			saveState()
		}
		result := r.mustAction(t, "product", attempt, "billing-deploy", 30*time.Minute, r.cp9Root,
			filepath.Join(r.cp9Root, "bin", "cp9"), "deploy", "service=billing-svc", "config="+r.config)
		after := r.tiltResource(t, attempt, "billing-svc", "billing-after")
		if !newBuild(after, billingBefore) || !strings.Contains(r.readCommand(t, "product", result), "triggered billing-svc") {
			r.productFatalf(t, "billing-svc deploy did not wait for a fresh direct image build")
		}
	})
}

type remoteSourcePod struct {
	Name    string
	UID     string
	ImageID string
}

type remoteSourceDeployState struct {
	PodBefore             remoteSourcePod          `json:"pod_before"`
	PodAfter              remoteSourcePod          `json:"pod_after"`
	PodAfterFallback      remoteSourcePod          `json:"pod_after_fallback"`
	BinaryBefore          string                   `json:"binary_before"`
	BinaryAfterFallback   string                   `json:"binary_after_fallback"`
	ImageBefore           remoteSourceTiltResource `json:"image_before"`
	ImageAfter            remoteSourceTiltResource `json:"image_after"`
	LiveBefore            remoteSourceTiltResource `json:"live_before"`
	BillingBefore         remoteSourceTiltResource `json:"billing_before"`
	BillingBeforeCaptured bool                     `json:"billing_before_captured"`
	FallbackBaseline      []byte                   `json:"fallback_baseline"`
	Marker                string                   `json:"marker"`
	TiltLogBefore         string                   `json:"tilt_log_before"`
}

type remoteSourceBuildOrder struct {
	ManualLiveBinaryTrigger  int  `json:"manual_live_binary_trigger"`
	LiveBinaryBuild          int  `json:"live_binary_build"`
	AutomaticLiveUpdate      int  `json:"automatic_live_update"`
	ManualImageTrigger       int  `json:"manual_service_image_trigger"`
	ManualLiveBinaryPosition int  `json:"manual_live_binary_position"`
	LiveBinaryBuildPosition  int  `json:"live_binary_build_position"`
	AutomaticUpdatePosition  int  `json:"automatic_live_update_position"`
	Ordered                  bool `json:"ordered"`
}

func (r *remoteSourceRun) networkResume(t *testing.T) {
	stage := "network-resume"
	attempt := r.stageAttemptDir(stage)
	probe := filepath.Join(r.sourceRoot, "cp9-network-resume-probe.bin")
	modeFile := filepath.Join(attempt, "ssh-mode")
	timingDir := filepath.Join(attempt, "ssh-timings")
	faultBin := filepath.Join(attempt, "fault-bin")
	var up *remoteSourceProcess
	var watch *remoteSourceProcess
	var retryWatch *remoteSourceProcess
	defer func() {
		_ = os.Remove(modeFile)
		_ = os.Remove(probe)
		for _, process := range []*remoteSourceProcess{retryWatch, watch, up} {
			r.stopProcess(process)
		}
	}()

	r.checkpoint(t, stage, "prepare-scoped-transport-fault", func(t *testing.T) {
		if err := os.MkdirAll(faultBin, 0o700); err != nil {
			r.harnessFatalf(t, "create fault wrapper directory: %v", err)
		}
		if err := os.MkdirAll(timingDir, 0o700); err != nil {
			r.harnessFatalf(t, "create timing directory: %v", err)
		}
		executable, err := os.Executable()
		if err != nil {
			r.harnessFatalf(t, "locate functional test executable: %v", err)
		}
		if err := os.Symlink(executable, filepath.Join(faultBin, "ssh")); err != nil {
			r.harnessFatalf(t, "install scoped SSH helper: %v", err)
		}
		realSSH, err := exec.LookPath("ssh")
		if err != nil {
			r.harnessFatalf(t, "locate real ssh: %v", err)
		}
		r.commandEnv[remoteSourceSSHHelper] = "1"
		r.commandEnv[remoteSourceSSHReal] = realSSH
		r.commandEnv[remoteSourceSSHMode] = modeFile
		r.commandEnv[remoteSourceSSHTimings] = timingDir
		r.commandEnv["PATH"] = faultBin + string(os.PathListSeparator) + os.Getenv("PATH")
		if err := writeRandomFile(probe, 128<<20); err != nil {
			r.harnessFatalf(t, "write transfer probe: %v", err)
		}
		if err := os.WriteFile(filepath.Join(attempt, "transfer-probe.sha256"), []byte(fileSHA256(t, probe)+"\n"), 0o600); err != nil {
			r.harnessFatalf(t, "write transfer probe hash: %v", err)
		}
	})

	r.checkpoint(t, stage, "interrupt-active-source-transfer", func(t *testing.T) {
		var err error
		up, err = r.startCommand(attempt, "up-interrupted", 3*time.Hour, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "up", "config="+r.config)
		if err != nil {
			r.productFatalf(t, "start source transfer: %v", err)
		}
		rsyncPID, command, err := waitForTransferProcess(up.cmd.Process.Pid, r.remoteSourceRoot, 15*time.Minute)
		if err != nil {
			r.harnessFatalf(t, "observe active source transfer: %v", err)
		}
		if err := os.WriteFile(filepath.Join(attempt, "transfer-active.txt"), []byte(fmt.Sprintf("pid=%d command=%s\n", rsyncPID, command)), 0o600); err != nil {
			r.harnessFatalf(t, "write active transfer evidence: %v", err)
		}
		if err := os.WriteFile(modeFile, []byte("network\n"), 0o600); err != nil {
			r.harnessFatalf(t, "activate scoped network fault: %v", err)
		}
		if err := syscall.Kill(rsyncPID, syscall.SIGTERM); err != nil {
			r.harnessFatalf(t, "interrupt source rsync %d: %v", rsyncPID, err)
		}
		outage := r.command(t, attempt, "status-during-outage", 2*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "config="+r.config)
		var status remoteSourceStatus
		r.decodeResult(t, "product", outage, &status)
		if status.SessionName != r.session || status.Layers.Tilt.Status != "unreachable" {
			r.productFatalf(t, "network fault status = session %q Tilt %q, want same session and unreachable", status.SessionName, status.Layers.Tilt.Status)
		}
		if err := os.Remove(modeFile); err != nil {
			r.harnessFatalf(t, "restore scoped network transport: %v", err)
		}
	})

	r.checkpoint(t, stage, "recover-same-remote-controller", func(t *testing.T) {
		restarted := false
		deadline := time.Now().Add(10 * time.Minute)
		for time.Now().Before(deadline) {
			status, ok := r.tryStatus(attempt, "status-before-detach")
			if ok && status.SessionName == r.session && status.Layers.Tilt.Controller == "remote" && status.Layers.Tilt.Running {
				break
			}
			if !processRunning(up) {
				if restarted {
					r.productFatalf(t, "source transfer command exited again before controller recovery")
				}
				_, _ = up.wait()
				var err error
				up, err = r.startCommand(attempt, "up-restarted", 3*time.Hour, r.cp9Root, nil,
					filepath.Join(r.cp9Root, "bin", "cp9"), "up", "config="+r.config)
				if err != nil {
					r.productFatalf(t, "restart same-session up: %v", err)
				}
				restarted = true
			}
			time.Sleep(5 * time.Second)
		}
		status := r.status(t, attempt, "status-controller-recovered")
		if status.SessionName != r.session || status.Layers.Tilt.Controller != "remote" || !status.Layers.Tilt.Running {
			r.productFatalf(t, "same remote controller did not recover")
		}
		if !processRunning(up) {
			r.productFatalf(t, "local up process exited before detach checkpoint")
		}
		r.stopProcess(up)
		up = nil
		status = r.status(t, attempt, "status-after-detach")
		if status.SessionName != r.session || status.Status == "failed" || status.Layers.Tilt.Controller != "remote" || !status.Layers.Tilt.Running {
			r.productFatalf(t, "detaching local up stopped or replaced the remote controller")
		}
		r.requireHealthyTunnels(t, status)
	})

	r.checkpoint(t, stage, "watch-logs-complete-transfer-and-deploy", func(t *testing.T) {
		var err error
		watch, err = r.startCommand(attempt, "status-watch-after-detach", 3*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "--watch", "config="+r.config)
		if err != nil {
			r.productFatalf(t, "start status watch after detach: %v", err)
		}
		if !waitForLineCount(watch.stdoutPath, 2, 100*time.Second, func() bool { return processRunning(watch) }) {
			r.productFatalf(t, "status watch did not remain usable after detach")
		}
		r.stopProcess(watch)
		watch = nil
		r.mustCommand(t, "product", attempt, "logs-after-detach", 2*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "logs", "limit=50", "config="+r.config)
		r.mustCommand(t, "product", attempt, "up-resume", 45*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "up", "config="+r.config)
		status := r.status(t, attempt, "status-ready-after-resume")
		r.requireHealthyStatus(t, status)
		remote := strings.Fields(r.readCommand(t, "product", r.mustCommand(t, "product", attempt, "remote-transfer-hash", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "sha256sum", filepath.Join(r.remoteSourceRoot, filepath.Base(probe)))))
		if len(remote) == 0 || remote[0] != fileSHA256(t, probe) {
			r.productFatalf(t, "resumed transfer is incomplete")
		}
		r.mustCommand(t, "product", attempt, "deploy-after-resume", 30*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "deploy", "service=config-center", "config="+r.config)
	})

	r.checkpoint(t, stage, "remove-transfer-probe", func(t *testing.T) {
		if err := os.Remove(probe); err != nil {
			r.harnessFatalf(t, "remove local transfer probe: %v", err)
		}
		r.mustCommand(t, "product", attempt, "up-remove-probe", 30*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "up", "config="+r.config)
		absent := r.command(t, attempt, "remote-transfer-probe-absent", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "test", "!", "-e", filepath.Join(r.remoteSourceRoot, filepath.Base(probe)))
		if absent.ExitCode != 0 {
			r.productFatalf(t, "removed transfer probe remains remotely")
		}
		r.requireHealthyStatus(t, r.status(t, attempt, "status-after-probe-removal"))
	})

	r.checkpoint(t, stage, "ssm-commands-stay-bounded-when-applicable", func(t *testing.T) {
		status := r.status(t, attempt, "status-before-ssm-check")
		if status.Layers.Host.Execution.Remote.SSHTransport != "ssm" {
			if err := os.WriteFile(filepath.Join(attempt, "ssm-checkpoint.txt"), []byte("NOT RUN: persisted transport is not SSM\n"), 0o600); err != nil {
				r.harnessFatalf(t, "write SSM not-run receipt: %v", err)
			}
			t.Skip("persisted transport is not SSM")
		}
		r.mustCommand(t, "product", attempt, "doctor-ssm", 2*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "doctor", "config="+r.config)
		r.requireHealthyStatus(t, r.status(t, attempt, "status-ssm"))
	})

	r.checkpoint(t, stage, "healthy-probes-exceed-one-aggregate-budget", func(t *testing.T) {
		if err := os.WriteFile(modeFile, []byte("delay\n"), 0o600); err != nil {
			r.harnessFatalf(t, "enable delayed healthy probes: %v", err)
		}
		result := r.mustCommand(t, "product", attempt, "status-delayed-healthy", 90*time.Second, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "config="+r.config)
		if err := os.Remove(modeFile); err != nil {
			r.harnessFatalf(t, "disable delayed healthy probes: %v", err)
		}
		var status remoteSourceStatus
		r.decodeResult(t, "harness", result, &status)
		if status.SessionName != r.session || status.Status != "ready" || status.Layers.Tilt.ReadyResources != status.Layers.Tilt.TotalResources || status.Layers.Tilt.TotalResources == 0 {
			r.productFatalf(t, "delayed healthy status lost real ready resource counts")
		}
		observation := readProbeTiming(t, filepath.Join(timingDir, "observation.json"))
		readiness := readProbeTiming(t, filepath.Join(timingDir, "readiness.json"))
		if observation.Total < 16*time.Second || observation.Total >= 30*time.Second || readiness.Total < 16*time.Second || readiness.Total >= 30*time.Second || observation.Total+readiness.Total <= 30*time.Second {
			r.harnessFatalf(t, "real probe timings do not satisfy the aggregate budget contract: observation=%s readiness=%s", observation.Total, readiness.Total)
		}
		if result.Duration <= 30*time.Second {
			r.harnessFatalf(t, "status wall time %s did not actually exceed one 30s command budget", result.Duration)
		}
	})

	r.checkpoint(t, stage, "readiness-timeout-is-retryable-and-watch-recovers", func(t *testing.T) {
		if err := os.WriteFile(modeFile, []byte("stall-readiness\n"), 0o600); err != nil {
			r.harnessFatalf(t, "stall readiness probe: %v", err)
		}
		var err error
		retryWatch, err = r.startCommand(attempt, "status-retryable-watch", 5*time.Minute, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "--watch", "config="+r.config)
		if err != nil {
			r.productFatalf(t, "start retryable status watch: %v", err)
		}
		if !waitForWatchEvent(retryWatch.stdoutPath, 75*time.Second, func(event remoteSourceWatchEvent) bool {
			return event.State.Status == "initializing" && event.State.TiltStatus == "unavailable" && event.State.ProbeRetryable && event.Action.Code == "wait" && strings.Contains(event.State.ProbeError, "remote Tilt resource readiness timed out after 30s")
		}, func() bool { return processRunning(retryWatch) }) {
			r.productFatalf(t, "watch did not report the retryable readiness 30s timeout contract")
		}
		if err := os.Remove(modeFile); err != nil {
			r.harnessFatalf(t, "remove readiness stall: %v", err)
		}
		if !waitForWatchEvent(retryWatch.stdoutPath, 2*time.Minute, func(event remoteSourceWatchEvent) bool {
			return event.State.Status == "ready"
		}, func() bool { return processRunning(retryWatch) }) {
			r.productFatalf(t, "same status watcher did not recover to ready")
		}
		r.stopProcess(retryWatch)
		retryWatch = nil
	})

	r.checkpoint(t, stage, "artifact-sync-timeout-is-retryable", func(t *testing.T) {
		if err := os.WriteFile(modeFile, []byte("stall-artifact-sync\n"), 0o600); err != nil {
			r.harnessFatalf(t, "stall artifact sync: %v", err)
		}
		result := r.mustCommand(t, "product", attempt, "status-artifact-timeout", 90*time.Second, r.cp9Root, nil,
			filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "config="+r.config)
		if err := os.Remove(modeFile); err != nil {
			r.harnessFatalf(t, "remove artifact-sync stall: %v", err)
		}
		var status remoteSourceStatus
		r.decodeResult(t, "harness", result, &status)
		if status.SessionName != r.session || status.Status != "initializing" || status.Layers.Tilt.Status != "unavailable" || !status.Layers.Tilt.Retryable || !strings.Contains(status.Layers.Tilt.ReadinessError, "remote Tilt artifact sync timed out after 30s") {
			r.productFatalf(t, "artifact-sync timeout did not expose the retryable initializing 30s contract")
		}
		r.requireHealthyStatus(t, r.status(t, attempt, "status-after-timeout-probes"))
	})
}

func (r *remoteSourceRun) ensurePrepared(t *testing.T) {
	t.Helper()
	if r.prepared {
		return
	}
	if r.prepareAttempted {
		r.failureKind = "prerequisite"
		t.Fatalf("session preparation already failed; refusing to create a second scene")
	}
	r.prepareAttempted = true
	metadataPath := filepath.Join(r.root, "default-804-session.json")
	if !enabled(os.Getenv(remoteSourceResume)) {
		result := r.mustCommand(t, "prerequisite", r.root, "prepare-default-session", 10*time.Minute, r.originRoot, nil,
			filepath.Join(r.originRoot, "spec-tests", "scripts", "default-804-session.sh"), "prepare", "source-development", r.root)
		_ = result
	}
	decodeFile(t, metadataPath, &r.metadata)
	if r.metadata.Head == "" || r.metadata.Session == "" || r.metadata.CP9Root == "" || r.metadata.Config == "" || r.metadata.OriginCP9 != r.originRoot {
		r.harnessFatalf(t, "prepared session metadata is incomplete or belongs to another checkout")
	}
	if r.metadata.Head != r.currentHead(t) {
		r.harnessFatalf(t, "prepared head %s does not match executing head %s", r.metadata.Head, r.currentHead(t))
	}
	r.cp9Root = r.metadata.CP9Root
	r.config = r.metadata.Config
	r.sourceRoot = filepath.Join(r.metadata.WorkRoot, "config-center")
	r.sourceFile = filepath.Join(r.sourceRoot, "cmd", "server", "main.go")
	r.session = r.metadata.Session
	for _, path := range []string{r.cp9Root, r.config, r.sourceRoot, r.sourceFile} {
		if _, err := os.Stat(path); err != nil {
			r.harnessFatalf(t, "prepared path %s is unavailable: %v", path, err)
		}
	}
	r.prepared = true
	if enabled(os.Getenv(remoteSourceResume)) {
		status := r.status(t, r.root, "resume-status")
		r.remoteSourceRoot = filepath.Join(status.Layers.Host.Execution.SourceSyncDestination, "controller", "work", "config-center")
	}
	r.writeState(t)
}

func (r *remoteSourceRun) preflight(t *testing.T, evidenceDir, label string) bool {
	t.Helper()
	if !r.prepared {
		r.ensurePrepared(t)
	}
	status, ok := r.tryStatus(evidenceDir, label+"-status")
	if !ok {
		r.failureKind = "harness"
		return false
	}
	expectedCompileFailure, validActionState := r.expectedCompileFailureAction()
	if !validActionState {
		return false
	}
	if status.SessionName != r.session || status.Layers.Tilt.Controller != "remote" || !status.Layers.Tilt.Running ||
		!healthyTunnels(status.Layers.Tilt.TunnelProcesses) || (status.Status != "ready" && !expectedCompileFailure) {
		r.failureKind = "blocked"
		return false
	}
	if r.remoteSourceRoot == "" {
		r.remoteSourceRoot = filepath.Join(status.Layers.Host.Execution.SourceSyncDestination, "controller", "work", "config-center")
	}
	if !r.sourceClean() && (!expectedCompileFailure || !r.expectedCompileFailureLocalSource()) {
		r.failureKind = "blocked"
		return false
	}
	localSource := fileSHA256(t, r.sourceFile)
	localDocker, expectedProbe, ok := r.remoteSourceExpectations()
	if !ok {
		return false
	}
	remoteSource := r.command(t, evidenceDir, label+"-remote-source", 2*time.Minute, r.cp9Root, nil,
		r.sessionSSH(), r.session, "sha256sum", filepath.Join(r.remoteSourceRoot, strings.TrimPrefix(r.sourceFile, r.sourceRoot+string(filepath.Separator))))
	remoteDocker := r.command(t, evidenceDir, label+"-remote-dockerfile", 2*time.Minute, r.cp9Root, nil,
		r.sessionSSH(), r.session, "sha256sum", filepath.Join(r.remoteSourceRoot, "Dockerfile"))
	if remoteSource.ExitCode != 0 || remoteDocker.ExitCode != 0 {
		r.failureKind = "harness"
		return false
	}
	sourceFields := strings.Fields(readFile(t, remoteSource.Stdout))
	dockerFields := strings.Fields(readFile(t, remoteDocker.Stdout))
	if len(sourceFields) == 0 || len(dockerFields) == 0 || sourceFields[0] != localSource || dockerFields[0] != localDocker {
		r.failureKind = "blocked"
		return false
	}
	probeName := "cmd/server/cp9_spec_probe.go"
	if expectedProbe != "" {
		probe := r.command(t, evidenceDir, label+"-expected-source-probe", 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "sha256sum", filepath.Join(r.remoteSourceRoot, probeName))
		fields := strings.Fields(readFile(t, probe.Stdout))
		if probe.ExitCode != 0 || len(fields) == 0 || fields[0] != expectedProbe {
			r.failureKind = "blocked"
			return false
		}
	} else {
		absent := r.command(t, evidenceDir, label+"-absent-"+sanitizeRemoteSourceLabel(probeName), 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "test", "!", "-e", filepath.Join(r.remoteSourceRoot, probeName))
		if absent.ExitCode != 0 {
			r.failureKind = "blocked"
			return false
		}
	}
	for _, name := range []string{"cp9-spec-untracked-probe.txt", "cp9-network-resume-probe.bin"} {
		absent := r.command(t, evidenceDir, label+"-absent-"+sanitizeRemoteSourceLabel(name), 2*time.Minute, r.cp9Root, nil,
			r.sessionSSH(), r.session, "test", "!", "-e", filepath.Join(r.remoteSourceRoot, name))
		if absent.ExitCode != 0 {
			r.failureKind = "blocked"
			return false
		}
	}
	if status.Status != "ready" && !r.validateExpectedCompileFailure(t, evidenceDir, status) {
		return false
	}
	return true
}

func (r *remoteSourceRun) expectedCompileFailureAction() (bool, bool) {
	if r.currentStage != "deploy-current-source" {
		return false, true
	}
	failed, exists, err := r.savedAction("deploy-compile-failure")
	if err != nil {
		r.failureKind = "harness"
		return false, false
	}
	if !exists {
		return false, true
	}
	restore, restoring, err := r.savedAction("deploy-restore")
	if err != nil {
		r.failureKind = "harness"
		return false, false
	}
	if restoring {
		if restore.Status == "started" {
			r.failureKind = "blocked"
			return false, false
		}
		return false, true
	}
	want := []string{filepath.Join(r.cp9Root, "bin", "cp9"), "deploy", "service=config-center", "config=" + r.config}
	if failed.Status != "completed" || failed.Head != r.currentHead(r.t) || failed.Directory != r.cp9Root ||
		!slices.Equal(failed.Command, want) || failed.ExitCode == 0 {
		r.failureKind = "blocked"
		return false, false
	}
	return true, true
}

func (r *remoteSourceRun) expectedCompileFailureLocalSource() bool {
	var state remoteSourceDeployState
	if decodeJSONFile(filepath.Join(r.root, "state", "deploy.json"), &state) != nil || state.Marker == "" || len(state.FallbackBaseline) == 0 {
		return false
	}
	status := exec.Command("git", "-C", r.sourceRoot, "status", "--porcelain")
	data, err := status.Output()
	if err != nil {
		return false
	}
	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
		if line == "" {
			continue
		}
		if len(line) < 4 {
			return false
		}
		path := strings.TrimSpace(line[2:])
		if path != "Dockerfile" && path != "cmd/server/cp9_spec_probe.go" {
			return false
		}
	}
	docker, err := os.ReadFile(filepath.Join(r.sourceRoot, "Dockerfile"))
	if err != nil {
		return false
	}
	markedDocker := append(append([]byte(nil), state.FallbackBaseline...), []byte("\n# "+state.Marker+" image fallback\n")...)
	if !bytes.Equal(docker, state.FallbackBaseline) && !bytes.Equal(docker, markedDocker) {
		return false
	}
	probe, err := os.ReadFile(filepath.Join(r.sourceRoot, "cmd", "server", "cp9_spec_probe.go"))
	if os.IsNotExist(err) {
		return true
	}
	expectedProbe := []byte(fmt.Sprintf("package main\n\nimport \"log\"\n\nfunc init() { log.Print(%q) }\n\nthis is not valid Go\n", state.Marker))
	return err == nil && bytes.Equal(probe, expectedProbe)
}

func (r *remoteSourceRun) validateExpectedCompileFailure(t *testing.T, evidenceDir string, status remoteSourceStatus) bool {
	t.Helper()
	const resourceName = "config-center-live-binary"
	if status.Status != "failed" || status.Layers.Tilt.Status != "failed" || status.Layers.Tilt.TotalResources < 1 ||
		status.Layers.Tilt.ReadyResources+1 != status.Layers.Tilt.TotalResources ||
		!slices.Equal(status.Layers.Tilt.NotReadyResources, []string{resourceName}) ||
		!slices.Equal(status.Layers.Tilt.FailedResources, []string{resourceName}) {
		r.failureKind = "blocked"
		return false
	}
	resource := r.command(t, evidenceDir, "expected-failed-live-binary", 2*time.Minute, r.cp9Root, nil,
		r.sessionTiltGet(), r.session, resourceName)
	if resource.ExitCode != 0 {
		r.failureKind = "harness"
		return false
	}
	var liveBinary remoteSourceTiltResource
	if decodeJSONFile(resource.Stdout, &liveBinary) != nil {
		r.failureKind = "harness"
		return false
	}
	if liveBinary.Metadata.Name != resourceName || !strings.EqualFold(liveBinary.Status.UpdateStatus, "error") {
		r.failureKind = "blocked"
		return false
	}
	var state remoteSourceDeployState
	if decodeJSONFile(filepath.Join(r.root, "state", "deploy.json"), &state) != nil ||
		state.PodAfterFallback.Name == "" || state.BinaryAfterFallback == "" {
		r.failureKind = "harness"
		return false
	}
	podsResult := r.command(t, evidenceDir, "expected-last-good-pod", 2*time.Minute, r.cp9Root, nil,
		r.sessionKubectl(), r.session, "-n", "cp9-global", "get", "pods", "-l", "app.kubernetes.io/name=config-center", "-o", "json")
	if podsResult.ExitCode != 0 {
		r.failureKind = "harness"
		return false
	}
	var pods remoteSourcePodList
	if decodeJSONFile(podsResult.Stdout, &pods) != nil || len(pods.Items) != 1 {
		r.failureKind = "harness"
		return false
	}
	pod := pods.Items[0]
	ready, containerReady, imageID := false, false, ""
	for _, condition := range pod.Status.Conditions {
		ready = ready || strings.EqualFold(condition.Type, "Ready") && strings.EqualFold(condition.Status, "True")
	}
	for _, container := range pod.Status.ContainerStatuses {
		if container.Name == "config-center" {
			containerReady, imageID = container.Ready, container.ImageID
		}
	}
	if pod.Metadata.Name != state.PodAfterFallback.Name || pod.Metadata.UID != state.PodAfterFallback.UID ||
		imageID != state.PodAfterFallback.ImageID || !strings.EqualFold(pod.Status.Phase, "Running") || !ready || !containerReady {
		r.failureKind = "blocked"
		return false
	}
	binary := r.command(t, evidenceDir, "expected-last-good-binary", 2*time.Minute, r.cp9Root, nil,
		r.sessionKubectl(), r.session, "-n", "cp9-global", "exec", pod.Metadata.Name, "-c", "config-center", "--", "sha256sum", "/workspace/service")
	if binary.ExitCode != 0 {
		r.failureKind = "harness"
		return false
	}
	data, err := os.ReadFile(binary.Stdout)
	if err != nil {
		r.failureKind = "harness"
		return false
	}
	fields := strings.Fields(string(data))
	if len(fields) == 0 {
		r.failureKind = "harness"
		return false
	}
	if fields[0] != state.BinaryAfterFallback {
		r.failureKind = "blocked"
		return false
	}
	return true
}

func (r *remoteSourceRun) remoteSourceExpectations() (string, string, bool) {
	docker, err := os.ReadFile(filepath.Join(r.sourceRoot, "Dockerfile"))
	if err != nil {
		r.failureKind = "harness"
		return "", "", false
	}
	if r.currentStage != "deploy-current-source" {
		return bytesSHA256(docker), "", true
	}
	actions := map[string]remoteSourceSavedAction{}
	for _, name := range []string{"deploy-source", "deploy-image-fallback", "deploy-compile-failure", "deploy-restore"} {
		action, exists, actionErr := r.savedAction(name)
		if actionErr != nil {
			r.failureKind = "harness"
			return "", "", false
		}
		if !exists {
			continue
		}
		if action.Status == "started" {
			r.failureKind = "blocked"
			return "", "", false
		}
		if action.Status != "completed" {
			r.failureKind = "harness"
			return "", "", false
		}
		actions[name] = action
	}
	if action, restored := actions["deploy-restore"]; restored {
		if action.ExitCode != 0 {
			r.failureKind = "blocked"
			return "", "", false
		}
		return bytesSHA256(docker), "", true
	}
	deployed, exists := actions["deploy-source"]
	if !exists {
		return bytesSHA256(docker), "", true
	}
	if deployed.ExitCode != 0 {
		r.failureKind = "blocked"
		return "", "", false
	}
	var state remoteSourceDeployState
	if err := decodeJSONFile(filepath.Join(r.root, "state", "deploy.json"), &state); err != nil || state.Marker == "" || len(state.FallbackBaseline) == 0 {
		r.failureKind = "harness"
		return "", "", false
	}
	docker = append([]byte(nil), state.FallbackBaseline...)
	probe := fmt.Sprintf("package main\n\nimport \"log\"\n\nfunc init() { log.Print(%q) }\n", state.Marker)
	if _, failed := actions["deploy-compile-failure"]; failed {
		probe += "\nthis is not valid Go\n"
	}
	if action, fallback := actions["deploy-image-fallback"]; fallback {
		if action.ExitCode != 0 {
			r.failureKind = "blocked"
			return "", "", false
		}
		docker = append(docker, []byte("\n# "+state.Marker+" image fallback\n")...)
	}
	return bytesSHA256(docker), bytesSHA256([]byte(probe)), true
}

func (r *remoteSourceRun) status(t *testing.T, evidenceDir, label string) remoteSourceStatus {
	t.Helper()
	result := r.mustCommand(t, "product", evidenceDir, label, 2*time.Minute, r.cp9Root, nil,
		filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "config="+r.config)
	var status remoteSourceStatus
	r.decodeResult(t, "harness", result, &status)
	return status
}

func (r *remoteSourceRun) tryStatus(evidenceDir, label string) (remoteSourceStatus, bool) {
	result := r.command(r.t, evidenceDir, label, 2*time.Minute, r.cp9Root, nil,
		filepath.Join(r.cp9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "config="+r.config)
	if result.ExitCode != 0 {
		return remoteSourceStatus{}, false
	}
	var status remoteSourceStatus
	if err := decodeJSONFile(result.Stdout, &status); err != nil {
		return remoteSourceStatus{}, false
	}
	return status, true
}

func (r *remoteSourceRun) requireHealthyStatus(t *testing.T, status remoteSourceStatus) {
	t.Helper()
	if status.SessionName != r.session || status.Status != "ready" || status.Layers.Tilt.Controller != "remote" || !status.Layers.Tilt.Running {
		r.productFatalf(t, "session status is not the same Ready remote controller")
	}
	r.requireHealthyTunnels(t, status)
}

func (r *remoteSourceRun) requireHealthyTunnels(t *testing.T, status remoteSourceStatus) {
	t.Helper()
	if !healthyTunnels(status.Layers.Tilt.TunnelProcesses) {
		r.productFatalf(t, "session does not have exactly one healthy controller and kube-api tunnel supervisor")
	}
}

func healthyTunnels(tunnels []remoteSourceTunnel) bool {
	if len(tunnels) != 2 {
		return false
	}
	roles := map[string]bool{}
	pids := map[int]bool{}
	for _, tunnel := range tunnels {
		if tunnel.PID <= 0 || tunnel.State != "running" || tunnel.ActionRequired || (tunnel.Role != "controller" && tunnel.Role != "kube-api") {
			return false
		}
		roles[tunnel.Role] = true
		pids[tunnel.PID] = true
	}
	return roles["controller"] && roles["kube-api"] && len(pids) == 2
}

func (r *remoteSourceRun) requireCleanSource(t *testing.T) {
	t.Helper()
	if !r.sourceClean() {
		r.harnessFatalf(t, "source checkout is not clean")
	}
}

func (r *remoteSourceRun) sourceClean() bool {
	cmd := exec.Command("git", "-C", r.sourceRoot, "status", "--porcelain")
	data, err := cmd.Output()
	return err == nil && strings.TrimSpace(string(data)) == ""
}

func (r *remoteSourceRun) sessionSSH() string {
	return filepath.Join(r.cp9Root, "skills", "get-started", "scripts", "session-ssh.sh")
}

func (r *remoteSourceRun) sessionKubectl() string {
	return filepath.Join(r.cp9Root, "skills", "get-started", "scripts", "session-kubectl.sh")
}

func (r *remoteSourceRun) sessionTiltGet() string {
	return filepath.Join(r.cp9Root, "spec-tests", "scripts", "session-tilt-get.sh")
}

func (r *remoteSourceRun) tiltResource(t *testing.T, evidenceDir, resource, label string) remoteSourceTiltResource {
	t.Helper()
	result := r.mustCommand(t, "harness", evidenceDir, label, 2*time.Minute, r.cp9Root, nil, r.sessionTiltGet(), r.session, resource)
	var value remoteSourceTiltResource
	r.decodeResult(t, "harness", result, &value)
	return value
}

func (r *remoteSourceRun) clearImageStamp(t *testing.T, evidenceDir, resource string) {
	t.Helper()
	var status struct {
		APIPort int `json:"api_port"`
	}
	decodeFile(t, filepath.Join(r.cp9Root, ".cp9", r.session, "tilt", "reports", "tilt-status.json"), &status)
	if status.APIPort <= 0 {
		r.harnessFatalf(t, "remote Tilt API port is missing")
	}
	r.mustAction(t, "harness", evidenceDir, "clear-image-stamp", 2*time.Minute, r.cp9Root,
		r.sessionSSH(), r.session, "tilt", "patch", "uiresource", resource, "--type=merge", "-p",
		`{"metadata":{"annotations":{"cp9.tidbcloud.com/image-stamp":null}}}`,
		"--host", "127.0.0.1", "--port", strconv.Itoa(status.APIPort))
}

func (r *remoteSourceRun) waitForTiltResource(t *testing.T, evidenceDir, resource string, requireHistory bool) remoteSourceTiltResource {
	t.Helper()
	deadline := time.Now().Add(2 * time.Minute)
	var previous string
	for attempt := 1; time.Now().Before(deadline); attempt++ {
		value := r.tiltResource(t, evidenceDir, resource, fmt.Sprintf("%s-before-%02d", sanitizeRemoteSourceLabel(resource), attempt))
		status := strings.ToLower(value.Status.UpdateStatus)
		terminal := status == "none" && !requireHistory || status == "ok" && len(value.Status.BuildHistory) > 0
		fingerprint, _ := json.Marshal(value)
		if terminal && previous == string(fingerprint) {
			return value
		}
		if terminal {
			previous = string(fingerprint)
		} else {
			previous = ""
		}
		time.Sleep(2 * time.Second)
	}
	r.harnessFatalf(t, "Tilt resource %s did not reach a stable terminal state", resource)
	return remoteSourceTiltResource{}
}

func (r *remoteSourceRun) singleConfigCenterPod(t *testing.T, evidenceDir, label string) remoteSourcePod {
	t.Helper()
	result := r.mustCommand(t, "product", evidenceDir, label, 2*time.Minute, r.cp9Root, nil,
		r.sessionKubectl(), r.session, "-n", "cp9-global", "get", "pods", "-l", "app.kubernetes.io/name=config-center", "-o", "json")
	var pods remoteSourcePodList
	r.decodeResult(t, "harness", result, &pods)
	if len(pods.Items) != 1 {
		r.productFatalf(t, "config-center Pod count = %d, want 1", len(pods.Items))
	}
	pod := remoteSourcePod{Name: pods.Items[0].Metadata.Name, UID: pods.Items[0].Metadata.UID}
	for _, container := range pods.Items[0].Status.ContainerStatuses {
		if container.Name == "config-center" {
			pod.ImageID = container.ImageID
		}
	}
	if pod.Name == "" || pod.UID == "" || pod.ImageID == "" {
		r.harnessFatalf(t, "config-center Pod identity is incomplete")
	}
	return pod
}

func (r *remoteSourceRun) podBinaryHash(t *testing.T, evidenceDir, label, pod string) string {
	t.Helper()
	result := r.mustCommand(t, "product", evidenceDir, label, 2*time.Minute, r.cp9Root, nil,
		r.sessionKubectl(), r.session, "-n", "cp9-global", "exec", pod, "-c", "config-center", "--", "sha256sum", "/workspace/service")
	fields := strings.Fields(r.readCommand(t, "harness", result))
	if len(fields) == 0 {
		r.harnessFatalf(t, "Pod binary hash is empty")
	}
	return fields[0]
}

func (r *remoteSourceRun) captureRemoteTiltLog(t *testing.T, evidenceDir, label string) string {
	t.Helper()
	var host struct {
		Remote struct {
			RemoteStateDir string `json:"remote_state_dir"`
		} `json:"remote"`
	}
	decodeFile(t, filepath.Join(r.cp9Root, ".cp9", r.session, "host.json"), &host)
	if host.Remote.RemoteStateDir == "" {
		r.harnessFatalf(t, "remote state directory is absent from host.json")
	}
	result := r.mustCommand(t, "harness", evidenceDir, label, 2*time.Minute, r.cp9Root, nil,
		r.sessionSSH(), r.session, "tail", "-n", "5000", "--", filepath.Join(host.Remote.RemoteStateDir, "controller", "session", "tilt-up.log"))
	return r.readCommand(t, "harness", result)
}

func (r *remoteSourceRun) decodeResult(t *testing.T, kind string, result remoteSourceCommandResult, target any) {
	t.Helper()
	if err := decodeJSONFile(result.Stdout, target); err != nil {
		r.failureKind = kind
		t.Fatalf("decode %s: %v", result.Stdout, err)
	}
}

func (r *remoteSourceRun) readCommand(t *testing.T, kind string, result remoteSourceCommandResult) string {
	t.Helper()
	data, err := os.ReadFile(result.Stdout)
	if err != nil {
		r.failureKind = kind
		t.Fatalf("read command output %s: %v", result.Stdout, err)
	}
	return string(data)
}

func (r *remoteSourceRun) mustCommand(t *testing.T, kind, evidenceDir, label string, timeout time.Duration, dir string, stdin io.Reader, argv ...string) remoteSourceCommandResult {
	t.Helper()
	result := r.command(t, evidenceDir, label, timeout, dir, stdin, argv...)
	if result.ExitCode != 0 {
		r.failureKind = kind
		t.Fatalf("command %s exited %d; see %s and %s", label, result.ExitCode, result.Stdout, result.Stderr)
	}
	return result
}

func (r *remoteSourceRun) mustAction(t *testing.T, kind, evidenceDir, label string, timeout time.Duration, dir string, argv ...string) remoteSourceCommandResult {
	t.Helper()
	result := r.action(t, evidenceDir, label, timeout, dir, argv...)
	if result.ExitCode != 0 {
		r.failureKind = kind
		t.Fatalf("action %s exited %d; see %s and %s", label, result.ExitCode, result.Stdout, result.Stderr)
	}
	return result
}

func (r *remoteSourceRun) action(t *testing.T, evidenceDir, label string, timeout time.Duration, dir string, argv ...string) remoteSourceCommandResult {
	t.Helper()
	if r.currentStage == "" {
		r.harnessFatalf(t, "action %s has no active stage", label)
	}
	receiptPath := r.actionReceiptPath(label)
	if enabled(os.Getenv(remoteSourceResume)) && fileExists(receiptPath) {
		saved, _, err := r.savedAction(label)
		if err != nil {
			r.harnessFatalf(t, "read saved action %s: %v", label, err)
		}
		if saved.Head != r.currentHead(t) || saved.Directory != dir || !slices.Equal(saved.Command, argv) {
			r.harnessFatalf(t, "saved action %s does not match the executing head or command", label)
		}
		if saved.Status == "started" {
			r.failureKind = "blocked"
			t.Fatalf("action %s has a non-terminal receipt; refusing to replay an action with an unknown outcome", label)
		}
		if saved.Status != "completed" {
			r.harnessFatalf(t, "saved action %s has invalid status %q", label, saved.Status)
		}
		stdout := filepath.Join(r.root, filepath.FromSlash(saved.Stdout))
		stderr := filepath.Join(r.root, filepath.FromSlash(saved.Stderr))
		if !fileExists(stdout) || !fileExists(stderr) {
			r.harnessFatalf(t, "saved action %s output is unavailable", label)
		}
		return remoteSourceCommandResult{ExitCode: saved.ExitCode, Stdout: stdout, Stderr: stderr, Duration: time.Duration(saved.DurationNS)}
	}
	writeJSON(t, receiptPath, remoteSourceSavedAction{
		Status: "started", Head: r.currentHead(t), Directory: dir, Command: append([]string(nil), argv...),
	})
	result := r.command(t, evidenceDir, label, timeout, dir, nil, argv...)
	stdout, stdoutErr := filepath.Rel(r.root, result.Stdout)
	stderr, stderrErr := filepath.Rel(r.root, result.Stderr)
	if stdoutErr != nil || stderrErr != nil || strings.HasPrefix(stdout, "..") || strings.HasPrefix(stderr, "..") {
		r.harnessFatalf(t, "action %s output is outside the run directory", label)
	}
	writeJSON(t, receiptPath, remoteSourceSavedAction{
		Status: "completed", Head: r.currentHead(t), Directory: dir, Command: append([]string(nil), argv...), ExitCode: result.ExitCode,
		Stdout: filepath.ToSlash(stdout), Stderr: filepath.ToSlash(stderr), DurationNS: int64(result.Duration),
	})
	return result
}

func (r *remoteSourceRun) actionReceiptPath(label string) string {
	return filepath.Join(r.root, "actions", r.currentStage, sanitizeRemoteSourceLabel(label)+".json")
}

func (r *remoteSourceRun) savedAction(label string) (remoteSourceSavedAction, bool, error) {
	path := r.actionReceiptPath(label)
	if !fileExists(path) {
		return remoteSourceSavedAction{}, false, nil
	}
	var saved remoteSourceSavedAction
	if err := decodeJSONFile(path, &saved); err != nil {
		return remoteSourceSavedAction{}, true, err
	}
	return saved, true, nil
}

func (r *remoteSourceRun) command(t *testing.T, evidenceDir, label string, timeout time.Duration, dir string, stdin io.Reader, argv ...string) remoteSourceCommandResult {
	t.Helper()
	process, err := r.startCommand(evidenceDir, label, timeout, dir, stdin, argv...)
	if err != nil {
		r.failureKind = "harness"
		t.Fatalf("start command %s: %v", label, err)
	}
	result, err := process.wait()
	if err != nil {
		r.failureKind = "harness"
		t.Fatalf("record command %s: %v", label, err)
	}
	return result
}

func (r *remoteSourceRun) startCommand(evidenceDir, label string, timeout time.Duration, dir string, stdin io.Reader, argv ...string) (*remoteSourceProcess, error) {
	if len(argv) == 0 {
		return nil, errors.New("command is empty")
	}
	if err := os.MkdirAll(evidenceDir, 0o700); err != nil {
		return nil, err
	}
	r.commandsMu.Lock()
	r.commandID++
	id := r.commandID
	r.commandsMu.Unlock()
	prefix := fmt.Sprintf("%03d-%s", id, sanitizeRemoteSourceLabel(label))
	stdoutPath := filepath.Join(evidenceDir, prefix+".stdout.log")
	stderrPath := filepath.Join(evidenceDir, prefix+".stderr.log")
	stdout, err := os.OpenFile(stdoutPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return nil, err
	}
	stderr, err := os.OpenFile(stderrPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		_ = stdout.Close()
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
	cmd.Dir = dir
	cmd.Env = mergeRemoteSourceEnv(r.commandEnv)
	cmd.Stdin = stdin
	cmd.Stdout = stdout
	cmd.Stderr = stderr
	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	cmd.Cancel = func() error {
		if cmd.Process == nil {
			return os.ErrProcessDone
		}
		err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
		if errors.Is(err, syscall.ESRCH) {
			return os.ErrProcessDone
		}
		return err
	}
	cmd.WaitDelay = 5 * time.Second
	started := time.Now()
	if err := cmd.Start(); err != nil {
		cancel()
		_ = stdout.Close()
		_ = stderr.Close()
		return nil, err
	}
	process := &remoteSourceProcess{
		cmd: cmd, ctx: ctx, cancel: cancel, stdout: stdout, stderr: stderr, stdoutPath: stdoutPath, stderrPath: stderrPath,
		receipt: filepath.Join(evidenceDir, prefix+".json"), argv: append([]string(nil), argv...), started: started,
	}
	r.processes = append(r.processes, process)
	return process, nil
}

func (p *remoteSourceProcess) wait() (remoteSourceCommandResult, error) {
	if p == nil {
		return remoteSourceCommandResult{}, errors.New("process is nil")
	}
	if p.waited {
		return remoteSourceCommandResult{}, errors.New("process was already waited")
	}
	p.waited = true
	runErr := p.cmd.Wait()
	finished := time.Now()
	contextErr := p.ctx.Err()
	p.cancel()
	closeErr := errors.Join(p.stdout.Close(), p.stderr.Close())
	exitCode := 0
	if runErr != nil {
		exitCode = -1
		var exitErr *exec.ExitError
		if errors.As(runErr, &exitErr) {
			exitCode = exitErr.ExitCode()
		}
	}
	receipt := remoteSourceCommandReceipt{
		Command: p.argv, StartedAt: p.started.UTC().Format(time.RFC3339Nano), FinishedAt: finished.UTC().Format(time.RFC3339Nano),
		DurationMS: finished.Sub(p.started).Milliseconds(), ExitCode: exitCode, Stdout: filepath.Base(p.stdoutPath), Stderr: filepath.Base(p.stderrPath),
	}
	if runErr != nil {
		receipt.Error = runErr.Error()
	}
	if contextErr != nil {
		receipt.Error = contextErr.Error()
	}
	data, marshalErr := json.MarshalIndent(receipt, "", "  ")
	if marshalErr == nil {
		data = append(data, '\n')
		marshalErr = os.WriteFile(p.receipt, data, 0o600)
	}
	return remoteSourceCommandResult{ExitCode: exitCode, Stdout: p.stdoutPath, Stderr: p.stderrPath, Duration: finished.Sub(p.started)}, errors.Join(closeErr, marshalErr)
}

func (r *remoteSourceRun) stopProcess(process *remoteSourceProcess) {
	if process == nil || process.waited {
		return
	}
	if process.cmd.Process != nil {
		_ = syscall.Kill(-process.cmd.Process.Pid, syscall.SIGINT)
	}
	done := make(chan struct{})
	go func() {
		_, _ = process.wait()
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		if process.cmd.Process != nil {
			_ = syscall.Kill(-process.cmd.Process.Pid, syscall.SIGKILL)
		}
		<-done
	}
}

func processRunning(process *remoteSourceProcess) bool {
	if process == nil || process.waited || process.cmd.Process == nil {
		return false
	}
	return process.cmd.Process.Signal(syscall.Signal(0)) == nil
}

func mergeRemoteSourceEnv(overrides map[string]string) []string {
	env := make([]string, 0, len(os.Environ())+len(overrides))
	for _, item := range os.Environ() {
		key, _, _ := strings.Cut(item, "=")
		if _, replaced := overrides[key]; !replaced {
			env = append(env, item)
		}
	}
	for key, value := range overrides {
		env = append(env, key+"="+value)
	}
	return env
}

func (r *remoteSourceRun) finalize(t *testing.T) {
	for _, process := range r.processes {
		r.stopProcess(process)
	}
	shouldDestroy := r.allPassed || enabled(os.Getenv(remoteSourceDestroy))
	metadataPath := filepath.Join(r.root, "default-804-session.json")
	if shouldDestroy && fileExists(metadataPath) {
		tunnelRuntime := filepath.Join(r.cp9Root, ".cp9", r.session, "tilt", "runtime")
		result := r.command(t, r.root, "cleanup", 45*time.Minute, r.originRoot, nil,
			filepath.Join(r.originRoot, "spec-tests", "scripts", "default-804-session.sh"), "cleanup", r.root)
		supervisors, supervisorErr := exactTunnelSupervisors(tunnelRuntime)
		if err := os.WriteFile(filepath.Join(r.root, "tunnel-supervisors-after-cleanup.txt"), []byte(supervisors), 0o600); err != nil {
			supervisorErr = errors.Join(supervisorErr, err)
		}
		if result.ExitCode == 0 && supervisorErr == nil && supervisors == "none\n" {
			r.cleanup = "pass"
			r.cleanupMessage = "exact session destroy, accepted cleanup handoff, generated worktree removal, and tunnel-supervisor absence recorded"
		} else {
			r.cleanup = "fail"
			r.cleanupMessage = fmt.Sprintf("cleanup exited %d or left tunnel evidence (%v); see %s and %s", result.ExitCode, supervisorErr, result.Stdout, result.Stderr)
			t.Errorf("%s", r.cleanupMessage)
		}
	} else if fileExists(metadataPath) {
		r.cleanup = "fail"
		r.cleanupMessage = "first-red scene preserved; set " + remoteSourceDestroy + "=1 only after disposition"
	} else {
		r.cleanup = "not_run"
		r.cleanupMessage = "no run-owned session metadata exists"
	}
	r.writeState(t)
	r.writeSummary(t)
}

func exactTunnelSupervisors(runtimeDir string) (string, error) {
	cmd := exec.Command("ps", "-ww", "-axo", "pid=,command=")
	data, err := cmd.Output()
	if err != nil {
		return "", err
	}
	var matches strings.Builder
	for _, line := range strings.Split(string(data), "\n") {
		if strings.Contains(line, filepath.Join(runtimeDir, "controller-tunnel-supervisor.sh")) || strings.Contains(line, filepath.Join(runtimeDir, "kube-api-tunnel-supervisor.sh")) {
			matches.WriteString(strings.TrimSpace(line))
			matches.WriteByte('\n')
		}
	}
	if matches.Len() == 0 {
		return "none\n", nil
	}
	return matches.String(), nil
}

func (r *remoteSourceRun) writeState(t *testing.T) {
	t.Helper()
	state := struct {
		Case             string                        `json:"case"`
		ExecutingHead    string                        `json:"executing_head"`
		Session          string                        `json:"session,omitempty"`
		Owner            string                        `json:"owner,omitempty"`
		Account          string                        `json:"account,omitempty"`
		Region           string                        `json:"region,omitempty"`
		CP9Root          string                        `json:"cp9_root,omitempty"`
		SourceRoot       string                        `json:"source_root,omitempty"`
		RemoteSourceRoot string                        `json:"remote_source_root,omitempty"`
		Stages           map[string]*remoteSourceStage `json:"stages"`
		Cleanup          string                        `json:"cleanup"`
		CleanupMessage   string                        `json:"cleanup_message,omitempty"`
	}{
		Case: remoteSourceCase, ExecutingHead: r.currentHead(t), Session: r.session,
		Owner: r.metadata.Owner, Account: r.metadata.Account, Region: r.metadata.Region,
		CP9Root: r.cp9Root, SourceRoot: r.sourceRoot, RemoteSourceRoot: r.remoteSourceRoot,
		Stages: r.stages, Cleanup: r.cleanup, CleanupMessage: r.cleanupMessage,
	}
	writeJSON(t, filepath.Join(r.root, "run-state.json"), state)
}

func (r *remoteSourceRun) loadState(t *testing.T) {
	t.Helper()
	var state struct {
		Case             string                        `json:"case"`
		ExecutingHead    string                        `json:"executing_head"`
		Session          string                        `json:"session"`
		CP9Root          string                        `json:"cp9_root"`
		SourceRoot       string                        `json:"source_root"`
		RemoteSourceRoot string                        `json:"remote_source_root"`
		Stages           map[string]*remoteSourceStage `json:"stages"`
	}
	decodeFile(t, filepath.Join(r.root, "run-state.json"), &state)
	if state.Case != remoteSourceCase {
		t.Fatalf("resume state belongs to %q, want %q", state.Case, remoteSourceCase)
	}
	if state.ExecutingHead != r.currentHead(t) {
		t.Fatalf("resume state head %s does not match executing head %s", state.ExecutingHead, r.currentHead(t))
	}
	for _, name := range []string{"input-refresh", "deploy-current-source", "network-resume"} {
		if state.Stages[name] == nil {
			t.Fatalf("resume state is missing stage %s", name)
		}
	}
	r.stages = state.Stages
	r.cp9Root = state.CP9Root
	r.sourceRoot = state.SourceRoot
	r.remoteSourceRoot = state.RemoteSourceRoot
	decodeFile(t, filepath.Join(r.root, "default-804-session.json"), &r.metadata)
	r.config = r.metadata.Config
	r.sourceFile = filepath.Join(r.sourceRoot, "cmd", "server", "main.go")
	r.session = state.Session
	r.prepared = true
}

func (r *remoteSourceRun) writeSummary(t *testing.T) {
	t.Helper()
	product := "pass"
	harness := "pass"
	for _, name := range []string{"input-refresh", "deploy-current-source", "network-resume"} {
		stage := r.stages[name]
		stage.Product, stage.Harness = remoteSourceStageVerdicts(stage)
		if stage.Product == "fail" {
			product = "fail"
		} else if stage.Product != "pass" && product != "fail" {
			product = "not_run"
		}
		if stage.Harness == "fail" {
			harness = "fail"
		} else if stage.Harness != "pass" && harness != "fail" {
			harness = "not_run"
		}
	}
	var summary strings.Builder
	summary.WriteString("# Remote source development session\n\n")
	fmt.Fprintf(&summary, "- Executing head: `%s`\n- Session: `%s`\n- Product: **%s**\n- Harness: **%s**\n- Cleanup: **%s** — %s\n\n", r.currentHead(t), r.session, product, harness, r.cleanup, r.cleanupMessage)
	summary.WriteString("| Phase | Executed | Product | Harness | Latest evidence |\n| --- | --- | --- | --- | --- |\n")
	for _, name := range []string{"input-refresh", "deploy-current-source", "network-resume"} {
		stage := r.stages[name]
		evidence := "—"
		if len(stage.Attempts) > 0 {
			evidence = stage.Attempts[len(stage.Attempts)-1].Evidence + "/"
		}
		fmt.Fprintf(&summary, "| %s | %t | %s | %s | %s |\n", name, stage.Executed, stage.Product, stage.Harness, evidence)
	}
	if err := os.WriteFile(filepath.Join(r.root, "summary.md"), []byte(summary.String()), 0o600); err != nil {
		t.Errorf("write run summary: %v", err)
	}
}

func remoteSourceStageVerdicts(stage *remoteSourceStage) (string, string) {
	if len(stage.Attempts) == 0 {
		return stage.Product, stage.Harness
	}
	product, harness := stage.Product, stage.Harness
	for _, attempt := range stage.Attempts {
		product = aggregateRemoteSourceLane(product, attempt.Product)
		harness = aggregateRemoteSourceLane(harness, attempt.Harness)
	}
	return product, harness
}

func aggregateRemoteSourceLane(current, next string) string {
	if current == "fail" || next == "fail" {
		return "fail"
	}
	if current == "pass" || next == "pass" {
		return "pass"
	}
	return "not_run"
}

func (r *remoteSourceRun) currentHead(t *testing.T) string {
	t.Helper()
	cmd := exec.Command("git", "-C", r.originRoot, "rev-parse", "HEAD")
	data, err := cmd.Output()
	if err != nil {
		t.Fatalf("resolve executing head: %v", err)
	}
	return strings.TrimSpace(string(data))
}

func (r *remoteSourceRun) harnessFatalf(t *testing.T, format string, args ...any) {
	t.Helper()
	r.failureKind = "harness"
	t.Fatalf(format, args...)
}

func (r *remoteSourceRun) productFatalf(t *testing.T, format string, args ...any) {
	t.Helper()
	r.failureKind = "product"
	t.Fatalf(format, args...)
}

func imageStamp(resource remoteSourceTiltResource) string {
	return resource.Metadata.Annotations["cp9.tidbcloud.com/image-stamp"]
}

func newBuild(after, before remoteSourceTiltResource) bool {
	if strings.ToLower(after.Status.UpdateStatus) != "ok" || len(after.Status.BuildHistory) == 0 || after.Status.BuildHistory[0].FinishTime == "" {
		return false
	}
	if len(before.Status.BuildHistory) == 0 {
		return true
	}
	return after.Status.BuildHistory[0].SpanID != before.Status.BuildHistory[0].SpanID || after.Status.BuildHistory[0].StartTime != before.Status.BuildHistory[0].StartTime
}

func sourceDeployOrder(before, after string) remoteSourceBuildOrder {
	type positions struct {
		ManualLiveBinary int
		LiveBinaryBuild  int
		AutomaticUpdate  int
	}
	count := func(text string) (remoteSourceBuildOrder, positions) {
		lines := strings.Split(text, "\n")
		var result remoteSourceBuildOrder
		last := positions{ManualLiveBinary: -1, LiveBinaryBuild: -1, AutomaticUpdate: -1}
		for index, line := range lines {
			end := min(index+4, len(lines))
			window := strings.Join(lines[index:end], "\n")
			if strings.Contains(line, "CLI Trigger") {
				if strings.Contains(window, "build-live-binary.sh") {
					result.ManualLiveBinaryTrigger++
					last.ManualLiveBinary = index
				} else if strings.Contains(window, "STEP 1/2 — Force update") {
					result.ManualImageTrigger++
				}
			}
			if strings.Contains(line, "[cp9 persona live-update] config-center: go build ./cmd/server") {
				result.LiveBinaryBuild++
				last.LiveBinaryBuild = index
			}
			if strings.Contains(line, "1 File Changed: [.cp9-build/config-center/live/service]") {
				result.AutomaticLiveUpdate++
				last.AutomaticUpdate = index
			}
		}
		return result, last
	}
	prior, _ := count(before)
	current, last := count(after)
	result := remoteSourceBuildOrder{
		ManualLiveBinaryTrigger:  current.ManualLiveBinaryTrigger - prior.ManualLiveBinaryTrigger,
		LiveBinaryBuild:          current.LiveBinaryBuild - prior.LiveBinaryBuild,
		AutomaticLiveUpdate:      current.AutomaticLiveUpdate - prior.AutomaticLiveUpdate,
		ManualImageTrigger:       current.ManualImageTrigger - prior.ManualImageTrigger,
		ManualLiveBinaryPosition: last.ManualLiveBinary,
		LiveBinaryBuildPosition:  last.LiveBinaryBuild,
		AutomaticUpdatePosition:  last.AutomaticUpdate,
	}
	result.Ordered = result.LiveBinaryBuildPosition >= 0 && result.LiveBinaryBuildPosition < result.AutomaticUpdatePosition
	if result.ManualLiveBinaryTrigger == 1 {
		result.Ordered = result.ManualLiveBinaryPosition >= 0 && result.ManualLiveBinaryPosition < result.LiveBinaryBuildPosition && result.Ordered
	}
	return result
}

func fileSHA256(t *testing.T, path string) string {
	t.Helper()
	file, err := os.Open(path)
	if err != nil {
		t.Fatalf("open %s for hashing: %v", path, err)
	}
	defer func() { _ = file.Close() }()
	hash := sha256.New()
	if _, err := io.Copy(hash, file); err != nil {
		t.Fatalf("hash %s: %v", path, err)
	}
	return hex.EncodeToString(hash.Sum(nil))
}

func bytesSHA256(data []byte) string {
	hash := sha256.Sum256(data)
	return hex.EncodeToString(hash[:])
}

func writeRandomFile(path string, size int64) error {
	file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return err
	}
	_, copyErr := io.CopyN(file, rand.Reader, size)
	return errors.Join(copyErr, file.Close())
}

func appendFile(path, value string) error {
	file, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
	if err != nil {
		return err
	}
	_, writeErr := file.WriteString(value)
	return errors.Join(writeErr, file.Close())
}

func decodeFile(t *testing.T, path string, target any) {
	t.Helper()
	if err := decodeJSONFile(path, target); err != nil {
		t.Fatalf("decode %s: %v", path, err)
	}
}

func decodeJSONFile(path string, target any) error {
	file, err := os.Open(path)
	if err != nil {
		return err
	}
	defer func() { _ = file.Close() }()
	return json.NewDecoder(file).Decode(target)
}

func writeJSON(t *testing.T, path string, value any) {
	t.Helper()
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		t.Fatalf("create parent for %s: %v", path, err)
	}
	data, err := json.MarshalIndent(value, "", "  ")
	if err != nil {
		t.Fatalf("encode %s: %v", path, err)
	}
	data = append(data, '\n')
	if err := os.WriteFile(path, data, 0o600); err != nil {
		t.Fatalf("write %s: %v", path, err)
	}
}

func readFile(t *testing.T, path string) string {
	t.Helper()
	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("read %s: %v", path, err)
	}
	return string(data)
}

func fileExists(path string) bool {
	_, err := os.Stat(path)
	return err == nil
}

func sanitizeRemoteSourceLabel(label string) string {
	var result strings.Builder
	for _, value := range strings.ToLower(label) {
		if value >= 'a' && value <= 'z' || value >= '0' && value <= '9' || value == '-' {
			result.WriteRune(value)
		} else {
			result.WriteByte('-')
		}
	}
	return strings.Trim(result.String(), "-")
}

func waitForLineCount(path string, count int, timeout time.Duration, alive func() bool) bool {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		file, err := os.Open(path)
		if err == nil {
			scanner := bufio.NewScanner(file)
			lines := 0
			for scanner.Scan() {
				lines++
			}
			_ = file.Close()
			if lines >= count {
				return true
			}
		}
		if !alive() {
			return false
		}
		time.Sleep(time.Second)
	}
	return false
}

func waitForWatchEvent(path string, timeout time.Duration, accept func(remoteSourceWatchEvent) bool, alive func() bool) bool {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		file, err := os.Open(path)
		if err == nil {
			scanner := bufio.NewScanner(file)
			for scanner.Scan() {
				var event remoteSourceWatchEvent
				if json.Unmarshal(scanner.Bytes(), &event) == nil && accept(event) {
					_ = file.Close()
					return true
				}
			}
			_ = file.Close()
		}
		if !alive() {
			return false
		}
		time.Sleep(time.Second)
	}
	return false
}

func waitForTransferProcess(parentPID int, remoteRoot string, timeout time.Duration) (int, string, error) {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		pgrep := exec.Command("pgrep", "-P", strconv.Itoa(parentPID), "-x", "rsync")
		data, _ := pgrep.Output()
		for _, field := range strings.Fields(string(data)) {
			pid, err := strconv.Atoi(field)
			if err != nil {
				continue
			}
			ps := exec.Command("ps", "-ww", "-p", field, "-o", "command=")
			command, err := ps.Output()
			if err == nil && strings.Contains(string(command), remoteRoot+".cp9-incoming/") {
				return pid, strings.TrimSpace(string(command)), nil
			}
		}
		time.Sleep(100 * time.Millisecond)
	}
	return 0, "", errors.New("timed out waiting for source rsync")
}

func readProbeTiming(t *testing.T, path string) remoteSourceProbeTiming {
	t.Helper()
	var raw struct {
		ActualNS  int64 `json:"actual_ns"`
		PaddingNS int64 `json:"padding_ns"`
		TotalNS   int64 `json:"total_ns"`
	}
	decodeFile(t, path, &raw)
	return remoteSourceProbeTiming{Actual: time.Duration(raw.ActualNS), Padding: time.Duration(raw.PaddingNS), Total: time.Duration(raw.TotalNS)}
}

func runRemoteSourceSSHHelper() int {
	realSSH := os.Getenv(remoteSourceSSHReal)
	modePath := os.Getenv(remoteSourceSSHMode)
	timingDir := os.Getenv(remoteSourceSSHTimings)
	if realSSH == "" || modePath == "" || timingDir == "" {
		fmt.Fprintln(os.Stderr, "remote source SSH helper environment is incomplete")
		return 64
	}
	modeData, _ := os.ReadFile(modePath)
	mode := strings.TrimSpace(string(modeData))
	if mode == "network" {
		fmt.Fprintln(os.Stderr, "cp9 spec: scoped network fault is active")
		return 255
	}
	arguments := strings.Join(os.Args[1:], " ")
	probe := ""
	if strings.Contains(arguments, "--property=LoadState,ActiveState,SubState,Result,MainPID") {
		probe = "observation"
	} else if strings.Contains(arguments, "uiresources") {
		probe = "readiness"
	} else if strings.Contains(arguments, "session/reports/evidence") && strings.Contains(arguments, "tar -C") {
		probe = "artifact-sync"
	}
	if mode == "stall-readiness" && probe == "readiness" || mode == "stall-artifact-sync" && probe == "artifact-sync" {
		time.Sleep(45 * time.Second)
		return 0
	}
	started := time.Now()
	command := exec.Command(realSSH, os.Args[1:]...)
	command.Stdin = os.Stdin
	command.Stdout = os.Stdout
	command.Stderr = os.Stderr
	err := command.Run()
	actual := time.Since(started)
	padding := time.Duration(0)
	if mode == "delay" && (probe == "observation" || probe == "readiness") && actual < 16*time.Second {
		padding = 16*time.Second - actual
		time.Sleep(padding)
	}
	total := time.Since(started)
	if probe != "" && mode == "delay" {
		_ = os.MkdirAll(timingDir, 0o700)
		data, marshalErr := json.MarshalIndent(struct {
			ActualNS  int64 `json:"actual_ns"`
			PaddingNS int64 `json:"padding_ns"`
			TotalNS   int64 `json:"total_ns"`
		}{ActualNS: int64(actual), PaddingNS: int64(padding), TotalNS: int64(total)}, "", "  ")
		if marshalErr == nil {
			data = append(data, '\n')
			_ = os.WriteFile(filepath.Join(timingDir, probe+".json"), data, 0o600)
		}
	}
	if err == nil {
		return 0
	}
	var exitErr *exec.ExitError
	if errors.As(err, &exitErr) {
		return exitErr.ExitCode()
	}
	fmt.Fprintln(os.Stderr, err)
	return 1
}