package pipeline

import (
	"bufio"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"textmachine/backend/internal/chunk"
	"textmachine/backend/internal/llm"
	"textmachine/backend/internal/obs"
	"textmachine/backend/internal/store"

	_ "modernc.org/sqlite"
)

// cutcall_test.go: the nine outcomes of a call, walked on a real provider socket, with the CALL COUNT
// taken from the server and a paired assertion about what the RESUME does.
//
// ⛔ WHY THE COUNT COMES FROM THE SERVER. A client-side counter counts intentions; only the server
// knows how many generations were actually asked for, and «how many times did we pay for this» is the
// whole question. Every count below is `srv.calls()`.
//
// ⛔ AND WHY THREE «BEFORE HEADERS» ROWS COST NOTHING. The money is drawn on the provider's own
// acknowledgement — a 2xx object in our hands — and not on our write: bytes entering the peer's TCP
// window say nothing about the application behind it, and settling for that charges a reader for a call
// nobody ran. Those rows keep their class, their flag and their resume; only the number is zero. On the
// shipping provider nothing is lost by it, because DeepSeek answers 200 on acceptance, so every self-cut
// there is post-header and paid.
//
// ⛔ AND WHY EVERY LOST-BUT-DELIVERED ROW HAS A RESUME ASSERTION. This design can fail in two
// directions and only one of them is loud. Re-buying a call we already paid for is the defect it
// fixes; NEVER re-doing an interrupted call is the same construction failing the other way, and it
// would leave every test here green while quietly turning a stopped run into permanent data loss. So
// each row states both: what the first run cost, and what the second run calls.

// --- the fixture provider ---

// cutServer is an OpenAI-compatible endpoint whose behaviour is chosen per REQUEST NUMBER, so one
// fixture can cut the first call and answer the second — which is exactly the shape a resume test needs.
type cutServer struct {
	mu      sync.Mutex
	n       int
	bodies  []string
	behave  func(i int, w http.ResponseWriter, r *http.Request) // i is 1-based
	arrived chan struct{}                                       // closed when the FIRST request lands
	once    sync.Once
	// flushed is closed when a handler FLUSHES its first response byte — the server side of the «a reply
	// came back» boundary two of the nine rows stand on. See flushWatch.
	flushed   chan struct{}
	flushOnce sync.Once
	srv       *httptest.Server
}

// flushWatch wraps the handler's ResponseWriter so the FIRST flush says so on a channel.
//
// ⛔ IT EXISTS BECAUSE ONE ROW'S BOUNDARY WAS ARRANGED BY HOPE. «The stop landed AFTER a response byte
// came back» was built by sleeping 50 ms from the request's ARRIVAL — a window that had to cover the
// handler being scheduled, the header write, the flush and the client's read — and it missed 2 runs in 8
// under load. It then failed on the MONEY, three steps from the cause, which is how a fixture that did
// not build its own situation reads as a defect in the code.
//
// Flush and Hijack are forwarded EXPLICITLY: wrapping a ResponseWriter hides the optional interfaces, and
// three of the nine rows hijack the connection — a hijack that stopped working would turn them into
// silent passes, which is worse than the flake this removes.
type flushWatch struct {
	http.ResponseWriter
	cs *cutServer
}

func (f flushWatch) Flush() {
	f.ResponseWriter.(http.Flusher).Flush()
	f.cs.flushOnce.Do(func() { close(f.cs.flushed) })
}

func (f flushWatch) Hijack() (net.Conn, *bufio.ReadWriter, error) {
	return f.ResponseWriter.(http.Hijacker).Hijack()
}

func newCutServer(t *testing.T, behave func(i int, w http.ResponseWriter, r *http.Request)) *cutServer {
	t.Helper()
	cs := &cutServer{behave: behave, arrived: make(chan struct{}), flushed: make(chan struct{})}
	cs.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		body, _ := io.ReadAll(r.Body)
		cs.mu.Lock()
		cs.n++
		i := cs.n
		cs.bodies = append(cs.bodies, string(body))
		cs.mu.Unlock()
		cs.once.Do(func() { close(cs.arrived) })
		cs.behave(i, flushWatch{w, cs}, r)
	}))
	t.Cleanup(cs.srv.Close)
	return cs
}

func (c *cutServer) calls() int { c.mu.Lock(); defer c.mu.Unlock(); return c.n }
func (c *cutServer) allBodies() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return append([]string(nil), c.bodies...)
}

// hold blocks the handler until the client walks away — with a bound of its own, because
// httptest.Close waits for outstanding handlers and a handler waiting for a disconnect nobody has
// noticed yet deadlocks the test. A hang has no colour: neither the battery nor a mutation reports one.
func hold(r *http.Request) {
	select {
	case <-r.Context().Done():
	case <-time.After(4 * time.Second):
	}
}

func answerWhole(w http.ResponseWriter) {
	w.Header().Set("Content-Type", "application/json")
	fmt.Fprint(w, `{"id":"fake","model":"fake-model","choices":[{"message":{"content":"Тихое утро."},"finish_reason":"stop"}],
		"usage":{"prompt_tokens":1000,"completion_tokens":500}}`)
}

// dropAfterHeaders sends a 2xx and part of a body, then kills the socket: a connection that broke with
// the provider mid-answer and both deadlines still alive.
func dropAfterHeaders(t *testing.T, w http.ResponseWriter) {
	t.Helper()
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	fmt.Fprint(w, `{"id":"fake","choices":[`)
	w.(http.Flusher).Flush()
	hijackClose(t, w)
}

// dropBeforeHeaders kills the socket with no status line at all — the request was delivered, nothing
// came back.
func dropBeforeHeaders(t *testing.T, w http.ResponseWriter) {
	t.Helper()
	hijackClose(t, w)
}

func hijackClose(t *testing.T, w http.ResponseWriter) {
	t.Helper()
	hj, ok := w.(http.Hijacker)
	if !ok {
		t.Error("the fixture needs a hijackable response writer")
		return
	}
	conn, _, err := hj.Hijack()
	if err != nil {
		t.Error(err)
		return
	}
	conn.Close()
}

// early200 is the vendor-documented «accepted, still queued» reply: a 200 whose body is empty lines.
func early200(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	fmt.Fprint(w, "\n")
	w.(http.Flusher).Flush()
	hold(r)
}

// deadTLS is a socket that completes the TCP connection and then says nothing. A client speaking TLS
// to it blocks in the handshake, so the request is NEVER WRITTEN.
//
// ⛔ THIS IS THE FIXTURE THE WHOLE FILE IS MOST AT RISK FROM. «Delivered» and «not delivered» are one
// bit apart, and the easy way to build the second is an HTTP handler that sleeps — which RECEIVES the
// request first and is therefore the FIRST case wearing the second's name. Every row here would then
// pass while the boundary they all stand on was never tested at all.
func deadTLS(t *testing.T) string {
	t.Helper()
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	go func() {
		var held []net.Conn
		for {
			c, err := ln.Accept()
			if err != nil {
				for _, h := range held {
					h.Close()
				}
				return
			}
			held = append(held, c)
		}
	}()
	t.Cleanup(func() { ln.Close() })
	return "https://" + ln.Addr().String()
}

// --- the project ---

// setupCutProject writes a ONE-STAGE book so a call count is unambiguous: every request the server
// sees is this chunk's draft, and nothing else is buying anything.
//
// tok_s_floor is set absurdly high on purpose. These rows are about the CUT, not about the deadline
// arithmetic — the derivation is pinned in internal/llm — and left at the vendor default every fixture
// would wait its derived deadline instead of the second attempt_s asks for.
func setupCutProject(t *testing.T, dir, providerURL string) string {
	t.Helper()
	writeFile(t, filepath.Join(dir, "prompts", "translator.md"),
		"Переводи с {{source_lang}} на {{target_lang}}.\n---USER---\n{{text}}")
	writeFile(t, filepath.Join(dir, "models.yaml"), fmt.Sprintf(`
prices_checked: %q
default_model: fake-model
providers:
  fake:
    kind: openai
    base_url: %q
    timeouts: { attempt_s: 1, max_attempts: 3, backoff_cap_s: 1, tok_s_floor: 1000000 }
models:
  fake-model:
    provider: fake
    price: { input_per_m: 1.0, cached_per_m: 0.1, cache_write_per_m: 0, output_per_m: 2.0 }
`, time.Now().UTC().Format("2006-01-02"), providerURL))
	writeFile(t, filepath.Join(dir, "pipeline.yaml"), `
core: C1
version: 1
defaults: { max_output_ratio: 2.0, min_max_tokens: 512 }
retries: { regenerate_before_escalate: 0 }
stages:
  - { name: draft, role: translator, model: fake-model, prompt_override: prompts/translator.md, prompt_version: v-test, temperature: 0.3, reasoning: "off" }
`)
	writeFile(t, filepath.Join(dir, "source.txt"), "静かな朝。")
	writeFile(t, filepath.Join(dir, "book.yaml"), `
book_id: test-book
title: Тест
source_lang: ja
target_lang: ru
genre: ранобэ
audience: тест
venuti: 0.5
honorifics: keep
transcription: polivanov
footnotes: minimal
pipeline: pipeline.yaml
models: models.yaml
source_file: source.txt
ceilings: { book_usd: 5.0, day_usd: 10.0 }
`)
	return filepath.Join(dir, "book.yaml")
}

