//go:build functional

package spectests_test

// Nested structs keep single-use JSON wire projections beside their assertions.
//revive:disable:nested-structs

import (
	"crypto/sha256"
	"crypto/x509"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"maps"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"regexp"
	"slices"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/tidbcloud/controlplane9/internal/host"
)

/*
# Dedicated GCP regional runtime survives an isolated controller loss

This case proves the GCP-specific public contract without duplicating the
Public UI Product journey. One Dev Full session keeps its AWS primary runtime,
adds the supported `pingcap-gardener/us-west1` GCE+k3d runtime, routes day-2
commands through the persisted two-runtime graph, and resumes only the failed
GCP controller.

## Agent execution

**Mode: single agent.** The checkout, two controllers, scoped fault, and cleanup
share one session. Do not spawn sub-agents.

Run the Go journey from the exact checkout with existing caller-owned key inputs:

```bash
CP9_GCP_RUNTIME_ENABLED=1 \
  CP9_UI_API_AWS_KEY_PAIR="$CP9_UI_API_AWS_KEY_PAIR" \
  CP9_UI_API_SSH_PRIVATE_KEY="$CP9_UI_API_SSH_PRIVATE_KEY" \
  GOTOOLCHAIN=go1.26.1 GO_FUNCTIONAL_TEST_PACKAGES=./spec-tests \
  GO_FUNCTIONAL_TEST_RUN='^TestFunctional_DedicatedGCPRegionalRuntime$$' make test-functional
```

The opt-in is checked before any build, preparation, or cloud action. Evidence
is written under `artifacts/spec-tests/<UTC>-dedicated-gcp-regional-runtime`.
All checkpoints use explicit named `t.Run` calls and run sequentially.

After diagnosis, continue the same run by adding
`CP9_GCP_RUNTIME_RUN_DIR=/absolute/path/to/the/original/run` and
`CP9_GCP_RUNTIME_RETRY_STAGE=<failed-checkpoint>` to that command. Passing
checkpoints and completed mutations reuse their receipts; an interrupted action
with unknown outcome cannot be replayed. Missing prerequisites, filtered
checkpoints, original failures, and mixed harness revisions remain visible.
No prior shell run is imported as a Go journal.

For an explicitly authorized same-scene cleanup after failure, set
`CP9_GCP_RUNTIME_DESTROY=1` and the original `CP9_GCP_RUNTIME_RUN_DIR` with the
same opt-in and top-level Go selection. Archive failure preserves the scene.
Cleanup records accepted dispatch without waiting for cloud zero.

## Defaults and authority

- Use AWS account `804837308162` in `us-west-2` and the existing caller-owned
  EC2 key pair/private key inputs documented in this directory.
- Use GCP Project `pingcap-gardener`, region `us-west1`, zone `us-west1-a`, and
  the already-bootstrapped keyless `cp9-developer` service account.
- The instruction to execute this case authorizes one fresh 10-hour session,
  one real deploy of `gcp-regional-server`, one GCP owner-artifact refresh, one
  restart of the session-owned Google-to-AWS token refresher, one run-owned GCS
  ADC write/read/delete probe, one deliberate stop of only its GCP Tilt systemd
  unit, the same-session `cp9 up` repair, and one normal `cp9 destroy`. It does
  not authorize Product creation or deletion, old-session cleanup, or a
  destructive cleanup-workflow invocation.
- Freeze further test mutation at the first unexpected result. The Go cleanup hook
  records the failure and preserves the exact generated session for diagnosis;
  cleanup resumes only through an explicit continuation of that same run.
- Prerequisites are `aws`, `curl`, `gcloud`, `git`, `jq`, `make`, and `python3`,
  an active caller identity for both configured clouds, the existing EC2 key
  pair/private key inputs, and the already-bootstrapped shared GCP runtime
  identity. A missing prerequisite is `NOT RUN`, not Product failure.

## Execute

This is one ordered phase. The numbered sections are checkpoints inside that
phase, not independently runnable phases; the first unexpected red stops later
mutation and preserves the session. Cleanup runs only after every preceding
checkpoint passes, or through an explicit same-session continuation.

### 1. Create the dual runtime once

Freeze the checkout and record the dry-run owner, session, TTL, AWS account,
GCP projects, region, zone, keyless identity, and the exact six session-owned
GCP substrate resources, with no Project IAM or service-account creation.
Before any session state or cloud mutation, prove that GCP-disabled config
retains the primary-only shape; reject unsupported region/zone, invalid
project, and non-remote primary host; and reject a request to source-build the
immutable `gcp-regional-server` owner artifact. Every negative case must leave
no session directory. Then run setup, doctor, and `cp9 up` once. Status must
report the same session `ready`, the primary remote Tilt controller running,
and the GCP layer `runtime_ready` with a nonempty ready service graph.

### 2. Prove the day-2 surface across both runtimes

Require a persisted runtime plan containing exactly `primary` and
`gcp-regional`. Its GCP membership must equal the complete current 38-resource
live-active graph: the five namespaces; Dedicated API bootstrap and both retained
GCP/AWS CRD/provider paths; Crossplane plus all six Upbound GCP packages; Regional
Server, cluster-control/webhook, TCIDM, TiCDC, O11Y, and Kong services. It must
contain no legacy `gke-infra-provider`, live-binary, Global, Starter, Premium,
Essential, or BYOC resource.
A resource unique to GCP must route there without a selector. Deploy
`gcp-regional-server` once and require its GCP Tilt build identity to change
while the complete primary build-identity receipt remains unchanged. A resource in
both memberships must fail as ambiguous without `runtime=`, then resolve under
each explicit runtime. The routing probes other than the named GCP deploy use
`dry-run=true`.

Query logs for the GCP-only service and require its attempted sources to name
`gcp-regional`. Run one bounded trace-only query and require both `primary` and
`gcp-regional` to be queried and labeled even when no line matches. Refresh the
`gcp-regional-server` owner artifact with `runtime=gcp-regional`, require the
result to report that runtime and the persisted snapshot digest. Prove the
primary remote context is byte-identical; if the upstream digest did not move,
the GCP context must also remain byte-identical, otherwise only the GCP context
must change. Finally, observe all nine resource families directly through `gcloud`, with
explicit JSON output, successful command exits, and valid list/policy shapes.
Retain raw observations plus a summary with scanned/matched counts and no item
payloads. Require session-attributable addresses, boot disks, firewalls,
instances, networks, and subnets. GKE, service accounts, and IAM bindings may
match zero because this case creates no Product or shared identity. This is
Harness inventory evidence: the former proof-only `cp9 inventory gcp` command
was removed in PR #1278 and is not a Product acceptance surface.

### 3. Stop and resume only GCP

Record both controller PIDs as Harness evidence. Stop only the exact GCP
runtime's persisted systemd unit through its GCE IAP transport. Status must keep
the same session and the primary controller running while reporting the GCP
runtime unavailable. The same bounded trace-only log query must return
`partial`, with a completed primary attempt and a failed GCP attempt. Run the
same `cp9 up` command once. The session must return to `ready`; Harness evidence
must show a new positive GCP PID and an unchanged primary PID.

### 4. Prove the identity and network boundaries

Require the GCE SSH firewall to allow only Google IAP TCP/22, the runtime
service account to have no user-managed keys, the GCP ProviderConfig to use
`InjectedIdentity`, and the exact six Upbound GCP Providers to be Installed and
Healthy, with their current ProviderRevisions Healthy. Require the active classic GCP,
upstream GCP, and retained AWS provider deployments to be Ready, and require
the selected GCP Cluster/NodePool/ProviderConfig plus retained AWS
Route53/PCA/ProviderConfig CRDs to be Established. From the GCE host, use ADC
for one session-owned GCS object write/read/delete and require a subsequent
404 observation. Go owns the sequence and assertions; the small remote Python
transport returns only HTTP status and the test payload, never the token.
An unexpected result preserves the first red, including an object whose delete
has not yet run, for same-scene continuation or normal session cleanup.

Require the Regional HTTPS firewall to allow only the primary AWS public
address `/32`: a request from the primary host must complete TLS while the same
direct request from the runner must not connect. The runner resolves the
persisted Regional IP; set `CP9_RUNNER_NEGATIVE_INTERFACE` (for example, `en0`)
only for a bounded interface-specific proof, otherwise the default route is
used. Decode only the non-secret JWT
claims and require the configured service-account subject and audience, a
fresh bounded lifetime, and an active 45-minute refresh unit. Restart that unit
once and require an atomic newer token-file receipt. From that same GCE web-
identity principal, require AWS STS caller identity, the exact Route53 A record,
and certificate access to the exact session PCA. Fetch the live Regional leaf
certificate from the primary AWS host, require its fingerprint to match the
Terraform certificate, verify it against the PCA chain, and check its hostname.

### 5. Prove cleanup ownership

Before invoking cleanup, retain the session journal, runtime plan, both host
records, both Tilt status records, and the token-refresh mtime receipt outside
the session worktree. An early up/graph failure uses the existing host-state
layout without requiring successful graph locators. Record absent early artifacts
explicitly; required or previously observed files must remain readable. If this
archive fails, preserve the session and stop.
Run both GCP cleanup scripts in exact-session dry-run mode only. Then destroy
the one session and retain the destroy receipt and accepted hosted-cleanup
handoff; do not poll the workflow.

## Verdicts

### Product

`PASS` requires invalid configuration and GCP source-build requests to fail
before mutation; the dual runtime to become Ready; deploy and artifact refresh
commands to resolve the correct runtimes; the isolated GCP
failure to retain the healthy primary runtime; and the same session to recover.
A failed prerequisite is `NOT RUN`; a changed session, mutable GCP source
resource, misroute, lost primary runtime, or failed repair is `FAIL`.

### Harness

`PASS` requires the frozen head, scope, negative probes, runtime plan, exact
graph/provider/CRD receipts, status snapshots, bounded command results,
controller PIDs/context hashes, and cleanup dry-runs in the generated evidence
directory. PID stability is Harness evidence, not a Product contract. A bad
locator, parser, fault target, or incomplete capture is Harness `FAIL` and
cannot manufacture a Product verdict.

### Security

`PASS` requires IAP-only TCP/22 ingress, primary-AWS-/32-only Product ingress,
no user-managed key, injected GCP identity, bounded ADC object access with
cleanup, exact Google JWT subject/audience and live refresh, GCE-principal AWS
STS/Route53/PCA access, and a live hostname-valid Regional certificate chained
to that PCA. A broader firewall, key, static credential source, stale/wrong
token, or identity failure is Security `FAIL`.

### O11Y

`PASS` requires service-specific logs to query only `gcp-regional`, a trace
query to report both labeled runtimes, and the same trace query during the
controlled GCP outage to preserve the primary result while labeling the GCP
attempt failed. Missing or mislabeled sources are O11Y `FAIL`, not Product or
Harness failure.

### Cleanup

`PASS` requires the one normal destroy to succeed and record the accepted
hosted-cleanup handoff. Do not wait for cloud zero. Cleanup evidence is separate
from runtime evidence and cannot change an observed Product result.
*/

