@@ -13,9 +13,6 @@ FIXED
 `Any` rather than promising the child function's annotated result type.
 Custom result reconstruction still requires `return_type`; runtime decoding is
 unchanged.
-- Async dispatch cleanup no longer hangs when async-generator finalizers
-continually start successor generators; exceeding the cleanup budget is reported
-without changing the snapshotted response.
 - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe
 orchestration time, preventing retries from expiring solely because history is
 replayed later. Retry calculations also accept timezone-aware timestamps.
 
@@ -308,11 +308,17 @@ continue as new, encounter a history failure, or reject a version.
 Local cleanup can await ready work such as `sleep(0)`; it must not wait for new
 durable or external work.
 Scheduling/awaiting durable work during async teardown raises an actionable
-error. Teardown must cooperate with cancellation; a non-quiescent cleanup
-queue is bounded and reported, not allowed to run indefinitely. Each
-async-generator sweep has a step budget fixed at sweep entry; tasks or generators
-created during that sweep cannot increase it. This bounds cooperative scheduling
-steps, not wall-clock time or code that never yields.
+error. Teardown must cooperate with cancellation.
+
+> [!WARNING]
+> Async teardown is capped at about 10,000 event-loop steps per cleanup phase
+> (more for dispatches with many tasks or async generators). Cleanup that is
+> still running when the cap is reached is truncated: remaining callbacks are
+> dropped, unfinished coroutines are closed, and a warning is logged. The
+> response decided before teardown is unchanged. Each async-generator sweep's
+> budget is fixed at sweep entry; tasks or generators created during that sweep
+> cannot increase it. The cap bounds cooperative scheduling steps, not
+> wall-clock time or code that never yields.
 
 Ordinary teardown failures (including `CancelledError`) are logged without
 changing the decided response or skipping other cooperative finalizers.
 
@@ -9,9 +9,6 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 FIXED
 
