//go:build linux

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/bnema/vev/pkg/rawterm"
	"github.com/bnema/vev/pkg/safedir"
)

type launchedProcess interface {
	Warmup([]byte) error
	Measure([]byte, func() error, func() error) error
	Close() error
}

// roleCommand contains only argv accepted by app.parseArgs. Transport routing
// metadata configures the harness's public-command seam; it is never passed as
// an unrecognised vev flag.
type roleCommand struct {
	Args      []string
	Transport transport
}

type launcher interface {
	Launch(processMapping, roleCommand) (launchedProcess, error)
}

type launchedRole struct {
	mapping processMapping
	process launchedProcess
}

// closeLaunchedRoles closes every role in reverse launch order. errors.Join
// retains every cleanup failure in that stable order.
func closeLaunchedRoles(processes []launchedRole) error {
	var failures []error
	for i := len(processes) - 1; i >= 0; i-- {
		if err := processes[i].process.Close(); err != nil {
			failures = append(failures, fmt.Errorf("close %s role: %w", processes[i].mapping.Role, err))
		}
	}
	return errors.Join(failures...)
}

// cliLauncher is deliberately limited to executing the supplied public vev
// binary. Scenario orchestration never reaches into an in-process daemon.
type cliLauncher struct {
	bin      string
	mu       sync.Mutex
	peers    map[string]peerRoute
	runtimes map[string]string
}

type peerRoute struct {
	mapping    processMapping
	command    roleCommand
	runtimeDir string
	stateDir   string
}

type preparedPeer struct {
	route          peerRoute
	cleanupRuntime func() error
}

func (p preparedPeer) Warmup([]byte) error { return nil }

func (p preparedPeer) Measure([]byte, func() error, func() error) error { return nil }

func (p preparedPeer) Close() error {
	if p.cleanupRuntime != nil {
		return p.cleanupRuntime()
	}
	return nil
}

type terminalOutput interface {
	io.Writer
	Close() error
}

type capturedTerminalOutput struct {
	file *os.File
	mu   sync.Mutex
	data bytes.Buffer
}

type synchronizedBuffer struct {
	mu   sync.Mutex
	data bytes.Buffer
}

func (b *synchronizedBuffer) Write(p []byte) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.data.Write(p)
}

func (b *synchronizedBuffer) WriteString(value string) {
	_, _ = b.Write([]byte(value))
}

func (b *synchronizedBuffer) String() string {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.data.String()
}

func (o *capturedTerminalOutput) Write(p []byte) (int, error) {
	o.mu.Lock()
	_, _ = o.data.Write(p)
	o.mu.Unlock()
	return o.file.Write(p)
}
func (o *capturedTerminalOutput) Close() error { return o.file.Close() }
func (o *capturedTerminalOutput) String() string {
	o.mu.Lock()
	defer o.mu.Unlock()
	return o.data.String()
}

var setPTYWinsize = rawterm.SetWinsize

type cliProcess struct {
	cmd              *exec.Cmd
	pty              io.ReadWriteCloser
	output           terminalOutput
	chunks           chan []byte
	done             chan struct{}
	closed           sync.Once
	waitErr          chan error
	waitTimeout      func() <-chan time.Time
	gracefulShutdown func()
	forceCleanup     func()
	forceKill        func()
	shutdown         func() error
	cleanupRuntime   func() error
	terminalReady    bool
	readyCommand     *exec.Cmd
	startCommand     *exec.Cmd
	configureCommand *exec.Cmd
	serverOutput     *synchronizedBuffer
	peerCommand      *exec.Cmd
	peerDone         chan error
	closeResult      error
}

func (p *cliProcess) Resize(cols, rows uint16) error {
	fd, ok := p.pty.(interface{ Fd() uintptr })
	if !ok {
		return errors.New("client PTY does not expose a file descriptor")
	}
	return setPTYWinsize(int(fd.Fd()), cols, rows)
}

