package schedule

import (
	"context"
	"errors"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
)

const (
	DeliveryTimeout  = 30 * time.Minute
	TickEvery        = 15 * time.Second
	MaxAttempts      = 5
	MaxParallelHosts = 4
	dueBatch         = 64
	backoffBase      = time.Minute
	backoffCap       = 15 * time.Minute
	// pruneEvery is the prune leg's cadence — see Runner.prune. Daily
	// because the thing it deletes is a 7-day-old finished row: a sweep an
	// hour earlier or later changes nothing a member can see.
	pruneEvery = 24 * time.Hour
)

// ErrTimeout is what a Deliverer returns when the handler ran past the
// delivery timeout. The policy treats it as delivered (spec §6.3): the job
// may have finished, or still be running — a retry would double it.
var ErrTimeout = errors.New("schedule: delivery timed out")

// Deliverer sends one tick and reports the handler's status, or an error.
//
// Status 0 with an error means NO RESPONSE — the tick never got an HTTP
// status line back. On the wake path that is a wake the activator refused
// ("instance is waking", "failed to wake"), on any path a delivery that ran
// past its deadline, and either way a handler that returned having written
// nothing. It does NOT mean the instance is unreachable in any lasting
// sense: the commonest cause is a wake still in progress, and the next
// attempt usually lands. Nor is it every failure — a socket that will not
// dial comes back as the proxy's own 502, a status like any other.
type Deliverer interface {
	Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error)
}

// Runner fires due timers: at-least-once, serialized per host, bounded
// parallelism across hosts (spec §6.3). Every seam is a field so a tick is
// testable with no sockets or clocks.
type Runner struct {
	StartDelivery DeliveryAdmission
	Reg           *registry.Registry
	Deliver       Deliverer
	Now           func() time.Time
	Timeout       time.Duration
	Token         func(host string) (string, error)
	Event         func(rt registry.Route, level, msg string)
	Report        func(rt registry.Route, st Status)
	Box           string
	Build         string
	Idle          time.Duration // the activator's idle window; zero when no hibernation is armed on this box
	HostWrites    hostwrite.Config
	// Gate is the per-host gate shared with the converger and a move's
	// drain (HostGate): claim enters it for the whole delivery. Nil: no
	// gate, main's runner. Independent of StartDelivery: a held row is
	// skipped before claim and fire, and the delivery admission is asked
	// inside fire for the rows that get that far, so a tick delivers only
	// when the hold and the admission both allow it.
	Gate *HostGate
	// Skip names a host this box must not touch at all, whatever the hold
	// wiring says: its due timers are neither delivered nor written, no
	// token is minted for it and nothing is reported. The agent sets it for
	// a held row and for a host whose offline move's record names another
	// region (harness Run 9, defect 8: after open, the source's tick minted
	// a fresh <host>.env in the instance directory the move had parked).
	// Asked at the tick's route read and again just before the token, so a
	// hold written in between is seen. Nil: none is skipped.
	Skip func(rt registry.Route) bool

	// beforeClaim is the tests' pause between a tick's route read and its
	// claim (TestDrainRaceCannotAdvanceBookkeeping). Nil in the agent.
	beforeClaim func(host string)

	// Per-host work is DETACHED from the tick: Tick launches one goroutine
	// per host with due rows and returns without waiting for any of them.
	// A handler is allowed to run for up to DeliveryTimeout (30m), and a
	// tick that waited on it would hold every other host's due timers
	// hostage for that long. busy is what keeps "one delivery at a time per
	// host" true across ticks: a host whose goroutine is still running is
	// skipped, and its rows (still due — nothing has been recorded yet) are
	// picked up by the first tick after it finishes. sem bounds hosts in
	// flight across ALL ticks, not per tick. inflight is for tests, which
	// need "everything this tick launched has finished" — see wait.
	mu       sync.Mutex
	busy     map[string]bool
	sem      chan struct{}
	inflight sync.WaitGroup

	// lastPruned is the prune leg's stamp — see prune.
	lastPruned time.Time
}

