fix: ignore late terminal tool updates (#1740)
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 `<return-code>0</return-code>`, **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.
|
||||
|
||||
Reference in New Issue
Block a user