//go:build functional

package spectests_test

import (
	"encoding/json"
	"errors"
	"io/fs"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
)

const (
	devOwnerArtifactsEnabled = "CP9_DEV_OWNER_ARTIFACTS_ENABLED"
	devOwnerArtifactsRunDir  = "CP9_DEV_OWNER_ARTIFACTS_RUN_DIR"
	devOwnerArtifactsRetry   = "CP9_DEV_OWNER_ARTIFACTS_RETRY_STAGE"
	devOwnerArtifactsDestroy = "CP9_DEV_OWNER_ARTIFACTS_DESTROY_SESSION"
)

type devOwnerRun struct {
	*ownerSessionRun
	configRelative       string
	sharedCRDRelative    string
	dedicatedCRDRelative string
	dataflowRelative     string
}

// # A Dev session uses owner artifacts through refresh and override
//
// This case proves one complete Dev journey in one Remote session: a fresh
// session starts from an immutable owner-artifact snapshot, focused refreshes
// change only their selections and reject malformed owner inputs, and a temporary
// service-image override returns exactly to the owner-resolved state.
//
// It covers issue [#1002](https://github.com/tidbcloud/controlplane9/issues/1002).
//
// ## Agent execution
//
// **Mode: single agent.** Run every phase in order against the same owner fixtures
// and session. Create the session once and clean it once. Spawn no sub-agents
// because artifact publication, snapshot changes, workload observations, 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, plus the phase's
// required snapshot/owner-state check. Run the phase when those checks pass.
// Otherwise record that phase as `BLOCKED`, Product `NOT RUN`, with the failed
// health evidence, continue recording any later independently safe phase, and
// perform the one cleanup.
//
// Native Go entrypoint:
//
// ```bash
// CP9_DEV_OWNER_ARTIFACTS_ENABLED=1 \
//   CP9_UI_API_AWS_KEY_PAIR='<existing-key-pair>' \
//   CP9_UI_API_SSH_PRIVATE_KEY='<readable-private-key-path>' \
//   GOTOOLCHAIN=go1.26.1 GO_FUNCTIONAL_TEST_PACKAGES=./spec-tests \
//   GO_FUNCTIONAL_TEST_RUN='^TestFunctional_DevOwnerArtifacts$$' \
//   make test-functional
// ```
//
// After diagnosing a first red, continue only that same evidence directory and
// named failed Go subtest, then rerun the entrypoint above. Completed product
// actions are read from their receipts and are not replayed; an unknown action or
// missing intermediate snapshot fails closed. Valid retry names are
// `fresh-owner-snapshot`, `focused-owner-refresh`, and
// `image-override-and-restore`.
// A recorded failed initial `up` can resume only when the same session still
// reports a failed `session_ready` lifecycle step. Each recovery gets a separate
// `up-resume-N` receipt; the original failure remains. A successful or unknown
// action is never replayed, and other failed lifecycle steps require diagnosis.
// Before selecting this recovery, preserve the lifecycle state and confirm all
// other steps are done, the host/config are unchanged, and the controller is
// healthy: ordinary `up` can otherwise repair tunnels or restart the controller.
//
// ```bash
// export CP9_DEV_OWNER_ARTIFACTS_RUN_DIR='<existing-evidence-directory>'
// export CP9_DEV_OWNER_ARTIFACTS_RETRY_STAGE='focused-owner-refresh'
// ```
//
// To clean up a preserved first-red scene without retrying product work, clear
// the retry selector, request exact destruction, and rerun the same entrypoint.
//
// ```bash
// export CP9_DEV_OWNER_ARTIFACTS_RUN_DIR='<existing-evidence-directory>'
// unset CP9_DEV_OWNER_ARTIFACTS_RETRY_STAGE
// export CP9_DEV_OWNER_ARTIFACTS_DESTROY_SESSION=1
// ```
//
// ## 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 Dev
//   `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.
// - Freeze the Dev owner repositories. Starter/Essential v1 activation and
//   runtime proof are outside this case.
// - The instruction to execute this case authorizes the session, private owner
//   fixtures, focused refreshes, rejected malformed inputs, temporary image
//   override, restoration, and exact cleanup. GitHub Device Flow is the only
//   possible human step.
//
// Do not create a TiDB cluster.
//
// ## Execute
//
// ### 1. Create the fresh owner-artifact baseline once
//
// Before phase 1, record the CP9 commit, upstream and consumed owner commits,
// account, region, session, TTL, and Dev artifact environment. Prepare unique
// markers in one shared and one Dedicated CRD plus a known prior `config-center`
// owner image.
//
// If these run prerequisites do not complete, do not run any phase. Record all
// three phases as unexecuted with Product and per-phase Harness `NOT RUN` and a
// prerequisite reason. Run-level Harness is `FAIL` when the prerequisite failure
// is a harness defect and `NOT RUN` otherwise. Attempt cleanup for any run-owned
// state, emit the run summary with all three phases listed as unexecuted, and
// return a nonzero result.
//
// After the prerequisites complete, run setup and the initial `cp9 up` once;
// require the generated session to reach `ready`. A failure here is an executed
// phase 1 result; later phases follow their normal health gates.
//
// Read its immutable snapshot. For every in-scope declared service, observe the
// current running workload and require every active instance to use the snapshot
// image. For every CRD group owned by an active API, require the live definition
// to match the snapshot; this AWS session excludes the inactive Dedicated GCP
// provider group. Require both unique CRD markers and at least one live owner
// image on the shared path to differ from its old fallback.
//
// ### 2. Refresh only the selected owner artifacts
//
// Publish the next `config-center` owner image and refresh only that service.
// Require its digest and current Ready Pods to change while every peer image and
// all live CRDs remain unchanged.
//
// Publish the next `cluster-ng-service-crds` content and refresh only that group.
// Require its digest and live definitions to change while every peer CRD and all
// Pods remain unchanged.
//
// Publish one ambiguous service input and one non-CRD document in turn. Each
// selected refresh must return nonzero and leave the complete snapshot, running
// Pods, and live CRDs unchanged. Restore the owner inputs before continuing.
//
// ### 3. Override one image and return to the owner state
//
// Record the current owner-resolved snapshot and `config-center` Pods. Submit an
// invalid image and require rejection with no change. Submit a known valid image
// different from the owner image and require only `config-center` plus its current
// Ready Pods to use that exact override.
//
// Refresh `config-center` again without an image. Require the complete snapshot to
// return byte-for-byte to the recorded owner snapshot and every current Ready Pod
// to run its restored snapshot image.
//
// ### 4. Clean up once
//
// Destroy only this session and generated worktree, leave the caller-owned key
// pair untouched, discard the private owner mirrors, and record the exact cleanup
// receipt plus durable accepted hosted-cleanup handoff. Do not wait for the
// asynchronous hosted cleanup run.
//
// ## Verdicts
//
// ### Phase 1: fresh owner snapshot
//
// #### Product
//
// `PASS` requires the fresh session snapshot, every in-scope active workload, and
// every live CRD to match the frozen owner inputs, including both unique CRD
// markers and a live shared image that differs from its old fallback. A stale
// image, copied CRD, or partial snapshot is `FAIL`.
//
// #### Harness
//
// `PASS` requires the owner mirrors, immutable snapshot, workload/CRD locators,
// and evidence under `session/` plus `phase-results/fresh-owner-snapshot.json` to
// complete. A fixture, locator, parser, or evidence defect is Harness `FAIL` and
// leaves unobserved Product assertions `NOT RUN`.
//
// ### Phase 2: focused owner refresh
//
// #### Product
//
// `PASS` requires each selected valid refresh to change only its target and its
// live consumers, and each malformed input to return nonzero while the complete
// snapshot and live consumers remain unchanged. Peer mutation, accepted malformed
// input, or a partial snapshot is `FAIL`. A failed pre-phase health/state check is
// `BLOCKED` and Product `NOT RUN`, not `FAIL`.
//
// #### Harness
//
// `PASS` requires the published owner commits, before/after snapshots, workload
// and CRD inventories, rejection logs, restoration, and
// `phase-results/focused-owner-refresh.json` to complete. Fixture control or
// comparison failure is Harness `FAIL`.
//
// ### Phase 3: image override and restore
//
// #### Product
//
// `PASS` requires invalid-image rejection, the valid override to affect only
// `config-center` and its Ready Pods, and an owner refresh to restore the complete
// snapshot byte-for-byte. A retained override, peer mutation, wrong Pod image, or
// partial restoration is `FAIL`. A failed pre-phase health/state check is
// `BLOCKED` and Product `NOT RUN`.
//
// #### Harness
//
// `PASS` requires the override/restore snapshots, live workload inventories,
// image observations, and `phase-results/image-override-and-restore.json` to
// complete. A locator, comparison, or evidence 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`.
//
// A missing tool, expired AWS login, inaccessible owner 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.
//
// Starter/Essential v1 activation and runtime behavior are not run or claimed by
// this case.
//
// ### Cleanup
//
// Cleanup `PASS` requires private fixtures and local run state removed, the exact
// session cleanup accepted, and required receipts recorded. Known residue or a
// missing handoff is cleanup `FAIL`, not product `FAIL`. Cleanup is `NOT RUN` only
// when the case created no run-owned state.

