package schedule

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

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

type fakeDeliverer struct {
	mu     sync.Mutex
	status int
	err    error
	calls  []*http.Request
	hosts  []string
}

func (f *fakeDeliverer) Deliver(_ context.Context, rt registry.Route, req *http.Request) (int, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.calls = append(f.calls, req)
	f.hosts = append(f.hosts, rt.Host)
	return f.status, f.err
}

// testClock is the Now seam for the tests that advance time WHILE a delivery
// goroutine is running: the plain `now := …; r.Now = func() …` shape those
// tests would otherwise use is a data race under -race, and the runner reads
// its clock from inside every host goroutine.
type testClock struct {
	mu sync.Mutex
	t  time.Time
}

func (c *testClock) now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.t
}

func (c *testClock) advance(d time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.t = c.t.Add(d)
}

// firedHosts counts deliveries per host — the counting assertion the
// parallelism and exactly-once tests need.
func (f *fakeDeliverer) firedHosts() map[string]int {
	f.mu.Lock()
	defer f.mu.Unlock()
	n := map[string]int{}
	for _, h := range f.hosts {
		n[h]++
	}
	return n
}

func runnerFixture(t *testing.T, d Deliverer) (*registry.Registry, *Runner) {
	t.Helper()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	if err := reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
		t.Fatal(err)
	}
	r := &Runner{Reg: reg, Deliver: d, Timeout: time.Minute, Box: "box1", Build: "test",
		Token: func(string) (string, error) { return "tok", nil }}
	return reg, r
}

func TestClassify(t *testing.T) {
	cases := []struct {
		status int
		err    error
		want   Outcome
	}{
		{204, nil, Outcome{Status: "204", Retry: false}},
		{200, nil, Outcome{Status: "200", Retry: false}},
		{404, nil, Outcome{Status: "404 (app refused)", Retry: false}},
		{403, nil, Outcome{Status: "403 (app refused)", Retry: false}},
		{302, nil, Outcome{Status: "302 (app refused)", Retry: false}},
		{500, nil, Outcome{Status: "500", Retry: true}},
		{503, nil, Outcome{Status: "503", Retry: true}},
		{429, nil, Outcome{Status: "429", Retry: true}},
		{408, nil, Outcome{Status: "408", Retry: true}},
		{0, errors.New("instance failed to wake"), Outcome{Status: "503: instance failed to wake", Retry: true}},
		{0, ErrTimeout, Outcome{Status: "timeout after 1m0s", Retry: false}},
	}
	for _, c := range cases {
		if got := classify(c.status, c.err, time.Minute); got != c.want {
			t.Errorf("classify(%d, %v) = %+v, want %+v", c.status, c.err, got, c.want)
		}
	}
}

func TestBuildRequestCarriesTheContract(t *testing.T) {
	rt := registry.Route{Host: "a.bab.oncarlos.com"}
	tm := registry.Timer{Name: "sync", Kind: KindSchedule, Path: "/jobs/sync?x=1", NextAt: 1787004060, DueAt: 1787004000}
	req, err := BuildRequest(rt, tm, "tok", "abc")
	if err != nil {
		t.Fatal(err)
	}
	if req.Method != http.MethodPost || req.Host != rt.Host || req.URL.Path != "/jobs/sync" || req.URL.RawQuery != "x=1" {
		t.Fatalf("req = %s %s %s", req.Method, req.Host, req.URL)
	}
	h := req.Header
	if h.Get("Authorization") != "Bearer tok" || h.Get("X-Carlos-Schedule") != "sync" || h.Get("X-Carlos-Schedule-Kind") != "schedule" ||
		h.Get("X-Carlos-Schedule-At") != "1787004000" || h.Get("User-Agent") != "carlos-schedule/abc" || req.ContentLength != 0 {
		t.Fatalf("headers = %v", h)
	}
	// A row from before due_at existed carries 0 there: the header falls
	// back to next_at, which at attempt 0 is the occurrence.
	tm.DueAt = 0
	if req, _ := BuildRequest(rt, tm, "tok", "abc"); req.Header.Get("X-Carlos-Schedule-At") != "1787004060" {
		t.Fatalf("legacy row At = %q", req.Header.Get("X-Carlos-Schedule-At"))
	}
	// A path the parser refuses is an error, never a POST to "/".
	tm.Path = "/jobs/%zz"
	if _, err := BuildRequest(rt, tm, "tok", "abc"); err == nil {
		t.Fatal("bad path built a request")
	}
}

func TestTickFiresDueAdvancesScheduleAndFinishesOnce(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix() - 5})
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "later", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/l", NextAt: now.Unix() + 60})
	var events []string
	r.Event = func(_ registry.Route, level, msg string) { events = append(events, level+": "+msg) }
	var reports int
	r.Report = func(registry.Route, Status) { reports++ }
	r.tick()
	if len(d.calls) != 2 {
		t.Fatalf("delivered %d, want 2 (not the future one)", len(d.calls))
	}
	sync, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
	if sync.LastStatus != "204" || sync.LastAt != now.Unix() || sync.NextAt != at("2026-08-23T18:00:00Z").Unix() || sync.Attempts != 0 {
		t.Fatalf("sync = %+v", sync)
	}
	once, _ := reg.GetTimer("a.bab.oncarlos.com", "remind")
	if once.NextAt != 0 || once.LastStatus != "204" {
		t.Fatalf("once = %+v", once)
	}
	if len(events) != 2 || events[0][:5] != "info:" {
		t.Fatalf("events = %v", events)
	}
	if reports != 1 { // one report per host per tick, not per timer
		t.Fatalf("reports = %d", reports)
	}
}

func TestTickRetriesWithBackoffThenGivesUp(t *testing.T) {
	d := &fakeDeliverer{status: 503}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix()})
	wantBackoff := []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute}
	for i, b := range wantBackoff {
		r.tick()
		tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
		if tm.Attempts != i+1 || tm.NextAt != now.Add(b).Unix() || tm.LastStatus != "503" {
			t.Fatalf("attempt %d: %+v (want next=+%s)", i+1, tm, b)
		}
		now = now.Add(b)
	}
	r.tick() // 5th failure: give up, advance to the natural next
	tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
	if tm.Attempts != 0 || tm.LastStatus != "failed after 5 attempts: 503" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
		t.Fatalf("after giving up: %+v", tm)
	}
}

func TestBackoffNeverPassesTheNaturalNext(t *testing.T) {
	d := &fakeDeliverer{status: 503}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:59:30Z") // next natural 1m fire is 14:00:00, 30s away
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "m", Kind: KindSchedule, Spec: "every 1m0s", Path: "/m", NextAt: now.Unix()})
	r.tick()
	tm, _ := reg.GetTimer("a.bab.oncarlos.com", "m")
	if tm.NextAt != at("2026-08-23T14:00:00Z").Unix() {
		t.Fatalf("next = %d, want the natural next, not now+1m", tm.NextAt)
	}
}

