diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index 42e8ead23..a8e6cb94c 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -646,18 +646,23 @@ async def list_thread_messages( event_store = get_run_event_store(request) messages = await event_store.list_messages(thread_id, limit=limit, before_seq=before_seq, after_seq=after_seq) - # Attach feedback to the last AI message of each run - feedback_repo = get_feedback_repo(request) - user_id = await get_current_user(request) - feedback_map = await feedback_repo.list_by_thread_grouped(thread_id, user_id=user_id) - - # Find the last ai_message per run_id + # Find the last AI message per run_id. AI messages are persisted by + # RunJournal with event_type "llm.ai.response" (see runtime/journal.py); + # the event store returns that value verbatim, so match on it here. last_ai_per_run: dict[str, int] = {} # run_id -> index in messages list for i, msg in enumerate(messages): - if msg.get("event_type") == "ai_message": + if msg.get("event_type") == "llm.ai.response": last_ai_per_run[msg["run_id"]] = i - # Attach feedback field + # Attach feedback to the last AI message of each run. Only query when there + # is an AI message to attach it to — threads with no completed AI turn yet + # would otherwise pay for a grouped feedback lookup whose result is unused. + feedback_map: dict[str, dict] = {} + if last_ai_per_run: + feedback_repo = get_feedback_repo(request) + user_id = await get_current_user(request) + feedback_map = await feedback_repo.list_by_thread_grouped(thread_id, user_id=user_id) + last_ai_indices = set(last_ai_per_run.values()) for i, msg in enumerate(messages): if i in last_ai_indices: diff --git a/backend/tests/test_thread_messages_feedback.py b/backend/tests/test_thread_messages_feedback.py new file mode 100644 index 000000000..b539a5912 --- /dev/null +++ b/backend/tests/test_thread_messages_feedback.py @@ -0,0 +1,76 @@ +"""Regression for the thread-messages feedback attachment. + +GET /api/threads/{thread_id}/messages attaches user feedback to the last AI +message of each run. AI messages are stored by RunJournal with event_type +"llm.ai.response"; the endpoint previously matched the non-existent +"ai_message", so feedback was never attached and the grouped-feedback query +ran on every request for nothing. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +from _router_auth_helpers import make_authed_test_app +from fastapi.testclient import TestClient + +from app.gateway.routers import thread_runs + + +def _make_app(messages, feedback_grouped): + app = make_authed_test_app() + app.include_router(thread_runs.router) + + event_store = MagicMock() + event_store.list_messages = AsyncMock(return_value=messages) + app.state.run_event_store = event_store + + feedback_repo = MagicMock() + feedback_repo.list_by_thread_grouped = AsyncMock(return_value=feedback_grouped) + app.state.feedback_repo = feedback_repo + + return app, feedback_repo + + +def _ai(run_id: str, seq: int, content: str) -> dict: + return {"seq": seq, "run_id": run_id, "event_type": "llm.ai.response", "category": "message", "content": content} + + +def _human(run_id: str, seq: int) -> dict: + return {"seq": seq, "run_id": run_id, "event_type": "llm.human.input", "category": "message", "content": "hi"} + + +def test_feedback_attached_to_last_ai_message_per_run(): + messages = [ + _human("r1", 1), + _ai("r1", 2, "first"), + _ai("r1", 3, "final answer"), # last AI of r1 -> should get feedback + _ai("r2", 4, "other run"), # last AI of r2 -> no feedback row + ] + grouped = {"r1": {"feedback_id": "fb-1", "rating": "up", "comment": "nice"}} + app, feedback_repo = _make_app(messages, grouped) + + resp = TestClient(app).get("/api/threads/t1/messages") + assert resp.status_code == 200 + data = resp.json() + + by_seq = {m["seq"]: m for m in data} + # The bug: this used to be None for every message. + assert by_seq[3]["feedback"] == {"feedback_id": "fb-1", "rating": "up", "comment": "nice"} + # Earlier AI message of the same run and the human message get no feedback. + assert by_seq[2]["feedback"] is None + assert by_seq[1]["feedback"] is None + # r2's last AI message has no feedback row. + assert by_seq[4]["feedback"] is None + feedback_repo.list_by_thread_grouped.assert_awaited_once() + + +def test_no_feedback_query_when_thread_has_no_ai_message(): + messages = [_human("r1", 1)] + app, feedback_repo = _make_app(messages, {}) + + resp = TestClient(app).get("/api/threads/t1/messages") + assert resp.status_code == 200 + assert resp.json()[0]["feedback"] is None + # No AI message -> the grouped feedback query must not run. + feedback_repo.list_by_thread_grouped.assert_not_awaited()