package pipeline

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"strings"
	"time"

	"textmachine/backend/internal/config"
	"textmachine/backend/internal/ledger"
	"textmachine/backend/internal/llm"
	"textmachine/backend/internal/obs"
	"textmachine/backend/internal/store"
)

// runner.go: the Веха-2 book runner — every (chapter, chunk) from the chunker
// through the configured stage list (C1: draft→edit) with full money discipline:
// reserve → call → settle+checkpoint (одна транзакция) → request_log. A bad
// chunk is FLAGGED (disposition, chunk_status) and the loop continues (D2), not
// a run-crash; only an infra failure aborts. Resume reads chunk_status BEFORE
// rendering, so a done or terminally-flagged chunk is never re-attacked or
// re-billed. Retries walk the `attempt` axis (bigger max_tokens for the
// retryable length/empty subset, up to the regenerate cap, then flag).
// Escalation and the real linguistic chunker are later steps of Фаза 1.

// Runner executes a book's pipeline.
type Runner struct {
	Book     *config.Book
	Models   *config.Models
	Pipeline *config.Pipeline
	Store    *store.Store
	Pricer   *ledger.Pricer
	Log      *slog.Logger
	// Resnapshot re-pins existing jobs to the CURRENT snapshot when configs/
	// prompts changed since the job started (tmctl --resnapshot). Без флага
	// расхождение — громкая ошибка: смена контекста инвалидирует чекпоинты и
	// пере-оплачивает вызовы, это делается только осознанно (Р6).
	Resnapshot bool

	clients   map[string]llm.LLMClient
	templates map[string]*PromptTemplate
}

// NewRunner loads all three configs rooted at book.yaml and opens the project
// store. Caller owns Close.
func NewRunner(bookPath string, logger *slog.Logger) (*Runner, error) {
	book, err := config.LoadBook(bookPath)
	if err != nil {
		return nil, err
	}
	models, err := config.LoadModels(book.ModelsFile)
	if err != nil {
		return nil, err
	}
	pipe, err := config.LoadPipeline(book.Pipeline, models)
	if err != nil {
		return nil, err
	}
	// Fail-fast ДО открытия store (взятия flock и side-effect'ов): исполнима
	// ли механика конфига и заданы ли ключи используемых моделей.
	if err := pipe.CheckRunnable(); err != nil {
		return nil, err
	}
	if err := models.CheckKeys(pipe); err != nil {
		return nil, err
	}
	pricer, err := models.Prices()
	if err != nil {
		return nil, err
	}
	st, err := store.Open(book.ProjectDB)
	if err != nil {
		return nil, err
	}
	r := &Runner{
		Book:      book,
		Models:    models,
		Pipeline:  pipe,
		Store:     st,
		Pricer:    pricer,
		Log:       logger,
		clients:   map[string]llm.LLMClient{},
		templates: map[string]*PromptTemplate{},
	}
	if err := r.loadTemplates(); err != nil {
		st.Close()
		return nil, err
	}
	return r, nil
}

func (r *Runner) Close() error { return r.Store.Close() }

func (r *Runner) loadTemplates() error {
	for _, st := range r.Pipeline.Stages {
		tpl, err := LoadPromptTemplate(st.Prompt)
		if err != nil {
			return err
		}
		r.templates[st.Name] = tpl
	}
	return nil
}

func (r *Runner) client(model string) (llm.LLMClient, error) {
	if c, ok := r.clients[model]; ok {
		return c, nil
	}
	c, err := BuildClient(r.Models, model, r.Log)
	if err != nil {
		return nil, err
	}
	r.clients[model] = c
	return c, nil
}

// contextSnap freezes the context-assembly knobs (§3.8) inside the snapshot.
// Fixed field order — it is part of a content hash.
type contextSnap struct {
	GlossaryInjection   string `json:"glossary_injection"`
	GlossaryTokenBudget int    `json:"glossary_token_budget"`
	STMDepth            int    `json:"stm_depth"`
	OverlapTokens       int    `json:"overlap_tokens"`
	CacheTTL            string `json:"cache_ttl"`
}

// memoryVersion is the content-hash of the deterministically materialized
// injected memory (approved glossary + summaries + series-bible), the memory
// component of the snapshot (D5.2/D8). Until the memory bank v2 migration lands
// the materialization is empty, so the hash is a stable constant; once memory
// enters msgs this changes and the resnapshot-gate fires loudly instead of a
// stale checkpoint re-paying a diverged translation. STM is excluded (rebuilt
// from checkpoints, §3.2).
func (r *Runner) memoryVersion() string {
	h := sha256.New()
	// Domain tag + empty materialization; the store query lands with the v2
	// memory schema and feeds ORDER BY-stable rows here.
	h.Write([]byte("tm-memory-v1\x00"))
	return hex.EncodeToString(h.Sum(nil))
}

