diff --git a/core/src/main/java/com/google/adk/agents/BaseAgent.java b/core/src/main/java/com/google/adk/agents/BaseAgent.java
index fc1f0f31e..e5deac934 100644
--- a/core/src/main/java/com/google/adk/agents/BaseAgent.java
+++ b/core/src/main/java/com/google/adk/agents/BaseAgent.java
@@ -23,6 +23,7 @@
 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;
@@ -38,6 +39,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;
@@ -459,6 +461,43 @@ public Flowable<Event> runLive(InvocationContext parentContext) {
     return run(parentContext, this::runLiveImpl);
   }
 
+  /**
+   * Builds an end-of-agent checkpoint event for a resumable invocation, marking that this agent
+   * finished its run. Emitting it (and recording it via {@link InvocationContext#setAgentState})
+   * lets a later run skip a completed agent.
+   *
+   * @param context Current invocation context.
+   * @return an event whose actions carry {@code endOfAgent = true}.
+   */
+  protected Event endOfAgentEvent(InvocationContext context) {
+    return Event.builder()
+        .id(Event.generateEventId())
+        .invocationId(context.invocationId())
+        .author(name())
+        .branch(context.branch().orElse(null))
+        .actions(EventActions.builder().endOfAgent(true).build())
+        .build();
+  }
+
+  /**
+   * Builds a checkpoint event carrying this agent's serialized state for a resumable invocation.
+   * Callers also record the state via {@link InvocationContext#setAgentState} so a later run
+   * resumes at the right point.
+   *
+   * @param context Current invocation context.
+   * @param agentState The serialized agent state to persist.
+   * @return an event whose actions carry {@code agentState}.
+   */
+  protected Event createStateEvent(InvocationContext context, Map<String, Object> agentState) {
+    return Event.builder()
+        .id(Event.generateEventId())
+        .invocationId(context.invocationId())
+        .author(name())
+        .branch(context.branch().orElse(null))
+        .actions(EventActions.builder().agentState(agentState).build())
+        .build();
+  }
+
   /**
    * Agent-specific asynchronous logic.
    *
diff --git a/core/src/main/java/com/google/adk/agents/InvocationContext.java b/core/src/main/java/com/google/adk/agents/InvocationContext.java
index 456758b95..3ad5a5524 100644
--- a/core/src/main/java/com/google/adk/agents/InvocationContext.java
+++ b/core/src/main/java/com/google/adk/agents/InvocationContext.java
@@ -20,6 +20,7 @@
 
 import com.google.adk.apps.ResumabilityConfig;
 import com.google.adk.artifacts.BaseArtifactService;
+import com.google.adk.events.Event;
 import com.google.adk.memory.BaseMemoryService;
 import com.google.adk.models.LlmCallsLimitExceededException;
 import com.google.adk.plugins.Plugin;
@@ -27,11 +28,19 @@
 import com.google.adk.sessions.BaseSessionService;
 import com.google.adk.sessions.Session;
 import com.google.adk.summarizer.EventsCompactionConfig;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.genai.types.Content;
+import com.google.genai.types.FunctionCall;
+import com.google.genai.types.FunctionResponse;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -56,6 +65,10 @@ public class InvocationContext {
   private final @Nullable ResumabilityConfig resumabilityConfig;
   private final InvocationCostManager invocationCostManager;
   private final Map<String, Object> callbackContextData;
+  // Resumability checkpoints, shared by reference across derived contexts so a sub-agent's
+  // checkpoint is visible to its parent and the runner.
+  private final Map<String, Map<String, Object>> agentStates;
+  private final Map<String, Boolean> endOfAgents;
 
   @Nullable private String branch;
   private BaseAgent agent;
@@ -83,6 +96,8 @@ protected InvocationContext(Builder builder) {
     // invocation invocation so that Plugins can access the same data it during the invocation
     // across all types of callbacks.
     this.callbackContextData = builder.callbackContextData;
+    this.agentStates = builder.agentStates;
+    this.endOfAgents = builder.endOfAgents;
   }
 
   /** Returns a new {@link Builder} for creating {@link InvocationContext} instances. */
@@ -230,6 +245,150 @@ public boolean isResumable() {
     return resumabilityConfig != null && resumabilityConfig.isResumable();
   }
 
+  /**
+   * Returns the per-agent resumability checkpoint states for this invocation, keyed by agent name.
+   * Shared by reference across derived contexts within the invocation.
+   */
+  public Map<String, Map<String, Object>> agentStates() {
+    return agentStates;
+  }
+
+  /** Returns the per-agent end-of-agent flags for this invocation, keyed by agent name. */
+  public Map<String, Boolean> endOfAgents() {
+    return endOfAgents;
+  }
+
+  /**
+   * Sets the checkpoint state of an agent explicitly. Does not implicitly initialize.
+   *
+   * @param agentName the agent whose state to set.
+   * @param agentState the serialized agent state to store; ignored when {@code endOfAgent} is true.
+   * @param endOfAgent when true, marks the agent finished and drops any stored state.
+   */
+  void setAgentState(
+      String agentName, @Nullable Map<String, Object> agentState, boolean endOfAgent) {
+    if (endOfAgent) {
+      endOfAgents.put(agentName, true);
+      agentStates.remove(agentName);
+    } else if (agentState != null) {
+      agentStates.put(agentName, agentState);
+      endOfAgents.put(agentName, false);
+    } else {
+      endOfAgents.remove(agentName);
+      agentStates.remove(agentName);
+    }
+  }
+
+  /** Recursively resets the checkpoint state of all sub-agents of the given agent. */
+  void resetSubAgentStates(String agentName) {
+    Optional<BaseAgent> target = agent.findAgent(agentName);
+    if (target.isEmpty()) {
+      return;
+    }
+    for (BaseAgent subAgent : target.get().subAgents()) {
+      setAgentState(subAgent.name(), null, false);
+      resetSubAgentStates(subAgent.name());
+    }
+  }
+
+  /**
+   * Rehydrates {@link #agentStates()} and {@link #endOfAgents()} from the current invocation's
+   * history when this invocation is resumable. For each event carrying agent-state information,
+   * sets the authoring agent's checkpoint; for a non-workflow author that already produced content,
+   * seeds an empty state so it is treated as mid-run.
+   */
+  public void populateInvocationAgentStates() {
+    if (!isResumable()) {
+      return;
+    }
+    for (Event event : getEvents(/* currentInvocation= */ true, /* currentBranch= */ false)) {
+      String author = event.author();
+      if (author == null) {
+        continue;
+      }
+      Optional<Map<String, Object>> agentState = event.actions().agentState();
+      if (event.actions().endOfAgent()) {
+        endOfAgents.put(author, true);
+        agentStates.remove(author);
+      } else if (agentState.isPresent()) {
+        agentStates.put(author, agentState.get());
+        endOfAgents.put(author, false);
+      } else if (!author.equals("user")
+          && event.content().isPresent()
+          && !agentStates.containsKey(author)) {
+        agentStates.put(author, new HashMap<>());
+        endOfAgents.put(author, false);
+      }
+    }
+  }
+
+  /**
+   * Returns the current session's events, optionally filtered to the current invocation and/or the
+   * current branch. Reads the in-memory {@link Session#events()} list, which {@link
+   * BaseSessionService#appendEvent} keeps in sync. A {@code null}-branch event is visible on any
+   * branch.
+   *
+   * @param currentInvocation whether to filter to events from this invocation.
+   * @param currentBranch whether to filter to events on this branch (or with no branch).
+   */
+  List<Event> getEvents(boolean currentInvocation, boolean currentBranch) {
+    List<Event> results = new ArrayList<>(session.events());
+    if (currentInvocation) {
+      results.removeIf(event -> !invocationId.equals(event.invocationId()));
+    }
+    if (currentBranch) {
+      results.removeIf(
+          event -> event.branch().isPresent() && !event.branch().get().equals(this.branch));
+    }
+    return results;
+  }
+
+  /**
+   * Returns whether to pause the invocation right after this event. Pausing (unlike ending) leaves
+   * the invocation resumable. Both conditions must hold: the app is {@link #isResumable()} and the
+   * event carries a long-running function call (including a synthetic {@code
+   * adk_request_confirmation} HITL request).
+   */
+  boolean shouldPauseInvocation(Event event) {
+    if (!isResumable()) {
+      return false;
+    }
+    Set<String> longRunningIds = event.longRunningToolIds().orElse(ImmutableSet.of());
+    if (longRunningIds.isEmpty()) {
+      return false;
+    }
+    ImmutableList<FunctionCall> functionCalls = event.functionCalls();
+    if (functionCalls.isEmpty()) {
+      return false;
+    }
+    return functionCalls.stream()
+        .anyMatch(call -> call.id().isPresent() && longRunningIds.contains(call.id().get()));
+  }
+
+  /**
+   * Finds the current-invocation event whose function call matches the first function response id
+   * in {@code functionResponseEvent}, searching newest-first.
+   */
+  Optional<Event> findMatchingFunctionCall(Event functionResponseEvent) {
+    ImmutableList<FunctionResponse> responses = functionResponseEvent.functionResponses();
+    if (responses.isEmpty()) {
+      return Optional.empty();
+    }
+    Optional<String> targetId = responses.get(0).id();
+    if (targetId.isEmpty()) {
+      return Optional.empty();
+    }
+    List<Event> events = getEvents(/* currentInvocation= */ true, /* currentBranch= */ false);
+    for (int i = events.size() - 1; i >= 0; i--) {
+      for (FunctionCall call : events.get(i).functionCalls()) {
+        if (call.id().isPresent() && call.id().get().equals(targetId.get())) {
+          return Optional.of(events.get(i));
+        }
+      }
+    }
+    return Optional.empty();
+  }
+
   private static class InvocationCostManager {
     private final AtomicInteger numberOfLlmCalls = new AtomicInteger(0);
 
@@ -289,6 +448,10 @@ private Builder(InvocationContext context) {
       // invocation invocation so that Plugins can access the same data it during the invocation
       // across all types of callbacks.
       this.callbackContextData = context.callbackContextData;
+      // Shared by reference so a sub-agent's checkpoint is visible to its parent and the runner
+      // within one invocation.
+      this.agentStates = context.agentStates;
+      this.endOfAgents = context.endOfAgents;
     }
 
     private BaseSessionService sessionService;
@@ -309,6 +472,8 @@ private Builder(InvocationContext context) {
     private @Nullable ResumabilityConfig resumabilityConfig;
     private InvocationCostManager invocationCostManager = new InvocationCostManager();
     private Map<String, Object> callbackContextData = new ConcurrentHashMap<>();
+    private Map<String, Map<String, Object>> agentStates = new ConcurrentHashMap<>();
+    private Map<String, Boolean> endOfAgents = new ConcurrentHashMap<>();
 
     /**
      * Sets the session service for managing session state.
diff --git a/core/src/main/java/com/google/adk/agents/LlmAgent.java b/core/src/main/java/com/google/adk/agents/LlmAgent.java
index fa754e0c0..011432f1b 100644
--- a/core/src/main/java/com/google/adk/agents/LlmAgent.java
+++ b/core/src/main/java/com/google/adk/agents/LlmAgent.java
@@ -70,6 +70,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
@@ -660,7 +661,93 @@ private static boolean isThought(Part part) {
 
   @Override
   protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
-    return llmFlow.run(invocationContext).doOnNext(this::maybeSaveOutputToState);
+    if (!invocationContext.isResumable()) {
+      return llmFlow.run(invocationContext).doOnNext(this::maybeSaveOutputToState);
+    }
+    return Flowable.defer(
+        () -> {
+          // Resumed after a transfer: continue the transferred sub-agent instead of re-invoking
+          // the model, then mark this agent done.
+          if (invocationContext.agentStates().containsKey(name())) {
+            Optional<BaseAgent> resumeTarget = getSubagentToResume(invocationContext);
+            if (resumeTarget.isPresent()) {
+              return resumeTarget
+                  .get()
+                  .runAsync(invocationContext)
+                  .concatWith(
+                      Flowable.defer(
+                          () -> {
+                            invocationContext.setAgentState(name(), /* agentState= */ null, true);
+                            return Flowable.just(endOfAgentEvent(invocationContext));
+                          }));
+            }
+          }
+          // Normal path: emit an end-of-agent checkpoint on completion so a later run can skip
+          // this agent, unless it paused on a long-running call (then suppress it so it can
+          // resume).
+          Flowable<Event> events =
+              llmFlow.run(invocationContext).doOnNext(this::maybeSaveOutputToState);
+          AtomicBoolean paused = new AtomicBoolean(false);
+          return events
+              .doOnNext(
+                  event -> {
+                    if (invocationContext.shouldPauseInvocation(event)) {
+                      paused.set(true);
+                    }
+                  })
+              .concatWith(
+                  Flowable.defer(
+                      () -> {
+                        if (paused.get() || invocationContext.agent() != this) {
+                          return Flowable.empty();
+                        }
+                        invocationContext.setAgentState(name(), /* agentState= */ null, true);
+                        return Flowable.just(endOfAgentEvent(invocationContext));
+                      }));
+        });
+  }
+
+  private Optional<BaseAgent> getTransferToAgentOrNull(Event event, String fromAgent) {
+    if (fromAgent.equals(event.author())) {
+      Optional<String> transferTo = event.actions().transferToAgent();
+      if (transferTo.isPresent() && !transferTo.get().equals(fromAgent)) {
+        return rootAgent().findAgent(transferTo.get());
+      }
+    }
+    return Optional.empty();
+  }
+
+  /**
+   * When this agent is being resumed, returns the sub-agent it had transferred to (so the resume
+   * continues that sub-agent), or empty when this agent should continue itself.
+   */
+  private Optional<BaseAgent> getSubagentToResume(InvocationContext context) {
+    List<Event> events =
+        context.getEvents(/* currentInvocation= */ true, /* currentBranch= */ true);
+    if (events.isEmpty()) {
+      return Optional.empty();
+    }
+    Event lastEvent = events.get(events.size() - 1);
+    if (name().equals(lastEvent.author())) {
+      return getTransferToAgentOrNull(lastEvent, name());
+    }
+    if (Objects.equals(lastEvent.author(), "user")) {
+      Optional<Event> functionCallEvent = context.findMatchingFunctionCall(lastEvent);
+      if (functionCallEvent.isEmpty()) {
+        throw new IllegalArgumentException(
+            "No matching function call to resume agent " + name() + " from a function response.");
+      }
+      if (name().equals(functionCallEvent.get().author())) {
+        return Optional.empty();
+      }
+    }
+    for (int i = events.size() - 2; i >= 0; i--) {
+      Optional<BaseAgent> agent = getTransferToAgentOrNull(events.get(i), name());
+      if (agent.isPresent()) {
+        return agent;
+      }
+    }
+    return Optional.empty();
   }
 
   @Override
