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)
	}
	spec, err := s.spec(l, bookCap)
	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). Passed in rather than computed here because
// computing it costs a call to the engine, and this function performs nothing.
func (s *Service) spec(l pgstore.LiveRun, bookCap money.MicroUSD) (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
	}
	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, 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 {
		return 0, fmt.Errorf("runs: stat journal: %w", err)
	}
	return st.Size(), nil
}