package pipeline import ( "context" "fmt" "textmachine/backend/internal/llm" ) // bookrun.go: цикл уровня КНИГИ (Веха 2) — ingest → сид/материализация памяти → // снапшот → последовательный обход чанков с D2-дисциплиной (флаг не роняет прогон) // и sticky-окном сцены (A5). Плюс типы результатов прогона и exit-контракт CLI. // StageResult reports one executed stage of a chunk. type StageResult struct { Stage string Role string Model string // фактически ответившая модель FromResume bool // no provider call was made THIS run (fully served from checkpoints) Usage llm.Usage CostUSD float64 // THIS run's spend (0 when fully served from checkpoints) CumCostUSD float64 // sum across ALL attempts of this chunk×stage (F3-honest, incl. retries) LatencyMS int FinishReason string Text string // the usable output (only on an ok disposition) // RecoveredText is the cosmetic-stripped export text (D35.4a): non-empty ONLY for a // FlagSanitizerStripped final stage, where the sanitizer removed a leading markdown header / // CJK-leak and committed the cleaned remainder. It carries the export forward on BOTH the // fresh path (runStage) and resume (resumeFromChunkStatus reads the derived checkpoint), so // translateChunk assembles the identical FinalText either way. "" for every other disposition. RecoveredText string Disposition Disposition FlagReason FlagReason // "" when ok Detail string Attempts int // Escalated — a single-hop fallback draft was tried this stage (D12); when the // fallback passed the re-gate, Model above is the fallback (it answered). Escalated bool EscalationModel string // the fallback model when Escalated ("" otherwise) } // ChunkOutcome is one chunk's result across the stage list. type ChunkOutcome struct { Chapter int ChunkIdx int Stages []StageResult // FinalText is the exported text: the last stage's output on ok; the cosmetic-stripped // remainder on a FlagSanitizerStripped flag (D35.4a — the chunk is exported flagged, not lost // to an empty placeholder); "" on any other flag (the contaminated output never exports). FinalText string Disposition Disposition // ok | flagged (a chunk has no "skipped" — that is a per-later-stage state) FlagReason FlagReason // the flagging stage's reason ("" when ok) CostUSD float64 // THIS run's spend on this chunk } // BookResult aggregates a whole run over every chapter×chunk. type BookResult struct { BookID string Chunks []ChunkOutcome TotalUSD float64 // THIS run's spend across all chunks Flagged int // number of flagged chunks (acceptance допускает N) } // ExitCode is the run's shell disposition: 0 clean, 2 completed-with-flags // (acceptance допускает N флагов — приёмка это разрешает). An infra failure is // an error from TranslateBook, which the CLI maps to 1. func (b *BookResult) ExitCode() int { if b.Flagged > 0 { return 2 } return 0 } // CompletedWithFlags is the typed sentinel the CLI maps to exit code 2: the book // finished end-to-end but N chunks were flagged for a human. It is NOT an infra // failure (a plain error → exit 1); it is a "clean run, attention needed" signal // carried up through `func run() error` in the idiomatic Go way. type CompletedWithFlags struct { Flagged int Total int } func (e *CompletedWithFlags) Error() string { return fmt.Sprintf("completed with %d/%d chunk(s) flagged for review", e.Flagged, e.Total) } // TranslateBook runs the whole book: split the normalized source into // chapter/chunk units (chunker.go) and drive each chunk through the stage list. // A bad chunk is flagged and the loop CONTINUES (D2); only an infra failure // (ceiling, config change without --resnapshot, unbilled call failure, store // error) aborts with an error, from which resume continues. func (r *Runner) TranslateBook(ctx context.Context) (*BookResult, error) { // Ingest reads + normalizes the source (txt/epub) into ordered per-chapter text // and captures ruby readings (ingest.go), decoding the txt source per book.encoding // (auto/utf8/gb18030). Offline and deterministic ($0, no LLM). doc, err := IngestEncoded(r.Book.SourceFile, r.Book.Encoding, r.Book.SourceLang) if err != nil { return nil, err } // Persist captured ruby readings (idempotent; consumed by seedGlossary below into // auto glossary candidates — шаг 4). Never injected into a prompt here (§7d); a // resume re-persists the same rows at $0. if err := r.persistRuby(doc.Ruby); err != nil { return nil, fmt.Errorf("pipeline: persist ruby readings: %w", err) } // Seed + materialize the glossary BEFORE the snapshot: the frozen approved rows feed // memoryVersion() → the snapshot (F1), so the snapshot must be computed with the bank // already materialized. Deterministic and $0 (seed file + classified ruby, no LLM). if err := r.seedGlossary(ctx); err != nil { return nil, err } // Snapshot is book-level (brief + stage plan + memory), computed once and // upserted before the loop; every job pins to it. memoryVersion() now reflects the // materialized bank, so a changed approved glossary re-pins loudly (F1). snapID, snapPayload, err := r.snapshotID() if err != nil { return nil, err } if err := r.Store.UpsertSnapshot(snapID, r.Book.BriefHash(), snapPayload); err != nil { return nil, fmt.Errorf("pipeline: upsert snapshot %.12s: %w", snapID, err) } chunks := SplitChunks(doc.Chapters) if len(chunks) == 0 { return nil, fmt.Errorf("pipeline: source file %s produced no chunks after normalization", r.Book.SourceFile) } // Масштаб прогона — одной строкой на stderr (боль smoke-прогона: N/M и знаменатель // прогресса не появлялись в логах вообще, только в stdout/status). r.Log.InfoContext(ctx, "book run started", "book", r.Book.BookID, "snapshot", snapID[:12], "chapters", len(doc.Chapters), "chunks", len(chunks), "stages_per_chunk", len(r.Pipeline.Stages)) res := &BookResult{BookID: r.Book.BookID} // Sticky scene-inertia state (A5): the exact-matched ids of the recent chunks in // the CURRENT chapter, reset at a chapter boundary (a new chapter is a scene change). // Rebuilt deterministically each run because translateChunk recomputes the ($0) // selection for EVERY chunk, resumed ones included — so sticky is resume-stable. var stickyWin []map[string]injectionDisposition prevChapter := 0 for i, ch := range chunks { if ch.Chapter != prevChapter { stickyWin = nil prevChapter = ch.Chapter // Пульс прогресса на границе главы: done/total + деньги ЭТОГО прогона — // оператор длинной книги видит движение, не открывая status. r.Log.InfoContext(ctx, "chapter started", "chapter", ch.Chapter, "chunks_done", i, "chunks_total", len(chunks), "run_usd", fmt.Sprintf("%.6f", res.TotalUSD)) } outcome, activeIDs, err := r.translateChunk(ctx, snapID, ch, unionSticky(stickyWin)) if err != nil { // Infra failure: abort. Chunks already committed are checkpointed; // resume continues from here at $0 for the done work. return res, err } res.Chunks = append(res.Chunks, *outcome) res.TotalUSD += outcome.CostUSD if outcome.Disposition == DispFlagged { res.Flagged++ } // Advance the sticky window (keep the last stickyDepth chunks' exact matches). stickyWin = append(stickyWin, activeIDs) if len(stickyWin) > stickyDepth { stickyWin = stickyWin[len(stickyWin)-stickyDepth:] } } r.Log.InfoContext(ctx, "book run finished", "book", r.Book.BookID, "chunks", len(res.Chunks), "flagged", res.Flagged, "run_usd", fmt.Sprintf("%.6f", res.TotalUSD)) return res, nil } // unionSticky merges the recent chunks' exact-matched ids into one sticky_prev, carrying // each id's DISPOSITION (D16.2). When an id fired in several recent chunks, the MOST RECENT // firing wins (win is ordered oldest→newest), so a re-established CONFIRMED match overrides // an older collision-prone AMBIGUOUS one — and vice-versa, never silently upgrading. func unionSticky(win []map[string]injectionDisposition) map[string]injectionDisposition { if len(win) == 0 { return nil } out := map[string]injectionDisposition{} for _, s := range win { for id, disp := range s { out[id] = disp } } return out }