diff --git a/backend/packages/harness/deerflow/persistence/feedback/__init__.py b/backend/packages/harness/deerflow/persistence/feedback/__init__.py
index ee958b027..6a1ca9f3f 100644
--- a/backend/packages/harness/deerflow/persistence/feedback/__init__.py
+++ b/backend/packages/harness/deerflow/persistence/feedback/__init__.py
@@ -1,6 +1,11 @@
-"""Feedback persistence — ORM and SQL repository."""
+"""Feedback persistence — ORM row only.
+
+The legacy SQL repository was replaced by the hexagonal slice:
+port `deerflow.domain.feedback.ports.FeedbackRepository`, adapter
+`app.infra.persistence.feedback.SqlFeedbackRepository`. The ORM row stays
+here until PR-N moves engine/models/migrations into app/infra.
+"""
 
 from deerflow.persistence.feedback.model import FeedbackRow
-from deerflow.persistence.feedback.sql import FeedbackRepository
 
-__all__ = ["FeedbackRepository", "FeedbackRow"]
+__all__ = ["FeedbackRow"]
diff --git a/backend/packages/harness/deerflow/persistence/feedback/sql.py b/backend/packages/harness/deerflow/persistence/feedback/sql.py
deleted file mode 100644
index 5cd03d0e9..000000000
--- a/backend/packages/harness/deerflow/persistence/feedback/sql.py
+++ /dev/null
@@ -1,240 +0,0 @@
-"""SQLAlchemy-backed feedback storage.
-
-Each method acquires its own short-lived session.
-"""
-
-from __future__ import annotations
-
-import uuid
-from datetime import UTC, datetime
-
-from sqlalchemy import case, func, select
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
-
-from deerflow.persistence.feedback.model import FeedbackRow
-from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
-from deerflow.utils.time import coerce_iso
-
-
-class FeedbackRepository:
-    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
-        self._sf = session_factory
-
-    @staticmethod
-    def _row_to_dict(row: FeedbackRow) -> dict:
-        d = row.to_dict()
-        val = d.get("created_at")
-        if isinstance(val, datetime):
-            # SQLite drops tzinfo on read; normalize via ``coerce_iso`` so output is always tz-aware.
-            d["created_at"] = coerce_iso(val)
-        return d
-
-    async def create(
-        self,
-        *,
-        run_id: str,
-        thread_id: str,
-        rating: int,
-        user_id: str | None | _AutoSentinel = AUTO,
-        message_id: str | None = None,
-        comment: str | None = None,
-    ) -> dict:
-        """Create a feedback record. rating must be +1 or -1."""
-        if rating not in (1, -1):
-            raise ValueError(f"rating must be +1 or -1, got {rating}")
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.create")
-        row = FeedbackRow(
-            feedback_id=str(uuid.uuid4()),
-            run_id=run_id,
-            thread_id=thread_id,
-            user_id=resolved_user_id,
-            message_id=message_id,
-            rating=rating,
-            comment=comment,
-            created_at=datetime.now(UTC),
-        )
-        async with self._sf() as session:
-            session.add(row)
-            await session.commit()
-            await session.refresh(row)
-            return self._row_to_dict(row)
-
-    async def get(
-        self,
-        feedback_id: str,
-        *,
-        user_id: str | None | _AutoSentinel = AUTO,
-    ) -> dict | None:
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.get")
-        async with self._sf() as session:
-            row = await session.get(FeedbackRow, feedback_id)
-            if row is None:
-                return None
-            if resolved_user_id is not None and row.user_id != resolved_user_id:
-                return None
-            return self._row_to_dict(row)
-
-    async def list_by_run(
-        self,
-        thread_id: str,
-        run_id: str,
-        *,
-        limit: int = 100,
-        user_id: str | None | _AutoSentinel = AUTO,
-    ) -> list[dict]:
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_run")
-        stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id, FeedbackRow.run_id == run_id)
-        if resolved_user_id is not None:
-            stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
-        stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            return [self._row_to_dict(r) for r in result.scalars()]
-
-    async def list_by_thread(
-        self,
-        thread_id: str,
-        *,
-        limit: int = 100,
-        user_id: str | None | _AutoSentinel = AUTO,
-    ) -> list[dict]:
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_thread")
-        stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id)
-        if resolved_user_id is not None:
-            stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
-        stmt = stmt.order_by(FeedbackRow.created_at.asc()).limit(limit)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            return [self._row_to_dict(r) for r in result.scalars()]
-
-    async def delete(
-        self,
-        feedback_id: str,
-        *,
-        user_id: str | None | _AutoSentinel = AUTO,
-    ) -> bool:
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.delete")
-        async with self._sf() as session:
-            row = await session.get(FeedbackRow, feedback_id)
-            if row is None:
-                return False
-            if resolved_user_id is not None and row.user_id != resolved_user_id:
-                return False
-            await session.delete(row)
-            await session.commit()
-            return True
-
-    async def upsert(
-        self,
-        *,
-        run_id: str,
-        thread_id: str,
-        rating: int,
-        user_id: str | None | _AutoSentinel = AUTO,
-        comment: str | None = None,
-    ) -> dict:
-        """Create or update feedback for (thread_id, run_id, user_id). rating must be +1 or -1."""
-        if rating not in (1, -1):
-            raise ValueError(f"rating must be +1 or -1, got {rating}")
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.upsert")
-        async with self._sf() as session:
-            stmt = select(FeedbackRow).where(
-                FeedbackRow.thread_id == thread_id,
-                FeedbackRow.run_id == run_id,
-                FeedbackRow.user_id == resolved_user_id,
-            )
-            result = await session.execute(stmt)
-            row = result.scalar_one_or_none()
-            if row is not None:
-                row.rating = rating
-                row.comment = comment
-                row.created_at = datetime.now(UTC)
-            else:
-                row = FeedbackRow(
-                    feedback_id=str(uuid.uuid4()),
-                    run_id=run_id,
-                    thread_id=thread_id,
-                    user_id=resolved_user_id,
-                    rating=rating,
-                    comment=comment,
-                    created_at=datetime.now(UTC),
-                )
-                session.add(row)
-            await session.commit()
-            await session.refresh(row)
-            return self._row_to_dict(row)
-
-    async def delete_by_run(
-        self,
-        *,
-        thread_id: str,
-        run_id: str,
-        user_id: str | None | _AutoSentinel = AUTO,
-    ) -> bool:
-        """Delete the current user's feedback for a run. Returns True if a record was deleted."""
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.delete_by_run")
-        async with self._sf() as session:
-            stmt = select(FeedbackRow).where(
-                FeedbackRow.thread_id == thread_id,
-                FeedbackRow.run_id == run_id,
-                FeedbackRow.user_id == resolved_user_id,
-            )
-            result = await session.execute(stmt)
-            row = result.scalar_one_or_none()
-            if row is None:
-                return False
-            await session.delete(row)
-            await session.commit()
-            return True
-
-    async def list_by_thread_grouped(
-        self,
-        thread_id: str,
-        *,
-        user_id: str | None | _AutoSentinel = AUTO,
-    ) -> dict[str, dict]:
-        """Return feedback grouped by run_id for a thread: {run_id: feedback_dict}."""
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_thread_grouped")
-        stmt = select(FeedbackRow).where(FeedbackRow.thread_id == thread_id)
-        if resolved_user_id is not None:
-            stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            return {row.run_id: self._row_to_dict(row) for row in result.scalars()}
-
-    async def list_by_run_ids(
-        self,
-        thread_id: str,
-        run_ids: set[str],
-        *,
-        user_id: str | None | _AutoSentinel = AUTO,
-    ) -> dict[str, dict]:
-        """Return feedback for only the selected runs in one thread."""
-        if not run_ids:
-            return {}
-        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.list_by_run_ids")
-        stmt = select(FeedbackRow).where(
-            FeedbackRow.thread_id == thread_id,
-            FeedbackRow.run_id.in_(run_ids),
-        )
-        if resolved_user_id is not None:
-            stmt = stmt.where(FeedbackRow.user_id == resolved_user_id)
-        async with self._sf() as session:
-            result = await session.execute(stmt)
-            return {row.run_id: self._row_to_dict(row) for row in result.scalars()}
-
-    async def aggregate_by_run(self, thread_id: str, run_id: str) -> dict:
-        """Aggregate feedback stats for a run using database-side counting."""
-        stmt = select(
-            func.count().label("total"),
-            func.coalesce(func.sum(case((FeedbackRow.rating == 1, 1), else_=0)), 0).label("positive"),
-            func.coalesce(func.sum(case((FeedbackRow.rating == -1, 1), else_=0)), 0).label("negative"),
-        ).where(FeedbackRow.thread_id == thread_id, FeedbackRow.run_id == run_id)
-        async with self._sf() as session:
-            row = (await session.execute(stmt)).one()
-            return {
-                "run_id": run_id,
-                "total": row.total,
-                "positive": row.positive,
-                "negative": row.negative,
-            }
diff --git a/backend/tests/test_feedback.py b/backend/tests/test_feedback.py
index d9a34fbf1..2c952228a 100644
--- a/backend/tests/test_feedback.py
+++ b/backend/tests/test_feedback.py
@@ -1,258 +1,154 @@
-"""Tests for FeedbackRepository and follow-up association.
+"""Contract tests for the FeedbackRepository port.
 
