from __future__ import annotations import importlib from datetime import datetime from types import SimpleNamespace import pytest import pytest_asyncio import sqlalchemy as sa from alembic.migration import MigrationContext from alembic.operations import Operations from fastapi import HTTPException from pydantic import ValidationError from sqlalchemy import func, select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from starlette.requests import Request from app.gateway.routers import feedback as feedback_router from app.gateway.routers.feedback import ( FeedbackWriteRequest, _find_ai_message_events, ) from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.feedback.scoring import calculate_answer_quality_score from deerflow.persistence.feedback.sql import FeedbackRepository @pytest_asyncio.fixture async def feedback_repository(tmp_path): engine = create_async_engine( f"sqlite+aiosqlite:///{tmp_path / 'feedback.db'}", ) async with engine.begin() as connection: await connection.run_sync(FeedbackRow.__table__.create) session_factory = async_sessionmaker(engine, expire_on_commit=False) try: yield FeedbackRepository(session_factory), session_factory finally: await engine.dispose() def test_feedback_request_enforces_reason_contract() -> None: positive = FeedbackWriteRequest(rating=1) assert positive.reasons == [] negative = FeedbackWriteRequest( rating=-1, reasons=["incorrect_or_incomplete", "incorrect_or_incomplete"], detail="事实不准确", ) assert [reason.value for reason in negative.reasons] == [ "incorrect_or_incomplete" ] assert negative.normalized_detail() == "事实不准确" with pytest.raises(ValidationError): FeedbackWriteRequest(rating=-1) with pytest.raises(ValidationError): FeedbackWriteRequest(rating=1, reasons=["other"]) def test_answer_quality_score_is_null_without_real_evidence() -> None: result = calculate_answer_quality_score([], agent_id="agent-a") assert result["score"] is None assert result["task_score"] is None assert result["score_band"]["key"] == "insufficient_data" assert result["confidence"] == 0 assert result["confidence_level"] == "none" assert all(item["score"] is None for item in result["dimensions"].values()) def test_answer_quality_score_maps_only_documented_reasons() -> None: result = calculate_answer_quality_score( [ {"rating": 1, "reasons": []}, {"rating": -1, "reasons": ["incorrect_or_incomplete"]}, {"rating": -1, "reasons": ["style_or_tone"]}, ], agent_id="agent-a", ) assert result["score"] == 69.2 assert result["task_success_rate"] == pytest.approx(1 / 3, abs=0.0001) assert result["confidence"] == pytest.approx(3 / 13, abs=0.0001) assert result["dimensions"]["correctness"]["negative_count"] == 1 assert result["dimensions"]["completeness"]["negative_count"] == 1 assert result["dimensions"]["evidence_grounding"]["negative_count"] == 1 assert result["dimensions"]["relevance"]["negative_count"] == 0 assert result["dimensions"]["relevance"]["evidence_count"] == 1 assert result["reason_breakdown"]["style_or_tone"] == { "count": 1, "rate": 0.5, } class _AgentScoreRepository: def __init__(self) -> None: self.agent_ids: list[str] = [] async def aggregate_by_agent(self, agent_id: str) -> dict: self.agent_ids.append(agent_id) return calculate_answer_quality_score([], agent_id=agent_id) @pytest.mark.asyncio async def test_agent_feedback_score_rejects_non_admin( monkeypatch, ) -> None: async def current_user(_request): return SimpleNamespace(system_role="user") repository = _AgentScoreRepository() request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(feedback_repo=repository)) ) monkeypatch.setattr(feedback_router, "get_optional_user_from_request", current_user) with pytest.raises(HTTPException) as exc_info: await feedback_router.get_agent_feedback_score("agent-a", request) assert exc_info.value.status_code == 403 assert repository.agent_ids == [] @pytest.mark.asyncio async def test_agent_feedback_score_allows_admin(monkeypatch) -> None: async def current_user(_request): return SimpleNamespace(system_role="admin") repository = _AgentScoreRepository() request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(feedback_repo=repository)) ) monkeypatch.setattr(feedback_router, "get_optional_user_from_request", current_user) result = await feedback_router.get_agent_feedback_score(" agent-a ", request) assert result["agent_id"] == "agent-a" assert repository.agent_ids == ["agent-a"] @pytest.mark.asyncio async def test_repository_upserts_per_user_thread_message_and_agent( feedback_repository, ) -> None: repository, session_factory = feedback_repository first = await repository.upsert( run_id="run-1", thread_id="thread-1", message_id="message-1", agent_id="agent-a", user_id="user-a", rating=-1, reasons=["incorrect_or_incomplete"], comment="missing evidence", ) updated = await repository.upsert( run_id="run-1", thread_id="thread-1", message_id="message-1", agent_id="agent-a", user_id="user-a", rating=1, reasons=[], comment=None, ) second_answer = await repository.upsert( run_id="run-1", thread_id="thread-1", message_id="message-2", agent_id="agent-a", user_id="user-a", rating=-1, reasons=["unmet_expectations"], ) other_user = await repository.upsert( run_id="run-1", thread_id="thread-1", message_id="message-1", agent_id="agent-a", user_id="user-b", rating=-1, reasons=["other"], ) assert updated["feedback_id"] == first["feedback_id"] assert updated["rating"] == 1 assert second_answer["feedback_id"] != first["feedback_id"] assert other_user["feedback_id"] != first["feedback_id"] async with session_factory() as session: assert await session.scalar(select(func.count()).select_from(FeedbackRow)) == 3 score = await repository.aggregate_by_agent("agent-a") assert score["sample_count"] == 3 assert score["positive_count"] == 1 assert score["negative_count"] == 2 @pytest.mark.asyncio async def test_legacy_run_level_target_remains_idempotent(feedback_repository) -> None: repository, _ = feedback_repository first = await repository.upsert( run_id="run-legacy", thread_id="thread-legacy", agent_id="lead_agent", user_id="user-a", rating=1, ) updated = await repository.upsert( run_id="run-legacy", thread_id="thread-legacy", agent_id="lead_agent", user_id="user-a", rating=-1, reasons=["other"], ) assert updated["feedback_id"] == first["feedback_id"] class _FakeEventStore: def __init__(self, rows: list[dict]) -> None: self.rows = rows async def list_messages( self, thread_id: str, *, limit: int, before_seq: int | None, ) -> list[dict]: rows = [row for row in self.rows if row["thread_id"] == thread_id] if before_seq is not None: rows = [row for row in rows if row["seq"] < before_seq] return rows[-limit:] @pytest.mark.asyncio async def test_batch_message_resolver_uses_serialized_ai_message_id() -> None: event_store = _FakeEventStore( [ { "thread_id": "thread-1", "run_id": "run-1", "seq": 1, "event_type": "llm.tool.result", "content": {"id": "tool-1"}, }, { "thread_id": "thread-1", "run_id": "run-1", "seq": 2, "event_type": "llm.ai.response", "content": {"id": "answer-1", "type": "ai"}, }, ] ) app = SimpleNamespace(state=SimpleNamespace(run_event_store=event_store)) request = Request({"type": "http", "app": app}) found = await _find_ai_message_events( request, thread_id="thread-1", message_ids={"answer-1", "tool-1", "missing"}, ) assert set(found) == {"answer-1"} assert found["answer-1"]["run_id"] == "run-1" def test_answer_feedback_migration_upgrades_and_downgrades_sqlite( tmp_path, monkeypatch, ) -> None: engine = sa.create_engine(f"sqlite:///{tmp_path / 'migration.db'}") metadata = sa.MetaData() runs = sa.Table( "runs", metadata, sa.Column("run_id", sa.String(64), primary_key=True), sa.Column("assistant_id", sa.String(128)), ) feedback = sa.Table( "feedback", metadata, sa.Column("feedback_id", sa.String(64), primary_key=True), sa.Column("run_id", sa.String(64), nullable=False), sa.Column("thread_id", sa.String(64), nullable=False), sa.Column("user_id", sa.String(64)), sa.Column("message_id", sa.String(64)), sa.Column("rating", sa.Integer, nullable=False), sa.Column("comment", sa.Text), sa.Column("created_at", sa.DateTime, nullable=False), sa.Index( "uq_feedback_thread_run_user", "thread_id", "run_id", "user_id", unique=True, ), ) metadata.create_all(engine) migration = importlib.import_module( "deerflow.persistence.migrations.versions." "20260916_02_answer_feedback_scoring" ) with engine.begin() as connection: connection.execute( runs.insert().values(run_id="run-1", assistant_id="agent-a") ) connection.execute( feedback.insert().values( feedback_id="feedback-1", run_id="run-1", thread_id="thread-1", user_id="user-1", message_id="message-1", rating=-1, comment="bad", created_at=datetime(2026, 9, 16, 12, 0, 0), ) ) connection.execute( feedback.insert(), [ { "feedback_id": "anonymous-1", "run_id": "run-1", "thread_id": "thread-1", "user_id": None, "message_id": "message-1", "rating": -1, "comment": "legacy anonymous", "created_at": datetime(2026, 9, 16, 12, 1, 0), }, { "feedback_id": "anonymous-2", "run_id": "run-1", "thread_id": "thread-1", "user_id": None, "message_id": "message-1", "rating": 1, "comment": None, "created_at": datetime(2026, 9, 16, 12, 2, 0), }, ], ) operations = Operations(MigrationContext.configure(connection)) monkeypatch.setattr(migration, "op", operations) migration.upgrade() inspector = sa.inspect(connection) columns = {column["name"] for column in inspector.get_columns("feedback")} indexes = {index["name"] for index in inspector.get_indexes("feedback")} assert {"agent_id", "target_key", "reasons", "updated_at"} <= columns assert "uq_feedback_target_key" in indexes assert "uq_feedback_thread_run_user" not in indexes upgraded = connection.execute( sa.text( "SELECT agent_id, target_key, reasons, updated_at " "FROM feedback WHERE feedback_id = 'feedback-1'" ) ).mappings().one() assert upgraded["agent_id"] == "agent-a" assert len(upgraded["target_key"]) == 64 assert upgraded["reasons"] == "[]" assert upgraded["updated_at"] is not None target_keys = connection.execute( sa.text("SELECT target_key FROM feedback") ).scalars().all() assert len(target_keys) == 3 assert len(set(target_keys)) == 3 migration.downgrade() downgraded_columns = { column["name"] for column in sa.inspect(connection).get_columns("feedback") } assert "agent_id" not in downgraded_columns assert "target_key" not in downgraded_columns engine.dispose()