func TestAppRefusalDoesNotRetry(t *testing.T) {
	d := &fakeDeliverer{status: 404}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "o", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 1})
	var levels []string
	r.Event = func(_ registry.Route, level, _ string) { levels = append(levels, level) }
	r.tick()
	tm, _ := reg.GetTimer("a.bab.oncarlos.com", "o")
	if tm.NextAt != 0 || tm.LastStatus != "404 (app refused)" || tm.Attempts != 0 {
		t.Fatalf("%+v", tm)
	}
	if len(levels) != 1 || levels[0] != "warn" {
		t.Fatalf("levels = %v", levels)
	}
}

func TestTimeoutCountsAsDelivered(t *testing.T) {
	d := &fakeDeliverer{err: ErrTimeout}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
	r.tick()
	tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
	if tm.Attempts != 0 || tm.LastStatus != "timeout after 1m0s" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
		t.Fatalf("%+v", tm)
	}
}

func TestNoTokenIsRecordedNotDelivered(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	r.Token = func(string) (string, error) { return "", errors.New("no CARLOS_ADMIN_TOKEN") }
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: now.Unix()})
	r.tick()
	if len(d.calls) != 0 {
		t.Fatal("delivered without a token")
	}
	tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
	if tm.Attempts != 1 || tm.LastStatus != "no instance token: no CARLOS_ADMIN_TOKEN" {
		t.Fatalf("%+v", tm)
	}
}

func TestTickSkipsRowsWhoseRouteIsGone(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/s", NextAt: 1})
	r.tick()
	if len(d.calls) != 0 {
		t.Fatal("delivered to a host with no route")
	}
}

// X-Carlos-Schedule-At is the OCCURRENCE, the key an app dedupes on (spec
// §6.3, §7.1): a retry re-sends the same value, a fire that advances the row
// moves it to the new natural occurrence, and a `run` makes now the
// occurrence. next_at is none of those things after the first retry.
func TestScheduleAtHeaderIsTheOccurrenceNotNextAt(t *testing.T) {
	d := &fakeDeliverer{status: 503}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	occurrence := now.Unix() - 5
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: occurrence, DueAt: occurrence})
	atHeader := func(i int) string { return d.calls[i].Header.Get("X-Carlos-Schedule-At") }

	r.tick() // fails → retry in 1m
	tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync")
	if tm.NextAt != now.Add(time.Minute).Unix() || tm.DueAt != occurrence {
		t.Fatalf("after first failure: %+v", tm)
	}
	now = now.Add(time.Minute)
	r.tick() // the retry: same occurrence on the wire
	if len(d.calls) != 2 || atHeader(0) != itoa(occurrence) || atHeader(1) != itoa(occurrence) {
		t.Fatalf("retry header = %q, first = %q, want both %d", atHeader(1), atHeader(0), occurrence)
	}

	d.status = 204
	now = now.Add(2 * time.Minute)
	r.tick() // succeeds → advances
	tm, _ = reg.GetTimer("a.bab.oncarlos.com", "sync")
	next := at("2026-08-23T18:00:00Z").Unix()
	if tm.NextAt != next || tm.DueAt != next || atHeader(2) != itoa(occurrence) {
		t.Fatalf("after advance: %+v (header %q)", tm, atHeader(2))
	}

	// `run`: FireTimerNow makes now the occurrence, and the tick sends it.
	now = now.Add(time.Minute)
	reg.FireTimerNow("a.bab.oncarlos.com", "sync", now.Unix())
	r.tick()
	if len(d.calls) != 4 || atHeader(3) != itoa(now.Unix()) {
		t.Fatalf("run header = %q, want %d", atHeader(3), now.Unix())
	}
}

// A row whose path the URL parser refuses is recorded, not retried and not
// delivered anywhere else: ValidPath runs the same parser at declaration, so
// such a row was written some other way and is exactly as bad next time.
func TestBadPathIsRecordedNotDelivered(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/%zz", NextAt: now.Unix()})
	r.tick()
	if len(d.calls) != 0 {
		t.Fatal("delivered a request the parser refused")
	}
	tm, _ := reg.GetTimer("a.bab.oncarlos.com", "s")
	if tm.Attempts != 0 || !strings.HasPrefix(tm.LastStatus, "bad path: ") || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
		t.Fatalf("%+v", tm)
	}
}

// tick is Tick plus a wait for every host goroutine it launched — the shape
// every test here asserts against, since Tick itself returns as soon as the
// work is in flight.
func (r *Runner) tick() {
	r.Tick(context.Background())
	r.wait()
}

// blockingDeliverer answers 204 at once for every host except `slow`, whose
// delivery parks on `gate` until the test closes it — a 20-minute handler,
// in miniature.
type blockingDeliverer struct {
	fakeDeliverer
	slow string
	gate chan struct{}
}

func (b *blockingDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {
	if rt.Host == b.slow {
		<-b.gate
	}
	return b.fakeDeliverer.Deliver(ctx, rt, req)
}

// One long handler must not hold every other host's timers: the tick
// launches per-host work and returns; while one host is parked, the others
// all fire; and a later tick does NOT start a second delivery to the parked
// host (its rows are still due — per-host serialization spans ticks).
func TestOneSlowHostDoesNotBlockTheOthersOrRefire(t *testing.T) {
	d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: "slow.bab.oncarlos.com", gate: make(chan struct{})}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	clock := &testClock{t: now}
	r.Now = clock.now
	hosts := []string{"slow.bab.oncarlos.com", "b.bab.oncarlos.com", "c.bab.oncarlos.com", "d.bab.oncarlos.com", "e.bab.oncarlos.com", "f.bab.oncarlos.com", "g.bab.oncarlos.com"}
	for i, h := range hosts {
		if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
			t.Fatal(err)
		}
		// The slow host is due FIRST, so it takes a slot before anyone else.
		reg.UpsertTimer(registry.Timer{Host: h, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 100 + int64(i)})
	}
	if len(hosts) <= MaxParallelHosts {
		t.Fatalf("fixture has %d hosts; the test needs more than MaxParallelHosts (%d) so slots recycle under the parked one", len(hosts), MaxParallelHosts)
	}

	done := make(chan struct{})
	go func() { r.Tick(context.Background()); close(done) }()
	select {
	case <-done:
	case <-time.After(10 * time.Second):
		t.Fatal("Tick did not return while one host was parked")
	}
	// Every other host fires AND is recorded while slow is parked — record
	// writes the row after Deliver returns, so poll the recorded state
	// itself rather than the delivery count (which races ahead of it).
	deadline := time.Now().Add(10 * time.Second)
	for {
		recorded := 0
		for _, h := range hosts[1:] {
			if tm, _ := reg.GetTimer(h, "j"); tm != nil && tm.LastStatus != "" {
				recorded++
			}
		}
		if recorded == len(hosts)-1 && d.firedHosts()[d.slow] == 0 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("fired = %v, want every host but %s recorded", d.firedHosts(), d.slow)
		}
		time.Sleep(5 * time.Millisecond)
	}
	for _, h := range hosts[1:] {
		if tm, _ := reg.GetTimer(h, "j"); tm.LastStatus != "204" {
			t.Fatalf("%s not recorded while slow was parked: %+v", h, tm)
		}
	}

	// The parked host's row is still due; a second tick must skip it.
	clock.advance(time.Minute)
	r.Tick(context.Background())
	close(d.gate)
	r.wait()
	if n := d.firedHosts()[d.slow]; n != 1 {
		t.Fatalf("slow host delivered %d times, want exactly 1 — a busy host was re-fired", n)
	}
	if tm, _ := reg.GetTimer(d.slow, "j"); tm.LastStatus != "204" || tm.NextAt != at("2026-08-23T14:00:00Z").Unix() {
		t.Fatalf("slow row after release: %+v", tm)
	}
	// And once released, it is fair game again.
	clock.advance(at("2026-08-23T14:00:01Z").Sub(clock.now()))
	r.tick()
	if n := d.firedHosts()[d.slow]; n != 2 {
		t.Fatalf("released host delivered %d times, want 2", n)
	}
}