-- With the corresponding core SDK update, async dispatch cleanup no longer
-hangs when async-generator finalizers continually start successor generators.
-Exceeding the cleanup budget is reported without changing the snapshotted response.
 - With the corresponding core SDK update, activity and sub-orchestration
 `retry_timeout` decisions use replay-safe orchestration time rather than
 wall-clock time. See the [core upgrade guidance](../CHANGELOG.md#unreleased)
 
@@ -39,7 +39,6 @@
 from typing import Any, TypeVar, cast
 
 from durabletask import task
-from durabletask.internal.exceptions import is_control_flow_exception
 
 T = TypeVar("T")
 
@@ -169,6 +168,7 @@ def __init__(
         self._closing = False
         self._fatal: BaseException | None = None
         self._cleanup_failure: BaseException | None = None
+        self._cleanup_truncated = False
         self._root: asyncio.Task[Any] | None = None
 
     def _now(self) -> datetime:
@@ -343,7 +343,7 @@ def resume(self) -> None:
     def _has_durable_work(self) -> bool:
         return any(not durable.is_complete for durable, _ in self._waits.values()) or bool(self._timers)
 
-    def _drain(self) -> int:
+    def _drain(self, cleanup_limit: int | None = None) -> int:
         previous_loop = asyncio._get_running_loop()
         previous_task = asyncio.current_task(previous_loop) if previous_loop is not None else None
         previous_hooks = sys.get_asyncgen_hooks()
@@ -353,7 +353,8 @@ def _drain(self) -> int:
         sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter, finalizer=self._asyncgen_finalizer)
         try:
             steps = 0
-            cleanup_limit = max(10000, 4 * (len(self._tasks) + len(self._asyncgens)))
+            if cleanup_limit is None:
+                cleanup_limit = self._cleanup_budget()
             while self._ready or self._cancel_ready:
                 self._ready.extend(handle for _, _, handle in sorted(self._cancel_ready))
                 self._cancel_ready.clear()
@@ -379,10 +380,7 @@ def _drain(self) -> int:
                     self._deliver_completions()
                 steps += 1
                 if self._closing and steps >= cleanup_limit:
-                    self._record_cleanup_error(RuntimeError(
-                        "Async cleanup did not quiesce; cleanup must cooperate with cancellation."))
-                    self._ready.clear()
-                    self._cancel_ready.clear()
+                    self._report_unquiesced_cleanup()
                     break
             return steps
         finally:
@@ -411,35 +409,50 @@ def call_exception_handler(self, context: dict[str, Any]) -> None:
             error = failure
         self._fatal = error if isinstance(error, BaseException) else RuntimeError(context["message"])
 
+    def _cleanup_budget(self) -> int:
+        # Cooperative cleanup is capped in scheduling steps, not wall-clock time.
+        return max(10000, 4 * (len(self._tasks) + len(self._asyncgens)))
+
+    def _report_unquiesced_cleanup(self) -> None:
+        self._ready.clear()
+        self._cancel_ready.clear()
+        if not self._cleanup_truncated:
+            self._cleanup_truncated = True
+            self._record_cleanup_error(RuntimeError(
+                "Async cleanup did not quiesce; cleanup must cooperate with cancellation."))
+
     def _record_cleanup_error(self, error: BaseException, source: str = "cleanup callback") -> None:
         self._logger.warning("%s: Async %s failed.", self.ctx.instance_id, source, exc_info=error)
         if not isinstance(error, (Exception, asyncio.CancelledError)):
             if _is_process_control(self._cleanup_failure):
                 return
-            if self._cleanup_failure is None or is_control_flow_exception(error):
+            # Keep the first failure unless a process-control exit supersedes it.
+            if self._cleanup_failure is None or _is_process_control(error):
                 self._cleanup_failure = error
 
     def _close_asyncgens(self, closed: dict[int, Any]) -> None:
         # Closing one generator may first-iterate another; sweep until no new generators appear.
         # Closed generators stay referenced so their ids cannot be reused during cleanup.
         steps = 0
         # Cleanup-created tasks and generators must not replenish the sweep's budget.
-        limit = max(10000, 4 * (len(self._tasks) + len(self._asyncgens)))
+        limit = self._cleanup_budget()
         while True:
             pending = [
                 generator for key, generator in self._asyncgens.items() if closed.get(key) is not generator]
             if not pending:
                 return
             if steps >= limit:
-                self._record_cleanup_error(RuntimeError(
-                    "Async cleanup did not quiesce; cleanup must cooperate with cancellation."))
+                self._report_unquiesced_cleanup()
                 return
             for generator in pending:
                 closed[id(generator)] = generator
                 self.create_task(generator.aclose())
-            steps += max(1, self._drain())
+            # Each drain spends from this sweep's remaining budget.
+            steps += max(1, self._drain(limit - steps))
 
     def close(self) -> None:
+        if self._closed:
+            return
         self._closing = True
         try:
             completed_at_close = {future for future in self._tasks if future.done()}
 
@@ -1126,6 +1126,60 @@ async def branch(index: int) -> None:
     assert cleaned == [0, 1]
 
 
+def test_cleanup_keeps_first_custom_exit_over_later_generator_exit() -> None:
+    class CleanupAbort(BaseException):
+        pass
+
+    errors: list[BaseException] = [CleanupAbort("first"), GeneratorExit("later")]
+
+    async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
+        async def branch(index: int) -> None:
+            try:
+                await ctx.wait_for_external_event(f"never-{index}")
+            finally:
+                raise errors[index]
+
+        asyncio.create_task(branch(0))
+        asyncio.create_task(branch(1))
+        await ctx.call_activity("wait")
+
+    complete = _completion(_execute(orchestrator, [], _start()))
+    assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED
+    assert complete.failureDetails.errorType.endswith("CleanupAbort")
+
+
+def test_dispatch_restores_host_async_generator_hooks() -> None:
+    def firstiter(generator: Any) -> None:
+        pass
+
+    def finalizer(generator: Any) -> None:
+        pass
+
+    async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
+        async def resource() -> AsyncIterator[int]:
+            yield 1
+
+        await anext(resource())
+        await ctx.call_activity("wait")
+
+    previous = sys.get_asyncgen_hooks()
+    sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=finalizer)
+    try:
+        _execute(orchestrator, [], _start())
+        assert sys.get_asyncgen_hooks() == (firstiter, finalizer)
+    finally:
+        sys.set_asyncgen_hooks(*previous)
+
+
+def test_loop_close_is_idempotent(monkeypatch: pytest.MonkeyPatch) -> None:
+    ctx = cast(task.OrchestrationContext, SimpleNamespace(current_utc_datetime=NOW, instance_id="id"))
+    loop = DurableEventLoop(ctx, is_complete=lambda: False)
+    loop.close()
+    monkeypatch.setattr(loop, "_drain", lambda *args: pytest.fail("closed loop drained again"))
+    loop.close()
+    assert loop.is_closed()
+
+
 @pytest.mark.parametrize("count", [2, 32, 6000])
 def test_cleanup_closes_async_generators_started_during_generator_cleanup(
         count: int, caplog: pytest.LogCaptureFixture) -> None:
@@ -1146,10 +1200,13 @@ async def resource(index: int) -> AsyncIterator[int]:
 
     actions = _execute(orchestrator, [], _start()).actions
     assert [action.scheduleTask.name for action in actions] == ["wait"]
-    assert closed == list(range(count))
     if count == 6000:
-        assert "Async cleanup did not quiesce" in caplog.text
+        # Each sweep's step budget is fixed at entry, so a long chain is truncated.
+        assert caplog.text.count("Async cleanup did not quiesce") == 1
+        assert closed[:3000] == list(range(3000))
+        assert len(closed) < count
     else:
+        assert closed == list(range(count))
         assert "Async" not in caplog.text