// snapshotID materializes the job-context snapshot (§3.2): brief_hash, chunker
// version, context-assembly knobs, the memory version and the full stage plan
// (model, prompt version + CONTENT hash, sampling, resolved capability).
// Payload is rendered with a fixed field order — the id is a content hash, so
// identical context re-upserts idempotently.
func (r *Runner) snapshotID() (id, payload string, err error) {
	type stageSnap struct {
		Name          string  `json:"name"`
		Role          string  `json:"role"`
		Model         string  `json:"model"`
		PromptVersion string  `json:"prompt_version"`
		PromptSHA256  string  `json:"prompt_sha256"`
		Temperature   float64 `json:"temperature"`
		Reasoning     string  `json:"reasoning"`
		// ExtraBody модели меняет ТЕЛО запроса (GLM thinking-off и т.п.) —
		// без него правка ручки в models.yaml молча не инвалидировала бы
		// чекпоинты (находка ревью). json.Marshal карты сортирует ключи →
		// детерминированный рендер.
		ModelExtra json.RawMessage `json:"model_extra,omitempty"`
		// Провайдер-уровневые оверрайды (local kind переопределяет wire
		// temperature/max_tokens ПОСЛЕ вычисления request-hash) — тоже влияют
		// на фактический запрос; без них правка providers.local.max_tokens
		// давала бы тот же snapshot и ложный checkpoint-hit на стендовом
		// local-пути (находка внешнего ревью F2).
		ProviderTemp   float64 `json:"provider_temp,omitempty"`
		ProviderMaxTok int     `json:"provider_max_tok,omitempty"`
		// Capability — резолвнутая wire-форма модели (D3.1): budget-ключ,
		// temperature-режим, reasoning-контроль. Меняет ТЕЛО запроса (max_tokens
		// vs max_completion_tokens, отправлять ли temperature, thinking-выключа-
		// тель), поэтому обязана входить в снапшот — иначе правка каппы молча
		// false-хитит чекпоинты (тот же класс D5.2, что и payload ниже).
		Capability json.RawMessage `json:"capability,omitempty"`
	}
	snap := struct {
		BriefHash        string `json:"brief_hash"`
		ChunkerVersion   string `json:"chunker_version"`
		EstimatorVersion string `json:"estimator_version"`
		// MaxTokensPolicy versions the attempt→max_tokens scaling (Веха 2): a
		// change to the retry doubling shifts attempt≥1 request_hashes, so it
		// belongs in the snapshot as a loud invalidation (same class as
		// estimator_version, applied to the regeneration axis).
		MaxTokensPolicy string `json:"max_tokens_policy"`
		PipelineCore    string `json:"pipeline_core"`
		// Defaults влияют на maxTokens, а тот входит в request-hash: без них
		// правка max_output_ratio молча инвалидировала бы все чекпоинты в
		// обход snapshot-гейта (находка ревью).
		MaxOutputRatio float64 `json:"max_output_ratio"`
		MinMaxTokens   int     `json:"min_max_tokens"`
		// ContextAssembly — ручки сборки контекста (глоссарий/STM/overlap/TTL,
		// §3.8). Их правка меняет ВХОД модели, когда память войдёт в msgs; сворачи-
		// ваем ДО инъекции, чтобы смена budget'а инъекции не промахнулась мимо
		// resnapshot-гейта (D5.2). Фикс-порядок полей — это content-hash.
		ContextAssembly contextSnap `json:"context_assembly"`
		// MemoryVersion — content-hash детерминированно материализованной инъекти-
		// руемой памяти (approved-глоссарий+резюме+series-bible). Не bump-счётчик
		// (D8): забыть пересчитать нельзя, а забытый bump переоткрыл бы класс тихой
		// переоплаты D5.2. До миграции банка памяти (v2) материализация пуста —
		// хэш стабилен; когда память войдёт в msgs, поле сменится и resnapshot-
		// гейт сработает громко. STM сюда НЕ входит (пересобирается из чекпоинтов).
		MemoryVersion string      `json:"memory_version"`
		Stages        []stageSnap `json:"stages"`
	}{
		BriefHash:        r.Book.BriefHash(),
		ChunkerVersion:   chunkerVersion,
		EstimatorVersion: estimatorVersion,
		MaxTokensPolicy:  maxTokensPolicyVersion,
		PipelineCore:     r.Pipeline.Core,
		MaxOutputRatio:   r.Pipeline.Defaults.MaxOutputRatio,
		MinMaxTokens:     r.Pipeline.Defaults.MinMaxTokens,
		ContextAssembly: contextSnap{
			GlossaryInjection:   r.Pipeline.Context.GlossaryInjection,
			GlossaryTokenBudget: r.Pipeline.Context.GlossaryTokenBudget,
			STMDepth:            r.Pipeline.Context.STMDepth,
			OverlapTokens:       r.Pipeline.Context.OverlapTokens,
			CacheTTL:            r.Pipeline.Context.CacheTTL,
		},
		MemoryVersion: r.memoryVersion(),
	}
	for _, st := range r.Pipeline.Stages {
		ss := stageSnap{
			Name: st.Name, Role: st.Role, Model: st.Model,
			PromptVersion: st.PromptVersion, PromptSHA256: r.templates[st.Name].SHA256,
			Temperature: st.Temperature, Reasoning: st.Reasoning,
		}
		if prov, ok := r.Models.Providers[r.Models.Models[st.Model].Provider]; ok {
			ss.ProviderTemp, ss.ProviderMaxTok = prov.Temperature, prov.MaxTokens
		}
		if extra := r.Models.Models[st.Model].ExtraBody; len(extra) > 0 {
			raw, merr := json.Marshal(extra)
			if merr != nil {
				return "", "", merr
			}
			ss.ModelExtra = raw
		}
		// Резолвнутая каппа — тем же ResolveCapability, что и у клиента, поэтому
		// хэшируемое всегда совпадает с отправляемым. json.Marshal сортирует
		// ключи карт (off_extra_body) → детерминированный рендер.
		capRaw, cerr := json.Marshal(r.Models.ResolveCapability(st.Model))
		if cerr != nil {
			return "", "", cerr
		}
		ss.Capability = capRaw
		snap.Stages = append(snap.Stages, ss)
	}
	data, err := json.Marshal(snap)
	if err != nil {
		return "", "", err
	}
	sum := sha256.Sum256(data)
	return hex.EncodeToString(sum[:]), string(data), nil
}

