package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/Benjamin2037/tipoc-tv/core/contracts"
	"github.com/Benjamin2037/tipoc-tv/core/execution"
	"github.com/Benjamin2037/tipoc-tv/core/scenario"
)

const (
	envInternalPrepareGenerationWorker = "TIPOC_INTERNAL_PREPARE_GENERATION_WORKER"
	envPrepareGenerationNodeID         = "TIPOC_PREPARE_GENERATION_NODE_ID"
	envPrepareGenerationEndpoint       = "TIPOC_PREPARE_GENERATION_COORDINATOR_ENDPOINT"
	envPrepareGenerationListen         = "TIPOC_PREPARE_GENERATION_COORDINATOR_LISTEN"
	envPrepareGenerationResult         = "TIPOC_PREPARE_GENERATION_RESULT"
	envPrepareOwnerPrefixes            = "TIPOC_PREPARE_OWNER_PREFIXES"
	envRunPressureNodeID               = "TIPOC_RUN_PRESSURE_NODE_ID"
	envRunPressureEndpoint             = "TIPOC_RUN_PRESSURE_COORDINATOR_ENDPOINT"
	envRunPressureListen               = "TIPOC_RUN_PRESSURE_COORDINATOR_LISTEN"
	envRerunScriptPath                 = "TIPOC_RERUN_SCRIPT"

	tipocCommandRerun                    = "rerun"
	tipocCommandOps                      = "ops"
	tipocCommandRUAcceptance             = "ru-acceptance"
	tipocCommandInternal                 = "internal"
	tipocInternalWorkloadProfile         = "workload-profile"
	tipocInternalWorkloadProfileMaterial = "materialize"
	tipocRunPreflight                    = "preflight"
	tipocRunRecover                      = "recover"
)

var runExternalStageCommand = func(ctx context.Context, command []string) ([]byte, error) {
	if len(command) == 0 {
		return nil, fmt.Errorf("command is empty")
	}
	cmd := exec.CommandContext(ctx, command[0], command[1:]...)
	return cmd.CombinedOutput()
}

var runRerunScriptCommand = func(ctx context.Context, args []string, stdout, stderr io.Writer) int {
	scriptPath, err := findRerunScriptPath()
	if err != nil {
		fmt.Fprintf(stderr, "error_code=invalid_stage_input reason=%s\n", contracts.SanitizeErrorDetail(err.Error()))
		return 2
	}
	cmd := exec.CommandContext(ctx, scriptPath, args...)
	cmd.Stdout = stdout
	cmd.Stderr = stderr
	if err := cmd.Run(); err != nil {
		var exitErr *exec.ExitError
		if errors.As(err, &exitErr) {
			return exitErr.ExitCode()
		}
		fmt.Fprintf(stderr, "error_code=external_command_failed reason=%s\n", contracts.SanitizeErrorDetail(err.Error()))
		return 11
	}
	return 0
}

func main() {
	os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))
}

func run(args []string, stdout, stderr io.Writer) int {
	if len(args) == 0 {
		printUsage(stderr)
		return 2
	}

	switch args[0] {
	case scenario.PublicStageInit:
		return runInit(args[1:], stdout, stderr)
	case scenario.PublicStagePrepare, scenario.PublicStageRun, scenario.PublicStageCleanup:
		return runGuardedStage(args[0], args[1:], stdout, stderr)
	case tipocCommandRerun:
		return runRerun(args[1:], stdout, stderr)
	case tipocCommandOps:
		return runOperationControl(args[1:], stdout, stderr)
	case tipocCommandRUAcceptance:
		return runRUAcceptance(args[1:], stdout, stderr)
	case tipocCommandInternal:
		return runInternal(args[1:], stdout, stderr)
	default:
		err := scenario.ValidatePublicStage(args[0])
		writeError(stderr, err)
		printUsage(stderr)
		return 2
	}
}

func runOperationControl(args []string, stdout, stderr io.Writer) int {
	if len(args) == 0 || args[0] == "-h" || args[0] == "--help" {
		fmt.Fprintln(stdout, "usage: tipoc ops <operation-id> --action <start|stop|action-id> --manifest <path> [--wait] [--timeout <duration>] [--request-id <id>]")
		if len(args) == 0 {
			return 2
		}
		return 0
	}
	operationID := strings.TrimSpace(args[0])
	fs := flag.NewFlagSet("ops", flag.ContinueOnError)
	fs.SetOutput(stderr)
	action := fs.String("action", "", "worker command or resolved runtime action ID")
	manifestPath := fs.String("manifest", "", "path to run-profile manifest")
	wait := fs.Bool("wait", false, "wait for the control request to reach a terminal state")
	timeout := fs.Duration("timeout", 5*time.Minute, "maximum time to wait for a terminal receipt")
	requestID := fs.String("request-id", "", "idempotent operation control request ID")
	if err := fs.Parse(args[1:]); err != nil {
		return 2
	}
	if fs.NArg() != 0 || operationID == "" || strings.TrimSpace(*action) == "" || strings.TrimSpace(*manifestPath) == "" {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageRun, "ops", "arguments", "ops requires one operation ID, --action, and --manifest", nil))
	}
	if *timeout <= 0 {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageRun, "ops", "timeout", "ops timeout must be positive", nil))
	}
	id := strings.TrimSpace(*requestID)
	if id == "" {
		id = fmt.Sprintf("ops-%d-%d", os.Getpid(), time.Now().UTC().UnixNano())
	}
	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
	defer cancel()
	request := contracts.OperationControlRequest{
		RequestID:   id,
		OperationID: operationID,
		Action:      strings.TrimSpace(*action),
		Wait:        *wait,
	}
	receipt, err := execution.SubmitOperationControl(ctx, *manifestPath, request)
	if err != nil {
		manifest, loadErr := scenario.LoadManifest(*manifestPath)
		if loadErr == nil {
			if _, statErr := os.Stat(execution.OperationControlEndpointPath(manifest)); os.IsNotExist(statErr) && manifest.Stage.CurrentState == scenario.StagePrepareComplete {
				wiring, wiringErr := newApplicationWiring()
				if wiringErr == nil {
					receipt, err = wiring.executor().ExecutePrepareOperationControl(ctx, *manifestPath, request)
				}
			}
		}
	}
	if err != nil {
		return writeError(stderr, err)
	}
	fmt.Fprintf(stdout, "operation=%s action=%s request_id=%s status=%s\n", receipt.OperationID, receipt.Action, receipt.RequestID, receipt.Status)
	if receipt.ActionReceiptRef != "" {
		fmt.Fprintf(stdout, "action_receipt=%s\n", receipt.ActionReceiptRef)
	}
	for _, ref := range receipt.WorkerReceiptRefs {
		fmt.Fprintf(stdout, "worker_receipt=%s\n", ref)
	}
	return 0
}

func runInternal(args []string, stdout, stderr io.Writer) int {
	if len(args) < 2 ||
		args[0] != tipocInternalWorkloadProfile ||
		args[1] != tipocInternalWorkloadProfileMaterial {
		return writeError(stderr, contracts.NewProductError(
			contracts.ErrorCodeInvalidStageInput,
			scenario.PublicStageRun,
			"internal",
			"command",
			"unknown internal command",
			nil,
		))
	}
	return runInternalWorkloadProfileMaterialize(args[2:], stdout, stderr)
}

