package schedule

import (
	"context"
	"path/filepath"
	"sync"
	"testing"
	"time"

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

func TestHoldsAwakeWhenTheIntervalIsInsideTheIdleWindow(t *testing.T) {
	now := time.Unix(1788120000, 0).UTC()
	cases := []struct {
		name string
		kind string
		spec string
		idle time.Duration
		want bool
	}{
		{"tighter than the window", KindSchedule, "every 5m", 15 * time.Minute, true},
		{"exactly the window", KindSchedule, "every 15m", 15 * time.Minute, true},
		{"wider than the window", KindSchedule, "every 1h", 15 * time.Minute, false},
		{"tight, short window", KindSchedule, "every 5m", 90 * time.Second, false},
		{"a once-timer never holds", KindOnce, "at 1788130000", 15 * time.Minute, false},
		{"hibernation not armed", KindSchedule, "every 1m", 0, false},
		{"hourly cron", KindSchedule, "cron 0 * * * *", 15 * time.Minute, false},
		{"every-minute cron", KindSchedule, "cron * * * * *", 15 * time.Minute, true},
		{"unparseable spec is not a claim", KindSchedule, "banana", 15 * time.Minute, false},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			got := holdsAwake(registry.Timer{Kind: tc.kind, Spec: tc.spec}, tc.idle, now)
			if got != tc.want {
				t.Errorf("holdsAwake(%q, idle=%s) = %v, want %v", tc.spec, tc.idle, got, tc.want)
			}
		})
	}
}

// Hibernation is armed PER ROUTE (rt.Hibernate), not just per box: a route
// with Hibernate off was never going to sleep, so no schedule on it can
// hold anything awake, however tight. StatusFor must zero the window before
// holdsAwake ever sees it — a box-wide idle window alone is not enough.
func TestStatusForNeverClaimsHoldsAwakeOffARouteWithHibernateOff(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	rt := registry.Route{Host: "resident.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: false}
	if err := reg.Upsert(rt); err != nil {
		t.Fatal(err)
	}
	now := at("2026-08-23T13:07:00Z")
	if err := reg.UpsertTimer(registry.Timer{Host: rt.Host, Name: "sweep", Kind: KindSchedule, Spec: "every 5m", Path: "/jobs/sweep", CreatedAt: now.Unix()}); err != nil {
		t.Fatal(err)
	}
	st, err := StatusFor(reg, rt, "box1", now, 15*time.Minute)
	if err != nil {
		t.Fatal(err)
	}
	if len(st.Timers) != 1 || st.Timers[0].HoldsAwake {
		t.Fatalf("HoldsAwake = %v on a Hibernate:false route, want false regardless of the box's idle window", st.Timers[0].HoldsAwake)
	}
}

func convergeFixture(t *testing.T) (*registry.Registry, *release.DirStore) {
	t.Helper()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	for _, rt := range []registry.Route{
		{Host: "a.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
		{Host: "b.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Hibernate: true},
		{Host: "u.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", Backing: registry.BackingUnit},
		{Host: "other.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "other", Status: "ready"},
		{Host: "site.bab.oncarlos.com", Kind: "static", Account: "bab", App: "site", Status: "ready"},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	return reg, release.NewDirStore(t.TempDir())
}

func TestConvergeMaterialisesDeclarationPerExecInstance(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	var reports []Status
	c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
		tm, err := reg.GetTimer(h, "sync")
		if err != nil {
			t.Fatalf("%s: %v", h, err)
		}
		if tm.Kind != KindSchedule || tm.Spec != "every 6h" || tm.Path != "/jobs/sync" || tm.NextAt != at("2026-08-23T18:00:00Z").Unix() {
			t.Fatalf("%s: %+v", h, tm)
		}
	}
	if rows, _ := reg.TimersFor("u.bab.oncarlos.com"); len(rows) != 0 {
		t.Fatal("unit-backed host got rows")
	}
	if rows, _ := reg.TimersFor("other.bab.oncarlos.com"); len(rows) != 0 {
		t.Fatal("undeclared app got rows")
	}
	// Reports: a and b with a timer each, u as unsupported. Nothing for site.
	byHost := map[string]Status{}
	for _, r := range reports {
		byHost[r.Host] = r
	}
	if len(byHost["a.bab.oncarlos.com"].Timers) != 1 || byHost["a.bab.oncarlos.com"].Box != "box1" {
		t.Fatalf("report a = %+v", byHost["a.bab.oncarlos.com"])
	}
	if byHost["u.bab.oncarlos.com"].Unsupported == "" {
		t.Fatalf("unit host not reported unsupported: %+v", byHost)
	}
	if _, ok := byHost["site.bab.oncarlos.com"]; ok {
		t.Fatal("static route reported")
	}

	// A second pass with nothing changed: no new reports, next_at untouched.
	reports = nil
	reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 10, 0, 999, 999)
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if len(reports) != 0 {
		t.Fatalf("unchanged pass reported %d hosts", len(reports))
	}
	if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 999 {
		t.Fatalf("unchanged pass moved next_at: %+v", tm)
	}
}

func TestConvergeRemovesUndeclaredAndKeepsOnce(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "gone", Every: "1h", Path: "/jobs/gone"}, now)
	reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "remind", Kind: KindOnce, Spec: "at 99999999999", Path: "/r", NextAt: 99999999999})
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	c.Pass(ctx)
	RemoveSchedule(ctx, st, "bab", "app", "gone", now)
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	rows, _ := reg.TimersFor("a.bab.oncarlos.com")
	names := map[string]bool{}
	for _, r := range rows {
		names[r.Name] = true
	}
	if !names["sync"] || !names["remind"] || names["gone"] {
		t.Fatalf("rows = %v", names)
	}
}