// runOnce opens a runner over the project and translates, returning the error verbatim (a cut run
// FAILS on purpose for two of the causes, and swallowing that would hide the pause).
func runOnce(t *testing.T, ctx context.Context, bookPath string) error {
	t.Helper()
	r, err := NewRunner(bookPath, obs.NewLogger())
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()
	_, terr := r.TranslateBook(ctx)
	return terr
}

// money reads the ledger for the book — the raw spend row and the checkpoints behind it, never an
// aggregate somebody else derived.
type money struct {
	committed, reserved float64
	checkpoints         []store.CheckpointUsage
	statuses            []store.ChunkStatus
	estRows             int
	estUSD              float64
}

func readMoney(t *testing.T, bookPath string) money {
	t.Helper()
	r, err := NewReadOnlyRunner(bookPath, obs.NewLogger())
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()
	var m money
	if m.committed, m.reserved, err = r.Store.SpentUSD(r.Book.BookID); err != nil {
		t.Fatal(err)
	}
	if m.checkpoints, err = r.Store.CheckpointUsageForBook(r.Book.BookID); err != nil {
		t.Fatal(err)
	}
	if m.statuses, err = r.Store.ChunkStatusesForBook(r.Book.BookID); err != nil {
		t.Fatal(err)
	}
	m.estRows, m.estUSD = estimatedSpend(m.checkpoints, m.committed)
	return m
}

// --- the nine ---

// cutRow is one outcome: how the provider misbehaves, what the FIRST run must cost and mark, and what
// the RESUME must call.
type cutRow struct {
	name string
	// behave drives the first run. Request numbering is per-server, so `i` is how many calls this
	// fixture has already taken.
	behave func(t *testing.T, i int, w http.ResponseWriter, r *http.Request)
	// cancelOnArrival stops the run as soon as the provider has the request — the «a person pressed
	// stop» cause, which cannot be produced by a handler.
	cancelOnArrival bool
	// notDelivered points the first run at a socket that never reads the request.
	notDelivered bool

	wantFirstCalls int  // asked of the SERVER
	wantPaid       bool // did the first run book money for it
	// wantNoStatus: an infra pause resolves nothing and must leave NO chunk_status row. It is its own
	// field rather than an empty wantFlag: reasonOK is itself the empty string, so overloading the zero
	// value would make «the chunk shipped» and «the chunk was never marked» the same expectation — and
	// the ok row would then pass while asserting the opposite of what it means.
	wantNoStatus   bool
	wantFlag       FlagReason
	wantDisp       Disposition
	wantEstimated  bool // is the money an ESTIMATE — asserted on BOTH sides, so the flag cannot be constant
	wantResumeCall int  // fresh calls the SECOND run makes
	// wantSameBudget: the re-done call must carry the max_tokens of the one that was cut — a doubling
	// here would buy twice the call for a health nobody lost.
	wantSameBudget bool
	// assertBoundary says this row's money stands on WHICH SIDE of the «a response byte came back»
	// boundary the stop landed, so the row asserts the cut's own evidence and PRINTS it instead of
	// letting a money mismatch three steps away speak for a fixture that missed its moment.
	assertBoundary bool
	// wantAfterHeaders is that side. It also decides what the canceller waits for: a row that must stop
	// AFTER the boundary waits for the handler's own flush rather than for a sleep to have covered it.
	wantAfterHeaders bool
}

func TestTheNineOutcomesOfACall(t *testing.T) {
	rows := []cutRow{{
		name: "after 2xx · whole body, JSON ok",
		behave: func(_ *testing.T, _ int, w http.ResponseWriter, _ *http.Request) {
			answerWhole(w)
		},
		wantFirstCalls: 1, wantPaid: true, wantFlag: reasonOK, wantDisp: DispOK, wantEstimated: false, wantResumeCall: 0,
	}, {
		name: "after 2xx · whole body, JSON broken",
		behave: func(_ *testing.T, i int, w http.ResponseWriter, _ *http.Request) {
			w.Header().Set("Content-Type", "application/json")
			fmt.Fprint(w, `{"id":"fake","choices":`) // complete read, undecodable content
		},
		// The billed-decode cap: one re-bill and no more. Unchanged by this pack — its read error is
		// nil, so it is the SECOND row of the table and not a cut at all.
		wantFirstCalls: 2, wantPaid: true, wantFlag: FlagDecodeError, wantDisp: DispFlagged, wantEstimated: true, wantResumeCall: 0,
	}, {
		name: "after 2xx · self-cut (early 200, then our deadline)",
		behave: func(_ *testing.T, _ int, w http.ResponseWriter, r *http.Request) {
			early200(w, r)
		},
		wantFirstCalls: 1, wantPaid: true, wantFlag: FlagAttemptTimeout, wantDisp: DispFlagged,
		wantEstimated: true, wantResumeCall: 0,
	}, {
		name: "after 2xx · cancelled by the operator",
		behave: func(_ *testing.T, _ int, w http.ResponseWriter, r *http.Request) {
			early200(w, r)
		},
		cancelOnArrival: true,
		wantFirstCalls:  1, wantPaid: true, wantFlag: FlagCancelled, wantDisp: DispFlagged,
		wantEstimated: true, wantResumeCall: 1, wantSameBudget: true,
		assertBoundary: true, wantAfterHeaders: true,
	}, {
		name: "after 2xx · connection lost",
		behave: func(t *testing.T, _ int, w http.ResponseWriter, _ *http.Request) {
			dropAfterHeaders(t, w)
		},
		// One retry, keyed on DELIVERY rather than on a 2xx, then an infra pause: no chunk_status row.
		wantFirstCalls: 2, wantPaid: true, wantNoStatus: true, wantEstimated: true, wantResumeCall: 1, wantSameBudget: true,
	}, {
		name: "before headers · self-cut (nothing ever came back)",
		behave: func(_ *testing.T, _ int, w http.ResponseWriter, r *http.Request) {
			hold(r) // delivered; not one response byte
		},
		wantFirstCalls: 1, wantPaid: false, wantFlag: FlagAttemptTimeout, wantDisp: DispFlagged,
		wantEstimated: false, wantResumeCall: 0,
	}, {
		name: "before headers · cancelled by the operator",
		behave: func(_ *testing.T, _ int, w http.ResponseWriter, r *http.Request) {
			hold(r)
		},
		cancelOnArrival: true,
		wantFirstCalls:  1, wantPaid: false, wantFlag: FlagCancelled, wantDisp: DispFlagged,
		wantEstimated: false, wantResumeCall: 1, wantSameBudget: true,
		assertBoundary: true, wantAfterHeaders: false,
	}, {
		name: "before headers · connection lost",
		behave: func(t *testing.T, _ int, w http.ResponseWriter, _ *http.Request) {
			dropBeforeHeaders(t, w)
		},
		wantFirstCalls: 2, wantPaid: false, wantNoStatus: true, wantEstimated: false, wantResumeCall: 1, wantSameBudget: true,
	}, {
		name:         "NOT delivered · the request never went out",
		notDelivered: true,
		// Every attempt of the retry chain, because nothing was bought and there is nothing to protect.
		wantFirstCalls: 0, wantPaid: false, wantNoStatus: true, wantEstimated: false, wantResumeCall: 1,
	}}

	if len(rows) != 9 {
		t.Fatalf("the table is the contract: five outcomes after a 2xx and four before headers, got %d rows", len(rows))
	}

	for _, row := range rows {
		t.Run(row.name, func(t *testing.T) {
			dir := t.TempDir()
			// The SECOND run always gets a healthy provider: the question a resume answers is «what does
			// it call», and a fixture that kept misbehaving would answer «it fails again» instead.
			// ⚠ ATOMIC, and not a plain bool. The handler runs on the server's goroutine and may still be
			// parked in `hold` when the test flips this — a data race the detector reports as a FAILED
			// TEST with no assertion in it, which is the most confusing kind of red there is.
			var resumed atomic.Bool
			srv := newCutServer(t, func(i int, w http.ResponseWriter, r *http.Request) {
				if resumed.Load() {
					answerWhole(w)
					return
				}
				row.behave(t, i, w, r)
			})
			base := srv.srv.URL
			if row.notDelivered {
				base = deadTLS(t)
			}
			bookPath := setupCutProject(t, dir, base)

			// --- the first run ---
			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()
			if row.cancelOnArrival {
				go func() {
					<-srv.arrived
					// The request is with the provider; this is the delivered-and-stopped case.
					if row.wantAfterHeaders {
						// …and the boundary must be CROSSED first. Waiting for the handler's own flush
						// leaves only the client's read of an already-sent byte inside the sleep below,
						// instead of the whole handler in it. The bound is the fixture's own: a flush
						// that never comes must not hang the test (a hang has no colour).
						select {
						case <-srv.flushed:
						case <-time.After(4 * time.Second):
						}
					}
					time.Sleep(50 * time.Millisecond)
					cancel()
				}()
			}
			firstErr := runOnce(t, ctx, bookPath)
			firstCalls := srv.calls()
			if firstCalls != row.wantFirstCalls {
				t.Fatalf("the SERVER was asked %d time(s), want %d (first run err: %v)", firstCalls, row.wantFirstCalls, firstErr)
			}

			if row.assertBoundary {
				var cut *llm.AttemptCutError
				if !errors.As(firstErr, &cut) {
					t.Fatalf("this row's money stands on a cut call's own evidence and the run left none: %v", firstErr)
				}
				// The величина the acceptance criterion asks for: the boundary this run actually crossed,
				// printed beside the side the row is about.
				t.Logf("the boundary crossed: after_headers=%v billable=%v bytes_read=%d whitespace_only=%v elapsed=%s cause=%s",
					cut.AfterHeaders, cut.Billable, cut.BytesRead, cut.WhitespaceOnly,
					cut.Elapsed.Round(time.Millisecond), cut.Cause)
				if cut.AfterHeaders != row.wantAfterHeaders {
					t.Fatalf("the stop landed on the WRONG side of the «a reply came back» boundary "+
						"(after_headers=%v, want %v): this fixture did not build the situation the row "+
						"asserts money about, so the money check below would answer about another one",
						cut.AfterHeaders, row.wantAfterHeaders)
				}
				// And the money bit itself, at the source rather than in the ledger: `Billable` is what
				// decides whether the settle books the estimate, so the row's wantPaid must be ITS value.
				if cut.Billable != row.wantPaid {
					t.Fatalf("the cut says billable=%v and the row expects paid=%v — the ledger assertion "+
						"below would then be measuring a different call", cut.Billable, row.wantPaid)
				}
			}

			m := readMoney(t, bookPath)
			assertCutMoney(t, row, m)

			// --- the resume, on the same store and the same snapshot ---
			resumed.Store(true)
			if row.notDelivered {
				// Only the endpoint moves, and base_url is not a snapshot input (buildSnapshotID folds
				// the provider's temperature/max_tokens/model and the resolved capability, never the
				// address) — so this is a RESUME and not a re-pin. The call count below is the proof:
				// a moved snapshot would re-run everything, not one stage.
				setupCutProject(t, dir, srv.srv.URL)
			}
			before := srv.calls()
			if err := runOnce(t, context.Background(), bookPath); err != nil {
				t.Fatalf("the resume must complete: %v", err)
			}
			gotResume := srv.calls() - before
			if gotResume != row.wantResumeCall {
				t.Fatalf("the resume asked the server %d time(s), want %d — a cut call must be re-done "+
					"exactly once and a resolved one not at all", gotResume, row.wantResumeCall)
			}
			if row.wantSameBudget {
				assertSameBudget(t, srv.allBodies(), before)
			}
		})
	}
}

