feat(harness): add feedback domain slice (model, ports, service)

Hexagonal inner ring for the feedback bounded context: frozen aggregate
with construct-time invariants (rating, reason-tag slugs), technology-
neutral ports (typing.Protocol), and the application service with
explicit user_id. Guarded by an AST purity test that keeps domain/ free
of infrastructure imports.

backend/packages/harness/deerflow/domain/__init__.py

@@ -0,0 +1,9 @@
+"""Hexagonal inner ring: domain models, ports, and application services.
+
+Every package under this namespace is one bounded context laid out as
+`{model,ports,service}.py`. Code here has zero infrastructure
+dependencies -- no sqlalchemy/fastapi/pydantic, no `app.*`, and no
+harness infrastructure modules (enforced in CI by
+`tests/test_harness_domain_purity.py`). Adapters implementing the ports
+live in `app/infra/`; wiring happens only in `app/gateway/deps.py`.
+"""

backend/packages/harness/deerflow/domain/feedback/__init__.py

@@ -0,0 +1,27 @@
+"""Feedback bounded context: a user's rating of a single run.
+
+Public API of the context. Import domain objects and the service from
+here; import ports (FeedbackRepository, RunLookup) from
+`deerflow.domain.feedback.ports` -- they are contracts consumed by
+adapters and tests, not everyday call-site symbols.
+"""
+
+from deerflow.domain.feedback.model import (
+    DuplicateFeedbackError,
+    Feedback,
+    FeedbackError,
+    InvalidRatingError,
+    InvalidTagError,
+    RunNotFoundError,
+)
+from deerflow.domain.feedback.service import FeedbackService
+
+__all__ = [
+    "DuplicateFeedbackError",
+    "Feedback",
+    "FeedbackError",
+    "FeedbackService",
+    "InvalidRatingError",
+    "InvalidTagError",
+    "RunNotFoundError",
+]

backend/packages/harness/deerflow/domain/feedback/model.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+VALID_RATINGS = (-1, 1)
+
+# Language-neutral reason slugs for thumbs-down feedback. The UI translates
+# them for display; storage and analytics only ever see the slug, so feedback
+# submitted under different UI languages stays aggregatable.
+VALID_FEEDBACK_TAGS = frozenset(
+    {
+        "incorrect",
+        "not_as_expected",
+        "slow",
+        "style_tone",
+        "safety_legal",
+        "other",
+    }
+)
+
+
+class FeedbackError(Exception): ...
+
+
+class InvalidRatingError(FeedbackError): ...
+
+
+class InvalidTagError(FeedbackError): ...
+
+
+class DuplicateFeedbackError(FeedbackError): ...
+
+
+class RunNotFoundError(FeedbackError): ...
+
+
+@dataclass(frozen=True)
+class Feedback:
+    """A user's rating of a single run: at most one per (thread, run, user).
+
+    Single-entity aggregate. ``message_id`` optionally narrows the rating
+    to one message within the run instead of the whole run. ``tags`` carry
+    optional thumbs-down reason slugs from the feedback dialog.
+    """
+
+    feedback_id: str
+    run_id: str
+    thread_id: str
+    rating: int
+    user_id: str | None = None
+    message_id: str | None = None
+    comment: str | None = None
+    tags: tuple[str, ...] = ()
+    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+
+    def __post_init__(self):
+        if self.rating not in VALID_RATINGS:
+            raise InvalidRatingError(f"rating must be +1 or -1, got {self.rating}")
+        unknown = set(self.tags) - VALID_FEEDBACK_TAGS
+        if unknown:
+            raise InvalidTagError(f"unknown feedback tags: {sorted(unknown)}")
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        run_id: str,
+        thread_id: str,
+        rating: int,
+        user_id: str | None = None,
+        message_id: str | None = None,
+        comment: str | None = None,
+        tags: tuple[str, ...] | list[str] = (),
+    ) -> Feedback:
+        """Factory: generate identity and validate invariants."""
+        return cls(
+            feedback_id=str(uuid.uuid4()),
+            run_id=run_id,
+            thread_id=thread_id,
+            rating=rating,
+            user_id=user_id,
+            message_id=message_id,
+            comment=comment,
+            tags=tuple(tags),
+        )

backend/packages/harness/deerflow/domain/feedback/ports.py

