textmachine/platform/internal/runs/control_test.go

package runs

import (
	"errors"
	"os"
	"path/filepath"
	"slices"
	"sync"
	"testing"
	"time"

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

// The control handles — stop and resume — are the two places where a USER's decision reaches a run
// that is not this process's child. What they have to get right is not the systemd call, which is
// one line, but the record around it: a stop that is not written down before the signal cannot be
// told from a crash afterwards, and a resume that forgets the previous attempt's hold reserves the
// run's ceiling twice.

// stopped drives a fixture to a run the user stopped: admitted, spawned, asked to stop, and then
// reconciled from the marker the unit left. It is the state resume starts from.
//
// spent is what the engine reports AFTER the spawn, which is the only order that models the real
// one: the spawn reads the book's meter to take this attempt's baseline, so a figure set before it
// would be read as money some earlier run had spent and this attempt would settle at nothing.
func (f *fixture) stopped(t *testing.T, chapters int, spent money.MicroUSD, marker runner.Marker) string {
	t.Helper()
	return f.stoppedRun(t, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: chapters}, spent, marker)
}

// stoppedRun is the same walk for a run whose START matters — a signing run, say.
func (f *fixture) stoppedRun(t *testing.T, in StartRequest, spent money.MicroUSD, marker runner.Marker) string {
	t.Helper()
	run, err := f.svc.Start(f.ctx, in)
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	marker.Unit = live.UnitName
	if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo), marker); err != nil {
		t.Fatal(err)
	}
	f.engine.set(statusSpending(spent), nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	return run.ID
}

// A run is REFUSED over a book whose chapter tree was never materialised, and refused BEFORE its
// hold — which is the whole point of where the guard sits (PD-405).
//
// The window is the intake's own and it is ordinary: `not_started` is committed in one transaction
// and the tree is materialised after it, outside. On the healthy path that is milliseconds (measured
// at 0.018 s, act D39.162); when the materialisation FAILED or was deferred it is unbounded, and that
// is the shape that costs money. A run admitted there can never move: every counter the screen shows
// is a count over `chapters` and there are no rows to count, so the bar reads 0/total for the run's
// whole life — and the tree's own debt is frozen meanwhile, because the sweep that would pay it skips
// a book with a live run. The user pays for a run that shows nothing and cannot be repaired.
//
// Mutation caught: moving the guard below the hold; dropping the `ChapterCount > 0` term (which would
// refuse a book that legitimately declares no chapters).
func TestARunIsRefusedOverABookWhoseChaptersWereNeverMaterialised(t *testing.T) {
	f := newFixture(t, "10", 500)
	book := f.bookID(t)
	// The tree goes away: the state the intake is in between its commit and its materialisation, and
	// the state it STAYS in when that materialisation broke.
	if _, err := f.store.Pool().Exec(f.ctx, `delete from chapters where book_id = $1`, book); err != nil {
		t.Fatal(err)
	}
	before := f.account(t)
	_, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, CeilingChapters: 10})
	if !errors.Is(err, ErrBookNotReady) {
		t.Fatalf("a run over a book with no chapter tree was answered %v, want book_not_ready: its bar"+
			" can never move and its tree's debt is frozen for as long as it runs", err)
	}
	// BEFORE the hold, and this is the half that is money rather than honesty.
	after := f.account(t)
	if after.Balance != before.Balance || after.LedgerSum != before.LedgerSum {
		t.Errorf("the refusal moved money: balance %s → %s, ledger %s → %s",
			before.Balance.USD(), after.Balance.USD(), before.LedgerSum.USD(), after.LedgerSum.USD())
	}
	// And the same book with its tree back is startable: the guard refuses the STATE, not the book.
	if _, err := f.store.Pool().Exec(f.ctx, `
		insert into chapters (id, book_id, number, units_total)
		select 'c' || g, $1, g, 1 from generate_series(1, 500) g`, book); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, CeilingChapters: 10}); err != nil {
		t.Fatalf("the book was refused with its tree materialised: %v", err)
	}
}

// The OTHER half of the same guard, and it exists because this pack's adversarial pass showed the
// header above claiming a mutation nothing caught: dropping the `ChapterCount > 0` term left the
// whole eighteen-package battery green, because no test built the book it would newly refuse.
//
// A book that declares NO chapters has no tree to be missing, so the guard must not touch it. It is
// refused, but by the thing that should refuse it — there is nothing to buy — and the two refusals
// carry different remedies: "wait for the intake" is a lie told about a book whose intake is over.
//
// Mutation caught: dropping the `ChapterCount > 0` term from the admission guard.
func TestABookThatDeclaresNoChaptersIsNotRefusedAsUnmaterialised(t *testing.T) {
	f := newFixture(t, "10", 0)
	book := f.bookID(t)
	var count int
	if err := f.store.Pool().QueryRow(f.ctx,
		`select chapter_count from books where id = $1`, book).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatalf("the fixture book declares %d chapters: this test is about the one that declares none", count)
	}
	_, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, CeilingChapters: 1})
	if errors.Is(err, ErrBookNotReady) {
		t.Errorf("a book that declares no chapters was told its chapters are still being materialised:"+
			" %v — its intake is OVER, and the honest refusal is that there is nothing to buy", err)
	}
	if err == nil {
		t.Error("a run was admitted over a book with nothing in it")
	}
}

// secondBook registers another book of the same account, with a directory of its own.
func (f *fixture) secondBook(t *testing.T) string {
	t.Helper()
	id, err := f.store.AddBook(f.ctx, pgstore.NewBook{OwnerID: "u1", Title: "second", SourceLang: "zh",
		TargetLang: "ru", ChapterCount: 100, Workdir: t.TempDir(), Now: f.now})
	if err != nil {
		t.Fatal(err)
	}
	// …with its chapter TREE, for the same reason the main fixture has one: a book that declares
	// chapters and materialised none cannot be started over (PD-405).
	if _, err := f.store.Pool().Exec(f.ctx, `
		insert into chapters (id, book_id, number, units_total)
		select $1 || ':c' || g, $1, g, 1 from generate_series(1, 100) g`, id); err != nil {
		t.Fatal(err)
	}
	return id
}

// lastAttempt is the snapshot a slow sweep pass would still be holding: the run as it was, with the
// attempt that was live at the time.
func (f *fixture) lastAttempt(t *testing.T, runID string) pgstore.LiveRun {
	t.Helper()
	l, err := f.store.ReadRunForResume(f.ctx, "u1", runID)
	if err != nil {
		t.Fatal(err)
	}
	return l
}

func (f *fixture) run(t *testing.T, runID string) pgstore.Run {
	t.Helper()
	r, err := f.store.ReadRun(f.ctx, "u1", runID)
	if err != nil {
		t.Fatal(err)
	}
	return r
}

