textmachine/platform/internal/books/parse.go

package books

import (
	"context"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"os/exec"
	"time"

	"textmachine/platform/internal/ingest"
	"textmachine/platform/internal/jobs"
	"textmachine/platform/internal/pgstore"
)

// Graces and budgets of the intake walk. Constants rather than settings: they are properties of the
// walk itself, and every one of them is a number an operator would have to reason about the
// reconciler to choose.
const (
	// claimGrace is how long a parse may be somebody's business before another pass may take it. It
	// covers the ordinary case — the queue job is claimed in milliseconds — and the failure it exists
	// for: the process holding the claim was restarted mid-parse.
	//
	// ⚠ It MUST outlive the queue's own job timeout, and it is written as that constant plus a margin
	// so the two cannot drift apart. Shorter, the sweep steals the claim from a parse that is still
	// legitimately running: thief and holder meet on one project directory and the loser dies on the
	// engine's exclusive lock. That is no longer the data-loss it was — the lock has its own exit code
	// now (`project_locked`, ingest.ExitProjectLocked) and reads as the host's state rather than as a
	// verdict about the book — but it still spends an attempt of the budget on nothing.
	claimGrace = jobs.JobTimeout + 5*time.Minute
	// UploadGrace is how long a book may stay `uploading`. A request that is still arriving holds the
	// row, so anything older than this is an upload whose request is gone — and that reasoning holds
	// only while the route's own read deadline is SHORTER. Exported so the boot can refuse a
	// configuration where it is not: an operator who raises TM_PLATFORM_UPLOAD_DEADLINE past this
	// would have the sweep delete a book's row and directory out from under a request still writing
	// into it.
	UploadGrace = time.Hour
	// parseAttempts is how many times a book may be handed to the engine before intake gives up. It
	// bounds how long a broken host re-runs the engine over every book uploaded to it, and how many
	// chances a genuinely unreadable file gets before its source is removed.
	//
	// ⚠ A refusal of the SOURCE still goes through the whole budget even though the engine's verdict
	// is now unambiguous (intakeReason). Keeping it is a deliberate choice and not an oversight left
	// over from when it had to be one: the budget costs a genuinely empty book five $0 calls spread
	// over the claim grace, and costs a mistaken verdict nothing at all, whereas removing it buys
	// only latency — on the single path where being wrong deletes a user's file. If the wait is worth
	// more than the insurance, that is the owner's trade to make and it is one constant.
	parseAttempts = 5
)

// Reasons a book is rejected. The platform's OWN closed vocabulary: the engine's text reads like
// pipeline internals and never crosses this seam (contract §Problem, PT-33), and these are stored
// for an operator rather than projected — contract v0 gives a rejected book no reason field.
const (
	// ReasonSourceUnreadable — the engine read the file, cut it, and found no book in it. The ONE
	// reason that ends with the user's source being DELETED, and therefore the one the engine has to
	// say unambiguously: it is exit code 11 and nothing else (intakeReason). Terminal, but not on the
	// first answer — see parseAttempts for why the budget is kept.
	ReasonSourceUnreadable = ingest.RejectSourceUnreadable
	// ReasonNotConfigured — the book has no project configuration, so there is nothing to parse
	// against. See ErrNotProvisioned: today that is a deployment's state, not a user's mistake.
	ReasonNotConfigured = ingest.RejectNotConfigured
	// ReasonParserUnavailable — the engine could not be RUN, repeatedly, until the attempt budget was
	// spent.
	ReasonParserUnavailable = ingest.RejectParserUnavailable
	// ReasonStorageUnavailable — the storage root itself is not there. Never terminal, and never
	// stored on a book: it says something about the host, and the book it happened to be read for is
	// no more at fault than any other. See ErrStorageGone.
	ReasonStorageUnavailable = ingest.RejectStorageUnavailable
	// ReasonSchemaMismatch — the book's project database is not the schema this engine build speaks
	// (`tmctl migrate`, unified backlog row 174). Like the two above it WAITS: nothing is wrong with
	// the book, the repair is an operator's `tmctl migrate` or a newer binary, and a host mid-upgrade
	// would otherwise reject every book it holds. It is also the state the deploy note's own step
	// exists to prevent (`tmplatformctl books --migratable`).
	ReasonSchemaMismatch = ingest.RejectSchemaMismatch
)