diff --git a/core/src/main/java/com/google/adk/agents/LoopAgent.java b/core/src/main/java/com/google/adk/agents/LoopAgent.java
index 19fd4c497..4c362e62c 100644
--- a/core/src/main/java/com/google/adk/agents/LoopAgent.java
+++ b/core/src/main/java/com/google/adk/agents/LoopAgent.java
@@ -18,9 +18,11 @@
 
 import com.google.adk.agents.ConfigAgentUtils.ConfigurationException;
 import com.google.adk.events.Event;
+import com.google.common.collect.ImmutableMap;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.reactivex.rxjava3.core.Flowable;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.jspecify.annotations.Nullable;
@@ -149,29 +151,105 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
           .takeUntil(LoopAgent::hasEscalateAction);
     }
 
-    // Resumable: stop looping once a sub-agent emits a pending long-running call (e.g. HITL),
-    // matching Python ADK v1 and avoiding a runaway loop. The current sub-agent still finishes;
-    // resuming into the paused iteration needs persisted state (future work).
-    AtomicBoolean paused = new AtomicBoolean(false);
-    AtomicInteger timesLooped = new AtomicInteger(0);
-    return Flowable.fromIterable(subAgents)
-        .concatMap(
-            subAgent ->
-                paused.get()
-                    ? Flowable.empty()
-                    : subAgent
-                        .runAsync(invocationContext)
-                        .doOnNext(
-                            event -> {
-                              if (WorkflowAgentResumption.hasPendingLongRunningCall(event)) {
-                                paused.set(true);
-                              }
-                            }))
-        .repeatUntil(
-            () ->
-                paused.get()
-                    || (maxIterations != null && timesLooped.incrementAndGet() >= maxIterations))
-        .takeUntil(LoopAgent::hasEscalateAction);
+    // Resumable: checkpoint {current_sub_agent, times_looped} before each sub-agent, resume into
+    // the checkpointed iteration, pause (not end) on a long-running call, and reset sub-agent
+    // state between iterations.
+    return Flowable.defer(
+        () -> {
+          Map<String, Object> state = invocationContext.agentStates().get(name());
+          String startSubAgentName =
+              state == null ? null : (String) state.get(WorkflowAgentStates.CURRENT_SUB_AGENT);
+          int startTimesLooped =
+              state == null || state.get(WorkflowAgentStates.TIMES_LOOPED) == null
+                  ? 0
+                  : ((Number) state.get(WorkflowAgentStates.TIMES_LOOPED)).intValue();
+          int startIndex =
+              WorkflowAgentStates.findIndexForResumption(subAgents, startSubAgentName, logger);
+          AtomicInteger timesLooped = new AtomicInteger(startTimesLooped);
+          AtomicBoolean shouldExit = new AtomicBoolean(false);
+          AtomicBoolean paused = new AtomicBoolean(false);
+          if (maxIterations != null && startTimesLooped >= maxIterations) {
+            return Flowable.just(endOfAgentEvent(invocationContext));
+          }
+          return runLoopIteration(
+              invocationContext,
+              subAgents,
+              startIndex,
+              new AtomicBoolean(startSubAgentName != null),
+              timesLooped,
+              shouldExit,
+              paused);
+        });
+  }
+
+  /**
+   * Runs one loop iteration over the sub-agents from {@code startIndex}, then either recurses for
+   * the next iteration or terminates (emitting end-of-agent unless paused). Shared holders carry
+   * the loop's mutable state across iterations.
+   */
+  private Flowable<Event> runLoopIteration(
+      InvocationContext context,
+      List<? extends BaseAgent> subAgents,
+      int startIndex,
+      AtomicBoolean resumingFirst,
+      AtomicInteger timesLooped,
+      AtomicBoolean shouldExit,
+      AtomicBoolean paused) {
+    Flowable<Event> iteration =
+        Flowable.fromIterable(subAgents.subList(startIndex, subAgents.size()))
+            .concatMap(
+                subAgent ->
+                    Flowable.defer(
+                        () -> {
+                          if (shouldExit.get() || paused.get()) {
+                            return Flowable.<Event>empty();
+                          }
+                          Flowable<Event> checkpoint = Flowable.empty();
+                          if (!resumingFirst.getAndSet(false)) {
+                            ImmutableMap<String, Object> subState =
+                                ImmutableMap.of(
+                                    WorkflowAgentStates.CURRENT_SUB_AGENT, subAgent.name(),
+                                    WorkflowAgentStates.TIMES_LOOPED, timesLooped.get());
+                            context.setAgentState(name(), subState, /* endOfAgent= */ false);
+                            checkpoint = Flowable.just(createStateEvent(context, subState));
+                          }
+                          Flowable<Event> run =
+                              subAgent
+                                  .runAsync(context)
+                                  .doOnNext(
+                                      event -> {
+                                        if (hasEscalateAction(event)) {
+                                          shouldExit.set(true);
+                                        }
+                                        if (context.shouldPauseInvocation(event)) {
+                                          paused.set(true);
+                                        }
+                                      });
+                          return checkpoint.concatWith(run);
+                        }));
+    return iteration.concatWith(
+        Flowable.defer(
+            () -> {
+              if (paused.get()) {
+                return Flowable.<Event>empty();
+              }
+              if (shouldExit.get()) {
+                return Flowable.just(endOfAgentEvent(context));
+              }
+              int looped = timesLooped.incrementAndGet();
+              context.resetSubAgentStates(name());
+              if (maxIterations != null && looped >= maxIterations) {
+                return Flowable.just(endOfAgentEvent(context));
+              }
+              return runLoopIteration(
+                  context,
+                  subAgents,
+                  /* startIndex= */ 0,
+                  new AtomicBoolean(false),
+                  timesLooped,
+                  shouldExit,
+                  paused);
+            }));
   }
 
   @Override
diff --git a/core/src/main/java/com/google/adk/agents/ParallelAgent.java b/core/src/main/java/com/google/adk/agents/ParallelAgent.java
index e1382a317..4927a52b0 100644
--- a/core/src/main/java/com/google/adk/agents/ParallelAgent.java
+++ b/core/src/main/java/com/google/adk/agents/ParallelAgent.java
@@ -19,12 +19,14 @@
 
 import com.google.adk.agents.ConfigAgentUtils.ConfigurationException;
 import com.google.adk.events.Event;
+import com.google.common.collect.ImmutableMap;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Scheduler;
 import io.reactivex.rxjava3.schedulers.Schedulers;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -175,13 +177,78 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
       return Flowable.empty();
     }
 
-    var updatedInvocationContext = setBranchForCurrentAgent(this, invocationContext);
-    List<Flowable<Event>> agentFlowables = new ArrayList<>();
-    for (BaseAgent subAgent : currentSubAgents) {
-      agentFlowables.add(subAgent.runAsync(updatedInvocationContext).subscribeOn(scheduler));
+    if (!invocationContext.isResumable()) {
+      var updatedInvocationContext = setBranchForCurrentAgent(this, invocationContext);
+      List<Flowable<Event>> agentFlowables = new ArrayList<>();
+      for (BaseAgent subAgent : currentSubAgents) {
+        agentFlowables.add(subAgent.runAsync(updatedInvocationContext).subscribeOn(scheduler));
+      }
+      return Flowable.merge(agentFlowables)
+          .takeUntil((Event event) -> event.actions().escalate().orElse(false));
     }
-    return Flowable.merge(agentFlowables)
-        .takeUntil((Event event) -> event.actions().escalate().orElse(false));
+
+    // Resumable: skip completed branches, checkpoint that this agent started, pause (without
+    // ending) if any branch pauses, and end only once every active branch finished.
+    return Flowable.defer(
+        () -> {
+          List<BaseAgent> activeSubAgents = new ArrayList<>();
+          for (BaseAgent subAgent : currentSubAgents) {
+            if (!invocationContext.endOfAgents().getOrDefault(subAgent.name(), false)) {
+              activeSubAgents.add(subAgent);
+            }
+          }
+
+          Flowable<Event> initialCheckpoint = Flowable.empty();
+          if (!invocationContext.agentStates().containsKey(name())) {
+            ImmutableMap<String, Object> emptyState = ImmutableMap.of();
+            invocationContext.setAgentState(name(), emptyState, /* endOfAgent= */ false);
+            initialCheckpoint = Flowable.just(createStateEvent(invocationContext, emptyState));
+          }
+
+          var updatedInvocationContext = setBranchForCurrentAgent(this, invocationContext);
+          AtomicBoolean paused = new AtomicBoolean(false);
+          List<Flowable<Event>> agentFlowables = new ArrayList<>();
+          for (BaseAgent subAgent : activeSubAgents) {
+            agentFlowables.add(
+                subAgent
+                    .runAsync(updatedInvocationContext)
+                    .subscribeOn(scheduler)
+                    .doOnNext(
+                        event -> {
+                          if (invocationContext.shouldPauseInvocation(event)) {
+                            paused.set(true);
+                          }
+                        }));
+          }
+          Flowable<Event> merged =
+              Flowable.merge(agentFlowables)
+                  .takeUntil((Event event) -> event.actions().escalate().orElse(false));
+
+          return initialCheckpoint
+              .concatWith(merged)
+              .concatWith(
+                  Flowable.defer(
+                      () -> {
+                        if (paused.get()) {
+                          return Flowable.<Event>empty();
+                        }
+                        // Assumes active sub-agents record endOfAgent (a custom BaseAgent may not).
+                        boolean allEnded = true;
+                        for (BaseAgent subAgent : activeSubAgents) {
+                          if (!invocationContext
+                              .endOfAgents()
+                              .getOrDefault(subAgent.name(), false)) {
+                            allEnded = false;
+                            break;
+                          }
+                        }
+                        if (allEnded) {
+                          invocationContext.setAgentState(name(), null, /* endOfAgent= */ true);
+                          return Flowable.just(endOfAgentEvent(invocationContext));
+                        }
+                        return Flowable.empty();
+                      }));
+        });
   }
 
   /**
diff --git a/core/src/main/java/com/google/adk/agents/SequentialAgent.java b/core/src/main/java/com/google/adk/agents/SequentialAgent.java
index 963c3d109..2c68ce8c0 100644
--- a/core/src/main/java/com/google/adk/agents/SequentialAgent.java
+++ b/core/src/main/java/com/google/adk/agents/SequentialAgent.java
@@ -17,8 +17,11 @@
 
 import com.google.adk.agents.ConfigAgentUtils.ConfigurationException;
 import com.google.adk.events.Event;
+import com.google.common.collect.ImmutableMap;
 import io.reactivex.rxjava3.core.Flowable;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -108,22 +111,58 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
       return Flowable.fromIterable(subAgents)
           .concatMap(subAgent -> subAgent.runAsync(invocationContext));
     }
-    int startIndex =
-        WorkflowAgentResumption.resumeSubAgentIndex(invocationContext, subAgents).orElse(0);
+    // Resumable: checkpoint each sub-agent before it runs, fast-forward to the checkpoint on
+    // resume, and pause (without ending) on a long-running call.
+    Map<String, Object> state = invocationContext.agentStates().get(name());
+    String startSubAgentName =
+        state == null ? null : (String) state.get(WorkflowAgentStates.CURRENT_SUB_AGENT);
+    int startIndex;
+    boolean isResuming;
+    if (startSubAgentName != null) {
+      startIndex = WorkflowAgentStates.findIndexForResumption(subAgents, startSubAgentName, logger);
+      isResuming = true;
+    } else {
+      // Back-compat: a session paused before checkpoints existed has no agentState; reconstruct
+      // the resume point from history so it still fast-forwards past completed sub-agents.
+      Optional<Integer> reconstructed =
+          WorkflowAgentResumption.resumeSubAgentIndex(invocationContext, subAgents);
+      startIndex = reconstructed.orElse(0);
+      isResuming = reconstructed.isPresent();
+    }
     AtomicBoolean paused = new AtomicBoolean(false);
+    AtomicBoolean resuming = new AtomicBoolean(isResuming);
     return Flowable.fromIterable(subAgents.subList(startIndex, subAgents.size()))
         .concatMap(
             subAgent ->
-                paused.get()
-                    ? Flowable.<Event>empty()
-                    : subAgent
-                        .runAsync(invocationContext)
-                        .doOnNext(
-                            event -> {
-                              if (WorkflowAgentResumption.hasPendingLongRunningCall(event)) {
-                                paused.set(true);
-                              }
-                            }));
+                Flowable.defer(
+                    () -> {
+                      if (paused.get()) {
+                        return Flowable.<Event>empty();
+                      }
+                      Flowable<Event> checkpoint = Flowable.empty();
+                      if (!resuming.getAndSet(false)) {
+                        ImmutableMap<String, Object> subState =
+                            ImmutableMap.of(WorkflowAgentStates.CURRENT_SUB_AGENT, subAgent.name());
+                        invocationContext.setAgentState(name(), subState, /* endOfAgent= */ false);
+                        checkpoint = Flowable.just(createStateEvent(invocationContext, subState));
+                      }
+                      Flowable<Event> run =
+                          subAgent
+                              .runAsync(invocationContext)
+                              .doOnNext(
+                                  event -> {
+                                    if (invocationContext.shouldPauseInvocation(event)) {
+                                      paused.set(true);
+                                    }
+                                  });
+                      return checkpoint.concatWith(run);
+                    }))
+        .concatWith(
+            Flowable.defer(
+                () ->
+                    paused.get()
+                        ? Flowable.<Event>empty()
+                        : Flowable.just(endOfAgentEvent(invocationContext))));
   }
 
   /**
diff --git a/core/src/main/java/com/google/adk/agents/WorkflowAgentResumption.java b/core/src/main/java/com/google/adk/agents/WorkflowAgentResumption.java
index 2bff47803..66ff62a10 100644
--- a/core/src/main/java/com/google/adk/agents/WorkflowAgentResumption.java
+++ b/core/src/main/java/com/google/adk/agents/WorkflowAgentResumption.java
@@ -48,12 +48,5 @@ static Optional<Integer> resumeSubAgentIndex(
     return Optional.empty();
   }
 
-  /**
-   * Whether the event emits a long-running call still awaiting a response (e.g. a HITL request).
-   */
-  static boolean hasPendingLongRunningCall(Event event) {
-    return Functions.hasPendingLongRunningCall(event);
-  }
-
   private WorkflowAgentResumption() {}
 }
