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
pip install agent-framework-core==1.14.0 anyio- Run the script below.
- 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
What happened?
FUNCTIONAL_WORKFLOWS'run(stream=True)returns aResponseStream[WorkflowEvent, WorkflowRunResult]backed by an async generator (FunctionalWorkflowDefinition._run_stream_async-style generator in_functional.py) that holds two kinds of context open across ayield:with create_workflow_span(...) as span:(OpenTelemetry) wraps the entire generator body.with _framework_event_origin(): yield <event>pairs (_events.py, used from_functional.pyand_executor.py) each set aContextVarandyieldwhile it's still set.If a consumer stops iterating before the generator is exhausted — e.g.
async for event in stream: breakafter seeing the firstrequest_infoevent, a completely normal "peek then decide" pattern — the generator is only closed later, on GC. Python's async-generator finalization throwsGeneratorExitinto the suspended frame from whatever task/Contexthappens to be running garbage collection, which is not necessarily the task that was iterating the stream. EveryContextVar.reset(token)the generator was about to run (both_framework_event_origin's own and OpenTelemetry'suse_span/start_as_current_span) then raises, because the token was created in a differentContext.This reproduces with no cross-task handoff at all — a single
anyio.run(main)coroutine, one task, no ASGI, noasyncio.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 aboutAgent.run(..., stream=True)+observability.py'sinner_response_telemetry_captured_fields, triggered by a different consuming task fully exhausting the stream). This report is about the functional-workflowsResponseStreamoverWorkflowEvents (_functional.py/_events.py/_executor.py), triggered by early abandonment in the same task.Expected behavior
Abandoning a
ResponseStreambefore 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:Steps to reproduce
pip install agent-framework-core==1.14.0 anyioTask exception was never retrieved/RuntimeError: async generator ignored GeneratorExit, plus (ifopentelemetry-apiis installed, which is a transitive dependency)Failed to detach context/ValueError: ... was created in a different Contextfrom both OpenTelemetry'suse_spanand MAF's own_framework_event_origin(_events.py:55).Code Sample
Error Messages / Stack Traces
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@contextmanagerthat doestoken = _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 inwith create_workflow_span(...) as span:and additionally doeswith _framework_event_origin(): yield WorkflowEvent...at each of thestarted/status/output/failedevent points. The samewith _framework_event_origin(): yield ...pattern also appears inagent_framework/_workflows/_executor.py(around lines 303-314).with ...: yieldsites 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 (aContextVartoken reset assumes it's running in theContextthat created it) — but a different subsystem (_functional.py/_events.pyfunctional-workflow event streaming, notobservability.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 (movingContextVarcreation into the awaited coroutine) may not by themselves cover this path, since here the problem is a context manager left open across ayieldinside 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