Python: [Bug]: Closing a FunctionalWorkflowAgent stream leaves the workflow marked as running #10164

Observed Behavior

Closing the response stream returned by FunctionalWorkflow.as_agent().run(..., stream=True) leaves the underlying workflow marked as running. A subsequent call on the same caller-scoped agent raises RuntimeError, even though await response.close() has returned.

This affects cancellation or disconnect cleanup followed by another turn. Closing before the first pull reproduces it without executing the workflow function. Closing after the first output also makes an immediate follow-up run fail.

Expected Behavior

Explicitly closing the agent response stream should close its owned workflow stream and release the run state before returning. The next sequential run should succeed, as it does when closing a stream returned directly by FunctionalWorkflow.run().

Steps to Reproduce

1. Use agent-framework-core from commit 924a6e50ec0336f3c965ee8b1151c54085888b6e (current main, package version 1.25.0).
2. Run the snippet below. No model, credentials or external service is needed.
3. The second run raises instead of returning result: 2.

Minimal Reproduction

import asyncio

from agent_framework import workflow


@workflow
async def pipeline(value: int) -> str:
    return f"result: {value}"


async def main() -> None:
    agent = pipeline.build().as_agent()
    response = agent.run(1, stream=True)
    await response.close()
    print((await agent.run(2)).text)


asyncio.run(main())

Error Messages and Stack Traces

  File "agent_framework/_workflows/_functional.py", line 2062, in run
    self._ensure_not_running()
  File "agent_framework/_workflows/_functional.py", line 2664, in _ensure_not_running
    raise RuntimeError("Workflow is already running. Concurrent executions are not allowed.")
RuntimeError: Workflow is already running. Concurrent executions are not allowed.

Package Versions

• agent-framework-core: 1.25.0, installed from the unmodified checkout above
• msgspec: 0.22.0; pydantic: 2.14.3; opentelemetry-api: 1.45.2

Python Version

CPython 3.12.13

Operating System

macOS, arm64

Regression

Unknown

Additional Context

Five local controls used the same workflow and a fresh instance for each case:

After deleting the closed stream, calling gc.collect() and yielding three event-loop ticks, the before-first-pull agent case still raises. The partially consumed agent case recovers only after that delayed cleanup.

FunctionalWorkflowAgent._run_streaming() eagerly creates the inner workflow stream, acquiring the run guard. It then returns a separate ResponseStream around _generate_updates(), with no cleanup hook or finally that closes the inner stream. The direct workflow stream has the cleanup hook that releases its guard.

This concerns the current experimental FUNCTIONAL_WORKFLOWS adapter. #8949 tracks the broader vNext design; this report asks whether propagating close to the owned inner stream is an appropriate bounded fix for the current API. No implementation has been started. The direct-workflow ContextVar issue in #7787 is separate; the direct close controls above pass.

I personally ran the minimal reproduction above on this checkout and observed the reported RuntimeError. Codex assisted with the source analysis and report preparation, and executed the additional five-case controls locally.

Acknowledgements

• I searched existing issues and did not find a duplicate.
• I personally verified this behavior and the reproduction details are authentic.
• I will wait for explicit maintainer agreement before starting implementation of a non-trivial change.