func runInternalWorkloadProfileMaterialize(args []string, stdout, stderr io.Writer) int {
	fs := flag.NewFlagSet("internal workload-profile materialize", flag.ContinueOnError)
	fs.SetOutput(stderr)
	manifestPath := fs.String("manifest", "", "new try-local run profile manifest")
	manifestTemplatePath := fs.String("manifest-template", "", "prepared manifest template for the new try")
	overridePath := fs.String("workload-profile-override", "", "path to a JSON or YAML workload profile override")
	profileName := fs.String("workload-profile", "", "compatibility workload profile name")
	attemptRoot := fs.String("attempt-root", "", "local attempt root for a rerun manifest")
	tryRoot := fs.String("try-root", "", "output root for a rerun manifest")
	continuationSource := fs.String("continuation-source", "", "previous run continuation receipt")
	remoteTryManifest := fs.String("remote-try-manifest", "", "remote try-local manifest path")
	remoteTryLogDir := fs.String("remote-try-log-dir", "", "remote try-local log directory")
	durationSeconds := fs.String("duration-seconds", "", "rerun duration override")
	warmupSeconds := fs.String("warmup-seconds", "", "rerun warmup override")
	cooldownSeconds := fs.String("cooldown-seconds", "", "rerun cooldown override")
	recoveryTimeoutSeconds := fs.String("recovery-timeout-seconds", "", "rerun recovery timeout override")
	progressIntervalSecs := fs.String("progress-interval-seconds", "", "rerun progress interval override")
	queryCostMode := fs.String("query-cost-mode", "", "rerun query-cost mode override")
	queryCostAttemptRUSource := fs.String("query-cost-attempt-ru-source", "", "rerun query-cost attempt RU source override")
	queryCostAttemptRUTimeoutMS := fs.String("query-cost-attempt-ru-timeout-milliseconds", "", "rerun query-cost attempt RU timeout override")
	queryCostAttemptRUFailureBudget := fs.String("query-cost-attempt-ru-failure-budget", "", "rerun query-cost attempt RU failure budget override")
	queryCostDeploymentAttestationPath := fs.String("query-cost-deployment-attestation-path", "", "rerun query-cost deployment attestation path")
	queryCostDeploymentAttestationSHA256 := fs.String("query-cost-deployment-attestation-sha256", "", "rerun query-cost deployment attestation SHA-256 pin")
	queryCostPollIntervalSeconds := fs.String("query-cost-poll-interval-seconds", "", "rerun query-cost poll interval override")
	queryCostPollTimeoutSeconds := fs.String("query-cost-poll-timeout-seconds", "", "rerun query-cost poll timeout override")
	queryCostMaxLeafGroups := fs.String("query-cost-max-leaf-groups", "", "rerun query-cost leaf catalog capacity override")
	queryCostMaxRollupGroups := fs.String("query-cost-max-rollup-groups", "", "rerun query-cost rollup catalog capacity override")
	queryCostMaxPlanVariants := fs.String("query-cost-max-plan-variants", "", "rerun query-cost plan variant limit override")
	queryCostIsolationStatus := fs.String("query-cost-isolation-status", "", "rerun query-cost isolation status override")
	queryCostIsolationReason := fs.String("query-cost-isolation-reason", "", "rerun query-cost isolation reason override")
	ruAcceptanceScenarioID := fs.String("ru-acceptance-scenario-id", "", "rerun RU acceptance scenario ID override")
	ruAcceptanceScenarioVariant := fs.String("ru-acceptance-scenario-variant", "", "rerun RU acceptance scenario variant override")
	concurrency := fs.String("concurrency", "", "rerun concurrency override")
	workloadMix := fs.String("workload-mix", "", "rerun workload mix override")
	exchangeMatchingWritePercent := fs.String("exchange-matching-write-percent", "", "rerun exchange matching write percentage override")
	exchangeMatchingSQLWritePercent := fs.String("exchange-matching-sql-write-percent", "", "rerun exchange matching SQL write percentage override")
	pressureNodesJSON := fs.String("pressure-nodes-json", "", "typed rerun pressure node overrides")
	sqlPoolSize := fs.String("sql-pool-size", "", "rerun SQL pool size override")
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if strings.TrimSpace(*manifestPath) == "" {
		return writeError(stderr, contracts.NewProductError(
			contracts.ErrorCodeInvalidStageInput,
			scenario.PublicStageRun,
			"materialize-workload-profile",
			"manifest",
			"internal workload profile materialization requires --manifest",
			nil,
		))
	}
	if strings.TrimSpace(*manifestTemplatePath) == "" {
		return writeError(stderr, contracts.NewProductError(
			contracts.ErrorCodeInvalidStageInput,
			scenario.PublicStageRun,
			"materialize-workload-profile",
			"manifest_template",
			"internal workload profile materialization requires --manifest-template",
			nil,
		))
	}
	var override *contracts.WorkloadProfileOverride
	if strings.TrimSpace(*overridePath) != "" {
		parsed, err := loadWorkloadProfileOverride(*overridePath)
		if err != nil {
			return writeError(stderr, err)
		}
		override = &parsed
	}
	if override != nil && strings.TrimSpace(*profileName) != "" {
		return writeError(stderr, contracts.NewProductError(
			contracts.ErrorCodeInvalidStageInput,
			scenario.PublicStageRun,
			"materialize-workload-profile",
			"workload_profile",
			"workload profile name and structured override cannot be combined",
			nil,
		))
	}
	var rerunOverrides *scenario.RunAgainManifestOverrides
	if strings.TrimSpace(*attemptRoot) != "" || strings.TrimSpace(*tryRoot) != "" || strings.TrimSpace(*pressureNodesJSON) != "" || strings.TrimSpace(*durationSeconds) != "" || strings.TrimSpace(*warmupSeconds) != "" || strings.TrimSpace(*cooldownSeconds) != "" || strings.TrimSpace(*recoveryTimeoutSeconds) != "" || strings.TrimSpace(*progressIntervalSecs) != "" || strings.TrimSpace(*queryCostMode) != "" || strings.TrimSpace(*queryCostAttemptRUSource) != "" || strings.TrimSpace(*queryCostAttemptRUTimeoutMS) != "" || strings.TrimSpace(*queryCostAttemptRUFailureBudget) != "" || strings.TrimSpace(*queryCostDeploymentAttestationPath) != "" || strings.TrimSpace(*queryCostDeploymentAttestationSHA256) != "" || strings.TrimSpace(*queryCostPollIntervalSeconds) != "" || strings.TrimSpace(*queryCostPollTimeoutSeconds) != "" || strings.TrimSpace(*queryCostMaxLeafGroups) != "" || strings.TrimSpace(*queryCostMaxRollupGroups) != "" || strings.TrimSpace(*queryCostMaxPlanVariants) != "" || strings.TrimSpace(*queryCostIsolationStatus) != "" || strings.TrimSpace(*queryCostIsolationReason) != "" || strings.TrimSpace(*ruAcceptanceScenarioID) != "" || strings.TrimSpace(*ruAcceptanceScenarioVariant) != "" || strings.TrimSpace(*concurrency) != "" || strings.TrimSpace(*workloadMix) != "" || strings.TrimSpace(*exchangeMatchingWritePercent) != "" || strings.TrimSpace(*exchangeMatchingSQLWritePercent) != "" || strings.TrimSpace(*sqlPoolSize) != "" {
		parsed, err := parseRunAgainManifestOverrides(
			*attemptRoot,
			*tryRoot,
			*continuationSource,
			*remoteTryManifest,
			*remoteTryLogDir,
			*durationSeconds,
			*warmupSeconds,
			*cooldownSeconds,
			*recoveryTimeoutSeconds,
			*progressIntervalSecs,
			*queryCostMode,
			*queryCostAttemptRUSource,
			*queryCostAttemptRUTimeoutMS,
			*queryCostAttemptRUFailureBudget,
			*queryCostDeploymentAttestationPath,
			*queryCostDeploymentAttestationSHA256,
			*queryCostPollIntervalSeconds,
			*queryCostPollTimeoutSeconds,
			*queryCostMaxLeafGroups,
			*queryCostMaxRollupGroups,
			*queryCostMaxPlanVariants,
			*queryCostIsolationStatus,
			*queryCostIsolationReason,
			*ruAcceptanceScenarioID,
			*ruAcceptanceScenarioVariant,
			*concurrency,
			*workloadMix,
			*exchangeMatchingWritePercent,
			*exchangeMatchingSQLWritePercent,
			*pressureNodesJSON,
			*sqlPoolSize,
		)
		if err != nil {
			return writeError(stderr, err)
		}
		rerunOverrides = &parsed
	}
	result, err := scenario.MaterializeWorkloadProfile(*manifestTemplatePath, *manifestPath, scenario.WorkloadProfileMaterializationInput{
		Override:      override,
		ProfileName:   *profileName,
		RerunManifest: rerunOverrides,
	})
	if err != nil {
		return writeError(stderr, err)
	}
	fmt.Fprintf(stdout, "manifest=%s\n", result.ManifestPath)
	fmt.Fprintf(stdout, "workload_profile_resolution=%s\n", result.ResolutionPath)
	fmt.Fprintf(stdout, "workload_profile_materialization=%s\n", result.CommitPath)
	fmt.Fprintf(stdout, "workload_profile_source_digest=%s\n", result.Resolution.SourceDigest)
	fmt.Fprintf(stdout, "workload_profile_resolved_digest=%s\n", result.Resolution.ResolvedDigest)
	return 0
}

