package console

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"reflect"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

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

func (fx *runFx) record(t *testing.T) *provision.Record {
	t.Helper()
	rec, err := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost)
	if err != nil {
		t.Fatal(err)
	}
	return rec
}

// plainStore hides every optional extension of the store under it (ListMod
// among them), as the test fakes and the fleet store door do.
type plainStore struct{ release.Store }

func (fx *runFx) putRunJSON(t *testing.T, key string, v any) {
	t.Helper()
	raw, _ := json.Marshal(v)
	if _, err := fx.rel.Put(context.Background(), key, raw, ""); err != nil {
		t.Fatal(err)
	}
}

func TestRunSweepExpiresAQueuedRunWhoseCallerLeft(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.srv.runSweep(context.Background(), time.Now().Add(30*time.Second))
	if r := fx.record(t).ActiveRun; r.State != provision.RunQueued {
		t.Fatalf("expired before its deadline: %+v", r)
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute))
	if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
		t.Fatalf("not expired 60 s past the request (codex R3-3): %+v", r)
	}
	events, _ := fx.st.Events(OpsAccountID, "shop", 20)
	found := false
	for _, e := range events {
		found = found || (e.Action == "run-end" && strings.Contains(e.Detail, id) && strings.Contains(e.Detail, `"expired"`))
	}
	if !found {
		t.Error("no run-end expired row")
	}
	// The expire route's own result: a follower can read why it ended.
	res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
	if err != nil || res == nil || res.Refused != runs.RefusedExpired {
		t.Fatalf("expired result %+v, %v", res, err)
	}
	if fx.startRun(t) == "" {
		t.Fatal("an expired run blocked the next")
	}
}

// The deadline that counts is the record's, stamped at admission after the
// upload. request.json's requested_at, which an upload can leave minutes
// old, must not expire a run whose window has only just opened.
func TestRunSweepExpiresByTheRecordsDeadlineNotTheRequests(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", id)
	if err != nil {
		t.Fatal(err)
	}
	req.RequestedAt = time.Now().Add(-10 * time.Minute)
	raw, _ := json.Marshal(req)
	_, etag, err := fx.rel.Get(context.Background(), runs.RequestKey(release.OpsAccount, "shop", id))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := fx.rel.Put(context.Background(), runs.RequestKey(release.OpsAccount, "shop", id), raw, etag); err != nil {
		t.Fatal(err)
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(10*time.Second))
	if r := fx.record(t).ActiveRun; r.State != provision.RunQueued {
		t.Fatalf("expired by requested_at: %+v", r)
	}
}

func TestRunSweepExpiryLosesToALatePickup(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute))
	if r := fx.record(t).ActiveRun; r.State != provision.RunPicked {
		t.Fatalf("the sweep overrode a pickup: %+v", r)
	}
	if res, _ := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id); res != nil {
		t.Fatalf("the losing sweep wrote a result: %+v", res)
	}
	if ends := fx.runEnds(t, id); len(ends) != 0 {
		t.Fatalf("the losing sweep wrote run-end rows: %q", ends)
	}
}

// The other order: the sweep's CAS lands first, and the box's pickup is the
// one that changes nothing.
func TestRunSweepExpiryBeatsALatePickupTheOtherWay(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute))
	_, err := provision.SetRecordRun(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost,
		provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	if !errors.Is(err, provision.ErrRunConflict) {
		t.Fatalf("a pickup after the sweep expired the run: %v", err)
	}
	if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
		t.Fatalf("%+v", r)
	}
}