func TestFunctional_DevOwnerArtifacts(t *testing.T) {
	if os.Getenv(devOwnerArtifactsEnabled) != "1" {
		t.Skipf("set %s=1 with the existing key pair and private key inputs to run the Dev owner-artifact journey", devOwnerArtifactsEnabled)
	}
	keyPair, privateKey := os.Getenv("CP9_UI_API_AWS_KEY_PAIR"), os.Getenv("CP9_UI_API_SSH_PRIVATE_KEY")
	if keyPair == "" || privateKey == "" {
		t.Skip("CP9_UI_API_AWS_KEY_PAIR and CP9_UI_API_SSH_PRIVATE_KEY are required together; no run-owned state was created")
	}
	if info, err := os.Stat(privateKey); err != nil || !info.Mode().IsRegular() {
		t.Skipf("CP9_UI_API_SSH_PRIVATE_KEY must name a readable regular file: %v", err)
	}
	run := &devOwnerRun{ownerSessionRun: newOwnerSessionRun(t, "dev-owner-artifacts-session", "dev", devOwnerArtifactsRunDir, devOwnerArtifactsRetry, devOwnerArtifactsDestroy,
		"fresh-owner-snapshot", "focused-owner-refresh", "image-override-and-restore")}
	t.Cleanup(func() { run.finalize(t) })
	if run.cleanupRequested() {
		t.Run("cleanup", func(t *testing.T) {
			run.activeFailure = "harness"
			run.cleanup(t)
		})
		return
	}
	if run.actionCompleted("prepare-session") {
		run.loadFixturePaths(t)
	}

	t.Run("fresh-owner-snapshot", func(t *testing.T) {
		run.runStage(t, "fresh-owner-snapshot", run.actionsReplaySafe("prepare-session", "configure-session", "setup", "up"), func(t *testing.T) {
			if !run.actionCompleted("prepare-session") {
				run.prepareDevFixture(t)
			}
			run.startSession(t, "dev")
			run.activeFailure = "product"
			snapshot := run.assertSnapshot(t, "dev")
			run.copyFile(t, filepath.Join(run.root, "app-delivery-fixture.txt"), filepath.Join(run.root, "app-delivery-consumed.txt"))
			evidence := filepath.Join(run.sessionDir, "inventory-initial")
			run.assertRunningWorkloads(t, evidence, snapshot)
			run.assertLiveCRDs(t, evidence, snapshot)
			run.assertDevMarkers(t)
			if !run.hasLiveOwnerImageDifferentFromFallback(t, snapshot) {
				run.activeFailure = "product"
				t.Fatal("no shared-path live owner image differs from its checked-in fallback")
			}
		})
	})

	t.Run("focused-owner-refresh", func(t *testing.T) {
		run.runStage(t, "focused-owner-refresh", run.actionsReplaySafe("refresh-config-center", "refresh-cluster-ng-crds", "refresh-malformed-service", "refresh-malformed-crd"), func(t *testing.T) {
			if !run.actionsSucceeded("prepare-session", "configure-session", "setup", "up") {
				run.activeFailure = "blocked"
				t.Skip("phase blocked: the original session actions are incomplete")
			}
			run.requireActionOrder(t, "refresh-config-center", "refresh-cluster-ng-crds", "refresh-malformed-service", "refresh-malformed-crd")
			run.requireActionOrder(t, "refresh-cluster-ng-crds", "refresh-malformed-service", "refresh-malformed-crd")
			run.requireActionOrder(t, "refresh-malformed-service", "refresh-malformed-crd")
			run.validateFixturePaths(t, true)
			run.restoreRejectedFixtureInputs(t)
			if !run.preflight(t, "dev", true) {
				t.Skip("phase blocked: the original session or owner state is not safely observable")
			}
			run.focusedRefresh(t)
		})
	})

	t.Run("image-override-and-restore", func(t *testing.T) {
		run.runStage(t, "image-override-and-restore", run.actionsReplaySafe("override-invalid-image", "override-config-center", "restore-config-center-owner"), func(t *testing.T) {
			overrideName, overrideImage := "", ""
			if run.actionCompleted("override-config-center") && !run.actionCompleted("restore-config-center-owner") {
				overrideName, overrideImage = "config-center", run.contractStagingImage(t, "config-center")
			}
			if !run.actionsSucceeded("prepare-session", "configure-session", "setup", "up") {
				run.activeFailure = "blocked"
				t.Skip("phase blocked: the original session actions are incomplete")
			}
			if !run.preflightState(t, "dev", overrideName, overrideImage, true) {
				t.Skip("phase blocked: the original session or owner state is not safely observable")
			}
			if _, err := os.Stat(filepath.Join(run.sessionDir, "snapshot-before-override.json")); errors.Is(err, os.ErrNotExist) {
				run.copyCurrentSnapshot(t, "snapshot-before-override.json")
			} else if err != nil {
				t.Fatalf("inspect snapshot before override: %v", err)
			}
			run.imageOverrideAndRestore(t)
		})
	})

	t.Run("cleanup", func(t *testing.T) {
		_, sessionStarted := run.journal.Actions["prepare-session"]
		if os.Getenv(run.destroyEnv) != "1" && sessionStarted && !run.stagesPassed("fresh-owner-snapshot", "focused-owner-refresh", "image-override-and-restore") {
			t.Skip("first red is preserved; exact cleanup remains pending")
		}
		if !run.hasOwnedState() {
			t.Skip("no run-owned state needs cleanup")
		}
		run.activeFailure = "harness"
		run.cleanup(t)
	})
}

