textmachine/platform/internal/runs/reconcile.go

package runs

import (
	"context"
	"errors"
	"fmt"
	"path/filepath"
	"time"

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

// spawnGrace is how long after an attempt was admitted the reconciler still believes systemd simply
// has not got to it. Under it, "no unit and no marker" means "not started yet"; over it, the same
// two facts mean the unit died without being able to say so — which is what a reboot looks like,
// since a transient unit does not survive one and its ExecStopPost never runs.
const spawnGrace = 60 * time.Second

// Sweep reconciles every run the database believes is live, then settles the money of every run
// that finished without it.
//
// It runs on a ticker AND at boot, and the boot pass is not a special case: the same reading of the
// same three sources — Postgres, the book's journal, the exit marker — restarts what a reboot
// interrupted (unified backlog row 138). systemd is asked only "is this unit still there", and its
// answer is evidence rather than truth (research/25 §Опс).
//
// One run's failure never stops the sweep: these are independent runs belonging to independent
// accounts, and a book whose directory an operator moved must not freeze everyone else's progress.
func (s *Service) Sweep(ctx context.Context) error {
	live, err := s.Store.ListLiveRuns(ctx)
	if err != nil {
		return err
	}
	for _, l := range live {
		if err := ctx.Err(); err != nil {
			return err
		}
		if err := s.withBudget(ctx, func(ctx context.Context) error { return s.reconcile(ctx, l) }); err != nil {
			s.log().ErrorContext(ctx, "run could not be reconciled", "run", l.RunID, "err", err)
		}
	}
	unsettled, err := s.Store.UnsettledRuns(ctx)
	if err != nil {
		return err
	}
	for _, u := range unsettled {
		if err := ctx.Err(); err != nil {
			return err
		}
		if err := s.withBudget(ctx, func(ctx context.Context) error { return s.settle(ctx, u) }); err != nil {
			s.log().ErrorContext(ctx, "run could not be settled", "run", u.RunID, "err", err)
		}
	}
	return nil
}

// runBudget is what ONE run may cost a pass of the sweep.
//
// It exists because the pass has a budget of its own and the work is not uniform: a spawn or a
// settlement is a `tmctl status` call, seconds of the engine's CPU each, and a handful of runs whose
// engine hangs used to spend the whole pass — after which the list, which is ordered the same way
// every time, was never reached past them. That is starvation with no upper bound, and it was
// invisible (register row PD-169; the counter that makes it visible is the sweep's own metric).
//
// Generous rather than tight: the call it bounds legitimately takes seconds on a large book, and the
// point is to bound the pathological case, not to race the ordinary one.
const defaultRunBudget = 60 * time.Second

func (s *Service) runBudget() time.Duration {
	if s.Cfg.RunBudget > 0 {
		return s.Cfg.RunBudget
	}
	return defaultRunBudget
}

// withBudget runs one item of a sweep under its own deadline. A cancelled item is not a failed one:
// the next pass reads the same world and tries again, and the money paths inside are transactional,
// so an item cut off mid-write leaves nothing half-done.
func (s *Service) withBudget(ctx context.Context, fn func(context.Context) error) error {
	c, cancel := context.WithTimeout(ctx, s.runBudget())
	defer cancel()
	return fn(c)
}

// Lag is how far the furthest-behind live run's journal is beyond this platform's cursor, in bytes.
//
// It is the tailer's own health, and until this pack nothing reported it: a projection that has
// stopped moving and a run that is simply quiet look the same from outside. Read separately from the
// sweep rather than folded into it because it is telemetry — a failure to measure must not be able
// to affect what the sweep does.
func (s *Service) Lag(ctx context.Context) (int64, error) {
	live, err := s.Store.ListLiveRuns(ctx)
	if err != nil {
		return 0, err
	}
	var worst int64
	for _, l := range live {
		size, err := journalSize(l.Workdir)
		if err != nil {
			continue // a book whose directory is unreadable is the reconciler's problem, not this one's
		}
		worst = max(worst, size-l.Position.Offset)
	}
	return worst, nil
}

func (s *Service) reconcile(ctx context.Context, l pgstore.LiveRun) error {
	// The cursor this call MOVED, not the one the sweep's snapshot was taken with. Everything below
	// that asks "has the stream said anything" has to ask about now: the snapshot is read before the
	// journal is drained, so a run whose first events arrive during this very sweep still looks
	// silent in it — and the repair channel would then be woken for a run that had just spoken.
	// Measured end to end, not reasoned about: a live run showed "edit 10/10" from a status report
	// seconds after its journal said "edit 0/10".
	// ⚠ A journal that cannot be read stops the PROJECTION and nothing else. It used to abort the
	// whole reconcile before the exit marker was even looked at, so one malformed line left a run
	// "translating" forever with its hold reserved — the engine long gone, the marker on disk, and
	// every sweep failing at the same byte. The lifecycle is decided from the marker and from
	// systemd; the journal only decides how fresh the numbers are.
	seq, drainErr := s.drainJournal(ctx, l)
	l.Position.LastSeq = max(l.Position.LastSeq, seq)
	if drainErr != nil {
		s.log().ErrorContext(ctx, "journal could not be materialized; the run's lifecycle continues without it",
			"run", l.RunID, "err", drainErr)
	}
	marker, err := runner.ReadMarker(s.markerPath(l.RunID, l.AttemptNo))
	switch {
	case err == nil, errors.Is(err, runner.ErrBadMarker):
		// There is a marker, so the unit ENDED, and the ending is decided here in both readings of it
		// — which is why the re-read is shared rather than repeated. A ceiling the stream announced
		// during this very pass outranks whatever the marker says, and putting the re-read on only the
		// readable branch buried exactly that case: a ceiling halt whose marker was corrupted came out
		// `failed` with no reason at all, on the one path where the exit code cannot say otherwise
		// either (acceptance dofix ФП-4).
		reason, rerr := s.freshPausedReason(ctx, l)
		if rerr != nil {
			return rerr // nothing is closed on facts we could not establish; the next pass tries again
		}
		l.PausedReason = reason
		if err != nil {
			// What the marker says cannot be read. Retrying is what the register row is about: the
			// write is atomic (temp+fsync+rename), so a marker that does not parse was changed by
			// something outside this platform, and no number of sweeps repairs it. The run is ended
			// with the fact recorded rather than re-read forever (register row PD-164).
			s.log().ErrorContext(ctx, "the exit marker cannot be read; the run is closed on what is left rather than reconciled forever",
				"run", l.RunID, "attempt", l.AttemptNo, "err", err)
			return s.finish(ctx, l, runner.Marker{Unit: l.UnitName, Result: runner.UnreadableMarkerResult})
		}
		if interruptedBySomeoneElse(l, marker) {
			// Exit 5 is a caught SIGTERM, and nothing here recorded asking for one. See the function.
			s.log().InfoContext(ctx, "the run was signalled by something that is not this platform's stop; restarting it",
				"run", l.RunID, "attempt", l.AttemptNo)
			return s.restart(ctx, l)
		}
		return s.finish(ctx, l, marker)
	case !errors.Is(err, runner.ErrNoMarker):
		return err
	}
	if l.UnitName == "" {
		if l.StopRequestedAt != nil {
			// Stopped before it ever started: the request was admitted, the money was held, and the
			// unit was never created. There is nothing to signal and nothing to wait for, so the run
			// ends here and its hold comes back whole through the ordinary settlement.
			return s.finishStopped(ctx, l)
		}
		// Admitted and never spawned: the platform stopped between the transaction and the unit, or
		// the queue entry was lost. The reconciler is the backstop for both.
		return s.spawnAttempt(ctx, l)
	}
	alive, err := s.Runner.Alive(ctx, l.UnitName)
	if err != nil {
		// The bus did not answer. "I could not ask" must never be read as "the run is gone": that
		// mistake restarts a live engine against its own project lock.
		return err
	}
	if alive {
		if l.StopRequestedAt != nil {
			// The intent is committed and the unit is still there. Either the signal never went out —
			// the platform died between writing the intent and asking systemd — or the engine is
			// finishing the chunk it has already paid for. Asking again is free and idempotent, and it
			// is the only thing that closes the first case.
			if err := s.Runner.Stop(ctx, l.UnitName); err != nil {
				return err
			}
			return nil
		}
		return s.maybeResync(ctx, l)
	}
	if s.now().Sub(l.AttemptStartedAt) < spawnGrace {
		return nil // systemd has not started it yet
	}
	if l.StopRequestedAt != nil {
		// The unit is gone and left no marker, which is the shape of a reboot — but this run was
		// asked to stop, and restarting it would spend the account's money on work its owner had
		// just cancelled. The intent decides, because it is the one fact here that is ours.
		return s.finishStopped(ctx, l)
	}
	reason, err := s.freshPausedReason(ctx, l)
	if err != nil {
		return err // same rule as above: a restart is a decision, and this one is not established yet
	}
	if l.PausedReason = reason; l.PausedReason != "" {
		// The stream already said a ceiling stopped this run, and a ceiling halt is not an
		// interruption: the engine ended on purpose and the next process would meet the same limit.
		// Restarting here would spend a hold on that, and it would also LOSE the reason — the restart
		// ends with a pause of its own, so a run stopped by the ENGINE's daily ceiling came back
		// wearing the account's. Found by the adversarial review of this pack; this branch is what
		// keeps `restart` a path only interruptions take.
		s.log().InfoContext(ctx, "the run's stream reported a ceiling and its unit is gone; closing it as paused",
			"run", l.RunID, "attempt", l.AttemptNo)
		return s.finish(ctx, l, runner.Marker{Unit: l.UnitName, Result: runner.UnitVanishedResult})
	}
	return s.restart(ctx, l)
}

// freshPausedReason re-reads what the run's projection says about a ceiling, because the sweep's
// snapshot predates the drain this pass just performed.
//
// It is asked at the two points where the answer DECIDES something — the ending of a run whose unit
// left a marker, readable or not, and the fate of one whose unit vanished without one — rather than
// after every drain: on those branches it is one query on a run that is about to change state, and
// everywhere else it would be one query per live run per tick for nothing.
//
// ⚠ A failed read is REPORTED, never absorbed, and the difference is a run's contract-visible state.
// It used to fall back to the snapshot, which is empty in exactly the case this exists for — the
// event landed in THIS drain — and the callers then closed the run on that emptiness. On the
// corrupt-marker branch there is no exit code to carry the ceiling independently and no next pass to
// correct it (the branch closes the run by design, PD-164), so one ordinary Postgres blip turned a
// resumable ceiling halt into `failed` with no reason. "I could not ask" is not an answer here any
// more than it is when systemd does not reply; the sweep logs it and the next pass reads the same
// world. Found by the state-machine lens of the dofix review, reproduced on live Postgres.
func (s *Service) freshPausedReason(ctx context.Context, l pgstore.LiveRun) (string, error) {
	reason, err := s.Store.RunPausedReason(ctx, l.RunID)
	if err != nil {
		return "", fmt.Errorf("runs: the run's paused reason could not be re-read after the drain: %w", err)
	}
	return reason, nil
}

// finishStopped closes a run whose end this platform asked for and whose unit left no marker to read:
// it was never created, or the manager went away with it.
//
// The two cases are not the same write. Where a unit EXISTED the ordinary finish applies. Where it
// never did, the close has to re-check that under a lock: the queue worker may have claimed the
// attempt between this sweep's snapshot and now, and closing the run then strands a live engine that
// no list looks at any more.
func (s *Service) finishStopped(ctx context.Context, l pgstore.LiveRun) error {
	if l.StopRequestedAt == nil {
		// The SNAPSHOT may predate the request — that is the whole reason one of the two callers is
		// here: the reconciler read the run, spent seconds settling, and the stop landed inside that
		// window. Reaching this function is itself the establishment of the fact, so the local copy is
		// made to say so rather than letting `outcome` read a nil and call the ending a failure.
		now := s.now()
		l.StopRequestedAt = &now
	}
	if l.UnitName != "" {
		return s.finish(ctx, l, runner.Marker{Unit: l.UnitName, Result: pgstore.StopRequestedResult})
	}
	// ⚠ An empty unit name is not proof that no process exists. `ReleaseSpawnClaim` clears the name
	// when the unit could NOT be created — and "could not be created" is not "was not created": a
	// systemd-run killed after it had already asked leaves an engine running, which is why the spend
	// baseline is deliberately kept on such an attempt (RecordSpawn). That baseline is the tombstone,
	// and where it exists the unit's name is derivable, so systemd is asked rather than assumed.
	// Closing over a live engine would leave it spending against a run this platform calls finished.
	if l.SpendBaseline != nil {
		unit := unitName(l.RunID, l.AttemptNo)
		alive, err := s.Runner.Alive(ctx, unit)
		if err != nil {
			// "I could not ask" is never "the run is gone" — the same rule the reconciler follows
			// everywhere else.
			return err
		}
		if alive {
			s.log().InfoContext(ctx, "a claim was given back but its unit exists; stopping it instead of closing the run",
				"run", l.RunID, "attempt", l.AttemptNo)
			return s.Runner.Stop(ctx, unit)
		}
	}
	finished, err := s.Store.FinishUnspawnedStop(ctx, l.RunID, l.AttemptID, s.now())
	if err != nil {
		return err
	}
	if !finished {
		// It was spawned after all. Nothing to do: the next pass sees a live unit and an intent that
		// says to stop it.
		s.log().InfoContext(ctx, "the stopped run was spawned after this sweep read it", "run", l.RunID)
		return nil
	}
	s.log().InfoContext(ctx, "run finished", "run", l.RunID, "status", "stopped",
		"result", pgstore.StopRequestedResult)
	return s.settle(ctx, l)
}

// drainJournal applies whatever the engine has written since the cursor, and returns where the
// cursor now stands.
func (s *Service) drainJournal(ctx context.Context, l pgstore.LiveRun) (int64, error) {
	if l.Quarantined {
		return l.Position.LastSeq, nil
	}
	path := filepath.Join(l.Workdir, ingest.JournalFile)
	sink := s.Store.NewRunSink(l.AttemptID, l.RunID, l.BookID)
	from := ingest.Position{Offset: l.Position.Offset, LastSeq: l.Position.LastSeq, LastHash: l.Position.LastHash}
	pos, _, err := ingest.Tail(ctx, path, l.EngineRunID, from, sink)
	switch {
	case errors.Is(err, ingest.ErrNoJournal):
		// The engine writes its first line whenever it gets there, and a run whose unit systemd has
		// not started yet has no journal at all. Ordinary, and not a failure: the tailer waits.
		return l.Position.LastSeq, nil
	case err != nil && !quarantines(err):
		// A moment we could not read it, not a stream we cannot read. Returned so the sweep logs it and
		// meets the same bytes again next pass.
		return l.Position.LastSeq, err
	case err != nil:
		// The RUN is left alone: it is spending money the account reserved, and our inability to read
		// its journal is not a reason to throw that away. Freshness falls back to the resync channel.
		s.log().ErrorContext(ctx, "journal cannot be materialized; falling back to resync",
			"run", l.RunID, "err", err)
		return l.Position.LastSeq, s.Store.Quarantine(ctx, l.AttemptID, err.Error())
	}
	if pos.Offset > l.Position.Offset {
		// Lines that were read and NOT applied — duplicates the cursor already covers — still move
		// the byte hint. Without this they are re-read on every sweep, forever.
		//
		// ⚠ There is one way for that to be a symptom rather than housekeeping, and it is free to
		// notice here: lines went past, none of them were ours, and none ever has been. The platform
		// names the stream before the unit starts and hands the name over as TM_TRACE_ID, so an
		// engine build that does not accept it announces itself under an id this cursor will never
		// match — and the run would then go quiet for its whole life, falling back to the resync
		// channel with nothing saying why.
		if l.EngineRunID != "" && pos.LastSeq == 0 {
			s.log().WarnContext(ctx, "the journal has lines and none belong to this attempt's stream; the engine may not be honouring the run id it was given (falling back to resync)",
				"run", l.RunID, "attempt", l.AttemptNo)
		}
		return pos.LastSeq, s.Store.SaveCursor(ctx, l.AttemptID, pgstore.Position{Offset: pos.Offset})
	}
	return pos.LastSeq, nil
}

// quarantines decides what a failure to materialize the journal MEANS: a stream this platform cannot
// read, or a moment in which it could not read one.
//
// That distinction is the whole decision, and getting it wrong is expensive in both directions. A
// gap, a changed payload, a malformed line, a line past the buffer: none is repaired by reading the
// same bytes again, and retrying them forever is what wedged a run — so those stop the projection.
// Our own shutdown and a lock Postgres broke are the opposite: the very next sweep reads exactly the
// same bytes successfully, and stopping the projection over one blinds a live, paying run for good.
func quarantines(err error) bool {
	switch {
	case err == nil:
		return false
	case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
		return false
	case pgstore.IsTransient(err):
		return false
	default:
		return true
	}
}

// maybeResync refreshes a live run from `tmctl status --json`, at most once every ResyncEvery.
//
// ⚠ What it produces is HONESTLY STALE, and by an amount the interval names: the run moves
// continuously and this reads it every few minutes. Now that the emitter writes (D39.131) it is what
// it was designed to be — the REPAIR path, reached only where the stream has said nothing or has
// been quarantined — rather than the only source of progress it had to be until then.
func (s *Service) maybeResync(ctx context.Context, l pgstore.LiveRun) error {
	// The stream, when there is one, is the FRESHER source and the free one. A status call costs
	// seconds of CPU on the engine's side, every time, and can only report what the journal has
	// already said — so the repair channel runs where there is nothing to repair from: an attempt
	// whose cursor has never moved, or one whose materialization was quarantined and for which this
	// is now the only source.
	//
	// ⚠ This gate used to carry a second job that it no longer has to: before the engine reported the
	// per-wave split (row 99, landed D39.122) a resync could only fold ONE aggregate over the two
	// counters the stream keeps apart, which is how a live run showed "edit 10/10" seconds after its
	// journal said "edit 0/10". ApplyStatus now materializes the same four counters as the stream.
	if l.Position.LastSeq > 0 && !l.Quarantined {
		return nil
	}
	if !s.dueForResync(l.RunID) {
		return nil
	}
	now := s.now()
	rep, err := s.Engine.Status(ctx, s.engineBinary(l), l.Workdir)
	if err != nil {
		// A status call that fails is not a run that failed. It is logged and retried on the next
		// interval; the read model keeps the last figures it had.
		s.log().WarnContext(ctx, "resync failed", "run", l.RunID, "err", err)
		return nil
	}
	return s.Store.ApplyStatus(ctx, l.RunID, l.BookID, rep, now)
}

// dueForResync is the rate limit, and it is the whole of what makes the resync affordable: the
// sweep runs every few seconds and each status call costs seconds of CPU. It records the decision as
// it makes it, so two sweeps cannot both conclude "due".
//
// In memory on purpose: this is a rate limit, not a fact — losing it on restart costs one extra
// status call, whereas persisting it would put a row write on a path that exists to avoid work.
func (s *Service) dueForResync(runID string) bool {
	if s.Cfg.ResyncEvery <= 0 || s.Engine == nil {
		return false
	}
	now := s.now()
	if s.resynced == nil {
		s.resynced = map[string]time.Time{}
	}
	if last, ok := s.resynced[runID]; ok && now.Sub(last) < s.Cfg.ResyncEvery {
		return false
	}
	s.resynced[runID] = now
	return true
}

// finish closes a run whose unit has ended.
func (s *Service) finish(ctx context.Context, l pgstore.LiveRun, m runner.Marker) error {
	delete(s.resynced, l.RunID) // a finished run keeps no rate-limit entry: the map is per process
	status, pausedReason, exit := outcome(l, m)
	closed, err := s.Store.FinishRun(ctx, pgstore.RunEnding{
		RunID:      l.RunID,
		AttemptID:  l.AttemptID,
		Status:     status,
		ExitResult: m.Result,
		ExitCode:   exit,
		// Written with the status and not by a second call: a `paused` run whose reason lands one
		// statement later is a run the resume path can read in between and misjudge.
		PausedReason: pausedReason,
		Now:          s.now(),
	})
	if err != nil {
		return err
	}
	if !closed {
		// The run was already finished, or this pass is holding a snapshot of an attempt that is no
		// longer the live one — a stop and a resume can both have happened since it was taken. Doing
		// anything further here would settle the money of a run somebody else is now running.
		s.log().InfoContext(ctx, "the run moved on since this pass read it; nothing to finish",
			"run", l.RunID, "attempt", l.AttemptNo)
		return nil
	}
	s.log().InfoContext(ctx, "run finished", "run", l.RunID, "status", status, "result", m.Result)
	// Settling immediately rather than waiting for the next sweep: the hold is the account's money
	// and every second it stays reserved is a second the user cannot start another book.
	return s.settle(ctx, l)
}

// outcome maps what systemd saw onto the product status of the run, and onto the reason it carries.
//
// The engine's exit contract is the input for a process that exited on its own (ingest/exit.go); a
// process that did NOT exit on its own has no exit code to map, and $SERVICE_RESULT is what tells a
// stop we asked for from a kill we did not.
//
// ⚠ The CEILING branch is the whole of PD-113, and it reads two independent channels because they
// fail independently: the stream's `ceiling` event and exit code 4. A journal that could not be
// written still leaves the code; a process killed before it could exit still leaves the event. Both
// answer `paused`, and neither may answer `failed` — the stop is resumable, and the contract forbids
// it (§BookStatus). Before the emitter landed there was only the event, and a ceiling halt with no
// journal was recorded as an infrastructure failure.
//
// A stop this platform ASKED for is a different question and is answered here, because the answer is
// ours: exit 5 says a signal wound the run down and says nothing about WHO sent it, so the record
// written before the signal went out is what tells a user's stop from a reboot (register row
// PD-152). What that record's ABSENCE means is decided one level up — see interruptedBySomeoneElse.
func outcome(l pgstore.LiveRun, m runner.Marker) (status, pausedReason string, exitCode *int) {
	if l.PausedReason != "" {
		// The stream already said this run stopped on a ceiling, and that survives ANY exit: the
		// process may have been killed on its way out, and the fact does not become less true.
		return "paused", l.PausedReason, nil
	}
	code, exited := m.Exited()
	if exited {
		switch ingest.OutcomeOf(code) {
		case ingest.OutcomeClean, ingest.OutcomeFlagged:
			// Flagged is a finished translation whose notes carry the flags. The engine's own clean
			// answer wins over a stop that arrived while it was already done.
			return "ready", "", &code
		case ingest.OutcomeBankStop:
			// The bank-signing stop. The ATTEMPT is over — the engine exits — and the run is left in
			// a state `resume` continues from once the decisions are complete.
			return "awaiting_bank", "", &code
		case ingest.OutcomeStopped:
			// A caught signal over a run this platform asked to stop. WHICH of the two records is
			// older decides nothing here, and reading the clock instead recorded `failed` for a stop
			// the user did ask for (acceptance dofix ФП-6): the guard below exists so that a run which
			// finished BY ITSELF a moment before the click is not called cancelled — and a run that
			// finished by itself does not exit 5. Its own endings are answered above, in this same
			// switch, before anything looks at a timestamp. What is left under exit 5 with an intent
			// on file is a stop, whether the signal that wound it down was ours or a reboot's that
			// arrived first; the two clocks being compared are not even the same one — the marker
			// carries the host's, the intent the platform's.
			if l.StopRequestedAt != nil {
				return "stopped", "", &code
			}
		case ingest.OutcomeCeiling:
			// No event reached us — the branch above would have taken it — so WHICH ceiling stopped the
			// run is genuinely unknown here, and that is what gets recorded. It is not a rare corner: a
			// quarantined projection never drains at all, so every ceiling halt of such a run arrives
			// this way. Guessing `credit_exhausted` would light the account-level halted flag for an
			// account that may have plenty of money (register row PD-203), and guessing `daily_ceiling`
			// would refuse a resume the user could have completed.
			return "paused", pgstore.PausedCeilingUnknown, &code
		}
	}
	if stoppedOnRequest(l, m) {
		// Exit 5 after our SIGTERM is the ordinary shape of a graceful stop; a kill after the stop
		// timeout, or an OOM in the same window, is an ungraceful one. Both are the stop the user
		// asked for, and the machine's own word for what happened stays in exit_result.
		return "stopped", "", nil
	}
	if !exited && m.Result == "success" {
		// Killed by a signal and systemd calls the result success: a stop nobody recorded, and no exit
		// code at all to read it from.
		return "stopped", "", nil
	}
	if exited {
		// Everything left, including the REFUSAL band. A refusal did no work and spent nothing, and
		// `failed` is the honest product status for it: the run cannot proceed, and the account is
		// charged the nothing it cost. It is deliberately NOT restarted — a refusal is reproducible by
		// construction (a configuration that will not load, a project another process holds), so a
		// restart would re-take a hold and meet the same answer on every sweep, forever. The class is
		// in exit_code for an operator, and the intake — the one path that acts destructively on a
		// refusal — reads the band itself (books.intakeReason).
		return "failed", "", &code
	}
	return "failed", "", nil
}

// interruptedBySomeoneElse reports an attempt that a signal wound down without this platform having
// asked for one.
//
// Exit 5 means the engine caught SIGINT/SIGTERM and shut down cleanly. It does NOT say who sent it,
// and there are two senders that leave no record: an ordinary host reboot, where the user manager
// stops every unit it owns, and an operator running `systemctl --user stop` by hand. The marker
// cannot tell them apart — both are `exit-code/exited/5` — so this is a decision and not a
// deduction, and it is written down as one.
//
// It reads as an INTERRUPTION: not a failure, not the user's stop, and therefore a candidate for the
// restart path that unified backlog row 138 exists for. The argument is asymmetry. Reading a reboot
// as a user's stop leaves every run on the host dead after a routine restart, with its budget
// unspent and nobody to resume it — which is the remaining half of PD-152 and the one thing row 138
// was built to prevent. Reading a hand-stop as an interruption costs one restart of a run whose
// owner never asked for it to end, and the documented way to end a run — the contract's stop handle
// — records an intent that this function then honours. The engine's own money is untouched either
// way: the restart settles the old attempt at what it actually spent and gives the new one what is
// left of the run's budget.
//
// A ceiling halt is never this: it has its own code and its own event, and both are read before.
func interruptedBySomeoneElse(l pgstore.LiveRun, m runner.Marker) bool {
	if l.StopRequestedAt != nil || l.PausedReason != "" {
		return false
	}
	code, exited := m.Exited()
	return exited && code == ingest.ExitStopped
}

// stoppedOnRequest reports whether this platform's own stop is what ended the attempt.
//
// The timestamp comparison is the race guard the design owes: a run that finished by itself a moment
// before someone pressed stop was not stopped by them, and calling it `stopped` would hide that
// ending behind a cancelled one. A marker with no timestamp is evidence we do not have, and then the
// intent — the only fact left — decides.
//
// ⚠ What reaches it is narrower than it looks, and deliberately so: every ending the engine has a
// code for is answered in `outcome`'s switch first, including the caught signal itself. So the
// clocks are compared only where the engine said something this platform cannot classify — an exit
// 1, a refusal, a process that did not exit at all — and where the comparison being wrong costs a
// label rather than a lifecycle decision.
func stoppedOnRequest(l pgstore.LiveRun, m runner.Marker) bool {
	if l.StopRequestedAt == nil {
		return false
	}
	return m.At.IsZero() || !l.StopRequestedAt.After(m.At)
}

// settle resolves the money of an attempt whose process is gone.
//
// The figure is the engine's OWN committed spend, read through `tmctl status --json` — the ratified
// repair channel — after the process has exited, so there is no lock to contend with and no
// inference from a stream that may have been truncated. If it cannot be read the reservation stays
// OPEN and the next sweep tries again: a hold that is still reserved is visible and recoverable,
// whereas a settlement against a guessed number is neither.
//
// ⚠ What this deliberately does NOT do is the escrow half of research/25 — write-ahead intent, the
// `uncertain` state, `closing` until the terminal seq. That is unified backlog row 136 and its own
// prompt; building half of it here would put a second, weaker answer next to the one being designed.
func (s *Service) settle(ctx context.Context, l pgstore.LiveRun) error {
	if s.Engine == nil {
		return nil
	}
	key := pgstore.ReservationKey(l.RunID, l.AttemptNo)
	// An attempt with no unit never reached the engine, so there is nothing to ask it about — and
	// asking anyway is what kept such a hold reserved for good on the very hosts that produce this
	// case: a deployment whose engine cannot be run fails `Status` on every pass, and the money of a
	// run that never ran would wait for an answer that never comes. The row is re-checked under the
	// lock inside ReleaseUnspawned, so a snapshot that was spawned in the meantime is refused there
	// and settles the ordinary way below. Found by cross-family review of the acceptance dofix.
	if l.UnitName == "" && l.SpendBaseline == nil {
		switch err := s.Store.ReleaseUnspawned(ctx, key, l.AttemptID, s.now()); {
		case err == nil, errors.Is(err, pgstore.ErrNoReservation):
			return s.Store.MarkSettled(ctx, l.RunID, s.now())
		case !errors.Is(err, pgstore.ErrAttemptSpawned):
			return err
		}
		s.log().InfoContext(ctx, "settlement deferred: the attempt was spawned after this sweep read it",
			"run", l.RunID, "attempt", l.AttemptNo)
		return nil // the next sweep sees the unit and settles against its baseline
	}
	rep, err := s.Engine.Status(ctx, s.engineBinary(l), l.Workdir)
	if err != nil {
		s.log().WarnContext(ctx, "settlement deferred: the engine's committed spend could not be read",
			"run", l.RunID, "err", err)
		return nil
	}
	if rep.Spend == nil {
		// Absent is not zero (PD-40): a settlement computed from a missing figure would release the
		// whole hold and charge nothing.
		s.log().WarnContext(ctx, "settlement deferred: the status report carries no committed spend",
			"run", l.RunID)
		return nil
	}
	// What the book's counter says NOW, bounded by what it said when a later attempt of the same book
	// started. Settlement is allowed to defer and a deferred one can be overtaken: the run is finished,
	// so nothing stops the account from starting another on the same book, and the counter this reads
	// is the BOOK's lifetime total. Without the bound the deferred settlement of the first run pays for
	// the second one's work as well, and the second then pays for it again.
	bound, err := s.Store.SpendBound(ctx, l.BookID, l.AttemptID)
	if err != nil {
		return err
	}
	committed := *rep.Spend
	if bound != nil && *bound < committed {
		committed = *bound
	}
	if l.SpendBaseline != nil && committed < *l.SpendBaseline {
		// The book's meter went BACKWARDS across this attempt, which no run produces: it is a project
		// database that was replaced or restored. attemptSpend clamps to zero rather than paying an
		// account for it, and this says so out loud — a settlement of nothing is not something to
		// discover from a balance. WARN and not INFO because the subject is money; the figures
		// themselves stay out of the line (D39.84).
		s.log().WarnContext(ctx, "the book's meter reads below this attempt's own baseline; it is settled at nothing",
			"run", l.RunID, "attempt", l.AttemptNo)
	}
	spent, err := attemptSpend(l, committed)
	if err != nil {
		if l.UnitName == "" {
			// No baseline because there was no spawn: the attempt was admitted and never started, so it
			// cost nothing and its hold comes back WHOLE. Without this the money of a run that never ran
			// is reserved forever.
			//
			// "Never started" is re-checked against the ROW, not against this snapshot: the sweep read
			// its list before working through it, and a run can be spawned, spend and exit inside that
			// window. Measured: the hold of an attempt that spent $0.50 came back in full.
			switch err := s.Store.ReleaseUnspawned(ctx, key, l.AttemptID, s.now()); {
			case errors.Is(err, pgstore.ErrAttemptSpawned):
				s.log().InfoContext(ctx, "settlement deferred: the attempt was spawned after this sweep read it",
					"run", l.RunID, "attempt", l.AttemptNo)
				return nil // the next sweep sees the unit and settles against its baseline
			case err != nil && !errors.Is(err, pgstore.ErrNoReservation):
				return err
			}
			return s.Store.MarkSettled(ctx, l.RunID, s.now())
		}
		// Spawned, but with no baseline — an attempt from before this column existed. Charging the
		// book's lifetime total would bill every earlier run again and charging zero would give this
		// one away, so it is left open and said out loud.
		s.log().ErrorContext(ctx, "settlement withheld: this attempt ran without a spend baseline",
			"run", l.RunID, "attempt", l.AttemptNo)
		return nil
	}
	if err := s.Store.Settle(ctx, key, spent, s.now()); err != nil {
		if errors.Is(err, pgstore.ErrNoReservation) {
			// Already closed by an earlier pass. Settling is idempotent by refusal, not by repetition.
			return s.Store.MarkSettled(ctx, l.RunID, s.now())
		}
		return err
	}
	return s.Store.MarkSettled(ctx, l.RunID, s.now())
}

// restart brings back a run whose unit vanished without a word — the case a reboot produces, since
// transient units do not survive one (research/25 §Форма, unified backlog row 138).
//
// The money is closed on the old attempt and reopened on the new one rather than carried over: the
// old hold was taken for a process that is gone, and what the account still owes for it is the
// engine's committed figure, not the ceiling. What remains of the run's budget is what the new
// attempt gets, so a restart cannot spend the run's ceiling twice.
//
// The LIMIT the resumed process is given is not that remainder: spawnAttempt reads the engine's
// meter again and hands it committed + the remainder, because the flag caps the book's cumulative
// spend and the interrupted attempt moved it (D39.122, and meter.bookCap for why the reserved figure
// is deliberately not in that sum).
func (s *Service) restart(ctx context.Context, l pgstore.LiveRun) error {
	if err := s.settle(ctx, l); err != nil {
		return err
	}
	next, v, err := s.reopen(ctx, l, fromALiveRun)
	if errors.Is(err, pgstore.ErrStopRequested) {
		// The user pressed stop while this pass was settling. The intent outranks the restart, and it
		// is still there — the refusal is what kept it — so the run ends here instead.
		s.log().InfoContext(ctx, "restart abandoned: the run was asked to stop", "run", l.RunID)
		return s.finishStopped(ctx, l)
	}
	if errors.Is(err, pgstore.ErrRunFinished) {
		// Another generation of the sweep closed the run while this one was settling. Its snapshot is
		// simply old, and there is nothing left to restart — bringing it back would take a second hold
		// and start an engine for a run whose owner has been told it ended.
		s.log().InfoContext(ctx, "restart abandoned: the run was finished by another pass", "run", l.RunID)
		return nil
	}
	if err != nil {
		return err
	}
	switch v {
	case exhausted:
		// `paused` WITH its reason: the contract describes PausedReason as the reason a run is paused,
		// and a paused run with a null reason gives the screen nothing to say. PauseRun is the path
		// that sets it; FinishRun does not. Done HERE and not inside reopen because the other caller —
		// a resume — works on a run that has already ended, and pausing one of those would rewrite how
		// it ended.
		//
		// A stop that arrived while this pass was settling outranks the pause: the store refuses and
		// the run ends as what it is, stopped.
		// The reason the STREAM gave wins over this path's own: overwriting a daily-ceiling halt with
		// the account's reason is what made it look like an exhausted account.
		//
		// ⚠ Unreachable today and kept anyway, which is stated rather than implied: `restart` is only
		// entered from branches that have already established the run reported no ceiling, so a landing
		// on these three lines SURVIVES. It is a belt against a later path reaching here with a reason
		// already set — the same shape as the ErrStopRequested answer in Resume below.
		reason := pgstore.PausedCreditExhausted
		if l.PausedReason != "" {
			reason = l.PausedReason
		}
		paused, err := s.Store.PauseRun(ctx, l.RunID, l.AttemptID, reason, s.now())
		if err != nil {
			if errors.Is(err, pgstore.ErrStopRequested) {
				s.log().InfoContext(ctx, "pause abandoned: the run was asked to stop", "run", l.RunID)
				return s.finishStopped(ctx, l)
			}
			return err
		}
		if !paused {
			// The run was already over, or the attempt this pass holds is no longer its live one.
			// Said rather than assumed: the line below would otherwise announce a pause that did not
			// happen, which is what the register row is about (PD-141).
			s.log().InfoContext(ctx, "the run moved on since this pass read it; nothing to pause",
				"run", l.RunID, "attempt", l.AttemptNo)
			return nil
		}
		s.log().InfoContext(ctx, "run paused", "run", l.RunID, "reason", reason)
		return nil
	case deferred:
		return nil
	}
	s.log().InfoContext(ctx, "interrupted run restarted", "run", l.RunID, "attempt", next.AttemptNo)
	return s.spawnAttempt(ctx, next)
}

// verdict is what reopening a run came to. Three outcomes because the callers want different things
// from the two that are not "it started": the reconciler treats both as "nothing more this pass",
// while a user's resume must tell "your money is still being counted, ask again" from "there is
// nothing left of this run's budget".
type verdict int

const (
	reopened verdict = iota
	// deferred — the previous attempt's money is not settled yet. Temporary by construction: the
	// settlement is retried by every sweep.
	deferred
	// exhausted — nothing is left of the run's budget, or the account cannot carry a new hold. What
	// to DO about it differs by caller, so reopen only reports it.
	exhausted
)

// reopen closes an attempt and opens the next one with what is LEFT of the run's budget.
//
// It is the shared body of the reconciler's restart and of the contract's resume, and sharing it is
// the point: both are "this run continues in a new process", both must give the new attempt the
// remainder rather than the whole ceiling — one run may not spend its ceiling twice — and both must
// refuse to open a second hold while the first is still open.
// requireLive says whether the CALLER is working from a snapshot of a run it believes is still
// going. The reconciler is; a resume is not — it continues a run that has ended on purpose.
type liveness bool

const (
	// fromALiveRun is the reconciler: its snapshot was taken before it did anything, and another
	// sweep generation can have finished the run in between (register row PD-181's class).
	fromALiveRun liveness = true
	// fromAFinishedRun is the resume handle: the run has ended and re-opening it is the point.
	fromAFinishedRun liveness = false
)

func (s *Service) reopen(ctx context.Context, l pgstore.LiveRun, from liveness) (pgstore.LiveRun, verdict, error) {
	// Settling is allowed to DEFER — the engine's figure may not be readable — and a deferral must not
	// become a second reservation. Asked rather than assumed: proceeding on an unsettled attempt holds
	// the ceiling twice and strands the first hold where no later sweep looks for it.
	open, err := s.Store.AttemptReservationOpen(ctx, l.RunID, l.AttemptNo)
	if err != nil {
		return pgstore.LiveRun{}, deferred, err
	}
	if open {
		s.log().WarnContext(ctx, "restart deferred: the previous attempt is not settled yet",
			"run", l.RunID, "attempt", l.AttemptNo)
		return pgstore.LiveRun{}, deferred, nil
	}
	budget := s.Pricing.Ceiling(l.CeilingChapters)
	spent, err := s.Store.RunSpent(ctx, l.RunID)
	if err != nil {
		return pgstore.LiveRun{}, deferred, err
	}
	remaining := budget - spent
	if remaining <= 0 {
		s.log().InfoContext(ctx, "the run has no budget left", "run", l.RunID)
		return pgstore.LiveRun{}, exhausted, nil
	}
	offset, err := journalSize(l.Workdir)
	if err != nil {
		return pgstore.LiveRun{}, deferred, err
	}
	// Nil = inherit the pinned build. A different one only when an operator asked for it (row 139).
	var version *string
	if s.Cfg.AllowEngineVersionChange && s.Cfg.EngineBinary != "" {
		version = &s.Cfg.EngineBinary
	}
	next, err := s.Store.RestartRun(ctx, pgstore.RestartInput{
		RunID:     l.RunID,
		AttemptID: l.AttemptID,
		// The reconciler's own guard against a stale snapshot: if another pass finished this run
		// while this one was settling, it must not be brought back to life.
		OnlyIfLive:   bool(from),
		UserID:       l.UserID,
		BookID:       l.BookID,
		Ceiling:      remaining,
		Offset:       offset,
		EngineBinary: version,
		Now:          s.now(),
	})
	if errors.Is(err, pgstore.ErrInsufficientCredit) {
		// The balance went elsewhere while this run was down. Honest state, not a failure.
		s.log().InfoContext(ctx, "the run cannot be continued on the current balance", "run", l.RunID)
		return pgstore.LiveRun{}, exhausted, nil
	}
	if err != nil {
		return pgstore.LiveRun{}, deferred, err
	}
	return next, reopened, nil
}

func (s *Service) enqueueNow(ctx context.Context, runID string) error {
	if s.Queue == nil {
		return nil // no queue configured: the reconciler still picks the run up on its next sweep
	}
	return s.Queue.EnqueueRunNow(ctx, runID)
}

// Stop is the contract's stop handle (§stopRun).
//
// What it does is record the INTENT and then ask systemd, in that order, and the order is the whole
// mechanism: the engine catches SIGTERM and exits 1, so the exit marker says the same thing for a
// stop and for a crash (register row PD-152). The row written before the signal is what tells them
// apart afterwards — and what stops the reconciler restarting a run whose owner just cancelled it.
//
// The systemd call is allowed to fail without failing the request: the intent is committed, the
// reconciler re-issues the stop on its next pass, and answering an error to a stop that WILL happen
// would invite the user to press it again.
func (s *Service) Stop(ctx context.Context, userID, runID string) (pgstore.Run, error) {
	run, unit, err := s.Store.RequestStop(ctx, userID, runID, s.now())
	if errors.Is(err, pgstore.ErrRunNotLive) {
		// No status in the message: RequestStop answers a zero Run alongside this error, so anything
		// read off it would be an empty string dressed up as a fact.
		return pgstore.Run{}, ErrNotStoppable
	}
	if err != nil {
		return pgstore.Run{}, err
	}
	s.log().InfoContext(ctx, "run stop requested", "run", runID)
	if unit == "" {
		// Admitted and not yet spawned. There is nothing to signal; the reconciler ends the run and
		// gives the hold back on its next pass.
		return run, nil
	}
	if err := s.Runner.Stop(ctx, unit); err != nil {
		s.log().ErrorContext(ctx, "the stop was recorded but systemd did not take it; the reconciler will retry",
			"run", runID, "err", err)
	}
	return run, nil
}

// Resume is the contract's resume handle (§resumeRun): it continues a run that was stopped.
//
// The mechanics are the reconciler's — a new attempt with what is LEFT of the run's budget — because
// a resume and a restart are the same event seen from two sides, and a second implementation of
// "give this run another process" is a second place for the money to be got wrong.
//
// What this adds is the state machine the contract describes, and its shape is dictated by that text
// rather than chosen:
//   - a run the user stopped continues;
//   - a bank stop answers 409 while the set of decisions is incomplete — and it is incomplete for
//     every book today, because nothing materializes the bank at all (companion §3: the export
//     artifact the read needs does not exist);
//   - a run paused by a ceiling "returns it to the same state", so it answers 202 with the run
//     unchanged rather than pretending an action happened;
//   - anything else — already going, finished, failed — is a 409.
func (s *Service) Resume(ctx context.Context, userID, runID string) (pgstore.Run, error) {
	// Ownership FIRST, deployment health second. The other order answers "this service cannot start
	// runs" to a caller asking about a run that is not theirs and to one asking about a run that does
	// not exist — which is both a worse answer and a small oracle about the shape of the deployment.
	l, err := s.Store.ReadRunForResume(ctx, userID, runID)
	if err != nil {
		return pgstore.Run{}, err
	}
	if err := s.runnable(); err != nil {
		return pgstore.Run{}, err
	}
	switch l.Status {
	case "stopped":
	case "paused":
		// A run stopped by the engine's own DAILY ceiling cannot be continued by anything this
		// platform can do, and re-opening it would be a loop with a cost. `ceilings.day_usd` lives in
		// a book.yaml the operator owns (D39.110 §2b); the platform neither sets it nor can read it,
		// so the next attempt meets the same limit within the same day and comes straight back — each
		// pass paying for a `tmctl status` and a transient unit.
		//
		// A deferred retry was the alternative and is rejected on the same fact: the boundary belongs
		// to the ENGINE's ledger day, which this side cannot see, so any timer here would be a guess
		// that spends spawns on a schedule nobody chose. 409 is the answer the contract already gives
		// for a run whose continuation depends on something outside this platform — the same shape as
		// the bank stop below — and it is the honest one.
		if l.PausedReason == pgstore.PausedDailyCeiling {
			return pgstore.Run{}, fmt.Errorf("%w: the engine's daily ceiling stopped it", ErrNotResumable)
		}
	case "awaiting_bank":
		// The stop clears ONLY on a complete set of decisions (contract §resumeRun, §BankDecisions).
		// There is no channel that could make one complete yet, so this is the honest answer rather
		// than a placeholder: resuming would hit the same stop and spend the start of a run doing it.
		return pgstore.Run{}, fmt.Errorf("%w: the glossary is not signed", ErrNotResumable)
	default:
		return pgstore.Run{}, fmt.Errorf("%w: it is %s", ErrNotResumable, l.Status)
	}
	next, v, err := s.reopen(ctx, l, fromAFinishedRun)
	if errors.Is(err, pgstore.ErrStopRequested) {
		// Cannot happen through this path today — a resume works on a run that has ENDED and the guard
		// only fires on a live one — but the answer is written rather than left to fall through as an
		// internal error if some later path reaches it.
		return pgstore.Run{}, fmt.Errorf("%w: it has been asked to stop", ErrNotResumable)
	}
	if errors.Is(err, pgstore.ErrNoRun) {
		// Another caller re-opened this run first — two clicks on the same button reach here together,
		// and the second loses the race for attempt N+1 on the unique index. Its answer is the WORK OF
		// THE FIRST: the run is continuing, so reporting "no such run" would be both false and
		// alarming, and re-opening again would take a second hold.
		return s.Store.ReadRun(ctx, userID, runID)
	}
	if err != nil {
		return pgstore.Run{}, err
	}
	switch v {
	case deferred:
		// The previous attempt's money is still being resolved. A second hold now would reserve the
		// ceiling twice, so the answer is "not yet" and the sweep clears it.
		return pgstore.Run{}, fmt.Errorf("%w: its previous attempt is still being settled", ErrNotResumable)
	case exhausted:
		// Nothing left to continue with: the call returns the run in the state it was in, and the state
		// is NOT rewritten on the way out — a run the user stopped stays stopped, because a resume that
		// changed nothing must not change what the run was.
		//
		// ⚠ A BOOK-ceiling pause does NOT reliably land here, and the arithmetic is worth stating
		// because the comment this replaces had it backwards (acceptance dofix ФП-6). The platform
		// sets that ceiling flush against the hold it took (PD-158), but the engine stops on the
		// RESERVATION it cannot take — so it halts with its cap not quite reached, the attempt settles
		// at less than the hold, and the run keeps the difference. `remaining` is therefore positive,
		// usually by less than one call: the resume re-opens, and either that remainder buys a call or
		// the engine meets the same ceiling at once and pauses again. The second case is a loop of one
		// hold, one spawn and one pause per click — no provider money, a `tmctl status` and a
		// transient unit each time — and it is a named residual (register row PD-223) rather than a
		// threshold invented here: what a reservation costs is the ENGINE's number and this side
		// cannot read it. Once the account is granted more, the same call continues the run properly.
		// The pause the platform CANNOT lift — the engine's daily ceiling — is refused above.
		return s.Store.ReadRun(ctx, userID, runID)
	}
	s.log().InfoContext(ctx, "run resumed", "run", runID, "attempt", next.AttemptNo)
	// QUEUED, not spawned here. Starting the engine inline would put two slow things inside a request
	// that answers 202: reading the book's meter costs seconds of the engine's CPU, and creating the
	// unit is a round trip to systemd. The admission path queues for the same reason, and the
	// reconciler is the backstop if the entry is lost.
	if err := s.enqueueNow(ctx, next.RunID); err != nil {
		// The run IS re-opened and the money IS held: the row says `translating` and the reconciler
		// picks up an attempt with no unit on its next pass, exactly as it does after an admission.
		// Reporting a failure here would tell the user nothing had started while the row said
		// otherwise.
		s.log().ErrorContext(ctx, "the resumed run could not be queued; the reconciler will retry",
			"run", runID, "err", err)
	}
	return s.Store.ReadRun(ctx, userID, runID)
}

// attemptSpend turns the engine's figure into what THIS attempt cost.
//
// ⚠ The engine's `committed_usd` is a lifetime total for the BOOK — literally
// `SELECT COALESCE(SUM(committed_usd),0) FROM spend WHERE book_id = ?`
// (backend/internal/store/ledger.go at HEAD, reached through pipeline/status.go) — across every run
// and every day. Settling with it verbatim charges an account for all of the book's earlier runs
// again, every time; the overcharge is bounded by the hold, and the ledger then reads "capped at the
// hold", which looks like an engine overspend rather than the platform's own arithmetic. So each
// attempt records where the meter stood when it started and owes the DIFFERENCE.
//
// Clamped at zero: a total that went DOWN is not a refund, it is a book whose project database was
// replaced, and paying an account for that is not something this code may decide.
func attemptSpend(l pgstore.LiveRun, committed money.MicroUSD) (money.MicroUSD, error) {
	if l.SpendBaseline == nil {
		return 0, errors.New("runs: the attempt has no spend baseline")
	}
	if committed < *l.SpendBaseline {
		return 0, nil
	}
	return committed - *l.SpendBaseline, nil
}