fix: wait for the Runner to persist a step's events before the ADK flow's next step (sequential-tool-execution race)

[copybara-service]

fix: wait for the Runner to persist a step's events before the ADK flow's next step (sequential-tool-execution race)

BaseLlmFlow.run() builds each step's request from the session, but the
Runner persists events asynchronously downstream of the flow. A slow
appendEvent could let the next step start from a stale session missing the
prior step's events (e.g. a tool's function response), making the model re-call
the tool or hallucinate its result.

The Runner stays the sole appendEvent caller and the flow waits: it calls
PersistBarrier.markPersisted(id) after each append (or markFailed(id, error)
if it fails), and the flow calls PersistBarrier.awaitPersisted(stepEvents)
between steps. The barrier is a reactive per-event signal in the shared
InvocationContext.callbackContextData and never blocks a thread; Contents is
unchanged. PersistBarrier.enable(), called by the Runner, keeps
awaitPersisted a no-op when the flow runs without a Runner.

Each event id maps to a CompletableSubject: pending until its append finishes,
then terminally completed or failed. The subject retains its terminal state, so
awaitPersisted/mark* may happen in any order and a late await -- e.g. at a
higher flow level across an agent transfer -- resolves immediately. If an append
fails, the matching await fails rather than blocking forever. It is thread-safe
and lock-free: markPersisted/markFailed may run off-thread when an async
appendEvent completes, and ConcurrentHashMap.computeIfAbsent hands both sides
the same subject.