// assertCutMoney is the money half of every row, read from the RAW ledger.
func assertCutMoney(t *testing.T, row cutRow, m money) {
	t.Helper()
	if row.wantPaid {
		if m.committed <= 0 {
			t.Fatalf("a DELIVERED request was booked at $0 — that is the whole defect: committed=%.8f", m.committed)
		}
	} else if m.committed != 0 {
		t.Fatalf("nothing was delivered, so nothing may be booked: committed=%.8f", m.committed)
	}
	// A reservation that was neither settled nor released silently tightens the book's own ceiling.
	if m.reserved != 0 {
		t.Fatalf("the reservation must be resolved either way, got reserved=%.8f", m.reserved)
	}
	// ESTIMATED, asserted on BOTH sides. The positive half is the disclosure the owner's word came
	// with; the negative half is what keeps it from being a constant — a build that marked every paid
	// row estimated would satisfy the positive half of all eight rows and say nothing.
	if row.wantEstimated {
		if m.estRows == 0 {
			t.Fatalf("money booked without a token count must publish as ESTIMATED; estimatedSpend saw %d row(s)", m.estRows)
		}
		if diff := m.estUSD - m.committed; diff > 1e-12 || diff < -1e-12 {
			t.Fatalf("all of this book's spend is estimated, so the published estimate must equal committed: est=%.8f committed=%.8f",
				m.estUSD, m.committed)
		}
	} else if m.estRows != 0 {
		t.Fatalf("a call the provider reported usage for is not an estimate; got %d estimated row(s) worth $%.8f",
			m.estRows, m.estUSD)
	}
	// The disposition.
	if row.wantNoStatus {
		if len(m.statuses) != 0 {
			t.Fatalf("an infra pause resolves nothing, so it must leave no chunk_status row; got %d: %+v", len(m.statuses), m.statuses)
		}
		return
	}
	if len(m.statuses) != 1 {
		t.Fatalf("want exactly one chunk_status row, got %d", len(m.statuses))
	}
	cs := m.statuses[0]
	if FlagReason(cs.FlagReason) != row.wantFlag {
		t.Fatalf("flag_reason = %q, want %q — a mark that names the wrong cause sends a person hunting "+
			"for a defect that is not there", cs.FlagReason, row.wantFlag)
	}
	if Disposition(cs.Disposition) != row.wantDisp {
		t.Fatalf("disposition = %q, want %q", cs.Disposition, row.wantDisp)
	}
}

// assertSameBudget proves the re-done call asked for the SAME output budget as the one that was cut —
// the reason the attempt index and the doubling count had to come apart. `from` is how many requests
// the server had taken before the resume.
func assertSameBudget(t *testing.T, bodies []string, from int) {
	t.Helper()
	if from == 0 || len(bodies) <= from {
		t.Fatalf("need a cut request and a re-done one to compare, have %d bodies with %d before the resume", len(bodies), from)
	}
	first := maxTokensOfBody(t, bodies[0])
	redone := maxTokensOfBody(t, bodies[from])
	if first != redone {
		t.Fatalf("the re-done call must carry the budget the cut one was granted: cut asked for %d, "+
			"the re-do asked for %d — a doubling here buys twice the call for a health nobody lost",
			first, redone)
	}
}

func maxTokensOfBody(t *testing.T, body string) int {
	t.Helper()
	var b struct {
		MaxTokens int `json:"max_tokens"`
	}
	if err := json.Unmarshal([]byte(body), &b); err != nil {
		t.Fatalf("unreadable recorded request body: %v", err)
	}
	if b.MaxTokens == 0 {
		t.Fatalf("recorded body carries no max_tokens: %s", body)
	}
	return b.MaxTokens
}

// TestSettleUSDForCutCallIsTheReservationEstimate pins the money POLICY on its own, because it is the
// one number in this pack that came from the owner rather than from the code, and it must be readable
// as a decision rather than inferred from a chain of fixtures.
func TestSettleUSDForCutCallIsTheReservationEstimate(t *testing.T) {
	for _, est := range []float64{0, 0.000001, 0.13, 12.5} {
		if got := settleUSDForCutCall(est); got != est {
			t.Fatalf("a call we cut is settled at the reservation estimate (D39.230 п.1): estimate %v → %v", est, got)
		}
	}
}

// TestAfterASelfCutTheResumeCallsNobody asserts ONE thing, and the narrowness is the design.
//
// The third leak this pack closes is not the retry — it is that the old zero branch wrote no
// checkpoint at all, so a resume walked back to the provider and BOUGHT THE CALL AGAIN through a
// different door. Removing the retry without the checkpoint would have closed the class halfway.
//
// ⛔ IT CARRIES NO MONEY ASSERTION ON PURPOSE. Break the checkpoint write and a test that also checks
// the ledger goes red about the money first, and the reader concludes the mutation was caught by the
// money pin — a right verdict for the wrong reason, which is a hole rather than a catch. Here the only
// sentence a failure can print is about a fresh call.
func TestAfterASelfCutTheResumeCallsNobody(t *testing.T) {
	dir := t.TempDir()
	var resumed atomic.Bool
	srv := newCutServer(t, func(_ int, w http.ResponseWriter, r *http.Request) {
		if resumed.Load() {
			answerWhole(w)
			return
		}
		hold(r) // delivered, and we walk away
	})
	bookPath := setupCutProject(t, dir, srv.srv.URL)

	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("a self-cut is a disposition, not an infra failure: %v", err)
	}
	before := srv.calls()
	resumed.Store(true)
	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("the resume must complete: %v", err)
	}
	if fresh := srv.calls() - before; fresh != 0 {
		t.Fatalf("the resume made %d fresh provider call(s) after a self-cut — the generation we already "+
			"paid for is being bought a second time through the resume door", fresh)
	}
}

