diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go index 57cfb6e794..8f31f59e59 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go @@ -14,7 +14,10 @@ import ( v1 "github.com/kandev/kandev/pkg/api/v1" ) -const toolStatusComplete = "complete" +const ( + toolStatusComplete = "complete" + toolStatusFailed = "failed" +) // handleMessageChunkEvent handles a "message_chunk" agent event, accumulating and flushing on newlines. func (m *Manager) handleMessageChunkEvent(execution *AgentExecution, event agentctl.AgentEvent) { @@ -415,6 +418,18 @@ var turnContentEventTypes = map[string]struct{}{ "permission_request": {}, } +func isTerminalToolUpdate(event agentctl.AgentEvent) bool { + if event.Type != "tool_update" { + return false + } + switch event.ToolStatus { + case toolStatusComplete, "completed", "success", "error", toolStatusFailed, "cancelled": + return true + default: + return false + } +} + // recordActivity updates the last-activity timestamp and, on the very first // event from an execution, publishes AgentRunning to transition STARTING → RUNNING. // @@ -449,6 +464,9 @@ func (m *Manager) recordActivity(execution *AgentExecution, event agentctl.Agent if m.executionStore == nil { return } + if isTerminalToolUpdate(event) { + return + } if _, ok := turnContentEventTypes[event.Type]; !ok { return } diff --git a/apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go b/apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go index fb46a30174..b4bee1219c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go @@ -187,6 +187,34 @@ func TestWakeup_PostBootMetadataEventsDoNotFlipToRunning(t *testing.T) { } } +func TestWakeup_LateTerminalToolUpdateDoesNotFlipToRunning(t *testing.T) { + for _, status := range []string{"complete", "completed", "success", "error", "failed", "cancelled"} { + t.Run(status, func(t *testing.T) { + mgr, eventBus := createTestManagerWithTracking() + execution := createTestExecution("exec-1", "task-1", "session-1") + _ = mgr.executionStore.Add(execution) + + execution.firstActivityOnce.Do(func() {}) + mgr.executionStore.UpdateStatus(execution.ID, v1.AgentStatusReady) + + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "tool_update", + ToolCallID: "tool-1", + ToolStatus: status, + }) + + for _, te := range eventBus.PublishedEvents { + if te.Event != nil && te.Event.Type == events.AgentRunning { + t.Fatal("standalone terminal tool update must not publish agent.running") + } + } + if execution.Status != v1.AgentStatusReady { + t.Fatalf("execution.Status = %q, want %q", execution.Status, v1.AgentStatusReady) + } + }) + } +} + // TestWakeup_EmptyTurnStillPublishesAgentReady covers the narrow edge case // where a wakeup turn produces *only* a `complete` event (no preceding // message_chunk/tool_call/etc) — e.g. when the model returns an empty diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_async_complete.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_async_complete.go index 7fe2321bce..07322e6fcf 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_async_complete.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_async_complete.go @@ -25,6 +25,12 @@ func currentAsyncTurnCompleteIdle() time.Duration { } func isAsyncTurnContentEvent(event AgentEvent) bool { + if event.Type == streams.EventTypeToolUpdate { + switch event.ToolStatus { + case toolStatusComplete, toolStatusError, toolStatusCancelled: + return false + } + } switch event.Type { case streams.EventTypeMessageChunk, streams.EventTypeReasoning, diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/async_complete_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/async_complete_test.go index 128a441968..0d101bc9ca 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/async_complete_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/async_complete_test.go @@ -68,6 +68,25 @@ func TestHandleACPUpdate_DoesNotEmitIdleCompleteWhilePromptActive(t *testing.T) }) } +func TestAsyncTurnComplete_TerminalToolUpdateDoesNotStartTurn(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + setAsyncTurnCompleteIdleForTest(t, 10*time.Millisecond) + a := newTestAdapter() + defer func() { _ = a.Close() }() + + a.maybeScheduleAsyncTurnComplete(AgentEvent{ + Type: streams.EventTypeToolUpdate, + SessionID: "s-late-tool", + ToolCallID: "tool-1", + ToolStatus: toolStatusComplete, + }) + + time.Sleep(50 * time.Millisecond) + synctest.Wait() + assertNoAdapterEvent(t, a, "after a standalone terminal tool update") + }) +} + func TestAsyncTurnComplete_CancelledByRealPromptCompletion(t *testing.T) { synctest.Test(t, func(t *testing.T) { setAsyncTurnCompleteIdleForTest(t, 50*time.Millisecond) diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 64d5002bc7..b16a993d25 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -387,36 +387,70 @@ func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle. // Determine message type from normalized payload for fallback creation msgType := toolKindToMessageType(payload.Data.Normalized) - - // Handle all status updates (running, complete, error, cancelled, pending, in_progress) - switch payload.Data.ToolStatus { + status := payload.Data.ToolStatus + switch status { case "running", agentEventComplete, agentEventCompleted, "success", agentEventError, agentEventFailed, "cancelled", "pending", "in_progress": - if err := s.messageCreator.UpdateToolCallMessage( - ctx, - payload.TaskID, - payload.Data.ToolCallID, - payload.Data.ParentToolCallID, // Pass parent for subagent nesting - payload.Data.ToolStatus, - "", // result - no longer used, tool results in NormalizedPayload - payload.SessionID, - payload.Data.ToolTitle, // Include title from update event - s.getActiveTurnID(payload.SessionID), // Turn ID for fallback creation - msgType, // Message type for fallback creation - payload.Data.Normalized, // Pass normalized tool data for message metadata - ); err != nil { - s.logger.Warn("failed to update tool call message", + default: + return + } + terminal := isTerminalToolUpdateStatus(status) + turnID := "" + if terminal { + var err error + turnID, err = s.peekActiveTurnID(ctx, payload.SessionID) + if err != nil { + s.logger.Warn("failed to look up active turn for terminal tool update", zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID), zap.String("tool_call_id", payload.Data.ToolCallID), zap.Error(err)) + // Fail closed: without a confirmed active turn, this must be an + // update-only reconciliation and cannot wake a settled session. + turnID = "" } + } else { + turnID = s.getActiveTurnID(payload.SessionID) + } + fallbackMsgType := msgType + if terminal && turnID == "" { + // A late terminal update can update its existing card, but must not + // create a message (and implicitly a turn) after the turn settled. + fallbackMsgType = "" + } - // Update session state for completion events. - // Use setSessionRunning so the task flips to IN_PROGRESS alongside — - // see comment in handleToolCallEvent for the REVIEW/RUNNING split bug. - if payload.Data.ToolStatus == agentEventComplete || payload.Data.ToolStatus == agentEventCompleted || - payload.Data.ToolStatus == "success" || payload.Data.ToolStatus == agentEventError || payload.Data.ToolStatus == agentEventFailed { - s.setSessionRunningForExecution(ctx, payload.TaskID, payload.SessionID, payload.ExecutionID) - } + if err := s.messageCreator.UpdateToolCallMessage( + ctx, + payload.TaskID, + payload.Data.ToolCallID, + payload.Data.ParentToolCallID, // Pass parent for subagent nesting + status, + "", // result - no longer used, tool results in NormalizedPayload + payload.SessionID, + payload.Data.ToolTitle, // Include title from update event + turnID, // Turn ID for fallback creation + fallbackMsgType, // Empty for settled terminal reconciliations + payload.Data.Normalized, // Pass normalized tool data for message metadata + ); err != nil { + s.logger.Warn("failed to update tool call message", + zap.String("task_id", payload.TaskID), + zap.String("tool_call_id", payload.Data.ToolCallID), + zap.Error(err)) + } + + // Terminal updates only wake an async turn that was established by prior + // substantive output. A standalone terminal reconciliation belongs to the + // already-settled turn that created the tool call. + if terminal && status != "cancelled" && turnID != "" { + s.setSessionRunningForExecution(ctx, payload.TaskID, payload.SessionID, payload.ExecutionID) + } +} + +func isTerminalToolUpdateStatus(status string) bool { + switch status { + case agentEventComplete, agentEventCompleted, "success", agentEventError, agentEventFailed, "cancelled": + return true + default: + return false } } diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go index b5e72a6835..e7383ca26b 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming_test.go @@ -16,6 +16,7 @@ import ( "github.com/kandev/kandev/internal/orchestrator/watcher" "github.com/kandev/kandev/internal/task/models" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + taskservice "github.com/kandev/kandev/internal/task/service" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -56,6 +57,50 @@ type mockAutomationRunService struct { failedErrorByTask map[string]string } +type serviceBackedMessageCreator struct { + mockMessageCreator + svc *taskservice.Service +} + +func (m *serviceBackedMessageCreator) UpdateToolCallMessage( + ctx context.Context, + taskID, toolCallID, parentToolCallID, status, result, agentSessionID, title, turnID, msgType string, + normalized *streams.NormalizedPayload, +) error { + return m.svc.UpdateToolCallMessageWithCreate( + ctx, + agentSessionID, + toolCallID, + parentToolCallID, + status, + result, + title, + normalized, + taskID, + turnID, + msgType, + ) +} + +func newServiceBackedMessageCreator(repo *sqliterepo.Repository) *serviceBackedMessageCreator { + services := taskservice.NewService(taskservice.Repos{ + Workspaces: repo, + Tasks: repo, + TaskRepos: repo, + Workflows: repo, + Messages: repo, + Turns: repo, + Sessions: repo, + GitSnapshots: repo, + RepoEntities: repo, + Executors: repo, + Environments: repo, + TaskEnvironments: repo, + Reviews: repo, + }, bus.NewMemoryEventBus(testLogger()), testLogger(), taskservice.RepositoryDiscoveryConfig{}) + return &serviceBackedMessageCreator{svc: services} +} + func (m *mockAutomationRunService) GetAutomation(context.Context, string) (*automation.Automation, error) { return nil, nil } @@ -432,8 +477,9 @@ func TestToolEventsWakeSessionAndTaskTogether(t *testing.T) { ctx := context.Background() cases := []struct { - name string - fire func(*Service) + name string + activeTurn bool + fire func(*Service) }{ { name: "tool_call event", @@ -449,7 +495,8 @@ func TestToolEventsWakeSessionAndTaskTogether(t *testing.T) { }, }, { - name: "tool_update completion event", + name: "tool_update completion event for active async turn", + activeTurn: true, fire: func(svc *Service) { svc.handleToolUpdateEvent(ctx, &lifecycle.AgentStreamEventPayload{ TaskID: "t1", @@ -477,6 +524,11 @@ func TestToolEventsWakeSessionAndTaskTogether(t *testing.T) { taskRepo := newMockTaskRepo() svc := createTestService(repo, newMockStepGetter(), taskRepo) svc.messageCreator = &mockMessageCreator{} + if tc.activeTurn { + svc.turnService = &repoTurnService{repo: repo} + _, err := svc.turnService.StartTurn(ctx, "s1") + require.NoError(t, err) + } tc.fire(svc) @@ -503,6 +555,11 @@ func TestToolEventsWakeSessionAndTaskTogether(t *testing.T) { taskRepo := newMockTaskRepo() svc := createTestService(repo, newMockStepGetter(), taskRepo) svc.messageCreator = &mockMessageCreator{} + if tc.activeTurn { + svc.turnService = &repoTurnService{repo: repo} + _, err := svc.turnService.StartTurn(ctx, "s1") + require.NoError(t, err) + } tc.fire(svc) @@ -558,6 +615,69 @@ func TestToolUpdateFromCompletedExecutionDoesNotWakeWaitingSession(t *testing.T) "late completed-execution tool event must not move the task back to IN_PROGRESS") } +func TestLateTerminalToolUpdateDoesNotCreateTurnOrWakeWaitingSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task1", "session1", "step1") + + session, err := repo.GetTaskSession(ctx, "session1") + require.NoError(t, err) + session.State = models.TaskSessionStateWaitingForInput + require.NoError(t, repo.UpdateTaskSession(ctx, session)) + + taskRepo := newMockTaskRepo() + svc := createTestService(repo, newMockStepGetter(), taskRepo) + svc.turnService = &repoTurnService{repo: repo} + svc.messageCreator = newServiceBackedMessageCreator(repo) + + svc.handleToolUpdateEvent(ctx, &lifecycle.AgentStreamEventPayload{ + TaskID: "task1", + SessionID: "session1", + ExecutionID: "exec-1", + Data: &lifecycle.AgentStreamEventData{ + Type: "tool_update", + ToolCallID: "tool-1", + ToolStatus: agentEventComplete, + }, + }) + + messages, err := repo.ListMessages(ctx, "session1") + require.NoError(t, err) + require.Empty(t, messages, + "late terminal status must not create a fallback message when the original is absent") + require.Zero(t, openTurnCount(t, repo, "session1"), + "late terminal status must not create a phantom turn") + updated, err := repo.GetTaskSession(ctx, "session1") + require.NoError(t, err) + require.Equal(t, models.TaskSessionStateWaitingForInput, updated.State, + "late terminal status must not wake the settled session") + require.Zero(t, taskRepo.stateWrites["task1"], + "late terminal status must not move the task back to IN_PROGRESS") +} + +func TestStatuslessToolUpdateDoesNotCreateTurn(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task1", "session1", "step1") + + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + svc.turnService = &repoTurnService{repo: repo} + messages := &mockMessageCreator{} + svc.messageCreator = messages + + svc.handleToolUpdateEvent(ctx, &lifecycle.AgentStreamEventPayload{ + TaskID: "task1", + SessionID: "session1", + Data: &lifecycle.AgentStreamEventData{ + Type: "tool_update", + ToolCallID: "tool-1", + }, + }) + + require.Zero(t, messages.toolUpdateWrites) + require.Zero(t, openTurnCount(t, repo, "session1")) +} + func TestToolUpdateFromCompletedExecutionDoesNotCreateMessage(t *testing.T) { ctx := context.Background() repo := setupTestRepo(t) diff --git a/docs/specs/ui/acp-shell-command-output.md b/docs/specs/ui/acp-shell-command-output.md index a7ad73abd5..07fbe7fbbe 100644 --- a/docs/specs/ui/acp-shell-command-output.md +++ b/docs/specs/ui/acp-shell-command-output.md @@ -118,6 +118,7 @@ The extension is additive: agents that ignore it continue through their current - Output exceeding the bound is truncated deterministically and remains valid UTF-8. - Unknown exit status never renders a success check or a failure cross solely because the value is absent. - ACP `failed` and `cancelled` statuses terminate active-call tracking even when no exit code is available. +- A terminal tool update received after its originating turn settled reconciles the existing tool message without opening a new turn or changing session or task state. - A failed on-demand request leaves the disclosure open, retains the last successful snapshot, and shows an output-unavailable state until a retry succeeds. It never falls back to a body from the normal message payload. - Poll responses that arrive after collapse, unmount, or a newer request, including the final terminal snapshot request, are ignored. @@ -133,6 +134,7 @@ Live output updates use the existing tool-message update path and are persisted - **GIVEN** Auggie returns XML-like output with `0`, **WHEN** the command completes, **THEN** the parsed output is visible and the row shows `Exit code 0` as success. - **GIVEN** an agent returns plain output with no structured or embedded exit status, **WHEN** the command completes, **THEN** the output is visible and the row shows `Exit code unavailable` with neutral status. - **GIVEN** a command is cancelled without an exit code, **WHEN** its terminal update is persisted, **THEN** the transcript remains expandable and shows `Exit code unavailable`. +- **GIVEN** a command's turn has settled and the provider later emits a terminal tool update, **WHEN** the update is persisted, **THEN** the existing transcript is reconciled without creating a turn, waking the session or task, or producing an empty-output warning. - **GIVEN** terminal output exceeds 256 KiB, **WHEN** live and final updates are normalized, **THEN** stored output remains within the bound, keeps the newest valid UTF-8 text, and the expanded row indicates truncation. - **GIVEN** a persisted shell message with a long command, **WHEN** chat opens on desktop or mobile, **THEN** the complete command wraps visibly while its output disclosure remains collapsed. - **GIVEN** a collapsed shell command with persisted output, **WHEN** task boot state, message REST/WS lists, and live message notifications reach the browser, **THEN** they contain byte counts and result metadata but no stdout or stderr body, and no output endpoint request occurs.