From 69d1fc116e195fe2d3f183333131bbbfaddae5dd Mon Sep 17 00:00:00 2001 From: shayne-snap Date: Fri, 7 Aug 2026 08:01:26 +0800 Subject: [PATCH] fix(agent): recover abrupt turns and trace interrupts --- internal/agent/cancel_repair_behavior_test.go | 16 ++++++- internal/agent/cancellation.go | 36 +++++++++++++++- internal/agent/cancellation_test.go | 43 +++++++++++++++++++ internal/agent/interrupt_marker.go | 13 +++++- internal/agent/premature_end_turn.go | 7 +++ internal/agent/premature_end_turn_test.go | 5 +++ internal/agent/stream_ingest.go | 17 +++++++- internal/agent/turn_loop.go | 4 +- internal/app/service/dispatch.go | 2 +- internal/app/service/events_test.go | 27 ++++++++++++ internal/app/service/protocol_dispatch.go | 1 + internal/app/service/service.go | 1 + internal/runtime/protocol/intents.go | 1 + internal/tui/model_busy_queue_test.go | 6 +-- internal/tui/model_event_handlers.go | 2 +- internal/tui/model_keys.go | 8 ++-- internal/tui/model_keys_chat.go | 2 +- internal/tui/model_keys_modals.go | 2 +- 18 files changed, 174 insertions(+), 19 deletions(-) create mode 100644 internal/agent/cancellation_test.go diff --git a/internal/agent/cancel_repair_behavior_test.go b/internal/agent/cancel_repair_behavior_test.go index 23016a72..a951b7f8 100644 --- a/internal/agent/cancel_repair_behavior_test.go +++ b/internal/agent/cancel_repair_behavior_test.go @@ -32,12 +32,12 @@ func TestRunStreamCancelCurrentTurn(t *testing.T) { store := NewInMemoryStore() prov := &cancelThenSummaryProvider{} a := NewAgent(prov, store, nil) - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancelCause(context.Background()) events, err := a.RunStream(ctx, "s-cancel", "hi") if err != nil { t.Fatalf("run stream failed: %v", err) } - time.AfterFunc(10*time.Millisecond, cancel) + time.AfterFunc(10*time.Millisecond, func() { cancel(NewUserInterrupt("esc")) }) seenCancelled := false for ev := range events { @@ -63,6 +63,18 @@ func TestRunStreamCancelCurrentTurn(t *testing.T) { if last.Role != RoleUser || !last.Hidden || last.FinishReason != FinishReasonCanceled || !strings.Contains(last.Text, "") { t.Fatalf("expected hidden interrupt marker, got: %+v", last) } + if !strings.Contains(last.Text, "source: esc") { + t.Fatalf("expected interrupt source in marker, got: %+v", last) + } + var canceledAssistant *Message + for i := range msgs { + if msgs[i].Role == RoleAssistant && msgs[i].FinishReason == FinishReasonCanceled { + canceledAssistant = &msgs[i] + } + } + if canceledAssistant == nil || !strings.Contains(canceledAssistant.ErrorDetail, "user interrupt (source: esc)") { + t.Fatalf("expected persisted cancellation cause, got: %+v", canceledAssistant) + } } func TestRunStreamServiceShutdownDoesNotPersistUserInterruptMarker(t *testing.T) { diff --git a/internal/agent/cancellation.go b/internal/agent/cancellation.go index 2c9a9633..d6901c77 100644 --- a/internal/agent/cancellation.go +++ b/internal/agent/cancellation.go @@ -1,12 +1,46 @@ package agent -import "errors" +import ( + "errors" + "strings" +) // ErrUserInterrupt identifies a cancellation requested for the active turn. // Plain context.WithCancel callers remain supported and retain the historical // interrupt behavior. var ErrUserInterrupt = errors.New("user interrupt") +type userInterruptError struct { + source string +} + +func (e userInterruptError) Error() string { + return "user interrupt (source: " + e.source + ")" +} + +func (e userInterruptError) Unwrap() error { + return ErrUserInterrupt +} + +// NewUserInterrupt records the UI or client path that requested cancellation +// while preserving errors.Is(err, ErrUserInterrupt) compatibility. +func NewUserInterrupt(source string) error { + source = strings.TrimSpace(source) + if source == "" { + return ErrUserInterrupt + } + return userInterruptError{source: source} +} + +// UserInterruptSource extracts the source attached by NewUserInterrupt. +func UserInterruptSource(err error) string { + var tagged userInterruptError + if errors.As(err, &tagged) { + return tagged.source + } + return "" +} + // ErrServiceShutdown identifies cancellation caused by tearing down the // service or its owning client. It must not be presented as a user interrupt. var ErrServiceShutdown = errors.New("service shutdown") diff --git a/internal/agent/cancellation_test.go b/internal/agent/cancellation_test.go new file mode 100644 index 00000000..8f7e4741 --- /dev/null +++ b/internal/agent/cancellation_test.go @@ -0,0 +1,43 @@ +package agent + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" +) + +func TestUserInterruptCarriesSource(t *testing.T) { + err := NewUserInterrupt(" ctrl+c ") + if !errors.Is(err, ErrUserInterrupt) { + t.Fatalf("expected user interrupt sentinel, got %v", err) + } + if got := UserInterruptSource(err); got != "ctrl+c" { + t.Fatalf("source = %q, want ctrl+c", got) + } + if got := UserInterruptSource(ErrUserInterrupt); got != "" { + t.Fatalf("sentinel source = %q, want empty", got) + } +} + +func TestCancellationErrorDetailIncludesContextCause(t *testing.T) { + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(NewUserInterrupt("esc")) + err := fmt.Errorf("request failed: %w", context.Canceled) + + got := cancellationErrorDetail(ctx, err) + for _, want := range []string{"request failed: context canceled", "cancel cause: user interrupt (source: esc)"} { + if !strings.Contains(got, want) { + t.Fatalf("detail %q does not contain %q", got, want) + } + } +} + +func TestCancellationErrorDetailLeavesPlainCancelStable(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if got := cancellationErrorDetail(ctx, context.Canceled); got != "context canceled" { + t.Fatalf("detail = %q, want context canceled", got) + } +} diff --git a/internal/agent/interrupt_marker.go b/internal/agent/interrupt_marker.go index ad45e679..f51262eb 100644 --- a/internal/agent/interrupt_marker.go +++ b/internal/agent/interrupt_marker.go @@ -2,12 +2,21 @@ package agent import ( "context" + "strings" "github.com/usewhale/whale/internal/core" ) const interruptedTurnMarkerText = "\nThe user interrupted the previous turn on purpose. Any running tools or commands may have partially executed; verify current state before retrying.\n" +func interruptedTurnMarkerTextForContext(ctx context.Context) string { + source := strings.TrimSpace(UserInterruptSource(context.Cause(ctx))) + if source == "" { + return interruptedTurnMarkerText + } + return "\nsource: " + source + "\nThe user interrupted the previous turn on purpose. Any running tools or commands may have partially executed; verify current state before retrying.\n" +} + func approvalDeniedMarkerText(toolName string) string { if toolName == "" { toolName = "unknown" @@ -15,11 +24,11 @@ func approvalDeniedMarkerText(toolName string) string { return "\nThe user denied a requested tool/action (tool: " + toolName + "). Treat the related task path as canceled. Do not retry, continue, or switch to another tool to bypass the denied action unless the user explicitly asks. If the user asks again, use the normal approval flow for the same capability instead of probing alternative tools that are known to be out of scope.\n" } -func (a *Agent) persistInterruptedTurnMarker(sessionID string) { +func (a *Agent) persistInterruptedTurnMarker(ctx context.Context, sessionID string) { _, _ = a.store.Create(context.Background(), core.Message{ SessionID: sessionID, Role: core.RoleUser, - Text: interruptedTurnMarkerText, + Text: interruptedTurnMarkerTextForContext(ctx), Hidden: true, FinishReason: core.FinishReasonCanceled, }) diff --git a/internal/agent/premature_end_turn.go b/internal/agent/premature_end_turn.go index f73692ef..023e5918 100644 --- a/internal/agent/premature_end_turn.go +++ b/internal/agent/premature_end_turn.go @@ -26,6 +26,7 @@ var prematureActionPrefixes = []string{ "now ", "next ", "then ", + "finally ", "continue ", "start ", "retry ", @@ -49,8 +50,14 @@ var prematureActionPrefixes = []string{ "先", "现在", "接下来", + "下一步", "然后", "再", + "最后是", + "最后一步", + "立即", + "马上", + "随后", "继续", "开始", "重新", diff --git a/internal/agent/premature_end_turn_test.go b/internal/agent/premature_end_turn_test.go index 740a1b65..bfb1ab48 100644 --- a/internal/agent/premature_end_turn_test.go +++ b/internal/agent/premature_end_turn_test.go @@ -133,9 +133,14 @@ func TestShouldRecoverPrematureEndTurn(t *testing.T) { {name: "observed workflow write lead-in", text: "项目已存在（mightty.pages.dev）。先写 workflow：", mode: session.ModeAgent, toolsAvailable: true, want: true}, {name: "observed rerun lead-in", text: "原始 formula 的 URL 不需要改。重新验证只改 version + sha256：", mode: session.ModeAgent, toolsAvailable: true, want: true}, {name: "observed retrigger lead-in", text: "权限已经修正。重新触发 release-please workflow 验证：", mode: session.ModeAgent, toolsAvailable: true, want: true}, + {name: "observed final picker lead-in", text: "最后是 session picker 的循环：", mode: session.ModeAgent, toolsAvailable: true, want: true}, + {name: "observed immediate merge lead-in", text: "CI 已通过，但 merge 没执行成功。立即用正确命令合并：", mode: session.ModeAgent, toolsAvailable: true, want: true}, + {name: "chinese next step lead-in", text: "下一步检查 CI：", mode: session.ModeAgent, toolsAvailable: true, want: true}, {name: "english action lead-in", text: "The config is present. Now verify the workflow:", mode: session.ModeAgent, toolsAvailable: true, want: true}, + {name: "english final action lead-in", text: "The checks passed. Finally merge the pull request:", mode: session.ModeAgent, toolsAvailable: true, want: true}, {name: "complete answer", text: "The workflow is valid.", mode: session.ModeAgent, toolsAvailable: true}, {name: "ordinary heading", text: "Details:", mode: session.ModeAgent, toolsAvailable: true}, + {name: "ordinary final heading", text: "最后：", mode: session.ModeAgent, toolsAvailable: true}, {name: "user choice prompt", text: "请选择：", mode: session.ModeAgent, toolsAvailable: true}, {name: "plan reply", text: "接下来执行：", mode: session.ModePlan, toolsAvailable: true}, {name: "ask reply", text: "现在检查：", mode: session.ModeAsk, toolsAvailable: true}, diff --git a/internal/agent/stream_ingest.go b/internal/agent/stream_ingest.go index c3ff7d63..fc84d574 100644 --- a/internal/agent/stream_ingest.go +++ b/internal/agent/stream_ingest.go @@ -183,7 +183,7 @@ func (a *Agent) collectAssistantStream(ctx context.Context, sessionID string, rt } else { assistant.FinishReason = core.FinishReasonError } - assistant.ErrorDetail = ev.Err.Error() + assistant.ErrorDetail = cancellationErrorDetail(ctx, ev.Err) assistant.ToolCalls = nil a.bestEffortUpdateAssistant(assistant) return core.Message{}, llm.Usage{}, "", nil, ev.Err @@ -212,6 +212,21 @@ func (a *Agent) collectAssistantStream(ctx context.Context, sessionID string, rt return assistant, lastUsage, lastModel, cacheShape, nil } +func cancellationErrorDetail(ctx context.Context, err error) string { + if err == nil { + return "" + } + detail := err.Error() + if !errors.Is(err, context.Canceled) { + return detail + } + cause := context.Cause(ctx) + if cause == nil || cause == context.Canceled { + return detail + } + return detail + " (cancel cause: " + cause.Error() + ")" +} + // messageUsageFrom converts provider usage into the persisted form, // returning nil when the provider reported nothing. func messageUsageFrom(u llm.Usage) *core.MessageUsage { diff --git a/internal/agent/turn_loop.go b/internal/agent/turn_loop.go index 099c191c..a9c41fbb 100644 --- a/internal/agent/turn_loop.go +++ b/internal/agent/turn_loop.go @@ -216,7 +216,7 @@ func (a *Agent) runStreamWithNewMessages(ctx context.Context, sessionID string, if sErr != nil { if errors.Is(sErr, context.Canceled) { if !isServiceShutdown(ctx) { - a.persistInterruptedTurnMarker(sessionID) + a.persistInterruptedTurnMarker(ctx, sessionID) } emit(AgentEvent{Type: AgentEventTypeTurnCancelled, Content: "turn cancelled"}) return @@ -249,7 +249,7 @@ func (a *Agent) runStreamWithNewMessages(ctx context.Context, sessionID string, if ctx.Err() != nil { if errors.Is(ctx.Err(), context.Canceled) { if !isServiceShutdown(ctx) { - a.persistInterruptedTurnMarker(sessionID) + a.persistInterruptedTurnMarker(ctx, sessionID) } return } diff --git a/internal/app/service/dispatch.go b/internal/app/service/dispatch.go index e1842fb9..48c29110 100644 --- a/internal/app/service/dispatch.go +++ b/internal/app/service/dispatch.go @@ -50,7 +50,7 @@ func (s *Service) Dispatch(in Intent) { case IntentShutdown: s.cancelMu.Lock() if s.cancelCause != nil { - s.cancelCause(agent.ErrUserInterrupt) + s.cancelCause(agent.NewUserInterrupt(in.InterruptSource)) } else if s.cancel != nil { s.cancel() } diff --git a/internal/app/service/events_test.go b/internal/app/service/events_test.go index 0e68b9d6..7bceb0f1 100644 --- a/internal/app/service/events_test.go +++ b/internal/app/service/events_test.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "encoding/json" + "errors" "fmt" "net/http" "net/http/httptest" @@ -1296,6 +1297,32 @@ func TestShutdownCancelsPendingInteractions(t *testing.T) { } } +func TestShutdownPreservesInterruptSource(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + turnCtx, turnCancel := context.WithCancelCause(ctx) + svc := &Service{ + ctx: ctx, + events: make(chan Event, 1), + cancelCause: turnCancel, + } + + svc.Dispatch(Intent{Kind: IntentShutdown, InterruptSource: "esc"}) + + select { + case <-turnCtx.Done(): + case <-time.After(2 * time.Second): + t.Fatal("shutdown did not cancel active turn") + } + cause := context.Cause(turnCtx) + if !errors.Is(cause, agent.ErrUserInterrupt) { + t.Fatalf("cause = %v, want user interrupt", cause) + } + if got := agent.UserInterruptSource(cause); got != "esc" { + t.Fatalf("source = %q, want esc", got) + } +} + func TestShutdownRejectsLateInteractions(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/internal/app/service/protocol_dispatch.go b/internal/app/service/protocol_dispatch.go index 3084f864..747846e4 100644 --- a/internal/app/service/protocol_dispatch.go +++ b/internal/app/service/protocol_dispatch.go @@ -9,6 +9,7 @@ import ( func (s *Service) DispatchProtocol(in protocol.Intent) { s.Dispatch(Intent{ Kind: serviceIntentKind(in.Kind), + InterruptSource: in.InterruptSource, Input: in.Input, ClientInputID: in.ClientInputID, HiddenInput: in.HiddenInput, diff --git a/internal/app/service/service.go b/internal/app/service/service.go index 2e113f41..46f48515 100644 --- a/internal/app/service/service.go +++ b/internal/app/service/service.go @@ -55,6 +55,7 @@ const ( type Intent struct { Kind IntentKind + InterruptSource string Input string ClientInputID string HiddenInput bool diff --git a/internal/runtime/protocol/intents.go b/internal/runtime/protocol/intents.go index 4e3894e2..e73673d7 100644 --- a/internal/runtime/protocol/intents.go +++ b/internal/runtime/protocol/intents.go @@ -41,6 +41,7 @@ const ( type Intent struct { Kind IntentKind `json:"kind"` + InterruptSource string `json:"interrupt_source,omitempty"` Input string `json:"input,omitempty"` ClientInputID string `json:"client_input_id,omitempty"` HiddenInput bool `json:"hidden_input,omitempty"` diff --git a/internal/tui/model_busy_queue_test.go b/internal/tui/model_busy_queue_test.go index be6bf246..39c755fa 100644 --- a/internal/tui/model_busy_queue_test.go +++ b/internal/tui/model_busy_queue_test.go @@ -581,7 +581,7 @@ func TestEscWithQueuedPromptSubmitsItAfterInterrupt(t *testing.T) { if !strings.Contains(renderedAfterEsc, "Interrupted to submit queued follow-up") { t.Fatalf("queued Esc interrupt should show queued follow-up notice:\n%s", renderedAfterEsc) } - if len(*intents) != 1 || (*intents)[0].Kind != protocol.IntentShutdown { + if len(*intents) != 1 || (*intents)[0].Kind != protocol.IntentShutdown || (*intents)[0].InterruptSource != "esc" { t.Fatalf("expected shutdown intent from Esc, got %+v", *intents) } @@ -612,7 +612,7 @@ func TestEscWithDraftQueuesAndSubmitsItAfterInterrupt(t *testing.T) { if !m.submitQueuedPromptAfterInterrupt { t.Fatal("expected Esc with draft to queue and request immediate submit") } - if len(*intents) != 1 || (*intents)[0].Kind != protocol.IntentShutdown { + if len(*intents) != 1 || (*intents)[0].Kind != protocol.IntentShutdown || (*intents)[0].InterruptSource != "esc" { t.Fatalf("expected shutdown intent from Esc, got %+v", *intents) } @@ -675,7 +675,7 @@ func TestCtrlCWhileBusyInterruptsWithoutArmingQuit(t *testing.T) { if !m.quitArmedUntil.IsZero() { t.Fatal("expected ctrl+c while busy not to arm quit") } - if len(*intents) != 1 || (*intents)[0].Kind != protocol.IntentShutdown { + if len(*intents) != 1 || (*intents)[0].Kind != protocol.IntentShutdown || (*intents)[0].InterruptSource != "ctrl+c" { t.Fatalf("expected ctrl+c while busy to dispatch shutdown intent, got %+v", *intents) } diff --git a/internal/tui/model_event_handlers.go b/internal/tui/model_event_handlers.go index 5cf2dd12..f1672c57 100644 --- a/internal/tui/model_event_handlers.go +++ b/internal/tui/model_event_handlers.go @@ -660,5 +660,5 @@ func (m *model) handleSessionHydratedEvent(ev protocol.Event) tea.Cmd { func (m *model) handleExitRequestedEvent() { m.clearProviderRetryStatus() - m.dispatchIntent(protocol.Intent{Kind: protocol.IntentShutdown}) + m.dispatchIntent(protocol.Intent{Kind: protocol.IntentShutdown, InterruptSource: "exit_requested"}) } diff --git a/internal/tui/model_keys.go b/internal/tui/model_keys.go index acf66bb3..7e2e33bc 100644 --- a/internal/tui/model_keys.go +++ b/internal/tui/model_keys.go @@ -25,7 +25,7 @@ func (m *model) handleKeyMsg(msg tea.KeyMsg) (tea.Cmd, bool, bool) { } if m.mode == modeChat && m.page == pageDiff { if msg.String() == "ctrl+c" && m.busy { - return m.interruptBusyTurn(), false, true + return m.interruptBusyTurn("ctrl+c"), false, true } if msg.String() == "ctrl+c" { return m.handleGlobalKey(msg) @@ -67,7 +67,7 @@ func (m *model) handleKeyMsg(msg tea.KeyMsg) (tea.Cmd, bool, bool) { // textarea, so a Ctrl+C arriving inside that window would otherwise // see an empty textarea and incorrectly interrupt the running turn. if msg.String() == "ctrl+c" && m.busy && (m.mode != modeChat || (m.input.Value() == "" && !m.hasWindowsPasteBuffer())) { - return m.interruptBusyTurn(), false, true + return m.interruptBusyTurn("ctrl+c"), false, true } if m.mode == modeChat { if cmd, handled := m.handleWindowsPasteFallbackKey(msg); handled { @@ -141,7 +141,7 @@ func (m model) shouldRouteWindowsPasteFallbackBeforeLayout(msg tea.KeyMsg) bool len(msg.Runes) > 0 } -func (m *model) interruptBusyTurn() tea.Cmd { +func (m *model) interruptBusyTurn(source string) tea.Cmd { alreadyStopping := m.stopping m.cancelBlockingModalForInterrupt(!alreadyStopping) if alreadyStopping { @@ -173,7 +173,7 @@ func (m *model) interruptBusyTurn() tea.Cmd { m.stoppingInterruptCount = 0 m.quitArmedUntil = time.Time{} if m.runtime != nil { - m.dispatchIntent(protocol.Intent{Kind: protocol.IntentShutdown}) + m.dispatchIntent(protocol.Intent{Kind: protocol.IntentShutdown, InterruptSource: source}) } m.status = "stopping" m.stopping = true diff --git a/internal/tui/model_keys_chat.go b/internal/tui/model_keys_chat.go index 13512f3e..d500644e 100644 --- a/internal/tui/model_keys_chat.go +++ b/internal/tui/model_keys_chat.go @@ -98,7 +98,7 @@ func (m *model) handleChatModeKey(msg tea.KeyMsg) (tea.Cmd, bool) { case "esc": if m.busy { m.prepareQueuedPromptAfterInterrupt() - return m.interruptBusyTurn(), true + return m.interruptBusyTurn("esc"), true } if m.page != pageChat { m.page = pageChat diff --git a/internal/tui/model_keys_modals.go b/internal/tui/model_keys_modals.go index bafdea77..89bc41b7 100644 --- a/internal/tui/model_keys_modals.go +++ b/internal/tui/model_keys_modals.go @@ -166,7 +166,7 @@ func (m *model) handleUserInputKey(msg tea.KeyMsg) tea.Cmd { switch msg.String() { case "esc": if m.busy { - return m.interruptBusyTurn() + return m.interruptBusyTurn("esc") } m.dispatchIntent(protocol.Intent{Kind: protocol.IntentCancelUserInput, ToolCallID: m.userInput.toolCallID}) m.mode = modeChat