func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {
	if len(role.Args) == 0 {
		return nil, errors.New("public CLI role has no arguments")
	}
	bin, err := filepath.Abs(l.bin)
	if err != nil {
		return nil, err
	}
	if info, err := os.Stat(bin); err != nil {
		return nil, err
	} else if info.IsDir() {
		return nil, fmt.Errorf("public CLI binary %q is a directory", bin)
	}
	if m.Role == "ssh_stdio_peer" {
		return l.preparePeer(m, role)
	}
	cmd := exec.Command(bin, role.Args...)
	runDir := filepath.Dir(m.TracePath)
	runtimeDir, err := l.runtimeDir(runDir)
	if err != nil {
		return nil, err
	}
	// All role inputs are absolute paths resolved at harness startup. Do not set
	// a role working directory: daemon-owned subprocess cleanup may otherwise
	// treat the evidence directory as its working tree and remove a preallocated
	// trace while a later repetition is being merged.
	cmd.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN", "VEV_PERF_BIN", "VEV_REMOTE_TRANSPORT", "XDG_RUNTIME_DIR", "XDG_STATE_HOME", "XDG_CONFIG_HOME"), traceEnvironment(m)...)
	cmd.Env = append(cmd.Env, "XDG_RUNTIME_DIR="+runtimeDir,
		"XDG_STATE_HOME="+filepath.Join(runDir, "state"), "XDG_CONFIG_HOME="+filepath.Join(runDir, "config"), "TERM=xterm-256color", "SHELL=/bin/sh", "VEV_PERF_BIN="+bin)
	l.mu.Lock()
	peer, routed := l.peers[runDir]
	l.mu.Unlock()
	if routed && (m.Role == "client" || m.Role == "daemon") {
		mode, err := remoteMode(peer.command.Transport)
		if err != nil {
			return nil, err
		}
		cmd.Env = append(withoutEnv(cmd.Env, "PATH", "VEV_REMOTE_TRANSPORT"), "PATH="+runDir+":"+os.Getenv("PATH"), "VEV_REMOTE_TRANSPORT="+mode)
	}
	if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
		return nil, err
	}
	if m.Role == "daemon" {
		configDir := filepath.Join(runDir, "config", "vev")
		if err := safedir.EnsurePrivate(configDir); err != nil {
			return nil, err
		}
		if err := os.WriteFile(filepath.Join(configDir, "broker.json"), []byte(`{"marker":"vev.broker.offline/v1","registrations":[]}`), 0o600); err != nil {
			return nil, err
		}
	}
	p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1), serverOutput: &synchronizedBuffer{}}
	if m.Role == "daemon" {
		// ls is read-only and cannot start an absent daemon. Start the daemon
		// explicitly before requiring its observed local catalogue.
		p.startCommand = exec.Command(bin, "--daemon-launcher")
		p.startCommand.Env = cmd.Env
		p.readyCommand = exec.Command(bin, "_broker-ready", "--require", "local-catalogue", "--timeout", "10s")
		p.readyCommand.Env = cmd.Env
		l.mu.Lock()
		peer, routed := l.peers[runDir]
		l.mu.Unlock()
		if routed {
			mode, modeErr := remoteMode(peer.command.Transport)
			if modeErr != nil {
				return nil, modeErr
			}
			p.configureCommand = exec.Command(bin, "host", "add", "harness@127.0.0.1")
			p.configureCommand.Env = append(withoutEnv(cmd.Env, "PATH", "VEV_REMOTE_TRANSPORT"), "PATH="+runDir+":"+os.Getenv("PATH"), "VEV_REMOTE_TRANSPORT="+mode)
			peerStart := exec.Command(bin, "--daemon")
			peerStart.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN", "XDG_RUNTIME_DIR", "XDG_STATE_HOME", "PATH"),
				"PATH="+runDir+":"+os.Getenv("PATH"),
				"XDG_RUNTIME_DIR="+peer.runtimeDir,
				"XDG_STATE_HOME="+peer.stateDir,
				"TERM=xterm-256color")
			peerStartOutput := &bytes.Buffer{}
			peerStart.Stdin, peerStart.Stdout, peerStart.Stderr = nil, peerStartOutput, peerStartOutput
			if err := peerStart.Start(); err != nil {
				return nil, fmt.Errorf("start remote peer daemon: %w", err)
			}
			peerSocket := filepath.Join(peer.runtimeDir, "vev", "daemonmux.sock")
			deadline := time.Now().Add(10 * time.Second)
			for {
				if info, err := os.Stat(peerSocket); err == nil && info.Mode()&os.ModeSocket != 0 {
					break
				}
				if time.Now().After(deadline) {
					return nil, fmt.Errorf("remote peer daemon did not publish %s (output=%q)", peerSocket, peerStartOutput.String())
				}
				time.Sleep(10 * time.Millisecond)
			}
			p.peerCommand = peerStart
			p.peerDone = make(chan error, 1)
			go func() {
				err := peerStart.Wait()
				if err != nil {
					p.serverOutput.WriteString(" remote-peer: " + err.Error() + " " + peerStartOutput.String())
				}
				p.peerDone <- err
			}()
		}
	}
	if m.Role == "daemon" {
		p.cleanupRuntime = func() error { return l.releaseRuntime(runDir) }
	}
	if m.Role == "client" {
		master, slave, err := openPTY()
		if err != nil {
			return nil, err
		}
		if err := setPTYWinsize(int(master.Fd()), 120, 40); err != nil {
			_ = master.Close()
			_ = slave.Close()
			return nil, fmt.Errorf("set canonical client PTY geometry: %w", err)
		}
		output, err := os.OpenFile(filepath.Join(filepath.Dir(m.TracePath), "terminal-output"), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
		if err != nil {
			_ = master.Close()
			_ = slave.Close()
			return nil, err
		}
		cmd.Stdin, cmd.Stdout, cmd.Stderr = slave, slave, slave
		// Ctty is the child descriptor number after Start remaps slave to stdin.
		cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true, Setctty: true, Ctty: 0}
		if err := cmd.Start(); err != nil {
			_ = output.Close()
			_ = master.Close()
			_ = slave.Close()
			if p.cleanupRuntime != nil {
				_ = p.cleanupRuntime()
			}
			return nil, err
		}
		_ = slave.Close()
		p.pty, p.output, p.chunks, p.done = master, &capturedTerminalOutput{file: output}, make(chan []byte, 32), make(chan struct{})
		// Stop workload generation through the attached shell before any signal.
		// A normal shell exit makes the daemon detach the client; the client then
		// closes and waits for its ssh transport, allowing a traced broker mux stdio
		// descendant to flush end marks and exit. Process-group signals are reserved
		// for timeout escalation because they would kill the descendant concurrently.
		p.gracefulShutdown = func() { _, _ = master.Write([]byte("exit\n")) }
		go p.copyTerminal()
	} else {
		// Every launched role owns a dedicated process group, so the bounded
		// cleanup path never signals the harness/test runner's group.
		cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
		devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
		if err != nil {
			return nil, err
		}
		cmd.Stdin, cmd.Stdout, cmd.Stderr = devNull, p.serverOutput, p.serverOutput
		if err := cmd.Start(); err != nil {
			_ = devNull.Close()
			if p.cleanupRuntime != nil {
				_ = p.cleanupRuntime()
			}
			return nil, err
		}
		_ = devNull.Close()
	}
	if m.Role == "daemon" {
		// Ask the daemon owned by the foreground broker to stop before the
		// broker's bounded process-group fallback. This lets blocked carriage
		// operations serialize their failed end marks before either process exits.
		p.shutdown = func() error {
			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			defer cancel()
			shutdown := exec.CommandContext(ctx, bin, "_daemon-stop")
			shutdown.Env = withoutEnv(cmd.Env, "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN")
			return shutdown.Run()
		}
	}
	go func() { p.waitErr <- cmd.Wait() }()
	return p, nil
}

