textmachine/backend/internal/pipeline/export.go

package pipeline

import (
	"fmt"
	"sort"
	"strings"
	"time"

	"textmachine/backend/internal/chunk"
	"textmachine/backend/internal/store"
)

// export.go: the read-only EXPORT projection (D39 layer 6 / D39.2-open №1; the deferred `tmctl export`
// of D15.2-stage-B in its MINIMAL read form). It emits every chunk's FINAL export text EXACTLY by prod
// semantics, so the polygon extractor (records.json / exp12_extract) and reader-samples read the SAME
// export-normalised bytes `tmctl translate` ships — instead of the RAW stored checkpoint, which skips
// the export contract for a clean chunk. The owner chose a BACKEND surface over a Python mirror of
// checks.ExportNormalize (which would drift against the Go contract). It is a PURE READ-ONLY projection like
// Status/QualityReport: $0, no LLM, no keys, no snapshot touch — it reads the persisted chunk_status +
// the $0 final checkpoint and applies the very same checks.ExportNormalize the ChunkOutcome does. Only the READ
// half is built here; the annotation/override half of the D15.2 `export` command is deliberately NOT.

// exportPending marks a manifest chunk that has not (yet) reached the final stage — no exported text.
const exportPending = "pending"

// ChunkExport is one chunk's final export record: the metadata plus the export-normalised final text.
type ChunkExport struct {
	Chapter    int    `json:"chapter"`
	ChunkIdx   int    `json:"chunk_idx"`
	SnapshotID string `json:"snapshot_id,omitempty"`
	// Disposition is the CHUNK-level verdict mirroring ChunkOutcome: "ok" only when the final stage is
	// ok; "flagged" for a flagged OR upstream-skipped final row; "pending" for a manifest chunk with no
	// final-stage row (not yet translated).
	Disposition string `json:"disposition"`
	FlagReason  string `json:"flag_reason,omitempty"`
	Detail      string `json:"detail,omitempty"`
	// FinalText is the export-normalised final text EXACTLY as `tmctl translate` ships it:
	// r.checkers.ExportNormalize(final checkpoint response) for an ok or cosmetic-stripped chunk, "" for a
	// substantive flag / upstream-skip / pending chunk (r.checkers.ExportNormalize("")=="", so the paths coincide
	// byte-for-byte with ChunkOutcome.FinalText).
	FinalText string `json:"final_text"`
	// Source is the chunk's normalized source text — populated ONLY under `tmctl export --pairs` (WS5
	// §5(д) / research/20 §E-3), the src↔target column the polygon needs to measure the DC1 (时辰) / DC2
	// (千万) source-vs-target checkers' false-positive rate. Omitted by default (the export contract is
	// target-only); the source is the $0 manifest re-chunk, never a stored/billed artifact.
	Source string `json:"source,omitempty"`
	// DroppedMembers counts the member chunks the c-lite editor left OUT of this unit's edit, so a
	// consumer can say "the text below is INCOMPLETE" without inferring it from FlagReason — which
	// cannot carry it: an edit flagged for its OWN reason (a cosmetic sanitizer strip) keeps that
	// reason even when a member also dropped, and the reason then describes the strip and is silent
	// about the missing text. Additive and omitempty: a unit that lost nothing serializes exactly as
	// before this field existed.
	DroppedMembers int `json:"dropped_members,omitempty"`
	// DroppedReason is why the FIRST dropped member flagged — the cause of the HOLE, which FlagReason
	// is not when the edit flagged on its own account as well. "" when nothing dropped.
	DroppedReason string `json:"dropped_reason,omitempty"`
	// Heading is the chapter title the engine renders for this unit's chapter — the SAME literal
	// chunk.ApplyHeading glued onto the front of FinalText (u.Members[0].Heading at assembly time), given
	// apart so a consumer that wants the title separate from the prose strips a KNOWN prefix instead of
	// guessing at the text. Non-empty only on a chapter-opening unit of a book with a heading rule
	// (D39.100 п.1: "" is a legal state — a book without a rule, a chapter without a structural marker),
	// and present on a pending unit too (it comes from the manifest, not from a run). Additive and
	// omitempty: the export of a book without headings serializes exactly as before this field existed.
	// ⚠ It is the engine's TEMPORARY render (D39.122 п.2д), not a label the book's data carries.
	Heading string `json:"heading,omitempty"`
}