// Start runs Tick every TickEvery until ctx ends.
func (r *Runner) Start(ctx context.Context) {
	go func() {
		t := time.NewTicker(TickEvery)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				r.Tick(ctx)
			}
		}
	}()
}

func (r *Runner) now() time.Time {
	if r.Now != nil {
		return r.Now()
	}
	return time.Now()
}

func (r *Runner) timeout() time.Duration {
	if r.Timeout > 0 {
		return r.Timeout
	}
	return DeliveryTimeout
}

// Tick is one pass: everything due, grouped by host; each host not already
// in flight gets a goroutine (at most MaxParallelHosts running across all
// ticks), timers within a host one after another. Tick returns once every
// goroutine is launched — it never waits for a delivery.
func (r *Runner) Tick(ctx context.Context) {
	now := r.now()
	r.prune(ctx, now)
	// Busy hosts are excluded IN THE QUERY, not after it. The batch is
	// capped at dueBatch rows: a host parked in a long handler (or one
	// holding more due rows than the batch) would otherwise fill the batch
	// with rows this tick is about to skip, and another host's due timer
	// would never be seen. The skip below stays as belt-and-braces — a host
	// can go busy between the query and the claim.
	//
	// With the hold wiring armed, held hosts are excluded in the query for
	// the same reason: a held row is skipped below with no bookkeeping, so
	// a held host's rows would sit at the front of every tick's batch and
	// starve the rest. Off, the query is main's.
	//
	// Skip set (the agent's scheduleSkip) is the same rule for the hosts it
	// names, before the limit too (Astra, moves-offline-fixes review 1): a
	// held host's rows are left out by the query, and a host Skip names on
	// other grounds is added to the exclusion and the batch read again, so
	// a skipped host's due rows can never fill the batch.
	due, err := r.dueRows(now)
	if err != nil {
		log.Printf("schedule: due query: %v", err)
		return
	}
	if len(due) == 0 {
		return
	}
	byHost := map[string][]registry.Timer{}
	var order []string
	for _, t := range due {
		if _, seen := byHost[t.Host]; !seen {
			order = append(order, t.Host)
		}
		byHost[t.Host] = append(byHost[t.Host], t)
	}
	for _, host := range order {
		work, err := r.HostWrites.Select(ctx, host, "timers")
		if err != nil {
			log.Printf("schedule: %s: select lifecycle authority: %v", host, err)
			continue
		}
		hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work})
		var rt *registry.Route
		err = withTimerWork(hctx, func() error {
			var readErr error
			rt, readErr = r.Reg.Get(host)
			return readErr
		})
		if err != nil {
			// A registry error is not "route gone": say so, or a broken
			// registry reads as a fleet of retired routes.
			log.Printf("schedule: %s: registry read: %v — skipping this tick", host, err)
			continue
		}
		if rt == nil {
			// Converge prunes these; until it does, they are not ours to fire.
			continue
		}
		if r.Skip != nil && r.Skip(*rt) {
			continue // not this box's host to touch: no bookkeeping either
		}
		// A held row (moves spec §5.2, §9.4 "frozen") is skipped with no
		// bookkeeping at all: before this, its delivery was refused by the
		// activator and recorded as a retry, so a held row's attempts
		// advanced while its instance was moving. This read is only the
		// cheap pre-filter; the decision is the read inside the claim below.
		// Only with the hold wiring armed: off, the row is main's.
		if r.holdsArmed() && registry.IsHold(rt.Status) {
			continue
		}
		// An uncommitted import's row gets no ticks (seeded-instances spec
		// §5.4): only verify requests reach it until commit. Whether or not
		// the hold wiring is armed.
		if rt.ImportGate != "" {
			continue
		}
		if r.beforeClaim != nil {
			r.beforeClaim(host)
		}
		if !r.claim(host) {
			continue // a previous tick's delivery to this host is still running, or a move is draining it
		}
		if r.holdsArmed() {
			// The decision read, inside the claim. A move holds the row and
			// then drains the host (HostGate.Drain), and the drain is sound
			// only if every tick that looks at the row after the hold does
			// so while it holds the claim: a tick that read the row ready
			// above and was descheduled before claim must not slip past the
			// drain and record an attempt.
			err = withTimerWork(hctx, func() error {
				var readErr error
				rt, readErr = r.Reg.Get(host)
				return readErr
			})
			// The import gate is part of the decision too: a swap that
			// landed after the pre-filter read gets no tick.
			if err != nil || rt == nil || registry.IsHold(rt.Status) || rt.ImportGate != "" {
				if err != nil {
					log.Printf("schedule: %s: registry read inside the claim: %v. skipping this tick", host, err)
				}
				r.release(host)
				continue
			}
		}
		// The slot is taken HERE, before the goroutine exists, so the bound
		// is on hosts in flight and never on goroutines waiting to be. Hosts
		// are in due order, so when every slot is busy the earliest-due host
		// is the one waiting for the next free one.
		select {
		case r.slots() <- struct{}{}:
		case <-ctx.Done():
			r.release(host)
			return
		}
		r.inflight.Add(1)
		go func(ctx context.Context, host string, rt registry.Route, timers []registry.Timer) {
			defer r.inflight.Done()
			defer func() { <-r.slots() }()
			defer r.release(host)
			for i, t := range timers {
				// A hold written while this host's earlier timers were
				// delivering stops the rest: each later timer re-reads the
				// row first. The drain waits for this goroutine either way.
				// Only with the hold wiring armed.
				if i > 0 && r.holdsArmed() && r.held(ctx, host) {
					break
				}
				r.fire(ctx, rt, t)
			}
			if r.Report != nil {
				// A report that cannot be built is a silent hole in `carlos
				// schedule ls` — the member's only view of this box. Say so:
				// the fires themselves still landed on the rows.
				var st Status
				err := withTimerWork(ctx, func() error {
					var statusErr error
					st, statusErr = StatusFor(r.Reg, rt, r.Box, r.now(), r.Idle)
					return statusErr
				})
				if err != nil {
					log.Printf("schedule: %s: build status report: %v", rt.Host, err)
				} else {
					r.Report(rt, st)
				}
			}
		}(hctx, host, *rt, byHost[host])
	}
}