// TestFunctional_DedicatedGCPRegionalRuntime owns one ordered, single-agent
// AWS primary + GCP regional session. See the contract below for authority.
func TestFunctional_DedicatedGCPRegionalRuntime(t *testing.T) {
	if os.Getenv("CP9_GCP_RUNTIME_ENABLED") != "1" {
		t.Skip("NOT RUN: set CP9_GCP_RUNTIME_ENABLED=1 to authorize the GCP runtime journey")
	}
	if (os.Getenv("CP9_GCP_RUNTIME_DESTROY") == "1" || os.Getenv("CP9_GCP_RUNTIME_RETRY_STAGE") != "") && os.Getenv("CP9_GCP_RUNTIME_RUN_DIR") == "" {
		t.Fatal("cleanup or retry requires the original CP9_GCP_RUNTIME_RUN_DIR")
	}
	g := newGCPRuntimeRun(t)
	t.Cleanup(func() { g.summary(t) })
	if g.cleanupRequested() {
		t.Run("cleanup", func(t *testing.T) { g.cleanup(t) })
		return
	}
	t.Run("prerequisites", func(t *testing.T) { g.stage(t, "prerequisites", "prerequisite", g.prerequisites) })
	t.Run("prepare-and-dry-run", func(t *testing.T) { g.stage(t, "prepare-and-dry-run", "product", g.prepare) })
	t.Run("reject-region", func(t *testing.T) {
		g.stage(t, "reject-region", "product", func(t *testing.T) {
			t.Helper()
			g.reject(t, "region", "    region: us-west1", "    region: europe-west1", `dedicated.gcp.region "europe-west1" is unsupported`)
		})
	})
	t.Run("reject-zone", func(t *testing.T) {
		g.stage(t, "reject-zone", "product", func(t *testing.T) {
			t.Helper()
			g.reject(t, "zone", "    zone: us-west1-a", "    zone: us-east1-a", `dedicated.gcp.zone "us-east1-a" is unsupported`)
		})
	})
	t.Run("reject-project", func(t *testing.T) {
		g.stage(t, "reject-project", "product", func(t *testing.T) {
			t.Helper()
			g.reject(
				t,
				"project",
				"    seed_project_id: pingcap-gardener",
				"    seed_project_id: INVALID",
				`dedicated.gcp.seed_project_id "INVALID" must be a lowercase GCP project ID`,
			)
		})
	})
	t.Run("reject-primary-host", func(t *testing.T) { g.stage(t, "reject-primary-host", "product", g.rejectPrimary) })
	t.Run("disabled-primary-only", func(t *testing.T) { g.stage(t, "disabled-primary-only", "product", g.disabled) })
	t.Run("reject-source-build", func(t *testing.T) { g.stage(t, "reject-source-build", "product", g.rejectSource) })
	t.Run("dual-runtime-ready", func(t *testing.T) { g.stage(t, "dual-runtime-ready", "product", g.start) })
	t.Run("runtime-graph", func(t *testing.T) { g.stage(t, "runtime-graph", "harness", g.graph) })
	t.Run("day2-routing-and-deploy", func(t *testing.T) { g.stage(t, "day2-routing-and-deploy", "product", g.deploy) })
	t.Run("logs-both-runtimes", func(t *testing.T) { g.stage(t, "logs-both-runtimes", "o11y", g.logs) })
	t.Run("focused-owner-refresh", func(t *testing.T) { g.stage(t, "focused-owner-refresh", "product", g.refresh) })
	t.Run("direct-cloud-inventory", func(t *testing.T) { g.stage(t, "direct-cloud-inventory", "harness", g.inventory) })
	t.Run("stop-only-gcp", func(t *testing.T) { g.stage(t, "stop-only-gcp", "product", g.stop) })
	t.Run("logs-partial", func(t *testing.T) { g.stage(t, "logs-partial", "o11y", g.partialLogs) })
	t.Run("resume-same-session", func(t *testing.T) { g.stage(t, "resume-same-session", "product", g.resume) })
	t.Run("keyless-and-providers", func(t *testing.T) { g.stage(t, "keyless-and-providers", "security", g.providers) })
	t.Run("https-boundary-and-tls", func(t *testing.T) { g.stage(t, "https-boundary-and-tls", "security", g.tls) })
	t.Run("google-to-aws-identity", func(t *testing.T) { g.stage(t, "google-to-aws-identity", "security", g.identity) })
	t.Run("gcs-adc-object", func(t *testing.T) { g.stage(t, "gcs-adc-object", "security", g.adc) })
	t.Run("cleanup-dry-runs", func(t *testing.T) { g.stage(t, "cleanup-dry-runs", "harness", g.cleanupDryRuns) })
	t.Run("cleanup", func(t *testing.T) { g.stage(t, "cleanup", "cleanup", g.cleanup) })
}

var gcpRuntimeStages = []string{
	"prerequisites",
	"prepare-and-dry-run",
	"reject-region",
	"reject-zone",
	"reject-project",
	"reject-primary-host",
	"disabled-primary-only",
	"reject-source-build",
	"dual-runtime-ready",
	"runtime-graph",
	"day2-routing-and-deploy",
	"logs-both-runtimes",
	"focused-owner-refresh",
	"direct-cloud-inventory",
	"stop-only-gcp",
	"logs-partial",
	"resume-same-session",
	"keyless-and-providers",
	"https-boundary-and-tls",
	"google-to-aws-identity",
	"gcs-adc-object",
	"cleanup-dry-runs",
	"cleanup",
}

type gcpRuntimeRun struct {
	*ownerSessionRun

	currentStage *ownerSessionStageReceipt
}

func newGCPRuntimeRun(t *testing.T) *gcpRuntimeRun {
	t.Helper()
	wd, err := os.Getwd()
	if err != nil {
		t.Fatal(err)
	}
	repo := filepath.Dir(wd)
	root := os.Getenv("CP9_GCP_RUNTIME_RUN_DIR")
	if root == "" {
		root = filepath.Join(repo, "artifacts", "spec-tests", time.Now().UTC().Format("20060102T150405Z")+"-dedicated-gcp-regional-runtime")
		if err := os.MkdirAll(filepath.Dir(root), 0o700); err != nil {
			t.Fatal(err)
		}
		if err := os.Mkdir(root, 0o700); err != nil {
			t.Fatal(err)
		}
	} else {
		root, err = filepath.Abs(root)
		if err != nil {
			t.Fatal(err)
		}
		canonical, e := filepath.EvalSymlinks(root)
		if e != nil || canonical != root || !ownerSessionChildPath(filepath.Join(repo, "artifacts", "spec-tests"), root) {
			t.Fatal("run directory must be an existing canonical child of artifacts/spec-tests")
		}
	}
	r := &ownerSessionRun{
		t:             t,
		repoRoot:      repo,
		root:          root,
		caseName:      "dedicated-gcp-regional-runtime",
		environment:   "dev",
		currentHead:   currentCommit(t, repo),
		sessionDir:    filepath.Join(root, "session"),
		activeFailure: "harness",
		retryStage:    os.Getenv("CP9_GCP_RUNTIME_RETRY_STAGE"),
		retryStageEnv: "CP9_GCP_RUNTIME_RETRY_STAGE",
		destroyEnv:    "CP9_GCP_RUNTIME_DESTROY",
		stageNames:    gcpRuntimeStages,
	}
	r.commander = &resolverRun{
		t:          t,
		repoRoot:   repo,
		root:       root,
		commandID:  ownerSessionPriorCommandID(root),
		failure:    &r.activeFailure,
		commandEnv: map[string]string{},
	}
	r.journal = ownerSessionJournal{
		Version:     ownerSessionJournalV1,
		Case:        r.caseName,
		Environment: "dev",
		CP9Head:     r.currentHead,
		Cleanup:     "not_run",
		Stages:      map[string][]ownerSessionStageReceipt{},
		Actions:     map[string]ownerSessionActionReceipt{},
		CreatedAt:   time.Now().UTC().Format(time.RFC3339Nano),
	}
	p := filepath.Join(root, ownerSessionJournalFile)
	//nolint:gosec // G703: fixed receipt under the validated, user-selected run directory.
	if _, e := os.Stat(p); e == nil {
		r.readJSON(t, p, &r.journal)
		if r.journal.Version != ownerSessionJournalV1 || r.journal.Case != r.caseName || r.journal.Environment != "dev" || r.journal.CP9Head == "" ||
			r.journal.Stages == nil ||
			r.journal.Actions == nil {
			t.Fatal("incompatible GCP journal")
		}
	} else if !errors.Is(e, os.ErrNotExist) {
		t.Fatal(e)
	} else if os.Getenv("CP9_GCP_RUNTIME_RUN_DIR") != "" {
		t.Fatal("resume directory has no GCP journal")
	}
	g := &gcpRuntimeRun{ownerSessionRun: r}
	//nolint:gosec // G703: fixed session metadata under the validated run directory.
	if _, e := os.Stat(filepath.Join(r.sessionDir, "default-804-session.json")); e == nil {
		g.loadSession(t)
	} else if !errors.Is(e, os.ErrNotExist) {
		t.Fatal(e)
	}
	if err := r.persistJournal(); err != nil {
		t.Fatal(err)
	}
	t.Logf("GCP runtime evidence: %s", root)
	return g
}
func (g *gcpRuntimeRun) loadSession(t *testing.T) {
	t.Helper()
	var m ownerSessionMetadata
	g.readJSON(t, filepath.Join(g.sessionDir, "default-804-session.json"), &m)
	g.check(
		t,
		"harness",
		m.Account == "804837308162" && m.Region == "us-west-2" && m.Owner != "" && m.Session != "" && m.Config != "" && m.CP9Root != "" &&
			m.Head == g.journal.CP9Head,
		"session metadata identity",
	)
	g.check(
		t,
		"harness",
		(g.journal.Session == "" || g.journal.Session == m.Session) && regexp.MustCompile(`^[a-z0-9][a-z0-9-]{2,31}$`).MatchString(m.Session),
		"preserved session identity",
	)
	g.cp9Root, g.cp9Config, g.sessionName = m.CP9Root, m.Config, m.Session
	g.journal.Session = m.Session
	g.stateDir = filepath.Join(m.CP9Root, ".cp9", m.Session)
	if g.journal.Cleanup != "pass" && !g.cleanupRequested() {
		g.check(t, "harness", g.gitOutput(t, "frozen-session-head", g.cp9Root, "rev-parse", "HEAD") == g.journal.CP9Head, "frozen session head")
	}
	g.commander.commandEnv["CP9_REPO_ROOT"] = g.cp9Root
}
func (g *gcpRuntimeRun) stage(t *testing.T, name, lane string, body func(*testing.T)) {
	t.Helper()
	if g.t.Failed() {
		t.Skip("NOT RUN: an earlier checkpoint or receipt write failed")
	}
	i := slices.Index(gcpRuntimeStages, name)
	if i > 0 && !g.stagesPassed(gcpRuntimeStages[:i]...) {
		t.Skip("NOT RUN: a preceding checkpoint did not pass")
	}
	receipts := g.journal.Stages[name]
	if len(receipts) > 0 {
		previous := receipts[len(receipts)-1]
		if previous.Status == "passed" {
			t.Skipf("reuse passing %s receipt from %s", name, previous.Head)
		}
		if previous.Status != "not_run" && g.retryStage != name {
			t.Fatalf("checkpoint %s is %s; diagnose the preserved scene before setting CP9_GCP_RUNTIME_RETRY_STAGE=%s", name, previous.Status, name)
		}
	}
	g.activeFailure = lane
	g.journal.Stages[name] = append(
		receipts,
		ownerSessionStageReceipt{
			Head:      g.currentHead,
			StartedAt: time.Now().UTC().Format(time.RFC3339Nano),
			Status:    "running",
			Product:   "not_run",
			Harness:   "not_run",
			Security:  "not_run",
			O11Y:      "not_run",
		},
	)
	stage := &g.journal.Stages[name][len(receipts)]
	g.currentStage = stage
	t.Cleanup(func() {
		defer func() { g.currentStage = nil }()
		stage.FinishedAt = time.Now().UTC().Format(time.RFC3339Nano)
		stage.Status, stage.Harness = "passed", "pass"
		verdict := "pass"
		if t.Failed() {
			stage.Status = "failed"
			verdict = "fail"
			lane = g.activeFailure
		} else if t.Skipped() {
			stage.Status, stage.Harness = "not_run", "not_run"
			verdict = "not_run"
		}
		gcpSetStageVerdict(stage, lane, verdict)
		if err := g.persistJournal(); err != nil {
			stage.Status, stage.Harness = "failed", "fail"
			retryErr := g.persistJournal()
			t.Errorf("finish GCP checkpoint %s: %v", name, errors.Join(err, retryErr))
		}
	})
	if err := g.persistJournal(); err != nil {
		g.activeFailure = "harness"
		t.Fatal(err)
	}
	body(t)
}