// runtimeDir returns one short, private runtime directory shared by the roles
// of one run. Keeping the Unix socket outside deeply nested evidence output
// avoids the platform socket-path limit without weakening run isolation.
func (l *cliLauncher) runtimeDir(runDir string) (string, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if path := l.runtimes[runDir]; path != "" {
		return path, nil
	}
	path, err := os.MkdirTemp("", "vev-harness-runtime-")
	if err != nil {
		return "", err
	}
	if err := safedir.EnsurePrivate(path); err != nil {
		_ = os.RemoveAll(path)
		return "", err
	}
	if l.runtimes == nil {
		l.runtimes = make(map[string]string)
	}
	l.runtimes[runDir] = path
	return path, nil
}

func (l *cliLauncher) releaseRuntime(runDir string) error {
	l.mu.Lock()
	path := l.runtimes[runDir]
	delete(l.runtimes, runDir)
	l.mu.Unlock()
	if path == "" {
		return nil
	}
	return os.RemoveAll(path)
}

// preparePeer installs a per-run ssh command seam. The ordinary public client
// still executes its documented remote attach command; its ssh child invokes
// only the broker mux stdio helper. This makes the peer
// carrying client traffic the declared, exclusively traced role rather than a
// separately launched but unused process.
func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedProcess, error) {
	runDir := filepath.Dir(m.TracePath)
	l.mu.Lock()
	runtimeDir := l.runtimes[runDir]
	l.mu.Unlock()
	peerOwnsRuntime := runtimeDir == ""
	fail := func(err error) error {
		if peerOwnsRuntime {
			return errors.Join(err, l.releaseRuntime(runDir))
		}
		return err
	}
	if peerOwnsRuntime {
		var err error
		runtimeDir, err = l.runtimeDir(runDir)
		if err != nil {
			return nil, fail(err)
		}
	}
	if err := safedir.EnsurePrivate(runDir); err != nil {
		return nil, fail(err)
	}
	if err := safedir.EnsurePrivate(runtimeDir); err != nil {
		return nil, fail(err)
	}
	if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
		return nil, fail(err)
	}
	if len(role.Args) != 2 || role.Args[0] != "_broker-mux-stdio" || role.Args[1] != "--production" {
		return nil, fail(fmt.Errorf("unsupported broker mux peer command %q", role.Args))
	}
	route := peerRoute{mapping: m, command: role, runtimeDir: filepath.Join(runDir, "peer-runtime"), stateDir: filepath.Join(runDir, "peer-state")}
	if err := safedir.EnsurePrivate(route.runtimeDir); err != nil {
		return nil, fail(err)
	}
	if err := safedir.EnsurePrivate(route.stateDir); err != nil {
		return nil, fail(err)
	}
	shim := filepath.Join(runDir, "ssh")
	// The SSH command seam forwards the broker mux stdio helper with the
	// peer's unique trace identity, so the peer carrying client traffic is the
	// declared, exclusively traced role.
	body := fmt.Sprintf(`#!/bin/sh
set -eu
case "$*" in
  *"_broker-mux-stdio"*)
    exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[6]q VEV_PERF_RUN=%[7]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _broker-mux-stdio --production --daemon-start existing-only
    ;;
  *) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;;
esac
`, m.TracePath, m.ProcessID, runtimeDir, filepath.Join(runDir, "state"), l.bin, m.Scenario, m.Run)
	if err := os.WriteFile(shim, []byte(body), 0o700); err != nil {
		return nil, fail(err)
	}
	l.mu.Lock()
	if l.peers == nil {
		l.peers = make(map[string]peerRoute)
	}
	if _, exists := l.peers[runDir]; exists {
		l.mu.Unlock()
		return nil, fail(fmt.Errorf("duplicate transport peer for %s", runDir))
	}
	l.peers[runDir] = route
	l.mu.Unlock()
	peer := preparedPeer{route: route}
	if peerOwnsRuntime {
		peer.cleanupRuntime = func() error { return l.releaseRuntime(runDir) }
	}
	return peer, nil
}

