textmachine/backend/internal/pipeline/waverun_test.go

package pipeline

import (
	"context"
	"errors"
	"math"
	"os"
	"path/filepath"
	"strings"
	"testing"

	"textmachine/backend/internal/obs"
)

// waverun_test.go: the R1 wave-driver behaviours the 1-chunk-per-chapter fixtures cannot exercise — a
// MULTI-CHUNK edit unit (the editor collapses several draft chunks into one edit call over their
// concatenation, addressed by the leader), parallel-worker money conservation, a flagged member draft
// skipping the unit's edit, and the bank-mining stop gate.

// cjkSource builds a single-chapter ja source of `paras` paragraphs, each `hanPerPara` Han chars + a full
// stop, separated by a blank line — enough to make the chunker split it into several draft chunks.
func cjkSource(paras, hanPerPara int) string {
	p := make([]string, paras)
	for i := range p {
		p[i] = strings.Repeat("文", hanPerPara) + "。"
	}
	return strings.Join(p, "\n\n")
}

// TestWaveMultiChunkEditUnit is the load-bearing R1 pin: a chapter that splits into TWO draft chunks but
// groups into ONE edit unit. The editor must run ONCE over the concatenation of the two members' drafts
// (never re-rendering either draft), the unit's outcome is a single ChunkOutcome at the leader, and export
// projects one row — not two "pending" chunks.
func TestWaveMultiChunkEditUnit(t *testing.T) {
	rec := &reqRec{}
	srv := newJSONProvider(rec, draftEdit)
	defer srv.Close()
	// para1 est_out ≈ 1677 (≤ draft budget 1797 → its own chunk); para2 est_out ≈ 1318 forces a second
	// chunk; together ≈ 2995 ≤ edit ceiling 3200 → ONE edit unit with both members.
	bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: cjkSource(1, 1400) + "\n\n" + strings.Repeat("字", 1100) + "。", regenerate: 0})
	ctx := obs.WithReqInfo(context.Background(), obs.ReqInfo{TraceID: obs.NewTraceID()})

	// Confirm the fixture actually produces the 2-chunk / 1-unit shape the test targets.
	r0 := newRunner(t, bookPath)
	manifest, err := r0.bookChunks()
	r0.Close()
	if err != nil {
		t.Fatal(err)
	}
	units := buildEditUnits(manifest)
	if len(manifest) != 2 || len(units) != 1 || len(units[0].Members) != 2 {
		t.Fatalf("fixture must be 2 draft chunks in 1 edit unit, got %d chunks / %d units", len(manifest), len(units))
	}

	r1 := newRunner(t, bookPath)
	res, err := r1.TranslateBook(ctx)
	if err != nil {
		t.Fatal(err)
	}
	r1.Close()

	// One outcome PER UNIT, at the leader chunk, carrying both member drafts + the single edit stage.
	if len(res.Chunks) != 1 {
		t.Fatalf("a 2-chunk unit must yield ONE unit outcome, got %d", len(res.Chunks))
	}
	oc := res.Chunks[0]
	if oc.ChunkIdx != 0 {
		t.Fatalf("the unit outcome must address the leader chunk (0), got %d", oc.ChunkIdx)
	}
	if len(oc.Stages) != 3 {
		t.Fatalf("the unit outcome must carry 2 member drafts + 1 edit, got %d stages", len(oc.Stages))
	}
	if oc.Stages[0].Role != roleTranslator || oc.Stages[1].Role != roleTranslator || oc.Stages[2].Role != roleEditor {
		t.Fatalf("stages must be [draft, draft, edit], got %s/%s/%s", oc.Stages[0].Role, oc.Stages[1].Role, oc.Stages[2].Role)
	}
	if oc.FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" {
		t.Fatalf("the unit's final text is the single edit output, got %q", oc.FinalText)
	}
	// EXACTLY 3 provider calls: 2 drafts + 1 edit — the editor did NOT re-render either draft.
	if rec.count() != 3 {
		t.Fatalf("multi-chunk unit must be 2 draft + 1 edit calls, got %d", rec.count())
	}
	// The editor's request carried the CONCATENATION of both members' drafts (joined by the unit separator).
	var editBody string
	for _, b := range rec.all() {
		if isEditBody(b) {
			editBody = b
		}
	}
	// Both members' drafts appear in the editor's input (the body JSON-escapes the join, so count the
	// draft marker rather than match the raw separator).
	if n := strings.Count(editBody, "ЧЕРНОВИК ПЕРЕВОДА"); n != 2 {
		t.Fatalf("the editor must read BOTH member drafts concatenated, found %d in the edit body", n)
	}

	// Export projects ONE unit row (not 2, and not a phantom "pending" for the non-leader member).
	r2 := newRunner(t, bookPath)
	exp, err := r2.Export(true)
	if err != nil {
		t.Fatal(err)
	}
	if exp.TotalChunks != 1 || exp.PendingChunks != 0 || len(exp.Chunks) != 1 {
		t.Fatalf("export must be 1 unit / 0 pending, got total=%d pending=%d rows=%d", exp.TotalChunks, exp.PendingChunks, len(exp.Chunks))
	}
	if exp.Chunks[0].FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" {
		t.Fatalf("export unit text mismatch: %q", exp.Chunks[0].FinalText)
	}
	// --pairs source is the members' joined source (the src↔target pair aligns to the editor's input).
	if !strings.Contains(exp.Chunks[0].Source, unitJoinSeparator) {
		t.Fatalf("export --pairs source must be the joined unit source, got %q", exp.Chunks[0].Source[:min(40, len(exp.Chunks[0].Source))])
	}

	// Status counts the unit as ONE done chunk (not two).
	st, err := r2.Status(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if st.TotalChunks != 1 || st.Done != 1 {
		t.Fatalf("status must be 1 unit done, got total=%d done=%d", st.TotalChunks, st.Done)
	}
	r2.Close()

	// Resume is $0 (all checkpoints hit; the editor re-derives its content-addressed id for free).
	r3 := newRunner(t, bookPath)
	defer r3.Close()
	before := rec.count()
	res2, err := r3.TranslateBook(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if rec.count() != before || res2.TotalUSD != 0 {
		t.Fatalf("resume must be $0 with no new calls: calls=%d usd=%v", rec.count()-before, res2.TotalUSD)
	}
}

// TestRunWaveSurfacesParentCancellation pins the cancellation contract: when the PARENT context is
// cancelled (a Ctrl-C / SIGTERM during a live run — main.go wires a signal.NotifyContext), runWave must
// surface that as an ERROR, never a silent nil. The feeder drops the remaining items on ctx.Done(), and a
// $0 resume worker never touches the cancellable ctx (its store ops use their own opContext), so firstErr
// stays nil — without a parent.Err() check runWave returns nil with items UNDONE, and the driver then
// indexes uninitialised result slots (an edit pipeline nil-derefs *oc; a draft-only run reports empty
// translations as exit-0 success). A cancelled parent → a non-nil error is the fix's whole point.
func TestRunWaveSurfacesParentCancellation(t *testing.T) {
	r := &Runner{} // runWave touches no Runner field
	ctx, cancel := context.WithCancel(context.Background())
	cancel() // the operator hit Ctrl-C before/at the wave boundary
	// Workers return nil (a $0 resume worker never observes the cancellable ctx) — so firstErr stays nil
	// and ONLY a parent-cancellation check can turn this into the error the driver needs.
	err := r.runWave(ctx, 2, 8, func(context.Context, int) error { return nil })
	if err == nil {
		t.Fatal("runWave returned nil on a cancelled parent ctx — items are left undone and the driver indexes uninitialised result slots (nil-deref panic on an edit pipeline / silent exit-0 partial success on draft-only)")
	}
}

// TestWaveParallelWorkersMoneyConserved runs the wave driver with several parallel workers over a book of
// many chunks and asserts the money invariant survives concurrency: committed spend == the sum of the
// settled checkpoints (the single-writer store serializes Reserve/Settle), and the per-unit result is
// complete + deterministic. Run under `-race` this also proves the waves share only read-only state.
func TestWaveParallelWorkersMoneyConserved(t *testing.T) {
	rec := &reqRec{}
	srv := newJSONProvider(rec, draftEdit)
	defer srv.Close()
	// 5 paragraphs of ~1400 Han each → 5 draft chunks, and since any two exceed the 3200 edit ceiling,
	// 5 edit units → 10 wave work items fanned over 4 workers.
	bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: cjkSource(5, 1400), regenerate: 0, waveWorkers: 4, bookUSD: 100})
	ctx := obs.WithReqInfo(context.Background(), obs.ReqInfo{TraceID: obs.NewTraceID()})

	r1 := newRunner(t, bookPath)
	res, err := r1.TranslateBook(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(res.Chunks) != 5 {
		t.Fatalf("5 chapters-worth of single-chunk units expected, got %d", len(res.Chunks))
	}
	for _, oc := range res.Chunks {
		if oc.Disposition != DispOK || oc.FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" {
			t.Fatalf("every parallel unit must be ok+edited, got %s / %q", oc.Disposition, oc.FinalText)
		}
	}
	// Money invariant under parallelism: the ledger's committed spend matches the run's own aggregate and
	// no reservation leaked — the single-writer store serialized every Reserve/Settle despite N workers
	// (the store-level atomicity + kill-9 durability is separately pinned by store/kill9_test.go).
	committed, reserved, err := r1.Store.SpentUSD("test-book")
	if err != nil {
		t.Fatal(err)
	}
	r1.Close()
	if reserved != 0 {
		t.Fatalf("no reservation may leak after a clean parallel run, reserved=%v", reserved)
	}
	if math.Abs(res.TotalUSD-committed) > 1e-9 {
		t.Fatalf("run total %v != ledger committed %v under parallel workers — a settle raced/leaked", res.TotalUSD, committed)
	}
}