// The heart of PD-152. The engine catches SIGTERM and exits 1, so the marker of a stop and the
// marker of a crash are the same bytes; what tells them apart is that the platform wrote down its
// own intent first.
func TestARunTheUserStoppedIsNotReportedAsFailed(t *testing.T) {
	f := newFixture(t, "10", 500)
	runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
		At: f.now.Add(time.Second)})
	if got := f.run(t, runID); got.Status != "stopped" {
		t.Fatalf("a stop the user asked for came back as %q, want stopped", got.Status)
	}
	// And the same marker WITHOUT the intent is still a failure: the discriminator has to be the
	// record, not a new reading of the exit code.
	g := newFixture(t, "10", 500)
	run, err := g.svc.Start(g.ctx, StartRequest{UserID: "u1", BookID: g.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	if err := g.svc.Spawn(g.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	live := g.live(t)
	if err := runner.WriteMarker(g.svc.markerPath(live.RunID, live.AttemptNo),
		runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "1"}); err != nil {
		t.Fatal(err)
	}
	if err := g.svc.Sweep(g.ctx); err != nil {
		t.Fatal(err)
	}
	if got := g.run(t, run.ID); got.Status != "failed" {
		t.Fatalf("a run nobody stopped came back as %q, want failed", got.Status)
	}
}

// The race the design owes an answer to: a run that finished by itself a moment before someone
// pressed stop was not stopped by them, and calling it `stopped` would hide a completed translation
// behind a cancelled one.
func TestARunThatEndedBeforeTheStopKeepsItsOwnOutcome(t *testing.T) {
	f := newFixture(t, "10", 500)
	// The marker is written a minute BEFORE the stop request the fixture makes.
	runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
		At: f.now.Add(-time.Minute)})
	if got := f.run(t, runID); got.Status != "failed" {
		t.Fatalf("a run that had already ended came back as %q, want failed", got.Status)
	}
}

// The engine's own clean answers outrank a stop that arrived while it was already finishing: a
// translation that COMPLETED must not be shown as cancelled.
func TestACleanExitOutranksAStopThatArrivedTooLate(t *testing.T) {
	f := newFixture(t, "10", 500)
	runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "0",
		At: f.now.Add(time.Second)})
	if got := f.run(t, runID); got.Status != "ready" {
		t.Fatalf("a run that finished cleanly came back as %q, want ready", got.Status)
	}
}

// The intent is committed BEFORE systemd is asked, and that order is the whole mechanism: a platform
// that died between the two must still know, on its next sweep, that the run was stopped on purpose.
func TestTheStopIsRecordedBeforeSystemdIsAsked(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	var recordedFirst bool
	f.runner.onStop = func(string) {
		var at *time.Time
		if err := f.store.Pool().QueryRow(f.ctx,
			`select stop_requested_at from runs where id = $1`, run.ID).Scan(&at); err != nil {
			t.Error(err)
			return
		}
		recordedFirst = at != nil
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	if !recordedFirst {
		t.Fatal("systemd was asked to stop the unit before the intent was committed")
	}
}

// A stop that systemd did not take must not be lost. The intent is durable, so the next sweep asks
// again — which is the only thing that closes "the platform died between the write and the call".
func TestAStopTheUnitDidNotTakeIsAskedAgainByTheSweep(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.stops()); n != 2 {
		t.Fatalf("the unit was asked to stop %d times, want 2 (the handle and the sweep)", n)
	}
}

// The dangerous half: a stopped unit that left no marker looks exactly like a reboot, and the reboot
// path RESTARTS the run — which would spend the account's money on work its owner had just
// cancelled.
func TestAStoppedRunIsNotRestartedWhenItsMarkerIsMissing(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	// The unit is gone, no marker was written, and the spawn grace is long past.
	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)
	}
	if got := f.run(t, run.ID); got.Status != "stopped" {
		t.Fatalf("run is %q, want stopped", got.Status)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units were started; a stopped run must not be restarted", n)
	}
	if acct := f.account(t); acct.Reserved != 0 {
		t.Fatalf("the hold of a stopped run is still open: %s", acct.Reserved.USD())
	}
}

// A run stopped before its unit ever existed: there is nothing to signal, and the money must come
// back whole rather than wait for a unit that will never be created.
func TestStoppingARunThatNeverSpawnedGivesTheHoldBackWhole(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.stops()); n != 0 {
		t.Fatalf("systemd was asked about a unit that does not exist (%d times)", n)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.run(t, run.ID); got.Status != "stopped" {
		t.Fatalf("run is %q, want stopped", got.Status)
	}
	acct := f.account(t)
	if acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000) {
		t.Fatalf("the hold of a run that never started did not come back whole: balance %s reserved %s",
			acct.Balance.USD(), acct.Reserved.USD())
	}
	if n := len(f.runner.starts()); n != 0 {
		t.Fatalf("a stopped run was spawned anyway (%d units)", n)
	}
}

// Ownership, on the handle that ENDS a paid run: another account's run is not "forbidden", it is not
// there at all (API1/BOLA, and the same rule the library reads by).
func TestARunOfAnotherAccountCannotBeStoppedOrResumed(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.store.Pool().Exec(f.ctx,
		`insert into users (id, email) values ('u2','u2@example.org')`); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u2", run.ID); !errors.Is(err, pgstore.ErrNoRun) {
		t.Fatalf("stop by a stranger: %v, want ErrNoRun", err)
	}
	if _, err := f.svc.Resume(f.ctx, "u2", run.ID); !errors.Is(err, pgstore.ErrNoRun) {
		t.Fatalf("resume by a stranger: %v, want ErrNoRun", err)
	}
}

// Stopping a run that is already over is a conflict and NOT a not-found: the owner can see the run,
// and answering 404 would tell them their own run does not exist.
func TestStoppingARunThatIsOverIsAConflict(t *testing.T) {
	f := newFixture(t, "10", 500)
	runID := f.stopped(t, 10, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
		At: f.now.Add(time.Second)})
	if _, err := f.svc.Stop(f.ctx, "u1", runID); !errors.Is(err, ErrNotStoppable) {
		t.Fatalf("stopping a finished run: %v, want ErrNotStoppable", err)
	}
}

