package rollout

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/status"
)

var markV2 = HoldRan{Account: "bab", App: "blog", Build: "v2"}

// Final review wave 2, N2: decideWake's rule for an instance with a
// hold-ran marker. The good build never boots in place of the installed
// build, and (Astra R3 N7, ruling 205) the marker lifts no failure record
// and the installed good build waits for a signed rollback.
func TestDecideWakeWithAHoldRanMarker(t *testing.T) {
	overlap := DefaultPolicy()
	book := func(recs ...FailureRecord) *FailureBook {
		b := &FailureBook{}
		for _, r := range recs {
			b.Record(r)
		}
		b.MarkHoldRan("a", markV2)
		return b
	}
	sibV2 := FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "x", At: t0}
	sibV3 := FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "x", At: t0}

	// N7: the marked build, blocked for the instance by a sibling's record,
	// with a good build: refused, never v1 and never v2.
	p, err := decideWake(inputs(wakeRoute("v2", "v1"), book(sibV2), overlap))
	if err != nil || p.Refuse != HeldDownRefusal("v2") || p.Boot.StagedPath != "" {
		t.Fatalf("installed is the marked build, blocked: want the wake refused in the hold-down words, got %+v, %v", p, err)
	}
	// With no good build a record alone refuses nothing (spec §8.5): the
	// marked build boots, hold, no fallback.
	noGood := wakeRoute("v2", "")
	noGood.GoodDigest = ""
	if p, err := decideWake(inputs(noGood, book(sibV2), overlap)); err != nil || p.Refuse != "" || p.Boot.Version != "v2" ||
		!p.FirstWake || p.Fallback != nil || p.Mode != activator.ModeHold {
		t.Fatalf("installed is the marked build, blocked, no good build: want a hold first wake of v2, got %+v, %v", p, err)
	}
	// Unblocked, and its release allows overlap: still hold, no fallback.
	if p, _ := decideWake(inputs(wakeRoute("v2", "v1"), book(), overlap)); p.Boot.Version != "v2" || p.Fallback != nil || p.Mode != activator.ModeHold {
		t.Fatalf("installed is the marked build, unblocked: want hold with no fallback, got %+v", p)
	}
	// Held down on the instance: the hold-down refusal still comes first.
	down := book(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, At: t0})
	if p, _ := decideWake(inputs(wakeRoute("v2", "v1"), down, overlap)); p.Refuse != HeldDownRefusal("v2") {
		t.Fatalf("marked build held down: want the hold-down refusal, got %+v", p)
	}
	// Another installed build held for the instance: refused, never v1.
	if p, _ := decideWake(inputs(wakeRoute("v3", "v1"), book(sibV3), overlap)); p.Refuse != HeldDownRefusal("v2") || p.Boot.StagedPath != "" {
		t.Fatalf("another held build: want the wake refused in the hold-down words, got %+v", p)
	}
	// Another, unblocked build: a first wake with no fallback, in its own mode.
	if p, _ := decideWake(inputs(wakeRoute("v3", "v1"), book(), overlap)); p.Refuse != "" || p.Boot.Version != "v3" || !p.FirstWake ||
		p.Fallback != nil || p.Mode != activator.ModeOverlap {
		t.Fatalf("another unblocked build: want an overlap first wake of v3 with no fallback, got %+v", p)
	}
	// Ruling 205: the good build installed is no proof of a rollback (root's
	// sibling repoint installs it too). The planner refuses while the marker
	// stands; the pointer pass lifts the marker on a signed rollback.
	if p, _ := decideWake(inputs(wakeRoute("v1", "v1"), book(), overlap)); p.Refuse != HeldDownRefusal("v2") || p.Boot.StagedPath != "" {
		t.Fatalf("the good build installed under a marker: want the wake refused, got %+v", p)
	}
	// Task 3 fix round 1: the marked build is also the good build, and it is
	// installed. The marked-build rule applies, not the good-build arm: it
	// boots, in hold mode with no fallback. A sibling's record of it does not
	// hold an instance that served it (spec §8.4, Blocks with served); only a
	// hold-down refuses it.
	for name, b := range map[string]*FailureBook{"no record": book(), "a sibling's record": book(sibV2)} {
		if p, err := decideWake(inputs(wakeRoute("v2", "v2"), b, overlap)); err != nil || p.Refuse != "" || p.Boot.Version != "v2" ||
			!p.FirstWake || p.Fallback != nil || p.Mode != activator.ModeHold {
			t.Fatalf("marked build = installed = good, %s: want a hold first wake of v2 with no fallback, got %+v, %v", name, p, err)
		}
	}
	downGood := book(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, At: t0})
	if p, _ := decideWake(inputs(wakeRoute("v2", "v2"), downGood, overlap)); p.Refuse != HeldDownRefusal("v2") || p.Boot.StagedPath != "" {
		t.Fatalf("marked build = installed = good, held down: want the hold-down refusal, got %+v", p)
	}
	// A staging refusal never boots the good build instead.
	in := inputs(wakeRoute("v2", "v1"), book(), overlap)
	in.Demand = func(string, string) (string, string, error) { return "", "", errors.New("digest mismatch") }
	if p, err := decideWake(in); err == nil || p.Refuse == "" || p.Boot.StagedPath != "" {
		t.Fatalf("a staging refusal under the marker: want the wake refused, got %+v, %v", p, err)
	}
	// Another app's marker under the same host name changes nothing.
	other := &FailureBook{}
	other.Record(sibV2)
	other.MarkHoldRan("a", HoldRan{Account: "bab", App: "other", Build: "v2"})
	if p, _ := decideWake(inputs(wakeRoute("v2", "v1"), other, overlap)); p.Boot.Version != "v1" || p.FirstWake {
		t.Fatalf("another app's marker: want the usual recovery onto v1, got %+v", p)
	}
}

