textmachine/backend/internal/pipeline/status.go

package pipeline

import (
	"context"
	"fmt"
	"sort"

	"textmachine/backend/internal/store"
)

// status.go: the READ-ONLY progress projection (`tmctl status`) and the targeted re-attack
// of flagged chunks (`tmctl redrive`) — the D12 "прогресс — единственный реальный пробел"
// tail, ratified by D15.3. Status makes ZERO LLM calls and ZERO checkpoint replays: it reads
// the book manifest (a deterministic, $0 re-chunk of the source) and projects it against the
// stored chunk_status / spend / retrieval_state. Redrive resets the flagged chunks' terminal
// state (their chunk_status + checkpoints) and re-runs the durable loop, so the DispOK work
// resumes at $0 and only the reset stages re-attack with a fresh retry/escalation budget.

// ChunkState is a chunk's resolved lifecycle state for the projection — a chunk-level view
// over its per-stage chunk_status rows. `skipped` is deliberately NOT a chunk state: it is a
// per-STAGE substate (the downstream stages of a flagged chunk), surfaced in the passport's
// stage counts, since the linear C1 core never skips a whole chunk.
type ChunkState string

const (
	ChunkDone       ChunkState = "done"        // every pipeline stage resolved ok
	ChunkFlagged    ChunkState = "flagged"     // a stage flagged (chunk not translated; ≠ infra failure)
	ChunkInProgress ChunkState = "in_progress" // some stages resolved, not all, none flagged (an interrupted run)
	ChunkPending    ChunkState = "pending"     // no stage attempted yet
)

// ChapterPassport is the per-chapter quality passport (D12): unit counts, the worst flag
// reason, a pass|attention|fail verdict (exp07 chapter rule) and the chapter's spend.
type ChapterPassport struct {
	Chapter         int     `json:"chapter"`
	ChunksTotal     int     `json:"chunks_total"`
	ChunksDone      int     `json:"chunks_done"`
	ChunksFlagged   int     `json:"chunks_flagged"`
	ChunksInFlight  int     `json:"chunks_in_progress"`
	ChunksPending   int     `json:"chunks_pending"`
	StagesSkipped   int     `json:"stages_skipped"` // per-stage skip count (downstream of a flag)
	Escalations     int     `json:"escalations"`
	PostcheckMisses int     `json:"postcheck_misses"` // confirmed post-check misses (retrieval_state)
	StyleFlags      int     `json:"style_flags"`      // cheap style/number gate hits (observability, not a disposition)
	WorstFlagReason string  `json:"worst_flag_reason,omitempty"`
	Verdict         string  `json:"verdict"` // pass | attention | fail (exp07: 0/1/≥2 flagged chunks)
	CostUSD         float64 `json:"cost_usd"`
}