diff --git a/core/src/main/java/com/google/adk/agents/WorkflowAgentStates.java b/core/src/main/java/com/google/adk/agents/WorkflowAgentStates.java
new file mode 100644
index 000000000..5c0e8a720
--- /dev/null
+++ b/core/src/main/java/com/google/adk/agents/WorkflowAgentStates.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.agents;
+
+import java.util.List;
+import org.jspecify.annotations.Nullable;
+import org.slf4j.Logger;
+
+/**
+ * Wire-format keys and helpers for workflow-agent resumability checkpoints. The keys match Python
+ * and Kotlin ADK so persisted state is portable across languages.
+ */
+final class WorkflowAgentStates {
+
+  /** Key holding the name of the current/next sub-agent in a Sequential or Loop checkpoint. */
+  static final String CURRENT_SUB_AGENT = "current_sub_agent";
+
+  /** Key holding the completed-iteration count in a Loop checkpoint. */
+  static final String TIMES_LOOPED = "times_looped";
+
+  /**
+   * Returns the index of the sub-agent to resume from by name, or 0 (with a warning) when the name
+   * is null or no longer present. Mirrors the Kotlin {@code findIndexForResumption}.
+   */
+  static int findIndexForResumption(
+      List<? extends BaseAgent> subAgents, @Nullable String agentName, Logger logger) {
+    if (agentName == null) {
+      return 0;
+    }
+    for (int i = 0; i < subAgents.size(); i++) {
+      if (agentName.equals(subAgents.get(i).name())) {
+        return i;
+      }
+    }
+    // Do not log the agent name (treat names as potentially sensitive); log only the shape.
+    logger.warn("Restored sub-agent not found in current sub-agents list; resuming from index 0.");
+    return 0;
+  }
+
+  private WorkflowAgentStates() {}
+}
diff --git a/core/src/main/java/com/google/adk/events/EventActions.java b/core/src/main/java/com/google/adk/events/EventActions.java
index cde23c10e..14f1f7252 100644
--- a/core/src/main/java/com/google/adk/events/EventActions.java
+++ b/core/src/main/java/com/google/adk/events/EventActions.java
@@ -44,6 +44,7 @@ public class EventActions extends JsonBaseModel {
   private ConcurrentMap<String, Map<String, Object>> requestedAuthConfigs;
   private ConcurrentMap<String, ToolConfirmation> requestedToolConfirmations;
   private boolean endOfAgent;
+  private @Nullable Map<String, Object> agentState;
   private @Nullable EventCompaction compaction;
 
   /** Default constructor for Jackson. */
@@ -66,6 +67,7 @@ private EventActions(Builder builder) {
     this.requestedAuthConfigs = builder.requestedAuthConfigs;
     this.requestedToolConfirmations = builder.requestedToolConfirmations;
     this.endOfAgent = builder.endOfAgent;
+    this.agentState = builder.agentState;
     this.compaction = builder.compaction;
   }
 
@@ -192,6 +194,19 @@ public void setEndInvocation(boolean endInvocation) {
     this.endOfAgent = endInvocation;
   }
 
+  /**
+   * The checkpointed state of the authoring agent at this event, used for session resumability.
+   * Only set by ADK workflow/agent machinery on resumable invocations.
+   */
+  @JsonProperty("agentState")
+  public Optional<Map<String, Object>> agentState() {
+    return Optional.ofNullable(agentState);
+  }
+
+  public void setAgentState(@Nullable Map<String, Object> agentState) {
+    this.agentState = agentState;
+  }
+
   @JsonProperty("compaction")
   public Optional<EventCompaction> compaction() {
     return Optional.ofNullable(compaction);
@@ -226,6 +241,7 @@ public boolean equals(Object o) {
         && Objects.equals(requestedAuthConfigs, that.requestedAuthConfigs)
         && Objects.equals(requestedToolConfirmations, that.requestedToolConfirmations)
         && (endOfAgent == that.endOfAgent)
+        && Objects.equals(agentState, that.agentState)
         && Objects.equals(compaction, that.compaction);
   }
 
@@ -241,6 +257,7 @@ public int hashCode() {
         requestedAuthConfigs,
         requestedToolConfirmations,
         endOfAgent,
+        agentState,
         compaction);
   }
 
@@ -255,6 +272,7 @@ public static class Builder {
     private ConcurrentMap<String, Map<String, Object>> requestedAuthConfigs;
     private ConcurrentMap<String, ToolConfirmation> requestedToolConfirmations;
     private boolean endOfAgent = false;
+    private @Nullable Map<String, Object> agentState;
     private @Nullable EventCompaction compaction;
 
     public Builder() {
@@ -276,6 +294,7 @@ private Builder(EventActions eventActions) {
       this.requestedToolConfirmations =
           new ConcurrentHashMap<>(eventActions.requestedToolConfirmations());
       this.endOfAgent = eventActions.endOfAgent;
+      this.agentState = eventActions.agentState;
       this.compaction = eventActions.compaction;
     }
 
@@ -376,6 +395,13 @@ public Builder endInvocation(boolean endInvocation) {
       return this;
     }
 
+    @CanIgnoreReturnValue
+    @JsonProperty("agentState")
+    public Builder agentState(@Nullable Map<String, Object> agentState) {
+      this.agentState = agentState;
+      return this;
+    }
+
     @CanIgnoreReturnValue
     @JsonProperty("compaction")
     public Builder compaction(@Nullable EventCompaction value) {
@@ -394,6 +420,7 @@ public Builder merge(EventActions other) {
       this.requestedAuthConfigs.putAll(other.requestedAuthConfigs());
       this.requestedToolConfirmations.putAll(other.requestedToolConfirmations());
       this.endOfAgent = this.endOfAgent || other.endOfAgent();
+      other.agentState().ifPresent(this::agentState);
       other.compaction().ifPresent(this::compaction);
       return this;
     }
diff --git a/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java b/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java
index 91cc225f2..66e8354bf 100644
--- a/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java
+++ b/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java
@@ -536,9 +536,10 @@ private Flowable<Event> run(
                     return Flowable.empty();
                   } else if (invocationContext.isResumable()
                       && Functions.hasPendingLongRunningCall(eventList)) {
-                    // When resumable, a pending long-running call (e.g. HITL) pauses the flow
-                    // instead of calling the model again, matching Python ADK v1 and avoiding a
-                    // runaway re-issue loop. The disabled path is unchanged.
+                    // When resumable, an unanswered long-running call (e.g. HITL) pauses the flow
+                    // instead of calling the model again; a call answered by a function response
+                    // continues so the model summarizes it, matching Python ADK 2.x decide_resume.
+                    // The disabled path is unchanged.
                     logger.debug("Pausing flow execution on a pending long-running call.");
                     return Flowable.empty();
                   } else {
diff --git a/core/src/main/java/com/google/adk/flows/llmflows/Functions.java b/core/src/main/java/com/google/adk/flows/llmflows/Functions.java
index 3f3b8ef86..a8d2d0e55 100644
--- a/core/src/main/java/com/google/adk/flows/llmflows/Functions.java
+++ b/core/src/main/java/com/google/adk/flows/llmflows/Functions.java
@@ -471,11 +471,34 @@ public static boolean hasPendingLongRunningCall(Event event) {
   }
 
   /**
-   * Returns whether the last one or two events hold a pending long-running call, meaning a
-   * resumable flow should pause instead of calling the model again. Mirrors Python ADK v1's
-   * flow-level pause check on {@code events[-1]} and {@code events[-2]}.
+   * Returns whether the last one or two events hold a long-running call still awaiting a response,
+   * meaning a resumable flow should pause instead of calling the model again. A response that
+   * resolves the call -- the tool's own same-turn value, or a later user-injected resume -- lets
+   * the flow continue and the model summarize; only a no-response return (null or empty result,
+   * which emits no function response) pauses. Mirrors Python ADK 2.x {@code decide_resume}.
    */
   static boolean hasPendingLongRunningCall(List<Event> events) {
+    if (events.isEmpty()) {
+      return false;
+    }
+    Event last = events.get(events.size() - 1);
+    if (events.size() >= 2 && !last.functionResponses().isEmpty()) {
+      Event pending = events.get(events.size() - 2);
+      Set<String> longRunningIds = pending.longRunningToolIds().orElse(ImmutableSet.of());
+      Set<String> pausedIds = new HashSet<>();
+      for (FunctionCall call : pending.functionCalls()) {
+        if (call.id().isPresent() && longRunningIds.contains(call.id().get())) {
+          pausedIds.add(call.id().get());
+        }
+      }
+      Set<String> resolvedIds = new HashSet<>();
+      for (FunctionResponse response : last.functionResponses()) {
+        response.id().ifPresent(resolvedIds::add);
+      }
+      if (!pausedIds.isEmpty() && resolvedIds.containsAll(pausedIds)) {
+        return false;
+      }
+    }
     int from = Math.max(0, events.size() - 2);
     for (int i = events.size() - 1; i >= from; i--) {
       if (hasPendingLongRunningCall(events.get(i))) {
diff --git a/core/src/main/java/com/google/adk/runner/Runner.java b/core/src/main/java/com/google/adk/runner/Runner.java
index 48eb9fad6..05ced235d 100644
--- a/core/src/main/java/com/google/adk/runner/Runner.java
+++ b/core/src/main/java/com/google/adk/runner/Runner.java
@@ -55,6 +55,8 @@
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.genai.types.AudioTranscriptionConfig;
 import com.google.genai.types.Content;
+import com.google.genai.types.FunctionCall;
+import com.google.genai.types.FunctionResponse;
 import com.google.genai.types.Modality;
 import com.google.genai.types.Part;
 import io.opentelemetry.api.trace.Span;
@@ -68,9 +70,11 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import org.jspecify.annotations.Nullable;
@@ -358,6 +362,22 @@ private Single<Event> appendNewMessageToSession(
       InvocationContext invocationContext,
       boolean saveInputBlobsAsArtifacts,
       @Nullable Map<String, Object> stateDelta) {
+    return appendNewMessageToSession(
+        session,
+        newMessage,
+        invocationContext,
+        saveInputBlobsAsArtifacts,
+        stateDelta,
+        /* branch= */ null);
+  }
+
+  private Single<Event> appendNewMessageToSession(
+      Session session,
+      Content newMessage,
+      InvocationContext invocationContext,
+      boolean saveInputBlobsAsArtifacts,
+      @Nullable Map<String, Object> stateDelta,
+      @Nullable String branch) {
     checkArgument(newMessage.parts().isPresent(), "No parts in the new_message.");
 
     Content messageToAppend = newMessage;
@@ -392,6 +412,7 @@ private Single<Event> appendNewMessageToSession(
             .id(Event.generateEventId())
             .invocationId(invocationContext.invocationId())
             .author("user")
+            .branch(branch)
             .content(messageToAppend);
 
     // Add state delta if provided
@@ -505,6 +526,62 @@ public Flowable<Event> runAsync(String userId, String sessionId, Content newMess
     return runAsync(userId, sessionId, newMessage, RunConfig.builder().build());
   }
 
+  /** See {@link #resumeAsync(String, String, String, Content, RunConfig)}. */
+  public Flowable<Event> resumeAsync(
+      String userId, String sessionId, @Nullable String invocationId) {
+    return resumeAsync(
+        userId, sessionId, invocationId, /* newMessage= */ null, RunConfig.builder().build());
+  }
+
+  /**
+   * Resumes a paused, resumable invocation instead of starting a new one: the message is optional
+   * and the run continues an existing invocation rather than minting a new id.
+   *
+   * <p>The invocation to resume is resolved from {@code newMessage} when it carries a function
+   * response, else from {@code invocationId}, else from the last event in the session. Agent
+   * checkpoints are rehydrated from history, and an invocation whose active agent already finished
+   * resolves to a no-op.
+   *
+   * @param userId the user id of the session.
+   * @param sessionId the session id.
+   * @param invocationId the invocation to resume; may be {@code null} when it can be inferred.
+   * @param newMessage an optional message (typically a function response) to append before running.
+   * @param runConfig the run configuration.
+   * @return the events generated while resuming, or an empty stream when there is nothing to
+   *     resume.
+   * @throws IllegalArgumentException if the app is not resumable or the invocation cannot be
+   *     resolved.
+   */
+  public Flowable<Event> resumeAsync(
+      String userId,
+      String sessionId,
+      @Nullable String invocationId,
+      @Nullable Content newMessage,
+      RunConfig runConfig) {
+    checkArgument(
+        isResumable(),
+        "resumeAsync requires an App configured with a resumable ResumabilityConfig.");
+    return Flowable.defer(
+            () ->
+                this.sessionService
+                    .getSession(appName, userId, sessionId, Optional.empty())
+                    .switchIfEmpty(
+                        Single.error(
+                            () ->
+                                new IllegalArgumentException(
+                                    String.format(
+                                        "Session not found: %s for user %s", sessionId, userId))))
+                    .flatMapPublisher(
+                        session ->
+                            runResumableFromSession(
+                                session,
+                                invocationId,
+                                newMessage,
+                                runConfig,
+                                /* stateDelta= */ null)))
+        .compose(Tracing.trace("invocation"));
+  }
+
   /**
    * Runs the agent asynchronously using a provided Session object.
    *
@@ -522,6 +599,21 @@ protected Flowable<Event> runAsyncImpl(
     Preconditions.checkNotNull(session, "session cannot be null");
     Preconditions.checkNotNull(newMessage, "newMessage cannot be null");
     Preconditions.checkNotNull(runConfig, "runConfig cannot be null");
+    // When resumable, a message that resolves to an existing invocation (e.g. a function response
+    // to a paused call) resumes it; any other message starts a new invocation. Disabled: unchanged.
+    if (isResumable()) {
+      return runResumableFromSession(
+          session, /* providedInvocationId= */ null, newMessage, runConfig, stateDelta);
+    }
+    return runNewInvocation(session, newMessage, runConfig, stateDelta);
+  }
+
+  /** Starts a brand-new invocation for {@code newMessage} (the default, non-resume flow). */
+  private Flowable<Event> runNewInvocation(
+      Session session,
+      Content newMessage,
+      RunConfig runConfig,
+      @Nullable Map<String, Object> stateDelta) {
     return Flowable.defer(
             () -> {
               Context capturedContext = Context.current();
@@ -670,6 +762,176 @@ private Completable compactEvents(Session session) {
         .orElseGet(Completable::complete);
   }
 
+  /**
+   * Resumes an existing invocation when one resolves from {@code providedInvocationId} or a
+   * function response in {@code newMessage}; otherwise starts a new invocation. Requires
+   * resumability.
+   */
+  private Flowable<Event> runResumableFromSession(
+      Session session,
+      @Nullable String providedInvocationId,
+      @Nullable Content newMessage,
+      RunConfig runConfig,
+      @Nullable Map<String, Object> stateDelta) {
+    return Flowable.defer(
+        () -> {
+          String resolvedInvocationId =
+              resolveInvocationId(session, newMessage, providedInvocationId);
+          if (resolvedInvocationId == null) {
+            if (newMessage == null) {
+              return Flowable.<Event>error(
+                  new IllegalArgumentException(
+                      "No new message provided and no resumable invocation to resume."));
+            }
+            return runNewInvocation(session, newMessage, runConfig, stateDelta);
+          }
+          return resumeCore(session, resolvedInvocationId, newMessage, runConfig, stateDelta);
+        });
+  }
+
+  /**
+   * Runs an existing invocation on the given session: optionally appends {@code newMessage},
+   * rehydrates agent checkpoints, skips a completed invocation, and runs the resolved agent under
+   * the resumed invocation id.
+   */
+  private Flowable<Event> resumeCore(
+      Session session,
+      String resolvedInvocationId,
+      @Nullable Content newMessage,
+      RunConfig runConfig,
+      @Nullable Map<String, Object> stateDelta) {
+    return Flowable.defer(
+        () -> {
+          Context capturedContext = Context.current();
+          if (stateDelta != null && !stateDelta.isEmpty()) {
+            stateDelta.forEach((key, value) -> session.state().put(key, value));
+          }
+
+          // Append the function-response message under the resumed invocation first, inheriting the
+          // branch of the call it answers, so routing and rehydration see it.
+          Completable appendMessage = Completable.complete();
+          if (newMessage != null) {
+            InvocationContext appendContext =
+                newInvocationContextBuilder(session)
+                    .invocationId(resolvedInvocationId)
+                    .runConfig(runConfig)
+                    .userContent(newMessage)
+                    .build();
+            String branch =
+                matchingFunctionCallEvent(session, newMessage).flatMap(Event::branch).orElse(null);
+            appendMessage =
+                appendNewMessageToSession(
+                        session,
+                        newMessage,
+                        appendContext,
+                        runConfig.saveInputBlobsAsArtifacts(),
+                        stateDelta,
+                        branch)
+                    .ignoreElement();
+          }
+
+          return appendMessage
+              .andThen(
+                  Flowable.defer(
+                      () -> {
+                        // Build the resumed context after any append so routing and rehydration see
+                        // the latest history.
+                        InvocationContext context =
+                            newInvocationContextBuilder(session)
+                                .invocationId(resolvedInvocationId)
+                                .runConfig(runConfig)
+                                .userContent(newMessage == null ? Content.fromParts() : newMessage)
+                                .build();
+                        context.populateInvocationAgentStates();
+
+                        // No-op guard: a completed invocation (its active agent already finished)
+                        // is not re-run.
+                        if (context.endOfAgents().getOrDefault(context.agent().name(), false)) {
+                          return Flowable.<Event>empty();
+                        }
+
+                        PersistBarrier.enable(context);
+
+                        return context
+                            .agent()
+                            .runAsync(context)
+                            .concatMap(
+                                agentEvent -> {
+                                  Single<Event> persistStep =
+                                      agentEvent.partial().orElse(false)
+                                          ? Single.just(agentEvent)
+                                          : this.sessionService.appendEvent(session, agentEvent);
+                                  return persistStep
+                                      .doOnSuccess(
+                                          unusedEvent ->
+                                              PersistBarrier.markPersisted(
+                                                  context, agentEvent.id()))
+                                      .doOnError(
+                                          error ->
+                                              PersistBarrier.markFailed(
+                                                  context, agentEvent.id(), error))
+                                      .flatMap(
+                                          registeredEvent ->
+                                              context
+                                                  .pluginManager()
+                                                  .onEventCallback(context, registeredEvent)
+                                                  .defaultIfEmpty(registeredEvent))
+                                      .toFlowable();
+                                })
+                            .concatWith(Completable.defer(() -> compactEvents(session)));
+                      }))
+              .compose(Tracing.<Event>withContext(capturedContext));
+        });
+  }
+
+  /**
+   * Resolves which invocation a request targets: the invocation that issued the function call
+   * matching {@code newMessage}'s function response, else the caller-supplied {@code invocationId}.
+   * Returns {@code null} when neither applies (a fresh message starts a new invocation).
+   */
+  private static @Nullable String resolveInvocationId(
+      Session session, @Nullable Content newMessage, @Nullable String invocationId) {
+    if (newMessage != null) {
+      Optional<String> fromResponse =
+          matchingFunctionCallEvent(session, newMessage).map(Event::invocationId);
+      if (fromResponse.isPresent()) {
+        return fromResponse.get();
+      }
+    }
+    return invocationId;
+  }
+
+  /**
+   * Returns the session event whose function call matches a function response id carried by {@code
+   * newMessage}, searching newest-first. Both the resumed invocation id and the branch of the
+   * appended function-response event are derived from it.
+   */
+  private static Optional<Event> matchingFunctionCallEvent(Session session, Content newMessage) {
+    Set<String> responseIds = new HashSet<>();
+    newMessage
+        .parts()
+        .ifPresent(
+            parts ->
+                parts.forEach(
+                    part ->
+                        part.functionResponse()
+                            .flatMap(FunctionResponse::id)
+                            .ifPresent(responseIds::add)));
+    if (responseIds.isEmpty()) {
+      return Optional.empty();
+    }
+    List<Event> events = session.events();
+    for (int i = events.size() - 1; i >= 0; i--) {
+      Event event = events.get(i);
+      for (FunctionCall call : event.functionCalls()) {
+        if (call.id().isPresent() && responseIds.contains(call.id().get())) {
+          return Optional.of(event);
+        }
+      }
+    }
+    return Optional.empty();
+  }
+
   private void copySessionStates(Session source, Session target) {
     // TODO: remove this hack when deprecating all runAsync with Session.
     target.state().putAll(source.state());
diff --git a/core/src/main/java/com/google/adk/sessions/SessionJsonConverter.java b/core/src/main/java/com/google/adk/sessions/SessionJsonConverter.java
index adc84fbb6..8365117ef 100644
--- a/core/src/main/java/com/google/adk/sessions/SessionJsonConverter.java
+++ b/core/src/main/java/com/google/adk/sessions/SessionJsonConverter.java
@@ -114,6 +114,7 @@ static String convertEventToJson(Event event, boolean useIsoString) {
       putIfNotEmpty(actionsJson, "requestedAuthConfigs", actions.requestedAuthConfigs());
       putIfNotEmpty(
           actionsJson, "requestedToolConfirmations", actions.requestedToolConfirmations());
+      actions.agentState().ifPresent(v -> actionsJson.put("agentState", v));
       eventJson.put("actions", actionsJson);
     }
     event.content().ifPresent(c -> eventJson.put("content", SessionUtils.encodeContent(c)));
@@ -192,6 +193,10 @@ static Event fromApiEvent(Map<String, Object> apiEvent) {
           Optional.ofNullable(actionsMap.get("requestedToolConfirmations"))
               .map(SessionJsonConverter::asConcurrentMapOfToolConfirmations)
               .orElse(new ConcurrentHashMap<>()));
+      Object agentState = actionsMap.get("agentState");
+      if (agentState != null) {
+        eventActionsBuilder.agentState((Map<String, Object>) agentState);
+      }
     }
 
     Event event =
diff --git a/core/src/test/java/com/google/adk/agents/InvocationContextTest.java b/core/src/test/java/com/google/adk/agents/InvocationContextTest.java
index e588a38ca..6c67f93fa 100644
--- a/core/src/test/java/com/google/adk/agents/InvocationContextTest.java
+++ b/core/src/test/java/com/google/adk/agents/InvocationContextTest.java
@@ -20,7 +20,10 @@
 import static org.junit.Assert.assertThrows;
 import static org.mockito.Mockito.mock;
 
+import com.google.adk.apps.ResumabilityConfig;
 import com.google.adk.artifacts.BaseArtifactService;
+import com.google.adk.events.Event;
+import com.google.adk.events.EventActions;
 import com.google.adk.memory.BaseMemoryService;
 import com.google.adk.models.LlmCallsLimitExceededException;
 import com.google.adk.plugins.PluginManager;
@@ -28,7 +31,10 @@
 import com.google.adk.sessions.Session;
 import com.google.adk.summarizer.EventsCompactionConfig;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.genai.types.Content;
+import com.google.genai.types.FunctionCall;
+import com.google.genai.types.Part;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -724,4 +730,332 @@ public void build_missingSessionService_throwsException() {
     IllegalStateException exception = assertThrows(IllegalStateException.class, builder::build);
     assertThat(exception).hasMessageThat().isEqualTo("Session service must be set.");
   }
+
+  // ---- Resumability: runtime checkpoint state (parity with Kotlin InvocationContextTest). ----
+
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  private InvocationContext resumableContext(Session eventSession, String invocationId) {
+    return InvocationContext.builder()
+        .sessionService(mockSessionService)
+        .artifactService(mockArtifactService)
+        .agent(mockAgent)
+        .session(eventSession)
+        .invocationId(invocationId)
+        .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+        .build();
+  }
+
+  private static Event agentEvent(
+      String invocationId, String author, EventActions actions, Content content) {
+    Event.Builder builder =
+        Event.builder().id(Event.generateEventId()).invocationId(invocationId).author(author);
+    if (actions != null) {
+      builder.actions(actions);
+    }
+    if (content != null) {
+      builder.content(content);
+    }
+    return builder.build();
+  }
+
+  @Test
+  public void isResumable_configTrue_returnsTrue() {
+    InvocationContext context = resumableContext(session, "inv");
+    assertThat(context.isResumable()).isTrue();
+  }
+
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void isResumable_configFalse_returnsFalse() {
+    InvocationContext context =
+        InvocationContext.builder()
+            .sessionService(mockSessionService)
+            .artifactService(mockArtifactService)
+            .agent(mockAgent)
+            .session(session)
+            .resumabilityConfig(ResumabilityConfig.builder().resumable(false).build())
+            .build();
+    assertThat(context.isResumable()).isFalse();
+  }
+
+  @Test
+  public void isResumable_nullConfig_returnsFalse() {
+    InvocationContext context =
+        InvocationContext.builder()
+            .sessionService(mockSessionService)
+            .artifactService(mockArtifactService)
+            .agent(mockAgent)
+            .session(session)
+            .build();
+    assertThat(context.isResumable()).isFalse();
+  }
+
+  @Test
+  public void setAgentState_storesStateAndClearsEnd() {
+    InvocationContext context = resumableContext(session, "inv");
+
+    context.setAgentState("a", ImmutableMap.of("k", "v"), /* endOfAgent= */ false);
+
+    assertThat(context.agentStates()).containsEntry("a", ImmutableMap.of("k", "v"));
+    assertThat(context.endOfAgents()).containsEntry("a", false);
+  }
+
+  @Test
+  public void setAgentState_endOfAgent_marksEndedAndDropsState() {
+    InvocationContext context = resumableContext(session, "inv");
+    context.setAgentState("a", ImmutableMap.of("k", "v"), /* endOfAgent= */ false);
+
+    context.setAgentState("a", /* agentState= */ null, /* endOfAgent= */ true);
+
+    assertThat(context.endOfAgents()).containsEntry("a", true);
+    assertThat(context.agentStates()).doesNotContainKey("a");
+  }
+
+  @Test
+  public void setAgentState_nullStateNotEnded_clearsBoth() {
+    InvocationContext context = resumableContext(session, "inv");
+    context.setAgentState("a", ImmutableMap.of("k", "v"), /* endOfAgent= */ false);
+
+    context.setAgentState("a", /* agentState= */ null, /* endOfAgent= */ false);
+
+    assertThat(context.agentStates()).doesNotContainKey("a");
+    assertThat(context.endOfAgents()).doesNotContainKey("a");
+  }
+
+  @Test
+  public void shouldPauseInvocation_resumableWithLongRunningCall_returnsTrue() {
+    InvocationContext context = resumableContext(session, "inv");
+    Event event =
+        Event.builder()
+            .id("e1")
+            .invocationId("inv")
+            .author("a")
+            .content(
+                Content.fromParts(
+                    Part.builder()
+                        .functionCall(FunctionCall.builder().id("c1").name("tool").build())
+                        .build()))
+            .longRunningToolIds(ImmutableSet.of("c1"))
+            .build();
+
+    assertThat(context.shouldPauseInvocation(event)).isTrue();
+  }
+
+  @Test
+  public void shouldPauseInvocation_notResumable_returnsFalse() {
+    InvocationContext context =
+        InvocationContext.builder()
+            .sessionService(mockSessionService)
+            .artifactService(mockArtifactService)
+            .agent(mockAgent)
+            .session(session)
+            .invocationId("inv")
+            .build();
+    Event event =
+        Event.builder()
+            .id("e1")
+            .invocationId("inv")
+            .author("a")
+            .content(
+                Content.fromParts(
+                    Part.builder()
+                        .functionCall(FunctionCall.builder().id("c1").name("tool").build())
+                        .build()))
+            .longRunningToolIds(ImmutableSet.of("c1"))
+            .build();
+
+    assertThat(context.shouldPauseInvocation(event)).isFalse();
+  }
+
+  @Test
+  public void shouldPauseInvocation_noLongRunningIds_returnsFalse() {
+    InvocationContext context = resumableContext(session, "inv");
+    Event event =
+        Event.builder()
+            .id("e1")
+            .invocationId("inv")
+            .author("a")
+            .content(
+                Content.fromParts(
+                    Part.builder()
+                        .functionCall(FunctionCall.builder().id("c1").name("tool").build())
+                        .build()))
+            .build();
+
+    assertThat(context.shouldPauseInvocation(event)).isFalse();
+  }
+
+  @Test
+  public void shouldPauseInvocation_callIdNotInLongRunningSet_returnsFalse() {
+    InvocationContext context = resumableContext(session, "inv");
+    Event event =
+        Event.builder()
+            .id("e1")
+            .invocationId("inv")
+            .author("a")
+            .content(
+                Content.fromParts(
+                    Part.builder()
+                        .functionCall(FunctionCall.builder().id("c1").name("tool").build())
+                        .build()))
+            .longRunningToolIds(ImmutableSet.of("other"))
+            .build();
+
+    assertThat(context.shouldPauseInvocation(event)).isFalse();
+  }
+
+  @Test
+  public void getEvents_filtersByInvocationAndBranch() {
+    Session eventSession = Session.builder("s").build();
+    Event thisInv = agentEvent("inv", "a", null, Content.fromParts(Part.fromText("x")));
+    Event otherInv = agentEvent("other", "a", null, Content.fromParts(Part.fromText("y")));
+    Event branchB =
+        Event.builder()
+            .id("e3")
+            .invocationId("inv")
+            .author("a")
+            .branch("branchB")
+            .content(Content.fromParts(Part.fromText("z")))
+            .build();
+    eventSession.events().add(thisInv);
+    eventSession.events().add(otherInv);
+    eventSession.events().add(branchB);
+    InvocationContext context = resumableContext(eventSession, "inv");
+
+    assertThat(context.getEvents(/* currentInvocation= */ true, /* currentBranch= */ false))
+        .containsExactly(thisInv, branchB)
+        .inOrder();
+    // A null-branch event is visible on any branch; the "branchB" event is filtered out.
+    assertThat(context.getEvents(/* currentInvocation= */ true, /* currentBranch= */ true))
+        .containsExactly(thisInv);
+  }
+
+  @Test
+  public void populateInvocationAgentStates_notResumable_doesNothing() {
+    Session eventSession = Session.builder("s").build();
+    eventSession
+        .events()
+        .add(
+            agentEvent(
+                "inv",
+                "a",
+                EventActions.builder().agentState(ImmutableMap.of("k", "v")).build(),
+                Content.fromParts(Part.fromText("x"))));
+    InvocationContext context =
+        InvocationContext.builder()
+            .sessionService(mockSessionService)
+            .artifactService(mockArtifactService)
+            .agent(mockAgent)
+            .session(eventSession)
+            .invocationId("inv")
+            .build();
+
+    context.populateInvocationAgentStates();
+
+    assertThat(context.agentStates()).isEmpty();
+    assertThat(context.endOfAgents()).isEmpty();
+  }
+
+  @Test
+  public void populateInvocationAgentStates_endOfAgentEvent_marksEndedAndRemovesState() {
+    Session eventSession = Session.builder("s").build();
+    eventSession
+        .events()
+        .add(
+            agentEvent(
+                "inv",
+                "a",
+                EventActions.builder().agentState(ImmutableMap.of("k", "v")).build(),
+                Content.fromParts(Part.fromText("x"))));
+    eventSession
+        .events()
+        .add(agentEvent("inv", "a", EventActions.builder().endOfAgent(true).build(), null));
+    InvocationContext context = resumableContext(eventSession, "inv");
+
+    context.populateInvocationAgentStates();
+
+    assertThat(context.endOfAgents()).containsEntry("a", true);
+    assertThat(context.agentStates()).doesNotContainKey("a");
+  }
+
+  @Test
+  public void populateInvocationAgentStates_agentStateEvent_setsStateAndClearsEnd() {
+    Session eventSession = Session.builder("s").build();
+    eventSession
+        .events()
+        .add(
+            agentEvent(
+                "inv",
+                "a",
+                EventActions.builder().agentState(ImmutableMap.of("k", "v")).build(),
+                Content.fromParts(Part.fromText("x"))));
+    InvocationContext context = resumableContext(eventSession, "inv");
+
+    context.populateInvocationAgentStates();
+
+    assertThat(context.agentStates()).containsEntry("a", ImmutableMap.of("k", "v"));
+    assertThat(context.endOfAgents()).containsEntry("a", false);
+  }
+
+  @Test
+  public void populateInvocationAgentStates_agentStateAndEndOfAgent_endOfAgentWins() {
+    Session eventSession = Session.builder("s").build();
+    eventSession
+        .events()
+        .add(
+            agentEvent(
+                "inv",
+                "a",
+                EventActions.builder()
+                    .endOfAgent(true)
+                    .agentState(ImmutableMap.of("k", "v"))
+                    .build(),
+                Content.fromParts(Part.fromText("x"))));
+    InvocationContext context = resumableContext(eventSession, "inv");
+
+    context.populateInvocationAgentStates();
+
+    assertThat(context.endOfAgents()).containsEntry("a", true);
+    assertThat(context.agentStates()).doesNotContainKey("a");
+  }
+
+  @Test
+  public void populateInvocationAgentStates_newContentFromNonUserAuthor_initializesEmptyState() {
+    Session eventSession = Session.builder("s").build();
+    eventSession
+        .events()
+        .add(agentEvent("inv", "a", null, Content.fromParts(Part.fromText("hello"))));
+    InvocationContext context = resumableContext(eventSession, "inv");
+
+    context.populateInvocationAgentStates();
+
+    assertThat(context.agentStates()).containsKey("a");
+    assertThat(context.agentStates().get("a")).isEmpty();
+    assertThat(context.endOfAgents()).containsEntry("a", false);
+  }
+
+  @Test
+  public void populateInvocationAgentStates_userMessage_ignoredForDefaultState() {
+    Session eventSession = Session.builder("s").build();
+    eventSession
+        .events()
+        .add(agentEvent("inv", "user", null, Content.fromParts(Part.fromText("hi"))));
+    InvocationContext context = resumableContext(eventSession, "inv");
+
+    context.populateInvocationAgentStates();
+
+    assertThat(context.agentStates()).isEmpty();
+  }
+
+  @Test
+  public void populateInvocationAgentStates_noContentNoState_ignored() {
+    Session eventSession = Session.builder("s").build();
+    eventSession.events().add(agentEvent("inv", "a", null, null));
+    InvocationContext context = resumableContext(eventSession, "inv");
+
+    context.populateInvocationAgentStates();
+
+    assertThat(context.agentStates()).isEmpty();
+    assertThat(context.endOfAgents()).isEmpty();
+  }
 }
diff --git a/core/src/test/java/com/google/adk/agents/LlmAgentTest.java b/core/src/test/java/com/google/adk/agents/LlmAgentTest.java
index 26843bb56..5777ef804 100644
--- a/core/src/test/java/com/google/adk/agents/LlmAgentTest.java
+++ b/core/src/test/java/com/google/adk/agents/LlmAgentTest.java
@@ -23,6 +23,7 @@
 import static com.google.adk.testing.TestUtils.createTestAgentBuilder;
 import static com.google.adk.testing.TestUtils.createTestLlm;
 import static com.google.adk.testing.TestUtils.createTextLlmResponse;
+import static com.google.adk.testing.TestUtils.simplifyEvents;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -33,7 +34,10 @@
 import com.google.adk.agents.Callbacks.BeforeToolCallback;
 import com.google.adk.agents.Callbacks.OnModelErrorCallback;
 import com.google.adk.agents.Callbacks.OnToolErrorCallback;
+import com.google.adk.apps.ResumabilityConfig;
+import com.google.adk.artifacts.InMemoryArtifactService;
 import com.google.adk.events.Event;
+import com.google.adk.events.EventActions;
 import com.google.adk.examples.Example;
 import com.google.adk.models.LlmRegistry;
 import com.google.adk.models.LlmRequest;
@@ -632,4 +636,110 @@ public void run_withExampleTool_doesNotAddFunctionDeclarations() {
     var config = request.config().get();
     assertThat(config.tools().isPresent()).isFalse();
   }
+
+  // ---- Resumability: resume into a transferred sub-agent (parity with ResumableLlmAgentTest).
+  // ----
+
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  private static InvocationContext resumableContextWithSeededEvent(
+      LlmAgent rootAgent, Event seededEvent) {
+    InMemorySessionService sessionService = new InMemorySessionService();
+    Session session = sessionService.createSession("app", "user").blockingGet();
+    var unused = sessionService.appendEvent(session, seededEvent).blockingGet();
+    return InvocationContext.builder()
+        .sessionService(sessionService)
+        .artifactService(new InMemoryArtifactService())
+        .invocationId("inv")
+        .agent(rootAgent)
+        .session(session)
+        .userContent(Content.fromParts(Part.fromText("hi")))
+        .runConfig(RunConfig.builder().build())
+        .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+        .build();
+  }
+
+  @Test
+  public void runAsync_resumeFromTransferCall_runsTransferredSubAgent() {
+    LlmAgent sub =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("sub response")))
+            .name("sub")
+            .build();
+    TestLlm rootLlm = createTestLlm(createTextLlmResponse("root should not run"));
+    LlmAgent root = createTestAgentBuilder(rootLlm).name("root").subAgents(sub).build();
+    Event transferEvent =
+        Event.builder()
+            .id("t1")
+            .invocationId("inv")
+            .author("root")
+            .actions(EventActions.builder().transferToAgent("sub").build())
+            .content(Content.fromParts(Part.fromText("transferring")))
+            .build();
+    InvocationContext context = resumableContextWithSeededEvent(root, transferEvent);
+    context.setAgentState("root", ImmutableMap.of(), /* endOfAgent= */ false);
+
+    List<Event> events = root.runAsync(context).toList().blockingGet();
+
+    // The transferred sub-agent runs; the root model is not re-invoked; root marks end-of-agent.
+    assertThat(simplifyEvents(events)).contains("sub: sub response");
+    assertThat(rootLlm.getRequests()).isEmpty();
+    assertThat(
+            events.stream()
+                .anyMatch(event -> event.author().equals("root") && event.actions().endOfAgent()))
+        .isTrue();
+  }
+
+  @Test
+  public void runAsync_resumeNoTransfer_continuesRootAgent() {
+    TestLlm rootLlm = createTestLlm(createTextLlmResponse("root continues"));
+    LlmAgent root = createTestAgentBuilder(rootLlm).name("root").build();
+    Event priorModelResponse =
+        Event.builder()
+            .id("m1")
+            .invocationId("inv")
+            .author("root")
+            .content(Content.fromParts(Part.fromText("earlier response")))
+            .build();
+    InvocationContext context = resumableContextWithSeededEvent(root, priorModelResponse);
+    context.setAgentState("root", ImmutableMap.of(), /* endOfAgent= */ false);
+
+    List<Event> events = root.runAsync(context).toList().blockingGet();
+
+    // No transfer recorded: the root agent continues by invoking its model.
+    assertThat(simplifyEvents(events)).contains("root: root continues");
+    assertThat(rootLlm.getRequests()).hasSize(1);
+  }
+
+  @Test
+  public void runAsync_resumeFromTransferToPeer_runsTransferredPeerAgent() {
+    LlmAgent root =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("root")))
+            .name("root")
+            .subAgents(
+                createTestAgentBuilder(
+                        createTestLlm(createTextLlmResponse("agent A should not re-run")))
+                    .name("agent_a")
+                    .build(),
+                createTestAgentBuilder(createTestLlm(createTextLlmResponse("agent B response")))
+                    .name("agent_b")
+                    .build())
+            .build();
+    // agent_a transferred to its peer agent_b (not a descendant), then the invocation paused.
+    LlmAgent agentA = (LlmAgent) root.findAgent("agent_a").get();
+    Event transferEvent =
+        Event.builder()
+            .id("t1")
+            .invocationId("inv")
+            .author("agent_a")
+            .actions(EventActions.builder().transferToAgent("agent_b").build())
+            .content(Content.fromParts(Part.fromText("transferring to peer")))
+            .build();
+    InvocationContext context = resumableContextWithSeededEvent(agentA, transferEvent);
+    context.setAgentState("agent_a", ImmutableMap.of(), /* endOfAgent= */ false);
+
+    List<Event> events = agentA.runAsync(context).toList().blockingGet();
+
+    // The transferred peer runs; agent_a does not re-run itself.
+    assertThat(simplifyEvents(events)).contains("agent_b: agent B response");
+    assertThat(simplifyEvents(events)).doesNotContain("agent_a: agent A should not re-run");
+  }
 }
diff --git a/core/src/test/java/com/google/adk/agents/LoopAgentTest.java b/core/src/test/java/com/google/adk/agents/LoopAgentTest.java
index b2d0778c6..b2454e367 100644
--- a/core/src/test/java/com/google/adk/agents/LoopAgentTest.java
+++ b/core/src/test/java/com/google/adk/agents/LoopAgentTest.java
@@ -19,6 +19,7 @@
 import static com.google.adk.testing.TestUtils.createEscalateEvent;
 import static com.google.adk.testing.TestUtils.createEvent;
 import static com.google.adk.testing.TestUtils.createInvocationContext;
+import static com.google.adk.testing.TestUtils.createResumableInvocationContext;
 import static com.google.adk.testing.TestUtils.createSubAgent;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.truth.Truth.assertThat;
@@ -27,8 +28,11 @@
 import com.google.adk.events.Event;
 import com.google.adk.testing.TestBaseAgent;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.genai.types.Content;
+import com.google.genai.types.FunctionCall;
 import com.google.genai.types.Part;
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Maybe;
@@ -240,4 +244,96 @@ public void runAsync_withEndInvocationInSubAgentCallback_stopsSubAgentButLoopCon
     assertThat(normalAgentRunCount.get()).isEqualTo(3);
     assertThat(subAgent2RunCount.get()).isEqualTo(1);
   }
+
+  // ---- Resumability: durable checkpoint resume (parity with Kotlin LoopAgentTest). ----
+
+  @Test
+  public void runAsync_resumable_emitsEndOfAgent() {
+    TestBaseAgent subAgent =
+        createSubAgent("sub", createEvent("e").toBuilder().author("sub").build());
+    LoopAgent loopAgent =
+        LoopAgent.builder().name("loop").subAgents(subAgent).maxIterations(1).build();
+    InvocationContext context = createResumableInvocationContext(loopAgent);
+
+    List<Event> events = loopAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(
+            events.stream()
+                .anyMatch(event -> event.author().equals("loop") && event.actions().endOfAgent()))
+        .isTrue();
+  }
+
+  @Test
+  public void runAsync_notResumable_doesNotEmitEndOfAgent() {
+    TestBaseAgent subAgent =
+        createSubAgent("sub", createEvent("e").toBuilder().author("sub").build());
+    LoopAgent loopAgent =
+        LoopAgent.builder().name("loop").subAgents(subAgent).maxIterations(1).build();
+    InvocationContext context = createInvocationContext(loopAgent);
+
+    List<Event> events = loopAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(events.stream().anyMatch(event -> event.actions().endOfAgent())).isFalse();
+  }
+
+  @Test
+  public void runAsync_resumingFromMiddle_restoresIterationAndAgent() {
+    TestBaseAgent agent1 =
+        createSubAgent(
+            "agent1", () -> Flowable.just(createEvent("a1").toBuilder().author("agent1").build()));
+    TestBaseAgent agent2 =
+        createSubAgent(
+            "agent2", () -> Flowable.just(createEvent("a2").toBuilder().author("agent2").build()));
+    LoopAgent loopAgent =
+        LoopAgent.builder().name("loop").subAgents(agent1, agent2).maxIterations(3).build();
+    InvocationContext context = createResumableInvocationContext(loopAgent);
+    // Resume mid-loop: iteration 1, at agent2.
+    context.setAgentState(
+        "loop",
+        ImmutableMap.of("current_sub_agent", "agent2", "times_looped", 1),
+        /* endOfAgent= */ false);
+
+    List<Event> events = loopAgent.runAsync(context).toList().blockingGet();
+
+    // The first sub-agent to run is agent2 (agent1 is skipped in the resumed iteration).
+    String firstSubAgentAuthor =
+        events.stream()
+            .filter(event -> event.content().isPresent())
+            .map(Event::author)
+            .filter(author -> author.equals("agent1") || author.equals("agent2"))
+            .findFirst()
+            .orElse(null);
+    assertThat(firstSubAgentAuthor).isEqualTo("agent2");
+    // The loop still finishes (reaches maxIterations) and marks end-of-agent.
+    assertThat(
+            events.stream()
+                .anyMatch(event -> event.author().equals("loop") && event.actions().endOfAgent()))
+        .isTrue();
+  }
+
+  @Test
+  public void runAsync_resumable_pausesOnLongRunningCall_doesNotEmitEndOfAgent() {
+    Event longRunningCall =
+        Event.builder()
+            .id("lro")
+            .invocationId("invocationId")
+            .author("sub")
+            .content(
+                Content.fromParts(
+                    Part.builder()
+                        .functionCall(FunctionCall.builder().id("c1").name("tool").build())
+                        .build()))
+            .longRunningToolIds(ImmutableSet.of("c1"))
+            .build();
+    TestBaseAgent subAgent = createSubAgent("sub", longRunningCall);
+    LoopAgent loopAgent =
+        LoopAgent.builder().name("loop").subAgents(subAgent).maxIterations(5).build();
+    InvocationContext context = createResumableInvocationContext(loopAgent);
+
+    List<Event> events = loopAgent.runAsync(context).toList().blockingGet();
+
+    // Paused on the long-running call: no end-of-agent, and the loop did not run 5 iterations.
+    assertThat(events.stream().anyMatch(event -> event.actions().endOfAgent())).isFalse();
+    assertThat(events.stream().filter(event -> event.author().equals("sub")).count()).isEqualTo(1L);
+  }
 }
diff --git a/core/src/test/java/com/google/adk/agents/ParallelAgentTest.java b/core/src/test/java/com/google/adk/agents/ParallelAgentTest.java
index e51240c45..e7f21bc2a 100644
--- a/core/src/test/java/com/google/adk/agents/ParallelAgentTest.java
+++ b/core/src/test/java/com/google/adk/agents/ParallelAgentTest.java
@@ -16,12 +16,20 @@
 
 package com.google.adk.agents;
 
+import static com.google.adk.testing.TestUtils.createFunctionCallLlmResponse;
 import static com.google.adk.testing.TestUtils.createInvocationContext;
+import static com.google.adk.testing.TestUtils.createResumableInvocationContext;
+import static com.google.adk.testing.TestUtils.createTestAgentBuilder;
+import static com.google.adk.testing.TestUtils.createTestLlm;
+import static com.google.adk.testing.TestUtils.createTextLlmResponse;
+import static com.google.adk.testing.TestUtils.simplifyEvents;
 import static com.google.common.truth.Truth.assertThat;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 import com.google.adk.events.Event;
+import com.google.adk.tools.FunctionTool;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.genai.types.Content;
 import com.google.genai.types.Part;
 import io.reactivex.rxjava3.core.Flowable;
@@ -30,6 +38,7 @@
 import io.reactivex.rxjava3.schedulers.TestScheduler;
 import io.reactivex.rxjava3.subscribers.TestSubscriber;
 import java.util.List;
+import org.jspecify.annotations.Nullable;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -194,4 +203,109 @@ public void runAsync_withTestScheduler_usesVirtualTime() {
     testSubscriber.assertValueCount(1);
     testSubscriber.assertComplete();
   }
+
+  // ---- Resumability: checkpoint / skip-completed / pause (parity with Kotlin ParallelAgentTest).
+
+  /** Tools for the resumability tests. */
+  public static final class Tools {
+    private Tools() {}
+
+    // A long-running tool awaiting an external result returns nothing yet, so a branch pauses.
+    @SuppressWarnings("unused") // Invoked reflectively by FunctionTool.
+    public static @Nullable ImmutableMap<String, Object> waitForApproval(String reason) {
+      return null;
+    }
+  }
+
+  @Test
+  public void runAsync_resumable_allSubAgentsEnd_emitsEndOfAgent() {
+    LlmAgent sub1 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("sub1 done")))
+            .name("sub1")
+            .build();
+    LlmAgent sub2 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("sub2 done")))
+            .name("sub2")
+            .build();
+    ParallelAgent parallelAgent =
+        ParallelAgent.builder().name("parallel").subAgents(sub1, sub2).build();
+    InvocationContext context = createResumableInvocationContext(parallelAgent);
+
+    List<Event> events = parallelAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(
+            events.stream()
+                .anyMatch(
+                    event -> event.author().equals("parallel") && event.actions().endOfAgent()))
+        .isTrue();
+  }
+
+  @Test
+  public void runAsync_notResumable_doesNotEmitEndOfAgent() {
+    LlmAgent sub1 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("sub1 done")))
+            .name("sub1")
+            .build();
+    ParallelAgent parallelAgent = ParallelAgent.builder().name("parallel").subAgents(sub1).build();
+    InvocationContext context = createInvocationContext(parallelAgent);
+
+    List<Event> events = parallelAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(events.stream().anyMatch(event -> event.actions().endOfAgent())).isFalse();
+  }
+
+  @Test
+  public void runAsync_resumable_oneBranchPauses_doesNotEmitEndOfAgent() {
+    LlmAgent pausing =
+        createTestAgentBuilder(
+                createTestLlm(
+                    createFunctionCallLlmResponse(
+                        "c1", "waitForApproval", ImmutableMap.of("reason", "x"))))
+            .name("pausing")
+            .tools(
+                FunctionTool.create(
+                    Tools.class,
+                    "waitForApproval",
+                    /* requireConfirmation= */ false,
+                    /* isLongRunning= */ true))
+            .build();
+    LlmAgent completing =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("done")))
+            .name("completing")
+            .build();
+    ParallelAgent parallelAgent =
+        ParallelAgent.builder().name("parallel").subAgents(pausing, completing).build();
+    InvocationContext context = createResumableInvocationContext(parallelAgent);
+
+    List<Event> events = parallelAgent.runAsync(context).toList().blockingGet();
+
+    // One branch paused: the parallel agent does not end.
+    assertThat(
+            events.stream()
+                .anyMatch(
+                    event -> event.author().equals("parallel") && event.actions().endOfAgent()))
+        .isFalse();
+  }
+
+  @Test
+  public void runAsync_resumable_skipsCompletedBranchesOnResume() {
+    LlmAgent sub1 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("sub1 done")))
+            .name("sub1")
+            .build();
+    LlmAgent sub2 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("sub2 done")))
+            .name("sub2")
+            .build();
+    ParallelAgent parallelAgent =
+        ParallelAgent.builder().name("parallel").subAgents(sub1, sub2).build();
+    InvocationContext context = createResumableInvocationContext(parallelAgent);
+    // sub1 already finished in a prior run.
+    context.setAgentState("sub1", /* agentState= */ null, /* endOfAgent= */ true);
+
+    List<Event> events = parallelAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(simplifyEvents(events)).doesNotContain("sub1: sub1 done");
+    assertThat(simplifyEvents(events)).contains("sub2: sub2 done");
+  }
 }