// dueRows is Tick's due query: busy hosts excluded, held hosts too when the
// hold wiring is armed or Skip is set, and, with Skip set, every host it
// names, read again with each one found added to the exclusion. Each read
// either has no skipped host or adds at least one host to the exclusion, so
// the loop ends within one read per host with due rows. A host whose row
// cannot be read is left in the batch: the tick's own read reports it.
func (r *Runner) dueRows(now time.Time) ([]registry.Timer, error) {
	exclude := r.busyHosts()
	for {
		var due []registry.Timer
		var err error
		if r.holdsArmed() || r.Skip != nil {
			due, err = r.Reg.DueUnheldTimers(now.Unix(), dueBatch, exclude)
		} else {
			due, err = r.Reg.DueTimers(now.Unix(), dueBatch, exclude)
		}
		if err != nil || r.Skip == nil || len(due) == 0 {
			return due, err
		}
		var skipped []string
		seen := map[string]bool{}
		for _, t := range due {
			if seen[t.Host] {
				continue
			}
			seen[t.Host] = true
			if rt, err := r.Reg.Get(t.Host); err == nil && rt != nil && r.Skip(*rt) {
				skipped = append(skipped, t.Host)
			}
		}
		if len(skipped) == 0 {
			return due, nil
		}
		exclude = append(exclude, skipped...)
	}
}

// busyHosts is the in-flight set as DueTimers wants it. A snapshot: a host
// that goes busy after this read is caught by claim instead.
func (r *Runner) busyHosts() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	if len(r.busy) == 0 {
		return nil
	}
	hosts := make([]string, 0, len(r.busy))
	for h := range r.busy {
		hosts = append(hosts, h)
	}
	return hosts
}