// The prune leg is the RUNNER's, not the converger's, because a box with the
// activator armed and no deploy store has no converger at all (spec §6.2) —
// and it is exactly the box whose rows arrive on the control socket. Without
// this leg its finished once-rows accumulate forever.
func TestRunnerPrunesWithoutAConverger(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	if r.Report != nil {
		t.Fatal("fixture has a reporter — this test is the storeless box")
	}
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	const host = "a.bab.oncarlos.com"
	old := now.Add(-8 * 24 * time.Hour).Unix()
	reg.UpsertTimer(registry.Timer{Host: host, Name: "done-long-ago", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 0, LastAt: old})
	reg.UpsertTimer(registry.Timer{Host: host, Name: "done-just-now", Kind: KindOnce, Spec: "at 1", Path: "/o", NextAt: 0, LastAt: now.Add(-time.Hour).Unix()})
	reg.UpsertTimer(registry.Timer{Host: "ghost.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: 0, LastAt: now.Unix()})
	r.tick()
	if tm, _ := reg.GetTimer(host, "done-long-ago"); tm != nil {
		t.Fatalf("a finished once-row 8 days old survived the runner's prune: %+v", tm)
	}
	if tm, _ := reg.GetTimer("ghost.bab.oncarlos.com", "s"); tm != nil {
		t.Fatalf("a row for a host with no route survived: %+v", tm)
	}
	if tm, _ := reg.GetTimer(host, "done-just-now"); tm == nil {
		t.Fatal("a finished once-row inside the retention window was pruned")
	}

	// Daily, not every tick: the stamp is what keeps a 15s tick from running
	// a DELETE 5,760 times a day.
	reg.UpsertTimer(registry.Timer{Host: "ghost2.bab.oncarlos.com", Name: "s", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: 0, LastAt: now.Unix()})
	now = now.Add(TickEvery)
	r.tick()
	if tm, _ := reg.GetTimer("ghost2.bab.oncarlos.com", "s"); tm == nil {
		t.Fatal("the prune leg ran again 15s later — the stamp is not holding")
	}
	now = now.Add(pruneEvery)
	r.tick()
	if tm, _ := reg.GetTimer("ghost2.bab.oncarlos.com", "s"); tm != nil {
		t.Fatalf("the prune leg never came round again: %+v", tm)
	}
}

// retainingDeliverer is a Deliverer that also keeps per-socket state — the
// agent's does (one connection pool per socket).
type retainingDeliverer struct {
	fakeDeliverer
	mu    sync.Mutex
	calls []map[string]bool
}

func (d *retainingDeliverer) RetainSockets(live map[string]bool) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.calls = append(d.calls, live)
}

func (d *retainingDeliverer) sweeps() []map[string]bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	return append([]map[string]bool(nil), d.calls...)
}

// The prune leg doubles as the delivery leg's eviction point: it has already
// read every route, so it hands the live SOCKET set to a Deliverer that keeps
// per-socket state. A retired route can be reclaimed no other way — nothing
// delivers to it, so nothing it holds can error its own way out.
func TestPruneSweepHandsTheLiveSocketsToTheDeliverer(t *testing.T) {
	d := &retainingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	if err := reg.Upsert(registry.Route{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Socket: "/run/carlos/a.sock"}); err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "site.bab.oncarlos.com", Kind: "static", Account: "bab", App: "site", Status: "ready"}); err != nil {
		t.Fatal(err)
	}
	r.tick()
	sweeps := d.sweeps()
	if len(sweeps) != 1 {
		t.Fatalf("%d sweeps, want the one the prune leg runs", len(sweeps))
	}
	if !sweeps[0]["/run/carlos/a.sock"] || len(sweeps[0]) != 1 {
		t.Fatalf("live sockets = %v, want exactly the routed instance's (a socketless route contributes nothing)", sweeps[0])
	}
	// Daily, with the prune itself: not once per tick.
	now = now.Add(TickEvery)
	r.tick()
	if got := len(d.sweeps()); got != 1 {
		t.Fatalf("%d sweeps after a second tick, want still 1", got)
	}
}

// A busy host must not crowd another host's due timer out of the batch. The
// runner skips a host whose delivery is still running, so its rows are dead
// weight in a capped query: the exclusion belongs in DueTimers, before the
// LIMIT. Sized to the real batch so the test fails if dueBatch and the
// fixture ever drift apart.
func TestBusyHostDoesNotStarveAnotherHostsTimer(t *testing.T) {
	d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: "slow.bab.oncarlos.com", gate: make(chan struct{})}
	reg, r := runnerFixture(t, d)
	clk := &testClock{t: at("2026-08-23T13:07:00Z")}
	now := clk.now()
	r.Now = clk.now
	for _, h := range []string{d.slow, "quiet.bab.oncarlos.com"} {
		if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
			t.Fatal(err)
		}
	}
	// A full batch of rows on the parked host, every one of them due BEFORE
	// the quiet host's single row.
	for i := 0; i < dueBatch; i++ {
		reg.UpsertTimer(registry.Timer{Host: d.slow, Name: fmt.Sprintf("j%02d", i), Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1000 + int64(i)})
	}
	reg.UpsertTimer(registry.Timer{Host: "quiet.bab.oncarlos.com", Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1})

	// First tick: the parked host claims a slot and stops there. Tick claims
	// synchronously, so the host is busy the moment it returns.
	r.Tick(context.Background())
	if busy := r.busyHosts(); len(busy) != 1 || busy[0] != d.slow {
		t.Fatalf("after the first tick busy = %v, want just the parked host", busy)
	}

	// Second tick, while it is still parked: the quiet host's row is due and
	// must be delivered and recorded.
	clk.advance(TickEvery)
	r.Tick(context.Background())
	deadline := time.Now().Add(10 * time.Second)
	for time.Now().Before(deadline) {
		if tm, _ := reg.GetTimer("quiet.bab.oncarlos.com", "j"); tm != nil && tm.LastStatus == "204" {
			break
		}
		time.Sleep(time.Millisecond)
	}
	if tm, _ := reg.GetTimer("quiet.bab.oncarlos.com", "j"); tm == nil || tm.LastStatus != "204" {
		t.Fatalf("the quiet host's timer never fired behind a full batch of a busy host's rows: %+v", tm)
	}
	close(d.gate)
	r.wait()
}