// UnitRef addresses one output unit by its manifest key.
type UnitRef struct {
	Chapter  int `json:"chapter"`
	ChunkIdx int `json:"chunk_idx"`
}

// exportVersion versions the SHAPE of the export document.
const exportVersion = "tm-export-v1"

// BookExport is the whole-book export projection (chunks in manifest — (chapter, chunk_idx) — order, so
// the JSON is deterministic across runs). The counters make a partial/empty/drifted book EXPLICIT
// instead of silently exporting as complete (F4/F3).
type BookExport struct {
	// Version is the SHAPE of this document (17-seam-inbound-law п.3: every JSON output of the engine
	// carries a versioned envelope). The law names `status --json` as the gap because that is where it
	// was noticed; the export was the second surface without one.
	Version string `json:"export_version"`
	BookID  string `json:"book_id"`
	// TotalUnits is the current source manifest size in output units (the honest denominator); the exported
	// units + PendingUnits == TotalUnits. GhostRows are stored final rows dropped as OUTSIDE the manifest.
	TotalUnits   int `json:"total_units"`
	PendingUnits int `json:"pending_units"`
	GhostRows    int `json:"ghost_rows,omitempty"`
	// ConfigDrift is true when the CURRENT config renders a snapshot different from the one the stored
	// rows carry (a gate flip / prompt bump / stage rename since the run) — the gate/stage re-derivation
	// below may then not match what translate actually did. CurrentSnapshot is that projected id.
	ConfigDrift bool `json:"config_drift"`
	// ConfigDriftBasis says what the boolean is a verdict OF — none | drift | unknown (driftbasis.go).
	// This surface needed it MOST: it is the one place in the engine that logged «drift state unknown
	// (reported as none)» and then reported none, and `build` GATES on the boolean (staleUnits), so an
	// un-run check silently produced `stale: 0` over an unknown state — a field without a basis
	// corrupting the neighbouring field that has one.
	ConfigDriftBasis string `json:"config_drift_basis"`
	CurrentSnapshot  string `json:"current_snapshot,omitempty"`
	// GhostUnits are the rows GhostRows counts, BY KEY — so a consumer can say WHICH chapter has
	// translated text the current cut of the book cannot place, not only that some chapter does. The
	// book writer marks those chapters (a coarsened cut leaves the joined leader row carrying the text
	// of a SHORTER unit than the current one; the rest of the source's translation sits in these rows
	// and reaches no file). Sorted by key; omitempty (a book without ghosts serializes as before).
	GhostUnits []UnitRef `json:"ghost_units,omitempty"`
	// TextModified is the newest chunk_status.updated_at among the FINAL rows this export joined — the
	// rows whose text (or whose withholding) is in Chunks — rendered as CCYY-MM-DDThh:mm:ssZ: the last
	// time the engine wrote the shipped state of this book. Draft rows, ghost rows and rows of stages
	// that do not ship are not consulted, so a $0 draft-wave re-write moves nothing here. A fact of the
	// store rather than of any process's clock, which is why the book writer uses it as the file's
	// modification time (two builds of one store then agree byte-for-byte). "" when no unit has a final
	// row (a book that has shipped nothing). Additive and omitempty. ⚠ «Last written», not «last
	// changed»: a resume that re-writes an unchanged final row moves it too (store.ChunkStatus.UpdatedAt).
	TextModified string `json:"text_modified,omitempty"`
	// Chunks is always non-nil (an empty book exports [] not null).
	Chunks []ChunkExport `json:"chunks"`
}