func remoteMode(t transport) (string, error) {
	switch t.Kind {
	case "ssh_stdio":
		return "stdio", nil
	default:
		return "", fmt.Errorf("transport %q cannot route through a remote peer", t.ID)
	}
}

// traceEnvironment is the complete process-local identity supplied by the
// persisted harness processMapping. App turns Scenario/Run into production
// mark correlation; sequence/request/epoch remain process-local operation IDs.
func traceEnvironment(m processMapping) []string {
	return []string{
		"VEV_PERF_TRACE=" + m.TracePath,
		"VEV_PERF_PROCESS_ID=" + m.ProcessID,
		"VEV_PERF_SCENARIO=" + m.Scenario,
		"VEV_PERF_RUN=" + strconv.Itoa(m.Run),
	}
}

func withoutEnv(env []string, names ...string) []string {
	remove := make(map[string]bool, len(names))
	for _, name := range names {
		remove[name] = true
	}
	out := make([]string, 0, len(env))
	for _, entry := range env {
		name, _, _ := strings.Cut(entry, "=")
		if !remove[name] {
			out = append(out, entry)
		}
	}
	return out
}

func (p *cliProcess) copyTerminal() {
	defer close(p.chunks)
	buf := make([]byte, 4096)
	for {
		n, err := p.pty.Read(buf)
		if n > 0 {
			chunk := append([]byte(nil), buf[:n]...)
			if _, writeErr := p.output.Write(chunk); writeErr != nil {
				return
			}
			select {
			case p.chunks <- chunk:
			case <-p.done:
				return
			}
		}
		if err != nil {
			return
		}
	}
}

