返回 DeepSeek-Reasonix
session_extract.go
根目录 / internal / agent / session_extract.go
1 package agent
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "math"
8 "slices"
9 "strings"
10 "sync"
11
12 "reasonix/internal/provider"
13 )
14
15 // Chunked session recovery summarizes an over-length transcript in
16 // exponentially decaying fragments and tree-reduces their digests.
17 // It powers the #9082 in-place compaction fallback.
18
19 const (
20 // Chunk byte budgets from newest to oldest (design doc 2026-08-18:
21 // 会话引用与压缩机制分析). Chunks older than the listed sizes stay at
22 // the last size - coarse folding is fine for ancient history.
23 extractChunkNewestBytes = 64 << 10
24 extractChunkNextBytes = 128 << 10
25 extractChunkThirdBytes = 256 << 10
26 extractChunkFourthBytes = 512 << 10
27 extractChunkOldestBytes = 768 << 10
28 // Adjacent chunks share this many bytes near their boundary so a fact
29 // spanning the cut is not lost between digests.
30 extractChunkOverlapBytes = 16 << 10
31 minMergeInputTokens = 400
32 // A chunked recovery owns the compaction lock and can issue multiple paid
33 // requests. Keep the exceptional path finite even when a provider repeatedly
34 // truncates a digest without making it smaller.
35 maxChunkedSummaryCalls = 64
36 maxChunkedMergeDepth = 8
37 )
38
39 const extractFragmentInstructionTmpl = `This is fragment %d/%d of an over-length session being recovered after its context exceeded the model window. Compact the preceding fragment into a durable briefing under these exact headings, omitting a heading only if it has no content:
40
41 ## Standing facts & constraints
42 ## Goal
43 ## Decisions & rationale
44 ## Files & code
45 ## Commands & outcomes
46 ## Errors & fixes
47 ## Pending & next step
48
49 Rules: be terse - bullet points and fragments, not prose. Preserve identifiers, paths, and numbers exactly. This fragment sits earlier in the session than newer ones will, so capture everything this fragment establishes even if it may be refined later. Do NOT invent anything not present in the messages; if something is unknown, leave it out rather than guessing. Output only the structured Markdown briefing. Do not call tools. Do not output reasoning.`
50
51 const extractMergeInstruction = `The following are sequential fragment briefings of one over-length session, ordered oldest to newest. Merge them into the session's final resume briefing under the same exact headings (## Standing facts & constraints / ## Goal / ## Decisions & rationale / ## Files & code / ## Commands & outcomes / ## Errors & fixes / ## Pending & next step). Later fragments supersede earlier ones - keep the final state of every fact, drop superseded entries, and preserve identifiers, paths, and numbers exactly. Output only the structured Markdown briefing. Do not call tools. Do not output reasoning.`
52
53 type extractMessageSpan struct{ lo, hi int }
54
55 // extractMessageUnits returns replay-safe units. An assistant tool-call message
56 // and all of its contiguous results are indivisible because every provider
57 // adapter sanitizes that pairing immediately before serialization.
58 func extractMessageUnits(msgs []provider.Message) []extractMessageSpan {
59 units := make([]extractMessageSpan, 0, len(msgs))
60 for i := 0; i < len(msgs); {
61 j := i + 1
62 switch {
63 case msgs[i].Role == provider.RoleAssistant && len(msgs[i].ToolCalls) > 0:
64 for j < len(msgs) && msgs[j].Role == provider.RoleTool {
65 j++
66 }
67 case msgs[i].Role == provider.RoleTool:
68 // Keep malformed/orphan result runs together too. Sanitization may drop
69 // them, but chunking must never create additional orphan boundaries.
70 for j < len(msgs) && msgs[j].Role == provider.RoleTool {
71 j++
72 }
73 }
74 units = append(units, extractMessageSpan{lo: i, hi: j})
75 i = j
76 }
77 return units
78 }
79
80 func extractUnitWireBytes(msgs []provider.Message, unit extractMessageSpan, policy provider.SharedWindowInputPolicy) int {
81 total := 0
82 for _, msg := range msgs[unit.lo:unit.hi] {
83 total += messageWireBytes(msg, policy)
84 }
85 return total
86 }
87
88 func extractInstructionWithFocus(base, instructions string) string {
89 if strings.TrimSpace(instructions) == "" {
90 return base
91 }
92 return base + "\n\nAdditional focus for this compaction (prioritize keeping this):\n" + strings.TrimSpace(instructions)
93 }
94
95 func extractFragmentInstruction(index, total int, instructions string) string {
96 return extractInstructionWithFocus(fmt.Sprintf(extractFragmentInstructionTmpl, index, total), instructions)
97 }
98
99 func extractMergeInstructionWithFocus(instructions string) string {
100 return extractInstructionWithFocus(extractMergeInstruction, instructions)
101 }
102
103 type chunkedSummaryRun struct {
104 a *Agent
105 calls int
106 usage *provider.Usage
107 }
108
109 func newChunkedSummaryRun(a *Agent) *chunkedSummaryRun {
110 return &chunkedSummaryRun{a: a}
111 }
112
113 func (r *chunkedSummaryRun) requireCalls(required int) error {
114 if required < 0 {
115 return fmt.Errorf("invalid chunked summary call reservation (%d)", required)
116 }
117 if r.calls+required > maxChunkedSummaryCalls {
118 return fmt.Errorf("chunked summary call budget exhausted (%d): %d used, %d required", maxChunkedSummaryCalls, r.calls, required)
119 }
120 return nil
121 }
122
123 func (r *chunkedSummaryRun) summarize(ctx context.Context, fold []provider.Message, instructions string, reserveAfter int) (foldSummary, error) {
124 if err := ctx.Err(); err != nil {
125 return foldSummary{}, err
126 }
127 if err := r.requireCalls(1 + reserveAfter); err != nil {
128 return foldSummary{}, err
129 }
130 r.calls++
131 res, err := r.a.foldToSummary(ctx, fold, instructions)
132 r.usage = mergeSamplingUsage(r.usage, res.Usage)
133 return res, err
134 }
135
136 // extractChunkSizes returns the per-chunk byte budgets from newest to oldest.
137 func extractChunkSizes() []int {
138 return []int{
139 extractChunkNewestBytes,
140 extractChunkNextBytes,
141 extractChunkThirdBytes,
142 extractChunkFourthBytes,
143 extractChunkOldestBytes,
144 }
145 }
146
147 func minimumChunkedSummaryCalls(chunkCount int) int {
148 if chunkCount <= 0 {
149 return 0
150 }
151 if chunkCount == 1 {
152 return 1
153 }
154 return chunkCount + 1
155 }
156
157 // messageWireBytes approximates the provider-visible transcript footprint of
158 // one message. chunkedFoldSummary first removes local-only/raw fields through
159 // modelInputMessages, so local storage metadata cannot distort boundaries.
160 func messageWireBytes(msg provider.Message, policy provider.SharedWindowInputPolicy) int {
161 chars, _, _ := requestCalibrationTextShape(provider.Request{Messages: []provider.Message{msg}}, policy)
162 n := int(chars) + len(msg.ReasoningID) + len(msg.ReasoningStatus) + len(msg.ReasoningSignature)
163 for _, tc := range msg.ToolCalls {
164 n += len(tc.ThoughtSignature)
165 }
166 for _, img := range msg.Images {
167 n += len(img)
168 }
169 return n
170 }
171
172 // splitExtractChunks splits messages newest-tail-first into chunks following
173 // the exponential size table, with adjacent chunks sharing `overlap` bytes
174 // around their boundary. Boundaries never split a replay-safe message unit:
175 // tool calls and their contiguous results stay together even when that
176 // overshoots the budget. Returns
177 // chunks oldest-first. A transcript smaller than the newest chunk yields a
178 // single chunk covering everything.
179 func splitExtractChunks(msgs []provider.Message, overlap int, policy provider.SharedWindowInputPolicy) [][]provider.Message {
180 if len(msgs) == 0 {
181 return nil
182 }
183 units := extractMessageUnits(msgs)
184 sizes := extractChunkSizes()
185 var spans []extractMessageSpan // unit indexes, newest -> oldest
186 end := len(units)
187 for i := 0; end > 0; i++ {
188 size := sizes[min(i, len(sizes)-1)]
189 if i > 0 {
190 size -= overlap // the shared boundary region is counted by the newer chunk
191 }
192 lo := end
193 acc := 0
194 for lo > 0 && acc < size {
195 lo--
196 acc += extractUnitWireBytes(msgs, units[lo], policy)
197 }
198 spans = append(spans, extractMessageSpan{lo: lo, hi: end})
199 end = lo
200 }
201 // Widen every older chunk's right edge into its newer neighbor's head so
202 // the shared overlap bytes are summarized twice, keeping boundary facts
203 // in both digests.
204 for j := 1; j < len(spans); j++ {
205 hi := spans[j].hi // the older chunk ends where the newer one begins
206 acc := 0
207 for hi < spans[j-1].hi && acc < overlap {
208 acc += extractUnitWireBytes(msgs, units[hi], policy)
209 hi++
210 }
211 spans[j].hi = hi
212 }
213 chunks := make([][]provider.Message, 0, len(spans))
214 for _, current := range slices.Backward(spans) { // oldest first
215 lo := units[current.lo].lo
216 hi := units[current.hi-1].hi
217 chunks = append(chunks, msgs[lo:hi])
218 }
219 return chunks
220 }
221
222 // chunkedFoldSummary summarizes a fold too large for one summarizer request:
223 // byte-budgeted fragments (exponentially decaying, newest finest), each
224 // summarized with the resilient half-split retry, and the digests merged via
225 // tree-reduce. It is the compaction fallback for over-length folds (#9082
226 // #9572 follow-up): the projection still installs in the same session, so
227 // work continues in place. progress, when non-nil, reports (chunks
228 // summarized, total chunks); the total grows when a fragment splits.
229 func (a *Agent) chunkedFoldSummary(ctx context.Context, fold []provider.Message, instructions string, progress func(done, total int)) (result foldSummary, err error) {
230 fold = modelInputMessages(fold)
231 result = foldSummary{
232 Mode: CompactionModeChunked,
233 FoldTokens: summaryInputTokens(fold),
234 InputMode: SummaryInputChunked,
235 }
236 if len(fold) == 0 {
237 return result, fmt.Errorf("fold is empty")
238 }
239 run := newChunkedSummaryRun(a)
240 defer func() {
241 result.Usage = run.usage
242 result.Spans = run.calls
243 }()
244 chunks := splitExtractChunks(fold, extractChunkOverlapBytes, sharedWindowInputPolicyOf(a.svc.prov))
245 if len(chunks) == 0 {
246 return result, fmt.Errorf("fold is empty")
247 }
248 text, err := a.summarizeExtractChunks(ctx, chunks, instructions, progress, run)
249 if err != nil {
250 return result, err
251 }
252 result.Text = text
253 return result, nil
254 }
255
256 func (a *Agent) summarizeExtractChunks(ctx context.Context, chunks [][]provider.Message, instructions string, progress func(done, total int), run *chunkedSummaryRun) (string, error) {
257 if err := ctx.Err(); err != nil {
258 return "", err
259 }
260 if len(chunks) == 0 {
261 return "", fmt.Errorf("no extract chunks to summarize")
262 }
263 if err := run.requireCalls(minimumChunkedSummaryCalls(len(chunks))); err != nil {
264 return "", err
265 }
266 report := orNoopProgress(progress)
267 // Fragments may split in half on summarizer failure (see
268 // extractFragmentResilient), so the progress total grows as splits happen.
269 var progressMu sync.Mutex
270 done, total := 0, len(chunks)
271 advance := func(grown bool) {
272 progressMu.Lock()
273 defer progressMu.Unlock()
274 if grown {
275 total++
276 } else {
277 done++
278 }
279 report(done, total)
280 }
281 parts := make([]string, 0, len(chunks))
282 mergeInstructions := extractMergeInstructionWithFocus(instructions)
283 for i, chunk := range chunks {
284 fragInstructions := extractFragmentInstruction(i+1, len(chunks), instructions)
285 reserveAfter := len(chunks) - i - 1
286 if len(chunks) > 1 {
287 reserveAfter++
288 }
289 res, err := a.extractFragmentResilient(ctx, chunk, fragInstructions, mergeInstructions, advance, run, reserveAfter)
290 if err != nil {
291 return "", fmt.Errorf("fragment %d/%d: %w", i+1, len(chunks), err)
292 }
293 parts = append(parts, res)
294 }
295 text, err := a.mergeFragmentsWithRun(ctx, parts, mergeInstructions, run, 0)
296 if err != nil {
297 return "", err
298 }
299 return text, nil
300 }
301
302 // extractFragmentResilient summarizes one extract fragment, splitting it in
303 // half and extracting each half when the fragment cannot be summarized whole:
304 // a very large fragment makes the summarizer output run into the provider's
305 // output-token limit (the same failure that blocks in-place compaction on
306 // over-length sessions), and on small-window models the fragment itself can
307 // overflow the input window. Both are fixed by smaller fragments, so the
308 // halves are extracted and their digests merged. Replay-safe units and the
309 // shared call budget bound the recovery. report(true) grows the progress total
310 // (one fragment became two); report(false) marks one leaf fragment summarized.
311 func (a *Agent) extractFragmentResilient(ctx context.Context, chunk []provider.Message, instructions, mergeInstructions string, report func(grown bool), run *chunkedSummaryRun, reserveAfter int) (string, error) {
312 res, err := run.summarize(ctx, chunk, instructions, reserveAfter)
313 if err == nil {
314 return strings.TrimSpace(res.Text), nil
315 }
316 leftChunk, rightChunk, splittable := splitExtractFragment(chunk)
317 if !summarySizeFailure(err) || !splittable {
318 return "", err
319 }
320 report(true)
321 left, err := a.extractFragmentResilient(ctx, leftChunk, instructions, mergeInstructions, report, run, reserveAfter+2)
322 if err != nil {
323 return "", err
324 }
325 right, err := a.extractFragmentResilient(ctx, rightChunk, instructions, mergeInstructions, report, run, reserveAfter+1)
326 if err != nil {
327 return "", err
328 }
329 merged, err := a.mergeFragmentsWithRun(ctx, []string{left, right}, mergeInstructions, run, reserveAfter)
330 if err != nil {
331 return "", fmt.Errorf("merge split fragments: %w", err)
332 }
333 return merged, nil
334 }
335
336 // summarySizeFailure reports a failure that a smaller summarizer input fixes:
337 // output truncation, local admission, or the provider's own overflow reply.
338 func summarySizeFailure(err error) bool {
339 return errors.Is(err, errSummaryOutputTruncated) || errors.Is(err, ErrCompactionRequired) ||
340 provider.AsContextLimitError(err) != nil
341 }
342
343 func splitExtractFragment(chunk []provider.Message) (left, right []provider.Message, ok bool) {
344 units := extractMessageUnits(chunk)
345 if len(units) < 2 {
346 return nil, nil, false
347 }
348 boundary := units[len(units)/2].lo
349 return chunk[:boundary], chunk[boundary:], true
350 }
351
352 // mergeInputBudget is the merge-request input ceiling in tokens: half of the
353 // safe summarizer input, leaving room for the merge instruction and the
354 // digest output inside the same request. An unknown window disables the
355 // pre-splitting (the request fails into the provider's own error then).
356 func (a *Agent) mergeInputBudget() int {
357 window := a.effectiveContextWindow()
358 if window <= 0 {
359 return math.MaxInt
360 }
361 return max(minMergeInputTokens, (window-a.summaryOutputBudget()-summaryPlanReserve(window))/2)
362 }
363
364 // mergeGroup merges one group of fragment briefings. A group that cannot be
365 // summarized whole (output truncation, input overflow) splits in half and
366 // recurses — the briefings are already in hand, so the merge must not fail
367 // with them discarded (#9082 follow-up).
368 func (a *Agent) mergeGroup(ctx context.Context, group []string, instructions string, run *chunkedSummaryRun, depth, reserveAfter int) (string, error) {
369 merged, err := run.summarize(ctx, mergeDigestMessages(group), instructions, reserveAfter)
370 if err == nil {
371 return strings.TrimSpace(merged.Text), nil
372 }
373 mergeErr := err
374 if !summarySizeFailure(err) || len(group) < 2 {
375 return "", err
376 }
377 if depth >= maxChunkedMergeDepth {
378 return "", fmt.Errorf("merge recovery depth exhausted (%d): %w", maxChunkedMergeDepth, err)
379 }
380 before := estimateMessagesTokens(mergeDigestMessages(group))
381 mid := len(group) / 2
382 left, err := a.mergeGroup(ctx, group[:mid], instructions, run, depth+1, reserveAfter+2)
383 if err != nil {
384 return "", err
385 }
386 right, err := a.mergeGroup(ctx, group[mid:], instructions, run, depth+1, reserveAfter+1)
387 if err != nil {
388 return "", err
389 }
390 next := []string{left, right}
391 after := estimateMessagesTokens(mergeDigestMessages(next))
392 if after >= before {
393 return "", fmt.Errorf("merge recovery made no progress (%d >= %d tokens): %w", after, before, mergeErr)
394 }
395 return a.mergeGroup(ctx, next, instructions, run, depth+1, reserveAfter)
396 }
397
398 // mergeFragments merges fragment briefings into one final briefing. When the
399 // whole set would overflow the merge request (many fragments, or a small
400 // provider window), it is merged pairwise first — tree-reduce, so the merge
401 // never fails with every fragment briefing already in hand.
402 func (a *Agent) mergeFragments(ctx context.Context, parts []string) (string, error) {
403 run := newChunkedSummaryRun(a)
404 return a.mergeFragmentsWithRun(ctx, parts, extractMergeInstruction, run, 0)
405 }
406
407 func (a *Agent) mergeFragmentsWithRun(ctx context.Context, parts []string, instructions string, run *chunkedSummaryRun, reserveAfter int) (string, error) {
408 if len(parts) == 0 {
409 return "", fmt.Errorf("no fragment briefings to merge")
410 }
411 parts = append([]string(nil), parts...)
412 for len(parts) > 1 && estimateMessagesTokens(mergeDigestMessages(parts)) > a.mergeInputBudget() {
413 pairCalls := len(parts) / 2
414 futureMerge := 0
415 if (len(parts)+1)/2 > 1 {
416 futureMerge = 1
417 }
418 if err := run.requireCalls(pairCalls + futureMerge + reserveAfter); err != nil {
419 return "", err
420 }
421 var next []string
422 pairIndex := 0
423 for i := 0; i < len(parts); i += 2 {
424 group := parts[i:min(i+2, len(parts))]
425 if len(group) == 1 {
426 // Odd tail: carried into the next round unchanged.
427 next = append(next, group[0])
428 continue
429 }
430 pairIndex++
431 pairReserve := pairCalls - pairIndex + futureMerge + reserveAfter
432 merged, err := a.mergeGroup(ctx, group, instructions, run, 0, pairReserve)
433 if err != nil {
434 return "", err
435 }
436 next = append(next, merged)
437 }
438 if len(next) >= len(parts) {
439 break // defensive: cannot shrink further; try the plain merge
440 }
441 parts = next
442 }
443 if len(parts) == 1 {
444 return parts[0], nil
445 }
446 merged, err := a.mergeGroup(ctx, parts, instructions, run, 0, reserveAfter)
447 if err != nil {
448 return "", fmt.Errorf("merge: %w", err)
449 }
450 return merged, nil
451 }
452
453 // mergeDigestMessages builds the merge request body: one user message per
454 // fragment briefing, oldest first, so the summarizer sees the timeline.
455 func mergeDigestMessages(parts []string) []provider.Message {
456 msgs := make([]provider.Message, 0, len(parts)+1)
457 msgs = append(msgs, provider.Message{
458 Role: provider.RoleUser,
459 Content: fmt.Sprintf("Session fragment briefings (%d fragments, oldest to newest):", len(parts)),
460 })
461 for i, part := range parts {
462 msgs = append(msgs, provider.Message{
463 Role: provider.RoleUser,
464 Content: fmt.Sprintf("<fragment index=%d>\n%s\n</fragment>", i+1, part),
465 })
466 }
467 return msgs
468 }
469
470 func orNoopProgress(progress func(done, total int)) func(done, total int) {
471 if progress != nil {
472 return progress
473 }
474 return func(done, total int) {}
475 }
476
476 lines GO