Expose a replay indicator on Context #7943

🔴 Required Information

Is your feature request related to a specific problem?

Yes. When a resumable Workflow resumes, ADK answers finished nodes from the session, but an orchestrator node (@node(rerun_on_resume=True), or any body that calls ctx.run_node) re-executes the node body from the top. Every log line, metric increment and alert hook between the ctx.run_node calls fires again on each resume, and nothing marks the emission as replayed. Observability backends show duplicated step lines, overcounted counters, repeated histogram samples and re-fired alerts.

Nothing in the public API says whether the current call is live or replayed (checked in 2.11.0, 2.12.0 and 2.13.0). Context (agents/context.py) exposes run_id, attempt_count, resume_inputs and node_path, and all of them read the same in the live and the replayed call: see the two identical orchestrator started lines in the output below. attempt_count is the retry counter of workflow/_node_runner.py:154 and :218, so it starts at 1 again on resume.

Plugins (plugins/base_plugin.py) get callbacks for agents, model calls and tools, none for a node run (#6779 asks for them). The node class (workflow/_base_node.py) has no lifecycle method to override.

Describe the Solution You'd Like

1. A read-only Context.is_replaying: bool (name open): True while the runtime re-executes a node body on resume and recorded child results remain to be delivered, False for a live call and from the moment the last recorded result was delivered. A line emitted after that point may never have been emitted live, so it must stay.
2. The current Context exposed through a public function over the context variable ADK already keeps (_SYNC_CALLABLE_CONTEXT in utils/_sync_runner.py:49, bound around every node body step in workflow/_base_node.py:201), so a process-wide logging.Filter or a metrics layer can read the flag without every node body checking it.

Other durable-execution runtimes either expose the same flag or handle replay inside their logger:

• Temporal: workflow.unsafe.is_replaying()
• Azure Durable Functions: context.is_replaying
• Dapr Workflow: ctx.is_replaying
• Restate SDK: the logger drops records during replay.
• Inngest: the function logger deduplicates records across resumes.

Impact on your work

We run resumable workflows in production with structured logging, metrics and alerts on every orchestrator step. After a resume, operators cannot trust any step-level signal of that run.

Willingness to contribute

Yes, for the Context indicator, once the maintainers agree on the shape.

🟡 Recommended Information

Describe Alternatives You've Considered

• Checking the session history inside the node body: if an output event for the next child already exists, the call will be replayed. This re-implements ADK's own replay decision, which changed in #7286, #7666 and #7771, so the home-made check silently diverges on upgrade.
• A state marker before each emission (ctx.state["step_1_logged"] = True). Every log line becomes a persisted state delta.
• Letting the duplicates through and removing them in the observability backend. That works for log lines, not for a counter already incremented, a histogram sample already recorded or an alert already fired.

Proposed API / Implementation

class Context:
  @property
  def is_replaying(self) -> bool:
    """True while this node body is re-executed on resume and recorded
    child results remain to be delivered; False otherwise."""

The accessor is one public function beside the context variable ADK already keeps (utils/_sync_runner.py:49) and binds around every node body step (workflow/_base_node.py:201):

def current_context() -> Context | None:
  return _SYNC_CALLABLE_CONTEXT.get()

An application attaches one filter to its log handlers at startup. logging.Filter.filter runs inside log.info(...), in the same task as the node body, so the context variable resolves to that body's ctx; a record logged outside a node sees None and passes:

class DropReplayedRecords(logging.Filter):

  def filter(self, record: logging.LogRecord) -> bool:
    ctx = current_context()
    return not (ctx is not None and ctx.is_replaying)


for handler in logging.getLogger().handlers:
  handler.addFilter(DropReplayedRecords())

ADK already knows at the start of a rerun which children have a recorded result (loop_state.recovered_executions in workflow/_workflow.py); the flag is true until the last of them has been handed back.

Additional Context

google-adk 2.13.0, Python 3.12.8, macOS, no LiteLLM, no model; function nodes and InMemorySessionService only.

Related: #6779 (node hooks), #7666 / #7771 and #7905 (a node that leaves no event reruns on resume; unchanged by this request), #6224 (around-call hooks for durable execution).

Minimal reproduction. An orchestrator logs, runs two children with ctx.run_node, logs after each, and crashes on its first attempt. The resume uses runner.run_async(invocation_id=..., new_message=None).

import asyncio
import logging
import sys

from google.adk.agents import Context
from google.adk.apps import App, ResumabilityConfig
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START, Workflow, node
from google.genai import types

logging.basicConfig(level=logging.INFO, format="%(name)s: %(message)s", stream=sys.stdout)
logging.getLogger("google_adk").setLevel(logging.CRITICAL)  # hides the crash traceback
log = logging.getLogger("app")

attempts = {"orchestrator": 0}
counters = {"orchestrator_started": 0, "child_a_finished": 0, "child_b_finished": 0, "child_runs": 0}


@node
def child_a(node_input):
    counters["child_runs"] += 1
    log.info("child_a ran")
    return "A"


@node
def child_b(node_input):
    counters["child_runs"] += 1
    log.info("child_b ran")
    return node_input + "B"


@node(rerun_on_resume=True)
async def orchestrator(ctx: Context, node_input):
    attempts["orchestrator"] += 1
    counters["orchestrator_started"] += 1
    log.info("orchestrator started (run_id=%s attempt_count=%s resume_inputs=%s)",
             ctx.run_id, ctx.attempt_count, ctx.resume_inputs)
    a = await ctx.run_node(child_a, node_input=node_input)
    counters["child_a_finished"] += 1
    log.info("orchestrator: child_a finished with %r", a)
    ab = await ctx.run_node(child_b, node_input=a)
    counters["child_b_finished"] += 1
    log.info("orchestrator: child_b finished with %r", ab)
    if attempts["orchestrator"] == 1:
        raise RuntimeError("simulated crash after the children finished")
    yield ab


async def main():
    workflow = Workflow(name="root", edges=[(START, orchestrator)])
    app = App(name="repro", root_agent=workflow,
              resumability_config=ResumabilityConfig(is_resumable=True))
    sessions = InMemorySessionService()
    runner = Runner(app=app, session_service=sessions)
    session = await sessions.create_session(app_name="repro", user_id="u")

    invocation_id = None
    print("--- first run ---")
    try:
        async for event in runner.run_async(user_id="u", session_id=session.id,
                                            new_message=types.UserContent("go")):
            invocation_id = event.invocation_id
    except Exception as exc:
        print(f"first run raised {type(exc).__name__}: {exc}")

    print("--- resume ---")
    async for event in runner.run_async(user_id="u", session_id=session.id,
                                        invocation_id=invocation_id, new_message=None):
        if event.output is not None:
            print(f"event: {event.node_info.path} output={event.output!r}")

    print("--- counters ---")
    print(counters)


asyncio.run(main())

Output on google-adk 2.13.0:

--- first run ---
app: orchestrator started (run_id=1 attempt_count=1 resume_inputs={})
app: child_a ran
app: orchestrator: child_a finished with 'A'
app: child_b ran
app: orchestrator: child_b finished with 'AB'
first run raised RuntimeError: simulated crash after the children finished
--- resume ---
app: orchestrator started (run_id=1 attempt_count=1 resume_inputs={})
app: orchestrator: child_a finished with 'A'
app: orchestrator: child_b finished with 'AB'
event: root@1/orchestrator@1 output='AB'
--- counters ---
{'orchestrator_started': 2, 'child_a_finished': 2, 'child_b_finished': 2, 'child_runs': 2}

The children ran once; the orchestrator's three lines and counters fired twice with identical ctx values. With the flag, the resumed started and child_a finished lines are droppable; child_b finished follows the last recorded result and stays.