cthulhu/internal/tentacle/process.go

// Copyright (c) 2026 Storm Dragon
// SPDX-License-Identifier: GPL-3.0-or-later

package tentacle

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"sync"
	"syscall"
	"time"
)

func (session *Session) run() {
	var cleanupError error
	starting := true
	ready := func() {
		if starting {
			starting = false
			<-session.manager.starting
			close(session.ready)
		}
	}
	defer func() {
		session.cancel()
		session.mutex.Lock()
		if cleanupError != nil {
			session.fail_locked(cleanupError)
			// Even a requested disable must report incomplete containment.
			session.state, session.failure = Failed, cleanupError
		} else if session.state == Stopping {
			session.state = Stopped
		} else if session.state != Failed {
			session.fail_locked(ErrExited)
		}
		session.mutex.Unlock()
		session.callbacks.Wait()
		// A retained status/session handle must not retain unsent argument text.
		for len(session.outbound) != 0 {
			<-session.outbound
		}
		ready()
		session.manager.mutex.Lock()
		if cleanupError != nil {
			// Keep the admission slot and reject further starts. Do not claim
			// capacity was reclaimed when owned processes may remain alive.
			session.manager.cleanupError = cleanupError
		} else {
			delete(session.manager.sessions, session.spec.ID)
		}
		close(session.done)
		session.manager.mutex.Unlock()
	}()
	select {
	case <-session.stop:
		return
	case <-session.ctx.Done():
		session.fail(session.ctx.Err())
		return
	default:
	}

	// Use owned files instead of exec's copier goroutines. Closing our ends
	// unblocks workers even when a descendant retains an inherited pipe.
	var pipes []*os.File
	defer func() {
		for _, pipe := range pipes {
			pipe.Close()
		}
	}()
	newPipe := func() (*os.File, *os.File, error) {
		reader, writer, err := os.Pipe()
		if err == nil {
			pipes = append(pipes, reader, writer)
		}
		return reader, writer, err
	}
	childInput, input, err := newPipe()
	if err != nil {
		session.fail(err)
		return
	}
	output, childOutput, err := newPipe()
	if err != nil {
		session.fail(err)
		return
	}
	diagnostics, childDiagnostics, err := newPipe()
	if err != nil {
		session.fail(err)
		return
	}
	command := exec.Command(session.spec.Program[0], session.spec.Program[1:]...)
	command.Dir = session.spec.Directory
	command.Stdin, command.Stdout, command.Stderr = childInput, childOutput, childDiagnostics
	command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
	var group *cgroup
	if containment := session.manager.containment; containment != nil {
		group, err = create_cgroup(filepath.Join(containment.group.path, session.identity), containment.limits.PerSession, 0)
		if err != nil {
			session.fail(err)
			return
		}
		defer func() { cleanupError = group.kill_and_remove(containment.limits.CleanupTimeout) }()
		// clone3(CLONE_INTO_CGROUP) applies limits before the child executes.
		// Moving an already-running PID would leave a fork/allocation race.
		command.SysProcAttr.UseCgroupFD = true
		command.SysProcAttr.CgroupFD = int(group.file.Fd())
	}
	var pid int
	var exited <-chan struct{}
	if session.manager.delegation != nil {
		var closeLaunch func()
		pid, exited, closeLaunch, err = session.manager.delegation.launch(session.ctx, session.spec, childInput, childOutput, childDiagnostics, group.file)
		if err == nil {
			defer closeLaunch()
		}
	} else {
		err = command.Start()
		if err == nil {
			pid = command.Process.Pid
			done := make(chan struct{})
			exited = done
			go func() { _ = command.Wait(); close(done) }()
		}
	}
	if err != nil {
		// Do not retain the command line, environment or child-supplied text.
		if group != nil {
			var errno syscall.Errno
			if errors.As(err, &errno) {
				session.fail(fmt.Errorf("%w: child placement/exec: %w", ErrContainment, errno))
			} else {
				session.fail(ErrContainment)
			}
		} else {
			session.fail(ErrExited)
		}
		return
	}
	childInput.Close()
	childOutput.Close()
	childDiagnostics.Close()
	var workers sync.WaitGroup
	workers.Add(3)
	go func() { defer workers.Done(); session.write_requests(input) }()
	go func() { defer workers.Done(); session.read_replies(output) }()
	go func() { defer workers.Done(); session.drain_diagnostics(diagnostics) }()
	defer func() {
		session.cancel()
		if group != nil {
			_ = write_control(group.path, "cgroup.kill", "1")
			if command.Process != nil {
				_ = command.Process.Kill()
			}
		} else {
			// Explicit trusted-fixture mode only; detached descendants escape.
			_ = syscall.Kill(-pid, syscall.SIGKILL)
		}
		input.Close()
		output.Close()
		diagnostics.Close()
		<-exited
		workers.Wait()
	}()

	session.mutex.Lock()
	var initialized <-chan Result
	if session.state == Starting && session.ctx.Err() == nil {
		apis := []string{"commands"}
		if session.spec.Presentation {
			apis = append(apis, "presentation")
		}
		if session.spec.Interaction {
			apis = append(apis, "interaction")
		}
		initialized, err = session.request_locked(session.ctx, "initialize", Initialization{
			Protocol: Version{Major: 1, Minor: 0}, Session: session.identity,
			APIs: apis, Capabilities: apis,
		}, session.manager.limits.StartupTimeout)
	} else {
		err = ErrUnavailable
	}
	session.mutex.Unlock()
	if err != nil {
		session.fail(err)
		return
	}
	select {
	case result := <-initialized:
		if result.Err != nil {
			session.fail(result.Err)
			return
		}
		var registration Registration
		if decode_strict(result.Value, &registration) != nil || validate_registration(registration, session.identity) != nil {
			session.fail(ErrProtocol)
			return
		}
		for _, command := range registration.Commands {
			if command.Presentation && !session.spec.Presentation || command.Interaction && (!command.Presentation || !session.spec.Interaction) {
				session.fail(ErrProtocol)
				return
			}
		}
		session.mutex.Lock()
		if session.state == Starting && session.ctx.Err() == nil {
			session.commands = registration.Commands
			session.state = Ready
		}
		session.mutex.Unlock()
		ready()
	case <-session.ctx.Done():
		session.fail(session.ctx.Err())
		return
	case <-exited:
		session.fail(ErrExited)
		return
	case <-session.stop:
		session.wait_shutdown(exited)
		return
	}
	select {
	case <-session.ctx.Done():
		session.fail(session.ctx.Err())
	case <-exited:
		session.fail(ErrExited)
	case <-session.stop:
		session.wait_shutdown(exited)
	}
}