diff --git a/core/src/test/java/com/google/adk/agents/SequentialAgentTest.java b/core/src/test/java/com/google/adk/agents/SequentialAgentTest.java
index 6bbd9e55b..3d525ac53 100644
--- a/core/src/test/java/com/google/adk/agents/SequentialAgentTest.java
+++ b/core/src/test/java/com/google/adk/agents/SequentialAgentTest.java
@@ -19,12 +19,18 @@
 import static com.google.adk.testing.TestUtils.createEvent;
 import static com.google.adk.testing.TestUtils.createInvocationContext;
 import static com.google.adk.testing.TestUtils.createLlmResponse;
+import static com.google.adk.testing.TestUtils.createResumableInvocationContext;
 import static com.google.adk.testing.TestUtils.createSubAgent;
 import static com.google.adk.testing.TestUtils.createTestAgent;
+import static com.google.adk.testing.TestUtils.createTestAgentBuilder;
 import static com.google.adk.testing.TestUtils.createTestLlm;
+import static com.google.adk.testing.TestUtils.createTextLlmResponse;
+import static com.google.adk.testing.TestUtils.simplifyEvents;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static com.google.common.truth.Truth.assertThat;
 
+import com.google.adk.apps.ResumabilityConfig;
+import com.google.adk.artifacts.InMemoryArtifactService;
 import com.google.adk.events.Event;
 import com.google.adk.sessions.InMemorySessionService;
 import com.google.adk.sessions.Session;