// Final review wave 2, N2: a planned wake onto the good build made before
// the marker existed (a recovery for the blocked v2) is refused at its exec
// once the marker stands. The marker comes from a hold first wake whose
// observation was cancelled after it ran. The refusal uses existing words.
func TestAHoldRanMarkerRefusesTheGoodBuildAtTheExec(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "h")
	h.stage("v2")
	h.install("v2", "h")
	h.sleepHost("h")
	h.pass()
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "sib", At: t0})
	h.wakeHost("h")
	recovery := h.m.Plan(ctx, h.rows.get("h"))
	if recovery.Boot.Version != "v1" || recovery.FirstWake || recovery.Check == nil {
		t.Fatalf("plan %+v: want the recovery onto the good v1 with its check (v2 is blocked, no marker yet)", recovery)
	}
	if err := recovery.Check("v1"); err != nil {
		t.Fatalf("before the marker the good build passes the check: %v", err)
	}
	h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("h"), Gen: activator.Gen{Build: "v2"},
		Mode: activator.ModeHold, FirstWake: true, Executed: true})
	err := recovery.Check("v1")
	if err == nil {
		t.Fatal("the recovery onto v1 passed its exec check after a hold build ran on h")
	}
	if want := HeldDownRefusal("v2"); err.Error() != want {
		t.Fatalf("refusal %q, want the existing words %q", err, want)
	}
}

// Final review wave 2, N2: a planned hold first wake whose observation was
// cancelled after its build ran (the activator's GenCancelled, Executed)
// leaves the marker, so a sibling's failure of the build afterwards does not
// send the instance's next wake to the good build. The marker is a
// restriction on the instance, not a failure: nothing is recorded or halted.
// The instance passing on the build (a settled first wake records it good)
// clears the marker, and with it the restriction.
func TestAFirstWakeCancelledAfterItsHoldBuildRanAndItsClearingByAPass(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "s")
	h.swap = appconfig.SwapHold
	h.stage("v2", "v3")
	h.install("v2", "s")
	h.sleepHost("s")
	h.pass()
	h.wakeHost("s")
	p := h.m.Plan(ctx, h.rows.get("s"))
	if !p.FirstWake || p.Boot.Version != "v2" || p.Mode != activator.ModeHold {
		t.Fatalf("plan %+v: want a hold first wake of v2", p)
	}
	g := activator.Gen{N: 7, Build: "v2", StagedPath: p.Boot.StagedPath, Digest: p.Boot.Digest}
	h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("s"), Gen: g, Mode: p.Mode, FirstWake: true, Token: p.Token})
	h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("s"), Gen: g, Mode: p.Mode, FirstWake: true,
		Token: p.Token, Executed: true})
	if got, ok := h.book.HoldRanOn("s"); !ok || !got.same(markV2) || !got.At.Equal(t0) {
		t.Fatalf("marker = %+v, %v: want v2's, stamped at its wake's plan", got, ok)
	}
	if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
		t.Fatal("a cancelled observation recorded a failure")
	}

	// A sibling then fails v2: the record holds v2 for s too (Astra R3 N7),
	// and the marker holds v1, so s stays down rather than boot either.
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "sib", PromotedAt: t0, At: t0})
	h.sleepHost("s")
	h.wakeHost("s")
	if p := h.m.Plan(ctx, h.rows.get("s")); p.Refuse != HeldDownRefusal("v2") {
		t.Fatalf("plan %+v: want the wake refused while the sibling's record of v2 stands", p)
	}
	// `carlos promote --retry` of v2 clears the record; v2 boots again, in
	// hold mode with nothing behind it, never the good v1.
	h.advance(time.Minute)
	h.pointer("stable", "v2", h.clock())
	h.pass()
	if _, ok := h.book.Lookup("bab", "blog", "v2"); ok {
		t.Fatal("the retry pointer left v2's record standing")
	}
	h.wakeHost("s")
	p = h.m.Plan(ctx, h.rows.get("s"))
	if p.Refuse != "" || p.Boot.Version != "v2" || !p.FirstWake || p.Fallback != nil || p.Mode != activator.ModeHold {
		t.Fatalf("plan %+v: want v2 again, never the good v1", p)
	}

	// It runs and settles: v2 is s's good build now, and the marker goes.
	h.act.serve("s", "v2")
	cur, _ := h.act.Current("s")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("s"), Gen: cur, Mode: p.Mode, FirstWake: true, Token: p.Token})
	h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: h.rows.get("s"), Gen: cur, Mode: p.Mode, FirstWake: true, Token: p.Token})
	if r := h.rows.get("s"); r.GoodVersion != "v2" {
		t.Fatalf("good build %q after the settle, want v2", r.GoodVersion)
	}
	if got, ok := h.book.HoldRanOn("s"); ok {
		t.Fatalf("marker %+v survived the instance passing on a build", got)
	}
	// The usual rule again: a held v3 sends the wake to the good build.
	h.install("v3", "s")
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "sib", At: t0})
	h.sleepHost("s")
	h.wakeHost("s")
	if p := h.m.Plan(ctx, h.rows.get("s")); p.Boot.Version != "v2" || p.FirstWake {
		t.Fatalf("plan %+v after the marker cleared: want the recovery onto the good v2", p)
	}
}

