diff --git a/coderd/apidoc/docs.go b/coderd/apidoc/docs.go index ba155d7992..2ac5aa5c0c 100644 --- a/coderd/apidoc/docs.go +++ b/coderd/apidoc/docs.go @@ -15970,6 +15970,9 @@ const docTemplate = `{ "result_delta": { "type": "string" }, + "result_reset": { + "type": "boolean" + }, "signature": { "type": "string" }, diff --git a/coderd/apidoc/swagger.json b/coderd/apidoc/swagger.json index 56b4e46414..81300bbedc 100644 --- a/coderd/apidoc/swagger.json +++ b/coderd/apidoc/swagger.json @@ -14391,6 +14391,9 @@ "result_delta": { "type": "string" }, + "result_reset": { + "type": "boolean" + }, "signature": { "type": "string" }, diff --git a/coderd/x/chatd/chatadvisor/runner.go b/coderd/x/chatd/chatadvisor/runner.go index a3d144967c..da8e0cb326 100644 --- a/coderd/x/chatd/chatadvisor/runner.go +++ b/coderd/x/chatd/chatadvisor/runner.go @@ -3,18 +3,29 @@ package chatadvisor import ( "context" "strings" + "time" "charm.land/fantasy" "golang.org/x/xerrors" "github.com/coder/coder/v2/coderd/x/chatd/chatloop" + "github.com/coder/coder/v2/coderd/x/chatd/chatretry" + "github.com/coder/coder/v2/codersdk" ) +// RunAdvisorOptions carries optional streaming callbacks for a +// single RunAdvisor invocation. +type RunAdvisorOptions struct { + OnAdviceDelta func(delta string) + OnAdviceReset func() +} + // RunAdvisor executes a single, tool-less nested advisor call. func (rt *Runtime) RunAdvisor( ctx context.Context, question string, conversationSnapshot []fantasy.Message, + opts *RunAdvisorOptions, ) (AdvisorResult, error) { // Model, MaxUsesPerRun, and MaxOutputTokens are validated by NewRuntime. // Runtime fields are unexported so callers cannot bypass that. @@ -37,7 +48,7 @@ func (rt *Runtime) RunAdvisor( resetProviderOptionsForNestedCall(nestedProviderOptions) var persistedStep chatloop.PersistedStep - runOpts := chatloop.RunOptions{ + chatLoopOpts := chatloop.RunOptions{ Model: rt.cfg.Model, Messages: BuildAdvisorMessages(question, conversationSnapshot), MaxSteps: 1, @@ -48,8 +59,23 @@ func (rt *Runtime) RunAdvisor( return nil }, } + if opts != nil && opts.OnAdviceDelta != nil { + chatLoopOpts.PublishMessagePart = func(role codersdk.ChatMessageRole, part codersdk.ChatMessagePart) { + if role != codersdk.ChatMessageRoleAssistant || + part.Type != codersdk.ChatMessagePartTypeText || + part.Text == "" { + return + } + opts.OnAdviceDelta(part.Text) + } + } + if opts != nil && opts.OnAdviceReset != nil { + chatLoopOpts.OnRetry = func(int, error, chatretry.ClassifiedError, time.Duration) { + opts.OnAdviceReset() + } + } - if err := chatloop.Run(ctx, runOpts); err != nil { + if err := chatloop.Run(ctx, chatLoopOpts); err != nil { // Refund the use so a transient provider failure does not // permanently exhaust the per-run advisor budget. rt.release() diff --git a/coderd/x/chatd/chatadvisor/runner_test.go b/coderd/x/chatd/chatadvisor/runner_test.go index ec81328274..c0fd90262e 100644 --- a/coderd/x/chatd/chatadvisor/runner_test.go +++ b/coderd/x/chatd/chatadvisor/runner_test.go @@ -48,7 +48,7 @@ func TestAdvisorRunAdvice(t *testing.T) { result, err := runtime.RunAdvisor(t.Context(), question, []fantasy.Message{ textMessage(fantasy.MessageRoleSystem, "existing system"), textMessage(fantasy.MessageRoleUser, "hello"), - }) + }, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeAdvice, result.Type) require.Equal(t, "Take the smallest safe change.", result.Advice) @@ -63,6 +63,122 @@ func TestAdvisorRunAdvice(t *testing.T) { require.Equal(t, question, singleText(t, capturedCall.Prompt[len(capturedCall.Prompt)-1])) } +func TestAdvisorRunStreamsAdviceDeltas(t *testing.T) { + t.Parallel() + + var deltas []string + runtime, err := chatadvisor.NewRuntime(chatadvisor.RuntimeConfig{ + Model: &chattest.FakeModel{ + ProviderName: "test-provider", + ModelName: "test-model", + StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) { + return streamFromParts([]fantasy.StreamPart{ + {Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "Use "}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "the smaller "}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "diff."}, + {Type: fantasy.StreamPartTypeTextEnd, ID: "text-1"}, + {Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop}, + }), nil + }, + }, + MaxUsesPerRun: 2, + MaxOutputTokens: 128, + }) + require.NoError(t, err) + + result, err := runtime.RunAdvisor(t.Context(), "what should I do?", nil, &chatadvisor.RunAdvisorOptions{ + OnAdviceDelta: func(delta string) { + deltas = append(deltas, delta) + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"Use ", "the smaller ", "diff."}, deltas) + require.Equal(t, chatadvisor.ResultTypeAdvice, result.Type) + require.Equal(t, "Use the smaller diff.", result.Advice) + require.Equal(t, 1, result.RemainingUses) +} + +func TestAdvisorRunResetsAdviceDeltasOnRetry(t *testing.T) { + t.Parallel() + + var ( + calls int + events []string + ) + runtime, err := chatadvisor.NewRuntime(chatadvisor.RuntimeConfig{ + Model: &chattest.FakeModel{ + ProviderName: "test-provider", + ModelName: "test-model", + StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) { + calls++ + if calls == 1 { + return streamFromParts([]fantasy.StreamPart{ + {Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "stale "}, + {Type: fantasy.StreamPartTypeError, Error: xerrors.New("received status 429 from upstream")}, + }), nil + } + return streamFromParts([]fantasy.StreamPart{ + {Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "fresh advice"}, + {Type: fantasy.StreamPartTypeTextEnd, ID: "text-1"}, + {Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop}, + }), nil + }, + }, + MaxUsesPerRun: 2, + MaxOutputTokens: 128, + }) + require.NoError(t, err) + + result, err := runtime.RunAdvisor(t.Context(), "what should I do?", nil, &chatadvisor.RunAdvisorOptions{ + OnAdviceDelta: func(delta string) { + events = append(events, "delta:"+delta) + }, + OnAdviceReset: func() { + events = append(events, "reset") + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"delta:stale ", "reset", "delta:fresh advice"}, events) + require.Equal(t, chatadvisor.ResultTypeAdvice, result.Type) + require.Equal(t, "fresh advice", result.Advice) +} + +func TestAdvisorRunErrorAfterPartialDelta(t *testing.T) { + t.Parallel() + + var deltas []string + runtime, err := chatadvisor.NewRuntime(chatadvisor.RuntimeConfig{ + Model: &chattest.FakeModel{ + ProviderName: "test-provider", + ModelName: "test-model", + StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) { + return streamFromParts([]fantasy.StreamPart{ + {Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "partial advice"}, + {Type: fantasy.StreamPartTypeError, Error: xerrors.New("boom after partial")}, + }), nil + }, + }, + MaxUsesPerRun: 1, + MaxOutputTokens: 128, + }) + require.NoError(t, err) + + result, err := runtime.RunAdvisor(t.Context(), "what should I do?", nil, &chatadvisor.RunAdvisorOptions{ + OnAdviceDelta: func(delta string) { + deltas = append(deltas, delta) + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"partial advice"}, deltas) + require.Equal(t, chatadvisor.ResultTypeError, result.Type) + require.Contains(t, result.Error, "boom after partial") + require.Equal(t, 1, result.RemainingUses) +} + func TestAdvisorRunLimitReached(t *testing.T) { t.Parallel() @@ -86,12 +202,12 @@ func TestAdvisorRunLimitReached(t *testing.T) { }) require.NoError(t, err) - first, err := runtime.RunAdvisor(t.Context(), "first?", nil) + first, err := runtime.RunAdvisor(t.Context(), "first?", nil, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeAdvice, first.Type) require.Equal(t, 0, first.RemainingUses) - second, err := runtime.RunAdvisor(t.Context(), "second?", nil) + second, err := runtime.RunAdvisor(t.Context(), "second?", nil, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeLimitReached, second.Type) require.Equal(t, 0, second.RemainingUses) @@ -114,7 +230,7 @@ func TestAdvisorRunError(t *testing.T) { }) require.NoError(t, err) - result, err := runtime.RunAdvisor(t.Context(), "what failed?", nil) + result, err := runtime.RunAdvisor(t.Context(), "what failed?", nil, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeError, result.Type) require.Contains(t, result.Error, "boom") @@ -149,12 +265,12 @@ func TestAdvisorRunError(t *testing.T) { }) require.NoError(t, err) - failed, err := runtime2.RunAdvisor(t.Context(), "first?", nil) + failed, err := runtime2.RunAdvisor(t.Context(), "first?", nil, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeError, failed.Type) require.Equal(t, 1, failed.RemainingUses) - retried, err := runtime2.RunAdvisor(t.Context(), "retry?", nil) + retried, err := runtime2.RunAdvisor(t.Context(), "retry?", nil, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeAdvice, retried.Type) require.Equal(t, "recovered", retried.Advice) @@ -251,7 +367,7 @@ func TestNewRuntimeDeepClonesOpenAIResponsesProviderOptions(t *testing.T) { }) require.NoError(t, err) - result, err := runtime.RunAdvisor(t.Context(), "anything?", nil) + result, err := runtime.RunAdvisor(t.Context(), "anything?", nil, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeAdvice, result.Type) @@ -316,7 +432,7 @@ func TestAdvisorRunStripsChainStateAndIsConsistentAcrossCalls(t *testing.T) { require.NoError(t, err) for i := range 2 { - result, err := runtime.RunAdvisor(t.Context(), fmt.Sprintf("q%d", i), nil) + result, err := runtime.RunAdvisor(t.Context(), fmt.Sprintf("q%d", i), nil, nil) require.NoError(t, err) require.Equal(t, chatadvisor.ResultTypeAdvice, result.Type) } diff --git a/coderd/x/chatd/chatadvisor/tool.go b/coderd/x/chatd/chatadvisor/tool.go index 8c8d25b14e..285bc836d4 100644 --- a/coderd/x/chatd/chatadvisor/tool.go +++ b/coderd/x/chatd/chatadvisor/tool.go @@ -24,6 +24,8 @@ const advisorQuestionMaxRunes = 2000 type ToolOptions struct { Runtime *Runtime GetConversationSnapshot func() []fantasy.Message + PublishAdviceDelta func(toolCallID string, delta string) + PublishAdviceReset func(toolCallID string) } // Tool returns a fantasy.AgentTool that asks a nested model for concise @@ -33,7 +35,7 @@ func Tool(opts ToolOptions) fantasy.AgentTool { return fantasy.NewAgentTool( ToolName, "Ask a separate advisor pass for strategic guidance about planning, architecture, tradeoffs, or debugging strategy. Provide a brief question. The advisor sees recent conversation context, runs without tools for a single step, and responds to the parent agent rather than the end user.", - func(ctx context.Context, args AdvisorArgs, _ fantasy.ToolCall) (fantasy.ToolResponse, error) { + func(ctx context.Context, args AdvisorArgs, call fantasy.ToolCall) (fantasy.ToolResponse, error) { if opts.Runtime == nil { return fantasy.NewTextErrorResponse("advisor runtime is not configured"), nil } @@ -51,7 +53,22 @@ func Tool(opts ToolOptions) fantasy.AgentTool { ), nil } - result, err := opts.Runtime.RunAdvisor(ctx, question, opts.GetConversationSnapshot()) + var runOpts *RunAdvisorOptions + if call.ID != "" && (opts.PublishAdviceDelta != nil || opts.PublishAdviceReset != nil) { + runOpts = &RunAdvisorOptions{} + if opts.PublishAdviceDelta != nil { + runOpts.OnAdviceDelta = func(delta string) { + opts.PublishAdviceDelta(call.ID, delta) + } + } + if opts.PublishAdviceReset != nil { + runOpts.OnAdviceReset = func() { + opts.PublishAdviceReset(call.ID) + } + } + } + + result, err := opts.Runtime.RunAdvisor(ctx, question, opts.GetConversationSnapshot(), runOpts) if err != nil { return fantasy.NewTextErrorResponse(err.Error()), nil } diff --git a/coderd/x/chatd/chatadvisor/tool_test.go b/coderd/x/chatd/chatadvisor/tool_test.go index 8208d054f8..03551cf5af 100644 --- a/coderd/x/chatd/chatadvisor/tool_test.go +++ b/coderd/x/chatd/chatadvisor/tool_test.go @@ -58,6 +58,126 @@ func TestAdvisorToolSuccess(t *testing.T) { require.Equal(t, 1, result.RemainingUses) } +func TestAdvisorToolPublishesAdviceDeltasWithToolCallID(t *testing.T) { + t.Parallel() + + type publishedDelta struct { + toolCallID string + delta string + } + var published []publishedDelta + + runtime, err := chatadvisor.NewRuntime(chatadvisor.RuntimeConfig{ + Model: &chattest.FakeModel{ + ProviderName: "test-provider", + ModelName: "test-model", + StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) { + return streamFromParts([]fantasy.StreamPart{ + {Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "Prefer "}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "the small diff."}, + {Type: fantasy.StreamPartTypeTextEnd, ID: "text-1"}, + {Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop}, + }), nil + }, + }, + MaxUsesPerRun: 2, + MaxOutputTokens: 128, + }) + require.NoError(t, err) + + tool := chatadvisor.Tool(chatadvisor.ToolOptions{ + Runtime: runtime, + GetConversationSnapshot: func() []fantasy.Message { return nil }, + PublishAdviceDelta: func(toolCallID string, delta string) { + published = append(published, publishedDelta{toolCallID: toolCallID, delta: delta}) + }, + }) + + resp := runAdvisorTool(t, tool, chatadvisor.AdvisorArgs{Question: "What's safest?"}) + require.False(t, resp.IsError) + require.Equal(t, []publishedDelta{ + {toolCallID: "call-1", delta: "Prefer "}, + {toolCallID: "call-1", delta: "the small diff."}, + }, published) + + var result chatadvisor.AdvisorResult + require.NoError(t, json.Unmarshal([]byte(resp.Content), &result)) + require.Equal(t, chatadvisor.ResultTypeAdvice, result.Type) + require.Equal(t, "Prefer the small diff.", result.Advice) +} + +func TestAdvisorToolPublishesAdviceResetWithToolCallID(t *testing.T) { + t.Parallel() + + type publishedEvent struct { + kind string + toolCallID string + delta string + } + var ( + calls int + published []publishedEvent + ) + + runtime, err := chatadvisor.NewRuntime(chatadvisor.RuntimeConfig{ + Model: &chattest.FakeModel{ + ProviderName: "test-provider", + ModelName: "test-model", + StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) { + calls++ + if calls == 1 { + return streamFromParts([]fantasy.StreamPart{ + {Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "stale "}, + {Type: fantasy.StreamPartTypeError, Error: xerrors.New("received status 429 from upstream")}, + }), nil + } + return streamFromParts([]fantasy.StreamPart{ + {Type: fantasy.StreamPartTypeTextStart, ID: "text-1"}, + {Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "fresh advice"}, + {Type: fantasy.StreamPartTypeTextEnd, ID: "text-1"}, + {Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop}, + }), nil + }, + }, + MaxUsesPerRun: 2, + MaxOutputTokens: 128, + }) + require.NoError(t, err) + + tool := chatadvisor.Tool(chatadvisor.ToolOptions{ + Runtime: runtime, + GetConversationSnapshot: func() []fantasy.Message { return nil }, + PublishAdviceDelta: func(toolCallID string, delta string) { + published = append(published, publishedEvent{ + kind: "delta", + toolCallID: toolCallID, + delta: delta, + }) + }, + PublishAdviceReset: func(toolCallID string) { + published = append(published, publishedEvent{ + kind: "reset", + toolCallID: toolCallID, + }) + }, + }) + + resp := runAdvisorTool(t, tool, chatadvisor.AdvisorArgs{Question: "What's safest?"}) + require.False(t, resp.IsError) + require.Equal(t, []publishedEvent{ + {kind: "delta", toolCallID: "call-1", delta: "stale "}, + {kind: "reset", toolCallID: "call-1"}, + {kind: "delta", toolCallID: "call-1", delta: "fresh advice"}, + }, published) + + var result chatadvisor.AdvisorResult + require.NoError(t, json.Unmarshal([]byte(resp.Content), &result)) + require.Equal(t, chatadvisor.ResultTypeAdvice, result.Type) + require.Equal(t, "fresh advice", result.Advice) +} + func TestAdvisorToolRejectsEmptyQuestion(t *testing.T) { t.Parallel() diff --git a/coderd/x/chatd/chatd.go b/coderd/x/chatd/chatd.go index e817e76f5a..5c15e33ab1 100644 --- a/coderd/x/chatd/chatd.go +++ b/coderd/x/chatd/chatd.go @@ -7250,6 +7250,28 @@ func (p *Server) runChat( // no tools. Strip it before handing the snapshot over. return stripAdvisorGuidanceBlock(slices.Clone(advisorPromptSnapshot)) }, + PublishAdviceDelta: func(toolCallID string, delta string) { + if toolCallID == "" || delta == "" { + return + } + p.publishMessagePart(chat.ID, codersdk.ChatMessageRoleTool, codersdk.ChatMessagePart{ + Type: codersdk.ChatMessagePartTypeToolResult, + ToolCallID: toolCallID, + ToolName: chatadvisor.ToolName, + ResultDelta: delta, + }) + }, + PublishAdviceReset: func(toolCallID string) { + if toolCallID == "" { + return + } + p.publishMessagePart(chat.ID, codersdk.ChatMessageRoleTool, codersdk.ChatMessagePart{ + Type: codersdk.ChatMessagePartTypeToolResult, + ToolCallID: toolCallID, + ToolName: chatadvisor.ToolName, + ResultReset: true, + }) + }, })) } diff --git a/coderd/x/chatd/chatd_test.go b/coderd/x/chatd/chatd_test.go index 413ea148d0..9d1512707a 100644 --- a/coderd/x/chatd/chatd_test.go +++ b/coderd/x/chatd/chatd_test.go @@ -9494,6 +9494,7 @@ func TestAdvisorHappyPath_RootChat(t *testing.T) { ctx := testutil.Context(t, testutil.WaitLong) const advisorReply = "break the problem into smaller pieces first" + advisorDeltas := []string{"break the problem ", "into smaller pieces first"} var ( streamedCallCount atomic.Int32 @@ -9526,7 +9527,7 @@ func TestAdvisorHappyPath_RootChat(t *testing.T) { streamedCallsMu.Unlock() advisorCallSeen.Store(true) return chattest.OpenAIStreamingResponse( - chattest.OpenAITextChunks(advisorReply)..., + chattest.OpenAITextChunks(advisorDeltas...)..., ) default: // Parent turn 2: observe the advisor tool result and close @@ -9612,6 +9613,36 @@ func TestAdvisorHappyPath_RootChat(t *testing.T) { } require.True(t, parentSawAdvisorResult, "parent must see the advisor reply in its continuation call") + + snapshot, _, cancelStream, ok := server.Subscribe(ctx, chat.ID, nil, 0) + require.True(t, ok) + cancelStream() + + var streamedAdvisorDeltas []string + for _, event := range snapshot { + if event.Type != codersdk.ChatStreamEventTypeMessagePart || event.MessagePart == nil { + continue + } + part := event.MessagePart.Part + if event.MessagePart.Role == codersdk.ChatMessageRoleTool && + part.Type == codersdk.ChatMessagePartTypeToolResult && + part.ToolName == chatadvisor.ToolName && + part.ResultDelta != "" { + streamedAdvisorDeltas = append(streamedAdvisorDeltas, part.ResultDelta) + } + } + require.Equal(t, advisorDeltas, streamedAdvisorDeltas, + "advisor nested text deltas must stream into the parent tool card") + + persisted, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ + ChatID: chat.ID, + AfterID: 0, + }) + require.NoError(t, err) + for _, msg := range persisted { + require.NotContains(t, string(msg.Content.RawMessage), "result_delta", + "advisor deltas are stream-only and must not be persisted") + } } // TestAdvisorGating_ChildChat guards the second dimension of the advisor diff --git a/codersdk/chats.go b/codersdk/chats.go index a86b72ccd6..fcbba8b5ec 100644 --- a/codersdk/chats.go +++ b/codersdk/chats.go @@ -255,7 +255,8 @@ type ChatMessagePart struct { Args json.RawMessage `json:"args,omitempty" variants:"tool-call?"` ArgsDelta string `json:"args_delta,omitempty" variants:"tool-call?"` Result json.RawMessage `json:"result,omitempty" variants:"tool-result?"` - ResultDelta string `json:"result_delta,omitempty"` + ResultDelta string `json:"result_delta,omitempty" variants:"tool-result?"` + ResultReset bool `json:"result_reset,omitempty" variants:"tool-result?"` IsError bool `json:"is_error,omitempty" variants:"tool-result?"` IsMedia bool `json:"is_media,omitempty" variants:"tool-result?"` SourceID string `json:"source_id,omitempty" variants:"source?"` @@ -327,9 +328,11 @@ type ChatMessagePart struct { // StripInternal removes internal-only fields that must not be // sent to API clients. Call before publishing via REST or SSE. // -// Note: ArgsDelta and ResultDelta are intentionally preserved. -// They are streaming-only fields consumed by the frontend via -// SSE message_part events (see processStepStream in chatloop). +// Note: ArgsDelta, ResultDelta, and ResultReset are intentionally preserved. +// They are streaming-only fields consumed by the frontend via SSE +// message_part events. ArgsDelta is produced by processStepStream in +// chatloop; ResultDelta and ResultReset are produced by the advisor +// streaming callbacks in chatd. func (p *ChatMessagePart) StripInternal() { p.ProviderMetadata = nil if p.FileID.Valid { diff --git a/codersdk/chats_test.go b/codersdk/chats_test.go index 3456c6ebb5..880094d65d 100644 --- a/codersdk/chats_test.go +++ b/codersdk/chats_test.go @@ -264,7 +264,6 @@ func TestChatMessagePartVariantTags(t *testing.T) { excludedFields := map[string]string{ "type": "discriminant, added automatically by codegen", "signature": "added in #22290, never populated by any code path", - "result_delta": "added in #22290, never populated by any code path", "provider_metadata": "internal only, stripped by db2sdk before API responses", "context_file_content": "internal only, stripped before API responses (typescript:\"-\")", "context_file_os": "internal only, used during prompt expansion (typescript:\"-\")", diff --git a/docs/reference/api/chats.md b/docs/reference/api/chats.md index 8263fb37db..a29c26f1da 100644 --- a/docs/reference/api/chats.md +++ b/docs/reference/api/chats.md @@ -125,6 +125,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -249,6 +250,7 @@ Status Code **200** | `»» provider_metadata` | array | false | | Provider metadata holds provider-specific response metadata (e.g. Anthropic cache control hints) as raw JSON. Internal only: stripped by db2sdk before API responses. | | `»» result` | array | false | | | | `»» result_delta` | string | false | | | +| `»» result_reset` | boolean | false | | | | `»» signature` | string | false | | | | `»» skill_description` | string | false | | Skill description is the short description from the skill's SKILL.md frontmatter. | | `»» skill_dir` | string | false | | Skill dir is the absolute path to the skill directory inside the workspace filesystem. Internal only: used by read_skill/read_skill_file tools to locate skill files. | @@ -455,6 +457,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -579,6 +582,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -854,6 +858,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1032,6 +1037,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1156,6 +1162,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1415,6 +1422,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1539,6 +1547,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1659,6 +1668,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1735,6 +1745,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1863,6 +1874,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -1938,6 +1950,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2063,6 +2076,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2196,6 +2210,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2268,6 +2283,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2329,6 +2345,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2577,6 +2594,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2701,6 +2719,7 @@ Experimental: this endpoint is subject to change. 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", diff --git a/docs/reference/api/schemas.md b/docs/reference/api/schemas.md index 97807cfc21..0cda9d7334 100644 --- a/docs/reference/api/schemas.md +++ b/docs/reference/api/schemas.md @@ -2199,6 +2199,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2323,6 +2324,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2659,6 +2661,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2748,6 +2751,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2791,6 +2795,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in | `provider_metadata` | array of integer | false | | Provider metadata holds provider-specific response metadata (e.g. Anthropic cache control hints) as raw JSON. Internal only: stripped by db2sdk before API responses. | | `result` | array of integer | false | | | | `result_delta` | string | false | | | +| `result_reset` | boolean | false | | | | `signature` | string | false | | | | `skill_description` | string | false | | Skill description is the short description from the skill's SKILL.md frontmatter. | | `skill_dir` | string | false | | Skill dir is the absolute path to the skill directory inside the workspace filesystem. Internal only: used by read_skill/read_skill_file tools to locate skill files. | @@ -2909,6 +2914,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -2985,6 +2991,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -3166,6 +3173,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -3311,6 +3319,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -3383,6 +3392,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -3444,6 +3454,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -3553,6 +3564,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -3736,6 +3748,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -4148,6 +4161,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -4223,6 +4237,7 @@ AuthorizationObject can represent a "set" of objects, such as: all workspaces in 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", @@ -6628,6 +6643,7 @@ CreateWorkspaceRequest provides options for creating a new workspace. Only one o 0 ], "result_delta": "string", + "result_reset": true, "signature": "string", "skill_description": "string", "skill_dir": "string", diff --git a/site/src/api/typesGenerated.ts b/site/src/api/typesGenerated.ts index ebd7da790d..31afea9094 100644 --- a/site/src/api/typesGenerated.ts +++ b/site/src/api/typesGenerated.ts @@ -2620,6 +2620,8 @@ export interface ChatToolResultPart { readonly tool_name?: string; readonly mcp_server_config_id?: string; readonly result?: Record; + readonly result_delta?: string; + readonly result_reset?: boolean; readonly is_error?: boolean; readonly is_media?: boolean; /** diff --git a/site/src/pages/AgentsPage/components/ChatConversation/streamState.test.ts b/site/src/pages/AgentsPage/components/ChatConversation/streamState.test.ts index b11984f90a..866c8df5ec 100644 --- a/site/src/pages/AgentsPage/components/ChatConversation/streamState.test.ts +++ b/site/src/pages/AgentsPage/components/ChatConversation/streamState.test.ts @@ -195,6 +195,169 @@ describe("applyMessagePartToStreamState", () => { }); }); + it("accumulates tool result deltas until a final result arrives", () => { + let state: StreamState | null = null; + state = applyMessagePartToStreamState(state, { + type: "tool-call", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + args: { question: "What is the safe path?" }, + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result_delta: "Use ", + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result_delta: "small steps.", + }); + + expect(state).not.toBeNull(); + expect(state!.toolResults["call-advisor-1"]).toMatchObject({ + id: "call-advisor-1", + name: "advisor", + result: "Use small steps.", + resultRaw: "Use small steps.", + isError: false, + isStreaming: true, + }); + expect( + buildStreamTools(state!.toolCalls, state!.toolResults)[0].status, + ).toBe("running"); + + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result: { + type: "advice", + advice: "Use small steps.", + advisor_model: "test-provider/test-model", + remaining_uses: "2", + }, + }); + + expect(state!.toolResults["call-advisor-1"]).toMatchObject({ + id: "call-advisor-1", + name: "advisor", + result: { + type: "advice", + advice: "Use small steps.", + advisor_model: "test-provider/test-model", + remaining_uses: "2", + }, + isError: false, + }); + expect(state!.toolResults["call-advisor-1"].isStreaming).toBeUndefined(); + expect( + buildStreamTools(state!.toolCalls, state!.toolResults)[0].status, + ).toBe("completed"); + }); + + it("resets streaming tool result deltas", () => { + let state: StreamState | null = null; + state = applyMessagePartToStreamState(state, { + type: "tool-call", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + args: { question: "What is the safe path?" }, + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result_delta: "stale advice", + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result_reset: true, + }); + + expect(state).not.toBeNull(); + expect(state!.toolResults["call-advisor-1"]).toBeUndefined(); + expect( + buildStreamTools(state!.toolCalls, state!.toolResults)[0].status, + ).toBe("running"); + + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result_delta: "fresh advice", + }); + expect(state!.toolResults["call-advisor-1"]).toMatchObject({ + result: "fresh advice", + resultRaw: "fresh advice", + isStreaming: true, + }); + }); + + it("replaces streamed tool result deltas with a final error", () => { + let state: StreamState | null = null; + state = applyMessagePartToStreamState(state, { + type: "tool-call", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + args: { question: "What is the safe path?" }, + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result_delta: "partial advice", + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result: { type: "error", error: "advisor failed" }, + is_error: true, + }); + + expect(state).not.toBeNull(); + expect(state!.toolResults["call-advisor-1"]).toMatchObject({ + result: { type: "error", error: "advisor failed" }, + isError: true, + }); + expect(state!.toolResults["call-advisor-1"].isStreaming).toBeUndefined(); + expect( + buildStreamTools(state!.toolCalls, state!.toolResults)[0].status, + ).toBe("error"); + }); + + it("clears streaming state for bare error tool results", () => { + let state: StreamState | null = null; + state = applyMessagePartToStreamState(state, { + type: "tool-call", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + args: { question: "What is the safe path?" }, + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + result_delta: "partial advice", + }); + state = applyMessagePartToStreamState(state, { + type: "tool-result", + tool_name: "advisor", + tool_call_id: "call-advisor-1", + is_error: true, + }); + + expect(state!.toolResults["call-advisor-1"].isStreaming).toBeUndefined(); + expect( + buildStreamTools(state!.toolCalls, state!.toolResults)[0].status, + ).toBe("error"); + }); + it("accumulates multiple tool calls in sequence", () => { let state: StreamState | null = null; state = applyMessagePartToStreamState(state, { diff --git a/site/src/pages/AgentsPage/components/ChatConversation/streamState.ts b/site/src/pages/AgentsPage/components/ChatConversation/streamState.ts index 8602c89981..9f4b6f03b7 100644 --- a/site/src/pages/AgentsPage/components/ChatConversation/streamState.ts +++ b/site/src/pages/AgentsPage/components/ChatConversation/streamState.ts @@ -106,13 +106,27 @@ export const applyMessagePartToStreamState = ( : null) || `tool-result-${Object.keys(nextState.toolResults).length + 1}-${++nextFallbackID}`; const existing = nextState.toolResults[toolCallID]; + if (part.result_reset) { + const toolResults = { ...nextState.toolResults }; + delete toolResults[toolCallID]; + return { + ...nextState, + blocks: ensureToolBlock(nextState.blocks, toolCallID), + toolResults, + }; + } + const nextResult = mergeStreamPayload( existing?.result, existing?.resultRaw, part.result, - undefined, // no delta: tool results arrive complete, not streamed incrementally + part.result_delta, ); const nextToolName = part.tool_name || existing?.name || "Tool"; + const isFinalResult = part.result !== undefined || part.is_error; + const isStreaming = isFinalResult + ? false + : existing?.isStreaming || Boolean(part.result_delta); const nextIsError = existing?.isError || parseToolResultIsError(nextToolName, part, nextResult.value); @@ -128,6 +142,7 @@ export const applyMessagePartToStreamState = ( result: nextResult.value, resultRaw: nextResult.rawText, isError: nextIsError, + isStreaming: isStreaming || undefined, mcpServerConfigId: part.mcp_server_config_id || existing?.mcpServerConfigId, }, @@ -193,6 +208,18 @@ export const applyMessagePartToStreamState = ( } }; +const getStreamToolStatus = ( + result: StreamState["toolResults"][string] | undefined, +): MergedTool["status"] => { + if (!result) { + return "running"; + } + if (result.isStreaming) { + return "running"; + } + return result.isError ? "error" : "completed"; +}; + export const buildStreamTools = ( toolCalls: StreamState["toolCalls"] | null | undefined, toolResults: StreamState["toolResults"] | null | undefined, @@ -213,7 +240,7 @@ export const buildStreamTools = ( args: call.args, result: result?.result, isError: result?.isError ?? false, - status: result ? (result.isError ? "error" : "completed") : "running", + status: getStreamToolStatus(result), mcpServerConfigId: call.mcpServerConfigId || result?.mcpServerConfigId, modelIntent: call.modelIntent, }); @@ -227,7 +254,7 @@ export const buildStreamTools = ( name: result.name, result: result.result, isError: result.isError, - status: result.isError ? "error" : "completed", + status: getStreamToolStatus(result), mcpServerConfigId: result.mcpServerConfigId, }); } diff --git a/site/src/pages/AgentsPage/components/ChatConversation/types.ts b/site/src/pages/AgentsPage/components/ChatConversation/types.ts index 424215b75a..80e89938b0 100644 --- a/site/src/pages/AgentsPage/components/ChatConversation/types.ts +++ b/site/src/pages/AgentsPage/components/ChatConversation/types.ts @@ -90,6 +90,8 @@ type StreamToolResult = { result?: unknown; resultRaw?: string; isError: boolean; + /** True while result deltas are still accumulating before the final result. */ + isStreaming?: boolean; mcpServerConfigId?: string; }; diff --git a/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.stories.tsx b/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.stories.tsx index 197e0877f2..396c287d5b 100644 --- a/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.stories.tsx +++ b/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.stories.tsx @@ -121,6 +121,30 @@ export const Running: Story = { }, }; +export const RunningWithStreamedAdvice: Story = { + args: { + status: "running", + args: { question: sampleQuestion }, + result: "Use the smaller diff while the advisor is still responding.", + }, + play: async ({ canvasElement }) => { + const canvas = within(canvasElement); + expect(canvas.getByText(sampleQuestion)).toBeInTheDocument(); + expect(canvas.getByText("Consulting advisor…")).toBeInTheDocument(); + expect( + await canvas.findByText( + "Use the smaller diff while the advisor is still responding.", + ), + ).toBeInTheDocument(); + expect( + canvas.queryByText("Advisor returned no guidance."), + ).not.toBeInTheDocument(); + expect( + canvas.queryByText("Reviewing context and preparing guidance."), + ).not.toBeInTheDocument(); + }, +}; + export const LimitReached: Story = { args: { status: "completed", diff --git a/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.tsx b/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.tsx index bf18de0fef..4ad6e0c30a 100644 --- a/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.tsx +++ b/site/src/pages/AgentsPage/components/ChatElements/tools/AdvisorTool.tsx @@ -110,7 +110,7 @@ export const AdvisorTool: React.FC = ({ data-testid="advisor-tool-scroll-area" >
- {isRunning ? ( + {isRunning && adviceText.length === 0 ? (
Reviewing context and preparing guidance.
@@ -147,7 +147,10 @@ export const AdvisorTool: React.FC = ({ Advice
- + {adviceText || EMPTY_ADVICE_MESSAGE}