[cyhandsome-123]

Link to Issue or Description of Change

Problem: Existing workflow tests check that siblings receive cancellation
and that a timed-out node retries. They do not hold an asynchronous finalizer
open to verify that the invocation waits for resource release, or that a retry
cannot overlap the previous attempt's cleanup.

Solution: Add Runner-based contract tests to the existing failure and timeout
test modules:

• Parameterize sibling failure and caller cancellation across static, nested,
  and dynamically scheduled children. Keep cleanup blocked, verify the
  invocation remains pending, then verify resource release, the original error
  or cancellation, and a single worker attempt despite a configured retry policy.
• Hold the first timed-out attempt's finalizer open, then verify cleanup precedes
  attempt 2 and only the recovered output reaches the downstream node.

The tests use real Workflow/Runner components and the existing invariant-checking
test helper. Events coordinate execution; a 100 ms non-cancelling observation
window checks that blocked cleanup prevents early completion. Five-second
watchdogs bound the positive waits. This change adds seven cases in two existing
files and leaves runtime behavior and public APIs unchanged.

Testing Plan

Unit Tests:

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

Windows, Python 3.12.14, base 557743c4484c7745f6c52b3bca07727011852e78:

python -m pytest tests/unittests/workflow/test_workflow_failures.py tests/unittests/workflow/test_workflow_node_timeout.py -q
# 48 passed

The same 48 cases also pass against a wheel installed into a fresh virtual
environment. Both pinned format checks (isort 8.0.1 and pyink 25.12.0) and the
repository compliance/link checks pass.

Two process-local mutation checks validate the assertions, without changing
the submitted source: skipping the workflow cleanup gather fails all six
cancellation cases; retrying immediately after requesting timeout cancellation
fails the timeout case. The unchanged implementation passes all seven cases.

Broader validation limitations:

• pytest tests/unittests/workflow -n 4 -q: 1170 passed, 1 xfailed, 2 failed,
  with no collection errors. Both failures are the unchanged
  test_execute_bash_tool_node_runs_command_after_approval variants, reproduced
  independently; ExecuteBashTool returns a POSIX-only error on Windows.
• The full tox matrix did not run tests: Python 3.11, 3.12 and 3.13 setup failed
  with uv's "Missing expected target directory for Python minor version link";
  the 3.14 setup attempt was stopped after the repeated setup failures.
• pre-commit run --files <the two changed tests> passed, and the normal
  commit hook passed without bypasses. The all-files run passed formatting,
  lint, compliance, documentation and spelling checks; two POSIX script hooks
  (check-new-py-prefix and update-constraints) could not start because
  /bin/bash is unavailable to Windows Python. This is not a claim that the
  full suite or every all-files hook is green.

Manual End-to-End (E2E) Tests:

Built the wheel with the project's flit backend, installed it and its runtime
dependencies in a fresh venv, and ran a standalone public InMemoryRunner
scenario with a dynamic worker, async resource finalizer and failing sibling.
No model endpoint, cloud credentials or external service is needed.

PASS: dynamic worker released its async resource before Runner returned.
PASS: original sibling failure propagated; cancelled worker was not retried.
lifecycle=['opened', 'closed']; attempts=[1]; events=1

Reproduce the offline Runner check

After installing the locally built wheel, save this as lifecycle_demo.py and
run python lifecycle_demo.py:

"""Offline public-Runner verification for workflow resource cleanup."""
import asyncio
from contextlib import asynccontextmanager

from google.adk.agents.context import Context
from google.adk.runners import InMemoryRunner
from google.adk.workflow import node
from google.adk.workflow import RetryConfig
from google.adk.workflow import START
from google.adk.workflow import Workflow
from google.genai import types


async def main():
    ready = asyncio.Event()
    lifecycle = []
    attempts = []

    @asynccontextmanager
    async def resource():
        lifecycle.append('opened')
        try:
            yield
        finally:
            # A real asynchronous close would yield control here as well.
            await asyncio.sleep(0)
            lifecycle.append('closed')

    @node(retry_config=RetryConfig(max_attempts=3, initial_delay=0, jitter=0))
    async def worker(ctx: Context):
        attempts.append(ctx.attempt_count)
        async with resource():
            ready.set()
            await asyncio.Event().wait()

    @node(rerun_on_resume=True)
    async def orchestrator(ctx: Context):
        return await ctx.run_node(worker)

    @node
    async def failing_sibling():
        await ready.wait()
        raise ValueError('intentional sibling failure')

    wf = Workflow(name='resource_demo', edges=[(START, orchestrator), (START, failing_sibling)])
    async with InMemoryRunner(node=wf) as runner:
        session = await runner.session_service.create_session(app_name=runner.app_name, user_id='demo')
        events = []
        try:
            async for event in runner.run_async(user_id='demo',session_id=session.id,new_message=types.Content(role='user',parts=[types.Part(text='start')])):
                events.append(event)
        except ValueError as error:
            assert str(error) == 'intentional sibling failure'
        else:
            raise AssertionError('Expected the sibling failure to propagate')
        assert lifecycle == ['opened','closed'], lifecycle
        assert attempts == [1], attempts
        assert not any(event.output is not None for event in events)
        print('PASS: dynamic worker released its async resource before Runner returned.')
        print('PASS: original sibling failure propagated; cancelled worker was not retried.')
        print(f'lifecycle={lifecycle}; attempts={attempts}; events={len(events)}')


asyncio.run(asyncio.wait_for(main(), timeout=10))

The logged ValueError is the deliberately injected sibling failure. The script
asserts that the original error is propagated and exits successfully.

Checklist

• I have read CONTRIBUTING.md.
• I have performed a self-review of my own code.
• I have commented hard-to-understand test synchronization.
• I have added tests that verify the existing lifecycle contract.
• New and existing unit tests pass locally with my changes (full suite).
• I have manually tested the Runner flow end-to-end, including a fresh wheel install.
• There are no dependent downstream changes.

Additional context

Prepared with AI assistance and verified locally. This is a focused test-only
contribution, with the validation limitations above left explicit for review.

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[adk-bot]

Thank you @cyhandsome-123 for your contribution! 🎉

Your changes have been successfully imported and merged via Copybara in commit 3e59429.

Closing this PR as the changes are now in the main branch.