textmachine/platform/internal/runner/engine.go

package runner

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"

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

// ConfigFile is the engine's per-book configuration, which lives in the book's project directory.
// `tmctl` requires --config on every command that touches a book; the platform never edits this
// file (D39.110 §2b), it only names it.
const ConfigFile = "book.yaml"

// CeilingTemplate is how the run's ceiling reaches the engine.
//
// The flag LANDED as `--ceiling-usd <usd>` (unified backlog row 145, engine commit 0e69bc1, ratified
// D39.122), and that form is the configured default. It stays a template because the form belongs to
// the engine's CLI and not to this package: an operator running a build whose flag is spelled
// differently changes one environment variable instead of waiting for a platform release.
//
// ⚠ What the number MEANS is the part worth reading before touching any of this. The flag is not a
// run budget: it overrides the book's own `ceilings.book_usd` and the engine compares it against the
// book's CUMULATIVE committed + reserved on every reservation it takes. Converting the user's
// increment into that absolute is the platform's duty and lives in runs.meter.bookCap.
//
// {{usd}} expands to the amount as a plain decimal number of dollars.
type CeilingTemplate []string

// ErrCeilingNotWired is a run that cannot be started because the engine has no way to be told its
// ceiling. A refusal, deliberately: the alternative is a paid run whose limit is the book's own,
// which is a different and possibly much larger number than the money the account reserved.
//
// Rare since the flag landed: the template has a default, and an override has to be actively made
// useless to reach this — an unset variable takes the default, but one holding only whitespace does
// not, and renders an empty template. Kept because the type admits the empty value at all, and a nil
// template must never mean "no limit".
var ErrCeilingNotWired = errors.New("runner: the engine's ceiling argument is not configured (row 145)")

// ParseCeilingTemplate reads the template from its configured form: argv separated by whitespace,
// with {{usd}} standing for the amount. Empty is legal and means "not wired yet".
func ParseCeilingTemplate(s string) (CeilingTemplate, error) {
	fields := strings.Fields(s)
	if len(fields) == 0 {
		return nil, nil
	}
	if !strings.Contains(s, "{{usd}}") {
		return nil, fmt.Errorf("runner: ceiling template %q does not contain {{usd}}", s)
	}
	return fields, nil
}

// Args renders the template for one amount.
func (t CeilingTemplate) Args(ceiling money.MicroUSD) ([]string, error) {
	if len(t) == 0 {
		return nil, ErrCeilingNotWired
	}
	if ceiling <= 0 {
		// R7 forbids a ledger with no ceiling, and the engine refuses a non-positive one. Catching it
		// here means the refusal names the platform's own arithmetic instead of arriving as an engine
		// start-up error hours later in a log nobody reads.
		return nil, fmt.Errorf("runner: ceiling must be positive, got %d micro-USD", ceiling)
	}
	usd := ceiling.USD()
	out := make([]string, len(t))
	for i, a := range t {
		out[i] = strings.ReplaceAll(a, "{{usd}}", usd)
	}
	return out, nil
}