func parseRunAgainManifestOverrides(attemptRoot, tryRoot, continuationSource, remoteTryManifest, remoteTryLogDir, durationSeconds, warmupSeconds, cooldownSeconds, recoveryTimeoutSeconds, progressIntervalSecs, queryCostMode, queryCostAttemptRUSource, queryCostAttemptRUTimeoutMS, queryCostAttemptRUFailureBudget, queryCostDeploymentAttestationPath, queryCostDeploymentAttestationSHA256, queryCostPollIntervalSeconds, queryCostPollTimeoutSeconds, queryCostMaxLeafGroups, queryCostMaxRollupGroups, queryCostMaxPlanVariants, queryCostIsolationStatus, queryCostIsolationReason, ruAcceptanceScenarioID, ruAcceptanceScenarioVariant, concurrency, workloadMix, exchangeMatchingWritePercent, exchangeMatchingSQLWritePercent, pressureNodesJSON, sqlPoolSize string) (scenario.RunAgainManifestOverrides, error) {
	if strings.TrimSpace(attemptRoot) == "" {
		return scenario.RunAgainManifestOverrides{}, fmt.Errorf("rerun manifest materialization requires --attempt-root")
	}
	parseInt64 := func(name, value string) (*int64, error) {
		if strings.TrimSpace(value) == "" {
			return nil, nil
		}
		parsed, err := strconv.ParseInt(value, 10, 64)
		if err != nil {
			return nil, fmt.Errorf("invalid %s %q: %w", name, value, err)
		}
		return &parsed, nil
	}
	parseInt := func(name, value string) (*int, error) {
		if strings.TrimSpace(value) == "" {
			return nil, nil
		}
		parsed, err := strconv.Atoi(value)
		if err != nil {
			return nil, fmt.Errorf("invalid %s %q: %w", name, value, err)
		}
		return &parsed, nil
	}
	duration, err := parseInt64("duration-seconds", durationSeconds)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	warmup, err := parseInt64("warmup-seconds", warmupSeconds)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	cooldown, err := parseInt64("cooldown-seconds", cooldownSeconds)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	recoveryTimeout, err := parseInt64("recovery-timeout-seconds", recoveryTimeoutSeconds)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	progress, err := parseInt64("progress-interval-seconds", progressIntervalSecs)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	queryCostAttemptTimeout, err := parseInt("query-cost-attempt-ru-timeout-milliseconds", queryCostAttemptRUTimeoutMS)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	queryCostAttemptFailureBudget, err := parseInt("query-cost-attempt-ru-failure-budget", queryCostAttemptRUFailureBudget)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	queryCostPollInterval, err := parseInt("query-cost-poll-interval-seconds", queryCostPollIntervalSeconds)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	queryCostPollTimeout, err := parseInt("query-cost-poll-timeout-seconds", queryCostPollTimeoutSeconds)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	queryCostLeafGroups, err := parseInt("query-cost-max-leaf-groups", queryCostMaxLeafGroups)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	queryCostRollupGroups, err := parseInt("query-cost-max-rollup-groups", queryCostMaxRollupGroups)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	queryCostPlanVariants, err := parseInt("query-cost-max-plan-variants", queryCostMaxPlanVariants)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	parallelism, err := parseInt("concurrency", concurrency)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	writePercent, err := parseInt("exchange-matching-write-percent", exchangeMatchingWritePercent)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	sqlWritePercent, err := parseInt("exchange-matching-sql-write-percent", exchangeMatchingSQLWritePercent)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	poolSize, err := parseInt("sql-pool-size", sqlPoolSize)
	if err != nil {
		return scenario.RunAgainManifestOverrides{}, err
	}
	overrides := scenario.RunAgainManifestOverrides{
		AttemptRoot:                          attemptRoot,
		TryRoot:                              tryRoot,
		ContinuationSource:                   continuationSource,
		RemoteTryManifest:                    remoteTryManifest,
		RemoteTryLogDir:                      remoteTryLogDir,
		DurationSeconds:                      duration,
		WarmupSeconds:                        warmup,
		CooldownSeconds:                      cooldown,
		RerunRecoveryTimeoutSeconds:          recoveryTimeout,
		ProgressIntervalSecs:                 progress,
		QueryCostMode:                        strings.TrimSpace(strings.ToLower(queryCostMode)),
		QueryCostAttemptRUSource:             strings.TrimSpace(strings.ToLower(queryCostAttemptRUSource)),
		QueryCostAttemptRUTimeoutMS:          queryCostAttemptTimeout,
		QueryCostAttemptRUFailureBudget:      queryCostAttemptFailureBudget,
		QueryCostDeploymentAttestationPath:   strings.TrimSpace(queryCostDeploymentAttestationPath),
		QueryCostDeploymentAttestationSHA256: strings.TrimSpace(strings.ToLower(queryCostDeploymentAttestationSHA256)),
		QueryCostPollIntervalSeconds:         queryCostPollInterval,
		QueryCostPollTimeoutSeconds:          queryCostPollTimeout,
		QueryCostMaxLeafGroups:               queryCostLeafGroups,
		QueryCostMaxRollupGroups:             queryCostRollupGroups,
		QueryCostMaxPlanVariants:             queryCostPlanVariants,
		QueryCostIsolationStatus:             queryCostIsolationStatus,
		QueryCostIsolationReason:             queryCostIsolationReason,
		RUAcceptanceScenarioID:               strings.TrimSpace(ruAcceptanceScenarioID),
		RUAcceptanceScenarioVariant:          strings.TrimSpace(ruAcceptanceScenarioVariant),
		Concurrency:                          parallelism,
		WorkloadMix:                          workloadMix,
		ExchangeMatchingWritePercent:         writePercent,
		ExchangeMatchingSQLWritePercent:      sqlWritePercent,
		SQLPoolSize:                          poolSize,
	}
	if strings.TrimSpace(pressureNodesJSON) != "" {
		if err := json.Unmarshal([]byte(pressureNodesJSON), &overrides.PressureNodes); err != nil {
			return scenario.RunAgainManifestOverrides{}, fmt.Errorf("invalid pressure-nodes-json: %w", err)
		}
		overrides.PressureNodesSet = true
	}
	return overrides, nil
}

func runInit(args []string, stdout, stderr io.Writer) int {
	fs := flag.NewFlagSet("init", flag.ContinueOnError)
	fs.SetOutput(stderr)
	specPath := fs.String("spec", "", "path to spec.yaml")
	internetSpecPath := fs.String("internet-spec", "", "path to the canonical Internet spec YAML")
	datasetIdentity := fs.String("dataset-identity", "", "immutable dataset identity for fresh Internet init")
	templateName := fs.String("template", "", "write a spec.yaml template by name")
	fromAttempt := fs.String("from-attempt", "", "previous attempt root containing spec.yaml")
	attemptID := fs.String("attempt-id", "", "override attempt id when deriving from a previous attempt")
	out := fs.String("out", "", "attempt output directory")
	remoteBootstrap := fs.String("remote-bootstrap", string(scenario.InitArtifactModePlan), "remote bootstrap mode: plan or execute")
	bigQueryFlags := bindBigQueryFlags(fs)
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if *out == "" {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageInit, "parse", "init.out", "init requires --out", nil))
	}
	if countNonEmpty(*specPath, *templateName, *fromAttempt) != 1 {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageInit, "parse", "init.source", "init requires exactly one of --spec, --template, or --from-attempt", nil))
	}
	if *templateName != "" {
		spec, err := scenario.TemplateSpec(*templateName)
		if err != nil {
			return writeError(stderr, err)
		}
		if err := applyBigQueryInitFlags(&spec, bigQueryFlags); err != nil {
			return writeError(stderr, err)
		}
		ingress, err := loadInternetInitIngress(spec, *internetSpecPath, *datasetIdentity, "")
		if err != nil {
			return writeError(stderr, err)
		}
		return writeInitializedAttempt(spec, ingress, *out, *attemptID, *remoteBootstrap, stdout, stderr)
	}
	if *fromAttempt != "" {
		priorManifest, err := scenario.LoadManifest(filepath.Join(*fromAttempt, scenario.ManifestFileName))
		if err != nil {
			return writeError(stderr, err)
		}
		if err := scenario.ValidateInternetManifestBridge(priorManifest); err != nil {
			return writeError(stderr, err)
		}
		spec, err := loadInitSpec("", *fromAttempt)
		if err != nil {
			return writeError(stderr, err)
		}
		if spec.Scenario.Industry != priorManifest.Workload.Industry || spec.Identity.RunID != priorManifest.RunID || spec.Identity.AttemptID != priorManifest.AttemptID {
			return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeManifestInput, scenario.PublicStageInit, "load-from-attempt", "spec", "saved spec identity differs from the prior run profile", nil))
		}
		if err := applyBigQueryInitFlags(&spec, bigQueryFlags); err != nil {
			return writeError(stderr, err)
		}
		ingress, err := loadInternetInitIngress(spec, *internetSpecPath, *datasetIdentity, *fromAttempt)
		if err != nil {
			return writeError(stderr, err)
		}
		return writeInitializedAttempt(spec, ingress, *out, *attemptID, *remoteBootstrap, stdout, stderr)
	}
	spec, err := loadInitSpec(*specPath, *fromAttempt)
	if err != nil {
		return writeError(stderr, err)
	}
	if err := applyBigQueryInitFlags(&spec, bigQueryFlags); err != nil {
		return writeError(stderr, err)
	}
	ingress, err := loadInternetInitIngress(spec, *internetSpecPath, *datasetIdentity, *fromAttempt)
	if err != nil {
		return writeError(stderr, err)
	}
	return writeInitializedAttempt(spec, ingress, *out, *attemptID, *remoteBootstrap, stdout, stderr)
}

