fix(orchestrator): fail never-started agent turns instead of reporting healthy (#2776)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
nova28
2026-08-19 04:57:00 +08:00
committed by GitHub
parent f95c7bf9a9
commit 4100b5b5b0
20 changed files with 535 additions and 45 deletions
@@ -29,19 +29,23 @@ type AgentEventPayload struct {
}
// AgentStalledPayload describes a prompt that has stopped receiving agent events.
// It is advisory only; the lifecycle remains running until the provider completes
// or the user cancels the turn.
// The activity epoch lets the consumer reject a stale watchdog snapshot. A
// never-started prompt is terminal only when its snapshot remains current.
type AgentStalledPayload struct {
AgentExecutionID string `json:"agent_execution_id"`
TaskID string `json:"task_id"`
SessionID string `json:"session_id"`
PromptGeneration uint64 `json:"prompt_generation"`
ActivityEpoch uint64 `json:"activity_epoch,omitempty"`
LastActivityAt time.Time `json:"last_activity_at"`
StalledFor time.Duration `json:"stalled_for"`
ToolCallID string `json:"tool_call_id,omitempty"`
ToolName string `json:"tool_name,omitempty"`
ToolTitle string `json:"tool_title,omitempty"`
ToolStatus string `json:"tool_status,omitempty"`
// NeverStarted is true when the agent has not emitted a single event since
// this prompt was dispatched — a terminal failure, not a mid-work pause.
NeverStarted bool `json:"never_started"`
}
// AgentctlEventPayload is the payload for agentctl lifecycle events (starting, ready, error).
@@ -34,13 +34,18 @@ func (p *EventPublisher) PublishAgentEvent(ctx context.Context, eventType string
p.publishAgentEventPayload(ctx, eventType, newAgentEventPayload(execution))
}
// PublishAgentStalled publishes one advisory inactivity signal for a prompt.
// PublishAgentStalled publishes one inactivity signal for a prompt.
// neverStarted reports whether the agent has emitted zero events since this
// prompt was dispatched, and activityEpoch lets the consumer revalidate that
// classification before applying a terminal transition.
func (p *EventPublisher) PublishAgentStalled(
ctx context.Context,
execution *AgentExecution,
promptGeneration uint64,
lastActivityAt time.Time,
stalledFor time.Duration,
activityEpoch uint64,
neverStarted bool,
) {
if p.eventBus == nil {
return
@@ -50,8 +55,10 @@ func (p *EventPublisher) PublishAgentStalled(
TaskID: execution.TaskID,
SessionID: execution.SessionID,
PromptGeneration: promptGeneration,
ActivityEpoch: activityEpoch,
LastActivityAt: lastActivityAt,
StalledFor: stalledFor,
NeverStarted: neverStarted,
}
if tool := execution.activeToolSnapshot(); tool != nil {
payload.ToolCallID = tool.ToolCallID
@@ -149,6 +149,26 @@ func (s *ExecutionStore) OwnsPromptGeneration(sessionID, executionID string, gen
return exists && generation != 0 && execution.promptGeneration == generation
}
// OwnsPromptActivity reports whether the execution still owns the prompt and
// its activity has not changed since the watchdog captured its snapshot.
func (s *ExecutionStore) OwnsPromptActivity(
sessionID, executionID string,
generation, activityEpoch uint64,
) bool {
s.mu.RLock()
defer s.mu.RUnlock()
currentExecutionID, exists := s.bySession[sessionID]
if !exists || currentExecutionID != executionID {
return false
}
execution, exists := s.executions[currentExecutionID]
if !exists || generation == 0 || activityEpoch == 0 || execution.promptGeneration != generation {
return false
}
return execution.promptActivityEpochSnapshot() == activityEpoch
}
// ActivePromptGeneration returns the generation of the prompt that is dispatched
// and still in flight for executionID, or 0 if there is none. A steer reuses this
// exact generation so the folded completion is attributed to the predecessor's
@@ -268,6 +288,7 @@ func (s *ExecutionStore) MarkPromptDispatched(executionID string, generation uin
return
}
execution.dispatchedPromptGeneration = generation
execution.armPromptActivity()
}
// UpdateError updates the error message of an agent execution and sets its status to failed.
@@ -189,3 +189,24 @@ func TestExecutionStore_BeginPromptClearsActiveTopLevelTool(t *testing.T) {
t.Fatalf("active tool after BeginPrompt = %#v, want nil", got)
}
}
func TestExecutionStore_OwnsPromptActivityRejectsChangedEpoch(t *testing.T) {
store := NewExecutionStore()
exec := &AgentExecution{
ID: "exec-1",
SessionID: "session-1",
promptGeneration: 4,
}
if err := store.Add(exec); err != nil {
t.Fatalf("Add: %v", err)
}
exec.armPromptActivity()
if !store.OwnsPromptActivity(exec.SessionID, exec.ID, 4, 1) {
t.Fatal("current activity epoch should own the prompt")
}
exec.markAgentActivity()
if store.OwnsPromptActivity(exec.SessionID, exec.ID, 4, 1) {
t.Fatal("changed activity epoch should invalidate the snapshot")
}
}
@@ -310,9 +310,7 @@ func (m *Manager) handleCompleteEvent(execution *AgentExecution, event *agentctl
}
m.releaseActivity(executionActivityKey(execution.ID))
execution.lastActivityAtMu.Lock()
execution.lastActivityAt = time.Now()
execution.lastActivityAtMu.Unlock()
execution.markAgentActivity()
// Check buffer content BEFORE any processing
execution.messageMu.Lock()
@@ -508,8 +506,13 @@ func isTerminalToolUpdate(event agentctl.AgentEvent) bool {
// (available_commands_update arriving 50ms after MarkBootReady, etc.) don't
// accidentally re-arm a freshly-booted no-prompt session as Running.
func (m *Manager) recordActivity(execution *AgentExecution, event agentctl.AgentEvent) {
_, isTurnContent := turnContentEventTypes[event.Type]
execution.lastActivityAtMu.Lock()
execution.lastActivityAt = time.Now()
if isTurnContent {
execution.agentEventSincePrompt = true
execution.promptActivityEpoch++
}
execution.lastActivityAtMu.Unlock()
// Gate firstActivityOnce on `Status != Ready` so a delayed metadata
@@ -1532,6 +1532,26 @@ func TestHandleAgentEvent_UpdatesLastActivityAt(t *testing.T) {
}
}
func TestRecordActivity_MetadataDoesNotMarkPromptStarted(t *testing.T) {
mgr, _ := createTestManagerWithTracking()
execution := createTestExecution("exec-metadata", "task-1", "session-1")
execution.agentEventSincePrompt = false
execution.lastActivityAt = time.Now().Add(-time.Minute)
mgr.recordActivity(execution, agentctl.AgentEvent{Type: "session_models"})
lastActivity, agentEventSeen, epoch := execution.promptActivitySnapshot()
if time.Since(lastActivity) > time.Second {
t.Fatalf("metadata event did not refresh last activity: %v ago", time.Since(lastActivity))
}
if agentEventSeen {
t.Fatal("metadata event marked the prompt as started")
}
if epoch != 0 {
t.Fatalf("metadata event advanced activity epoch to %d, want 0", epoch)
}
}
func TestHandleAgentEvent_TracksActiveTopLevelTool(t *testing.T) {
mgr, _ := createTestManagerWithTracking()
execution := createTestExecution("exec-1", "task-1", "session-1")
@@ -1325,6 +1325,15 @@ func (m *Manager) OwnsPromptGeneration(sessionID, executionID string, generation
return m.executionStore.OwnsPromptGeneration(sessionID, executionID, generation)
}
// OwnsPromptActivity reports whether a stall snapshot still belongs to the
// active prompt and no genuine event arrived after the snapshot was captured.
func (m *Manager) OwnsPromptActivity(
sessionID, executionID string,
generation, activityEpoch uint64,
) bool {
return m.executionStore.OwnsPromptActivity(sessionID, executionID, generation, activityEpoch)
}
// GetPromptGenerationForSession returns the generation currently owned by the
// active prompt for sessionID. Orchestrator cancellation uses this together
// with the execution and turn IDs to reject stale terminal events while the
@@ -1021,16 +1021,16 @@ func (sm *SessionManager) waitForPromptDone(
return nil, ctx.Err()
case <-stallTicker.C:
execution.lastActivityAtMu.Lock()
elapsed := time.Since(execution.lastActivityAt)
lastActivity := execution.lastActivityAt
execution.lastActivityAtMu.Unlock()
lastActivity, agentEventSeen, activityEpoch := execution.promptActivitySnapshot()
elapsed := time.Since(lastActivity)
neverStarted := !agentEventSeen
if elapsed >= 5*time.Minute && !stallReported {
sm.logger.Warn("agent stall detected: no events received",
zap.String("execution_id", execution.ID),
zap.Duration("elapsed_since_last_event", elapsed),
zap.Time("last_activity", lastActivity))
zap.Time("last_activity", lastActivity),
zap.Bool("never_started", neverStarted))
if sm.eventPublisher != nil {
sm.eventPublisher.PublishAgentStalled(
ctx,
@@ -1038,6 +1038,8 @@ func (sm *SessionManager) waitForPromptDone(
promptGeneration,
lastActivity,
elapsed,
activityEpoch,
neverStarted,
)
}
stallReported = true
@@ -1144,6 +1146,7 @@ func (sm *SessionManager) markPromptDispatched(execution *AgentExecution, genera
if generation != 0 && execution.promptGeneration == generation &&
execution.promptCompletionGeneration != generation {
execution.dispatchedPromptGeneration = generation
execution.armPromptActivity()
}
}
@@ -1286,6 +1289,10 @@ func (sm *SessionManager) dispatchSteerLocked(
// activity-timestamp bump for a steer that actually dispatched. Kept off the
// fallback path so a steer that degrades to an ordinary prompt is not recorded
// twice.
//
// Deliberately does not touch agentEventSincePrompt: a steer is user input
// injected into an already-running turn, not agent output, so it must not
// mask a stalled agent as having produced something.
func (sm *SessionManager) recordSteerActivity(execution *AgentExecution, prompt string) {
if sm.historyManager != nil && execution.historyEnabled && execution.SessionID != "" {
if err := sm.historyManager.AppendUserMessage(execution.SessionID, prompt); err != nil {
@@ -1484,9 +1491,6 @@ func (sm *SessionManager) preparePrompt(
sm.logger.Warn("failed to store user message to history", zap.Error(err))
}
}
execution.lastActivityAtMu.Lock()
execution.lastActivityAt = time.Now()
execution.lastActivityAtMu.Unlock()
return ctx, effectivePrompt, promptGeneration, nil
}
@@ -2131,6 +2131,138 @@ func TestWaitForPromptDone_PublishesSingleStall(t *testing.T) {
})
}
func TestMarkPromptDispatchedArmsActivityAfterDispatch(t *testing.T) {
store := NewExecutionStore()
execution := &AgentExecution{
ID: "test-exec",
SessionID: "test-session",
promptGeneration: 3,
agentEventSincePrompt: true,
promptActivityEpoch: 2,
promptCompletionGeneration: 0,
}
if err := store.Add(execution); err != nil {
t.Fatalf("add execution: %v", err)
}
sm := NewSessionManager(newSessionTestLogger(), make(chan struct{}))
sm.executionStore = store
sm.markPromptDispatched(execution, 3)
_, agentEventSeen, epoch := execution.promptActivitySnapshot()
if agentEventSeen {
t.Fatal("prompt dispatch left the previous activity marker armed")
}
if epoch != 3 {
t.Fatalf("activity epoch = %d, want 3 after dispatch", epoch)
}
}
// TestWaitForPromptDone_StallPayloadDiscriminatesNeverStarted verifies that
// agentEventSincePrompt (armed false on dispatch, set true only by a genuine
// agent event) is threaded into the published stall payload's NeverStarted
// field: dispatch-only silence reports NeverStarted true, and the same
// execution reports NeverStarted false once a real agent event has recorded
// activity for the current prompt.
func TestWaitForPromptDone_StallPayloadDiscriminatesNeverStarted(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
eventBus := &MockEventBusWithTracking{}
sm := NewSessionManager(newSessionTestLogger(), make(chan struct{}))
sm.eventPublisher = NewEventPublisher(eventBus, newSessionTestLogger())
execution := &AgentExecution{
ID: "test-exec",
TaskID: "test-task",
SessionID: "test-session",
promptDoneCh: make(chan PromptCompletionSignal, 1),
}
// Mirrors sendPrompt's dispatch bump: activity timestamp moves, but the
// discriminator is armed false because no agent event has arrived yet.
execution.lastActivityAt = time.Now()
execution.agentEventSincePrompt = false
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
waitResult := make(chan error, 1)
go func() {
_, err := sm.waitForPromptDone(ctx, execution, 7)
waitResult <- err
}()
time.Sleep(5 * time.Minute)
synctest.Wait()
payload := lastStalledPayload(t, eventBus)
if !payload.NeverStarted {
t.Fatalf("NeverStarted = false, want true for dispatch-only silence")
}
cancel()
if err := <-waitResult; !errors.Is(err, context.Canceled) {
t.Fatalf("waitForPromptDone error = %v, want context canceled", err)
}
})
}
func TestWaitForPromptDone_StallPayloadReportsRunningAfterAgentEvent(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
eventBus := &MockEventBusWithTracking{}
sm := NewSessionManager(newSessionTestLogger(), make(chan struct{}))
sm.eventPublisher = NewEventPublisher(eventBus, newSessionTestLogger())
execution := &AgentExecution{
ID: "test-exec",
TaskID: "test-task",
SessionID: "test-session",
promptDoneCh: make(chan PromptCompletionSignal, 1),
Status: v1.AgentStatusRunning,
}
execution.lastActivityAt = time.Now()
execution.agentEventSincePrompt = false
mgr := &Manager{eventPublisher: sm.eventPublisher}
mgr.recordActivity(execution, agentctl.AgentEvent{Type: "message_chunk"})
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
waitResult := make(chan error, 1)
go func() {
_, err := sm.waitForPromptDone(ctx, execution, 7)
waitResult <- err
}()
time.Sleep(5 * time.Minute)
synctest.Wait()
payload := lastStalledPayload(t, eventBus)
if payload.NeverStarted {
t.Fatalf("NeverStarted = true, want false once a genuine agent event recorded activity")
}
cancel()
if err := <-waitResult; !errors.Is(err, context.Canceled) {
t.Fatalf("waitForPromptDone error = %v, want context canceled", err)
}
})
}
func lastStalledPayload(t *testing.T, eventBus *MockEventBusWithTracking) AgentStalledPayload {
t.Helper()
eventBus.mu.Lock()
defer eventBus.mu.Unlock()
for i := len(eventBus.PublishedEvents) - 1; i >= 0; i-- {
te := eventBus.PublishedEvents[i]
if te.Subject != "agent.stalled" {
continue
}
payload, ok := te.Event.Data.(AgentStalledPayload)
if !ok {
t.Fatalf("agent.stalled event data = %#v, want AgentStalledPayload", te.Event.Data)
}
return payload
}
t.Fatalf("no agent.stalled event published")
return AgentStalledPayload{}
}
func countPublishedEvents(eventBus *MockEventBusWithTracking, subject string) int {
eventBus.mu.Lock()
defer eventBus.mu.Unlock()
@@ -211,10 +211,20 @@ type AgentExecution struct {
promptFinishedMu sync.Mutex
// Last time an agent event was received (for stall detection)
lastActivityAt time.Time
lastActivityAtMu sync.Mutex
activeTool *activeTopLevelTool
activeToolMu sync.RWMutex
lastActivityAt time.Time
// agentEventSincePrompt is armed (false) on each prompt dispatch and set
// true by the first genuine agent event (recordActivity/handleCompleteEvent)
// that follows. It lets the stall watchdog distinguish "the agent never
// produced a single frame for this prompt" from "it worked, then paused" —
// both cases otherwise bump the same lastActivityAt timestamp.
agentEventSincePrompt bool
// promptActivityEpoch changes when a prompt is armed or a genuine agent
// event arrives. Stall consumers use it to reject a snapshot that became
// stale while the event was crossing the bus.
promptActivityEpoch uint64
lastActivityAtMu sync.Mutex
activeTool *activeTopLevelTool
activeToolMu sync.RWMutex
// Fires once on the first agent event to publish AgentRunning.
firstActivityOnce sync.Once
@@ -339,6 +349,34 @@ func (e *AgentExecution) promptGenerationSnapshot() uint64 {
return e.promptGeneration
}
func (e *AgentExecution) armPromptActivity() {
e.lastActivityAtMu.Lock()
e.lastActivityAt = time.Now()
e.agentEventSincePrompt = false
e.promptActivityEpoch++
e.lastActivityAtMu.Unlock()
}
func (e *AgentExecution) markAgentActivity() {
e.lastActivityAtMu.Lock()
e.lastActivityAt = time.Now()
e.agentEventSincePrompt = true
e.promptActivityEpoch++
e.lastActivityAtMu.Unlock()
}
func (e *AgentExecution) promptActivitySnapshot() (time.Time, bool, uint64) {
e.lastActivityAtMu.Lock()
defer e.lastActivityAtMu.Unlock()
return e.lastActivityAt, e.agentEventSincePrompt, e.promptActivityEpoch
}
func (e *AgentExecution) promptActivityEpochSnapshot() uint64 {
e.lastActivityAtMu.Lock()
defer e.lastActivityAtMu.Unlock()
return e.promptActivityEpoch
}
// beginStartupAttempt starts a generation for a new ACP process. Generation
// zero is reserved for executions that predate startup tracking.
func (e *AgentExecution) beginStartupAttempt() uint64 {
@@ -2,6 +2,7 @@ package orchestrator
import (
"context"
"errors"
"fmt"
"strings"
@@ -15,10 +16,25 @@ import (
const (
stallCancelTurnButtonTestID = "stall-cancel-turn-button"
metaKeyActionVisibility = "action_visibility"
actionVisibilityRunning = "running"
neverStartedNoticeContent = "Agent produced no output since start."
)
// handleAgentStalled persists an advisory recovery affordance without changing
// the prompt, session, or task lifecycle.
// errAgentNeverStarted is the launch-failure error recorded when a stalled
// prompt's agent has emitted zero events since dispatch. Its Error() string
// is persisted as the session's error_message by recordSessionLaunchFailure;
// it is not a missing-branch error, so handleSessionLaunchFailed's
// branch-guidance side effect stays inert for this path.
var errAgentNeverStarted = errors.New("agent produced no output since start")
// handleAgentStalled persists an advisory recovery affordance for a turn that
// stalled after producing at least one agent event, leaving the prompt,
// session, and task lifecycle unchanged. When the agent has produced no
// output at all since the prompt was dispatched (payload.NeverStarted), the
// condition is terminal and unrecoverable: the notice is posted as an error
// and the session/task transition to FAILED via the same idempotent CAS path
// used for launch failures. The activity epoch prevents a late event from
// turning an old watchdog snapshot into a terminal failure.
func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.AgentStalledPayload) {
if s.messageCreator == nil || s.repo == nil || payload.TaskID == "" || payload.SessionID == "" {
return
@@ -47,6 +63,19 @@ func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.Agen
) {
return
}
if payload.ActivityEpoch != 0 {
activityOwner, ok := s.agentManager.(interface {
OwnsPromptActivity(sessionID, executionID string, generation, activityEpoch uint64) bool
})
if !ok || !activityOwner.OwnsPromptActivity(
payload.SessionID,
payload.AgentExecutionID,
payload.PromptGeneration,
payload.ActivityEpoch,
) {
return
}
}
turnID, err := s.peekActiveTurnID(ctx, payload.SessionID)
if err != nil {
return
@@ -55,10 +84,15 @@ func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.Agen
return
}
metadata := map[string]interface{}{
metaKeyActionVisibility: "running",
metaKeySessionID: payload.SessionID,
metaKeyTaskID: payload.TaskID,
"actions": []map[string]interface{}{{
metaKeySessionID: payload.SessionID,
metaKeyTaskID: payload.TaskID,
}
messageType := string(v1.MessageTypeStatus)
if payload.NeverStarted {
messageType = string(v1.MessageTypeError)
} else {
metadata[metaKeyActionVisibility] = actionVisibilityRunning
metadata["actions"] = []map[string]interface{}{{
actionMetaKeyType: "ws_request",
actionMetaKeyLabel: "Cancel turn",
actionMetaKeyTestID: stallCancelTurnButtonTestID,
@@ -66,14 +100,14 @@ func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.Agen
"method": "agent.cancel",
"payload": map[string]interface{}{"session_id": payload.SessionID},
},
}},
}}
}
if err := s.messageCreator.CreateSessionMessage(
ctx,
payload.TaskID,
stallNoticeContent(payload),
payload.SessionID,
string(v1.MessageTypeStatus),
messageType,
turnID,
metadata,
false,
@@ -83,9 +117,15 @@ func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.Agen
zap.String("session_id", payload.SessionID),
zap.Error(err))
}
if payload.NeverStarted {
s.recordSessionLaunchFailure(ctx, payload.TaskID, payload.SessionID, errAgentNeverStarted, session)
}
}
func stallNoticeContent(payload lifecycle.AgentStalledPayload) string {
if payload.NeverStarted {
return neverStartedNoticeContent
}
tool := strings.TrimSpace(payload.ToolTitle)
if tool == "" {
tool = strings.TrimSpace(payload.ToolName)
@@ -7,6 +7,7 @@ import (
"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
"github.com/kandev/kandev/internal/task/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) {
@@ -22,6 +23,7 @@ func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) {
currentPromptExecutionID: "execution-1",
}
agentMgr.currentPromptGeneration.Store(7)
agentMgr.currentPromptActivityEpoch.Store(1)
svc := createTestServiceWithScheduler(
repo,
newMockStepGetter(),
@@ -41,6 +43,7 @@ func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) {
TaskID: "task-1",
SessionID: "session-1",
PromptGeneration: 7,
ActivityEpoch: 1,
ToolName: "shell",
ToolTitle: "Start dev server",
ToolStatus: "in_progress",
@@ -50,10 +53,13 @@ func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) {
t.Fatalf("session messages = %d, want 1", len(messages.sessionMessages))
}
message := messages.sessionMessages[0]
if message.messageType != string(v1.MessageTypeStatus) {
t.Fatalf("message type = %q, want status", message.messageType)
}
if !strings.Contains(message.content, "Still waiting on Start dev server") {
t.Fatalf("notice content = %q, want sanitized tool title", message.content)
}
if message.metadata["action_visibility"] != "running" {
if message.metadata["action_visibility"] != actionVisibilityRunning {
t.Fatalf("action visibility = %v, want running", message.metadata["action_visibility"])
}
if message.turnID != activeTurn.ID {
@@ -84,6 +90,102 @@ func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) {
}
}
func TestHandleAgentStalled_NeverStartedFailsSessionAndTask(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedSession(t, repo, "task-1", "session-1", "step-1")
agentMgr := &mockAgentManager{
repoForExecutionLookup: repo,
currentPromptExecutionID: "execution-1",
}
agentMgr.currentPromptGeneration.Store(7)
agentMgr.currentPromptActivityEpoch.Store(1)
taskRepo := newMockTaskRepo()
seedMockTaskState(taskRepo, "task-1", v1.TaskStateInProgress)
svc := createTestServiceWithScheduler(
repo,
newMockStepGetter(),
taskRepo,
agentMgr,
)
svc.turnService = &repoTurnService{repo: repo}
activeTurn, err := svc.turnService.StartTurn(ctx, "session-1")
if err != nil {
t.Fatalf("start active turn: %v", err)
}
messages := &mockMessageCreator{}
svc.messageCreator = messages
svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
AgentExecutionID: "execution-1",
TaskID: "task-1",
SessionID: "session-1",
PromptGeneration: 7,
ActivityEpoch: 1,
NeverStarted: true,
})
if len(messages.sessionMessages) != 1 {
t.Fatalf("session messages = %d, want 1", len(messages.sessionMessages))
}
message := messages.sessionMessages[0]
if message.messageType != string(v1.MessageTypeError) {
t.Fatalf("message type = %q, want error", message.messageType)
}
if message.content != neverStartedNoticeContent {
t.Fatalf("notice content = %q, want the never-started condition named", message.content)
}
if message.turnID != activeTurn.ID {
t.Fatalf("notice turn ID = %q, want active turn %q", message.turnID, activeTurn.ID)
}
if _, hasVisibility := message.metadata["action_visibility"]; hasVisibility {
t.Fatalf("terminal notice has running visibility metadata: %#v", message.metadata)
}
if _, hasActions := message.metadata["actions"]; hasActions {
t.Fatalf("terminal notice has a running cancel action: %#v", message.metadata)
}
after, err := repo.GetTaskSession(ctx, "session-1")
if err != nil {
t.Fatalf("get session after handling stall: %v", err)
}
if after.State != models.TaskSessionStateFailed {
t.Fatalf("session state = %q, want FAILED", after.State)
}
taskRepo.mu.Lock()
taskState := taskRepo.updatedStates["task-1"]
taskRepo.mu.Unlock()
if taskState != v1.TaskStateFailed {
t.Fatalf("task state = %q, want FAILED", taskState)
}
}
func TestHandleAgentStalled_RejectsActivityEpochChangedAfterSnapshot(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedSession(t, repo, "task-1", "session-1", "step-1")
agentMgr := &mockAgentManager{currentPromptExecutionID: "execution-1"}
agentMgr.currentPromptGeneration.Store(7)
agentMgr.currentPromptActivityEpoch.Store(2)
svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
svc.turnService = &repoTurnService{repo: repo}
svc.messageCreator = &mockMessageCreator{}
svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{
AgentExecutionID: "execution-1",
TaskID: "task-1",
SessionID: "session-1",
PromptGeneration: 7,
ActivityEpoch: 1,
NeverStarted: true,
})
if got := len(svc.messageCreator.(*mockMessageCreator).sessionMessages); got != 0 {
t.Fatalf("stale activity snapshot created %d messages, want 0", got)
}
}
func TestHandleAgentStalled_RejectsSettledOrStalePrompt(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
@@ -122,3 +224,10 @@ func TestStallNoticeContentFallsBackWithoutTool(t *testing.T) {
t.Fatalf("stallNoticeContent() = %q, want generic fallback", got)
}
}
func TestStallNoticeContentNamesNeverStartedCondition(t *testing.T) {
payload := lifecycle.AgentStalledPayload{NeverStarted: true, ToolName: "shell", ToolTitle: "Start dev server"}
if got := stallNoticeContent(payload); got != neverStartedNoticeContent {
t.Fatalf("stallNoticeContent() = %q, want the never-started message regardless of tool info", got)
}
}
@@ -345,8 +345,9 @@ type mockAgentManager struct {
// ErrCancelEscalated sentinels handled inside cancelAgentSilent).
cancelAgentErr error
currentPromptGeneration atomic.Uint64
currentPromptExecutionID string
currentPromptGeneration atomic.Uint64
currentPromptActivityEpoch atomic.Uint64
currentPromptExecutionID string
// set_session_mode tracking (issue #1183). Records (sessionID, modeID) for
// every SetSessionModeBySessionID call. setSessionModeErr, when set, is
@@ -529,6 +530,15 @@ func (m *mockAgentManager) OwnsPromptGeneration(_ string, executionID string, ge
return executionID == m.currentPromptExecutionID && generation == m.currentPromptGeneration.Load()
}
func (m *mockAgentManager) OwnsPromptActivity(
_ string,
executionID string,
generation, activityEpoch uint64,
) bool {
return m.OwnsPromptGeneration("", executionID, generation) &&
activityEpoch == m.currentPromptActivityEpoch.Load()
}
func (m *mockAgentManager) GetPromptGenerationForSession(_ context.Context, _ string) (uint64, error) {
return m.currentPromptGeneration.Load(), nil
}
@@ -313,6 +313,15 @@ describe("ActionMessage — running stall notice", () => {
expect(container.firstChild).toBeNull();
});
it("keeps a terminal stall diagnostic visible after the session fails", () => {
const message = stalledMessage();
message.type = "error";
renderAction(message, "FAILED");
expect(screen.getByText("Still waiting on Start dev server.")).toBeTruthy();
});
it("sends agent.cancel when Cancel turn is activated", async () => {
renderAction(stalledMessage(), "RUNNING", undefined, "turn-1");
@@ -51,12 +51,20 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment:
const message = comment.content || t("task:anErrorOccurred");
if (metadata?.action_visibility === "running") {
if (sessionState !== "RUNNING" || !comment.turn_id || activeTurnId !== comment.turn_id) {
if (sessionState === "RUNNING" && comment.turn_id && activeTurnId === comment.turn_id) {
return (
<RunningActionNotice
actions={metadata.actions}
message={message}
taskId={comment.task_id}
/>
);
}
// A terminal error may have been persisted with the old running metadata
// shape. Let it use the settled renderer instead of hiding the diagnostic.
if (comment.type !== "error") {
return null;
}
return (
<RunningActionNotice actions={metadata.actions} message={message} taskId={comment.task_id} />
);
}
return (
@@ -1,6 +1,6 @@
# ADR-2026-07-29-agent-stall-user-controlled-recovery: Keep Agent Stall Recovery User Controlled
**Status:** accepted
**Status:** superseded by 2026-08-18-never-started-agent-stall-terminal
**Date:** 2026-07-29
**Area:** backend, frontend, protocol
@@ -0,0 +1,43 @@
# ADR-2026-08-18-never-started-agent-stall-terminal: Treat Never-Started Agent Stalls as Terminal
**Status:** accepted
**Date:** 2026-08-18
**Area:** backend, frontend, protocol
## Context
The five-minute stall watchdog normally reports advisory inactivity. A prompt
that has not produced any genuine agent event is different from a prompt that
started work and then became quiet. Without this distinction, a failed launch
can remain `RUNNING` and the user cannot recover it from the task view.
## Decision
Keep ordinary post-activity stalls advisory. When a current watchdog snapshot
shows no genuine turn event after the prompt was accepted, Kandev records the
existing launch-failure outcome and moves the session and task to `FAILED`.
The snapshot carries a prompt activity epoch. The orchestrator validates the
execution, prompt generation, and activity epoch before it applies the failure.
The terminal message uses settled error rendering and has no running-only cancel
action. Running notices keep their existing neutral cancel action.
## Consequences
Failed launches become visible and recoverable without waiting for a process
restart. A legitimate provider that emits no event before the threshold can be
classified as a failed launch, so the rule is limited to zero-event prompts and
does not change recovery for turns that already produced activity.
The lifecycle and orchestrator must keep the activity epoch in memory and must
reject stale watchdog events. Frontend rendering must not hide a terminal error
because it carries historical running metadata.
## Alternatives Considered
- **Keep every stall advisory.** Rejected because a never-started prompt can
remain `RUNNING` forever with no user-visible terminal result.
- **Fail every quiet prompt.** Rejected because long-running tools can be quiet
after they have already started.
- **Use only prompt generation.** Rejected because an agent event can arrive
after the watchdog snapshot and before the bus consumer handles it.
+1
View File
@@ -174,3 +174,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or
| 2026-08-15-office-mode-follows-active-workspace | [Office Mode Follows the Active Workspace](2026-08-15-office-mode-follows-active-workspace.md) | accepted | frontend, backend | 2026-08-15 |
| 2026-08-17-separate-task-activity-from-summary-freshness | [Separate Task Activity From Summary Freshness](2026-08-17-separate-task-activity-from-summary-freshness.md) | accepted | backend, frontend, protocol | 2026-08-17 |
| 2026-08-17-release-pr-ruleset-bypass | [Give Stable Release PRs an Administrator Token Bypass](2026-08-17-release-pr-ruleset-bypass.md) | accepted | infra, workflow, security | 2026-08-17 |
| 2026-08-18-never-started-agent-stall-terminal | [Treat Never-Started Agent Stalls as Terminal](2026-08-18-never-started-agent-stall-terminal.md) | accepted | backend, frontend, protocol | 2026-08-18 |
+10 -6
View File
@@ -1,6 +1,6 @@
---
spec: docs/specs/agent-stall-recovery/spec.md
decision: docs/decisions/2026-07-29-agent-stall-user-controlled-recovery.md
decision: docs/decisions/2026-08-18-never-started-agent-stall-terminal.md
created: 2026-07-29
status: draft
---
@@ -12,8 +12,9 @@ status: draft
Turn the lifecycle watchdog's existing inactivity observation into one
generation-scoped internal event, persist that event as an actionable notice,
and make the notice visible while the session is still `RUNNING`. The existing
cancel-escalation path remains the sole recovery mechanism; no timeout changes
session or process state automatically.
cancel-escalation path remains the recovery mechanism for turns that already
produced activity. A current zero-event snapshot uses the launch failure path
described by ADR-2026-08-18-never-started-agent-stall-terminal.
## Confirmed root cause
@@ -43,7 +44,8 @@ waits, force-releases a stuck prompt, and reconciles the session to
60 seconds, detect five minutes of inactivity, and publish `agent.stalled`
once per prompt generation. The payload carries task/session/execution IDs,
prompt generation, last-activity time, stalled duration, and optional tool
ID/name/title/status.
ID/name/title/status. Include an activity epoch so the orchestrator can
reject a snapshot that became stale after an agent event arrived.
- Add the payload and publisher support in
`apps/backend/internal/agent/runtime/lifecycle/event_types.go`,
`apps/backend/internal/agent/runtime/lifecycle/events.go`, and
@@ -66,6 +68,8 @@ waits, force-releases a stuck prompt, and reconciles the session to
the notice the active `turn_id` without lazily creating a successor turn.
- If message persistence fails, log the error and leave the running prompt
untouched.
- Never-started notices use settled error metadata and do not expose the
running-only cancel action.
---
@@ -191,8 +195,8 @@ requires them; the task-level commands remain the TDD evidence for each change.
## Risks and non-goals
- Event silence remains ambiguous, so the notice is advisory and cannot alter
lifecycle state by itself.
- Event silence remains advisory after genuine activity. A current zero-event
snapshot is the explicit launch-failure exception recorded in the ADR.
- Tool titles are already user-visible display data; raw command arguments must
not be copied into the notice.
- The notice is persisted once per prompt generation and assigned the active
+12 -5
View File
@@ -1,7 +1,7 @@
---
status: approved
created: 2026-07-29
updated: 2026-08-08
updated: 2026-08-18
owner: Kandev
---
@@ -13,6 +13,7 @@ Decisions:
- [ADR-2026-08-02-agent-terminal-diagnostics-over-stderr](../../decisions/2026-08-02-agent-terminal-diagnostics-over-stderr.md)
- [ADR-2026-08-07-allowlisted-provider-action-links](../../decisions/2026-08-07-allowlisted-provider-action-links.md)
- [ADR-2026-08-08-provider-neutral-agent-error-recovery](../../decisions/2026-08-08-provider-neutral-agent-error-recovery.md)
- [ADR-2026-08-18-never-started-agent-stall-terminal](../../decisions/2026-08-18-never-started-agent-stall-terminal.md)
Implementation plans:
@@ -47,8 +48,10 @@ mere inactivity so a failed turn is not presented as healthy work.
- The notice remains visible and actionable while the affected prompt's
`turn_id` is the active turn in a `RUNNING` session, including after a page
reload. It is hidden when that turn settles or a later turn becomes active.
- Detection alone does not change task state, session state, prompt admission,
or process liveness.
- Detection after genuine agent activity does not change task state, session
state, prompt admission, or process liveness. A current five-minute snapshot
with no genuine event since prompt dispatch is a launch failure and moves the
session and task to `FAILED`.
- The backend logs the first stall detected for a prompt generation and does
not emit another notice or log entry on every watchdog check.
@@ -178,6 +181,10 @@ sanitized diagnostic message for the collapsed technical-details surface.
- **GIVEN** a stall notice is visible on a phone viewport, **WHEN** the user taps
**Cancel turn**, **THEN** the same cancellation outcome is reachable through
an inline, content-width touch target of at least 44px.
- **GIVEN** a prompt has produced no genuine agent event since dispatch,
**WHEN** the current five-minute watchdog snapshot is handled, **THEN** Kandev
persists a terminal error, moves the session and task to `FAILED`, and does
not render a running-only cancel action.
- **GIVEN** a quiet but legitimate long-running turn, **WHEN** the inactivity
threshold passes and the user does not cancel, **THEN** Kandev leaves the
turn and process running.
@@ -214,8 +221,8 @@ sanitized diagnostic message for the collapsed technical-details surface.
## Out of scope
- Automatically timing out, failing, cancelling, or killing a turn based only
on inactivity.
- Automatically timing out, cancelling, or killing a turn based only on
inactivity.
- Making the inactivity threshold user-configurable.
- Reading, tailing, or exposing an agent vendor's private log files.
- Treating arbitrary stderr text as trusted terminal evidence.