package migration

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/observability/vm-migration/internal/exitcode"
	"github.com/tidbcloud/observability/vm-migration/internal/jsonio"
	"github.com/tidbcloud/observability/vm-migration/internal/retry"
	"github.com/tidbcloud/observability/vm-migration/internal/schema"
	"github.com/tidbcloud/observability/vm-migration/internal/status"
)

type loopVmctlRunner struct {
	calls int
}

func (r *loopVmctlRunner) Run(_ context.Context, _ []string, _ io.Writer) (int, int64) {
	r.calls++
	return 0, 42
}

func writeLoopPlan(t *testing.T, dir string, cmds []schema.MigrationLoopCmd) string {
	t.Helper()
	for i := range cmds {
		if cmds[i].Action == nil {
			cmds[i].Action = testAction(cmds[i].Step)
		}
	}
	return writeLoopPlanTenants(t, dir, []schema.MigrationLoopTenant{{
		Name:        "tenants-1",
		TenantsFile: "tenants-1.json",
		StatusFile:  "tenants-1-status.json",
		LogFile:     "tenants-1.log",
		Cmds:        cmds,
	}})
}

// contractCmds returns the full three-step command list every migration-loop
// artifact must carry, with default test actions attached.
func contractCmds() []schema.MigrationLoopCmd {
	steps := []string{"manifest", "migrate", "verify"}
	cmds := make([]schema.MigrationLoopCmd, 0, len(steps))
	for _, s := range steps {
		cmds = append(cmds, schema.MigrationLoopCmd{Step: s, Action: testAction(s)})
	}
	return cmds
}

// writeLoopPlanTenants writes a migration-loop.json with the given tenants
// under the standard runtime/vmselect scaffolding.
func writeLoopPlanTenants(t *testing.T, dir string, tenants []schema.MigrationLoopTenant) string {
	t.Helper()
	plan := schema.MigrationLoopPlan{
		Runtime: &schema.MigrationLoopRuntime{
			Region:    "prod",
			Vmcluster: "vmcluster",
			Vminsert:  "http://dst/{tenant}",
			Window: schema.MigrationLoopWindow{
				Start:        "2026-01-01T00:00:00Z",
				End:          "2026-01-02T00:00:00Z",
				StepInterval: "1d",
			},
			Verification: schema.MigrationLoopVerification{TimeoutSeconds: 60, Tolerance: 0.01},
		},
		Vmselect: &schema.MigrationLoopVmselect{
			Origin:    "http://origin/{tenant}",
			Migration: "http://migration/{tenant}",
			Candidate: "http://candidate/{tenant}",
		},
		Tenants: tenants,
	}
	data, err := json.MarshalIndent(plan, "", "  ")
	require.NoError(t, err)
	path := filepath.Join(dir, "migration-loop.json")
	require.NoError(t, os.WriteFile(path, append(data, '\n'), 0o644))
	return path
}

func testAction(step string) *schema.MigrationLoopAction {
	switch step {
	case "manifest":
		return &schema.MigrationLoopAction{Kind: schema.ActionManifest, TenantsFile: "tenants-1.json", Manifest: "tenants-1-manifest.json"}
	case "migrate":
		return &schema.MigrationLoopAction{Kind: schema.ActionMigrate, Manifest: "tenants-1-manifest.json", StepInterval: "day"}
	case "verify":
		return &schema.MigrationLoopAction{Kind: schema.ActionVerify, Manifest: "tenants-1-manifest.json"}
	default:
		return &schema.MigrationLoopAction{Kind: schema.MigrationLoopActionKind(step)}
	}
}

func succeedActions(_ context.Context, _ *schema.MigrationLoopPlan, _ string, action *schema.MigrationLoopAction, _ VmctlRunner, output ActionOutput) error {
	_, err := fmt.Fprintf(output.Log, "inline action=%s\n", action.Kind)
	return err
}

func combinedLoopOutput(w io.Writer) LoopOutput {
	return LoopOutput{Result: w, Progress: w, ActionLogObserver: ObserveWith(w)}
}

// runLoopForTest builds a Loop, combines every output channel for assertions,
// and runs one pass.
func runLoopForTest(t *testing.T, loopPath string) (*bytes.Buffer, error) {
	t.Helper()
	var stdout bytes.Buffer
	loop, err := NewLoop(loopPath, combinedLoopOutput(&stdout))
	if err != nil {
		return &stdout, err
	}
	loop.runAction = succeedActions
	return &stdout, loop.Run(context.Background())
}

func TestReconcileDoesNotCompleteFromTenantSummaryAlone(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())

	writeTenantStatusEvidence(t, dir, "tenants-1-status.json", 3, 0, 3)

	absPlan, err := filepath.Abs(loopPath)
	require.NoError(t, err)
	statusPath2 := DeriveLoopStatusPath(absPlan)
	plan, existing, err := loadInputs(absPlan, statusPath2)
	require.NoError(t, err)
	got, err := reconcileStatus(plan, existing, dir)
	require.NoError(t, err)

	require.Len(t, got.Tenants, 1)
	tenant := got.Tenants[0]
	assert.False(t, tenant.Completed, "tenant summary must not complete loop steps")
	assert.Equal(t, "manifest", tenant.CurrentStep)
	require.Len(t, tenant.Steps, 3)
	assert.Equal(t, StatusPending, tenant.Steps[0].Status)
}

