textmachine/platform/internal/runs/stalled_test.go

package runs

import (
	"context"
	"errors"
	"testing"
	"time"

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

// THE BLOCKER. A pass whose reconciliation phase is spent to the last millisecond still settles the
// money of the runs that ended.
//
// It was not, and the arithmetic of it is the whole defect: the pass had one budget (120 s) and each
// run inside it had another (60 s), so TWO runs whose engine hangs consumed the pass exactly. The
// loop then returned on `ctx.Err()` and `UnsettledRuns` — the ONLY retry a deferred settlement has —
// was never called. Not for those two books: for the WHOLE installation, on every pass, because the
// list is ordered oldest-first and a wedged run is by construction the oldest. Holds of accounts
// with no connection to the wedged books stayed open, and the books behind them could not start a
// run at all.
//
// The old pin over this ground (`TestOneSlowRunDoesNotEatThePassOfTheWholeSweep`) runs the sweep
// with NO pass deadline at all, so it proves the per-run budget and cannot observe this: the register
// row it closed stood `fixed` for two packs while the pass was dying. This one gives the pass a real
// deadline, which is the only way to see the phase boundary at all.
//
// Mutation caught: running both phases on one deadline (that is, `Sweep` calling the settlement
// straight after the loop on the same context); returning the phase's `ctx.Err()` out of `Sweep`.
func TestTheSettlementPhaseIsReachedWhenTheRunPhaseSpendsThePass(t *testing.T) {
	f := newFixture(t, "20", 500)
	f.runner.alive = true
	// The repair channel is rate-limited; this fixture wants it to fire on every pass, so the hang is
	// reached rather than skipped.
	f.svc.Cfg.ResyncEvery = time.Millisecond
	// TWO live runs whose engine hangs, because two is the arithmetic of the defect: the pass was
	// 120 s and one run's budget 60 s, so it took exactly two to consume a pass entirely. One hung
	// run never could, which is why a fixture with one observes nothing.
	hung, 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, hung.ID); err != nil {
		t.Fatal(err)
	}
	secondHungBook := f.secondBook(t)
	secondHung, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: secondHungBook, CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, secondHung.ID); err != nil {
		t.Fatal(err)
	}
	secondHungDir := f.workdirOf(t, secondHungBook)
	// …and a run of ANOTHER book that ENDED and whose money is still open, because the engine could
	// not be asked at the time. Another book and, on a real deployment, another account: this is the
	// work the wedge above was stopping for everyone.
	stuck := f.secondBook(t)
	ended, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: stuck, CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, ended.ID); err != nil {
		t.Fatal(err)
	}
	endedRun := f.lastAttempt(t, ended.ID)
	if err := runner.WriteMarker(f.svc.markerPath(ended.ID, endedRun.AttemptNo),
		runner.Marker{Unit: endedRun.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
		t.Fatal(err)
	}
	f.engine.set(ingest.StatusReport{TotalUnits: 10, Done: 10}, nil) // no committed figure: absent is not zero
	if err := f.svc.Sweep(f.ctx); 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 {
		t.Fatalf("the fixture has %d unsettled runs, want the one this test is about", len(open))
	}
	f.engine.set(statusSpending(money.MicroUSD(100_000)), nil)
	f.engine.block(hung.ID, f.workdir)
	f.engine.block(secondHung.ID, secondHungDir)
	// Two runs at half the pass each: exactly the ratio that shipped.
	f.svc.Cfg.RunBudget = 400 * time.Millisecond
	// The repair channel remembers when it last ran per run, on the SERVICE's clock, so the pass above
	// used up both turns. A second of that clock is what puts them back in front of the hanging engine
	// — and it is far short of the deferral, so nothing here is skipped for the wrong reason.
	soon := f.now.Add(time.Second)
	f.svc.Now = func() time.Time { return soon }

	// A pass with a real deadline, of which the reconciliation phase may take half — which the hung
	// run then takes all of.
	pass, cancel := context.WithTimeout(f.ctx, 800*time.Millisecond)
	defer cancel()
	start := time.Now()
	// Running out is REPORTED (the starvation counter is raised by exactly this error); anything else
	// is a real failure.
	if err := f.svc.Sweep(pass); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("the sweep failed for a reason other than running out: %v", err)
	}
	if elapsed := time.Since(start); elapsed < 300*time.Millisecond {
		t.Fatalf("the pass took %v: the hung run did not actually spend the phase, so this fixture observes nothing", elapsed)
	}
	open, err = f.store.UnsettledRuns(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(open) != 0 {
		t.Errorf("the money of a finished run was not settled because another run's engine hung: %+v", open)
	}
	if got, want := f.account(t).Reserved, 2*f.svc.Pricing.Ceiling(10); got != want {
		// Only the two hung runs' own holds are left: the settled one came back.
		t.Errorf("reserved %s after the pass, want the two live holds %s", got.USD(), want.USD())
	}
}

