textmachine/backend/internal/pipeline/waverun.go

package pipeline

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"

	"textmachine/backend/internal/checks"
	"textmachine/backend/internal/chunk"
	"textmachine/backend/internal/config"
	"textmachine/backend/internal/lang"
	"textmachine/backend/internal/membank"
	"textmachine/backend/internal/obs"
	"textmachine/backend/internal/runevents"
	"textmachine/backend/internal/store"
)

// waverun.go: the R1 WAVE EXECUTOR — the driver switch from the sequential per-chunk-all-stages loop
// (bookrun.go/chunkrun.go, retired) to the precompute pass ($0) → the draft wave (draft stage ∥ over draft CHUNKS) →
// the bank-mining stop → the edit wave (edit stage ∥ over edit UNITS). The load-bearing decoupling (plan §1/§2):
// the DRAFT is the fine per-chunk unit (COGS/coverage/alignment), the EDIT is the coarse per-unit unit
// (a chapter-scale grouping of whole draft chunks — cross-chunk cohesion, D39 point 4). the edit wave reads a unit's draft
// as the concatenation of its members' draft-wave outputs and does a FRESH memory.Select over the whole-unit source;
// it NEVER re-renders the draft stage (the load-bearing invariant — a re-render over the enriched bank would move
// the draft request_hash and re-bill the draft wave). Money stays single-writer-safe (Reserve/Settle serialize); the
// waves add only read-only shared state (clients/templates/memory/rate-guards built in the precompute pass). Per-wave
// snapshots (draft-wave snapshot base-bank, edit-wave snapshot enriched) keep «the re-payment stays ONE»: a the bank-mining stop enrichment moves
// only edit-wave snapshot, so draft-wave checkpoints stay valid.

// wavedStage carries a wave stage with its GLOBAL index in Pipeline.Stages — runStage needs the global
// index for isFinal (the LAST stage is the shipping stage the sanitizer runs on) and for the >0 sizing
// path (a later stage sizes max_tokens from its input draft, not the source, D2.5).
type wavedStage struct {
	st  config.Stage
	idx int
}

// waveStagesIndexed partitions Pipeline.Stages by role for a wave, carrying each stage's global index
// (mirrors waveStages but retains the index runStage needs). the draft wave = translator (draft) stages; the edit wave = every
// non-translator (editor/other) stage. Order preserved.
func (r *Runner) waveStagesIndexed(w wave) []wavedStage {
	var out []wavedStage
	for i, st := range r.Pipeline.Stages {
		isDraft := st.Role == roleTranslator
		if (w == waveDraft) == isDraft {
			out = append(out, wavedStage{st: st, idx: i})
		}
	}
	return out
}

// stageNameSet is the set of a wave's stage names — used by the read-models to classify a stored
// chunk_status row by wave (its snapshot belongs to that wave's per-wave snapshot).
func stageNameSet(staged []wavedStage) map[string]bool {
	m := make(map[string]bool, len(staged))
	for _, ws := range staged {
		m[ws.st.Name] = true
	}
	return m
}

// outputUnits is the SHIPPING granularity the read-models (export/status) project — matching the per-unit
// BookResult: edit units (member draft chunks grouped) for an edit pipeline, or one singleton unit per
// draft chunk for a draft-only pipeline (the draft itself ships). One helper so export and status agree.
func (r *Runner) outputUnits(chunks []chunk.Chunk) []editUnit {
	if r.finalStageWave() == waveEdit {
		return buildEditUnits(chunks)
	}
	out := make([]editUnit, len(chunks))
	for i, ch := range chunks {
		out[i] = editUnit{EditUnitID: ch.EditUnitID, Chapter: ch.Chapter, FirstChunkIdx: ch.ChunkIdx, Members: []chunk.Chunk{ch}}
	}
	return out
}

// WaveSignatureStop is the typed sentinel the driver returns when bank-mining proposes a delta holding
// terms no earlier stop has presented (the D39.144 flag model — see mining.go): the run completes the
// draft wave, writes the owner signature map, records the map in the presented memory, and STOPS before
// the edit wave (an explicit stop boundary, not a mid-run append). The owner reviews the map, decides
// whatever they wish through `tmctl bank-apply` — or nothing — and re-runs: the draft wave resumes at
// $0, the stop does not re-fire on what it already presented, and the edit wave runs with the undecided
// rows riding as unsigned auto-bank surfaces marked ⟨проверить⟩. NOT an infra failure — a deliberate
// human-in-the-loop pause. The CLI maps it (errors.As) to a DISTINCT exit code (3, unlike
// CompletedWithFlags's 2 or a crash's 1) and renders the terms + signature-map path to the operator
// (main.go/renderSignatureStop, R1-FL-A).
type WaveSignatureStop struct {
	Terms         int
	SignaturePath string
	// TablePath is the sidecar holding the FULL verification table; Rows is that same table in memory, so
	// the CLI can print a capped view without re-reading the file (the cap exists because emitRankCap is
	// 200 and a 200-term dump is not a review surface).
	TablePath string
	Rows      []BankStopRow
}

