Runner persists plugin replacement events according to the original event's partial/media flags

Problem

on_event_callback can mutate an Event or return a replacement. On main 5bc9e8c, equivalent transformations behave differently: turning a partial event into a final event in place persists the final event and state delta, while returning a replacement delivers the final event to the caller but loses both the event and state delta from the session.

The same stale-input check affects live media filtering: replacing inline audio with text still excludes the resulting text, while replacing text with inline audio stores data the live persistence filter would normally exclude.

Expected

Persistence decisions should inspect the post-callback event that Runner delivers to the caller. Mutation and replacement should have the same persistence effects for equivalent output events.

Reproducer

Run the following from an ADK development checkout with PYTHONPATH=src python repro.py. No model or network is used.

import asyncio,json
from google.adk.agents import BaseAgent
from google.adk.events import Event,EventActions
from google.adk.plugins.base_plugin import BasePlugin
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.live import LiveRequestQueue
from google.genai import types

async def probe(mode,style):
    class Agent(BaseAgent):
        async def _run_async_impl(self,ctx):
            yield Event(author=self.name,partial=True,
                        content=types.Content(role='model',parts=[types.Part(text='draft')]))
        _run_live_impl=_run_async_impl
    class Finalize(BasePlugin):
        async def on_event_callback(self,*,invocation_context,event):
            if not event.partial: return None
            if style=='in_place':
                event.partial=False
                event.content=types.Content(role='model',parts=[types.Part(text='final')])
                event.actions=EventActions(state_delta={'finalized':True})
                return None
            return Event(author=event.author,partial=False,
                         content=types.Content(role='model',parts=[types.Part(text='final')]),
                         actions=EventActions(state_delta={'finalized':True}))
    service=InMemorySessionService()
    runner=Runner(app_name='scout',agent=Agent(name='agent'),session_service=service,
                  plugins=[Finalize(name='finalize')])
    session=await service.create_session(app_name='scout',user_id='u')
    if mode=='live':
        gen=runner.run_live(user_id='u',session_id=session.id,live_request_queue=LiveRequestQueue())
    else:
        gen=runner.run_async(user_id='u',session_id=session.id,new_message=types.Content(role='user',parts=[types.Part(text='hello')]))
    output=[e async for e in gen]
    stored=await service.get_session(app_name='scout',user_id='u',session_id=session.id)
    await runner.close()
    return {'mode':mode,'style':style,'yielded_final':any(e.content and any(p.text=='final' for p in e.content.parts or []) for e in output),
            'stored_final':any(e.content and any(p.text=='final' for p in e.content.parts or []) for e in stored.events),'state':stored.state}

async def main():
    results=[await probe(m,s) for m in ['async','live'] for s in ['in_place','replacement']]
    print(json.dumps(results,indent=2))
if __name__=='__main__':asyncio.run(main())

On unmodified main, all four cases report yielded_final: true. Only in-place cases report stored_final: true and state { "finalized": true }; replacement cases report stored_final: false and empty state.

Diagnosis and verification

Persistence eligibility is currently evaluated against the pre-callback event rather than the effective event returned from on_event_callback. Once a replacement is accepted, persistence and live filtering decisions governing that output should use the effective event.

A focused fix is prepared locally that aligns persistence and live filtering decisions with the effective event. It includes 20 regression cases, including SSE and live media filtering on normal and before-run early-exit paths. Baseline: 7 fail, 13 pass. Patched: all 20 pass; related suites: 259 pass, 1 skip, 3 expected failures. Pre-commit passes. Python 3.11.9/macOS arm64; full supported-Python tox is pending.

I have a focused local fix and regression tests prepared and would be happy to send the PR if this direction is appropriate. I will link a draft pending maintainer confirmation of the persistence semantics and completion of validation.

Related but distinct: #3990 concerns which Event is persisted after callbacks; this report concerns which Event determines persistence eligibility. #5161 proposes a separate post-persistence/pre-yield hook; this report stays within the existing on_event_callback contract. No matching open issue/PR found in my search; please flag any overlapping work.