// workdirOf is where a book's project lives, which is what the fake engine hangs by.
func (f *fixture) workdirOf(t *testing.T, bookID string) string {
	t.Helper()
	var dir string
	if err := f.store.Pool().QueryRow(f.ctx, `select workdir from books where id = $1`, bookID).Scan(&dir); err != nil {
		t.Fatal(err)
	}
	return dir
}

// A run the sweep could not finish stops standing at the head of the list.
//
// The list is oldest-first and a wedged run is the oldest, so before this it was reconciled FIRST on
// every pass, for as long as it existed, spending its whole budget every time. Deferring is what
// takes it out of the head; the ordering is untouched, because oldest-first is right for everything
// that is merely slow.
//
// Mutation caught: the sweep reading ListLiveRuns instead of RunsToReconcile; DeferRun writing the
// deadline without incrementing the count, or `now()` instead of a real one.
func TestARunTheSweepCannotFinishStopsHoldingTheHeadOfTheList(t *testing.T) {
	f := newFixture(t, "20", 500)
	hung, 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, hung.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = true
	f.engine.block(hung.ID, f.workdir)
	f.svc.Cfg.RunBudget = 100 * time.Millisecond
	f.svc.Cfg.ResyncEvery = time.Millisecond // every pass would reach the hang, if it looked at all
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// ⚠ THE SWEEP's own list, not the store's: a second pass must not touch this attempt at all. The
	// store-level assertion below passes whatever the sweep reads, so on its own it proves nothing
	// about the sweep — which is how the first version of this test survived the mutation it names.
	calls := f.engine.called()
	soon := f.now.Add(time.Second) // past the resync interval, far short of the deferral
	f.svc.Now = func() time.Time { return soon }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.engine.called(); got != calls {
		t.Errorf("the next pass called the engine %d more times about a deferred run: it is still at the head of the list",
			got-calls)
	}
	due, err := f.store.RunsToReconcile(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 0 {
		t.Fatalf("the run that spent its budget is due again immediately: %+v", due)
	}
	// It is still LIVE, and telemetry still sees it. A run hidden from the number that says how far
	// behind the projection is would be a run nobody can tell is stuck.
	live, err := f.store.ListLiveRuns(f.ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(live) != 1 {
		t.Fatalf("the deferred run vanished from the live list: %+v", live)
	}
	if live[0].ReconcileFailures != 1 {
		t.Errorf("the attempt counts %d failures after ONE reconciliation and one skipped pass: either the failure is not counted, or the deferral is not honoured",
			live[0].ReconcileFailures)
	}
	// …and when the deferral lapses it comes back.
	back, err := f.store.RunsToReconcile(f.ctx, f.svc.now().Add(backoff(1)+time.Second))
	if err != nil {
		t.Fatal(err)
	}
	if len(back) != 1 {
		t.Fatalf("the run never becomes due again: %+v — a deferral is a delay, not a deletion", back)
	}
	// A pass that succeeds forgets the failures, or a run that hiccups once carries it forever.
	if err := f.store.ClearRunDeferral(f.ctx, live[0].AttemptID); err != nil {
		t.Fatal(err)
	}
	due, err = f.store.RunsToReconcile(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 1 || due[0].ReconcileFailures != 0 {
		t.Errorf("after a successful pass the attempt is %+v, want it due with no failures", due)
	}
}

// Enough failures in a row and the run becomes something an operator is TOLD about and can act on.
//
// The half that existed was the telling, and only just: `tm_platform_sweep_unfinished_total` says a
// PASS did not finish and never says on what — so an operator watched a counter climb with nothing
// to look at and nothing to do. What this asserts is the two halves that were missing: the state has
// a name (`StalledRuns`, the gauge and the CLI read it) and it carries what the decision needs —
// which run, how many times, what it said, and what it is holding.
//
// Mutation caught: counting failures per process rather than in the row; StalledRuns filtering on
// finished runs or forgetting the open reservation.
func TestARunThatKeepsFailingBecomesTheOperatorsProblem(t *testing.T) {
	f := newFixture(t, "20", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	live := f.live(t)
	for i := 1; i <= StalledAfter; i++ {
		got, err := f.store.DeferRun(f.ctx, live.AttemptID, f.svc.now().Add(backoff(i)), "the engine did not answer")
		if err != nil {
			t.Fatal(err)
		}
		if got != i {
			t.Fatalf("after %d deferrals the row counts %d", i, got)
		}
	}
	// Below the threshold the list is empty: an operator's list that includes every hiccup is one
	// they stop reading.
	none, err := f.store.StalledRuns(f.ctx, StalledAfter+1)
	if err != nil {
		t.Fatal(err)
	}
	if len(none) != 0 {
		t.Errorf("a run below the threshold is reported as stalled: %+v", none)
	}
	// …and a floor of ZERO lists every live run, which is what the plain `tmplatformctl runs` asks
	// for. A floor of one dropped the healthy half of what the usage line promises, so on a working
	// deployment the command answered as if nothing were running at all.
	all, err := f.store.StalledRuns(f.ctx, 0)
	if err != nil {
		t.Fatal(err)
	}
	if len(all) != 1 {
		t.Errorf("the operator's plain listing shows %d live runs, want the one that exists", len(all))
	}
	stalled, err := f.store.StalledRuns(f.ctx, StalledAfter)
	if err != nil {
		t.Fatal(err)
	}
	if len(stalled) != 1 {
		t.Fatalf("%d stalled runs, want the one this test wedged", len(stalled))
	}
	got := stalled[0]
	switch {
	case got.RunID != run.ID:
		t.Errorf("the stalled run is %s, want %s", got.RunID, run.ID)
	case got.Failures != StalledAfter:
		t.Errorf("it counts %d failures", got.Failures)
	case got.LastError == "":
		t.Error("it carries no last error: an operator cannot tell a hung engine from a missing workdir")
	case got.NextTry == nil:
		t.Error("it carries no next attempt time: an operator cannot tell a deferred run from an abandoned one")
	case money.MicroUSD(got.HeldMicroUSD) != f.svc.Pricing.Ceiling(10):
		t.Errorf("it reports %s held, want the run's own hold %s — the money frozen by the stall is the point",
			money.MicroUSD(got.HeldMicroUSD).USD(), f.svc.Pricing.Ceiling(10).USD())
	}
	// What the engine has reported spending on THIS ATTEMPT travels with the row, because the decision
	// the row informs — give the hold back WHOLE — is otherwise made blind to how much work is being
	// written off.
	//
	// ⚠ AND IT IS THE ATTEMPT'S SHARE, not the column: the engine's figure is the BOOK's lifetime
	// total, so a book on its second run carried the first run's spending in the same number. The
	// fixture puts a baseline under it because the difference is the answer.
	if _, err := f.store.Pool().Exec(f.ctx, `
		update run_attempts set spend_micro_usd = 123456, spend_baseline_micro_usd = 100000
		 where id = $1`, live.AttemptID); err != nil {
		t.Fatal(err)
	}
	again, err := f.store.StalledRuns(f.ctx, StalledAfter)
	if err != nil {
		t.Fatal(err)
	}
	if len(again) != 1 || again[0].SpentMicroUSD == nil || *again[0].SpentMicroUSD != 23456 {
		t.Errorf("the operator's row reports %+v: this attempt spent 23456, and the rest is what the book cost before it — `--release-hold` decided from the larger number writes off work that was already paid for", again)
	}
	// An attempt with NO baseline is the one the settlement refuses to price, and the row says "not
	// known" rather than naming a figure: a zero here is one an operator could act on.
	if _, err := f.store.Pool().Exec(f.ctx, `
		update run_attempts set spend_baseline_micro_usd = null where id = $1`, live.AttemptID); err != nil {
		t.Fatal(err)
	}
	unpriced, err := f.store.StalledRuns(f.ctx, StalledAfter)
	if err != nil {
		t.Fatal(err)
	}
	if len(unpriced) != 1 || unpriced[0].SpentMicroUSD != nil {
		t.Errorf("an attempt with no baseline reports %+v: the settlement withholds a verdict on exactly this attempt, and the table must not invent one", unpriced)
	}
	// The gauge behind the CLI counts the same thing.
	o, err := f.store.Observe(f.ctx, StalledAfter)
	if err != nil {
		t.Fatal(err)
	}
	if o.StalledRuns != 1 {
		t.Errorf("the observation counts %d stalled runs", o.StalledRuns)
	}

	// ⚠ AND THE DEFERRAL TRAVELS WITH THE ATTEMPT INTO THE SETTLEMENT LIST — the REVERSE of what this
	// pin asserted when it was written. The old claim (the two lists must stay asymmetric, because
	// deferred reconciliation still has money to close) reads well and left the settlement list with
	// the same starvation: it is ordered oldest-first too. An attempt already known to be unreachable
	// should not re-learn it at the price of a whole budget.
	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) != 0 {
		t.Errorf("the settlement list shows %d runs while the attempt is serving its backoff: one wedged settlement at the head of this list starves every other account's hold", len(open))
	}
	// And it comes back when the backoff lapses — deferred, never dropped: a hold nothing ever looks
	// at again is worse than one that is closed late.
	later, err := f.store.UnsettledRuns(f.ctx, f.svc.now().Add(backoff(StalledAfter)+time.Second))
	if err != nil {
		t.Fatal(err)
	}
	if len(later) != 1 {
		t.Errorf("the settlement list shows %d runs after the backoff lapsed: the money must be retried, not abandoned", len(later))
	}
}

// The operator's terminal verdict: what it refuses to close, and what it does with the money.
//
// It refuses a run whose attempt still NAMES a unit — that is a process systemd may still be
// running, and closing the run over it leaves an engine spending against something nothing looks at
// any more ("I could not ask" is never "it is gone").
//
// The hold comes back WHOLE either way, because that guard has already decided the money: only an
// attempt that never reached the engine can be abandoned, so nothing was spent. `--release-hold`
// chooses WHEN — now, or on the next sweep — and its own branch has to record the settlement, since
// it takes the run out of the settlement list before that list ever sees it.
//
// Mutation caught: dropping the unit guard; leaving the book without the reading-surface debt every
// other ending leaves; releasing the hold without stamping `settled_at`.
func TestAbandoningAStalledRunIsRefusedOverALiveUnitAndAlwaysGivesTheHoldBack(t *testing.T) {
	f := newFixture(t, "20", 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)
	}
	_, err = f.store.AbandonRun(f.ctx, run.ID, "the host is gone", false, f.now)
	if !errors.Is(err, pgstore.ErrRunMayHaveAProcess) {
		t.Fatalf("abandoning a run with a unit answered %v, want a refusal", err)
	}
	// ⚠ AND WITH THE NAME CLEARED, which is the case the name alone cannot see. `ReleaseSpawnClaim`
	// empties `unit_name` when the unit could not be CREATED — and a `systemd-run` killed after it had
	// already asked leaves an engine running, which is why the spend baseline is kept on such an
	// attempt as its tombstone (RecordSpawn). Closing the run here would leave that engine spending
	// against a run nothing looks at, and `--release-hold` would hand back the money it is spending.
	if _, err := f.store.Pool().Exec(f.ctx,
		`update run_attempts set unit_name = null where run_id = $1`, run.ID); err != nil {
		t.Fatal(err)
	}
	var baseline *int64
	if err := f.store.Pool().QueryRow(f.ctx,
		`select spend_baseline_micro_usd from run_attempts where run_id = $1`, run.ID).Scan(&baseline); err != nil {
		t.Fatal(err)
	}
	if baseline == nil {
		t.Fatal("the fixture's attempt carries no spend baseline: it cannot observe the tombstone case")
	}
	if _, err := f.store.AbandonRun(f.ctx, run.ID, "the host is gone", false, f.now); !errors.Is(err, pgstore.ErrRunMayHaveAProcess) {
		t.Errorf("abandoning a run whose claim was given back but whose baseline stands answered %v, want a refusal", err)
	}

	// The case the handle is for: an attempt that never became a process (register row PD-162 — a
	// book whose directory was removed refuses every spawn, forever, with the hold open).
	stuck := f.secondBook(t)
	unspawned, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: stuck, CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	// A ceiling the stream reported before the run wedged: the column the abandon has to clear.
	if _, err := f.store.Pool().Exec(f.ctx,
		`update runs set paused_reason = 'credit_exhausted' where id = $1`, unspawned.ID); err != nil {
		t.Fatal(err)
	}
	before := f.account(t)
	if _, err := f.store.AbandonRun(f.ctx, unspawned.ID, "the workdir was removed", false, f.now); err != nil {
		t.Fatal(err)
	}
	if got := f.account(t); got.Reserved != before.Reserved {
		t.Errorf("the hold moved without being asked to: %s → %s", before.Reserved.USD(), got.Reserved.USD())
	}
	// ⚠ WHAT it is left as is contract-visible and is asserted, not assumed: the book's status is that
	// of its current or last run (canon §BookStatus), so a run left `stopped` under a book left
	// `failed` gives a client two answers and no rule for choosing. `service_error` is the vocabulary's
	// word for "the deployment, and retrying alone does not clear it" — which is exactly what an
	// abandoned run is.
	var runStatus, runReason, bookStatus string
	if err := f.store.Pool().QueryRow(f.ctx, `
		select r.status, coalesce(r.failure_reason, ''), b.status
		  from runs r join books b on b.id = r.book_id where r.id = $1`, unspawned.ID).
		Scan(&runStatus, &runReason, &bookStatus); err != nil {
		t.Fatal(err)
	}
	if runStatus != "failed" || runReason != "service_error" {
		t.Errorf("the abandoned run is %q/%q, want failed/service_error", runStatus, runReason)
	}
	// …and it carries no PAUSE reason. The canon allows one only while a run is `paused`, and a live
	// run can have picked one up from its stream (a ceiling event) before it wedged — so leaving the
	// column alone puts a machine reason on the wire under a status that forbids it. `FinishRun`
	// nulls it for exactly this reason.
	var paused *string
	if err := f.store.Pool().QueryRow(f.ctx,
		`select paused_reason from runs where id = $1`, unspawned.ID).Scan(&paused); err != nil {
		t.Fatal(err)
	}
	if paused != nil {
		t.Errorf("the abandoned run carries paused_reason %q under status %q, which the canon allows only while paused",
			*paused, runStatus)
	}
	if bookStatus != runStatus {
		t.Errorf("the book says %q and its last run says %q: a client holding both has to decide which wins",
			bookStatus, runStatus)
	}
	// The run is over, which is what frees the BOOK: a live run is what `runs_one_live_per_book`
	// refuses a second one against.
	if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: stuck, CeilingChapters: 5}); err != nil {
		t.Fatalf("the book is still frozen after its run was abandoned: %v", err)
	}
	// …and the book owes a reading surface, like it does after every other ending: whatever was
	// translated before the stall is bought and paid for.
	var owed *time.Time
	if err := f.store.Pool().QueryRow(f.ctx,
		`select read_model_owed_at from books where id = $1`, stuck).Scan(&owed); err != nil {
		t.Fatal(err)
	}
	if owed == nil {
		t.Error("the abandoned run left its book owing nothing: its text will never reach a reader")
	}

	// Without the flag the money is not left open: the next sweep closes it, whole.
	swept, err := f.store.UnsettledRuns(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(swept) != 1 {
		t.Fatalf("the abandoned run is not in the settlement list (%d rows): its hold comes back from nowhere else", len(swept))
	}
	beforeSweep := f.account(t)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := f.account(t); got.Reserved != beforeSweep.Reserved-f.svc.Pricing.Ceiling(10) {
		t.Errorf("reserved %s → %s after the sweep: an abandoned run never reached the engine, so its hold is owed back whole",
			beforeSweep.Reserved.USD(), got.Reserved.USD())
	}

	// And with the flag, the same thing happens inside the command itself.
	third := f.secondBook(t)
	last, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: third, CeilingChapters: 10})
	if err != nil {
		t.Fatal(err)
	}
	held := f.account(t)
	if _, err := f.store.AbandonRun(f.ctx, last.ID, "the engine will never answer", true, f.now); err != nil {
		t.Fatal(err)
	}
	after := f.account(t)
	if after.Reserved != held.Reserved-f.svc.Pricing.Ceiling(10) {
		t.Errorf("reserved %s → %s, want the run's own hold given back", held.Reserved.USD(), after.Reserved.USD())
	}
	// …and the run is recorded as SETTLED. Nothing else will ever stamp it: this branch closes the
	// reservation, so the settlement list never sees the run and the column stayed null for good.
	var settled *time.Time
	if err := f.store.Pool().QueryRow(f.ctx,
		`select settled_at from runs where id = $1`, last.ID).Scan(&settled); err != nil {
		t.Fatal(err)
	}
	if settled == nil {
		t.Error("the run whose hold this command returned is still recorded as unsettled: nothing else will ever say otherwise")
	}
	if after.Balance != held.Balance+f.svc.Pricing.Ceiling(10) {
		t.Errorf("balance %s → %s, want the hold back whole", held.Balance.USD(), after.Balance.USD())
	}
	if after.Balance != after.LedgerSum {
		t.Errorf("the cached balance and the ledger disagree after an abandon: %s vs %s",
			after.Balance.USD(), after.LedgerSum.USD())
	}
	// ⚠ THE ANSWER CHANGED WITH PD-385 and the assertion moved with it, deliberately and not to get a
	// green: a finished run is no longer met with "there is no such run". It is met with the state it
	// is actually in — finished, money already closed, nothing to do — because answering the second
	// with the first is precisely what made a FROZEN hold read as a typo. What the test pins is
	// unchanged: abandoning twice is refused and the money moves once.
	if _, err := f.store.AbandonRun(f.ctx, last.ID, "again", true, f.now); !errors.Is(err, pgstore.ErrMoneyAlreadyClosed) {
		t.Errorf("abandoning an already-abandoned run answered %v, want ErrMoneyAlreadyClosed", err)
	}
}