func (e *WaveSignatureStop) Error() string {
	return fmt.Sprintf("bank-mining stopped on new terms — the signature map %s holds %d term(s), new and still-undecided together; decide via `tmctl bank-apply` (or not), then resume (the edit wave)", e.SignaturePath, e.Terms)
}

// translateBookWaves is the wave driver (R1). It runs the draft wave (draft ∥), the bank-mining stop, the edit wave (edit ∥) and
// assembles the BookResult. chunks + stickySel come from the precompute pass (chunk.SplitChunks + precomputeSticky), computed by
// the caller (TranslateBook) after ingest/seed/eager-build. Returns a *WaveSignatureStop when the bank-mining stop stops.
func (r *Runner) translateBookWaves(ctx context.Context, chunks []chunk.Chunk, stickySel []membank.Selection, scope *volumeScope) (*BookResult, error) {
	draftStages := r.waveStagesIndexed(waveDraft)
	editStages := r.waveStagesIndexed(waveEdit)
	workers := r.Pipeline.Waves.Workers
	editWave := len(editStages) > 0 // a real editor wave exists; else the draft IS the shipping output (draft-only)

	// The run-event counters (row 103), seeded from what the store already holds: the same output units
	// and the same per-wave arithmetic `status` projects, so the stream and the resync channel can never
	// quote two different numbers.
	r.beginRunEvents(ctx, chunks)

	// --- the draft wave: draft stage(s) ∥ over draft chunks under draft-wave snapshot (base-bank version) ---
	draftSnapshot, draftPayload, err := r.snapshotIDForWave(waveDraft)
	if err != nil {
		return nil, err
	}
	if err := r.Store.UpsertSnapshot(draftSnapshot, r.Book.BriefHash(), draftPayload); err != nil {
		return nil, fmt.Errorf("pipeline: upsert draft-wave snapshot %.12s: %w", draftSnapshot, err)
	}
	r.Log.InfoContext(ctx, "draft wave started", "snapshot", draftSnapshot[:12],
		"chunks", len(chunks), "workers", workers, "draft_stages", len(draftStages))
	draftResults := make([]stageSeqResult, len(chunks))
	draftUnits := newDraftUnitTracker(r.outputUnits(chunks), chunks)
	if err := r.runWave(ctx, workers, len(chunks), func(ctx context.Context, i int) error {
		// The VOLUME ceiling (volume.go), applied where the item BEGINS rather than after it: a chunk whose
		// output unit is outside this run's granted scope is never started, so no ceiling can be overshot by
		// one call. A nil scope means no ceiling is in force and this is a no-op.
		if !scope.allows(chunks[i]) {
			return nil
		}
		d, err := r.runDraftChunk(ctx, draftSnapshot, chunks[i], stickySel[i], draftStages, !editWave)
		if err != nil {
			return err
		}
		draftResults[i] = d
		// The wave fans out over CHUNKS but the seam counts OUTPUT UNITS (the manifest's granularity, and
		// therefore the reader's): a unit is announced by whichever member finishes last.
		if u, done := draftUnits.memberDone(chunks[i]); done {
			shipped, flagged, reason := draftUnits.outcome(u, draftResults)
			r.events.unitResolved(runevents.WaveDraft, u.Chapter, u.FirstChunkIdx, shipped, flagged, reason)
		}
		return nil
	}); err != nil {
		return nil, err
	}

	r.reportEvicted(ctx, "draft")

	// --- the bank-mining stop: bank-mining stop boundary (auto-continues when mining is not configured) ---
	if stopped, err := r.runBankMiningStop(ctx, chunks, draftSnapshot, editWave); err != nil {
		return nil, err
	} else if stopped {
		// The COUNT travels; the table does not (row 101 ships it as an artifact — a thousand rows are
		// not an event).
		r.events.emit(runevents.TypeBankStop, runevents.BankStop{TermsProposed: r.lastMinedCount})
		return nil, &WaveSignatureStop{
			Terms: r.lastMinedCount, SignaturePath: r.signatureMapPath(),
			TablePath: r.bankStopTablePath(), Rows: r.lastBankStopRows,
		}
	}

	res := &BookResult{BookID: r.Book.BookID}
	// The stop NAMES its ceiling (D39.165 §1б: OpenHands' silent iteration limit is the anti-pattern being
	// avoided). It rides on the result rather than on an error because a volume stop is a COMPLETION — the
	// run did what was bought — so it must not reach the exit-code mapper at all. Only set when the ceiling
	// actually held work back: a run granted more than the book had left simply finished.
	if scope.bound() {
		res.Volume = &scope.stop
		r.Log.InfoContext(ctx, "the run is bounded by the VOLUME ceiling: it will stop having done what was granted, not at the end of the book",
			"book", r.Book.BookID, "max_units", scope.stop.MaxUnits,
			"paid_units", scope.stop.Paid(), "delivered_new", scope.stop.Delivered, "re_made", scope.stop.Reworked,
			"free", scope.stop.Free,
			"left_never_delivered", scope.stop.LeftFresh, "left_delivered_not_re_made", scope.stop.LeftRework)
	}
	// The terminologist's spend is THIS run's spend and belongs in this run's total. It is not a chunk cost,
	// so the per-chunk accumulation below cannot see it — and a paid call that no total reports is a call
	// the operator cannot notice. (Zero when the gate is off or the calls replayed from checkpoints.)
	res.TotalUSD += r.lastTerminology.CostUSD

	// --- Draft-only pipeline: the draft IS the shipping output; assemble per draft chunk (no edit units) ---
	if !editWave {
		for i, ch := range chunks {
			if !scope.allows(ch) {
				continue // outside the volume scope: never drafted, so there is no outcome to assemble
			}
			oc := r.draftOnlyOutcome(ch, draftResults[i])
			res.Chunks = append(res.Chunks, oc)
			res.TotalUSD += oc.CostUSD
			if oc.Disposition == DispFlagged {
				res.Flagged++
			}
		}
		scope.reconcile(res.Chunks)
		r.Log.InfoContext(ctx, "book run finished (draft-only)", "book", r.Book.BookID,
			"chunks", len(res.Chunks), "flagged", res.Flagged, "run_usd", fmt.Sprintf("%.6f", res.TotalUSD))
		return res, nil
	}

	// --- the edit wave: edit stage(s) ∥ over edit units under edit-wave snapshot (enriched-bank version) ---
	editSnapshot, editPayload, err := r.snapshotIDForWave(waveEdit)
	if err != nil {
		return nil, err
	}
	if err := r.Store.UpsertSnapshot(editSnapshot, r.Book.BriefHash(), editPayload); err != nil {
		return nil, fmt.Errorf("pipeline: upsert edit-wave snapshot %.12s: %w", editSnapshot, err)
	}
	units := buildEditUnits(chunks)
	// ⛔ THE PLAN IS RE-JUDGED HERE, because the bank-mining stop above may have re-seeded the bank and
	// moved the snapshot this wave runs under. Units the plan called FREE were admitted outside the grant;
	// any that the new bank makes paying must take a slot or not run. See rescopeEditWave.
	if err := r.rescopeEditWave(ctx, scope, units, chunks, stickySel, editSnapshot); err != nil {
		return nil, err
	}
	// ⚠ AND THE RE-PLAN IS NOT OPTIONAL: the wave refuses to run a scope that was planned against a
	// different snapshot than the one it is about to use. Without this the call above is merely a call —
	// deleting it leaves every test green while the ceiling silently stops holding on any book that
	// re-seeds mid-run. With it, the invariant is the code's own, and removing the re-plan makes the run
	// say so instead of overspending quietly.
	if scope != nil && scope.editSnapshot != editSnapshot {
		return nil, fmt.Errorf("pipeline: the volume scope was planned against edit-wave snapshot %.12s but the wave is running under %.12s — units admitted as free were judged on ground this run has since replaced, and paying for them would escape the ceiling (wave sequencing bug: rescopeEditWave did not run)",
			scope.editSnapshot, editSnapshot)
	}
	draftByKey := make(map[chunkKey]stageSeqResult, len(chunks))
	for i, ch := range chunks {
		draftByKey[chunkKey{ch.Chapter, ch.ChunkIdx}] = draftResults[i]
	}
	r.Log.InfoContext(ctx, "edit wave started", "snapshot", editSnapshot[:12],
		"units", len(units), "workers", workers, "edit_stages", len(editStages))
	unitOutcomes := make([]*ChunkOutcome, len(units))
	if err := r.runWave(ctx, workers, len(units), func(ctx context.Context, i int) error {
		// Same admission as the draft wave, and the SAME set: a unit is either in this run's scope for both
		// waves or in neither, so the edit wave can never be handed a unit whose members were not drafted.
		if !scope.allowsUnit(units[i]) {
			return nil
		}
		oc, err := r.runEditUnit(ctx, editSnapshot, units[i], draftByKey, editStages)
		if err != nil {
			return err
		}
		unitOutcomes[i] = oc
		r.events.unitResolved(runevents.WaveEdit, oc.Chapter, oc.ChunkIdx,
			oc.FinalText != "", oc.Disposition == DispFlagged, string(oc.FlagReason))
		return nil
	}); err != nil {
		return nil, err
	}
	for _, oc := range unitOutcomes {
		if oc == nil {
			continue // outside the volume scope: the unit was never started this run
		}
		res.Chunks = append(res.Chunks, *oc)
		res.TotalUSD += oc.CostUSD
		if oc.Disposition == DispFlagged {
			res.Flagged++
		}
	}
	// The plan's counters become the run's counters only now, corrected by what actually shipped.
	scope.reconcile(res.Chunks)
	r.Log.InfoContext(ctx, "book run finished", "book", r.Book.BookID,
		"units", len(res.Chunks), "flagged", res.Flagged, "run_usd", fmt.Sprintf("%.6f", res.TotalUSD))
	return res, nil
}

