package pipeline import ( "context" "fmt" "sort" "strings" "textmachine/backend/internal/chunk" "textmachine/backend/internal/membank" "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 "progress is the only real gap" // 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"` UnitsTotal int `json:"units_total"` UnitsDone int `json:"units_done"` UnitsFlagged int `json:"units_flagged"` UnitsInProgress int `json:"units_in_progress"` UnitsPending int `json:"units_pending"` StagesSkipped int `json:"stages_skipped"` // per-stage skip count (downstream of a flag) Escalations int `json:"escalations"` PostcheckMisses int `json:"postcheck_misses"` StyleFlags int `json:"style_flags"` // cheap style/number gate hits (observability, not a disposition) // RepairApplied counts this chapter's units whose SHIPPED text carries an applied repair (pack-16). It is // DERIVED from the durable final_hash namespace, never stored, so it survives a resume like the rest of // the projection. omitempty: a chapter with no repairs renders exactly as before the field existed. RepairApplied int `json:"repair_applied,omitempty"` WorstFlagReason string `json:"worst_flag_reason,omitempty"` Verdict string `json:"verdict"` // pass | attention | fail (exp07: 0/1/≥2 flagged units) 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"` // RebillUnits/RebillUSD turn the drift BOOLEAN into the number the operator actually decides on // (spec D15.2 §9, taken 25.07): "the config drifted" says nothing about whether continuing costs a // cent or the whole book — these say "N chunk×stage units already billed under a superseded snapshot // would be paid for again, ~$X". Same projection the consent gate refuses on (projectRebill), so // status can never quote a different number than the one translate enforces. Both omitempty: a book // with no drift keeps its exact prior bytes. RebillUnits int `json:"rebill_units,omitempty"` RebillUSD float64 `json:"rebill_usd,omitempty"` TotalUnits int `json:"total_units"` 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"` // UnsignedBankTerms is how many bank rows carry a rendering NOBODY approved — the engine's own auto/draft // proposals, which the auto mode of the bank stop puts in front of the editor marked ⟨проверить⟩. Directive // item 4 asks for it HERE and not only in `report`: status is the command an operator runs before deciding // whether to keep paying, and "this book is translating against N unsigned terms" is that decision's input. UnsignedBankTerms int `json:"unsigned_bank_terms,omitempty"` // 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 // ContentLabels / Routing are the content-label PROVENANCE (B6): what the book declares and which // model each stage resolves to under it. They are a $0 projection of the config — deliberately NOT a // hash input (the labels' wire fate is the resolved model itself, D39.26 point 1), so this is where an // operator reads "which endpoint received this book". Both omitempty: an unlabelled book's report is // byte-identical to what it was before these fields existed. ContentLabels []string `json:"content_labels,omitempty"` // Routing is one "stage=model" entry per stage (plus "→hop" when a fallback is wired), in stage order. // A string list rather than a struct: it is read by humans and diffed by CI, and a labelled run's whole // routing decision fits on one line. Routing []string `json:"routing,omitempty"` // ContentRoutingProblems mirrors the read-path warning (D39.26 point 9): a book whose labels are not // runnable stays inspectable, and the reason travels WITH the projection instead of only in a log line // the operator may never see. ContentRoutingProblems []string `json:"content_routing_problems,omitempty"` Chapters []ChapterPassport `json:"chapters"` } // contentRoutingRows renders the per-stage routing projection ("stage=model" / "stage=model→hop") in // stage order. Shared by the status and quality read models so the two can never disagree about which // model a stage resolved to. func (r *Runner) contentRoutingRows() []string { rows := make([]string, 0, len(r.Pipeline.Stages)) for _, st := range r.Pipeline.Stages { row := st.Name + "=" + st.ResolvedModel if st.ResolvedHop != "" { row += "→" + st.ResolvedHop } rows = append(rows, row) } return rows } // 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) that was DROPPED 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 FlagSanitizerStripped: // A cosmetic leak the sanitizer STRIPPED and exported (D35.4a): the chunk shipped cleaned, // so it is the least alarming flag — an "auto-cleaned, glance to verify" signal, ranked below // a budget symptom (the chunk is not lost; a human need only spot-check the auto-clean). return 7 case FlagLength, FlagEmpty: return 6 } return 8 } // bookChunks re-derives the book's chunk manifest — Ingest + chunk.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.Chunk, error) { doc, err := chunk.IngestEncoded(r.Book.SourceFile, r.Book.Encoding, r.Book.SourceLang) if err != nil { return nil, err } return chunk.SplitChunks(doc.Chapters, r.segBudget(), r.headingRule(), r.sentenceAbbrevs()), nil } // chunkKey identifies a chunk positionally. type chunkKey struct { chapter, chunkIdx int } // chunkStateResolution is one output unit's resolved chunk-level state: what to report, why, what it // cost and how it got there. It is a struct rather than a 5-tuple because every caller needs a // different subset and a positional return made "which bool was escalated" a reading exercise. type chunkStateResolution struct { State ChunkState Reason string // the deciding flag reason; "" unless State is ChunkFlagged CostUSD float64 // the unit's spend across every stored row (all members, all stages) Escalated bool // any row escalated to a fallback model Skipped int // rows recorded as skipped (an upstream stage flagged) } // unitRows collects one output unit's stored rows in MEMBER order, leader first. The ORDER is // load-bearing, not cosmetic: "the first flagged row decides the unit" only means "the earliest member // that dropped" when the rows arrive in member order — that is what makes status, export and translate // name the SAME reason on a unit where several things went wrong. func unitRows(u editUnit, byChunk map[chunkKey][]store.ChunkStatus) []store.ChunkStatus { var rows []store.ChunkStatus for _, m := range u.Members { rows = append(rows, byChunk[chunkKey{m.Chapter, m.ChunkIdx}]...) } return rows } // droppedMember is one member chunk whose DRAFT flagged, so the edit wave dropped it from the unit // (c-lite, D39.17-fix): the editor still ran over the clean remainder and that text SHIPS, but the unit // is flagged and carries this member's reason and detail. type droppedMember struct { Key chunkKey Reason string Detail string } // memberDrops is the SINGLE definition of the c-lite rule "a dropped member flags its unit", in member // order. translate applies it live (runEditUnit); status applies it by folding the member rows through // resolveChunkState below (a dropped member's flagged draft row is the first flagged row of the unit); // export, which projects only the FINAL row, has no member rows to fold and calls this directly. Before // this helper the three read models agreed only by a comment asking future edits to keep them in step. // // A member with several draft-stage rows contributes its LAST flagged one (the rows are in stored // order), matching the map-overwrite the export projection has always used. func memberDrops(u editUnit, byChunk map[chunkKey][]store.ChunkStatus, draftStages map[string]bool) []droppedMember { var drops []droppedMember for _, m := range u.Members { key := chunkKey{m.Chapter, m.ChunkIdx} var d *droppedMember for _, cs := range byChunk[key] { if draftStages[cs.Stage] && cs.Disposition == string(DispFlagged) { d = &droppedMember{Key: key, Reason: cs.FlagReason, Detail: cs.Detail} } } if d != nil { drops = append(drops, *d) } } return drops } // resolveChunkState maps a unit's per-stage rows (in unitRows order) to its chunk-level state, plus its // cost, escalation flag and (when flagged) the flag reason. stagesTotal is the expected ok-row count. func resolveChunkState(rows []store.ChunkStatus, stagesTotal int) chunkStateResolution { if len(rows) == 0 { return chunkStateResolution{State: ChunkPending} } res := chunkStateResolution{} ok := 0 for _, cs := range rows { res.CostUSD += cs.CostUSD if cs.Escalated { res.Escalated = true } switch cs.Disposition { case string(DispOK): ok++ case string(DispFlagged): // The FIRST flagged stage decides the unit; keep its reason (the rows arrive in member/stage order // with the leader bucket first, so first-wins aligns status with translate/export, which both take // the first flag — a member drop's reason, or the edit's own when the edit flagged). Without the // guard the loop's last-flagged row would win, diverging from the other read-models on a multi-flag unit. if res.State != ChunkFlagged { res.State, res.Reason = ChunkFlagged, cs.FlagReason } case string(DispSkipped): res.Skipped++ } } if res.State == ChunkFlagged { return res } if ok == stagesTotal { res.State = ChunkDone return res } res.State = ChunkInProgress return res } // 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{} for _, cs := range statuses { byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}] = append(byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}], cs) } // Post-check misses + style flags (retrieval_state) — the unit-level signal the the edit-wave editor merged onto // its LEADER chunk's row (a non-leader member carries only its draft injection, post-check=0). 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 } // The shipping granularity is the OUTPUT UNIT (edit unit for an edit pipeline, draft chunk for a // draft-only one) — status projects it like export + the per-unit BookResult. A unit is DONE when every // member draft AND the unit's edit resolved ok, so its expected-ok count is len(members)·|draft stages| + // |edit stages| (a non-leader member has draft rows only; the single edit row lives at the leader). draftStages := r.waveStagesIndexed(waveDraft) editStages := r.waveStagesIndexed(waveEdit) draftStageNames, editStageNames := stageNameSet(draftStages), stageNameSet(editStages) units := r.outputUnits(chunks) // The post-check GATE (opt-in) flags a unit 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 unit as done/pass while `tmctl translate` exits 2 (finding #2). gateOn := r.Pipeline.Gates.Glossary.PostcheckGate rep := &StatusReport{BookID: r.Book.BookID, TotalUnits: len(units)} // The unsigned-bank exposure, same definition as the quality report's (a row with a rendering that is // not approved). Read-only, like everything else here; a read failure says so rather than reporting 0. if rows, gerr := r.Store.GlossaryForBook(r.Book.BookID); gerr != nil { r.Log.WarnContext(ctx, "status: could not read the bank; the unsigned-term count is unknown, not zero", "err", gerr) } else { for _, e := range rows { if e.Status != "approved" && strings.TrimSpace(e.Dst) != "" { rep.UnsignedBankTerms++ } } } passports := map[int]*ChapterPassport{} var chapterOrder []int for _, u := range units { p := passports[u.Chapter] if p == nil { p = &ChapterPassport{Chapter: u.Chapter, Verdict: "pass"} passports[u.Chapter] = p chapterOrder = append(chapterOrder, u.Chapter) } p.UnitsTotal++ leader := chunkKey{u.Chapter, u.FirstChunkIdx} expected := len(u.Members)*len(draftStages) + len(editStages) res := resolveChunkState(unitRows(u, byChunk), expected) state, reason := res.State, res.Reason p.CostUSD += res.CostUSD p.StagesSkipped += res.Skipped miss := missByChunk[leader] // the unit's post-check ran once, at the leader row p.PostcheckMisses += miss rep.PostcheckMisses += miss style := styleByChunk[leader] p.StyleFlags += style rep.StyleFlags += style // Applied repairs (pack-16): the unit's shipping row points at a repair export. Read off the durable // final_hash namespace — no counter column, so a resumed run reports the same number. for _, row := range unitRows(u, byChunk) { if strings.HasPrefix(row.FinalHash, repairDerivedNS+":") { p.RepairApplied++ break } } if gateOn && state == ChunkDone && miss > 0 { // Matches the wave editor: the gate flags only an otherwise-ok unit. 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 res.Escalated { p.Escalations++ rep.Escalations++ } switch state { case ChunkDone: rep.Done++ p.UnitsDone++ case ChunkFlagged: rep.Flagged++ p.UnitsFlagged++ if p.WorstFlagReason == "" || flagReasonSeverity(reason) < flagReasonSeverity(p.WorstFlagReason) { p.WorstFlagReason = reason } case ChunkInProgress: rep.InProgress++ p.UnitsInProgress++ case ChunkPending: rep.Pending++ p.UnitsPending++ } } // Chapter verdicts (exp07 chapter rule: 0 flagged = pass, 1 = attention, ≥2 = fail — over UNITS: a // flagged edit unit counts once, not per member chunk). sort.Ints(chapterOrder) for _, n := range chapterOrder { p := passports[n] switch { case p.UnitsFlagged >= 2: p.Verdict = "fail" case p.UnitsFlagged == 1: p.Verdict = "attention" default: p.Verdict = "pass" } rep.Chapters = append(rep.Chapters, *p) } if rep.TotalUnits > 0 { rep.PercentDone = 100 * float64(rep.Done) / float64(rep.TotalUnits) } // Per-wave snapshot drift (finding #3, wave-aware): the rows carry draft-wave snapshot (draft) + edit-wave snapshot // (edit), so a normal book has TWO snapshots — that is NOT drift. SnapshotDrift means a disagreement // WITHIN a wave (a config changed mid-book without a full re-pin). ConfigDrift compares each wave's // single stored snapshot against the CURRENT projection of THAT wave (materialize the stored glossary // once, then project both). rep.Snapshot reports the SHIPPING wave's snapshot. draftSnaps, editSnaps := map[string]bool{}, map[string]bool{} for _, cs := range statuses { if cs.SnapshotID == "" { continue } switch { case draftStageNames[cs.Stage]: draftSnaps[cs.SnapshotID] = true case editStageNames[cs.Stage]: editSnaps[cs.SnapshotID] = true } } rep.SnapshotDrift = len(draftSnaps) > 1 || len(editSnaps) > 1 finalSnaps := editSnaps if r.finalStageWave() == waveDraft { finalSnaps = draftSnaps } if len(finalSnaps) == 1 { for s := range finalSnaps { rep.Snapshot = s } } if !rep.SnapshotDrift && (len(draftSnaps) > 0 || len(editSnaps) > 0) { if err := r.projectStoredMemory(); err != nil { // Do not stay silent: a projection failure used to be read silently as "no drift" (chain audit). r.Log.WarnContext(ctx, "config-drift check failed; drift state unknown (reported as none)", "err", err) } else { checkWave := func(snaps map[string]bool, w wave) { if len(snaps) != 1 { return } var stored string for s := range snaps { stored = s } cur, _, serr := r.snapshotIDForWave(w) if serr != nil { r.Log.WarnContext(ctx, "config-drift check failed for a wave; drift state unknown", "err", serr) return } if cur != stored { rep.ConfigDrift = true rep.CurrentSnapshot = cur } } checkWave(draftSnaps, waveDraft) checkWave(editSnaps, waveEdit) } } // Re-payment projection (spec §9): what the drift above would COST. Computed whenever rows exist — // not only under ConfigDrift — because SnapshotDrift (rows split across snapshots within one wave) // re-bills too, and that is precisely the case the boolean pair leaves unpriced. $0 and read-only: // projectRebill only re-renders the wave snapshots and sums stored costs. if len(statuses) > 0 { if proj, perr := r.projectRebill(statuses, chunks); perr != nil { // Same discipline as the drift check above: a failed projection is reported, never // silently rendered as "nothing to re-pay". r.Log.WarnContext(ctx, "re-bill projection failed; the re-payment cost of the drift is unknown (reported as none)", "err", perr) } else { rep.RebillUnits, rep.RebillUSD = proj.Rows, proj.USD } } // 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-unit average over the whole book (done + // flagged = a unit fully attempted). NOT book committed: committed also carries partial spend on // IN-PROGRESS units (excluded from the denominator), which would over-estimate (finding #7). The // arithmetic lives in projectBookUSD (rebill.go) because the consent threshold is 5% of THIS number // — one definition, so the threshold can never be computed from a drifted copy of it. processed := rep.Done + rep.Flagged rep.ProjectedBookUSD = projectBookUSD(units, byChunk, len(draftStages), len(editStages)) // 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.TotalUnits - processed if processed > 0 && freshCalls > 0 && remaining > 0 { meanMSPerChunk := float64(totalMS) / float64(processed) rep.ETASeconds = meanMSPerChunk * float64(remaining) / 1000 } // Content-label provenance (B6): only for a book that declares labels, so an unlabelled report keeps // its exact prior bytes. The routing rows come from the shared projection, and the problems (if any) // are the same ones the write path refuses on. if len(r.Book.ContentLabels) > 0 { rep.ContentLabels = r.Book.ContentLabels rep.Routing = r.contentRoutingRows() rep.ContentRoutingProblems = r.Pipeline.ContentProblems } return rep, nil } // projectStoredMemory materializes r.memory from the STORED glossary (read-only: no re-seed, no write) — // the side effect the wave-snapshot projections need. 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) projectStoredMemory() error { rows, err := r.Store.GlossaryForBook(r.Book.BookID) if err != nil { return err } r.memory = membank.Materialize(rows, r.Pipeline.Gates.Glossary.PostcheckGate) return nil } // 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 (package №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 { // Per-wave drift (R1): a stored row carries its WAVE's snapshot (draft rows → draft-wave snapshot, edit // rows → edit-wave snapshot), never the whole-pipeline one — so compare each row against the current // projection of ITS wave (from the SEEDED r.memory seedGlossary just set, matching what the re-run // will render). A whole-pipeline comparison would abort every redrive spuriously. w1cur, _, e1 := r.snapshotIDForWave(waveDraft) if e1 != nil { return summary, nil, fmt.Errorf("pipeline: redrive draft-wave snapshot check: %w", e1) } w2cur, _, e2 := r.snapshotIDForWave(waveEdit) if e2 != nil { return summary, nil, fmt.Errorf("pipeline: redrive edit-wave snapshot check: %w", e2) } draftStageNames := stageNameSet(r.waveStagesIndexed(waveDraft)) for _, cs := range statuses { cur := w2cur if draftStageNames[cs.Stage] { cur = w1cur } if cs.SnapshotID != "" && cs.SnapshotID != cur { return summary, nil, fmt.Errorf("pipeline: redrive aborted — the book rows are under snapshot %.12s but the current config/seed renders %.12s (the config/prompts/capabilities/seed-glossary changed): resetting now would re-pay for the book and destroy the flag telemetry; re-run `translate --resnapshot` to accept the re-payment explicitly, or revert the config/seed", cs.SnapshotID, cur) } } } // Consent to a RE-PAYMENT (D20.2-Q2, rebill.go) — BEFORE the destructive reset, for the same reason // the drift guard is: a refusal must not leave the flag telemetry deleted (external-review 1c). It // runs on BOTH paths, --resnapshot included: that flag skips the snapshot COMPARISON above, never // the question of what the re-pin costs. The targets' own re-attack is not what this projects — that // spend is the command's declared purpose — but the book-wide re-payment a snapshot move brings is. rebillChunks, err := r.bookChunks() if err != nil { return summary, nil, fmt.Errorf("pipeline: redrive re-bill projection: %w", err) } if err := r.checkRebillConsent(ctx, rebillChunks); err != nil { return summary, nil, err } // 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 }