func TestConvergeRunSignalFiresOnceAcrossInstances(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	c.Pass(ctx)
	// A run signal that predates this converger is history, not a request.
	TouchRun(ctx, st, "bab", "app", "sync", now)
	c2 := NewConverger(reg, st, "box1", nil)
	c2.Now = func() time.Time { return now }
	c2.Pass(ctx)
	if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt == now.Unix() {
		t.Fatal("a pre-existing run signal was replayed at seed")
	}
	TouchRun(ctx, st, "bab", "app", "sync", now)
	c2.Pass(ctx)
	for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
		if tm, _ := reg.GetTimer(h, "sync"); tm.NextAt != now.Unix() {
			t.Fatalf("%s not fired: %+v", h, tm)
		}
	}
	reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
	c2.Pass(ctx) // same etag: must not re-fire
	if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
		t.Fatal("run signal replayed on an unchanged etag")
	}
}

func TestConvergePrunesRemovedHosts(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	reg.UpsertTimer(registry.Timer{Host: "removed.bab.oncarlos.com", Name: "x", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/p", NextAt: 1})
	c := NewConverger(reg, st, "box1", nil)
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("removed.bab.oncarlos.com"); len(rows) != 0 {
		t.Fatal("orphan rows survived")
	}
}

// TestConvergeNameCollisionWithOnceRowSkippedSilently pins the fix for a
// found bug: a declared schedule sharing a name with an existing once-row
// (the instance's own, registered off its control socket) must never be
// treated as "previous state" for the changed comparison — the once-row's
// spec text ("at ...") always differs from the declared schedule's, and
// comparing against it pinned `changed` (and hence a Report) on every
// single pass forever, never converging quiet even though nothing was
// ever actually written.
func TestConvergeNameCollisionWithOnceRowSkippedSilently(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	if err := reg.UpsertTimer(registry.Timer{Host: "a.bab.oncarlos.com", Name: "sync", Kind: KindOnce, Spec: "at 99999999999", Path: "/r", NextAt: 99999999999}); err != nil {
		t.Fatal(err)
	}
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	var reports []Status
	c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 || rows[0].Kind != KindOnce || rows[0].NextAt != 99999999999 {
		t.Fatalf("once row not left untouched / a schedule row was created: %+v", rows)
	}

	reports = nil
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 || rows[0].Kind != KindOnce {
		t.Fatalf("once row disturbed on the second pass: %+v", rows)
	}
	for _, r := range reports {
		if r.Host == "a.bab.oncarlos.com" {
			t.Fatalf("second pass reported the colliding host: %+v", r)
		}
	}
}

// TestConvergeDropsScheduleRowsOnUnitFlip pins the fix for a found bug: the
// unit-backed early return skipped TimersFor/DeleteTimersExcept entirely,
// so a host that flips from exec-backed to unit-backed (registry.SetBacking
// — the app-capabilities migration door) kept its old schedule rows
// forever, violating "a unit-backed host has no rows".
func TestConvergeDropsScheduleRowsOnUnitFlip(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
		t.Fatalf("setup: expected one schedule row before the flip, got %+v", rows)
	}

	if err := reg.SetBacking("a.bab.oncarlos.com", registry.BackingUnit); err != nil {
		t.Fatal(err)
	}
	var reports []Status
	c.Report = func(_ registry.Route, s Status) { reports = append(reports, s) }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
		t.Fatalf("schedule rows survived a flip to unit-backed: %+v", rows)
	}
	var got *Status
	for i := range reports {
		if reports[i].Host == "a.bab.oncarlos.com" {
			got = &reports[i]
		}
	}
	if got == nil || got.Unsupported == "" {
		t.Fatalf("no unsupported report for the flipped host: %+v", reports)
	}
}

