ReflectAndRetryModelPlugin counts one streamed model error twice #7872

🔴 Required Information

Describe the Bug:
With streaming on, ReflectAndRetryModelPlugin counts each model failure twice, so the agent gets about half the retries max_retries promises. With max_retries=1, the very first failure raises "the retry limit has been exceeded", before any retry happens.

Progressive SSE streaming is on by default. Under it, the chunk that ends with an error finish reason (e.g. MALFORMED_FUNCTION_CALL) reaches after_model_callback as a partial response. Then the final aggregated response repeats the same error_code and finish_reason. after_model_callback doesn't check llm_response.partial, so both responses go through _handle_model_error and increment the failure count.

LoggingPlugin and DebugLoggingPlugin already guard against this. Both only record the termination on the final response, so they don't record it twice.

Steps to Reproduce:

1. Run an agent with ReflectAndRetryModelPlugin(max_retries=1) and RunConfig(streaming_mode=StreamingMode.SSE).
2. Have the model end its first stream with finish_reason=MALFORMED_FUNCTION_CALL, then answer normally on the retry.

Expected Behavior:
The plugin sends one reflection call with retry_count=1, the model retries, and the run finishes.

Observed Behavior:

EVENT partial=True call=adk_handle_model_error retry_count=1
RAISED RuntimeError Error in plugin 'reflect_retry_model_plugin' during 'after_model_callback' callback: The model has failed consecutively 1 times and the retry limit has been exceeded.
model calls: 1

With max_retries=3, the same single failure is reported as retry_count=1 (partial) and then retry_count=2 (final).

Environment Details:

• ADK Library Version (pip show google-adk): 2.13.0 (also on main at e9e8f61)
• Desktop OS: macOS
• Python Version (python -V): 3.11.16

Model Information:

• Are you using LiteLLM: No
• Which model is being used: Gemini (gemini-2.5-flash), with SSE streaming

🟡 Optional Information

Minimal Reproduction Code:

calls = {"n": 0}

async def fake_stream(*, model, contents, config):
  calls["n"] += 1
  if calls["n"] == 1:
    chunk = types.GenerateContentResponse(candidates=[types.Candidate(
        finish_reason=types.FinishReason.MALFORMED_FUNCTION_CALL,
        finish_message="Malformed function call: print(foo(")])
  else:
    chunk = types.GenerateContentResponse(candidates=[types.Candidate(
        content=types.Content(role="model", parts=[types.Part(text="done")]),
        finish_reason=types.FinishReason.STOP)])

  async def gen():
    yield chunk
  return gen()

client = genai.Client(api_key="fake-key")
client.aio.models.generate_content_stream = fake_stream
agent = LlmAgent(name="a", model=Gemini(model="gemini-2.5-flash", client=client))
app = App(name="app", root_agent=agent, plugins=[ReflectAndRetryModelPlugin(max_retries=1)])

runner = InMemoryRunner(app=app)
# run_async(..., run_config=RunConfig(streaming_mode=StreamingMode.SSE))

How often has this issue occurred?:

• Always (100%), whenever a streamed response ends with one of the plugin's error finish reasons