// Export builds the read-only per-chunk export projection. It opens no jobs, reserves nothing, makes no
// LLM call — it joins the $0 source manifest against the persisted chunk_status (final-stage rows) and,
// for a chunk that produced exported text, the $0 final checkpoint, applying checks.ExportNormalize exactly as
// translateChunk does. Safe to run whenever `report`/`status` are (the same exclusive-lock rule).
// Deterministic over the store + source.
//
// CAVEATS (surfaced, never silent):
//   - CONFIG/GATE DRIFT (F3): the gate flag AND the final-stage NAME are read from the CURRENT config.
//     A config edit since the run (a glossary post-check gate FLIP, a prompt bump, a final-stage rename)
//     changes the projected snapshot; Export sets ConfigDrift + WARNs so the gate re-derivation / row
//     lookup mismatch is visible (a run under the same config reads correctly; `status` shows it too).
//   - MANIFEST (F4): a chunk absent from the current source is a GHOST row (dropped + counted + WARNed);
//     a manifest chunk with no final row is PENDING; both keep a partial/edited book from exporting as
//     if complete (the D37-skew the polygon extractor must not inherit).
func (r *Runner) Export(pairs bool) (*BookExport, error) {
	statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID)
	if err != nil {
		return nil, err
	}
	gateMisses, err := r.exportGateMisses()
	if err != nil {
		return nil, err
	}
	// F4: the $0 source manifest is authoritative. Under the wave model the export granularity is the EDIT
	// UNIT (the editor collapses a unit's member chunks into ONE final text), so join the manifest into
	// output units — each a (chapter, firstChunkIdx) key + the unit's joined source (the --pairs column
	// aligned byte-for-byte to how the edit wave concatenated the members). A draft-only pipeline has one output per
	// draft chunk (the draft ships). Iterating units keeps a PENDING unit explicit and a GHOST row visible.
	// The persisted manifest serves the DEFAULT export (row 100): the projection joins stored rows against
	// positions and prepends the chapter title, both of which the manifest carries. `--pairs` is the one
	// mode that emits the SOURCE column, so it demands the full re-chunk — a text-free manifest would ship
	// an empty src column to the FP-measure, which is worse than being slow.
	manifest, err := r.exportChunks(pairs)
	if err != nil {
		return nil, err
	}
	units := r.outputUnits(manifest)
	finalRow := r.finalStageRows(statuses)
	byChunk := map[chunkKey][]store.ChunkStatus{}
	for _, cs := range statuses {
		byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}] = append(byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}], cs)
	}
	draftStageNames := stageNameSet(r.waveStagesIndexed(waveDraft))

	exp := &BookExport{Version: exportVersion, BookID: r.Book.BookID, Chunks: []ChunkExport{}}
	inManifest := map[chunkKey]bool{}
	var shipped []store.ChunkStatus // the final rows the export joined — what TextModified is read from
	for _, u := range units {
		key := chunkKey{u.Chapter, u.FirstChunkIdx}
		inManifest[key] = true
		exp.TotalUnits++
		cs, ok := finalRow[key]
		if !ok {
			pend := ChunkExport{Chapter: u.Chapter, ChunkIdx: u.FirstChunkIdx, Disposition: exportPending, Heading: u.Members[0].Heading}
			if pairs {
				pend.Source = u.sourceText()
			}
			exp.PendingUnits++
			exp.Chunks = append(exp.Chunks, pend)
			continue
		}
		ce, err := r.exportUnit(u, cs, byChunk, draftStageNames, gateMisses[key], pairs)
		if err != nil {
			return nil, err
		}
		exp.Chunks = append(exp.Chunks, ce)
		shipped = append(shipped, cs)
	}
	// F4: GHOST rows — a stored final row whose unit-leader key is NOT in the current manifest units
	// (source shrunk since the run). Dropped (never exported as if live) but counted + WARNed.
	for k := range finalRow {
		if !inManifest[k] {
			exp.GhostRows++
			exp.GhostUnits = append(exp.GhostUnits, UnitRef{Chapter: k.chapter, ChunkIdx: k.chunkIdx})
		}
	}
	sort.Slice(exp.GhostUnits, func(i, j int) bool {
		a, b := exp.GhostUnits[i], exp.GhostUnits[j]
		return a.Chapter < b.Chapter || (a.Chapter == b.Chapter && a.ChunkIdx < b.ChunkIdx)
	})
	if exp.GhostRows > 0 {
		r.Log.Warn("export: dropped chunk_status rows outside the current source manifest (source shrunk since the run?)",
			"book", r.Book.BookID, "ghost_rows", exp.GhostRows)
	}
	modified, err := newestStatusWrite(shipped)
	if err != nil {
		return nil, err
	}
	exp.TextModified = modified
	r.exportConfigDrift(statuses, draftStageNames, exp)
	return exp, nil
}