// runWave fans `n` items out over `workers` goroutines, calling work(ctx, i) for each index. The first
// error cancels the rest (via a derived context) and is returned; work items are independent (a wave has no
// shared mutable input), and money/store writes serialize through the single-writer pool, so this is
// race-clean. Results are written by the callback into a caller-owned slice indexed by i, so the result
// ORDER is deterministic (item index) regardless of completion order — the golden capture sorts the wire/log
// side-effects separately. workers ≤ 0 is treated as 1 (LoadPipeline already clamps).
//
// PARENT CANCELLATION (Ctrl-C / SIGTERM — the CLI wires a signal.NotifyContext) is surfaced as an error too,
// NOT a silent nil: the feeder drops the remaining items on ctx.Done(), and a worker need not observe the
// cancellation to return (a $0 resume worker resolves entirely through its own opContext store reads, never
// touching the cancellable ctx), so firstErr can stay nil while items are left UNDONE. Returning nil then
// would let the caller index uninitialised result slots — a nil-deref on the edit-wave `*oc` assembly, or a
// draft-only run reporting empty translations as exit-0 success. So after the workers drain, a still-nil
// firstErr defers to parent.Err(): nil on a clean run, context.Canceled on an interrupted one (the durable
// store stays correct — the undone items were never checkpointed, so a re-run resumes them).
func (r *Runner) runWave(parent context.Context, workers, n int, work func(ctx context.Context, i int) error) error {
	if workers < 1 {
		workers = 1
	}
	if n == 0 {
		return nil
	}
	ctx, cancel := context.WithCancel(parent)
	defer cancel()
	idxCh := make(chan int)
	var wg sync.WaitGroup
	var mu sync.Mutex
	var firstErr error
	// A crash gets its OWN slot rather than competing for firstErr, and outranks it on the way out. Both
	// halves are load-bearing. Unrecovered, a worker panic takes the process down through the runtime's
	// handler, which exits 2 — the code the shell contract reserves for "completed with flags", so a run
	// that died mid-book was recorded as `ready` (row 176). But routing it through first-wins would only
	// move the lie: a sibling that already failed cancels the wave, and a panic in the code nobody expected
	// to run under cancellation would then be discarded with its stack, leaving the run to depart as the
	// sibling's ceiling halt (exit 4, `paused`) or cancellation (exit 5). A process that crashed is not
	// paused and did not stop gracefully.
	var panicErr error
	fail := func(err error) {
		mu.Lock()
		if firstErr == nil {
			firstErr = err
			cancel()
		}
		mu.Unlock()
	}
	failPanic := func(err error) {
		mu.Lock()
		if panicErr == nil {
			panicErr = err
		}
		mu.Unlock()
		cancel()
	}
	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			// NOT obs.SafeGo: this recover does not let the run continue, it makes the run fail loudly.
			defer func() {
				if p := recover(); p != nil {
					failPanic(obs.NewPanicError("wave worker", p))
				}
			}()
			for i := range idxCh {
				if err := work(ctx, i); err != nil {
					fail(err)
					return
				}
			}
		}()
	}
	for i := 0; i < n; i++ {
		select {
		case idxCh <- i:
		case <-ctx.Done():
		}
	}
	close(idxCh)
	wg.Wait()
	if panicErr != nil {
		return panicErr
	}
	if firstErr != nil {
		return firstErr
	}
	// No worker failed, but the PARENT may have been cancelled (Ctrl-C), which the feeder honoured by
	// dropping items — surface that as an error so the driver never assembles a partial result as success.
	return parent.Err()
}