// ErrNotProvisioned is a book with no usable engine configuration and no way for this platform to
// make one.
//
// The question it used to name — who writes the first `book.yaml`, given that D39.110 §2b says the
// platform does not own that file — was answered by D39.130 as form Б: the platform RENDERS one from
// a template the operator deploys, once, and never reads or edits it again (render.go). So this
// error is now the narrow remainder: a deployment with no template configured, or one whose template
// cannot be read or parsed.
//
// Both are the deployment's state and neither is the book's fault, which is why `not_configured`
// never ends an intake and never spends its budget — it applies to EVERY book on the host at once,
// and a human fixing one file clears all of them.
var ErrNotProvisioned = errors.New("books: the book has no engine configuration")

// ErrDirectoryGone is a book whose whole project directory is missing.
//
// Told apart from ErrNotProvisioned deliberately, and the difference is terminal-vs-not: a missing
// CONFIGURATION is a deployment question somebody can still answer, while a missing DIRECTORY means
// the source this platform received is not there any more and no amount of waiting brings it back.
// It is also the crash window of the rejection path — the directory is removed before the row is
// written — and reading it as "not configured" left such a book in `parsing` for good.
var ErrDirectoryGone = errors.New("books: the book's directory is gone")

// ErrStorageGone is the ROOT of the book storage missing — an unmounted volume, or a deployment
// pointed at a path that is not there yet.
//
// Told apart from ErrDirectoryGone because the two look identical from one book (both are ENOENT on
// the same Stat) and mean opposite things. A book's own directory being gone is that book's own
// terminal end; the root being gone is the host's, and reading it as the first would have ONE sweep
// reject EVERY book in intake with "the source cannot be read" — a reason that blames the user's
// file, is terminal by design, and has no way back (there is no re-parse and no un-reject).
var ErrStorageGone = errors.New("books: the book storage root is gone")

// Parse turns a received file into a chapter tree, or into a reason it is not one.
//
// It is the body of the queue's worker AND of the backstop sweep, and it is safe to call twice: the
// claim is a compare-and-set, so the loser does nothing. Losing that race is the ordinary case, not
// a failure — two `tmctl manifest` processes on one project directory would be two writers of a file
// the engine holds exclusively.
func (s *Service) Parse(ctx context.Context, bookID string) error {
	now := s.now()
	claim, err := s.Store.ClaimParse(ctx, bookID, now, now.Add(-claimGrace))
	if errors.Is(err, pgstore.ErrParseClaimed) {
		return nil
	}
	if err != nil {
		return err
	}
	return s.parseClaimed(ctx, claim, false)
}

// ErrNoBookInSource is a file the engine read, cut, and found no book in — the one verdict that is
// about the user's text rather than about this deployment (backlog row 285).
var ErrNoBookInSource = fmt.Errorf("%w: there is no book in this file", ErrBadIntake)

// ErrStructureNotDeliverable is a source the engine read and cut into fewer than two chapters.
//
// A refusal of OUR reach, not of the file: the reader's book is built one document per engine chapter
// (backend/internal/bookfile/epub.go), so a book nobody cut is delivered as a single canvas of any
// length. TEMPORARY — it goes when the delivery structure lands (backlog rows 283/325).
//
// The predicate is the engine's CUT and never the file extension: `.epub` is cut by nav or NCX
// (backend/internal/chunk/epubtoc.go), a txt whose language has structure data is cut by its
// headings, and anything else answers `structure: none`. Nothing here knows a format or a pair.
var ErrStructureNotDeliverable = fmt.Errorf("%w: this file has no chapter structure we can deliver yet", ErrBadIntake)

// errNotConclusive ends the intake's own pass without a verdict about the file: the queue finishes
// the book exactly as it did before this pass existed.
//
// Every class but the engine's two answers about the SOURCE leaves through here, and none of them
// touches `defer_` or `reject`: this pass must spend no attempt of a budget that bounds how often a
// broken HOST is asked, and it must leave no `rejected` row a 201 would then carry out. The caller
// gives the claim back on it (books.cutNow).
var errNotConclusive = errors.New("books: the intake's own cut reached no verdict about the file")

// notConclusive names the class in the message an operator reads, and keeps the sentinel matchable.
func notConclusive(reason string) error {
	return fmt.Errorf("%w (%s)", errNotConclusive, reason)
}