// countingDeliverer parks every delivery on a gate the test opens, and
// signals two things about how many ran at once: reached (exactly
// MaxParallelHosts in flight) and over (MORE than that — the regression).
type countingDeliverer struct {
	fakeDeliverer
	gate     chan struct{}
	mu2      sync.Mutex
	cur      int
	max      int
	reached  chan struct{}
	over     chan struct{}
	onceFull sync.Once
	onceOver sync.Once
}

func (c *countingDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) {
	c.mu2.Lock()
	c.cur++
	if c.cur > c.max {
		c.max = c.cur
	}
	full, over := c.cur == MaxParallelHosts, c.cur > MaxParallelHosts
	c.mu2.Unlock()
	if full {
		c.onceFull.Do(func() { close(c.reached) })
	}
	if over {
		c.onceOver.Do(func() { close(c.over) })
	}
	<-c.gate
	c.mu2.Lock()
	c.cur--
	c.mu2.Unlock()
	return c.fakeDeliverer.Deliver(ctx, rt, req)
}

func (c *countingDeliverer) peak() int {
	c.mu2.Lock()
	defer c.mu2.Unlock()
	return c.max
}

// boundWindow is how long the bound is watched with every delivery parked.
// It is a floor on how long we LOOK, never a deadline anything must meet:
// both assertions inside it are negative (a fifth delivery must not start,
// Tick must not return), so a slow machine can only make the test weaker,
// never redder. A widened bound trips both within microseconds.
const boundWindow = 500 * time.Millisecond

// The ≤4-parallel path: with more due hosts than MaxParallelHosts, exactly
// four deliveries run at once — no fewer (they are genuinely parallel, not
// one host at a time) and no more (the semaphore is the bound) — and every
// host still completes once the slots recycle.
func TestDeliveriesAreParallelToTheBoundAndAllComplete(t *testing.T) {
	d := &countingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, gate: make(chan struct{}), reached: make(chan struct{}), over: make(chan struct{})}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	const hosts = 2 * MaxParallelHosts
	var names []string
	for i := 0; i < hosts; i++ {
		h := fmt.Sprintf("h%d.bab.oncarlos.com", i)
		names = append(names, h)
		if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
			t.Fatal(err)
		}
		reg.UpsertTimer(registry.Timer{Host: h, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 100 + int64(i)})
	}

	tickDone := make(chan struct{})
	go func() { r.Tick(context.Background()); close(tickDone) }()
	select {
	case <-d.reached:
	case <-time.After(10 * time.Second):
		t.Fatalf("only %d deliveries ever ran at once, want %d in parallel", d.peak(), MaxParallelHosts)
	}
	// The BOUND, watched while the gate is still shut and all four slots are
	// held. Two independent ways a widened semaphore shows itself, and the
	// test must be red on either:
	//
	//   - a fifth delivery starts (`over`);
	//   - Tick RETURNS. With 8 hosts and 4 slots it cannot: it takes the slot
	//     before launching each host's goroutine, so it is parked on the
	//     semaphore until a delivery finishes — and none can while the gate
	//     is shut.
	//
	// Watching the peak after the fact cannot do this job: opening the gate
	// drains `cur` faster than a fifth delivery would ever be observed, which
	// is why the earlier shape passed 25/25 with the semaphore quadrupled.
	select {
	case <-d.over:
		t.Fatalf("%d deliveries in flight at once, want at most MaxParallelHosts (%d)", d.peak(), MaxParallelHosts)
	case <-tickDone:
		t.Fatalf("Tick returned with every delivery still parked — it launched more than MaxParallelHosts (%d) hosts (peak %d)", MaxParallelHosts, d.peak())
	case <-time.After(boundWindow):
	}

	close(d.gate) // every parked delivery finishes; the waiting hosts take the slots
	select {
	case <-tickDone:
	case <-time.After(10 * time.Second):
		t.Fatal("Tick never returned after the gate opened")
	}
	r.wait()
	if got := d.peak(); got > MaxParallelHosts { // belt-and-braces; the window above is the real assertion
		t.Fatalf("%d deliveries in flight at once, want at most MaxParallelHosts (%d)", got, MaxParallelHosts)
	}
	for _, h := range names {
		tm, _ := reg.GetTimer(h, "j")
		if tm == nil || tm.LastStatus != "204" {
			t.Fatalf("%s did not complete: %+v", h, tm)
		}
	}
	if fired := d.firedHosts(); len(fired) != hosts {
		t.Fatalf("delivered to %d hosts, want %d", len(fired), hosts)
	}
}

type timerMutationDeliverer struct {
	mutate func()
	status int
}

func (d timerMutationDeliverer) Deliver(context.Context, registry.Route, *http.Request) (int, error) {
	d.mutate()
	return d.status, nil
}

func TestDeliveryCompletionPreservesReplacementRegistration(t *testing.T) {
	for _, mutation := range []string{"identical", "timing", "path", "cancel", "recreate", "manual"} {
		for _, status := range []int{204, 503} {
			t.Run(fmt.Sprintf("%s/%d", mutation, status), func(t *testing.T) {
				reg, runner := runnerFixture(t, nil)
				now := at("2026-08-23T13:07:00Z")
				runner.Now = func() time.Time { return now }
				const host = "a.bab.oncarlos.com"
				mux := http.NewServeMux()
				Mount(mux, host, reg, func() (string, error) { return "tok", nil }, runner.Now)
				put := func(when time.Time, path string) {
					req := httptest.NewRequest(http.MethodPut, "/schedule/remind", strings.NewReader(fmt.Sprintf(`{"at":%q,"path":%q}`, when.Format(time.RFC3339), path)))
					req.Header.Set("Authorization", "Bearer tok")
					w := httptest.NewRecorder()
					mux.ServeHTTP(w, req)
					if w.Code != http.StatusOK {
						t.Errorf("PUT: %d %s", w.Code, w.Body.String())
					}
				}
				put(now, "/jobs/remind")
				before, err := reg.GetTimer(host, "remind")
				if err != nil {
					t.Fatal(err)
				}
				var replacement *registry.Timer
				var events []string
				runner.Event = func(_ registry.Route, _, msg string) { events = append(events, msg) }
				runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
					switch mutation {
					case "identical":
						put(now, "/jobs/remind")
					case "timing":
						put(now.Add(time.Hour), "/jobs/remind")
					case "path":
						put(now, "/jobs/changed")
					case "cancel", "recreate":
						req := httptest.NewRequest(http.MethodDelete, "/schedule/remind", nil)
						req.Header.Set("Authorization", "Bearer tok")
						w := httptest.NewRecorder()
						mux.ServeHTTP(w, req)
						if w.Code != http.StatusNoContent {
							t.Errorf("DELETE: %d", w.Code)
						}
						if mutation == "recreate" {
							put(now, "/jobs/remind")
						}
					case "manual":
						if err := reg.FireTimerNow(host, "remind", now.Unix()); err != nil {
							t.Error(err)
						}
					}
					replacement, err = reg.GetTimer(host, "remind")
					if mutation != "cancel" && err != nil {
						t.Error(err)
					}
				}}
				runner.tick()
				if len(events) != 1 || strings.Contains(events[0], "retrying at") {
					t.Fatalf("superseded delivery event: %v", events)
				}
				after, err := reg.GetTimer(host, "remind")
				if mutation == "cancel" {
					if !errors.Is(err, os.ErrNotExist) {
						t.Fatalf("cancelled timer: %+v, %v", after, err)
					}
					return
				}
				if err != nil {
					t.Fatal(err)
				}
				if replacement == nil {
					t.Fatal("replacement missing")
				}
				if after.Registration == before.Registration || after.Registration == "" {
					t.Fatalf("registration did not change: %+v", after)
				}
				if *after != *replacement {
					t.Fatalf("completion changed replacement: got %+v, want %+v", after, replacement)
				}
				runner.Deliver = &fakeDeliverer{status: 204}
				now = time.Unix(after.NextAt, 0)
				runner.tick()
				finished, err := reg.GetTimer(host, "remind")
				if err != nil {
					t.Fatal(err)
				}
				if finished.NextAt != 0 || finished.LastStatus != "204" {
					t.Fatalf("replacement did not fire: %+v", finished)
				}
			})
		}
	}
}