// StatusReport is the whole-book read-model.
type StatusReport struct {
	BookID   string `json:"book_id"`
	Snapshot string `json:"snapshot_id"` // the snapshot the stored rows were resolved under ("" = nothing run yet)
	// SnapshotDrift is true when the stored rows carry more than one snapshot id (a config
	// changed mid-book without a full --resnapshot re-pin) — a loud signal, not a silent skew.
	SnapshotDrift bool `json:"snapshot_drift"`
	// ConfigDrift is true when the CURRENT config renders a snapshot DIFFERENT from the one the
	// stored rows were resolved under — a wire/verdict-affecting edit since the last run (e.g. a
	// prompt-version bump, a gate flip) that `tmctl translate` would --resnapshot. Without this
	// the operator sees "done/pass" and false confidence (finding #3). CurrentSnapshot is what
	// the config renders now (empty when it could not be computed or matches).
	ConfigDrift     bool   `json:"config_drift"`
	CurrentSnapshot string `json:"current_snapshot,omitempty"`

	TotalChunks int     `json:"total_chunks"`
	Done        int     `json:"done"`
	InProgress  int     `json:"in_progress"`
	Flagged     int     `json:"flagged"`
	Pending     int     `json:"pending"`
	PercentDone float64 `json:"percent_done"` // 100·done/total, unit-count only (no synthetic time-bar)

	// GlossaryMissFlagged counts chunks the post-check GATE promoted to flagged/glossary_miss (a
	// CONFIRMED miss on an otherwise-ok chunk). A SUBSET of Flagged and NOT re-drivable: the miss
	// re-derives from the unchanged glossary, so the fix is a seed edit + `translate --resnapshot`,
	// never a redrive (a no-op for them). Flagged − GlossaryMissFlagged is the re-drivable remainder,
	// which is what the status CLI must advise `tmctl redrive` for (minor 1d: the old blanket hint
	// sent glossary_miss chunks into a redrive dead-end).
	GlossaryMissFlagged int `json:"glossary_miss_flagged"`

	Escalations     int `json:"escalations"`
	PostcheckMisses int `json:"postcheck_misses"`
	// StyleFlags is the book-wide total of the cheap deterministic style/number gate hits
	// (dialogue-dash, ё, translit-interjection, 万/億 magnitude). Observability, never a
	// disposition — a nonzero count is "attention worth a human glance", not a failed chunk.
	StyleFlags int `json:"style_flags"`

	CommittedUSD     float64 `json:"committed_usd"`
	ReservedUSD      float64 `json:"reserved_usd"`
	BookCeilingUSD   float64 `json:"book_ceiling_usd,omitempty"`
	CeilingPct       float64 `json:"ceiling_pct,omitempty"` // 100·(committed+reserved)/book_ceiling
	ProjectedBookUSD float64 `json:"projected_book_usd"`    // committed extrapolated over the whole book
	ETASeconds       float64 `json:"eta_seconds,omitempty"` // secondary: mean fresh-call throughput × remaining

	Chapters []ChapterPassport `json:"chapters"`
}

// flagReasonSeverity ranks flag reasons worst-first for the chapter's "worst flag" passport
// field. Lower = worse. Deterministic content-failures (refusal/echo/excision) outrank
// retryable budget symptoms (length/empty), which are the least alarming.
func flagReasonSeverity(reason string) int {
	switch FlagReason(reason) {
	case FlagHardRefusal, FlagSoftRefusal, FlagContentFilter, FlagHardBlock:
		return 0
	case FlagCJKArtifact, FlagExcisionSuspect, FlagCoverageFail:
		return 1
	case FlagLoopDegenerate:
		return 2
	case FlagDecodeError:
		return 3
	case FlagGlossaryMiss:
		return 4
	case FlagLength, FlagEmpty:
		return 5
	}
	return 6
}

// bookChunks re-derives the book's chunk manifest — Ingest + SplitChunks — for the honest N/M
// denominator. Pure and $0 (no LLM); unlike TranslateBook it does NOT persist ruby or seed the
// glossary, so status stays a pure read. A change to the chunk BOUNDARIES (added/removed text that
// shifts the manifest) or the chunker VERSION is reflected here.
//
// CAVEAT (minor 1d — narrowed overclaim): a source edit that leaves the boundaries unchanged (a
// typo fix inside a chunk) is INVISIBLE to status — it changes the chunk's content_hash but not the
// manifest and not the snapshot (which carries chunker_version, not the source bytes), so status
// still reads done/pass over a translation of the OLD text. Surfacing it would need a per-chunk
// content_hash re-render ($0) compared against the stored chunk_status.content_hash; deferred,
// because the resume fast-path already re-checks content_hash on the NEXT translate — which is
// where such an edit surfaces (as a re-bill of exactly the touched chunks), just not in status.
func (r *Runner) bookChunks() ([]Chunk, error) {
	doc, err := IngestEncoded(r.Book.SourceFile, r.Book.Encoding, r.Book.SourceLang)
	if err != nil {
		return nil, err
	}
	return SplitChunks(doc.Chapters), nil
}

// chunkKey identifies a chunk positionally.
type chunkKey struct {
	chapter, chunkIdx int
}