// StageResult reports one executed stage of a chunk.
type StageResult struct {
	Stage        string
	Role         string
	Model        string // фактически ответившая модель
	FromResume   bool   // no provider call was made THIS run (fully served from checkpoints)
	Usage        llm.Usage
	CostUSD      float64 // THIS run's spend (0 when fully served from checkpoints)
	CumCostUSD   float64 // sum across ALL attempts of this chunk×stage (F3-honest, incl. retries)
	LatencyMS    int
	FinishReason string
	Text         string // the usable output (only on an ok disposition)
	Disposition  Disposition
	FlagReason   FlagReason // "" when ok
	Detail       string
	Attempts     int
}

// ChunkOutcome is one chunk's result across the stage list.
type ChunkOutcome struct {
	Chapter     int
	ChunkIdx    int
	Stages      []StageResult
	FinalText   string      // the last stage's output; "" when the chunk is flagged
	Disposition Disposition // ok | flagged (a chunk has no "skipped" — that is a per-later-stage state)
	FlagReason  FlagReason  // the flagging stage's reason ("" when ok)
	CostUSD     float64     // THIS run's spend on this chunk
}

// BookResult aggregates a whole run over every chapter×chunk.
type BookResult struct {
	BookID   string
	Chunks   []ChunkOutcome
	TotalUSD float64 // THIS run's spend across all chunks
	Flagged  int     // number of flagged chunks (acceptance допускает N)
}

// ExitCode is the run's shell disposition: 0 clean, 2 completed-with-flags
// (acceptance допускает N флагов — приёмка это разрешает). An infra failure is
// an error from TranslateBook, which the CLI maps to 1.
func (b *BookResult) ExitCode() int {
	if b.Flagged > 0 {
		return 2
	}
	return 0
}

// CompletedWithFlags is the typed sentinel the CLI maps to exit code 2: the book
// finished end-to-end but N chunks were flagged for a human. It is NOT an infra
// failure (a plain error → exit 1); it is a "clean run, attention needed" signal
// carried up through `func run() error` in the idiomatic Go way.
type CompletedWithFlags struct {
	Flagged int
	Total   int
}

func (e *CompletedWithFlags) Error() string {
	return fmt.Sprintf("completed with %d/%d chunk(s) flagged for review", e.Flagged, e.Total)
}

