package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"mime/multipart"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

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

const runUsageRun = "usage: carlos run --app <app> --instance <host> [--file <path>]... [--timeout 30m] -- <arg>..."

const runUsageFollow = "carlos run --app <app> --instance <host> --follow <id>"

const runUsageKey = "add --idempotency-key <key> to start a run you can safely run again: the same key and request follow the run already started"

func runUsageText() string {
	return runUsageRun + "\n       " + runUsageFollow + "\n       " + runUsageKey + "\n"
}

const runDirectModeMsg = "carlos run works through a console, and this shell is in direct bucket mode. unset CARLOS_DEPLOYMENT_BUCKET and CARLOS_DEPLOYMENT_DIR, then run carlos auth login --console <url>"

// errRunUsage is a command line that does not parse: the usage text says
// what was wrong.
var errRunUsage = errors.New("carlos run: usage")

// runOpts is `carlos run` parsed.
type runOpts struct {
	app, instance, follow, console string
	idempotencyKey                 string
	acctRaw, acctIDRaw             *string
	files                          []string
	timeout                        time.Duration
	argv                           []string
}

// parseRunArgs splits at the first "--": carlos's flags before it, the
// command after it. Nothing may sit between the flags and the "--", so a
// forgotten "--" is a usage error rather than a command whose flags carlos
// swallowed.
func parseRunArgs(args []string, stderr io.Writer) (runOpts, error) {
	flagArgs, argv := args, []string(nil)
	if i := slices.Index(args, "--"); i >= 0 {
		flagArgs, argv = args[:i], args[i+1:]
	}
	fs := flag.NewFlagSet("run", flag.ContinueOnError)
	fs.SetOutput(stderr)
	app := appFlag(fs, "the app whose command to run")
	instance := fs.String("instance", "", "the instance to run it against")
	var files multiFlag
	fs.Var(&files, "file", "a file to upload with the run (repeatable; 64 MB in total)")
	timeout := fs.Duration("timeout", runs.DefaultTimeout, "how long the command may run (1s to 6h)")
	follow := fs.String("follow", "", "attach to a run already started, by its id")
	key := fs.String("idempotency-key", "", "a key of your choosing: run again with the same key and the same request, and you follow the run already started instead of starting a second")
	acctRaw, acctIDRaw, _ := accountFlags(fs)
	console := consoleOption(fs, "console to run through (default: the only logged-in one)")
	fs.Usage = flagUsage(fs, runUsageText())
	if err := fs.Parse(flagArgs); err != nil {
		return runOpts{}, err
	}
	if fs.NArg() > 0 {
		return runOpts{}, errRunUsage
	}
	o := runOpts{app: strings.TrimSpace(*app), instance: strings.TrimSpace(*instance), follow: strings.TrimSpace(*follow),
		console: *console, acctRaw: acctRaw, acctIDRaw: acctIDRaw, files: files, timeout: *timeout, argv: argv,
		idempotencyKey: *key}
	switch {
	case o.instance == "":
		return o, errors.New("--app and --instance name the instance to run against")
	case o.follow != "" && (len(o.argv) > 0 || len(o.files) > 0):
		return o, errors.New("--follow takes a run id and no command after --")
	case runs.CheckTimeout(o.timeout) != nil:
		return o, errors.New("--timeout must be between 1s and 6h")
	case o.idempotencyKey != "" && !runs.ValidKey(o.idempotencyKey):
		return o, errors.New(runKeyShapeMsg)
	case o.idempotencyKey != "" && o.follow != "":
		return o, errors.New("--idempotency-key is for starting a run. --follow starts none")
	}
	return o, nil
}

// runKeyShapeMsg is what the CLI says of a key that the console would refuse.
const runKeyShapeMsg = "--idempotency-key is 1 to 128 characters: letters, digits, dots, underscores, colons and hyphens"

// errRunNotRegular is a --file that is a directory or a device: only a
// regular file has a size to promise the console.
var errRunNotRegular = errors.New("not a regular file")

