Maven/internal/phraser/server.go

// phraser/server.go — the llama-server this phraser talks to: the child
// process it may own, and the startup handshake that waits for the port.
//
// Nothing here knows what a prompt is. Split out of llmphraser.go so the
// phrasing paths and the process lifetime read as two subjects.
package phraser

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
	"regexp"
	"strings"
	"sync"
	"syscall"
	"time"
)

var listenRE = regexp.MustCompile(`listening on (https?://\S+)`)

// backend — one llama-server this phraser talks to. Two implementations: a
// llamaProc we spawned and must reap, and a borrowedBackend someone else owns.
type backend interface {
	BaseURL() string
	Close() error
}

// borrowedBackend — a server started and owned by someone else (the phrasing
// scorer's shared llama-server). Closing it is a no-op by construction.
type borrowedBackend string

func (b borrowedBackend) BaseURL() string { return string(b) }
func (b borrowedBackend) Close() error    { return nil }

// llamaProc — a llama-server child process plus the goroutine reading its
// stderr. Close kills and reaps it; see the Pdeathsig note in spawnLlamaServer.
type llamaProc struct {
	base   string
	cmd    *exec.Cmd
	cancel context.CancelFunc
	wg     sync.WaitGroup
}

func (l *llamaProc) BaseURL() string { return l.base }

func (l *llamaProc) Close() error {
	l.cancel()
	if l.cmd != nil && l.cmd.Process != nil {
		_ = l.cmd.Process.Kill()
		_ = l.cmd.Wait() // reap the process — without Wait, the child becomes a zombie
	}
	l.wg.Wait()
	return nil
}

// lineTail keeps the last few startup lines so a server that dies before it
// listens can say why in the error, not just "EOF". Written by the reader
// goroutine and read by whoever gives up on startup, so it takes a lock.
type lineTail struct {
	mu    sync.Mutex
	lines []string
}

const lineTailMax = 12

func (t *lineTail) add(line string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	t.lines = append(t.lines, line)
	if len(t.lines) > lineTailMax {
		t.lines = t.lines[len(t.lines)-lineTailMax:]
	}
}

func (t *lineTail) String() string {
	t.mu.Lock()
	defer t.mu.Unlock()
	if len(t.lines) == 0 {
		return "(no output)"
	}
	return strings.Join(t.lines, " | ")
}

func extractPort(listen string) string {
	_, port, _ := strings.Cut(listen, ":")
	if port == "" {
		return "0"
	}
	return port
}

// spawnLlamaServer starts one llama-server for cfg and waits until it says which
// address it is listening on. ctx owns the process lifetime, so it must be the
// daemon's context, not a request's.
func spawnLlamaServer(ctx context.Context, cfg Config) (backend, error) {
	ctx, cancel := context.WithCancel(ctx)
	p, err := startLlamaProc(ctx, cfg)
	if err != nil {
		cancel()
		return nil, err
	}
	p.cancel = cancel
	return p, nil
}

// llamaArgs is the command line for one resident server. It is a function and
// not an inline literal because kill-maven.sh's orphan sweep matches against
// this exact line, and a test pins the two together.
func llamaArgs(cfg Config) []string {
	args := []string{
		"-m", cfg.ModelPath,
		"--host", "127.0.0.1",
		"--port", extractPort(cfg.Listen),
		"-c", fmt.Sprintf("%d", cfg.NCtx),
		"-ngl", fmt.Sprintf("%d", cfg.NGpuLayers),
		"--no-webui",
	}
	if cfg.CacheRAMMiB > 0 {
		args = append(args, "--cache-ram", fmt.Sprintf("%d", cfg.CacheRAMMiB))
	}
	return args
}

// defaultStartupTimeout — the wait for llama-server's listen line when Config
// does not set one. A cold model load off disk is the slow part.
const defaultStartupTimeout = 60 * time.Second

func startLlamaProc(ctx context.Context, cfg Config) (*llamaProc, error) {
	startupTimeout := cfg.StartupTimeout
	if startupTimeout <= 0 {
		startupTimeout = defaultStartupTimeout
	}
	p := &llamaProc{}
	cmd := exec.CommandContext(ctx, cfg.BinPath, llamaArgs(cfg)...)
	// Pdeathsig: the kernel SIGKILLs llama-server the moment mavend dies — by
	// ANY means, including SIGKILL/OOM/panic where our Close() never runs. Without
	// it a hard-killed mavend orphans its llama-server (reparented to init, keeps
	// eating GPU/RAM); repeated dev restarts pile up orphans until the box OOMs.
	// Setpgid isolates it in its own process group so a stray Ctrl-C on the
	// terminal group doesn't half-kill it out from under us. (Linux-only, like
	// the rest of the daemon.)
	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true, Pdeathsig: syscall.SIGKILL}
	p.cmd = cmd

	// One pipe for both streams. llama.cpp writes its buffer sizes, KV-cache
	// layout and offload lines to stderr and its request log to stdout, and
	// stdout used to go nowhere at all — so nothing about the model's memory was
	// diagnosable from a running box. Both ends land in mavend's log now.
	pr, pw, err := os.Pipe()
	if err != nil {
		return nil, fmt.Errorf("llm: output pipe: %w", err)
	}
	cmd.Stdout = pw
	cmd.Stderr = pw

	if err := cmd.Start(); err != nil {
		pr.Close()
		pw.Close()
		return nil, fmt.Errorf("llm: start: %w", err)
	}
	// The child holds the only other reference to the write end. Dropping ours
	// is what makes the reader see EOF when the child dies.
	pw.Close()

	portCh := make(chan string, 1)
	errCh := make(chan error, 1)
	tail := &lineTail{}
	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		defer pr.Close()
		sc := bufio.NewScanner(pr)
		// llama.cpp prints one prompt per line and a prompt can be long.
		sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
		listening := false
		for sc.Scan() {
			line := sc.Bytes()
			log.Printf("llama: %s", line)
			if !listening {
				tail.add(string(line))
				if m := listenRE.FindSubmatch(line); len(m) > 1 {
					listening = true
					portCh <- string(m[1])
					close(portCh)
				}
			}
		}
		err := sc.Err()
		if err == nil {
			err = io.EOF
		}
		errCh <- err
	}()

	fail := func(err error) (*llamaProc, error) {
		_ = cmd.Process.Kill()
		_ = cmd.Wait()
		return nil, err
	}
	select {
	case addr := <-portCh:
		p.base = addr
		return p, nil
	case err := <-errCh:
		// The tail is the whole diagnosis when the server dies during load: bare
		// "EOF" never said which layer or which allocation it choked on.
		return fail(fmt.Errorf("llm: server output: %w; last output: %s", err, tail.String()))
	case <-ctx.Done():
		return fail(ctx.Err())
	case <-time.After(startupTimeout):
		return fail(fmt.Errorf("llm: server did not start within %s; last output: %s", startupTimeout, tail.String()))
	}
}