// Final review wave 2, N2: `carlos instances retry` clears the host's
// marker, with or without a failure record to clear beside it.
func TestRetryHostClearsTheHoldRanMarker(t *testing.T) {
	t.Run("with a record", func(t *testing.T) {
		h := newHarness(t, "a")
		h.stage("v2")
		h.install("v2", "a")
		h.sleepHost("a")
		h.pass()
		h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "sib", At: t0})
		h.book.MarkHoldRan("a", markV2)
		if !h.retry("a", "v2") {
			t.Fatal("RetryHost reported nothing cleared")
		}
		if _, ok := h.book.HoldRanOn("a"); ok {
			t.Fatal("the marker survived the host's retry")
		}
	})
	t.Run("marker only", func(t *testing.T) {
		h := newHarness(t, "a", "b")
		h.stage("v2")
		h.install("v2", "a")
		h.sleepHost("a")
		h.pass()
		h.book.MarkHoldRan("a", markV2)
		h.book.MarkHoldRan("b", markV2)
		if !h.retry("a", "v2") {
			t.Fatal("RetryHost reported nothing cleared with a marker standing")
		}
		if _, ok := h.book.HoldRanOn("a"); ok {
			t.Fatal("the marker survived the host's retry")
		}
		if _, ok := h.book.HoldRanOn("b"); !ok {
			t.Fatal("a's retry cleared b's marker")
		}
	})
}

// Final review wave 2, N2, the scheduler's half: which attempt outcomes set
// the marker. A hold adoption that ran and did not pass, and was not held
// down, sets it; a failed one sets it in the same save as its record. One
// that ran nothing, an overlap attempt, a restart, a hold-down and a pass do
// not.
func TestWhichAttemptOutcomesLeaveTheHoldRanMarker(t *testing.T) {
	// Started is the dispatch's (scheduler.execute): the marker's At, on disk
	// too, whichever save sets it (Task 3 fix round 1).
	started := t0.Add(-time.Minute)
	adopt := dispatch{ID: Identity{Host: "a", Cause: activator.CauseRelease, Build: "v2"}, SchedRev: 1, Class: classAdopt, Cohort: cohortKey("v2", classAdopt), Started: started}
	restart := dispatch{ID: Identity{Host: "a", Cause: activator.CauseRestart, Build: "v1", RestartID: "r1"}, SchedRev: 1, Class: classRecover, Cohort: cohortKey("v1", classRecover)}
	for _, tc := range []struct {
		name string
		d    dispatch
		res  activator.SwapResult
		mode activator.SwapMode
		want bool
	}{
		{"hold, superseded after it ran", adopt, activator.SwapResult{Outcome: activator.SwapAbandoned, Executed: true}, activator.ModeHold, true},
		{"hold, failed after it ran, not held down", adopt, activator.SwapResult{Outcome: activator.SwapFailed, Reason: "health check failed", Executed: true}, activator.ModeHold, true},
		{"hold, abandoned before it ran", adopt, activator.SwapResult{Outcome: activator.SwapAbandoned}, activator.ModeHold, false},
		{"hold, down: held down instead", adopt, activator.SwapResult{Outcome: activator.SwapDown, Reason: "health check failed", Executed: true}, activator.ModeHold, false},
		{"overlap, superseded after it ran", adopt, activator.SwapResult{Outcome: activator.SwapAbandoned, Executed: true}, activator.ModeOverlap, false},
		{"hold restart, superseded after it ran", restart, activator.SwapResult{Outcome: activator.SwapAbandoned, Executed: true}, activator.ModeHold, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			h := newHarness(t, "a")
			dir := t.TempDir()
			b, err := OpenFailureBook(dir)
			if err != nil {
				t.Fatal(err)
			}
			h.book, h.m.d.Book = b, b
			s := h.m.scheduler(appKey{"bab", "blog"})
			s.mu.Lock()
			slot := s.q.slot("a", t0)
			slot.serving, slot.good = "v1", "v1"
			d := tc.d
			slot.running = &d
			s.q.running++
			s.mu.Unlock()
			s.done(tc.d, tc.res, tc.mode)

			again, err := OpenFailureBook(dir)
			if err != nil {
				t.Fatal(err)
			}
			got, ok := again.HoldRanOn("a")
			if ok != tc.want || (ok && (!got.same(markV2) || !got.At.Equal(started))) {
				t.Fatalf("marker on disk = %+v, %v; want set %v, stamped at the dispatch %v", got, ok, tc.want, started)
			}
			if tc.res.Outcome == activator.SwapFailed {
				if _, ok := again.Lookup("bab", "blog", "v2"); !ok {
					t.Fatal("the failure record is not on disk beside the marker")
				}
			}
		})
	}
}