@@ -32,6 +38,7 @@
 import com.google.adk.testing.TestLlm;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.genai.types.Content;
 import com.google.genai.types.FunctionCall;
 import com.google.genai.types.FunctionResponse;
@@ -284,4 +291,138 @@ private static InvocationContext contextResumingCall(BaseAgent rootAgent, String
     var unusedResponse = sessionService.appendEvent(session, responseEvent).blockingGet();
     return createInvocationContext(rootAgent, sessionService, session);
   }
+
+  // ---- Resumability: durable checkpoint resume (parity with Kotlin SequentialAgentTest). ----
+
+  @Test
+  public void runAsync_resumingFromMiddle_startsFromCorrectAgent() {
+    LlmAgent agent1 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("a1 done")))
+            .name("agent1")
+            .build();
+    LlmAgent agent2 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("a2 done")))
+            .name("agent2")
+            .build();
+    LlmAgent agent3 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("a3 done")))
+            .name("agent3")
+            .build();
+    SequentialAgent sequentialAgent =
+        SequentialAgent.builder().name("seq").subAgents(agent1, agent2, agent3).build();
+    InvocationContext context = createResumableInvocationContext(sequentialAgent);
+    // Seed the checkpoint: resume at agent2.
+    context.setAgentState(
+        "seq", ImmutableMap.of("current_sub_agent", "agent2"), /* endOfAgent= */ false);
+
+    List<Event> events = sequentialAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(simplifyEvents(events)).doesNotContain("agent1: a1 done");
+    assertThat(simplifyEvents(events)).contains("agent2: a2 done");
+    assertThat(simplifyEvents(events)).contains("agent3: a3 done");
+  }
+
+  @Test
+  public void runAsync_resumable_emitsEndOfAgent() {
+    LlmAgent agent1 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("done"))).name("agent1").build();
+    SequentialAgent sequentialAgent =
+        SequentialAgent.builder().name("seq").subAgents(agent1).build();
+    InvocationContext context = createResumableInvocationContext(sequentialAgent);
+
+    List<Event> events = sequentialAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(
+            events.stream()
+                .anyMatch(event -> event.author().equals("seq") && event.actions().endOfAgent()))
+        .isTrue();
+  }
+
+  @Test
+  public void runAsync_notResumable_doesNotEmitEndOfAgent() {
+    LlmAgent agent1 =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("done"))).name("agent1").build();
+    SequentialAgent sequentialAgent =
+        SequentialAgent.builder().name("seq").subAgents(agent1).build();
+    InvocationContext context = createInvocationContext(sequentialAgent);
+
+    List<Event> events = sequentialAgent.runAsync(context).toList().blockingGet();
+
+    assertThat(events.stream().anyMatch(event -> event.actions().endOfAgent())).isFalse();
+  }
+
+  // Back-compat: a session paused before checkpoints existed must still fast-forward past
+  // completed sub-agents (reconstructed from history) rather than re-running them.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void runAsync_resumeLegacySessionWithoutCheckpoints_doesNotRerunCompletedSubAgents() {
+    TestBaseAgent agent1 =
+        createSubAgent("agent1", createEvent("a1").toBuilder().author("agent1").build());
+    TestBaseAgent agent2 =
+        createSubAgent("agent2", createEvent("a2").toBuilder().author("agent2").build());
+    TestBaseAgent agent3 =
+        createSubAgent("agent3", createEvent("a3").toBuilder().author("agent3").build());
+    SequentialAgent sequentialAgent =
+        SequentialAgent.builder().name("seq").subAgents(agent1, agent2, agent3).build();
+
+    // Simulate a legacy paused session: agent2 issued a long-running call (no checkpoint events),
+    // and the user has now supplied the awaited function response.
+    InMemorySessionService sessionService = new InMemorySessionService();
+    Session session = sessionService.createSession("app", "user").blockingGet();
+    var unusedCall =
+        sessionService
+            .appendEvent(
+                session,
+                Event.builder()
+                    .id("fc")
+                    .invocationId("inv")
+                    .author("agent2")
+                    .content(
+                        Content.fromParts(
+                            Part.builder()
+                                .functionCall(FunctionCall.builder().id("c1").name("tool").build())
+                                .build()))
+                    .longRunningToolIds(ImmutableSet.of("c1"))
+                    .build())
+            .blockingGet();
+    var unusedResponse =
+        sessionService
+            .appendEvent(
+                session,
+                Event.builder()
+                    .id("fr")
+                    .invocationId("inv")
+                    .author("user")
+                    .content(
+                        Content.fromParts(
+                            Part.builder()
+                                .functionResponse(
+                                    FunctionResponse.builder()
+                                        .id("c1")
+                                        .name("tool")
+                                        .response(ImmutableMap.of())
+                                        .build())
+                                .build()))
+                    .build())
+            .blockingGet();
+    InvocationContext context =
+        InvocationContext.builder()
+            .sessionService(sessionService)
+            .artifactService(new InMemoryArtifactService())
+            .invocationId("inv")
+            .agent(sequentialAgent)
+            .session(session)
+            .userContent(Content.fromParts(Part.fromText("resume")))
+            .runConfig(RunConfig.builder().build())
+            .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+            .build();
+    // No populate/checkpoint state seeded -- this is the legacy case.
+
+    var unused = sequentialAgent.runAsync(context).toList().blockingGet();
+
+    // agent1 (already completed before the pause) must not re-run; agent2 and agent3 do.
+    assertThat(agent1.getInvocationCount()).isEqualTo(0);
+    assertThat(agent2.getInvocationCount()).isEqualTo(1);
+    assertThat(agent3.getInvocationCount()).isEqualTo(1);
+  }
 }
