[Bug]: ADK middleware re-executes already-answered messages when run with >1 replica (processed-message ledger is process-local)

### Pre-flight Checklist

- [x] I have searched [existing issues](https://github.com/ag-ui-protocol/ag-ui/issues) and this hasn't been reported yet.
- [x] I am using the **latest** version AG-UI.

### Describe the Bug

`SessionManager` keeps its "which client messages have I already run" ledger —
`_processed_message_ids` — in a plain in-process dict (`session_manager.py:96`),
while the ADK session it wraps is durable (`DatabaseSessionService`). That ledger
is the only input to `AdkAgent._get_unseen_messages()` (`adk_agent.py:1542-1567`
on `main`).

So when the middleware runs with more than one replica, a turn routed to a
process that has never served the thread finds the client's entire re-sent
history "unseen" and **re-executes already-answered user messages** before the
new one. The user sees an old message answered again in a fresh run, the
duplicate turns are persisted into the shared session, and the duplicated LLM
calls are billed.

Sharing one `SessionManager` across a process fixes the single-process case
(otherwise every request replays). It cannot fix N processes.

**How many messages replay is unbounded.** The only thing stopping a cold process
from re-running *every* past turn is an incidental heuristic in the run loop
(`adk_agent.py:1371-1397`): a historical user batch is skipped only when `tool`-role
messages immediately follow it, those carry `tool_call_id`s,
`_get_pending_tool_call_ids()` returns non-`None`, and none are pending. A past turn
answered *without* tool calls satisfies none of that and is re-executed. So the count
equals the number of such turns in the history — not capped at one.

This is the correctness counterpart to #2186. That issue treats the re-sent history as
harmless waste — "`_get_unseen_messages` deliberately ignores the re-sent history",
and "the LLM/token cost is not doubled". Both hold only while the ledger is warm. On a
cold process the history is not ignored, it is executed, and the cost *is* multiplied.

`SessionManager.__init__` accepts a durable `session_service` for sessions, and the
`use_thread_id_as_session_id` docstring says it exists so thread→session mapping
survives "middleware restarts". Restart/instance resilience was therefore considered
for session identity but not for this ledger, which is why this reads as an oversight
rather than a deliberate constraint. There is no pluggable store, no persistence hook,
and no constructor argument to back it with the injected `session_service`.

### Steps to Reproduce

1. Run the ADK middleware in **two** processes sharing one `DatabaseSessionService`,
each constructing its own `SessionManager` (i.e. any horizontally scaled
deployment). Front them with a load balancer that has no session affinity.
2. Send turn 1 for a thread to process A. It answers and marks the message
processed in A's memory.
3. Send turn 2 for the *same* thread to process B. The client re-sends the full
history (#2186).
4. Observe B re-execute turn 1's user message, then turn 2 — two
`_start_new_execution` calls under a single `run_id`.

Use a turn that requires **no tool calls** to see it reliably, per the heuristic above.

### Expected Behavior

A given user message is executed exactly once per thread, regardless of which
process serves the turn. Which replica handles a request should not change what
work is performed.

Concretely: `_get_unseen_messages()` should reach the same verdict on a cold
process as on a warm one — either because the ledger is durable alongside the
session, or because "already handled" is derived from the durable session rather
than from process-local state.

### Environment

```text
AG-UI package(s) & version(s):
ag-ui-adk 0.7.0 (current PyPI release; main is unchanged in the relevant code)
ag-ui-protocol 0.1.15
google-adk 2.7.0
Runtime: Python 3.13
Session service: DatabaseSessionService (PostgreSQL)
SessionManager: use_thread_id_as_session_id=True, one shared instance per process,
injected via ADKAgent.from_app(session_manager=...)
Deployment: 2 replicas behind a load balancer with no session affinity
```

### Screenshots

_No response_

### Logs & Errors

```shell
One thread over ~40 minutes, 2 replicas, no affinity. [EXEC] and [SESSION_DEBUG]
are the library's own INFO logs. Pod names and ids anonymized.

# pod run_id session events at exec start
1 A run-1 0
2 B <- switch run-2 31
3 B run-2 (DUPLICATE) 56
4 B (same pod) run-3 81
5 A <- switch run-4 106
6 A run-4 (DUPLICATE) 131
7 B <- switch run-5 156
8 B run-5 (DUPLICATE) 181

[EXEC] NEW_RUN - thread=, run=run-4, tool_results=[], message_batch_len=1
[SESSION_DEBUG] Session has 106 events
[EXEC] NEW_RUN - thread=, run=run-4, tool_results=[], message_batch_len=1
[SESSION_DEBUG] Session has 131 events

- The duplicate carries the SAME run_id: one HTTP request producing two
sequential _start_new_execution calls, not a client retry.
- Duplicates correlate 1:1 with the pod changing between turns. Consecutive
turns on the same pod never duplicate.
- Session event count rises monotonically across BOTH pods, which confirms the
session store is consistent. The only un-shared state is the ledger.
```

### Additional Context

**Suggested directions**

1. Give the ledger the same durability as the session — persist the processed ids in
the ADK session, or accept a pluggable store on `SessionManager`. Note
`mark_messages_processed` / `get_processed_message_ids` are currently sync, so a
persistent backing likely needs them async or hydrated at run entry.
2. Better: derive "already handled" from the durable session instead of a side ledger,
so there is no second source of truth to keep in sync.
3. Short term, document it. A note that `SessionManager` is process-affine and that
multi-replica deployments need sticky sessions.

**Relation to #2186.** This is evidence for that issue's options 1/2 (ack-based
sending, or a `sinceMessageId` cursor) over option 3 (`messageFilter`): the first two
remove the re-sent history a cold process can misread as new, whereas a client-side
filter — in #2186's own words — "has no notion of what the server already has", so the
replay survives it.

**Workaround for anyone hitting this before a fix.** Enable sticky sessions, or seed
the ledger at request entry: before `adk_agent.run(...)`, call
`mark_messages_processed()` for every message id up to and including the last
`assistant` message, leaving the trailing segment (the new user message, or a HITL
resume's tool results) untouched. No-op on a warm process. The `app_name` must match
`ADKAgent._get_app_name()` for the same run — the `App.name` passed to `from_app` —
or it silently does nothing.

**Possibly related, found while investigating:** `adk_agent.py` has no handling for the
`activity` or `reasoning` message roles (grep returns nothing). In the run loop they
are neither `tool` nor `assistant`, so they fall through to
`message_batch.append(candidate)` and are fed to the model as if they were user input.
Low reachability — the JS client strips `activity` per #2186 — but a client echoing
`reasoning` history back would hit it. Happy to split this into its own issue if you'd
prefer.