// prune is the runner's OWN sweep of dead timer rows, and it exists because
// the converger's identical sweep is not always there: the converger needs
// the deploy store, and a box with the activator armed and no store still
// accumulates rows — once-timers registered by instances on their control
// socket (spec §7.2). Without this leg those finished rows would sit in the
// table forever and could pin an app against the 1000-per-host cap on the
// one kind of box that has no other way to shed them.
//
// Daily, stamped rather than scheduled: it rides the 15s tick so there is no
// second goroutine to stop, and the first tick after start prunes once.
// Cheap either way (one DELETE over an indexed table), and idempotent with
// the converger's sweep on a box that has both.
func (r *Runner) prune(ctx context.Context, now time.Time) {
	r.mu.Lock()
	if !r.lastPruned.IsZero() && now.Sub(r.lastPruned) < pruneEvery {
		r.mu.Unlock()
		return
	}
	// Stamped before the work, not after: a registry that cannot be read is
	// not worth retrying every 15s, and the rows a failed sweep would have
	// dropped are inert until it next runs.
	r.lastPruned = now
	r.mu.Unlock()
	routes, err := r.Reg.List()
	if err != nil {
		log.Printf("schedule: runner prune: list routes: %v — leaving the rows alone", err)
		return
	}
	sockets := map[string]bool{}
	alive := map[string]bool{}
	for _, rt := range routes {
		alive[rt.Host] = true
		if rt.Socket != "" {
			sockets[rt.Socket] = true
		}
	}
	// The sweep is also the delivery leg's eviction point: a route that goes
	// away stops receiving deliveries, so nothing it caches for that route's
	// socket would ever error and be dropped on its own. Optional — a
	// Deliverer that caches nothing (every test's, and any future one) simply
	// does not implement it.
	if sr, ok := r.Deliver.(SocketRetainer); ok {
		sr.RetainSockets(sockets)
	}
	if len(routes) == 0 {
		// PruneTimers reads an empty alive set as "every host is gone" and
		// deletes the whole table — correct for a genuinely empty registry
		// (routes and timers live in the same file, so every row really is
		// orphaned), but it is a wipe reachable from two callers now, and a
		// caller that ever passed an empty set by mistake would take the
		// table with it. This leg declines it explicitly: rows on a box with
		// no routes are inert, and the next sweep after a route appears is
		// the one that can tell orphans from arrivals.
		log.Printf("schedule: runner prune: the registry has no routes — declining to sweep (that would delete every timer row)")
		return
	}
	if !r.HostWrites.Enabled {
		if n, err := r.Reg.PruneTimers(alive, now.Add(-onceRetention).Unix()); err != nil {
			log.Printf("schedule: runner prune: %v", err)
		} else if n > 0 {
			log.Printf("schedule: runner prune: pruned %d timer row(s)", n)
		}
		return
	}
	cutoff := now.Add(-onceRetention).Unix()
	for _, rt := range routes {
		work, err := r.HostWrites.Select(ctx, rt.Host, "timers")
		if err != nil {
			log.Printf("schedule: runner prune %s: select lifecycle authority: %v", rt.Host, err)
			continue
		}
		hctx := context.WithValue(ctx, timerWorkKey{}, timerWork{ctx: ctx, work: work})
		var rows []registry.Timer
		if err := withTimerWork(hctx, func() error {
			var readErr error
			rows, readErr = r.Reg.TimersFor(rt.Host)
			return readErr
		}); err != nil {
			log.Printf("schedule: runner prune %s: %v", rt.Host, err)
			continue
		}
		for _, row := range rows {
			if row.Kind != KindOnce || row.NextAt != 0 || row.LastAt >= cutoff {
				continue
			}
			if err := deleteSelectedTimer(hctx, r.Reg, row); err != nil {
				log.Printf("schedule: runner prune %s/%s: %v", row.Host, row.Name, err)
			}
		}
	}
}

