| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "testing" |
| 11 | |
| 12 | "reasonix/internal/event" |
| 13 | "reasonix/internal/provider" |
| 14 | "reasonix/internal/tool" |
| 15 | ) |
| 16 | |
| 17 | const deepSeekOverflowBody = `{"error":{"message":"This model's maximum context length is %d tokens. However, you requested %d tokens (%d in the messages, %d in the completion). Please reduce the length of the messages or completion.","type":"invalid_request_error","param":null,"code":"invalid_request_error"}}` |
| 18 | |
| 19 | // denseTokenizerProvider counts three characters per token where the agent's |
| 20 | // cold estimate assumes four, so a fold the estimator believes fits overflows |
| 21 | // on the wire exactly as #9818 reported. Its overflow reply is the parsed |
| 22 | // DeepSeek 400 body, so the feedback path sees what production sees. |
| 23 | // alwaysOverflow reports every prompt as at least the window, so each reply |
| 24 | // still justifies its rejection while no summary form can ever land. |
| 25 | type denseTokenizerProvider struct { |
| 26 | mu sync.Mutex |
| 27 | window int |
| 28 | alwaysOverflow bool |
| 29 | // unnumberedReplay rejects replay-form summaries with a bare overflow that |
| 30 | // carries no token numbers, the shape GLM reports (#9878). |
| 31 | unnumberedReplay bool |
| 32 | requests []provider.Request |
| 33 | overflows int |
| 34 | } |
| 35 | |
| 36 | func (p *denseTokenizerProvider) Name() string { return "dense-tokenizer" } |
| 37 | |
| 38 | func (p *denseTokenizerProvider) ContextBudgetPolicy() provider.ContextBudgetPolicy { |
| 39 | return provider.ContextBudgetPolicy{ |
| 40 | WindowMode: provider.ContextWindowShared, AutoOutputTokens: 8192, MaxOutputTokens: 8192, |
| 41 | LimitMode: provider.OutputLimitOmitWhenSafe, |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | func denseTokens(req provider.Request) int { |
| 46 | chars, _, _ := requestCalibrationTextShape(req, provider.SharedWindowInputPolicy{}) |
| 47 | return int(chars) / 3 |
| 48 | } |
| 49 | |
| 50 | func isSummaryRequest(req provider.Request) bool { |
| 51 | return len(req.Messages) > 0 && strings.Contains(req.Messages[len(req.Messages)-1].Content, "Compact the preceding conversation prefix") |
| 52 | } |
| 53 | |
| 54 | func (p *denseTokenizerProvider) Stream(_ context.Context, req provider.Request) (<-chan provider.Chunk, error) { |
| 55 | p.mu.Lock() |
| 56 | defer p.mu.Unlock() |
| 57 | req.Messages = append([]provider.Message(nil), req.Messages...) |
| 58 | p.requests = append(p.requests, req) |
| 59 | if p.unnumberedReplay && isSummaryRequest(req) && req.Messages[0].Content != slimSummarySystemPrompt { |
| 60 | p.overflows++ |
| 61 | return nil, &provider.ContextLimitError{APIError: &provider.APIError{ |
| 62 | Provider: p.Name(), Status: 400, Body: `{"error":{"code":"1261","message":"Prompt exceeds max length"}}`, |
| 63 | }} |
| 64 | } |
| 65 | prompt := denseTokens(req) |
| 66 | if p.unnumberedReplay { |
| 67 | prompt = prompt * 3 / 4 // an ordinary tokenizer: only the replay form was rejected |
| 68 | } |
| 69 | if p.alwaysOverflow { |
| 70 | prompt = max(prompt, p.window) |
| 71 | } |
| 72 | completion := req.MaxTokens |
| 73 | if completion <= 0 { |
| 74 | completion = 8192 |
| 75 | } |
| 76 | if prompt+completion > p.window { |
| 77 | p.overflows++ |
| 78 | body := fmt.Sprintf(deepSeekOverflowBody, p.window, prompt+completion, prompt, completion) |
| 79 | limit := provider.ParseContextLimitError(&provider.APIError{Provider: p.Name(), Status: 400, Body: body}) |
| 80 | if limit == nil { |
| 81 | return nil, fmt.Errorf("test body did not parse as a context limit: %s", body) |
| 82 | } |
| 83 | return nil, limit |
| 84 | } |
| 85 | text := "ok" |
| 86 | if isSummaryRequest(req) { |
| 87 | text = "- goal: keep going\n- pending: continue" |
| 88 | } |
| 89 | return chunks( |
| 90 | provider.Chunk{Type: provider.ChunkText, Text: text}, |
| 91 | provider.Chunk{Type: provider.ChunkUsage, Usage: &provider.Usage{PromptTokens: prompt, CompletionTokens: 8, TotalTokens: prompt + 8, RequestCount: 1}}, |
| 92 | provider.Chunk{Type: provider.ChunkDone}, |
| 93 | ), nil |
| 94 | } |
| 95 | |
| 96 | func (p *denseTokenizerProvider) summaryRequests() []provider.Request { |
| 97 | p.mu.Lock() |
| 98 | defer p.mu.Unlock() |
| 99 | var out []provider.Request |
| 100 | for _, req := range p.requests { |
| 101 | if isSummaryRequest(req) { |
| 102 | out = append(out, req) |
| 103 | } |
| 104 | } |
| 105 | return out |
| 106 | } |
| 107 | |
| 108 | func requestFingerprints(reqs []provider.Request) map[string]int { |
| 109 | seen := map[string]int{} |
| 110 | for _, req := range reqs { |
| 111 | seen[providerVisibleFingerprint(req.Messages)]++ |
| 112 | } |
| 113 | return seen |
| 114 | } |
| 115 | |
| 116 | func longASCIISession(turns int) *Session { |
| 117 | big := strings.Repeat("alpha beta gamma delta ", 200) |
| 118 | sess := NewSession("sys") |
| 119 | sess.Add(provider.Message{Role: provider.RoleUser, Content: "standing constraint: keep the public API stable"}) |
| 120 | for i := range turns { |
| 121 | sess.Add(provider.Message{Role: provider.RoleAssistant, Content: fmt.Sprintf("step %d: %s", i, big)}) |
| 122 | sess.Add(provider.Message{Role: provider.RoleUser, Content: "continue"}) |
| 123 | } |
| 124 | return sess |
| 125 | } |
| 126 | |
| 127 | // The estimator plans the largest prefix it believes fits; the provider counts |
| 128 | // denser and rejects it. The overflow must recalibrate the estimator and the |
| 129 | // re-planned request must be strictly smaller, landing a real digest without |
| 130 | // the fragment path and without ever repeating the rejected request. |
| 131 | func TestSummaryOverflowRecalibratesAndReplansSmaller(t *testing.T) { |
| 132 | prov := &denseTokenizerProvider{window: 20_000} |
| 133 | sess := longASCIISession(16) |
| 134 | a := New(prov, tool.NewRegistry(), sess, Options{ContextWindow: 20_000, CompactRatio: 0.8}, event.Discard) |
| 135 | if est, fold, hard := a.ContextUsedTokens(), a.compactTrigger(), a.hardInputCeiling(); est < fold || est >= hard { |
| 136 | t.Fatalf("fixture estimates %d tokens; want between the trigger %d and the ceiling %d", est, fold, hard) |
| 137 | } |
| 138 | |
| 139 | if err := prepareContext(context.Background(), a, CompactionTriggerPressure); err != nil { |
| 140 | t.Fatalf("prepare = %v", err) |
| 141 | } |
| 142 | summaries := prov.summaryRequests() |
| 143 | if prov.overflows != 1 || len(summaries) != 2 { |
| 144 | t.Fatalf("overflows=%d summaries=%d, want one rejected replay and one re-planned success", prov.overflows, len(summaries)) |
| 145 | } |
| 146 | if first, second := denseTokens(summaries[0]), denseTokens(summaries[1]); second >= first { |
| 147 | t.Fatalf("re-planned summary request %d tokens is not smaller than the rejected %d", second, first) |
| 148 | } |
| 149 | for fp, n := range requestFingerprints(summaries) { |
| 150 | if n > 1 { |
| 151 | t.Fatalf("summary request %s was sent %d times", fp, n) |
| 152 | } |
| 153 | } |
| 154 | if ratio := a.tokPerChar(); ratio < 0.3 { |
| 155 | t.Fatalf("calibration ratio %.3f did not learn the provider's denser tokenizer", ratio) |
| 156 | } |
| 157 | r := a.sess.compactionState.LastReceipt |
| 158 | if r == nil || r.Status != "applied" || r.Action != "summary" || latestDigest(a.sess.compactionState.Projection.Messages) == "" { |
| 159 | t.Fatalf("receipt = %+v, want an applied summary with a digest", r) |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | // An overflow reply without token numbers cannot recalibrate anything, so a |
| 164 | // re-plan would resend the same bytes. The ladder must skip straight to the |
| 165 | // transcript form and must not learn a ratio or window from zero fields. |
| 166 | func TestUnnumberedSummaryOverflowSkipsReplanToTranscript(t *testing.T) { |
| 167 | prov := &denseTokenizerProvider{window: 20_000, unnumberedReplay: true} |
| 168 | reg := tool.NewRegistry() |
| 169 | reg.Add(schemaTool{}) |
| 170 | sess := longASCIISession(16) |
| 171 | a := New(prov, reg, sess, Options{ContextWindow: 20_000, CompactRatio: 0.8}, event.Discard) |
| 172 | |
| 173 | if err := prepareContext(context.Background(), a, CompactionTriggerPressure); err != nil { |
| 174 | t.Fatalf("prepare = %v", err) |
| 175 | } |
| 176 | summaries := prov.summaryRequests() |
| 177 | if prov.overflows != 1 || len(summaries) != 2 { |
| 178 | t.Fatalf("overflows=%d summaries=%d, want one rejected replay and one transcript-form success", prov.overflows, len(summaries)) |
| 179 | } |
| 180 | if slim := summaries[1]; len(slim.Tools) != 0 || len(slim.Messages) != 2 { |
| 181 | t.Fatalf("second request = %d tools, %d messages; want the transcript form, not a re-planned replay", len(slim.Tools), len(slim.Messages)) |
| 182 | } |
| 183 | if ratio := a.tokPerChar(); ratio != fallbackTokPerChar { |
| 184 | t.Fatalf("calibration ratio %.3f changed on an overflow without token numbers", ratio) |
| 185 | } |
| 186 | if window := a.effectiveContextWindow(); window != 20_000 { |
| 187 | t.Fatalf("effective window %d changed on an overflow without token numbers", window) |
| 188 | } |
| 189 | r := a.sess.compactionState.LastReceipt |
| 190 | if r == nil || r.Status != "applied" || r.Action != "summary" || latestDigest(a.sess.compactionState.Projection.Messages) == "" { |
| 191 | t.Fatalf("receipt = %+v, want an applied summary with a digest", r) |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | // A provider that rejects every summary form must not trap /compact in a loop |
| 196 | // of identical requests: replay re-plans, then the transcript form, then the |
| 197 | // fragment path, and at the ceiling the truncation rescue finally lands. |
| 198 | func TestManualCompactOverCeilingFailsWithoutRepeatingRequests(t *testing.T) { |
| 199 | prov := &denseTokenizerProvider{window: 20_000, alwaysOverflow: true} |
| 200 | sess := longASCIISession(30) |
| 201 | a := New(prov, tool.NewRegistry(), sess, Options{ContextWindow: 20_000, CompactRatio: 0.8}, event.Discard) |
| 202 | if est, hard := a.ContextUsedTokens(), a.hardInputCeiling(); est < hard { |
| 203 | t.Fatalf("fixture estimates %d tokens against a %d ceiling; it is not over it", est, hard) |
| 204 | } |
| 205 | |
| 206 | if err := a.CompactNow(context.Background(), ""); !errors.Is(err, ErrCompactionRequired) { |
| 207 | t.Fatalf("CompactNow = %v, want a recoverable context failure", err) |
| 208 | } |
| 209 | assertNoFailedSummaryRewrite(t, a) |
| 210 | summaries := prov.summaryRequests() |
| 211 | if len(summaries) < 3 { |
| 212 | t.Fatalf("summary requests = %d, want replay re-plans and the transcript form before the rescue", len(summaries)) |
| 213 | } |
| 214 | for fp, n := range requestFingerprints(summaries) { |
| 215 | if n > 1 { |
| 216 | t.Fatalf("summary request %s was sent %d times", fp, n) |
| 217 | } |
| 218 | } |
| 219 | slim := 0 |
| 220 | for _, req := range summaries { |
| 221 | if len(req.Tools) == 0 && len(req.Messages) == 2 { |
| 222 | slim++ |
| 223 | } |
| 224 | } |
| 225 | if slim != 1 { |
| 226 | t.Fatalf("transcript-form summary requests = %d, want exactly one rung", slim) |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | type schemaTool struct{} |
| 231 | |
| 232 | func (schemaTool) Name() string { return "read_file" } |
| 233 | func (schemaTool) Description() string { return "Read a file." } |
| 234 | func (schemaTool) Schema() json.RawMessage { |
| 235 | return json.RawMessage(`{"type":"object","properties":{"path":{"type":"string"}}}`) |
| 236 | } |
| 237 | func (schemaTool) ReadOnly() bool { return true } |
| 238 | func (schemaTool) Execute(context.Context, json.RawMessage) (string, error) { |
| 239 | return "", nil |
| 240 | } |
| 241 | |
| 242 | func TestSlimSummaryRequestIsBoundedAndToolFree(t *testing.T) { |
| 243 | reg := tool.NewRegistry() |
| 244 | reg.Add(schemaTool{}) |
| 245 | a := New(&denseTokenizerProvider{window: 1 << 20}, reg, NewSession("sys"), Options{ContextWindow: 1 << 20}, event.Discard) |
| 246 | body := strings.Repeat("0123456789", 3000) |
| 247 | fold := []provider.Message{ |
| 248 | {Role: provider.RoleUser, Content: "read it"}, |
| 249 | {Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: "c1", Name: "read_file", Arguments: `{"path":"x"}`}}}, |
| 250 | {Role: provider.RoleTool, ToolCallID: "c1", Name: "read_file", Content: body, Images: []string{"data:image/png;base64,AAAA"}}, |
| 251 | } |
| 252 | |
| 253 | replay := a.summaryRequest(fold, "") |
| 254 | if len(replay.Tools) == 0 { |
| 255 | t.Fatal("replay form must carry the tool schemas the sampling request uses") |
| 256 | } |
| 257 | slim := a.slimSummaryRequest(fold, "") |
| 258 | if len(slim.Tools) != 0 || len(slim.Messages) != 2 { |
| 259 | t.Fatalf("slim form = %d tools, %d messages; want no schemas and one transcript turn", len(slim.Tools), len(slim.Messages)) |
| 260 | } |
| 261 | text := slim.Messages[1].Content |
| 262 | if !strings.Contains(text, "tool result truncated for summarization") || strings.Contains(text, "base64") { |
| 263 | t.Fatal("slim transcript must cut the tool body and drop images") |
| 264 | } |
| 265 | if len(text) > slimToolResultRunes+2000 { |
| 266 | t.Fatalf("slim transcript is %d bytes; the tool body should be bounded by %d runes", len(text), slimToolResultRunes) |
| 267 | } |
| 268 | if got, want := a.estimatedRequestTokens(slim), a.estimatedRequestTokens(replay); got >= want { |
| 269 | t.Fatalf("slim request estimates %d tokens, not smaller than the replay's %d", got, want) |
| 270 | } |
| 271 | } |
| 272 | |
| 273 | func TestChunkedFallbackAppliesOnlyAfterTranscriptForm(t *testing.T) { |
| 274 | overflow := &provider.ContextLimitError{WindowTokens: 10, PromptTokens: 20} |
| 275 | if chunkedFallbackApplies(overflow, SummaryInputCachePrefix) { |
| 276 | t.Fatal("a replay overflow should be re-planned, not fragmented") |
| 277 | } |
| 278 | if !chunkedFallbackApplies(overflow, SummaryInputSlim) { |
| 279 | t.Fatal("an overflow of the transcript form has no cheaper rung left") |
| 280 | } |
| 281 | if !chunkedFallbackApplies(errSummaryOutputTruncated, SummaryInputCachePrefix) || !chunkedFallbackApplies(ErrCompactionRequired, SummaryInputCachePrefix) { |
| 282 | t.Fatal("output truncation and local admission keep their direct fragment path") |
| 283 | } |
| 284 | } |
| 285 | |
| 286 | func TestActiveTurnFoldBoundaryKeepsNewestRounds(t *testing.T) { |
| 287 | round := func(i int) []provider.Message { |
| 288 | id := fmt.Sprintf("c%d", i) |
| 289 | return []provider.Message{ |
| 290 | {Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: id, Name: "read_file", Arguments: "{}"}}}, |
| 291 | {Role: provider.RoleTool, ToolCallID: id, Name: "read_file", Content: "body"}, |
| 292 | } |
| 293 | } |
| 294 | msgs := []provider.Message{{Role: provider.RoleSystem, Content: "sys"}, {Role: provider.RoleUser, Content: "task", CreatedAt: 7}} |
| 295 | for i := range 4 { |
| 296 | msgs = append(msgs, round(i)...) |
| 297 | } |
| 298 | // Rounds occupy [2,4) [4,6) [6,8) [8,10); the newest two stay verbatim. |
| 299 | if got := activeTurnFoldBoundary(msgs, 1, len(msgs)); got != 6 { |
| 300 | t.Fatalf("boundary = %d, want 6 (fold prompt + two oldest rounds)", got) |
| 301 | } |
| 302 | short := msgs[:6] |
| 303 | if got := activeTurnFoldBoundary(short, 1, len(short)); got != 1 { |
| 304 | t.Fatalf("boundary = %d, want the turn kept whole when it has only the rounds to keep", got) |
| 305 | } |
| 306 | if got := activeTurnFoldBoundary(msgs, 1, 7); got != 4 { |
| 307 | t.Fatalf("boundary = %d, want 4 when the fold end cuts the newest rounds off", got) |
| 308 | } |
| 309 | } |
| 310 | |
| 311 | func TestFailedReceiptLiftsWhenViewOutgrowsFailure(t *testing.T) { |
| 312 | const window = 10_000 |
| 313 | sess := &Session{Messages: []provider.Message{ |
| 314 | {Role: provider.RoleSystem, Content: "system"}, |
| 315 | {Role: provider.RoleUser, Content: "task"}, |
| 316 | {Role: provider.RoleAssistant, Content: strings.Repeat("old work ", 500)}, |
| 317 | {Role: provider.RoleUser, Content: "current"}, |
| 318 | {Role: provider.RoleAssistant, Content: "tail"}, |
| 319 | }} |
| 320 | prov := &failingSummaryProvider{} |
| 321 | a := New(prov, tool.NewRegistry(), sess, Options{ContextWindow: window, CompactRatio: 0.85, RecentKeep: 2}, event.Discard) |
| 322 | policy := ContextPreparePolicy{Trigger: CompactionTriggerPressure, ObservedInputTokens: 8600} |
| 323 | a.activeTurnCreatedAt.Store(11) |
| 324 | |
| 325 | if _, err := a.contextManager().Prepare(context.Background(), policy); err != nil { |
| 326 | t.Fatal(err) |
| 327 | } |
| 328 | sess.Add(provider.Message{Role: provider.RoleTool, Content: strings.Repeat("small output ", 40)}) |
| 329 | if _, err := a.contextManager().Prepare(context.Background(), policy); err != nil { |
| 330 | t.Fatal(err) |
| 331 | } |
| 332 | if prov.calls != 1 { |
| 333 | t.Fatalf("a small same-turn change made %d summary calls, want the backoff to hold", prov.calls) |
| 334 | } |
| 335 | sess.Add(provider.Message{Role: provider.RoleTool, Content: strings.Repeat("large output ", 400)}) |
| 336 | if _, err := a.contextManager().Prepare(context.Background(), policy); err != nil { |
| 337 | t.Fatal(err) |
| 338 | } |
| 339 | if prov.calls != 2 { |
| 340 | t.Fatalf("a view grown by over 5%% of the window made %d summary calls, want the backoff lifted", prov.calls) |
| 341 | } |
| 342 | } |
| 343 |