// Final review wave 2, N2: status shows an instance with a marker as held,
// in the existing words (HeldBy, AttemptDown), never new ones.
//
// Astra R3 N7 and ruling 205 add two down cases: the marked build itself
// held by a sibling's record (d), and the good build installed with no
// signed rollback yet (c). Their wakes are refused, so they are down.
func TestStatusShowsAHoldRanInstanceAsHeld(t *testing.T) {
	h := newHarness(t, "a", "b", "c")
	h.stage("v2", "v3")
	h.install("v2", "a")
	h.install("v3", "b")
	h.sleepHost("a")
	h.sleepHost("b")
	h.sleepHost("c")
	h.pass()
	h.book.MarkHoldRan("a", markV2)
	h.book.MarkHoldRan("b", markV2)
	h.book.MarkHoldRan("c", markV2)
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "sib", Reason: "exited early", At: t0})
	r := h.m.Status()
	if got := r.Hosts["a"]; got.HeldBy != "v2" || got.Attempt != "" {
		t.Fatalf("a = %+v: want held by v2 (its wake boots v2, nothing else)", got)
	}
	if got := r.Hosts["b"]; got.HeldBy != "v3" || got.Attempt != status.AttemptDown || got.Reason != status.ReasonExit {
		t.Fatalf("b = %+v: want down, held by v3 (its wakes are refused, not fallen back)", got)
	}
	if got := r.Hosts["c"]; got.HeldBy != "v2" || got.Attempt != status.AttemptDown {
		t.Fatalf("c = %+v: want down, held by v2 (its good build v1 is installed, but no signed rollback lifted the marker)", got)
	}

	d := newHarness(t, "d")
	d.stage("v2")
	d.install("v2", "d")
	d.sleepHost("d")
	d.pass()
	d.book.MarkHoldRan("d", markV2)
	d.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "sib", Reason: "exited early", At: t0})
	if got := d.m.Status().Hosts["d"]; got.HeldBy != "v2" || got.Attempt != status.AttemptDown || got.Reason != status.ReasonExit {
		t.Fatalf("d = %+v: want down, held by v2 (the marked build is held by a sibling's record)", got)
	}

	// Task 3 fix round 1: the marked build is also the installed good build.
	// Its wake boots it (the marked-build rule), so it is held, not down.
	e := newHarness(t, "e")
	e.stage("v2")
	r2 := e.rows.get("e")
	r2.Version, r2.GoodVersion, r2.GoodDigest = "v2", "v2", "sha-v2"
	e.rows.put(r2)
	e.sleepHost("e")
	e.pass()
	e.book.MarkHoldRan("e", markV2)
	e.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "sib", Reason: "exited early", At: t0})
	if got := e.m.Status().Hosts["e"]; got.HeldBy != "v2" || got.Attempt != "" {
		t.Fatalf("e = %+v: want held by v2 and not down (its marked build is its installed good build, which its wake boots)", got)
	}
}

// Task 3 fix round 1: when the marker names the installed good build, the
// marked-build rule applies at the plan and at the exec, even with a
// sibling's record of the build standing (the instance served it). The good
// build is still refused at the exec of any other plan while the marker
// stands.
func TestAMarkerOnTheInstalledGoodBuildBootsIt(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "h")
	h.stage("v2", "v3")
	r := h.rows.get("h")
	r.Version, r.GoodVersion, r.GoodDigest = "v2", "v2", "sha-v2"
	h.rows.put(r)
	h.book.MarkHoldRan("h", markV2)
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "sib", At: t0})
	h.sleepHost("h")
	h.wakeHost("h")
	p := h.m.Plan(ctx, h.rows.get("h"))
	if p.Refuse != "" || p.Boot.Version != "v2" || p.Mode != activator.ModeHold || p.Fallback != nil || p.Check == nil {
		t.Fatalf("plan %+v: want a hold wake of v2 with no fallback", p)
	}
	if err := p.Check("v2"); err != nil {
		t.Fatalf("the plan's check refuses v2, the marked and installed good build: %v", err)
	}
	// Another build installed: the good v2 is not the installed build, and
	// the good-build arm refuses it at that plan's exec.
	h.install("v3", "h")
	other := h.rows.get("h")
	check := h.m.planCheck(other, activator.WakePlan{Boot: p.Boot, Mode: activator.ModeHold, FirstWake: true})
	if err := check("v2"); err == nil || err.Error() != HeldDownRefusal("v2") {
		t.Fatalf("check of the good v2 with v3 installed = %v, want the hold-down refusal", err)
	}
}