// TestWaveEditUnitFlaggedMemberDraft: when ONE member draft of a multi-chunk unit flags, the whole unit is
// flagged and its edit is SKIPPED (no paid edit over a partial draft), while the good member's draft money
// stays committed. There is exactly ONE edit call fewer than a clean run.
func TestWaveEditUnitFlaggedMemberDraft(t *testing.T) {
	rec := &reqRec{}
	// The second paragraph carries a marker; the mock returns an untranslated CJK echo for it → cjk_artifact.
	const echoMarker = "禁"
	respond := func(body string) (string, string) {
		if isEditBody(body) {
			return "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД", "stop"
		}
		if strings.Contains(body, echoMarker) {
			return "这是完全没有翻译的中文内容。", "stop" // fully-CJK → classify flags cjk_artifact
		}
		return "ЧЕРНОВИК ПЕРЕВОДА", "stop"
	}
	srv := newJSONProvider(rec, respond)
	defer srv.Close()
	src := strings.Repeat("文", 1400) + "。\n\n" + strings.Repeat(echoMarker, 1100) + "。"
	bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: src, regenerate: 0})
	ctx := obs.WithReqInfo(context.Background(), obs.ReqInfo{TraceID: obs.NewTraceID()})

	r := newRunner(t, bookPath)
	defer r.Close()
	res, err := r.TranslateBook(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(res.Chunks) != 1 {
		t.Fatalf("still one unit outcome, got %d", len(res.Chunks))
	}
	oc := res.Chunks[0]
	if oc.Disposition != DispFlagged || oc.FlagReason != FlagCJKArtifact {
		t.Fatalf("a flagged member draft must flag the unit (cjk_artifact), got %s/%s", oc.Disposition, oc.FlagReason)
	}
	if oc.FinalText != "" {
		t.Fatalf("a flagged unit exports nothing, got %q", oc.FinalText)
	}
	// 2 draft calls (one ok, one echo), ZERO edit calls — the unit's edit was skipped over the partial draft.
	if rec.count() != 2 {
		t.Fatalf("a partial-draft unit runs 2 drafts + 0 edit, got %d calls", rec.count())
	}
}

