Fix tests and address comments

python/packages/core/agent_framework/_workflows/_runner.py

@@ -114,7 +114,7 @@ class Runner:
             WorkflowRunnerException: If the runner is already reserved or running.
         """
         if self._lifecycle is not _RunnerLifecycle.IDLE:
-            raise WorkflowRunnerException("Runner is already running.")
+            raise WorkflowRunnerException("Runner is already reserved or running.")
         self._lifecycle = _RunnerLifecycle.RESERVED
 
     def release(self) -> None:

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -556,8 +556,7 @@ class Workflow(DictConvertible):
                     self._runner.state.set(WORKFLOW_RUN_KWARGS_KEY, {})
                 self._runner.state.commit()  # Commit immediately so kwargs are available
 
-                # Set streaming mode (always set explicitly per run since
-                # reset_for_new_run() no longer runs to clear it).
+                # Explicitly set streaming mode per run
                 self._runner.context.set_streaming(streaming)
 
                 # Execute initial setup if provided

python/packages/core/tests/workflow/test_runner.py

@@ -335,7 +335,7 @@ def test_runner_reserve_twice_raises():
     """
     runner = _make_runner()
     runner.reserve()
-    with pytest.raises(WorkflowRunnerException, match="Runner is already running."):
+    with pytest.raises(WorkflowRunnerException, match="Runner is already reserved or running."):
         runner.reserve()
 
 
@@ -428,7 +428,7 @@ async def test_runner_rejects_concurrent_run_until_convergence():
     await started.wait()  # first run is now executing
 
     try:
-        with pytest.raises(WorkflowRunnerException, match="Runner is already running."):
+        with pytest.raises(WorkflowRunnerException, match="Runner is already reserved or running."):
             runner.reserve()
     finally:
         release.set()