func writeInitializedAttempt(spec contracts.ScenarioSpec, internet *scenario.InternetInitIngress, out, attemptID, remoteBootstrap string, stdout, stderr io.Writer) int {
	if attemptID != "" {
		spec.Identity.AttemptID = attemptID
	}
	attemptRoot, err := filepath.Abs(out)
	if err != nil {
		return writeError(stderr, err)
	}
	spec.Init.AttemptRoot = attemptRoot
	wiring, err := newApplicationWiring()
	if err != nil {
		return writeError(stderr, err)
	}
	manifest, err := wiring.compiler().Compile(context.Background(), spec)
	if err != nil {
		return writeError(stderr, err)
	}
	initOptions := scenario.InitArtifactOptions{
		AttemptRoot:             attemptRoot,
		RemoteBootstrapMode:     scenario.InitArtifactMode(remoteBootstrap),
		InternetSpec:            internetSpecBytes(internet),
		InternetDatasetIdentity: internetDatasetIdentity(internet),
	}
	if manifest.Workload.Industry == contracts.IndustryInternet {
		compiled, ref, compileErr := compileAndPublishInternetInit(context.Background(), manifest.Scenario, manifest, initOptions)
		if compileErr != nil {
			return writeError(stderr, compileErr)
		}
		if compiled.Scenario.RunID != manifest.RunID || compiled.Scenario.AttemptID != manifest.AttemptID {
			return writeError(stderr, fmt.Errorf("compiled Internet manifest identity differs from the public manifest"))
		}
		manifest.InternetManifestRef = coreInternetInitArtifactRef(ref)
	}
	result, err := scenario.WriteInitArtifacts(context.Background(), manifest.Scenario, manifest, scenario.InitArtifactOptions{
		AttemptRoot:             initOptions.AttemptRoot,
		RemoteBootstrapMode:     initOptions.RemoteBootstrapMode,
		InternetSpec:            initOptions.InternetSpec,
		InternetDatasetIdentity: initOptions.InternetDatasetIdentity,
	})
	if err != nil {
		return writeError(stderr, err)
	}
	fmt.Fprintf(stdout, "spec=%s\n", result.SpecPath)
	if result.InternetSpecPath != "" {
		fmt.Fprintf(stdout, "internet_spec=%s\n", result.InternetSpecPath)
	}
	fmt.Fprintf(stdout, "manifest=%s\n", result.ManifestPath)
	fmt.Fprintf(stdout, "init_receipt=%s\n", result.InitReceiptPath)
	fmt.Fprintf(stdout, "runbook=%s\n", result.RunbookPath)
	fmt.Fprintf(stdout, "remote_bootstrap_plan=%s\n", result.RemoteBootstrapPlanPath)
	fmt.Fprintf(stdout, "remote_bootstrap_script=%s\n", result.RemoteBootstrapScriptPath)
	for _, path := range result.NodeBootstrapPaths {
		fmt.Fprintf(stdout, "node_bootstrap=%s\n", path)
	}
	if result.RemoteBootstrapReceiptPath != "" {
		fmt.Fprintf(stdout, "remote_bootstrap_receipt=%s\n", result.RemoteBootstrapReceiptPath)
	}
	return 0
}

func loadInternetInitIngress(spec contracts.ScenarioSpec, internetSpecPath, datasetIdentity, fromAttempt string) (*scenario.InternetInitIngress, error) {
	isInternet := spec.Scenario.Industry == contracts.IndustryInternet
	if !isInternet {
		if strings.TrimSpace(internetSpecPath) != "" || strings.TrimSpace(datasetIdentity) != "" {
			return nil, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageInit, "parse", "init.internet", "--internet-spec and --dataset-identity are valid only for Internet init", nil)
		}
		return nil, nil
	}
	if strings.TrimSpace(fromAttempt) != "" {
		if strings.TrimSpace(internetSpecPath) != "" || strings.TrimSpace(datasetIdentity) != "" {
			return nil, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageInit, "parse", "init.internet", "Internet --from-attempt reloads the effective spec and dataset identity and rejects fresh overrides", nil)
		}
		ingress, err := scenario.LoadAttemptInternetInitIngress(context.Background(), fromAttempt)
		if err != nil {
			return nil, err
		}
		return &ingress, nil
	}
	ingress, err := scenario.LoadFreshInternetInitIngress(internetSpecPath, datasetIdentity)
	if err != nil {
		return nil, err
	}
	return &ingress, nil
}

func internetSpecBytes(ingress *scenario.InternetInitIngress) []byte {
	if ingress == nil {
		return nil
	}
	return append([]byte(nil), ingress.RawSpec...)
}

func internetDatasetIdentity(ingress *scenario.InternetInitIngress) string {
	if ingress == nil {
		return ""
	}
	return ingress.DatasetIdentity
}

func runRerun(args []string, stdout, stderr io.Writer) int {
	return runRerunScriptCommand(context.Background(), rerunArgsWithDefaultTipoc(args), stdout, stderr)
}

func rerunArgsWithDefaultTipoc(args []string) []string {
	out := append([]string(nil), args...)
	for _, arg := range out {
		if arg == "--tipoc" || strings.HasPrefix(arg, "--tipoc=") {
			return out
		}
	}
	exe, err := os.Executable()
	if err != nil || strings.TrimSpace(exe) == "" {
		return out
	}
	return append(out, "--tipoc", exe)
}

func findRerunScriptPath() (string, error) {
	if path := strings.TrimSpace(os.Getenv(envRerunScriptPath)); path != "" {
		return requireExecutableFile(path)
	}
	const rel = "scripts/rerun/run-again.sh"
	if wd, err := os.Getwd(); err == nil {
		if path, ok := findAncestorFile(wd, rel); ok {
			return requireExecutableFile(path)
		}
	}
	if exe, err := os.Executable(); err == nil {
		if path, ok := findAncestorFile(filepath.Dir(exe), rel); ok {
			return requireExecutableFile(path)
		}
	}
	return "", fmt.Errorf("rerun wrapper not found; set %s or run from the repository containing %s", envRerunScriptPath, rel)
}

func findAncestorFile(start, rel string) (string, bool) {
	dir, err := filepath.Abs(start)
	if err != nil {
		return "", false
	}
	for {
		candidate := filepath.Join(dir, rel)
		if info, err := os.Stat(candidate); err == nil && !info.IsDir() {
			return candidate, true
		}
		parent := filepath.Dir(dir)
		if parent == dir {
			return "", false
		}
		dir = parent
	}
}

func requireExecutableFile(path string) (string, error) {
	abs, err := filepath.Abs(path)
	if err != nil {
		return "", err
	}
	info, err := os.Stat(abs)
	if err != nil {
		return "", err
	}
	if info.IsDir() {
		return "", fmt.Errorf("rerun wrapper path is a directory: %s", abs)
	}
	if info.Mode()&0o111 == 0 {
		return "", fmt.Errorf("rerun wrapper is not executable: %s", abs)
	}
	return abs, nil
}

