package pipeline import ( "errors" "fmt" "log/slog" "os" "path/filepath" "sort" "strings" "sync" "unicode" "textmachine/backend/internal/checks" "textmachine/backend/internal/chunk" "textmachine/backend/internal/config" "textmachine/backend/internal/lang" "textmachine/backend/internal/ledger" "textmachine/backend/internal/llm" "textmachine/backend/internal/membank" "textmachine/backend/internal/store" ) // runner.go: runner setup — Runner and its wiring (config stack, store in the right // ownership mode, client/template caches). The book machinery itself is split across // related files (package №4): snapshot — snapshot.go, seed/ruby — seeding.go, the book // loop — bookrun.go, the chunk disposition loop — chunkrun.go, stage execution // with money (reserve → call → settle+checkpoint) — stagerun.go, single-hop // escalation with re-gating — escalation.go, $0-resume of a resolved one — resume.go. // roleTranslator is the stage role whose output is a direct source→target // translation — the only role the excision coverage-gate is meaningful on. const roleTranslator = "translator" // roleEditor is the BILINGUAL editor stage (D30.1 supersede of D1): editor.md feeds it BOTH the // source and the draft, plus the approved glossary's CONFIRMED dst forms as target-consistency // constraints. (The stale "monolingual, never sees the source" framing is D1, fully superseded.) const roleEditor = "editor" // Runner executes a book's pipeline. type Runner struct { Book *config.Book Models *config.Models Pipeline *config.Pipeline Store *store.Store Pricer *ledger.Pricer Log *slog.Logger // Resnapshot re-pins existing jobs to the CURRENT snapshot when configs/ // prompts changed since the job started (tmctl --resnapshot). Without the flag // a divergence is a loud error: a context change invalidates the checkpoints and // re-pays for the calls, so it is done only deliberately (Р6). Resnapshot bool // AcceptRebill is the operator's consent to the projected RE-PAYMENT of already-billed work // (tmctl --accept-rebill[=usd], D20.2-Q2). It is ORTHOGONAL to Resnapshot: that flag grants // permission to re-pin, this one consents to an AMOUNT (rebill.go). Zero value = no consent, which // only matters when the projection exceeds the book's threshold. AcceptRebill RebillConsent // VerifyBank is the operator's request to STOP at the bank-mining boundary and review the bank before // the edit wave (tmctl --verify-bank, D39.42 п.5). Default OFF: the run consolidates what it mined, // carries it forward marked-unverified, and finishes — «просто как намайнит и закончит». It is an // operator MODE, not a wire or snapshot axis: it changes when a human is asked, never what is sent. VerifyBank bool clients map[string]llm.LLMClient templates map[string]*PromptTemplate // repairTemplates holds one prompt per ENABLED repair class (pack-16), keyed by class. Empty when the // repair gate is off, which is the default; its SHAs are folded into the snapshot of the wave that owns // the shipping stage, because a repair prompt is not a stage and the per-stage fold cannot see it. repairTemplates map[checks.RepairClass]*PromptTemplate // terminologyTemplate is the pair's TERMINOLOGIST role prompt (pack-20), loaded only when the gate is on. // nil otherwise, which is what makes the whole role a no-op for every book that does not enable it. It is // NOT snapshot-folded: the role writes a signature map, not a checkpoint (config.TerminologyGate). terminologyTemplate *PromptTemplate // classifierTemplate is the pair's §2 type-classifier prompt, loaded only when classify_types is on. nil // otherwise → the classifier phase is a no-op and the draft heuristic type stands. classifierTemplate *PromptTemplate // targetScript is the target language's Unicode script, resolved once from gates.terminology // .target_script (config validates the name). nil when the book declares none — the answer-language // screen is then inert, which loadTargetScript says out loud if any channel could have used it. targetScript *unicode.RangeTable // rateGuards is the per-model wave-concurrency guard set (WS1 §1б), built once in the precompute pass and // read-only in the waves — a transport axis, never snapshot-folded. nil until buildRateGuards. rateGuards map[string]*rateGuard // bankSrc is the banknote parser's "is this surface in the book?" index (backlog 19), built once from // the chunk manifest in the precompute pass. nil outside a book run → the parser then judges a line by // the chunk text it was handed, and accepts nothing without one. bankSrc *bankSourceIndex // memory is the book's glossary FROZEN for this job (materialized once in // TranslateBook, after seeding, before snapshotID). Its Version() is the F1 // content-hash folded into the snapshot; its Select drives per-chunk injection. // nil until materialized (report path / no glossary) → memoryVersion() falls back // to the empty-materialization hash, a stable constant. memory *membank.Bank // baseMemory is the DRAFT-wave glossary bank: the BASE rows only (Source∈{seed,ruby,auto}, EXCLUDING // Source:mined). The draft wave's injection MUST be selected over this — not the enriched `memory` — // so it is BYTE-IDENTICAL across a bank-mining enrichment, matching the draft-wave snapshot which folds // baseMemoryVersion (mined-excluded). Otherwise signing a mined term would change the draft wire (the // injection is a message folded into request_hash) and silently re-bill the draft wave on the owner // re-run — even though the base snapshot is unchanged (the review-confirmed "ONE re-payment" hole). // When the book has NO mined rows (every $0 test / the golden), it is the SAME object as `memory` // (identical content) — no double materialization, the injection is unchanged. The editor keeps the // enriched `memory`. nil ⇔ memory is nil (materialized together in seedGlossary). baseMemory *membank.Bank // checkers is the compiled WS5/pack-13 observability checker spec (pair-14 data-out): the pair's // DETECTION patterns + tables (from pack) + the target-general lists (from embedded target data), // compiled once in loadLangPack. nil-safe: a no-pack / no-target-data book runs the checkers inert. checkers *checks.Checkers // pack is the book's language-data pack (internal/lang), loaded once in openRunner from // book.LangpackRoot (D39.15/16). The the bank-mining stop bank-miner reads its tables; pack.Version() is folded into // the snapshot (a pack edit is a loud --resnapshot). nil when the book declares no langpack_root, or // its pair has no catalog dir — then the miner is inert (the bank-mining stop auto-continues) and the fold is omitted. pack *lang.Pack // escMu serializes the single-hop escalation budget admission across the PARALLEL draft workers // (R1). escalationBudgetRemains is a non-atomic read-then-act over EscalationSpentUSD, so N concurrent // draft chunks could each read spent 0 { logger.Warn("a stage sends the SOURCE text to a model whose wire SUPPRESSES thinking — the ratified echo zone (D19.1 point 2: reasoning-off + dense CJK is a property of the model class, not a vendor quirk). Not a stop: the downstream echo gate (cjk_artifact) still catches an echo on every configuration (D19.2), but it costs a paid call per hit", "stages", strings.Join(exposed, ", "), "source_lang", r.Book.SourceLang) } return r, nil } // loadLangPack resolves the book's language-data pack (D39.15/16), loaded on BOTH the write path and the // read-only path so status/export/report reproduce the identical snapshot (pack.Version() is folded). The // contract mirrors the owner directive R1: a pair WITH a catalog directory (configs/langpacks/-/) // loads FAIL-LOUD (a missing/corrupt file stops the run, never a silently-empty miner); a pair WITHOUT a // catalog — or a book with no langpack_root — runs with a nil pack (the miner is inert, the bank-mining stop auto-continues, // and the snapshot fold is omitted). Presence of the pair directory is the "catalog exists" signal. func (r *Runner) loadLangPack() error { // The observability checkers compile from the pair pack (DETECTION patterns + tables, may be nil) AND the // embedded target data (target-general lists, keyed by target lang). Built here so it exists even for a // no-langpack book (a nil pack → inert pair checkers; the target lists still load). r.pack is set below. defer func() { r.checkers = checks.CompileCheckersFor(r.pack, lang.TargetChecksFor(r.Book.TargetLang)) // The SOURCE writing script(s) are a runner fact (the book's source lang), not target/pair data — set // after compile so the echo detector and the repair script-guard share one declared-source notion. r.checkers.SetSourceScripts(lang.LangScripts(r.Book.SourceLang)) }() if r.Book.LangpackRoot == "" { return nil // no langpack declared → nil pack, miner inert } pairDir := filepath.Join(r.Book.LangpackRoot, r.Book.LangPair()) if fi, err := os.Stat(pairDir); err != nil || !fi.IsDir() { // No catalog for this pair → nil-and-run (a ja book against a zh-only root just runs without mining). return nil } // A book-scoped overlay (LangpackExtend) unions the book's PRIVATE canon onto the shared pair pack // (pair-14 §1: 古月 is a 蛊真人 clan, not a shared 百家姓 surname). "" ⇒ plain Load. Folds into Version(). pack, err := lang.LoadWithOverlay(r.Book.LangpackRoot, r.Book.SourceLang, r.Book.TargetLang, r.Book.LangpackExtend) if err != nil { return fmt.Errorf("pipeline: load langpack for %s: %w", r.Book.LangPair(), err) } r.pack = pack // A pack that loaded but declares no name-morphology channel (its manifest opts out) runs the miner on an // INERT channel — empty tables, zero candidates. Say so out loud rather than mine silently (P4 contract). if !pack.HasMorphologyChannel() { r.Log.Warn("langpack declares no source-morphology channel — the name-miner is inert (no candidates); this is expected for a source without the pinyin name-miner schema", "pair", r.Book.LangPair()) } return nil } // packVersion is the snapshot-folded language-pack version: pack.Version() when a pack is loaded, "" when // not (omitted from the snapshot so a no-pack book is byte-stable and never re-billed for a feature it does // not use). A pack DATA edit changes Version() → the snapshot moves → a loud --resnapshot (R1, drift-proof). func (r *Runner) packVersion() string { if r.pack != nil { return r.pack.Version() } return "" } func (r *Runner) Close() error { return r.Store.Close() } func (r *Runner) loadTemplates() error { for _, st := range r.Pipeline.Stages { // The path was resolved at load — by the pair/role convention, or by the stage's deliberate // prompt_override — and a pair with no prompt pack already failed there, naming the file it // looked for (D39.23). Here it is only read. tpl, err := LoadPromptTemplate(st.PromptPath) if err != nil { return err } // Resolve the effective system once per stage: fold the few-shot block into // System unless the stage switched it off (D38.4). FewShot is retained so the // snapshot can fold the on/off state only for stages that actually have a block. tpl.System = tpl.SystemFor(fewShotEnabled(st)) r.templates[st.Name] = tpl } return nil } // headingRule is the book's deterministic chapter-heading rule (pack-13 title policy), or nil when the // book declares no langpack, its pair has no catalog, or the pair ships no heading.txt. The chunker uses // it to detect+strip a source header and render «Глава N»; nil ⇒ the feature is inert (byte-identical // chunks). Its bytes ride pack.Version() (folded into the snapshot via LangpackVersion), so a rule edit is // a loud --resnapshot and a book without a rule is never re-billed for a feature it does not use. func (r *Runner) headingRule() *lang.HeadingRule { if r.pack != nil { return r.pack.Heading } return nil } // sentenceAbbrevs is the BOOK's source-language abbreviation set for the sentence splitter (data-driven — // no hard-coded "en"; a CJK source ships none, so the set is empty and never consulted). func (r *Runner) sentenceAbbrevs() map[string]bool { return lang.SentenceAbbrev(r.Book.SourceLang) } // estOutTokens is the DISPLAY-ONLY fertility estimate of a text's ru output tokens (est_out = fertility // over the source char-classes, research/20 — NOT the char/4 EstimateTokens whose CJK undercount is the // «CJK-mine»). Used ONLY to annotate a settle-estimate telemetry row (pack-13 point-9): it feeds NOTHING // on the money path (not the reservation, not CostUSD), so it can never shift a verdict or a wire byte. func (r *Runner) estOutTokens(text string) int { cjk, other := chunk.TokenClassCounts(text) return int(r.segBudget().EstOut(cjk, other)) } // segBudget resolves the pipeline's config.Segmentation into the pipeline-package chunk.SegBudget the // chunker consumes (WS2). Kept in the runner (not chunker.go) so the segmenter stays config-free. func (r *Runner) segBudget() chunk.SegBudget { seg := r.Pipeline.Segmentation return chunk.SegBudget{ DraftBudgetOut: float64(seg.DraftBudgetOut), EditCeilingOut: float64(seg.EditCeilingOut), FertCJK: seg.Fertility.CJK, FertOther: seg.Fertility.Other, } } // reachableModels enumerates every model a run can call, LABEL ROUTING APPLIED: the resolved stage // models, their resolved single hops, and the repair model when that gate is on (config.ReachableModels // is the one definition — the capability invariant, the API-key preflight, the rate guards and this // eager client set all read it, so a model cannot be reachable for one and invisible to another). // The set is complete and static for a book, which is what lets the precompute pass pre-build every // client; a NEW call path must extend that definition, not this wrapper. Deterministic order. func (r *Runner) reachableModels() []string { return r.Pipeline.ReachableModels() } // sourcePlaceholder is the template marker that puts the SOURCE text on the wire (render.go's vals map). // A stage whose template carries it sends the raw source to its model — which is what makes the echo // class below reachable for that stage, editor included since the editor became bilingual (D30.1). const sourcePlaceholder = "{{text}}" // sourceEchoExposure lists `stage→model` pairs that ship the SOURCE text to a model whose RESOLVED wire // actively SUPPRESSES thinking — the other half of the reasoning-off hole (D39.26 добор B). The money // half is a hard load gate (config: an additive provider that thinks needs a reserved buffer); this half // cannot be a hard gate, because whether suppression actually costs quality is an open MEASUREMENT // (D19.1: reasoning-ON does not save archaic dense zh either) and the choice of editor is the owner's. // So it is made LOUD instead of silent, and the always-on downstream echo detector (classify → // cjk_artifact, mandatory on every configuration per D19.2) remains the enforcement. // // Generic by construction: no provider or label name appears here — it is "this stage sends the source" // × "this model's wire suppresses thinking" × "the source script is one where the class was measured". // It fires on no shipping path today (deepseek resolves to ReasoningNone, so "off" is a no-op and // thinking stays ON) and does fire for a reasoning-off editor arm on a CJK source, which is exactly the // exposure the stale "the editor only ever sees a Russian draft" comments used to deny. // It covers the ENGINE'S OWN calls too, not just the book's stages. Those calls carry a knob only since // D39.87, and the knob has an asymmetric empty value: on a ReasoningNone capability "" means the provider's // default (thinking ON), while on an extra_body_disable one (GLM) it means thinking DISABLED — the echo zone // — and gates.terminology.model is not restricted to a family. An unset knob pointed at such a model is // therefore a silent thinking-off with no config line describing it, which is precisely what this walk // exists to refuse to leave silent. func (r *Runner) sourceEchoExposure() []string { if !lang.IsCJKScriptLang(r.Book.SourceLang) { return nil } var out []string shipsSource := func(t *PromptTemplate) bool { return t != nil && strings.Contains(t.System+t.FewShot+t.User, sourcePlaceholder) } add := func(name, model, reasoning string) { if !r.Models.ThinksOnWire(model, reasoning) { out = append(out, name+"→"+model) } } for _, st := range r.Pipeline.Stages { if shipsSource(r.templates[st.Name]) { add(st.Name, st.ResolvedModel, st.Reasoning) } } // The bank roles ship source WITHOUT the placeholder: their candidate blocks carry KWIC contexts cut // from the book, which is the whole reason the roles can render a term at all. So membership is by // construction (the template is loaded ⇔ the role runs), not by scanning for a marker they do not use. bank := r.Pipeline.Gates.Terminology if r.terminologyTemplate != nil { add(roleTerminologist, bank.Model, bank.Reasoning) } if r.classifierTemplate != nil { add(roleClassifier, bank.ClassifierModel(), bank.Reasoning) } // Repair sends the source SPAN through the ordinary placeholder, per class — so it is scanned like a // stage, and a class whose prompt is monolingual is correctly not listed. for class, t := range r.repairTemplates { if shipsSource(t) { add(roleRepair+"/"+string(class), r.Pipeline.Gates.Repair.Model, r.Pipeline.Gates.Repair.Reasoning) } } sort.Strings(out) // map iteration above — the warning must not reorder between runs return out } // labelsFor returns the content labels of the text this run sends — BOOK-CONSTANT in v1 (D39.26 point 4). // The signature takes no chunk ON PURPOSE: per-chapter labels would need the per-stage model fold moved // off the per-wave snapshot (one snapshot is rendered per wave, before the chunk loop), so promising a // per-chunk shape here would promise what the mechanism does not hold. Granularity is an owner decision // (a labelled book routes ENTIRELY through label-capable models, which is what makes the ToS guarantee // structural rather than dependent on the accuracy of manual per-chapter reading). func (r *Runner) labelsFor() []string { return r.Book.ContentLabels } // buildClients EAGER-constructs every reachable LLM client BEFORE any wave goroutine starts (the precompute pass). // After this the clients map is READ-ONLY in the waves, so r.client is lock-free and a miss is a // loud error, not a lazy build under a data race — closing the runner.go lazy-init race (D12) and // tripwiring a future un-enumerated model axis. Idempotent; BuildClient needs only the provider // config (not the API key), so eager-build is safe even on a key-less resume. func (r *Runner) buildClients() error { for _, m := range r.reachableModels() { if _, ok := r.clients[m]; ok { continue } c, err := BuildClient(r.Models, m, r.Log) if err != nil { return fmt.Errorf("pipeline: eager-build client for model %q (the precompute pass): %w", m, err) } r.clients[m] = c } return nil } // clientFor returns the PRE-BUILT client for a model, and REFUSES when that model's provider may not // receive the labels of the content about to be sent. The label set is a PARAMETER rather than something // the function reads for itself, and this is the only way to obtain a client: a new call path physically // cannot reach a provider without stating what it is sending (the compile-enforced half of the routing // guarantee, D39.26 point 8). Load-time validation already refuses such a config, so a failure here means // a path bypassed it — an infra error, never a content verdict, so the book pauses durably (D4) instead // of flagging a chunk. // // Read-only (no lazy build, no lock): buildClients constructed every reachable client in the precompute // pass, so the map is only read in the waves; a miss means an un-enumerated model reached the wire and is // a loud error, never a silent lazy build under a race. func (r *Runner) clientFor(model string, labels []string) (llm.LLMClient, error) { if missing := r.Models.MissingLabels(model, labels); len(missing) > 0 { return nil, fmt.Errorf("pipeline: refusing to send content labelled %v to model %q (provider %s) — it does not accept %v. A provider without the label must not RECEIVE such content at all, whatever it would answer (D39.25); this is a routing bug: load-time validation should have caught it", labels, model, r.Models.ProviderOf(model), missing) } if c, ok := r.clients[model]; ok { return c, nil } return nil, fmt.Errorf("pipeline: no pre-built client for model %q — the precompute pass buildClients enumerates the resolved stage models ∪ their hops ∪ the repair model; a model outside that set reached the wire, add it to the eager set to keep the waves race-free", model) }