// gateStore blocks every Get until open is closed, recording the keys it was
// asked for. It is how the two tests below observe whether a second Pass got
// past the lock while the first one was parked mid-pass in the store.
type gateStore struct {
	release.Store
	entered chan string
	open    chan struct{}
}

func (g *gateStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	g.entered <- key
	<-g.open
	return g.Store.Get(ctx, key)
}

// TestConvergePassHoldsTheLockForTheWholePass pins the serialization the agent
// depends on: fastConvergeLoop (seed + serial leg) and adoptLoop (hourly floor)
// call Pass on ONE shared Converger from two goroutines, and they genuinely
// overlap. Without the lock the etag maps take a concurrent read and write —
// a runtime throw that kills the agent and the activator with it — and two
// interleaved passes can fire the same manual run twice.
//
// The store parks the first pass inside its very first Get, still holding the
// lock; a second Pass must not reach the store at all until it is released.
func TestConvergePassHoldsTheLockForTheWholePass(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	gate := &gateStore{Store: st, entered: make(chan string, 64), open: make(chan struct{})}
	c := NewConverger(reg, gate, "box1", nil)
	c.Now = func() time.Time { return now }

	first := make(chan error, 1)
	go func() { first <- c.Pass(ctx) }()
	<-gate.entered // pass 1 is inside a Get, holding the lock

	second := make(chan error, 1)
	go func() { second <- c.Pass(ctx) }()
	select {
	case k := <-gate.entered:
		// Only pass 2 can be here: pass 1 is parked inside its Get and cannot
		// issue another until open closes.
		t.Fatalf("a second Pass reached the store (%s) while the first held the lock", k)
	case <-time.After(100 * time.Millisecond):
		// Blocked, as it must be. A slow machine can only weaken this check,
		// never fail it spuriously.
	}
	close(gate.open)
	if err := <-first; err != nil {
		t.Fatalf("first pass: %v", err)
	}
	if err := <-second; err != nil {
		t.Fatalf("second pass: %v", err)
	}
	if len(gate.entered) == 0 {
		t.Fatal("the gate store was never consulted — this test pinned nothing")
	}
}