// Complete a lane before later, independent evidence work can fail.
func (g *gcpRuntimeRun) complete(t *testing.T, lane string) {
	t.Helper()
	if g.currentStage == nil {
		t.Fatal("lane completion requires an active checkpoint")
	}
	gcpSetStageVerdict(g.currentStage, lane, "pass")
	if err := g.persistJournal(); err != nil {
		g.activeFailure = "harness"
		t.Fatal(err)
	}
}
func gcpSetStageVerdict(r *ownerSessionStageReceipt, lane, verdict string) {
	switch lane {
	case "product":
		r.Product = verdict
	case "harness":
		r.Harness = verdict
	case "security":
		r.Security = verdict
	case "o11y":
		r.O11Y = verdict
	}
}
func (g *gcpRuntimeRun) recordHarnessFailure(name, message string) error {
	now := time.Now().UTC().Format(time.RFC3339Nano)
	g.journal.Stages[name] = append(g.journal.Stages[name], ownerSessionStageReceipt{
		Head: g.currentHead, StartedAt: now, FinishedAt: now, Status: "failed",
		Product: "not_run", Harness: "fail", Security: "not_run", O11Y: "not_run", Message: message,
	})
	return g.persistJournal()
}
func (g *gcpRuntimeRun) check(t *testing.T, lane string, ok bool, message string) {
	t.Helper()
	g.activeFailure = lane
	if !ok {
		t.Fatal(message)
	}
}
func (g *gcpRuntimeRun) save(t *testing.T, name string, value any) {
	t.Helper()
	b, e := json.MarshalIndent(value, "", "  ")
	if e != nil {
		g.activeFailure = "harness"
		t.Fatal(e)
	}
	g.writeText(t, name, string(b))
}
func gcpStageVerdict(r *ownerSessionStageReceipt, lane string) string {
	switch lane {
	case "product":
		return r.Product
	case "harness":
		return r.Harness
	case "security":
		return r.Security
	case "o11y":
		return r.O11Y
	default:
		return "not_run"
	}
}
func (g *gcpRuntimeRun) verdict(lane string, names ...string) string {
	complete := true
	for _, name := range names {
		receipts := g.journal.Stages[name]
		if len(receipts) == 0 || gcpStageVerdict(&receipts[len(receipts)-1], lane) != "pass" {
			complete = false
		}
	}
	for _, receipts := range g.journal.Stages {
		for i := range receipts {
			if gcpStageVerdict(&receipts[i], lane) == "fail" {
				return "fail"
			}
		}
	}
	if complete {
		return "pass"
	}
	return "not_run"
}
func (g *gcpRuntimeRun) summary(t *testing.T) {
	t.Helper()
	verdict := map[string]string{
		"product": g.verdict(
			"product",
			"prepare-and-dry-run",
			"reject-region",
			"reject-zone",
			"reject-project",
			"reject-primary-host",
			"disabled-primary-only",
			"reject-source-build",
			"dual-runtime-ready",
			"day2-routing-and-deploy",
			"focused-owner-refresh",
			"stop-only-gcp",
			"resume-same-session",
		),
		"harness":  g.verdict("harness", gcpRuntimeStages[:len(gcpRuntimeStages)-1]...),
		"security": g.verdict("security", "keyless-and-providers", "https-boundary-and-tls", "google-to-aws-identity", "gcs-adc-object"),
		"o11y":     g.verdict("o11y", "logs-both-runtimes", "logs-partial"),
		"cleanup":  g.journal.Cleanup,
	}
	publish := func() error {
		var b strings.Builder
		b.WriteString(
			"# Dedicated GCP regional runtime\n\n| Checkpoint | Product | Harness | Security | O11Y | Status | Head |\n| --- | --- | --- | --- | --- | --- | --- |\n",
		)
		names := slices.Clone(gcpRuntimeStages)
		for _, name := range []string{"summary", "cleanup-evidence"} {
			if len(g.journal.Stages[name]) > 0 {
				names = append(names, name)
			}
		}
		for _, name := range names {
			receipts := g.journal.Stages[name]
			if len(receipts) == 0 {
				fmt.Fprintf(&b, "| %s | not_run | not_run | not_run | not_run | not_run | — |\n", name)
			}
			for i := range receipts {
				r := &receipts[i]
				fmt.Fprintf(&b, "| %s | %s | %s | %s | %s | %s | %s |\n", name, r.Product, r.Harness, r.Security, r.O11Y, r.Status, r.Head)
			}
		}
		fmt.Fprintf(
			&b,
			"\nProduct head: %s; harness head: %s; session: %s.\n\nVerdicts: %v\n",
			g.journal.CP9Head,
			g.currentHead,
			g.sessionName,
			verdict,
		)
		data, err := json.MarshalIndent(verdict, "", "  ")
		if err != nil {
			return err
		}
		// Attempt both destinations so one unavailable file does not hide the result.
		return errors.Join(os.WriteFile(filepath.Join(g.root, "summary.json"), append(data, '\n'), 0o600),
			os.WriteFile(filepath.Join(g.root, "summary.md"), []byte(b.String()), 0o600))
	}
	if err := publish(); err != nil {
		journalErr := g.recordHarnessFailure("summary", err.Error())
		verdict["harness"] = "fail"
		t.Errorf("publish final evidence: %v", errors.Join(err, journalErr, publish()))
	}
}