// parseClaimed is the body of a parse, shared by the queue worker and the intake.
//
// `atIntake` selects the only difference: what to do with a verdict about the SOURCE. A sweep must
// spend the attempt budget, because being wrong there deletes a file whose owner has gone home
// (parseAttempts). At intake the uploader still holds the file, so being wrong costs a retry — which
// is why the budget's reasoning does not reach here.
func (s *Service) parseClaimed(ctx context.Context, claim pgstore.ParseClaim, atIntake bool) error {
	m, err := s.manifest(ctx, claim)
	if err == nil {
		// THE FLOOR, and it is FIRST for a reason that is the whole of it: below this line a document
		// that could not be read correctly is indistinguishable from a book with nothing in it, and
		// that reading DELETES the user's upload after the attempt budget.
		//
		// It asks two things at once because both have the same answer (ingest.Readable): is this the
		// manifest shape this build reads, and does the document describe its own contents. The first
		// is the mine register row PD-213 names — a renamed key decodes to zeroes and the zeroes read
		// as an empty book. The second is PD-367: the MATERIALISER has always refused a document whose
		// counts and contents disagree, and the intake accepted the same document, founded a book on
		// its counts and let a run be started and PAID FOR over an empty tree. One document must not
		// get two answers from two ends of the same intake.
		//
		// `parser_unavailable` and never `source_unreadable`: a count read from the wrong key is not
		// evidence about the user's text, and the class that keeps the file is the only honest one for
		// «this build cannot read what the engine sent». It still spends the attempt budget, so a
		// deployment that is genuinely broken stops rather than retries forever.
		if rerr := m.Readable(); rerr != nil {
			s.log().ErrorContext(ctx, "the manifest does not describe itself, so it is not being read correctly",
				"chapters", m.ChaptersTotal, "units", m.UnitsTotal, "manifest_version", m.Version,
				"reason", ReasonParserUnavailable, "err", rerr)
			if atIntake {
				return notConclusive(ReasonParserUnavailable)
			}
			return s.defer_(ctx, claim, ReasonParserUnavailable)
		}
		if m.ChaptersTotal < 2 && atIntake {
			// Which of the two it is matters: the remedies are opposite — a different file, versus
			// waiting for delivery to learn this shape.
			if m.ChaptersTotal < 1 {
				return ErrNoBookInSource
			}
			s.log().InfoContext(ctx, "an upload was refused because nothing cut it into chapters",
				"structure", m.Structure, "units", m.UnitsTotal)
			return ErrStructureNotDeliverable
		}
		if m.ChaptersTotal < 1 {
			// The engine SUCCEEDED and reported a book with no chapters in it. Same verdict as exit 11
			// and the same budget: a manifest is not a place a deployment fault can hide.
			//
			// Reachable only for a document that PASSED the floor above — the version this build reads,
			// and every count agreeing with its own contents — which is what makes «there is no book in
			// these bytes» a statement about the user's text rather than about our reader.
			//
			// ⚠ The second conjunct this branch used to carry (`&& emptyBook(m)`, i.e. `UnitsTotal < 1`)
			// is GONE rather than kept for safety, because past the floor it can no longer be false and
			// a condition nothing can falsify is a condition no test can defend. Whole() forces
			// `ChaptersTotal == len(Chapters)` and `UnitsTotal == sum(len(c.Units))`, so zero chapters
			// implies zero units by arithmetic. What the conjunct used to guard — a document counting
			// units while counting no chapters, the shape of a key that moved — is refused ABOVE now,
			// non-destructively, which is the same verdict it used to produce and one branch earlier.
			// Removed on the finding of this pack's own adversarial pass, which showed it surviving
			// deletion against the whole battery.
			return s.defer_(ctx, claim, ReasonSourceUnreadable)
		}
		// Counts and versions, no book id: the same rule as everywhere else on this side of the log
		// (ENGINEERING_STANDARDS §Наблюдаемость). What an operator needs per book — including why a
		// book was rejected — is a column, not a log line.
		s.log().InfoContext(ctx, "book parsed", "chapters", m.ChaptersTotal,
			"manifest_version", m.Version, "chunker", m.ChunkerVersion)
		c, cancel := s.writeCtx(ctx)
		defer cancel()
		owed, err := s.Store.FinishParse(c, claim.BookID, claim.At, pgstore.ParsedBook{
			Chapters:       m.ChaptersTotal,
			SourceSHA256:   m.SourceSHA256Bytes(),
			ChunkerVersion: m.ChunkerVersion,
		})
		if err != nil {
			return err
		}
		// The tree the reader screen shows is materialized here, at the end of the parse. A failure is
		// logged and not returned — the parse succeeded, and re-running it to get the tree would spend
		// an attempt of a budget that exists for something else. The debt FinishParse recorded is what
		// brings the materializer back to it.
		//
		// NOT on the intake's own pass: materializing runs the engine twice more (readmodel.MaterializeBudget
		// is five minutes), and the uploader is holding the request open. The debt is already recorded,
		// so the materializer's sweep takes it — which is what makes the upload's tail bounded by
		// UploadSettle.
		if s.Reader != nil && !atIntake {
			// Detached from the job's deadline, which has already paid for the ingest and the cut:
			// materializing runs the engine twice more, and sharing what was left made a large book's
			// tree land empty. It bounds itself from there.
			c := context.WithoutCancel(ctx)
			b := pgstore.OwedBook{ID: claim.BookID, Workdir: claim.Workdir, OwedAt: owed}
			// CLAIMED first: the debt this parse just recorded is visible to the materializer's own
			// sweep the moment it commits, and without the claim every book slower to materialize than
			// one sweep interval was read by both at once.
			switch b, held, err := s.Reader.Claim(c, b); {
			case err != nil:
				s.log().ErrorContext(ctx, "the reading surface could not be claimed; the sweep will take it", "err", err)
			case !held:
				s.log().InfoContext(ctx, "the reading surface of this parse is already being materialized")
			default:
				if err := s.Reader.RefreshCut(c, b, m); err != nil {
					s.log().ErrorContext(ctx, "the reading surface of a parsed book could not be materialized", "err", err)
				}
			}
		}
		return nil
	}
	if errors.Is(err, ErrStorageGone) {
		// The host cannot see its own storage. Nothing about this book is known yet, so it waits with
		// the budget untouched — exactly like a book waiting for a configuration.
		s.log().ErrorContext(ctx, "the book storage root is not there: intake waits rather than rejects",
			"reason", ReasonStorageUnavailable)
		if atIntake {
			return notConclusive(ReasonStorageUnavailable)
		}
		return s.defer_(ctx, claim, ReasonStorageUnavailable)
	}
	if errors.Is(err, ErrDirectoryGone) {
		// Nothing to wait for and nothing to retry: what this platform received is not on disk. It is
		// also how the rejection path heals from a crash between removing the directory and writing
		// the row — the next pass finds no directory and finishes the job.
		s.log().ErrorContext(ctx, "book rejected: its directory is gone", "reason", ReasonSourceUnreadable)
		if atIntake {
			// Not answered as «no book in this file» either: the directory this platform wrote is gone
			// while the request that wrote it is still open, which is the host's state and not the
			// user's file.
			return notConclusive(ReasonSourceUnreadable)
		}
		return s.reject(ctx, claim, ReasonSourceUnreadable)
	}
	reason := ReasonNotConfigured
	if !errors.Is(err, ErrNotProvisioned) {
		reason = intakeReason(err)
	}
	if atIntake {
		// Exit 11 and nothing else: the engine's refusal contract makes that class unambiguous
		// (intakeReason).
		if reason == ReasonSourceUnreadable {
			return ErrNoBookInSource
		}
		return notConclusive(reason)
	}
	// ⚠ err can carry the book's path (the engine is asked about a directory), which is the open class
	// of PD-139, so it travels only at WARN/ERROR and never at INFO.
	return s.defer_(ctx, claim, reason)
}

