textmachine/platform/internal/runs/spawn.go

package runs

import (
	"context"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"

	"textmachine/platform/internal/ingest"
	"textmachine/platform/internal/money"
	"textmachine/platform/internal/pgstore"
	"textmachine/platform/internal/runner"
)

// Spawn starts the engine for a run that has already been admitted. It is the body of the queue's
// worker, and it does exactly one thing beyond starting a process: it refuses to start a SECOND one.
//
// That refusal is the point. A queue job is retried — after a platform restart, after a lease
// expires — and the run it names may well still be going, because the run is not the job's child.
// Spawning again would put two engines on one book, and the second would die on the project file's
// exclusive lock after the first had already been billed for the work.
func (s *Service) Spawn(ctx context.Context, runID string) error {
	run, err := s.Store.ReadRunForSpawn(ctx, runID)
	if errors.Is(err, pgstore.ErrNoRun) {
		// Finished, or never admitted. Either way there is nothing to start, and reporting a failure
		// would have the queue retry a run that is over.
		s.log().InfoContext(ctx, "queued run is no longer live", "run", runID)
		return nil
	}
	if err != nil {
		return err
	}
	if run.AlreadySpawned {
		s.log().InfoContext(ctx, "run already has a unit", "run", runID, "unit", run.UnitName)
		return nil
	}
	return s.spawnAttempt(ctx, run.LiveRun)
}

// spawnAttempt pins the binary, records what is about to happen, and creates the unit.
//
// The record is written BEFORE the unit exists, and that order is deliberate: a crash between the
// two leaves a row that says "an attempt with this unit name was about to start", which the
// reconciler can check against systemd and against the exit marker. The other order leaves a
// running engine that nothing in the database refers to.
func (s *Service) spawnAttempt(ctx context.Context, l pgstore.LiveRun) error {
	// The DEPLOYMENT's refusals first, because they are free and the next line is not: a status call
	// costs seconds of the engine's CPU, and an instance that cannot start runs at all would pay that
	// for every run on every sweep before arriving at the same answer.
	if err := s.runnable(); err != nil {
		return err
	}
	// A run the user has already asked to stop is not started. The claim in RecordSpawn refuses the
	// window this cannot see — the intent written after this read — but the ordinary case is a stop
	// pressed on a queued run before its worker got to it, and starting an engine there would spend
	// the account's money on work that was cancelled before it began.
	if l.StopRequestedAt != nil {
		s.log().InfoContext(ctx, "not starting a run that has been asked to stop", "run", l.RunID)
		return nil
	}
	// The engine's own money state for this book, read BEFORE anything is started, because afterwards
	// both of its numbers have already moved. It decides two different things: what this attempt will
	// owe when it ends (the difference from the committed figure) and what limit it may be given (the
	// cumulative cap the engine's flag actually means).
	//
	// A meter that cannot be read REFUSES the spawn. That is the deliberate half: a paid run this
	// platform could not bill correctly is worse than a run that starts one sweep later, and the
	// reconciler retries.
	m, err := s.bookMeter(ctx, l)
	if err != nil {
		return err
	}
	if m.leftover() {
		// Worth a line because it is evidence of a process that died mid-call, and because the cap
		// below is computed as if it were already gone — which it will be, the moment this attempt
		// opens the book for writing.
		s.log().InfoContext(ctx, "the engine's ledger carries a reservation from a process that is gone; the engine clears it at start",
			"run", l.RunID, "attempt", l.AttemptNo)
	}
	baseline, bookCap := m.committed, m.bookCap(l.Ceiling)
	if l.SpendBaseline != nil && l.CeilingArg > 0 {
		// A previous claim of THIS attempt already decided both, and this is a retry of it. The
		// numbers are re-used rather than recomputed because "the unit could not be created" does not
		// mean it was not: a systemd-run killed after it had already asked leaves an engine running,
		// and a fresh reading of the counter therefore contains that engine's own spend. Recomputing
		// hands it a second, larger limit and leaves ceiling_arg_micro_usd — the column whose whole
		// job is to answer "what limit did that process have" — describing neither of the two.
		baseline, bookCap = *l.SpendBaseline, l.CeilingArg
		s.log().InfoContext(ctx, "re-using the limit a previous claim of this attempt recorded",
			"run", l.RunID, "attempt", l.AttemptNo)
	}
	// The book's order, resolved against the tree as it stands NOW. Read here rather than carried on
	// every LiveRun: resolving it walks the book's units, and the reconciler's list of live runs must
	// not pay that for every run on every sweep (pgstore.ReadOrderForSpawn).
	order, err := s.Store.ReadOrderForSpawn(ctx, l.RunID)
	if err != nil {
		return err
	}
	spec, err := s.spec(l, bookCap, order)
	if err != nil {
		return err
	}
	claimed, err := s.Store.RecordSpawn(ctx, pgstore.SpawnRecord{
		AttemptID: l.AttemptID, Unit: spec.Unit, Binary: spec.Binary,
		EngineRunID: engineStreamID(l.RunID, l.AttemptNo),
		Ceiling:     l.Ceiling, CeilingArg: bookCap, Baseline: baseline,
	})
	if err != nil {
		return err
	}
	if !claimed {
		// Someone else — the queue worker, or a previous pass of the reconciler — got here first.
		// Losing this race is the ordinary case, not a failure.
		s.log().InfoContext(ctx, "another caller had already claimed this attempt", "run", l.RunID)
		return nil
	}
	// A marker left over from a previous attempt with the same name would be read as this one's
	// ending the moment the reconciler looked. Names carry the attempt number, so this only fires
	// after a re-run of the same attempt, but "only" is not "never".
	if err := os.Remove(spec.ExitMarker); err != nil && !errors.Is(err, fs.ErrNotExist) {
		return s.unclaim(ctx, l, fmt.Errorf("runs: clear stale exit marker: %w", err))
	}
	if err := s.Runner.Start(ctx, spec); err != nil {
		return s.unclaim(ctx, l, err)
	}
	return nil
}