func (r *devOwnerRun) prepareDevFixture(t *testing.T) {
	t.Helper()
	r.prepareRepositories(t)
	r.activeFailure = "harness"
	r.configRelative = filepath.Join("config-center", "envs", "dev", "kustomization.yaml")
	r.sharedCRDRelative = r.firstYAML(t, filepath.Join("dev", "dedicated-cluster-ng-service", "cluster-ng-service-crds", "envs", "dev", "base"))
	r.dedicatedCRDRelative = filepath.Join("dedicated-cluster-service", "cluster-controller-crd", "envs", "dev", "bases", "cloud.pingcap.com_backups.yaml")
	r.dataflowRelative = filepath.Join("dev", "dataflow-service-ng", "env", "dev", "http-version.yaml")
	r.validateFixturePaths(t, true)
	for name, value := range map[string]string{
		"config": r.configRelative, "shared-crd": r.sharedCRDRelative,
		"dedicated-crd": r.dedicatedCRDRelative, "dataflow": r.dataflowRelative,
	} {
		r.writeText(t, "dev-"+name+"-relative.txt", value)
	}
	r.copyOwnerFile(t, r.configRelative, "config-center-original.yaml")
	r.copyOwnerFile(t, r.sharedCRDRelative, "shared-crd-original.yaml")
	r.copyOwnerFile(t, r.dedicatedCRDRelative, "dedicated-crd-original.yaml")
	r.copyOwnerFile(t, r.dataflowRelative, "dataflow-original.yaml")
	replaceConfigTag(t, filepath.Join(r.ownerWork, r.configRelative), resolverConfigTag)
	annotateOwnerCRD(t, filepath.Join(r.ownerWork, r.sharedCRDRelative), "fresh-dev")
	annotateOwnerCRD(t, filepath.Join(r.ownerWork, r.dedicatedCRDRelative), "fresh-dev")
	r.commitFixture(t, "prepare Dev owner artifact inputs", r.configRelative, r.sharedCRDRelative, r.dedicatedCRDRelative)
	r.freezeRepositories(t)
}

