fix(dashboard): stop _run_chat crashing when GC closes an orphaned turn

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:

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) ──