Merge pull request #8031 from esengine/feat/segment-digests

feat(agent): carry digests verbatim across incremental folds

internal/agent/compact.go

@@ -34,6 +34,7 @@ const (
 	fallbackTokPerChar         = 0.25  // ~4 chars/token, used before any usage is available to calibrate
 	maxPinnedFirstUserTokens   = 1500  // ceiling on pinning the first user turn verbatim; larger first turns (pasted content) stay foldable
 	pinnedFirstUserWindowFrac  = 0.15  // and never pin a first turn worth more than this fraction of the window
+	maxCarriedDigestTokens     = 12000 // ceiling on digests carried verbatim across folds before one consolidating fold merges them
 	maxEarlyUserTurns          = 3     // small user turns hoisted verbatim ahead of the digest; position-fixed (the first N of the fold region, never "the latest N") so the projection prefix stays byte-stable
 )
 

internal/agent/compact_incremental_test.go

@@ -18,8 +18,11 @@ func foldAgent(t *testing.T, prov provider.Provider, incremental bool) (*Agent,
 	t.Helper()
 	sess := NewSession("sys")
 	sess.Add(provider.Message{Role: provider.RoleUser, Content: "task"})
+	// Sized so one generation of work outgrows the verbatim tail without a large
+	// fixture: these tests assert what a fold reads, not how much, and
+	// internal/agent already runs closest to the Windows CI timeout.
 	a := New(prov, tool.NewRegistry(), sess, Options{
-		ContextWindow: 128_000,
+		ContextWindow: 11_000,
 		RecentKeep:    2,
 		SessionPath:   t.TempDir() + "/session.jsonl",
 		ArchiveDir:    t.TempDir(),
@@ -36,11 +39,11 @@ func foldArmSet(incremental bool) ablation.Set {
 }
 
 func addWork(sess *Session, gen int) {
-	for i := range 8 {
+	for i := range 6 {
 		id := fmt.Sprintf("g%dc%d", gen, i)
 		sess.Add(provider.Message{Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: id, Name: "read_file", Arguments: "{}"}}})
 		sess.Add(provider.Message{Role: provider.RoleTool, ToolCallID: id, Name: "read_file",
-			Content: strings.Repeat(fmt.Sprintf("gen %d line %d\n", gen, i), 1200)})
+			Content: strings.Repeat(fmt.Sprintf("gen %d line %d\n", gen, i), 400)})
 	}
 }
 
@@ -83,18 +86,94 @@ func TestFullFoldRereadsTheOldestHistoryEveryTime(t *testing.T) {
 	}
 }
 