func (r *devOwnerRun) loadFixturePaths(t *testing.T) {
	t.Helper()
	r.configRelative = r.readText(t, "dev-config-relative.txt")
	r.sharedCRDRelative = r.readText(t, "dev-shared-crd-relative.txt")
	r.dedicatedCRDRelative = r.readText(t, "dev-dedicated-crd-relative.txt")
	r.dataflowRelative = r.readText(t, "dev-dataflow-relative.txt")
	r.validateFixturePaths(t, false)
}

func (r *devOwnerRun) validateFixturePaths(t *testing.T, requireFiles bool) {
	t.Helper()
	for name, relative := range map[string]string{
		"config": r.configRelative, "shared-crd": r.sharedCRDRelative,
		"dedicated-crd": r.dedicatedCRDRelative, "dataflow": r.dataflowRelative,
	} {
		relative = filepath.ToSlash(relative)
		if !fs.ValidPath(relative) || relative == "." {
			t.Fatalf("invalid saved %s fixture path %q", name, relative)
		}
		if requireFiles {
			if _, err := secureResolverPath(r.ownerWork, relative, false); err != nil {
				t.Fatalf("unsafe saved %s fixture path: %v", name, err)
			}
		}
	}
}

func (r *devOwnerRun) restoreRejectedFixtureInputs(t *testing.T) {
	t.Helper()
	if !r.actionCompleted("refresh-malformed-service") && !r.actionCompleted("refresh-malformed-crd") {
		return
	}
	r.copyFile(t, filepath.Join(r.root, "config-center-original.yaml"), filepath.Join(r.ownerWork, r.configRelative))
	r.copyFile(t, filepath.Join(r.root, "shared-crd-original.yaml"), filepath.Join(r.ownerWork, r.sharedCRDRelative))
	r.copyFile(t, filepath.Join(r.root, "dataflow-original.yaml"), filepath.Join(r.ownerWork, r.dataflowRelative))
	r.publishFixture(t, "resume-valid-focused-inputs", r.configRelative, r.sharedCRDRelative, r.dataflowRelative)
}

