fix: propagate A2A request metadata into the run config in AgentExecutor

During A2A communication, `AgentExecutor` invoked the 4-argument
`Runner.runAsync`, which hardcodes `stateDelta = null` and silently dropped the
caller's incoming request metadata.

Route the incoming request metadata (`RequestContext.getParams().metadata()`)
into `RunConfig.customMetadata` under the `a2a_metadata` key, so downstream
processing can read it. This mirrors ADK Python and Go, which pass A2A request
metadata through the run config and leave the session state delta null, rather
than writing framework IDs into session state. Add tests for the metadata and
no-metadata cases.

PiperOrigin-RevId: 945055254

Author: MiloszSobczyk

a2a/src/main/java/com/google/adk/a2a/executor/AgentExecutor.java

@@ -20,6 +20,7 @@
 import com.google.adk.a2a.converters.EventConverter;
 import com.google.adk.a2a.converters.PartConverter;
 import com.google.adk.agents.BaseAgent;
+import com.google.adk.agents.RunConfig;
 import com.google.adk.apps.App;
 import com.google.adk.artifacts.BaseArtifactService;
 import com.google.adk.events.Event;
@@ -38,6 +39,7 @@
 import io.a2a.spec.Artifact;
 import io.a2a.spec.InvalidAgentResponseError;
 import io.a2a.spec.Message;
+import io.a2a.spec.MessageSendParams;
 import io.a2a.spec.Part;
 import io.a2a.spec.TaskArtifactUpdateEvent;
 import io.a2a.spec.TaskState;
@@ -63,6 +65,7 @@
 public class AgentExecutor implements io.a2a.server.agentexecution.AgentExecutor {
   private static final Logger logger = LoggerFactory.getLogger(AgentExecutor.class);
   private static final String USER_ID_PREFIX = "A2A_USER_";
+  private static final String A2A_METADATA_KEY = "a2a_metadata";
   private final Map<String, Disposable> activeTasks = new ConcurrentHashMap<>();
   private final Runner.Builder runnerBuilder;
   private final AgentExecutorConfig agentExecutorConfig;
@@ -217,7 +220,7 @@ public void execute(RequestContext ctx, EventQueue eventQueue) {
                                 getUserId(ctx),
                                 session.id(),
                                 content,
-                                agentExecutorConfig.runConfig());
+                                runConfigWithA2aMetadata(ctx));
                           });
                 })
             .concatMap(
@@ -273,6 +276,23 @@ private String getUserId(RequestContext ctx) {
     return USER_ID_PREFIX + ctx.getContextId();
   }
 
+  /**
+   * Returns the configured run config enriched with the caller's incoming A2A request metadata
+   * under the {@code a2a_metadata} key, so downstream processing can read it via {@link
+   * RunConfig#customMetadata()}.
+   */
+  private RunConfig runConfigWithA2aMetadata(RequestContext ctx) {
+    RunConfig runConfig = agentExecutorConfig.runConfig();
+    MessageSendParams params = ctx.getParams();
+    Map<String, Object> requestMetadata = params == null ? null : params.metadata();
+    if (requestMetadata == null || requestMetadata.isEmpty()) {
+      return runConfig;
+    }
+    Map<String, Object> customMetadata = new HashMap<>(runConfig.customMetadata());
+    customMetadata.put(A2A_METADATA_KEY, requestMetadata);
+    return runConfig.toBuilder().customMetadata(customMetadata).build();
+  }
+
   private Maybe<Session> prepareSession(
       RequestContext ctx, String appName, BaseSessionService service) {
     return service

a2a/src/test/java/com/google/adk/a2a/executor/AgentExecutorTest.java

@@ -12,6 +12,7 @@
 
 import com.google.adk.agents.BaseAgent;
 import com.google.adk.agents.InvocationContext;
+import com.google.adk.agents.RunConfig;
 import com.google.adk.apps.App;
 import com.google.adk.artifacts.InMemoryArtifactService;
 import com.google.adk.events.Event;
@@ -24,6 +25,7 @@
 import io.a2a.server.agentexecution.RequestContext;
 import io.a2a.server.events.EventQueue;
 import io.a2a.spec.Message;
+import io.a2a.spec.MessageSendParams;
 import io.a2a.spec.TaskArtifactUpdateEvent;
 import io.a2a.spec.TaskState;
 import io.a2a.spec.TaskStatus;
@@ -342,6 +344,62 @@ public void execute_runnerSucceeds_registerCompletedTaskFails_noFailedTaskRegist
     assertThat(statusEvents).isEmpty();
   }
 
+  @Test
+  public void execute_propagatesRequestMetadataIntoRunConfig() {
+    testAgent.setEventsToEmit(Flowable.empty());
+    AgentExecutor executor =
+        new AgentExecutor.Builder()
+            .agentExecutorConfig(AgentExecutorConfig.builder().build())
+            .app(App.builder().name("test_app").rootAgent(testAgent).build())
+            .sessionService(new InMemorySessionService())
+            .artifactService(new InMemoryArtifactService())
+            .build();
+
+    Message message =
+        new Message.Builder()
+            .messageId("msg-1")
+            .role(Message.Role.USER)
+            .parts(ImmutableList.of(new TextPart("trigger")))
+            .build();
+    MessageSendParams params =
+        new MessageSendParams.Builder()
+            .message(message)
+            .metadata(ImmutableMap.of("key", "value"))
+            .build();
+    RequestContext ctx = mock(RequestContext.class);
+    when(ctx.getMessage()).thenReturn(message);
+    when(ctx.getTaskId()).thenReturn("task-1");
+    when(ctx.getContextId()).thenReturn("ctx-1");
+    when(ctx.getParams()).thenReturn(params);
+
+    executor.execute(ctx, eventQueue);
+
+    // The runner passes the enriched run config down to the agent's invocation context.
+    RunConfig runConfig = testAgent.lastInvocationContext.runConfig();
+    assertThat(runConfig.customMetadata())
+        .containsEntry("a2a_metadata", ImmutableMap.of("key", "value"));
+  }
+
+  @Test
+  public void execute_withoutRequestMetadata_leavesRunConfigCustomMetadataEmpty() {
+    testAgent.setEventsToEmit(Flowable.empty());
+    AgentExecutor executor =
+        new AgentExecutor.Builder()
+            .agentExecutorConfig(AgentExecutorConfig.builder().build())
+            .app(App.builder().name("test_app").rootAgent(testAgent).build())
+            .sessionService(new InMemorySessionService())
+            .artifactService(new InMemoryArtifactService())
+            .build();
+
+    // createRequestContext() does not stub getParams(), mirroring a request with no metadata.
+    RequestContext ctx = createRequestContext();
+
+    executor.execute(ctx, eventQueue);
+
+    RunConfig runConfig = testAgent.lastInvocationContext.runConfig();
+    assertThat(runConfig.customMetadata()).doesNotContainKey("a2a_metadata");
+  }
+
   private RequestContext createRequestContext() {
     Message message =
         new Message.Builder()
@@ -507,6 +565,7 @@ public void execute_withDefaultArtifactPerRun_emitsMessageAndLastChunk() {
 
   private static final class TestAgent extends BaseAgent {
     private Flowable<Event> eventsToEmit;
+    private volatile InvocationContext lastInvocationContext;
 
     TestAgent() {
       this(Flowable.empty());
@@ -524,6 +583,7 @@ void setEventsToEmit(Flowable<Event> events) {
 
     @Override
     protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
+      this.lastInvocationContext = invocationContext;
       return eventsToEmit;
     }