diff --git a/core/src/commonMain/kotlin/com/google/adk/kt/processors/HistoryRewriterProcessor.kt b/core/src/commonMain/kotlin/com/google/adk/kt/processors/HistoryRewriterProcessor.kt index ab2b8021..7989df32 100644 --- a/core/src/commonMain/kotlin/com/google/adk/kt/processors/HistoryRewriterProcessor.kt +++ b/core/src/commonMain/kotlin/com/google/adk/kt/processors/HistoryRewriterProcessor.kt @@ -120,12 +120,16 @@ internal class HistoryRewriterProcessor { ) } - // Pass 2: append raw (non-compaction) events that don't fall into a kept compaction range. + // Pass 2: append raw (non-compaction) events that don't fall into a kept compaction range. A + // raw event is only covered by a compaction that appears after it in the stream (`index < + // it.index`). finalItems += events .withIndex() .filter { (_, event) -> event.actions.compaction == null } - .filter { (_, event) -> keptCompactionRanges.none { event.timestamp in it.start..it.end } } + .filter { (index, event) -> + keptCompactionRanges.none { index < it.index && event.timestamp in it.start..it.end } + } .map { (index, event) -> Item(event.timestamp, index, event) } return finalItems.sortedWith(compareBy({ it.timestamp }, { it.index })).map { it.event } diff --git a/core/src/commonMain/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactor.kt b/core/src/commonMain/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactor.kt index 56699d68..2614825d 100644 --- a/core/src/commonMain/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactor.kt +++ b/core/src/commonMain/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactor.kt @@ -95,7 +95,21 @@ class TokenThresholdEventCompactor( if (candidateEvents.size <= eventRetentionSize) return null // Where the kept-raw tail begins; candidates before it are compacted. - val firstRetainedIndex = candidateEvents.size - eventRetentionSize + var firstRetainedIndex = candidateEvents.size + if (eventRetentionSize > 0) { + firstRetainedIndex -= eventRetentionSize + val retentionBoundaryTimestamp = candidateEvents[firstRetainedIndex].timestamp + + // Move the cut back so it never splits a same-timestamp group; the summary's endTimestamp + // must stay below every retained event, or coverage would drop the event that ties it. + while ( + firstRetainedIndex > 0 && + candidateEvents[firstRetainedIndex - 1].timestamp >= retentionBoundaryTimestamp + ) { + firstRetainedIndex-- + } + } + val eventsToCompact = longestSelfContainedPrefix(candidateEvents.subList(0, firstRetainedIndex)) if (eventsToCompact.isEmpty()) return null diff --git a/core/src/commonTest/kotlin/com/google/adk/kt/processors/HistoryRewriterProcessorTest.kt b/core/src/commonTest/kotlin/com/google/adk/kt/processors/HistoryRewriterProcessorTest.kt new file mode 100644 index 00000000..a5ad7b93 --- /dev/null +++ b/core/src/commonTest/kotlin/com/google/adk/kt/processors/HistoryRewriterProcessorTest.kt @@ -0,0 +1,86 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.google.adk.kt.processors + +import com.google.adk.kt.testing.compactionEvent +import com.google.adk.kt.testing.userEvent +import kotlin.test.Test +import kotlin.test.assertEquals + +class HistoryRewriterProcessorTest { + + @Test + fun rewrite_laterEventSharingCompactionEndTimestamp_isKept() { + // Forward same-millisecond tie: a summary covers [100, 200], and a later raw event shares the + // summary's endTimestamp (200). Because it appears after the summary in the stream it is not + // covered by it, so it must survive in the rebuilt context -- while the earlier event it does + // cover is replaced by the summary. + val events = + listOf( + userEvent("covered", timestamp = 100L), + compactionEvent(startTs = 100L, endTs = 200L, summary = "SUM"), + userEvent("later", timestamp = 200L), + ) + + val contents = + HistoryRewriterProcessor().rewrite(events, agentName = "agent", currentBranch = null) + val texts = contents.flatMap { it.parts }.mapNotNull { it.text } + + assertEquals(listOf("SUM", "later"), texts) + } + + @Test + fun rewrite_retainedEventsAboveSummaryRange_areKept() { + // Token-threshold tail-retention layout: the summary is appended last, after the retained tail, + // and covers only [100, 100]. The retained events (200) precede the summary in the stream but + // sit above its range, so they must be kept, and the summary is placed at its endTimestamp. + val events = + listOf( + userEvent("covered", timestamp = 100L), + userEvent("a", timestamp = 200L), + userEvent("b", timestamp = 200L), + compactionEvent(startTs = 100L, endTs = 100L, summary = "SUM"), + ) + + val contents = + HistoryRewriterProcessor().rewrite(events, agentName = "agent", currentBranch = null) + val texts = contents.flatMap { it.parts }.mapNotNull { it.text } + + assertEquals(listOf("SUM", "a", "b"), texts) + } + + @Test + fun rewrite_multipleSummaries_keepBothAndDropCoveredEvents() { + // Two summaries with a forward tie: S1 covers [100, 200] and S2 covers [200, 300]. The event at + // 200 that follows S1 ties S1's endTimestamp but is covered by S2 (not S1). Both summaries are + // kept and every covered raw event is dropped. + val events = + listOf( + userEvent("u1", timestamp = 100L), + userEvent("u2", timestamp = 200L), + compactionEvent(startTs = 100L, endTs = 200L, summary = "S1"), + userEvent("u3", timestamp = 200L), + userEvent("u4", timestamp = 300L), + compactionEvent(startTs = 200L, endTs = 300L, summary = "S2"), + ) + + val contents = + HistoryRewriterProcessor().rewrite(events, agentName = "agent", currentBranch = null) + val texts = contents.flatMap { it.parts }.mapNotNull { it.text } + + assertEquals(listOf("S1", "S2"), texts) + } +} diff --git a/core/src/commonTest/kotlin/com/google/adk/kt/runners/AbstractRunnerTest.kt b/core/src/commonTest/kotlin/com/google/adk/kt/runners/AbstractRunnerTest.kt index 6586559c..12880499 100644 --- a/core/src/commonTest/kotlin/com/google/adk/kt/runners/AbstractRunnerTest.kt +++ b/core/src/commonTest/kotlin/com/google/adk/kt/runners/AbstractRunnerTest.kt @@ -1030,7 +1030,10 @@ class AbstractRunnerTest { eventRetentionSize = 1, summarizer = summarizer, ), - ) + ), + // Deterministic, strictly-increasing timestamps so the retention boundary never lands on a + // same-millisecond wall-clock tie (which would empty the window and skip compaction). + sessionService = MonotonicTimestampSessionService(), ) // First turn: no prior usage metadata, the char estimate is tiny, so no compaction fires. @@ -1091,7 +1094,10 @@ class AbstractRunnerTest { eventRetentionSize = 1, summarizer = summarizer, ), - ) + ), + // Deterministic, strictly-increasing timestamps so the token-threshold retention boundary + // does not land on a same-millisecond wall-clock tie (which would retain "resp" too). + sessionService = MonotonicTimestampSessionService(), ) // Turn 1: neither fires yet -- no reported token count, and below the sliding interval. diff --git a/core/src/commonTest/kotlin/com/google/adk/kt/runners/InMemoryRunnerTest.kt b/core/src/commonTest/kotlin/com/google/adk/kt/runners/InMemoryRunnerTest.kt index f93146bb..d5b937b5 100644 --- a/core/src/commonTest/kotlin/com/google/adk/kt/runners/InMemoryRunnerTest.kt +++ b/core/src/commonTest/kotlin/com/google/adk/kt/runners/InMemoryRunnerTest.kt @@ -228,7 +228,9 @@ class InMemoryRunnerTest { summarizer = summarizer, ), ) - val runner = InMemoryRunner(app = app) + // Monotonic timestamps so the retention boundary never lands on a same-millisecond tie, which + // would empty the window and skip compaction (see MonotonicTimestampSessionService). + val runner = InMemoryRunner(app = app, sessionService = MonotonicTimestampSessionService()) // Turn 1 records a prompt token count; turn 2's pre-call token-threshold compaction then fires. runner diff --git a/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/SlidingWindowEventCompactorTest.kt b/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/SlidingWindowEventCompactorTest.kt index bb1ccbb1..764a6b42 100644 --- a/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/SlidingWindowEventCompactorTest.kt +++ b/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/SlidingWindowEventCompactorTest.kt @@ -467,6 +467,48 @@ class SlidingWindowEventCompactorTest { } } + @Test + fun compact_newInvocationTiesPriorSummaryBoundary_includesTiedInvocationInWindow() = runTest { + // Window-selection boundary tie. A prior summary ends at ts=200, and a genuinely-new invocation + // (inv_3) lands in the same wall-clock millisecond (ts=200). The crossing check uses `<=`, so + // this pins that the tied new invocation is still included in the compaction window -- not + // skipped as "already compacted". + val summarizer = + object : EventSummarizer { + val windows = mutableListOf>() + + override suspend fun summarizeEvents(events: List): Event { + windows.add(events.toList()) + return compactionEvent( + startTs = events.first().timestamp, + endTs = events.last().timestamp, + summary = "S2", + ) + } + } + val sessionService = RecordingSessionService() + val compactor = + SlidingWindowEventCompactor( + EventsCompactionConfig(compactionInterval = 2, overlapSize = 0, summarizer = summarizer) + ) + val session = testSession() + // A prior summary S1 already covers inv_1 and inv_2 (endTimestamp = 200). + session.events.add(userEvent("u1", invocationId = "inv_1", timestamp = 100L)) + session.events.add(userEvent("u2", invocationId = "inv_2", timestamp = 200L)) + session.events.add(compactionEvent(startTs = 100L, endTs = 200L, summary = "S1")) + // Two new invocations; inv_3 lands in the same millisecond as S1's endTimestamp. + session.events.add(userEvent("u3", invocationId = "inv_3", timestamp = 200L)) + session.events.add(userEvent("u4", invocationId = "inv_4", timestamp = 300L)) + + compactor.compact(session, sessionService) + + // The tied new invocation (u3) is included in the compaction window, not skipped. + assertEquals(1, sessionService.appended.size) + val windowTexts = + summarizer.windows.single().flatMap { it.content?.parts.orEmpty() }.mapNotNull { it.text } + assertEquals(listOf("u3", "u4"), windowTexts) + } + // ----- helpers ----- /** diff --git a/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactorTest.kt b/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactorTest.kt index cf0f5222..7ef7470b 100644 --- a/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactorTest.kt +++ b/core/src/commonTest/kotlin/com/google/adk/kt/summarizer/TokenThresholdEventCompactorTest.kt @@ -343,6 +343,43 @@ class TokenThresholdEventCompactorTest { assertSame(b, latestCompactionEvent(events)) } + @Test + fun compact_sameTimestampAtRetentionBoundary_retainsWholeTieGroup() = runTest { + // A same-timestamp group is never split across the compaction boundary: only events strictly + // below the first retained event's timestamp are summarized. A and B share ts=200, so with + // retention=1 both stay retained and only e1 (ts=100) is summarized, keeping the summary's + // endTimestamp below every retained event. + val summarizer = + object : EventSummarizer { + val windows = mutableListOf>() + + override suspend fun summarizeEvents(events: List): Event { + windows.add(events.toList()) + return compactionEvent( + startTs = events.first().timestamp, + endTs = events.last().timestamp, + summary = "SUM", + ) + } + } + val sessionService = RecordingSessionService() + val compactor = + tokenThresholdCompactor(tokenThreshold = 100, eventRetentionSize = 1, summarizer) + val session = testSession() + val e1 = userEvent("u1", invocationId = "inv_1", timestamp = 100L) + val a = modelEvent("A", invocationId = "inv_2", timestamp = 200L) + val b = modelEventWithPromptTokens(150, text = "B", invocationId = "inv_2", timestamp = 200L) + session.events.addAll(listOf(e1, a, b)) + + compactor.compact(session, sessionService) + + // Only e1 is summarized; the whole same-timestamp group (A, B) is retained, and the summary's + // endTimestamp (100) stays strictly below the retained events' timestamp (200). + assertEquals(1, sessionService.appended.size) + assertEquals(listOf(e1), summarizer.windows.single()) + assertEquals(100L, sessionService.appended.single().actions.compaction?.endTimestamp) + } + // ----- helpers ----- private fun tokenThresholdCompactor(