// Resume is the reconciler's restart with a user behind it: a NEW attempt, holding what is LEFT of
// the run's budget. The money is the assertion — one run may not reserve its ceiling twice.
func TestResumeContinuesTheRunWithWhatIsLeftOfItsBudget(t *testing.T) {
	f := newFixture(t, "10", 500)
	spent := money.MicroUSD(500_000)
	runID := f.stopped(t, 100, spent, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
		At: f.now.Add(time.Second)})
	budget := f.svc.Pricing.Ceiling(100)
	if acct := f.account(t); acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000)-spent {
		t.Fatalf("before the resume: balance %s reserved %s", acct.Balance.USD(), acct.Reserved.USD())
	}
	got, err := f.svc.Resume(f.ctx, "u1", runID)
	if err != nil {
		t.Fatal(err)
	}
	if got.Status != "translating" || got.FinishedAt != nil {
		t.Fatalf("the resumed run is %+v, want a live translating run", got)
	}
	// The new attempt holds the REMAINDER, not the whole ceiling: the first attempt already spent
	// part of it and settled at what it spent.
	if acct := f.account(t); acct.Reserved != budget-spent {
		t.Fatalf("the resumed run holds %s, want %s (the ceiling less what was already spent)",
			acct.Reserved.USD(), (budget - spent).USD())
	}
	// The engine is QUEUED, not started inside the request: reading the book's meter costs seconds of
	// the engine's CPU and creating a unit is a round trip to systemd, and a call that answers 202
	// must not hold either. (Measured, not reasoned about: the live probe's resume held its request
	// until the engine exited.) With no queue wired here, the reconciler is the backstop — which is
	// the same backstop the admission path relies on.
	live := f.live(t)
	if live.AttemptNo != 2 || live.UnitName != "" {
		t.Fatalf("the resumed run's attempt is %+v, want a second attempt waiting to be spawned", live)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started; the resume spawned the engine inside the request", n)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.starts()); n != 2 {
		t.Fatalf("%d units started across the run, want 2 once the backstop ran", n)
	}
	// The book follows the run back into translation, and the library's revision moves with it.
	book, _, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if book.Status != "translating" {
		t.Fatalf("the book of a resumed run is %q, want translating", book.Status)
	}
}

// The bank stop is the one resume the contract describes in detail, and its condition — a COMPLETE
// set of decisions — cannot be met by any deployment today: nothing materializes the bank at all
// (companion §3). Answering 202 would start a run that walks straight into the same stop.
// Signing the bank is ONE act over the whole of it, and `resume` lifts that stop with the decisions
// AS THEY STAND (owner 16.08, D39.144). This test replaces the one that pinned the opposite: the
// gate it protected — "the stop clears only on a complete set of decisions" — was an invention of
// the contract line, and the engine has always continued with an unsigned bank by default.
//
// Mutation caught: restoring the completeness gate in Resume; treating `awaiting_bank` as a status
// that falls through to the default 409.
func TestResumeLiftsABankStopWithTheDecisionsAsTheyStand(t *testing.T) {
	f := newFixture(t, "10", 500)
	runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "3",
		At: f.now.Add(time.Second)})
	if got := f.run(t, runID); got.Status != "awaiting_bank" {
		t.Fatalf("run after the bank stop is %q", got.Status)
	}
	// No decision has been recorded at all, which is the state every book is in until somebody opens
	// the screen — and the one the old gate refused.
	got, err := f.svc.Resume(f.ctx, "u1", runID)
	if err != nil {
		t.Fatalf("resume of a bank stop: %v, want it continued", err)
	}
	if got.Status != "translating" {
		t.Fatalf("the resumed run is %q, want it going again", got.Status)
	}
}

// A run that spent the whole ceiling it bought is REFUSED: there is no work this call could pay for,
// so the remedy is a new run. Nothing moves — no second hold, no second unit — and the run's own
// state is not rewritten on the way out: a run the user stopped stays stopped.
//
// ⚠ It used to answer 202 with the run unchanged, which is a success a client cannot tell from a
// continuation (PD-282). The canon settled it the other way and now says the opposite in as many
// words: `ceiling_reached` in EVERY status, and a 202 means work was actually re-opened.
func TestResumeOfARunWithNothingLeftIsRefusedWithTheCeilingReached(t *testing.T) {
	f := newFixture(t, "10", 500)
	// The attempt spends its entire ceiling, so the run has no budget left at all.
	runID := f.stopped(t, 10, f.svc.Pricing.Ceiling(10), runner.Marker{Result: "exit-code",
		Code: "exited", Status: "1", At: f.now.Add(time.Second)})
	before := f.account(t)
	_, err := f.svc.Resume(f.ctx, "u1", runID)
	if !errors.Is(err, ErrCeilingReached) {
		t.Fatalf("resume of a run with nothing left: %v, want ErrCeilingReached", err)
	}
	if got := f.run(t, runID); got.Status != "stopped" || got.PausedReason != "" {
		t.Fatalf("the refusal rewrote the run to %q/%q, want the stopped run it was", got.Status, got.PausedReason)
	}
	if acct := f.account(t); acct.Reserved != 0 || acct.Balance != before.Balance {
		t.Fatalf("a resume that changed nothing moved money: %+v -> %+v", before, acct)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started, want 1: nothing was resumed", n)
	}
}

// The OTHER refusal, and the whole reason the two are told apart: this run has chapters left in the
// ceiling it bought and the ACCOUNT cannot cover them. The remedy is money — after which this same
// call continues this same run — so answering `ceiling_reached` would send the user off to buy a run
// they do not need.
//
// Mutation caught: folding the two verdicts back into one.
func TestResumeOverAnEmptyAccountSaysTheCreditIsUnavailableAndNotTheCeiling(t *testing.T) {
	f := newFixture(t, "10", 500)
	// Half the ceiling spent, so the RUN has room left; then the account is emptied under it.
	runID := f.stopped(t, 10, f.svc.Pricing.Ceiling(5), runner.Marker{Result: "exit-code",
		Code: "exited", Status: "1", At: f.now.Add(time.Second)})
	if _, err := f.store.Adjust(f.ctx, "u1", -f.account(t).Balance, "test", "drain", "spent", f.now); err != nil {
		t.Fatal(err)
	}
	_, err := f.svc.Resume(f.ctx, "u1", runID)
	if errors.Is(err, ErrCeilingReached) {
		t.Fatalf("a run with room left was refused as if its ceiling were spent: %v", err)
	}
	if !errors.Is(err, ErrCreditUnavailable) {
		t.Fatalf("resume over an empty account: %v, want ErrCreditUnavailable", err)
	}
	// …and topping the account up is all it takes: the same call then continues the same run.
	if _, err := f.store.Grant(f.ctx, "u1", money.MicroUSD(5_000_000), "test", "topup", "", f.now); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Resume(f.ctx, "u1", runID); err != nil {
		t.Fatalf("a topped-up account still could not continue the run: %v", err)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started: the resume is queued, not spawned inline", n)
	}
	if got := f.run(t, runID); got.Status != "translating" {
		t.Errorf("the resumed run is %q, want it going again", got.Status)
	}
}