// resolveChunkState maps a chunk's per-stage rows to its chunk-level state, plus its cost,
// escalation flag and (when flagged) the flag reason. stagesTotal is the pipeline stage count.
func resolveChunkState(rows []store.ChunkStatus, stagesTotal int) (st ChunkState, reason string, cost float64, escalated bool, skipped int) {
	if len(rows) == 0 {
		return ChunkPending, "", 0, false, 0
	}
	ok := 0
	for _, cs := range rows {
		cost += cs.CostUSD
		if cs.Escalated {
			escalated = true
		}
		switch cs.Disposition {
		case string(DispOK):
			ok++
		case string(DispFlagged):
			// The first (only) flagged stage decides the chunk; keep its reason.
			st, reason = ChunkFlagged, cs.FlagReason
		case string(DispSkipped):
			skipped++
		}
	}
	if st == ChunkFlagged {
		return st, reason, cost, escalated, skipped
	}
	if ok == stagesTotal {
		return ChunkDone, "", cost, escalated, skipped
	}
	return ChunkInProgress, "", cost, escalated, skipped
}

// Status builds the read-only progress projection. It opens no jobs, reserves nothing, makes
// no LLM call and replays no checkpoint — the D12 query-handler analogue. Safe to run whenever
// the project file is not held by a running tmctl (the same exclusive-lock rule as `report`).
func (r *Runner) Status(ctx context.Context) (*StatusReport, error) {
	chunks, err := r.bookChunks()
	if err != nil {
		return nil, err
	}
	statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID)
	if err != nil {
		return nil, err
	}
	byChunk := map[chunkKey][]store.ChunkStatus{}
	snapSeen := map[string]bool{}
	for _, cs := range statuses {
		byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}] = append(byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}], cs)
		if cs.SnapshotID != "" {
			snapSeen[cs.SnapshotID] = true
		}
	}

	// Per-chunk post-check misses (retrieval_state) — the glossary-consistency signal that is
	// re-derived each run and NOT a chunk_status disposition (default flagger mode).
	states, err := r.Store.RetrievalStatesForBook(r.Book.BookID)
	if err != nil {
		return nil, err
	}
	missByChunk := map[chunkKey]int{}
	styleByChunk := map[chunkKey]int{}
	for _, rs := range states {
		missByChunk[chunkKey{rs.Chapter, rs.ChunkIdx}] = rs.NPostcheckMiss
		styleByChunk[chunkKey{rs.Chapter, rs.ChunkIdx}] = rs.NStyleFlags
	}

	stagesTotal := len(r.Pipeline.Stages)
	// The post-check GATE (opt-in) flags a chunk at the CHUNK level AFTER the stage loop, so it
	// is NEVER written as a chunk_status row (every stage stays DispOK). Without accounting for
	// it here, status would report a gate-flagged chunk as done/pass while `tmctl translate`
	// exits 2 (finding #2). When the gate is ON, promote an otherwise-done chunk with a CONFIRMED
	// post-check miss to flagged/glossary_miss so status matches the run.
	gateOn := r.Pipeline.Gates.Glossary.PostcheckGate
	rep := &StatusReport{BookID: r.Book.BookID, TotalChunks: len(chunks)}
	passports := map[int]*ChapterPassport{}
	var chapterOrder []int
	var processedCost float64 // spend of PROCESSED chunks (done+flagged) — the projection base (finding #7)

	for _, ch := range chunks {
		p := passports[ch.Chapter]
		if p == nil {
			p = &ChapterPassport{Chapter: ch.Chapter, Verdict: "pass"}
			passports[ch.Chapter] = p
			chapterOrder = append(chapterOrder, ch.Chapter)
		}
		p.ChunksTotal++
		key := chunkKey{ch.Chapter, ch.ChunkIdx}
		state, reason, cost, escalated, skipped := resolveChunkState(byChunk[key], stagesTotal)
		p.CostUSD += cost
		p.StagesSkipped += skipped
		miss := missByChunk[key]
		p.PostcheckMisses += miss
		rep.PostcheckMisses += miss
		style := styleByChunk[key]
		p.StyleFlags += style
		rep.StyleFlags += style
		if gateOn && state == ChunkDone && miss > 0 {
			// Matches translateChunk: the gate flags only an otherwise-ok chunk. Not re-drivable
			// (the miss re-derives from the unchanged glossary; a fix is a glossary edit =
			// --resnapshot, not a redrive) — counted separately so the CLI advises the right action.
			state, reason = ChunkFlagged, string(FlagGlossaryMiss)
			rep.GlossaryMissFlagged++
		}
		if escalated {
			p.Escalations++
			rep.Escalations++
		}
		switch state {
		case ChunkDone:
			rep.Done++
			p.ChunksDone++
		case ChunkFlagged:
			rep.Flagged++
			p.ChunksFlagged++
			if p.WorstFlagReason == "" || flagReasonSeverity(reason) < flagReasonSeverity(p.WorstFlagReason) {
				p.WorstFlagReason = reason
			}
		case ChunkInProgress:
			rep.InProgress++
			p.ChunksInFlight++
		case ChunkPending:
			rep.Pending++
			p.ChunksPending++
		}
		if state == ChunkDone || state == ChunkFlagged {
			processedCost += cost // a fully-attempted chunk's cost feeds the projection
		}
	}

	// Chapter verdicts (exp07 chapter rule: 0 flagged = pass, 1 = attention, ≥2 = fail).
	sort.Ints(chapterOrder)
	for _, n := range chapterOrder {
		p := passports[n]
		switch {
		case p.ChunksFlagged >= 2:
			p.Verdict = "fail"
		case p.ChunksFlagged == 1:
			p.Verdict = "attention"
		default:
			p.Verdict = "pass"
		}
		rep.Chapters = append(rep.Chapters, *p)
	}

	if rep.TotalChunks > 0 {
		rep.PercentDone = 100 * float64(rep.Done) / float64(rep.TotalChunks)
	}
	if len(snapSeen) == 1 {
		for s := range snapSeen {
			rep.Snapshot = s
		}
	} else if len(snapSeen) > 1 {
		rep.SnapshotDrift = true
	}

	// Config-drift (finding #3): compute the CURRENT config's snapshot READ-ONLY (materialize
	// the STORED glossary — no re-seed, no write) and compare to the single snapshot the stored
	// rows carry. A wire/verdict-affecting config edit since the last run (a prompt bump, a gate
	// flip — this very package bumps the editor prompt_version) is otherwise hidden behind
	// "done/pass" until translate forces a --resnapshot. Only meaningful when a single snapshot
	// is on record (multi-snapshot is already flagged as SnapshotDrift). Note: a change to the
	// seed FILE that was not re-run is NOT detected here (the stored glossary is what status
	// projects); it surfaces on the next translate's re-seed.
	if rep.Snapshot != "" {
		if curSnap, serr := r.currentSnapshotProjected(); serr == nil && curSnap != rep.Snapshot {
			rep.ConfigDrift = true
			rep.CurrentSnapshot = curSnap
		}
	}

	// Money.
	committed, reserved, err := r.Store.SpentUSD(r.Book.BookID)
	if err != nil {
		return nil, err
	}
	rep.CommittedUSD, rep.ReservedUSD = committed, reserved
	rep.BookCeilingUSD = r.Book.Ceilings.BookUSD
	if rep.BookCeilingUSD > 0 {
		rep.CeilingPct = 100 * (committed + reserved) / rep.BookCeilingUSD
	}
	// Projected book cost: extrapolate the per-PROCESSED-chunk average over the whole book
	// (done + flagged = a chunk fully attempted). Uses processedCost, NOT book committed:
	// committed also carries partial spend on IN-PROGRESS chunks (excluded from the denominator),
	// which would over-estimate (finding #7). The clean average × total is the honest estimate.
	processed := rep.Done + rep.Flagged
	if processed > 0 {
		rep.ProjectedBookUSD = processedCost / float64(processed) * float64(rep.TotalChunks)
	}

	// ETA (secondary): mean fresh-call throughput × remaining processing. No synthetic bar.
	// DEVIATION from D12 (which ratified an EWMA) — minor 1d, made explicit: this is a plain
	// arithmetic mean, not an exponentially-weighted one. For a batch book run throughput is
	// ~stationary, so the mean and an EWMA converge; a recency-weighted EWMA (which would adapt to
	// a mid-run provider slowdown) is deferred because it needs the ORDERED per-call latencies, not
	// the sum+count FreshCallLatencyMS returns. Kept secondary/advisory in the output so it is never
	// mistaken for a committed completion time.
	totalMS, freshCalls, err := r.Store.FreshCallLatencyMS(r.Book.BookID)
	if err != nil {
		return nil, err
	}
	remaining := rep.TotalChunks - processed
	if processed > 0 && freshCalls > 0 && remaining > 0 {
		meanMSPerChunk := float64(totalMS) / float64(processed)
		rep.ETASeconds = meanMSPerChunk * float64(remaining) / 1000
	}
	return rep, nil
}