func TestDeliveryCompletionSurvivesUnchangedDeclaration(t *testing.T) {
	for _, status := range []int{204, 503} {
		t.Run(fmt.Sprint(status), func(t *testing.T) {
			reg, runner := runnerFixture(t, nil)
			now := at("2026-08-23T13:07:00Z")
			runner.Now = func() time.Time { return now }
			tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindSchedule, Spec: "every 6h0m0s", Path: "/jobs/sync", NextAt: now.Unix(), DueAt: now.Unix()}
			if err := reg.UpsertTimer(tm); err != nil {
				t.Fatal(err)
			}
			before, err := reg.GetTimer(tm.Host, tm.Name)
			if err != nil {
				t.Fatal(err)
			}
			tm.NextAt, tm.DueAt = 0, 0
			runner.Deliver = timerMutationDeliverer{status: status, mutate: func() {
				if err := reg.UpsertTimer(tm); err != nil {
					t.Error(err)
				}
			}}
			runner.tick()
			after, err := reg.GetTimer(tm.Host, tm.Name)
			if err != nil {
				t.Fatal(err)
			}
			if after.Registration != before.Registration || after.LastStatus != fmt.Sprint(status) {
				t.Fatalf("completion missing: %+v", after)
			}
			if status == 503 && (after.Attempts != 1 || after.DueAt != before.DueAt || after.NextAt != now.Add(time.Minute).Unix()) {
				t.Fatalf("retry changed: %+v", after)
			}
			if err := reg.UpsertTimer(tm); err != nil {
				t.Fatal(err)
			}
			converged, err := reg.GetTimer(tm.Host, tm.Name)
			if err != nil {
				t.Fatal(err)
			}
			if *converged != *after {
				t.Fatalf("converge changed completed state: got %+v, want %+v", converged, after)
			}
		})
	}
}

func TestInvalidTimerSnapshotCannotDeleteReplacement(t *testing.T) {
	reg, runner := runnerFixture(t, &fakeDeliverer{status: 204})
	tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "invalid", Path: "/jobs/remind", NextAt: 100}
	if err := reg.UpsertTimer(tm); err != nil {
		t.Fatal(err)
	}
	old, err := reg.GetTimer(tm.Host, tm.Name)
	if err != nil {
		t.Fatal(err)
	}
	tm.Spec = "at 100"
	if err := reg.UpsertTimer(tm); err != nil {
		t.Fatal(err)
	}
	replacement, err := reg.GetTimer(tm.Host, tm.Name)
	if err != nil {
		t.Fatal(err)
	}
	runner.fire(context.Background(), registry.Route{Host: tm.Host}, *old)
	after, err := reg.GetTimer(tm.Host, tm.Name)
	if err != nil {
		t.Fatal(err)
	}
	if *after != *replacement {
		t.Fatalf("invalid snapshot changed replacement: got %+v, want %+v", after, replacement)
	}
}

func TestBatchSkipsTimerChangedByEarlierDelivery(t *testing.T) {
	for _, mutation := range []string{"cancel", "rearm"} {
		t.Run(mutation, func(t *testing.T) {
			reg, runner := runnerFixture(t, nil)
			now := at("2026-08-23T13:07:00Z")
			runner.Now = func() time.Time { return now }
			const host = "a.bab.oncarlos.com"
			for _, name := range []string{"first", "later"} {
				if err := reg.UpsertTimer(registry.Timer{Host: host, Name: name, Kind: KindOnce, Spec: "at 100", Path: "/jobs/" + name, NextAt: 100}); err != nil {
					t.Fatal(err)
				}
			}
			calls := 0
			runner.Deliver = timerMutationDeliverer{status: 204, mutate: func() {
				calls++
				if mutation == "cancel" {
					if err := reg.DeleteTimer(host, "later"); err != nil {
						t.Error(err)
					}
				} else {
					if err := reg.FireTimerNow(host, "later", now.Add(time.Hour).Unix()); err != nil {
						t.Error(err)
					}
				}
			}}
			runner.tick()
			if calls != 1 {
				t.Fatalf("delivered %d timers, want first only", calls)
			}
		})
	}
}

func TestConcurrentTicksSkipSnapshotChangedDuringTokenLookup(t *testing.T) {
	for _, mutation := range []string{"cancel", "rearm", "retry", "retry-still-due", "done"} {
		t.Run(mutation, func(t *testing.T) {
			deliver := &fakeDeliverer{status: 204}
			reg, runner := runnerFixture(t, deliver)
			now := at("2026-08-23T13:07:00Z")
			runner.Now = func() time.Time { return now }
			tm := registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 100", Path: "/jobs/remind", NextAt: 100}
			if err := reg.UpsertTimer(tm); err != nil {
				t.Fatal(err)
			}
			before, err := reg.GetTimer(tm.Host, tm.Name)
			if err != nil {
				t.Fatal(err)
			}
			entered, release := make(chan struct{}), make(chan struct{})
			var tokenOnce sync.Once
			runner.Token = func(string) (string, error) { tokenOnce.Do(func() { close(entered) }); <-release; return "tok", nil }
			runner.Tick(context.Background())
			<-entered
			switch mutation {
			case "cancel":
				err = reg.DeleteTimer(tm.Host, tm.Name)
			case "rearm":
				err = reg.FireTimerNow(tm.Host, tm.Name, now.Add(time.Hour).Unix())
			case "retry":
				_, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Add(time.Minute).Unix(), 100)
			case "retry-still-due":
				_, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "503", 0, 1, now.Unix()-1, 100)
			case "done":
				_, err = reg.MarkTimerFired(tm.Host, tm.Name, before.Registration, now.Unix(), "204", 0, 0, 0, 0)
			}
			if err != nil {
				close(release)
				runner.wait()
				t.Fatal(err)
			}
			runner.Tick(context.Background())
			close(release)
			runner.wait()
			runner.fire(context.Background(), registry.Route{Host: tm.Host}, *before)
			if len(deliver.calls) != 0 {
				t.Fatalf("delivered stale snapshot %d times", len(deliver.calls))
			}
		})
	}
}