// backoff grows and stops growing. A delay that doubled without a ceiling would take a run that
// healed hours ago and leave it waiting days.
func TestTheBackoffGrowsAndIsCapped(t *testing.T) {
	if backoff(1) != time.Minute {
		t.Errorf("the first deferral is %v, want a minute", backoff(1))
	}
	if backoff(2) <= backoff(1) {
		t.Errorf("the deferral does not grow: %v then %v", backoff(1), backoff(2))
	}
	if got := backoff(1000); got != 30*time.Minute {
		t.Errorf("the deferral of a long-dead run is %v, want the cap", got)
	}
}

// THE DEFERRAL ENGAGES AT THE RATIO THIS ZONE ACTUALLY SHIPS, and that is a different assertion from
// "the deferral engages".
//
// It is here because the mechanism did NOT engage on any default deployment, and the arithmetic is
// the whole of it: a 2-minute pass gives the reconciliation phase half — 60 seconds — and one run's
// budget is 60 seconds. The item's context is a CHILD of the phase's and is created later, so its
// deadline is never the later of the two; at these numbers they are the same instant. A test that
// judged the overrun by asking whether the PARENT still had time therefore answered "no" exactly
// when the item had hung — and the code took the other branch and recorded the wedged run as
// SUCCESSFULLY reconciled, clearing whatever count it had. Mechanisms 2, 3 and 4 of this pack —
// deferral, the stalled gauge, the operator's handle — were unreachable in production while every
// test of them passed, because every one of those tests set a run budget far below the pass.
//
// So this fixture sets the SHIPPED RATIO (one run's budget is exactly the phase's share) and nothing
// else. Found by a reviewer's pin, not by reading.
//
// Mutation caught: judging the overrun by `ctx.Err() == nil` instead of by how long the item ran.
func TestTheDeferralEngagesAtTheRatioThisZoneShips(t *testing.T) {
	f := newFixture(t, "20", 500)
	f.runner.alive = true
	f.svc.Cfg.ResyncEvery = time.Millisecond
	hung, 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, hung.ID); err != nil {
		t.Fatal(err)
	}
	f.engine.block(hung.ID, f.workdir)

	// 1 : 2, exactly as the defaults ship (TM_PLATFORM_RUN_BUDGET 60s inside TM_PLATFORM_SWEEP_BUDGET
	// 2m, of which the phase takes half).
	f.svc.Cfg.RunBudget = 200 * time.Millisecond
	pass, cancel := context.WithTimeout(f.ctx, 400*time.Millisecond)
	defer cancel()
	if err := f.svc.Sweep(pass); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("the sweep failed for a reason other than running out: %v", err)
	}
	live, err := f.store.ListLiveRuns(f.ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(live) != 1 {
		t.Fatalf("%d live runs", len(live))
	}
	if live[0].ReconcileFailures != 1 {
		t.Fatalf("the hung run counts %d failures at the shipped ratio: the deferral never engages, so nothing ever reaches StalledAfter and the operator's handle is unreachable",
			live[0].ReconcileFailures)
	}
	due, err := f.store.RunsToReconcile(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 0 {
		t.Errorf("the hung run is due again immediately: it holds the head of the list on every pass")
	}
}

// THE BLOCKER, SECOND HALF: one settlement nobody can finish stops holding the head of ITS list too.
//
// The pass was split so a wedged reconciliation could not starve the money, and the money phase kept
// the same defect one level down: `UnsettledRuns` is ordered oldest-first, each item costs a whole
// budget, and `settle` answers nil when the engine's figure cannot be read — so the wedge reports
// success while spending the pass, and the same items are first again next time.
//
// Mutation caught: `settleOne` not deferring on an overrun; `UnsettledRuns` ignoring the deferral.
func TestASettlementNobodyCanFinishStopsHoldingTheHeadOfTheMoneyList(t *testing.T) {
	f := newFixture(t, "20", 500)
	// Three books that all END with their money still open — the state a settlement deferral leaves,
	// reached the ordinary way: the engine answers without a committed figure, and absent is not zero.
	f.engine.set(statusSpending(money.MicroUSD(0)), nil) // a spawn reads the meter and needs a figure
	dirs := make([]string, 0, 3)
	runIDs := make([]string, 0, 3)
	books := []string{f.bookID(t), f.secondBook(t), f.secondBook(t)} // read the first BEFORE adding any
	for _, book := range books {
		run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, CeilingChapters: 10})
		if err != nil {
			t.Fatal(err)
		}
		if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
			t.Fatal(err)
		}
		dirs = append(dirs, f.workdirOf(t, book))
		runIDs = append(runIDs, run.ID)
	}
	// Now the engine stops reporting a committed figure, and each run is ended in its own second so
	// that "oldest first" means something to assert against.
	f.engine.set(ingest.StatusReport{TotalUnits: 10, Done: 10}, nil)
	for i, id := range runIDs {
		a := f.lastAttempt(t, id)
		if err := runner.WriteMarker(f.svc.markerPath(id, a.AttemptNo),
			runner.Marker{Unit: a.UnitName, Result: "exit-code", Code: "exited", Status: "0"}); err != nil {
			t.Fatal(err)
		}
		ended := f.now.Add(time.Duration(i+1) * time.Second)
		f.svc.Now = func() time.Time { return ended }
		if err := f.svc.Sweep(f.ctx); err != nil {
			t.Fatal(err)
		}
	}
	// ⚠ READ PAST THE BACKOFF, and the arithmetic is worth writing down because it is the fixture's
	// own doing rather than a change of behaviour. The setup sweeps once per run, and each sweep's
	// settlement phase meets EVERY run already on the list — so the first book collects three
	// failures, the second two, the third one. Since PD-384 a blocked settlement is counted, and from
	// the SECOND failure it backs off (the first is retried at once, so a transient fault still costs
	// fifteen seconds — TestTheFirstFailedSettlementIsRetriedAtOnce...). Three minutes clears the
	// longest of those. What this test is about begins after that: they are all due again, and two of
	// them are still unreadable.
	measuring := f.now.Add(3 * time.Minute)
	f.svc.Now = func() time.Time { return measuring }
	open, err := f.store.UnsettledRuns(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(open) != 3 {
		t.Fatalf("%d unsettled runs, want the three this fixture is about", len(open))
	}
	// The two OLDEST are the ones whose project stopped answering — by construction the two at the
	// head of a list ordered by `ended_at`. The third is an ordinary settlement waiting behind them.
	f.engine.set(statusSpending(money.MicroUSD(100_000)), nil)
	f.engine.block("", dirs[0])
	f.engine.block("", dirs[1])
	f.svc.Cfg.RunBudget = 200 * time.Millisecond

	// Pass one: the two wedges spend it, exactly as they did before this fix. What is new is that
	// they are now COUNTED, which is the only thing that can get them out of the way.
	first, cancelFirst := context.WithTimeout(f.ctx, 400*time.Millisecond)
	defer cancelFirst()
	if err := f.svc.Sweep(first); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("the sweep failed for a reason other than running out: %v", err)
	}
	// Pass two, with the clock unmoved: the deferred pair is out of the list, so the third book's
	// money — a different account's, on a real deployment — is finally reached.
	second, cancelSecond := context.WithTimeout(f.ctx, 400*time.Millisecond)
	defer cancelSecond()
	if err := f.svc.Sweep(second); err != nil && !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("the second sweep failed for a reason other than running out: %v", err)
	}
	left, err := f.store.UnsettledRuns(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(left) != 0 {
		t.Errorf("%d settlements are still due: the wedged pair holds the head of the money list, so every other account's hold waits behind a project nobody can read", len(left))
	}
	// And the money moved: only the two unreachable holds are still reserved.
	if got, want := f.account(t).Reserved, 2*f.svc.Pricing.Ceiling(10); got != want {
		t.Errorf("reserved %s, want the two wedged holds %s", got.USD(), want.USD())
	}
}