// storeTimeLayout is how SQLite's datetime('now') spells chunk_status.updated_at (UTC to the second).
const storeTimeLayout = "2006-01-02 15:04:05"

// newestStatusWrite renders the newest updated_at among the given rows (the joined FINAL rows) as
// BookExport.TextModified. Rows carry the store's own spelling, which sorts lexically, so the newest is
// the string maximum; it is re-parsed rather than re-spelled by hand so a row that is NOT in the store's
// form is a loud inconsistent-store error and not a silently mangled timestamp. "" when there is no row.
func newestStatusWrite(statuses []store.ChunkStatus) (string, error) {
	newest := ""
	for _, cs := range statuses {
		if cs.UpdatedAt > newest {
			newest = cs.UpdatedAt
		}
	}
	if newest == "" {
		return "", nil
	}
	t, err := time.Parse(storeTimeLayout, newest)
	if err != nil {
		return "", fmt.Errorf("pipeline: export: chunk_status.updated_at %q is not in the store's form (inconsistent store): %w", newest, err)
	}
	return t.UTC().Format("2006-01-02T15:04:05Z"), nil
}

// exportChunks resolves the manifest the export projects over: the persisted one when the export does
// not need source text, the full ingest+split when it does (`--pairs`). One helper so the "who needs
// text" rule lives in one place instead of being re-decided at each call.
func (r *Runner) exportChunks(pairs bool) ([]chunk.Chunk, error) {
	if pairs {
		return r.bookChunks()
	}
	chunks, _, err := r.readModelChunks()
	return chunks, err
}

// exportGateMisses returns the CONFIRMED post-check miss count per chunk, or an empty map when the
// glossary post-check gate is off.
//
// The gate (opt-in) flips a chunk to flagged/glossary_miss at the CHUNK level AFTER the stage loop, so it
// is NEVER written to chunk_status — every stage row stays DispOK (chunkrun.translateChunk records only
// retrieval_state). Export must re-derive it, exactly as Status does: a raw chunk_status read would
// report a gate-flagged chunk as clean ok AND ship the contaminated text `tmctl translate` deliberately
// withheld. retrieval_state is read only when the gate is on (else the map stays empty and every lookup
// is a no-op).
func (r *Runner) exportGateMisses() (map[chunkKey]int, error) {
	misses := map[chunkKey]int{}
	if !r.Pipeline.Gates.Glossary.PostcheckGate {
		return misses, nil
	}
	states, err := r.Store.RetrievalStatesForBook(r.Book.BookID)
	if err != nil {
		return nil, err
	}
	for _, rs := range states {
		misses[chunkKey{rs.Chapter, rs.ChunkIdx}] = rs.NPostcheckMiss // CONFIRMED-miss count
	}
	return misses, nil
}

// finalStageRows indexes the FINAL stage's rows by their addressing key — where the exported text and the
// unit's verdict live. Under the R1 wave model the shipping stage is the editor, run per EDIT UNIT, so its
// rows exist only at each unit's LEADER (chapter, firstChunkIdx), NOT per draft chunk: the result holds one
// entry per output unit (per draft chunk for a draft-only pipeline). All of them carry the FINAL wave's
// snapshot (the edit-wave snapshot for an edit pipeline, the draft-wave one for draft-only).
func (r *Runner) finalStageRows(statuses []store.ChunkStatus) map[chunkKey]store.ChunkStatus {
	lastStage := ""
	if n := len(r.Pipeline.Stages); n > 0 {
		lastStage = r.Pipeline.Stages[n-1].Name
	}
	rows := map[chunkKey]store.ChunkStatus{}
	for _, cs := range statuses {
		if cs.Stage != lastStage {
			continue
		}
		rows[chunkKey{cs.Chapter, cs.ChunkIdx}] = cs
	}
	return rows
}