// stageSeqResult is the outcome of running one chunk/unit through an ORDERED list of stages (one wave):
// the per-stage results, the final OK stage's text, and the terminal flag state.
type stageSeqResult struct {
	stages     []StageResult
	finalText  string // the last OK stage's Text (empty until a stage runs; unused when flagged)
	flagged    bool
	flagReason FlagReason
	recovered  string       // sanitizer-stripped export text from the flagging stage (D35.4a)
	cost       float64      // THIS run's spend across the sequence
	bankTelem  bankFlags    // the translator-role stage's banknote telemetry (WS4 point 10)
	bankProps  string       // JSON of that stage's PARSED banknote entries (the WHAT channel, D39.36 fix)
	repair     repairResult // the shipping stage's repair outcome (pack-16); zero when the gate is off
}

// runStageSequence runs `staged` in order over one chunk/unit, feeding each stage's output to the next as
// `prev` (starting from startPrev), stopping the sequence at the FIRST flagged stage (later stages are
// recorded `skipped` — no paid edit over a garbage draft, D2). It is the generalized inner loop of the
// retired translateChunk, reused by BOTH waves (draft over a chunk, the edit-wave editor over a unit). injectionByRole
// maps a stage role to its already-rendered memory-injection message (the translator's src→dst glossary in
// the draft wave, the editor's CONFIRMED-dst constraint block in the edit wave). Deterministic; content-verified resume via runStage.
func (r *Runner) runStageSequence(ctx context.Context, staged []wavedStage, snapID string, ch chunk.Chunk, startPrev string, injectionByRole map[string]string, injected []membank.PickedEntry) (stageSeqResult, error) {
	var res stageSeqResult
	prev := startPrev
	flagged := false
	var flagReason FlagReason
	recovered := ""
	for _, ws := range staged {
		st := ws.st
		if flagged {
			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 res, fmt.Errorf("pipeline: record skipped chunk_status: %w", err)
			}
			res.stages = append(res.stages, StageResult{
				Stage: st.Name, Role: st.Role, Model: st.ResolvedModel,
				Disposition: DispSkipped, FlagReason: flagReason, Detail: detail,
			})
			continue
		}
		sr, err := r.runStage(ctx, st, ws.idx, snapID, ch, prev, injectionByRole[st.Role], injected)
		if err != nil {
			return res, err
		}
		res.stages = append(res.stages, *sr)
		res.cost += sr.CostUSD
		if sr.Repair.touched() {
			res.repair = sr.Repair
		}
		if st.Role == roleTranslator {
			res.bankTelem, res.bankProps = sr.BankFlags, sr.BankProposals
			// FL-2: the resume fast-path serves the banknote-cleaned final_hash and leaves BankFlags zero;
			// restore the telemetry the fresh run persisted so persistRetrievalState does not zero the three
			// banknote columns on every resume of a ready chunk (guarded on an empty BankFlags so the
			// attempt-loop path — which re-derives telemetry from the raw checkpoint — is never overwritten).
			if res.bankTelem == (bankFlags{}) && sr.FromResume && r.Pipeline.Gates.Banknote.Enabled {
				if prevRS, err := r.Store.GetRetrievalState(r.Book.BookID, ch.Chapter, ch.ChunkIdx); err == nil && prevRS != nil {
					res.bankTelem = bankFlags{
						NLines: prevRS.NBanknoteLines, ParseFail: prevRS.BanknoteParseFail != 0, Truncated: prevRS.BanknoteTruncated != 0,
					}
					// The proposals restore with the counters and for the same reason: the fast path never
					// touches the raw checkpoint, so re-writing the row from a resumed run would erase the
					// very WHAT the signature map is about to read.
					res.bankProps = prevRS.BanknoteDetail
				}
			}
		}
		if sr.Disposition == DispFlagged {
			flagged = true
			flagReason = sr.FlagReason
			recovered = sr.RecoveredText
			continue
		}
		prev = sr.Text
	}
	res.finalText = prev
	res.flagged = flagged
	res.flagReason = flagReason
	res.recovered = recovered
	return res, nil
}