func TestReconcileDoesNotOverrideFailedStepWithCompletedSummary(t *testing.T) {
	dir := t.TempDir()
	plan := &schema.MigrationLoopPlan{Tenants: []schema.MigrationLoopTenant{{
		Name:       "tenants-1",
		StatusFile: "tenants-1-status.json",
		Cmds: []schema.MigrationLoopCmd{{
			Step:   "migrate",
			Action: testAction("migrate"),
		}},
	}}}
	existing := &schema.MigrationLoopStatus{Tenants: []schema.MigrationLoopTenantStatus{{
		Name:      "tenants-1",
		Completed: true,
		Failed:    true,
		Steps: []schema.MigrationLoopStepStatus{{
			Step:   "migrate",
			Status: StatusFailed,
		}},
	}}}
	writeTenantStatusEvidence(t, dir, "tenants-1-status.json", 1, 1, 1)

	reconciled, err := reconcileStatus(plan, existing, dir)
	require.NoError(t, err)
	got := reconciled.Tenants[0]
	assert.False(t, got.Completed, "explicit failed step must win over completed shard summary")
	assert.Equal(t, "migrate", got.CurrentStep)
}

func TestReconcileDoesNotCompleteOnEmptySummary(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())
	statusPath := filepath.Join(dir, "tenants-1-status.json")
	require.NoError(t, os.WriteFile(statusPath, []byte(`{"summary":{}}`), 0o644))
	absPlan, err := filepath.Abs(loopPath)
	require.NoError(t, err)
	statusPath2 := DeriveLoopStatusPath(absPlan)
	plan, existing, err := loadInputs(absPlan, statusPath2)
	require.NoError(t, err)
	got, err := reconcileStatus(plan, existing, dir)
	require.NoError(t, err)

	tenant := got.Tenants[0]
	assert.False(t, tenant.Completed, "empty summary should not complete")
	assert.Equal(t, "manifest", tenant.CurrentStep)
}

func TestReconcilePreservesTenantFinishedAt(t *testing.T) {
	dir := t.TempDir()
	plan := &schema.MigrationLoopPlan{Tenants: []schema.MigrationLoopTenant{{
		Name:       "tenants-1",
		StatusFile: "tenants-1-status.json",
		Cmds:       contractCmds(),
	}}}
	started := time.Date(2026, 8, 27, 10, 0, 0, 0, time.UTC)
	finished := started.Add(time.Hour)
	existing := &schema.MigrationLoopStatus{Tenants: []schema.MigrationLoopTenantStatus{{
		Name:        "tenants-1",
		Completed:   true,
		CurrentStep: StepDone,
		StartedAt:   &started,
		FinishedAt:  &finished,
		Steps: []schema.MigrationLoopStepStatus{
			{Step: "manifest", Status: StatusCompleted},
			{Step: "migrate", Status: StatusCompleted},
			{Step: "verify", Status: StatusCompleted},
		},
	}}}
	writeTenantStatusEvidence(t, dir, "tenants-1-status.json", 1, 1, 1)

	reconciled, err := reconcileStatus(plan, existing, dir)
	require.NoError(t, err)
	got := reconciled.Tenants[0]
	assert.True(t, got.Completed)
	require.NotNil(t, got.FinishedAt)
	assert.True(t, got.FinishedAt.Equal(finished))
}

func TestReconcileReopensCompletedStepsWhenShardSummaryHasGap(t *testing.T) {
	cases := []struct {
		name        string
		secondPhase schema.Phase
		wantCurrent string
	}{
		{name: "migration gap", secondPhase: schema.PhasePending, wantCurrent: "migrate"},
		{name: "verification gap", secondPhase: schema.PhaseImported, wantCurrent: "verify"},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			dir := t.TempDir()
			finished := time.Date(2026, 8, 27, 11, 0, 0, 0, time.UTC)
			plan := &schema.MigrationLoopPlan{Tenants: []schema.MigrationLoopTenant{{
				Name:       "tenants-1",
				StatusFile: "tenants-1-status.json",
				Cmds: []schema.MigrationLoopCmd{
					{Step: "manifest"},
					{Step: "migrate", Action: testAction("migrate")},
					{Step: "verify"},
				},
			}}}
			existing := &schema.MigrationLoopStatus{Tenants: []schema.MigrationLoopTenantStatus{{
				Name:        "tenants-1",
				Completed:   true,
				CurrentStep: StepDone,
				FinishedAt:  &finished,
				Steps: []schema.MigrationLoopStepStatus{
					{Step: "manifest", Status: StatusCompleted},
					{Step: "migrate", Status: StatusCompleted},
					{Step: "verify", Status: StatusCompleted},
				},
			}}}
			imported := 2
			if tc.secondPhase == schema.PhasePending {
				imported = 1
			}
			writeTenantStatusEvidence(t, dir, "tenants-1-status.json", 2, imported, 1)

			reconciled, err := reconcileStatus(plan, existing, dir)
			require.NoError(t, err)
			got := reconciled.Tenants[0]
			assert.False(t, got.Completed, "tenant with a shard progress gap must be reopened")
			assert.Equal(t, tc.wantCurrent, got.CurrentStep)
			assert.Nil(t, got.FinishedAt, "reopened tenant drops its recorded finish time")
			for _, step := range got.Steps {
				if step.Step == tc.wantCurrent {
					assert.Equal(t, StatusPending, step.Status)
				}
			}
		})
	}
}