func loadInitSpec(specPath, fromAttempt string) (contracts.ScenarioSpec, error) {
	if specPath != "" {
		return scenario.LoadSpecFile(specPath)
	}
	path := filepath.Join(fromAttempt, scenario.SpecFileName)
	spec, err := scenario.LoadSpecFile(path)
	if err != nil {
		return contracts.ScenarioSpec{}, err
	}
	spec.Init.AttemptRoot = ""
	return spec, nil
}

func loadWorkloadProfileOverride(path string) (contracts.WorkloadProfileOverride, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return contracts.WorkloadProfileOverride{}, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageRun, "load-workload-profile-override", "workload_profile_override", "read workload profile override file failed", err)
	}
	override, err := scenario.ParseWorkloadProfileOverride(data, filepath.Ext(path))
	if err != nil {
		return contracts.WorkloadProfileOverride{}, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageRun, "parse-workload-profile-override", "workload_profile_override", "parse workload profile override file failed", err)
	}
	return override, nil
}

func countNonEmpty(values ...string) int {
	count := 0
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			count++
		}
	}
	return count
}

func runGuardedStage(stage string, args []string, stdout, stderr io.Writer) int {
	if stage == scenario.PublicStageRun && len(args) > 0 && args[0] == tipocRunPreflight {
		return runRunPreflight(args[1:], stdout, stderr)
	}
	if stage == scenario.PublicStageRun && len(args) > 0 && args[0] == tipocRunRecover {
		return runInternetRerunRecovery(args[1:], stdout, stderr)
	}
	fs := flag.NewFlagSet(stage, flag.ContinueOnError)
	fs.SetOutput(stderr)
	manifestPath := fs.String("manifest", "", "path to run-profile manifest")
	prepareOps := fs.String("ops", "", "prepare operation list: generate,import,add-index,backup,restore")
	prepareFinish := fs.Bool("finish", false, "validate externally completed prepare import and indexes, then mark prepare complete")
	continuationFrom := fs.String("continuation-from", "", "prior terminal run-profile manifest used to validate a post-run prepare continuation")
	bucketRepair := fs.Bool("bucket-repair", false, "repair internet capacity bucket prepare data and write formal coverage evidence")
	restoreBackupURI := fs.String("restore-backup-uri", "", "prepare restore backup URI override")
	restoreBackupReceipt := fs.String("restore-backup-receipt", "", "prepare restore backup receipt JSON")
	restoreConfirm := fs.String("restore-confirm", "", "destructive restore confirmation token")
	workloadProfileOverridePath := fs.String("workload-profile-override", "", "path to a JSON or YAML workload profile override")
	operationRuntimeOverridesPath := ""
	operationRuntimeResolutionPath := ""
	if stage == scenario.PublicStageRun {
		fs.StringVar(&operationRuntimeOverridesPath, "operation-runtime-overrides", "", "path to typed operation runtime overrides JSON")
		fs.StringVar(&operationRuntimeResolutionPath, "operation-runtime-resolution", "", "path to a resolved operation runtime artifact")
	}
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if stage == scenario.PublicStagePrepare && *bucketRepair {
		switch {
		case *prepareOps != "":
			return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidPrepareOperation, scenario.PublicStagePrepare, "bucket-repair", "prepare.ops", "prepare --bucket-repair cannot be combined with --ops", nil))
		case *prepareFinish:
			return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidPrepareOperation, scenario.PublicStagePrepare, "bucket-repair", "prepare.finish", "prepare --bucket-repair cannot be combined with --finish", nil))
		case *restoreBackupURI != "" || *restoreBackupReceipt != "" || *restoreConfirm != "":
			return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidPrepareOperation, scenario.PublicStagePrepare, "bucket-repair", "prepare.restore", "prepare --bucket-repair cannot be combined with restore options", nil))
		case strings.TrimSpace(*continuationFrom) != "":
			return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidPrepareOperation, scenario.PublicStagePrepare, "bucket-repair", "prepare.continuation_from", "prepare --bucket-repair cannot be combined with --continuation-from", nil))
		}
	}
	if stage == scenario.PublicStagePrepare && strings.TrimSpace(*continuationFrom) != "" && !*prepareFinish {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidPrepareOperation, scenario.PublicStagePrepare, "continuation", "prepare.continuation_from", "prepare --continuation-from requires --finish", nil))
	}
	if stage != scenario.PublicStagePrepare && (*prepareOps != "" || *prepareFinish || *bucketRepair || *restoreBackupURI != "" || *restoreBackupReceipt != "" || *restoreConfirm != "" || strings.TrimSpace(*continuationFrom) != "") {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, stage, "parse", "prepare.options", fmt.Sprintf("%s does not accept prepare operation options", stage), nil))
	}
	if stage != scenario.PublicStageRun && strings.TrimSpace(*workloadProfileOverridePath) != "" {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, stage, "parse", "workload_profile_override", fmt.Sprintf("%s does not accept workload profile runtime options", stage), nil))
	}
	if *manifestPath == "" {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, stage, "parse", "manifest", fmt.Sprintf("%s requires --manifest", stage), nil))
	}

	if stage == scenario.PublicStageRun {
		if err := scenario.ValidateWorkloadProfileMaterialization(*manifestPath); err != nil {
			return writeError(stderr, err)
		}
	}
	manifest, err := scenario.LoadManifest(*manifestPath)
	if err != nil {
		return writeError(stderr, err)
	}
	runOptions := execution.RunOptions{}
	workloadProfileEnvelope := ""
	if stage == scenario.PublicStageRun {
		if strings.TrimSpace(*workloadProfileOverridePath) != "" {
			override, overrideErr := loadWorkloadProfileOverride(*workloadProfileOverridePath)
			if overrideErr != nil {
				return writeError(stderr, overrideErr)
			}
			runOptions.WorkloadProfile = &override
			_, workloadProfileEnvelope, err = execution.BuildWorkloadProfileResolutionEnvelope(manifest, &override)
			if err != nil {
				return writeError(stderr, err)
			}
		}
		if strings.TrimSpace(operationRuntimeOverridesPath) != "" {
			overrides, overrideErr := execution.LoadOperationRuntimeOverrides(operationRuntimeOverridesPath)
			if overrideErr != nil {
				return writeError(stderr, overrideErr)
			}
			runOptions.OperationRuntimeOverrides = overrides
		}
		runOptions.OperationRuntimeResolutionPath = strings.TrimSpace(operationRuntimeResolutionPath)
		if err := runOptions.ValidateOperationRuntimeInputs(); err != nil {
			return writeError(stderr, err)
		}
	}
	if delegated, code := maybeDelegateStage(stage, *manifestPath, manifest, prepareCLIOptions{
		ops:                         *prepareOps,
		finish:                      *prepareFinish,
		continuationFrom:            *continuationFrom,
		bucketRepair:                *bucketRepair,
		restoreBackupURI:            *restoreBackupURI,
		restoreBackupReceipt:        *restoreBackupReceipt,
		restoreConfirm:              *restoreConfirm,
		workloadProfileOverridePath: *workloadProfileOverridePath,
		workloadProfileEnvelope:     workloadProfileEnvelope,
		operationRuntimeOverrides:   operationRuntimeOverridesPath,
		operationRuntimeResolution:  operationRuntimeResolutionPath,
	}, stdout, stderr); delegated {
		return code
	}

	wiring, err := newApplicationWiring()
	if err != nil {
		return writeError(stderr, err)
	}
	executor := wiring.executor()
	executor.ProgressWriter = stderr
	var (
		result execution.StageResult
	)
	switch stage {
	case scenario.PublicStagePrepare:
		options := execution.PrepareOptions{
			OpsCSV:               *prepareOps,
			RestoreBackupURI:     *restoreBackupURI,
			RestoreBackupReceipt: *restoreBackupReceipt,
			RestoreConfirm:       *restoreConfirm,
			ContinuationFrom:     *continuationFrom,
			Finish:               *prepareFinish,
			BucketRepair:         *bucketRepair,
		}
		result, err = executor.PrepareWithOptions(context.Background(), *manifestPath, options)
	case scenario.PublicStageRun:
		result, err = executor.RunWithOptions(context.Background(), *manifestPath, runOptions)
	case scenario.PublicStageCleanup:
		result, err = executor.Cleanup(context.Background(), *manifestPath)
	default:
		err = scenario.ValidatePublicStage(stage)
	}
	if err != nil {
		return writeError(stderr, err)
	}
	fmt.Fprintf(stdout, "stage=%s status=%s manifest=%s\n", result.Stage, result.Status, result.ManifestPath)
	for _, artifact := range visibleStageArtifacts(stage, result.Artifacts) {
		fmt.Fprintf(stdout, "artifact=%s\n", artifact)
	}
	if summary := runSummaryArtifactPath(stage, result.Artifacts); summary != "" {
		fmt.Fprintf(stdout, "run_summary=%s\n", summary)
	}
	if result.Message != "" {
		fmt.Fprintf(stdout, "message=%s\n", result.Message)
	}
	return 0
}

