fix(workflow): keep session identity across single-turn node contexts

src/google/adk/workflow/_llm_agent_wrapper.py

@@ -294,7 +294,12 @@ def prepare_llm_agent_context(agent: LlmAgent, ctx: Context) -> Context:
   )
   agent_ctx.isolation_scope = ctx.isolation_scope
 
-  ic.session = ic.session.model_copy(deep=False)
+  # Do not copy `ic.session`: it stays the same object shared with the
+  # parent context. A shallow copy here duplicates `last_update_time` and
+  # the storage revision marker, so a DatabaseSessionService write made
+  # through this node's session (e.g. mid-invocation compaction) would
+  # never be reflected on the parent's session object, causing the next
+  # node's write through the parent copy to be rejected as stale.
   return agent_ctx
 
 

tests/unittests/apps/test_compaction_runner_e2e.py

@@ -24,7 +24,10 @@
 from google.adk.apps.llm_event_summarizer import LlmEventSummarizer
 from google.adk.events.event import Event
 from google.adk.runners import Runner
+from google.adk.sessions.database_session_service import DatabaseSessionService
 from google.adk.sessions.in_memory_session_service import InMemorySessionService
+from google.adk.workflow import START
+from google.adk.workflow._workflow import Workflow
 from google.genai import types
 from google.genai.types import Content
 from google.genai.types import Part
@@ -210,3 +213,80 @@ async def test_runner_appends_sliding_window_compaction_event():
   assert (
       compaction_events
   ), "runner did not append the sliding-window compaction event"
+
+
+@pytest.mark.asyncio
+async def test_mid_workflow_compaction_does_not_stale_later_node_append():
+  """Compacting a non-last Workflow node must not stale-fail later nodes.
+
+  Each single-turn LlmAgent node in a Workflow runs against its own copy of
+  the InvocationContext (see ``prepare_llm_agent_context``). Token-threshold
+  compaction writes through that per-node context's session, so a
+  ``DatabaseSessionService`` (which rejects an ``append_event`` whose
+  in-memory revision marker is behind storage) must still accept later
+  writes made through the shared session object: they should see the marker
+  the compaction write left behind, not a stale copy of it.
+  """
+  agent1_model = testing_utils.MockModel.create(
+      responses=["agent1 turn 1", "agent1 turn 2"]
+  )
+  agent2_model = testing_utils.MockModel.create(
+      responses=["agent2 turn 1", "agent2 turn 2"]
+  )
+  agent1 = Agent(name="agent1", model=agent1_model, mode="single_turn")
+  agent2 = Agent(name="agent2", model=agent2_model, mode="single_turn")
+  workflow = Workflow(
+      name="wf",
+      edges=[(START, agent1), (agent1, agent2)],
+  )
+  app = App(
+      name="test_app",
+      root_agent=workflow,
+      events_compaction_config=EventsCompactionConfig(
+          token_threshold=100,
+          event_retention_size=0,
+          summarizer=LlmEventSummarizer(
+              llm=testing_utils.MockModel.create(responses=["summary"])
+          ),
+      ),
+  )
+  session_service = DatabaseSessionService("sqlite+aiosqlite:///:memory:")
+  await session_service.create_session(
+      app_name="test_app", user_id="u1", session_id="s1"
+  )
+  runner = Runner(app=app, session_service=session_service)
+
+  # Turn 1: short message, well below the token threshold estimate. No
+  # compaction triggered.
+  async for _ in runner.run_async(
+      user_id="u1",
+      session_id="s1",
+      new_message=Content(role="user", parts=[Part(text="hi")]),
+  ):
+    pass
+
+  # Turn 2: a long message pushes agent1's estimated prompt token count
+  # above the threshold, so its compaction request-processor compacts
+  # mid-invocation, before agent2 runs.
+  long_message = "lorem ipsum dolor sit amet " * 40
+  events = [
+      event
+      async for event in runner.run_async(
+          user_id="u1",
+          session_id="s1",
+          new_message=Content(role="user", parts=[Part(text=long_message)]),
+      )
+  ]
+
+  agent2_events = [event for event in events if event.author == "agent2"]
+  assert agent2_events, "agent2's response was not produced/persisted"
+
+  refreshed = await session_service.get_session(
+      app_name="test_app", user_id="u1", session_id="s1"
+  )
+  persisted_agent2_events = [
+      event for event in refreshed.events if event.author == "agent2"
+  ]
+  assert (
+      len(persisted_agent2_events) == 2
+  ), "agent2's response was not persisted to storage"