// WaitReady executes the production dial-only readiness contract under the
// exact XDG environment shared by the servers and clients. Its bounded JSON
// stdout, stderr and owner-process death are retained in any failure.
func (p *cliProcess) WaitReady() error {
	if p.startCommand != nil {
		// Starting the daemon is separate from readiness: only the semantic
		// catalogue probe below can establish that it is serving.
		if output, err := p.startCommand.CombinedOutput(); err != nil {
			return fmt.Errorf("start daemon for broker readiness: %w (output=%q)", err, output)
		}
	}
	if p.readyCommand == nil {
		return errors.New("broker readiness command is empty")
	}
	var stdout, stderr bytes.Buffer
	p.readyCommand.Stdout, p.readyCommand.Stderr = &stdout, &stderr
	probeDone := make(chan error, 1)
	if err := p.readyCommand.Start(); err != nil {
		return fmt.Errorf("start broker readiness probe: %w", err)
	}
	go func() { probeDone <- p.readyCommand.Wait() }()
	select {
	case err := <-probeDone:
		if err != nil {
			return fmt.Errorf("broker readiness probe: %w (stdout=%q stderr=%q)", err, stdout.String(), stderr.String())
		}
		var report struct {
			Status string `json:"status"`
		}
		if err := json.Unmarshal(bytes.TrimSpace(stdout.Bytes()), &report); err != nil || report.Status != "ready" {
			return fmt.Errorf("broker readiness invalid JSON/status: %v (stdout=%q stderr=%q)", err, stdout.String(), stderr.String())
		}
		if p.configureCommand != nil {
			if output, err := p.configureCommand.CombinedOutput(); err != nil {
				return fmt.Errorf("configure remote broker membership: %w (output=%q)", err, output)
			}
		}
		return nil
	case err := <-p.waitErr:
		p.waitErr <- err
		_ = p.readyCommand.Process.Kill()
		probeErr := <-probeDone
		return fmt.Errorf("production server group died before readiness: %v (probe=%v stdout=%q stderr=%q servers=%q)", err, probeErr, stdout.String(), stderr.String(), p.serverOutput.String())
	}
}

func (p *cliProcess) Warmup(input []byte) error {
	if p.pty == nil || len(input) == 0 {
		return nil
	}
	// A client cannot write application output until it has received Welcome and
	// entered raw mode. Require that observable output before submitting any
	// warmup input, so the PTY's line discipline cannot echo it locally.
	if !p.terminalReady {
		if err := p.waitForTerminalReady(); err != nil {
			return err
		}
		p.terminalReady = true
	}
	if _, err := p.pty.Write(input); err != nil {
		return err
	}
	// Do not begin measured input until the public client has rendered a real
	// shell response to warmup input. This is intentionally not a measurement.
	return p.waitForTerminalMarker(inputMarker(input), input)
}