func runRunPreflight(args []string, stdout, stderr io.Writer) int {
	fs := flag.NewFlagSet("run preflight", flag.ContinueOnError)
	fs.SetOutput(stderr)
	manifestPath := fs.String("manifest", "", "path to run-profile manifest")
	workloadProfileOverridePath := fs.String("workload-profile-override", "", "path to a JSON or YAML workload profile override")
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if fs.NArg() != 0 {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageRun, "parse", "run.preflight.args", "run preflight accepts only flags", nil))
	}
	if *manifestPath == "" {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageRun, "parse", "manifest", "run preflight requires --manifest", nil))
	}
	options := execution.RunOptions{}
	if strings.TrimSpace(*workloadProfileOverridePath) != "" {
		override, err := loadWorkloadProfileOverride(*workloadProfileOverridePath)
		if err != nil {
			return writeError(stderr, err)
		}
		options.WorkloadProfile = &override
	}
	wiring, err := newApplicationWiring()
	if err != nil {
		return writeError(stderr, err)
	}
	executor := wiring.executor()
	result, err := executor.RunPreflightWithOptions(context.Background(), *manifestPath, options)
	if err != nil {
		return writeError(stderr, err)
	}
	fmt.Fprintf(stdout, "stage=%s status=%s manifest=%s\n", result.Stage, result.Status, result.ManifestPath)
	for _, artifact := range result.Artifacts {
		fmt.Fprintf(stdout, "artifact=%s\n", artifact)
	}
	if receipt := runPreflightArtifactPath(result.Artifacts); receipt != "" {
		fmt.Fprintf(stdout, "preflight_receipt=%s\n", receipt)
	}
	if result.Message != "" {
		fmt.Fprintf(stdout, "message=%s\n", result.Message)
	}
	return 0
}

// runInternetRerunRecovery is an internal Run capability used by run-again.
// It retains the four public stages: it only writes pre-run recovery evidence
// and must finish before the later ordinary `tipoc run` starts measurement.
func runInternetRerunRecovery(args []string, stdout, stderr io.Writer) int {
	fs := flag.NewFlagSet("run recover", flag.ContinueOnError)
	fs.SetOutput(stderr)
	manifestPath := fs.String("manifest", "", "path to rerun run-profile manifest")
	if err := fs.Parse(args); err != nil {
		return 2
	}
	if fs.NArg() != 0 || strings.TrimSpace(*manifestPath) == "" {
		return writeError(stderr, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageRun,
			"parse", "manifest", "run recover requires --manifest", nil))
	}
	wiring, err := newApplicationWiring()
	if err != nil {
		return writeError(stderr, err)
	}
	result, err := wiring.executor().RecoverInternetRerunState(context.Background(), *manifestPath)
	if err != nil {
		return writeError(stderr, err)
	}
	fmt.Fprintf(stdout, "stage=%s status=%s manifest=%s\n", result.Stage, result.Status, result.ManifestPath)
	for _, artifact := range result.Artifacts {
		fmt.Fprintf(stdout, "artifact=%s\n", artifact)
	}
	if result.Message != "" {
		fmt.Fprintf(stdout, "message=%s\n", result.Message)
	}
	return 0
}

type prepareCLIOptions struct {
	ops                         string
	finish                      bool
	continuationFrom            string
	bucketRepair                bool
	restoreBackupURI            string
	restoreBackupReceipt        string
	restoreConfirm              string
	workloadProfileOverridePath string
	workloadProfileEnvelope     string
	operationRuntimeOverrides   string
	operationRuntimeResolution  string
}

type bigQueryCLIFlags struct {
	enabled        string
	startAfter     int64
	startAfterSet  bool
	concurrency    int
	concurrencySet bool
	queryIDs       string
}

func bindBigQueryFlags(fs *flag.FlagSet) *bigQueryCLIFlags {
	flags := &bigQueryCLIFlags{}
	fs.StringVar(&flags.enabled, "big-query-enabled", "", "enable Big Query during init: true or false")
	fs.Func("big-query-start-after-foreground-ready-seconds", "seconds to wait after foreground_ready before big queries start", func(value string) error {
		parsed, err := strconv.ParseInt(value, 10, 64)
		if err != nil {
			return err
		}
		flags.startAfter = parsed
		flags.startAfterSet = true
		return nil
	})
	fs.Func("big-query-concurrency", "big query worker concurrency, from 1 to 10", func(value string) error {
		parsed, err := strconv.Atoi(value)
		if err != nil {
			return err
		}
		flags.concurrency = parsed
		flags.concurrencySet = true
		return nil
	})
	fs.StringVar(&flags.queryIDs, "big-query-queries", "", "comma-separated report ids, for example report_01,report_06,report_09")
	return flags
}

func (f bigQueryCLIFlags) anySet() bool {
	return strings.TrimSpace(f.enabled) != "" || f.startAfterSet || f.concurrencySet || strings.TrimSpace(f.queryIDs) != ""
}

func (f bigQueryCLIFlags) queryIDList() []string {
	if strings.TrimSpace(f.queryIDs) == "" {
		return nil
	}
	ids := strings.Split(f.queryIDs, ",")
	out := make([]string, 0, len(ids))
	for _, id := range ids {
		if trimmed := strings.TrimSpace(id); trimmed != "" {
			out = append(out, trimmed)
		}
	}
	return out
}

func (f bigQueryCLIFlags) enabledValue() (*bool, error) {
	if strings.TrimSpace(f.enabled) == "" {
		return nil, nil
	}
	value, err := strconv.ParseBool(strings.TrimSpace(f.enabled))
	if err != nil {
		return nil, contracts.NewProductError(contracts.ErrorCodeInvalidStageInput, scenario.PublicStageInit, "parse", "big_query_interference.enabled", "--big-query-enabled must be true or false", err)
	}
	return &value, nil
}

func applyBigQueryInitFlags(spec *contracts.ScenarioSpec, flags *bigQueryCLIFlags) error {
	if flags == nil || !flags.anySet() {
		return nil
	}
	for _, operation := range spec.Operations {
		if strings.TrimSpace(operation.ID) == "big_query" || strings.TrimSpace(operation.ID) == "securities/tpce/big_query" {
			return contracts.NewProductError(contracts.ErrorCodeInvalidOperationConfig, scenario.PublicStageInit, "parse", "operations", "init Big Query flags cannot be combined with canonical operations[]", nil)
		}
	}
	enabled, err := flags.enabledValue()
	if err != nil {
		return err
	}
	config := &spec.Pressure.BigQueryInterference
	if enabled != nil {
		config.Enabled = *enabled
	}
	if flags.startAfterSet {
		config.StartAfterForegroundReadySeconds = flags.startAfter
	}
	if flags.concurrencySet {
		config.Concurrency = flags.concurrency
	}
	if strings.TrimSpace(flags.queryIDs) != "" {
		config.QueryIDs = flags.queryIDList()
	}
	return nil
}

func maybeDelegateStage(stage, manifestPath string, manifest contracts.RunProfileManifest, prepareOptions prepareCLIOptions, stdout, stderr io.Writer) (bool, int) {
	if internalStageWorker(stage, prepareOptions) {
		return false, 0
	}
	currentNodeID := strings.TrimSpace(os.Getenv("TIPOC_NODE_ID"))
	coordinatorNodeID := strings.TrimSpace(manifest.Init.CoordinatorNodeID)
	if currentNodeID == "" || coordinatorNodeID == "" || currentNodeID == coordinatorNodeID {
		return false, 0
	}
	if stage == scenario.PublicStageRun &&
		strings.TrimSpace(prepareOptions.workloadProfileOverridePath) != "" &&
		strings.TrimSpace(prepareOptions.workloadProfileEnvelope) == "" {
		err := contracts.NewProductError(
			contracts.ErrorCodeCoordinatorDelegationFailed,
			stage,
			"delegate-workload-profile-override",
			"workload_profile_override",
			"workload profile override could not be converted to the typed coordinator transport",
			nil,
		)
		return true, writeError(stderr, err)
	}
	code, err := delegateStageToCoordinator(context.Background(), stage, manifestPath, manifest, prepareOptions, stdout)
	if err != nil {
		return true, writeError(stderr, err)
	}
	return true, code
}

