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