// TranslateArgs is the engine invocation for a run.
//
// `--config` is not optional and its absence is not tolerated: tmctl requires it on every book
// command, so a call without it fails at argument parsing and never reaches the book.
//
// keysFile is the DEPLOYMENT's provider-key file, passed as `--keys-file` (row 211). It goes on
// `translate` alone because the engine refuses the flag on every other command (D20.4: the $0 read
// verbs must not demand keys), and it goes as an ARGUMENT rather than into the unit's environment:
// keys never pass through this process. Empty means the flag is not passed and the engine falls back
// to its conventional `.env` files — which on the SaaS path nothing writes, so a deployment that
// leaves this unset is one whose runs fail at the first provider call.
//
// maxUnits is the run's VOLUME allowance: no more than that many output units may take a slot of this
// run's grant (backend/internal/pipeline/volume.go). Zero passes no flag and means "no volume bound",
// which is what the WHOLE-BOOK order is.
//
// ⛔ IT IS THE SECOND HALF OF THE ORDER AND IT IS WHAT MAKES THE FIRST HALF TRUE. Until now the only
// bound the engine got was a DOLLAR one, so «buy ten chapters» reached it as a sum, and a sum buys
// whatever it buys: measured on real ledgers, ten chapters' worth of money bought sixteen to
// twenty-six (D39.165 §1, and volume.go opens with the same arithmetic). The knob said chapters and
// meant money. With this flag the seller's promise and the engine's stop are the same quantity, in
// the same unit, on both sides of the seam — the manifest's `units_total` is where this platform's
// per-chapter unit count comes from, so the conversion is exact rather than estimated.
//
// It is computed at every SPAWN from the book's order minus what has been delivered, never frozen at
// admission: a restart of a run that already delivered half of its order must not hand the engine the
// whole of it a second time.
//
// resnapshot and acceptRebill are the re-pass consents (P10, D39.165 §3), decided at ADMISSION and
// stored on the run row — this function only renders what the row says, so a respawned attempt
// carries the same argv. They are the engine's own ORTHOGONAL pair (tmctl: --resnapshot re-pins
// jobs onto the current snapshot, --accept-rebill consents to the re-payment): a run over a moved
// bank needs both, because the snapshot guard and the consent gate refuse independently. The
// consent is always CAPPED — `--accept-rebill=<usd>`, the concrete sum the admission agreed to —
// never the bare blanket form: a projection grown past it (deploy drift on top of the bank's move)
// must refuse, not be bought silently. Zero means no consent and no flag.
func TranslateArgs(workdir string, verifyBank bool, keysFile string, resnapshot bool,
	acceptRebill money.MicroUSD, maxUnits int, ceiling []string) []string {
	args := []string{"translate", "--config", filepath.Join(workdir, ConfigFile)}
	if verifyBank {
		args = append(args, "--verify-bank")
	}
	if keysFile != "" {
		args = append(args, "--keys-file", keysFile)
	}
	if resnapshot {
		args = append(args, "--resnapshot")
	}
	if acceptRebill > 0 {
		args = append(args, "--accept-rebill="+acceptRebill.USD())
	}
	if maxUnits > 0 {
		args = append(args, "--max-units", strconv.Itoa(maxUnits))
	}
	return append(args, ceiling...)
}

// StatusArgs is the reconciliation channel: `tmctl status --json` on a run that is not moving.
//
// Read-only and free of money, but NOT free of CPU — every call re-ingests and re-chunks the source
// (~1.4-1.5 s on a 23 MB book, unified backlog row 100), so it is a repair path and a slow poll,
// never a live feed.
func StatusArgs(workdir string) []string {
	return []string{"status", "--config", filepath.Join(workdir, ConfigFile), "--json"}
}

// ManifestArgs is the intake channel: `tmctl manifest --config <book.yaml> --json`.
//
// $0 and key-less like `status` — it ingests and cuts the source and writes a sidecar, with no
// provider call and no wave — and it is the ONE engine command the platform runs before a book has
// ever been translated. ⚠ Its first touch of a project CREATES and migrates the engine's database
// (ratified D39.122), which is why intake needs the book's directory writable and why it is this
// command rather than `status`: `status` has the same side effect and answers a different question.
func ManifestArgs(workdir string) []string {
	return []string{"manifest", "--config", filepath.Join(workdir, ConfigFile), "--json"}
}

