feat(acp): JSON-RPC 2.0 codec (number/string id, LF-delimited, concurrent-safe encoder)

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}}
+}

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)
+}