package runs

import (
	"errors"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

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

// The consumer half of the run-event seam (П-15), driven through the WHOLE sweep and not through the
// pure function next door. The decision tables in reconcile_test.go say what `outcome` returns; these
// say that a run really does end up in that state, with its money settled and its next action taken
// — which is the gap a reviewer found once already, by deleting a branch a table test could not miss.

// spawned admits a run and starts its unit, returning the live attempt.
func spawned(t *testing.T, f *fixture, chapters int) pgstore.LiveRun {
	t.Helper()
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(chapters)})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.runner.alive = false // the unit is gone by the time the sweep looks, which is what a marker means
	return f.live(t)
}

// ends writes the exit marker of a live attempt, as ExecStopPost would.
func ends(t *testing.T, f *fixture, l pgstore.LiveRun, m runner.Marker) {
	t.Helper()
	m.Unit = l.UnitName
	if m.At.IsZero() {
		m.At = f.now.Add(time.Minute)
	}
	if err := runner.WriteMarker(f.svc.markerPath(l.RunID, l.AttemptNo), m); err != nil {
		t.Fatal(err)
	}
}

func exitedWith(code string) runner.Marker {
	return runner.Marker{Result: "exit-code", Code: "exited", Status: code}
}

// card reads the run the book's card carries.
func card(t *testing.T, f *fixture) pgstore.Run {
	t.Helper()
	_, run, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
	if err != nil {
		t.Fatal(err)
	}
	if run == nil {
		t.Fatal("the book has no run")
	}
	return *run
}

// PD-113, end to end and with NO event on the stream at all — which is the half only the exit code
// can carry, and the half that did not exist before D39.131. A journal the engine could not write
// still leaves exit 4 behind.
//
// Three things have to be true together, and the third is why this is a sweep test: the run is
// `paused`, it carries a reason (a paused run with a null one gives the screen nothing to say), and
// it is NOT restarted — a ceiling halt is a run that has spent what was bought for it, and starting
// a second engine would walk it into the same stop and charge for the privilege.
//
// Mutation caught: dropping ingest.OutcomeCeiling from outcome(), and dropping PausedReason from
// RunEnding.
func TestACeilingHaltWithNoEventIsPausedWithItsReasonAndNotRestarted(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	ends(t, f, l, exitedWith("4"))
	spent := fixtureHold(100)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 40, Spend: usd(spent), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	got := card(t, f)
	if got.Status != "paused" {
		t.Fatalf("a ceiling halt was recorded as %q; the contract forbids `failed` for a resumable stop", got.Status)
	}
	// The reason the platform can stand behind: a ceiling stopped it, and with no event to say WHOSE
	// ceiling, that is exactly what is recorded. `credit_exhausted` would be a guess that lights the
	// account's own halted flag.
	if got.PausedReason != pgstore.PausedCeilingUnknown {
		t.Errorf("paused_reason %q, want %q", got.PausedReason, pgstore.PausedCeilingUnknown)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Errorf("%d units started: a ceiling halt must not be respawned", n)
	}
	if acct := f.account(t); acct.Reserved != 0 {
		t.Errorf("the hold is still open after a ceiling halt: %s", acct.Reserved.USD())
	}
}

// The other channel, and the reason the re-read exists. The ceiling event arrives in the drain of
// THIS very sweep, so the snapshot the pass started from still says nothing — and judging the ending
// from that snapshot answered `failed` for a run whose own stream had just said `ceiling`
// (acceptance of D39.131, п.3). It also carries the SCOPE, which the exit code cannot.
//
// Mutation caught: reading paused_reason from the sweep's snapshot instead of re-reading it after
// the drain.
func TestACeilingEventArrivingInThisVerySweepStillDecidesTheEnding(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	body := hello(t, f) +
		`{"seq":2,"type":"ceiling","data":{"halted":true,"scope":"day"}}` + "\n" +
		`{"seq":3,"type":"finished","data":{"outcome":"ceiling"}}` + "\n"
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	// The marker says only that the process ended; the class it ended with is on the stream. Written
	// as exit 1 on purpose — that is the shape this platform saw for a ceiling halt before the engine
	// had a code for it, and the branch must not depend on the code being there.
	ends(t, f, l, exitedWith("1"))
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 40, Spend: usd(1_000_000), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	got := card(t, f)
	if got.Status != "paused" || got.PausedReason != pgstore.PausedDailyCeiling {
		t.Fatalf("the run ended as %q/%q, want paused/%s", got.Status, got.PausedReason, pgstore.PausedDailyCeiling)
	}
}