// TranslateBook runs the whole book: split the normalized source into
// chapter/chunk units (chunker.go) and drive each chunk through the stage list.
// A bad chunk is flagged and the loop CONTINUES (D2); only an infra failure
// (ceiling, config change without --resnapshot, unbilled call failure, store
// error) aborts with an error, from which resume continues.
func (r *Runner) TranslateBook(ctx context.Context) (*BookResult, error) {
	srcRaw, err := os.ReadFile(r.Book.SourceFile)
	if err != nil {
		return nil, fmt.Errorf("pipeline: read source: %w", err)
	}
	// Нормализуем (BOM/CRLF/NFC) — request-hash стабилен между ОС/редакторами.
	source := NormalizeSource(string(srcRaw))
	if source == "" {
		return nil, fmt.Errorf("pipeline: source file %s is empty", r.Book.SourceFile)
	}

	// Snapshot is book-level (brief + stage plan + memory), computed once and
	// upserted before the loop; every job pins to it.
	snapID, snapPayload, err := r.snapshotID()
	if err != nil {
		return nil, err
	}
	if err := r.Store.UpsertSnapshot(snapID, r.Book.BriefHash(), snapPayload); err != nil {
		return nil, err
	}

	chunks := SplitChunks(source)
	if len(chunks) == 0 {
		return nil, fmt.Errorf("pipeline: source file %s produced no chunks after normalization", r.Book.SourceFile)
	}

	res := &BookResult{BookID: r.Book.BookID}
	for _, ch := range chunks {
		outcome, err := r.translateChunk(ctx, snapID, ch)
		if err != nil {
			// Infra failure: abort. Chunks already committed are checkpointed;
			// resume continues from here at $0 for the done work.
			return res, err
		}
		res.Chunks = append(res.Chunks, *outcome)
		res.TotalUSD += outcome.CostUSD
		if outcome.Disposition == DispFlagged {
			res.Flagged++
		}
	}
	return res, nil
}

// translateChunk drives ONE chunk through the stage list. Stages run in order,
// each feeding the next; the FIRST flagged stage stops the chunk — later stages
// are recorded `skipped` (no paid edit over a garbage draft, D2). Returns an
// error only on an infra failure.
func (r *Runner) translateChunk(ctx context.Context, snapID string, ch Chunk) (*ChunkOutcome, error) {
	out := &ChunkOutcome{Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Disposition: DispOK}
	prev := ""
	flagged := false
	var flagReason FlagReason

	for stageIdx, st := range r.Pipeline.Stages {
		if flagged {
			// An earlier stage flagged → this stage is not attempted or billed.
			detail := fmt.Sprintf("skipped: an upstream stage was flagged (%s)", flagReason)
			if err := r.Store.UpsertChunkStatus(store.ChunkStatus{
				BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name,
				SnapshotID: snapID, Disposition: string(DispSkipped), FlagReason: string(flagReason),
				Detail: detail,
			}); err != nil {
				return out, fmt.Errorf("pipeline: record skipped chunk_status: %w", err)
			}
			out.Stages = append(out.Stages, StageResult{
				Stage: st.Name, Role: st.Role, Model: st.Model,
				Disposition: DispSkipped, FlagReason: flagReason, Detail: detail,
			})
			continue
		}

		sr, err := r.runStage(ctx, st, stageIdx, snapID, ch, prev)
		if err != nil {
			return out, err
		}
		out.Stages = append(out.Stages, *sr)
		out.CostUSD += sr.CostUSD
		if sr.Disposition == DispFlagged {
			flagged = true
			flagReason = sr.FlagReason
			continue
		}
		prev = sr.Text
	}

	if flagged {
		out.Disposition = DispFlagged
		out.FlagReason = flagReason
		out.FinalText = "" // garbage/refusal never propagates to the next chunk or export
	} else {
		out.FinalText = prev
	}
	return out, nil
}

