codespace/internal/acp/service.go

package acp

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"codespace/internal/fs"
	"codespace/internal/process"
	"codespace/internal/workspace"
)

// Service is the backend facade for the ACP endpoints.
type Service interface {
	Status(workspaceID string) (StatusResponse, error)
	History(workspaceID string) (HistoryResponse, error)
	Prompt(workspaceID string, content string) (PromptResponse, error)
	Stream(workspaceID string, content string, out chan<- StreamEvent) error
	Cancel(workspaceID string) error
	EnsureReady(workspaceID string) error
}

// StatusResponse is returned by Service.Status.
type StatusResponse struct {
	WorkspaceID string
	Ready       bool
	SessionID   string
	Running     bool
	PID         int
	Error       string
}

// HistoryResponse is returned by Service.History.
type HistoryResponse struct {
	SessionID string
	Messages  []Message
}

// PromptResponse is returned by Service.Prompt.
type PromptResponse struct {
	SessionID  string
	StopReason string
	Text       string
}

type localService struct {
	processes  process.Manager
	workspaces workspace.Manager
	lg         *slog.Logger
	mu         sync.Mutex
	clients    map[string]*Client
	locks      map[string]*sync.Mutex
}

// NewService creates an ACP service that owns one Client per workspace.
func NewService(processes process.Manager, workspaces workspace.Manager, lg *slog.Logger) Service {
	if lg == nil {
		lg = slog.Default()
	}
	return &localService{
		processes:  processes,
		workspaces: workspaces,
		lg:         lg,
		clients:    make(map[string]*Client),
		locks:      make(map[string]*sync.Mutex),
	}
}

func (s *localService) clientFor(workspaceID string) (*Client, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	c, ok := s.clients[workspaceID]
	if !ok {
		ws, err := s.workspaces.Get(workspaceID)
		if err != nil {
			return nil, err
		}
		c = NewClient(workspaceID, ws.Root, s.processes, fs.NewLocal(ws.Root), s.lg)
		s.clients[workspaceID] = c
		s.locks[workspaceID] = &sync.Mutex{}
	}
	return c, nil
}

func (s *localService) workspaceLock(workspaceID string) *sync.Mutex {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.locks[workspaceID] == nil {
		s.locks[workspaceID] = &sync.Mutex{}
	}
	return s.locks[workspaceID]
}

// Status returns the current ACP and process status without starting anything.
func (s *localService) Status(workspaceID string) (StatusResponse, error) {
	if _, err := s.workspaces.Get(workspaceID); err != nil {
		return StatusResponse{}, err
	}
	c, err := s.clientFor(workspaceID)
	if err != nil {
		return StatusResponse{}, err
	}
	ready, sessionID := c.Status()
	ps := s.processes.Status(workspaceID)
	return StatusResponse{
		WorkspaceID: workspaceID,
		Ready:       ready,
		SessionID:   sessionID,
		Running:     ps.Running,
		PID:         ps.PID,
	}, nil
}

// History returns the accumulated messages for the workspace.
func (s *localService) History(workspaceID string) (HistoryResponse, error) {
	if _, err := s.workspaces.Get(workspaceID); err != nil {
		return HistoryResponse{}, err
	}
	c, err := s.clientFor(workspaceID)
	if err != nil {
		return HistoryResponse{}, err
	}
	_, sessionID := c.Status()
	return HistoryResponse{
		SessionID: sessionID,
		Messages:  c.History(),
	}, nil
}

// EnsureReady initializes the process and ACP session for the workspace.
func (s *localService) EnsureReady(workspaceID string) error {
	if _, err := s.workspaces.Get(workspaceID); err != nil {
		return err
	}
	c, err := s.clientFor(workspaceID)
	if err != nil {
		return err
	}
	lock := s.workspaceLock(workspaceID)
	lock.Lock()
	defer lock.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	return c.Initialize(ctx)
}

// Prompt sends a user prompt and returns the agent's response text.
func (s *localService) Prompt(workspaceID string, content string) (PromptResponse, error) {
	if _, err := s.workspaces.Get(workspaceID); err != nil {
		return PromptResponse{}, err
	}
	c, err := s.clientFor(workspaceID)
	if err != nil {
		return PromptResponse{}, err
	}
	lock := s.workspaceLock(workspaceID)
	lock.Lock()
	defer lock.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	if err := c.Initialize(ctx); err != nil {
		return PromptResponse{}, err
	}
	res, err := c.Prompt(ctx, content)
	if err != nil {
		return PromptResponse{}, err
	}
	_, sessionID := c.Status()
	return PromptResponse{
		SessionID:  sessionID,
		StopReason: res.StopReason,
		Text:       res.Text,
	}, nil
}

// Cancel interrupts an in-flight prompt.
func (s *localService) Cancel(workspaceID string) error {
	if _, err := s.workspaces.Get(workspaceID); err != nil {
		return err
	}
	c, err := s.clientFor(workspaceID)
	if err != nil {
		return err
	}
	return c.Cancel()
}

// Stream sends a user prompt and streams response chunks to out.
func (s *localService) Stream(workspaceID string, content string, out chan<- StreamEvent) error {
	if _, err := s.workspaces.Get(workspaceID); err != nil {
		return err
	}
	c, err := s.clientFor(workspaceID)
	if err != nil {
		return err
	}
	lock := s.workspaceLock(workspaceID)
	lock.Lock()
	defer lock.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	if err := c.Initialize(ctx); err != nil {
		return err
	}
	return c.Stream(ctx, content, out)
}