// TestANonTwoHundredIsNotAPurchase pins the ORDER inside the transport: the status line is read before
// the body's read error. A 4xx/5xx says the provider refused or failed — nothing was generated and
// nothing is owed — so a body cut short underneath one is a detail of a failure, not a purchase. Read
// the other way round, every terminal 4xx whose tiny body happened to land on the deadline would settle
// an estimate, and the money boundary would leak through the one door that is supposed to be free.
func TestANonTwoHundredIsNotAPurchase(t *testing.T) {
	dir := t.TempDir()
	srv := newCutServer(t, func(_ int, w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusInternalServerError)
		fmt.Fprint(w, `{"error":{"message":"upstream b`) // body never finishes
		w.(http.Flusher).Flush()
		hold(r)
	})
	bookPath := setupCutProject(t, dir, srv.srv.URL)

	if err := runOnce(t, context.Background(), bookPath); err == nil {
		t.Fatal("a provider failing every attempt must surface as an infra failure")
	}
	m := readMoney(t, bookPath)
	if m.committed != 0 {
		t.Fatalf("a non-2xx generated nothing and owes nothing; committed=%.8f", m.committed)
	}
	if m.reserved != 0 {
		t.Fatalf("the reservation of a failed call must be given back; reserved=%.8f", m.reserved)
	}
	if len(m.checkpoints) != 0 {
		t.Fatalf("nothing was bought, so nothing may be checkpointed; got %d", len(m.checkpoints))
	}
}

// countingHandler counts log records whose message carries a substring. A handler rather than a scan
// over a shared buffer: an assert on a substring of one big buffer is the shape that gives both false
// reds and quiet greens (D39.171), and «how many times was this said» is a question a handler answers
// exactly.
type countingHandler struct {
	n      *atomic.Int32
	needle string
	// alt counts a SECOND substring, so one fixture can assert both that a line is said and that it
	// does not say something it must not. Nil when unused.
	alt       *atomic.Int32
	altNeedle string
	// attrs collects the ATTRIBUTE KEYS of the matched records. The keys are what an operator reads
	// beside the sentence, and a key that names a per-attempt figure as the whole wait is as untrue as
	// a sentence would be — `waited=4s of=1s` was printed eighteen times before this was pinned.
	attrs *sync.Map
}

func (h countingHandler) Enabled(context.Context, slog.Level) bool { return true }
func (h countingHandler) Handle(_ context.Context, r slog.Record) error {
	if strings.Contains(r.Message, h.needle) {
		h.n.Add(1)
		if h.alt != nil && strings.Contains(r.Message, h.altNeedle) {
			h.alt.Add(1)
		}
		if h.attrs != nil {
			r.Attrs(func(a slog.Attr) bool {
				h.attrs.Store(a.Key, struct{}{})
				return true
			})
		}
	}
	return nil
}
func (h countingHandler) WithAttrs([]slog.Attr) slog.Handler { return h }
func (h countingHandler) WithGroup(string) slog.Handler      { return h }

// TestTheWaitSaysSoWhileItLastsAndIsSilentOtherwise: with a deadline derived from the budget a call
// may now legitimately run for a quarter of an hour, and the single "calling model" line then leaves
// an operator watching a silence he cannot tell from a wedged process.
//
// ⛔ BOTH HALVES, IN ONE TEST. A conditional message needs a fixture where it MUST be said and one
// where it must be SILENT; a test that asserts only the silence is vacuous — it passes on a build that
// deleted the line — and one that asserts only the speech passes on a build that says it on every
// healthy call, which is how a warning becomes something an operator scrolls past.
func TestTheWaitSaysSoWhileItLastsAndIsSilentOtherwise(t *testing.T) {
	const needle = "still waiting for the provider"
	// The second needle is the assertion the first one cannot make: the line must not state a fact the
	// runner never asked about. It has no view of the transport's trace, so «a call that has been
	// delivered» was an assertion nobody had checked — and on a request that never went out it printed
	// eighteen times, telling the operator the opposite of what happened.
	const claimsDelivery = "delivered"
	var keys sync.Map
	run := func(t *testing.T, slow bool) (said, claimed int) {
		t.Helper()
		dir := t.TempDir()
		srv := newCutServer(t, func(_ int, w http.ResponseWriter, r *http.Request) {
			if slow {
				hold(r) // held until our own deadline cuts it
				return
			}
			answerWhole(w)
		})
		bookPath := setupCutProject(t, dir, srv.srv.URL)
		var seen, claims atomic.Int32
		r, err := NewRunner(bookPath, slog.New(countingHandler{n: &seen, needle: needle, alt: &claims, altNeedle: claimsDelivery, attrs: &keys}))
		if err != nil {
			t.Fatal(err)
		}
		defer r.Close()
		if _, err := r.TranslateBook(context.Background()); err != nil {
			t.Fatalf("translate: %v", err)
		}
		return int(seen.Load()), int(claims.Load())
	}
	said, claimed := run(t, true)
	if said == 0 {
		t.Fatalf("a call the engine waited its whole deadline for said nothing while it waited — the "+
			"operator cannot tell that silence from a dead process (heartbeats seen: %d)", said)
	}
	if claimed != 0 {
		t.Fatalf("the waiting line stated %d time(s) that the call had been DELIVERED — the runner never "+
			"asked the transport, and on a request that never went out this tells the operator the "+
			"opposite of what happened", claimed)
	}
	// The KEYS the operator reads beside the sentence. This timer wraps the whole retry chain while the
	// only deadline it can name is one attempt's, so a key called `of` turned a truthful pair of numbers
	// into «waited 4s of 1s» — the hung-process signal this line exists to remove, printed by the line
	// itself. And the transport logs an `attempt` of its own, so the pipeline's must not share the key.
	for _, bad := range []string{"of", "attempt"} {
		if _, found := keys.Load(bad); found {
			t.Fatalf("the waiting line carries the key %q: a per-attempt figure presented as the whole "+
				"wait, or a number that collides with the transport's own under one name", bad)
		}
	}
	for _, want := range []string{"waited", "attempt_deadline", "stage_attempt"} {
		if _, found := keys.Load(want); !found {
			t.Fatalf("the waiting line must carry %q — without it the reader cannot tell which of the two "+
				"clocks the number belongs to", want)
		}
	}
	if said, _ = run(t, false); said != 0 {
		t.Fatalf("a call that answered at once must print no waiting line; got %d", said)
	}
}

// TestTheHeartbeatIsPacedByTheWaitItReports: the interval is a quarter of the deadline, capped at a
// minute. Derived rather than fixed, so the line exists on every call the engine is willing to wait
// for — including the short ones, which is the only reason the fixture above can observe it at all.
func TestTheHeartbeatIsPacedByTheWaitItReports(t *testing.T) {
	if got := heartbeatEvery(time.Hour); got != waitHeartbeat {
		t.Fatalf("a long wait reports at the cap, got %s", got)
	}
	if got := heartbeatEvery(4 * time.Second); got != time.Second {
		t.Fatalf("a short wait reports four times over its life, got %s", got)
	}
	// A zero or absurd deadline must not become a zero ticker — time.NewTicker panics on one, and a
	// panic inside a logging goroutine kills the run over a log line.
	for _, d := range []time.Duration{0, -time.Second, time.Nanosecond} {
		if got := heartbeatEvery(d); got <= 0 {
			t.Fatalf("deadline %s produced a non-positive interval %s — time.NewTicker panics on that", d, got)
		}
	}
}