@@ -0,0 +1,69 @@
+from typing import Protocol, runtime_checkable
+
+from deerflow.domain.feedback.model import Feedback
+
+
+@runtime_checkable
+class FeedbackRepository(Protocol):
+    """Output port for feedback persistence.
+
+    Implementations exchange domain objects only and translate storage
+    failures into domain errors -- no storage vocabulary (SQL, tables,
+    files) may leak through this contract.
+    """
+
+    async def save(self, feedback: Feedback) -> Feedback:
+        """Store the user's current rating for a run (idempotent upsert).
+
+        Creates the entry if absent; otherwise replaces rating/comment and
+        refreshes created_at while keeping the aggregate identity
+        (thread_id, run_id, user_id). Returns the stored state.
+        """
+        ...
+
+    async def latest_per_run_in_thread(self, thread_id: str, *, user_id: str | None) -> dict[str, Feedback]:
+        """Return each run's current feedback across a whole thread, keyed by run_id.
+
+        Single bulk read used to badge the message list -- avoids one
+        query per run. Ownership filter: a non-None user_id restricts results to the
+        user's entries; None means no filtering (no-auth mode).
+        """
+        ...
+
+    async def latest_for_runs(self, thread_id: str, run_ids: set[str], *, user_id: str | None) -> dict[str, Feedback]:
+        """Return current feedback for only the selected runs of a thread.
+
+        Paged variant of latest_per_run_in_thread: the message-list page
+        endpoint only needs badges for the runs on the current page.
+        Returns an empty mapping when run_ids is empty. Same ownership
+        filter as find_for_run.
+        """
+        ...
+
+    async def remove_for_run(self, thread_id: str, run_id: str, *, user_id: str | None) -> bool:
+        """Retract the user's feedback for a run.
+
+        Returns True if an entry was removed, False if none existed.
+        Ownership filter: a non-None user_id restricts results to the
+        user's entries; None means no filtering (no-auth mode).
+        """
+        ...
+
+
+class RunLookup(Protocol):
+    """Narrow output port: the only question the feedback context asks
+    about runs -- which thread does a run belong to.
+
+    Deliberately not the full run store: depending on this one-method
+    contract keeps the feedback context decoupled from the execution
+    context's wide repository interface (interface segregation).
+    """
+
+    async def thread_of(self, run_id: str) -> str | None:
+        """Return the thread_id owning the given run, or None if the run
+        does not exist.
+
+        Used by the service to verify run ownership before writing
+        feedback (rejects cross-thread run ids as RunNotFoundError).
+        """
+        ...

backend/packages/harness/deerflow/domain/feedback/service.py

@@ -0,0 +1,71 @@
+from deerflow.domain.feedback.model import Feedback, RunNotFoundError
+from deerflow.domain.feedback.ports import FeedbackRepository, RunLookup
+
+
+class FeedbackService:
+    """Input port of the feedback context (application service).
+
+    Orchestrates use cases only: fetch/verify -> apply domain rules ->
+    persist through output ports. Holds no business rules itself (those
+    live on the Feedback aggregate) and knows nothing about HTTP or
+    storage. user_id is always passed in explicitly -- resolving the
+    current user is the primary adapter's job.
+    """
+
+    def __init__(self, repository: FeedbackRepository, runs: RunLookup):
+        self._repository = repository
+        self._runs = runs
+
+    async def _require_run(self, thread_id: str, run_id: str) -> None:
+        """Reject runs that do not exist or belong to another thread."""
+        if await self._runs.thread_of(run_id) != thread_id:
+            raise RunNotFoundError("Run does not belong to the specified thread")
+
+    async def rate_run(
+        self,
+        thread_id: str,
+        run_id: str,
+        *,
+        rating: int,
+        comment: str | None,
+        user_id: str | None,
+        tags: tuple[str, ...] | list[str] = (),
+    ) -> Feedback:
+        """Set the user's current rating for a run (idempotent).
+
+        Backs the PUT endpoint: "my current verdict on this run is X".
+        Verifies run ownership first, then creates the aggregate (which
+        validates the rating) and stores it with upsert-by-identity
+        semantics -- repeated calls replace the previous rating.
+
+        Raises:
+            RunNotFoundError: the run does not exist or does not belong
+                to the given thread (cross-thread ids are rejected).
+            InvalidRatingError: rating is not +1 or -1 (raised by the
+                aggregate factory before any I/O happens).
+        """
+        await self._require_run(thread_id, run_id)
+        feedback = Feedback.create(
+            run_id=run_id,
+            thread_id=thread_id,
+            rating=rating,
+            user_id=user_id,
+            comment=comment,
+            tags=tags,
+        )
+        return await self._repository.save(feedback)
+
+    async def retract_run_rating(self, thread_id: str, run_id: str, *, user_id: str | None) -> bool:
+        """Withdraw the user's rating for a run (clicking the active button
+        again). Returns False when there was nothing to retract."""
+        return await self._repository.remove_for_run(thread_id, run_id, user_id=user_id)
+
+    async def latest_per_run_in_thread(self, thread_id: str, *, user_id: str | None) -> dict[str, Feedback]:
+        """Current feedback per run across a thread -- powers the message-list
+        thumb badges (full-list path)."""
+        return await self._repository.latest_per_run_in_thread(thread_id, user_id=user_id)
+
+    async def latest_for_runs(self, thread_id: str, run_ids: set[str], *, user_id: str | None) -> dict[str, Feedback]:
+        """Current feedback for the selected runs only -- powers the paged
+        message-list badges."""
+        return await self._repository.latest_for_runs(thread_id, run_ids, user_id=user_id)

backend/tests/test_feedback_domain.py

