//go:build functional

package spectests_test

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"sort"
	"strconv"
	"strings"
	"syscall"
	"testing"
	"time"

	"gopkg.in/yaml.v3"
)

const (
	sharedSourceEnabledEnv    = "CP9_SOURCE_BUILD_SHARED_REPO_ENABLED"
	sharedSourceRunDirEnv     = "CP9_SOURCE_BUILD_SHARED_REPO_RUN_DIR"
	sharedSourceRetryStageEnv = "CP9_SOURCE_BUILD_SHARED_REPO_RETRY_STAGE"
	sharedSourceDestroyEnv    = "CP9_SOURCE_BUILD_SHARED_REPO_DESTROY_SESSION"
	sharedSourceJournalFile   = "source-build-shared-repo-state.json"
	sharedSourceJournalV1     = 1
)

var sharedSourceProductStages = []string{
	"source-identities",
	"bounded-parallel-builds",
	"deploy-config-center",
	"deploy-config-center-regional",
}

type sharedSourceRun struct {
	repoRoot     string
	runDir       string
	currentHead  string
	retryStage   string
	metadata     sharedSourceMetadata
	journal      sharedSourceJournal
	commandCount int
}

type sharedSourceJournal struct {
	Version   int                                   `json:"version"`
	Case      string                                `json:"case"`
	CP9Head   string                                `json:"cp9_head,omitempty"`
	Session   string                                `json:"session,omitempty"`
	Cleanup   string                                `json:"cleanup"`
	Stages    map[string][]sharedSourceStageReceipt `json:"stages"`
	Actions   map[string]sharedSourceActionReceipt  `json:"actions"`
	CreatedAt string                                `json:"created_at"`
	UpdatedAt string                                `json:"updated_at"`
}

type sharedSourceStageReceipt struct {
	Head       string `json:"head"`
	StartedAt  string `json:"started_at"`
	FinishedAt string `json:"finished_at,omitempty"`
	Status     string `json:"status"`
	Product    string `json:"product"`
	Harness    string `json:"harness"`
	Message    string `json:"message,omitempty"`
}

type sharedSourceActionReceipt struct {
	Head       string `json:"head"`
	StartedAt  string `json:"started_at"`
	FinishedAt string `json:"finished_at,omitempty"`
	Status     string `json:"status"`
}

