cthulhu/internal/tentacle/process.go

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

package tentacle

import (
	"bufio"
	"encoding/json"
	"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:]...)
	supervised := len(session.manager.supervisorProgram) != 0
	var lifetime, childLifetime, configuration, childConfiguration *os.File
	if supervised {
		childLifetime, lifetime, err = newPipe()
		if err != nil {
			session.fail(ErrSupervision)
			return
		}
		childConfiguration, configuration, err = newPipe()
		if err != nil {
			session.fail(ErrSupervision)
			return
		}
		program := session.manager.supervisorProgram
		command = exec.Command(program[0], program[1:]...)
		command.ExtraFiles = []*os.File{childLifetime, childConfiguration}
	} else {
		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 waitError error
	exited := make(chan struct{})
	err = command.Start()
	if err == nil {
		go func() { waitError = command.Wait(); close(exited) }()
	}

	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
	if supervised {
		childLifetime.Close()
		childConfiguration.Close()
		workers.Add(1)
		go func() {
			defer workers.Done()
			_ = json.NewEncoder(configuration).Encode(session.spec)
			configuration.Close()
		}()
	}
	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 supervised {
			lifetime.Close()
			configuration.Close()
		}
		if group != nil {
			_ = write_control(group.path, "cgroup.kill", "1")
			if command.Process != nil {
				_ = command.Process.Kill()
			}
		} else if !supervised {
			// Explicit trusted-fixture mode only; detached descendants escape.
			_ = syscall.Kill(-command.Process.Pid, syscall.SIGKILL)
		}
		input.Close()
		output.Close()
		diagnostics.Close()
		if supervised && group == nil {
			select {
			case <-exited:
				if waitError != nil {
					cleanupError = ErrSupervision
				}
			case <-time.After(DefaultResourceLimits().CleanupTimeout + time.Second):
				cleanupError = ErrSupervision
				_ = command.Process.Kill()
				<-exited
			}
		} else {
			<-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")
		}
		if session.spec.ReaderSettings {
			apis = append(apis, "reader-settings")
		}
		initialized, err = session.request_locked(session.ctx, "initialize", Initialization{
			Protocol: session.spec.protocol(), Session: session.identity,
			APIs: session.spec.apis(apis), Capabilities: session.spec.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_version(registration, session.identity, session.spec.protocol()) != nil {
			session.fail(ErrProtocol)
			return
		}
		for _, command := range registration.Commands {
			if command.Reader && !session.spec.ReaderSettings || command.Presentation && !session.spec.Presentation || command.Interaction && (!command.Presentation || !session.spec.Interaction) {
				session.fail(ErrProtocol)
				return
			}
		}
		if registration.Extensions != nil && !session.spec.ReaderSettings {
			session.fail(ErrProtocol)
			return
		}
		session.mutex.Lock()
		if session.state == Starting && session.ctx.Err() == nil {
			session.commands = registration.Commands
			if registration.Extensions != nil {
				session.extensions = *registration.Extensions
			}
			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 {
					session.mutex.Lock()
					session.queuedBytes -= len(item.frame)
					session.mutex.Unlock()
					continue
				}
			}
			n, err := writer.Write(item.frame)
			if err == nil && n != len(item.frame) {
				err = io.ErrShortWrite
			}
			session.mutex.Lock()
			session.queuedBytes -= len(item.frame)
			session.mutex.Unlock()
			if 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
		}
		var fields map[string]json.RawMessage
		if validate_json(frame) != nil || json.Unmarshal(frame, &fields) != nil {
			session.fail(ErrProtocol)
			return
		}
		if _, request := fields["method"]; request {
			if err := session.accept_child_request(frame); err != nil {
				session.fail(err)
				return
			}
			continue
		}
		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)
	}
}