package pipeline import ( "context" "encoding/json" "errors" "fmt" "strings" "sync" "textmachine/backend/internal/checks" "textmachine/backend/internal/chunk" "textmachine/backend/internal/config" "textmachine/backend/internal/lang" "textmachine/backend/internal/membank" "textmachine/backend/internal/obs" "textmachine/backend/internal/runevents" "textmachine/backend/internal/store" ) // waverun.go: the R1 WAVE EXECUTOR — the driver switch from the sequential per-chunk-all-stages loop // (bookrun.go/chunkrun.go, retired) to the precompute pass ($0) → the draft wave (draft stage ∥ over draft CHUNKS) → // the bank-mining stop → the edit wave (edit stage ∥ over edit UNITS). The load-bearing decoupling (plan §1/§2): // the DRAFT is the fine per-chunk unit (COGS/coverage/alignment), the EDIT is the coarse per-unit unit // (a chapter-scale grouping of whole draft chunks — cross-chunk cohesion, D39 point 4). the edit wave reads a unit's draft // as the concatenation of its members' draft-wave outputs and does a FRESH memory.Select over the whole-unit source; // it NEVER re-renders the draft stage (the load-bearing invariant — a re-render over the enriched bank would move // the draft request_hash and re-bill the draft wave). Money stays single-writer-safe (Reserve/Settle serialize); the // waves add only read-only shared state (clients/templates/memory/rate-guards built in the precompute pass). Per-wave // snapshots (draft-wave snapshot base-bank, edit-wave snapshot enriched) keep «the re-payment stays ONE»: a the bank-mining stop enrichment moves // only edit-wave snapshot, so draft-wave checkpoints stay valid. // wavedStage carries a wave stage with its GLOBAL index in Pipeline.Stages — runStage needs the global // index for isFinal (the LAST stage is the shipping stage the sanitizer runs on) and for the >0 sizing // path (a later stage sizes max_tokens from its input draft, not the source, D2.5). type wavedStage struct { st config.Stage idx int } // waveStagesIndexed partitions Pipeline.Stages by role for a wave, carrying each stage's global index // (mirrors waveStages but retains the index runStage needs). the draft wave = translator (draft) stages; the edit wave = every // non-translator (editor/other) stage. Order preserved. func (r *Runner) waveStagesIndexed(w wave) []wavedStage { var out []wavedStage for i, st := range r.Pipeline.Stages { isDraft := st.Role == roleTranslator if (w == waveDraft) == isDraft { out = append(out, wavedStage{st: st, idx: i}) } } return out } // stageNameSet is the set of a wave's stage names — used by the read-models to classify a stored // chunk_status row by wave (its snapshot belongs to that wave's per-wave snapshot). func stageNameSet(staged []wavedStage) map[string]bool { m := make(map[string]bool, len(staged)) for _, ws := range staged { m[ws.st.Name] = true } return m } // outputUnits is the SHIPPING granularity the read-models (export/status) project — matching the per-unit // BookResult: edit units (member draft chunks grouped) for an edit pipeline, or one singleton unit per // draft chunk for a draft-only pipeline (the draft itself ships). One helper so export and status agree. func (r *Runner) outputUnits(chunks []chunk.Chunk) []editUnit { if r.finalStageWave() == waveEdit { return buildEditUnits(chunks) } out := make([]editUnit, len(chunks)) for i, ch := range chunks { out[i] = editUnit{EditUnitID: ch.EditUnitID, Chapter: ch.Chapter, FirstChunkIdx: ch.ChunkIdx, Members: []chunk.Chunk{ch}} } return out } // WaveSignatureStop is the typed sentinel the driver returns when bank-mining proposes a delta holding // terms no earlier stop has presented (the D39.144 flag model — see mining.go): the run completes the // draft wave, writes the owner signature map, records the map in the presented memory, and STOPS before // the edit wave (an explicit stop boundary, not a mid-run append). The owner reviews the map, decides // whatever they wish through `tmctl bank-apply` — or nothing — and re-runs: the draft wave resumes at // $0, the stop does not re-fire on what it already presented, and the edit wave runs with the undecided // rows riding as unsigned auto-bank surfaces, used on the wire like any other row. NOT an infra failure — a deliberate // human-in-the-loop pause. The CLI maps it (errors.As) to a DISTINCT exit code (3, unlike // CompletedWithFlags's 2 or a crash's 1) and renders the terms + signature-map path to the operator // (main.go/renderSignatureStop, R1-FL-A). type WaveSignatureStop struct { Terms int SignaturePath string // TablePath is the sidecar holding the FULL verification table; Rows is that same table in memory, so // the CLI can print a capped view without re-reading the file (the cap exists because emitRankCap is // 200 and a 200-term dump is not a review surface). TablePath string Rows []BankStopRow } func (e *WaveSignatureStop) Error() string { return fmt.Sprintf("bank-mining stopped on new terms — the signature map %s holds %d term(s), new and still-undecided together; decide via `tmctl bank-apply` (or not), then resume (the edit wave)", e.SignaturePath, e.Terms) } // translateBookWaves is the wave driver (R1). It runs the draft wave (draft ∥), the bank-mining stop, the edit wave (edit ∥) and // assembles the BookResult. chunks + stickySel come from the precompute pass (chunk.SplitChunks + precomputeSticky), computed by // the caller (TranslateBook) after ingest/seed/eager-build. Returns a *WaveSignatureStop when the bank-mining stop stops. func (r *Runner) translateBookWaves(ctx context.Context, chunks []chunk.Chunk, stickySel []membank.Selection, scope *volumeScope) (*BookResult, error) { draftStages := r.waveStagesIndexed(waveDraft) editStages := r.waveStagesIndexed(waveEdit) workers := r.Pipeline.Waves.Workers editWave := len(editStages) > 0 // a real editor wave exists; else the draft IS the shipping output (draft-only) // The run-event counters (row 103), seeded from what the store already holds: the same output units // and the same per-wave arithmetic `status` projects, so the stream and the resync channel can never // quote two different numbers. r.beginRunEvents(ctx, chunks) // --- the draft wave: draft stage(s) ∥ over draft chunks under draft-wave snapshot (base-bank version) --- draftSnapshot, draftPayload, err := r.snapshotIDForWave(waveDraft) if err != nil { return nil, err } if err := r.Store.UpsertSnapshot(draftSnapshot, r.Book.BriefHash(), draftPayload); err != nil { return nil, fmt.Errorf("pipeline: upsert draft-wave snapshot %.12s: %w", draftSnapshot, err) } r.Log.InfoContext(ctx, "draft wave started", "snapshot", draftSnapshot[:12], "chunks", len(chunks), "workers", workers, "draft_stages", len(draftStages)) draftResults := make([]stageSeqResult, len(chunks)) // ⛔ WHICH CHUNKS ACTUALLY RAN, as its own fact rather than as a shape of draftResults. The zero // stageSeqResult is indistinguishable from a real one that flagged nothing and cost nothing, and the // difference stopped being academic the moment a wave could end with items unstarted: the edit wave // reads a member's draft out of a map built over ALL chunks, so an unstarted member arrives as a // present-and-clean empty draft and the editor is PAID to edit a hole. drafted := make([]bool, len(chunks)) draftUnits := newDraftUnitTracker(r.outputUnits(chunks), chunks) if err := r.runWave(ctx, workers, len(chunks), func(ctx context.Context, i int) error { // The VOLUME ceiling (volume.go), applied where the item BEGINS rather than after it: a chunk whose // output unit is outside this run's granted scope is never started, so no ceiling can be overshot by // one call. A nil scope means no ceiling is in force and this is a no-op. if !scope.allows(chunks[i]) { return nil } d, err := r.runDraftChunk(ctx, draftSnapshot, chunks[i], stickySel[i], draftStages, !editWave) if err != nil { return err } draftResults[i] = d drafted[i] = true // The wave fans out over CHUNKS but the seam counts OUTPUT UNITS (the manifest's granularity, and // therefore the reader's): a unit is announced by whichever member finishes last. if u, done := draftUnits.memberDone(chunks[i]); done { shipped, flagged, reason := draftUnits.outcome(u, draftResults) r.events.unitResolved(runevents.WaveDraft, u.Chapter, u.FirstChunkIdx, shipped, flagged, reason) } return nil }); err != nil { var halt *CeilingHalt if !errors.As(err, &halt) { return nil, err } // ⛔ THE MONEY STOPPED IN THE DRAFT WAVE, SO THE RUN ENDS HERE — it does not go on to the mining // stop or the edit wave, and this is a correctness rule rather than thrift. The edit wave looks a // member's draft up in a map built over EVERY chunk, so an unstarted member is found (`ok` is // true) holding the zero result: not flagged, empty text. It would be joined into the unit's draft // as a clean member and the editor would be PAID to edit a unit with a hole in it, then ship it. // The mining stop is the same class one level up: its input is the draft of the WHOLE book, and // re-seeding the bank from a partial one moves the edit-wave snapshot for every later run. r.reportEvicted(ctx, "draft") return r.moneyStoppedResult(ctx, chunks, drafted, draftResults, editWave), err } r.reportEvicted(ctx, "draft") // --- the bank-mining stop: bank-mining stop boundary (auto-continues when mining is not configured) --- if stopped, err := r.runBankMiningStop(ctx, chunks, draftSnapshot, editWave); err != nil { return nil, err } else if stopped { // The COUNT travels; the table does not (row 101 ships it as an artifact — a thousand rows are // not an event). r.events.emit(runevents.TypeBankStop, runevents.BankStop{TermsProposed: r.lastMinedCount}) return nil, &WaveSignatureStop{ Terms: r.lastMinedCount, SignaturePath: r.signatureMapPath(), TablePath: r.bankStopTablePath(), Rows: r.lastBankStopRows, } } res := &BookResult{BookID: r.Book.BookID} // The stop NAMES its ceiling (D39.165 §1б: OpenHands' silent iteration limit is the anti-pattern being // avoided). It rides on the result rather than on an error because a volume stop is a COMPLETION — the // run did what was bought — so it must not reach the exit-code mapper at all. Set when the grant held // work back OR when the run did work outside it (bound()): a run granted more than the book had left, // and doing nothing beyond it, simply finished and reports nothing here. // ⚠ ASKED BEFORE reconcile, AND THAT ORDER IS LOAD-BEARING — found by planting, and it is the kind of // thing a tidy-up removes without noticing. bound() reads the PLAN's counters; reconcile then trues them // up against what actually happened, and a carried unit that came back FLAGGED has its Carried moved to // Flagged. So a run whose ONLY work outside the grant was that unit has Carried==0 by the time reconcile // is done: deciding res.Volume from the trued-up counters would leave it silent about a unit it paid for // and finished outside its grant — the disclosure this report exists for. The natural refactor «report // from the reconciled numbers» is therefore wrong, and TestReconcileMovesAFlaggedCarryOutOfCarried is // where a reader meets the consequence. if scope.bound() { res.Volume = &scope.stop // The log takes the same fork as the operator line for the same reason: a run that reached the end // of the book did NOT stop short of it, and saying so of every bounded run makes the sentence // worthless exactly where the grant behaved unusually. msg := "the run is bounded by the VOLUME ceiling: it will stop having done what was granted, not at the end of the book" if scope.stop.Left() == 0 { msg = "the run reached the end of the book under a VOLUME ceiling that held nothing back; it is reported because units an earlier run began were finished outside this grant" } r.Log.InfoContext(ctx, msg, "book", r.Book.BookID, "max_units", scope.stop.MaxUnits, "paid_units", scope.stop.Paid(), "delivered_new", scope.stop.Delivered, "re_made", scope.stop.Reworked, "finished_from_an_earlier_grant", scope.stop.Carried, "free", scope.stop.Free, "left_never_delivered", scope.stop.LeftFresh, "left_delivered_not_re_made", scope.stop.LeftRework) } // The terminologist's spend is THIS run's spend and belongs in this run's total. It is not a chunk cost, // so the per-chunk accumulation below cannot see it — and a paid call that no total reports is a call // the operator cannot notice. (Zero when the gate is off or the calls replayed from checkpoints.) res.TotalUSD += r.lastTerminology.CostUSD // --- Draft-only pipeline: the draft IS the shipping output; assemble per draft chunk (no edit units) --- if !editWave { for i, ch := range chunks { if !scope.allows(ch) { continue // outside the volume scope: never drafted, so there is no outcome to assemble } oc := r.draftOnlyOutcome(ch, draftResults[i]) res.Chunks = append(res.Chunks, oc) res.TotalUSD += oc.CostUSD if oc.Disposition == DispFlagged { res.Flagged++ } } scope.reconcile(res.Chunks) r.Log.InfoContext(ctx, "book run finished (draft-only)", "book", r.Book.BookID, "chunks", len(res.Chunks), "flagged", res.Flagged, "run_usd", fmt.Sprintf("%.6f", res.TotalUSD)) return res, nil } // --- the edit wave: edit stage(s) ∥ over edit units under edit-wave snapshot (enriched-bank version) --- editSnapshot, editPayload, err := r.snapshotIDForWave(waveEdit) if err != nil { return nil, err } if err := r.Store.UpsertSnapshot(editSnapshot, r.Book.BriefHash(), editPayload); err != nil { return nil, fmt.Errorf("pipeline: upsert edit-wave snapshot %.12s: %w", editSnapshot, err) } units := buildEditUnits(chunks) // ⛔ THE PLAN IS RE-JUDGED HERE, because the bank-mining stop above may have re-seeded the bank and // moved the snapshot this wave runs under. Units the plan called FREE were admitted outside the grant; // any that the new bank makes paying must take a slot or not run. See rescopeEditWave. if err := r.rescopeEditWave(ctx, scope, units, chunks, stickySel, editSnapshot); err != nil { return nil, err } // ⚠ AND THE RE-PLAN IS NOT OPTIONAL: the wave refuses to run a scope that was planned against a // different snapshot than the one it is about to use. Without this the call above is merely a call — // deleting it leaves every test green while the ceiling silently stops holding on any book that // re-seeds mid-run. With it, the invariant is the code's own, and removing the re-plan makes the run // say so instead of overspending quietly. if scope != nil && scope.editSnapshot != editSnapshot { return nil, fmt.Errorf("pipeline: the volume scope was planned against edit-wave snapshot %.12s but the wave is running under %.12s — units admitted as free were judged on ground this run has since replaced, and paying for them would escape the ceiling (wave sequencing bug: rescopeEditWave did not run)", scope.editSnapshot, editSnapshot) } draftByKey := make(map[chunkKey]stageSeqResult, len(chunks)) for i, ch := range chunks { draftByKey[chunkKey{ch.Chapter, ch.ChunkIdx}] = draftResults[i] } r.Log.InfoContext(ctx, "edit wave started", "snapshot", editSnapshot[:12], "units", len(units), "workers", workers, "edit_stages", len(editStages)) unitOutcomes := make([]*ChunkOutcome, len(units)) if err := r.runWave(ctx, workers, len(units), func(ctx context.Context, i int) error { // Same admission as the draft wave, and the SAME set: a unit is either in this run's scope for both // waves or in neither, so the edit wave can never be handed a unit whose members were not drafted. if !scope.allowsUnit(units[i]) { return nil } oc, err := r.runEditUnit(ctx, editSnapshot, units[i], draftByKey, editStages) if err != nil { return err } unitOutcomes[i] = oc r.events.unitResolved(runevents.WaveEdit, oc.Chapter, oc.ChunkIdx, oc.FinalText != "", oc.Disposition == DispFlagged, string(oc.FlagReason)) return nil }); err != nil { var halt *CeilingHalt if !errors.As(err, &halt) { return nil, err } foldUnitOutcomes(res, unitOutcomes) // The drafts of units that never reached an edit outcome were paid for by THIS run and ship // nothing; a total that omits them under-reports the run's own spend on the one line an operator // reads. (Zero when those drafts were replayed from an earlier run's checkpoints — cost is this // run's spend, not the book's.) for i, u := range units { if unitOutcomes[i] != nil { continue } for _, m := range u.Members { res.TotalUSD += draftByKey[chunkKey{m.Chapter, m.ChunkIdx}].cost } } r.noteMoneyStop(ctx, res) return res, err } foldUnitOutcomes(res, unitOutcomes) // The plan's counters become the run's counters only now, corrected by what actually shipped. scope.reconcile(res.Chunks) r.Log.InfoContext(ctx, "book run finished", "book", r.Book.BookID, "units", len(res.Chunks), "flagged", res.Flagged, "run_usd", fmt.Sprintf("%.6f", res.TotalUSD)) return res, nil } // runWave fans `n` items out over `workers` goroutines, calling work(ctx, i) for each index. The first // error cancels the rest (via a derived context) and is returned; work items are independent (a wave has no // shared mutable input), and money/store writes serialize through the single-writer pool, so this is // race-clean. Results are written by the callback into a caller-owned slice indexed by i, so the result // ORDER is deterministic (item index) regardless of completion order — the golden capture sorts the wire/log // side-effects separately. workers ≤ 0 is treated as 1 (LoadPipeline already clamps). // // PARENT CANCELLATION (Ctrl-C / SIGTERM — the CLI wires a signal.NotifyContext) is surfaced as an error too, // NOT a silent nil: the feeder drops the remaining items on ctx.Done(), and a worker need not observe the // cancellation to return (a $0 resume worker resolves entirely through its own opContext store reads, never // touching the cancellable ctx), so firstErr can stay nil while items are left UNDONE. Returning nil then // would let the caller index uninitialised result slots — a nil-deref on the edit-wave `*oc` assembly, or a // draft-only run reporting empty translations as exit-0 success. So after the workers drain, a still-nil // firstErr defers to parent.Err(): nil on a clean run, context.Canceled on an interrupted one (the durable // store stays correct — the undone items were never checkpointed, so a re-run resumes them). func (r *Runner) runWave(parent context.Context, workers, n int, work func(ctx context.Context, i int) error) error { if workers < 1 { workers = 1 } if n == 0 { return nil } ctx, cancel := context.WithCancel(parent) defer cancel() idxCh := make(chan int) // latched is the SPEND CEILING's stop, and it is a different verb from cancel(). Closing it makes the // feeder stop handing out indices — nothing new begins — while every worker already holding an item // runs it to completion under a context nobody cancelled. That is the whole of backlog row 277: a // refused reservation used to reach `fail` like any other error, cancel the derived context and kill // the W-1 calls the money WAS there for, so a book whose ceiling could not admit the W'th call // produced nothing at all. // // A channel and not a boolean, because the feeder BLOCKS on `idxCh <- i` when every worker is busy — // which is exactly the state a ceiling stop is reached in. A flag would be read after the send it is // meant to prevent. latched := make(chan struct{}) var latchOnce sync.Once var wg sync.WaitGroup var mu sync.Mutex var firstErr error // haltErr is the ceiling's OWN slot, and the separation is load-bearing. A ceiling no longer cancels, // so siblings run on and one of them may hit a genuine infra failure afterwards. Sharing one slot // would make whichever happened first speak for both: a first-wins ceiling would report a broken run // as «paused, add money», and a first-wins failure would erase a money fact the platform's contract // says must never be recorded as `failed` (PD-113). var haltErr error // A crash gets its OWN slot rather than competing for firstErr, and outranks it on the way out. Both // halves are load-bearing. Unrecovered, a worker panic takes the process down through the runtime's // handler, which exits 2 — the code the shell contract reserves for "completed with flags", so a run // that died mid-book was recorded as `ready` (row 176). But routing it through first-wins would only // move the lie: a sibling that already failed cancels the wave, and a panic in the code nobody expected // to run under cancellation would then be discarded with its stack, leaving the run to depart as the // sibling's ceiling halt (exit 4, `paused`) or cancellation (exit 5). A process that crashed is not // paused and did not stop gracefully. var panicErr error fail := func(err error) { var halt *CeilingHalt if errors.As(err, &halt) { mu.Lock() if haltErr == nil { haltErr = err } mu.Unlock() // ⚠ THE MONEY FACT IS PUBLISHED HERE, where it happens, and not only from the error the driver // eventually returns. PD-113 was closed with two channels that fail INDEPENDENTLY — the event // and the exit code — and if a sibling's infra failure outranks the halt on the way out (it // does, see below), deriving the event from the returned error alone would let one broken // worker take BOTH channels, which is the single failure mode that fix exists to rule out. // terminal() will not emit a second one. r.events.ceilingReached(halt) latchOnce.Do(func() { close(latched) }) return } mu.Lock() if firstErr == nil { firstErr = err cancel() } mu.Unlock() } failPanic := func(err error) { mu.Lock() if panicErr == nil { panicErr = err } mu.Unlock() cancel() } for w := 0; w < workers; w++ { wg.Add(1) go func() { defer wg.Done() // NOT obs.SafeGo: this recover does not let the run continue, it makes the run fail loudly. defer func() { if p := recover(); p != nil { failPanic(obs.NewPanicError("wave worker", p)) } }() for i := range idxCh { if err := work(ctx, i); err != nil { fail(err) return } } }() } for i := 0; i < n; i++ { // ⚠ THE LATCH IS ASKED FIRST, NON-BLOCKINGLY, AND THAT IS NOT BELT-AND-BRACES. Go chooses // UNIFORMLY AT RANDOM among the ready cases of a select, so with a worker idle and `latched` // closed, the three-way select below hands out an index about half the time — and the wave would // keep starting items after the ceiling said stop, up to one per idle worker. The same shape has // always been true of `ctx.Done()` here; on the money path it is the difference between «nothing // new was started» and «a few more things were». select { case <-latched: continue default: } select { case idxCh <- i: case <-ctx.Done(): case <-latched: // The spend ceiling stopped admissions. Drain the remaining indices WITHOUT offering them: // the loop runs to n so the channel is closed exactly once below, and every skipped item is // simply not started — it has no checkpoint, no chunk_status row, and a resume picks it up. } } close(idxCh) wg.Wait() if panicErr != nil { // ⚠ THE CRASH RANKS, BUT IT DOES NOT ERASE. A worker that panicked while a SIBLING had already // caught a spend ceiling used to depart carrying the crash alone, and with it went the money fact: // `errors.As(err, &CeilingHalt)` went false, the driver's terminal() fell through to `failed` and // emitted no `ceiling` event, and the platform — which reads that event first and lets it survive // ANY exit (runs/reconcile.go, «the fact does not become less true») — recorded a run stopped by a // ceiling as failed, which PD-113 forbids (fix list ФЧ-7). The redundancy PD-113 was closed with is // two channels that fail INDEPENDENTLY; one panic took both. // // So both facts travel and the CONSUMERS rank, which they already do: exitCode checks the panic // first («even if a layer joined it with one») and returns 1, and terminal() does the same before // it looks for a halt. A crashed process is still not reported as paused — the outcome mirrors the // exit code — while the money that ran out stays on the record. // // ⚠ ONLY A CEILING JOINS, and the narrowness is the rule rather than a shortcut. A sibling's // ceiling is a durable statement about MONEY that happened BEFORE the crash and has a channel of // its own to reach. Every other firstErr is either an infra failure — about which the crash // already says everything — or the context.Canceled the panic itself caused, and carrying THAT // would let a crash read as the graceful stop it triggered, which is the masking trap // TestWaveWorkerPanicSurfacesAsAFailingError exists to hold. if haltErr != nil { return errors.Join(panicErr, haltErr) } return panicErr } if firstErr != nil { // ⛔ A REAL FAILURE OUTRANKS THE CEILING, AND THEY ARE NOT JOINED. The panic above joins because a // panic has a TYPE that exitCode and terminal() both check FIRST, so the ranking survives the // join. An infra failure has no type: `errors.As(join, &CeilingHalt)` would match, the run would // depart exit 4 and the platform would record `paused` for a run that BROKE, and the remedy the // operator is handed («add money») would be the wrong one. So the exit code tells the truth about // the failure, and the money fact travels on the channel it was given for exactly this case — the // `ceiling` event, already written by fail() at the moment of the refusal. return firstErr } if haltErr != nil { return haltErr } // No worker failed, but the PARENT may have been cancelled (Ctrl-C), which the feeder honoured by // dropping items — surface that as an error so the driver never assembles a partial result as success. return parent.Err() } // stageSeqResult is the outcome of running one chunk/unit through an ORDERED list of stages (one wave): // the per-stage results, the final OK stage's text, and the terminal flag state. type stageSeqResult struct { stages []StageResult finalText string // the last OK stage's Text (empty until a stage runs; unused when flagged) flagged bool flagReason FlagReason recovered string // sanitizer-stripped export text from the flagging stage (D35.4a) cost float64 // THIS run's spend across the sequence bankTelem bankFlags // the translator-role stage's banknote telemetry (WS4 point 10) bankProps string // JSON of that stage's PARSED banknote entries (the WHAT channel, D39.36 fix) repair repairResult // the shipping stage's repair outcome (pack-16); zero when the gate is off } // runStageSequence runs `staged` in order over one chunk/unit, feeding each stage's output to the next as // `prev` (starting from startPrev), stopping the sequence at the FIRST flagged stage (later stages are // recorded `skipped` — no paid edit over a garbage draft, D2). It is the generalized inner loop of the // retired translateChunk, reused by BOTH waves (draft over a chunk, the edit-wave editor over a unit). injectionByRole // maps a stage role to its already-rendered memory-injection message (the translator's src→dst glossary in // the draft wave, the editor's CONFIRMED-dst constraint block in the edit wave). Deterministic; content-verified resume via runStage. func (r *Runner) runStageSequence(ctx context.Context, staged []wavedStage, snapID string, ch chunk.Chunk, startPrev string, injectionByRole map[string]string, injected []membank.PickedEntry) (stageSeqResult, error) { var res stageSeqResult prev := startPrev flagged := false var flagReason FlagReason recovered := "" for _, ws := range staged { st := ws.st if flagged { detail := fmt.Sprintf("skipped: an upstream stage was flagged (%s)", flagReason) if err := r.Store.UpsertChunkStatus(store.ChunkStatus{ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name, SnapshotID: snapID, Disposition: string(DispSkipped), FlagReason: string(flagReason), Detail: detail, }); err != nil { return res, fmt.Errorf("pipeline: record skipped chunk_status: %w", err) } res.stages = append(res.stages, StageResult{ Stage: st.Name, Role: st.Role, Model: st.ResolvedModel, Disposition: DispSkipped, FlagReason: flagReason, Detail: detail, }) continue } sr, err := r.runStage(ctx, st, ws.idx, snapID, ch, prev, injectionByRole[st.Role], injected) if err != nil { return res, err } res.stages = append(res.stages, *sr) res.cost += sr.CostUSD if sr.Repair.touched() { res.repair = sr.Repair } if st.Role == roleTranslator { res.bankTelem, res.bankProps = sr.BankFlags, sr.BankProposals // FL-2: the resume fast-path serves the banknote-cleaned final_hash and leaves BankFlags zero; // restore the telemetry the fresh run persisted so persistRetrievalState does not zero the three // banknote columns on every resume of a ready chunk (guarded on an empty BankFlags so the // attempt-loop path — which re-derives telemetry from the raw checkpoint — is never overwritten). if res.bankTelem == (bankFlags{}) && sr.FromResume && r.Pipeline.Gates.Banknote.Enabled { if prevRS, err := r.Store.GetRetrievalState(r.Book.BookID, ch.Chapter, ch.ChunkIdx); err == nil && prevRS != nil { res.bankTelem = bankFlags{ NLines: prevRS.NBanknoteLines, ParseFail: prevRS.BanknoteParseFail != 0, Truncated: prevRS.BanknoteTruncated != 0, } // The proposals restore with the counters and for the same reason: the fast path never // touches the raw checkpoint, so re-writing the row from a resumed run would erase the // very WHAT the signature map is about to read. res.bankProps = prevRS.BanknoteDetail } } } if sr.Disposition == DispFlagged { flagged = true flagReason = sr.FlagReason recovered = sr.RecoveredText continue } prev = sr.Text } res.finalText = prev res.flagged = flagged res.flagReason = flagReason res.recovered = recovered return res, nil } // runDraftChunk runs the draft stage(s) over one chunk under the draft-wave snapshot (plan §1). It renders // the translator's src→dst glossary injection from the precomputed per-chunk selection over the BASE bank // (baseMemory — mined-excluded, so the injection is stable across a bank-mining enrichment), runs the // sequence, and persists the draft retrieval_state (injection counts + banknote telemetry). When this is the // FINAL wave (a draft-only pipeline with no editor), the draft IS the shipping text, so the post-check + // cheap gates run here over the draft; otherwise they belong to the edit wave (the edit unit's final output). func (r *Runner) runDraftChunk(ctx context.Context, draftSnapshot string, ch chunk.Chunk, memSel membank.Selection, draftStages []wavedStage, isFinalWave bool) (stageSeqResult, error) { injectionByRole := map[string]string{} if r.baseMemory != nil { // Serialize the per-role injection blocks via the role→renderer registry (D39 layer 7) from the // precomputed base-bank selection: the translator gets its src→dst glossary block; any other // role's block is rendered too but consumed only if a stage of that role runs in this wave. tx := lang.InjectionTextsFor(r.Book.TargetLang) for role, render := range roleInjectionRenderers { injectionByRole[role] = render(memSel.Injected, tx) // pure → map-order-independent } if len(memSel.TrustGated) > 0 { r.Log.WarnContext(ctx, "memory: lower-trust longer key refused from suppressing a higher-trust nested key (approved term preserved; reconcile the seed)", "chapter", ch.Chapter, "chunk", ch.ChunkIdx, "trust_gated", len(memSel.TrustGated), "first", memSel.TrustGated[0].Suppressor+"⊃"+memSel.TrustGated[0].Protected) } } seq, err := r.runStageSequence(ctx, draftStages, draftSnapshot, ch, "", injectionByRole, memSel.Injected) if err != nil { return seq, err } var postMisses membank.PostcheckResult outputChecked := false var cheap checks.CheapGateResult if isFinalWave && r.baseMemory != nil && !seq.flagged && seq.finalText != "" { // Draft-only pipeline: the draft is the shipping output → run the FINAL-output checks here, over the // SAME base bank whose terms the draft was injected with (post-check parity with the injection). outputChecked = true postMisses = r.baseMemory.Postcheck(memSel.Injected, seq.finalText) if r.Pipeline.Gates.Glossary.PostcheckGate && postMisses.ConfirmedCount() > 0 { seq.flagged = true seq.flagReason = FlagGlossaryMiss r.Log.WarnContext(ctx, "glossary post-check gate flagged the chunk", "chapter", ch.Chapter, "chunk", ch.ChunkIdx, "confirmed_misses", postMisses.ConfirmedCount()) } } if isFinalWave && !seq.flagged && seq.finalText != "" && r.checkers.TargetActive() { cheap = checks.RunCheapGates(ch.Text, seq.finalText, seq.finalText, r.cheapGateConfig()) } var voice checks.VoiceResult var leaks []membank.SpoilerLeak if isFinalWave && !seq.flagged && seq.finalText != "" { voice = r.runVoiceChecks(r.baseMemory, ch.Chapter, ch.Text, seq.finalText) leaks = r.spoilerLeaks(r.baseMemory, memSel, seq.finalText) } if r.baseMemory != nil { if err := r.persistRetrievalState(draftSnapshot, ch, memSel, postMisses, outputChecked, cheap, seq.bankTelem, seq.bankProps, voice, leaks); err != nil { return seq, err } } return seq, nil } // runEditUnit runs the editor stage(s) over one edit unit under the edit-wave snapshot (plan §1/§2). It reads // the unit's draft as the concatenation of its member chunks' draft outputs (NEVER re-rendering the draft // stage — the load-bearing invariant), does a FRESH memory.Select over the unit source over the ENRICHED bank // (sticky degenerate at unit scope → nil), and runs the editor over (unit source, unit draft). Post-check + // cheap gates run over the edit output at unit granularity, merged into the LEADER chunk's retrieval_state row. // // A flagged member draft is DROPPED from the edit, not blanked-whole-unit (c-lite, D39.17-fix): the editor // still runs over the CLEAN members' source+draft, so the good (already-paid) drafts are edited and shipped — // only when EVERY member flagged does the unit ship "". The edit row records the edit's honest outcome; the // UNIT is flagged whenever a member dropped (or the edit itself flagged), a fact the export/status read-models // re-derive from the member draft rows (status.resolveChunkState) so all three agree. The pre-c-lite behaviour // (any flagged member → whole unit blank) discarded a chapter-scale unit + its paid sibling drafts over one bad // chunk — a regression from the retired per-chunk model, closed here. func (r *Runner) runEditUnit(ctx context.Context, editSnapshot string, unit editUnit, draftByKey map[chunkKey]stageSeqResult, editStages []wavedStage) (*ChunkOutcome, error) { out := &ChunkOutcome{Chapter: unit.Chapter, ChunkIdx: unit.FirstChunkIdx, Disposition: DispOK} var draftStages []StageResult var draftCost float64 var cleanSources []string // the non-flagged members' source (the editor's {{text}}, aligned to the clean draft) var draftParts []string // the non-flagged members' draft (the editor's input) memberFlagged := false var memberFlagReason FlagReason for _, m := range unit.Members { d, ok := draftByKey[chunkKey{m.Chapter, m.ChunkIdx}] if !ok { return nil, fmt.Errorf("pipeline: edit unit ch%d chunk%d: missing draft result (wave sequencing bug)", m.Chapter, m.ChunkIdx) } draftStages = append(draftStages, d.stages...) draftCost += d.cost if d.flagged { if !memberFlagged { memberFlagged = true memberFlagReason = d.flagReason } out.DroppedMembers++ // counted for EVERY drop, including when the edit itself flags too out.DroppedReason = memberFlagReason // the HOLE's cause, which out.FlagReason may not be continue // the flagged member is DROPPED from the edit — not skipped-whole-unit (c-lite) } cleanSources = append(cleanSources, m.Text) draftParts = append(draftParts, d.finalText) } out.CostUSD = draftCost // Only when EVERY member draft flagged is there nothing clean to edit → skip the edit and ship "" (the whole // unit is unusable). Record a skipped edit chunk_status at the leader so the read-models see a flagged unit. if len(draftParts) == 0 { leader := chunk.Chunk{Chapter: unit.Chapter, ChunkIdx: unit.FirstChunkIdx} skipped, err := r.recordSkippedStages(ctx, editStages, editSnapshot, leader, memberFlagReason) if err != nil { return nil, err } out.Stages = append(draftStages, skipped...) out.Disposition = DispFlagged out.FlagReason = memberFlagReason out.FinalText = "" return out, nil } // The edit is ADDRESSED at the manifest leader (chapter, FirstChunkIdx — where the read-models look for the // unit's final row) but its SOURCE + DRAFT are the CLEAN members only. With NO flagged member cleanSources // join == unit.sourceText() byte-for-byte, so the normal path (and the golden) is unchanged. leader := chunk.Chunk{Chapter: unit.Chapter, ChunkIdx: unit.FirstChunkIdx, Text: strings.Join(cleanSources, unitJoinSeparator)} unitDraft := strings.Join(draftParts, unitJoinSeparator) var editSel membank.Selection injectionByRole := map[string]string{} if r.memory != nil { // FRESH Select over the WHOLE-unit source over the ENRICHED bank (§1(б)/F3): NOT the the precompute pass per-chunk // memSel (that was over the base bank). Sticky is degenerate at unit scope (a unit == a chapter or a // sub-chapter split, so the intra-chapter sticky window does not apply) → nil sticky_prev. editSel = r.memory.Select(leader.Text, unit.Chapter, nil, r.Pipeline.Context.GlossaryTokenBudget) // Render the per-role injection from the ENRICHED unit selection via the registry (D39 layer 7): the // editor gets its CONFIRMED-dst constraint block; other roles' blocks are rendered but consumed // only if a stage of that role runs in the edit wave. tx := lang.InjectionTextsFor(r.Book.TargetLang) for role, render := range roleInjectionRenderers { injectionByRole[role] = render(editSel.Injected, tx) } } seq, err := r.runStageSequence(ctx, editStages, editSnapshot, leader, unitDraft, injectionByRole, editSel.Injected) if err != nil { return nil, err } out.Stages = append(draftStages, seq.stages...) out.CostUSD += seq.cost finalText, flagged, flagReason, recovered := seq.finalText, seq.flagged, seq.flagReason, seq.recovered var postMisses membank.PostcheckResult outputChecked := false if r.memory != nil && !flagged && finalText != "" { outputChecked = true postMisses = r.memory.Postcheck(editSel.Injected, finalText) if r.Pipeline.Gates.Glossary.PostcheckGate && postMisses.ConfirmedCount() > 0 { flagged = true flagReason = FlagGlossaryMiss r.Log.WarnContext(ctx, "glossary post-check gate flagged the edit unit", "chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx, "confirmed_misses", postMisses.ConfirmedCount()) } } var cheap checks.CheapGateResult if !flagged && finalText != "" && r.checkers.TargetActive() { cheap = checks.RunCheapGates(leader.Text, unitDraft, finalText, r.cheapGateConfig()) if cheap.Total() > 0 { r.Log.InfoContext(ctx, "cheap style gates flagged the edit unit (observability, not a gate)", "chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx, "style_flags", cheap.Total()) } } // The pack-19 flaggers, on the same shipped text and under the same "clean unit" condition as the // cheap gates: a flagged unit ships nothing, so measuring its prose would report defects nobody reads. var voice checks.VoiceResult var leaks []membank.SpoilerLeak if !flagged && finalText != "" { voice = r.runVoiceChecks(r.memory, unit.Chapter, leader.Text, finalText) if voice.Total() > 0 { r.Log.InfoContext(ctx, "voice flagger fired on the edit unit (observability, not a gate)", "chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx, "voice_flags", voice.Total(), "attributed_replies", voice.Attributed, "version", checks.VoiceCheckVersion) } if leaks = r.spoilerLeaks(r.memory, editSel, finalText); len(leaks) > 0 { r.Log.WarnContext(ctx, "spoiler leak: a rendering this chapter must not know yet reached the output", "chapter", unit.Chapter, "leader_chunk", unit.FirstChunkIdx, "leaks", len(leaks), "first", leaks[0].Src) } } if r.memory != nil { if err := r.mergeUnitRetrievalState(unit.Chapter, unit.FirstChunkIdx, postMisses, outputChecked, cheap, voice, leaks); err != nil { return nil, err } } // Unit disposition: the EDIT itself flagged → ship its recovered/""; else a MEMBER dropped → still ship the // edited clean remainder, flagged for review (c-lite); else clean → ship the edit output ok. switch { case flagged: out.Disposition = DispFlagged out.FlagReason = flagReason out.FinalText = r.checkers.ExportNormalize(recovered) case memberFlagged: out.Disposition = DispFlagged out.FlagReason = memberFlagReason out.FinalText = r.checkers.ExportNormalize(finalText) default: out.FinalText = r.checkers.ExportNormalize(finalText) } // Title policy (pack-13): prepend the chapter's deterministic «Глава N» to its FIRST unit's non-empty // export text. unit.Members[0] is the unit's leader chunk; only a chapter's opening chunk (ChunkIdx 0) // carries a Heading, so non-leader units are a no-op. Applied to the export projection identically in // export.go — both derive the heading from the same deterministic chunker, so translate and export agree. out.FinalText = chunk.ApplyHeading(unit.Members[0].Heading, out.FinalText) return out, nil } // recordSkippedStages writes a `skipped` chunk_status row (+ a skipped StageResult) for each stage in a // wave, at chunk `ch` under `snapID` — used when a unit's member draft flagged, so the unit's edit is // skipped (mirrors runStageSequence's flagged-skip path but for a pre-decided skip). func (r *Runner) recordSkippedStages(ctx context.Context, staged []wavedStage, snapID string, ch chunk.Chunk, flagReason FlagReason) ([]StageResult, error) { detail := fmt.Sprintf("skipped: a member draft chunk of this edit unit was flagged (%s)", flagReason) var out []StageResult for _, ws := range staged { st := ws.st if err := r.Store.UpsertChunkStatus(store.ChunkStatus{ BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name, SnapshotID: snapID, Disposition: string(DispSkipped), FlagReason: string(flagReason), Detail: detail, }); err != nil { return nil, fmt.Errorf("pipeline: record skipped edit chunk_status: %w", err) } out = append(out, StageResult{ Stage: st.Name, Role: st.Role, Model: st.ResolvedModel, Disposition: DispSkipped, FlagReason: flagReason, Detail: detail, }) } return out, nil } // mergeUnitRetrievalState folds a edit unit's post-check + cheap-gate observability into the LEADER chunk's // retrieval_state row — the row the draft wave already wrote with the leader chunk's draft-injection counts + banknote // telemetry. A read-modify-write (single-writer safe; each unit owns a distinct leader chunk_idx) preserves // the draft fields and adds the unit-level post-check/style fields, so the read-models aggregate injection // per draft chunk and post-check/style per unit. A missing row (r.memory nil path) is a no-op. func (r *Runner) mergeUnitRetrievalState(chapter, firstChunkIdx int, misses membank.PostcheckResult, outputChecked bool, cheap checks.CheapGateResult, voice checks.VoiceResult, leaks []membank.SpoilerLeak) error { rs, err := r.Store.GetRetrievalState(r.Book.BookID, chapter, firstChunkIdx) if err != nil { return fmt.Errorf("pipeline: read leader retrieval_state ch%d/chunk%d: %w", chapter, firstChunkIdx, err) } if rs == nil { return nil // no draft row (no memory / no draft) → nothing to merge onto } rs.NStyleFlags = cheap.Total() rs.StyleDetail = "" if cheap.Total() > 0 { if b, err := json.Marshal(cheap); err == nil { rs.StyleDetail = string(b) } } setVoiceState(rs, voice, leaks) rs.NPostcheckMiss = 0 rs.PostcheckDetail = "" if outputChecked { rs.NPostcheckMiss = misses.ConfirmedCount() rs.NUnverifiedShown, rs.NUnverifiedFollowed = misses.Shown, misses.Followed if all := misses.All(); len(all) > 0 { if b, err := json.Marshal(all); err == nil { rs.PostcheckDetail = string(b) } } } return r.Store.UpsertRetrievalState(*rs) } // draftOnlyOutcome assembles a ChunkOutcome for a draft-only pipeline (no editor): the draft is the shipping // output, so the per-chunk draft result maps straight to the outcome (post-check/cheap already ran in // runDraftChunk as the final wave). Mirrors the export contract: a cosmetic strip exports its recovered // text; every other flag exports "". func (r *Runner) draftOnlyOutcome(ch chunk.Chunk, d stageSeqResult) ChunkOutcome { oc := ChunkOutcome{Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stages: d.stages, CostUSD: d.cost, Disposition: DispOK} if d.flagged { oc.Disposition = DispFlagged oc.FlagReason = d.flagReason oc.FinalText = r.checkers.ExportNormalize(d.recovered) } else { oc.FinalText = r.checkers.ExportNormalize(d.finalText) } oc.FinalText = chunk.ApplyHeading(ch.Heading, oc.FinalText) // pack-13 title policy (draft-only shipping path) return oc } // foldUnitOutcomes folds the edit wave's results into the book result. A nil outcome is a unit this run // never started — outside the volume scope, or past the point a spend ceiling stopped admissions. func foldUnitOutcomes(res *BookResult, unitOutcomes []*ChunkOutcome) { for _, oc := range unitOutcomes { if oc == nil { continue } res.Chunks = append(res.Chunks, *oc) res.TotalUSD += oc.CostUSD if oc.Disposition == DispFlagged { res.Flagged++ } } } // moneyStoppedResult assembles what a run stopped by a SPEND ceiling IN THE DRAFT WAVE actually has. // // For an edit pipeline that is nothing shippable — a draft is not the product — but the drafts were paid // for, so their cost is carried into the run's total. For a draft-only pipeline the drafts ARE the // product and the ones that finished ship exactly as they would have. // // Only chunks the wave actually ran are folded. The zero stageSeqResult of an unstarted chunk would // otherwise ship as an empty, unflagged, free translation — a unit the reader receives blank and the // ledger reports as delivered. func (r *Runner) moneyStoppedResult(ctx context.Context, chunks []chunk.Chunk, drafted []bool, draftResults []stageSeqResult, editWave bool) *BookResult { res := &BookResult{BookID: r.Book.BookID} for i, ch := range chunks { if !drafted[i] { continue } if editWave { res.TotalUSD += draftResults[i].cost continue } oc := r.draftOnlyOutcome(ch, draftResults[i]) res.Chunks = append(res.Chunks, oc) res.TotalUSD += oc.CostUSD if oc.Disposition == DispFlagged { res.Flagged++ } } r.noteMoneyStop(ctx, res) return res } // noteMoneyStop finishes a spend-ceiling stop — and states what such a stop must NOT claim. // // ⛔ THE VOLUME LEDGER IS WITHHELD, and it is withheld rather than corrected because it cannot be // corrected. VolumeStop's seven counters are a PLAN: planVolume classifies every unit before the waves // start, and reconcile only trues up outcomes that exist. A spend ceiling stops the run in the middle of // work the plan said would happen, so units that were admitted and never finished stay counted as // Delivered, Reworked, Carried or Free while LeftFresh and LeftRework are short by exactly the same // units — and a run that stopped short of the book would then announce «reached the end of the book … // held nothing back» and invite no further purchase. There is no bucket in those seven for «admitted, // not finished», so the honest move is to say nothing in that vocabulary and to say what happened in one // of its own: the `money` ledger on the stream's terminal line, counted from units the run RESOLVED. func (r *Runner) noteMoneyStop(ctx context.Context, res *BookResult) { res.Volume = nil r.Log.InfoContext(ctx, "the run stopped on a SPEND ceiling: every admitted call was finished and nothing new was started — the book is resumable and the stream says how much was missing", "book", r.Book.BookID, "units_shipped_this_run", len(res.Chunks), "run_usd", fmt.Sprintf("%.6f", res.TotalUSD)) }