type sharedSourceMetadata struct {
	Case      string `json:"case"`
	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 sharedSourceDryRun struct {
	NoMutation bool                     `json:"no_mutation"`
	Session    string                   `json:"session"`
	Owner      string                   `json:"owner"`
	TTL        string                   `json:"ttl"`
	AWS        sharedSourceDryRunAWS    `json:"aws"`
	Services   []sharedSourceDryService `json:"services"`
}

type sharedSourceDryRunAWS struct {
	Account string `json:"account"`
	Region  string `json:"region"`
}

type sharedSourceDryService struct {
	Name              string `json:"name"`
	Role              string `json:"role"`
	ConfiguredPath    string `json:"configured_path"`
	CanonicalPath     string `json:"canonical_path"`
	GitHEAD           string `json:"git_head"`
	Dirty             bool   `json:"dirty"`
	BuildMain         string `json:"build_main"`
	TiltResource      string `json:"tilt_resource"`
	WorkloadKind      string `json:"workload_kind"`
	WorkloadNamespace string `json:"workload_namespace"`
	WorkloadName      string `json:"workload_name"`
	WorkloadContainer string `json:"workload_container"`
}

type sharedSourceStatus struct {
	Session string `json:"session_name"`
	Status  string `json:"status"`
	Layers  struct {
		Host struct {
			Execution struct {
				SourceSyncDestination string `json:"source_sync_destination"`
			} `json:"execution"`
		} `json:"host"`
	} `json:"layers"`
}

type sharedSourceUIResource struct {
	Metadata struct {
		Name string `json:"name"`
	} `json:"metadata"`
	Status struct {
		RuntimeStatus string                     `json:"runtimeStatus"`
		UpdateStatus  string                     `json:"updateStatus"`
		TriggerMode   int                        `json:"triggerMode"`
		BuildHistory  []sharedSourceBuildHistory `json:"buildHistory"`
		Conditions    []struct {
			Type   string `json:"type"`
			Status string `json:"status"`
		} `json:"conditions"`
	} `json:"status"`
}

type sharedSourceBuildHistory struct {
	SpanID     string `json:"spanID"`
	StartTime  string `json:"startTime"`
	FinishTime string `json:"finishTime"`
}

type sharedSourceBuildInterval struct {
	Resource string    `json:"resource"`
	SpanID   string    `json:"span_id"`
	Start    time.Time `json:"start"`
	Finish   time.Time `json:"finish"`
}

type sharedSourceCommandReceipt struct {
	Command    []string `json:"command"`
	Directory  string   `json:"directory"`
	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 sharedSourceCommandResult struct {
	ExitCode int
	Stdout   string
	Stderr   string
}

type sharedSourceFileIdentity struct {
	Kind   string `json:"kind"`
	Digest string `json:"digest,omitempty"`
	Target string `json:"target,omitempty"`
}

type sharedSourceWorkload struct {
	Namespace          string `json:"namespace"`
	Name               string `json:"name"`
	UID                string `json:"uid"`
	Generation         int64  `json:"generation"`
	ObservedGeneration int64  `json:"observed_generation"`
	Replicas           int32  `json:"replicas"`
	ReadyReplicas      int32  `json:"ready_replicas"`
	AvailableReplicas  int32  `json:"available_replicas"`
	Container          string `json:"container"`
	Image              string `json:"image"`
}

// # Source build maps one repository to several services with bounded parallelism
//
// This case proves one user journey: a user selects two Tilt services from one
// Git repository plus a service from another repository, then `cp9 up` preserves
// all mappings while enforcing the configured source-build concurrency limit.
//
// It covers issues [#977](https://github.com/tidbcloud/controlplane9/issues/977),
// [#1000](https://github.com/tidbcloud/controlplane9/issues/1000), and
// [#1005](https://github.com/tidbcloud/controlplane9/issues/1005), with fixes in
// PRs [#969](https://github.com/tidbcloud/controlplane9/pull/969),
// [#983](https://github.com/tidbcloud/controlplane9/pull/983), and
// [#1010](https://github.com/tidbcloud/controlplane9/pull/1010).
//
// ## Agent execution
//
// **Mode: single agent.** Spawn no sub-agents. One parent owns the session,
// observes the three selected builds, runs both deploys, decides the verdict, and
// cleans up. Tilt creates the concurrency under test; agent concurrency would add
// no independent work and would complicate the one-session mutation boundary.
//
// Native Go functional entrypoint:
//
// ```text
// CP9_SOURCE_BUILD_SHARED_REPO_ENABLED=1 \
// CP9_UI_API_AWS_KEY_PAIR=<existing-key-pair> \
// CP9_UI_API_SSH_PRIVATE_KEY=<matching-private-key-file> \
// GOTOOLCHAIN=go1.26.1 GO_FUNCTIONAL_TEST_PACKAGES=./spec-tests \
// GO_FUNCTIONAL_TEST_RUN='^TestFunctional_SourceBuildSharedRepo$$' make test-functional
// ```
//
// The opt-in is checked before any build, setup, cloud read, or session mutation.
// To continue a preserved run, also set
// `CP9_SOURCE_BUILD_SHARED_REPO_RUN_DIR=<absolute-run-evidence-directory>` and
// select the next safe subtest. A failed stage is never retried implicitly; after
// same-scene diagnosis, set `CP9_SOURCE_BUILD_SHARED_REPO_RETRY_STAGE` to that
// stage only when its journal shows that the product action did not start or
// completed; an action left in `started` state is never replayed automatically.
// The summary retains every attempt and a Product or Harness `FAIL` remains sticky
// after repaired evidence passes.
//
// After an explicit reviewer or owner disposition, a retained failed scene can be
// cleaned without making failed Product stages pass:
//
// ```text
// CP9_SOURCE_BUILD_SHARED_REPO_ENABLED=1 \
// CP9_SOURCE_BUILD_SHARED_REPO_DESTROY_SESSION=1 \
// CP9_SOURCE_BUILD_SHARED_REPO_RUN_DIR=<absolute-run-evidence-directory> \
// GOTOOLCHAIN=go1.26.1 GO_FUNCTIONAL_TEST_PACKAGES=./spec-tests \
// GO_FUNCTIONAL_TEST_RUN='^TestFunctional_SourceBuildSharedRepo$$/cleanup$$' make test-functional
// ```
//
// ## Default run
//
// - AWS account `804837308162`, region `us-west-2`, source profile
//   `dev-offline-afterworld`, role `cp9-developer`, and source-build parallelism
//   `2`.
// - A fresh 10-hour `remote-ec2-k3d` session with a dynamically selected public
//   subnet and a caller-owned existing EC2 key pair plus matching private key
//   supplied through `CP9_UI_API_AWS_KEY_PAIR` and `CP9_UI_API_SSH_PRIVATE_KEY`.
// - An isolated exact-head CP9 worktree, one clean `tidbcloud/config-center`
//   checkout shared by `config-center` and `config-center-regional`, and one
//   clean `tidbcloud/cluster-service-ng` checkout for `cluster-ng-server`.
// - The instruction to execute this case authorizes session setup, both targeted
//   deploys, and exact-session cleanup. Do not ask again.
//
// The machine needs `gh`, `jq`, `python3`, `ssh`, and AWS CLI. GitHub Device Flow
// is the only possible human step during the source builds.
//
// ## Execute
//
// `TestFunctional_SourceBuildSharedRepo` owns the ordered phase control, parses
// the structured source identities and timestamped build histories, preserves a
// first-red journal, and records exact cleanup. It reuses the existing default
// 804 session preparation and remote Tilt transport helpers; there is no second
// shell implementation of this journey.
//
// ### 1. Freeze and inspect the selected sources
//
// Use this test's existing harness or an equivalent method. Freeze the selected CP9,
// `config-center`, and `cluster-service-ng` revisions and create one fresh
// session. Before mutation:
//
// - require clean selected source checkouts;
// - build CP9, run setup and doctor, and confirm the plan uses the declared
//   account, region, session, TTL, and cleanup scope;
// - require the three selected source-built services:
//   `config-center`, `config-center-regional`, and `cluster-ng-server`;
// - require every service's planned source identity to match its selected clean
//   checkout;
// - require the two `config-center` services to share one complete Git root and
//   revision while keeping distinct service and workload identities;
// - fail closed when any configured path is unreadable;
// - arrange exact-session cleanup for every exit path.
//
// ### 2. Prove bounded parallel source builds
//
// Bring the session to `ready`. Observe the start and completion of each selected
// source build and require no more than the configured limit of two to run at
// once. Unrelated Tilt graph work may still run in parallel. The executor or
// harness chooses how to locate the builds and calculate overlap.
//
// Both services from `config-center` must preserve their separate
// resource/workload mappings while using one complete remote Git root. The
// harness compares every locally tracked regular file and symlink in both source
// repositories with the actual remote sync destinations. A missing or mismatched
// tracked source path, missing sibling source, or selected build that never
// completes is `FAIL`.
//
// ### 3. Prove both shared-root services remain deployable
//
// Deploy `config-center`, then `config-center-regional`, against the same
// session.
//
// The first deploy must refresh the shared root and update only the selected
// service resource. The sibling must remain present and healthy, and the second
// targeted deploy must also succeed. Before and after each deploy, query the
// sibling's exact Deployment from the frozen dry-run mapping and require its
// namespace, name, UID, generation, replica count, selected container, and image
// to remain unchanged and fully Ready/Available. Deleting, hiding, or remapping
// the sibling is `FAIL`.
//
// ### 4. Clean up
//
// Destroy only the recorded session and isolated worktree, and record cleanup
// receipts. The caller-owned key pair is outside CP9 cleanup; no key-pair
// ownership or deletion receipt is expected. A completed run requires passing
// destroy and either a passing or unnecessary worktree receipt.
//
// ## Verdicts
//
// ### Product
//
// `PASS` requires exact dry-run provenance, fail-closed unreadable-path handling,
// one complete shared Git root, three distinct service mappings, at most two
// concurrent source builds, top-level `ready`, and two successful targeted
// deploys without deleting or remapping the sibling service. Missing sibling
// source, more than two directly observed overlapping builds, a selected build
// that never completes, or a deleted or remapped sibling is `FAIL`.
//
// Expired `dev-offline-afterworld` login, missing tool, inaccessible source
// repository, or unavailable Device Flow human is `NOT RUN` for the affected
// assertion. Missing or unparseable observation data leaves only the affected
// product assertion `NOT RUN` unless direct behavior disproves it.
//
// ### Harness
//
// Harness `PASS` requires the chosen source-identity, build-overlap, and deploy
// observations to complete correctly. A locator, parser, interval calculation,
// script, or evidence-capture defect is harness `FAIL`, not product `FAIL`.
// Preserve the current session and direct observations, and mark only
// still-unobserved product assertions `NOT RUN`. Harness is `NOT RUN` only when a
// prerequisite stops the case before any probe begins.
//
// ### Cleanup
//
// Cleanup `PASS` requires the exact session and isolated worktree removed with the
// required receipts recorded. The caller-owned key pair is not a run-owned
// resource and must not be deleted. 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_SourceBuildSharedRepo(t *testing.T) {
	if os.Getenv(sharedSourceEnabledEnv) != "1" {
		t.Skipf("set %s=1 to run the shared-repository source-build journey", sharedSourceEnabledEnv)
	}

	run := newSharedSourceRun(t)
	t.Cleanup(func() { run.writeSummary(t) })

	t.Run("source-identities", func(t *testing.T) {
		stage := run.beginStage(t, "source-identities", true)
		defer run.finishStage(t, stage)
		run.assertSourceIdentities(t, stage)
	})

	t.Run("bounded-parallel-builds", func(t *testing.T) {
		run.requirePassed(t, "source-identities")
		stage := run.beginStage(t, "bounded-parallel-builds", run.actionReplaySafe("cp9-up"))
		defer run.finishStage(t, stage)
		run.assertBoundedParallelBuilds(t, stage)
	})

	t.Run("deploy-config-center", func(t *testing.T) {
		run.requirePassed(t, "bounded-parallel-builds")
		stage := run.beginStage(t, "deploy-config-center", run.actionReplaySafe("deploy-config-center"))
		defer run.finishStage(t, stage)
		run.assertTargetedDeploy(t, stage, "config-center", "config-center-regional")
	})

	t.Run("deploy-config-center-regional", func(t *testing.T) {
		run.requirePassed(t, "deploy-config-center")
		stage := run.beginStage(t, "deploy-config-center-regional", run.actionReplaySafe("deploy-config-center-regional"))
		defer run.finishStage(t, stage)
		run.assertTargetedDeploy(t, stage, "config-center-regional", "config-center")
	})

	t.Run("cleanup", func(t *testing.T) {
		if os.Getenv(sharedSourceDestroyEnv) != "1" {
			for _, dependency := range sharedSourceProductStages {
				run.requirePassed(t, dependency)
			}
		} else if run.journal.Session == "" {
			t.Skip("no run-owned session or worktree to clean")
		}
		if run.journal.Cleanup == "pass" {
			t.Skip("reuse passing exact-session cleanup receipt")
		}
		run.cleanup(t)
	})
}

func newSharedSourceRun(t *testing.T) *sharedSourceRun {
	t.Helper()
	wd, err := os.Getwd()
	if err != nil {
		t.Fatalf("resolve spec-tests working directory: %v", err)
	}
	repoRoot := filepath.Dir(wd)
	if _, err := os.Stat(filepath.Join(repoRoot, "go.mod")); err != nil {
		t.Fatalf("resolve repository root from %s: %v", wd, err)
	}
	currentHead := currentCommit(t, repoRoot)
	root := filepath.Join(repoRoot, "artifacts", "spec-tests")
	if err := os.MkdirAll(root, 0o700); err != nil {
		t.Fatalf("create spec evidence root: %v", err)
	}

	runDir := strings.TrimSpace(os.Getenv(sharedSourceRunDirEnv))
	if runDir == "" {
		runDir = filepath.Join(root, time.Now().UTC().Format("20060102T150405Z")+"-shared-repo-parallel-build")
		if err := os.Mkdir(runDir, 0o700); err != nil {
			t.Fatalf("create shared-source evidence directory %s: %v", runDir, err)
		}
	} else {
		runDir, err = filepath.Abs(runDir)
		if err != nil {
			t.Fatalf("resolve %s: %v", sharedSourceRunDirEnv, err)
		}
		canonicalRoot, rootErr := filepath.EvalSymlinks(root)
		canonicalRun, runErr := filepath.EvalSymlinks(runDir)
		if rootErr != nil || runErr != nil || !sharedSourceChildPath(canonicalRoot, canonicalRun) {
			t.Fatalf("%s must name an existing child of %s", sharedSourceRunDirEnv, root)
		}
	}

	run := &sharedSourceRun{
		repoRoot:     repoRoot,
		runDir:       runDir,
		currentHead:  currentHead,
		retryStage:   strings.TrimSpace(os.Getenv(sharedSourceRetryStageEnv)),
		commandCount: sharedSourcePriorCommandCount(runDir),
	}
	journalPath := filepath.Join(runDir, sharedSourceJournalFile)
	data, err := os.ReadFile(journalPath)
	if errors.Is(err, os.ErrNotExist) {
		now := time.Now().UTC().Format(time.RFC3339Nano)
		run.journal = sharedSourceJournal{
			Version:   sharedSourceJournalV1,
			Case:      "source-build-shared-repo-and-parallelism",
			Cleanup:   "not_run",
			Stages:    map[string][]sharedSourceStageReceipt{},
			Actions:   map[string]sharedSourceActionReceipt{},
			CreatedAt: now,
			UpdatedAt: now,
		}
		if err := run.persistJournal(); err != nil {
			t.Fatalf("create shared-source journal: %v", err)
		}
	} else if err != nil {
		t.Fatalf("read shared-source journal %s: %v", journalPath, err)
	} else {
		if err := json.Unmarshal(data, &run.journal); err != nil {
			t.Fatalf("decode shared-source journal %s: %v", journalPath, err)
		}
		if run.journal.Version != sharedSourceJournalV1 || run.journal.Case != "source-build-shared-repo-and-parallelism" ||
			run.journal.Stages == nil || run.journal.Actions == nil {
			t.Fatalf("shared-source journal %s has an incompatible identity or shape", journalPath)
		}
	}
	if _, err := os.Stat(filepath.Join(runDir, "default-804-session.json")); err == nil {
		run.loadMetadata(t)
	} else if !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("inspect shared-source metadata: %v", err)
	}
	return run
}

