Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions tests/unittests/apps/test_compaction.py
Original file line number Diff line number Diff line change
Expand Up @@ -813,6 +813,51 @@ def test_get_contents_with_multiple_compactions(self):
self.assertEqual(actual_texts, expected_texts)
# Verify timestamps are in order

def test_replay_contract_compacted_and_raw_events_match_effective_prompt(
self,
):
raw_events = [
self._create_event(1.0, 'inv1', 'User asks about weather'),
self._create_event(2.0, 'inv1', 'Agent asks clarifying question'),
self._create_event(3.0, 'inv2', 'User clarifies location'),
self._create_event(4.0, 'inv2', 'Agent proposes plan'),
self._create_event(5.0, 'inv3', 'User asks for final answer'),
]

compacted_events = [
self._create_compacted_event(
1.0,
4.0,
(
'User asks about weather\n'
'Agent asks clarifying question\n'
'User clarifies location\n'
'Agent proposes plan'
),
appended_ts=4.5,
),
self._create_event(5.0, 'inv3', 'User asks for final answer'),
]

raw_prompt = '\n'.join(
part.text
for content in contents._get_contents(None, raw_events)
for part in content.parts
if part.text
)
compacted_prompt = '\n'.join(
part.text
for content in contents._get_contents(None, compacted_events)
for part in content.parts
if part.text
)

self.assertEqual(
compacted_prompt,
raw_prompt,
'Compaction should preserve deterministic replay prompt semantics.',
)
Comment on lines +816 to +859
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To improve maintainability and reduce code duplication, you can extract the logic for reconstructing the prompt from a list of events into a local helper function. This makes the test cleaner and avoids repeating the same generator expression.

  def test_replay_contract_compacted_and_raw_events_match_effective_prompt(
      self,
  ):
    raw_events = [
        self._create_event(1.0, 'inv1', 'User asks about weather'),
        self._create_event(2.0, 'inv1', 'Agent asks clarifying question'),
        self._create_event(3.0, 'inv2', 'User clarifies location'),
        self._create_event(4.0, 'inv2', 'Agent proposes plan'),
        self._create_event(5.0, 'inv3', 'User asks for final answer'),
    ]

    compacted_events = [
        self._create_compacted_event(
            1.0,
            4.0,
            (
                'User asks about weather\n'
                'Agent asks clarifying question\n'
                'User clarifies location\n'
                'Agent proposes plan'
            ),
            appended_ts=4.5,
        ),
        self._create_event(5.0, 'inv3', 'User asks for final answer'),
    ]

    def _reconstruct_prompt(events_list):
      return '\n'.join(
          part.text
          for content in contents._get_contents(None, events_list)
          for part in content.parts
          if part.text
      )

    raw_prompt = _reconstruct_prompt(raw_events)
    compacted_prompt = _reconstruct_prompt(compacted_events)

    self.assertEqual(
        compacted_prompt,
        raw_prompt,
        'Compaction should preserve deterministic replay prompt semantics.',
    )


def test_get_contents_subsumed_compaction_is_hidden(self):
events = [
self._create_event(1.0, 'inv1', 'Event 1'),
Expand Down