// The reconciler's half of the same decision, which the shared body must not have taken away: a run
// that was LIVE and has nothing left is `paused` with the reason the contract has a word for, and a
// resume of that comes back paused too.
func TestAnInterruptedRunWithNothingLeftIsPausedAndStaysPausedThroughAResume(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 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 and then its unit vanished without a marker, which is what a
	// reboot leaves behind.
	f.engine.set(statusSpending(f.svc.Pricing.Ceiling(10)), 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)
	}
	got := f.run(t, run.ID)
	if got.Status != "paused" || got.PausedReason != pgstore.PausedCreditExhausted {
		t.Fatalf("an interrupted run with no budget is %q/%q, want paused/credit_exhausted",
			got.Status, got.PausedReason)
	}
	// ⚠ 409 and NOT a silent 202 with the run unchanged. The limit travels with the START of a run
	// and nothing changes it afterwards, so this call can never move such a run — and an answer a
	// client cannot tell from a real continuation is the defect 0.3.0 named (canon §resumeRun). The
	// remedy is a NEW run with a larger ceiling, which is legal from any paused book.
	if _, err := f.svc.Resume(f.ctx, "u1", run.ID); !errors.Is(err, ErrCeilingReached) {
		t.Fatalf("resume of a run stopped at a limit: %v, want ErrCeilingReached", err)
	}
	if got := f.run(t, run.ID); got.Status != "paused" || got.PausedReason != pgstore.PausedCreditExhausted {
		t.Fatalf("the refused resume rewrote the run to %q/%q", got.Status, got.PausedReason)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started, want 1", n)
	}
}

// A settlement is allowed to defer, and a resume that ignored that would open a SECOND hold on a run
// whose first is still reserved.
func TestResumeRefusesWhileThePreviousAttemptIsStillHoldingMoney(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", 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: "1",
			At: f.now.Add(time.Second)}); err != nil {
		t.Fatal(err)
	}
	// The engine cannot be asked what it spent, so the run finishes and its hold stays open.
	f.engine.set(statusSpending(0), errors.New("status unavailable"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if acct := f.account(t); acct.Reserved == 0 {
		t.Fatal("the hold was released even though the engine could not be asked")
	}
	if _, err := f.svc.Resume(f.ctx, "u1", run.ID); !errors.Is(err, ErrNotResumable) {
		t.Fatalf("resume while unsettled: %v, want ErrNotResumable", err)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started: an unsettled run must not be resumed", n)
	}
}

// A book that is still arriving, or one the engine refused, has no chapter tree to translate — and
// while it is `uploading` it has half a file on disk. The refusal happens BEFORE the money moves.
func TestARunIsRefusedOnABookThatIsNotThroughIntake(t *testing.T) {
	f := newFixture(t, "10", 500)
	for _, status := range []string{"uploading", "parsing", "rejected"} {
		if _, err := f.store.Pool().Exec(f.ctx,
			`update books set status = $1 where id = $2`, status, f.bookID(t)); err != nil {
			t.Fatal(err)
		}
		_, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
		if !errors.Is(err, ErrBookNotReady) {
			t.Fatalf("starting a run on a %s book: %v, want ErrBookNotReady", status, err)
		}
		if acct := f.account(t); acct.Reserved != 0 {
			t.Fatalf("a refused run held money: %s", acct.Reserved.USD())
		}
	}
}

// statusSpending is a report of an engine that has committed exactly this much on the book.
func statusSpending(v money.MicroUSD) ingest.StatusReport {
	return ingest.StatusReport{Spend: usd(v), Reserved: usd(0)}
}

// PD-169. A pass of the sweep has one budget for every run in it, and the list is ordered the same
// way every time: a run whose engine hangs used to eat the pass, and the tail of the list — other
// people's books — was never reached at all. The per-run budget is what bounds that, and this is the
// assertion that it does.
func TestOneSlowRunDoesNotEatThePassOfTheWholeSweep(t *testing.T) {
	f := newFixture(t, "20", 500)
	f.svc.Cfg.RunBudget = 100 * time.Millisecond
	// Two books, two live runs, and the sweep visits them in the order they started.
	first, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	second := f.secondBook(t)
	if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: second, CeilingChapters: 10}); err != nil {
		t.Fatal(err)
	}
	// The engine hangs for the FIRST run only — until the context that carries it is cancelled.
	f.engine.block(first.ID, f.workdir)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// The second run was spawned in the same pass, which is the whole property: the slow one cost its
	// own budget and nothing else's.
	started := f.runner.starts()
	if len(started) != 1 {
		t.Fatalf("%d units started; the second run was starved by the first", len(started))
	}
}

// A run standing at the bank-signing stop is NOT an interruption, so the sweep does not restart it —
// it closes it AT the stop, and only the user's resume takes it further.
//
// ⚠ RE-SIGNED by pack P12 (unified-backlog row 240), and the re-signing is the point rather than an
// edit to keep a test green. This test used to pin a guard that keyed on WHICH BIT a restart wrote
// (`LiftBankStop`), and that guard was written against an engine that re-halted on everything still
// undecided: keeping `--verify-bank` on the restarted attempt was what re-created the pause. The
// engine's presented memory (storage v16, D39.158) ended that — a map whose clusters were already
// shown does not stop again, whatever the flag says — so from that day the guard protected NOTHING:
// the sweep's restart marched straight through the paid pause with one WARN in the engine's own
// journal. What holds instead is the decision one level up, and it is the same rule the ceiling
// branch beside it already follows: an engine that ended ON PURPOSE is not restarted.
//
// Mutation caught: restarting a run whose status is `awaiting_bank` when its unit vanishes.
func TestTheReconcilerDoesNotRestartPastABankStopNobodySigned(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), VerifyBank: true,
		CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	// What the sink writes when the engine reports the bank stop: the status moves and the run stays
	// LIVE, because the attempt has not ended yet (pgstore/sink.go, TypeBankStop).
	if _, err := f.store.Pool().Exec(f.ctx,
		`update runs set status = 'awaiting_bank' where id = $1`, run.ID); err != nil {
		t.Fatal(err)
	}
	// …and then the unit dies leaving no marker — a reboot, or a SIGTERM from outside. The SERVICE's
	// clock is what the grace is measured on, and it is not f.now.
	f.runner.alive = false
	late := f.now.Add(2 * time.Hour)
	f.svc.Now = func() time.Time { return late }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	var status string
	var attempts int
	var finished *time.Time
	if err := f.store.Pool().QueryRow(f.ctx,
		`select r.status, r.finished_at, (select count(*) from run_attempts where run_id = r.id)
		   from runs r where r.id = $1`, run.ID).Scan(&status, &finished, &attempts); err != nil {
		t.Fatal(err)
	}
	// The sweep must have LOOKED at this run, or the test observes nothing: the same guard against a
	// fixture that never reaches the decision which caught the first version of this test.
	if !f.runner.askedAlive() {
		t.Fatal("the sweep never reached this run: the fixture cannot observe the property")
	}
	if attempts != 1 {
		t.Errorf("the sweep restarted a run standing at the signing stop (%d attempts): the pause the user paid for was spent on a march through it", attempts)
	}
	if status != "awaiting_bank" {
		t.Errorf("the run left the signing stop as %q, want it still standing there for the user to answer", status)
	}
	if finished == nil {
		t.Error("the run was left LIVE at a stop whose engine is gone: nothing ends it and nothing resumes it")
	}
	// Whatever the sweep did spawn, it carried the flag: after D39.158 the platform never masks it,
	// and where the engine stops is the engine's decision.
	for i, spec := range f.runner.starts() {
		if !slices.Contains(spec.Args, "--verify-bank") {
			t.Errorf("attempt %d was spawned without the signing stop: %v", i+1, spec.Args)
		}
	}
}