type timerWorkKey struct{}
type timerWork struct {
	ctx  context.Context
	work hostwrite.Work
}

func withTimerWork(ctx context.Context, fn func() error) error {
	selected, ok := ctx.Value(timerWorkKey{}).(timerWork)
	if !ok || selected.ctx == nil || selected.work == nil {
		return hostwrite.ErrUnavailable
	}
	return selected.work.With(selected.ctx, hostwrite.Identity{}, fn)
}

// SocketRetainer is the optional half of Deliverer that keeps per-socket
// state (a connection pool, in the agent's case). The runner's prune leg
// hands it the set of sockets still routed on this box; everything else the
// implementation holds is for a route that is gone.
type SocketRetainer interface {
	RetainSockets(live map[string]bool)
}

// holdsArmed reports whether the move-hold wiring is on. The host package
// sets Gate (armScheduleHolds) only with the Phase 1 switch on and
// scheduled work armed; Gate nil is the switch off, and then every hold
// check in Tick is skipped and the runner is main's code path exactly: the
// due query is DueTimers, there is no decision read inside the claim, no
// check between a host's timers, and a row whose status is a hold is
// delivered and recorded like any other.
func (r *Runner) holdsArmed() bool { return r.Gate != nil }

// claim marks host in flight and enters its gate; false if it already was in
// flight, or a move's drain holds or waits for the host (HostGate).
func (r *Runner) claim(host string) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.busy[host] {
		return false
	}
	if !r.Gate.tryEnter(host) {
		return false
	}
	if r.busy == nil {
		r.busy = map[string]bool{}
	}
	r.busy[host] = true
	return true
}

// Busy reports whether a delivery to host is in flight: the per-host claim
// (moves spec §9.4 "drained"). A snapshot: the cut's drain is
// Gate.Drain, which also waits for a claim taken after this answer.
func (r *Runner) Busy(host string) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.busy[host]
}

func (r *Runner) release(host string) {
	r.mu.Lock()
	delete(r.busy, host)
	r.Gate.leave(host)
	r.mu.Unlock()
}

// held reports whether host's row is held by an operation now, or cannot be
// read: either way the rest of the host's timers wait for the next tick.
func (r *Runner) held(ctx context.Context, host string) bool {
	var rt *registry.Route
	err := withTimerWork(ctx, func() error {
		var readErr error
		rt, readErr = r.Reg.Get(host)
		return readErr
	})
	return err != nil || rt == nil || registry.IsHold(rt.Status)
}

// slots is the cross-tick semaphore, made on first use so a zero Runner
// literal (every test, and the agent's) needs no constructor.
func (r *Runner) slots() chan struct{} {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.sem == nil {
		r.sem = make(chan struct{}, MaxParallelHosts)
	}
	return r.sem
}

// wait blocks until every host goroutine launched so far has finished —
// the tests' seam, so a Tick's post-conditions can be asserted without
// sleeping. Nothing in the agent calls it.
func (r *Runner) wait() { r.inflight.Wait() }

func (r *Runner) timerStillDue(ctx context.Context, t registry.Timer) bool {
	var current *registry.Timer
	err := withTimerWork(ctx, func() error {
		var readErr error
		current, readErr = r.Reg.GetTimer(t.Host, t.Name)
		return readErr
	})
	return err == nil && current.Registration == t.Registration && current.NextAt == t.NextAt && current.NextAt > 0 && current.NextAt <= r.now().Unix()
}

