feat: add supervisor prototype with embedded frontend

internal/session/manager.go

@@ -0,0 +1,158 @@
+package session
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"supervisor/internal/domain"
+	"supervisor/internal/store"
+	"supervisor/internal/util"
+)
+
+var ErrSessionNotFound = errors.New("session not found")
+var ErrSessionRunning = errors.New("session is running")
+
+type Manager struct {
+	mu              sync.RWMutex
+	store           store.SessionStore
+	factory         PTYFactory
+	runtimes        map[string]*Session
+	scrollbackLimit int
+}
+
+func NewManager(store store.SessionStore, factory PTYFactory) *Manager {
+	if factory == nil {
+		factory = DefaultPTYFactory{}
+	}
+	return &Manager{
+		store:           store,
+		factory:         factory,
+		runtimes:        make(map[string]*Session),
+		scrollbackLimit: 512 * 1024,
+	}
+}
+
+func (m *Manager) CreateSession(ctx context.Context, params CreateSessionParams) (domain.Session, error) {
+	command := params.Command
+	if command == "" {
+		command = "bash"
+	}
+	s := domain.Session{
+		ID:        util.NewID("sess"),
+		Name:      params.Name,
+		AgentID:   params.AgentID,
+		Command:   command,
+		Status:    domain.SessionStatusCreated,
+		CreatedAt: time.Now().UTC(),
+	}
+	if s.Name == "" {
+		s.Name = s.ID
+	}
+
+	runtime := NewSession(s, m.scrollbackLimit, func(session domain.Session) {
+		_ = m.store.Upsert(context.Background(), session)
+	})
+
+	m.mu.Lock()
+	m.runtimes[s.ID] = runtime
+	m.mu.Unlock()
+
+	if err := m.store.Upsert(ctx, s); err != nil {
+		return domain.Session{}, err
+	}
+	return s, nil
+}
+
+func (m *Manager) StartSession(_ context.Context, id string) error {
+	runtime, err := m.runtimeByID(id)
+	if err != nil {
+		return err
+	}
+	return runtime.Start(m.factory)
+}
+
+func (m *Manager) StopSession(_ context.Context, id string) error {
+	runtime, err := m.runtimeByID(id)
+	if err != nil {
+		return err
+	}
+	return runtime.Stop()
+}
+
+func (m *Manager) ListSessions(ctx context.Context) ([]domain.Session, error) {
+	return m.store.List(ctx)
+}
+
+func (m *Manager) GetSession(ctx context.Context, id string) (domain.Session, error) {
+	s, ok, err := m.store.Get(ctx, id)
+	if err != nil {
+		return domain.Session{}, err
+	}
+	if !ok {
+		return domain.Session{}, ErrSessionNotFound
+	}
+	return s, nil
+}
+
+func (m *Manager) WriteInput(_ context.Context, id string, input string) error {
+	runtime, err := m.runtimeByID(id)
+	if err != nil {
+		return err
+	}
+	return runtime.WriteInput(input)
+}
+
+func (m *Manager) Subscribe(id string) (<-chan domain.Event, func(), error) {
+	runtime, err := m.runtimeByID(id)
+	if err != nil {
+		return nil, nil, err
+	}
+	_, ch, cancel := runtime.Subscribe()
+	return ch, cancel, nil
+}
+
+func (m *Manager) Resize(_ context.Context, id string, cols, rows int) error {
+	runtime, err := m.runtimeByID(id)
+	if err != nil {
+		return err
+	}
+	return runtime.Resize(cols, rows)
+}
+
+func (m *Manager) Scrollback(id string) ([]byte, error) {
+	runtime, err := m.runtimeByID(id)
+	if err != nil {
+		return nil, err
+	}
+	return runtime.Scrollback(), nil
+}
+
+func (m *Manager) DeleteSession(ctx context.Context, id string) error {
+	runtime, err := m.runtimeByID(id)
+	if err != nil {
+		return err
+	}
+
+	snapshot := runtime.Snapshot()
+	if snapshot.Status == domain.SessionStatusRunning {
+		return ErrSessionRunning
+	}
+
+	m.mu.Lock()
+	delete(m.runtimes, id)
+	m.mu.Unlock()
+
+	return m.store.Delete(ctx, id)
+}
+
+func (m *Manager) runtimeByID(id string) (*Session, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	runtime, ok := m.runtimes[id]
+	if !ok {
+		return nil, ErrSessionNotFound
+	}
+	return runtime, nil
+}

internal/session/models.go

@@ -0,0 +1,20 @@
+package session
+
+import "supervisor/internal/domain"
+
+type CreateSessionParams struct {
+	Name    string `json:"name"`
+	AgentID string `json:"agentId"`
+	Command string `json:"command"`
+}
+
+type InputRequest struct {
+	Input string `json:"input"`
+}
+
+type ResizeRequest struct {
+	Cols int `json:"cols"`
+	Rows int `json:"rows"`
+}
+
+type SessionSummary = domain.Session

