diff --git a/cmd/server/main.go b/cmd/server/main.go index bcbb2c8..ab1b39f 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -34,15 +34,16 @@ func main() { } workspaces := workspace.NewLocalManager(cfg.Workspace.Root) - processes := process.NewManager(cfg.Process.OpenCodeCommand) + processes := process.NewManager(cfg.Process.OpenCodeCommand, cfg.Process.Args) shellMgr := shell.NewManager(cfg.Shell.Command, cfg.Shell.Args) workspaceSvc := service.NewWorkspaceService(workspaces, processes, shellMgr, lg) fileSvc := service.NewFileService(workspaces, cfg.File.MaxWriteBytes) processSvc := service.NewProcessService(workspaces, processes, lg) shellSvc := service.NewShellService(workspaces, shellMgr, lg) + acpSvc := service.NewAcpService(processes, workspaces, lg) - router := api.NewRouter(workspaceSvc, fileSvc, processSvc, shellSvc, lg, cfg.Gin.Mode) + router := api.NewRouter(workspaceSvc, fileSvc, processSvc, shellSvc, acpSvc, lg, cfg.Gin.Mode) server := &http.Server{ Addr: cfg.Server.Addr, diff --git a/configs/config.yaml b/configs/config.yaml index 43b7d7e..64bc2da 100644 --- a/configs/config.yaml +++ b/configs/config.yaml @@ -8,6 +8,8 @@ workspace: root: "./workspaces" process: opencodeCommand: "opencode" + args: + - "acp" shell: command: "bash" diff --git a/internal/acp/client.go b/internal/acp/client.go new file mode 100644 index 0000000..df531ee --- /dev/null +++ b/internal/acp/client.go @@ -0,0 +1,269 @@ +package acp + +import ( + "context" + "encoding/json" + "log/slog" + "strings" + "sync" + + "codespace/internal/fs" + "codespace/internal/process" + "codespace/internal/util" +) + +// PromptResult is the outcome of a single prompt turn. +type PromptResult struct { + StopReason string + Text string +} + +// Client manages one ACP session for a single workspace process. +type Client struct { + workspaceID string + root string + processes process.Manager + fs fs.FileSystem + lg *slog.Logger + + mu sync.RWMutex + ready bool + sessionID string + transport *transport + history *History +} + +// NewClient creates an ACP client for the given workspace. +func NewClient(workspaceID, root string, processes process.Manager, filesystem fs.FileSystem, lg *slog.Logger) *Client { + if lg == nil { + lg = slog.Default() + } + return &Client{ + workspaceID: workspaceID, + root: root, + processes: processes, + fs: filesystem, + lg: lg, + history: NewHistory(), + } +} + +// Initialize starts the process (if necessary), performs the ACP handshake, +// and creates a session. It is safe to call repeatedly; the client will reset +// and re-initialize if the process is not running. +func (c *Client) Initialize(ctx context.Context) error { + c.mu.RLock() + already := c.ready && c.sessionID != "" && c.processes.Status(c.workspaceID).Running + c.mu.RUnlock() + if already { + return nil + } + + c.mu.Lock() + c.ready = false + c.sessionID = "" + c.history.Clear() + old := c.transport + c.transport = nil + c.mu.Unlock() + + if old != nil { + old.onRequest = nil + old.onNotification = nil + old.onClose = nil + _ = old.Close() + } + + if !c.processes.Status(c.workspaceID).Running { + if err := c.processes.Start(c.workspaceID, c.root); err != nil { + return util.Wrap(util.CodeInternal, "failed to start opencode", err) + } + } + + stdin, err := c.processes.Stdin(c.workspaceID) + if err != nil { + return util.Wrap(util.CodeInternal, "failed to get opencode stdin", err) + } + sub, err := c.processes.Subscribe(c.workspaceID) + if err != nil { + return util.Wrap(util.CodeInternal, "failed to subscribe to opencode output", err) + } + + tr := newTransport(stdin, sub, c.lg) + tr.onRequest = c.handleRequest + tr.onNotification = c.handleNotification + tr.onClose = c.onTransportClose + tr.start() + + initParams := InitializeParams{ + ProtocolVersion: 1, + ClientCapabilities: ClientCapabilities{ + FS: ClientFSCapabilities{ReadTextFile: true, WriteTextFile: true}, + }, + ClientInfo: ClientInfo{Name: "codespace", Version: "0.1.0"}, + } + var initRes InitializeResult + if err := tr.call(ctx, "initialize", initParams, &initRes); err != nil { + _ = tr.Close() + return util.Wrap(util.CodeInternal, "initialize failed", err) + } + + sessParams := SessionNewParams{Cwd: c.root, MCPServers: []any{}} + var sessRes SessionNewResult + if err := tr.call(ctx, "session/new", sessParams, &sessRes); err != nil { + _ = tr.Close() + return util.Wrap(util.CodeInternal, "session/new failed", err) + } + + c.mu.Lock() + c.transport = tr + c.sessionID = sessRes.SessionID + c.ready = true + c.mu.Unlock() + + c.lg.Info("acp initialized", "workspace_id", c.workspaceID, "session_id", sessRes.SessionID) + return nil +} + +// Prompt sends a user message and returns the agent's final text for this turn. +func (c *Client) Prompt(ctx context.Context, content string) (PromptResult, error) { + var result PromptResult + c.mu.RLock() + if !c.ready { + c.mu.RUnlock() + return result, util.New(util.CodeConflict, "client not ready") + } + sessionID := c.sessionID + tr := c.transport + startIdx := c.history.Len() + c.mu.RUnlock() + + c.history.Add("user", content) + + params := SessionPromptParams{ + SessionID: sessionID, + Prompt: []PromptMessage{{Type: "text", Text: content}}, + } + if err := tr.call(ctx, "session/prompt", params, &result); err != nil { + return result, util.Wrap(util.CodeInternal, "session/prompt failed", err) + } + + msgs := c.history.Since(startIdx) + var sb strings.Builder + for _, m := range msgs { + if m.Role == "agent" { + sb.WriteString(m.Text) + } + } + result.Text = sb.String() + return result, nil +} + +// Cancel sends a best-effort session/cancel notification. +func (c *Client) Cancel() error { + c.mu.RLock() + defer c.mu.RUnlock() + if !c.ready || c.transport == nil { + return nil + } + return c.transport.sendNotification("session/cancel", SessionCancelParams{SessionID: c.sessionID}) +} + +// Status returns whether the client has a ready session and its id. +func (c *Client) Status() (ready bool, sessionID string) { + c.mu.RLock() + defer c.mu.RUnlock() + return c.ready, c.sessionID +} + +// History returns the in-memory conversation history. +func (c *Client) History() []Message { + return c.history.List() +} + +// handleRequest dispatches agent-initiated JSON-RPC requests. +func (c *Client) handleRequest(method string, params json.RawMessage, id int) { + var result any + var rpcErr *rpcError + + switch method { + case "fs/read_text_file": + var p FSReadParams + if err := json.Unmarshal(params, &p); err != nil { + rpcErr = &rpcError{Code: -32700, Message: "Parse error"} + break + } + data, err := c.fs.Read(p.Path) + if err != nil { + rpcErr = c.rpcErrorFromErr(err) + } else { + result = FSReadResult{Content: string(data)} + } + case "fs/write_text_file": + var p FSWriteParams + if err := json.Unmarshal(params, &p); err != nil { + rpcErr = &rpcError{Code: -32700, Message: "Parse error"} + break + } + if err := c.fs.Write(p.Path, []byte(p.Content)); err != nil { + rpcErr = c.rpcErrorFromErr(err) + } else { + result = FSWriteResult{} + } + default: + rpcErr = &rpcError{Code: -32601, Message: "Method not found: " + method} + } + + c.mu.RLock() + tr := c.transport + c.mu.RUnlock() + if tr != nil { + if err := tr.sendResponse(id, result, rpcErr); err != nil { + c.lg.Error("acp failed to send response", "workspace_id", c.workspaceID, "method", method, "error", err) + } + } +} + +// handleNotification processes agent-initiated notifications. +func (c *Client) handleNotification(method string, params json.RawMessage) { + if method != "session/update" { + c.lg.Debug("acp drop notification", "workspace_id", c.workspaceID, "method", method) + return + } + var up SessionUpdateParams + if err := json.Unmarshal(params, &up); err != nil { + c.lg.Error("acp invalid session/update", "workspace_id", c.workspaceID, "error", err) + return + } + switch up.Update.SessionUpdate { + case "agent_message_chunk": + c.history.Add("agent", up.Update.Content) + case "user_message_chunk": + c.history.Add("user", up.Update.Content) + default: + c.lg.Debug("acp drop update", "workspace_id", c.workspaceID, "type", up.Update.SessionUpdate) + } +} + +// onTransportClose is invoked when the process output stream closes. +func (c *Client) onTransportClose() { + c.mu.Lock() + defer c.mu.Unlock() + if c.ready { + c.ready = false + c.sessionID = "" + c.transport = nil + c.history.Clear() + c.lg.Info("acp transport closed", "workspace_id", c.workspaceID) + } +} + +// rpcErrorFromErr maps a filesystem error to a JSON-RPC error. +func (c *Client) rpcErrorFromErr(err error) *rpcError { + switch util.CodeOf(err) { + case util.CodeNotFound, util.CodeBadRequest: + return &rpcError{Code: -32602, Message: err.Error()} + default: + return &rpcError{Code: -32603, Message: err.Error()} + } +} diff --git a/internal/acp/client_test.go b/internal/acp/client_test.go new file mode 100644 index 0000000..5ad796c --- /dev/null +++ b/internal/acp/client_test.go @@ -0,0 +1,278 @@ +package acp + +import ( + "bytes" + "encoding/json" + "log/slog" + "os" + "strings" + "sync" + "testing" + "time" + + "codespace/internal/fs" +) + +// fakeSubscription implements process.Subscription for tests. +type fakeSubscription struct { + ch chan []byte + once sync.Once +} + +func newFakeSubscription() *fakeSubscription { + return &fakeSubscription{ch: make(chan []byte, 16)} +} + +func (s *fakeSubscription) Output() <-chan []byte { + return s.ch +} + +func (s *fakeSubscription) send(b []byte) { + s.ch <- b +} + +func (s *fakeSubscription) Close() error { + s.once.Do(func() { close(s.ch) }) + return nil +} + +// recordingWriteCloser captures every byte written to it. +type recordingWriteCloser struct { + mu sync.Mutex + buf bytes.Buffer + closed bool +} + +func (w *recordingWriteCloser) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.Write(p) +} + +func (w *recordingWriteCloser) Close() error { + w.closed = true + return nil +} + +func (w *recordingWriteCloser) String() string { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.String() +} + +// stubFS implements fs.FileSystem for read/write tests. +type stubFS struct { + mu sync.Mutex + files map[string]string +} + +func newStubFS() *stubFS { + return &stubFS{files: make(map[string]string)} +} + +func (s *stubFS) Read(path string) ([]byte, error) { + s.mu.Lock() + defer s.mu.Unlock() + if data, ok := s.files[path]; ok { + return []byte(data), nil + } + return nil, os.ErrNotExist +} + +func (s *stubFS) Write(path string, data []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + s.files[path] = string(data) + return nil +} + +func (s *stubFS) List(path string) ([]fs.FileInfo, error) { return nil, nil } +func (s *stubFS) Mkdir(path string) error { return nil } +func (s *stubFS) Remove(path string) error { return nil } +func (s *stubFS) Rename(oldPath, newPath string) error { return nil } +func (s *stubFS) Stat(path string) (*fs.FileInfo, error) { return nil, nil } + +func TestNDJSONFramingHandlesSplitLines(t *testing.T) { + sub := newFakeSubscription() + stdin := &recordingWriteCloser{} + tr := newTransport(stdin, sub, slog.Default()) + tr.start() + defer tr.Close() + + got := make(chan struct { + method string + content string + }, 2) + tr.onNotification = func(method string, params json.RawMessage) { + if method != "session/update" { + return + } + var up SessionUpdateParams + if err := json.Unmarshal(params, &up); err != nil { + t.Errorf("unmarshal notification: %v", err) + return + } + got <- struct { + method string + content string + }{method, up.Update.Content} + } + + line1 := `{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"s1","update":{"sessionUpdate":"agent_message_chunk","content":"hello"}}}` + line2 := `{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"s1","update":{"sessionUpdate":"agent_message_chunk","content":" world"}}}` + + // Send one complete line plus a partial second line. + sub.send([]byte(line1 + "\n" + line2[:len(line2)-2])) + + select { + case n := <-got: + if n.content != "hello" { + t.Fatalf("first content = %q, want hello", n.content) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for first notification") + } + + // Complete the second line. + sub.send([]byte(line2[len(line2)-2:] + "\n")) + + select { + case n := <-got: + if n.content != " world" { + t.Fatalf("second content = %q, want ' world'", n.content) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for second notification") + } +} + +func TestRequestResponseCorrelation(t *testing.T) { + sub := newFakeSubscription() + stdin := &recordingWriteCloser{} + tr := newTransport(stdin, sub, slog.Default()) + tr.start() + defer tr.Close() + + ch := make(chan *jsonRPCMessage, 1) + tr.pendingMu.Lock() + tr.pending[1] = ch + tr.pendingMu.Unlock() + + sub.send([]byte(`{"jsonrpc":"2.0","id":1,"result":{"sessionId":"sess-1"}}` + "\n")) + + select { + case resp := <-ch: + if resp.ID == nil || *resp.ID != 1 { + t.Fatalf("id = %v, want 1", resp.ID) + } + var res SessionNewResult + if err := json.Unmarshal(resp.Result, &res); err != nil { + t.Fatal(err) + } + if res.SessionID != "sess-1" { + t.Fatalf("sessionId = %q, want sess-1", res.SessionID) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for response") + } +} + +func TestNotificationDispatch(t *testing.T) { + sub := newFakeSubscription() + stdin := &recordingWriteCloser{} + tr := newTransport(stdin, sub, slog.Default()) + tr.start() + defer tr.Close() + + got := make(chan string, 1) + tr.onNotification = func(method string, params json.RawMessage) { + got <- method + } + + sub.send([]byte(`{"jsonrpc":"2.0","method":"ping"}` + "\n")) + + select { + case m := <-got: + if m != "ping" { + t.Fatalf("method = %q, want ping", m) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for notification") + } +} + +func TestAgentRequestTerminalReturnsMethodNotFound(t *testing.T) { + c := &Client{ + workspaceID: "ws", + fs: newStubFS(), + history: NewHistory(), + lg: slog.Default(), + } + stdin := &recordingWriteCloser{} + tr := newTransport(stdin, newFakeSubscription(), slog.Default()) + c.transport = tr + c.ready = true + c.sessionID = "s1" + + c.handleRequest("terminal/create", json.RawMessage(`{}`), 7) + + line := strings.TrimSpace(stdin.String()) + if line == "" { + t.Fatal("expected a response to be written") + } + var resp jsonRPCMessage + if err := json.Unmarshal([]byte(line), &resp); err != nil { + t.Fatalf("unmarshal response: %v", err) + } + if resp.ID == nil || *resp.ID != 7 { + t.Fatalf("id = %v, want 7", resp.ID) + } + if resp.Error == nil { + t.Fatal("expected error response") + } + if resp.Error.Code != -32601 { + t.Fatalf("error code = %d, want -32601", resp.Error.Code) + } +} + +func TestAgentRequestReadFile(t *testing.T) { + stub := newStubFS() + stub.files["main.go"] = "package main\n" + + c := &Client{ + workspaceID: "ws", + fs: stub, + history: NewHistory(), + lg: slog.Default(), + } + stdin := &recordingWriteCloser{} + tr := newTransport(stdin, newFakeSubscription(), slog.Default()) + c.transport = tr + c.ready = true + c.sessionID = "s1" + + params, _ := json.Marshal(FSReadParams{Path: "main.go", SessionID: "s1"}) + c.handleRequest("fs/read_text_file", params, 3) + + line := strings.TrimSpace(stdin.String()) + if line == "" { + t.Fatal("expected a response to be written") + } + var resp jsonRPCMessage + if err := json.Unmarshal([]byte(line), &resp); err != nil { + t.Fatalf("unmarshal response: %v", err) + } + if resp.ID == nil || *resp.ID != 3 { + t.Fatalf("id = %v, want 3", resp.ID) + } + if resp.Error != nil { + t.Fatalf("unexpected error: %v", resp.Error) + } + var res FSReadResult + if err := json.Unmarshal(resp.Result, &res); err != nil { + t.Fatal(err) + } + if res.Content != "package main\n" { + t.Fatalf("content = %q, want 'package main\\n'", res.Content) + } +} diff --git a/internal/acp/history.go b/internal/acp/history.go new file mode 100644 index 0000000..98866cc --- /dev/null +++ b/internal/acp/history.go @@ -0,0 +1,69 @@ +package acp + +import ( + "sync" + "time" +) + +// Message is one entry in the conversation history. +type Message struct { + Role string `json:"role"` + Text string `json:"text"` + Time time.Time `json:"time"` +} + +// History accumulates prompt/response messages in memory for a session. +type History struct { + mu sync.RWMutex + messages []Message +} + +// NewHistory creates an empty History. +func NewHistory() *History { + return &History{} +} + +// Add appends a message with the current time. +func (h *History) Add(role, text string) { + h.mu.Lock() + defer h.mu.Unlock() + h.messages = append(h.messages, Message{Role: role, Text: text, Time: time.Now()}) +} + +// List returns a copy of all messages, oldest first. +func (h *History) List() []Message { + h.mu.RLock() + defer h.mu.RUnlock() + out := make([]Message, len(h.messages)) + copy(out, h.messages) + return out +} + +// Len returns the number of stored messages. +func (h *History) Len() int { + h.mu.RLock() + defer h.mu.RUnlock() + return len(h.messages) +} + +// Since returns messages starting at index idx. +func (h *History) Since(idx int) []Message { + h.mu.RLock() + defer h.mu.RUnlock() + if idx < 0 { + idx = 0 + } + if idx > len(h.messages) { + idx = len(h.messages) + } + out := make([]Message, len(h.messages)-idx) + copy(out, h.messages[idx:]) + return out +} + +// Clear removes all stored messages. +func (h *History) Clear() { + h.mu.Lock() + defer h.mu.Unlock() + h.messages = h.messages[:0] +} diff --git a/internal/acp/messages.go b/internal/acp/messages.go new file mode 100644 index 0000000..5a0d08d --- /dev/null +++ b/internal/acp/messages.go @@ -0,0 +1,121 @@ +package acp + +import "encoding/json" + +// jsonRPCMessage is the raw JSON-RPC 2.0 envelope used over NDJSON. +type jsonRPCMessage struct { + JSONRPC string `json:"jsonrpc"` + ID *int `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *rpcError `json:"error,omitempty"` +} + +// rpcError is a JSON-RPC 2.0 error object. +type rpcError struct { + Code int `json:"code"` + Message string `json:"message"` + Data any `json:"data,omitempty"` +} + +// InitializeParams is sent once to negotiate the ACP session. +type InitializeParams struct { + ProtocolVersion int `json:"protocolVersion"` + ClientCapabilities ClientCapabilities `json:"clientCapabilities"` + ClientInfo ClientInfo `json:"clientInfo"` +} + +// ClientCapabilities describes tools the client exposes to the agent. +type ClientCapabilities struct { + FS ClientFSCapabilities `json:"fs"` +} + +// ClientFSCapabilities declares filesystem support. +type ClientFSCapabilities struct { + ReadTextFile bool `json:"readTextFile"` + WriteTextFile bool `json:"writeTextFile"` +} + +// ClientInfo identifies this client to the agent. +type ClientInfo struct { + Name string `json:"name"` + Version string `json:"version"` +} + +// InitializeResult is returned by the agent for initialize. +type InitializeResult struct { + ProtocolVersion int `json:"protocolVersion"` + AgentCapabilities any `json:"agentCapabilities"` + AgentInfo any `json:"agentInfo"` +} + +// SessionNewParams starts a new agent session for a workspace. +type SessionNewParams struct { + Cwd string `json:"cwd"` + MCPServers []any `json:"mcpServers"` +} + +// SessionNewResult returns the session identifier. +type SessionNewResult struct { + SessionID string `json:"sessionId"` +} + +// PromptMessage is a single user message part. +type PromptMessage struct { + Type string `json:"type"` + Text string `json:"text"` +} + +// SessionPromptParams sends user input to the agent. +type SessionPromptParams struct { + SessionID string `json:"sessionId"` + Prompt []PromptMessage `json:"prompt"` +} + +// SessionPromptResult is the final response for a prompt turn. +type SessionPromptResult struct { + StopReason string `json:"stopReason"` +} + +// SessionCancelParams interrupts an in-flight turn. +type SessionCancelParams struct { + SessionID string `json:"sessionId"` +} + +// SessionUpdateParams is the payload of a session/update notification. +type SessionUpdateParams struct { + SessionID string `json:"sessionId"` + Update Update `json:"update"` +} + +// Update carries a single streamed update from the agent. +type Update struct { + SessionUpdate string `json:"sessionUpdate"` + MessageID string `json:"messageId,omitempty"` + Content string `json:"content,omitempty"` + ToolCallID string `json:"toolCallId,omitempty"` +} + +// FSReadParams is sent by the agent to read a workspace file. +type FSReadParams struct { + Path string `json:"path"` + SessionID string `json:"sessionId"` + Line *int `json:"line,omitempty"` + Limit *int `json:"limit,omitempty"` +} + +// FSReadResult returns file contents to the agent. +type FSReadResult struct { + Content string `json:"content"` +} + +// FSWriteParams is sent by the agent to write a workspace file. +type FSWriteParams struct { + Path string `json:"path"` + SessionID string `json:"sessionId"` + Content string `json:"content"` +} + +// FSWriteResult is the empty success response for fs/write_text_file. +type FSWriteResult struct{} diff --git a/internal/acp/service.go b/internal/acp/service.go new file mode 100644 index 0000000..d06daac --- /dev/null +++ b/internal/acp/service.go @@ -0,0 +1,188 @@ +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) + 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() +} diff --git a/internal/acp/transport.go b/internal/acp/transport.go new file mode 100644 index 0000000..c610893 --- /dev/null +++ b/internal/acp/transport.go @@ -0,0 +1,198 @@ +package acp + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "sync" + "sync/atomic" + + "codespace/internal/process" +) + +// transport reads NDJSON frames from a process subscription and writes NDJSON +// frames to the process stdin. +type transport struct { + stdin io.WriteCloser + sub process.Subscription + frames chan *jsonRPCMessage + pending map[int]chan *jsonRPCMessage + pendingMu sync.Mutex + nextID atomic.Int64 + onRequest func(method string, params json.RawMessage, id int) + onNotification func(method string, params json.RawMessage) + onClose func() + lg *slog.Logger + closeOnce sync.Once +} + +// newTransport creates a transport bound to the given stdin writer and output +// subscription. Call start to begin the read/dispatch goroutines. +func newTransport(stdin io.WriteCloser, sub process.Subscription, lg *slog.Logger) *transport { + if lg == nil { + lg = slog.Default() + } + return &transport{ + stdin: stdin, + sub: sub, + frames: make(chan *jsonRPCMessage, 64), + pending: make(map[int]chan *jsonRPCMessage), + lg: lg, + } +} + +// start launches the read and dispatch loops. +func (t *transport) start() { + go t.readLoop() + go t.dispatchLoop() +} + +// readLoop consumes raw bytes from the subscription, splits on newlines, and +// feeds parsed JSON-RPC frames to the frames channel. +func (t *transport) readLoop() { + defer close(t.frames) + buf := bytes.NewBuffer(make([]byte, 0, 4096)) + for chunk := range t.sub.Output() { + buf.Write(chunk) + for { + data := buf.Bytes() + idx := bytes.IndexByte(data, '\n') + if idx < 0 { + break + } + line := make([]byte, idx) + copy(line, data[:idx]) + buf.Next(idx + 1) + msg := &jsonRPCMessage{} + if err := json.Unmarshal(line, msg); err != nil { + t.lg.Error("acp invalid ndjson line", "error", err) + continue + } + t.frames <- msg + } + } +} + +// dispatchLoop routes frames to pending request channels, notification +// handlers, or incoming-request handlers. +func (t *transport) dispatchLoop() { + defer func() { + if t.onClose != nil { + t.onClose() + } + }() + for msg := range t.frames { + switch { + case msg.ID != nil && msg.Method != "": + // Incoming request from the agent. + if t.onRequest != nil { + id := *msg.ID + go t.onRequest(msg.Method, msg.Params, id) + } + case msg.ID == nil && msg.Method != "": + // Notification from the agent. + if t.onNotification != nil { + go t.onNotification(msg.Method, msg.Params) + } + case msg.ID != nil && (msg.Result != nil || msg.Error != nil): + // Response to a client-initiated request. + id := *msg.ID + t.pendingMu.Lock() + ch, ok := t.pending[id] + delete(t.pending, id) + t.pendingMu.Unlock() + if ok { + select { + case ch <- msg: + default: + } + } else { + t.lg.Debug("acp unsolicited response", "id", id) + } + default: + t.lg.Debug("acp unhandled frame", "method", msg.Method, "id", msg.ID) + } + } +} + +// call sends a JSON-RPC request and waits for its response. +func (t *transport) call(ctx context.Context, method string, params, result any) error { + id := int(t.nextID.Add(1)) + rawParams, err := json.Marshal(params) + if err != nil { + return err + } + msg := &jsonRPCMessage{JSONRPC: "2.0", ID: &id, Method: method, Params: rawParams} + ch := make(chan *jsonRPCMessage, 1) + t.pendingMu.Lock() + t.pending[id] = ch + t.pendingMu.Unlock() + defer func() { + t.pendingMu.Lock() + delete(t.pending, id) + t.pendingMu.Unlock() + }() + if err := t.write(msg); err != nil { + return err + } + select { + case <-ctx.Done(): + return ctx.Err() + case resp := <-ch: + if resp.Error != nil { + return errors.New(resp.Error.Message) + } + if result != nil && resp.Result != nil { + return json.Unmarshal(resp.Result, result) + } + return nil + } +} + +// sendNotification writes a JSON-RPC notification (no id). +func (t *transport) sendNotification(method string, params any) error { + rawParams, err := json.Marshal(params) + if err != nil { + return err + } + msg := &jsonRPCMessage{JSONRPC: "2.0", Method: method, Params: rawParams} + return t.write(msg) +} + +// sendResponse writes a JSON-RPC response for an agent-initiated request. +func (t *transport) sendResponse(id int, result any, rpcErr *rpcError) error { + msg := &jsonRPCMessage{JSONRPC: "2.0", ID: &id} + if rpcErr != nil { + msg.Error = rpcErr + } else { + raw, err := json.Marshal(result) + if err != nil { + return err + } + msg.Result = raw + } + return t.write(msg) +} + +// write marshals a JSON-RPC message and appends a newline. +func (t *transport) write(msg *jsonRPCMessage) error { + data, err := json.Marshal(msg) + if err != nil { + return err + } + data = append(data, '\n') + _, err = t.stdin.Write(data) + return err +} + +// Close shuts down the transport by closing the subscription. This causes the +// read loop to exit and the dispatch loop to follow. +func (t *transport) Close() error { + t.closeOnce.Do(func() { + _ = t.sub.Close() + }) + return nil +} diff --git a/internal/api/acp_handler.go b/internal/api/acp_handler.go new file mode 100644 index 0000000..5f43074 --- /dev/null +++ b/internal/api/acp_handler.go @@ -0,0 +1,69 @@ +package api + +import ( + "net/http" + + "codespace/internal/model" + "codespace/internal/service" + + "github.com/gin-gonic/gin" +) + +type acpHandler struct { + svc *service.AcpService +} + +func (h *acpHandler) status(c *gin.Context) { + id := c.Param("id") + status, err := h.svc.Status(id) + if err != nil { + writeError(c, err) + return + } + c.JSON(http.StatusOK, model.AcpStatusResponse{ + WorkspaceID: status.WorkspaceID, + Ready: status.Ready, + SessionID: status.SessionID, + Running: status.Running, + PID: status.PID, + Error: status.Error, + }) +} + +func (h *acpHandler) history(c *gin.Context) { + id := c.Param("id") + hist, err := h.svc.History(id) + if err != nil { + writeError(c, err) + return + } + c.JSON(http.StatusOK, hist) +} + +func (h *acpHandler) prompt(c *gin.Context) { + id := c.Param("id") + var req model.AcpPromptRequest + if err := c.ShouldBindJSON(&req); err != nil { + writeBadRequest(c, "invalid json") + return + } + if req.Content == "" { + writeBadRequest(c, "content is required") + return + } + res, err := h.svc.Prompt(id, req.Content) + if err != nil { + writeError(c, err) + return + } + c.JSON(http.StatusOK, res) +} + +func (h *acpHandler) cancel(c *gin.Context) { + id := c.Param("id") + if err := h.svc.Cancel(id); err != nil { + writeError(c, err) + return + } + c.Status(http.StatusNoContent) +} diff --git a/internal/api/process_ws_test.go b/internal/api/process_ws_test.go index 02cd469..ebf03b9 100644 --- a/internal/api/process_ws_test.go +++ b/internal/api/process_ws_test.go @@ -29,14 +29,15 @@ func TestProcessWS(t *testing.T) { wsRoot := filepath.Join(tmpDir, "workspaces") wsMgr := workspace.NewLocalManager(wsRoot) - procMgr := process.NewManager(opencodePath) + procMgr := process.NewManager(opencodePath, nil) shellMgr := shell.NewManager("bash", []string{"-i"}) wsSvc := service.NewWorkspaceService(wsMgr, procMgr, shellMgr, nil) fileSvc := service.NewFileService(wsMgr, 1<<20) procSvc := service.NewProcessService(wsMgr, procMgr, nil) shellSvc := service.NewShellService(wsMgr, shellMgr, nil) + acpSvc := service.NewAcpService(procMgr, wsMgr, nil) - r := NewRouter(wsSvc, fileSvc, procSvc, shellSvc, nil, gin.TestMode) + r := NewRouter(wsSvc, fileSvc, procSvc, shellSvc, acpSvc, nil, gin.TestMode) srv := httptest.NewServer(r) defer srv.Close() @@ -104,14 +105,15 @@ func TestProcessWSMultiSubscriber(t *testing.T) { wsRoot := filepath.Join(tmpDir, "workspaces") wsMgr := workspace.NewLocalManager(wsRoot) - procMgr := process.NewManager(opencodePath) + procMgr := process.NewManager(opencodePath, nil) shellMgr := shell.NewManager("bash", []string{"-i"}) wsSvc := service.NewWorkspaceService(wsMgr, procMgr, shellMgr, nil) fileSvc := service.NewFileService(wsMgr, 1<<20) procSvc := service.NewProcessService(wsMgr, procMgr, nil) shellSvc := service.NewShellService(wsMgr, shellMgr, nil) + acpSvc := service.NewAcpService(procMgr, wsMgr, nil) - r := NewRouter(wsSvc, fileSvc, procSvc, shellSvc, nil, gin.TestMode) + r := NewRouter(wsSvc, fileSvc, procSvc, shellSvc, acpSvc, nil, gin.TestMode) srv := httptest.NewServer(r) defer srv.Close() diff --git a/internal/api/router.go b/internal/api/router.go index 36a275e..2ba87a0 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -10,7 +10,7 @@ import ( ) // NewRouter builds a Gin engine with all API routes registered. -func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, processes *service.ProcessService, shells *service.ShellService, lg *slog.Logger, ginMode string) *gin.Engine { +func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, processes *service.ProcessService, shells *service.ShellService, acpSvc *service.AcpService, lg *slog.Logger, ginMode string) *gin.Engine { gin.SetMode(ginMode) r := gin.New() r.Use(gin.Recovery()) @@ -20,6 +20,7 @@ func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, fileHandler := &fileHandler{svc: files} procHandler := &processHandler{svc: processes} shellHandler := &shellHandler{svc: shells} + acpHandler := &acpHandler{svc: acpSvc} r.GET("/healthz", healthHandler) @@ -48,6 +49,10 @@ func NewRouter(workspaces *service.WorkspaceService, files *service.FileService, api.POST("/workspaces/:id/shell/restart", shellHandler.restart) api.GET("/workspaces/:id/shell/status", shellHandler.status) api.GET("/workspaces/:id/shell/ws", shellHandler.ws) + api.GET("/workspaces/:id/acp/status", acpHandler.status) + api.GET("/workspaces/:id/acp/history", acpHandler.history) + api.POST("/workspaces/:id/acp/prompt", acpHandler.prompt) + api.POST("/workspaces/:id/acp/cancel", acpHandler.cancel) return r } diff --git a/internal/api/router_test.go b/internal/api/router_test.go index 75fccf9..bb2ed1e 100644 --- a/internal/api/router_test.go +++ b/internal/api/router_test.go @@ -24,13 +24,14 @@ func setupTestRouterWithMaxWriteBytes(t *testing.T, maxWriteBytes int64) http.Ha t.Helper() root := t.TempDir() wsMgr := workspace.NewLocalManager(root) - procMgr := process.NewManager("") + procMgr := process.NewManager("", nil) shellMgr := shell.NewManager("bash", []string{"-i"}) wsSvc := service.NewWorkspaceService(wsMgr, procMgr, shellMgr, nil) fileSvc := service.NewFileService(wsMgr, maxWriteBytes) procSvc := service.NewProcessService(wsMgr, procMgr, nil) shellSvc := service.NewShellService(wsMgr, shellMgr, nil) - return NewRouter(wsSvc, fileSvc, procSvc, shellSvc, nil, gin.TestMode) + acpSvc := service.NewAcpService(procMgr, wsMgr, nil) + return NewRouter(wsSvc, fileSvc, procSvc, shellSvc, acpSvc, nil, gin.TestMode) } func createWorkspaceForTest(t *testing.T, router http.Handler, id string) { diff --git a/internal/model/acp.go b/internal/model/acp.go new file mode 100644 index 0000000..af20d58 --- /dev/null +++ b/internal/model/acp.go @@ -0,0 +1,38 @@ +package model + +import "time" + +// AcpStatusResponse is the API response for ACP status. +type AcpStatusResponse struct { + WorkspaceID string `json:"workspaceId"` + Ready bool `json:"ready"` + SessionID string `json:"sessionId,omitempty"` + Running bool `json:"running"` + PID int `json:"pid,omitempty"` + Error string `json:"error,omitempty"` +} + +// AcpMessage is a single role/text/time entry in the ACP history. +type AcpMessage struct { + Role string `json:"role"` + Text string `json:"text"` + Time time.Time `json:"time"` +} + +// AcpHistoryResponse is the API response for ACP history. +type AcpHistoryResponse struct { + SessionID string `json:"sessionId"` + Messages []AcpMessage `json:"messages"` +} + +// AcpPromptRequest is the body for POST /acp/prompt. +type AcpPromptRequest struct { + Content string `json:"content"` +} + +// AcpPromptResponse is the API response for a prompt turn. +type AcpPromptResponse struct { + SessionID string `json:"sessionId"` + StopReason string `json:"stopReason"` + Text string `json:"text"` +} diff --git a/internal/process/manager.go b/internal/process/manager.go index b9f17f5..92c4da7 100644 --- a/internal/process/manager.go +++ b/internal/process/manager.go @@ -36,10 +36,10 @@ type LocalManager struct { // NewManager creates a LocalManager with the given command. // If command is empty, it defaults to "opencode". -func NewManager(command string) *LocalManager { +func NewManager(command string, args []string) *LocalManager { return &LocalManager{ command: normalizeCommand(command), - args: []string{}, + args: args, sessions: make(map[string]*Session), } } @@ -267,4 +267,3 @@ func (m *LocalManager) exitOf(workspaceID string) (ExitInfo, error) { defer sess.mu.Unlock() return sess.Exit, nil } - diff --git a/internal/process/manager_test.go b/internal/process/manager_test.go index b3e41ac..84ca0c6 100644 --- a/internal/process/manager_test.go +++ b/internal/process/manager_test.go @@ -8,7 +8,7 @@ import ( ) func TestStatusBeforeStartIsNotRunning(t *testing.T) { - m := NewManager("opencode") + m := NewManager("opencode", nil) s := m.Status("ws1") if s.Running { t.Error("expected Running=false before start") @@ -19,7 +19,7 @@ func TestStatusBeforeStartIsNotRunning(t *testing.T) { } func TestStartRejectsDuplicateRunningSession(t *testing.T) { - m := NewManager("") + m := NewManager("", nil) // Override command for testing using the test binary itself. m.command = os.Args[0] m.args = []string{"-test.run=TestHelperProcess", "--"} @@ -43,7 +43,7 @@ func TestStartRejectsDuplicateRunningSession(t *testing.T) { } func TestStopMissingSessionIsNotFound(t *testing.T) { - m := NewManager("opencode") + m := NewManager("opencode", nil) err := m.Stop("nonexistent") if err == nil { t.Fatal("expected error stopping missing session") diff --git a/internal/service/acp_service.go b/internal/service/acp_service.go new file mode 100644 index 0000000..8578f4e --- /dev/null +++ b/internal/service/acp_service.go @@ -0,0 +1,88 @@ +package service + +import ( + "log/slog" + + "codespace/internal/acp" + "codespace/internal/model" + "codespace/internal/process" + "codespace/internal/workspace" +) + +// AcpService wraps the low-level acp.Service with logging and model mapping. +type AcpService struct { + svc acp.Service + lg *slog.Logger +} + +// NewAcpService creates an AcpService. +// If lg is nil, slog.Default() is used. +func NewAcpService(processes process.Manager, workspaces workspace.Manager, lg *slog.Logger) *AcpService { + if lg == nil { + lg = slog.Default() + } + return &AcpService{svc: acp.NewService(processes, workspaces, lg), lg: lg} +} + +// Status returns the ACP status for the workspace. +func (s *AcpService) Status(workspaceID string) (model.AcpStatusResponse, error) { + status, err := s.svc.Status(workspaceID) + if err != nil { + s.lg.Error("acp status failed", "workspace_id", workspaceID, "error", err) + return model.AcpStatusResponse{}, err + } + return model.AcpStatusResponse{ + WorkspaceID: status.WorkspaceID, + Ready: status.Ready, + SessionID: status.SessionID, + Running: status.Running, + PID: status.PID, + Error: status.Error, + }, nil +} + +// History returns the ACP conversation history for the workspace. +func (s *AcpService) History(workspaceID string) (model.AcpHistoryResponse, error) { + hist, err := s.svc.History(workspaceID) + if err != nil { + s.lg.Error("acp history failed", "workspace_id", workspaceID, "error", err) + return model.AcpHistoryResponse{}, err + } + messages := make([]model.AcpMessage, len(hist.Messages)) + for i, m := range hist.Messages { + messages[i] = model.AcpMessage{Role: m.Role, Text: m.Text, Time: m.Time} + } + return model.AcpHistoryResponse{SessionID: hist.SessionID, Messages: messages}, nil +} + +// Prompt sends a prompt to the agent and returns its response. +func (s *AcpService) Prompt(workspaceID string, content string) (model.AcpPromptResponse, error) { + res, err := s.svc.Prompt(workspaceID, content) + if err != nil { + s.lg.Error("acp prompt failed", "workspace_id", workspaceID, "error", err) + return model.AcpPromptResponse{}, err + } + return model.AcpPromptResponse{ + SessionID: res.SessionID, + StopReason: res.StopReason, + Text: res.Text, + }, nil +} + +// Cancel interrupts an in-flight prompt. +func (s *AcpService) Cancel(workspaceID string) error { + if err := s.svc.Cancel(workspaceID); err != nil { + s.lg.Error("acp cancel failed", "workspace_id", workspaceID, "error", err) + return err + } + return nil +} + +// EnsureReady ensures the ACP process is started and initialized. +func (s *AcpService) EnsureReady(workspaceID string) error { + if err := s.svc.EnsureReady(workspaceID); err != nil { + s.lg.Error("acp ensure ready failed", "workspace_id", workspaceID, "error", err) + return err + } + return nil +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 3b2c4e0..e218625 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -37,7 +37,8 @@ type WorkspaceConfig struct { // ProcessConfig holds OpenCode process settings. type ProcessConfig struct { - OpenCodeCommand string `yaml:"opencodeCommand"` + OpenCodeCommand string `yaml:"opencodeCommand"` + Args []string `yaml:"args"` } // ShellConfig holds interactive shell settings. @@ -93,7 +94,7 @@ func Default() Config { MaxHeaderBytes: 1 << 20, // 1 MiB }, Workspace: WorkspaceConfig{Root: "./workspaces"}, - Process: ProcessConfig{OpenCodeCommand: "opencode"}, + Process: ProcessConfig{OpenCodeCommand: "opencode", Args: []string{"acp"}}, Shell: ShellConfig{Command: "bash", Args: []string{"-i"}}, File: FileConfig{MaxWriteBytes: 1 << 20}, Gin: GinConfig{Mode: "release"}, @@ -150,6 +151,9 @@ func applyRaw(cfg *Config, raw rawConfig) error { if raw.Process.OpenCodeCommand != "" { cfg.Process.OpenCodeCommand = raw.Process.OpenCodeCommand } + if raw.Process.Args != nil { + cfg.Process.Args = raw.Process.Args + } cfg.Shell.Command = raw.Shell.Command cfg.Shell.Args = raw.Shell.Args if raw.File.MaxWriteBytes != 0 { @@ -211,6 +215,9 @@ func applyEnv(cfg *Config) error { if v := os.Getenv("CODESPACE_OPENCODE_COMMAND"); v != "" { cfg.Process.OpenCodeCommand = v } + if v := os.Getenv("CODESPACE_OPENCODE_ARGS"); v != "" { + cfg.Process.Args = strings.Split(v, ",") + } if v := os.Getenv("CODESPACE_SHELL_COMMAND"); v != "" { cfg.Shell.Command = v }