func (r *sharedSourceRun) assertSourceIdentities(t *testing.T, stage *sharedSourceStageReceipt) {
	t.Helper()
	if dirty := r.gitOutput(t, stage, "cp9-source-status", r.repoRoot, "status", "--porcelain"); dirty != "" {
		r.prerequisiteFatalf(t, stage, "CP9 checkout must be clean before freezing the journey")
	}
	r.prepare(t, stage)
	r.mustPrerequisiteCommand(t, stage, "build-cp9", 30*time.Minute, r.metadata.CP9Root, nil,
		"make", "build")
	r.mustPrerequisiteCommand(t, stage, "setup", 30*time.Minute, r.metadata.CP9Root, nil,
		filepath.Join(r.metadata.CP9Root, "bin", "cp9"), "setup", "checkout=false", "config="+r.metadata.Config)
	r.mustPrerequisiteCommand(t, stage, "doctor", 15*time.Minute, r.metadata.CP9Root, nil,
		filepath.Join(r.metadata.CP9Root, "bin", "cp9"), "doctor", "config="+r.metadata.Config)
	if head := r.gitOutput(t, stage, "cp9-product-head", r.metadata.CP9Root, "rev-parse", "HEAD"); head != r.metadata.Head {
		r.harnessFatalf(t, stage, "prepared CP9 worktree head=%s, want %s", head, r.metadata.Head)
	}

	r.assertUnreadableSourceRejected(t, stage)
	result := r.mustProductCommand(t, stage, "dry-run", 15*time.Minute, r.metadata.CP9Root, nil,
		filepath.Join(r.metadata.CP9Root, "bin", "cp9"), "{sys.output.format=json}", "up", "dry-run=true", "config="+r.metadata.Config)
	data := r.readCommandStdout(t, stage, result)
	r.writeEvidence(t, stage, "dry-run.json", data)
	var dryRun sharedSourceDryRun
	if err := json.Unmarshal(data, &dryRun); err != nil {
		r.harnessFatalf(t, stage, "parse successful dry-run JSON: %v", err)
	}
	r.assertDryRun(t, stage, dryRun)
	if _, err := os.Stat(filepath.Join(r.metadata.CP9Root, ".cp9", r.metadata.Session)); !errors.Is(err, os.ErrNotExist) {
		r.productFatalf(t, stage, "dry-run created session state at .cp9/%s", r.metadata.Session)
	}
}

func (r *sharedSourceRun) prepare(t *testing.T, stage *sharedSourceStageReceipt) {
	t.Helper()
	if r.metadata.CP9Root != "" {
		return
	}
	for _, tool := range []string{"aws", "bash", "gh", "git", "jq", "make", "python3", "ssh"} {
		if _, err := exec.LookPath(tool); err != nil {
			r.prerequisiteFatalf(t, stage, "required tool %s is unavailable: %v", tool, err)
		}
	}
	keyPair := strings.TrimSpace(os.Getenv("CP9_UI_API_AWS_KEY_PAIR"))
	privateKey := strings.TrimSpace(os.Getenv("CP9_UI_API_SSH_PRIVATE_KEY"))
	if keyPair == "" || privateKey == "" {
		r.prerequisiteFatalf(t, stage, "CP9_UI_API_AWS_KEY_PAIR and CP9_UI_API_SSH_PRIVATE_KEY are required together")
	}
	if info, err := os.Stat(privateKey); err != nil || !info.Mode().IsRegular() {
		r.prerequisiteFatalf(t, stage, "CP9_UI_API_SSH_PRIVATE_KEY must name a readable regular file")
	}
	helper := filepath.Join(r.repoRoot, "spec-tests", "scripts", "default-804-session.sh")
	r.mustPrerequisiteCommand(t, stage, "prepare-default-session", 20*time.Minute, r.repoRoot, nil,
		"bash", helper, "prepare", "parallel", r.runDir)
	r.loadMetadata(t)
	if r.metadata.Head != r.currentHead {
		r.harnessFatalf(t, stage, "prepared CP9 head=%s, want current %s", r.metadata.Head, r.currentHead)
	}
}

func (r *sharedSourceRun) loadMetadata(t *testing.T) {
	t.Helper()
	data, err := os.ReadFile(filepath.Join(r.runDir, "default-804-session.json"))
	if err != nil {
		t.Fatalf("read shared-source session metadata: %v", err)
	}
	if err := json.Unmarshal(data, &r.metadata); err != nil {
		t.Fatalf("decode shared-source session metadata: %v", err)
	}
	if r.metadata.Case != "parallel" || r.metadata.Account != "804837308162" ||
		r.metadata.Region != "us-west-2" || r.metadata.Profile != "dev-offline-afterworld" ||
		r.metadata.Session == "" || r.metadata.Owner == "" || r.metadata.TTL == "" ||
		r.metadata.Head == "" || r.metadata.CP9Root == "" || r.metadata.WorkRoot == "" ||
		r.metadata.Config == "" || r.metadata.OriginCP9 != r.repoRoot {
		t.Fatalf("shared-source session metadata has an unexpected identity or incomplete scope")
	}
	if !sharedSourceChildPath(r.metadata.WorkRoot, r.metadata.CP9Root) {
		t.Fatalf("CP9 worktree %s is outside work root %s", r.metadata.CP9Root, r.metadata.WorkRoot)
	}
	if r.journal.Session != "" && (r.journal.Session != r.metadata.Session || r.journal.CP9Head != r.metadata.Head) {
		t.Fatalf("shared-source journal identity does not match session metadata")
	}
	r.journal.Session = r.metadata.Session
	r.journal.CP9Head = r.metadata.Head
	if r.journal.Cleanup != "pass" {
		r.journal.Cleanup = "fail"
	}
	if err := r.persistJournal(); err != nil {
		t.Fatalf("record shared-source session identity: %v", err)
	}
}

func (r *sharedSourceRun) assertUnreadableSourceRejected(t *testing.T, stage *sharedSourceStageReceipt) {
	t.Helper()
	stateDir := filepath.Join(r.metadata.CP9Root, ".cp9", r.metadata.Session)
	if _, err := os.Stat(stateDir); !errors.Is(err, os.ErrNotExist) {
		r.harnessFatalf(t, stage, "cannot run pre-mutation negative check: session state already exists at %s", stateDir)
	}
	source := filepath.Join(r.metadata.WorkRoot, "cluster-service-ng")
	hidden := source + ".unreadable"
	if err := os.Rename(source, hidden); err != nil {
		r.harnessFatalf(t, stage, "hide cluster-service-ng checkout for negative check: %v", err)
	}
	restored := false
	defer func() {
		if restored {
			return
		}
		if err := os.Rename(hidden, source); err != nil {
			stage.Harness = "fail"
			t.Errorf("restore cluster-service-ng checkout after negative check: %v", err)
		}
	}()
	result, err := r.command("reject-unreadable-source", 5*time.Minute, r.metadata.CP9Root, nil,
		filepath.Join(r.metadata.CP9Root, "bin", "cp9"), "{sys.output.format=json}", "up", "dry-run=true", "config="+r.metadata.Config)
	if err != nil {
		r.harnessFatalf(t, stage, "run unreadable-source negative check: %v", err)
	}
	if result.ExitCode == 0 {
		r.productFatalf(t, stage, "dry-run accepted missing cluster-service-ng checkout")
	}
	combined := append(r.readFile(t, stage, result.Stdout), r.readFile(t, stage, result.Stderr)...)
	if !strings.Contains(string(combined), "cluster-ng-server") ||
		!strings.Contains(string(combined), "not a readable canonical path") {
		r.productFatalf(t, stage, "unreadable-source rejection did not identify cluster-ng-server and its unreadable source")
	}
	if _, err := os.Stat(stateDir); !errors.Is(err, os.ErrNotExist) {
		r.productFatalf(t, stage, "rejected dry-run created session state at %s", stateDir)
	}
	if err := os.Rename(hidden, source); err != nil {
		r.harnessFatalf(t, stage, "restore cluster-service-ng checkout: %v", err)
	}
	restored = true
}

