fix: pre-merge stateDelta before onUserMessageCallback in Runner

Previously, stateDelta passed to runAsync was merged into the session
only after pluginManager.onUserMessageCallback was invoked, so plugins
observed a stale session state during the user-message callback.

PiperOrigin-RevId: 921350398

Author: damianmomotgoogle

core/src/main/java/com/google/adk/runner/Runner.java

@@ -485,6 +485,12 @@ protected Flowable<Event> runAsyncImpl(
               BaseAgent rootAgent = this.agent;
               String invocationId = InvocationContext.newInvocationContextId();
 
+              // Pre-merge stateDelta so onUserMessageCallback can access it.
+              // Safe: session is a copy; persistence still happens via appendNewMessageToSession.
+              if (stateDelta != null && !stateDelta.isEmpty()) {
+                stateDelta.forEach((key, value) -> session.state().put(key, value));
+              }
+
               // Create initial context
               InvocationContext initialContext =
                   newInvocationContextBuilder(session)

core/src/test/java/com/google/adk/runner/RunnerTest.java

@@ -1122,6 +1122,42 @@ public void beforeRunCallback_withStateDelta_seesMergedState() {
     assertThat(sessionInCallback.state()).containsEntry("number", 123);
   }
 
+  @Test
+  public void onUserMessageCallback_withStateDelta_seesMergedState() {
+    // Snapshot the session state *inside* the callback, otherwise the assertion would
+    // observe the post-runAsync state which is mutated by appendEvent regardless of whether
+    // the pre-merge in Runner is applied.
+    AtomicReference<ConcurrentHashMap<String, Object>> stateInCallback = new AtomicReference<>();
+    when(plugin.onUserMessageCallback(any(), any()))
+        .thenAnswer(
+            invocation -> {
+              InvocationContext ctx = invocation.getArgument(0);
+              stateInCallback.set(new ConcurrentHashMap<>(ctx.session().state()));
+              return Maybe.empty();
+            });
+
+    ImmutableMap<String, Object> stateDelta =
+        ImmutableMap.of("callback_key", "callback_value", "number", 123);
+
+    var unused =
+        runner
+            .runAsync(
+                "user",
+                session.id(),
+                createContent("test with state"),
+                RunConfig.builder().build(),
+                stateDelta)
+            .toList()
+            .blockingGet();
+
+    // Verify onUserMessageCallback was called
+    verify(plugin).onUserMessageCallback(any(), any());
+
+    // Verify state delta was merged before onUserMessageCallback was invoked
+    assertThat(stateInCallback.get()).containsEntry("callback_key", "callback_value");
+    assertThat(stateInCallback.get()).containsEntry("number", 123);
+  }
+
   @Test
   public void runAsync_ensureEventsAreAppendedInOrder() throws Exception {
     Event event1 = TestUtils.createEvent("1");