// Measure records a boundary after copyTerminal has read the marker from the
// real client PTY and the harness-owned output writer has accepted that chunk.
// This models successful terminal-byte delivery without adding filesystem
// durability latency. The marker is never fabricated by the harness, and PTY
// local echo is discarded so it cannot satisfy the boundary.
func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error {
	if p.pty == nil {
		return errors.New("measured process has no terminal PTY")
	}
	if _, err := p.pty.Write(input); err != nil {
		return err
	}
	// The injected callback runs only after the owned PTY accepts the bytes.
	if err := injected(); err != nil {
		return err
	}
	if err := p.waitForTerminalMarker(inputMarker(input), input); err != nil {
		return err
	}
	return flushed()
}

// waitForTerminalReady waits for public client output produced before any
// harness input. The client enters raw mode before its output loop can render
// this initial prompt/state, so this is an observable readiness boundary.
func (p *cliProcess) waitForTerminalReady() error {
	timer := time.NewTimer(10 * time.Second)
	defer timer.Stop()
	var observed []byte
	for {
		select {
		case chunk, ok := <-p.chunks:
			if !ok {
				return p.terminalClosedError("readiness output")
			}
			observed = append(observed, chunk...)
			if bytes.Contains(observed, []byte("$")) || bytes.Contains(observed, []byte("#")) {
				return nil
			}
			if len(observed) > 64*1024 {
				observed = append([]byte(nil), observed[len(observed)-32*1024:]...)
			}
		case err := <-p.waitErr:
			// Keep the exit result available to Close, which owns reaping.
			p.waitErr <- err
			if err == nil {
				return errors.New("client exited before readiness output")
			}
			return fmt.Errorf("client exited before readiness output: %w", err)
		case <-timer.C:
			if captured, ok := p.output.(*capturedTerminalOutput); ok {
				return fmt.Errorf("timed out waiting for client readiness output (terminal=%q)", tailString(captured.String(), 4096))
			}
			return errors.New("timed out waiting for client readiness output")
		}
	}
}

func tailString(value string, limit int) string {
	if len(value) <= limit {
		return value
	}
	return value[len(value)-limit:]
}

func (p *cliProcess) terminalClosedError(boundary string) error {
	if p.waitErr != nil {
		select {
		case err := <-p.waitErr:
			// Close owns reaping and may need the result after a failed boundary.
			p.waitErr <- err
			if err != nil {
				return fmt.Errorf("client terminal closed before %s: %w", boundary, err)
			}
		case <-time.After(10 * time.Millisecond):
		}
	}
	return fmt.Errorf("client terminal closed before %s", boundary)
}

func (p *cliProcess) waitForTerminalMarker(marker, input []byte) error {
	var received []byte
	echo := newPTYLocalEcho(input)
	timer := time.NewTimer(10 * time.Second)
	defer timer.Stop()
	for !bytes.Contains(received, marker) {
		select {
		case chunk, ok := <-p.chunks:
			if !ok {
				return p.terminalClosedError("application output")
			}
			received = append(received, echo.applicationOutput(chunk)...)
		case err := <-p.waitErr:
			// Keep the exit result available to Close, which owns reaping.
			p.waitErr <- err
			if err == nil {
				return errors.New("client exited before application output")
			}
			return fmt.Errorf("client exited before application output: %w", err)
		case <-timer.C:
			if captured, ok := p.output.(*capturedTerminalOutput); ok {
				return fmt.Errorf("timed out waiting for application terminal output (terminal=%q)", captured.String())
			}
			return errors.New("timed out waiting for application terminal output")
		}
	}
	return nil
}