func (r *sharedSourceRun) assertDryRun(t *testing.T, stage *sharedSourceStageReceipt, dryRun sharedSourceDryRun) {
	t.Helper()
	configData := r.readFile(t, stage, r.metadata.Config)
	var declared struct {
		Services struct {
			SourceBuildParallelism int `yaml:"source_build_parallelism"`
		} `yaml:"services"`
	}
	if err := yaml.Unmarshal(configData, &declared); err != nil {
		r.harnessFatalf(t, stage, "parse generated session config: %v", err)
	}
	if declared.Services.SourceBuildParallelism != 2 {
		r.harnessFatalf(t, stage, "generated source-build parallelism=%d, want 2", declared.Services.SourceBuildParallelism)
	}
	if !dryRun.NoMutation || dryRun.Session != r.metadata.Session || dryRun.Owner != r.metadata.Owner ||
		dryRun.AWS.Account != r.metadata.Account || dryRun.AWS.Region != r.metadata.Region || dryRun.TTL == "" {
		r.productFatalf(t, stage, "dry-run scope does not match the recorded no-mutation session scope")
	}

	type expectedIdentity struct {
		repository string
		main       string
		tilt       string
		namespace  string
		workload   string
		container  string
	}
	expected := map[string]expectedIdentity{
		"config-center": {
			repository: "config-center", main: "cmd/server/main.go", tilt: "config-center-image",
			namespace: "cp9-global", workload: "config-center", container: "config-center",
		},
		"config-center-regional": {
			repository: "config-center", main: "cmd/server/main.go", tilt: "config-center-regional-image",
			namespace: "cp9-premium", workload: "config-center-regional", container: "config-center-regional",
		},
		"cluster-ng-server": {
			repository: "cluster-service-ng", main: "cmd/cluster_ng_server/main.go", tilt: "cluster-ng-server-image",
			namespace: "cp9-premium", workload: "cluster-ng-server", container: "server",
		},
	}
	if len(dryRun.Services) != len(expected) {
		r.productFatalf(t, stage, "dry-run reported %d services, want exactly %d", len(dryRun.Services), len(expected))
	}
	seen := map[string]sharedSourceDryService{}
	for _, service := range dryRun.Services {
		want, ok := expected[service.Name]
		if !ok || seen[service.Name].Name != "" {
			r.productFatalf(t, stage, "dry-run reported unexpected or duplicate service %q", service.Name)
		}
		canonical, err := filepath.EvalSymlinks(filepath.Join(r.metadata.WorkRoot, want.repository))
		if err != nil {
			r.harnessFatalf(t, stage, "resolve expected %s source root: %v", service.Name, err)
		}
		canonical, err = filepath.Abs(canonical)
		if err != nil {
			r.harnessFatalf(t, stage, "canonicalize expected %s source root: %v", service.Name, err)
		}
		head := r.gitOutput(t, stage, "source-head-"+service.Name, canonical, "rev-parse", "HEAD")
		dirty := r.gitOutput(t, stage, "source-status-"+service.Name, canonical, "status", "--porcelain")
		configured, err := filepath.Abs(filepath.Join(r.metadata.WorkRoot, want.repository))
		if err != nil {
			r.harnessFatalf(t, stage, "resolve configured %s source root: %v", service.Name, err)
		}
		if service.Role != "source-build" || service.ConfiguredPath != configured || service.CanonicalPath != canonical ||
			service.GitHEAD != head || service.Dirty || dirty != "" || service.BuildMain != want.main ||
			service.TiltResource != want.tilt || service.WorkloadKind != "kubernetes-deployment" ||
			service.WorkloadNamespace != want.namespace || service.WorkloadName != want.workload ||
			service.WorkloadContainer != want.container {
			r.productFatalf(t, stage, "dry-run source identity for %s does not match its selected clean checkout and workload contract", service.Name)
		}
		seen[service.Name] = service
		r.writeEvidence(t, stage, "source-head-"+service.Name+".txt", []byte(head+"\n"))
	}
	global, regional := seen["config-center"], seen["config-center-regional"]
	if global.CanonicalPath != regional.CanonicalPath || global.GitHEAD != regional.GitHEAD {
		r.productFatalf(t, stage, "config-center services do not share one complete Git root and revision")
	}
	if global.TiltResource == regional.TiltResource ||
		global.WorkloadNamespace+"/"+global.WorkloadName == regional.WorkloadNamespace+"/"+regional.WorkloadName {
		r.productFatalf(t, stage, "config-center services lost their distinct Tilt or workload identities")
	}
	data, err := json.MarshalIndent(dryRun.Services, "", "  ")
	if err != nil {
		r.harnessFatalf(t, stage, "encode selected source identities: %v", err)
	}
	r.writeEvidence(t, stage, "selected-services.json", append(data, '\n'))
}

func (r *sharedSourceRun) assertBoundedParallelBuilds(t *testing.T, stage *sharedSourceStageReceipt) {
	t.Helper()
	if !r.actionCompleted("cp9-up") {
		r.startAction(t, "cp9-up")
		r.mustSourceBuildCommand(t, stage, "up", 6*time.Hour,
			filepath.Join(r.metadata.CP9Root, "bin", "cp9"), "up", "config="+r.metadata.Config)
		r.completeAction(t, "cp9-up")
	}
	status := r.readReadyStatus(t, stage, "status-ready")
	if status.Session != r.metadata.Session || status.Status != "ready" {
		r.productFatalf(t, stage, "cp9 status=%q session=%q, want ready for %s", status.Status, status.Session, r.metadata.Session)
	}
	if strings.TrimSpace(status.Layers.Host.Execution.SourceSyncDestination) == "" {
		r.productFatalf(t, stage, "ready session has no remote source-sync destination")
	}
	remoteRoots, err := json.MarshalIndent(map[string]string{
		"config-center":      status.Layers.Host.Execution.SourceSyncDestination + "/controller/work/config-center",
		"cluster-service-ng": status.Layers.Host.Execution.SourceSyncDestination + "/controller/work/cluster-service-ng",
	}, "", "  ")
	if err != nil {
		r.harnessFatalf(t, stage, "encode remote source roots: %v", err)
	}
	r.writeEvidence(t, stage, "remote-source-roots.json", append(remoteRoots, '\n'))
	r.assertCompleteRemoteSource(t, stage, "config-center", "config-center",
		filepath.Join(r.metadata.WorkRoot, "config-center"), status.Layers.Host.Execution.SourceSyncDestination+"/controller/work/config-center")
	r.assertCompleteRemoteSource(t, stage, "cluster-service-ng", "cluster-ng-server",
		filepath.Join(r.metadata.WorkRoot, "cluster-service-ng"), status.Layers.Host.Execution.SourceSyncDestination+"/controller/work/cluster-service-ng")

	resources := []string{"config-center-image", "config-center-regional-image", "cluster-ng-server-image"}
	intervals := make([]sharedSourceBuildInterval, 0, len(resources))
	for _, name := range resources {
		resource := r.readUIResource(t, stage, "initial-"+name, name)
		if resource.Metadata.Name != name {
			r.harnessFatalf(t, stage, "Tilt returned resource %q for %s", resource.Metadata.Name, name)
		}
		if len(resource.Status.BuildHistory) == 0 {
			r.productFatalf(t, stage, "selected source build %s has no build history", name)
		}
		for _, build := range resource.Status.BuildHistory {
			start, startErr := time.Parse(time.RFC3339Nano, build.StartTime)
			finish, finishErr := time.Parse(time.RFC3339Nano, build.FinishTime)
			if startErr != nil || finishErr != nil {
				r.harnessFatalf(t, stage, "parse %s build interval %q..%q: start=%v finish=%v", name, build.StartTime, build.FinishTime, startErr, finishErr)
			}
			if !finish.After(start) {
				r.productFatalf(t, stage, "source build %s has non-positive interval %s..%s", name, start, finish)
			}
			intervals = append(intervals, sharedSourceBuildInterval{Resource: name, SpanID: build.SpanID, Start: start, Finish: finish})
		}
	}
	maximum := sharedSourceMaximumOverlap(intervals)
	if maximum > 2 {
		r.productFatalf(t, stage, "source builds reached overlap %d, configured limit is 2", maximum)
	}
	data, err := json.MarshalIndent(struct {
		ConfiguredLimit int                         `json:"configured_limit"`
		MaximumOverlap  int                         `json:"maximum_overlap"`
		Intervals       []sharedSourceBuildInterval `json:"intervals"`
	}{ConfiguredLimit: 2, MaximumOverlap: maximum, Intervals: intervals}, "", "  ")
	if err != nil {
		r.harnessFatalf(t, stage, "encode source-build intervals: %v", err)
	}
	r.writeEvidence(t, stage, "build-concurrency.json", append(data, '\n'))
}

func sharedSourceMaximumOverlap(intervals []sharedSourceBuildInterval) int {
	type event struct {
		at    time.Time
		delta int
	}
	events := make([]event, 0, len(intervals)*2)
	for _, interval := range intervals {
		events = append(events, event{at: interval.Start, delta: 1}, event{at: interval.Finish, delta: -1})
	}
	sort.Slice(events, func(i, j int) bool {
		if events[i].at.Equal(events[j].at) {
			return events[i].delta < events[j].delta
		}
		return events[i].at.Before(events[j].at)
	})
	active, maximum := 0, 0
	for _, event := range events {
		active += event.delta
		maximum = max(maximum, active)
	}
	return maximum
}

func (r *sharedSourceRun) assertCompleteRemoteSource(t *testing.T, stage *sharedSourceStageReceipt, repository, service, localRoot, remoteRoot string) {
	t.Helper()
	expectedHead := strings.TrimSpace(string(r.readFile(t, stage, filepath.Join(r.runDir, "source-head-"+service+".txt"))))
	if head := r.gitOutput(t, stage, "remote-source-head-"+repository, localRoot, "rev-parse", "HEAD"); head != expectedHead {
		r.harnessFatalf(t, stage, "local %s source head changed after its dry-run identity was frozen", repository)
	}
	if dirty := r.gitOutput(t, stage, "remote-source-status-"+repository, localRoot, "status", "--porcelain"); dirty != "" {
		r.harnessFatalf(t, stage, "local %s source changed after its dry-run identity was frozen", repository)
	}
	expected := r.trackedSourceManifest(t, stage, repository, localRoot)
	remote := r.remoteSourceManifest(t, stage, repository, remoteRoot, expected)
	for path, want := range expected {
		got, ok := remote[path]
		if !ok || got != want {
			r.productFatalf(t, stage, "remote %s source differs at tracked path %q", repository, path)
		}
	}
	data, err := json.MarshalIndent(remote, "", "  ")
	if err != nil {
		r.harnessFatalf(t, stage, "encode observed %s source manifest: %v", repository, err)
	}
	r.writeEvidence(t, stage, "remote-source-manifest-"+repository+".json", append(data, '\n'))
}