func TestNonpositiveOccurrenceDeliveredAndRetriedWithOriginalTimestamp(t *testing.T) {
	for _, requested := range []int64{0, -1, -62135596800} {
		t.Run(fmt.Sprint(requested), func(t *testing.T) {
			deliver := &fakeDeliverer{status: 503}
			reg, runner := runnerFixture(t, deliver)
			now := at("2026-08-23T13:07:00Z")
			runner.Now = func() time.Time { return now }
			mux := http.NewServeMux()
			Mount(mux, "a.bab.oncarlos.com", reg, func() (string, error) { return "tok", nil }, runner.Now)
			rec := do(mux, "PUT", "/schedule/remind", "Bearer tok", `{"at":"`+time.Unix(requested, 0).UTC().Format(time.RFC3339)+`","path":"/jobs/remind"}`)
			if rec.Code != http.StatusOK {
				t.Fatalf("PUT: %d %s", rec.Code, rec.Body.String())
			}
			runner.tick()
			row, err := reg.GetTimer("a.bab.oncarlos.com", "remind")
			if err != nil {
				t.Fatal(err)
			}
			if row.Attempts != 1 || row.NextAt != now.Add(time.Minute).Unix() || row.DueAt != requested {
				t.Fatalf("retry: %+v", row)
			}
			now = now.Add(time.Minute)
			deliver.status = 204
			runner.tick()
			if len(deliver.calls) != 2 {
				t.Fatalf("deliveries: %d", len(deliver.calls))
			}
			for _, req := range deliver.calls {
				if got := req.Header.Get("X-Carlos-Schedule-At"); got != fmt.Sprint(requested) {
					t.Fatalf("occurrence: %q, want %d", got, requested)
				}
			}
			row, err = reg.GetTimer("a.bab.oncarlos.com", "remind")
			if err != nil {
				t.Fatal(err)
			}
			if row.NextAt != 0 || row.LastStatus != "204" {
				t.Fatalf("completed: %+v", row)
			}
		})
	}
}

// ---- move holds (moves spec §9.4) ----

// holdHost is the runner fixture's host, held by a move in these tests.
const holdHost = "a.bab.oncarlos.com"

// timersOf is the host's whole timers table, for before/after comparisons.
func timersOf(t *testing.T, reg *registry.Registry, host string) []registry.Timer {
	t.Helper()
	rows, err := reg.TimersFor(host)
	if err != nil {
		t.Fatal(err)
	}
	return rows
}

func sameTimers(a, b []registry.Timer) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

// Witness wTimerFreeze: a held row's due timers are not delivered and their
// bookkeeping does not move (before this, a refused delivery advanced
// attempts); a frozen host's declared schedule is neither upserted nor
// pruned by Converger.Pass; once the hold lifts, both resume.
func TestHeldRowsFreezeTimers(t *testing.T) {
	d := &fakeDeliverer{status: 503} // a delivery would retry and advance attempts
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	r.Gate = &HostGate{}
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1, DueAt: 1})
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "stale", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: now.Unix() - 5})
	if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	before := timersOf(t, reg, holdHost)
	r.tick()
	if len(d.calls) != 0 {
		t.Fatalf("a held row's timers were delivered %d time(s)", len(d.calls))
	}
	if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
		t.Fatalf("a held row's bookkeeping moved:\n got %+v\nwant %+v", got, before)
	}
	if r.Busy(holdHost) {
		t.Fatal("the runner kept a claim on a held host")
	}

	// The converger, with the host frozen: the declaration is not upserted
	// and the undeclared "stale" schedule is not pruned.
	st := release.NewDirStore(t.TempDir())
	ctx := context.Background()
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	frozen := true
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	c.Gate = r.Gate
	c.Frozen = func(host string) bool { return host == holdHost && frozen }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
		t.Fatalf("Converger.Pass wrote a frozen host's timers:\n got %+v\nwant %+v", got, before)
	}

	// The hold lifts: the runner delivers and the converger converges.
	if ok, err := reg.ReleaseHold(holdHost, "mv_a"); err != nil || !ok {
		t.Fatalf("release hold = %v %v", ok, err)
	}
	frozen = false
	r.tick()
	if len(d.calls) == 0 {
		t.Fatal("nothing was delivered after the hold lifted")
	}
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := reg.GetTimer(holdHost, "sync"); err != nil {
		t.Fatalf("the declaration did not converge after the freeze: %v", err)
	}
	if tm, _ := reg.GetTimer(holdHost, "stale"); tm != nil {
		t.Fatalf("the undeclared schedule survived the converge after the freeze: %+v", tm)
	}
}

// A hold written while a host's goroutine is delivering stops the rest of
// that host's due timers: the next timer re-reads the row first.
func TestHoldMidDeliveryStopsTheRest(t *testing.T) {
	reg, r := runnerFixture(t, nil)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	r.Gate = &HostGate{} // the hold wiring armed; off, both would be delivered (TestRunnerWithoutTheGateIsMain)
	var delivered []string
	r.Deliver = deliverFunc(func(rt registry.Route, req *http.Request) (int, error) {
		delivered = append(delivered, req.Header.Get("X-Carlos-Schedule"))
		if err := reg.SetHold(rt.Host, "moving", "mv_a"); err != nil {
			t.Error(err)
		}
		return 204, nil
	})
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "first", Kind: KindOnce, Spec: "at 1", Path: "/1", NextAt: 1})
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "second", Kind: KindOnce, Spec: "at 2", Path: "/2", NextAt: 2})
	r.tick()
	if len(delivered) != 1 || delivered[0] != "first" {
		t.Fatalf("delivered %v, want only the first", delivered)
	}
	second, _ := reg.GetTimer(holdHost, "second")
	if second.NextAt != 2 || second.Attempts != 0 || second.LastAt != 0 {
		t.Fatalf("the second timer's bookkeeping moved under the hold: %+v", second)
	}
}

type deliverFunc func(rt registry.Route, req *http.Request) (int, error)

func (f deliverFunc) Deliver(_ context.Context, rt registry.Route, req *http.Request) (int, error) {
	return f(rt, req)
}

// The cut's drain race: a tick reads the route while it is ready and pauses
// before its claim; the cut sets the hold and finds the host not busy, and
// its drain returns; the tick resumes, claims, re-reads the row inside the
// claim, sees the hold and lets go. Nothing is delivered and attempts does
// not move.
func TestDrainRaceCannotAdvanceBookkeeping(t *testing.T) {
	d := &fakeDeliverer{status: 503}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	r.Gate = &HostGate{}
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
	before := timersOf(t, reg, holdHost)
	reached, resume := make(chan struct{}), make(chan struct{})
	r.beforeClaim = func(host string) {
		if host == holdHost {
			close(reached)
			<-resume
		}
	}
	ticked := make(chan struct{})
	go func() { r.Tick(context.Background()); close(ticked) }()
	<-reached
	// The cut: hold, then drain.
	if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	if r.Busy(holdHost) {
		t.Fatal("the paused tick already held the claim")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := r.Gate.Drain(ctx, holdHost); err != nil {
		t.Fatalf("drain = %v", err)
	}
	close(resume)
	<-ticked
	r.wait()
	if len(d.calls) != 0 {
		t.Fatalf("the paused tick delivered %d time(s) after the drain", len(d.calls))
	}
	if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
		t.Fatalf("the paused tick moved bookkeeping after the drain:\n got %+v\nwant %+v", got, before)
	}
	if r.Busy(holdHost) {
		t.Fatal("the tick kept its claim after seeing the hold")
	}
}