-Uses temp SQLite DB for ORM tests.
+Every implementation (SQL adapter, in-memory fake) runs the same suite, so
+port semantics and implementations cannot drift apart. The in-memory fake
+also serves service-level tests as a fast zero-IO double.
 """
 
+from dataclasses import replace
+
 import pytest
 
-from deerflow.persistence.feedback import FeedbackRepository
+from deerflow.domain.feedback import Feedback
+from deerflow.domain.feedback.ports import FeedbackRepository
 
 
-async def _make_feedback_repo(tmp_path):
-    from deerflow.persistence.engine import get_session_factory, init_engine
+class InMemoryFeedbackRepository:
+    """Zero-IO fake keyed by aggregate identity (thread, run, user)."""
 
-    url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}"
-    await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
-    return FeedbackRepository(get_session_factory())
+    def __init__(self) -> None:
+        self._rows: dict[tuple[str, str, str | None], Feedback] = {}
+
+    async def save(self, feedback: Feedback) -> Feedback:
+        key = (feedback.thread_id, feedback.run_id, feedback.user_id)
+        existing = self._rows.get(key)
+        if existing is not None:
+            feedback = replace(feedback, feedback_id=existing.feedback_id)
+        self._rows[key] = feedback
+        return feedback
+
+    async def latest_per_run_in_thread(self, thread_id: str, *, user_id: str | None) -> dict[str, Feedback]:
+        return {fb.run_id: fb for (thread, _run, user), fb in self._rows.items() if thread == thread_id and (user_id is None or user == user_id)}
+
+    async def latest_for_runs(self, thread_id: str, run_ids: set[str], *, user_id: str | None) -> dict[str, Feedback]:
+        if not run_ids:
+            return {}
+        per_run = await self.latest_per_run_in_thread(thread_id, user_id=user_id)
+        return {run_id: fb for run_id, fb in per_run.items() if run_id in run_ids}
+
+    async def remove_for_run(self, thread_id: str, run_id: str, *, user_id: str | None) -> bool:
+        return self._rows.pop((thread_id, run_id, user_id), None) is not None
 
 
-async def _cleanup():
-    from deerflow.persistence.engine import close_engine
-
-    await close_engine()
-
-
-# -- FeedbackRepository --
-
-
-class TestFeedbackRepository:
-    @pytest.mark.anyio
-    async def test_create_positive(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        record = await repo.create(run_id="r1", thread_id="t1", rating=1)
-        assert record["feedback_id"]
-        assert record["rating"] == 1
-        assert record["run_id"] == "r1"
-        assert record["thread_id"] == "t1"
-        assert "created_at" in record
-        await _cleanup()
+class FeedbackRepositoryContract:
+    """Shared port-semantics suite; subclasses provide a ``repo`` fixture."""
 
     @pytest.mark.anyio
-    async def test_create_negative_with_comment(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        record = await repo.create(
-            run_id="r1",
-            thread_id="t1",
-            rating=-1,
-            comment="Response was inaccurate",
+    async def test_satisfies_port(self, repo):
+        assert isinstance(repo, FeedbackRepository)
+
+    @pytest.mark.anyio
+    async def test_save_creates(self, repo):
+        fb = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
+        assert fb.feedback_id
+        assert fb.rating == 1
+        assert fb.created_at.tzinfo is not None
+
+    @pytest.mark.anyio
+    async def test_save_updates_keeping_identity(self, repo):
+        first = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
+        second = await repo.save(
+            Feedback.create(
+                run_id="r1",
+                thread_id="t1",
+                rating=-1,
+                user_id="u1",
+                comment="changed my mind",
+                tags=["incorrect", "slow"],
+            )
         )
-        assert record["rating"] == -1
-        assert record["comment"] == "Response was inaccurate"
-        await _cleanup()
+        assert second.feedback_id == first.feedback_id
+        assert second.rating == -1
+        assert second.comment == "changed my mind"
+        assert second.tags == ("incorrect", "slow")
 
     @pytest.mark.anyio
-    async def test_create_with_message_id(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        record = await repo.create(run_id="r1", thread_id="t1", rating=1, message_id="msg-42")
-        assert record["message_id"] == "msg-42"
-        await _cleanup()
+    async def test_save_different_users_separate(self, repo):
+        a = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
+        b = await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=-1, user_id="u2"))
+        assert a.feedback_id != b.feedback_id
 
     @pytest.mark.anyio
-    async def test_create_with_owner(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        record = await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-1")
-        assert record["user_id"] == "user-1"
-        await _cleanup()
+    async def test_latest_per_run_in_thread_scopes_thread_and_owner(self, repo):
+        await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
+        await repo.save(Feedback.create(run_id="r2", thread_id="t1", rating=-1, user_id="u1"))
+        await repo.save(Feedback.create(run_id="r3", thread_id="t2", rating=1, user_id="u1"))
+        await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=-1, user_id="u2"))
 
-    @pytest.mark.anyio
-    async def test_create_invalid_rating_zero(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        with pytest.raises(ValueError):
-            await repo.create(run_id="r1", thread_id="t1", rating=0)
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_create_invalid_rating_five(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        with pytest.raises(ValueError):
-            await repo.create(run_id="r1", thread_id="t1", rating=5)
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_get(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        created = await repo.create(run_id="r1", thread_id="t1", rating=1)
-        fetched = await repo.get(created["feedback_id"])
-        assert fetched is not None
-        assert fetched["feedback_id"] == created["feedback_id"]
-        assert fetched["rating"] == 1
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_get_nonexistent(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        assert await repo.get("nonexistent") is None
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_list_by_run(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-1")
-        await repo.create(run_id="r1", thread_id="t1", rating=-1, user_id="user-2")
-        await repo.create(run_id="r2", thread_id="t1", rating=1, user_id="user-1")
-        results = await repo.list_by_run("t1", "r1", user_id=None)
-        assert len(results) == 2
-        assert all(r["run_id"] == "r1" for r in results)
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_list_by_thread(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        await repo.create(run_id="r1", thread_id="t1", rating=1)
-        await repo.create(run_id="r2", thread_id="t1", rating=-1)
-        await repo.create(run_id="r3", thread_id="t2", rating=1)
-        results = await repo.list_by_thread("t1")
-        assert len(results) == 2
-        assert all(r["thread_id"] == "t1" for r in results)
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_delete(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        created = await repo.create(run_id="r1", thread_id="t1", rating=1)
-        deleted = await repo.delete(created["feedback_id"])
-        assert deleted is True
-        assert await repo.get(created["feedback_id"]) is None
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_delete_nonexistent(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        deleted = await repo.delete("nonexistent")
-        assert deleted is False
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_aggregate_by_run(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-1")
-        await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="user-2")
-        await repo.create(run_id="r1", thread_id="t1", rating=-1, user_id="user-3")
-        stats = await repo.aggregate_by_run("t1", "r1")
-        assert stats["total"] == 3
-        assert stats["positive"] == 2
-        assert stats["negative"] == 1
-        assert stats["run_id"] == "r1"
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_aggregate_empty(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        stats = await repo.aggregate_by_run("t1", "r1")
-        assert stats["total"] == 0
-        assert stats["positive"] == 0
-        assert stats["negative"] == 0
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_upsert_creates_new(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        record = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
-        assert record["rating"] == 1
-        assert record["feedback_id"]
-        assert record["user_id"] == "u1"
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_upsert_updates_existing(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        first = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
-        second = await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u1", comment="changed my mind")
-        assert second["feedback_id"] == first["feedback_id"]
-        assert second["rating"] == -1
-        assert second["comment"] == "changed my mind"
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_upsert_different_users_separate(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        r1 = await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
-        r2 = await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u2")
-        assert r1["feedback_id"] != r2["feedback_id"]
-        assert r1["rating"] == 1
-        assert r2["rating"] == -1
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_upsert_invalid_rating(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        with pytest.raises(ValueError):
-            await repo.upsert(run_id="r1", thread_id="t1", rating=0, user_id="u1")
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_delete_by_run(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
-        deleted = await repo.delete_by_run(thread_id="t1", run_id="r1", user_id="u1")
-        assert deleted is True
-        results = await repo.list_by_run("t1", "r1", user_id="u1")
-        assert len(results) == 0
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_delete_by_run_nonexistent(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        deleted = await repo.delete_by_run(thread_id="t1", run_id="r1", user_id="u1")
-        assert deleted is False
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_list_by_thread_grouped(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
-        await repo.upsert(run_id="r2", thread_id="t1", rating=-1, user_id="u1")
-        await repo.upsert(run_id="r3", thread_id="t2", rating=1, user_id="u1")
-        grouped = await repo.list_by_thread_grouped("t1", user_id="u1")
-        assert "r1" in grouped
-        assert "r2" in grouped
-        assert "r3" not in grouped
-        assert grouped["r1"]["rating"] == 1
-        assert grouped["r2"]["rating"] == -1
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_list_by_thread_grouped_empty(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        grouped = await repo.list_by_thread_grouped("t1", user_id="u1")
-        assert grouped == {}
-        await _cleanup()
-
-    @pytest.mark.anyio
-    async def test_list_by_run_ids_is_thread_and_owner_scoped(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
-        await repo.upsert(run_id="r1", thread_id="t1", rating=1, user_id="u1")
-        await repo.upsert(run_id="r2", thread_id="t1", rating=-1, user_id="u1")
-        await repo.upsert(run_id="r3", thread_id="t1", rating=1, user_id="u1")
-        await repo.upsert(run_id="r1", thread_id="t1", rating=-1, user_id="u2")
-        await repo.upsert(run_id="r2", thread_id="t2", rating=1, user_id="u1")
-
-        grouped = await repo.list_by_run_ids("t1", {"r1", "r2"}, user_id="u1")
+        grouped = await repo.latest_per_run_in_thread("t1", user_id="u1")
 
         assert set(grouped) == {"r1", "r2"}
-        assert grouped["r1"]["rating"] == 1
-        assert grouped["r2"]["rating"] == -1
-        await _cleanup()
+        assert grouped["r1"].rating == 1
+        assert grouped["r2"].rating == -1
 
     @pytest.mark.anyio
-    async def test_list_by_run_ids_empty_skips_query(self, tmp_path):
-        repo = await _make_feedback_repo(tmp_path)
+    async def test_latest_per_run_in_thread_empty(self, repo):
+        assert await repo.latest_per_run_in_thread("t1", user_id="u1") == {}
 
-        assert await repo.list_by_run_ids("t1", set(), user_id="u1") == {}
-        await _cleanup()
+    @pytest.mark.anyio
+    async def test_latest_for_runs_scopes_selection(self, repo):
+        await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
+        await repo.save(Feedback.create(run_id="r2", thread_id="t1", rating=-1, user_id="u1"))
+        await repo.save(Feedback.create(run_id="r3", thread_id="t1", rating=1, user_id="u1"))
+
+        grouped = await repo.latest_for_runs("t1", {"r1", "r2"}, user_id="u1")
+
+        assert set(grouped) == {"r1", "r2"}
+
+    @pytest.mark.anyio
+    async def test_latest_for_runs_empty_set(self, repo):
+        assert await repo.latest_for_runs("t1", set(), user_id="u1") == {}
+
+    @pytest.mark.anyio
+    async def test_remove_for_run(self, repo):
+        await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
+        assert await repo.remove_for_run("t1", "r1", user_id="u1") is True
+        assert await repo.latest_per_run_in_thread("t1", user_id="u1") == {}
+
+    @pytest.mark.anyio
+    async def test_remove_for_run_nonexistent(self, repo):
+        assert await repo.remove_for_run("t1", "r1", user_id="u1") is False
+
+    @pytest.mark.anyio
+    async def test_remove_for_run_other_user_denied(self, repo):
+        await repo.save(Feedback.create(run_id="r1", thread_id="t1", rating=1, user_id="u1"))
+        assert await repo.remove_for_run("t1", "r1", user_id="u2") is False
+        assert set(await repo.latest_per_run_in_thread("t1", user_id="u1")) == {"r1"}
 
 
-# -- Follow-up association --
+class TestSqlFeedbackRepository(FeedbackRepositoryContract):
+    @pytest.fixture
+    async def repo(self, tmp_path):
+        from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+        import deerflow.persistence.feedback.model  # noqa: F401  (register FeedbackRow)
+        from app.infra.persistence.feedback import SqlFeedbackRepository
+        from deerflow.persistence.base import Base
+
+        engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'test.db'}")
+        async with engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        yield SqlFeedbackRepository(async_sessionmaker(engine, expire_on_commit=False))
+        await engine.dispose()
+
+
+class TestInMemoryFeedbackRepository(FeedbackRepositoryContract):
+    @pytest.fixture
+    def repo(self):
+        return InMemoryFeedbackRepository()
+
+
+# -- Follow-up association (unrelated to the feedback repository) --
 
 
 class TestFollowUpAssociation:
diff --git a/backend/tests/test_feedback_service.py b/backend/tests/test_feedback_service.py
new file mode 100644
index 000000000..3aab6ec63
--- /dev/null
+++ b/backend/tests/test_feedback_service.py
@@ -0,0 +1,96 @@
+"""FeedbackService tests with injected fakes — zero IO, no engine.
+
+The seam the ports create is what makes these tests possible: the service
+is exercised end-to-end against dict-backed doubles.
+"""
+
+import pytest
+from test_feedback import InMemoryFeedbackRepository
+
+from deerflow.domain.feedback import FeedbackService, InvalidRatingError, RunNotFoundError
+
+
+class FakeRunLookup:
+    """RunLookup double backed by a run_id -> thread_id mapping."""
+
+    def __init__(self, runs: dict[str, str]):
+        self._runs = runs
+
+    async def thread_of(self, run_id: str) -> str | None:
+        return self._runs.get(run_id)
+
+
+def _service(runs: dict[str, str] | None = None) -> FeedbackService:
+    return FeedbackService(
+        repository=InMemoryFeedbackRepository(),
+        runs=FakeRunLookup(runs if runs is not None else {"r1": "t1"}),
+    )
+
+
+class TestRateRun:
+    @pytest.mark.anyio
+    async def test_stores_rating(self):
+        svc = _service()
+        fb = await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
+        assert fb.rating == 1
+        assert fb.user_id == "u1"
+
+    @pytest.mark.anyio
+    async def test_idempotent_upsert_keeps_identity(self):
+        svc = _service()
+        first = await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
+        second = await svc.rate_run("t1", "r1", rating=-1, comment="meh", user_id="u1")
+        assert second.feedback_id == first.feedback_id
+        assert second.rating == -1
+
+    @pytest.mark.anyio
+    async def test_enrich_with_tags(self):
+        svc = _service()
+        await svc.rate_run("t1", "r1", rating=-1, comment=None, user_id="u1")
+        fb = await svc.rate_run("t1", "r1", rating=-1, comment="wrong", user_id="u1", tags=["incorrect"])
+        assert fb.tags == ("incorrect",)
+
+    @pytest.mark.anyio
+    async def test_unknown_run_rejected(self):
+        svc = _service({})
+        with pytest.raises(RunNotFoundError):
+            await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
+
+    @pytest.mark.anyio
+    async def test_cross_thread_run_rejected(self):
+        """A run id belonging to another thread must not be ratable through
+        this thread's URL (ownership check)."""
+        svc = _service({"r1": "other-thread"})
+        with pytest.raises(RunNotFoundError):
+            await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
+
+    @pytest.mark.anyio
+    async def test_invalid_rating_propagates_before_persistence(self):
+        svc = _service()
+        with pytest.raises(InvalidRatingError):
+            await svc.rate_run("t1", "r1", rating=0, comment=None, user_id="u1")
+
+
+class TestRetractAndReads:
+    @pytest.mark.anyio
+    async def test_retract(self):
+        svc = _service()
+        await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
+        assert await svc.retract_run_rating("t1", "r1", user_id="u1") is True
+        assert await svc.retract_run_rating("t1", "r1", user_id="u1") is False
+
+    @pytest.mark.anyio
+    async def test_latest_per_run_in_thread(self):
+        svc = _service({"r1": "t1", "r2": "t1"})
+        await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
+        await svc.rate_run("t1", "r2", rating=-1, comment=None, user_id="u1")
+        grouped = await svc.latest_per_run_in_thread("t1", user_id="u1")
+        assert {run_id: fb.rating for run_id, fb in grouped.items()} == {"r1": 1, "r2": -1}
+
+    @pytest.mark.anyio
+    async def test_latest_for_runs(self):
+        svc = _service({"r1": "t1", "r2": "t1"})
+        await svc.rate_run("t1", "r1", rating=1, comment=None, user_id="u1")
+        await svc.rate_run("t1", "r2", rating=-1, comment=None, user_id="u1")
+        grouped = await svc.latest_for_runs("t1", {"r2"}, user_id="u1")
+        assert set(grouped) == {"r2"}
diff --git a/backend/tests/test_owner_isolation.py b/backend/tests/test_owner_isolation.py
index ac190bbdf..18ca80367 100644
--- a/backend/tests/test_owner_isolation.py
+++ b/backend/tests/test_owner_isolation.py
@@ -334,53 +334,32 @@ async def test_run_events_cross_user_delete_denied(tmp_path):
 @pytest.mark.anyio
 @pytest.mark.no_auto_user
 async def test_feedback_cross_user_isolation(tmp_path):