// TestConvergePassIsSafeUnderConcurrentCallers is the -race half of the same
// finding, and the exactly-once half of it: four passes racing on one
// Converger must leave the manual run fired ONCE, with the etag recorded so a
// later pass does not replay it. The Report closure appends to a plain slice
// on purpose — Pass calls it under the lock, so a caller's ordinary,
// non-thread-safe callback is safe, and this would flag under -race if that
// ever stopped being true.
func TestConvergePassIsSafeUnderConcurrentCallers(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	var reports []Status
	c := NewConverger(reg, st, "box1", func(_ registry.Route, s Status) { reports = append(reports, s) })
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil { // seed: a run signal older than the converger is history
		t.Fatal(err)
	}
	TouchRun(ctx, st, "bab", "app", "sync", now)

	var wg sync.WaitGroup
	errs := make(chan error, 8)
	for i := 0; i < 4; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := c.Pass(ctx); err != nil {
				errs <- err
			}
		}()
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		t.Fatalf("concurrent Pass: %v", err)
	}
	for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
		if tm, _ := reg.GetTimer(h, "sync"); tm.NextAt != now.Unix() {
			t.Fatalf("%s: run signal not fired: %+v", h, tm)
		}
	}
	// Exactly-once, COUNTED: a run signal that fired twice is one that got
	// delivered twice, so the assertion runs the rows through a real Runner
	// and counts the deliveries rather than inferring from next_at. Four
	// racing passes, one tick each host — and a later pass, which sees the
	// recorded etag, adds no second delivery.
	d := &fakeDeliverer{status: 204}
	r := &Runner{Reg: reg, Deliver: d, Timeout: time.Minute, Box: "box1", Build: "test",
		Now: func() time.Time { return now }, Token: func(string) (string, error) { return "tok", nil }}
	r.tick()
	fired := d.firedHosts()
	for _, h := range []string{"a.bab.oncarlos.com", "b.bab.oncarlos.com"} {
		if fired[h] != 1 {
			t.Fatalf("%s delivered %d times, want exactly 1 (fired = %v)", h, fired[h], fired)
		}
	}
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	r.tick()
	if got := d.firedHosts(); len(got) != len(fired) || got["a.bab.oncarlos.com"] != 1 || got["b.bab.oncarlos.com"] != 1 {
		t.Fatalf("the run signal replayed after the concurrent passes: deliveries now %v, were %v", got, fired)
	}
	if len(reports) == 0 {
		t.Fatal("no reports at all — the fixture converged nothing")
	}
}

// The declaration object itself can go away (not just move to an empty
// schedules list — that is TestConvergeHandlesPresentButEmptyDeclaration
// below) and later reappear: e.g. hand-deleted from the bucket, or an app
// re-created after deletion (spec §4.1: absence is legal forever). Its
// absence must clear a host's rows exactly like an empty declaration does,
// and the cached "" etag from the absent read must not be mistaken for a
// match against whatever real etag the object comes back with.
func TestConvergeDeclarationGoingAbsentThenPresentAgain(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
		t.Fatalf("setup: want one row, got %+v", rows)
	}

	if err := st.Delete(ctx, DeclarationKey("bab", "app")); err != nil {
		t.Fatal(err)
	}
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
		t.Fatalf("row survived the declaration going absent: %+v", rows)
	}

	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
		t.Fatalf("row not re-materialised once the declaration came back: %+v", rows)
	}
}

// RemoveSchedule rewrites schedules.json with an empty Schedules slice; it
// never deletes the key (spec §4.1's "absence is legal forever" is about a
// key that was never written, not one a member emptied out). A
// present-but-empty declaration must prune existing rows exactly like an
// absent one, without erroring on the zero-length Schedules slice.
func TestConvergeHandlesPresentButEmptyDeclaration(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 1 {
		t.Fatalf("setup: want one row, got %+v", rows)
	}

	if _, present, err := RemoveSchedule(ctx, st, "bab", "app", "sync", now); err != nil || !present {
		t.Fatalf("remove: present=%v err=%v", present, err)
	}
	if _, etag, err := ReadDeclaration(ctx, st, "bab", "app"); err != nil || etag == "" {
		t.Fatalf("declaration should still be present (non-empty etag) after removing its only schedule: %q %v", etag, err)
	}

	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if rows, _ := reg.TimersFor("a.bab.oncarlos.com"); len(rows) != 0 {
		t.Fatalf("row survived a present-but-empty declaration: %+v", rows)
	}
}

// The run signal object can itself go away between passes (no deletion API
// is designed for it, but nothing stops a hand-edited bucket). Its absence
// must read as "no movement" — not as a fire, and not as a poisoned cache
// that can no longer recognise a later, genuinely new touch.
func TestConvergeRunSignalGoingAbsentIsNotAFire(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil { // seed
		t.Fatal(err)
	}
	TouchRun(ctx, st, "bab", "app", "sync", now)
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	reg.MarkTimerFired("a.bab.oncarlos.com", "sync", scheduleTimerRegistration(t, reg, "a.bab.oncarlos.com", "sync"), now.Unix(), "204", 1, 0, 777, 777)
	if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
		t.Fatalf("setup: run signal did not fire: %+v", tm)
	}

	if err := st.Delete(ctx, RunKey("bab", "app")); err != nil {
		t.Fatal(err)
	}
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != 777 {
		t.Fatalf("run signal going absent misfired: %+v", tm)
	}

	// A later, genuinely new touch must still be recognised.
	TouchRun(ctx, st, "bab", "app", "sync", now)
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if tm, _ := reg.GetTimer("a.bab.oncarlos.com", "sync"); tm.NextAt != now.Unix() {
		t.Fatalf("run signal did not fire once the object came back: %+v", tm)
	}
}