internal/session/process.go

@@ -0,0 +1,53 @@
+package session
+
+import (
+	"os"
+	"os/exec"
+	"syscall"
+
+	"github.com/creack/pty"
+)
+
+type DefaultPTYFactory struct{}
+
+type shellProcess struct {
+	cmd *exec.Cmd
+	pty *os.File
+}
+
+func (f DefaultPTYFactory) Start(command string) (PTYProcess, error) {
+	cmd := exec.Command("bash", "-lc", command)
+	cmd.Env = os.Environ()
+	ptmx, err := pty.Start(cmd)
+	if err != nil {
+		return nil, err
+	}
+	return &shellProcess{cmd: cmd, pty: ptmx}, nil
+}
+
+func (p *shellProcess) Read(b []byte) (int, error) {
+	return p.pty.Read(b)
+}
+
+func (p *shellProcess) Write(b []byte) (int, error) {
+	return p.pty.Write(b)
+}
+
+func (p *shellProcess) Close() error {
+	return p.pty.Close()
+}
+
+func (p *shellProcess) Wait() error {
+	return p.cmd.Wait()
+}
+
+func (p *shellProcess) Resize(cols, rows uint16) error {
+	return pty.Setsize(p.pty, &pty.Winsize{Cols: cols, Rows: rows})
+}
+
+func (p *shellProcess) SignalStop() error {
+	if p.cmd.Process == nil {
+		return nil
+	}
+	return p.cmd.Process.Signal(syscall.SIGTERM)
+}

internal/session/pty.go

@@ -0,0 +1,14 @@
+package session
+
+import "io"
+
+type PTYProcess interface {
+	io.ReadWriteCloser
+	Wait() error
+	Resize(cols, rows uint16) error
+	SignalStop() error
+}
+
+type PTYFactory interface {
+	Start(command string) (PTYProcess, error)
+}

internal/session/session.go

@@ -0,0 +1,273 @@
+package session
+
+import (
+	"errors"
+	"io"
+	"os/exec"
+	"sync"
+	"time"
+
+	"supervisor/internal/domain"
+	"supervisor/internal/util"
+)
+
+type StateChangeFn func(domain.Session)
+
+type Session struct {
+	mu             sync.RWMutex
+	meta           domain.Session
+	process        PTYProcess
+	subscribers    map[string]chan domain.Event
+	scrollback     []byte
+	scrollbackSize int
+	onStateChange  StateChangeFn
+}
+
+func NewSession(meta domain.Session, scrollbackSize int, onStateChange StateChangeFn) *Session {
+	if scrollbackSize <= 0 {
+		scrollbackSize = 256 * 1024
+	}
+	return &Session{
+		meta:           meta,
+		subscribers:    make(map[string]chan domain.Event),
+		scrollbackSize: scrollbackSize,
+		onStateChange:  onStateChange,
+	}
+}
+
+func (s *Session) Snapshot() domain.Session {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.meta
+}
+
+func (s *Session) Scrollback() []byte {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	cpy := make([]byte, len(s.scrollback))
+	copy(cpy, s.scrollback)
+	return cpy
+}
+
+func (s *Session) Start(factory PTYFactory) error {
+	s.mu.Lock()
+	if s.meta.Status == domain.SessionStatusRunning {
+		s.mu.Unlock()
+		return nil
+	}
+	if s.meta.Command == "" {
+		s.mu.Unlock()
+		return errors.New("empty command")
+	}
+	proc, err := factory.Start(s.meta.Command)
+	if err != nil {
+		now := time.Now().UTC()
+		s.meta.Status = domain.SessionStatusError
+		s.meta.ExitedAt = &now
+		s.mu.Unlock()
+		s.emitStateChange()
+		s.publish(domain.Event{
+			Type:      domain.EventError,
+			SessionID: s.meta.ID,
+			Payload: domain.ErrorEvent{
+				Message: err.Error(),
+			},
+			At: time.Now().UTC(),
+		})
+		return err
+	}
+	now := time.Now().UTC()
+	s.process = proc
+	s.meta.Status = domain.SessionStatusRunning
+	s.meta.StartedAt = &now
+	s.meta.ExitedAt = nil
+	s.meta.ExitCode = nil
+	s.mu.Unlock()
+
+	s.emitStateChange()
+	s.publishStatus()
+
+	go s.readLoop(proc)
+	go s.waitLoop(proc)
+	return nil
+}
+
+func (s *Session) Stop() error {
+	s.mu.Lock()
+	if s.process == nil {
+		s.mu.Unlock()
+		return nil
+	}
+	proc := s.process
+	if s.meta.Status == domain.SessionStatusRunning {
+		s.meta.Status = domain.SessionStatusStopped
+	}
+	s.mu.Unlock()
+
+	s.emitStateChange()
+	s.publishStatus()
+	return proc.SignalStop()
+}
+
+func (s *Session) WriteInput(input string) error {
+	s.mu.RLock()
+	proc := s.process
+	status := s.meta.Status
+	s.mu.RUnlock()
+
+	if proc == nil || status != domain.SessionStatusRunning {
+		return errors.New("session is not running")
+	}
+	_, err := proc.Write([]byte(input))
+	return err
+}
+
+func (s *Session) Resize(cols, rows int) error {
+	if cols <= 0 || rows <= 0 {
+		return errors.New("invalid terminal size")
+	}
+	s.mu.RLock()
+	proc := s.process
+	s.mu.RUnlock()
+	if proc == nil {
+		return errors.New("session has no process")
+	}
+	return proc.Resize(uint16(cols), uint16(rows))
+}
+
+func (s *Session) Subscribe() (string, <-chan domain.Event, func()) {
+	id := util.NewID("sub")
+	ch := make(chan domain.Event, 128)
+	s.mu.Lock()
+	s.subscribers[id] = ch
+	s.mu.Unlock()
+	cancel := func() {
+		s.mu.Lock()
+		sub, ok := s.subscribers[id]
+		if ok {
+			delete(s.subscribers, id)
+			close(sub)
+		}
+		s.mu.Unlock()
+	}
+	return id, ch, cancel
+}
+
+func (s *Session) readLoop(proc PTYProcess) {
+	buf := make([]byte, 4096)
+	for {
+		n, err := proc.Read(buf)
+		if n > 0 {
+			chunk := append([]byte(nil), buf[:n]...)
+			s.appendScrollback(chunk)
+			s.publish(domain.Event{
+				Type:      domain.EventTerminalOutput,
+				SessionID: s.Snapshot().ID,
+				Payload: domain.TerminalOutputEvent{
+					Data: string(chunk),
+				},
+				At: time.Now().UTC(),
+			})
+		}
+		if err != nil {
+			if !errors.Is(err, io.EOF) {
+				s.publish(domain.Event{
+					Type:      domain.EventError,
+					SessionID: s.Snapshot().ID,
+					Payload: domain.ErrorEvent{Message: err.Error()},
+					At:        time.Now().UTC(),
+				})
+			}
+			return
+		}
+	}
+}
+
+func (s *Session) waitLoop(proc PTYProcess) {
+	err := proc.Wait()
+	_ = proc.Close()
+
+	var exitCode *int
+	if err != nil {
+		if exitErr, ok := err.(*exec.ExitError); ok {
+			code := exitErr.ExitCode()
+			exitCode = &code
+		}
+	}
+	if err == nil {
+		code := 0
+		exitCode = &code
+	}
+
+	now := time.Now().UTC()
+	s.mu.Lock()
+	if s.meta.Status != domain.SessionStatusError {
+		s.meta.Status = domain.SessionStatusExited
+	}
+	s.meta.ExitedAt = &now
+	s.meta.ExitCode = exitCode
+	s.process = nil
+	s.mu.Unlock()
+
+	s.emitStateChange()
+	s.publishStatus()
+
+	if err != nil {
+		s.publish(domain.Event{
+			Type:      domain.EventError,
+			SessionID: s.Snapshot().ID,
+			Payload: domain.ErrorEvent{Message: err.Error()},
+			At:        time.Now().UTC(),
+		})
+	}
+}
+
+func (s *Session) appendScrollback(data []byte) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(data) >= s.scrollbackSize {
+		s.scrollback = append([]byte(nil), data[len(data)-s.scrollbackSize:]...)
+		return
+	}
+	s.scrollback = append(s.scrollback, data...)
+	if len(s.scrollback) > s.scrollbackSize {
+		extra := len(s.scrollback) - s.scrollbackSize
+		s.scrollback = append([]byte(nil), s.scrollback[extra:]...)
+	}
+}
+
+func (s *Session) publishStatus() {
+	snap := s.Snapshot()
+	s.publish(domain.Event{
+		Type:      domain.EventSessionStatus,
+		SessionID: snap.ID,
+		Payload: domain.SessionStatusEvent{
+			Status:   snap.Status,
+			ExitCode: snap.ExitCode,
+		},
+		At: time.Now().UTC(),
+	})
+}
+
+func (s *Session) publish(event domain.Event) {
+	s.mu.RLock()
+	subs := make([]chan domain.Event, 0, len(s.subscribers))
+	for _, ch := range s.subscribers {
+		subs = append(subs, ch)
+	}
+	s.mu.RUnlock()
+
+	for _, ch := range subs {
+		select {
+		case ch <- event:
+		default:
+		}
+	}
+}
+
+func (s *Session) emitStateChange() {
+	if s.onStateChange == nil {
+		return
+	}
+	s.onStateChange(s.Snapshot())
+}