+    """Ownership is an explicit user_id filter on the new feedback port —
+    the AUTO-sentinel context no longer applies (hexagonal slice)."""
+    from app.infra.persistence.feedback import SqlFeedbackRepository
+    from deerflow.domain.feedback import Feedback
     from deerflow.persistence.engine import get_session_factory
-    from deerflow.persistence.feedback import FeedbackRepository
 
     cleanup = await _make_engines(tmp_path)
     try:
-        repo = FeedbackRepository(get_session_factory())
+        repo = SqlFeedbackRepository(get_session_factory())
 
-        # User A submits positive feedback.
-        with _as_user(USER_A):
-            a_feedback = await repo.create(
-                run_id="run-a1",
-                thread_id="t-alpha",
-                rating=1,
-                comment="A liked this",
-            )
-
-        # User B submits negative feedback.
-        with _as_user(USER_B):
-            b_feedback = await repo.create(
-                run_id="run-b1",
-                thread_id="t-beta",
-                rating=-1,
-                comment="B disliked this",
-            )
+        await repo.save(Feedback.create(run_id="run-a1", thread_id="t-alpha", rating=1, user_id=USER_A.id, comment="A liked this"))
+        await repo.save(Feedback.create(run_id="run-b1", thread_id="t-beta", rating=-1, user_id=USER_B.id, comment="B disliked this"))
 
         # User A must see only A's feedback.