// TestWaveEscalationBudgetSerializedUnderParallelism pins the escMu: two draft chunks escalate
// CONCURRENTLY under a budget that admits only ONE hop. The escalation soft-cap is a non-atomic
// read-then-act over EscalationSpentUSD, so without escMu both parallel workers could read spent<budget and
// both be admitted (overshoot by a whole hop). escMu serializes the admission through the paid settle, so
// exactly one hop fires — the same guarantee the sequential TestRunnerEscalationBudgetExhaustionDeniesLaterHop
// gives, now under parallel workers. (Run under -race this also proves the shared spend path is race-clean.)
func TestWaveEscalationBudgetSerializedUnderParallelism(t *testing.T) {
	rec := &reqRec{}
	srv := newJSONProvider(rec, echoOrClean) // primary echoes (cjk_artifact); fake-fallback cleans
	defer srv.Close()
	bookPath := setupTwoChapterEscalation(t, srv.URL, 0.001) // budget 0.001 < one hop (≈0.00182)
	// Bump to parallel workers so both chapters' drafts escalate at once.
	pipePath := filepath.Join(filepath.Dir(bookPath), "pipeline.yaml")
	raw, err := os.ReadFile(pipePath)
	if err != nil {
		t.Fatal(err)
	}
	writeFile(t, pipePath, strings.Replace(string(raw), "core: C1", "core: C1\nwaves: { workers: 4 }", 1))

	r := newRunner(t, bookPath)
	defer r.Close()
	res, err := r.TranslateBook(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	esc, err := r.Store.EscalationSpentUSD("test-book")
	if err != nil {
		t.Fatal(err)
	}
	if diff := esc - fakeCallUSD; diff > 1e-12 || diff < -1e-12 {
		t.Fatalf("escMu must serialize escalation admission under parallel workers: want exactly one hop (%.6f), got %.6f", fakeCallUSD, esc)
	}
	// Exactly one unit escalated-OK and one stayed flagged (WHICH one wins the single hop is
	// scheduling-dependent, so assert the totals, not the identity).
	escalatedOK, flagged := 0, 0
	for _, oc := range res.Chunks {
		if oc.Disposition == DispOK && len(oc.Stages) > 0 && oc.Stages[0].Escalated {
			escalatedOK++
		}
		if oc.Disposition == DispFlagged {
			flagged++
		}
	}
	if escalatedOK != 1 || flagged != 1 {
		t.Fatalf("exactly one unit escalates-OK and one stays flagged, got ok=%d flagged=%d", escalatedOK, flagged)
	}
}

// TestWaveMinedSignDoesNotRebillDraft is the «переоплата ОДНА» pin AT THE INJECTION LEVEL (the checkpoint-
// review found the split was implemented only at the version-hash level): signing a mined term and re-running
// must move ONLY the edit wave, leaving the draft wave's checkpoints byte-stable. The draft selects over the
// BASE bank (mined-excluded), so a mined addition does NOT change the draft's injected wire → its content_hash
// / final_hash / snapshot are unchanged → the draft resumes at $0. The editor selects over the ENRICHED bank,
// so the mined term DOES move the edit-wave snapshot (the single, --resnapshot-gated overpay). Reverting the
// fix (draft over the enriched bank) makes the draft content_hash change here → the test fails.
func TestWaveMinedSignDoesNotRebillDraft(t *testing.T) {
	rec := &reqRec{}
	srv := newJSONProvider(rec, draftEdit)
	defer srv.Close()
	// Source carries a BASE seed term (魔法学院) and a to-be-mined term (方源); both are ≥2-char Han keys.
	src := "方源走进了魔法学院的图书馆。"
	seed := "terms:\n  - src: 魔法学院\n    dst: Академия магии\n    status: approved\n"
	bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: src, glossarySeed: seed, regenerate: 0})
	ctx := obs.WithReqInfo(context.Background(), obs.ReqInfo{TraceID: obs.NewTraceID()})

	r1 := newRunner(t, bookPath)
	if _, err := r1.TranslateBook(ctx); err != nil {
		t.Fatal(err)
	}
	draftBefore, err := r1.Store.GetChunkStatus("test-book", 1, 0, "draft")
	if err != nil || draftBefore == nil {
		t.Fatalf("draft chunk_status after run 1: %v / %v", draftBefore, err)
	}
	editBefore, err := r1.Store.GetChunkStatus("test-book", 1, 0, "edit")
	if err != nil || editBefore == nil {
		t.Fatalf("edit chunk_status after run 1: %v / %v", editBefore, err)
	}
	r1.Close()

	// Owner signs 方源 → Фан Юань (approved) into a mined-delta file (loaded as Source:mined) and re-runs
	// with --resnapshot (the edit wave's enriched snapshot legitimately moves; the draft's must not).
	dir := filepath.Dir(bookPath)
	writeFile(t, filepath.Join(dir, "mined-delta.yaml"), "terms:\n  - src: 方源\n    dst: Фан Юань\n    status: approved\n")
	rawBook, err := os.ReadFile(bookPath)
	if err != nil {
		t.Fatal(err)
	}
	writeFile(t, bookPath, strings.Replace(string(rawBook), "pipeline: pipeline.yaml", "pipeline: pipeline.yaml\nmined_delta: mined-delta.yaml", 1))

	r2 := newRunner(t, bookPath)
	defer r2.Close()
	r2.Resnapshot = true
	if _, err := r2.TranslateBook(ctx); err != nil {
		t.Fatal(err)
	}
	draftAfter, err := r2.Store.GetChunkStatus("test-book", 1, 0, "draft")
	if err != nil || draftAfter == nil {
		t.Fatalf("draft chunk_status after run 2: %v / %v", draftAfter, err)
	}
	editAfter, err := r2.Store.GetChunkStatus("test-book", 1, 0, "edit")
	if err != nil || editAfter == nil {
		t.Fatalf("edit chunk_status after run 2: %v / %v", editAfter, err)
	}

	// «переоплата ОДНА»: the DRAFT wave is byte-stable across the mined sign — same snapshot, same rendered
	// content, same authoritative checkpoint → it resumed at $0 (a mined term never entered the draft wire).
	if draftAfter.SnapshotID != draftBefore.SnapshotID {
		t.Fatalf("draft-wave snapshot moved on a mined sign (%.12s → %.12s) — base bank not excluding mined", draftBefore.SnapshotID, draftAfter.SnapshotID)
	}
	if draftAfter.ContentHash != draftBefore.ContentHash {
		t.Fatalf("draft injection CHANGED on a mined sign — the draft selects over the ENRICHED bank (regression): «переоплата ОДНА» broken, the whole draft wave re-bills")
	}
	if draftAfter.FinalHash != draftBefore.FinalHash {
		t.Fatalf("draft checkpoint re-created on a mined sign (%.12s → %.12s) — the draft was re-billed", draftBefore.FinalHash, draftAfter.FinalHash)
	}
	// The single, gated overpay DID land on the edit wave: the enriched snapshot moved.
	if editAfter.SnapshotID == editBefore.SnapshotID {
		t.Fatalf("edit-wave snapshot did NOT move on a mined sign — the enriched bank is not folding the mined term")
	}
}

// TestWaveMiningUnconfiguredAutoContinues: with no langpack / no contrast (every existing fixture), the
// bank-mining stop is a no-op — the driver runs straight through to the edit wave and writes no signature
// map. Guards that the mining wiring never perturbs a normal $0 run.
func TestWaveMiningUnconfiguredAutoContinues(t *testing.T) {
	rec := &reqRec{}
	srv := newJSONProvider(rec, draftEdit)
	defer srv.Close()
	bookPath := setupProjectOpts(t, srv.URL, projectOpts{regenerate: 0})
	ctx := obs.WithReqInfo(context.Background(), obs.ReqInfo{TraceID: obs.NewTraceID()})

	r := newRunner(t, bookPath)
	defer r.Close()
	if r.pack != nil {
		t.Fatal("no langpack_root → pack must be nil")
	}
	stopped, err := r.runBankMiningStop(ctx, nil)
	if err != nil || stopped {
		t.Fatalf("unconfigured mining must auto-continue, got stopped=%v err=%v", stopped, err)
	}
	if _, err := r.TranslateBook(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := os.Stat(r.signatureMapPath()); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("no signature map may be written when mining is unconfigured (err=%v)", err)
	}
}