// Package pipeline is the C-core runner: deterministic prompt rendering, // request hashing, chunk checkpoints and the stage loop. Phase 0 is the // mini-runner C1 (draft→edit, one chunk); the per-chapter/per-chunk loops, // gates and escalation grow here in Phase 1 (Р2: this is runner code, not config). package pipeline import ( "crypto/sha256" "encoding/binary" "encoding/hex" "fmt" "os" "strconv" "strings" "unicode" "textmachine/backend/internal/config" "textmachine/backend/internal/llm" ) // render.go enforces the determinism invariant (03-implementation-notes §3.1): // the rendered prompt is a PURE function of (snapshot, committed prior // outputs, chunk). Timestamps/UUIDs are forbidden; no map iteration — only a // fixed list of placeholders. Verified by a "two renders are byte-for-byte // identical" test. Break determinism → break the request-hash → resume // re-translates and re-pays for the chapter, and the DeepSeek cache (byte prefix // match) misses. // chunkerVersion versions the segmentation/ingestion rules (including the source // normalization text.NormalizeSource, the txt/epub ingest layer and the target chunk // size targetChunkTokens); it is part of the snapshot, so re-chunking a book is an // explicit re-translation, not a silent cache miss (§3.4 — [DECISION NEEDED] item // 2в). Step 3a replaced the Milestone-2 char-budget paragraph packer with the real // source segmenter (sentence-aware, ~1–2k-token packing over ingest-split // chapters — chunker.go/ingest.go), so the version bumps: any earlier project DB // re-pins loudly via the --resnapshot gate (a re-chunk is a deliberate // re-translation), never a silent divergent re-pay. NOTE: the chunk target is a // const covered by THIS string; if it ever becomes config-tunable it must ALSO be // folded into the top-level snapshot (like coverageSnap), never left to the // version alone (§7d). Bumped v3→v4 closing external-review 3a: invalid-UTF-8 is // now normalized once in text.NormalizeSource (#3), the ASCII sentence boundary honors // quote-depth (#4), and
inside no longer leaks a newline into the body // (#5) — all segmentation/ingest behaviour changes → a loud --resnapshot. // Bumped v4→v5 (WS2): the packing budget moved from input-token heuristic (const 1500) to OUTPUT // (ru) tokens via fertility (est_out), draft chunks decoupled from coarse EDIT units (grouped whole // chunks to EditCeilingOut), oversized-sentence flag added. The budget itself is ALSO folded via // segmentationSnap (snapshot.go), so this version covers only the algorithm shape. // Ш-2 EXTENSION (see text/norm.go; beyond the owner-named normalizer/classifier/style set — justified: same // silent-drift class): chunk.TokenClassCounts (chunker.go) classifies source runes via unicode.Han/Hiragana/ // Katakana/Hangul to compute est_out, which SETS chunk boundaries → the wire. A toolchain Unicode bump that // reclassifies a rune would silently re-chunk the book; weaving unicode.Version makes it a loud --resnapshot. const chunkerVersion = "chunker-v5-output-budget-editunit+u" + unicode.Version // estimatorVersion versions EstimateTokens: its output feeds into max_tokens and // through it into the request-hash, so re-calibrating the weights is likewise an // explicit invalidation via the snapshot, not a silent miss of every checkpoint // (review finding: a code-only edit to the estimator would re-pay for half the book). // Ш-2 EXTENSION (see text/norm.go): EstimateTokens classifies runes via the same unicode ranges to size // max_tokens (∈ request_hash), so a toolchain Unicode reclassification shifts the wire — folded loudly. const estimatorVersion = "estimator-v0+u" + unicode.Version // maxTokensPolicyVersion versions the attempt→max_tokens scaling // (maxTokensForAttempt, disposition.go). attempt-0 budget is already covered by // estimatorVersion + defaults, but a change to the RETRY scaling would silently // shift the request_hash of every attempt≥1 (a missed checkpoint → re-pay on // retried chunks). Folding this version into the snapshot makes such a change a // loud --resnapshot instead — the same discipline as estimatorVersion, applied // to the regeneration axis (Milestone 2). const maxTokensPolicyVersion = "maxtok-v1-double-per-attempt" // userSeparator splits a prompt template file into the system part and the // user part. Both parts are versioned templates in prompts/ (the editorial team // later edits the files, not the code). const userSeparator = "\n---USER---\n" // fewShotSeparator optionally splits the SYSTEM part into a core prefix and a // trailing few-shot example block (D38.4). A stage may drop the block (few_shot:false) // for a model whose own CoT is disrupted by hand examples (deepseek-thinking, exp14 §2а). // Absent → the whole system part is core; the toggle is a no-op. const fewShotSeparator = "\n---FEWSHOT---\n" // PromptTemplate is one loaded stage template. System is the core system prefix; // FewShot is the optional example block (empty when the file has no ---FEWSHOT---). type PromptTemplate struct { System string FewShot string User string SHA256 string // hash of the raw file — part of the snapshot } // LoadPromptTemplate reads and splits a template file into core-system / few-shot / user. func LoadPromptTemplate(path string) (*PromptTemplate, error) { raw, err := os.ReadFile(path) if err != nil { return nil, fmt.Errorf("pipeline: read prompt %s: %w", path, err) } sum := sha256.Sum256(raw) parts := strings.SplitN(string(raw), userSeparator, 2) if len(parts) != 2 { return nil, fmt.Errorf("pipeline: prompt %s lacks the %q separator between system and user parts", path, strings.TrimSpace(userSeparator)) } t := &PromptTemplate{User: strings.TrimSpace(parts[1]), SHA256: hex.EncodeToString(sum[:])} // The few-shot block, if present, is the tail of the SYSTEM part (before ---USER---). head := strings.SplitN(parts[0], fewShotSeparator, 2) t.System = strings.TrimSpace(head[0]) if len(head) == 2 { t.FewShot = strings.TrimSpace(head[1]) } return t, nil } // SystemFor returns the effective system prompt for a stage: the core prefix, plus the // few-shot block when the stage keeps it on (the default). The whole raw file is still // snapshot-pinned via SHA256, and the few_shot on/off state is folded separately, so // dropping the block is a loud --resnapshot, not a silent divergence. func (t *PromptTemplate) SystemFor(fewShotOn bool) string { if fewShotOn && t.FewShot != "" { return t.System + "\n\n" + t.FewShot } return t.System } // fewShotEnabled resolves a stage's few_shot toggle: absent (nil) defaults to ON, so // every existing config keeps the few-shot examples without touching its yaml. func fewShotEnabled(st config.Stage) bool { return st.FewShot == nil || *st.FewShot } // RenderVars are the ONLY placeholders a template may use. A fixed struct, not // a map: map iteration order would randomize the render. type RenderVars struct { Book *config.Book Text string // the source chunk Draft string // prior stage output ("" on the first stage) } // Render substitutes placeholders in a SINGLE pass over the template: values // are inserted verbatim and never re-scanned, so a literal «{{…}}» inside the // source or the LLM draft is just text, not a marker (review finding: a sequential // ReplaceAll substituted the draft into a «{{draft}}» from the source and choked on // «{{TN: примечание}}» in the model output). Unknown {{…}} markers in the TEMPLATE // are a hard error — a silently empty placeholder is worse than a crash. func Render(tpl string, v RenderVars) (string, error) { // Lookup-only map (no iteration — determinism is not harmed). vals := map[string]string{ "book_id": v.Book.BookID, "title": v.Book.Title, "source_lang": v.Book.SourceLang, "target_lang": v.Book.TargetLang, "genre": v.Book.Genre, "audience": v.Book.Audience, "venuti": strconv.FormatFloat(v.Book.Venuti, 'f', 2, 64), "honorifics": v.Book.Honorifics, "transcription": v.Book.Transcription, "footnotes": v.Book.Footnotes, "text": v.Text, "draft": v.Draft, } var b strings.Builder rest := tpl for { i := strings.Index(rest, "{{") if i < 0 { b.WriteString(rest) return b.String(), nil } b.WriteString(rest[:i]) rest = rest[i:] end := strings.Index(rest, "}}") if end < 0 { return "", fmt.Errorf("pipeline: unterminated placeholder %q in template", snippetStr(rest)) } name := rest[2:end] val, ok := vals[name] if !ok { return "", fmt.Errorf("pipeline: unknown placeholder {{%s}} in template", name) } b.WriteString(val) rest = rest[end+2:] } } func snippetStr(s string) string { if len(s) > 40 { return s[:40] + "…" } return s } // Messages builds the wire-neutral message list for a stage with no memory injection // (the Milestone-0/2 behaviour). Layout per Р5: system (stable prefix, cache // boundary) → user (volatile tail). Kept as the zero-injection convenience over // MessagesWithInjection so existing callers/tests are untouched. func Messages(tpl *PromptTemplate, v RenderVars) ([]llm.Message, error) { return MessagesWithInjection(tpl, v, "") } // MessagesWithInjection builds the stage messages with the memory bank v2 injection // surface (step 4 §C, orchestrator variant "а" — a code-assembled message, NOT a new // RenderVars placeholder, so the fixed template stays stable). Layout per Р5/A6: // // system(stable prefix, CacheBoundary) → [injection: glossary/STM] → user(volatile tail) // // The injection is its OWN message placed AFTER the cache boundary (which stays on the // stable system prefix — the DeepSeek byte-prefix cache still hits; Anthropic's removed, // no cache_control concern). It is NOT part of ch.Text, so it never leaks into the // coverage len_ratio or the {{text}} placeholder (the 3a ch.Text contract). Because the // injection is a message, it is folded into request_hash/content_hash automatically — a // changed injection is a changed request (a resumed chunk reproduces it deterministically // from the frozen bank). An empty injection yields the exact 2-message list of Messages. func MessagesWithInjection(tpl *PromptTemplate, v RenderVars, injection string) ([]llm.Message, error) { sys, err := Render(tpl.System, v) if err != nil { return nil, err } user, err := Render(tpl.User, v) if err != nil { return nil, err } msgs := make([]llm.Message, 0, 3) msgs = append(msgs, llm.Message{Role: "system", Content: sys, CacheBoundary: true}) if strings.TrimSpace(injection) != "" { msgs = append(msgs, llm.Message{Role: "system", Content: injection, CacheBoundary: false}) } msgs = append(msgs, llm.Message{Role: "user", Content: user}) return msgs, nil } // Request is everything that determines ONE model call: the addressing (book/chapter/chunk/attempt), // the stage's wire parameters and the rendered messages. It is the input of RequestHash — a struct // rather than a positional argument list because the fields are 13 mostly-scalar values whose ORDER is // the only thing that told them apart at a call site (a swapped chapter/chunkIdx or role/model pair // compiles and silently addresses a DIFFERENT checkpoint: a re-billed chunk or, worse, a served one). // Field NAMES are free; the hash is over the values in the fixed order RequestHash writes them. type Request struct { BookID string Chapter int ChunkIdx int Attempt int // the regeneration dimension (a gate failed → attempt+1 → a new key) Stage string Role string Model string Temperature float64 Reasoning string JSONOnly bool MaxTokens int SnapshotID string // freezes the volatile context (config/prompts/bank/gates) Messages []llm.Message } // RequestHash is the checkpoint/resume key (§3.1): a stable hash of everything // that determines the call. The snapshot id freezes the volatile context, so // identical re-renders after a restart find their checkpoint and are neither // repeated nor re-billed. Attempt is the regeneration dimension (Phase 1: a gate // failed → attempt+1 → a new key; without it a regeneration would land on the // just-failed checkpoint and instantly escalate to premium). func RequestHash(req Request) string { h := sha256.New() // Each field is length-prefixed (8 bytes LE) instead of a NUL separator: // content with a \x00 byte can no longer shift field boundaries or forge // extra messages (review finding). Version v2 pins the hash-format // change. var lb [8]byte w := func(parts ...string) { for _, p := range parts { binary.LittleEndian.PutUint64(lb[:], uint64(len(p))) h.Write(lb[:]) h.Write([]byte(p)) } } w("tm-request-v2", req.BookID, strconv.Itoa(req.Chapter), strconv.Itoa(req.ChunkIdx), strconv.Itoa(req.Attempt), req.Stage, req.Role, req.Model, strconv.FormatFloat(req.Temperature, 'f', -1, 64), req.Reasoning, strconv.FormatBool(req.JSONOnly), strconv.Itoa(req.MaxTokens), req.SnapshotID) // Explicit message count: the list length is also part of the key. binary.LittleEndian.PutUint64(lb[:], uint64(len(req.Messages))) h.Write(lb[:]) for _, m := range req.Messages { w(m.Role, m.Content, strconv.FormatBool(m.CacheBoundary)) } return hex.EncodeToString(h.Sum(nil)) } // msgsContentHash is a content signature of the rendered messages, INDEPENDENT // of attempt/max_tokens. It guards the chunk_status resume fast-path (stagerun.go): // the source bytes are NOT folded into the snapshot (only the chunker RULES and // the semantic brief are), so a positional chunk_status row must be re-validated // against the current rendered content — otherwise an edited source would serve // a stale, divergent translation with no cache miss (the determinism invariant's // named failure; self-review Milestone 2). Length-prefixed like RequestHash so // content with NUL bytes cannot forge field/message boundaries. The template is // already snapshot-pinned, so this only has to catch source/draft edits. func msgsContentHash(msgs []llm.Message) string { h := sha256.New() var lb [8]byte w := func(s string) { binary.LittleEndian.PutUint64(lb[:], uint64(len(s))) h.Write(lb[:]) h.Write([]byte(s)) } w("tm-content-v1") binary.LittleEndian.PutUint64(lb[:], uint64(len(msgs))) h.Write(lb[:]) for _, m := range msgs { w(m.Role) w(m.Content) } return hex.EncodeToString(h.Sum(nil)) } // EstimateTokens is a cheap, deterministic token estimate for reservation // sizing and max_tokens derivation (NOT for billing — billing uses the API's // usage). Eval calibration: a CJK glyph ≈0.9 tokens, Russian/Latin ≈3 chars per // token. func EstimateTokens(s string) int { cjk, other := 0, 0 for _, r := range s { switch { case unicode.In(r, unicode.Han, unicode.Hiragana, unicode.Katakana, unicode.Hangul): cjk++ case unicode.IsSpace(r): // whitespace mostly folds into neighbouring tokens default: other++ } } est := cjk + other/3 if est < 16 { est = 16 } return est }