func (r *Runner) fire(ctx context.Context, rt registry.Route, t registry.Timer) {
	if !r.timerStillDue(ctx, t) {
		return
	}

	sp, err := ParseSpec(t.Spec)
	if err != nil {
		log.Printf("schedule: %s/%s: bad spec %q: %v — dropping the row", rt.Host, t.Name, t.Spec, err)
		if err := deleteSelectedTimer(ctx, r.Reg, t); err != nil {
			log.Printf("schedule: %s/%s: drop bad row: %v", rt.Host, t.Name, err)
		}
		return
	}
	if r.Skip != nil {
		// The token is the one write a tick makes on the instance's own
		// ground, so the row is read again right before it.
		var current *registry.Route
		err := withTimerWork(ctx, func() error {
			var readErr error
			current, readErr = r.Reg.Get(rt.Host)
			return readErr
		})
		if err != nil || current == nil || r.Skip(*current) {
			return
		}
	}
	start := r.now()
	token, err := r.Token(rt.Host)
	if err != nil {
		r.record(ctx, rt, t, sp, start, Outcome{Status: "no instance token: " + err.Error(), Retry: true}, 0)
		return
	}
	req, err := BuildRequest(rt, t, token, r.Build)
	if err != nil {
		// Not retried: the path is the row's, and it will be exactly as bad
		// next time. The row advances (or finishes) with the reason on it.
		r.record(ctx, rt, t, sp, start, Outcome{Status: "bad path: " + err.Error()}, 0)
		return
	}
	if !r.timerStillDue(ctx, t) {
		return
	}
	dctx, cancel := context.WithTimeout(ctx, r.timeout())
	req = req.WithContext(dctx)
	var status int
	var derr error
	if r.HostWrites.Enabled {
		status, derr = r.deliverClaim(dctx, rt, t, req)
	} else {
		status, derr = r.Deliver.Deliver(dctx, rt, req)
	}
	cancel()
	if errors.Is(derr, hostwrite.ErrUnavailable) {
		return
	}
	if errors.Is(derr, context.DeadlineExceeded) {
		derr = ErrTimeout
	}
	r.record(ctx, rt, t, sp, start, classify(status, derr, r.timeout()), r.now().Sub(start))
}

// record applies the policy table to one outcome and writes the row, the
// member-facing event, and nothing else.
func (r *Runner) record(ctx context.Context, rt registry.Route, t registry.Timer, sp Spec, at time.Time, o Outcome, took time.Duration) {
	attempts := 0
	status := o.Status
	natural := sp.Next(at) // zero for a finished once
	next := natural.Unix()
	if natural.IsZero() {
		next = 0
	}
	// due is the occurrence the NEXT delivery is for (X-Carlos-Schedule-At,
	// the app's dedupe key): the new natural occurrence when the row
	// advances, and the one just attempted when it is retried.
	due := next
	level := "info"
	if o.Retry {
		attempts = t.Attempts + 1
		if attempts >= MaxAttempts {
			status = fmt.Sprintf("failed after %d attempts: %s", MaxAttempts, o.Status)
			attempts = 0
			level = "warn"
		} else {
			backoff := backoffBase << (attempts - 1)
			if backoff > backoffCap {
				backoff = backoffCap
			}
			retryAt := at.Add(backoff)
			// Never let a retry outlive the next natural occurrence: past it
			// the retry and the occurrence would be the same delivery, and a
			// 15m backoff on a 1m schedule would skip 14 fires. A once row
			// has no natural next (zero), so its retries are never clamped.
			if !natural.IsZero() && retryAt.After(natural) {
				retryAt = natural
			}
			next = retryAt.Unix()
			due = dueOf(t)
			level = "warn"
		}
	} else if !isSuccess(o.Status) {
		level = "warn"
	}
	var applied bool
	err := withTimerWork(ctx, func() error {
		var writeErr error
		applied, writeErr = r.Reg.MarkTimerFired(rt.Host, t.Name, t.Registration, at.Unix(), status, took.Milliseconds(), attempts, next, due)
		return writeErr
	})
	if err != nil {
		log.Printf("schedule: %s/%s: record fire: %v", rt.Host, t.Name, err)
		// Flag off keeps the old behaviour: the fire event still goes out.
		// Under the claim, a refused or failed guarded write stops here.
		if selected, ok := ctx.Value(timerWorkKey{}).(timerWork); !ok || !hostwrite.Legacy(selected.work) {
			return
		}
	}
	if r.Event != nil {
		msg := fmt.Sprintf("schedule %s fired POST %s → %s in %s", t.Name, t.Path, status, took.Truncate(time.Millisecond))
		if attempts > 0 && applied {
			msg += fmt.Sprintf(" (attempt %d of %d; retrying at %s)", attempts, MaxAttempts, time.Unix(next, 0).UTC().Format(time.RFC3339))
		}
		r.Event(rt, level, msg)
	}
}