// exportUnit projects ONE output unit with a stored final row into its export record: the final-row
// verdict and text (chunkExport), the c-lite member-drop override, the deterministic chapter title, and —
// under --pairs — the source column aligned to what the editor actually saw.
func (r *Runner) exportUnit(u editUnit, cs store.ChunkStatus, byChunk map[chunkKey][]store.ChunkStatus,
	draftStageNames map[string]bool, gateMiss int, pairs bool) (ChunkExport, error) {
	ce, err := r.chunkExport(cs, r.Pipeline.Gates.Glossary.PostcheckGate, gateMiss)
	if err != nil {
		return ChunkExport{}, err
	}
	// c-lite (D39.17-fix): the wave editor edits the CLEAN members of a unit even when one member draft
	// flagged — the edit row is DispOK and SHIPS the edited clean remainder, but the UNIT is flagged because
	// a member was dropped. The rule has ONE definition (memberDrops, status.go); export applies it here
	// because its projection sees only the final row, never the member rows status folds. The record carries
	// the FIRST dropped member's reason AND detail, so it is self-consistent rather than the ok edit's blank
	// detail, and ce.FinalText (the edited clean remainder) still ships.
	drops := memberDrops(u, byChunk, draftStageNames)
	// ⚠ A DRAFT-ONLY pipeline has no members to drop: the unit IS one draft chunk and that chunk's own
	// draft row is its FINAL row, so memberDrops — which looks for a flagged DRAFT row among the unit's
	// members — reports the unit's own flag as a lost member. Reading that as a hole would announce
	// «a fragment is missing» over a chunk whose text is entirely present (a cosmetically stripped one
	// ships all of its prose), which is the anti-scope this pack is explicitly bound by: a marker on
	// complete text misinforms. Before the DroppedMembers field the mistake was impossible by accident
	// — droppedAny also demanded an ok edit — so the guard has to be stated now that the count is kept
	// on its own.
	if r.finalStageWave() == waveEdit {
		// Recorded for EVERY drop, not only for the one that overrides an ok edit: the unit whose edit
		// flagged on its own account lost the member's text just the same, and that is the case the
		// reason-only reading was blind to.
		ce.DroppedMembers = len(drops)
		if len(drops) > 0 {
			ce.DroppedReason = drops[0].Reason
		}
	} else {
		drops = nil
	}
	droppedAny := len(drops) > 0 && ce.Disposition == string(DispOK)
	if droppedAny {
		ce.Disposition = string(DispFlagged)
		ce.FlagReason = drops[0].Reason
		ce.Detail = drops[0].Detail
	}
	// Title policy (pack-13): prepend the chapter's deterministic «Глава N» to its FIRST unit's non-empty
	// export text — the SAME projection waverun's outcome assembly applies, both derived from the
	// deterministic chunker (u.Members[0] is the leader; only a chapter-opening chunk carries a Heading),
	// so `tmctl export` and `tmctl translate` ship byte-identical text. The --pairs Source column is left
	// as the (heading-stripped) source manifest — the DC FP-measure aligns src↔target on the body prose,
	// not the deterministic title (which no checker inspects).
	ce.FinalText = chunk.ApplyHeading(u.Members[0].Heading, ce.FinalText)
	ce.Heading = u.Members[0].Heading // the same literal, given apart (see the field)
	if !pairs {
		return ce, nil
	}
	ce.Source = u.sourceText() // the unit src↔target column for the DC1/DC2 FP-measure (--pairs)
	if droppedAny {
		// The target is the edit over the CLEAN members only — align the --pairs source to match (exclude
		// the dropped members, byte-for-byte as runEditUnit's cleanSources join) so the DC1/DC2 FP-measure
		// sees a real src↔target pair, not the dropped member's source terms with no target counterpart.
		dropped := make(map[chunkKey]bool, len(drops))
		for _, d := range drops {
			dropped[d.Key] = true
		}
		var clean []string
		for _, m := range u.Members {
			if !dropped[chunkKey{m.Chapter, m.ChunkIdx}] {
				clean = append(clean, m.Text)
			}
		}
		ce.Source = strings.Join(clean, unitJoinSeparator)
	}
	return ce, nil
}