// A declaration can exist for an app the registry has never heard of
// (nothing shipped there yet, or every instance since removed) — byApp
// never gains an entry for it, convergeApp is never called for it, and
// Pass must return cleanly rather than erroring. The weak version of this
// test is a wholly empty registry, where that's true trivially because
// NOTHING converges; the real claim is that a routeless app is skipped
// while a SIBLING app with live routes still converges normally in the
// same pass, so this seeds both: "ghost" (declared, zero routes) beside
// "other" (declared here, and the fixture already gives it a real route).
func TestConvergeZeroRouteAppDoesNotBlockASiblingApp(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	UpsertSchedule(ctx, st, "bab", "ghost", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	UpsertSchedule(ctx, st, "bab", "other", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatalf("Pass with one routeless app declared alongside a routed sibling: %v", err)
	}
	rows, err := reg.TimersFor("other.bab.oncarlos.com")
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 || rows[0].Name != "sync" || rows[0].Kind != KindSchedule {
		t.Fatalf("the routed sibling app (other) did not converge: %+v", rows)
	}
	// "ghost" has no host anywhere to hold a row, so there is nothing to
	// query for it directly — Pass returning cleanly with "other" actually
	// converged above IS the assertion that byApp skipped "ghost" without
	// that skip taking its sibling down with it.
}

// A declaration whose cron never matches (written past Validate — an older
// console, or by hand) must not become a row due at the zero time's unix
// value: it is parked at next_at=0 with last_status "never fires", and the
// runner therefore never sees it.
func TestConvergeParksAScheduleThatNeverFires(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	raw := []byte(`{"schema":1,"schedules":[{"name":"never","cron":"0 0 30 2 *","path":"/x"}]}`)
	if _, err := st.Put(ctx, DeclarationKey("bab", "app"), raw, ""); err != nil {
		t.Fatal(err)
	}
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	tm, err := reg.GetTimer("a.bab.oncarlos.com", "never")
	if err != nil {
		t.Fatal(err)
	}
	if tm.NextAt != 0 || tm.DueAt != 0 || tm.LastStatus != StatusNeverFires {
		t.Fatalf("parked row = %+v, want next_at 0 and %q", tm, StatusNeverFires)
	}
	if due, _ := reg.DueTimers(now.Unix(), 10, nil); len(due) != 0 {
		t.Fatalf("a parked row is due: %+v", due)
	}
}

func scheduleTimerRegistration(t *testing.T, reg *registry.Registry, host, name string) string {
	t.Helper()
	row, err := reg.GetTimer(host, name)
	if err != nil {
		t.Fatal(err)
	}
	return row.Registration
}

// The converger and the cut's barrier: a convergeHost paused after its
// Frozen check, inside the gate, holds the drain until it has finished
// writing; a converger that starts after the drain sees Frozen and writes
// nothing, so the timers snapshot taken after the drain equals the table.
func TestConvergerCannotWriteAfterTheBarrier(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	const host = "a.bab.oncarlos.com"
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	gate := &HostGate{}
	var frozen sync.Map
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	c.Gate = gate
	c.Frozen = func(h string) bool { _, ok := frozen.Load(h); return ok }
	reached, resume := make(chan struct{}), make(chan struct{})
	var once sync.Once
	c.afterFrozenCheck = func(h string) {
		if h == host {
			once.Do(func() { close(reached); <-resume })
		}
	}
	passed := make(chan error, 1)
	go func() { passed <- c.Pass(ctx) }()
	<-reached

	// The cut: hold, freeze, drain. The paused convergence holds the drain.
	if err := reg.SetHold(host, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	frozen.Store(host, true)
	drained := make(chan error, 1)
	go func() { drained <- gate.Drain(ctx, host) }()
	select {
	case err := <-drained:
		t.Fatalf("the drain returned (%v) while a convergence was inside the gate", err)
	case <-time.After(50 * time.Millisecond):
	}
	close(resume)
	if err := <-drained; err != nil {
		t.Fatal(err)
	}
	snap, err := reg.TimersFor(host)
	if err != nil {
		t.Fatal(err)
	}
	if err := <-passed; err != nil {
		t.Fatal(err)
	}
	if len(snap) != 1 || snap[0].Name != "sync" {
		t.Fatalf("the paused convergence's write is not in the snapshot: %+v", snap)
	}

	// A later pass, with the declaration changed: frozen, it writes nothing.
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "nightly", Every: "24h", Path: "/jobs/nightly"}, now)
	if _, _, err := RemoveSchedule(ctx, st, "bab", "app", "sync", now); err != nil {
		t.Fatal(err)
	}
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	got, err := reg.TimersFor(host)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != len(snap) || got[0] != snap[0] {
		t.Fatalf("the table moved after the barrier:\n got %+v\nwant %+v", got, snap)
	}
	// The other host of the app, not frozen, converged as usual.
	if _, err := reg.GetTimer("b.bab.oncarlos.com", "nightly"); err != nil {
		t.Fatalf("an unfrozen host stopped converging: %v", err)
	}
}