// runStage runs ONE stage of ONE chunk to a terminal disposition. It first
// resumes a resolved verdict (chunk_status read BEFORE any render — anti-wedge
// #1), otherwise walks the attempt axis: render → per-attempt checkpoint resume
// or a fresh reserve/call/settle → classify → retry the retryable {length,empty}
// subset with a doubled budget up to the regenerate cap, then flag. Returns an
// error ONLY on an infra failure; a bad completion is a disposition, never an
// error.
func (r *Runner) runStage(ctx context.Context, st config.Stage, stageIdx int, snapID string, ch Chunk, prev string) (*StageResult, error) {
	job, err := r.Store.EnsureJob(r.Book.BookID, ch.Chapter, st.Name, snapID)
	if err != nil {
		return nil, err
	}
	// Snapshot pinning (Р6): a job frozen on a stale snapshot is a loud stop
	// unless --resnapshot explicitly accepts the re-translation.
	if job.SnapshotID != snapID {
		if !r.Resnapshot {
			return nil, fmt.Errorf("pipeline: job %s/ch%d/%s was started under snapshot %.12s, current config renders snapshot %.12s — конфиг/промпты изменились; уже оплаченные чекпоинты станут недействительны и вызовы будут пере-оплачены; повторить с --resnapshot, чтобы принять это явно",
				r.Book.BookID, ch.Chapter, st.Name, job.SnapshotID, snapID)
		}
		if err := r.Store.UpdateJobSnapshot(job.ID, snapID); err != nil {
			return nil, err
		}
		r.Log.WarnContext(ctx, "job re-pinned to new snapshot (--resnapshot)", "stage", st.Name, "old", job.SnapshotID[:12], "new", snapID[:12])
	}

	// Render the wire messages up front. This is cheap (string substitution — the
	// expensive part is the LLM call, still gated below) and it lets the resume
	// fast-path be CONTENT-VERIFIED: the source bytes are not in the snapshot, so
	// a positional chunk_status row must be checked against the current rendered
	// content, else an edited source would serve a stale, divergent translation.
	msgs, err := Messages(r.templates[st.Name], RenderVars{Book: r.Book, Text: ch.Text, Draft: prev})
	if err != nil {
		return nil, err
	}
	contentHash := msgsContentHash(msgs)

	// Anti-wedge #1 + content-guard (self-review): resume resolves from
	// chunk_status BEFORE any LLM call. A flagged chunk is not in TM (garbage
	// never commits), so absence-in-TM ≠ "not done" — the disposition row is the
	// resume authority, and a terminally-flagged chunk is NOT re-attacked (no
	// infinite paid loop, even if the regenerate budget was raised). The row is
	// trusted ONLY when snapshot AND content both match: a stale-snapshot row
	// (post --resnapshot) or a stale-content row (source edited, not in the
	// snapshot) is ignored → the attempt loop below re-derives it content-safely
	// (a source edit becomes a silent per-chunk re-translate, §3.4, not a stale
	// serve). A `skipped` row is ignored too: translateChunk re-derives the skip.
	if cs, err := r.Store.GetChunkStatus(r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name); err != nil {
		return nil, err
	} else if cs != nil && cs.SnapshotID == snapID && cs.ContentHash == contentHash && cs.Disposition != string(DispSkipped) {
		return r.resumeFromChunkStatus(ctx, st, ch, cs)
	}

	// max_tokens base is sized from the text THIS stage processes: the source for
	// the translator, the prior draft for later stages (D2.5 — the monolingual
	// editor works over the Russian draft ≈1.9× the CJK source, so sizing edit
	// from source under-budgets and false-triggers a length retry).
	sizingText := ch.Text
	if stageIdx > 0 {
		sizingText = prev
	}
	baseMaxTokens := int(float64(EstimateTokens(sizingText)) * r.Pipeline.Defaults.MaxOutputRatio)
	if baseMaxTokens < r.Pipeline.Defaults.MinMaxTokens {
		baseMaxTokens = r.Pipeline.Defaults.MinMaxTokens
	}

	// Дополняем ReqInfo, СОХРАНЯЯ решения допуска (LogBodies) — перезапись с нуля
	// отрезала бы задокументированный debug-канал (находка ревью).
	ri, _ := obs.ReqInfoFromContext(ctx)
	ri.Book, ri.Chapter, ri.Chunk, ri.Stage, ri.Role = r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, st.Role
	ctx = obs.WithReqInfo(ctx, ri)

	maxRegen := r.Pipeline.Retries.RegenerateBeforeEscalate
	if maxRegen < 0 {
		maxRegen = 0
	}

	var cumCost, runCost float64
	var last stageAttempt
	anyFresh := false
	attemptsMade := 0
	for attempt := 0; ; attempt++ {
		maxTokens := maxTokensForAttempt(baseMaxTokens, attempt)
		att, err := r.runAttempt(ctx, st, snapID, ch, job, attempt, maxTokens, msgs)
		if err != nil {
			return nil, err // infra failure
		}
		attemptsMade = attempt + 1
		cumCost += att.cumCost
		runCost += att.runCost
		anyFresh = anyFresh || att.freshCall
		last = att
		if att.cls.ok() {
			break
		}
		// Flagged: re-attack only the retryable subset, only while regenerations
		// remain (a bigger budget on the attempt axis, D2.3). Everything else is
		// deterministic — a same-model retry would re-refuse and re-bill (D2.2).
		if att.cls.Reason.retryable() && attempt < maxRegen {
			r.Log.WarnContext(ctx, "stage flagged, regenerating with a larger budget",
				"stage", st.Name, "chapter", ch.Chapter, "chunk", ch.ChunkIdx,
				"attempt", attempt, "reason", string(att.cls.Reason), "next_max_tokens", maxTokensForAttempt(baseMaxTokens, attempt+1))
			continue
		}
		break
	}

	disposition := last.cls.Reason.disposition()
	finalHash := ""
	if disposition == DispOK {
		finalHash = last.reqHash
	}
	// chunk_status is a RESOLVE over the checkpoints: cost_usd sums every attempt
	// (F3-honest — retries are counted), final_hash points at the authoritative
	// checkpoint the ok path serves on resume.
	if err := r.Store.UpsertChunkStatus(store.ChunkStatus{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name,
		SnapshotID: snapID, ContentHash: contentHash, Disposition: string(disposition), FlagReason: string(last.cls.Reason),
		Attempts: attemptsMade, FinalHash: finalHash, CostUSD: cumCost, Detail: last.cls.Detail,
	}); err != nil {
		return nil, fmt.Errorf("pipeline: record chunk_status: %w", err)
	}

	sr := &StageResult{
		Stage: st.Name, Role: st.Role, Model: last.modelActual,
		FromResume:   !anyFresh,
		Usage:        last.usage,
		CostUSD:      runCost,
		CumCostUSD:   cumCost,
		LatencyMS:    last.latency,
		FinishReason: last.finish,
		Disposition:  disposition,
		FlagReason:   last.cls.Reason,
		Detail:       last.cls.Detail,
		Attempts:     attemptsMade,
	}
	if disposition == DispOK {
		sr.Text = last.text
	} else {
		r.Log.WarnContext(ctx, "stage flagged", "stage", st.Name, "chapter", ch.Chapter,
			"chunk", ch.ChunkIdx, "reason", string(last.cls.Reason), "attempts", attemptsMade,
			"cum_cost_usd", fmt.Sprintf("%.6f", cumCost))
	}
	return sr, nil
}

