fix(runners): apply state_delta when resuming without a new_message

[lwangverizon]

Link to Issue or Description of Change

1. Link to an existing issue (if applicable):

• Closes: state_delta is silently ignored when resuming an invocation without a new_message #6644
• Related: state_delta passed to /run endpoint is silently ignored in Workflow (NodeRunner) path #5763, fix(runners): Preserve state_delta in NodeRunner path #5767 (the new_message half of the same root cause)

Problem:

Runner.run_async accepts a state_delta argument and documents it as "Optional state
changes to apply to the session", but the delta is only ever applied while appending the
user message event — so both dispatch paths gate it on if new_message::

• node path — runners.py:628 → _append_user_event(..., state_delta=...)
• legacy path — runners.py:2142 in _setup_context_for_resumed_invocation →
  _handle_new_message(..., state_delta=...)

Resuming an invocation by invocation_id with no new_message is a supported call, so in
that case a caller-supplied state_delta is accepted and then silently discarded — no
warning, no error, no event. #5767 fixed this for the new_message path; this is the
remaining half.

Solution:

Apply the delta via a content-less event when there is no user message to carry it.

This mirrors the existing rewind path, which already appends
Event(author='user', actions=EventActions(state_delta=...)) with no content
(runners.py:1373-1381) — so the shape is established in this file rather than new.

The new _append_state_delta_event helper sits next to _append_user_event and reuses the
same surrounding machinery (_apply_run_config_custom_metadata,
stamp_event_branch_context, session_service.append_event), then it is called from the
two if new_message: sites as an elif state_delta: branch. Net effect: a documented
parameter stops losing data, and nothing changes on any path that already worked.

I deliberately did not make this raise or warn on an unsupported combination, and did not
change when state_delta is accepted — the parameter is already public and documented, so
honouring it is the smaller and more backward-compatible change.

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

Added test_run_async_applies_state_delta_when_resuming_without_new_message as the sibling
of the existing test_run_passes_state_delta (tests/unittests/test_runners.py:734), which
covers only the new_message path.

Confirmed the test actually pins the fix — with the runners.py change stashed it fails,
and passes with it:

# fix stashed
1 failed, 1 passed
# fix applied
2 passed

Full file:

$ pytest tests/unittests/test_runners.py -q
79 passed, 17 warnings in 1.53s

Adjacent suites, compared against pristine main @ a5864a0e in a separate worktree:

# main @ a5864a0e   44 failed, 868 passed
# this branch       44 failed, 869 passed   (+1 = the new test)

The 44 failures are pre-existing on main and unrelated to this change (mostly
test_vertex_ai_session_service.py); I diffed the two FAILED lists and they are
identical, so this branch introduces no regressions. test_audio_transcriber.py is excluded
from both runs — it fails to collect on main for a missing test dependency.

pyink --check reports both changed files unchanged.

Manual End-to-End (E2E) Tests:

Ran a standalone script that exercises both runner dispatch paths, since LlmAgent is a
BaseNode and takes the node path while a plain BaseAgent takes the legacy path. It uses a
stub model, so it needs no network, API key, or model access. Each case does a normal run to
create an invocation, then resumes it by invocation_id with a state_delta and no
new_message.

repro.py

import asyncio
from typing import AsyncGenerator

from google.adk.agents.base_agent import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.agents.llm_agent import LlmAgent
from google.adk.apps.app import App, ResumabilityConfig
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.events.event import Event
from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_response import LlmResponse
from google.adk.runners import Runner
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai import types

USER, SESSION = "u1", "s1"


class EchoLlm(BaseLlm):
  """Minimal offline model so the repro needs no network or API key."""

  model: str = "echo"

  async def generate_content_async(
      self, llm_request, stream: bool = False
  ) -> AsyncGenerator[LlmResponse, None]:
    yield LlmResponse(
        content=types.Content(role="model", parts=[types.Part(text="ok")])
    )


class EchoAgent(BaseAgent):
  """Plain BaseAgent (not a BaseNode) -> takes the legacy runner path."""

  async def _run_async_impl(
      self, ctx: InvocationContext
  ) -> AsyncGenerator[Event, None]:
    yield Event(
        invocation_id=ctx.invocation_id,
        author=self.name,
        content=types.Content(role="model", parts=[types.Part(text="ok")]),
    )


async def check(app_name: str, label: str, agent: BaseAgent) -> bool:
  session_service = InMemorySessionService()
  runner = Runner(
      app=App(
          name=app_name,
          root_agent=agent,
          resumability_config=ResumabilityConfig(is_resumable=True),
      ),
      session_service=session_service,
      artifact_service=InMemoryArtifactService(),
  )
  await session_service.create_session(
      app_name=app_name, user_id=USER, session_id=SESSION
  )

  # Turn 1: a normal run, so there is an invocation to resume.
  async for _ in runner.run_async(
      user_id=USER,
      session_id=SESSION,
      new_message=types.Content(role="user", parts=[types.Part(text="hi")]),
  ):
    pass

  session = await session_service.get_session(
      app_name=app_name, user_id=USER, session_id=SESSION
  )
  invocation_id = session.events[0].invocation_id

  # Turn 2: resume by invocation_id, with a state_delta and NO new_message.
  async for _ in runner.run_async(
      user_id=USER,
      session_id=SESSION,
      invocation_id=invocation_id,
      state_delta={"resumed_key": "resumed_value"},
  ):
    pass

  session = await session_service.get_session(
      app_name=app_name, user_id=USER, session_id=SESSION
  )
  applied = session.state.get("resumed_key") == "resumed_value"
  print(f"  {label:<28} state={dict(session.state)!r:<34} applied={applied}")
  return applied


async def main():
  print("resume by invocation_id + state_delta, no new_message:")
  a = await check(
      "app_node", "A) node path (LlmAgent)", LlmAgent(name="root", model=EchoLlm())
  )
  b = await check(
      "app_legacy", "B) legacy path (BaseAgent)", EchoAgent(name="root")
  )
  print(f"\nRESULT: {'PASS' if (a and b) else 'FAIL (state_delta dropped)'}")


asyncio.run(main())

Before — pristine main @ a5864a0e, both paths drop the delta:

resume by invocation_id + state_delta, no new_message:
  A) node path (LlmAgent)      state={}                                 applied=False
  B) legacy path (BaseAgent)   state={}                                 applied=False

RESULT: FAIL (state_delta dropped)

After — this branch, both paths apply it:

resume by invocation_id + state_delta, no new_message:
  A) node path (LlmAgent)      state={'resumed_key': 'resumed_value'}   applied=True
  B) legacy path (BaseAgent)   state={'resumed_key': 'resumed_value'}   applied=True

RESULT: PASS

Checklist

• I have read the CONTRIBUTING.md document.
• I have performed a self-review of my own code.
• I have commented my code, particularly in hard-to-understand areas.
• I have added tests that prove my fix is effective or that my feature works.
• New and existing unit tests pass locally with my changes.
• I have manually tested my changes end-to-end.
• Any dependent changes have been merged and published in downstream modules. — N/A, no dependent changes.

Additional context

Scope note: this only adds an elif branch on the two sites that were already dropping the
delta, plus the helper they call. No existing path changes behavior, and the event it appends
is the same shape the rewind path already writes — so it should be a no-op for anyone not
passing state_delta without a new_message.

While tracing this I checked a third if new_message: site, _handle_new_message
(runners.py:2296 on main), and deliberately left it alone: types.Content defines no
__bool__, so bool(types.Content()) and bool(types.Content(parts=[])) are both True
and the guard there can never be falsy when reached. Adding a branch would have been dead
code.