// unclaim gives the attempt back when the unit was NOT created.
//
// Without it the claim is a lie the reconciler believes: a recorded unit name with no unit and no
// exit marker is exactly the shape of an interrupted run, so every sweep would RESTART the run —
// settling, taking a fresh hold, failing to spawn again — and a run whose engine never started would
// eat its whole ceiling a sweep at a time. Undone, the same attempt is simply retried.
func (s *Service) unclaim(ctx context.Context, l pgstore.LiveRun, cause error) error {
	if err := s.Store.ReleaseSpawnClaim(ctx, l.AttemptID); err != nil {
		return errors.Join(cause, err)
	}
	return cause
}

// spec is everything the unit will be, decided and nothing performed. Separate from spawnAttempt
// because what goes into the unit — the pinned binary, the ceiling, the marker command — is the part
// worth reading and worth asserting, and welding it to three side effects makes it neither.
//
// bookCap is the engine's ceiling ARGUMENT, already converted from the user's increment: the flag is
// a cumulative book cap, not a run budget (D39.122). `order` is the book's order resolved against the
// tree as it stands. BOTH are passed in rather than read here, and for one reason: this function
// PERFORMS NOTHING. Computing the cap costs a call to the engine and resolving the order costs a
// query, and a decision function that reaches for either stops being one — the tests that assert
// what argv a run gets would then need a database to ask about a run that does not exist.
func (s *Service) spec(l pgstore.LiveRun, bookCap money.MicroUSD, order pgstore.SpawnOrder) (runner.Spec, error) {
	if err := s.runnable(); err != nil {
		return runner.Spec{}, err
	}
	ceiling, err := s.ceilingFor(bookCap)
	if err != nil {
		return runner.Spec{}, err
	}
	maxUnits, err := maxUnitsFor(l, order)
	if err != nil {
		return runner.Spec{}, err
	}
	unit := unitName(l.RunID, l.AttemptNo)
	marker := s.markerPath(l.RunID, l.AttemptNo)
	return runner.Spec{
		Unit: unit,
		// The path this ATTEMPT is pinned to, which for a resume is the one the run started with
		// (row 139). Falling back to the configured path is for the first attempt, which has none yet.
		Binary: s.engineBinary(l),
		// ⚠ `--verify-bank` rides EVERY attempt of a run that asked for it, resumes included, and the
		// platform no longer decides where the engine stops. That decision moved into the engine with
		// its presented memory (storage schema v16, D39.158): a stop fires only on a cluster no
		// earlier stop has shown, so a resumed attempt carrying the flag auto-continues at the same
		// boundary by the engine's own predicate and its undecided rows ride to the editor marked
		// ⟨проверить⟩ (backend/internal/pipeline/mining.go, D39.42 п.3). Masking the flag here was
		// this platform's stand-in for that predicate while the engine still re-halted on everything
		// undecided; keeping the stand-in after the engine grew the memory would mean two answers to
		// one question, and the law of the seam gives it to the engine.
		Args: runner.TranslateArgs(l.Workdir, l.VerifyBank, s.Cfg.KeysFile,
			l.Resnapshot, l.AcceptRebill, maxUnits, ceiling),
		Workdir:    l.Workdir,
		Env:        engineEnv(engineStreamID(l.RunID, l.AttemptNo)),
		ExitMarker: marker,
		MarkerArgv: append(append([]string{}, s.Cfg.MarkerArgv...), marker, unit),
		MemoryMax:  s.Cfg.MemoryMax,
		TasksMax:   s.Cfg.TasksMax,
	}, nil
}