func TestLoopRunWritesStepLogBlockAndMarksCompleted(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())

	stdout, err := runLoopForTest(t, loopPath)
	assert.NoError(t, err, "Run failed")
	out := stdout.String()
	for _, want := range []string{
		"tenant=tenants-1 step=manifest attempt=1 status=running",
		"inline action=manifest",
		"result: status=completed exitCode=0",
		"all-tenants-done",
	} {
		assert.Contains(t, out, want)
	}
	statusPath := filepath.Join(dir, "migration-loop-status.json")
	data, err := os.ReadFile(statusPath)
	require.NoError(t, err)
	var doc schema.MigrationLoopStatus
	require.NoError(t, json.Unmarshal(data, &doc))
	require.Len(t, doc.Tenants, 1)
	assert.True(t, doc.Tenants[0].Completed, "tenant should be completed")
	require.NotEmpty(t, doc.Tenants[0].Steps)
	step := doc.Tenants[0].Steps[0]
	if step.Status != StatusCompleted {
		t.Fatalf("step status: got %s want %s", step.Status, StatusCompleted)
	}
	require.NotNil(t, step.LastExitCode, "completed step should record LastExitCode, got nil")
	exitCode := *step.LastExitCode
	assert.Equal(t, 0, exitCode)
	logPath := filepath.Join(dir, "tenants-1.log")
	logData, err := os.ReadFile(logPath)
	require.NoError(t, err)
	logStr := string(logData)
	assert.Contains(t, logStr, "inline action=manifest", "log missing output")
	assert.Contains(t, logStr, "result: status=completed exitCode=0", "log missing result line")
}

func TestLoopRoutesResultProgressAndActionLogSeparately(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())
	var result, progress bytes.Buffer
	loop, err := NewLoop(loopPath, LoopOutput{Result: &result, Progress: &progress})
	require.NoError(t, err)
	loop.runAction = succeedActions

	require.NoError(t, loop.Run(context.Background()))
	actionLog, err := os.ReadFile(filepath.Join(dir, "tenants-1.log"))
	require.NoError(t, err)

	assert.Contains(t, result.String(), "all-tenants-done")
	assert.NotContains(t, result.String(), "inline action")
	assert.Contains(t, progress.String(), "status=running")
	assert.Contains(t, progress.String(), "status=completed")
	assert.NotContains(t, progress.String(), "inline action")
	assert.Contains(t, string(actionLog), "inline action=manifest")
	assert.NotContains(t, string(actionLog), "all-tenants-done")
}

// TestLoopRunsTypedMigrateStepInProcess drives the real action dispatch for a
// single migrate step. The Loop is constructed directly: a one-step plan is
// not a loadable artifact under the step contract, and seeding the other
// three steps as completed would contradict the fresh-shard evidence this
// test needs.
func TestLoopRunsTypedMigrateStepInProcess(t *testing.T) {
	dir := t.TempDir()
	manifestPath := filepath.Join(dir, "tenants-1-manifest.json")
	manifest := `{
  "region":"prod","namespace":"observability","vmcluster":"vmcluster","stepInterval":"1d",
  "tenants":[{"tenant":"1001:0","sourceAddr":"http://src","destinationAddr":"http://dst","shards":[
    {"shardID":"1001:0:1","timeStart":"2026-06-16T00:00:00Z","timeEnd":"2026-06-17T00:00:00Z"}
  ]}]
}`
	require.NoError(t, os.WriteFile(manifestPath, []byte(manifest), 0o644))
	plan := &schema.MigrationLoopPlan{
		Runtime: &schema.MigrationLoopRuntime{
			Region:    "prod",
			Vmcluster: "vmcluster",
			Vminsert:  "http://dst/{tenant}",
			Window: schema.MigrationLoopWindow{
				Start:        "2026-01-01T00:00:00Z",
				End:          "2026-01-02T00:00:00Z",
				StepInterval: "1d",
			},
		},
		Vmselect: &schema.MigrationLoopVmselect{
			Origin:    "http://origin/{tenant}",
			Migration: "http://migration/{tenant}",
			Candidate: "http://candidate/{tenant}",
		},
		Tenants: []schema.MigrationLoopTenant{{
			Name:       "tenants-1",
			StatusFile: "tenants-1-status.json",
			LogFile:    "tenants-1.log",
			Cmds: []schema.MigrationLoopCmd{{
				Step: "migrate",
				Action: &schema.MigrationLoopAction{
					Kind:         schema.ActionMigrate,
					Manifest:     "tenants-1-manifest.json",
					StepInterval: "day",
				},
			}},
		}},
	}
	loopPath := filepath.Join(dir, "migration-loop.json")
	var stdout bytes.Buffer
	vmctl := &loopVmctlRunner{}
	loop := newLoop(loopPath, plan, &schema.MigrationLoopStatus{
		Tenants: []schema.MigrationLoopTenantStatus{{Name: "tenants-1", Steps: []schema.MigrationLoopStepStatus{{Step: "migrate", Status: StatusPending}}}},
	}, combinedLoopOutput(&stdout))
	loop.vmctlRunner = vmctl
	assert.NoError(t, loop.Run(context.Background()), "Run failed")
	assert.Equal(t, 1, vmctl.calls)
	doc, err := status.LoadDocument(status.PathForManifest(manifestPath))
	require.NoError(t, err)
	assert.Equal(t, 1, doc.Summary.Imported)
}

