| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "net/http" |
| 11 | "net/http/httptest" |
| 12 | "os" |
| 13 | "strconv" |
| 14 | "strings" |
| 15 | "testing" |
| 16 | |
| 17 | "reasonix/internal/event" |
| 18 | "reasonix/internal/provider" |
| 19 | "reasonix/internal/provider/openai" |
| 20 | "reasonix/internal/tool" |
| 21 | ) |
| 22 | |
| 23 | // echoTool is a trivial read-only tool used to drive a multi-step tool loop: |
| 24 | // each call appends an assistant(tool_call) + tool(result) pair to the history, |
| 25 | // growing the request prefix the way a real multi-turn session does. |
| 26 | type echoTool struct{} |
| 27 | |
| 28 | func (echoTool) Name() string { return "echo" } |
| 29 | func (echoTool) Description() string { return "echo back the given text" } |
| 30 | func (echoTool) Schema() json.RawMessage { |
| 31 | return json.RawMessage(`{"type":"object","properties":{"text":{"type":"string"}},"required":["text"]}`) |
| 32 | } |
| 33 | func (echoTool) ReadOnly() bool { return true } |
| 34 | func (echoTool) Execute(_ context.Context, args json.RawMessage) (string, error) { |
| 35 | var a struct { |
| 36 | Text string `json:"text"` |
| 37 | } |
| 38 | _ = json.Unmarshal(args, &a) |
| 39 | return "echoed: " + a.Text, nil |
| 40 | } |
| 41 | |
| 42 | // collectSink captures the per-turn Usage events plus any compaction notices the |
| 43 | // agent emits, so the test can replay exactly what the status line would show. |
| 44 | type collectSink struct { |
| 45 | usages []*provider.Usage |
| 46 | notices []string |
| 47 | maintenance []event.ContextMaintenance |
| 48 | blocked bool |
| 49 | } |
| 50 | |
| 51 | func (s *collectSink) Emit(e event.Event) { |
| 52 | switch e.Kind { |
| 53 | case event.Usage: |
| 54 | if e.Usage != nil { |
| 55 | s.usages = append(s.usages, e.Usage) |
| 56 | } |
| 57 | case event.Notice: |
| 58 | s.notices = append(s.notices, e.Text) |
| 59 | case event.ContextMaintenanceEvent: |
| 60 | if e.Maintenance != nil { |
| 61 | s.maintenance = append(s.maintenance, *e.Maintenance) |
| 62 | if e.Maintenance.Status == "blocked" { |
| 63 | s.blocked = true |
| 64 | } |
| 65 | } |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | // mockDeepSeek derives cache hits from the byte-identical prefix shared with |
| 70 | // the previous conversation request, directly measuring prefix stability. |
| 71 | |
| 72 | type mockDeepSeek struct { |
| 73 | t *testing.T |
| 74 | prevMessages []json.RawMessage // last conversation request's messages |
| 75 | reqChars []int // total prompt chars per conversation request |
| 76 | hitChars []int // cached prefix chars per conversation request |
| 77 | withTools bool // advertise the echo tool (and emit tool calls) |
| 78 | reasoning string // chain-of-thought echoed every turn (round-tripped) |
| 79 | toolRounds int // remaining tool-call rounds before a final answer |
| 80 | } |
| 81 | |
| 82 | func (m *mockDeepSeek) handler(w http.ResponseWriter, r *http.Request) { |
| 83 | body, _ := io.ReadAll(r.Body) |
| 84 | |
| 85 | // Compaction appends one final instruction to the ordinary cached prefix. |
| 86 | // Answer it with a short summary and do not let it replace conversation |
| 87 | // bookkeeping; the replayed prefix itself must match the prior request. |
| 88 | if isSummarizeRequest(body) { |
| 89 | msgs := decodeMessages(body) |
| 90 | replayed := msgs[:len(msgs)-1] |
| 91 | if len(m.prevMessages) > 0 && commonPrefixMsgs(m.prevMessages, replayed) != len(replayed) { |
| 92 | m.t.Errorf("summary request did not replay a byte-identical cached conversation prefix") |
| 93 | } |
| 94 | writeSSE(w, m.t, |
| 95 | streamChunk(deltaText("- goal: keep going\n- decisions: none\n- pending: continue")), |
| 96 | finishChunk("stop"), |
| 97 | usageChunk(100, 40, 0, 100), |
| 98 | ) |
| 99 | return |
| 100 | } |
| 101 | |
| 102 | msgs := decodeMessages(body) |
| 103 | common := commonPrefixMsgs(m.prevMessages, msgs) |
| 104 | hitChars := charsOf(msgs[:common]) |
| 105 | totalChars := charsOf(msgs) |
| 106 | m.prevMessages = msgs |
| 107 | m.reqChars = append(m.reqChars, totalChars) |
| 108 | m.hitChars = append(m.hitChars, hitChars) |
| 109 | |
| 110 | promptTok := totalChars / 4 |
| 111 | hitTok := hitChars / 4 |
| 112 | missTok := promptTok - hitTok |
| 113 | |
| 114 | emitTool := m.withTools && m.toolRounds > 0 |
| 115 | if emitTool { |
| 116 | m.toolRounds-- |
| 117 | } |
| 118 | |
| 119 | chunks := []sseResp{streamChunk(deltaReasoning(m.reasoning))} |
| 120 | if emitTool { |
| 121 | idx := len(m.reqChars) |
| 122 | chunks = append(chunks, |
| 123 | streamChunk(deltaToolCall(idx, "echo", fmt.Sprintf(`{"text":"round-%d"}`, idx))), |
| 124 | finishChunk("tool_calls")) |
| 125 | } else { |
| 126 | chunks = append(chunks, |
| 127 | streamChunk(deltaText("Done.")), |
| 128 | finishChunk("stop")) |
| 129 | } |
| 130 | chunks = append(chunks, usageChunk(promptTok, 50, hitTok, missTok)) |
| 131 | writeSSE(w, m.t, chunks...) |
| 132 | } |
| 133 | |
| 134 | func (m *mockDeepSeek) tools() *tool.Registry { |
| 135 | reg := tool.NewRegistry() |
| 136 | if m.withTools { |
| 137 | reg.Add(echoTool{}) |
| 138 | } |
| 139 | return reg |
| 140 | } |
| 141 | |
| 142 | // hitRate is the status-line formula: hit / (hit+miss), falling back to prompt. |
| 143 | func hitRate(u *provider.Usage) int { |
| 144 | denom := u.CacheHitTokens + u.CacheMissTokens |
| 145 | if denom == 0 { |
| 146 | denom = u.PromptTokens |
| 147 | } |
| 148 | if denom == 0 { |
| 149 | return 0 |
| 150 | } |
| 151 | return u.CacheHitTokens * 100 / denom |
| 152 | } |
| 153 | |
| 154 | const systemPrompt = "You are reasonix, a coding agent. Be concise and follow project conventions. " + |
| 155 | "This system prompt is the cacheable head of every request and must never change between turns." |
| 156 | |
| 157 | // longReasoning stands in for a deepseek-reasoner chain-of-thought that the agent |
| 158 | // round-trips onto the assistant turn (agent.go round-trips ReasoningContent). |
| 159 | const longReasoning = "Let me reason about this carefully. I will weigh the constraints, " + |
| 160 | "enumerate the candidate approaches, reject the ones that violate a requirement, and then " + |
| 161 | "commit to the most defensible option, double-checking it against the original goal before answering." |
| 162 | |
| 163 | // TestCacheHitPrefixStable proves the standard path keeps a byte-stable prefix: |
| 164 | // every request re-sends the full prior history untouched, and the displayed |
| 165 | // hit% equals hit/prompt%. This rules out "something is breaking the cache" and |
| 166 | // "the display math is wrong" for the no-compaction path. |
| 167 | func TestCacheHitPrefixStable(t *testing.T) { |
| 168 | mock := &mockDeepSeek{t: t, withTools: true, reasoning: longReasoning, toolRounds: 2} |
| 169 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 170 | defer srv.Close() |
| 171 | |
| 172 | a, sink := newAgent(t, srv.URL, mock.tools(), 0 /*no compaction*/, 0) |
| 173 | if err := a.Run(context.Background(), "echo a couple things then finish"); err != nil { |
| 174 | t.Fatalf("Run: %v", err) |
| 175 | } |
| 176 | |
| 177 | // Reconstruct the requests to check prefix stability. Replay equality is |
| 178 | // already encoded in hitChars==full-previous-prefix, but assert it directly. |
| 179 | for i := 1; i < len(mock.reqChars); i++ { |
| 180 | // On request i the cached prefix should be the ENTIRE request i-1. |
| 181 | if mock.hitChars[i] != mock.reqChars[i-1] { |
| 182 | t.Errorf("PREFIX BROKEN at req %d: cached %d chars but the full prior request was %d chars", |
| 183 | i, mock.hitChars[i], mock.reqChars[i-1]) |
| 184 | } |
| 185 | } |
| 186 | t.Logf("prefix STABLE across %d requests — nothing in the client breaks the cache", len(mock.reqChars)) |
| 187 | |
| 188 | t.Logf("==== reported usage (what the status line renders) ====") |
| 189 | for i, u := range sink.usages { |
| 190 | want := -1 |
| 191 | if u.PromptTokens > 0 { |
| 192 | want = 100 * u.CacheHitTokens / u.PromptTokens |
| 193 | } |
| 194 | t.Logf("turn %d: prompt=%d hit=%d miss=%d → 'cache %d%%' (hit/prompt=%d%%) | %s", |
| 195 | i, u.PromptTokens, u.CacheHitTokens, u.CacheMissTokens, hitRate(u), want, |
| 196 | strings.TrimSpace(FormatUsageLine(u, nil, nil))) |
| 197 | if u.CacheHitTokens+u.CacheMissTokens != u.PromptTokens { |
| 198 | t.Errorf("display denominator mismatch: hit+miss=%d != prompt=%d (status%% would read wrong)", |
| 199 | u.CacheHitTokens+u.CacheMissTokens, u.PromptTokens) |
| 200 | } |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | // TestCacheHitClimbsWithoutCompaction runs a long multi-turn conversation with |
| 205 | // compaction DISABLED and prints the hit-rate curve. With a stable prefix the |
| 206 | // rate should climb past 90% as history dwarfs each turn's fresh tail. |
| 207 | func TestCacheHitClimbsWithoutCompaction(t *testing.T) { |
| 208 | mock := &mockDeepSeek{t: t, reasoning: longReasoning} |
| 209 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 210 | defer srv.Close() |
| 211 | |
| 212 | a, sink := newAgent(t, srv.URL, mock.tools(), 0 /*no compaction*/, 0) |
| 213 | |
| 214 | const turns = 14 |
| 215 | for i := range turns { |
| 216 | userMsg := "Turn " + fmt.Sprint(i) + ": " + strings.Repeat("please consider this requirement. ", 6) |
| 217 | if err := a.Run(context.Background(), userMsg); err != nil { |
| 218 | t.Fatalf("Run %d: %v", i, err) |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | t.Logf("==== hit-rate curve, NO compaction (%d turns) ====", turns) |
| 223 | peak := 0 |
| 224 | for i, u := range sink.usages { |
| 225 | r := hitRate(u) |
| 226 | if r > peak { |
| 227 | peak = r |
| 228 | } |
| 229 | t.Logf("turn %2d: prompt=%5d hit=%5d miss=%4d → cache %d%%", i, u.PromptTokens, u.CacheHitTokens, u.CacheMissTokens, r) |
| 230 | } |
| 231 | t.Logf("peak hit rate without compaction: %d%%", peak) |
| 232 | if peak < 90 { |
| 233 | t.Logf("NOTE: even with a perfectly stable prefix the rate plateaus below 90%% — "+ |
| 234 | "each turn's fresh tail (incl. %d-char round-tripped reasoning) is too large a share", len(longReasoning)) |
| 235 | } |
| 236 | } |
| 237 | |
| 238 | // A window too small to hold even the system and active tail cannot be repaired |
| 239 | // by fabricating a mechanical digest. |
| 240 | func TestTooSmallWindowReturnsCompactionRequired(t *testing.T) { |
| 241 | mock := &mockDeepSeek{t: t, withTools: true, reasoning: longReasoning, toolRounds: 30} |
| 242 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 243 | defer srv.Close() |
| 244 | |
| 245 | a, sink := newAgent(t, srv.URL, mock.tools(), 900 /*window tok*/, 4 /*recentKeep*/) |
| 246 | |
| 247 | if err := a.Run(context.Background(), strings.Repeat("please consider this requirement. ", 6)); !errors.Is(err, ErrCompactionRequired) { |
| 248 | t.Fatalf("Run = %v, want ErrCompactionRequired", err) |
| 249 | } |
| 250 | _ = sink |
| 251 | } |
| 252 | |
| 253 | // TestReasoningRoundTripCost contrasts the hit-rate curve WITH vs WITHOUT the |
| 254 | // reasoning_content round-trip (agent.go re-sends the assistant chain-of-thought |
| 255 | // every turn). It quantifies how much that round-tripped CoT — assuming DeepSeek |
| 256 | // counts it as uncached prompt — drags the hit rate down at each turn. |
| 257 | func TestReasoningRoundTripCost(t *testing.T) { |
| 258 | curve := func(reasoning string) []int { |
| 259 | mock := &mockDeepSeek{t: t, reasoning: reasoning} |
| 260 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 261 | defer srv.Close() |
| 262 | a, sink := newAgent(t, srv.URL, mock.tools(), 0, 0) |
| 263 | const turns = 12 |
| 264 | for i := range turns { |
| 265 | if err := a.Run(context.Background(), strings.Repeat("please consider this requirement. ", 6)); err != nil { |
| 266 | t.Fatalf("Run %d: %v", i, err) |
| 267 | } |
| 268 | } |
| 269 | out := make([]int, len(sink.usages)) |
| 270 | for i, u := range sink.usages { |
| 271 | out[i] = hitRate(u) |
| 272 | } |
| 273 | return out |
| 274 | } |
| 275 | |
| 276 | withCoT := curve(longReasoning) |
| 277 | without := curve("") |
| 278 | |
| 279 | t.Logf("==== reasoning round-trip: hit-rate cost per turn ====") |
| 280 | t.Logf("turn | with reasoning round-trip | without (stripped) | delta") |
| 281 | firstCross := func(c []int) int { |
| 282 | for i, r := range c { |
| 283 | if r >= 90 { |
| 284 | return i |
| 285 | } |
| 286 | } |
| 287 | return -1 |
| 288 | } |
| 289 | for i := range withCoT { |
| 290 | t.Logf(" %2d | %3d%% | %3d%% | +%d pts", |
| 291 | i, withCoT[i], without[i], without[i]-withCoT[i]) |
| 292 | } |
| 293 | t.Logf("turns needed to reach 90%%: with round-trip = %d, stripped = %d", firstCross(withCoT), firstCross(without)) |
| 294 | } |
| 295 | |
| 296 | // TestSessionAggregateCacheRate verifies the session-aggregate hit-rate the |
| 297 | // status line now shows: Agent.SessionCache() accumulates every turn's hit/miss |
| 298 | // (so it equals the sum of the per-turn usages), and the aggregate rate is the |
| 299 | // steadier, higher number compared to the volatile single-turn rate. |
| 300 | func TestSessionAggregateCacheRate(t *testing.T) { |
| 301 | mock := &mockDeepSeek{t: t, reasoning: longReasoning} |
| 302 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 303 | defer srv.Close() |
| 304 | |
| 305 | a, sink := newAgent(t, srv.URL, mock.tools(), 0, 0) |
| 306 | const turns = 8 |
| 307 | for i := range turns { |
| 308 | if err := a.Run(context.Background(), strings.Repeat("please consider this requirement. ", 6)); err != nil { |
| 309 | t.Fatalf("Run %d: %v", i, err) |
| 310 | } |
| 311 | } |
| 312 | |
| 313 | // The agent's cumulative counters must equal the sum of the per-turn usages. |
| 314 | var sumHit, sumMiss int |
| 315 | for _, u := range sink.usages { |
| 316 | sumHit += u.CacheHitTokens |
| 317 | sumMiss += u.CacheMissTokens |
| 318 | } |
| 319 | hit, miss := a.SessionCache() |
| 320 | if hit != sumHit || miss != sumMiss { |
| 321 | t.Errorf("SessionCache()=%d/%d but per-turn sums are %d/%d", hit, miss, sumHit, sumMiss) |
| 322 | } |
| 323 | |
| 324 | agg := 100 * hit / (hit + miss) |
| 325 | last := sink.usages[len(sink.usages)-1] |
| 326 | single := 100 * last.CacheHitTokens / last.PromptTokens |
| 327 | t.Logf("after %d turns: aggregate(session) = %d%% vs single(last turn) = %d%%", turns, agg, single) |
| 328 | if agg <= 0 || agg > 100 { |
| 329 | t.Errorf("aggregate rate out of range: %d%%", agg) |
| 330 | } |
| 331 | } |
| 332 | |
| 333 | func TestSetSessionResetsSessionCache(t *testing.T) { |
| 334 | mock := &mockDeepSeek{t: t, reasoning: longReasoning} |
| 335 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 336 | defer srv.Close() |
| 337 | |
| 338 | a, _ := newAgent(t, srv.URL, mock.tools(), 0, 0) |
| 339 | if err := a.Run(context.Background(), strings.Repeat("please consider this requirement. ", 6)); err != nil { |
| 340 | t.Fatalf("Run: %v", err) |
| 341 | } |
| 342 | hit, miss := a.SessionCache() |
| 343 | if hit+miss == 0 { |
| 344 | t.Fatalf("SessionCache()=%d/%d before reset, want telemetry to record the turn", hit, miss) |
| 345 | } |
| 346 | a.SetSession(NewSession("system")) |
| 347 | hit, miss = a.SessionCache() |
| 348 | if hit != 0 || miss != 0 { |
| 349 | t.Fatalf("SessionCache()=%d/%d after SetSession, want reset", hit, miss) |
| 350 | } |
| 351 | } |
| 352 | |
| 353 | func TestReleaseCacheHitGuard(t *testing.T) { |
| 354 | if os.Getenv("REASONIX_RELEASE_CACHE_GUARD") == "" { |
| 355 | t.Skip("set REASONIX_RELEASE_CACHE_GUARD=1 to run the release cache guard") |
| 356 | } |
| 357 | |
| 358 | threshold := envInt("REASONIX_CACHE_GUARD_THRESHOLD", 90) |
| 359 | maxLowCases := envInt("REASONIX_CACHE_GUARD_MAX_LOW_CASES", 1) |
| 360 | for _, size := range []int{64 << 10, 256 << 10} { |
| 361 | verifyLargeToolOutputCacheContract(t, size) |
| 362 | t.Logf("CACHE_GUARD_RESULT: case=large-tool-%dk status=pass provider_bytes_max=%d", size>>10, maxToolOutputBytes) |
| 363 | } |
| 364 | |
| 365 | cases := []struct { |
| 366 | name string |
| 367 | run func(*testing.T) []int |
| 368 | }{ |
| 369 | { |
| 370 | name: "plain-dialogue", |
| 371 | run: func(t *testing.T) []int { |
| 372 | return cacheCurve(t, &mockDeepSeek{t: t, reasoning: longReasoning}, 14) |
| 373 | }, |
| 374 | }, |
| 375 | { |
| 376 | name: "plain-dialogue-no-reasoning", |
| 377 | run: func(t *testing.T) []int { |
| 378 | return cacheCurve(t, &mockDeepSeek{t: t}, 14) |
| 379 | }, |
| 380 | }, |
| 381 | { |
| 382 | name: "long-dialogue", |
| 383 | run: func(t *testing.T) []int { |
| 384 | return cacheCurveWithMessages(t, &mockDeepSeek{t: t, reasoning: longReasoning}, repeatedMessages(18, 18)) |
| 385 | }, |
| 386 | }, |
| 387 | { |
| 388 | name: "mixed-message-sizes", |
| 389 | run: func(t *testing.T) []int { |
| 390 | msgs := make([]string, 0, 20) |
| 391 | for i := range 20 { |
| 392 | repeats := 4 |
| 393 | if i%3 == 2 { |
| 394 | repeats = 20 |
| 395 | } |
| 396 | msgs = append(msgs, fmt.Sprintf("Turn %d: ", i)+strings.Repeat("preserve the request prefix while handling varied input. ", repeats)) |
| 397 | } |
| 398 | return cacheCurveWithMessages(t, &mockDeepSeek{t: t, reasoning: longReasoning}, msgs) |
| 399 | }, |
| 400 | }, |
| 401 | { |
| 402 | name: "tool-loop", |
| 403 | run: func(t *testing.T) []int { |
| 404 | return toolLoopCurve(t, &mockDeepSeek{t: t, withTools: true, reasoning: longReasoning, toolRounds: 14}) |
| 405 | }, |
| 406 | }, |
| 407 | { |
| 408 | name: "tool-loop-no-reasoning", |
| 409 | run: func(t *testing.T) []int { |
| 410 | return toolLoopCurve(t, &mockDeepSeek{t: t, withTools: true, toolRounds: 14}) |
| 411 | }, |
| 412 | }, |
| 413 | { |
| 414 | name: "long-tool-loop", |
| 415 | run: func(t *testing.T) []int { |
| 416 | return toolLoopCurve(t, &mockDeepSeek{t: t, withTools: true, reasoning: longReasoning, toolRounds: 24}) |
| 417 | }, |
| 418 | }, |
| 419 | { |
| 420 | name: "long-tool-loop-no-reasoning", |
| 421 | run: func(t *testing.T) []int { |
| 422 | return toolLoopCurve(t, &mockDeepSeek{t: t, withTools: true, toolRounds: 24}) |
| 423 | }, |
| 424 | }, |
| 425 | } |
| 426 | |
| 427 | type result struct { |
| 428 | name string |
| 429 | rate int |
| 430 | all []int |
| 431 | } |
| 432 | var lows []result |
| 433 | for _, c := range cases { |
| 434 | rates := c.run(t) |
| 435 | rate := tailAverage(rates, 3) |
| 436 | status := "pass" |
| 437 | if rate < threshold { |
| 438 | status = "low" |
| 439 | lows = append(lows, result{name: c.name, rate: rate, all: rates}) |
| 440 | } |
| 441 | t.Logf("CACHE_GUARD_RESULT: case=%s tail_avg=%d threshold=%d status=%s rates=%v", |
| 442 | c.name, rate, threshold, status, rates) |
| 443 | } |
| 444 | |
| 445 | if len(lows) > maxLowCases { |
| 446 | var parts []string |
| 447 | for _, low := range lows { |
| 448 | parts = append(parts, fmt.Sprintf("%s=%d%%", low.name, low.rate)) |
| 449 | } |
| 450 | msg := fmt.Sprintf("%d cache guard cases are below %d%%: %s", len(lows), threshold, strings.Join(parts, ", ")) |
| 451 | t.Logf("CACHE_GUARD_WARNING: %s", msg) |
| 452 | if os.Getenv("REASONIX_CACHE_GUARD_STRICT") != "" { |
| 453 | t.Fatal(msg) |
| 454 | } |
| 455 | } |
| 456 | } |
| 457 | |
| 458 | func TestLargeToolOutputCachePrefixContract(t *testing.T) { |
| 459 | visibleSizes := make([]int, 0, 2) |
| 460 | for _, size := range []int{64 << 10, 256 << 10} { |
| 461 | visibleSizes = append(visibleSizes, verifyLargeToolOutputCacheContract(t, size)) |
| 462 | } |
| 463 | if visibleSizes[1]-visibleSizes[0] > 128 { |
| 464 | t.Fatalf("provider-visible growth tracks RawContent: 64KiB=%d 256KiB=%d", visibleSizes[0], visibleSizes[1]) |
| 465 | } |
| 466 | } |
| 467 | |
| 468 | func verifyLargeToolOutputCacheContract(t *testing.T, size int) int { |
| 469 | t.Helper() |
| 470 | callID := fmt.Sprintf("large-%d", size) |
| 471 | const rawSentinel = "UNIQUE-RAW-MIDDLE-SENTINEL" |
| 472 | raw := strings.Repeat("R", size/2) + rawSentinel + strings.Repeat("R", size/2-len(rawSentinel)) |
| 473 | bounded, notice := truncateToolOutputFor(raw, "large_result", callID) |
| 474 | if notice == "" || len(bounded) > maxToolOutputBytes { |
| 475 | t.Fatalf("%d-byte result was not bounded: content=%d notice=%q", size, len(bounded), notice) |
| 476 | } |
| 477 | canonical := []provider.Message{ |
| 478 | {Role: provider.RoleSystem, Content: systemPrompt}, |
| 479 | {Role: provider.RoleUser, Content: "produce a large result"}, |
| 480 | {Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: callID, Name: "large_result", Arguments: `{}`}}}, |
| 481 | {Role: provider.RoleTool, ToolCallID: callID, Name: "large_result", Content: bounded, RawContent: raw}, |
| 482 | } |
| 483 | first := modelInputMessages(canonical) |
| 484 | firstWire := encodeProviderMessages(t, first) |
| 485 | if bytes.Contains(mustCacheJSON(t, firstWire), []byte(rawSentinel)) { |
| 486 | t.Fatal("provider-visible request contains RawContent sentinel") |
| 487 | } |
| 488 | if got := first[len(first)-1]; got.RawContent != "" || len(got.Content) > maxToolOutputBytes { |
| 489 | t.Fatalf("provider tool result is not bounded: content=%d raw=%d", len(got.Content), len(got.RawContent)) |
| 490 | } |
| 491 | |
| 492 | secondCanonical := append(append([]provider.Message(nil), canonical...), provider.Message{Role: provider.RoleUser, Content: "continue"}) |
| 493 | second := modelInputMessages(secondCanonical) |
| 494 | secondWire := encodeProviderMessages(t, second) |
| 495 | if common := commonPrefixMsgs(firstWire, secondWire); common != len(firstWire) { |
| 496 | t.Fatalf("%d-byte result changed old provider prefix at message %d/%d", size, common, len(firstWire)) |
| 497 | } |
| 498 | |
| 499 | reg := tool.NewRegistry() |
| 500 | reg.Add(echoTool{}) |
| 501 | beforeSchemas, afterSchemas := reg.Schemas(), reg.Schemas() |
| 502 | beforeShape := CaptureShape(systemPrompt, beforeSchemas, 0) |
| 503 | afterShape := CaptureShape(systemPrompt, afterSchemas, 0) |
| 504 | if !bytes.Equal(mustCacheJSON(t, beforeSchemas), mustCacheJSON(t, afterSchemas)) { |
| 505 | t.Fatal("large tool result changed provider tool schema bytes or order") |
| 506 | } |
| 507 | if diag := CompareShape(beforeShape, afterShape, nil, nil); diag.PrefixChanged { |
| 508 | t.Fatalf("large append-only result reported PrefixChanged: %+v", diag) |
| 509 | } |
| 510 | if canonical[len(canonical)-1].RawContent != raw { |
| 511 | t.Fatal("canonical RawContent was not retained") |
| 512 | } |
| 513 | return charsOf(firstWire) |
| 514 | } |
| 515 | |
| 516 | func encodeProviderMessages(t *testing.T, msgs []provider.Message) []json.RawMessage { |
| 517 | t.Helper() |
| 518 | out := make([]json.RawMessage, len(msgs)) |
| 519 | for i, msg := range msgs { |
| 520 | out[i] = append(json.RawMessage(nil), mustCacheJSON(t, msg)...) |
| 521 | } |
| 522 | return out |
| 523 | } |
| 524 | |
| 525 | func mustCacheJSON(t *testing.T, value any) []byte { |
| 526 | t.Helper() |
| 527 | b, err := json.Marshal(value) |
| 528 | if err != nil { |
| 529 | t.Fatal(err) |
| 530 | } |
| 531 | return b |
| 532 | } |
| 533 | |
| 534 | func cacheCurve(t *testing.T, mock *mockDeepSeek, turns int) []int { |
| 535 | return cacheCurveWithMessages(t, mock, repeatedMessages(turns, 6)) |
| 536 | } |
| 537 | |
| 538 | func repeatedMessages(turns, repeats int) []string { |
| 539 | msgs := make([]string, 0, turns) |
| 540 | for i := range turns { |
| 541 | msgs = append(msgs, "Turn "+fmt.Sprint(i)+": "+strings.Repeat("please consider this requirement. ", repeats)) |
| 542 | } |
| 543 | return msgs |
| 544 | } |
| 545 | |
| 546 | func cacheCurveWithMessages(t *testing.T, mock *mockDeepSeek, messages []string) []int { |
| 547 | t.Helper() |
| 548 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 549 | defer srv.Close() |
| 550 | |
| 551 | a, sink := newAgent(t, srv.URL, mock.tools(), 0, 0) |
| 552 | for i, userMsg := range messages { |
| 553 | if err := a.Run(context.Background(), userMsg); err != nil { |
| 554 | t.Fatalf("Run %d: %v", i, err) |
| 555 | } |
| 556 | } |
| 557 | return usageRates(sink.usages) |
| 558 | } |
| 559 | |
| 560 | func toolLoopCurve(t *testing.T, mock *mockDeepSeek) []int { |
| 561 | t.Helper() |
| 562 | srv := httptest.NewServer(http.HandlerFunc(mock.handler)) |
| 563 | defer srv.Close() |
| 564 | |
| 565 | a, sink := newAgent(t, srv.URL, mock.tools(), 0, 0) |
| 566 | if err := a.Run(context.Background(), strings.Repeat("please consider this requirement. ", 6)); err != nil { |
| 567 | t.Fatalf("Run: %v", err) |
| 568 | } |
| 569 | return usageRates(sink.usages) |
| 570 | } |
| 571 | |
| 572 | func usageRates(usages []*provider.Usage) []int { |
| 573 | out := make([]int, len(usages)) |
| 574 | for i, u := range usages { |
| 575 | out[i] = hitRate(u) |
| 576 | } |
| 577 | return out |
| 578 | } |
| 579 | |
| 580 | func tailAverage(xs []int, n int) int { |
| 581 | if len(xs) == 0 { |
| 582 | return 0 |
| 583 | } |
| 584 | if n > len(xs) { |
| 585 | n = len(xs) |
| 586 | } |
| 587 | sum := 0 |
| 588 | for _, x := range xs[len(xs)-n:] { |
| 589 | sum += x |
| 590 | } |
| 591 | return sum / n |
| 592 | } |
| 593 | |
| 594 | func envInt(name string, fallback int) int { |
| 595 | raw := os.Getenv(name) |
| 596 | if raw == "" { |
| 597 | return fallback |
| 598 | } |
| 599 | n, err := strconv.Atoi(raw) |
| 600 | if err != nil || n < 0 { |
| 601 | return fallback |
| 602 | } |
| 603 | return n |
| 604 | } |
| 605 | |
| 606 | // newAgent wires a real openai.Provider at url into a real Agent. |
| 607 | func newAgent(t *testing.T, url string, reg *tool.Registry, contextWindow, recentKeep int) (*Agent, *collectSink) { |
| 608 | t.Helper() |
| 609 | prov, err := openai.New(provider.Config{ |
| 610 | Name: "deepseek", |
| 611 | BaseURL: url, |
| 612 | Model: "deepseek-reasoner", |
| 613 | APIKey: "test", |
| 614 | Extra: map[string]any{"api_key_env": "DEEPSEEK_API_KEY"}, |
| 615 | }) |
| 616 | if err != nil { |
| 617 | t.Fatalf("provider New: %v", err) |
| 618 | } |
| 619 | sink := &collectSink{} |
| 620 | a := New(prov, reg, NewSession(systemPrompt), Options{ |
| 621 | Temperature: 0, |
| 622 | ContextWindow: contextWindow, |
| 623 | RecentKeep: recentKeep, |
| 624 | }, sink) |
| 625 | return a, sink |
| 626 | } |
| 627 | |
| 628 | // request inspection helpers |
| 629 | |
| 630 | func decodeMessages(body []byte) []json.RawMessage { |
| 631 | var req struct { |
| 632 | Messages []json.RawMessage `json:"messages"` |
| 633 | } |
| 634 | _ = json.Unmarshal(body, &req) |
| 635 | return req.Messages |
| 636 | } |
| 637 | |
| 638 | func isSummarizeRequest(body []byte) bool { |
| 639 | msgs := decodeMessages(body) |
| 640 | if len(msgs) == 0 { |
| 641 | return false |
| 642 | } |
| 643 | var m struct { |
| 644 | Role string `json:"role"` |
| 645 | Content string `json:"content"` |
| 646 | } |
| 647 | _ = json.Unmarshal(msgs[len(msgs)-1], &m) |
| 648 | return m.Role == "user" && strings.Contains(m.Content, "Compact the preceding conversation prefix") |
| 649 | } |
| 650 | |
| 651 | func commonPrefixMsgs(a, b []json.RawMessage) int { |
| 652 | n := 0 |
| 653 | for n < len(a) && n < len(b) && bytes.Equal(a[n], b[n]) { |
| 654 | n++ |
| 655 | } |
| 656 | return n |
| 657 | } |
| 658 | |
| 659 | func charsOf(msgs []json.RawMessage) int { |
| 660 | total := 0 |
| 661 | for _, m := range msgs { |
| 662 | total += len(m) |
| 663 | } |
| 664 | return total |
| 665 | } |
| 666 | |
| 667 | // SSE chunk builders matching the streamResponse shape the provider parses |
| 668 | |
| 669 | type sseDelta struct { |
| 670 | Content string `json:"content,omitempty"` |
| 671 | ReasoningContent string `json:"reasoning_content,omitempty"` |
| 672 | ToolCalls []sseToolCall `json:"tool_calls,omitempty"` |
| 673 | } |
| 674 | |
| 675 | type sseToolCall struct { |
| 676 | Index int `json:"index"` |
| 677 | ID string `json:"id"` |
| 678 | Type string `json:"type"` |
| 679 | Function struct { |
| 680 | Name string `json:"name"` |
| 681 | Arguments string `json:"arguments"` |
| 682 | } `json:"function"` |
| 683 | } |
| 684 | |
| 685 | type sseChoice struct { |
| 686 | Delta sseDelta `json:"delta"` |
| 687 | FinishReason *string `json:"finish_reason"` |
| 688 | } |
| 689 | |
| 690 | type sseResp struct { |
| 691 | Choices []sseChoice `json:"choices"` |
| 692 | Usage *sseUsage `json:"usage,omitempty"` |
| 693 | } |
| 694 | |
| 695 | type sseUsage struct { |
| 696 | PromptTokens int `json:"prompt_tokens"` |
| 697 | CompletionTokens int `json:"completion_tokens"` |
| 698 | TotalTokens int `json:"total_tokens"` |
| 699 | PromptCacheHitTokens int `json:"prompt_cache_hit_tokens"` |
| 700 | PromptCacheMissTokens int `json:"prompt_cache_miss_tokens"` |
| 701 | } |
| 702 | |
| 703 | func deltaReasoning(s string) sseDelta { return sseDelta{ReasoningContent: s} } |
| 704 | func deltaText(s string) sseDelta { return sseDelta{Content: s} } |
| 705 | func deltaToolCall(idx int, name, args string) sseDelta { |
| 706 | tc := sseToolCall{Index: idx, ID: fmt.Sprintf("call_%d", idx), Type: "function"} |
| 707 | tc.Function.Name = name |
| 708 | tc.Function.Arguments = args |
| 709 | return sseDelta{ToolCalls: []sseToolCall{tc}} |
| 710 | } |
| 711 | |
| 712 | func streamChunk(d sseDelta) sseResp { return sseResp{Choices: []sseChoice{{Delta: d}}} } |
| 713 | func finishChunk(reason string) sseResp { |
| 714 | return sseResp{Choices: []sseChoice{{FinishReason: &reason}}} |
| 715 | } |
| 716 | func usageChunk(prompt, completion, hit, miss int) sseResp { |
| 717 | return sseResp{Usage: &sseUsage{ |
| 718 | PromptTokens: prompt, |
| 719 | CompletionTokens: completion, |
| 720 | TotalTokens: prompt + completion, |
| 721 | PromptCacheHitTokens: hit, |
| 722 | PromptCacheMissTokens: miss, |
| 723 | }} |
| 724 | } |
| 725 | |
| 726 | func writeSSE(w http.ResponseWriter, t *testing.T, chunks ...sseResp) { |
| 727 | t.Helper() |
| 728 | w.Header().Set("Content-Type", "text/event-stream") |
| 729 | f, ok := w.(http.Flusher) |
| 730 | if !ok { |
| 731 | t.Fatal("ResponseWriter is not a Flusher") |
| 732 | } |
| 733 | for _, c := range chunks { |
| 734 | b, _ := json.Marshal(c) |
| 735 | fmt.Fprintf(w, "data: %s\n\n", b) |
| 736 | f.Flush() |
| 737 | } |
| 738 | fmt.Fprint(w, "data: [DONE]\n\n") |
| 739 | f.Flush() |
| 740 | } |
| 741 |