| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "slices" |
| 7 | "strings" |
| 8 | "testing" |
| 9 | "time" |
| 10 | "unicode/utf8" |
| 11 | |
| 12 | "reasonix/internal/event" |
| 13 | "reasonix/internal/provider" |
| 14 | "reasonix/internal/tool" |
| 15 | ) |
| 16 | |
| 17 | // foldableSessionOverForce builds a transcript whose bulk is assistant text, so |
| 18 | // the free prune pass cannot reclaim it and Prepare must reach the summarizer. |
| 19 | func foldableSessionOverForce(turns int) *Session { |
| 20 | big := strings.Repeat("word ", 400) |
| 21 | msgs := []provider.Message{ |
| 22 | {Role: provider.RoleSystem, Content: "sys"}, |
| 23 | {Role: provider.RoleUser, Content: "standing constraint: never change the public API"}, |
| 24 | } |
| 25 | for range turns { |
| 26 | msgs = append(msgs, |
| 27 | provider.Message{Role: provider.RoleAssistant, Content: big}, |
| 28 | provider.Message{Role: provider.RoleUser, Content: "continue"}, |
| 29 | ) |
| 30 | } |
| 31 | return &Session{Messages: msgs} |
| 32 | } |
| 33 | |
| 34 | func agentOverForce(t *testing.T, prov provider.Provider, sess *Session) *Agent { |
| 35 | t.Helper() |
| 36 | return agentOverForceWindow(t, prov, sess, 5000) |
| 37 | } |
| 38 | |
| 39 | // agentOverForceWindow sits the session above the force ratio. A folded |
| 40 | // transcript lands back under the trigger, so a blocked turn can only mean the |
| 41 | // fold itself failed. |
| 42 | func agentOverForceWindow(t *testing.T, prov provider.Provider, sess *Session, window int) *Agent { |
| 43 | t.Helper() |
| 44 | return New(prov, tool.NewRegistry(), sess, Options{ |
| 45 | ContextWindow: window, |
| 46 | CompactRatio: 0.5, |
| 47 | CompactForceRatio: 0.5, |
| 48 | RecentKeep: 2, |
| 49 | ArchiveDir: t.TempDir(), |
| 50 | }, event.Discard) |
| 51 | } |
| 52 | |
| 53 | // degradedFold reports whether a fold was committed with the mechanical digest |
| 54 | // standing in for the summary. The receipt is the host record that the |
| 55 | // projection was installed; the digest text is what the model is actually told. |
| 56 | func degradedFold(a *Agent) bool { |
| 57 | r := a.sess.compactionState.LastReceipt |
| 58 | return r != nil && r.Status == "applied" && |
| 59 | strings.Contains(latestDigest(a.sess.compactionState.Projection.Messages), "summary was unavailable") |
| 60 | } |
| 61 | |
| 62 | func prepareContext(ctx context.Context, a *Agent, trigger string) error { |
| 63 | _, err := a.contextManager().Prepare(ctx, ContextPreparePolicy{Trigger: trigger}) |
| 64 | return err |
| 65 | } |
| 66 | |
| 67 | // foldRegionOf is the region the next compaction would hand the summarizer. |
| 68 | func foldRegionOf(a *Agent) []provider.Message { |
| 69 | canonical, version := a.sess.conversation.snapshotMessagesVersion() |
| 70 | msgs, _ := a.visibleInputForFold(a.sess.compactionState, canonical, version) |
| 71 | head, start, ok := a.planFoldRegion(msgs, false, false) |
| 72 | if !ok { |
| 73 | return nil |
| 74 | } |
| 75 | _, fold, _ := a.partitionFoldForProjection(msgs[head:start]) |
| 76 | return fold |
| 77 | } |
| 78 | |
| 79 | // latestDigest returns the text of the last compaction digest in a projection. |
| 80 | func latestDigest(msgs []provider.Message) string { |
| 81 | for _, m := range slices.Backward(msgs) { |
| 82 | if isCompactionSummary(m) { |
| 83 | return m.Content |
| 84 | } |
| 85 | } |
| 86 | return "" |
| 87 | } |
| 88 | |
| 89 | // projectionTokens reports what the model would actually see. |
| 90 | func projectionTokens(a *Agent) int { |
| 91 | msgs, _ := a.sess.conversation.snapshotMessagesVersion() |
| 92 | return estimateMessagesTokens(provider.ModelMessages(modelVisibleFromProjection(a.sess.compactionState.Projection, msgs))) |
| 93 | } |
| 94 | |
| 95 | func TestSummarizerCancellationAtOverflowPropagatesWithoutFallback(t *testing.T) { |
| 96 | sess := foldableSessionOverForce(6) |
| 97 | a := agentOverForce(t, &fakeProvider{hang: true}, sess) |
| 98 | before := estimateMessagesTokens(provider.ModelMessages(sess.Messages)) |
| 99 | |
| 100 | ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) |
| 101 | defer cancel() |
| 102 | |
| 103 | if err := prepareContext(ctx, a, CompactionTriggerOverflow); !errors.Is(err, context.DeadlineExceeded) { |
| 104 | t.Fatalf("prepare = %v, want context deadline", err) |
| 105 | } |
| 106 | if after := projectionTokens(a); after != 0 { |
| 107 | t.Fatalf("cancellation installed projection tokens=%d (source=%d)", after, before) |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | // Thinking-mode providers (DeepSeek vision SKUs) may answer the summary |
| 112 | // request with reasoning_content only and an empty content block. The |
| 113 | // summarizer must surface the reasoning instead of failing with "summarizer |
| 114 | // returned empty output" and retrying forever (observed on a 2M-token session: |
| 115 | // chunked fallback reached fragment 2/14 and died on the same empty-output |
| 116 | // check). |
| 117 | func TestSummarizerReasoningOnlyIsSurfacedNotEmptied(t *testing.T) { |
| 118 | sess := foldableSessionOverForce(6) |
| 119 | a := agentOverForce(t, &fakeProvider{reasoningReply: "- kept: alpha constraint\n- kept: beta file path"}, sess) |
| 120 | before := estimateMessagesTokens(provider.ModelMessages(sess.Messages)) |
| 121 | |
| 122 | if err := prepareContext(context.Background(), a, CompactionTriggerOverflow); err != nil { |
| 123 | t.Fatalf("prepare with reasoning-only summary = %v, want applied fold", err) |
| 124 | } |
| 125 | if after := projectionTokens(a); after == 0 || after >= before { |
| 126 | t.Fatalf("reasoning-only summary installed projection tokens=%d (source=%d)", after, before) |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | // A reasoning-only reply that also opened a tool call is not a briefing: the |
| 131 | // empty-output rejection must survive, and an opened call counts even when |
| 132 | // the stream never completed it. At the ceiling that rejection blocks the turn. |
| 133 | func TestSummarizerReasoningWithToolCallStaysEmpty(t *testing.T) { |
| 134 | sess := foldableSessionOverForce(6) |
| 135 | a := agentOverForce(t, &fakeProvider{reasoningReply: "let me call a tool first", reasoningTool: true}, sess) |
| 136 | var rejected string |
| 137 | a.svc.sink = event.FuncSink(func(e event.Event) { |
| 138 | if e.Kind == event.ContextMaintenanceEvent && e.Maintenance != nil && e.Maintenance.Status == "failed" { |
| 139 | rejected = e.Maintenance.Reason |
| 140 | } |
| 141 | }) |
| 142 | |
| 143 | if err := prepareContext(context.Background(), a, CompactionTriggerOverflow); !errors.Is(err, ErrCompactionRequired) { |
| 144 | t.Fatalf("prepare = %v, want a recoverable context failure", err) |
| 145 | } |
| 146 | if !strings.Contains(rejected, "summarizer returned empty output") { |
| 147 | t.Fatalf("failed receipt reason = %q, want the empty-output rejection for reasoning with a tool call", rejected) |
| 148 | } |
| 149 | assertNoFailedSummaryRewrite(t, a) |
| 150 | } |
| 151 | |
| 152 | // The reasoning clamp cuts on rune boundaries so a CJK briefing stays valid |
| 153 | // UTF-8 for the provider request that replays the digest. |
| 154 | func TestSummarizerReasoningClampKeepsValidUTF8(t *testing.T) { |
| 155 | sess := foldableSessionOverForce(6) |
| 156 | a := agentOverForce(t, &fakeProvider{reasoningReply: strings.Repeat("上下文摘要要点。", 3000)}, sess) |
| 157 | |
| 158 | summary, _, err := a.summarize(context.Background(), sess.Messages[1:], "") |
| 159 | if err != nil { |
| 160 | t.Fatalf("summarize = %v", err) |
| 161 | } |
| 162 | if len(summary) > summaryReasoningMaxBytes || !utf8.ValidString(summary) { |
| 163 | t.Fatalf("clamped reasoning is %d bytes valid=%v, want <= %d bytes of valid UTF-8", len(summary), utf8.ValidString(summary), summaryReasoningMaxBytes) |
| 164 | } |
| 165 | } |
| 166 | |
| 167 | // truncatedRescue reports whether the last maintenance installed the lossy |
| 168 | // truncation projection instead of any digest. |
| 169 | func truncatedRescue(a *Agent) bool { |
| 170 | r := a.sess.compactionState.LastReceipt |
| 171 | return r != nil && r.Status == "applied" && r.Action == "truncate" && |
| 172 | latestDigest(a.sess.compactionState.Projection.Messages) == "" |
| 173 | } |
| 174 | |
| 175 | // Overflow recovery failure preserves the view and blocks the current attempt. |
| 176 | func TestOverflowSummarizerFailurePreservesContext(t *testing.T) { |
| 177 | sess := foldableSessionOverForce(6) |
| 178 | a := agentOverForce(t, &fakeProvider{streamErr: errors.New("provider down")}, sess) |
| 179 | before := estimateMessagesTokens(provider.ModelMessages(sess.Messages)) |
| 180 | |
| 181 | if err := prepareContext(context.Background(), a, CompactionTriggerOverflow); !errors.Is(err, ErrCompactionRequired) { |
| 182 | t.Fatalf("prepare = %v, want a recoverable context failure", err) |
| 183 | } |
| 184 | assertNoFailedSummaryRewrite(t, a) |
| 185 | if after := estimateMessagesTokens(provider.ModelMessages(a.ModelHistorySnapshot())); after != before { |
| 186 | t.Fatalf("model history changed: %d -> %d", before, after) |
| 187 | } |
| 188 | } |
| 189 | |
| 190 | // An oversized complete-prefix request fails admission and must not fabricate |
| 191 | // a summary or privately shorten its input. The failed attempt preserves the view. |
| 192 | func TestSummarizerFailureOnOversizedFoldDoesNotFabricateDigest(t *testing.T) { |
| 193 | sess := foldableSessionOverForce(120) |
| 194 | a := agentOverForceWindow(t, &fakeProvider{streamErr: errors.New("provider exploded")}, sess, 60000) |
| 195 | if tokens, budget := a.guardedSummaryInputTokens(foldRegionOf(a)), a.summaryInputBudget(""); budget <= 0 || tokens <= budget { |
| 196 | t.Fatalf("fixture fold is %d tokens against a %d budget; the shortening path is not exercised", tokens, budget) |
| 197 | } |
| 198 | |
| 199 | if err := prepareContext(context.Background(), a, CompactionTriggerOverflow); !errors.Is(err, ErrCompactionRequired) { |
| 200 | t.Fatalf("prepare = %v, want a recoverable context failure", err) |
| 201 | } |
| 202 | if degradedFold(a) || latestDigest(a.sess.compactionState.Projection.Messages) != "" { |
| 203 | t.Errorf("failed summary fabricated a digest: receipt=%+v", a.sess.compactionState.LastReceipt) |
| 204 | } |
| 205 | assertNoFailedSummaryRewrite(t, a) |
| 206 | } |
| 207 | |
| 208 | // Below the hard ceiling the turn still goes out, so a failed summary must stay |
| 209 | // a failure: the recoverable view proceeds unchanged below the hard ceiling. |
| 210 | func TestPressureBelowHardCeilingKeepsTheFailure(t *testing.T) { |
| 211 | sess := foldableSessionOverForce(6) |
| 212 | a := agentOverForce(t, &fakeProvider{streamErr: errors.New("provider down")}, sess) |
| 213 | if est, hard := a.estimatedPromptTokens(sess.Messages), a.hardInputCeiling(); est >= hard { |
| 214 | t.Fatalf("fixture estimates %d tokens against a %d ceiling; it is not below it", est, hard) |
| 215 | } |
| 216 | |
| 217 | if err := prepareContext(context.Background(), a, CompactionTriggerPressure); err != nil { |
| 218 | t.Fatalf("prepare = %v, want the turn to proceed unfolded", err) |
| 219 | } |
| 220 | if degradedFold(a) { |
| 221 | t.Error("a recoverable view was folded without a summary") |
| 222 | } |
| 223 | if r := a.sess.compactionState.LastReceipt; r == nil || (r.Status != "blocked" && r.Status != "failed") { |
| 224 | t.Errorf("receipt = %+v, want the failure recorded so the summary is not paid for twice", r) |
| 225 | } |
| 226 | } |
| 227 | |
| 228 | // The receipt recorded below the ceiling must not outlive the ceiling itself: |
| 229 | // once growing usage crosses the hard ceiling the fold is the only way out, so |
| 230 | // recovery has to run even with a standing failed receipt. If the summarizer is |
| 231 | // still down, hard pressure stops without installing a fallback projection. |
| 232 | func TestFailedSummaryReceiptRetriesAtHardCeilingWithoutFallback(t *testing.T) { |
| 233 | sess := foldableSessionOverForce(6) |
| 234 | a := agentOverForce(t, &fakeProvider{streamErr: errors.New("provider down")}, sess) |
| 235 | a.activeTurnCreatedAt.Store(42) |
| 236 | if est, hard := a.estimatedPromptTokens(sess.Messages), a.hardInputCeiling(); est >= hard { |
| 237 | t.Fatalf("fixture estimates %d tokens against a %d ceiling; it is not below it", est, hard) |
| 238 | } |
| 239 | |
| 240 | // The failed pressure summary is recorded, not fatal: the turn goes out. |
| 241 | if err := prepareContext(context.Background(), a, CompactionTriggerPressure); err != nil { |
| 242 | t.Fatalf("prepare = %v, want the turn to proceed unfolded", err) |
| 243 | } |
| 244 | if r := a.sess.compactionState.LastReceipt; r == nil || (r.Status != "blocked" && r.Status != "failed") { |
| 245 | t.Fatalf("receipt = %+v, want the failure recorded", r) |
| 246 | } |
| 247 | |
| 248 | // The session keeps growing past the ceiling while the receipt stands. |
| 249 | big := strings.Repeat("word ", 400) |
| 250 | for range 4 { |
| 251 | sess.Add(provider.Message{Role: provider.RoleAssistant, Content: big}) |
| 252 | sess.Add(provider.Message{Role: provider.RoleUser, Content: "continue"}) |
| 253 | } |
| 254 | if est, hard := a.estimatedPromptTokens(sess.Messages), a.hardInputCeiling(); est < hard { |
| 255 | t.Fatalf("grown fixture estimates %d tokens against a %d ceiling; it is not past it", est, hard) |
| 256 | } |
| 257 | |
| 258 | if err := prepareContext(context.Background(), a, CompactionTriggerPressure); !errors.Is(err, ErrCompactionRequired) { |
| 259 | t.Fatalf("over-ceiling prepare = %v, want a recoverable context failure", err) |
| 260 | } |
| 261 | if degradedFold(a) { |
| 262 | t.Fatal("hard-ceiling failure installed a mechanical digest") |
| 263 | } |
| 264 | assertNoFailedSummaryRewrite(t, a) |
| 265 | } |
| 266 | |
| 267 | func assertNoFailedSummaryRewrite(t *testing.T, a *Agent) { |
| 268 | t.Helper() |
| 269 | if a.currentProjectionVersion() != 0 || truncatedRescue(a) || latestDigest(a.sess.compactionState.Projection.Messages) != "" { |
| 270 | t.Fatal("failed summary changed the model projection") |
| 271 | } |
| 272 | } |
| 273 | |
| 274 | // A ceiling recovery that lands under the fold trigger clears the stuck |
| 275 | // latch: the next pressure round above the trigger must compact again instead |
| 276 | // of coasting back to the physical ceiling. |
| 277 | func TestCeilingRecoveryClearsStuckLatch(t *testing.T) { |
| 278 | sess := foldableSessionOverForce(10) |
| 279 | a := agentOverForce(t, &fakeProvider{reply: "digest"}, sess) |
| 280 | if est, hard := a.estimatedPromptTokens(sess.Messages), a.hardInputCeiling(); est < hard { |
| 281 | t.Fatalf("fixture estimates %d tokens against a %d ceiling; it is not past it", est, hard) |
| 282 | } |
| 283 | a.sess.compaction.stuck = true |
| 284 | |
| 285 | if err := prepareContext(context.Background(), a, CompactionTriggerPressure); err != nil { |
| 286 | t.Fatalf("ceiling recovery = %v, want a fold despite the latch", err) |
| 287 | } |
| 288 | if a.sess.compaction.stuck { |
| 289 | t.Fatal("stale stuck latch survived a recovery that landed under the trigger") |
| 290 | } |
| 291 | version := a.currentProjectionVersion() |
| 292 | if version == 0 { |
| 293 | t.Fatal("recovery installed no projection") |
| 294 | } |
| 295 | |
| 296 | // One big append jumps from under the trigger straight past it, still |
| 297 | // below the ceiling: the pressure round must compact, not coast. |
| 298 | big := strings.Repeat("word ", 400) |
| 299 | for range 20 { |
| 300 | sess.Add(provider.Message{Role: provider.RoleAssistant, Content: big}) |
| 301 | sess.Add(provider.Message{Role: provider.RoleUser, Content: "continue"}) |
| 302 | current := a.contextManager().currentPrepared() |
| 303 | if current.InputTokens >= a.compactTrigger() && current.InputTokens < a.hardInputCeiling() { |
| 304 | break |
| 305 | } |
| 306 | } |
| 307 | if est, fold := a.contextManager().currentPrepared().InputTokens, a.compactTrigger(); est < fold { |
| 308 | t.Fatalf("grown fixture estimates %d tokens, below fold %d", est, fold) |
| 309 | } |
| 310 | if err := prepareContext(context.Background(), a, CompactionTriggerPressure); err != nil { |
| 311 | t.Fatalf("pressure round above the trigger: %v", err) |
| 312 | } |
| 313 | if a.currentProjectionVersion() == version { |
| 314 | t.Fatal("pressure round above the trigger did not compact; the stale latch suppressed it") |
| 315 | } |
| 316 | } |
| 317 | |
| 318 | // Cancellation is the user's decision, not a summarizer failure: it must keep |
| 319 | // its error and leave the projection alone. |
| 320 | func TestCallerCancellationDoesNotDegrade(t *testing.T) { |
| 321 | sess := foldableSessionOverForce(6) |
| 322 | a := agentOverForce(t, &fakeProvider{hang: true}, sess) |
| 323 | |
| 324 | ctx, cancel := context.WithCancel(context.Background()) |
| 325 | cancel() |
| 326 | |
| 327 | if err := prepareContext(ctx, a, CompactionTriggerOverflow); err == nil { |
| 328 | t.Fatal("cancelled prepare reported success") |
| 329 | } |
| 330 | if degradedFold(a) { |
| 331 | t.Error("cancellation installed a degraded fold; it should change nothing") |
| 332 | } |
| 333 | } |
| 334 |