// errRunFileChanged is a --file whose size is no longer what meta promised.
var errRunFileChanged = errors.New("the file changed while carlos was reading it")

// errRunLocalFile is what errors.Is finds in a --file that cannot be
// uploaded: a failure on this machine (exit 2), not the console's answer.
var errRunLocalFile = errors.New("a --file could not be read")

// runFileError is a --file failure carrying the RUN-C53 text.
type runFileError struct{ err error }

func (e *runFileError) Error() string        { return e.err.Error() }
func (e *runFileError) Unwrap() error        { return e.err }
func (e *runFileError) Is(target error) bool { return target == errRunLocalFile }

// runReadError is a --file that cannot be uploaded (RUN-C53). It is a local
// failure, and either nothing has been sent or the upload is aborted before
// its last byte.
func runReadError(path string, err error) error {
	return &runFileError{fmt.Errorf("could not read %s: %w", path, err)}
}

// runMetaFor is the create request's meta: argv, the timeout, and each
// file's base name and size. A run with an idempotency key also declares
// each file's sha256: the console needs it to tell a retry of the same
// request from a different one. The console checks all of it again.
func runMetaFor(o runOpts) (runs.Meta, error) {
	m := runs.Meta{Argv: o.argv, TimeoutS: int64(o.timeout / time.Second)}
	for _, p := range o.files {
		st, err := os.Stat(p)
		if err != nil {
			return m, runReadError(p, err)
		}
		if !st.Mode().IsRegular() {
			return m, runReadError(p, errRunNotRegular)
		}
		f := runs.MetaFile{Name: filepath.Base(p), Size: st.Size()}
		if o.idempotencyKey != "" {
			if f.SHA256, err = runFileSum(p); err != nil {
				return m, runReadError(p, err)
			}
		}
		m.Files = append(m.Files, f)
	}
	return m, nil
}