func TestNewLoopRejectsUnknownAction(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, []schema.MigrationLoopCmd{{
		Step:   "legacy",
		Action: &schema.MigrationLoopAction{Kind: "shell"},
	}})
	_, err := NewLoop(loopPath, combinedLoopOutput(&bytes.Buffer{}))
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "rerun prepare")
}

func TestNewLoopRejectsActionKindThatDoesNotMatchStep(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, []schema.MigrationLoopCmd{{
		Step:   "manifest",
		Action: &schema.MigrationLoopAction{Kind: schema.ActionVerify, Manifest: "tenants-1-manifest.json"},
	}})
	_, err := NewLoop(loopPath, combinedLoopOutput(io.Discard))
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "does not match")
}

func TestNewLoopRejectsStepWithoutAction(t *testing.T) {
	dir := t.TempDir()
	plan := schema.MigrationLoopPlan{
		Runtime:  &schema.MigrationLoopRuntime{},
		Vmselect: &schema.MigrationLoopVmselect{},
		Tenants: []schema.MigrationLoopTenant{{
			Name: "t1",
			Cmds: []schema.MigrationLoopCmd{{
				Step: "manifest",
			}},
		}},
	}
	data, err := json.Marshal(plan)
	require.NoError(t, err)
	path := filepath.Join(dir, "migration-loop.json")
	require.NoError(t, os.WriteFile(path, data, 0o644))
	_, err = NewLoop(path, combinedLoopOutput(&bytes.Buffer{}))
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "rerun prepare")
}

// writeTenantStatusEvidence writes a per-tenant shard status file whose
// summary backs the loop-status step markers: the first `imported` shards
// are imported and the first `verified` shards are verified.
func writeTenantStatusEvidence(t *testing.T, dir, file string, total, imported, verified int) {
	t.Helper()
	doc := status.BuildInitialStatus(&schema.Manifest{Tenants: []schema.ManifestTenant{{
		Tenant: "1001:0",
	}}}, "tenants-1-manifest.json")
	for i := range total {
		shard := schema.Shard{ShardID: fmt.Sprintf("1001:0:%d", i+1)}
		if i < imported {
			shard.Phase = schema.PhaseImported
		}
		if i < verified {
			shard.Verified = true
		}
		doc.Shards = append(doc.Shards, shard)
	}
	require.NoError(t, status.WriteJSONAtomic(filepath.Join(dir, file), doc))
}

func TestLoopResumesFromFailedStep(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())

	statusPath := filepath.Join(dir, "migration-loop-status.json")
	seed := schema.MigrationLoopStatus{
		UpdatedAt: "2026-07-08T12:00:00Z",
		Tenants: []schema.MigrationLoopTenantStatus{
			{
				Name:        "tenants-1",
				Completed:   false,
				Failed:      true,
				CurrentStep: "manifest",
				Steps: []schema.MigrationLoopStepStatus{
					{
						Step:     "manifest",
						Status:   StatusFailed,
						Attempts: 1,
					},
					{
						Step:     "migrate",
						Status:   StatusCompleted,
						Attempts: 1,
					},
					{
						Step:     "verify",
						Status:   StatusCompleted,
						Attempts: 1,
					},
				},
			},
		},
	}
	data, err := json.MarshalIndent(seed, "", "  ")
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(statusPath, append(data, '\n'), 0o644))
	writeTenantStatusEvidence(t, dir, "tenants-1-status.json", 2, 2, 2)

	var stdout bytes.Buffer
	loop, err := NewLoop(loopPath, combinedLoopOutput(&stdout))
	require.NoError(t, err)
	_ = loop.Run(context.Background())

	finalPath := filepath.Join(dir, "migration-loop-status.json")
	finalData, err := os.ReadFile(finalPath)
	require.NoError(t, err)
	var finalDoc schema.MigrationLoopStatus
	require.NoError(t, json.Unmarshal(finalData, &finalDoc))
	tenant := finalDoc.Tenants[0]
	assert.Equal(t, 1, tenant.Steps[2].Attempts, "verify attempts: should not re-run")
	assert.Equal(t, 2, tenant.Steps[0].Attempts, "manifest attempts: one retry on resume")
}