// The failure count reaches the threshold it is measured against.
//
// It could not: `maybeResync` asks the engine at most once every ResyncEvery and skips it while the
// stream moves, so a hung run failed once, was deferred, came back to a pass that asked nothing, and
// had its count cleared by that silence — 1, 0, 1, 0. StalledAfter was unreachable on any
// deployment.
//
// Mutation caught: clearing the deferral on a pass that established nothing (that is,
// `ClearRunDeferral` on plain `err == nil`).
func TestTheFailureCountSurvivesThePassesThatAskTheEngineNothing(t *testing.T) {
	f := newFixture(t, "20", 500)
	f.runner.alive = true
	f.svc.Cfg.ResyncEvery = time.Hour // the repair channel is due once and then never again here
	hung, 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, hung.ID); err != nil {
		t.Fatal(err)
	}
	f.engine.block(hung.ID, f.workdir)
	f.svc.Cfg.RunBudget = 100 * time.Millisecond
	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].ReconcileFailures != 1 {
		t.Fatalf("the hung run counts %d failures after the pass that reached it", live[0].ReconcileFailures)
	}
	// Now the passes that cost nothing and prove nothing: the deferral has lapsed, so the run IS
	// looked at, but the repair channel is not due, so the engine is never asked. Before this fix each
	// of these reset the count to zero.
	for i := range 3 {
		later := f.now.Add(backoff(1) + time.Duration(i+1)*time.Second)
		f.svc.Now = func() time.Time { return later }
		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 live[0].ReconcileFailures != 1 {
		t.Errorf("the count is %d after three passes that asked the engine nothing: silence is not evidence of health, and a count reset by it never reaches StalledAfter",
			live[0].ReconcileFailures)
	}
}