func internalStageWorker(stage string, prepareOptions prepareCLIOptions) bool {
	switch stage {
	case scenario.PublicStagePrepare:
		if !prepareWorkerMayRunLocally(prepareOptions) {
			return false
		}
		if strings.TrimSpace(os.Getenv(envInternalPrepareGenerationWorker)) != "" {
			return true
		}
		if strings.TrimSpace(os.Getenv(envPrepareGenerationResult)) != "" && strings.TrimSpace(os.Getenv(envPrepareOwnerPrefixes)) != "" {
			return true
		}
		return strings.TrimSpace(os.Getenv(envPrepareGenerationNodeID)) != "" &&
			(strings.TrimSpace(os.Getenv(envPrepareGenerationEndpoint)) != "" || strings.TrimSpace(os.Getenv(envPrepareGenerationListen)) != "")
	case scenario.PublicStageRun:
		return strings.TrimSpace(os.Getenv(envRunPressureNodeID)) != "" &&
			(strings.TrimSpace(os.Getenv(envRunPressureEndpoint)) != "" || strings.TrimSpace(os.Getenv(envRunPressureListen)) != "")
	default:
		return false
	}
}

func prepareWorkerMayRunLocally(options prepareCLIOptions) bool {
	// Non-coordinator nodes may only execute explicit generation slices; full
	// prepare, finish, restore, import, index, and backup stay coordinator-owned.
	if options.bucketRepair {
		return prepareDistributedGenerationEnvPresent()
	}
	if options.finish ||
		strings.TrimSpace(options.restoreBackupURI) != "" ||
		strings.TrimSpace(options.restoreBackupReceipt) != "" ||
		strings.TrimSpace(options.restoreConfirm) != "" {
		return false
	}
	ops, err := scenario.ParsePrepareOps(options.ops)
	if err != nil {
		return false
	}
	return len(ops) == 1 && ops[0] == contracts.PrepareOpGenerate
}

func prepareDistributedGenerationEnvPresent() bool {
	return strings.TrimSpace(os.Getenv(envPrepareGenerationNodeID)) != "" &&
		(strings.TrimSpace(os.Getenv(envPrepareGenerationEndpoint)) != "" ||
			strings.TrimSpace(os.Getenv(envPrepareGenerationListen)) != "")
}

func delegateStageToCoordinator(ctx context.Context, stage, manifestPath string, manifest contracts.RunProfileManifest, prepareOptions prepareCLIOptions, stdout io.Writer) (int, error) {
	target := sshTarget(manifest.Init.SSHUser, manifest.Init.CoordinatorAddress)
	if strings.TrimSpace(target) == "" {
		return 0, contracts.NewProductError(contracts.ErrorCodeCoordinatorDelegationFailed, stage, "missing_coordinator_address", "init.coordinator_address", fmt.Sprintf("node %s cannot delegate %s because coordinator address is empty", os.Getenv("TIPOC_NODE_ID"), stage), nil)
	}
	remoteManifestPath := filepath.ToSlash(filepath.Join(firstNonEmpty(manifest.Init.RemoteAttemptRoot, filepath.Dir(manifestPath)), scenario.ManifestFileName))
	logDir := filepath.ToSlash(firstNonEmpty(manifest.Init.CoordinatorLogDir, filepath.Join(firstNonEmpty(manifest.Init.RemoteAttemptRoot, filepath.Dir(manifestPath)), "logs", "coordinator")))
	stageLog := filepath.ToSlash(filepath.Join(logDir, stage+".log"))
	stageExit := filepath.ToSlash(filepath.Join(logDir, stage+".exit"))
	if strings.TrimSpace(prepareOptions.continuationFrom) != "" {
		remoteContinuation, err := coordinatorContinuationManifestPath(prepareOptions.continuationFrom)
		if err != nil {
			return 0, contracts.NewProductError(contracts.ErrorCodeCoordinatorDelegationFailed, stage, "continuation_path", "prepare.continuation_from", "resolve prior continuation manifest on coordinator", err)
		}
		prepareOptions.continuationFrom = remoteContinuation
	}
	remoteArgs := stageArgs(stage, remoteManifestPath, prepareOptions)
	remoteCommand := shellJoin(remoteArgs)
	if strings.TrimSpace(prepareOptions.workloadProfileEnvelope) != "" {
		remoteCommand = execution.WorkloadProfileResolutionEnvelopeEnv + "=" + shellQuote(prepareOptions.workloadProfileEnvelope) + " " + remoteCommand
	}
	// The remote stage writes its own exit file; the SSH start command must not
	// replace that with the launch exit status.
	remoteRun := fmt.Sprintf("%s > %s 2>&1; echo $? > %s", remoteCommand, shellQuote(stageLog), shellQuote(stageExit))
	remoteStart := fmt.Sprintf("mkdir -p %s && rm -f %s && nohup %s > /dev/null 2>&1 &",
		shellQuote(logDir),
		shellQuote(stageExit),
		shellJoin([]string{"sh", "-c", remoteRun}),
	)
	if output, err := runExternalStageCommand(ctx, []string{"ssh", target, remoteStart}); err != nil {
		return 0, contracts.NewProductError(contracts.ErrorCodeCoordinatorDelegationFailed, stage, "start_failed", "coordinator", fmt.Sprintf("delegate %s start on coordinator node %s failed", stage, manifest.Init.CoordinatorNodeID), fmt.Errorf("%w: %s", err, strings.TrimSpace(string(output))))
	}
	fmt.Fprintf(stdout, "delegated %s from node %s to coordinator %s\n", stage, os.Getenv("TIPOC_NODE_ID"), manifest.Init.CoordinatorNodeID)
	pollInterval := time.Duration(manifest.Init.StatusPollIntervalSecs) * time.Second
	if pollInterval <= 0 {
		pollInterval = 30 * time.Second
	}
	for {
		if output, err := runExternalStageCommand(ctx, []string{"ssh", target, fmt.Sprintf("test -f %s && tail -n 40 %s || true", shellQuote(stageLog), shellQuote(stageLog))}); err == nil && len(output) > 0 {
			writeSafeCoordinatorTail(stdout, string(output))
		}
		if _, err := runExternalStageCommand(ctx, []string{"ssh", target, fmt.Sprintf("test -f %s", shellQuote(stageExit))}); err == nil {
			output, err := runExternalStageCommand(ctx, []string{"ssh", target, fmt.Sprintf("cat %s", shellQuote(stageExit))})
			if err != nil {
				return 0, contracts.NewProductError(contracts.ErrorCodeCoordinatorDelegationFailed, stage, "exit_status_read_failed", "coordinator", fmt.Sprintf("read %s exit status from coordinator node %s failed", stage, manifest.Init.CoordinatorNodeID), err)
			}
			exitCode, err := strconv.Atoi(strings.TrimSpace(string(output)))
			if err != nil {
				return 0, contracts.NewProductError(contracts.ErrorCodeCoordinatorDelegationFailed, stage, "exit_status_parse_failed", "coordinator", fmt.Sprintf("parse %s coordinator exit status failed", stage), fmt.Errorf("%w: %s", err, strings.TrimSpace(string(output))))
			}
			return exitCode, nil
		}
		timer := time.NewTimer(pollInterval)
		select {
		case <-ctx.Done():
			timer.Stop()
			return 0, contracts.NewProductError(contracts.ErrorCodeCoordinatorDelegationFailed, stage, "context_cancelled", "coordinator", fmt.Sprintf("delegate %s wait was interrupted", stage), ctx.Err())
		case <-timer.C:
		}
	}
}

func coordinatorContinuationManifestPath(path string) (string, error) {
	path = strings.TrimSpace(path)
	prior, err := scenario.LoadManifest(path)
	if err != nil {
		return "", err
	}
	abs, err := filepath.Abs(path)
	if err != nil {
		return "", err
	}
	remoteRoot := strings.TrimSpace(prior.Init.RemoteAttemptRoot)
	if remoteRoot == "" {
		return filepath.ToSlash(abs), nil
	}
	remoteAbs, err := filepath.Abs(remoteRoot)
	if err == nil {
		if rel, relErr := filepath.Rel(remoteAbs, abs); relErr == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
			return filepath.ToSlash(abs), nil
		}
	}
	localRoots := []string{strings.TrimSpace(prior.Init.LocalAttemptRoot), strings.TrimSpace(prior.EvidenceRoot)}
	for _, localRoot := range localRoots {
		if localRoot == "" {
			continue
		}
		localAbs, err := filepath.Abs(localRoot)
		if err != nil {
			continue
		}
		rel, err := filepath.Rel(localAbs, abs)
		if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
			continue
		}
		return filepath.ToSlash(filepath.Join(remoteRoot, rel)), nil
	}
	return "", fmt.Errorf("prior manifest %s is not within its local attempt root", path)
}