// ptyLocalEcho removes the one ordered input echo that a not-yet-raw slave PTY
// can produce. It normalizes NL to the conventional ONLCR echo spelling; any
// divergence is retained as application output. The line discipline emits the
// echo before the child can read and respond to the same write.
type ptyLocalEcho struct {
	expected []byte
	matched  int
	done     bool
}

func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte {
	if e.done || len(e.expected) == 0 {
		return chunk
	}
	for i, b := range chunk {
		if b != e.expected[e.matched] {
			e.done = true
			return append([]byte(nil), chunk[i:]...)
		}
		e.matched++
		if e.matched == len(e.expected) {
			e.done = true
			return append([]byte(nil), chunk[i+1:]...)
		}
	}
	return nil
}

// Close tears down a launched role along a bounded escalation ladder. Each rung
// performs one action, then waits a single closeDeadline for the process to be
// reaped before the next rung escalates. The graceful rungs — the public daemon
// stop and a client shell exit — let a role close its blocked transport
// operations and drain descendants' end marks. Process-group SIGTERM is reserved
// for timeout escalation; only a SIGKILL rung fails teardown. Every path closes
// the PTY exactly once and finalizes output and runtime state.
func (p *cliProcess) Close() error {
	p.closed.Do(func() {
		// 1. Drain terminal output for the whole teardown. Without an active
		// consumer a full chunks channel stalls copyTerminal, which then stops
		// reading the PTY master and blocks the client mid-restore before it can
		// close its observed receive operation. The drain stops with p.done.
		drainDone := p.drainTerminalOutput()

		exited := false
		// waitForExit blocks up to one bounded deadline for the reap.
		waitForExit := func() {
			if p.waitErr == nil || exited {
				return
			}
			select {
			case <-p.waitErr:
				exited = true
			case <-p.closeDeadline():
			}
		}
		// reaped reports whether the process is already gone; a role with no
		// process to reap needs no escalation at all.
		reaped := func() bool {
			if p.waitErr == nil || exited {
				return true
			}
			select {
			case <-p.waitErr:
				exited = true
				return true
			default:
				return false
			}
		}
		ptyClosed := false
		closePTY := func() {
			if p.pty != nil && !ptyClosed {
				_ = p.pty.Close()
				ptyClosed = true
			}
		}

		// 2. Daemon role: ask the public daemon to stop before any signal so its
		// blocked carriage can close and serialize failed end marks.
		if p.shutdown != nil && !reaped() {
			_ = p.shutdown()
			waitForExit()
		}
		// 3. Client role: stop the shell normally. Session teardown detaches the
		// client, which closes and waits for ssh/broker mux stdio before it exits.
		if p.gracefulShutdown != nil && !reaped() {
			p.gracefulShutdown()
			waitForExit()
		}
		// 4. Escalate a role that ignored graceful shutdown to process-group
		// SIGTERM. This remains bounded and cleans up hung descendants.
		if !reaped() {
			p.forceProcessGroupCleanup()
			waitForExit()
		}
		// 5. SIGKILL the role's process group. This severs the process; an
		// in-flight observed operation cannot be closed, so its trace may truncate.
		killed := false
		if !reaped() {
			killed = true
			p.forceProcessGroupKill()
			waitForExit()
		}
		// 6. A SIGKILL escalation cannot guarantee a complete trace. Fail teardown
		// with a clear cause rather than letting the truncation surface later as a
		// downstream merge mystery. The graceful rungs never error on their own.
		if killed {
			if exited {
				p.closeResult = errors.New("public CLI process did not exit after graceful shutdown and SIGTERM; killed")
			} else {
				p.closeResult = errors.New("timed out reaping public CLI process")
			}
		}

		// Close the PTY exactly once on every path: an early graceful exit skips
		// the hangup rung, so close it here before the drain stops.
		closePTY()
		if p.done != nil {
			close(p.done)
		}
		if drainDone != nil {
			<-drainDone
		}
		if p.peerCommand != nil && p.peerCommand.Process != nil {
			_ = p.peerCommand.Process.Signal(syscall.SIGTERM)
			if p.peerDone != nil {
				select {
				case <-p.peerDone:
				case <-time.After(2 * time.Second):
					_ = p.peerCommand.Process.Kill()
					<-p.peerDone
				}
			}
		}
		if p.output != nil {
			if err := p.output.Close(); p.closeResult == nil {
				p.closeResult = err
			}
		}
		if p.cleanupRuntime != nil {
			if err := p.cleanupRuntime(); p.closeResult == nil {
				p.closeResult = err
			}
		}
	})
	return p.closeResult
}

