Python: fix: prevent colliding message_id values in incremental compaction

python/packages/core/agent_framework/_compaction.py

@@ -110,6 +110,12 @@ def group_messages(messages: list[Message]) -> list[dict[str, Any]]:
     Returns:
         Ordered list of lightweight span dicts with keys:
         ``group_id``, ``kind``, ``start_index``, ``end_index``, ``has_reasoning``.
+
+    Note:
+        When called standalone, this function will assign ``message_id``
+        values to messages that lack them (using list index).  When called
+        via ``annotate_message_groups``, IDs are pre-assigned with absolute
+        indices to avoid collisions across incremental calls.
     """
     _ensure_message_ids(messages)
     spans: list[dict[str, Any]] = []
@@ -439,6 +445,13 @@ def annotate_message_groups(
         if previous_group_index is not None:
             group_index_offset = previous_group_index + 1
 
+    # Assign message IDs only to the *new* suffix using absolute indices
+    # so that IDs are globally unique without re-scanning the full list
+    # on every incremental call (fixes #5237).
+    for absolute_index, message in enumerate(messages[start_index:], start=start_index):
+        if not message.message_id:
+            message.message_id = f"msg_{absolute_index}"
+
     spans = group_messages(messages[start_index:])
     for span_index, span in enumerate(spans):
         group_id = str(span["group_id"])

python/packages/core/tests/core/test_compaction.py

@@ -952,3 +952,31 @@ async def test_in_memory_history_provider_default_loads_all() -> None:
 
     loaded = await provider.get_messages(session_id="test", state=state)
     assert len(loaded) == 3
+
+
+def test_incremental_annotation_produces_unique_message_ids() -> None:
+    """Incremental calls to annotate_message_groups must not produce colliding message_id values.
+
+    Previously, _ensure_message_ids was called inside group_messages on a
+    *slice* of the full list, so the slice always started at index 0 and
+    produced msg_0, msg_1, ... colliding with IDs assigned in earlier calls.
+
+    Fixes #5237.
+    """
+    messages: list[Message] = []
+
+    # Simulate a multi-turn conversation where messages are appended and
+    # annotated incrementally (as the CompactionProvider does).
+    for turn in range(4):
+        messages.append(Message(role="user", contents=[f"Turn {turn + 1} user"]))
+        messages.append(Message(role="assistant", contents=[f"Turn {turn + 1} assistant"]))
+        annotate_message_groups(messages)
+
+    # Every message should have a message_id
+    assert all(m.message_id for m in messages)
+
+    # All message_ids must be unique
+    ids = [m.message_id for m in messages]
+    assert len(ids) == len(set(ids)), (
+        f"Colliding message_ids detected: {ids}"
+    )