textmachine/backend/internal/pipeline/seeding.go

package pipeline

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"textmachine/backend/internal/store"
)

// seeding.go: the job's deterministic $0 preludes — REPLACE-seed the glossary from the
// manual file + ruby readings (D16.4) and persist the ruby aggregates. Run BEFORE
// snapshotID: the frozen approved rows enter memoryVersion (F1).

// seedGlossary REPLACES the book's glossary from its deterministic inputs — the manual
// seed file (curated approved/draft terms) and the captured ruby readings (classified
// into auto candidates) — then MATERIALIZES the frozen bank for this job (r.memory). Run
// once before snapshotID: the frozen APPROVED rows are hashed into memoryVersion (F1), so
// editing the seed is a loud --resnapshot. Idempotent (full replace), $0, no LLM. B2
// approved dst-collisions are logged (not fatal — some collisions are legitimate).
func (r *Runner) seedGlossary(ctx context.Context) error {
	var entries []store.GlossaryEntry
	if r.Book.GlossarySeed != "" {
		seed, err := loadGlossarySeed(r.Book.GlossarySeed)
		if err != nil {
			return err
		}
		entries = append(entries, seed...)
	}
	manualSrcs := map[string]bool{}
	for _, e := range entries {
		manualSrcs[e.Src] = true
	}
	ruby, err := r.Store.RubyReadingsForBook(r.Book.BookID)
	if err != nil {
		return fmt.Errorf("pipeline: read ruby readings for %s: %w", r.Book.BookID, err)
	}
	// D16.4: attach a manual term's kana ruby-reading as an alias (kana spelling matchable)
	// BEFORE appending the auto-candidates, so it only touches the curated manual entries. A
	// reading that would collide with a different seeded term (homophone) is skipped+logged,
	// not attached (which would fail the book loud on an alias the operator cannot edit out).
	if skipped := attachRubyAliasesToManual(entries, ruby); len(skipped) > 0 {
		r.Log.WarnContext(ctx, "ruby kana-alias skipped as a homophone collision (kana form left unmatchable; disambiguate in the seed if needed)",
			"skipped", strings.Join(skipped, "; "))
	}
	// Mined-write path (R1, plan §1(в)/F2): the owner-curated mined-delta file joins the seed as
	// Source:"mined" (loadMinedDelta re-stamps the seed loader's Source), so its terms fold into the
	// ENRICHED bank version but NOT the base — a re-run that adds signed mined terms moves ONLY snapshot_W2.
	// Loaded AFTER the seed/ruby so manualSrcs already reflects the curated seed. A `mined` term that
	// duplicates a seed src is caught by approvedSharedKeyCollisions below like any other collision.
	minedDelta, err := r.loadMinedDelta()
	if err != nil {
		return err
	}
	// D39.20 deviation-#1 fix: a mined-delta term whose UNIQUE key (src, sense, since_ch, until_ch —
	// store/migrate.go glossary UNIQUE) already exists in the SIGNED seed makes the flat INSERT in
	// ReplaceGlossary crash on that constraint and abort the whole paid run. approvedSharedKeyCollisions
	// below does NOT catch it (it skips a SAME-dst duplicate, and keys on the firing surface, not the
	// UNIQUE tuple). Fail LOUD here with the duplicate list + a fix hint (edit the seed OR drop it from
	// the delta) — NEVER a silent merge over the signed seed. Checked BEFORE the append so the delta rows
	// are still separable. Deterministic (seed order).
	if dups := minedDeltaSeedCollisions(entries, minedDelta); len(dups) > 0 {
		return fmt.Errorf("pipeline: mined-delta %s duplicates term(s) already in the signed seed (would crash ReplaceGlossary on the glossary UNIQUE(book_id,src,sense,since_ch,until_ch)):\n  - %s\n  fix: correct the term in the seed, or remove it from the mined-delta file — never both (no silent merge over the signed seed)",
			r.Book.MinedDelta, strings.Join(dups, "\n  - "))
	}
	entries = append(entries, minedDelta...)
	for i := range entries {
		entries[i].BookID = r.Book.BookID
	}
	// Task-6 re-audit: fail loud on a firing key shared by two DIFFERENT approved terms with
	// different dst + overlapping windows (the alias generalization of the D16.1 polysemy
	// livelock) — checked over the FULL entry set (incl. ruby-attached aliases) before persisting.
	if cols := approvedSharedKeyCollisions(entries); len(cols) > 0 {
		return fmt.Errorf("pipeline: glossary shared-key collisions (A2 / D16.1 livelock class):\n  - %s", strings.Join(cols, "\n  - "))
	}
	if err := r.Store.ReplaceGlossary(r.Book.BookID, entries); err != nil {
		return fmt.Errorf("pipeline: replace glossary for %s (%d entries): %w", r.Book.BookID, len(entries), err)
	}
	rows, err := r.Store.GlossaryForBook(r.Book.BookID)
	if err != nil {
		return fmt.Errorf("pipeline: read glossary for %s: %w", r.Book.BookID, err)
	}
	r.memory = materializeMemory(rows, r.Pipeline.Gates.Glossary.PostcheckGate)
	// The DRAFT wave selects over a BASE-scoped bank (Source:mined excluded) so its injection is
	// byte-identical across a bank-mining enrichment — matching the draft-wave snapshot (baseMemoryVersion),
	// which keeps «one re-payment» honest at the WIRE level, not only the version-hash level. Only the
	// editor sees mined terms (the enriched `memory`). When there are no mined rows (every $0 test / the
	// golden) the base bank IS the enriched one — share the object, no double materialization, no drift.
	baseRows := rows[:0:0]
	hasMined := false
	for _, row := range rows {
		if row.Source == "mined" {
			hasMined = true
			continue
		}
		baseRows = append(baseRows, row)
	}
	if hasMined {
		r.baseMemory = materializeMemory(baseRows, r.Pipeline.Gates.Glossary.PostcheckGate)
	} else {
		r.baseMemory = r.memory
	}
	if cols := injectivityCollisions(rows); len(cols) > 0 {
		r.Log.WarnContext(ctx, "glossary approved dst-collisions (B2: two source terms share one Russian surface — the reader cannot tell them apart)",
			"collisions", strings.Join(cols, "; "))
	}
	r.Log.InfoContext(ctx, "glossary materialized", "book", r.Book.BookID,
		"entries", len(rows), "memory_version", r.memory.Version()[:12])
	return nil
}