// stageAttempt is the result of one attempt (a checkpoint hit or a fresh call).
type stageAttempt struct {
	reqHash     string
	cls         classification
	text        string
	usage       llm.Usage
	finish      string
	modelActual string
	latency     int
	runCost     float64 // billed THIS run (0 on a checkpoint hit)
	cumCost     float64 // this attempt's cost (checkpoint cost on a hit, fresh cost on a call)
	freshCall   bool    // a provider call was made this run
}

// runAttempt executes exactly one attempt on the request-hash axis: a checkpoint
// hit is classified for free (self-heal, incl. legacy Фаза-0 empty/decode
// checkpoints — no re-billing), otherwise a fresh reserve → call → settle+
// checkpoint. It returns an error only on an infra failure; a bad completion
// comes back as a classification on the attempt.
func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, ch Chunk, job *store.Job, attempt, maxTokens int, msgs []llm.Message) (stageAttempt, error) {
	reqHash := RequestHash(r.Book.BookID, ch.Chapter, ch.ChunkIdx, attempt, st.Name, st.Role, st.Model,
		st.Temperature, st.Reasoning, false, maxTokens, snapID, msgs)
	att := stageAttempt{reqHash: reqHash, modelActual: st.Model}

	// Resume on the attempt axis: a checkpoint means THIS attempt already happened
	// and was billed — classify its text and never re-bill (kill -9 loses ≤1 call;
	// legacy empty/decode checkpoints self-heal by classification, not a crash).
	if cp, err := r.Store.GetCheckpoint(reqHash); err != nil {
		return att, err
	} else if cp != nil {
		var usage llm.Usage
		_ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
		att.usage = usage
		att.text = cp.ResponseText
		att.finish = cp.FinishReason
		att.modelActual = cp.ModelActual
		att.cumCost = cp.CostUSD
		att.cls = classify(classifyInput{Source: ch.Text, Output: cp.ResponseText, Finish: cp.FinishReason, TargetLang: r.Book.TargetLang})
		r.Store.LogRequest(ctx, r.Log, store.RequestLog{
			BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
			Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: cp.ModelActual,
			RequestHash: reqHash, TMHit: true, OK: att.cls.ok(), FinishReason: cp.FinishReason,
			Degraded: degradedTag(att.cls),
		})
		_ = r.Store.SetJobStatus(job.ID, "done")
		r.Log.InfoContext(ctx, "attempt served from checkpoint", "stage", st.Name,
			"attempt", attempt, "hash", reqHash[:12], "disposition", string(att.cls.Reason.disposition()))
		return att, nil
	}

	// Fresh call: reserve → call → settle+checkpoint (atomic, §3.3).
	price := r.Pricer.PriceFor(st.Model)
	promptEst := 0
	for _, m := range msgs {
		promptEst += EstimateTokens(m.Content)
	}
	estimate := ledger.EstimateUSD(price, promptEst, maxTokens)

	resv, verdict, err := r.Store.Reserve(r.Book.BookID, estimate, store.Ceilings{
		BookUSD: r.Book.Ceilings.BookUSD, DayUSD: r.Book.Ceilings.DayUSD,
	})
	if err != nil {
		_ = r.Store.SetJobStatus(job.ID, "failed")
		return att, err
	}
	switch verdict {
	case store.ReserveDeniedBook:
		// Ceiling is a hard, book-wide stop (not a per-chunk flag): the job stays
		// 'pending' and resume continues once the ceiling is raised.
		return att, fmt.Errorf("pipeline: book USD ceiling reached (%.2f$) — raise ceilings.book_usd or stop", r.Book.Ceilings.BookUSD)
	case store.ReserveDeniedDay:
		return att, fmt.Errorf("pipeline: daily USD ceiling reached (%.2f$)", r.Book.Ceilings.DayUSD)
	}

	client, err := r.client(st.Model)
	if err != nil {
		_ = r.Store.Release(resv)
		_ = r.Store.SetJobStatus(job.ID, "failed")
		return att, err
	}
	if err := r.Store.SetJobStatus(job.ID, "running"); err != nil {
		_ = r.Store.Release(resv)
		return att, err
	}
	att.freshCall = true

	start := time.Now()
	resp, err := client.Complete(ctx, llm.LLMRequest{
		Model:           st.Model,
		Messages:        msgs,
		MaxTokens:       maxTokens,
		Temperature:     st.Temperature,
		ReasoningEffort: st.Reasoning,
	})
	att.latency = int(time.Since(start).Milliseconds())
	if err != nil {
		var bde *llm.BilledDecodeError
		if errors.As(err, &bde) {
			// 2xx with an unreadable body: the provider ALREADY billed. Do not
			// release; conservatively settle the ESTIMATE with an empty decode
			// checkpoint, then FLAG (anti-wedge #2 — the loop continues, this is
			// not an infra crash). A settle failure here is a real infra fault.
			if serr := r.Store.SettleWithCheckpoint(resv, estimate, store.Checkpoint{
				RequestHash: reqHash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: attempt,
				Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: st.Model,
				ResponseText: "", UsageJSON: "{}", CostUSD: estimate, FinishReason: decodeErrorFinish,
			}); serr != nil {
				return att, fmt.Errorf("pipeline: settle after billed decode failure: %w", serr)
			}
			att.finish = decodeErrorFinish
			att.cumCost, att.runCost = estimate, estimate
			att.cls = classification{FlagDecodeError, "billed 2xx with an unreadable body: " + err.Error()}
			r.Store.LogRequest(ctx, r.Log, store.RequestLog{
				BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
				Stage: st.Name, Role: st.Role, ModelRequested: st.Model,
				RequestHash: reqHash, CostUSD: estimate, LatencyMS: att.latency,
				Degraded: "billed_2xx_decode_failed", Err: err.Error(), OK: false, FinishReason: decodeErrorFinish,
			})
			_ = r.Store.SetJobStatus(job.ID, "done")
			return att, nil
		}
		// No 2xx ever arrived: nothing was billed. Release and surface as an INFRA
		// failure — a long book pauses/resumes on an outage or terminal 4xx (D4),
		// rather than flag-storming every remaining chunk on a dead provider.
		if rerr := r.Store.Release(resv); rerr != nil {
			r.Log.ErrorContext(ctx, "release after failed call also failed", "err", rerr)
		}
		r.Store.LogRequest(ctx, r.Log, store.RequestLog{
			BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
			Stage: st.Name, Role: st.Role, ModelRequested: st.Model,
			RequestHash: reqHash, LatencyMS: att.latency, Err: err.Error(), OK: false,
		})
		_ = r.Store.SetJobStatus(job.ID, "failed")
		return att, fmt.Errorf("pipeline: stage %s call: %w", st.Name, err)
	}

	modelActual := resp.Model
	if modelActual == "" {
		modelActual = st.Model
	}
	att.modelActual = modelActual
	att.text = resp.Text
	att.finish = resp.FinishReason
	att.usage = resp.Usage

	// Цена по фактически ответившей модели, с fallback на ЗАПРОШЕННУЮ (не на
	// дешёвый глобальный якорь), если провайдер вернул канонизированный слаг.
	price = r.Pricer.PriceForResponse(st.Model, modelActual)
	cost := ledger.CostUSD(price, resp.Usage)
	// Платная модель (InputPerM>0) вернула 2xx с нулевым usage — $0 ослепил бы
	// потолок: берём консервативную оценку. Для local ($0 цена) нулевой usage
	// штатен — остаётся $0.
	if cost == 0 && price.InputPerM > 0 && resp.Usage.PromptTokens == 0 && resp.Usage.CompletionTokens == 0 {
		r.Log.WarnContext(ctx, "paid 2xx with zero usage; settling the reservation estimate to keep the ceiling honest",
			"stage", st.Name, "model", modelActual, "estimate_usd", fmt.Sprintf("%.6f", estimate))
		cost = estimate
	}
	usageJSON, err := json.Marshal(resp.Usage)
	if err != nil {
		return att, err
	}
	att.cumCost, att.runCost = cost, cost

	// Деньги: settle + сырой ответ — одна транзакция (§3.3). ВСЕГДА, даже для
	// пустого/усечённого/отказного ответа: он оплачен провайдером, чекпоинт
	// хранит его до цента, а годность решает classify ПОСЛЕ (деньги ≠ вердикт).
	if err := r.Store.SettleWithCheckpoint(resv, cost, store.Checkpoint{
		RequestHash: reqHash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: attempt,
		Stage: st.Name, Role: st.Role,
		ModelRequested: st.Model, ModelActual: modelActual, ResponseText: resp.Text,
		UsageJSON: string(usageJSON), CostUSD: cost, FinishReason: resp.FinishReason,
		ProviderRequestID: resp.ProviderRequestID,
	}); err != nil {
		// The provider HAS billed this 2xx; failing to persist means the money
		// state is behind reality — fail loud, never continue on top.
		return att, fmt.Errorf("pipeline: settle stage %s: %w", st.Name, err)
	}

	// Classify AFTER the money is durably settled+checkpointed. F4 lives here: a
	// non-empty truncated length draft is now classified (flagged/retried), never
	// silently passed downstream as OK; an empty completion is flagged too, not a
	// run-crash.
	att.cls = classify(classifyInput{Source: ch.Text, Output: resp.Text, Finish: resp.FinishReason, TargetLang: r.Book.TargetLang})

	r.Store.LogRequest(ctx, r.Log, store.RequestLog{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
		Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: modelActual,
		RequestHash:  reqHash,
		PromptTokens: resp.Usage.PromptTokens, CachedTokens: resp.Usage.CachedTokens,
		CacheCreationTokens: resp.Usage.CacheCreationTokens,
		CompletionTokens:    resp.Usage.CompletionTokens, ReasoningTokens: resp.Usage.ReasoningTokens,
		CostUSD: cost, LatencyMS: att.latency, FinishReason: resp.FinishReason,
		OK: att.cls.ok(), Degraded: degradedTag(att.cls),
	})
	_ = r.Store.SetJobStatus(job.ID, "done")
	r.Log.InfoContext(ctx, "attempt completed", "stage", st.Name, "attempt", attempt, "model", modelActual,
		"cost_usd", fmt.Sprintf("%.6f", cost), "latency_ms", att.latency,
		"disposition", string(att.cls.Reason.disposition()), "reason", string(att.cls.Reason))
	return att, nil
}