// Final review wave 2, N2: the markers survive a reopen; a record written
// with MarkHoldRan carries its marker in the same file and never stores the
// instruction; a clear is saved too.
func TestTheHoldRanMarkerSurvivesReopen(t *testing.T) {
	dir := t.TempDir()
	b, err := OpenFailureBook(dir)
	if err != nil {
		t.Fatal(err)
	}
	if err := b.MarkHoldRan("a.example.com", markV2); err != nil {
		t.Fatal(err)
	}
	r := rec("v3", "c.example.com")
	r.MarkHoldRan = true
	if _, err := b.Record(r); err != nil {
		t.Fatal(err)
	}
	raw, err := os.ReadFile(filepath.Join(dir, FailureFile))
	if err != nil {
		t.Fatal(err)
	}
	if strings.Contains(strings.ToLower(string(raw)), "markholdran") || strings.Contains(string(raw), "mark_hold_ran") {
		t.Fatalf("the instruction was stored:\n%s", raw)
	}
	again, err := OpenFailureBook(dir)
	if err != nil {
		t.Fatal(err)
	}
	if got, ok := again.HoldRanOn("a.example.com"); !ok || !got.same(markV2) {
		t.Fatalf("a's marker after reopen = %+v, %v", got, ok)
	}
	if got, ok := again.HoldRanOn("c.example.com"); !ok || got.Build != "v3" {
		t.Fatalf("c's marker (set with its record) after reopen = %+v, %v", got, ok)
	}
	if rec, _ := again.Lookup("bab", "blog", "v3"); rec.MarkHoldRan {
		t.Fatal("a record came back carrying the instruction")
	}
	if cleared, err := again.ClearHoldRan("a.example.com"); !cleared || err != nil {
		t.Fatalf("ClearHoldRan = %v, %v", cleared, err)
	}
	third, err := OpenFailureBook(dir)
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := third.HoldRanOn("a.example.com"); ok {
		t.Fatal("the clear was not saved")
	}
	if _, ok := third.HoldRanOn("c.example.com"); !ok {
		t.Fatal("clearing a's marker dropped c's")
	}
}

// Final review wave 2, N2: the markers are additive to schema 1. A file an
// older agent wrote (no hold_ran key) opens with no markers. A file with
// markers still parses into the old shape (plain encoding/json skips the
// unknown key) with every record intact. Since Task 5 it is written as
// schema 2, and since the crash-epoch hard limit as schema 3, which an older
// agent's reader refuses on purpose rather than read it and drop the
// markers and intents (TestTheBookIsUnreadableToAnOlderReader).
func TestTheHoldRanMarkerIsSchemaAdditive(t *testing.T) {
	dir := t.TempDir()
	old := `{"schema": 1, "records": [{"account": "bab", "app": "blog", "build": "v2", "host": "a", "cause": "release",
	  "mode": "hold", "reason": "health check failed", "promoted_at": "2023-11-14T22:13:20Z", "at": "2023-11-14T22:13:20Z"}]}`
	if err := os.WriteFile(filepath.Join(dir, FailureFile), []byte(old), 0o644); err != nil {
		t.Fatal(err)
	}
	b, err := OpenFailureBook(dir)
	if err != nil {
		t.Fatalf("an older agent's file: %v", err)
	}
	if _, ok := b.HoldRanOn("a"); ok {
		t.Fatal("a file without markers came back with one")
	}
	if !b.Blocks("bab", "blog", "v2", "x", false) {
		t.Fatal("the older file's record was lost")
	}
	if err := b.MarkHoldRan("a", markV2); err != nil {
		t.Fatal(err)
	}
	raw, err := os.ReadFile(filepath.Join(dir, FailureFile))
	if err != nil {
		t.Fatal(err)
	}
	// The schema 1 shape an agent before the markers decodes into.
	var older struct {
		Schema  int             `json:"schema"`
		Records []FailureRecord `json:"records"`
	}
	if err := json.Unmarshal(raw, &older); err != nil {
		t.Fatalf("an older agent cannot parse the file with markers: %v", err)
	}
	if older.Schema != 3 || len(older.Records) != 1 || older.Records[0].Build != "v2" {
		t.Fatalf("older agent's view = %+v: want schema 3 and the one record", older)
	}
	if !strings.Contains(string(raw), `"hold_ran"`) {
		t.Fatalf("the marker is not in the file:\n%s", raw)
	}
}

