Description of Change

RedisSessionService.append_event() currently serializes the caller's event list as the entire stored session. If the caller loaded only recent events with GetSessionConfig, the first append permanently deletes the excluded history. Runner.run_async() with RunConfig.get_session_config reaches the same path.

Reload the stored event history inside a Redis WATCH/MULTI/EXEC transaction, append the processed event once, and retain the caller's filtered view. A retry reloads the canonical history so another append is preserved. The stored JSON format, public API, state handling and configured TTL remain unchanged. Redis's existing behavior for missing or expired sessions is retained.

Reproduction

Environment details: Current main 3a06a8621d04fd87aad9f6b078c999c78a96fb66, ADK 2.9.0 package metadata, macOS arm64, Python 3.12.10, Redis 8.4.0, redis-py 8.1.0. No LiteLLM or model call.

Start a disposable Redis instance, for example redis-server --port 0 --unixsocket /tmp/adk-history-example.sock --save "" --appendonly no. Install ADK and redis, save the following as repro.py, then run REDIS_SOCKET=/tmp/adk-history-example.sock python repro.py. The custom agent returns a local reply without a model or credentials. It exercises both direct appends and Runner.run_async().

import asyncio
import os
from uuid import uuid4

from google.adk.agents.base_agent import BaseAgent
from google.adk.agents.run_config import RunConfig
from google.adk.events import Event
from google.adk.integrations.redis import RedisSessionService, RedisSessionServiceConfig
from google.adk.runners import Runner
from google.adk.sessions.base_session_service import GetSessionConfig
from google.genai import types
import redis.asyncio as redis


class LocalReplyAgent(BaseAgent):
    async def _run_async_impl(self, ctx):
        yield Event(
            invocation_id=ctx.invocation_id,
            author=self.name,
            content=types.Content(role="model", parts=[types.Part(text="reply")]),
        )


def messages(session):
    return [event.content.parts[0].text for event in session.events]


async def main():
    failures = []
    async with redis.Redis(
        unix_socket_path=os.environ["REDIS_SOCKET"], decode_responses=True
    ) as client:
        service = RedisSessionService(
            redis_client=client,
            config=RedisSessionServiceConfig(
                key_prefix=f"history-repro:{uuid4()}:", ttl_seconds=60
            ),
        )
        config = GetSessionConfig(num_recent_events=2)
        for mode in ("direct", "runner"):
            scope = dict(app_name="demo", user_id="user", session_id=mode)
            session = await service.create_session(**scope)
            for i in range(4):
                await service.append_event(
                    session,
                    Event(author="user", invocation_id=f"turn-{i}", message=f"old-{i}"),
                )
            view = await service.get_session(**scope, config=config)
            print(f"{mode} loaded:", messages(view))
            if mode == "direct":
                await service.append_event(
                    view, Event(author="user", invocation_id="new-turn", message="new")
                )
            else:
                runner = Runner(
                    app_name=scope["app_name"],
                    agent=LocalReplyAgent(name="local_reply"),
                    session_service=service,
                )
                async for _ in runner.run_async(
                    user_id=scope["user_id"], session_id=scope["session_id"],
                    new_message=types.Content(role="user", parts=[types.Part(text="new")]),
                    run_config=RunConfig(get_session_config=config),
                ):
                    pass
            stored = await service.get_session(**scope)
            print(f"{mode} stored:", messages(stored))
            expected = [f"old-{i}" for i in range(4)] + ["new"]
            if mode == "runner":
                expected.append("reply")
            if messages(stored) != expected:
                failures.append(mode)
    assert not failures, failures


asyncio.run(main())

Before the fix, both workflows lose the two excluded events (exit 1):

direct loaded: ['old-2', 'old-3']
direct stored: ['old-2', 'old-3', 'new']
runner loaded: ['old-2', 'old-3']
runner stored: ['old-2', 'old-3', 'new', 'reply']
AssertionError: ['direct', 'runner']

Expected behavior, verified with the fixed wheel (exit 0):

direct loaded: ['old-2', 'old-3']
direct stored: ['old-0', 'old-1', 'old-2', 'old-3', 'new']
runner loaded: ['old-2', 'old-3']
runner stored: ['old-0', 'old-1', 'old-2', 'old-3', 'new', 'reply']

Testing Plan

• Final regression tests reject the original source: 8 fail, with the unfiltered control passing. Fixed Redis and session suites: 522 passed, 3 existing expected failures. Coverage includes recent-event, zero-event and timestamp filters, two appends, full Event equality, state scopes, partial events, intervening append/delete/expiry and TTL refresh.
• Adjacent Runner, local-storage forwarding, compaction, rewind, live-flow and A2A tests: 1017 passed, 50 skipped, 3 expected failures. These runs emitted upstream/dependency warnings; skipped paths are not claimed as verified.
• Built baseline and fixed wheels and installed them in a clean environment. A real Redis and deterministic custom-agent Runner check covers 48 cases across Redis, memory, SQLite and SQLAlchemy SQLite. The baseline loses history in 8 Redis cases; the fixed wheel passes all 48. Actual Redis WATCH conflicts, delete/expiry retries and 20 concurrent filtered appends pass. The same 52 checks pass with redis-py 4.2.0 on Python 3.10 and 8.1.0 on Python 3.12. The SQLAlchemy SQLite control requires its asyncio extra on this host.
• Full pre-commit passes on both baseline and fixed trees; changed-file and commit hooks pass.
• Full-extra mypy 2.3.1 comparisons on Python 3.10–3.13 find no new diagnostics. Baseline and fixed each report the same 842 errors after line normalization, so raw mypy exits remain 1. These macOS runs use the CI lancedb omission; they are not the GitHub-hosted Linux jobs.
• Full unit suite completed through tox on every supported Python version, using the upstream pytest tests/unittests command. All five tox environments exit 1 with the failures detailed below; this is not an all-green matrix.

Full-suite failures and environment:

• All five versions fail tests/unittests/evaluation/test_local_eval_service.py::test_eval_injects_session_input_state_into_instruction with empty inferences. Running the unchanged baseline's entire test_local_eval_service.py module with each corresponding tox environment reproduces the same failure: 1 failed, 26 passed. This test uses the default in-memory session service.
• Python 3.12 and 3.13 additionally fail both test_entry_point_loads_only_allowlisted_packages cases because Homebrew Python loads sitecustomize. Both cases fail identically on the unchanged baseline. A fresh interpreter already has Homebrew's sitecustomize loaded before any ADK import.
• Dependencies were locked locally in a separate verification checkout with UV_EXCLUDE_NEWER=2026-09-16T09:00:00Z; no lockfile or generated artifact is submitted. The final runs use a tool-only PATH without gcloud because an existing CLI login test otherwise invokes an interactive ADC login on this host. The evaluation and import tests were run unchanged, with no additional test deselection. Earlier interrupted runs are not counted as completed checks.
• Alternate A2A/MCP dependency matrices and GitHub-hosted CI have not been run locally.

Documentation

The companion documentation PR clarifies that retrieval filters limit the loaded view without deleting persisted events. Its strict build and served-page inspection pass. The two changes should be coordinated so the documentation reflects the Redis correction.