94 lines
4.9 KiB
Go
94 lines
4.9 KiB
Go
package pipeline
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"textmachine/backend/internal/config"
|
||
"textmachine/backend/internal/llm"
|
||
"textmachine/backend/internal/store"
|
||
)
|
||
|
||
// resume.go: обслуживание уже РЕШЁННОГО чанк×стадии из chunk_status/чекпоинтов —
|
||
// $0, без вызова провайдера; ok-стадия отдаёт авторитетный текст следующей стадии,
|
||
// терминальный флаг никогда не переатакуется (анти-ведж №1).
|
||
|
||
// resumeFromChunkStatus serves a chunk×stage whose disposition is already
|
||
// resolved (read before any render — no re-billing). An ok stage returns its
|
||
// authoritative checkpoint text to feed the next stage; a flagged stage returns
|
||
// the flag WITHOUT re-attacking (a terminal flag never re-enters the paid loop).
|
||
func (r *Runner) resumeFromChunkStatus(ctx context.Context, st config.Stage, ch Chunk, cs *store.ChunkStatus) (*StageResult, error) {
|
||
sr := &StageResult{
|
||
Stage: st.Name, Role: st.Role, Model: st.Model, FromResume: true,
|
||
CostUSD: 0, CumCostUSD: cs.CostUSD,
|
||
Disposition: Disposition(cs.Disposition), FlagReason: FlagReason(cs.FlagReason),
|
||
Detail: cs.Detail, Attempts: cs.Attempts,
|
||
// Escalation attribution now travels on chunk_status (D15.3), so it is preserved
|
||
// across resume for BOTH an ok escalation AND a flagged-then-escalated chunk (the
|
||
// status read-model reads it without re-opening the checkpoint).
|
||
Escalated: cs.Escalated, EscalationModel: cs.EscalationModel,
|
||
}
|
||
if cs.Disposition == string(DispOK) {
|
||
cp, err := r.Store.GetCheckpoint(cs.FinalHash)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pipeline: read final checkpoint %.12s for %s/ch%d/chunk%d/%s: %w",
|
||
cs.FinalHash, r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, err)
|
||
}
|
||
if cp == nil || strings.TrimSpace(cp.ResponseText) == "" {
|
||
// chunk_status ok is only written AFTER its checkpoint is durably
|
||
// settled, so this is a torn/tampered store — fail loud rather than
|
||
// feed the next stage an empty draft.
|
||
return nil, fmt.Errorf("pipeline: chunk_status ok for %s/ch%d/chunk%d/%s references checkpoint %.12s which is missing/empty — inconsistent store",
|
||
r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, cs.FinalHash)
|
||
}
|
||
var usage llm.Usage
|
||
if uerr := json.Unmarshal([]byte(cp.UsageJSON), &usage); uerr != nil {
|
||
r.Log.DebugContext(ctx, "checkpoint usage_json unreadable; tokens report as zero", "hash", cs.FinalHash[:12], "err", uerr)
|
||
}
|
||
sr.Usage = usage
|
||
sr.Model = cp.ModelActual
|
||
sr.FinishReason = cp.FinishReason
|
||
sr.Text = cp.ResponseText
|
||
} else if cs.FlagReason == string(FlagSanitizerStripped) && cs.FinalHash != "" {
|
||
// A cosmetic sanitizer strip (D35.4a): the cleaned export text lives in the derived
|
||
// checkpoint final_hash points at (written durably before this chunk_status row), so resume
|
||
// re-serves the identical FinalText WITHOUT re-stripping — deterministic and $0. A missing
|
||
// derived checkpoint is a torn store (final_hash is written after it), so fail loud like the
|
||
// ok branch rather than silently export empty.
|
||
cp, err := r.Store.GetCheckpoint(cs.FinalHash)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("pipeline: read sanitized-export checkpoint %.12s for %s/ch%d/chunk%d/%s: %w",
|
||
cs.FinalHash, r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, err)
|
||
}
|
||
if cp == nil || cp.ResponseText == "" {
|
||
return nil, fmt.Errorf("pipeline: chunk_status sanitizer_stripped for %s/ch%d/chunk%d/%s references derived checkpoint %.12s which is missing/empty — inconsistent store",
|
||
r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, cs.FinalHash)
|
||
}
|
||
sr.RecoveredText = cp.ResponseText
|
||
// Mirror the ok branch: the resume request_log row's request_hash points at this derived
|
||
// checkpoint, so report ITS finish_reason ("sanitized_export") rather than a bare "" that
|
||
// contradicts the referenced row (adversarial review NIT; telemetry-only).
|
||
sr.FinishReason = cp.FinishReason
|
||
// F7 (D39.4): mirror the ok branch's model attribution — the derived checkpoint carries the
|
||
// ACTUAL model that produced the flagged output; without this the resumed telemetry row is
|
||
// attributed to the config slug (st.Model) instead of the model that actually answered.
|
||
sr.Model = cp.ModelActual
|
||
}
|
||
rl := r.baseRequestLog(st, ch, st.Model, cs.FinalHash)
|
||
rl.ModelActual = sr.Model
|
||
rl.TMHit, rl.OK = true, cs.Disposition == string(DispOK)
|
||
rl.FinishReason, rl.Degraded = sr.FinishReason, nonOKTag(cs.Disposition, cs.FlagReason)
|
||
r.Store.LogRequest(ctx, r.Log, rl)
|
||
r.Log.InfoContext(ctx, "stage resolved from chunk_status", "stage", st.Name,
|
||
"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "disposition", cs.Disposition, "reason", cs.FlagReason)
|
||
return sr, nil
|
||
}
|
||
|
||
// nonOKTag is the request_log `degraded` value for a resumed disposition row.
|
||
func nonOKTag(disposition, reason string) string {
|
||
if disposition == string(DispOK) {
|
||
return ""
|
||
}
|
||
return reason
|
||
}
|