Add single-hop escalation routing a deterministic content-failure to a named fallback model once, re-gated, budget-capped, with channel-B isolation by type
This commit is contained in:
Claude (backend session)
parent
6dc8f13c6c
commit
5cbf75cae8
8 changed files with 497 additions and 37 deletions
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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`).
|
||||
|
||||
## Полигон
|
||||
(секция параллельной сессии — записи добавлять сюда)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue