codespace/internal/process/manager.go

package process

import (
	"fmt"
	"io"
	"os"
	"os/exec"
	"sync"
	"syscall"

	"codespace/internal/util"
)

const (
	outputBufferSize = 4 * 1024
	ringBufferSize   = 64 * 1024
)

// Manager manages OpenCode processes per workspace.
type Manager interface {
	Start(workspaceID string, workspaceRoot string) error
	Stop(workspaceID string) error
	Restart(workspaceID string, workspaceRoot string) error
	Status(workspaceID string) Status
	Subscribe(workspaceID string) (Subscription, error)
}

// LocalManager implements Manager using local OS processes.
type LocalManager struct {
	command  string
	args     []string
	mu       sync.Mutex
	sessions map[string]*Session
}

// NewManager creates a LocalManager with the given command.
// If command is empty, it defaults to "opencode".
func NewManager(command string) *LocalManager {
	return &LocalManager{
		command:  normalizeCommand(command),
		args:     []string{},
		sessions: make(map[string]*Session),
	}
}

// Start launches a process for the given workspace.
// Returns CodeConflict if a session is already running.
func (m *LocalManager) Start(workspaceID string, workspaceRoot string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if sess, ok := m.sessions[workspaceID]; ok && sess.Cmd.Process != nil {
		if sess.Cmd.ProcessState == nil {
			return util.New(util.CodeConflict, "process already running")
		}
	}

	cmd := exec.Command(m.command, m.args...)
	cmd.Dir = workspaceRoot
	cmd.Env = append(os.Environ(), fmt.Sprintf("HOME=%s", workspaceRoot))

	stdinR, stdinW, err := os.Pipe()
	if err != nil {
		return util.Wrap(util.CodeInternal, "failed to create stdin pipe", err)
	}

	stdoutR, stdoutW, err := os.Pipe()
	if err != nil {
		stdinR.Close()
		stdinW.Close()
		return util.Wrap(util.CodeInternal, "failed to create stdout pipe", err)
	}

	cmd.Stdin = stdinR
	cmd.Stdout = stdoutW
	cmd.Stderr = stdoutW

	if err := cmd.Start(); err != nil {
		stdinR.Close()
		stdinW.Close()
		stdoutR.Close()
		stdoutW.Close()
		return util.Wrap(util.CodeInternal, "failed to start process", err)
	}

	stdinR.Close()
	stdoutW.Close()

	sess := &Session{
		WorkspaceID: workspaceID,
		Root:        workspaceRoot,
		Cmd:         cmd,
		Stdin:       stdinW,
	}
	m.sessions[workspaceID] = sess

	outputDone := make(chan struct{})
	go m.captureOutput(sess, stdoutR, outputDone)
	go m.waitExit(sess, outputDone)

	return nil
}

// captureOutput reads from the merged stdout/stderr pipe, keeps the most
// recent ringBufferSize bytes in an in-memory ring buffer, and fans out each
// chunk to all subscribers using non-blocking sends.
func (m *LocalManager) captureOutput(sess *Session, stdoutR *os.File, done chan<- struct{}) {
	defer close(done)
	defer stdoutR.Close()

	buf := make([]byte, outputBufferSize)
	ring := newRingBuffer(ringBufferSize)

	for {
		n, err := stdoutR.Read(buf)
		if n > 0 {
			chunk := make([]byte, n)
			copy(chunk, buf[:n])
			ring.Write(chunk)

			sess.mu.Lock()
			subs := make([]Subscription, len(sess.Subscribers))
			copy(subs, sess.Subscribers)
			sess.mu.Unlock()

			for _, sub := range subs {
				sub.(*subscription).send(chunk)
			}
		}
		if err != nil {
			if err != io.EOF {
				// Ignore read errors; the pipe is closing.
			}
			break
		}
	}
}

// waitExit waits for the process to exit, then records the exit status, closes
// the stdin writer, and closes every subscriber channel exactly once.
func (m *LocalManager) waitExit(sess *Session, outputDone <-chan struct{}) {
	_ = sess.Cmd.Wait()
	<-outputDone

	sess.mu.Lock()
	defer sess.mu.Unlock()

	if sess.Cmd.ProcessState != nil {
		sess.Exit.Code = sess.Cmd.ProcessState.ExitCode()
		if ws, ok := sess.Cmd.ProcessState.Sys().(syscall.WaitStatus); ok && ws.Signaled() {
			sess.Exit.Signal = ws.Signal().String()
		}
	}

	if sess.Stdin != nil {
		_ = sess.Stdin.Close()
	}

	for _, sub := range sess.Subscribers {
		sub.(*subscription).closeChan()
	}
}

// Stop kills the process for the given workspace.
// Returns CodeNotFound if no session exists.
func (m *LocalManager) Stop(workspaceID string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	sess, ok := m.sessions[workspaceID]
	if !ok || sess.Cmd.Process == nil {
		return util.New(util.CodeNotFound, "no running process for workspace")
	}

	if err := sess.Cmd.Process.Kill(); err != nil {
		return util.Wrap(util.CodeInternal, "failed to stop process", err)
	}

	delete(m.sessions, workspaceID)
	return nil
}

// Restart stops (if running) then starts the process.
func (m *LocalManager) Restart(workspaceID string, workspaceRoot string) error {
	m.mu.Lock()
	if sess, ok := m.sessions[workspaceID]; ok && sess.Cmd.Process != nil {
		if sess.Cmd.ProcessState == nil {
			_ = sess.Cmd.Process.Kill()
			delete(m.sessions, workspaceID)
		}
	}
	m.mu.Unlock()

	return m.Start(workspaceID, workspaceRoot)
}

// Status returns the current process status for the workspace.
func (m *LocalManager) Status(workspaceID string) Status {
	m.mu.Lock()
	defer m.mu.Unlock()

	sess, ok := m.sessions[workspaceID]
	if !ok || sess.Cmd.Process == nil {
		return Status{WorkspaceID: workspaceID, Running: false}
	}

	if sess.Cmd.ProcessState != nil {
		return Status{WorkspaceID: workspaceID, Running: false}
	}

	return Status{
		WorkspaceID: workspaceID,
		Running:     true,
		PID:         sess.Cmd.Process.Pid,
	}
}

// Subscribe creates a new output subscription for the workspace.
// Returns CodeNotFound if the workspace has no session.
func (m *LocalManager) Subscribe(workspaceID string) (Subscription, error) {
	m.mu.Lock()
	sess, ok := m.sessions[workspaceID]
	m.mu.Unlock()

	if !ok {
		return nil, util.New(util.CodeNotFound, "no running process")
	}

	sess.mu.Lock()
	sub := newSubscription(sess)
	sess.Subscribers = append(sess.Subscribers, sub)
	if sess.Cmd.ProcessState != nil {
		sub.closeChan()
	}
	sess.mu.Unlock()

	return sub, nil
}

// ringBuffer is a fixed-size byte buffer that drops the oldest bytes when full.
type ringBuffer struct {
	mu   sync.Mutex
	buf  []byte
	maxN int
}

func newRingBuffer(maxN int) *ringBuffer {
	return &ringBuffer{maxN: maxN}
}

func (r *ringBuffer) Write(p []byte) {
	r.mu.Lock()
	defer r.mu.Unlock()

	if len(p) >= r.maxN {
		r.buf = append(r.buf[:0], p[len(p)-r.maxN:]...)
		return
	}

	r.buf = append(r.buf, p...)
	if len(r.buf) > r.maxN {
		r.buf = r.buf[len(r.buf)-r.maxN:]
	}
}