// THE PAID STOP SURVIVES THE TWO PATHS THAT REACH IT THROUGH THIS SWEEP'S OWN DRAIN, and both were
// open until this pack's adversarial pass measured them.
//
// The trap they share: the sweep's `LiveRun` is a by-value snapshot taken BEFORE the journal is
// drained, and the bank-stop event lands in that very drain — it writes `awaiting_bank` on the row
// and nothing the snapshot carries. A guard that asked the snapshot therefore asked a value that
// could not yet know, and the run was restarted past the pause the user paid for. After D39.158 that
// restart does not even re-stop: it carries `--verify-bank` and the engine's presented memory
// auto-continues over a map it has already shown, one WARN in the engine's own journal.
//
// Sub-test A — the unit VANISHED, and the stop arrived in this pass's drain.
// Sub-test B — a marker exists reading exit 5, an outside SIGTERM. That branch (`interruptedBySomeoneElse`)
// answers BEFORE `outcome` is ever consulted, so the rule written into `outcome` could not see it.
//
// Mutation caught: reading l.Status from the pre-drain snapshot; dropping the awaiting_bank exclusion
// from the interrupted-by-someone-else branch.
func TestABankStopArrivingInThisSweepsOwnDrainIsNotRestartedPast(t *testing.T) {
	bankStop := func(t *testing.T, f *fixture) {
		t.Helper()
		body := hello(t, f) + `{"seq":2,"type":"bank_stop","data":{}}` + "\n"
		if err := os.WriteFile(filepath.Join(f.workdir, ingest.JournalFile), []byte(body), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	start := func(t *testing.T) (*fixture, string) {
		t.Helper()
		f := newFixture(t, "10", 500)
		run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), VerifyBank: true,
			CeilingChapters: 10})
		if err != nil {
			t.Fatal(err)
		}
		if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
			t.Fatal(err)
		}
		bankStop(t, f)
		return f, run.ID
	}
	assertStandsAtTheStop := func(t *testing.T, f *fixture, runID string) {
		t.Helper()
		var status string
		var attempts int
		if err := f.store.Pool().QueryRow(f.ctx,
			`select r.status, (select count(*) from run_attempts where run_id = r.id)
			   from runs r where r.id = $1`, runID).Scan(&status, &attempts); err != nil {
			t.Fatal(err)
		}
		if attempts != 1 {
			t.Errorf("the sweep opened attempt %d over a run standing at the signing stop: the pause the"+
				" user paid for was spent on a march through it", attempts)
		}
		if status != "awaiting_bank" {
			t.Errorf("the run left the signing stop as %q, want it still standing there", status)
		}
	}

	t.Run("the unit vanished", func(t *testing.T) {
		f, runID := start(t)
		f.runner.alive = false
		late := f.now.Add(2 * time.Hour)
		f.svc.Now = func() time.Time { return late }
		if err := f.svc.Sweep(f.ctx); err != nil {
			t.Fatal(err)
		}
		assertStandsAtTheStop(t, f, runID)
	})

	t.Run("an outside signal left a marker", func(t *testing.T) {
		f, runID := start(t)
		f.runner.alive = false
		live := f.live(t)
		// Exit 5 with NO stop recorded by this platform: the shape interruptedBySomeoneElse exists for.
		if err := runner.WriteMarker(f.svc.markerPath(runID, live.AttemptNo),
			runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "5"}); err != nil {
			t.Fatal(err)
		}
		late := f.now.Add(2 * time.Hour)
		f.svc.Now = func() time.Time { return late }
		if err := f.svc.Sweep(f.ctx); err != nil {
			t.Fatal(err)
		}
		assertStandsAtTheStop(t, f, runID)
	})
}

// `blocked` answers ONE question — why is this scale shorter than the account could afford — so it
// names another book only when giving that book's hold back would lengthen the scale. A hold that
// costs this book nothing must not send the user off to stop a run for no gain.
//
// Mutation caught: filling BlockedBy from CreditHeldBy alone, which is what shipped.
func TestBlockedNamesAnotherBookOnlyWhenItsHoldShortensTheScale(t *testing.T) {
	// $20 buys far more than either book has chapters, so nothing money does can shorten this scale.
	f := newFixture(t, "20", 4)
	own := f.bookID(t)
	second := f.secondBook(t)
	if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: second, CeilingChapters: 10}); err != nil {
		t.Fatal(err)
	}
	opts, err := f.svc.Bounds(f.ctx, "u1", own)
	if err != nil {
		t.Fatal(err)
	}
	if opts.Ceiling.Max != 4 {
		t.Fatalf("the scale is %d chapters, want the book's own 4 — the fixture does not test what it claims", opts.Ceiling.Max)
	}
	if opts.BlockedBy != "" {
		t.Errorf("blocked names %q while the scale is bounded by the book's length, not by money", opts.BlockedBy)
	}
	// Now money IS the bound: a balance that buys less than the book has left, with a hold open
	// elsewhere that would have bought more.
	poor := newFixture(t, "0.3", 500)
	poorOwn := poor.bookID(t)
	other := poor.secondBook(t)
	if _, err := poor.svc.Start(poor.ctx, StartRequest{UserID: "u1", BookID: other, CeilingChapters: 5}); err != nil {
		t.Fatal(err)
	}
	opts, err = poor.svc.Bounds(poor.ctx, "u1", poorOwn)
	if err != nil {
		t.Fatal(err)
	}
	if opts.BlockedBy != other {
		t.Errorf("blocked is %q while another book's hold is what shortens the scale (max %d)",
			opts.BlockedBy, opts.Ceiling.Max)
	}
}

