package main

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"mime/multipart"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/runs"
)

func TestParseRunArgs(t *testing.T) {
	ok, err := parseRunArgs([]string{"--app", "shop", "--instance", "h.example.com", "--file", "a.csv", "--file", "b.csv", "--timeout", "2m", "--", "import", "--dry-run"}, io.Discard)
	if err != nil {
		t.Fatal(err)
	}
	if ok.app != "shop" || ok.instance != "h.example.com" || len(ok.files) != 2 || ok.timeout != 2*time.Minute || strings.Join(ok.argv, " ") != "import --dry-run" {
		t.Fatalf("opts %+v", ok)
	}
	def, err := parseRunArgs([]string{"--instance", "h.example.com", "--", "x"}, io.Discard)
	if err != nil || def.timeout != 30*time.Minute {
		t.Fatalf("default timeout %s, %v", def.timeout, err)
	}
	for args, want := range map[string]string{
		"--app shop --instance h.example.com import --dry-run": "",
		"--app shop -- import":                                 "--app and --instance name the instance to run against",
		"--instance h.example.com --timeout 7h -- x":           "--timeout must be between 1s and 6h",
		"--instance h.example.com --timeout 0s -- x":           "--timeout must be between 1s and 6h",
		"--instance h.example.com --follow run_x -- x":         "--follow takes a run id and no command after --",
		"--instance h.example.com --follow run_x --file a":     "--follow takes a run id and no command after --",
	} {
		_, err := parseRunArgs(strings.Fields(args), io.Discard)
		if err == nil {
			t.Errorf("%q accepted", args)
			continue
		}
		if want == "" && !errors.Is(err, errRunUsage) {
			t.Errorf("%q: %v, want the usage error", args, err)
		}
		if want != "" && err.Error() != want {
			t.Errorf("%q: %q, want %q", args, err, want)
		}
	}
}

func TestRunMetaForStatsEveryFile(t *testing.T) {
	dir := t.TempDir()
	p := filepath.Join(dir, "checkins.csv")
	if err := os.WriteFile(p, []byte("a,b\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	m, err := runMetaFor(runOpts{argv: []string{"import"}, timeout: 90 * time.Second, files: []string{p}})
	if err != nil || m.TimeoutS != 90 || len(m.Files) != 1 || m.Files[0].Name != "checkins.csv" || m.Files[0].Size != 4 {
		t.Fatalf("meta %+v, %v", m, err)
	}
	missing := filepath.Join(dir, "nope.csv")
	if _, err := runMetaFor(runOpts{files: []string{missing}}); !errors.Is(err, errRunLocalFile) || !strings.HasPrefix(err.Error(), "could not read "+missing+": ") {
		t.Fatalf("missing file: %v", err)
	}
}

func TestRunAPIWireShapes(t *testing.T) {
	type seen struct{ method, path, auth, body string }
	var got []seen
	var metaJSON, fileName, fileBody string
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		s := seen{method: r.Method, path: r.URL.RequestURI(), auth: r.Header.Get("Authorization")}
		if strings.HasPrefix(r.Header.Get("Content-Type"), "multipart/") {
			mr, err := r.MultipartReader()
			if err != nil {
				t.Fatal(err)
			}
			for {
				part, err := mr.NextPart()
				if err != nil {
					break
				}
				raw, _ := io.ReadAll(part)
				if part.FormName() == "meta" {
					metaJSON = string(raw)
				} else {
					fileName, fileBody = part.FileName(), string(raw)
				}
			}
			got = append(got, s)
			w.WriteHeader(http.StatusAccepted)
			json.NewEncoder(w).Encode(runs.Created{ID: "run_a", Host: "h.example.com", State: "queued"})
			return
		}
		raw, _ := io.ReadAll(r.Body)
		s.body = string(raw)
		got = append(got, s)
		switch {
		case strings.Contains(s.path, "/expire"):
			w.WriteHeader(http.StatusConflict)
			json.NewEncoder(w).Encode(map[string]string{"error": "run run_a has already started"})
		case r.Method == http.MethodGet:
			json.NewEncoder(w).Encode(runs.Page{ID: "run_a", State: "picked"})
		default:
			json.NewEncoder(w).Encode(runs.StateAnswer{State: "cancel"})
		}
	}))
	defer ts.Close()
	dir := t.TempDir()
	p := filepath.Join(dir, "x.csv")
	os.WriteFile(p, []byte("1,2"), 0o600)
	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com"}
	ctx := context.Background()
	created, err := c.create(ctx, runs.Meta{Argv: []string{"import"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "x.csv", Size: 3}}}, []string{p})
	if err != nil || created.ID != "run_a" {
		t.Fatalf("create %+v, %v", created, err)
	}
	if !strings.Contains(metaJSON, `"argv":["import"]`) || fileName != "x.csv" || fileBody != "1,2" {
		t.Fatalf("meta %s file %s %q", metaJSON, fileName, fileBody)
	}
	if _, err := c.page(ctx, "run_a", 3); err != nil {
		t.Fatal(err)
	}
	if st, err := c.cancel(ctx, "run_a"); err != nil || st != "cancel" {
		t.Fatalf("cancel %q, %v", st, err)
	}
	_, err = c.expire(ctx, "run_a")
	var ce *runConsoleError
	if !errors.As(err, &ce) || ce.status != http.StatusConflict || ce.msg != "run run_a has already started" {
		t.Fatalf("expire %v", err)
	}
	if err := c.ack(ctx, "run_a", 7); err != nil {
		t.Fatal(err)
	}
	base := "/api/cli/apps/acme/shop/instances/h.example.com/runs"
	want := []seen{
		{http.MethodPost, base, "Bearer cli-token", ""},
		{http.MethodGet, base + "/run_a?after=3", "Bearer cli-token", ""},
		{http.MethodPost, base + "/run_a/cancel", "Bearer cli-token", "{}"},
		{http.MethodPost, base + "/run_a/expire", "Bearer cli-token", "{}"},
		{http.MethodPost, base + "/run_a/ack", "Bearer cli-token", `{"last_seq":7}`},
	}
	if len(got) != len(want) {
		t.Fatalf("requests %+v", got)
	}
	for i := range want {
		if got[i] != want[i] {
			t.Errorf("request %d = %+v, want %+v", i, got[i], want[i])
		}
	}
}

func TestRunMetaForRefusesWhatIsNotAFile(t *testing.T) {
	dir := t.TempDir()
	_, err := runMetaFor(runOpts{files: []string{dir}})
	if !errors.Is(err, errRunLocalFile) || !errors.Is(err, errRunNotRegular) || !strings.HasPrefix(err.Error(), "could not read "+dir+": ") {
		t.Fatalf("directory: %v", err)
	}
}

// A part that disagrees with meta must never reach the console as a whole
// request: the file changed between the stat and the upload.
func TestRunAPICreateFailsWhenAFileChangesSize(t *testing.T) {
	for name, tc := range map[string]struct {
		content string
		size    int64
	}{
		"shrunk": {"12", 3},
		"grown":  {"12345", 3},
	} {
		t.Run(name, func(t *testing.T) {
			var complete atomic.Bool
			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				mr, err := r.MultipartReader()
				if err != nil {
					return
				}
				for {
					part, err := mr.NextPart()
					if err == io.EOF {
						break
					}
					if err != nil {
						return
					}
					if _, err := io.Copy(io.Discard, part); err != nil {
						return
					}
				}
				complete.Store(true)
				w.WriteHeader(http.StatusAccepted)
				json.NewEncoder(w).Encode(runs.Created{ID: "run_a"})
			}))
			p := filepath.Join(t.TempDir(), "x.csv")
			if err := os.WriteFile(p, []byte(tc.content), 0o600); err != nil {
				t.Fatal(err)
			}
			c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com"}
			_, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "x.csv", Size: tc.size}}}, []string{p})
			// Close waits for the handler, so complete is settled when read.
			ts.Close()
			var uerr *url.Error
			if !errors.Is(err, errRunLocalFile) || errors.As(err, &uerr) || !strings.HasPrefix(err.Error(), "could not read "+p+": ") {
				t.Fatalf("create: %v, want the bare could-not-read error", err)
			}
			if complete.Load() {
				t.Fatal("the console received a whole request whose part disagrees with meta")
			}
		})
	}
}