@@ -0,0 +1,49 @@
+"""Pure unit tests for the Feedback aggregate — zero IO, no engine."""
+
+import pytest
+
+from deerflow.domain.feedback import Feedback, InvalidRatingError, InvalidTagError
+
+
+class TestFeedbackAggregate:
+    def test_factory_generates_identity(self):
+        fb = Feedback.create(run_id="r1", thread_id="t1", rating=1)
+        assert fb.feedback_id
+        assert fb.run_id == "r1"
+        assert fb.thread_id == "t1"
+
+    def test_factory_distinct_identities(self):
+        a = Feedback.create(run_id="r1", thread_id="t1", rating=1)
+        b = Feedback.create(run_id="r1", thread_id="t1", rating=1)
+        assert a.feedback_id != b.feedback_id
+
+    def test_created_at_is_tz_aware(self):
+        fb = Feedback.create(run_id="r1", thread_id="t1", rating=1)
+        assert fb.created_at.tzinfo is not None
+
+    @pytest.mark.parametrize("rating", [0, 5, -2, 2])
+    def test_invalid_rating_raises_before_any_io(self, rating):
+        with pytest.raises(InvalidRatingError):
+            Feedback.create(run_id="r1", thread_id="t1", rating=rating)
+
+    @pytest.mark.parametrize("rating", [1, -1])
+    def test_valid_ratings(self, rating):
+        assert Feedback.create(run_id="r1", thread_id="t1", rating=rating).rating == rating
+
+    def test_valid_tags_accepted(self):
+        fb = Feedback.create(run_id="r1", thread_id="t1", rating=-1, tags=["incorrect", "slow"])
+        assert fb.tags == ("incorrect", "slow")
+
+    def test_unknown_tag_rejected(self):
+        with pytest.raises(InvalidTagError):
+            Feedback.create(run_id="r1", thread_id="t1", rating=-1, tags=["bogus"])
+
+    def test_direct_construction_also_validates(self):
+        # Invariants live on the aggregate, not only on the factory.
+        with pytest.raises(InvalidRatingError):
+            Feedback(feedback_id="x", run_id="r1", thread_id="t1", rating=0)
+
+    def test_frozen(self):
+        fb = Feedback.create(run_id="r1", thread_id="t1", rating=1)
+        with pytest.raises(AttributeError):
+            fb.rating = -1  # type: ignore[misc]

backend/tests/test_harness_domain_purity.py

@@ -0,0 +1,72 @@
+"""Purity check: the hexagonal inner ring must stay infrastructure-free.
+
+``packages/harness/deerflow/domain/`` holds domain models, ports, and
+application services. Per the hexagonal layering rules it may depend on the
+standard library (and other domain modules) only — no ORM/HTTP/validation
+frameworks, no app layer, and no harness infrastructure modules. This is the
+machine-checkable form of "the domain doesn't depend on any other module".
+
+Same AST-scan strength (and same accepted blind spots, e.g. dynamic imports)
+as test_harness_boundary.py.
+"""
+
+import ast
+from pathlib import Path
+
+DOMAIN_ROOT = Path(__file__).parent.parent / "packages" / "harness" / "deerflow" / "domain"
+
+BANNED_PREFIXES = (
+    # third-party infrastructure
+    "sqlalchemy",
+    "alembic",
+    "aiosqlite",
+    "fastapi",
+    "starlette",
+    "pydantic",
+    "langchain",
+    "langgraph",
+    # app layer (also covered by test_harness_boundary, restated for clarity)
+    "app",
+    # harness infrastructure modules
+    "deerflow.persistence",
+    "deerflow.runtime",
+    "deerflow.mcp",
+    "deerflow.sandbox",
+    "deerflow.agents",
+    "deerflow.tools",
+    "deerflow.skills",
+    "deerflow.community",
+    "deerflow.tui",
+)
+
+
+def _collect_imports(filepath: Path) -> list[tuple[int, str]]:
+    """Return (line_number, module_path) for every import in *filepath*."""
+    source = filepath.read_text(encoding="utf-8")
+    try:
+        tree = ast.parse(source, filename=str(filepath))
+    except SyntaxError:
+        return []
+
+    results: list[tuple[int, str]] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                results.append((node.lineno, alias.name))
+        elif isinstance(node, ast.ImportFrom):
+            if node.module:
+                results.append((node.lineno, node.module))
+    return results
+
+
+def test_domain_has_no_infrastructure_imports():
+    assert DOMAIN_ROOT.is_dir(), f"domain package missing: {DOMAIN_ROOT}"
+    violations: list[str] = []
+
+    for py_file in sorted(DOMAIN_ROOT.rglob("*.py")):
+        for lineno, module in _collect_imports(py_file):
+            if any(module == prefix or module.startswith(prefix + ".") for prefix in BANNED_PREFIXES):
+                rel = py_file.relative_to(DOMAIN_ROOT.parent.parent.parent)
+                violations.append(f"  {rel}:{lineno}  imports {module}")
+
+    assert not violations, "Hexagonal inner ring (domain/) must stay infrastructure-free:\n" + "\n".join(violations)