// defer_ spends one attempt of the budget and, when the budget is gone, ends the intake.
//
// ONE path for every way a parse can fail, and the reason has outlived the defect that produced it.
// It was written when the engine mapped ALL of its failures onto exit 1, so "this source cannot be
// cut" was indistinguishable from "the disk was full" or "an operator's own tmctl held the lock",
// and rejecting on the first answer turned any of those into irreversible data loss. The engine now
// says which is which (intakeReason), and the single path stays because the BUDGET is still what
// bounds a broken host — what changed is that only one class ever reaches the destructive end of it.
//
// ⚠ `not_configured` and `storage_unavailable` NEVER become terminal, and never spend an attempt.
// Both say something about the deployment rather than about the book: a configuration that is
// missing or will not load (the platform renders it from the operator's template — see render.go —
// and a template that is absent or wrong is the operator's to fix), or a storage root that is not
// mounted. Rejecting either would destroy uploads over a gap the user cannot see, and it would do it
// to EVERY book on the host at once. They stay `parsing`, visible in the intake metric, and a human
// fixing the deployment is all it takes.
func (s *Service) defer_(ctx context.Context, claim pgstore.ParseClaim, reason string) error {
	if !waitsForTheDeployment(reason) && claim.Attempts >= parseAttempts {
		s.log().ErrorContext(ctx, "book rejected: intake has spent its attempts on it",
			"attempts", claim.Attempts, "reason", reason)
		return s.reject(ctx, claim, reason)
	}
	// The claim is NOT given back: it is what spaces the retries. Released, the sweep would re-offer
	// the book on its very next tick (the staleness predicate falls back to `added_at`, which is
	// already old), and the attempts would burn in as many ticks — 75 seconds at the default sweep
	// interval — rather than over the time this budget is for.
	//
	// The ATTEMPT, though, is given back when the engine was never asked. `ClaimParse` counts every
	// claim, and a book waiting for its configuration claims once per grace forever — so without this
	// the budget was spent by WAITING, and the first real answer from the engine afterwards was
	// terminal on arrival. A typo in a hand-written `book.yaml` would then delete the user's upload
	// on the first attempt at reading it. Waiting must not bring deletion closer.
	if waitsForTheDeployment(reason) {
		c, cancel := s.writeCtx(ctx)
		defer cancel()
		if err := s.Store.RefundParseAttempt(c, claim.BookID, claim.At); err != nil {
			s.log().ErrorContext(ctx, "the parse attempt could not be refunded", "err", err)
		}
	}
	s.log().WarnContext(ctx, "parse deferred", "attempts", claim.Attempts, "reason", reason)
	return nil
}