// currentSnapshotProjected computes the CURRENT config's snapshotID from the STORED glossary
// (read-only: materialize what is persisted, no re-seed, no write) — the single projection
// Status uses for ConfigDrift and Redrive uses to guard its destructive reset. Side effect
// (deliberate, on the read path): it sets r.memory to the stored-glossary materialization, the
// same as Status did inline. A seed-FILE edit not yet re-run is NOT reflected here (the STORED
// glossary is projected, not the seed file) — that drift surfaces on the next translate's
// re-seed, exactly as it does for `translate` itself.
func (r *Runner) currentSnapshotProjected() (string, error) {
	rows, err := r.Store.GlossaryForBook(r.Book.BookID)
	if err != nil {
		return "", err
	}
	r.memory = materializeMemory(rows, r.Pipeline.Gates.Glossary.PostcheckGate)
	id, _, err := r.snapshotID()
	return id, err
}

// RedriveSelector picks the flagged chunks to re-attack. -1 on Chapter/ChunkIdx means "any"
// (a valid chunk index is ≥0, a chapter ≥1, so -1 is an unambiguous "unset"); an empty Reason
// matches any flag_reason. DryRun reports what WOULD be reset without touching anything.
type RedriveSelector struct {
	Chapter  int
	ChunkIdx int
	Reason   string
	DryRun   bool
}