func (r *sharedSourceRun) trackedSourceManifest(t *testing.T, stage *sharedSourceStageReceipt, repository, root string) map[string]sharedSourceFileIdentity {
	t.Helper()
	result := r.mustHarnessCommand(t, stage, "tracked-source-files-"+repository, 2*time.Minute, root, nil,
		"git", "-C", root, "ls-files", "-z")
	manifest := map[string]sharedSourceFileIdentity{}
	for _, raw := range bytes.Split(r.readCommandStdout(t, stage, result), []byte{0}) {
		if len(raw) == 0 {
			continue
		}
		path := string(raw)
		full := filepath.Join(root, filepath.FromSlash(path))
		if !sharedSourceChildPath(root, full) {
			r.harnessFatalf(t, stage, "tracked %s path escapes its source root: %q", repository, path)
		}
		info, err := os.Lstat(full)
		if err != nil {
			r.harnessFatalf(t, stage, "inspect tracked %s path %q: %v", repository, path, err)
		}
		switch mode := info.Mode(); {
		case mode.IsRegular():
			digest := sha256.Sum256(r.readFile(t, stage, full))
			manifest[path] = sharedSourceFileIdentity{Kind: "file", Digest: fmt.Sprintf("%x", digest)}
		case mode&os.ModeSymlink != 0:
			target, err := os.Readlink(full)
			if err != nil {
				r.harnessFatalf(t, stage, "read tracked %s symlink %q: %v", repository, path, err)
			}
			manifest[path] = sharedSourceFileIdentity{Kind: "symlink", Target: target}
		default:
			r.harnessFatalf(t, stage, "tracked %s path %q has unsupported mode %s", repository, path, mode)
		}
	}
	if len(manifest) == 0 {
		r.harnessFatalf(t, stage, "tracked %s source manifest is empty", repository)
	}
	data, err := json.MarshalIndent(manifest, "", "  ")
	if err != nil {
		r.harnessFatalf(t, stage, "encode tracked %s source manifest: %v", repository, err)
	}
	r.writeEvidence(t, stage, "local-source-manifest-"+repository+".json", append(data, '\n'))
	return manifest
}

func (r *sharedSourceRun) remoteSourceManifest(t *testing.T, stage *sharedSourceStageReceipt, repository, root string, expected map[string]sharedSourceFileIdentity) map[string]sharedSourceFileIdentity {
	t.Helper()
	helper := filepath.Join(r.metadata.CP9Root, "skills", "get-started", "scripts", "session-ssh.sh")
	paths := make([]string, 0, len(expected))
	for path := range expected {
		paths = append(paths, path)
	}
	sort.Strings(paths)
	var input bytes.Buffer
	for _, path := range paths {
		input.WriteString(path)
		input.WriteByte(0)
	}
	const inspect = `set -euo pipefail
cd "$1"
files=$(mktemp)
trap 'unlink "$files"' EXIT
while IFS= read -r -d '' path; do
  if [[ "$path" = /* || "$path" = ../* || "$path" = */../* ]]; then exit 65; fi
  if [[ -L "$path" ]]; then
    printf 'symlink\0%s\0' "$path"
    readlink -z -- "$path"
  elif [[ -f "$path" ]]; then
    printf '%s\0' "$path" >>"$files"
  else
    printf 'tracked remote source path is missing: %s\n' "$path" >&2
    exit 66
  fi
done
if [[ -s "$files" ]]; then
  xargs -0 sha256sum -z -- <"$files" |
    while IFS= read -r -d '' record; do
      digest=${record%% *}
      path=${record#* }
      printf 'file\0%s\0%s\0' "${path# }" "$digest"
    done
fi`
	result := r.mustRemoteSourceCommand(t, stage, "remote-source-files-"+repository, 10*time.Minute, input.Bytes(),
		map[string]string{"CP9_REPO_ROOT": r.metadata.CP9Root}, "bash", helper, r.metadata.Session,
		"bash", "-c", inspect, "shared-source-inspect", root)
	manifest := map[string]sharedSourceFileIdentity{}
	parts := bytes.Split(r.readCommandStdout(t, stage, result), []byte{0})
	if len(parts) > 0 && len(parts[len(parts)-1]) == 0 {
		parts = parts[:len(parts)-1]
	}
	if len(parts)%3 != 0 {
		r.harnessFatalf(t, stage, "parse remote %s source records", repository)
	}
	for index := 0; index < len(parts); index += 3 {
		kind, path, value := string(parts[index]), string(parts[index+1]), string(parts[index+2])
		if _, duplicate := manifest[path]; duplicate {
			r.harnessFatalf(t, stage, "remote %s source path has multiple identities: %q", repository, path)
		}
		switch kind {
		case "file":
			manifest[path] = sharedSourceFileIdentity{Kind: kind, Digest: value}
		case "symlink":
			manifest[path] = sharedSourceFileIdentity{Kind: kind, Target: value}
		default:
			r.harnessFatalf(t, stage, "remote %s source path %q has unknown kind %q", repository, path, kind)
		}
	}
	return manifest
}

func (r *sharedSourceRun) mustRemoteSourceCommand(t *testing.T, stage *sharedSourceStageReceipt, label string, timeout time.Duration, input []byte, env map[string]string, argv ...string) sharedSourceCommandResult {
	t.Helper()
	result, err := r.commandWithInput(label, timeout, r.metadata.CP9Root, input, env, argv...)
	if err != nil {
		r.harnessFatalf(t, stage, "observe remote source with %s: %v", label, err)
	}
	if result.ExitCode == 0 {
		return result
	}
	stderr := string(r.readFile(t, stage, result.Stderr))
	if result.ExitCode == 66 && strings.Contains(stderr, "tracked remote source path is missing:") {
		r.productFatalf(t, stage, "remote source observation %s found missing tracked source", label)
	}
	r.harnessFatalf(t, stage, "remote source inspection %s exited %d", label, result.ExitCode)
	return result
}

func (r *sharedSourceRun) assertTargetedDeploy(t *testing.T, stage *sharedSourceStageReceipt, target, sibling string) {
	t.Helper()
	action := "deploy-" + target
	freshBefore := !r.actionCompleted(action)
	beforeTarget := r.readOrLoadUIResource(t, stage, action+"-before-"+target, target, freshBefore)
	beforeTargetBuild := r.readOrLoadUIResource(t, stage, action+"-before-"+target+"-live-binary", target+"-live-binary", freshBefore)
	beforeSibling := r.readOrLoadUIResource(t, stage, action+"-before-"+sibling, sibling, freshBefore)
	beforeSiblingBuild := r.readOrLoadUIResource(t, stage, action+"-before-"+sibling+"-live-binary", sibling+"-live-binary", freshBefore)
	beforeSiblingWorkload := r.readOrLoadWorkload(t, stage, action+"-before-"+sibling+"-workload", sibling, freshBefore)
	r.assertServiceReady(t, stage, beforeTarget, target)
	r.assertServiceReady(t, stage, beforeSibling, sibling)
	r.assertWorkloadReady(t, stage, beforeSiblingWorkload, sibling)

	if !r.actionCompleted(action) {
		r.startAction(t, action)
		r.mustSourceBuildCommand(t, stage, action, 30*time.Minute,
			filepath.Join(r.metadata.CP9Root, "bin", "cp9"), "deploy", "service="+target, "config="+r.metadata.Config)
		r.completeAction(t, action)
	}

	afterTarget := r.readUIResource(t, stage, action+"-after-"+target, target)
	afterTargetBuild := r.readUIResource(t, stage, action+"-after-"+target+"-live-binary", target+"-live-binary")
	afterSibling := r.readUIResource(t, stage, action+"-after-"+sibling, sibling)
	afterSiblingBuild := r.readUIResource(t, stage, action+"-after-"+sibling+"-live-binary", sibling+"-live-binary")
	afterSiblingWorkload := r.readWorkload(t, stage, action+"-after-"+sibling+"-workload", sibling)
	r.assertServiceReady(t, stage, afterTarget, target)
	r.assertServiceReady(t, stage, afterSibling, sibling)
	r.assertWorkloadReady(t, stage, afterSiblingWorkload, sibling)
	r.assertOneNewCompletedBuild(t, stage, beforeTargetBuild, afterTargetBuild, target+"-live-binary")
	if !slices.Equal(beforeSiblingBuild.Status.BuildHistory, afterSiblingBuild.Status.BuildHistory) {
		r.productFatalf(t, stage, "targeted deploy of %s also rebuilt sibling %s", target, sibling)
	}
	if beforeSibling.Metadata.Name != afterSibling.Metadata.Name || afterSibling.Metadata.Name != sibling ||
		!sharedSourceSameWorkloadMapping(beforeSiblingWorkload, afterSiblingWorkload) {
		r.productFatalf(t, stage, "targeted deploy of %s deleted or remapped sibling %s", target, sibling)
	}
	status := r.readReadyStatus(t, stage, action+"-status-ready")
	if status.Session != r.metadata.Session || status.Status != "ready" {
		r.productFatalf(t, stage, "session left ready state after targeted deploy of %s", target)
	}
}

func (r *sharedSourceRun) readOrLoadUIResource(t *testing.T, stage *sharedSourceStageReceipt, label, name string, fresh bool) sharedSourceUIResource {
	t.Helper()
	if fresh {
		return r.readUIResource(t, stage, label, name)
	}
	path := filepath.Join(r.runDir, label+".json")
	data := r.readFile(t, stage, path)
	var resource sharedSourceUIResource
	if err := json.Unmarshal(data, &resource); err != nil {
		r.harnessFatalf(t, stage, "parse preserved Tilt resource %s: %v", path, err)
	}
	return resource
}