func (g *gcpRuntimeRun) command(t *testing.T, label, lane string, action bool, argv ...string) resolverCommandResult {
	t.Helper()
	g.activeFailure = lane
	dir := g.cp9Root
	if dir == "" {
		dir = g.repoRoot
	}
	if action {
		return g.mustActionDir(t, label, ownerSessionUpTimeout, dir, argv...)
	}
	return g.mustDir(t, label, ownerSessionCommandTimeout, dir, argv...)
}
func (g *gcpRuntimeRun) cp9(t *testing.T, label, lane string, action bool, args ...string) resolverCommandResult {
	t.Helper()
	argv := append([]string{filepath.Join(g.cp9Root, "bin", "cp9")}, args...)
	argv = append(argv, "config="+g.cp9Config)
	return g.command(t, label, lane, action, argv...)
}
func (g *gcpRuntimeRun) decode(t *testing.T, label string, result resolverCommandResult, out any) {
	t.Helper()
	g.writeText(t, label+".json", result.Stdout)
	if err := json.Unmarshal([]byte(result.Stdout), out); err != nil {
		g.activeFailure = "harness"
		t.Fatalf("parse %s: %v", label, err)
	}
}
func gcpShell(value string) string { return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'" }
func (g *gcpRuntimeRun) primary(args ...string) []string {
	return append([]string{filepath.Join(g.cp9Root, "skills", "get-started", "scripts", "session-ssh.sh"), g.sessionName}, args...)
}

type gcpDryRun struct {
	NoMutation bool   `json:"no_mutation"`
	Session    string `json:"session"`
	Owner      string `json:"owner"`
	TTL        string `json:"ttl"`
	AWS        struct {
		Account string `json:"account"`
		Region  string `json:"region"`
	} `json:"aws"`
	GCP *struct {
		Enabled   bool     `json:"enabled"`
		Apply     bool     `json:"apply"`
		Seed      string   `json:"seed_project_id"`
		Dataplane string   `json:"dataplane_project_id"`
		Region    string   `json:"region"`
		Zone      string   `json:"zone"`
		Identity  string   `json:"runtime_identity"`
		Resources []string `json:"resources"`
	} `json:"dedicated_gcp"`
}

func (g *gcpRuntimeRun) prerequisites(t *testing.T) {
	t.Helper()
	g.activeFailure = "prerequisite"
	for _, name := range []string{"aws", "curl", "gcloud", "git", "jq", "make", "python3"} {
		if _, e := exec.LookPath(name); e != nil {
			t.Skipf("NOT RUN: missing %s", name)
		}
	}
	if os.Getenv("CP9_UI_API_AWS_KEY_PAIR") == "" || os.Getenv("CP9_UI_API_SSH_PRIVATE_KEY") == "" {
		t.Skip("NOT RUN: existing CP9_UI_API_AWS_KEY_PAIR and CP9_UI_API_SSH_PRIVATE_KEY required")
	}
	//nolint:gosec // G703: the caller explicitly supplies this existing read-only key path.
	f, e := os.Open(os.Getenv("CP9_UI_API_SSH_PRIVATE_KEY"))
	if e != nil {
		t.Skip("NOT RUN: caller-owned SSH private key unreadable")
	}
	if e = f.Close(); e != nil {
		t.Fatal(e)
	}
	aws := g.run(t, "prerequisite-aws", time.Minute, "aws", "sts", "get-caller-identity", "--profile", "dev-offline-afterworld", "--output", "json")
	var caller struct{ Account string }
	if aws.ExitCode != 0 || json.Unmarshal([]byte(aws.Stdout), &caller) != nil || caller.Account != "804837308162" {
		t.Skip("NOT RUN: AWS profile must resolve to account 804837308162")
	}
	auth := g.run(t, "prerequisite-gcp-auth", time.Minute, "gcloud", "auth", "list", "--filter=status:ACTIVE", "--format=json")
	var accounts []struct {
		Account string `json:"account"`
	}
	if auth.ExitCode != 0 || json.Unmarshal([]byte(auth.Stdout), &accounts) != nil || len(accounts) == 0 {
		t.Skip("NOT RUN: active GCP login required")
	}
	project := g.run(t, "prerequisite-gcp-project", time.Minute, "gcloud", "projects", "describe", "pingcap-gardener", "--format=json")
	var p struct {
		ProjectID string `json:"projectId"`
	}
	if project.ExitCode != 0 || json.Unmarshal([]byte(project.Stdout), &p) != nil || p.ProjectID != "pingcap-gardener" {
		t.Skip("NOT RUN: pingcap-gardener project unavailable")
	}
	sa := g.run(
		t,
		"prerequisite-gcp-service-account",
		time.Minute,
		"gcloud",
		"iam",
		"service-accounts",
		"describe",
		"cp9-developer@pingcap-gardener.iam.gserviceaccount.com",
		"--project=pingcap-gardener",
		"--format=json",
	)
	var identity struct {
		Email string `json:"email"`
	}
	if sa.ExitCode != 0 || json.Unmarshal([]byte(sa.Stdout), &identity) != nil ||
		identity.Email != "cp9-developer@pingcap-gardener.iam.gserviceaccount.com" {
		t.Skip("NOT RUN: shared keyless GCP service account unavailable")
	}
}
func (g *gcpRuntimeRun) prepare(t *testing.T) {
	t.Helper()
	g.mustActionDir(
		t,
		"prepare-session",
		ownerSessionCommandTimeout,
		g.repoRoot,
		filepath.Join(g.repoRoot, "spec-tests", "scripts", "default-804-session.sh"),
		"prepare",
		"gcp-regional",
		g.sessionDir,
	)
	g.loadSession(t)
	g.command(t, "build", "harness", false, "make", "build")
	var d gcpDryRun
	g.decode(t, "dry-run", g.cp9(t, "dry-run", "product", false, "{sys.output.format=json}", "up", "dry-run=true"), &d)
	g.check(
		t,
		"product",
		d.NoMutation && d.Session == g.sessionName && d.Owner != "" && d.TTL != "" && d.AWS.Account == "804837308162" && d.AWS.Region == "us-west-2" &&
			d.GCP != nil,
		"dry-run owner/session/cloud scope",
	)
	c := d.GCP
	g.check(
		t,
		"product",
		c.Enabled && !c.Apply && c.Seed == "pingcap-gardener" && c.Dataplane == "pingcap-gardener" && c.Region == "us-west1" && c.Zone == "us-west1-a" &&
			c.Identity == "shared Project-bootstrap cp9-developer service account via ADC; no JSON key",
		"GCP dry-run scope and keyless identity",
	)
	g.check(
		t,
		"product",
		gcpSameNames(
			c.Resources,
			[]string{
				"custom VPC",
				"regional subnet",
				"static public address",
				"IAP-only SSH firewall",
				"AWS-host-/32 Kong firewall",
				"one GCE k3d runtime",
			},
		),
		"exact six session-owned resources; no project IAM or service-account creation",
	)
	g.complete(t, "product")
	g.save(t, "scope.json", d)
}
func gcpSameNames(got, want []string) bool {
	got = slices.Clone(got)
	want = slices.Clone(want)
	slices.Sort(got)
	slices.Sort(want)
	return slices.Equal(got, want)
}
func (g *gcpRuntimeRun) configText(t *testing.T) string {
	t.Helper()
	b, e := os.ReadFile(g.cp9Config)
	if e != nil {
		g.activeFailure = "harness"
		t.Fatal(e)
	}
	return string(b)
}
func (g *gcpRuntimeRun) replace(t *testing.T, text, before, after string) string {
	t.Helper()
	g.check(t, "harness", strings.Count(text, before) == 1, "fixture replacement must match exactly once: "+before)
	return strings.Replace(text, before, after, 1)
}
func (g *gcpRuntimeRun) noState(t *testing.T) {
	t.Helper()
	_, e := os.Lstat(g.stateDir)
	if e != nil && !errors.Is(e, os.ErrNotExist) {
		g.activeFailure = "harness"
		t.Fatalf("observe pre-mutation session path: %v", e)
	}
	g.check(t, "product", errors.Is(e, os.ErrNotExist), "pre-mutation probe created session state or state is unobservable")
}
func (g *gcpRuntimeRun) rejected(t *testing.T, name, text, expected string) {
	t.Helper()
	path := filepath.Join(g.root, "invalid-"+name+".yaml")
	g.writeBytesAt(t, path, []byte(text))
	g.activeFailure = "product"
	r := g.runDir(
		t,
		"invalid-"+name,
		ownerSessionCommandTimeout,
		g.cp9Root,
		filepath.Join(g.cp9Root, "bin", "cp9"),
		"up",
		"dry-run=true",
		"config="+path,
	)
	g.check(t, "product", r.ExitCode != 0, "invalid "+name+" unexpectedly accepted")
	g.check(t, "product", strings.Contains(r.Stdout+"\n"+r.Stderr, expected), "invalid "+name+" did not report its target error")
	g.noState(t)
}
func (g *gcpRuntimeRun) reject(t *testing.T, name, before, after, expected string) {
	t.Helper()
	g.rejected(t, name, g.replace(t, g.configText(t), before, after), expected)
}
func (g *gcpRuntimeRun) rejectPrimary(t *testing.T) {
	t.Helper()
	text := g.replace(t, g.configText(t), "  mode: remote-ec2-k3d", "  mode: local-k3d")
	text = g.replace(t, text, "  exposure:\n    enabled: true", "  exposure:\n    enabled: false")
	g.rejected(t, "primary-host", text, "dedicated.gcp requires the existing AWS primary host.mode remote-ec2-k3d")
}
func (g *gcpRuntimeRun) disabled(t *testing.T) {
	t.Helper()
	text := g.configText(t)
	start, end := strings.Index(text, "\ndedicated:\n"), strings.Index(text, "\nhost:\n")
	g.check(t, "harness", start >= 0 && end > start, "dedicated fixture block")
	path := filepath.Join(g.root, "gcp-disabled.yaml")
	g.writeBytesAt(t, path, []byte(text[:start]+text[end:]))
	var d gcpDryRun
	g.decode(
		t,
		"gcp-disabled",
		g.command(
			t,
			"gcp-disabled",
			"product",
			false,
			filepath.Join(g.cp9Root, "bin", "cp9"),
			"{sys.output.format=json}",
			"up",
			"dry-run=true",
			"config="+path,
		),
		&d,
	)
	g.check(t, "product", d.NoMutation && d.GCP == nil, "disabled GCP must preserve primary-only dry-run")
	g.noState(t)
}
func (g *gcpRuntimeRun) rejectSource(t *testing.T) {
	t.Helper()
	text := g.configText(t)
	var tilt string
	for line := range strings.SplitSeq(text, "\n") {
		if file, ok := strings.CutPrefix(line, "  file: "); ok {
			tilt = file
		}
	}
	g.check(t, "harness", tilt != "" && ownerSessionChildPath(g.cp9Root, tilt), "source Tiltfile locator")
	b, e := os.ReadFile(tilt)
	if e != nil {
		t.Fatal(e)
	}
	re := regexp.MustCompile(`(?s)source_build_services = \[.*?\n\]`)
	g.check(t, "harness", len(re.FindAll(b, -1)) == 1, "source-build fixture selection")
	invalid := filepath.Join(g.cp9Root, "personas", "dev-full", "runtime", ".cp9-spec-"+g.sessionName+"-invalid-source.Tiltfile")
	g.writeBytesAt(t, invalid, re.ReplaceAll(b, []byte("source_build_services = [\n    \"gcp-regional-server\",\n]")))
	g.copyFile(t, invalid, filepath.Join(g.root, "invalid-gcp-source-build.Tiltfile"))
	// The fixture is local and run-owned; keep it on failure with the same scene.
	g.rejected(
		t,
		"gcp-source-build",
		g.replace(t, text, "  file: "+tilt, "  file: "+invalid),
		"source_build_services selected unknown service gcp-regional-server",
	)
	g.complete(t, "product")
	if e := os.Remove(invalid); e != nil {
		g.activeFailure = "harness"
		t.Fatal(e)
	}
}

type gcpTiltStatus struct {
	APIPort int    `json:"api_port"`
	Unit    string `json:"remote_unit"`
	Context string `json:"remote_context_index_path"`
}
type gcpHost struct {
	Remote struct {
		Instance   string `json:"instance_name"`
		IP         string `json:"public_ip"`
		Kubeconfig string `json:"remote_kubeconfig_path"`
		StateDir   string `json:"remote_state_dir"`
	} `json:"remote"`
	GCP struct {
		Project        string `json:"project_id"`
		Zone           string `json:"zone"`
		ServiceAccount string `json:"service_account_email"`
		UniqueID       string `json:"service_account_unique_id"`
		Role           string `json:"aws_role_arn"`
		URL            string `json:"regional_endpoint_url"`
		Host           string `json:"regional_endpoint_host"`
		Bucket         string `json:"o11y_bucket"`
	} `json:"gcp"`
	DedicatedAPI struct {
		Kubeconfig string `json:"remote_admin_kubeconfig_path"`
	} `json:"dedicated_api"`
}
type gcpGraph struct {
	APIVersion string `json:"api_version"`
	Session    string `json:"session_name"`
	Runtimes   []struct {
		ID        string   `json:"id"`
		StateDir  string   `json:"state_dir"`
		Resources []string `json:"resources"`
	} `json:"runtimes"`
}
type gcpLocators struct {
	Primary   gcpTiltStatus `json:"primary"`
	Regional  gcpTiltStatus `json:"regional"`
	Host      gcpHost       `json:"host"`
	StateDir  string        `json:"state_dir"`
	Ambiguous string        `json:"ambiguous"`
}

func (g *gcpRuntimeRun) locators(t *testing.T) gcpLocators {
	t.Helper()
	var l gcpLocators
	g.readJSON(t, filepath.Join(g.root, "locators.json"), &l)
	g.check(
		t,
		"harness",
		l.Host.Remote.Instance != "" && l.Host.GCP.Project == "pingcap-gardener" && l.Host.GCP.Zone == "us-west1-a" &&
			ownerSessionChildPath(g.stateDir, l.StateDir),
		"persisted GCP transport scope",
	)
	return l
}
func (g *gcpRuntimeRun) remote(t *testing.T, args ...string) []string {
	t.Helper()
	l := g.locators(t)
	quoted := make([]string, len(args))
	for i, arg := range args {
		quoted[i] = gcpShell(arg)
	}
	return []string{
		"gcloud",
		"compute",
		"ssh",
		l.Host.Remote.Instance,
		"--project=" + l.Host.GCP.Project,
		"--zone=" + l.Host.GCP.Zone,
		"--tunnel-through-iap",
		"--quiet",
		"--command=" + strings.Join(quoted, " "),
	}
}
func (g *gcpRuntimeRun) status(t *testing.T, label, want, gcpWant string) {
	t.Helper()
	var s struct {
		Session string `json:"session_name"`
		Status  string `json:"status"`
		Layers  struct {
			Tilt struct {
				Controller string `json:"controller"`
				Running    bool   `json:"running"`
			} `json:"tilt"`
			GCP struct {
				Status string `json:"status"`
				Tilt   struct {
					Controller string `json:"controller"`
					Running    bool   `json:"running"`
					Total      int    `json:"total_resources"`
					Ready      int    `json:"ready_resources"`
				} `json:"tilt"`
			} `json:"dedicated_gcp"`
		} `json:"layers"`
	}
	// Partial status can return nonzero; parse its public status contract separately.
	r := g.runDir(
		t,
		label,
		ownerSessionCommandTimeout,
		g.cp9Root,
		filepath.Join(g.cp9Root, "bin", "cp9"),
		"{sys.output.format=json}",
		"status",
		"config="+g.cp9Config,
	)
	g.decode(t, label, r, &s)
	g.check(t, "product", want == "partial" || r.ExitCode == 0, "Ready status command failed")
	g.check(
		t,
		"product",
		s.Session == g.sessionName && s.Status == want && s.Layers.Tilt.Controller == "remote" && s.Layers.Tilt.Running &&
			s.Layers.GCP.Status == gcpWant,
		"same-session dual runtime status",
	)
	if want == "ready" {
		g.check(
			t,
			"product",
			s.Layers.GCP.Tilt.Controller == "remote" && s.Layers.GCP.Tilt.Running && s.Layers.GCP.Tilt.Total > 0 &&
				s.Layers.GCP.Tilt.Ready == s.Layers.GCP.Tilt.Total,
			"GCP remote controller and complete ready graph",
		)
	}
}
func (g *gcpRuntimeRun) start(t *testing.T) {
	t.Helper()
	g.cp9(t, "setup", "harness", true, "setup", "checkout=false")
	g.cp9(t, "doctor", "harness", false, "doctor")
	g.cp9(t, "up", "product", true, "up")
	g.status(t, "status-ready-before", "ready", "runtime_ready")
}
func (g *gcpRuntimeRun) graph(t *testing.T) {
	t.Helper()
	var plan gcpGraph
	g.readJSON(t, filepath.Join(g.stateDir, "runtime-plan.json"), &plan)
	g.check(
		t,
		"harness",
		plan.APIVersion == "cp9.tidbcloud.com/runtime-plan/v1" && plan.Session == g.sessionName && len(plan.Runtimes) == 2,
		"persisted two-runtime plan identity",
	)
	var primary, regional []string
	var l gcpLocators
	for _, r := range plan.Runtimes {
		switch r.ID {
		case "primary":
			primary = r.Resources
		case "gcp-regional":
			regional = r.Resources
			l.StateDir = filepath.Join(g.stateDir, r.StateDir)
		}
	}
	g.check(t, "harness", len(primary) > 0 && ownerSessionChildPath(g.stateDir, l.StateDir), "runtime memberships and regional state directory")
	g.check(
		t,
		"harness",
		gcpSameNames(
			regional,
			[]string{
				"(Tiltfile)",
				"ns-cp9-system",
				"ns-cp9-dedicated",
				"ns-crossplane-system",
				"ns-kong-system",
				"ns-observability",
				"dedicated-api-host-ready",
				"dedicated-crossplane-gcp-crds",
				"dedicated-crossplane-aws-crds",
				"dedicated-infra-api-crds",
				"gcp-classic-provider",
				"gcp-upstream-provider",
				"dedicated-aws-provider",
				"gcp-crossplane",
				"gcp-crossplane-ready",
				"gcp-upbound-provider-runtime",
				"gcp-upbound-provider-family",
				"gcp-upbound-provider-compute",
				"gcp-upbound-provider-container",
				"gcp-upbound-provider-dns",
				"gcp-upbound-provider-iam",
				"gcp-upbound-provider-storage",
				"gcp-upbound-provider-config",
				"gcp-gateway-api-crds",
				"gcp-o11y-kong-plugin",
				"gcp-regional-kong-controller",
				"gcp-o11y-runtime-secret",
				"gcp-o11y-platform-crds",
				"gcp-o11y-platform-image-pull-secret",
				"gcp-o11y-platform",
				"gcp-o11y-agent-control-server",
				"gcp-webhook-server",
				"gcp-cluster-control-server",
				"gcp-regional-server",
				"gcp-gke-provider",
				"gcp-regional-tcidm-service",
				"gcp-dedicated-ticdc-service",
				"gcp-regional-kong",
			},
		),
		"exact active GCP graph membership",
	)
	g.check(
		t,
		"harness",
		!slices.Contains(primary, "gcp-regional-server") && slices.Contains(regional, "gcp-regional-server"),
		"unique GCP service membership",
	)
	for _, name := range regional {
		if slices.Contains(primary, name) {
			l.Ambiguous = name
			break
		}
	}
	g.check(t, "harness", l.Ambiguous != "", "shared runtime resource")
	g.readJSON(t, filepath.Join(g.stateDir, "tilt", "reports", "tilt-status.json"), &l.Primary)
	g.readJSON(t, filepath.Join(l.StateDir, "tilt", "reports", "tilt-status.json"), &l.Regional)
	g.readJSON(t, filepath.Join(l.StateDir, "host.json"), &l.Host)
	unit := regexp.MustCompile(`^cp9-tilt-[A-Za-z0-9_-]+\.service$`)
	g.check(
		t,
		"harness",
		unit.MatchString(l.Primary.Unit) && unit.MatchString(l.Regional.Unit) && l.Primary.APIPort > 0 && l.Regional.APIPort > 0 && l.Primary.Context != "" &&
			l.Regional.Context != "",
		"remote controller locators",
	)
	g.save(t, "locators.json", l)
	g.locators(t)
	g.copyFile(t, filepath.Join(g.stateDir, "runtime-plan.json"), filepath.Join(g.root, "runtime-plan.json"))
}

func (g *gcpRuntimeRun) before(t *testing.T, label, action string, argv ...string) string {
	t.Helper()
	path := filepath.Join(g.root, label+".txt")
	_, started := g.journal.Actions[action]
	if started {
		b, e := os.ReadFile(path)
		g.check(t, "harness", e == nil, "missing pre-action evidence for "+action+"; refusing to manufacture it after mutation")
		return string(b)
	}
	r := g.command(t, label, "harness", false, argv...)
	g.writeText(t, label+".txt", r.Stdout)
	return r.Stdout
}

type gcpBuildResource struct {
	Metadata struct {
		Name string `json:"name"`
	} `json:"metadata"`
	Status struct {
		BuildHistory []struct {
			Span string `json:"spanID"`
		} `json:"buildHistory"`
	} `json:"status"`
}

func (g *gcpRuntimeRun) buildIdentity(t *testing.T, text string, primary bool) string {
	t.Helper()
	if !primary {
		var r gcpBuildResource
		g.check(
			t,
			"harness",
			json.Unmarshal([]byte(text), &r) == nil && len(r.Status.BuildHistory) > 0 && r.Status.BuildHistory[0].Span != "",
			"GCP build identity capture",
		)
		return r.Status.BuildHistory[0].Span
	}
	var list struct {
		Items []gcpBuildResource `json:"items"`
	}
	g.check(t, "harness", json.Unmarshal([]byte(text), &list) == nil && len(list.Items) > 0, "primary build identity capture")
	// Preserve the complete ordered UI-resource identity projection from the old case.
	type build struct {
		Name  string `json:"name"`
		Build string `json:"build"`
	}
	values := make([]build, 0, len(list.Items))
	for _, item := range list.Items {
		g.check(t, "harness", item.Metadata.Name != "", "primary resource name")
		span := ""
		if len(item.Status.BuildHistory) > 0 {
			span = item.Status.BuildHistory[0].Span
		}
		values = append(values, build{Name: item.Metadata.Name, Build: span})
	}
	b, e := json.Marshal(values)
	if e != nil {
		t.Fatal(e)
	}
	return fmt.Sprintf("%x", sha256.Sum256(b))
}
func (g *gcpRuntimeRun) route(t *testing.T, label, service, runtime string) {
	t.Helper()
	args := []string{"{sys.output.format=json}", "deploy", "service=" + service, "dry-run=true"}
	if runtime != "" {
		args = append(args, "runtime="+runtime)
	}
	var r struct {
		Runtime string `json:"runtime"`
		DryRun  bool   `json:"dry_run"`
	}
	g.decode(t, label, g.cp9(t, label, "product", false, args...), &r)
	if runtime == "" {
		runtime = "gcp-regional"
	}
	g.check(t, "product", r.Runtime == runtime && r.DryRun, "explicit or unique runtime route: "+label)
}
func (g *gcpRuntimeRun) deploy(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	g.route(t, "route-unique-gcp", "gcp-regional-server", "")
	ambiguous := g.runDir(
		t,
		"route-ambiguous",
		ownerSessionCommandTimeout,
		g.cp9Root,
		filepath.Join(g.cp9Root, "bin", "cp9"),
		"deploy",
		"service="+l.Ambiguous,
		"dry-run=true",
		"config="+g.cp9Config,
	)
	g.check(
		t,
		"product",
		ambiguous.ExitCode != 0 && strings.Contains(ambiguous.Stdout+"\n"+ambiguous.Stderr, "exists in runtimes"),
		"ambiguous resource must require runtime selector",
	)
	g.route(t, "route-explicit-primary", l.Ambiguous, "primary")
	g.route(t, "route-explicit-gcp", l.Ambiguous, "gcp-regional")
	primary := g.primary("tilt", "get", "uiresources", "-o", "json", "--host", "127.0.0.1", "--port", strconv.Itoa(l.Primary.APIPort))
	regional := g.remote(
		t,
		"tilt",
		"get",
		"uiresource",
		"gcp-regional-server",
		"-o",
		"json",
		"--host",
		"127.0.0.1",
		"--port",
		strconv.Itoa(l.Regional.APIPort),
	)
	pb := g.buildIdentity(t, g.before(t, "primary-build-before", "deploy-unique-gcp", primary...), true)
	gb := g.buildIdentity(t, g.before(t, "gcp-build-before", "deploy-unique-gcp", regional...), false)
	g.cp9(t, "deploy-unique-gcp", "product", true, "deploy", "service=gcp-regional-server")
	g.complete(t, "product")
	ga := g.buildIdentity(t, g.command(t, "gcp-build-after", "harness", false, regional...).Stdout, false)
	pa := g.buildIdentity(t, g.command(t, "primary-build-after", "harness", false, primary...).Stdout, true)
	g.save(t, "build-identity-receipt.json", map[string]string{"primary_before": pb, "primary_after": pa, "gcp_before": gb, "gcp_after": ga})
	g.check(t, "harness", gb != ga && pb == pa, "GCP deploy must change only its build identity")
}

type gcpLogs struct {
	Service  string   `json:"service"`
	TraceID  string   `json:"trace_id"`
	Limit    int      `json:"limit"`
	Status   string   `json:"status"`
	Sources  []string `json:"sources"`
	Attempts []struct {
		Substrate string `json:"substrate"`
		Status    string `json:"status"`
	} `json:"attempted_sources"`
}

func (g *gcpRuntimeRun) trace(t *testing.T, label string) gcpLogs {
	t.Helper()
	var l gcpLogs
	g.decode(
		t,
		label,
		g.cp9(t, label, "o11y", false, "{sys.output.format=json}", "logs", "trace-id=cp9-spec-"+g.sessionName+"-no-match", "since=10m", "limit=25"),
		&l,
	)
	g.check(t, "o11y", l.TraceID == "cp9-spec-"+g.sessionName+"-no-match" && l.Limit == 25, "bounded trace identity")
	return l
}
func (g *gcpRuntimeRun) attempts(t *testing.T, l *gcpLogs, gcpStatus string) {
	t.Helper()
	primary, regional := 0, 0
	for _, a := range l.Attempts {
		if a.Substrate == "primary" && a.Status == "queried" {
			primary++
		}
		if a.Substrate == "gcp-regional" && a.Status == gcpStatus {
			regional++
		}
	}
	g.check(t, "o11y", primary == 1 && regional == 1 && len(l.Attempts) == 2, "trace query must account for both labeled runtimes")
}
func (g *gcpRuntimeRun) logs(t *testing.T) {
	t.Helper()
	var l gcpLogs
	g.decode(
		t,
		"logs-unique-gcp",
		g.cp9(t, "logs-unique-gcp", "o11y", false, "{sys.output.format=json}", "logs", "service=gcp-regional-server", "since=10m", "limit=20"),
		&l,
	)
	g.check(t, "o11y", l.Service == "gcp-regional-server" && l.Limit == 20 && len(l.Sources) > 0 && len(l.Attempts) > 0, "bounded GCP service logs")
	for _, s := range l.Sources {
		g.check(t, "o11y", strings.HasPrefix(s, "gcp-regional/"), "GCP-only log source")
	}
	for _, a := range l.Attempts {
		g.check(t, "o11y", a.Substrate == "gcp-regional", "GCP-only attempted log source")
	}
	trace := g.trace(t, "logs-trace-both-runtimes")
	g.attempts(t, &trace, "queried")
}
func (g *gcpRuntimeRun) hash(t *testing.T, text string) string {
	t.Helper()
	fields := strings.Fields(text)
	g.check(t, "harness", len(fields) > 0 && regexp.MustCompile(`^[a-f0-9]{64}$`).MatchString(fields[0]), "SHA-256 observation")
	return fields[0]
}
func (g *gcpRuntimeRun) refresh(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	primary := g.primary("sha256sum", l.Primary.Context)
	regional := g.remote(t, "sha256sum", l.Regional.Context)
	pb := g.hash(t, g.before(t, "primary-context-before", "refresh-gcp", primary...))
	gb := g.hash(t, g.before(t, "gcp-context-before", "refresh-gcp", regional...))
	beforePath := filepath.Join(g.root, "snapshot-before-refresh.json")
	if _, e := os.Stat(beforePath); errors.Is(e, os.ErrNotExist) {
		_, started := g.journal.Actions["refresh-gcp"]
		g.check(t, "harness", !started, "snapshot before refresh is missing")
		g.copyFile(t, filepath.Join(g.stateDir, "artifacts", "snapshot.json"), beforePath)
	} else if e != nil {
		t.Fatal(e)
	}
	var before, after struct {
		Digest string `json:"digest"`
	}
	g.readJSON(t, beforePath, &before)
	result := g.cp9(t, "refresh-gcp", "product", true, "artifacts.refresh", "artifacts=gcp-regional-server", "runtime=gcp-regional")
	g.readJSON(t, filepath.Join(g.stateDir, "artifacts", "snapshot.json"), &after)
	g.copyFile(t, filepath.Join(g.stateDir, "artifacts", "snapshot.json"), filepath.Join(g.root, "snapshot-after-refresh.json"))
	g.check(t, "harness", before.Digest != "" && after.Digest != "", "owner snapshot digests")
	g.check(
		t,
		"product",
		strings.HasPrefix(result.Stdout, "refreshed dev/gcp-regional artifacts gcp-regional-server ("+after.Digest+"); snapshot "),
		"refresh output runtime and persisted digest",
	)
	g.complete(t, "product")
	pa := g.hash(t, g.command(t, "primary-context-after", "harness", false, primary...).Stdout)
	ga := g.hash(t, g.command(t, "gcp-context-after", "harness", false, regional...).Stdout)
	g.save(
		t,
		"artifact-refresh-receipt.json",
		map[string]string{
			"primary_before": pb,
			"primary_after":  pa,
			"gcp_before":     gb,
			"gcp_after":      ga,
			"digest_before":  before.Digest,
			"digest_after":   after.Digest,
		},
	)
	g.check(t, "harness", pb == pa, "refresh changed primary context")
	g.check(t, "harness", (before.Digest == after.Digest) == (gb == ga), "only changed GCP artifact digest may change GCP context")
}

type gcpInventoryItem struct {
	Name   string            `json:"name"`
	Labels map[string]string `json:"labels"`
}
type gcpInventoryCount struct {
	Kind    string `json:"kind"`
	Scanned int    `json:"scanned"`
	Matched int    `json:"matched"`
}

func (g *gcpRuntimeRun) inventoryFamily(t *testing.T, kind string, required bool, args ...string) gcpInventoryCount {
	t.Helper()
	argv := append([]string{"gcloud"}, args...)
	argv = append(argv, "--project=pingcap-gardener", "--format=json")
	var items []gcpInventoryItem
	g.decode(t, "inventory-"+kind, g.command(t, "inventory-"+kind, "harness", false, argv...), &items)
	g.check(t, "harness", items != nil, "inventory must be a JSON array, not null or empty command output")
	count := gcpInventoryCount{Kind: kind, Scanned: len(items)}
	for _, item := range items {
		g.check(t, "harness", item.Name != "", "inventory item missing name")
		name := filepath.Base(item.Name)
		if name == g.sessionName || strings.HasPrefix(name, g.sessionName+"-") || item.Labels["cp9-session"] == g.sessionName {
			count.Matched++
		}
	}
	g.check(t, "harness", !required || count.Matched > 0, "missing session-attributable "+kind)
	return count
}
func (g *gcpRuntimeRun) inventory(t *testing.T) {
	t.Helper()
	// Direct cloud observations replace the deleted proof-only cp9 inventory command.
	addresses := g.inventoryFamily(t, "addresses", true, "compute", "addresses", "list", "--regions=us-west1")
	disks := g.inventoryFamily(t, "disks", true, "compute", "disks", "list", "--zones=us-west1-a")
	firewalls := g.inventoryFamily(t, "firewalls", true, "compute", "firewall-rules", "list")
	clusters := g.inventoryFamily(t, "gke", false, "container", "clusters", "list", "--region=us-west1")
	instances := g.inventoryFamily(t, "instances", true, "compute", "instances", "list", "--zones=us-west1-a")
	networks := g.inventoryFamily(t, "networks", true, "compute", "networks", "list")
	accounts := g.inventoryFamily(t, "service-accounts", false, "iam", "service-accounts", "list")
	subnets := g.inventoryFamily(t, "subnets", true, "compute", "networks", "subnets", "list", "--regions=us-west1")
	var policy struct {
		Bindings []struct {
			Role    string   `json:"role"`
			Members []string `json:"members"`
		} `json:"bindings"`
	}
	g.decode(
		t,
		"inventory-iam-bindings",
		g.command(t, "inventory-iam-bindings", "harness", false, "gcloud", "projects", "get-iam-policy", "pingcap-gardener", "--format=json"),
		&policy,
	)
	g.check(t, "harness", policy.Bindings != nil, "IAM inventory must contain bindings")
	bindings := gcpInventoryCount{Kind: "iam-bindings", Scanned: len(policy.Bindings)}
	for _, b := range policy.Bindings {
		g.check(t, "harness", b.Role != "" && b.Members != nil, "IAM binding shape")
		for _, member := range b.Members {
			if strings.Contains(member, g.sessionName) {
				bindings.Matched++
				break
			}
		}
	}
	g.save(t, "gcp-inventory-summary.json", struct {
		Project  string              `json:"project"`
		Region   string              `json:"region"`
		Session  string              `json:"session"`
		Families []gcpInventoryCount `json:"families"`
	}{"pingcap-gardener", "us-west1", g.sessionName, []gcpInventoryCount{addresses, disks, firewalls, clusters, bindings, instances, networks, accounts, subnets}})
}
func (g *gcpRuntimeRun) pid(t *testing.T, text string) int {
	t.Helper()
	n, e := strconv.Atoi(strings.TrimSpace(text))
	g.check(t, "harness", e == nil && n > 0, "positive controller PID")
	return n
}
func (g *gcpRuntimeRun) stop(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	g.pid(t, g.before(t, "primary-pid-before", "stop-gcp", g.primary("systemctl", "show", l.Primary.Unit, "--property=MainPID", "--value")...))
	g.pid(t, g.before(t, "gcp-pid-before", "stop-gcp", g.remote(t, "systemctl", "show", l.Regional.Unit, "--property=MainPID", "--value")...))
	g.command(t, "stop-gcp", "harness", true, g.remote(t, "sudo", "systemctl", "stop", l.Regional.Unit)...)
	g.status(t, "status-gcp-stopped", "partial", "runtime_failed")
}
func (g *gcpRuntimeRun) partialLogs(t *testing.T) {
	t.Helper()
	l := g.trace(t, "logs-trace-gcp-stopped")
	g.check(t, "o11y", l.Status == "partial", "trace status during GCP outage")
	g.attempts(t, &l, "failed")
}
func (g *gcpRuntimeRun) resume(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	g.cp9(t, "resume-gcp", "product", true, "up")
	g.status(t, "status-ready-after", "ready", "runtime_ready")
	g.complete(t, "product")
	g.activeFailure = "harness"
	pb, e := os.ReadFile(filepath.Join(g.root, "primary-pid-before.txt"))
	if e != nil {
		t.Fatal(e)
	}
	gb, e := os.ReadFile(filepath.Join(g.root, "gcp-pid-before.txt"))
	if e != nil {
		t.Fatal(e)
	}
	pa := g.pid(
		t,
		g.command(
			t,
			"primary-pid-after",
			"harness",
			false,
			g.primary("systemctl", "show", l.Primary.Unit, "--property=MainPID", "--value")...).Stdout,
	)
	ga := g.pid(
		t,
		g.command(t, "gcp-pid-after", "harness", false, g.remote(t, "systemctl", "show", l.Regional.Unit, "--property=MainPID", "--value")...).Stdout,
	)
	pBefore, gBefore := g.pid(t, string(pb)), g.pid(t, string(gb))
	g.save(t, "controller-pid-receipt.json", map[string]int{"primary_before": pBefore, "primary_after": pa, "gcp_before": gBefore, "gcp_after": ga})
	g.check(t, "harness", pa == pBefore && ga != gBefore, "repair must restart only GCP controller")
}

func gcpRemotePath(path string) string {
	if tail, ok := strings.CutPrefix(path, "~/"); ok {
		return `"${HOME}/"` + gcpShell(tail)
	}
	return gcpShell(path)
}
func (g *gcpRuntimeRun) kubectl(t *testing.T, dedicated bool, args ...string) []string {
	t.Helper()
	l := g.locators(t)
	path := l.Host.Remote.Kubeconfig
	argv := []string{"kubectl"}
	if dedicated {
		path = l.Host.DedicatedAPI.Kubeconfig
		argv = append(argv, "--context", "cp9-dedicated")
	}
	g.check(t, "harness", path != "", "GCP kubeconfig locator")
	argv = append(argv, args...)
	quoted := make([]string, len(argv))
	for i, a := range argv {
		quoted[i] = gcpShell(a)
	}
	return g.remote(t, "sh", "-c", "KUBECONFIG="+gcpRemotePath(path)+" exec "+strings.Join(quoted, " "))
}
func (g *gcpRuntimeRun) objects(t *testing.T, label string, argv ...string) ownerObjectList {
	t.Helper()
	var list ownerObjectList
	g.decode(t, label, g.command(t, label, "harness", false, argv...), &list)
	g.check(t, "harness", list.Items != nil, "Kubernetes list shape for "+label)
	return list
}
func (g *gcpRuntimeRun) objectNames(t *testing.T, list ownerObjectList, want []string) {
	t.Helper()
	names := make([]string, 0, len(list.Items))
	for i := range list.Items {
		names = append(names, list.Items[i].Metadata.Name)
	}
	g.check(t, "harness", gcpSameNames(names, want), "exact live object membership")
}

type gcpFirewall struct {
	Network string   `json:"network"`
	Sources []string `json:"sourceRanges"`
	Allowed []struct {
		Protocol string   `json:"IPProtocol"`
		Ports    []string `json:"ports"`
	} `json:"allowed"`
}

func (g *gcpRuntimeRun) firewall(t *testing.T, label, suffix, source, port string) {
	t.Helper()
	var f gcpFirewall
	g.decode(
		t,
		label,
		g.command(
			t,
			label,
			"harness",
			false,
			"gcloud",
			"compute",
			"firewall-rules",
			"describe",
			g.sessionName+suffix,
			"--project=pingcap-gardener",
			"--format=json",
		),
		&f,
	)
	g.check(
		t,
		"security",
		filepath.Base(f.Network) == g.sessionName+"-gcp-regional" && slices.Equal(f.Sources, []string{source}) && len(f.Allowed) == 1 && f.Allowed[0].Protocol == "tcp" &&
			slices.Equal(f.Allowed[0].Ports, []string{port}),
		"exact session network and ingress allowlist: "+label,
	)
}
func (g *gcpRuntimeRun) providers(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	g.firewall(t, "gcp-iap-firewall", "-gcp-regional-iap-ssh", "35.235.240.0/20", "22")
	g.check(
		t,
		"harness",
		l.Host.GCP.ServiceAccount == "cp9-developer@pingcap-gardener.iam.gserviceaccount.com",
		"shared runtime service-account locator",
	)
	var keys []struct {
		KeyType string `json:"keyType"`
	}
	g.decode(
		t,
		"gcp-service-account-keys",
		g.command(
			t,
			"gcp-service-account-keys",
			"harness",
			false,
			"gcloud",
			"iam",
			"service-accounts",
			"keys",
			"list",
			"--iam-account="+l.Host.GCP.ServiceAccount,
			"--project=pingcap-gardener",
			"--format=json",
		),
		&keys,
	)
	g.check(t, "harness", keys != nil, "service-account key list shape")
	for _, k := range keys {
		g.check(t, "security", k.KeyType != "" && k.KeyType != "USER_MANAGED", "no user-managed runtime service-account keys")
	}
	var config struct {
		Spec struct {
			Project     string `json:"projectID"`
			Credentials struct {
				Source string `json:"source"`
			} `json:"credentials"`
		} `json:"spec"`
	}
	g.decode(
		t,
		"gcp-providerconfig",
		g.command(
			t,
			"gcp-providerconfig",
			"harness",
			false,
			g.kubectl(t, false, "get", "providerconfig.gcp.upbound.io", "pingcap-gardener", "-o", "json")...),
		&config,
	)
	g.check(
		t,
		"security",
		config.Spec.Project == "pingcap-gardener" && config.Spec.Credentials.Source == "InjectedIdentity",
		"injected GCP provider identity",
	)
	g.complete(t, "security")
	var crd ownerObject
	g.decode(
		t,
		"gcp-upbound-crd",
		g.command(t, "gcp-upbound-crd", "harness", false, g.kubectl(t, false, "get", "crd", "providerconfigs.gcp.upbound.io", "-o", "json")...),
		&crd,
	)
	g.check(
		t,
		"harness",
		crd.Metadata.Name == "providerconfigs.gcp.upbound.io" && hasOwnerCondition(crd, "Established"),
		"Upbound ProviderConfig CRD Established",
	)
	crds := []string{
		"providerconfigs.gcp.crossplane.io",
		"providerconfigs.aws.crossplane.io",
		"clusters.container.gcp.crossplane.io",
		"nodepools.container.gcp.crossplane.io",
		"hostedzones.route53.aws.crossplane.io",
		"resourcerecordsets.route53.aws.crossplane.io",
		"certificateauthorities.acmpca.aws.crossplane.io",
		"certificateauthoritypermissions.acmpca.aws.crossplane.io",
	}
	args := append([]string{"get", "crd"}, crds...)
	args = append(args, "-o", "json")
	live := g.objects(t, "gcp-required-crds", g.kubectl(t, true, args...)...)
	g.objectNames(t, live, crds)
	for i := range live.Items {
		g.check(t, "harness", hasOwnerCondition(live.Items[i], "Established"), "Established CRD: "+live.Items[i].Metadata.Name)
	}
	providers := g.objects(t, "gcp-providers", g.kubectl(t, false, "get", "providers.pkg.crossplane.io", "-o", "json")...)
	g.objectNames(
		t,
		providers,
		[]string{
			"provider-family-gcp",
			"upbound-provider-gcp-compute",
			"upbound-provider-gcp-container",
			"upbound-provider-gcp-dns",
			"provider-gcp-iam",
			"upbound-provider-gcp-storage",
		},
	)
	revisions := g.objects(t, "gcp-provider-revisions", g.kubectl(t, false, "get", "providerrevisions.pkg.crossplane.io", "-o", "json")...)
	for i := range providers.Items {
		p := &providers.Items[i]
		g.check(
			t,
			"harness",
			hasOwnerCondition(*p, "Healthy") && hasOwnerCondition(*p, "Installed") && p.Status.CurrentRevision != "",
			"Installed Healthy provider: "+p.Metadata.Name,
		)
		found := false
		for j := range revisions.Items {
			revision := &revisions.Items[j]
			if revision.Metadata.Name == p.Status.CurrentRevision {
				g.check(t, "harness", hasOwnerCondition(*revision, "Healthy"), "Healthy current provider revision: "+revision.Metadata.Name)
				found = true
			}
		}
		g.check(t, "harness", found, "current provider revision missing: "+p.Status.CurrentRevision)
	}
	deployments := g.objects(
		t,
		"gcp-classic-provider-deployments",
		g.kubectl(
			t,
			false,
			"-n",
			"cp9-dedicated",
			"get",
			"deployment",
			"gcp-provider",
			"gcp-provider-upstream",
			"aws-provider-upstream",
			"-o",
			"json",
		)...)
	g.objectNames(t, deployments, []string{"aws-provider-upstream", "gcp-provider", "gcp-provider-upstream"})
	for i := range deployments.Items {
		d := &deployments.Items[i]
		g.check(
			t,
			"harness",
			d.Spec.Replicas != nil && *d.Spec.Replicas > 0 && d.Status.ReadyReplicas == *d.Spec.Replicas,
			"classic provider deployment Ready: "+d.Metadata.Name,
		)
	}
}

type gcpIdentityOutputs struct {
	Identity struct {
		Value struct {
			Audience string `json:"audience"`
			Role     string `json:"role_arn"`
			UniqueID string `json:"service_account_unique_id"`
		} `json:"value"`
	} `json:"identity"`
	TLS struct {
		Value struct {
			Certificate string `json:"certificate"`
			Chain       string `json:"certificate_chain"`
			PrivateKey  string `json:"private_key"`
		} `json:"value"`
	} `json:"regional_tls"`
}

func (g *gcpRuntimeRun) identityOutputs(t *testing.T) gcpIdentityOutputs {
	t.Helper()
	var o gcpIdentityOutputs
	g.readJSON(t, filepath.Join(g.stateDir, "terraform-gcp-aws-identity", "outputs.json"), &o)
	g.check(
		t,
		"security",
		o.Identity.Value.Audience != "" && o.Identity.Value.Role != "" && o.Identity.Value.UniqueID != "" && o.TLS.Value.Certificate != "" && o.TLS.Value.Chain != "" &&
			o.TLS.Value.PrivateKey != "",
		"Google-to-AWS identity and TLS outputs present",
	)
	// Never copy the private key or full Terraform outputs into evidence.
	g.save(t, "gcp-aws-identity-contract.json", struct {
		Identity           any  `json:"identity"`
		CertificatePresent bool `json:"certificate_present"`
		ChainPresent       bool `json:"chain_present"`
		PrivateKeyPresent  bool `json:"private_key_present"`
	}{o.Identity.Value, true, true, true})
	return o
}

type gcpCurlObservation struct {
	ExitCode int      `json:"exitcode"`
	Error    string   `json:"errormsg"`
	Connect  *float64 `json:"time_connect"`
	TLS      *float64 `json:"time_appconnect"`
}

func gcpCurlClassification(c gcpCurlObservation) string {
	if c.Connect == nil || c.TLS == nil {
		return "incomplete_curl_receipt"
	}
	switch {
	case c.ExitCode == 6:
		return "dns_failure"
	case c.ExitCode == 45 || strings.Contains(strings.ToLower(c.Error), "bind") || strings.Contains(strings.ToLower(c.Error), "interface"):
		return "binding_failure"
	case *c.Connect > 0 && *c.TLS == 0:
		return "tcp_connected_tls_or_tun_failure"
	case *c.Connect == 0 && *c.TLS == 0 && (c.ExitCode == 7 || c.ExitCode == 28):
		return "tcp_unconnected"
	case *c.TLS > 0:
		return "tls_connected_request_failed"
	default:
		return fmt.Sprintf("unexpected_curl_exit_%d", c.ExitCode)
	}
}
func (g *gcpRuntimeRun) tls(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	var primary gcpHost
	g.readJSON(t, filepath.Join(g.stateDir, "host.json"), &primary)
	g.check(
		t,
		"harness",
		primary.Remote.IP != "" && l.Host.GCP.Host != "" && l.Host.Remote.IP != "" && l.Host.GCP.URL == "https://"+l.Host.GCP.Host,
		"Regional ingress locators",
	)
	g.firewall(t, "gcp-kong-firewall", "-gcp-regional-kong", primary.Remote.IP+"/32", "443")
	argv := []string{
		"curl",
		"--noproxy",
		"*",
		"--resolve",
		l.Host.GCP.Host + ":443:" + l.Host.Remote.IP,
		"--insecure",
		"--silent",
		"--show-error",
		"--connect-timeout",
		"5",
		"--max-time",
		"10",
		"--output",
		"/dev/null",
		"--write-out",
		"%{json}\n",
	}
	if iface := os.Getenv("CP9_RUNNER_NEGATIVE_INTERFACE"); iface != "" {
		argv = append(argv, "--interface", iface)
	}
	argv = append(argv, l.Host.GCP.URL+"/")
	old := g.commander.commandEnv
	env := maps.Clone(old)
	for _, k := range []string{"HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy"} {
		env[k] = ""
	}
	env["NO_PROXY"], env["no_proxy"] = l.Host.GCP.Host, l.Host.GCP.Host
	g.commander.commandEnv = env
	r := g.runDir(t, "gcp-regional-local-negative", time.Minute, g.cp9Root, argv...)
	g.commander.commandEnv = old
	var c gcpCurlObservation
	g.decode(t, "gcp-regional-local-negative", r, &c)
	classification := gcpCurlClassification(c)
	g.save(
		t,
		"gcp-regional-local-negative-classification.json",
		map[string]any{
			"classification": classification,
			"curl_rc":        r.ExitCode,
			"interface":      os.Getenv("CP9_RUNNER_NEGATIVE_INTERFACE"),
			"resolved_ip":    l.Host.Remote.IP,
		},
	)
	g.check(t, "harness", c.ExitCode == r.ExitCode, "curl process/JSON exit agreement")
	g.check(t, "security", r.ExitCode != 0, "Regional ingress reachable outside the primary AWS /32")
	g.check(t, "harness", classification == "tcp_unconnected", "negative network proof is inconclusive: "+classification)
	o := g.identityOutputs(t)
	live := g.command(
		t,
		"gcp-regional-live-cert",
		"harness",
		false,
		g.primary("openssl", "s_client", "-connect", l.Host.GCP.Host+":443", "-servername", l.Host.GCP.Host, "-showcerts")...)
	leaf := g.parseCertificate(t, live.Stdout)
	contract := g.parseCertificate(t, o.TLS.Value.Certificate)
	g.writeText(t, "gcp-regional-live-cert.pem", string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: leaf.Raw})))
	g.writeText(t, "gcp-regional-contract-cert.pem", o.TLS.Value.Certificate)
	g.writeText(t, "gcp-regional-ca-chain.pem", o.TLS.Value.Chain)
	roots := x509.NewCertPool()
	g.check(t, "harness", roots.AppendCertsFromPEM([]byte(o.TLS.Value.Chain)), "parse session PCA chain")
	_, err := leaf.Verify(x509.VerifyOptions{Roots: roots, DNSName: l.Host.GCP.Host})
	g.check(t, "security", err == nil, "live Regional certificate chain and hostname verification: "+fmt.Sprint(err))
	g.check(t, "security", reflect.DeepEqual(leaf.Raw, contract.Raw), "live Regional leaf fingerprint differs from Terraform certificate")
	g.complete(t, "security")
	g.save(
		t,
		"gcp-regional-tls-receipt.json",
		map[string]any{"hostname": l.Host.GCP.Host, "chain_verified": true, "sha256": fmt.Sprintf("%x", sha256.Sum256(leaf.Raw))},
	)
}
func (g *gcpRuntimeRun) parseCertificate(t *testing.T, text string) *x509.Certificate {
	t.Helper()
	block, _ := pem.Decode([]byte(text))
	g.check(t, "harness", block != nil && block.Type == "CERTIFICATE", "PEM certificate capture")
	c, e := x509.ParseCertificate(block.Bytes)
	g.check(t, "harness", e == nil, "X.509 certificate parse")
	return c
}