// drainTerminalOutput consumes copyTerminal's chunks for the duration of a
// teardown so a full channel can never wedge copyTerminal (and thus the client's
// PTY). It returns nil when there is no terminal to drain, and otherwise a
// channel closed once the consumer has stopped on p.done.
func (p *cliProcess) drainTerminalOutput() <-chan struct{} {
	if p.chunks == nil || p.done == nil {
		return nil
	}
	done := make(chan struct{})
	chunks := p.chunks
	go func() {
		defer close(done)
		for {
			select {
			case _, ok := <-chunks:
				if !ok {
					// copyTerminal closed the channel; a closed channel is always
					// ready, so stop selecting it and wait only for the stop signal.
					chunks = nil
				}
			case <-p.done:
				return
			}
		}
	}()
	return done
}

func (p *cliProcess) closeDeadline() <-chan time.Time {
	if p.waitTimeout != nil {
		return p.waitTimeout()
	}
	if p.shutdown != nil {
		// A public kill asks the daemon to finish bounded detached notices and
		// close blocked carriage before it exits; do not turn that graceful
		// protocol into a SIGKILL at the two-second client deadline.
		return time.After(5 * time.Second)
	}
	return time.After(2 * time.Second)
}

func (p *cliProcess) forceProcessGroupCleanup() {
	if p.forceCleanup != nil {
		p.forceCleanup()
		return
	}
	if p.cmd != nil && p.cmd.Process != nil {
		// The client is a session/process-group leader; forced cleanup must
		// include its ssh seam/broker mux stdio descendants.
		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
	}
}

func (p *cliProcess) forceProcessGroupKill() {
	if p.forceKill != nil {
		p.forceKill()
		return
	}
	if p.cmd != nil && p.cmd.Process != nil {
		_ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL)
	}
}

// openPTY is the small Linux PTY boundary needed to drive the public terminal
// client. No daemon or adapter implementation is imported by this command.
func openPTY() (*os.File, *os.File, error) {
	// O_CLOEXEC keeps these harness-owned descriptors out of the launched client:
	// exec dup2s the slave onto the child's stdio, but an inherited master (or a
	// second slave) would otherwise leak in and hold the terminal open, so closing
	// the harness master could never hang up the client. cmd.Start still remaps the
	// slave onto fds 0-2, which are exempt from close-on-exec.
	masterFD, err := syscall.Open("/dev/ptmx", syscall.O_RDWR|syscall.O_NOCTTY|syscall.O_CLOEXEC, 0)
	if err != nil {
		return nil, nil, err
	}
	closeMaster := func(err error) (*os.File, *os.File, error) { _ = syscall.Close(masterFD); return nil, nil, err }
	number, err := rawterm.PtsNumber(masterFD)
	if err != nil {
		return closeMaster(err)
	}
	if err := rawterm.UnlockPt(masterFD); err != nil {
		return closeMaster(err)
	}
	slaveFD, err := syscall.Open(fmt.Sprintf("/dev/pts/%d", number), syscall.O_RDWR|syscall.O_NOCTTY|syscall.O_CLOEXEC, 0)
	if err != nil {
		return closeMaster(err)
	}
	return os.NewFile(uintptr(masterFD), "vev-client-master"), os.NewFile(uintptr(slaveFD), "vev-client-slave"), nil
}

func inputMarker(input []byte) []byte {
	start := bytes.Index(input, []byte("__VEV_HARNESS_"))
	if start < 0 {
		return input
	}
	end := bytes.Index(input[start:], []byte(`\n`))
	if end < 0 {
		return input[start:]
	}
	return input[start : start+end]
}