func TestRunSweepMarksALostRunDone(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
	fx.chunk(t, id, 1, runs.StreamOut, "partial")
	fx.chunk(t, id, 2, runs.StreamErr, "more")
	fx.srv.runSweep(context.Background(), time.Now().Add(30*time.Minute))
	if r := fx.record(t).ActiveRun; r.State != provision.RunPicked {
		t.Fatalf("lost before an hour past its deadline: %+v", r)
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
	if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
		t.Fatalf("lost run not done: %+v", r)
	}
	res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
	if err != nil || res == nil || res.Interrupted != runs.InterruptedLost {
		t.Fatalf("lost result %+v, %v", res, err)
	}
	if res.LastSeq != 2 || res.Box != "box-1" {
		t.Fatalf("lost result does not name the last chunk and box: %+v", res)
	}
	if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
		t.Fatalf("run-end rows %q", ends)
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(3*time.Hour))
	if ends := fx.runEnds(t, id); len(ends) != 1 {
		t.Fatalf("a second sweep wrote another row: %q", ends)
	}
}

func TestRunSweepMarksACancellingRunLost(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
	fx.transition(t, provision.RunCancelTransition(id))
	fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
	if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
		t.Fatalf("a cancelling run whose box is gone is not done: %+v", r)
	}
}

// A box that wrote its own result and died before moving the run to done:
// its result stands, and the row says result, not lost.
func TestRunSweepLostRunKeepsABoxResultThatWonTheWrite(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
	own := runs.Result{Exit: 3, Started: time.Now().Add(-time.Hour), Ended: time.Now().Add(-59 * time.Minute)}
	if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, own); err != nil {
		t.Fatal(err)
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
	if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
		t.Fatalf("%+v", r)
	}
	res, _ := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
	if res == nil || res.Exit != 3 || res.Interrupted != "" {
		t.Fatalf("the box's result was overwritten: %+v", res)
	}
	if ends := fx.runEnds(t, id); len(ends) != 1 || strings.Contains(ends[0], `"lost"`) {
		t.Fatalf("run-end rows %q", ends)
	}
}

func TestRunSweepClearsTerminalRunsAfterADay(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	ended := time.Now()
	if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Ended: ended}); err != nil {
		t.Fatal(err)
	}
	fx.transition(t, provision.RunFinish(id))
	fx.srv.runSweep(context.Background(), ended.Add(23*time.Hour))
	if fx.record(t).ActiveRun == nil {
		t.Fatal("cleared before 24 hours")
	}
	fx.srv.runSweep(context.Background(), ended.Add(25*time.Hour))
	if r := fx.record(t).ActiveRun; r != nil {
		t.Fatalf("not cleared after 24 hours: %+v", r)
	}
	if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id)); len(left) != 0 {
		t.Fatalf("objects left %q", left)
	}
}

// A cancelled or expired run is ended by the console, which wrote its result
// at that moment: the day counts from there.
func TestRunSweepClearsACancelledRunADayAfterItsResult(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	if status, body := fx.post(t, id, "cancel", ""); status != 200 {
		t.Fatalf("cancel: %d %s", status, body)
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(23*time.Hour))
	if fx.record(t).ActiveRun == nil {
		t.Fatal("cleared before 24 hours")
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(25*time.Hour))
	if r := fx.record(t).ActiveRun; r != nil {
		t.Fatalf("not cleared after 24 hours: %+v", r)
	}
	if left := fx.left(id); len(left) != 0 {
		t.Fatalf("objects left %q", left)
	}
}

// A console that died between the record's CAS and the result and run-end
// row leaves a terminal run with neither. The sweep writes them, once.
func TestRunSweepFinishesAnExpiredRunThatHasNoResult(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunExpire(id))
	fx.srv.runSweep(context.Background(), time.Now())
	res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
	if err != nil || res == nil || res.Refused != runs.RefusedExpired {
		t.Fatalf("result %+v, %v", res, err)
	}
	fx.srv.runSweep(context.Background(), time.Now())
	if ends := fx.runEnds(t, id); len(ends) != 1 {
		t.Fatalf("run-end rows %q", ends)
	}
}