// fakeRunFile is an open file whose reader and reported sizes the test
// controls.
type fakeRunFile struct {
	io.Reader
	mode  os.FileMode
	sizes []int64 // what successive Stat calls report; the last repeats
}

type fakeInfo struct {
	os.FileInfo
	mode os.FileMode
	size int64
}

func (i fakeInfo) Mode() os.FileMode { return i.mode }
func (i fakeInfo) Size() int64       { return i.size }

func (f *fakeRunFile) Stat() (os.FileInfo, error) {
	size := f.sizes[0]
	if len(f.sizes) > 1 {
		f.sizes = f.sizes[1:]
	}
	return fakeInfo{mode: f.mode, size: size}, nil
}

func TestCopyRunFilePartBranches(t *testing.T) {
	want := runs.MetaFile{Name: "x.csv", Size: 4}
	for name, tc := range map[string]struct {
		f    *fakeRunFile
		fail error // nil: the part is written
	}{
		"as promised":          {&fakeRunFile{strings.NewReader("1234"), 0, []int64{4, 4}}, nil},
		"stat disagrees":       {&fakeRunFile{strings.NewReader("1234"), 0, []int64{5}}, errRunFileChanged},
		"reader short":         {&fakeRunFile{strings.NewReader("12"), 0, []int64{4, 4}}, errRunFileChanged},
		"grown after the copy": {&fakeRunFile{strings.NewReader("1234"), 0, []int64{4, 9}}, errRunFileChanged},
		"not a regular file":   {&fakeRunFile{strings.NewReader("1234"), os.ModeDir, []int64{4}}, errRunNotRegular},
		"reader fails midway":  {&fakeRunFile{io.MultiReader(strings.NewReader("12"), iotestErrReader{}), 0, []int64{4, 4}}, errRunReaderBroke},
	} {
		t.Run(name, func(t *testing.T) {
			var buf bytes.Buffer
			mw := multipart.NewWriter(&buf)
			err := copyRunFilePart(mw, tc.f, "/data/x.csv", want)
			if tc.fail == nil {
				if err != nil || !strings.Contains(buf.String(), "1234") {
					t.Fatalf("err %v, body %q", err, buf.String())
				}
				return
			}
			if !errors.Is(err, tc.fail) || !errors.Is(err, errRunLocalFile) || !strings.HasPrefix(err.Error(), "could not read /data/x.csv: ") {
				t.Fatalf("err %v, want %v as a local file error", err, tc.fail)
			}
		})
	}
}

var errRunReaderBroke = errors.New("disk fell over")

type iotestErrReader struct{}

func (iotestErrReader) Read([]byte) (int, error) { return 0, errRunReaderBroke }

// The wire order the console's door requires: meta first, then one file
// part per meta file in meta's order, each named as meta names it.
func TestRunAPICreateSendsMetaThenFilesInOrder(t *testing.T) {
	type part struct{ form, name, body string }
	var got []part
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		mr, err := r.MultipartReader()
		if err != nil {
			t.Error(err)
			return
		}
		for {
			p, err := mr.NextPart()
			if err != nil {
				break
			}
			raw, _ := io.ReadAll(p)
			got = append(got, part{p.FormName(), p.FileName(), string(raw)})
		}
		w.WriteHeader(http.StatusAccepted)
		json.NewEncoder(w).Encode(runs.Created{ID: "run_a"})
	}))
	defer ts.Close()
	dir := t.TempDir()
	var paths []string
	var files []runs.MetaFile
	for _, f := range []struct{ name, body string }{{"b.csv", "bb"}, {"a.csv", "a"}} {
		p := filepath.Join(dir, f.name)
		if err := os.WriteFile(p, []byte(f.body), 0o600); err != nil {
			t.Fatal(err)
		}
		paths = append(paths, p)
		files = append(files, runs.MetaFile{Name: f.name, Size: int64(len(f.body))})
	}
	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com"}
	if _, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60, Files: files}, paths); err != nil {
		t.Fatal(err)
	}
	if len(got) != 3 || got[0].form != "meta" || got[0].name != "" ||
		got[1] != (part{"file", "b.csv", "bb"}) || got[2] != (part{"file", "a.csv", "a"}) {
		t.Fatalf("parts %+v", got)
	}
}

func TestRunAPICreateRefusesFilesThatDisagreeWithMeta(t *testing.T) {
	c := runAPI{s: newTestSession("http://127.0.0.1:1"), a: apiAcct{Seg: "acme"}, app: "a", host: "h.example.com"}
	_, err := c.create(context.Background(), runs.Meta{Argv: []string{"x"}, TimeoutS: 60}, []string{"/nowhere"})
	if err == nil {
		t.Fatal("create accepted more paths than meta files")
	}
}

// scriptedRun is a runClient that answers pages from a script.
type scriptedRun struct {
	pages      []runs.Page
	pageErr    error   // every page call fails with it
	pageErrs   []error // the first page calls fail with these, in order
	cancelErrs []error // one per cancel call, in order; cancelErr once spent
	expireTo   string
	expireErr  error
	cancelErr  error
	createErr  error
	ackErrs    []error // one per ack call, in order; nil once spent
	acked      []int64
	cancels    int
	expires    int
	created    runs.Meta

	// mu guards cancels and pageCalls, which a test reads from its own
	// goroutine while followRun's goroutine writes them.
	mu        sync.Mutex
	pageCalls int
}

// cancelsMade and pagesRead are safe to call while followRun is running.
func (s *scriptedRun) cancelsMade() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.cancels
}

func (s *scriptedRun) pagesRead() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.pageCalls
}

func (s *scriptedRun) create(_ context.Context, m runs.Meta, _ []string) (runs.Created, error) {
	s.created = m
	if s.createErr != nil {
		return runs.Created{}, s.createErr
	}
	return runs.Created{ID: "run_a", State: "queued"}, nil
}
func (s *scriptedRun) page(context.Context, string, int64) (runs.Page, error) {
	s.mu.Lock()
	s.pageCalls++
	s.mu.Unlock()
	if len(s.pageErrs) > 0 {
		err := s.pageErrs[0]
		s.pageErrs = s.pageErrs[1:]
		if err != nil { // nil: this call is answered from pages
			return runs.Page{}, err
		}
	}
	if s.pageErr != nil {
		return runs.Page{}, s.pageErr
	}
	p := s.pages[0]
	if len(s.pages) > 1 {
		s.pages = s.pages[1:]
	}
	return p, nil
}
func (s *scriptedRun) cancel(context.Context, string) (string, error) {
	s.mu.Lock()
	s.cancels++
	s.mu.Unlock()
	if len(s.cancelErrs) > 0 {
		err := s.cancelErrs[0]
		s.cancelErrs = s.cancelErrs[1:]
		return "cancel", err
	}
	return "cancel", s.cancelErr
}
func (s *scriptedRun) expire(context.Context, string) (string, error) {
	s.expires++
	return s.expireTo, s.expireErr
}
func (s *scriptedRun) ack(_ context.Context, _ string, last int64) error {
	s.acked = append(s.acked, last)
	if len(s.ackErrs) == 0 {
		return nil
	}
	err := s.ackErrs[0]
	s.ackErrs = s.ackErrs[1:]
	return err
}