// waitsForTheDeployment reports the reasons that never end an intake and never spend its budget: the
// engine was not asked at all, or it was asked and refused over something that belongs to the
// DEPLOYMENT: a configuration nobody has written yet, a storage root that is not mounted, a project
// database an engine upgrade has not migrated. Every one of them applies to every book on the host
// at once and every one is answerable by a human, so ending an intake over one would destroy
// uploads over a gap their owners cannot see — in bulk.
func waitsForTheDeployment(reason string) bool {
	switch reason {
	case ReasonNotConfigured, ReasonStorageUnavailable, ReasonSchemaMismatch:
		return true
	}
	return false
}

// writeCtx is the context a TERMINAL write uses: detached from the caller's deadline and bounded on
// its own.
//
// The engine call this follows can legitimately consume the whole budget of the pass — and then the
// write that records what happened would run on an already-expired context and be lost, leaving the
// book to be retried and the attempt to be spent again, forever. What must survive is the record.
func (s *Service) writeCtx(ctx context.Context) (context.Context, context.CancelFunc) {
	budget := writeBudget
	if s.writeBudget > 0 {
		budget = s.writeBudget // tests shorten it; nothing outside this package can set it
	}
	return context.WithTimeout(context.WithoutCancel(ctx), budget)
}

// writeBudget is what a terminal write gets. Short: it is one statement against a database this
// process is already connected to.
const writeBudget = 30 * time.Second

// UploadSettle is the worst case of everything an upload still does once its body has arrived, and
// the boot leaves room for it (internal/config): the upload's deadline bounds the BODY, and the
// windows an upload must finish inside bound all of it.
//
// The steps, in order and by the budget each actually gets, because the sum is only as right as this
// list — and an earlier edition of it dropped the first one:
//
//  1. StartParsing, which moves the book out of `uploading` — writeBudget;
//  2. the synchronous cut — CutBudget;
//  3. whichever end the cut reaches (FinishParse, ReleaseParseClaim or DeleteRefusedIntake) — writeBudget;
//  4. the re-read that gives the response the row as it stands — writeBudget;
//  5. the idempotency receipt the HTTP surface writes (httpapi.settleCtx, 10s), which is shorter
//     than a writeBudget and is covered by the fourth one here.
//
// None of them is bounded BY another: every one takes a detached context of its own, so the worst
// case is the sum.
//
// The reading surface is deliberately NOT in this list: materializing it runs the engine twice more
// (readmodel.MaterializeBudget), and the intake leaves that debt to the materializer's sweep rather
// than making an uploader wait for it (parseClaimed, the `!atIntake` guard).
const UploadSettle = CutBudget + 4*writeBudget