func (r *sharedSourceRun) readOrLoadWorkload(t *testing.T, stage *sharedSourceStageReceipt, label, service string, fresh bool) sharedSourceWorkload {
	t.Helper()
	if fresh {
		return r.readWorkload(t, stage, label, service)
	}
	return r.parseWorkload(t, stage, label, service, r.readFile(t, stage, filepath.Join(r.runDir, label+".json")))
}

func (r *sharedSourceRun) readWorkload(t *testing.T, stage *sharedSourceStageReceipt, label, service string) sharedSourceWorkload {
	t.Helper()
	selected := r.selectedService(t, stage, service)
	if selected.WorkloadKind != "kubernetes-deployment" {
		r.harnessFatalf(t, stage, "selected service %s has unsupported workload kind %q", service, selected.WorkloadKind)
	}
	helper := filepath.Join(r.metadata.CP9Root, "skills", "get-started", "scripts", "session-kubectl.sh")
	result, err := r.command(label, 3*time.Minute, r.metadata.CP9Root,
		map[string]string{"CP9_REPO_ROOT": r.metadata.CP9Root}, "bash", helper, r.metadata.Session,
		"-n", selected.WorkloadNamespace, "get", "deployment", selected.WorkloadName, "-o", "json")
	if err != nil {
		r.harnessFatalf(t, stage, "observe workload for sibling %s: %v", service, err)
	}
	if result.ExitCode != 0 {
		stderr := string(r.readFile(t, stage, result.Stderr))
		if strings.Contains(stderr, "NotFound") || strings.Contains(stderr, "not found") {
			r.productFatalf(t, stage, "frozen sibling workload for %s is missing", service)
		}
		r.harnessFatalf(t, stage, "workload observation for sibling %s exited %d", service, result.ExitCode)
	}
	data := r.readCommandStdout(t, stage, result)
	r.writeEvidence(t, stage, label+".json", data)
	return r.parseWorkload(t, stage, label, service, data)
}

func (r *sharedSourceRun) selectedService(t *testing.T, stage *sharedSourceStageReceipt, name string) sharedSourceDryService {
	t.Helper()
	var services []sharedSourceDryService
	if err := json.Unmarshal(r.readFile(t, stage, filepath.Join(r.runDir, "selected-services.json")), &services); err != nil {
		r.harnessFatalf(t, stage, "parse frozen selected services: %v", err)
	}
	var selected sharedSourceDryService
	for _, service := range services {
		if service.Name != name {
			continue
		}
		if selected.Name != "" {
			r.harnessFatalf(t, stage, "frozen selected services contain duplicate %s", name)
		}
		selected = service
	}
	if selected.Name == "" {
		r.harnessFatalf(t, stage, "frozen selected services omit %s", name)
	}
	return selected
}

func (r *sharedSourceRun) parseWorkload(t *testing.T, stage *sharedSourceStageReceipt, label, service string, data []byte) sharedSourceWorkload {
	t.Helper()
	selected := r.selectedService(t, stage, service)
	var resource struct {
		Metadata struct {
			Namespace  string `json:"namespace"`
			Name       string `json:"name"`
			UID        string `json:"uid"`
			Generation int64  `json:"generation"`
		} `json:"metadata"`
		Spec struct {
			Replicas int32 `json:"replicas"`
			Template struct {
				Spec struct {
					Containers []struct {
						Name  string `json:"name"`
						Image string `json:"image"`
					} `json:"containers"`
				} `json:"spec"`
			} `json:"template"`
		} `json:"spec"`
		Status struct {
			ObservedGeneration int64 `json:"observedGeneration"`
			ReadyReplicas      int32 `json:"readyReplicas"`
			AvailableReplicas  int32 `json:"availableReplicas"`
		} `json:"status"`
	}
	if err := json.Unmarshal(data, &resource); err != nil {
		r.harnessFatalf(t, stage, "parse workload observation %s: %v", label, err)
	}
	workload := sharedSourceWorkload{
		Namespace: resource.Metadata.Namespace, Name: resource.Metadata.Name, UID: resource.Metadata.UID,
		Generation: resource.Metadata.Generation, ObservedGeneration: resource.Status.ObservedGeneration,
		Replicas: resource.Spec.Replicas, ReadyReplicas: resource.Status.ReadyReplicas,
		AvailableReplicas: resource.Status.AvailableReplicas, Container: selected.WorkloadContainer,
	}
	for _, container := range resource.Spec.Template.Spec.Containers {
		if container.Name == selected.WorkloadContainer {
			if workload.Image != "" {
				r.harnessFatalf(t, stage, "workload %s has duplicate container %s", service, selected.WorkloadContainer)
			}
			workload.Image = container.Image
		}
	}
	if workload.Namespace != selected.WorkloadNamespace || workload.Name != selected.WorkloadName ||
		workload.UID == "" || workload.Generation == 0 || workload.Image == "" {
		r.productFatalf(t, stage, "workload observation for %s does not match its frozen namespace, name, and container mapping", service)
	}
	return workload
}

func (r *sharedSourceRun) assertWorkloadReady(t *testing.T, stage *sharedSourceStageReceipt, workload sharedSourceWorkload, service string) {
	t.Helper()
	if workload.Replicas < 1 || workload.ObservedGeneration < workload.Generation ||
		workload.ReadyReplicas < workload.Replicas || workload.AvailableReplicas < workload.Replicas {
		r.productFatalf(t, stage, "workload for sibling %s is not fully observed, Ready, and Available", service)
	}
}

func sharedSourceSameWorkloadMapping(before, after sharedSourceWorkload) bool {
	return before.Namespace == after.Namespace && before.Name == after.Name && before.UID == after.UID &&
		before.Generation == after.Generation && before.Replicas == after.Replicas &&
		before.Container == after.Container && before.Image == after.Image
}

func (r *sharedSourceRun) readUIResource(t *testing.T, stage *sharedSourceStageReceipt, label, name string) sharedSourceUIResource {
	t.Helper()
	helper := filepath.Join(r.metadata.CP9Root, "spec-tests", "scripts", "session-tilt-get.sh")
	result := r.mustHarnessCommand(t, stage, label, 3*time.Minute, r.metadata.CP9Root,
		map[string]string{"CP9_REPO_ROOT": r.metadata.CP9Root}, "bash", helper, r.metadata.Session, name)
	data := r.readCommandStdout(t, stage, result)
	r.writeEvidence(t, stage, label+".json", data)
	var resource sharedSourceUIResource
	if err := json.Unmarshal(data, &resource); err != nil {
		r.harnessFatalf(t, stage, "parse Tilt resource %s: %v", name, err)
	}
	return resource
}

func (r *sharedSourceRun) assertServiceReady(t *testing.T, stage *sharedSourceStageReceipt, resource sharedSourceUIResource, name string) {
	t.Helper()
	if resource.Metadata.Name != name {
		r.productFatalf(t, stage, "Tilt service resource %s is missing or remapped to %q", name, resource.Metadata.Name)
	}
	for _, condition := range resource.Status.Conditions {
		if strings.EqualFold(condition.Type, "Ready") && strings.EqualFold(condition.Status, "True") {
			return
		}
	}
	r.productFatalf(t, stage, "Tilt service resource %s is not Ready", name)
}

func (r *sharedSourceRun) assertOneNewCompletedBuild(t *testing.T, stage *sharedSourceStageReceipt, before, after sharedSourceUIResource, name string) {
	t.Helper()
	if before.Metadata.Name != name || after.Metadata.Name != name {
		r.productFatalf(t, stage, "Tilt build resource %s is missing or remapped", name)
	}
	old := map[string]bool{}
	for _, build := range before.Status.BuildHistory {
		old[sharedSourceBuildKey(build)] = true
	}
	var added []sharedSourceBuildHistory
	for _, build := range after.Status.BuildHistory {
		if !old[sharedSourceBuildKey(build)] {
			added = append(added, build)
		}
	}
	if len(added) != 1 {
		r.productFatalf(t, stage, "targeted deploy produced %d new %s invocations, want 1", len(added), name)
	}
	start, startErr := time.Parse(time.RFC3339Nano, added[0].StartTime)
	finish, finishErr := time.Parse(time.RFC3339Nano, added[0].FinishTime)
	if startErr != nil || finishErr != nil {
		r.harnessFatalf(t, stage, "parse new %s invocation interval: start=%v finish=%v", name, startErr, finishErr)
	}
	if !finish.After(start) || !strings.EqualFold(after.Status.UpdateStatus, "ok") {
		r.productFatalf(t, stage, "new %s invocation did not reach a successful terminal state", name)
	}
}

func sharedSourceBuildKey(build sharedSourceBuildHistory) string {
	return build.SpanID + "\x00" + build.StartTime + "\x00" + build.FinishTime
}

func (r *sharedSourceRun) readReadyStatus(t *testing.T, stage *sharedSourceStageReceipt, label string) sharedSourceStatus {
	t.Helper()
	result := r.mustHarnessCommand(t, stage, label, 5*time.Minute, r.metadata.CP9Root, nil,
		filepath.Join(r.metadata.CP9Root, "bin", "cp9"), "{sys.output.format=json}", "status", "config="+r.metadata.Config)
	data := r.readCommandStdout(t, stage, result)
	r.writeEvidence(t, stage, label+".json", data)
	var status sharedSourceStatus
	if err := json.Unmarshal(data, &status); err != nil {
		r.harnessFatalf(t, stage, "parse %s: %v", label, err)
	}
	return status
}