func TestRunSweepDeletesOrphanRunDirectories(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)}
	fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
	young := old
	young.ID, young.RequestedAt = "run_dddddddddddddddddddddddddd", time.Now()
	fx.putRunJSON(t, runs.RequestKey(young.Account, young.App, young.ID), young)
	fx.srv.runSweep(ctx, time.Now())
	if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) != 0 {
		t.Errorf("a day-old orphan survived: %q", left)
	}
	if left, _ := fx.rel.List(ctx, runs.Prefix(young.Account, young.App, young.ID)); len(left) == 0 {
		t.Error("a fresh orphan (a create in flight) was deleted")
	}
}

// The run the record names is never an orphan, however old its request.
func TestRunSweepKeepsTheRunTheRecordNames(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(100*time.Hour)))
	fx.srv.runSweep(context.Background(), time.Now().Add(48*time.Hour))
	if left := fx.left(id); len(left) == 0 {
		t.Fatal("the picked run's objects were deleted")
	}
}

// A directory with no request.json (a stray result.json a stale retry can
// leave, or an audited marker left by a cut-short delete) is reaped once
// its newest object is a day old, when the store lists modification times.
func TestRunSweepReapsAStrayRunDirectoryByItsNewestObject(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	id := "run_eeeeeeeeeeeeeeeeeeeeeeeeee"
	if _, err := fx.rel.Put(ctx, runs.ResultKey(release.OpsAccount, "shop", id), []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	fx.srv.runSweep(ctx, time.Now())
	fx.srv.runSweep(ctx, time.Now())
	if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
		t.Fatal("a stray written a moment ago was deleted")
	}
	fx.srv.runSweep(ctx, time.Now().Add(25*time.Hour))
	if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) != 0 {
		t.Fatalf("a day-old stray survived: %q", left)
	}
}

// A store that cannot list modification times leaves strays alone.
func TestRunSweepLeavesAStrayAloneWithoutModTimes(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	fx.srv.cfg.Releases = plainStore{fx.rel}
	if _, ok := fx.srv.cfg.Releases.(release.ModLister); ok {
		t.Fatal("the wrapper still lists modification times")
	}
	id := "run_ffffffffffffffffffffffffff"
	if _, err := fx.rel.Put(ctx, runs.ResultKey(release.OpsAccount, "shop", id), []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 3; i++ {
		fx.srv.runSweep(ctx, time.Now().Add(48*time.Hour))
	}
	if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
		t.Fatal("a stray was deleted by a store that cannot say how old it is")
	}
}

func TestRunSweepLeavesAnUnreadableRequestAlone(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	id := "run_hhhhhhhhhhhhhhhhhhhhhhhhhh"
	if _, err := fx.rel.Put(ctx, runs.RequestKey(release.OpsAccount, "shop", id), bytes.Repeat([]byte("x"), 8), ""); err != nil {
		t.Fatal(err)
	}
	fx.srv.runSweep(ctx, time.Now().Add(48*time.Hour))
	if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
		t.Fatal("a directory whose request could not be read was deleted")
	}
}

// A follower's poll that lands between the sweep's lost result and its done
// CAS wins the audit marker. The row must still say lost.
func TestRunSweepLostRunRowSaysLostWhenAPollGotThereFirst(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
	lost := runs.Result{Interrupted: runs.InterruptedLost, Box: "box-1", Ended: time.Now().Add(2 * time.Hour)}
	if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, lost); err != nil {
		t.Fatal(err)
	}
	if p := fx.page(t, id, 0); p.Result == nil {
		t.Fatal("the poll did not see the result")
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
	if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
		t.Fatalf("%+v", r)
	}
	if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
		t.Fatalf("run-end rows %q", ends)
	}
}