// A run the ENGINE'S OWN daily ceiling stopped cannot be continued by anything this platform can do:
// `ceilings.day_usd` lives in a book.yaml the operator owns. Resuming would open an attempt that
// meets the same limit inside the same day and comes straight back, paying for a status call and a
// transient unit each time — so the answer is a refusal with a reason, not a loop.
// ⚠ 0.3.0 WIDENED this: EVERY paused run is refused with `ceiling_reached`, not only the one the
// engine's daily ceiling stopped. The limit travels with the start of a run, so this call cannot
// move any of them, and the remedy the client offers is a NEW run with a larger ceiling — legal from
// any paused book, whatever the reason says, including `null` (canon §resumeRun).
//
// Mutation caught: re-opening a `paused` run instead of refusing it; refusing only the daily
// ceiling and letting the platform's own reason fall through into a re-open.
func TestAResumeOfARunPausedAtALimitIsRefusedWhoseverLimitItWas(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	// Only a little of the budget is gone, so `exhausted` is NOT what refuses this: the run has money
	// left and would happily re-open without the guard.
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 5, Spend: usd(100_000), Reserved: usd(0)}, nil)
	if _, err := f.store.PauseRun(f.ctx, l.RunID, l.AttemptID, pgstore.PausedDailyCeiling, f.now); err != nil {
		t.Fatal(err)
	}
	if err := f.svc.Sweep(f.ctx); err != nil { // settles the attempt, so `deferred` is not what refuses either
		t.Fatal(err)
	}
	if _, err := f.svc.Resume(f.ctx, "u1", l.RunID); !errors.Is(err, ErrCeilingReached) {
		t.Fatalf("resume of a run the engine's daily ceiling stopped: %v, want ErrCeilingReached", err)
	}
	if got := card(t, f); got.Status != "paused" || got.PausedReason != pgstore.PausedDailyCeiling {
		t.Errorf("the refused resume changed the run to %q/%q", got.Status, got.PausedReason)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Errorf("%d units started: the refusal must not have spawned anything", n)
	}
	// The ceiling the PLATFORM set answers the same way and for the same reason. It used to be the
	// opposite case — re-opened silently — and the client could not tell that 202 from a run that had
	// actually continued.
	if _, err := f.store.Pool().Exec(f.ctx,
		`update runs set paused_reason = $2 where id = $1`, l.RunID, pgstore.PausedCreditExhausted); err != nil {
		t.Fatal(err)
	}
	if _, err := f.svc.Resume(f.ctx, "u1", l.RunID); !errors.Is(err, ErrCeilingReached) {
		t.Fatalf("resume of a run the platform's own ceiling stopped: %v, want ErrCeilingReached", err)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Errorf("%d units started after the second refusal", n)
	}
}