// A daemon shutting down mid-pass is not a starved installation.
//
// Both phases report running out with `DeadlineExceeded`, which becomes `sweep_unfinished_total`. A
// cancelled context is the same `ctx.Done()` and a different fact, and reading them alike raised
// that counter on every restart and wrote failures to whatever the pass was holding.
//
// Mutation caught: reporting on any expired context rather than on the phase's own clock; deferring
// an item whose context was cancelled.
func TestAPassTheDaemonStoppedIsNotReportedAsStarvation(t *testing.T) {
	f := newFixture(t, "20", 500)
	f.runner.alive = true
	f.svc.Cfg.ResyncEvery = time.Millisecond
	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)
	}
	f.engine.block(run.ID, f.workdir)
	f.svc.Cfg.RunBudget = time.Minute // far longer than the shutdown below, so only the cancel can end it
	pass, stop := context.WithCancel(f.ctx)
	go func() {
		time.Sleep(100 * time.Millisecond)
		stop() // the daemon goes down while the sweep is inside the hung engine
	}()
	defer stop()
	if err := f.svc.Sweep(pass); err != nil {
		t.Errorf("a sweep the daemon stopped answered %v: the starvation counter rises on exactly this error, so a routine restart reads as a wedged installation", err)
	}
	live, err := f.store.ListLiveRuns(f.ctx)
	if err != nil {
		t.Fatal(err)
	}
	if live[0].ReconcileFailures != 0 {
		t.Errorf("the run counts %d failures after a shutdown: an orderly restart must not write runs a failure they did not have", live[0].ReconcileFailures)
	}
}