// resumeFromChunkStatus serves a chunk×stage whose disposition is already
// resolved (read before any render — no re-billing). An ok stage returns its
// authoritative checkpoint text to feed the next stage; a flagged stage returns
// the flag WITHOUT re-attacking (a terminal flag never re-enters the paid loop).
func (r *Runner) resumeFromChunkStatus(ctx context.Context, st config.Stage, ch Chunk, cs *store.ChunkStatus) (*StageResult, error) {
	sr := &StageResult{
		Stage: st.Name, Role: st.Role, Model: st.Model, FromResume: true,
		CostUSD: 0, CumCostUSD: cs.CostUSD,
		Disposition: Disposition(cs.Disposition), FlagReason: FlagReason(cs.FlagReason),
		Detail: cs.Detail, Attempts: cs.Attempts,
	}
	if cs.Disposition == string(DispOK) {
		cp, err := r.Store.GetCheckpoint(cs.FinalHash)
		if err != nil {
			return nil, err
		}
		if cp == nil || strings.TrimSpace(cp.ResponseText) == "" {
			// chunk_status ok is only written AFTER its checkpoint is durably
			// settled, so this is a torn/tampered store — fail loud rather than
			// feed the next stage an empty draft.
			return nil, fmt.Errorf("pipeline: chunk_status ok for %s/ch%d/chunk%d/%s references checkpoint %.12s which is missing/empty — inconsistent store",
				r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, cs.FinalHash)
		}
		var usage llm.Usage
		_ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
		sr.Usage = usage
		sr.Model = cp.ModelActual
		sr.FinishReason = cp.FinishReason
		sr.Text = cp.ResponseText
	}
	r.Store.LogRequest(ctx, r.Log, store.RequestLog{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
		Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: sr.Model,
		RequestHash: cs.FinalHash, TMHit: true, OK: cs.Disposition == string(DispOK),
		FinishReason: sr.FinishReason, Degraded: nonOKTag(cs.Disposition, cs.FlagReason),
	})
	r.Log.InfoContext(ctx, "stage resolved from chunk_status", "stage", st.Name,
		"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "disposition", cs.Disposition, "reason", cs.FlagReason)
	return sr, nil
}

// degradedTag surfaces a flag reason into the request_log `degraded` column.
func degradedTag(cls classification) string {
	if cls.ok() {
		return ""
	}
	return string(cls.Reason)
}

// nonOKTag is the request_log `degraded` value for a resumed disposition row.
func nonOKTag(disposition, reason string) string {
	if disposition == string(DispOK) {
		return ""
	}
	return reason
}