Don't drop notification answers when the session is busy

nerve/notifications/service.py

@@ -472,22 +472,18 @@ async def handle_answer(
             "content": injected_message,
         })
 
+        # Dispatch unconditionally — ``engine.run`` serializes per
+        # session, so a mid-turn session picks the answer up when its
+        # current turn finishes (FIFO), same pattern as the wakeup
+        # dispatcher. The old ``is_running`` skip here silently dropped
+        # answers that arrived while the session was busy.
         try:
-            if not self.engine.sessions.is_running(session_id):
-                task = asyncio.create_task(
-                    self.engine.run(
-                        session_id=session_id,
-                        user_message=injected_message,
-                        source=f"notification:{answered_by}",
-                        channel=answered_by,
-                    )
-                )
-                task.add_done_callback(self._on_answer_task_done)
-            else:
-                logger.info(
-                    "Session %s running — answer stored, not injected",
-                    session_id,
-                )
+            self._dispatch_into_session(
+                session_id,
+                injected_message,
+                source=f"notification:{answered_by}",
+                channel=answered_by,
+            )
         except Exception as e:
             logger.error(
                 "Failed to inject answer for %s into session %s: %s",
@@ -662,6 +658,34 @@ async def _append_approval_audit(self, event: dict[str, Any]) -> None:
                 exc, event.get("event"),
             )
 
+    def _dispatch_into_session(
+        self,
+        session_id: str,
+        message: str,
+        *,
+        source: str,
+        channel: str | None = None,
+        internal: bool = False,
+    ) -> None:
+        """Fire-and-forget ``engine.run()`` into a session.
+
+        Safe to call while the session is mid-turn: ``engine.run``
+        holds a per-session lock, so the injected message waits behind
+        any in-flight turn and runs when it finishes (FIFO) — the
+        wakeup-dispatcher pattern. Never skip-on-busy here; that drops
+        the message.
+        """
+        task = asyncio.create_task(
+            self.engine.run(
+                session_id=session_id,
+                user_message=message,
+                source=source,
+                channel=channel,
+                internal=internal,
+            )
+        )
+        task.add_done_callback(self._on_answer_task_done)
+
     def _on_answer_task_done(self, task: asyncio.Task) -> None:
         """Log errors from answer injection tasks.
 
@@ -1182,11 +1206,11 @@ async def _inject_expiry_note(
 
         Dispatched unconditionally — ``engine.run`` serializes per
         session, so a mid-turn session just processes the note when its
-        current turn finishes (same pattern as the wakeup dispatcher;
-        deliberately NOT the stale ``is_running`` skip that drops
-        answers). Skipped for external (satellite) sessions, whose
-        conversation loop Nerve doesn't own, and archived/missing
-        sessions — those got the web broadcast and nothing more.
+        current turn finishes (same pattern as the wakeup dispatcher
+        and ``handle_answer``). Skipped for external (satellite)
+        sessions, whose conversation loop Nerve doesn't own, and
+        archived/missing sessions — those got the web broadcast and
+        nothing more.
         """
         try:
             session = await self.db.get_session(session_id)
@@ -1225,15 +1249,12 @@ async def _inject_expiry_note(
             )
             message = f"[Questions expired unanswered]\n{titles}"
 
-        task = asyncio.create_task(
-            self.engine.run(
-                session_id=session_id,
-                user_message=message,
-                source="notification:expiry",
-                internal=True,
-            )
+        self._dispatch_into_session(
+            session_id,
+            message,
+            source="notification:expiry",
+            internal=True,
         )
-        task.add_done_callback(self._on_answer_task_done)
 
     async def _edit_telegram_expired(self, notif: dict[str, Any]) -> None:
         """Best-effort edit of the Telegram card to show it expired.

tests/test_notifications_actionable.py

@@ -559,6 +559,103 @@ async def test_legacy_question_path_still_injects_answer(
         fake_engine.run.assert_called()
 
 
+# ----------------------------------------------------------------------
+#  Busy-session answer injection
+# ----------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+class TestBusySessionAnswerInjection:
+    """Answers must be injected even when the target session is mid-turn.
+
+    Regression tests for the stale ``is_running`` skip: an answer that
+    arrived while the session was busy was marked answered in the DB
+    but never dispatched into the session — silently lost. Dispatch is
+    now unconditional; ``engine.run``'s per-session lock queues the
+    injection behind the in-flight turn (FIFO).
+    """
+
+    async def test_busy_session_still_dispatches_answer(
+        self,
+        db: Database,
+        fake_config: NerveConfig,
+        fake_engine: MagicMock,
+        patch_broadcaster: list,
+    ):
+        await db.create_session("s1")
+        svc = NotificationService(fake_config, db, fake_engine)
+        result = await svc.ask_question(
+            session_id="s1",
+            title="busy question",
+            body="pick one",
+            options=["yes", "no"],
+        )
+        nid = result["notification_id"]
+
+        # Session is mid-turn — the old code skipped injection here
+        # and the answer never reached the agent.
+        fake_engine.sessions.is_running.return_value = True
+
+        ok = await svc.handle_answer(nid, "yes", "web")
+        assert ok is True
+
+        notif = await db.get_notification(nid)
+        assert notif["status"] == "answered"
+
+        # The injected run was dispatched despite the busy session.
+        await asyncio.sleep(0)  # let the fire-and-forget task settle
+        fake_engine.run.assert_called_once()
+        kwargs = fake_engine.run.call_args.kwargs
+        assert kwargs["session_id"] == "s1"
+        assert kwargs["source"] == "notification:web"
+        assert kwargs["channel"] == "web"
+        assert "busy question" in kwargs["user_message"]
+        assert "yes" in kwargs["user_message"]
+
+        # The chat UI still sees the answer immediately, even though
+        # the injected turn is queued behind the in-flight one.
+        injected = [
+            m for _, m in patch_broadcaster
+            if m.get("type") == "answer_injected"
+            and m.get("notification_id") == nid
+        ]
+        assert injected
+
+    async def test_answers_during_busy_turn_dispatch_in_arrival_order(
+        self,
+        db: Database,
+        fake_config: NerveConfig,
+        fake_engine: MagicMock,
+        patch_broadcaster: list,
+    ):
+        await db.create_session("s1")
+        svc = NotificationService(fake_config, db, fake_engine)
+        first = await svc.ask_question(
+            session_id="s1", title="first question", options=["a", "b"],
+        )
+        second = await svc.ask_question(
+            session_id="s1", title="second question", options=["a", "b"],
+        )
+
+        fake_engine.sessions.is_running.return_value = True
+
+        assert await svc.handle_answer(
+            first["notification_id"], "a", "web",
+        )
+        assert await svc.handle_answer(
+            second["notification_id"], "b", "web",
+        )
+
+        await asyncio.sleep(0)
+        assert fake_engine.run.call_count == 2
+        messages = [
+            c.kwargs["user_message"]
+            for c in fake_engine.run.call_args_list
+        ]
+        assert "first question" in messages[0]
+        assert "second question" in messages[1]
+
+
 # ----------------------------------------------------------------------
 #  Handler registry sanity
 # ----------------------------------------------------------------------