// Two clicks on one button. Both calls pass the state check together and race for attempt N+1 on the
// unique index; the loser must answer with the run the winner re-opened, not with a not-found — and
// exactly ONE hold may be taken.
func TestTwoResumesOfOneRunTakeOneHoldAndBothAnswer(t *testing.T) {
	f := newFixture(t, "10", 500)
	runID := f.stopped(t, 100, money.MicroUSD(500_000), runner.Marker{Result: "exit-code",
		Code: "exited", Status: "1", At: f.now.Add(time.Second)})
	var wg sync.WaitGroup
	runs := make([]pgstore.Run, 2)
	errs := make([]error, 2)
	for i := range runs {
		wg.Add(1)
		go func() {
			defer wg.Done()
			runs[i], errs[i] = f.svc.Resume(f.ctx, "u1", runID)
		}()
	}
	wg.Wait()
	for i, err := range errs {
		if err != nil {
			t.Fatalf("resume %d: %v, want the run back", i, err)
		}
		if runs[i].Status != "translating" {
			t.Errorf("resume %d answered %q, want translating", i, runs[i].Status)
		}
	}
	if acct := f.account(t); acct.Reserved != f.svc.Pricing.Ceiling(100)-money.MicroUSD(500_000) {
		t.Fatalf("two resumes reserved %s, want exactly one remainder", acct.Reserved.USD())
	}
	if acct := f.account(t); acct.Balance != acct.LedgerSum {
		t.Fatalf("the cached balance and the ledger disagree: %s vs %s", acct.Balance.USD(), acct.LedgerSum.USD())
	}
	var attempts int
	if err := f.store.Pool().QueryRow(f.ctx,
		`select count(*) from run_attempts where run_id = $1`, runID).Scan(&attempts); err != nil {
		t.Fatal(err)
	}
	if attempts != 2 {
		t.Fatalf("%d attempts after two resumes, want 2", attempts)
	}
}

