textmachine/platform/internal/runs/sweep_test.go

package runs

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"

	"textmachine/platform/internal/ingest"
	"textmachine/platform/internal/money"
	"textmachine/platform/internal/pgstore"
	"textmachine/platform/internal/pricing"
	"textmachine/platform/internal/runner"
)

// The reconciler is mostly a decision, and its decisions are tested next door without a database.
// What these tests add is the part no fake can prove: that the decision, the read model and the
// MONEY end up consistent — a run that finished has its hold settled, and one that was interrupted
// gets a second attempt with what is left rather than a second full reservation.
func sweepDB(t *testing.T) (*pgstore.Store, context.Context, string) {
	t.Helper()
	admin := os.Getenv("TM_PLATFORM_TEST_DSN")
	if admin == "" {
		t.Skip("TM_PLATFORM_TEST_DSN not set: the reconciler's money path needs a live Postgres")
	}
	ctx := t.Context()
	var suffix [6]byte
	if _, err := rand.Read(suffix[:]); err != nil {
		t.Fatal(err)
	}
	name := "tm_runs_test_" + hex.EncodeToString(suffix[:])
	conn, err := pgx.Connect(ctx, admin)
	if err != nil {
		t.Fatalf("connect: %v", err)
	}
	if _, err := conn.Exec(ctx, "create database "+pgx.Identifier{name}.Sanitize()); err != nil {
		conn.Close(ctx)
		t.Skipf("cannot create a scratch database (%v): grant CREATEDB or point the DSN at one", err)
	}
	t.Cleanup(func() {
		c, cancel := context.WithTimeout(context.Background(), 15*time.Second)
		defer cancel()
		_, _ = conn.Exec(c, "drop database if exists "+pgx.Identifier{name}.Sanitize()+" with (force)")
		conn.Close(c)
	})
	u, err := url.Parse(admin)
	if err != nil {
		t.Fatal(err)
	}
	u.Path = "/" + name
	if err := pgstore.Migrate(ctx, u.String()); err != nil {
		t.Fatalf("migrate: %v", err)
	}
	s, err := pgstore.Open(ctx, u.String())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(s.Close)
	// The scratch DSN travels with the store because one test needs a SECOND connection to the same
	// database: it takes the service's own pool away mid-pass, and the state it then asserts about has
	// to be read by something the test did not break.
	return s, ctx, u.String()
}

// fixture is one funded account with one book on disk and a service wired to it.
type fixture struct {
	svc     *Service
	store   *pgstore.Store
	ctx     context.Context
	dsn     string
	runner  *fakeRunner
	engine  *fakeEngine
	workdir string
	now     time.Time
}

func newFixture(t *testing.T, balance string, chapters int) *fixture {
	t.Helper()
	store, ctx, dsn := sweepDB(t)
	now := time.Now().UTC().Truncate(time.Millisecond)
	if _, err := store.Pool().Exec(ctx, `insert into users (id, email) values ('u1','u1@example.org')`); err != nil {
		t.Fatal(err)
	}
	amount, err := money.ParseUSD(balance)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := store.Grant(ctx, "u1", amount, "test", "seed", "", now); err != nil {
		t.Fatal(err)
	}
	workdir := t.TempDir()
	book, err := store.AddBook(ctx, pgstore.NewBook{OwnerID: "u1", Title: "蛊真人", SourceLang: "zh",
		TargetLang: "ru", ChapterCount: chapters, Workdir: workdir, Now: now})
	if err != nil {
		t.Fatal(err)
	}
	// ⚠ The book gets its chapter TREE, and the fixture was extended rather than the guard relaxed
	// (PD-405). Before this it declared N chapters and materialised none — the exact state a run may
	// no longer be started over, because every counter the screen shows is a count over `chapters`
	// and a run admitted there reads 0/total for its whole life while spending. The whole runs battery
	// was riding that state, which is why the guard's arrival is what surfaced it.
	//
	// The rows are EMPTY (nothing drafted, nothing edited), which is what the fixture already meant:
	// with no rows at all every derived counter was zero too, so no assertion here changes.
	if _, err := store.Pool().Exec(ctx, `
		insert into chapters (id, book_id, number, units_total)
		select 'c' || g, $1, g, 1 from generate_series(1, $2) g`, book, chapters); err != nil {
		t.Fatal(err)
	}
	// ⚠ AND THE BOOK GETS ITS PRICE, for the same reason it got its tree: a state the fixture used to
	// leave it in is one a run may no longer be started over. An unpriced book is REFUSED now
	// (runs.ErrNotPriced) rather than sold at a per-chapter constant, so a fixture without a
	// projection would test the refusal in every test that meant to test something else.
	//
	// The numbers are the shipped shapes rather than round ones: `fixtureChapterUSD` is the old
	// constant's $0.03 so that money assertions elsewhere stay legible against a familiar figure,
	// `fixtureStepMaxUSD` is the editor reservation the engine actually refused on 04.09 ($0.069828),
	// and the book-level bound is ZERO so that these tests measure the ORDER's own arithmetic and not
	// the terminology bond, which has its own tests in `pricing`.
	if _, err := store.Pool().Exec(ctx, `
		update books set expected_micro_usd = $2, book_once_micro_usd = 0, step_max_micro_usd = $3,
		                 source_chars = $4, structure = 'detected'
		 where id = $1`,
		book, int64(fixtureChapterUSD)*int64(max(chapters, 1)), int64(fixtureStepMaxUSD),
		int64(max(chapters, 1))*1000); err != nil {
		t.Fatal(err)
	}
	if _, err := store.Pool().Exec(ctx, `
		insert into units (id, chapter_id, ordinal, source, target, state, source_chars, expected_micro_usd)
		select 'u' || g, 'c' || g, 0, 'src', '', 'pending', 1000, $2 from generate_series(1, $1) g`,
		chapters, int64(fixtureChapterUSD)); err != nil {
		t.Fatal(err)
	}
	model, err := pricing.New(pricing.DefaultHoldFactorPercent)
	if err != nil {
		t.Fatal(err)
	}
	// The engine answers a committed AND a reserved figure from the start: the spawn reads the book's
	// meter BEFORE it starts anything, and a book that has never run reads zero for both.
	f := &fixture{store: store, ctx: ctx, dsn: dsn, runner: &fakeRunner{},
		engine: &fakeEngine{report: ingest.StatusReport{Spend: usd(0), Reserved: usd(0)}}, workdir: workdir, now: now}
	f.svc = service(t, f.runner, f.engine, now)
	f.svc.Store = store
	f.svc.Pricing = model
	return f
}

func (f *fixture) bookID(t *testing.T) string {
	t.Helper()
	lib, err := f.store.ListBooks(f.ctx, "u1", 10, "")
	if err != nil || len(lib.Books) != 1 {
		t.Fatalf("library: %+v (%v)", lib, err)
	}
	return lib.Books[0].ID
}

