diff --git a/src/kiro_crew/dashboard/chat_runner.py b/src/kiro_crew/dashboard/chat_runner.py index fb163a094c0..2fd5ce46dc6 100644 --- a/src/kiro_crew/dashboard/chat_runner.py +++ b/src/kiro_crew/dashboard/chat_runner.py @@ -5193,9 +5193,54 @@ def _drop_stale_admissions(state: DashboardState, slot: _ChatSlot) -> None: ) +def _on_serving_loop(state: DashboardState) -> bool: + """True only when the CURRENTLY running loop is this dashboard's own. + + The queue-cycle tail can be reached three ways: (1) normally, on the + dashboard's serving loop; (2) off any loop, when a turn coroutine + orphaned as an unjoined background task is finalized by the GC via + ``coro.close()`` after its loop has closed; (3) ON A DIFFERENT loop, + when that GC finalization happens to run while a later, unrelated loop + is active (successive event loops in a test process are the ordinary + way this occurs). Only case (1) may dispatch: cases (2) and (3) would + schedule the successor turn and its background tasks onto a loop that + does not own this turn, consuming the queued turn and running work on + the wrong loop. A bare ``get_running_loop() is not None`` test cannot + tell (1) from (3), so identity against the bound serving loop is what + the guard must key on. + + Returns False when no loop is running, when no serving loop was ever + bound, or when the running loop is not the serving loop. + """ + running = state._running_loop() + if running is None: + return False + serving = state.serving_loop + return serving is not None and running is serving + + async def _start_next_queued_turn(state: DashboardState, slot: _ChatSlot) -> bool: """Dequeue and start one ready Kiro turn, preserving queue semantics.""" + if not _on_serving_loop(state): + # Not on this dashboard's serving loop: nothing may be dispatched. + # This function is reached off the serving loop when a turn's + # coroutine, orphaned as an unjoined background task (a test/harness + # hygiene bug, not a production path -- ``spawn_guarded_turn`` always + # registers its task and ``finish_turn_task`` always retrieves it), + # is finalized by the GC via ``coro.close()`` after the owning loop + # closed. ``close()`` throws ``GeneratorExit`` at whatever await + # point the coroutine was suspended on, and that can land with NO + # loop running or with a LATER, unrelated loop running (successive + # event loops in a test process). Neither owns this turn: dispatching + # would schedule the successor turn and its background tasks onto the + # wrong loop, or -- with no loop -- make ``asyncio.create_task`` raise + # from inside ``_run_chat``'s ``finally`` while it unwinds the + # ``GeneratorExit``, the unraisable exception this guards against. + # Bail out before any queue mutation so the queue is left exactly as + # a legitimate later drain (running ON the serving loop) will find it. + return False + # FIRST, before anything reads the queue: re-assert each entry's # admission-time containment and drop what no longer qualifies (#5911). # Everything below — the note flush peeking at queue[0], the user-intervention @@ -5660,6 +5705,18 @@ async def _run_pending_synthesis(state: DashboardState, slot: _ChatSlot) -> None def _finish_queue_cycle(state: DashboardState, slot: _ChatSlot) -> None: """Start synthesis when eligible, otherwise mark a queue cycle idle.""" + # Every asyncio.create_task below needs THIS dashboard's serving loop + # under it, and this function's tail also runs from inside _run_chat's + # own finally while that finally is unwinding an off-loop GeneratorExit + # (see _start_next_queued_turn and _on_serving_loop). A later unrelated + # loop is not this turn's owner, so identity against the serving loop -- + # not a bare running-loop test -- is what gates dispatch. Folded into + # will_synthesize itself (not just its create_task) so the + # flush-deferred-notes gate right below stays in lockstep with the actual + # dispatch -- a note withheld for a synthesis turn that never gets + # scheduled would otherwise never flush. + loop_running = _on_serving_loop(state) + will_synthesize = ( slot._pending_synthesis and not slot._synthesis_inflight @@ -5669,6 +5726,7 @@ def _finish_queue_cycle(state: DashboardState, slot: _ChatSlot) -> None: and state.subagents is not None and not state.subagents.running_agents_for(f"dashboard:{slot.key}") and slot._subagent_deliveries_inflight == 0 + and loop_running ) # Before any successor is dispatched. A held note's CONTEXT half drains into @@ -5717,24 +5775,27 @@ def _finish_queue_cycle(state: DashboardState, slot: _ChatSlot) -> None: state.refresh_slot_source_status(slot.key) state.push_refresh("history") if not slot._titled: - title_task = asyncio.create_task(_maybe_auto_title(state, slot)) - state._background_tasks.add(title_task) - title_task.add_done_callback(state._background_tasks.discard) + if loop_running: + title_task = asyncio.create_task(_maybe_auto_title(state, slot)) + state._background_tasks.add(title_task) + title_task.add_done_callback(state._background_tasks.discard) else: # Already titled: re-examine an AUTO title at bounded milestones so # long sessions aren't stuck with a name generated from their very # first message. Self-guarding (origin/milestone/in-flight checks in # maybe_refresh_title) — the common case returns without any LLM call. - refresh_task = asyncio.create_task(maybe_refresh_title(state, slot)) - state._background_tasks.add(refresh_task) - refresh_task.add_done_callback(state._background_tasks.discard) + if loop_running: + refresh_task = asyncio.create_task(maybe_refresh_title(state, slot)) + state._background_tasks.add(refresh_task) + refresh_task.add_done_callback(state._background_tasks.discard) # Intent summary for the chat summary panel. Self-guarding: the common case # (feature disabled) returns before any work, and an unchanged transcript is # served from the sidecar cache without a model call. - summary_task = asyncio.create_task(generate_session_summary(state, slot)) - state._background_tasks.add(summary_task) - summary_task.add_done_callback(state._background_tasks.discard) + if loop_running: + summary_task = asyncio.create_task(generate_session_summary(state, slot)) + state._background_tasks.add(summary_task) + summary_task.add_done_callback(state._background_tasks.discard) def _emit_ttft_metric(t0: float, session_key: str, *, is_new: bool, resumed: bool) -> None: diff --git a/test/test_dashboard_chat.py b/test/test_dashboard_chat.py index 7ae6e8f1c98..55dc3557c53 100644 --- a/test/test_dashboard_chat.py +++ b/test/test_dashboard_chat.py @@ -3675,6 +3675,214 @@ async def stream(stream_message: str): ) +class TestQueueDrainOffLoopGuard: + """Regression coverage for the off-loop GeneratorExit crash guarded by + ``_on_serving_loop`` in ``_start_next_queued_turn`` / + ``_finish_queue_cycle``. + + In production a turn's coroutine is never abandoned mid-flight: + ``spawn_guarded_turn`` always registers its task in + ``state._background_tasks`` and ``finish_turn_task`` always retrieves it. + The one way this fires is a test/harness bug -- a ``_run_chat`` task + dropped while still pending, after its own event loop already closed. + Python then finalizes the coroutine via ``close()``, which throws + ``GeneratorExit`` at whatever await point it was suspended on. Before the + fix, unwinding that through ``_run_chat``'s own ``finally`` reached the + queue-drain tail (``_start_next_queued_turn`` / ``_finish_queue_cycle``), + and both unconditionally called ``asyncio.create_task`` -- which calls + ``get_running_loop()`` and raised ``RuntimeError`` with no loop to + schedule onto, replacing the in-flight ``GeneratorExit`` and surfacing as + an unraisable exception blamed on whatever test the GC happened to run + in. + + The guard keys on IDENTITY against the dashboard's bound serving loop, + not on whether *any* loop is running: the GC can finalize the orphaned + coroutine while a LATER, unrelated loop is active (successive event loops + in a test process), and dispatching onto that loop -- which does not own + the turn -- would consume the queued turn and run background work on the + wrong loop. These tests pin the guard directly for both the no-loop case + and the wrong-loop case (without needing the real coroutine machinery) + plus one end-to-end reproduction of the actual race against ``_run_chat``. + """ + + @staticmethod + def _break_running_loop(state, monkeypatch) -> None: + # Simulate GC finalization with NO loop running at all. + monkeypatch.setattr(state, "_running_loop", lambda: None) + + @staticmethod + def _running_loop_is_a_foreign_loop(state, monkeypatch) -> None: + # Simulate GC finalization while a LATER, unrelated loop is running: + # a serving loop was bound at startup, but the loop now under us is a + # different one. ``_on_serving_loop`` must reject this by identity. + import asyncio as _asyncio + + foreign = _asyncio.new_event_loop() + serving = _asyncio.new_event_loop() + state.bind_serving_loop(serving) + monkeypatch.setattr(state, "_running_loop", lambda: foreign) + + @pytest.mark.asyncio + async def test_start_next_queued_turn_returns_false_off_loop_and_leaves_queue_intact( + self, tmp_path, monkeypatch + ) -> None: + from kiro_crew.dashboard import chat_runner + + state = _make_state(tmp_path) + slot = state.get_or_create_slot("off-loop-start") + slot.queue_append("keep this queued") + self._break_running_loop(state, monkeypatch) + + started = await chat_runner._start_next_queued_turn(state, slot) + + assert started is False + assert len(slot._queue) == 1 + assert slot._queue[0]["content"] == "keep this queued" + + @pytest.mark.asyncio + async def test_start_next_queued_turn_returns_false_on_a_foreign_loop_and_leaves_queue_intact( + self, tmp_path, monkeypatch + ) -> None: + """A later, unrelated loop is running -- but it does not own this turn. + + This is the case a bare ``_running_loop() is None`` check misses: the + GC finalizes the orphaned coroutine while a DIFFERENT loop is active, + so ``_running_loop()`` returns a real (non-None) loop that is not the + serving loop. Dispatching here would consume the queued turn onto a + loop that does not own it. ``_on_serving_loop`` must reject it by + identity and leave the queue exactly as a legitimate later drain + (running on the serving loop) will find it. + """ + from kiro_crew.dashboard import chat_runner + + state = _make_state(tmp_path) + slot = state.get_or_create_slot("foreign-loop-start") + slot.queue_append("keep this queued") + self._running_loop_is_a_foreign_loop(state, monkeypatch) + + started = await chat_runner._start_next_queued_turn(state, slot) + + assert started is False + assert len(slot._queue) == 1 + assert slot._queue[0]["content"] == "keep this queued" + + @pytest.mark.asyncio + async def test_finish_queue_cycle_on_a_foreign_loop_schedules_no_background_tasks( + self, tmp_path, monkeypatch + ) -> None: + """Same wrong-loop case for the synthesis/title/summary dispatch tail: + a non-None foreign loop must still schedule nothing.""" + from kiro_crew.dashboard import chat_runner + + state = _make_state(tmp_path) + slot = state.get_or_create_slot("foreign-loop-finish") + slot._titled = False + self._running_loop_is_a_foreign_loop(state, monkeypatch) + + chat_runner._finish_queue_cycle(state, slot) + + assert state._background_tasks == set() + + @pytest.mark.asyncio + async def test_finish_queue_cycle_off_loop_schedules_no_background_tasks( + self, tmp_path, monkeypatch + ) -> None: + from kiro_crew.dashboard import chat_runner + + state = _make_state(tmp_path) + slot = state.get_or_create_slot("off-loop-finish") + slot._titled = False + self._break_running_loop(state, monkeypatch) + + # Every branch below normally schedules a background task (auto-title + # and session-summary); off-loop, none of them may. + chat_runner._finish_queue_cycle(state, slot) + + assert state._background_tasks == set() + + @pytest.mark.asyncio + async def test_finish_queue_cycle_off_loop_skips_synthesis_dispatch( + self, tmp_path, monkeypatch + ) -> None: + from kiro_crew.dashboard import chat_runner + + state = _make_state(tmp_path) + slot = state.get_or_create_slot("off-loop-synth") + slot._pending_synthesis = True + slot._titled = True + state.subagents = MagicMock() + state.subagents.running_agents_for = MagicMock(return_value=[]) + self._break_running_loop(state, monkeypatch) + + chat_runner._finish_queue_cycle(state, slot) + + # Not claimed: nothing would ever clear it, since no synthesis task + # was scheduled to run _run_pending_synthesis's own finally. + assert slot._synthesis_inflight is False + assert state._background_tasks == set() + + def test_run_chat_close_after_loop_closed_does_not_raise(self, tmp_path) -> None: + """The real race: close() a suspended _run_chat coroutine after the + loop it was parked on has already closed. + + Plain sync test (no ``@pytest.mark.asyncio``) so this owns its event + loop outright -- ``loop.run_until_complete`` cannot run nested inside + another already-running loop. + """ + from kiro_crew.dashboard.chat import _run_chat + + state = _make_state(tmp_path) + state.kiro_prerequisite_service = object() + state.broadcast_ws = MagicMock() + state.push_slots_update = MagicMock() + state.context_builder = None + state.consolidator = None + state._hook_store = None + state._yolo = False + + async def _hang(*_args, **_kwargs): + # A real, never-resolving suspension (an actual Future via + # loop.call_later), not an instantly-resolved mock -- this parks + # _run_chat the way a stalled ACP call would, rather than letting + # it complete synchronously. + await asyncio.sleep(999) + + state.sessions.get_or_create = AsyncMock(side_effect=_hang) + slot = state.get_or_create_slot("orphaned-turn") + slot._titled = True + slot.queue_append("keep this queued") + + loop = asyncio.new_event_loop() + try: + task = loop.create_task(_run_chat(state, slot, "first message")) + # Real (short) wall-clock wait so _run_chat actually runs, on + # THIS loop, all the way up to its first genuine suspension -- + # not a single sleep(0) pump relying on scheduling order. + loop.run_until_complete(asyncio.sleep(0.05)) + assert not task.done(), "the hang stub must keep the turn suspended, not finished" + coro = task.get_coro() + # This task is deliberately abandoned pending -- that's the whole + # scenario under test. Left alone, its own eventual __del__ (at + # some later, unpredictable GC moment) logs "Task was destroyed + # but it is pending!" via the (by-then-closed) loop's exception + # handler -- a separate, expected asyncio noise source, not the + # unraisable exception this test verifies is gone. Silence it + # here, deterministically, rather than letting it surface later. + task._log_destroy_pending = False + finally: + loop.close() + + # coro is now suspended with its owning loop already closed -- the + # exact state a GC sweep finds an orphaned spawn_guarded_turn Task + # in. This must not raise: before the fix, unwinding the + # GeneratorExit close() throws here reached _run_chat's queue-drain + # tail, which tried to asyncio.create_task() a successor turn with no + # loop to schedule it onto and raised RuntimeError instead of + # letting the close finish cleanly. + coro.close() + assert coro.cr_frame is None + + # ── History save on close (not per-turn) ──