mirror of
https://github.com/coder/coder.git
synced 2026-10-16 14:48:56 +08:00
fix(coderd/x/chatd): stream compaction summaries (#28886)
Chats could get permanently stuck once their context crossed the compaction threshold on adaptive-thinking Claude models: every compaction attempt failed with `compaction produced an empty summary`, and both automatic and manual compaction take the same path, so the chat could never recover. ## Problem `compactionSummaryCall` deliberately set `MaxOutputTokens = nil` because the summary request was non-streaming and the Anthropic SDK rejects non-streaming requests whose `max_tokens` implies a completion longer than ten minutes. But a nil cap makes fantasy's Anthropic provider silently default to 4096 output tokens, and adaptive-thinking Claude models can spend that entire budget on a thinking block, leaving zero summary text. ## Fix - Stream the compaction summary request. Streaming lifts the Anthropic SDK's non-streaming duration restriction, which was the only reason the cap was nil'd. - Keep the resolved output cap (32k default) on the summary call instead of nil'ing it. - When the summary comes back empty with truncation evidence (length finish reason, or reasoning with no text), fail with `compaction summary was truncated at the output token cap` instead of the generic empty-summary error, so the next occurrence is attributable at a glance. Unit tests cover text-block joining and the truncation classification; the compaction e2e fixtures now serve streamed summaries and assert the summary request streams with `max_tokens: 32000`. Dogfood UAT on a deployment built from this branch verified manual compaction with Claude Opus 5 (adaptive thinking, twice, with reload persistence) and an OpenAI regression chat. > 🤖 Opened by [Xum](https://mux.coder.com) on behalf of @ibetitsmike.
This commit is contained in:
@@ -5236,22 +5236,25 @@ func highUsageTextResponse(text string) chattest.AnthropicResponse {
|
||||
}, text)...)
|
||||
}
|
||||
|
||||
func anthropicCompactionResponse(text string) chattest.AnthropicResponse {
|
||||
return chattest.AnthropicResponse{Response: &chattest.AnthropicMessage{
|
||||
ID: "msg-compaction",
|
||||
Type: "message",
|
||||
Role: "assistant",
|
||||
Content: text,
|
||||
Model: "claude-3-opus-20240229",
|
||||
StopReason: "end_turn",
|
||||
}}
|
||||
func anthropicCompactionResponse(t testing.TB, req *chattest.AnthropicRequest, text string) chattest.AnthropicResponse {
|
||||
t.Helper()
|
||||
require.True(t, req.Stream)
|
||||
// The summary cap is the configured cap with headroom bounded by
|
||||
// the remaining context window, so it varies per test fixture; exact
|
||||
// values are pinned in the chatloop unit tests and the hook test.
|
||||
require.Positive(t, req.MaxTokens)
|
||||
return chattest.AnthropicStreamingResponse(chattest.AnthropicTextChunks(text)...)
|
||||
}
|
||||
|
||||
func highUsageReadFileResponse(path string) chattest.AnthropicResponse {
|
||||
return readFileResponseWithInputTokens(path, 80)
|
||||
}
|
||||
|
||||
func readFileResponseWithInputTokens(path string, inputTokens int) chattest.AnthropicResponse {
|
||||
chunks := chattest.AnthropicToolCallChunks("read_file", fmt.Sprintf(`{"path":%q}`, path))
|
||||
for i := range chunks {
|
||||
if chunks[i].Type == "message_start" {
|
||||
chunks[i].Message.Usage = map[string]int{"input_tokens": 80}
|
||||
chunks[i].Message.Usage = map[string]int{"input_tokens": inputTokens}
|
||||
}
|
||||
if chunks[i].Type == "message_delta" {
|
||||
chunks[i].UsageMap = map[string]int{"output_tokens": 5}
|
||||
@@ -5275,10 +5278,10 @@ func TestActiveServer_RoutingPreservesAPIKeyAfterCompaction(t *testing.T) {
|
||||
var streamCount atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return chattest.AnthropicNonStreamingResponse(compactionSummary)
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("AI Gateway Compaction")
|
||||
}
|
||||
|
||||
@@ -5398,10 +5401,10 @@ func TestActiveServer_CompactionRecordsMetric(t *testing.T) {
|
||||
var streamCount atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
switch streamCount.Add(1) {
|
||||
@@ -5490,12 +5493,12 @@ func TestActiveServer_Compaction(t *testing.T) {
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
requests.record(req)
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
require.Contains(t, body, "read_file")
|
||||
require.Contains(t, body, "package main")
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
require.Contains(t, body, "read_file")
|
||||
require.Contains(t, body, "package main")
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
switch streamCount.Add(1) {
|
||||
@@ -5586,7 +5589,7 @@ func TestActiveServer_Compaction(t *testing.T) {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionRequests.Add(1)
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
@@ -5625,10 +5628,10 @@ func TestActiveServer_Compaction(t *testing.T) {
|
||||
var streamCount atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
switch streamCount.Add(1) {
|
||||
@@ -5727,13 +5730,13 @@ func TestActiveServer_ManualCompaction(t *testing.T) {
|
||||
var compactionRequests atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionRequests.Add(1)
|
||||
require.Contains(t, body, "hello from the user")
|
||||
require.True(t, anthropicMessageHasEphemeralCacheControl(t, req.Messages[len(req.Messages)-1]))
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionRequests.Add(1)
|
||||
require.Contains(t, body, "hello from the user")
|
||||
require.True(t, anthropicMessageHasEphemeralCacheControl(t, req.Messages[len(req.Messages)-1]))
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
streamCount.Add(1)
|
||||
@@ -5819,11 +5822,11 @@ func TestActiveServer_ManualCompaction(t *testing.T) {
|
||||
var compactionRequests atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionRequests.Add(1)
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionRequests.Add(1)
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
return chattest.AnthropicStreamingResponse(chattest.AnthropicTextChunks("assistant answer")...)
|
||||
@@ -5879,22 +5882,22 @@ func TestActiveServer_ManualCompaction(t *testing.T) {
|
||||
var compactionToolCounts []int
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
mu.Lock()
|
||||
compactionToolCounts = append(compactionToolCounts, len(req.Tools))
|
||||
mu.Unlock()
|
||||
// The tool definitions push the summary request past
|
||||
// the window that the tool-less request still fits.
|
||||
if len(req.Tools) > 0 {
|
||||
return chattest.AnthropicResponse{Error: &chattest.ErrorResponse{
|
||||
StatusCode: http.StatusBadRequest,
|
||||
Type: "invalid_request_error",
|
||||
Message: "prompt is too long: 20000 tokens > 16385 maximum",
|
||||
}}
|
||||
}
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
mu.Lock()
|
||||
compactionToolCounts = append(compactionToolCounts, len(req.Tools))
|
||||
mu.Unlock()
|
||||
// The tool definitions push the summary request past
|
||||
// the window that the tool-less request still fits.
|
||||
if len(req.Tools) > 0 {
|
||||
return chattest.AnthropicResponse{Error: &chattest.ErrorResponse{
|
||||
StatusCode: http.StatusBadRequest,
|
||||
Type: "invalid_request_error",
|
||||
Message: "prompt is too long: 20000 tokens > 16385 maximum",
|
||||
}}
|
||||
}
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
return chattest.AnthropicStreamingResponse(chattest.AnthropicTextChunks("assistant answer")...)
|
||||
@@ -6119,11 +6122,11 @@ func TestActiveServer_ManualClear(t *testing.T) {
|
||||
var compactionRequests atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionRequests.Add(1)
|
||||
return anthropicCompactionResponse(t, req, "unexpected summary")
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionRequests.Add(1)
|
||||
return anthropicCompactionResponse("unexpected summary")
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
// The first turn exceeds the threshold, so stale usage would
|
||||
@@ -6169,10 +6172,10 @@ func TestActiveServer_ManualClear(t *testing.T) {
|
||||
db, ps := dbtestutil.NewDB(t)
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse(t, req, "compaction summary")
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse("compaction summary")
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
return chattest.AnthropicStreamingResponse(chattest.AnthropicTextChunks("assistant answer")...)
|
||||
@@ -6394,14 +6397,17 @@ func TestActiveServer_CompactionModelOverride(t *testing.T) {
|
||||
},
|
||||
},
|
||||
{
|
||||
// 3276 is 0.8 (high) of the summary call's default 4096 max_tokens.
|
||||
// High effort maps to 0.8 of the summary call's max_tokens
|
||||
// (the remaining-window clamp; its exact value is pinned in
|
||||
// the chatloop unit tests).
|
||||
name: "legacy budget-thinking override model",
|
||||
overrideModel: "claude-haiku-4-5",
|
||||
effort: "high",
|
||||
assertSummaryRequest: func(t *testing.T, req *chattest.AnthropicRequest) {
|
||||
require.Empty(t, string(req.OutputConfig))
|
||||
require.Contains(t, string(req.Thinking), `"type":"enabled"`)
|
||||
require.Contains(t, string(req.Thinking), `"budget_tokens":3276`)
|
||||
wantBudget := int64(float64(req.MaxTokens) * 0.8)
|
||||
require.Contains(t, string(req.Thinking), fmt.Sprintf(`"budget_tokens":%d`, wantBudget))
|
||||
},
|
||||
},
|
||||
{
|
||||
@@ -6434,23 +6440,27 @@ func TestActiveServer_CompactionModelOverride(t *testing.T) {
|
||||
var streamCount atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
require.Equal(t, tc.overrideModel, req.Model)
|
||||
if tc.keepsTools {
|
||||
require.NotEmpty(t, req.Tools)
|
||||
} else {
|
||||
require.Empty(t, req.Tools, "a different model shares no prompt cache, so the summary carries no tool definitions")
|
||||
}
|
||||
tc.assertSummaryRequest(t, req)
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
require.Equal(t, tc.overrideModel, req.Model)
|
||||
if tc.keepsTools {
|
||||
require.NotEmpty(t, req.Tools)
|
||||
} else {
|
||||
require.Empty(t, req.Tools, "a different model shares no prompt cache, so the summary carries no tool definitions")
|
||||
}
|
||||
tc.assertSummaryRequest(t, req)
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
require.Equal(t, chatModelName, req.Model)
|
||||
switch streamCount.Add(1) {
|
||||
case 1:
|
||||
return highUsageReadFileResponse("/tmp/a.txt")
|
||||
// A large window keeps the summary cap clamp
|
||||
// (limit minus usage and reserves) above the
|
||||
// minimum legacy thinking budget so the effort
|
||||
// mapping stays observable on the summary request.
|
||||
return readFileResponseWithInputTokens("/tmp/a.txt", 80_000)
|
||||
default:
|
||||
require.Contains(t, body, compactionSummary)
|
||||
require.Empty(t, string(req.OutputConfig),
|
||||
@@ -6464,7 +6474,7 @@ func TestActiveServer_CompactionModelOverride(t *testing.T) {
|
||||
}
|
||||
})
|
||||
user, org, model := seedAnthropicChatDependencies(t, db, anthropicURL)
|
||||
model = updateChatModelCompressionThreshold(t, db, model, 100, thresholdPercent)
|
||||
model = updateChatModelCompressionThreshold(t, db, model, 100_000, thresholdPercent)
|
||||
overrideModel := seedOverrideModel(ctx, t, db, model, tc.overrideModel, tc.effort, 1_000_000)
|
||||
ws, dbAgent := seedWorkspaceWithAgent(t, db, user.ID)
|
||||
|
||||
@@ -6548,11 +6558,11 @@ func TestActiveServer_CompactionModelOverride(t *testing.T) {
|
||||
var streamCount atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
require.Equal(t, overrideModelName, req.Model)
|
||||
return anthropicCompactionResponse(t, req, compactionSummary)
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
require.Equal(t, overrideModelName, req.Model)
|
||||
return anthropicCompactionResponse(compactionSummary)
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
switch streamCount.Add(1) {
|
||||
@@ -11775,10 +11785,10 @@ func TestActiveServer_ChatTurnDebugRunRecordsMCPConnectOnDecisionError(t *testin
|
||||
// terminally with the still-over-limit error.
|
||||
var streamCount atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
if strings.Contains(anthropicRequestBody(t, *req), "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse(t, req, "summary text for compaction")
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(anthropicRequestBody(t, *req), "You are performing a context compaction") {
|
||||
return anthropicCompactionResponse("summary text for compaction")
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
if streamCount.Add(1) == 1 {
|
||||
|
||||
@@ -232,6 +232,10 @@ type GenerateCompactionOptions struct {
|
||||
// Clock measures the summary call duration. Required.
|
||||
Clock quartz.Clock
|
||||
|
||||
// StreamSilenceTimeout bounds how long the summary stream may go
|
||||
// without yielding a part. Defaults to defaultStreamSilenceTimeout.
|
||||
StreamSilenceTimeout time.Duration
|
||||
|
||||
// OnModelStreamStart runs immediately before the summary model call,
|
||||
// at the instant CompactionResult.Runtime starts measuring.
|
||||
OnModelStreamStart func()
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
|
||||
"github.com/coder/coder/v2/coderd/x/chatd/chatdebug"
|
||||
"github.com/coder/coder/v2/codersdk"
|
||||
"github.com/coder/quartz"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -102,6 +103,12 @@ type CompactionOptions struct {
|
||||
// summary call uses the exact same ordered definitions.
|
||||
ToolDefinitions []fantasy.Tool
|
||||
|
||||
// Clock and StreamSilenceTimeout guard the summary stream against a
|
||||
// provider that opens the stream but stops yielding parts. Zero values
|
||||
// fall back to a real clock and DefaultStreamSilenceTimeout.
|
||||
Clock quartz.Clock
|
||||
StreamSilenceTimeout time.Duration
|
||||
|
||||
// Force skips the threshold gate (including the threshold=100
|
||||
// disable and the zero-usage early return). Set for manual,
|
||||
// user-requested compactions.
|
||||
@@ -171,6 +178,24 @@ func GenerateCompaction(ctx context.Context, opts GenerateCompactionOptions) (Co
|
||||
return CompactionResult{}, nil
|
||||
}
|
||||
|
||||
// Sum-enforcing providers reject requests whose input plus
|
||||
// max_tokens exceeds the context window, so bound the summary cap
|
||||
// by the remaining window. contextTokens covers only the trigger
|
||||
// step's prompt, so also reserve that step's output, the tool
|
||||
// results executed after it, and the summary prompt appended by
|
||||
// generateCompactionSummary, all of which become input to the
|
||||
// summary request. Degenerate cases (unknown limit, no room left)
|
||||
// leave the cap unchanged.
|
||||
if config.SummaryCall.MaxOutputTokens != nil {
|
||||
promptBytes := len(config.SummaryPrompt) + len(config.SummaryHint)
|
||||
promptTokens := int64((promptBytes + bytesPerTokenEstimate - 1) / bytesPerTokenEstimate)
|
||||
reserved := opts.StepUsage.OutputTokens + promptTokens + trailingToolResultTokens(opts.Messages)
|
||||
remaining := contextLimit - contextTokens - reserved
|
||||
if remaining > 0 && remaining < *config.SummaryCall.MaxOutputTokens {
|
||||
config.SummaryCall.MaxOutputTokens = &remaining
|
||||
}
|
||||
}
|
||||
|
||||
if config.PublishMessagePart != nil && config.ToolCallID != "" {
|
||||
config.PublishMessagePart(
|
||||
codersdk.ChatMessageRoleAssistant,
|
||||
@@ -226,24 +251,26 @@ func GenerateCompaction(ctx context.Context, opts GenerateCompactionOptions) (Co
|
||||
|
||||
func normalizedCompactionGenerateConfig(opts GenerateCompactionOptions) (CompactionOptions, bool) {
|
||||
config := CompactionOptions{
|
||||
ThresholdPercent: opts.ThresholdPercent,
|
||||
ContextLimit: opts.ContextLimit,
|
||||
SummaryPrompt: opts.SummaryPrompt,
|
||||
SummaryHint: opts.SummaryHint,
|
||||
SystemSummaryPrefix: opts.SystemSummaryPrefix,
|
||||
DebugSvc: opts.DebugSvc,
|
||||
ChatID: opts.ChatID,
|
||||
HistoryTipMessageID: opts.HistoryTipMessageID,
|
||||
ResolvedProvider: opts.ResolvedProvider,
|
||||
ResolvedModel: opts.ResolvedModel,
|
||||
ModelConfigID: opts.ModelConfigID,
|
||||
SummaryCall: opts.SummaryCall,
|
||||
ToolDefinitions: opts.ToolDefinitions,
|
||||
Force: opts.Force,
|
||||
Source: opts.Source,
|
||||
ToolCallID: opts.ToolCallID,
|
||||
ToolName: opts.ToolName,
|
||||
PublishMessagePart: opts.PublishMessagePart,
|
||||
ThresholdPercent: opts.ThresholdPercent,
|
||||
ContextLimit: opts.ContextLimit,
|
||||
SummaryPrompt: opts.SummaryPrompt,
|
||||
SummaryHint: opts.SummaryHint,
|
||||
SystemSummaryPrefix: opts.SystemSummaryPrefix,
|
||||
DebugSvc: opts.DebugSvc,
|
||||
ChatID: opts.ChatID,
|
||||
HistoryTipMessageID: opts.HistoryTipMessageID,
|
||||
ResolvedProvider: opts.ResolvedProvider,
|
||||
ResolvedModel: opts.ResolvedModel,
|
||||
ModelConfigID: opts.ModelConfigID,
|
||||
SummaryCall: opts.SummaryCall,
|
||||
ToolDefinitions: opts.ToolDefinitions,
|
||||
Force: opts.Force,
|
||||
Source: opts.Source,
|
||||
ToolCallID: opts.ToolCallID,
|
||||
ToolName: opts.ToolName,
|
||||
PublishMessagePart: opts.PublishMessagePart,
|
||||
Clock: opts.Clock,
|
||||
StreamSilenceTimeout: opts.StreamSilenceTimeout,
|
||||
}
|
||||
if strings.TrimSpace(config.SummaryPrompt) == "" {
|
||||
config.SummaryPrompt = defaultCompactionSummaryPrompt
|
||||
@@ -281,6 +308,37 @@ func publishCompactionError(config CompactionOptions, msg string) {
|
||||
)
|
||||
}
|
||||
|
||||
// trailingToolResultTokens estimates tokens for the tool-result
|
||||
// messages that follow the usage-measured assistant step. Local tools
|
||||
// execute before compaction, so their results are input to the summary
|
||||
// request but absent from StepUsage.
|
||||
func trailingToolResultTokens(messages []fantasy.Message) int64 {
|
||||
totalBytes := 0
|
||||
for i := len(messages) - 1; i >= 0 && messages[i].Role == fantasy.MessageRoleTool; i-- {
|
||||
for _, part := range messages[i].Content {
|
||||
result, ok := part.(fantasy.ToolResultPart)
|
||||
if !ok {
|
||||
resultPtr, okPtr := part.(*fantasy.ToolResultPart)
|
||||
if !okPtr || resultPtr == nil {
|
||||
continue
|
||||
}
|
||||
result = *resultPtr
|
||||
}
|
||||
switch output := result.Output.(type) {
|
||||
case fantasy.ToolResultOutputContentText:
|
||||
totalBytes += len(output.Text)
|
||||
case fantasy.ToolResultOutputContentError:
|
||||
if output.Error != nil {
|
||||
totalBytes += len(output.Error.Error())
|
||||
}
|
||||
case fantasy.ToolResultOutputContentMedia:
|
||||
totalBytes += len(output.Data) + len(output.Text)
|
||||
}
|
||||
}
|
||||
}
|
||||
return int64((totalBytes + bytesPerTokenEstimate - 1) / bytesPerTokenEstimate)
|
||||
}
|
||||
|
||||
// contextTokensFromUsage returns the total context token count from
|
||||
// a step's usage report. It sums input, cache-read, and
|
||||
// cache-creation tokens when available, falling back to TotalTokens
|
||||
@@ -442,7 +500,7 @@ func generateCompactionSummary(
|
||||
|
||||
summaryCtx, finishDebugRun := startCompactionDebugRun(ctx, options)
|
||||
defer func() {
|
||||
// If model.Generate (or anything else below) panics, the
|
||||
// If model.Stream (or anything else below) panics, the
|
||||
// named err return is still nil at this point. Without the
|
||||
// recover hook we would finalize the debug run as Completed
|
||||
// in the exact crash path operators rely on to diagnose
|
||||
@@ -459,33 +517,116 @@ func generateCompactionSummary(
|
||||
call := options.SummaryCall
|
||||
call.Prompt = summaryPrompt
|
||||
call.Tools = options.ToolDefinitions
|
||||
response, err := model.Generate(summaryCtx, call)
|
||||
clock := options.Clock
|
||||
if clock == nil {
|
||||
clock = quartz.NewReal()
|
||||
}
|
||||
timeout := options.StreamSilenceTimeout
|
||||
if timeout == 0 {
|
||||
timeout = DefaultStreamSilenceTimeout
|
||||
}
|
||||
// NopMetrics: TTFT is an assistant-generation metric, so the summary
|
||||
// stream must not record into it.
|
||||
streamSummaryText := func() (string, error) {
|
||||
attempt, err := guardedStream(
|
||||
summaryCtx,
|
||||
options.ResolvedProvider,
|
||||
options.ResolvedModel,
|
||||
clock,
|
||||
timeout,
|
||||
func(attemptCtx context.Context) (fantasy.StreamResponse, error) {
|
||||
return model.Stream(attemptCtx, call)
|
||||
},
|
||||
NopMetrics(),
|
||||
)
|
||||
if err != nil {
|
||||
return "", xerrors.Errorf("stream summary text: %w", wrapProviderStreamError(options.ResolvedProvider, err))
|
||||
}
|
||||
defer attempt.release()
|
||||
|
||||
textPartIndexes := make(map[string]int)
|
||||
textParts := make([]string, 0, 1)
|
||||
var (
|
||||
reasoningSeen bool
|
||||
finishSeen bool
|
||||
finishReason fantasy.FinishReason
|
||||
streamErr error
|
||||
)
|
||||
for part := range attempt.stream {
|
||||
switch part.Type {
|
||||
case fantasy.StreamPartTypeTextStart:
|
||||
if _, ok := textPartIndexes[part.ID]; ok {
|
||||
continue
|
||||
}
|
||||
textPartIndexes[part.ID] = len(textParts)
|
||||
textParts = append(textParts, part.Delta)
|
||||
case fantasy.StreamPartTypeTextDelta:
|
||||
index, ok := textPartIndexes[part.ID]
|
||||
if !ok {
|
||||
index = len(textParts)
|
||||
textPartIndexes[part.ID] = index
|
||||
textParts = append(textParts, "")
|
||||
}
|
||||
textParts[index] += part.Delta
|
||||
case fantasy.StreamPartTypeReasoningStart,
|
||||
fantasy.StreamPartTypeReasoningDelta,
|
||||
fantasy.StreamPartTypeReasoningEnd:
|
||||
reasoningSeen = true
|
||||
case fantasy.StreamPartTypeFinish:
|
||||
finishSeen = true
|
||||
finishReason = part.FinishReason
|
||||
case fantasy.StreamPartTypeError:
|
||||
streamErr = part.Error
|
||||
if streamErr == nil {
|
||||
streamErr = xerrors.New("model returned an error part")
|
||||
}
|
||||
}
|
||||
if streamErr != nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
if err := attempt.finish(streamErr); err != nil {
|
||||
// Providers can surface remote stream resets as bare
|
||||
// context.Canceled; wrap them like GenerateAssistant so the
|
||||
// generation loop retries instead of terminally erroring.
|
||||
return "", xerrors.Errorf("stream summary text: %w", wrapProviderStreamError(options.ResolvedProvider, err))
|
||||
}
|
||||
if !finishSeen {
|
||||
// A stream that ends without a finish part was interrupted.
|
||||
// Committing its partial text would compact history against an
|
||||
// incomplete summary.
|
||||
if ctxErr := summaryCtx.Err(); ctxErr != nil {
|
||||
return "", xerrors.Errorf("stream summary text: %w", ctxErr)
|
||||
}
|
||||
return "", xerrors.New("compaction summary stream ended without a finish part")
|
||||
}
|
||||
|
||||
parts := make([]string, 0, len(textParts))
|
||||
for _, text := range textParts {
|
||||
text = strings.TrimSpace(text)
|
||||
if text != "" {
|
||||
parts = append(parts, text)
|
||||
}
|
||||
}
|
||||
joined := strings.TrimSpace(strings.Join(parts, " "))
|
||||
if joined == "" && (finishReason == fantasy.FinishReasonLength || reasoningSeen) {
|
||||
return "", xerrors.New("compaction summary was truncated at the output token cap")
|
||||
}
|
||||
return joined, nil
|
||||
}
|
||||
summary, err = streamSummaryText()
|
||||
if err != nil && len(call.Tools) > 0 && isContextTooLargeError(err) {
|
||||
// Tool definitions keep the summary request on the parent turn's
|
||||
// cacheable prefix, but they also make it larger than the turn
|
||||
// that just overflowed. Compaction is the recovery path for an
|
||||
// over-limit conversation, so fall back to the tool-less request,
|
||||
// which still fits whenever the history alone does.
|
||||
// which still fits whenever the history alone does. The rejection
|
||||
// can arrive from the stream open or as an in-stream error part,
|
||||
// so the retry wraps the whole attempt.
|
||||
call.Tools = nil
|
||||
response, err = model.Generate(summaryCtx, call)
|
||||
summary, err = streamSummaryText()
|
||||
}
|
||||
if err != nil {
|
||||
return "", xerrors.Errorf("generate summary text: %w", err)
|
||||
}
|
||||
|
||||
parts := make([]string, 0, len(response.Content))
|
||||
for _, block := range response.Content {
|
||||
textBlock, ok := fantasy.AsContentType[fantasy.TextContent](block)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
text := strings.TrimSpace(textBlock.Text)
|
||||
if text == "" {
|
||||
continue
|
||||
}
|
||||
parts = append(parts, text)
|
||||
}
|
||||
return strings.TrimSpace(strings.Join(parts, " ")), nil
|
||||
return summary, err
|
||||
}
|
||||
|
||||
// contextTooLargePhrases are context-window rejections fantasy does not
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -19,12 +20,32 @@ import (
|
||||
"github.com/coder/coder/v2/coderd/database"
|
||||
"github.com/coder/coder/v2/coderd/database/dbmock"
|
||||
"github.com/coder/coder/v2/coderd/x/chatd/chatdebug"
|
||||
"github.com/coder/coder/v2/coderd/x/chatd/chaterror"
|
||||
"github.com/coder/coder/v2/coderd/x/chatd/chattest"
|
||||
"github.com/coder/coder/v2/codersdk"
|
||||
"github.com/coder/coder/v2/testutil"
|
||||
"github.com/coder/quartz"
|
||||
)
|
||||
|
||||
func compactionStream(parts ...fantasy.StreamPart) fantasy.StreamResponse {
|
||||
return func(yield func(fantasy.StreamPart) bool) {
|
||||
for _, part := range parts {
|
||||
if !yield(part) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func summaryTextStream() fantasy.StreamResponse {
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "summary"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
)
|
||||
}
|
||||
|
||||
func TestStartCompactionDebugRun_DoesNotReportDebugErrors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -184,7 +205,7 @@ func TestGenerateCompactionSummary_PanicFinalizesAsError(t *testing.T) {
|
||||
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
GenerateFn: func(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
panic("compaction model crash")
|
||||
},
|
||||
}
|
||||
@@ -245,9 +266,9 @@ func TestGenerateCompactionSummaryPreservesCallOptions(t *testing.T) {
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
GenerateFn: func(_ context.Context, call fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
got = call
|
||||
return &fantasy.Response{Content: []fantasy.Content{fantasy.TextContent{Text: "summary"}}}, nil
|
||||
return summaryTextStream(), nil
|
||||
},
|
||||
}
|
||||
|
||||
@@ -301,9 +322,9 @@ func TestGenerateCompactionSummaryUsesToolDefinitions(t *testing.T) {
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
GenerateFn: func(_ context.Context, call fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
got = call
|
||||
return &fantasy.Response{Content: []fantasy.Content{fantasy.TextContent{Text: "summary"}}}, nil
|
||||
return summaryTextStream(), nil
|
||||
},
|
||||
}
|
||||
|
||||
@@ -335,9 +356,9 @@ func TestGenerateCompactionSummaryAppliesAnthropicPromptCaching(t *testing.T) {
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: fantasyanthropic.Name,
|
||||
ModelName: "claude",
|
||||
GenerateFn: func(_ context.Context, call fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
got = call
|
||||
return &fantasy.Response{Content: []fantasy.Content{fantasy.TextContent{Text: "summary"}}}, nil
|
||||
return summaryTextStream(), nil
|
||||
},
|
||||
}
|
||||
|
||||
@@ -370,7 +391,7 @@ func TestGenerateCompactionSummaryRetriesWithoutToolsWhenContextTooLarge(t *test
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
GenerateFn: func(_ context.Context, call fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
calls = append(calls, call)
|
||||
if len(call.Tools) > 0 {
|
||||
return nil, &fantasy.ProviderError{
|
||||
@@ -379,7 +400,7 @@ func TestGenerateCompactionSummaryRetriesWithoutToolsWhenContextTooLarge(t *test
|
||||
StatusCode: http.StatusBadRequest,
|
||||
}
|
||||
}
|
||||
return &fantasy.Response{Content: []fantasy.Content{fantasy.TextContent{Text: "summary"}}}, nil
|
||||
return summaryTextStream(), nil
|
||||
},
|
||||
}
|
||||
|
||||
@@ -446,7 +467,7 @@ func TestGenerateCompactionSummaryDoesNotRetryOtherFailures(t *testing.T) {
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
GenerateFn: func(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
calls++
|
||||
return nil, tc.err
|
||||
},
|
||||
@@ -564,16 +585,21 @@ func TestGenerateCompactionSummary_UsesCallerContext(t *testing.T) {
|
||||
|
||||
type contextKey string
|
||||
testCtx := context.WithValue(context.Background(), contextKey("key"), "value")
|
||||
var ctxSeen context.Context
|
||||
var (
|
||||
ctxSeen context.Context
|
||||
errAtCall error
|
||||
)
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
GenerateFn: func(ctx context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(ctx context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
ctxSeen = ctx
|
||||
return &fantasy.Response{
|
||||
Content: []fantasy.Content{
|
||||
fantasy.TextContent{Text: "summary"},
|
||||
},
|
||||
}, nil
|
||||
errAtCall = ctx.Err()
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "summary"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
|
||||
@@ -583,13 +609,161 @@ func TestGenerateCompactionSummary_UsesCallerContext(t *testing.T) {
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "summary", summary)
|
||||
require.Same(t, testCtx, ctxSeen)
|
||||
require.NoError(t, ctxSeen.Err())
|
||||
// The silence guard wraps the caller context in a cancelable child that
|
||||
// is released after the summary completes, so assert inheritance rather
|
||||
// than identity: values propagate, the context is live at call time, and
|
||||
// no deadline is attached (the guard uses a timer, not a context
|
||||
// deadline).
|
||||
require.NoError(t, errAtCall)
|
||||
_, ok := ctxSeen.Deadline()
|
||||
require.False(t, ok)
|
||||
require.Equal(t, "value", ctxSeen.Value(contextKey("key")))
|
||||
}
|
||||
|
||||
func TestGenerateCompactionSummary_Stream(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
t.Run("joins text blocks", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "first"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "first", Delta: " first "},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "first"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "second"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "second", Delta: " second "},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "second"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "first second", summary)
|
||||
})
|
||||
|
||||
t.Run("reports output cap truncation", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeReasoningStart, ID: "reasoning"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeReasoningDelta, ID: "reasoning", Delta: "thinking"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeReasoningEnd, ID: "reasoning"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonLength},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
require.Empty(t, summary)
|
||||
require.EqualError(t, err, "compaction summary was truncated at the output token cap")
|
||||
})
|
||||
|
||||
t.Run("rejects stream without finish part", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "partial summary"},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
require.Empty(t, summary)
|
||||
require.EqualError(t, err, "compaction summary stream ended without a finish part")
|
||||
})
|
||||
|
||||
t.Run("classifies canceled stream open as transport reset", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return nil, context.Canceled
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
ResolvedProvider: "fake",
|
||||
})
|
||||
require.Empty(t, summary)
|
||||
require.ErrorIs(t, err, chaterror.ErrProviderTransportReset)
|
||||
classified := chaterror.Classify(err)
|
||||
require.True(t, classified.Retryable)
|
||||
require.Equal(t, "fake", classified.Provider)
|
||||
})
|
||||
|
||||
t.Run("classifies canceled error part as transport reset", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeError, Error: context.Canceled},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
ResolvedProvider: "fake",
|
||||
})
|
||||
require.Empty(t, summary)
|
||||
require.ErrorIs(t, err, chaterror.ErrProviderTransportReset)
|
||||
classified := chaterror.Classify(err)
|
||||
require.True(t, classified.Retryable)
|
||||
require.Equal(t, "fake", classified.Provider)
|
||||
})
|
||||
|
||||
t.Run("classifies stream silence", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := testutil.Context(t, testutil.WaitLong)
|
||||
clock := quartz.NewMock(t)
|
||||
trap := clock.Trap().AfterFunc(streamSilenceGuardTimerTag)
|
||||
defer trap.Close()
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "openai",
|
||||
StreamFn: func(ctx context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return func(yield func(fantasy.StreamPart) bool) {
|
||||
<-ctx.Done()
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
ResolvedProvider: "openai",
|
||||
Clock: clock,
|
||||
StreamSilenceTimeout: 5 * time.Millisecond,
|
||||
})
|
||||
done <- err
|
||||
}()
|
||||
|
||||
trap.MustWait(ctx).MustRelease(ctx)
|
||||
_, waiter := clock.AdvanceNext()
|
||||
waiter.MustWait(ctx)
|
||||
err := <-done
|
||||
require.Error(t, err)
|
||||
classified := chaterror.Classify(err)
|
||||
require.Equal(t, codersdk.ChatErrorKindStreamSilenceTimeout, classified.Kind)
|
||||
require.Equal(t, "openai", classified.Provider)
|
||||
require.True(t, classified.Retryable)
|
||||
})
|
||||
}
|
||||
|
||||
// TestGenerateCompaction_ForceBypassesThresholdGates verifies the
|
||||
// manual-compaction contract: Force runs the summary even when usage
|
||||
// is below threshold, when usage is zero, and when threshold=100
|
||||
@@ -602,13 +776,14 @@ func TestGenerateCompaction_ForceBypassesThresholdGates(t *testing.T) {
|
||||
return &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
GenerateFn: func(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
*calls++
|
||||
return &fantasy.Response{
|
||||
Content: []fantasy.Content{
|
||||
fantasy.TextContent{Text: "forced summary"},
|
||||
},
|
||||
}, nil
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "forced summary"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -670,6 +845,77 @@ func TestGenerateCompaction_ForceBypassesThresholdGates(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestGenerateCompaction_ClampsSummaryCapToRemainingWindow pins the
|
||||
// summary output cap bound: sum-enforcing providers reject requests
|
||||
// whose input plus max_tokens exceeds the context window, so the cap
|
||||
// shrinks to the remaining window minus the trigger step's output and
|
||||
// the appended summary prompt (ceil(bytes/3) tokens), and degenerate
|
||||
// cases stay unchanged.
|
||||
func TestGenerateCompaction_ClampsSummaryCapToRemainingWindow(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
contextLimit int64
|
||||
inputTokens int64
|
||||
outputTokens int64
|
||||
toolResultBytes int
|
||||
cap int64
|
||||
wantCap int64
|
||||
}{
|
||||
{name: "clamps to remaining window", contextLimit: 100, inputTokens: 75, outputTokens: 10, cap: 64_000, wantCap: 13},
|
||||
{name: "reserves trailing tool results", contextLimit: 100, inputTokens: 72, toolResultBytes: 30, cap: 64_000, wantCap: 16},
|
||||
{name: "keeps cap that fits", contextLimit: 200_000, inputTokens: 140_000, outputTokens: 500, cap: 50_000, wantCap: 50_000},
|
||||
{name: "usage at limit leaves cap unchanged", contextLimit: 100, inputTokens: 100, cap: 64_000, wantCap: 64_000},
|
||||
{name: "reserves leave no room, cap unchanged", contextLimit: 100, inputTokens: 80, outputTokens: 25, cap: 64_000, wantCap: 64_000},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var gotCap *int64
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
StreamFn: func(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
gotCap = call.MaxOutputTokens
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "summary"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
capTokens := tc.cap
|
||||
messages := []fantasy.Message{textMessage(fantasy.MessageRoleUser, "hello")}
|
||||
if tc.toolResultBytes > 0 {
|
||||
messages = append(messages, fantasy.Message{
|
||||
Role: fantasy.MessageRoleTool,
|
||||
Content: []fantasy.MessagePart{fantasy.ToolResultPart{
|
||||
ToolCallID: "call-1",
|
||||
Output: fantasy.ToolResultOutputContentText{Text: strings.Repeat("x", tc.toolResultBytes)},
|
||||
}},
|
||||
})
|
||||
}
|
||||
result, err := GenerateCompaction(context.Background(), GenerateCompactionOptions{
|
||||
Model: model,
|
||||
Messages: messages,
|
||||
Clock: quartz.NewMock(t),
|
||||
ThresholdPercent: 70,
|
||||
ContextLimit: tc.contextLimit,
|
||||
SummaryPrompt: "prompt",
|
||||
StepUsage: fantasy.Usage{InputTokens: tc.inputTokens, OutputTokens: tc.outputTokens},
|
||||
SummaryCall: fantasy.Call{MaxOutputTokens: &capTokens},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "summary", result.SummaryReport)
|
||||
require.NotNil(t, gotCap)
|
||||
require.Equal(t, tc.wantCap, *gotCap)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestGenerateCompaction_DefaultSourceAutomatic verifies an unforced
|
||||
// over-threshold run reports the automatic source by default.
|
||||
func TestGenerateCompaction_DefaultSourceAutomatic(t *testing.T) {
|
||||
@@ -678,12 +924,13 @@ func TestGenerateCompaction_DefaultSourceAutomatic(t *testing.T) {
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
GenerateFn: func(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
return &fantasy.Response{
|
||||
Content: []fantasy.Content{
|
||||
fantasy.TextContent{Text: "auto summary"},
|
||||
},
|
||||
}, nil
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "auto summary"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
), nil
|
||||
},
|
||||
}
|
||||
result, err := GenerateCompaction(context.Background(), GenerateCompactionOptions{
|
||||
@@ -707,11 +954,13 @@ func TestGenerateCompaction_SummaryEstimate(t *testing.T) {
|
||||
var parts []codersdk.ChatMessagePart
|
||||
result, err := GenerateCompaction(t.Context(), GenerateCompactionOptions{
|
||||
Model: &chattest.FakeModel{
|
||||
GenerateFn: func(context.Context, fantasy.Call) (*fantasy.Response, error) {
|
||||
return &fantasy.Response{
|
||||
Content: []fantasy.Content{fantasy.TextContent{Text: "界x"}},
|
||||
Usage: fantasy.Usage{InputTokens: 900, OutputTokens: 400},
|
||||
}, nil
|
||||
StreamFn: func(context.Context, fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return compactionStream(
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "界x"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
), nil
|
||||
},
|
||||
},
|
||||
Messages: []fantasy.Message{textMessage(fantasy.MessageRoleUser, "hello")},
|
||||
|
||||
@@ -130,12 +130,13 @@ func TestGenerateCompaction_RecordsRuntime(t *testing.T) {
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "test-provider",
|
||||
ModelName: "test-model",
|
||||
GenerateFn: func(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
clock.Advance(1500 * time.Millisecond)
|
||||
return &fantasy.Response{
|
||||
Content: []fantasy.Content{
|
||||
fantasy.TextContent{Text: "summary"},
|
||||
},
|
||||
return func(yield func(fantasy.StreamPart) bool) {
|
||||
yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"})
|
||||
yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "summary"})
|
||||
yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"})
|
||||
yield(fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop})
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
|
||||
@@ -146,11 +146,11 @@ func TestManualCompactionPostCompactEffects(t *testing.T) {
|
||||
var compactionCalls atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionCalls.Add(1)
|
||||
return anthropicCompactionResponse(t, req, "manual hook compaction summary")
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionCalls.Add(1)
|
||||
return anthropicCompactionResponse("manual hook compaction summary")
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
// Low usage keeps automatic compaction out of the way, so
|
||||
@@ -232,12 +232,16 @@ func startCompactionHookChat(
|
||||
var streamCalls atomic.Int32
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
body := anthropicRequestBody(t, *req)
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionCalls.Add(1)
|
||||
inspectCompaction(t, body)
|
||||
// The 32000 default cap with 50% headroom. The remaining
|
||||
// window (100 limit - 80 input - trigger output - summary
|
||||
// prompt reserve) is non-positive, so the cap is unchanged.
|
||||
require.Equal(t, 48000, req.MaxTokens)
|
||||
return anthropicCompactionResponse(t, req, "hook compaction summary")
|
||||
}
|
||||
if !req.Stream {
|
||||
if strings.Contains(body, "You are performing a context compaction") {
|
||||
compactionCalls.Add(1)
|
||||
inspectCompaction(t, body)
|
||||
return anthropicCompactionResponse("hook compaction summary")
|
||||
}
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
if streamCalls.Add(1) == 1 {
|
||||
|
||||
@@ -182,9 +182,8 @@ func TestPrepareGenerationClampsRequestedReasoningEffortToMax(t *testing.T) {
|
||||
require.Equal(t, prepared.CallTemplate.ProviderOptions, summaryCall.ProviderOptions)
|
||||
require.NotNil(t, summaryCall.ToolChoice)
|
||||
require.Equal(t, fantasy.ToolChoiceNone, *summaryCall.ToolChoice)
|
||||
// Non-streaming summaries must not inherit the default output cap the
|
||||
// Anthropic SDK rejects.
|
||||
require.Nil(t, summaryCall.MaxOutputTokens)
|
||||
require.NotNil(t, summaryCall.MaxOutputTokens)
|
||||
require.Equal(t, 3*defaultChatMaxOutputTokens/2, *summaryCall.MaxOutputTokens)
|
||||
}
|
||||
|
||||
func TestPrepareGenerationReplacesUnsupportedToolMedia(t *testing.T) {
|
||||
|
||||
@@ -199,14 +199,22 @@ func (r resolvedModelCall) newCall() fantasy.Call {
|
||||
}
|
||||
|
||||
// compactionSummaryCall follows the resolved call template, except summaries
|
||||
// must not call tools and must not carry the default output cap: the summary
|
||||
// request is non-streaming, and the Anthropic SDK rejects non-streaming
|
||||
// requests whose max_tokens implies a completion longer than ten minutes.
|
||||
// must not call tools. Streaming avoids the Anthropic SDK's non-streaming
|
||||
// duration limit, while the explicit cap prevents adaptive thinking from
|
||||
// exhausting fantasy's smaller provider default before producing summary text.
|
||||
// The cap gets 50% headroom because the summary must fit reasoning plus
|
||||
// summary text in one response; a low configured chat cap could otherwise
|
||||
// truncate the summary that replaces pruned history. The headroom stays
|
||||
// modest because a cap configured at a provider's output ceiling overshoots
|
||||
// it; that residual overshoot is an accepted tradeoff. GenerateCompaction
|
||||
// bounds the increased cap by the remaining context window at trigger time.
|
||||
func compactionSummaryCall(resolved resolvedModelCall) fantasy.Call {
|
||||
call := resolved.newCall()
|
||||
toolChoiceNone := fantasy.ToolChoiceNone
|
||||
call.ToolChoice = &toolChoiceNone
|
||||
call.MaxOutputTokens = nil
|
||||
if call.MaxOutputTokens != nil {
|
||||
call.MaxOutputTokens = ptr.Ref(*call.MaxOutputTokens * 3 / 2)
|
||||
}
|
||||
return call
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user