func (r *sharedSourceRun) cleanup(t *testing.T) {
	t.Helper()
	helper := filepath.Join(r.repoRoot, "spec-tests", "scripts", "default-804-session.sh")
	result, err := r.command("cleanup", 90*time.Minute, r.repoRoot, nil, "bash", helper, "cleanup", r.runDir)
	if err != nil || result.ExitCode != 0 {
		r.journal.Cleanup = "fail"
		_ = r.persistJournal()
		t.Fatalf("clean exact shared-source session: error=%v exit=%d", err, result.ExitCode)
	}
	data, err := os.ReadFile(filepath.Join(r.runDir, "cleanup-receipt.json"))
	if err != nil {
		r.journal.Cleanup = "fail"
		_ = r.persistJournal()
		t.Fatalf("read cleanup receipt: %v", err)
	}
	var receipt struct {
		Destroy  string `json:"destroy"`
		Worktree string `json:"worktree"`
	}
	err = json.Unmarshal(data, &receipt)
	_, upStarted := r.journal.Actions["cp9-up"]
	destroyPassed := receipt.Destroy == "pass" || receipt.Destroy == "not_needed" && !upStarted
	if err != nil || !destroyPassed ||
		(receipt.Worktree != "pass" && receipt.Worktree != "not_needed") {
		r.journal.Cleanup = "fail"
		_ = r.persistJournal()
		t.Fatalf("cleanup receipt is not passing: parse=%v destroy=%q worktree=%q", err, receipt.Destroy, receipt.Worktree)
	}
	r.journal.Cleanup = "pass"
	if err := r.persistJournal(); err != nil {
		t.Fatalf("record passing cleanup: %v", err)
	}
}

func (r *sharedSourceRun) beginStage(t *testing.T, name string, replaySafe bool) *sharedSourceStageReceipt {
	t.Helper()
	receipts := r.journal.Stages[name]
	if len(receipts) > 0 {
		latest := receipts[len(receipts)-1]
		if latest.Status == "passed" {
			t.Skipf("reuse passing %s receipt from head %s", name, latest.Head)
		}
		if r.retryStage != name {
			t.Fatalf("stage %s is %s; set %s=%s only after same-scene diagnosis establishes a safe replay", name, latest.Status, sharedSourceRetryStageEnv, name)
		}
		if !replaySafe {
			t.Fatalf("stage %s cannot replay because its product action is not recorded complete", name)
		}
	}
	now := time.Now().UTC().Format(time.RFC3339Nano)
	r.journal.Stages[name] = append(receipts, sharedSourceStageReceipt{
		Head: r.currentHead, StartedAt: now, Status: "running", Product: "not_run", Harness: "not_run",
	})
	if err := r.persistJournal(); err != nil {
		t.Fatalf("begin shared-source stage %s: %v", name, err)
	}
	return &r.journal.Stages[name][len(r.journal.Stages[name])-1]
}

func (r *sharedSourceRun) finishStage(t *testing.T, stage *sharedSourceStageReceipt) {
	t.Helper()
	stage.FinishedAt = time.Now().UTC().Format(time.RFC3339Nano)
	if t.Failed() {
		stage.Status = "failed"
		if stage.Product == "not_run" && stage.Harness == "not_run" && stage.Message == "" {
			stage.Harness = "fail"
			stage.Message = "subtest failed without a classified result"
		}
	} else if !t.Skipped() {
		stage.Status = "passed"
		stage.Product = "pass"
		stage.Harness = "pass"
	}
	if err := r.persistJournal(); err != nil {
		t.Errorf("finish shared-source stage: %v", err)
	}
}

func (r *sharedSourceRun) requirePassed(t *testing.T, name string) {
	t.Helper()
	receipts := r.journal.Stages[name]
	if len(receipts) == 0 || receipts[len(receipts)-1].Status != "passed" {
		t.Skipf("requires passing %s receipt from this run", name)
	}
}

func (r *sharedSourceRun) actionCompleted(name string) bool {
	return r.journal.Actions[name].Status == "completed"
}

func (r *sharedSourceRun) actionReplaySafe(name string) bool {
	status := r.journal.Actions[name].Status
	return status == "" || status == "completed"
}

func (r *sharedSourceRun) startAction(t *testing.T, name string) {
	t.Helper()
	if current := r.journal.Actions[name]; current.Status != "" {
		t.Fatalf("product action %s already has status %s", name, current.Status)
	}
	r.journal.Actions[name] = sharedSourceActionReceipt{
		Head: r.currentHead, StartedAt: time.Now().UTC().Format(time.RFC3339Nano), Status: "started",
	}
	if err := r.persistJournal(); err != nil {
		t.Fatalf("record start of product action %s: %v", name, err)
	}
}

func (r *sharedSourceRun) completeAction(t *testing.T, name string) {
	t.Helper()
	action := r.journal.Actions[name]
	if action.Status != "started" {
		t.Fatalf("product action %s status=%q, want started", name, action.Status)
	}
	action.Status = "completed"
	action.FinishedAt = time.Now().UTC().Format(time.RFC3339Nano)
	r.journal.Actions[name] = action
	if err := r.persistJournal(); err != nil {
		t.Fatalf("record completion of product action %s: %v", name, err)
	}
}

func (r *sharedSourceRun) prerequisiteFatalf(t *testing.T, stage *sharedSourceStageReceipt, format string, args ...any) {
	t.Helper()
	stage.Product = "not_run"
	stage.Harness = "not_run"
	stage.Message = fmt.Sprintf(format, args...)
	t.Fatalf(format, args...)
}

func (r *sharedSourceRun) productFatalf(t *testing.T, stage *sharedSourceStageReceipt, format string, args ...any) {
	t.Helper()
	stage.Product = "fail"
	stage.Harness = "pass"
	stage.Message = fmt.Sprintf(format, args...)
	t.Fatalf(format, args...)
}

func (r *sharedSourceRun) harnessFatalf(t *testing.T, stage *sharedSourceStageReceipt, format string, args ...any) {
	t.Helper()
	stage.Product = "not_run"
	stage.Harness = "fail"
	stage.Message = fmt.Sprintf(format, args...)
	t.Fatalf(format, args...)
}

func (r *sharedSourceRun) mustPrerequisiteCommand(t *testing.T, stage *sharedSourceStageReceipt, label string, timeout time.Duration, directory string, env map[string]string, argv ...string) sharedSourceCommandResult {
	t.Helper()
	result, err := r.command(label, timeout, directory, env, argv...)
	if err != nil || result.ExitCode != 0 {
		r.prerequisiteFatalf(t, stage, "prerequisite command %s failed: error=%v exit=%d", label, err, result.ExitCode)
	}
	return result
}

func (r *sharedSourceRun) mustProductCommand(t *testing.T, stage *sharedSourceStageReceipt, label string, timeout time.Duration, directory string, env map[string]string, argv ...string) sharedSourceCommandResult {
	t.Helper()
	result, err := r.command(label, timeout, directory, env, argv...)
	if err != nil {
		r.harnessFatalf(t, stage, "run product command %s: %v", label, err)
	}
	if result.ExitCode != 0 {
		r.productFatalf(t, stage, "product command %s exited %d", label, result.ExitCode)
	}
	return result
}

func (r *sharedSourceRun) mustSourceBuildCommand(t *testing.T, stage *sharedSourceStageReceipt, label string, timeout time.Duration, argv ...string) sharedSourceCommandResult {
	t.Helper()
	result, err := r.command(label, timeout, r.metadata.CP9Root, nil, argv...)
	if err != nil {
		r.harnessFatalf(t, stage, "run source-build command %s: %v", label, err)
	}
	if result.ExitCode == 0 {
		return result
	}
	combined := append(r.readFile(t, stage, result.Stdout), r.readFile(t, stage, result.Stderr)...)
	if strings.Contains(string(combined), "GitHub device authorization expired") {
		r.prerequisiteFatalf(t, stage, "source-build command %s stopped because GitHub Device Flow expired", label)
	}
	r.productFatalf(t, stage, "source-build command %s exited %d", label, result.ExitCode)
	return result
}

func (r *sharedSourceRun) mustHarnessCommand(t *testing.T, stage *sharedSourceStageReceipt, label string, timeout time.Duration, directory string, env map[string]string, argv ...string) sharedSourceCommandResult {
	t.Helper()
	result, err := r.command(label, timeout, directory, env, argv...)
	if err != nil || result.ExitCode != 0 {
		r.harnessFatalf(t, stage, "harness command %s failed: error=%v exit=%d", label, err, result.ExitCode)
	}
	return result
}

func (r *sharedSourceRun) command(label string, timeout time.Duration, directory string, overrides map[string]string, argv ...string) (sharedSourceCommandResult, error) {
	return r.commandWithInput(label, timeout, directory, nil, overrides, argv...)
}

