diff --git a/backend/cmd/tmctl/main.go b/backend/cmd/tmctl/main.go index 527731c..09f0f9c 100644 --- a/backend/cmd/tmctl/main.go +++ b/backend/cmd/tmctl/main.go @@ -31,6 +31,7 @@ func run() error { fs := flag.NewFlagSet(cmd, flag.ExitOnError) 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 { return err } @@ -43,11 +44,16 @@ func run() error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - ctx = obs.WithReqInfo(ctx, obs.ReqInfo{TraceID: obs.NewTraceID()}) + // LogBodies решается один раз на допуске (privacy by default): содержимое + // LLM-обменов попадает в логи только при LOG_LLM_BODIES=1 И LOG_LEVEL=debug. + ctx = obs.WithReqInfo(ctx, obs.ReqInfo{ + TraceID: obs.NewTraceID(), + LogBodies: os.Getenv("LOG_LLM_BODIES") == "1", + }) switch cmd { case "translate": - return translate(ctx, *cfgPath) + return translate(ctx, *cfgPath, *resnapshot) case "report": return report(*cfgPath) default: @@ -55,12 +61,13 @@ func run() error { } } -func translate(ctx context.Context, cfgPath string) error { +func translate(ctx context.Context, cfgPath string, resnapshot bool) error { r, err := pipeline.NewRunner(cfgPath, obs.NewLogger()) if err != nil { return err } defer r.Close() + r.Resnapshot = resnapshot res, err := r.TranslateOneChunk(ctx) if err != nil { @@ -147,7 +154,12 @@ func loadDotEnv(path string) { if !ok { continue } - k, v = strings.TrimSpace(k), strings.Trim(strings.TrimSpace(v), `"'`) + k, v = strings.TrimSpace(k), strings.TrimSpace(v) + // Снимаем только ПАРНЫЕ обрамляющие кавычки: Trim по набору символов + // откусил бы легитимную кавычку в конце ключа (находка ревью). + if n := len(v); n >= 2 && (v[0] == '"' || v[0] == '\'') && v[n-1] == v[0] { + v = v[1 : n-1] + } if os.Getenv(k) == "" { os.Setenv(k, v) } diff --git a/backend/internal/config/models.go b/backend/internal/config/models.go index 3072900..455f7b4 100644 --- a/backend/internal/config/models.go +++ b/backend/internal/config/models.go @@ -39,7 +39,11 @@ type Provider struct { 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"` - Timeouts Timeouts `yaml:"timeouts"` + // Temperature (local kind): override запроса — ollama чтит request- + // температуру поверх Modelfile, и температура облачной роли молча + // расстроила бы локальную модель. 0 = наследовать запрос. + Temperature float64 `yaml:"temperature"` + Timeouts Timeouts `yaml:"timeouts"` } // Timeouts is the retry profile per provider (профиль per-провайдер из @@ -154,9 +158,18 @@ func LoadModels(path string) (*Models, error) { return &m, nil } -// Prices builds the ledger pricer from the config. +// Prices builds the ledger pricer from the config. Besides the model entries, +// each LOCAL provider's own backend tag is registered at $0: адаптер локальной +// ноги отдаёт LLMResponse.Model = свой тег (huihui_ai/qwen3-…), и без этой +// записи PriceFor свалился бы на платный default-якорь — бесплатные локальные +// вызовы книжились бы по облачной цене, фантомно съедая потолки. func (m *Models) Prices() (*ledger.Pricer, error) { - prices := make(map[string]ledger.ModelPrice, len(m.Models)) + prices := make(map[string]ledger.ModelPrice, len(m.Models)+len(m.Providers)) + for _, p := range m.Providers { + if p.Kind == "local" && p.Model != "" { + prices[p.Model] = ledger.ModelPrice{} + } + } for name, mod := range m.Models { prices[name] = mod.Price.ToLedger() } diff --git a/backend/internal/ledger/pricing.go b/backend/internal/ledger/pricing.go index f9bc7ca..5e15dca 100644 --- a/backend/internal/ledger/pricing.go +++ b/backend/internal/ledger/pricing.go @@ -75,9 +75,17 @@ func CostUSD(price ModelPrice, u llm.Usage) float64 { } // EstimateUSD is the pre-call reservation estimate: assume the whole prompt -// misses the cache and the completion runs to maxTokens. Deliberately -// pessimistic — the reservation is released down to the real cost at settle. +// misses the cache — or, worse, is entirely WRITTEN to the cache (Anthropic +// write ×1.25/×2 дороже input) — and the completion runs to maxTokens. +// Deliberately pessimistic: реальная цена не должна превышать резерв, иначе +// settle пробивает потолок, пропущенный на допуске. Известный остаточный +// случай: additive reasoning-токены (xAI) биллятся ПОВЕРХ completion и оценкой +// не покрыты — допустимый овершут ≤ одной резервации, как у донора. func EstimateUSD(price ModelPrice, promptTokens, maxTokens int) float64 { - return float64(promptTokens)*price.InputPerM/1_000_000 + + inPerM := price.InputPerM + if price.CacheWritePerM > inPerM { + inPerM = price.CacheWritePerM + } + return float64(promptTokens)*inPerM/1_000_000 + float64(maxTokens)*price.OutputPerM/1_000_000 } diff --git a/backend/internal/llm/httpllm.go b/backend/internal/llm/httpllm.go index 4a384e8..46a8033 100644 --- a/backend/internal/llm/httpllm.go +++ b/backend/internal/llm/httpllm.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "log/slog" @@ -24,9 +25,16 @@ import ( // константы: захардкоженный 60-секундный per-attempt дедлайн донора убивал // бы реальные чанки 63–278 с и трижды бесплатно их ретраил; // - Retry-After на 429 читается и уважается (донор капил backoff на 8 с — -// молоток по rate-limit'ам массового прогона глав); +// молоток по rate-limit'ам массового прогона глав), но сам капится +// maxRetryAfterWait — иначе враждебный «Retry-After: 7200» вешает стадию +// на часы с удержанным резервом; // - тело ответа читается через LimitReader (у донора кап был только в // gemini-native адаптере). +// +// Задокументированное отступление от донора: one-shot self-heal параметра +// reasoning_effort (модель, вернувшая 400, запоминалась и параметр срезался) +// НЕ портирован — у нас reasoning задаётся per-стадия в конфиге с fail-fast +// валидацией, и 400 от провайдера должен чиниться в конфиге, а не глотаться. // RetryProfile bounds one client's retry loop. Zero fields fall back to // defaults; the profile comes from models.yaml per provider (per-role @@ -62,6 +70,59 @@ func (p RetryProfile) withDefaults() RetryProfile { // keeping a misbehaving endpoint from exhausting memory. const maxResponseBytes = 16 << 20 +// maxRetryAfterWait bounds an honoured Retry-After: the provider's hint wins +// over our backoff schedule, but never for longer than this — a stage holding +// a USD reservation must stay interruptible-by-timeout, not wedged for hours. +const maxRetryAfterWait = 5 * time.Minute + +// retryLoop is the ONE retry engine shared by every transport (OpenAI-compat +// and the native Anthropic adapter): exponential backoff with cap, Retry-After +// override, jitter, ctx-done select. Один экземпляр политики — чтобы фикс +// ретраев не «уезжал» от одного провайдера, забыв другой. +func retryLoop[T any](ctx context.Context, profile RetryProfile, name string, log *slog.Logger, attempt func() (T, bool, error)) (T, error) { + var zero T + var lastErr error + for att := 0; att < profile.MaxAttempts; att++ { + if att > 0 { + shift := att - 1 + if shift > 20 { + shift = 20 // защита сдвига от переполнения при больших max_attempts + } + backoff := profile.BackoffBase << uint(shift) + if backoff <= 0 || backoff > profile.BackoffCap { + backoff = profile.BackoffCap + } + if ra := retryAfterOf(lastErr); ra > 0 { + if ra > maxRetryAfterWait { + ra = maxRetryAfterWait + } + backoff = ra + } + backoff += time.Duration(rand.Intn(250)) * time.Millisecond + select { + case <-ctx.Done(): + return zero, ctx.Err() + case <-time.After(backoff): + } + } + resp, retryable, err := attempt() + if err == nil { + return resp, nil + } + lastErr = err + if ctx.Err() != nil { + return zero, ctx.Err() + } + if !retryable { + return zero, err + } + if log != nil { + log.WarnContext(ctx, name+" attempt failed, will retry", "attempt", att+1, "max", profile.MaxAttempts, "err", err) + } + } + return zero, fmt.Errorf("%s: exhausted %d attempts: %w", name, profile.MaxAttempts, lastErr) +} + // openAIClient performs OpenAI-compatible /chat/completions calls with retry. type openAIClient struct { name string // provider label for logs/errors ("deepseek", "glm", "local") @@ -99,11 +160,14 @@ type openAIMessage struct { } type openAIRequest struct { - Model string `json:"model"` - Messages []openAIMessage `json:"messages"` - MaxTokens int `json:"max_tokens"` - Temperature float64 `json:"temperature,omitempty"` - Stream bool `json:"stream"` + Model string `json:"model"` + Messages []openAIMessage `json:"messages"` + MaxTokens int `json:"max_tokens"` + // Temperature is ALWAYS sent (no omitempty): дропнутый explicit T=0 + // молча превращал детерминированную стадию (судья C2) в сэмплирование + // провайдерским дефолтом ~1.0. + Temperature float64 `json:"temperature"` + Stream bool `json:"stream"` // Optional; omitempty keeps the plain body minimal. ReasoningEffort string `json:"reasoning_effort,omitempty"` ResponseFormat any `json:"response_format,omitempty"` @@ -200,43 +264,9 @@ func (c *openAIClient) complete(ctx context.Context, reqBody openAIRequest) (*op if err != nil { return nil, err } - - var lastErr error - for attempt := 0; attempt < c.profile.MaxAttempts; attempt++ { - if attempt > 0 { - backoff := c.profile.BackoffBase << uint(attempt-1) - if backoff > c.profile.BackoffCap { - backoff = c.profile.BackoffCap - } - // A server-provided Retry-After overrides our schedule: the provider - // knows its rate-limit window better than our exponential guess. - if ra := retryAfterOf(lastErr); ra > 0 { - backoff = ra - } - backoff += time.Duration(rand.Intn(250)) * time.Millisecond - select { - case <-ctx.Done(): - return nil, ctx.Err() - case <-time.After(backoff): - } - } - - resp, retryable, err := c.attempt(ctx, payload) - if err == nil { - return resp, nil - } - lastErr = err - if ctx.Err() != nil { - return nil, ctx.Err() - } - if !retryable { - return nil, err - } - if c.log != nil { - c.log.WarnContext(ctx, c.name+" attempt failed, will retry", "attempt", attempt+1, "max", c.profile.MaxAttempts, "err", err) - } - } - return nil, fmt.Errorf("%s: exhausted %d attempts: %w", c.name, c.profile.MaxAttempts, lastErr) + return retryLoop(ctx, c.profile, c.name, c.log, func() (*openAIResponse, bool, error) { + return c.attempt(ctx, payload) + }) } // attempt performs one HTTP call. Returns retryable=true for 429/5xx and @@ -280,7 +310,11 @@ func (c *openAIClient) attempt(ctx context.Context, payload []byte) (*openAIResp var out openAIResponse if err := json.Unmarshal(data, &out); err != nil { - return nil, false, fmt.Errorf("%s decode: %w", c.name, err) + // 2xx с нечитаемым телом (обрыв/прокси/обрезка LimitReader): провайдер + // УЖЕ списал деньги. Retryable (повтор может прийти целым); типизировано, + // чтобы раннер на исчерпании попыток НЕ вернул резерв, а консервативно + // заселтлил оценку — недоучёт хуже переучёта для потолков. + return nil, true, &BilledDecodeError{Provider: c.name, Err: err} } // A 2xx is a billed call even when the model returns empty content // (content filter, finish_reason=length with no text). Return it as a @@ -305,13 +339,50 @@ func (e *HTTPStatusError) Error() string { return fmt.Sprintf("%s http %d: %s", e.Provider, e.Status, e.Body) } +// BilledDecodeError is a 2xx whose body could not be decoded: the provider has +// billed the call, but usage/text are unknown. The runner treats an exhausted +// retry chain ending in this error as BILLED (settle at the reservation +// estimate) rather than releasing the reservation. +type BilledDecodeError struct { + Provider string + Err error +} + +func (e *BilledDecodeError) Error() string { + return fmt.Sprintf("%s: 2xx body decode failed (call IS billed): %v", e.Provider, e.Err) +} + +func (e *BilledDecodeError) Unwrap() error { return e.Err } + func retryAfterOf(err error) time.Duration { - if se, ok := err.(*HTTPStatusError); ok { + var se *HTTPStatusError + if errors.As(err, &se) { return se.RetryAfter } return 0 } +// toOpenAIMessages maps neutral messages onto the wire. CacheBoundary is +// meaningless here: OpenAI-compatible caches (DeepSeek: prefix-match с 0-го +// токена блоками по 64 токена) are automatic; the stable-prefix ORDER the +// assembler produced is all that matters. +func toOpenAIMessages(msgs []Message) []openAIMessage { + out := make([]openAIMessage, len(msgs)) + for i, m := range msgs { + out[i] = openAIMessage{Role: m.Role, Content: m.Content} + } + return out +} + +// jsonResponseFormat returns the response_format value for JSONOnly requests +// (nil otherwise, so the field serializes away). +func jsonResponseFormat(jsonOnly bool) any { + if jsonOnly { + return map[string]string{"type": "json_object"} + } + return nil +} + func parseRetryAfter(v string) time.Duration { if v == "" { return 0 diff --git a/backend/internal/llm/httpllm_test.go b/backend/internal/llm/httpllm_test.go index 6cfccb9..967a1e8 100644 --- a/backend/internal/llm/httpllm_test.go +++ b/backend/internal/llm/httpllm_test.go @@ -3,6 +3,7 @@ package llm import ( "context" "encoding/json" + "errors" "net/http" "net/http/httptest" "sync/atomic" @@ -163,7 +164,7 @@ func TestTerminal4xxNoRetry(t *testing.T) { t.Fatal("terminal 4xx must fail") } var se *HTTPStatusError - if !asHTTPStatus(err, &se) || se.Status != 400 { + if !errors.As(err, &se) || se.Status != 400 { t.Fatalf("want typed 400, got %v", err) } if calls.Load() != 1 { @@ -210,17 +211,23 @@ func TestRetryOn5xxThenSuccess(t *testing.T) { } } -func asHTTPStatus(err error, target **HTTPStatusError) bool { - for err != nil { - if se, ok := err.(*HTTPStatusError); ok { - *target = se - return true - } - u, ok := err.(interface{ Unwrap() error }) - if !ok { - return false - } - err = u.Unwrap() +func TestBilledDecodeErrorOn2xxGarbage(t *testing.T) { + // 2xx с нечитаемым телом: ретраится, а на исчерпании отдаёт типизированную + // billed-ошибку — раннер по ней селтлит оценку вместо возврата резерва. + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.Write([]byte(`{"truncated`)) // обрыв тела через прокси + })) + defer srv.Close() + + c := NewOpenAICompatClient(OpenAICompatConfig{Name: "test", BaseURL: srv.URL, Profile: fastProfile()}, nil) + _, err := c.Complete(context.Background(), LLMRequest{Model: "m", Messages: []Message{{Role: "user", Content: "u"}}, MaxTokens: 16}) + var bde *BilledDecodeError + if err == nil || !errors.As(err, &bde) { + t.Fatalf("want BilledDecodeError, got %v", err) + } + if calls.Load() != 3 { + t.Fatalf("garbled 2xx must be retried to exhaustion, calls=%d", calls.Load()) } - return false } diff --git a/backend/internal/llm/provider_anthropic.go b/backend/internal/llm/provider_anthropic.go index df94f9f..650f0eb 100644 --- a/backend/internal/llm/provider_anthropic.go +++ b/backend/internal/llm/provider_anthropic.go @@ -7,9 +7,7 @@ import ( "fmt" "io" "log/slog" - "math/rand" "net/http" - "time" "textmachine/backend/internal/obs" ) @@ -93,12 +91,15 @@ type anthropicMessage struct { Content []anthropicTextBlock `json:"content"` } +// anthropicRequest deliberately carries NO sampling parameters: актуальные +// Claude-модели (Sonnet 5, Opus 4.7+) отклоняют temperature/top_p с 400 — +// управление стилем идёт промптом. Конфигная temperature стадии остаётся в +// request-hash (детерминизм ключа), но на этот wire не попадает. type anthropicRequest struct { - Model string `json:"model"` - MaxTokens int `json:"max_tokens"` - System []anthropicTextBlock `json:"system,omitempty"` - Messages []anthropicMessage `json:"messages"` - Temperature float64 `json:"temperature,omitempty"` + Model string `json:"model"` + MaxTokens int `json:"max_tokens"` + System []anthropicTextBlock `json:"system,omitempty"` + Messages []anthropicMessage `json:"messages"` } type anthropicResponse struct { @@ -126,52 +127,26 @@ func (c *anthropicClient) Complete(ctx context.Context, req LLMRequest) (*LLMRes if err != nil { return nil, err } - - var lastErr error - for attempt := 0; attempt < c.profile.MaxAttempts; attempt++ { - if attempt > 0 { - backoff := c.profile.BackoffBase << uint(attempt-1) - if backoff > c.profile.BackoffCap { - backoff = c.profile.BackoffCap - } - if ra := retryAfterOf(lastErr); ra > 0 { - backoff = ra - } - backoff += time.Duration(rand.Intn(250)) * time.Millisecond - select { - case <-ctx.Done(): - return nil, ctx.Err() - case <-time.After(backoff): - } - } - resp, retryable, err := c.attempt(ctx, payload) - if err == nil { - return resp, nil - } - lastErr = err - if ctx.Err() != nil { - return nil, ctx.Err() - } - if !retryable { - return nil, err - } - if c.log != nil { - c.log.WarnContext(ctx, "anthropic attempt failed, will retry", "attempt", attempt+1, "max", c.profile.MaxAttempts, "err", err) - } - } - return nil, fmt.Errorf("anthropic: exhausted %d attempts: %w", c.profile.MaxAttempts, lastErr) + return retryLoop(ctx, c.profile, "anthropic", c.log, func() (*LLMResponse, bool, error) { + return c.attempt(ctx, payload) + }) } // buildRequest maps the neutral request onto the Messages wire. Leading // system-role messages become system blocks; CacheBoundary flags become -// cache_control on that block. Thinking stays off (Фаза 0: Anthropic — только -// SFW-роли редактора/эскалации, thinking-глубина не нужна и биллится как -// output; ReasoningEffort пока не маппится — задокументированное ограничение). +// cache_control on that block. Thinking is not mapped yet (Фаза 0: Anthropic — +// только SFW-роли редактора/эскалации): ""/"off" проходят (без thinking- +// конфигурации), а low/medium/high падают громко — молча оплаченная стадия БЕЗ +// заказанной глубины рассуждений хуже ошибки конфига. func (c *anthropicClient) buildRequest(req LLMRequest) (*anthropicRequest, error) { + switch req.ReasoningEffort { + case "", "off": + default: + return nil, fmt.Errorf("anthropic: reasoning %q is not mapped by this adapter yet (thinking — Фаза 2)", req.ReasoningEffort) + } wire := &anthropicRequest{ - Model: req.Model, - MaxTokens: req.MaxTokens, - Temperature: req.Temperature, + Model: req.Model, + MaxTokens: req.MaxTokens, } boundaries := 0 block := func(m Message) anthropicTextBlock { @@ -242,7 +217,8 @@ func (c *anthropicClient) attempt(ctx context.Context, payload []byte) (*LLMResp var out anthropicResponse if err := json.Unmarshal(data, &out); err != nil { - return nil, false, fmt.Errorf("anthropic decode: %w", err) + // 2xx уже оплачен провайдером — типизируем, retryable (см. httpllm.go). + return nil, true, &BilledDecodeError{Provider: "anthropic", Err: err} } var text string diff --git a/backend/internal/llm/provider_anthropic_test.go b/backend/internal/llm/provider_anthropic_test.go index 64631a1..b466469 100644 --- a/backend/internal/llm/provider_anthropic_test.go +++ b/backend/internal/llm/provider_anthropic_test.go @@ -68,6 +68,21 @@ func TestAnthropicWireShapeAndUsage(t *testing.T) { if resp.Text != "перевод" || resp.FinishReason != FinishStop || resp.Model != "claude-sonnet-5" { t.Fatalf("resp = %+v", resp) } + // Sampling-параметры на этот wire не отправляются: актуальные Claude-модели + // отклоняют temperature с 400. + if _, has := got["temperature"]; has { + t.Fatal("temperature must not be sent to the Anthropic wire") + } +} + +func TestAnthropicRejectsUnmappedReasoning(t *testing.T) { + c := NewAnthropicClient(AnthropicConfig{BaseURL: "http://unused", APIKey: "k", Profile: fastProfile()}, nil) + _, err := c.Complete(context.Background(), LLMRequest{ + Model: "m", Messages: []Message{{Role: "user", Content: "u"}}, MaxTokens: 16, ReasoningEffort: "high", + }) + if err == nil { + t.Fatal("unmapped reasoning depth must fail loud, not silently run without thinking") + } } func TestAnthropicStopReasonMapping(t *testing.T) { diff --git a/backend/internal/llm/provider_local.go b/backend/internal/llm/provider_local.go index 87c76d7..e11ff8c 100644 --- a/backend/internal/llm/provider_local.go +++ b/backend/internal/llm/provider_local.go @@ -63,10 +63,6 @@ func NewLocalClient(cfg LocalConfig, logger *slog.Logger) LLMClient { } func (c *localClient) Complete(ctx context.Context, req LLMRequest) (*LLMResponse, error) { - msgs := make([]openAIMessage, len(req.Messages)) - for i, m := range req.Messages { - msgs[i] = openAIMessage{Role: m.Role, Content: m.Content} - } maxTok := req.MaxTokens if c.maxTok > 0 { maxTok = c.maxTok @@ -75,10 +71,6 @@ func (c *localClient) Complete(ctx context.Context, req LLMRequest) (*LLMRespons if c.temp > 0 { temp = c.temp } - var respFormat any - if req.JSONOnly { - respFormat = map[string]string{"type": "json_object"} - } effort := req.ReasoningEffort if effort == "off" { // ollama's /v1 maps reasoning_effort onto Qwen3 thinking; "none" is its @@ -87,12 +79,12 @@ func (c *localClient) Complete(ctx context.Context, req LLMRequest) (*LLMRespons } resp, err := c.http.complete(ctx, openAIRequest{ Model: c.model, - Messages: msgs, + Messages: toOpenAIMessages(req.Messages), MaxTokens: maxTok, Temperature: temp, Stream: false, ReasoningEffort: effort, - ResponseFormat: respFormat, + ResponseFormat: jsonResponseFormat(req.JSONOnly), }) if err != nil { return nil, err diff --git a/backend/internal/llm/provider_openai.go b/backend/internal/llm/provider_openai.go index cf6e07a..c611eb6 100644 --- a/backend/internal/llm/provider_openai.go +++ b/backend/internal/llm/provider_openai.go @@ -64,18 +64,6 @@ func NewOpenAICompatClient(cfg OpenAICompatConfig, logger *slog.Logger) LLMClien } func (c *openAICompatClient) Complete(ctx context.Context, req LLMRequest) (*LLMResponse, error) { - msgs := make([]openAIMessage, len(req.Messages)) - for i, m := range req.Messages { - // CacheBoundary is meaningless on this wire: OpenAI-compatible caches - // (DeepSeek: prefix-match с 0-го токена блоками по 64 токена) are - // automatic. The stable-prefix ordering the assembler produced is all - // that matters here. - msgs[i] = openAIMessage{Role: m.Role, Content: m.Content} - } - var respFormat any - if req.JSONOnly { - respFormat = map[string]string{"type": "json_object"} - } effort := req.ReasoningEffort if effort == "off" { // "off" is our neutral value; OpenAI-compat providers with a @@ -84,12 +72,12 @@ func (c *openAICompatClient) Complete(ctx context.Context, req LLMRequest) (*LLM } resp, err := c.http.complete(ctx, openAIRequest{ Model: req.Model, - Messages: msgs, + Messages: toOpenAIMessages(req.Messages), MaxTokens: req.MaxTokens, Temperature: req.Temperature, Stream: false, ReasoningEffort: effort, - ResponseFormat: respFormat, + ResponseFormat: jsonResponseFormat(req.JSONOnly), extra: c.extra, }) if err != nil { diff --git a/backend/internal/obs/logging.go b/backend/internal/obs/logging.go index e408b27..8bebd85 100644 --- a/backend/internal/obs/logging.go +++ b/backend/internal/obs/logging.go @@ -30,15 +30,23 @@ func NewLogger() *slog.Logger { } // contextHandler wraps a slog.Handler so a record logged with one of the -// *Context methods automatically gets the call's trace_id attached — the -// userver-style trail, set once at the top of a unit of work and carried -// through to the model call. Records logged without a traced context simply -// carry no trace_id. +// *Context methods automatically gets the call's trace_id AND the translation +// axis (book/chapter/chunk/stage/role) attached — the userver-style trail, set +// once at the top of a unit of work and carried through to the model call. +// Records logged without a traced context simply carry none of these. type contextHandler struct{ slog.Handler } func (h contextHandler) Handle(ctx context.Context, r slog.Record) error { - if id := TraceFromContext(ctx); id != "" { - r.AddAttrs(slog.String("trace_id", id)) + if ri, ok := ReqInfoFromContext(ctx); ok { + if ri.TraceID != "" { + r.AddAttrs(slog.String("trace_id", ri.TraceID)) + } + if ri.Book != "" { + r.AddAttrs(slog.String("book", ri.Book), slog.Int("chapter", ri.Chapter), slog.Int("chunk", ri.Chunk)) + } + if ri.Stage != "" { + r.AddAttrs(slog.String("stage", ri.Stage), slog.String("role", ri.Role)) + } } return h.Handler.Handle(ctx, r) } diff --git a/backend/internal/pipeline/clients.go b/backend/internal/pipeline/clients.go index 656677e..63362bc 100644 --- a/backend/internal/pipeline/clients.go +++ b/backend/internal/pipeline/clients.go @@ -38,11 +38,12 @@ func BuildClient(models *config.Models, modelName string, logger *slog.Logger) ( }, logger), nil case "local": return llm.NewLocalClient(llm.LocalConfig{ - BaseURL: prov.BaseURL, - APIKey: prov.APIKey(), - Model: prov.Model, - MaxTokens: prov.MaxTokens, - Profile: prov.Timeouts.Profile(), + BaseURL: prov.BaseURL, + APIKey: prov.APIKey(), + Model: prov.Model, + Temperature: prov.Temperature, + MaxTokens: prov.MaxTokens, + Profile: prov.Timeouts.Profile(), }, logger), nil default: return nil, fmt.Errorf("pipeline: provider %q has unknown kind %q", mod.Provider, prov.Kind) diff --git a/backend/internal/pipeline/render.go b/backend/internal/pipeline/render.go index 896a632..a596c3a 100644 --- a/backend/internal/pipeline/render.go +++ b/backend/internal/pipeline/render.go @@ -30,6 +30,12 @@ import ( // (§3.4 — [НУЖНО РЕШЕНИЕ] п.2в). const chunkerVersion = "chunker-v0-wholefile" +// estimatorVersion versions EstimateTokens: его выход входит в max_tokens и +// через него в request-hash, поэтому перекалибровка весов — тоже явная +// инвалидация через snapshot, а не тихий промах всех чекпоинтов (находка +// ревью: code-only правка эстиматора пере-оплатила бы полкниги). +const estimatorVersion = "estimator-v0" + // userSeparator splits a prompt template file into the system part and the // user part. Обе части — версионируемые шаблоны в prompts/ («Редакция» позже // правит файлы, не код). @@ -67,37 +73,57 @@ type RenderVars struct { Draft string // prior stage output ("" on the first stage) } -// Render substitutes placeholders. Unknown {{…}} markers are a hard error — -// молчаливо пустой плейсхолдер в промпте хуже падения. +// Render substitutes placeholders in a SINGLE pass over the template: values +// are inserted verbatim and never re-scanned, so literal «{{…}}» внутри +// исходника или черновика LLM — это просто текст, а не маркер (находка ревью: +// последовательный ReplaceAll подставлял черновик в «{{draft}}» из сорса и +// падал на «{{TN: примечание}}» в выводе модели). Unknown {{…}} markers in +// the TEMPLATE are a hard error — молчаливо пустой плейсхолдер хуже падения. func Render(tpl string, v RenderVars) (string, error) { - pairs := [...][2]string{ - {"{{book_id}}", v.Book.BookID}, - {"{{title}}", v.Book.Title}, - {"{{source_lang}}", v.Book.SourceLang}, - {"{{target_lang}}", v.Book.TargetLang}, - {"{{genre}}", v.Book.Genre}, - {"{{audience}}", v.Book.Audience}, - {"{{venuti}}", strconv.FormatFloat(v.Book.Venuti, 'f', 2, 64)}, - {"{{honorifics}}", v.Book.Honorifics}, - {"{{transcription}}", v.Book.Transcription}, - {"{{footnotes}}", v.Book.Footnotes}, - {"{{text}}", v.Text}, - {"{{draft}}", v.Draft}, + // Lookup-only map (никакой итерации — детерминизм не страдает). + vals := map[string]string{ + "book_id": v.Book.BookID, + "title": v.Book.Title, + "source_lang": v.Book.SourceLang, + "target_lang": v.Book.TargetLang, + "genre": v.Book.Genre, + "audience": v.Book.Audience, + "venuti": strconv.FormatFloat(v.Book.Venuti, 'f', 2, 64), + "honorifics": v.Book.Honorifics, + "transcription": v.Book.Transcription, + "footnotes": v.Book.Footnotes, + "text": v.Text, + "draft": v.Draft, } - out := tpl - for _, p := range pairs { - out = strings.ReplaceAll(out, p[0], p[1]) - } - if i := strings.Index(out, "{{"); i >= 0 { - end := strings.Index(out[i:], "}}") - if end < 0 { - end = len(out) - i - } else { - end += 2 + var b strings.Builder + rest := tpl + for { + i := strings.Index(rest, "{{") + if i < 0 { + b.WriteString(rest) + return b.String(), nil } - return "", fmt.Errorf("pipeline: unknown placeholder %q in template", out[i:i+end]) + b.WriteString(rest[:i]) + rest = rest[i:] + end := strings.Index(rest, "}}") + if end < 0 { + return "", fmt.Errorf("pipeline: unterminated placeholder %q in template", snippetStr(rest)) + } + name := rest[2:end] + val, ok := vals[name] + if !ok { + return "", fmt.Errorf("pipeline: unknown placeholder {{%s}} in template", name) + } + b.WriteString(val) + rest = rest[end+2:] } - return out, nil +} + +func snippetStr(s string) string { + if len(s) > 40 { + return s[:40] + "…" + } + return s } // Messages builds the wire-neutral message list for a stage. Layout по Р5: @@ -122,8 +148,10 @@ func Messages(tpl *PromptTemplate, v RenderVars) ([]llm.Message, error) { // RequestHash is the checkpoint/resume key (§3.1): a stable hash of everything // that determines the call. The snapshot id freezes the volatile context, so // identical re-renders after a restart find their checkpoint and are neither -// repeated nor re-billed. -func RequestHash(bookID string, chapter, chunkIdx int, stage, role, model string, temperature float64, reasoning string, jsonOnly bool, maxTokens int, snapshotID string, msgs []llm.Message) string { +// repeated nor re-billed. attempt — измерение регенераций (Фаза 1: гейт +// провалился → attempt+1 → новый ключ; без него регенерация попадала бы в +// только что провалившийся чекпоинт и мгновенно эскалировала в премиум). +func RequestHash(bookID string, chapter, chunkIdx, attempt int, stage, role, model string, temperature float64, reasoning string, jsonOnly bool, maxTokens int, snapshotID string, msgs []llm.Message) string { h := sha256.New() w := func(parts ...string) { for _, p := range parts { @@ -131,7 +159,7 @@ func RequestHash(bookID string, chapter, chunkIdx int, stage, role, model string h.Write([]byte{0}) } } - w("tm-request-v1", bookID, strconv.Itoa(chapter), strconv.Itoa(chunkIdx), stage, role, model, + w("tm-request-v1", bookID, strconv.Itoa(chapter), strconv.Itoa(chunkIdx), strconv.Itoa(attempt), stage, role, model, strconv.FormatFloat(temperature, 'f', -1, 64), reasoning, strconv.FormatBool(jsonOnly), strconv.Itoa(maxTokens), snapshotID) for _, m := range msgs { diff --git a/backend/internal/pipeline/render_test.go b/backend/internal/pipeline/render_test.go index a93d40c..919e863 100644 --- a/backend/internal/pipeline/render_test.go +++ b/backend/internal/pipeline/render_test.go @@ -3,6 +3,7 @@ package pipeline import ( "os" "path/filepath" + "strings" "testing" "textmachine/backend/internal/config" @@ -55,8 +56,8 @@ func TestRenderIsDeterministic(t *testing.T) { } } } - h1 := RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", m1) - h2 := RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", m1) + h1 := RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", m1) + h2 := RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", m1) if h1 != h2 { t.Fatal("request hash is not stable") } @@ -64,15 +65,15 @@ func TestRenderIsDeterministic(t *testing.T) { func TestRequestHashSensitivity(t *testing.T) { msgs := []llm.Message{{Role: "system", Content: "s", CacheBoundary: true}, {Role: "user", Content: "u"}} - base := RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", msgs) + base := RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", msgs) variants := map[string]string{ - "model": RequestHash("b", 1, 0, "draft", "translator", "m2", 0.3, "off", false, 1024, "snap", msgs), - "temp": RequestHash("b", 1, 0, "draft", "translator", "m", 0.4, "off", false, 1024, "snap", msgs), - "snapshot": RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap2", msgs), - "maxTokens": RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 2048, "snap", msgs), - "chunk": RequestHash("b", 1, 1, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", msgs), - "content": RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", []llm.Message{{Role: "system", Content: "s", CacheBoundary: true}, {Role: "user", Content: "u2"}}), + "model": RequestHash("b", 1, 0, 0, "draft", "translator", "m2", 0.3, "off", false, 1024, "snap", msgs), + "temp": RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.4, "off", false, 1024, "snap", msgs), + "snapshot": RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap2", msgs), + "maxTokens": RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 2048, "snap", msgs), + "chunk": RequestHash("b", 1, 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", msgs), + "content": RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", []llm.Message{{Role: "system", Content: "s", CacheBoundary: true}, {Role: "user", Content: "u2"}}), } for name, h := range variants { if h == base { @@ -81,9 +82,9 @@ func TestRequestHashSensitivity(t *testing.T) { } // Конкатенация с разделителем не склеивается: ("ab","c") != ("a","bc"). - a := RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", + a := RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", []llm.Message{{Role: "user", Content: "ab"}, {Role: "user", Content: "c"}}) - b := RequestHash("b", 1, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", + b := RequestHash("b", 1, 0, 0, "draft", "translator", "m", 0.3, "off", false, 1024, "snap", []llm.Message{{Role: "user", Content: "a"}, {Role: "user", Content: "bc"}}) if a == b { t.Fatal("field separator is not injective") @@ -98,6 +99,27 @@ func TestRenderRejectsUnknownPlaceholder(t *testing.T) { } } +// Литеральные «{{…}}» в ИСХОДНИКЕ/черновике — просто текст: значения +// вставляются verbatim и не пере-сканируются (однопроходный рендер). +func TestRenderValuesAreNotRescanned(t *testing.T) { + tpl := writeTemplate(t, "Система.\n---USER---\nТекст: {{text}}\nЧерновик: {{draft}}") + v := RenderVars{ + Book: testBook(), + Text: "герой сказал {{draft}} и ушёл", // literal braces in source + Draft: "примечание {{TN: сноска}} осталось", // literal braces in prior LLM output + } + msgs, err := Messages(tpl, v) + if err != nil { + t.Fatalf("literal braces inside values must not fail: %v", err) + } + user := msgs[1].Content + for _, want := range []string{"герой сказал {{draft}} и ушёл", "примечание {{TN: сноска}} осталось"} { + if !strings.Contains(user, want) { + t.Fatalf("value %q must survive verbatim, got:\n%s", want, user) + } + } +} + func TestTemplateRequiresUserSeparator(t *testing.T) { path := filepath.Join(t.TempDir(), "bad.md") if err := os.WriteFile(path, []byte("только системная часть"), 0o644); err != nil { diff --git a/backend/internal/pipeline/runner.go b/backend/internal/pipeline/runner.go index e258aad..befa92c 100644 --- a/backend/internal/pipeline/runner.go +++ b/backend/internal/pipeline/runner.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "log/slog" "os" @@ -32,6 +33,11 @@ type Runner struct { Store *store.Store Pricer *ledger.Pricer Log *slog.Logger + // Resnapshot re-pins existing jobs to the CURRENT snapshot when configs/ + // prompts changed since the job started (tmctl --resnapshot). Без флага + // расхождение — громкая ошибка: смена контекста инвалидирует чекпоинты и + // пере-оплачивает вызовы, это делается только осознанно (Р6). + Resnapshot bool clients map[string]llm.LLMClient templates map[string]*PromptTemplate @@ -117,14 +123,16 @@ func (r *Runner) snapshotID() (id, payload string, err error) { Reasoning string `json:"reasoning"` } snap := struct { - BriefHash string `json:"brief_hash"` - ChunkerVersion string `json:"chunker_version"` - PipelineCore string `json:"pipeline_core"` - Stages []stageSnap `json:"stages"` + BriefHash string `json:"brief_hash"` + ChunkerVersion string `json:"chunker_version"` + EstimatorVersion string `json:"estimator_version"` + PipelineCore string `json:"pipeline_core"` + Stages []stageSnap `json:"stages"` }{ - BriefHash: r.Book.BriefHash(), - ChunkerVersion: chunkerVersion, - PipelineCore: r.Pipeline.Core, + BriefHash: r.Book.BriefHash(), + ChunkerVersion: chunkerVersion, + EstimatorVersion: estimatorVersion, + PipelineCore: r.Pipeline.Core, } for _, st := range r.Pipeline.Stages { snap.Stages = append(snap.Stages, stageSnap{ @@ -207,6 +215,19 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c if err != nil { return nil, err } + // Snapshot pinning (Р6): джоба заморожена на снапшоте своего старта. Если + // конфиги/промпты с тех пор изменились — это инвалидация чекпоинтов и + // пере-оплата; выполняем её только по явной команде. + 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) + } + 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]) + } if err := r.Store.SetJobStatus(job.ID, "running"); err != nil { return nil, err } @@ -225,19 +246,27 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c maxTokens = r.Pipeline.Defaults.MinMaxTokens } - reqHash := RequestHash(r.Book.BookID, chapter, chunkIdx, st.Name, st.Role, st.Model, + // 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) - ctx = obs.WithReqInfo(ctx, obs.ReqInfo{ - TraceID: obs.TraceFromContext(ctx), Book: r.Book.BookID, - Chapter: chapter, Chunk: chunkIdx, Stage: st.Name, Role: st.Role, - }) + // Дополняем 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 + 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{ @@ -289,6 +318,27 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c }) latency := int(time.Since(start).Milliseconds()) if err != nil { + // 2xx с нечитаемым телом: провайдер УЖЕ списал деньги — резерв не + // возвращаем, консервативно селтлим ОЦЕНКУ с пустым чекпоинтом + // (переучёт ≤1 оценки безопаснее слепого потолка) и падаем громко. + var bde *llm.BilledDecodeError + if errors.As(err, &bde) { + if serr := r.Store.SettleWithCheckpoint(resv, estimate, store.Checkpoint{ + RequestHash: reqHash, JobID: job.ID, ChunkIdx: chunkIdx, Attempt: attempt, + Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: st.Model, + ResponseText: "", UsageJSON: "{}", CostUSD: estimate, FinishReason: "decode_error", + }); serr != nil { + r.Log.ErrorContext(ctx, "settle after billed decode failure also failed", "err", serr) + } + 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, + RequestHash: reqHash, CostUSD: estimate, LatencyMS: latency, + Degraded: "billed_2xx_decode_failed", Err: err.Error(), OK: false, + }) + _ = r.Store.SetJobStatus(job.ID, "failed") + return nil, fmt.Errorf("pipeline: stage %s: %w (оценка $%.6f заселтлена консервативно)", st.Name, err, estimate) + } // No 2xx ever arrived: nothing was billed by the provider — release the // reservation and surface the failure with a telemetry row. if rerr := r.Store.Release(resv); rerr != nil { @@ -315,7 +365,8 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c // Деньги: settle + сырой ответ — одна транзакция (§3.3). if err := r.Store.SettleWithCheckpoint(resv, cost, store.Checkpoint{ - RequestHash: reqHash, JobID: job.ID, ChunkIdx: chunkIdx, Stage: st.Name, Role: st.Role, + RequestHash: reqHash, JobID: job.ID, ChunkIdx: 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, ProviderRequestID: resp.ProviderRequestID, @@ -325,6 +376,25 @@ func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, c return nil, 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) + } + 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, diff --git a/backend/internal/pipeline/runner_test.go b/backend/internal/pipeline/runner_test.go index b1c0bef..c081fe3 100644 --- a/backend/internal/pipeline/runner_test.go +++ b/backend/internal/pipeline/runner_test.go @@ -168,6 +168,63 @@ func TestRunnerEndToEndWithResume(t *testing.T) { } } +// Р6: правка конфига/промпта после старта джоб — не тихая инвалидация +// чекпоинтов, а громкая ошибка; пере-перевод только явным --resnapshot. +func TestRunnerSnapshotPinning(t *testing.T) { + var calls atomic.Int32 + srv := newFakeProvider(t, &calls) + 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 { + t.Fatal(err) + } + r1.Close() + if calls.Load() != 2 { + t.Fatalf("run1 calls = %d", calls.Load()) + } + + // Меняем промпт переводчика → новый 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.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()) + } + + // Явное согласие: джобы перепривязываются, вызовы повторяются и оплачиваются. + r3, err := NewRunner(bookPath, obs.NewLogger()) + if err != nil { + t.Fatal(err) + } + defer r3.Close() + r3.Resnapshot = true + res, err := r3.TranslateOneChunk(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 res.TotalUSD <= 0 { + t.Fatal("re-translation must be billed") + } +} + func TestRunnerCeilingDenies(t *testing.T) { var calls atomic.Int32 srv := newFakeProvider(t, &calls) diff --git a/backend/internal/store/jobs.go b/backend/internal/store/jobs.go index c9468a1..5cbe796 100644 --- a/backend/internal/store/jobs.go +++ b/backend/internal/store/jobs.go @@ -71,3 +71,14 @@ func (s *Store) SetJobStatus(jobID int64, status string) error { `UPDATE jobs SET status = ?, updated_at = datetime('now') WHERE id = ?`, status, jobID) return err } + +// UpdateJobSnapshot re-pins a job to a new snapshot — ЯВНАЯ команда оператора +// (tmctl --resnapshot): пере-перевод уже оплаченных чанков случается только +// осознанно, никогда как тихий побочный эффект правки конфига (Р6). +func (s *Store) UpdateJobSnapshot(jobID int64, snapshotID string) error { + ctx, cancel := opContext() + defer cancel() + _, err := s.w.ExecContext(ctx, + `UPDATE jobs SET snapshot_id = ?, updated_at = datetime('now') WHERE id = ?`, snapshotID, jobID) + return err +} diff --git a/backend/internal/store/ledger.go b/backend/internal/store/ledger.go index e7edc4d..dbadeaf 100644 --- a/backend/internal/store/ledger.go +++ b/backend/internal/store/ledger.go @@ -99,6 +99,7 @@ type Checkpoint struct { RequestHash string JobID int64 ChunkIdx int + Attempt int Stage string Role string ModelRequested string @@ -132,12 +133,12 @@ func (s *Store) SettleWithCheckpoint(res Reservation, cost float64, cp Checkpoin ins, err := tx.ExecContext(ctx, ` INSERT INTO checkpoints ( - request_hash, job_id, chunk_idx, stage, role, + 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (request_hash) DO NOTHING`, - cp.RequestHash, cp.JobID, cp.ChunkIdx, cp.Stage, cp.Role, + 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) if err != nil { @@ -173,10 +174,10 @@ func (s *Store) GetCheckpoint(requestHash string) (*Checkpoint, error) { defer cancel() cp := Checkpoint{RequestHash: requestHash} err := s.r.QueryRowContext(ctx, ` - SELECT job_id, chunk_idx, stage, role, model_requested, model_actual, + SELECT job_id, chunk_idx, attempt, stage, role, model_requested, model_actual, response_text, usage_json, cost_usd, finish_reason, provider_request_id FROM checkpoints WHERE request_hash = ?`, requestHash).Scan( - &cp.JobID, &cp.ChunkIdx, &cp.Stage, &cp.Role, &cp.ModelRequested, &cp.ModelActual, + &cp.JobID, &cp.ChunkIdx, &cp.Attempt, &cp.Stage, &cp.Role, &cp.ModelRequested, &cp.ModelActual, &cp.ResponseText, &cp.UsageJSON, &cp.CostUSD, &cp.FinishReason, &cp.ProviderRequestID) if errors.Is(err, sql.ErrNoRows) { return nil, nil diff --git a/backend/internal/store/migrate.go b/backend/internal/store/migrate.go index 3e97ac5..f26a37e 100644 --- a/backend/internal/store/migrate.go +++ b/backend/internal/store/migrate.go @@ -46,6 +46,7 @@ var migrations = []string{ request_hash TEXT PRIMARY KEY, job_id INTEGER NOT NULL REFERENCES jobs(id), chunk_idx INTEGER NOT NULL, + attempt INTEGER NOT NULL DEFAULT 0, -- регенерации Фазы 1: attempt входит в request-hash stage TEXT NOT NULL, role TEXT NOT NULL, model_requested TEXT NOT NULL, diff --git a/backend/internal/store/requestlog.go b/backend/internal/store/requestlog.go index 813f2ab..13f5119 100644 --- a/backend/internal/store/requestlog.go +++ b/backend/internal/store/requestlog.go @@ -55,7 +55,7 @@ func (s *Store) InsertRequestLog(rl RequestLog) error { rl.ModelRequested, rl.ModelActual, rl.RequestHash, rl.PromptTokens, rl.CachedTokens, rl.CacheCreationTokens, rl.CompletionTokens, rl.ReasoningTokens, - rl.CostUSD, rl.LatencyMS, rl.FinishReason, boolToInt(rl.TMHit), rl.Degraded, rl.Err, boolToInt(rl.OK)) + rl.CostUSD, rl.LatencyMS, rl.FinishReason, rl.TMHit, rl.Degraded, rl.Err, rl.OK) return err } @@ -80,10 +80,3 @@ func (s *Store) RequestLogRows(bookID string) (*sql.Rows, error) { cost_usd, latency_ms, finish_reason, tm_hit, ok FROM request_log WHERE book_id = ? ORDER BY id`, bookID) } - -func boolToInt(b bool) int { - if b { - return 1 - } - return 0 -} diff --git a/backend/internal/store/store.go b/backend/internal/store/store.go index ec76279..74a3f00 100644 --- a/backend/internal/store/store.go +++ b/backend/internal/store/store.go @@ -18,6 +18,8 @@ import ( "database/sql" "fmt" "net/url" + "os" + "syscall" "time" _ "modernc.org/sqlite" @@ -30,13 +32,23 @@ const opTimeout = 10 * time.Second // Store is one project database. type Store struct { - w *sql.DB // single-connection write pool; all mutations go through it - r *sql.DB // read pool + w *sql.DB // single-connection write pool; all mutations go through it + r *sql.DB // read pool + lock *os.File } // Open opens (or creates) the project database at path, applies pending // migrations and recovers stale reservations from a crashed run. +// +// Ownership: Open takes an EXCLUSIVE flock on .lock — «один процесс +// владеет файлом проекта» здесь принуждается, а не предполагается. Без замка +// параллельный `tmctl report` во время перевода обнулил бы recovery-проходом +// живые резервы бегущего процесса и ослепил бы его потолки (находка ревью). func Open(path string) (*Store, error) { + lock, err := acquireLock(path + ".lock") + if err != nil { + return nil, err + } dsn := func(txlock string) string { v := url.Values{} v.Add("_pragma", "journal_mode(WAL)") @@ -51,6 +63,7 @@ func Open(path string) (*Store, error) { w, err := sql.Open("sqlite", dsn("immediate")) if err != nil { + releaseLock(lock) return nil, fmt.Errorf("store: open write pool: %w", err) } // One writer connection: SQLite has a single writer per file anyway; @@ -60,11 +73,12 @@ func Open(path string) (*Store, error) { r, err := sql.Open("sqlite", dsn("")) if err != nil { w.Close() + releaseLock(lock) return nil, fmt.Errorf("store: open read pool: %w", err) } r.SetMaxOpenConns(4) - s := &Store{w: w, r: r} + s := &Store{w: w, r: r, lock: lock} ctx, cancel := context.WithTimeout(context.Background(), opTimeout) defer cancel() if err := s.migrate(ctx); err != nil { @@ -81,12 +95,36 @@ func Open(path string) (*Store, error) { func (s *Store) Close() error { err1 := s.w.Close() err2 := s.r.Close() + releaseLock(s.lock) if err1 != nil { return err1 } return err2 } +// acquireLock takes a non-blocking exclusive flock. Kernel releases it on +// process death (kill -9 включительно), so a crashed run never wedges the +// project. +func acquireLock(path string) (*os.File, error) { + f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644) + if err != nil { + return nil, fmt.Errorf("store: open lock file: %w", err) + } + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + f.Close() + return nil, fmt.Errorf("store: project database is in use by another tmctl process (lock %s): %w", path, err) + } + return f, nil +} + +func releaseLock(f *os.File) { + if f == nil { + return + } + _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) + _ = f.Close() +} + func opContext() (context.Context, context.CancelFunc) { return context.WithTimeout(context.Background(), opTimeout) } diff --git a/backend/internal/store/store_test.go b/backend/internal/store/store_test.go index f44d216..ca530b6 100644 --- a/backend/internal/store/store_test.go +++ b/backend/internal/store/store_test.go @@ -188,3 +188,23 @@ func TestMigrateIsIdempotent(t *testing.T) { s.Close() } } + +func TestSingleProcessOwnership(t *testing.T) { + // Второй Open того же проекта обязан отказать: параллельный процесс + // (report во время translate) recovery-проходом обнулил бы живые резервы. + path := filepath.Join(t.TempDir(), "test.db") + s1, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer s1.Close() + if _, err := Open(path); err == nil { + t.Fatal("second Open must be denied while the project is locked") + } + s1.Close() + s2, err := Open(path) + if err != nil { + t.Fatalf("after Close the lock must be free: %v", err) + } + s2.Close() +}