diff --git a/backend/internal/config/models.go b/backend/internal/config/models.go index b5eca47..80b7ff1 100644 --- a/backend/internal/config/models.go +++ b/backend/internal/config/models.go @@ -49,10 +49,17 @@ type Provider struct { // thinking:disabled is fine because its input is a Russian draft (no CJK to // echo): GLM's provider carries no such flag. Validation-only metadata — it // never reaches the wire, so it is NOT part of the snapshot (no --resnapshot). - EchoesWhenThinkingOff bool `yaml:"echoes_when_thinking_off"` - CacheTTL string `yaml:"cache_ttl"` // anthropic kind: "", "5m", "1h" - Model string `yaml:"model"` // local kind: the backend's own tag - MaxTokens int `yaml:"max_tokens"` + EchoesWhenThinkingOff bool `yaml:"echoes_when_thinking_off"` + // Permissive marks a provider allowed to serve the 18+ channel B (the most + // permissive large APIs: xAI/Grok, DeepSeek official, local abliterated — D3). + // It is the TYPE that enforces channel-B isolation (D4.1): a channel=adult stage + // may only run on, and escalate to, a permissive provider — a structural gate, + // not a comment, so an SFW-only provider can never silently serve/answer a + // permissive chunk. Validation-only metadata (never on the wire, not in snapshot). + Permissive bool `yaml:"permissive"` + CacheTTL string `yaml:"cache_ttl"` // anthropic kind: "", "5m", "1h" + Model string `yaml:"model"` // local kind: the backend's own tag + MaxTokens int `yaml:"max_tokens"` // Temperature (local kind): override запроса — ollama чтит request- // температуру поверх Modelfile, и температура облачной роли молча // расстроила бы локальную модель. 0 = наследовать запрос. @@ -229,6 +236,17 @@ func (m *Models) Prices() (*ledger.Pricer, error) { return ledger.NewPricer(prices, m.Models[m.DefaultModel].Price.ToLedger()) } +// providerPermissive reports whether a model's provider is marked permissive +// (channel-B-eligible, D4.1). Unknown model → false (fail closed: an undefined +// model is never treated as permissive). +func (m *Models) providerPermissive(modelName string) bool { + mod, ok := m.Models[modelName] + if !ok { + return false + } + return m.Providers[mod.Provider].Permissive +} + // APIKey resolves a provider's key from the environment ("" if the provider // declares no env var — the local backend). func (p Provider) APIKey() string { @@ -387,6 +405,16 @@ func (m *Models) CheckKeys(pipe *Pipeline) error { } } } + // Single-hop fallback models are reachable only once an escalation budget is set + // (opt-in); preflight their keys then, so an escalation does not 401 mid-book + // after the primary already flagged. + if pipe.Escal.BudgetUSD > 0 { + for _, st := range pipe.Stages { + if st.EscalateTo != "" { + needed[st.EscalateTo] = struct{}{} + } + } + } var problems []string seen := map[string]bool{} // dedupe per provider for name := range needed { diff --git a/backend/internal/config/pipeline.go b/backend/internal/config/pipeline.go index 21a6457..e268590 100644 --- a/backend/internal/config/pipeline.go +++ b/backend/internal/config/pipeline.go @@ -64,6 +64,18 @@ type Stage struct { PromptVersion string `yaml:"prompt_version"` Temperature float64 `yaml:"temperature"` Reasoning string `yaml:"reasoning"` // "", off, low, medium, high + // EscalateTo is the SINGLE-HOP fallback model tried ONCE when this stage's + // output is a deterministic content-failure another model might fix (echo / + // excision / refusal — D12). Empty = no escalation (e.g. the editor is pinned + // per book — its style must not drift to a foreign model, D12/2605.13368). The + // fallback is its OWN model, which is the request_hash axis (a distinct + // checkpoint, never the failed primary's). Gated by escalation.budget_usd. + EscalateTo string `yaml:"escalate_to"` + // Channel selects the safety channel: "" | sfw (default) | adult (18+). An + // adult stage is isolated by TYPE (D4.1): its model AND its escalate_to must be + // on permissive providers, enforced at load — never a fall-through to a refusing + // SFW provider. + Channel string `yaml:"channel"` } // Gates is the QA-gate config skeleton (пороги полигона; исполнение — Фаза 1). @@ -197,6 +209,28 @@ func LoadPipeline(path string, models *Models) (*Pipeline, error) { default: bad("stage %q: reasoning must be off|low|medium|high, got %q", st.Name, st.Reasoning) } + switch st.Channel { + case "", "sfw", "adult": + default: + bad("stage %q: channel must be sfw|adult, got %q", st.Name, st.Channel) + } + if st.EscalateTo != "" { + if _, ok := models.Models[st.EscalateTo]; !ok { + bad("stage %q: escalate_to model %q is not defined in models.yaml", st.Name, st.EscalateTo) + } else if st.EscalateTo == st.Model { + bad("stage %q: escalate_to must differ from the primary model %q (a same-model hop is a guaranteed repeat)", st.Name, st.Model) + } + } + // D4.1: channel-B (18+) isolation is enforced by TYPE — a permissive stage + // may only run on, and escalate to, a permissive provider. + if st.Channel == "adult" { + if !models.providerPermissive(st.Model) { + bad("stage %q: channel=adult requires model %q on a permissive provider (D4.1)", st.Name, st.Model) + } + if st.EscalateTo != "" && !models.providerPermissive(st.EscalateTo) { + bad("stage %q: channel=adult escalate_to %q must be on a permissive provider (D4.1 — no fall-through to a refusing SFW channel)", st.Name, st.EscalateTo) + } + } } for chain, ms := range p.Escal.Chains { for _, m := range ms { diff --git a/backend/internal/pipeline/disposition.go b/backend/internal/pipeline/disposition.go index ee505f5..24028d5 100644 --- a/backend/internal/pipeline/disposition.go +++ b/backend/internal/pipeline/disposition.go @@ -88,6 +88,23 @@ func (r FlagReason) retryable() bool { return r == FlagLength || r == FlagEmpty } +// escalatable reports whether a flag is a DETERMINISTIC content-failure that a +// DIFFERENT model might fix — the single-hop escalation trigger (D12, the +// deterministic-content-failure class). These arrive as HTTP 200 (the provider +// billed a "translation" that is echo / excision / a refusal), so a same-model +// retry just re-produces them (which is why they are non-retryable) — escalation +// routes them to another model ONCE. length/empty are excluded (retryable on the +// same model with a bigger budget); decode_error is transport-shaped (a billed +// unreadable body), not a content verdict another model would reliably avoid. +func (r FlagReason) escalatable() bool { + switch r { + case FlagCJKArtifact, FlagExcisionSuspect, FlagLoopDegenerate, + FlagHardRefusal, FlagSoftRefusal, FlagContentFilter: + return true + } + return false +} + // disposition maps a reason to the resolved chunk×stage state. func (r FlagReason) disposition() Disposition { if r == reasonOK { diff --git a/backend/internal/pipeline/runner.go b/backend/internal/pipeline/runner.go index 6ec436e..4c33837 100644 --- a/backend/internal/pipeline/runner.go +++ b/backend/internal/pipeline/runner.go @@ -179,6 +179,24 @@ func (r *Runner) classifyOutput(source, output, finish string) classification { return cls } +// escalationBudgetRemains reports whether the book may still spend on a single-hop +// fallback draft: escalation is OPT-IN via escalation.budget_usd (0 = disabled, the +// boevoy default — a stage's escalate_to is inert until a premium budget is set), and +// capped at that budget summed over the book's escalation checkpoints (money-path +// durable, resume-safe). A soft cap: the hop already in flight may cross it slightly, +// but the NEXT chunk's escalation is then denied. +func (r *Runner) escalationBudgetRemains() (bool, error) { + budget := r.Pipeline.Escal.BudgetUSD + if budget <= 0 { + return false, nil + } + spent, err := r.Store.EscalationSpentUSD(r.Book.BookID) + if err != nil { + return false, err + } + return spent < budget, nil +} + // memoryVersion is the content-hash of the deterministically materialized // injected memory (approved glossary + summaries + series-bible), the memory // component of the snapshot (D5.2/D8). Until the memory bank v2 migration lands @@ -226,6 +244,14 @@ func (r *Runner) snapshotID() (id, payload string, err error) { // тель), поэтому обязана входить в снапшот — иначе правка каппы молча // false-хитит чекпоинты (тот же класс D5.2, что и payload ниже). Capability json.RawMessage `json:"capability,omitempty"` + // EscalateTo + EscalateCapability — the single-hop fallback model and its + // resolved wire shape (Веха 2.5). The fallback's request_hash uses ITS model, + // and its wire body uses ITS capability — both must be in the snapshot, or + // changing the escalation model / its caps would silently false-hit an + // escalated chunk's checkpoint (closes the D5.2 escalation-capability techdebt: + // stages were folded, escalation models were not, until the cycle landed). + EscalateTo string `json:"escalate_to,omitempty"` + EscalateCapability json.RawMessage `json:"escalate_capability,omitempty"` } snap := struct { BriefHash string `json:"brief_hash"` @@ -301,6 +327,17 @@ func (r *Runner) snapshotID() (id, payload string, err error) { return "", "", cerr } ss.Capability = capRaw + // Fold the single-hop fallback model + its resolved capability (Веха 2.5): + // both are wire-affecting inputs of an escalated call, so changing them is a + // loud --resnapshot, not a silent false-hit on an escalated checkpoint. + if st.EscalateTo != "" { + ss.EscalateTo = st.EscalateTo + escCap, eerr := json.Marshal(r.Models.ResolveCapability(st.EscalateTo)) + if eerr != nil { + return "", "", eerr + } + ss.EscalateCapability = escCap + } snap.Stages = append(snap.Stages, ss) } data, err := json.Marshal(snap) @@ -327,6 +364,10 @@ type StageResult struct { FlagReason FlagReason // "" when ok Detail string Attempts int + // Escalated — a single-hop fallback draft was tried this stage (D12); when the + // fallback passed the re-gate, Model above is the fallback (it answered). + Escalated bool + EscalationModel string // the fallback model when Escalated ("" otherwise) } // ChunkOutcome is one chunk's result across the stage list. @@ -553,7 +594,7 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, stageIdx int, sn attemptsMade := 0 for attempt := 0; ; attempt++ { maxTokens := maxTokensForAttempt(baseMaxTokens, attempt) - att, err := r.runAttempt(ctx, st, snapID, ch, job, attempt, maxTokens, msgs) + att, err := r.runAttempt(ctx, st, st.Model, snapID, ch, job, attempt, maxTokens, msgs, false) if err != nil { return nil, err // infra failure } @@ -577,6 +618,42 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, stageIdx int, sn break } + // Single-hop escalation (D12 deterministic-content-failure class): a flag that a + // DIFFERENT model might fix (echo / excision / refusal) is routed ONCE to the + // stage's named fallback, under the book's escalation.budget_usd, and RE-GATED + // (classifyOutput runs on the fallback output too — §3.8). The editor declares no + // escalate_to → pinned per book (its style must not drift to a foreign model, + // D12/2605.13368). The fallback uses its OWN model (the request_hash axis), so its + // checkpoint never collides with the failed primary's, and it is exactly ONE call + // (not a fresh retry budget): total calls/chunk = primary attempts + 1 hop, never + // reset on the fallback (the LiteLLM #19985 retry×fallback blow-up guard). + escalated, escModel := false, "" + if last.cls.Reason.escalatable() && st.EscalateTo != "" { + remains, err := r.escalationBudgetRemains() + if err != nil { + return nil, err + } + if remains { + fb, err := r.runAttempt(ctx, st, st.EscalateTo, snapID, ch, job, 0, baseMaxTokens, msgs, true) + if err != nil { + return nil, err + } + cumCost += fb.cumCost + runCost += fb.runCost + anyFresh = anyFresh || fb.freshCall + escalated, escModel = true, st.EscalateTo + r.Log.WarnContext(ctx, "stage escalated to a fallback model", + "stage", st.Name, "chapter", ch.Chapter, "chunk", ch.ChunkIdx, + "primary_reason", string(last.cls.Reason), "fallback", st.EscalateTo, + "fallback_disposition", string(fb.cls.Reason.disposition())) + if fb.cls.ok() { + last = fb // the fallback draft passed the re-gate → it is authoritative + } + // else: the fallback also failed → keep the primary flag (last unchanged); + // the fallback call is billed and counted, the chunk stays flagged (1 hop). + } + } + disposition := last.cls.Reason.disposition() finalHash := "" if disposition == DispOK { @@ -595,16 +672,18 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, stageIdx int, sn 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, + 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, + Escalated: escalated, + EscalationModel: escModel, } if disposition == DispOK { sr.Text = last.text @@ -635,10 +714,10 @@ type stageAttempt struct { // 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, +func (r *Runner) runAttempt(ctx context.Context, st config.Stage, model, snapID string, ch Chunk, job *store.Job, attempt, maxTokens int, msgs []llm.Message, escalation bool) (stageAttempt, error) { + reqHash := RequestHash(r.Book.BookID, ch.Chapter, ch.ChunkIdx, attempt, st.Name, st.Role, model, st.Temperature, st.Reasoning, false, maxTokens, snapID, msgs) - att := stageAttempt{reqHash: reqHash, modelActual: st.Model} + att := stageAttempt{reqHash: reqHash, modelActual: 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; @@ -656,7 +735,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, att.cls = r.classifyOutput(ch.Text, cp.ResponseText, cp.FinishReason) 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, + Stage: st.Name, Role: st.Role, ModelRequested: model, ModelActual: cp.ModelActual, RequestHash: reqHash, TMHit: true, OK: att.cls.ok(), FinishReason: cp.FinishReason, Degraded: degradedTag(att.cls), }) @@ -667,7 +746,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, } // Fresh call: reserve → call → settle+checkpoint (atomic, §3.3). - price := r.Pricer.PriceFor(st.Model) + price := r.Pricer.PriceFor(model) promptEst := 0 for _, m := range msgs { promptEst += EstimateTokens(m.Content) @@ -690,7 +769,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, return att, fmt.Errorf("pipeline: daily USD ceiling reached (%.2f$)", r.Book.Ceilings.DayUSD) } - client, err := r.client(st.Model) + client, err := r.client(model) if err != nil { _ = r.Store.Release(resv) _ = r.Store.SetJobStatus(job.ID, "failed") @@ -704,7 +783,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, start := time.Now() resp, err := client.Complete(ctx, llm.LLMRequest{ - Model: st.Model, + Model: model, Messages: msgs, MaxTokens: maxTokens, Temperature: st.Temperature, @@ -720,8 +799,9 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, // 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: ch.ChunkIdx, Attempt: attempt, - Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: st.Model, + Stage: st.Name, Role: st.Role, ModelRequested: model, ModelActual: model, ResponseText: "", UsageJSON: "{}", CostUSD: estimate, FinishReason: decodeErrorFinish, + Escalation: escalation, }); serr != nil { return att, fmt.Errorf("pipeline: settle after billed decode failure: %w", serr) } @@ -730,7 +810,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, 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: ch.Chapter, ChunkIdx: ch.ChunkIdx, - Stage: st.Name, Role: st.Role, ModelRequested: st.Model, + Stage: st.Name, Role: st.Role, ModelRequested: model, RequestHash: reqHash, CostUSD: estimate, LatencyMS: att.latency, Degraded: "billed_2xx_decode_failed", Err: err.Error(), OK: false, FinishReason: decodeErrorFinish, }) @@ -745,7 +825,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, } 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, + Stage: st.Name, Role: st.Role, ModelRequested: model, RequestHash: reqHash, LatencyMS: att.latency, Err: err.Error(), OK: false, }) _ = r.Store.SetJobStatus(job.ID, "failed") @@ -754,7 +834,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, modelActual := resp.Model if modelActual == "" { - modelActual = st.Model + modelActual = model } att.modelActual = modelActual att.text = resp.Text @@ -763,7 +843,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, // Цена по фактически ответившей модели, с fallback на ЗАПРОШЕННУЮ (не на // дешёвый глобальный якорь), если провайдер вернул канонизированный слаг. - price = r.Pricer.PriceForResponse(st.Model, modelActual) + price = r.Pricer.PriceForResponse(model, modelActual) cost := ledger.CostUSD(price, resp.Usage) // Платная модель (InputPerM>0) вернула 2xx с нулевым usage — $0 ослепил бы // потолок: берём консервативную оценку. Для local ($0 цена) нулевой usage @@ -785,9 +865,9 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, if err := r.Store.SettleWithCheckpoint(resv, cost, store.Checkpoint{ RequestHash: reqHash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: attempt, Stage: st.Name, Role: st.Role, - ModelRequested: st.Model, ModelActual: modelActual, ResponseText: resp.Text, + ModelRequested: model, ModelActual: modelActual, ResponseText: resp.Text, UsageJSON: string(usageJSON), CostUSD: cost, FinishReason: resp.FinishReason, - ProviderRequestID: resp.ProviderRequestID, + ProviderRequestID: resp.ProviderRequestID, Escalation: escalation, }); err != nil { // The provider HAS billed this 2xx; failing to persist means the money // state is behind reality — fail loud, never continue on top. @@ -802,7 +882,7 @@ func (r *Runner) runAttempt(ctx context.Context, st config.Stage, snapID string, 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: modelActual, + Stage: st.Name, Role: st.Role, ModelRequested: model, ModelActual: modelActual, RequestHash: reqHash, PromptTokens: resp.Usage.PromptTokens, CachedTokens: resp.Usage.CachedTokens, CacheCreationTokens: resp.Usage.CacheCreationTokens, diff --git a/backend/internal/pipeline/runner_test.go b/backend/internal/pipeline/runner_test.go index 710caf8..c59a54c 100644 --- a/backend/internal/pipeline/runner_test.go +++ b/backend/internal/pipeline/runner_test.go @@ -961,3 +961,266 @@ func TestRunnerRejectsCoverageGateWithoutThresholds(t *testing.T) { t.Fatalf("a rejected config must not reach the provider, calls=%d", rec.count()) } } + +// --- Веха 2.5: single-hop escalation (B, D12) ------------------------------- + +// setupEscalationProject wires a ja→ru book whose draft stage escalates to a second +// model (fake-fallback) on a deterministic content-failure, under escalation +// budget_usd. Both models sit on the same fake provider; the mock distinguishes them +// by the "model" field in the request body. Optional permissive marks the provider +// permissive and channel=adult on the draft (D4.1 isolation test). +func setupEscalationProject(t *testing.T, providerURL string, budgetUSD float64, permissive bool, channel string) string { + t.Helper() + dir := t.TempDir() + writeFile(t, filepath.Join(dir, "prompts", "translator.md"), + "Переводи с {{source_lang}} на {{target_lang}}.\n---USER---\n{{text}}") + writeFile(t, filepath.Join(dir, "prompts", "editor.md"), + "Редактируй перевод.\n---USER---\nИсходник: {{text}}\nЧерновик перевода для редактуры: {{draft}}") + + permLine := "" + if permissive { + permLine = "\n permissive: true" + } + writeFile(t, filepath.Join(dir, "models.yaml"), fmt.Sprintf(` +prices_checked: %q +default_model: fake-model +providers: + fake: + kind: openai + base_url: %q%s + timeouts: { attempt_s: 5, max_attempts: 2, backoff_cap_s: 1 } +models: + fake-model: + provider: fake + price: { input_per_m: 1.0, cached_per_m: 0.1, cache_write_per_m: 0, output_per_m: 2.0 } + fake-fallback: + provider: fake + 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, permLine)) + + chanLine := "" + if channel != "" { + chanLine = ", channel: " + channel + } + 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: 0 } +stages: + - { name: draft, role: translator, model: fake-model, prompt: prompts/translator.md, prompt_version: v-test, temperature: 0.3, reasoning: "off", escalate_to: fake-fallback%s } + - { name: edit, role: editor, model: fake-model, prompt: prompts/editor.md, prompt_version: v-test, temperature: 0.4, reasoning: "off" } +escalation: { budget_usd: %g } +`, chanLine, budgetUSD)) + + writeFile(t, filepath.Join(dir, "source.txt"), "静かな図書館の朝。") + writeFile(t, filepath.Join(dir, "book.yaml"), ` +book_id: test-book +title: Тест +source_lang: ja +target_lang: ru +genre: ранобэ +audience: тест +venuti: 0.5 +honorifics: keep +transcription: polivanov +footnotes: minimal +pipeline: pipeline.yaml +models: models.yaml +source_file: source.txt +ceilings: { book_usd: 5.0, day_usd: 10.0 } +`) + return filepath.Join(dir, "book.yaml") +} + +// echoOrClean is the escalation mock: the primary (fake-model) ECHOES the CJK source +// (cjk_artifact); the fallback (fake-fallback) returns a clean Russian translation. +func echoOrClean(body string) (string, string) { + if isEditBody(body) { + return "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД", "stop" + } + if strings.Contains(body, "fake-fallback") { + return "Тихое утро в библиотеке.", "stop" + } + return "静かな図書館の朝。", "stop" // echo of the CJK source → cjk_artifact +} + +// A deterministic echo (cjk_artifact) is escalated ONCE to the fallback model, which +// translates cleanly → the chunk is OK, billed to the fallback, and the edit runs. +func TestRunnerEscalationFixesEcho(t *testing.T) { + rec := &reqRec{} + srv := newJSONProvider(rec, echoOrClean) + defer srv.Close() + bookPath := setupEscalationProject(t, srv.URL, 1.0, false, "") + ctx := context.Background() + + r := newRunner(t, bookPath) + defer r.Close() + res, err := r.TranslateBook(ctx) + if err != nil { + t.Fatal(err) + } + // draft(fake-model)=echo → escalate draft(fake-fallback)=clean → edit = 3 calls. + if rec.count() != 3 { + t.Fatalf("want 3 calls (primary + fallback + edit), got %d", rec.count()) + } + ch := res.Chunks[0] + if ch.Disposition != DispOK || res.Flagged != 0 || res.ExitCode() != 0 { + t.Fatalf("escalated chunk must resolve OK, got %+v (flagged=%d)", ch, res.Flagged) + } + draft := ch.Stages[0] + // The mock canonicalizes every response model to "fake-model", so Model reflects + // that; the escalation is proven by Escalated + EscalationModel + the fallback's text. + if !draft.Escalated || draft.EscalationModel != "fake-fallback" || draft.Text != "Тихое утро в библиотеке." { + t.Fatalf("draft must record the successful escalation to fake-fallback, got %+v", draft) + } + if ch.FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" { + t.Fatalf("final text must come from the edit over the fallback draft, got %q", ch.FinalText) + } + esc, err := r.Store.EscalationSpentUSD("test-book") + if err != nil || esc <= 0 { + t.Fatalf("the escalation call must be tagged and summable, got %v (err %v)", esc, err) + } +} + +// If the fallback ALSO fails the re-gate, the chunk stays flagged (primary reason), +// the edit is skipped, and it was exactly ONE hop (primary + fallback, no more). +func TestRunnerEscalationFallbackAlsoFailsFlags(t *testing.T) { + rec := &reqRec{} + srv := newJSONProvider(rec, func(body string) (string, string) { + if isEditBody(body) { + return "ред", "stop" + } + return "静かな図書館の朝。", "stop" // BOTH models echo → cjk_artifact + }) + defer srv.Close() + bookPath := setupEscalationProject(t, srv.URL, 1.0, false, "") + + r := newRunner(t, bookPath) + defer r.Close() + res, err := r.TranslateBook(context.Background()) + if err != nil { + t.Fatal(err) + } + if rec.count() != 2 { + t.Fatalf("want exactly 2 calls (primary + 1 fallback hop, edit skipped), got %d", rec.count()) + } + ch := res.Chunks[0] + if ch.Disposition != DispFlagged || ch.FlagReason != FlagCJKArtifact || res.ExitCode() != 2 { + t.Fatalf("a failed escalation must keep the primary flag, got %+v", ch) + } + if !ch.Stages[0].Escalated { + t.Fatalf("the draft must record that escalation was tried, got %+v", ch.Stages[0]) + } + if ch.Stages[1].Disposition != DispSkipped { + t.Fatalf("edit must be skipped after a failed escalation, got %+v", ch.Stages[1]) + } +} + +// escalation.budget_usd = 0 disables escalation (opt-in): the deterministic flag +// stays a flag, no fallback call is made. +func TestRunnerEscalationBudgetZeroDisables(t *testing.T) { + rec := &reqRec{} + srv := newJSONProvider(rec, echoOrClean) + defer srv.Close() + bookPath := setupEscalationProject(t, srv.URL, 0, false, "") // budget 0 + + r := newRunner(t, bookPath) + defer r.Close() + res, err := r.TranslateBook(context.Background()) + if err != nil { + t.Fatal(err) + } + if rec.count() != 1 { + t.Fatalf("budget=0 must not make a fallback call, got %d calls", rec.count()) + } + ch := res.Chunks[0] + if ch.Disposition != DispFlagged || ch.FlagReason != FlagCJKArtifact || ch.Stages[0].Escalated { + t.Fatalf("with no budget the echo must stay flagged, unescalated, got %+v", ch.Stages[0]) + } +} + +// A successful escalation is resume-safe: the chunk resolves from chunk_status and +// its final_hash points at the FALLBACK checkpoint, served at $0 with no re-call. +func TestRunnerEscalationResumeServesFallback(t *testing.T) { + rec := &reqRec{} + srv := newJSONProvider(rec, echoOrClean) + defer srv.Close() + bookPath := setupEscalationProject(t, srv.URL, 1.0, false, "") + ctx := context.Background() + + r1 := newRunner(t, bookPath) + if _, err := r1.TranslateBook(ctx); err != nil { + t.Fatal(err) + } + r1.Close() + callsAfterRun1 := rec.count() + + r2 := newRunner(t, bookPath) + defer r2.Close() + res, err := r2.TranslateBook(ctx) + if err != nil { + t.Fatal(err) + } + if rec.count() != callsAfterRun1 { + t.Fatalf("resume must not re-call after a successful escalation, calls %d -> %d", callsAfterRun1, rec.count()) + } + if res.TotalUSD != 0 || res.Chunks[0].Disposition != DispOK { + t.Fatalf("resume must serve the escalated chunk at $0, got %+v", res.Chunks[0]) + } + if res.Chunks[0].FinalText != "ОТРЕДАКТИРОВАННЫЙ ПЕРЕВОД" { + t.Fatalf("resume text differs: %q", res.Chunks[0].FinalText) + } +} + +// The fallback model is folded into the snapshot: removing escalate_to changes the +// snapshot, so resume fails loud until --resnapshot (D5.2 escalation-capability closure). +func TestRunnerEscalationEntersSnapshot(t *testing.T) { + rec := &reqRec{} + srv := newJSONProvider(rec, echoOrClean) + defer srv.Close() + bookPath := setupEscalationProject(t, srv.URL, 1.0, false, "") + ctx := context.Background() + + r1 := newRunner(t, bookPath) + if _, err := r1.TranslateBook(ctx); err != nil { + t.Fatal(err) + } + r1.Close() + + 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), ", escalate_to: fake-fallback", "", 1)) + + r2 := newRunner(t, bookPath) + _, err = r2.TranslateBook(ctx) + r2.Close() + if err == nil || !strings.Contains(err.Error(), "resnapshot") { + t.Fatalf("changing the fallback model must fail loud mentioning --resnapshot, got: %v", err) + } +} + +// D4.1: channel-B (18+) isolation is enforced by TYPE — a channel=adult stage on a +// NON-permissive provider is rejected at load, never a silent fall-through. +func TestRunnerChannelBRequiresPermissive(t *testing.T) { + rec := &reqRec{} + srv := newJSONProvider(rec, echoOrClean) + defer srv.Close() + + // channel=adult, provider NOT permissive → NewRunner must fail. + nonPerm := setupEscalationProject(t, srv.URL, 1.0, false, "adult") + if _, err := NewRunner(nonPerm, obs.NewLogger()); err == nil || !strings.Contains(err.Error(), "permissive") { + t.Fatalf("channel=adult on a non-permissive provider must fail-fast, got: %v", err) + } + + // channel=adult, provider permissive → accepted. + perm := setupEscalationProject(t, srv.URL, 1.0, true, "adult") + r, err := NewRunner(perm, obs.NewLogger()) + if err != nil { + t.Fatalf("channel=adult on a permissive provider must load, got: %v", err) + } + r.Close() +} diff --git a/backend/internal/store/ledger.go b/backend/internal/store/ledger.go index dbadeaf..4ae9479 100644 --- a/backend/internal/store/ledger.go +++ b/backend/internal/store/ledger.go @@ -17,9 +17,9 @@ import ( type ReserveResult int const ( - ReserveOK ReserveResult = iota - ReserveDeniedBook // per-book USD ceiling hit - ReserveDeniedDay // daily USD ceiling hit + ReserveOK ReserveResult = iota + ReserveDeniedBook // per-book USD ceiling hit + ReserveDeniedDay // daily USD ceiling hit ) // Reservation is the handle Settle/Release need to undo the admission. @@ -109,6 +109,9 @@ type Checkpoint struct { CostUSD float64 FinishReason string ProviderRequestID string + // Escalation marks a single-hop fallback-draft checkpoint (D12): summed against + // escalation.budget_usd, independent of the primary translation spend. + Escalation bool } // SettleWithCheckpoint atomically (ONE transaction, one file) converts the @@ -135,12 +138,12 @@ func (s *Store) SettleWithCheckpoint(res Reservation, cost float64, cp Checkpoin INSERT INTO checkpoints ( request_hash, job_id, chunk_idx, attempt, stage, role, model_requested, model_actual, response_text, usage_json, - cost_usd, finish_reason, provider_request_id - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + cost_usd, finish_reason, provider_request_id, escalation + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (request_hash) DO NOTHING`, cp.RequestHash, cp.JobID, cp.ChunkIdx, cp.Attempt, cp.Stage, cp.Role, cp.ModelRequested, cp.ModelActual, cp.ResponseText, cp.UsageJSON, - cp.CostUSD, cp.FinishReason, cp.ProviderRequestID) + cp.CostUSD, cp.FinishReason, cp.ProviderRequestID, cp.Escalation) if err != nil { return fmt.Errorf("store: checkpoint insert: %w", err) } @@ -197,3 +200,19 @@ func (s *Store) SpentUSD(bookID string) (committed, reserved float64, err error) bookID).Scan(&committed, &reserved) return } + +// EscalationSpentUSD sums the settled cost of a book's escalation (fallback-draft) +// checkpoints — the money-path figure the runner gates against escalation.budget_usd +// (D12). It joins checkpoints→jobs by book_id and counts only escalation=1 rows, so +// it is durable across resume and never double-counts a checkpoint-hit (a resumed +// hop re-serves its checkpoint without a new settle). +func (s *Store) EscalationSpentUSD(bookID string) (float64, error) { + ctx, cancel := opContext() + defer cancel() + var sum float64 + err := s.r.QueryRowContext(ctx, ` + SELECT COALESCE(SUM(c.cost_usd), 0) + FROM checkpoints c JOIN jobs j ON c.job_id = j.id + WHERE j.book_id = ? AND c.escalation = 1`, bookID).Scan(&sum) + return sum, err +} diff --git a/backend/internal/store/migrate.go b/backend/internal/store/migrate.go index d7acc6d..efa7750 100644 --- a/backend/internal/store/migrate.go +++ b/backend/internal/store/migrate.go @@ -126,6 +126,13 @@ var migrations = []string{ ); CREATE INDEX IF NOT EXISTS chunk_status_book_idx ON chunk_status (book_id, disposition); `, + // v3: escalation flag on checkpoints (Веха 2.5, D12 single-hop). The fallback + // draft (a deterministic content-failure routed to another model ONCE) is tagged + // so the book's escalation.budget_usd can be enforced by summing ONLY escalation + // spend, independent of the primary translation cost. On the checkpoint (written + // in the same tx as the settle), NOT request_log, so the sum is durable across + // resume / kill -9 (telemetry is fire-and-forget; money-path budgeting is not). + `ALTER TABLE checkpoints ADD COLUMN escalation INTEGER NOT NULL DEFAULT 0;`, } // migrate runs all pending migrations on the write pool, one transaction per diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index 9566b87..2810990 100644 --- a/docs/PROGRESS.md +++ b/docs/PROGRESS.md @@ -322,7 +322,19 @@ keep-alive (Ф1–2), инъекция глоссария (`selective`), пор `reasoning_content`-vs-`content`, cache-поля, `content_filter`-эмиссия требуют **платного** completion → не гоняю здесь; проверит live-conformance харнесс (C) при ручном прогоне оператором. Код уже корректен по докам (`httpllm.go` читает `content`, парсит оба cache-варианта DeepSeek). -Дальше: B (single-hop эскалация) → C (live-conformance). +### 2026-07-04 — Веха 2.5 (B): single-hop эскалация детерминированных провалов + +Реализован класс «детерминированный контент-провал» из D12: флаг, который другая модель может починить (`cjk_artifact`/`excision_suspect`/`loop_degenerate`/`hard_refusal`/`soft_refusal`/`content_filter` — `FlagReason.escalatable()`), эскалирует **на другую модель ОДИН раз**, а не флажок. + +**Механика (раннер):** после primary-attempt-loop, если вердикт escalatable И у стадии задан `escalate_to` И бюджет остаётся — один вызов `runAttempt(escalate_to, escalation=true)`, результат **ре-гейтится** (`classifyOutput` гоняется и на фолбэке). Прошёл → фолбэк авторитетный (его чекпоинт = `final_hash`); не прошёл → primary-флаг остаётся, фолбэк оплачен и посчитан. **Ось эскалации = модель в `request_hash`** (не attempt): чекпоинт фолбэка `(model=escalate_to, attempt=0)` не коллизится с провалившимся primary. **Ровно 1 хоп** (не свежий retry-бюджет): вызовов/чанк = primary attempts + 1 (guard LiteLLM #19985). `runAttempt` параметризован `model`+`escalation`. + +**Бюджет (D12):** `escalation.budget_usd` — opt-in (0 = выкл, боевой дефолт). Money-path-durable: миграция v3 добавляет колонку `escalation` на checkpoints (пишется в той же tx, что settle), `EscalationSpentUSD` суммирует только escalation-чекпоинты (resume-safe, не двойной счёт на checkpoint-hit). Soft cap. + +**Канал B (D4.1) типом, не комментарием:** `Provider.permissive` + `Stage.channel` (sfw|adult); `LoadPipeline` валит `channel=adult`, если primary или `escalate_to` НЕ на пермиссив-провайдере. **Editor pinned per book:** editor не имеет `escalate_to` → не эскалирует (стиль не плывёт на чужую модель, 2605.13368). **D5.2-закрытие:** `escalate_to` + его резолвнутая каппа свёрнуты в `stageSnap` — смена эскалационной модели/каппы = громкий `--resnapshot`. `CheckKeys` префлайтит ключи `escalate_to` при budget>0. `escalate_to == primary` запрещён валидацией. + +Тесты (мок-провайдер, два fake-модели): эхо→фолбэк-починка→OK+edit; фолбэк тоже эхнул→primary-флаг+skip (ровно 2 вызова); budget=0→не эскалирует; resume подаёт фолбэк-чекпоинт за $0; смена fallback→`--resnapshot`; channel=adult без permissive→fail-fast. Всё зелёное `-race`, `tmctl report` $0. + +Дальше: C (live-conformance харнесс, build-tag `live`). ## Полигон (секция параллельной сессии — записи добавлять сюда)