type termFx struct {
	out, err strings.Builder
	sig      chan os.Signal
	now      time.Time
}

func newTermFx() *termFx {
	return &termFx{sig: make(chan os.Signal, 2), now: time.Unix(1_700_000_000, 0)}
}

func (f *termFx) term() runTerm {
	return runTerm{stdout: &f.out, stderr: &f.err, interrupts: f.sig,
		after: func(d time.Duration) <-chan time.Time {
			f.now = f.now.Add(d)
			c := make(chan time.Time, 1)
			c <- f.now
			return c
		},
		now: func() time.Time { return f.now }}
}

func chunk(seq int64, stream, data string) runs.Chunk {
	return runs.Chunk{Seq: seq, Stream: stream, Data: []byte(data)}
}

func TestFollowRunInterleavesAndAcks(t *testing.T) {
	f := newTermFx()
	s := &scriptedRun{pages: []runs.Page{
		{State: "picked", Chunks: []runs.Chunk{chunk(1, "o", "one\n"), chunk(2, "e", "warn\n")}},
		{State: "picked", Chunks: []runs.Chunk{chunk(3, "o", "two\n")}, Result: &runs.Result{Exit: 0, LastSeq: 3}},
	}}
	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 0 {
		t.Fatalf("exit %d", code)
	}
	if f.out.String() != "one\ntwo\n" || f.err.String() != "warn\n" {
		t.Fatalf("stdout %q stderr %q", f.out.String(), f.err.String())
	}
	if len(s.acked) != 1 || s.acked[0] != 3 {
		t.Fatalf("acked %v", s.acked)
	}
}

func TestFollowRunPagesUntilTheResult(t *testing.T) {
	f := newTermFx()
	res := &runs.Result{Exit: 0, LastSeq: 3}
	s := &scriptedRun{pages: []runs.Page{
		{State: "done", Chunks: []runs.Chunk{chunk(1, "o", "a")}, Result: res, More: true},
		{State: "done", Chunks: []runs.Chunk{chunk(2, "o", "b")}, Result: res, More: true},
		{State: "done", Chunks: []runs.Chunk{chunk(3, "o", "c")}, Result: res},
	}}
	followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
	if f.out.String() != "abc" || len(s.acked) != 1 || s.acked[0] != 3 {
		t.Fatalf("stdout %q acked %v (every page before the ack)", f.out.String(), s.acked)
	}
}

func TestFollowRunExitCodes(t *testing.T) {
	started := time.Unix(1_700_000_000, 0)
	for _, c := range []struct {
		res  runs.Result
		code int
		says string
	}{
		{runs.Result{Exit: 7}, 7, ""},
		{runs.Result{Signal: "9"}, 137, ""},
		{runs.Result{TimedOut: true, Started: started, Ended: started.Add(30 * time.Minute)}, 124, "shop.example.com: the command timed out after 30m0s"},
		{runs.Result{Interrupted: "the instance was restarted"}, 125, "shop.example.com: the run was interrupted: the instance was restarted. it may have done part of its work"},
		{runs.Result{Interrupted: runs.InterruptedLost}, 125, "shop.example.com: the run was interrupted: its box stopped reporting. it may have done part of its work"},
		{runs.Result{Cancelled: true, Started: started}, 130, "shop.example.com: the command was cancelled"},
		{runs.Result{Refused: runs.RefusedExpired}, 3, "shop.example.com: not picked up within 60 seconds. the run did not start. try again"},
		{runs.Result{Refused: runs.RefusedFlat}, 3, "shop.example.com: the box is not on the nested layout yet, so carlos run cannot run there"},
		{runs.Result{Refused: runs.RefusedClaim}, 3, "shop.example.com: the box has host claims armed. carlos run cannot run there yet"},
		{runs.Result{Refused: runs.RefusedWake, Detail: "wake failed: restore"}, 3, "shop.example.com: the instance did not wake. wake failed: restore"},
		{runs.Result{Refused: runs.RefusedIncarnation}, 3, "shop.example.com was deleted and created again after this run was requested. the run did not start"},
		{runs.Result{Refused: runs.RefusedImport}, 3, "shop.example.com is an import that is not committed yet. the run did not start"},
		{runs.Result{Exit: 0, Truncated: true}, 0, "shop.example.com: the output passed 64 MB and was cut off there. the command ran on to its end"},
		{runs.Result{Exit: 5, Truncated: true}, 5, "shop.example.com: the output passed 64 MB and was cut off there. the command ran on to its end"},
	} {
		code, lines := runExit("shop.example.com", c.res)
		if code != c.code || (c.says != "" && !slices.Contains(lines, c.says)) || (c.says == "" && len(lines) != 0) {
			t.Errorf("%+v: %d %q, want %d %q", c.res, code, lines, c.code, c.says)
		}
	}
}

func TestFollowRunTellsCancelledBeforeStartFromCancelledAfter(t *testing.T) {
	f := newTermFx()
	s := &scriptedRun{pages: []runs.Page{{State: "cancelled", Result: &runs.Result{Cancelled: true, LastSeq: 0}}}}
	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 {
		t.Fatalf("exit %d", code)
	}
	if got := f.err.String(); got != "shop.example.com: run run_a was cancelled before it started\n" {
		t.Fatalf("stderr %q", got)
	}
	if len(s.acked) != 1 {
		t.Fatalf("acked %v", s.acked)
	}
}

// A run whose deadline passed before any box launched it has no start
// time: it is told as timed out before it started, never with a duration
// measured from the zero time (re-review 8). Exit 124 either way.
func TestFollowRunTellsTimedOutBeforeStartFromTimedOutAfter(t *testing.T) {
	started := time.Unix(1_700_000_000, 0)
	for _, c := range []struct {
		res  runs.Result
		says string
	}{
		{runs.Result{TimedOut: true, Ended: started}, "shop.example.com: run run_a timed out before it started\n"},
		{runs.Result{TimedOut: true, Started: started, Ended: started.Add(30 * time.Minute)}, "shop.example.com: the command timed out after 30m0s\n"},
	} {
		f := newTermFx()
		s := &scriptedRun{pages: []runs.Page{{State: "done", Result: &c.res}}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 124 {
			t.Fatalf("%+v: exit %d", c.res, code)
		}
		if got := f.err.String(); got != c.says {
			t.Fatalf("%+v: stderr %q, want %q", c.res, got, c.says)
		}
	}
}

func TestFollowRunExpiresAQueuedRunAfterSixtySeconds(t *testing.T) {
	f := newTermFx()
	s := &scriptedRun{pages: []runs.Page{{State: "queued"}}, expireTo: "expired"}
	code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", f.now.Add(60*time.Second))
	if code != 3 || s.expires != 1 || !strings.Contains(f.err.String(), "shop.example.com: not picked up within 60 seconds. the run did not start. try again") {
		t.Fatalf("exit %d expires %d stderr %q", code, s.expires, f.err.String())
	}
	if len(s.acked) != 1 || s.acked[0] != 0 {
		t.Fatalf("acked %v (the uploaded files go now, not with the sweep)", s.acked)
	}
	if f.now.Sub(time.Unix(1_700_000_000, 0)) < 60*time.Second {
		t.Fatal("expired before 60 seconds")
	}
}

func TestFollowRunKeepsFollowingWhenExpiryLoses(t *testing.T) {
	f := newTermFx()
	queued := runs.Page{State: "queued"}
	pages := make([]runs.Page, 0, 62)
	for range 61 {
		pages = append(pages, queued)
	}
	pages = append(pages, runs.Page{State: "done", Result: &runs.Result{Exit: 4}})
	s := &scriptedRun{pages: pages, expireErr: &runConsoleError{status: http.StatusConflict, msg: "run run_a has already started"}}
	if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", f.now.Add(60*time.Second)); code != 4 || s.expires != 1 {
		t.Fatalf("exit %d expires %d (a late pickup wins over the expiry)", code, s.expires)
	}
}