// A run a box finished that nobody followed or acked gets its one row from
// the sweep, before the sweep clears it.
func TestRunSweepWritesTheRowOfAFinishedRunNobodyFollowed(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
	ended := time.Now()
	if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 2, Ended: ended}); err != nil {
		t.Fatal(err)
	}
	fx.transition(t, provision.RunFinish(id))
	fx.srv.runSweep(context.Background(), ended.Add(time.Hour))
	if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"result"`) {
		t.Fatalf("run-end rows %q", ends)
	}
	fx.srv.runSweep(context.Background(), ended.Add(25*time.Hour))
	if r := fx.record(t).ActiveRun; r != nil {
		t.Fatalf("not cleared: %+v", r)
	}
	if ends := fx.runEnds(t, id); len(ends) != 1 {
		t.Fatalf("run-end rows after the clear %q", ends)
	}
}

// A run the sweep marked done and then died: no row. A later pass writes it,
// as lost.
func TestRunSweepWritesTheRowOfALostRunItsSweepDiedOn(t *testing.T) {
	fx := runFixture(t)
	id := fx.startRun(t)
	fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
	lost := runs.Result{Interrupted: runs.InterruptedLost, Box: "box-1", Ended: time.Now()}
	if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, lost); err != nil {
		t.Fatal(err)
	}
	fx.transition(t, provision.RunFinish(id))
	fx.srv.runSweep(context.Background(), time.Now().Add(time.Hour))
	if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
		t.Fatalf("run-end rows %q", ends)
	}
}

// Expiry and lost-run closing run on every pass; retention and orphans only
// on housekeeping passes.
func TestRunSweepHousekeepingPhasesSkipOffPasses(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)}
	fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
	fx.startRun(t)
	fx.srv.runSweepPass(ctx, time.Now().Add(2*time.Minute), false)
	if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
		t.Fatalf("expiry did not run on an off pass: %+v", r)
	}
	if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) == 0 {
		t.Fatal("the orphan phase ran on an off pass")
	}
	fx.srv.runSweepPass(ctx, time.Now().Add(26*time.Hour), false)
	if fx.record(t).ActiveRun == nil {
		t.Fatal("the retention phase ran on an off pass")
	}
	fx.srv.runSweepPass(ctx, time.Now().Add(26*time.Hour), true)
	if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) != 0 {
		t.Fatalf("the orphan phase did not run on a housekeeping pass: %q", left)
	}
	if r := fx.record(t).ActiveRun; r != nil {
		t.Fatalf("the retention phase did not run on a housekeeping pass: %+v", r)
	}
}

// hangingStore blocks every call until its context is done.
type hangingStore struct{ release.Store }

func (hangingStore) Get(ctx context.Context, _ string) ([]byte, string, error) {
	<-ctx.Done()
	return nil, "", ctx.Err()
}
func (hangingStore) List(ctx context.Context, _ string) ([]string, error) {
	<-ctx.Done()
	return nil, ctx.Err()
}
func (hangingStore) ListDirs(ctx context.Context, _ string) ([]string, error) {
	<-ctx.Done()
	return nil, ctx.Err()
}

func TestRunSweepPassIsBoundedByATimeout(t *testing.T) {
	fx := runFixture(t)
	fx.srv.cfg.Releases = hangingStore{fx.rel}
	old := runSweepTimeout
	runSweepTimeout = 50 * time.Millisecond
	defer func() { runSweepTimeout = old }()
	done := make(chan struct{})
	go func() { fx.srv.runSweep(context.Background(), time.Now()); close(done) }()
	select {
	case <-done:
	case <-time.After(10 * time.Second):
		t.Fatal("a hung store stalled the pass")
	}
}

// A run a box finished with no follower, replaced by a new admission before
// a housekeeping pass, is an orphan when the sweep comes to it. Its row is
// written before its objects go.
func TestRunSweepWritesTheRowOfAReplacedRunBeforeDeletingIt(t *testing.T) {
	fx := runFixture(t)
	id := fx.finishedRun(t)
	fx.startRun(t) // the admission replaces the finished run on the record
	if ends := fx.runEnds(t, id); len(ends) != 0 {
		t.Fatalf("run-end rows before the sweep %q", ends)
	}
	fx.srv.runSweep(context.Background(), time.Now().Add(25*time.Hour))
	if left := fx.left(id); len(left) != 0 {
		t.Fatalf("the orphan survived: %q", left)
	}
	if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"result"`) {
		t.Fatalf("run-end rows %q", ends)
	}
}

