fix(agent): reject interrupted summaries at EOF

Problem: compaction could report success and accept a partial digest after
the summary deadline or caller cancellation.

Root cause: providers close their channel when context ends. Selecting EOF
instead of the simultaneously ready cancellation accepted nonempty output;
the deferred ceiling classifier only corrected non-nil errors.

Fix: keep EOF validation at one result boundary and check the request context
before accepting a digest or classifying truncation. Preserve typed ceiling
and cancellation failures, usage accounting, and stable provider requests.

Verification: force EOF/cancellation overlap with a progress barrier and fake
clock, cover normal completion/truncation and charged usage, and exercise
the real Build-to-notice path. Baseline production source fails on macOS
and Linux; fixed Linux cases pass 20 repetitions. Focused race/cache tests,
full root tests, required tool/boot tests, vet, lint, make check, and release
script checks pass. A separate combined tree also validates this fix with
the post-answer finalization change from PR #14140.
This commit is contained in:
SivanCola committed 2026-10-22 13:19:02 +08:00
1 parent a708a88a75
commit 8380c9b67c
5 files changed
+184 -14

No files matched your search

@@ -17,6 +17,7 @@ type pacedDigestProvider struct {
interval time.Duration
words int
stall bool
closed chan struct{}
}
func (p pacedDigestProvider) Name() string { return "boot-paced-digest" }
@@ -25,6 +26,9 @@ func (p pacedDigestProvider) Stream(ctx context.Context, req provider.Request) (
ch := make(chan provider.Chunk)
digest := req.Summary
go func() {
if digest && p.closed != nil {
defer close(p.closed)
}
defer close(ch)
if !digest {
ch <- provider.Chunk{Type: provider.ChunkText, Text: strings.Repeat("work output line with detail. ", 400)}
@@ -53,6 +57,13 @@ func (p pacedDigestProvider) Stream(ctx context.Context, req provider.Request) (
}
func compactOnPacedProvider(t *testing.T, prov pacedDigestProvider) *noticeSink {
t.Helper()
sink := &noticeSink{}
compactOnDigestProvider(t, prov, sink)
return sink
}
func compactOnDigestProvider(t *testing.T, prov provider.Provider, sink event.Sink) {
t.Helper()
isolateConfigHome(t)
dir := robustTempDir(t)
@@ -72,7 +83,6 @@ model = "x"
context_window = 200000
`)
approveWorkspace(t, dir)
sink := &noticeSink{}
ctrl, err := Build(context.Background(), Options{Sink: sink})
if err != nil {
t.Fatalf("Build: %v", err)
@@ -82,7 +92,6 @@ context_window = 200000
_ = ctrl.Run(context.Background(), "keep going")
}
ctrl.Submit("/compact")
return sink
}
func waitNotice(t *testing.T, sink *noticeSink, codes ...string) event.Event {
@@ -136,11 +145,26 @@ func TestEffectCeilingNoticeIsItsOwnCode(t *testing.T) {
prev := agent.SummaryBounds
agent.SummaryBounds = agent.SummaryLimits{Idle: time.Second, Ceiling: 300 * time.Millisecond}
defer func() { agent.SummaryBounds = prev }()
sink := compactOnPacedProvider(t, pacedDigestProvider{interval: 30 * time.Millisecond, words: 1 << 20})
closed := make(chan struct{})
sink := &noticeSink{}
gate := event.FuncSink(func(e event.Event) {
if e.Kind == event.CompactionProgress {
select {
case <-closed:
case <-time.After(5 * time.Second):
t.Error("provider did not close after its ceiling")
}
}
sink.Emit(e)
})
compactOnDigestProvider(t, pacedDigestProvider{interval: 30 * time.Millisecond, words: 1 << 20, closed: closed}, gate)
got := waitNotice(t, sink, event.NoticeCodeCompactFailed)
if got.Detail != "summary_ceiling" {
t.Fatalf("detail = %q, want summary_ceiling", got.Detail)
}
if _, ok := sink.notice(event.NoticeCodeCompacted); ok {
t.Fatal("an interrupted digest was reported as successfully compacted")
}
}
func TestBoundWordingStatesTheBoundsTheKernelEnforces(t *testing.T) {
+2 -11
View File
@@ -686,17 +686,8 @@ func (a *contextWindow) summarizeAtMode(ctx context.Context, region []provider.M
case chunk, ok := <-ch:
idle.Reset(bounds.Idle)
if !ok {
if usage != nil && usage.FinishReason == "length" {
return "", usage, &summaryTruncation{Cap: req.MaxTokens}
}
s := strings.TrimSpace(b.String())
if s == "" {
return "", usage, fmt.Errorf("summarizer returned empty output")
}
if reused && !hasDigestHeading(s) {
return "", usage, errSummaryPrefixNotDigest
}
return s, usage, nil
s, err := summaryAtEOF(ctx, b.String(), usage, req.MaxTokens, reused)
return s, usage, err
}
switch chunk.Type {
case provider.ChunkText:
@@ -0,0 +1,28 @@
package agent
import (
"context"
"fmt"
"strings"
"reasonix/internal/contract/provider"
)
func summaryAtEOF(ctx context.Context, text string, usage *provider.Usage, maxTokens int, reused bool) (string, error) {
// Providers also close their stream on cancellation; EOF alone cannot
// distinguish a finished digest from a partial one at the request deadline.
if err := ctx.Err(); err != nil {
return "", err
}
if usage != nil && usage.FinishReason == "length" {
return "", &summaryTruncation{Cap: maxTokens}
}
s := strings.TrimSpace(text)
if s == "" {
return "", fmt.Errorf("summarizer returned empty output")
}
if reused && !hasDigestHeading(s) {
return "", errSummaryPrefixNotDigest
}
return s, nil
}
@@ -0,0 +1,116 @@
package agent
import (
"context"
"errors"
"testing"
"testing/synctest"
"time"
"reasonix/internal/contract/event"
"reasonix/internal/contract/provider"
"reasonix/internal/contract/tool"
"reasonix/internal/state/sessionstore"
)
type terminalSummaryProvider struct {
closed chan struct{}
usage *provider.Usage
}
func (*terminalSummaryProvider) Name() string { return "terminal-summary" }
func (p *terminalSummaryProvider) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Chunk, error) {
ch := make(chan provider.Chunk, 2)
ch <- provider.Chunk{Type: provider.ChunkUsage, Usage: p.usage}
ch <- provider.Chunk{Type: provider.ChunkText, Text: "## Digest\npartial"}
go func() {
<-ctx.Done()
close(ch)
close(p.closed)
}()
return ch, nil
}
func TestSummaryEOFDoesNotAcceptAnInterruptedDigest(t *testing.T) {
for _, tc := range []struct {
name string
ceiling time.Duration
cancel bool
finishReason string
want CompactionNoopReason
}{
{name: "ceiling", ceiling: time.Second, want: FailSummaryCeiling},
{name: "cancel", ceiling: time.Minute, cancel: true, want: FailCancelled},
{name: "ceiling_after_length", ceiling: time.Second, finishReason: "length", want: FailSummaryCeiling},
{name: "cancel_after_length", ceiling: time.Minute, cancel: true, finishReason: "length", want: FailCancelled},
} {
t.Run(tc.name, func(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
prev := SummaryBounds
SummaryBounds = SummaryLimits{Idle: time.Minute, Ceiling: tc.ceiling}
t.Cleanup(func() { SummaryBounds = prev })
// Both select arms are ready on release. Repetition exercises either
// choice without relying on scheduler delays to create the overlap.
for attempt := range 32 {
ctx, cancel := context.WithCancel(context.Background())
prov := &terminalSummaryProvider{closed: make(chan struct{}), usage: &provider.Usage{TotalTokens: 7, FinishReason: tc.finishReason}}
var spend []event.Event
sink := event.FuncSink(func(e event.Event) {
if e.Kind == event.Usage {
spend = append(spend, e)
}
if e.Kind != event.CompactionProgress {
return
}
if tc.cancel {
cancel()
}
select {
case <-prov.closed:
case <-time.After(5 * time.Second):
t.Fatal("provider did not close after interruption")
}
})
a := New(prov, tool.NewRegistry(), &sessionstore.Session{}, Options{}, sink)
s, usage, err := a.window().summarize(ctx, []provider.Message{{Role: provider.RoleUser, Content: "x"}}, "")
cancel()
if got := compactionFailureCode(err); got != tc.want || s != "" {
t.Fatalf("attempt %d: summary = %q, code = %q (err %v), want empty / %q", attempt, s, got, err, tc.want)
}
if usage == nil || usage.TotalTokens != 7 {
t.Fatalf("usage = %+v, want the interrupted request's spend", usage)
}
if len(spend) != 1 || spend[0].Usage.TotalTokens != 7 || spend[0].UsageSource != event.UsageSourceCompaction {
t.Fatalf("spend = %+v, want exactly one compaction usage event", spend)
}
if tc.cancel && !errors.Is(err, context.Canceled) {
t.Fatalf("err = %v, want caller cancellation", err)
}
}
})
})
}
}
func TestSummaryEOFPreservesLiveSuccessAndTruncation(t *testing.T) {
for _, tc := range []struct {
name string
prov provider.Provider
want CompactionNoopReason
}{
{name: "complete", prov: &fakeProvider{reply: "complete"}},
{name: "truncated", prov: &truncatingProvider{truncateAtOrBelow: summaryOutputMaxTokens}, want: FailSummaryTruncated},
} {
t.Run(tc.name, func(t *testing.T) {
a := New(tc.prov, tool.NewRegistry(), &sessionstore.Session{}, Options{}, event.Discard)
s, _, err := a.window().summarize(context.Background(), []provider.Message{{Role: provider.RoleUser, Content: "x"}}, "")
if got := compactionFailureCode(err); got != tc.want {
t.Fatalf("code = %q (err %v), want %q", got, err, tc.want)
}
if (s != "") != (tc.want == "") {
t.Fatalf("summary = %q, want output only on a complete stream", s)
}
})
}
}
+11
View File
@@ -0,0 +1,11 @@
## 修复
- 上下文压缩超时或取消时,不再因摘要流同时关闭而误报成功或采用未完成的摘要。关联 #14140。
---
<!-- lang:en -->
## Fixed
- Context compaction no longer reports success or accepts an unfinished digest when its summary stream closes at the same time as a timeout or cancellation. Refs #14140.