func TestFollowRunAsksAgainWhenTheExpiryCallFailsToArrive(t *testing.T) {
	f := newTermFx()
	s := &scriptedRun{pages: []runs.Page{{State: "queued"}}, expireTo: "expired"}
	calls := 0
	c := &countingExpire{scriptedRun: s, fail: 2, calls: &calls}
	if code := followRun(context.Background(), c, f.term(), "shop.example.com", "run_a", f.now.Add(time.Second)); code != 3 || calls != 3 {
		t.Fatalf("exit %d, expire calls %d", code, calls)
	}
}

// countingExpire fails its first `fail` expire calls as a lost connection.
type countingExpire struct {
	*scriptedRun
	fail  int
	calls *int
}

func (c *countingExpire) expire(ctx context.Context, id string) (string, error) {
	*c.calls++
	if *c.calls <= c.fail {
		return "", errors.New("connection reset")
	}
	return c.scriptedRun.expire(ctx, id)
}

func TestFollowRunStatesWithoutAResult(t *testing.T) {
	for state, want := range map[string]struct {
		code int
		says string
	}{
		"expired":   {3, "shop.example.com: not picked up within 60 seconds. the run did not start. try again"},
		"cancelled": {130, "shop.example.com: run run_a was cancelled before it started"},
	} {
		f := newTermFx()
		s := &scriptedRun{pages: []runs.Page{{State: state}}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != want.code || !strings.Contains(f.err.String(), want.says) {
			t.Errorf("%s: exit %d stderr %q", state, code, f.err.String())
		}
	}
}

// A page with no state and no result is a run the record no longer names.
// A result may still arrive, so the loop asks for a while, then gives up
// with RUN-C86 instead of polling until the sweep deletes the run.
func TestFollowRunGivesUpOnARunTheRecordLetGo(t *testing.T) {
	t.Run("no result comes", func(t *testing.T) {
		f := newTermFx()
		start := f.now
		s := &scriptedRun{pages: []runs.Page{{Chunks: []runs.Chunk{chunk(1, "o", "part\n")}}, {}}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 125 {
			t.Fatalf("exit %d stderr %q", code, f.err.String())
		}
		if got := f.err.String(); got != "shop.example.com: the instance no longer holds run run_a, and no result came back. it may have done part of its work\n" {
			t.Fatalf("stderr %q", got)
		}
		if waited := f.now.Sub(start); waited < runPageGiveUp || waited > runPageGiveUp+2*time.Second {
			t.Fatalf("gave up after %s, want %s", waited, runPageGiveUp)
		}
		if f.out.String() != "part\n" || len(s.acked) != 0 {
			t.Fatalf("stdout %q acked %v", f.out.String(), s.acked)
		}
	})
	t.Run("a result arrives in time", func(t *testing.T) {
		f := newTermFx()
		pages := make([]runs.Page, 0, 91)
		for range 90 {
			pages = append(pages, runs.Page{})
		}
		pages = append(pages, runs.Page{Result: &runs.Result{Interrupted: runs.InterruptedLost}})
		s := &scriptedRun{pages: pages}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 125 || !strings.Contains(f.err.String(), "its box stopped reporting") {
			t.Fatalf("exit %d stderr %q", code, f.err.String())
		}
		if len(s.acked) != 1 {
			t.Fatalf("acked %v", s.acked)
		}
	})
	t.Run("a state in between starts the wait again", func(t *testing.T) {
		f := newTermFx()
		var pages []runs.Page
		for range 100 {
			pages = append(pages, runs.Page{})
		}
		pages = append(pages, runs.Page{State: "picked"})
		for range 100 {
			pages = append(pages, runs.Page{})
		}
		pages = append(pages, runs.Page{State: "done", Result: &runs.Result{Exit: 4}})
		s := &scriptedRun{pages: pages}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 4 {
			t.Fatalf("exit %d stderr %q", code, f.err.String())
		}
	})
}

func TestFollowRunPageErrors(t *testing.T) {
	for name, c := range map[string]struct {
		err  error
		code int
		says string
	}{
		"no run":      {&runConsoleError{status: http.StatusNotFound, msg: "shop.example.com has no run run_a"}, 2, "shop.example.com has no run run_a, or this console has no carlos run yet"},
		"console 4xx": {&runConsoleError{status: http.StatusConflict, msg: "carlos run is not available here"}, 2, "carlos run: carlos run is not available here"},
		"network":     {errors.New("connection refused"), 1, "carlos run: connection refused"},
	} {
		f := newTermFx()
		s := &scriptedRun{pageErr: c.err}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != c.code || f.err.String() != c.says+"\n" {
			t.Errorf("%s: exit %d stderr %q", name, code, f.err.String())
		}
	}
}

func TestFollowRunRetriesAPageThatMayPass(t *testing.T) {
	done := runs.Page{State: "done", Chunks: []runs.Chunk{chunk(1, "o", "out\n")}, Result: &runs.Result{Exit: 9, LastSeq: 1}}
	t.Run("a 503 then success keeps following", func(t *testing.T) {
		f := newTermFx()
		start := f.now
		s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{errors.New("could not read the run. try again"), errors.New("connection reset")}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 {
			t.Fatalf("exit %d stderr %q", code, f.err.String())
		}
		if f.out.String() != "out\n" || f.err.String() != "" || len(s.acked) != 1 || f.now.Sub(start) != 2*time.Second {
			t.Fatalf("stdout %q stderr %q acked %v waited %s", f.out.String(), f.err.String(), s.acked, f.now.Sub(start))
		}
	})
	t.Run("a 429 waits as long as Retry-After says", func(t *testing.T) {
		f := newTermFx()
		start := f.now
		s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{&runConsoleError{status: http.StatusTooManyRequests, msg: "too many run requests. wait a minute and try again", retryAfter: 17 * time.Second}}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 || f.now.Sub(start) != 17*time.Second {
			t.Fatalf("exit %d waited %s", code, f.now.Sub(start))
		}
	})
	t.Run("a 429 without Retry-After waits longer than a second, and no longer than a minute", func(t *testing.T) {
		f := newTermFx()
		start := f.now
		s := &scriptedRun{pages: []runs.Page{done}, pageErrs: []error{
			&runConsoleError{status: http.StatusTooManyRequests, msg: "slow down"},
			&runConsoleError{status: http.StatusTooManyRequests, msg: "slow down", retryAfter: time.Hour},
		}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 || f.now.Sub(start) != runRateWait+runRateWaitMax {
			t.Fatalf("exit %d waited %s", code, f.now.Sub(start))
		}
	})
	t.Run("a page that keeps failing is given up on after two minutes with exit 1", func(t *testing.T) {
		f := newTermFx()
		start := f.now
		s := &scriptedRun{pageErr: errors.New("could not read the run. try again")}
		code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
		if waited := f.now.Sub(start); code != 1 || waited < 2*time.Minute || waited > 2*time.Minute+5*time.Second || f.err.String() != "carlos run: could not read the run. try again\n" {
			t.Fatalf("exit %d waited %s stderr %q", code, waited, f.err.String())
		}
	})
	t.Run("the bound counts failures in a row", func(t *testing.T) {
		f := newTermFx()
		var errs []error
		for range 100 {
			errs = append(errs, errors.New("blip"))
		}
		// 100s of failures, a page that works, 100s more, then the result:
		// 200s in all, but never two minutes in a row.
		all := append(append(append([]error{}, errs...), nil), errs...)
		s := &scriptedRun{pageErrs: append(all, nil), pages: []runs.Page{{State: "picked"}, done}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 9 {
			t.Fatalf("exit %d stderr %q", code, f.err.String())
		}
	})
	t.Run("a refusal that cannot pass is not retried", func(t *testing.T) {
		f := newTermFx()
		s := &scriptedRun{pageErrs: []error{&runConsoleError{status: http.StatusForbidden, msg: "no"}}, pages: []runs.Page{done}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 2 || f.err.String() != "carlos run: no\n" {
			t.Fatalf("exit %d stderr %q", code, f.err.String())
		}
	})
	t.Run("a failing page does not stop Ctrl-C", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		f.sig <- os.Interrupt
		s := &scriptedRun{pageErr: errors.New("connection refused")}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels < 1 {
			t.Fatalf("exit %d cancels %d", code, s.cancels)
		}
	})
}

func TestFollowRunPacesAPageCutShort(t *testing.T) {
	f := newTermFx()
	start := f.now
	res := &runs.Result{LastSeq: 3}
	s := &scriptedRun{pages: []runs.Page{
		{Chunks: []runs.Chunk{chunk(1, "o", "a")}, Result: res, More: true},
		{Chunks: []runs.Chunk{chunk(2, "o", "b")}, Result: res, More: true},
		{Chunks: []runs.Chunk{chunk(3, "o", "c")}, Result: res},
	}}
	followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
	if f.out.String() != "abc" || f.now.Sub(start) != 2*runMorePause {
		t.Fatalf("stdout %q paused %s", f.out.String(), f.now.Sub(start))
	}
}

func TestRunAPIPageReadsRetryAfter(t *testing.T) {
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Retry-After", "7")
		w.WriteHeader(http.StatusTooManyRequests)
		json.NewEncoder(w).Encode(map[string]string{"error": "too many run requests. wait a minute and try again"})
	}))
	defer ts.Close()
	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "shop.example.com"}
	_, err := c.page(context.Background(), "run_a", 0)
	var ce *runConsoleError
	if !errors.As(err, &ce) || ce.status != http.StatusTooManyRequests || ce.retryAfter != 7*time.Second {
		t.Fatalf("%#v", err)
	}
}