// CutBudget bounds the cut an upload waits for. Past it the book is accepted `parsing` and the queue
// finishes the job, so overrunning costs a less informative response and never a failed upload.
//
// A constant rather than a setting because it is part of UploadSettle above, which the boot compares
// against the idempotency claim window: a knob here would let an operator push the tail of an upload
// past the window in which its key can still be replayed.
const CutBudget = 90 * time.Second

// manifest asks the engine to cut the book, once its configuration is there to cut it against —
// rendering that configuration first, if this deployment carries a template (form Б, D39.130).
func (s *Service) manifest(ctx context.Context, claim pgstore.ParseClaim) (ingest.Manifest, error) {
	if s.Engine == nil {
		return ingest.Manifest{}, errors.New("books: no engine is configured")
	}
	workdir := claim.Workdir
	if _, err := os.Stat(workdir); err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			// Which of the two absences this is decides whether the book dies, so it is decided by the
			// SENTINEL rather than by the root's own existence (see storageIsThere).
			if !s.storageIsThere() {
				return ingest.Manifest{}, ErrStorageGone
			}
			return ingest.Manifest{}, ErrDirectoryGone
		}
		return ingest.Manifest{}, fmt.Errorf("books: read book directory: %w", err)
	}
	// The provisioning seam. It runs on EVERY pass and not only on the first, which is what makes a
	// deployment repairable: an operator who fixes a broken template has the books already waiting on
	// it rendered by the next sweep, with no attempt spent in the meantime (defer_).
	if err := s.provision(ctx, workdir, bookConfig{
		ID: claim.BookID, Title: claim.Title, SourceLang: claim.SourceLang,
		TargetLang: claim.TargetLang,
	}); err != nil {
		return ingest.Manifest{}, err
	}
	return s.Engine.Manifest(ctx, s.Cfg.EngineBinary, workdir)
}

// intakeReason turns what the engine ANSWERED into this platform's own word for it.
//
// ⚠ This is the consumer half of PD-196, and the whole defect lived in the sentence this function
// used to be. The engine mapped every failure onto exit 1, so "there is no book in these bytes",
// "this configuration will not load" and "another process holds the project" arrived as one number —
// and the only thing this side could do with an ExitError was call it `source_unreadable`, which the
// budget below turns into a DELETED upload. An operator's typo in a book.yaml was five attempts away
// from destroying a user's file.
//
// The engine now answers with a class (D39.131): a reserved band of exit codes, where each number
// names why the invocation was turned down before it did any work. So the verdict is read from the
// CODE, and exactly one code — `source_unreadable` — is allowed to mean the user's text is at fault.
// Everything else, INCLUDING a refusal class this build has never heard of and including a plain
// exit 1, is about the deployment or the host, and none of those may cost anyone their upload.
func intakeReason(err error) string {
	var exit *exec.ExitError
	// Not an answer at all: no binary at that path, no permission to execute it, a working directory
	// that is gone, a context that expired. And a signal is not an answer either — an engine the
	// machine killed said nothing about the book.
	if !errors.As(err, &exit) || !exit.Exited() {
		return ReasonParserUnavailable
	}
	switch exit.ExitCode() {
	case ingest.ExitSourceUnreadable:
		// The ONE class about the user's text: the source was read and cut and there is no book in
		// it. No `encoding`, `source_lang` or path setting explains an empty result from a successful
		// read, which is what makes it safe to act on (the engine's own refusal.go says so).
		return ReasonSourceUnreadable
	case ingest.ExitSchemaMismatch:
		// The project's database is not this binary's schema — an engine upgrade that has not been
		// migrated yet (row 174). The book is blameless and the repair is an operator's, so this waits
		// with the other deployment classes rather than spending a budget that ends in a rejection.
		return ReasonSchemaMismatch
	case ingest.ExitConfigInvalid:
		// The book's configuration will not load. That is the deployment's file — rendered from the
		// operator's template, or dropped in by hand — and it is the same kind of gap as no
		// configuration at all: a human fixes it and the next sweep succeeds. So it WAITS, spends no
		// attempt and never deletes anything.
		return ReasonNotConfigured
	default:
		// Everything else: a lock another tmctl holds, an unrecognised refusal class, an ordinary
		// exit 1. Bounded by the attempt budget, and a rejection on that budget keeps the file —
		// only `source_unreadable` removes it.
		return ReasonParserUnavailable
	}
}