// meter is the engine's own money state for a book: what it has already paid for, and what its
// ledger has promised and not yet paid. Both are LIFETIME figures for the BOOK, across every run.
type meter struct {
	committed money.MicroUSD
	reserved  money.MicroUSD
}

// bookCap turns the increment the user bought into the number the engine's `--ceiling-usd` means.
//
// The flag is not a run budget. It replaces the book's own `ceilings.book_usd` and is compared
// against the book's CUMULATIVE committed + reserved on every reservation the engine takes
// (backend/internal/store/ledger.go Reserve, backend/cmd/tmctl/invocation.go). Handing it the
// increment therefore denies the first reservation of every run after the first — the engine exits
// 1, which this platform can only report as `failed`, having done no work at all.
//
// ⚠ The RESERVED half of the engine's own comparison is deliberately not added, and the reason is a
// line the first version of the formula did not account for. `store.Open` — the WRITE path every
// `translate` takes — runs `recoverReservations`, which zeroes every `reserved_usd` of the book
// before the first reservation is judged (backend/internal/store/store.go:110 and :278 — re-aimed
// 04.09, the targets had drifted from :88 and :214). `tmctl status` is read-only and deliberately
// does NOT run that pass (store.go:117-132 `OpenReadOnly` says so), so the figure
// the platform reads is a LEFTOVER of a crashed process, guaranteed to be gone by the time the cap
// is compared against anything. Adding it hands the run that much headroom BEYOND its hold: the
// engine stops late, settlement is capped at the hold, and the account underpays while the ledger
// reads like an engine overspend.
//
// This shipped as a named deviation and was RATIFIED on 09.08 after the orchestrator measured both
// formulas against the engine's own gate: the one with the reserved term overpaid by exactly the
// leftover (PD-158).
//
// The reserved figure is still read and still required (absent ≠ zero) because it is what makes the
// deviation safe to reason about: at spawn there is no other writer — the project file is held
// exclusively and the platform admits one live run per book — so anything reserved is by
// construction a leftover, never a live promise.
func (m meter) bookCap(increment money.MicroUSD) money.MicroUSD {
	return m.committed + increment
}

// leftover reports that a previous process died holding a reservation the engine will clear when it
// next opens the book for writing. The FACT only: money never reaches an INFO log (D39.84).
func (m meter) leftover() bool { return m.reserved > 0 }