func TestFollowRunAck(t *testing.T) {
	notDone := &runConsoleError{status: http.StatusConflict, msg: "run run_a has not finished yet"}
	gone := &runConsoleError{status: http.StatusNotFound, msg: "shop.example.com has no run run_a"}
	page := runs.Page{State: "done", Result: &runs.Result{Exit: 6, LastSeq: 2}}
	t.Run("a lost 200 retried finds the objects gone and is success", func(t *testing.T) {
		f := newTermFx()
		s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{gone}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 || f.err.String() != "" || len(s.acked) != 1 {
			t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked)
		}
	})
	t.Run("an ack before the box's done is retried", func(t *testing.T) {
		f := newTermFx()
		s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{notDone, notDone}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 || f.err.String() != "" || len(s.acked) != 3 {
			t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked)
		}
	})
	t.Run("a box that never finishes is given up on, the exit code stands", func(t *testing.T) {
		f := newTermFx()
		errs := make([]error, 100)
		for i := range errs {
			errs[i] = notDone
		}
		s := &scriptedRun{pages: []runs.Page{page}, ackErrs: errs}
		code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
		if code != 6 || len(s.acked) < 2 || len(s.acked) > 20 || !strings.Contains(f.err.String(), "carlos run: run run_a has not finished yet") {
			t.Fatalf("exit %d acked %d stderr %q", code, len(s.acked), f.err.String())
		}
	})
	t.Run("a refused ack is said and the exit code stands", func(t *testing.T) {
		f := newTermFx()
		s := &scriptedRun{pages: []runs.Page{page}, ackErrs: []error{&runConsoleError{status: http.StatusConflict, msg: "run run_a ended at chunk 2, not 3"}}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 6 || len(s.acked) != 1 || f.err.String() != "carlos run: run run_a ended at chunk 2, not 3\n" {
			t.Fatalf("exit %d stderr %q acked %v", code, f.err.String(), s.acked)
		}
	})
}

func TestFollowRunInterrupts(t *testing.T) {
	queued := runs.Page{State: "picked"}
	done := runs.Page{State: "done", Result: &runs.Result{Cancelled: true, Started: time.Unix(1_700_000_000, 0)}}
	t.Run("the first cancels and keeps following", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		s := &scriptedRun{pages: []runs.Page{queued, done}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 1 {
			t.Fatalf("exit %d cancels %d", code, s.cancels)
		}
		if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\nshop.example.com: the command was cancelled\n" {
			t.Fatalf("stderr %q", got)
		}
	})
	t.Run("the second stops waiting", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		f.sig <- os.Interrupt
		s := &scriptedRun{pages: []runs.Page{queued}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 1 || len(s.acked) != 0 {
			t.Fatalf("exit %d cancels %d acked %v", code, s.cancels, s.acked)
		}
	})
	t.Run("a cancel that finds the run already ended is silent success", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		s := &scriptedRun{pages: []runs.Page{queued, done}, cancelErr: &runConsoleError{status: http.StatusConflict, msg: "run run_a has already ended"}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 {
			t.Fatalf("exit %d", code)
		}
		if got := f.err.String(); strings.Contains(got, "already ended") {
			t.Fatalf("stderr %q", got)
		}
	})
	t.Run("a cancel that did not arrive is sent again on the next poll, and said once", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		s := &scriptedRun{pages: []runs.Page{queued, queued, queued, done}, cancelErrs: []error{errors.New("connection reset"), errors.New("connection reset"), nil}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 3 {
			t.Fatalf("exit %d cancels %d", code, s.cancels)
		}
		if got := strings.Count(f.err.String(), "carlos run: connection reset"); got != 1 {
			t.Fatalf("said %d times: %q", got, f.err.String())
		}
	})
	t.Run("the second Ctrl-C tries an unanswered cancel once more", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		f.sig <- os.Interrupt
		s := &scriptedRun{pages: []runs.Page{queued}, cancelErrs: []error{errors.New("connection reset"), nil}}
		if code := followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}); code != 130 || s.cancels != 2 {
			t.Fatalf("exit %d cancels %d", code, s.cancels)
		}
	})
	t.Run("a cancel that failed is said", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		s := &scriptedRun{pages: []runs.Page{queued, done}, cancelErr: errors.New("connection reset")}
		followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{})
		if !strings.Contains(f.err.String(), "carlos run: connection reset") {
			t.Fatalf("stderr %q", f.err.String())
		}
	})
}

// gatedRun holds one request open, as a request in flight: entered is closed
// when the held call arrives, and the call returns when release is closed or
// its context is cancelled, as a real request does.
type gatedRun struct {
	*scriptedRun
	holdPage   int           // the page call (1-based) to hold, 0 for none
	holdCancel int           // the cancel call (1-based) to hold, 0 for none
	holdAck    bool          // hold the first ack call
	entered    chan struct{} // the held page or ack call has arrived
	cancelIn   chan struct{} // the held cancel call has arrived
	release    chan struct{}
	ctxErr     error // the held page's context error when it returned
	cancelCtx  error // the held cancel's context error as it arrived
	acksHeld   int
}

func (g *gatedRun) hold(ctx context.Context, entered chan struct{}) error {
	if entered == g.cancelIn {
		g.cancelCtx = ctx.Err()
	}
	close(entered)
	select {
	case <-g.release:
	case <-ctx.Done():
	}
	g.ctxErr = ctx.Err()
	return g.ctxErr
}