// exportConfigDrift sets ConfigDrift/CurrentSnapshot when the CURRENT config would render a different
// snapshot than the stored rows carry (F3).
//
// The stored rows carry PER-WAVE snapshots (draft rows → draft-wave snapshot, edit rows → edit-wave
// snapshot), so each wave is compared against its own current projection (like status). A change in
// EITHER wave means `translate` would --resnapshot (a draft-config change re-pins the draft wave and
// cascades to the edit via content-hash). The check is meaningful only when a wave's rows carry a single
// snapshot — a mid-book multi-snapshot drift is already visible in the per-row snapshot_id.
func (r *Runner) exportConfigDrift(statuses []store.ChunkStatus, draftStageNames map[string]bool, exp *BookExport) {
	editStageNames := stageNameSet(r.waveStagesIndexed(waveEdit))
	draftSnaps, editSnaps := map[string]bool{}, map[string]bool{}
	for _, cs := range statuses {
		if cs.SnapshotID == "" {
			continue
		}
		switch {
		case draftStageNames[cs.Stage]:
			draftSnaps[cs.SnapshotID] = true
		case editStageNames[cs.Stage]:
			editSnaps[cs.SnapshotID] = true
		}
	}
	// ⚠ THE STORED GLOSSARY ON PURPOSE, and NOT the fold `status` uses. The two surfaces answer two
	// different questions and this pack deliberately moved only one of them.
	//
	// status asks "what would the NEXT run cost", so it must fold the decision files: money is the point,
	// and a bank-apply that has not been run yet is exactly the spend it exists to reveal (row 231).
	// export asks whether THIS document is consistent with the run that produced it — its consumer is the
	// polygon's extraction, which uses ConfigDrift to decide whether an export is trustworthy to MEASURE.
	// An export whose text is byte-identical to what its run shipped is perfectly measurable even with a
	// decision file sitting un-run beside the database, and folding that file here would mark every such
	// export drifted and quietly disqualify sound measurements in another zone.
	//
	// This was briefly changed to the fold for symmetry and changed back: symmetry between two surfaces
	// that answer different questions is not a property worth having, and redefining a field another zone
	// consumes was not this pack's order. The divergence is deliberate, and it is named here so the next
	// reader does not "fix" it either.
	exp.ConfigDriftBasis = DriftBasisUnknown // until the check actually runs, `false` is not an answer
	if err := r.projectStoredMemory(); err != nil {
		r.Log.Warn("export: config-drift check failed; drift state is UNKNOWN, not none (config_drift_basis=unknown)", "err", err)
		return
	}
	// The SAME precondition `status` applies, from one definition: a book with no stored snapshot has
	// nothing to compare, and `false` there is not an answer.
	ran := driftCheckable(statuses)
	if !ran {
		r.Log.Warn("export: config-drift not checked: no stored row carries a snapshot id, so drift is UNKNOWN, not none",
			"book", r.Book.BookID)
	}
	checkWave := func(snaps map[string]bool, w wave) {
		if len(snaps) != 1 {
			return
		}
		var stored string
		for s := range snaps {
			stored = s
		}
		cur, _, serr := r.snapshotIDForWave(w)
		if serr != nil {
			r.Log.Warn("export: config-drift check failed for a wave; drift state is UNKNOWN, not none", "err", serr)
			ran = false
			return
		}
		if cur != stored {
			exp.ConfigDrift = true
			exp.CurrentSnapshot = cur
			r.Log.Warn("export: CONFIG-DRIFT — current config renders a different snapshot than the stored rows; the gate/stage re-derivation may not match the run (translate would require --resnapshot)",
				"book", r.Book.BookID, "stored", stored, "current", cur, "wave", w)
		}
	}
	checkWave(draftSnaps, waveDraft)
	checkWave(editSnaps, waveEdit)
	// A stored row for a STAGE the current pipeline does not run is drift of the loudest kind — a stage
	// renamed or removed since the run (the CAVEATS' «final-stage rename»). The per-wave check above only
	// compares the snapshots of stages that still exist, so without this the removal of the editor stage
	// makes the draft rows the shipping rows and nothing fires: an unedited book reads as complete.
	// ONE definition of the orphan-stage rule, shared with `status` (orphanStageRows) — the divergence
	// between the two surfaces is what backlog row 239 was.
	if stage, orphan := orphanStageRows(statuses, draftStageNames, editStageNames); orphan {
		exp.ConfigDrift = true
		if exp.CurrentSnapshot == "" {
			if cur, _, serr := r.snapshotIDForWave(r.finalStageWave()); serr == nil {
				exp.CurrentSnapshot = cur
			}
		}
		r.Log.Warn("export: CONFIG-DRIFT — stored rows carry a stage the current config does not run (renamed or removed since the run); the shipping rows are not the ones the run shipped",
			"book", r.Book.BookID, "stage", stage)
	}
	exp.ConfigDriftBasis = driftBasisFor(ran, exp.ConfigDrift)
}

