feat(backend): add opencode ACP (Agent Client Protocol) client
Adds a minimal but real ACP stack for the opencode process: - pkg/config: process.args default ["acp"] (opencodeCommand still "opencode") - internal/process: NewManager(command, args) — exec.Command uses args - internal/acp (new): NDJSON transport + JSON-RPC client over the existing process stdio. Implements initialize / session/new / session/prompt / session/cancel. Serves fs/read_text_file and fs/write_text_file from the workspace's fs.FileSystem. terminal/* requests get MethodNotFound. - internal/service/acp_service: per-workspace Client + mutex; starts the process on first prompt; transparently re-init on restart. - internal/api/acp_handler: GET /acp/status, GET /acp/history, POST /acp/prompt, POST /acp/cancel. - internal/model/acp: API DTOs. - internal/acp/client_test: NDJSON split-lines, request/response correlation, notification dispatch, agent-initiated request handling (fs + terminal). Existing process WS endpoint and Shell subsystem are unchanged. Conversation: 019f354c-a51b-7ec3-83ad-c647e9b50b19
This commit is contained in:
tao.chen
@@ -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()}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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]
|
||||
}
|
||||
@@ -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{}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user