func (f *fixture) live(t *testing.T) pgstore.LiveRun {
	t.Helper()
	live, err := f.store.ListLiveRuns(f.ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(live) != 1 {
		t.Fatalf("%d live runs, want 1", len(live))
	}
	return live[0]
}

// hello is the handshake line of the attempt the fixture has spawned, carrying the stream id the
// PLATFORM gave it. Written from the bound id rather than from a literal because that binding is the
// whole guard against adopting somebody else's stream: a fixture with a made-up id would pass while
// the reader ignored the id entirely.
func hello(t *testing.T, f *fixture) string {
	t.Helper()
	return `{"seq":1,"type":"hello","data":{"stream_version":"1.1","engine_run_id":"` +
		f.live(t).EngineRunID + `","book_id":"b"}}` + "\n"
}

func (f *fixture) account(t *testing.T) pgstore.Account {
	t.Helper()
	a, err := f.store.ReadAccount(f.ctx, "u1")
	if err != nil {
		t.Fatal(err)
	}
	return a
}

// The whole loop: admit, spawn, the unit ends, the run finishes and the money settles at what the
// engine actually spent — with the rest of the hold given back.
func TestARunThatEndsIsFinishedAndSettledAtWhatTheEngineSpent(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if got := f.account(t); got.Reserved != fixtureHold(100) {
		t.Fatalf("the hold was not taken before the spawn: reserved %s", got.Reserved.USD())
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if len(f.runner.starts()) != 1 {
		t.Fatalf("%d units started", len(f.runner.starts()))
	}
	// A second delivery of the same queue job must NOT put a second engine on the book.
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("a retried queue job started %d units", n)
	}
	// The unit ends cleanly, and the engine reports what it spent.
	live := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo),
		runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	spent := money.MicroUSD(1_234_567)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 100, Spend: usd(spent), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	_, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if card == nil || card.Status != "ready" || card.FinishedAt == nil {
		t.Fatalf("run after the unit ended: %+v", card)
	}
	acct := f.account(t)
	if acct.Reserved != 0 {
		t.Errorf("the hold is still open after settlement: %s", acct.Reserved.USD())
	}
	if want := money.MicroUSD(10_000_000) - spent; acct.Balance != want {
		t.Errorf("balance %s, want %s — settled at what the engine spent, not at the ceiling", acct.Balance.USD(), want.USD())
	}
	if acct.Balance != acct.LedgerSum {
		t.Errorf("the cached balance and the ledger disagree: %s vs %s", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// Two callers legitimately reach the spawn at once: the queue worker that was handed the run, and
// the reconciler that finds it unspawned on its next pass. Exactly one may create a unit — the other
// would put a second engine on a book whose project file the first holds an exclusive lock on.
func TestOnlyOneOfTwoConcurrentSpawnersStartsTheEngine(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
	if err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	var wg sync.WaitGroup
	errs := make([]error, 8)
	for i := range errs {
		wg.Add(1)
		go func() {
			defer wg.Done()
			errs[i] = f.svc.spawnAttempt(f.ctx, live)
		}()
	}
	wg.Wait()
	for i, err := range errs {
		if err != nil {
			t.Errorf("spawner %d: %v", i, err)
		}
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started for one attempt, want exactly 1", n)
	}
	if got := f.live(t); got.RunID != run.ID || got.UnitName == "" {
		t.Errorf("after the race the attempt carries %+v", got)
	}
}

// A settlement figure that cannot be read leaves the hold OPEN for the next sweep. The alternative —
// settling against a guess — is the one outcome nobody can undo.
func TestASettlementThatCannotBeReadLeavesTheHoldOpen(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(run.ID, live.AttemptNo),
		runner.Marker{Unit: "u", Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	// The report carries no committed figure: absent is NOT zero (PD-40), and treating it as zero
	// would release the whole hold and charge nothing.
	f.engine.set(ingest.StatusReport{TotalUnits: 10, Done: 10}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.account(t); got.Reserved != fixtureHold(10) {
		t.Fatalf("the hold was resolved without a figure: reserved %s", got.Reserved.USD())
	}
	open, err := f.store.UnsettledRuns(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(open) != 1 {
		t.Fatalf("the unsettled run is not listed for a retry: %+v", open)
	}
	// Now the engine can answer, and the same sweep resolves it.
	spent := money.MicroUSD(200_000)
	f.engine.set(ingest.StatusReport{TotalUnits: 10, Done: 10, Spend: usd(spent), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.account(t); got.Reserved != 0 {
		t.Errorf("reserved %s after a successful retry", got.Reserved.USD())
	}
}

// A reboot takes every transient unit with it and runs no ExecStopPost, so an interrupted run has
// neither a marker nor a unit. It is restarted (unified backlog row 138) with what is LEFT of its
// budget — reserving the full ceiling again would let one run spend it twice.
func TestARunInterruptedByARebootComesBackWithTheBudgetItHasLeft(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	// Time passes and the machine reboots: the unit is gone, no marker was written, and the engine
	// reports what it managed to spend before it died.
	spent := money.MicroUSD(900_000)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 30, Spend: usd(spent), Reserved: usd(0)}, nil)
	f.runner.alive = false
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if live.AttemptNo != 2 {
		t.Fatalf("the interrupted run was not restarted: attempt %d", live.AttemptNo)
	}
	budget := fixtureHold(100)
	if want := budget - spent; live.Ceiling != want {
		t.Errorf("the second attempt reserved %s, want %s — the budget minus what was already spent",
			live.Ceiling.USD(), want.USD())
	}
	if n := len(f.runner.starts()); n != 2 {
		t.Fatalf("%d units started in all, want the original and the restart", n)
	}
	acct := f.account(t)
	if acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree after a restart", acct.Balance.USD(), acct.LedgerSum.USD())
	}
	// $10 − $0.9 spent − the new hold.
	if want := money.MicroUSD(10_000_000) - spent - live.Ceiling; acct.Balance != want {
		t.Errorf("balance %s, want %s", acct.Balance.USD(), want.USD())
	}
}

// An interrupted run whose budget is gone cannot be resumed, and the honest state for that is
// `paused` with the reason the contract has a word for — not `failed`.
func TestAnInterruptedRunWithNothingLeftIsPausedRatherThanFailed(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	// The engine spent the whole ceiling before the machine went down.
	spent := fixtureHold(10)
	f.engine.set(ingest.StatusReport{TotalUnits: 10, Done: 10, Spend: usd(spent), Reserved: usd(0)}, nil)
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	_, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if card == nil || card.Status != "paused" {
		t.Fatalf("a run with no budget left: %+v", card)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Errorf("%d units started: a run with no budget must not be respawned", n)
	}
}

// The tailer, the sink and the cursor, driven by the reconciler rather than by a test calling them
// directly: this is what proves the journal path is actually wired into the sweep.
func TestTheSweepMaterializesWhateverTheJournalHasGained(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	body := hello(t, f) +
		`{"seq":2,"type":"progress","data":{"draft":{"done":7,"total":20},"edit":{"done":1,"total":20},"eta_seconds":42}}` + "\n"
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// ⚠ What the WIRE carries is one counter in chapters, and a book with no chapter tree has
	// nothing to count — so what proves the line was materialized is the FRAME it produced. Pinning
	// the frame is also the stronger claim: it is what a watching client actually receives.
	if n := framesOfKind(t, f, pgstore.FrameProgress); n == 0 {
		t.Fatalf("the progress line materialized no frame")
	}
	// ⚠ And the NUMBERS behind that frame. Counting frames only says one was emitted; the projection
	// folding the journal wrong emits exactly the same frame. What a progress line materialises is the
	// ETA it carried and the SHAPE it announced — the per-wave counters it also used to write had no
	// reader and went with PD-411 — so those are what is asserted: read the wrong field of the line
	// and both come out wrong.
	if eta := f.runETA(t, run.ID); eta == nil || *eta != 42 {
		t.Fatalf("the journal's eta folded to %v, want 42", eta)
	}
	if editor, _ := f.waveShape(t, f.bookID(t)); editor == nil || !*editor {
		t.Fatalf("the line announced an edit wave of 20 and the book's shape folded to %v", editor)
	}
	live := f.live(t)
	if live.EngineRunID != engineStreamID(run.ID, 1) || live.Position.LastSeq != 2 || live.Position.Offset != int64(len(body)) {
		t.Errorf("cursor after the sweep: %+v", live.Position)
	}
	// Sweeping again re-reads nothing and changes nothing: at-least-once is the norm and the cursor
	// is what makes it free.
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if n := framesOfKind(t, f, pgstore.FrameProgress); n != 1 {
		t.Errorf("a second sweep re-materialized the line: %d progress frames", n)
	}
	if _, epoch := f.waveShape(t, f.bookID(t)); epoch != 0 {
		t.Errorf("a second sweep crossed a shape boundary that was never crossed: epoch=%d, want 0", epoch)
	}
}

// A deadlock Postgres broke must not stop the projection — driven through the WHOLE path, because
// the decision itself is pinned next door on a pure function and a table test cannot notice that the
// decision stopped being consulted.
//
// ⚠ That gap was real: a reviewer removed the branch in drainJournal that asks, and the table test
// passed. So this one manufactures an ACTUAL deadlock: a transaction takes the two rows in the
// inverted order while the materializer takes them in the package's order, and Postgres kills a
// side. When it kills ours, the sweep must come back with a transient error and an attempt that is
// still being materialized.
func TestADeadlockDoesNotStopTheProjection(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	// The handshake is materialized FIRST and on its own. It is the one line that does not go through
	// Apply — it binds the engine's run id with a statement of its own — so a contended attempt row
	// would block the tailer before it had taken the book lock, and there would be no cycle to break.
	body := hello(t, f)
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, _, err := f.svc.drainJournal(f.ctx, f.live(t)); err != nil {
		t.Fatal(err)
	}

	pool := f.store.Pool()
	var victim error
	for seq := 2; seq <= 9 && victim == nil; seq++ {
		// One fresh line per round: a round our side loses rolls back, a round it wins moves the cursor.
		line := fmt.Sprintf(`{"seq":%d,"type":"progress","data":{"draft":{"done":%d,"total":10}}}`, seq, seq-1) + "\n"
		fh, err := os.OpenFile(journal, os.O_APPEND|os.O_WRONLY, 0o600)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := fh.WriteString(line); err != nil {
			t.Fatal(err)
		}
		if err := fh.Close(); err != nil {
			t.Fatal(err)
		}
		live := f.live(t)
		// The inverted order: the attempt first, the book second. Nothing in the package does this —
		// that is the point, it is the shape the package's own order exists to prevent.
		evil, err := pool.Begin(f.ctx)
		if err != nil {
			t.Fatal(err)
		}
		var id int64
		if err := evil.QueryRow(f.ctx,
			`select id from run_attempts where id = $1 for update`, live.AttemptID).Scan(&id); err != nil {
			t.Fatal(err)
		}
		drained := make(chan error, 1)
		go func() {
			_, _, err := f.svc.drainJournal(f.ctx, live)
			drained <- err
		}()
		waitBlocked(t, f)
		var bookID string
		evilErr := evil.QueryRow(f.ctx,
			`select id from books where id = $1 for update`, live.BookID).Scan(&bookID)
		// The inverted transaction is over the moment the deadlock resolved, and it is let go BEFORE the
		// materializer is waited on: it holds the attempt row, and a materializer that goes on to touch
		// that row — which is exactly what the defect does — would otherwise hang instead of failing.
		_ = evil.Rollback(f.ctx)
		err = <-drained
		// Asserted every round, not only at the end: a quarantine here is the defect, and catching it
		// on the round that caused it is the difference between a failure and a hang.
		if got := f.live(t); got.Quarantined {
			t.Fatalf("the projection was quarantined over a lock that resolved itself: %v", err)
		}
		switch {
		case pgstore.IsTransient(err):
			victim = err // ours was the side Postgres killed: that is the round this test is about
		case evilErr == nil && err == nil:
			t.Fatal("no deadlock was produced: the two transactions did not contend")
		}
	}
	if victim == nil {
		t.Fatal("Postgres killed the other side in every round; the branch under test was never reached")
	}
	if got := f.live(t); got.Quarantined {
		t.Fatalf("a deadlock stopped the projection of a live run for good: %v", victim)
	}
	// And the very next sweep materializes the line the deadlock cost us.
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if n := framesOfKind(t, f, pgstore.FrameProgress); n == 0 {
		t.Errorf("the sweep after the deadlock materialized nothing")
	}
}

// framesOfKind counts the frames of one kind this book has produced. It reads the live buffer the
// stream is served from, which is the same thing a watching client would have been sent.
func framesOfKind(t *testing.T, f *fixture, kind string) int {
	t.Helper()
	frames, err := f.store.ReadFrames(f.ctx, f.bookID(t), 0, 100)
	if err != nil {
		t.Fatal(err)
	}
	n := 0
	for _, fr := range frames {
		if fr.Event == kind {
			n++
		}
	}
	return n
}

// waitBlocked waits until some transaction of this test database is waiting for a lock.
func waitBlocked(t *testing.T, f *fixture) {
	t.Helper()
	deadline := time.Now().Add(20 * time.Second)
	for {
		var blocked int
		if err := f.store.Pool().QueryRow(f.ctx, `
			select count(*) from pg_stat_activity
			 where datname = current_database() and wait_event_type = 'Lock'`).Scan(&blocked); err != nil {
			t.Fatal(err)
		}
		if blocked > 0 {
			return
		}
		if time.Now().After(deadline) {
			t.Fatal("nothing ever blocked on a lock")
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// A journal that contradicts itself quarantines the ATTEMPT and leaves the RUN alone: the engine is
// spending money the account reserved, and our inability to read its journal is not a reason to
// throw that away.
func TestAContradictoryJournalQuarantinesTheProjectionAndNotTheRun(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	handshake := hello(t, f)
	first := handshake + `{"seq":2,"type":"progress","data":{"draft":{"done":7,"total":20}}}` + "\n"
	if err := os.WriteFile(journal, []byte(first), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// The same seq comes back saying something else, and the byte hint is dropped so the reader meets
	// it where its cursor stands.
	rewritten := handshake + `{"seq":2,"type":"progress","data":{"draft":{"done":99,"total":20}}}` + "\n"
	if err := os.WriteFile(journal, []byte(rewritten), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := f.store.Pool().Exec(f.ctx, `update run_attempts set last_offset = 0`); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if !live.Quarantined {
		t.Fatal("a journal that contradicted itself did not quarantine the attempt")
	}
	_, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if card == nil || card.FinishedAt != nil {
		t.Errorf("the run was ended over a projection failure: %+v", card)
	}
	if n := framesOfKind(t, f, pgstore.FrameProgress); n != 1 {
		t.Errorf("the contradicting figure was materialized anyway: %d progress frames", n)
	}
	if got := f.account(t); got.Reserved == 0 {
		t.Error("the hold was released while the engine is still running")
	}
}

// The engine's committed figure is a LIFETIME total for the book, so the second run of a book must
// be charged for what IT spent and not for what the first one did.
//
// ⚠ Written after an adversarial review measured the opposite: two runs costing $1.00 and $0.50 were
// charged $2.50 between them. The overcharge was bounded by the hold, so it never exceeded what was
// reserved — and the ledger recorded it as "capped at the hold", which reads as an engine overspend
// rather than as the platform's own arithmetic.
func TestASecondRunOnABookIsChargedOnlyForWhatItSpent(t *testing.T) {
	f := newFixture(t, "10", 500)
	book := f.bookID(t)
	lifetime := money.MicroUSD(0)
	f.engine.set(ingest.StatusReport{Spend: usd(lifetime), Reserved: usd(0)}, nil)

	runOnce := func(spend money.MicroUSD) {
		t.Helper()
		run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, Chapters: order(100)})
		if err != nil {
			t.Fatal(err)
		}
		if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
			t.Fatal(err)
		}
		live := f.live(t)
		// The book's lifetime meter climbs; what this run cost is the difference.
		lifetime += spend
		f.engine.set(ingest.StatusReport{Spend: usd(lifetime), Reserved: usd(0)}, nil)
		if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo),
			runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
			t.Fatal(err)
		}
		if err := f.svc.Sweep(f.ctx); err != nil {
			t.Fatal(err)
		}
	}
	runOnce(money.MicroUSD(1_000_000))
	runOnce(money.MicroUSD(500_000))

	acct := f.account(t)
	if want := money.MicroUSD(10_000_000 - 1_500_000); acct.Balance != want {
		t.Fatalf("balance %s after two runs costing $1.00 and $0.50, want %s", acct.Balance.USD(), want.USD())
	}
	if acct.Reserved != 0 {
		t.Errorf("a hold is still open: %s", acct.Reserved.USD())
	}
	if acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// ceilingJudge is a fake engine start that judges `--ceiling-usd` THE WAY THE ENGINE DOES, in two
// steps that both matter:
//
//   - opening the book for writing ZEROES every leftover `reserved_usd` (store.Open →
//     recoverReservations), so what the platform read through the read-only status channel is gone
//     before a single reservation is judged;
//   - every reservation is then compared against the book's CUMULATIVE committed + reserved, so a
//     cap that is not strictly above what the book stands at denies the first one and the process
//     exits 1 having done nothing (ledger.go Reserve).
//
// It exists because a fake that merely records the argument cannot fail: the platform handed the
// engine the INCREMENT for two whole packs and every test passed, because no test modelled the one
// rule that makes the two numbers different.
type ceilingJudge struct {
	*fakeRunner
	meter  func() meter
	denied []money.MicroUSD
}

func (c *ceilingJudge) Start(ctx context.Context, s runner.Spec) error {
	got, err := ceilingArg(s.Args)
	if err != nil {
		return err
	}
	// The recovery pass, modelled: reserved is gone by the time anything is compared.
	standing := c.meter().committed
	if got <= standing {
		c.denied = append(c.denied, got)
		return fmt.Errorf("engine: reservation denied: book ceiling %s is at or below what the book stands at (%s)",
			got.USD(), standing.USD())
	}
	return c.fakeRunner.Start(ctx, s)
}

func ceilingArg(args []string) (money.MicroUSD, error) {
	for i, a := range args {
		if a == "--ceiling-usd" && i+1 < len(args) {
			return money.ParseUSD(args[i+1])
		}
	}
	return 0, errors.New("engine: --ceiling-usd is required and was not passed")
}

// The engine's ceiling flag is a CUMULATIVE book cap, not a run budget (D39.122): it overrides
// `ceilings.book_usd` and is judged against everything the book has ever committed or reserved. So
// the second run of a book must be told the sum, and handing it the increment denies its first
// reservation — the engine exits 1 and the platform can only report a `failed` run that did no work.
//
// ⚠ Written after acceptance measured exactly that, on both sides of the seam.
func TestTheSecondRunOfABookIsGivenTheCumulativeCapAndNotItsOwnIncrement(t *testing.T) {
	f := newFixture(t, "10", 500)
	lifetime := money.MicroUSD(0)
	judge := &ceilingJudge{fakeRunner: f.runner, meter: func() meter {
		return meter{committed: lifetime}
	}}
	f.svc.Runner = judge

	runOnce := func(spend money.MicroUSD) {
		t.Helper()
		run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
		if err != nil {
			t.Fatal(err)
		}
		if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
			t.Fatalf("the engine refused the ceiling it was given: %v", err)
		}
		live := f.live(t)
		lifetime += spend
		f.engine.set(ingest.StatusReport{Spend: usd(lifetime), Reserved: usd(0)}, nil)
		if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo),
			runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
			t.Fatal(err)
		}
		if err := f.svc.Sweep(f.ctx); err != nil {
			t.Fatal(err)
		}
	}
	runOnce(money.MicroUSD(3_000_000))
	runOnce(money.MicroUSD(2_000_000))

	if len(judge.denied) != 0 {
		t.Fatalf("the engine denied %v: the argument was the increment, not the cumulative cap", judge.denied)
	}
	starts := f.runner.starts()
	if len(starts) != 2 {
		t.Fatalf("%d units started", len(starts))
	}
	// $3 already committed by run one, plus run two's own increment — the hold its order was quoted
	// at, which is what the flag's cumulative arithmetic adds to the committed figure.
	second, err := ceilingArg(starts[1].Args)
	if err != nil {
		t.Fatal(err)
	}
	if want := money.MicroUSD(3_000_000) + fixtureHold(100); second != want {
		t.Errorf("the second run was given %s, want %s", second.USD(), want.USD())
	}
	// And what was actually sent is what was written down, because the meter it was computed from
	// keeps moving and the question "what limit did that process have" has to stay answerable.
	var stored int64
	if err := f.store.Pool().QueryRow(f.ctx, `
		select ceiling_arg_micro_usd from run_attempts order by id desc limit 1`).Scan(&stored); err != nil {
		t.Fatal(err)
	}
	if money.MicroUSD(stored) != second {
		t.Errorf("the attempt records a ceiling of %s and the engine got %s",
			money.MicroUSD(stored).USD(), second.USD())
	}
}

// A resumed attempt is a new process against a book whose meter the interrupted one MOVED, so its
// cap is recomputed from a fresh reading — and the reservation the dead process left behind must NOT
// inflate it.
//
// That leftover is the interesting half, and it is why this test exists next to the one above.
// `tmctl status` reports it, because the read-only path deliberately skips the recovery pass; the
// resumed engine's own `store.Open` then zeroes it before judging anything. Counting it would hand
// the resume that much room BEYOND its hold: the engine stops late, settlement caps at the hold, and
// the account underpays. See meter.bookCap — named deviation from D39.122, PD-158.
func TestAResumeIsGivenACapComputedFromTheMeterAsItStandsNow(t *testing.T) {
	f := newFixture(t, "10", 500)
	m := meter{}
	judge := &ceilingJudge{fakeRunner: f.runner, meter: func() meter { return m }}
	f.svc.Runner = judge
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	// The machine goes down: $0.90 committed, and $0.20 the dead process never released.
	m = meter{committed: 900_000, reserved: 200_000}
	f.engine.set(ingest.StatusReport{Spend: usd(m.committed), Reserved: usd(m.reserved)}, nil)
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	starts := f.runner.starts()
	if len(starts) != 2 {
		t.Fatalf("%d units started, want the original and the resume", len(starts))
	}
	resumed, err := ceilingArg(starts[1].Args)
	if err != nil {
		t.Fatal(err)
	}
	remaining := fixtureHold(100) - m.committed
	if want := m.committed + remaining; resumed != want {
		t.Errorf("the resume was given %s, want %s — what the book stands at plus what is left of the budget",
			resumed.USD(), want.USD())
	}
	if resumed >= m.committed+m.reserved+remaining {
		t.Errorf("the leftover reservation inflated the cap to %s: the engine clears it at start, so that is headroom beyond the hold",
			resumed.USD())
	}
	if len(judge.denied) != 0 {
		t.Errorf("the engine denied the resume's ceiling: %v", judge.denied)
	}
}

// A settlement that DEFERRED can be overtaken: the run is finished, so nothing stops the account
// from starting another run on the same book, and the figure the retry then reads is the BOOK's
// lifetime counter, which the second run has been moving.
//
// ⚠ Measured before the bound existed: a run that cost $0.10 was charged $2.10 — its own spend plus
// everything its successor had spent by the time the retry landed — and the successor then paid that
// same amount again. Found by two independent reviewers of this pack, reproduced by both.
func TestADeferredSettlementIsNotChargedForTheNextRunOfTheSameBook(t *testing.T) {
	f := newFixture(t, "10", 500)
	book := f.bookID(t)
	run1, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run1.ID); err != nil {
		t.Fatal(err)
	}
	live1 := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(live1.RunID, live1.AttemptNo),
		runner.Marker{Unit: live1.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	// The engine cannot be asked at the moment the run ends, so the settlement defers — the ordinary
	// deferral, already pinned elsewhere. What matters here is what happens NEXT.
	f.engine.set(ingest.StatusReport{}, errors.New("tmctl: status: database is locked"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// The book has cost $0.10 in all, and a second run starts on it.
	f.engine.set(ingest.StatusReport{Spend: usd(100_000), Reserved: usd(0)}, nil)
	run2, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, Chapters: order(100)})
	if err != nil {
		t.Fatalf("a second run on a book whose predecessor is unsettled: %v", err)
	}
	if err := f.svc.Spawn(f.ctx, run2.ID); err != nil {
		t.Fatal(err)
	}
	// The second run spends $2.00 and is still going when the first one's settlement is retried.
	f.engine.set(ingest.StatusReport{Spend: usd(2_100_000), Reserved: usd(0)}, nil)
	f.runner.alive = true
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	spent1, err := f.store.RunSpent(f.ctx, run1.ID)
	if err != nil {
		t.Fatal(err)
	}
	if want := money.MicroUSD(100_000); spent1 != want {
		t.Fatalf("the first run was charged %s for work that cost %s", spent1.USD(), want.USD())
	}
	// ...and the second run then pays for its own, once.
	live2 := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(live2.RunID, live2.AttemptNo),
		runner.Marker{Unit: live2.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	spent2, err := f.store.RunSpent(f.ctx, run2.ID)
	if err != nil {
		t.Fatal(err)
	}
	if want := money.MicroUSD(2_000_000); spent2 != want {
		t.Errorf("the second run was charged %s, want %s", spent2.USD(), want.USD())
	}
	acct := f.account(t)
	if want := money.MicroUSD(10_000_000 - 2_100_000); acct.Balance != want {
		t.Errorf("balance %s after two runs costing $2.10 between them, want %s", acct.Balance.USD(), want.USD())
	}
	if acct.Reserved != 0 {
		t.Errorf("a hold is still open: %s", acct.Reserved.USD())
	}
	if acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// A book meter that reads BELOW an attempt's own baseline is not a refund — it is a project database
// that was replaced — so the attempt settles at nothing. Silently charging nothing is the part that
// is not acceptable: a settlement of zero has to be findable in a log rather than in a balance.
func TestAMeterThatWentBackwardsSettlesAtNothingAndSaysSo(t *testing.T) {
	f := newFixture(t, "10", 500)
	var log bytes.Buffer
	f.svc.Log = slog.New(slog.NewTextHandler(&log, &slog.HandlerOptions{Level: slog.LevelWarn}))
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	// The attempt starts on a book that has already cost $1.00...
	f.engine.set(ingest.StatusReport{Spend: usd(1_000_000), Reserved: usd(0)}, nil)
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo),
		runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	// ...and the database it was translating is replaced by an older copy.
	f.engine.set(ingest.StatusReport{Spend: usd(200_000), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	acct := f.account(t)
	if acct.Balance != money.MicroUSD(10_000_000) {
		t.Errorf("balance %s: a meter that went backwards must not charge and must not refund", acct.Balance.USD())
	}
	if acct.Reserved != 0 {
		t.Errorf("the hold is still open: %s", acct.Reserved.USD())
	}
	if !strings.Contains(log.String(), "below this attempt's own baseline") {
		t.Errorf("a settlement of nothing was silent; the log said: %s", log.String())
	}
	if strings.Contains(log.String(), "1.000000") || strings.Contains(log.String(), "0.200000") {
		t.Errorf("the line carries money: %s", log.String())
	}
}

// A restart while the engine's figure cannot be read must NOT open a second reservation: the first
// one is still open, and nothing would ever come back for it.
//
// ⚠ Also written after an adversarial review measured it: a $3.00 run showed $6.00 reserved and ended
// with $3.00 reserved permanently, listed by nothing.
func TestARestartIsDeferredWhileTheInterruptedAttemptIsUnsettled(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	held := f.account(t).Reserved
	if held == 0 {
		t.Fatal("nothing was reserved")
	}
	// The machine rebooted and the engine can no longer be asked.
	f.engine.set(ingest.StatusReport{}, errors.New("tmctl: config: no such file"))
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.account(t).Reserved; got != held {
		t.Fatalf("reserved %s after a deferred restart, want the original %s", got.USD(), held.USD())
	}
	if live := f.live(t); live.AttemptNo != 1 {
		t.Fatalf("a second attempt was opened on an unsettled one: attempt %d", live.AttemptNo)
	}
	// Once the engine answers again the run restarts, with the budget its predecessor left.
	spent := money.MicroUSD(900_000)
	f.engine.set(ingest.StatusReport{Spend: usd(spent), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if live.AttemptNo != 2 {
		t.Fatalf("attempt %d after the engine came back", live.AttemptNo)
	}
	acct := f.account(t)
	if want := fixtureHold(100) - spent; live.Ceiling != want {
		t.Errorf("the new attempt reserved %s, want %s", live.Ceiling.USD(), want.USD())
	}
	if acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// An attempt that was replaced leaves its reservation open while its RUN goes on. A list keyed on the
// run being over never looked at it again — the hold stayed reserved for the life of the account.
func TestAnInterruptedAttemptsHoldIsStillFoundWhileItsRunGoesOn(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	// Close the attempt behind the reconciler's back, as an operator or a crash-repair would, leaving
	// its reservation open and the run alive.
	if _, err := f.store.Pool().Exec(f.ctx,
		`update run_attempts set ended_at = now() where id = $1`, live.AttemptID); err != nil {
		t.Fatal(err)
	}
	open, err := f.store.UnsettledRuns(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(open) != 1 || open[0].AttemptNo != 1 {
		t.Fatalf("the stranded hold is not listed: %+v", open)
	}
}

// A run that was admitted and never started cost nothing, so its hold comes back WHOLE. Without this
// the money of a run that never ran stays reserved for the life of the account.
func TestTheHoldOfARunThatNeverStartedComesBackWhole(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if live.UnitName != "" {
		t.Fatal("the attempt was spawned; this test is about one that never was")
	}
	if err := runner.WriteMarker(f.svc.markerPath(run.ID, live.AttemptNo),
		runner.Marker{Unit: "u", Result: "exit-code", Code: "exited", Status: "1"}); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	acct := f.account(t)
	if acct.Reserved != 0 {
		t.Errorf("reserved %s for a run that never started", acct.Reserved.USD())
	}
	if acct.Balance != money.MicroUSD(10_000_000) {
		t.Errorf("balance %s: a run that never started must cost nothing", acct.Balance.USD())
	}
}

// A unit that CANNOT be created must leave the attempt retryable. The claim used to survive the
// failure, and a recorded unit name with no unit and no marker is exactly the shape of an interrupted
// run — so every sweep restarted the run, settled, took a fresh hold and failed to spawn again, and a
// run whose engine never started ate its whole ceiling a sweep at a time.
func TestAUnitThatCannotBeCreatedDoesNotEatTheRunsBudget(t *testing.T) {
	f := newFixture(t, "10", 500)
	f.runner.startErr = errors.New("Failed to start transient scope unit")
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err == nil {
		t.Fatal("a failed spawn was reported as success")
	}
	held := f.account(t).Reserved
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	for range 6 {
		if err := f.svc.Sweep(f.ctx); err != nil {
			t.Fatal(err)
		}
	}
	live := f.live(t)
	if live.AttemptNo != 1 {
		t.Fatalf("six sweeps of a run that never started produced attempt %d", live.AttemptNo)
	}
	if got := f.account(t).Reserved; got != held {
		t.Errorf("reserved %s after six failed spawns, want the original %s", got.USD(), held.USD())
	}
	if got := f.account(t).Balance; got != money.MicroUSD(10_000_000)-held {
		t.Errorf("balance %s: a run whose engine never started must cost nothing", got.USD())
	}
}

// A spawn that reported failure did not necessarily fail: systemd-run can be killed after it has
// already asked for the unit, and then an engine is running while the platform believes none is. The
// claim is given back so the attempt can be retried — and the RE-claim must keep the baseline the
// first one recorded, because the meter it would read now includes work this very attempt has done.
//
// ⚠ Found by an outside-the-map reviewer of this pack, reading the path rather than the report.
func TestAReclaimedAttemptKeepsTheBaselineItFirstRecorded(t *testing.T) {
	f := newFixture(t, "10", 500)
	f.runner.startErr = errors.New("Failed to start transient scope unit: connection timed out")
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err == nil {
		t.Fatal("the spawn reported success")
	}
	live := f.live(t)
	if live.UnitName != "" {
		t.Fatal("the claim was not given back")
	}
	// The unit was created after all, and the engine it holds has spent $0.40 by the next sweep.
	f.engine.set(ingest.StatusReport{Spend: usd(400_000), Reserved: usd(0)}, nil)
	f.runner.startErr = nil
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	var baseline, stored *int64
	if err := f.store.Pool().QueryRow(f.ctx, `
		select spend_baseline_micro_usd, ceiling_arg_micro_usd from run_attempts where id = $1`,
		live.AttemptID).Scan(&baseline, &stored); err != nil {
		t.Fatal(err)
	}
	if baseline == nil || stored == nil {
		t.Fatal("the attempt has no baseline at all")
	}
	if *baseline != 0 {
		t.Errorf("the re-claim moved the baseline to %s; the attempt would then be billed the difference from its own work",
			money.MicroUSD(*baseline).USD())
	}
	// And the limit the engine was HANDED on the retry is the one the row records. Recomputing it from
	// a counter the first claim's own engine has been moving hands that engine a second, larger limit
	// and leaves the forensic column describing neither.
	starts := f.runner.starts()
	if len(starts) != 1 {
		t.Fatalf("%d units started", len(starts))
	}
	handed, err := ceilingArg(starts[0].Args)
	if err != nil {
		t.Fatal(err)
	}
	if handed != money.MicroUSD(*stored) {
		t.Errorf("the retry handed the engine %s and the row records %s",
			handed.USD(), money.MicroUSD(*stored).USD())
	}
	if want := fixtureHold(100); handed != want {
		t.Errorf("the retry handed %s, want the limit the first claim decided (%s)", handed.USD(), want.USD())
	}
}

// A journal that cannot be read stops the PROJECTION and nothing else. One malformed line used to
// abort the reconcile before the exit marker was even looked at, so the run stayed "translating"
// forever with its hold reserved — the engine long gone and the marker on disk.
func TestAnUnreadableJournalDoesNotStopTheRunFromFinishing(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(filepath.Join(f.workdir, ingest.JournalFile),
		[]byte("this is not an event at all\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo),
		runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	spent := money.MicroUSD(400_000)
	f.engine.set(ingest.StatusReport{Spend: usd(spent), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	_, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if card == nil || card.FinishedAt == nil {
		t.Fatalf("a run whose journal is unreadable never finished: %+v", card)
	}
	if got := f.account(t); got.Reserved != 0 {
		t.Errorf("its hold is still open: %s", got.Reserved.USD())
	}
}

// The settlement's TWO deferral branches are different code and both must leave the money alone: the
// status call failing, and the report carrying no figure. Only the second was pinned, and the first
// is the commoner one.
func TestAFailedStatusCallLeavesTheMoneyExactlyWhereItWas(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(run.ID, live.AttemptNo),
		runner.Marker{Unit: "u", Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	held := f.account(t).Reserved
	f.engine.set(ingest.StatusReport{}, errors.New("tmctl: status: database is locked"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.account(t).Reserved; got != held {
		t.Errorf("reserved %s after a failed status call, want the original %s", got.USD(), held.USD())
	}
	open, err := f.store.UnsettledRuns(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(open) != 1 {
		t.Fatalf("the run is not on the settlement worklist: %+v", open)
	}
	var settled *time.Time
	if err := f.store.Pool().QueryRow(f.ctx, `select settled_at from runs where id=$1`, run.ID).Scan(&settled); err != nil {
		t.Fatal(err)
	}
	if settled != nil {
		t.Error("a run was stamped settled although its money never resolved")
	}
}

// A run that cannot be resumed on the CURRENT balance is paused with the reason the contract has a
// word for — not left live to be retried on every sweep for the rest of its life.
func TestAnInterruptedRunThatTheBalanceCannotCarryIsPaused(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	// The engine spent almost nothing, but the rest of the balance went elsewhere while the run was
	// down, so the restart cannot reserve what is left of the budget.
	spent := money.MicroUSD(100_000)
	f.engine.set(ingest.StatusReport{Spend: usd(spent), Reserved: usd(0)}, nil)
	if _, err := f.store.Adjust(f.ctx, "u1", money.MicroUSD(-9_000_000), "test", "elsewhere", "spent", f.now); err != nil {
		t.Fatal(err)
	}
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	_, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if card == nil || card.Status != "paused" || card.PausedReason != "credit_exhausted" {
		t.Fatalf("a run the balance cannot carry: %+v", card)
	}
}

// Lines read and NOT applied still move the byte hint. Without that they are re-read on every sweep
// for the life of the run — the case the code names and the one no test reached, because a sweep that
// applies events moves the hint through the sink instead.
func TestASweepOverAlreadyAppliedLinesStillMovesTheByteHint(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	body := hello(t, f) +
		`{"seq":2,"type":"progress","data":{"draft":{"done":1,"total":10}}}` + "\n"
	if err := os.WriteFile(filepath.Join(f.workdir, ingest.JournalFile), []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// Drop the hint but keep the seq, which is what a lost hint looks like. The next sweep re-reads
	// every line, applies none of them — and must still leave the hint at the end of the file.
	if _, err := f.store.Pool().Exec(f.ctx, `update run_attempts set last_offset = 0`); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.live(t); got.Position.Offset != int64(len(body)) {
		t.Errorf("the byte hint is at %d after a duplicate-only sweep, want %d", got.Position.Offset, len(body))
	}
}

// The spawn reads the book's meter BEFORE it starts anything, and a meter it cannot read REFUSES the
// attempt: with no baseline the settlement charges this run for everything the book has ever cost,
// and with no reserved figure the engine is handed a cap below its own ledger. Starting anyway is
// how a run gets paid for and billed wrong; starting one sweep later costs a sweep.
//
// ⚠ The refusal itself was built and left unpinned, and a mutation that answered "zero, no error"
// survived the whole battery.
func TestAnAttemptWhoseMeterCannotBeReadIsNotStartedAtAll(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	for _, tc := range []struct {
		name string
		rep  ingest.StatusReport
		err  error
	}{
		{"the call fails", ingest.StatusReport{}, errors.New("tmctl: status: database is locked")},
		{"no committed figure", ingest.StatusReport{Reserved: usd(0)}, nil},
		{"no reserved figure", ingest.StatusReport{Spend: usd(0)}, nil},
	} {
		f.engine.set(tc.rep, tc.err)
		if err := f.svc.Spawn(f.ctx, run.ID); err == nil {
			t.Fatalf("%s: the attempt was started anyway", tc.name)
		}
		if n := len(f.runner.starts()); n != 0 {
			t.Fatalf("%s: %d units started", tc.name, n)
		}
		if live := f.live(t); live.UnitName != "" {
			t.Fatalf("%s: the attempt was claimed with unit %q, so no later sweep will retry it", tc.name, live.UnitName)
		}
	}
	// The engine answers again and the same attempt starts, which is what makes the refusal a delay
	// rather than a lost run.
	f.engine.set(ingest.StatusReport{Spend: usd(0), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started once the engine could be read", n)
	}
}

// A marker left behind by an earlier run of the SAME attempt is read as this one's ending the moment
// the reconciler looks — a run that has just been started is finished and settled while its engine
// is alive and spending. The names carry the attempt number, so this needs a re-run of one attempt;
// "only then" is not "never", and the clearing was built and left unpinned.
func TestAStaleExitMarkerIsClearedBeforeTheUnitStarts(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	// The leftover: attempt 1 of this run ended once before, and its marker is still on disk.
	marker := f.svc.markerPath(live.RunID, live.AttemptNo)
	if err := runner.WriteMarker(marker, runner.Marker{Unit: unitName(live.RunID, live.AttemptNo),
		Result: "exit-code", Code: "exited", Status: "1"}); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := runner.ReadMarker(marker); !errors.Is(err, runner.ErrNoMarker) {
		t.Fatalf("the stale marker survived the spawn: %v", err)
	}
	f.runner.alive = true
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	_, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if card == nil || card.FinishedAt != nil {
		t.Fatalf("a run whose engine had just been started was finished from an old marker: %+v", card)
	}
	if got := f.account(t).Reserved; got == 0 {
		t.Error("its hold was resolved while the engine is running")
	}
}

// The sweep reconciles from a list it read BEFORE working through it, and a run can be spawned,
// spend and exit inside that window — a fast run with other runs ahead of it in the pass. The
// settlement's "admitted and never started" branch must therefore re-check the row, not the
// snapshot, or the hold of a run that really spent money comes back whole.
//
// ⚠ Measured by acceptance: an attempt that spent $0.50 was charged $0.000000.
func TestAStaleSnapshotDoesNotGiveBackTheHoldOfAnAttemptThatSpent(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	stale := f.live(t)
	if stale.UnitName != "" || stale.SpendBaseline != nil {
		t.Fatalf("the snapshot must predate the spawn: %+v", stale)
	}
	// Meanwhile the queue worker starts the run, the engine spends $0.50 and the unit exits.
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	fresh := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(fresh.RunID, fresh.AttemptNo),
		runner.Marker{Unit: fresh.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	spent := money.MicroUSD(500_000)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 100, Spend: usd(spent), Reserved: usd(0)}, nil)
	// The sweep finally reaches this run, carrying the snapshot it started with.
	if _, err := f.svc.reconcile(f.ctx, stale); err != nil {
		t.Fatal(err)
	}
	if got := f.account(t); got.Balance == money.MicroUSD(10_000_000) {
		t.Fatal("the whole hold came back for an attempt that spent $0.50")
	}
	// The next sweep, with a snapshot that has the unit in it, settles at what was actually spent.
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	acct := f.account(t)
	if want := money.MicroUSD(10_000_000) - spent; acct.Balance != want {
		t.Errorf("balance %s, want %s", acct.Balance.USD(), want.USD())
	}
	if acct.Reserved != 0 {
		t.Errorf("a hold is still open: %s", acct.Reserved.USD())
	}
	if acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// Unified backlog row 139: a run is PINNED to the engine build it started with, and a resume stays on
// it. The engine is deployed more often than a translation finishes, so the quiet behaviour would be
// for a resumed run to continue under a program nobody chose for it.
func TestAResumeStaysOnTheEngineBuildTheRunStartedWith(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	started := f.svc.Cfg.EngineBinary
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if got := f.runner.starts()[0].Binary; got != started {
		t.Fatalf("the first attempt ran %q, want the configured build", got)
	}
	// The engine is redeployed while the run is down.
	f.svc.Cfg.EngineBinary = "/opt/engine/2026.09.01/tmctl"
	spent := money.MicroUSD(100_000)
	f.engine.set(ingest.StatusReport{Spend: usd(spent), Reserved: usd(0)}, nil)
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	starts := f.runner.starts()
	if len(starts) != 2 {
		t.Fatalf("%d units started", len(starts))
	}
	if got := starts[1].Binary; got != started {
		t.Errorf("the resumed attempt ran %q, want the pinned %q", got, started)
	}
	if got := f.live(t).EngineBinary; got != started {
		t.Errorf("the new attempt records %q as its build", got)
	}
}

// ...and moving to another build is allowed, but only by saying so.
func TestAResumeMovesToANewEngineBuildOnlyWhenItIsAllowed(t *testing.T) {
	f := newFixture(t, "10", 500)
	f.svc.Cfg.AllowEngineVersionChange = true
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	next := "/opt/engine/2026.09.01/tmctl"
	f.svc.Cfg.EngineBinary = next
	spent := money.MicroUSD(100_000)
	f.engine.set(ingest.StatusReport{Spend: usd(spent), Reserved: usd(0)}, nil)
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	starts := f.runner.starts()
	if len(starts) != 2 || starts[1].Binary != next {
		t.Fatalf("the resumed attempt ran %q, want the new build", starts[len(starts)-1].Binary)
	}
}

// Cross-family review of the acceptance dofix (M3): the same hold, on the host that actually produces
// this case. An attempt is unspawned BECAUSE the engine could not be run — and settlement used to ask
// that same engine for a committed spend before it would give the money back, so every pass failed on
// the call and the hold of a run that never ran stayed reserved for good.
func TestTheHoldOfARunThatNeverStartedComesBackOnAHostWhoseEngineCannotAnswer(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if live.UnitName != "" {
		t.Fatal("the attempt was spawned; this test is about one that never was")
	}
	f.engine.set(ingest.StatusReport{}, errors.New("tmctl status: exec: no such file or directory"))
	if err := runner.WriteMarker(f.svc.markerPath(run.ID, live.AttemptNo),
		runner.Marker{Unit: "u", Result: "exit-code", Code: "exited", Status: "1"}); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	acct := f.account(t)
	if acct.Reserved != 0 {
		t.Fatalf("reserved %s: the hold of a run that never started waits on an engine that cannot answer",
			acct.Reserved.USD())
	}
	if acct.Balance != money.MicroUSD(10_000_000) {
		t.Errorf("balance %s: a run that never started must cost nothing", acct.Balance.USD())
	}
	if f.engine.called() != 0 {
		t.Errorf("the engine was asked %d times about an attempt that never reached it", f.engine.called())
	}
}

// A continuation is priced at what the run was SOLD for, not at what the deployment's own setting
// says on the day it continues. The hold cushion is a deployment setting; a restart that derives the
// budget from it again re-prices a paid run in both directions (PD-168: with the setting doubled a
// $2.50 remainder holds $5.50, and with it cut below what was spent a run with chapters left pauses
// as exhausted). The budget is the first attempt's hold, read back.
//
// ⚠ The moving setting is now the CUSHION and not a per-chapter rate — the rate is gone entirely and
// what a chapter costs is read from the engine (unified backlog row 280). The defect this guards is
// unchanged: it was never about which setting moved, only about a continuation being re-priced from
// ANY of them.
//
// Mutation caught: `budget, err := s.Store.RunBudget(...)` in reopen replaced by a fresh quote of the order.
func TestARestartHoldsWhatTheRunWasSoldForWhenTheRateHasMovedSince(t *testing.T) {
	for name, factor := range map[string]int{
		"doubled": 2 * pricing.DefaultHoldFactorPercent,
		"raised":  pricing.DefaultHoldFactorPercent + 60,
		"at the floor, well under what was spent": 100,
	} {
		t.Run(name, func(t *testing.T) {
			f := newFixture(t, "10", 500)
			run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
			if err != nil {
				t.Fatal(err)
			}
			if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
				t.Fatal(err)
			}
			sold := fixtureHold(100) // $3.00: the hold the user agreed to
			spent := money.MicroUSD(500_000)
			f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 30, Spend: usd(spent), Reserved: usd(0)}, nil)
			f.runner.alive = false
			f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
			// The deployment's rate moves between the purchase and the reboot.
			moved, err := pricing.New(factor)
			if err != nil {
				t.Fatal(err)
			}
			f.svc.Pricing = moved
			if err := f.svc.Sweep(f.ctx); err != nil {
				t.Fatal(err)
			}
			live, err := f.store.ListLiveRuns(f.ctx)
			if err != nil {
				t.Fatal(err)
			}
			if len(live) != 1 || live[0].AttemptNo != 2 {
				t.Fatalf("the interrupted run was not restarted (%+v): a run sold with $2.50 left was read at today's rate", live)
			}
			if want := sold - spent; live[0].Ceiling != want {
				t.Errorf("the second attempt reserved %s, want %s — what the run was sold for less what it spent, not %s at today's rate",
					live[0].Ceiling.USD(), want.USD(), (fixtureHoldAt(factor, 100) - spent).USD())
			}
			acct := f.account(t)
			if acct.Balance != acct.LedgerSum {
				t.Errorf("balance %s and ledger %s disagree after a restart", acct.Balance.USD(), acct.LedgerSum.USD())
			}
			if want := money.MicroUSD(10_000_000) - spent - (sold - spent); acct.Balance != want {
				t.Errorf("balance %s, want %s", acct.Balance.USD(), want.USD())
			}
		})
	}
}

// A quarantine is lifted by the operator (PD-426) and the next sweep materializes the journal again
// from the cursor the projection stopped at — not from the start, and not from wherever the file has
// grown to. Driven through the sweep because `drainJournal`'s first line is the quarantine check: a
// lift that cleared the column while nothing read it would look lifted and do nothing.
//
// Mutation caught: Unquarantine clearing nothing; drainJournal ignoring `Quarantined`; the lift
// resetting the cursor; a second lift answering success.
func TestALiftedQuarantineMaterializesTheJournalAgainFromTheCursor(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	body := hello(t, f) +
		`{"seq":2,"type":"progress","data":{"draft":{"done":7,"total":20},"edit":{"done":1,"total":20},"eta_seconds":42}}` + "\n"
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	before := f.live(t)
	if before.Position.LastSeq != 2 {
		t.Fatalf("cursor after the first sweep: %+v", before.Position)
	}
	// The projection is quarantined — by whatever the reader could not read at the time — and the
	// journal grows meanwhile.
	const reason = "ingest: unsupported stream version: stream is 9.9, this build speaks 1.1"
	if err := f.store.Quarantine(f.ctx, before.AttemptID, reason); err != nil {
		t.Fatal(err)
	}
	more := `{"seq":3,"type":"progress","data":{"draft":{"done":9,"total":20},"edit":{"done":1,"total":20},"eta_seconds":41}}` + "\n"
	if err := os.WriteFile(journal, []byte(body+more), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if l := f.live(t); !l.Quarantined || l.Position.LastSeq != 2 {
		t.Fatalf("a quarantined attempt was read anyway: %+v", l.Position)
	}
	lifted, err := f.store.Unquarantine(f.ctx, run.ID)
	if err != nil {
		t.Fatal(err)
	}
	if lifted.AttemptNo != 1 || lifted.Reason != reason || lifted.Position.LastSeq != 2 || lifted.Position.Offset != int64(len(body)) {
		t.Fatalf("the lift reported %+v, want attempt 1, the reason it carried, and the cursor at seq 2 / offset %d", lifted, len(body))
	}
	// The cursor itself, read back from the database BEFORE the next sweep: the lift clears the column
	// and nothing else, and "nothing else" is both halves — the seq and the byte hint. A lift that
	// reset the hint would send the reader back over an earlier attempt's lines, where a foreign seq
	// meets ours and quarantines the very attempt the lift was for.
	held := f.live(t)
	if held.Quarantined || held.Position.LastSeq != before.Position.LastSeq ||
		held.Position.Offset != before.Position.Offset || !bytes.Equal(held.Position.LastHash, before.Position.LastHash) {
		t.Fatalf("after the lift the stored cursor is %+v (quarantined=%v), want it untouched at %+v",
			held.Position, held.Quarantined, before.Position)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	after := f.live(t)
	if after.Quarantined || after.Position.LastSeq != 3 || after.Position.Offset != int64(len(body+more)) {
		t.Fatalf("after the lift the sweep left the cursor at %+v (quarantined=%v), want seq 3 at the end of the journal", after.Position, after.Quarantined)
	}
	if n := framesOfKind(t, f, pgstore.FrameProgress); n != 2 {
		t.Errorf("%d progress frames, want the one before the quarantine and the one after the lift", n)
	}
	// Lifting again is refused with its own word: there is nothing to lift.
	if _, err := f.store.Unquarantine(f.ctx, run.ID); !errors.Is(err, pgstore.ErrNotQuarantined) {
		t.Fatalf("a second lift answered %v, want ErrNotQuarantined", err)
	}
}

// The budget is the FIRST attempt's hold, and only the first: a run interrupted twice is re-held
// against what it was sold for less everything it has spent, not against the previous attempt's
// remainder less that attempt's spend — which is the same number once and a smaller one every time
// after, and under-holds the third attempt by the first attempt's spend. Two interruptions are the
// smallest shape on which the two readings differ.
//
// Mutation caught: RunBudget reading the run's LATEST reservation instead of attempt 1's.
func TestATwiceInterruptedRunIsStillHeldAgainstWhatItWasSoldFor(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	sold := fixtureHold(100)
	// First interruption: $0.50 spent. The meter is the BOOK's lifetime counter, so it only grows.
	first := money.MicroUSD(500_000)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 20, Spend: usd(first), Reserved: usd(0)}, nil)
	f.runner.alive = false
	f.svc.Now = func() time.Time { return f.now.Add(time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	second := f.live(t)
	if second.AttemptNo != 2 || second.Ceiling != sold-first {
		t.Fatalf("after the first interruption: attempt %d holds %s, want attempt 2 holding %s", second.AttemptNo, second.Ceiling.USD(), (sold - first).USD())
	}
	// The second attempt runs and is interrupted too, another $0.30 later.
	f.runner.alive = true
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	more := money.MicroUSD(300_000)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 35, Spend: usd(first + more), Reserved: usd(0)}, nil)
	f.runner.alive = false
	f.svc.Now = func() time.Time { return f.now.Add(3 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	third := f.live(t)
	if third.AttemptNo != 3 {
		t.Fatalf("after the second interruption: attempt %d, want 3", third.AttemptNo)
	}
	if want := sold - first - more; third.Ceiling != want {
		t.Errorf("the third attempt holds %s, want %s — sold for %s, spent %s in all; a budget read from the previous attempt's hold would give %s",
			third.Ceiling.USD(), want.USD(), sold.USD(), (first + more).USD(), ((sold - first) - (first + more)).USD())
	}
	acct := f.account(t)
	if acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
	if want := money.MicroUSD(10_000_000) - first - more - third.Ceiling; acct.Balance != want {
		t.Errorf("balance %s, want %s", acct.Balance.USD(), want.USD())
	}
}

// A PARKED attempt keeps its freshness. The tailer stops where another stream begins, which leaves a
// cursor that does not move, no error and no quarantine — so nothing in the row says the projection
// has stopped, and the repair channel's guard («the stream is speaking, do not ask the engine») reads
// that silence as speech unless the pass carries the park to it. The sweep carries it, and the
// operator gets a WARN naming the run.
//
// ⚠ The stranger may be THIS RUN: a respawn of the same attempt is handed the same stream id and the
// engine mints a fresh one when that id has already written for this book — so «parked» is not
// «our process is gone», and a run can sit here for its whole life.
//
// Mutation caught: dropping `parked` from the resync guard; swallowing ErrForeignStreamAhead in
// drainJournal (the park then reads as "caught up" and the guard shuts the repair channel again).
func TestAParkedAttemptStillGetsTheRepairChannel(t *testing.T) {
	f := newFixture(t, "10", 500)
	// The WARN is the operator's only sight of this state, so it is asserted as an EMISSION and not as
	// a call: a guard around the line, or a line put back on every pass, leaves a function-level pin
	// green while the deliverable changes.
	var log bytes.Buffer
	f.svc.Log = slog.New(slog.NewTextHandler(&log, &slog.HandlerOptions{Level: slog.LevelWarn}))
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	// Our own handshake and one progress line, then a stranger's handshake: the shape a respawn of
	// this same attempt writes once the engine has re-minted its id.
	body := hello(t, f) +
		`{"seq":2,"type":"progress","data":{"draft":{"done":3,"total":20},"eta_seconds":99}}` + "\n" +
		`{"seq":1,"type":"hello","data":{"stream_version":"1.1","engine_run_id":"SOMEONE-ELSE","book_id":"b"}}` + "\n"
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	// The engine answers the repair channel with figures the stream never delivered.
	f.engine.set(ingest.StatusReport{TotalUnits: 20, Done: 7, ETASeconds: 42, Spend: usd(0), Reserved: usd(0)}, nil)
	before := f.engine.called()
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if live.Quarantined {
		t.Fatal("the park was written down as a quarantine: it is neither corruption nor a misread")
	}
	if live.Position.LastSeq != 2 {
		t.Fatalf("cursor at seq %d, want it parked on our own last line", live.Position.LastSeq)
	}
	// The DELTA the sweep itself caused, not the running total: the spawn above already asked the
	// engine once (it reads the book's meter), so a total can never be zero and an assertion on it
	// asserts nothing.
	if asked := f.engine.called() - before; asked == 0 {
		t.Fatal("the sweep did not ask the repair channel: a parked attempt has no other source of freshness")
	}
	eta := f.runETA(t, run.ID)
	if eta == nil || *eta != 42 {
		got := "nil"
		if eta != nil {
			got = fmt.Sprint(*eta)
		}
		t.Fatalf("the run's eta is %s, want the repair channel's 42 — the park left the screen frozen", got)
	}
	// A SECOND sweep inside the same resync interval: the park is still there, and the operator's line
	// is said once for the crossing and not again — the cadence is part of the signal.
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if said := strings.Count(log.String(), "another stream id"); said != 1 {
		t.Fatalf("the parked attempt was announced %d times over two sweeps, want once — at the crossing:\n%s", said, log.String())
	}
}

// The parked attempt's WARN is the only signal it has, and its cadence is part of the signal. Said on
// every pass it is four lines a minute for the life of a park that can last the run's — the line an
// operator filters, which this package already says of the deferral. Said once it is invisible to
// anyone who starts watching afterwards, and unlike a deferral the park is written down NOWHERE.
// So: at the crossing, then no more often than the repair channel itself speaks.
//
// Mutation caught: dropping the throttle (a line per sweep); making it once-only (silence after the
// interval); keying the throttle on anything but the attempt (one attempt silencing another).
func TestTheParkedAttemptSaysItselfOnceThenAtTheRepairChannelsCadence(t *testing.T) {
	svc := &Service{Cfg: Config{ResyncEvery: 5 * time.Minute}}
	now := time.Now().UTC()
	svc.Now = func() time.Time { return now }
	// Four sweeps inside one interval, at the sweep's own cadence: the crossing speaks, the rest are
	// silent. Counted by what the call ANSWERS, so an answer that never changes cannot pass.
	said := 0
	for range 4 {
		if svc.sayParked(7) {
			said++
		}
		now = now.Add(30 * time.Second)
	}
	if said != 1 {
		t.Fatalf("the park said itself %d times inside one resync interval, want once — at the crossing", said)
	}
	// A DIFFERENT attempt is a different fact: its own crossing is never swallowed by the neighbour's
	// throttle, and saying it does not silence the first one either.
	if !svc.sayParked(8) {
		t.Fatal("another attempt's park was swallowed by the first one's throttle")
	}
	if svc.sayParked(7) {
		t.Fatal("the first attempt spoke again inside its interval: the throttle is not keyed on the attempt")
	}
	// Past the interval it speaks again: a park that outlives the operator's attention has to be
	// findable by someone who starts watching now.
	now = now.Add(5*time.Minute + time.Second)
	if !svc.sayParked(7) {
		t.Fatal("the parked attempt fell silent for good: nothing else records the state")
	}
}

// A parked attempt has a row, a gauge and a cell (PD-438). Before them the state had a name
// (ErrForeignStreamAhead) and a throttled WARN, so an operator reading `tmplatformctl runs` saw a run
// whose numbers had stopped moving and an empty quarantine cell.
//
// A column is affordable because the park is not a lifecycle state anyone has to clear: it is
// re-derived from (cursor, journal) on every pass, so the sweep whose verdict changes clears it. The
// second half of this test measures exactly that.
//
// Removing the CALLER's `parked != l.Parked` threshold changes nothing observable — the write's own
// set-once guard still holds the stamp — so the last assertion asks the write directly rather than
// letting the two guards cover for each other.
func TestAParkedAttemptIsVisibleInTheRowTheGaugeAndTheListing(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	parking := hello(t, f) +
		`{"seq":2,"type":"progress","data":{"draft":{"done":3,"total":20},"eta_seconds":99}}` + "\n" +
		`{"seq":1,"type":"hello","data":{"stream_version":"1.1","engine_run_id":"SOMEONE-ELSE","book_id":"b"}}` + "\n"
	if err := os.WriteFile(journal, []byte(parking), 0o600); err != nil {
		t.Fatal(err)
	}
	f.engine.set(ingest.StatusReport{TotalUnits: 20, Done: 7, ETASeconds: 42, Spend: usd(0), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// Read through the operator's own listing, not the column: a column nothing selects is not a
	// signal.
	row := stalledRow(t, f, run.ID)
	if row.ParkedAt == nil {
		t.Fatal("the parked attempt carries no ParkedAt: the operator's table shows it exactly as a healthy run")
	}
	if row.QuarantineReason != "" {
		t.Fatal("the park was recorded as a quarantine, which offers a lift that would refuse")
	}
	stamped := *row.ParkedAt
	// Its own series: a park and a quarantine freeze a run's figures the same way and take opposite
	// actions.
	o, err := f.store.Observe(f.ctx, StalledAfter)
	if err != nil {
		t.Fatal(err)
	}
	if o.ParkedAttempts != 1 || o.QuarantinedAttempts != 0 {
		t.Fatalf("parked=%d quarantined=%d, want the park counted once and in its own series",
			o.ParkedAttempts, o.QuarantinedAttempts)
	}
	// A second sweep still parks — and must not move the stamp, or «since when» becomes «as of now».
	f.svc.Now = func() time.Time { return f.now.Add(time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if again := stalledRow(t, f, run.ID); again.ParkedAt == nil || !again.ParkedAt.Equal(stamped) {
		t.Fatalf("the park stamp moved from %v to %v across a pass that changed nothing", stamped, again.ParkedAt)
	}
	// The same property asked of the write itself: two guards stand between a sweep and a moved stamp
	// and either alone keeps the assertion above green. The caller's is a cost guard; the property
	// lives here.
	if err := f.store.MarkParked(f.ctx, f.live(t).AttemptID, true, f.now.Add(2*time.Hour)); err != nil {
		t.Fatal(err)
	}
	if direct := stalledRow(t, f, run.ID); direct.ParkedAt == nil || !direct.ParkedAt.Equal(stamped) {
		t.Fatalf("a second mark moved the stamp from %v to %v: «parked since» would then mean «parked as of "+
			"the last sweep», which is a fact nobody needs and the operator's actual question unanswered",
			stamped, direct.ParkedAt)
	}
	// The stream resumes under the id the platform gave it, so the pass that stops deriving the park is
	// the pass that clears it.
	if err := os.WriteFile(journal, []byte(hello(t, f)+
		`{"seq":2,"type":"progress","data":{"draft":{"done":4,"total":20},"eta_seconds":90}}`+"\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := f.store.Pool().Exec(f.ctx,
		`update run_attempts set last_offset = 0, last_seq = 0, last_line_sha256 = '' where run_id = $1 and ended_at is null`,
		run.ID); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if cleared := stalledRow(t, f, run.ID); cleared.ParkedAt != nil {
		t.Fatalf("the attempt is materializing again and the row still says parked since %v: "+
			"a projection that only ever goes ON is a state somebody has to lift", cleared.ParkedAt)
	}
}

// stalledRow is the operator's view of one run: the same call `tmplatformctl runs` makes.
func stalledRow(t *testing.T, f *fixture, runID string) pgstore.StalledRun {
	t.Helper()
	list, err := f.store.StalledRuns(f.ctx, 0)
	if err != nil {
		t.Fatal(err)
	}
	for _, r := range list {
		if r.RunID == runID && !r.Settling {
			return r
		}
	}
	t.Fatalf("run %s is not in the operator's live listing at all", runID)
	return pgstore.StalledRun{}
}

// The fixture's priced shapes, named rather than repeated. See newFixture for why each is what it is.
const (
	fixtureChapterUSD = money.MicroUSD(30_000) // $0.03 — the old constant, kept so figures stay legible
	fixtureStepMaxUSD = money.MicroUSD(69_828) // the editor reservation the engine refused on 04.09
)

// order is a chapter order of n chapters, as the service now takes one.
func order(n int) *int { return &n }

// fixtureHold is what the fixture's own model reserves for an order of n of its chapters — written
// as the arithmetic rather than as a literal, so a test that asserts money says WHY it expects the
// figure it expects.
func fixtureHold(n int) money.MicroUSD { return fixtureHoldAt(pricing.DefaultHoldFactorPercent, n) }

// fixtureHoldAt is the same arithmetic under a DIFFERENT cushion — what a re-pricing at today's
// setting would have reserved. Used only in failure messages, where naming the number a defect would
// have produced is what makes the assertion legible.
func fixtureHoldAt(percent, n int) money.MicroUSD {
	return fixtureChapterUSD*money.MicroUSD(n)*money.MicroUSD(percent)/100 + fixtureStepMaxUSD
}

// multiUnitBook is a priced book whose chapters hold MORE THAN ONE unit, and it exists because
// nothing else in this package did.
//
// ⛔ EVERY OTHER FIXTURE HERE WRITES `units_total = 1`, so chapters and units were the same number in
// every test of this zone — and an order, a hold and `--max-units` all live in UNITS while the bar
// and the delivered counters live in CHAPTERS. A whole axis of this pack could not go red: swapping
// the unit of measure in `maxUnitsFor` left the battery green. A fixture where the two numbers
// DIFFER is what makes the difference assertable at all.
//
// Chapter n holds `unitsPer` units; every unit is priced alike so the arithmetic stays legible.
func multiUnitBook(t *testing.T, f *fixture, id string, chapters, unitsPer int) string {
	t.Helper()
	book, err := f.store.AddBook(f.ctx, pgstore.NewBook{OwnerID: "u1", Title: "多单元", SourceLang: "zh",
		TargetLang: "ru", ChapterCount: chapters, Workdir: t.TempDir(), Now: f.now})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.store.Pool().Exec(f.ctx, `
		insert into chapters (id, book_id, number, units_total)
		select $1 || ':c' || g, $1, g, $2 from generate_series(1, $3) g`,
		book, unitsPer, chapters); err != nil {
		t.Fatal(err)
	}
	if _, err := f.store.Pool().Exec(f.ctx, `
		insert into units (id, chapter_id, ordinal, source, target, state, source_chars, expected_micro_usd)
		select $1 || ':u' || c || '-' || u, $1 || ':c' || c, u - 1, 'src', '', 'pending', 1000, $2
		  from generate_series(1, $3) c, generate_series(1, $4) u`,
		book, int64(fixtureChapterUSD), chapters, unitsPer); err != nil {
		t.Fatal(err)
	}
	if _, err := f.store.Pool().Exec(f.ctx, `
		update books set expected_micro_usd = $2, book_once_micro_usd = 0, step_max_micro_usd = $3,
		                 source_chars = $4, structure = 'detected'
		 where id = $1`,
		book, int64(fixtureChapterUSD)*int64(chapters*unitsPer), int64(fixtureStepMaxUSD),
		int64(chapters*unitsPer)*1000); err != nil {
		t.Fatal(err)
	}
	return book
}

// ⛔ THE BASIS IS READ FROM THE ENDING THIS PASS JUST DECIDED, not from the status the run had while
// it was still going. `finish` computes the outcome, writes it, and settles in the same call — and on
// a pre-finish snapshot `settlementBasis` sees `translating`, which is not one of the endings the
// engine chooses, so every clean finish was labelled as a cut-off one (PD-441).
//
// That is the most common ending there is, so the label was wrong on the majority of settlements: the
// ledger said «this figure is short because the attempt was cut off mid-work» about runs that were
// not cut off at all.
//
// Asserted on the ledger ROW, not on the classifier: the classifier was already right, and what was
// wrong was what reached it.
func TestACleanEndingIsSettledOnTheBasisOfTheEndingItReached(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo),
		runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 100, Spend: usd(money.MicroUSD(1_234_567)), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	var note string
	if err := f.store.Pool().QueryRow(f.ctx,
		`select note from credit_ledger where kind = 'settlement' and source_id = $1`,
		pgstore.ReservationKey(run.ID, live.AttemptNo)).Scan(&note); err != nil {
		t.Fatal(err)
	}
	if strings.Contains(note, "cut off mid-work") {
		t.Fatalf("a run that ended `ready` was settled as cut off mid-work: %q", note)
	}
	if !strings.Contains(note, "at least") {
		t.Fatalf("the settlement of a clean ending says nothing about being a floor: %q", note)
	}
}