// minedDeltaSeedCollisions returns a human message for every mined-delta entry whose store UNIQUE key
// (src, sense, since_ch, until_ch) already exists among the seed/ruby entries — the D39.20 deviation-#1
// crash class. It keys on the FULL uniqueness tuple (not the firing surface, unlike
// approvedSharedKeyCollisions, and regardless of dst — a SAME-dst duplicate crashes the INSERT just the
// same, yet approvedSharedKeyCollisions deliberately skips it). Pure and deterministic (seed order); an
// empty result means the delta is disjoint from the seed on the uniqueness axis and can be appended safely.
func minedDeltaSeedCollisions(seedEntries, mined []store.GlossaryEntry) []string {
	type key struct {
		src, sense   string
		since, until int
	}
	seen := map[key]store.GlossaryEntry{}
	for _, e := range seedEntries {
		seen[key{e.Src, e.Sense, e.SinceCh, e.UntilCh}] = e
	}
	var out []string
	for _, m := range mined {
		if prior, ok := seen[key{m.Src, m.Sense, m.SinceCh, m.UntilCh}]; ok {
			out = append(out, fmt.Sprintf("src %q (sense %q, window [%d,%d]) is in both the seed (%q→%q) and the mined-delta (%q→%q)",
				m.Src, m.Sense, m.SinceCh, m.UntilCh, prior.Src, prior.Dst, m.Src, m.Dst))
		}
	}
	return out
}

// persistRuby aggregates the ingested ruby occurrences into one row per
// (base, reading) — first_chapter = MIN, occurrences = full-book count — and
// REPLACES the book's whole ruby set (store.ReplaceRubyReadings). Idempotent: the
// aggregation is recomputed identically on every ingest, so a resume re-writes the
// same rows; a source edit converges every column — a pair removed by the edit
// disappears instead of lingering as a phantom (external-review #6). Called
// unconditionally (even for zero readings — a txt book or a source that dropped its
// furigana) so the full-replace clears any stale rows. The write order is sorted for
// deterministic, test-stable behavior. Memory v2 (step 4) consumes ruby_readings into
// a glossary name-lock (D9); nothing here injects into a prompt (§7d).
func (r *Runner) persistRuby(readings []RubyReading) error {
	type agg struct {
		first int
		count int
	}
	seen := map[[2]string]*agg{}
	order := make([][2]string, 0, len(readings))
	for _, rr := range readings {
		key := [2]string{rr.Base, rr.Reading}
		a, ok := seen[key]
		if !ok {
			a = &agg{first: rr.Chapter}
			seen[key] = a
			order = append(order, key)
		}
		if rr.Chapter < a.first {
			a.first = rr.Chapter
		}
		a.count++
	}
	sort.Slice(order, func(i, j int) bool {
		if order[i][0] != order[j][0] {
			return order[i][0] < order[j][0]
		}
		return order[i][1] < order[j][1]
	})
	rows := make([]store.RubyReading, 0, len(order))
	for _, key := range order {
		a := seen[key]
		rows = append(rows, store.RubyReading{
			BookID: r.Book.BookID, Base: key[0], Reading: key[1],
			FirstChapter: a.first, Occurrences: a.count,
		})
	}
	return r.Store.ReplaceRubyReadings(r.Book.BookID, rows)
}