// A delivery in flight when the hold is written holds up the drain until it
// has recorded its outcome: the snapshot after the drain sees the
// delivery's bookkeeping, never a write that lands after it.
func TestDrainWaitsForTheDeliveryInFlight(t *testing.T) {
	d := &blockingDeliverer{fakeDeliverer: fakeDeliverer{status: 204}, slow: holdHost, gate: make(chan struct{})}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	r.Gate = &HostGate{}
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
	r.Tick(context.Background())
	if !r.Busy(holdHost) {
		t.Fatal("the delivery is not in flight")
	}
	if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	drained := make(chan error, 1)
	go func() { drained <- r.Gate.Drain(context.Background(), holdHost) }()
	select {
	case err := <-drained:
		t.Fatalf("the drain returned (%v) with a delivery in flight", err)
	case <-time.After(50 * time.Millisecond):
	}
	close(d.gate)
	if err := <-drained; err != nil {
		t.Fatal(err)
	}
	snap := timersOf(t, reg, holdHost)
	r.wait()
	if snap[0].LastStatus != "204" || snap[0].NextAt != 0 {
		t.Fatalf("the drain returned before the delivery recorded: %+v", snap[0])
	}
	if got := timersOf(t, reg, holdHost); !sameTimers(got, snap) {
		t.Fatalf("the table changed after the drain:\n got %+v\nwant %+v", got, snap)
	}
}

// Fix round 1 (task-6-review, Important 1): a held host's rows must not
// consume the due batch. DueTimers applies its LIMIT before the runner's
// hold filter, so a full batch of held rows due before an unrelated host's
// row would be selected on every tick and that row never seen. With the
// hold wiring armed the query itself leaves held hosts out, and the
// unrelated timer runs this tick.
func TestHeldHostsDoNotStarveAnotherHostsTimer(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	r.Gate = &HostGate{}
	const quiet = "quiet.bab.oncarlos.com"
	if err := reg.Upsert(registry.Route{Host: quiet, Kind: "instance", Account: "bab", App: "app", Status: "ready"}); err != nil {
		t.Fatal(err)
	}
	// A full batch of rows on the held host, every one due BEFORE the quiet
	// host's single row.
	for i := 0; i < dueBatch; i++ {
		reg.UpsertTimer(registry.Timer{Host: holdHost, Name: fmt.Sprintf("j%02d", i), Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1000 + int64(i)})
	}
	reg.UpsertTimer(registry.Timer{Host: quiet, Name: "j", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/j", NextAt: now.Unix() - 1})
	if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	before := timersOf(t, reg, holdHost)
	r.tick()
	if tm, _ := reg.GetTimer(quiet, "j"); tm == nil || tm.LastStatus != "204" {
		t.Fatalf("the quiet host's timer did not fire this tick behind a full batch of a held host's rows: %+v", tm)
	}
	for _, h := range d.hosts {
		if h != quiet {
			t.Fatalf("delivered to %s, a held host", h)
		}
	}
	if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
		t.Fatalf("the held host's bookkeeping moved:\n got %+v\nwant %+v", got, before)
	}
}

// Fix round 1 (task-6-review, Important 3): with no gate (the Phase 1
// switch off, or no hold wiring on this box) the runner is main's code
// path. A row whose status is a hold is delivered like any other and its
// bookkeeping moves: the batch includes it, the pre-filter and the decision
// read do not look at the hold, and the between-timer check does not run.
func TestRunnerWithoutTheGateIsMain(t *testing.T) {
	d := &fakeDeliverer{status: 503}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "first", Kind: KindOnce, Spec: "at 1", Path: "/1", NextAt: 1, DueAt: 1})
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "second", Kind: KindOnce, Spec: "at 2", Path: "/2", NextAt: 2, DueAt: 2})
	if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	r.tick()
	if len(d.calls) != 2 {
		t.Fatalf("with the hold wiring off a held row's due timers were delivered %d time(s), want both", len(d.calls))
	}
	for _, name := range []string{"first", "second"} {
		tm, err := reg.GetTimer(holdHost, name)
		if err != nil {
			t.Fatal(err)
		}
		if tm.Attempts != 1 || tm.LastStatus != "503" || tm.NextAt != now.Add(backoffBase).Unix() {
			t.Fatalf("%s: bookkeeping did not advance as main's does: %+v", name, tm)
		}
	}
}

// ---- the import gate (seeded-instances spec §5.4) ----

const gateSeedID = "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"

// gateForImport swaps host's row for a gated import row in one journal
// transaction, the way the swap does.
func gateForImport(t *testing.T, reg *registry.Registry, host string) {
	t.Helper()
	stored, err := reg.Get(host)
	if err != nil || stored == nil {
		t.Fatalf("get %s: %v %v", host, stored, err)
	}
	next := *stored
	next.Provisioned = true
	next.ImportGate = strings.Repeat("a", 64)
	next.SeedSource = `{"id":"` + gateSeedID + `","kind":"import","state":"swapped"}`
	if err := reg.SwapForImport(registry.ImportJournal{Host: host, SeedID: gateSeedID, Original: *stored, At: 1,
		RecordKey: "apps/bab/app/instances/" + host + ".json"}, next); err != nil {
		t.Fatal(err)
	}
}

// A gated row gets no ticks, whether or not the hold wiring is armed, and
// its bookkeeping does not move; commit opens it.
func TestTickSkipsGatedRows(t *testing.T) {
	for _, armed := range []bool{false, true} {
		t.Run(fmt.Sprintf("holds armed %v", armed), func(t *testing.T) {
			reg, r := runnerFixture(t, nil)
			now := at("2026-08-23T13:07:00Z")
			r.Now = func() time.Time { return now }
			if armed {
				r.Gate = &HostGate{}
			}
			delivered := 0
			r.Deliver = deliverFunc(func(registry.Route, *http.Request) (int, error) { delivered++; return 204, nil })
			gateForImport(t, reg, holdHost)
			reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
			before := timersOf(t, reg, holdHost)
			r.tick()
			if delivered != 0 {
				t.Fatalf("delivered %d tick(s) to a gated row", delivered)
			}
			if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
				t.Fatalf("a gated row's bookkeeping moved:\n got %+v\nwant %+v", got, before)
			}
			if r.Busy(holdHost) {
				t.Fatal("the runner kept a claim on a gated host")
			}
			if err := reg.CommitImport(holdHost, gateSeedID); err != nil {
				t.Fatal(err)
			}
			r.tick()
			if delivered != 1 {
				t.Fatalf("after commit delivered %d, want 1", delivered)
			}
		})
	}
}