func (g *gcpRuntimeRun) identity(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	o := g.identityOutputs(t)
	g.check(
		t,
		"security",
		o.Identity.Value.UniqueID == l.Host.GCP.UniqueID && o.Identity.Value.Role == l.Host.GCP.Role,
		"identity contract agrees with GCP host",
	)
	token := strings.TrimSuffix(l.Host.Remote.StateDir, "/") + "/identity/token"
	unit := "cp9-" + g.sessionName + "-gcp-aws-token.service"
	g.check(t, "harness", l.Host.Remote.StateDir != "", "remote token file locator")
	active := g.command(t, "gcp-aws-token-unit", "harness", false, g.remote(t, "systemctl", "is-active", unit)...)
	g.check(t, "security", strings.TrimSpace(active.Stdout) == "active", "Google-to-AWS token refresh unit active")
	scriptPath := strings.TrimSuffix(l.Host.Remote.StateDir, "/") + "/identity/refresh-aws-web-identity.sh"
	installed := g.command(t, "gcp-token-exec-start", "harness", false, g.remote(t, "systemctl", "show", unit, "--property=ExecStart", "--value")...)
	script := g.command(t, "gcp-token-refresher-script", "harness", false, g.remote(t, "cat", scriptPath)...)
	g.check(
		t,
		"security",
		strings.Contains(installed.Stdout, "path="+scriptPath+" ;") &&
			len(regexp.MustCompile(`(?m)^[ \t]*sleep 2700[ \t]*$`).FindAllString(script.Stdout, -1)) == 1,
		"active token unit uses the installed 45-minute refresher",
	)
	before := g.before(t, "gcp-token-mtime-before", "restart-token", g.remote(t, "stat", "-c", "%Y", token)...)
	b, e := strconv.ParseInt(strings.TrimSpace(before), 10, 64)
	g.check(t, "harness", e == nil && b > 0, "token mtime before restart")
	g.command(t, "restart-token", "security", true, g.remote(t, "sudo", "systemctl", "restart", unit)...)
	deadline := time.Now().Add(30 * time.Second)
	var after int64
	for {
		remaining := time.Until(deadline)
		if remaining <= 0 {
			break
		}
		g.activeFailure = "harness"
		observed := g.mustDir(t, "gcp-token-mtime-after", min(remaining, 10*time.Second), g.cp9Root, g.remote(t, "stat", "-c", "%Y", token)...)
		after, e = strconv.ParseInt(strings.TrimSpace(observed.Stdout), 10, 64)
		g.check(t, "harness", e == nil && after > 0, "token mtime after restart")
		if after > b || time.Now().After(deadline) {
			break
		}
		time.Sleep(time.Second)
	}
	g.save(t, "gcp-token-refresh-receipt.json", map[string]any{"unit": unit, "token_file": token, "mtime_before": b, "mtime_after": after})
	g.check(t, "security", after > b, "token refresher did not publish a newer token within 30 seconds")
	// Decode only approved non-secret claims on the host; the bearer never crosses SSH.
	const claimsTransport = `import base64,json,os,sys,time
path=sys.argv[1]
s=open(path).read().strip().split(".")[1]
p=json.loads(base64.urlsafe_b64decode(s+"="*((4-len(s)%4)%4)))
print(json.dumps({"sub":p["sub"],"aud":p["aud"],"lifetime_seconds":p["exp"]-p["iat"],"remaining_seconds":p["exp"]-int(time.time()),"age_seconds":int(time.time()-os.stat(path).st_mtime)}))`
	var claims struct {
		Sub       string `json:"sub"`
		Aud       string `json:"aud"`
		Lifetime  int    `json:"lifetime_seconds"`
		Remaining int    `json:"remaining_seconds"`
		Age       *int   `json:"age_seconds"`
	}
	g.decode(
		t,
		"gcp-aws-token-claims",
		g.command(t, "gcp-aws-token-claims", "harness", false, g.remote(t, "python3", "-c", claimsTransport, token)...),
		&claims,
	)
	g.check(
		t,
		"security",
		claims.Sub == l.Host.GCP.UniqueID && claims.Aud == o.Identity.Value.Audience && claims.Lifetime >= 3000 && claims.Remaining > 300 && claims.Age != nil && *claims.Age >= 0 &&
			*claims.Age < 3000,
		"Google JWT subject, audience and freshness",
	)
	aws := []string{
		"env",
		"-u", "AWS_ACCESS_KEY_ID", "-u", "AWS_SECRET_ACCESS_KEY", "-u", "AWS_SESSION_TOKEN", "-u", "AWS_SECURITY_TOKEN",
		"-u", "AWS_PROFILE", "-u", "AWS_DEFAULT_PROFILE",
		"-u", "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI", "-u", "AWS_CONTAINER_CREDENTIALS_FULL_URI",
		"-u", "AWS_CONTAINER_AUTHORIZATION_TOKEN", "-u", "AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE",
		"AWS_CONFIG_FILE=/dev/null", "AWS_SHARED_CREDENTIALS_FILE=/dev/null", "AWS_CREDENTIAL_FILE=/dev/null", "BOTO_CONFIG=/dev/null",
		"AWS_ROLE_ARN=" + l.Host.GCP.Role,
		"AWS_WEB_IDENTITY_TOKEN_FILE=" + token,
		"AWS_ROLE_SESSION_NAME=cp9-spec",
		"AWS_REGION=us-west-2",
		"AWS_DEFAULT_REGION=us-west-2",
		"AWS_EC2_METADATA_DISABLED=true",
		"aws",
	}
	caller := g.command(
		t,
		"gcp-aws-caller",
		"security",
		false,
		g.remote(t, append(slices.Clone(aws), "sts", "get-caller-identity", "--query", "Arn", "--output", "text")...)...)
	role := strings.SplitN(l.Host.GCP.Role, ":", 6)
	g.check(t, "harness", len(role) == 6 && role[0] == "arn" && role[1] != "" && role[2] == "iam" && role[3] == "" &&
		role[4] == "804837308162" && strings.HasPrefix(role[5], "role/") && filepath.Base(role[5]) != "role", "expected full AWS role ARN")
	prefix := "arn:" + role[1] + ":sts::" + role[4] + ":assumed-role/" + filepath.Base(role[5]) + "/"
	session, matched := strings.CutPrefix(strings.TrimSpace(caller.Stdout), prefix)
	g.check(t, "security", matched && regexp.MustCompile(`^[A-Za-z0-9_+=,.@-]{2,64}$`).MatchString(session), "GCE web-identity AWS STS principal")
	var state struct {
		Paths struct {
			Contract string `json:"terraform_contract"`
		} `json:"paths"`
	}
	g.readJSON(t, filepath.Join(g.stateDir, "state.json"), &state)
	var contract struct {
		DNS struct {
			Zone string `json:"zone_id"`
		} `json:"dedicated_public_dns"`
		PCA struct {
			ARN string `json:"arn"`
		} `json:"session_product_pca"`
	}
	g.check(t, "harness", ownerSessionChildPath(g.stateDir, state.Paths.Contract), "primary Terraform contract must belong to this session")
	g.readJSON(t, state.Paths.Contract, &contract)
	g.check(t, "harness", contract.DNS.Zone != "" && contract.PCA.ARN != "", "session Route53/PCA locators")
	var records struct {
		Items []struct {
			Name    string `json:"Name"`
			Type    string `json:"Type"`
			Records []struct {
				Value string `json:"Value"`
			} `json:"ResourceRecords"`
		} `json:"ResourceRecordSets"`
	}
	g.decode(
		t,
		"gcp-regional-route53",
		g.command(
			t,
			"gcp-regional-route53",
			"security",
			false,
			g.remote(
				t,
				append(slices.Clone(aws), "route53", "list-resource-record-sets", "--hosted-zone-id", contract.DNS.Zone, "--output", "json")...)...),
		&records,
	)
	found := false
	for _, r := range records.Items {
		if r.Name == l.Host.GCP.Host+"." && r.Type == "A" && len(r.Records) == 1 && r.Records[0].Value == l.Host.Remote.IP {
			found = true
		}
	}
	g.check(t, "security", found, "exact session Regional Route53 A record via GCE principal")
	var pca struct {
		Certificate string `json:"Certificate"`
		Chain       string `json:"CertificateChain"`
	}
	g.decode(
		t,
		"gcp-regional-pca",
		g.command(
			t,
			"gcp-regional-pca",
			"security",
			false,
			g.remote(
				t,
				append(
					slices.Clone(aws),
					"acm-pca",
					"get-certificate-authority-certificate",
					"--certificate-authority-arn",
					contract.PCA.ARN,
					"--output",
					"json",
				)...)...),
		&pca,
	)
	g.check(t, "security", pca.Certificate != "" && pca.Chain != "", "session PCA certificate access via GCE principal")
}