// runFileSum is the hex sha256 of the file at path.
func runFileSum(path string) (string, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", err
	}
	defer f.Close()
	h := sha256.New()
	if _, err := io.Copy(h, f); err != nil {
		return "", err
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

// runClient is the console's five run routes (run spec §5.1, §5.6, §5.7).
type runClient interface {
	create(ctx context.Context, meta runs.Meta, paths []string) (runs.Created, error)
	page(ctx context.Context, id string, after int64) (runs.Page, error)
	cancel(ctx context.Context, id string) (string, error)
	expire(ctx context.Context, id string) (string, error)
	ack(ctx context.Context, id string, lastSeq int64) error
}

// runConsoleError is a 4xx the console answered, with its message: a
// refusal by the console, exit 2.
type runConsoleError struct {
	status     int
	msg        string
	retryAfter time.Duration // a 429's Retry-After, when the console sent one
}

func (e *runConsoleError) Error() string { return e.msg }

// runAnswer decodes a want-status answer into out, or turns any other
// answer into an error: a 4xx into a runConsoleError.
func runAnswer(status int, body []byte, err error, want int, out any) error {
	if err != nil {
		return err
	}
	if status != want {
		e := apiErrorFromBody(status, body)
		if status >= 400 && status < 500 {
			return &runConsoleError{status: status, msg: e.Error()}
		}
		return e
	}
	if out == nil {
		return nil
	}
	return json.Unmarshal(body, out)
}

// runAPI is runClient over the signed-in console.
type runAPI struct {
	s    *apiSession
	a    apiAcct
	app  string
	host string
	key  string // the Idempotency-Key a create carries; "" sends none
}

func (c runAPI) path(tail string) string {
	return apiPath(c.a, c.app, "/instances/"+url.PathEscape(c.host)+"/runs"+tail)
}

// create sends meta, then one file part per meta file in meta's order, each
// named as meta names it and exactly as long as meta says. A file whose size
// is no longer meta's fails the call before its part is written, or aborts
// the upload part-way: the console never gets a whole request whose parts
// disagree with its meta.
func (c runAPI) create(ctx context.Context, meta runs.Meta, paths []string) (runs.Created, error) {
	if len(paths) != len(meta.Files) {
		return runs.Created{}, errors.New("carlos run: the files to upload and the files in meta differ")
	}
	var mu sync.Mutex
	var writeErr error
	var hdr http.Header
	if c.key != "" {
		hdr = http.Header{"Idempotency-Key": {c.key}}
	}
	status, body, err := c.s.c.doMultipartHeaders(ctx, http.MethodPost, c.path(""), hdr, func(mw *multipart.Writer) (err error) {
		defer func() {
			mu.Lock()
			writeErr = err
			mu.Unlock()
		}()
		mp, err := mw.CreateFormField("meta")
		if err != nil {
			return err
		}
		if err := json.NewEncoder(mp).Encode(meta); err != nil {
			return err
		}
		for i, p := range paths {
			if err := writeRunFilePart(mw, p, meta.Files[i]); err != nil {
				return err
			}
		}
		return nil
	})
	if err != nil {
		// The http client wraps a failed body in a *url.Error. When the
		// failure was a file the writer could not read, the user is told that,
		// not which URL the request was for.
		mu.Lock()
		we := writeErr
		mu.Unlock()
		if errors.Is(we, errRunLocalFile) {
			return runs.Created{}, we
		}
	}
	var out runs.Created
	return out, runAnswer(status, body, err, http.StatusAccepted, &out)
}

// runFileHandle is the open file writeRunFilePart reads: the reader and the
// file's current size.
type runFileHandle interface {
	io.Reader
	Stat() (os.FileInfo, error)
}

// writeRunFilePart writes one file part: exactly want.Size bytes of the file
// at path, under want.Name. The size is checked on the open file, before and
// after the copy, so a file that grew or shrank since the stat is an error
// and not a part that disagrees with meta.
func writeRunFilePart(mw *multipart.Writer, path string, want runs.MetaFile) error {
	f, err := os.Open(path)
	if err != nil {
		return runReadError(path, err)
	}
	defer f.Close()
	return copyRunFilePart(mw, f, path, want)
}

// copyRunFilePart is writeRunFilePart over an open file.
func copyRunFilePart(mw *multipart.Writer, f runFileHandle, path string, want runs.MetaFile) error {
	st, err := f.Stat()
	if err != nil {
		return runReadError(path, err)
	}
	if !st.Mode().IsRegular() {
		return runReadError(path, errRunNotRegular)
	}
	if st.Size() != want.Size {
		return runReadError(path, errRunFileChanged)
	}
	fp, err := mw.CreateFormFile("file", want.Name)
	if err != nil {
		return err
	}
	if n, err := io.CopyN(fp, f, want.Size); err != nil {
		if n < want.Size && errors.Is(err, io.EOF) {
			err = errRunFileChanged
		}
		return runReadError(path, err)
	}
	if st, err := f.Stat(); err != nil || st.Size() != want.Size {
		return runReadError(path, errRunFileChanged)
	}
	return nil
}

func (c runAPI) page(ctx context.Context, id string, after int64) (runs.Page, error) {
	status, h, body, err := c.s.getJSONHeader(ctx, c.path("/"+url.PathEscape(id)+fmt.Sprintf("?after=%d", after)))
	var out runs.Page
	err = runAnswer(status, body, err, http.StatusOK, &out)
	var ce *runConsoleError
	if errors.As(err, &ce) && status == http.StatusTooManyRequests {
		if n, perr := strconv.Atoi(h.Get("Retry-After")); perr == nil && n > 0 {
			ce.retryAfter = time.Duration(n) * time.Second
		}
	}
	return out, err
}

func (c runAPI) cancel(ctx context.Context, id string) (string, error) {
	status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/cancel"), struct{}{})
	var out runs.StateAnswer
	return out.State, runAnswer(status, body, err, http.StatusOK, &out)
}

func (c runAPI) expire(ctx context.Context, id string) (string, error) {
	status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/expire"), struct{}{})
	var out runs.StateAnswer
	return out.State, runAnswer(status, body, err, http.StatusOK, &out)
}

func (c runAPI) ack(ctx context.Context, id string, lastSeq int64) error {
	status, body, err := c.s.postJSON(ctx, c.path("/"+url.PathEscape(id)+"/ack"), runs.Ack{LastSeq: lastSeq})
	return runAnswer(status, body, err, http.StatusOK, nil)
}

// runTerm is how the follow loop meets the terminal and the clock; tests
// replace all of it.
type runTerm struct {
	stdout, stderr io.Writer
	interrupts     <-chan os.Signal
	after          func(time.Duration) <-chan time.Time
	now            func() time.Time
}

func (t runTerm) say(s string) { fmt.Fprintln(t.stderr, s) }

// runAckTries is how many times an ack that the console cannot take yet, or
// that did not arrive, is sent; runAckWait is the pause between them.
const (
	runAckTries = 10
	runAckWait  = time.Second
)

// How the follow loop treats a console that is slow to answer. A page that
// fails for a reason that may pass (a 5xx, a lost connection, a 429) is asked
// for again; after runPageGiveUp of failing in a row the follow is given up
// (the run carries on, and --follow picks it up again). A 429 waits longer,
// or as long as the console's Retry-After says, at most runRateWaitMax. A
// page cut short is followed after a short pause, so that catching up on a
// long run cannot spend the console's per-token budget in a burst.
const (
	runPageGiveUp  = 2 * time.Minute
	runRateWait    = 5 * time.Second
	runRateWaitMax = time.Minute
	runMorePause   = 200 * time.Millisecond
)

// runNotPickedUp is RUN-C55.
func runNotPickedUp(host string) string {
	return fmt.Sprintf("%s: not picked up within 60 seconds. the run did not start. try again", host)
}

// runLetGo is RUN-C86: the record no longer names the run and no result
// came back for it within runPageGiveUp.
func runLetGo(host, id string) string {
	return fmt.Sprintf("%s: the instance no longer holds run %s, and no result came back. it may have done part of its work", host, id)
}

// runConflict reports whether err is the console's 409 with exactly msg.
func runConflict(err error, msg string) bool {
	var ce *runConsoleError
	return errors.As(err, &ce) && ce.status == http.StatusConflict && ce.msg == msg
}

// ackRun tells the console the output has been read, so it can delete the
// run's objects. The output is already on the terminal and the exit code is
// the command's, so an ack that cannot be made is said and nothing more.
// An ack retried after a lost 200 finds the run gone (404), which is done.
// An ack that beats the box's last write (409, not finished) or does not
// arrive is sent again a few times.
func ackRun(ctx context.Context, c runClient, t runTerm, id string, lastSeq int64) {
	var err error
	for try := 0; try < runAckTries && ctx.Err() == nil; try++ {
		if try > 0 {
			select {
			case <-t.after(runAckWait):
			case <-ctx.Done():
				return
			}
		}
		err = c.ack(ctx, id, lastSeq)
		var ce *runConsoleError
		switch {
		case err == nil:
			return
		case errors.As(err, &ce) && ce.status == http.StatusNotFound:
			return
		case errors.As(err, &ce) && !runConflict(err, fmt.Sprintf("run %s has not finished yet", id)):
			t.say(fmt.Sprintf("carlos run: %s", err))
			return
		}
	}
	if ctx.Err() != nil {
		return // cut off by Ctrl-C: nothing hangs on the ack
	}
	t.say(fmt.Sprintf("carlos run: %s", err))
}

// followRun prints a run's output until its result, acknowledges it, and
// returns the exit code (run spec §5.6). It polls once a second; a page
// cut short is followed at once; the result is in hand only when a page
// carries it and is not cut short, because the console reads the result
// before it lists chunks. A run still queued at expireAt is expired (§5.2);
// zero never expires. A page with no state and no result is a run the
// record no longer names (the instance was deleted, or the run was dropped):
// a result may still arrive, so it is asked for again for runPageGiveUp, and
// then the follow ends with RUN-C86 and 125. The first Ctrl-C cancels the
// run on the box and keeps following; the second stops waiting (§5.7).
func followRun(ctx context.Context, c runClient, t runTerm, host, id string, expireAt time.Time) int {
	var after int64
	watch := &runWatch{ch: t.interrupts}
	cancel := runCancel{w: watch}
	var failingSince, letGoSince time.Time
	for {
		var p runs.Page
		var err error
		watch.run(ctx, func(ctx context.Context) { p, err = c.page(ctx, id, after) })
		if watch.stopped() {
			return stopWaiting(ctx, t, c, id, &cancel)
		}
		if err != nil {
			var ce *runConsoleError
			switch {
			case errors.As(err, &ce) && ce.status == http.StatusNotFound:
				t.say(fmt.Sprintf("%s has no run %s, or this console has no carlos run yet", host, id))
				return 2
			case errors.As(err, &ce) && ce.status != http.StatusTooManyRequests:
				t.say(fmt.Sprintf("carlos run: %s", err))
				return 2
			}
			// A 5xx, a 429 or a connection that failed may pass: ask again,
			// for a while.
			if failingSince.IsZero() {
				failingSince = t.now()
			}
			if t.now().Sub(failingSince) >= runPageGiveUp {
				t.say(fmt.Sprintf("carlos run: %s", err))
				return 1
			}
			wait := time.Second
			if ce != nil {
				wait = min(max(runRateWait, ce.retryAfter), runRateWaitMax)
			}
			if interrupted(ctx, t, c, id, &cancel, wait) {
				return 130
			}
			continue
		}
		failingSince = time.Time{}
		for _, ch := range p.Chunks {
			if ch.Seq <= after {
				continue
			}
			w := t.stdout
			if ch.Stream == runs.StreamErr {
				w = t.stderr
			}
			w.Write(ch.Data)
			after = ch.Seq
		}
		if p.More {
			if interrupted(ctx, t, c, id, &cancel, runMorePause) {
				return 130
			}
			continue
		}
		if p.Result != nil {
			// The result is in hand and the ack is a courtesy: Ctrl-C during
			// it ends the ack, not the exit code, and cancels nothing.
			watch.run(ctx, func(ctx context.Context) { ackRun(ctx, c, t, id, p.Result.LastSeq) })
			code, lines := runExit(host, *p.Result)
			switch {
			case p.Result.Cancelled && p.Result.Started.IsZero():
				// Cancelled before any box launched it: there was no command.
				lines = []string{fmt.Sprintf("%s: run %s was cancelled before it started", host, id)}
			case p.Result.TimedOut && p.Result.Started.IsZero():
				// Its deadline passed before any box launched it: there was
				// no command, and so no duration to report (re-review 8).
				lines = []string{fmt.Sprintf("%s: run %s timed out before it started", host, id)}
			}
			for _, l := range lines {
				t.say(l)
			}
			return code
		}
		if p.State != "" {
			letGoSince = time.Time{}
		} else {
			if letGoSince.IsZero() {
				letGoSince = t.now()
			}
			if t.now().Sub(letGoSince) >= runPageGiveUp {
				t.say(runLetGo(host, id))
				return 125
			}
		}
		switch p.State {
		case provision.RunExpired:
			t.say(runNotPickedUp(host))
			return 3
		case provision.RunCancelled:
			t.say(fmt.Sprintf("%s: run %s was cancelled before it started", host, id))
			return 130
		case provision.RunQueued:
			if !expireAt.IsZero() && !t.now().Before(expireAt) {
				var state string
				var err error
				watch.run(ctx, func(ctx context.Context) { state, err = c.expire(ctx, id) })
				if watch.stopped() {
					return stopWaiting(ctx, t, c, id, &cancel)
				}
				var ce *runConsoleError
				switch {
				case err == nil && state == provision.RunExpired:
					// The run's uploaded files go now, not with the sweep.
					// The ack is a courtesy: nothing hangs on it.
					watch.run(ctx, func(ctx context.Context) { _ = c.ack(ctx, id, 0) })
					t.say(runNotPickedUp(host))
					return 3
				case err == nil || errors.As(err, &ce):
					// A box picked the run first (409), or the console will
					// not expire it: follow it. A call that did not arrive is
					// asked again on the next poll.
					expireAt = time.Time{}
				}
			}
		}
		if interrupted(ctx, t, c, id, &cancel, time.Second) {
			return 130
		}
	}
}

// runWatch is the follow loop's view of Ctrl-C. Between requests the loop
// takes a Ctrl-C itself; while a request is in flight run takes it, so the
// second one can end that request at once (run spec §5.7). n counts every
// Ctrl-C taken, and handled counts those the loop has acted on.
type runWatch struct {
	ch         <-chan os.Signal
	n, handled int
}

// run calls f with a context that the second Ctrl-C cancels, and reports
// whether it did. Ctrl-Cs that arrive meanwhile are counted for the loop. If
// the second has already been taken, the context starts cancelled: no request
// gets going after the user said stop.
func (w *runWatch) run(ctx context.Context, f func(context.Context)) bool {
	return w.do(ctx, 0, f)
}

// do is run with a deadline when limit is set. A request made after the
// second Ctrl-C is bounded this way instead of started cancelled: it is still
// attempted, and a further Ctrl-C ends it.
func (w *runWatch) do(ctx context.Context, limit time.Duration, f func(context.Context)) (aborted bool) {
	rctx, cancel := context.WithCancel(ctx)
	if limit > 0 {
		var stopTimer context.CancelFunc
		rctx, stopTimer = context.WithTimeout(rctx, limit)
		defer stopTimer()
	} else if w.n >= 2 {
		aborted = true
		cancel()
	}
	stop, fin := make(chan struct{}), make(chan struct{})
	go func() {
		defer close(fin)
		for {
			select {
			case <-w.ch:
				w.n++
				if w.n >= 2 && !aborted {
					aborted = true
					cancel()
				}
			case <-stop:
				return
			}
		}
	}()
	f(rctx)
	close(stop)
	<-fin
	cancel()
	return aborted
}

// stopped reports whether the second Ctrl-C has been taken.
func (w *runWatch) stopped() bool { return w.n >= 2 }

// take claims a Ctrl-C that is waiting, taken by run or still in the channel.
func (w *runWatch) take() bool {
	if w.n > w.handled {
		w.handled++
		return true
	}
	select {
	case <-w.ch:
		w.n++
		w.handled++
		return true
	default:
		return false
	}
}

// runLateCancelWait is how long a cancel sent after the second Ctrl-C may
// take. A variable so a test can shorten it.
var runLateCancelWait = 3 * time.Second

// runCancel is the state of the cancel a Ctrl-C asks for.
type runCancel struct {
	w                 *runWatch
	asked, done, said bool
	abandoned         bool // the second Ctrl-C cut the request off: do not send it again
}

// send makes the cancel, if one was asked for and has not been answered. A
// cancel retried after a lost 200 finds the run already ended, which is what
// was asked for. One that did not arrive is sent again on the next call, so
// the cancel still happens (run spec §5.7); it is said once. A cancel the
// second Ctrl-C cut off is not sent again: that Ctrl-C said stop waiting.
func (k *runCancel) send(ctx context.Context, t runTerm, c runClient, id string) {
	if !k.asked || k.done || k.abandoned {
		return
	}
	var err error
	var aborted bool
	// A cancel sent after the second Ctrl-C is still attempted, but only for
	// a moment: the user has said stop waiting.
	late := k.w.stopped()
	if late {
		aborted = k.w.do(ctx, runLateCancelWait, func(ctx context.Context) { _, err = c.cancel(ctx, id) })
	} else {
		aborted = k.w.run(ctx, func(ctx context.Context) { _, err = c.cancel(ctx, id) })
	}
	if (aborted && errors.Is(err, context.Canceled)) || (late && errors.Is(err, context.DeadlineExceeded)) {
		k.abandoned = true
		return
	}
	var ce *runConsoleError
	switch {
	case err == nil, runConflict(err, fmt.Sprintf("run %s has already ended", id)):
		k.done = true
	case errors.As(err, &ce):
		k.done = true
		t.say(fmt.Sprintf("carlos run: %s", err))
	case !k.said:
		k.said = true
		t.say(fmt.Sprintf("carlos run: %s", err))
	}
}

// press acts on one Ctrl-C that has been taken. The first asks for the cancel
// and returns false so following goes on; the second tries the cancel once
// more if it is unanswered and returns true: stop waiting.
func (k *runCancel) press(ctx context.Context, t runTerm, c runClient, id string) bool {
	if k.asked {
		k.send(ctx, t, c, id)
		return true
	}
	k.asked = true
	t.say(fmt.Sprintf("cancelling run %s. press Ctrl-C again to stop waiting", id))
	k.send(ctx, t, c, id)
	return false
}

// stopWaiting is the exit once the second Ctrl-C has been taken, possibly
// while a request was in flight and the first not yet acted on.
func stopWaiting(ctx context.Context, t runTerm, c runClient, id string, k *runCancel) int {
	for k.w.take() {
		if k.press(ctx, t, c, id) {
			break
		}
	}
	return 130
}

// interrupted waits out d, the pause before the next request, and handles a
// Ctrl-C that arrives first or was already waiting: a pending Ctrl-C is
// taken before the next tick, never after it. The first asks for the cancel
// and returns false so following goes on; the second stops waiting and
// returns true. A cancel still unanswered when the wait ends is sent again.
func interrupted(ctx context.Context, t runTerm, c runClient, id string, k *runCancel, d time.Duration) bool {
	if !k.w.take() {
		select {
		case <-t.after(d):
			k.send(ctx, t, c, id)
			// A second Ctrl-C during the resend ends the wait here, not
			// after another request.
			return k.w.stopped()
		case <-k.w.ch:
			k.w.n++
			k.w.handled++
		}
	}
	if k.press(ctx, t, c, id) {
		return true
	}
	// A second Ctrl-C already waiting is not left for the next poll.
	if k.w.take() {
		k.press(ctx, t, c, id)
		return true
	}
	return false
}

// runExit is a result's exit code and what stderr says about it (run spec
// §5.6, §8; index item 6).
func runExit(host string, r runs.Result) (int, []string) {
	var lines []string
	if r.Truncated {
		lines = append(lines, fmt.Sprintf("%s: the output passed 64 MB and was cut off there. the command ran on to its end", host))
	}
	switch {
	case r.Refused == runs.RefusedExpired:
		return 3, append(lines, runNotPickedUp(host))
	case r.Refused == runs.RefusedFlat:
		return 3, append(lines, fmt.Sprintf("%s: the box is not on the nested layout yet, so carlos run cannot run there", host))
	case r.Refused == runs.RefusedClaim:
		return 3, append(lines, fmt.Sprintf("%s: the box has host claims armed. carlos run cannot run there yet", host))
	case r.Refused == runs.RefusedWake:
		return 3, append(lines, fmt.Sprintf("%s: the instance did not wake. %s", host, r.Detail))
	case r.Refused == runs.RefusedIncarnation:
		return 3, append(lines, fmt.Sprintf("%s was deleted and created again after this run was requested. the run did not start", host))
	case r.Refused == runs.RefusedImport:
		return 3, append(lines, fmt.Sprintf("%s is an import that is not committed yet. the run did not start", host))
	case r.Refused != "":
		return 3, append(lines, fmt.Sprintf("carlos run: %s", r.Refused))
	case r.Cancelled:
		return 130, append(lines, fmt.Sprintf("%s: the command was cancelled", host))
	case r.TimedOut:
		return 124, append(lines, fmt.Sprintf("%s: the command timed out after %s", host, r.Ended.Sub(r.Started).Round(time.Second)))
	case r.Interrupted != "":
		return 125, append(lines, fmt.Sprintf("%s: the run was interrupted: %s. it may have done part of its work", host, r.Interrupted))
	case r.Signal != "":
		n, _ := strconv.Atoi(r.Signal)
		return 128 + n, lines
	}
	return r.Exit, lines
}

// startRun sends the run and follows it. A console refusal, or a --file that
// cannot be read, is exit 2.
func startRun(ctx context.Context, c runClient, t runTerm, o runOpts, app string) int {
	meta, err := runMetaFor(o)
	if err != nil {
		t.say(err.Error())
		return 2
	}
	created, err := c.create(ctx, meta, o.files)
	if err != nil {
		var ce *runConsoleError
		switch {
		case errors.Is(err, errRunLocalFile):
			t.say(err.Error())
			return 2
		case errors.As(err, &ce):
			t.say(fmt.Sprintf("carlos run: %s", err))
			return 2
		}
		t.say(fmt.Sprintf("carlos run: %s", err))
		return 1
	}
	if created.Replayed {
		// An earlier attempt with this key started the run: this one starts
		// nothing, and follows it.
		if created.State == "" && runGone(ctx, c, created.ID) {
			// The record no longer names the run and its objects are gone: it
			// was acknowledged or swept. The key proves it started.
			t.say(runReplayGone(created.ID))
			return 2
		}
		t.say(fmt.Sprintf("already started: run %s on %s. to follow it again: carlos run --app %s --instance %s --follow %s", created.ID, o.instance, app, o.instance, created.ID))
	} else {
		t.say(fmt.Sprintf("run %s on %s. to follow it again: carlos run --app %s --instance %s --follow %s", created.ID, o.instance, app, o.instance, created.ID))
	}
	return followRun(ctx, c, t, o.instance, created.ID, t.now().Add(runs.PickupWindow))
}

// runReplayGone is what a retry says when its key started a run that is no
// longer there to follow.
func runReplayGone(id string) string {
	return fmt.Sprintf("run %s was already started with this key, and its output is no longer kept", id)
}

// runGone reports whether the console says it has no run id (a 404 on the
// first page). Any other answer, an error included, is not "gone": the
// follow that comes next meets it again and deals with it.
func runGone(ctx context.Context, c runClient, id string) bool {
	_, err := c.page(ctx, id, 0)
	var ce *runConsoleError
	return errors.As(err, &ce) && ce.status == http.StatusNotFound
}

// runRun is `carlos run` (run spec §3).
func runRun(args []string) {
	o, err := parseRunArgs(args, os.Stderr)
	if err != nil {
		if errors.Is(err, flag.ErrHelp) {
			os.Exit(0)
		}
		if !errors.Is(err, errRunUsage) {
			fmt.Fprintln(os.Stderr, err.Error())
		}
		fmt.Fprint(os.Stderr, runUsageText())
		os.Exit(2)
	}
	sess, err := resolveMode(o.console)
	if err != nil {
		log.Fatal(err)
	}
	if sess == nil {
		log.Fatal(runDirectModeMsg)
	}
	printUsing(os.Stderr, sess)
	acct := resolvedAccount(sess, o.acctRaw, o.acctIDRaw)
	app := resolvedApp(sess, acct, o.app, runUsageText())
	interrupts := make(chan os.Signal, 2)
	signal.Notify(interrupts, os.Interrupt, syscall.SIGTERM)
	t := runTerm{stdout: os.Stdout, stderr: os.Stderr, interrupts: interrupts, after: time.After, now: time.Now}
	c := runAPI{s: sess, a: acct, app: app, host: o.instance, key: o.idempotencyKey}
	ctx := context.Background()
	if o.follow != "" {
		os.Exit(followRun(ctx, c, t, o.instance, o.follow, time.Time{}))
	}
	os.Exit(startRun(ctx, c, t, o, app))
}