package pipeline import ( "context" "fmt" "sort" "textmachine/backend/internal/store" ) // status.go: the READ-ONLY progress projection (`tmctl status`) and the targeted re-attack // of flagged chunks (`tmctl redrive`) — the D12 "прогресс — единственный реальный пробел" // tail, ratified by D15.3. Status makes ZERO LLM calls and ZERO checkpoint replays: it reads // the book manifest (a deterministic, $0 re-chunk of the source) and projects it against the // stored chunk_status / spend / retrieval_state. Redrive resets the flagged chunks' terminal // state (their chunk_status + checkpoints) and re-runs the durable loop, so the DispOK work // resumes at $0 and only the reset stages re-attack with a fresh retry/escalation budget. // ChunkState is a chunk's resolved lifecycle state for the projection — a chunk-level view // over its per-stage chunk_status rows. `skipped` is deliberately NOT a chunk state: it is a // per-STAGE substate (the downstream stages of a flagged chunk), surfaced in the passport's // stage counts, since the linear C1 core never skips a whole chunk. type ChunkState string const ( ChunkDone ChunkState = "done" // every pipeline stage resolved ok ChunkFlagged ChunkState = "flagged" // a stage flagged (chunk not translated; ≠ infra failure) ChunkInProgress ChunkState = "in_progress" // some stages resolved, not all, none flagged (an interrupted run) ChunkPending ChunkState = "pending" // no stage attempted yet ) // ChapterPassport is the per-chapter quality passport (D12): unit counts, the worst flag // reason, a pass|attention|fail verdict (exp07 chapter rule) and the chapter's spend. type ChapterPassport struct { Chapter int `json:"chapter"` ChunksTotal int `json:"chunks_total"` ChunksDone int `json:"chunks_done"` ChunksFlagged int `json:"chunks_flagged"` ChunksInFlight int `json:"chunks_in_progress"` ChunksPending int `json:"chunks_pending"` StagesSkipped int `json:"stages_skipped"` // per-stage skip count (downstream of a flag) Escalations int `json:"escalations"` PostcheckMisses int `json:"postcheck_misses"` // confirmed post-check misses (retrieval_state) StyleFlags int `json:"style_flags"` // cheap style/number gate hits (observability, not a disposition) WorstFlagReason string `json:"worst_flag_reason,omitempty"` Verdict string `json:"verdict"` // pass | attention | fail (exp07: 0/1/≥2 flagged chunks) CostUSD float64 `json:"cost_usd"` } // StatusReport is the whole-book read-model. type StatusReport struct { BookID string `json:"book_id"` Snapshot string `json:"snapshot_id"` // the snapshot the stored rows were resolved under ("" = nothing run yet) // SnapshotDrift is true when the stored rows carry more than one snapshot id (a config // changed mid-book without a full --resnapshot re-pin) — a loud signal, not a silent skew. SnapshotDrift bool `json:"snapshot_drift"` // ConfigDrift is true when the CURRENT config renders a snapshot DIFFERENT from the one the // stored rows were resolved under — a wire/verdict-affecting edit since the last run (e.g. a // prompt-version bump, a gate flip) that `tmctl translate` would --resnapshot. Without this // the operator sees "done/pass" and false confidence (finding #3). CurrentSnapshot is what // the config renders now (empty when it could not be computed or matches). ConfigDrift bool `json:"config_drift"` CurrentSnapshot string `json:"current_snapshot,omitempty"` TotalChunks int `json:"total_chunks"` Done int `json:"done"` InProgress int `json:"in_progress"` Flagged int `json:"flagged"` Pending int `json:"pending"` PercentDone float64 `json:"percent_done"` // 100·done/total, unit-count only (no synthetic time-bar) // GlossaryMissFlagged counts chunks the post-check GATE promoted to flagged/glossary_miss (a // CONFIRMED miss on an otherwise-ok chunk). A SUBSET of Flagged and NOT re-drivable: the miss // re-derives from the unchanged glossary, so the fix is a seed edit + `translate --resnapshot`, // never a redrive (a no-op for them). Flagged − GlossaryMissFlagged is the re-drivable remainder, // which is what the status CLI must advise `tmctl redrive` for (minor 1d: the old blanket hint // sent glossary_miss chunks into a redrive dead-end). GlossaryMissFlagged int `json:"glossary_miss_flagged"` Escalations int `json:"escalations"` PostcheckMisses int `json:"postcheck_misses"` // StyleFlags is the book-wide total of the cheap deterministic style/number gate hits // (dialogue-dash, ё, translit-interjection, 万/億 magnitude). Observability, never a // disposition — a nonzero count is "attention worth a human glance", not a failed chunk. StyleFlags int `json:"style_flags"` CommittedUSD float64 `json:"committed_usd"` ReservedUSD float64 `json:"reserved_usd"` BookCeilingUSD float64 `json:"book_ceiling_usd,omitempty"` CeilingPct float64 `json:"ceiling_pct,omitempty"` // 100·(committed+reserved)/book_ceiling ProjectedBookUSD float64 `json:"projected_book_usd"` // committed extrapolated over the whole book ETASeconds float64 `json:"eta_seconds,omitempty"` // secondary: mean fresh-call throughput × remaining Chapters []ChapterPassport `json:"chapters"` } // flagReasonSeverity ranks flag reasons worst-first for the chapter's "worst flag" passport // field. Lower = worse. Deterministic content-failures (refusal/echo/excision) outrank // retryable budget symptoms (length/empty), which are the least alarming. func flagReasonSeverity(reason string) int { switch FlagReason(reason) { case FlagHardRefusal, FlagSoftRefusal, FlagContentFilter, FlagHardBlock: return 0 case FlagCJKArtifact, FlagExcisionSuspect, FlagCoverageFail: return 1 case FlagSanitizerDefect: // A contaminated output (leaked preamble / notes / ###) is unreadable-as-shipped — // ranked with the deterministic content failures, above a mere budget symptom. return 2 case FlagLoopDegenerate: return 3 case FlagDecodeError: return 4 case FlagGlossaryMiss: return 5 case FlagLength, FlagEmpty: return 6 } return 7 } // bookChunks re-derives the book's chunk manifest — Ingest + SplitChunks — for the honest N/M // denominator. Pure and $0 (no LLM); unlike TranslateBook it does NOT persist ruby or seed the // glossary, so status stays a pure read. A change to the chunk BOUNDARIES (added/removed text that // shifts the manifest) or the chunker VERSION is reflected here. // // CAVEAT (minor 1d — narrowed overclaim): a source edit that leaves the boundaries unchanged (a // typo fix inside a chunk) is INVISIBLE to status — it changes the chunk's content_hash but not the // manifest and not the snapshot (which carries chunker_version, not the source bytes), so status // still reads done/pass over a translation of the OLD text. Surfacing it would need a per-chunk // content_hash re-render ($0) compared against the stored chunk_status.content_hash; deferred, // because the resume fast-path already re-checks content_hash on the NEXT translate — which is // where such an edit surfaces (as a re-bill of exactly the touched chunks), just not in status. func (r *Runner) bookChunks() ([]Chunk, error) { doc, err := IngestEncoded(r.Book.SourceFile, r.Book.Encoding, r.Book.SourceLang) if err != nil { return nil, err } return SplitChunks(doc.Chapters), nil } // chunkKey identifies a chunk positionally. type chunkKey struct { chapter, chunkIdx int } // resolveChunkState maps a chunk's per-stage rows to its chunk-level state, plus its cost, // escalation flag and (when flagged) the flag reason. stagesTotal is the pipeline stage count. func resolveChunkState(rows []store.ChunkStatus, stagesTotal int) (st ChunkState, reason string, cost float64, escalated bool, skipped int) { if len(rows) == 0 { return ChunkPending, "", 0, false, 0 } ok := 0 for _, cs := range rows { cost += cs.CostUSD if cs.Escalated { escalated = true } switch cs.Disposition { case string(DispOK): ok++ case string(DispFlagged): // The first (only) flagged stage decides the chunk; keep its reason. st, reason = ChunkFlagged, cs.FlagReason case string(DispSkipped): skipped++ } } if st == ChunkFlagged { return st, reason, cost, escalated, skipped } if ok == stagesTotal { return ChunkDone, "", cost, escalated, skipped } return ChunkInProgress, "", cost, escalated, skipped } // Status builds the read-only progress projection. It opens no jobs, reserves nothing, makes // no LLM call and replays no checkpoint — the D12 query-handler analogue. Safe to run whenever // the project file is not held by a running tmctl (the same exclusive-lock rule as `report`). func (r *Runner) Status(ctx context.Context) (*StatusReport, error) { chunks, err := r.bookChunks() if err != nil { return nil, err } statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID) if err != nil { return nil, err } byChunk := map[chunkKey][]store.ChunkStatus{} snapSeen := map[string]bool{} for _, cs := range statuses { byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}] = append(byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}], cs) if cs.SnapshotID != "" { snapSeen[cs.SnapshotID] = true } } // Per-chunk post-check misses (retrieval_state) — the glossary-consistency signal that is // re-derived each run and NOT a chunk_status disposition (default flagger mode). states, err := r.Store.RetrievalStatesForBook(r.Book.BookID) if err != nil { return nil, err } missByChunk := map[chunkKey]int{} styleByChunk := map[chunkKey]int{} for _, rs := range states { missByChunk[chunkKey{rs.Chapter, rs.ChunkIdx}] = rs.NPostcheckMiss styleByChunk[chunkKey{rs.Chapter, rs.ChunkIdx}] = rs.NStyleFlags } stagesTotal := len(r.Pipeline.Stages) // The post-check GATE (opt-in) flags a chunk at the CHUNK level AFTER the stage loop, so it // is NEVER written as a chunk_status row (every stage stays DispOK). Without accounting for // it here, status would report a gate-flagged chunk as done/pass while `tmctl translate` // exits 2 (finding #2). When the gate is ON, promote an otherwise-done chunk with a CONFIRMED // post-check miss to flagged/glossary_miss so status matches the run. gateOn := r.Pipeline.Gates.Glossary.PostcheckGate rep := &StatusReport{BookID: r.Book.BookID, TotalChunks: len(chunks)} passports := map[int]*ChapterPassport{} var chapterOrder []int var processedCost float64 // spend of PROCESSED chunks (done+flagged) — the projection base (finding #7) for _, ch := range chunks { p := passports[ch.Chapter] if p == nil { p = &ChapterPassport{Chapter: ch.Chapter, Verdict: "pass"} passports[ch.Chapter] = p chapterOrder = append(chapterOrder, ch.Chapter) } p.ChunksTotal++ key := chunkKey{ch.Chapter, ch.ChunkIdx} state, reason, cost, escalated, skipped := resolveChunkState(byChunk[key], stagesTotal) p.CostUSD += cost p.StagesSkipped += skipped miss := missByChunk[key] p.PostcheckMisses += miss rep.PostcheckMisses += miss style := styleByChunk[key] p.StyleFlags += style rep.StyleFlags += style if gateOn && state == ChunkDone && miss > 0 { // Matches translateChunk: the gate flags only an otherwise-ok chunk. Not re-drivable // (the miss re-derives from the unchanged glossary; a fix is a glossary edit = // --resnapshot, not a redrive) — counted separately so the CLI advises the right action. state, reason = ChunkFlagged, string(FlagGlossaryMiss) rep.GlossaryMissFlagged++ } if escalated { p.Escalations++ rep.Escalations++ } switch state { case ChunkDone: rep.Done++ p.ChunksDone++ case ChunkFlagged: rep.Flagged++ p.ChunksFlagged++ if p.WorstFlagReason == "" || flagReasonSeverity(reason) < flagReasonSeverity(p.WorstFlagReason) { p.WorstFlagReason = reason } case ChunkInProgress: rep.InProgress++ p.ChunksInFlight++ case ChunkPending: rep.Pending++ p.ChunksPending++ } if state == ChunkDone || state == ChunkFlagged { processedCost += cost // a fully-attempted chunk's cost feeds the projection } } // Chapter verdicts (exp07 chapter rule: 0 flagged = pass, 1 = attention, ≥2 = fail). sort.Ints(chapterOrder) for _, n := range chapterOrder { p := passports[n] switch { case p.ChunksFlagged >= 2: p.Verdict = "fail" case p.ChunksFlagged == 1: p.Verdict = "attention" default: p.Verdict = "pass" } rep.Chapters = append(rep.Chapters, *p) } if rep.TotalChunks > 0 { rep.PercentDone = 100 * float64(rep.Done) / float64(rep.TotalChunks) } if len(snapSeen) == 1 { for s := range snapSeen { rep.Snapshot = s } } else if len(snapSeen) > 1 { rep.SnapshotDrift = true } // Config-drift (finding #3): compute the CURRENT config's snapshot READ-ONLY (materialize // the STORED glossary — no re-seed, no write) and compare to the single snapshot the stored // rows carry. A wire/verdict-affecting config edit since the last run (a prompt bump, a gate // flip — this very package bumps the editor prompt_version) is otherwise hidden behind // "done/pass" until translate forces a --resnapshot. Only meaningful when a single snapshot // is on record (multi-snapshot is already flagged as SnapshotDrift). Note: a change to the // seed FILE that was not re-run is NOT detected here (the stored glossary is what status // projects); it surfaces on the next translate's re-seed. if rep.Snapshot != "" { if curSnap, serr := r.currentSnapshotProjected(); serr != nil { // Не молчать: провал проекции раньше тихо читался как «дрифта нет», и // оператор верил чистому статусу при реально изменённом конфиге (аудит // цепочек). Drift остаётся false (проекция неизвестна), но с WARN. r.Log.WarnContext(ctx, "config-drift check failed; drift state unknown (reported as none)", "err", serr) } else if curSnap != rep.Snapshot { rep.ConfigDrift = true rep.CurrentSnapshot = curSnap } } // Money. committed, reserved, err := r.Store.SpentUSD(r.Book.BookID) if err != nil { return nil, err } rep.CommittedUSD, rep.ReservedUSD = committed, reserved rep.BookCeilingUSD = r.Book.Ceilings.BookUSD if rep.BookCeilingUSD > 0 { rep.CeilingPct = 100 * (committed + reserved) / rep.BookCeilingUSD } // Projected book cost: extrapolate the per-PROCESSED-chunk average over the whole book // (done + flagged = a chunk fully attempted). Uses processedCost, NOT book committed: // committed also carries partial spend on IN-PROGRESS chunks (excluded from the denominator), // which would over-estimate (finding #7). The clean average × total is the honest estimate. processed := rep.Done + rep.Flagged if processed > 0 { rep.ProjectedBookUSD = processedCost / float64(processed) * float64(rep.TotalChunks) } // ETA (secondary): mean fresh-call throughput × remaining processing. No synthetic bar. // DEVIATION from D12 (which ratified an EWMA) — minor 1d, made explicit: this is a plain // arithmetic mean, not an exponentially-weighted one. For a batch book run throughput is // ~stationary, so the mean and an EWMA converge; a recency-weighted EWMA (which would adapt to // a mid-run provider slowdown) is deferred because it needs the ORDERED per-call latencies, not // the sum+count FreshCallLatencyMS returns. Kept secondary/advisory in the output so it is never // mistaken for a committed completion time. totalMS, freshCalls, err := r.Store.FreshCallLatencyMS(r.Book.BookID) if err != nil { return nil, err } remaining := rep.TotalChunks - processed if processed > 0 && freshCalls > 0 && remaining > 0 { meanMSPerChunk := float64(totalMS) / float64(processed) rep.ETASeconds = meanMSPerChunk * float64(remaining) / 1000 } return rep, nil } // currentSnapshotProjected computes the CURRENT config's snapshotID from the STORED glossary // (read-only: materialize what is persisted, no re-seed, no write) — the single projection // Status uses for ConfigDrift and Redrive uses to guard its destructive reset. Side effect // (deliberate, on the read path): it sets r.memory to the stored-glossary materialization, the // same as Status did inline. A seed-FILE edit not yet re-run is NOT reflected here (the STORED // glossary is projected, not the seed file) — that drift surfaces on the next translate's // re-seed, exactly as it does for `translate` itself. func (r *Runner) currentSnapshotProjected() (string, error) { rows, err := r.Store.GlossaryForBook(r.Book.BookID) if err != nil { return "", err } r.memory = materializeMemory(rows, r.Pipeline.Gates.Glossary.PostcheckGate) id, _, err := r.snapshotID() return id, err } // RedriveSelector picks the flagged chunks to re-attack. -1 on Chapter/ChunkIdx means "any" // (a valid chunk index is ≥0, a chapter ≥1, so -1 is an unambiguous "unset"); an empty Reason // matches any flag_reason. DryRun reports what WOULD be reset without touching anything. type RedriveSelector struct { Chapter int ChunkIdx int Reason string DryRun bool } func (s RedriveSelector) matches(cs store.ChunkStatus) bool { if s.Chapter >= 0 && cs.Chapter != s.Chapter { return false } if s.ChunkIdx >= 0 && cs.ChunkIdx != s.ChunkIdx { return false } if s.Reason != "" && cs.FlagReason != s.Reason { return false } return true } // RedriveTarget is one chunk selected for re-attack and the stages that were (or would be) reset. type RedriveTarget struct { Chapter int `json:"chapter"` ChunkIdx int `json:"chunk_idx"` FlagReason string `json:"flag_reason"` Stages []string `json:"reset_stages"` } // RedriveSummary reports the plan/outcome of the reset half of a redrive. type RedriveSummary struct { Targets []RedriveTarget `json:"targets"` DryRun bool `json:"dry_run"` ResetRun bool `json:"reset_run"` // the re-translate ran (false on dry-run / no targets) } // Redrive re-attacks the FLAGGED chunks matching sel (D15.3 / D12 Step-Functions redrive). It // NEVER touches DispOK work (D12): for each selected chunk it resets only the flagged stage and // its downstream skipped stages — their chunk_status + checkpoints are deleted (ResetChunkStages), // so the durable re-run re-derives them with a FRESH retry/escalation budget (a fresh provider // call, not a deterministic replay of the flagged completion) while the upstream ok stages resume // at $0. Money bills normally (reserve/settle+checkpoint); the previously-spent money on the // discarded attempts stays committed (honest — it was really billed), so the ceilings remain // accurate. Requires the CURRENT config to render the SAME snapshot the flagged rows carry (no // --resnapshot): a config change makes the re-run fail loud with the resnapshot guidance instead // of silently re-pricing the book. Returns the reset plan, and the re-run's BookResult (nil on // dry-run or when nothing matched). func (r *Runner) Redrive(ctx context.Context, sel RedriveSelector) (*RedriveSummary, *BookResult, error) { statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID) if err != nil { return nil, nil, err } // Group by chunk and find the targets: a chunk with a FLAGGED stage matching the selector. byChunk := map[chunkKey][]store.ChunkStatus{} var order []chunkKey for _, cs := range statuses { k := chunkKey{cs.Chapter, cs.ChunkIdx} if _, seen := byChunk[k]; !seen { order = append(order, k) } byChunk[k] = append(byChunk[k], cs) } sort.Slice(order, func(i, j int) bool { if order[i].chapter != order[j].chapter { return order[i].chapter < order[j].chapter } return order[i].chunkIdx < order[j].chunkIdx }) // Build the reset PLAN first — pure, no mutation — so the destructive reset happens only // after the drift guard below clears (Task 1c). Each target resets its flagged stage AND its // downstream skipped stages; upstream DispOK stages are never included (D12 — never re-pay ok // work), which is the invariant the redrive tests mutation-lock. summary := &RedriveSummary{DryRun: sel.DryRun} for _, k := range order { rows := byChunk[k] targeted := false reason := "" for _, cs := range rows { if cs.Disposition == string(DispFlagged) && sel.matches(cs) { targeted, reason = true, cs.FlagReason break } } if !targeted { continue } var stages []string for _, cs := range rows { if cs.Disposition == string(DispFlagged) || cs.Disposition == string(DispSkipped) { stages = append(stages, cs.Stage) } } sort.Strings(stages) summary.Targets = append(summary.Targets, RedriveTarget{ Chapter: k.chapter, ChunkIdx: k.chunkIdx, FlagReason: reason, Stages: stages, }) } if sel.DryRun || len(summary.Targets) == 0 { return summary, nil, nil } // Config/seed-drift guard BEFORE the destructive reset (external-review 1c). A redrive re-runs // under the CURRENT config; if it drifted from the snapshot the stored rows carry, TranslateBook // fails loud in runStage with the resnapshot guidance — but the OLD code had already deleted the // flagged chunk_status rows + checkpoints (ResetChunkStages), so that fail-loud left the flag // telemetry destroyed and a later `status` read the chunk as pending/pass. // // Re-seed the glossary from the seed FILE BEFORE the destructive reset — on BOTH paths, incl. // --resnapshot (пакет №3 fix: the redrive-resnapshot wiring in cmd/tmctl exposed that the old code // nested this seed inside `if !r.Resnapshot`, so under --resnapshot a seed-FILE edit that // introduces a collision would fail loud only in TranslateBook's re-seed AFTER ResetChunkStages // already deleted the flag telemetry — re-opening the external-review 1c torn-state class for the // resnapshot path). seedGlossary is idempotent and touches only glossary rows (never // chunk_status/checkpoints), so a seed error (e.g. a new shared-key collision) surfaces HERE, // before any reset; TranslateBook re-seeds identically afterwards (no double cost). It also // materializes r.memory so the snapshot comparison below matches what TranslateBook renders. if err := r.seedGlossary(ctx); err != nil { return summary, nil, fmt.Errorf("pipeline: redrive seed glossary: %w", err) } // Config/seed-drift guard BEFORE the destructive reset (external-review 1c): if the current config // drifted from the snapshot the stored rows carry, TranslateBook would fail loud in runStage — but // the reset (ResetChunkStages) would already have deleted the flagged rows/checkpoints, so a later // `status` reads the chunk as pending/pass. Refuse loud here instead. Skipped under --resnapshot // (the operator explicitly accepts the re-pin/re-pay) — but ONLY the snapshot COMPARISON is skipped, // never the seed-error-before-reset protection above. if !r.Resnapshot { curSnap, _, serr := r.snapshotID() if serr != nil { return summary, nil, fmt.Errorf("pipeline: redrive snapshot check: %w", serr) } for _, cs := range statuses { if cs.SnapshotID != "" && cs.SnapshotID != curSnap { return summary, nil, fmt.Errorf("pipeline: redrive прерван — строки книги под снапшотом %.12s, а текущий конфиг/сид рендерит %.12s (конфиг/промпты/капабилити/сид-глоссарий изменились): сброс сейчас пере-оплатил бы книгу и уничтожил бы флаг-телеметрию; повторите `translate --resnapshot`, чтобы принять пере-оплату явно, либо верните конфиг/сид", cs.SnapshotID, curSnap) } } } // Destructive reset — safe now that drift has been ruled out. for _, t := range summary.Targets { if err := r.Store.ResetChunkStages(r.Book.BookID, t.Chapter, t.ChunkIdx, t.Stages); err != nil { return summary, nil, fmt.Errorf("pipeline: redrive reset ch%d/chunk%d: %w", t.Chapter, t.ChunkIdx, err) } } // Re-run the durable loop: reset chunks re-attack fresh; everything else resumes at $0. res, err := r.TranslateBook(ctx) summary.ResetRun = true return summary, res, err }