// A user's stop outranks a deferral: new evidence beats a statement about the past.
//
// The sweep is the only reader of live runs, so an attempt serving a backoff is one nothing looks
// at — and `Stop` promises the reconciler will re-issue on its next pass. Without this the stopped
// run sat at `translating` with its hold reserved until the backoff lapsed, which on a run already
// deferred a few times is minutes, and the backoff caps at half an hour.
//
// Mutation caught: dropping `or r.stop_requested_at is not null` from RunsToReconcile.
func TestAStopOutranksADeferral(t *testing.T) {
	f := newFixture(t, "20", 500)
	f.runner.alive = true
	f.svc.Cfg.ResyncEvery = time.Millisecond
	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)
	}
	f.engine.block(run.ID, f.workdir)
	f.svc.Cfg.RunBudget = 100 * time.Millisecond
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	// Deferred, and therefore invisible to the sweep until the backoff lapses.
	due, err := f.store.RunsToReconcile(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(due) != 0 {
		t.Fatalf("the hung run is due immediately: this fixture is not the one it claims")
	}
	// The user presses stop. That is evidence that arrived AFTER the backoff was set.
	if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
		t.Fatal(err)
	}
	back, err := f.store.RunsToReconcile(f.ctx, f.svc.now())
	if err != nil {
		t.Fatal(err)
	}
	if len(back) != 1 {
		t.Errorf("a stopped run is still serving its backoff: its hold stays reserved and the run reads `translating` for as long as the deferral lasts")
	}
}