func (r *sharedSourceRun) commandWithInput(label string, timeout time.Duration, directory string, input []byte, overrides map[string]string, argv ...string) (sharedSourceCommandResult, error) {
	r.commandCount++
	prefix := fmt.Sprintf("%03d-%s", r.commandCount, sharedSourceLabel(label))
	commandsDir := filepath.Join(r.runDir, "commands")
	if err := os.MkdirAll(commandsDir, 0o700); err != nil {
		return sharedSourceCommandResult{}, err
	}
	stdoutPath := filepath.Join(commandsDir, prefix+".stdout.log")
	stderrPath := filepath.Join(commandsDir, prefix+".stderr.log")
	stdout, err := os.OpenFile(stdoutPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return sharedSourceCommandResult{}, err
	}
	stderr, err := os.OpenFile(stderrPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		_ = stdout.Close()
		return sharedSourceCommandResult{}, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	started := time.Now()
	command := exec.CommandContext(ctx, argv[0], argv[1:]...)
	command.Dir = directory
	command.Env = sharedSourceCommandEnv(overrides)
	if input != nil {
		command.Stdin = bytes.NewReader(input)
	}
	command.Stdout = io.MultiWriter(stdout, os.Stdout)
	command.Stderr = io.MultiWriter(stderr, os.Stderr)
	command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	command.Cancel = func() error {
		if command.Process == nil {
			return os.ErrProcessDone
		}
		err := syscall.Kill(-command.Process.Pid, syscall.SIGKILL)
		if errors.Is(err, syscall.ESRCH) {
			return os.ErrProcessDone
		}
		return err
	}
	command.WaitDelay = 5 * time.Second
	runErr := command.Run()
	finished := time.Now()
	closeErr := errors.Join(stdout.Close(), stderr.Close())
	exitCode := 0
	if runErr != nil {
		exitCode = -1
		var exitErr *exec.ExitError
		if errors.As(runErr, &exitErr) {
			exitCode = exitErr.ExitCode()
		}
	}
	receipt := sharedSourceCommandReceipt{
		Command: argv, Directory: directory,
		StartedAt: started.UTC().Format(time.RFC3339Nano), FinishedAt: finished.UTC().Format(time.RFC3339Nano),
		DurationMS: finished.Sub(started).Milliseconds(), ExitCode: exitCode,
		Stdout: filepath.ToSlash(filepath.Join("commands", filepath.Base(stdoutPath))),
		Stderr: filepath.ToSlash(filepath.Join("commands", filepath.Base(stderrPath))),
	}
	if runErr != nil {
		receipt.Error = runErr.Error()
	}
	if ctx.Err() != nil {
		receipt.Error = ctx.Err().Error()
	}
	data, marshalErr := json.MarshalIndent(receipt, "", "  ")
	if marshalErr == nil {
		marshalErr = os.WriteFile(filepath.Join(commandsDir, prefix+".json"), append(data, '\n'), 0o600)
	}
	if closeErr != nil || marshalErr != nil {
		return sharedSourceCommandResult{}, errors.Join(closeErr, marshalErr)
	}
	result := sharedSourceCommandResult{ExitCode: exitCode, Stdout: stdoutPath, Stderr: stderrPath}
	if ctx.Err() != nil {
		return result, ctx.Err()
	}
	if runErr != nil {
		var exitErr *exec.ExitError
		if !errors.As(runErr, &exitErr) {
			return result, runErr
		}
	}
	return result, nil
}

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

func (r *sharedSourceRun) gitOutput(t *testing.T, stage *sharedSourceStageReceipt, label, directory string, args ...string) string {
	t.Helper()
	argv := append([]string{"git", "-C", directory}, args...)
	result := r.mustHarnessCommand(t, stage, label, 2*time.Minute, r.repoRoot, nil, argv...)
	return strings.TrimSpace(string(r.readCommandStdout(t, stage, result)))
}

func (r *sharedSourceRun) readCommandStdout(t *testing.T, stage *sharedSourceStageReceipt, result sharedSourceCommandResult) []byte {
	t.Helper()
	return r.readFile(t, stage, result.Stdout)
}

func (r *sharedSourceRun) readFile(t *testing.T, stage *sharedSourceStageReceipt, path string) []byte {
	t.Helper()
	data, err := os.ReadFile(path)
	if err != nil {
		r.harnessFatalf(t, stage, "read evidence %s: %v", path, err)
	}
	return data
}

func (r *sharedSourceRun) writeEvidence(t *testing.T, stage *sharedSourceStageReceipt, name string, data []byte) {
	t.Helper()
	path := filepath.Join(r.runDir, name)
	if !sharedSourceChildPath(r.runDir, path) {
		r.harnessFatalf(t, stage, "evidence path %q escapes run directory", name)
	}
	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
		r.harnessFatalf(t, stage, "create evidence parent for %s: %v", name, err)
	}
	if err := os.WriteFile(path, data, 0o600); err != nil {
		r.harnessFatalf(t, stage, "write evidence %s: %v", name, err)
	}
}

func (r *sharedSourceRun) persistJournal() error {
	r.journal.UpdatedAt = time.Now().UTC().Format(time.RFC3339Nano)
	data, err := json.MarshalIndent(r.journal, "", "  ")
	if err != nil {
		return err
	}
	temporary, err := os.CreateTemp(r.runDir, ".shared-source-state-")
	if err != nil {
		return err
	}
	temporaryPath := temporary.Name()
	defer os.Remove(temporaryPath)
	if err := temporary.Chmod(0o600); err != nil {
		_ = temporary.Close()
		return err
	}
	if _, err := temporary.Write(append(data, '\n')); err != nil {
		_ = temporary.Close()
		return err
	}
	if err := temporary.Sync(); err != nil {
		_ = temporary.Close()
		return err
	}
	if err := temporary.Close(); err != nil {
		return err
	}
	return os.Rename(temporaryPath, filepath.Join(r.runDir, sharedSourceJournalFile))
}

func (r *sharedSourceRun) writeSummary(t *testing.T) {
	t.Helper()
	product, harness := "pass", "pass"
	for _, name := range sharedSourceProductStages {
		productStage := sharedSourceAttemptVerdict(r.journal.Stages[name], true)
		harnessStage := sharedSourceAttemptVerdict(r.journal.Stages[name], false)
		if productStage == "not_run" {
			if product == "pass" {
				product = "not_run"
			}
		} else if productStage == "fail" {
			product = "fail"
		}
		if harnessStage == "not_run" {
			if harness == "pass" {
				harness = "not_run"
			}
		} else if harnessStage == "fail" {
			harness = "fail"
		}
	}
	var summary strings.Builder
	fmt.Fprintf(&summary, "# Shared-repository source builds and bounded parallelism\n\n")
	fmt.Fprintf(&summary, "- Product verdict: %s\n- Harness verdict: %s\n- Cleanup verdict: %s\n", product, harness, r.journal.Cleanup)
	fmt.Fprintf(&summary, "- Product CP9 head: %s\n- Current harness head: %s\n- Session: %s\n", sharedSourceValue(r.journal.CP9Head), r.currentHead, sharedSourceValue(r.journal.Session))
	summary.WriteString("\n| Go subtest | Product | Harness | Status |\n| --- | --- | --- | --- |\n")
	for _, name := range sharedSourceProductStages {
		receipts := r.journal.Stages[name]
		if len(receipts) == 0 {
			fmt.Fprintf(&summary, "| %s | not_run | not_run | not_run |\n", name)
			continue
		}
		for index, receipt := range receipts {
			label := name
			if len(receipts) > 1 {
				label = fmt.Sprintf("%s attempt %d", name, index+1)
			}
			fmt.Fprintf(&summary, "| %s | %s | %s | %s |\n", label, receipt.Product, receipt.Harness, receipt.Status)
		}
	}
	summary.WriteString("\n- Evidence: [run evidence](.)\n")
	if err := os.WriteFile(filepath.Join(r.runDir, "summary.md"), []byte(summary.String()), 0o600); err != nil {
		t.Errorf("write shared-source summary: %v", err)
	}
	marker := filepath.Join(r.runDir, "scene-preserved.txt")
	if r.journal.Session != "" && r.journal.Cleanup != "pass" {
		_ = os.WriteFile(marker, []byte("Scene preserved; cleanup is pending.\n"), 0o600)
	} else if r.journal.Cleanup == "pass" {
		_ = os.Remove(marker)
	}
}

func sharedSourceAttemptVerdict(receipts []sharedSourceStageReceipt, product bool) string {
	verdict := "not_run"
	for _, receipt := range receipts {
		value := receipt.Harness
		if product {
			value = receipt.Product
		}
		if value == "fail" {
			return "fail"
		}
		if value == "pass" {
			verdict = "pass"
		}
	}
	return verdict
}

func (r *sharedSourceRun) latestStage(name string) (sharedSourceStageReceipt, bool) {
	receipts := r.journal.Stages[name]
	if len(receipts) == 0 {
		return sharedSourceStageReceipt{}, false
	}
	return receipts[len(receipts)-1], true
}

func sharedSourceChildPath(root, path string) bool {
	relative, err := filepath.Rel(root, path)
	return err == nil && relative != "." && relative != ".." && !strings.HasPrefix(relative, ".."+string(filepath.Separator))
}

func sharedSourcePriorCommandCount(runDir string) int {
	entries, err := os.ReadDir(filepath.Join(runDir, "commands"))
	if err != nil {
		return 0
	}
	maximum := 0
	for _, entry := range entries {
		prefix, _, ok := strings.Cut(entry.Name(), "-")
		if !ok {
			continue
		}
		value, err := strconv.Atoi(prefix)
		if err == nil {
			maximum = max(maximum, value)
		}
	}
	return maximum
}

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

func sharedSourceValue(value string) string {
	if value == "" {
		return "not_recorded"
	}
	return value
}