dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

fix: prevent duplicate messages when resuming ACP sessions (#392)

Browse Source
This commit is contained in:
João Salavisa
2026-03-08 18:13:11 +00:00
committed by GitHub
parent b48471e063
commit c70e97c104
1 changed files with 24 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go
+24 -8
View File
@@ -297,7 +297,9 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string) error {
defer span.End()
// Suppress history replay notifications during load.
// ACP agents stream the entire conversation history during session/load.
// ACP session/load replays the entire conversation history asynchronously.
// We set a flag to suppress these notifications to avoid duplicating messages in the database.
// The flag will be cleared when we send the next prompt (see Prompt method).
a.mu.Lock()
a.isLoadingSession = true
a.mu.Unlock()
@@ -308,11 +310,6 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string) error {
McpServers: []acp.McpServer{}, // MCPs configured via CLI args; empty satisfies the required field
})
// Clear the loading flag (even on error) before processing response
a.mu.Lock()
a.isLoadingSession = false
a.mu.Unlock()
if err != nil {
span.RecordError(err)
return fmt.Errorf("failed to load session: %w", err)
@@ -389,6 +386,18 @@ func (a *Adapter) Prompt(ctx context.Context, message string, attachments []v1.M
)
a.setPromptTraceCtx(promptCtx)
// Clear the loading flag before sending the prompt.
// If we're resuming a session, history replay is complete by the time we send a new prompt.
a.mu.Lock()
wasLoading := a.isLoadingSession
a.isLoadingSession = false
a.mu.Unlock()
if wasLoading {
a.logger.Info("cleared session load suppression flag before sending new prompt",
zap.String("session_id", sessionID))
}
a.logger.Info("sending prompt",
zap.String("session_id", sessionID),
zap.Int("content_blocks", len(contentBlocks)),
@@ -601,15 +610,22 @@ func (a *Adapter) handleACPUpdate(n acp.SessionNotification) {
// During session/load, suppress history replay notifications.
// ACP agents stream the entire conversation history during load, which should not
// be emitted as new message events to avoid duplicating messages in the UI.
// We suppress: message chunks, thinking chunks, tool calls, and tool updates.
a.mu.RLock()
isLoading := a.isLoadingSession
a.mu.RUnlock()
if isLoading {
u := n.Update
if u.AgentMessageChunk != nil || u.UserMessageChunk != nil || u.AgentThoughtChunk != nil {
// Suppress all conversation history events during load
if u.AgentMessageChunk != nil || u.UserMessageChunk != nil || u.AgentThoughtChunk != nil ||
u.ToolCall != nil || u.ToolCallUpdate != nil {
a.logger.Debug("suppressing history replay notification during session load",
zap.String("session_id", string(n.SessionId)))
zap.String("session_id", string(n.SessionId)),
zap.Bool("is_message", u.AgentMessageChunk != nil || u.UserMessageChunk != nil),
zap.Bool("is_thinking", u.AgentThoughtChunk != nil),
zap.Bool("is_tool_call", u.ToolCall != nil),
zap.Bool("is_tool_update", u.ToolCallUpdate != nil))
return
}
}