222 lines
12 KiB
Go
222 lines
12 KiB
Go
package pipeline
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"textmachine/backend/internal/chunk"
|
||
"textmachine/backend/internal/llm"
|
||
)
|
||
|
||
// bookrun.go: the BOOK-level loop (Milestone 2) — ingest → memory seed/materialization →
|
||
// snapshot → sequential walk over chunks with D2 discipline (a flag does not drop the run)
|
||
// and a sticky scene window (A5). Plus the run-result types and the CLI exit contract.
|
||
|
||
// StageResult reports one executed stage of a chunk.
|
||
type StageResult struct {
|
||
Stage string
|
||
Role string
|
||
Model string // the model that actually answered
|
||
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 ATTEMPTED this stage (D12); when the
|
||
// fallback passed the re-gate, Model above is the fallback (it answered).
|
||
Escalated bool
|
||
// EscalationModel is the fallback model whose output became AUTHORITATIVE ("" when no hop ran, and
|
||
// "" when the hop ran but also failed — then the primary's flag stands and nothing of the hop's
|
||
// output is used; `Escalated` alone carries the fact of the attempt). See stagerun.go.
|
||
EscalationModel string
|
||
// BankFlags carries the translator draft's banknote telemetry (WS4 point 10): accepted line count +
|
||
// parse-fail / truncation flags. Zero-valued on every non-translator stage and every channel-off run.
|
||
BankFlags bankFlags
|
||
// BankProposals is the JSON of the same draft's PARSED banknote entries (the WHAT channel). It rides
|
||
// beside the counters because D39.36 showed counters alone are useless to the person who has to sign:
|
||
// they say a proposal existed, not what it was.
|
||
BankProposals string
|
||
// Repair is the addressable-defect repair outcome of this stage (pack-16): zero for every stage that is
|
||
// not the shipping one, and for every run with the gate off. Counters are DERIVED state — they are never
|
||
// stored in their own row, because the durable artifacts (the role='repair' checkpoints and the
|
||
// tm-repair-v1 final_hash namespace) already record what happened and survive a resume, whereas a column
|
||
// on retrieval_state would be zeroed by the draft wave's unconditional rewrite on every resumed run.
|
||
Repair repairResult
|
||
}
|
||
|
||
// 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
|
||
// DroppedMembers counts the unit's member chunks whose draft flagged and which the c-lite editor
|
||
// therefore left OUT of the edit (waverun.go runEditUnit). It is what makes "the shipped text is
|
||
// INCOMPLETE" a fact the renderers can state instead of infer: FlagReason cannot carry it, because
|
||
// a unit can be flagged for the EDIT's own reason (a cosmetic sanitizer strip) while a member was
|
||
// dropped as well, and then the reason names the strip and says nothing about the missing member.
|
||
// 0 for every unit that lost nothing.
|
||
DroppedMembers int
|
||
// DroppedReason is why the FIRST dropped member's draft flagged — the cause of the HOLE, which is
|
||
// not the same fact as FlagReason. When the edit ALSO flags on its own account (a cosmetic
|
||
// sanitizer strip), FlagReason is the strip's and says nothing about the missing text; printing it
|
||
// as the cause of the loss tells the reader a clean-up ate a chunk of the book, which is the very
|
||
// false claim this pack exists to remove. "" when nothing dropped.
|
||
DroppedReason FlagReason
|
||
}
|
||
|
||
// 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 allows N)
|
||
}
|
||
|
||
// ExitCode is the run's shell disposition: 0 clean, 2 completed-with-flags
|
||
// (acceptance allows N flags — acceptance permits this). 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.
|
||
//
|
||
// It is also the boundary of the run-event seam (row 103): the journal is opened here rather than with
|
||
// the runner, because a run's identity IS the call's trace id and that lives on the context — and
|
||
// because a command that never runs a book (a dry-run redrive, a read-only projection) must leave no
|
||
// trace in a stream that describes runs. Every exit of the run passes through terminal() below, so the
|
||
// stream's last line says WHY it ended and a reader never has to do exit-code archaeology.
|
||
func (r *Runner) TranslateBook(ctx context.Context) (*BookResult, error) {
|
||
r.openEvents(ctx)
|
||
res, err := r.translateBook(ctx)
|
||
r.events.terminal(res, err)
|
||
return res, err
|
||
}
|
||
|
||
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).
|
||
// Fingerprint the source BEFORE reading it (row 100): the manifest written below must be stamped with
|
||
// the identity of the bytes it was actually cut from, and the cut takes seconds on a large book.
|
||
// nil = it could not be taken, which costs this run its manifest and nothing else.
|
||
srcBefore := r.sourceFingerprintBeforeIngest(ctx)
|
||
doc, err := r.ingestSource()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// the precompute pass: eager-build every reachable client BEFORE any wave goroutine, so the clients map is
|
||
// read-only in the waves (r.client is lock-free, a miss is loud — closes the D12 lazy-init race).
|
||
if err := r.buildClients(); err != nil {
|
||
return nil, err
|
||
}
|
||
// the precompute pass: build the per-model rate-guards (transport axis, read-only in the waves; a no-op until a
|
||
// model configures rate_limit — WS1 §1б, the mistral-arm precondition WS6).
|
||
r.buildRateGuards()
|
||
|
||
// Persist captured ruby readings (idempotent; consumed by seedGlossary below into
|
||
// auto glossary candidates — step 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
|
||
}
|
||
// The bank read-out (backlog row 125), at the FIRST boundary where a bank exists. Doing it here rather
|
||
// than only at the stops means a book that never mines anything — or a run that dies before the first
|
||
// wave — still leaves a readable bank behind, which is the state a reader has to be able to see.
|
||
r.exportBank(ctx, "run-start/seeded")
|
||
|
||
chunks, chapterTexts := chunk.SplitChunksWithChapters(doc.Chapters, r.segBudget(), r.headingRule(), r.sentenceAbbrevs())
|
||
if len(chunks) == 0 {
|
||
return nil, sourceHasNoContent(fmt.Errorf("pipeline: source file %s produced no chunks after normalization", r.Book.SourceFile))
|
||
}
|
||
// The chapter/chunk manifest (backlog row 100): persisted HERE, where the split that every paid byte
|
||
// is addressed against was just computed, so the artifact and the run can never describe different
|
||
// books. Loud but not fatal — see persistManifest.
|
||
r.persistManifest(ctx, srcBefore, chapterTexts, chunks)
|
||
// the precompute pass: the banknote parser's source index (backlog 19). Built here, in the composite
|
||
// root, because the rule "a bank line must name something the book contains" needs the whole chunk
|
||
// manifest, and building it anywhere else would let a call path exist without it.
|
||
r.bankSrc = newBankSourceIndex(chunks)
|
||
|
||
// Consent to a RE-PAYMENT (D20.2-Q2, rebill.go) — BEFORE the waves, hence before the first Reserve:
|
||
// if this run would pay again for units already billed under a superseded snapshot, and the amount
|
||
// is over the book's threshold, it stops here with the sum instead of quietly re-buying the book.
|
||
// It needs the materialized memory (the per-wave snapshots fold it), so it sits after seedGlossary,
|
||
// and the chunk manifest (the projected book cost is per output unit), so it sits after the split.
|
||
if err := r.checkRebillConsent(ctx, chunks); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// The run scale — one line to stderr (the smoke-run pain: N/M and the progress
|
||
// denominator never appeared in the logs at all, only in stdout/status).
|
||
r.Log.InfoContext(ctx, "book run started", "book", r.Book.BookID,
|
||
"chapters", len(doc.Chapters), "chunks", len(chunks), "stages", len(r.Pipeline.Stages))
|
||
|
||
// the precompute pass: pre-compute the sticky-chain memory selection for EVERY chunk (WS1 §1б, precomputeSticky). The
|
||
// sticky window (A5 scene-inertia — the recent chunks' exact matches in the current chapter, reset at
|
||
// a chapter boundary) is a CROSS-CHUNK sequential dependency, so it is computed once here in the precompute pass and
|
||
// consumed by the parallel draft-wave workers (it cannot be recomputed inside a wave). Byte-identical to the
|
||
// retired inline computation (Select is $0 and pure — the golden test proves the injected bytes hold).
|
||
// The draft wave selects over the BASE bank (mined-excluded) so its injection is byte-identical across
|
||
// a bank-mining enrichment (the review-confirmed «re-paid ONCE» fix); the editor uses the enriched bank.
|
||
stickySel := precomputeSticky(chunks, r.baseMemory, r.Pipeline.Context.GlossaryTokenBudget)
|
||
|
||
// The wave executor (R1, waverun.go): the draft wave (draft ∥) → the bank-mining stop → the edit wave (edit ∥). It computes +
|
||
// upserts the per-wave snapshots (draft-wave snapshot base-bank / edit-wave snapshot enriched) itself and pins each
|
||
// wave's jobs to its own — a the bank-mining stop enrichment moves only edit-wave snapshot, keeping draft-wave checkpoints valid
|
||
// («re-paid ONCE»). Returns a *WaveSignatureStop when the bank-mining stop stops for owner sign.
|
||
res, err := r.translateBookWaves(ctx, chunks, stickySel)
|
||
if err == nil {
|
||
// End of the run (row 125). The stop paths refresh the read-out themselves, so this is the boundary
|
||
// they do not cover: a run that finished, whose last bank change was the auto-continue re-seed or a
|
||
// signed mined-delta loaded at start.
|
||
r.exportBank(ctx, "run-finished")
|
||
}
|
||
return res, err
|
||
}
|