func TestClassifyError(t *testing.T) {
	cases := []struct {
		name     string
		err      error
		category string
	}{
		{"typed transient", fmt.Errorf("connection refused: %w", retry.ErrTransient), CategoryTransient},
		{"typed permanent", fmt.Errorf("bad config: %w", retry.ErrPermanent), CategoryPermanent},
		{"unclassified defaults permanent", errors.New("invalid manifest"), CategoryPermanent},
		{"context cancellation", context.Canceled, CategoryInfrastructure},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			got := classifyError(tc.err)
			assert.Equal(t, tc.category, got.Category)
		})
	}
}

func TestLoopAllTenantsAlreadyDone(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())
	writeTenantStatusEvidence(t, dir, "tenants-1-status.json", 1, 1, 1)
	statusPath := filepath.Join(dir, "migration-loop-status.json")
	seed := schema.MigrationLoopStatus{
		UpdatedAt: "2026-07-08T12:00:00Z",
		Tenants: []schema.MigrationLoopTenantStatus{
			{
				Name:        "tenants-1",
				Completed:   true,
				CurrentStep: StepDone,
				Steps: []schema.MigrationLoopStepStatus{
					{Step: "manifest", Status: StatusCompleted, Attempts: 1},
					{Step: "migrate", Status: StatusCompleted, Attempts: 1},
					{Step: "verify", Status: StatusCompleted, Attempts: 1},
				},
			},
		},
	}
	data, err := json.MarshalIndent(seed, "", "  ")
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(statusPath, append(data, '\n'), 0o644))
	stdout, err := runLoopForTest(t, loopPath)
	assert.NoError(t, err, "Run failed")
	assert.Contains(t, stdout.String(), "all-tenants-done")
}

// The second runLoopForTest rebuilds via NewLoop — the reconcile path that
// used to wipe finishedAt. Shard evidence keeps it from failing closed.
func TestSummaryDurationFrozenAcrossPasses(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())

	_, err := runLoopForTest(t, loopPath)
	require.NoError(t, err, "pass 1 failed")
	writeTenantStatusEvidence(t, dir, "tenants-1-status.json", 1, 1, 1)
	_, err = runLoopForTest(t, loopPath)
	require.NoError(t, err, "pass 2 failed")

	var doc schema.MigrationLoopStatus
	require.NoError(t, jsonio.ReadFile(filepath.Join(dir, "migration-loop-status.json"), &doc))
	tenant := doc.Tenants[0]
	require.NotNil(t, tenant.StartedAt)
	require.NotNil(t, tenant.FinishedAt, "second pass must not wipe tenant finishedAt")

	s, err := ReadLoopSummary(loopPath)
	require.NoError(t, err)
	row := s.Rows[0]
	assert.Equal(t, StatusCompleted, row.Status)
	assert.Equal(t, tenant.FinishedAt.Sub(*tenant.StartedAt).Seconds(), row.Duration)
}

// TestLoopAnchorsArtifactsOnPlanDir runs a loop from a process cwd that differs
// from the plan's directory and asserts action cwd, per-tenant log, and loop
// status all anchor on the plan dir.
func TestLoopAnchorsArtifactsOnPlanDir(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())

	other := t.TempDir()
	orig, err := os.Getwd()
	require.NoError(t, err)
	require.NoError(t, os.Chdir(other))
	t.Cleanup(func() { _ = os.Chdir(orig) })

	var stdout bytes.Buffer
	loop, err := NewLoop(loopPath, combinedLoopOutput(&stdout))
	require.NoError(t, err)
	loop.runAction = func(_ context.Context, _ *schema.MigrationLoopPlan, planDir string, _ *schema.MigrationLoopAction, _ VmctlRunner, _ ActionOutput) error {
		return os.WriteFile(filepath.Join(planDir, "cwd-probe.txt"), []byte("ok"), 0o644)
	}
	assert.NoError(t, loop.Run(context.Background()), "Run failed")

	for _, artifact := range []string{"cwd-probe.txt", "tenants-1.log", "migration-loop-status.json"} {
		_, err := os.Stat(filepath.Join(dir, artifact))
		assert.NoError(t, err, "artifact %s should land next to the plan", artifact)
	}
	_, err = os.Stat(filepath.Join(other, "cwd-probe.txt"))
	assert.Error(t, err, "action wrote into the process cwd, want plan dir")
}