// Astra R3 N7: a marker never turns into permission to run a build a
// sibling's failure record holds. The instance has a good build, so it stays
// down, at the plan and at the exec of a plan made before the record.
func TestAMarkedBuildStillHonoursASiblingsFailureRecord(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "h", "s")
	h.stage("v2")
	h.install("v2", "h", "s")
	h.book.MarkHoldRan("h", markV2)
	h.sleepHost("h")
	h.wakeHost("h")
	before := h.m.Plan(ctx, h.rows.get("h"))
	if before.Refuse != "" || before.Boot.Version != "v2" || before.Mode != activator.ModeHold || before.Fallback != nil {
		t.Fatalf("plan %+v: want a hold first wake of v2 while no record holds it", before)
	}
	h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", At: t0}) // a sibling's plain failure
	if err := before.Check("v2"); err == nil {
		t.Fatal("the earlier plan's check lets v2 boot after a sibling's record of it landed")
	}
	h.wakeHost("h")
	if p := h.m.Plan(ctx, h.rows.get("h")); p.Refuse != HeldDownRefusal("v2") {
		t.Fatalf("plan %+v: want the hold-down refusal for v2", p)
	}
}

// Ruling 205 (Astra R3): the installed good build is no proof of an explicit
// rollback, because root's automatic sibling repoint installs it too. Only a
// signed pointer naming the good build, newer than the marker (`carlos
// rollback` re-signs one), lifts the marker; the pointer pass does it. Until
// then the plan is refused, and so is the good build at the exec of any
// earlier plan.
func TestASignedRollbackClearsTheMarkerAndTheGoodBuildBoots(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "h")
	h.stage("v2")
	h.sleepHost("h")
	h.wakeHost("h")
	early := h.m.Plan(ctx, h.rows.get("h"))
	if early.Refuse != "" || early.Boot.Version != "v1" || early.FirstWake || early.Check == nil {
		t.Fatalf("plan %+v: want the ordinary wake of the installed good v1 before any marker", early)
	}
	// Hold v2 ran on h, and root repointed h to its good build: v1 is
	// installed, under a marker stamped at t0. v1's pointer predates it.
	h.book.MarkHoldRan("h", markV2)
	h.pointer("stable", "v1", t0.Add(-time.Hour))
	h.advance(time.Minute)
	h.pass()
	if _, ok := h.book.HoldRanOn("h"); !ok {
		t.Fatal("a pointer older than the marker cleared it")
	}
	h.wakeHost("h")
	if p := h.m.Plan(ctx, h.rows.get("h")); p.Refuse != HeldDownRefusal("v2") {
		t.Fatalf("no rollback pointer: plan %+v, want refused", p)
	}
	if err := early.Check("v1"); err == nil || err.Error() != HeldDownRefusal("v2") {
		t.Fatalf("the earlier plan's check on v1 = %v, want the hold-down refusal for v2", err)
	}

	// `carlos rollback` signs v1's pointer afresh.
	h.pointer("stable", "v1", h.clock())
	h.advance(pointerReadEvery)
	h.pass()
	if got, ok := h.book.HoldRanOn("h"); ok {
		t.Fatalf("marker %+v survived a signed rollback newer than it", got)
	}
	h.wakeHost("h")
	p := h.m.Plan(ctx, h.rows.get("h"))
	if p.Refuse != "" || p.Boot.Version != "v1" || p.FirstWake {
		t.Fatalf("after a signed rollback: plan %+v, want v1's ordinary wake", p)
	}
	if err := p.Check("v1"); err != nil {
		t.Fatalf("after a signed rollback the check refuses v1: %v", err)
	}
}