// An orphan with no result (its record let it go before any box wrote one)
// gets a lost result and its row says lost. A run that already has its row
// does not get a second.
func TestRunSweepWritesALostRowForAnOrphanWithNoResult(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, By: "alice@keymail.dev", RequestedAt: time.Now().Add(-25 * time.Hour)}
	fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
	fx.chunk(t, old.ID, 1, runs.StreamOut, "a")
	rowed := old
	rowed.ID = "run_dddddddddddddddddddddddddd"
	fx.putRunJSON(t, runs.RequestKey(rowed.Account, rowed.App, rowed.ID), rowed)
	if _, err := fx.rel.Put(ctx, runs.AuditedKey(rowed.Account, rowed.App, rowed.ID), []byte("result"), ""); err != nil {
		t.Fatal(err)
	}
	fx.srv.runSweep(ctx, time.Now())
	for _, id := range []string{old.ID, rowed.ID} {
		if left := fx.left(id); len(left) != 0 {
			t.Fatalf("%s survived: %q", id, left)
		}
	}
	ends := fx.runEnds(t, old.ID)
	if len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) || !strings.Contains(ends[0], `"last_seq":1`) {
		t.Fatalf("run-end rows %q", ends)
	}
	if ends := fx.runEnds(t, rowed.ID); len(ends) != 0 {
		t.Fatalf("a run with its row got another: %q", ends)
	}
}

// A marker the bucket refuses leaves the orphan for a later pass, so its
// row is never lost to the delete.
func TestRunSweepKeepsAnOrphanWhoseRowCouldNotBeMarked(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)}
	fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
	refuse := true
	fx.srv.cfg.Releases = &faultStore{Store: fx.rel, put: func(key string) (bool, error) {
		if refuse && strings.HasSuffix(key, "/audited") {
			return false, errors.New("access denied")
		}
		return false, nil
	}}
	fx.srv.runSweep(ctx, time.Now())
	if left := fx.left(old.ID); len(left) == 0 {
		t.Fatal("the orphan was deleted with no row")
	}
	refuse = false
	fx.srv.runSweep(ctx, time.Now())
	if left := fx.left(old.ID); len(left) != 0 {
		t.Fatalf("the orphan survived: %q", left)
	}
	if ends := fx.runEnds(t, old.ID); len(ends) != 1 {
		t.Fatalf("run-end rows %q", ends)
	}
}

// One record that will not read does not stop expiry for the rest of the
// fleet, but it does stop the orphan phase: a run that record names could
// look like an orphan.
func TestRunSweepReadsPastAnUnreadableRecordButDeletesNoOrphan(t *testing.T) {
	fx := runFixture(t)
	ctx := context.Background()
	putShopRecord(t, fx.consoleFixture, provision.Record{Host: "other.example.com", Channel: "stable", CreatedAt: time.Now().UTC()})
	fx.startRun(t)
	old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: "other.example.com", RequestedAt: time.Now().Add(-25 * time.Hour)}
	fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
	unreadable := provision.RecordKey(release.OpsAccount, "shop", "other.example.com")
	broken := true
	fx.srv.cfg.Releases = &faultStore{Store: fx.rel, get: func(key string) error {
		if broken && key == unreadable {
			return errors.New("slow down")
		}
		return nil
	}}
	fx.srv.runSweep(ctx, time.Now().Add(2*time.Minute))
	if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
		t.Fatalf("one unreadable record stopped the expiry: %+v", r)
	}
	if left := fx.left(old.ID); len(left) == 0 {
		t.Fatal("an orphan was deleted on a pass that could not read every record")
	}
	broken = false
	fx.srv.runSweep(ctx, time.Now().Add(2*time.Minute))
	if left := fx.left(old.ID); len(left) != 0 {
		t.Fatalf("the orphan survived a full read: %q", left)
	}
}