-        with _as_user(USER_A):
-            retrieved = await repo.get(a_feedback["feedback_id"])
-            assert retrieved is not None
-            assert retrieved["comment"] == "A liked this"
+        a_view = await repo.latest_per_run_in_thread("t-alpha", user_id=USER_A.id)
+        assert set(a_view) == {"run-a1"}
+        assert a_view["run-a1"].comment == "A liked this"
 
-            # CRITICAL: cannot read B's feedback by id.
-            leaked = await repo.get(b_feedback["feedback_id"])
-            assert leaked is None, "User A leaked User B's feedback"
-
-            # list_by_run for B's run must be empty.
-            empty = await repo.list_by_run("t-beta", "run-b1")
-            assert empty == []
+        # CRITICAL: A cannot see B's feedback through B's thread.
+        leaked = await repo.latest_per_run_in_thread("t-beta", user_id=USER_A.id)
+        assert leaked == {}, "User A leaked User B's feedback"
 
         # User B must see only B's feedback.
-        with _as_user(USER_B):
-            leaked = await repo.get(a_feedback["feedback_id"])
-            assert leaked is None, "User B leaked User A's feedback"
-
-            b_list = await repo.list_by_run("t-beta", "run-b1")
-            assert len(b_list) == 1
-            assert b_list[0]["comment"] == "B disliked this"
+        assert await repo.latest_per_run_in_thread("t-alpha", user_id=USER_B.id) == {}
+        b_view = await repo.latest_per_run_in_thread("t-beta", user_id=USER_B.id)
+        assert b_view["run-b1"].comment == "B disliked this"
     finally:
         await cleanup()
 