// TestLoopSummaryGapAnchorsOnPlanDir constructs a Loop from a process cwd that
// differs from the plan's directory and asserts a per-tenant summary gap still
// reopens work by resolving the summary next to the plan.
func TestLoopSummaryGapAnchorsOnPlanDir(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())
	summaryPath := filepath.Join(dir, "tenants-1-status.json")
	doc := status.BuildInitialStatus(&schema.Manifest{
		Tenants: []schema.ManifestTenant{
			{Tenant: "1001:0", Shards: []schema.ManifestShard{{ShardID: "1001:0:1"}}},
		},
	}, "tenants-1-manifest.json")
	require.NoError(t, status.WriteJSONAtomic(summaryPath, doc))
	existing := &schema.MigrationLoopStatus{Tenants: []schema.MigrationLoopTenantStatus{{
		Name:        "tenants-1",
		Completed:   true,
		CurrentStep: StepDone,
		Steps: []schema.MigrationLoopStepStatus{
			{Step: "manifest", Status: StatusCompleted},
			{Step: "migrate", Status: StatusCompleted},
			{Step: "verify", Status: StatusCompleted},
		},
	}}}
	require.NoError(t, dumpStatusAtomic(DeriveLoopStatusPath(loopPath), existing))

	other := t.TempDir()
	orig, err := os.Getwd()
	require.NoError(t, err)
	require.NoError(t, os.Chdir(other))
	t.Cleanup(func() { _ = os.Chdir(orig) })

	var loop *Loop
	loop, err = NewLoop(loopPath, combinedLoopOutput(&bytes.Buffer{}))
	require.NoError(t, err)
	got := loop.status.Tenants[0]
	assert.False(t, got.Completed)
	assert.Equal(t, "migrate", got.CurrentStep)
	assert.Equal(t, StatusCompleted, got.Steps[0].Status)
	assert.Equal(t, StatusPending, got.Steps[1].Status)
}

func TestLoopUsesPlanLogFileName(t *testing.T) {
	dir := t.TempDir()
	plan := schema.MigrationLoopPlan{
		Runtime:  &schema.MigrationLoopRuntime{},
		Vmselect: &schema.MigrationLoopVmselect{},
		Tenants: []schema.MigrationLoopTenant{{
			Name:        "1001:0",
			TenantsFile: "tenants-1001_0.json",
			StatusFile:  "tenants-1001_0-status.json",
			LogFile:     "tenants-1001_0.log",
			Cmds: []schema.MigrationLoopCmd{
				{Step: "manifest", Action: testAction("manifest")},
				{Step: "migrate", Action: testAction("migrate")},
				{Step: "verify", Action: testAction("verify")},
			},
		}},
	}
	data, err := json.MarshalIndent(plan, "", "  ")
	require.NoError(t, err)
	loopPath := filepath.Join(dir, "migration-loop.json")
	require.NoError(t, os.WriteFile(loopPath, append(data, '\n'), 0o644))

	var stdout bytes.Buffer
	loop, err := NewLoop(loopPath, combinedLoopOutput(&stdout))
	require.NoError(t, err)
	loop.runAction = succeedActions
	assert.NoError(t, loop.Run(context.Background()), "Run failed")

	_, err = os.Stat(filepath.Join(dir, "tenants-1001_0.log"))
	require.NoError(t, err)
	_, err = os.Stat(filepath.Join(dir, "1001:0.log"))
	assert.Error(t, err)
}

func TestReconcileFailsClosedOnMissingEvidence(t *testing.T) {
	// seed writes a loop-status whose tenant claims every step up to and
	// including the given step completed, with an optional per-tenant status
	// file on disk.
	seed := func(t *testing.T, throughStep, statusContent string) string {
		t.Helper()
		dir := t.TempDir()
		loopPath := writeLoopPlan(t, dir, contractCmds())
		if statusContent != "" {
			require.NoError(t, os.WriteFile(filepath.Join(dir, "tenants-1-status.json"), []byte(statusContent), 0o644))
		}
		steps := make([]schema.MigrationLoopStepStatus, 0, 4)
		for _, c := range contractCmds() {
			steps = append(steps, schema.MigrationLoopStepStatus{Step: c.Step, Status: StatusCompleted, Attempts: 1})
			if c.Step == throughStep {
				break
			}
		}
		doc := schema.MigrationLoopStatus{
			UpdatedAt: "2026-07-08T12:00:00Z",
			Tenants: []schema.MigrationLoopTenantStatus{{
				Name:        "tenants-1",
				Completed:   false,
				CurrentStep: throughStep,
				Steps:       steps,
			}},
		}
		data, err := json.MarshalIndent(doc, "", "  ")
		require.NoError(t, err)
		require.NoError(t, os.WriteFile(filepath.Join(dir, "migration-loop-status.json"), append(data, '\n'), 0o644))
		return loopPath
	}
	f := func(t *testing.T, throughStep, statusContent, wantErr string) {
		t.Helper()
		_, err := NewLoop(seed(t, throughStep, statusContent), combinedLoopOutput(io.Discard))
		if wantErr == "" {
			require.NoError(t, err)
			return
		}
		assert.ErrorContains(t, err, wantErr)
	}
	t.Run("MissingStatusFileWithCompletedVerify", func(t *testing.T) {
		f(t, "verify", "", "claims completed migrate/verify steps")
	})
	t.Run("CorruptStatusFile", func(t *testing.T) {
		f(t, "verify", "not json at all", "is unreadable")
	})
	t.Run("CompletedManifestAloneDoesNotTrip", func(t *testing.T) {
		f(t, "manifest", "", "")
	})
	t.Run("FreshTenantWithoutEvidenceIsFine", func(t *testing.T) {
		dir := t.TempDir()
		loopPath := writeLoopPlan(t, dir, contractCmds())
		_, err := NewLoop(loopPath, combinedLoopOutput(io.Discard))
		assert.NoError(t, err, "no loop-status yet: everything pending, no claims to verify")
	})
}

