diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index 1527fb9..1da4a17 100644 --- a/nerve/notifications/service.py +++ b/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. diff --git a/tests/test_notifications_actionable.py b/tests/test_notifications_actionable.py index a84564b..3e456a1 100644 --- a/tests/test_notifications_actionable.py +++ b/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 # ----------------------------------------------------------------------