diff --git a/internal/handler/web.go b/internal/handler/web.go index 1734e308..93cf5222 100644 --- a/internal/handler/web.go +++ b/internal/handler/web.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "strings" "sync" @@ -446,9 +447,11 @@ type WebSubagentProgressData struct { // WebDoneData signals agent completion. Error is a short user-facing summary; // Detail carries the full raw error text for a collapsible "details" view. +// Stopped marks a user-initiated stop — the UI shows a calm notice, not an error. type WebDoneData struct { - Error string `json:"error,omitempty"` - Detail string `json:"detail,omitempty"` + Error string `json:"error,omitempty"` + Detail string `json:"detail,omitempty"` + Stopped bool `json:"stopped,omitempty"` } // WebApprovalRequestData carries an approval request. ToolCallID (when known) @@ -626,6 +629,12 @@ func (h *WebHandler) OnAgentDone(err error) { h.emit("agent_done", WebDoneData{}) return } + // User-initiated stop (the runner reports the clean context error): show a + // calm "stopped" notice, not a red error card. + if errors.Is(err, context.Canceled) { + h.emit("agent_done", WebDoneData{Stopped: true}) + return + } // Raw run errors (eino NodeRunError wrapping go-openai API errors) are too // noisy for the timeline — send a one-line summary plus the raw detail. summary, detail := internalmodel.SummarizeRunError(err) diff --git a/internal/runner/cancel_backfill_test.go b/internal/runner/cancel_backfill_test.go new file mode 100644 index 00000000..77882224 --- /dev/null +++ b/internal/runner/cancel_backfill_test.go @@ -0,0 +1,201 @@ +package runner + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/cloudwego/eino/adk" + einomodel "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/components/tool" + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/schema" + + "github.com/cnjack/jcode/internal/handler" + "github.com/cnjack/jcode/internal/session" +) + +// blockingTool simulates a long-running tool: it blocks until its context is +// canceled (user stop) and then returns the context error. +type blockingTool struct { + info *schema.ToolInfo +} + +func newBlockingTool() *blockingTool { + return &blockingTool{info: &schema.ToolInfo{ + Name: "block", + Desc: "blocks until the run is cancelled", + ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{ + "note": {Type: schema.String, Desc: "ignored"}, + }), + }} +} + +func (bt *blockingTool) Info(context.Context) (*schema.ToolInfo, error) { return bt.info, nil } + +func (bt *blockingTool) InvokableRun(ctx context.Context, _ string, _ ...tool.Option) (string, error) { + <-ctx.Done() + return "", ctx.Err() +} + +// cancelModel streams one assistant message carrying a call to the blocking +// tool; if the run ever continues past the tool it answers with plain text. +type cancelModel struct{} + +func (m *cancelModel) WithTools([]*schema.ToolInfo) (einomodel.ToolCallingChatModel, error) { + return m, nil +} + +func (m *cancelModel) Generate(context.Context, []*schema.Message, ...einomodel.Option) (*schema.Message, error) { + return nil, errors.New("Generate is not used: streaming is enabled") +} + +func (m *cancelModel) Stream(_ context.Context, input []*schema.Message, _ ...einomodel.Option) (*schema.StreamReader[*schema.Message], error) { + if last := input[len(input)-1]; last.Role == schema.Tool { + return schema.StreamReaderFromArray([]*schema.Message{ + {Role: schema.Assistant, Content: "done"}, + }), nil + } + return schema.StreamReaderFromArray([]*schema.Message{{ + Role: schema.Assistant, + ToolCalls: []schema.ToolCall{{ + ID: "call-block-1", + Function: schema.FunctionCall{Name: "block", Arguments: "{}"}, + }}, + }}), nil +} + +// cancelRecordingHandler cancels the run as soon as the tool call is +// announced (the user hitting Stop) and records every tool result. +type cancelRecordingHandler struct { + stubHandler + cancel context.CancelFunc + done chan struct{} + mu sync.Mutex + results []handler.ToolResultEvent + doneErr error +} + +func (h *cancelRecordingHandler) OnToolCall(handler.ToolCallEvent) { h.cancel() } + +func (h *cancelRecordingHandler) OnToolResult(ev handler.ToolResultEvent) { + h.mu.Lock() + defer h.mu.Unlock() + h.results = append(h.results, ev) +} + +func (h *cancelRecordingHandler) OnAgentDone(err error) { + h.mu.Lock() + h.doneErr = err + h.mu.Unlock() + close(h.done) +} + +// TestRunInnerCancellationBackfillsToolResult stops the run while a tool is +// still executing. The announced tool call must not be left dangling: exactly +// one result reaches the handler, and the persisted session reconstructs into +// a history that satisfies the model API's tool-call/tool-message invariant +// (previously a resume of such a session failed with "assistant message with +// 'tool_calls' must be followed by tool messages..."). +func TestRunInnerCancellationBackfillsToolResult(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ag, err := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{ + Name: "cancel-test", + Description: "cancel-test", + Instruction: "test", + Model: &cancelModel{}, + ToolsConfig: adk.ToolsConfig{ + ToolsNodeConfig: compose.ToolsNodeConfig{ + Tools: []tool.BaseTool{newBlockingTool()}, + }, + }, + MaxIterations: 5, + }) + if err != nil { + t.Fatalf("create agent: %v", err) + } + + rec, err := session.NewRecorder("cancel-test", "test", "test") + if err != nil { + t.Fatalf("create recorder: %v", err) + } + h := &cancelRecordingHandler{cancel: cancel, done: make(chan struct{})} + + finished := make(chan bool, 1) + go func() { + _, done := runInner(ctx, ag, []adk.Message{schema.UserMessage("go")}, h, rec) + finished <- done + }() + + select { + case done := <-finished: + if !done { + t.Errorf("runInner done = false, want true after cancellation") + } + case <-time.After(10 * time.Second): + t.Fatal("runInner did not return after cancellation") + } + + select { + case <-h.done: + case <-time.After(time.Second): + t.Fatal("OnAgentDone was not called") + } + h.mu.Lock() + defer h.mu.Unlock() + if !errors.Is(h.doneErr, context.Canceled) { + t.Errorf("OnAgentDone err = %v, want context.Canceled", h.doneErr) + } + + // The announced call got exactly one result (drain backfill or a folded + // framework result — either satisfies the invariant). + if len(h.results) != 1 { + t.Fatalf("tool results = %d, want exactly 1", len(h.results)) + } + if h.results[0].ToolCallID != "call-block-1" { + t.Errorf("result ToolCallID = %q, want call-block-1", h.results[0].ToolCallID) + } + + // The session on disk pairs the recorded call with a recorded result. + entries, err := session.LoadSession(rec.UUID()) + if err != nil { + t.Fatalf("load session: %v", err) + } + var calls, results int + for _, e := range entries { + switch e.Type { + case session.EntryToolCall: + calls++ + case session.EntryToolResult: + results++ + if e.ToolCallID != "call-block-1" { + t.Errorf("recorded result for %q, want call-block-1", e.ToolCallID) + } + } + } + if calls != 1 || results != 1 { + t.Errorf("recorded calls=%d results=%d, want 1/1", calls, results) + } + + state := session.ReconstructState(entries) + for i, m := range state.History { + if m.Role != schema.Assistant || len(m.ToolCalls) == 0 { + continue + } + answered := map[string]bool{} + for j := i + 1; j < len(state.History) && state.History[j].Role == schema.Tool; j++ { + answered[state.History[j].ToolCallID] = true + } + for _, tc := range m.ToolCalls { + if !answered[tc.ID] { + t.Errorf("reconstructed history: tool_call %s has no answering tool message", tc.ID) + } + } + } +} diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 09945887..5a5b6630 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -238,7 +238,11 @@ func runInner( // toolStarts records when each tool call was announced so results can // carry a call→result latency. The event loop below is the only reader // and writer (single goroutine), so no locking is needed. - toolStarts := make(map[string]time.Time) + type toolStart struct { + at time.Time + name string + } + toolStarts := make(map[string]toolStart) // meter carries approval wait/denied outcomes from the approval path (see // Run) so the emitted Duration is pure execution time and denied calls are // flagged. emitToolResult also owns session recording so the persisted @@ -247,7 +251,7 @@ func runInner( emitToolResult := func(name, output, toolCallID string, err error) { ev := handler.ToolResultEvent{Name: name, Output: output, ToolCallID: toolCallID, Err: err} if started, ok := toolStarts[toolCallID]; ok { - ev.Duration = time.Since(started) + ev.Duration = time.Since(started.at) delete(toolStarts, toolCallID) } applyApprovalOutcome(&ev, meter) @@ -256,6 +260,19 @@ func runInner( rec.RecordToolResult(name, output, toolCallID, err, ev.Denied, ev.Duration) } } + // drainDanglingToolResults backfills a result for every announced tool + // call that never produced one (user stop, fatal tool-node failure). The + // session must never persist a tool_call without its tool_result: the next + // resume would reconstruct a history the model API rejects ("assistant + // message with 'tool_calls' must be followed by tool messages..."). The + // backfill carries no error: an interrupted call is not a failed call, and + // a non-nil Err would paint every front-end's tool row red (raw + // "context.Canceled" text) right next to the calm stop notice. + drainDanglingToolResults := func() { + for id, started := range toolStarts { + emitToolResult(started.name, session.InterruptedToolOutput, id, nil) + } + } config.Logger().Printf("[runner] runInner start, messages=%d", len(messages)) iterator := ag.Run(ctx, input) @@ -270,6 +287,7 @@ func runInner( // calm "Stopped". runInner owns this OnAgentDone (returns done=true), // so Run does not emit a second one. config.Logger().Printf("[runner] context cancelled, stopping iteration") + drainDanglingToolResults() h.OnAgentDone(ctx.Err()) return assistantText.String(), true default: @@ -290,6 +308,7 @@ func runInner( // "[NodeRunError] context canceled"); report the clean context // error instead of the noisy wrapped one. config.Logger().Printf("[runner] event error during cancellation: %v", event.Err) + drainDanglingToolResults() h.OnAgentDone(ctx.Err()) return assistantText.String(), true } @@ -298,6 +317,7 @@ func runInner( // so wrapping here fixes the display in the TUI, the web UI and ACP // at once — and stops the next frontend from having to remember. config.Logger().Printf("[runner] event error: %v", event.Err) + drainDanglingToolResults() h.OnAgentDone(internalmodel.WrapFriendly(event.Err, "", "")) return assistantText.String(), true } @@ -401,7 +421,7 @@ func runInner( startedAt := time.Now() for i, idx := range indices { p := pending[idx] - toolStarts[p.id] = startedAt + toolStarts[p.id] = toolStart{at: startedAt, name: p.name} h.OnToolCall(handler.ToolCallEvent{ Name: p.name, Args: p.args.String(), @@ -422,7 +442,7 @@ func runInner( startedAt := time.Now() size := len(mo.Message.ToolCalls) for i, tc := range mo.Message.ToolCalls { - toolStarts[tc.ID] = startedAt + toolStarts[tc.ID] = toolStart{at: startedAt, name: tc.Function.Name} h.OnToolCall(handler.ToolCallEvent{ Name: tc.Function.Name, Args: tc.Function.Arguments, diff --git a/internal/session/history.go b/internal/session/history.go index 38dd2dee..22c95b39 100644 --- a/internal/session/history.go +++ b/internal/session/history.go @@ -2,6 +2,7 @@ package session import ( "encoding/json" + "slices" "github.com/cloudwego/eino/adk" "github.com/cloudwego/eino/schema" @@ -138,7 +139,11 @@ type SessionState struct { // It is compact-aware: if a compact entry is found, messages before it are // replaced with the compact summary. // -// Subagent-internal entries are skipped. +// Subagent marker entries (subagent_start/result/async) carry no conversation +// content and are skipped — but entries BETWEEN them are NOT skipped: subagent +// internal tool calls are never persisted (only the main runner records +// tool_call/tool_result), so entries in that window are the main agent's own +// parallel tool calls and must be kept. func ReconstructState(entries []Entry) *SessionState { state := &SessionState{ EnvTarget: "local", @@ -146,25 +151,11 @@ func ReconstructState(entries []Entry) *SessionState { var msgs []adk.Message var lastTarget string - var subagentDepth int for _, e := range entries { - // Track subagent boundaries first. + // Subagent boundary markers are informational (TUI replay) only. switch e.Type { - case EntrySubagentStart: - subagentDepth++ - continue - case EntrySubagentResult: - if subagentDepth > 0 { - subagentDepth-- - } - continue - case EntrySubagentAsync: - continue - } - - // Skip entries that belong to a running subagent. - if subagentDepth > 0 { + case EntrySubagentStart, EntrySubagentResult, EntrySubagentAsync: continue } @@ -279,6 +270,49 @@ func ReconstructState(entries []Entry) *SessionState { } } - state.History = msgs + state.History = repairDanglingToolCalls(msgs) return state } + +// InterruptedToolOutput is the placeholder content backfilled for tool calls +// whose result never made it to disk (user stop, process kill, or a recording +// gap). It tells the model what happened instead of fabricating an output. +// The runner records the same marker for interrupted calls so live sessions +// and reconstructed history stay identical. +const InterruptedToolOutput = "[Interrupted before result was recorded]" + +// repairDanglingToolCalls returns msgs with placeholder tool messages inserted +// so every assistant tool_call is answered by a matching tool message. A +// session recorded mid-run (user stop, process kill) otherwise reconstructs +// into a history the model API rejects: "an assistant message with +// 'tool_calls' must be followed by tool messages responding to each +// 'tool_call_id'". +func repairDanglingToolCalls(msgs []adk.Message) []adk.Message { + for i := 0; i < len(msgs); i++ { + m := msgs[i] + if m.Role != schema.Assistant || len(m.ToolCalls) == 0 { + continue + } + // Collect the call IDs answered by the tool messages of this group. + answered := make(map[string]bool, len(m.ToolCalls)) + j := i + 1 + for j < len(msgs) && msgs[j].Role == schema.Tool { + answered[msgs[j].ToolCallID] = true + j++ + } + var fill []adk.Message + for _, tc := range m.ToolCalls { + if !answered[tc.ID] { + fill = append(fill, schema.ToolMessage( + InterruptedToolOutput, tc.ID, schema.WithToolName(tc.Function.Name))) + } + } + if len(fill) == 0 { + continue + } + // Insert right after the group's last tool message (or directly after + // the assistant message when no result was recorded at all). + msgs = slices.Insert(msgs, j, fill...) + } + return msgs +} diff --git a/internal/session/history_test.go b/internal/session/history_test.go index 79c047a2..b4ec18c6 100644 --- a/internal/session/history_test.go +++ b/internal/session/history_test.go @@ -3,9 +3,31 @@ package session import ( "testing" + "github.com/cloudwego/eino/adk" "github.com/cloudwego/eino/schema" ) +// assertToolCallInvariant fails the test when any assistant message carries a +// tool_call that is not answered by a tool message in the group immediately +// following it — the exact condition model APIs reject with a 400. +func assertToolCallInvariant(t *testing.T, msgs []adk.Message) { + t.Helper() + for i, m := range msgs { + if m.Role != schema.Assistant || len(m.ToolCalls) == 0 { + continue + } + answered := make(map[string]bool, len(m.ToolCalls)) + for j := i + 1; j < len(msgs) && msgs[j].Role == schema.Tool; j++ { + answered[msgs[j].ToolCallID] = true + } + for _, tc := range m.ToolCalls { + if !answered[tc.ID] { + t.Errorf("msg[%d]: tool_call %s(%s) has no answering tool message", i, tc.Function.Name, tc.ID) + } + } + } +} + // TestReconstructState_CompactKeepsTail verifies that replaying a compact entry // keeps the KeptN most recent messages after the summary, matching what the // live agent kept in memory at compaction time (previously the tail was @@ -104,6 +126,98 @@ func TestReconstructState_CompactKeptNOverflow(t *testing.T) { } } +// TestReconstructState_ParallelToolResultInSubagentWindow reproduces the +// session-86e09b12 failure: the main agent issued grep+subagent in one +// parallel batch and the fast grep's result landed between subagent_start and +// subagent_result. The old subagentDepth skip window swallowed it, leaving a +// dangling grep tool_call the model API rejected on resume ("tool_call_ids +// did not have response messages: grep:53"). +func TestReconstructState_ParallelToolResultInSubagentWindow(t *testing.T) { + entries := []Entry{ + {Type: EntryUser, Content: "u"}, + {Type: EntryToolCall, Name: "grep", Args: "{}", ToolCallID: "c-grep"}, + {Type: EntryToolCall, Name: "subagent", Args: "{}", ToolCallID: "c-sub"}, + {Type: EntrySubagentStart, SubagentName: "s", SubagentType: "explore"}, + {Type: EntryToolResult, Name: "grep", Output: "grep-out", ToolCallID: "c-grep"}, + {Type: EntrySubagentResult, SubagentName: "s", Output: "sub-out"}, + {Type: EntryToolResult, Name: "subagent", Output: "sub-out", ToolCallID: "c-sub"}, + {Type: EntryAssistant, Content: "done"}, + } + state := ReconstructState(entries) + + assertToolCallInvariant(t, state.History) + // Both results must survive, in recorded order, right after the assistant. + if len(state.History) != 5 { + t.Fatalf("History length = %d, want 5 (user, assistant, 2 tool results, assistant)", len(state.History)) + } + if state.History[2].Role != schema.Tool || state.History[2].ToolCallID != "c-grep" || state.History[2].Content != "grep-out" { + t.Errorf("History[2] = %v %q, want grep tool result", state.History[2].Role, state.History[2].Content) + } + if state.History[3].Role != schema.Tool || state.History[3].ToolCallID != "c-sub" { + t.Errorf("History[3] = %v, want subagent tool result", state.History[3].Role) + } +} + +// TestReconstructState_BackfillsInterruptedToolCall covers a session recorded +// mid-run (user stop / process kill): the tool_call is on disk but its result +// never arrived. Reconstruction must insert a placeholder tool message so the +// rebuilt history satisfies the model API's tool-call invariant. +func TestReconstructState_BackfillsInterruptedToolCall(t *testing.T) { + entries := []Entry{ + {Type: EntryUser, Content: "u"}, + {Type: EntryToolCall, Name: "grep", Args: "{}", ToolCallID: "c1"}, + {Type: EntryToolCall, Name: "execute", Args: "{}", ToolCallID: "c2"}, + {Type: EntryToolResult, Name: "grep", Output: "ok", ToolCallID: "c1"}, + // execute never produced a result. + } + state := ReconstructState(entries) + + assertToolCallInvariant(t, state.History) + if len(state.History) != 4 { + t.Fatalf("History length = %d, want 4 (user, assistant, 2 tool messages)", len(state.History)) + } + fill := state.History[3] + if fill.Role != schema.Tool || fill.ToolCallID != "c2" || fill.ToolName != "execute" { + t.Errorf("backfilled message = %v name=%q id=%q, want execute tool message for c2", fill.Role, fill.ToolName, fill.ToolCallID) + } + if fill.Content != InterruptedToolOutput { + t.Errorf("backfilled content = %q, want %q", fill.Content, InterruptedToolOutput) + } +} + +// TestReconstructState_BackfillPreservesFollowingMessages ensures the +// placeholder lands inside the right tool group when later turns exist. +func TestReconstructState_BackfillPreservesFollowingMessages(t *testing.T) { + entries := []Entry{ + {Type: EntryUser, Content: "u1"}, + {Type: EntryToolCall, Name: "grep", Args: "{}", ToolCallID: "c1"}, + // interrupted here; then the session was resumed and continued. + {Type: EntryUser, Content: "u2"}, + {Type: EntryAssistant, Content: "a2"}, + } + state := ReconstructState(entries) + + assertToolCallInvariant(t, state.History) + want := []struct { + role schema.RoleType + content string + }{ + {schema.User, "u1"}, + {schema.Assistant, ""}, + {schema.Tool, InterruptedToolOutput}, + {schema.User, "u2"}, + {schema.Assistant, "a2"}, + } + if len(state.History) != len(want) { + t.Fatalf("History length = %d, want %d", len(state.History), len(want)) + } + for i, w := range want { + if state.History[i].Role != w.role || state.History[i].Content != w.content { + t.Errorf("History[%d] = %v %q, want %v %q", i, state.History[i].Role, state.History[i].Content, w.role, w.content) + } + } +} + func TestPruneOldToolOutputsClearsScreenshotPixels(t *testing.T) { encoded := "base64-must-not-survive" shot := schema.ToolMessage("", "shot-call", schema.WithToolName("computer_screenshot")) diff --git a/internal/web/chat.go b/internal/web/chat.go index 6017f0b3..e5d55411 100644 --- a/internal/web/chat.go +++ b/internal/web/chat.go @@ -314,8 +314,9 @@ func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) { cancel() } - // Notify clients on that task's channel. - eng.handler.OnAgentDone(fmt.Errorf("stopped by user")) + // The runner owns the run lifecycle: it observes the cancellation and emits + // the single OnAgentDone(context.Canceled), which the web handler surfaces + // as a calm "stopped" notice. Emitting one here too would double-report. writeJSON(w, http.StatusOK, map[string]string{"status": "stopped"}) } diff --git a/web/src/app/store.ts b/web/src/app/store.ts index b599da6a..58e76e52 100644 --- a/web/src/app/store.ts +++ b/web/src/app/store.ts @@ -231,7 +231,7 @@ const chatSlice = createSlice({ // Session was deleted — its stash can never drain again. delete s.queuedBySession[a.payload] }, - agentDone(s, a: { payload: { error?: string; detail?: string } | undefined }) { + agentDone(s, a: { payload: { error?: string; detail?: string; stopped?: boolean } | undefined }) { // Stamp duration on the last assistant message. for (let i = s.timeline.length - 1; i >= 0; i--) { const item = s.timeline[i] @@ -250,7 +250,20 @@ const chatSlice = createSlice({ item.data.awaitingApproval = undefined } } - if (a.payload?.error) { + if (a.payload?.stopped) { + // Manual stop — a calm muted notice, not an error card. + s.timeline.push({ + kind: 'message', + data: { + id: genId('sys'), + role: 'system', + content: 'Stopped by user', + timestamp: Date.now(), + level: 'notice', + }, + seq: nextSeq(), + }) + } else if (a.payload?.error) { s.timeline.push({ kind: 'message', data: { diff --git a/web/src/app/wsBridge.ts b/web/src/app/wsBridge.ts index d51c6f9d..078e084e 100644 --- a/web/src/app/wsBridge.ts +++ b/web/src/app/wsBridge.ts @@ -70,7 +70,7 @@ export function createWSHandlers( const activeId = getState().session.currentSessionId const isForeground = !taskId || taskId === activeId if (isForeground) { - dispatch(chatActions.agentDone(d ? { error: d.error, detail: d.detail } : undefined)) + dispatch(chatActions.agentDone(d ? { error: d.error, detail: d.detail, stopped: d.stopped } : undefined)) } // Refresh sidebar metadata (title / updated_at / running) after a turn. void dispatch(loadTasks() as never) diff --git a/web/src/lib/ws.ts b/web/src/lib/ws.ts index 88556bd7..5df9f46b 100644 --- a/web/src/lib/ws.ts +++ b/web/src/lib/ws.ts @@ -46,7 +46,7 @@ export interface WSHandlers { presentation?: import('./types').ToolResultPresentation }) => void onTokenUpdate?: (data: import('./types').TokenUpdateData) => void - onAgentDone?: (data: { error?: string; detail?: string; task_id?: string }) => void + onAgentDone?: (data: { error?: string; detail?: string; stopped?: boolean; task_id?: string }) => void onTodoUpdate?: () => void onGoalUpdate?: (data: import('jcode-ui-core').Goal | null) => void onApprovalRequest?: (data: import('./types').ApprovalRequestData) => void