func (g *gatedRun) page(ctx context.Context, id string, after int64) (runs.Page, error) {
	if g.pagesRead()+1 == g.holdPage {
		if err := g.hold(ctx, g.entered); err != nil {
			g.scriptedRun.mu.Lock()
			g.scriptedRun.pageCalls++
			g.scriptedRun.mu.Unlock()
			return runs.Page{}, err
		}
	}
	return g.scriptedRun.page(ctx, id, after)
}

func (g *gatedRun) cancel(ctx context.Context, id string) (string, error) {
	if g.holdCancel != 0 && g.cancelsMade()+1 == g.holdCancel {
		g.scriptedRun.mu.Lock()
		g.scriptedRun.cancels++
		g.scriptedRun.mu.Unlock()
		return "", g.hold(ctx, g.cancelIn)
	}
	return g.scriptedRun.cancel(ctx, id)
}

func (g *gatedRun) ack(ctx context.Context, id string, last int64) error {
	if g.holdAck && g.acksHeld == 0 {
		g.acksHeld++
		g.scriptedRun.acked = append(g.scriptedRun.acked, last)
		return g.hold(ctx, g.entered)
	}
	return g.scriptedRun.ack(ctx, id, last)
}

func newGatedRun(s *scriptedRun) *gatedRun {
	return &gatedRun{scriptedRun: s, entered: make(chan struct{}), cancelIn: make(chan struct{}), release: make(chan struct{})}
}

func awaitFollowRun(t *testing.T, done <-chan int, what string) int {
	t.Helper()
	select {
	case code := <-done:
		return code
	case <-time.After(2 * time.Second):
		t.Fatal(what)
		return 0
	}
}

func awaitSignal(t *testing.T, ch <-chan struct{}, what string) {
	t.Helper()
	select {
	case <-ch:
	case <-time.After(2 * time.Second):
		t.Fatal(what)
	}
}

func TestFollowRunCtrlCOnceCancelsTwiceExits(t *testing.T) {
	f := newTermFx()
	s := &scriptedRun{pages: []runs.Page{{State: "picked"}}}
	done := make(chan int, 1)
	go func() { done <- followRun(context.Background(), s, f.term(), "shop.example.com", "run_a", time.Time{}) }()
	f.sig <- os.Interrupt
	deadline := time.Now().Add(2 * time.Second)
	for s.cancelsMade() == 0 && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	if got := s.cancelsMade(); got != 1 {
		t.Fatalf("the first Ctrl-C made %d cancels, want 1", got)
	}
	// The first Ctrl-C does not end the wait: the run is still being polled.
	seen := s.pagesRead()
	for s.pagesRead() < seen+3 && time.Now().Before(deadline) {
		time.Sleep(5 * time.Millisecond)
	}
	if s.pagesRead() < seen+3 {
		t.Fatal("followRun stopped polling after the first Ctrl-C")
	}
	select {
	case code := <-done:
		t.Fatalf("followRun returned %d after one Ctrl-C", code)
	default:
	}
	f.sig <- os.Interrupt
	if code := awaitFollowRun(t, done, "the second Ctrl-C did not stop waiting"); code != 130 {
		t.Fatalf("exit %d, want 130", code)
	}
	if got := s.cancelsMade(); got != 1 {
		t.Fatalf("%d cancels, want 1: the second Ctrl-C does not cancel again", got)
	}
	if len(s.acked) != 0 {
		t.Fatalf("acked %v: nothing was read to the end", s.acked)
	}
	if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
		t.Fatalf("stderr %q", got)
	}
}

