diff --git a/src/kiro_crew/dashboard/turn_dispatch.py b/src/kiro_crew/dashboard/turn_dispatch.py index 639b0092f46..d790d2c2432 100644 --- a/src/kiro_crew/dashboard/turn_dispatch.py +++ b/src/kiro_crew/dashboard/turn_dispatch.py @@ -344,6 +344,7 @@ def _on_deadline() -> None: _TURN_DEADLINE.set(loop.time() + timeout_secs) task: "asyncio.Task[Any] | None" = None handle: "asyncio.TimerHandle | None" = None + _generator_exit = False try: try: task = asyncio.ensure_future(coro) @@ -370,6 +371,21 @@ def _on_deadline() -> None: # the production path: without this the ceiling is silently absorbed. raise TimeoutError(f"turn exceeded the {timeout_secs:.0f}s ceiling") return result + except GeneratorExit: + # This wrapper is being torn down directly -- its own coroutine object + # is being ``close()``-d, which happens when nothing ever awaited or + # cancelled it through a live Task and the garbage collector reclaims + # it instead (an orphaned ``spawn_guarded_turn`` dispatch nobody + # joined). Unlike a live ``CancelledError`` unwind, there is no + # guarantee the event loop that would drive ``task`` to completion is + # even still running -- ``close()`` resumes this frame synchronously + # from whatever thread the collector runs on, not from a loop + # callback. A coroutine that suspends again while unwinding a + # GeneratorExit gets "coroutine ignored GeneratorExit" from the + # interpreter, so ``_generator_exit`` below tells the ``finally`` to + # skip the join and only cancel best-effort. + _generator_exit = True + raise finally: if handle is not None: handle.cancel() @@ -382,13 +398,20 @@ def _on_deadline() -> None: # The wrapper itself was cancelled, or setup failed after Task # creation. Cancel AND join it: cancellation alone can leave an # unstarted coroutine pending until a later GC cycle. - task.cancel() try: - await task - except BaseException: - # Cleanup must preserve the exception already leaving the - # wrapper (setup failure or caller cancellation). + task.cancel() + except RuntimeError: + # The loop that owned ``task`` is already closed (the + # GeneratorExit case, or a shutdown race). Nothing left to + # schedule the cancellation on. pass + if not _generator_exit: + try: + await task + except BaseException: + # Cleanup must preserve the exception already leaving the + # wrapper (setup failure or caller cancellation). + pass async def bounded_chat_turn(coro: "Coroutine[Any, Any, Any]") -> Any: diff --git a/test/test_turn_dispatch.py b/test/test_turn_dispatch.py index c3a455e6a7a..22a4e17a724 100644 --- a/test/test_turn_dispatch.py +++ b/test/test_turn_dispatch.py @@ -202,6 +202,101 @@ def _record_task(coro): td._TURN_DEADLINE.set(original_deadline) +class TestBoundedTurnGeneratorExit: + """A ``_bounded_turn`` wrapper nobody ever awaited or cancelled through a + live Task -- e.g. a ``spawn_guarded_turn`` dispatch left in + ``state._background_tasks`` when the test's event loop closes -- is + reclaimed by the garbage collector instead, which calls ``close()`` on + its still-suspended coroutine directly. That throws ``GeneratorExit`` at + the ``await task`` suspension point, resumed synchronously by the + collector rather than by a loop callback. The old cleanup path answered + with ``task.cancel(); await task`` unconditionally -- fine for a live + ``CancelledError`` unwind, fatal here: suspending on that second + ``await`` while already unwinding a ``GeneratorExit`` is exactly what + raises "coroutine ignored GeneratorExit" as an unraisable exception + nobody can catch (surfaced in CI as ``PytestUnraisableExceptionWarning``, + misattributed to whichever later test happened to trigger the GC sweep). + """ + + def test_close_after_the_owning_loop_has_closed_does_not_raise(self) -> None: + """Faithful reproduction: a real loop, closed while the wrapper is + still suspended on it, then ``close()`` from entirely outside any + running loop -- exactly what a garbage-collected orphan goes through. + + Not ``@pytest.mark.asyncio``: the whole point is driving the wrapper + to suspension on one real loop, closing THAT loop, and only then + calling ``close()`` -- which needs managing the loop by hand rather + than the fixture's own. + """ + + async def _inner() -> None: + await asyncio.sleep(3600) + + async def _driver(): + bt = td._bounded_turn(_inner(), 60) + # Drive the wrapper to its suspension point at `await task` for + # real, on a live loop -- exactly where a caller leaves it if + # nothing ever awaits or cancels the wrapper itself. + bt.send(None) + return bt + + loop = asyncio.new_event_loop() + try: + bt = loop.run_until_complete(_driver()) + finally: + loop.close() + + # What the garbage collector does to an unreferenced, still-suspended + # coroutine once its owning loop is gone. Must not raise "coroutine + # ignored GeneratorExit" or "RuntimeError: Event loop is closed". + bt.close() + + @pytest.mark.asyncio + async def test_owned_tasks_loop_already_closed_does_not_raise(self, monkeypatch) -> None: + """``Task.cancel()`` itself can raise once its owning loop is closed. + + ``call_soon`` checks ``_check_closed()`` synchronously, so cancelling a + task whose loop closed between dispatch and collection raises + ``RuntimeError: Event loop is closed`` right out of ``task.cancel()``. + Mirrors ``TestBoundedTurnSetup``'s timer-failure setup: forcing entry + into the cleanup path via a rejected timer, but with the owned task's + own ``cancel`` failing too, the way it would against a closed loop. + """ + original_deadline = td._TURN_DEADLINE.get() + created_tasks: list[asyncio.Task[None]] = [] + real_ensure_future = asyncio.ensure_future + + async def _turn() -> None: + raise AssertionError("failed setup must not start the turn") + + def _record_task(coro): + task = real_ensure_future(coro) + monkeypatch.setattr( + task, "cancel", MagicMock(side_effect=RuntimeError("Event loop is closed")) + ) + created_tasks.append(task) + return task + + loop = asyncio.get_running_loop() + turn = _turn() + monkeypatch.setattr(td.asyncio, "ensure_future", _record_task) + monkeypatch.setattr( + loop, + "call_later", + MagicMock(side_effect=RuntimeError("timer rejected deadline")), + ) + try: + # Must not raise the mocked "Event loop is closed" instead of the + # real setup failure -- task.cancel() failing is swallowed, not + # propagated over the exception already unwinding the wrapper. + with pytest.raises(RuntimeError, match="timer rejected deadline"): + await td._bounded_turn(turn, 60) + assert len(created_tasks) == 1 + created_tasks[0].cancel.assert_called_once() + finally: + td._TURN_DEADLINE.set(original_deadline) + + class TestTimeoutCard: def test_names_the_limit_in_hours(self) -> None: assert "2-hour" in td.format_turn_timeout_card(7200.0)