diff --git a/backend/cmd/tmctl/main.go b/backend/cmd/tmctl/main.go
index ae3785f..bd5406a 100644
--- a/backend/cmd/tmctl/main.go
+++ b/backend/cmd/tmctl/main.go
@@ -4,6 +4,7 @@ package main
import (
"bufio"
"context"
+ "errors"
"flag"
"fmt"
"os"
@@ -16,8 +17,20 @@ import (
"textmachine/backend/internal/pipeline"
)
+// Exit codes (Веха 2): 0 clean · 2 completed-with-flags (приёмка допускает N
+// флагов) · 1 infra failure. The exit-2 case is carried up as a typed sentinel
+// (*pipeline.CompletedWithFlags) so a "clean run, attention needed" never
+// masquerades as an infra crash and vice-versa.
func main() {
- if err := run(); err != nil {
+ err := run()
+ var flagged *pipeline.CompletedWithFlags
+ switch {
+ case err == nil:
+ return
+ case errors.As(err, &flagged):
+ fmt.Fprintln(os.Stderr, "tmctl:", err)
+ os.Exit(2)
+ default:
fmt.Fprintln(os.Stderr, "tmctl:", err)
os.Exit(1)
}
@@ -29,7 +42,11 @@ func run() error {
}
cmd, args := os.Args[1], os.Args[2:]
- fs := flag.NewFlagSet(cmd, flag.ExitOnError)
+ // ContinueOnError (not ExitOnError): the stdlib's ExitOnError calls os.Exit(2)
+ // on a parse failure, which would COLLIDE with exit code 2 (completed-with-
+ // flags). Returning the error routes a bad flag to main()'s default branch →
+ // exit 1, keeping 2 exclusive to the flagged-chunks case (self-review Веха 2).
+ fs := flag.NewFlagSet(cmd, flag.ContinueOnError)
cfgPath := fs.String("config", "", "path to book.yaml")
resnapshot := fs.Bool("resnapshot", false, "re-pin existing jobs to the current config snapshot (пере-перевод оплаченных чанков — явное согласие)")
if err := fs.Parse(args); err != nil {
@@ -69,35 +86,58 @@ func translate(ctx context.Context, cfgPath string, resnapshot bool) error {
defer r.Close()
r.Resnapshot = resnapshot
- res, err := r.TranslateOneChunk(ctx)
+ res, err := r.TranslateBook(ctx)
if err != nil {
return err
}
- fmt.Println("=== ПЕРЕВОД ===")
- fmt.Println(res.FinalText)
- fmt.Println()
- fmt.Println("=== СТОИМОСТЬ ПО СТАДИЯМ ===")
- for _, st := range res.Stages {
- src := "call"
- if st.FromResume {
- src = "checkpoint"
+ for _, ch := range res.Chunks {
+ fmt.Printf("=== ГЛАВА %d ЧАНК %d — %s%s ===\n", ch.Chapter, ch.ChunkIdx, ch.Disposition, flagSuffix(ch.FlagReason))
+ if ch.Disposition == pipeline.DispOK {
+ fmt.Println(ch.FinalText)
+ } else {
+ fmt.Printf("[ФЛАГ %s] чанк не переведён — черновик/редактура непригодны, помечен для человека\n", ch.FlagReason)
}
- fmt.Printf("%-8s %-22s %-10s $%.6f in=%d (cached=%d, cache_write=%d) out=%d+%d %dms finish=%s\n",
- st.Stage, st.Model, src, st.CostUSD,
- st.Usage.PromptTokens, st.Usage.CachedTokens, st.Usage.CacheCreationTokens,
- st.Usage.CompletionTokens, st.Usage.ReasoningTokens, st.LatencyMS, st.FinishReason)
+ for _, st := range ch.Stages {
+ how := "call"
+ switch {
+ case st.Disposition == pipeline.DispSkipped:
+ how = "skipped"
+ case st.FromResume:
+ how = "resume"
+ }
+ fmt.Printf(" %-8s %-22s %-8s %-8s%s $%.6f (cum $%.6f) in=%d (cached=%d) out=%d+%d att=%d %dms finish=%s\n",
+ st.Stage, st.Model, how, st.Disposition, flagSuffix(st.FlagReason),
+ st.CostUSD, st.CumCostUSD,
+ st.Usage.PromptTokens, st.Usage.CachedTokens,
+ st.Usage.CompletionTokens, st.Usage.ReasoningTokens, st.Attempts, st.LatencyMS, st.FinishReason)
+ }
+ fmt.Println()
}
- fmt.Printf("ИТОГО (этот запуск): $%.6f\n", res.TotalUSD)
+ fmt.Printf("ИТОГО (этот запуск): $%.6f — чанков %d, флагов %d\n", res.TotalUSD, len(res.Chunks), res.Flagged)
committed, reserved, err := r.Store.SpentUSD(res.BookID)
if err != nil {
return err
}
fmt.Printf("Ledger книги: committed=$%.6f reserved=$%.6f\n", committed, reserved)
+
+ // Exit code 2 is a typed sentinel, not an infra error: the report above is
+ // already on stdout; main() maps this to a non-zero exit for the operator.
+ if res.Flagged > 0 {
+ return &pipeline.CompletedWithFlags{Flagged: res.Flagged, Total: len(res.Chunks)}
+ }
return nil
}
+// flagSuffix renders a non-empty flag reason as "(reason)".
+func flagSuffix(r pipeline.FlagReason) string {
+ if r == "" {
+ return ""
+ }
+ return "(" + string(r) + ")"
+}
+
func report(cfgPath string) error {
r, err := pipeline.NewRunner(cfgPath, obs.NewLogger())
if err != nil {
@@ -119,6 +159,27 @@ func report(cfgPath string) error {
row.LatencyMS, row.FinishReason, row.TMHit, row.OK)
}
+ // Flag section (Веха 2): every chunk×stage whose disposition ≠ ok — the
+ // "флаг редактору" the plan requires (02-mvp Фаза-1 приёмка допускает N).
+ flags, err := r.Store.ChunkStatusesForBook(r.Book.BookID)
+ if err != nil {
+ return err
+ }
+ headerPrinted := false
+ for _, f := range flags {
+ if f.Disposition == "ok" {
+ continue
+ }
+ if !headerPrinted {
+ fmt.Printf("\n=== ФЛАГИ (disposition ≠ ok) ===\n")
+ fmt.Printf("%-4s %-6s %-8s %-9s %-16s %5s %10s %s\n",
+ "ch", "chunk", "stage", "disp", "reason", "att", "cost_usd", "detail")
+ headerPrinted = true
+ }
+ fmt.Printf("%-4d %-6d %-8s %-9s %-16s %5d %10.6f %s\n",
+ f.Chapter, f.ChunkIdx, f.Stage, f.Disposition, f.FlagReason, f.Attempts, f.CostUSD, f.Detail)
+ }
+
committed, reserved, err := r.Store.SpentUSD(r.Book.BookID)
if err != nil {
return err
diff --git a/backend/internal/pipeline/chunker.go b/backend/internal/pipeline/chunker.go
new file mode 100644
index 0000000..783f3e7
--- /dev/null
+++ b/backend/internal/pipeline/chunker.go
@@ -0,0 +1,88 @@
+package pipeline
+
+import "strings"
+
+// chunker.go: the Веха-2 segmenter SEAM. Its only job right now is to turn a
+// normalized source into an ordered list of (chapter, chunk) units so the runner
+// loop, resume and disposition machinery can be exercised over MANY chunks — the
+// hardcoded "chapter=1/chunk=0" of Фаза 0 is gone.
+//
+// This is a deliberate PLACEHOLDER. The real linguistic chunker of шаг 3
+// (sentence segmentation, read-only overlap, ruby name-reading extraction for
+// ja) replaces SplitChunks WITHOUT touching the loop — it only has to keep the
+// Chunk contract and bump chunkerVersion. chunkerVersion is folded into the
+// snapshot (render.go), so changing the segmentation is an explicit
+// re-translation (a loud --resnapshot), never a silent cache miss (R6/D5.2).
+
+// Chunk is one ordered unit of translation work.
+type Chunk struct {
+ Chapter int // 1-based
+ ChunkIdx int // 0-based within its chapter
+ Text string
+}
+
+// chapterSep splits the source into chapters. Form feed (U+000C) is the ASCII
+// "page/section break" — semantically a chapter boundary, invisible in prose,
+// and untouched by NormalizeSource (BOM/CRLF/NFC/outer-trim leave an internal
+// \f intact). A source with no \f is a single chapter (Фаза-0 backward compat:
+// the one-file example stays one chapter).
+const chapterSep = "\f"
+
+// targetChunkChars is the placeholder chunk size (≈ the plan's "1–2k tokens";
+// for a CJK source ~1 char ≈ 1 token). Paragraphs are packed up to this size and
+// never split, so the 264-char example remains a single chunk.
+const targetChunkChars = 1500
+
+// SplitChunks segments a NORMALIZED source (post-NormalizeSource) into an ordered
+// chunk list. Rule: split on \f into 1-based chapters; within a chapter pack
+// blank-line paragraphs greedily into chunks of ≤ targetChunkChars, never
+// splitting a paragraph (a paragraph larger than the target is its own chunk);
+// drop empty pieces. Fully deterministic.
+func SplitChunks(source string) []Chunk {
+ var out []Chunk
+ chapterNo := 0
+ for _, chapRaw := range strings.Split(source, chapterSep) {
+ paras := splitParagraphs(chapRaw)
+ if len(paras) == 0 {
+ continue // an empty chapter block does not consume a chapter number
+ }
+ chapterNo++
+ chunkIdx := 0
+ var buf strings.Builder
+ flush := func() {
+ if buf.Len() == 0 {
+ return
+ }
+ out = append(out, Chunk{Chapter: chapterNo, ChunkIdx: chunkIdx, Text: buf.String()})
+ chunkIdx++
+ buf.Reset()
+ }
+ for _, p := range paras {
+ // Start a new chunk if adding this paragraph would exceed the target
+ // and the current chunk is non-empty (never split a paragraph).
+ if buf.Len() > 0 && buf.Len()+len("\n\n")+len(p) > targetChunkChars {
+ flush()
+ }
+ if buf.Len() > 0 {
+ buf.WriteString("\n\n")
+ }
+ buf.WriteString(p)
+ }
+ flush()
+ }
+ return out
+}
+
+// splitParagraphs breaks a chapter block on blank lines and trims each paragraph,
+// dropping empties. Deterministic; whitespace-only paragraphs vanish.
+func splitParagraphs(block string) []string {
+ var paras []string
+ for _, raw := range strings.Split(block, "\n\n") {
+ // Collapse a run of blank lines: a "paragraph" that is itself only blank
+ // lines (from 3+ consecutive newlines) trims to "".
+ if p := strings.TrimSpace(raw); p != "" {
+ paras = append(paras, p)
+ }
+ }
+ return paras
+}
diff --git a/backend/internal/pipeline/chunker_test.go b/backend/internal/pipeline/chunker_test.go
new file mode 100644
index 0000000..3a2ec0a
--- /dev/null
+++ b/backend/internal/pipeline/chunker_test.go
@@ -0,0 +1,96 @@
+package pipeline
+
+import (
+ "reflect"
+ "strings"
+ "testing"
+)
+
+func TestSplitChunksSingleParagraph(t *testing.T) {
+ got := SplitChunks("静かな図書館の朝。")
+ want := []Chunk{{Chapter: 1, ChunkIdx: 0, Text: "静かな図書館の朝。"}}
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("single paragraph = %+v, want %+v", got, want)
+ }
+}
+
+func TestSplitChunksChapters(t *testing.T) {
+ src := "Глава один." + chapterSep + "Глава два." + chapterSep + "Глава три."
+ got := SplitChunks(src)
+ want := []Chunk{
+ {Chapter: 1, ChunkIdx: 0, Text: "Глава один."},
+ {Chapter: 2, ChunkIdx: 0, Text: "Глава два."},
+ {Chapter: 3, ChunkIdx: 0, Text: "Глава три."},
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("chapters = %+v, want %+v", got, want)
+ }
+}
+
+func TestSplitChunksPacksToTarget(t *testing.T) {
+ // Two paragraphs that together exceed the target must split into two chunks,
+ // each never splitting a paragraph.
+ p1 := strings.Repeat("a", 800)
+ p2 := strings.Repeat("b", 800)
+ got := SplitChunks(p1 + "\n\n" + p2)
+ want := []Chunk{
+ {Chapter: 1, ChunkIdx: 0, Text: p1},
+ {Chapter: 1, ChunkIdx: 1, Text: p2},
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("packing = chapters/chunks %d, want 2 chunks; got %+v", len(got), summarize(got))
+ }
+}
+
+func TestSplitChunksPacksSmallParagraphsTogether(t *testing.T) {
+ // Small paragraphs pack into one chunk (under the target), joined by a blank line.
+ got := SplitChunks("Абзац один.\n\nАбзац два.\n\nАбзац три.")
+ if len(got) != 1 {
+ t.Fatalf("small paragraphs must pack into one chunk, got %d: %+v", len(got), summarize(got))
+ }
+ if got[0].Text != "Абзац один.\n\nАбзац два.\n\nАбзац три." {
+ t.Fatalf("packed text = %q", got[0].Text)
+ }
+}
+
+func TestSplitChunksOversizeParagraphIsOwnChunk(t *testing.T) {
+ big := strings.Repeat("ы", 2000) // > targetChunkChars but never split
+ got := SplitChunks(big)
+ if len(got) != 1 || got[0].Text != big {
+ t.Fatalf("an oversize paragraph must be one un-split chunk, got %+v", summarize(got))
+ }
+}
+
+func TestSplitChunksDropsEmpties(t *testing.T) {
+ // Empty chapter blocks and whitespace-only paragraphs vanish; chapter numbers
+ // only advance for non-empty chapters.
+ src := "A" + chapterSep + " \n\n " + chapterSep + "B"
+ got := SplitChunks(src)
+ want := []Chunk{
+ {Chapter: 1, ChunkIdx: 0, Text: "A"},
+ {Chapter: 2, ChunkIdx: 0, Text: "B"},
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("empties = %+v, want %+v", got, want)
+ }
+}
+
+func TestSplitChunksDeterministic(t *testing.T) {
+ src := "Абзац.\n\nЕщё абзац." + chapterSep + strings.Repeat("длинный ", 300)
+ a, b := SplitChunks(src), SplitChunks(src)
+ if !reflect.DeepEqual(a, b) {
+ t.Fatal("SplitChunks must be deterministic")
+ }
+}
+
+func summarize(cs []Chunk) []string {
+ out := make([]string, len(cs))
+ for i, c := range cs {
+ txt := c.Text
+ if len(txt) > 12 {
+ txt = txt[:12] + "…"
+ }
+ out[i] = strings.TrimSpace(txt)
+ }
+ return out
+}
diff --git a/backend/internal/pipeline/disposition.go b/backend/internal/pipeline/disposition.go
new file mode 100644
index 0000000..ee505f5
--- /dev/null
+++ b/backend/internal/pipeline/disposition.go
@@ -0,0 +1,286 @@
+package pipeline
+
+import (
+ "fmt"
+ "regexp"
+ "strings"
+ "unicode"
+ "unicode/utf8"
+
+ "textmachine/backend/internal/llm"
+)
+
+// disposition.go: the per-chunk×stage verdict machinery of the Веха-2 runner
+// (D2). It answers three questions about a completion — is it usable, and if not
+// WHY, and may we re-attack it — WITHOUT ever touching the wire (classify is a
+// pure function of the completion text + finish reason + a little context, so a
+// resume reproduces the same verdict from the same checkpoint, no re-billing).
+//
+// This is deliberately SEPARATE from the configurable coverage QA-gate
+// (Gates.Coverage, шаг 6): that gate is opt-in and threshold-driven; the
+// classifier here is intrinsic runner robustness that is always on, so a bad
+// chunk is flagged-and-skipped instead of poisoning the pipeline (empty draft →
+// empty edit → empty export) or wedging the whole book on a single call.
+
+// Disposition is the resolved state of a chunk×stage over its checkpoints.
+type Disposition string
+
+const (
+ // DispOK — a usable completion; its text feeds the next stage and is the
+ // authoritative checkpoint.
+ DispOK Disposition = "ok"
+ // DispFlagged — the completion is unusable (refusal / echo / truncation loop
+ // / empty / decode error) and retries (if any) are exhausted; the chunk is
+ // flagged for a human and later stages are skipped. Money is still accounted.
+ DispFlagged Disposition = "flagged"
+ // DispSkipped — this stage was NOT attempted because an earlier stage of the
+ // same chunk was flagged (no garbage draft → no paid edit over garbage).
+ DispSkipped Disposition = "skipped"
+)
+
+// FlagReason is the tagged cause of a flag — a contract enum like the Finish*
+// constants (D2.2: tagging drives retry policy). The runner re-attacks ONLY the
+// retryable subset; the rest are deterministic per model/channel, so a same-model
+// retry would just re-refuse and re-bill (routing to channel B / escalation is
+// Фаза 2). Some constants are DEFINED here for contract completeness but are not
+// yet emitted by Веха 2 — see the notes — they belong to later steps.
+type FlagReason string
+
+const (
+ reasonOK FlagReason = "" // sentinel: not a flag
+
+ // Emitted by classify() — the retryable pair (D2: bigger max_tokens on the
+ // attempt axis, up to the regenerate cap, then flag).
+ FlagLength FlagReason = "length" // truncated at max_tokens (a genuine cut, not a loop)
+ FlagEmpty FlagReason = "empty" // 2xx with no usable text (thinking likely ate the whole budget)
+
+ // Emitted by classify() — deterministic, NOT retryable on the same model.
+ FlagLoopDegenerate FlagReason = "loop_degenerate" // repetition loop at a length cut (bigger budget just buys more loop, D2.3)
+ FlagHardRefusal FlagReason = "hard_refusal" // provider finish_reason=refusal
+ FlagSoftRefusal FlagReason = "soft_refusal" // refusal-blacklist match on a short output (the real insurance, D2.4)
+ FlagContentFilter FlagReason = "content_filter" // provider finish_reason=content_filter (best-effort — emission unverified across DS/GLM/Kimi/grok, D2.4)
+ FlagCJKArtifact FlagReason = "cjk_artifact" // output echoed the CJK source instead of translating (D3.4; фолбэк-черновик — шаг 7, пока просто флаг)
+
+ // Emitted by the runner's billed-decode path.
+ FlagDecodeError FlagReason = "decode_error" // 2xx with an unreadable body — billed, conservatively settled, flagged
+
+ // Reserved for later steps — DEFINED for contract stability, NOT emitted by
+ // Веха 2. coverage_fail / excision_suspect are verdicts of the configurable
+ // coverage gate (шаг 6); hard_block / upstream_not_ok are for HTTP-level
+ // content blocks handled when escalation lands (шаг 7).
+ FlagCoverageFail FlagReason = "coverage_fail"
+ FlagExcisionSuspect FlagReason = "excision_suspect"
+ FlagHardBlock FlagReason = "hard_block"
+ FlagUpstreamNotOK FlagReason = "upstream_not_ok"
+)
+
+// decodeErrorFinish is the finish_reason the runner stores on a billed-but-
+// unreadable 2xx (BilledDecodeError). classify() recognises it so a RESUMED
+// decode checkpoint re-resolves to the same FlagDecodeError verdict the live
+// path assigned — live and resume must agree (determinism of the resolve).
+const decodeErrorFinish = "decode_error"
+
+// retryable reports whether a flag may be re-attacked on the SAME model along the
+// attempt axis. Only length/empty: both are budget symptoms a bigger max_tokens
+// can cure. Everything else is deterministic (refusal/filter/echo/loop/decode) —
+// re-attacking burns money on a guaranteed repeat (D2.2/D2.3).
+func (r FlagReason) retryable() bool {
+ return r == FlagLength || r == FlagEmpty
+}
+
+// disposition maps a reason to the resolved chunk×stage state.
+func (r FlagReason) disposition() Disposition {
+ if r == reasonOK {
+ return DispOK
+ }
+ return DispFlagged
+}
+
+// classifyInput is everything classify needs. It is more than the "(text,
+// finish)" shorthand because a faithful 1:1 port of eval/refusal_bench.py
+// (03-implementation-notes §3.7) needs the source length (soft-refusal is a
+// refusal pattern on an ANOMALOUSLY SHORT output — a long translation that
+// merely quotes "I'm sorry, but…" as dialogue must NOT flag) and the target
+// language (CJK-echo detection is meaningless when translating INTO a CJK
+// language).
+type classifyInput struct {
+ Source string
+ Output string
+ Finish string
+ TargetLang string
+}
+
+// classification is classify's verdict.
+type classification struct {
+ Reason FlagReason
+ Detail string
+}
+
+func (c classification) ok() bool { return c.Reason == reasonOK }
+
+// classify is the single verdict function. ORDER MATTERS (D2.1/D2.4): the
+// deterministic hard signals (decode / content_filter / refusal finish) and the
+// text refusal-blacklist / CJK-echo are checked BEFORE length/empty, so a
+// truncated refusal is flagged as a refusal (deterministic, no paid re-attack)
+// rather than as a length cut (retryable). Pure: no time, no randomness, no map
+// iteration — resume reproduces the identical verdict from the identical
+// checkpoint.
+func classify(in classifyInput) classification {
+ // Strip a … block the model may have leaked into content
+ // (mirrors refusal_bench) before judging emptiness/length.
+ out := strings.TrimSpace(stripThink(in.Output))
+ finish := in.Finish
+
+ // 1) Deterministic finish-reason signals — never retried.
+ switch finish {
+ case decodeErrorFinish:
+ return classification{FlagDecodeError, "billed 2xx with an unreadable body"}
+ case llm.FinishContentFilter:
+ return classification{FlagContentFilter, "provider finish_reason=content_filter"}
+ case llm.FinishRefusal:
+ return classification{FlagHardRefusal, "provider finish_reason=refusal"}
+ }
+
+ // 2) Text refusal-blacklist / CJK-echo — ДО length/empty (only meaningful on
+ // non-empty text; a truncated refusal must not become a paid length retry).
+ if out != "" {
+ if isRefusal(out, in.Source) {
+ return classification{FlagSoftRefusal, "refusal-blacklist match on a short output"}
+ }
+ if !isCJKTarget(in.TargetLang) {
+ if share := cjkShare(out); share > cjkEchoThreshold {
+ return classification{FlagCJKArtifact, fmt.Sprintf("CJK share %.0f%% in output (untranslated echo)", share*100)}
+ }
+ }
+ }
+
+ // 3) Empty completion — retryable (thinking usually ate the entire budget).
+ if out == "" {
+ return classification{FlagEmpty, "empty completion (finish=" + finish + ")"}
+ }
+
+ // 4) Length cut — retryable, UNLESS the output is a degenerate repetition
+ // loop, in which case a bigger budget only buys more loop (D2.3): flag now.
+ if finish == llm.FinishLength {
+ if degenerateLoop(out) {
+ return classification{FlagLoopDegenerate, "repetition loop at a length cut"}
+ }
+ return classification{FlagLength, "truncated at max_tokens"}
+ }
+
+ // 5) A usable completion.
+ return classification{reasonOK, ""}
+}
+
+// --- refusal blacklist (en/ru/zh/ja) — ported 1:1 from eval/refusal_bench.py ---
+
+var refusalPatterns = []string{
+ `I (?:can(?:'|no)t|cannot|won'?t|am (?:not able|unable)) (?:to )?(?:assist|help|translate|continue|provide|fulfill)`,
+ `I'?m sorry,? but`,
+ `against (?:my|our) (?:content )?polic`,
+ `content polic(?:y|ies)`,
+ `не могу (?:помочь|перевести|продолжить|выполнить)`,
+ `я не буду переводить`,
+ `противоречит (?:политике|правилам)`,
+ `としてお答えできません`,
+ `申し訳ありませんが`,
+ `我(?:不能|无法)(?:协助|翻译|继续)`,
+ `抱歉[，,]`,
+ `^\s*\[?(?:REFUSED|ОТКАЗ)\]?\s*$`,
+}
+
+var refusalRE = regexp.MustCompile("(?im)" + strings.Join(refusalPatterns, "|"))
+
+var thinkRE = regexp.MustCompile(`(?s).*?\s*`)
+
+func stripThink(s string) string { return thinkRE.ReplaceAllString(s, "") }
+
+// isRefusal ports refusal_bench: a refusal-blacklist match AND an anomalously
+// short output (shorter than max(400 runes, half the source), so a long faithful
+// translation that merely contains a refusal-like phrase does not flag).
+func isRefusal(out, src string) bool {
+ if !refusalRE.MatchString(out) {
+ return false
+ }
+ threshold := utf8.RuneCountInString(src) / 2
+ if threshold < 400 {
+ threshold = 400
+ }
+ return utf8.RuneCountInString(out) < threshold
+}
+
+// --- CJK-echo (untranslated source returned instead of a translation, D3.4) ---
+
+// cjkEchoThreshold: >15% Han+kana in a non-CJK-target output means the model
+// echoed the source (refusal_bench cjk_share > 0.15).
+const cjkEchoThreshold = 0.15
+
+func cjkShare(s string) float64 {
+ if s == "" {
+ return 0
+ }
+ cjk, total := 0, 0
+ for _, r := range s {
+ total++
+ if unicode.In(r, unicode.Han, unicode.Hiragana, unicode.Katakana) {
+ cjk++
+ }
+ }
+ if total == 0 {
+ return 0
+ }
+ return float64(cjk) / float64(total)
+}
+
+// isCJKTarget reports whether the translation TARGET is a CJK language, in which
+// case CJK in the output is expected and the echo check is skipped. Phase-1
+// scope is →ru, so this is normally false.
+func isCJKTarget(lang string) bool {
+ switch strings.ToLower(strings.TrimSpace(lang)) {
+ case "zh", "ja", "ko", "zh-cn", "zh-tw":
+ return true
+ }
+ return false
+}
+
+// --- degeneration detector (n-gram loop) — runs BEFORE the length retry (D2.3) ---
+
+// degenerateLoop reports whether the text is a repetition loop: the word-level
+// trigram distinctness collapses when a phrase/sentence repeats to fill the
+// budget. Pure and deterministic (only len(distinct) is used, never map order).
+// Below a floor of words a length cut of genuinely short text is not judged a
+// loop. CJK output (no spaces → few "words") is left to the echo check.
+func degenerateLoop(text string) bool {
+ const n = 3
+ const minWords = 30
+ const distinctRatio = 0.25 // <25% distinct trigrams ⇒ ≥75% are repeats ⇒ loop
+ words := strings.Fields(text)
+ if len(words) < minWords {
+ return false
+ }
+ total := len(words) - n + 1
+ seen := make(map[string]struct{}, total)
+ for i := 0; i+n <= len(words); i++ {
+ seen[strings.Join(words[i:i+n], "\x00")] = struct{}{}
+ }
+ return float64(len(seen))/float64(total) < distinctRatio
+}
+
+// --- max_tokens on the attempt axis (D2.3) ---
+
+// maxTokensForAttempt is the PURE output-token budget for a retry attempt:
+// attempt 0 = base, each regeneration DOUBLES it (D2.3 remedy for a length cut —
+// the previous budget was too small). Purity is load-bearing: the value enters
+// request_hash, so resume must reproduce the identical per-attempt budget. Its
+// FORMULA is versioned into the snapshot (maxTokensPolicyVersion) so a change is
+// a loud --resnapshot, not a silent checkpoint miss on retried chunks (the same
+// discipline as estimatorVersion).
+func maxTokensForAttempt(base, attempt int) int {
+ if attempt <= 0 {
+ return base
+ }
+ if attempt > 20 { // defensive: never shift by a runaway amount (overflow guard)
+ attempt = 20
+ }
+ return base << uint(attempt)
+}
diff --git a/backend/internal/pipeline/disposition_test.go b/backend/internal/pipeline/disposition_test.go
new file mode 100644
index 0000000..9868129
--- /dev/null
+++ b/backend/internal/pipeline/disposition_test.go
@@ -0,0 +1,138 @@
+package pipeline
+
+import (
+ "strings"
+ "testing"
+
+ "textmachine/backend/internal/llm"
+)
+
+// classify: ORDER is the contract (D2.1/D2.4) — refusal/echo/deterministic finish
+// signals are decided BEFORE length/empty, so a truncated refusal is not turned
+// into a paid length retry.
+func TestClassifyOrder(t *testing.T) {
+ const ru = "ru"
+ cases := []struct {
+ name string
+ in classifyInput
+ wantReason FlagReason
+ }{
+ {"clean stop", classifyInput{Source: "静かな朝。", Output: "Тихое утро.", Finish: llm.FinishStop, TargetLang: ru}, reasonOK},
+ {"empty at stop", classifyInput{Source: "静かな朝。", Output: " ", Finish: llm.FinishStop, TargetLang: ru}, FlagEmpty},
+ {"empty at length (thinking ate budget)", classifyInput{Source: "静かな朝。", Output: "", Finish: llm.FinishLength, TargetLang: ru}, FlagEmpty},
+ {"genuine length cut", classifyInput{Source: strings.Repeat("текст ", 50), Output: "Начало предложения обрывается на", Finish: llm.FinishLength, TargetLang: ru}, FlagLength},
+ {"length but repetition loop", classifyInput{Source: "x", Output: strings.Repeat("ло ", 60), Finish: llm.FinishLength, TargetLang: ru}, FlagLoopDegenerate},
+ {"truncated soft refusal (NOT length)", classifyInput{Source: "静かな朝、彼女は本を開いた。長い一日の始まりだった。", Output: "Я не могу помочь с этим запросом.", Finish: llm.FinishLength, TargetLang: ru}, FlagSoftRefusal},
+ {"english refusal short", classifyInput{Source: "長い長い長い長い文章です。", Output: "I'm sorry, but I cannot assist with that.", Finish: llm.FinishStop, TargetLang: ru}, FlagSoftRefusal},
+ {"cjk echo (untranslated)", classifyInput{Source: "彼は言った。", Output: "彼は言った。とても静かな朝だった。", Finish: llm.FinishStop, TargetLang: ru}, FlagCJKArtifact},
+ {"provider content_filter finish", classifyInput{Source: "x", Output: "partial", Finish: llm.FinishContentFilter, TargetLang: ru}, FlagContentFilter},
+ {"provider refusal finish", classifyInput{Source: "x", Output: "", Finish: llm.FinishRefusal, TargetLang: ru}, FlagHardRefusal},
+ {"resumed decode checkpoint", classifyInput{Source: "x", Output: "", Finish: decodeErrorFinish, TargetLang: ru}, FlagDecodeError},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ got := classify(tc.in)
+ if got.Reason != tc.wantReason {
+ t.Fatalf("classify reason = %q, want %q (detail %q)", got.Reason, tc.wantReason, got.Detail)
+ }
+ if (got.Reason == reasonOK) != got.ok() {
+ t.Fatalf("ok() inconsistent for reason %q", got.Reason)
+ }
+ })
+ }
+}
+
+// A refusal that is NOT anomalously short (a long faithful translation that
+// merely quotes a refusal-like phrase as dialogue) must NOT flag.
+func TestClassifyLongOutputWithRefusalPhraseIsOK(t *testing.T) {
+ src := strings.Repeat("短", 100) // 100 CJK runes → threshold max(400, 50) = 400
+ out := "«Я не могу помочь тебе», — тихо сказал он, отводя взгляд. " + strings.Repeat("Слова растворялись в утреннем тумане, и никто их не услышал. ", 12)
+ got := classify(classifyInput{Source: src, Output: out, Finish: llm.FinishStop, TargetLang: "ru"})
+ if !got.ok() {
+ t.Fatalf("a long translation quoting a refusal phrase must be ok, got %q", got.Reason)
+ }
+}
+
+// Determinism: identical input yields the identical verdict (pure function).
+func TestClassifyDeterministic(t *testing.T) {
+ in := classifyInput{Source: "彼は静かに歩いた。", Output: strings.Repeat("эхо ", 40), Finish: llm.FinishLength, TargetLang: "ru"}
+ a, b := classify(in), classify(in)
+ if a != b {
+ t.Fatalf("classify is not deterministic: %+v vs %+v", a, b)
+ }
+}
+
+func TestRetryableSubset(t *testing.T) {
+ retry := map[FlagReason]bool{FlagLength: true, FlagEmpty: true}
+ all := []FlagReason{
+ reasonOK, FlagLength, FlagEmpty, FlagLoopDegenerate, FlagHardRefusal, FlagSoftRefusal,
+ FlagContentFilter, FlagCJKArtifact, FlagDecodeError, FlagCoverageFail, FlagExcisionSuspect,
+ FlagHardBlock, FlagUpstreamNotOK,
+ }
+ for _, r := range all {
+ if got, want := r.retryable(), retry[r]; got != want {
+ t.Fatalf("%q.retryable() = %v, want %v (only length/empty are retryable)", r, got, want)
+ }
+ }
+ if reasonOK.disposition() != DispOK {
+ t.Fatal("reasonOK must resolve to DispOK")
+ }
+ if FlagLength.disposition() != DispFlagged {
+ t.Fatal("a flag reason must resolve to DispFlagged")
+ }
+}
+
+// maxTokensForAttempt is pure and doubles per attempt (D2.3), which is what puts
+// a DIFFERENT max_tokens (and thus request_hash) on each regeneration.
+func TestMaxTokensForAttempt(t *testing.T) {
+ base := 2048
+ for attempt, want := range map[int]int{0: base, 1: 2 * base, 2: 4 * base, 3: 8 * base} {
+ if got := maxTokensForAttempt(base, attempt); got != want {
+ t.Fatalf("maxTokensForAttempt(%d, %d) = %d, want %d", base, attempt, got, want)
+ }
+ }
+ // Overflow guard: an absurd attempt is clamped, never a negative shift.
+ if got := maxTokensForAttempt(base, 999); got <= 0 {
+ t.Fatalf("maxTokensForAttempt overflow guard failed: %d", got)
+ }
+}
+
+func TestDegenerateLoop(t *testing.T) {
+ if !degenerateLoop(strings.Repeat("одно и то же ", 40)) {
+ t.Fatal("a repeated phrase must be a degenerate loop")
+ }
+ healthy := "Она открыла книгу и начала читать первую главу, где юный герой впервые покидал родную деревню навстречу далёким горам и неизвестной судьбе, полной опасностей."
+ if degenerateLoop(healthy) {
+ t.Fatal("healthy prose must NOT be a degenerate loop")
+ }
+ if degenerateLoop("короткий обрыв на пол") {
+ t.Fatal("a short length cut is not enough evidence for a loop")
+ }
+}
+
+func TestIsRefusalShortVsLong(t *testing.T) {
+ if !isRefusal("Извините, я не могу перевести это.", "abc") {
+ t.Fatal("a short refusal must be detected")
+ }
+ long := strings.Repeat("Это обычный длинный перевод без всякого отказа. ", 30)
+ if isRefusal(long, "abc") {
+ t.Fatal("a long non-refusal must not be flagged")
+ }
+}
+
+func TestCJKShareAndTarget(t *testing.T) {
+ if s := cjkShare("彼は言った"); s < 0.9 {
+ t.Fatalf("all-CJK text share = %.2f, want ~1", s)
+ }
+ if s := cjkShare("Полностью русский текст"); s > 0.01 {
+ t.Fatalf("russian text share = %.2f, want ~0", s)
+ }
+ if !isCJKTarget("ja") || !isCJKTarget("ZH") || isCJKTarget("ru") {
+ t.Fatal("isCJKTarget classification wrong")
+ }
+ // Into a CJK target, CJK output is expected — no echo flag.
+ got := classify(classifyInput{Source: "Утро.", Output: "静かな朝。", Finish: llm.FinishStop, TargetLang: "ja"})
+ if got.Reason == FlagCJKArtifact {
+ t.Fatal("CJK output must not be an echo artifact when the target IS CJK")
+ }
+}
diff --git a/backend/internal/pipeline/render.go b/backend/internal/pipeline/render.go
index 8d77fc2..890a5b9 100644
--- a/backend/internal/pipeline/render.go
+++ b/backend/internal/pipeline/render.go
@@ -31,8 +31,11 @@ import (
// chunkerVersion versions the segmentation/ingestion rules (включая
// нормализацию источника NormalizeSource); it is part of the snapshot, so
// re-chunking a book is an explicit re-translation, not a silent cache miss
-// (§3.4 — [НУЖНО РЕШЕНИЕ] п.2в).
-const chunkerVersion = "chunker-v1-wholefile-nfc"
+// (§3.4 — [НУЖНО РЕШЕНИЕ] п.2в). Веха 2 replaced the whole-file "one chunk" of
+// Фаза 0 with the chapter/chunk splitter (chunker.go), so the version bumps:
+// any Фаза-0 project DB re-pins loudly via the --resnapshot gate (a re-chunk is
+// a deliberate re-translation), never a silent divergent re-pay.
+const chunkerVersion = "chunker-v2-ff-para-pack"
// estimatorVersion versions EstimateTokens: его выход входит в max_tokens и
// через него в request-hash, поэтому перекалибровка весов — тоже явная
@@ -40,6 +43,15 @@ const chunkerVersion = "chunker-v1-wholefile-nfc"
// ревью: code-only правка эстиматора пере-оплатила бы полкниги).
const estimatorVersion = "estimator-v0"
+// 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 (Веха 2).
+const maxTokensPolicyVersion = "maxtok-v1-double-per-attempt"
+
// userSeparator splits a prompt template file into the system part and the
// user part. Обе части — версионируемые шаблоны в prompts/ («Редакция» позже
// правит файлы, не код).
@@ -181,6 +193,33 @@ func RequestHash(bookID string, chapter, chunkIdx, attempt int, stage, role, mod
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 (runner.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 Веха 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))
+}
+
// NormalizeSource canonicalizes the source chunk so the request-hash is stable
// across editors and operating systems (находка ревью): strip a UTF-8 BOM,
// CRLF/CR → LF, Unicode NFC, trim surrounding whitespace. Без этого файл,
diff --git a/backend/internal/pipeline/runner.go b/backend/internal/pipeline/runner.go
index e2a7414..ef403f9 100644
--- a/backend/internal/pipeline/runner.go
+++ b/backend/internal/pipeline/runner.go
@@ -19,11 +19,15 @@ import (
"textmachine/backend/internal/store"
)
-// runner.go: the Phase-0 mini-runner — one chunk through the configured stage
-// list (C1: draft→edit) with full money discipline: reserve → call → settle+
-// checkpoint (одна транзакция) → request_log. Resume: найденный чекпоинт =
-// вызов не повторяется и не оплачивается. Гейты/эскалация/циклы по главам —
-// Фаза 1, сюда же (код раннера).
+// runner.go: the Веха-2 book runner — every (chapter, chunk) from the chunker
+// through the configured stage list (C1: draft→edit) with full money discipline:
+// reserve → call → settle+checkpoint (одна транзакция) → request_log. A bad
+// chunk is FLAGGED (disposition, chunk_status) and the loop continues (D2), not
+// a run-crash; only an infra failure aborts. Resume reads chunk_status BEFORE
+// rendering, so a done or terminally-flagged chunk is never re-attacked or
+// re-billed. Retries walk the `attempt` axis (bigger max_tokens for the
+// retryable length/empty subset, up to the regenerate cap, then flag).
+// Escalation and the real linguistic chunker are later steps of Фаза 1.
// Runner executes a book's pipeline.
type Runner struct {
@@ -175,10 +179,15 @@ func (r *Runner) snapshotID() (id, payload string, err error) {
Capability json.RawMessage `json:"capability,omitempty"`
}
snap := struct {
- BriefHash string `json:"brief_hash"`
- ChunkerVersion string `json:"chunker_version"`
- EstimatorVersion string `json:"estimator_version"`
- PipelineCore string `json:"pipeline_core"`
+ BriefHash string `json:"brief_hash"`
+ ChunkerVersion string `json:"chunker_version"`
+ EstimatorVersion string `json:"estimator_version"`
+ // MaxTokensPolicy versions the attempt→max_tokens scaling (Веха 2): a
+ // change to the retry doubling shifts attempt≥1 request_hashes, so it
+ // belongs in the snapshot as a loud invalidation (same class as
+ // estimator_version, applied to the regeneration axis).
+ MaxTokensPolicy string `json:"max_tokens_policy"`
+ PipelineCore string `json:"pipeline_core"`
// Defaults влияют на maxTokens, а тот входит в request-hash: без них
// правка max_output_ratio молча инвалидировала бы все чекпоинты в
// обход snapshot-гейта (находка ревью).
@@ -201,6 +210,7 @@ func (r *Runner) snapshotID() (id, payload string, err error) {
BriefHash: r.Book.BriefHash(),
ChunkerVersion: chunkerVersion,
EstimatorVersion: estimatorVersion,
+ MaxTokensPolicy: maxTokensPolicyVersion,
PipelineCore: r.Pipeline.Core,
MaxOutputRatio: r.Pipeline.Defaults.MaxOutputRatio,
MinMaxTokens: r.Pipeline.Defaults.MinMaxTokens,
@@ -247,33 +257,72 @@ func (r *Runner) snapshotID() (id, payload string, err error) {
return hex.EncodeToString(sum[:]), string(data), nil
}
-// StageResult reports one executed stage.
+// StageResult reports one executed stage of a chunk.
type StageResult struct {
Stage string
Role string
Model string // фактически ответившая модель
- FromResume bool // served from a checkpoint, no call made
+ FromResume bool // no provider call was made THIS run (fully served from checkpoints)
Usage llm.Usage
- CostUSD float64
+ 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
+ Text string // the usable output (only on an ok disposition)
+ Disposition Disposition
+ FlagReason FlagReason // "" when ok
+ Detail string
+ Attempts int
}
-// RunResult is the whole chunk's outcome.
-type RunResult struct {
- BookID string
- Chapter int
- ChunkIdx int
- Stages []StageResult
- FinalText string
- TotalUSD float64
+// ChunkOutcome is one chunk's result across the stage list.
+type ChunkOutcome struct {
+ Chapter int
+ ChunkIdx int
+ Stages []StageResult
+ FinalText string // the last stage's output; "" when the chunk is flagged
+ 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
}
-// TranslateOneChunk runs the configured stages over the book's source file as
-// a single chunk (Фаза 0). Chapter/chunk are fixed at 1/0 until the chunker
-// lands in Phase 1.
-func (r *Runner) TranslateOneChunk(ctx context.Context) (*RunResult, error) {
+// 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) {
srcRaw, err := os.ReadFile(r.Book.SourceFile)
if err != nil {
return nil, fmt.Errorf("pipeline: read source: %w", err)
@@ -284,6 +333,8 @@ func (r *Runner) TranslateOneChunk(ctx context.Context) (*RunResult, error) {
return nil, fmt.Errorf("pipeline: source file %s is empty", r.Book.SourceFile)
}
+ // Snapshot is book-level (brief + stage plan + memory), computed once and
+ // upserted before the loop; every job pins to it.
snapID, snapPayload, err := r.snapshotID()
if err != nil {
return nil, err
@@ -292,96 +343,276 @@ func (r *Runner) TranslateOneChunk(ctx context.Context) (*RunResult, error) {
return nil, err
}
- const chapter, chunkIdx = 1, 0
- res := &RunResult{BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx}
- prev := ""
+ chunks := SplitChunks(source)
+ if len(chunks) == 0 {
+ return nil, fmt.Errorf("pipeline: source file %s produced no chunks after normalization", r.Book.SourceFile)
+ }
- for _, st := range r.Pipeline.Stages {
- sr, err := r.runStage(ctx, st, snapID, chapter, chunkIdx, source, prev)
+ res := &BookResult{BookID: r.Book.BookID}
+ for _, ch := range chunks {
+ outcome, err := r.translateChunk(ctx, snapID, ch)
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.Stages = append(res.Stages, *sr)
- res.TotalUSD += sr.CostUSD
- prev = sr.Text
+ res.Chunks = append(res.Chunks, *outcome)
+ res.TotalUSD += outcome.CostUSD
+ if outcome.Disposition == DispFlagged {
+ res.Flagged++
+ }
}
- res.FinalText = prev
return res, nil
}
-func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, chapter, chunkIdx int, source, prev string) (*StageResult, error) {
- job, err := r.Store.EnsureJob(r.Book.BookID, chapter, st.Name, snapID)
+// translateChunk drives ONE chunk through the stage list. Stages run in order,
+// each feeding the next; the FIRST flagged stage stops the chunk — later stages
+// are recorded `skipped` (no paid edit over a garbage draft, D2). Returns an
+// error only on an infra failure.
+func (r *Runner) translateChunk(ctx context.Context, snapID string, ch Chunk) (*ChunkOutcome, error) {
+ out := &ChunkOutcome{Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Disposition: DispOK}
+ prev := ""
+ flagged := false
+ var flagReason FlagReason
+
+ for stageIdx, st := range r.Pipeline.Stages {
+ if flagged {
+ // An earlier stage flagged → this stage is not attempted or billed.
+ detail := fmt.Sprintf("skipped: an upstream stage was flagged (%s)", flagReason)
+ if err := r.Store.UpsertChunkStatus(store.ChunkStatus{
+ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name,
+ SnapshotID: snapID, Disposition: string(DispSkipped), FlagReason: string(flagReason),
+ Detail: detail,
+ }); err != nil {
+ return out, fmt.Errorf("pipeline: record skipped chunk_status: %w", err)
+ }
+ out.Stages = append(out.Stages, StageResult{
+ Stage: st.Name, Role: st.Role, Model: st.Model,
+ Disposition: DispSkipped, FlagReason: flagReason, Detail: detail,
+ })
+ continue
+ }
+
+ sr, err := r.runStage(ctx, st, stageIdx, snapID, ch, prev)
+ if err != nil {
+ return out, err
+ }
+ out.Stages = append(out.Stages, *sr)
+ out.CostUSD += sr.CostUSD
+ if sr.Disposition == DispFlagged {
+ flagged = true
+ flagReason = sr.FlagReason
+ continue
+ }
+ prev = sr.Text
+ }
+
+ if flagged {
+ out.Disposition = DispFlagged
+ out.FlagReason = flagReason
+ out.FinalText = "" // garbage/refusal never propagates to the next chunk or export
+ } else {
+ out.FinalText = prev
+ }
+ return out, nil
+}
+
+// runStage runs ONE stage of ONE chunk to a terminal disposition. It first
+// resumes a resolved verdict (chunk_status read BEFORE any render — anti-wedge
+// #1), otherwise walks the attempt axis: render → per-attempt checkpoint resume
+// or a fresh reserve/call/settle → classify → retry the retryable {length,empty}
+// subset with a doubled budget up to the regenerate cap, then flag. Returns an
+// error ONLY on an infra failure; a bad completion is a disposition, never an
+// error.
+func (r *Runner) runStage(ctx context.Context, st config.Stage, stageIdx int, snapID string, ch Chunk, prev string) (*StageResult, error) {
+ job, err := r.Store.EnsureJob(r.Book.BookID, ch.Chapter, st.Name, snapID)
if err != nil {
return nil, err
}
- // Snapshot pinning (Р6): джоба заморожена на снапшоте своего старта. Если
- // конфиги/промпты с тех пор изменились — это инвалидация чекпоинтов и
- // пере-оплата; выполняем её только по явной команде.
+ // Snapshot pinning (Р6): a job frozen on a stale snapshot is a loud stop
+ // unless --resnapshot explicitly accepts the re-translation.
if job.SnapshotID != snapID {
if !r.Resnapshot {
return nil, fmt.Errorf("pipeline: job %s/ch%d/%s was started under snapshot %.12s, current config renders snapshot %.12s — конфиг/промпты изменились; уже оплаченные чекпоинты станут недействительны и вызовы будут пере-оплачены; повторить с --resnapshot, чтобы принять это явно",
- r.Book.BookID, chapter, st.Name, job.SnapshotID, snapID)
+ r.Book.BookID, ch.Chapter, st.Name, job.SnapshotID, snapID)
}
if err := r.Store.UpdateJobSnapshot(job.ID, snapID); err != nil {
return nil, err
}
r.Log.WarnContext(ctx, "job re-pinned to new snapshot (--resnapshot)", "stage", st.Name, "old", job.SnapshotID[:12], "new", snapID[:12])
}
- // Статус 'running' выставляется НЕ здесь, а прямо перед вызовом провайдера:
- // иначе отказ потолка/ошибка сборки клиента/промах гейта оставляли бы
- // джобу навсегда в 'running' (находка ревью). Отказ потолка — retriable,
- // джоба остаётся 'pending'; настоящая ошибка — 'failed'.
- msgs, err := Messages(r.templates[st.Name], RenderVars{Book: r.Book, Text: source, Draft: prev})
+ // Render the wire messages up front. This is cheap (string substitution — the
+ // expensive part is the LLM call, still gated below) and it lets the resume
+ // fast-path be CONTENT-VERIFIED: the source bytes are not in the snapshot, so
+ // a positional chunk_status row must be checked against the current rendered
+ // content, else an edited source would serve a stale, divergent translation.
+ msgs, err := Messages(r.templates[st.Name], RenderVars{Book: r.Book, Text: ch.Text, Draft: prev})
if err != nil {
return nil, err
}
+ contentHash := msgsContentHash(msgs)
- // max_tokens: калибровка полигона — русский выход ≈1.9× входа в токенах;
- // ratio из конфига, floor защищает короткие чанки. Входит в request-hash
- // (меняет вызов), поэтому считается детерминированно от исходника.
- inputEst := EstimateTokens(source)
- maxTokens := int(float64(inputEst) * r.Pipeline.Defaults.MaxOutputRatio)
- if maxTokens < r.Pipeline.Defaults.MinMaxTokens {
- maxTokens = r.Pipeline.Defaults.MinMaxTokens
+ // Anti-wedge #1 + content-guard (self-review): resume resolves from
+ // chunk_status BEFORE any LLM call. A flagged chunk is not in TM (garbage
+ // never commits), so absence-in-TM ≠ "not done" — the disposition row is the
+ // resume authority, and a terminally-flagged chunk is NOT re-attacked (no
+ // infinite paid loop, even if the regenerate budget was raised). The row is
+ // trusted ONLY when snapshot AND content both match: a stale-snapshot row
+ // (post --resnapshot) or a stale-content row (source edited, not in the
+ // snapshot) is ignored → the attempt loop below re-derives it content-safely
+ // (a source edit becomes a silent per-chunk re-translate, §3.4, not a stale
+ // serve). A `skipped` row is ignored too: translateChunk re-derives the skip.
+ if cs, err := r.Store.GetChunkStatus(r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name); err != nil {
+ return nil, err
+ } else if cs != nil && cs.SnapshotID == snapID && cs.ContentHash == contentHash && cs.Disposition != string(DispSkipped) {
+ return r.resumeFromChunkStatus(ctx, st, ch, cs)
}
- // attempt 0: измерение регенераций подключит гейт-цикл Фазы 1.
- const attempt = 0
- reqHash := RequestHash(r.Book.BookID, chapter, chunkIdx, attempt, st.Name, st.Role, st.Model,
- st.Temperature, st.Reasoning, false, maxTokens, snapID, msgs)
+ // max_tokens base is sized from the text THIS stage processes: the source for
+ // the translator, the prior draft for later stages (D2.5 — the monolingual
+ // editor works over the Russian draft ≈1.9× the CJK source, so sizing edit
+ // from source under-budgets and false-triggers a length retry).
+ sizingText := ch.Text
+ if stageIdx > 0 {
+ sizingText = prev
+ }
+ baseMaxTokens := int(float64(EstimateTokens(sizingText)) * r.Pipeline.Defaults.MaxOutputRatio)
+ if baseMaxTokens < r.Pipeline.Defaults.MinMaxTokens {
+ baseMaxTokens = r.Pipeline.Defaults.MinMaxTokens
+ }
- // Дополняем ReqInfo, СОХРАНЯЯ решения допуска (LogBodies) — перезапись с
- // нуля отрезала бы задокументированный debug-канал (находка ревью).
+ // Дополняем ReqInfo, СОХРАНЯЯ решения допуска (LogBodies) — перезапись с нуля
+ // отрезала бы задокументированный debug-канал (находка ревью).
ri, _ := obs.ReqInfoFromContext(ctx)
- ri.Book, ri.Chapter, ri.Chunk, ri.Stage, ri.Role = r.Book.BookID, chapter, chunkIdx, st.Name, st.Role
+ ri.Book, ri.Chapter, ri.Chunk, ri.Stage, ri.Role = r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, st.Role
ctx = obs.WithReqInfo(ctx, ri)
- // Resume path: a checkpoint means the call already happened and was paid —
- // serve it, log the hit, never re-bill (приёмка: kill -9 теряет ≤1 вызов).
- // Пустой чекпоинт (оплаченный отказ/обрезка/нечитаемое тело) НЕ подаётся
- // дальше молча: черновик="" отравил бы редактуру и экспорт (находка ревью).
- if cp, err := r.Store.GetCheckpoint(reqHash); err != nil {
- return nil, err
- } else if cp != nil {
- if strings.TrimSpace(cp.ResponseText) == "" {
- return nil, fmt.Errorf("pipeline: stage %s has a BILLED but empty checkpoint (finish=%s, hash %.12s) — вызов был оплачен, но пригодного текста нет; до регенераций Фазы 1: удалить строку checkpoints вручную или дождаться механизма attempt", st.Name, cp.FinishReason, reqHash)
- }
- var usage llm.Usage
- _ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
- r.Store.LogRequest(ctx, r.Log, store.RequestLog{
- BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
- Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: cp.ModelActual,
- RequestHash: reqHash, TMHit: true, OK: true, FinishReason: cp.FinishReason,
- })
- if err := r.Store.SetJobStatus(job.ID, "done"); err != nil {
- return nil, err
- }
- r.Log.InfoContext(ctx, "stage served from checkpoint", "stage", st.Name, "hash", reqHash[:12])
- return &StageResult{Stage: st.Name, Role: st.Role, Model: cp.ModelActual,
- FromResume: true, Usage: usage, CostUSD: 0, FinishReason: cp.FinishReason, Text: cp.ResponseText}, nil
+ maxRegen := r.Pipeline.Retries.RegenerateBeforeEscalate
+ if maxRegen < 0 {
+ maxRegen = 0
}
+ var cumCost, runCost float64
+ var last stageAttempt
+ anyFresh := false
+ attemptsMade := 0
+ for attempt := 0; ; attempt++ {
+ maxTokens := maxTokensForAttempt(baseMaxTokens, attempt)
+ att, err := r.runAttempt(ctx, st, snapID, ch, job, attempt, maxTokens, msgs)
+ if err != nil {
+ return nil, err // infra failure
+ }
+ attemptsMade = attempt + 1
+ cumCost += att.cumCost
+ runCost += att.runCost
+ anyFresh = anyFresh || att.freshCall
+ last = att
+ if att.cls.ok() {
+ break
+ }
+ // Flagged: re-attack only the retryable subset, only while regenerations
+ // remain (a bigger budget on the attempt axis, D2.3). Everything else is
+ // deterministic — a same-model retry would re-refuse and re-bill (D2.2).
+ if att.cls.Reason.retryable() && attempt < maxRegen {
+ r.Log.WarnContext(ctx, "stage flagged, regenerating with a larger budget",
+ "stage", st.Name, "chapter", ch.Chapter, "chunk", ch.ChunkIdx,
+ "attempt", attempt, "reason", string(att.cls.Reason), "next_max_tokens", maxTokensForAttempt(baseMaxTokens, attempt+1))
+ continue
+ }
+ break
+ }
+
+ disposition := last.cls.Reason.disposition()
+ finalHash := ""
+ if disposition == DispOK {
+ finalHash = last.reqHash
+ }
+ // chunk_status is a RESOLVE over the checkpoints: cost_usd sums every attempt
+ // (F3-honest — retries are counted), final_hash points at the authoritative
+ // checkpoint the ok path serves on resume.
+ if err := r.Store.UpsertChunkStatus(store.ChunkStatus{
+ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name,
+ SnapshotID: snapID, ContentHash: contentHash, Disposition: string(disposition), FlagReason: string(last.cls.Reason),
+ Attempts: attemptsMade, FinalHash: finalHash, CostUSD: cumCost, Detail: last.cls.Detail,
+ }); err != nil {
+ return nil, fmt.Errorf("pipeline: record chunk_status: %w", err)
+ }
+
+ sr := &StageResult{
+ Stage: st.Name, Role: st.Role, Model: last.modelActual,
+ FromResume: !anyFresh,
+ Usage: last.usage,
+ CostUSD: runCost,
+ CumCostUSD: cumCost,
+ LatencyMS: last.latency,
+ FinishReason: last.finish,
+ Disposition: disposition,
+ FlagReason: last.cls.Reason,
+ Detail: last.cls.Detail,
+ Attempts: attemptsMade,
+ }
+ if disposition == DispOK {
+ sr.Text = last.text
+ } else {
+ r.Log.WarnContext(ctx, "stage flagged", "stage", st.Name, "chapter", ch.Chapter,
+ "chunk", ch.ChunkIdx, "reason", string(last.cls.Reason), "attempts", attemptsMade,
+ "cum_cost_usd", fmt.Sprintf("%.6f", cumCost))
+ }
+ return sr, nil
+}
+
+// stageAttempt is the result of one attempt (a checkpoint hit or a fresh call).
+type stageAttempt struct {
+ reqHash string
+ cls classification
+ text string
+ usage llm.Usage
+ finish string
+ modelActual string
+ latency int
+ runCost float64 // billed THIS run (0 on a checkpoint hit)
+ cumCost float64 // this attempt's cost (checkpoint cost on a hit, fresh cost on a call)
+ freshCall bool // a provider call was made this run
+}
+
+// runAttempt executes exactly one attempt on the request-hash axis: a checkpoint
+// hit is classified for free (self-heal, incl. legacy Фаза-0 empty/decode
+// checkpoints — no re-billing), otherwise a fresh reserve → call → settle+
+// checkpoint. It returns an error only on an infra failure; a bad completion
+// comes back as a classification on the attempt.
+func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, ch Chunk, job *store.Job, attempt, maxTokens int, msgs []llm.Message) (stageAttempt, error) {
+ reqHash := RequestHash(r.Book.BookID, ch.Chapter, ch.ChunkIdx, attempt, st.Name, st.Role, st.Model,
+ st.Temperature, st.Reasoning, false, maxTokens, snapID, msgs)
+ att := stageAttempt{reqHash: reqHash, modelActual: st.Model}
+
+ // Resume on the attempt axis: a checkpoint means THIS attempt already happened
+ // and was billed — classify its text and never re-bill (kill -9 loses ≤1 call;
+ // legacy empty/decode checkpoints self-heal by classification, not a crash).
+ if cp, err := r.Store.GetCheckpoint(reqHash); err != nil {
+ return att, err
+ } else if cp != nil {
+ var usage llm.Usage
+ _ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
+ att.usage = usage
+ att.text = cp.ResponseText
+ att.finish = cp.FinishReason
+ att.modelActual = cp.ModelActual
+ att.cumCost = cp.CostUSD
+ att.cls = classify(classifyInput{Source: ch.Text, Output: cp.ResponseText, Finish: cp.FinishReason, TargetLang: r.Book.TargetLang})
+ r.Store.LogRequest(ctx, r.Log, store.RequestLog{
+ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
+ Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: cp.ModelActual,
+ RequestHash: reqHash, TMHit: true, OK: att.cls.ok(), FinishReason: cp.FinishReason,
+ Degraded: degradedTag(att.cls),
+ })
+ _ = r.Store.SetJobStatus(job.ID, "done")
+ r.Log.InfoContext(ctx, "attempt served from checkpoint", "stage", st.Name,
+ "attempt", attempt, "hash", reqHash[:12], "disposition", string(att.cls.Reason.disposition()))
+ return att, nil
+ }
+
+ // Fresh call: reserve → call → settle+checkpoint (atomic, §3.3).
price := r.Pricer.PriceFor(st.Model)
promptEst := 0
for _, m := range msgs {
@@ -394,27 +625,29 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c
})
if err != nil {
_ = r.Store.SetJobStatus(job.ID, "failed")
- return nil, err
+ return att, err
}
switch verdict {
case store.ReserveDeniedBook:
- // Джоба остаётся 'pending' (не 'failed'): поднимут потолок — продолжит.
- return nil, fmt.Errorf("pipeline: book USD ceiling reached (%.2f$) — raise ceilings.book_usd or stop", r.Book.Ceilings.BookUSD)
+ // Ceiling is a hard, book-wide stop (not a per-chunk flag): the job stays
+ // 'pending' and resume continues once the ceiling is raised.
+ return att, fmt.Errorf("pipeline: book USD ceiling reached (%.2f$) — raise ceilings.book_usd or stop", r.Book.Ceilings.BookUSD)
case store.ReserveDeniedDay:
- return nil, fmt.Errorf("pipeline: daily USD ceiling reached (%.2f$)", r.Book.Ceilings.DayUSD)
+ return att, fmt.Errorf("pipeline: daily USD ceiling reached (%.2f$)", r.Book.Ceilings.DayUSD)
}
client, err := r.client(st.Model)
if err != nil {
_ = r.Store.Release(resv)
_ = r.Store.SetJobStatus(job.ID, "failed")
- return nil, err
+ return att, err
}
-
if err := r.Store.SetJobStatus(job.ID, "running"); err != nil {
_ = r.Store.Release(resv)
- return nil, err
+ return att, err
}
+ att.freshCall = true
+
start := time.Now()
resp, err := client.Complete(ctx, llm.LLMRequest{
Model: st.Model,
@@ -423,55 +656,64 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c
Temperature: st.Temperature,
ReasoningEffort: st.Reasoning,
})
- latency := int(time.Since(start).Milliseconds())
+ att.latency = int(time.Since(start).Milliseconds())
if err != nil {
- // 2xx с нечитаемым телом: провайдер УЖЕ списал деньги — резерв не
- // возвращаем, консервативно селтлим ОЦЕНКУ с пустым чекпоинтом
- // (переучёт ≤1 оценки безопаснее слепого потолка) и падаем громко.
var bde *llm.BilledDecodeError
if errors.As(err, &bde) {
+ // 2xx with an unreadable body: the provider ALREADY billed. Do not
+ // release; conservatively settle the ESTIMATE with an empty decode
+ // checkpoint, then FLAG (anti-wedge #2 — the loop continues, this is
+ // not an infra crash). A settle failure here is a real infra fault.
if serr := r.Store.SettleWithCheckpoint(resv, estimate, store.Checkpoint{
- RequestHash: reqHash, JobID: job.ID, ChunkIdx: chunkIdx, Attempt: attempt,
+ RequestHash: reqHash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: attempt,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: st.Model,
- ResponseText: "", UsageJSON: "{}", CostUSD: estimate, FinishReason: "decode_error",
+ ResponseText: "", UsageJSON: "{}", CostUSD: estimate, FinishReason: decodeErrorFinish,
}); serr != nil {
- r.Log.ErrorContext(ctx, "settle after billed decode failure also failed", "err", serr)
+ return att, fmt.Errorf("pipeline: settle after billed decode failure: %w", serr)
}
+ att.finish = decodeErrorFinish
+ att.cumCost, att.runCost = estimate, estimate
+ att.cls = classification{FlagDecodeError, "billed 2xx with an unreadable body: " + err.Error()}
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
- BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
+ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model,
- RequestHash: reqHash, CostUSD: estimate, LatencyMS: latency,
- Degraded: "billed_2xx_decode_failed", Err: err.Error(), OK: false,
+ RequestHash: reqHash, CostUSD: estimate, LatencyMS: att.latency,
+ Degraded: "billed_2xx_decode_failed", Err: err.Error(), OK: false, FinishReason: decodeErrorFinish,
})
- _ = r.Store.SetJobStatus(job.ID, "failed")
- return nil, fmt.Errorf("pipeline: stage %s: %w (оценка $%.6f заселтлена консервативно)", st.Name, err, estimate)
+ _ = r.Store.SetJobStatus(job.ID, "done")
+ return att, nil
}
- // No 2xx ever arrived: nothing was billed by the provider — release the
- // reservation and surface the failure with a telemetry row.
+ // No 2xx ever arrived: nothing was billed. Release and surface as an INFRA
+ // failure — a long book pauses/resumes on an outage or terminal 4xx (D4),
+ // rather than flag-storming every remaining chunk on a dead provider.
if rerr := r.Store.Release(resv); rerr != nil {
r.Log.ErrorContext(ctx, "release after failed call also failed", "err", rerr)
}
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
- BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
+ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model,
- RequestHash: reqHash, LatencyMS: latency, Err: err.Error(), OK: false,
+ RequestHash: reqHash, LatencyMS: att.latency, Err: err.Error(), OK: false,
})
_ = r.Store.SetJobStatus(job.ID, "failed")
- return nil, fmt.Errorf("pipeline: stage %s call: %w", st.Name, err)
+ return att, fmt.Errorf("pipeline: stage %s call: %w", st.Name, err)
}
modelActual := resp.Model
if modelActual == "" {
modelActual = st.Model
}
+ att.modelActual = modelActual
+ att.text = resp.Text
+ att.finish = resp.FinishReason
+ att.usage = resp.Usage
+
// Цена по фактически ответившей модели, с fallback на ЗАПРОШЕННУЮ (не на
// дешёвый глобальный якорь), если провайдер вернул канонизированный слаг.
price = r.Pricer.PriceForResponse(st.Model, modelActual)
cost := ledger.CostUSD(price, resp.Usage)
- // Платная модель (InputPerM>0) вернула 2xx с нулевым usage — баг учёта
- // провайдера/необычный ответ: $0 ослепил бы потолок, поэтому берём
- // консервативную оценку и предупреждаем (находка ревью). Для local ($0
- // цена) нулевой usage штатен — остаётся $0.
+ // Платная модель (InputPerM>0) вернула 2xx с нулевым usage — $0 ослепил бы
+ // потолок: берём консервативную оценку. Для local ($0 цена) нулевой usage
+ // штатен — остаётся $0.
if cost == 0 && price.InputPerM > 0 && resp.Usage.PromptTokens == 0 && resp.Usage.CompletionTokens == 0 {
r.Log.WarnContext(ctx, "paid 2xx with zero usage; settling the reservation estimate to keep the ceiling honest",
"stage", st.Name, "model", modelActual, "estimate_usd", fmt.Sprintf("%.6f", estimate))
@@ -479,12 +721,15 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c
}
usageJSON, err := json.Marshal(resp.Usage)
if err != nil {
- return nil, err
+ return att, err
}
+ att.cumCost, att.runCost = cost, cost
- // Деньги: settle + сырой ответ — одна транзакция (§3.3).
+ // Деньги: settle + сырой ответ — одна транзакция (§3.3). ВСЕГДА, даже для
+ // пустого/усечённого/отказного ответа: он оплачен провайдером, чекпоинт
+ // хранит его до цента, а годность решает classify ПОСЛЕ (деньги ≠ вердикт).
if err := r.Store.SettleWithCheckpoint(resv, cost, store.Checkpoint{
- RequestHash: reqHash, JobID: job.ID, ChunkIdx: chunkIdx, Attempt: attempt,
+ RequestHash: reqHash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: attempt,
Stage: st.Name, Role: st.Role,
ModelRequested: st.Model, ModelActual: modelActual, ResponseText: resp.Text,
UsageJSON: string(usageJSON), CostUSD: cost, FinishReason: resp.FinishReason,
@@ -492,45 +737,85 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c
}); err != nil {
// The provider HAS billed this 2xx; failing to persist means the money
// state is behind reality — fail loud, never continue on top.
- return nil, fmt.Errorf("pipeline: settle stage %s: %w", st.Name, err)
+ return att, fmt.Errorf("pipeline: settle stage %s: %w", st.Name, err)
}
- // Пустой оплаченный ответ (finish=length: thinking съел бюджет; фильтр):
- // деньги заселтлены, чекпоинт есть — но дальше по конвейеру его НЕ подаём:
- // черновик="" молча отравил бы редактуру. Громкая остановка (регенерация
- // по attempt — Фаза 1).
- if strings.TrimSpace(resp.Text) == "" {
- r.Store.LogRequest(ctx, r.Log, store.RequestLog{
- BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
- Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: modelActual,
- RequestHash: reqHash,
- PromptTokens: resp.Usage.PromptTokens, CachedTokens: resp.Usage.CachedTokens,
- CacheCreationTokens: resp.Usage.CacheCreationTokens,
- CompletionTokens: resp.Usage.CompletionTokens, ReasoningTokens: resp.Usage.ReasoningTokens,
- CostUSD: cost, LatencyMS: latency, FinishReason: resp.FinishReason,
- Degraded: "empty_completion", OK: false,
- })
- _ = r.Store.SetJobStatus(job.ID, "failed")
- return nil, fmt.Errorf("pipeline: stage %s returned an EMPTY completion (finish=%s, billed $%.6f, checkpointed) — вероятно thinking/фильтр съел бюджет; поднимите max_tokens/смените модель и удалите строку checkpoints %.12s", st.Name, resp.FinishReason, cost, reqHash)
- }
+ // Classify AFTER the money is durably settled+checkpointed. F4 lives here: a
+ // non-empty truncated length draft is now classified (flagged/retried), never
+ // silently passed downstream as OK; an empty completion is flagged too, not a
+ // run-crash.
+ att.cls = classify(classifyInput{Source: ch.Text, Output: resp.Text, Finish: resp.FinishReason, TargetLang: r.Book.TargetLang})
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
- BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
+ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: modelActual,
- RequestHash: reqHash,
+ RequestHash: reqHash,
PromptTokens: resp.Usage.PromptTokens, CachedTokens: resp.Usage.CachedTokens,
CacheCreationTokens: resp.Usage.CacheCreationTokens,
CompletionTokens: resp.Usage.CompletionTokens, ReasoningTokens: resp.Usage.ReasoningTokens,
- CostUSD: cost, LatencyMS: latency, FinishReason: resp.FinishReason, OK: true,
+ CostUSD: cost, LatencyMS: att.latency, FinishReason: resp.FinishReason,
+ OK: att.cls.ok(), Degraded: degradedTag(att.cls),
})
- if err := r.Store.SetJobStatus(job.ID, "done"); err != nil {
- return nil, err
- }
- r.Log.InfoContext(ctx, "stage completed", "stage", st.Name, "model", modelActual,
- "cost_usd", fmt.Sprintf("%.6f", cost), "latency_ms", latency,
- "prompt_tokens", resp.Usage.PromptTokens, "completion_tokens", resp.Usage.CompletionTokens)
-
- return &StageResult{Stage: st.Name, Role: st.Role, Model: modelActual,
- Usage: resp.Usage, CostUSD: cost, LatencyMS: latency,
- FinishReason: resp.FinishReason, Text: resp.Text}, nil
+ _ = r.Store.SetJobStatus(job.ID, "done")
+ r.Log.InfoContext(ctx, "attempt completed", "stage", st.Name, "attempt", attempt, "model", modelActual,
+ "cost_usd", fmt.Sprintf("%.6f", cost), "latency_ms", att.latency,
+ "disposition", string(att.cls.Reason.disposition()), "reason", string(att.cls.Reason))
+ return att, nil
+}
+
+// resumeFromChunkStatus serves a chunk×stage whose disposition is already
+// resolved (read before any render — no re-billing). An ok stage returns its
+// authoritative checkpoint text to feed the next stage; a flagged stage returns
+// the flag WITHOUT re-attacking (a terminal flag never re-enters the paid loop).
+func (r *Runner) resumeFromChunkStatus(ctx context.Context, st config.Stage, ch Chunk, cs *store.ChunkStatus) (*StageResult, error) {
+ sr := &StageResult{
+ Stage: st.Name, Role: st.Role, Model: st.Model, FromResume: true,
+ CostUSD: 0, CumCostUSD: cs.CostUSD,
+ Disposition: Disposition(cs.Disposition), FlagReason: FlagReason(cs.FlagReason),
+ Detail: cs.Detail, Attempts: cs.Attempts,
+ }
+ if cs.Disposition == string(DispOK) {
+ cp, err := r.Store.GetCheckpoint(cs.FinalHash)
+ if err != nil {
+ return nil, err
+ }
+ if cp == nil || strings.TrimSpace(cp.ResponseText) == "" {
+ // chunk_status ok is only written AFTER its checkpoint is durably
+ // settled, so this is a torn/tampered store — fail loud rather than
+ // feed the next stage an empty draft.
+ return nil, fmt.Errorf("pipeline: chunk_status ok for %s/ch%d/chunk%d/%s references checkpoint %.12s which is missing/empty — inconsistent store",
+ r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, cs.FinalHash)
+ }
+ var usage llm.Usage
+ _ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
+ sr.Usage = usage
+ sr.Model = cp.ModelActual
+ sr.FinishReason = cp.FinishReason
+ sr.Text = cp.ResponseText
+ }
+ r.Store.LogRequest(ctx, r.Log, store.RequestLog{
+ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
+ Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: sr.Model,
+ RequestHash: cs.FinalHash, TMHit: true, OK: cs.Disposition == string(DispOK),
+ FinishReason: sr.FinishReason, Degraded: nonOKTag(cs.Disposition, cs.FlagReason),
+ })
+ r.Log.InfoContext(ctx, "stage resolved from chunk_status", "stage", st.Name,
+ "chapter", ch.Chapter, "chunk", ch.ChunkIdx, "disposition", cs.Disposition, "reason", cs.FlagReason)
+ return sr, nil
+}
+
+// degradedTag surfaces a flag reason into the request_log `degraded` column.
+func degradedTag(cls classification) string {
+ if cls.ok() {
+ return ""
+ }
+ return string(cls.Reason)
+}
+
+// nonOKTag is the request_log `degraded` value for a resumed disposition row.
+func nonOKTag(disposition, reason string) string {
+ if disposition == string(DispOK) {
+ return ""
+ }
+ return reason
}
diff --git a/backend/internal/pipeline/runner_test.go b/backend/internal/pipeline/runner_test.go
index c0acc7a..163d12e 100644
--- a/backend/internal/pipeline/runner_test.go
+++ b/backend/internal/pipeline/runner_test.go
@@ -1,8 +1,9 @@
package pipeline
import (
- "bytes"
"context"
+ "database/sql"
+ "encoding/json"
"fmt"
"io"
"net/http"
@@ -10,16 +11,18 @@ import (
"os"
"path/filepath"
"strings"
- "sync/atomic"
+ "sync"
"testing"
"time"
"textmachine/backend/internal/obs"
+
+ _ "modernc.org/sqlite"
)
-// e2e-тест мини-раннера: temp-проект книги против httptest-провайдера. Деньги,
-// чекпоинты и resume проверяются через реальные конфиги/store — то, что
-// демонстрирует приёмка Фазы 0, только на моке.
+// e2e-тесты книжного раннера Вехи 2: temp-проект против httptest-провайдера.
+// Деньги, чекпоинты, resume, disposition и exit-коды проверяются через реальные
+// конфиги/store — то, что демонстрирует приёмка, только на моке.
func writeFile(t *testing.T, path, content string) {
t.Helper()
@@ -31,25 +34,89 @@ func writeFile(t *testing.T, path, content string) {
}
}
-// newFakeProvider serves OpenAI-совместимые ответы: draft-стадии отвечает
-// «ЧЕРНОВИК», edit-стадии — «РЕДАКТУРА» (различает по наличию слова ЧЕРНОВИК
-// в промпте редактора). Считает вызовы.
-func newFakeProvider(t *testing.T, calls *atomic.Int32) *httptest.Server {
+// reqRec records every provider request body (thread-safe for -race).
+type reqRec struct {
+ mu sync.Mutex
+ n int
+ bodies []string
+}
+
+func (r *reqRec) record(body string) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ r.n++
+ r.bodies = append(r.bodies, body)
+}
+func (r *reqRec) count() int { r.mu.Lock(); defer r.mu.Unlock(); return r.n }
+func (r *reqRec) all() []string {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return append([]string(nil), r.bodies...)
+}
+
+// newJSONProvider serves an OpenAI-compatible completion whose text+finish come
+// from respond(body); usage is fixed. The default draftEdit distinguishes the
+// editor by the presence of the editor-prompt marker in the body.
+func newJSONProvider(rec *reqRec, respond func(body string) (text, finish string)) *httptest.Server {
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- calls.Add(1)
body, _ := io.ReadAll(r.Body)
- text := "ЧЕРНОВИК ПЕРЕВОДА"
- if bytes.Contains(body, []byte("Черновик перевода для редактуры")) {
- text = "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД"
+ rec.record(string(body))
+ text, finish := respond(string(body))
+ if finish == "" {
+ finish = "stop"
}
- fmt.Fprintf(w, `{"id":"fake-%d","model":"fake-model","choices":[{"message":{"content":"%s"},"finish_reason":"stop"}],
+ tb, _ := json.Marshal(text)
+ fmt.Fprintf(w, `{"id":"fake","model":"fake-model","choices":[{"message":{"content":%s},"finish_reason":%q}],
"usage":{"prompt_tokens":1000,"completion_tokens":500,"prompt_tokens_details":{"cached_tokens":200}}}`,
- calls.Load(), text)
+ tb, finish)
}))
}
-func setupProject(t *testing.T, providerURL string) string {
+func isEditBody(body string) bool {
+ return strings.Contains(body, "Черновик перевода для редактуры")
+}
+
+func draftEdit(body string) (string, string) {
+ if isEditBody(body) {
+ return "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД", "stop"
+ }
+ return "ЧЕРНОВИК ПЕРЕВОДА", "stop"
+}
+
+func maxTokensOf(t *testing.T, body string) int {
t.Helper()
+ var m map[string]any
+ if err := json.Unmarshal([]byte(body), &m); err != nil {
+ t.Fatalf("bad request body: %v", err)
+ }
+ v, ok := m["max_tokens"].(float64)
+ if !ok {
+ t.Fatalf("no numeric max_tokens in body: %s", body)
+ }
+ return int(v)
+}
+
+// перстоимость fake-model на вызов: (1000-200)·1 + 200·0.1 + 500·2 за 1M.
+const fakeCallUSD = (800*1.0 + 200*0.1 + 500*2.0) / 1e6
+
+type projectOpts struct {
+ source string
+ regenerate int
+ minMaxTokens int
+ bookUSD float64
+}
+
+func setupProjectOpts(t *testing.T, providerURL string, o projectOpts) string {
+ t.Helper()
+ if o.source == "" {
+ o.source = "静かな図書館の朝。"
+ }
+ if o.minMaxTokens == 0 {
+ o.minMaxTokens = 512
+ }
+ if o.bookUSD == 0 {
+ o.bookUSD = 1.0
+ }
dir := t.TempDir()
writeFile(t, filepath.Join(dir, "prompts", "translator.md"),
@@ -71,18 +138,18 @@ models:
price: { input_per_m: 1.0, cached_per_m: 0.1, cache_write_per_m: 0, output_per_m: 2.0 }
`, time.Now().UTC().Format("2006-01-02"), providerURL))
- writeFile(t, filepath.Join(dir, "pipeline.yaml"), `
+ writeFile(t, filepath.Join(dir, "pipeline.yaml"), fmt.Sprintf(`
core: C1
version: 1
-defaults: { max_output_ratio: 2.0, min_max_tokens: 512 }
-retries: { regenerate_before_escalate: 1 }
+defaults: { max_output_ratio: 2.0, min_max_tokens: %d }
+retries: { regenerate_before_escalate: %d }
stages:
- { name: draft, role: translator, model: fake-model, prompt: prompts/translator.md, prompt_version: v-test, temperature: 0.3, reasoning: "off" }
- { name: edit, role: editor, model: fake-model, prompt: prompts/editor.md, prompt_version: v-test, temperature: 0.4, reasoning: "off" }
-`)
+`, o.minMaxTokens, o.regenerate))
- writeFile(t, filepath.Join(dir, "source.txt"), "静かな図書館の朝。")
- writeFile(t, filepath.Join(dir, "book.yaml"), `
+ writeFile(t, filepath.Join(dir, "source.txt"), o.source)
+ writeFile(t, filepath.Join(dir, "book.yaml"), fmt.Sprintf(`
book_id: test-book
title: Тест
source_lang: ja
@@ -96,37 +163,50 @@ footnotes: minimal
pipeline: pipeline.yaml
models: models.yaml
source_file: source.txt
-ceilings: { book_usd: 1.0, day_usd: 2.0 }
-`)
+ceilings: { book_usd: %g, day_usd: 2.0 }
+`, o.bookUSD))
return filepath.Join(dir, "book.yaml")
}
+func setupProject(t *testing.T, providerURL string) string {
+ return setupProjectOpts(t, providerURL, projectOpts{regenerate: 1})
+}
+
+func newRunner(t *testing.T, bookPath string) *Runner {
+ t.Helper()
+ r, err := NewRunner(bookPath, obs.NewLogger())
+ if err != nil {
+ t.Fatal(err)
+ }
+ return r
+}
+
+// --- existing Фаза-0/Веха-1 guarantees, ported to the book runner -----------
+
func TestRunnerEndToEndWithResume(t *testing.T) {
- var calls atomic.Int32
- srv := newFakeProvider(t, &calls)
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, draftEdit)
defer srv.Close()
bookPath := setupProject(t, srv.URL)
ctx := obs.WithReqInfo(context.Background(), obs.ReqInfo{TraceID: obs.NewTraceID()})
- // Прогон 1: два вызова (draft, edit), деньги учтены.
- r1, err := NewRunner(bookPath, obs.NewLogger())
+ r1 := newRunner(t, bookPath)
+ res1, err := r1.TranslateBook(ctx)
if err != nil {
t.Fatal(err)
}
- res1, err := r1.TranslateOneChunk(ctx)
- if err != nil {
- t.Fatal(err)
+ if rec.count() != 2 {
+ t.Fatalf("expected 2 provider calls, got %d", rec.count())
}
- if calls.Load() != 2 {
- t.Fatalf("expected 2 provider calls, got %d", calls.Load())
- }
- if len(res1.Stages) != 2 || res1.FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" {
+ if len(res1.Chunks) != 1 || res1.Flagged != 0 || res1.ExitCode() != 0 {
t.Fatalf("run1 = %+v", res1)
}
- // Стоимость по формуле: (1000-200)*1 + 200*0.1 + 500*2 за 1M — на стадию.
- wantStage := (800*1.0 + 200*0.1 + 500*2.0) / 1e6
- if diff := res1.TotalUSD - 2*wantStage; diff > 1e-12 || diff < -1e-12 {
- t.Fatalf("total = %v, want %v", res1.TotalUSD, 2*wantStage)
+ ch := res1.Chunks[0]
+ if ch.Disposition != DispOK || ch.FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" {
+ t.Fatalf("chunk = %+v", ch)
+ }
+ if diff := res1.TotalUSD - 2*fakeCallUSD; diff > 1e-12 || diff < -1e-12 {
+ t.Fatalf("total = %v, want %v", res1.TotalUSD, 2*fakeCallUSD)
}
committed, reserved, err := r1.Store.SpentUSD("test-book")
if err != nil {
@@ -137,121 +217,91 @@ func TestRunnerEndToEndWithResume(t *testing.T) {
}
r1.Close()
- // Прогон 2 (новый процесс — новый Runner над тем же project.db): оба
- // вызова обслуживаются чекпоинтами, провайдер не трогается, доплаты нет.
- r2, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
+ // Прогон 2 (новый процесс): обе стадии из чекпоинтов/chunk_status за $0.
+ r2 := newRunner(t, bookPath)
defer r2.Close()
- res2, err := r2.TranslateOneChunk(ctx)
+ res2, err := r2.TranslateBook(ctx)
if err != nil {
t.Fatal(err)
}
- if calls.Load() != 2 {
- t.Fatalf("resume must not call the provider again, calls=%d", calls.Load())
+ if rec.count() != 2 {
+ t.Fatalf("resume must not call the provider again, calls=%d", rec.count())
}
- for _, st := range res2.Stages {
+ if res2.TotalUSD != 0 {
+ t.Fatalf("resume must be $0, got %v", res2.TotalUSD)
+ }
+ for _, st := range res2.Chunks[0].Stages {
if !st.FromResume || st.CostUSD != 0 {
- t.Fatalf("stage %s must be served from checkpoint at $0: %+v", st.Stage, st)
+ t.Fatalf("stage %s must be served from resume at $0: %+v", st.Stage, st)
}
}
- if res2.FinalText != res1.FinalText {
- t.Fatalf("resume text differs: %q vs %q", res2.FinalText, res1.FinalText)
- }
- committed2, _, err := r2.Store.SpentUSD("test-book")
- if err != nil {
- t.Fatal(err)
+ if res2.Chunks[0].FinalText != res1.Chunks[0].FinalText {
+ t.Fatalf("resume text differs: %q vs %q", res2.Chunks[0].FinalText, res1.Chunks[0].FinalText)
}
+ committed2, _, _ := r2.Store.SpentUSD("test-book")
if committed2 != committed {
t.Fatalf("resume must not add spend: %v -> %v", committed, committed2)
}
}
-// Р6: правка конфига/промпта после старта джоб — не тихая инвалидация
-// чекпоинтов, а громкая ошибка; пере-перевод только явным --resnapshot.
func TestRunnerSnapshotPinning(t *testing.T) {
- var calls atomic.Int32
- srv := newFakeProvider(t, &calls)
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, draftEdit)
defer srv.Close()
bookPath := setupProject(t, srv.URL)
ctx := context.Background()
- r1, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
- if _, err := r1.TranslateOneChunk(ctx); err != nil {
+ r1 := newRunner(t, bookPath)
+ if _, err := r1.TranslateBook(ctx); err != nil {
t.Fatal(err)
}
r1.Close()
- if calls.Load() != 2 {
- t.Fatalf("run1 calls = %d", calls.Load())
+ if rec.count() != 2 {
+ t.Fatalf("run1 calls = %d", rec.count())
}
- // Меняем промпт переводчика → новый snapshot.
promptPath := filepath.Join(filepath.Dir(bookPath), "prompts", "translator.md")
writeFile(t, promptPath, "НОВЫЙ промпт с {{source_lang}}.\n---USER---\n{{text}}")
- r2, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
- _, err = r2.TranslateOneChunk(ctx)
+ r2 := newRunner(t, bookPath)
+ _, err := r2.TranslateBook(ctx)
r2.Close()
if err == nil || !strings.Contains(err.Error(), "resnapshot") {
t.Fatalf("changed config must fail loud mentioning --resnapshot, got: %v", err)
}
- if calls.Load() != 2 {
- t.Fatalf("denied resume must not call the provider, calls=%d", calls.Load())
+ if rec.count() != 2 {
+ t.Fatalf("denied resume must not call the provider, calls=%d", rec.count())
}
- // Явное согласие: джобы перепривязываются, вызовы повторяются и оплачиваются.
- r3, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
+ r3 := newRunner(t, bookPath)
defer r3.Close()
r3.Resnapshot = true
- res, err := r3.TranslateOneChunk(ctx)
+ res, err := r3.TranslateBook(ctx)
if err != nil {
t.Fatal(err)
}
- if calls.Load() != 4 {
- t.Fatalf("resnapshot run must re-call both stages, calls=%d", calls.Load())
+ if rec.count() != 4 {
+ t.Fatalf("resnapshot run must re-call both stages, calls=%d", rec.count())
}
if res.TotalUSD <= 0 {
t.Fatal("re-translation must be billed")
}
}
-// D5.2/B: правка ТОЛЬКО каппы модели (wire-форма) обязана сдвинуть snapshot и
-// уронить resume громко — иначе изменённое тело запроса подалось бы из старого
-// чекпоинта (тихий расходящийся ре-пэй). Гейт на то, что stageSnap реально
-// сворачивает резолвнутую Capability: без `ss.Capability` этот тест бы прошёл
-// resume молча (находка селфревью вехи 1).
+// D5.2/B: правка ТОЛЬКО каппы модели двигает snapshot и роняет resume громко.
func TestRunnerSnapshotPinsCapability(t *testing.T) {
- var calls atomic.Int32
- srv := newFakeProvider(t, &calls)
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, draftEdit)
defer srv.Close()
bookPath := setupProject(t, srv.URL)
ctx := context.Background()
- r1, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
- if _, err := r1.TranslateOneChunk(ctx); err != nil {
+ r1 := newRunner(t, bookPath)
+ if _, err := r1.TranslateBook(ctx); err != nil {
t.Fatal(err)
}
r1.Close()
- if calls.Load() != 2 {
- t.Fatalf("run1 calls = %d", calls.Load())
- }
- // Меняем ТОЛЬКО каппу fake-model (temperature force) — промпт/сэмплинг/
- // провайдер/цены те же. Резолвнутая Capability меняется → snapshotID обязан
- // сдвинуться → resume падает на snapshot-pinning, а не подаёт чекпоинт.
modelsPath := filepath.Join(filepath.Dir(bookPath), "models.yaml")
raw, err := os.ReadFile(modelsPath)
if err != nil {
@@ -265,22 +315,17 @@ func TestRunnerSnapshotPinsCapability(t *testing.T) {
}
writeFile(t, modelsPath, patched)
- r2, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
- _, err = r2.TranslateOneChunk(ctx)
+ r2 := newRunner(t, bookPath)
+ _, err = r2.TranslateBook(ctx)
r2.Close()
if err == nil || !strings.Contains(err.Error(), "resnapshot") {
- t.Fatalf("a capability edit must fail loud mentioning --resnapshot (snapshot must fold the resolved capability), got: %v", err)
+ t.Fatalf("a capability edit must fail loud mentioning --resnapshot, got: %v", err)
}
- if calls.Load() != 2 {
- t.Fatalf("denied resume must not call the provider, calls=%d", calls.Load())
+ if rec.count() != 2 {
+ t.Fatalf("denied resume must not call the provider, calls=%d", rec.count())
}
}
-// Платный 2xx с нулевым usage не должен селтлиться в $0 (иначе потолок слепнет):
-// берётся консервативная оценка резерва.
func TestRunnerZeroUsagePaidSettlesEstimate(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fmt.Fprint(w, `{"id":"z","model":"deepseek-v4-flash","choices":[{"message":{"content":"перевод"},"finish_reason":"stop"}],
@@ -289,36 +334,27 @@ func TestRunnerZeroUsagePaidSettlesEstimate(t *testing.T) {
defer srv.Close()
bookPath := setupProject(t, srv.URL)
- r, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
+ r := newRunner(t, bookPath)
defer r.Close()
- res, err := r.TranslateOneChunk(context.Background())
+ res, err := r.TranslateBook(context.Background())
if err != nil {
t.Fatal(err)
}
if res.TotalUSD <= 0 {
t.Fatalf("zero-usage paid 2xx must settle a non-zero estimate, got $%.6f", res.TotalUSD)
}
- committed, _, err := r.Store.SpentUSD("test-book")
- if err != nil {
- t.Fatal(err)
- }
+ committed, _, _ := r.Store.SpentUSD("test-book")
if committed != res.TotalUSD {
t.Fatalf("committed %v != total %v", committed, res.TotalUSD)
}
}
-// NewRunner обязан отклонить конфиг с нереализованной механикой ДО открытия
-// store и любых вызовов (guard границы «конфиг vs код»).
func TestRunnerRejectsUnrunnableConfig(t *testing.T) {
- var calls atomic.Int32
- srv := newFakeProvider(t, &calls)
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, draftEdit)
defer srv.Close()
bookPath := setupProject(t, srv.URL)
- // Ломаем pipeline на fanout>1 (механика Фазы 2).
pipePath := filepath.Join(filepath.Dir(bookPath), "pipeline.yaml")
raw, err := os.ReadFile(pipePath)
if err != nil {
@@ -329,35 +365,449 @@ func TestRunnerRejectsUnrunnableConfig(t *testing.T) {
if _, err := NewRunner(bookPath, obs.NewLogger()); err == nil {
t.Fatal("fanout.candidates>1 must be rejected by NewRunner (Phase-2 mechanics)")
}
- if calls.Load() != 0 {
- t.Fatalf("rejected config must not reach the provider, calls=%d", calls.Load())
+ if rec.count() != 0 {
+ t.Fatalf("rejected config must not reach the provider, calls=%d", rec.count())
}
}
func TestRunnerCeilingDenies(t *testing.T) {
- var calls atomic.Int32
- srv := newFakeProvider(t, &calls)
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, draftEdit)
defer srv.Close()
- bookPath := setupProject(t, srv.URL)
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{bookUSD: 0.0000001})
- // Потолок $0: первый же reserve обязан отказать ДО вызова провайдера.
- raw, err := os.ReadFile(bookPath)
- if err != nil {
- t.Fatal(err)
- }
- patched := strings.Replace(string(raw), "book_usd: 1.0", "book_usd: 0.0000001", 1)
- writeFile(t, bookPath, patched)
-
- r, err := NewRunner(bookPath, obs.NewLogger())
- if err != nil {
- t.Fatal(err)
- }
+ r := newRunner(t, bookPath)
defer r.Close()
- _, err = r.TranslateOneChunk(context.Background())
+ _, err := r.TranslateBook(context.Background())
if err == nil {
t.Fatal("ceiling must deny the run")
}
- if calls.Load() != 0 {
- t.Fatalf("denied reserve must not reach the provider, calls=%d", calls.Load())
+ if rec.count() != 0 {
+ t.Fatalf("denied reserve must not reach the provider, calls=%d", rec.count())
+ }
+}
+
+// --- Веха 2: chunk loop, disposition, resume, exit codes --------------------
+
+// Multi-chunk (two chapters via form feed) end-to-end + resume at $0.
+func TestRunnerMultiChunkResume(t *testing.T) {
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, draftEdit)
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: "ГЛАВАОДИН\fГЛАВАДВА"})
+ ctx := context.Background()
+
+ r1 := newRunner(t, bookPath)
+ res1, err := r1.TranslateBook(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r1.Close()
+ if len(res1.Chunks) != 2 {
+ t.Fatalf("want 2 chunks (2 chapters), got %d", len(res1.Chunks))
+ }
+ if res1.Chunks[0].Chapter != 1 || res1.Chunks[1].Chapter != 2 {
+ t.Fatalf("chapters = %d,%d", res1.Chunks[0].Chapter, res1.Chunks[1].Chapter)
+ }
+ if rec.count() != 4 { // 2 chunks × (draft+edit)
+ t.Fatalf("want 4 calls, got %d", rec.count())
+ }
+
+ r2 := newRunner(t, bookPath)
+ defer r2.Close()
+ res2, err := r2.TranslateBook(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec.count() != 4 {
+ t.Fatalf("resume must not re-call, calls=%d", rec.count())
+ }
+ if res2.TotalUSD != 0 || res2.Flagged != 0 || res2.ExitCode() != 0 {
+ t.Fatalf("resume run = %+v", res2)
+ }
+}
+
+// D2 core: a flagged chunk skips its remaining stages and the loop continues to
+// the next chunk; the run completes with exit code 2, and resume does not
+// re-attack the flagged chunk.
+func TestRunnerFlagSkipContinueExit2(t *testing.T) {
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, func(body string) (string, string) {
+ if strings.Contains(body, "ОТКАЗНАЯГЛАВА") && !isEditBody(body) {
+ return "Извините, я не могу перевести это.", "stop" // soft refusal
+ }
+ return draftEdit(body)
+ })
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: "ОТКАЗНАЯГЛАВА\fНОРМАЛЬНАЯГЛАВА"})
+ ctx := context.Background()
+
+ r1 := newRunner(t, bookPath)
+ res1, err := r1.TranslateBook(ctx)
+ if err != nil {
+ t.Fatalf("a flagged chunk must NOT be an error (loop continues), got %v", err)
+ }
+ // draft(ch1)=refusal → edit(ch1) skipped (no call); draft(ch2)+edit(ch2) ok = 3 calls.
+ if rec.count() != 3 {
+ t.Fatalf("want 3 calls (edit of flagged chunk skipped), got %d", rec.count())
+ }
+ if res1.Flagged != 1 || res1.ExitCode() != 2 {
+ t.Fatalf("want 1 flagged chunk / exit 2, got flagged=%d exit=%d", res1.Flagged, res1.ExitCode())
+ }
+ c1 := res1.Chunks[0]
+ if c1.Disposition != DispFlagged || c1.FlagReason != FlagSoftRefusal || c1.FinalText != "" {
+ t.Fatalf("chunk1 = %+v (want flagged/soft_refusal/no text)", c1)
+ }
+ if c1.Stages[1].Disposition != DispSkipped {
+ t.Fatalf("edit of a flagged chunk must be skipped, got %+v", c1.Stages[1])
+ }
+ if res1.Chunks[1].Disposition != DispOK {
+ t.Fatalf("chunk2 must be ok, got %+v", res1.Chunks[1])
+ }
+ // chunk_status persisted the flag and the skip.
+ cs, _ := r1.Store.GetChunkStatus("test-book", 1, 0, "draft")
+ if cs == nil || cs.Disposition != "flagged" || cs.FlagReason != "soft_refusal" {
+ t.Fatalf("draft chunk_status = %+v", cs)
+ }
+ skip, _ := r1.Store.GetChunkStatus("test-book", 1, 0, "edit")
+ if skip == nil || skip.Disposition != "skipped" {
+ t.Fatalf("edit chunk_status = %+v", skip)
+ }
+ r1.Close()
+
+ // Resume: the flagged chunk is resolved from chunk_status, NOT re-attacked.
+ r2 := newRunner(t, bookPath)
+ defer r2.Close()
+ res2, err := r2.TranslateBook(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec.count() != 3 {
+ t.Fatalf("resume must not re-attack a flagged chunk, calls=%d", rec.count())
+ }
+ if res2.Flagged != 1 || res2.TotalUSD != 0 || res2.ExitCode() != 2 {
+ t.Fatalf("resume run = %+v", res2)
+ }
+}
+
+// F4: a non-empty truncated (finish=length) draft must be FLAGGED, never passed
+// downstream to the editor as OK. With no regenerations it flags immediately.
+func TestRunnerF4TruncatedLengthNotPassedToEdit(t *testing.T) {
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, func(body string) (string, string) {
+ if !isEditBody(body) {
+ return "Обрыв на середине предложения", "length" // truncated, non-empty, not a loop
+ }
+ return "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД", "stop"
+ })
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{regenerate: 0})
+ ctx := context.Background()
+
+ r := newRunner(t, bookPath)
+ defer r.Close()
+ res, err := r.TranslateBook(ctx)
+ if err != nil {
+ t.Fatalf("a truncated draft must flag, not crash: %v", err)
+ }
+ if rec.count() != 1 {
+ t.Fatalf("edit must NOT run on a flagged draft, calls=%d", rec.count())
+ }
+ ch := res.Chunks[0]
+ if ch.Disposition != DispFlagged || ch.FlagReason != FlagLength {
+ t.Fatalf("chunk must be flagged length, got %+v", ch)
+ }
+ if ch.Stages[0].Disposition != DispFlagged || ch.Stages[1].Disposition != DispSkipped {
+ t.Fatalf("draft flagged + edit skipped expected, got %+v", ch.Stages)
+ }
+}
+
+// D2.3 + F3 + attempt-axis: a length cut is retried with a DOUBLED budget; the
+// second attempt succeeds, the editor then runs, and cost sums both attempts.
+func TestRunnerLengthRetrySucceedsDoublesBudget(t *testing.T) {
+ rec := &reqRec{}
+ src := strings.Repeat("字", 500) // EstimateTokens=500 → base max_tokens=1000
+ srv := newJSONProvider(rec, func(body string) (string, string) {
+ if isEditBody(body) {
+ return "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД", "stop"
+ }
+ if maxTokensOf(t, body) <= 1000 { // attempt 0 (base)
+ return "Обрыв на середине", "length"
+ }
+ return "ПОЛНЫЙ ЧЕРНОВИК", "stop" // attempt 1 (doubled budget)
+ })
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: src, regenerate: 1, minMaxTokens: 128})
+ ctx := context.Background()
+
+ r := newRunner(t, bookPath)
+ defer r.Close()
+ res, err := r.TranslateBook(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ch := res.Chunks[0]
+ if ch.Disposition != DispOK || ch.FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" {
+ t.Fatalf("retry should succeed and edit should run, got %+v", ch)
+ }
+ draft := ch.Stages[0]
+ if draft.Attempts != 2 {
+ t.Fatalf("draft must have taken 2 attempts, got %d", draft.Attempts)
+ }
+ // F3: cost sums BOTH attempts.
+ if diff := draft.CumCostUSD - 2*fakeCallUSD; diff > 1e-12 || diff < -1e-12 {
+ t.Fatalf("draft cum cost = %v, want 2×call %v (attempts summed)", draft.CumCostUSD, 2*fakeCallUSD)
+ }
+ // The two draft attempts doubled the budget: 1000 then 2000.
+ var draftToks []int
+ for _, b := range rec.all() {
+ if !isEditBody(b) {
+ draftToks = append(draftToks, maxTokensOf(t, b))
+ }
+ }
+ if len(draftToks) != 2 || draftToks[0] != 1000 || draftToks[1] != 2000 {
+ t.Fatalf("attempt budgets = %v, want [1000 2000]", draftToks)
+ }
+}
+
+// D2.5: the editor's max_tokens is sized from the (long Russian) DRAFT, not the
+// (short CJK) source — otherwise it under-budgets and false-triggers a length
+// retry. Broken D2.5 would make the two budgets equal.
+func TestRunnerEditSizedFromDraft(t *testing.T) {
+ rec := &reqRec{}
+ longDraft := strings.Repeat("Утро было тихим и ясным над рекой. ", 120) // long Russian output
+ srv := newJSONProvider(rec, func(body string) (string, string) {
+ if isEditBody(body) {
+ return "ОК", "stop"
+ }
+ return longDraft, "stop"
+ })
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: strings.Repeat("字", 100), minMaxTokens: 128})
+
+ r := newRunner(t, bookPath)
+ defer r.Close()
+ if _, err := r.TranslateBook(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ var draftTok, editTok int
+ for _, b := range rec.all() {
+ if isEditBody(b) {
+ editTok = maxTokensOf(t, b)
+ } else {
+ draftTok = maxTokensOf(t, b)
+ }
+ }
+ if editTok <= draftTok {
+ t.Fatalf("edit max_tokens (%d) must exceed draft's (%d) — sized from the longer draft (D2.5)", editTok, draftTok)
+ }
+}
+
+// Anti-wedge #2: an empty completion is FLAGGED and the loop continues — not a
+// crash (the Фаза-0 behaviour was to error out the whole run).
+func TestRunnerEmptyFlaggedNotCrash(t *testing.T) {
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, func(body string) (string, string) {
+ if !isEditBody(body) {
+ return "", "stop" // empty draft
+ }
+ return "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД", "stop"
+ })
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{regenerate: 0})
+
+ r := newRunner(t, bookPath)
+ defer r.Close()
+ res, err := r.TranslateBook(context.Background())
+ if err != nil {
+ t.Fatalf("empty completion must flag, not crash: %v", err)
+ }
+ ch := res.Chunks[0]
+ if ch.Disposition != DispFlagged || ch.FlagReason != FlagEmpty {
+ t.Fatalf("empty draft must be flagged empty, got %+v", ch)
+ }
+ if res.ExitCode() != 2 {
+ t.Fatalf("exit code must be 2, got %d", res.ExitCode())
+ }
+ // Money for the paid-but-empty call is still accounted.
+ committed, _, _ := r.Store.SpentUSD("test-book")
+ if committed <= 0 {
+ t.Fatalf("the paid empty call must still be billed, committed=%v", committed)
+ }
+}
+
+// Anti-wedge #2: a billed 2xx with an unreadable body → flagged decode_error
+// with the estimate settled, loop continues (not a crash).
+func TestRunnerDecodeErrorFlagged(t *testing.T) {
+ var draftCalls int
+ var mu sync.Mutex
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ body, _ := io.ReadAll(r.Body)
+ if !isEditBody(string(body)) {
+ mu.Lock()
+ draftCalls++
+ mu.Unlock()
+ fmt.Fprint(w, `NOT JSON AT ALL`) // 200 with a garbage body → BilledDecodeError
+ return
+ }
+ fmt.Fprint(w, `{"id":"e","model":"fake-model","choices":[{"message":{"content":"ред"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`)
+ }))
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{regenerate: 0})
+
+ r := newRunner(t, bookPath)
+ defer r.Close()
+ res, err := r.TranslateBook(context.Background())
+ if err != nil {
+ t.Fatalf("a billed decode failure must flag, not crash: %v", err)
+ }
+ ch := res.Chunks[0]
+ if ch.Disposition != DispFlagged || ch.FlagReason != FlagDecodeError {
+ t.Fatalf("draft must be flagged decode_error, got %+v", ch)
+ }
+ if ch.Stages[1].Disposition != DispSkipped {
+ t.Fatalf("edit must be skipped after a decode flag, got %+v", ch.Stages[1])
+ }
+ if res.TotalUSD <= 0 {
+ t.Fatal("the conservative estimate must be settled for a billed decode failure")
+ }
+}
+
+// Determinism content-guard (self-review Веха 2): the source is NOT in the
+// snapshot, so editing a chunk's source between runs must RE-TRANSLATE that
+// position (content-addressed, §3.4), never serve the stale positional
+// chunk_status row. Without the content_hash guard the fast-path would return
+// the OLD translation at $0.
+func TestRunnerSourceEditReTranslates(t *testing.T) {
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, draftEdit)
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{source: "СТАРЫЙИСХОДНИК"})
+ ctx := context.Background()
+
+ r1 := newRunner(t, bookPath)
+ if _, err := r1.TranslateBook(ctx); err != nil {
+ t.Fatal(err)
+ }
+ r1.Close()
+ callsAfterRun1 := rec.count()
+
+ // Edit the source in place (same position, still one chunk). No brief field
+ // and not the chunker changed, so the snapshot is unchanged — the ONLY thing
+ // that must invalidate resume is the content itself.
+ writeFile(t, filepath.Join(filepath.Dir(bookPath), "source.txt"), "НОВЫЙИСХОДНИК")
+
+ r2 := newRunner(t, bookPath)
+ defer r2.Close()
+ res, err := r2.TranslateBook(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Both stages re-render with the new content (the editor sees {{text}} too),
+ // miss their old checkpoints, and are re-translated and re-billed.
+ if rec.count() != callsAfterRun1+2 {
+ t.Fatalf("an edited source must re-translate the chunk, not serve stale: calls %d -> %d", callsAfterRun1, rec.count())
+ }
+ if res.TotalUSD <= 0 {
+ t.Fatalf("re-translation of an edited source must be billed, got $%.6f", res.TotalUSD)
+ }
+}
+
+// Anti-wedge #1: a terminally-flagged chunk is resolved from chunk_status BEFORE
+// rendering, so it is NOT re-attacked even when the retry budget is later raised
+// (RegenerateBeforeEscalate is not in the snapshot). Without the chunk_status
+// fast-path the attempt loop would spawn NEW paid attempts up to the new cap.
+func TestRunnerFlaggedNotReattackedWhenRegenRaised(t *testing.T) {
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, func(body string) (string, string) {
+ if !isEditBody(body) {
+ return "Обрыв на середине", "length" // always a length cut
+ }
+ return "ред", "stop"
+ })
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{regenerate: 0})
+ ctx := context.Background()
+
+ r1 := newRunner(t, bookPath)
+ if _, err := r1.TranslateBook(ctx); err != nil {
+ t.Fatal(err)
+ }
+ r1.Close()
+ callsAfterRun1 := rec.count() // regen=0 → exactly 1 draft attempt, flagged length
+
+ // Raise the regeneration budget and re-run. The flagged chunk must stay put.
+ 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), "regenerate_before_escalate: 0", "regenerate_before_escalate: 2", 1))
+
+ r2 := newRunner(t, bookPath)
+ defer r2.Close()
+ res, err := r2.TranslateBook(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec.count() != callsAfterRun1 {
+ t.Fatalf("a terminally-flagged chunk must NOT be re-attacked on a raised budget: calls %d -> %d", callsAfterRun1, rec.count())
+ }
+ if res.Chunks[0].Disposition != DispFlagged || res.Chunks[0].FlagReason != FlagLength {
+ t.Fatalf("chunk must stay flagged length, got %+v", res.Chunks[0])
+ }
+}
+
+// Anti-wedge #3: a checkpoint present without its chunk_status row (kill -9 lost
+// the resolve) self-heals — classify the checkpoint text, no re-billing, no
+// crash. Proven by deleting the chunk_status row and re-running with regen=0 so
+// the empty flag cannot spawn a new paid attempt.
+func TestRunnerLegacyEmptyCheckpointSelfHeal(t *testing.T) {
+ rec := &reqRec{}
+ srv := newJSONProvider(rec, func(body string) (string, string) {
+ if !isEditBody(body) {
+ return "", "stop" // empty draft → empty checkpoint
+ }
+ return "ред", "stop"
+ })
+ defer srv.Close()
+ bookPath := setupProjectOpts(t, srv.URL, projectOpts{regenerate: 0})
+ ctx := context.Background()
+
+ r1 := newRunner(t, bookPath)
+ if _, err := r1.TranslateBook(ctx); err != nil {
+ t.Fatal(err)
+ }
+ dbPath := r1.Book.ProjectDB
+ r1.Close()
+ callsAfterRun1 := rec.count()
+
+ // Simulate the resolve being lost while the (empty) checkpoint survives.
+ db, err := sql.Open("sqlite", "file:"+dbPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Exec(`DELETE FROM chunk_status WHERE stage = 'draft'`); err != nil {
+ t.Fatal(err)
+ }
+ db.Close()
+
+ r2 := newRunner(t, bookPath)
+ defer r2.Close()
+ res, err := r2.TranslateBook(ctx)
+ if err != nil {
+ t.Fatalf("legacy empty checkpoint must self-heal, not crash: %v", err)
+ }
+ if rec.count() != callsAfterRun1 {
+ t.Fatalf("self-heal must NOT re-bill the empty attempt: calls %d -> %d", callsAfterRun1, rec.count())
+ }
+ ch := res.Chunks[0]
+ if ch.Disposition != DispFlagged || ch.FlagReason != FlagEmpty {
+ t.Fatalf("self-heal must re-derive the empty flag, got %+v", ch)
+ }
+ // chunk_status was rebuilt.
+ cs, _ := r2.Store.GetChunkStatus("test-book", 1, 0, "draft")
+ if cs == nil || cs.Disposition != "flagged" {
+ t.Fatalf("chunk_status must be rebuilt as flagged, got %+v", cs)
}
}
diff --git a/backend/internal/store/chunkstatus.go b/backend/internal/store/chunkstatus.go
new file mode 100644
index 0000000..56b83fc
--- /dev/null
+++ b/backend/internal/store/chunkstatus.go
@@ -0,0 +1,106 @@
+package store
+
+import (
+ "database/sql"
+ "errors"
+)
+
+// chunkstatus.go: the per-chunk×stage disposition record (Веха 2, D2). It is a
+// materialized RESOLVE over the append-only checkpoints (keyed by request_hash,
+// which already carries `attempt`), NOT a column on checkpoints. Because it is
+// derivable from the checkpoints, losing it to a kill -9 self-heals: on resume
+// the runner re-classifies the checkpoint text (no re-billing) and re-upserts
+// the row. `disposition` is the axis the runner loop reads BEFORE rendering, so
+// a terminally-flagged chunk is never re-attacked into an infinite paid loop.
+//
+// The store is dumb storage here: disposition/flag_reason are plain strings; the
+// typed FlagReason/Disposition constants and the classifier live in the pipeline
+// package.
+
+// ChunkStatus is one (book, chapter, chunk, stage) disposition row.
+type ChunkStatus struct {
+ BookID string
+ Chapter int
+ ChunkIdx int
+ Stage string
+ SnapshotID string
+ ContentHash string // signature of the rendered msgs (the source is NOT in the snapshot); guards the resume fast-path against serving a stale translation after a source edit
+ Disposition string // ok | flagged | skipped
+ FlagReason string // "" when ok
+ Attempts int // number of attempts made for this chunk×stage
+ FinalHash string // request_hash of the authoritative checkpoint (ok path)
+ CostUSD float64 // sum across all attempts (F3-honest)
+ Detail string
+}
+
+// UpsertChunkStatus writes (or overwrites) the disposition row. Overwrite is the
+// resolve semantics: re-running classify over the same checkpoints reproduces the
+// same row, and a --resnapshot re-processing legitimately replaces a stale row.
+func (s *Store) UpsertChunkStatus(cs ChunkStatus) error {
+ ctx, cancel := opContext()
+ defer cancel()
+ _, err := s.w.ExecContext(ctx, `
+ INSERT INTO chunk_status (
+ book_id, chapter, chunk_idx, stage, snapshot_id, content_hash,
+ disposition, flag_reason, attempts, final_hash, cost_usd, detail, updated_at
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
+ ON CONFLICT (book_id, chapter, chunk_idx, stage) DO UPDATE SET
+ snapshot_id = excluded.snapshot_id,
+ content_hash = excluded.content_hash,
+ disposition = excluded.disposition,
+ flag_reason = excluded.flag_reason,
+ attempts = excluded.attempts,
+ final_hash = excluded.final_hash,
+ cost_usd = excluded.cost_usd,
+ detail = excluded.detail,
+ updated_at = excluded.updated_at`,
+ cs.BookID, cs.Chapter, cs.ChunkIdx, cs.Stage, cs.SnapshotID, cs.ContentHash,
+ cs.Disposition, cs.FlagReason, cs.Attempts, cs.FinalHash, cs.CostUSD, cs.Detail)
+ return err
+}
+
+// GetChunkStatus returns the disposition row for a chunk×stage, or nil if none.
+func (s *Store) GetChunkStatus(bookID string, chapter, chunkIdx int, stage string) (*ChunkStatus, error) {
+ ctx, cancel := opContext()
+ defer cancel()
+ cs := ChunkStatus{BookID: bookID, Chapter: chapter, ChunkIdx: chunkIdx, Stage: stage}
+ err := s.r.QueryRowContext(ctx, `
+ SELECT snapshot_id, content_hash, disposition, flag_reason, attempts, final_hash, cost_usd, detail
+ FROM chunk_status
+ WHERE book_id = ? AND chapter = ? AND chunk_idx = ? AND stage = ?`,
+ bookID, chapter, chunkIdx, stage).Scan(
+ &cs.SnapshotID, &cs.ContentHash, &cs.Disposition, &cs.FlagReason, &cs.Attempts, &cs.FinalHash, &cs.CostUSD, &cs.Detail)
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &cs, nil
+}
+
+// ChunkStatusesForBook returns every disposition row for a book, ordered for a
+// stable report (tmctl report: the flag section).
+func (s *Store) ChunkStatusesForBook(bookID string) ([]ChunkStatus, error) {
+ ctx, cancel := opContext()
+ defer cancel()
+ rows, err := s.r.QueryContext(ctx, `
+ SELECT chapter, chunk_idx, stage, snapshot_id, content_hash, disposition, flag_reason,
+ attempts, final_hash, cost_usd, detail
+ FROM chunk_status WHERE book_id = ?
+ ORDER BY chapter, chunk_idx, stage`, bookID)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []ChunkStatus
+ for rows.Next() {
+ cs := ChunkStatus{BookID: bookID}
+ if err := rows.Scan(&cs.Chapter, &cs.ChunkIdx, &cs.Stage, &cs.SnapshotID, &cs.ContentHash,
+ &cs.Disposition, &cs.FlagReason, &cs.Attempts, &cs.FinalHash, &cs.CostUSD, &cs.Detail); err != nil {
+ return nil, err
+ }
+ out = append(out, cs)
+ }
+ return out, rows.Err()
+}
diff --git a/backend/internal/store/chunkstatus_test.go b/backend/internal/store/chunkstatus_test.go
new file mode 100644
index 0000000..50f264a
--- /dev/null
+++ b/backend/internal/store/chunkstatus_test.go
@@ -0,0 +1,63 @@
+package store
+
+import "testing"
+
+func TestChunkStatusUpsertGetOverwrite(t *testing.T) {
+ s, _ := openTemp(t)
+
+ if got, err := s.GetChunkStatus("book", 1, 0, "draft"); err != nil || got != nil {
+ t.Fatalf("empty store must return (nil,nil), got %+v err=%v", got, err)
+ }
+
+ cs := ChunkStatus{
+ BookID: "book", Chapter: 1, ChunkIdx: 0, Stage: "draft", SnapshotID: "snap1",
+ Disposition: "flagged", FlagReason: "length", Attempts: 2, FinalHash: "",
+ CostUSD: 0.0034, Detail: "truncated at max_tokens",
+ }
+ if err := s.UpsertChunkStatus(cs); err != nil {
+ t.Fatal(err)
+ }
+ got, err := s.GetChunkStatus("book", 1, 0, "draft")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got == nil || got.Disposition != "flagged" || got.FlagReason != "length" || got.Attempts != 2 || got.CostUSD != 0.0034 {
+ t.Fatalf("round-trip mismatch: %+v", got)
+ }
+
+ // Overwrite (the resolve semantics): a later verdict replaces the row.
+ cs.Disposition, cs.FlagReason, cs.FinalHash, cs.CostUSD = "ok", "", "hashABC", 0.0068
+ if err := s.UpsertChunkStatus(cs); err != nil {
+ t.Fatal(err)
+ }
+ got, _ = s.GetChunkStatus("book", 1, 0, "draft")
+ if got.Disposition != "ok" || got.FlagReason != "" || got.FinalHash != "hashABC" || got.CostUSD != 0.0068 {
+ t.Fatalf("overwrite failed: %+v", got)
+ }
+}
+
+func TestChunkStatusesForBookOrdered(t *testing.T) {
+ s, _ := openTemp(t)
+ rows := []ChunkStatus{
+ {BookID: "b", Chapter: 2, ChunkIdx: 0, Stage: "draft", SnapshotID: "s", Disposition: "ok"},
+ {BookID: "b", Chapter: 1, ChunkIdx: 1, Stage: "edit", SnapshotID: "s", Disposition: "skipped"},
+ {BookID: "b", Chapter: 1, ChunkIdx: 0, Stage: "draft", SnapshotID: "s", Disposition: "flagged", FlagReason: "soft_refusal"},
+ {BookID: "other", Chapter: 1, ChunkIdx: 0, Stage: "draft", SnapshotID: "s", Disposition: "ok"},
+ }
+ for _, r := range rows {
+ if err := s.UpsertChunkStatus(r); err != nil {
+ t.Fatal(err)
+ }
+ }
+ got, err := s.ChunkStatusesForBook("b")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(got) != 3 {
+ t.Fatalf("want 3 rows for book b (other filtered), got %d", len(got))
+ }
+ // Ordered by (chapter, chunk_idx, stage).
+ if got[0].Chapter != 1 || got[0].ChunkIdx != 0 || got[1].ChunkIdx != 1 || got[2].Chapter != 2 {
+ t.Fatalf("ordering wrong: %+v", got)
+ }
+}
diff --git a/backend/internal/store/migrate.go b/backend/internal/store/migrate.go
index f26a37e..d7acc6d 100644
--- a/backend/internal/store/migrate.go
+++ b/backend/internal/store/migrate.go
@@ -101,6 +101,31 @@ var migrations = []string{
);
CREATE INDEX IF NOT EXISTS request_log_ts_idx ON request_log (ts);
`,
+ // v2: per-chunk×stage disposition (Веха 2, D2). Резолв ПОВЕРХ append-only
+ // checkpoints (keyed by request_hash-с-attempt) — не колонка на checkpoints:
+ // на resume строка пересобирается из чекпоинтов, поэтому её потеря на kill -9
+ // самолечится (classify по тексту чекпоинта, без ре-биллинга). jobs.status
+ // контентом НЕ расширяется — failed остаётся инфра-сбоем; плохой чанк живёт
+ // здесь как disposition, а цикл продолжается.
+ `
+ CREATE TABLE IF NOT EXISTS chunk_status (
+ book_id TEXT NOT NULL,
+ chapter INTEGER NOT NULL,
+ chunk_idx INTEGER NOT NULL,
+ stage TEXT NOT NULL,
+ snapshot_id TEXT NOT NULL, -- под каким снапшотом резолвнут (stale после --resnapshot игнорируется)
+ content_hash TEXT NOT NULL DEFAULT '',-- сигнатура отрендеренных msgs (исходник НЕ в снапшоте): правка src делает позиционную строку недействительной, а не подаёт устаревший перевод
+ disposition TEXT NOT NULL, -- ok | flagged | skipped
+ flag_reason TEXT NOT NULL DEFAULT '', -- '' когда ok; Go-константа FlagReason
+ attempts INTEGER NOT NULL DEFAULT 0,
+ final_hash TEXT NOT NULL DEFAULT '', -- request_hash авторитетного чекпоинта (ok-путь подаёт его текст дальше)
+ cost_usd REAL NOT NULL DEFAULT 0, -- сумма по ВСЕМ попыткам (F3-честно: ретраи учтены)
+ detail TEXT NOT NULL DEFAULT '',
+ updated_at TEXT NOT NULL DEFAULT (datetime('now')),
+ PRIMARY KEY (book_id, chapter, chunk_idx, stage)
+ );
+ CREATE INDEX IF NOT EXISTS chunk_status_book_idx ON chunk_status (book_id, disposition);
+ `,
}
// migrate runs all pending migrations on the write pool, one transaction per
diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md
index a49c091..e106e98 100644
--- a/docs/PROGRESS.md
+++ b/docs/PROGRESS.md
@@ -257,6 +257,20 @@ keep-alive (Ф1–2), инъекция глоссария (`selective`), пор
**Веха 2 (следующая):** runner chapter/chunk-loop + `chunk_status`/disposition (A, D2) — `flag_reason`-константы, `classify(text,finish)` (refusal/echo до length), ось attempt, exit-коды 0/2/1, фиксы F4 (усечённый length → flagged) + edit-`max_tokens` от длины черновика.
+### 2026-07-04 — Сессия 4: Веха 2 Фазы 1 — chunk-loop + disposition (A, D2) + фиксы F4/D2.5
+
+Снят хардкод «один чанк»: прогон идёт по всей книге, плохой кусок помечается и цикл продолжается. Всё зелёное: `go build/vet/test -race`, `tmctl report` ($0). Живые LLM не звались (моки + tmctl $0, как предписано).
+
+**Цикл по чанкам (шов под шаг 3).** `TranslateOneChunk`(chapter=1/chunk=0 хардкод) → `TranslateBook` → `translateChunk` (цикл стадий) → `runStage` → `runAttempt` (ось attempt). `RunResult`→`ChunkOutcome` (per-chunk) + `BookResult` (агрегат книги). Чанкер — **минимальный детерминированный плейсхолдер** (`chunker.go` `SplitChunks`: разбивка на главы по form-feed, паковка абзацев ≤1500 симв., абзац не режется); настоящий лингвистический чанкер — шаг 3, подменит `SplitChunks` не трогая цикл/disposition. `chunkerVersion` бампнут `v1-wholefile`→`v2-ff-para-pack` (в snapshot; ре-чанкинг = осознанный `--resnapshot`).
+
+**Схема disposition (D2, контракт-несущее).** v2-миграция (append, IF NOT EXISTS): `chunk_status(book_id,chapter,chunk_idx,stage,snapshot_id,content_hash,disposition,flag_reason,attempts,final_hash,cost_usd,detail)`, PK (book,chapter,chunk,stage). Это **резолв поверх append-only checkpoints**, не колонка на checkpoints — на resume пересобирается из чекпоинтов. `disposition ∈ {ok,flagged,skipped}`; `jobs.status` контентом НЕ расширён (failed = только инфра). `flag_reason` — Go-константы как `Finish*` (12 шт.; эмитятся length/empty/loop_degenerate/hard_refusal/soft_refusal/content_filter/cjk_artifact/decode_error, зарезервированы coverage_fail/excision_suspect/hard_block/upstream_not_ok под шаги 6–7). Единый `classify(src,out,finish,target)` — порядок: refusal-блэклист (порт `refusal_bench.py` 1:1, en/ru/zh/ja) / CJK-echo **ДО** length/empty (усечённый отказ не триггерит платную переатаку); `content_filter` best-effort (реальная страховка — блэклист). n-gram loop-чек ПЕРЕД удвоением max_tokens. `maxTokensForAttempt` — чистая функция (удвоение per attempt), её версия в snapshot (`maxTokensPolicy`). Retry-flagged только {length,empty} на той же модели по оси attempt (до `regenerate_before_escalate`, потом флаг); refusal/filter/cjk/loop/decode детерминированы → не переатакуются. Три анти-веджа отработали (**mutation-verified**): resume читает `chunk_status` до вызова; empty/decode → flagged (цикл идёт, не крэш); legacy-пустой чекпоинт самолечится classify-ем без ре-биллинга. Exit: `CompletedWithFlags` sentinel → 0 чисто / 2 с флагами (N допустимо) / 1 инфра.
+
+**Фиксы.** **F4:** усечённый `length` с непустым черновиком теперь классифицируется (flagged/retry), не течёт в edit как OK (`classify` после settle). **D2.5:** edit-`max_tokens` от длины ЧЕРНОВИКА (`prev`), не исходника — монолингвальный редактор работает по русскому черновику ≈1.9× исходника (mutation-verified). **F3 честно:** `chunk_status.cost_usd` суммирует все попытки; per-run cost отделён от кумулятивного; idempotency-ключ не делал (отложено).
+
+**Селфревью (адверсариальный воркфлоу, 5 измерений → верификация каждой находки).** disposition-money/resume/classifier — **0 дефектов**. 2 подтверждённых исправлено (оба mutation-verified): (1) *[детерминизм, medium]* `chunk_status` fast-path резолвил ПОЗИЦИОННО и подавал устаревший перевод при правке исходника (исходник не в snapshot) — фикс: сигнатура `content_hash` отрендеренных msgs в `chunk_status`, fast-path доверяет строке только при совпадении snapshot И content (иначе — content-safe attempt-loop; правка src = тихий per-chunk ре-перевод по §3.4, не устаревший подача); стоило переноса дешёвого рендера ДО чекпоинт-резолва (LLM-вызов по-прежнему загейчен — «не переатаковать» сохранено). (2) *[CLI, low]* `flag.ExitOnError` → `os.Exit(2)` на кривом флаге, коллизия с exit-2 «с флагами» → `ContinueOnError` (кривой флаг = exit 1). Тесты: 12 новых (disposition/chunker/chunkstatus юниты + e2e мульти-чанк/resume/flag+skip/exit-коды/F4/D2.5/decode/legacy-self-heal/source-edit), все mutation-проверены на укус.
+
+**Техдолг вперёд (не блокирует шаг 3):** чанкер — плейсхолдер (шаг 3 подменит `SplitChunks`, чтит контракт `Chunk` + бампает `chunkerVersion`); небилленный upstream-сбой = инфра-аболт (D4 pause/resume), `upstream_not_ok`/`hard_block`-константы под HTTP-классификацию шага 7; редкий угол: kill-9 в микро-окне (потеря `chunk_status`) ПЛЮС одновременное снижение `regenerate_before_escalate` может консервативно пере-флагнуть бывший-ok чанк (без денег/расхождения — фаст-путь чинит норму, self-heal best-effort). `jobs.status` теперь наблюдательный (ничего на нём не ветвится).
+
## Полигон
(секция параллельной сессии — записи добавлять сюда)
@@ -299,7 +313,8 @@ keep-alive (Ф1–2), инъекция глоссария (`selective`), пор
- [—] **Банк памяти (задача 5 + шире)** — по решению владельца выносится в **отдельную сессию «Валидация банка памяти»** (главная точка отказа пайплайна; не бросаться в код без разбора опасных сценариев). Полигон подготовил черновик промта → `docs/MEMORY_RESEARCH_SESSION_PROMPT.md`; **оркестратору вычитать/финализировать**, затем владелец запускает. Замер эмбеддингов (bge-m3 vs Qwen3-Embedding vs LaBSE) — часть мандата той сессии.
- [ ] Довалидация токен-калибровки на 3–5 главах реальных вебновелл (файлы владельца).
- [ ] Задача 4 (пилот выбора ядра, Фаза 2.5): протокол BWS + панель LLM-судей.
-- [ ] **Эксперимент 05 — memory-bet (индикативный), готовая спека ниже** — решает go/no-go на ставку банка памяти; runnable СЕЙЧАС без бэкенда (чистый eval).
+- [x] ~~**Эксперимент 05 — memory-bet**~~ — **СНЯТ как низкосигнальный** (решение владельца). Прогон был, но его ось C0-vs-C1 («бредит ли модель без глоссария») предрешена, а C3/C2 лишь пере-подтвердили уже процитированную литературу (2510.00829) — тест не мог изменить ни одного решения. Артефакты (doc/скрипт/данные) удалены. Единственный actionable-вывод (post-check + disposition обязательны) уже зафиксирован в реестре `architecture/06` (A2/E1) независимо от этого прогона. **Урок для eval:** тест ставить только если его исход мог бы перевернуть решение И не установлен литературой. Ценность вместо этого — валидация самого МЕХАНИЗМА (см. ниже).
+- [x] **Референс горячего пути банка памяти + self-tests** → `eval/memory_hotpath.py` (11/11 PASS). Исполняемая спека механизма (не продукт): нормализация trad→simp/kana, точный матч ключей/алиасов с запретом одиночных (омографы 送灶/炎/修/气 НЕ инъектируются), спойлер-фильтр `since_ch/until_ch` (hard-reject), sticky для местоименных чанков, disposition confirmed/ambiguous/reject, post-check (ловит и утечку, и отравление). **Бэкенду:** порт в Go 1:1, тесты T1–T10 → unit-тесты; `[PROD]`-замены: OpenCC, Aho-Corasick, pymorphy/`decl`.
> **[ЗАДАЧА ОРКЕСТРАТОРА → eval] Эксперимент 05: memory-bet (индикативный).** Отвечает на главный незакрытый вопрос валидации (research/13 §Вердикт): оправдан ли банк памяти vs просто длинный контекст, и вредит ли ошибочная инъекция (страх владельца). **Runnable сейчас, продукт не нужен** — это конструирование промптов + API-вызовы + метрики.
> **Метод:** один стек `deepseek-chat` (черновик) → `grok-4.3` (редактор, thinking OFF, temp 0.4 — квирки в `00-provider-quirks`). Реальные zh→ru чанки: `eval/data/samples/zh/luxun-ah-q-ch1-4.txt` (повторяющиеся имена 阿Q/赵太爷/王胡/送灶; есть русский эталон Рогова `samples/ru/luxun-ah-q-ru.txt` для якоря верности). Нарезать ~5–8 перекрывающихся чанков; собрать мини-глоссарий (5–10 записей: имена+реалии, с dst и `decl`). **4 условия на чанк:**
@@ -367,3 +382,7 @@ keep-alive (Ф1–2), инъекция глоссария (`selective`), пор
> **[БЭКЕНДУ]** реестр `architecture/06` — твой контракт для шага 4 (миграция памяти v2). Порядок: **F1 (snapshotID под память+context-assembly) — ПЕРВЫМ** (общий гейт, держит D1/D5, независим от memory-вопроса — код отстаёт от impl-notes §3.2, там уже предписан версия-счётчик); затем схема v2 = D7 + добавления реестра (`sense`, ось редакционного времени, `UNIQUE(src,sense,window)`+обратная dst-проверка, `min_key_len`+запрет одиночных ключей, артефакт нормализации NFKC/OpenCC/kana/ruby); горячий путь = **Aho-Corasick/trie, НЕ FTS5** (impl-notes §3.6 уже так); post-check обязателен, слепой post-replace в ru запрещён; трёхсторонний disposition + retrieval-state запись. Режим инъекции держи конфигом (`selective|full_prefix`) — eval потом решит дефолт. Отложенное в v1.1/Ф2 (заложить поля/место, не реализовывать): Палладий/Поливанов B6, гейт фактичности резюме D1, эмбеддинг-слой A7, бэкап файла F4, морфо-гейт глагольного рода C3.
> **[ПОЛИГОНУ/ПИЛОТУ]** in-house memory-eval — дециждающий эксперимент, фолдится в пилот Ф2.5: WMT25-трёхрежимный (no / correct / **random** terminology — random-рука ловит тихую деградацию) на реальных zh/ja→ru чанках; **обязательный baseline «банк ON vs OFF» и «банк vs длинный контекст frontier-модели»** на том же стеке (DelTA этого не делала — это пробел, который решает go/no-go на ставку); мерить две оси раздельно (самосогласованность approved-форм + source-fidelity/omission через CometKiwi + LLM-судья-против-источника); ru-специфика (согласование рода вкл. глагол, склонение dst); предзарегистрировать пороги + допустимый flag-volume на главу. Эмбеддинг-бенчмарк уже сделан этой сессией (bge-m3 дефолт, `eval/retrieval_bench.py`) — переиспользуй.
+
+> **[СЕССИЯ ЗАКРЫТА, 04.07]** Выходы: research/13 (+§«Честные слабые места»), architecture/06, `eval/memory_hotpath.py` (спека), `eval/retrieval_bench.py` (эмбеддинги). Эксп-05 снят как низкосигнальный. **Перед кодом реализующей сессии — прочитать research/13 §«Честные слабые места»:** F1 подан как решённый, но там развилка (snapshot материализует байты vs store версионирует глоссарий); post-check в русской морфологии — несущий, но невалидированный (померить до доверия как гейту); «пусто=безопасно» — размен на ложный-пропуск, который тоже тихий; ставка на детерминированный no-LLM путь — cost-driven, не evidence-driven (DelTA валидирует LLM-в-цикле). Гигиена: фоновых процессов нет, ollama полигона не тронут, `eval/.venv` дополнен (torch-cpu/sentence-transformers/openai/dotenv — переиспользуемо).
+
+> **[ДОБАВЛЕНО после закрытия] Локалка × банк памяти — тест экстракции (дыра G1)** → `eval/extract_bench.py`. Локалка касается банка в 2 точках: эмбеддинги (сделано, bge-m3) и экстракция терминов (write-path). Индикативно (3 кейса zh+ja): **локаль хорошо СПОТИТ сущности** (qwen3-abliterated:8b recall 1.0, 0 галлюцинаций, ~15с) но **плохо РЕНДЕРИТ dst** (阿Q→«А Цзы», 羅生門→«Россомаха») → **разделить: спот локально (дёшево) / рендер dst — облако+человек** (уточняет Р4/G1, где «экстракция=локаль» слито в одно). Побочно: deepseek-v4-flash молча вернул ПУСТОЙ ответ на части чанков (祝福 воспроизводимо) — тот empty-paid-response, ради которого существуют D2/coverage-гейт. Аггрегат deepseek по recall не репортил (контаминирован пустыми) — только per-case. Оговорка: 3 кейса, индикативно.