106 lines
4.3 KiB
Go
106 lines
4.3 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
|
||
}
|
||
|
||
// 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, 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,
|
||
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)
|
||
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}
|
||
err := s.r.QueryRowContext(ctx, `
|
||
SELECT snapshot_id, content_hash, disposition, flag_reason, attempts, final_hash, cost_usd, detail
|
||
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)
|
||
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) {
|
||
ctx, cancel := opContext()
|
||
defer cancel()
|
||
rows, err := s.r.QueryContext(ctx, `
|
||
SELECT chapter, chunk_idx, stage, snapshot_id, content_hash, disposition, flag_reason,
|
||
attempts, final_hash, cost_usd, detail
|
||
FROM chunk_status WHERE book_id = ?
|
||
ORDER BY chapter, chunk_idx, stage`, bookID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
var out []ChunkStatus
|
||
for rows.Next() {
|
||
cs := ChunkStatus{BookID: bookID}
|
||
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); err != nil {
|
||
return nil, err
|
||
}
|
||
out = append(out, cs)
|
||
}
|
||
return out, rows.Err()
|
||
}
|