153 lines
6.8 KiB
Go
153 lines
6.8 KiB
Go
package store
|
||
|
||
import (
|
||
"database/sql"
|
||
"errors"
|
||
)
|
||
|
||
// chunkstatus.go: the per-chunk×stage disposition record (Веха 2, D2). It is a
|
||
// materialized RESOLVE over the append-only checkpoints (keyed by request_hash,
|
||
// which already carries `attempt`), NOT a column on checkpoints. Because it is
|
||
// derivable from the checkpoints, losing it to a kill -9 self-heals: on resume
|
||
// the runner re-classifies the checkpoint text (no re-billing) and re-upserts
|
||
// the row. `disposition` is the axis the runner loop reads BEFORE rendering, so
|
||
// a terminally-flagged chunk is never re-attacked into an infinite paid loop.
|
||
//
|
||
// The store is dumb storage here: disposition/flag_reason are plain strings; the
|
||
// typed FlagReason/Disposition constants and the classifier live in the pipeline
|
||
// package.
|
||
|
||
// ChunkStatus is one (book, chapter, chunk, stage) disposition row.
|
||
type ChunkStatus struct {
|
||
BookID string
|
||
Chapter int
|
||
ChunkIdx int
|
||
Stage string
|
||
SnapshotID string
|
||
ContentHash string // signature of the rendered msgs (the source is NOT in the snapshot); guards the resume fast-path against serving a stale translation after a source edit
|
||
Disposition string // ok | flagged | skipped
|
||
FlagReason string // "" when ok
|
||
Attempts int // number of attempts made for this chunk×stage
|
||
FinalHash string // request_hash of the authoritative checkpoint (ok path)
|
||
CostUSD float64 // sum across all attempts (F3-honest)
|
||
Detail string
|
||
Escalated bool // a single-hop fallback was used for this chunk×stage (D12/D15.3 telemetry)
|
||
EscalationModel string // the fallback model that answered ("" when not escalated)
|
||
}
|
||
|
||
// UpsertChunkStatus writes (or overwrites) the disposition row. Overwrite is the
|
||
// resolve semantics: re-running classify over the same checkpoints reproduces the
|
||
// same row, and a --resnapshot re-processing legitimately replaces a stale row.
|
||
func (s *Store) UpsertChunkStatus(cs ChunkStatus) error {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
_, err := s.w.ExecContext(ctx, `
|
||
INSERT INTO chunk_status (
|
||
book_id, chapter, chunk_idx, stage, snapshot_id, content_hash,
|
||
disposition, flag_reason, attempts, final_hash, cost_usd, detail,
|
||
escalated, escalation_model, updated_at
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
|
||
ON CONFLICT (book_id, chapter, chunk_idx, stage) DO UPDATE SET
|
||
snapshot_id = excluded.snapshot_id,
|
||
content_hash = excluded.content_hash,
|
||
disposition = excluded.disposition,
|
||
flag_reason = excluded.flag_reason,
|
||
attempts = excluded.attempts,
|
||
final_hash = excluded.final_hash,
|
||
cost_usd = excluded.cost_usd,
|
||
detail = excluded.detail,
|
||
escalated = excluded.escalated,
|
||
escalation_model = excluded.escalation_model,
|
||
updated_at = excluded.updated_at`,
|
||
cs.BookID, cs.Chapter, cs.ChunkIdx, cs.Stage, cs.SnapshotID, cs.ContentHash,
|
||
cs.Disposition, cs.FlagReason, cs.Attempts, cs.FinalHash, cs.CostUSD, cs.Detail,
|
||
boolToInt(cs.Escalated), cs.EscalationModel)
|
||
return err
|
||
}
|
||
|
||
// GetChunkStatus returns the disposition row for a chunk×stage, or nil if none.
|
||
func (s *Store) GetChunkStatus(bookID string, chapter, chunkIdx int, stage string) (*ChunkStatus, error) {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
cs := ChunkStatus{BookID: bookID, Chapter: chapter, ChunkIdx: chunkIdx, Stage: stage}
|
||
var escalated int
|
||
err := s.r.QueryRowContext(ctx, `
|
||
SELECT snapshot_id, content_hash, disposition, flag_reason, attempts, final_hash, cost_usd, detail,
|
||
escalated, escalation_model
|
||
FROM chunk_status
|
||
WHERE book_id = ? AND chapter = ? AND chunk_idx = ? AND stage = ?`,
|
||
bookID, chapter, chunkIdx, stage).Scan(
|
||
&cs.SnapshotID, &cs.ContentHash, &cs.Disposition, &cs.FlagReason, &cs.Attempts, &cs.FinalHash, &cs.CostUSD, &cs.Detail,
|
||
&escalated, &cs.EscalationModel)
|
||
cs.Escalated = escalated != 0
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return nil, nil
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &cs, nil
|
||
}
|
||
|
||
// ChunkStatusesForBook returns every disposition row for a book, ordered for a
|
||
// stable report (tmctl report: the flag section).
|
||
func (s *Store) ChunkStatusesForBook(bookID string) ([]ChunkStatus, error) {
|
||
return queryAll(s.r, `
|
||
SELECT chapter, chunk_idx, stage, snapshot_id, content_hash, disposition, flag_reason,
|
||
attempts, final_hash, cost_usd, detail, escalated, escalation_model
|
||
FROM chunk_status WHERE book_id = ?
|
||
ORDER BY chapter, chunk_idx, stage`,
|
||
func(rows *sql.Rows) (ChunkStatus, error) {
|
||
cs := ChunkStatus{BookID: bookID}
|
||
var escalated int
|
||
if err := rows.Scan(&cs.Chapter, &cs.ChunkIdx, &cs.Stage, &cs.SnapshotID, &cs.ContentHash,
|
||
&cs.Disposition, &cs.FlagReason, &cs.Attempts, &cs.FinalHash, &cs.CostUSD, &cs.Detail,
|
||
&escalated, &cs.EscalationModel); err != nil {
|
||
return cs, err
|
||
}
|
||
cs.Escalated = escalated != 0
|
||
return cs, nil
|
||
}, bookID)
|
||
}
|
||
|
||
// ResetChunkStages deletes the chunk_status rows AND the checkpoints for the given stages of
|
||
// ONE chunk, in a single transaction — the durable half of `tmctl redrive` (D15.3). After
|
||
// this, the next translate re-derives those stages: their chunk_status is gone (the runner
|
||
// re-runs instead of resuming the flag) and their checkpoints are gone (a FRESH provider call
|
||
// is made, not a deterministic replay of the flagged completion). Only the passed stages are
|
||
// touched — an upstream OK stage keeps its checkpoint and resumes at $0 (D12: never re-pay
|
||
// DispOK work). MONEY (documented, honest): the money already spent on the discarded attempts
|
||
// STAYS committed in `spend` (it was really billed at the provider), so the spend ceilings
|
||
// remain honest; only the resume cache and the escalation-budget accounting are reset, which
|
||
// is the "fresh retry/escalation budget" redrive grants an explicit operator command. Thus
|
||
// after a redrive committed(spend) >= SUM(checkpoints) — the safe direction (a ceiling never
|
||
// under-counts). Checkpoints are joined to jobs by (book, chapter, stage); request_log
|
||
// (append-only telemetry) is deliberately NOT touched, so the paid history is still auditable.
|
||
func (s *Store) ResetChunkStages(bookID string, chapter, chunkIdx int, stages []string) error {
|
||
if len(stages) == 0 {
|
||
return nil
|
||
}
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
tx, err := s.w.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
for _, stage := range stages {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
DELETE FROM checkpoints
|
||
WHERE chunk_idx = ? AND job_id IN (
|
||
SELECT id FROM jobs WHERE book_id = ? AND chapter = ? AND stage = ?)`,
|
||
chunkIdx, bookID, chapter, stage); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
DELETE FROM chunk_status
|
||
WHERE book_id = ? AND chapter = ? AND chunk_idx = ? AND stage = ?`,
|
||
bookID, chapter, chunkIdx, stage); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return tx.Commit()
|
||
}
|