// bookMeter asks the engine where the book's money stands. Both figures come from ONE call: they
// have to be consistent with each other, and a second call is a second CPU-seconds-long re-ingest of
// the source (unified backlog row 100).
func (s *Service) bookMeter(ctx context.Context, l pgstore.LiveRun) (meter, error) {
	if s.Engine == nil {
		return meter{}, nil // no repair channel configured: the dev path, where nothing settles either
	}
	rep, err := s.Engine.Status(ctx, s.engineBinary(l), l.Workdir)
	if err != nil {
		return meter{}, fmt.Errorf("runs: the book's spend could not be read, so the attempt is not started: %w", err)
	}
	// Absent is not zero (PD-40), and the two absences fail differently: a missing committed figure
	// makes this attempt pay for every earlier run of the book, a missing reserved figure hands the
	// engine a cap BELOW what its own ledger has already promised, which it refuses at once.
	if rep.Spend == nil {
		return meter{}, errors.New("runs: the status report carries no committed spend, so the attempt is not started")
	}
	if rep.Reserved == nil {
		return meter{}, errors.New("runs: the status report carries no reserved spend, so the attempt is not started")
	}
	return meter{committed: *rep.Spend, reserved: *rep.Reserved}, nil
}

// engineBinary is the path an attempt is PINNED to (unified backlog row 139), falling back to the
// configured one only for an attempt that has not been spawned yet. The engine ships more often than
// a run finishes, so asking the current binary about a book an older one is translating — or resuming
// with it — is a different thing from what was started.
func (s *Service) engineBinary(l pgstore.LiveRun) string {
	if l.EngineBinary != "" {
		return l.EngineBinary
	}
	return s.Cfg.EngineBinary
}

// unitName is the transient unit of one ATTEMPT. The attempt number is part of it because a resumed
// run is a new process and a new unit, and systemd will not accept a name that is still loaded.
func unitName(runID string, attempt int) string {
	return fmt.Sprintf("tm-run-%s-%d", runID, attempt)
}

// engineStreamID is the id this attempt's engine announces its event stream under. The format lives
// in pgstore because that is the package which writes it — the row is named when the ATTEMPT is
// created, not when its unit is, so that a drain happening in between has nothing to adopt (see
// pgstore.EngineStreamID). This is the same string, for the environment the unit gets.
func engineStreamID(runID string, attempt int) string {
	return pgstore.EngineStreamID(runID, attempt)
}

// engineEnv is the environment of the run's unit. Provider keys are deliberately NOT here — they
// reach the engine as the `--keys-file` ARGUMENT of `translate` (row 211, spec above), read by the
// engine itself, so they never pass through this process or the unit's environment — and the only
// thing this platform puts in it is the run's identity.
func engineEnv(streamID string) []string { return []string{"TM_TRACE_ID=" + streamID} }

func (s *Service) markerPath(runID string, attempt int) string {
	return filepath.Join(s.Cfg.StateDir, "runs", fmt.Sprintf("%s-%d.exit", runID, attempt))
}

// journalSize is where an attempt's own lines begin. The journal is per BOOK and append-only
// (D39.106 §2), so a resumed run appends its handshake after everything the previous attempt wrote.
func journalSize(workdir string) (int64, error) {
	st, err := os.Stat(filepath.Join(workdir, ingest.JournalFile))
	if errors.Is(err, fs.ErrNotExist) {
		return 0, nil // the emitter has not written anything yet, which is the ordinary first run
	}
	if err != nil {
		// ⚠ The PATH is dropped and the operation and the errno are kept, exactly as the admission
		// guard does it (sourceThere). This error is wrapped into a spawn failure that reaches an
		// ERROR line, and the path a book is at IS that book's identity: the directory of every book
		// this platform took in is `<books dir>/<book id>` (books.Receive), so a line carrying it
		// carries the identifier the zone's own standard keeps out of logs (PD-139, PD-99). The run
		// id the line already carries is the handle an operator resolves through `tmplatformctl`.
		var pe *fs.PathError
		if errors.As(err, &pe) {
			return 0, fmt.Errorf("runs: stat journal: %s: %w", pe.Op, pe.Err)
		}
		return 0, fmt.Errorf("runs: stat journal: %w", err)
	}
	return st.Size(), nil
}