// runDraftChunk runs the draft stage(s) over one chunk under the draft-wave snapshot (plan §1). It renders
// the translator's src→dst glossary injection from the precomputed per-chunk selection over the BASE bank
// (baseMemory — mined-excluded, so the injection is stable across a bank-mining enrichment), runs the
// sequence, and persists the draft retrieval_state (injection counts + banknote telemetry). When this is the
// FINAL wave (a draft-only pipeline with no editor), the draft IS the shipping text, so the post-check +
// cheap gates run here over the draft; otherwise they belong to the edit wave (the edit unit's final output).
func (r *Runner) runDraftChunk(ctx context.Context, draftSnapshot string, ch chunk.Chunk, memSel membank.Selection, draftStages []wavedStage, isFinalWave bool) (stageSeqResult, error) {
	injectionByRole := map[string]string{}
	if r.baseMemory != nil {
		// Serialize the per-role injection blocks via the role→renderer registry (D39 layer 7) from the
		// precomputed base-bank selection: the translator gets its src→dst glossary block; any other
		// role's block is rendered too but consumed only if a stage of that role runs in this wave.
		tx := lang.InjectionTextsFor(r.Book.TargetLang)
		for role, render := range roleInjectionRenderers {
			injectionByRole[role] = render(memSel.Injected, tx) // pure → map-order-independent
		}
		if len(memSel.TrustGated) > 0 {
			r.Log.WarnContext(ctx, "memory: lower-trust longer key refused from suppressing a higher-trust nested key (approved term preserved; reconcile the seed)",
				"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "trust_gated", len(memSel.TrustGated),
				"first", memSel.TrustGated[0].Suppressor+"⊃"+memSel.TrustGated[0].Protected)
		}
	}
	seq, err := r.runStageSequence(ctx, draftStages, draftSnapshot, ch, "", injectionByRole, memSel.Injected)
	if err != nil {
		return seq, err
	}

	var postMisses membank.PostcheckResult
	outputChecked := false
	var cheap checks.CheapGateResult
	if isFinalWave && r.baseMemory != nil && !seq.flagged && seq.finalText != "" {
		// Draft-only pipeline: the draft is the shipping output → run the FINAL-output checks here, over the
		// SAME base bank whose terms the draft was injected with (post-check parity with the injection).
		outputChecked = true
		postMisses = r.baseMemory.Postcheck(memSel.Injected, seq.finalText)
		if r.Pipeline.Gates.Glossary.PostcheckGate && postMisses.ConfirmedCount() > 0 {
			seq.flagged = true
			seq.flagReason = FlagGlossaryMiss
			r.Log.WarnContext(ctx, "glossary post-check gate flagged the chunk",
				"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "confirmed_misses", postMisses.ConfirmedCount())
		}
	}
	if isFinalWave && !seq.flagged && seq.finalText != "" && r.checkers.TargetActive() {
		cheap = checks.RunCheapGates(ch.Text, seq.finalText, seq.finalText, r.cheapGateConfig())
	}
	var voice checks.VoiceResult
	var leaks []membank.SpoilerLeak
	if isFinalWave && !seq.flagged && seq.finalText != "" {
		voice = r.runVoiceChecks(r.baseMemory, ch.Chapter, ch.Text, seq.finalText)
		leaks = r.spoilerLeaks(r.baseMemory, memSel, seq.finalText)
	}
	if r.baseMemory != nil {
		if err := r.persistRetrievalState(draftSnapshot, ch, memSel, postMisses, outputChecked, cheap, seq.bankTelem, seq.bankProps, voice, leaks); err != nil {
			return seq, err
		}
	}
	return seq, nil
}

// runEditUnit runs the editor stage(s) over one edit unit under the edit-wave snapshot (plan §1/§2). It reads
// the unit's draft as the concatenation of its member chunks' draft outputs (NEVER re-rendering the draft
// stage — the load-bearing invariant), does a FRESH memory.Select over the unit source over the ENRICHED bank
// (sticky degenerate at unit scope → nil), and runs the editor over (unit source, unit draft). Post-check +
// cheap gates run over the edit output at unit granularity, merged into the LEADER chunk's retrieval_state row.
//
// A flagged member draft is DROPPED from the edit, not blanked-whole-unit (c-lite, D39.17-fix): the editor
// still runs over the CLEAN members' source+draft, so the good (already-paid) drafts are edited and shipped —
// only when EVERY member flagged does the unit ship "". The edit row records the edit's honest outcome; the
// UNIT is flagged whenever a member dropped (or the edit itself flagged), a fact the export/status read-models
// re-derive from the member draft rows (status.resolveChunkState) so all three agree. The pre-c-lite behaviour
// (any flagged member → whole unit blank) discarded a chapter-scale unit + its paid sibling drafts over one bad
// chunk — a regression from the retired per-chunk model, closed here.
func (r *Runner) runEditUnit(ctx context.Context, editSnapshot string, unit editUnit, draftByKey map[chunkKey]stageSeqResult, editStages []wavedStage) (*ChunkOutcome, error) {
	out := &ChunkOutcome{Chapter: unit.Chapter, ChunkIdx: unit.FirstChunkIdx, Disposition: DispOK}

	var draftStages []StageResult
	var draftCost float64
	var cleanSources []string // the non-flagged members' source (the editor's {{text}}, aligned to the clean draft)
	var draftParts []string   // the non-flagged members' draft (the editor's input)
	memberFlagged := false
	var memberFlagReason FlagReason
	for _, m := range unit.Members {
		d, ok := draftByKey[chunkKey{m.Chapter, m.ChunkIdx}]
		if !ok {
			return nil, fmt.Errorf("pipeline: edit unit ch%d chunk%d: missing draft result (wave sequencing bug)", m.Chapter, m.ChunkIdx)
		}
		draftStages = append(draftStages, d.stages...)
		draftCost += d.cost
		if d.flagged {
			if !memberFlagged {
				memberFlagged = true
				memberFlagReason = d.flagReason
			}
			out.DroppedMembers++                 // counted for EVERY drop, including when the edit itself flags too
			out.DroppedReason = memberFlagReason // the HOLE's cause, which out.FlagReason may not be
			continue                             // the flagged member is DROPPED from the edit — not skipped-whole-unit (c-lite)
		}
		cleanSources = append(cleanSources, m.Text)
		draftParts = append(draftParts, d.finalText)
	}
	out.CostUSD = draftCost

	// Only when EVERY member draft flagged is there nothing clean to edit → skip the edit and ship "" (the whole
	// unit is unusable). Record a skipped edit chunk_status at the leader so the read-models see a flagged unit.
	if len(draftParts) == 0 {
		leader := chunk.Chunk{Chapter: unit.Chapter, ChunkIdx: unit.FirstChunkIdx}
		skipped, err := r.recordSkippedStages(ctx, editStages, editSnapshot, leader, memberFlagReason)
		if err != nil {
			return nil, err
		}
		out.Stages = append(draftStages, skipped...)
		out.Disposition = DispFlagged
		out.FlagReason = memberFlagReason
		out.FinalText = ""
		return out, nil
	}

	// The edit is ADDRESSED at the manifest leader (chapter, FirstChunkIdx — where the read-models look for the
	// unit's final row) but its SOURCE + DRAFT are the CLEAN members only. With NO flagged member cleanSources
	// join == unit.sourceText() byte-for-byte, so the normal path (and the golden) is unchanged.
	leader := chunk.Chunk{Chapter: unit.Chapter, ChunkIdx: unit.FirstChunkIdx, Text: strings.Join(cleanSources, unitJoinSeparator)}
	unitDraft := strings.Join(draftParts, unitJoinSeparator)
	var editSel membank.Selection
	injectionByRole := map[string]string{}
	if r.memory != nil {
		// FRESH Select over the WHOLE-unit source over the ENRICHED bank (§1(б)/F3): NOT the the precompute pass per-chunk
		// memSel (that was over the base bank). Sticky is degenerate at unit scope (a unit == a chapter or a
		// sub-chapter split, so the intra-chapter sticky window does not apply) → nil sticky_prev.
		editSel = r.memory.Select(leader.Text, unit.Chapter, nil, r.Pipeline.Context.GlossaryTokenBudget)
		// Render the per-role injection from the ENRICHED unit selection via the registry (D39 layer 7): the
		// editor gets its CONFIRMED-dst constraint block; other roles' blocks are rendered but consumed
		// only if a stage of that role runs in the edit wave.
		tx := lang.InjectionTextsFor(r.Book.TargetLang)
		for role, render := range roleInjectionRenderers {
			injectionByRole[role] = render(editSel.Injected, tx)
		}
	}
	seq, err := r.runStageSequence(ctx, editStages, editSnapshot, leader, unitDraft, injectionByRole, editSel.Injected)
	if err != nil {
		return nil, err
	}
	out.Stages = append(draftStages, seq.stages...)
	out.CostUSD += seq.cost

	finalText, flagged, flagReason, recovered := seq.finalText, seq.flagged, seq.flagReason, seq.recovered
	var postMisses membank.PostcheckResult
	outputChecked := false
	if r.memory != nil && !flagged && finalText != "" {
		outputChecked = true
		postMisses = r.memory.Postcheck(editSel.Injected, finalText)
		if r.Pipeline.Gates.Glossary.PostcheckGate && postMisses.ConfirmedCount() > 0 {
			flagged = true
			flagReason = FlagGlossaryMiss
			r.Log.WarnContext(ctx, "glossary post-check gate flagged the edit unit",
				"chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx, "confirmed_misses", postMisses.ConfirmedCount())
		}
	}
	var cheap checks.CheapGateResult
	if !flagged && finalText != "" && r.checkers.TargetActive() {
		cheap = checks.RunCheapGates(leader.Text, unitDraft, finalText, r.cheapGateConfig())
		if cheap.Total() > 0 {
			r.Log.InfoContext(ctx, "cheap style gates flagged the edit unit (observability, not a gate)",
				"chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx, "style_flags", cheap.Total())
		}
	}
	// The pack-19 flaggers, on the same shipped text and under the same "clean unit" condition as the
	// cheap gates: a flagged unit ships nothing, so measuring its prose would report defects nobody reads.
	var voice checks.VoiceResult
	var leaks []membank.SpoilerLeak
	if !flagged && finalText != "" {
		voice = r.runVoiceChecks(r.memory, unit.Chapter, leader.Text, finalText)
		if voice.Total() > 0 {
			r.Log.InfoContext(ctx, "voice flagger fired on the edit unit (observability, not a gate)",
				"chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx,
				"voice_flags", voice.Total(), "attributed_replies", voice.Attributed, "version", checks.VoiceCheckVersion)
		}
		if leaks = r.spoilerLeaks(r.memory, editSel, finalText); len(leaks) > 0 {
			r.Log.WarnContext(ctx, "spoiler leak: a rendering this chapter must not know yet reached the output",
				"chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx, "leaks", len(leaks), "first", leaks[0].Src)
		}
	}
	if r.memory != nil {
		if err := r.mergeUnitRetrievalState(unit.Chapter, unit.FirstChunkIdx, postMisses, outputChecked, cheap, voice, leaks); err != nil {
			return nil, err
		}
	}

	// Unit disposition: the EDIT itself flagged → ship its recovered/""; else a MEMBER dropped → still ship the
	// edited clean remainder, flagged for review (c-lite); else clean → ship the edit output ok.
	switch {
	case flagged:
		out.Disposition = DispFlagged
		out.FlagReason = flagReason
		out.FinalText = r.checkers.ExportNormalize(recovered)
	case memberFlagged:
		out.Disposition = DispFlagged
		out.FlagReason = memberFlagReason
		out.FinalText = r.checkers.ExportNormalize(finalText)
	default:
		out.FinalText = r.checkers.ExportNormalize(finalText)
	}
	// Title policy (pack-13): prepend the chapter's deterministic «Глава N» to its FIRST unit's non-empty
	// export text. unit.Members[0] is the unit's leader chunk; only a chapter's opening chunk (ChunkIdx 0)
	// carries a Heading, so non-leader units are a no-op. Applied to the export projection identically in
	// export.go — both derive the heading from the same deterministic chunker, so translate and export agree.
	out.FinalText = chunk.ApplyHeading(unit.Members[0].Heading, out.FinalText)
	return out, nil
}

// recordSkippedStages writes a `skipped` chunk_status row (+ a skipped StageResult) for each stage in a
// wave, at chunk `ch` under `snapID` — used when a unit's member draft flagged, so the unit's edit is
// skipped (mirrors runStageSequence's flagged-skip path but for a pre-decided skip).
func (r *Runner) recordSkippedStages(ctx context.Context, staged []wavedStage, snapID string, ch chunk.Chunk, flagReason FlagReason) ([]StageResult, error) {
	detail := fmt.Sprintf("skipped: a member draft chunk of this edit unit was flagged (%s)", flagReason)
	var out []StageResult
	for _, ws := range staged {
		st := ws.st
		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 nil, fmt.Errorf("pipeline: record skipped edit chunk_status: %w", err)
		}
		out = append(out, StageResult{
			Stage: st.Name, Role: st.Role, Model: st.ResolvedModel,
			Disposition: DispSkipped, FlagReason: flagReason, Detail: detail,
		})
	}
	return out, nil
}

// mergeUnitRetrievalState folds a edit unit's post-check + cheap-gate observability into the LEADER chunk's
// retrieval_state row — the row the draft wave already wrote with the leader chunk's draft-injection counts + banknote
// telemetry. A read-modify-write (single-writer safe; each unit owns a distinct leader chunk_idx) preserves
// the draft fields and adds the unit-level post-check/style fields, so the read-models aggregate injection
// per draft chunk and post-check/style per unit. A missing row (r.memory nil path) is a no-op.
func (r *Runner) mergeUnitRetrievalState(chapter, firstChunkIdx int, misses membank.PostcheckResult, outputChecked bool, cheap checks.CheapGateResult, voice checks.VoiceResult, leaks []membank.SpoilerLeak) error {
	rs, err := r.Store.GetRetrievalState(r.Book.BookID, chapter, firstChunkIdx)
	if err != nil {
		return fmt.Errorf("pipeline: read leader retrieval_state ch%d/chunk%d: %w", chapter, firstChunkIdx, err)
	}
	if rs == nil {
		return nil // no draft row (no memory / no draft) → nothing to merge onto
	}
	rs.NStyleFlags = cheap.Total()
	rs.StyleDetail = ""
	if cheap.Total() > 0 {
		if b, err := json.Marshal(cheap); err == nil {
			rs.StyleDetail = string(b)
		}
	}
	setVoiceState(rs, voice, leaks)
	rs.NPostcheckMiss = 0
	rs.PostcheckDetail = ""
	if outputChecked {
		rs.NPostcheckMiss = misses.ConfirmedCount()
		rs.NUnverifiedShown, rs.NUnverifiedFollowed = misses.Shown, misses.Followed
		if all := misses.All(); len(all) > 0 {
			if b, err := json.Marshal(all); err == nil {
				rs.PostcheckDetail = string(b)
			}
		}
	}
	return r.Store.UpsertRetrievalState(*rs)
}

// draftOnlyOutcome assembles a ChunkOutcome for a draft-only pipeline (no editor): the draft is the shipping
// output, so the per-chunk draft result maps straight to the outcome (post-check/cheap already ran in
// runDraftChunk as the final wave). Mirrors the export contract: a cosmetic strip exports its recovered
// text; every other flag exports "".
func (r *Runner) draftOnlyOutcome(ch chunk.Chunk, d stageSeqResult) ChunkOutcome {
	oc := ChunkOutcome{Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stages: d.stages, CostUSD: d.cost, Disposition: DispOK}
	if d.flagged {
		oc.Disposition = DispFlagged
		oc.FlagReason = d.flagReason
		oc.FinalText = r.checkers.ExportNormalize(d.recovered)
	} else {
		oc.FinalText = r.checkers.ExportNormalize(d.finalText)
	}
	oc.FinalText = chunk.ApplyHeading(ch.Heading, oc.FinalText) // pack-13 title policy (draft-only shipping path)
	return oc
}