fix(acp): group agent chunks by messageId (backend) and coalesce consecutive same-role entries in the UI (frontend)
The opencode acp binary streams agent_message_chunk notifications with an EMPTY messageId. The backend was already correct (it groups when a messageId is present), but the UI rendered every chunk as its own bubble, so a single 'count 1 to 5' reply showed as 9 separate bubbles. Backend: - internal/acp/history.go: add MessageID to Message; new AppendOrCreate method that finds or creates an entry by messageId and appends text to it (preserving the original timestamp). - internal/acp/client.go: agent_message_chunk and user_message_chunk use AppendOrCreate. - internal/model/acp.go: expose MessageID in the wire response. - internal/service/acp_service.go: copy MessageID into the API DTO. - internal/acp/client_test.go: add TestHistoryAppendOrCreateGroupsByMessageID. Frontend: - web/src/components/acp/groupMessages.ts: new pure helper that coalesces consecutive same-role history entries into a single bubble (keeps the first chunk's time + messageId). - web/src/components/acp/AcpPanel.tsx: render coalesced messages; auto-scroll effect now depends on coalesced so it fires as the agent streams. Verified via e2e: 'count from 1 to 5' now returns a single agent entry with text '1\n2\n3\n4\n5' instead of 9 separate entries.
This commit is contained in:
tao.chen
@@ -238,13 +238,13 @@ func (c *Client) handleNotification(method string, params json.RawMessage) {
|
||||
switch up.Update.SessionUpdate {
|
||||
case "agent_message_chunk":
|
||||
if up.Update.Content.Type == "text" {
|
||||
c.history.Add("agent", up.Update.Content.Text)
|
||||
c.history.AppendOrCreate("agent", up.Update.MessageID, up.Update.Content.Text)
|
||||
} else {
|
||||
c.lg.Debug("acp drop non-text agent chunk", "workspace_id", c.workspaceID, "type", up.Update.Content.Type)
|
||||
}
|
||||
case "user_message_chunk":
|
||||
if up.Update.Content.Type == "text" {
|
||||
c.history.Add("user", up.Update.Content.Text)
|
||||
c.history.AppendOrCreate("user", up.Update.MessageID, up.Update.Content.Text)
|
||||
} else {
|
||||
c.lg.Debug("acp drop non-text user chunk", "workspace_id", c.workspaceID, "type", up.Update.Content.Type)
|
||||
}
|
||||
|
||||
@@ -276,3 +276,78 @@ func TestAgentRequestReadFile(t *testing.T) {
|
||||
t.Fatalf("content = %q, want 'package main\\n'", res.Content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHistoryAppendOrCreateGroupsByMessageID(t *testing.T) {
|
||||
h := NewHistory()
|
||||
|
||||
// 1. First chunk for a new messageId creates an entry.
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
beforeFirst := time.Now()
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
firstTime := h.AppendOrCreate("agent", "msg-1", "hello")
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
afterFirst := time.Now()
|
||||
|
||||
if firstTime.Before(beforeFirst) || firstTime.After(afterFirst) {
|
||||
t.Fatalf("first chunk timestamp %v out of range [%v, %v]", firstTime, beforeFirst, afterFirst)
|
||||
}
|
||||
msgs := h.List()
|
||||
if len(msgs) != 1 {
|
||||
t.Fatalf("after first chunk: len = %d, want 1", len(msgs))
|
||||
}
|
||||
if msgs[0].Role != "agent" || msgs[0].Text != "hello" || msgs[0].MessageID != "msg-1" {
|
||||
t.Fatalf("first chunk message = %+v, want role=agent text=hello messageId=msg-1", msgs[0])
|
||||
}
|
||||
|
||||
// 2. Second chunk for the same messageId appends and preserves timestamp.
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
secondTime := h.AppendOrCreate("agent", "msg-1", " world")
|
||||
if !secondTime.Equal(firstTime) {
|
||||
t.Fatalf("second chunk timestamp changed: got %v, want %v", secondTime, firstTime)
|
||||
}
|
||||
msgs = h.List()
|
||||
if len(msgs) != 1 {
|
||||
t.Fatalf("after second chunk: len = %d, want 1", len(msgs))
|
||||
}
|
||||
if msgs[0].Text != "hello world" {
|
||||
t.Fatalf("after second chunk: text = %q, want \"hello world\"", msgs[0].Text)
|
||||
}
|
||||
|
||||
// 3. A different messageId creates a new entry.
|
||||
h.AppendOrCreate("agent", "msg-2", "second")
|
||||
msgs = h.List()
|
||||
if len(msgs) != 2 {
|
||||
t.Fatalf("after different id: len = %d, want 2", len(msgs))
|
||||
}
|
||||
if msgs[1].Text != "second" || msgs[1].MessageID != "msg-2" {
|
||||
t.Fatalf("second message = %+v, want text=second messageId=msg-2", msgs[1])
|
||||
}
|
||||
|
||||
// 4. Empty messageId always appends a new entry and never collides.
|
||||
h.AppendOrCreate("user", "", "a")
|
||||
h.AppendOrCreate("user", "", "b")
|
||||
msgs = h.List()
|
||||
if len(msgs) != 4 {
|
||||
t.Fatalf("after two empty-id chunks: len = %d, want 4", len(msgs))
|
||||
}
|
||||
if msgs[2].Text != "a" || msgs[2].MessageID != "" {
|
||||
t.Fatalf("empty-id message[2] = %+v, want text=a messageId=\"\"", msgs[2])
|
||||
}
|
||||
if msgs[3].Text != "b" || msgs[3].MessageID != "" {
|
||||
t.Fatalf("empty-id message[3] = %+v, want text=b messageId=\"\"", msgs[3])
|
||||
}
|
||||
|
||||
// 5. Clear resets the index map so the same id can be reused.
|
||||
h.Clear()
|
||||
if h.Len() != 0 {
|
||||
t.Fatalf("after Clear: len = %d, want 0", h.Len())
|
||||
}
|
||||
h.AppendOrCreate("agent", "msg-1", "after clear")
|
||||
msgs = h.List()
|
||||
if len(msgs) != 1 {
|
||||
t.Fatalf("after clear + chunk: len = %d, want 1", len(msgs))
|
||||
}
|
||||
if msgs[0].Text != "after clear" || msgs[0].MessageID != "msg-1" {
|
||||
t.Fatalf("after clear message = %+v, want text=\"after clear\" messageId=msg-1", msgs[0])
|
||||
}
|
||||
}
|
||||
|
||||
+33
-8
@@ -7,27 +7,51 @@ import (
|
||||
|
||||
// Message is one entry in the conversation history.
|
||||
type Message struct {
|
||||
Role string `json:"role"`
|
||||
Text string `json:"text"`
|
||||
Time time.Time `json:"time"`
|
||||
Role string `json:"role"`
|
||||
Text string `json:"text"`
|
||||
Time time.Time `json:"time"`
|
||||
MessageID string `json:"messageId,omitempty"`
|
||||
}
|
||||
|
||||
// History accumulates prompt/response messages in memory for a session.
|
||||
type History struct {
|
||||
mu sync.RWMutex
|
||||
messages []Message
|
||||
mu sync.RWMutex
|
||||
messages []Message
|
||||
indexByID map[string]int
|
||||
}
|
||||
|
||||
// NewHistory creates an empty History.
|
||||
func NewHistory() *History {
|
||||
return &History{}
|
||||
return &History{indexByID: make(map[string]int)}
|
||||
}
|
||||
|
||||
// Add appends a message with the current time.
|
||||
// Add appends a message with the current time. It is used for messages that
|
||||
// have no messageId (e.g. the user prompt path).
|
||||
func (h *History) Add(role, text string) {
|
||||
h.AppendOrCreate(role, "", text)
|
||||
}
|
||||
|
||||
// AppendOrCreate finds the existing entry with the same messageId and appends
|
||||
// text to it; otherwise creates a new entry. When messageId is empty this
|
||||
// behaves like Add(role, text). It returns the time of the created or existing
|
||||
// entry.
|
||||
func (h *History) AppendOrCreate(role, messageId, text string) time.Time {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.messages = append(h.messages, Message{Role: role, Text: text, Time: time.Now()})
|
||||
|
||||
if messageId == "" {
|
||||
h.messages = append(h.messages, Message{Role: role, Text: text, Time: time.Now()})
|
||||
return h.messages[len(h.messages)-1].Time
|
||||
}
|
||||
|
||||
if idx, ok := h.indexByID[messageId]; ok {
|
||||
h.messages[idx].Text += text
|
||||
return h.messages[idx].Time
|
||||
}
|
||||
|
||||
h.messages = append(h.messages, Message{Role: role, Text: text, Time: time.Now(), MessageID: messageId})
|
||||
h.indexByID[messageId] = len(h.messages) - 1
|
||||
return h.messages[len(h.messages)-1].Time
|
||||
}
|
||||
|
||||
// List returns a copy of all messages, oldest first.
|
||||
@@ -66,4 +90,5 @@ func (h *History) Clear() {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.messages = h.messages[:0]
|
||||
h.indexByID = make(map[string]int)
|
||||
}
|
||||
|
||||
@@ -14,9 +14,10 @@ type AcpStatusResponse struct {
|
||||
|
||||
// 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"`
|
||||
Role string `json:"role"`
|
||||
Text string `json:"text"`
|
||||
Time time.Time `json:"time"`
|
||||
MessageID string `json:"messageId,omitempty"`
|
||||
}
|
||||
|
||||
// AcpHistoryResponse is the API response for ACP history.
|
||||
|
||||
@@ -50,7 +50,7 @@ func (s *AcpService) History(workspaceID string) (model.AcpHistoryResponse, erro
|
||||
}
|
||||
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}
|
||||
messages[i] = model.AcpMessage{Role: m.Role, Text: m.Text, Time: m.Time, MessageID: m.MessageID}
|
||||
}
|
||||
return model.AcpHistoryResponse{SessionID: hist.SessionID, Messages: messages}, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user