// reject records the terminal end of an intake and removes what is left of it.
//
// The source is deleted, and it is a decision rather than housekeeping: the file cannot be parsed,
// no path in the contract re-parses or downloads it, and an authenticated route that writes bytes to
// an operator's disk and never removes them is a hole this pack would otherwise be opening. The ROW
// stays — the user has to be able to see that the book they uploaded did not make it.
func (s *Service) reject(ctx context.Context, claim pgstore.ParseClaim, reason string) error {
	// The DIRECTORY goes first and the row second, and the order is the crash window: dying between
	// the two then leaves a book still `parsing` with no source, which the next attempt fails on and
	// rejects properly. The other order leaves a directory with a full source and a row that says
	// `rejected` — and nothing ever looks at a rejected book again, so those bytes stay forever.
	//
	// ⚠ Only where the source is the BOOK's fault. A deployment that could not run the engine keeps
	// the file: deleting a user's upload because this host was misconfigured is not a decision to
	// make on their behalf.
	if reason == ReasonSourceUnreadable {
		s.removeDir(claim.Workdir)
	}
	// On a context of its own: this write is the only record that the intake is over, and the engine
	// call that led here may have used up everything the pass had.
	c, cancel := s.writeCtx(ctx)
	defer cancel()
	return s.Store.RejectBook(c, claim.BookID, claim.At, reason)
}

// Sweep finishes the intake walk of every book whose own process did not.
//
// The two halves are the two ways a walk stops: a request that went away mid-upload, and a parse
// whose process is gone. Neither is reachable from the thing that started it, which is what makes
// this the only cure — the same reason the run reconciler exists. The third way — the parse
// finished and its tree did not land — is not here: that book carries a debt, and the materializer's
// own pass answers it (readmodel.Drain).
//
// One book's failure never stops the sweep: these are independent books of independent accounts.
func (s *Service) Sweep(ctx context.Context) error {
	now := s.now()
	stuck, err := s.Store.StuckIntake(ctx, now.Add(-UploadGrace), now.Add(-claimGrace))
	if err != nil {
		return err
	}
	for _, b := range stuck {
		if err := ctx.Err(); err != nil {
			return err
		}
		switch b.Status {
		case "uploading":
			s.log().InfoContext(ctx, "removing an upload that never finished")
			s.abandon(ctx, b.ID, b.Workdir)
		case "parsing":
			// Each book gets its OWN budget, and it has to be one a parse can live inside: the engine
			// call is the same one the queue gives fifteen minutes. Sharing the pass's deadline meant a
			// large book was killed by it, the kill was read as a host that cannot run the engine, and
			// the attempt was spent — every pass, until the book was rejected for being big.
			//
			// ⚠ A budget the book gets is not a budget the PASS still has, and that gap was the rest of
			// the same defect (re-check of the dofix): the second book of a pass inherited whatever the
			// first one left, so a slow first parse handed the second a stub of a deadline and burned
			// its attempt on the timeout. A book that cannot be given its whole budget is therefore not
			// STARTED — the claim is taken inside Parse, so a book left for the next tick has spent
			// nothing — and the pass says so once rather than per book.
			if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) < jobs.JobTimeout {
				s.log().InfoContext(ctx, "intake pass ends early: what is left of it is shorter than a parse",
					"remaining", time.Until(deadline).String())
				return nil
			}
			c, cancel := context.WithTimeout(ctx, jobs.JobTimeout)
			err := s.Parse(c, b.ID)
			cancel()
			if err != nil {
				s.log().ErrorContext(ctx, "book could not be parsed", "err", err)
			}
		}
	}
	return nil
}