func (r *devOwnerRun) focusedRefresh(t *testing.T) {
	t.Helper()
	r.activeFailure = "harness"
	r.copyOwnerFile(t, r.configRelative, "config-center-before-focused.yaml")
	r.copyOwnerFile(t, r.sharedCRDRelative, "shared-crd-before-focused.yaml")
	configState := filepath.Join(r.root, "config-center-before-focused.yaml")
	crdState := filepath.Join(r.root, "shared-crd-before-focused.yaml")
	defer func() {
		if r.actionSucceeded("refresh-config-center") {
			configState = filepath.Join(r.root, "config-center-original.yaml")
		}
		if r.actionSucceeded("refresh-cluster-ng-crds") {
			crdState = filepath.Join(r.root, "shared-crd-original.yaml")
		}
		r.restoreFocusedState(t, configState, crdState)
	}()
	beforeConfig := r.snapshotBeforeAction(t, "snapshot-before-config-center.json", "refresh-config-center")
	var nextConfigCommit string
	if r.actionCompleted("refresh-config-center") {
		nextConfigCommit = r.readText(t, filepath.Join("read-only", "publish-next-config-center-image-commit.txt"))
	} else {
		r.copyFile(t, filepath.Join(r.root, "config-center-original.yaml"), filepath.Join(r.ownerWork, r.configRelative))
		nextConfigCommit = r.publishFixture(t, "publish-next-config-center-image", r.configRelative)
	}
	r.mustRefresh(t, "refresh-config-center", "artifacts=config-center")
	afterConfig := r.snapshotAfterAction(t, "snapshot-after-config-center.json", "refresh-config-center",
		"refresh-cluster-ng-crds", "refresh-malformed-service", "refresh-malformed-crd",
		"override-invalid-image", "override-config-center", "restore-config-center-owner")
	r.activeFailure = "product"
	assertSelectedOwnerChange(t, beforeConfig, afterConfig, "config-center", nextConfigCommit)
	if !r.actionCompleted("refresh-cluster-ng-crds") {
		r.assertCurrentOwnerSnapshot(t, "dev", "", "")
		configEvidence := filepath.Join(r.sessionDir, "inventory-after-config-center")
		r.assertRunningWorkloads(t, configEvidence, afterConfig)
		r.assertLiveCRDs(t, configEvidence, afterConfig)
	}

	r.activeFailure = "harness"
	beforeCRD := r.snapshotBeforeAction(t, "snapshot-before-crds.json", "refresh-cluster-ng-crds")
	var nextCRDCommit string
	if r.actionCompleted("refresh-cluster-ng-crds") {
		nextCRDCommit = r.readText(t, filepath.Join("read-only", "publish-next-cluster-ng-crds-commit.txt"))
	} else {
		r.copyFile(t, filepath.Join(r.root, "shared-crd-original.yaml"), filepath.Join(r.ownerWork, r.sharedCRDRelative))
		nextCRDCommit = r.publishFixture(t, "publish-next-cluster-ng-crds", r.sharedCRDRelative)
	}
	r.mustRefresh(t, "refresh-cluster-ng-crds", "artifacts=cluster-ng-service-crds")
	afterCRD := r.snapshotAfterAction(t, "snapshot-after-crds.json", "refresh-cluster-ng-crds",
		"refresh-malformed-service", "refresh-malformed-crd",
		"override-invalid-image", "override-config-center", "restore-config-center-owner")
	r.activeFailure = "product"
	assertSelectedOwnerChange(t, beforeCRD, afterCRD, "cluster-ng-service-crds", nextCRDCommit)
	r.assertCurrentOwnerSnapshot(t, "dev", "", "")
	crdEvidence := filepath.Join(r.sessionDir, "inventory-after-crds")
	r.assertRunningWorkloads(t, crdEvidence, afterCRD)
	r.assertLiveCRDs(t, crdEvidence, afterCRD)
	r.assertSharedMarker(t, "refreshed-marker-crd", nil)

	r.activeFailure = "harness"
	dataflowPath := filepath.Join(r.ownerWork, r.dataflowRelative)
	originalDataflow, err := os.ReadFile(filepath.Join(r.root, "dataflow-original.yaml"))
	if err != nil {
		t.Fatalf("read original dataflow fixture: %v", err)
	}
	if !r.actionCompleted("refresh-malformed-service") {
		currentDataflow, err := os.ReadFile(dataflowPath)
		if err != nil {
			t.Fatalf("read current dataflow fixture: %v", err)
		}
		r.writeBytesAt(t, dataflowPath, append(currentDataflow, originalDataflow...))
	}
	if !r.actionCompleted("refresh-malformed-service") {
		r.publishFixture(t, "publish-malformed-dataflow", r.dataflowRelative)
	}
	r.expectRefreshRejection(t, "refresh-malformed-service", "dataflow")
	r.writeBytesAt(t, dataflowPath, originalDataflow)
	r.publishFixture(t, "restore-dataflow", r.dataflowRelative)

	crdPath := filepath.Join(r.ownerWork, r.sharedCRDRelative)
	if !r.actionCompleted("refresh-malformed-crd") {
		r.writeBytesAt(t, crdPath, []byte("apiVersion: v1\nkind: ConfigMap\nmetadata:\n  name: not-a-crd\n"))
	}
	if !r.actionCompleted("refresh-malformed-crd") {
		r.publishFixture(t, "publish-malformed-crd", r.sharedCRDRelative)
	}
	r.expectRefreshRejection(t, "refresh-malformed-crd", "cluster-ng-service-crds")
	r.copyFile(t, filepath.Join(r.root, "shared-crd-original.yaml"), crdPath)
	r.publishFixture(t, "restore-after-malformed-crd", r.sharedCRDRelative)
}