// Exit 5 is a caught SIGTERM and says nothing about who sent it. With nothing recorded on our side
// it is an INTERRUPTION — a host reboot, or an operator's `systemctl stop` — and the run comes back
// with what is left of its budget (unified backlog row 138). Burying it as `stopped` would leave
// every run on the host dead after a routine restart, which is the remaining half of PD-152.
//
// Mutation caught: removing the interruptedBySomeoneElse branch from reconcile().
func TestAGracefulSignalNobodyAskedForBringsTheRunBack(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	spent := money.MicroUSD(900_000)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 30, Spend: usd(spent), Reserved: usd(0)}, nil)
	ends(t, f, l, exitedWith("5"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	next := f.live(t)
	if next.AttemptNo != 2 {
		t.Fatalf("a run wound down by a signal nobody recorded was not restarted: attempt %d", next.AttemptNo)
	}
	if want := fixtureHold(100) - spent; next.Ceiling != want {
		t.Errorf("the new attempt reserved %s, want the budget minus what was spent (%s)",
			next.Ceiling.USD(), want.USD())
	}
	if n := len(f.runner.starts()); n != 2 {
		t.Errorf("%d units started in all, want the original and the restart", n)
	}
	if acct := f.account(t); acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree after the restart", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// …and the same exit code with an intent recorded is the user's stop, which is never restarted: the
// money would go on work its owner had just cancelled.
func TestAGracefulSignalWeAskedForEndsTheRun(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 30, Spend: usd(900_000), Reserved: usd(0)}, nil)
	if _, err := f.svc.Stop(f.ctx, "u1", l.RunID); err != nil {
		t.Fatal(err)
	}
	ends(t, f, l, exitedWith("5"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := card(t, f); got.Status != "stopped" {
		t.Fatalf("a stop the user asked for ended as %q", got.Status)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Errorf("%d units started: a run its owner stopped must not come back", n)
	}
}

// A marker that EXISTS and does not parse used to fail the reconciliation of its run on every sweep,
// for good: the write is atomic, so nothing this platform does produces one, and nothing it does
// repairs one either (register row PD-164). The run is ended with the fact recorded instead.
//
// Mutation caught: mapping a decode failure back onto a plain error return.
func TestAnUnreadableExitMarkerEndsTheRunInsteadOfWedgingIt(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	path := f.svc.markerPath(l.RunID, l.AttemptNo)
	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil {
		t.Fatal(err)
	}
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 10, Spend: usd(300_000), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	got := card(t, f)
	if got.Status != "failed" || got.FinishedAt == nil {
		t.Fatalf("a run whose marker cannot be read is %q (finished: %v)", got.Status, got.FinishedAt)
	}
	// The machine's word and this platform's conclusion have to be tellable apart by an operator.
	var result string
	if err := f.store.Pool().QueryRow(f.ctx,
		`select exit_result from run_attempts where id = $1`, l.AttemptID).Scan(&result); err != nil {
		t.Fatal(err)
	}
	if result != runner.UnreadableMarkerResult {
		t.Errorf("exit_result %q, want %q", result, runner.UnreadableMarkerResult)
	}
	// And the money is resolved rather than left holding: the whole cost of the old behaviour was a
	// hold that no sweep could ever close.
	if acct := f.account(t); acct.Reserved != 0 {
		t.Errorf("the hold is still open: %s", acct.Reserved.USD())
	}
}

// The same corrupted marker over a run whose STREAM said a ceiling stopped it — and the stream is
// then the only witness there is, since a marker that does not parse carries no exit code either.
// Closing such a run `failed` with no reason contradicts the contract twice over: the stop is
// resumable, and the screen is left with a run that ended for no stated reason (acceptance dofix
// ФП-4).
//
// Mutation caught: re-reading the paused reason only on the branch where the marker parses.
func TestACeilingSurvivesAMarkerThatCannotBeRead(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	body := hello(t, f) +
		`{"seq":2,"type":"ceiling","data":{"halted":true,"scope":"day"}}` + "\n" +
		`{"seq":3,"type":"finished","data":{"outcome":"ceiling"}}` + "\n"
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	path := f.svc.markerPath(l.RunID, l.AttemptNo)
	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil {
		t.Fatal(err)
	}
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 10, Spend: usd(300_000), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	got := card(t, f)
	if got.Status != "paused" || got.PausedReason != pgstore.PausedDailyCeiling {
		t.Fatalf("the run ended as %q/%q, want paused/%s", got.Status, got.PausedReason, pgstore.PausedDailyCeiling)
	}
	// The unreadable marker is still recorded as what it was: the ceiling decides the STATUS, not
	// what an operator gets to know about the marker.
	var result string
	if err := f.store.Pool().QueryRow(f.ctx,
		`select exit_result from run_attempts where id = $1`, l.AttemptID).Scan(&result); err != nil {
		t.Fatal(err)
	}
	if result != runner.UnreadableMarkerResult {
		t.Errorf("exit_result %q, want %q", result, runner.UnreadableMarkerResult)
	}
}

// WHICH build is asked about a book, after its run is over, is a money question and not a detail.
//
// A run started before an engine deploy and finished after it leaves its project file in the schema
// its own build wrote. The currently deployed binary refuses a file older than itself — that refusal
// is the whole of the v15 deadlock — so a settlement that asked the CONFIGURED path would never get
// the committed figure, and the hold would stay reserved with no sweep able to close it. The attempt
// records the path it ran with (row 139) and every question about that attempt goes back to it.
//
// Mutation caught: `s.Engine.Status(ctx, s.Cfg.EngineBinary, …)` in settle — and in bookMeter and
// maybeResync, which are the same question asked earlier.
func TestTheMoneyOfAFinishedRunIsReadWithTheBuildItRanWith(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	const pinned = "/opt/engine/2026.08.01/tmctl"
	if l.EngineBinary != pinned {
		t.Fatalf("the attempt was pinned to %q", l.EngineBinary)
	}
	// The deployment moves on while the run is still going, exactly as an engine upgrade does.
	f.svc.Cfg.EngineBinary = "/opt/engine/2026.08.14/tmctl"
	ends(t, f, l, exitedWith("0"))
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 100, Spend: usd(1_000_000), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	asked := f.engine.askedAs()
	if len(asked) == 0 {
		t.Fatal("the engine was never asked about the finished run")
	}
	for _, bin := range asked {
		if bin != pinned {
			t.Errorf("the engine was asked as %q; every question about this attempt goes to %q", bin, pinned)
		}
	}
	// And the settlement actually happened, so the assertion above is about a path that ran.
	if acct := f.account(t); acct.Reserved != 0 {
		t.Errorf("the hold is still open: %s", acct.Reserved.USD())
	}
}

// A ceiling the stream announced during THIS pass, a marker that does not parse — and a projection
// the reconciler cannot re-read. Nothing may be decided from that.
//
// The corrupt-marker branch is where it costs the most: it carries no exit code to say `ceiling`
// independently, and it CLOSES the run by design (PD-164), so there is no next pass to correct it.
// A re-read that answered "" on failure therefore turned an ordinary Postgres blip into a resumable
// ceiling halt recorded as `failed` with no reason at all. Found by the state-machine lens of the
// dofix review, on live Postgres.
//
// What the mutation moves is WHICH decision the reconciler refuses to make: with the database gone
// nothing can be written either way, so the assertion is on the refusal itself.
//
// Mutation caught: freshPausedReason falling back to the snapshot instead of reporting the failure.
func TestAnEndingIsNeverDecidedFromAReadThatFailed(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	body := hello(t, f) +
		`{"seq":2,"type":"ceiling","data":{"halted":true,"scope":"day"}}` + "\n"
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	// The drain of this very pass materializes the ceiling; the marker cannot say anything about it.
	if _, err := f.svc.reconcile(f.ctx, l); err != nil {
		t.Fatal(err)
	}
	path := f.svc.markerPath(l.RunID, l.AttemptNo)
	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil {
		t.Fatal(err)
	}
	// …and now the database goes away before the pass can establish what the projection says.
	f.store.Pool().Close()
	_, err := f.svc.reconcile(f.ctx, l)
	if err == nil {
		t.Fatal("a pass that could not read the run's projection reported success")
	}
	if !strings.Contains(err.Error(), "paused reason") {
		t.Fatalf("the ending was decided past the read that failed: %v", err)
	}
	// The run is still there to be reconciled by the next pass — read with a connection this test did
	// not break.
	store, err := pgstore.Open(f.ctx, f.dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer store.Close()
	var finished *time.Time
	if err := store.Pool().QueryRow(f.ctx, `select finished_at from runs where id = $1`, l.RunID).Scan(&finished); err != nil {
		t.Fatal(err)
	}
	if finished != nil {
		t.Errorf("the run was closed on a fact the pass could not establish (finished %v)", finished)
	}
}

// engineScript writes a `tmctl` whose whole behaviour is two files the test rewrites between calls:
// what it prints and the code it exits with. Used where the property under test is a disposition of
// the SHELL contract, which the stub engine cannot have — it answers in Go values, on the far side
// of the exit code.
func engineScript(t *testing.T) (bin string, answers func(report string, code int)) {
	t.Helper()
	dir := t.TempDir()
	report, code := filepath.Join(dir, "report.json"), filepath.Join(dir, "code")
	bin = filepath.Join(dir, "tmctl")
	script := "#!/bin/sh\ncat '" + report + "'\nexit \"$(cat '" + code + "')\"\n"
	if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
		t.Fatal(err)
	}
	answers = func(r string, c int) {
		t.Helper()
		if err := os.WriteFile(report, []byte(r), 0o600); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(code, []byte(strconv.Itoa(c)+"\n"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	answers(`{"committed_usd":0,"reserved_usd":0}`, 0)
	return bin, answers
}

// The money of a book that flagged a unit, through the REAL engine channel and the real exit
// contract (acceptance dofix ФП-1).
//
// This is the ordinary path, not a corner: the engine exits 2 from `status --json` for any book with
// a flagged unit, and the platform asks that same command for the committed figure of every run that
// ends. Reading the code as a refusal deferred the settlement forever — the hold stayed reserved,
// the account could not start another run of that book, and nothing in the read model said why. The
// stub engine cannot catch that class at all, so this test runs a process.
//
// Mutation caught: any non-zero exit read as an error in runner.Status.
func TestTheMoneyOfAFlaggedBookIsSettledThroughTheRealExitContract(t *testing.T) {
	f := newFixture(t, "10", 500)
	bin, answers := engineScript(t)
	f.svc.Engine = runner.New(nil)
	f.svc.Cfg.EngineBinary = bin

	l := spawned(t, f, 100) // the meter reads zero on a book that has never run
	hold := fixtureHold(100)
	if got := f.account(t); got.Reserved != hold {
		t.Fatalf("the hold was not taken: reserved %s", got.Reserved.USD())
	}
	// The run finished and one unit needs a human: exit 2 from the unit AND from every read of the
	// book afterwards.
	const spent = 1_250_000
	answers(`{"book_id":"gu","total_units":10,"done":9,"flagged":1,"committed_usd":1.25,"reserved_usd":0}`, 2)
	ends(t, f, l, exitedWith("2"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}

	if got := card(t, f); got.Status != "ready" {
		t.Fatalf("a completed run with flagged units ended as %q", got.Status)
	}
	acct := f.account(t)
	if acct.Reserved != 0 {
		t.Fatalf("the hold of a flagged run is still open: %s", acct.Reserved.USD())
	}
	// Twice, from two independent facts: the balance row the account screen reads, and the ledger it
	// is derived from. A settlement that wrote one without the other would leave the account able to
	// spend money it has already spent.
	granted, err := money.ParseUSD("10")
	if err != nil {
		t.Fatal(err)
	}
	if want := granted - spent; acct.Balance != want {
		t.Errorf("balance %s, want %s", acct.Balance.USD(), want.USD())
	}
	var ledger money.MicroUSD
	if err := f.store.Pool().QueryRow(f.ctx,
		`select coalesce(sum(amount_micro_usd), 0) from credit_ledger where user_id = 'u1'`).Scan(&ledger); err != nil {
		t.Fatal(err)
	}
	if ledger != acct.Balance {
		t.Errorf("the ledger sums to %s and the balance says %s", ledger.USD(), acct.Balance.USD())
	}
}

// Two generations of the sweep on one run, which is what an overlapping deploy produces: one pass
// closes it while the other is still holding a snapshot that says it is live.
//
// The stale pass must find NOTHING to do. Re-opening would clear `finished_at`, take a second hold
// and start an engine for a run whose owner has already been told it ended — the class of PD-181,
// reached through the branch this pack added (exit 5 with no recorded intent restarts a run).
//
// Mutation caught: dropping OnlyIfLive from the reconciler's RestartInput.
func TestAStalePassDoesNotResurrectARunAnotherPassHasFinished(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 30, Spend: usd(900_000), Reserved: usd(0)}, nil)
	// The snapshot this pass will decide from — taken while the run was live and BEFORE its owner
	// pressed stop. That is the whole window: everything below happens inside it.
	stale := f.live(t)

	// The user stops the run, and another generation of the sweep closes it as stopped.
	if _, err := f.svc.Stop(f.ctx, "u1", l.RunID); err != nil {
		t.Fatal(err)
	}
	ends(t, f, l, exitedWith("5"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	before := card(t, f)
	if before.Status != "stopped" {
		t.Fatalf("the first pass ended the run as %q", before.Status)
	}
	spawns := len(f.runner.starts())

	// …and now the stale pass acts on its own snapshot. Driven through reconcile directly, because
	// that is exactly what the second generation would call with the list it read before.
	if _, err := f.svc.reconcile(f.ctx, stale); err != nil {
		t.Fatalf("the stale pass failed instead of doing nothing: %v", err)
	}
	after := card(t, f)
	if after.Status != before.Status || after.FinishedAt == nil {
		t.Fatalf("a stale pass re-opened a run another pass had finished: %q → %q (finished: %v)",
			before.Status, after.Status, after.FinishedAt)
	}
	if n := len(f.runner.starts()); n != spawns {
		t.Errorf("the stale pass started %d more engines", n-spawns)
	}
	acct := f.account(t)
	if acct.Reserved != 0 {
		t.Errorf("the stale pass took a second hold: %s reserved", acct.Reserved.USD())
	}
	if acct.Balance != acct.LedgerSum {
		t.Errorf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
	}
}

// The refusal band: the engine was turned down before it did any work, so nothing was spent and the
// hold comes back. `failed` for the run — it cannot proceed — and never a restart, which would meet
// the same answer on every sweep for as long as the deployment stays broken.
func TestARefusedInvocationEndsTheRunAndGivesTheWholeHoldBack(t *testing.T) {
	f := newFixture(t, "10", 500)
	l := spawned(t, f, 100)
	ends(t, f, l, exitedWith("10")) // the configuration will not load
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 0, Spend: usd(0), Reserved: usd(0)}, nil)
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	if got := card(t, f); got.Status != "failed" {
		t.Fatalf("a refused invocation ended as %q", got.Status)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Errorf("%d units started: a refusal is reproducible, so a restart is a loop", n)
	}
	acct := f.account(t)
	if acct.Reserved != 0 {
		t.Errorf("the hold is still open after a refusal: %s", acct.Reserved.USD())
	}
	if acct.Balance != money.MicroUSD(10_000_000) {
		t.Errorf("balance %s: a refusal spent nothing and must cost nothing", acct.Balance.USD())
	}
}

// A run whose STREAM already reported a ceiling, whose unit then vanishes without a marker — a
// reboot in the seconds between the engine's last line and its exit.
//
// It must be closed as the pause it already is, and NOT restarted: the engine stopped on purpose and
// the next process would meet the same limit. The second half is the one the adversarial review of
// this pack found: the restart path's own pause hard-codes `credit_exhausted`, so a run stopped by
// the ENGINE's daily ceiling used to come back wearing the ACCOUNT's reason — which lights the
// account-level halted flag and re-opens the resume the day-ceiling guard exists to refuse.
//
// Mutation caught: removing the PausedReason branch from reconcile's no-marker path, and
// hard-coding PausedCreditExhausted in restart's exhausted branch.
func TestACeilingThatTheStreamReportedSurvivesAUnitThatVanished(t *testing.T) {
	f := newFixture(t, "10", 500)
	spawned(t, f, 100)
	journal := filepath.Join(f.workdir, ingest.JournalFile)
	body := hello(t, f) + `{"seq":2,"type":"ceiling","data":{"halted":true,"scope":"day"}}` + "\n"
	if err := os.WriteFile(journal, []byte(body), 0o600); err != nil {
		t.Fatal(err)
	}
	// No marker at all, and long enough ago that the spawn grace has expired: the shape of a reboot.
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 5, Spend: usd(100_000), Reserved: usd(0)}, nil)
	f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	got := card(t, f)
	if got.Status != "paused" || got.PausedReason != pgstore.PausedDailyCeiling {
		t.Fatalf("the run ended as %q/%q, want paused/%s", got.Status, got.PausedReason, pgstore.PausedDailyCeiling)
	}
	if n := len(f.runner.starts()); n != 1 {
		t.Errorf("%d units started: a run the engine stopped on purpose must not be restarted", n)
	}
	if acct := f.account(t); acct.Reserved != 0 {
		t.Errorf("the hold is still open: %s", acct.Reserved.USD())
	}
}

// The stream is named when the ATTEMPT ROW is written, not when its unit is — so there is no window
// in which a drain has no name of its own and adopts whatever handshake it meets. Between admission
// and spawn the reconciler really does drain (an admitted attempt is in ListLiveRuns), and that
// window is seconds of a `tmctl status` call or a whole sweep interval.
//
// Mutation caught: moving EngineStreamID back out of the attempt INSERTs into RecordSpawn.
func TestAnAttemptIsNamedBeforeAnythingCanBeSpawnedForIt(t *testing.T) {
	f := newFixture(t, "10", 500)
	run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
	if err != nil {
		t.Fatal(err)
	}
	// Nothing has been spawned yet — this is exactly the window.
	if n := len(f.runner.starts()); n != 0 {
		t.Fatalf("%d units started before the assertion", n)
	}
	live := f.live(t)
	if live.EngineRunID != pgstore.EngineStreamID(live.RunID, live.AttemptNo) {
		t.Fatalf("an admitted attempt carries engine_run_id %q; a drain in this window would adopt a stranger's stream", live.EngineRunID)
	}
	// …and a restarted attempt is named the same way, in the same statement that creates it.
	if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
		t.Fatal(err)
	}
	f.engine.set(ingest.StatusReport{TotalUnits: 100, Done: 1, Spend: usd(10_000), Reserved: usd(0)}, nil)
	ends(t, f, f.live(t), exitedWith("5"))
	if err := f.svc.Sweep(f.ctx); err != nil {
		t.Fatal(err)
	}
	next := f.live(t)
	if next.AttemptNo != 2 {
		t.Fatalf("the run was not restarted: attempt %d", next.AttemptNo)
	}
	if next.EngineRunID != pgstore.EngineStreamID(next.RunID, 2) {
		t.Errorf("attempt 2 carries engine_run_id %q", next.EngineRunID)
	}
}