// TestACancelledRunNeverReportsAMoneyStop drives the admission path DIRECTLY, with no wave and no
// race, because that is the only way to be sure which line is under test.
//
// ⛔ THE FIRST VERSION OF THIS TEST WAS VACUOUS AND A MUTATION SAID SO. It cancelled the context before
// TranslateBook and asserted the outcome — but the wave refuses a dead context before it dispatches
// anything (measured: 0 requests reached the server), so the reservation loop was never entered, the
// guard was never executed, and deleting the guard left the test GREEN. A test that reaches the
// situation only through a wave cannot say which of the two it is pinning.
//
// WHAT IT PINS. A refusal the ceiling cannot lift skips the wait entirely and walks to the halt. If the
// run is over by then, the truth is the cancellation and not the money: a `ceiling` event outlives any
// exit code, so the platform would record `paused` and tell a person to add money for a run that person
// had stopped themselves. Reachable at all only because a cut call is now PAID for (D39.230 п.1) — a
// stop commits every flying call's estimate, which is what can carry a book past its own ceiling.
func TestACancelledRunNeverReportsAMoneyStop(t *testing.T) {
	dir := t.TempDir()
	srv := newCutServer(t, func(_ int, w http.ResponseWriter, _ *http.Request) { answerWhole(w) })
	bookPath := setupCutProject(t, dir, srv.srv.URL)

	r, err := NewRunner(bookPath, obs.NewLogger())
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()
	// A ceiling no single call fits under: the refusal is one no settle could ever lift, which is the
	// branch that skips the wait and goes straight to the halt.
	r.CeilingUSD = 1e-9

	// The eager client set a real run precomputes before its first wave; runAttempt resolves the client
	// before any money moves, so without it the call fails on the wrong thing entirely.
	if berr := r.buildClients(); berr != nil {
		t.Fatal(berr)
	}
	st := r.Pipeline.Stages[0]
	ch := chunk.Chunk{Chapter: 1, ChunkIdx: 0, Text: "静かな朝。"}
	const snapID = "snapshot-under-test"
	// A job references a snapshot row; the id is opaque to everything this test exercises.
	if serr := r.Store.UpsertSnapshot(snapID, r.Book.BriefHash(), "{}"); serr != nil {
		t.Fatal(serr)
	}
	job, jerr := r.Store.EnsureJob(r.Book.BookID, ch.Chapter, st.Name, snapID)
	if jerr != nil {
		t.Fatal(jerr)
	}
	msgs, merr := MessagesWithInjection(r.templates[st.Name], RenderVars{Book: r.Book, Text: ch.Text}, "")
	if merr != nil {
		t.Fatal(merr)
	}

	ctx, cancel := context.WithCancel(context.Background())
	cancel() // the state every sibling of a stopped call is in
	_, aerr := r.runAttempt(ctx, st, st.Model, snapID, ch, job, 0, 512, msgs, false, true, true)

	if aerr == nil {
		t.Fatal("a refused reservation under a dead context must not report success")
	}
	var halt *CeilingHalt
	if errors.As(aerr, &halt) {
		t.Fatalf("a run that was CANCELLED reported a MONEY stop: %v", aerr)
	}
	if !errors.Is(aerr, context.Canceled) {
		t.Fatalf("the cancellation must be what leaves the call, got %T %v", aerr, aerr)
	}
	if srv.calls() != 0 {
		t.Fatalf("a refused reservation must reach no provider, got %d call(s)", srv.calls())
	}
}

// TestABurnedCheckpointIsNeverReadAsAnAnswer is the regression an adversarial pass found, and it is
// the worst thing this pack could have shipped.
//
// A burned checkpoint records MONEY and no result. `runStage` knows that from the `burned` flag — but
// three other callers of runAttempt do not read it, and the zero classification they see IS «ok». The
// escalation hop then adopted a cut call as authoritative: the chunk shipped `ok` with an EMPTY text,
// the primary flag (a content filter!) was erased, and the run after that died on its own chunk_status
// pointing at a textless checkpoint. Money spent, text lost, book unresumable — all three at once, and
// silently.
//
// The fixture: the draft comes back content-filtered (an escalatable flag), the hop is held on the wire
// and the run is stopped, then the book is resumed twice.
func TestABurnedCheckpointIsNeverReadAsAnAnswer(t *testing.T) {
	dir := t.TempDir()
	var resumed atomic.Bool
	hopSeen := make(chan struct{})
	var once sync.Once
	var srv *cutServer
	srv = newCutServer(t, func(_ int, w http.ResponseWriter, r *http.Request) {
		if strings.Contains(lastBody(srv), `"fake-hop"`) && !resumed.Load() {
			once.Do(func() { close(hopSeen) })
			hold(r) // the HOP is the call the stop catches
			return
		}
		if resumed.Load() {
			answerWhole(w)
			return
		}
		// The primary draft: a content filter — deterministic, escalatable.
		w.Header().Set("Content-Type", "application/json")
		fmt.Fprint(w, `{"id":"fake","model":"fake-model","choices":[{"message":{"content":"нет"},"finish_reason":"content_filter"}],
			"usage":{"prompt_tokens":100,"completion_tokens":10}}`)
	})
	bookPath := setupHopProject(t, dir, srv.srv.URL)

	// ⛔ SYNCHRONISED ON THE HOP'S OWN REQUEST, not on a clock. A sleep here lands on the DRAFT under
	// load and leaves the test green about a moment it never reached — clocks standing in for the thing
	// being measured is the shape D39.171 names, and this fixture had it.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		<-hopSeen
		cancel()
	}()
	_ = runOnce(t, ctx, bookPath)

	// The first resume must not ship an empty chunk as ok.
	resumed.Store(true)
	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("the first resume must complete: %v", err)
	}
	m := readMoney(t, bookPath)
	for _, cs := range m.statuses {
		if cs.Disposition == string(DispOK) {
			cp, cerr := readFinalText(t, bookPath, cs.FinalHash)
			if cerr != nil {
				t.Fatalf("an `ok` row must point at a readable checkpoint: %v", cerr)
			}
			if strings.TrimSpace(cp) == "" {
				t.Fatalf("chunk_status ch%d/chunk%d/%s shipped `ok` with an EMPTY text — a call that was "+
					"cut off was adopted as the answer", cs.Chapter, cs.ChunkIdx, cs.Stage)
			}
		}
	}
	// And the SECOND resume must not die on what the first one wrote.
	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("the second resume died on the state the first one left — the book is unresumable: %v", err)
	}
}

func lastBody(c *cutServer) string {
	b := c.allBodies()
	if len(b) == 0 {
		return ""
	}
	return b[len(b)-1]
}

func readFinalText(t *testing.T, bookPath, hash string) (string, error) {
	t.Helper()
	r, err := NewReadOnlyRunner(bookPath, obs.NewLogger())
	if err != nil {
		return "", err
	}
	defer r.Close()
	cp, cerr := r.Store.GetCheckpoint(hash)
	if cerr != nil {
		return "", cerr
	}
	if cp == nil {
		return "", fmt.Errorf("checkpoint %.12s is missing", hash)
	}
	return cp.ResponseText, nil
}

// setupHopProject is setupCutProject with an escalation hop, so a cut can land on the FALLBACK call —
// the caller that reads `cls.ok()` and never hears about `burned`.
func setupHopProject(t *testing.T, dir, providerURL string) string {
	t.Helper()
	writeFile(t, filepath.Join(dir, "prompts", "translator.md"),
		"Переводи с {{source_lang}} на {{target_lang}}.\n---USER---\n{{text}}")
	writeFile(t, filepath.Join(dir, "models.yaml"), fmt.Sprintf(`
prices_checked: %q
default_model: fake-model
providers:
  fake:
    kind: openai
    base_url: %q
    timeouts: { attempt_s: 1, max_attempts: 3, backoff_cap_s: 1, tok_s_floor: 1000000 }
models:
  fake-model:
    provider: fake
    price: { input_per_m: 1.0, cached_per_m: 0.1, cache_write_per_m: 0, output_per_m: 2.0 }
  fake-hop:
    provider: fake
    price: { input_per_m: 1.0, cached_per_m: 0.1, cache_write_per_m: 0, output_per_m: 2.0 }
`, time.Now().UTC().Format("2006-01-02"), providerURL))
	writeFile(t, filepath.Join(dir, "pipeline.yaml"), `
core: C1
version: 1
defaults: { max_output_ratio: 2.0, min_max_tokens: 512 }
retries: { regenerate_before_escalate: 0 }
stages:
  - { name: draft, role: translator, model: fake-model, prompt_override: prompts/translator.md, prompt_version: v-test, temperature: 0.3, reasoning: "off", escalate_to: fake-hop }
escalation: { budget_usd: 5.0 }
`)
	writeFile(t, filepath.Join(dir, "source.txt"), "静かな朝。")
	writeFile(t, filepath.Join(dir, "book.yaml"), `
book_id: test-book
title: Тест
source_lang: ja
target_lang: ru
genre: ранобэ
audience: тест
venuti: 0.5
honorifics: keep
transcription: polivanov
footnotes: minimal
pipeline: pipeline.yaml
models: models.yaml
source_file: source.txt
ceilings: { book_usd: 5.0, day_usd: 10.0 }
`)
	return filepath.Join(dir, "book.yaml")
}