func (r *devOwnerRun) snapshotBeforeAction(t *testing.T, name, action string) resolverSnapshot {
	t.Helper()
	r.activeFailure = "harness"
	path := filepath.Join(r.sessionDir, name)
	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
		if _, started := r.journal.Actions[action]; started {
			t.Fatalf("%s is missing for recorded action %s", name, action)
		}
		r.copyFile(t, r.snapshot, path)
	} else if err != nil {
		t.Fatalf("inspect %s: %v", name, err)
	}
	var snapshot resolverSnapshot
	r.readJSON(t, path, &snapshot)
	return snapshot
}

func (r *devOwnerRun) snapshotAfterAction(t *testing.T, name, action string, later ...string) resolverSnapshot {
	t.Helper()
	r.activeFailure = "harness"
	path := filepath.Join(r.sessionDir, name)
	if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
		for _, candidate := range later {
			if _, started := r.journal.Actions[candidate]; started {
				t.Fatalf("%s is missing after %s and later action %s is already recorded", name, action, candidate)
			}
		}
		r.copyFile(t, r.snapshot, path)
	} else if err != nil {
		t.Fatalf("inspect %s: %v", name, err)
	}
	var snapshot resolverSnapshot
	r.readJSON(t, path, &snapshot)
	return snapshot
}

func (r *devOwnerRun) requireActionOrder(t *testing.T, action string, later ...string) {
	t.Helper()
	for _, candidate := range later {
		if _, recorded := r.journal.Actions[candidate]; recorded && !r.actionCompleted(action) {
			t.Fatalf("action %s is recorded without completed predecessor %s", candidate, action)
		}
	}
}

func (r *devOwnerRun) restoreFocusedState(t *testing.T, configState, crdState string) {
	t.Helper()
	failure := r.activeFailure
	r.activeFailure = "harness"
	r.copyFile(t, configState, filepath.Join(r.ownerWork, r.configRelative))
	r.copyFile(t, crdState, filepath.Join(r.ownerWork, r.sharedCRDRelative))
	r.copyFile(t, filepath.Join(r.root, "dataflow-original.yaml"), filepath.Join(r.ownerWork, r.dataflowRelative))
	r.publishFixture(t, "restore-after-focused-refresh", r.configRelative, r.sharedCRDRelative, r.dataflowRelative)
	r.copyCurrentSnapshot(t, "snapshot-before-override.json")
	r.activeFailure = failure
}