var (
	safeCoordinatorProgressKeyPattern     = regexp.MustCompile(`^[A-Za-z0-9_.:-]+$`)
	safeCoordinatorProgressValuePattern   = regexp.MustCompile(`^[A-Za-z0-9_.:,%+-]+$`)
	safeCoordinatorGenerationValuePattern = regexp.MustCompile(`^[A-Za-z0-9_.:/,%+-]+$`)
	safePrepareProgressKeys               = map[string]struct{}{
		"operation":   {},
		"phase":       {},
		"status":      {},
		"table":       {},
		"updated_at":  {},
		"started_at":  {},
		"finished_at": {},
		"detail":      {},
		"completed":   {},
		"total":       {},
	}
	safeGenerationProgressKeys = map[string]struct{}{
		"node":                      {},
		"event":                     {},
		"assigned_load_units":       {},
		"completed_objects":         {},
		"completed_bytes":           {},
		"completed_gib":             {},
		"expected_bytes_estimate":   {},
		"expected_gib_estimate":     {},
		"remaining_bytes_estimate":  {},
		"remaining_gib_estimate":    {},
		"progress_percent_estimate": {},
		"throughput_mib_per_sec":    {},
		"elapsed":                   {},
		"eta":                       {},
		"last_table":                {},
		"last_object":               {},
	}
)

func writeSafeCoordinatorTail(stdout io.Writer, output string) {
	for _, line := range strings.Split(output, "\n") {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		if safeCoordinatorProgressLineSafe(line) {
			fmt.Fprintln(stdout, line)
		}
	}
}

func safeCoordinatorProgressLineSafe(line string) bool {
	fields := strings.Fields(line)
	if len(fields) < 2 {
		return false
	}
	prefix := fields[0]
	if prefix != "prepare_progress" && prefix != "generation-progress" {
		return false
	}
	for _, field := range fields[1:] {
		key, value, ok := strings.Cut(field, "=")
		if !ok || key == "" || !safeCoordinatorProgressKeyPattern.MatchString(key) || !safeCoordinatorProgressKeyAllowed(prefix, key) {
			return false
		}
		if value == "" {
			continue
		}
		if strings.Contains(value, "://") {
			return false
		}
		valuePattern := safeCoordinatorProgressValuePattern
		if prefix == "generation-progress" {
			valuePattern = safeCoordinatorGenerationValuePattern
		}
		if !valuePattern.MatchString(value) {
			return false
		}
	}
	return true
}

func safeCoordinatorProgressKeyAllowed(prefix, key string) bool {
	switch prefix {
	case "prepare_progress":
		_, ok := safePrepareProgressKeys[key]
		return ok
	case "generation-progress":
		_, ok := safeGenerationProgressKeys[key]
		return ok
	default:
		return false
	}
}

func visibleStageArtifacts(stage string, artifacts []string) []string {
	if stage != scenario.PublicStagePrepare {
		return artifacts
	}
	allowed := map[string]struct{}{
		"prepare-progress.json":               {},
		"prepare-progress.jsonl":              {},
		"prepare-receipt.json":                {},
		"prepare-completion-validation.json":  {},
		"prepare-finish-validation.json":      {},
		"post-run-continuation-baseline.json": {},
		"statistics-validation-receipt.json":  {},
		"br-checkpoint-receipt.json":          {},
		"stage-log.jsonl":                     {},
	}
	out := make([]string, 0, len(artifacts))
	for _, artifact := range artifacts {
		if _, ok := allowed[filepath.Base(artifact)]; ok {
			out = append(out, artifact)
		}
	}
	return out
}

func runSummaryArtifactPath(stage string, artifacts []string) string {
	if stage != scenario.PublicStageRun {
		return ""
	}
	for _, artifact := range artifacts {
		if filepath.Base(artifact) == "summary.json" {
			return artifact
		}
	}
	return ""
}

func runPreflightArtifactPath(artifacts []string) string {
	for _, artifact := range artifacts {
		if filepath.Base(artifact) == "preflight-receipt.json" {
			return artifact
		}
	}
	return ""
}

func stageArgs(stage, manifestPath string, prepareOptions prepareCLIOptions) []string {
	args := []string{"tipoc", stage, "--manifest", manifestPath}
	if stage != scenario.PublicStagePrepare {
		if strings.TrimSpace(prepareOptions.operationRuntimeOverrides) != "" {
			args = append(args, "--operation-runtime-overrides", prepareOptions.operationRuntimeOverrides)
		}
		if strings.TrimSpace(prepareOptions.operationRuntimeResolution) != "" {
			args = append(args, "--operation-runtime-resolution", prepareOptions.operationRuntimeResolution)
		}
		return args
	}
	if strings.TrimSpace(prepareOptions.ops) != "" {
		args = append(args, "--ops", prepareOptions.ops)
	}
	if prepareOptions.finish {
		args = append(args, "--finish")
	}
	if strings.TrimSpace(prepareOptions.continuationFrom) != "" {
		args = append(args, "--continuation-from", prepareOptions.continuationFrom)
	}
	if prepareOptions.bucketRepair {
		args = append(args, "--bucket-repair")
	}
	if strings.TrimSpace(prepareOptions.restoreBackupURI) != "" {
		args = append(args, "--restore-backup-uri", prepareOptions.restoreBackupURI)
	}
	if strings.TrimSpace(prepareOptions.restoreBackupReceipt) != "" {
		args = append(args, "--restore-backup-receipt", prepareOptions.restoreBackupReceipt)
	}
	if strings.TrimSpace(prepareOptions.restoreConfirm) != "" {
		args = append(args, "--restore-confirm", prepareOptions.restoreConfirm)
	}
	return args
}

func sshTarget(user, address string) string {
	address = strings.TrimSpace(address)
	user = strings.TrimSpace(user)
	if user == "" || address == "" {
		return address
	}
	return user + "@" + address
}

func shellJoin(args []string) string {
	quoted := make([]string, 0, len(args))
	for _, arg := range args {
		quoted = append(quoted, shellQuote(arg))
	}
	return strings.Join(quoted, " ")
}

func shellQuote(value string) string {
	if value == "" {
		return "''"
	}
	return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'"
}

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

func writeError(stderr io.Writer, err error) int {
	var coded contracts.CodedError
	if errors.As(err, &coded) {
		code := coded.ProductErrorCode()
		fmt.Fprintf(stderr, "error_code=%s %s\n", code, coded.Error())
		return exitCodeForProductError(code)
	}
	fmt.Fprintf(stderr, "error_code=unclassified_error reason=%s\n", contracts.SanitizeErrorDetail(err.Error()))
	return 1
}

func exitCodeForProductError(code string) int {
	switch code {
	case contracts.ErrorCodeUnknownStage, contracts.ErrorCodeInvalidStageInput, contracts.ErrorCodeInvalidPrepareOperation:
		return 2
	case contracts.ErrorCodeUnsupportedPrepareSource:
		return 3
	case contracts.ErrorCodeIllegalStageTransition:
		return 4
	case contracts.ErrorCodeStageNotImplemented:
		return 5
	case contracts.ErrorCodeSpecInput, contracts.ErrorCodeManifestInput, contracts.ErrorCodeInvalidScenarioSpec:
		return 6
	case contracts.ErrorCodeCoordinatorDelegationFailed:
		return 7
	case contracts.ErrorCodePreparePrerequisiteMissing, contracts.ErrorCodeRuntimePrerequisiteMissing:
		return 8
	case contracts.ErrorCodePrepareCompletionValidationFailed:
		return 9
	case contracts.ErrorCodeDistributedRuntimeCoordinationFailed:
		return 10
	case contracts.ErrorCodeExternalCommandFailed:
		return 11
	default:
		return 1
	}
}

func printUsage(w io.Writer) {
	fmt.Fprintln(w, "usage: tipoc <init|prepare|run|cleanup|rerun|ops|ru-acceptance> [options]")
}