// listCountingStore is a DirStore with the change index on that records every
// listing made through it, so a test can say which prefixes a pass walked.
// Its embedded DirStore keeps every optional extension (conditional reads,
// ETag listings, ModLister) a record view and the sweeps use.
type listCountingStore struct {
	*release.DirStore
	mu    sync.Mutex
	lists []string // "op prefix", in call order
}

func (c *listCountingStore) note(op, prefix string) {
	c.mu.Lock()
	c.lists = append(c.lists, op+" "+prefix)
	c.mu.Unlock()
}

// taken returns the listings made since the last call and forgets them.
func (c *listCountingStore) taken() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	out := c.lists
	c.lists = nil
	return out
}

func (c *listCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
	c.note("List", prefix)
	return c.DirStore.List(ctx, prefix)
}

func (c *listCountingStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	c.note("ListPage", prefix)
	return c.DirStore.ListPage(ctx, prefix, startAfter, limit)
}

func (c *listCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	c.note("ListDirs", prefix)
	return c.DirStore.ListDirs(ctx, prefix)
}

func (c *listCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
	c.note("ListETags", prefix)
	return c.DirStore.ListETags(ctx, prefix)
}

// indexedDeployment swaps fx's deployment store for a listCountingStore over the
// same bucket, with the change index on, as the console's opener builds it.
func indexedDeployment(t *testing.T, fx *consoleFixture) *listCountingStore {
	t.Helper()
	d, ok := fx.rel.(*deletableStore)
	if !ok {
		t.Fatalf("fixture store is %T", fx.rel)
	}
	ds := release.NewDirStore(d.root)
	ds.EnableChangeIndex()
	cs := &listCountingStore{DirStore: ds}
	fx.rel, fx.srv.cfg.Releases = cs, cs
	return cs
}

// armRecordView arms srv's record view over live, as cmd/console does with
// CARLOS_CHANGE_INDEX=read, and runs the backstop's first rebuild.
func armRecordView(t *testing.T, srv *Server, live release.Store) *provision.RecordView {
	t.Helper()
	v := provision.NewRecordView(live, release.NewChangeWatch(live))
	if err := v.Rebuild(context.Background()); err != nil {
		t.Fatal(err)
	}
	srv.recordView = v
	return v
}

// runsListings keeps the listings under a runs/ prefix.
func runsListings(lists []string) []string {
	var out []string
	for _, l := range lists {
		if strings.Contains(l, "/runs/") {
			out = append(out, l)
		}
	}
	sort.Strings(out)
	return out
}

// Armed, the run sweep reads its records from the record view: once the
// view has caught up, a pass with nothing written lists nothing at all.
// Disarmed, the same pass walks the bucket as before the index.
func TestRunSweepReadsViewWhenArmed(t *testing.T) {
	t.Parallel()
	fx := runFixture(t)
	cs := indexedDeployment(t, fx.consoleFixture)
	ctx := context.Background()
	id := fx.startRun(t)

	cs.taken()
	fx.srv.runSweepPass(ctx, time.Now(), false)
	if lists := cs.taken(); len(lists) == 0 {
		t.Fatal("a disarmed pass listed nothing: the counter does not see the walk")
	}

	armRecordView(t, fx.srv, cs)
	// The expiry acts on the run the view holds, and its write bumps the
	// app's scope, which the next pass's refresh re-reads.
	fx.srv.runSweepPass(ctx, time.Now().Add(2*time.Minute), false)
	if r := fx.record(t).ActiveRun; r == nil || r.ID != id || r.State != provision.RunExpired {
		t.Fatalf("armed pass did not expire the run it read from the view: %+v", r)
	}
	fx.srv.runSweepPass(ctx, time.Now().Add(3*time.Minute), false)
	cs.taken()
	for i := 0; i < 2; i++ {
		fx.srv.runSweepPass(ctx, time.Now().Add(4*time.Minute), false)
	}
	if lists := cs.taken(); len(lists) != 0 {
		t.Fatalf("armed passes with nothing written listed the bucket: %q", lists)
	}
}

