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.
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:
[A, B][A, B, C][A, B, C][A, B, A, B, C]This logical result is inferred from directly observing that the second save submits three new Cosmos upserts (
A,B, andC) 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
Error Messages / Stack Traces
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.