// Ruling 205, in the order it happens (Task 3 fix round 1): hold build v2 is
// dispatched and running on h; `carlos rollback` signs v1 and root installs
// v1 on h; the attempt is superseded after v2 ran, and marks h. The marker's
// At is the attempt's dispatch, so the rollback signed while v2 ran is newer
// than it, and the pass after the mark clears it at once (setting the marker
// made the pointer read due; the pass that saw the install came before any
// marker). The sleeping h's next wake boots v1.
//
// A pointer naming v1 signed before the attempt started clears nothing: it
// says nothing about recovering from v2.
func TestAnInstallOntoTheGoodBuildMakesTheRollbackReadDueAtOnce(t *testing.T) {
	ctx := context.Background()
	for _, tc := range []struct {
		name     string
		signedAt time.Duration // the v1 pointer's PromotedAt, from the attempt's dispatch
		cleared  bool
	}{
		{"rollback signed while the hold build ran", 5 * time.Second, true},
		{"pointer signed before the hold build started", -time.Minute, false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			h := newHarness(t, "h")
			h.swap = appconfig.SwapHold
			h.stage("v2", "v3")
			h.act.mode = activator.ModeHold
			h.act.executed = func(a activator.Attempt, _ activator.SwapOutcome) bool { return a.Build == "v2" }
			// A record of another build keeps the app's pointers read; the
			// first pass reads them, so no read is due for pointerReadEvery
			// unless something makes it so.
			h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "sib", PromotedAt: t0, At: t0})
			gate := h.act.hold("h")
			h.install("v2", "h")
			h.pass()
			waitFor(t, "h's hold attempt onto v2", func() bool {
				_, started, _ := h.act.snapshot()
				return len(started) == 1 && started[0] == "h:v2"
			})
			start := h.clock()

			h.advance(5 * time.Second)
			h.pointer("stable", "v1", start.Add(tc.signedAt)) // `carlos rollback` to v1 (or an older pointer)
			h.install("v1", "h")                              // root installs it
			h.pass()                                          // the newer desire supersedes the attempt
			h.advance(5 * time.Second)
			close(gate)
			waitFor(t, "h's attempt to finish", func() bool { return !h.m.attemptRunning("h") })
			got, ok := h.book.HoldRanOn("h")
			if !ok || !got.same(markV2) || !got.At.Equal(start) {
				t.Fatalf("marker = %+v, %v: want v2's, stamped at the attempt's dispatch %v", got, ok, start)
			}

			h.sleepHost("h")
			h.pass()
			if _, still := h.book.HoldRanOn("h"); still == tc.cleared {
				t.Fatalf("after the pass the marker stands: %v, want %v", still, !tc.cleared)
			}
			h.wakeHost("h")
			p := h.m.Plan(ctx, h.rows.get("h"))
			if tc.cleared {
				if p.Refuse != "" || p.Boot.Version != "v1" || p.FirstWake {
					t.Fatalf("plan %+v after the rollback: want v1's ordinary wake", p)
				}
				if err := p.Check("v1"); err != nil {
					t.Fatalf("after the rollback the check refuses v1: %v", err)
				}
				return
			}
			if p.Refuse != HeldDownRefusal("v2") {
				t.Fatalf("plan %+v with only an older pointer: want refused", p)
			}
		})
	}
}

// Ruling 205 (Task 3 fix round 1): a planned wake's marker is stamped with
// its own plan's time, the start of its execution, kept per token: a newer
// plan for the host does not change an older wake's start. A wake whose
// start is not known (no token) falls back to the book's clock, the safe
// direction.
func TestAWakesHoldRanMarkerIsStampedWithItsPlanTime(t *testing.T) {
	ctx := context.Background()
	h := newHarness(t, "s")
	h.swap = appconfig.SwapHold
	h.stage("v2")
	h.install("v2", "s")
	h.sleepHost("s")
	h.pass()
	h.wakeHost("s")
	first := h.m.Plan(ctx, h.rows.get("s"))
	if !first.FirstWake || first.Boot.Version != "v2" || first.Mode != activator.ModeHold {
		t.Fatalf("plan %+v: want a hold first wake of v2", first)
	}
	planned := h.clock()
	h.advance(time.Minute)
	h.wakeHost("s")
	if second := h.m.Plan(ctx, h.rows.get("s")); second.Token == first.Token || second.Boot.Version != "v2" {
		t.Fatalf("second plan %+v: want another plan of v2", second)
	}
	h.advance(time.Minute)
	h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("s"), Gen: activator.Gen{Build: "v2"},
		Mode: activator.ModeHold, FirstWake: true, Token: first.Token, Executed: true})
	if got, ok := h.book.HoldRanOn("s"); !ok || !got.same(markV2) || !got.At.Equal(planned) {
		t.Fatalf("marker = %+v, %v: want v2's, stamped at its wake's plan %v", got, ok, planned)
	}

	h.book.ClearHoldRan("s")
	h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get("s"), Gen: activator.Gen{Build: "v2"},
		Mode: activator.ModeHold, FirstWake: true, Executed: true})
	if got, ok := h.book.HoldRanOn("s"); !ok || !got.At.Equal(h.clock()) {
		t.Fatalf("marker = %+v, %v: a wake of unknown start, want the book's clock %v", got, ok, h.clock())
	}
}

