Resumable flow: a parallel tool call that never executed is never replayed when a sibling call was answered

## 🔴 Required Information

**Describe the Bug:**

In resumable mode, when one event carries **parallel** function calls and only **some** of them produced a response before the interruption, `decide_step_resume()` returns `CONTINUE`. The calls that never executed are never replayed: they do not run, no error is raised, and the flow proceeds to the next LLM call as if it had answers it never received.

The two extremes are handled correctly — nothing answered replays, everything answered continues. Only the partial case, which is the case resumption exists for, is lost.

The decision asks *whether **any** answer came back* rather than *whether **every** call was answered*:

- [`_resume_utils.py:192`](https://github.com/google/adk-python/blob/main/src/google/adk/flows/llm_flows/_resume_utils.py#L192) — `any(fr.name not in call_names for fr in answers)` matches on **names**, so an answer to one call reads as an answer to the event.
- [`_resume_utils.py:251`](https://github.com/google/adk-python/blob/main/src/google/adk/flows/llm_flows/_resume_utils.py#L251) — `not call_ids & answered_ids` is an **intersection**, so one answered id clears the whole set.

`call_ids` and `answered_ids` are both already in scope at that point.

**Steps to Reproduce:**

1. `pip install google-adk==2.9.0`
2. Save the script under *Minimal Reproduction Code* as `repro.py`
3. `python repro.py`

**Expected Behavior:**

A call with no response is replayed, whether or not a sibling call in the same event was answered — cases A and B below should be `replay_calls`.

**Observed Behavior:**

```
A. parallel calls, different names, only c1 executed
executed=['c1'] never executed=['c2']
decide_step_resume -> continue expected replay_calls LOST

B. parallel calls, same name twice, only c1 executed
executed=['c1'] never executed=['c2']
decide_step_resume -> continue expected replay_calls LOST

C. NEGATIVE CONTROL — all executed (mirrors your own test)
executed=['c1', 'c2'] never executed=none
decide_step_resume -> continue expected continue ok

D. NEGATIVE CONTROL — none executed
executed=none never executed=['c1', 'c2']
decide_step_resume -> replay_calls expected replay_calls ok

E. NEGATIVE CONTROL — single call, not executed
executed=none never executed=['c1']
decide_step_resume -> replay_calls expected replay_calls ok
```

**Environment Details:**

- ADK Library Version: `google-adk 2.9.0` (the same lines are present on `main`)
- Desktop OS: macOS 26.6.2 (arm64)
- Python Version: 3.12.13

**Model Information:**

- Are you using LiteLLM: No
- Which model is being used: N/A — the repro calls `decide_step_resume()` directly and needs no model

---

## 🟡 Optional Information

**Additional Context:**

The correct idiom is already in this file, 130 lines above, in `_pause_left_calls_unanswered`:

```python
# `issubset`, not `&`: this asks whether *any* awaited id is still open, so a
# partially answered pause keeps waiting. `decide_resume` asks the opposite
# question of its own ids -- whether *none* are answered -- and drops
# `issubset` for that reason. The two are not interchangeable.
return bool(awaited) and not awaited.issubset(answered)
```

That comment states the distinction exactly; the replay decision is the second place that needs it.

`test_parallel_calls_all_answered_continue` covers the fully answered case, and its comment says it exists so a fully answered event is not replayed and the tools do not run twice. That guard is right — the partial case appears to be the gap it left, and it has no test.

I have deliberately **not** proposed a patch, because the fix is a design choice I am not in a position to make: replaying the event runs *all* of its calls, so avoiding duplicate execution of the calls that already succeeded means either replaying per call or filtering the event down to the unanswered ids. Both change behaviour beyond this function.

**Scope — what I did and did not verify:**

- Verified: `decide_step_resume()` in isolation, with the event shapes used by `tests/unittests/flows/llm_flows/test_resume_utils.py`.
- Not verified: an end-to-end run against a live model, or how often a real interruption lands between sibling responses. Whether this is reachable in practice depends on when responses are persisted relative to the crash, which I have not measured.
- The `Ctx` class in the repro is mine, standing in for `InvocationContext`; it supplies only the three members `decide_step_resume` reads.

**Minimal Reproduction Code:**

```python
"""ADK resumable flow, real entry point decide_step_resume().
A parallel tool call that never executed is never replayed: the flow continues
to the LLM as if it had an answer it never got."""
from google.genai import types
from google.adk.events.event import Event
from google.adk.flows.llm_flows._resume_utils import decide_step_resume, ResumeAction

class Ctx:
"""Only what decide_step_resume touches."""
def __init__(self, events): self._events, self.is_resumable = events, True
def _get_events(self, current_invocation=True, current_branch=True): return self._events
def should_pause_invocation(self, ev): return False

def call_event(pairs):
return Event(author='agent', invocation_id='inv-1',
content=types.Content(role='model', parts=[
types.Part(function_call=types.FunctionCall(id=i, name=n, args={}))
for i, n in pairs]))

def response_event(name, cid):
return Event(author='user', invocation_id='inv-1',
content=types.Content(role='user', parts=[
types.Part(function_response=types.FunctionResponse(
id=cid, name=name, response={'r': 'ok'}))]))

def run(label, calls, ran, tools, expect):
events = [call_event(calls)] + [response_event(n, i) for i, n in calls if i in ran]
d = decide_step_resume(Ctx(events), {t: object() for t in tools})
missing = [i for i, _ in calls if i not in ran]
bad = missing and d.action is ResumeAction.CONTINUE
print(f'{label}\n executed={sorted(ran) or "none"} never executed={missing or "none"}')
print(f' decide_step_resume -> {d.action.value:12s} expected {expect:12s}'
f' {"LOST" if bad else "ok"}\n')

run('A. parallel calls, different names, only c1 executed',
[('c1','ask'), ('c2','fetch')], {'c1'}, ['ask','fetch'], 'replay_calls')
run('B. parallel calls, same name twice, only c1 executed',
[('c1','ask'), ('c2','ask')], {'c1'}, ['ask'], 'replay_calls')
run('C. NEGATIVE CONTROL — all executed (mirrors their own test)',
[('c1','ask'), ('c2','fetch')], {'c1','c2'}, ['ask','fetch'], 'continue')
run('D. NEGATIVE CONTROL — none executed',
[('c1','ask'), ('c2','fetch')], set(), ['ask','fetch'], 'replay_calls')
run('E. NEGATIVE CONTROL — single call, not executed',
[('c1','ask')], set(), ['ask'], 'replay_calls')
```

**How often has this issue occurred?:**

- Always (100%) — deterministic for the inputs above.

**Related:** #7076 is a different defect in the same subsystem (who may author a dispatched call); this one is about which calls are replayed.