// ExportArgs is the TEXT channel: `tmctl export --config <book.yaml> --json --pairs`.
//
// $0 and key-less like the other two — a pure read projection of the persisted rows joined onto the
// source cut — and the only channel that carries a pair's source and translation at all. `--pairs`
// is what adds the SOURCE column, and it costs the full re-chunk: the manifest alone carries no
// text, so a run without the flag would answer the reading surface with an empty source for every
// pair.
//
// ⚠ Run at the BOUNDARIES of the work only. The engine holds its project exclusively while a run is
// going, and the contract declares exactly this freshness for a pair's translation: updated at the
// boundaries and at stops, never continuously.
func ExportArgs(workdir string) []string {
	return []string{"export", "--config", filepath.Join(workdir, ConfigFile), "--json", "--pairs"}
}

// maxManifest bounds what the platform will read from the manifest command. The document is counts
// and identities — no source text — so a real one is single-digit megabytes even for a 2283-chapter
// book; the cap is against a process that is not the engine, not against the engine.
const maxManifest = 64 << 20

// maxExport bounds the export document, and it is the largest of the three because it is the only
// one that carries TEXT — both columns of a whole book, tens of megabytes on a corpus book
// (PLATFORM_DIRECTION §5б). Still a refusal of a process that is not the engine.
const maxExport = 512 << 20

// Manifest builds the book's chapter tree and returns the allowlisted summary of it.
//
// It reads the SAME rule as Status about exit 2 — the command completed and something in the book
// needs a human — and not because the manifest renderer can produce it: it cannot today (the
// flagged sentinel is built in four places in backend/cmd/tmctl/render.go, belonging to `translate`,
// `status` and `redrive`, and `renderManifest` returns nil for every document it prints). It reads
// it because the alternative is an invariant of another zone that nothing here tests: were a later
// engine to flag something during a cut, intake would spend the book's whole budget and end its
// walk, for every book on the host at once. Applying the rule costs one condition; depending on the
// invariant costs a comment that has to stay true (raised by the seam lens of the dofix review).
func (r *Runner) Manifest(ctx context.Context, binary, workdir string) (ingest.Manifest, error) {
	doc, _, err := readEngine(ctx, binary, workdir, ManifestArgs(workdir), maxManifest, "manifest")
	if err != nil {
		return ingest.Manifest{}, err
	}
	return ingest.DecodeManifest(doc)
}

// Export reads a book's pairs — source, translation and the state derived from them.
//
// It carries the SAME exit-2 rule as Manifest and Status, and here the rule is load-bearing rather
// than defensive: a book with a flagged unit is the ordinary case for this command, and reading a
// non-zero code as "the engine did not answer" would leave exactly those books without a reading
// surface (the shape of ФП-1, which cost the money path of every book that ever flagged a unit).
func (r *Runner) Export(ctx context.Context, binary, workdir string) (ingest.Export, error) {
	doc, _, err := readEngine(ctx, binary, workdir, ExportArgs(workdir), maxExport, "export")
	if err != nil {
		return ingest.Export{}, err
	}
	return ingest.DecodeExport(doc)
}

// readEngine runs one $0 read command and returns its document, plus the exit error when the engine
// completed WITH FLAGS — an answer rather than a refusal, which only `status` has to weigh further.
//
// One function for all three commands, because everything below is the same problem three times: a
// pipe nobody drains blocks the child, a cap at the mercy of the thing it bounds is not a cap, exit
// 2 is an answer, and the engine's stderr must not become a platform log line verbatim.
func readEngine(ctx context.Context, binary, workdir string, args []string, limit int64, what string) ([]byte, error, error) {
	cmd := exec.CommandContext(ctx, binary, args...)
	cmd.Dir = workdir
	var errOut bytes.Buffer
	cmd.Stderr = &errOut
	out, err := cmd.StdoutPipe()
	if err != nil {
		return nil, nil, fmt.Errorf("runner: tmctl %s: %w", what, err)
	}
	if err := cmd.Start(); err != nil {
		return nil, nil, fmt.Errorf("runner: tmctl %s: %w", what, err)
	}
	doc, readErr := io.ReadAll(io.LimitReader(out, limit+1))
	overflowed := drain(cmd, out, int64(len(doc)) > limit)
	// The fact and the first line of stderr, never the whole of it. The error carries an
	// *exec.ExitError underneath, because the caller tells "the engine answered no" from "the engine
	// could not be run" by exactly that type.
	var flagged error
	if err := cmd.Wait(); overflowed == nil && err != nil {
		refused := fmt.Errorf("runner: tmctl %s: %w: %s", what, err, firstLine(errOut.Bytes()))
		if !ingest.CompletedWithFlags(err) {
			return nil, nil, refused
		}
		flagged = refused
	}
	if overflowed != nil {
		return nil, nil, fmt.Errorf("runner: tmctl %s: %w", what, overflowed)
	}
	if readErr != nil {
		return nil, nil, fmt.Errorf("runner: read %s: %w", what, readErr)
	}
	return doc, flagged, nil
}