// TestABurnIsMoneyWithoutAnAnswerAndNotJustAWord: the finish_reason string shares a namespace with
// whatever a vendor decides to print — the adapter already normalises invented values — so a provider
// answering 200 with a real translation under one of OUR names must keep its answer. A genuine burn is
// written by this engine and is always textless, so the text is part of the test and costs nothing.
//
// Both sides, because a predicate asserted in one direction says nothing: the textless rows must burn
// and the answered ones must not.
func TestABurnIsMoneyWithoutAnAnswerAndNotJustAWord(t *testing.T) {
	burns := []string{cancelledFinish, connectionLostFinish}
	for _, finish := range burns {
		if !burnedByCut(&store.Checkpoint{FinishReason: finish, ResponseText: ""}) {
			t.Fatalf("a textless %q checkpoint is money without a result and must burn", finish)
		}
		if burnedByCut(&store.Checkpoint{FinishReason: finish, ResponseText: "Тихое утро."}) {
			t.Fatalf("a %q checkpoint that CARRIES a translation is an answer, whatever the provider "+
				"called its finish reason — burning it throws the text away and buys it again", finish)
		}
	}
	// And an ordinary reply is never a burn, textless or not: an empty completion is a `empty` flag,
	// which is retryable, and reading it as money-without-a-result would change what a retry costs.
	for _, finish := range []string{"stop", "length", decodeErrorFinish, attemptTimeoutFinish} {
		if burnedByCut(&store.Checkpoint{FinishReason: finish, ResponseText: ""}) {
			t.Fatalf("%q is a verdict the resume must serve, not a key to spend: burning it re-buys the call", finish)
		}
	}
}

// TestACancelledRowIsNotAFreeResume: the volume planner asks the same question runStage asks — «does
// this row resume without a provider call» — and it used to ask it with its own copy, which did not
// know about `cancelled`. A unit whose last row is one was then counted as costing nothing, so a grant
// of one unit paid for two AND the volume report, which speaks only when something was carried, said
// nothing at all. This pins that both ends now go through the one predicate.
func TestACancelledRowIsNotAFreeResume(t *testing.T) {
	var r Runner
	draft := map[string]bool{"draft": true}
	edit := map[string]bool{}
	const snap, hash = "snap", "content"
	hashes := map[chunkKey]map[string]string{{1, 0}: {"draft": hash}}
	row := func(flag FlagReason, disp Disposition) []store.ChunkStatus {
		return []store.ChunkStatus{{
			Chapter: 1, ChunkIdx: 0, Stage: "draft",
			SnapshotID: snap, ContentHash: hash,
			Disposition: string(disp), FlagReason: string(flag),
		}}
	}
	// The control first: an ordinary resolved row IS free, so a failure below means the predicate
	// changed its answer for `cancelled` and not for everything.
	if !r.rowsResumeFree(row(reasonOK, DispOK), draft, edit, snap, "", hashes) {
		t.Fatal("a settled ok row resumes for nothing — this fixture is not measuring what it thinks")
	}
	if !r.rowsResumeFree(row(FlagAttemptTimeout, DispFlagged), draft, edit, snap, "", hashes) {
		t.Fatal("a terminal flag is never re-attacked, so it resumes for nothing")
	}
	if r.rowsResumeFree(row(FlagCancelled, DispFlagged), draft, edit, snap, "", hashes) {
		t.Fatal("a `cancelled` row records a STOP, not an answer: the resume re-does that call for money, " +
			"and counting it free lets a volume grant pay for a unit it never charged for")
	}
}

// TestMoneyWithNoCheckpointLeftIsStillAnEstimate: a redrive DELETES the checkpoints of the stages it
// re-attacks and leaves their spend committed — «after a redrive committed(spend) >= SUM(checkpoints),
// the safe direction» (store/chunkstatus.go). Derived from the surviving rows alone, the estimated
// share then falls to zero and the platform is told that money nobody can account for was MEASURED.
// The gap belongs in the estimate by the same rule everything else here does: a cost with no token
// count to justify it.
func TestMoneyWithNoCheckpointLeftIsStillAnEstimate(t *testing.T) {
	// ⚠ THE USAGE IS MARSHALLED FROM THE REAL TYPE, not hand-written. llm.Usage carries no json tags, so
	// its wire form is Go field names — and a hand-written snake_case fixture unmarshals to all zeros,
	// i.e. it silently becomes the OTHER case and the control arm certifies nothing. Caught by this test
	// failing on its own control.
	usageJSON, err := json.Marshal(llm.Usage{PromptTokens: 100, CompletionTokens: 50})
	if err != nil {
		t.Fatal(err)
	}
	measured := []store.CheckpointUsage{{CostUSD: 0.02, UsageJSON: string(usageJSON)}}
	// The control first: while the evidence is intact, a measured call is not an estimate.
	if rows, usd := estimatedSpend(measured, 0.02); rows != 0 || usd != 0 {
		t.Fatalf("a call the provider reported usage for is not an estimate, got %d row(s) / $%.8f", rows, usd)
	}
	// Now the same book after a redrive threw that checkpoint away and kept the money.
	rows, usd := estimatedSpend(nil, 0.02)
	if rows == 0 {
		t.Fatal("committed money with no checkpoint behind it is money nobody can account for; publishing " +
			"it as measured tells the platform the opposite of what is true")
	}
	if diff := usd - 0.02; diff > 1e-12 || diff < -1e-12 {
		t.Fatalf("the unaccounted gap is the whole of it, got $%.8f", usd)
	}
	// And float slack must not invent one out of a sum of prices.
	if rows, usd := estimatedSpend(measured, 0.02+1e-15); rows != 0 || usd != 0 {
		t.Fatalf("a nanodollar of float noise is not unaccounted money, got %d row(s) / $%.12f", rows, usd)
	}
}

// TestADroppedStageIsFreeEvenWhenItsRowSaysCancelled is the other direction of the volume predicate, and
// the direction the first version of that fix got wrong.
//
// The per-row walk drops the stages this pipeline no longer runs — they cannot cost anything, because
// nothing will call them. Asking «does this row resume for free» BEFORE that drop made a unit paid for a
// `cancelled` row of a stage that will never run again, so an operator who edits the pipeline over a
// stopped run would spend volume slots on nothing. Asked after, both readings are right.
func TestADroppedStageIsFreeEvenWhenItsRowSaysCancelled(t *testing.T) {
	var r Runner
	draft := map[string]bool{"draft": true}
	edit := map[string]bool{}
	const snap, hash = "snap", "content"
	hashes := map[chunkKey]map[string]string{{1, 0}: {"draft": hash}}
	row := func(stage string, flag FlagReason) []store.ChunkStatus {
		return []store.ChunkStatus{{
			Chapter: 1, ChunkIdx: 0, Stage: stage,
			SnapshotID: snap, ContentHash: hash,
			Disposition: string(DispFlagged), FlagReason: string(flag),
		}}
	}
	// The control: on a stage this pipeline STILL runs, `cancelled` costs money — that is the fix.
	if r.rowsResumeFree(row("draft", FlagCancelled), draft, edit, snap, "", hashes) {
		t.Fatal("a cancelled row of a LIVE stage is re-done for money and must take a slot")
	}
	// And on one it no longer runs, it costs nothing, because nothing will call it.
	if !r.rowsResumeFree(row("retired", FlagCancelled), draft, edit, snap, "", hashes) {
		t.Fatal("a stage this pipeline no longer runs cannot cost anything, whatever its last row says — " +
			"charging a volume slot for it spends the grant on a call that will never be made")
	}
}

