diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools.go index 949750fc9d..bbcc7e2a89 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools.go +++ b/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, }) } diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools_test.go index f64fbda901..4af6d26bf3 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_tools_test.go +++ b/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 diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index 2cb1f95ff2..a9d24f7481 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/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{ diff --git a/apps/backend/internal/integration/test_server_test.go b/apps/backend/internal/integration/test_server_test.go index 1eb8cded24..df6374c33b 100644 --- a/apps/backend/internal/integration/test_server_test.go +++ b/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, diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 9aa64f4539..02d10a1177 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/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)) diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index dd567022b2..6fade959ee 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 945a479464..045739e7d2 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 9ab7ead7a7..0e3c35a9fa 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/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 } diff --git a/apps/backend/internal/task/repository/repoerrors/errors.go b/apps/backend/internal/task/repository/repoerrors/errors.go index 4dbcc2f299..9fe01d2aa6 100644 --- a/apps/backend/internal/task/repository/repoerrors/errors.go +++ b/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") diff --git a/apps/backend/internal/task/repository/sqlite/base.go b/apps/backend/internal/task/repository/sqlite/base.go index 63a206c005..ab2e815cbb 100644 --- a/apps/backend/internal/task/repository/sqlite/base.go +++ b/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 diff --git a/apps/backend/internal/task/repository/sqlite/message_agent_plan.go b/apps/backend/internal/task/repository/sqlite/message_agent_plan.go new file mode 100644 index 0000000000..c1acddfed8 --- /dev/null +++ b/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 +} diff --git a/apps/backend/internal/task/repository/sqlite/message_agent_plan_postgres_test.go b/apps/backend/internal/task/repository/sqlite/message_agent_plan_postgres_test.go new file mode 100644 index 0000000000..94cb69c4f6 --- /dev/null +++ b/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", + }, + } +} diff --git a/apps/backend/internal/task/service/service_agent_plan_messages_test.go b/apps/backend/internal/task/service/service_agent_plan_messages_test.go new file mode 100644 index 0000000000..ce453485c7 --- /dev/null +++ b/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 + } +} diff --git a/apps/backend/internal/task/service/service_messages.go b/apps/backend/internal/task/service/service_messages.go index 0825629651..dbceae0ab2 100644 --- a/apps/backend/internal/task/service/service_messages.go +++ b/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). diff --git a/apps/web/e2e/tests/chat/agent-plan-coalescing.spec.ts b/apps/web/e2e/tests/chat/agent-plan-coalescing.spec.ts new file mode 100644 index 0000000000..50ebc345e5 --- /dev/null +++ b/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.", + }); + } + }); +}); diff --git a/apps/web/hooks/processed-message-filtering.test.ts b/apps/web/hooks/processed-message-filtering.test.ts index 87c8e59b5b..80a51d3764 100644 --- a/apps/web/hooks/processed-message-filtering.test.ts +++ b/apps/web/hooks/processed-message-filtering.test.ts @@ -135,6 +135,23 @@ function baseMessage(overrides: Partial): 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 { + 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(), new Set()); + +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"; diff --git a/apps/web/hooks/processed-message-filtering.ts b/apps/web/hooks/processed-message-filtering.ts index 65c0e4af7b..1163383efd 100644 --- a/apps/web/hooks/processed-message-filtering.ts +++ b/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(); + 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, + ), ), ); } diff --git a/docs/decisions/2026-09-17-coalesce-streamed-agent-plans.md b/docs/decisions/2026-09-17-coalesce-streamed-agent-plans.md new file mode 100644 index 0000000000..e79e9f82a4 --- /dev/null +++ b/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. diff --git a/docs/plans/agent-plan-stream-coalescing/plan.md b/docs/plans/agent-plan-stream-coalescing/plan.md new file mode 100644 index 0000000000..a5b897816f --- /dev/null +++ b/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. diff --git a/docs/plans/agent-plan-stream-coalescing/task-01-coalesce-streamed-agent-plans.md b/docs/plans/agent-plan-stream-coalescing/task-01-coalesce-streamed-agent-plans.md new file mode 100644 index 0000000000..cc5f55810d --- /dev/null +++ b/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. diff --git a/docs/specs/agents/requirements/agent-plan-stream-coalescing.md b/docs/specs/agents/requirements/agent-plan-stream-coalescing.md new file mode 100644 index 0000000000..ac070769ea --- /dev/null +++ b/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. diff --git a/docs/specs/agents/system-design/agent-plan-stream-coalescing.md b/docs/specs/agents/system-design/agent-plan-stream-coalescing.md new file mode 100644 index 0000000000..5a4c05ff0d --- /dev/null +++ b/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)