func TestLoadInputsRejectsNonContractStepSequence(t *testing.T) {
	cases := []struct {
		name    string
		steps   []string
		wantErr string
	}{
		{"MissingMigrate", []string{"manifest", "verify"}, "steps must be exactly"},
		{"MissingVerify", []string{"manifest", "migrate"}, "steps must be exactly"},
		{"DuplicateMigrate", []string{"manifest", "migrate", "migrate", "verify"}, "steps must be exactly"},
		{"VerifyBeforeMigrate", []string{"manifest", "verify", "migrate"}, "steps must be exactly"},
		{"ManifestOnly", []string{"manifest"}, "steps must be exactly"},
		{"Empty", nil, "steps must be exactly"},
		// summary-status is no longer a supported action kind at all.
		{"SummaryPresent", []string{"manifest", "migrate", "verify", "summary-status"}, `unsupported action kind "summary-status"`},
		// precheck is no longer a supported action kind at all.
		{"PrecheckPresent", []string{"manifest", "precheck", "migrate", "verify"}, `unsupported action kind "precheck"`},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			dir := t.TempDir()
			cmds := make([]schema.MigrationLoopCmd, 0, len(tc.steps))
			for _, s := range tc.steps {
				cmds = append(cmds, schema.MigrationLoopCmd{Step: s, Action: testAction(s)})
			}
			loopPath := writeLoopPlan(t, dir, cmds)
			_, _, err := loadInputs(loopPath, DeriveLoopStatusPath(loopPath))
			assert.ErrorContains(t, err, tc.wantErr)
		})
	}
}

func TestLoadInputsRejectsMissingTenantPlaceholder(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, []schema.MigrationLoopCmd{{Step: "manifest"}})
	planBytes, err := os.ReadFile(loopPath)
	require.NoError(t, err)
	corrupted := strings.Replace(string(planBytes), "http://dst/{tenant}", "http://dst/insert/0/prometheus", 1)
	assert.NotEqual(t, string(planBytes), corrupted, "fixture must actually drop the placeholder")
	require.NoError(t, os.WriteFile(loopPath, []byte(corrupted), 0o644))

	_, _, err = loadInputs(loopPath, DeriveLoopStatusPath(loopPath))
	assert.ErrorContains(t, err, "vminsert URL template must contain {tenant}")
}

func TestLoopRunCancelledPassLeavesRemainingTenantsUntouched(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlanTenants(t, dir, []schema.MigrationLoopTenant{
		{
			Name:        "tenants-1",
			TenantsFile: "tenants-1.json",
			StatusFile:  "tenants-1-status.json",
			Cmds:        contractCmds(),
		},
		{
			Name:        "tenants-2",
			TenantsFile: "tenants-2.json",
			StatusFile:  "tenants-2-status.json",
			Cmds:        contractCmds(),
		},
	})

	ctx, cancel := context.WithCancel(context.Background())
	loop, err := NewLoop(loopPath, combinedLoopOutput(io.Discard))
	require.NoError(t, err)
	blocked := make(chan struct{})
	loop.runAction = func(ctx context.Context, _ *schema.MigrationLoopPlan, _ string, _ *schema.MigrationLoopAction, _ VmctlRunner, _ ActionOutput) error {
		close(blocked)
		<-ctx.Done()
		return ctx.Err()
	}
	runErr := make(chan error, 1)
	go func() {
		runErr <- loop.Run(ctx)
	}()
	<-blocked
	cancel()
	err = <-runErr

	var ec *exitcode.Error
	require.ErrorAs(t, err, &ec)
	assert.Equal(t, exitcode.Cancelled, ec.Code, "cancelled pass must exit Cancelled, got err=%v", err)

	finalData, err := os.ReadFile(filepath.Join(dir, "migration-loop-status.json"))
	require.NoError(t, err)
	var final schema.MigrationLoopStatus
	require.NoError(t, json.Unmarshal(finalData, &final))
	second := final.Tenants[1]
	assert.False(t, second.Failed, "remaining tenant must not be marked failed by a cancel unwind")
	assert.Equal(t, StatusPending, second.Steps[0].Status)
}