// chunkExport computes one chunk's export record from its final-stage row, exactly by prod FinalText
// semantics (chunkrun.translateChunk): an ok chunk exports r.checkers.ExportNormalize(primary checkpoint), a
// cosmetic sanitizer-strip exports r.checkers.ExportNormalize(its derived stripped checkpoint) — both via
// final_hash — the glossary gate override exports "", and every other flag / upstream skip exports "".
func (r *Runner) chunkExport(cs store.ChunkStatus, gateOn bool, gateMiss int) (ChunkExport, error) {
	ce := ChunkExport{Chapter: cs.Chapter, ChunkIdx: cs.ChunkIdx, SnapshotID: cs.SnapshotID, Detail: cs.Detail}
	// Post-check gate override (mirrors translateChunk): an otherwise-ok chunk (DispOK ⇒ every stage ok)
	// with a CONFIRMED miss is flagged glossary_miss and exports NOTHING — the withheld text never ships.
	if gateOn && cs.Disposition == string(DispOK) && gateMiss > 0 {
		ce.Disposition = string(DispFlagged)
		ce.FlagReason = string(FlagGlossaryMiss)
		return ce, nil
	}
	if cs.Disposition == string(DispOK) {
		ce.Disposition = string(DispOK)
		if cs.FinalHash == "" {
			// F9: an ok chunk MUST carry a final_hash — an empty one is an inconsistent store. Fail LOUD
			// (parity with the missing-checkpoint case), never silently export "".
			return ce, fmt.Errorf("pipeline: export ch%d/chunk%d: DispOK row has an empty final_hash (inconsistent store)", cs.Chapter, cs.ChunkIdx)
		}
	} else {
		// A flagged final row OR a skipped one (an upstream stage flagged) is a flagged CHUNK; the final
		// row carries the propagated flag reason (chunkrun records it on the skipped row too).
		ce.Disposition = string(DispFlagged)
		ce.FlagReason = cs.FlagReason
	}
	if cs.FinalHash != "" && (cs.Disposition == string(DispOK) || cs.FlagReason == string(FlagSanitizerStripped)) {
		cp, err := r.Store.GetCheckpoint(cs.FinalHash)
		if err != nil {
			return ce, err
		}
		if cp == nil {
			return ce, fmt.Errorf("pipeline: export ch%d/chunk%d: chunk_status.final_hash %.12s has no checkpoint (inconsistent store)", cs.Chapter, cs.ChunkIdx, cs.FinalHash)
		}
		ce.FinalText = r.checkers.ExportNormalize(cp.ResponseText)
	}
	return ce, nil
}