textmachine/backend/internal/pipeline/volume.go

581 lines
32 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package pipeline
import (
"context"
"fmt"
"textmachine/backend/internal/chunk"
"textmachine/backend/internal/membank"
"textmachine/backend/internal/store"
)
// volume.go: the VOLUME ceiling — the run's second stop, beside the money one (D39.165 §1б, backlog row
// «потолок объёма»).
//
// WHY A SECOND CEILING AT ALL. The platform sells CHAPTERS and hands the engine a DOLLAR bound
// (--ceiling-usd), which is the only bound the engine had. Measured on real ledgers (D39.165 §1), a
// chapter costs $0.0115$0.0190 and the constant sold against it is $0.03 — so a purchase of ten chapters
// hands over $0.30, and $0.30 buys sixteen to twenty-six. The knob said chapters and meant money. This
// gives the engine a bound in the unit it actually ships, so the seller's promise and the engine's stop
// are the same quantity.
//
// WHAT IT IS MEASURED IN, and why not the other two candidates. The unit is the OUTPUT UNIT — editUnit,
// the granularity `outputUnits` returns and `buildManifest` publishes as the manifest's units_total, which
// is where the platform's per-chapter unit count comes from. So "N units" means the same thing on both
// sides of the seam, and the platform's chapters→units conversion is EXACT rather than an estimate: the
// manifest is $0, needs no keys and exists before the first paid call.
// - NOT chunk×stage, which is the BILLING unit (the one the re-payment estimate counts in). Cutting the
// wave in the unit one PAYS in, while selling the unit one SHIPS in, reproduces the very defect above
// one layer down: a unit costs len(Members)·nDraftStages + nEditStages positions, so "10" would buy a
// different amount of book on a different pipeline shape.
// - NOT the chunk, which is an internal artefact of the cut: the buyer never sees it and the manifest
// does not count it.
//
// ⚠ The unit count DOES depend on the pipeline's shape — an editor pipeline groups draft chunks into
// coarse edit units, a draft-only pipeline ships one unit per draft chunk (outputUnits). That is a
// property of the configuration, not a branch per language pair, and it is safe because the manifest the
// platform counted and the run it bounds come from the SAME deployment. It would stop being safe if a
// ceiling were carried across two deployments with different pipelines.
//
// HOW REPINS AND RETRIES COUNT — the question this design has to answer out loud.
// - A RETRY and an ESCALATION hop do NOT count. They live inside one unit (the attempt loop in
// stagerun.go writes ONE chunk_status row per chunk×stage however many attempts it took), and the
// volume ceiling bounds DELIVERY while the money ceiling bounds SPEND. The measured quality tail
// (retries+escalations swinging 1.5%→23% between runs) is a fact about money and already has its own
// bound; if it burned volume, a buyer of ten chapters would receive eight because two of them
// stuttered — paying in BOOK for the engine's own tail, which is the substitution this pack exists to
// remove.
// - A RE-PIN and a $0 RESUME do NOT count. A re-pass that re-pins five hundred units for $0 (rebill.go)
// would otherwise exhaust the ceiling having delivered nothing.
//
// So: THE CEILING COUNTS OUTPUT UNITS FOR WHICH THIS RUN WILL MAKE AT LEAST ONE PROVIDER CALL. Units that
// resolve for free are admitted regardless — they cost nothing, and holding them back would leave the
// book's snapshots stale for no saving.
//
// AND IT COUNTS THEM IN TWO KINDS, which is the difference between a report and a true report. A paying
// unit is either book that did not exist (DELIVERY) or book that existed and is being made again under a
// moved snapshot (REWORK). unitClass below is that distinction, and VolumeStop carries it all the way to
// the operator: a purchase spent entirely on rework delivers no chapter, and neither the counters nor the
// line the CLI prints may let that read as a delivery. The remainder splits the same way — units never
// delivered are the only ones anybody may be invited to buy, while delivered-but-not-re-made units are
// unrefreshed, not unbought.
//
// ⚠ ON THE PREDICATE'S RELATION TO projectRebill: the free/paying test here is runStage's, not the
// re-payment estimate's, and they are NOT the same test. projectRebill deliberately ignores the CONTENT
// axis (its own doc comment says so: a source edit moves content_hash but not the snapshot, and is
// invisible there). This one checks the rendered content hash on every row, so a source edit correctly
// makes a unit paying here while the estimate would call it free. The difference is in the safe
// direction — the ceiling over-counts cost rather than under-counting it — and it is why this predicate
// is written against the executor rather than borrowed from the projection.
//
// CHECKED BEFORE, NOT AFTER. The scope is computed ONCE, before either wave starts, and the waves simply
// do not begin an item outside it. That is structural rather than a guard: there is no moment at which a
// unit past the ceiling has started, so the ceiling cannot be overshot by one call the way a post-hoc
// token check always is (LiteLLM's known behaviour, named in D39.165 §1б). It is also why the admission
// is not a counter the wave workers decrement — the waves are parallel, and a worker refused a slot would
// have to abandon its item permanently while a slot freed by a $0 re-pin went to nobody.
//
// THE STOP IS A COMPLETION, NOT A HALT — ratified by the orchestrator with this pack and NOT re-decided
// here. A money ceiling stops the run in the MIDDLE of work it meant to do: that is exit 4, the platform
// records `paused`, and the remedy is more money. A volume ceiling means the run did exactly what was
// bought: exit 0, the frozen exit-code dictionary (cmd/tmctl/main.go) is not touched at all, and the
// platform records the run `ready` — which does NOT declare the book finished, because the book's progress
// travels as unit events, so the next purchase starts a new run normally. What DOES have to be said is
// WHICH ceiling stopped it (OpenHands' iteration limit is silent, and that is a standing complaint named
// in the same note): that lives in the run's result, its log line and the CLI's rendering, not in a new
// exit code and not in a new word of the event stream's outcome vocabulary.
// VolumeStop is a run's report that it stopped on the volume ceiling rather than on the end of the book.
// It is a RESULT, not an error: the run did what it was granted, so it travels on BookResult and never
// through the error path, where the exit-code mapper would have to invent a word for it.
//
// ⚠ IT COUNTS PAID UNITS IN TWO KINDS, and the split is a MODEL rather than a nicety. A paying unit is
// either book that did not exist before (DELIVERY) or book that existed and is being made again under a
// moved snapshot (REWORK), and the two are not the same product: the first is what "buy ten chapters"
// means, the second is what a re-pass means. The first version of this struct had one counter for both,
// and every consequence of that was a lie the operator could act on — a purchase that went entirely into
// re-translating the opening chapters reported ten units bought and exited 0 like a delivery, while the
// remaining count invited buying chapters the reader already owned. A struct that cannot tell the two
// apart cannot report either one honestly, so it carries both.
type VolumeStop struct {
// MaxUnits is the ceiling that was in force.
MaxUnits int
// Delivered is paying units that had never been completed before — NEW book, the thing a purchase of
// chapters is actually for.
Delivered int
// Reworked is paying units that WERE already completed and are being paid for a second time because
// their snapshot moved. Real work and a real charge, but not new book: a purchase made entirely of
// these advances the reader not at all.
Reworked int
// Flagged is paid units that resolved FLAGGED and shipped no text. They cost money and delivered
// nothing, and calling them delivered is the lie this counter exists to stop: a purchase of two units
// one of which flagged used to print "2 NEW unit(s) delivered" over a single readable unit.
//
// ⚠ It is filled AFTER the waves, from what actually happened, not from the plan. Everything else on
// this struct is a plan-time decision about admission — correct for bounding money, and unable on its
// own to know whether the work it authorised produced a chapter.
Flagged int
// Free is units that rode along at $0 — resumed or re-pinned. Reported because otherwise the operator
// sees a run that touched far more units than it was granted and cannot tell why.
Free int
// LeftFresh is undelivered units still in the book. THIS is what "there is more to buy" honestly
// means, and it is the only one of the two remainders a buyer should ever be invited to purchase.
LeftFresh int
// LeftRework is completed units still carrying a superseded snapshot. They are "unrefreshed", not
// "unbought", and presenting them as stock for sale is how re-payment becomes a product.
LeftRework int
}
// Paid is every unit this run was charged for, of either kind.
func (v VolumeStop) Paid() int { return v.Delivered + v.Reworked }
// reconcile corrects the plan-time counters against what the run actually produced.
//
// Admission is a decision made BEFORE the work, and it has to be — that is what bounds the money. But a
// decision to pay for a unit is not a fact about a chapter existing: a unit can be paid for and come back
// flagged with nothing shippable. Reporting the plan as though it were the outcome is how "2 NEW unit(s)
// delivered" ended up printed over one readable unit. So the counters are trued up here, from the
// outcomes, before anyone reads them.
func (s *volumeScope) reconcile(outcomes []ChunkOutcome) {
if s == nil {
return
}
for _, oc := range outcomes {
if oc.Disposition != DispFlagged || oc.FinalText != "" {
continue // it shipped something readable; the plan's word for it stands
}
switch {
case s.stop.Delivered > 0 && s.class[chunkKey{oc.Chapter, oc.ChunkIdx}] == unitFresh:
s.stop.Delivered--
case s.stop.Reworked > 0:
s.stop.Reworked--
case s.stop.Delivered > 0:
s.stop.Delivered--
default:
continue // it was never counted as paid (a free unit that flagged on resume); nothing to move
}
s.stop.Flagged++
}
}
// Left is every unit the ceiling held back, of either kind.
func (v VolumeStop) Left() int { return v.LeftFresh + v.LeftRework }
func (v VolumeStop) String() string {
// ⚠ EVERY COUNT HERE IS IN OUTPUT UNITS, AND THE WORDS MUST SAY SO. An earlier version of this line
// read "%d NEW chapter(s) delivered" over a counter that increments per editUnit — and a chapter is
// more than one unit whenever the cut closes a unit at the edit ceiling, and is many units in a
// draft-only pipeline. That is the unit↔chapter substitution this whole ceiling exists to remove,
// committed by the ceiling's own report. Chapters are the PLATFORM's word; it converts them through
// the manifest. The engine speaks units and nothing else.
s := fmt.Sprintf("stopped on the VOLUME ceiling (--max-units %d), not on money and not at the end of the book: %d paying output unit(s) — %d NEW unit(s) delivered, %d already-delivered unit(s) re-made under a moved snapshot",
v.MaxUnits, v.Paid()+v.Flagged, v.Delivered, v.Reworked)
if v.Flagged > 0 {
s += fmt.Sprintf(", and %d PAID FOR BUT FLAGGED — money spent, no readable text produced (buying more will not fix them; `tmctl redrive` re-attacks a flag)", v.Flagged)
}
if v.Free > 0 {
s += fmt.Sprintf("; %d rode along at $0 (resumed or re-pinned)", v.Free)
}
s += fmt.Sprintf(". Still in the book: %d unit(s) NEVER delivered", v.LeftFresh)
if v.LeftRework > 0 {
s += fmt.Sprintf(" and %d already delivered but not yet re-made (those are unrefreshed, NOT unbought)", v.LeftRework)
}
return s
}
// volumeScope is the run's admitted set of output units, decided before any work begins. A nil scope
// means no volume ceiling is in force and nothing anywhere changes behaviour.
type volumeScope struct {
admitted map[chunkKey]bool // unit leader key → admitted
leader map[chunkKey]bool // member chunk key → its unit is admitted (flattened for the draft wave)
stop VolumeStop
// class is what each unit was judged to be at PLAN time, kept because the edit wave has to re-judge
// the free ones (see rescopeEditWave).
class map[chunkKey]unitClass
// editSnapshot is the edit-wave snapshot the classification above was made AGAINST. The run can move
// it after planning — the bank-mining stop re-seeds mid-run — so this is what rescopeEditWave compares
// with to know whether the plan is still standing on the ground it was made on.
editSnapshot string
// editBlocked are units whose EDIT must not run even though they were admitted: they were judged free,
// the bank then moved under them, and no grant was left to pay for what they turned out to cost.
editBlocked map[chunkKey]bool
}
// allows reports whether a DRAFT chunk may be started: it may when the unit it belongs to was admitted.
func (s *volumeScope) allows(ch chunk.Chunk) bool {
if s == nil {
return true
}
return s.leader[chunkKey{ch.Chapter, ch.ChunkIdx}]
}
// allowsUnit reports whether an EDIT unit may be started.
func (s *volumeScope) allowsUnit(u editUnit) bool {
if s == nil {
return true
}
key := chunkKey{u.Chapter, u.FirstChunkIdx}
return s.admitted[key] && !s.editBlocked[key]
}
// admittedStatuses narrows stored rows to the units this run is actually going to work on, so a money
// projection over them describes THIS run rather than the whole book. A nil scope is the identity, which
// is what keeps every un-bounded run's figures byte-identical to what they were before the ceiling
// existed. An edit row lives at its unit's LEADER chunk, and a leader is one of the unit's members, so
// the member map covers both waves' rows.
func (s *volumeScope) admittedStatuses(in []store.ChunkStatus) []store.ChunkStatus {
if s == nil {
return in
}
out := in[:0:0]
for _, cs := range in {
if s.leader[chunkKey{cs.Chapter, cs.ChunkIdx}] {
out = append(out, cs)
}
}
return out
}
// bound reports whether the ceiling actually held work back. A run granted more than the book had left is
// an ordinary complete run and must not be reported as a volume stop — otherwise every generously-bounded
// run would end claiming it had been cut short.
func (s *volumeScope) bound() bool { return s != nil && s.stop.Left() > 0 }
// planVolume decides, before any wave starts, which output units this run may work on.
//
// It is $0 by construction: store reads, string rendering and hashing, no provider anywhere. It is also
// skipped entirely when no ceiling is set, so a run without --max-units pays nothing for this existing —
// not even the content-hash reproduction, which is the expensive half.
func (r *Runner) planVolume(ctx context.Context, chunks []chunk.Chunk, stickySel []membank.Selection) (*volumeScope, error) {
if r.MaxUnits <= 0 {
return nil, nil
}
// ⚠ REFUSED on a pipeline whose SHIPPING stage is a translator while editor stages also exist, because
// in that one shape the two granularities this ceiling straddles stop being the same object. outputUnits
// keys off finalStageWave and returns a singleton per DRAFT CHUNK there, while the edit wave still
// groups those chunks into buildEditUnits — so a ceiling granted in singletons would admit one member's
// chunk and then let the edit wave run the whole multi-member unit over drafts that were never made,
// paying full price for half-empty input. config validates stage ROLES but imposes no order, so the
// shape loads; refusing loudly is the only answer that neither miscounts a purchase nor silently
// changes what a unit means. (It is also the shape in which the platform's chapters→units conversion
// would stop being exact, which is the whole basis of measuring in units at all.)
if len(r.waveStagesIndexed(waveEdit)) > 0 && r.finalStageWave() == waveDraft {
return nil, fmt.Errorf("pipeline: --max-units cannot bound this pipeline: its last stage is a translator while editor stages exist, so the shipping unit is the draft CHUNK while the edit wave still works in grouped edit UNITS — a ceiling counted in one would admit work measured in the other. Put the shipping stage last (an editor/other role), or run this pipeline without a volume ceiling")
}
// ⚠ THE ONE COMPOSITION THIS CEILING MAKES WORSE, said out loud where an operator will see it.
//
// A bounded purchase drafts only part of the book, and the bank-mining stop consolidates over the
// chunks drafted SO FAR — so the auto-bank it writes grows with every purchase. The next purchase folds
// that larger auto-bank into the ENRICHED bank, which moves the edit-wave snapshot, and the edit jobs
// the PREVIOUS purchase created are then pinned to a superseded one: without --resnapshot the run stops
// loudly, and with it the units whose injected bytes the new terms changed are re-translated and re-paid.
//
// None of that is new machinery — it is the mine D39.165 §3 named and errata 28.08-и narrowed to "an
// edit made AFTER edit jobs exist". What IS new is the frequency: without a volume ceiling that state
// needs an interrupted run (which usually has no edit jobs yet), while WITH one it is the ordinary
// shape of selling a book a few chapters at a time. Warned rather than refused, because the run is
// still correct — it re-pays through the consent gate like any other re-payment — and because refusing
// would take the ceiling away from exactly the books that most need selling in parts.
if r.pack != nil && r.Pipeline.Mining.ContrastPath != "" {
r.Log.WarnContext(ctx, "a VOLUME ceiling on a book that MINES its bank: each purchase drafts more, so the bank-mining stop writes a larger auto-bank, and the next purchase moves the edit-wave snapshot the previous purchase's edit jobs are pinned to. Expect that run to need --resnapshot and to re-pay the units the new terms actually touch (the re-payment consent gate still bounds it)",
"book", r.Book.BookID, "max_units", r.MaxUnits)
}
units := r.outputUnits(chunks)
statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID)
if err != nil {
return nil, fmt.Errorf("pipeline: read chunk_status for the volume ceiling: %w", err)
}
class, curEditAtPlan, err := r.classifyUnits(units, chunks, statuses, stickySel)
if err != nil {
return nil, err
}
s := &volumeScope{
admitted: make(map[chunkKey]bool, len(units)),
leader: make(map[chunkKey]bool, len(chunks)),
stop: VolumeStop{MaxUnits: r.MaxUnits},
class: class,
editSnapshot: curEditAtPlan,
editBlocked: map[chunkKey]bool{},
}
admit := func(u editUnit) {
key := chunkKey{u.Chapter, u.FirstChunkIdx}
s.admitted[key] = true
for _, m := range u.Members {
s.leader[chunkKey{m.Chapter, m.ChunkIdx}] = true
}
}
// ⚠ A MISSING KEY READS AS unitFresh, and that is deliberate rather than incidental: classifyUnits
// returns an EMPTY map for a book with no stored rows, so the zero value has to be the class such a
// book's units actually are. unitFresh is iota 0 for exactly this reason — do not reorder the enum.
classOf := func(u editUnit) unitClass { return class[chunkKey{u.Chapter, u.FirstChunkIdx}] }
// Free units first and unconditionally: they spend nothing, so holding them back would only leave
// their rows pinned to a superseded snapshot for no saving at all.
for _, u := range units {
if classOf(u) == unitFree {
s.stop.Free++
admit(u)
}
}
// ⚠ DELIVERY BEFORE REWORK, and this ORDER is the answer to "what did the buyer actually get".
//
// Both kinds are paying work, so a single pass in book order admits whichever comes first — and after
// any bank edit the already-delivered units ARE the early ones. Measured consequence, on a fixture
// where one chapter is one unit: a purchase of two units on a five-unit book with two delivered
// re-made units 1 and 2 and never started unit 3, reporting Delivered=0, Reworked=2, LeftFresh=3. The
// buyer paid for two units, received none, and three units that had never been translated at all sat
// untouched. A ceiling sold as "the seller's promise and the engine's stop are the same quantity"
// cannot behave that way.
//
// So the grant goes to NEW book while any is left, and only then to re-making. The reason it is safe to
// serve units out of book order is that nothing in a unit's rendered bytes depends on which other units
// ran: the sticky-window selection is precomputed over the WHOLE book before either wave
// (precomputeSticky), so running unit 3 before unit 1 renders exactly what running them in order would.
//
// ⚠ THE TRADE IS NAMED, not hidden: a purchase whose PURPOSE was a re-pass will now spend its grant on
// undelivered units if the book still has any. That is the right default — an undelivered unit is
// worth more to a reader than a re-made one, and the re-payment consent gate bounds rework by money in
// its own right — but it IS a default, and a caller who needs "re-make only" needs a flag that says so.
// Reported to the orchestrator as a chosen default rather than assumed.
for _, pass := range []unitClass{unitFresh, unitRework} {
for _, u := range units {
if classOf(u) != pass {
continue
}
if s.stop.Paid() >= r.MaxUnits {
if pass == unitFresh {
s.stop.LeftFresh++
} else {
s.stop.LeftRework++
}
continue
}
if pass == unitFresh {
s.stop.Delivered++
} else {
s.stop.Reworked++
}
admit(u)
}
}
return s, nil
}
// rescopeEditWave re-judges the FREE units against the snapshot the edit wave is actually about to run
// under, and makes them pay for a grant slot if they turn out to cost money.
//
// ⛔ WITHOUT THIS THE CEILING DOES NOT HOLD, and the report lies while it fails. planVolume classifies
// before the draft wave, but the bank-mining stop sits between the two waves and RE-SEEDS the bank
// mid-run (mining.go, the auto-continue branch), after which waverun recomputes the edit-wave snapshot.
// Every unit admitted as FREE — and free units are admitted OUTSIDE the grant, because free work costs
// nothing — was judged against a snapshot the run itself then replaced. Any of them whose injected bytes
// the new bank changes gets a fresh PAID editor call that no grant ever authorised, and the stop line
// calls it "rode along at $0". Found by the acceptance hunter, who measured four provider calls under a
// grant of one and $0.007280 of spend reported as free.
//
// The fix is not a guard but a re-plan: the free/paying question is asked again, against the snapshot
// that is now real, at the last moment before the edit wave begins — so "check before the unit" still
// holds. A unit that has become paying takes a slot if one is left; if none is, its edit does not run at
// all and it is reported as an already-delivered unit still awaiting its re-make, which is what it is.
//
// It is a no-op on the ordinary path: with no mid-run re-seed the snapshot is unchanged and every free
// unit stays free, so a book that does not mine pays nothing for this existing.
func (r *Runner) rescopeEditWave(ctx context.Context, s *volumeScope, units []editUnit, chunks []chunk.Chunk, stickySel []membank.Selection, editSnapshot string) error {
if s == nil || editSnapshot == s.editSnapshot {
return nil // the ground the plan was made on is still the ground the wave runs on
}
statuses, err := r.Store.ChunkStatusesForBook(r.Book.BookID)
if err != nil {
return fmt.Errorf("pipeline: re-read chunk_status for the volume ceiling's edit-wave re-plan: %w", err)
}
draftStages, editStages := r.waveStagesIndexed(waveDraft), r.waveStagesIndexed(waveEdit)
draftNames, editNames := stageNameSet(draftStages), stageNameSet(editStages)
curDraft, _, err := r.snapshotIDForWave(waveDraft)
if err != nil {
return fmt.Errorf("pipeline: render the draft-wave snapshot for the edit-wave re-plan: %w", err)
}
// The memo was dropped by the re-seed that moved the snapshot (materializeBanks clears it), so this
// renders against the bank the edit wave will actually use.
hashes := r.cachedRenderedContentHashes(chunks, stickySel)
byChunk := map[chunkKey][]store.ChunkStatus{}
for _, cs := range statuses {
byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}] = append(byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}], cs)
}
moved, paid, blocked := 0, 0, 0
for _, u := range units {
key := chunkKey{u.Chapter, u.FirstChunkIdx}
if !s.admitted[key] || s.class[key] != unitFree {
continue // only the units admitted WITHOUT a slot can be wrong about being free
}
rows := unitRows(u, byChunk)
if r.rowsResumeFree(rows, draftNames, editNames, curDraft, editSnapshot, hashes) {
continue // still free under the snapshot that is now real
}
moved++
s.stop.Free--
if s.stop.Paid() < s.stop.MaxUnits {
// It costs money now, and there is grant left to pay for it: it becomes what it is — an
// already-delivered unit being re-made.
s.stop.Reworked++
s.class[key] = unitRework
paid++
continue
}
// No grant left. Its edit does not run; it stays delivered and stale, which is LeftRework.
s.editBlocked[key] = true
s.stop.LeftRework++
s.class[key] = unitRework
blocked++
}
if moved > 0 {
r.Log.WarnContext(ctx, "the bank moved between planning and the edit wave, so units judged FREE are no longer free: they have been re-judged against the snapshot the edit wave actually uses",
"book", r.Book.BookID, "no_longer_free", moved, "took_a_grant_slot", paid, "held_back_for_want_of_grant", blocked,
"planned_against", s.editSnapshot[:12], "running_under", editSnapshot[:12])
}
s.editSnapshot = editSnapshot
return nil
}
// unitClass is what this run would DO to an output unit, which is the distinction the ceiling both
// admits and reports on.
type unitClass int
const (
// unitFresh — never completed. Paying for it DELIVERS a chapter that did not exist.
unitFresh unitClass = iota
// unitFree — fully recorded and every row resumes or re-pins at $0. Costs nothing, so it rides along
// regardless of the ceiling.
unitFree
// unitRework — fully recorded, but at least one row will be paid for again under a moved snapshot.
// Real work and a real charge, and NOT new book: it replaces a chapter the reader already has.
unitRework
)
// classifyUnits sorts every output unit into the three things this run can do to it.
//
// The free/paying predicate is deliberately the one the run itself applies, read off runStage's resume
// fast-path: a stored row is served for free when its rendered content hash is unchanged AND its snapshot
// either matches the current one or is a re-pinnable bank-only move. A unit is free when every one of its
// rows is, and when every position it will execute already HAS a row — a half-done unit resumes what it
// has and pays for the rest.
//
// The fresh/rework split falls straight out of the same completeness test, which is why it costs nothing
// to make: a unit that was never fully recorded has never been delivered, so paying for it is delivery; a
// unit that WAS fully recorded has been delivered once already, so paying for it again is rework.
//
// Every uncertainty resolves to "this unit will cost", never to "this unit is free". That direction is
// forced: mistaking a paying unit for a free one lets the run spend past the ceiling, which is the defect
// the ceiling exists to prevent, while the opposite merely makes a run stop one unit early.
func (r *Runner) classifyUnits(units []editUnit, chunks []chunk.Chunk, statuses []store.ChunkStatus, stickySel []membank.Selection) (map[chunkKey]unitClass, string, error) {
class := map[chunkKey]unitClass{}
if len(statuses) == 0 {
return class, "", nil // a book nothing has run is entirely fresh; skip the rendering entirely
}
draftStages, editStages := r.waveStagesIndexed(waveDraft), r.waveStagesIndexed(waveEdit)
draftNames, editNames := stageNameSet(draftStages), stageNameSet(editStages)
curDraft, _, err := r.snapshotIDForWave(waveDraft)
if err != nil {
return nil, "", fmt.Errorf("pipeline: render the draft-wave snapshot for the volume ceiling: %w", err)
}
curEdit := curDraft
if len(editStages) > 0 {
if curEdit, _, err = r.snapshotIDForWave(waveEdit); err != nil {
return nil, "", fmt.Errorf("pipeline: render the edit-wave snapshot for the volume ceiling: %w", err)
}
}
// The same reproduction the re-payment estimate uses (repin.go): what the run WOULD render for every
// live position. A position whose inputs cannot be reproduced is simply absent, and an absent hash is
// read below as "cannot conclude", i.e. the paying answer.
hashes := r.cachedRenderedContentHashes(chunks, stickySel)
byChunk := map[chunkKey][]store.ChunkStatus{}
for _, cs := range statuses {
byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}] = append(byChunk[chunkKey{cs.Chapter, cs.ChunkIdx}], cs)
}
for _, u := range units {
rows := unitRows(u, byChunk)
key := chunkKey{u.Chapter, u.FirstChunkIdx}
if !unitFullyRecorded(u, rows, draftStages, editStages) {
class[key] = unitFresh // a position with no row is a position this run will call for
continue
}
if r.rowsResumeFree(rows, draftNames, editNames, curDraft, curEdit, hashes) {
class[key] = unitFree
continue
}
class[key] = unitRework
}
return class, curEdit, nil
}
// unitFullyRecorded reports whether EVERY position this run would execute for the unit already has a
// stored row — the completeness half of "this unit costs nothing".
//
// ⚠ IT DELIBERATELY DOES NOT USE resolveChunkState, and the difference is money. That resolver answers
// "what does a reader call this unit", and it returns ChunkFlagged the moment ANY row is flagged, before
// it ever tests whether the expected positions are all present. That is right for a progress projection
// and wrong here: a unit whose draft flagged for one member and whose EDIT row does not exist yet — the
// state every book is left in by a bank-signing stop, a Ctrl-C, or any abort in the edit wave — would be
// read as terminal, judged free, admitted without consuming a grant, and then charged for a full editor
// call. The run would pay for more units than were bought and, with Deferred still 0, would not even
// report a volume stop. Found by this pack's own adversarial pass, reproduced before it was fixed.
//
// Positions are counted, not dispositions: a completed unit always leaves a row for every position it
// has — ok, flagged, or the `skipped` rows recordSkippedStages writes downstream of a flag, including the
// all-members-flagged case where the editor never runs. So full coverage IS terminality, and it is the
// property that actually predicts "no provider call".
func unitFullyRecorded(u editUnit, rows []store.ChunkStatus, draftStages, editStages []wavedStage) bool {
type pos struct {
chapter, chunkIdx int
stage string
}
have := make(map[pos]bool, len(rows))
for _, cs := range rows {
have[pos{cs.Chapter, cs.ChunkIdx, cs.Stage}] = true
}
for _, m := range u.Members {
for _, ws := range draftStages {
if !have[pos{m.Chapter, m.ChunkIdx, ws.st.Name}] {
return false
}
}
}
// The unit's edit rows all live at its LEADER chunk — that is where the read-models look for them.
for _, ws := range editStages {
if !have[pos{u.Chapter, u.FirstChunkIdx, ws.st.Name}] {
return false
}
}
return true
}
// rowsResumeFree is the per-row half of the predicate above.
func (r *Runner) rowsResumeFree(rows []store.ChunkStatus, draftNames, editNames map[string]bool,
curDraft, curEdit string, hashes map[chunkKey]map[string]string) bool {
for _, cs := range rows {
if cs.Disposition == string(DispSkipped) {
continue // a skipped stage is re-derived from the flag above it; it never reaches a provider
}
var cur string
var w wave
switch {
case draftNames[cs.Stage]:
cur, w = curDraft, waveDraft
case editNames[cs.Stage]:
cur, w = curEdit, waveEdit
default:
continue // a stage this pipeline no longer runs cannot cost anything (rebill.go's rule)
}
h, ok := hashes[chunkKey{cs.Chapter, cs.ChunkIdx}][cs.Stage]
if !ok || h != cs.ContentHash {
return false // the wire bytes moved, or could not be reproduced — either way, a fresh call
}
if cs.SnapshotID != cur && !r.repinnable(cs.SnapshotID, cur, w) {
return false
}
}
return true
}