func TestFollowRunCtrlCWhileARequestIsInFlight(t *testing.T) {
	picked := runs.Page{State: "picked"}
	done := runs.Page{State: "done", Result: &runs.Result{Cancelled: true, Started: time.Unix(1_700_000_000, 0)}}
	start := func(g *gatedRun, f *termFx) <-chan int {
		codes := make(chan int, 1)
		go func() {
			codes <- followRun(context.Background(), g, f.term(), "shop.example.com", "run_a", time.Time{})
		}()
		return codes
	}
	// promptly is how long a second Ctrl-C may take to end a request that
	// never answers; the held request would otherwise wait for ever.
	const promptly = 500 * time.Millisecond
	quick := func(t *testing.T, codes <-chan int) int {
		t.Helper()
		select {
		case code := <-codes:
			return code
		case <-time.After(promptly):
			t.Fatal("the second Ctrl-C did not end the request in flight")
			return 0
		}
	}
	t.Run("the first is taken when the request returns, before the next poll", func(t *testing.T) {
		f := newTermFx()
		g := newGatedRun(&scriptedRun{pages: []runs.Page{picked, done}})
		g.holdPage = 1
		codes := start(g, f)
		awaitSignal(t, g.entered, "the first page request was never made")
		f.sig <- os.Interrupt
		// Wait until the watcher has taken the Ctrl-C, then give the loop a
		// moment to act on it: it must not, while the request is open.
		for end := time.Now().Add(2 * time.Second); len(f.sig) > 0 && time.Now().Before(end); {
			time.Sleep(time.Millisecond)
		}
		time.Sleep(50 * time.Millisecond)
		// The request is not abandoned, and nothing is cancelled until it is back.
		if got := g.cancelsMade(); got != 0 {
			t.Fatalf("%d cancels while the page request was in flight", got)
		}
		close(g.release)
		if code := awaitFollowRun(t, codes, "the run was not followed to its result"); code != 130 {
			t.Fatalf("exit %d, want 130", code)
		}
		if g.ctxErr != nil {
			t.Fatalf("the request in flight was cut off: %v", g.ctxErr)
		}
		if got := g.cancelsMade(); got != 1 {
			t.Fatalf("%d cancels, want 1", got)
		}
		if got := g.pagesRead(); got != 2 {
			t.Fatalf("%d page requests, want 2: the one in flight and the one that carries the result", got)
		}
		if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\nshop.example.com: the command was cancelled\n" {
			t.Fatalf("stderr %q", got)
		}
	})
	t.Run("the second ends a page request in flight at once", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt // the first, handled after the first page
		g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
		g.holdPage = 2
		codes := start(g, f)
		awaitSignal(t, g.entered, "the second page request was never made")
		if got := g.cancelsMade(); got != 1 {
			t.Fatalf("%d cancels before the second page request, want 1", got)
		}
		f.sig <- os.Interrupt // the request is never released
		if code := quick(t, codes); code != 130 {
			t.Fatalf("exit %d, want 130", code)
		}
		if !errors.Is(g.ctxErr, context.Canceled) {
			t.Fatalf("the request in flight ended with %v, want it cut off", g.ctxErr)
		}
		if got := g.pagesRead(); got != 2 {
			t.Fatalf("%d page requests, want 2: no poll after the second Ctrl-C", got)
		}
		if got := g.cancelsMade(); got != 1 {
			t.Fatalf("%d cancels, want 1: the one already sent stays sent and is not repeated", got)
		}
		if len(g.acked) != 0 {
			t.Fatalf("acked %v", g.acked)
		}
		if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
			t.Fatalf("stderr %q", got)
		}
	})
	t.Run("both arriving during one page request still cancel, then stop", func(t *testing.T) {
		f := newTermFx()
		g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
		g.holdPage = 1
		codes := start(g, f)
		awaitSignal(t, g.entered, "the first page request was never made")
		f.sig <- os.Interrupt
		f.sig <- os.Interrupt
		if code := quick(t, codes); code != 130 {
			t.Fatalf("exit %d, want 130", code)
		}
		if got := g.cancelsMade(); got != 1 {
			t.Fatalf("%d cancels, want 1: the first Ctrl-C still asks for the cancel", got)
		}
		if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
			t.Fatalf("stderr %q", got)
		}
	})
	t.Run("the second ends a cancel request in flight and does not send it again", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
		g.holdCancel = 1
		codes := start(g, f)
		awaitSignal(t, g.cancelIn, "the cancel request was never made")
		f.sig <- os.Interrupt
		if code := quick(t, codes); code != 130 {
			t.Fatalf("exit %d, want 130", code)
		}
		if got := g.cancelsMade(); got != 1 {
			t.Fatalf("%d cancels, want 1", got)
		}
		if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
			t.Fatalf("stderr %q: a cut-off cancel is not an error to report", got)
		}
	})
	t.Run("both during the ack end the ack, not the exit code, and cancel nothing", func(t *testing.T) {
		f := newTermFx()
		g := newGatedRun(&scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{LastSeq: 3}}}})
		g.holdAck = true
		codes := start(g, f)
		awaitSignal(t, g.entered, "the ack request was never made")
		f.sig <- os.Interrupt
		f.sig <- os.Interrupt
		if code := quick(t, codes); code != 0 {
			t.Fatalf("exit %d, want the command's own 0", code)
		}
		if got := g.cancelsMade(); got != 0 {
			t.Fatalf("%d cancels sent to a finished run", got)
		}
		if got := f.err.String(); got != "" {
			t.Fatalf("stderr %q", got)
		}
	})
	t.Run("a second Ctrl-C during the cancel resend ends the wait before another request", func(t *testing.T) {
		f := newTermFx()
		f.sig <- os.Interrupt
		g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}, cancelErrs: []error{errors.New("connection reset")}})
		g.holdCancel = 2 // the first cancel fails fast; its resend never answers
		g.holdPage = 3   // any page request after that would hang too
		codes := start(g, f)
		awaitSignal(t, g.cancelIn, "the cancel was never sent again")
		f.sig <- os.Interrupt
		if code := quick(t, codes); code != 130 {
			t.Fatalf("exit %d, want 130", code)
		}
		if got := g.pagesRead(); got != 2 {
			t.Fatalf("%d page requests, want 2: none after the second Ctrl-C", got)
		}
	})
	late := func(t *testing.T) {
		old := runLateCancelWait
		runLateCancelWait = 100 * time.Millisecond
		t.Cleanup(func() { runLateCancelWait = old })
	}
	t.Run("a cancel first sent after the second Ctrl-C is attempted, but only briefly", func(t *testing.T) {
		late(t)
		f := newTermFx()
		g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}})
		g.holdPage = 1
		g.holdCancel = 1 // never answers
		codes := start(g, f)
		awaitSignal(t, g.entered, "the page request was never made")
		f.sig <- os.Interrupt
		f.sig <- os.Interrupt
		select {
		case code := <-codes:
			if code != 130 {
				t.Fatalf("exit %d, want 130", code)
			}
		case <-time.After(time.Second):
			t.Fatal("a cancel sent after the second Ctrl-C held the exit")
		}
		if got := g.cancelsMade(); got != 1 {
			t.Fatalf("%d cancels, want 1: still attempted", got)
		}
		if g.cancelCtx != nil {
			t.Fatalf("the cancel was sent on a context already %v: it was not attempted", g.cancelCtx)
		}
		if got := f.err.String(); got != "cancelling run run_a. press Ctrl-C again to stop waiting\n" {
			t.Fatalf("stderr %q", got)
		}
	})
	t.Run("a cancel resent on the second Ctrl-C is attempted, but only briefly", func(t *testing.T) {
		late(t)
		f := newTermFx()
		f.sig <- os.Interrupt
		g := newGatedRun(&scriptedRun{pages: []runs.Page{picked}, cancelErrs: []error{errors.New("connection reset")}})
		g.holdPage = 2   // the second Ctrl-C arrives during this page request
		g.holdCancel = 2 // the resend never answers
		codes := start(g, f)
		awaitSignal(t, g.entered, "the second page request was never made")
		f.sig <- os.Interrupt
		select {
		case code := <-codes:
			if code != 130 {
				t.Fatalf("exit %d, want 130", code)
			}
		case <-time.After(time.Second):
			t.Fatal("a cancel resent after the second Ctrl-C held the exit")
		}
		if got := g.cancelsMade(); got != 2 {
			t.Fatalf("%d cancels, want 2: the resend is still attempted", got)
		}
		if g.cancelCtx != nil {
			t.Fatalf("the resend was sent on a context already %v: it was not attempted", g.cancelCtx)
		}
	})
}

func TestStartRunPrintsTheFollowCommand(t *testing.T) {
	f := newTermFx()
	s := &scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{Exit: 0}}}}
	o := runOpts{instance: "shop.example.com", argv: []string{"import"}, timeout: time.Minute}
	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 0 {
		t.Fatalf("exit %d", code)
	}
	if !strings.Contains(f.err.String(), "run run_a on shop.example.com. to follow it again: carlos run --app shop --instance shop.example.com --follow run_a") {
		t.Fatalf("stderr %q", f.err.String())
	}
	if s.created.TimeoutS != 60 {
		t.Fatalf("meta %+v", s.created)
	}
}

func TestStartRunFailures(t *testing.T) {
	missing := filepath.Join(t.TempDir(), "absent.csv")
	for name, c := range map[string]struct {
		o         runOpts
		createErr error
		code      int
		says      string
	}{
		"a file that cannot be read": {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute, files: []string{missing}}, nil, 2, "could not read " + missing},
		"a console refusal":          {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute}, &runConsoleError{status: http.StatusConflict, msg: "an instance can run one command at a time"}, 2, "carlos run: an instance can run one command at a time"},
		"a lost connection":          {runOpts{instance: "shop.example.com", argv: []string{"x"}, timeout: time.Minute}, errors.New("connection refused"), 1, "carlos run: connection refused"},
	} {
		f := newTermFx()
		s := &scriptedRun{createErr: c.createErr}
		if code := startRun(context.Background(), s, f.term(), c.o, "shop"); code != c.code || !strings.Contains(f.err.String(), c.says) || strings.Contains(f.err.String(), "to follow it again") {
			t.Errorf("%s: exit %d stderr %q", name, code, f.err.String())
		}
	}
}

// A keyed run sends the key as the Idempotency-Key header, and declares each
// file's sha256 in the meta so the console can tell a retry of the same
// request from a different one under the same key.
func TestRunIdempotencyKeyFlag(t *testing.T) {
	o, err := parseRunArgs([]string{"--instance", "h.example.com", "--idempotency-key", "import-42:a.b_c", "--", "import"}, io.Discard)
	if err != nil || o.idempotencyKey != "import-42:a.b_c" {
		t.Fatalf("opts %+v, %v", o, err)
	}
	dir := t.TempDir()
	p := filepath.Join(dir, "x.csv")
	os.WriteFile(p, []byte("1,2"), 0o600)
	sum := sha256.Sum256([]byte("1,2"))
	want := hex.EncodeToString(sum[:])

	m, err := runMetaFor(runOpts{argv: []string{"import"}, timeout: time.Minute, files: []string{p}, idempotencyKey: "k1"})
	if err != nil || len(m.Files) != 1 || m.Files[0].SHA256 != want {
		t.Fatalf("keyed meta %+v, %v, want sha256 %s", m, err, want)
	}
	m, err = runMetaFor(runOpts{argv: []string{"import"}, timeout: time.Minute, files: []string{p}})
	if err != nil || m.Files[0].SHA256 != "" {
		t.Fatalf("an unkeyed run declares no hash: %+v, %v", m, err)
	}

	var header, metaJSON string
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		header = r.Header.Get("Idempotency-Key")
		mr, err := r.MultipartReader()
		if err != nil {
			t.Error(err)
			return
		}
		for {
			part, err := mr.NextPart()
			if err != nil {
				break
			}
			raw, _ := io.ReadAll(part)
			if part.FormName() == "meta" {
				metaJSON = string(raw)
			}
		}
		w.WriteHeader(http.StatusAccepted)
		json.NewEncoder(w).Encode(runs.Created{ID: "run_a", Host: "h.example.com", State: "queued"})
	}))
	defer ts.Close()
	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com", key: "k1"}
	meta, err := runMetaFor(runOpts{argv: []string{"import"}, timeout: time.Minute, files: []string{p}, idempotencyKey: "k1"})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := c.create(context.Background(), meta, []string{p}); err != nil {
		t.Fatal(err)
	}
	if header != "k1" || !strings.Contains(metaJSON, `"sha256":"`+want+`"`) {
		t.Fatalf("header %q, meta %s", header, metaJSON)
	}
	c.key = ""
	header = "unset"
	if _, err := c.create(context.Background(), meta, []string{p}); err != nil || header != "" {
		t.Fatalf("an unkeyed create sent the header %q, %v", header, err)
	}
}