@@ -388,25 +367,21 @@ async def test_feedback_cross_user_isolation(tmp_path):
 @pytest.mark.anyio
 @pytest.mark.no_auto_user
 async def test_feedback_cross_user_delete_denied(tmp_path):
+    from app.infra.persistence.feedback import SqlFeedbackRepository
+    from deerflow.domain.feedback import Feedback
     from deerflow.persistence.engine import get_session_factory
-    from deerflow.persistence.feedback import FeedbackRepository
 
     cleanup = await _make_engines(tmp_path)
     try:
-        repo = FeedbackRepository(get_session_factory())
+        repo = SqlFeedbackRepository(get_session_factory())
+        await repo.save(Feedback.create(run_id="run-a1", thread_id="t-alpha", rating=1, user_id=USER_A.id))
 
-        with _as_user(USER_A):
-            fb = await repo.create(run_id="run-a1", thread_id="t-alpha", rating=1)
+        # User B tries to retract A's feedback — must return False (no-op).
+        deleted = await repo.remove_for_run("t-alpha", "run-a1", user_id=USER_B.id)
+        assert deleted is False, "User B deleted User A's feedback"
 
-        # User B tries to delete A's feedback — must return False (no-op).
-        with _as_user(USER_B):
-            deleted = await repo.delete(fb["feedback_id"])
-            assert deleted is False, "User B deleted User A's feedback"
-
-        # A's feedback still retrievable.
-        with _as_user(USER_A):
-            row = await repo.get(fb["feedback_id"])
-            assert row is not None
+        # A's feedback still present.
+        assert set(await repo.latest_per_run_in_thread("t-alpha", user_id=USER_A.id)) == {"run-a1"}
     finally:
         await cleanup()
 