// maxStatus bounds what the platform will read from the status command, for the same reason and
// against the same thing as maxManifest: a process at the configured path that is not the engine.
// Measured rather than guessed — a real 2283-chapter book's `status --json` is 1.1 MB — so the cap
// is three orders of magnitude above the largest report anyone produces.
const maxStatus = 16 << 20

// Status runs the reconciliation channel and decodes the allowlisted subset of its report.
//
// ⚠ Exit 2 is an ANSWER here, not a refusal: `status --json` prints the whole report and then exits
// 2 for a book with flagged units (ingest.CompletedWithFlags). THREE things have to hold together
// for that answer to be taken — the code, a report that parses, and the report agreeing that the
// book has a flagged unit, which is the only condition under which the engine emits the code
// (backend/cmd/tmctl/render.go `if rep.Flagged > 0`). Anything else is a refusal, as every other
// non-zero code is.
//
// The third check is what the money review of this dofix asked for, and it is not ceremony: this
// call settles a run, so a process at the pinned path that is not that build — a wrapper, a
// version directory replaced in place, a half-finished deploy — printing any JSON object with a
// committed figure and exiting 2 would be CHARGED rather than deferred.
func (r *Runner) Status(ctx context.Context, binary, workdir string) (ingest.StatusReport, error) {
	doc, flagged, err := readEngine(ctx, binary, workdir, StatusArgs(workdir), maxStatus, "status")
	if err != nil {
		return ingest.StatusReport{}, err
	}
	rep, decodeErr := ingest.DecodeStatus(doc)
	if flagged != nil && (decodeErr != nil || rep.Flagged == 0) {
		// It exited 2 and what came back is not the report that code means. The EXIT error is the more
		// informative half — the decoder can only say that the bytes are not a report.
		return ingest.StatusReport{}, flagged
	}
	return rep, decodeErr
}

// drain finishes with the child's stdout once the caller has read what it wanted, and reports the
// overflow when there was one.
//
// The ordinary case is a DRAIN, and that is not tidiness: a pipe nobody reads blocks the child on its
// next write and Wait then blocks on the child. Closing it instead kills the engine with EPIPE, which
// this side would then report as a host that cannot run the engine.
//
// Over the cap the process is KILLED instead, and that half was found by writing the test for the
// cap: a writer that never stops is exactly what the cap exists for, and draining it never returns —
// the caller hangs until its own deadline, which on the intake path is fifteen minutes of a worker.
// A cap that is at the mercy of the thing it bounds is not a cap.
func drain(cmd *exec.Cmd, out io.Reader, over bool) error {
	if !over {
		_, _ = io.Copy(io.Discard, out)
		return nil
	}
	if cmd.Process != nil {
		_ = cmd.Process.Kill()
	}
	_, _ = io.Copy(io.Discard, out) // returns at once now: the writer is gone
	return errors.New("the engine printed more than this platform will read, which no book produces")
}

func firstLine(b []byte) []byte {
	if i := bytes.IndexByte(b, '\n'); i >= 0 {
		return b[:i]
	}
	return b
}