diff --git a/core/src/test/java/com/google/adk/events/EventActionsTest.java b/core/src/test/java/com/google/adk/events/EventActionsTest.java
index c5949caf7..77fb805d4 100644
--- a/core/src/test/java/com/google/adk/events/EventActionsTest.java
+++ b/core/src/test/java/com/google/adk/events/EventActionsTest.java
@@ -111,6 +111,61 @@ public void merge_mergesAllFields() {
     assertThat(merged.compaction()).hasValue(COMPACTION);
   }
 
+  @Test
+  public void agentState_roundTripsThroughToBuilder() {
+    EventActions actions =
+        EventActions.builder().agentState(ImmutableMap.of("current_sub_agent", "b")).build();
+
+    EventActions rebuilt = actions.toBuilder().build();
+
+    assertThat(rebuilt).isEqualTo(actions);
+    assertThat(rebuilt.agentState()).hasValue(ImmutableMap.of("current_sub_agent", "b"));
+  }
+
+  @Test
+  public void agentState_roundTripsThroughJson() {
+    EventActions actions =
+        EventActions.builder().agentState(ImmutableMap.of("times_looped", 2)).build();
+
+    EventActions deserialized = EventActions.fromJsonString(actions.toJson(), EventActions.class);
+
+    assertThat(deserialized.agentState()).isPresent();
+    assertThat(deserialized.agentState().get()).containsEntry("times_looped", 2);
+  }
+
+  @Test
+  public void agentState_absentByDefault_andOmittedFromJson() {
+    EventActions actions = EventActions.builder().build();
+
+    assertThat(actions.agentState()).isEmpty();
+    // Kept out of the serialized form so pre-existing events stay byte-identical.
+    assertThat(actions.toJson()).doesNotContain("agentState");
+  }
+
+  @Test
+  public void merge_agentState_lastWins() {
+    EventActions first =
+        EventActions.builder().agentState(ImmutableMap.of("current_sub_agent", "a")).build();
+    EventActions second =
+        EventActions.builder().agentState(ImmutableMap.of("current_sub_agent", "b")).build();
+
+    EventActions merged = first.toBuilder().merge(second).build();
+
+    assertThat(merged.agentState()).hasValue(ImmutableMap.of("current_sub_agent", "b"));
+  }
+
+  @Test
+  public void merge_agentState_disjointKeys_replacesWholeMap() {
+    // agentState is a single checkpoint payload: merge replaces it wholesale (last-wins) rather
+    // than deep-merging keys.
+    EventActions first = EventActions.builder().agentState(ImmutableMap.of("a", 1)).build();
+    EventActions second = EventActions.builder().agentState(ImmutableMap.of("b", 2)).build();
+
+    EventActions merged = first.toBuilder().merge(second).build();
+
+    assertThat(merged.agentState()).hasValue(ImmutableMap.of("b", 2));
+  }
+
   @Test
   public void merge_endOfAgentIsOrderIndependent() {
     // A tool that ends the invocation, and one that leaves the flag at its default false. Folding
diff --git a/core/src/test/java/com/google/adk/flows/llmflows/FunctionsTest.java b/core/src/test/java/com/google/adk/flows/llmflows/FunctionsTest.java
index 8e8555114..12cf1845a 100644
--- a/core/src/test/java/com/google/adk/flows/llmflows/FunctionsTest.java
+++ b/core/src/test/java/com/google/adk/flows/llmflows/FunctionsTest.java
@@ -568,10 +568,44 @@ public void hasPendingLongRunningCall_emptyList_returnsFalse() {
     assertThat(Functions.hasPendingLongRunningCall(ImmutableList.<Event>of())).isFalse();
   }
 
+  @Test
+  public void hasPendingLongRunningCall_list_responseResolvesCall_returnsFalse() {
+    // The trailing function response resolves the pending long-running call, so the flow continues.
+    ImmutableList<Event> events =
+        ImmutableList.of(longRunningCallEvent("call1"), functionResponseEvent("call1"));
+    assertThat(Functions.hasPendingLongRunningCall(events)).isFalse();
+  }
+
+  @Test
+  public void hasPendingLongRunningCall_list_responseForDifferentCall_returnsTrue() {
+    // The response does not resolve the pending call, so the long-running call still pauses.
+    ImmutableList<Event> events =
+        ImmutableList.of(longRunningCallEvent("call1"), functionResponseEvent("other"));
+    assertThat(Functions.hasPendingLongRunningCall(events)).isTrue();
+  }
+
   private static Event longRunningCallEvent(String callId) {
     return functionCallEvent(callId, callId);
   }
 
+  private static Event functionResponseEvent(String callId) {
+    return Event.builder()
+        .id("response_" + callId)
+        .invocationId("invocation1")
+        .author("user")
+        .content(
+            Content.fromParts(
+                Part.builder()
+                    .functionResponse(
+                        FunctionResponse.builder()
+                            .id(callId)
+                            .name("tool")
+                            .response(ImmutableMap.of())
+                            .build())
+                    .build()))
+        .build();
+  }
+
   // Event with a function call; longRunningId, when non-null, is marked long-running.
   private static Event functionCallEvent(String callId, String longRunningId) {
     Event.Builder builder =
diff --git a/core/src/test/java/com/google/adk/runner/RunnerTest.java b/core/src/test/java/com/google/adk/runner/RunnerTest.java
index 3870d3461..c2c4bb77f 100644
--- a/core/src/test/java/com/google/adk/runner/RunnerTest.java
+++ b/core/src/test/java/com/google/adk/runner/RunnerTest.java
@@ -23,12 +23,14 @@
 import static com.google.adk.testing.TestUtils.createTestLlm;
 import static com.google.adk.testing.TestUtils.createTextLlmResponse;
 import static com.google.adk.testing.TestUtils.simplifyEvents;
+import static com.google.adk.testing.TestUtils.simplifyResumableEvents;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.truth.Truth.assertThat;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Arrays.stream;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -54,6 +56,7 @@
 import com.google.adk.artifacts.BaseArtifactService;
 import com.google.adk.artifacts.InMemoryArtifactService;
 import com.google.adk.events.Event;
+import com.google.adk.events.EventActions;
 import com.google.adk.flows.llmflows.Functions;
 import com.google.adk.models.LlmRequest;
 import com.google.adk.models.LlmResponse;
@@ -67,6 +70,7 @@
 import com.google.adk.sessions.SessionKey;
 import com.google.adk.summarizer.EventsCompactionConfig;
 import com.google.adk.telemetry.Tracing;
+import com.google.adk.testing.TestBaseAgent;
 import com.google.adk.testing.TestLlm;
 import com.google.adk.testing.TestUtils;
 import com.google.adk.testing.TestUtils.EchoTool;
@@ -2428,12 +2432,17 @@ public void runAsync_withToolConfirmation_inSequentialAgent_runsLaterSubAgentsAf
             .toList()
             .blockingGet();
 
-    // Turn 2: B resumes and executes the tool, then C runs. A is not re-run.
-    assertThat(simplifyEvents(eventsAfterConfirmation))
+    // Turn 2: B resumes and executes the tool, then C runs (A is not re-run), with per-agent and
+    // workflow checkpoints.
+    assertThat(simplifyResumableEvents(eventsAfterConfirmation))
         .containsExactly(
             "b_agent: FunctionResponse(name=echoTool, response={message=hello})",
             "b_agent: Response after user confirmed.",
-            "c_agent: agent C done")
+            "b_agent: end_of_agent",
+            "workflow_agent: agent_state={current_sub_agent=c_agent}",
+            "c_agent: agent C done",
+            "c_agent: end_of_agent",
+            "workflow_agent: end_of_agent")
         .inOrder();
   }
 