// TestAStopOverAnEscalationHopLeavesAMarkAndIsRedone is the blocker an acceptance verifier measured,
// and both halves of it were invisible to everything this pack had built.
//
// A run stopped over the HOP settled the money and left NO chunk_status row: `committed=0.001176`,
// `checkpoints=2`, `chunk_status_rows=0` — the unmarked hole §4.2 forbids. And the resume made ZERO
// fresh calls, because the hop addresses a fixed attempt 0 and the burned key there is hit forever.
// The mark lived at the attempt loop's error return, and the hop leaves runStage through another one;
// the burn-walk lived in the loop, and the hop is not in it.
//
// ⛔ THE STOP IS SYNCHRONISED ON THE HOP'S OWN REQUEST, not on a clock. A sleep here would land on the
// draft under load and leave the test green about a moment it never reached (D39.171).
func TestAStopOverAnEscalationHopLeavesAMarkAndIsRedone(t *testing.T) {
	dir := t.TempDir()
	var resumed atomic.Bool
	hopSeen := make(chan struct{})
	var once sync.Once
	var srv *cutServer
	srv = newCutServer(t, func(_ int, w http.ResponseWriter, r *http.Request) {
		if resumed.Load() {
			answerWhole(w)
			return
		}
		if strings.Contains(lastBody(srv), `"fake-hop"`) {
			once.Do(func() { close(hopSeen) }) // the HOP is on the wire — now, and not on a timer
			hold(r)
			return
		}
		// The primary draft: content-filtered, which is deterministic and escalatable.
		w.Header().Set("Content-Type", "application/json")
		fmt.Fprint(w, `{"id":"fake","model":"fake-model","choices":[{"message":{"content":"нет"},"finish_reason":"content_filter"}],
			"usage":{"prompt_tokens":100,"completion_tokens":10}}`)
	})
	bookPath := setupHopProject(t, dir, srv.srv.URL)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		<-hopSeen
		cancel()
	}()
	_ = runOnce(t, ctx, bookPath)

	m := readMoney(t, bookPath)
	if m.committed <= 0 {
		t.Fatalf("the hop was delivered and cut, so it is owed: committed=%.8f", m.committed)
	}
	if len(m.statuses) != 1 {
		t.Fatalf("a stop over the HOP left %d chunk_status row(s): money is booked and the position "+
			"exports as an unexplained gap, which is exactly the invisible hole this pack forbids", len(m.statuses))
	}
	if got := FlagReason(m.statuses[0].FlagReason); got != FlagCancelled {
		t.Fatalf("the mark must name the stop, got %q", got)
	}

	// And the resume re-does it: the burned hop key is spent, the next one is asked at the same budget.
	before := srv.calls()
	resumed.Store(true)
	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("the resume must complete: %v", err)
	}
	if fresh := srv.calls() - before; fresh == 0 {
		t.Fatal("the resume made NO fresh call after a stop over the hop — the interrupted work is " +
			"never re-done and the money for it is spent, which is this construction failing the other way")
	}
}

// TestASelfCutResumesFromITSCHECKPOINTAndNotOnlyFromChunkStatus is the pin §4.3 was supposed to have and
// did not: an acceptance verifier planted both mutations that attack the checkpoint path and BOTH
// SURVIVED on a green package.
//
// The reason is the ordinary resume's own design. runStage resolves from chunk_status BEFORE any render
// (anti-wedge #1), so `TestAfterASelfCutTheResumeCallsNobody` never reaches the checkpoint at all: it
// proves the chunk_status door is shut and says nothing about the second one. Corrupt the checkpoint's
// key, or delete classify's verdict for a cut finish, and that test stays green while the engine is
// broken — in the second case broken expensively, because a cut checkpoint then reads as an EMPTY
// completion, `empty` is RETRYABLE, and the chunk is re-bought on a doubled budget.
//
// So this one enters through the door the checkpoint exists for: the torn store the fast path is
// documented against («kill -9 loses ≤1 call»). The chunk_status row is removed and the resume is asked
// what it does with the checkpoint alone.
func TestASelfCutResumesFromITSCHECKPOINTAndNotOnlyFromChunkStatus(t *testing.T) {
	dir := t.TempDir()
	var resumed atomic.Bool
	srv := newCutServer(t, func(_ int, w http.ResponseWriter, r *http.Request) {
		if resumed.Load() {
			answerWhole(w)
			return
		}
		hold(r) // delivered, and our own deadline walks away
	})
	bookPath := setupCutProject(t, dir, srv.srv.URL)

	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("a self-cut is a disposition, not an infra failure: %v", err)
	}
	before := srv.calls()
	m := readMoney(t, bookPath)
	if len(m.checkpoints) != 1 || len(m.statuses) != 1 {
		t.Fatalf("the fixture must leave exactly one checkpoint and one status row, got %d/%d",
			len(m.checkpoints), len(m.statuses))
	}

	// The torn store: the settle landed, the read-model row did not. Everything the resume can go on is
	// the checkpoint.
	dropped := dropChunkStatus(t, bookPath)
	if dropped != 1 {
		t.Fatalf("the fixture removed %d chunk_status row(s), want 1 — it is not measuring the torn store", dropped)
	}

	resumed.Store(true)
	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("the resume must complete: %v", err)
	}
	if fresh := srv.calls() - before; fresh != 0 {
		t.Fatalf("the resume made %d fresh provider call(s) with the checkpoint of a self-cut in front of "+
			"it: the generation we already paid for is being bought again through the door chunk_status "+
			"was not covering", fresh)
	}
	after := readMoney(t, bookPath)
	if diff := after.committed - m.committed; diff > 1e-12 || diff < -1e-12 {
		t.Fatalf("the resume moved the money: %.8f → %.8f", m.committed, after.committed)
	}
	if FlagReason(after.statuses[0].FlagReason) != FlagAttemptTimeout {
		t.Fatalf("the rebuilt row must carry the cut's own cause, got %q — read as an empty completion it "+
			"becomes RETRYABLE and the chunk is re-bought on a doubled budget", after.statuses[0].FlagReason)
	}
}

