Workflow paused on ADK 2.10 or earlier fails to resume when an output_schema agent's stored reply is fenced JSON #7971

Required Information

Describe the Bug:

A workflow session paused on ADK 2.10 or earlier fails to resume on 2.11.0 through 2.13.0 and main when an LlmAgent node with an output_schema had replied with JSON in a Markdown code fence (```json ... ```). The live run accepted that reply. The resumed run raises WorkflowDataError.

• Live: process_llm_agent_output parses the reply with validate_schema (_llm_agent_wrapper.py#L410), which strips the fence first (_schema_utils.py#L253).
• Stored by 2.10 and earlier: the node set message_as_output on every reply, so the runner saved the event with output=None and only the raw, fenced text in content (runners.py#L828-L834). Since 2.11.0 (6b96ba1, fix: preserve single-turn structured output #6100) a schema node no longer sets it (_llm_agent_wrapper.py#L420-L421), so new sessions store the parsed output and resume fine.
• Resume: for an event stored the old way, rehydration takes the message as the node's output (_rehydration_utils.py#L427-L434, #L450-L451). The replay interceptor (_replay_interceptor.py#L129) passes it to _process_rehydrated_output, which parses the text with TypeAdapter(...).validate_json(text) without stripping the fence (#L194-L214). The json.loads fallback fails as well, and the run raises.

So any app that keeps sessions in a persistent store and upgrades from 2.10 or earlier cannot resume the sessions that were paused after such a reply.

Steps to Reproduce:

1. Save the script under Minimal Reproduction Code as repro.py. It uses a scripted model and a SQLite DatabaseSessionService, so no API key is needed.
2. Pause on 2.10.0:
   uv run --no-project --python 3.11 --with google-adk==2.10.0 --with 'sqlalchemy[asyncio]>=2' --with aiosqlite python repro.py pause sessions.db state.json
3. Resume the same session on 2.13.0:
   uv run --no-project --python 3.11 --with google-adk==2.13.0 --with 'sqlalchemy[asyncio]>=2' --with aiosqlite python repro.py resume sessions.db state.json

Expected Behavior:

The resume replays the output the live run parsed, approve gets the same dict again, and the run ends with:

approve got: {'title': 'Weekly', 'bullets': ['one']}
resumed output: {'digest': {'title': 'Weekly', 'bullets': ['one']}, 'approved': True}

Observed Behavior:

  File ".../google/adk/workflow/utils/_replay_interceptor.py", line 137, in check_interception
  File ".../google/adk/workflow/utils/_rehydration_utils.py", line 180, in _process_rehydrated_output
google.adk.workflow._errors.WorkflowDataError: Validation failed for rehydrated output against schema: 1 validation error for Digest
  Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value='```json\n{"title": "Week...bullets": ["one"]}\n```', input_type=str]

Main at 3565ad3 raises the same error for the same stored session.

Environment Details:

• ADK Library Version (pip show google-adk): paused on 2.10.0, resumed on 2.13.0 (latest on PyPI, released 2026-10-23); also resumed on main at 3565ad3 (2026-11-02)
• Desktop OS: macOS
• Python Version (python -V): 3.11.14

Model Information:

• Are you using LiteLLM: No
• Which model is being used: a scripted BaseLlm in the repro. We first saw fenced replies from Claude through google.adk.models.anthropic_llm.Claude on 2.9.

Optional Information

Regression:

No. Up to 2.10.0 every resume after a fenced schema reply failed this way, also on the same version (2.9.2 and 2.10.0 fail when the pause and the resume both run on them). 2.11.0 fixed new sessions by storing the parsed output. Sessions stored before that still go through the path that does not strip the fence.

Additional Context:

_process_rehydrated_output already falls back to plain json.loads "to prevent blocking resumption on schema drift" (#L200-L210), so it is meant to tolerate stored data, and a fence is the same kind of gap.

I have a fix with tests and will open a PR: strip the fence in _process_rehydrated_output with _strip_json_code_fence, the helper validate_schema uses (_schema_utils.py#L215).

Minimal Reproduction Code:

"""Pause on one ADK version, resume on another, with a SQLite session DB.

python repro.py pause  <db_path> <state_json>
python repro.py resume <db_path> <state_json>
"""

import asyncio
import json
import sys
from typing import Any

from google.adk.agents.context import Context
from google.adk.agents.llm_agent import LlmAgent
from google.adk.apps.app import App
from google.adk.events.event import Event
from google.adk.events.request_input import RequestInput
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.database_session_service import DatabaseSessionService
from google.adk.workflow import node
from google.adk.workflow import START
from google.adk.workflow import Workflow
from google.genai import types
from pydantic import BaseModel
import google.adk

FENCED = '```json\n{"title": "Weekly", "bullets": ["one"]}\n```'


class Digest(BaseModel):
  title: str
  bullets: list[str]


class FencedModel(BaseLlm):
  """Replies with fenced JSON, as a model without schema enforcement can."""

  async def generate_content_async(self, llm_request, stream=False):
    yield LlmResponse(
        content=types.Content(role='model', parts=[types.Part(text=FENCED)])
    )


@node(rerun_on_resume=True)
def approve(ctx: Context, node_input: Any):
  print('approve got:', node_input)
  if 'approve' not in ctx.resume_inputs:
    yield RequestInput(interrupt_id='approve', message='Publish?')
    return
  yield Event(output={'digest': node_input, **ctx.resume_inputs['approve']})


async def main(phase: str, db_path: str, state_path: str):
  print('google-adk', google.adk.__version__, phase)
  summarise = LlmAgent(
      name='summarise', model=FencedModel(model='fenced'), output_schema=Digest
  )
  app = App(
      name='repro',
      root_agent=Workflow(name='digest', edges=[(START, summarise, approve)]),
  )
  sessions = DatabaseSessionService(db_url=f'sqlite+aiosqlite:///{db_path}')
  runner = Runner(app=app, session_service=sessions)

  if phase == 'pause':
    session = await sessions.create_session(app_name='repro', user_id='u')
    events = [
        e
        async for e in runner.run_async(
            user_id='u',
            session_id=session.id,
            new_message=types.Content(
                role='user', parts=[types.Part(text='go')]
            ),
        )
    ]
    with open(state_path, 'w') as f:
      json.dump(
          {'session_id': session.id, 'invocation_id': events[0].invocation_id},
          f,
      )
    return

  with open(state_path) as f:
    state = json.load(f)
  answer = types.Part(
      function_response=types.FunctionResponse(
          id='approve', name='adk_request_input', response={'approved': True}
      )
  )
  async for e in runner.run_async(
      user_id='u',
      session_id=state['session_id'],
      new_message=types.Content(role='user', parts=[answer]),
      invocation_id=state['invocation_id'],
  ):
    if e.output is not None:
      print('resumed output:', e.output)


asyncio.run(main(*sys.argv[1:4]))

How often has this issue occurred?:

• Always (100%), for a session paused on 2.10 or earlier after a fenced reply from a schema agent node, then resumed on 2.11.0 or later.