// While any routed host is frozen, Pass leaves the table-wide prune alone,
// so a frozen host's finished once-rows stay as the snapshot saw them; the
// runner's own daily sweep is the one prune that is not frozen (joint N14).
func TestConvergerSkipsTheTableWidePruneWhileAHostIsFrozen(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	const host = "a.bab.oncarlos.com"
	old := now.Add(-8 * 24 * time.Hour).Unix()
	reg.UpsertTimer(registry.Timer{Host: host, Name: "done", Kind: KindOnce, Spec: "at 1", Path: "/d", NextAt: 0, LastAt: old})
	if _, err := reg.MarkTimerFired(host, "done", scheduleTimerRegistration(t, reg, host, "done"), old, "204", 1, 0, 0, 0); err != nil {
		t.Fatal(err)
	}
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	c.Frozen = func(h string) bool { return h == host }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if tm, _ := reg.GetTimer(host, "done"); tm == nil {
		t.Fatal("the frozen host's finished once-row was pruned")
	}
	c.Frozen = func(string) bool { return false }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if tm, _ := reg.GetTimer(host, "done"); tm != nil {
		t.Fatalf("the prune did not resume after the freeze: %+v", tm)
	}
}

// Fix round 1 (task-6-review, Important 3): with neither Frozen nor Gate
// set (the Phase 1 switch off) the converger is main's. A host whose row
// status is a hold has its declaration converged and its undeclared
// schedule pruned, and the table-wide prune runs.
func TestConvergerWithoutTheHooksIsMain(t *testing.T) {
	reg, st := convergeFixture(t)
	ctx := context.Background()
	now := at("2026-08-23T13:07:00Z")
	const host = "a.bab.oncarlos.com"
	if err := reg.SetHold(host, "moving", "mv_a"); err != nil {
		t.Fatal(err)
	}
	reg.UpsertTimer(registry.Timer{Host: host, Name: "stale", Kind: KindSchedule, Spec: "every 1h0m0s", Path: "/s", NextAt: now.Unix() - 5})
	old := now.Add(-8 * 24 * time.Hour).Unix()
	reg.UpsertTimer(registry.Timer{Host: host, Name: "done", Kind: KindOnce, Spec: "at 1", Path: "/d", NextAt: 0, LastAt: old})
	if _, err := reg.MarkTimerFired(host, "done", scheduleTimerRegistration(t, reg, host, "done"), old, "204", 1, 0, 0, 0); err != nil {
		t.Fatal(err)
	}
	UpsertSchedule(ctx, st, "bab", "app", Schedule{Name: "sync", Every: "6h", Path: "/jobs/sync"}, now)
	c := NewConverger(reg, st, "box1", nil)
	c.Now = func() time.Time { return now }
	if err := c.Pass(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := reg.GetTimer(host, "sync"); err != nil {
		t.Fatalf("with the hold wiring off the held host's declaration did not converge: %v", err)
	}
	if tm, _ := reg.GetTimer(host, "stale"); tm != nil {
		t.Fatalf("with the hold wiring off the undeclared schedule survived: %+v", tm)
	}
	if tm, _ := reg.GetTimer(host, "done"); tm != nil {
		t.Fatalf("with the hold wiring off the finished once-row survived the table-wide prune: %+v", tm)
	}
}