// Ruling 205's book half: a marker keeps the At its caller gives (the
// execution's start), and one given none is stamped by the book's clock; a
// repeated report of the same execution keeps the first stamp, and writes
// nothing (another execution renews it: bookseq_test.go). The rollback
// clear lifts only (acct, app)'s marker, and only for a pointer newer than
// it.
func TestTheHoldRanMarkerStampAndItsRollbackClear(t *testing.T) {
	now := t0
	b := &FailureBook{now: func() time.Time { return now }}
	one := markV2
	one.Exec = "a1"
	b.MarkHoldRan("a", one)
	now = now.Add(time.Minute)
	b.MarkHoldRan("a", one)
	if got, _ := b.HoldRanOn("a"); !got.At.Equal(t0) {
		t.Fatalf("marker At %v after a repeated report of its execution, want %v", got.At, t0)
	}
	if ok, _ := b.ClearHoldRanOnPointer("a", "bab", "blog", t0); ok {
		t.Fatal("a pointer as old as the marker cleared it")
	}
	if ok, _ := b.ClearHoldRanOnPointer("a", "bab", "other", t0.Add(time.Second)); ok {
		t.Fatal("another app's pointer cleared the marker")
	}
	if ok, _ := b.ClearHoldRanOnPointer("a", "bab", "blog", t0.Add(time.Second)); !ok {
		t.Fatal("a newer pointer left the marker standing")
	}
	b.MarkHoldRan("a", HoldRan{Account: "bab", App: "blog", Build: "v3"})
	if got, _ := b.HoldRanOn("a"); got.Build != "v3" || !got.At.Equal(now) {
		t.Fatalf("marker %+v: want v3's, stamped at %v", got, now)
	}
	// A report of an execution that started before the standing marker's
	// (Task 4 fix round 1) names its build but keeps the later stamp: At
	// never moves back.
	started := now.Add(-30 * time.Second)
	b.MarkHoldRan("a", HoldRan{Account: "bab", App: "blog", Build: "v4", At: started})
	if got, _ := b.HoldRanOn("a"); got.Build != "v4" || !got.At.Equal(now) {
		t.Fatalf("marker %+v: want v4's, keeping the later stamp %v", got, now)
	}
	b.ClearHoldRan("a")
	b.MarkHoldRan("a", HoldRan{Account: "bab", App: "blog", Build: "v4", At: started})
	if got, _ := b.HoldRanOn("a"); got.Build != "v4" || !got.At.Equal(started) {
		t.Fatalf("marker %+v: want v4's, stamped at its execution's start %v, not the book's clock", got, started)
	}
}

// Ruling 205 at the scheduler (Task 3 fix round 1): an awake instance (a
// socket instance, or a worker) serving v2 under v2's hold-ran marker, with
// its good v1 installed (root's repoint, say), is not moved onto v1 until a
// signed rollback lifts the marker. The gated attempt ends abandoned with
// nothing run, is derived again after the abandon deferral, and is gated
// again; once the pointer pass sees `carlos rollback`'s pointer and clears
// the marker, the next derivation runs the move.
func TestAMoveOntoTheInstalledGoodBuildWaitsForASignedRollback(t *testing.T) {
	for _, worker := range []bool{false, true} {
		name := "socket instance"
		if worker {
			name = "worker"
		}
		t.Run(name, func(t *testing.T) {
			var h *harness
			if worker {
				h = workerHarness(t, "h")
			} else {
				h = newHarness(t, "h")
			}
			var (
				mu    sync.Mutex
				gated int
			)
			h.m.d.Logf = func(format string, args ...any) {
				if strings.Contains(fmt.Sprintf(format, args...), "waiting for a signed rollback") {
					mu.Lock()
					gated++
					mu.Unlock()
				}
			}
			gatedTimes := func() int {
				mu.Lock()
				defer mu.Unlock()
				return gated
			}
			moves := func() int {
				calls, _, _ := h.act.snapshot()
				return len(calls) + len(h.act.replacements())
			}
			h.stage("v2")
			h.act.serve("h", "v2")
			h.book.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2", At: t0})
			h.pass()
			waitFor(t, "the move onto v1 to be gated", func() bool { return gatedTimes() == 1 && !h.m.attemptRunning("h") })
			if n := moves(); n != 0 {
				t.Fatalf("%d Swap/ReplaceWorker calls while v2's marker stands, want none", n)
			}
			// After the abandon deferral it is derived again, and gated again.
			h.advance(abandonDeferral + time.Second)
			h.pass()
			waitFor(t, "the move onto v1 to be gated again", func() bool { return gatedTimes() == 2 && !h.m.attemptRunning("h") })
			if n := moves(); n != 0 {
				t.Fatalf("%d Swap/ReplaceWorker calls while v2's marker stands, want none", n)
			}

			// `carlos rollback` signs v1; the pointer pass clears the marker,
			// and the next derivation moves h onto v1.
			h.pointer("stable", "v1", h.clock())
			h.advance(pointerReadEvery)
			h.pass()
			if got, ok := h.book.HoldRanOn("h"); ok {
				t.Fatalf("marker %+v survived the signed rollback", got)
			}
			waitFor(t, "the move onto v1", func() bool { return moves() == 1 })
			if calls, _, _ := h.act.snapshot(); len(calls) == 1 && (worker || calls[0].a.Build != "v1") {
				t.Fatalf("swap %+v: want one onto v1 (and none for a worker)", calls[0].a)
			}
			if r := h.act.replacements(); len(r) == 1 && (!worker || r[0].a.Build != "v1") {
				t.Fatalf("replacement %+v: want one onto v1 (and none for a socket instance)", r[0].a)
			}
		})
	}
}