func (s RedriveSelector) matches(cs store.ChunkStatus) bool {
	if s.Chapter >= 0 && cs.Chapter != s.Chapter {
		return false
	}
	if s.ChunkIdx >= 0 && cs.ChunkIdx != s.ChunkIdx {
		return false
	}
	if s.Reason != "" && cs.FlagReason != s.Reason {
		return false
	}
	return true
}

// RedriveTarget is one chunk selected for re-attack and the stages that were (or would be) reset.
type RedriveTarget struct {
	Chapter    int      `json:"chapter"`
	ChunkIdx   int      `json:"chunk_idx"`
	FlagReason string   `json:"flag_reason"`
	Stages     []string `json:"reset_stages"`
}

// RedriveSummary reports the plan/outcome of the reset half of a redrive.
type RedriveSummary struct {
	Targets  []RedriveTarget `json:"targets"`
	DryRun   bool            `json:"dry_run"`
	ResetRun bool            `json:"reset_run"` // the re-translate ran (false on dry-run / no targets)
}

// Redrive re-attacks the FLAGGED chunks matching sel (D15.3 / D12 Step-Functions redrive). It
// NEVER touches DispOK work (D12): for each selected chunk it resets only the flagged stage and
// its downstream skipped stages — their chunk_status + checkpoints are deleted (ResetChunkStages),
// so the durable re-run re-derives them with a FRESH retry/escalation budget (a fresh provider
// call, not a deterministic replay of the flagged completion) while the upstream ok stages resume
// at $0. Money bills normally (reserve/settle+checkpoint); the previously-spent money on the
// discarded attempts stays committed (honest — it was really billed), so the ceilings remain
// accurate. Requires the CURRENT config to render the SAME snapshot the flagged rows carry (no
// --resnapshot): a config change makes the re-run fail loud with the resnapshot guidance instead
// of silently re-pricing the book. Returns the reset plan, and the re-run's BookResult (nil on
// dry-run or when nothing matched).
func (r *Runner) Redrive(ctx context.Context, sel RedriveSelector) (*RedriveSummary, *BookResult, error) {
	statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID)
	if err != nil {
		return nil, nil, err
	}
	// Group by chunk and find the targets: a chunk with a FLAGGED stage matching the selector.
	byChunk := map[chunkKey][]store.ChunkStatus{}
	var order []chunkKey
	for _, cs := range statuses {
		k := chunkKey{cs.Chapter, cs.ChunkIdx}
		if _, seen := byChunk[k]; !seen {
			order = append(order, k)
		}
		byChunk[k] = append(byChunk[k], cs)
	}
	sort.Slice(order, func(i, j int) bool {
		if order[i].chapter != order[j].chapter {
			return order[i].chapter < order[j].chapter
		}
		return order[i].chunkIdx < order[j].chunkIdx
	})

	// Build the reset PLAN first — pure, no mutation — so the destructive reset happens only
	// after the drift guard below clears (Task 1c). Each target resets its flagged stage AND its
	// downstream skipped stages; upstream DispOK stages are never included (D12 — never re-pay ok
	// work), which is the invariant the redrive tests mutation-lock.
	summary := &RedriveSummary{DryRun: sel.DryRun}
	for _, k := range order {
		rows := byChunk[k]
		targeted := false
		reason := ""
		for _, cs := range rows {
			if cs.Disposition == string(DispFlagged) && sel.matches(cs) {
				targeted, reason = true, cs.FlagReason
				break
			}
		}
		if !targeted {
			continue
		}
		var stages []string
		for _, cs := range rows {
			if cs.Disposition == string(DispFlagged) || cs.Disposition == string(DispSkipped) {
				stages = append(stages, cs.Stage)
			}
		}
		sort.Strings(stages)
		summary.Targets = append(summary.Targets, RedriveTarget{
			Chapter: k.chapter, ChunkIdx: k.chunkIdx, FlagReason: reason, Stages: stages,
		})
	}

	if sel.DryRun || len(summary.Targets) == 0 {
		return summary, nil, nil
	}

	// Config/seed-drift guard BEFORE the destructive reset (external-review 1c). A redrive re-runs
	// under the CURRENT config; if it drifted from the snapshot the stored rows carry, TranslateBook
	// fails loud in runStage with the resnapshot guidance — but the OLD code had already deleted the
	// flagged chunk_status rows + checkpoints (ResetChunkStages), so that fail-loud left the flag
	// telemetry destroyed and a later `status` read the chunk as pending/pass.
	//
	// The snapshot MUST be computed exactly the way TranslateBook will compute it — i.e. AFTER
	// re-seeding the glossary from the seed FILE (self-review major: a read-only stored-glossary
	// projection misses a seed-FILE edit — precisely what the glossary_miss CLI hint tells the
	// operator to make — so the guard would pass, the reset would fire, and the re-run's re-seed
	// would then fail loud AFTER the destructive delete). seedGlossary is idempotent and touches
	// only glossary rows (never chunk_status/checkpoints), so re-seeding here is safe; TranslateBook
	// re-seeds again identically. A seed error (e.g. a new collision) also surfaces here, before any
	// reset. Skipped under --resnapshot (the operator explicitly accepts the re-pin/re-pay).
	if !r.Resnapshot {
		if err := r.seedGlossary(ctx); err != nil {
			return summary, nil, fmt.Errorf("pipeline: redrive seed glossary: %w", err)
		}
		curSnap, _, serr := r.snapshotID()
		if serr != nil {
			return summary, nil, fmt.Errorf("pipeline: redrive snapshot check: %w", serr)
		}
		for _, cs := range statuses {
			if cs.SnapshotID != "" && cs.SnapshotID != curSnap {
				return summary, nil, fmt.Errorf("pipeline: redrive прерван — строки книги под снапшотом %.12s, а текущий конфиг/сид рендерит %.12s (конфиг/промпты/капабилити/сид-глоссарий изменились): сброс сейчас пере-оплатил бы книгу и уничтожил бы флаг-телеметрию; повторите `translate --resnapshot`, чтобы принять пере-оплату явно, либо верните конфиг/сид", cs.SnapshotID, curSnap)
			}
		}
	}

	// Destructive reset — safe now that drift has been ruled out.
	for _, t := range summary.Targets {
		if err := r.Store.ResetChunkStages(r.Book.BookID, t.Chapter, t.ChunkIdx, t.Stages); err != nil {
			return summary, nil, fmt.Errorf("pipeline: redrive reset ch%d/chunk%d: %w", t.Chapter, t.ChunkIdx, err)
		}
	}

	// Re-run the durable loop: reset chunks re-attack fresh; everything else resumes at $0.
	res, err := r.TranslateBook(ctx)
	summary.ResetRun = true
	return summary, res, err
}