-func TestIncrementalFoldFeedsThePriorDigestInsteadOfRereading(t *testing.T) {
+func TestIncrementalFoldSummarizesOnlyTheNewWork(t *testing.T) {
+	// Re-summarizing a digest is the one step that can drop a fact for good, so
+	// an incremental fold sends the summarizer neither the old raw history nor
+	// the digest that replaced it — only what is new.
 	prov := &countingProvider{reply: "digest"}
 	a, sess := foldAgent(t, prov, true)
 	var last string
 	for gen := range 3 {
 		last = foldInput(t, a, sess, prov, gen)
 	}
-	if !strings.Contains(last, summaryTagOpen) {
-		t.Fatalf("an incremental fold must feed the prior digest back in:\n%.200q", last)
+	if strings.Contains(last, summaryTagOpen) {
+		t.Fatalf("a carried digest must not be re-summarized:\n%.200q", last)
 	}
 	if strings.Contains(last, "gen 0 line 0") {
-		t.Fatal("an incremental fold must not re-read history the prior digest already covers")
+		t.Fatal("an incremental fold must not re-read history a prior digest already covers")
+	}
+	if !strings.Contains(last, "gen 2 line 0") {
+		t.Fatal("the fold must still summarize the newest work")
+	}
+}
+
+func TestCarriedDigestsSurviveVerbatimAndInOrder(t *testing.T) {
+	prov := &countingProvider{reply: "digest"}
+	a, sess := foldAgent(t, prov, true)
+	var prev []string
+	for gen := range 3 {
+		foldInput(t, a, sess, prov, gen)
+		got := projectionDigests(a)
+		if len(got) != gen+1 {
+			t.Fatalf("after fold %d the projection carries %d digests, want %d", gen+1, len(got), gen+1)
+		}
+		for i, was := range prev {
+			if got[i] != was {
+				t.Fatalf("fold %d rewrote carried digest %d:\n was %.80q\n now %.80q", gen+1, i+1, was, got[i])
+			}
+		}
+		prev = got
+	}
+}
+
+func projectionDigests(a *Agent) []string {
+	var out []string
+	for _, m := range a.compactionState.Projection.Messages {
+		if isCompactionSummary(m) {
+			out = append(out, m.Content)
+		}
+	}
+	return out
+}
+
+// The reason to carry digests instead of merging them is not only fidelity: the
+// bytes ahead of the newest digest stop changing, so a fold appends to the
+// prefix instead of rewriting it.
+func TestCarriedDigestsKeepTheProjectionPrefixStable(t *testing.T) {
+	prov := &countingProvider{reply: "digest"}
+	a, sess := foldAgent(t, prov, true)
+	foldInput(t, a, sess, prov, 0)
+	before := renderTranscript(a.compactionState.Projection.Messages[:digestEnd(a)])
+	foldInput(t, a, sess, prov, 1)
+	after := renderTranscript(a.compactionState.Projection.Messages[:digestEnd(a)-1])
+	if !strings.HasPrefix(after, before) {
+		t.Fatalf("a fold rewrote the projection prefix instead of appending to it:\nbefore %.120q\nafter  %.120q", before, after)
+	}
+}
+
+// digestEnd is one past the last carried digest in the current projection.
+func digestEnd(a *Agent) int {
+	end := 0
+	for i, m := range a.compactionState.Projection.Messages {
+		if isCompactionSummary(m) {
+			end = i + 1
+		}
+	}
+	return end
+}
+
+func TestCarriedDigestsConsolidateWhenTheyOutgrowTheirBudget(t *testing.T) {
+	prov := &countingProvider{reply: strings.Repeat("carried digest body ", 400)} // ~8k tokens est. each
+	a, sess := foldAgent(t, prov, true)
+	var last string
+	for gen := range 3 {
+		last = foldInput(t, a, sess, prov, gen)
+	}
+	if !strings.Contains(last, summaryTagOpen) {
+		t.Fatal("digests past the carry budget must be merged by one consolidating fold")
+	}
+	if n := len(projectionDigests(a)); n != 1 {
+		t.Fatalf("projection carries %d digests after consolidation, want 1", n)
 	}
 }
 

internal/agent/compact_partition_test.go

@@ -15,7 +15,8 @@ func bigTurn() string { return strings.Repeat("paste ", 4000) }
 // every provider-visible message in the region lands in exactly one group.
 func partitionCoversRegion(t *testing.T, a *Agent, region []provider.Message) (early, kept, fold []provider.Message) {
 	t.Helper()
-	early, kept, fold = a.partitionFoldForProjection(region)
+	early, carried, kept, fold := a.partitionFoldForProjection(region)
+	kept = append(carried, kept...)
 	seen := map[string]int{}
 	for _, group := range [][]provider.Message{early, kept, fold} {
 		for _, m := range group {

internal/agent/compact_projection.go

@@ -409,7 +409,7 @@ func (a *Agent) compactToProjection(ctx context.Context, trigger, instructions s
 		return CompactionNoop, nil
 	}
 	region := msgs[head:start]
-	early, kept, fold := a.partitionFoldForProjection(region)
+	early, carried, kept, fold := a.partitionFoldForProjection(region)
 	if len(fold) == 0 {
 		return CompactionNoop, nil
 	}
@@ -468,9 +468,10 @@ func (a *Agent) compactToProjection(ctx context.Context, trigger, instructions s
 		return CompactionNoop, err
 	}
 
-	projMsgs := make([]provider.Message, 0, head+len(early)+1+len(kept)+len(msgs)-start)
+	projMsgs := make([]provider.Message, 0, head+len(early)+len(carried)+1+len(kept)+len(msgs)-start)
 	projMsgs = append(projMsgs, msgs[:head]...)
 	projMsgs = append(projMsgs, early...)
+	projMsgs = append(projMsgs, carried...)
 	projMsgs = append(projMsgs, formatSummaryMessage(summary))
 	projMsgs = append(projMsgs, kept...)
 	projMsgs = append(projMsgs, msgs[start:]...)
@@ -555,21 +556,51 @@ func (a *Agent) foldSource(canonical []provider.Message) []provider.Message {
 // the remainder that folds (prior digests included, so a merge yields one
 // summary). The groups partition the region — one pass decides each message
 // once, so no turn can fall between a hoist rule and a fold rule that disagree.
-func (a *Agent) partitionFoldForProjection(region []provider.Message) (early, kept, fold []provider.Message) {
+func (a *Agent) partitionFoldForProjection(region []provider.Message) (early, carried, kept, fold []provider.Message) {
 	policyKeep := keepIndexes(region, a.keepPolicy)
 	hoist := a.earlyUserTurns(region)
+	carryDigests := a.carryPriorDigests(region)
 	for i, m := range region {
 		switch {
 		case m.LocalOnly: // display-only output never reaches a provider
 		case hoist[i]:
 			early = append(early, m)
-		case policyKeep[i] && !isCompactionSummary(m):
+		case isCompactionSummary(m):
+			if carryDigests {
+				carried = append(carried, m)
+				continue
+			}
+			fold = append(fold, m)
+		case policyKeep[i]:
 			kept = append(kept, m)
 		default:
 			fold = append(fold, m)
 		}
 	}
-	return early, kept, fold
+	return early, carried, kept, fold
+}
+
+// carryPriorDigests reports whether the digests already in the region survive
+// this fold verbatim instead of being re-summarized. Re-summarizing a digest is
+// the only step in a fold where a fact can be dropped and never recovered, so
+// an incremental fold carries them — until they outgrow their budget, when one
+// consolidating fold merges them and the chain starts over.
+func (a *Agent) carryPriorDigests(region []provider.Message) bool {
+	if !a.ablation.Off(ablation.FullFold) {
+		return false
+	}
+	total := 0
+	for _, m := range region {
+		if isCompactionSummary(m) {
+			total += summaryInputTokens([]provider.Message{m})
+		}
+	}
+	if total > maxCarriedDigestTokens {
+		a.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelInfo, Text: fmt.Sprintf(
+			"consolidating %d tokens est. of carried digests into one; earlier facts now depend on this merge", total)})
+		return false
+	}
+	return true
 }
 
 // earlyUserTurns marks the region positions of the small user turns hoisted

internal/agent/compact_test.go

@@ -486,9 +486,9 @@ func TestInterruptedDisplayStaysOutOfCompactionPromptAndProjection(t *testing.T)
 		InterruptedTurn: &provider.InterruptedTurnRecovery{Pending: true},
 	}
 	a := &Agent{}
-	early, kept, fold := a.partitionFoldForProjection([]provider.Message{local})
-	if len(early) != 0 || len(kept) != 0 || len(fold) != 0 {
-		t.Fatalf("compaction partition early=%+v kept=%+v fold=%+v, want display-only output in none of them", early, kept, fold)
+	early, carried, kept, fold := a.partitionFoldForProjection([]provider.Message{local})
+	if len(early) != 0 || len(carried) != 0 || len(kept) != 0 || len(fold) != 0 {
+		t.Fatalf("compaction partition early=%+v carried=%+v kept=%+v fold=%+v, want display-only output in none of them", early, carried, kept, fold)
 	}
 	if transcript := renderTranscript([]provider.Message{local}); transcript != "" {
 		t.Fatalf("local interrupted output leaked into compaction prompt: %q", transcript)