Python: [Bug]: Functional workflow ResponseStream leaks ContextVar tokens across tasks when abandoned mid-iteration (break before exhaustion) #7787

What happened?

FUNCTIONAL_WORKFLOWS' run(stream=True) returns a ResponseStream[WorkflowEvent, WorkflowRunResult] backed by an async generator (FunctionalWorkflowDefinition._run_stream_async-style generator in _functional.py) that holds two kinds of context open across a yield:

• with create_workflow_span(...) as span: (OpenTelemetry) wraps the entire generator body.
• Many individual with _framework_event_origin(): yield <event> pairs (_events.py, used from _functional.py and _executor.py) each set a ContextVar and yield while it's still set.

If a consumer stops iterating before the generator is exhausted — e.g. async for event in stream: break after seeing the first request_info event, a completely normal "peek then decide" pattern — the generator is only closed later, on GC. Python's async-generator finalization throws GeneratorExit into the suspended frame from whatever task/Context happens to be running garbage collection, which is not necessarily the task that was iterating the stream. Every ContextVar.reset(token) the generator was about to run (both _framework_event_origin's own and OpenTelemetry's use_span/start_as_current_span) then raises, because the token was created in a different Context.

This reproduces with no cross-task handoff at all — a single anyio.run(main) coroutine, one task, no ASGI, no asyncio.create_task. The trigger is only "stop iterating early", not "iterate on a different task", so I believe it's a distinct defect from #6866 / #7767 (both of which are about Agent.run(..., stream=True) + observability.py's inner_response_telemetry_captured_fields, triggered by a different consuming task fully exhausting the stream). This report is about the functional-workflows ResponseStream over WorkflowEvents (_functional.py / _events.py / _executor.py), triggered by early abandonment in the same task.

Expected behavior

Abandoning a ResponseStream before it's exhausted (break, a caller-side timeout, an exception in the consumer, etc.) should not raise anything, on any task. The most common ask-then-answer HITL pattern is exactly this shape:

async for event in stream:
    if event.type == "request_info":
        break
# ... decide what to answer, resume later with run(responses=...)

Steps to reproduce

1. pip install agent-framework-core==1.14.0 anyio
2. Run the script below.
3. Observe Task exception was never retrieved / RuntimeError: async generator ignored GeneratorExit, plus (if opentelemetry-api is installed, which is a transitive dependency) Failed to detach context / ValueError: ... was created in a different Context from both OpenTelemetry's use_span and MAF's own _framework_event_origin (_events.py:55).

Code Sample

import gc
import anyio
from agent_framework import step, workflow


@step
async def increment(state: dict) -> dict:
    return {"n": state["n"] + 1}


@workflow(name="increment-pipeline")
async def pipeline(state: dict) -> dict:
    state = await increment(state)
    return await increment(state)


async def main() -> None:
    stream = pipeline.build().run(stream=True, message=({"n": 0},))

    async for _event in stream:
        break  # a normal "peek at the first event, decide" pattern

    del stream
    gc.collect()
    await anyio.sleep(0.1)
    print("done")


anyio.run(main)

Error Messages / Stack Traces

Task exception was never retrieved
future: <Task finished name='Task-2' coro=<<async_generator_athrow without __name__>()> exception=RuntimeError('async generator ignored GeneratorExit')>
RuntimeError: async generator ignored GeneratorExit

Failed to detach context
Traceback (most recent call last):
  File ".../opentelemetry/trace/__init__.py", line 608, in use_span
    yield span
  File ".../opentelemetry/trace/__init__.py", line 508, in start_as_current_span
    yield span
  File ".../opentelemetry/trace/__init__.py", line 443, in start_as_current_span
    yield span
GeneratorExit

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File ".../opentelemetry/context/__init__.py", line 135, in detach
    _RUNTIME_CONTEXT.detach(token)
  File ".../opentelemetry/context/contextvars_context.py", line 42, in detach
    self._current_context.reset(token)
ValueError: <Token var=<ContextVar name='current_context' default={} ...>> was created in a different Context

Exception ignored while closing generator <generator object _framework_event_origin at ...>:
Traceback (most recent call last):
  File ".../agent_framework/_workflows/_events.py", line 55, in _framework_event_origin
    _event_origin_context.reset(token)
ValueError: <Token var=<ContextVar name='workflow_event_origin' default=<WorkflowEventSource.EXECUTOR: 'EXECUTOR'> ...>> was created in a different Context

Package Versions

agent-framework-core: 1.14.0
anyio: 4.14.2
opentelemetry-api: 1.44.0 (transitive)

Python Version

Reproduces on both Python 3.12.3 and 3.14.3.

Additional Context

Likely root cause, for reference:

• agent_framework/_workflows/_events.py:48-55 — _framework_event_origin() is a @contextmanager that does token = _event_origin_context.set(...); yield; finally: _event_origin_context.reset(token).
• agent_framework/_workflows/_functional.py (around lines 1036-1123 in 1.14.0) — the workflow-run generator wraps its entire body in with create_workflow_span(...) as span: and additionally does with _framework_event_origin(): yield WorkflowEvent... at each of the started/status/output/failed event points. The same with _framework_event_origin(): yield ... pattern also appears in agent_framework/_workflows/_executor.py (around lines 303-314).
• Any of these with ...: yield sites can be the one still open when the generator is abandoned; which one depends on how far the consumer got before stopping.

This looks related to #6866 (closed, likely-fixed) and #7767 (open) — same underlying class of bug (a ContextVar token reset assumes it's running in the Context that created it) — but a different subsystem (_functional.py/_events.py functional-workflow event streaming, not observability.py's agent-response telemetry) and a different trigger (early abandonment in the same task vs. full exhaustion in a different task). The fixes referenced there (moving ContextVar creation into the awaited coroutine) may not by themselves cover this path, since here the problem is a context manager left open across a yield inside a generator that can be GC'd from anywhere, not a token created before a task boundary.

Package Versions

agent-framework-core: 1.14.0

Python Version

3.12