From cf314021e7f6461f439c8e457f301fdb776ff857 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Thu, 7 May 2026 12:04:27 +0800 Subject: [PATCH] feat(acp): JSON-RPC 2.0 codec (number/string id, LF-delimited, concurrent-safe encoder) --- internal/acp/jsonrpc.go | 201 +++++++++++++++++++++++++++++++++++ internal/acp/jsonrpc_test.go | 156 +++++++++++++++++++++++++++ 2 files changed, 357 insertions(+) create mode 100644 internal/acp/jsonrpc.go create mode 100644 internal/acp/jsonrpc_test.go diff --git a/internal/acp/jsonrpc.go b/internal/acp/jsonrpc.go new file mode 100644 index 0000000..b230b0e --- /dev/null +++ b/internal/acp/jsonrpc.go @@ -0,0 +1,201 @@ +// jsonrpc.go 实现 ACP 用到的最小 JSON-RPC 2.0 codec。spec §9.5 选定自写最小 +// 实现，理由：ACP 协议简单，无需引入完整 JSON-RPC 库。 +// +// 关键设计： +// - id 类型同时支持 number(int64) 和 string（spec §5.2 命名空间约定） +// - DecodeMessage 返回统一 *Message，调用方按 IsRequest/IsResponse/IsNotification 分发 +// - 单条消息长度上限：DefaultMaxMessageBytes（防 OOM） +package acp + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "io" + "sync" +) + +// JSONRPC 错误码（标准 + ACP 自定义） +const ( + JSONRPCParseError = -32700 + JSONRPCInvalidRequest = -32600 + JSONRPCMethodNotFound = -32601 + JSONRPCInvalidParams = -32602 + JSONRPCInternalError = -32603 + + // ACP 自定义（-32000 到 -32099 区间为 server 自定义） + ACPFsPathOutsideWorktree = -32001 +) + +// DefaultMaxMessageBytes 单条 JSON-RPC message 最大字节数（防 OOM）。 +const DefaultMaxMessageBytes = 8 * 1024 * 1024 // 8 MB + +// Message 是 JSON-RPC 2.0 envelope 统一表示。Request 含 ID + Method； +// Notification 含 Method 但 ID = nil；Response 含 ID + (Result 或 Error)。 +type Message struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` + Error *JSONRPCError `json:"error,omitempty"` +} + +// JSONRPCError 是 JSON-RPC 错误对象。 +type JSONRPCError struct { + Code int `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data,omitempty"` +} + +func (e *JSONRPCError) Error() string { + return fmt.Sprintf("jsonrpc error %d: %s", e.Code, e.Message) +} + +// IsRequest reports whether m is a request (has ID + Method). +func (m *Message) IsRequest() bool { + return m.Method != "" && len(m.ID) > 0 && m.Result == nil && m.Error == nil +} + +// IsNotification reports whether m is a notification (has Method but no ID). +func (m *Message) IsNotification() bool { + return m.Method != "" && len(m.ID) == 0 +} + +// IsResponse reports whether m is a response (has ID, no Method). +func (m *Message) IsResponse() bool { + return m.Method == "" && len(m.ID) > 0 +} + +// IDInt64 attempts to extract a numeric ID. Returns (n, true) if it's a JSON +// number; (0, false) otherwise. +func (m *Message) IDInt64() (int64, bool) { + if len(m.ID) == 0 { + return 0, false + } + var n int64 + if err := json.Unmarshal(m.ID, &n); err == nil { + return n, true + } + return 0, false +} + +// IDString attempts to extract a string ID. +func (m *Message) IDString() (string, bool) { + if len(m.ID) == 0 { + return "", false + } + var s string + if err := json.Unmarshal(m.ID, &s); err == nil { + return s, true + } + return "", false +} + +// Decoder reads newline-delimited JSON-RPC messages from a stream. +type Decoder struct { + r *bufio.Reader + maxSize int +} + +// NewDecoder constructs a Decoder. maxSize <= 0 → DefaultMaxMessageBytes. +func NewDecoder(r io.Reader, maxSize int) *Decoder { + if maxSize <= 0 { + maxSize = DefaultMaxMessageBytes + } + br := bufio.NewReaderSize(r, 64*1024) + return &Decoder{r: br, maxSize: maxSize} +} + +// DecodeMessage reads one LF-delimited JSON message. +func (d *Decoder) DecodeMessage() (*Message, error) { + line, err := d.r.ReadBytes('\n') + if err != nil { + if errors.Is(err, io.EOF) && len(line) == 0 { + return nil, io.EOF + } + if !errors.Is(err, io.EOF) { + return nil, err + } + } + if len(line) > d.maxSize { + return nil, fmt.Errorf("acp.jsonrpc: message exceeds max size %d", d.maxSize) + } + var m Message + if err := json.Unmarshal(line, &m); err != nil { + return nil, fmt.Errorf("acp.jsonrpc: parse: %w", err) + } + if m.JSONRPC != "2.0" { + return nil, fmt.Errorf("acp.jsonrpc: unsupported jsonrpc version %q", m.JSONRPC) + } + return &m, nil +} + +// Encoder serializes Messages to a writer with LF separator. +type Encoder struct { + w io.Writer + mu sync.Mutex +} + +// NewEncoder constructs an Encoder. Writes are mutex-protected so concurrent +// callers (e.g. response handler + scheduled prompt) don't interleave bytes. +func NewEncoder(w io.Writer) *Encoder { + return &Encoder{w: w} +} + +// Encode marshals m as JSON and writes it followed by '\n'. +func (e *Encoder) Encode(m *Message) error { + if m.JSONRPC == "" { + m.JSONRPC = "2.0" + } + b, err := json.Marshal(m) + if err != nil { + return fmt.Errorf("acp.jsonrpc: marshal: %w", err) + } + e.mu.Lock() + defer e.mu.Unlock() + if _, err := e.w.Write(b); err != nil { + return err + } + if _, err := e.w.Write([]byte{'\n'}); err != nil { + return err + } + return nil +} + +// NewRequestInt builds a Message with an integer id (server-initiated calls). +func NewRequestInt(id int64, method string, params any) (*Message, error) { + idJSON, err := json.Marshal(id) + if err != nil { + return nil, err + } + pb, err := json.Marshal(params) + if err != nil { + return nil, err + } + return &Message{JSONRPC: "2.0", ID: idJSON, Method: method, Params: pb}, nil +} + +// NewNotification builds a Message without id. +func NewNotification(method string, params any) (*Message, error) { + pb, err := json.Marshal(params) + if err != nil { + return nil, err + } + return &Message{JSONRPC: "2.0", Method: method, Params: pb}, nil +} + +// NewResponseOK builds a success response. +func NewResponseOK(id json.RawMessage, result any) (*Message, error) { + rb, err := json.Marshal(result) + if err != nil { + return nil, err + } + return &Message{JSONRPC: "2.0", ID: id, Result: rb}, nil +} + +// NewResponseErr builds an error response. +func NewResponseErr(id json.RawMessage, code int, msg string) *Message { + return &Message{JSONRPC: "2.0", ID: id, Error: &JSONRPCError{Code: code, Message: msg}} +} diff --git a/internal/acp/jsonrpc_test.go b/internal/acp/jsonrpc_test.go new file mode 100644 index 0000000..050e5b1 --- /dev/null +++ b/internal/acp/jsonrpc_test.go @@ -0,0 +1,156 @@ +package acp_test + +import ( + "bytes" + "encoding/json" + "io" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/yan1h/agent-coding-workflow/internal/acp" +) + +func TestJSONRPC_RoundtripRequestNumberID(t *testing.T) { + t.Parallel() + req, err := acp.NewRequestInt(42, "session/prompt", map[string]any{"text": "hi"}) + require.NoError(t, err) + + var buf bytes.Buffer + require.NoError(t, acp.NewEncoder(&buf).Encode(req)) + + dec := acp.NewDecoder(&buf, 0) + msg, err := dec.DecodeMessage() + require.NoError(t, err) + + assert.True(t, msg.IsRequest()) + assert.Equal(t, "session/prompt", msg.Method) + id, ok := msg.IDInt64() + require.True(t, ok) + assert.Equal(t, int64(42), id) +} + +func TestJSONRPC_RoundtripStringID(t *testing.T) { + t.Parallel() + idJSON, _ := json.Marshal("client-abc") + req := &acp.Message{JSONRPC: "2.0", ID: idJSON, Method: "session/prompt"} + + var buf bytes.Buffer + require.NoError(t, acp.NewEncoder(&buf).Encode(req)) + + dec := acp.NewDecoder(&buf, 0) + msg, _ := dec.DecodeMessage() + s, ok := msg.IDString() + require.True(t, ok) + assert.Equal(t, "client-abc", s) + + _, ok2 := msg.IDInt64() + assert.False(t, ok2) +} + +func TestJSONRPC_Notification_NoID(t *testing.T) { + t.Parallel() + notif, _ := acp.NewNotification("session/update", map[string]string{"x": "y"}) + + var buf bytes.Buffer + require.NoError(t, acp.NewEncoder(&buf).Encode(notif)) + + msg, _ := acp.NewDecoder(&buf, 0).DecodeMessage() + assert.True(t, msg.IsNotification()) + assert.False(t, msg.IsRequest()) + assert.False(t, msg.IsResponse()) +} + +func TestJSONRPC_Response(t *testing.T) { + t.Parallel() + idJSON, _ := json.Marshal(int64(7)) + resp, _ := acp.NewResponseOK(idJSON, map[string]string{"sessionId": "abc"}) + + var buf bytes.Buffer + require.NoError(t, acp.NewEncoder(&buf).Encode(resp)) + + msg, _ := acp.NewDecoder(&buf, 0).DecodeMessage() + assert.True(t, msg.IsResponse()) + + var result struct { + SessionID string `json:"sessionId"` + } + require.NoError(t, json.Unmarshal(msg.Result, &result)) + assert.Equal(t, "abc", result.SessionID) +} + +func TestJSONRPC_ErrorResponse(t *testing.T) { + t.Parallel() + idJSON, _ := json.Marshal("client-1") + resp := acp.NewResponseErr(idJSON, acp.JSONRPCMethodNotFound, "no such method") + + var buf bytes.Buffer + require.NoError(t, acp.NewEncoder(&buf).Encode(resp)) + + msg, _ := acp.NewDecoder(&buf, 0).DecodeMessage() + require.NotNil(t, msg.Error) + assert.Equal(t, acp.JSONRPCMethodNotFound, msg.Error.Code) + assert.Equal(t, "no such method", msg.Error.Message) +} + +func TestJSONRPC_DecoderEOF(t *testing.T) { + t.Parallel() + dec := acp.NewDecoder(bytes.NewReader(nil), 0) + _, err := dec.DecodeMessage() + assert.ErrorIs(t, err, io.EOF) +} + +func TestJSONRPC_DecoderMaxSize(t *testing.T) { + t.Parallel() + huge := make([]byte, 200) + for i := range huge { + huge[i] = 'a' + } + body := []byte(`{"jsonrpc":"2.0","method":"x","params":"`) + body = append(body, huge...) + body = append(body, []byte(`"}`+"\n")...) + + dec := acp.NewDecoder(bytes.NewReader(body), 100) + _, err := dec.DecodeMessage() + require.Error(t, err) + assert.Contains(t, err.Error(), "exceeds max size") +} + +func TestJSONRPC_DecoderBadJSON(t *testing.T) { + t.Parallel() + dec := acp.NewDecoder(bytes.NewReader([]byte("not-json\n")), 0) + _, err := dec.DecodeMessage() + require.Error(t, err) +} + +func TestJSONRPC_EncoderConcurrentSafe(t *testing.T) { + t.Parallel() + var buf bytes.Buffer + enc := acp.NewEncoder(&buf) + + const N = 100 + done := make(chan struct{}, N) + for i := 0; i < N; i++ { + go func(id int64) { + req, _ := acp.NewRequestInt(id, "m", nil) + _ = enc.Encode(req) + done <- struct{}{} + }(int64(i)) + } + for i := 0; i < N; i++ { + <-done + } + + dec := acp.NewDecoder(&buf, 0) + count := 0 + for { + _, err := dec.DecodeMessage() + if err == io.EOF { + break + } + require.NoError(t, err) + count++ + } + assert.Equal(t, N, count) +}