// putAppRecord writes an instance record of app in the operator account.
func putAppRecord(t *testing.T, s release.Store, app, host string) {
	t.Helper()
	rec := provision.Record{Schema: 1, Host: host, Slug: strings.Split(host, ".")[0], App: app, Account: release.OpsAccount,
		Channel: "stable", CreatedAt: time.Now().UTC()}
	if err := provision.CreateRecord(context.Background(), s, rec); err != nil {
		t.Fatal(err)
	}
}

// Between daily full sweeps, orphan housekeeping lists runs/ only for apps
// with a run-start row in the last 48 hours (codex R2-1): shop ran an hour
// ago, cafe never did, so only shop's runs/ is listed.
func TestOrphanHousekeepingListsOnlyRecentRunApps(t *testing.T) {
	t.Parallel()
	fx := runFixture(t)
	cs := indexedDeployment(t, fx.consoleFixture)
	putAppRecord(t, cs, "cafe", "cafe.example.com")
	now := time.Now()
	if err := fx.st.RecordActivity(OpsAccountID, "shop", "alice@keymail.dev", Activity{Action: "run-start"}, now.Add(-time.Hour)); err != nil {
		t.Fatal(err)
	}
	armRecordView(t, fx.srv, cs)
	fx.srv.lastOrphanFull = now.Add(-time.Hour)
	cs.taken()
	fx.srv.runSweepPass(context.Background(), now, true)
	want := []string{"ListDirs " + runs.RunsPrefix(release.OpsAccount, "shop")}
	if got := runsListings(cs.taken()); !reflect.DeepEqual(got, want) {
		t.Fatalf("runs/ listings %q, want %q", got, want)
	}
	if !fx.srv.lastOrphanFull.Equal(now.Add(-time.Hour)) {
		t.Fatalf("a narrowed sweep moved the full sweep's clock to %s", fx.srv.lastOrphanFull)
	}
}

// The first housekeeping after start, and then one a day, sweeps every app
// with records, whatever the activity rows say.
func TestOrphanHousekeepingFullSweepDaily(t *testing.T) {
	t.Parallel()
	fx := runFixture(t)
	cs := indexedDeployment(t, fx.consoleFixture)
	putAppRecord(t, cs, "cafe", "cafe.example.com")
	armRecordView(t, fx.srv, cs)
	ctx := context.Background()
	all := []string{
		"ListDirs " + runs.RunsPrefix(release.OpsAccount, "cafe"),
		"ListDirs " + runs.RunsPrefix(release.OpsAccount, "shop"),
	}
	now := time.Now()

	cs.taken()
	fx.srv.runSweepPass(ctx, now, true)
	if got := runsListings(cs.taken()); !reflect.DeepEqual(got, all) {
		t.Fatalf("first housekeeping listed %q, want every app with records %q", got, all)
	}
	if !fx.srv.lastOrphanFull.Equal(now) {
		t.Fatalf("lastOrphanFull = %s after a full sweep at %s", fx.srv.lastOrphanFull, now)
	}

	fx.srv.runSweepPass(ctx, now.Add(10*time.Minute), true)
	if got := runsListings(cs.taken()); len(got) != 0 {
		t.Fatalf("housekeeping with no recent run listed %q", got)
	}

	later := now.Add(orphanFullSweepEvery + time.Minute)
	fx.srv.runSweepPass(ctx, later, true)
	if got := runsListings(cs.taken()); !reflect.DeepEqual(got, all) {
		t.Fatalf("a day later housekeeping listed %q, want %q", got, all)
	}
	if !fx.srv.lastOrphanFull.Equal(later) {
		t.Fatalf("lastOrphanFull = %s, want %s", fx.srv.lastOrphanFull, later)
	}
}