@@ -2446,12 +2455,12 @@ public void runAsync_withLongRunningCall_inSequentialAgent_runsLaterSubAgentsAft
         createTestAgentBuilder(createTestLlm(createTextLlmResponse("agent A done")))
             .name("a_agent")
             .build();
-    // With resumability on, B pauses right after the long-running call (no extra model call), so a
-    // single follow-up response covers the resume.
+    // With resumability on, B pauses right after the no-result long-running call (no extra model
+    // call), so a single follow-up response covers the resume.
     TestLlm bTestLlm =
         createTestLlm(
             createFunctionCallLlmResponse(
-                "lro_call_id", "echoTool", ImmutableMap.of("message", "hello")),
+                "lro_call_id", "pendingTool", ImmutableMap.of("message", "hello")),
             createTextLlmResponse("agent B resumed"));
     LlmAgent agentB =
         createTestAgentBuilder(bTestLlm)
@@ -2459,7 +2468,7 @@ public void runAsync_withLongRunningCall_inSequentialAgent_runsLaterSubAgentsAft
             .tools(
                 FunctionTool.create(
                     Tools.class,
-                    "echoTool",
+                    "pendingTool",
                     /* requireConfirmation= */ false,
                     /* isLongRunning= */ true))
             .build();
@@ -2506,34 +2515,36 @@ public void runAsync_withLongRunningCall_inSequentialAgent_runsLaterSubAgentsAft
                         .functionResponse(
                             FunctionResponse.builder()
                                 .id("lro_call_id")
-                                .name("echoTool")
+                                .name("pendingTool")
                                 .response(ImmutableMap.of("message", "hello")))
                         .build()))
             .toList()
             .blockingGet();
 
-    // Turn 2: B resumes from the long-running response, then C runs. A is not re-run.
-    assertThat(simplifyEvents(eventsAfterResume))
-        .containsExactly("b_agent: agent B resumed", "c_agent: agent C done")
+    // Turn 2: B resumes from the long-running response and C runs (A is not re-run), with per-agent
+    // and workflow checkpoints.
+    assertThat(simplifyResumableEvents(eventsAfterResume))
+        .containsExactly(
+            "b_agent: agent B resumed",
+            "b_agent: end_of_agent",
+            "workflow_agent: agent_state={current_sub_agent=c_agent}",
+            "c_agent: agent C done",
+            "c_agent: end_of_agent",
+            "workflow_agent: end_of_agent")
         .inOrder();
   }
 
-  // Regression: a pending long-running call must pause the LLM flow after a single model call when
-  // resumability is on. Before the flow-level pause, the flow kept re-calling the model (re-issuing
-  // the call), burning tokens. The scripted model would re-issue the call if the flow did not
-  // pause;
-  // we assert exactly one model call was made and the later responses were never consumed.
+  // A value-returning long-running tool is not a pending request: it resolves the call in the same
+  // turn, so even with resumability on the flow continues and the model summarizes the result (two
+  // model calls) rather than pausing. Only a no-result long-running tool pauses.
   @Test
   @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
-  public void runAsync_withLongRunningCall_resumable_pausesAfterSingleModelCall() {
+  public void runAsync_withLongRunningCall_resumable_valueReturn_continuesAndSummarizes() {
     TestLlm testLlm =
         createTestLlm(
             createFunctionCallLlmResponse(
                 "lro_call_id", "echoTool", ImmutableMap.of("message", "hello")),
-            // Extra responses the flow must NOT consume; reaching them means it looped.
-            createFunctionCallLlmResponse(
-                "lro_call_id", "echoTool", ImmutableMap.of("message", "hello")),
-            createTextLlmResponse("should not be reached"));
+            createTextLlmResponse("summarized echo"));
     LlmAgent agent =
         createTestAgentBuilder(testLlm)
             .name("agent")
@@ -2561,9 +2572,10 @@ public void runAsync_withLongRunningCall_resumable_pausesAfterSingleModelCall()
             .toList()
             .blockingGet();
 
-    // The flow paused after the single long-running call instead of re-calling the model.
-    assertThat(testLlm.getRequests()).hasSize(1);
-    assertThat(simplifyEvents(events)).doesNotContain("agent: should not be reached");
+    // The value-returning call was summarized in the same turn: the model was re-invoked (two
+    // calls) and the summary surfaced, with no pause.
+    assertThat(testLlm.getRequests()).hasSize(2);
+    assertThat(simplifyEvents(events)).contains("agent: summarized echo");
   }
 
   // Gating: with resumability OFF (default) the flow does NOT pause on a long-running call; it
@@ -2638,6 +2650,427 @@ public void runAsync_withLongRunningCall_noImmediateResult_endsAfterSingleModelC
     assertThat(simplifyEvents(events)).doesNotContain("agent: should not be reached");
   }
 