// A stopped run can be resumed — unless the account started ANOTHER run of the same book in the
// meantime. Re-opening then puts two live runs under an index that forbids exactly that, and the
// answer has to be the conflict a second admission gets, not an internal error. The money is the
// other half: the hold of the refused resume is taken in the same transaction and dies with it.
func TestResumeIsRefusedWhenTheBookHasAnotherLiveRun(t *testing.T) {
	f := newFixture(t, "10", 500)
	stoppedRun := f.stopped(t, 10, money.MicroUSD(10_000), runner.Marker{Result: "exit-code",
		Code: "exited", Status: "1", At: f.now.Add(time.Second)})
	if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10}); err != nil {
		t.Fatal(err)
	}
	before := f.account(t)
	if _, err := f.svc.Resume(f.ctx, "u1", stoppedRun); !errors.Is(err, pgstore.ErrRunInFlight) {
		t.Fatalf("resume while another run is live: %v, want ErrRunInFlight", err)
	}
	acct := f.account(t)
	if acct.Reserved != before.Reserved || acct.Balance != before.Balance {
		t.Fatalf("a refused resume moved money: %+v -> %+v", before, acct)
	}
	if acct.Balance != acct.LedgerSum {
		t.Fatalf("the cached balance and the ledger disagree: %s vs %s", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// One counter per book, and every run-carrying answer uses it (contract §Revision). The stop handle
// used to answer from the run's own column, which sits BELOW the card's number the moment anything
// materializes — and a client that has applied the card's revision is required by the contract to
// drop the lower one, i.e. to drop the answer to the button it just pressed.
func TestEveryRunCarryingAnswerUsesTheBooksRevision(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	// Something materializes: the book's counter moves and the run's own column does not.
	if _, err := f.store.Pool().Exec(f.ctx,
		`update books set revision = revision + 5 where id = $1`, f.bookID(t)); err != nil {
		t.Fatal(err)
	}
	book, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil || card == nil {
		t.Fatalf("card: %v", err)
	}
	if card.Revision != book.Revision {
		t.Fatalf("the card's run carries revision %d and its book %d: one counter per book", card.Revision, book.Revision)
	}
	stopped, err := f.svc.Stop(f.ctx, "u1", run.ID)
	if err != nil {
		t.Fatal(err)
	}
	if stopped.Revision != book.Revision {
		t.Fatalf("stop answered revision %d, the card answers %d — the client drops the lower one",
			stopped.Revision, book.Revision)
	}
	read, err := f.store.ReadRun(f.ctx, "u1", run.ID)
	if err != nil {
		t.Fatal(err)
	}
	if read.Revision != book.Revision {
		t.Fatalf("ReadRun answered revision %d, the card answers %d", read.Revision, book.Revision)
	}
}

// A run that is not the caller's, or one that does not exist, is a 404 — even on a deployment that
// could not resume anything anyway. The other order answers "this service cannot start runs" to a
// question about someone else's run, which is both a worse answer and a small oracle.
func TestOwnershipIsJudgedBeforeTheDeploymentsHealth(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	f.svc.Cfg.MarkerArgv = nil // a deployment that cannot record how a run ends
	if _, err := f.svc.Resume(f.ctx, "u1", "run_does_not_exist"); !errors.Is(err, pgstore.ErrNoRun) {
		t.Fatalf("resume of a missing run on an unwired deployment: %v, want ErrNoRun", err)
	}
	if _, err := f.svc.Resume(f.ctx, "u1", run.ID); !errors.Is(err, ErrRunnerIncomplete) {
		t.Fatalf("resume of an OWN run on an unwired deployment: %v, want the deployment's refusal", err)
	}
}

// The reconciler decides from a SNAPSHOT and writes seconds later; in between, the user presses
// stop. Three windows, one cure — the row is re-read under the lock that does the write. Each of
// these was reproduced by an independent reviewer before it was closed.

// Window 1 (money): a stop pressed while the reconciler is settling an interrupted attempt. The
// restart used to clear the fresh intent, open a second attempt and take a new hold — a 202 for a
// stop that never happened, and the user paying for the work they had just cancelled.
func TestAStopPressedWhileTheReconcilerRestartsIsNotLost(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	// The unit is gone with no marker — the shape of a reboot — and the grace is past, so the
	// reconciler will restart. The stop lands while it is settling.
	f.runner.alive = false
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	f.engine.onStatus = func() {
		f.engine.onStatus = nil
		if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
			t.Error(err)
		}
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	got := f.run(t, run.ID)
	if got.Status != "stopped" {
		t.Fatalf("the run is %q, want stopped: the stop outranks the restart", got.Status)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Fatalf("%d units started; the restart went ahead over a stop", n)
	}
	if acct := f.account(t); acct.Reserved != 0 {
		t.Fatalf("a run stopped mid-restart still holds %s", acct.Reserved.USD())
	}
	if acct := f.account(t); acct.Balance != acct.LedgerSum {
		t.Fatalf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// Window 2 (money): the sweep is about to close a run that was stopped before it spawned, and the
// queue worker creates the unit in between. Closing it anyway strands a live engine that no list
// looks at — not the live runs (finished), not the unsettled ones (its hold was released).
func TestARunSpawnedWhileTheSweepWasClosingItIsNotAbandoned(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	// The snapshot a sweep would decide from: taken while the attempt still had no unit.
	live := f.live(t)
	// The worker claims and creates the unit — BEFORE the stop, because the claim itself refuses once
	// an intent is standing. This is the window that remains: the sweep is holding a snapshot older
	// than the unit.
	claimed, err := f.store.RecordSpawn(f.ctx, pgstore.SpawnRecord{AttemptID: live.AttemptID,
		Unit: "tm-run-raced", Binary: "/opt/engine/tmctl", Ceiling: live.Ceiling,
		CeilingArg: live.Ceiling, Baseline: 1})
	if err != nil || !claimed {
		t.Fatalf("the worker could not claim the attempt: %v %v", claimed, err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.finishStopped(f.ctx, live); err != nil {
		t.Fatal(err)
	}
	got := f.run(t, run.ID)
	if got.Status == "stopped" || got.FinishedAt != nil {
		t.Fatalf("a run whose engine had just started was closed anyway: %+v", got)
	}
	if acct := f.account(t); acct.Reserved == 0 {
		t.Fatal("the hold of a run whose engine is running was released")
	}
	// And the next pass does the right thing with it: a live unit plus a standing intent is a stop.
	f.runner.alive = true
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.stops()); n == 0 {
		t.Fatal("the raced run was never asked to stop")
	}
}

// Window 3: a stop pressed on a run whose queue job has not been picked up yet. The worker must not
// start an engine for it — the stop arrived before the run did anything at all.
func TestTheQueueWorkerDoesNotStartARunThatWasAlreadyStopped(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.starts()); n != 0 {
		t.Fatalf("%d units started for a run that was stopped before its worker ran", n)
	}
	// The claim itself refuses too, which is the half that covers an intent written after the read.
	live := f.live(t)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	claimed, err := f.store.RecordSpawn(f.ctx, pgstore.SpawnRecord{AttemptID: live.AttemptID,
		Unit: "tm-run-late", Binary: "/opt/engine/tmctl", Ceiling: live.Ceiling, CeilingArg: live.Ceiling})
	if err != nil {
		t.Fatal(err)
	}
	if claimed {
		t.Fatal("the spawn claim was granted for an attempt of a run that is over and settled")
	}
	if acct := f.account(t); acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000) {
		t.Fatalf("the hold did not come back whole: balance %s reserved %s",
			acct.Balance.USD(), acct.Reserved.USD())
	}
}

// The window a RESUME opened and nothing else could: a run comes back to life, and the exit marker of
// the attempt that ended is still on disk — nothing deletes markers of attempts that are over. A pass
// holding the old snapshot would close the resumed run from it, and attempt 2's hold would then be
// in NO worklist at all: live runs are selected by `finished_at is null`, unsettled ones by an
// attempt with `ended_at`, and a re-finished resumed run matches neither.
func TestAStaleSweepDoesNotReFinishAResumedRunFromTheOldAttemptsMarker(t *testing.T) {
	f := newFixture(t, "10", 500)
	spent := money.MicroUSD(100_000)
	runID := f.stopped(t, 100, spent, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
		At: f.now.Add(time.Second)})
	stale := f.lastAttempt(t, runID) // the snapshot a slow pass is still holding: attempt 1
	if _, err := f.svc.Resume(f.ctx, "u1", runID); err != nil {
		t.Fatal(err)
	}
	held := f.account(t).Reserved
	if held == 0 {
		t.Fatal("the resumed run holds nothing")
	}
	// The slow pass arrives with attempt 1 and its marker, which is still on disk.
	if _, err := f.svc.reconcile(f.ctx, stale); err != nil {
		t.Fatal(err)
	}
	got := f.run(t, runID)
	if got.Status != "translating" || got.FinishedAt != nil {
		t.Fatalf("a stale pass closed the resumed run: %+v", got)
	}
	if acct := f.account(t); acct.Reserved != held {
		t.Fatalf("the resumed run's hold moved: %s → %s", held.USD(), acct.Reserved.USD())
	}
	// And the run is still findable, which is the half that makes the money recoverable at all.
	live, err := f.store.ListLiveRuns(f.ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(live) != 1 || live[0].AttemptNo != 2 {
		t.Fatalf("live runs after the stale pass: %+v", live)
	}
}

// A claim that was GIVEN BACK is not proof that no engine exists: `systemd-run` killed after it had
// already asked leaves one running, which is why the spend baseline is kept on such an attempt. The
// stop path must ask systemd about the unit's deterministic name rather than close over it.
func TestAStopDoesNotCloseARunWhoseGivenBackClaimLeftAnEngineRunning(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	// The unit could "not be created" — but it was: the claim is given back and the baseline stays.
	f.runner.startErr = errors.New("systemd-run was killed after it had asked")
	if err := f.svc.Spawn(f.ctx, run.ID); err == nil {
		t.Fatal("a failed spawn reported success")
	}
	live := f.live(t)
	if live.UnitName != "" || live.SpendBaseline == nil {
		t.Fatalf("after a given-back claim: unit %q, baseline %v", live.UnitName, live.SpendBaseline)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	// systemd says the orphan is alive.
	f.runner.alive = true
	f.runner.startErr = nil
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	got := f.run(t, run.ID)
	if got.FinishedAt != nil {
		t.Fatalf("the run was closed while its engine was still running: %+v", got)
	}
	if n := len(f.runner.stops()); n == 0 {
		t.Fatal("the orphaned unit was never asked to stop")
	}
	if acct := f.account(t); acct.Reserved == 0 {
		t.Fatal("the hold of a run whose engine is alive was released")
	}
}

// The window the queue worker opens: it reads the run, then spends seconds in `bookMeter` asking the
// engine what the book has cost. A stop committed inside that window has no unit to signal — so the
// only thing that can refuse the engine is the claim itself.
func TestAStopCommittedWhileTheWorkerReadsTheMeterStopsTheSpawn(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	// The stop lands while the worker is inside the meter call, i.e. after it read its snapshot.
	f.engine.onStatus = func() {
		f.engine.onStatus = nil
		if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
			t.Error(err)
		}
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	if n := len(f.runner.starts()); n != 0 {
		t.Fatalf("%d units started for a run stopped while the worker was reading the meter", n)
	}
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.run(t, run.ID); got.Status != "stopped" {
		t.Fatalf("the run is %q, want stopped", got.Status)
	}
	if acct := f.account(t); acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000) {
		t.Fatalf("the hold did not come back whole: balance %s reserved %s",
			acct.Balance.USD(), acct.Reserved.USD())
	}
}

// FP5-2 (acceptance, money). The guard FinishRun received (PD-181) was missing from the path that
// closes a stop which never spawned: a stale pass could close a RESUMED run through it, and attempt
// 2's hold would then be in no worklist at all — with the added twist that every later resume answers
// 409 forever, because the run it would continue is finished.
func TestAStaleUnspawnedStopDoesNotCloseAResumedRun(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 100})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	stale := f.live(t) // attempt 1, no unit — the snapshot a slow pass holds
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Resume(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	held := f.account(t).Reserved
	if held == 0 {
		t.Fatal("the resumed run holds nothing")
	}
	// The slow pass arrives with attempt 1 and the intent that was on it.
	if err := f.svc.finishStopped(f.ctx, stale); err != nil {
		t.Fatal(err)
	}
	got := f.run(t, run.ID)
	if got.Status != "translating" || got.FinishedAt != nil {
		t.Fatalf("a stale unspawned-stop closed the resumed run: %+v", got)
	}
	if acct := f.account(t); acct.Reserved != held {
		t.Fatalf("the resumed run's hold moved: %s → %s", held.USD(), acct.Reserved.USD())
	}
	live, err := f.store.ListLiveRuns(f.ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(live) != 1 || live[0].AttemptNo != 2 {
		t.Fatalf("live runs after the stale pass: %+v", live)
	}
}

// FP5-8(а) (acceptance). A stop that lands while the reconciler is settling an exhausted run must not
// come back as `paused/credit_exhausted`: that says the money ran out, when what happened is that the
// owner stopped it.
func TestAStopDuringSettlementOutranksThePause(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	// The attempt spends the whole ceiling, so the restart path will find nothing left; the stop lands
	// while that settlement is in flight.
	f.engine.set(statusSpending(f.svc.Pricing.Ceiling(10)), nil)
	f.engine.onStatus = func() {
		f.engine.onStatus = nil
		if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
			t.Error(err)
		}
	}
	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)
	}
	got := f.run(t, run.ID)
	if got.Status != "stopped" || got.PausedReason != "" {
		t.Fatalf("the run came back as %q/%q, want stopped", got.Status, got.PausedReason)
	}
	if acct := f.account(t); acct.Balance != acct.LedgerSum {
		t.Fatalf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// waveShape reads what a progress line actually folds into the read model: the ETA it carried, and
// the SHAPE the engine announced with it — whether this book's pipeline has an editor, and which
// generation of that answer the book is on.
//
// ⚠ It replaces `waveCounts`, which read `runs.draft_done/edit_done`. Those columns were dropped with
// register row PD-411 — two writers, no reader — so an assertion over them proved that a fold had
// happened, not that anything a client can see was right. The re-point is to the numbers this event
// STILL materialises, which is the honest remainder rather than a weakened claim: read the wrong
// field of the line and the shape comes out wrong, which is what the assertion catches.
func (f *fixture) waveShape(t *testing.T, bookID string) (editor *bool, epoch int) {
	t.Helper()
	if err := f.store.Pool().QueryRow(f.ctx,
		`select epoch_editor, shape_epoch from books where id = $1`, bookID).Scan(&editor, &epoch); err != nil {
		t.Fatal(err)
	}
	return editor, epoch
}

// runETA is the other half of what a progress line leaves behind.
func (f *fixture) runETA(t *testing.T, runID string) *int {
	t.Helper()
	var eta *int
	if err := f.store.Pool().QueryRow(f.ctx,
		`select eta_seconds from runs where id = $1`, runID).Scan(&eta); err != nil {
		t.Fatal(err)
	}
	return eta
}

// The budget of a RESUME is what the run was sold for — PD-168's second caller, the one a user
// presses. The shape is PD-168's own measurement: 100 chapters bought at $0.03, $0.50 spent, the rate
// doubled between the stop and the resume — a rate-derived budget holds $5.50 for a $2.50 remainder.
//
// Mutation caught: `budget := s.Pricing.Ceiling(l.CeilingChapters)` in reopen.
func TestAResumeHoldsWhatTheRunWasSoldForWhenTheRateHasMovedSince(t *testing.T) {
	f := newFixture(t, "10", 500)
	spent := money.MicroUSD(500_000)
	runID := f.stopped(t, 100, spent, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
		At: f.now.Add(time.Second)})
	sold := f.svc.Pricing.Ceiling(100)
	doubled, err := pricing.New(2 * pricing.DefaultPerChapter)
	if err != nil {
		t.Fatal(err)
	}
	f.svc.Pricing = doubled
	if _, err := f.svc.Resume(f.ctx, "u1", runID); err != nil {
		t.Fatal(err)
	}
	if acct := f.account(t); acct.Reserved != sold-spent {
		t.Fatalf("the resumed run holds %s, want %s (sold for %s, spent %s); priced again at today's rate it would hold %s",
			acct.Reserved.USD(), (sold - spent).USD(), sold.USD(), spent.USD(), (doubled.Ceiling(100) - spent).USD())
	}
	if acct := f.account(t); acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree after a resume", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// A run whose first hold cannot be found is NOT priced again from the rate: that would be the defect
// in another coat. Every admission opens the row in the same transaction as the run and closing it
// keeps the row, so its absence is a broken invariant — answered by name, with nothing moved.
//
// Mutation caught: falling back to `s.Pricing.Ceiling` on ErrNoFirstHold.
func TestAContinuationWithoutTheFirstHoldIsRefusedRatherThanRepriced(t *testing.T) {
	f := newFixture(t, "10", 500)
	runID := f.stopped(t, 100, 500_000, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
		At: f.now.Add(time.Second)})
	if _, err := f.store.Pool().Exec(f.ctx, `delete from reservations where engine_run_id = $1`, runID+"#1"); err != nil {
		t.Fatal(err)
	}
	before := f.account(t)
	_, err := f.svc.Resume(f.ctx, "u1", runID)
	if !errors.Is(err, pgstore.ErrNoFirstHold) {
		t.Fatalf("a resume with no first hold answered %v, want ErrNoFirstHold", err)
	}
	if live, err := f.store.ListLiveRuns(f.ctx); err != nil || len(live) != 0 {
		t.Fatalf("a run whose budget cannot be read was re-opened: %+v (%v)", live, err)
	}
	if after := f.account(t); after.Reserved != before.Reserved || after.Balance != before.Balance {
		t.Fatalf("money moved on a refused resume: before %+v, after %+v", before, after)
	}
}