diff --git a/backend/packages/harness/deerflow/domain/__init__.py b/backend/packages/harness/deerflow/domain/__init__.py new file mode 100644 index 000000000..75edaa1c9 --- /dev/null +++ b/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`. +""" diff --git a/backend/packages/harness/deerflow/domain/feedback/__init__.py b/backend/packages/harness/deerflow/domain/feedback/__init__.py new file mode 100644 index 000000000..172eeed4b --- /dev/null +++ b/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", +] diff --git a/backend/packages/harness/deerflow/domain/feedback/model.py b/backend/packages/harness/deerflow/domain/feedback/model.py new file mode 100644 index 000000000..277c0c3cb --- /dev/null +++ b/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), + ) diff --git a/backend/packages/harness/deerflow/domain/feedback/ports.py b/backend/packages/harness/deerflow/domain/feedback/ports.py new file mode 100644 index 000000000..3a81126f6 --- /dev/null +++ b/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). + """ + ... diff --git a/backend/packages/harness/deerflow/domain/feedback/service.py b/backend/packages/harness/deerflow/domain/feedback/service.py new file mode 100644 index 000000000..85d33a00d --- /dev/null +++ b/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) diff --git a/backend/tests/test_feedback_domain.py b/backend/tests/test_feedback_domain.py new file mode 100644 index 000000000..9c85c98f3 --- /dev/null +++ b/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] diff --git a/backend/tests/test_harness_domain_purity.py b/backend/tests/test_harness_domain_purity.py new file mode 100644 index 000000000..ff83857c6 --- /dev/null +++ b/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)