// writePreviewableLoopPlan stages the standard loop fixture plus its tenants
// inventory file.
func writePreviewableLoopPlan(t *testing.T, dir string) string {
	t.Helper()
	loopPath := writeLoopPlan(t, dir, contractCmds())
	require.NoError(t, os.WriteFile(filepath.Join(dir, "tenants-1.json"), []byte(`["1001:0"]`), 0o644))
	return loopPath
}

func previewMigrateStep(t *testing.T, pending []schema.MigrationLoopPending) *schema.MigrationLoopPendingStep {
	t.Helper()
	for i := range pending[0].Steps {
		if pending[0].Steps[i].Step == "migrate" {
			return &pending[0].Steps[i]
		}
	}
	t.Fatal("migrate step not found in pending steps")
	return nil
}

func TestPendingStepsPreviewsMigrateCommands(t *testing.T) {
	dir := t.TempDir()
	loopPath := writePreviewableLoopPlan(t, dir)

	loop, err := NewLoop(loopPath, combinedLoopOutput(io.Discard))
	require.NoError(t, err)
	pending := loop.PendingSteps()
	assert.Len(t, pending, 1)

	migrate := previewMigrateStep(t, pending)
	// The 1d fixture window previews as a single shard.
	require.Len(t, migrate.Cmds, 1, "cmds=%v", migrate.Cmds)
	assert.Contains(t, migrate.Cmds[0], "vmctl vm-native")

	for _, name := range []string{"tenants-1-manifest.json", "tenants-1-status.json"} {
		_, err := os.Stat(filepath.Join(dir, name))
		assert.True(t, os.IsNotExist(err), "preview must not write %s", name)
	}
}

func TestPendingStepsPrefersOnDiskManifest(t *testing.T) {
	dir := t.TempDir()
	loopPath := writePreviewableLoopPlan(t, dir)
	// Two shards on disk where the 1d window previews one.
	manifest := `{"region":"prod","vmcluster":"vmcluster","stepInterval":"1d","tenants":[
		{"tenant":"1001:0","sourceAddr":"http://src","destinationAddr":"http://dst",
		 "shards":[
			{"shardID":"a","timeStart":"2026-01-01T00:00:00Z","timeEnd":"2026-01-01T12:00:00Z"},
			{"shardID":"b","timeStart":"2026-01-01T12:00:00Z","timeEnd":"2026-01-02T00:00:00Z"}]}]}`
	require.NoError(t, os.WriteFile(filepath.Join(dir, "tenants-1-manifest.json"), []byte(manifest), 0o644))

	loop, err := NewLoop(loopPath, combinedLoopOutput(io.Discard))
	require.NoError(t, err)
	migrate := previewMigrateStep(t, loop.PendingSteps())
	assert.Len(t, migrate.Cmds, 2, "on-disk manifest must win over the computed preview")
}

func TestLoopWarnings(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())
	var out bytes.Buffer
	loop, err := NewLoop(loopPath, LoopOutput{Progress: &out})
	require.NoError(t, err)

	assert.Empty(t, loop.Warnings())
	loop.recordWarning("stale tmp file left behind")
	loop.recordWarning("shard gap in summary")
	assert.Equal(t, []string{"stale tmp file left behind", "shard gap in summary"}, loop.Warnings())
	assert.Contains(t, out.String(), "warning: stale tmp file left behind")

	w := loop.Warnings()
	w[0] = "mutated"
	assert.Equal(t, "stale tmp file left behind", loop.Warnings()[0], "Warnings must return a copy")
}

func TestWithActionExecutor(t *testing.T) {
	dir := t.TempDir()
	loopPath := writeLoopPlan(t, dir, contractCmds())

	var dispatched []string
	var out bytes.Buffer
	loop, err := NewLoop(loopPath, combinedLoopOutput(&out), WithActionExecutor(
		func(_ context.Context, _ *schema.MigrationLoopPlan, _ string, action *schema.MigrationLoopAction, _ VmctlRunner, output ActionOutput) error {
			dispatched = append(dispatched, string(action.Kind))
			_, err := fmt.Fprintln(output.Log, "injected")
			return err
		}))
	require.NoError(t, err)

	// Constructed before Run: afterwards the on-disk loop-status claims
	// completed steps that the no-op executor left no evidence for, and
	// NewLoop correctly fails closed on a reload.
	nilOpt, err := NewLoop(loopPath, combinedLoopOutput(&bytes.Buffer{}), WithActionExecutor(nil))
	require.NoError(t, err)
	assert.NotNil(t, nilOpt.runAction, "nil executor must leave the default dispatch in place")

	require.NoError(t, loop.Run(context.Background()))
	assert.Equal(t, []string{"manifest", "migrate", "verify"}, dispatched)
	assert.Contains(t, out.String(), "all-tenants-done")
}