Python: [Bug]: CosmosHistoryProvider re-persists replayed transcript prefixes, causing duplicate history growth #8268

Description

CosmosHistoryProvider.save_messages() persists every supplied message, including messages that are already stored for the same session and provider source. A caller that supplies a cumulative transcript can therefore silently add another copy of the persisted prefix.

For example:

• Existing persisted history: [A, B]
• Later cumulative input: [A, B, C]
• Expected persisted history: [A, B, C]
• Resulting logical persisted history under current write behavior: [A, B, A, B, C]

This logical result is inferred from directly observing that the second save submits three new Cosmos upserts (A, B, and C) instead of only one (C); the reproduction does not perform a live Cosmos readback. Each document receives a fresh UUID, so Cosmos upsert semantics do not replace the earlier logical messages. Replaying successively longer transcripts can consequently produce superlinear persistent-history growth.

The shared history helper filter_new_messages(existing, incoming) defines the established behavior for both append-only inputs and full-transcript replay while preserving legitimate repeated turns. The in-memory, file, and Redis history providers apply that behavior, but the Cosmos provider currently does not.

Code Sample

import asyncio
from collections.abc import AsyncIterator
from unittest.mock import AsyncMock, MagicMock

from agent_framework import Message
from agent_framework_azure_cosmos import CosmosHistoryProvider


async def items(messages: list[Message]) -> AsyncIterator[dict[str, object]]:
    for message in messages:
        yield {"message": message.to_dict()}


async def main() -> None:
    existing = [
        Message(role="user", contents=["A"]),
        Message(role="assistant", contents=["B"]),
    ]
    incoming = [
        Message(role="user", contents=["A"]),
        Message(role="assistant", contents=["B"]),
        Message(role="user", contents=["C"]),
    ]
    container = MagicMock()
    container.query_items.return_value = items(existing)
    container.execute_item_batch = AsyncMock(return_value=[])
    provider = CosmosHistoryProvider(source_id="mem", container_client=container)

    await provider.save_messages("s1", incoming)

    operations = container.execute_item_batch.await_args.kwargs["batch_operations"]
    assert len(operations) == 1  # Fails on current main: 3 == 1
    assert operations[0][1][0]["message"]["contents"][0]["text"] == "C"


asyncio.run(main())

Error Messages / Stack Traces

There is no runtime exception. The failure is silent duplicate persistence and history growth. The deterministic regression assertion on current main is:

E       AssertionError: assert 3 == 1

Package Versions

agent-framework: 1.18.0; agent-framework-core: 1.18.0; agent-framework-azure-cosmos: 1.0.0b260910; source: c37de51

Python Version

Python 3.12.12

Additional Context

This appears to be a remaining provider-specific instance of the history replay behavior reported in #7211 and addressed for the in-memory, file, and Redis providers in #7242. The shared helper introduced there explicitly handles append-only input and full-transcript replay while preserving legitimate duplicate turns.

A related follow-up, #8223, was fixed by #8224, which applied the same helper to the MessagePack branch of FileHistoryProvider; Cosmos remains a separate affected persistence backend.

At the direct provider level, supplying only message deltas avoids replaying the stored prefix. That does not protect call paths that validly pass an accumulated transcript.