func isSuccess(status string) bool { return len(status) == 3 && status[0] == '2' }

// Outcome is one line of the policy table (spec §6.3).
type Outcome struct {
	Status string
	Retry  bool
}

func classify(status int, err error, timeout time.Duration) Outcome {
	switch {
	case errors.Is(err, ErrTimeout):
		return Outcome{Status: "timeout after " + timeout.String()}
	case err != nil:
		// No response / transport error (status 0). Recorded as 503 because
		// that is what a member's own client would have seen, and retried
		// for the same reason: nothing says the handler ran.
		return Outcome{Status: "503: " + err.Error(), Retry: true}
	case status >= 200 && status < 300:
		return Outcome{Status: strconv.Itoa(status)}
	case status >= 500, status == 408, status == 429:
		return Outcome{Status: strconv.Itoa(status), Retry: true}
	default:
		return Outcome{Status: strconv.Itoa(status) + " (app refused)"}
	}
}

// Classify is the exported policy for the host package's tests.
func Classify(status int, err error) Outcome { return classify(status, err, DeliveryTimeout) }

func dueOf(t registry.Timer) int64 {
	if t.DueAt != 0 {
		return t.DueAt
	}
	if t.Kind == KindOnce {
		if sp, err := ParseSpec(t.Spec); err == nil && sp.Kind == KindOnce && sp.At.Unix() == 0 {
			return 0
		}
	}
	return t.NextAt
}

// BuildRequest is the tick (spec §6.3's header table). Content-Length 0 and
// no body: a tick carries no data, by design. A path the URL parser refuses
// is an error, never a quiet POST somewhere else — ValidPath runs the same
// parser at declaration, so this only fires for a row written some other
// way, and the runner records it on the row rather than delivering.
func BuildRequest(rt registry.Route, t registry.Timer, token, build string) (*http.Request, error) {
	if err := ValidPath(t.Path); err != nil {
		return nil, err
	}
	req, err := http.NewRequest(http.MethodPost, "http://"+rt.Host+t.Path, http.NoBody)
	if err != nil {
		return nil, err
	}
	req.Host = rt.Host
	req.ContentLength = 0
	req.Header.Set("Authorization", "Bearer "+token)
	req.Header.Set("X-Carlos-Schedule", t.Name)
	req.Header.Set("X-Carlos-Schedule-Kind", t.Kind)
	req.Header.Set("X-Carlos-Schedule-At", strconv.FormatInt(dueOf(t), 10))
	req.Header.Set("User-Agent", "carlos-schedule/"+build)
	return req, nil
}

// TokenFromEnvPaths is the production Token with the env file's location
// taken from envPath rather than assumed flat: isolated-exec's nested rows
// keep <host>.env inside their own instance directory (design §7), and the
// host wiring (schedule_wiring.go) resolves that through the same
// classification the exec backend used to decide where it spawned the
// instance. Every caller passes a resolver here now; a caller that predates
// the nested layout gets one that always names the flat path.
func TokenFromEnvPaths(envPath func(host string) string) func(host string) (string, error) {
	return func(host string) (string, error) {
		env, err := readEnvFile(envPath(host))
		if err != nil {
			return "", err
		}
		if env["CARLOS_ADMIN_TOKEN"] == "" {
			return "", errors.New("no CARLOS_ADMIN_TOKEN")
		}
		return env["CARLOS_ADMIN_TOKEN"], nil
	}
}