feat: support pausing and resuming a session invocation

When enabled via ResumabilityConfig, an invocation that issues a long-running
function call pauses and can be resumed later from durable per-agent checkpoints
for LlmAgent, SequentialAgent, LoopAgent, and ParallelAgent, via a runAsync
overload that takes an invocation id. A resumed transfer continues at the
transferred-to sub-agent; a call a previous run never executed is replayed rather
than re-requested from the model; and resuming persists the incoming message and
any state delta before running. Behavior is unchanged unless resumability is
enabled.

PiperOrigin-RevId: 973694952

Author: damianmomotgoogle

core/src/main/java/com/google/adk/agents/BaseAgent.java

@@ -16,13 +16,15 @@
 
 package com.google.adk.agents;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Strings.isNullOrEmpty;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.lang.String.format;
 
 import com.google.adk.agents.Callbacks.AfterAgentCallback;
 import com.google.adk.agents.Callbacks.BeforeAgentCallback;
 import com.google.adk.events.Event;
+import com.google.adk.events.EventActions;
 import com.google.adk.plugins.Plugin;
 import com.google.adk.telemetry.Instrumentation;
 import com.google.adk.telemetry.Instrumentation.AgentInvocation;
@@ -39,6 +41,7 @@
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
 import java.util.regex.Pattern;
@@ -136,10 +139,8 @@ private static void validateAgentName(String name) {
       throw new IllegalArgumentException(
           format("Agent name '%s' does not match regex '%s'.", name, IDENTIFIER_REGEX));
     }
-    if (name.equals(Role.USER)) {
-      throw new IllegalArgumentException(
-          "Agent name cannot be 'user'; reserved for end-user input.");
-    }
+    checkArgument(
+        !name.equals(Role.USER), "Agent name cannot be 'user'; reserved for end-user input.");
   }
 
   /**
@@ -468,6 +469,44 @@ public Flowable<Event> runLive(InvocationContext parentContext) {
     return run(parentContext, this::runLiveImpl);
   }
 
+  /**
+   * Records this agent's end-of-agent checkpoint and returns it as a single-event stream. The
+   * recorded state is cleared and the agent marked finished, so a later run skips it.
+   *
+   * @param context Current invocation context.
+   * @return stream of the single {@code endOfAgent = true} checkpoint event.
+   */
+  final Flowable<Event> endOfAgentAndRecord(InvocationContext context) {
+    context.setAgentState(name(), /* agentState= */ null, /* endOfAgent= */ true);
+    return Flowable.just(checkpointEvent(context, EventActions.builder().endOfAgent(true).build()));
+  }
+
+  /**
+   * Records {@code agentState} for this agent and returns the matching checkpoint event as a
+   * single-event stream. The agent is left unfinished, so a later run resumes from this checkpoint.
+   *
+   * @param context Current invocation context.
+   * @param agentState The serialized agent state to persist.
+   * @return stream of the single checkpoint event carrying {@code agentState}.
+   */
+  final Flowable<Event> checkpointAndRecord(
+      InvocationContext context, Map<String, Object> agentState) {
+    context.setAgentState(name(), agentState, /* endOfAgent= */ false);
+    return Flowable.just(
+        checkpointEvent(context, EventActions.builder().agentState(agentState).build()));
+  }
+
+  /** Builds a resumability checkpoint event authored by this agent carrying {@code actions}. */
+  private Event checkpointEvent(InvocationContext context, EventActions actions) {
+    return Event.builder()
+        .id(Event.generateEventId())
+        .invocationId(context.invocationId())
+        .author(name())
+        .branch(context.branch().orElse(null))
+        .actions(actions)
+        .build();
+  }
+
   /**
    * Agent-specific asynchronous logic.
    *