// A pass that ran out with runs it never reached SAYS SO, because the counter that shows starvation
// is raised by the caller testing for exactly that error (runner.pass → metrics.ObserveSweep).
//
// Returning nil made the sweep quiet and the counter unable to rise — and `tm_platform_sweep_unfinished_total`
// is the one signal register row PD-169 names as the visibility of this whole class. Losing it while
// fixing the class is the worst possible trade.
//
// Mutation caught: returning nil from the phase when it runs out with work left.
func TestAPassThatRanOutOfTimeSaysSo(t *testing.T) {
	f := newFixture(t, "20", 500)
	f.runner.alive = true
	f.svc.Cfg.ResyncEvery = time.Millisecond
	// Two runs and a phase that can hold neither: the second is never reached.
	for _, book := range []string{f.bookID(t), f.secondBook(t)} {
		run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, CeilingChapters: 10})
		if err != nil {
			t.Fatal(err)
		}
		if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
			t.Fatal(err)
		}
		var dir string
		if err := f.store.Pool().QueryRow(f.ctx, `select workdir from books where id = $1`, book).Scan(&dir); err != nil {
			t.Fatal(err)
		}
		f.engine.block(run.ID, dir)
	}
	f.svc.Cfg.RunBudget = 300 * time.Millisecond
	pass, cancel := context.WithTimeout(f.ctx, 400*time.Millisecond)
	defer cancel()
	err := f.svc.Sweep(pass)
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Errorf("a pass that never reached its second run answered %v; the counter that shows starvation is raised by exactly this error and can no longer rise",
			err)
	}
}