diff --git a/backend/tests/test_persistence_timezone.py b/backend/tests/test_persistence_timezone.py
index 7cd7b3310..4adb6404a 100644
--- a/backend/tests/test_persistence_timezone.py
+++ b/backend/tests/test_persistence_timezone.py
@@ -76,12 +76,17 @@ async def test_run_repository_emits_tz_aware_timestamps(tmp_path):
 
 @pytest.mark.anyio
 async def test_feedback_repository_emits_tz_aware_timestamps(tmp_path):
-    from deerflow.persistence.feedback import FeedbackRepository
+    from app.infra.persistence.feedback import SqlFeedbackRepository
+    from deerflow.domain.feedback import Feedback
 
-    repo = FeedbackRepository(await _init_sqlite(tmp_path))
+    repo = SqlFeedbackRepository(await _init_sqlite(tmp_path))
     try:
-        record = await repo.create(run_id="r-tz", thread_id="t-tz", rating=1, user_id="u1")
-        _assert_tz_aware(record["created_at"], context="feedback.create.created_at")
+        fb = await repo.save(Feedback.create(run_id="r-tz", thread_id="t-tz", rating=1, user_id="u1"))
+        # The port exchanges domain objects (datetime), not ISO strings;
+        # serialization to ISO happens in the router. Assert tz-awareness here.
+        assert fb.created_at.tzinfo is not None, "feedback.save.created_at lacks tzinfo"
+        read_back = (await repo.latest_per_run_in_thread("t-tz", user_id="u1"))["r-tz"]
+        assert read_back.created_at.tzinfo is not None, "feedback read-back lacks tzinfo"
     finally:
         await _cleanup()