Skip to content
Open
Show file tree
Hide file tree
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
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from __future__ import annotations

import json
import logging
import warnings
from collections.abc import Awaitable, Sequence
from inspect import isawaitable
Expand All @@ -23,6 +24,8 @@

from ._feature_usage import FeatureIndex

logger = logging.getLogger(__name__)

_T = TypeVar("_T")


Expand Down Expand Up @@ -271,7 +274,18 @@ async def get_messages(
redis_messages = cast("list[str]", await _redis_result(cast("Any", self._redis_client).lrange(key, 0, -1)))
messages: list[Message] = []
for serialized in redis_messages:
messages.append(Message.from_dict(self._deserialize_json(serialized)))
try:
payload = self._deserialize_json(serialized)
if not isinstance(payload, dict):
raise ValueError(f"Expected a JSON object, got {type(payload).__name__}")
messages.append(Message.from_dict(payload))
except (ValueError, TypeError, AttributeError, json.JSONDecodeError) as exc:

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

lisry-1 Skipping a malformed middle record can duplicate the valid history on the next save — For stored history [A, malformed, C], get_messages() now returns [A, C]. If the caller then saves the full transcript [A, B, C, D], the current contiguous-overlap check can append all four messages and persist [A, C, A, B, C, D]. Please make save-time deduplication account for skipped records and add this replay regression.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Skipping one record here can break function-call/result pairing: dropping a malformed tool-result record leaves its assistant call without a terminal result, while dropping a malformed call record can leave an orphaned tool result. Those lossy histories can survive serialization and make the next provider request invalid. Could we recover or reject complete tool exchanges atomically rather than individual records, and add corrupted call/result regressions consistent with the function-calling loop contract?

# Log the exception class, not its message: ``Message.from_dict``
# embeds the payload's ``type`` value in mismatch errors, so logging
# ``str(exc)`` could copy untrusted history data into application logs.
logger.warning(
"Skipping malformed Redis history entry (%s)", type(exc).__name__
)
return messages

async def save_messages(
Expand Down
44 changes: 44 additions & 0 deletions python/packages/redis/tests/test_providers.py
Original file line number Diff line number Diff line change
Expand Up @@ -681,6 +681,50 @@ async def test_empty_returns_empty(self, mock_redis_client: MagicMock):
messages = await provider.get_messages("s1")
assert messages == []

async def test_skips_malformed_entries(self, mock_redis_client: MagicMock):
msg = Message(role="user", contents=["Hello"])
mock_redis_client.lrange = AsyncMock(
return_value=[
"",
"not-json",
"null", # valid JSON, but deserializes to None
"[]", # valid JSON, but deserializes to a list
"{}", # valid JSON, but missing the required "role" field
json.dumps(msg.to_dict()),
]
)

with patch("agent_framework_redis._history_provider.redis.from_url") as mock_from_url:
mock_from_url.return_value = mock_redis_client
provider = RedisHistoryProvider("mem", redis_url="redis://localhost:6379", application_id="test-app")

messages = await provider.get_messages("s1")
assert len(messages) == 1
assert messages[0].text == "Hello"

async def test_malformed_entries_do_not_leak_payload_in_logs(
self, mock_redis_client: MagicMock, caplog: pytest.LogCaptureFixture
):
# A payload whose ``type`` field carries a private value must not be echoed
# into the skip warning: ``Message.from_dict`` embeds ``type`` in mismatch
# errors, and logging ``str(exc)`` would copy it into application logs.
sensitive = "secret-internal-token"
mock_redis_client.lrange = AsyncMock(
return_value=[json.dumps({"type": sensitive, "role": "unknown"})]
)

with patch("agent_framework_redis._history_provider.redis.from_url") as mock_from_url:
mock_from_url.return_value = mock_redis_client
provider = RedisHistoryProvider("mem", redis_url="redis://localhost:6379", application_id="test-app")

with caplog.at_level("WARNING", logger="agent_framework_redis._history_provider"):
messages = await provider.get_messages("s1")

assert messages == []
warning_text = "\n".join(r.message for r in caplog.records)
assert "Skipping malformed Redis history entry" in warning_text
assert sensitive not in warning_text

async def test_returns_messages_when_lrange_is_synchronous(self, mock_redis_client: MagicMock):
"""redis-py types several commands as returning a value or an awaitable; handle both."""
msg = Message(role="user", contents=["Hello"])
Expand Down
Loading