func (r *devOwnerRun) imageOverrideAndRestore(t *testing.T) {
	t.Helper()
	r.requireActionOrder(t, "override-invalid-image", "override-config-center", "restore-config-center-owner")
	r.requireActionOrder(t, "override-config-center", "restore-config-center-owner")
	var before resolverSnapshot
	r.readJSON(t, filepath.Join(r.sessionDir, "snapshot-before-override.json"), &before)
	if r.actionCompleted("override-config-center") || r.actionCompleted("restore-config-center-owner") {
		result := r.refresh(t, "override-invalid-image", "artifacts=config-center", "image=not-an-image")
		if result.ExitCode == 0 {
			r.activeFailure = "product"
			t.Fatal("override-invalid-image unexpectedly accepted invalid input")
		}
	} else {
		r.expectRefreshRejection(t, "override-invalid-image", "config-center", "image=not-an-image")
	}
	ownerImage := findSnapshotArtifact(t, before, "config-center").Image
	overrideImage := r.contractStagingImage(t, "config-center")
	if ownerImage == overrideImage {
		r.activeFailure = "harness"
		t.Fatalf("override image %q equals the current owner image", overrideImage)
	}
	r.mustRefresh(t, "override-config-center", "artifacts=config-center", "image="+overrideImage)
	afterOverride := r.snapshotAfterAction(t, "snapshot-after-image-override.json", "override-config-center", "restore-config-center-owner")
	r.activeFailure = "product"
	assertSelectedOwnerChange(t, before, afterOverride, "config-center", findSnapshotArtifact(t, before, "config-center").Commit)
	if !r.actionCompleted("restore-config-center-owner") {
		r.assertCurrentOwnerSnapshot(t, "dev", "config-center", overrideImage)
		overrideEvidence := filepath.Join(r.sessionDir, "inventory-after-override")
		r.assertRunningWorkloads(t, overrideEvidence, afterOverride)
	}

	r.mustRefresh(t, "restore-config-center-owner", "artifacts=config-center")
	afterRestore := r.snapshotAfterAction(t, "snapshot-after-image-restore.json", "restore-config-center-owner")
	r.activeFailure = "product"
	assertOwnerRestore(t, before, afterRestore, r.fixtureHead)
	r.assertCurrentOwnerSnapshot(t, "dev", "", "")
	restoreEvidence := filepath.Join(r.sessionDir, "inventory-after-restore")
	r.assertRunningWorkloads(t, restoreEvidence, afterRestore)
}

func (r *devOwnerRun) expectRefreshRejection(t *testing.T, label, artifact string, extra ...string) {
	t.Helper()
	before := r.snapshotBeforeAction(t, "snapshot-before-"+sanitizeResolverLabel(label)+".json", label)
	alreadyCompleted := r.actionCompleted(label)
	args := []string{"artifacts=" + artifact}
	args = append(args, extra...)
	r.activeFailure = "product"
	result := r.refresh(t, label, args...)
	if result.ExitCode == 0 {
		t.Fatalf("%s unexpectedly accepted invalid input", label)
	}
	afterPath := filepath.Join(r.sessionDir, "snapshot-after-"+sanitizeResolverLabel(label)+".json")
	if _, err := os.Stat(afterPath); errors.Is(err, os.ErrNotExist) {
		if alreadyCompleted {
			r.activeFailure = "harness"
			t.Fatalf("snapshot after recorded rejection %s is unavailable", label)
		}
		r.copyFile(t, r.snapshot, afterPath)
	} else if err != nil {
		r.activeFailure = "harness"
		t.Fatalf("inspect snapshot after %s: %v", label, err)
	}
	var after resolverSnapshot
	r.readJSON(t, afterPath, &after)
	if !reflect.DeepEqual(after, before) {
		r.activeFailure = "product"
		t.Fatalf("%s changed the immutable snapshot", label)
	}
	evidence := filepath.Join(r.sessionDir, "inventory-after-"+sanitizeResolverLabel(label))
	r.assertRunningWorkloads(t, evidence, before)
	r.assertLiveCRDs(t, evidence, before)
}

func (r *devOwnerRun) mustRefresh(t *testing.T, label string, args ...string) {
	t.Helper()
	r.activeFailure = "product"
	result := r.refresh(t, label, args...)
	if result.ExitCode != 0 {
		t.Fatalf("%s failed with exit %d: %s", label, result.ExitCode, result.Stderr)
	}
}

func (r *devOwnerRun) copyCurrentSnapshot(t *testing.T, name string) resolverSnapshot {
	t.Helper()
	r.activeFailure = "harness"
	target := filepath.Join(r.sessionDir, name)
	r.copyFile(t, r.snapshot, target)
	var snapshot resolverSnapshot
	r.readJSON(t, target, &snapshot)
	return snapshot
}

func (r *devOwnerRun) copyOwnerFile(t *testing.T, relative, evidence string) {
	t.Helper()
	r.copyFile(t, filepath.Join(r.ownerWork, relative), filepath.Join(r.root, evidence))
}

func (r *devOwnerRun) assertDevMarkers(t *testing.T) {
	t.Helper()
	r.assertSharedMarker(t, "shared-marker-crd", stringPointer("fresh-dev"))
	dedicatedName := r.firstCRDName(t, filepath.Join(r.root, "dedicated-crd-original.yaml"))
	var sessionContext ownerSessionContext
	r.readJSON(t, r.contextPath, &sessionContext)
	object := r.kubectlObject(t, r.sessionDir, "dedicated-marker-crd", r.dedicatedKubectl(sessionContext.Host.DedicatedAPI.RemoteAdminKubeconfigPath), "get", "customresourcedefinition", dedicatedName, "-o", "json")
	if object.Metadata.Annotations["cp9.tidbcloud.com/spec-artifact"] != "fresh-dev" {
		r.activeFailure = "product"
		t.Fatalf("Dedicated Dev CRD %s lacks the live owner marker", dedicatedName)
	}
}