// ErrOrderUnresolvable is a run whose ORDER no longer names anything in the book.
//
// ⛔ A REFUSAL TO SPAWN, and neither of the two numbers that could be substituted is honest. The
// order is stored as the IDENTITY of its boundary precisely so that a book cut again makes it stop
// resolving instead of quietly meaning something else (migration 00033); reading the dangling
// reference as «the whole book» would sell more than was bought, and reading it as «nothing» would
// sell less. What the buyer needs is to be asked again, over the new cut, at the new price — which
// is what the order form does the moment anybody looks at it.
//
// The run is left alone rather than failed here: the reconciler retries, the refusal is loud, and a
// re-cut cannot land under a LIVE run in the first place (the materializer's sweep skips a book that
// has one), so reaching this at all is already a state worth an operator's eyes.
var ErrOrderUnresolvable = errors.New("runs: the book was cut again and this run's order no longer names a boundary in it")

// maxUnitsFor is this run's VOLUME allowance: the book's order, minus what has already been
// delivered. Zero — no flag, no volume bound — is the WHOLE-BOOK order, which is what an order with
// no limit means and the only thing it can mean.
//
// ⛔ IT IS COMPUTED HERE, AT EVERY SPAWN, AND NOT FROZEN AT ADMISSION. A run is respawned — after a
// restart, after a resume, after a platform deploy — and by then some of its order may be delivered.
// Handing the engine the whole order again would let a run that stopped and continued twice buy the
// same volume three times over; the money ceiling would still bound the dollars, but the promise
// «you bought N units» would stop being about N.
//
// ⚠ AND IT NEVER GOES TO ZERO BY ARITHMETIC, which is the one way this could turn into "no bound at
// all". An order fully delivered floors to 1 rather than to 0, because 0 is the flag's own word for
// «unbounded»: a run respawned after its order was completed would otherwise be handed a whole book.
// One unit is the smallest honest allowance, and such a run has nothing fresh to spend it on — the
// engine finishes what it carried and stops.
func maxUnitsFor(l pgstore.LiveRun, o pgstore.SpawnOrder) (int, error) {
	if l.OrderedChapters == 0 {
		// ⛔ THE RE-PASS, and it is the ONE shape that writes a zero chapter count (pgstore.StartRun).
		// It buys no volume at all — it re-makes what a correction touched, up to the whole book — so
		// the book's own order says nothing about it. Bounding it by that order would be worse than
		// leaving it unbounded: a book already delivered through its ordered boundary has nothing
		// LEFT of that order, the allowance would floor at one unit, and the re-pass the buyer paid
		// a whole book's projection for would re-make exactly one.
		return 0, nil
	}
	if o.Order.Whole() {
		// ⛔ THE WHOLE BOOK IS BOUNDED TOO, on a continuation, and it is not belt-and-braces: with no
		// ceiling in force the engine builds no volume scope, and the scope is what carries its
		// protective ORDER of work — new book before re-made book. Without one, a continuation
		// carrying `--resnapshot` walks the edit wave in BOOK order and re-pays the beginning of the
		// book before editing the chapters that were just bought. The bound is the remainder itself,
		// so it takes nothing away from an order for everything.
		//
		// The FIRST run of a book gets none: nothing is delivered, so nothing can be re-made, and a
		// flag that bounds the whole of what exists bounds nothing.
		if o.HasPriorRun && o.UnitsLeft > 0 {
			return o.UnitsLeft, nil
		}
		return 0, nil
	}
	if !o.Resolved {
		return 0, fmt.Errorf("%w (run %s)", ErrOrderUnresolvable, l.RunID)
	}
	if left := o.Units - o.Delivered; left > 0 {
		return left, nil
	}
	return 1, nil
}