fix: coalesce streamed agent plan messages (#3764)

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Kandev Agent <agent@kandev.dev>

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools.go

@@ -819,6 +819,7 @@ func (a *Adapter) convertToolCallResultUpdate(sessionID string, tcu *acp.Session
 				a.sendUpdate(AgentEvent{
 					Type:        streams.EventTypeAgentPlan,
 					SessionID:   sessionID,
+					ToolCallID:  emittedToolCallID,
 					PlanContent: planContent,
 				})
 			}

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/coder/acp-go-sdk"
+	"github.com/kandev/kandev/internal/agentctl/types/streams"
 )
 
 func TestLocationsArgsFromACP(t *testing.T) {
@@ -53,6 +54,25 @@ func TestToolCallUpdateSupplemental(t *testing.T) {
 	}
 }
 
+// @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.1
+func TestConvertToolCallResultUpdateAgentPlanCarriesToolCallID(t *testing.T) {
+	adapter := newTestAdapter()
+	t.Cleanup(func() { _ = adapter.Close() })
+
+	adapter.convertToolCallResultUpdate("session-1", &acp.SessionToolCallUpdate{
+		ToolCallId: "plan-call-1",
+		RawInput:   map[string]any{"plan": "# Plan\n\n1. Read"},
+	})
+
+	event := <-adapter.updatesCh
+	if event.Type != streams.EventTypeAgentPlan {
+		t.Fatalf("event type = %q, want %q", event.Type, streams.EventTypeAgentPlan)
+	}
+	if event.ToolCallID != "plan-call-1" {
+		t.Fatalf("tool call id = %q, want plan-call-1", event.ToolCallID)
+	}
+}
+
 func TestPathFromLocationSlice(t *testing.T) {
 	tests := []struct {
 		name  string

apps/backend/internal/backendapp/adapters.go

@@ -1431,6 +1431,15 @@ func (a *messageCreatorAdapter) UpdateToolCallMessage(ctx context.Context, taskI
 	return a.svc.UpdateToolCallMessageWithCreate(ctx, agentSessionID, toolCallID, parentToolCallID, status, result, title, normalized, taskID, turnID, msgType)
 }
 
+func (a *messageCreatorAdapter) UpsertAgentPlanMessage(
+	ctx context.Context,
+	taskID, sourceToolCallID, agentSessionID, content, turnID string,
+) error {
+	return a.svc.UpsertAgentPlanMessage(
+		ctx, taskID, sourceToolCallID, agentSessionID, content, turnID,
+	)
+}
+
 // CreateSessionMessage creates a message for non-chat session updates (status/progress/error/etc).
 func (a *messageCreatorAdapter) CreateSessionMessage(ctx context.Context, taskID, content, agentSessionID, messageType, turnID string, metadata map[string]interface{}, requestsInput bool) error {
 	_, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{

apps/backend/internal/integration/test_server_test.go

@@ -216,6 +216,15 @@ func (a *testMessageCreatorAdapter) UpdateToolCallMessage(ctx context.Context, t
 	return a.svc.UpdateToolCallMessageWithCreate(ctx, agentSessionID, toolCallID, parentToolCallID, status, result, title, normalized, taskID, turnID, msgType)
 }
 
+func (a *testMessageCreatorAdapter) UpsertAgentPlanMessage(
+	ctx context.Context,
+	taskID, sourceToolCallID, agentSessionID, content, turnID string,
+) error {
+	return a.svc.UpsertAgentPlanMessage(
+		ctx, taskID, sourceToolCallID, agentSessionID, content, turnID,
+	)
+}
+
 func (a *testMessageCreatorAdapter) CreateSessionMessage(ctx context.Context, taskID, content, agentSessionID, messageType, turnID string, metadata map[string]interface{}, requestsInput bool) error {
 	_, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{
 		TaskSessionID: agentSessionID,

apps/backend/internal/orchestrator/event_handlers_streaming.go

@@ -3282,18 +3282,29 @@ func (s *Service) handleOfficeTurnComplete(
 	return true
 }
 
-// handleAgentPlanEvent handles agent_plan events from tool calls (e.g. ExitPlanMode)
-// and creates a dedicated agent_plan message in the session.
+// handleAgentPlanEvent handles agent_plan events from tool calls (e.g. ExitPlanMode).
 func (s *Service) handleAgentPlanEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) {
 	if payload.SessionID == "" || payload.Data.PlanContent == "" || s.messageCreator == nil {
 		return
 	}
 	sessionID := payload.SessionID
-	if err := s.messageCreator.CreateSessionMessage(
-		ctx, payload.TaskID, payload.Data.PlanContent, sessionID,
-		string(models.MessageTypeAgentPlan), s.getActiveTurnID(sessionID), nil, false,
+	turnID := s.getActiveTurnID(sessionID)
+	if payload.Data.ToolCallID == "" {
+		if err := s.messageCreator.CreateSessionMessage(
+			ctx, payload.TaskID, payload.Data.PlanContent, sessionID,
+			string(models.MessageTypeAgentPlan), turnID, nil, false,
+		); err != nil {
+			s.logger.Error("failed to create uncorrelated agent plan message",
+				zap.String("task_id", payload.TaskID),
+				zap.String("session_id", sessionID),
+				zap.Error(err))
+		}
+		return
+	}
+	if err := s.messageCreator.UpsertAgentPlanMessage(
+		ctx, payload.TaskID, payload.Data.ToolCallID, sessionID, payload.Data.PlanContent, turnID,
 	); err != nil {
-		s.logger.Error("failed to create agent plan message",
+		s.logger.Error("failed to upsert agent plan message",
 			zap.String("task_id", payload.TaskID),
 			zap.String("session_id", sessionID),
 			zap.Error(err))

apps/backend/internal/orchestrator/event_handlers_streaming_test.go

@@ -311,6 +311,15 @@ func (m *serviceBackedMessageCreator) UpdateToolCallMessage(
 	)
 }
 
+func (m *serviceBackedMessageCreator) UpsertAgentPlanMessage(
+	ctx context.Context,
+	taskID, sourceToolCallID, agentSessionID, content, turnID string,
+) error {
+	return m.svc.UpsertAgentPlanMessage(
+		ctx, taskID, sourceToolCallID, agentSessionID, content, turnID,
+	)
+}
+
 func newServiceBackedMessageCreator(repo *sqliterepo.Repository) *serviceBackedMessageCreator {
 	services := taskservice.NewService(taskservice.Repos{
 		Workspaces:       repo,
@@ -377,6 +386,28 @@ func (b *recordingEventBus) Request(context.Context, string, *bus.Event, time.Du
 func (b *recordingEventBus) Close()            {}
 func (b *recordingEventBus) IsConnected() bool { return true }
 
+// @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.1
+func TestHandleAgentPlanEventUsesCorrelatedMessageUpdate(t *testing.T) {
+	messages := &mockMessageCreator{}
+	service := &Service{messageCreator: messages, logger: testLogger()}
+	service.activeTurns.Store("session-plan", "turn-plan")
+
+	service.handleAgentPlanEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
+		TaskID:    "task-plan",
+		SessionID: "session-plan",
+		Data: &lifecycle.AgentStreamEventData{
+			ToolCallID:  "call-plan",
+			PlanContent: "# Plan\n\n1. Read",
+		},
+	})
+
+	require.Zero(t, messages.sessionMessageAttempts)
+	require.Zero(t, messages.toolUpdateWrites)
+	require.Equal(t, 1, messages.agentPlanUpserts)
+	require.Equal(t, "call-plan", messages.lastAgentPlanToolCallID)
+	require.Equal(t, "# Plan\n\n1. Read", messages.lastAgentPlanContent)
+}
+
 func TestUpdateTaskSessionStatePublishesPersistedUpdatedAt(t *testing.T) {
 	ctx := context.Background()
 	repo := setupTestRepo(t)

apps/backend/internal/orchestrator/service.go

@@ -123,6 +123,7 @@ type MessageCreator interface {
 	// normalized contains the typed tool payload data.
 	// parentToolCallID is the parent Task tool call ID for subagent nesting (empty for top-level).
 	UpdateToolCallMessage(ctx context.Context, taskID, toolCallID, parentToolCallID, status, result, agentSessionID, title, turnID, msgType string, normalized *streams.NormalizedPayload) error
+	UpsertAgentPlanMessage(ctx context.Context, taskID, sourceToolCallID, agentSessionID, content, turnID string) error
 	CreateSessionMessage(ctx context.Context, taskID, content, agentSessionID, messageType, turnID string, metadata map[string]interface{}, requestsInput bool) error
 	CreateSessionMessageIdempotent(ctx context.Context, messageID, taskID, content, agentSessionID, messageType, turnID string, metadata map[string]interface{}, requestsInput bool) error
 	CreatePermissionRequestMessage(ctx context.Context, taskID, sessionID, requestID, pendingID, toolCallID, title, turnID string, options []map[string]interface{}, actionType string, actionDetails map[string]interface{}) (string, error)

apps/backend/internal/orchestrator/task_operations_test.go

@@ -4753,6 +4753,12 @@ type mockMessageCreator struct {
 	thinkingWrites            int
 	toolCallWrites            int
 	toolUpdateWrites          int
+	lastToolUpdateID          string
+	lastToolUpdateTitle       string
+	lastToolUpdateType        string
+	agentPlanUpserts          int
+	lastAgentPlanToolCallID   string
+	lastAgentPlanContent      string
 	userMessageErr            error
 	idempotentUserMessages    map[string]struct{}
 	permissionClaimFn         func(context.Context, models.PermissionResolutionClaimRequest) (*models.PermissionResolutionClaimResult, error)
@@ -4816,8 +4822,25 @@ func (m *mockMessageCreator) CreateToolCallMessage(context.Context, string, stri
 	return nil
 }
 
-func (m *mockMessageCreator) UpdateToolCallMessage(context.Context, string, string, string, string, string, string, string, string, string, *streams.NormalizedPayload) error {
+func (m *mockMessageCreator) UpdateToolCallMessage(
+	_ context.Context,
+	_, toolCallID, _, _, _, _, title, _, msgType string,
+	_ *streams.NormalizedPayload,
+) error {
 	m.toolUpdateWrites++
+	m.lastToolUpdateID = toolCallID
+	m.lastToolUpdateTitle = title
+	m.lastToolUpdateType = msgType
+	return nil
+}
+
+func (m *mockMessageCreator) UpsertAgentPlanMessage(
+	_ context.Context,
+	_, sourceToolCallID, _, content, _ string,
+) error {
+	m.agentPlanUpserts++
+	m.lastAgentPlanToolCallID = sourceToolCallID
+	m.lastAgentPlanContent = content
 	return nil
 }
 

apps/backend/internal/task/repository/repoerrors/errors.go

@@ -23,6 +23,10 @@ var ErrInitialTaskBriefStale = errors.New("initial task brief is stale")
 // ErrMessageNotFound reports that no message row matched the supplied id.
 var ErrMessageNotFound = errors.New("message not found")
 
+// ErrMessageIdentityConflict reports that a deterministic message id already
+// belongs to a different immutable message identity.
+var ErrMessageIdentityConflict = errors.New("message identity conflict")
+
 // ErrTaskParentMismatch reports that a task no longer has the parent/workspace
 // relation a cross-task mutation was authorized against.
 var ErrTaskParentMismatch = errors.New("task parent relation no longer matches")

apps/backend/internal/task/repository/sqlite/base.go

@@ -138,6 +138,9 @@ type Repository struct {
 	// step instead of returning as if there were none. Nil in production
 	// and in every test but the one that sets it.
 	taskRowReconfirmHook func()
+	// agentPlanUpsertAfterRead is a test-only synchronization seam used to
+	// pause a plan upsert while its identity lock and transaction are held.
+	agentPlanUpsertAfterRead func()
 	// stepEntryDispatcher fires a step's session-independent on_enter
 	// sequence after a registered step-transition writer commits. Nil-safe
 	// (see dispatchStepEntry in step_entry_dispatch.go): unset in every

apps/backend/internal/task/repository/sqlite/message_agent_plan.go

@@ -0,0 +1,181 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/jmoiron/sqlx"
+
+	"github.com/kandev/kandev/internal/db/dialect"
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/repository/repoerrors"
+)
+
+const agentPlanMessageLockNamespace = "agent-plan-message:"
+
+// UpsertAgentPlanMessageWithConversationReceipt serializes the complete
+// read/create/update decision for one deterministic agent-plan message.
+func (r *Repository) UpsertAgentPlanMessageWithConversationReceipt(
+	ctx context.Context,
+	message *models.Message,
+) (*models.Message, *models.ConversationMutationReceipt, bool, error) {
+	if message == nil || message.ID == "" {
+		return nil, nil, false, errors.New("agent plan message identity is required")
+	}
+	metadataJSON, err := json.Marshal(message.Metadata)
+	if err != nil {
+		return nil, nil, false, fmt.Errorf("serialize agent plan metadata: %w", err)
+	}
+	tx, err := r.db.BeginTxx(ctx, nil)
+	if err != nil {
+		return nil, nil, false, fmt.Errorf("begin agent plan upsert: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+	if err := r.lockAgentPlanMessageIdentity(ctx, tx, message); err != nil {
+		return nil, nil, false, err
+	}
+	base, err := r.prepareConversationMutation(ctx, tx, message.TaskSessionID)
+	if err != nil {
+		return nil, nil, false, err
+	}
+
+	existing, err := r.readConversationMessageTx(ctx, tx, message.ID)
+	if errors.Is(err, sql.ErrNoRows) {
+		return r.insertAgentPlanMessage(ctx, tx, base, message, string(metadataJSON))
+	}
+	if err != nil {
+		return nil, nil, false, fmt.Errorf("read agent plan message: %w", err)
+	}
+	if r.agentPlanUpsertAfterRead != nil {
+		r.agentPlanUpsertAfterRead()
+	}
+	return r.updateAgentPlanMessage(ctx, tx, base, existing, message)
+}
+
+func (r *Repository) updateAgentPlanMessage(
+	ctx context.Context,
+	tx *sqlx.Tx,
+	base int64,
+	existing, incoming *models.Message,
+) (*models.Message, *models.ConversationMutationReceipt, bool, error) {
+	if !sameAgentPlanMessageIdentity(existing, incoming) {
+		return nil, nil, false, repoerrors.ErrMessageIdentityConflict
+	}
+	if existing.Content == incoming.Content {
+		return existing, nil, false, nil
+	}
+
+	existing.Content = incoming.Content
+	existing.UpdatedAt = r.nowUTC()
+	existingMetadata, err := json.Marshal(existing.Metadata)
+	if err != nil {
+		return nil, nil, false, fmt.Errorf("serialize stored agent plan metadata: %w", err)
+	}
+	requestsInput := 0
+	if existing.RequestsInput {
+		requestsInput = 1
+	}
+	if err := r.updateMessageWithPayloadGuardTx(ctx, tx, existing, existingMetadata, requestsInput); err != nil {
+		return nil, nil, false, err
+	}
+	receipt := &models.ConversationMutationReceipt{}
+	if err := r.populateConversationMessageReceipt(
+		ctx, tx, receipt, base, existing, models.ConversationMutationUpsert,
+	); err != nil {
+		return nil, nil, false, err
+	}
+	if err := r.finishConversationMutation(tx); err != nil {
+		return nil, nil, false, err
+	}
+	return existing, receipt, false, nil
+}
+
+func (r *Repository) lockAgentPlanMessageIdentity(
+	ctx context.Context,
+	tx *sqlx.Tx,
+	message *models.Message,
+) error {
+	if dialect.IsPostgres(r.db.DriverName()) {
+		if _, err := tx.ExecContext(
+			ctx,
+			`SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`,
+			agentPlanMessageLockNamespace+message.ID,
+		); err != nil {
+			return fmt.Errorf("lock agent plan message %q: %w", message.ID, err)
+		}
+		return nil
+	}
+	result, err := tx.ExecContext(
+		ctx,
+		tx.Rebind(`UPDATE task_sessions SET id = id WHERE id = ?`),
+		message.TaskSessionID,
+	)
+	if err != nil {
+		return fmt.Errorf("lock agent plan session %q: %w", message.TaskSessionID, err)
+	}
+	if affected, err := result.RowsAffected(); err != nil {
+		return err
+	} else if affected == 0 {
+		return fmt.Errorf("task session %s not found", message.TaskSessionID)
+	}
+	return nil
+}
+
+func (r *Repository) insertAgentPlanMessage(
+	ctx context.Context,
+	tx *sqlx.Tx,
+	base int64,
+	message *models.Message,
+	metadataJSON string,
+) (*models.Message, *models.ConversationMutationReceipt, bool, error) {
+	now := r.nowUTC()
+	if message.CreatedAt.IsZero() {
+		message.CreatedAt = now
+	}
+	if message.UpdatedAt.IsZero() {
+		message.UpdatedAt = message.CreatedAt
+	}
+	requestsInput := 0
+	if message.RequestsInput {
+		requestsInput = 1
+	}
+	if err := r.insertMessageWithPayloadGuardTx(
+		ctx, tx, message, requestsInput, string(message.Type), metadataJSON,
+	); err != nil {
+		return nil, nil, false, err
+	}
+	receipt := &models.ConversationMutationReceipt{}
+	if err := r.populateConversationMessageReceipt(
+		ctx, tx, receipt, base, message, models.ConversationMutationUpsert,
+	); err != nil {
+		return nil, nil, false, err
+	}
+	if err := r.finishConversationMutation(tx); err != nil {
+		return nil, nil, false, err
+	}
+	return message, receipt, true, nil
+}
+
+func sameAgentPlanMessageIdentity(existing, incoming *models.Message) bool {
+	if existing.ID != incoming.ID ||
+		existing.TaskSessionID != incoming.TaskSessionID ||
+		existing.TaskID != incoming.TaskID ||
+		existing.TurnID != incoming.TurnID ||
+		existing.AuthorType != incoming.AuthorType ||
+		existing.AuthorID != incoming.AuthorID ||
+		existing.Type != incoming.Type ||
+		existing.RequestsInput != incoming.RequestsInput {
+		return false
+	}
+	for _, key := range []string{"tool_call_id", "agent_plan_tool_call_id"} {
+		existingValue, existingOK := existing.Metadata[key].(string)
+		incomingValue, incomingOK := incoming.Metadata[key].(string)
+		if !existingOK || !incomingOK || existingValue != incomingValue {
+			return false
+		}
+	}
+	return true
+}

apps/backend/internal/task/repository/sqlite/message_agent_plan_postgres_test.go

@@ -0,0 +1,116 @@
+package sqlite
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/testutil"
+)
+
+func TestPostgresAgentPlanUpsertSerializesAcrossConnections(t *testing.T) {
+	dsn := testutil.PostgresDSNFromEnv(t)
+	repoA := openPostgresRepo(t)
+	secondDB := openSecondPostgresConnection(t, dsn, repoA.db)
+	repoB, err := NewWithDB(secondDB, secondDB, nil)
+	if err != nil {
+		t.Fatalf("new second postgres repo: %v", err)
+	}
+	observerDB := openSecondPostgresConnection(t, dsn, repoA.db)
+	observer, err := NewWithDB(observerDB, observerDB, nil)
+	if err != nil {
+		t.Fatalf("new observer postgres repo: %v", err)
+	}
+	ctx := context.Background()
+	seedPostgresSession(
+		t, repoA, "task-agent-plan-pg", "session-agent-plan-pg", "turn-agent-plan-pg", time.Now().UTC(),
+	)
+
+	initial := postgresAgentPlanMessage("initial snapshot")
+	if _, _, created, err := repoA.UpsertAgentPlanMessageWithConversationReceipt(ctx, initial); err != nil {
+		t.Fatalf("create initial agent plan: %v", err)
+	} else if !created {
+		t.Fatal("initial agent plan was not created")
+	}
+
+	olderRead := make(chan struct{})
+	releaseOlder := make(chan struct{})
+	var releaseOnce sync.Once
+	release := func() { releaseOnce.Do(func() { close(releaseOlder) }) }
+	t.Cleanup(release)
+	repoA.agentPlanUpsertAfterRead = func() {
+		close(olderRead)
+		<-releaseOlder
+	}
+
+	type upsertResult struct {
+		created bool
+		err     error
+	}
+	olderDone := make(chan upsertResult, 1)
+	go func() {
+		_, _, created, upsertErr := repoA.UpsertAgentPlanMessageWithConversationReceipt(
+			ctx, postgresAgentPlanMessage("older snapshot"),
+		)
+		olderDone <- upsertResult{created: created, err: upsertErr}
+	}()
+	select {
+	case <-olderRead:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for older upsert to read")
+	}
+
+	newerDone := make(chan upsertResult, 1)
+	go func() {
+		_, _, created, upsertErr := repoB.UpsertAgentPlanMessageWithConversationReceipt(
+			ctx, postgresAgentPlanMessage("newer snapshot"),
+		)
+		newerDone <- upsertResult{created: created, err: upsertErr}
+	}()
+	waitForPostgresLockWait(t, ctx, observer, "pg_advisory_xact_lock")
+	release()
+
+	for name, done := range map[string]<-chan upsertResult{
+		"older": olderDone,
+		"newer": newerDone,
+	} {
+		select {
+		case result := <-done:
+			if result.err != nil {
+				t.Fatalf("%s upsert: %v", name, result.err)
+			}
+			if result.created {
+				t.Fatalf("%s upsert unexpectedly created a row", name)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("timed out waiting for %s upsert", name)
+		}
+	}
+
+	stored, err := repoA.GetMessage(ctx, initial.ID)
+	if err != nil {
+		t.Fatalf("read final agent plan: %v", err)
+	}
+	if stored.Content != "newer snapshot" {
+		t.Fatalf("final agent plan = %q, want newer snapshot", stored.Content)
+	}
+}
+
+func postgresAgentPlanMessage(content string) *models.Message {
+	const messageID = "47f568f5-c467-51cc-b63c-71fd2253ed93"
+	return &models.Message{
+		ID:            messageID,
+		TaskSessionID: "session-agent-plan-pg",
+		TaskID:        "task-agent-plan-pg",
+		TurnID:        "turn-agent-plan-pg",
+		AuthorType:    models.MessageAuthorAgent,
+		Content:       content,
+		Type:          models.MessageTypeAgentPlan,
+		Metadata: map[string]interface{}{
+			"tool_call_id":            "agent-plan:" + messageID,
+			"agent_plan_tool_call_id": "source-agent-plan-pg",
+		},
+	}
+}

apps/backend/internal/task/service/service_agent_plan_messages_test.go

@@ -0,0 +1,215 @@
+package service
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/repository"
+	"github.com/kandev/kandev/internal/task/repository/plancommenttx"
+)
+
+type agentPlanRaceRepository struct {
+	repository.MessageRepository
+	receiptWriter conversationMessageReceiptWriter
+
+	admissionMu   sync.Mutex
+	stateMu       sync.Mutex
+	admissionHeld bool
+
+	olderUpdateReady chan struct{}
+	releaseOlder     chan struct{}
+	newerUpdateDone  chan struct{}
+}
+
+func newAgentPlanRaceRepository(base repository.MessageRepository) *agentPlanRaceRepository {
+	return &agentPlanRaceRepository{
+		MessageRepository: base,
+		receiptWriter:     base.(conversationMessageReceiptWriter),
+		olderUpdateReady:  make(chan struct{}),
+		releaseOlder:      make(chan struct{}),
+		newerUpdateDone:   make(chan struct{}),
+	}
+}
+
+func (r *agentPlanRaceRepository) AcquireMessageAdmission(
+	ctx context.Context,
+	messageID string,
+) (context.Context, func(), error) {
+	key := "message:" + messageID
+	if plancommenttx.AdmissionLeaseHeld(ctx, key) {
+		return ctx, func() {}, nil
+	}
+	r.admissionMu.Lock()
+	r.stateMu.Lock()
+	r.admissionHeld = true
+	r.stateMu.Unlock()
+	return plancommenttx.WithAdmissionLease(ctx, key), func() {
+		r.stateMu.Lock()
+		r.admissionHeld = false
+		r.stateMu.Unlock()
+		r.admissionMu.Unlock()
+	}, nil
+}
+
+func (r *agentPlanRaceRepository) CreateMessageWithConversationReceipt(
+	ctx context.Context,
+	message *models.Message,
+) (*models.ConversationMutationReceipt, error) {
+	return r.receiptWriter.CreateMessageWithConversationReceipt(ctx, message)
+}
+
+func (r *agentPlanRaceRepository) UpdateMessageWithConversationReceipt(
+	ctx context.Context,
+	message *models.Message,
+) (*models.ConversationMutationReceipt, error) {
+	if message.Content == "older snapshot" {
+		close(r.olderUpdateReady)
+		<-r.releaseOlder
+	}
+	receipt, err := r.receiptWriter.UpdateMessageWithConversationReceipt(ctx, message)
+	if message.Content == "newer snapshot" {
+		close(r.newerUpdateDone)
+	}
+	return receipt, err
+}
+
+func (r *agentPlanRaceRepository) DeleteMessageWithConversationReceipt(
+	ctx context.Context,
+	messageID string,
+) (*models.ConversationMutationReceipt, error) {
+	return r.receiptWriter.DeleteMessageWithConversationReceipt(ctx, messageID)
+}
+
+func (r *agentPlanRaceRepository) isAdmissionHeld() bool {
+	r.stateMu.Lock()
+	defer r.stateMu.Unlock()
+	return r.admissionHeld
+}
+
+// @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.1
+// @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.2
+func TestUpsertAgentPlanMessageCoalescesSnapshotsAndSeparatesToolCalls(t *testing.T) {
+	service, eventBus, repo := newMessageTestService(t)
+	ctx := context.Background()
+
+	for _, snapshot := range []string{"# Plan\n\n1. Read", "# Plan\n\n1. Read\n2. Write"} {
+		err := service.UpsertAgentPlanMessage(
+			ctx, "task-msg", "source-call-1", "sess-msg", snapshot, "",
+		)
+		if err != nil {
+			t.Fatalf("upsert correlated plan: %v", err)
+		}
+	}
+	publishedBeforeReplay := len(eventBus.GetPublishedEvents())
+	if err := service.UpsertAgentPlanMessage(
+		ctx, "task-msg", "source-call-1", "sess-msg", "# Plan\n\n1. Read\n2. Write", "",
+	); err != nil {
+		t.Fatalf("replay identical plan: %v", err)
+	}
+	if published := len(eventBus.GetPublishedEvents()); published != publishedBeforeReplay {
+		t.Fatalf("identical replay published %d new events, want 0", published-publishedBeforeReplay)
+	}
+
+	if err := service.UpsertAgentPlanMessage(
+		ctx, "task-msg", "source-call-2", "sess-msg", "# Other plan", "",
+	); err != nil {
+		t.Fatalf("upsert separate plan: %v", err)
+	}
+
+	messages, err := repo.ListMessages(ctx, "sess-msg")
+	if err != nil {
+		t.Fatalf("list messages: %v", err)
+	}
+	if len(messages) != 2 {
+		t.Fatalf("message count = %d, want 2", len(messages))
+	}
+	if messages[0].Content != "# Plan\n\n1. Read\n2. Write" {
+		t.Fatalf("first plan content = %q", messages[0].Content)
+	}
+	if messages[1].Content != "# Other plan" {
+		t.Fatalf("second plan content = %q", messages[1].Content)
+	}
+	if correlation, _ := messages[0].Metadata["tool_call_id"].(string); !strings.HasPrefix(correlation, "agent-plan:") {
+		t.Fatalf("plan correlation = %q, want agent-plan namespace", correlation)
+	}
+	if source := messages[0].Metadata["agent_plan_tool_call_id"]; source != "source-call-1" {
+		t.Fatalf("source tool call id = %v, want source-call-1", source)
+	}
+}
+
+// @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.2
+func TestUpsertAgentPlanMessageSerializesReadThroughUpdate(t *testing.T) {
+	service, eventBus, baseRepo := newMessageTestService(t)
+	raceRepo := newAgentPlanRaceRepository(baseRepo)
+	service.messages = raceRepo
+	ctx := context.Background()
+
+	if err := service.UpsertAgentPlanMessage(
+		ctx, "task-msg", "source-call-race", "sess-msg", "initial snapshot", "turn-msg",
+	); err != nil {
+		t.Fatalf("seed plan: %v", err)
+	}
+	eventBus.ClearEvents()
+
+	olderDone := make(chan error, 1)
+	go func() {
+		olderDone <- service.UpsertAgentPlanMessage(
+			ctx, "task-msg", "source-call-race", "sess-msg", "older snapshot", "turn-msg",
+		)
+	}()
+	waitForAgentPlanRace(t, raceRepo.olderUpdateReady, "older snapshot update")
+
+	newerDone := make(chan error, 1)
+	go func() {
+		newerDone <- service.UpsertAgentPlanMessage(
+			ctx, "task-msg", "source-call-race", "sess-msg", "newer snapshot", "turn-msg",
+		)
+	}()
+	if raceRepo.isAdmissionHeld() {
+		close(raceRepo.releaseOlder)
+	} else {
+		waitForAgentPlanRace(t, raceRepo.newerUpdateDone, "newer snapshot update")
+		close(raceRepo.releaseOlder)
+	}
+	if err := waitForAgentPlanResult(t, olderDone, "older upsert"); err != nil {
+		t.Fatalf("older upsert: %v", err)
+	}
+	if err := waitForAgentPlanResult(t, newerDone, "newer upsert"); err != nil {
+		t.Fatalf("newer upsert: %v", err)
+	}
+
+	messages, err := baseRepo.ListMessages(ctx, "sess-msg")
+	if err != nil {
+		t.Fatalf("list messages: %v", err)
+	}
+	if len(messages) != 1 {
+		t.Fatalf("message count = %d, want 1", len(messages))
+	}
+	if messages[0].Content != "newer snapshot" {
+		t.Fatalf("plan content = %q, want newer snapshot", messages[0].Content)
+	}
+}
+
+func waitForAgentPlanRace(t *testing.T, ready <-chan struct{}, operation string) {
+	t.Helper()
+	select {
+	case <-ready:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timed out waiting for %s", operation)
+	}
+}
+
+func waitForAgentPlanResult(t *testing.T, result <-chan error, operation string) error {
+	t.Helper()
+	select {
+	case err := <-result:
+		return err
+	case <-time.After(2 * time.Second):
+		t.Fatalf("timed out waiting for %s", operation)
+		return nil
+	}
+}

apps/backend/internal/task/service/service_messages.go

@@ -30,6 +30,8 @@ const (
 
 var ErrMessageIDConflict = errors.New("client message id is already used")
 
+var agentPlanMessageIDNamespace = uuid.MustParse("138966de-88bc-49c0-b65f-cbfbac17f729")
+
 type planCommentMessageWriter interface {
 	CreateMessageWithPlanComments(
 		context.Context,
@@ -90,6 +92,13 @@ type conversationMessageReceiptWriter interface {
 	DeleteMessageWithConversationReceipt(context.Context, string) (*models.ConversationMutationReceipt, error)
 }
 
+type agentPlanMessageWriter interface {
+	UpsertAgentPlanMessageWithConversationReceipt(
+		context.Context,
+		*models.Message,
+	) (*models.Message, *models.ConversationMutationReceipt, bool, error)
+}
+
 type planCommentAdmissionLocker interface {
 	AcquirePlanCommentAdmission(context.Context, string) (context.Context, func(), error)
 }
@@ -1020,6 +1029,130 @@ func (s *Service) UpdateToolCallMessage(ctx context.Context, sessionID, toolCall
 	return s.UpdateToolCallMessageWithCreate(ctx, sessionID, toolCallID, "", status, result, title, normalized, "", "", "")
 }
 
+// UpsertAgentPlanMessage persists the latest snapshot for one plan-producing
+// tool call. Its deterministic identity makes first delivery and retries use
+// the same row without the missing-tool retry delay.
+func (s *Service) UpsertAgentPlanMessage(
+	ctx context.Context,
+	taskID, sourceToolCallID, sessionID, content, turnID string,
+) error {
+	name := fmt.Sprintf("%s\x00%s\x00%s", sessionID, turnID, sourceToolCallID)
+	messageID := uuid.NewSHA1(agentPlanMessageIDNamespace, []byte(name)).String()
+	correlationID := "agent-plan:" + messageID
+	req := &CreateMessageRequest{
+		TaskSessionID: sessionID,
+		TaskID:        taskID,
+		TurnID:        turnID,
+		Content:       content,
+		AuthorType:    "agent",
+		Type:          string(models.MessageTypeAgentPlan),
+		Metadata: map[string]interface{}{
+			"tool_call_id":            correlationID,
+			"agent_plan_tool_call_id": sourceToolCallID,
+		},
+	}
+	if err := s.authorizeMessageCreate(ctx, req); err != nil {
+		return err
+	}
+	session, err := s.getSessionWithRetry(
+		ctx, sessionID, messageID, messageCreateMaxRetries, messageCreateRetryDelay,
+	)
+	if err != nil {
+		return err
+	}
+	message, err := s.buildMessage(ctx, messageID, req, session)
+	if err != nil {
+		return err
+	}
+	if writer, ok := s.messages.(agentPlanMessageWriter); ok {
+		persisted, receipt, created, err := writer.UpsertAgentPlanMessageWithConversationReceipt(ctx, message)
+		if errors.Is(err, repoerrors.ErrMessageIdentityConflict) {
+			return ErrMessageIDConflict
+		}
+		if err != nil || receipt == nil {
+			return err
+		}
+		return s.publishAgentPlanMessage(ctx, persisted, receipt, created)
+	}
+	return s.upsertAgentPlanMessageFallback(ctx, message, req)
+}
+
+func (s *Service) upsertAgentPlanMessageFallback(
+	ctx context.Context,
+	message *models.Message,
+	req *CreateMessageRequest,
+) error {
+	admissionCtx, releaseAdmission, err := s.AcquireMessageAdmission(ctx, message.ID)
+	if err != nil {
+		return err
+	}
+	defer releaseAdmission()
+
+	existing, err := s.messages.GetMessageWithPromptIndex(admissionCtx, message.ID)
+	if err == nil && existing != nil {
+		if !sameAgentPlanMessageIdentity(existing, message) {
+			return ErrMessageIDConflict
+		}
+		if existing.Content == message.Content {
+			return nil
+		}
+		existing.Content = message.Content
+		receipt, err := s.updateMessageWithReceipt(admissionCtx, existing)
+		if err != nil {
+			return err
+		}
+		return s.publishAgentPlanMessage(admissionCtx, existing, receipt, false)
+	}
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("check existing agent plan message: %w", err)
+	}
+	_, receipt, err := s.createMessageWithRequestRetry(
+		admissionCtx, message, req, messageCreateMaxRetries, messageCreateRetryDelay,
+	)
+	if err != nil {
+		return err
+	}
+	return s.publishAgentPlanMessage(admissionCtx, message, receipt, true)
+}
+
+func (s *Service) publishAgentPlanMessage(
+	ctx context.Context,
+	message *models.Message,
+	receipt *models.ConversationMutationReceipt,
+	created bool,
+) error {
+	eventType := events.MessageUpdated
+	if created {
+		eventType = events.MessageAdded
+	}
+	err := s.publishMessageEvent(ctx, eventType, message, receipt)
+	if created {
+		return nil
+	}
+	return err
+}
+
+func sameAgentPlanMessageIdentity(existing, incoming *models.Message) bool {
+	if existing.ID != incoming.ID ||
+		existing.TaskSessionID != incoming.TaskSessionID ||
+		existing.TaskID != incoming.TaskID ||
+		existing.TurnID != incoming.TurnID ||
+		existing.AuthorType != incoming.AuthorType ||
+		existing.AuthorID != incoming.AuthorID ||
+		existing.Type != incoming.Type ||
+		existing.RequestsInput != incoming.RequestsInput {
+		return false
+	}
+	for _, key := range []string{"tool_call_id", "agent_plan_tool_call_id"} {
+		existingValue, existingOK := existing.Metadata[key].(string)
+		incomingValue, incomingOK := incoming.Metadata[key].(string)
+		if !existingOK || !incomingOK || existingValue != incomingValue {
+			return false
+		}
+	}
+	return true
+}
+
 // UpdateToolCallMessageWithCreate is like UpdateToolCallMessage but can create the message if not found.
 // If taskID, turnID, and msgType are provided, the message will be created if it doesn't exist.
 // parentToolCallID is used for subagent nesting (empty for top-level).

apps/web/e2e/tests/chat/agent-plan-coalescing.spec.ts

@@ -0,0 +1,64 @@
+import { test, expect } from "../../fixtures/test-base";
+import { SessionPage } from "../../pages/session-page";
+
+const FIRST_SNAPSHOT = "# Batch ES reads\n\n1. Read the first index";
+const LATEST_SNAPSHOT =
+  "# Batch ES reads\n\n1. Read the first index\n2. Read the second index\n3. Merge results";
+
+test.describe("agent plan coalescing", () => {
+  // The focused unit RED proves the projection defect; this browser case proves
+  // persisted replay through the shared desktop/phone transcript path.
+  // @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.3
+  test("shows one latest card for persisted cumulative plan snapshots", async ({
+    testPage,
+    apiClient,
+    seedData,
+    prCapture,
+  }) => {
+    const task = await apiClient.createTask(seedData.workspaceId, "Agent plan coalescing", {
+      description: "Persisted cumulative agent plan snapshots",
+      workflow_id: seedData.workflowId,
+      workflow_step_id: seedData.startStepId,
+      agent_profile_id: seedData.agentProfileId,
+      repository_ids: [seedData.repositoryId],
+    });
+    const { session_id: sessionId } = await apiClient.seedTaskSession(task.id, {
+      state: "IDLE",
+      agentProfileId: seedData.agentProfileId,
+    });
+    await apiClient.seedSessionMessage(sessionId, {
+      type: "agent_plan",
+      content: FIRST_SNAPSHOT,
+    });
+    await apiClient.seedSessionMessage(sessionId, {
+      type: "agent_plan",
+      content: LATEST_SNAPSHOT,
+    });
+
+    await testPage.goto(`/t/${task.id}`);
+    const session = new SessionPage(testPage);
+    await session.waitForLoad();
+
+    const chat = session.activeChat();
+    await expect(chat.getByText("Batch ES reads", { exact: true })).toHaveCount(1);
+    await expect(chat.getByText("Merge results", { exact: true })).toBeVisible();
+
+    await testPage.reload();
+    await session.waitForLoad();
+    await expect(session.activeChat().getByText("Batch ES reads", { exact: true })).toHaveCount(1);
+    await expect(session.activeChat().getByText("Merge results", { exact: true })).toBeVisible();
+
+    await prCapture.screenshot("desktop-agent-plan-coalescing", {
+      caption: "Desktop transcript shows one latest agent plan card after replay.",
+    });
+    if (prCapture.capturing) {
+      await testPage.setViewportSize({ width: 393, height: 851 });
+      await expect(session.activeChat().getByText("Batch ES reads", { exact: true })).toHaveCount(
+        1,
+      );
+      await prCapture.screenshot("phone-agent-plan-coalescing", {
+        caption: "Phone transcript uses the same single-card agent plan projection.",
+      });
+    }
+  });
+});

apps/web/hooks/processed-message-filtering.test.ts

@@ -135,6 +135,23 @@ function baseMessage(overrides: Partial<Message>): Message {
   } as Message;
 }
 
+const PLAN_TURN_ID = "turn-1";
+const AGENT_PLAN_TYPE = "agent_plan";
+const FIRST_PLAN_CONTENT = "# Plan\n\n1. Read";
+const LATEST_PLAN_CONTENT = `${FIRST_PLAN_CONTENT}\n2. Write`;
+const LEGACY_PLAN_1_ID = "legacy-plan-1";
+const LEGACY_PLAN_2_ID = "legacy-plan-2";
+
+function planMessage(id: string, content: string, overrides: Partial<Message> = {}): Message {
+  return baseMessage({
+    id,
+    turn_id: PLAN_TURN_ID,
+    type: AGENT_PLAN_TYPE,
+    content,
+    ...overrides,
+  });
+}
+
 function emptyTurnNotice(turnId: string): Message {
   return baseMessage({
     id: `empty-turn-${turnId}`,
@@ -241,6 +258,89 @@ describe("filterVisibleMessages empty-turn notice supersession", () => {
   });
 });
 
+const filterPlanMessages = (messages: Message[]) =>
+  filterVisibleMessages(messages, new Set<string>(), new Set<string>());
+
+describe("filterVisibleMessages correlated agent plans", () => {
+  // @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.1
+  it("keeps only the latest snapshot for a correlated plan stream", () => {
+    const correlation = { tool_call_id: "agent-plan:correlation-1" };
+    const first = planMessage("plan-1", FIRST_PLAN_CONTENT, { metadata: correlation });
+    const latest = planMessage("plan-2", LATEST_PLAN_CONTENT, { metadata: correlation });
+
+    expect(filterPlanMessages([first, latest]).map((message) => message.id)).toEqual(["plan-2"]);
+  });
+
+  it("keeps only the final delivery when repeated snapshots share a message id", () => {
+    const correlation = { tool_call_id: "agent-plan:correlation-1" };
+    const first = planMessage("plan-1", FIRST_PLAN_CONTENT, { metadata: correlation });
+    const latest = planMessage("plan-1", LATEST_PLAN_CONTENT, { metadata: correlation });
+
+    expect(filterPlanMessages([first, latest]).map((message) => message.content)).toEqual([
+      LATEST_PLAN_CONTENT,
+    ]);
+  });
+
+  // @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.2
+  it("keeps separate correlated plan streams in one turn", () => {
+    const first = planMessage("plan-1", "# First plan", {
+      metadata: { tool_call_id: "agent-plan:correlation-1" },
+    });
+    const second = planMessage("plan-2", "# Second plan", {
+      metadata: { tool_call_id: "agent-plan:correlation-2" },
+    });
+
+    expect(filterPlanMessages([first, second]).map((message) => message.id)).toEqual([
+      "plan-1",
+      "plan-2",
+    ]);
+  });
+});
+
+describe("filterVisibleMessages legacy agent plans", () => {
+  // @covers AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.4
+  it("collapses a contiguous same-turn legacy prefix chain", () => {
+    const first = planMessage(LEGACY_PLAN_1_ID, FIRST_PLAN_CONTENT);
+    const latest = planMessage(LEGACY_PLAN_2_ID, LATEST_PLAN_CONTENT);
+
+    expect(filterPlanMessages([first, latest]).map((message) => message.id)).toEqual([
+      LEGACY_PLAN_2_ID,
+    ]);
+  });
+
+  it("keeps legacy plans separated by another conversation item", () => {
+    const first = planMessage(LEGACY_PLAN_1_ID, FIRST_PLAN_CONTENT);
+    const message = baseMessage({ id: "message-1", turn_id: PLAN_TURN_ID, content: "Working" });
+    const latest = planMessage(LEGACY_PLAN_2_ID, LATEST_PLAN_CONTENT);
+
+    expect(filterPlanMessages([first, message, latest]).map((item) => item.id)).toEqual([
+      LEGACY_PLAN_1_ID,
+      "message-1",
+      LEGACY_PLAN_2_ID,
+    ]);
+  });
+
+  it("keeps legacy prefix plans from different turns", () => {
+    const first = planMessage(LEGACY_PLAN_1_ID, "# Plan");
+    const second = planMessage(LEGACY_PLAN_2_ID, FIRST_PLAN_CONTENT, { turn_id: "turn-2" });
+
+    expect(filterPlanMessages([first, second]).map((message) => message.id)).toEqual([
+      LEGACY_PLAN_1_ID,
+      LEGACY_PLAN_2_ID,
+    ]);
+  });
+
+  it("keeps same-turn legacy plans when the content is not a prefix", () => {
+    const first = planMessage(LEGACY_PLAN_1_ID, "# First plan");
+    const second = planMessage(LEGACY_PLAN_2_ID, "# Different plan");
+
+    expect(filterPlanMessages([first, second]).map((message) => message.id)).toEqual([
+      LEGACY_PLAN_1_ID,
+      LEGACY_PLAN_2_ID,
+    ]);
+  });
+});
+
 describe("filterVisibleMessages recovery history", () => {
   const RECOVERY_MESSAGE = "Could not resume the saved session.";
   const FIRST_RECOVERY_ID = "recovery-1";

apps/web/hooks/processed-message-filtering.ts

@@ -111,6 +111,52 @@ function deduplicateRecoveryMessages(messages: Message[]): Message[] {
   });
 }
 
+function agentPlanCorrelationKey(message: Message): string | null {
+  if (message.type !== "agent_plan") return null;
+  const metadata = message.metadata as { tool_call_id?: unknown } | undefined;
+  const toolCallId = metadata?.tool_call_id;
+  return typeof toolCallId === "string" && toolCallId.startsWith("agent-plan:") ? toolCallId : null;
+}
+
+function collapseCorrelatedAgentPlans(messages: Message[]): Message[] {
+  const latestIndexByCorrelation = new Map<string, number>();
+  for (const [index, message] of messages.entries()) {
+    const correlation = agentPlanCorrelationKey(message);
+    if (correlation) latestIndexByCorrelation.set(correlation, index);
+  }
+  if (latestIndexByCorrelation.size === 0) return messages;
+  return messages.filter((message, index) => {
+    const correlation = agentPlanCorrelationKey(message);
+    return !correlation || latestIndexByCorrelation.get(correlation) === index;
+  });
+}
+
+function collapseLegacyAgentPlanPrefixes(messages: Message[]): Message[] {
+  const collapsed: Message[] = [];
+  for (const message of messages) {
+    const previous = collapsed[collapsed.length - 1];
+    const isLegacyPrefix =
+      previous?.type === "agent_plan" &&
+      message.type === "agent_plan" &&
+      !agentPlanCorrelationKey(previous) &&
+      !agentPlanCorrelationKey(message) &&
+      Boolean(previous.turn_id) &&
+      previous.turn_id === message.turn_id &&
+      previous.content !== message.content &&
+      message.content.startsWith(previous.content);
+    if (isLegacyPrefix) {
+      collapsed[collapsed.length - 1] = message;
+    } else {
+      collapsed.push(message);
+    }
+  }
+  return collapsed;
+}
+
+function collapseAgentPlanSnapshots(messages: Message[]): Message[] {
+  return collapseLegacyAgentPlanPrefixes(collapseCorrelatedAgentPlans(messages));
+}
+
 export function isAgentBootResumeMessage(message: Message): boolean {
   if (message.type !== "script_execution") return false;
   const metadata = message.metadata as { script_type?: string; is_resuming?: boolean } | undefined;
@@ -359,10 +405,12 @@ export function filterVisibleMessages(
     if (message.type === "permission_request") return isPermissionVisible(message, toolCallIds);
     return false;
   });
-  return collapseTodoSnapshotsPerTurn(
-    dropSupersededEmptyTurnNotices(
-      deduplicateAgentBootResumes(deduplicateRecoveryMessages(filtered)),
-      messages,
+  return collapseAgentPlanSnapshots(
+    collapseTodoSnapshotsPerTurn(
+      dropSupersededEmptyTurnNotices(
+        deduplicateAgentBootResumes(deduplicateRecoveryMessages(filtered)),
+        messages,
+      ),
     ),
   );
 }

docs/decisions/2026-09-17-coalesce-streamed-agent-plans.md

@@ -0,0 +1,54 @@
+# ADR-2026-09-17-coalesce-streamed-agent-plans: Coalesce streamed agent plans
+
+**Status:** accepted
+**Date:** 2026-09-17
+**Area:** protocol
+
+## Context
+
+Some ACP agents stream a plan by repeatedly updating the `plan` input of one
+tool call. Kandev currently converts every snapshot into an `agent_plan` event
+without the source tool-call identity, then persists every event as a new
+conversation message. A single plan can therefore appear as many cumulative
+cards in live and replayed transcripts.
+
+The plan remains useful before the tool call completes, distinct plan
+invocations must remain distinct, and existing stored transcripts cannot be
+assumed to contain correlation metadata.
+
+## Decision
+
+Every agent-plan event produced from a tool update carries that source
+`ToolCallID`. The durable conversation boundary identifies one plan stream by
+session, turn, and tool call, and upserts its latest accepted snapshot into one
+`agent_plan` message.
+
+The web transcript projection uses the durable correlation identity when it is
+present. For historical uncorrelated rows, it collapses only contiguous
+same-turn `agent_plan` messages whose content forms a strict cumulative prefix
+chain. It does not rewrite stored historical rows.
+
+## Consequences
+
+Live updates and replay use the same durable message identity, so reloading does
+not reveal intermediate snapshots. Separate tool calls retain separate cards.
+The ACP-to-conversation path must preserve tool-call identity and test
+idempotent updates.
+
+Legacy cleanup is deliberately conservative. Historical snapshots that are not
+contiguous prefixes remain visible rather than risking accidental data loss.
+The compatibility projection can be removed only after the retained history no
+longer contains uncorrelated snapshots.
+
+## Alternatives Considered
+
+- Persist only the final tool input. This removes duplication but withholds a
+  useful plan while a long-running tool call is still in progress.
+- Deduplicate only in the frontend. This hides the symptom in one projection
+  while retaining duplicate durable messages for reload, pagination, APIs, and
+  future clients.
+- Coalesce by content similarity or turn alone. This can merge independent plan
+  invocations and makes identity depend on mutable text.
+- Rewrite all historical rows. This adds destructive migration risk for a
+  rendering compatibility issue that can be handled conservatively at read
+  time.

docs/plans/agent-plan-stream-coalescing/plan.md

@@ -0,0 +1,118 @@
+---
+status: done
+requirements:
+  - REQ-AGENTS-AGENT-PLAN-STREAM-COALESCING-001
+system_design:
+  - ../../specs/agents/system-design/agent-plan-stream-coalescing.md
+---
+
+# Agent plan stream coalescing
+
+## Overview
+
+Turn repeated ACP plan-input snapshots from one tool call into one durable,
+updating conversation message. Retain a conservative frontend projection for
+historical duplicate rows that lack correlation metadata.
+
+## Delivery order
+
+1. Add a failing ACP adapter test proving an emitted plan event carries the
+   source tool-call identity.
+2. Add failing orchestrator and task-service tests proving repeated snapshots
+   upsert one `agent_plan` message while distinct tool calls remain separate.
+3. Implement adapter correlation and durable message upsert.
+4. Add failing frontend tests for correlated rows and contiguous legacy prefix
+   chains, then implement the compatibility projection.
+5. Prove replay behavior through a focused Playwright transcript scenario and
+   run the targeted backend, frontend, and specification checks.
+
+The work is one vertical slice because adapter correlation, persistence, and
+projection are all required for one stable live-and-replay outcome.
+
+## Technical approach
+
+Use the existing `AgentEvent.ToolCallID` field for
+`EventTypeAgentPlan`. The orchestrator shall pass the task, session, active
+turn, source tool call, and snapshot to a dedicated task-service upsert path.
+The task service derives the deterministic message ID and namespaced
+correlation metadata from the session, turn, and source tool call. The
+repository serializes the read/create/update decision for that identity in one
+transaction; the first event creates one `agent_plan` row and each later
+accepted event updates that row's content.
+
+Keep uncorrelated adapter events append-compatible. In
+`use-processed-messages`, normalize visible plans before activity grouping:
+retain only the newest row for a repeated durable correlation key, and collapse
+uncorrelated rows only when they are adjacent, share a turn, and form a strict
+content-prefix chain.
+
+Do not add a database migration or modify `AgentPlanMessage`.
+
+## ASCII UI preview
+
+`UI-01: Streamed agent plan transcript`, entry point: desktop or phone task
+conversation after one plan-producing tool call.
+
+Current observed behavior:
+
+```text
+Conversation (single scroll owner)
++ Agent plan: Batch ES reads       [collapsed]
++ Agent plan: Batch ES reads       [collapsed]
++ Agent plan: Batch ES reads       [collapsed]
++ ...one card per partial snapshot
+```
+
+Proposed shared desktop and phone behavior:
+
+```text
+Conversation (single scroll owner)
++ Agent plan: Batch ES reads       [collapsed]
+  latest complete snapshot
+```
+
+The one-card cardinality is required by
+`AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.1` and `.5`. Spacing, card
+geometry, collapse controls, scroll behavior, and touch targets are unchanged.
+
+## Verification strategy
+
+- ACP adapter unit tests cover `ToolCallID` propagation and empty-plan guards.
+- Task-service and orchestrator tests cover create, update, identical no-op,
+  retry, distinct tool calls, serialized concurrent updates, and uncorrelated
+  compatibility. A PostgreSQL multi-connection regression proves the identity
+  lock spans the read/update transaction.
+- Frontend unit tests cover correlated duplicates, legacy prefix chains, turn
+  and adjacency boundaries, and non-prefix plans.
+- One Chromium Playwright scenario loads persisted duplicate snapshots and
+  observes one latest plan card. The shared normalization hook makes a second
+  mobile scenario redundant because no responsive presentation changes.
+- Specification validation and diff checks close the package.
+
+## Work orders
+
+- [x] [Task 01: Coalesce streamed agent plans](task-01-coalesce-streamed-agent-plans.md)
+
+## Verification results
+
+- ACP adapter, task-service, and orchestrator plan regressions passed.
+- Agent-plan repository regressions passed, including a real PostgreSQL
+  multi-connection lock-order case and the identical-snapshot no-op assertion.
+- Task-service and orchestrator plan regressions passed under the Go race
+  detector.
+- Backend application and integration message adapters compiled.
+- 87 focused frontend tests, TypeScript typecheck, and changed-file ESLint
+  passed.
+- The fresh production-build Chromium replay scenario passed before and after
+  reload.
+
+## Risks and exclusions
+
+- A correlation key that is not namespaced can collide with the visible tool
+  call message.
+- Content-length heuristics can merge unrelated plans; the legacy path must use
+  strict adjacency, same-turn, and prefix checks.
+- Retried or concurrent delivery can race with message creation and update;
+  the persistence boundary must serialize the full mutation by plan identity.
+- Do not rewrite historical rows, change task-plan documents, change plan-card
+  UI, or broaden transcript deduplication to other message types.

docs/plans/agent-plan-stream-coalescing/task-01-coalesce-streamed-agent-plans.md

@@ -0,0 +1,146 @@
+---
+id: "01-coalesce-streamed-agent-plans"
+title: "Coalesce streamed agent plans"
+status: done
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-AGENTS-AGENT-PLAN-STREAM-COALESCING-001
+acceptance_criteria:
+  - AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.1
+  - AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.2
+  - AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.3
+  - AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.4
+  - AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.5
+system_design:
+  - ../../specs/agents/system-design/agent-plan-stream-coalescing.md
+---
+
+# Task 01: Coalesce streamed agent plans
+
+## Summary
+
+Correlate ACP plan snapshots to their source tool call, persist one updating
+`agent_plan` message per invocation, and collapse conservative legacy snapshot
+chains during transcript projection.
+
+## In scope
+
+- Propagate source `ToolCallID` on ACP agent-plan events.
+- Add a task-service agent-plan upsert keyed by session, turn, and source tool
+  call, and route correlated orchestrator events through it.
+- Preserve append behavior for uncorrelated plan events.
+- Normalize correlated and qualifying legacy snapshots before message grouping.
+- Add red-first backend and frontend tests and one replay-focused Chromium
+  Playwright regression.
+
+## Out of scope
+
+- Database migrations or historical-row rewrites.
+- Task-plan document history or safe-edit behavior.
+- Plan-card markup, localization, layout, responsive branches, controls,
+  scrolling, or touch behavior.
+- Deduplicating other message types.
+
+## Acceptance
+
+- Repeated snapshots from one source tool call create one durable plan message
+  whose content is the latest accepted snapshot; retries remain idempotent and
+  separate tool calls remain separate.
+- Live, replayed, and paginated transcripts project correlated plans once and
+  fold only adjacent same-turn strict-prefix legacy chains.
+- Focused browser evidence displays one latest plan card from persisted
+  duplicates without changing desktop or phone composition.
+
+## ASCII UI preview
+
+`UI-01: Streamed agent plan transcript` from the
+[combined plan preview](plan.md#ascii-ui-preview):
+
+```text
+Desktop and phone task conversation
++ Agent plan: Batch ES reads       [collapsed]
+  latest complete snapshot
+```
+
+Card controls and the existing single transcript scroll owner are unchanged.
+
+## Verification
+
+```bash
+cd apps/backend
+go test ./internal/agentctl/server/adapter/transport/acp -run 'AgentPlan'
+go test ./internal/task/service ./internal/orchestrator -run 'AgentPlan'
+go test ./internal/task/repository/sqlite -run 'AgentPlan'
+cd ../
+pnpm install --frozen-lockfile
+pnpm --filter @kandev/web test -- hooks/processed-message-filtering.test.ts hooks/use-processed-messages.test.ts
+pnpm --filter @kandev/web run typecheck
+cd web
+pnpm exec eslint hooks/processed-message-filtering.ts hooks/processed-message-filtering.test.ts hooks/use-processed-messages.ts hooks/use-processed-messages.test.ts e2e/tests/chat/agent-plan-coalescing.spec.ts
+pnpm e2e:run --host --project chromium tests/chat/agent-plan-coalescing.spec.ts -- --retries=0
+cd ../..
+python3 scripts/list-docs.py decisions --format paths
+python3 scripts/lint-spec-files.py --all
+python3 scripts/list-docs.py validate
+git diff --check
+```
+
+## Files likely touched
+
+- `apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools.go`
+- `apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools_test.go`
+- `apps/backend/internal/orchestrator/event_handlers_streaming.go`
+- `apps/backend/internal/orchestrator/event_handlers_streaming_test.go`
+- `apps/backend/internal/task/service/service_messages.go`
+- `apps/backend/internal/task/service/service_messages_test.go`
+- `apps/web/hooks/processed-message-filtering.ts`
+- `apps/web/hooks/processed-message-filtering.test.ts`
+- `apps/web/hooks/use-processed-messages.ts`
+- `apps/web/hooks/use-processed-messages.test.ts`
+- `apps/web/e2e/tests/chat/agent-plan-coalescing.spec.ts`
+
+## Dependencies
+
+None.
+
+## Risks
+
+- Tool-call IDs are not globally unique; session and turn scope must remain part
+  of the durable identity.
+- The first snapshot and a retry can arrive close together; creation and update
+  cannot be separate non-idempotent operations.
+- The legacy compatibility rule must fail closed at any ambiguity.
+
+## Parallelism
+
+`sequential`
+
+## Inputs
+
+- `REQ-AGENTS-AGENT-PLAN-STREAM-COALESCING-001` and all acceptance criteria.
+- The agent-plan stream system design and
+  `ADR-2026-09-17-coalesce-streamed-agent-plans`.
+- Existing tool-call message update patterns and transcript visibility filters.
+
+## Results
+
+- RED: the ACP adapter emitted plan snapshots with an empty tool-call identity,
+  the orchestrator appended a new session message, and transcript filtering
+  retained both correlated and legacy cumulative snapshots.
+- GREEN: ACP plan events now carry the emitted source tool-call ID. The
+  orchestrator routes correlated plans to a deterministic task-service upsert,
+  while uncorrelated adapter events retain append compatibility.
+- GREEN: the repository serializes each correlated plan's read/create/update
+  and conversation receipt in one identity-locked transaction. Identical
+  snapshots skip both the write and update event.
+- GREEN: transcript filtering keeps the final delivery for each durable
+  correlation and folds only adjacent same-turn strict-prefix legacy chains.
+- Focused backend tests passed in the ACP adapter, task service, and
+  orchestrator; the task-service and orchestrator cases also passed with
+  `-race`. Backend application and integration message adapters compiled.
+- Focused frontend tests passed: 87 tests across the filtering and processed
+  message suites. TypeScript typecheck and changed-file ESLint passed.
+- The production-build Chromium replay scenario passed: 1 test confirmed one
+  latest plan card before and after page reload.

docs/specs/agents/requirements/agent-plan-stream-coalescing.md

@@ -0,0 +1,58 @@
+---
+status: active
+system: agents
+created: 2026-09-17
+owners:
+  - kandev
+---
+
+# Agent Plan Stream Coalescing Requirements
+
+## Overview
+
+Agent providers can stream a growing plan through repeated updates for one tool
+invocation. The agent system owns the provider-facing event contract and must
+present that logical plan once, without exposing transport snapshots as
+separate conversation entries.
+
+## Terminology
+
+- **Plan snapshot:** One complete plan value observed on an agent tool update.
+- **Plan stream identity:** The session, turn, and source tool-call identity
+  that together identify one logical plan-producing invocation.
+
+## Requirements
+
+### REQ-AGENTS-AGENT-PLAN-STREAM-COALESCING-001: One transcript item per plan stream
+
+**Intent:** Users can read and revisit an agent-authored plan without repeated
+cards created by incremental provider transport updates.
+
+**User story:** As a task user, I want one current plan card for each plan
+invocation, so that streamed drafting does not overwhelm the conversation.
+
+#### Acceptance criteria
+
+- **AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.1:** When one plan-producing
+  tool invocation emits multiple snapshots, the conversation shall contain one
+  plan item for that invocation and shall show its latest accepted snapshot.
+- **AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.2:** When separate plan-producing
+  tool invocations occur in the same session or turn, the conversation shall
+  keep them as separate plan items.
+- **AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.3:** After live delivery, page
+  reload, session resume, or history pagination, the conversation shall show
+  the same one-item-per-invocation result.
+- **AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.4:** When historical messages
+  lack a plan stream identity, a contiguous same-turn sequence whose content
+  grows strictly by prefix shall render only its final snapshot; non-prefix
+  plans and plans separated by another conversation item shall remain distinct.
+- **AC-AGENTS-AGENT-PLAN-STREAM-COALESCING-001.5:** Desktop and phone task
+  conversations shall apply the same plan-item cardinality without changing
+  the existing plan card controls, scrolling, or touch behavior.
+
+## Out of scope
+
+- Deleting or rewriting historical conversation rows.
+- Changing task-plan document storage, revision history, or agent plan-write
+  safety.
+- Changing plan-card layout, controls, copy, or responsive composition.

docs/specs/agents/system-design/agent-plan-stream-coalescing.md

@@ -0,0 +1,123 @@
+---
+status: current
+system: agents
+requirements:
+  - REQ-AGENTS-AGENT-PLAN-STREAM-COALESCING-001
+---
+
+# Agent Plan Stream Coalescing System Design
+
+## Purpose and boundaries
+
+The Agents system owns conversion of provider updates into durable conversation
+messages. This design covers ACP plan correlation, transcript persistence, and
+the compatibility projection used by the web conversation. It does not change
+task-plan document storage or the plan-card component.
+
+## Requirement mapping
+
+| Requirement | Design section |
+| --- | --- |
+| `REQ-AGENTS-AGENT-PLAN-STREAM-COALESCING-001` | [Data and contracts](#data-and-contracts), [Control flow](#control-flow), [Persistence and compatibility](#persistence-and-compatibility) |
+
+## Components and responsibilities
+
+- `internal/agentctl/server/adapter/transport/acp` extracts `rawInput.plan` from
+  ACP tool updates and attaches the source `ToolCallID` to
+  `streams.EventTypeAgentPlan`.
+- `internal/orchestrator` resolves the active session and turn, then passes
+  those IDs with the task, source tool call, and snapshot to the task service.
+- `internal/task/service` derives the deterministic message ID and namespaced
+  correlation metadata. Its agent-plan upsert uses the same durable mutation
+  path for first delivery, retries, and later snapshots.
+- `internal/task/repository/sqlite` owns the atomic read/create/update decision
+  and conversation receipt. PostgreSQL takes a transaction-scoped advisory
+  lock for the deterministic plan-message identity before its first read;
+  SQLite takes its writer lock before reading.
+- `apps/web/hooks/use-processed-messages.ts` projects correlated rows and
+  conservatively folds legacy cumulative snapshots before render grouping.
+- `AgentPlanMessage` remains the presentation component for the resulting
+  message and does not perform deduplication.
+
+## Data and contracts
+
+`streams.AgentEvent` already carries `ToolCallID`. An
+`EventTypeAgentPlan` emitted from an ACP tool update shall set:
+
+- `SessionID` to the agent session;
+- `ToolCallID` to the source ACP tool call;
+- `PlanContent` to the complete plan snapshot from that update.
+
+The task service derives a namespaced plan-message correlation key from
+`ToolCallID`. The namespace prevents collision with the visible tool-call
+message while preserving the original source identity in message metadata. The
+durable message identity additionally includes the session and active turn, so
+providers that reuse tool-call identifiers cannot merge unrelated plans.
+
+An uncorrelated `EventTypeAgentPlan` remains supported for adapters that do not
+yet provide a tool-call identity. It follows the existing append behavior and
+is eligible only for the conservative web compatibility projection.
+
+## Control flow
+
+1. The ACP adapter receives a tool-call update containing a non-empty
+   `rawInput.plan`.
+2. It emits an agent-plan event with the tool-call identity on every accepted
+   snapshot.
+3. The orchestrator resolves the active turn and asks the task service to
+   upsert the correlated `agent_plan` message.
+4. The task service derives the durable message identity and asks the
+   repository to apply the snapshot. The repository holds one identity-scoped
+   transaction lock across the first read, create or update, and conversation
+   receipt. An identical snapshot is a no-op. Concurrent calls therefore read
+   the prior committed snapshot and cannot overwrite in reverse after reading
+   the same stale value.
+5. WebSocket delivery and conversation replay both expose that same message.
+6. Before grouping render items, the web projection retains the latest message
+   for a repeated durable correlation key and applies the legacy prefix rule
+   only to uncorrelated rows.
+
+## Failure and recovery
+
+Missing session IDs, empty plan content, and unavailable message persistence
+remain non-actionable events under the existing handler guard. A correlated
+retry is idempotent and cannot create another plan row. If a provider omits
+`ToolCallID`, Kandev preserves the plan as a separate message rather than
+discarding it.
+
+Legacy folding requires all messages to be `agent_plan` entries in the same
+turn, adjacent after normal visibility filtering, and each previous content to
+be a strict prefix of the next. Any failed condition ends the chain. This
+failure-closed rule may leave ambiguous duplicates visible but cannot merge
+unrelated plans.
+
+## Persistence and compatibility
+
+No schema migration is required. The existing message content and metadata
+fields store the latest snapshot and its correlation identity. Upsert identity
+is deterministic across retries and process restarts. The repository's
+transaction-scoped identity lock also spans multiple backend instances that
+share PostgreSQL.
+
+Existing duplicate rows remain unchanged. The web compatibility projection
+selects the final row from qualifying legacy chains, which preserves immutable
+history while repairing conversation readability on live and paginated reads.
+
+## Observability
+
+Existing agent event and message update logs expose the event type, task,
+session, and tool-call update outcome. Failures use the current
+`failed to create agent plan message` error path, renamed to describe the
+upsert operation when correlation is present.
+
+## Responsive presentation
+
+This change only normalizes transcript data before the existing plan card is
+rendered. Desktop and phone use the same message hook and therefore receive the
+same one-card projection. The plan-card composition, scroll owner, controls,
+touch targets, safe-area behavior, and navigation do not change. Targeted hook
+tests satisfy mobile parity because no viewport-dependent UI behavior changes.
+
+## Related decisions
+
+- [ADR-2026-09-17-coalesce-streamed-agent-plans](../../../decisions/2026-09-17-coalesce-streamed-agent-plans.md)