From 86e4baccf8eea4c467444bb92e12929e44b7d70d Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Thu, 23 Jul 2026 15:30:40 +0800 Subject: [PATCH] feat(backend): add feedback SQL adapter with tags column and migration SqlFeedbackRepository implements the domain port (queries migrated unchanged from the legacy repository); RunStoreRunLookup adapts the run store for ownership checks. FeedbackRow gains a JSON tags column with an idempotent alembic migration. Legacy repository untouched in this step. --- backend/app/infra/persistence/__init__.py | 0 backend/app/infra/persistence/feedback.py | 154 ++++++++++++++++++ .../deerflow/persistence/feedback/model.py | 6 +- .../migrations/versions/0007_feedback_tags.py | 33 ++++ 4 files changed, 192 insertions(+), 1 deletion(-) create mode 100644 backend/app/infra/persistence/__init__.py create mode 100644 backend/app/infra/persistence/feedback.py create mode 100644 backend/packages/harness/deerflow/persistence/migrations/versions/0007_feedback_tags.py diff --git a/backend/app/infra/persistence/__init__.py b/backend/app/infra/persistence/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/infra/persistence/feedback.py b/backend/app/infra/persistence/feedback.py new file mode 100644 index 000000000..fd80f664a --- /dev/null +++ b/backend/app/infra/persistence/feedback.py @@ -0,0 +1,154 @@ +"""SQL adapters for the feedback bounded context. + +Secondary adapters implementing the ports declared in +``deerflow.domain.feedback.ports``. SQL/ORM vocabulary stops at this +file: methods exchange domain objects, translate ``IntegrityError`` into +domain errors, and normalize SQLite's tz-naive reads. Queries were +migrated unchanged from the legacy repository (now removed) to preserve +behavior. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.domain.feedback.model import DuplicateFeedbackError, Feedback +from deerflow.domain.feedback.ports import FeedbackRepository, RunLookup + +# Transitional: the ORM row stays in the harness until PR-N moves engine, +# models, and migrations into app/infra. +from deerflow.persistence.feedback.model import FeedbackRow + + +def _tz_aware(value: datetime) -> datetime: + """SQLite drops tzinfo on read; stored values are always UTC.""" + return value if value.tzinfo is not None else value.replace(tzinfo=UTC) + + +class SqlFeedbackRepository(FeedbackRepository): + """SQL implementation of the ``FeedbackRepository`` port. + + Explicit inheritance is a readability aid only: a missing method would + still instantiate fine (Protocol bodies are inherited), so the contract + test suite must cover every port method. + """ + + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._session_factory = session_factory + + @staticmethod + def _to_domain(row: FeedbackRow) -> Feedback: + """ORM row -> aggregate. The only place reads are normalized.""" + return Feedback( + feedback_id=row.feedback_id, + run_id=row.run_id, + thread_id=row.thread_id, + rating=row.rating, + user_id=row.user_id, + message_id=row.message_id, + comment=row.comment, + tags=tuple(row.tags or ()), + created_at=_tz_aware(row.created_at), + ) + + @staticmethod + def _to_row(feedback: Feedback) -> FeedbackRow: + """Aggregate -> ORM row. Explicit field list: new columns stay + private until deliberately mapped here.""" + return FeedbackRow( + feedback_id=feedback.feedback_id, + run_id=feedback.run_id, + thread_id=feedback.thread_id, + user_id=feedback.user_id, + message_id=feedback.message_id, + rating=feedback.rating, + comment=feedback.comment, + tags=list(feedback.tags) or None, + created_at=feedback.created_at, + ) + + async def save(self, feedback: Feedback) -> Feedback: + # Migrated from legacy ``upsert``: look up by aggregate identity, + # then update in place or insert. + async with self._session_factory() as session: + stmt = select(FeedbackRow).where( + FeedbackRow.thread_id == feedback.thread_id, + FeedbackRow.run_id == feedback.run_id, + FeedbackRow.user_id == feedback.user_id, + ) + row = (await session.execute(stmt)).scalar_one_or_none() + if row is not None: + row.rating = feedback.rating + row.comment = feedback.comment + row.tags = list(feedback.tags) or None + row.created_at = feedback.created_at + else: + row = self._to_row(feedback) + session.add(row) + try: + await session.commit() + except IntegrityError as exc: + # Two concurrent upserts can both miss the lookup and insert; + # the loser hits the unique constraint. Translate instead of + # leaking the driver exception (legacy code returned a 500). + raise DuplicateFeedbackError(f"concurrent feedback upsert for run {feedback.run_id}") from exc + await session.refresh(row) + return self._to_domain(row) + + async def latest_per_run_in_thread(self, thread_id: str, *, user_id: str | None) -> dict[str, Feedback]: + # Migrated from legacy ``list_by_thread_grouped``. With an ownership + # filter the unique constraint guarantees at most one row per run. + stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id) + if user_id is not None: + stmt = stmt.where(FeedbackRow.user_id == user_id) + async with self._session_factory() as session: + result = await session.execute(stmt) + return {row.run_id: self._to_domain(row) for row in result.scalars()} + + async def latest_for_runs(self, thread_id: str, run_ids: set[str], *, user_id: str | None) -> dict[str, Feedback]: + # Migrated from legacy ``list_by_run_ids`` (paged message list). + if not run_ids: + return {} + stmt = select(FeedbackRow).where( + FeedbackRow.thread_id == thread_id, + FeedbackRow.run_id.in_(run_ids), + ) + if user_id is not None: + stmt = stmt.where(FeedbackRow.user_id == user_id) + async with self._session_factory() as session: + result = await session.execute(stmt) + return {row.run_id: self._to_domain(row) for row in result.scalars()} + + async def remove_for_run(self, thread_id: str, run_id: str, *, user_id: str | None) -> bool: + # Migrated from legacy ``delete_by_run``. + async with self._session_factory() as session: + stmt = select(FeedbackRow).where( + FeedbackRow.thread_id == thread_id, + FeedbackRow.run_id == run_id, + FeedbackRow.user_id == user_id, + ) + row = (await session.execute(stmt)).scalar_one_or_none() + if row is None: + return False + await session.delete(row) + await session.commit() + return True + + +class RunStoreRunLookup(RunLookup): + """Adapts the framework ``RunStore`` (wide interface) to the narrow + ``RunLookup`` port -- reuses the existing lookup, no new SQL. + + Used by the service for run-ownership checks before writing feedback. + """ + + def __init__(self, run_store) -> None: + self._run_store = run_store + + async def thread_of(self, run_id: str) -> str | None: + run = await self._run_store.get(run_id) + return run.get("thread_id") if run else None diff --git a/backend/packages/harness/deerflow/persistence/feedback/model.py b/backend/packages/harness/deerflow/persistence/feedback/model.py index a9b6479b3..a7968fbf9 100644 --- a/backend/packages/harness/deerflow/persistence/feedback/model.py +++ b/backend/packages/harness/deerflow/persistence/feedback/model.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import UTC, datetime -from sqlalchemy import DateTime, String, Text, UniqueConstraint +from sqlalchemy import JSON, DateTime, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column from deerflow.persistence.base import Base @@ -29,4 +29,8 @@ class FeedbackRow(Base): comment: Mapped[str | None] = mapped_column(Text) # Optional text feedback from the user + tags: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) + # Optional thumbs-down reason slugs from the feedback dialog + # (language-neutral, e.g. ["incorrect", "slow"]) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC)) diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0007_feedback_tags.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0007_feedback_tags.py new file mode 100644 index 000000000..cc3c861d9 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0007_feedback_tags.py @@ -0,0 +1,33 @@ +"""feedback tags. + +Revision ID: 0007_feedback_tags +Revises: 0006_agents +Create Date: 2026-07-23 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0007_feedback_tags" +down_revision: str | Sequence[str] | None = "0006_agents" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + columns = {col["name"] for col in inspector.get_columns("feedback")} + if "tags" in columns: + # Idempotent: a DB whose full-metadata create_all already provisioned + # the column must not have it re-added here. + return + op.add_column("feedback", sa.Column("tags", sa.JSON(), nullable=True)) + + +def downgrade() -> None: + op.drop_column("feedback", "tags")