// dropChunkStatus deletes the book's chunk_status rows and returns how many went, leaving the
// checkpoints alone. It reaches the file directly because that is the only way to reproduce the state
// the checkpoint fast path exists for — a process killed between the settle and the read-model write.
// The redrive primitive cannot stand in: it deletes the checkpoints too, which is the opposite fixture.
func dropChunkStatus(t *testing.T, bookPath string) int64 {
	t.Helper()
	r, err := NewReadOnlyRunner(bookPath, obs.NewLogger())
	if err != nil {
		t.Fatal(err)
	}
	dbPath := r.Book.ProjectDB
	r.Close()

	db, err := sql.Open("sqlite", dbPath)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	res, err := db.Exec(`DELETE FROM chunk_status`)
	if err != nil {
		t.Fatal(err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		t.Fatal(err)
	}
	return n
}

// TestAnAcceptedSocketThatNobodyReadCostsNothing is the fixture an acceptance verifier built to expose
// the over-charge, turned into the pin for its remedy.
//
// A socket that ACCEPTS the connection and never reads a byte is what a load balancer in front of a
// dead backend looks like: our write lands in the peer's TCP window, `WroteRequest` fires, and nothing
// behind it ever sees the request. Measured before the money predicate was narrowed: 3 cancelled
// in-flight calls in 25 settled an estimate for a request no handler had entered.
//
// The class is still named — a delivered request that we cut is a `flagged(attempt_timeout)` position
// either way — and only the number is zero. Charging a reader for a call nobody ran is the one
// direction the canon forbids (D39.196 п.2а); under-counting is the direction it absorbs.
func TestAnAcceptedSocketThatNobodyReadCostsNothing(t *testing.T) {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	var accepted atomic.Int32
	go func() {
		var held []net.Conn
		for {
			c, aerr := ln.Accept()
			if aerr != nil {
				for _, h := range held {
					h.Close()
				}
				return
			}
			accepted.Add(1)
			held = append(held, c) // accepted, and not one byte is ever read from it
		}
	}()
	t.Cleanup(func() { ln.Close() })

	dir := t.TempDir()
	bookPath := setupCutProject(t, dir, "http://"+ln.Addr().String())
	_ = runOnce(t, context.Background(), bookPath)

	if n := accepted.Load(); n == 0 {
		t.Fatal("the socket was never connected to — this fixture measured nothing")
	}
	m := readMoney(t, bookPath)
	if m.committed != 0 {
		t.Fatalf("no reply ever came back, so no provider acknowledged this request; charging $%.8f for "+
			"it bills a reader for a call nobody ran", m.committed)
	}
	if m.reserved != 0 {
		t.Fatalf("the reservation must be resolved either way, got %.8f", m.reserved)
	}
	// The class is still named — the position is not a silent hole just because it cost nothing.
	if len(m.statuses) != 1 || FlagReason(m.statuses[0].FlagReason) != FlagAttemptTimeout {
		t.Fatalf("a delivered call our own deadline cut is flagged whatever it cost; got %d row(s): %+v",
			len(m.statuses), m.statuses)
	}
	if m.estRows != 0 {
		t.Fatalf("nothing was booked, so nothing is an estimate; got %d row(s)", m.estRows)
	}
}

// TestABurnedBankBatchIsNotCountedAsPaid: the bank pass gates its role sub-budget on «does a checkpoint
// exist at this batch's key». Cut calls are now settled at that key, so the answer stopped meaning what
// the gate reads it as — the batch WILL be bought again (runAttempt walks past a burned key), and a
// probe answering «already paid» lets that purchase escape the only budget bounding bank spend.
//
// Before cut calls were settled, «a checkpoint exists» and «this batch is done» were the same statement.
// This pack made them different, so the probe had to learn the difference.
func TestABurnedBankBatchIsNotCountedAsPaid(t *testing.T) {
	// The control first: an ordinary answered checkpoint IS a paid batch, or this test would pass on a
	// probe that simply always says no.
	answered := &store.Checkpoint{FinishReason: "stop", ResponseText: "термин\tterm", CostUSD: 0.01}
	if burnedByCut(answered) {
		t.Fatal("an answered batch must read as paid — the control of this fixture")
	}
	for _, finish := range []string{cancelledFinish, connectionLostFinish} {
		burned := &store.Checkpoint{FinishReason: finish, ResponseText: "", CostUSD: 0.01}
		if !burnedByCut(burned) {
			t.Fatalf("a %q checkpoint records money and no result; counting it as a paid batch lets the "+
				"re-purchase escape the role sub-budget", finish)
		}
	}
}

// TestTheBankPaidProbeSeesThroughABurnedCheckpoint pins the PROBE, not the predicate under it. The
// first version of this test asserted `burnedByCut` directly — true, and vacuous: removing the probe's
// use of it left the package green and the mutation SURVIVED.
//
// What the probe gates is the role sub-budget: «already paid» means «do not charge this batch again».
// Cut calls are settled at the batch's own key now, so a burned row there is money with no result — the
// batch WILL be bought again (runAttempt walks past it), and a probe that answers «paid» lets that
// purchase escape the only budget bounding bank spend.
func TestTheBankPaidProbeSeesThroughABurnedCheckpoint(t *testing.T) {
	dir := t.TempDir()
	srv := newCutServer(t, func(_ int, w http.ResponseWriter, _ *http.Request) { answerWhole(w) })
	bookPath := setupCutProject(t, dir, srv.srv.URL)
	r, err := NewRunner(bookPath, obs.NewLogger())
	if err != nil {
		t.Fatal(err)
	}
	defer r.Close()

	st := r.Pipeline.Stages[0]
	ch := chunk.Chunk{Chapter: 0, ChunkIdx: 0}
	msgs := []llm.Message{{Role: "user", Content: "термины"}}
	const snapID = "snapshot-under-test"
	if serr := r.Store.UpsertSnapshot(snapID, r.Book.BriefHash(), "{}"); serr != nil {
		t.Fatal(serr)
	}
	job, jerr := r.Store.EnsureJob(r.Book.BookID, ch.Chapter, st.Name, snapID)
	if jerr != nil {
		t.Fatal(jerr)
	}
	_, maxTokens := r.bankCallBudget(st.Model, msgs)
	hash := RequestHash(r.attemptRequest(st, st.Model, snapID, ch, 0, maxTokens, msgs))

	// Nothing there yet: the control, so a probe that always answered «no» would not pass this.
	if paid, perr := r.bankCheckpointExists(st, snapID, ch, msgs); perr != nil || paid {
		t.Fatalf("an empty key is not a paid batch: paid=%t err=%v", paid, perr)
	}
	write := func(finish, text string) {
		t.Helper()
		resv, verdict, rerr := r.Store.Reserve(r.Book.BookID, 0.001, store.Ceilings{BookUSD: 100, DayUSD: 100})
		if rerr != nil || verdict != store.ReserveOK {
			t.Fatalf("reserve: %v %v", verdict, rerr)
		}
		if serr := r.Store.SettleWithCheckpoint(resv, 0.001, store.Checkpoint{
			RequestHash: hash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: 0,
			Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: st.Model,
			ResponseText: text, UsageJSON: "{}", CostUSD: 0.001, FinishReason: finish,
		}, nil); serr != nil {
			t.Fatalf("settle: %v", serr)
		}
	}
	write(cancelledFinish, "")
	if paid, perr := r.bankCheckpointExists(st, snapID, ch, msgs); perr != nil || paid {
		t.Fatalf("a burned key is money with no result and the batch will be bought again; answering "+
			"«paid» lets that purchase escape the role sub-budget. paid=%t err=%v", paid, perr)
	}
	// And the other side: a real answer at the same key IS a paid batch, or the probe now says no to
	// everything and the sub-budget gate has simply been turned off.
	if derr := r.Store.PutDerivedCheckpoint(store.Checkpoint{
		RequestHash: hash + "-answered", JobID: job.ID, ChunkIdx: ch.ChunkIdx, Stage: st.Name, Role: st.Role,
		ResponseText: "термин\tterm", UsageJSON: "{}", FinishReason: "stop",
	}); derr != nil {
		t.Fatal(derr)
	}
	if !burnedByCut(&store.Checkpoint{FinishReason: cancelledFinish}) ||
		burnedByCut(&store.Checkpoint{FinishReason: "stop"}) {
		t.Fatal("the predicate under the probe must still tell the two apart")
	}
}

// TestABurnFollowedByARegenerationDoesNotOverBuy is the compound case, and it exists because a
// SURVIVING mutation said the simple one had stopped proving anything.
//
// Moving the burn walk inside runAttempt made the straightforward re-do correct BY CONSTRUCTION: the
// caller's index never advances for a burn, so «budget from the attempt index» and «budget from the
// doubling count» agree and the mutation that swaps them is a no-op. They diverge only AFTER the loop
// copies the walked index back — a burn, then a length flag, then a regeneration. That is where the
// axis split still earns its keep, and where over-buying is real: the regeneration would ask for FOUR
// times the base instead of two, and reserve four times the money with it.
//
// The sequence: a cancelled call at attempt 0 (burned key) → resume walks to attempt 1 and calls at the
// SAME budget → that answer is truncated → one regeneration at exactly TWICE the base.
func TestABurnFollowedByARegenerationDoesNotOverBuy(t *testing.T) {
	dir := t.TempDir()
	var resumed atomic.Bool
	var afterBurn atomic.Int32
	srv := newCutServer(t, func(_ int, w http.ResponseWriter, r *http.Request) {
		if !resumed.Load() {
			hold(r) // run 1: delivered, and the operator stops the run
			return
		}
		if afterBurn.Add(1) == 1 {
			// The re-done call: truncated, which is retryable and buys exactly one doubling.
			w.Header().Set("Content-Type", "application/json")
			fmt.Fprint(w, `{"id":"fake","model":"fake-model","choices":[{"message":{"content":"половина главы"},"finish_reason":"length"}],
				"usage":{"prompt_tokens":100,"completion_tokens":10}}`)
			return
		}
		answerWhole(w)
	})
	bookPath := setupRegenProject(t, dir, srv.srv.URL)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		<-srv.arrived
		cancel()
	}()
	_ = runOnce(t, ctx, bookPath)

	cutBudget := maxTokensOfBody(t, srv.allBodies()[0])
	// ⛔ THE PREMISE, and without it this fixture measures a different scenario with the same numbers. The
	// budgets below agree just as well on a run that was stopped BEFORE delivery: there is no burned key
	// then, the resume simply makes a first call and regenerates it, and every assertion still passes.
	// What makes this the compound case is that run 1 left a key holding money-or-nothing and NO result,
	// which the resume must walk past rather than replay.
	if n := len(readMoney(t, bookPath).checkpoints); n != 1 {
		t.Fatalf("premise broken: run 1 must leave exactly ONE checkpoint — the cut call's burned key — "+
			"and it left %d. Without that key the resume below is an ordinary first call and this fixture "+
			"proves nothing about the axis it exists for", n)
	}
	before := srv.calls()
	resumed.Store(true)
	if err := runOnce(t, context.Background(), bookPath); err != nil {
		t.Fatalf("the resume must complete: %v", err)
	}

	after := srv.allBodies()[before:]
	if len(after) != 2 {
		t.Fatalf("the resume must re-do the cut call and then regenerate it once: %d call(s)", len(after))
	}
	redone, regenerated := maxTokensOfBody(t, after[0]), maxTokensOfBody(t, after[1])
	if redone != cutBudget {
		t.Fatalf("the re-done call must carry the budget the cut one was granted: cut %d, re-do %d",
			cutBudget, redone)
	}
	if regenerated != 2*cutBudget {
		t.Fatalf("one regeneration is ONE doubling: want %d, got %d — keyed on the attempt index instead "+
			"of the doubling count it asks for %d and reserves the money for it",
			2*cutBudget, regenerated, 4*cutBudget)
	}
	// And the walk really happened: the burned key is still there, beside the two the resume bought.
	if n := len(readMoney(t, bookPath).checkpoints); n != 3 {
		t.Fatalf("premise broken: the burned key plus the re-done call plus its regeneration is three "+
			"checkpoints; got %d, so the resume did not walk past the burn the way this test assumes", n)
	}
}

// setupRegenProject is setupCutProject with one regeneration allowed, so a truncated answer buys a
// doubling instead of a flag.
func setupRegenProject(t *testing.T, dir, providerURL string) string {
	t.Helper()
	bookPath := setupCutProject(t, dir, providerURL)
	writeFile(t, filepath.Join(dir, "pipeline.yaml"), `
core: C1
version: 1
defaults: { max_output_ratio: 2.0, min_max_tokens: 512 }
retries: { regenerate_before_escalate: 1 }
stages:
  - { name: draft, role: translator, model: fake-model, prompt_override: prompts/translator.md, prompt_version: v-test, temperature: 0.3, reasoning: "off" }
`)
	return bookPath
}