func (session *Session) wait_shutdown(exited <-chan struct{}) {
	timer := time.NewTimer(session.manager.limits.ShutdownTimeout)
	defer timer.Stop()
	select {
	case <-exited:
	case <-timer.C:
	case <-session.ctx.Done():
	}
}

func (session *Session) write_requests(writer io.Writer) {
	for {
		select {
		case <-session.ctx.Done():
			return
		case item := <-session.outbound:
			if item.id != "" {
				session.mutex.Lock()
				call := session.pending[item.id]
				eligible := call != nil && call.ctx.Err() == nil
				session.mutex.Unlock()
				if !eligible {
					continue
				}
			}
			if _, err := writer.Write(item.frame); err != nil {
				session.fail(ErrExited)
				return
			}
		}
	}
}

func (session *Session) read_replies(reader io.Reader) {
	buffer := bufio.NewReaderSize(reader, 4096)
	for session.ctx.Err() == nil {
		frame, err := read_frame(buffer)
		if err != nil {
			if err == io.EOF {
				err = ErrExited
			}
			session.fail(err)
			return
		}
		reply, err := decode_response(frame)
		if err != nil {
			session.fail(err)
			return
		}
		session.accept_reply(reply)
	}
}

func (session *Session) drain_diagnostics(reader io.Reader) {
	count, err := io.Copy(io.Discard, io.LimitReader(reader, MaxDiagnosticBytes+1))
	if count > MaxDiagnosticBytes {
		session.fail(ErrLimit)
	} else if err != nil && session.ctx.Err() == nil {
		session.fail(ErrExited)
	}
}