func (g *gcpRuntimeRun) adc(t *testing.T) {
	t.Helper()
	l := g.locators(t)
	g.check(t, "harness", l.Host.GCP.Bucket != "", "GCS ADC bucket locator")
	object := "spec-tests/" + g.sessionName + "/adc.txt"
	payload := "adc-" + g.sessionName
	// This transport performs one HTTP request with the instance's metadata token.
	// Go owns write/read/delete ordering and every assertion. No token is printed.
	const transport = `import json,sys,urllib.request,urllib.parse,urllib.error
method,bucket,obj,payload=sys.argv[1:]
metadata=urllib.request.Request("http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token",headers={"Metadata-Flavor":"Google"})
token=json.load(urllib.request.urlopen(metadata,timeout=20))["access_token"]
name=urllib.parse.quote(obj,safe="")
url="https://storage.googleapis.com/storage/v1/b/"+bucket+"/o/"+name
if method=="POST": url="https://storage.googleapis.com/upload/storage/v1/b/"+bucket+"/o?uploadType=media&name="+name
if method=="GET": url+="?alt=media"
request=urllib.request.Request(url,data=payload.encode() if method=="POST" else None,headers={"Authorization":"Bearer "+token,"Content-Type":"text/plain"},method=method)
try:
 with urllib.request.urlopen(request,timeout=30) as response:
  print(json.dumps({"status":response.status,"body":response.read().decode() if method=="GET" else ""}))
except urllib.error.HTTPError as error:
 print(json.dumps({"status":error.code,"body":""}))`
	invoke := func(t *testing.T, label, method string, action bool) (int, string) {
		t.Helper()
		var r struct {
			Status int    `json:"status"`
			Body   string `json:"body"`
		}
		argv := g.remote(t, "python3", "-c", transport, method, l.Host.GCP.Bucket, object, payload)
		if label == "gcs-adc-read" {
			raw := g.before(t, label, "gcs-adc-delete", argv...)
			g.decode(t, label, resolverCommandResult{Stdout: raw}, &r)
		} else {
			g.decode(t, label, g.command(t, label, "security", action, argv...), &r)
		}
		g.check(t, "harness", r.Status >= 100 && r.Status <= 599, "GCS HTTP status capture")
		return r.Status, r.Body
	}
	status, _ := invoke(t, "gcs-adc-write", "POST", true)
	g.check(t, "security", status == 200, "ADC object upload")
	status, body := invoke(t, "gcs-adc-read", "GET", false)
	g.check(t, "security", status == 200 && body == payload, "ADC object read matches uploaded payload")
	status, _ = invoke(t, "gcs-adc-delete", "DELETE", true)
	g.check(t, "security", status == 204, "ADC object cleanup")
	status, _ = invoke(t, "gcs-adc-absent", "GET", false)
	g.check(t, "security", status == 404, "ADC object absent after delete")
	g.complete(t, "security")
	g.save(t, "gcp-adc-read-write.json", map[string]bool{"write": true, "read": true, "cleanup": true})
}
func (g *gcpRuntimeRun) cleanupDryRuns(t *testing.T) {
	t.Helper()
	var d gcpDryRun
	g.readJSON(t, filepath.Join(g.root, "dry-run.json"), &d)
	g.check(t, "harness", d.Owner != "" && d.Session == g.sessionName, "exact cleanup dry-run owner and session")
	previous := g.commander.commandEnv
	env := maps.Clone(previous)
	env["CP9_CLEANUP_GCP_PROJECT_ID"] = "pingcap-gardener"
	env["CP9_CLEANUP_GCP_REGION"] = "us-west1"
	env["CP9_CLEANUP_OWNER"] = d.Owner
	env["CP9_CLEANUP_SESSION"] = g.sessionName
	env["CP9_CLEANUP_DRY_RUN"] = "true"
	g.commander.commandEnv = env
	defer func() { g.commander.commandEnv = previous }()
	g.command(t, "gcp-runtime-cleanup-dry-run", "harness", false, filepath.Join(g.cp9Root, ".github", "scripts", "cleanup-gcp-runtime.sh"))
	g.command(t, "gcp-product-cleanup-dry-run", "harness", false, filepath.Join(g.cp9Root, ".github", "scripts", "cleanup-gcp-dedicated-products.sh"))
}
func (g *gcpRuntimeRun) cleanup(t *testing.T) {
	t.Helper()
	defer func() {
		if t.Failed() && g.activeFailure == "harness" {
			if err := g.recordHarnessFailure(
				"cleanup-evidence",
				"required cleanup evidence failed; preserve the scene and repair before retry",
			); err != nil {
				t.Errorf("record cleanup evidence failure: %v", err)
			}
		}
	}()
	if g.journal.Cleanup == "pass" {
		t.Skip("reuse passing exact-session cleanup")
	}
	g.activeFailure = "harness"
	_, upStarted := g.journal.Actions["up"]
	archive := filepath.Join(g.root, "pre-destroy-archive")
	if upStarted && g.latestCleanupActionLabel() == "" {
		// Early up/graph failures may precede transport locators. The host layout
		// already identifies the regional state, independently of graph acceptance.
		regional := host.GCPRegionalStateDir(g.stateDir)
		_, locatorErr := os.Lstat(filepath.Join(g.root, "locators.json"))
		if locatorErr != nil && !errors.Is(locatorErr, os.ErrNotExist) {
			t.Fatal(locatorErr)
		}
		up := g.journal.Actions["up"]
		required := locatorErr == nil || (up.Status == "completed" && up.ExitCode == 0) || g.stagesPassed("dual-runtime-ready")
		observed := map[string]bool{}
		manifest := filepath.Join(archive, "availability.json")
		if _, err := os.Lstat(manifest); err == nil {
			g.readJSON(t, manifest, &observed)
			g.check(t, "harness", observed != nil, "archive availability receipt")
		} else if !errors.Is(err, os.ErrNotExist) {
			t.Fatal(err)
		}
		files := map[string]string{
			"state.json":                     filepath.Join(g.stateDir, "state.json"),
			"runtime-plan.json":              filepath.Join(g.stateDir, "runtime-plan.json"),
			"host.json":                      filepath.Join(g.stateDir, "host.json"),
			"gcp-regional-host.json":         filepath.Join(regional, "host.json"),
			"primary-tilt-status.json":       filepath.Join(g.stateDir, "tilt", "reports", "tilt-status.json"),
			"gcp-regional-tilt-status.json":  filepath.Join(regional, "tilt", "reports", "tilt-status.json"),
			"gcp-token-refresh-receipt.json": filepath.Join(g.root, "gcp-token-refresh-receipt.json"),
			"locators.json":                  filepath.Join(g.root, "locators.json"),
		}
		for name, source := range files {
			info, err := os.Lstat(source)
			if errors.Is(err, os.ErrNotExist) {
				g.check(t, "harness", !observed[name] && (!required || name == "gcp-token-refresh-receipt.json" || name == "locators.json") &&
					(name != "gcp-token-refresh-receipt.json" || !g.stagesPassed("google-to-aws-identity")) &&
					(name != "locators.json" || !g.stagesPassed("runtime-graph")), "required or previously observed archive source disappeared: "+source)
				observed[name] = false // Not observed after early failure; not a passing graph.
				continue
			}
			g.check(t, "harness", err == nil && info.Mode().IsRegular(), "archive source must be a readable regular file: "+source)
			observed[name] = true
		}
		g.save(t, "pre-destroy-archive/availability.json", observed)
		for name, source := range files {
			if observed[name] {
				g.copyFile(t, source, filepath.Join(archive, name))
			}
		}
		g.copyFile(t, filepath.Join(g.root, ownerSessionJournalFile), filepath.Join(archive, ownerSessionJournalFile))
	}
	// Reuse the existing exact-session cleanup and immutable attempt receipts.
	g.activeFailure = "cleanup"
	g.ownerSessionRun.cleanup(t)
}