func (r *devOwnerRun) assertSharedMarker(t *testing.T, label string, want *string) {
	t.Helper()
	name := r.firstCRDName(t, filepath.Join(r.root, "shared-crd-original.yaml"))
	object := r.kubectlObject(t, r.sessionDir, label, r.hostKubectl(), "get", "customresourcedefinition", name, "-o", "json")
	got, present := object.Metadata.Annotations["cp9.tidbcloud.com/spec-artifact"]
	if want == nil && present {
		r.activeFailure = "product"
		t.Fatalf("shared Dev CRD %s retained marker %q", name, got)
	}
	if want != nil && (!present || got != *want) {
		r.activeFailure = "product"
		t.Fatalf("shared Dev CRD %s marker=%q; want %q", name, got, *want)
	}
}

func (r *devOwnerRun) hasLiveOwnerImageDifferentFromFallback(t *testing.T, snapshot resolverSnapshot) bool {
	t.Helper()
	data, err := os.ReadFile(filepath.Join(r.cp9Root, "internal", "sourcebacked", "contracts.json"))
	if err != nil {
		r.activeFailure = "harness"
		t.Fatalf("read source-backed contracts: %v", err)
	}
	var contracts ownerContractsFile
	if err := json.Unmarshal(data, &contracts); err != nil {
		r.activeFailure = "harness"
		t.Fatalf("parse source-backed contracts: %v", err)
	}
	fallbacks := map[string]string{}
	backends := map[string]string{}
	for _, contract := range contracts.Contracts {
		fallbacks[contract.Name] = contract.StagingImage
		backends[contract.Name] = contract.Backend.Kind
	}
	var rows []string
	for _, artifact := range snapshot.Artifacts {
		if artifact.Kind == "service" && backends[artifact.Name] == "k3d-control-plane" && fallbacks[artifact.Name] != "" && artifact.Image != fallbacks[artifact.Name] {
			rows = append(rows, strings.Join([]string{backends[artifact.Name], artifact.Name, fallbacks[artifact.Name], artifact.Image}, "\t"))
		}
	}
	r.writeBytesAt(t, filepath.Join(r.sessionDir, "owner-images-different-from-fallback.tsv"), []byte(strings.Join(rows, "\n")+"\n"))
	return len(rows) != 0
}

func assertSelectedOwnerChange(t *testing.T, before, after resolverSnapshot, selected, commit string) {
	t.Helper()
	if before.Digest == after.Digest {
		t.Fatalf("%s refresh did not change the snapshot digest", selected)
	}
	beforeByName := map[string]resolverSnapshotArtifact{}
	afterByName := map[string]resolverSnapshotArtifact{}
	for _, artifact := range before.Artifacts {
		beforeByName[artifact.Name] = artifact
	}
	for _, artifact := range after.Artifacts {
		afterByName[artifact.Name] = artifact
	}
	if len(beforeByName) != len(afterByName) {
		t.Fatalf("%s refresh changed the artifact set", selected)
	}
	for name, beforeArtifact := range beforeByName {
		afterArtifact, ok := afterByName[name]
		if !ok {
			t.Fatalf("%s refresh removed artifact %s", selected, name)
		}
		if name == selected {
			if beforeArtifact.Digest == afterArtifact.Digest || afterArtifact.Commit != commit {
				t.Fatalf("%s refresh digest/commit did not advance exactly", selected)
			}
			continue
		}
		if !reflect.DeepEqual(beforeArtifact, afterArtifact) {
			t.Fatalf("%s refresh changed peer %s", selected, name)
		}
	}
}

func assertOwnerRestore(t *testing.T, before, after resolverSnapshot, commit string) {
	t.Helper()
	want := before
	want.Artifacts = append([]resolverSnapshotArtifact(nil), before.Artifacts...)
	found := false
	for i := range want.Artifacts {
		if want.Artifacts[i].Name == "config-center" {
			want.Artifacts[i].Commit = commit
			found = true
		}
	}
	if !found {
		t.Fatal("owner snapshot has no config-center artifact")
	}
	want.Digest = after.Digest
	if !reflect.DeepEqual(want, after) {
		t.Fatal("owner refresh did not restore the complete snapshot")
	}
	digest, err := resolverSnapshotDigest(after)
	if err != nil {
		t.Fatalf("calculate restored snapshot digest: %v", err)
	}
	if after.Digest != digest {
		t.Fatalf("restored snapshot digest=%q, want %q", after.Digest, digest)
	}
}

func stringPointer(value string) *string { return &value }