+  // A resumable LlmAgent that completes normally emits a trailing end-of-agent checkpoint so a
+  // later run can tell the invocation finished.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void runAsync_resumable_completedLlmAgent_emitsEndOfAgentCheckpoint() {
+    LlmAgent agent =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("all done")))
+            .name("agent")
+            .build();
+    Runner runner =
+        Runner.builder()
+            .app(
+                App.builder()
+                    .name("test")
+                    .rootAgent(agent)
+                    .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+                    .build())
+            .build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+
+    List<Event> events =
+        runner
+            .runAsync("user", session.id(), Content.fromParts(Part.fromText("from user")))
+            .toList()
+            .blockingGet();
+
+    Event last = Iterables.getLast(events);
+    assertThat(last.author()).isEqualTo("agent");
+    assertThat(last.actions().endOfAgent()).isTrue();
+  }
+
+  // Gating: with resumability OFF (default) a completed LlmAgent emits no end-of-agent checkpoint,
+  // keeping the event stream identical to before. Pairs with the resumable test above.
+  @Test
+  public void runAsync_resumabilityDisabled_completedLlmAgent_emitsNoEndOfAgent() {
+    LlmAgent agent =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("all done")))
+            .name("agent")
+            .build();
+    Runner runner =
+        Runner.builder().app(App.builder().name("test").rootAgent(agent).build()).build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+
+    List<Event> events =
+        runner
+            .runAsync("user", session.id(), Content.fromParts(Part.fromText("from user")))
+            .toList()
+            .blockingGet();
+
+    assertThat(events.stream().anyMatch(event -> event.actions().endOfAgent())).isFalse();
+  }
+
+  // resumeAsync on a completed invocation is a no-op: the active agent already ended, so nothing
+  // re-runs. Mirrors the Python/Kotlin resume no-op guard.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void resumeAsync_completedInvocation_isNoOp() {
+    LlmAgent agent =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("all done")))
+            .name("agent")
+            .build();
+    Runner runner =
+        Runner.builder()
+            .app(
+                App.builder()
+                    .name("test")
+                    .rootAgent(agent)
+                    .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+                    .build())
+            .build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+
+    List<Event> firstTurn =
+        runner
+            .runAsync("user", session.id(), Content.fromParts(Part.fromText("from user")))
+            .toList()
+            .blockingGet();
+    String invocationId = firstTurn.get(0).invocationId();
+
+    List<Event> resumed =
+        runner.resumeAsync("user", session.id(), invocationId).toList().blockingGet();
+
+    assertThat(resumed).isEmpty();
+  }
+
+  // resumeAsync with a function response resumes the SAME invocation that issued the matching call
+  // (rather than minting a new invocation id) and runs it to completion.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void resumeAsync_withFunctionResponse_resumesSameInvocation() {
+    TestLlm testLlm =
+        createTestLlm(
+            createFunctionCallLlmResponse(
+                "lro_call_id", "pendingTool", ImmutableMap.of("message", "hello")),
+            createTextLlmResponse("resumed answer"));
+    LlmAgent agent =
+        createTestAgentBuilder(testLlm)
+            .name("agent")
+            .tools(
+                FunctionTool.create(
+                    Tools.class,
+                    "pendingTool",
+                    /* requireConfirmation= */ false,
+                    /* isLongRunning= */ true))
+            .build();
+    Runner runner =
+        Runner.builder()
+            .app(
+                App.builder()
+                    .name("test")
+                    .rootAgent(agent)
+                    .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+                    .build())
+            .build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+
+    List<Event> pausedTurn =
+        runner
+            .runAsync("user", session.id(), Content.fromParts(Part.fromText("from user")))
+            .toList()
+            .blockingGet();
+    String pausedInvocationId = pausedTurn.get(0).invocationId();
+    assertThat(simplifyEvents(pausedTurn)).doesNotContain("agent: resumed answer");
+
+    List<Event> resumed =
+        runner
+            .resumeAsync(
+                "user",
+                session.id(),
+                /* invocationId= */ null,
+                Content.fromParts(
+                    Part.builder()
+                        .functionResponse(
+                            FunctionResponse.builder()
+                                .id("lro_call_id")
+                                .name("pendingTool")
+                                .response(ImmutableMap.of("message", "hello")))
+                        .build()),
+                RunConfig.builder().build())
+            .toList()
+            .blockingGet();
+
+    // The resumed events belong to the original (paused) invocation, not a fresh one.
+    assertThat(resumed).isNotEmpty();
+    assertThat(resumed.stream().allMatch(event -> event.invocationId().equals(pausedInvocationId)))
+        .isTrue();
+    assertThat(simplifyEvents(resumed)).contains("agent: resumed answer");
+    assertThat(resumed.stream().anyMatch(event -> event.actions().endOfAgent())).isTrue();
+  }
+
+  // ResumeInvocationTest parity: resume an OLDER paused invocation (not the latest) via its
+  // long-running function response; the resumed run belongs to that older invocation.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void resumeAsync_resumesAnyInvocation_notJustTheLatest() {
+    TestLlm testLlm =
+        createTestLlm(
+            createFunctionCallLlmResponse(
+                "call-1", "pendingTool", ImmutableMap.of("message", "hi")),
+            createTextLlmResponse("llm response in invocation 2"),
+            createFunctionCallLlmResponse(
+                "call-3", "pendingTool", ImmutableMap.of("message", "hi")),
+            createTextLlmResponse("llm response after resuming invocation 1"));
+    LlmAgent agent =
+        createTestAgentBuilder(testLlm)
+            .name("agent")
+            .tools(
+                FunctionTool.create(
+                    Tools.class,
+                    "pendingTool",
+                    /* requireConfirmation= */ false,
+                    /* isLongRunning= */ true))
+            .build();
+    Runner runner =
+        Runner.builder()
+            .app(
+                App.builder()
+                    .name("test")
+                    .rootAgent(agent)
+                    .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+                    .build())
+            .build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+
+    // Invocation 1 pauses on the long-running call.
+    List<Event> inv1 =
+        runner
+            .runAsync("user", session.id(), Content.fromParts(Part.fromText("q1")))
+            .toList()
+            .blockingGet();
+    String inv1Id = inv1.get(0).invocationId();
+    // Invocation 2 finishes; invocation 3 pauses again.
+    Object unusedInv2 =
+        runner
+            .runAsync("user", session.id(), Content.fromParts(Part.fromText("q2")))
+            .toList()
+            .blockingGet();
+    Object unusedInv3 =
+        runner
+            .runAsync("user", session.id(), Content.fromParts(Part.fromText("q3")))
+            .toList()
+            .blockingGet();
+
+    // Resume invocation 1 (the oldest, not the latest) via its function response.
+    List<Event> resumed =
+        runner
+            .resumeAsync(
+                "user",
+                session.id(),
+                inv1Id,
+                Content.fromParts(
+                    Part.builder()
+                        .functionResponse(
+                            FunctionResponse.builder()
+                                .id("call-1")
+                                .name("pendingTool")
+                                .response(ImmutableMap.of("message", "hi")))
+                        .build()),
+                RunConfig.builder().build())
+            .toList()
+            .blockingGet();
+
+    assertThat(simplifyEvents(resumed)).contains("agent: llm response after resuming invocation 1");
+    assertThat(resumed.stream().allMatch(event -> event.invocationId().equals(inv1Id))).isTrue();
+  }
+
+  // InMemoryRunnerTest parity: resume by invocationId rehydrates the agent's checkpoint state from
+  // history so the running agent observes it.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void resumeAsync_restoresAgentStateFromHistory() {
+    TestBaseAgent agent =
+        new TestBaseAgent(
+            "test_agent",
+            "desc",
+            () -> Flowable.<Event>empty(),
+            /* subAgents= */ null,
+            /* beforeAgentCallbacks= */ null,
+            /* afterAgentCallbacks= */ null);
+    Runner runner =
+        Runner.builder()
+            .app(
+                App.builder()
+                    .name("test")
+                    .rootAgent(agent)
+                    .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+                    .build())
+            .build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+    Object unusedUser =
+        runner
+            .sessionService()
+            .appendEvent(
+                session,
+                Event.builder()
+                    .id("u1")
+                    .invocationId("test-inv")
+                    .author("user")
+                    .content(Content.fromParts(Part.fromText("hi")))
+                    .build())
+            .blockingGet();
+    Object unusedState =
+        runner
+            .sessionService()
+            .appendEvent(
+                session,
+                Event.builder()
+                    .id("s1")
+                    .invocationId("test-inv")
+                    .author("test_agent")
+                    .actions(
+                        EventActions.builder()
+                            .agentState(ImmutableMap.of("saved", "state"))
+                            .build())
+                    .content(Content.fromParts(Part.fromText("previous response")))
+                    .build())
+            .blockingGet();
+
+    Object unused = runner.resumeAsync("user", session.id(), "test-inv").toList().blockingGet();
+
+    assertThat(agent.getLastInvocationContext().agentStates())
+        .containsEntry("test_agent", ImmutableMap.of("saved", "state"));
+  }
+
+  // InMemoryRunnerTest parity: resume by invocationId with a new user message appends that content
+  // under the resumed invocation.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void resumeAsync_withNewMessage_appendsUserContentUnderResumedInvocation() {
+    TestBaseAgent agent =
+        new TestBaseAgent("test_agent", "desc", () -> Flowable.<Event>empty(), null, null, null);
+    Runner runner =
+        Runner.builder()
+            .app(
+                App.builder()
+                    .name("test")
+                    .rootAgent(agent)
+                    .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+                    .build())
+            .build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+    Object unusedUser =
+        runner
+            .sessionService()
+            .appendEvent(
+                session,
+                Event.builder()
+                    .id("u1")
+                    .invocationId("test-inv")
+                    .author("user")
+                    .content(Content.fromParts(Part.fromText("hi")))
+                    .build())
+            .blockingGet();
+
+    Object unused =
+        runner
+            .resumeAsync(
+                "user",
+                session.id(),
+                "test-inv",
+                Content.fromParts(Part.fromText("New message")),
+                RunConfig.builder().build())
+            .toList()
+            .blockingGet();
+
+    Session reloaded =
+        runner
+            .sessionService()
+            .getSession("test", "user", session.id(), Optional.empty())
+            .blockingGet();
+    assertThat(reloaded.events()).hasSize(2);
+    assertThat(
+            Iterables.getLast(reloaded.events())
+                .content()
+                .flatMap(Content::parts)
+                .get()
+                .get(0)
+                .text())
+        .hasValue("New message");
+  }
+
+  // RunnerTest parity (disabled counterpart): resuming a non-resumable app throws.
+  @Test
+  public void resumeAsync_notResumable_throwsException() {
+    LlmAgent agent =
+        createTestAgentBuilder(createTestLlm(createTextLlmResponse("x"))).name("agent").build();
+    Runner runner =
+        Runner.builder().app(App.builder().name("test").rootAgent(agent).build()).build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+
+    assertThrows(
+        IllegalArgumentException.class, () -> runner.resumeAsync("user", session.id(), "some-inv"));
+  }
+
+  // InMemoryRunnerTest parity: the appended function-response inherits the branch of the function
+  // call it answers.
+  @Test
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public void resumeAsync_withFunctionResponse_copiesBranchFromMatchingCall() {
+    TestBaseAgent agent =
+        new TestBaseAgent("test_agent", "desc", () -> Flowable.<Event>empty(), null, null, null);
+    Runner runner =
+        Runner.builder()
+            .app(
+                App.builder()
+                    .name("test")
+                    .rootAgent(agent)
+                    .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+                    .build())
+            .build();
+    Session session = runner.sessionService().createSession("test", "user").blockingGet();
+    Object unusedFc =
+        runner
+            .sessionService()
+            .appendEvent(
+                session,
+                Event.builder()
+                    .id("fc1")
+                    .invocationId("test-inv")
+                    .author("test_agent")
+                    .branch("my_special_branch")
+                    .content(
+                        Content.fromParts(
+                            Part.builder()
+                                .functionCall(
+                                    FunctionCall.builder().id("call_abc").name("test_func").build())
+                                .build()))
+                    .build())
+            .blockingGet();
+
+    Object unused =
+        runner
+            .resumeAsync(
+                "user",
+                session.id(),
+                /* invocationId= */ null,
+                Content.fromParts(
+                    Part.builder()
+                        .functionResponse(
+                            FunctionResponse.builder()
+                                .id("call_abc")
+                                .name("test_func")
+                                .response(ImmutableMap.of("result", "ok")))
+                        .build()),
+                RunConfig.builder().build())
+            .toList()
+            .blockingGet();
+
+    Session reloaded =
+        runner
+            .sessionService()
+            .getSession("test", "user", session.id(), Optional.empty())
+            .blockingGet();
+    Event lastUser =
+        reloaded.events().stream()
+            .filter(event -> event.author().equals("user"))
+            .reduce((first, second) -> second)
+            .get();
+    assertThat(lastUser.branch()).hasValue("my_special_branch");
+  }
+
   // The long-running call event is now a final response, but it carries no text. An agent with an
   // outputKey must not overwrite that key with an empty string. Matches ADK Python's output_key
   // guard, which skips final events that have no text part.
@@ -2766,7 +3199,7 @@ public void runAsync_loopAgentWithLongRunningSubAgent_resumable_stopsAfterFirstI
                 calls.incrementAndGet() <= 5
                     ? Flowable.just(
                         createFunctionCallLlmResponse(
-                            "lro_call_id", "echoTool", ImmutableMap.of("message", "hello")))
+                            "lro_call_id", "pendingTool", ImmutableMap.of("message", "hello")))
                     : Flowable.just(createTextLlmResponse("stop")));
     LlmAgent inner =
         createTestAgentBuilder(loopLlm)
@@ -2774,7 +3207,7 @@ public void runAsync_loopAgentWithLongRunningSubAgent_resumable_stopsAfterFirstI
             .tools(
                 FunctionTool.create(
                     Tools.class,
-                    "echoTool",
+                    "pendingTool",
                     /* requireConfirmation= */ false,
                     /* isLongRunning= */ true))
             .build();
@@ -2814,7 +3247,7 @@ public void runAsync_parallelAgentWithLongRunningBranch_resumable_otherBranchCom
     TestLlm longRunningLlm =
         createTestLlm(
             createFunctionCallLlmResponse(
-                "lro_call_id", "echoTool", ImmutableMap.of("message", "hello")),
+                "lro_call_id", "pendingTool", ImmutableMap.of("message", "hello")),
             createTextLlmResponse("unexpected"));
     LlmAgent longRunningBranch =
         createTestAgentBuilder(longRunningLlm)
@@ -2822,7 +3255,7 @@ public void runAsync_parallelAgentWithLongRunningBranch_resumable_otherBranchCom
             .tools(
                 FunctionTool.create(
                     Tools.class,
-                    "echoTool",
+                    "pendingTool",
                     /* requireConfirmation= */ false,
                     /* isLongRunning= */ true))
             .build();
diff --git a/core/src/test/java/com/google/adk/sessions/SessionJsonConverterTest.java b/core/src/test/java/com/google/adk/sessions/SessionJsonConverterTest.java
index a63e3b38d..90ca6a175 100644
--- a/core/src/test/java/com/google/adk/sessions/SessionJsonConverterTest.java
+++ b/core/src/test/java/com/google/adk/sessions/SessionJsonConverterTest.java
@@ -276,6 +276,46 @@ public void fromApiEvent_complexActions_success() {
     assertThat(eventActions.endOfAgent()).isTrue();
   }
 
+  @Test
+  public void convertEventToJson_agentState_success() throws JsonProcessingException {
+    EventActions actions =
+        EventActions.builder()
+            .agentState(ImmutableMap.of("current_sub_agent", "b_agent", "times_looped", 2))
+            .build();
+    Event event =
+        Event.builder()
+            .author("agent")
+            .invocationId("inv-1")
+            .timestamp(Instant.parse("2023-01-01T00:00:00.123Z").toEpochMilli())
+            .actions(actions)
+            .build();
+
+    String json = SessionJsonConverter.convertEventToJson(event, true);
+    JsonNode actionsNode = objectMapper.readTree(json).get("actions");
+
+    assertThat(actionsNode.get("agentState").get("current_sub_agent").asText())
+        .isEqualTo("b_agent");
+    assertThat(actionsNode.get("agentState").get("times_looped").asInt()).isEqualTo(2);
+  }
+
+  @Test
+  public void fromApiEvent_agentState_success() {
+    Map<String, Object> apiEvent = new HashMap<>();
+    apiEvent.put("name", "sessions/123/events/456");
+    apiEvent.put("invocationId", "inv-1");
+    apiEvent.put("author", "agent");
+    apiEvent.put("timestamp", "2023-01-01T00:00:00.123Z");
+    Map<String, Object> actions = new HashMap<>();
+    actions.put("agentState", ImmutableMap.of("current_sub_agent", "b_agent", "times_looped", 2));
+    apiEvent.put("actions", actions);
+
+    Event event = SessionJsonConverter.fromApiEvent(apiEvent);
+
+    assertThat(event.actions().agentState()).isPresent();
+    assertThat(event.actions().agentState().get()).containsEntry("current_sub_agent", "b_agent");
+    assertThat(event.actions().agentState().get()).containsEntry("times_looped", 2);
+  }
+
   @Test
   public void fromApiEvent_minimalEvent_success() {
     Map<String, Object> apiEvent = new HashMap<>();
diff --git a/core/src/test/java/com/google/adk/testing/TestUtils.java b/core/src/test/java/com/google/adk/testing/TestUtils.java
index daed8d2e4..aa55c72c9 100644
--- a/core/src/test/java/com/google/adk/testing/TestUtils.java
+++ b/core/src/test/java/com/google/adk/testing/TestUtils.java
@@ -24,6 +24,7 @@
 import com.google.adk.agents.InvocationContext;
 import com.google.adk.agents.LlmAgent;
 import com.google.adk.agents.RunConfig;
+import com.google.adk.apps.ResumabilityConfig;
 import com.google.adk.artifacts.InMemoryArtifactService;
 import com.google.adk.events.Event;
 import com.google.adk.events.EventActions;
@@ -71,6 +72,22 @@ public static InvocationContext createInvocationContext(BaseAgent agent) {
     return createInvocationContext(agent, RunConfig.builder().build());
   }
 
+  /** Like {@link #createInvocationContext(BaseAgent)} but with resumability enabled. */
+  @SuppressWarnings("deprecation") // Resumability flag is intentionally deprecated (partial).
+  public static InvocationContext createResumableInvocationContext(BaseAgent agent) {
+    InMemorySessionService sessionService = new InMemorySessionService();
+    return InvocationContext.builder()
+        .sessionService(sessionService)
+        .artifactService(new InMemoryArtifactService())
+        .invocationId("invocationId")
+        .agent(agent)
+        .session(sessionService.createSession("test_app", "test-user").blockingGet())
+        .userContent(Content.fromParts(Part.fromText("user content")))
+        .runConfig(RunConfig.builder().build())
+        .resumabilityConfig(ResumabilityConfig.builder().resumable(true).build())
+        .build();
+  }
+
   public static InvocationContext createInvocationContext(
       BaseAgent agent, BaseSessionService sessionService, Session session) {
     return InvocationContext.builder()
@@ -105,6 +122,30 @@ public static ImmutableList<String> simplifyEvents(List<Event> events) {
         .collect(toImmutableList());
   }
 
+  /** Marker rendered for an end-of-agent checkpoint event by {@link #simplifyResumableEvents}. */
+  public static final String END_OF_AGENT = "end_of_agent";
+
+  /**
+   * Like {@link #simplifyEvents} but renders resumability checkpoint events distinctly: an
+   * end-of-agent event as {@link #END_OF_AGENT} and an agent-state event as {@code
+   * agent_state=...}. Mirrors the Kotlin {@code simplifyResumableEvents} test helper.
+   */
+  public static ImmutableList<String> simplifyResumableEvents(List<Event> events) {
+    return events.stream()
+        .map(event -> event.author() + ": " + formatResumableEvent(event))
+        .collect(toImmutableList());
+  }
+
+  private static String formatResumableEvent(Event event) {
+    if (event.actions().endOfAgent()) {
+      return END_OF_AGENT;
+    }
+    if (event.actions().agentState().isPresent()) {
+      return "agent_state=" + event.actions().agentState().get();
+    }
+    return formatEventContent(event);
+  }
+
   private static String formatEventContent(Event event) {
     return formatContent(
         event