// M14: the swap lands between the tick's pre-filter read and its claim. The
// decision read inside the claim sees the gate and lets go: nothing is
// delivered and the bookkeeping does not move.
func TestGateLandingBeforeTheClaimGetsNoTick(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	now := at("2026-08-23T13:07:00Z")
	r.Now = func() time.Time { return now }
	r.Gate = &HostGate{}
	reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 1})
	before := timersOf(t, reg, holdHost)
	swapped := false
	r.beforeClaim = func(host string) {
		if host == holdHost && !swapped {
			swapped = true
			gateForImport(t, reg, holdHost)
		}
	}
	r.tick()
	if !swapped {
		t.Fatal("test bug: the tick never reached its claim")
	}
	if len(d.calls) != 0 {
		t.Fatalf("a row gated before the claim got %d tick(s)", len(d.calls))
	}
	if got := timersOf(t, reg, holdHost); !sameTimers(got, before) {
		t.Fatalf("the gated row's bookkeeping moved:\n got %+v\nwant %+v", got, before)
	}
	if r.Busy(holdHost) {
		t.Fatal("the tick kept its claim after seeing the gate")
	}
}

// Final review F5: 64 gated due timers (four uncommitted imports with 16
// each, due first) and one ordinary due timer on another host. The batch
// leaves the gated rows out before its limit, so the ordinary timer is
// delivered on the first tick, and the gated rows' bookkeeping does not
// move.
func TestGatedTimersDoNotStarveAnotherHost(t *testing.T) {
	for _, armed := range []bool{false, true} {
		t.Run(fmt.Sprintf("holds armed %v", armed), func(t *testing.T) {
			reg, r := runnerFixture(t, nil)
			now := at("2026-08-23T13:07:00Z")
			r.Now = func() time.Time { return now }
			if armed {
				r.Gate = &HostGate{}
			}
			var delivered []string
			r.Deliver = deliverFunc(func(rt registry.Route, _ *http.Request) (int, error) {
				delivered = append(delivered, rt.Host)
				return 204, nil
			})
			var gated []string
			for i := range 4 {
				host := fmt.Sprintf("g%d.bab.oncarlos.com", i)
				gated = append(gated, host)
				if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
					t.Fatal(err)
				}
				gateForImport(t, reg, host)
				for n := range dueBatch / 4 {
					reg.UpsertTimer(registry.Timer{Host: host, Name: fmt.Sprintf("t%d", n), Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: int64(1 + n)})
				}
			}
			reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "remind", Kind: KindOnce, Spec: "at 1", Path: "/r", NextAt: 100})
			before := map[string][]registry.Timer{}
			for _, h := range gated {
				before[h] = timersOf(t, reg, h)
			}
			r.tick()
			r.wait()
			if len(delivered) != 1 || delivered[0] != holdHost {
				t.Fatalf("delivered %q, want the ordinary host's timer alone", delivered)
			}
			for _, h := range gated {
				if got := timersOf(t, reg, h); !sameTimers(got, before[h]) {
					t.Fatalf("%s's gated rows moved:\n got %+v\nwant %+v", h, got, before[h])
				}
			}
		})
	}
}

// Skip (harness Run 9, defect 8): a host the agent says not to touch, here a
// held row with the move-hold wiring off (an offline move's source after
// open), gets no token, no delivery and no bookkeeping. And a hold written
// between the tick's route read and the fire is seen before the token, the
// one write a tick makes on the instance's own ground.
func TestSkipMintsNoTokenAndRecordsNothing(t *testing.T) {
	d := &fakeDeliverer{status: 204}
	reg, r := runnerFixture(t, d)
	now := at("2026-10-20T17:27:04Z")
	r.Now = func() time.Time { return now }
	tokens := 0
	r.Token = func(string) (string, error) { tokens++; return "tok", nil }
	r.Skip = func(rt registry.Route) bool { return registry.IsHold(rt.Status) }
	if err := reg.UpsertTimer(registry.Timer{Host: holdHost, Name: "every-minute", Kind: KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: now.Unix() - 5}); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	before := timersOf(t, reg, holdHost)
	r.tick()
	if tokens != 0 || len(d.calls) != 0 || !sameTimers(timersOf(t, reg, holdHost), before) || r.Busy(holdHost) {
		t.Fatalf("a skipped host: %d tokens, %d deliveries, timers %+v (were %+v)", tokens, len(d.calls), timersOf(t, reg, holdHost), before)
	}

	// The row reads ready at the tick's read; the hold lands before the claim.
	if ok, err := reg.ReleaseHold(holdHost, "mv_a"); err != nil || !ok {
		t.Fatalf("release hold = %v %v", ok, err)
	}
	r.beforeClaim = func(host string) {
		if err := reg.SetHold(host, "moving", "mv_b"); err != nil {
			t.Error(err)
		}
	}
	r.tick()
	if tokens != 0 || len(d.calls) != 0 || !sameTimers(timersOf(t, reg, holdHost), before) {
		t.Fatalf("a hold written before the fire: %d tokens, %d deliveries", tokens, len(d.calls))
	}

	// Ready again, nothing skipped: the tick fires as it always has.
	if ok, err := reg.ReleaseHold(holdHost, "mv_b"); err != nil || !ok {
		t.Fatalf("release hold = %v %v", ok, err)
	}
	r.beforeClaim = nil
	r.tick()
	if tokens != 1 || len(d.calls) != 1 {
		t.Fatalf("a ready host: %d tokens, %d deliveries", tokens, len(d.calls))
	}
}

// Astra's review of moves-offline-fixes, finding 1: with the hold wiring off
// (Gate nil, as on an offline move's boxes), a host Skip names must not fill
// the due batch. 64 earlier due rows of a skipped host, held or not, and one
// later row of an ordinary host: the ordinary one fires on the first tick.
func TestSkippedHostsDoNotStarveAnotherHostsTimer(t *testing.T) {
	for _, held := range []bool{true, false} {
		d := &fakeDeliverer{status: 204}
		reg, r := runnerFixture(t, d)
		now := at("2026-10-20T17:27:04Z")
		r.Now = func() time.Time { return now }
		const other = "b.bab.oncarlos.com"
		if err := reg.Upsert(registry.Route{Host: other, Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true}); err != nil {
			t.Fatal(err)
		}
		if held {
			if err := reg.SetHold(holdHost, "moving", "mv_a"); err != nil {
				t.Fatal(err)
			}
			r.Skip = func(rt registry.Route) bool { return registry.IsHold(rt.Status) }
		} else {
			// Ready, but named elsewhere (an offline move's record names
			// another region): only Skip knows.
			r.Skip = func(rt registry.Route) bool { return rt.Host == holdHost }
		}
		for i := range dueBatch {
			if err := reg.UpsertTimer(registry.Timer{Host: holdHost, Name: fmt.Sprintf("t%02d", i), Kind: KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: now.Unix() - 1000 + int64(i)}); err != nil {
				t.Fatal(err)
			}
		}
		if err := reg.UpsertTimer(registry.Timer{Host: other, Name: "later", Kind: KindSchedule, Spec: "every 1m0s", Path: "/tick", NextAt: now.Unix() - 5}); err != nil {
			t.Fatal(err)
		}
		if r.Gate != nil {
			t.Fatal("fixture: the hold wiring is armed")
		}
		r.tick()
		if len(d.calls) != 1 || d.hosts[0] != other {
			t.Fatalf("held %v: deliveries to %v, want exactly one to %s", held, d.hosts, other)
		}
	}
}