// An invalid key is refused by the CLI, before any network call: runRun
// parses its arguments before it resolves a console.
func TestRunIdempotencyKeyIsCheckedBeforeAnyCall(t *testing.T) {
	for name, key := range map[string]string{
		"a space":   "two words",
		"a slash":   "a/b",
		"too long":  strings.Repeat("k", 129),
		"non-ascii": "klé",
	} {
		_, err := parseRunArgs([]string{"--instance", "h.example.com", "--idempotency-key", key, "--", "x"}, io.Discard)
		if err == nil || !strings.Contains(err.Error(), "--idempotency-key") {
			t.Errorf("%s: %v, want a refusal naming --idempotency-key", name, err)
		}
	}
	// An empty value is the same as not passing the flag.
	if o, err := parseRunArgs([]string{"--instance", "h.example.com", "--idempotency-key", "", "--", "x"}, io.Discard); err != nil || o.idempotencyKey != "" {
		t.Errorf("empty key: %+v, %v", o, err)
	}
	// A key belongs to starting a run; --follow starts none.
	if _, err := parseRunArgs([]string{"--instance", "h.example.com", "--follow", "run_x", "--idempotency-key", "k"}, io.Discard); err == nil {
		t.Error("--follow with a key accepted")
	}
}

// keyedConsole is a console that admits a keyed create once: the second
// create under the same key answers 202 with the same run and replayed.
type keyedConsole struct {
	mu       sync.Mutex
	creates  int // create requests made
	started  int // runs started
	admitted map[string]string
	pages    []string // run ids paged
}

func (k *keyedConsole) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	k.mu.Lock()
	defer k.mu.Unlock()
	if r.Method == http.MethodPost && strings.HasPrefix(r.Header.Get("Content-Type"), "multipart/") {
		k.creates++
		key := r.Header.Get("Idempotency-Key")
		if id, ok := k.admitted[key]; ok {
			w.WriteHeader(http.StatusAccepted)
			json.NewEncoder(w).Encode(runs.Created{ID: id, Host: "h.example.com", State: "done", Replayed: true})
			return
		}
		k.started++
		id := "run_" + strconv.Itoa(k.started)
		k.admitted[key] = id
		w.WriteHeader(http.StatusAccepted)
		json.NewEncoder(w).Encode(runs.Created{ID: id, Host: "h.example.com", State: "queued"})
		return
	}
	if r.Method == http.MethodGet {
		k.pages = append(k.pages, r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:])
		json.NewEncoder(w).Encode(runs.Page{State: "done", Result: &runs.Result{Exit: 0}})
		return
	}
	json.NewEncoder(w).Encode(runs.StateAnswer{State: "done"}) // ack
}

func TestRunCLIRetryFollowsTheReplayedRun(t *testing.T) {
	k := &keyedConsole{admitted: map[string]string{}}
	ts := httptest.NewServer(k)
	defer ts.Close()
	o := runOpts{instance: "h.example.com", argv: []string{"import"}, timeout: time.Minute, idempotencyKey: "import-42"}
	c := runAPI{s: newTestSession(ts.URL), a: apiAcct{Seg: "acme"}, app: "shop", host: "h.example.com", key: o.idempotencyKey}

	first := newTermFx()
	if code := startRun(context.Background(), c, first.term(), o, "shop"); code != 0 {
		t.Fatalf("first: exit %d, stderr %q", code, first.err.String())
	}
	if strings.Contains(first.err.String(), "already started") || !strings.Contains(first.err.String(), "run run_1 on h.example.com. to follow it again") {
		t.Fatalf("a fresh run says %q", first.err.String())
	}

	second := newTermFx()
	if code := startRun(context.Background(), c, second.term(), o, "shop"); code != 0 {
		t.Fatalf("retry: exit %d, stderr %q", code, second.err.String())
	}
	if !strings.Contains(second.err.String(), "already started: run run_1 on h.example.com. to follow it again: carlos run --app shop --instance h.example.com --follow run_1") {
		t.Fatalf("the retry says %q", second.err.String())
	}
	if k.creates != 2 || k.started != 1 {
		t.Fatalf("%d creates, %d runs started: want 2 and 1", k.creates, k.started)
	}
	if want := []string{"run_1", "run_1"}; !slices.Equal(k.pages, want) {
		t.Fatalf("followed %v, want %v: the retry follows the same run", k.pages, want)
	}
}

// replayedRun is a console whose create always answers a replay of run_a
// with the given state.
type replayedRun struct {
	*scriptedRun
	state string
}

func (r replayedRun) create(context.Context, runs.Meta, []string) (runs.Created, error) {
	return runs.Created{ID: "run_a", Host: "shop.example.com", State: r.state, Replayed: true}, nil
}

// A retry whose key started a run that was since acknowledged or swept has
// nothing to follow: it says so, rather than the generic "has no run".
func TestRunCLIRetryOfARunWhoseOutputIsGone(t *testing.T) {
	o := runOpts{instance: "shop.example.com", argv: []string{"import"}, timeout: time.Minute, idempotencyKey: "k"}
	f := newTermFx()
	s := replayedRun{scriptedRun: &scriptedRun{pageErr: &runConsoleError{status: http.StatusNotFound, msg: "no such run"}}}
	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 2 {
		t.Fatalf("exit %d, stderr %q", code, f.err.String())
	}
	got := f.err.String()
	if !strings.Contains(got, "run run_a was already started with this key, and its output is no longer kept") ||
		strings.Contains(got, "has no run") || strings.Contains(got, "to follow it again") {
		t.Fatalf("stderr %q", got)
	}

	// A replay the record still names is followed, with no probe.
	f = newTermFx()
	s = replayedRun{scriptedRun: &scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{Exit: 0}}}}, state: "done"}
	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 0 || s.pagesRead() != 1 {
		t.Fatalf("named replay: exit %d, %d pages, stderr %q", code, s.pagesRead(), f.err.String())
	}
	// A replay the record no longer names, whose output is still there, is
	// followed too.
	f = newTermFx()
	s = replayedRun{scriptedRun: &scriptedRun{pages: []runs.Page{{State: "done", Result: &runs.Result{Exit: 0}}}}}
	if code := startRun(context.Background(), s, f.term(), o, "shop"); code != 0 || !strings.Contains(f.err.String(), "already started: run run_a") {
		t.Fatalf("unnamed replay: exit %d, stderr %q", code, f.err.String())
	}
}